fix(cluster): pin the departure retention to the loop that applies it

The membership tick's call to PurgeDepartedBefore was the only production
wiring this change introduced, and removing it left the suite green. A
retention nothing applies is a departure that never ages out, which is
the state the sweep's held-ness filter exists to make reachable at all.
A spec now ages a released row past DepartedRetention on the database
clock, starts a real Membership, and waits for the row to go.

Release and Deregister leaned on "no owner id is ever empty" to avoid
touching an already-departed row, which is the accident Owner refuses to
lean on. A departed row keeps its epoch and carries an empty owner, so a
release or a deregistration naming an empty id matched it and stamped a
fresh departure over the old one, making a worker that left long ago look
like one that has only just gone. Both now filter on connectionIsHeld.

The comment on DisconnectedAt claimed a held row never carries a
departure. A binary from before this column existed claims without
clearing the stamp, so a rolling upgrade produces exactly that row. The
comment now says what holds, and says to ask held-ness first and read the
stamp second.

The sweep's vocabulary follows the code: it records departures where the
comments still said it deleted rows, and its log line separates the
instance rows it deleted from the connection rows it left behind.

Assisted-by: Claude Opus 5 [claude-code]
Signed-off-by: Ettore Di Giacinto <mudler@localai.io>

core/services/cluster/membership.go

