diff --git a/go/delivery/buildrun/asks_test.go b/go/delivery/buildrun/asks_test.go index 3365202..44281b1 100644 --- a/go/delivery/buildrun/asks_test.go +++ b/go/delivery/buildrun/asks_test.go @@ -48,6 +48,35 @@ func TestAnAskerPutsOnceAndReadsTheStep(t *testing.T) { if len(g.Asks()) != 0 { t.Fatal("a wait put an ask") } + // Issue 734: an ask the build seat gave up on (DEAD_LETTERS keeps its copy with the ask's id) ends its wait as failed + // with the words, though a node had taken it; under NotTakenBound one no node takes ends at the bound; a step written + // first is answered as the node wrote it. + dead := &DeadLetters{} + lw := LostAskWaiter{Inner: g, Dead: dead} + k = Asker{Publisher: g, Waiter: lw, Bound: time.Minute} + g.Write(delivery.Step{ID: "mvl0frv18d1970", Node: "ace", State: delivery.StepRunning, Started: time.Now()}) + given, ok := GivenUpOf(map[string][]string{"Mesh-Dead-Msg-Id": {"mvl0frv18d1970"}, "Mesh-Dead-Deliveries": {"5"}, + "Mesh-Dead-Stream": {"SEAT_NODE_BUILD_GO"}, "Mesh-Dead-Gave-Up": {"2026-10-26T12:26:40Z"}}) + if !ok { + t.Fatal("a dead letter's headers were not read") + } + go dead.Heard(given) + step, err = k.Wait(context.Background(), goAsk("mvl0frv18d1970")) + if err != nil || step.State != delivery.StepFailed || !IsLost(step) || + !strings.Contains(step.Reason, LostWords+" after 5 deliveries at 2026-10-26T12:26:40Z") { + t.Fatalf("a given-up ask: %+v, %v", step, err) + } + if _, err := k.Passed(step, nil); err == nil || !strings.Contains(err.Error(), "on "+LostNode+" failed: "+LostWords) { + t.Fatalf("a given-up ask's error: %v", err) + } + step, err = k.Wait(NotTakenBound(context.Background(), 20*time.Millisecond), goAsk("mvl0frv18d1971")) + if err != nil || !IsNotTaken(step) || IsLost(step) { + t.Fatalf("an ask no node took: %+v, %v", step, err) + } + g.Write(delivery.Step{ID: "mvl0frv18d1972", Node: "g14", State: delivery.StepPassed}) + if step, err = k.Wait(NotTakenBound(context.Background(), time.Minute), goAsk("mvl0frv18d1972")); err != nil || step.Node != "g14" { + t.Fatalf("a step written: %+v, %v", step, err) + } } // Review (issue 734 path): an ask taken again by the same node (its build module replaced mid-run) is bounded from the diff --git a/go/delivery/buildrun/lostask.go b/go/delivery/buildrun/lostask.go new file mode 100644 index 0000000..52f7ee8 --- /dev/null +++ b/go/delivery/buildrun/lostask.go @@ -0,0 +1,296 @@ +package buildrun + +import ( + "context" + "fmt" + "strconv" + "strings" + "sync" + "time" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// **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. +// +// An asker of the build seats (the checker, the builder) hears those copies (HearDeadLetters: an ordered consumer on +// DEAD_LETTERS filtered to the build seats' streams, granted to a user of a seat pulled by kind, buscompose +// DeadLetterReads) into DeadLetters, and LostAskWaiter ends the waiting step at once as failed on the node LostNode, its +// reason saying LostWords; the asker decides what follows (the checker ends a try as error and asks a check again, the +// builder asks its part of a build again). +// +// Belt and braces, where the asker says so in the context (NotTakenBound; the checker's tries): an ask no build node took +// within a bound ends the same way, saying NotTakenWords. An ask waits in the stream as long as the backlog is otherwise +// (ADR 0525 §9): a backlog is no fault. + +// LostNode is the node a step ended for a lost ask names: no build node wrote it. +const LostNode = "bus" + +// The words of a step ended for its ask. +const ( + LostWords = "its ask was given up by the build seat" + NotTakenWords = "no build node took its ask" +) + +// 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 (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 + ".>" } + +// SeatStream is the stream of a build seat's language (buscompose seatStreamName of node-build-). +func SeatStream(language string) string { return "SEAT_NODE_BUILD_" + strings.ToUpper(language) } + +// 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 an 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 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: LostNode, 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 +} + +// IsLost says whether a step was ended by LostAskWaiter because the build seat gave its ask up (LostWords). +func IsLost(s delivery.Step) bool { return s.Node == LostNode && strings.Contains(s.Reason, LostWords) } + +// IsNotTaken says whether a step was ended by LostAskWaiter because no build node took its ask within the bound. +func IsNotTaken(s delivery.Step) bool { + return s.Node == LostNode && strings.Contains(s.Reason, NotTakenWords) +} + +// HearDeadLetters follows, for each language's build seat stream, the copies DEAD_LETTERS keeps of the asks its workers +// gave up on, into dead: an ordered consumer from now, which resumes where it was after a reconnect, and starts here again +// when it is reset before it heard any, so a bus reload loses none. Where the bus refuses it (the grant not yet composed) +// it says so and goes on; the stop it answers ends every consumer it started. +func HearDeadLetters(ctx context.Context, js jetstream.JetStream, languages []string, dead *DeadLetters, logf func(string, ...any)) func() { + var stops []func() + since := time.Now() + for _, language := range languages { + stream := SeatStream(language) + filter := DeadLetterFilter(stream) + notHeard := func(err error) { + logf("the asks %s gives up on are not heard (%s of %s), so a wait for one ends only at its bound: %v", + stream, filter, DeadLettersStream, err) + } + cons, err := js.OrderedConsumer(ctx, 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 := GivenUpOf(m.Headers()); ok { + logf("the ask %s on %s was given up after %d deliveries at %s: the wait for its step 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", DeadLettersStream, err) })) + if err != nil { + notHeard(err) + continue + } + stops = append(stops, cc.Stop) + } + return func() { + for _, stop := range stops { + stop() + } + } +}