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.
2026-10-15 14:47:09 +02:00
parent 45b6c347b2
commit 868e3ae377
3 changed files with 179 additions and 6 deletions
@@ -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,
@@ -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) {
@@ -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) {