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.

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

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)
+}

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,

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}}

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

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)
 				}

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

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

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