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" ],