@@ -20,10 +20,11 @@ const (
 	// InstanceLiveness is how long a replica may go without a heartbeat before
 	// its peers treat it as gone: six consecutive misses.
 	//
-	// The window is generous on purpose. Declaring a replica dead deletes the
-	// connection rows it owned, and a worker whose row is deleted while its
-	// owner is merely slow has to be re-homed for nothing. The cost of waiting
-	// is bounded and symmetric: traffic for that worker is retried, not lost.
+	// The window is generous on purpose. Declaring a replica dead records a
+	// departure for every connection row it owned, and a worker recorded as
+	// departed while its owner is merely slow has to be re-homed for nothing.
+	// The cost of waiting is bounded and symmetric: traffic for that worker is
+	// retried, not lost.
 	InstanceLiveness = 30 * time.Second
 
 	// deregisterTimeout bounds the deregistration Stop performs. Shutdown is
@@ -192,10 +193,10 @@ func (m *Membership) tick(ctx context.Context) {
 		// carries no address, so the row has to be rebuilt from scratch.
 		//
 		// Register rebuilds the instance row ONLY. The sweep that removed it
-		// removed the connections this replica owned in the same transaction,
-		// so the tunnels still held here have to be claimed again or this
-		// replica serves workers that, as far as every other replica can see,
-		// are connected nowhere.
+		// recorded a departure for every connection this replica owned, in the
+		// same transaction, so the tunnels still held here have to be claimed
+		// again or this replica serves workers that, as far as every other
+		// replica can see, are connected nowhere.
 		xlog.Warn("Cluster instance row was reaped, re-registering", "id", m.id)
 		if err := m.reg.Register(ctx, m.id, m.addr, m.version); err == nil {
 			m.reclaimTunnels(ctx)
@@ -217,7 +218,10 @@ func (m *Membership) tick(ctx context.Context) {
 		return
 	}
 	if instances > 0 || connections > 0 {
-		xlog.Info("Reaped cluster state left by dead replicas", "instances", instances, "connections", connections)
+		// Two verbs because the sweep does two things: the instance rows are
+		// gone, the connection rows are still there and now record a departure.
+		xlog.Info("Swept cluster state left by dead replicas",
+			"instances_deleted", instances, "connections_departed", connections)
 	}
 
 	// The retention has an owner, and it is this sweep. A departure that
@@ -235,8 +239,9 @@ func (m *Membership) tick(ctx context.Context) {
 }
 
 // reclaimTunnels re-writes a claim for every worker tunnel this replica still
-// holds, after the sweep that deleted them. It is separate from tick only so
-// the lock around the registry reference is not held across the database work.
+// holds, after the sweep that recorded them as departed. It is separate from
+// tick only so the lock around the registry reference is not held across the
+// database work.
 func (m *Membership) reclaimTunnels(ctx context.Context) {
 	m.mu.Lock()
 	tunnels := m.tunnels
@@ -285,9 +290,12 @@ 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)
 		}
-		// 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).
+		// Held rows only, for the reason Owner filters rather than leaning on
+		// the join: an id column that is never empty is an accident of who
+		// registers, not a property, and an empty id here would match every
+		// departed row in the table and reset every departure's age.
+		if err := tx.Model(&NodeConnection{}).
+			Where("owner_instance_id = ? AND "+connectionIsHeld, id).
 			Updates(departure()).Error; err != nil {
 			return fmt.Errorf("recording departures for connections owned by %q: %w", id, err)
 		}

core/services/cluster/membership_test.go

@@ -178,6 +178,56 @@ var _ = Describe("Reaping dead replicas", func() {
 		Expect(owner).To(Equal("staying"), "deregistering one replica took another replica's claim")
 	})
 
+	It("does not restamp a departure when a replica with no id deregisters", func() {
+		// Deregister matches an owner id, and a departed row carries an empty
+		// one, so an empty id would match every departure in the table and reset
+		// each one's age. Nothing generates an empty id today, which is exactly
+		// why the filter has to be in the query rather than in that habit.
+		Expect(reg.Register(ctx, "leaving", "10.0.0.2:8080", "v1")).To(Succeed())
+		epoch, err := reg.Claim(ctx, "w1", "leaving")
+		Expect(err).ToNot(HaveOccurred())
+		Expect(reg.Release(ctx, "w1", "leaving", epoch)).To(Succeed())
+		var before cluster.NodeConnection
+		Expect(db.WithContext(ctx).Where("node_id = ?", "w1").First(&before).Error).To(Succeed())
+
+		Expect(reg.Deregister(ctx, "")).To(Succeed())
+
+		var after cluster.NodeConnection
+		Expect(db.WithContext(ctx).Where("node_id = ?", "w1").First(&after).Error).To(Succeed())
+		Expect(after.DisconnectedAt).ToNot(BeNil())
+		Expect(*after.DisconnectedAt).To(Equal(*before.DisconnectedAt),
+			"a departure that had already been recorded was stamped again, so its age restarted")
+	})
+
+	It("purges a departure that has aged past the retention, from the loop that sweeps it", func() {
+		// The retention exists only if something applies it, and this loop is
+		// the only thing that does. A PurgeDepartedBefore nobody calls leaves a
+		// row per worker that ever dialled this deployment, which is the state
+		// the sweep's held-ness filter exists to make reachable in the first
+		// place.
+		Expect(reg.Register(ctx, "me", "10.0.0.1:8080", "v1")).To(Succeed())
+		epoch, err := reg.Claim(ctx, "w1", "me")
+		Expect(err).ToNot(HaveOccurred())
+		Expect(reg.Release(ctx, "w1", "me", epoch)).To(Succeed())
+		// Aged on the database clock, past the retention the loop applies.
+		Expect(db.WithContext(ctx).Exec(
+			`UPDATE node_connections SET disconnected_at = now() - make_interval(secs => ?) WHERE node_id = ?`,
+			(cluster.DepartedRetention + time.Minute).Seconds(), "w1").Error).To(Succeed())
+
+		membership := cluster.NewMembership(reg, "me", "10.0.0.1:8080", "v1")
+		Expect(membership.Start(ctx)).To(Succeed())
+		DeferCleanup(membership.Stop)
+
+		// Polled rather than slept: the loop ticks on its own schedule, and the
+		// spec is about the call happening at all, not about when.
+		Eventually(func() (int64, error) {
+			var rows int64
+			err := db.WithContext(ctx).Model(&cluster.NodeConnection{}).Count(&rows).Error
+			return rows, err
+		}, "20s", "250ms").Should(BeZero(),
+			"the loop that sweeps dead replicas never applied the departure retention")
+	})
+
 	It("deregisters when the membership loop stops", func() {
 		membership := cluster.NewMembership(reg, "me", "10.0.0.1:8080", "v1")
 		Expect(membership.Start(ctx)).To(Succeed())

core/services/cluster/ownership.go

@@ -70,8 +70,16 @@ type NodeConnection struct {
 	// 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.
+	// Every writer in this package leaves it null while the row is held: Claim
+	// clears it in the statement that writes the owner, and only a departure
+	// sets it. That is a property of these writers, not of the table, and a
+	// mixed-version deployment breaks it: a replica running a binary from
+	// before this column existed claims without clearing the stamp, so a held
+	// row can carry the departure a newer replica recorded.
+	//
+	// So held-ness is the question, and the stamp only refines it. Ask
+	// connectionIsHeld first and read this second; a reader that reads the
+	// stamp alone reports a connected worker as gone.
 	DisconnectedAt *time.Time `gorm:"index" json:"disconnected_at,omitempty"`
 }
 
@@ -83,8 +91,11 @@ type NodeConnection struct {
 // 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 writes ask it too, and for the same reason Owner does rather than leaning
+// on its join: Release and Deregister already match an owner id, but that only
+// excludes a departed row while no owner id is ever empty, which is an accident
+// of who registers. An empty one would match every departure in the table and
+// reset its age.
 //
 // 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
@@ -340,7 +351,11 @@ func (r *Registry) Release(ctx context.Context, nodeID, ownerID string, epoch in
 	// read off RowsAffected.
 	res := r.db.WithContext(ctx).
 		Model(&NodeConnection{}).
-		Where("node_id = ? AND owner_instance_id = ? AND epoch = ?", nodeID, ownerID, epoch).
+		// The held-ness filter is not implied by the owner match: a departed row
+		// keeps its epoch, so a release naming an empty owner would match it and
+		// stamp a fresh departure over the old one, making a worker that left
+		// long ago look like one that has only just gone.
+		Where("node_id = ? AND owner_instance_id = ? AND epoch = ? AND "+connectionIsHeld, nodeID, ownerID, epoch).
 		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)

core/services/cluster/ownership_test.go

@@ -491,6 +491,26 @@ var _ = Describe("Connection ownership", func() {
 				"a replica that only just noticed its dead socket marked a live tunnel as departed")
 		})
 
+		It("refuses a release that names no owner, so a recorded departure cannot be aged backwards", func() {
+			// A departed row keeps the epoch of the claim that left, and its
+			// owner is the empty string, so a release naming an empty owner
+			// matches it on both columns. Stamping a fresh departure there would
+			// make a worker that left long ago look like one that has only just
+			// gone, which is the difference every window above is measured from.
+			epoch, err := reg.Claim(ctx, "w1", "inst-a")
+			Expect(err).ToNot(HaveOccurred())
+			Expect(reg.Release(ctx, "w1", "inst-a", epoch)).To(Succeed())
+			var before cluster.NodeConnection
+			Expect(db.WithContext(ctx).Where("node_id = ?", "w1").First(&before).Error).To(Succeed())
+
+			Expect(reg.Release(ctx, "w1", "", epoch)).To(MatchError(cluster.ErrNoConnection))
+
+			var after cluster.NodeConnection
+			Expect(db.WithContext(ctx).Where("node_id = ?", "w1").First(&after).Error).To(Succeed())
+			Expect(after.DisconnectedAt).ToNot(BeNil())
+			Expect(*after.DisconnectedAt).To(Equal(*before.DisconnectedAt))
+		})
+
 		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())