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 d602f8e..f003aae 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -92,7 +92,16 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro hearing := buildrun.NewDeadLetterHearing(ctx, buildrun.JetStreamDeadLetters{JS: js}, delivery.AskLanguages, dead, logger.Printf, buildrun.SayHearing(ctx, hearingKV, logger.Printf)) defer hearing.Stop() - conn.SetReconnectHandler(func(*nats.Conn) { hearing.Again() }) + // Each ask's step is heard by its own watch and, at every step end of its build module and every reconnect, read + // (buildrun.HearEnds): a watch that goes deaf loses no end (the lost check of 2026-10-28). + var buildModules []string + for _, l := range delivery.AskLanguages { + if m, err := delivery.BuildModuleOf(l); err == nil { + buildModules = append(buildModules, m) + } + } + heard := buildrun.HearEnds(ctx, asks, asks, buildModules, logger.Printf) + conn.SetReconnectHandler(func(*nats.Conn) { hearing.Again(); heard.Nudge("") }) // The builder asks the controller, as its own account, whether the artifact store still holds a done key's blobs // (the seat verb held, novox/hq issue 1937) before it leaves the key done. calls, err := process.Connect(cred, nats.Name(cred.Node+"."+cred.Module+".calls"), nats.RetryOnFailedConnect(true)) @@ -102,7 +111,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro defer calls.Close() b := &build.Builder{ Claimer: build.Claimer{Store: kvStore{builds}, By: cred.Node + "." + cred.Module}, - Asks: buildrun.Asker{Publisher: asks, Waiter: buildrun.LostAskWaiter{Inner: asks, Dead: dead, Hearing: hearing}, Bound: bound}, + Asks: buildrun.Asker{Publisher: asks, Waiter: buildrun.LostAskWaiter{Inner: heard, Dead: dead, Hearing: hearing}, Bound: bound}, Held: heldOnTheBus{js}, // 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}, diff --git a/modules/mesh-delivery-builder/go.mod b/modules/mesh-delivery-builder/go.mod index ddfb7c0..a70a9fd 100644 --- a/modules/mesh-delivery-builder/go.mod +++ b/modules/mesh-delivery-builder/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-builder go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028174713-38dbe1d3a388 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028192930-6aeb40109529 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-builder/go.sum b/modules/mesh-delivery-builder/go.sum index 6245202..88f3a59 100644 --- a/modules/mesh-delivery-builder/go.sum +++ b/modules/mesh-delivery-builder/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028174713-38dbe1d3a388 h1:ycu/sNieTdLpD3bzhZOCXdzcoqXUOxPRmqWV019REpo= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028174713-38dbe1d3a388/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028192930-6aeb40109529 h1:Xx5mcDPWcRuPelUxdjvwkmVcfyGww0rRajA9JjFA/hk= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028192930-6aeb40109529/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-builder/internal/build/run_test.go b/modules/mesh-delivery-builder/internal/build/run_test.go index 481d925..6cbb41c 100644 --- a/modules/mesh-delivery-builder/internal/build/run_test.go +++ b/modules/mesh-delivery-builder/internal/build/run_test.go @@ -691,6 +691,7 @@ func TestABuildAskWaitingInItsStreamIsNotBounded(t *testing.T) { byHand bool always bool // Gone says every ask cannot be answered, not only the first dead bool // DEAD_LETTERS keeps the first ask: the seat gave it up (issue 734) + deaf bool // the ask's own watch hears nothing though its step ends (the lost check of 2026-10-28) puts int done bool asking int // the asking the event says, 0 when none is said @@ -739,6 +740,7 @@ func TestABuildAskWaitingInItsStreamIsNotBounded(t *testing.T) { {name: "given up by its stream: asked again under a new id and done from it", puts: 2, done: true, asking: 2}, {name: "asked again by hand: the same, with no check bound", byHand: true, puts: 2, done: true, asking: 2}, {name: "given up into DEAD_LETTERS: its wait ends failed with the words and it is asked again at once", dead: true, puts: 2, done: true, asking: 2}, + {name: "its step ended while its watch was deaf: read at its build module's next step end, done with no new ask", deaf: true, puts: 1, done: true}, {name: "never answerable: failed after its askings", askings: 2, always: true, puts: 2}, {name: "taken over at its last asking: failed, not asked again", askings: 3, always: true, step: "asked mesh-build-go again (asking 3): x", puts: 0}, } { @@ -829,7 +831,12 @@ func TestABuildAskWaitingInItsStreamIsNotBounded(t *testing.T) { if c.dead { r.b.Asks = buildrun.Asker{Publisher: r.asks, Waiter: buildrun.LostAskWaiter{Inner: r.asks, Dead: dead}, Bound: 2 * time.Second} } - if !c.byHand && !c.dead { + var heard *buildrun.HeardEnds + if c.deaf { + heard = &buildrun.HeardEnds{Inner: silentWatch{}, Read: r.asks} + r.b.Asks = buildrun.Asker{Publisher: r.asks, Waiter: heard, Bound: 2 * time.Second} + } + if !c.byHand && !c.dead && !c.deaf { gone.why = func(a delivery.StepAsk) string { if c.always || a.ID == first || first == "" { return "it waits first in line while a node took a later ask" @@ -851,7 +858,16 @@ func TestABuildAskWaitingInItsStreamIsNotBounded(t *testing.T) { } } } - if c.done { + if c.deaf { + r.asks.End(r.answer(waitForAsks(t, r, 1)[0])) + for deadline := time.Now().Add(2 * time.Second); heard.Waiting() == 0; time.Sleep(time.Millisecond) { + if time.Now().After(deadline) { + t.Fatal("no ask is waited on") + } + } + heard.Nudge(buildrun.GoBuildModule) // a step of mesh-build-go ended + } + if c.done && !c.deaf { asks := waitForAsks(t, r, c.puts) if first == "" && !c.byHand { first = asks[0].ID @@ -876,7 +892,7 @@ func TestABuildAskWaitingInItsStreamIsNotBounded(t *testing.T) { mu.Lock() defer mu.Unlock() // Note 62: the old message of an ask asked again because its stream gave it up is dropped from the stream. - if gone.mu.Lock(); c.done && !c.byHand && !c.dead && (len(gone.dropped) != 1 || gone.dropped[0] != first) { + if gone.mu.Lock(); c.done && !c.byHand && !c.dead && !c.deaf && (len(gone.dropped) != 1 || gone.dropped[0] != first) { t.Errorf("dropped %v from the stream, want the old ask %s", gone.dropped, first) } gone.mu.Unlock() @@ -1086,3 +1102,10 @@ func TestATakeOverThatCannotReadTheStreamWaitsForTheAsk(t *testing.T) { t.Fatalf("the key = %+v after %d asks, want done from the ask's step, none put", v, len(r.asks.Asks())) } } + +// silentWatch is a watch of an ask's step that stays open and hears nothing. +type silentWatch struct{} + +func (silentWatch) WatchAsk(context.Context, string, string) (<-chan delivery.Step, error) { + return make(chan delivery.Step), nil +} diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go index 1563dcb..6647277 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go @@ -145,9 +145,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro logger.Printf, buildrun.SayHearing(ctx, hearingKV, logger.Printf)) defer hearing.Stop() // Each ask's step is heard by its own watch and, at every step end of its build module and every reconnect, read - // (check.HeardEnds): a watch that goes deaf loses no end (the lost check of 2026-10-28). - heard := &check.HeardEnds{Inner: asks, Read: askSteps{js: js}, Logf: logger.Printf} - hearStepEnds(ctx, js, []string{buildrun.GoBuildModule, buildrun.NodeBuildModule}, heard, logger.Printf) + // (buildrun.HearEnds): a watch that goes deaf loses no end (the lost check of 2026-10-28). + heard := buildrun.HearEnds(ctx, asks, asks, []string{buildrun.GoBuildModule, buildrun.NodeBuildModule}, logger.Printf) conn.SetReconnectHandler(func(*nats.Conn) { hearing.Again(); heard.Nudge("") }) waiter := buildrun.LostAskWaiter{Inner: heard, Dead: dead, Hearing: hearing} ch := &check.Checker{ diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/stepends.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/stepends.go deleted file mode 100644 index 0193082..0000000 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/stepends.go +++ /dev/null @@ -1,83 +0,0 @@ -package main - -import ( - "context" - "fmt" - - "github.com/nats-io/nats.go/jetstream" - - "git.novox.be/novox/mesh-sdk/go/delivery" - "git.novox.be/novox/mesh-sdk/go/process" - - "mesh-delivery-checker/internal/check" -) - -// askSteps reads an ask's step keys now, from its build module's steps state (check.StepReader): a read, not a watch. -type askSteps struct{ js jetstream.JetStream } - -func (a askSteps) AskSteps(ctx context.Context, module, askID string) ([]delivery.Step, error) { - pattern, err := delivery.AskStepWatch(askID) - if err != nil { - return nil, err - } - kv, err := a.js.KeyValue(ctx, delivery.StepsBucket(module)) - if err != nil { - return nil, fmt.Errorf("the bucket %s: %w", delivery.StepsBucket(module), err) - } - lister, err := kv.ListKeysFiltered(ctx, pattern) - if err != nil { - return nil, err - } - var out []delivery.Step - for key := range lister.Keys() { - e, err := kv.Get(ctx, key) - if err != nil { - continue // deleted between the list and the read - } - if step, err := delivery.DecodeStep(e.Value()); err == nil { - out = append(out, step) - } - } - return out, nil -} - -// hearStepEnds nudges every ask of a build module the checker waits on each time a step of that module ends (any key of -// its steps state written not running): an ask whose own watch went deaf reads its end then (check.HeardEnds). One watch -// per build module, updates only; one that closes is opened again while it delivered something since it was opened, and -// said and left otherwise. No timer. -func hearStepEnds(ctx context.Context, js process.BucketOpener, modules []string, heard *check.HeardEnds, logf func(string, ...any)) { - for _, module := range modules { - kv := process.OptionalBucket(ctx, js, delivery.StepsBucket(module), "a check whose own watch of its step goes deaf is read only at a reconnect or by hand", logf) - if kv == nil { - continue - } - go func() { - for ctx.Err() == nil { - w, err := kv.WatchAll(ctx, jetstream.UpdatesOnly()) - if err != nil { - logf("the steps of %s cannot be watched, so a check whose own watch goes deaf is read only at a reconnect or by hand: %v", module, err) - return - } - heardAny := false - for e := range w.Updates() { - if e == nil || e.Operation() != jetstream.KeyValuePut { - continue - } - heardAny = true - if step, err := delivery.DecodeStep(e.Value()); err == nil && step.State != delivery.StepRunning { - heard.Nudge(module) - } - } - _ = w.Stop() - if ctx.Err() == nil { - heard.Nudge(module) // a step may have ended while the watch was closing - if !heardAny { - logf("the watch of the steps of %s closed having heard nothing, so it is left: a check whose own watch goes deaf is read only at a reconnect or by hand", module) - return - } - logf("the watch of the steps of %s closed; opened again", module) - } - } - }() - } -} diff --git a/modules/mesh-delivery-checker/go.mod b/modules/mesh-delivery-checker/go.mod index 85ac234..a1dad59 100644 --- a/modules/mesh-delivery-checker/go.mod +++ b/modules/mesh-delivery-checker/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-checker go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028174713-38dbe1d3a388 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028192930-6aeb40109529 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-checker/go.sum b/modules/mesh-delivery-checker/go.sum index 6245202..88f3a59 100644 --- a/modules/mesh-delivery-checker/go.sum +++ b/modules/mesh-delivery-checker/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028174713-38dbe1d3a388 h1:ycu/sNieTdLpD3bzhZOCXdzcoqXUOxPRmqWV019REpo= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028174713-38dbe1d3a388/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028192930-6aeb40109529 h1:Xx5mcDPWcRuPelUxdjvwkmVcfyGww0rRajA9JjFA/hk= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028192930-6aeb40109529/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-checker/internal/check/heardends.go b/modules/mesh-delivery-checker/internal/check/heardends.go deleted file mode 100644 index d122600..0000000 --- a/modules/mesh-delivery-checker/internal/check/heardends.go +++ /dev/null @@ -1,144 +0,0 @@ -package check - -import ( - "context" - "sync" - - "git.novox.be/novox/mesh-sdk/go/delivery" - "git.novox.be/novox/mesh-sdk/go/delivery/buildrun" -) - -// **A step's end is never only its own watch's to hear** (the lost check of 2026-10-28: mesh-catalog #553's verdict -// stayed running after its step mvsgthyjed5786 had passed on shanks at 19:09:31Z; the watch of that one key, opened at -// 18:46:05Z, heard neither the step's start nor its end and never closed, while a watch opened at 18:58:48Z heard its own -// step end). HeardEnds stands between the asker (buildrun.Asker) and the watch of each ask's step key: when it is -// nudged, every ask it waits on reads its step keys now (a read, not the watch) and hands what it reads to its wait, so a -// step that ended is judged even when its own watch went deaf. A watch that closes under a wait is opened again, at most -// reopenTries times. It is nudged on an event (Nudge): any step of the same build module ending (the checker's one watch of -// each steps state), a reconnect to the bus, and the hand (checker_ask_again's verdict). It holds no timer. - -// StepReader reads, now, every step a build module's steps state holds for an ask: its keys `*.`. -type StepReader interface { - AskSteps(ctx context.Context, module, askID string) ([]delivery.Step, error) -} - -// reopenTries is how often a wait opens its ask's watch again after it closed, before the wait ends failed. -const reopenTries = 3 - -// HeardEnds is an AskWaiter over Inner that also reads an ask's steps when nudged. Safe for concurrent use. -type HeardEnds struct { - Inner buildrun.AskWaiter - Read StepReader - Logf func(format string, args ...any) - - mu sync.Mutex - waiting map[string]*waitedAsk // by ask id -} - -type waitedAsk struct { - module string - nudge chan struct{} -} - -func (h *HeardEnds) logf(format string, args ...any) { - if h.Logf != nil { - h.Logf(format, args...) - } -} - -// WatchAsk is AskWaiter's: what Inner's watch hands, and, at each nudge, what the ask's keys hold now. -func (h *HeardEnds) WatchAsk(ctx context.Context, module, askID string) (<-chan delivery.Step, error) { - in, err := h.Inner.WatchAsk(ctx, module, askID) - if err != nil { - return nil, err - } - w := &waitedAsk{module: module, nudge: make(chan struct{}, 1)} - h.mu.Lock() - if h.waiting == nil { - h.waiting = map[string]*waitedAsk{} - } - h.waiting[askID] = w - h.mu.Unlock() - out := make(chan delivery.Step) - go func() { - defer close(out) - defer func() { - h.mu.Lock() - if h.waiting[askID] == w { - delete(h.waiting, askID) - } - h.mu.Unlock() - }() - send := func(s delivery.Step) bool { - select { - case out <- s: - return true - case <-ctx.Done(): - return false - } - } - reopened := 0 - for { - select { - case <-ctx.Done(): - return - case s, ok := <-in: - if !ok { - if ctx.Err() != nil || reopened >= reopenTries { - h.logf("the watch of the ask %s's step on %s closed %d times; its wait ends", askID, module, reopened+1) - return - } - reopened++ - if in, err = h.Inner.WatchAsk(ctx, module, askID); err != nil { - h.logf("the watch of the ask %s's step on %s closed and cannot be opened again, so its wait ends: %v", askID, module, err) - return - } - h.logf("the watch of the ask %s's step on %s closed; opened again (%d of %d)", askID, module, reopened, reopenTries) - continue - } - if !send(s) { - return - } - case <-w.nudge: - if h.Read == nil { - continue - } - steps, err := h.Read.AskSteps(ctx, module, askID) - if err != nil { - h.logf("the steps of the ask %s on %s cannot be read: %v", askID, module, err) - continue - } - for _, s := range steps { - if s.State != delivery.StepRunning { - h.logf("the ask %s's step on %s ended %s, read at a nudge: its watch had not handed it over", askID, s.Node, s.State) - } - if !send(s) { - return - } - } - } - } - }() - return out, nil -} - -// Nudge has every ask waited on of module read its steps now; module "" nudges every ask. -func (h *HeardEnds) Nudge(module string) { - h.mu.Lock() - defer h.mu.Unlock() - for _, w := range h.waiting { - if module == "" || w.module == module { - select { - case w.nudge <- struct{}{}: - default: // a nudge is pending already - } - } - } -} - -// Waiting is how many asks are waited on now. -func (h *HeardEnds) Waiting() int { - h.mu.Lock() - defer h.mu.Unlock() - return len(h.waiting) -} diff --git a/modules/mesh-delivery-checker/internal/check/lostask_test.go b/modules/mesh-delivery-checker/internal/check/lostask_test.go index bf32de5..3539b8c 100644 --- a/modules/mesh-delivery-checker/internal/check/lostask_test.go +++ b/modules/mesh-delivery-checker/internal/check/lostask_test.go @@ -269,7 +269,7 @@ func TestACheckWhoseAskJudgedNothingIsAskedAgainOrRead(t *testing.T) { t.Run(row.name, func(t *testing.T) { b := newBench(t) var asked func() int - var heard *HeardEnds + var heard *buildrun.HeardEnds var tr *troubled if row.lose > 0 { l := losing(b, row.lose) @@ -277,7 +277,7 @@ func TestACheckWhoseAskJudgedNothingIsAskedAgainOrRead(t *testing.T) { } else { tr = &troubled{FakeAsks: &buildrun.FakeAsks{}, answer: b.answer, restarts: row.restarts, deaf: row.deaf, deafIDs: map[string]bool{}, steps: map[string]delivery.Step{}} - heard = &HeardEnds{Inner: tr, Read: tr} + heard = &buildrun.HeardEnds{Inner: tr, Read: tr} b.ex.Asks = buildrun.Asker{Publisher: tr, Waiter: heard, Bound: 5 * time.Second} asked = func() int { return len(tr.Asks()) } }