A-282 corrected (join 11): a moved step's bound is written again by a revision-checked update, never a delete

c5a6871f created the key once and never wrote it again, so a step that kept moving was bounded from its first movement (issue 1930). Arm now creates the key when absent; one ending no earlier is armed already; one ending earlier is written again by an update against the revision read, carrying the bound's time to live; a waiter whose update loses reads the key again and takes the newer bound as armed. The test moves a step twice past its first bound, with two waiters each time, and reads one write per movement, its time to live, and no delete. Builds: design 47 G A-282, A-115, §544 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
2026-11-12 13:16:22 +01:00
parent c663b27848
commit 339964920b
3 changed files with 126 additions and 32 deletions
@@ -183,7 +183,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		Claimer: build.Claimer{Store: kvStore{buildsKV}, By: cred.Node + "." + cred.Module},
 		Seats:   seats,
 		Asks: buildrun.Asker{Seats: seats, Publisher: asks, Waiter: buildrun.LostAskWaiter{Inner: heard, Dead: dead, Hearing: hearing}, Bound: bound, Dropper: buildrun.SeatDrops{Caller: calls},
-			Bounds: &stepBounds{bounds: pubBoundsKV.KV, now: time.Now, logf: logger.Printf}, Expiry: stepExpiry},
+			Bounds: &stepBounds{bounds: pubBoundsKV.KV, js: js, now: time.Now, logf: logger.Printf}, Expiry: stepExpiry},
 		Held: heldOnTheBus{js: js, seats: seats},
 		// An ask that can no longer be answered is asked again (issue 1945), read at the step bound: a context's bound.
 		Gone:       heldOnTheBus{js: js, seats: seats},
@@ -2,10 +2,12 @@ package main

 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"time"

+	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"

 	"mesh-delivery-builder/internal/versions"
