From 16594feff1197a3c7932d3b93a7399e96cf2a6cd Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 28 Oct 2026 20:36:25 +0100 Subject: [PATCH] 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 },