From a816bf9b84bccaed6470f6931307f6df51eb6600 Mon Sep 17 00:00:00 2001 From: Ettore Di Giacinto Date: Tue, 1 Sep 2026 11:17:48 +0000 Subject: [PATCH] fix(testutil): stop the shared-database change from disarming two regressions Two advisory-lock specs named their database by literal, ALTER DATABASE testdb. Once the test helper started handing every spec its own database on a shared server, that statement landed on the maintenance database and did nothing to the one the spec was holding, so both specs went green having never reproduced the condition they exist for. They regress a model-load advisory-lock wedge that has already shipped to production once, so the previous commit's de-flaking silently disarmed a regression test for a real deployed bug. Both sites now read the name back with current_database() and, more importantly, assert the override actually landed before relying on it. A literal name can go stale again; an assertion that the setting is in force cannot pass while it is not. Removing either production override now fails the matching spec with the real 55P03 and 57014 again. That literal also meant every CREATE DATABASE and every DROP ... WITH (FORCE) ran under the 300ms bound it set on the maintenance database, which is a new load-dependent single-spec flake inside the change that was meant to remove one. The helper's maintenance connections now pin one connection and clear both timeouts on it, so no setting a spec makes can bound them, and a white-box spec imposes the leak deliberately and proves it does not reach them. Also pins the reclaimOne gate deferral the previous commit added without a test, by panicking inside the re-claim's own claim statement, and drops the per-dial empty-token log line to debug now that the boot warning says it once. Assisted-by: Claude Opus 5 [claude-code] Signed-off-by: Ettore Di Giacinto --- core/http/endpoints/cluster/connect.go | 7 ++- .../advisorylock/advisorylock_test.go | 61 ++++++++++++++---- core/services/cluster/tunnel_test.go | 42 +++++++++++++ core/services/testutil/testdb.go | 58 +++++++++++++++-- .../services/testutil/testdb_internal_test.go | 63 +++++++++++++++++++ 5 files changed, 214 insertions(+), 17 deletions(-) create mode 100644 core/services/testutil/testdb_internal_test.go diff --git a/core/http/endpoints/cluster/connect.go b/core/http/endpoints/cluster/connect.go index e76450692..ca94ead48 100644 --- a/core/http/endpoints/cluster/connect.go +++ b/core/http/endpoints/cluster/connect.go @@ -111,7 +111,12 @@ func ConnectHandler(registry *nodes.NodeRegistry, tunnels *clustersvc.TunnelRegi // secret. One log line for both leaves an operator reading "wrong token" // while every worker fails identically. if node.TokenHash == "" { - xlog.Warn("Refusing a worker tunnel: this node has no stored token, which means it registered with no registration token configured", + // Debug, not Warn. Every worker in such a deployment fails this way + // on every reconnect, so warning per dial buries the log; the fact + // is stated once, at boot, where core/http/app.go warns that no + // registration token is configured. What this line adds is which + // node, for an operator who has already read that warning. + xlog.Debug("refusing a worker tunnel: this node has no stored token, so it registered with no registration token configured", "node", nodeID, "knob", "LOCALAI_REGISTRATION_TOKEN") return echo.NewHTTPError(http.StatusUnauthorized, "unauthorized") } diff --git a/core/services/advisorylock/advisorylock_test.go b/core/services/advisorylock/advisorylock_test.go index f1bd3e75e..536666f6c 100644 --- a/core/services/advisorylock/advisorylock_test.go +++ b/core/services/advisorylock/advisorylock_test.go @@ -2,7 +2,9 @@ package advisorylock import ( "context" + "fmt" "runtime" + "strings" "sync" "sync/atomic" "time" @@ -14,6 +16,51 @@ import ( "gorm.io/gorm" ) +// alterThisDatabase applies a server-side setting to the database this handle is +// actually connected to, and proves it landed. +// +// The name is read back from the connection rather than written as a literal. +// The test helper hands each spec its own database on a shared server, so a +// hard-coded name ALTERs a database this handle never touches: the statement +// succeeds, the override does nothing, and the two specs below go green having +// exercised none of the condition they exist for. They regress a model-load +// advisory-lock wedge that has already shipped to production once, so a green +// spec that proves nothing is the worst outcome available here. +// +// The read-back is the guard. Idle connections are dropped first so the next one +// is opened fresh and inherits the new database-level default; SHOW then reports +// what a waiter's own connection would inherit. If that ever stops matching, the +// spec fails here rather than passing for the wrong reason. +func alterThisDatabase(db *gorm.DB, setting, value string) { + GinkgoHelper() + + var name string + Expect(db.Raw("SELECT current_database()").Scan(&name).Error).ToNot(HaveOccurred()) + Expect(name).ToNot(BeEmpty()) + + Expect(db.Exec(fmt.Sprintf("ALTER DATABASE %q SET %s = %s", name, setting, quoteLiteral(value))).Error). + ToNot(HaveOccurred()) + + sqlDB, err := db.DB() + Expect(err).ToNot(HaveOccurred()) + // database/sql retains no idle connections at 0, closing the ones it is + // already holding, so every connection after this point is opened fresh and + // inherits the new database-level default. + sqlDB.SetMaxIdleConns(0) + + var applied string + Expect(db.Raw("SHOW " + setting).Scan(&applied).Error).ToNot(HaveOccurred()) + Expect(applied).To(Equal(value), + "the %s override did not reach the database this spec is holding (%s), so the spec below would pass without ever reproducing the condition it regresses", + setting, name) +} + +// quoteLiteral wraps a settings value as a SQL string literal. The values here +// are spec constants, so this only has to be correct, not hostile-input-proof. +func quoteLiteral(v string) string { + return "'" + strings.ReplaceAll(v, "'", "''") + "'" +} + var _ = Describe("AdvisoryLock", func() { Context("PostgreSQL advisory locks", func() { var db *gorm.DB @@ -166,12 +213,7 @@ var _ = Describe("AdvisoryLock", func() { // blocked on pg_advisory_lock() is aborted by the server after this // window and surfaces SQLSTATE 55P03 ("canceling statement due to // lock timeout") to the caller instead of waiting for its turn. - Expect(db.Exec("ALTER DATABASE testdb SET lock_timeout = '300ms'").Error).ToNot(HaveOccurred()) - sqlDB, err := db.DB() - Expect(err).ToNot(HaveOccurred()) - // Drop pooled connections so subsequent ones reconnect and inherit - // the new database-level lock_timeout default. - sqlDB.SetMaxIdleConns(0) + alterThisDatabase(db, "lock_timeout", "300ms") holding := make(chan struct{}) released := make(chan struct{}) @@ -214,12 +256,7 @@ var _ = Describe("AdvisoryLock", func() { // statement_timeout=60s; a cold model load holds the lock far longer, // so every concurrent caller died with SQLSTATE 57014 ("canceling // statement due to statement timeout") rather than waiting its turn. - Expect(db.Exec("ALTER DATABASE testdb SET statement_timeout = '300ms'").Error).ToNot(HaveOccurred()) - sqlDB, err := db.DB() - Expect(err).ToNot(HaveOccurred()) - // Drop pooled connections so subsequent ones reconnect and inherit - // the new database-level statement_timeout default. - sqlDB.SetMaxIdleConns(0) + alterThisDatabase(db, "statement_timeout", "300ms") holding := make(chan struct{}) released := make(chan struct{}) diff --git a/core/services/cluster/tunnel_test.go b/core/services/cluster/tunnel_test.go index acc81d818..7d0136b47 100644 --- a/core/services/cluster/tunnel_test.go +++ b/core/services/cluster/tunnel_test.go @@ -715,6 +715,48 @@ var _ = Describe("Re-claiming tunnels after this replica's rows were reaped", fu } Consistently(stored, 2*cluster.InstanceHeartbeat, time.Second).Should(Equal(epoch)) }) + + It("frees the node's gate when a re-claim panics", func() { + // The same property Attach's gate specs pin, on the other function that + // takes the gate. Reclaim runs from the heartbeat loop, so a wedged gate + // here is worse than one wedged by a dial: nothing retries it, and the + // worker can never re-attach to this replica because its Attach blocks + // in enterClaim until its own context expires. + // + // The panic is thrown from inside the re-claim's own Claim statement, + // on that statement's goroutine, using the same gorm Trace hook the + // interleaving specs above use. That is the window a real panic under + // Claim would land in: the row is written and the gate is held. + hook := newClaimHook(isClaimOf("w1")) + hooked := cluster.NewTunnelRegistry( + cluster.NewRegistry(db.Session(&gorm.Session{Logger: hook})), "me") + + frontend, _ := workerTunnel() + _, err := hooked.Attach(ctx, "w1", frontend) + Expect(err).ToNot(HaveOccurred()) + + // Installed after the attach so the hook fires on the RE-claim, not on + // the claim that set the tunnel up. + hook.setAction(func(string) { panic("claim exploded") }) + + panicked := func() (p bool) { + defer func() { p = recover() != nil }() + _, _ = hooked.Reclaim(ctx) + return + }() + Expect(panicked).To(BeTrue(), + "the hook did not fire inside the re-claim, so this spec is no longer testing what it claims") + + // A wedged gate is indistinguishable from a slow one except by waiting. + // The hook has already fired once and will not fire again, so this + // Attach either completes or never reaches Claim at all. + bounded, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + next, _ := workerTunnel() + _, err = hooked.Attach(bounded, "w1", next) + Expect(err).ToNot(HaveOccurred(), + "the panicking re-claim left this node's gate closed, so the worker can never attach to this replica again") + }) }) var _ = Describe("The worker tunnel registry's claim gate", func() { diff --git a/core/services/testutil/testdb.go b/core/services/testutil/testdb.go index 755589cd4..7e2c061dd 100644 --- a/core/services/testutil/testdb.go +++ b/core/services/testutil/testdb.go @@ -146,7 +146,7 @@ func SetupTestDB() *gorm.DB { // teardown. closePool(db) - drop, err := gorm.Open(postgres.Open(dsn), &gorm.Config{Logger: logger.Discard}) + drop, err := openTolerantPool(dsn) if err != nil { // Reported, never asserted. A cleanup that fails the spec turns one // database hiccup into a failure that buries whatever the spec was @@ -165,16 +165,66 @@ func SetupTestDB() *gorm.DB { return db } -// openPool connects to dsn with logging off. Used for the short-lived -// maintenance connections only; the database a spec is handed keeps gorm's -// silent logger so a caller can still swap it. +// openPool connects to dsn with logging off and with every server-side timeout +// disabled on the session. Used for the short-lived maintenance connections +// only; the database a spec is handed keeps gorm's silent logger and the +// server's defaults, because setting timeouts on it is a thing specs do on +// purpose. +// +// The timeouts are cleared because CREATE DATABASE and DROP DATABASE must not +// be bounded by anything a spec configured. A spec that sets a short +// statement_timeout on ITS database cannot reach this one, but a spec that +// names the maintenance database by mistake can, and that is not hypothetical: +// two advisory-lock specs did exactly that until this round. The consequence +// there is a load-dependent failure in another spec's setup or a silently +// swallowed DROP, which is the same invisible single-spec flake this helper was +// rewritten to remove. +// +// MaxOpenConns(1) is what makes the SET reach the statement that follows it: a +// session setting lives on one connection, and with a single connection in the +// pool there is no other one for CREATE or DROP to land on. func openPool(dsn string) *gorm.DB { GinkgoHelper() db, err := gorm.Open(postgres.Open(dsn), &gorm.Config{Logger: logger.Discard}) Expect(err).ToNot(HaveOccurred()) + + sqlDB, err := db.DB() + Expect(err).ToNot(HaveOccurred()) + sqlDB.SetMaxOpenConns(1) + + Expect(db.Exec("SET statement_timeout = 0").Error).To(Succeed()) + Expect(db.Exec("SET lock_timeout = 0").Error).To(Succeed()) return db } +// openTolerantPool is openPool for the cleanup path, which must report a +// failure rather than assert one: an assertion here would fail a spec that had +// already passed, and bury whatever the next real failure was. +func openTolerantPool(dsn string) (*gorm.DB, error) { + db, err := gorm.Open(postgres.Open(dsn), &gorm.Config{Logger: logger.Discard}) + if err != nil { + return nil, err + } + sqlDB, err := db.DB() + if err != nil { + closePool(db) + return nil, err + } + sqlDB.SetMaxOpenConns(1) + // The DROP below is the statement most likely to be slow, since it waits on + // FORCE terminating other sessions, so it is the one a leaked timeout would + // abort. Measured at up to 169ms under load, against a 300ms bound. + if err := db.Exec("SET statement_timeout = 0").Error; err != nil { + closePool(db) + return nil, err + } + if err := db.Exec("SET lock_timeout = 0").Error; err != nil { + closePool(db) + return nil, err + } + return db, nil +} + func closePool(db *gorm.DB) { if db == nil { return diff --git a/core/services/testutil/testdb_internal_test.go b/core/services/testutil/testdb_internal_test.go new file mode 100644 index 000000000..457d6f7d5 --- /dev/null +++ b/core/services/testutil/testdb_internal_test.go @@ -0,0 +1,63 @@ +package testutil + +import ( + "fmt" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +// These are white-box on purpose: the property is about the connection this +// package makes for itself, which no caller can reach. +var _ = Describe("the maintenance connection", func() { + It("cannot be bounded by a timeout set on the maintenance database", func() { + // The leak this pins is not hypothetical. Two advisory-lock specs named + // a database by literal, and once the helper started handing out + // per-spec databases those ALTERs landed on the maintenance database + // instead, so every CREATE DATABASE and every DROP ... WITH (FORCE) ran + // under a 300ms bound. A CREATE that trips it fails another spec's + // setup; a DROP that trips it is swallowed and leaks a database. Both + // are load-dependent single-spec failures, which is the exact shape + // this helper was rewritten to remove. + dsn := sharedPostgres() + + var maintenance string + func() { + probe := openPool(dsn) + defer closePool(probe) + Expect(probe.Raw("SELECT current_database()").Scan(&maintenance).Error).To(Succeed()) + }() + Expect(maintenance).ToNot(BeEmpty()) + + // Impose the leak, then assert a fresh maintenance connection is + // unaffected. Reset first so a failure below cannot leave the bound in + // place for the rest of the suite. + DeferCleanup(func() { + reset := openPool(dsn) + defer closePool(reset) + Expect(reset.Exec(fmt.Sprintf("ALTER DATABASE %q RESET statement_timeout", maintenance)).Error).To(Succeed()) + Expect(reset.Exec(fmt.Sprintf("ALTER DATABASE %q RESET lock_timeout", maintenance)).Error).To(Succeed()) + }) + func() { + impose := openPool(dsn) + defer closePool(impose) + Expect(impose.Exec(fmt.Sprintf("ALTER DATABASE %q SET statement_timeout = '1ms'", maintenance)).Error).To(Succeed()) + Expect(impose.Exec(fmt.Sprintf("ALTER DATABASE %q SET lock_timeout = '1ms'", maintenance)).Error).To(Succeed()) + }() + + fresh := openPool(dsn) + defer closePool(fresh) + var statementTimeout, lockTimeout string + Expect(fresh.Raw("SHOW statement_timeout").Scan(&statementTimeout).Error).To(Succeed()) + Expect(fresh.Raw("SHOW lock_timeout").Scan(&lockTimeout).Error).To(Succeed()) + Expect(statementTimeout).To(Equal("0"), + "a statement_timeout on the maintenance database reached the helper's own connection, so CREATE and DROP DATABASE are bounded by whatever a spec configured") + Expect(lockTimeout).To(Equal("0"), + "a lock_timeout on the maintenance database reached the helper's own connection") + + // And the thing the timeouts would actually abort still works while the + // bound is in force. A 1ms statement_timeout is far below the 14-26ms a + // CREATE DATABASE takes here, so this could not pass by being fast. + Expect(SetupTestDB()).ToNot(BeNil()) + }) +})