From eaa578a4b9198a12a9fd9992bf8a483ea377da76 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 01:07:35 +0100 Subject: [PATCH] Builder: an ask's step bounds are keys with their own time to live buildrun.Asker takes optional Bounds and Expiry: the stand-still bound and the cap of three bounds are the keys stand. and cap. of the builder's state bounds, whose expiry the builder's watch of that state tells the Expiry, and Wait ends then; no timer of the process. With no Bounds (the checker's Asker, which the lead's A-118 leaves as it is) Wait keeps its timers. Builds: design 47 G A-77, A-101, A-118 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../mesh-delivery-builder/buildrun/asks.go | 139 +++++++++++++++--- .../buildrun/asks_test.go | 58 ++++++++ .../cmd/mesh-delivery-builder/main.go | 12 +- .../cmd/mesh-delivery-builder/stepbounds.go | 42 ++++++ 4 files changed, 229 insertions(+), 22 deletions(-) create mode 100644 modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stepbounds.go diff --git a/modules/mesh-delivery-builder/buildrun/asks.go b/modules/mesh-delivery-builder/buildrun/asks.go index ed688d09..879cf595 100644 --- a/modules/mesh-delivery-builder/buildrun/asks.go +++ b/modules/mesh-delivery-builder/buildrun/asks.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "log" + "strings" "sync" "time" @@ -45,10 +46,79 @@ type Asker struct { // again that the bus did not give within its bound, or a wait ended with a cause its caller names (a superseded // fingerprint). Never a wait ended by a shutdown (a plain cancel): the ask is then taken over by the next run. Dropper AskDropper + // Bounds and Expiry keep the step bounds as keys with their own time to live (design 47 G A-77, A-101, A-118): the stand-still + // bound as StandBoundKey and the cap of three Bounds as CapBoundKey in the asking part's own state, whose expiry the part's watch + // of that state tells Expiry (Expiry.Ended), and Wait then ends. Nil Bounds keeps the bounds as timers of the process, as the + // checker has them. + Bounds StepBounds + Expiry *Expiry // Logf says a drop that failed (the log's Printf when unset): the ask then stays on its stream, and it is said. Logf func(format string, args ...any) } +// StepBounds keeps one bound as a key with its own time to live: Arm writes key afresh to end after bound, End deletes it. +type StepBounds interface { + Arm(ctx context.Context, key, why string, bound time.Duration) error + End(ctx context.Context, key string) +} + +// The keys of an ask's step bounds in the asking part's state: `stand.` (the step stood still) and `cap.` (a step +// that kept moving never ended). +const ( + standBoundPrefix = "stand." + capBoundPrefix = "cap." +) + +// StandBoundKey is the key of the stand-still bound of an ask's step. +func StandBoundKey(askID string) string { return standBoundPrefix + askID } + +// CapBoundKey is the key of the cap on an ask's step. +func CapBoundKey(askID string) string { return capBoundPrefix + askID } + +// StepBoundOf says whether key is a step bound's, as the part's watch of its bounds state reads it. +func StepBoundOf(key string) bool { + return strings.HasPrefix(key, standBoundPrefix) || strings.HasPrefix(key, capBoundPrefix) +} + +// Expiry is where the part's watch of its bounds state tells the end of a bound key (the bus's marker of its time to live), and the +// waits for those keys hear it. Safe for concurrent use; the zero value is ready. +type Expiry struct { + mu sync.Mutex + waits map[string]chan struct{} +} + +// Ended tells the end of key to the wait that holds it, and says whether one did. +func (e *Expiry) Ended(key string) bool { + e.mu.Lock() + c, ok := e.waits[key] + e.mu.Unlock() + if ok { + select { + case c <- struct{}{}: + default: + } + } + return ok +} + +// hold is the channel key's end is told on, and its release. +func (e *Expiry) hold(key string) (<-chan struct{}, func()) { + e.mu.Lock() + defer e.mu.Unlock() + if e.waits == nil { + e.waits = map[string]chan struct{}{} + } + c := make(chan struct{}, 1) + e.waits[key] = c + return c, func() { + e.mu.Lock() + if e.waits[key] == c { + delete(e.waits, key) + } + e.mu.Unlock() + } +} + // Withdrawn is the cause a caller gives the context of an ask's wait to withdraw the ask (ADR 0664 §4): a build superseded by a // newer fingerprint, a candidate taken out, an ask asked again by hand. Asker.Wait drops the ask on this cause alone: a // plain cancel (a shutdown) and any other cause (a bound passing, a check of whether the ask is gone) leave it on its stream @@ -155,16 +225,51 @@ func (k Asker) Wait(ctx context.Context, a buildstep.StepAsk) (buildstep.Step, e if err != nil { return buildstep.Step{}, err } - var bound, outerCap <-chan time.Time - var timer, capTimer *time.Timer + // The two bounds: keys with their own time to live whose expiry is told by Expiry, or, with no Bounds, timers. + var bound, outerCap <-chan struct{} + keyed := k.Bounds != nil && k.Expiry != nil + var timers [2]*time.Timer + standKey, capKey := StandBoundKey(a.ID), CapBoundKey(a.ID) + if keyed { + var releaseStand, releaseCap func() + bound, releaseStand = k.Expiry.hold(standKey) + outerCap, releaseCap = k.Expiry.hold(capKey) + defer func() { + releaseStand() + releaseCap() + k.Bounds.End(context.WithoutCancel(ctx), standKey) + k.Bounds.End(context.WithoutCancel(ctx), capKey) + }() + } defer func() { - if timer != nil { - timer.Stop() - } - if capTimer != nil { - capTimer.Stop() + for _, t := range timers { + if t != nil { + t.Stop() + } } }() + // arm ends the bound i (0 stand-still, 1 cap) after d: a key's time to live, or a timer. + arm := func(i int, d time.Duration) error { + d = max(d, time.Millisecond) + if keyed { + key, why := standKey, "the step of the ask "+a.ID+" stood still" + if i == 1 { + key, why = capKey, "the step of the ask "+a.ID+" never ended" + } + return k.Bounds.Arm(ctx, key, why, d) + } + if timers[i] != nil { + timers[i].Stop() + } + c := make(chan struct{}, 1) + timers[i] = time.AfterFunc(d, func() { c <- struct{}{} }) + if i == 0 { + bound = c + } else { + outerCap = c + } + return nil + } running, since, phase, phaseAt := "", time.Time{}, "", time.Time{} var agains []string // the node of each run that ended on a transport fault, in order (ADR 0664 §3) standStill := func() error { @@ -203,11 +308,9 @@ func (k Asker) Wait(ctx context.Context, a buildstep.StepAsk) (buildstep.Step, e agains = append(agains, step.Node) running, since, phase, phaseAt = step.Node, step.Started, "given again after a transport fault", step.AgainAt if k.Bound > 0 { - if timer != nil { - timer.Stop() + if err := arm(0, time.Until(step.AgainAt.Add(k.Bound))); err != nil { + return buildstep.Step{}, fmt.Errorf("the bound of the ask %s's step cannot be kept: %w", a.ID, err) } - timer = time.NewTimer(max(time.Until(step.AgainAt.Add(k.Bound)), 0)) - bound = timer.C } continue } @@ -240,21 +343,17 @@ func (k Asker) Wait(ctx context.Context, a buildstep.StepAsk) (buildstep.Step, e if !last.IsZero() { left = time.Until(last.Add(k.Bound)) } - if timer != nil { - timer.Stop() + if err := arm(0, left); err != nil { + return buildstep.Step{}, fmt.Errorf("the bound of the ask %s's step cannot be kept: %w", a.ID, err) } - timer = time.NewTimer(max(left, 0)) - bound = timer.C if newRun { - if capTimer != nil { - capTimer.Stop() - } outer := 3 * k.Bound if !step.Started.IsZero() { outer = time.Until(step.Started.Add(3 * k.Bound)) } - capTimer = time.NewTimer(max(outer, 0)) - outerCap = capTimer.C + if err := arm(1, outer); err != nil { + return buildstep.Step{}, fmt.Errorf("the bound of the ask %s's step cannot be kept: %w", a.ID, err) + } } } } diff --git a/modules/mesh-delivery-builder/buildrun/asks_test.go b/modules/mesh-delivery-builder/buildrun/asks_test.go index cafe4c55..379a0579 100644 --- a/modules/mesh-delivery-builder/buildrun/asks_test.go +++ b/modules/mesh-delivery-builder/buildrun/asks_test.go @@ -397,3 +397,61 @@ func (p *subjectsOf) PublishAsk(ctx context.Context, subject, id string, data [] p.seen = append(p.seen, subject) return p.inner.PublishAsk(ctx, subject, id, data) } + +// keyBounds is a StepBounds that keeps the armed keys and the ones ended. +type keyBounds struct { + mu sync.Mutex + armed, ends []string +} + +func (b *keyBounds) Arm(_ context.Context, key, _ string, _ time.Duration) error { + b.mu.Lock() + defer b.mu.Unlock() + b.armed = append(b.armed, key) + return nil +} + +func (b *keyBounds) End(_ context.Context, key string) { + b.mu.Lock() + defer b.mu.Unlock() + b.ends = append(b.ends, key) +} + +// Design 47 G A-77, A-101: a step's bounds are keys with their own time to live; the end of the stand-still key, told by the part's +// watch of its state, ends the wait, and no timer of the process does. +func TestAStepsBoundsAreKeysWhoseExpiryEndsTheWait(t *testing.T) { + h := &FakeAsks{} + id := "mvl0frv18d1990" + h.Write(buildstep.Step{ID: id, Node: "ace", State: buildstep.StepRunning, Started: time.Now(), Phase: "fetching", PhaseAt: time.Now()}) + bounds, expiry := &keyBounds{}, &Expiry{} + k := Asker{Seats: testSeats(), Publisher: h, Waiter: h, Bound: time.Hour, Bounds: bounds, Expiry: expiry} + done := make(chan error, 1) + go func() { _, err := k.Wait(context.Background(), goAsk(id)); done <- err }() + for deadline := time.Now().Add(5 * time.Second); ; time.Sleep(time.Millisecond) { + bounds.mu.Lock() + n := len(bounds.armed) + bounds.mu.Unlock() + if n >= 2 { + break + } + if time.Now().After(deadline) { + t.Fatalf("armed %v, want the stand-still and the cap keys", bounds.armed) + } + } + select { + case err := <-done: + t.Fatalf("the wait ended with no expiry told: %v", err) + case <-time.After(20 * time.Millisecond): // nothing moves it: the wait is on the key, not a timer + } + if !expiry.Ended(StandBoundKey(id)) { + t.Fatal("no wait held the stand-still key") + } + if err := <-done; err == nil || !strings.Contains(err.Error(), "stood still") { + t.Errorf("the wait ended with %v, want its step stood still", err) + } + bounds.mu.Lock() + defer bounds.mu.Unlock() + if len(bounds.ends) != 2 || bounds.armed[0] != StandBoundKey(id) { + t.Errorf("armed %v, ended %v; want both keys ended with the wait", bounds.armed, bounds.ends) + } +} 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 1bd7f2e7..40948fb0 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -176,6 +176,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro stepReads.bounds, stepReads.findings = func() jetstream.KeyValue { return pubBoundsKV.KV() }, found pubs := &publishes{bounds: pubBoundsKV.KV, versions: numberer.Versions, findings: found, bound: versions.PublishBound, logf: logger.Printf} numberer.Publishing = pubs.arm + stepExpiry := &buildrun.Expiry{} var b *build.Builder seatsChanged := func(ctx context.Context) { heard.Watch(seats.Modules()) @@ -186,8 +187,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro b = &build.Builder{ 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}}, - Held: heldOnTheBus{js: js, 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}, + 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}, CheckAfter: bound, @@ -463,6 +465,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro unput.expired(ctx, e) return } + if buildrun.StepBoundOf(e.Key()) { + if e.Operation() == jetstream.KeyValuePurge { // its time to live ended: the ask's wait ends + stepExpiry.Ended(e.Key()) + } + return + } if _, _, ok := versions.StepsBoundOf(e.Key()); ok { stepReads.expired(ctx, e) return diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stepbounds.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stepbounds.go new file mode 100644 index 00000000..30c58647 --- /dev/null +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stepbounds.go @@ -0,0 +1,42 @@ +package main + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/nats-io/nats.go/jetstream" + + "mesh-delivery-builder/internal/versions" +) + +// stepBounds keeps the bounds of an ask's step as keys of the state bounds with their own time to live (design 47 G A-77, A-101, +// 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) +} + +// Arm is buildrun.StepBounds' Arm: the key is written afresh, so a step that moved has its bound counted from its movement. +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) + } + if err := kv.Delete(ctx, key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) { + return err + } + _, err := kv.Create(ctx, key, versions.EncodeBound(why, s.now().Add(bound)), jetstream.KeyTTL(bound)) + return err +} + +// End is buildrun.StepBounds' End. +func (s *stepBounds) End(ctx context.Context, key string) { + if kv := s.bounds(); kv != nil { + if err := kv.Delete(ctx, key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) { + s.logf("the step bound %s is not deleted: %v", key, err) + } + } +}