From dc024ae5527d41c04ab721dbc43eaf1242787c73 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 23:33:20 +0100 Subject: [PATCH] BL17e: the read of an ask's steps is a kept call: a fault ends no key, the watch is taken up again from its last revision, past five minutes unanswered it is call-unanswered on a bound key MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit JetStreamAsks.WatchAsk opens the watch as a kept call and takes a closed watch up after the last revision heard; the first fault arms a key of the state bounds with its own time to live (StepsBoundKey), whose end while the read is unanswered raises call-unanswered about the ask, and the answer ends it. Builds: design 47 G §568, A-77 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../mesh-delivery-builder/buildrun/asks.go | 108 ++++++++++++++---- .../buildrun/steps_resume_test.go | 88 ++++++++++++++ .../cmd/mesh-delivery-builder/main.go | 10 +- .../cmd/mesh-delivery-builder/stepsread.go | 95 +++++++++++++++ .../mesh-delivery-builder/stepsread_test.go | 43 +++++++ .../internal/versions/publish.go | 16 +++ 6 files changed, 336 insertions(+), 24 deletions(-) create mode 100644 modules/mesh-delivery-builder/buildrun/steps_resume_test.go create mode 100644 modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stepsread.go create mode 100644 modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stepsread_test.go diff --git a/modules/mesh-delivery-builder/buildrun/asks.go b/modules/mesh-delivery-builder/buildrun/asks.go index 7db40ad7..85efdecf 100644 --- a/modules/mesh-delivery-builder/buildrun/asks.go +++ b/modules/mesh-delivery-builder/buildrun/asks.go @@ -299,6 +299,14 @@ func (k Asker) Passed(step buildstep.Step, err error) (buildstep.Step, error) { // of `*.` in the build module's steps bucket. type JetStreamAsks struct { JS jetstream.JetStream + // Unanswered is told each time the read of an ask's steps meets a transport fault it keeps waiting out (design 47 G §568, ADR + // 0668 §3): the builder arms the call's bound, a key with its own time to live (A-77), and raises call-unanswered when it + // ends unanswered. Answered is told when the read is answered again. Both may be nil. + Unanswered func(module, askID string, fault error) + Answered func(module, askID string) + // OpenWatch opens the watch of keys filtered by pattern in the build module's steps state, from the revision after `after` + // when after is not 0; nil is the bus's. + OpenWatch func(ctx context.Context, module, pattern string, after uint64) (jetstream.KeyWatcher, error) } // PublishAsk is AskPublisher's. @@ -309,46 +317,100 @@ func (s JetStreamAsks) PublishAsk(ctx context.Context, subject, id string, data return err } -// WatchAsk is AskWaiter's: the watch hands over what the keys hold now, so a step that ended before it began is read at -// once, and every change after. -func (s JetStreamAsks) WatchAsk(ctx context.Context, module, askID string) (<-chan buildstep.Step, error) { - pattern, err := buildstep.AskStepWatch(askID) - if err != nil { - return nil, err +// openWatch opens the watch of pattern in the steps state of module, from the revision after `after` when it is not 0. +func (s JetStreamAsks) openWatch(ctx context.Context, module, pattern string, after uint64) (jetstream.KeyWatcher, error) { + if s.OpenWatch != nil { + return s.OpenWatch(ctx, module, pattern, after) } kv, err := s.JS.KeyValue(ctx, StepsBucket(module)) if err != nil { return nil, fmt.Errorf("the bucket %s: %w", StepsBucket(module), err) } - w, err := kv.Watch(ctx, pattern) + if after > 0 { + return kv.Watch(ctx, pattern, jetstream.ResumeFromRevision(after)) + } + return kv.Watch(ctx, pattern) +} + +// transportFault says err is a transport fault of ADR 0664 §1's classes or the read's own deadline: nothing was answered. +func transportFault(err error) bool { + return busretry.Undelivered(err) || busretry.TimedOut(err) +} + +// openKept is a kept call: the watch is opened, and opened again while the bus answers nothing, until it is or ctx ends +// (design 47 G §568: a transport fault of the builder's own read ends no build key). Each fault past the call's own asking again +// is told to Unanswered; the answer, to Answered. An error that is no transport fault is the caller's. +func (s JetStreamAsks) openKept(ctx context.Context, module, askID, pattern string, after uint64) (jetstream.KeyWatcher, error) { + for { + w, err := busretry.DoRepeatable("opening the watch of the ask "+askID+"'s step on "+module, func() (jetstream.KeyWatcher, error) { + return s.openWatch(ctx, module, pattern, after) + }, nil) + if err == nil { + if s.Answered != nil { + s.Answered(module, askID) + } + return w, nil + } + if ctx.Err() != nil { + return nil, ctx.Err() + } + if !transportFault(err) { + return nil, err + } + if s.Unanswered != nil { + s.Unanswered(module, askID, err) + } + } +} + +// WatchAsk is AskWaiter's: the watch hands over what the keys hold now, so a step that ended before it began is read at +// once, and every change after. A watch that cannot be opened for a transport fault is opened again once the bus answers, and one +// that closes is taken up again from the revision after the last it handed over (design 47 G §568), so a fault fails no key. +func (s JetStreamAsks) WatchAsk(ctx context.Context, module, askID string) (<-chan buildstep.Step, error) { + pattern, err := buildstep.AskStepWatch(askID) + if err != nil { + return nil, err + } + w, err := s.openKept(ctx, module, askID, pattern, 0) if err != nil { return nil, err } out := make(chan buildstep.Step) go func() { defer close(out) - defer w.Stop() + var last uint64 for { - select { - case <-ctx.Done(): - return - case e, ok := <-w.Updates(): - if !ok { - return - } - if e == nil || e.Operation() != jetstream.KeyValuePut { - continue - } - step, err := buildstep.DecodeStep(e.Value()) - if err != nil { - continue // not a step (a node's pause key never matches an ask id; anything else is not this ask's) - } + closed := false + for !closed { select { - case out <- step: case <-ctx.Done(): + _ = w.Stop() return + case e, ok := <-w.Updates(): + if !ok { + closed = true + break + } + if e == nil || e.Operation() != jetstream.KeyValuePut { + continue + } + last = max(last, e.Revision()) + step, err := buildstep.DecodeStep(e.Value()) + if err != nil { + continue // not a step (a node's pause key never matches an ask id; anything else is not this ask's) + } + select { + case out <- step: + case <-ctx.Done(): + _ = w.Stop() + return + } } } + _ = w.Stop() + if w, err = s.openKept(ctx, module, askID, pattern, last+1); err != nil { + return + } } }() return out, nil diff --git a/modules/mesh-delivery-builder/buildrun/steps_resume_test.go b/modules/mesh-delivery-builder/buildrun/steps_resume_test.go new file mode 100644 index 00000000..cb8c4203 --- /dev/null +++ b/modules/mesh-delivery-builder/buildrun/steps_resume_test.go @@ -0,0 +1,88 @@ +package buildrun + +import ( + "context" + "sync" + "testing" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-delivery/sdk/buildstep" + "git.novox.be/novox/mesh-sdk/go/busretry" +) + +type fakeEntry struct { + jetstream.KeyValueEntry + rev uint64 + value []byte +} + +func (e fakeEntry) Revision() uint64 { return e.rev } +func (e fakeEntry) Value() []byte { return e.value } +func (e fakeEntry) Operation() jetstream.KeyValueOp { return jetstream.KeyValuePut } + +type fakeWatcher struct { + jetstream.KeyWatcher + updates chan jetstream.KeyValueEntry +} + +func (w fakeWatcher) Updates() <-chan jetstream.KeyValueEntry { return w.updates } +func (w fakeWatcher) Stop() error { return nil } + +// Design 47 G §568: the read of an ask's steps is a kept call. A transport fault opening the watch ends no key (the wait is +// not failed): it is opened again once the bus answers, said unanswered meanwhile and answered after; and a watch that closes +// is taken up again from the revision after the last it handed over, so the step that ended meanwhile is heard. +func TestTheWatchOfAStepResumesAfterAFault(t *testing.T) { + t.Cleanup(busretry.Waits(20*time.Millisecond, time.Millisecond, 2*time.Millisecond)) + passed, _ := buildstep.EncodeStep(buildstep.Step{ID: "mvl0frv18d1958", Tool: "build", Node: "ace", State: buildstep.StepPassed}) + running, _ := buildstep.EncodeStep(buildstep.Step{ID: "mvl0frv18d1958", Tool: "build", Node: "ace", State: buildstep.StepRunning}) + var mu sync.Mutex + var opens []uint64 + var unanswered, answered int + asks := JetStreamAsks{ + Unanswered: func(string, string, error) { mu.Lock(); unanswered++; mu.Unlock() }, + Answered: func(string, string) { mu.Lock(); answered++; mu.Unlock() }, + OpenWatch: func(_ context.Context, _, _ string, after uint64) (jetstream.KeyWatcher, error) { + mu.Lock() + defer mu.Unlock() + opens = append(opens, after) + ch := make(chan jetstream.KeyValueEntry, 2) + // The bus answers nothing past the kept call's own asking again (its bound), so the fault is told and the watch tried again. + switch n := len(opens); { + case n <= 40: + return nil, nats.ErrNoResponders + case n == 41: + ch <- fakeEntry{rev: 5, value: running} + close(ch) // the watch closes under the wait + default: + ch <- fakeEntry{rev: 6, value: passed} + } + return fakeWatcher{updates: ch}, nil + }} + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + steps, err := asks.WatchAsk(ctx, "mesh-build-go", "mvl0frv18d1958") + if err != nil { + t.Fatalf("a transport fault opening the watch failed the wait: %v", err) + } + var states []string + for s := range steps { + states = append(states, s.State) + if buildstep.Ended(s) { + break + } + } + mu.Lock() + defer mu.Unlock() + if len(states) != 2 || states[0] != buildstep.StepRunning || states[1] != buildstep.StepPassed { + t.Fatalf("steps heard = %v, want running then passed", states) + } + if len(opens) != 42 || opens[41] != 6 { + t.Errorf("opens after revisions %v, want the last taken up again after the last revision heard (6)", opens) + } + if unanswered == 0 || answered != 2 { + t.Errorf("unanswered %d, answered %d, want the fault said and each answer said", unanswered, answered) + } +} 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 0b856d1a..42179b45 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -140,7 +140,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro tools := &buildrun.ToolchainSet{} // Each build module's part of a build is an ask on its build seat's stream, pulled by a build node with a free slot // (novox/hq ADR 0525): the builder chooses no node, so it reads neither presence nor have and holds no capacity. - asks := buildrun.JetStreamAsks{JS: js} + // The read of an ask's steps is a kept call (design 47 G §568): a fault ends no key, and past five minutes unanswered it is a + // finding; its bound is a key of the state bounds, bound once the bus holds it (set below, with the findings). + stepReads := &stepsRead{bounds: func() jetstream.KeyValue { return nil }, bound: meshconditionsclient.CallUnansweredAfter, now: time.Now, logf: logger.Printf} + asks := buildrun.JetStreamAsks{JS: js, Unanswered: stepReads.unanswered, Answered: stepReads.answered} // The asks the build seats gave up on (issue 734): heard from DEAD_LETTERS, each ends the wait for its step at once, // and the lane asks its part again (Builder.waitOnce) instead of waiting for the check bound. // Heard in the background, and again at a reconnect or an ask while it is not; whether it is heard is kept in the @@ -181,6 +184,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // A publish that ends badly is a finding (design 51 *Health and conditions*): a holder's refused, a version no holder // answered within its bound (a key of the state bounds with its own time to live), a seat's outcome that does not read. pubBoundsKV := process.AwaitBucket(ctx, js, conn, versions.BoundsBucket, "a publish no holder answers, and a node holding the old form seven days, are not raised", logger.Printf) + 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 b := &build.Builder{ @@ -443,6 +447,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro unput.expired(ctx, e) return } + if _, _, ok := versions.StepsBoundOf(e.Key()); ok { + stepReads.expired(ctx, e) + return + } if node, ok := enginereads.BoundOf(e.Key()); ok { if e.Operation() == jetstream.KeyValuePurge { // its five minutes ended reads.expiredCall(node) diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stepsread.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stepsread.go new file mode 100644 index 00000000..378a8db2 --- /dev/null +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stepsread.go @@ -0,0 +1,95 @@ +package main + +import ( + "context" + "errors" + "fmt" + "sync" + "time" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" + + "mesh-delivery-builder/internal/versions" +) + +// **The builder's read of an ask's steps is a kept call, unanswered past five minutes a finding** (design 47 G §568, A-77, ADR +// 0668 §3): a transport fault opening or taking up the watch of `steps` ends no build key; the read is made again once the bus +// answers. The first fault arms a key of the state bounds (StepsBoundKey) with its own time to live; the bus's marker of its end, +// heard on the watch of the state, raises call-unanswered about the ask while the read is still unanswered; the answer deletes +// the key and clears the finding. No timer. +type stepsRead struct { + bounds func() jetstream.KeyValue + findings *findings + bound time.Duration + now func() time.Time + logf func(string, ...any) + + mu sync.Mutex + open map[string]string // the asks whose read is unanswered, by ask id: the last fault +} + +// unanswered is told each fault of the read of the ask's steps on module: the first arms the bound. +func (s *stepsRead) unanswered(module, askID string, fault error) { + s.mu.Lock() + if s.open == nil { + s.open = map[string]string{} + } + _, armed := s.open[askID] + s.open[askID] = fault.Error() + s.mu.Unlock() + if armed { + return + } + ctx := context.Background() + var err error + if kv := s.bounds(); kv == nil { + err = fmt.Errorf("%s is not on the bus yet", versions.BoundsBucket) + } else if _, err = kv.Create(ctx, versions.StepsBoundKey(module, askID), versions.EncodeBound("the read of the steps of the ask "+askID+" on "+module, s.now().Add(s.bound)), jetstream.KeyTTL(s.bound)); errors.Is(err, jetstream.ErrKeyExists) { + err = nil + } + if err != nil { + s.logf("the bound of the read of the steps of the ask %s is not written, so it is raised now: %v", askID, err) + s.raise(ctx, module, askID, fault.Error(), s.now()) + } +} + +// answered ends the read's bound and clears its finding. +func (s *stepsRead) answered(module, askID string) { + s.mu.Lock() + _, was := s.open[askID] + delete(s.open, askID) + s.mu.Unlock() + if !was { + return + } + ctx := context.Background() + if kv := s.bounds(); kv != nil { + if err := kv.Delete(ctx, versions.StepsBoundKey(module, askID)); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) { + s.logf("the bound of the read of the steps of the ask %s is not deleted: %v", askID, err) + } + } + s.findings.clear(ctx, meshconditionsclient.KindCallUnanswered, meshconditionsclient.ScopeCall, askID) +} + +// expired is told each change of the state bounds: the end of a steps bound raises call-unanswered while its read is unanswered. +func (s *stepsRead) expired(ctx context.Context, e jetstream.KeyValueEntry) { + module, askID, ok := versions.StepsBoundOf(e.Key()) + if !ok || e.Operation() != jetstream.KeyValuePurge { + return + } + s.mu.Lock() + fault, open := s.open[askID] + s.mu.Unlock() + if !open { + return + } + s.raise(ctx, module, askID, fault, s.now().Add(-s.bound)) +} + +func (s *stepsRead) raise(ctx context.Context, module, askID, fault string, since time.Time) { + o := callUnanswered(askID, "the read of the steps of "+module, "", fault, "builder."+meshconditionsclient.KindCallUnanswered, since) + o.needs = "the bus answers the read again by itself; builder_stuck shows the key waiting on the ask." + s.findings.raise(ctx, o) +} diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stepsread_test.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stepsread_test.go new file mode 100644 index 00000000..d47b9c4d --- /dev/null +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stepsread_test.go @@ -0,0 +1,43 @@ +package main + +import ( + "context" + "errors" + "strings" + "testing" + "time" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" + + "mesh-delivery-builder/internal/versions" +) + +// Design 47 G §568, A-77: the first fault of the read of an ask's steps arms a bound key with its own time to live; its end +// while the read is unanswered raises call-unanswered about the ask; the answer deletes the bound and clears the finding. +func TestTheStepsReadIsCallUnansweredAfterItsBound(t *testing.T) { + st := &memFindings{vals: map[string][]byte{}} + f := &findings{w: st, logf: t.Logf, now: func() time.Time { return time.Unix(0, 0) }} + marks := &memBounds{keys: map[string]bool{}} + s := &stepsRead{bounds: func() jetstream.KeyValue { return marks }, findings: f, bound: meshconditionsclient.CallUnansweredAfter, + now: func() time.Time { return time.Unix(600, 0) }, logf: t.Logf} + ask := "mvl0frv18d1958" + callKey := meshconditionsclient.Key(meshconditionsclient.ScopeCall, ask, meshconditionsclient.KindCallUnanswered) + s.unanswered("mesh-build-go", ask, errors.New("nats: no responders available for request")) + if !marks.keys[versions.StepsBoundKey("mesh-build-go", ask)] || st.vals[callKey] != nil { + t.Fatalf("a first fault: bound %v, finding %v; want the bound armed, no finding yet", marks.keys, st.vals[callKey] != nil) + } + s.expired(context.Background(), purged{key: versions.StepsBoundKey("mesh-build-go", ask)}) + if got, err := decodeFinding(st.vals[callKey]); err != nil || got.Kind != meshconditionsclient.KindCallUnanswered || !strings.Contains(got.Summary, "the steps of mesh-build-go") || !strings.Contains(got.Summary, ask) { + t.Fatalf("the bound's end: the finding = %+v, %v", got, err) + } + s.answered("mesh-build-go", ask) + if marks.keys[versions.StepsBoundKey("mesh-build-go", ask)] || st.vals[callKey] != nil { + t.Errorf("the read answered: bound %v, finding left %v", marks.keys, st.vals[callKey] != nil) + } + s.expired(context.Background(), purged{key: versions.StepsBoundKey("mesh-build-go", ask)}) + if st.vals[callKey] != nil { + t.Error("a bound's end after the answer raised call-unanswered") + } +} diff --git a/modules/mesh-delivery-builder/internal/versions/publish.go b/modules/mesh-delivery-builder/internal/versions/publish.go index 7b401b8a..43457b3b 100644 --- a/modules/mesh-delivery-builder/internal/versions/publish.go +++ b/modules/mesh-delivery-builder/internal/versions/publish.go @@ -37,6 +37,7 @@ const ( publishBoundPrefix = "publish." oldFormBoundPrefix = "old-form." unputBoundPrefix = "unput." + stepsBoundPrefix = "steps." ) // UnputBoundKey is the bound of a build key a look left building because its end could not be written (issue 1966, @@ -44,6 +45,21 @@ const ( // finding call-unanswered about the key. func UnputBoundKey(buildKey string) string { return unputBoundPrefix + buildKey } +// StepsBoundKey is the bound of the builder's read of an ask's steps while the bus does not answer it (design 47 G §568, A-77): +// `steps..`. Its end, the bus's marker of its time to live, while the read is unanswered, is the finding +// call-unanswered about the ask. +func StepsBoundKey(module, askID string) string { return stepsBoundPrefix + module + "." + askID } + +// StepsBoundOf is the build module and ask a steps bound names, and whether key is one. +func StepsBoundOf(key string) (module, askID string, ok bool) { + rest, ok := strings.CutPrefix(key, stepsBoundPrefix) + if !ok { + return "", "", false + } + module, askID, ok = strings.Cut(rest, ".") + return module, askID, ok && module != "" && askID != "" +} + // UnputBoundOf is the build key an unput bound names, and whether key is one. func UnputBoundOf(key string) (string, bool) { return strings.CutPrefix(key, unputBoundPrefix) }