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) + } + } +}