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",