@@ -15,22 +17,61 @@ import (
 // A-118): the bus's marker of a key's end, heard on the watch of the state, is told to buildrun.Asker's Expiry. No timer.
 type stepBounds struct {
 	bounds func() jetstream.KeyValue
-	now    func() time.Time
-	logf   func(string, ...any)
+	// js writes a moved step's bound again: an update against the revision read, carrying the bound's time to live.
+	js   jetstream.JetStream
+	now  func() time.Time
+	logf func(string, ...any)
 }

-// Arm is buildrun.StepBounds' Arm: the key is created once (design 47 G A-282). A waiter that finds it already there, made by
-// another waiter of the same ask, takes it as armed and waits on the same key; it is never deleted to be made again.
+// Arm is buildrun.StepBounds' Arm (design 47 G A-282, issue 1930): the key is never deleted to be made again. It is created
+// when absent; one already there that ends no earlier was armed by another waiter of the same ask and is taken as armed; one
+// that ends earlier is the step's bound before it moved, and is written again by an update against the revision read, with
+// the bound's own time to live, so the bound counts from the step's last movement. A waiter whose update loses to another's
+// reads the key again and takes the newer bound as armed.
 func (s *stepBounds) Arm(ctx context.Context, key, why string, bound time.Duration) error {
 	kv := s.bounds()
 	if kv == nil {
 		return fmt.Errorf("%s is not on the bus yet", versions.BoundsBucket)
 	}
-	_, err := kv.Create(ctx, key, versions.EncodeBound(why, s.now().Add(bound)), jetstream.KeyTTL(bound))
-	if errors.Is(err, jetstream.ErrKeyExists) {
-		return nil // armed already
+	ends := s.now().Add(bound)
+	value := versions.EncodeBound(why, ends)
+	for range 2 { // a key that ended between the create and the read is created once more
+		_, err := kv.Create(ctx, key, value, jetstream.KeyTTL(bound))
+		if !errors.Is(err, jetstream.ErrKeyExists) {
+			return err
+		}
+		held, err := kv.Get(ctx, key)
+		if errors.Is(err, jetstream.ErrKeyNotFound) {
+			continue
+		}
+		if err != nil {
+			return err
+		}
+		var was versions.Bound
+		if err := json.Unmarshal(held.Value(), &was); err == nil && !was.Ends.Before(ends) {
+			return nil // armed already, at this movement or a later one
+		}
+		if s.js == nil {
+			return fmt.Errorf("the bound %s moved and is not written again: no JetStream to update it with", key)
+		}
+		_, err = s.js.PublishMsg(ctx, &nats.Msg{Subject: "$KV." + kv.Bucket() + "." + key, Data: value},
+			jetstream.WithExpectLastSequencePerSubject(held.Revision()), jetstream.WithMsgTTL(bound))
+		if wrongLast(err) {
+			if _, err := kv.Get(ctx, key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
+				return err
+			}
+			return nil // another waiter wrote the newer bound: armed
+		}
+		return err
 	}
-	return err
+	return fmt.Errorf("the bound %s ended each time it was read", key)
+}
+
+// wrongLast is a write that lost to another's: the revision it was made against is no longer the key's last.
+func wrongLast(err error) bool {
+	var api *jetstream.APIError
+	return errors.As(err, &api) && (api.ErrorCode == jetstream.JSErrCodeStreamWrongLastSequence ||
+		api.ErrorCode == jetstream.JSErrCodeStreamWrongLastSequenceConstant)
 }

 // End is buildrun.StepBounds' End.
@@ -3,6 +3,7 @@
 package main

 import (
+	"encoding/json"
 	"sync"
 	"testing"
 	"time"
@@ -12,42 +13,94 @@ import (
 	"mesh-delivery-builder/internal/versions"
 )

-// A-282: an ask's step bound key is created once. Two waiters on one ask arm it side by side and each takes it as armed; a
-// waiter that arms it again finds it there and leaves it as it stands, never deleted to be made again.
+// A-282 (corrected), issue 1930: an ask's step bound key is never deleted to be made again. Two waiters on one ask arm it side
+// by side and each takes it as armed; a waiter arming the same movement again writes nothing. A step that moves twice past its
+// first bound has its key written again each time, by an update against the revision read with the bound's own time to live,
+// so the bound counts from the last movement and the step is not judged standing still; two waiters seeing one movement write
+// it once.
 func TestTwoWaitersOnOneAskArmItsStepBoundOnce(t *testing.T) {
 	b := startBed(t)
 	kv, err := b.js.KeyValue(b.ctx, versions.BoundsBucket)
 	if err != nil {
 		t.Fatal(err)
 	}
+	var mu sync.Mutex
 	at := time.Date(2026, 11, 12, 12, 0, 0, 0, time.UTC)
+	clock := func() time.Time { mu.Lock(); defer mu.Unlock(); return at }
+	moveTo := func(d time.Duration) { mu.Lock(); at = at.Add(d); mu.Unlock() }
 	waiter := func() *stepBounds {
-		return &stepBounds{bounds: func() jetstream.KeyValue { return kv }, now: func() time.Time { return at }, logf: t.Logf}
+		return &stepBounds{bounds: func() jetstream.KeyValue { return kv }, js: b.js, now: clock, logf: t.Logf}
 	}
-	const key = "stand.mwaskaaaaaaaaa"
-	var wg sync.WaitGroup
-	errs := make(chan error, 2)
-	for range 2 {
-		wg.Add(1)
-		go func() { defer wg.Done(); errs <- waiter().Arm(b.ctx, key, "the step stood still", time.Minute) }()
-	}
-	wg.Wait()
-	close(errs)
-	for err := range errs {
-		if err != nil {
-			t.Errorf("a waiter of the ask did not take the bound as armed: %v", err)
+	const key, bound = "stand.mwaskaaaaaaaaa", time.Minute
+	side := func(what string) {
+		t.Helper()
+		var wg sync.WaitGroup
+		errs := make(chan error, 2)
+		for range 2 {
+			wg.Add(1)
+			go func() { defer wg.Done(); errs <- waiter().Arm(b.ctx, key, "the step stood still", bound) }()
+		}
+		wg.Wait()
+		close(errs)
+		for err := range errs {
+			if err != nil {
+				t.Errorf("%s: a waiter of the ask did not take the bound as armed: %v", what, err)
+			}
 		}
 	}
-	first, err := kv.Get(b.ctx, key)
+	read := func() (jetstream.KeyValueEntry, versions.Bound) {
+		t.Helper()
+		e, err := kv.Get(b.ctx, key)
+		if err != nil {
+			t.Fatal(err)
+		}
+		var v versions.Bound
+		if err := json.Unmarshal(e.Value(), &v); err != nil {
+			t.Fatal(err)
+		}
+		return e, v
+	}
+	side("the first arming")
+	first, v := read()
+	if !v.Ends.Equal(clock().Add(bound)) {
+		t.Errorf("the bound ends %s", v.Ends)
+	}
+	if err := waiter().Arm(b.ctx, key, "the step stood still", bound); err != nil {
+		t.Fatal(err)
+	}
+	if again, _ := read(); again.Revision() != first.Revision() {
+		t.Errorf("the same movement armed again wrote the key: revision %d, was %d", again.Revision(), first.Revision())
+	}
+	last := first
+	for move := range 2 {
+		moveTo(bound + 10*time.Second) // past the bound before
+		side("a movement")
+		e, v := read()
+		if e.Revision() != last.Revision()+1 || !v.Ends.Equal(clock().Add(bound)) || e.Operation() != jetstream.KeyValuePut {
+			t.Fatalf("movement %d: revision %d (was %d), ends %s, %s: the bound does not count from the step's movement", move+1, e.Revision(), last.Revision(), v.Ends, e.Operation())
+		}
+		msg, err := b.js.Stream(b.ctx, "KV_"+versions.BoundsBucket)
+		if err != nil {
+			t.Fatal(err)
+		}
+		raw, err := msg.GetLastMsgForSubject(b.ctx, "$KV."+versions.BoundsBucket+"."+key)
+		if err != nil || raw.Header.Get("Nats-TTL") == "" {
+			t.Fatalf("movement %d: the key written again keeps no time to live (%v, %v)", move+1, raw, err)
+		}
+		last = e
+	}
+	// Never deleted: the key's history holds no delete marker.
+	w, err := kv.WatchAll(b.ctx, jetstream.IncludeHistory())
 	if err != nil {
 		t.Fatal(err)
 	}
-	at = at.Add(30 * time.Second)
-	if err := waiter().Arm(b.ctx, key, "the step stood still", time.Minute); err != nil {
-		t.Fatalf("a later waiter: %v", err)
-	}
-	again, err := kv.Get(b.ctx, key)
-	if err != nil || again.Revision() != first.Revision() || string(again.Value()) != string(first.Value()) {
-		t.Errorf("the bound was made again: revision %d (was %d), %v", again.Revision(), first.Revision(), err)
+	defer w.Stop()
+	for e := range w.Updates() {
+		if e == nil {
+			break
+		}
+		if e.Key() == key && e.Operation() != jetstream.KeyValuePut {
+			t.Errorf("the bound key was %s at revision %d", e.Operation(), e.Revision())
+		}
 	}
 }