diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go index 4930788..3bb3fd1 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go @@ -15,6 +15,7 @@ import ( stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/delivery/buildrun" "mesh-delivery-builder/internal/build" ) @@ -31,7 +32,8 @@ func Tools(r build.StuckReader, askAgain func(key string, value []byte) error, q Description: "Every build key that waits (building or queued), oldest first: its module, fingerprint, step words and since when, and each ask it " + "waits on by language with its id, when it was put, its asking (1 for the first; the builder asks an ask again when it can no longer be " + "answered, at most 3 times) and what its build module's steps hold of it (none, or its state on its node since its start). The " + - "seat workers' delivery counts are not readable by the builder. With module, that module's keys only. Read-only.", + "seat workers' delivery counts are not readable by the builder. dead_letters says whether the builder hears the asks the build seats give up " + + "on (issue 734), and why not: heard, a given-up ask is asked again at once; deaf, at its check bound. With module, that module's keys only. Read-only.", Input: map[string]any{"type": "object", "properties": map[string]any{ "module": map[string]any{"type": "string", "description": "only this module's keys"}}}, Run: func(args map[string]any) (any, error) { return r.Stuck(str(args, "module")) }}, @@ -166,7 +168,8 @@ func main() { steps[bm] = read(stdio.State(bm + ".steps")) } reasks := stdio.State(build.ReAsksState) - r := build.StuckReader{Logf: func(f string, a ...any) { fmt.Fprintf(os.Stderr, "[mesh-delivery-builder-tools] "+f+"\n", a...) }, Builds: read(stdio.State("builds")), ReAsks: read(reasks), Steps: steps} + r := build.StuckReader{Logf: func(f string, a ...any) { fmt.Fprintf(os.Stderr, "[mesh-delivery-builder-tools] "+f+"\n", a...) }, Builds: read(stdio.State("builds")), ReAsks: read(reasks), Steps: steps, + Hearing: read(stdio.State(buildrun.HearingState))} put := func(key string, value []byte) error { _, err := reasks.Put(key, json.RawMessage(value)) return err diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go index 81c973d..7d71872 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -149,6 +149,16 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // Each build module's part of a build is an ask on its build seat's stream, pulled by a build node with a free slot // (novox/hq ADR 0525): the builder chooses no node, so it reads neither presence nor have and holds no capacity. asks := buildrun.JetStreamAsks{JS: js} + // The asks the build seats gave up on (issue 734): heard from DEAD_LETTERS, each ends the wait for its step at once, + // and the lane asks its part again (Builder.waitOnce) instead of waiting for the check bound. + // Heard in the background, and again at a reconnect or an ask while it is not; whether it is heard is kept in the + // state hearing, which builder_stuck says (dead_letters). + dead := &buildrun.DeadLetters{} + hearingKV := process.OptionalBucket(ctx, js, delivery.BuilderModule+"_"+buildrun.HearingState, "builder_stuck cannot say whether DEAD_LETTERS is heard", logger.Printf) + 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() }) // 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. account, err := process.ReadAccount(path) @@ -162,7 +172,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: asks, Bound: bound}, + Asks: buildrun.Asker{Publisher: asks, Waiter: buildrun.LostAskWaiter{Inner: asks, Dead: dead, Hearing: hearing}, Bound: bound}, Held: heldOnTheBus{js}, // An ask that can no longer be answered is asked again (issue 1945), read at the step bound: a context's bound. Gone: heldOnTheBus{js}, diff --git a/modules/mesh-delivery-builder/go.mod b/modules/mesh-delivery-builder/go.mod index fb89b4d..b206980 100644 --- a/modules/mesh-delivery-builder/go.mod +++ b/modules/mesh-delivery-builder/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-builder go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026134035-21252194d530 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-builder/go.sum b/modules/mesh-delivery-builder/go.sum index 2dfa4fd..9a6ce0a 100644 --- a/modules/mesh-delivery-builder/go.sum +++ b/modules/mesh-delivery-builder/go.sum @@ -1,7 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024020722-34f210e5eba1 h1:fzOkrz9CSp3mEjeV7NfZx8yoivI1sD1AOtIPtyyF8RI= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024020722-34f210e5eba1/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585 h1:7QoFH5qjXRpx5M4uPdYsVJ5qKl2082gq8n0lFxhoNIg= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026134035-21252194d530 h1:NwCb+gOh0UYqlyG6hG2/jffE/Gd0c7YWd4iJlaCB0lU= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026134035-21252194d530/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-builder/internal/build/askagain.go b/modules/mesh-delivery-builder/internal/build/askagain.go index e048541..1b03da9 100644 --- a/modules/mesh-delivery-builder/internal/build/askagain.go +++ b/modules/mesh-delivery-builder/internal/build/askagain.go @@ -12,6 +12,7 @@ import ( "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" ) // **An ask that can no longer be answered is asked again** (novox/hq issue 1945). A build seat's stream gives an ask up @@ -20,7 +21,8 @@ import ( // restart take over the same id and wait again. The builder notices it at the check bound (CheckAfter, the step bound), // a context's bound and no timer of its own: it asks Gone, which reads the bus, and asks such an ask again under a new // id, written into the key, said in its log and as the event build.asked-again, at most Askings times; then the key is -// failed with why. A person asks the same by hand with the tool builder_ask_again (AskAgain). +// failed with why. An ask DEAD_LETTERS says the seat gave up (buildrun.LostAskWaiter, issue 734) is asked again at once. +// A person asks the same by hand with the tool builder_ask_again (AskAgain). // DefaultAskings is how many times a build module's part of a build is asked, the first included, when Builder.Askings is 0. const DefaultAskings = 3 @@ -315,6 +317,11 @@ func (b *Builder) waitOnce(ctx context.Context, ask delivery.StepAsk, again <-ch defer stop() } step, err := b.Asks.Wait(wait, ask) + if err == nil && buildrun.IsLost(step) { + // The build seat gave the ask up (issue 734): it can no longer be answered, so it is asked again now, with the + // words of why, not at the check bound. + return step, step.Reason, errAgain + } if err == nil || ctx.Err() != nil { return step, "", err } diff --git a/modules/mesh-delivery-builder/internal/build/run_test.go b/modules/mesh-delivery-builder/internal/build/run_test.go index c6ba95a..c8fcd3e 100644 --- a/modules/mesh-delivery-builder/internal/build/run_test.go +++ b/modules/mesh-delivery-builder/internal/build/run_test.go @@ -690,12 +690,14 @@ func TestABuildAskWaitingInItsStreamIsNotBounded(t *testing.T) { step string 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) puts int done bool asking int // the asking the event says, 0 when none is said }{ {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: "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}, } { @@ -714,7 +716,11 @@ func TestABuildAskWaitingInItsStreamIsNotBounded(t *testing.T) { r.st.vals[r.key] = data } gone := &goneAsks{} - if !c.byHand { + dead := &buildrun.DeadLetters{} + 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 { 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" @@ -724,6 +730,10 @@ func TestABuildAskWaitingInItsStreamIsNotBounded(t *testing.T) { r.b.Gone, r.b.CheckAfter = gone, 5*time.Millisecond } r.b.Module(context.Background(), "m", moduleValue(t, "modules/m")) + if c.dead { + dead.Heard(buildrun.GivenUp{AskID: waitForAsks(t, r, 1)[0].ID, Stream: "SEAT_NODE_BUILD_GO", Deliveries: 5, + At: time.Date(2026, 10, 26, 12, 26, 40, 0, time.UTC)}) + } if c.byHand { waitForAsks(t, r, 1) for deadline := time.Now().Add(2 * time.Second); len(r.b.AskAgain(r.key, "stuck since 19:06")) == 0; time.Sleep(time.Millisecond) { @@ -757,13 +767,16 @@ 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 && (len(gone.dropped) != 1 || gone.dropped[0] != first) { + if gone.mu.Lock(); c.done && !c.byHand && !c.dead && (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() if c.asking > 0 && (len(said) != 1 || said[0].Asking != c.asking || said[0].New == said[0].Old || said[0].Key != r.key || said[0].Why == "") { t.Errorf("said %+v, want once at asking %d", said, c.asking) } + if c.dead && (len(said) != 1 || !strings.Contains(said[0].Why, buildrun.LostWords+" after 5 deliveries at 2026-10-26T12:26:40Z")) { + t.Errorf("said %+v, want the ask asked again saying the seat gave it up", said) + } }) } t.Run("what the bus says of an ask", func(t *testing.T) { // 1945 @@ -836,12 +849,17 @@ func TestABuildAskWaitingInItsStreamIsNotBounded(t *testing.T) { running, _ := json.Marshal(delivery.Step{V: delivery.Version, ID: "mvo53jcm70b5f8", Tool: "delivery ask", Node: "g14", State: delivery.StepRunning, Started: at}) _, reask, _ := NewReAsk("m", fp1, "stuck", "shanks.lead", at) r := StuckReader{Builds: state(map[string]string{key: string(building), done: string(finished)}), ReAsks: state(map[string]string{key: string(reask)}), - Steps: map[string]StateRead{buildrun.GoBuildModule: state(map[string]string{}), ImageBuildModule: state(map[string]string{"g14.mvo53jcm70b5f8": string(running)})}, - Now: func() time.Time { return at.Add(time.Hour) }} + Steps: map[string]StateRead{buildrun.GoBuildModule: state(map[string]string{}), ImageBuildModule: state(map[string]string{"g14.mvo53jcm70b5f8": string(running)})}, + Hearing: state(map[string]string{buildrun.HearingKey: `{"heard":false,"streams":["SEAT_NODE_BUILD_GO"],"why":"DEAD_LETTERS is not heard: SEAT_NODE_BUILD_GO: permissions violation","at":"2026-10-26T13:00:00Z"}`}), + Now: func() time.Time { return at.Add(time.Hour) }} out, err := r.Stuck("") if err != nil { t.Fatal(err) } + // Issue 734, review round 1: a builder that does not hear DEAD_LETTERS says so, with why. + if dl := out["dead_letters"].(map[string]any); dl["heard"] != false || !strings.Contains(dl["why"].(string), "permissions violation") { + t.Errorf("dead_letters = %v, want not heard, saying why", dl) + } stuck := out["stuck"].([]Stuck) if len(stuck) != 1 || stuck[0].Key != key || stuck[0].Age != "1h0m0s" || stuck[0].NoneStep != 1 || stuck[0].ReAsk == nil || stuck[0].ReAsk.Why != "stuck" { t.Fatalf("stuck = %+v, want the building key alone, an hour old, one ask with no step, with its ask again", stuck) diff --git a/modules/mesh-delivery-builder/internal/build/stuck.go b/modules/mesh-delivery-builder/internal/build/stuck.go index 41184c4..50ed667 100644 --- a/modules/mesh-delivery-builder/internal/build/stuck.go +++ b/modules/mesh-delivery-builder/internal/build/stuck.go @@ -106,7 +106,9 @@ type StuckReader struct { ReAsks StateRead // Steps are the build modules' steps states, by build module. Steps map[string]StateRead - Now func() time.Time + // Hearing is the builder's state hearing: whether it hears DEAD_LETTERS (issue 734); without it none is said. + Hearing StateRead + Now func() time.Time // Logf, when set, gets the one line a call logs: the values it read and the time it took. Logf func(format string, args ...any) } @@ -149,6 +151,20 @@ func parallel(n int, read func(i int) error) error { } // Stuck is every key building or queued with its asks, oldest first; with module, that module's only. +// deadLetters says whether the builder hears the asks the build seats give up on (issue 734), and why not: deaf, a lane +// whose ask is given up waits for its check bound (buildrun.HearingHealth over the kept hearing). +func (r StuckReader) deadLetters() map[string]any { + if r.Hearing.Get == nil { + return map[string]any{"heard": false, "why": "the builder's state hearing is not readable here"} + } + raw, _, err := r.Hearing.Get(buildrun.HearingKey) + if err != nil { + return map[string]any{"heard": false, "why": "the builder's state hearing cannot be read: " + err.Error()} + } + a := buildrun.HearingHealth(raw) + return map[string]any{"heard": a.Healthy, "why": a.Why} +} + func (r StuckReader) Stuck(module string) (map[string]any, error) { started := time.Now() reads := 0 @@ -268,6 +284,7 @@ func (r StuckReader) Stuck(module string) (map[string]any, error) { }) answer := map[string]any{"stuck": out, "count": len(out), "delivery_counts": "not readable by the builder: the build seats' workers' information is granted to the delivery view alone (its backlog); an ask given up on is told by the builder at its check bound and asked again"} + answer["dead_letters"] = r.deadLetters() if cut { answer["cut"], answer["note"] = true, fmt.Sprintf("read %d of %d build keys in key order and stopped; give a module to read its keys alone", MaxReads, total) } diff --git a/modules/mesh-delivery-builder/manifest_test.go b/modules/mesh-delivery-builder/manifest_test.go index b629f5c..6679a16 100644 --- a/modules/mesh-delivery-builder/manifest_test.go +++ b/modules/mesh-delivery-builder/manifest_test.go @@ -87,7 +87,7 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1 if m.Module != "mesh-delivery-builder" { t.Errorf("module = %q", m.Module) } - if want := []state{{Name: "builds", History: 5}, {Name: "instances", History: 1}, {Name: "queues", History: 1}, {Name: "reasks", History: 1}}; !reflect.DeepEqual(m.State, want) { // delivery.BuilderBuildsHistory + if want := []state{{Name: "builds", History: 5}, {Name: "hearing", History: 1}, {Name: "instances", History: 1}, {Name: "queues", History: 1}, {Name: "reasks", History: 1}}; !reflect.DeepEqual(m.State, want) { // delivery.BuilderBuildsHistory t.Errorf("state = %+v, want %+v", m.State, want) } // 1887/A4c: the builder chooses no node, so it reads neither presence nor have, invokes no build module's tool and diff --git a/modules/mesh-delivery-builder/module.json b/modules/mesh-delivery-builder/module.json index 94426fc..e4d2859 100644 --- a/modules/mesh-delivery-builder/module.json +++ b/modules/mesh-delivery-builder/module.json @@ -20,6 +20,7 @@ ], "state": [ { "name": "builds", "history": 5 }, + { "name": "hearing", "history": 1 }, { "name": "instances", "history": 1 }, { "name": "queues", "history": 1 }, { "name": "reasks", "history": 1 } 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 5972f6f..8e5ec44 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 @@ -12,6 +12,7 @@ import ( stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/delivery/buildrun" "mesh-delivery-checker/internal/check" ) @@ -76,6 +77,13 @@ func (r runsOf) CommitOf(module string) string { type stateStore struct { check.StateRead create func(key string, value []byte) error + entry func(key string) (json.RawMessage, uint64, error) + update func(key string, value []byte, revision uint64) error +} + +func (s stateStore) Entry(key string) (json.RawMessage, uint64, error) { return s.entry(key) } +func (s stateStore) Update(key string, value []byte, revision uint64) error { + return s.update(key, value, revision) } func (s stateStore) Keys() ([]string, error) { return s.StateRead.Keys() } @@ -85,11 +93,12 @@ func (s stateStore) Create(key string, value []byte) error { return s.create(k // Trying is what the try's tools work on: the tries, and the gate's trunk heads. type Trying struct { Tries check.TryStore + Ender check.TryEnder Trunks func() (map[string]string, error) Now func() time.Time } -// TryTools are the try's two tools (ADR 0501 §5): ask one, read one. mesh-controller's try calls them. +// TryTools are the try's tools (ADR 0501 §5): ask one, read one, and end one by hand (issue 734). mesh-controller's try calls them. func TryTools(tr Trying) []stdio.Tool { str := func(args map[string]any, k string) string { s, _ := args[k].(string); return s } return []stdio.Tool{ @@ -97,7 +106,8 @@ func TryTools(tr Trying) []stdio.Tool { Description: "Try a pushed head before its pull request is opened: the checker judges it as it judges a pull request's head, merged onto its trunk's head, " + "its unit and integration levels through the build modules, and owes the forge nothing. Answers the try's id at once; " + check.TryReadTool + " reads " + "where it stands. A try of the same head that waits or runs is the answer to a second ask; at most 4 wait or run at once, mesh-wide. A repository " + - "the checker checks no level of (one whose trunk the gate does not name) is refused.", + "the checker checks no level of (one whose trunk the gate does not name) is refused. A try whose ask no build node took within the step " + + "bound ends as error; its ask stays in the build seat's stream and may still run later, judged by nobody (issue 734).", Input: map[string]any{"type": "object", "required": []string{"repository", "head"}, "properties": map[string]any{ "repository": map[string]any{"type": "string", "description": "/"}, "head": map[string]any{"type": "string", "description": "the full commit to try, 40 or 64 hex digits"}, @@ -134,6 +144,28 @@ func TryTools(tr Trying) []stdio.Tool { } return check.ReadTry(tr.Tries, str(args, "id")) }}, + {Name: check.TryEndTool, + Description: "End a try that waits or runs, as error, by hand: say why (kept in what it says). Its place among the 4 is free at once, " + + "the checker stops waiting for its asks, and the next " + check.TryTool + " of its head asks anew with a new id. For a try whose ask " + + "was lost (no step runs for it in checker_status); the checker ends one by itself when the build seat gives its ask up, or no node " + + "takes it within the step bound. A try that ended already is refused.", + Input: map[string]any{"type": "object", "required": []string{"id", "why"}, "properties": map[string]any{ + "id": map[string]any{"type": "string", "description": "the try's id, try-"}, + "why": map[string]any{"type": "string", "description": "why it is ended, in words"}}}, + Run: func(args map[string]any) (any, error) { + if tr.Ender == nil { + return nil, fmt.Errorf("the tries state cannot be written here") + } + now := time.Now + if tr.Now != nil { + now = tr.Now + } + t, err := check.EndTry(tr.Ender, str(args, "id"), str(args, "why"), now()) + if err != nil { + return nil, err + } + return map[string]any{"id": t.ID, "state": t.State, "said": t.Said}, nil + }}, } } @@ -195,7 +227,8 @@ func Tools(r check.Reader) []stdio.Tool { "left_running with why, and not counted. Nothing holds the checker and it has no capacity of its own: a check is an ask on the " + "build seats' streams, whose backlog the delivery view shows (novox/hq ADR 0525). held_by_turn names the candidates whose check an approved " + "merge's turn holds off (novox/hq ADR 0548 §3), with the repository and the merge that waits; the gate's turns tool reads every turn and its " + - "hand-now tool steers one. Read-only.", + "hand-now tool steers one. dead_letters says whether the checker hears the asks the build seats give up on (issue 734), and why not: " + + "deaf, a try or check whose ask is given up waits until its bound. Read-only.", Input: map[string]any{"type": "object", "properties": map[string]any{}}, Run: func(map[string]any) (any, error) { return r.Status() }}, } @@ -215,7 +248,7 @@ 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"))} + TurnState: read(stdio.State("mesh-delivery-gate.turns")), HearingState: read(stdio.State(buildrun.HearingState))} 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") @@ -228,13 +261,21 @@ func main() { return s.Heads, err } r.Trunks = trunks - trying := Trying{ - Tries: stateStore{StateRead: read(tries), create: func(key string, value []byte) error { - _, err := tries.Create(key, json.RawMessage(value)) - return err - }}, - Trunks: trunks, - } + store := stateStore{StateRead: read(tries), create: func(key string, value []byte) error { + _, err := tries.Create(key, json.RawMessage(value)) + return err + }, entry: func(key string) (json.RawMessage, uint64, error) { + reads.Add(1) + e, err := tries.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 := tries.Update(key, json.RawMessage(value), revision) + return err + }} + trying := Trying{Tries: store, Ender: store, Trunks: trunks} if err := stdio.Serve("", logged(append(Tools(r), TryTools(trying)...))); 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 4bafd10..9f250df 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 @@ -53,13 +53,19 @@ func (m *memStore) Create(key string, value []byte) error { return nil } +func (m *memStore) Entry(key string) (json.RawMessage, uint64, error) { return m.vals[key], 1, nil } +func (m *memStore) Update(key string, value []byte, revision uint64) error { + m.vals[key] = value + return nil +} + func TestTheTryToolsAskAndReadATry(t *testing.T) { // ADR 0501 §5, the agreed interface with mesh-controller's try store := &memStore{vals: map[string]json.RawMessage{}} - tools := TryTools(Trying{Tries: store, Now: func() time.Time { return time.Unix(1, 5) }, + tools := TryTools(Trying{Tries: store, Ender: store, Now: func() time.Time { return time.Unix(1, 5) }, Trunks: func() (map[string]string, error) { return map[string]string{"novox/mesh-delivery": strings.Repeat("a", 40)}, nil }}) - if len(tools) != 2 || tools[0].Name != "checker_try" || tools[1].Name != "checker_try_read" { + if len(tools) != 3 || tools[0].Name != "checker_try" || tools[1].Name != "checker_try_read" || tools[2].Name != "checker_try_end" { t.Fatalf("tools = %+v", tools) } out, err := tools[0].Run(map[string]any{"repository": "novox/mesh-delivery", "head": strings.Repeat("1", 40), "trunk": "main"}) @@ -74,6 +80,14 @@ func TestTheTryToolsAskAndReadATry(t *testing.T) { // ADR 0501 §5, the agreed i if err != nil || !strings.Contains(string(raw.(json.RawMessage)), `"state":"queued"`) { t.Errorf("read = %s, %v", raw, err) } + // Issue 734: a try whose ask was lost is ended by hand with why, and its head is asked anew. + ended, err := tools[2].Run(map[string]any{"id": id, "why": "its ask was lost"}) + if err != nil || ended.(map[string]any)["state"] != "error" { + t.Errorf("end = %v, %v", ended, err) + } + if _, err := tools[2].Run(map[string]any{"id": id, "why": "again"}); err == nil { + t.Error("an ended try was ended again") + } if _, err := tools[0].Run(map[string]any{"repository": "novox/hq", "head": strings.Repeat("1", 40)}); err == nil || err.Error() != "no level of novox/hq is checked by the checker" { t.Errorf("hq: %v", err) diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go index 8268215..4621c46 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go @@ -24,6 +24,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/announce" "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" "mesh-delivery-checker/internal/check" ) @@ -79,6 +80,8 @@ type fakeBuildModules struct { js jetstream.JetStream ctx context.Context conflict atomic.Bool + // hold is a commit whose fetch never answers: its ask stays taken, as one lost in a bus reload (issue 734). + hold atomic.Value mu sync.Mutex calls map[string]int // by `.` testTool *testing.T @@ -98,6 +101,9 @@ func (f *fakeBuildModules) result(t *testing.T, tool string, args map[string]any var v any switch { case strings.HasSuffix(tool, "_fetch"): + if h, _ := f.hold.Load().(string); h != "" && args["commit"] == h { + <-f.ctx.Done() + } name := "go-fetch-clean" if f.conflict.Load() { name = "go-fetch-conflict" @@ -298,6 +304,10 @@ func startBed(t *testing.T) *bed { } b.tools[module] = kv } + // What the controller keeps of the messages a consumer gave up on (issue 734). + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: buildrun.DeadLettersStream, Subjects: []string{"mesh.events.dead.>"}}); err != nil { + t.Fatal(err) + } // The checker's stream of events, which the indexer and the forge bridge hear. if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "CHECKS", Subjects: []string{"mesh.stream.mesh-delivery-checker.event.>"}}); err != nil { t.Fatal(err) @@ -716,6 +726,49 @@ func TestATryIsJudgedAndItsEndWrittenWithNoStatusOwed(t *testing.T) { t.Errorf("a try wrote %s: %v", name, keys) } } + // Issue 734: a try whose ask the build seat gives up on (the controller keeps its copy in DEAD_LETTERS with the ask's + // message id) ends as error saying so, and its head asked again is a new try. + b.fake.hold.Store(head2) + lost, err := check.AskTry(store, map[string]string{repo: trunk}, check.TryAsk{Repository: repo, Head: head2}, time.Now()) + if err != nil { + t.Fatal(err) + } + steps, err := b.js.KeyValue(b.ctx, delivery.StepsBucket("mesh-build-go")) + if err != nil { + t.Fatal(err) + } + askID := "" + for deadline := time.Now().Add(20 * time.Second); askID == "" && time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { + keys, _ := steps.Keys(b.ctx) + for _, k := range keys { + if e, err := steps.Get(b.ctx, k); err == nil && strings.Contains(string(e.Value()), `"running"`) { + askID = strings.TrimPrefix(k, "n1.") + } + } + } + if askID == "" { + t.Fatal("the held try's ask was never taken") + } + dead := &nats.Msg{Subject: "mesh.events.dead.SEAT_NODE_BUILD_GO.SEAT_NODE_BUILD_GO_CHECK_worker", Header: nats.Header{}} + dead.Header.Set(buildrun.DeadHeaderMsgID, askID) + dead.Header.Set(buildrun.DeadHeaderDeliveries, "5") + dead.Header.Set(buildrun.DeadHeaderStream, "SEAT_NODE_BUILD_GO") + dead.Header.Set(buildrun.DeadHeaderGaveUp, time.Now().UTC().Format(time.RFC3339Nano)) + if _, err := b.js.PublishMsg(b.ctx, dead); err != nil { + t.Fatal(err) + } + for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { + raw, _ := store.Get(lost) + if last, err = check.DecodeTry(raw); err == nil && last.Over() { + break + } + } + if last.State != check.TryError || !strings.Contains(last.Said, buildrun.LostWords+" after 5 deliveries") { + t.Fatalf("the try whose ask was given up = %+v, want error saying so", last) + } + if again, err := check.AskTry(store, map[string]string{repo: trunk}, check.TryAsk{Repository: repo, Head: head2}, time.Now()); err != nil || again == lost { + t.Errorf("its head asked again answered %q, %v, want a new try", again, err) + } } // TestTheCheckerSaysOwedReadableOnTheBus: started, the checker says on its own states subject what it keeps, owed among @@ -742,7 +795,7 @@ 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") || len(said.States) != 5 { + !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 five states with kept, owed, previews and verdicts readable (1929/AC4; 1887 A5b)", said) } } 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 851b2d5..e274d4c 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go @@ -116,6 +116,8 @@ func durationEnv(getenv func(string) string, name string, def time.Duration) (ti var keptStates = []announce.State{ // The kept verdicts are read by the Go build module, which takes them for a check ask's unchanged packages (issue 1887's // 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"}, {Name: "kept", Readable: true}, {Name: "owed", Readable: true}, {Name: "previews", Readable: true}, @@ -198,6 +200,16 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // A check is one ask per pull request and build module on the build seats' check stream, pulled by a build node with a // free slot (novox/hq ADR 0525 §1, §3): the checker chooses no node and starts no step. asks := buildrun.JetStreamAsks{JS: js} + // The asks the build seats gave up on (issue 734): heard from DEAD_LETTERS, each ends the step its asker waits for. + // Heard in the background, and again at a reconnect or an ask while it is not; whether it is heard is kept in the + // state hearing, which checker_status says (dead_letters). + dead := &buildrun.DeadLetters{} + hearingKV := process.OptionalBucket(ctx, js, check.HearingBucket, "checker_status cannot say whether DEAD_LETTERS is heard", logger.Printf) + 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} ch := &check.Checker{ Verdicts: kvStore{buckets[delivery.CheckerVerdictsBucket]}, Owed: kvStore{buckets[delivery.CheckerOwedBucket]}, @@ -231,7 +243,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return cw, nil }} // The tries (ADR 0501 §5) are judged by the same judge, held by the same deliveries. - tryer := &check.Tryer{Tries: kvStore{buckets[check.TriesBucket]}, Bound: checkBound, Logf: logger.Printf} + tryer := &check.Tryer{Tries: kvStore{buckets[check.TriesBucket]}, Bound: checkBound, NotTaken: stepBound, Logf: logger.Printf} announce := func(ctx context.Context, event, key string) { body, _ := json.Marshal(map[string]string{"key": key}) if _, err := js.Publish(ctx, eventSubject(event), body); err != nil { @@ -241,7 +253,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro repos := &repoModules{manifests: kvStore{buckets[delivery.ManifestsBucket]}, logf: logger.Printf} ch.Runs = repos // the node-engine a check reads beside at its running commit (ADR 0548 §6) ch.Judge = newJudge(judgeDeps{ - asks: buildrun.Asker{Publisher: asks, Waiter: asks, Bound: stepBound}, tools: tools, trunks: trunks, + asks: buildrun.Asker{Publisher: asks, Waiter: waiter, Bound: stepBound}, tools: tools, trunks: trunks, kept: kvStore{buckets[delivery.CheckerKeptBucket]}, previews: kvStore{buckets[delivery.PreviewsBucket]}, repos: repos, builds: builds, forge: forgeReads{call: caller.Call}, logf: logger.Printf, announce: announce, compositions: compositions, diff --git a/modules/mesh-delivery-checker/go.mod b/modules/mesh-delivery-checker/go.mod index a517885..012f639 100644 --- a/modules/mesh-delivery-checker/go.mod +++ b/modules/mesh-delivery-checker/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-checker go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026130530-33395f68b309 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026134035-21252194d530 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-checker/go.sum b/modules/mesh-delivery-checker/go.sum index 9d7f193..9a6ce0a 100644 --- a/modules/mesh-delivery-checker/go.sum +++ b/modules/mesh-delivery-checker/go.sum @@ -1,11 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026111253-9c61918ac36f h1:9+TZpNq8Ifj1+qmbidE6l/7qkbua4yLdzRMrezNvf1A= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026111253-9c61918ac36f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026123512-bb713103371a h1:ZxKOtc8LusZEEpbbu41lYxJiaITN8lZCIcasA5kdgGQ= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026123512-bb713103371a/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026130105-45564e69a648 h1:UUhWHYnaTJwgkgoYhQtJrm+r5S5UNAQ4DRMlt4x21Hg= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026130105-45564e69a648/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026130530-33395f68b309 h1:mKJToG8E6ZnZj9700mtfu6YUNNnxzSj0KTfZplQkUCo= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026130530-33395f68b309/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026134035-21252194d530 h1:NwCb+gOh0UYqlyG6hG2/jffE/Gd0c7YWd4iJlaCB0lU= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026134035-21252194d530/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-checker/internal/check/checker.go b/modules/mesh-delivery-checker/internal/check/checker.go index aad5c53..e74f944 100644 --- a/modules/mesh-delivery-checker/internal/check/checker.go +++ b/modules/mesh-delivery-checker/internal/check/checker.go @@ -502,6 +502,17 @@ func (ch *Checker) try(ctx context.Context, key string, cs *Case) { began := ch.now() ch.writeOwed(ctx, cs, delivery.Verdict{State: delivery.VerdictRunning}, Outcome{}) out := ch.judge(ctx, 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. + for try := 1; IsLostAsk(out.Verdict) && ctx.Err() == nil; try++ { + if try >= LostAskTries { + out.Verdict.Failed.Words = strings.TrimSpace(out.Verdict.Failed.Words + "\nasked " + fmt.Sprint(LostAskTries) + + " times and given up each time: " + UnheardWords) + break + } + ch.logf("%s: %s; asked again (%d of %d)", cs.Key, out.Verdict.Failed.Words, try+1, LostAskTries) + out = ch.judge(ctx, cs) + } if ctx.Err() != nil { return } diff --git a/modules/mesh-delivery-checker/internal/check/lostask.go b/modules/mesh-delivery-checker/internal/check/lostask.go new file mode 100644 index 0000000..4db60c6 --- /dev/null +++ b/modules/mesh-delivery-checker/internal/check/lostask.go @@ -0,0 +1,34 @@ +package check + +import ( + "strings" + + "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/delivery/buildrun" +) + +// **An ask the build seat gave up on ends its step** (novox/hq issue 734): mesh-sdk buildrun.LostAskWaiter hands the +// waiting asker a failed step on the node buildrun.LostNode when DEAD_LETTERS keeps the ask's copy (buildrun.HearDeadLetters, +// granted by buscompose DeadLetterReads), and, under buildrun.NotTakenBound, when no build node took the ask within the +// bound. What follows is the checker's: a try ends as error (Tryer.run), with the not-taken bound at the step bound +// (Tryer.NotTaken); a check is asked again under its claim, at most LostAskTries times, then fails as not heard and is +// judged again when the nodes change (Checker.try). A check's ask has no not-taken bound: it waits in the stream as long as +// the backlog is (ADR 0525 §9), and a check ended for a backlog would owe its pull request a failure. + +// LostAskTries is how often a check asks again whose ask was given up, before it fails as not heard (UnheardWords) and is +// 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). +func IsLostAsk(v delivery.Verdict) bool { + return v.State == delivery.VerdictFail && strings.Contains(v.Failed.Words, buildrun.LostWords) +} + +// IsNotTaken says whether a verdict failed because no build node took its ask within the bound (buildrun.NotTakenWords). +func IsNotTaken(v delivery.Verdict) bool { + return v.State == delivery.VerdictFail && strings.Contains(v.Failed.Words, buildrun.NotTakenWords) +} + +// HearingBucket is the checker's state hearing: whether it hears DEAD_LETTERS (buildrun.Hearing under +// buildrun.HearingKey), which checker_status says (dead_letters). +const HearingBucket = delivery.CheckerModule + "_" + buildrun.HearingState diff --git a/modules/mesh-delivery-checker/internal/check/lostask_test.go b/modules/mesh-delivery-checker/internal/check/lostask_test.go new file mode 100644 index 0000000..35ea6a5 --- /dev/null +++ b/modules/mesh-delivery-checker/internal/check/lostask_test.go @@ -0,0 +1,170 @@ +package check + +import ( + "context" + "encoding/json" + "strings" + "sync" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/delivery/buildrun" +) + +// losingAsks is the build seat over a bus that gives up the first `lose` asks put (DEAD_LETTERS keeps them, as on +// 2026-10-26 during a bus reload: try-1793017340003993788) and answers the rest; with never set it neither answers nor +// gives up (an ask no node takes). +type losingAsks struct { + *buildrun.FakeAsks + answer func(delivery.StepAsk) delivery.Step + dead *buildrun.DeadLetters + never bool + + mu sync.Mutex + lose int +} + +func (l *losingAsks) PublishAsk(ctx context.Context, subject, id string, data []byte) error { + if err := l.FakeAsks.PublishAsk(ctx, subject, id, data); err != nil { + return err + } + l.mu.Lock() + lost := l.lose > 0 + if lost { + l.lose-- + } + l.mu.Unlock() + switch { + case l.never: + case lost: + h := map[string][]string{buildrun.DeadHeaderMsgID: {id}, buildrun.DeadHeaderDeliveries: {"5"}, buildrun.DeadHeaderStream: {"SEAT_NODE_BUILD_GO"}, + buildrun.DeadHeaderGaveUp: {"2026-10-26T12:26:40Z"}} + g, ok := buildrun.GivenUpOf(h) + if !ok { + return nil + } + go l.dead.Heard(g) + default: + a, err := delivery.DecodeStepAsk(data) + if err != nil { + return err + } + l.Write(l.answer(a)) + } + return nil +} + +func losing(b *bench, lose int) *losingAsks { + l := &losingAsks{FakeAsks: &buildrun.FakeAsks{}, answer: b.answer, dead: &buildrun.DeadLetters{}, lose: lose} + b.ex.Asks = buildrun.Asker{Publisher: l, Waiter: buildrun.LostAskWaiter{Inner: l.FakeAsks, Dead: l.dead}, Bound: 5 * time.Second} + return l +} + +// endable is the tries state as the end tool sees it. +type endable struct{ kv *memKV } + +func (e endable) Entry(key string) (json.RawMessage, uint64, error) { + v, rev, _, err := e.kv.Get(context.Background(), key) + return v, rev, err +} +func (e endable) Update(key string, value []byte, revision uint64) error { + _, err := e.kv.Update(context.Background(), key, value, revision) + return err +} + +// Issue 734: a try whose ask the build seat gave up on ends as error saying so, and the next ask of its head is a new try; +// one no node takes ends at the step bound; one ended by hand frees its place and stops being waited for. +func TestAnAskGivenUpOrNeverTakenEndsItsTryAsErrorAndItsHeadIsAskedAnew(t *testing.T) { + ctx := context.Background() + for _, row := range []struct { + name string + lose int + never bool + want string + }{ + {"given up by the build seat", 1, false, buildrun.LostWords + " after 5 deliveries at 2026-10-26T12:26:40Z"}, + {"taken by no build node within the bound", 0, true, buildrun.NotTakenWords + " within 200ms"}, + } { + t.Run(row.name, func(t *testing.T) { + kv := newMemKV() + at := time.Date(2026, 10, 26, 12, 22, 20, 0, time.UTC) + id, err := AskTry(memTries{kv}, tryTrunks, TryAsk{Repository: repoA, Head: head1}, at) + if err != nil { + t.Fatal(err) + } + b := newBench(t) + l := losing(b, row.lose) + l.never = row.never + tr := &Tryer{Tries: kv, Judge: b.ex, NotTaken: 200 * time.Millisecond} + raw, _, _, _ := kv.Get(ctx, id) + tr.Put(ctx, id, raw) + tr.Wait() + got := readTry(t, kv, id) + if got.State != TryError || len(got.Failed) != 1 || !strings.Contains(got.Failed[0], row.want) || !strings.Contains(got.Said, row.want) { + t.Fatalf("try = %+v, want error saying %q", got, row.want) + } + again, err := AskTry(memTries{kv}, tryTrunks, TryAsk{Repository: repoA, Head: head1}, at.Add(time.Minute)) + if err != nil || again == id { + t.Errorf("the head asked again answered %q, %v, want a new try", again, err) + } + }) + } + t.Run("ended by hand", func(t *testing.T) { + kv := newMemKV() + id, _ := AskTry(memTries{kv}, tryTrunks, TryAsk{Repository: repoA, Head: head1}, time.Now()) + b := newBench(t) + losing(b, 0).never = true + tr := &Tryer{Tries: kv, Judge: b.ex} // no bound: only the hand ends it + raw, _, _, _ := kv.Get(ctx, id) + tr.Put(ctx, id, raw) + deadline := time.Now().Add(2 * time.Second) + for readTry(t, kv, id).State != TryRunning && time.Now().Before(deadline) { + time.Sleep(5 * time.Millisecond) + } + if _, err := EndTry(endable{kv}, id, "", time.Now()); err == nil { + t.Error("an end without a why was taken") + } + ended, err := EndTry(endable{kv}, id, "its ask was lost in the bus reload", time.Now()) + if err != nil || ended.State != TryError || !strings.HasPrefix(ended.Said, "ended by hand ("+TryEndTool+"): its ask was lost") { + t.Fatalf("ended = %+v, %v", ended, err) + } + raw, _, _, _ = kv.Get(ctx, id) + tr.Put(ctx, id, raw) // the watch of the tries hands the end over + tr.Wait() // returns: the try's judging stopped + if got := readTry(t, kv, id); got.State != TryError || got.Said != ended.Said { + t.Errorf("try = %+v, want the hand's end kept", got) + } + if _, err := EndTry(endable{kv}, id, "again", time.Now()); err == nil || !strings.Contains(err.Error(), "ended already") { + t.Errorf("a second end answered %v", err) + } + }) +} + +// 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) { + ctx := context.Background() + for _, row := range []struct { + lose int + state string + asks int + }{ + {1, delivery.VerdictPass, 2}, + {LostAskTries, delivery.VerdictFail, LostAskTries}, + } { + 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) + } + } +} diff --git a/modules/mesh-delivery-checker/internal/check/read.go b/modules/mesh-delivery-checker/internal/check/read.go index ed7eaec..28ed623 100644 --- a/modules/mesh-delivery-checker/internal/check/read.go +++ b/modules/mesh-delivery-checker/internal/check/read.go @@ -8,6 +8,7 @@ import ( "time" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/delivery/buildrun" ) // The read tools (novox/hq ADR 0484 clauses 8 and 13, issue 1686): what the checker stored, as stored, with the line of the @@ -44,7 +45,9 @@ type Reader struct { // Runs says which commit the mesh runs of a module (the indexer's modules), as the checker's decision reads it: the // node-engine a verdict names is compared at its running commit, not at the trunk's (ADR 0548 §6); nil knows none. Runs interface{ CommitOf(module string) string } - Now func() time.Time + // HearingState is the checker's state hearing: whether it hears DEAD_LETTERS (issue 734). + HearingState 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 @@ -320,7 +323,29 @@ type Status struct { // HeldByTurn are the candidates whose check an approved merge's turn holds off (ADR 0548 §3), by group: the repository and // the group of the merge that waits. Their check is claimed when that turn goes. HeldByTurn map[string]string `json:"held_by_turn"` - Notes []string `json:"notes,omitempty"` + // 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"` + Notes []string `json:"notes,omitempty"` +} + +// DeadLettersHeard is the checker's hearing of DEAD_LETTERS as the status says it: healthy or not, and why. +type DeadLettersHeard struct { + Heard bool `json:"heard"` + Why string `json:"why"` +} + +// deadLetters reads the checker's kept hearing (buildrun.HearingHealth). +func (r Reader) deadLetters() DeadLettersHeard { + if r.HearingState.Get == nil { + return DeadLettersHeard{Why: "the checker's state hearing is not readable here"} + } + raw, err := r.HearingState.Get(buildrun.HearingKey) + if err != nil { + return DeadLettersHeard{Why: "the checker's state hearing cannot be read: " + err.Error()} + } + a := buildrun.HearingHealth(raw) + return DeadLettersHeard{Heard: a.Healthy, Why: a.Why} } // turns reads the gate's merge turns, by repository; nil when they cannot be read, said in a note. @@ -501,6 +526,7 @@ func (r Reader) Status() (Status, error) { } } } + st.DeadLetters = r.deadLetters() st.Notes = append(st.Notes, "nothing holds the checker and it holds no place: running_steps are the build modules' keys that can still be running (started within the step bound, their build module running on the node); a candidate without a verdict has its check ask in a build seat's stream, waiting for a node with a free slot, whose backlog the delivery view shows (ADR 0525 §4, §9)") return st, nil } diff --git a/modules/mesh-delivery-checker/internal/check/read_test.go b/modules/mesh-delivery-checker/internal/check/read_test.go index 0f113c1..6f1b3f9 100644 --- a/modules/mesh-delivery-checker/internal/check/read_test.go +++ b/modules/mesh-delivery-checker/internal/check/read_test.go @@ -144,7 +144,10 @@ func TestTheStatusToolReadsRunningSteps(t *testing.T) { CandidateState: fakeState{ckey: cand}.read(), VerdictState: fakeState{}.read(), HaveState: fakeState{"shanks.mesh-build-go": haveOf(t, delivery.PhaseRunning)}.read(), - Now: func() time.Time { return time.Date(2026, 10, 22, 15, 10, 0, 0, time.UTC) }, + HearingState: fakeState{"dead-letters": []byte(`{"heard":false,"streams":["SEAT_NODE_BUILD_GO"],` + + `"why":"DEAD_LETTERS is not heard, so an ask a build seat gives up on is waited for until its bound: SEAT_NODE_BUILD_GO: permissions violation",` + + `"at":"2026-10-26T13:00:00Z"}`)}.read(), + Now: func() time.Time { return time.Date(2026, 10, 22, 15, 10, 0, 0, time.UTC) }, } st, err := rd.Status() if err != nil { @@ -156,9 +159,13 @@ func TestTheStatusToolReadsRunningSteps(t *testing.T) { if len(st.Unjudged) != 1 || st.Unjudged[0] != "feat/x" { t.Errorf("unjudged = %v", st.Unjudged) } + // Issue 734, review round 1: a checker that does not hear DEAD_LETTERS says so in its status, with why. + if st.DeadLetters.Heard || !strings.Contains(st.DeadLetters.Why, "permissions violation") { + t.Errorf("dead letters = %+v, want not heard, saying why", st.DeadLetters) + } // What cannot be read is said, not guessed. st, err = Reader{}.Status() - if err != nil || len(st.Notes) == 0 { + if err != nil || len(st.Notes) == 0 || st.DeadLetters.Heard { t.Errorf("empty status = %+v, %v", st, err) } } diff --git a/modules/mesh-delivery-checker/internal/check/try.go b/modules/mesh-delivery-checker/internal/check/try.go index e2b4f5d..6ae09ac 100644 --- a/modules/mesh-delivery-checker/internal/check/try.go +++ b/modules/mesh-delivery-checker/internal/check/try.go @@ -13,6 +13,7 @@ import ( "time" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/delivery/buildrun" ) // A try is a pushed head judged by the checker as a pull request's head is, before its pull request is opened (ADR 0501 @@ -24,6 +25,7 @@ import ( const ( TryTool = "checker_try" TryReadTool = "checker_try_read" + TryEndTool = "checker_try_end" // TriesState is the checker's state of tries: the bucket mesh-delivery-checker_tries, keys try-. TriesState = "tries" TriesBucket = delivery.CheckerModule + "_" + TriesState @@ -222,6 +224,52 @@ func ReadTry(store TryStore, id string) (json.RawMessage, error) { return raw, nil } +// TryEnder is the tries state as the end tool sees it: one key with its revision, and a write at that revision. +type TryEnder interface { + Entry(key string) (json.RawMessage, uint64, error) + Update(key string, value []byte, revision uint64) error +} + +// EndTry ends a try that waits or runs as error, saying why and that it was ended by hand (issue 734): its place is +// free at once, and the next checker_try of its head asks anew. A try that ended already is refused, naming its end. +func EndTry(store TryEnder, id, why string, now time.Time) (Try, error) { + if !tryKeyOK.MatchString(id) { + return Try{}, fmt.Errorf("%q is not a try's id, try-", id) + } + why = strings.TrimSpace(why) + if why == "" { + return Try{}, errors.New("say why the try is ended: the why is kept in what it says") + } + raw, rev, err := store.Entry(id) + if err != nil { + return Try{}, err + } + if len(raw) == 0 { + return Try{}, fmt.Errorf("no try %s: it was never asked, or it ended more than %s ago and was removed", id, TryKeep) + } + t, err := DecodeTry(raw) + if err != nil { + return Try{}, err + } + if t.Over() { + return Try{}, fmt.Errorf("%s ended already, %s at %s", id, t.State, t.Ended.Format(time.RFC3339)) + } + t.State, t.Ended = TryError, now.UTC() + if !t.Started.IsZero() { + t.Took = t.Ended.Sub(t.Started).Round(time.Second).String() + } + t.Said = strings.TrimSpace("ended by hand (" + TryEndTool + "): " + why + " | " + t.Said) + t.Said = strings.TrimSuffix(t.Said, " |") + data, err := encodeTry(t) + if err != nil { + return Try{}, err + } + if err := store.Update(id, data, rev); err != nil { + return Try{}, fmt.Errorf("%s changed while it was ended, read it and end it again: %w", id, err) + } + return t, nil +} + // TryOptions narrow what a try runs; a pull request's check has none. type TryOptions struct { Level string @@ -234,12 +282,14 @@ type Tryer struct { Tries KV Judge Judge Bound time.Duration - Now func() time.Time - Logf func(format string, args ...any) + // NotTaken ends a try whose ask no build node took within it as error (issue 734); none when 0. + NotTaken time.Duration + Now func() time.Time + Logf func(format string, args ...any) mu sync.Mutex wg sync.WaitGroup - inflight map[string]bool + inflight map[string]context.CancelFunc seen map[string]bool // looked at since this process started: a try left running is taken over the first time } @@ -267,25 +317,31 @@ func (tr *Tryer) Put(ctx context.Context, key string, value []byte) { tr.logf("tries.%s holds what is not a try, left: %v", key, err) return } - if t.Over() { - return - } tr.mu.Lock() if tr.inflight == nil { - tr.inflight, tr.seen = map[string]bool{}, map[string]bool{} + tr.inflight, tr.seen = map[string]context.CancelFunc{}, map[string]bool{} } - if tr.inflight[key] || (t.State == TryRunning && tr.seen[key]) { + if t.Over() { + // Ended while it ran (checker_try_end): its judging stops, and its asks are no longer waited for. + if stop := tr.inflight[key]; stop != nil { + stop() + } tr.mu.Unlock() return } + if tr.inflight[key] != nil || (t.State == TryRunning && tr.seen[key]) { + tr.mu.Unlock() + return + } + run, stop := context.WithCancel(ctx) tr.seen[key] = true - tr.inflight[key] = true + tr.inflight[key] = stop tr.mu.Unlock() tr.wg.Add(1) go func() { defer tr.wg.Done() - defer func() { tr.mu.Lock(); delete(tr.inflight, key); tr.mu.Unlock() }() - tr.run(ctx, key) + defer func() { tr.mu.Lock(); delete(tr.inflight, key); tr.mu.Unlock(); stop() }() + tr.run(run, key) }() } @@ -342,9 +398,13 @@ func (tr *Tryer) run(ctx context.Context, key string) { Queued: t.Asked, Progress: &Progress{}, Try: &TryOptions{Level: t.Level, Run: t.Tests}} - out, judged := judgeWithin(ctx, tr.Judge, cs, tr.Bound) + judging := ctx + if tr.NotTaken > 0 { + judging = buildrun.NotTakenBound(ctx, tr.NotTaken) + } + out, judged := judgeWithin(judging, tr.Judge, cs, tr.Bound) if ctx.Err() != nil { - return // left running: the checker takes it over when it starts + return // left running: the checker takes it over when it starts; or ended by hand, which wrote its end } v := out.Verdict t.Ended = tr.now().UTC() @@ -355,8 +415,12 @@ func (tr *Tryer) run(ctx context.Context, key string) { t.Said += " | " + n } switch { - case !judged || IsNodesFailure(v): + case !judged || IsNodesFailure(v) || IsLostAsk(v) || IsNotTaken(v): + // Not the change's: its ask was given up or never taken (issue 734), or a part of the mesh was not heard. t.State = TryError + if IsLostAsk(v) || IsNotTaken(v) { + t.Failed = []string{v.Failed.Words} + } case v.State == delivery.VerdictPass: t.State = TryPass default: diff --git a/modules/mesh-delivery-checker/manifest_test.go b/modules/mesh-delivery-checker/manifest_test.go index 07bb4aa..815ce91 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: "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: "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,8 @@ 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 two tools (ADR 0501 §5) beside them: mesh-controller's try calls them. - if want := []string{"checker_kept", "checker_previews", "checker_pull", "checker_status", "checker_try", "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. + 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) { 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 eafca18..a968770 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": "hearing", "history": 1 }, { "name": "kept", "history": 1 }, { "name": "owed", "history": 5 }, { "name": "previews", "history": 5 }, @@ -42,6 +43,7 @@ "checker_pull", "checker_status", "checker_try", + "checker_try_end", "checker_try_read", "checker_verdicts" ], diff --git a/modules/mesh-grants/go.mod b/modules/mesh-grants/go.mod index 3d6dc66..76ac8cc 100644 --- a/modules/mesh-grants/go.mod +++ b/modules/mesh-grants/go.mod @@ -3,7 +3,7 @@ module mesh-grants go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026112527-c2a410bb6376 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026134035-21252194d530 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-grants/go.sum b/modules/mesh-grants/go.sum index 8f26273..78176f6 100644 --- a/modules/mesh-grants/go.sum +++ b/modules/mesh-grants/go.sum @@ -1,11 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026080443-136fa8b2a896 h1:5rHfru5NJ7SK20QOt8YUJSglcn1gsjtLc9AGmWYGnRQ= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026080443-136fa8b2a896/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026105311-175a5567b8bd h1:ef6y09csev6xY/suFrIFGUXbgvDA29kummFtHcIb3B8= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026105311-175a5567b8bd/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026110029-83ee7769d8ee h1:ZdBMkF4WYJi82jb3xYi9Z2lXN/iMTn9r3SFC122ROY4= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026110029-83ee7769d8ee/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026112527-c2a410bb6376 h1:by2zBASLb8h5jA1IiYJWykXwLnHYpuZh/zFNa4NsbOU= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026112527-c2a410bb6376/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026134035-21252194d530 h1:NwCb+gOh0UYqlyG6hG2/jffE/Gd0c7YWd4iJlaCB0lU= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026134035-21252194d530/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=