From 49e77447fbba6b5fc17286505f5608cad32dcd72 Mon Sep 17 00:00:00 2001 From: Ettore Di Giacinto Date: Wed, 2 Sep 2026 12:03:55 +0000 Subject: [PATCH] feat(cluster): record a departure instead of erasing the connection Releasing a worker tunnel deleted its node_connections row, so "this worker's link dropped a moment ago" and "this worker has never connected here" were one observation: no row. Nothing above could tell a worker re-homing between replicas from a worker that is gone, and any grace period built on top would have had nothing to measure from. The row now survives a departure. Release clears owner_instance_id and stamps disconnected_at on the database clock; the membership sweep and Deregister do the same for every connection a dead or departing replica held; Claim clears the stamp in the same upsert that writes the owner, so a reconnect is never observed half-applied. PurgeDepartedBefore deletes a departure once it is older than DepartedRetention, and the membership tick owns that schedule. Owner and OwnerRow report a departed row as ErrNoConnection, through the one predicate connectionIsHeld, the way instanceIsLive is the one predicate for replica liveness. This change records the departure and does not interpret it: how long ago it happened is nobody's answer yet. The sweep only clears rows that are still held. An empty owner is in no instance's id, so without that filter every heartbeat would restamp every departed row and no departure could ever age out. Assisted-by: Claude Opus 5 [claude-code] Signed-off-by: Ettore Di Giacinto --- core/services/cluster/membership.go | 94 ++++++--- core/services/cluster/membership_test.go | 67 ++++++- core/services/cluster/ownership.go | 152 +++++++++++++-- core/services/cluster/ownership_test.go | 225 ++++++++++++++++++++-- docs/content/features/distributed-mode.md | 2 +- 5 files changed, 481 insertions(+), 59 deletions(-) diff --git a/core/services/cluster/membership.go b/core/services/cluster/membership.go index 6a8650927..84544e713 100644 --- a/core/services/cluster/membership.go +++ b/core/services/cluster/membership.go @@ -29,6 +29,18 @@ const ( // deregisterTimeout bounds the deregistration Stop performs. Shutdown is // not the place to wait on a database. deregisterTimeout = 5 * time.Second + + // DepartedRetention is how long a connection row outlives the tunnel it + // recorded before the sweep deletes it. + // + // A multiple of the liveness window rather than the window itself. The row + // survives so that how long ago a tunnel went can be answered at all, and a + // retention as short as the window a reader compares that age against would + // let the purge delete the row out from under the reader, turning a worker + // that is re-dialling back into a worker that was never here. Ten windows + // is far enough clear of any such comparison, and short enough that a + // worker retired for good does not sit in the table for a day. + DepartedRetention = 10 * InstanceLiveness ) // Membership publishes this replica's address and keeps the instances table @@ -207,6 +219,19 @@ func (m *Membership) tick(ctx context.Context) { if instances > 0 || connections > 0 { xlog.Info("Reaped cluster state left by dead replicas", "instances", instances, "connections", connections) } + + // The retention has an owner, and it is this sweep. A departure that + // nothing ever deletes is a row per worker that ever dialled this + // deployment, and it is the same loop that decides a replica is gone, so + // there is one schedule rather than two. + purged, err := m.reg.PurgeDepartedBefore(ctx, DepartedRetention) + if err != nil { + xlog.Warn("Purging departed worker connections failed", "error", err) + return + } + if purged > 0 { + xlog.Info("Purged worker connections whose departure aged out", "connections", purged, "retention", DepartedRetention) + } } // reclaimTunnels re-writes a claim for every worker tunnel this replica still @@ -233,13 +258,18 @@ func (m *Membership) reclaimTunnels(ctx context.Context) { } } -// Deregister removes one replica and the connections it owned. +// Deregister removes one replica and records a departure for every connection +// it owned. // -// It deletes both, in one transaction, for the same reason ReapStale does: a -// replica that is gone owns nothing, and leaving its connection rows behind -// would point every reader at an owner that no longer exists. This is the +// Both in one transaction, for the same reason ReapStale does it in one: a +// replica that is gone owns nothing, and leaving its connection rows pointing +// at it would point every reader at an owner that no longer exists. This is the // announced form of what the sweeper does by inference, and the two must not -// disagree about what "gone" removes. +// disagree about what "gone" leaves behind. +// +// The connection rows are cleared, not deleted: a worker whose frontend shut +// down is about to re-dial the load balancer, and erasing its row would make +// the seconds in between look like a worker that had never connected. // // Instances first, then connections, which is deliberate and is the same order // ReapStale takes. The two paths run concurrently in the ordinary case, a @@ -255,8 +285,11 @@ func (r *Registry) Deregister(ctx context.Context, id string) error { if err := tx.Where("id = ?", id).Delete(&Instance{}).Error; err != nil { return fmt.Errorf("deleting instance %q: %w", id, err) } - if err := tx.Where("owner_instance_id = ?", id).Delete(&NodeConnection{}).Error; err != nil { - return fmt.Errorf("deleting connections owned by %q: %w", id, err) + // No held-ness filter: only rows this replica owns match, and a + // departed row carries no owner, so it cannot be restamped here. + if err := tx.Model(&NodeConnection{}).Where("owner_instance_id = ?", id). + Updates(departure()).Error; err != nil { + return fmt.Errorf("recording departures for connections owned by %q: %w", id, err) } return nil }); err != nil { @@ -266,33 +299,39 @@ func (r *Registry) Deregister(ctx context.Context, id string) error { } // ReapStale deletes the replicas that have not heartbeated within the liveness -// window, and the connection rows whose owner is no longer among the survivors. +// window, and records a departure for every connection row whose owner is no +// longer among the survivors. // -// The two deletes are one sweeper on purpose. A connection row is only ever -// orphaned by its owner dying, so the moment that is decided is the moment to -// clean up after it; a second sweeper with its own schedule would either lag -// this one or race it, and would need its own answer to "is that replica -// alive", which is the one fact this table already owns. +// The two are one sweeper on purpose. A connection row is only ever orphaned by +// its owner dying, so the moment that is decided is the moment to clean up +// after it; a second sweeper with its own schedule would either lag this one or +// race it, and would need its own answer to "is that replica alive", which is +// the one fact this table already owns. +// +// The connection rows are cleared and not deleted, and the returned count is of +// rows this sweep cleared. A worker whose owning replica died has not gone +// anywhere: it is re-dialling the load balancer, and deleting its row would +// erase the departure that says how long ago that started. // // self is never reaped. This process may fail to heartbeat for longer than the // window (a long stall, a database blip) and still be serving: deleting its own -// row would then delete the connections of workers that are, at that moment, +// row would then mark as departed the workers that are, at that moment, // connected to it. // // That protection is one-sided. A replica that stalls long enough is reaped BY -// ANOTHER replica, taking its connection rows with it, and Register rebuilds -// the instance row and nothing else. What restores the rest is the re-claim in -// tick, which writes a fresh claim for every tunnel the tunnel registry still -// holds; until it runs, this replica holds sockets the table records nobody -// holding. +// ANOTHER replica, which records a departure for every connection it held, and +// Register rebuilds the instance row and nothing else. What restores the rest +// is the re-claim in tick, which writes a fresh claim for every tunnel the +// tunnel registry still holds; until it runs, this replica holds sockets the +// table records nobody holding. // // PostgreSQL only, like Live: distributed mode requires it, and the interval // arithmetic is measured on the database's clock because liveness is compared // across replicas. // -// Instances are deleted before connections, and Deregister takes the same order +// Instances are written before connections, and Deregister takes the same order // on purpose, so the two paths cannot deadlock against each other. Here the -// order is also forced: the connection delete asks which instance rows survived, +// order is also forced: the connection sweep asks which instance rows survived, // so it has to run second. func (r *Registry) ReapStale(ctx context.Context, self string, within time.Duration) (instances int64, connections int64, err error) { err = r.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { @@ -309,10 +348,17 @@ func (r *Registry) ReapStale(ctx context.Context, self string, within time.Durat // Whatever survived the delete above is the live set, so this needs no // second liveness rule and cannot disagree with the first one. - res = tx.Where("owner_instance_id NOT IN (SELECT id FROM instances)"). - Delete(&NodeConnection{}) + // + // Held rows only. An empty owner is in no instance's id, so a departed + // row matches the set difference too, and re-clearing it every sweep + // would push its departure forward five seconds at a time: it would + // never age out of any window measured from it, and every sweep would + // report clearing a connection that had already gone. + res = tx.Model(&NodeConnection{}). + Where("owner_instance_id NOT IN (SELECT id FROM instances) AND " + connectionIsHeld). + Updates(departure()) if res.Error != nil { - return fmt.Errorf("deleting orphaned node connections: %w", res.Error) + return fmt.Errorf("recording departures for orphaned node connections: %w", res.Error) } connections = res.RowsAffected return nil diff --git a/core/services/cluster/membership_test.go b/core/services/cluster/membership_test.go index c331eafe2..22bd37f00 100644 --- a/core/services/cluster/membership_test.go +++ b/core/services/cluster/membership_test.go @@ -51,6 +51,69 @@ var _ = Describe("Reaping dead replicas", func() { Expect(err).To(MatchError(cluster.ErrNoConnection)) }) + It("records a departure rather than erasing the connection when a stale replica is swept", func() { + // The row is what tells "this worker's owner died a moment ago" from + // "this worker has never connected". Deleting it on the sweep erased + // exactly the departure a reconnect grace has to be measured from. + Expect(reg.Register(ctx, "live", "10.0.0.1:8080", "v1")).To(Succeed()) + Expect(reg.Register(ctx, "dead", "10.0.0.2:8080", "v1")).To(Succeed()) + _, err := reg.Claim(ctx, "w1", "dead") + Expect(err).ToNot(HaveOccurred()) + age("dead", time.Hour) + + _, _, err = reg.ReapStale(ctx, "live", time.Minute) + Expect(err).ToNot(HaveOccurred()) + + var row cluster.NodeConnection + Expect(db.WithContext(ctx).Where("node_id = ?", "w1").First(&row).Error).To(Succeed(), + "the sweep deleted the row, so the worker its owner left behind looks like one that never dialled") + Expect(row.OwnerInstanceID).To(BeEmpty()) + Expect(row.DisconnectedAt).ToNot(BeNil()) + }) + + It("does not restamp a departure it has already recorded", func() { + // The sweep runs every heartbeat. Re-clearing a row it already cleared + // would push its departure forward on every pass, so the departure + // would never age out of any window and the row would be reported as + // swept for as long as it existed. + Expect(reg.Register(ctx, "live", "10.0.0.1:8080", "v1")).To(Succeed()) + Expect(reg.Register(ctx, "dead", "10.0.0.2:8080", "v1")).To(Succeed()) + _, err := reg.Claim(ctx, "w1", "dead") + Expect(err).ToNot(HaveOccurred()) + age("dead", time.Hour) + + _, connections, err := reg.ReapStale(ctx, "live", time.Minute) + Expect(err).ToNot(HaveOccurred()) + Expect(connections).To(Equal(int64(1))) + var first cluster.NodeConnection + Expect(db.WithContext(ctx).Where("node_id = ?", "w1").First(&first).Error).To(Succeed()) + + _, connections, err = reg.ReapStale(ctx, "live", time.Minute) + Expect(err).ToNot(HaveOccurred()) + Expect(connections).To(BeZero(), "the sweeper reported clearing a connection that was already departed") + + var second cluster.NodeConnection + Expect(db.WithContext(ctx).Where("node_id = ?", "w1").First(&second).Error).To(Succeed()) + Expect(second.DisconnectedAt).ToNot(BeNil()) + Expect(*second.DisconnectedAt).To(Equal(*first.DisconnectedAt), + "the second sweep moved the departure forward, so it can never age past a grace window") + }) + + It("records a departure rather than erasing the connection when a replica deregisters", func() { + // The announced form of the same thing the sweeper does by inference, + // and the two must not disagree about what "gone" leaves behind. + Expect(reg.Register(ctx, "leaving", "10.0.0.2:8080", "v1")).To(Succeed()) + _, err := reg.Claim(ctx, "w1", "leaving") + Expect(err).ToNot(HaveOccurred()) + + Expect(reg.Deregister(ctx, "leaving")).To(Succeed()) + + var row cluster.NodeConnection + Expect(db.WithContext(ctx).Where("node_id = ?", "w1").First(&row).Error).To(Succeed()) + Expect(row.OwnerInstanceID).To(BeEmpty()) + Expect(row.DisconnectedAt).ToNot(BeNil()) + }) + It("leaves the connections of a live replica alone", func() { Expect(reg.Register(ctx, "live", "10.0.0.1:8080", "v1")).To(Succeed()) Expect(reg.Register(ctx, "other", "10.0.0.2:8080", "v1")).To(Succeed()) @@ -150,8 +213,8 @@ var _ = Describe("Reaping dead replicas", func() { ReapStale(ctx, "sweeper", time.Minute) Expect(err).ToNot(HaveOccurred()) - Expect(deregRec.deleteOrder()).To(Equal([]string{"instances", "node_connections"})) - Expect(reapRec.deleteOrder()).To(Equal(deregRec.deleteOrder()), + Expect(deregRec.writeOrder()).To(Equal([]string{"instances", "node_connections"})) + Expect(reapRec.writeOrder()).To(Equal(deregRec.writeOrder()), "the sweeper and deregistration must lock the same two tables in the same order") }) diff --git a/core/services/cluster/ownership.go b/core/services/cluster/ownership.go index 8a80adede..d4a544fcc 100644 --- a/core/services/cluster/ownership.go +++ b/core/services/cluster/ownership.go @@ -28,11 +28,12 @@ func isPostgres(db *gorm.DB) bool { } // epochSequence is the PostgreSQL sequence every claim draws its epoch from. -// A sequence rather than a per-row counter because a released row is deleted: -// with `epoch = epoch + 1` the numbering restarts at 1 for the next claim, so a +// A sequence rather than a per-row counter because a connection row does not +// outlive the deployment: a departure is purged once it is old enough, and with +// `epoch = epoch + 1` the numbering would restart at 1 for the next claim, so a // replica that claimed, lost the worker, and claimed again could be handed an // epoch it already held, and a delayed cleanup from the first claim would then -// match, and delete, the live one. +// match, and clear, the live one. const epochSequence = "node_connection_epochs" // NodeConnection records which frontend replica currently holds a worker's @@ -50,15 +51,46 @@ const epochSequence = "node_connection_epochs" // one is answered by the epoch; a second liveness clock for the same fact would // only drift from the first. type NodeConnection struct { - NodeID string `gorm:"primaryKey;size:36" json:"node_id"` + NodeID string `gorm:"primaryKey;size:36" json:"node_id"` + // Empty when nobody holds this tunnel. A departed row keeps the empty + // string rather than NULL so there is one spelling of "held by nobody"; + // connectionIsHeld is the only place that spelling is written down. OwnerInstanceID string `gorm:"size:36;index;not null" json:"owner_instance_id"` Epoch int64 `gorm:"not null" json:"epoch"` // No column DEFAULT: now() is PostgreSQL syntax and would reach the DDL, // which breaks AutoMigrate on the SQLite single-binary path. Claim writes // the database clock as an expression instead, the way Register does. ConnectedAt time.Time `gorm:"not null" json:"connected_at"` + // DisconnectedAt is when the tunnel LEFT, and it exists because "gone for + // thirty seconds" and "never here" have to be different answers. + // + // A released row used to be deleted, so a worker re-homing between replicas + // was indistinguishable from one that had never dialled, and any grace + // period built on top would have had nothing to measure from. It is NOT a + // liveness clock for the owner: whether the OWNING replica is alive is + // still Instance.LastSeen and nothing else. It ticks once, on departure. + // + // Null while the row is held, and stamped on the database clock by whoever + // records the departure, so a row with an owner never also carries one. + DisconnectedAt *time.Time `gorm:"index" json:"disconnected_at,omitempty"` } +// connectionIsHeld is the one predicate that separates a row recording a tunnel +// somebody holds from a row recording a departure. Everything that asks that +// question asks it here, or asks for its negation: Owner and OwnerRow refuse a +// row it rejects, ReapStale clears only rows it accepts, and +// PurgeDepartedBefore deletes only rows it rejects. Two spellings of one fact +// drift, and the drift here would show up as a departed worker resolving to a +// replica that holds nothing. +// +// Deregister is the one write that does not use it, because it matches an owner +// id, which is narrower: a departed row carries no owner and so cannot match. +// +// The column is table-qualified because Owner reads it across a join, where an +// unqualified name is ambiguous; qualifying it costs the single-table readers +// nothing. +const connectionIsHeld = `node_connections.owner_instance_id <> ''` + // Migrate creates every table and sequence this package owns. It is the one // call a caller has to remember: gorm's AutoMigrate models tables and columns // but has no notion of a sequence, and the connection fence draws its epochs @@ -145,6 +177,10 @@ func (r *Registry) Claim(ctx context.Context, nodeID, ownerID string) (int64, er "owner_instance_id": ownerID, "epoch": nextEpoch, "connected_at": gorm.Expr("now()"), + // In the SAME statement as the owner, so a reconnect is never + // observed half-applied: a row that names a holder while still + // carrying a departure would answer "held" and "gone" at once. + "disconnected_at": nil, }), }, clause.Returning{Columns: []clause.Column{{Name: "epoch"}}}, @@ -179,10 +215,15 @@ func (r *Registry) Claim(ctx context.Context, nodeID, ownerID string) (int64, er // liveness. Its callers today are this package's specs, including the one that // holds the two reads apart, and the e2e cluster spec that watches ownership // move between replicas. No production caller reads it, and the sweeper is not -// one: ReapStale deletes orphans with a set difference in SQL. +// one: ReapStale finds orphans with a set difference in SQL. +// +// A row that records a departure is ErrNoConnection here too. The row survives +// so that something can decide how old the departure is; what it says about who +// holds the tunnel is nobody, and this read reports exactly that. func (r *Registry) OwnerRow(ctx context.Context, nodeID string) (string, int64, error) { var conn NodeConnection - err := r.db.WithContext(ctx).Where("node_id = ?", nodeID).First(&conn).Error + err := r.db.WithContext(ctx). + Where("node_id = ? AND "+connectionIsHeld, nodeID).First(&conn).Error if errors.Is(err, gorm.ErrRecordNotFound) { return "", 0, fmt.Errorf("looking up owner of node %q: %w", nodeID, ErrNoConnection) } @@ -202,10 +243,15 @@ func (r *Registry) OwnerRow(ctx context.Context, nodeID string) (string, int64, // is gone, and a relay built on it would dial a corpse and report the worker as // unreachable rather than as absent. // -// A missing row and a dead owner are one answer on purpose. Both mean "no -// replica here holds this worker's tunnel", which is what a caller decides on; -// they differ only in which sweep has already run, and that is the sweeper's -// business rather than the caller's. +// A missing row, a departed row and a dead owner are one answer on purpose. All +// three mean "no live replica holds this worker's tunnel", which is what a +// caller decides on; they differ only in which sweep has already run, and that +// is the sweeper's business rather than the caller's. +// +// That answer is emphatically not "this worker is gone". How long ago the +// departure was recorded is what separates a worker re-homing between replicas +// from one that has left, and this read does not report it: a caller that has +// to tell those apart reads the departure, not this. // // One statement, joined, not a row read followed by an instance lookup: between // two statements the owner can die, and the caller would act on an owner the @@ -235,7 +281,11 @@ func (r *Registry) Owner(ctx context.Context, nodeID string) (string, int64, err // is here to filter and the row scanned back must stay this table's. Select("node_connections.*"). Joins("JOIN instances ON instances.id = node_connections.owner_instance_id AND "+instanceIsLive, InstanceLiveness.Seconds()). - Where("node_connections.node_id = ?", nodeID). + // The departure filter is this query's own, not the join's. An empty + // owner matches no instance today only because no replica registers + // under an empty id, which is an accident of who registers rather than + // a property of ownership. + Where("node_connections.node_id = ? AND "+connectionIsHeld, nodeID). Take(&conn).Error if errors.Is(err, gorm.ErrRecordNotFound) { return "", 0, fmt.Errorf("looking up live owner of node %q: %w", nodeID, ErrNoConnection) @@ -246,19 +296,52 @@ func (r *Registry) Owner(ctx context.Context, nodeID string) (string, int64, err return conn.OwnerInstanceID, conn.Epoch, nil } -// Release drops the claim identified by ownerID and epoch. Both are in the -// WHERE so a replica that has only just noticed its dead socket cannot delete -// the claim a later reconnect established elsewhere: the row it is trying to -// clean up no longer exists, and deleting the live one would strand a worker -// that is in fact connected. A claim that is no longer the live one is reported -// as ErrNoConnection rather than silently ignored, because the caller learning -// it has been fenced out is the point. +// departure is the single assignment that turns a held row into a departed one. +// Release writes it for one claim and the membership sweep writes it for every +// claim a dead replica held, and the two must not disagree about what a +// departure looks like: a row cleared without a stamp, or stamped without being +// cleared, is a state every reader here has no answer for. +// +// The timestamp is the database's, never this process's, for the same reason +// instance liveness is: it is compared across replicas, so it has to be +// measured on the one clock they share. +func departure() map[string]any { + return map[string]any{ + "owner_instance_id": "", + "disconnected_at": gorm.Expr("now()"), + } +} + +// Release drops the claim identified by ownerID and epoch, and records the +// departure: the row survives with no owner and a disconnected_at stamp. +// +// An UPDATE and not a DELETE, because the row is the only place a departure can +// be recorded. Deleting it made a worker re-homing between replicas look like +// one that had never connected, so nothing above could tell a two-second blip +// from a worker that is gone. +// +// Both ownerID and epoch are in the WHERE so a replica that has only just +// noticed its dead socket cannot touch the claim a later reconnect established +// elsewhere: it must not clear a live claim, and it must not stamp a departure +// onto one. A claim that is no longer the live one is reported as +// ErrNoConnection rather than silently ignored, because the caller learning it +// has been fenced out is the point. func (r *Registry) Release(ctx context.Context, nodeID, ownerID string, epoch int64) error { + // Refused rather than attempted, for the reason Claim refuses: the + // departure is stamped with now(), which on the single-binary SQLite path + // fails as a missing function and reads as a missing migration. It is + // deliberately not ErrNoConnection: a deployment with no cluster holds no + // claims, and reporting a fenced-out release would tell the caller it lost + // one. + if !isPostgres(r.db) { + return fmt.Errorf("releasing connection for node %q held by %q: connection ownership requires PostgreSQL, this deployment runs on %q", nodeID, ownerID, r.db.Dialector.Name()) + } // gorm reports no error when a Where matches nothing, so the miss has to be // read off RowsAffected. res := r.db.WithContext(ctx). + Model(&NodeConnection{}). Where("node_id = ? AND owner_instance_id = ? AND epoch = ?", nodeID, ownerID, epoch). - Delete(&NodeConnection{}) + Updates(departure()) if res.Error != nil { return fmt.Errorf("releasing connection for node %q held by %q at epoch %d: %w", nodeID, ownerID, epoch, res.Error) } @@ -267,3 +350,34 @@ func (r *Registry) Release(ctx context.Context, nodeID, ownerID string, epoch in } return nil } + +// PurgeDepartedBefore deletes the connection rows whose departure is older than +// olderThan, and returns how many went. +// +// Departures are kept so that absence can be decided, not kept forever. The +// retention has to outlast every window measured from a departure by enough +// that a purge can never turn a worker inside its reconnect grace into a worker +// that was never here; the caller passes a multiple of that grace, never the +// grace itself. +// +// A held row is never touched, whatever timestamp it carries: deleting one +// strands a worker that is connected at that moment. The age is measured by the +// database, like every other window here, so no replica's clock decides how old +// another replica's departure is. +func (r *Registry) PurgeDepartedBefore(ctx context.Context, olderThan time.Duration) (int64, error) { + // Refused rather than attempted, for the reason Claim refuses: now() and + // make_interval are PostgreSQL, and on the single-binary SQLite path this + // would fail with "no such function: now", which reads as a missing + // migration. + if !isPostgres(r.db) { + return 0, fmt.Errorf("purging departed connections: connection ownership requires PostgreSQL, this deployment runs on %q", r.db.Dialector.Name()) + } + res := r.db.WithContext(ctx). + Where("NOT ("+connectionIsHeld+") AND node_connections.disconnected_at < now() - make_interval(secs => ?)", + olderThan.Seconds()). + Delete(&NodeConnection{}) + if res.Error != nil { + return 0, fmt.Errorf("purging departed connections: %w", res.Error) + } + return res.RowsAffected, nil +} diff --git a/core/services/cluster/ownership_test.go b/core/services/cluster/ownership_test.go index cd422199c..17f88617b 100644 --- a/core/services/cluster/ownership_test.go +++ b/core/services/cluster/ownership_test.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "path/filepath" + "regexp" "strings" "sync" "time" @@ -57,25 +58,28 @@ func (r *sqlRecorder) only() string { return r.statements[0] } -// deleteOrder returns the tables the recorded statements deleted from, in the -// order they were issued. It is how a spec pins a lock order: the order two -// paths take the same tables in is a property of the SQL, and asserting it on -// an outcome instead would mean racing two transactions into a real deadlock. -func (r *sqlRecorder) deleteOrder() []string { +// writeTarget matches the table a statement writes to, anchored at the verb so +// the UPDATE inside an upsert's ON CONFLICT clause cannot be mistaken for one. +var writeTarget = regexp.MustCompile(`^\s*(?i:delete\s+from|update)\s+"?([a-z_]+)"?`) + +// writeOrder returns the tables the recorded statements wrote to, in the order +// they were issued. It is how a spec pins a lock order: the order two paths +// take the same tables in is a property of the SQL, and asserting it on an +// outcome instead would mean racing two transactions into a real deadlock. +// +// Deletes and updates count alike. What deadlocks two transactions is the order +// they take row locks in, and an update takes the same lock a delete does, so a +// path that stopped deleting a table and started updating it would still have +// to keep the order. +func (r *sqlRecorder) writeOrder() []string { r.mu.Lock() defer r.mu.Unlock() ExpectWithOffset(1, r.errs).To(BeEmpty(), "a recorded statement failed") var tables []string for _, stmt := range r.statements { - idx := strings.Index(strings.ToLower(stmt), "delete from ") - if idx < 0 { - continue + if m := writeTarget.FindStringSubmatch(stmt); m != nil { + tables = append(tables, m[1]) } - fields := strings.Fields(stmt[idx+len("delete from "):]) - if len(fields) == 0 { - continue - } - tables = append(tables, strings.Trim(fields[0], `"`)) } return tables } @@ -380,6 +384,179 @@ var _ = Describe("Connection ownership", func() { Expect(rows).To(HaveLen(1)) Expect(seen).To(HaveKey(rows[0].Epoch), "the stored epoch was never handed to any claimant") }) + + // A tunnel that goes away has to leave a mark. Deleting the row made "this + // worker's link dropped a moment ago" and "this worker has never connected + // here" one observation, so nothing above could tell a reconnect in flight + // from a departure, and every grace period built on top would have had + // nothing to measure from. + Describe("recording a departure", func() { + It("keeps the row and stamps disconnected_at when a claim is released", func() { + epoch, err := reg.Claim(ctx, "w1", "inst-a") + Expect(err).ToNot(HaveOccurred()) + + Expect(reg.Release(ctx, "w1", "inst-a", epoch)).To(Succeed()) + + var row cluster.NodeConnection + Expect(db.WithContext(ctx).Where("node_id = ?", "w1").First(&row).Error).To(Succeed(), + "the released row was deleted, so a worker that just left is indistinguishable from one that never dialled") + Expect(row.OwnerInstanceID).To(BeEmpty()) + Expect(row.DisconnectedAt).ToNot(BeNil()) + }) + + It("records the departure with one update measured on the database clock", func() { + epoch, err := reg.Claim(ctx, "w1", "inst-a") + Expect(err).ToNot(HaveOccurred()) + + rec := newSQLRecorder() + recording := cluster.NewRegistry(db.Session(&gorm.Session{Logger: rec})) + Expect(recording.Release(ctx, "w1", "inst-a", epoch)).To(Succeed()) + + sql := strings.ToLower(rec.only()) + Expect(sql).To(ContainSubstring("update")) + Expect(sql).ToNot(ContainSubstring("delete")) + // The departure is compared against a grace window by other + // replicas, so it has to be stamped on the one clock they share. A + // Go-side time.Now() would appear as a bound parameter instead, and + // clock skew would then widen or narrow every window built on it. + Expect(sql).To(ContainSubstring("now()")) + Expect(sql).ToNot(MatchRegexp(`disconnected_at"?\s*=\s*'`), + "the departure must not be a literal timestamp from this process's clock") + }) + + It("reports a released row as no connection from both reads", func() { + Expect(reg.Register(ctx, "inst-a", "10.0.0.1:8080", "v1")).To(Succeed()) + epoch, err := reg.Claim(ctx, "w1", "inst-a") + Expect(err).ToNot(HaveOccurred()) + Expect(reg.Release(ctx, "w1", "inst-a", epoch)).To(Succeed()) + + // Both reads: the row surviving is for whoever decides how old the + // departure is, and until something does, a departed row is no + // connection to a dialer and no connection to an observer. + _, _, err = reg.Owner(ctx, "w1") + Expect(err).To(MatchError(cluster.ErrNoConnection)) + _, _, err = reg.OwnerRow(ctx, "w1") + Expect(err).To(MatchError(cluster.ErrNoConnection)) + }) + + It("does not resolve a departed row through an instance whose id is empty", func() { + // Owner rejects a departed row itself rather than leaning on the + // instances join to miss it. The join only misses an empty owner + // for as long as no instance row carries an empty id, which is an + // accident of who registers rather than a property of ownership. + Expect(reg.Register(ctx, "", "10.0.0.1:8080", "v1")).To(Succeed()) + epoch, err := reg.Claim(ctx, "w1", "inst-a") + Expect(err).ToNot(HaveOccurred()) + Expect(reg.Release(ctx, "w1", "inst-a", epoch)).To(Succeed()) + + _, _, err = reg.Owner(ctx, "w1") + Expect(err).To(MatchError(cluster.ErrNoConnection)) + }) + + It("clears the departure when the worker reconnects", func() { + Expect(reg.Register(ctx, "inst-a", "10.0.0.1:8080", "v1")).To(Succeed()) + epoch, err := reg.Claim(ctx, "w1", "inst-a") + Expect(err).ToNot(HaveOccurred()) + Expect(reg.Release(ctx, "w1", "inst-a", epoch)).To(Succeed()) + + again, err := reg.Claim(ctx, "w1", "inst-a") + Expect(err).ToNot(HaveOccurred()) + Expect(again).ToNot(Equal(epoch), "uniqueness, never ordering") + + var row cluster.NodeConnection + Expect(db.WithContext(ctx).Where("node_id = ?", "w1").First(&row).Error).To(Succeed()) + Expect(row.DisconnectedAt).To(BeNil(), + "a row with an owner still carried a departure, so a reader has two answers to choose from") + owner, _, err := reg.Owner(ctx, "w1") + Expect(err).ToNot(HaveOccurred()) + Expect(owner).To(Equal("inst-a")) + }) + + It("does not let a fenced-out replica stamp a departure onto the live claim", func() { + Expect(reg.Register(ctx, "inst-a", "10.0.0.1:8080", "v1")).To(Succeed()) + stale, err := reg.Claim(ctx, "w1", "inst-a") + Expect(err).ToNot(HaveOccurred()) + live, err := reg.Claim(ctx, "w1", "inst-a") + Expect(err).ToNot(HaveOccurred()) + + Expect(reg.Release(ctx, "w1", "inst-a", stale)).To(MatchError(cluster.ErrNoConnection)) + + owner, epoch, err := reg.Owner(ctx, "w1") + Expect(err).ToNot(HaveOccurred()) + Expect(owner).To(Equal("inst-a")) + Expect(epoch).To(Equal(live)) + var row cluster.NodeConnection + Expect(db.WithContext(ctx).Where("node_id = ?", "w1").First(&row).Error).To(Succeed()) + Expect(row.DisconnectedAt).To(BeNil(), + "a replica that only just noticed its dead socket marked a live tunnel as departed") + }) + + It("purges a departure older than the retention and keeps a recent one", func() { + oldEpoch, err := reg.Claim(ctx, "old", "inst-a") + Expect(err).ToNot(HaveOccurred()) + Expect(reg.Release(ctx, "old", "inst-a", oldEpoch)).To(Succeed()) + // Aged on the database clock, which is the clock the purge measures + // on. Sleeping out a retention window in a spec is forbidden, and + // would be measuring this process's clock rather than the query. + Expect(db.WithContext(ctx).Exec( + `UPDATE node_connections SET disconnected_at = now() - make_interval(secs => ?) WHERE node_id = ?`, + 3600, "old").Error).To(Succeed()) + + freshEpoch, err := reg.Claim(ctx, "fresh", "inst-a") + Expect(err).ToNot(HaveOccurred()) + Expect(reg.Release(ctx, "fresh", "inst-a", freshEpoch)).To(Succeed()) + + purged, err := reg.PurgeDepartedBefore(ctx, 10*time.Minute) + Expect(err).ToNot(HaveOccurred()) + Expect(purged).To(Equal(int64(1))) + + var remaining []cluster.NodeConnection + Expect(db.WithContext(ctx).Find(&remaining).Error).To(Succeed()) + Expect(remaining).To(HaveLen(1)) + Expect(remaining[0].NodeID).To(Equal("fresh"), + "the purge took a departure that is still inside every window built on it") + }) + + It("measures the retention on the database clock", func() { + rec := newSQLRecorder() + recording := cluster.NewRegistry(db.Session(&gorm.Session{Logger: rec})) + _, err := recording.PurgeDepartedBefore(ctx, 10*time.Minute) + Expect(err).ToNot(HaveOccurred()) + + sql := strings.ToLower(rec.only()) + // Asserted on the statement because no outcome can separate the two + // here: every replica in a spec shares this host's clock, so a + // cutoff computed in Go agrees with the database's until two + // machines disagree, and then one replica purges departures its + // peers still consider recent. + Expect(sql).To(ContainSubstring("now()")) + Expect(sql).To(ContainSubstring("make_interval")) + Expect(sql).ToNot(MatchRegexp(`disconnected_at"?\s*<\s*'`), + "the retention cutoff must not be a literal timestamp from this process's clock") + }) + + It("never purges a row that is still held, whatever timestamp it carries", func() { + Expect(reg.Register(ctx, "inst-a", "10.0.0.1:8080", "v1")).To(Succeed()) + claimed, err := reg.Claim(ctx, "w1", "inst-a") + Expect(err).ToNot(HaveOccurred()) + // No writer produces a held row carrying a departure, so it is + // written here directly: the purge has to be pinned on the owner + // column rather than on the timestamp happening to be null, because + // deleting a held row strands a worker that is connected right now. + Expect(db.WithContext(ctx).Exec( + `UPDATE node_connections SET disconnected_at = now() - make_interval(secs => ?) WHERE node_id = ?`, + 3600, "w1").Error).To(Succeed()) + + purged, err := reg.PurgeDepartedBefore(ctx, time.Minute) + Expect(err).ToNot(HaveOccurred()) + Expect(purged).To(BeZero()) + + owner, epoch, err := reg.Owner(ctx, "w1") + Expect(err).ToNot(HaveOccurred(), "the purge deleted the row of a tunnel somebody holds") + Expect(owner).To(Equal("inst-a")) + Expect(epoch).To(Equal(claimed)) + }) + }) }) var _ = Describe("Connection ownership on a non-PostgreSQL dialect", func() { @@ -420,4 +597,26 @@ var _ = Describe("Connection ownership on a non-PostgreSQL dialect", func() { Expect(err).To(HaveOccurred()) Expect(err.Error()).To(ContainSubstring("requires PostgreSQL")) }) + + It("refuses to release, rather than clearing a claim outside the fence", func() { + Expect(cluster.Migrate(ctx, db)).To(Succeed()) + + err := cluster.NewRegistry(db).Release(ctx, "w1", "inst-a", 1) + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("requires PostgreSQL")) + Expect(err.Error()).ToNot(ContainSubstring("no such function"), + "a dialect that cannot record a departure must say so, not surface as a missing migration") + Expect(err).ToNot(MatchError(cluster.ErrNoConnection), + "a deployment with no cluster holds no claims; reporting a fenced-out release would let a caller conclude it lost one") + }) + + It("refuses to purge departures, rather than failing as a missing function", func() { + Expect(cluster.Migrate(ctx, db)).To(Succeed()) + + _, err := cluster.NewRegistry(db).PurgeDepartedBefore(ctx, time.Minute) + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("requires PostgreSQL")) + Expect(err.Error()).ToNot(ContainSubstring("no such function"), + "a dialect that cannot answer must say so, not surface as a missing migration") + }) }) diff --git a/docs/content/features/distributed-mode.md b/docs/content/features/distributed-mode.md index 9075c97a4..ee815c2ca 100644 --- a/docs/content/features/distributed-mode.md +++ b/docs/content/features/distributed-mode.md @@ -137,7 +137,7 @@ A tunnel credential does not replace `LOCALAI_REGISTRATION_TOKEN`. Without one, Only **backend** nodes are issued one. An agent worker has no inbound surface for the tunnel to replace and no client for it, so minting one would widen the credential surface for nothing; its row keeps an empty tunnel credential and the tunnel route refuses it like any other node without one. -The tunnel lands on exactly one frontend replica, and that replica records itself as the owner of the worker's connection in the `node_connections` table. When the socket dies the claim is dropped with it. If the replica stalls long enough for its peers to reap it, it re-claims the tunnels it still holds on a live session as soon as it re-registers, skipping any whose socket has already gone. That re-claim needs the replica to have an advertised address: without one it never had an instance row to begin with, and its tunnels are usable only by the replica holding them. +The tunnel lands on exactly one frontend replica, and that replica records itself as the owner of the worker's connection in the `node_connections` table. When the socket dies the claim is dropped, but the row stays behind with no owner and a `disconnected_at` stamp, so a worker that is re-dialling the load balancer can be told from one that has never connected. The row is deleted once that departure is older than ten liveness windows (five minutes). If the replica stalls long enough for its peers to reap it, it re-claims the tunnels it still holds on a live session as soon as it re-registers, skipping any whose socket has already gone. That re-claim needs the replica to have an advertised address: without one it never had an instance row to begin with, and its tunnels are usable only by the replica holding them. | Method | Path | Description | |--------|------|-------------|