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 {