From 339964920b4753f846ed6ce5e6b3c4f7a5df8ccf Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 13:16:22 +0100 Subject: [PATCH] A-282 corrected (join 11): a moved step's bound is written again by a revision-checked update, never a delete MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- .../cmd/mesh-delivery-builder/main.go | 2 +- .../cmd/mesh-delivery-builder/stepbounds.go | 57 +++++++++-- .../stepbounds_integration_test.go | 99 ++++++++++++++----- 3 files changed, 126 insertions(+), 32 deletions(-) diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go index afa9d083..9406b833 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -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}, diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stepbounds.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stepbounds.go index e5d87a5e..1ea37585 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stepbounds.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stepbounds.go @@ -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. diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stepbounds_integration_test.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stepbounds_integration_test.go index 9702da08..9aad7b9f 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stepbounds_integration_test.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stepbounds_integration_test.go @@ -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()) + } } }