Merge pull request #201 from matrix-org/kegan/max-conn-env-var

Add SYNCV3_MAX_DB_CONN: use it in e2e tests

.github/workflows/tests.yml

@@ -88,6 +88,10 @@ jobs:
                 if-no-files-found: error
     end_to_end:
         runs-on: ubuntu-latest
+        strategy:
+          matrix:
+            # test with unlimited + 1 + 2 max db conns. If we end up double transacting in the tests anywhere, conn=1 tests will fail.
+            max_db_conns: [0,1,2]
         services:
             synapse:
                 # Custom image built from https://github.com/matrix-org/synapse/tree/v1.72.0/docker/complement with a dummy /complement/ca set
@@ -97,6 +101,11 @@ jobs:
                     SERVER_NAME: synapse
                 ports:
                     - 8008:8008
+                # Set health checks to wait until synapse has started
+                options: >-
+                    --health-interval 10s
+                    --health-timeout 5s
+                    --health-retries 5
             # Label used to access the service container
             postgres:
                 # Docker Hub image
@@ -142,6 +151,7 @@ jobs:
                   SYNCV3_DB: user=postgres dbname=syncv3 sslmode=disable password=postgres host=localhost
                   SYNCV3_SERVER: http://localhost:8008
                   SYNCV3_SECRET: itsasecret
+                  SYNCV3_MAX_DB_CONN: ${{ matrix.max_db_conns }}
                   E2E_TEST_SERVER_STDOUT: test-e2e-server.log
 
             - name: Upload test log

cmd/syncv3/main.go

