fix(db): Harden embedded and ephemeral database test lifecycle

Bound database probes and operations to prevent hung test runs, and clean up started servers when lock release fails.
Harden stale-database cleanup against malformed or future-dated names and parse PostgreSQL settings correctly.
Extend the e2e timeout for slower macOS runners and expanded coverage.

Author: moshloop

.github/workflows/e2e.yml

@@ -35,7 +35,9 @@ jobs:
       matrix:
         os: [ubuntu-latest, macos-latest]
         go-version: ["1.26"]
-    timeout-minutes: 30
+    # Ginkgo e2e (10m) + embedded pool tests (10m) + helper unit tests (5m) +
+    # a full coverage re-run of the e2e suite, on a slower macOS runner.
+    timeout-minutes: 45
 
     # Exercises the default embedded-postgres path on both platforms. The
     # external-DB path is covered by e2e-external, which needs a Linux-only

db/embedded_lifecycle.go

@@ -8,6 +8,7 @@ import (
 	"net"
 	"os"
 	"path/filepath"
+	"strconv"
 	"sync"
 	"time"
 
@@ -20,6 +21,7 @@ import (
 const (
 	embeddedStartupLockTimeout = 5 * time.Minute
 	embeddedStartupRetryDelay  = 100 * time.Millisecond
+	embeddedProbeTimeout       = 200 * time.Millisecond
 )
 
 var embeddedStartupMu sync.Mutex
@@ -123,9 +125,14 @@ func (r *embeddedRuntime) withStartupLock(
 
 	dsn, stop, startErr := fn()
 	unlockErr := lock.Unlock()
-	if startErr != nil || unlockErr != nil {
+	if startErr != nil {
 		return "", nil, errors.Join(startErr, unlockErr)
 	}
+	if unlockErr != nil {
+		// The server did start, so its stop closer must not be dropped on the
+		// floor: shut it back down rather than leaking an unmanaged cluster.
+		return "", nil, errors.Join(unlockErr, stop())
+	}
 	return dsn, stop, nil
 }
 
@@ -178,7 +185,9 @@ func (r *embeddedRuntime) resolveLockedConfig() error {
 }
 
 func (r *embeddedRuntime) reuseRunningServer(dsn string) (bool, error) {
-	if !portIsListening(r.port) {
+	ctx, cancel := context.WithTimeout(context.Background(), embeddedProbeTimeout)
+	defer cancel()
+	if !portIsListening(ctx, r.port) {
 		return false, nil
 	}
 	if err := r.validateServer(dsn); err != nil {
@@ -245,7 +254,6 @@ func validateEmbeddedDataDirectory(ctx context.Context, conn *pgx.Conn, expected
 
 func validateFastTestingServer(ctx context.Context, conn *pgx.Conn) error {
 	var fsync, synchronousCommit, fullPageWrites string
-	var maxConnections int
 	if err := conn.QueryRow(ctx, "SHOW fsync").Scan(&fsync); err != nil {
 		return fmt.Errorf("read fsync: %w", err)
 	}
@@ -255,12 +263,29 @@ func validateFastTestingServer(ctx context.Context, conn *pgx.Conn) error {
 	if err := conn.QueryRow(ctx, "SHOW full_page_writes").Scan(&fullPageWrites); err != nil {
 		return fmt.Errorf("read full_page_writes: %w", err)
 	}
-	if err := conn.QueryRow(ctx, "SHOW max_connections").Scan(&maxConnections); err != nil {
-		return fmt.Errorf("read max_connections: %w", err)
+	maxConnections, err := showInt(ctx, conn, "max_connections")
+	if err != nil {
+		return err
 	}
 	return validateFastTestingSettings(fsync, synchronousCommit, fullPageWrites, maxConnections)
 }
 
+// showInt reads a numeric server setting. SHOW always reports text (OID 25)
+// regardless of the underlying GUC type, so the value has to be parsed rather
+// than scanned straight into an int. setting is always a package-level literal;
+// SHOW takes no bind parameters.
+func showInt(ctx context.Context, conn *pgx.Conn, setting string) (int, error) {
+	var raw string
+	if err := conn.QueryRow(ctx, "SHOW "+setting).Scan(&raw); err != nil {
+		return 0, fmt.Errorf("read %s: %w", setting, err)
+	}
+	value, err := strconv.Atoi(raw)
+	if err != nil {
+		return 0, fmt.Errorf("parse %s %q: %w", setting, raw, err)
+	}
+	return value, nil
+}
+
 func (r *embeddedRuntime) serverConfig() embeddedpostgres.Config {
 	config := embeddedpostgres.DefaultConfig().
 		Port(r.port).
@@ -299,8 +324,9 @@ func (r *embeddedRuntime) dsn() string {
 	)
 }
 
-func portIsListening(port uint32) bool {
-	conn, err := net.DialTimeout("tcp", fmt.Sprintf("localhost:%d", port), 200*time.Millisecond)
+func portIsListening(ctx context.Context, port uint32) bool {
+	var dialer net.Dialer
+	conn, err := dialer.DialContext(ctx, "tcp", fmt.Sprintf("localhost:%d", port))
 	if err != nil {
 		return false
 	}

db/embedded_lifecycle_integration_test.go

@@ -47,12 +47,12 @@ var _ = Describe("embedded PostgreSQL lifecycle", Label("integration"), func() {
 		defer conn.Close(context.Background()) //nolint:errcheck
 
 		var dataDirectory, fsync, synchronousCommit, fullPageWrites string
-		var maxConnections int
 		Expect(conn.QueryRow(ctx, "SHOW data_directory").Scan(&dataDirectory)).To(Succeed())
 		Expect(conn.QueryRow(ctx, "SHOW fsync").Scan(&fsync)).To(Succeed())
 		Expect(conn.QueryRow(ctx, "SHOW synchronous_commit").Scan(&synchronousCommit)).To(Succeed())
 		Expect(conn.QueryRow(ctx, "SHOW full_page_writes").Scan(&fullPageWrites)).To(Succeed())
-		Expect(conn.QueryRow(ctx, "SHOW max_connections").Scan(&maxConnections)).To(Succeed())
+		maxConnections, err := showInt(ctx, conn, "max_connections")
+		Expect(err).NotTo(HaveOccurred())
 
 		expectedDataDirectory, err := filepath.EvalSymlinks(filepath.Join(config.DataDir, "data"))
 		Expect(err).NotTo(HaveOccurred())

dbtest/dbtest.go

@@ -35,6 +35,10 @@ const (
 	EnvCreate = "COMMONS_DB_CREATE"
 )
 
+// connectTimeout bounds the blocking Postgres calls made while resolving a test
+// database, so an unreachable server fails the test instead of hanging it.
+const connectTimeout = 30 * time.Second
+
 // Options configures how a test database is resolved.
 type Options struct {
 	// Name seeds the database name. Required.
@@ -84,7 +88,9 @@ func (d *DB) SQL() *sql.DB {
 		d.fail(fmt.Errorf("open %s: %w", redact(d.dsn), err))
 		return nil
 	}
-	if err := handle.Ping(); err != nil {
+	ctx, cancel := context.WithTimeout(context.Background(), connectTimeout)
+	defer cancel()
+	if err := handle.PingContext(ctx); err != nil {
 		_ = handle.Close()
 		d.fail(fmt.Errorf("ping %s: %w", redact(d.dsn), err))
 		return nil
@@ -164,7 +170,7 @@ func Open(opts Options) (*DB, func() error, error) {
 		if err != nil {
 			return nil, nil, err
 		}
-		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+		ctx, cancel := context.WithTimeout(context.Background(), connectTimeout)
 		defer cancel()
 		dsn, drop, err := acquirePooledScratch(ctx, adminURL, opts.Name, db.unique, time.Now())
 		if err != nil {

dbtest/ephemeral.go

@@ -49,8 +49,13 @@ func managedDatabaseName(prefix string, created time.Time, unique, name string)
 	if name == "" {
 		return base
 	}
-	description := sanitize(name)
+	// The identity is what keeps concurrent runs apart, so when it fills the
+	// identifier there is no room left for a human-readable description.
 	room := maxIdentifier - len(base) - 1
+	if room <= 0 {
+		return base
+	}
+	description := sanitize(name)
 	if len(description) > room {
 		description = description[:room]
 	}

dbtest/ephemeral_pool.go

@@ -8,6 +8,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/flanksource/commons/logger"
 	"github.com/jackc/pgerrcode"
 	"github.com/lib/pq"
 )
@@ -217,16 +218,22 @@ func (p *ephemeralPool) cleanupStale(
 		return err
 	}
 	for _, name := range names {
+		// An unparseable or future-dated entry is skipped rather than fatal.
+		// cleanupStale runs under the pool-wide advisory lock, so aborting here
+		// would wedge every checkout for this prefix pair until someone cleaned
+		// up by hand - and a clock step-back is enough to produce one.
 		created, err := managedDatabaseCreatedWithPrefixes(
 			name,
 			p.config.PoolPrefix,
 			p.config.TestPrefix,
 		)
 		if err != nil {
-			return err
+			logger.Warnf("dbtest: skipping unrecognised managed database %q: %v", name, err)
+			continue
 		}
 		if created.After(now) {
-			return fmt.Errorf("managed database name %q has a future timestamp", name)
+			logger.Warnf("dbtest: skipping managed database %q with a future timestamp %s", name, created)
+			continue
 		}
 		if now.Sub(created) <= p.config.MaxAge {
 			continue

dbtest/ephemeral_test.go

@@ -1,6 +1,7 @@
 package dbtest
 
 import (
+	"fmt"
 	"path/filepath"
 	"strings"
 	"time"
@@ -58,6 +59,23 @@ var _ = Describe("ephemeral database configuration", func() {
 		Expect(name).To(ContainSubstring(testUnique))
 	})
 
+	It("drops the description entirely when the identity fills the identifier", func() {
+		// prefix + unix seconds + "_" + unique consumes all 63 characters, so
+		// there is no room left for even a one-character description.
+		identity := fmt.Sprintf("%s%d_", testDatabasePrefix, created.Unix())
+		unique := strings.Repeat("u", maxIdentifier-len(identity))
+
+		name := managedDatabaseName(testDatabasePrefix, created, unique, testName)
+
+		Expect(len(name)).To(Equal(maxIdentifier))
+		Expect(name).To(Equal(identity + unique))
+		Expect(name).NotTo(ContainSubstring("migration_runner"))
+
+		parsed, err := managedDatabaseCreated(name)
+		Expect(err).NotTo(HaveOccurred())
+		Expect(parsed).To(Equal(created))
+	})
+
 	DescribeTable("rejects names outside the managed namespace",
 		func(name string) {
 			_, err := managedDatabaseCreated(name)

dbtest/scratch.go

@@ -1,6 +1,7 @@
 package dbtest
 
 import (
+	"context"
 	"crypto/rand"
 	"database/sql"
 	"encoding/hex"
@@ -36,11 +37,14 @@ func createScratch(adminURL, name, unique string) (string, func() error, error)
 	}
 	defer admin.Close() //nolint:errcheck
 
+	ctx, cancel := context.WithTimeout(context.Background(), connectTimeout)
+	defer cancel()
+
 	quoted := pq.QuoteIdentifier(dbName)
-	if _, err := admin.Exec(`DROP DATABASE IF EXISTS ` + quoted + ` WITH (FORCE)`); err != nil {
+	if _, err := admin.ExecContext(ctx, `DROP DATABASE IF EXISTS `+quoted+` WITH (FORCE)`); err != nil {
 		return "", nil, fmt.Errorf("drop stale database %s: %w", dbName, err)
 	}
-	if _, err := admin.Exec(`CREATE DATABASE ` + quoted); err != nil {
+	if _, err := admin.ExecContext(ctx, `CREATE DATABASE `+quoted); err != nil {
 		return "", nil, fmt.Errorf("create database %s: %w", dbName, err)
 	}
 
@@ -55,7 +59,9 @@ func createScratch(adminURL, name, unique string) (string, func() error, error)
 			return fmt.Errorf("open %s to drop %s: %w", redact(adminURL), dbName, err)
 		}
 		defer cleanup.Close() //nolint:errcheck
-		if _, err := cleanup.Exec(`DROP DATABASE IF EXISTS ` + quoted + ` WITH (FORCE)`); err != nil {
+		dropCtx, dropCancel := context.WithTimeout(context.Background(), connectTimeout)
+		defer dropCancel()
+		if _, err := cleanup.ExecContext(dropCtx, `DROP DATABASE IF EXISTS `+quoted+` WITH (FORCE)`); err != nil {
 			return fmt.Errorf("drop database %s: %w", dbName, err)
 		}
 		return nil