Elect primary that is most caught up with replication

Run a query for new primary selection that calculates a replication
queue "size" that uses a heuristic for number of incomplete jobs
(dead, in_progress, ready, failed) since the last completed replication job.

2 files changed, 14 insertions, 13 deletions

diff --git a/internal/praefect/config/config.go b/internal/praefect/config/config.go
index 4d948f699..3508c4a23 100644
--- a/internal/praefect/config/config.go
+++ b/internal/praefect/config/config.go
@@ -148,20 +148,21 @@ func (c Config) NeedsSQL() bool {
 
 // DB holds Postgres client configuration data.
 type DB struct {
-	Host        string `toml:"host"`
-	Port        int    `toml:"port"`
-	User        string `toml:"user"`
-	Password    string `toml:"password"`
-	DBName      string `toml:"dbname"`
-	SSLMode     string `toml:"sslmode"`
-	SSLCert     string `toml:"sslcert"`
-	SSLKey      string `toml:"sslkey"`
-	SSLRootCert string `toml:"sslrootcert"`
+	Host                         string `toml:"host"`
+	Port                         int    `toml:"port"`
+	User                         string `toml:"user"`
+	Password                     string `toml:"password"`
+	DBName                       string `toml:"dbname"`
+	SSLMode                      string `toml:"sslmode"`
+	SSLCert                      string `toml:"sslcert"`
+	SSLKey                       string `toml:"sslkey"`
+	SSLRootCert                  string `toml:"sslrootcert"`
+	StatementTimeoutMilliseconds int    `toml:"default_timeout_ms"`
 }
 
 // ToPQString returns a connection string that can be passed to github.com/lib/pq.
 func (db DB) ToPQString() string {
-	var fields []string
+	fields := []string{fmt.Sprintf("statement_timeout=%d", db.StatementTimeoutMilliseconds)}
 	if db.Port > 0 {
 		fields = append(fields, fmt.Sprintf("port=%d", db.Port))
 	}
diff --git a/internal/praefect/config/config_test.go b/internal/praefect/config/config_test.go
index eed4abe3b..041c30188 100644
--- a/internal/praefect/config/config_test.go
+++ b/internal/praefect/config/config_test.go
@@ -199,7 +199,7 @@ func TestToPQString(t *testing.T) {
 		in   DB
 		out  string
 	}{
-		{desc: "empty", in: DB{}, out: "binary_parameters=yes"},
+		{desc: "empty", in: DB{}, out: "statement_timeout=0 binary_parameters=yes"},
 		{
 			desc: "basic example",
 			in: DB{
@@ -213,14 +213,14 @@ func TestToPQString(t *testing.T) {
 				SSLKey:      "/path/to/key",
 				SSLRootCert: "/path/to/root-cert",
 			},
-			out: `port=2345 host=1.2.3.4 user=praefect-user password=secret dbname=praefect_production sslmode=require sslcert=/path/to/cert sslkey=/path/to/key sslrootcert=/path/to/root-cert binary_parameters=yes`,
+			out: `statement_timeout=0 port=2345 host=1.2.3.4 user=praefect-user password=secret dbname=praefect_production sslmode=require sslcert=/path/to/cert sslkey=/path/to/key sslrootcert=/path/to/root-cert binary_parameters=yes`,
 		},
 		{
 			desc: "with spaces and quotes",
 			in: DB{
 				Password: "secret foo'bar",
 			},
-			out: `password=secret\ foo\'bar binary_parameters=yes`,
+			out: `statement_timeout=0 password=secret\ foo\'bar binary_parameters=yes`,
 		},
 	}