@@ -6,6 +6,7 @@ import (
 	_ "net/http/pprof"
 	"os"
 	"os/signal"
+	"strconv"
 	"strings"
 	"syscall"
 	"time"
@@ -40,6 +41,7 @@ const (
 	EnvJaeger     = "SYNCV3_JAEGER_URL"
 	EnvSentryDsn  = "SYNCV3_SENTRY_DSN"
 	EnvLogLevel   = "SYNCV3_LOG_LEVEL"
+	EnvMaxConns   = "SYNCV3_MAX_DB_CONN"
 )
 
 var helpMsg = fmt.Sprintf(`
@@ -55,7 +57,8 @@ Environment var
 %s Default: unset. The Jaeger URL to send spans to e.g http://localhost:14268/api/traces - if unset does not send OTLP traces.
 %s Default: unset. The Sentry DSN to report events to e.g https://[email protected]/123 - if unset does not send sentry events.
 %s  Default: info. The level of verbosity for messages logged. Available values are trace, debug, info, warn, error and fatal
-`, EnvServer, EnvDB, EnvSecret, EnvBindAddr, EnvTLSCert, EnvTLSKey, EnvPPROF, EnvPrometheus, EnvJaeger, EnvSentryDsn, EnvLogLevel)
+%s Default: unset. Max database connections to use when communicating with postgres. Unset or 0 means no limit.
+`, EnvServer, EnvDB, EnvSecret, EnvBindAddr, EnvTLSCert, EnvTLSKey, EnvPPROF, EnvPrometheus, EnvJaeger, EnvSentryDsn, EnvLogLevel, EnvMaxConns)
 
 func defaulting(in, dft string) string {
 	if in == "" {
@@ -81,6 +84,7 @@ func main() {
 		EnvJaeger:     os.Getenv(EnvJaeger),
 		EnvSentryDsn:  os.Getenv(EnvSentryDsn),
 		EnvLogLevel:   os.Getenv(EnvLogLevel),
+		EnvMaxConns:   defaulting(os.Getenv(EnvMaxConns), "0"),
 	}
 	requiredEnvVars := []string{EnvServer, EnvDB, EnvSecret, EnvBindAddr}
 	for _, requiredEnvVar := range requiredEnvVars {
@@ -136,6 +140,8 @@ func main() {
 		}
 	}
 
+	fmt.Printf("Debug=%v LogLevel=%v MaxConns=%v\n", args[EnvDebug] == "1", args[EnvLogLevel], args[EnvMaxConns])
+
 	if args[EnvDebug] == "1" {
 		zerolog.SetGlobalLevel(zerolog.TraceLevel)
 	} else {
@@ -162,9 +168,13 @@ func main() {
 		panic(err)
 	}
 
+	maxConnsInt, err := strconv.Atoi(args[EnvMaxConns])
+	if err != nil {
+		panic("invalid value for " + EnvMaxConns + ": " + args[EnvMaxConns])
+	}
 	h2, h3 := syncv3.Setup(args[EnvServer], args[EnvDB], args[EnvSecret], syncv3.Opts{
 		AddPrometheusMetrics: args[EnvPrometheus] != "",
-		DBMaxConns:           100,
+		DBMaxConns:           maxConnsInt,
 		DBConnMaxIdleTime:    time.Hour,
 	})
 

state/storage.go

@@ -57,6 +57,10 @@ func NewStorage(postgresURI string) *Storage {
 		// TODO: if we panic(), will sentry have a chance to flush the event?
 		logger.Panic().Err(err).Str("uri", postgresURI).Msg("failed to open SQL DB")
 	}
+	return NewStorageWithDB(db)
+}
+
+func NewStorageWithDB(db *sqlx.DB) *Storage {
 	acc := &Accumulator{
 		db:            db,
 		roomsTable:    NewRoomsTable(db),

sync2/storage.go

@@ -26,6 +26,10 @@ func NewStore(postgresURI, secret string) *Storage {
 		// TODO: if we panic(), will sentry have a chance to flush the event?
 		logger.Panic().Err(err).Str("uri", postgresURI).Msg("failed to open SQL DB")
 	}
+	return NewStoreWithDB(db, secret)
+}
+
+func NewStoreWithDB(db *sqlx.DB, secret string) *Storage {
 	return &Storage{
 		DevicesTable: NewDevicesTable(db),
 		TokensTable:  NewTokensTable(db, secret),

v3.go

@@ -77,20 +77,24 @@ func Setup(destHomeserver, postgresURI, secret string, opts Opts) (*handler2.Han
 		},
 		DestinationServer: destHomeserver,
 	}
-	store := state.NewStorage(postgresURI)
-	storev2 := sync2.NewStore(postgresURI, secret)
-	for _, db := range []*sqlx.DB{store.DB, storev2.DB} {
-		if opts.DBMaxConns > 0 {
-			// https://github.com/go-sql-driver/mysql#important-settings
-			// "db.SetMaxIdleConns() is recommended to be set same to db.SetMaxOpenConns(). When it is smaller
-			// than SetMaxOpenConns(), connections can be opened and closed much more frequently than you expect."
-			db.SetMaxOpenConns(opts.DBMaxConns)
-			db.SetMaxIdleConns(opts.DBMaxConns)
-		}
-		if opts.DBConnMaxIdleTime > 0 {
-			db.SetConnMaxIdleTime(opts.DBConnMaxIdleTime)
-		}
+	db, err := sqlx.Open("postgres", postgresURI)
+	if err != nil {
+		sentry.CaptureException(err)
+		// TODO: if we panic(), will sentry have a chance to flush the event?
+		logger.Panic().Err(err).Str("uri", postgresURI).Msg("failed to open SQL DB")
+	}
+	if opts.DBMaxConns > 0 {
+		// https://github.com/go-sql-driver/mysql#important-settings
+		// "db.SetMaxIdleConns() is recommended to be set same to db.SetMaxOpenConns(). When it is smaller
+		// than SetMaxOpenConns(), connections can be opened and closed much more frequently than you expect."
+		db.SetMaxOpenConns(opts.DBMaxConns)
+		db.SetMaxIdleConns(opts.DBMaxConns)
+	}
+	if opts.DBConnMaxIdleTime > 0 {
+		db.SetConnMaxIdleTime(opts.DBConnMaxIdleTime)
 	}
+	store := state.NewStorageWithDB(db)
+	storev2 := sync2.NewStoreWithDB(db, secret)
 	bufferSize := 50
 	deviceDataUpdateFrequency := time.Second
 	if opts.TestingSynchronousPubsub {