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()) + }) +})