From e927469460e4c851835ae3d0473ba4afea61716c Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 23 Oct 2026 16:34:39 +0200 Subject: [PATCH 1/5] Pulse a build ask only from when a node takes it, and end the pulse with its step, so a waiting ask is a backlog and not a stuck row (issue 1908) --- .../integration_test.go | 22 ++-- .../cmd/mesh-delivery-watchman/main.go | 30 +---- .../internal/watch/mirror.go | 64 ++++++---- .../internal/watch/watch_test.go | 114 +++++++++++++----- 4 files changed, 139 insertions(+), 91 deletions(-) diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go index b4a4207..a7b2943 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go @@ -167,23 +167,23 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8 // An ask put on a build seat gets a pulse; one no node takes lets the pulse expire, and that writes a stuck key. askStream, _ := delivery.AskStream(delivery.AskGo) - askSubject, _ := delivery.AskSubject(delivery.AskGo, delivery.AskCheck) - putAsk := func(id string) { + steps := kvs[delivery.StepsBucket("mesh-build-go")] + // A node takes an ask: it writes the step key running (and passed when the ask ends). + putStep := func(id, state string) { t.Helper() - body, err := delivery.EncodeStepAsk(delivery.StepAsk{ID: id, Kind: delivery.AskCheck, Language: delivery.AskGo, Asker: "mesh-delivery-checker", At: time.Now().UTC(), - Check: &delivery.CheckJob{Fetch: map[string]any{"repository": "novox/x", "commit": "0123456789abcdef0123456789abcdef01234567"}, Levels: []string{delivery.LevelUnit}}}) + body, err := delivery.EncodeStep(delivery.Step{ID: id, Tool: "build_go_test", Node: "n1", State: state, Started: time.Now().UTC()}) if err != nil { t.Fatal(err) } - if err := ctl.Publish(askSubject, body); err != nil { + if _, err := steps.Put(ctx, "n1."+id, body); err != nil { t.Fatal(err) } } const ask1, ask2 = "m0000001aaaaaa", "m0000002bbbbbb" - putAsk(ask1) - until("a pulse for the ask waiting", func() bool { _, ok := has(delivery.PulseBucket, pulseOf(askStream, ask1)); return ok }) + putStep(ask1, delivery.StepRunning) + until("a pulse for the ask a node took", func() bool { _, ok := has(delivery.PulseBucket, pulseOf(askStream, ask1)); return ok }) var stuck delivery.Stuck - until("a stuck key for the ask no node took", func() bool { + until("a stuck key for the ask no node ended", func() bool { e, ok := has(delivery.StuckBucket, stuckOf(askStream, ask1)) if !ok { return false @@ -210,11 +210,9 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8 until("node n2 online again", func() bool { on, ok := online("n2"); return ok && on }) until("the pulse of node n2 deleted", func() bool { _, ok := has(delivery.PulseBucket, pulseOf(delivery.PresenceBucket, "n2")); return !ok }) // An ask a node takes at once is the same: no stuck key, however long the bound passes. - putAsk(ask2) + putStep(ask2, delivery.StepRunning) until("a pulse for the second ask", func() bool { _, ok := has(delivery.PulseBucket, pulseOf(askStream, ask2)); return ok }) - if _, err := kvs[delivery.StepsBucket("mesh-build-go")].Put(ctx, "n1."+ask2, []byte(`{}`)); err != nil { - t.Fatal(err) - } + putStep(ask2, delivery.StepPassed) until("the pulse of the second ask deleted", func() bool { _, ok := has(delivery.PulseBucket, pulseOf(askStream, ask2)); return !ok }) time.Sleep(6 * time.Second) // twice the bound and the marker time for _, k := range []string{stuckOf(delivery.PresenceBucket, "n2"), stuckOf(askStream, ask2)} { diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go index ac5ec78..9a9be02 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go @@ -1,5 +1,5 @@ // Command mesh-delivery-watchman is delivery's watchman (novox/hq ADR 0429 §1, ADR 0431, ADR 0433, issue 1135): it -// hears the bus's connect and disconnect events and writes presence, mirrors every move a machine owes (a build ask waiting for a node, +// hears the bus's connect and disconnect events and writes presence, mirrors every move a machine owes (a build ask a node is running, // a node or part gone offline) into the pulse bucket with a time to live, and writes // a stuck key when the server expires one. It holds no timer: every reaction is an event, a watch or a request at // start or at a reconnect, and the server's expiry is the only clock. @@ -179,15 +179,10 @@ func follow(ctx context.Context, w jetstream.KeyWatcher, each func(jetstream.Key } } -// stepTaken is a change of a build module's steps: a step key written for an ask is that ask taken by a node. The node's -// `paused.` key and a key that is no step key are left. +// stepTaken is a change of a build module's steps, handed to the watchman as it is (watch.Step): a running step is its ask +// taken by a node and pulsed, one that ended ends the pulse. An ask still waiting in a stream has no pulse (issue 1908). func stepTaken(w *watch.Watchman, language string, e jetstream.KeyValueEntry) { - if opOf(e) != watch.Put { - return - } - if _, id, err := delivery.ParseStepKey(e.Key()); err == nil { - w.Taken(language, id) - } + w.Step(language, e.Key(), e.Value(), opOf(e)) } // run is the program: it returns when ctx ends or the bus cannot be used. @@ -318,23 +313,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } } } - // The asks put on the build seats' streams: a plain subscription hears each as it is published, as the stream captures it. - for _, lang := range delivery.AskLanguages { - for _, kind := range delivery.AskKinds { - subject, _ := delivery.AskSubject(lang, kind) - lang := lang - if _, err := conn.Subscribe(subject, func(m *nats.Msg) { - a, err := delivery.DecodeStepAsk(m.Data) - if err != nil { - logger.Printf("an ask on %s is not readable: %v", m.Subject, err) - return - } - w.Asked(lang, a.ID, a.At) - }); err != nil { - return err - } - } - } for subject, open := range map[string]bool{connectSubject: true, disconnectSubject: false} { if _, err := conn.Subscribe(subject, onEvent(open)); err != nil { return err diff --git a/modules/mesh-delivery-watchman/internal/watch/mirror.go b/modules/mesh-delivery-watchman/internal/watch/mirror.go index 8297755..485162c 100644 --- a/modules/mesh-delivery-watchman/internal/watch/mirror.go +++ b/modules/mesh-delivery-watchman/internal/watch/mirror.go @@ -2,41 +2,53 @@ package watch import ( "git.novox.be/novox/mesh-sdk/go/delivery" - "time" ) -// A waiting ask is a move (novox/hq ADR 0525 §4, issue 1887 A8b): a message on a build seat's stream (mesh-sdk go/delivery -// AskStream) that a node of the seat must pull. Its pulse is the pulse of the bucket named as the stream, under the ask's id; -// the ask put on the seat is the move beginning at its time, and the ask taken (its step key `.` written in -// the build module's steps, delivery.StepsBucket) is the move done. An ask no node takes within the bound is stuck, as any -// move. The watchman reads no bucket of the controller for it. +// A build ask taken by a node is a move (novox/hq ADR 0525 §4, issues 1887 A8b, 1908, 1909): the node writes the ask's step +// key `.` running in the build module's steps (delivery.StepsBucket) when it takes the ask, and writes it passed, +// failed or refused when the ask ends. The pulse of the move is the pulse of the bucket named as the seat's stream +// (delivery.AskStream), under the ask's id, and it begins when the node started the step. +// +// **An ask still waiting in the stream is a backlog, not a stall** (issue 1908): the rebuild after the switch puts about 170 +// asks on a few nodes' slots, and most wait for hours with every worker healthy. No pulse is made for a waiting ask, so none +// can expire into a false stuck row; the view shows the backlog and the oldest waiting ask's age (delivery view, A8a). What +// is read of the steps at the watchman's start is mirrored the same way, so an ask running when the watchman was down is +// pulsed again (issue 1909). -// AskedKey names a waiting ask for the pulse and stuck keys: the seat's stream and the ask's id. -func askedBucket(language string) (string, error) { return delivery.AskStream(language) } - -// Asked is an ask of a build seat of language put on its stream at time at (the ask's own time). -func (w *Watchman) Asked(language, id string, at time.Time) { - bucket, err := askedBucket(language) +// Step is a change of a key of the steps bucket of language's build module, read from the watch (at the start the keys that +// stand, then each change). A running step makes the pulse of its ask, from the step's own start; a step that ended, or a +// key deleted, ends it. The node's `paused.` key, a key that is no step key and a value that cannot be read are left. +func (w *Watchman) Step(language, key string, value []byte, op Op) { + bucket, err := delivery.AskStream(language) if err != nil { - w.say("no pulse for the ask %s: %v", id, err) + w.say("no pulse for the step %s: %v", key, err) return } - w.mu.Lock() - defer w.mu.Unlock() - if at.IsZero() { - at = w.now() - } - w.mirror(bucket, id, describe("the ask %s to be taken by a node", id), "", at, 0) -} - -// Taken is the ask taken by a node: its step key written. The pulse is deleted, so it never expires, and a stuck key it left is -// written as moved. -func (w *Watchman) Taken(language, id string) { - bucket, err := askedBucket(language) + node, id, err := delivery.ParseStepKey(key) if err != nil { return } w.mu.Lock() defer w.mu.Unlock() - w.moved(bucket, id) + switch op { + case Expired: + return + case Delete: + w.moved(bucket, id) + return + } + step, err := delivery.DecodeStep(value) + if err != nil { + w.say("the step %s cannot be read, and is left: %v", key, err) + return + } + if step.State != delivery.StepRunning { + w.moved(bucket, id) + return + } + since := step.Started + if since.IsZero() { + since = w.now() + } + w.mirror(bucket, id, describe("the ask %s to end on %s", id, node), node, since, 0) } diff --git a/modules/mesh-delivery-watchman/internal/watch/watch_test.go b/modules/mesh-delivery-watchman/internal/watch/watch_test.go index 1c8a6be..8abd76b 100644 --- a/modules/mesh-delivery-watchman/internal/watch/watch_test.go +++ b/modules/mesh-delivery-watchman/internal/watch/watch_test.go @@ -78,6 +78,23 @@ const ( presenceNS = delivery.PresenceBucket ) +// runningStep is a build module's step key as a node writes it when it takes an ask: the ask's id, running, since started. +func runningStep(id string, started time.Time) []byte { + b, err := delivery.EncodeStep(delivery.Step{ID: id, Tool: "build_go_test", Node: "n1", State: delivery.StepRunning, Started: started}) + if err != nil { + panic(err) + } + return b +} + +func endedStep(id, state string, started time.Time) []byte { + b, err := delivery.EncodeStep(delivery.Step{ID: id, Tool: "build_go_test", Node: "n1", State: state, Started: started, Ended: started.Add(time.Second)}) + if err != nil { + panic(err) + } + return b +} + func pulseKeyOf(t *testing.T, bucket, key string) string { t.Helper() k, err := delivery.PulseKey(bucket, key) @@ -207,7 +224,7 @@ func TestRebuildTakesThePresenceOfWhoIsGone(t *testing.T) { // 1135/AC2 func TestRebuildPulsesFromCurrentKeys(t *testing.T) { // 1135/AC2 w, o, c := setup() c.t = t0.Add(30 * time.Second) - w.Asked("go", "askid1", t0.Add(10*time.Second)) + w.Step("go", "n1.askid1", runningStep("askid1", t0.Add(10*time.Second)), Put) w.SeedPresence("ace", false, t0) w.Reconcile() if err := w.Rebuild(func([]byte) ([]byte, error) { return connzAnswer(), nil }); err != nil { @@ -226,7 +243,7 @@ func TestReconcileDropsPulsesNothingMirrorsAnyMore(t *testing.T) { // 1135/AC2 live := pulseKeyOf(t, checksKey, "askid1") w.SeedPulse(stale) w.SeedPulse(live) - w.Asked("go", "askid1", t0) + w.Step("go", "n1.askid1", runningStep("askid1", t0), Put) w.Reconcile() if !slices.Equal(o.pulseDeletes, []string{stale}) { t.Fatalf("deleted = %v, want only %s", o.pulseDeletes, stale) @@ -239,7 +256,7 @@ func TestReconcileDropsPulsesNothingMirrorsAnyMore(t *testing.T) { // 1135/AC2 func TestPulseTimeToLiveIsTheBoundMinusTheAge(t *testing.T) { // 1135/AC3 w, o, c := setup() c.t = t0.Add(30 * time.Second) - w.Asked("go", "askid1", t0) // the key found at start, last renewed 30s ago + w.Step("go", "n1.askid1", runningStep("askid1", t0), Put) // the key found at start, last renewed 30s ago k := pulseKeyOf(t, checksKey, "askid1") if o.ttls[k] != 90*time.Second { t.Fatalf("ttl = %v, want 90s (last renewed 30s ago)", o.ttls[k]) @@ -257,7 +274,7 @@ func TestPulseTimeToLiveIsTheBoundMinusTheAge(t *testing.T) { // 1135/AC3 func TestAMoveAtTheBoundIsStuckAtOnce(t *testing.T) { // 1135/AC3 w, o, c := setup() c.t = t0.Add(120 * time.Second) - w.Asked("go", "askid1", t0) + w.Step("go", "n1.askid1", runningStep("askid1", t0), Put) if len(o.pulses) != 0 { t.Fatalf("a pulse was written for a heartbeat 120s old: %v", o.pulses) } @@ -265,7 +282,7 @@ func TestAMoveAtTheBoundIsStuckAtOnce(t *testing.T) { // 1135/AC3 if !ok { t.Fatalf("no stuck key; have %v", o.stuck) } - if s.Moved || !s.Since.Equal(t0) || s.Waited != "the ask askid1 to be taken by a node" { + if s.Moved || !s.Since.Equal(t0) || s.Waited != "the ask askid1 to end on n1" { t.Fatalf("stuck = %+v", s) } } @@ -273,13 +290,13 @@ func TestAMoveAtTheBoundIsStuckAtOnce(t *testing.T) { // 1135/AC3 func TestWaitedInWords(t *testing.T) { // 1135/AC3 w, o, c := setup() c.t = t0.Add(200 * time.Second) - w.Asked("go", "askid1", t0) + w.Step("go", "n1.askid1", runningStep("askid1", t0), Put) w.SeedPresence("ace", false, t0) w.SeedPresence("part.mesh-delivery-view.ace", false, t0) w.Reconcile() _ = w.Rebuild(func([]byte) ([]byte, error) { return connzAnswer(), nil }) want := map[string]string{ - stuckKeyOf(t, checksKey, "askid1"): "the ask askid1 to be taken by a node", + stuckKeyOf(t, checksKey, "askid1"): "the ask askid1 to end on n1", stuckKeyOf(t, presenceNS, "ace"): "ace to come back online", stuckKeyOf(t, presenceNS, "part.mesh-delivery-view.ace"): "mesh-delivery-view on ace to come back online", } @@ -293,13 +310,13 @@ func TestWaitedInWords(t *testing.T) { // 1135/AC3 func TestEveryHeartbeatRenewsThePulse(t *testing.T) { // 1135/AC3 w, o, c := setup() k := pulseKeyOf(t, checksKey, "askid1") - w.Asked("go", "askid1", t0) + w.Step("go", "n1.askid1", runningStep("askid1", t0), Put) if o.ttls[k] != 120*time.Second { t.Fatalf("ttl = %v, want the full 120s", o.ttls[k]) } o.ttls[k] = 0 c.t = t0.Add(100 * time.Second) - w.Asked("go", "askid1", c.t) // the builder re-puts the key: same check, started long ago + w.Step("go", "n1.askid1", runningStep("askid1", c.t), Put) // the builder re-puts the key: same check, started long ago if o.ttls[k] != 120*time.Second { t.Fatalf("a heartbeat left the pulse as it was: ttl = %v, want the full 120s again", o.ttls[k]) } @@ -308,7 +325,7 @@ func TestEveryHeartbeatRenewsThePulse(t *testing.T) { // 1135/AC3 } // The same revision heard twice changes nothing. o.ttls[k] = 0 - w.Asked("go", "askid1", c.t) + w.Step("go", "n1.askid1", runningStep("askid1", c.t), Put) if o.ttls[k] != 0 { t.Fatal("one revision rewrote the pulse twice") } @@ -317,10 +334,10 @@ func TestEveryHeartbeatRenewsThePulse(t *testing.T) { // 1135/AC3 func TestMirroredMovesDeleteThePulse(t *testing.T) { // 1135/AC4 w, o, c := setup() c.t = t0.Add(10 * time.Second) - w.Asked("go", "askid1", t0) + w.Step("go", "n1.askid1", runningStep("askid1", t0), Put) w.Connect("node.ace") w.Disconnect("node.ace") - w.Taken("go", "askid1") // the check ended + w.Step("go", "n1.askid1", nil, Delete) // the check ended w.Connect("node.ace") if len(o.pulses) != 0 { t.Fatalf("pulses left: %v", o.pulses) @@ -335,9 +352,9 @@ func TestMirroredMovesDeleteThePulse(t *testing.T) { // 1135/AC4 func TestACheckWhoseBuilderStopsIsStuckFromItsLastHeartbeat(t *testing.T) { // 1135/AC4 w, o, c := setup() - w.Asked("go", "askid1", t0) + w.Step("go", "n1.askid1", runningStep("askid1", t0), Put) c.t = t0.Add(50 * time.Second) - w.Asked("go", "askid1", c.t) // the last heartbeat + w.Step("go", "n1.askid1", runningStep("askid1", c.t), Put) // the last heartbeat c.t = t0.Add(170 * time.Second) w.Pulse(pulseKeyOf(t, checksKey, "askid1"), Expired, c.t) s := o.stuck[stuckKeyOf(t, checksKey, "askid1")] @@ -348,7 +365,7 @@ func TestACheckWhoseBuilderStopsIsStuckFromItsLastHeartbeat(t *testing.T) { // 1 func TestAStuckKeyIsWrittenMovedNeverDeleted(t *testing.T) { // 1135/AC4 w, o, c := setup() - w.Asked("go", "askid1", t0) + w.Step("go", "n1.askid1", runningStep("askid1", t0), Put) pk := pulseKeyOf(t, checksKey, "askid1") c.t = t0.Add(120 * time.Second) w.Pulse(pk, Expired, c.t) @@ -357,24 +374,24 @@ func TestAStuckKeyIsWrittenMovedNeverDeleted(t *testing.T) { // 1135/AC4 t.Fatal("stuck written as moved at expiry") } c.t = t0.Add(300 * time.Second) - w.Taken("go", "askid1") + w.Step("go", "n1.askid1", nil, Delete) s := o.stuck[sk] - if !s.Moved || !s.Since.Equal(t0) || s.Waited != "the ask askid1 to be taken by a node" { + if !s.Moved || !s.Since.Equal(t0) || s.Waited != "the ask askid1 to end on n1" { t.Fatalf("stuck after the move = %+v, want the same key moved", s) } n := o.stuckWrites - w.Taken("go", "askid1") + w.Step("go", "n1.askid1", nil, Delete) if o.stuckWrites != n { t.Fatal("a stuck key already moved was written again") } } -func TestAnAskTakenAfterStuckMovesIt(t *testing.T) { // 1135/AC4 +func TestAnAskEndedAfterStuckMovesIt(t *testing.T) { // 1135/AC4 w, o, c := setup() - w.Asked("go", "askid1", t0) + w.Step("go", "n1.askid1", runningStep("askid1", t0), Put) c.t = t0.Add(120 * time.Second) w.Pulse(pulseKeyOf(t, checksKey, "askid1"), Expired, c.t) - w.Taken("go", "askid1") // a node took it at last + w.Step("go", "n1.askid1", nil, Delete) // the ask ended at last if !o.stuck[stuckKeyOf(t, checksKey, "askid1")].Moved { t.Fatal("an ask taken after it was stuck is still stuck") } @@ -398,23 +415,23 @@ func TestPresenceStuckThenBackOnline(t *testing.T) { // 1135/AC4 func TestExpiryMarkerWritesStuck(t *testing.T) { // 1135/AC5 w, o, c := setup() - w.Asked("go", "askid1", t0) + w.Step("go", "n1.askid1", runningStep("askid1", t0), Put) c.t = t0.Add(120 * time.Second) w.Pulse(pulseKeyOf(t, checksKey, "askid1"), Expired, c.t) s, ok := o.stuck[stuckKeyOf(t, checksKey, "askid1")] if !ok { t.Fatalf("no stuck key; have %v", o.stuck) } - if s.Bucket != checksKey || s.Key != "askid1" || s.Waited != "the ask askid1 to be taken by a node" || !s.Since.Equal(t0) || s.Moved { + if s.Bucket != checksKey || s.Key != "askid1" || s.Waited != "the ask askid1 to end on n1" || !s.Since.Equal(t0) || s.Moved { t.Fatalf("stuck = %+v", s) } } func TestOwnDeleteOfAPulseWritesNothing(t *testing.T) { // 1135/AC5 w, o, c := setup() - w.Asked("go", "askid1", t0) + w.Step("go", "n1.askid1", runningStep("askid1", t0), Put) pk := pulseKeyOf(t, checksKey, "askid1") - w.Taken("go", "askid1") + w.Step("go", "n1.askid1", nil, Delete) w.Pulse(pk, Delete, c.t) // the delete the watchman made, heard on its own watch w.Pulse(pk, Expired, c.t) w.Pulse("mesh-delivery-watchman_pulse.unknown.key", Expired, c.t) @@ -427,13 +444,56 @@ func TestTheBoundCanBeShortenedForATest(t *testing.T) { // 1135/AC8 w, o, c := setup() w.SetBound(2 * time.Second) c.t = t0.Add(500 * time.Millisecond) - w.Asked("go", "askid1", t0) + w.Step("go", "n1.askid1", runningStep("askid1", t0), Put) if got := o.ttls[pulseKeyOf(t, checksKey, "askid1")]; got != 1500*time.Millisecond { t.Fatalf("ttl = %v, want 1.5s", got) } c.t = t0.Add(3 * time.Second) - w.Asked("go", "askid2", t0) + w.Step("go", "n1.askid2", runningStep("askid2", t0), Put) if _, ok := o.stuck[stuckKeyOf(t, checksKey, "askid2")]; !ok { t.Fatalf("a heartbeat older than the bound is stuck at once; have %v", o.stuck) } } + +func TestAnAskTakenByANodeHasAPulseFromWhenItStartedAndEndsWithTheStep(t *testing.T) { // 1908, 1909 + w, o, c := setup() + c.t = t0.Add(10 * time.Second) + w.Step("go", "n1.askid1", runningStep("askid1", t0), Put) + if len(o.pulses) != 1 { + t.Fatalf("pulses = %v, want the taken ask's", o.pulses) + } + for _, state := range []string{delivery.StepPassed, delivery.StepFailed, delivery.StepRefused} { + w2, o2, c2 := setup() + c2.t = t0.Add(10 * time.Second) + w2.Step("go", "n1.askid1", runningStep("askid1", t0), Put) + w2.Step("go", "n1.askid1", endedStep("askid1", state, t0), Put) + if len(o2.pulses) != 0 && o2.pulses[pulseKeyOf(t, checksKey, "askid1")].Key != "" { + t.Errorf("%s: the pulse of an ended ask stands: %v", state, o2.pulses) + } + _ = c2 + } + _ = c +} + +func TestARunningAskFoundAtStartGetsItsPulse(t *testing.T) { // 1909 + w, o, c := setup() + c.t = t0.Add(30 * time.Second) + w.Step("go", "n1.askid1", runningStep("askid1", t0), Put) // what the steps held when the watchman started + p, ok := o.pulses[pulseKeyOf(t, checksKey, "askid1")] + if !ok || p.Key != "askid1" { + t.Fatalf("pulses = %v", o.pulses) + } + if got := o.ttls[pulseKeyOf(t, checksKey, "askid1")]; got <= 0 || got >= 120*time.Second { + t.Errorf("ttl = %v, want the bound less its 30s", got) + } +} + +func TestAPausedKeyOrAnotherNodesNonStepIsLeft(t *testing.T) { // 1908 + w, o, _ := setup() + w.Step("go", "paused.n1", []byte(`{}`), Put) + w.Step("go", "n1.not-an-id!", []byte(`{}`), Put) + w.Step("go", "n1.askid1", []byte(`{"v":99}`), Put) + if len(o.pulses) != 0 || len(o.stuck) != 0 { + t.Errorf("pulses %v stuck %v", o.pulses, o.stuck) + } +} From 76bc810bda3ec7c7e1d51ca35fe183849459a70c Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 23 Oct 2026 16:34:50 +0200 Subject: [PATCH 2/5] Test that an ask running when the watchman starts gets its pulse from the steps that stand (issue 1909) --- .../integration_test.go | 81 +++++++++++++++++++ 1 file changed, 81 insertions(+) diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go index a7b2943..5d56874 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go @@ -221,3 +221,84 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8 } } } + +// TestAnAskRunningBeforeTheStartGetsItsPulse puts a running step in the build module's steps before the watchman starts, and +// sees the pulse of its ask once it is up: what stood at the start is mirrored as what happens later is (issue 1909). +func TestAnAskRunningBeforeTheStartGetsItsPulse(t *testing.T) { // 1909 + dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "w") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { os.RemoveAll(dir) }) + conf := filepath.Join(dir, "bus.conf") + if err := os.WriteFile(conf, []byte(fmt.Sprintf(busConf, filepath.Join(dir, "js"))), 0o600); err != nil { + t.Fatal(err) + } + opts, err := server.ProcessConfigFile(conf) + if err != nil { + t.Fatal(err) + } + opts.NoLog, opts.NoSigs = true, true + srv, err := server.NewServer(opts) + if err != nil { + t.Fatal(err) + } + go srv.Start() + if !srv.ReadyForConnections(10 * time.Second) { + t.Fatal("the bus did not come up") + } + t.Cleanup(srv.Shutdown) + ctl, err := nats.Connect(srv.ClientURL(), nats.UserInfo("ctl", "pw")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(ctl.Close) + js, err := jetstream.New(ctl) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + kvs := map[string]jetstream.KeyValue{} + for _, c := range []jetstream.KeyValueConfig{ + {Bucket: delivery.PresenceBucket, History: 20}, + {Bucket: delivery.StuckBucket, History: 20}, + {Bucket: delivery.PulseBucket, LimitMarkerTTL: time.Second}, + {Bucket: delivery.StepsBucket("mesh-build-go"), History: 5}, + } { + kv, err := js.CreateKeyValue(ctx, c) + if err != nil { + t.Fatal(err) + } + kvs[c.Bucket] = kv + } + const id = "m0000009ffffff" + body, err := delivery.EncodeStep(delivery.Step{ID: id, Tool: "build_go_test", Node: "n1", State: delivery.StepRunning, Started: time.Now().UTC()}) + if err != nil { + t.Fatal(err) + } + if _, err := kvs[delivery.StepsBucket("mesh-build-go")].Put(ctx, "n1."+id, body); err != nil { + t.Fatal(err) + } + broker := filepath.Join(dir, "broker") + cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n1", Module: "mesh-delivery-watchman", User: "n1.mesh-delivery-watchman", Password: "pw"}) + if err := os.WriteFile(broker, cred, 0o600); err != nil { + t.Fatal(err) + } + env := map[string]string{"MESH_BROKER_FILE": broker} + runCtx, stop := context.WithCancel(ctx) + done := make(chan error, 1) + go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }() + t.Cleanup(func() { stop(); <-done }) + stream, _ := delivery.AskStream(delivery.AskGo) + pk, err := delivery.PulseKey(stream, id) + if err != nil { + t.Fatal(err) + } + for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { + if _, err := kvs[delivery.PulseBucket].Get(ctx, pk); err == nil { + return + } + } + t.Fatalf("no pulse for the ask running before the start (%s)", pk) +} From c0f24f56b5e773bcc3e8821e26da5b86d152e1ca Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 23 Oct 2026 16:35:05 +0200 Subject: [PATCH 3/5] Drop the view's use of the build seats: it reads their workers and publishes no ask (issue 1887 A8a) --- modules/mesh-delivery-view/manifest_test.go | 6 +++--- modules/mesh-delivery-view/module.json | 5 ----- 2 files changed, 3 insertions(+), 8 deletions(-) diff --git a/modules/mesh-delivery-view/manifest_test.go b/modules/mesh-delivery-view/manifest_test.go index de423f5..871bca3 100644 --- a/modules/mesh-delivery-view/manifest_test.go +++ b/modules/mesh-delivery-view/manifest_test.go @@ -94,9 +94,9 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7 t.Errorf("the manifest declares %q: the view writes only its answers", k) } } - // The view reads the build seats' streams for the backlog of asks (issue 1887 A8a): the three seats it uses. - if want := []string{"node-build-go", "node-build-image", "node-build-node"}; !reflect.DeepEqual(m.Uses, want) { - t.Errorf("uses = %v, want %v", m.Uses, want) + // The view uses no build seat: it reads their workers by the broker's named grant and publishes no ask (ADR 0525 §2). + if len(m.Uses) != 0 { + t.Errorf("uses = %v, want none", m.Uses) } // The one state the view writes: the operator's answers to the asks that need no proof (ADR 0528 clause 6, 1297/B9), which the teller reads. if want := []state{{Name: "answers", History: 5}}; !reflect.DeepEqual(m.State, want) { diff --git a/modules/mesh-delivery-view/module.json b/modules/mesh-delivery-view/module.json index 0d5ed50..61ed274 100644 --- a/modules/mesh-delivery-view/module.json +++ b/modules/mesh-delivery-view/module.json @@ -28,11 +28,6 @@ "mesh-delivery-watchman.stuck", "node-engine.have" ], - "uses": [ - "node-build-go", - "node-build-image", - "node-build-node" - ], "own-secrets": { "broker": "${dir:state}/broker" }, From 84b2c92ccf8cb4170086f9cc002f6784818f65c6 Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 23 Oct 2026 16:39:56 +0200 Subject: [PATCH 4/5] Put again, under its own id, an ask a build key names that neither its stream holds nor any node ran, so a builder that stopped between naming and putting it leaves no key waiting for an ask nobody put (issue 1910) --- .../cmd/mesh-delivery-builder/held.go | 74 +++++++++++++++++++ .../cmd/mesh-delivery-builder/main.go | 1 + .../internal/build/run.go | 21 ++++++ .../internal/build/run_test.go | 35 +++++++++ 4 files changed, 131 insertions(+) create mode 100644 modules/mesh-delivery-builder/cmd/mesh-delivery-builder/held.go diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/held.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/held.go new file mode 100644 index 0000000..6316cfe --- /dev/null +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/held.go @@ -0,0 +1,74 @@ +package main + +import ( + "context" + "errors" + "fmt" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// heldOnTheBus says whether an ask a build key names is on its build seat's stream, or a build node wrote a step for it +// (issue 1910): read once, at the take-over of a key at start, never polled. +type heldOnTheBus struct{ js jetstream.JetStream } + +// Held is build.Builder's Held: a step key `.` in the build module's steps, or a message of the stream under +// the ask's subject whose Nats-Msg-Id is the ask's id. +func (h heldOnTheBus) Held(ctx context.Context, a delivery.StepAsk) (bool, error) { + module, err := delivery.BuildModuleOf(a.Language) + if err != nil { + return false, err + } + pattern, err := delivery.AskStepWatch(a.ID) + if err != nil { + return false, err + } + steps, err := h.js.KeyValue(ctx, delivery.StepsBucket(module)) + if err != nil { + return false, fmt.Errorf("the bucket %s: %w", delivery.StepsBucket(module), err) + } + keys, err := steps.ListKeysFiltered(ctx, pattern) + if err != nil { + return false, err + } + for range keys.Keys() { + keys.Stop() + return true, nil // a node took it + } + subject, err := delivery.AskSubject(a.Language, a.Kind) + if err != nil { + return false, err + } + name, err := h.js.StreamNameBySubject(ctx, subject) + if errors.Is(err, jetstream.ErrStreamNotFound) { + return false, nil + } + if err != nil { + return false, err + } + stream, err := h.js.Stream(ctx, name) + if err != nil { + return false, err + } + // A seat's stream is a work queue, which takes no second consumer over its subjects: its messages are read by sequence + // instead, each a direct read that consumes nothing. + info, err := stream.Info(ctx) + if err != nil { + return false, err + } + for seq := info.State.FirstSeq; seq != 0 && seq <= info.State.LastSeq; seq++ { + m, err := stream.GetMsg(ctx, seq) + if errors.Is(err, jetstream.ErrMsgNotFound) { + continue // taken and acknowledged, or removed + } + if err != nil { + return false, err + } + if m.Subject == subject && m.Header.Get(jetstream.MsgIDHeader) == a.ID { + return true, nil + } + } + return false, nil +} 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 6d597a3..0cd20c8 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -155,6 +155,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro b := &build.Builder{ Claimer: build.Claimer{Store: kvStore{builds}, By: cred.Node + "." + cred.Module}, Asks: buildrun.Asker{Publisher: asks, Waiter: asks, Bound: bound}, + Held: heldOnTheBus{js}, Toolchains: tools, Manifests: manifestReader{manifests}, Logf: logger.Printf, diff --git a/modules/mesh-delivery-builder/internal/build/run.go b/modules/mesh-delivery-builder/internal/build/run.go index 64e3bcd..9ca9dbf 100644 --- a/modules/mesh-delivery-builder/internal/build/run.go +++ b/modules/mesh-delivery-builder/internal/build/run.go @@ -35,6 +35,12 @@ type Builder struct { Put(ctx context.Context, a delivery.StepAsk) error Wait(ctx context.Context, a delivery.StepAsk) (delivery.Step, error) } + // Held says whether an ask a key names is on its stream or a build node wrote a step for it (issue 1910): a key taken + // over at start whose ask is neither was left by a builder that stopped between naming the ask and putting it, and + // the ask is put then, under the same id. Nil takes every named ask as held. + Held interface { + Held(ctx context.Context, a delivery.StepAsk) (bool, error) + } Toolchains ToolchainEntries Manifests Manifests // Limit is how many builds the builder has asked at once; their asks wait in the build seats' streams, which are the @@ -668,6 +674,21 @@ func (b *Builder) runLane(ctx context.Context, l *lane, repository string, m del ask := delivery.StepAsk{ID: buildstep.NewID(), Kind: kind, Language: language, Asker: delivery.BuilderModule, At: time.Now(), Build: job} if id := asked.resume(language); id != "" { ask.ID = id // put before this builder started again: the stream holds it, or a node already ran it + if b.Held != nil { + held, err := b.Held.Held(ctx, ask) + if err != nil { + l.err = fmt.Errorf("whether the ask %s of %s the key names is on its stream cannot be read: %w", ask.ID, l.module, err) + return + } + if !held { + // The builder that named it stopped before it put it (issue 1910): put it now, under the same id. + b.logf("the ask %s of %s that the key names was never put: it is put now", ask.ID, l.module) + if err := b.Asks.Put(ctx, ask); err != nil { + l.err = err + return + } + } + } } else { if err := asked.note(ctx, language, ask.ID, "asked "+l.module); err != nil { l.err = fmt.Errorf("the ask of %s could not be written to the key: %w", l.module, err) diff --git a/modules/mesh-delivery-builder/internal/build/run_test.go b/modules/mesh-delivery-builder/internal/build/run_test.go index ba94b0e..4cc1a5c 100644 --- a/modules/mesh-delivery-builder/internal/build/run_test.go +++ b/modules/mesh-delivery-builder/internal/build/run_test.go @@ -613,3 +613,38 @@ func TestAFailedPublishOfACandidatesImageIsWrittenToTheKey(t *testing.T) { // re t.Errorf("build = %+v, want done with a reference", v) } } + +// heldAsks says an ask is held when its id is among held. +type heldAsks map[string]bool + +func (h heldAsks) Held(_ context.Context, a delivery.StepAsk) (bool, error) { return h[a.ID], nil } + +// 1910: a builder that stopped between naming the ask in its key and putting it left a key naming an ask neither the +// stream holds nor any node ran: started again, the builder puts that ask, under the same id, once, and the key is +// built from its step; an ask the stream holds is not put again. +func TestAnAskTheKeyNamesAndNobodyPutIsPutAtStart(t *testing.T) { + for _, held := range []bool{false, true} { + r := newRig(t, goBundle) + const id = "mvl0frv18d1958" + data, _ := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildBuilding, By: "novox.builder", Manifest: manifestD, + Step: "asked mesh-build-go", Asks: map[string]string{delivery.AskGo: id}}) + r.st.vals[r.key] = data + r.b.Held = heldAsks{id: held} + if held { + r.asks.Answer = nil + r.asks.End(delivery.Step{ID: id, Node: "shanks", State: delivery.StepPassed, + Result: []byte(`{"artifacts":[{"artifact":"tool","digest":"` + digest + `","reference":"artifact-store://m/tool/blobs/` + digest + `"}]}`)}) + } + v := r.consider(t, "modules/m") + asks := r.asks.Asks() + switch { + case held && len(asks) != 0: + t.Errorf("an ask the stream holds was put again: %+v", asks) + case !held && (len(asks) != 1 || asks[0].ID != id): + t.Errorf("the ask nobody put: asks %+v, want one under %s", asks, id) + } + if v.State != delivery.BuildDone { + t.Errorf("held %v: the key = %+v, want done", held, v) + } + } +} From 37860d46a0b6a682f020071715c3ae358e5e26fe Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 23 Oct 2026 16:41:34 +0200 Subject: [PATCH 5/5] Stop watching the node-engine's have in the checker: a check's asks wait in the build seats' streams, which redeliver a lost node's ask, so no check fails for want of a node; presence stays for an unheard forge and the assigner's presence (issue 1912) --- .../cmd/mesh-delivery-checker/main.go | 29 ++++--------------- .../cmd/mesh-delivery-checker/nodes_test.go | 21 ++++++++++++++ .../internal/check/checker.go | 10 +++---- .../internal/check/checker_test.go | 6 ++-- .../internal/check/rank_test.go | 2 +- 5 files changed, 35 insertions(+), 33 deletions(-) create mode 100644 modules/mesh-delivery-checker/cmd/mesh-delivery-checker/nodes_test.go 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 6e1f203..34db5d7 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go @@ -197,7 +197,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } tools := &buildrun.ToolchainSet{} presence := &buildrun.PresenceSet{} - have := &buildrun.HaveSet{} trunks := &check.Trunks{} // 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. @@ -279,8 +278,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro defer cc.Stop() } errs := make(chan error, 8) - // The watchman's presence says which nodes are asleep. Where its bucket is not there yet no node is online, so nothing - // is checked; the checker says so and is started again with the watchman. + // The watchman's presence: a node coming back online asks again a check whose forge was not heard (UnheardWords), and + // the assigner's own presence says whether a composition is being made (compositionWatch, ADR 0537 §4). A check is no + // longer held for a node: its asks wait in the build seats' streams, which redeliver a lost node's ask (ADR 0525 §6, + // issue 1912), so the node-engine's have is not watched here. switch kv, err := js.KeyValue(ctx, delivery.PresenceBucket); { case errors.Is(err, jetstream.ErrBucketNotFound): logger.Printf("the bucket %s does not exist yet: no node counts as online until the checker starts again", delivery.PresenceBucket) @@ -294,27 +295,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro go func() { errs <- follow(ctx, pw, func(e jetstream.KeyValueEntry) { if presence.Put(e.Key(), e.Value(), !opOf(e)) { - ch.NodesChanged(ctx) // a node came online or went: a check that waited for one may go on - } - }) - }() - } - // The node-engine's have says whether a node's build module runs and is healthy: a node whose build module is not is not - // chosen. Where its bucket is not there yet no build module counts as healthy, so nothing is checked. - switch kv, err := js.KeyValue(ctx, delivery.HaveBucket); { - case errors.Is(err, jetstream.ErrBucketNotFound): - logger.Printf("the bucket %s does not exist yet: no build module counts as healthy until the checker starts again", delivery.HaveBucket) - case err != nil: - return fmt.Errorf("the bucket %s: %w", delivery.HaveBucket, err) - default: - hw, err := settle(ctx, kv, func(e jetstream.KeyValueEntry) { have.Put(e.Key(), e.Value(), !opOf(e)) }) - if err != nil { - return err - } - go func() { - errs <- follow(ctx, hw, func(e jetstream.KeyValueEntry) { - if have.Put(e.Key(), e.Value(), !opOf(e)) { - ch.NodesChanged(ctx) + ch.NodesChanged(ctx) // a node came online or went: a check whose forge was not heard may now be heard } }) }() diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/nodes_test.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/nodes_test.go new file mode 100644 index 0000000..2a7b78d --- /dev/null +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/nodes_test.go @@ -0,0 +1,21 @@ +package main + +import ( + "os" + "strings" + "testing" +) + +// 1912: with checks as asks on the build seats' streams, the stream redelivers a lost node's ask (ADR 0525 §6), so the +// checker program opens no node-engine have to start again a check that failed for want of a node. +func TestTheCheckerWatchesNoHave(t *testing.T) { + raw, err := os.ReadFile("main.go") + if err != nil { + t.Fatal(err) + } + for _, s := range []string{"delivery.HaveBucket", "buildrun.HaveSet"} { + if strings.Contains(string(raw), s) { + t.Errorf("the checker program still reads %s", s) + } + } +} diff --git a/modules/mesh-delivery-checker/internal/check/checker.go b/modules/mesh-delivery-checker/internal/check/checker.go index 577ed0e..6b3e220 100644 --- a/modules/mesh-delivery-checker/internal/check/checker.go +++ b/modules/mesh-delivery-checker/internal/check/checker.go @@ -14,10 +14,6 @@ import ( // DefaultBound is how long a check may run before it is ended failed naming the step it was stuck at (ADR 0484, issue 1678). const DefaultBound = 30 * time.Minute -// NodesWords is in the Words of a failure that is the build nodes' (none qualifies, the toolchains disagree): such a verdict -// is judged again when the nodes change, by a watch and never a timer (ADR 0437 §8). -const NodesWords = "build node qualifies" - // UnheardWords end the Words of a failure because a part of the mesh the check reads was not heard (the forge, issue 1624): // like a failure of the nodes, it is not the change's, and is judged again when the nodes change. const UnheardWords = "not heard; judged again when the nodes change" @@ -307,9 +303,11 @@ func (ch *Checker) claim(ctx context.Context, cs *Case, first bool, epoch uint64 return &claimed{rev}, nil } -// IsNodesFailure says whether a verdict failed because no build node could take the check. +// IsNodesFailure says whether a verdict failed because a part of the mesh the check reads was not heard (UnheardWords): +// it is judged again when the nodes change. No check fails for want of a build node any more: its asks wait in the build +// seats' streams (ADR 0525 §6, issue 1912). func IsNodesFailure(v delivery.Verdict) bool { - return v.State == delivery.VerdictFail && (strings.Contains(v.Failed.Words, NodesWords) || strings.Contains(v.Failed.Words, UnheardWords)) + return v.State == delivery.VerdictFail && strings.Contains(v.Failed.Words, UnheardWords) } // try claims the check, judges it within its bound and writes the verdict and the owed statuses. A check that ends diff --git a/modules/mesh-delivery-checker/internal/check/checker_test.go b/modules/mesh-delivery-checker/internal/check/checker_test.go index 1184a1b..e8126c9 100644 --- a/modules/mesh-delivery-checker/internal/check/checker_test.go +++ b/modules/mesh-delivery-checker/internal/check/checker_test.go @@ -208,11 +208,13 @@ func TestACheckPastItsBoundFailsNamingTheStuckStep(t *testing.T) { // N6/AC3 } } -func TestNoQualifyingNodeFailsWithTheReasonAndIsJudgedAgainWhenTheNodesChange(t *testing.T) { // N6/AC4 +// A check whose forge was not heard is judged again when the nodes change (issue 1912: no check fails for want of a build +// node any more; its asks wait in the streams). +func TestAnUnheardForgeFailsWithTheReasonAndIsJudgedAgainWhenTheNodesChange(t *testing.T) { // N6/AC4 ctx := context.Background() j := &judgeFunc{f: func(_ context.Context, _ *Case, n int) (Outcome, error) { if n == 1 { - return Outcome{Verdict: failure("shanks", "no go "+NodesWords+": none is online")}, nil + return Outcome{Verdict: failure("the forge", "novox/mesh-delivery#7: go.mod cannot be read: no responders (the forge was "+UnheardWords+")")}, nil } return Outcome{Verdict: passing()}, nil }} diff --git a/modules/mesh-delivery-checker/internal/check/rank_test.go b/modules/mesh-delivery-checker/internal/check/rank_test.go index 10d306c..801fe8e 100644 --- a/modules/mesh-delivery-checker/internal/check/rank_test.go +++ b/modules/mesh-delivery-checker/internal/check/rank_test.go @@ -39,7 +39,7 @@ func TestCheckerStatusNamesTheWaitingChecksByKind(t *testing.T) { // 1897/AC4 ctx := context.Background() j := &judgeFunc{f: func(_ context.Context, c *Case, _ int) (Outcome, error) { if c.Group == group(2) { - return Outcome{Verdict: failure("checker", "no "+NodesWords+": every node is offline")}, nil + return Outcome{Verdict: failure("the forge", "go.mod cannot be read: no responders (the forge was "+UnheardWords+")")}, nil } return Outcome{Verdict: passing()}, nil }}