From 6b85220fa4394e491248557b85a231fdcdf899e2 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 26 Oct 2026 13:51:04 +0100 Subject: [PATCH 1/4] Checker: an ask the build seat gives up on ends its try as error and asks a check again; a try no node takes ends at the step bound; checker_try_end ends one by hand (issue 734) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A check ask handed out five times during a bus reload was given up into DEAD_LETTERS and no build node wrote a step for it, so try-1793017340003993788 stayed running for ever, held one of the four try places, and checker_try of its head kept answering its id. The checker now hears the copies DEAD_LETTERS keeps of the build seats' given-up asks (an ordered consumer from its start; the grant is mesh-sdk/mesh-controller DeadLetterReads, branch fix/734-lost-ask) and ends the step it waits for: a try ends as error saying why, a check is asked again under its claim, up to three times, then fails as not heard. A try whose ask no node takes within the step bound ends the same way (a check's backlog stays no fault, ADR 0525 §9). checker_try_end ends a try by hand, freeing its place and stopping its judging. --- .../cmd/mesh-delivery-checker-tools/main.go | 54 +++- .../mesh-delivery-checker-tools/main_test.go | 18 +- .../mesh-delivery-checker/integration_test.go | 52 ++++ .../cmd/mesh-delivery-checker/main.go | 48 +++- .../internal/check/checker.go | 11 + .../internal/check/lostask.go | 257 ++++++++++++++++++ .../internal/check/lostask_test.go | 170 ++++++++++++ .../internal/check/try.go | 91 ++++++- .../mesh-delivery-checker/manifest_test.go | 4 +- modules/mesh-delivery-checker/module.json | 1 + 10 files changed, 678 insertions(+), 28 deletions(-) create mode 100644 modules/mesh-delivery-checker/internal/check/lostask.go create mode 100644 modules/mesh-delivery-checker/internal/check/lostask_test.go 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..c36295c 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 @@ -76,6 +76,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 +92,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{ @@ -134,6 +142,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 + }}, } } @@ -228,13 +258,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..55f2b3d 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 @@ -79,6 +79,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 +100,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 +303,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: check.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 +725,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(check.DeadHeaderMsgID, askID) + dead.Header.Set(check.DeadHeaderDeliveries, "5") + dead.Header.Set(check.DeadHeaderStream, "SEAT_NODE_BUILD_GO") + dead.Header.Set(check.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, check.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 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..db516ed 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go @@ -198,6 +198,10 @@ 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. + dead := &check.DeadLetters{} + defer hearDeadLetters(ctx, js, dead, logger.Printf)() + waiter := check.LostAskWaiter{Inner: asks, Dead: dead} ch := &check.Checker{ Verdicts: kvStore{buckets[delivery.CheckerVerdictsBucket]}, Owed: kvStore{buckets[delivery.CheckerOwedBucket]}, @@ -231,7 +235,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 +245,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, @@ -397,3 +401,43 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return err } } + +// hearDeadLetters follows, for each build seat's stream, the copies DEAD_LETTERS keeps of the asks its workers gave up on +// (issue 734): an ordered consumer from this start, which resumes where it was after a reconnect, so a bus reload loses none. Its +// grant is a user's of a seat pulled by kind (mesh-sdk buscompose DeadLetterReads); where the bus refuses it, the checker +// says so and goes on: a try still ends when no node took its ask within the step bound, and a check waits as before. +func hearDeadLetters(ctx context.Context, js jetstream.JetStream, dead *check.DeadLetters, logf func(string, ...any)) func() { + var stops []func() + // From this start on: a step this process waits for was asked since, and a reset of the consumer before it heard any + // copy starts here again rather than at the reset. + since := time.Now() + for _, stream := range check.BuildSeatStreams { + filter := check.DeadLetterFilter(stream) + notHeard := func(err error) { + logf("the asks %s gives up on are not heard (%s of %s), so a check whose ask it gives up waits until the checker starts again: %v", + stream, filter, check.DeadLettersStream, err) + } + cons, err := js.OrderedConsumer(ctx, check.DeadLettersStream, jetstream.OrderedConsumerConfig{ + FilterSubjects: []string{filter}, DeliverPolicy: jetstream.DeliverByStartTimePolicy, OptStartTime: &since}) + if err != nil { + notHeard(err) + continue + } + cc, err := cons.Consume(func(m jetstream.Msg) { + if g, ok := check.GivenUpOf(m.Headers()); ok { + logf("the ask %s on %s was given up after %d deliveries at %s: the step that waits for it ends", g.AskID, g.Stream, g.Deliveries, g.At.Format(time.RFC3339)) + dead.Heard(g) + } + }, jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logf("hearing %s: %v", check.DeadLettersStream, err) })) + if err != nil { + notHeard(err) + continue + } + stops = append(stops, cc.Stop) + } + return func() { + for _, stop := range stops { + stop() + } + } +} 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..e33f8b9 --- /dev/null +++ b/modules/mesh-delivery-checker/internal/check/lostask.go @@ -0,0 +1,257 @@ +package check + +import ( + "context" + "fmt" + "strconv" + "strings" + "sync" + "time" + + "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). A build seat's worker offers an ask +// delivery.AskMaxDeliver times; one handed out that often without an answer (a bus reload that drops every hand-out, a node +// that dies taking it each time) is given up by the server, and the controller keeps a copy of it in DEAD_LETTERS under +// mesh.events.dead.., with the ask's message id (its ask id), how often it was handed out and when it +// was given up. No build node writes a step for it, so an asker that waits for the step waits for ever: a try stays +// running and holds a try place, a check's verdict stays running and its pull request waits. +// +// The checker hears those copies (DeadLetters, an ordered consumer on DEAD_LETTERS filtered to the build seats' streams, +// granted to a user of a seat pulled by kind, mesh-sdk buscompose DeadLetterReads) and LostAskWaiter ends the waiting step +// at once as failed, its words saying LostWords. A try ends as error; a check asks again (Checker.try). +// +// Belt and braces, for a try only (NotTakenBound): an ask no build node took within the step bound ends the same way, +// saying NotTakenWords. A check's ask waits in the stream as long as the backlog is (ADR 0525 §9): a backlog is no fault, +// and a check ended for one would owe its pull request a failure. + +// The words of a step the checker ended for its ask. +const ( + LostWords = "its ask was given up by the build seat" + NotTakenWords = "no build node took its ask" +) + +// 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 + +// The headers the controller keeps a given-up message's copy with in DEAD_LETTERS (mesh-controller internal/link +// deadletters.go): the original message id (an ask's is its ask id), how often it was handed out, when it was given up. +const ( + DeadHeaderMsgID = "Mesh-Dead-Msg-Id" + DeadHeaderDeliveries = "Mesh-Dead-Deliveries" + DeadHeaderGaveUp = "Mesh-Dead-Gave-Up" + DeadHeaderStream = "Mesh-Dead-Stream" +) + +// DeadLettersStream is the stream the controller keeps given-up messages in, and DeadLetterFilter the subjects of one seat +// stream's copies there (mesh-sdk buscompose DeadLetterSubject). +const DeadLettersStream = "DEAD_LETTERS" + +// DeadLetterFilter is the subjects DEAD_LETTERS keeps a seat stream's given-up messages under. +func DeadLetterFilter(stream string) string { return "mesh.events.dead." + stream + ".>" } + +// BuildSeatStreams are the build seats' streams the checker asks on (mesh-sdk buscompose seatStreamName of node-build-go +// and node-build-node). +var BuildSeatStreams = []string{"SEAT_NODE_BUILD_GO", "SEAT_NODE_BUILD_NODE"} + +// GivenUp is one ask a build seat gave up on, as DEAD_LETTERS keeps it. +type GivenUp struct { + AskID string + Stream string + Deliveries int + At time.Time +} + +// GivenUpOf reads a DEAD_LETTERS copy's headers; false when it names no message id. +func GivenUpOf(header map[string][]string) (GivenUp, bool) { + get := func(k string) string { + if v := header[k]; len(v) > 0 { + return strings.TrimSpace(v[0]) + } + return "" + } + g := GivenUp{AskID: get(DeadHeaderMsgID), Stream: get(DeadHeaderStream)} + if g.AskID == "" { + return GivenUp{}, false + } + g.Deliveries, _ = strconv.Atoi(get(DeadHeaderDeliveries)) + g.At, _ = time.Parse(time.RFC3339Nano, get(DeadHeaderGaveUp)) + return g, true +} + +// Words are what a step ended for the given-up ask says. +func (g GivenUp) Words() string { + at := "at an unknown time" + if !g.At.IsZero() { + at = "at " + g.At.UTC().Format(time.RFC3339) + } + return fmt.Sprintf("%s after %d deliveries %s (kept in %s); ask again", LostWords, g.Deliveries, at, DeadLettersStream) +} + +// keptGivenUp is how many given-up asks DeadLetters remembers for a waiter that starts after the copy was heard. +const keptGivenUp = 1024 + +// DeadLetters holds the given-up asks heard, and hands each to the waiter of its ask. Safe for concurrent use. +type DeadLetters struct { + mu sync.Mutex + heard map[string]GivenUp + order []string + watchers map[string][]chan GivenUp +} + +// Heard takes one given-up ask in. +func (d *DeadLetters) Heard(g GivenUp) { + d.mu.Lock() + if d.heard == nil { + d.heard, d.watchers = map[string]GivenUp{}, map[string][]chan GivenUp{} + } + if _, ok := d.heard[g.AskID]; !ok { + d.order = append(d.order, g.AskID) + if len(d.order) > keptGivenUp { + delete(d.heard, d.order[0]) + d.order = d.order[1:] + } + } + d.heard[g.AskID] = g + ws := d.watchers[g.AskID] + delete(d.watchers, g.AskID) + d.mu.Unlock() + for _, w := range ws { + w <- g // buffered, one each + } +} + +// watch answers a channel that gets the ask's given-up copy once (at once when it was heard already), and its cancel. +func (d *DeadLetters) watch(askID string) (<-chan GivenUp, func()) { + ch := make(chan GivenUp, 1) + d.mu.Lock() + defer d.mu.Unlock() + if d.heard == nil { + d.heard, d.watchers = map[string]GivenUp{}, map[string][]chan GivenUp{} + } + if g, ok := d.heard[askID]; ok { + ch <- g + return ch, func() {} + } + d.watchers[askID] = append(d.watchers[askID], ch) + return ch, func() { + d.mu.Lock() + defer d.mu.Unlock() + ws := d.watchers[askID] + for i, w := range ws { + if w == ch { + d.watchers[askID] = append(ws[:i], ws[i+1:]...) + break + } + } + if len(d.watchers[askID]) == 0 { + delete(d.watchers, askID) + } + } +} + +type notTakenKey struct{} + +// NotTakenBound says, in ctx, that an ask no build node took within bound ends its step (a try's, never a check's). +func NotTakenBound(ctx context.Context, bound time.Duration) context.Context { + return context.WithValue(ctx, notTakenKey{}, bound) +} + +func notTakenBoundOf(ctx context.Context) time.Duration { + d, _ := ctx.Value(notTakenKey{}).(time.Duration) + return d +} + +// LostAskWaiter is a buildrun.AskWaiter that ends an ask's step when the build seat gave the ask up (DeadLetters), or, under +// NotTakenBound, when no build node took it within the bound: it hands the asker a failed step whose reason says why, on +// the node "bus". Every step the build module writes is handed on as it is. +type LostAskWaiter struct { + Inner buildrun.AskWaiter + Dead *DeadLetters + Now func() time.Time +} + +func (w LostAskWaiter) now() time.Time { + if w.Now != nil { + return w.Now() + } + return time.Now() +} + +// WatchAsk is AskWaiter's. +func (w LostAskWaiter) WatchAsk(ctx context.Context, module, askID string) (<-chan delivery.Step, error) { + inner, err := w.Inner.WatchAsk(ctx, module, askID) + if err != nil { + return nil, err + } + var dead <-chan GivenUp + cancel := func() {} + if w.Dead != nil { + dead, cancel = w.Dead.watch(askID) + } + // The not-taken bound is a context's (ADR 0437 §8: no timer), ended when a node takes the ask. + var notTaken <-chan struct{} + bound := notTakenBoundOf(ctx) + stopBound := func() {} + if bound > 0 { + var nt context.Context + nt, stopBound = context.WithTimeout(context.Background(), bound) + notTaken = nt.Done() + } + out := make(chan delivery.Step) + go func() { + defer close(out) + defer cancel() + defer func() { stopBound() }() + send := func(s delivery.Step) bool { + select { + case out <- s: + return true + case <-ctx.Done(): + return false + } + } + ended := func(reason string) delivery.Step { + at := w.now().UTC() + return delivery.Step{ID: askID, Node: "bus", State: delivery.StepFailed, Reason: reason, Ended: at} + } + for { + select { + case <-ctx.Done(): + return + case s, ok := <-inner: + if !ok { + return + } + if notTaken != nil { + stopBound() // a node took it: its step is bounded by the asker from its start + notTaken = nil + } + if !send(s) { + return + } + case g := <-dead: + dead = nil + send(ended(g.Words())) + return + case <-notTaken: + send(ended(fmt.Sprintf("%s within %s of its put; it was lost or waits behind a backlog; ask again", NotTakenWords, bound))) + return + } + } + }() + return out, nil +} + +// IsLostAsk says whether a verdict failed because the build seat gave its ask up (LostWords). +func IsLostAsk(v delivery.Verdict) bool { + return v.State == delivery.VerdictFail && strings.Contains(v.Failed.Words, LostWords) +} + +// IsNotTaken says whether a verdict failed because no build node took its ask within the bound (NotTakenWords). +func IsNotTaken(v delivery.Verdict) bool { + return v.State == delivery.VerdictFail && strings.Contains(v.Failed.Words, NotTakenWords) +} 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..e12566e --- /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 *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{DeadHeaderMsgID: {id}, DeadHeaderDeliveries: {"5"}, DeadHeaderStream: {"SEAT_NODE_BUILD_GO"}, + DeadHeaderGaveUp: {"2026-10-26T12:26:40Z"}} + g, ok := 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: &DeadLetters{}, lose: lose} + b.ex.Asks = buildrun.Asker{Publisher: l, Waiter: 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, LostWords + " after 5 deliveries at 2026-10-26T12:26:40Z"}, + {"taken by no build node within the bound", 0, true, 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/try.go b/modules/mesh-delivery-checker/internal/check/try.go index e2b4f5d..c27c449 100644 --- a/modules/mesh-delivery-checker/internal/check/try.go +++ b/modules/mesh-delivery-checker/internal/check/try.go @@ -24,6 +24,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 +223,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 +281,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 +316,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 +397,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 = 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 +414,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..de36137 100644 --- a/modules/mesh-delivery-checker/manifest_test.go +++ b/modules/mesh-delivery-checker/manifest_test.go @@ -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..53e50b4 100644 --- a/modules/mesh-delivery-checker/module.json +++ b/modules/mesh-delivery-checker/module.json @@ -42,6 +42,7 @@ "checker_pull", "checker_status", "checker_try", + "checker_try_end", "checker_try_read", "checker_verdicts" ], -- 2.54.0 From 33de9dde4ff9bca1075407a11875db231f53ea4a Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 26 Oct 2026 14:03:01 +0100 Subject: [PATCH 2/4] Checker and builder: wait on the build seats through mesh-sdk's LostAskWaiter, so an ask the seat gave up on ends every asker's wait (issue 734) The waiter moved to mesh-sdk buildrun (fix/734-lost-ask, d96b9fd6), where every asker of the build seats shares it. The builder hears DEAD_LETTERS for the go, node and image seats and asks a given-up part of a build again at once, with the seat's words as why, instead of waiting for the check bound. The checker keeps what it does with a lost ask (a try ends as error, a check is asked again) and its verdict words. --- .../cmd/mesh-delivery-builder/main.go | 6 +- modules/mesh-delivery-builder/go.mod | 2 +- modules/mesh-delivery-builder/go.sum | 6 +- .../internal/build/askagain.go | 9 +- .../internal/build/run_test.go | 17 +- .../mesh-delivery-checker/integration_test.go | 13 +- .../cmd/mesh-delivery-checker/main.go | 46 +--- .../internal/check/lostask.go | 249 +----------------- .../internal/check/lostask_test.go | 16 +- .../internal/check/try.go | 3 +- 10 files changed, 62 insertions(+), 305 deletions(-) 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..c3223b3 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,10 @@ 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. + dead := &buildrun.DeadLetters{} + defer buildrun.HearDeadLetters(ctx, js, delivery.AskLanguages, dead, logger.Printf)() // 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 +166,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}, 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..2357313 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.20261026130046-d96b9fd66069 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..9a97538 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.20261026130046-d96b9fd66069 h1:dPwZwGbW8LuT4p2baXuiyVNOWU6XO46kbw46VU5TAUQ= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026130046-d96b9fd66069/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..a65a63b 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 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 55f2b3d..d3e8e23 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" ) @@ -304,7 +305,7 @@ 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: check.DeadLettersStream, Subjects: []string{"mesh.events.dead.>"}}); err != nil { + 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. @@ -749,10 +750,10 @@ func TestATryIsJudgedAndItsEndWrittenWithNoStatusOwed(t *testing.T) { 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(check.DeadHeaderMsgID, askID) - dead.Header.Set(check.DeadHeaderDeliveries, "5") - dead.Header.Set(check.DeadHeaderStream, "SEAT_NODE_BUILD_GO") - dead.Header.Set(check.DeadHeaderGaveUp, time.Now().UTC().Format(time.RFC3339Nano)) + 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) } @@ -762,7 +763,7 @@ func TestATryIsJudgedAndItsEndWrittenWithNoStatusOwed(t *testing.T) { break } } - if last.State != check.TryError || !strings.Contains(last.Said, check.LostWords+" after 5 deliveries") { + 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 { 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 db516ed..21a0d2c 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go @@ -199,9 +199,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // 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. - dead := &check.DeadLetters{} - defer hearDeadLetters(ctx, js, dead, logger.Printf)() - waiter := check.LostAskWaiter{Inner: asks, Dead: dead} + dead := &buildrun.DeadLetters{} + defer buildrun.HearDeadLetters(ctx, js, []string{delivery.AskGo, delivery.AskNode}, dead, logger.Printf)() + waiter := buildrun.LostAskWaiter{Inner: asks, Dead: dead} ch := &check.Checker{ Verdicts: kvStore{buckets[delivery.CheckerVerdictsBucket]}, Owed: kvStore{buckets[delivery.CheckerOwedBucket]}, @@ -401,43 +401,3 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return err } } - -// hearDeadLetters follows, for each build seat's stream, the copies DEAD_LETTERS keeps of the asks its workers gave up on -// (issue 734): an ordered consumer from this start, which resumes where it was after a reconnect, so a bus reload loses none. Its -// grant is a user's of a seat pulled by kind (mesh-sdk buscompose DeadLetterReads); where the bus refuses it, the checker -// says so and goes on: a try still ends when no node took its ask within the step bound, and a check waits as before. -func hearDeadLetters(ctx context.Context, js jetstream.JetStream, dead *check.DeadLetters, logf func(string, ...any)) func() { - var stops []func() - // From this start on: a step this process waits for was asked since, and a reset of the consumer before it heard any - // copy starts here again rather than at the reset. - since := time.Now() - for _, stream := range check.BuildSeatStreams { - filter := check.DeadLetterFilter(stream) - notHeard := func(err error) { - logf("the asks %s gives up on are not heard (%s of %s), so a check whose ask it gives up waits until the checker starts again: %v", - stream, filter, check.DeadLettersStream, err) - } - cons, err := js.OrderedConsumer(ctx, check.DeadLettersStream, jetstream.OrderedConsumerConfig{ - FilterSubjects: []string{filter}, DeliverPolicy: jetstream.DeliverByStartTimePolicy, OptStartTime: &since}) - if err != nil { - notHeard(err) - continue - } - cc, err := cons.Consume(func(m jetstream.Msg) { - if g, ok := check.GivenUpOf(m.Headers()); ok { - logf("the ask %s on %s was given up after %d deliveries at %s: the step that waits for it ends", g.AskID, g.Stream, g.Deliveries, g.At.Format(time.RFC3339)) - dead.Heard(g) - } - }, jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logf("hearing %s: %v", check.DeadLettersStream, err) })) - if err != nil { - notHeard(err) - continue - } - stops = append(stops, cc.Stop) - } - return func() { - for _, stop := range stops { - stop() - } - } -} diff --git a/modules/mesh-delivery-checker/internal/check/lostask.go b/modules/mesh-delivery-checker/internal/check/lostask.go index e33f8b9..4335752 100644 --- a/modules/mesh-delivery-checker/internal/check/lostask.go +++ b/modules/mesh-delivery-checker/internal/check/lostask.go @@ -1,257 +1,30 @@ package check import ( - "context" - "fmt" - "strconv" "strings" - "sync" - "time" "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). A build seat's worker offers an ask -// delivery.AskMaxDeliver times; one handed out that often without an answer (a bus reload that drops every hand-out, a node -// that dies taking it each time) is given up by the server, and the controller keeps a copy of it in DEAD_LETTERS under -// mesh.events.dead.., with the ask's message id (its ask id), how often it was handed out and when it -// was given up. No build node writes a step for it, so an asker that waits for the step waits for ever: a try stays -// running and holds a try place, a check's verdict stays running and its pull request waits. -// -// The checker hears those copies (DeadLetters, an ordered consumer on DEAD_LETTERS filtered to the build seats' streams, -// granted to a user of a seat pulled by kind, mesh-sdk buscompose DeadLetterReads) and LostAskWaiter ends the waiting step -// at once as failed, its words saying LostWords. A try ends as error; a check asks again (Checker.try). -// -// Belt and braces, for a try only (NotTakenBound): an ask no build node took within the step bound ends the same way, -// saying NotTakenWords. A check's ask waits in the stream as long as the backlog is (ADR 0525 §9): a backlog is no fault, -// and a check ended for one would owe its pull request a failure. - -// The words of a step the checker ended for its ask. -const ( - LostWords = "its ask was given up by the build seat" - NotTakenWords = "no build node took its ask" -) +// **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 -// The headers the controller keeps a given-up message's copy with in DEAD_LETTERS (mesh-controller internal/link -// deadletters.go): the original message id (an ask's is its ask id), how often it was handed out, when it was given up. -const ( - DeadHeaderMsgID = "Mesh-Dead-Msg-Id" - DeadHeaderDeliveries = "Mesh-Dead-Deliveries" - DeadHeaderGaveUp = "Mesh-Dead-Gave-Up" - DeadHeaderStream = "Mesh-Dead-Stream" -) - -// DeadLettersStream is the stream the controller keeps given-up messages in, and DeadLetterFilter the subjects of one seat -// stream's copies there (mesh-sdk buscompose DeadLetterSubject). -const DeadLettersStream = "DEAD_LETTERS" - -// DeadLetterFilter is the subjects DEAD_LETTERS keeps a seat stream's given-up messages under. -func DeadLetterFilter(stream string) string { return "mesh.events.dead." + stream + ".>" } - -// BuildSeatStreams are the build seats' streams the checker asks on (mesh-sdk buscompose seatStreamName of node-build-go -// and node-build-node). -var BuildSeatStreams = []string{"SEAT_NODE_BUILD_GO", "SEAT_NODE_BUILD_NODE"} - -// GivenUp is one ask a build seat gave up on, as DEAD_LETTERS keeps it. -type GivenUp struct { - AskID string - Stream string - Deliveries int - At time.Time -} - -// GivenUpOf reads a DEAD_LETTERS copy's headers; false when it names no message id. -func GivenUpOf(header map[string][]string) (GivenUp, bool) { - get := func(k string) string { - if v := header[k]; len(v) > 0 { - return strings.TrimSpace(v[0]) - } - return "" - } - g := GivenUp{AskID: get(DeadHeaderMsgID), Stream: get(DeadHeaderStream)} - if g.AskID == "" { - return GivenUp{}, false - } - g.Deliveries, _ = strconv.Atoi(get(DeadHeaderDeliveries)) - g.At, _ = time.Parse(time.RFC3339Nano, get(DeadHeaderGaveUp)) - return g, true -} - -// Words are what a step ended for the given-up ask says. -func (g GivenUp) Words() string { - at := "at an unknown time" - if !g.At.IsZero() { - at = "at " + g.At.UTC().Format(time.RFC3339) - } - return fmt.Sprintf("%s after %d deliveries %s (kept in %s); ask again", LostWords, g.Deliveries, at, DeadLettersStream) -} - -// keptGivenUp is how many given-up asks DeadLetters remembers for a waiter that starts after the copy was heard. -const keptGivenUp = 1024 - -// DeadLetters holds the given-up asks heard, and hands each to the waiter of its ask. Safe for concurrent use. -type DeadLetters struct { - mu sync.Mutex - heard map[string]GivenUp - order []string - watchers map[string][]chan GivenUp -} - -// Heard takes one given-up ask in. -func (d *DeadLetters) Heard(g GivenUp) { - d.mu.Lock() - if d.heard == nil { - d.heard, d.watchers = map[string]GivenUp{}, map[string][]chan GivenUp{} - } - if _, ok := d.heard[g.AskID]; !ok { - d.order = append(d.order, g.AskID) - if len(d.order) > keptGivenUp { - delete(d.heard, d.order[0]) - d.order = d.order[1:] - } - } - d.heard[g.AskID] = g - ws := d.watchers[g.AskID] - delete(d.watchers, g.AskID) - d.mu.Unlock() - for _, w := range ws { - w <- g // buffered, one each - } -} - -// watch answers a channel that gets the ask's given-up copy once (at once when it was heard already), and its cancel. -func (d *DeadLetters) watch(askID string) (<-chan GivenUp, func()) { - ch := make(chan GivenUp, 1) - d.mu.Lock() - defer d.mu.Unlock() - if d.heard == nil { - d.heard, d.watchers = map[string]GivenUp{}, map[string][]chan GivenUp{} - } - if g, ok := d.heard[askID]; ok { - ch <- g - return ch, func() {} - } - d.watchers[askID] = append(d.watchers[askID], ch) - return ch, func() { - d.mu.Lock() - defer d.mu.Unlock() - ws := d.watchers[askID] - for i, w := range ws { - if w == ch { - d.watchers[askID] = append(ws[:i], ws[i+1:]...) - break - } - } - if len(d.watchers[askID]) == 0 { - delete(d.watchers, askID) - } - } -} - -type notTakenKey struct{} - -// NotTakenBound says, in ctx, that an ask no build node took within bound ends its step (a try's, never a check's). -func NotTakenBound(ctx context.Context, bound time.Duration) context.Context { - return context.WithValue(ctx, notTakenKey{}, bound) -} - -func notTakenBoundOf(ctx context.Context) time.Duration { - d, _ := ctx.Value(notTakenKey{}).(time.Duration) - return d -} - -// LostAskWaiter is a buildrun.AskWaiter that ends an ask's step when the build seat gave the ask up (DeadLetters), or, under -// NotTakenBound, when no build node took it within the bound: it hands the asker a failed step whose reason says why, on -// the node "bus". Every step the build module writes is handed on as it is. -type LostAskWaiter struct { - Inner buildrun.AskWaiter - Dead *DeadLetters - Now func() time.Time -} - -func (w LostAskWaiter) now() time.Time { - if w.Now != nil { - return w.Now() - } - return time.Now() -} - -// WatchAsk is AskWaiter's. -func (w LostAskWaiter) WatchAsk(ctx context.Context, module, askID string) (<-chan delivery.Step, error) { - inner, err := w.Inner.WatchAsk(ctx, module, askID) - if err != nil { - return nil, err - } - var dead <-chan GivenUp - cancel := func() {} - if w.Dead != nil { - dead, cancel = w.Dead.watch(askID) - } - // The not-taken bound is a context's (ADR 0437 §8: no timer), ended when a node takes the ask. - var notTaken <-chan struct{} - bound := notTakenBoundOf(ctx) - stopBound := func() {} - if bound > 0 { - var nt context.Context - nt, stopBound = context.WithTimeout(context.Background(), bound) - notTaken = nt.Done() - } - out := make(chan delivery.Step) - go func() { - defer close(out) - defer cancel() - defer func() { stopBound() }() - send := func(s delivery.Step) bool { - select { - case out <- s: - return true - case <-ctx.Done(): - return false - } - } - ended := func(reason string) delivery.Step { - at := w.now().UTC() - return delivery.Step{ID: askID, Node: "bus", State: delivery.StepFailed, Reason: reason, Ended: at} - } - for { - select { - case <-ctx.Done(): - return - case s, ok := <-inner: - if !ok { - return - } - if notTaken != nil { - stopBound() // a node took it: its step is bounded by the asker from its start - notTaken = nil - } - if !send(s) { - return - } - case g := <-dead: - dead = nil - send(ended(g.Words())) - return - case <-notTaken: - send(ended(fmt.Sprintf("%s within %s of its put; it was lost or waits behind a backlog; ask again", NotTakenWords, bound))) - return - } - } - }() - return out, nil -} - -// IsLostAsk says whether a verdict failed because the build seat gave its ask up (LostWords). +// 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, LostWords) + 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 (NotTakenWords). +// 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, NotTakenWords) + return v.State == delivery.VerdictFail && strings.Contains(v.Failed.Words, buildrun.NotTakenWords) } diff --git a/modules/mesh-delivery-checker/internal/check/lostask_test.go b/modules/mesh-delivery-checker/internal/check/lostask_test.go index e12566e..35ea6a5 100644 --- a/modules/mesh-delivery-checker/internal/check/lostask_test.go +++ b/modules/mesh-delivery-checker/internal/check/lostask_test.go @@ -18,7 +18,7 @@ import ( type losingAsks struct { *buildrun.FakeAsks answer func(delivery.StepAsk) delivery.Step - dead *DeadLetters + dead *buildrun.DeadLetters never bool mu sync.Mutex @@ -38,9 +38,9 @@ func (l *losingAsks) PublishAsk(ctx context.Context, subject, id string, data [] switch { case l.never: case lost: - h := map[string][]string{DeadHeaderMsgID: {id}, DeadHeaderDeliveries: {"5"}, DeadHeaderStream: {"SEAT_NODE_BUILD_GO"}, - DeadHeaderGaveUp: {"2026-10-26T12:26:40Z"}} - g, ok := GivenUpOf(h) + 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 } @@ -56,8 +56,8 @@ func (l *losingAsks) PublishAsk(ctx context.Context, subject, id string, data [] } func losing(b *bench, lose int) *losingAsks { - l := &losingAsks{FakeAsks: &buildrun.FakeAsks{}, answer: b.answer, dead: &DeadLetters{}, lose: lose} - b.ex.Asks = buildrun.Asker{Publisher: l, Waiter: LostAskWaiter{Inner: l.FakeAsks, Dead: l.dead}, Bound: 5 * time.Second} + 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 } @@ -83,8 +83,8 @@ func TestAnAskGivenUpOrNeverTakenEndsItsTryAsErrorAndItsHeadIsAskedAnew(t *testi never bool want string }{ - {"given up by the build seat", 1, false, LostWords + " after 5 deliveries at 2026-10-26T12:26:40Z"}, - {"taken by no build node within the bound", 0, true, NotTakenWords + " within 200ms"}, + {"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() diff --git a/modules/mesh-delivery-checker/internal/check/try.go b/modules/mesh-delivery-checker/internal/check/try.go index c27c449..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 @@ -399,7 +400,7 @@ func (tr *Tryer) run(ctx context.Context, key string) { Try: &TryOptions{Level: t.Level, Run: t.Tests}} judging := ctx if tr.NotTaken > 0 { - judging = NotTakenBound(ctx, tr.NotTaken) + judging = buildrun.NotTakenBound(ctx, tr.NotTaken) } out, judged := judgeWithin(judging, tr.Judge, cs, tr.Bound) if ctx.Err() != nil { -- 2.54.0 From 60c7ffc981418dd6c1c07777c48bdc33e8f5a64a Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 26 Oct 2026 14:22:15 +0100 Subject: [PATCH 3/4] Checker, builder and grants part: hear DEAD_LETTERS again until heard, say when they do not, and compose the same grant (issue 734, review round 1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit mesh-grants moves to the same mesh-sdk (3203358c), so its shadow user list grants the checker and the builder the DEAD_LETTERS reads the controller's list will, and the take-over is not blocked by a difference. The checker and the builder hear DEAD_LETTERS through buildrun.DeadLetterHearing: in the background, again at every reconnect and every ask waited for while deaf, and keep whether they hear it in their new state hearing, which checker_status and builder_stuck say as dead_letters. checker_try says that a try ended by the not-taken bound leaves its ask in the stream. Not done as asked: a module health tool. The catalogue refuses a tool check on a module with no runtime of its own and no other check beside it (ADR 0227 rule 8, ADR 0366 §1), so the deaf state is said in the status tools and the log. --- .../cmd/mesh-delivery-builder-tools/main.go | 7 +++-- .../cmd/mesh-delivery-builder/main.go | 10 +++++-- modules/mesh-delivery-builder/go.mod | 2 +- modules/mesh-delivery-builder/go.sum | 4 +-- .../internal/build/run_test.go | 9 ++++-- .../internal/build/stuck.go | 19 +++++++++++- .../mesh-delivery-builder/manifest_test.go | 2 +- modules/mesh-delivery-builder/module.json | 1 + .../cmd/mesh-delivery-checker-tools/main.go | 9 ++++-- .../mesh-delivery-checker/integration_test.go | 2 +- .../cmd/mesh-delivery-checker/main.go | 12 ++++++-- .../internal/check/lostask.go | 4 +++ .../internal/check/read.go | 30 +++++++++++++++++-- .../internal/check/read_test.go | 11 +++++-- .../mesh-delivery-checker/manifest_test.go | 2 +- modules/mesh-delivery-checker/module.json | 1 + modules/mesh-grants/go.mod | 2 +- modules/mesh-grants/go.sum | 10 ++----- 18 files changed, 106 insertions(+), 31 deletions(-) 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 c3223b3..7d71872 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -151,8 +151,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro 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{} - defer buildrun.HearDeadLetters(ctx, js, delivery.AskLanguages, dead, logger.Printf)() + 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) @@ -166,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: buildrun.LostAskWaiter{Inner: asks, Dead: dead}, 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 2357313..dd89931 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.20261026130046-d96b9fd66069 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026131841-3203358c4cb5 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 9a97538..ad2ab1a 100644 --- a/modules/mesh-delivery-builder/go.sum +++ b/modules/mesh-delivery-builder/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026130046-d96b9fd66069 h1:dPwZwGbW8LuT4p2baXuiyVNOWU6XO46kbw46VU5TAUQ= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026130046-d96b9fd66069/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026131841-3203358c4cb5 h1:sBSS509nZDRczmKb/IcHoLtFKJR36eZdQoTMK9tScsU= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026131841-3203358c4cb5/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/run_test.go b/modules/mesh-delivery-builder/internal/build/run_test.go index a65a63b..c8fcd3e 100644 --- a/modules/mesh-delivery-builder/internal/build/run_test.go +++ b/modules/mesh-delivery-builder/internal/build/run_test.go @@ -849,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 c36295c..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" ) @@ -105,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"}, @@ -225,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() }}, } @@ -245,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") 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 d3e8e23..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 @@ -795,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 21a0d2c..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}, @@ -199,9 +201,15 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // 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{} - defer buildrun.HearDeadLetters(ctx, js, []string{delivery.AskGo, delivery.AskNode}, dead, logger.Printf)() - waiter := buildrun.LostAskWaiter{Inner: asks, Dead: dead} + 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]}, diff --git a/modules/mesh-delivery-checker/internal/check/lostask.go b/modules/mesh-delivery-checker/internal/check/lostask.go index 4335752..4db60c6 100644 --- a/modules/mesh-delivery-checker/internal/check/lostask.go +++ b/modules/mesh-delivery-checker/internal/check/lostask.go @@ -28,3 +28,7 @@ func IsLostAsk(v delivery.Verdict) bool { 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/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/manifest_test.go b/modules/mesh-delivery-checker/manifest_test.go index de36137..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", diff --git a/modules/mesh-delivery-checker/module.json b/modules/mesh-delivery-checker/module.json index 53e50b4..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 }, diff --git a/modules/mesh-grants/go.mod b/modules/mesh-grants/go.mod index 3d6dc66..423f7c2 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.20261026131841-3203358c4cb5 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..fa3242c 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.20261026131841-3203358c4cb5 h1:sBSS509nZDRczmKb/IcHoLtFKJR36eZdQoTMK9tScsU= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026131841-3203358c4cb5/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= -- 2.54.0 From 3993175e457f94270721657784a2065e6f18e6b0 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 26 Oct 2026 14:45:33 +0100 Subject: [PATCH 4/4] Checker, builder and grants part: pin mesh-sdk main, which holds both the dead-letter hearing (#92) and the placement (#91) (issue 734) --- modules/mesh-delivery-builder/go.mod | 2 +- modules/mesh-delivery-builder/go.sum | 4 ++-- modules/mesh-delivery-checker/go.mod | 2 +- modules/mesh-delivery-checker/go.sum | 10 ++-------- modules/mesh-grants/go.mod | 2 +- modules/mesh-grants/go.sum | 4 ++-- 6 files changed, 9 insertions(+), 15 deletions(-) diff --git a/modules/mesh-delivery-builder/go.mod b/modules/mesh-delivery-builder/go.mod index dd89931..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.20261026131841-3203358c4cb5 + 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 ad2ab1a..9a6ce0a 100644 --- a/modules/mesh-delivery-builder/go.sum +++ b/modules/mesh-delivery-builder/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026131841-3203358c4cb5 h1:sBSS509nZDRczmKb/IcHoLtFKJR36eZdQoTMK9tScsU= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026131841-3203358c4cb5/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/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-grants/go.mod b/modules/mesh-grants/go.mod index 423f7c2..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.20261026131841-3203358c4cb5 + 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 fa3242c..78176f6 100644 --- a/modules/mesh-grants/go.sum +++ b/modules/mesh-grants/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026131841-3203358c4cb5 h1:sBSS509nZDRczmKb/IcHoLtFKJR36eZdQoTMK9tScsU= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026131841-3203358c4cb5/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= -- 2.54.0