From 6f079f13a82703f45e07342366843924c8c6b93e Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 28 Oct 2026 20:27:41 +0100 Subject: [PATCH 1/7] Checker: never lose a step's end, and ask a lost check again (issue 734's kind) mesh-catalog #553's verdict stayed running after its step mvsgthyjed5786 passed on shanks at 19:09:31Z: the watch of that one step key, opened at 18:46:05Z, heard neither its start nor its end and never closed, so the wait had no bound running and nothing would ever judge it. - HeardEnds stands between the asker and each ask's watch: at every step end of the same build module (one watch per steps state), at a reconnect, the asks waited on read their keys now and are judged; a watch that closes is opened again, at most 3 times. - A step the build module's next process ended (RestartReason, a delivery replacing it) counts as a lost ask: asked again under the claim, at most LostAskTries, then failed as not heard, judged again when the nodes change. - checker_ask_again {group, why} ends a running verdict as asked again by hand; the checker hears its verdicts, ends the wait and claims it anew. Refused while a step of its pull requests runs, or for a verdict that is not running. - checker_status names such checks under lost, and each step's what. Tests: one table replaces the given-up test (rows for given up, replaced, deaf and nudged, deaf and asked by hand); counts unchanged (147 unit, 10 integration in the checker). --- .../cmd/mesh-delivery-checker-tools/main.go | 35 ++- .../mesh-delivery-checker-tools/main_test.go | 8 + .../cmd/mesh-delivery-checker/main.go | 20 +- .../cmd/mesh-delivery-checker/stepends.go | 83 +++++++ .../internal/check/askagain.go | 132 +++++++++++ .../internal/check/checker.go | 14 +- .../internal/check/heardends.go | 144 ++++++++++++ .../internal/check/lostask.go | 9 +- .../internal/check/lostask_test.go | 208 ++++++++++++++++-- .../internal/check/read.go | 185 +++++++++++----- .../mesh-delivery-checker/manifest_test.go | 5 +- modules/mesh-delivery-checker/module.json | 1 + 12 files changed, 753 insertions(+), 91 deletions(-) create mode 100644 modules/mesh-delivery-checker/cmd/mesh-delivery-checker/stepends.go create mode 100644 modules/mesh-delivery-checker/internal/check/askagain.go create mode 100644 modules/mesh-delivery-checker/internal/check/heardends.go diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go index a6a6936..54ac89d 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go @@ -1,6 +1,6 @@ // Command mesh-delivery-checker-tools serves the checker's read tools (novox/hq ADR 0484 clause 13, issue 1686): the runtime // launches it for a call, it reads the checker's verdicts and kept states and the indexer's candidates through the checker's -// state and reads, and writes nothing. The checker is the other program: it serves no tool (ADR 0437 §4). +// state and reads. It writes only the tries (the try's tools) and, asking a check again by hand, one verdict. The checker is the other program: it serves no tool (ADR 0437 §4). package main import ( @@ -235,6 +235,24 @@ func Tools(r check.Reader) []stdio.Tool { } } +// RecoveryTools are the tools that steer a check by hand (tools-are-how-a-module-is-spoken-to): ask a check again whose +// wait judged nothing (the lost check of 2026-10-28). They write the checker's verdicts through w. +func RecoveryTools(r check.Reader, w check.VerdictEntry) []stdio.Tool { + str := func(args map[string]any, k string) string { s, _ := args[k].(string); return s } + return []stdio.Tool{{Name: check.AskAgainTool, + Description: "Ask a candidate's check again by hand, whose verdict stays running while nothing runs for it (" + check.StatusReadTool + " names it " + + "under lost): its verdict is ended as asked again, saying why, and the checker, which hears its verdicts, stops the wait that judged nothing " + + "and asks the check anew, whole. Refused for a group with no candidate, a verdict that is not running (a pass or failure stands until a push or " + + "a trunk move), and while a step of its pull requests can still run on a build module (its end judges it). The checker reads a step that " + + "ended while its watch was deaf by itself at the next step end of its build module; this is for what that does not reach.", + Input: map[string]any{"type": "object", "required": []string{"group", "why"}, "properties": map[string]any{ + "group": map[string]any{"type": "string", "description": "the delivery group's branch name"}, + "why": map[string]any{"type": "string", "description": "why it is asked again, in words; kept in the verdict it ends"}}}, + Run: func(args map[string]any) (any, error) { + return r.AskAgain(w, str(args, "group"), str(args, "why")) + }}} +} + // stepBound is the checker's step bound as the checker reads it (MESH_CHECKER_STEP_BOUND), its default when unset or not a // duration. func stepBound() time.Duration { @@ -277,7 +295,20 @@ func main() { return err }} trying := Trying{Tries: store, Ender: store, Trunks: trunks} - if err := stdio.Serve("", logged(append(Tools(r), TryTools(trying)...))); err != nil { + verdicts := stdio.State("verdicts") + written := stateStore{entry: func(key string) (json.RawMessage, uint64, error) { + reads.Add(1) + e, err := verdicts.Get(key) + if err != nil || e == nil { + return nil, 0, err + } + return e.Value, e.Revision, nil + }, update: func(key string, value []byte, revision uint64) error { + _, err := verdicts.Update(key, json.RawMessage(value), revision) + return err + }} + tools := append(append(Tools(r), TryTools(trying)...), RecoveryTools(r, written)...) + if err := stdio.Serve("", logged(tools)); err != nil { fmt.Fprintf(os.Stderr, "[mesh-delivery-checker-tools] %v\n", err) os.Exit(1) } diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main_test.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main_test.go index 9f250df..282db23 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main_test.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main_test.go @@ -23,6 +23,14 @@ func TestTheToolsServedAreTheFiveReadTools(t *testing.T) { if !slices.Equal(names, want) { t.Errorf("tools = %v, want %v", names, want) } + // The recovery tool beside them asks a lost check again; with nothing to write it refuses. + recovery := RecoveryTools(check.Reader{}, nil) + if len(recovery) != 1 || recovery[0].Name != check.AskAgainTool || recovery[0].Description == "" { + t.Errorf("recovery tools = %+v", recovery) + } + if _, err := recovery[0].Run(map[string]any{"group": "feat/x", "why": "lost"}); err == nil { + t.Errorf("%s answered with no state to write", check.AskAgainTool) + } // With nothing to read from a tool says so and changes nothing. for _, tl := range Tools(check.Reader{}) { if tl.Name == check.StatusReadTool { 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 52530d5..1563dcb 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go @@ -144,8 +144,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro hearing := buildrun.NewDeadLetterHearing(ctx, buildrun.JetStreamDeadLetters{JS: js}, []string{delivery.AskGo, delivery.AskNode}, dead, logger.Printf, buildrun.SayHearing(ctx, hearingKV, logger.Printf)) defer hearing.Stop() - conn.SetReconnectHandler(func(*nats.Conn) { hearing.Again() }) - waiter := buildrun.LostAskWaiter{Inner: asks, Dead: dead, Hearing: hearing} + // 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) + conn.SetReconnectHandler(func(*nats.Conn) { hearing.Again(); heard.Nudge("") }) + waiter := buildrun.LostAskWaiter{Inner: heard, Dead: dead, Hearing: hearing} ch := &check.Checker{ Verdicts: kvStore{buckets[delivery.CheckerVerdictsBucket]}, Owed: kvStore{buckets[delivery.CheckerOwedBucket]}, @@ -331,6 +335,18 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } }) }() + // The checker's own verdicts, updates only: one checker_ask_again wrote asks its check again (Checker.VerdictWritten). + vw, err := buckets[delivery.CheckerVerdictsBucket].WatchAll(ctx, jetstream.UpdatesOnly()) + if err != nil { + return fmt.Errorf("the checker's verdicts cannot be watched, so no check can be asked again by hand: %w", err) + } + go func() { + errs <- follow(ctx, vw, func(e jetstream.KeyValueEntry) { + if opOf(e) { + ch.VerdictWritten(ctx, e.Key(), e.Value()) + } + }) + }() logger.Printf("started on %s", cred.Node) go func() { errs <- follow(ctx, w, each) }() select { diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/stepends.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/stepends.go new file mode 100644 index 0000000..0193082 --- /dev/null +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/stepends.go @@ -0,0 +1,83 @@ +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/internal/check/askagain.go b/modules/mesh-delivery-checker/internal/check/askagain.go new file mode 100644 index 0000000..3a4df17 --- /dev/null +++ b/modules/mesh-delivery-checker/internal/check/askagain.go @@ -0,0 +1,132 @@ +package check + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// **A check is asked again by hand** (the lost check of 2026-10-28; tools-are-how-a-module-is-spoken-to: every recovery +// has a tool). checker_ask_again ends a candidate's running verdict, whose wait judged nothing, with AskedAgainWords and +// the why; the checker hears it on its watch of its verdicts (Checker.VerdictWritten), ends the check of those heads +// running in it, and claims and asks it anew, a whole check (IsEnded). It is refused while a step of the candidate's pull +// requests can still run (checker_status's running_steps): that step's end judges it. + +// AskAgainTool asks a candidate's check again by hand. +const AskAgainTool = "checker_ask_again" + +// AskedAgainWords open the words of a verdict AskAgainTool wrote; the why follows. +const AskedAgainWords = EndedWords + "asked again by hand (" + AskAgainTool + "): " + +// errAskedAgain is the cause a running check is ended with when its verdict was asked again by hand. +var errAskedAgain = errors.New("its verdict was asked again by hand (" + AskAgainTool + ")") + +// IsAskedAgain says whether a verdict was written by AskAgainTool. +func IsAskedAgain(v delivery.Verdict) bool { + return IsEnded(v) && strings.HasPrefix(v.Failed.Words, AskedAgainWords) +} + +// VerdictEntry is the verdicts state as the ask-again tool writes it: a key's value and revision (nil and 0 for none), +// and a compare-and-set. +type VerdictEntry interface { + Entry(key string) (json.RawMessage, uint64, error) + Update(key string, value []byte, revision uint64) error +} + +// AskedAgain is the answer of AskAgainTool. +type AskedAgain struct { + Key string `json:"key"` + Was string `json:"was"` + Said string `json:"said"` + Note string `json:"note"` +} + +// AskAgain ends the running verdict of the group's candidate as asked again by hand, saying why. It refuses a group with +// no candidate, a verdict that is not running (a pass or a failure stands until a push or a trunk move), and one a step +// of whose pull requests can still run. +func (r Reader) AskAgain(w VerdictEntry, group, why string) (AskedAgain, error) { + why = strings.TrimSpace(why) + if group == "" || why == "" { + return AskedAgain{}, fmt.Errorf("give the group and why it is asked again") + } + if w == nil { + return AskedAgain{}, fmt.Errorf("the verdicts state cannot be written here") + } + c, ok := r.candidate(group) + if !ok { + return AskedAgain{}, fmt.Errorf("no candidate is held for the group %q", group) + } + key, err := delivery.CheckerVerdictKey(c.Group, c.HeadsHash) + if err != nil { + return AskedAgain{}, err + } + raw, rev, err := w.Entry(key) + if err != nil { + return AskedAgain{}, fmt.Errorf("%s cannot be read: %w", key, err) + } + if len(raw) == 0 { + return AskedAgain{}, fmt.Errorf("%s holds no verdict: the checker claims its check when it looks at the candidate (%s)", key, StatusReadTool) + } + v, err := delivery.DecodeVerdict(raw) + if err != nil { + return AskedAgain{}, fmt.Errorf("%s holds what is not a verdict: %w", key, err) + } + if v.State != delivery.VerdictRunning { + return AskedAgain{}, fmt.Errorf("%s is %s, not running: a verdict stands until a push or a trunk move asks it again", key, v.State) + } + for _, rs := range r.readSteps().running { + if asksOf(c, rs.What) { + return AskedAgain{}, fmt.Errorf("a step of it runs: %s %s on %s since %s (%s); its end judges it", rs.Module, rs.ID, rs.Node, rs.Since, rs.What) + } + } + words := AskedAgainWords + why + ended := delivery.Verdict{Group: v.Group, Heads: v.Heads, Trunk: v.Trunk, State: delivery.VerdictFail, + Failed: delivery.FailedFirst{Machine: "checker", Words: words}} + data, err := delivery.EncodeVerdict(ended) + if err != nil { + return AskedAgain{}, err + } + if err := w.Update(key, data, rev); err != nil { + return AskedAgain{}, fmt.Errorf("%s changed while it was asked again, so it was not; read it again: %w", key, err) + } + return AskedAgain{Key: key, Was: v.State, Said: words, + Note: "the checker hears it on its watch of its verdicts, ends the wait that judged nothing and asks the check anew, whole; " + + VerdictsReadTool + " shows it running again"}, nil +} + +// VerdictWritten takes a change of the checker's verdicts in (its watch, updates only): a verdict AskAgainTool wrote ends +// the check of those heads running in this process, which then considers them again; with none running, the candidate +// that holds them is considered at once. Any other verdict is the checker's own, and left. +func (ch *Checker) VerdictWritten(ctx context.Context, key string, value []byte) { + v, err := delivery.DecodeVerdict(value) + if err != nil || !IsAskedAgain(v) { + return + } + ch.init() + ch.mu.Lock() + if l, ok := ch.live[key]; ok { + l.end(errAskedAgain) + ch.mu.Unlock() + ch.logf("%s: %s; the check running for it is ended and asked again", key, v.Failed.Words) + return + } + var group string + var cand delivery.Candidate + for k, c := range ch.known { + if vk, err := delivery.CheckerVerdictKey(c.Group, c.HeadsHash); err == nil && vk == key { + group, cand = k, c + break + } + } + ch.mu.Unlock() + if group == "" { + ch.logf("%s: %s, but no candidate holds these heads, so nothing is asked", key, v.Failed.Words) + return + } + ch.logf("%s: %s; asked again", key, v.Failed.Words) + ch.consider(ctx, group, cand) +} diff --git a/modules/mesh-delivery-checker/internal/check/checker.go b/modules/mesh-delivery-checker/internal/check/checker.go index 3e5b1da..aa9a8c7 100644 --- a/modules/mesh-delivery-checker/internal/check/checker.go +++ b/modules/mesh-delivery-checker/internal/check/checker.go @@ -328,7 +328,11 @@ func (ch *Checker) consider(ctx context.Context, key string, c delivery.Candidat delete(ch.again, vkey) delete(ch.live, vkey) ch.mu.Unlock() + again := errors.Is(context.Cause(run), errAskedAgain) end(nil) + if again && ctx.Err() == nil { + ch.consider(ctx, key, ch.candidate(key, c)) // asked again by hand (AskAgainTool): claimed anew + } }() for { // Every check is asked at once: the build seats' streams hold the backlog, a new head's check and a check after a @@ -713,8 +717,10 @@ func (ch *Checker) try(ctx, run context.Context, key string, cs *Case) { began := ch.now() ch.writeOwed(ctx, cs, delivery.Verdict{State: delivery.VerdictRunning}, Outcome{}) out := ch.judge(run, cs) - // An ask the build seat gave up on judged nothing (issue 734): the check is asked again under the same claim, its status - // pending, and after LostAskTries it fails as not heard, judged again when the nodes change. + // An ask the build seat gave up on, or whose step the build module's next process ended because it was replaced under + // it, judged nothing (issue 734, IsLostAsk): the check is asked again under the same claim, its status pending, and after + // LostAskTries it fails as not heard, judged again when the nodes change. A step that ended while its watch was deaf is + // read at the next nudge (HeardEnds) and judged. for try := 1; IsLostAsk(out.Verdict) && run.Err() == nil; try++ { if try >= LostAskTries { out.Verdict.Failed.Words = strings.TrimSpace(out.Verdict.Failed.Words + "\nasked " + fmt.Sprint(LostAskTries) + @@ -724,8 +730,8 @@ func (ch *Checker) try(ctx, run context.Context, key string, cs *Case) { ch.logf("%s: %s; asked again (%d of %d)", cs.Key, out.Verdict.Failed.Words, try+1, LostAskTries) out = ch.judge(run, cs) } - if ctx.Err() != nil { - return + if ctx.Err() != nil || errors.Is(context.Cause(run), errAskedAgain) { + return // the checker stops, or the verdict was asked again by hand, which wrote it: its next check claims it anew } var why superseded if errors.As(context.Cause(run), &why) { diff --git a/modules/mesh-delivery-checker/internal/check/heardends.go b/modules/mesh-delivery-checker/internal/check/heardends.go new file mode 100644 index 0000000..d122600 --- /dev/null +++ b/modules/mesh-delivery-checker/internal/check/heardends.go @@ -0,0 +1,144 @@ +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.go b/modules/mesh-delivery-checker/internal/check/lostask.go index 4db60c6..8328e46 100644 --- a/modules/mesh-delivery-checker/internal/check/lostask.go +++ b/modules/mesh-delivery-checker/internal/check/lostask.go @@ -3,6 +3,7 @@ package check import ( "strings" + "git.novox.be/novox/mesh-sdk/go/buildstep" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/delivery/buildrun" ) @@ -19,9 +20,13 @@ import ( // judged again when the nodes change. const LostAskTries = 3 -// IsLostAsk says whether a verdict failed because the build seat gave its ask up (buildrun.LostWords). +// IsLostAsk says whether a verdict failed because its ask judged nothing: the build seat gave it up (buildrun.LostWords), or +// the build module that ran its step was replaced under it, and its next process ended the step it left running +// (buildstep.RestartReason; a delivery of the build module, as on 2026-10-28). Neither is the change's failure: the check +// is asked again, as for a given-up ask. func IsLostAsk(v delivery.Verdict) bool { - return v.State == delivery.VerdictFail && strings.Contains(v.Failed.Words, buildrun.LostWords) + return v.State == delivery.VerdictFail && (strings.Contains(v.Failed.Words, buildrun.LostWords) || + strings.Contains(v.Failed.Words, buildstep.RestartReason)) } // IsNotTaken says whether a verdict failed because no build node took its ask within the bound (buildrun.NotTakenWords). diff --git a/modules/mesh-delivery-checker/internal/check/lostask_test.go b/modules/mesh-delivery-checker/internal/check/lostask_test.go index 35ea6a5..bf32de5 100644 --- a/modules/mesh-delivery-checker/internal/check/lostask_test.go +++ b/modules/mesh-delivery-checker/internal/check/lostask_test.go @@ -8,6 +8,7 @@ import ( "testing" "time" + "git.novox.be/novox/mesh-sdk/go/buildstep" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/delivery/buildrun" ) @@ -141,30 +142,195 @@ func TestAnAskGivenUpOrNeverTakenEndsItsTryAsErrorAndItsHeadIsAskedAnew(t *testi }) } -// Issue 734: a pull request's check whose ask the build seat gave up on is asked again under its claim, its status -// pending, and judged; one given up every time fails as not heard and is judged again when the nodes change. -func TestACheckWhoseAskWasGivenUpIsAskedAgain(t *testing.T) { +// troubled is the build seat as a check meets it when its step's end does not reach its wait: the first `restarts` asks' +// steps are ended by the build module's next process (buildstep.RestartReason), and the first `deaf` asks' watches hear +// nothing though their steps end (the lost check of 2026-10-28). Its steps state is read through AskSteps and Steps. +type troubled struct { + *buildrun.FakeAsks + answer func(delivery.StepAsk) delivery.Step + + mu sync.Mutex + restarts, deaf int + deafIDs map[string]bool + steps map[string]delivery.Step +} + +func (tr *troubled) PublishAsk(ctx context.Context, subject, id string, data []byte) error { + if err := tr.FakeAsks.PublishAsk(ctx, subject, id, data); err != nil { + return err + } + a, err := delivery.DecodeStepAsk(data) + if err != nil { + return err + } + step := tr.answer(a) + step.What = buildstep.AskWhat(a) + tr.mu.Lock() + if tr.restarts > 0 { + tr.restarts-- + step.State, step.Reason, step.Result = delivery.StepFailed, buildstep.RestartReason, nil + } + deaf := tr.deaf > 0 + if deaf { + tr.deaf-- + tr.deafIDs[id] = true + } + tr.steps[id] = step + tr.mu.Unlock() + if !deaf { + tr.Write(step) + } + return nil +} + +func (tr *troubled) WatchAsk(ctx context.Context, module, askID string) (<-chan delivery.Step, error) { + tr.mu.Lock() + deaf := tr.deafIDs[askID] + tr.mu.Unlock() + if deaf { + return make(chan delivery.Step), nil // open, and silent for good + } + return tr.FakeAsks.WatchAsk(ctx, module, askID) +} + +func (tr *troubled) AskSteps(_ context.Context, _, askID string) ([]delivery.Step, error) { + tr.mu.Lock() + defer tr.mu.Unlock() + if s, ok := tr.steps[askID]; ok { + return []delivery.Step{s}, nil + } + return nil, nil +} + +// Steps is the steps state as the read tools read it, with running in place of every step the hand has not let end. +func (tr *troubled) Steps(running bool) StateRead { + tr.mu.Lock() + defer tr.mu.Unlock() + vals := map[string]json.RawMessage{} + for id, s := range tr.steps { + if running { + s.State, s.Ended, s.Started = delivery.StepRunning, time.Time{}, time.Now() + } + raw, _ := delivery.EncodeStep(s) + vals[s.Node+"."+id] = raw + } + return StateRead{Keys: func() ([]string, error) { + var out []string + for k := range vals { + out = append(out, k) + } + return out, nil + }, Get: func(k string) (json.RawMessage, error) { return vals[k], nil }} +} + +// kvRead is a bucket in memory as the read tools read it, live. +func kvRead(kv *memKV) StateRead { + return StateRead{Keys: func() ([]string, error) { return kv.Keys(context.Background()) }, + Get: func(k string) (json.RawMessage, error) { + v, _, _, err := kv.Get(context.Background(), k) + return v, err + }} +} + +// asVerdicts is the rig's verdicts as the ask-again tool reads and writes them. +type asVerdicts struct{ kv *memKV } + +func (a asVerdicts) Entry(key string) (json.RawMessage, uint64, error) { + v, rev, _, err := a.kv.Get(context.Background(), key) + return v, rev, err +} +func (a asVerdicts) Update(key string, value []byte, revision uint64) error { + _, err := a.kv.Update(context.Background(), key, value, revision) + return err +} + +// Issue 734 and the lost check of 2026-10-28: a check whose ask judged nothing is asked again, and one whose step ended +// unheard is judged. Given up by the build seat once, or its step ended by the build module's next process after a +// delivery replaced it, the check is asked again under its claim and passes; given up every time it fails loudly as not +// heard, judged again when the nodes change. A step whose watch went deaf is read at the next step end of its build module +// (a nudge) and judged with no new ask; one no event reaches is named lost by checker_status, refused by checker_ask_again +// while its step still runs, and asked again by it once its step ended. +func TestACheckWhoseAskJudgedNothingIsAskedAgainOrRead(t *testing.T) { ctx := context.Background() for _, row := range []struct { - lose int - state string - asks int + name string + lose, restarts, deaf int + how string // "nudge": a step of the build module ends; "hand": checker_ask_again + state string + asks int }{ - {1, delivery.VerdictPass, 2}, - {LostAskTries, delivery.VerdictFail, LostAskTries}, + {name: "given up once", lose: 1, state: delivery.VerdictPass, asks: 2}, + {name: "given up every time", lose: LostAskTries, state: delivery.VerdictFail, asks: LostAskTries}, + {name: "its build module replaced under it", restarts: 1, state: delivery.VerdictPass, asks: 2}, + {name: "replaced under every ask", restarts: LostAskTries, state: delivery.VerdictFail, asks: LostAskTries}, + {name: "its watch deaf, a step of its build module ends", deaf: 1, how: "nudge", state: delivery.VerdictPass, asks: 1}, + {name: "its watch deaf, asked again by hand", deaf: 1, how: "hand", state: delivery.VerdictPass, asks: 2}, } { - b := newBench(t) - l := losing(b, row.lose) - r := newRig(t, b.ex) - cand := candidateOf(t, "feat/x", pullOf(repoA, 7, head1)) - r.ch.Candidate(ctx, "feat%2Fx", cand) - r.ch.Wait() - v := r.verdict(t, keyOf(t, "feat/x", cand)) - if v.State != row.state || len(l.Asks()) != row.asks { - t.Errorf("lose %d: verdict %s (%s) after %d asks, want %s after %d", row.lose, v.State, v.Failed.Words, len(l.Asks()), row.state, row.asks) - } - if row.state == delivery.VerdictFail && (!IsLostAsk(v) || !IsNodesFailure(v)) { - t.Errorf("a check given up every time = %q, want lost and judged again when the nodes change", v.Failed.Words) - } + t.Run(row.name, func(t *testing.T) { + b := newBench(t) + var asked func() int + var heard *HeardEnds + var tr *troubled + if row.lose > 0 { + l := losing(b, row.lose) + asked = func() int { return len(l.Asks()) } + } 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} + b.ex.Asks = buildrun.Asker{Publisher: tr, Waiter: heard, Bound: 5 * time.Second} + asked = func() int { return len(tr.Asks()) } + } + r := newRig(t, b.ex) + cand := candidateOf(t, "feat/x", pullOf(repoA, 7, head1)) + key := keyOf(t, "feat/x", cand) + r.ch.Candidate(ctx, "feat%2Fx", cand) + if row.deaf > 0 { + for deadline := time.Now().Add(2 * time.Second); heard.Waiting() == 0 && time.Now().Before(deadline); { + time.Sleep(2 * time.Millisecond) + } + if v := r.verdict(t, key); v.State != delivery.VerdictRunning { + t.Fatalf("a check whose step ended unheard = %s, want running until it is read", v.State) + } + } + switch row.how { + case "nudge": + heard.Nudge(buildrun.NodeBuildModule) // another build module's step: not this one's to read + heard.Nudge(buildrun.GoBuildModule) + case "hand": + cands := newMemKV() + ckey, _ := delivery.CandidatesKey("feat/x") + _, _ = cands.Create(ctx, ckey, cand) + rd := Reader{CandidateState: kvRead(cands), VerdictState: kvRead(r.verdicts), + StepStates: map[string]StateRead{buildrun.GoBuildModule: tr.Steps(true)}} + if _, err := rd.AskAgain(asVerdicts{r.verdicts}, "feat/x", "its watch went deaf"); err == nil || !strings.Contains(err.Error(), "a step of it runs") { + t.Errorf("asked again while its step runs: %v", err) + } + rd.StepStates[buildrun.GoBuildModule] = tr.Steps(false) + if st, _ := rd.Status(); !strings.Contains(st.Lost["feat/x"], AskAgainTool) { + t.Errorf("status lost = %v, want feat/x named with %s", st.Lost, AskAgainTool) + } + got, err := rd.AskAgain(asVerdicts{r.verdicts}, "feat/x", "its watch went deaf") + if err != nil || got.Key != key || !strings.HasPrefix(got.Said, AskedAgainWords) { + t.Fatalf("asked again = %+v, %v", got, err) + } + raw, _, _, _ := r.verdicts.Get(ctx, key) + r.ch.VerdictWritten(ctx, key, raw) // the checker's watch of its verdicts hands it over + } + r.ch.Wait() + v := r.verdict(t, key) + if v.State != row.state || asked() != row.asks { + t.Errorf("verdict %s (%s) after %d asks, want %s after %d", v.State, v.Failed.Words, asked(), row.state, row.asks) + } + if row.state == delivery.VerdictFail && (!IsLostAsk(v) || !IsNodesFailure(v)) { + t.Errorf("a check that judged nothing every time = %q, want lost and judged again when the nodes change", v.Failed.Words) + } + if row.how == "hand" { + rd := Reader{VerdictState: kvRead(r.verdicts), CandidateState: kvRead(newMemKV())} + if _, err := rd.AskAgain(asVerdicts{r.verdicts}, "feat/x", "again"); err == nil { + t.Error("a judged check was asked again") + } + } + }) } } diff --git a/modules/mesh-delivery-checker/internal/check/read.go b/modules/mesh-delivery-checker/internal/check/read.go index cbe4581..f5d82e5 100644 --- a/modules/mesh-delivery-checker/internal/check/read.go +++ b/modules/mesh-delivery-checker/internal/check/read.go @@ -306,6 +306,8 @@ type RunningStep struct { ID string `json:"id"` Tool string `json:"tool"` Since string `json:"since,omitempty"` + // What says what the step is (buildstep.AskWhat): an ask's kind, repository@head and levels. + What string `json:"what,omitempty"` } // LeftStep is a key a build module left running that no step behind it runs any more, and why. @@ -331,6 +333,9 @@ type Status struct { // CannotBeJudged are the candidates no pull request of which can be judged, by group, with why: the gate's source.main // names no trunk head of their repositories (issue 1422). They are judged once it does; until then their status is error. CannotBeJudged map[string]string `json:"cannot_be_judged"` + // Lost are the candidates whose verdict is running while no step of theirs can still run and their newest step ended, by + // group, with that step: nothing will judge them (the lost check of 2026-10-28); checker_ask_again asks one again. + Lost map[string]string `json:"lost"` // DeadLetters says whether the checker hears the asks the build seats give up on (issue 734): deaf, a try or check // whose ask is given up waits until its bound. DeadLetters DeadLettersHeard `json:"dead_letters"` @@ -394,63 +399,11 @@ func heldBy(t *Turns, c delivery.Candidate, heads map[string]string) string { // Status reads the build modules' steps and the candidates against their verdicts. func (r Reader) Status() (Status, error) { - st := Status{RunningByNode: map[string]int{}} - mods := make([]string, 0, len(r.StepStates)) - for m := range r.StepStates { - mods = append(mods, m) - } - sort.Strings(mods) - st.Running, st.LeftRunning = []RunningStep{}, []LeftStep{} - now, bound := time.Now(), r.StepBound - if r.Now != nil { - now = r.Now() - } - if bound <= 0 { - bound = DefaultStepBound - } - for _, m := range mods { - sr := r.StepStates[m] - if sr.Keys == nil || sr.Get == nil { - continue - } - keys, err := sr.Keys() - if err != nil { - st.Notes = append(st.Notes, "the steps of "+m+" cannot be listed: "+err.Error()) - continue - } - // A step that runs started lately: the newest ids are read first, at most MaxReads of them. - newestFirst(keys) - keys, cut := capKeys(keys, "") - if cut != "" { - st.Notes = append(st.Notes, "the steps of "+m+": read the newest "+fmt.Sprint(MaxReads)+" of its keys; a step older than those is not shown") - } - steps := make([]delivery.Step, len(keys)) - isRunning := make([]bool, len(keys)) - _ = parallel(len(keys), func(i int) error { - raw, err := sr.Get(keys[i]) - if err != nil || len(raw) == 0 { - return nil - } - step, err := delivery.DecodeStep(raw) - steps[i], isRunning[i] = step, err == nil && step.State == delivery.StepRunning - return nil - }) - for i := range keys { - if !isRunning[i] { - continue - } - step := steps[i] - rs := RunningStep{Module: m, Node: step.Node, ID: step.ID, Tool: step.Tool} - if !step.Started.IsZero() { - rs.Since = step.Started.UTC().Format("2006-01-02T15:04:05Z") - } - if why := r.leftWhy(m, step, now, bound); why != "" { - st.LeftRunning = append(st.LeftRunning, LeftStep{RunningStep: rs, Why: why}) - continue - } - st.Running = append(st.Running, rs) - st.RunningByNode[step.Node]++ - } + st := Status{RunningByNode: map[string]int{}, Lost: map[string]string{}} + read := r.readSteps() + st.Running, st.LeftRunning, st.Notes = read.running, read.left, read.notes + for _, rs := range st.Running { + st.RunningByNode[rs.Node]++ } if r.HaveState.Get == nil { st.Notes = append(st.Notes, "the node-engine's have is not readable here, so a step is not checked against its build module being on its node") @@ -530,7 +483,11 @@ func (r Reader) Status() (Status, error) { } v, err := delivery.DecodeVerdict(raw) switch { - case err != nil || v.State == delivery.VerdictRunning: + case err != nil: + case v.State == delivery.VerdictRunning: + if why := read.lostWhy(c); why != "" { + st.Lost[c.Group] = why + } case IsNodesFailure(v): st.Waiting[KindNodes] = append(st.Waiting[KindNodes], c.Group) case IsUntrunked(v): @@ -551,6 +508,118 @@ func (r Reader) Status() (Status, error) { return st, nil } +// stepsRead is what the build modules' steps states hold: the steps that can still run, those left running, and every +// ask step read, by module, for the candidate it belongs to (stepsOf). +type stepsRead struct { + running []RunningStep + left []LeftStep + all []moduleStep + notes []string +} + +type moduleStep struct { + module string + step delivery.Step +} + +// readSteps reads the newest MaxReads keys of each build module's steps state. +func (r Reader) readSteps() stepsRead { + out := stepsRead{running: []RunningStep{}, left: []LeftStep{}} + mods := make([]string, 0, len(r.StepStates)) + for m := range r.StepStates { + mods = append(mods, m) + } + sort.Strings(mods) + now, bound := time.Now(), r.StepBound + if r.Now != nil { + now = r.Now() + } + if bound <= 0 { + bound = DefaultStepBound + } + for _, m := range mods { + sr := r.StepStates[m] + if sr.Keys == nil || sr.Get == nil { + continue + } + keys, err := sr.Keys() + if err != nil { + out.notes = append(out.notes, "the steps of "+m+" cannot be listed: "+err.Error()) + continue + } + // A step that runs started lately: the newest ids are read first, at most MaxReads of them. + newestFirst(keys) + keys, cut := capKeys(keys, "") + if cut != "" { + out.notes = append(out.notes, "the steps of "+m+": read the newest "+fmt.Sprint(MaxReads)+" of its keys; a step older than those is not shown") + } + steps := make([]delivery.Step, len(keys)) + decoded := make([]bool, len(keys)) + _ = parallel(len(keys), func(i int) error { + raw, err := sr.Get(keys[i]) + if err != nil || len(raw) == 0 { + return nil + } + step, err := delivery.DecodeStep(raw) + steps[i], decoded[i] = step, err == nil + return nil + }) + for i := range keys { + if !decoded[i] { + continue + } + step := steps[i] + out.all = append(out.all, moduleStep{module: m, step: step}) + if step.State != delivery.StepRunning { + continue + } + rs := RunningStep{Module: m, Node: step.Node, ID: step.ID, Tool: step.Tool, What: step.What} + if !step.Started.IsZero() { + rs.Since = step.Started.UTC().Format("2006-01-02T15:04:05Z") + } + if why := r.leftWhy(m, step, now, bound); why != "" { + out.left = append(out.left, LeftStep{RunningStep: rs, Why: why}) + continue + } + out.running = append(out.running, rs) + } + } + return out +} + +// asksOf says whether a step is an ask of one of the candidate's pull requests: its What names `@` +// (buildstep.AskWhat). +func asksOf(c delivery.Candidate, what string) bool { + for _, p := range c.Pulls { + if p.Repository != "" && strings.Contains(what, p.Repository+"@"+short12(p.Head.Head)) { + return true + } + } + return false +} + +// lostWhy says why a candidate whose verdict is running has a check that nothing will judge, or "": no step of its pull +// requests can still run, and its newest one ended. +func (s stepsRead) lostWhy(c delivery.Candidate) string { + for _, rs := range s.running { + if asksOf(c, rs.What) { + return "" + } + } + var last *delivery.Step + for i := range s.all { + st := s.all[i].step + if st.State != delivery.StepRunning && asksOf(c, st.What) && (last == nil || st.Ended.After(last.Ended)) { + last = &s.all[i].step + } + } + if last == nil { + return "" // its ask waits in its build seat's stream: no node took it yet + } + return fmt.Sprintf("its verdict is running, but no step of it runs: its last step %s on %s ended %s at %s and no verdict followed; %s asks it again", + last.ID, last.Node, last.State, last.Ended.UTC().Format(time.RFC3339), AskAgainTool) +} + // DefaultStepBound is how long one check ask's step may run once a build node took it (its key written running; the fetch, // the digests and the levels), never counting its wait in the stream; past it the check fails naming the node. const DefaultStepBound = 30 * time.Minute diff --git a/modules/mesh-delivery-checker/manifest_test.go b/modules/mesh-delivery-checker/manifest_test.go index 815ce91..a56a841 100644 --- a/modules/mesh-delivery-checker/manifest_test.go +++ b/modules/mesh-delivery-checker/manifest_test.go @@ -124,8 +124,9 @@ func TestManifestDeclarations(t *testing.T) { // 1678/AC1 t.Errorf("streams = %+v, want %+v", m.Streams, want) } // The read tools (issue 1686) are served by the tools bundle, not by the checker's process. - // The try's tools (ADR 0501 §5; the end by hand, issue 734) beside them: mesh-controller's try calls them. - if want := []string{"checker_kept", "checker_previews", "checker_pull", "checker_status", "checker_try", "checker_try_end", "checker_try_read", "checker_verdicts"}; !reflect.DeepEqual(sorted(m.Tools), want) { + // The try's tools (ADR 0501 §5; the end by hand, issue 734) beside them: mesh-controller's try calls them. A lost check + // is asked again by hand (checker_ask_again). + if want := []string{"checker_ask_again", "checker_kept", "checker_previews", "checker_pull", "checker_status", "checker_try", "checker_try_end", "checker_try_read", "checker_verdicts"}; !reflect.DeepEqual(sorted(m.Tools), want) { t.Errorf("tools = %v, want %v", m.Tools, want) } for _, k := range []string{"logs", "emits", "claims", "provides", "requires", "defines-seats", "seat", "seats", "bus", "config"} { diff --git a/modules/mesh-delivery-checker/module.json b/modules/mesh-delivery-checker/module.json index a968770..c5fe482 100644 --- a/modules/mesh-delivery-checker/module.json +++ b/modules/mesh-delivery-checker/module.json @@ -38,6 +38,7 @@ "node-build-node" ], "tools": [ + "checker_ask_again", "checker_kept", "checker_previews", "checker_pull", From cb46ff8735cd1176402d6f12c14a5eac49a90c0f Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 28 Oct 2026 20:31:04 +0100 Subject: [PATCH 2/7] Checker and builder wait through mesh-sdk's HearEnds, so a deaf watch loses no step end in either The checker's HeardEnds moved to mesh-sdk buildrun (fix/lost-check, 6aeb401): both modules pin it, and the builder, which waits on the same Asker, now reads an ask's step at each step end of its build module too. checker_ask_again and checker_status's lost stay in the checker. Tests: one row in the builder's ask table (its watch deaf, read at a nudge, done with no new ask); counts unchanged. --- .../cmd/mesh-delivery-builder/main.go | 13 +- modules/mesh-delivery-builder/go.mod | 2 +- modules/mesh-delivery-builder/go.sum | 4 +- .../internal/build/run_test.go | 29 +++- .../cmd/mesh-delivery-checker/main.go | 5 +- .../cmd/mesh-delivery-checker/stepends.go | 83 ---------- modules/mesh-delivery-checker/go.mod | 2 +- modules/mesh-delivery-checker/go.sum | 4 +- .../internal/check/heardends.go | 144 ------------------ .../internal/check/lostask_test.go | 4 +- 10 files changed, 47 insertions(+), 243 deletions(-) delete mode 100644 modules/mesh-delivery-checker/cmd/mesh-delivery-checker/stepends.go delete mode 100644 modules/mesh-delivery-checker/internal/check/heardends.go 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()) } } From 16594feff1197a3c7932d3b93a7399e96cf2a6cd Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 28 Oct 2026 20:36:25 +0100 Subject: [PATCH 3/7] checker_ask_again: refuse while an ask of the check waits in its build seat's stream Review of fix/lost-check: an ask no node took yet has no step key, so asking again then would run the check twice (issue 1791's kind). The checker keeps the asks each running check put in a new state asks, and the tool refuses unless every one of them has a step key. Asking again under the same ask id is not done: the stream drops a second put of an id within its de-duplication window and keeps both past it, so the waiting copy is never replaced. Tests: the hand row of the lost-ask table refuses while its ask waits; counts unchanged. --- .../cmd/mesh-delivery-checker-tools/main.go | 3 +- .../cmd/mesh-delivery-checker/bus.go | 45 +++++++++++++ .../cmd/mesh-delivery-checker/main.go | 4 ++ .../internal/check/askagain.go | 63 +++++++++++++++++-- .../internal/check/checker.go | 10 +++ .../internal/check/lostask_test.go | 11 +++- .../internal/check/read.go | 4 +- .../mesh-delivery-checker/manifest_test.go | 2 +- modules/mesh-delivery-checker/module.json | 1 + 9 files changed, 134 insertions(+), 9 deletions(-) diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go index 54ac89d..660b6d5 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go @@ -267,7 +267,8 @@ func main() { PreviewState: read(stdio.State("previews")), StepStates: map[string]check.StateRead{"mesh-build-go": read(stdio.State("mesh-build-go.steps")), "mesh-build-node": read(stdio.State("mesh-build-node.steps"))}, HaveState: read(stdio.State("node-engine.have")), StepBound: stepBound(), - TurnState: read(stdio.State("mesh-delivery-gate.turns")), HearingState: read(stdio.State(buildrun.HearingState))} + TurnState: read(stdio.State("mesh-delivery-gate.turns")), HearingState: read(stdio.State(buildrun.HearingState)), + AskState: read(stdio.State(check.AsksState))} r.Runs = runsOf{read(stdio.State("mesh-delivery-indexer.modules"))} // the node-engine at the commit the mesh runs, as the checker reads it tries := stdio.State(check.TriesState) source := stdio.State("mesh-delivery-gate.source") diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/bus.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/bus.go index d81d219..13f2089 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/bus.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/bus.go @@ -9,6 +9,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/process" "mesh-delivery-checker/internal/check" ) @@ -130,3 +131,47 @@ func (m manifestReader) Manifest(ctx context.Context, digest string) (catalogue. func parseManifest(raw string) (catalogue.Manifest, error) { return catalogue.ParseManifestForInputs([]byte(raw)) } + +// awaitedKV is a bucket that may appear after the start (process.AwaitBucket) as a check.KV: a call while it is not on the +// bus fails saying so. +type awaitedKV struct{ b *process.AwaitedBucket } + +func (a awaitedKV) kv() (kvStore, error) { + kv := a.b.KV() + if kv == nil { + return kvStore{}, errors.New("the bucket is not on the bus yet") + } + return kvStore{kv}, nil +} + +func (a awaitedKV) Get(ctx context.Context, key string) ([]byte, uint64, bool, error) { + s, err := a.kv() + if err != nil { + return nil, 0, false, err + } + return s.Get(ctx, key) +} + +func (a awaitedKV) Create(ctx context.Context, key string, value []byte) (uint64, error) { + s, err := a.kv() + if err != nil { + return 0, err + } + return s.Create(ctx, key, value) +} + +func (a awaitedKV) Update(ctx context.Context, key string, value []byte, revision uint64) (uint64, error) { + s, err := a.kv() + if err != nil { + return 0, err + } + return s.Update(ctx, key, value, revision) +} + +func (a awaitedKV) Keys(ctx context.Context) ([]string, error) { + s, err := a.kv() + if err != nil { + return nil, err + } + return s.Keys(ctx) +} 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 6647277..a485138 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go @@ -69,6 +69,8 @@ var keptStates = []announce.State{ // design limit 1); the checker alone writes them (ADR 0444 clause 5). // Whether DEAD_LETTERS is heard (issue 734): read by the checker's own status tool alone. {Name: "hearing"}, + // The asks each running check put (check.AsksState): read by the checker's own checker_ask_again alone. + {Name: "asks"}, {Name: "kept", Readable: true}, {Name: "owed", Readable: true}, {Name: "previews", Readable: true}, @@ -149,7 +151,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro 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} + asksKV := process.AwaitBucket(ctx, js, conn, check.AsksBucket, "checker_ask_again refuses every check, as it cannot tell whether its ask waits in a stream", logger.Printf) ch := &check.Checker{ + Asks: awaitedKV{asksKV}, Verdicts: kvStore{buckets[delivery.CheckerVerdictsBucket]}, Owed: kvStore{buckets[delivery.CheckerOwedBucket]}, Trunks: trunks, diff --git a/modules/mesh-delivery-checker/internal/check/askagain.go b/modules/mesh-delivery-checker/internal/check/askagain.go index 3a4df17..e4a3004 100644 --- a/modules/mesh-delivery-checker/internal/check/askagain.go +++ b/modules/mesh-delivery-checker/internal/check/askagain.go @@ -14,7 +14,40 @@ import ( // has a tool). checker_ask_again ends a candidate's running verdict, whose wait judged nothing, with AskedAgainWords and // the why; the checker hears it on its watch of its verdicts (Checker.VerdictWritten), ends the check of those heads // running in it, and claims and asks it anew, a whole check (IsEnded). It is refused while a step of the candidate's pull -// requests can still run (checker_status's running_steps): that step's end judges it. +// requests can still run (checker_status's running_steps): that step's end judges it; and while an ask the check put is +// still in its build seat's stream, taken by no node (it has no step key yet): asking again then would run the check twice +// (issue 1791's kind). The checker keeps the asks each running check put in its state asks (AsksState, by verdict key). +// An ask is not put again under its own id instead: the stream de-duplicates an id within its window, so the copy waiting +// stays and the new one is dropped, and past the window both would run; nothing takes a put ask back (buildrun.Asker). + +// AsksState is the checker's state of the asks each running check put, by verdict key (CheckAsks); AsksBucket its bucket. +const ( + AsksState = "asks" + AsksBucket = delivery.CheckerModule + "_" + AsksState +) + +// CheckAsks are the asks a check put, in order, as the state asks keeps them; none once it ended. +type CheckAsks struct { + Asks []string `json:"asks"` +} + +// recordAsks keeps the asks a check put under its verdict key; nil Asks keeps nothing. A record that cannot be written +// is said: checker_ask_again then refuses the check, as it cannot tell whether its ask waits in a stream. +func (ch *Checker) recordAsks(ctx context.Context, key string, asks []string) { + if ch.Asks == nil { + return + } + if asks == nil { + asks = []string{} + } + data, err := json.Marshal(CheckAsks{Asks: asks}) + if err == nil { + _, err = putOnce(ctx, ch.Asks, key, data, func(cur []byte) bool { return string(cur) == string(data) }) + } + if err != nil { + ch.logf("%s: the asks it put cannot be kept (%v), so %s refuses it", key, err, AskAgainTool) + } +} // AskAgainTool asks a candidate's check again by hand. const AskAgainTool = "checker_ask_again" @@ -46,8 +79,8 @@ type AskedAgain struct { } // AskAgain ends the running verdict of the group's candidate as asked again by hand, saying why. It refuses a group with -// no candidate, a verdict that is not running (a pass or a failure stands until a push or a trunk move), and one a step -// of whose pull requests can still run. +// no candidate, a verdict that is not running (a pass or a failure stands until a push or a trunk move), one a step of +// whose pull requests can still run, and one whose asks are not known or one of which no node took yet. func (r Reader) AskAgain(w VerdictEntry, group, why string) (AskedAgain, error) { why = strings.TrimSpace(why) if group == "" || why == "" { @@ -78,11 +111,33 @@ func (r Reader) AskAgain(w VerdictEntry, group, why string) (AskedAgain, error) if v.State != delivery.VerdictRunning { return AskedAgain{}, fmt.Errorf("%s is %s, not running: a verdict stands until a push or a trunk move asks it again", key, v.State) } - for _, rs := range r.readSteps().running { + steps := r.readSteps() + for _, rs := range steps.running { if asksOf(c, rs.What) { return AskedAgain{}, fmt.Errorf("a step of it runs: %s %s on %s since %s (%s); its end judges it", rs.Module, rs.ID, rs.Node, rs.Since, rs.What) } } + if r.AskState.Get == nil { + return AskedAgain{}, fmt.Errorf("the checker's asks are not readable here, so whether its ask waits in a build seat's stream cannot be told") + } + kept, err := r.AskState.Get(key) + var put CheckAsks + if err == nil && len(kept) > 0 { + err = json.Unmarshal(kept, &put) + } + if err != nil || len(put.Asks) == 0 { + return AskedAgain{}, fmt.Errorf("the checker names no ask of %s (%v), so whether one waits in a build seat's stream cannot be told; asking again could run it twice", key, err) + } + taken := map[string]bool{} + for _, s := range steps.all { + taken[s.step.ID] = true + } + for _, id := range put.Asks { + if !taken[id] { + return AskedAgain{}, fmt.Errorf("its ask %s waits in its build seat's stream, taken by no node yet: asking again would run it twice; "+ + "the seats' backlog is in mesh-delivery-view.build_usage", id) + } + } words := AskedAgainWords + why ended := delivery.Verdict{Group: v.Group, Heads: v.Heads, Trunk: v.Trunk, State: delivery.VerdictFail, Failed: delivery.FailedFirst{Machine: "checker", Words: words}} diff --git a/modules/mesh-delivery-checker/internal/check/checker.go b/modules/mesh-delivery-checker/internal/check/checker.go index aa9a8c7..2dd21ce 100644 --- a/modules/mesh-delivery-checker/internal/check/checker.go +++ b/modules/mesh-delivery-checker/internal/check/checker.go @@ -79,6 +79,8 @@ type Checker struct { } // Bound is how long one check may run; none when 0 (DefaultBound). Bound time.Duration + // Asks keeps the asks each running check put, by verdict key (AsksState), for checker_ask_again; nil keeps none. + Asks KV // Link is the base of the address a status points at; the verdict key follows. Link string // Announce says a status was written (EventOwedWritten); nil says nothing. @@ -715,6 +717,14 @@ func (ch *Checker) try(ctx, run context.Context, key string, cs *Case) { return } began := ch.now() + if cs.Watch == nil && ch.Asks != nil { + cs.Watch = func(_ string, step *delivery.Step) { + if step == nil { // put: kept before its wait starts + ch.recordAsks(ctx, cs.Key, cs.Asks()) + } + } + defer ch.recordAsks(ctx, cs.Key, nil) + } ch.writeOwed(ctx, cs, delivery.Verdict{State: delivery.VerdictRunning}, Outcome{}) out := ch.judge(run, cs) // An ask the build seat gave up on, or whose step the build module's next process ended because it was replaced under diff --git a/modules/mesh-delivery-checker/internal/check/lostask_test.go b/modules/mesh-delivery-checker/internal/check/lostask_test.go index 3539b8c..2df4431 100644 --- a/modules/mesh-delivery-checker/internal/check/lostask_test.go +++ b/modules/mesh-delivery-checker/internal/check/lostask_test.go @@ -282,6 +282,8 @@ func TestACheckWhoseAskJudgedNothingIsAskedAgainOrRead(t *testing.T) { asked = func() int { return len(tr.Asks()) } } r := newRig(t, b.ex) + asksKept := newMemKV() + r.ch.Asks = asksKept cand := candidateOf(t, "feat/x", pullOf(repoA, 7, head1)) key := keyOf(t, "feat/x", cand) r.ch.Candidate(ctx, "feat%2Fx", cand) @@ -301,8 +303,13 @@ func TestACheckWhoseAskJudgedNothingIsAskedAgainOrRead(t *testing.T) { cands := newMemKV() ckey, _ := delivery.CandidatesKey("feat/x") _, _ = cands.Create(ctx, ckey, cand) - rd := Reader{CandidateState: kvRead(cands), VerdictState: kvRead(r.verdicts), - StepStates: map[string]StateRead{buildrun.GoBuildModule: tr.Steps(true)}} + none := StateRead{Keys: func() ([]string, error) { return nil, nil }, Get: func(string) (json.RawMessage, error) { return nil, nil }} + rd := Reader{CandidateState: kvRead(cands), VerdictState: kvRead(r.verdicts), AskState: kvRead(asksKept), + StepStates: map[string]StateRead{buildrun.GoBuildModule: none}} // no node took its ask yet + if _, err := rd.AskAgain(asVerdicts{r.verdicts}, "feat/x", "its watch went deaf"); err == nil || !strings.Contains(err.Error(), "waits in its build seat's stream") { + t.Errorf("asked again while its ask waits in the stream (issue 1791's kind): %v", err) + } + rd.StepStates[buildrun.GoBuildModule] = tr.Steps(true) if _, err := rd.AskAgain(asVerdicts{r.verdicts}, "feat/x", "its watch went deaf"); err == nil || !strings.Contains(err.Error(), "a step of it runs") { t.Errorf("asked again while its step runs: %v", err) } diff --git a/modules/mesh-delivery-checker/internal/check/read.go b/modules/mesh-delivery-checker/internal/check/read.go index f5d82e5..250b6b9 100644 --- a/modules/mesh-delivery-checker/internal/check/read.go +++ b/modules/mesh-delivery-checker/internal/check/read.go @@ -47,7 +47,9 @@ type Reader struct { Runs interface{ CommitOf(module string) string } // HearingState is the checker's state hearing: whether it hears DEAD_LETTERS (issue 734). HearingState StateRead - Now func() time.Time + // AskState is the checker's state asks: the asks each running check put (AsksState), which checker_ask_again reads. + AskState StateRead + Now func() time.Time } // VerdictRead is a verdict as stored with the full report of the status of each pull request it judged (Report), when the diff --git a/modules/mesh-delivery-checker/manifest_test.go b/modules/mesh-delivery-checker/manifest_test.go index a56a841..404988d 100644 --- a/modules/mesh-delivery-checker/manifest_test.go +++ b/modules/mesh-delivery-checker/manifest_test.go @@ -93,7 +93,7 @@ func TestManifestDeclarations(t *testing.T) { // 1678/AC1 t.Errorf("module = %q", m.Module) } // delivery.CheckerKeptHistory, CheckerOwedHistory, PreviewsHistory, CheckerVerdictsHistory; the names follow the SDK's buckets. - if want := []state{{Name: "hearing", History: 1}, {Name: "kept", History: 1}, {Name: "owed", History: 5}, {Name: "previews", History: 5}, {Name: "tries", History: 1}, {Name: "verdicts", History: 5}}; !reflect.DeepEqual(m.State, want) { + if want := []state{{Name: "asks", History: 1}, {Name: "hearing", History: 1}, {Name: "kept", History: 1}, {Name: "owed", History: 5}, {Name: "previews", History: 5}, {Name: "tries", History: 1}, {Name: "verdicts", History: 5}}; !reflect.DeepEqual(m.State, want) { t.Errorf("state = %+v, want %+v", m.State, want) } if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-node.steps", "mesh-build-node.toolchains", "mesh-controller.assignments", diff --git a/modules/mesh-delivery-checker/module.json b/modules/mesh-delivery-checker/module.json index c5fe482..6d12e3b 100644 --- a/modules/mesh-delivery-checker/module.json +++ b/modules/mesh-delivery-checker/module.json @@ -10,6 +10,7 @@ { "name": "checks", "events": ["preview.**", "owed.**"] } ], "state": [ + { "name": "asks", "history": 1 }, { "name": "hearing", "history": 1 }, { "name": "kept", "history": 1 }, { "name": "owed", "history": 5 }, From a58a73bb55932c99aa74c25e33ec567e062fde82 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 28 Oct 2026 20:35:22 +0100 Subject: [PATCH 4/7] =?UTF-8?q?Checker=20integration=20tests:=20give=20the?= =?UTF-8?q?=20bed's=20node=20a=20record=20in=20the=20have,=20so=20the=20na?= =?UTF-8?q?mes=20check=20finds=20a=20name=20to=20ask=20with=20(ADR=200567?= =?UTF-8?q?=20=C2=A712,=20issue=201126)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Since 9db9499 the checker asks no check while no node record is on the bus; the bed wrote none, so eight integration tests failed on main (PR #101's check). The fake build module now records the names an ask carries, and the first test asserts n1 is named. --- .../mesh-delivery-checker/integration_test.go | 32 ++++++++++++++++++- 1 file changed, 31 insertions(+), 1 deletion(-) diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go index 6b39ca3..be4d4da 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go @@ -89,6 +89,8 @@ type fakeBuildModules struct { // beside is, by workspace, the siblings the fetch placed in it as ` `, sorted: a digest echoes them // (1929/AC2; mesh-sdk buildstep.besideOf), or the level runs whole and keeps nothing. beside map[string][]string + // names are the nodes' names the last check ask carried, for the names check (ADR 0567 §12). + names []string } func (f *fakeBuildModules) count(tool string) int { @@ -191,7 +193,14 @@ func (f *fakeBuildModules) serve(t *testing.T, module string, kept jetstream.Key } p := &buildstep.Puller{Source: &buildstep.JetStreamAsks{JS: f.js, Language: language}, Language: language, Node: "n1", Put: put, Capacity: func() int { return 2 }, PullWait: 200 * time.Millisecond, - Run: func(ctx context.Context, a delivery.StepAsk) (any, error) { return buildstep.RunAsk(ctx, steps, a) }} + Run: func(ctx context.Context, a delivery.StepAsk) (any, error) { + if a.Check != nil { + f.mu.Lock() + f.names = a.Check.NodeNames + f.mu.Unlock() + } + return buildstep.RunAsk(ctx, steps, a) + }} go func() { _ = p.Pull(f.ctx) }() } @@ -334,6 +343,7 @@ func startBed(t *testing.T) *bed { t.Fatal(err) } b.setOnline(t, "n1", true) + b.setNode(t, "n1") for _, module := range []string{"mesh-build-go", "mesh-build-node"} { b.setHealth(t, "n1", module) } @@ -366,6 +376,20 @@ func (b *bed) setOnline(t *testing.T, node string, online bool) { } } +// setNode writes the node's record in the node-engine's have, as the node-engine does, so the checker names it in each +// check ask (ADR 0567 §12); with no record the checker asks no check. +func (b *bed) setNode(t *testing.T, node string) { + t.Helper() + data, err := delivery.EncodeNodeRecord(delivery.NodeRecord{V: 1, Node: node}) + if err != nil { + t.Fatal(err) + } + key, _ := delivery.HaveNodeKey(node) + if _, err := b.buckets[delivery.HaveBucket].Put(b.ctx, key, data); err != nil { + t.Fatal(err) + } +} + func (b *bed) setHealth(t *testing.T, node, module string) { t.Helper() data, err := delivery.EncodeHave(delivery.Have{Phase: delivery.PhaseRunning, Health: delivery.HealthOK}) @@ -493,6 +517,12 @@ func TestCheckerJudgesACandidateAndOwesItsStatus(t *testing.T) { // N6/AC5, 1300 if v.Composition == nil || v.Composition.State != delivery.CompositionNotRun || len(v.Machines) != 1 || v.Machines[0] != "n1" { t.Errorf("composition %+v machines %v", v.Composition, v.Machines) } + b.fake.mu.Lock() + names := b.fake.names + b.fake.mu.Unlock() + if len(names) != 1 || names[0] != "n1" { + t.Errorf("the check ask named the nodes %v, want [n1] from the have's node record", names) + } o := b.owed(t, 7, head1, "success") if o.Context != delivery.StatusContextCheck || !strings.HasPrefix(o.Text, "pass 111111111111 on aaaaaaaaaaaa") || !strings.Contains(o.Text, "unit 2 run") { t.Errorf("owed = %+v", o) From ca74639a7445ed0cbae76609c689ad735437638c Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 28 Oct 2026 20:37:07 +0100 Subject: [PATCH 5/7] Checker integration test: the checker says its seventh state, asks, as its own The new state asks (checker_ask_again's record of each running check's asks) is kept and not readable, as hearing is. --- .../cmd/mesh-delivery-checker/integration_test.go | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go index be4d4da..a3bf800 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go @@ -827,7 +827,8 @@ func TestTheCheckerSaysOwedReadableOnTheBus(t *testing.T) { // R2-1 break } if said.Module != "mesh-delivery-checker" || said.Node != "n1" || !said.Keeps("owed") || !said.Readable("owed") || - !said.Readable("previews") || !said.Readable("verdicts") || !said.Readable("kept") || !said.Readable("tries") || said.Readable("hearing") || len(said.States) != 6 { - t.Fatalf("the checker said %+v, want its six states with kept, owed, previews, tries and verdicts readable, tries by the delivery view (1929/AC4; 1887 A5b; build_usage)", said) + !said.Readable("previews") || !said.Readable("verdicts") || !said.Readable("kept") || !said.Readable("tries") || said.Readable("hearing") || + !said.Keeps("asks") || said.Readable("asks") || len(said.States) != 7 { + t.Fatalf("the checker said %+v, want its seven states, asks and hearing its own, with kept, owed, previews, tries and verdicts readable, tries by the delivery view (1929/AC4; 1887 A5b; build_usage)", said) } } From 2007a79c45c31c9151d689c174822570071801f5 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 28 Oct 2026 20:38:31 +0100 Subject: [PATCH 6/7] Checker and builder pin mesh-sdk f91340f, which opens a silent steps watch again --- modules/mesh-delivery-builder/go.mod | 2 +- modules/mesh-delivery-builder/go.sum | 4 ++-- modules/mesh-delivery-checker/go.mod | 2 +- modules/mesh-delivery-checker/go.sum | 4 ++-- 4 files changed, 6 insertions(+), 6 deletions(-) diff --git a/modules/mesh-delivery-builder/go.mod b/modules/mesh-delivery-builder/go.mod index a70a9fd..b8b03ce 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.20261028192930-6aeb40109529 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028193502-f91340fe82ef 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 88f3a59..cbc4905 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.20261028192930-6aeb40109529 h1:Xx5mcDPWcRuPelUxdjvwkmVcfyGww0rRajA9JjFA/hk= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028192930-6aeb40109529/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028193502-f91340fe82ef h1:ggwHXCvAOhBCKAY+ErehKEhu/IQTS4poJBRDokGYnDc= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028193502-f91340fe82ef/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/go.mod b/modules/mesh-delivery-checker/go.mod index a1dad59..0eb617e 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.20261028192930-6aeb40109529 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028193502-f91340fe82ef 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 88f3a59..cbc4905 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.20261028192930-6aeb40109529 h1:Xx5mcDPWcRuPelUxdjvwkmVcfyGww0rRajA9JjFA/hk= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028192930-6aeb40109529/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028193502-f91340fe82ef h1:ggwHXCvAOhBCKAY+ErehKEhu/IQTS4poJBRDokGYnDc= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028193502-f91340fe82ef/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= From e7888024a8264f5d1e6ac46a2e9cc0d112ff60be Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 28 Oct 2026 20:39:50 +0100 Subject: [PATCH 7/7] Checker: delete a check's asks key at its end, so the state asks holds only the checks that run Re-review of fix/lost-check: writing an empty list at the end kept one key for every verdict key ever checked. Tests: every row of the lost-ask table proves the key gone at the end; one row more (deaf after its build module replaced it, asked again by hand); counts unchanged. --- .../cmd/mesh-delivery-checker/bus.go | 8 ++++++++ .../internal/check/askagain.go | 19 +++++++++++++++---- .../internal/check/checker.go | 4 ++-- .../internal/check/fakes_test.go | 8 ++++++++ .../internal/check/lostask_test.go | 4 ++++ 5 files changed, 37 insertions(+), 6 deletions(-) diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/bus.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/bus.go index 13f2089..a5453a6 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/bus.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/bus.go @@ -175,3 +175,11 @@ func (a awaitedKV) Keys(ctx context.Context) ([]string, error) { } return s.Keys(ctx) } + +func (a awaitedKV) Delete(ctx context.Context, key string) error { + s, err := a.kv() + if err != nil { + return err + } + return s.Delete(ctx, key) +} diff --git a/modules/mesh-delivery-checker/internal/check/askagain.go b/modules/mesh-delivery-checker/internal/check/askagain.go index e4a3004..7540bbe 100644 --- a/modules/mesh-delivery-checker/internal/check/askagain.go +++ b/modules/mesh-delivery-checker/internal/check/askagain.go @@ -26,20 +26,31 @@ const ( AsksBucket = delivery.CheckerModule + "_" + AsksState ) -// CheckAsks are the asks a check put, in order, as the state asks keeps them; none once it ended. +// CheckAsks are the asks a check put, in order, as the state asks keeps them while it runs; its end deletes the key. type CheckAsks struct { Asks []string `json:"asks"` } +// AsksKV is the state asks: a KV whose key a check's end deletes, so the state holds only the checks that run. +type AsksKV interface { + KV + Delete(ctx context.Context, key string) error +} + +// forgetAsks deletes the asks a check put at its end; one that cannot be deleted is said, and the next check of the same +// heads writes the key anew. +func (ch *Checker) forgetAsks(ctx context.Context, key string) { + if err := ch.Asks.Delete(ctx, key); err != nil { + ch.logf("%s: the asks it put could not be forgotten at its end: %v", key, err) + } +} + // recordAsks keeps the asks a check put under its verdict key; nil Asks keeps nothing. A record that cannot be written // is said: checker_ask_again then refuses the check, as it cannot tell whether its ask waits in a stream. func (ch *Checker) recordAsks(ctx context.Context, key string, asks []string) { if ch.Asks == nil { return } - if asks == nil { - asks = []string{} - } data, err := json.Marshal(CheckAsks{Asks: asks}) if err == nil { _, err = putOnce(ctx, ch.Asks, key, data, func(cur []byte) bool { return string(cur) == string(data) }) diff --git a/modules/mesh-delivery-checker/internal/check/checker.go b/modules/mesh-delivery-checker/internal/check/checker.go index 2dd21ce..da952d3 100644 --- a/modules/mesh-delivery-checker/internal/check/checker.go +++ b/modules/mesh-delivery-checker/internal/check/checker.go @@ -80,7 +80,7 @@ type Checker struct { // Bound is how long one check may run; none when 0 (DefaultBound). Bound time.Duration // Asks keeps the asks each running check put, by verdict key (AsksState), for checker_ask_again; nil keeps none. - Asks KV + Asks AsksKV // Link is the base of the address a status points at; the verdict key follows. Link string // Announce says a status was written (EventOwedWritten); nil says nothing. @@ -723,7 +723,7 @@ func (ch *Checker) try(ctx, run context.Context, key string, cs *Case) { ch.recordAsks(ctx, cs.Key, cs.Asks()) } } - defer ch.recordAsks(ctx, cs.Key, nil) + defer ch.forgetAsks(ctx, cs.Key) } ch.writeOwed(ctx, cs, delivery.Verdict{State: delivery.VerdictRunning}, Outcome{}) out := ch.judge(run, cs) diff --git a/modules/mesh-delivery-checker/internal/check/fakes_test.go b/modules/mesh-delivery-checker/internal/check/fakes_test.go index 2965b72..d53f2a1 100644 --- a/modules/mesh-delivery-checker/internal/check/fakes_test.go +++ b/modules/mesh-delivery-checker/internal/check/fakes_test.go @@ -60,3 +60,11 @@ func (m *memKV) Keys(context.Context) ([]string, error) { sort.Strings(keys) return keys, nil } + +func (m *memKV) Delete(_ context.Context, key string) error { + m.mu.Lock() + defer m.mu.Unlock() + delete(m.vals, key) + delete(m.revs, key) + return nil +} diff --git a/modules/mesh-delivery-checker/internal/check/lostask_test.go b/modules/mesh-delivery-checker/internal/check/lostask_test.go index 2df4431..ddf4d33 100644 --- a/modules/mesh-delivery-checker/internal/check/lostask_test.go +++ b/modules/mesh-delivery-checker/internal/check/lostask_test.go @@ -265,6 +265,7 @@ func TestACheckWhoseAskJudgedNothingIsAskedAgainOrRead(t *testing.T) { {name: "replaced under every ask", restarts: LostAskTries, state: delivery.VerdictFail, asks: LostAskTries}, {name: "its watch deaf, a step of its build module ends", deaf: 1, how: "nudge", state: delivery.VerdictPass, asks: 1}, {name: "its watch deaf, asked again by hand", deaf: 1, how: "hand", state: delivery.VerdictPass, asks: 2}, + {name: "its watch deaf, asked again by hand after its build module replaced it", deaf: 1, restarts: 1, how: "hand", state: delivery.VerdictPass, asks: 2}, } { t.Run(row.name, func(t *testing.T) { b := newBench(t) @@ -329,6 +330,9 @@ func TestACheckWhoseAskJudgedNothingIsAskedAgainOrRead(t *testing.T) { if v.State != row.state || asked() != row.asks { t.Errorf("verdict %s (%s) after %d asks, want %s after %d", v.State, v.Failed.Words, asked(), row.state, row.asks) } + if _, _, kept, _ := asksKept.Get(ctx, key); kept { + t.Errorf("the asks of the ended check are still kept under %s", key) + } if row.state == delivery.VerdictFail && (!IsLostAsk(v) || !IsNodesFailure(v)) { t.Errorf("a check that judged nothing every time = %q, want lost and judged again when the nodes change", v.Failed.Words) }