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