Merge pull request 'The checker and the builder use it; checker_ask_again asks a lost check again; the checker's integration bed writes a node record (issue 734's kind)' (#103) from fix/lost-check into main

This commit was merged in pull request #103.
This commit is contained in:
2026-10-28 20:41:01 +00:00
19 changed files with 761 additions and 108 deletions
@@ -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},
+1 -1
View File
@@ -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.20261028193502-f91340fe82ef
github.com/nats-io/nats-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0
)
+2 -2
View File
@@ -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.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=
@@ -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
}
@@ -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 {
@@ -249,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")
@@ -277,7 +296,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)
}
@@ -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 {
@@ -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,55 @@ 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)
}
func (a awaitedKV) Delete(ctx context.Context, key string) error {
s, err := a.kv()
if err != nil {
return err
}
return s.Delete(ctx, key)
}
@@ -89,6 +89,8 @@ type fakeBuildModules struct {
// beside is, by workspace, the siblings the fetch placed in it as `<owner/repo> <commit>`, 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)
@@ -797,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)
}
}
@@ -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},
@@ -144,9 +146,14 @@ 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
// (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}
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,
@@ -331,6 +338,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 {
+1 -1
View File
@@ -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.20261028193502-f91340fe82ef
github.com/nats-io/nats-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0
)
+2 -2
View File
@@ -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.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=
@@ -0,0 +1,198 @@
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; 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 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
}
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"
// 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), 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 == "" {
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)
}
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}}
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)
}
@@ -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 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.
@@ -328,7 +330,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
@@ -711,10 +717,20 @@ 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.forgetAsks(ctx, cs.Key)
}
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 +740,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) {
@@ -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
}
@@ -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).
@@ -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,206 @@ 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},
{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},
} {
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 *buildrun.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 = &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()) }
}
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)
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)
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)
}
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 _, _, 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)
}
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")
}
}
})
}
}
@@ -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
@@ -306,6 +308,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 +335,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 +401,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 +485,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 +510,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 `<repository>@<head12>`
// (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
@@ -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",
@@ -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"} {
@@ -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 },
@@ -38,6 +39,7 @@
"node-build-node"
],
"tools": [
"checker_ask_again",
"checker_kept",
"checker_previews",
"checker_pull",