From 868e3ae37780fcca5f0fb9c7b0c9a1ddc59c2725 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 15 Oct 2026 14:47:09 +0200 Subject: [PATCH] Never mark a call abandoned over its recorded finish when a controller starts (issue 711) A starting controller's sweep read each call it saw running and wrote it back abandoned unconditionally, so a finish written in between, with its kept answer, was lost. The mark is now a compare-and-set on the call's record: a finish already recorded, or racing the mark, stands. --- internal/link/calls.go | 20 +++-- internal/link/calls_kept.go | 40 ++++++++++ internal/link/calls_kept_test.go | 125 +++++++++++++++++++++++++++++++ 3 files changed, 179 insertions(+), 6 deletions(-) diff --git a/internal/link/calls.go b/internal/link/calls.go index 262defee..6332c705 100644 --- a/internal/link/calls.go +++ b/internal/link/calls.go @@ -106,6 +106,13 @@ type CallKeeper interface { Kept(ctx context.Context, id string) (Call, bool, error) // Recent is the kept calls, newest first, without their answers. Recent(ctx context.Context) ([]Call, error) + // MarkAbandoned marks the call id abandoned, unless its record already says it finished (novox/hq + // issue 711): a finish recorded before, or racing, the mark must stand — a call whose answer was + // kept never reads abandoned. It is the record's own compare-and-set that decides the race, not + // the order these two writers happen to run in, so it retries against a finish landing in + // between. It returns the record as it stands once this returns: marked abandoned, the finish + // that won instead, or a zero Call when the record is gone. + MarkAbandoned(ctx context.Context, id string) (Call, error) } // CallLog keeps the latest calls a holder served. @@ -171,14 +178,15 @@ func (l *CallLog) Durably(ctx context.Context, keeper CallKeeper, holder string, if c.State != CallRunning || c.Holder == holder { continue } - whole, found, err := keeper.Kept(ctx, c.ID) - if err != nil || !found { - whole = c - } - whole.State = CallAbandoned - if err := keeper.Keep(ctx, whole); err != nil { + final, err := keeper.MarkAbandoned(ctx, c.ID) + if err != nil { return fmt.Errorf("marking %s abandoned: %w", c.ID, err) } + if final.State != CallAbandoned { + // Its finish was recorded before this sweep reached it, or while it raced this mark + // (novox/hq issue 711): that word stands, with its kept answer, and is never overwritten. + continue + } if logger != nil { logger.Printf("%s (%s.%s, asked %s by %s) was running under %s (epoch %d), which stopped: marked "+ "abandoned — it may have done part of what it was asked, and nothing will finish it", c.ID, c.Seat, diff --git a/internal/link/calls_kept.go b/internal/link/calls_kept.go index 06e4fff2..2cd9b787 100644 --- a/internal/link/calls_kept.go +++ b/internal/link/calls_kept.go @@ -87,6 +87,46 @@ func (b *BusCalls) Kept(ctx context.Context, id string) (Call, bool, error) { return c, true, nil } +// MarkAbandoned marks id abandoned unless its record already says it finished (novox/hq issue 711): +// a controller starting finds a call running under a holder that stopped, but that holder may have +// written the finish — with its kept answer — around the same moment. The record's own +// compare-and-set decides the race rather than the order these two writers happen to run in: this +// reads the record, and if it still says running, writes abandoned at that revision; a finish landing +// in between fails the write with a revision mismatch, and this rereads and finds the finish, which +// then stands. A record already resolved, or gone, is returned as found, so the caller never writes +// abandoned over it. +func (b *BusCalls) MarkAbandoned(ctx context.Context, id string) (Call, error) { + for { + entry, err := b.kv.Get(ctx, id) + if errors.Is(err, jetstream.ErrKeyNotFound) || errors.Is(err, jetstream.ErrInvalidKey) { + return Call{}, nil + } + if err != nil { + return Call{}, err + } + var c Call + if err := json.Unmarshal(entry.Value(), &c); err != nil { + return Call{}, fmt.Errorf("the record of %s on the bus is not a call: %w", id, err) + } + if c.State != CallRunning { + // A finish (or an earlier mark) is already the record's word: it stands, untouched. + return c, nil + } + c.State = CallAbandoned + record, err := json.Marshal(c) + if err != nil { + return Call{}, err + } + if _, err := b.kv.Update(ctx, id, record, entry.Revision()); err != nil { + if errors.Is(err, jetstream.ErrKeyRevisionMismatch) { + continue // something — a finish, most likely — wrote since: reread and judge it again + } + return Call{}, err + } + return c, nil + } +} + // Recent is every call the bucket holds, newest first, without answers: read once through a watch // of the records, which hands over the current value of each and then says it has. func (b *BusCalls) Recent(ctx context.Context) ([]Call, error) { diff --git a/internal/link/calls_kept_test.go b/internal/link/calls_kept_test.go index 84a23be7..7652e778 100644 --- a/internal/link/calls_kept_test.go +++ b/internal/link/calls_kept_test.go @@ -105,6 +105,131 @@ func TestNatsACallsOutcomeOutlivesItsController(t *testing.T) { } } +// racingRecent wraps a CallKeeper so a finish lands right after its Recent() snapshot is taken, as +// if that finish raced the moment the snapshot was read (novox/hq issue 711): the snapshot still +// says a call is running, but by the time anything acts on that entry the record has already moved +// on. +type racingRecent struct { + CallKeeper + thenLand func() +} + +func (r racingRecent) Recent(ctx context.Context) ([]Call, error) { + out, err := r.CallKeeper.Recent(ctx) + if err == nil && r.thenLand != nil { + r.thenLand() + } + return out, err +} + +// **A finish recorded before the sweep reaches a call stands, with its kept answer** (novox/hq issue +// 711): a stale "running" snapshot is no excuse to overwrite a record that has since said how the +// call ended. A caller reading `abandoned` for a call that succeeded would redo work already done. +func TestNatsAFinishRacingTheSweepsSnapshotReadsFinished(t *testing.T) { + keeper := callsBucket(t) + running := Call{ID: "call-711-1", Seat: "mesh-controller", Verb: "push", State: CallRunning, Holder: "controller@one"} + if err := keeper.Keep(t.Context(), running); err != nil { + t.Fatal(err) + } + waitKept(t, keeper, running.ID, CallRunning) + + finished := running + finished.State = CallAnswered + finished.Answer = json.RawMessage(`"anchor told"`) + wrapped := racingRecent{CallKeeper: keeper, thenLand: func() { + if err := keeper.Keep(context.Background(), finished); err != nil { + t.Error(err) + } + waitKept(t, keeper, running.ID, CallAnswered) + }} + + var said bytes.Buffer + second := NewCallLog() + if err := second.Durably(t.Context(), wrapped, "controller@two", log.New(&said, "", 0)); err != nil { + t.Fatal(err) + } + got, found, err := second.Get(running.ID) + if err != nil || !found || got.State != CallAnswered || !strings.Contains(string(got.Answer), "anchor told") { + t.Fatalf("its finish did not stand: %+v %v %v", got, found, err) + } + if strings.Contains(said.String(), running.ID) { + t.Errorf("a finished call was said abandoned: %q", said.String()) + } +} + +// A call left running with no finish recorded is still marked abandoned — unchanged (novox/hq issue +// 711): only a finish actually recorded for it may stand instead. +func TestNatsARunningCallWithNoFinishIsStillAbandoned(t *testing.T) { + keeper := callsBucket(t) + running := Call{ID: "call-711-2", Seat: "mesh-controller", Verb: "plans", State: CallRunning, Holder: "controller@one"} + if err := keeper.Keep(t.Context(), running); err != nil { + t.Fatal(err) + } + waitKept(t, keeper, running.ID, CallRunning) + + var said bytes.Buffer + second := NewCallLog() + if err := second.Durably(t.Context(), keeper, "controller@two", log.New(&said, "", 0)); err != nil { + t.Fatal(err) + } + got, found, err := second.Get(running.ID) + if err != nil || !found || got.State != CallAbandoned || !strings.Contains(said.String(), running.ID) { + t.Fatalf("got %+v %v %v; said %q", got, found, err, said.String()) + } +} + +// delayedKV wraps a jetstream.KeyValue so a hook runs just before Update, to land a write between a +// reader's Get and its own compare-and-set write. +type delayedKV struct { + jetstream.KeyValue + before func() +} + +func (d delayedKV) Update(ctx context.Context, key string, value []byte, revision uint64) (uint64, error) { + d.before() + return d.KeyValue.Update(ctx, key, value, revision) +} + +// **The race where a finish lands after the sweep's own read but before its write: the finish +// wins** (novox/hq issue 711). The sweep's write is a compare-and-set on the revision it read, so a +// finish landing in between invalidates it; the sweep rereads, finds the finish, and leaves it alone. +func TestNatsAFinishRacingTheSweepsWriteWins(t *testing.T) { + keeper := callsBucket(t) + running := Call{ID: "call-711-3", Seat: "mesh-controller", Verb: "push", State: CallRunning, Holder: "controller@one"} + if err := keeper.Keep(t.Context(), running); err != nil { + t.Fatal(err) + } + waitKept(t, keeper, running.ID, CallRunning) + + finished := running + finished.State = CallFinishedAfter + finished.Answer = json.RawMessage(`"anchor told"`) + landed := false + racey := &BusCalls{kv: delayedKV{KeyValue: keeper.kv, before: func() { + if landed { + return + } + landed = true + if err := keeper.Keep(context.Background(), finished); err != nil { + t.Error(err) + } + waitKept(t, keeper, running.ID, CallFinishedAfter) + }}} + + var said bytes.Buffer + second := NewCallLog() + if err := second.Durably(t.Context(), racey, "controller@two", log.New(&said, "", 0)); err != nil { + t.Fatal(err) + } + got, found, err := second.Get(running.ID) + if err != nil || !found || got.State != CallFinishedAfter || !strings.Contains(string(got.Answer), "anchor told") { + t.Fatalf("the finish did not win the race: %+v %v %v", got, found, err) + } + if strings.Contains(said.String(), running.ID) { + t.Errorf("a finished call was said abandoned: %q", said.String()) + } +} + // The bucket holds the last thousand calls or fourteen days: a call is two keys, so the stream under // it holds twice as many messages, and asserting it again keeps the count. func TestNatsTheCallsBucketIsBounded(t *testing.T) {