From 81844a3210029892d9d6f6843762fe60a5bac5dc Mon Sep 17 00:00:00 2001 From: Peli de Halleux <4175913+pelikhan@users.noreply.github.com> Date: Thu, 12 Nov 2026 09:59:38 -0800 Subject: [PATCH] Clone per-test Postgres databases from a provisioned template; trim redundant fixtures Since the SQLite removal, every testutil.Postgres fixture provisions a full canonical store (~360 daily partitions, ~5s), pushing internal/server past the 10m go test timeout. Provision one template database per test process through the production store and clone it per test (CREATE DATABASE ... TEMPLATE), falling back to a private schema when the role cannot create databases. Reclaim databases left by exited processes. Also share one app across TestSessionArchiveResponseClosesReader subtests and stop re-executing repository queries already covered by the full dashboard corpus test. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../collect/native_control_process_test.go | 4 +- .../collect/run_enumeration_process_test.go | 3 +- server/internal/collect/run_worker_test.go | 4 +- server/internal/postgresx/native_test.go | 15 ++ .../server/dashboard_sql_corpus_test.go | 30 ++- .../internal/server/session_archive_test.go | 6 +- .../testutil/pgtemplate/pgtemplate.go | 208 ++++++++++++++++++ server/internal/testutil/postgres.go | 41 +++- 8 files changed, 287 insertions(+), 24 deletions(-) create mode 100644 server/internal/testutil/pgtemplate/pgtemplate.go diff --git a/server/internal/collect/native_control_process_test.go b/server/internal/collect/native_control_process_test.go index 31e906334e..297b985331 100644 --- a/server/internal/collect/native_control_process_test.go +++ b/server/internal/collect/native_control_process_test.go @@ -38,7 +38,7 @@ func TestNativeControlRecoveryUsesSeparateProcessAndDurableValueHint(t *testing. // #nosec G204 -- invokes this test executable without external collector tooling. child := exec.CommandContext(t.Context(), executable, "-test.run=^TestNativeControlRecoverySubprocess$", "-test.v") child.Env = append(os.Environ(), "PATH=/nonexistent", "CAO_NATIVE_CONTROL_DSN="+dsn, - "CAO_NATIVE_CONTROL_NAMESPACE="+namespace, "CAO_NATIVE_CONTROL_SCHEMA="+config.RuntimeParams["search_path"]) + "CAO_NATIVE_CONTROL_NAMESPACE="+namespace, "CAO_NATIVE_CONTROL_SCHEMA="+config.RuntimeParams["search_path"], "CAO_NATIVE_CONTROL_CANONICAL_DSN="+config.ConnString()) if output, err := child.CombinedOutput(); err != nil { t.Fatalf("separate native control process failed: %v\n%s", err, output) } @@ -68,7 +68,7 @@ func TestNativeControlRecoverySubprocess(t *testing.T) { t.Fatal(err) } defer func() { _ = store.Close() }() - config, err := pgx.ParseConfig(dsn) + config, err := pgx.ParseConfig(os.Getenv("CAO_NATIVE_CONTROL_CANONICAL_DSN")) if err != nil { t.Fatal(err) } diff --git a/server/internal/collect/run_enumeration_process_test.go b/server/internal/collect/run_enumeration_process_test.go index 5c7d77e25e..866bf280cd 100644 --- a/server/internal/collect/run_enumeration_process_test.go +++ b/server/internal/collect/run_enumeration_process_test.go @@ -40,6 +40,7 @@ func TestNativeEnumerationResumesAbandonedPageInSeparateProcess(t *testing.T) { child.Env = append(os.Environ(), "PATH=/nonexistent", "CAO_NATIVE_DISCOVERY_DSN="+dsn, "CAO_NATIVE_DISCOVERY_NAMESPACE="+namespace, "CAO_NATIVE_DISCOVERY_SCHEMA="+config.RuntimeParams["search_path"], + "CAO_NATIVE_DISCOVERY_CANONICAL_DSN="+config.ConnString(), "CAO_NATIVE_DISCOVERY_BEFORE="+before.Format(time.RFC3339)) if output, err := child.CombinedOutput(); err != nil { t.Fatalf("separate native enumeration process failed: %v\n%s", err, output) @@ -67,7 +68,7 @@ func TestNativeEnumerationRecoverySubprocess(t *testing.T) { t.Fatal(err) } defer func() { _ = store.Close() }() - config, err := pgx.ParseConfig(dsn) + config, err := pgx.ParseConfig(os.Getenv("CAO_NATIVE_DISCOVERY_CANONICAL_DSN")) if err != nil { t.Fatal(err) } diff --git a/server/internal/collect/run_worker_test.go b/server/internal/collect/run_worker_test.go index b8249a188b..a37c7dbd08 100644 --- a/server/internal/collect/run_worker_test.go +++ b/server/internal/collect/run_worker_test.go @@ -149,7 +149,7 @@ func TestNativeRunWorkerCrashAfterPublicationRecoversInAnotherProcess(t *testing child := exec.CommandContext(t.Context(), executable, "-test.run=^TestNativeRunWorkerRecoverySubprocess$", "-test.v") child.Env = []string{ "PATH=/nonexistent", "CAO_NATIVE_RUN_DSN=" + dsn, "CAO_NATIVE_RUN_NAMESPACE=" + namespace, - "CAO_NATIVE_RUN_SCHEMA=" + config.RuntimeParams["search_path"], "CAO_NATIVE_RUN_RECORDED=" + recorded, + "CAO_NATIVE_RUN_SCHEMA=" + config.RuntimeParams["search_path"], "CAO_NATIVE_RUN_CANONICAL_DSN=" + config.ConnString(), "CAO_NATIVE_RUN_RECORDED=" + recorded, } output, err := child.CombinedOutput() if err != nil || !strings.Contains(string(output), "historical durable handoff verified") { @@ -187,7 +187,7 @@ func TestNativeRunWorkerRecoverySubprocess(t *testing.T) { if err != nil || len(leases) != 1 || leases[0].Task.Attempt != 3 || leases[0].Task.Failures != 0 { t.Fatalf("recovered exact historical attempt: %v %v", leases, err) } - config, err := pgx.ParseConfig(dsn) + config, err := pgx.ParseConfig(os.Getenv("CAO_NATIVE_RUN_CANONICAL_DSN")) if err != nil { t.Fatal(err) } diff --git a/server/internal/postgresx/native_test.go b/server/internal/postgresx/native_test.go index 0c2a94a5c9..3e84dd6364 100644 --- a/server/internal/postgresx/native_test.go +++ b/server/internal/postgresx/native_test.go @@ -15,6 +15,7 @@ import ( "github.com/githubnext/gh-cao/server/internal/model" "github.com/githubnext/gh-cao/server/internal/query" + "github.com/githubnext/gh-cao/server/internal/testutil/pgtemplate" ) func nativeTestStore(t *testing.T) (*Store, *pgx.ConnConfig) { @@ -23,6 +24,20 @@ func nativeTestStore(t *testing.T) (*Store, *pgx.ConnConfig) { if endpoint == "" { t.Skip("POSTGRES_URL is unset") } + if config, ok := pgtemplate.Clone(t, t.Context(), endpoint, func(ctx context.Context, config *pgx.ConnConfig) error { + store, err := NewConfig(ctx, config) + if err != nil { + return err + } + return store.Close() + }); ok { + store, err := NewConfig(t.Context(), config) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = store.Close() }) + return store, config + } config, err := pgx.ParseConfig(endpoint) if err != nil { t.Fatal(err) diff --git a/server/internal/server/dashboard_sql_corpus_test.go b/server/internal/server/dashboard_sql_corpus_test.go index 2a1c11a694..2bf55716eb 100644 --- a/server/internal/server/dashboard_sql_corpus_test.go +++ b/server/internal/server/dashboard_sql_corpus_test.go @@ -16,19 +16,33 @@ import ( ) func TestDashboardQueryCorpusUsesPostgres(t *testing.T) { - assertDashboardQueriesUsePostgres(t, nil) + assertDashboardQueriesUsePostgres(t) } +// TestRepositoryQueriesUsePostgres guards that repository queries stay in the +// deployed corpus, which TestDashboardQueryCorpusUsesPostgres executes in full. func TestRepositoryQueriesUsePostgres(t *testing.T) { - assertDashboardQueriesUsePostgres(t, []string{ + definitions, _, err := dashboardSQLCorpus() + if err != nil { + t.Fatal(err) + } + deployed := make(map[string]bool, len(definitions)) + for _, definition := range definitions { + deployed[definition.Name] = true + } + for _, name := range []string{ "repository-run-statistics", "repository-run-identities", "repository-import-totals", "repository-run-totals", "repository-aic-totals", "repository-activity", "repository-observed-run-counts", "repository-run-inventory", - }) + } { + if !deployed[name] { + t.Errorf("repository query %q is missing from the deployed dashboard corpus", name) + } + } } -func assertDashboardQueriesUsePostgres(t *testing.T, requested []string) { +func assertDashboardQueriesUsePostgres(t *testing.T) { t.Helper() store := integrationDatabase(t) definitions, parameters, err := dashboardSQLCorpus() @@ -41,11 +55,9 @@ func assertDashboardQueriesUsePostgres(t *testing.T, requested []string) { if err := query.Validate(definitions); err != nil { t.Fatalf("validate deployed dashboard query corpus: %v", err) } - if requested == nil { - requested = make([]string, len(definitions)) - for index, definition := range definitions { - requested[index] = definition.Name - } + requested := make([]string, len(definitions)) + for index, definition := range definitions { + requested[index] = definition.Name } runtimeSource := func(_ context.Context, name string) (model.Source, error) { return model.Source{ diff --git a/server/internal/server/session_archive_test.go b/server/internal/server/session_archive_test.go index 0f97a30260..bc0b194209 100644 --- a/server/internal/server/session_archive_test.go +++ b/server/internal/server/session_archive_test.go @@ -223,9 +223,13 @@ func TestSessionArchiveResultBudgetsErrorsAndRevocation(t *testing.T) { } func TestSessionArchiveResponseClosesReader(t *testing.T) { + // Outcomes share one app; each subtest restores the credential it may revoke + // and installs its own archive service. + app, _ := newLocalPostgresTestApp(t, false) + accessToken := app.accessToken for _, outcome := range []string{"success", "withdrawn", "oversize", "authorization-revoked", "authorization-revoked-at-delivery", "query-error"} { t.Run(outcome, func(t *testing.T) { - app, _ := newLocalPostgresTestApp(t, false) + app.accessToken = accessToken reader := &sessionArchiveReadProbe{} delivered := false reader.deliver = func(deliver func() error) error { diff --git a/server/internal/testutil/pgtemplate/pgtemplate.go b/server/internal/testutil/pgtemplate/pgtemplate.go new file mode 100644 index 0000000000..27eb37696c --- /dev/null +++ b/server/internal/testutil/pgtemplate/pgtemplate.go @@ -0,0 +1,208 @@ +// Package pgtemplate clones per-test PostgreSQL databases from a template +// provisioned once per test process. +// +// Provisioning a canonical store creates hundreds of daily partitions, which +// costs seconds per test. Cloning a provisioned template costs a fraction of a +// second while still giving every test a private database. The package has no +// store dependency so the store's own package tests can use it. +// +// Clones live in separate databases, so packages tested concurrently against one +// server no longer serialize on the store's database-scoped advisory locks. +// Dashboard queries lock every daily partition, so concurrent package runs +// (go test -p N) need a larger max_locks_per_transaction, for example 512; +// CI runs Postgres packages with -p 1. +package pgtemplate + +import ( + "context" + "database/sql" + "errors" + "fmt" + "hash/fnv" + "net/url" + "os" + "strings" + "sync" + "sync/atomic" + "syscall" + "testing" + "time" + + "github.com/jackc/pgx/v5" + _ "github.com/jackc/pgx/v5/stdlib" // registers the "pgx" database/sql driver +) + +// Schema is the search_path of every template and clone. +const Schema = "cao_test" + +// Provision initializes the template database through production code. It +// receives a configuration whose search_path is Schema and must release every +// connection before returning. +type Provision func(context.Context, *pgx.ConnConfig) error + +type template struct { + name string + unsupported bool +} + +var ( + mu sync.Mutex + templates = map[string]*template{} + sequence atomic.Uint64 +) + +// Clone returns a configuration for a private database cloned from this +// process's template for endpoint, provisioning the template on first use. It +// returns false when the configured role cannot create databases. The clone is +// dropped when the test ends. +func Clone(t testing.TB, parent context.Context, endpoint string, provision Provision) (*pgx.ConnConfig, bool) { + t.Helper() + admin, err := sql.Open("pgx", endpoint) + if err != nil { + t.Fatal("configure Postgres test connection") + } + t.Cleanup(func() { _ = admin.Close() }) + mu.Lock() + defer mu.Unlock() + current := templates[endpoint] + if current == nil { + current = provisionTemplate(t, parent, admin, endpoint, provision) + templates[endpoint] = current + } + if current.unsupported { + return nil, false + } + ctx, cancel := context.WithTimeout(parent, 30*time.Second) + defer cancel() + name := fmt.Sprintf("%s_%d", processPrefix, sequence.Add(1)) + if _, err := admin.ExecContext(ctx, "CREATE DATABASE "+pgx.Identifier{name}.Sanitize()+" TEMPLATE "+pgx.Identifier{current.name}.Sanitize()); err != nil { + t.Fatalf("clone Postgres test template %s: %v", current.name, err) + } + t.Cleanup(func() { + cleanup, stop := context.WithTimeout(context.WithoutCancel(parent), 30*time.Second) + defer stop() + if _, err := admin.ExecContext(cleanup, "DROP DATABASE IF EXISTS "+pgx.Identifier{name}.Sanitize()+" WITH (FORCE)"); err != nil { + t.Errorf("remove isolated Postgres test database %s: %v", name, err) + } + }) + return DatabaseConfig(t, endpoint, name), true +} + +func provisionTemplate(t testing.TB, parent context.Context, admin *sql.DB, endpoint string, provision Provision) *template { + t.Helper() + ctx, cancel := context.WithTimeout(context.WithoutCancel(parent), 2*time.Minute) + defer cancel() + started := time.Now() + removeStale(t, ctx, admin) + name := fmt.Sprintf("%s_tmpl", processPrefix) + if _, err := admin.ExecContext(ctx, "CREATE DATABASE "+pgx.Identifier{name}.Sanitize()); err != nil { + t.Logf("Postgres test role cannot create databases (%v); provisioning a private schema per test", err) + return &template{unsupported: true} + } + config := DatabaseConfig(t, endpoint, name) + connection, err := pgx.ConnectConfig(ctx, config.Copy()) + if err == nil { + _, err = connection.Exec(ctx, "CREATE SCHEMA "+Schema) + _ = connection.Close(ctx) + } + if err == nil { + err = provision(ctx, config) + } + if err == nil { + _, err = admin.ExecContext(ctx, "ALTER DATABASE "+pgx.Identifier{name}.Sanitize()+" WITH IS_TEMPLATE true ALLOW_CONNECTIONS false") + } + if err != nil { + drop(ctx, admin, name) + t.Fatalf("provision Postgres test template: %v", err) + } + t.Logf("provisioned Postgres test template %s in %s", name, time.Since(started).Round(time.Millisecond)) + return &template{name: name} +} + +// Database names identify the owning host and process so later processes can +// reclaim databases left by interrupted runs: cao_test____…. +var ( + hostTag = hostHash() + processPrefix = fmt.Sprintf("cao_test_%s_%d_%d", hostTag, os.Getpid(), time.Now().Unix()) +) + +func hostHash() string { + host, _ := os.Hostname() + digest := fnv.New32a() + _, _ = digest.Write([]byte(host)) + return fmt.Sprintf("%08x", digest.Sum32()) +} + +// removeStale drops templates and clones whose owning process on this host has +// exited, and any older than a day. Live processes' databases stay intact. +func removeStale(t testing.TB, ctx context.Context, admin *sql.DB) { + t.Helper() + rows, err := admin.QueryContext(ctx, `SELECT datname, datistemplate FROM pg_database WHERE datname LIKE 'cao\_test\_%'`) + if err != nil { + return + } + type candidate struct { + name string + template bool + } + var stale []candidate + cutoff := time.Now().Add(-24 * time.Hour).Unix() + for rows.Next() { + var current candidate + if rows.Scan(¤t.name, ¤t.template) != nil { + break + } + var host string + var pid int + var started int64 + if _, err := fmt.Sscanf(strings.ReplaceAll(current.name, "_", " "), "cao test %s %d %d", &host, &pid, &started); err != nil { + continue + } + if started < cutoff || (host == hostTag && pid != os.Getpid() && !processAlive(pid)) { + stale = append(stale, current) + } + } + _ = rows.Close() + for _, current := range stale { + if current.template { + if _, err := admin.ExecContext(ctx, "ALTER DATABASE "+pgx.Identifier{current.name}.Sanitize()+" WITH IS_TEMPLATE false"); err != nil { + continue + } + } + drop(ctx, admin, current.name) + t.Logf("removed stale Postgres test database %s", current.name) + } +} + +func processAlive(pid int) bool { + process, err := os.FindProcess(pid) + if err != nil { + return false + } + // Treat unknown outcomes, such as unsupported signals, as alive. + err = process.Signal(syscall.Signal(0)) + return !errors.Is(err, os.ErrProcessDone) && !errors.Is(err, syscall.ESRCH) +} + +func drop(ctx context.Context, admin *sql.DB, name string) { + _, _ = admin.ExecContext(ctx, "DROP DATABASE IF EXISTS "+pgx.Identifier{name}.Sanitize()+" WITH (FORCE)") +} + +// DatabaseConfig targets database name with search_path Schema while +// preserving every other endpoint setting, so ConnString identifies the clone +// for subprocesses and adapters. +func DatabaseConfig(t testing.TB, endpoint, name string) *pgx.ConnConfig { + t.Helper() + target := endpoint + " dbname='" + name + "'" + if parsed, err := url.Parse(endpoint); err == nil && (parsed.Scheme == "postgres" || parsed.Scheme == "postgresql") { + parsed.Path = "/" + name + parsed.RawPath = "" + target = parsed.String() + } + config, err := pgx.ParseConfig(target) + if err != nil { + t.Fatal("parse Postgres test configuration") + } + config.RuntimeParams["search_path"] = Schema + return config +} diff --git a/server/internal/testutil/postgres.go b/server/internal/testutil/postgres.go index 022e656519..be30e393ac 100644 --- a/server/internal/testutil/postgres.go +++ b/server/internal/testutil/postgres.go @@ -12,6 +12,7 @@ import ( "github.com/jackc/pgx/v5" "github.com/githubnext/gh-cao/server/internal/postgresx" + "github.com/githubnext/gh-cao/server/internal/testutil/pgtemplate" ) func ConnectionString(t testing.TB, config *pgx.ConnConfig) string { @@ -28,16 +29,20 @@ func ConnectionString(t testing.TB, config *pgx.ConnConfig) string { return endpoint.String() } -// Postgres creates a private schema, including when initialization fails. -// Service-backed tests never replace another test's or the operator's data. +// Postgres creates a private database or schema, including when initialization +// fails. Service-backed tests never replace another test's or the operator's data. func Postgres(t testing.TB, parent context.Context, environment ...string) *postgresx.Store { t.Helper() store, _ := PostgresWithConfig(t, parent, environment...) return store } -// PostgresWithConfig also exposes the isolated schema connection for operational +// PostgresWithConfig also exposes the isolated connection for operational // adapters that need their own tables in the same disposable database. +// +// Each test receives a database cloned from a per-process template provisioned +// through the production store; see pgtemplate. Roles that cannot create +// databases fall back to a private schema in the named database. func PostgresWithConfig(t testing.TB, parent context.Context, environment ...string) (*postgresx.Store, *pgx.ConnConfig) { t.Helper() endpoint := "" @@ -49,13 +54,36 @@ func PostgresWithConfig(t testing.TB, parent context.Context, environment ...str if endpoint == "" { t.Skipf("set one of %v to run Postgres integration tests", environment) } + config, ok := pgtemplate.Clone(t, parent, endpoint, func(ctx context.Context, config *pgx.ConnConfig) error { + store, err := postgresx.NewConfig(ctx, config) + if err != nil { + return err + } + return store.Close() + }) + if !ok { + config = isolatedSchema(t, parent, endpoint) + } ctx, cancel := context.WithTimeout(parent, 30*time.Second) defer cancel() + store, err := postgresx.NewConfig(ctx, config) + if err != nil { + t.Fatal("initialize isolated Postgres test store") + } + t.Cleanup(func() { _ = store.Close() }) + return store, config +} + +// isolatedSchema provisions a private schema in the configured database. +func isolatedSchema(t testing.TB, parent context.Context, endpoint string) *pgx.ConnConfig { + t.Helper() admin, err := sql.Open("pgx", endpoint) if err != nil { t.Fatal("configure Postgres test connection") } t.Cleanup(func() { _ = admin.Close() }) + ctx, cancel := context.WithTimeout(parent, 30*time.Second) + defer cancel() schema := fmt.Sprintf("cao_test_%d", time.Now().UnixNano()) if _, err := admin.ExecContext(ctx, "CREATE SCHEMA "+schema); err != nil { t.Fatal("create isolated Postgres test schema") @@ -104,10 +132,5 @@ func PostgresWithConfig(t testing.TB, parent context.Context, environment ...str t.Fatal("parse Postgres test configuration") } config.RuntimeParams["search_path"] = schema - store, err := postgresx.NewConfig(ctx, config) - if err != nil { - t.Fatal("initialize isolated Postgres test schema") - } - t.Cleanup(func() { _ = store.Close() }) - return store, config + return config }