From 5e2938ebf0f54a91cd0568d5b80113630c8b620c Mon Sep 17 00:00:00 2001 From: Ettore Di Giacinto Date: Tue, 1 Sep 2026 06:46:02 +0000 Subject: [PATCH] feat(cluster): resolve tunnel ownership against a live owner OwnerRow is a bare row read of node_connections. A connection row outlives the replica that wrote it: a replica that dies stops heartbeating, but its rows survive until a peer's sweep removes them, which is up to InstanceLiveness plus one InstanceHeartbeat later. For that whole window the table names a process that is gone. The next component phase 2 builds is the relaying dialer, and a dialer reading OwnerRow would relay into a corpse for roughly 35 seconds after every replica death, then report the worker as unreachable when it is in fact absent, which is the distinction the phase 1 end-to-end specs pinned. Owner is the resolving read: one statement joining instances, returning ErrNoConnection when the row is missing OR its owner is not live. Both cases are one answer on purpose, since both mean no replica here holds this tunnel; they differ only in which sweep has run. It is one statement, not a row read followed by an instance lookup, because between two statements the owner can die and the caller would act on an owner the second read would have rejected. OwnerRow stays, unjoined, for readers that need the row itself, and a spec holds the two apart: with an aged-out owner, OwnerRow still names it and Owner refuses, so neither can quietly become the other. The liveness predicate is now one string, instanceIsLive, shared by Live and by Owner's join. Two spellings of one fact drift, and this drift would show as a relay to a replica one query calls dead and another calls alive. It is table-qualified so it is unambiguous inside the join, and the cutoff stays on the database clock, so replica clock skew cannot widen or narrow the window. Both mutations were run. Dropping the liveness predicate from the join fails 3 specs, the aged-owner one among them. Replacing the database clock with a Go-side time.Now() fails 1: the aged-owner specs still pass, because the two clocks agree on one host, and only the recorded-SQL spec sees the literal timestamp. That is why that spec exists. Assisted-by: Claude Opus 5 [claude-code] Signed-off-by: Ettore Di Giacinto --- core/services/cluster/instance.go | 20 ++++- core/services/cluster/ownership.go | 55 ++++++++++-- core/services/cluster/ownership_test.go | 114 ++++++++++++++++++++++++ 3 files changed, 179 insertions(+), 10 deletions(-) diff --git a/core/services/cluster/instance.go b/core/services/cluster/instance.go index 4140ce6bc..c991d0da8 100644 --- a/core/services/cluster/instance.go +++ b/core/services/cluster/instance.go @@ -91,13 +91,27 @@ func (r *Registry) Heartbeat(ctx context.Context, id string) error { return nil } +// instanceIsLive is the one predicate that decides whether a replica is still +// alive, and it takes the window in seconds as its single bind parameter. Every +// reader of that fact uses this string: Live to list the survivors, Owner to +// refuse an owner that is not among them. Two spellings of one fact drift, and +// the drift would show up as a relay to a replica one query calls dead and +// another calls alive. +// +// The column is table-qualified because Owner reads it across a join, where an +// unqualified last_seen would be ambiguous. Postgres folds the unquoted name to +// the same table gorm quotes, so the qualification costs Live nothing. +// +// The cutoff is computed by the database for the same reason Register stamps +// there: liveness is compared across replicas, so a reader's own clock must not +// decide whether another replica is alive. +const instanceIsLive = `instances.last_seen > now() - make_interval(secs => ?)` + // Live returns the instances whose LastSeen is newer than now-within. func (r *Registry) Live(ctx context.Context, within time.Duration) ([]Instance, error) { var out []Instance - // The cutoff is computed by the database for the same reason Register stamps - // there: a reader's clock must not decide whether another replica is alive. if err := r.db.WithContext(ctx). - Where("last_seen > now() - make_interval(secs => ?)", within.Seconds()). + Where(instanceIsLive, within.Seconds()). Order("id"). Find(&out).Error; err != nil { return nil, fmt.Errorf("listing live instances: %w", err) diff --git a/core/services/cluster/ownership.go b/core/services/cluster/ownership.go index 131b3d0b9..48d9f94c5 100644 --- a/core/services/cluster/ownership.go +++ b/core/services/cluster/ownership.go @@ -168,14 +168,12 @@ func (r *Registry) Claim(ctx context.Context, nodeID, ownerID string) (int64, er // It answers "what does the table say", NOT "who holds this tunnel". The owner // it names may be dead: a replica that dies stops heartbeating, and its rows // survive until another replica's sweep removes them, which is up to -// InstanceLiveness plus one InstanceHeartbeat later. Any caller that ACTS on -// the answer must join instances itself and treat a non-live owner as -// ErrNoConnection; relaying to the row without that check is relaying into a -// process that is gone. +// InstanceLiveness plus one InstanceHeartbeat later. // -// The name says row on purpose, so that the joining version can take the plain -// name when phase 2 introduces the first caller that needs it. Nothing in -// phase 1 reads this outside tests, which is why the join is not here yet. +// Callers that ACT on the answer want Owner, which joins instances and treats a +// non-live owner as ErrNoConnection. This one is for callers that need to see +// the row itself, such as a sweeper deciding what to clean up, or a spec +// proving the two reads differ. 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 @@ -188,6 +186,49 @@ func (r *Registry) OwnerRow(ctx context.Context, nodeID string) (string, int64, return conn.OwnerInstanceID, conn.Epoch, nil } +// Owner returns the replica that holds nodeID's tunnel AND is still live, with +// the epoch of that claim, or ErrNoConnection when there is no such replica. +// +// This is the read anything that ACTS on the answer must use. A connection row +// outlives its owner: a replica that dies stops heartbeating but its rows stay +// until a peer's sweep removes them, which is up to InstanceLiveness plus one +// InstanceHeartbeat later. For that whole window OwnerRow names a process that +// 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. +// +// 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 +// second read would have rejected. The join makes the two facts one snapshot. +// +// The window is InstanceLiveness rather than a parameter, which is the window +// the membership loop sweeps with. A caller free to pick its own could keep +// relaying to a replica the sweeper has already declared dead, or give up on +// one the sweeper is still keeping. +func (r *Registry) Owner(ctx context.Context, nodeID string) (string, int64, error) { + var conn NodeConnection + err := r.db.WithContext(ctx). + Model(&NodeConnection{}). + // Only the connection's own columns are selected: the join exists to + // filter, and SELECT * across it would hand gorm the instances columns + // to scan into a NodeConnection. + Select("node_connections.*"). + Joins("JOIN instances ON instances.id = node_connections.owner_instance_id AND "+instanceIsLive, InstanceLiveness.Seconds()). + Where("node_connections.node_id = ?", nodeID). + Take(&conn).Error + if errors.Is(err, gorm.ErrRecordNotFound) { + return "", 0, fmt.Errorf("looking up live owner of node %q: %w", nodeID, ErrNoConnection) + } + if err != nil { + return "", 0, fmt.Errorf("looking up live owner of node %q: %w", nodeID, 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 diff --git a/core/services/cluster/ownership_test.go b/core/services/cluster/ownership_test.go index cd59f8056..b7abbbd6a 100644 --- a/core/services/cluster/ownership_test.go +++ b/core/services/cluster/ownership_test.go @@ -188,6 +188,120 @@ var _ = Describe("Connection ownership", func() { Expect(err).To(MatchError(cluster.ErrNoConnection)) }) + // Owner is the resolving read: it answers "who holds this tunnel and can be + // relayed to", where OwnerRow answers "what does the table say". The gap + // between the two is a whole liveness window wide, because a replica that + // dies leaves its connection rows behind until a peer's sweep removes them. + Describe("resolving the owner that can actually be relayed to", func() { + // Aged far enough past InstanceLiveness that the exact window boundary + // is not what these specs are measuring. + agedOut := 10 * time.Minute + + // age rewrites an instance's heartbeat into the past. Sleeping for a + // liveness window is forbidden in a spec, and would be measuring the + // clock rather than the query. + age := func(id string, by time.Duration) { + ExpectWithOffset(1, db.Model(&cluster.Instance{}).Where("id = ?", id). + Update("last_seen", time.Now().Add(-by)).Error).To(Succeed()) + } + + It("names an owner whose replica is live", 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()) + + owner, epoch, err := reg.Owner(ctx, "w1") + Expect(err).ToNot(HaveOccurred()) + Expect(owner).To(Equal("inst-a")) + Expect(epoch).To(Equal(claimed), "the resolved epoch must be the fence token the claim was handed") + }) + + It("refuses to name an owner that has no instance row at all", func() { + // What a completed sweep leaves for the moment between deleting the + // instance row and deleting the connections it orphaned, and what a + // re-registering replica's own connection rows look like meanwhile. + _, err := reg.Claim(ctx, "w1", "inst-gone") + Expect(err).ToNot(HaveOccurred()) + + _, _, err = reg.Owner(ctx, "w1") + Expect(err).To(MatchError(cluster.ErrNoConnection)) + }) + + It("refuses to name an owner whose heartbeat has aged past the liveness window", func() { + // The window this task exists to close: the replica is dead, no peer + // has swept it yet, and the row still names it. + Expect(reg.Register(ctx, "inst-a", "10.0.0.1:8080", "v1")).To(Succeed()) + _, err := reg.Claim(ctx, "w1", "inst-a") + Expect(err).ToNot(HaveOccurred()) + age("inst-a", agedOut) + + _, _, err = reg.Owner(ctx, "w1") + Expect(err).To(MatchError(cluster.ErrNoConnection)) + }) + + It("names an owner again once its heartbeat comes back", func() { + // Liveness is a window, not a latch: a replica that stalls and + // recovers still owns the sockets it never dropped, so resolution + // has to follow last_seen rather than remember a verdict. + 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()) + age("inst-a", agedOut) + Expect(reg.Heartbeat(ctx, "inst-a")).To(Succeed()) + + owner, epoch, err := reg.Owner(ctx, "w1") + Expect(err).ToNot(HaveOccurred()) + Expect(owner).To(Equal("inst-a")) + Expect(epoch).To(Equal(claimed)) + }) + + It("still reports the dead owner through OwnerRow, which is why the two reads are separate", 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()) + age("inst-a", agedOut) + + owner, epoch, err := reg.OwnerRow(ctx, "w1") + Expect(err).ToNot(HaveOccurred(), "OwnerRow reads the row and nothing else; hiding the dead owner here would leave the sweeper with no way to see what it has to clean up") + Expect(owner).To(Equal("inst-a")) + Expect(epoch).To(Equal(claimed)) + + _, _, err = reg.Owner(ctx, "w1") + Expect(err).To(MatchError(cluster.ErrNoConnection), "Owner and OwnerRow must not agree here, or one of them is redundant") + }) + + It("reports a node with no connection at all the same way", func() { + _, _, err := reg.Owner(ctx, "ghost") + Expect(err).To(MatchError(cluster.ErrNoConnection)) + }) + + It("resolves in one joined statement measured on the database clock", func() { + Expect(reg.Register(ctx, "inst-a", "10.0.0.1:8080", "v1")).To(Succeed()) + _, err := reg.Claim(ctx, "w1", "inst-a") + Expect(err).ToNot(HaveOccurred()) + + rec := newSQLRecorder() + recording := cluster.NewRegistry(db.Session(&gorm.Session{Logger: rec})) + _, _, err = recording.Owner(ctx, "w1") + Expect(err).ToNot(HaveOccurred()) + + sql := strings.ToLower(rec.only()) + // only() rules out the read-then-look-up shape: two statements + // leave a window in which the owner dies between them, which is the + // race the join closes. + Expect(sql).To(ContainSubstring("join")) + Expect(sql).To(ContainSubstring("instances")) + // Liveness is compared across replicas, so the cutoff has to be + // computed on the one clock they all share. A Go-side time.Now() + // would appear as a bound parameter and a plain comparison instead, + // and replica clock skew would then widen or narrow the window. + Expect(sql).To(ContainSubstring("now()")) + Expect(sql).To(ContainSubstring("make_interval")) + Expect(sql).ToNot(MatchRegexp(`last_seen\s*>\s*'`), + "the liveness cutoff must not be a literal timestamp from this process's clock") + }) + }) + It("claims in one statement that draws its epoch from the database sequence and stamps on the database clock", func() { rec := newSQLRecorder() recording := cluster.NewRegistry(db.Session(&gorm.Session{Logger: rec}))