From e6f91ed3494ad424e2cd6e458d28be2cdd0bb768 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 02:18:50 +0100 Subject: [PATCH] Watchman: time a build seat's ask from its step-progress, not from the holders' steps (fix 10, A-136) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §483, §496, §602, A-136 --- .../cmd/mesh-delivery-watchman/consumes.go | 24 +++ .../integration_test.go | 60 ++++---- .../cmd/mesh-delivery-watchman/main.go | 43 ------ .../internal/watch/findings_test.go | 5 +- .../internal/watch/mirror.go | 100 ++++++++----- .../internal/watch/names.go | 3 - .../internal/watch/watch.go | 10 +- .../internal/watch/watch_test.go | 138 ++++++++++-------- .../mesh-delivery-watchman/manifest_test.go | 28 ++-- modules/mesh-delivery-watchman/module.json | 17 ++- 10 files changed, 225 insertions(+), 203 deletions(-) diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go index 13e2c919..b3b39e33 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go @@ -8,6 +8,7 @@ import ( "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo" "strings" "sync" @@ -46,6 +47,7 @@ func consumed(w *watch.Watchman, subject string, body []byte, logf func(string, w.Accepted(a) case pipelineSubject(w, subject, body): case recordSubject(w, subject, body): + case buildSeatSubject(w, subject, body): case brokerSubject(subject) && strings.HasSuffix(subject, ".event."+eventCreated): if s, err := watch.ReadBrokerSaid(body); err != nil { logf("%v; left", err) @@ -187,3 +189,25 @@ func recordSubject(w *watch.Watchman, subject string, body []byte) bool { } return true } + +// buildSeatSubject hands the build seats' `ask-taken`, `step-progress` and `step-ended`, on +// mesh.seat..event.., to the watchman (design 47 G §483, §496, §602, A-136): it times every ask of a +// build seat from them. It says whether the subject was one of them. +func buildSeatSubject(w *watch.Watchman, subject string, body []byte) bool { + parts := strings.Split(subject, ".") + if len(parts) != 6 || parts[0] != "mesh" || parts[1] != "seat" || parts[3] != "event" || !strings.HasPrefix(parts[2], "node-build-") { + return false + } + seat, node := parts[2], parts[5] + switch parts[4] { + case nodebuildgo.EventAskTaken: + w.AskTaken(seat, node, body) + case nodebuildgo.EventStepProgress: + w.StepProgress(seat, node, body) + case nodebuildgo.EventStepEnded: + w.AskEnded(seat, body) + default: + return false + } + return true +} 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 97ef4b04..15c84ddc 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 @@ -69,13 +69,27 @@ func serveOwners(t *testing.T, ctx context.Context, js jetstream.JetStream, nc * t.Helper() viewstest.Serve(t, ctx, js, nc, watch.AssignerModule, wantsState, watch.AssignerModule+".assigner_read_wants", "n1_"+watch.WatchmanModule) viewstest.Serve(t, ctx, js, nc, watch.IndexerModule, modulesState, watch.IndexerModule+".indexer_read_modules", "n1_"+watch.WatchmanModule) - // The seat that places modules says who holds each build seat (design 47 G A-126): the module whose steps the watchman watches. - held, _ := json.Marshal(map[string]any{"result": map[string]any{"seats": []map[string]any{ - {"seat": "node-build-go", "holders": []map[string]string{{"module": "mesh-build-go"}}}}}, "node": "ctl"}) - for _, subject := range []string{"mesh.seat.mesh-module-assigner.call.assignments", "mesh.seat.mesh-module-assigner.call.assignments.>"} { - if _, err := nc.Subscribe(subject, func(m *nats.Msg) { _ = m.Respond(held) }); err != nil { - t.Fatal(err) - } +} + +// makeEvents makes EVENTS and the watchman's durable consumer of the build seats' events, as the controller and the grants part +// make them (design 47 G §483, §496, §602); the owners served before it made EVENTS for the modules' events. +func makeEvents(t *testing.T, ctx context.Context, js jetstream.JetStream) { + t.Helper() + if _, err := js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>", "mesh.seat.*.event.>"}}); err != nil { + t.Fatal(err) + } + if _, err := js.CreateOrUpdateConsumer(ctx, "EVENTS", jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-watchman", + AckPolicy: jetstream.AckExplicitPolicy}); err != nil { + t.Fatal(err) + } +} + +// sayAsk is what the holder of the Go build seat says on node n1 about an ask: ask-taken, step-progress or step-ended. +func sayAsk(t *testing.T, ctx context.Context, js jetstream.JetStream, event, id string) { + t.Helper() + body, _ := json.Marshal(map[string]any{"ask": id, "asker": "a1", "at": time.Now().UTC(), "phase": "build"}) + if _, err := js.Publish(ctx, "mesh.seat.node-build-go.event."+event+".n1", body); err != nil { + t.Fatal(err) } } @@ -135,7 +149,6 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8 {Bucket: watch.PresenceBucket, History: 20}, {Bucket: watch.StuckBucket, History: 20}, {Bucket: watch.PulseBucket, LimitMarkerTTL: time.Second}, - {Bucket: watch.StepsBucket("mesh-build-go"), History: 5}, {Bucket: watch.AssignerWantsBucket, History: 20}, {Bucket: watch.HaveBucket, History: 20}, {Bucket: watch.FindingsBucket, History: 5}, @@ -149,6 +162,7 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8 } serveOwners(t, ctx, js, ctl) findingsStream(t, ctx, js) + makeEvents(t, ctx, js) // The controller's consumer on DEAD_LETTER_NOTICES, the dead letters' streams and a source stream with one message a // consumer will give up on (1952/WM5); no lease bucket at all (issue 1126). @@ -237,20 +251,8 @@ 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, _ := buildstep.AskStream(buildstep.AskGo) - steps := kvs[watch.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 := buildstep.EncodeStep(buildstep.Step{ID: id, Tool: "build_go_test", Node: "n1", State: state, Started: time.Now().UTC()}) - if err != nil { - t.Fatal(err) - } - if _, err := steps.Put(ctx, "n1."+id, body); err != nil { - t.Fatal(err) - } - } const ask1, ask2 = "m0000001aaaaaa", "m0000002bbbbbb" - putStep(ask1, buildstep.StepRunning) + sayAsk(t, ctx, js, "ask-taken", ask1) until("a pulse for the ask a node took", func() bool { _, ok := has(watch.PulseBucket, pulseOf(askStream, ask1)); return ok }) var stuck meshwatchdogsclient.Stuck until("a stuck key for the ask no node ended", func() bool { @@ -280,9 +282,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(watch.PulseBucket, pulseOf(watch.PresenceBucket, "n2")); return !ok }) // An ask a node takes at once is the same: no stuck key, however long the bound passes. - putStep(ask2, buildstep.StepRunning) + sayAsk(t, ctx, js, "ask-taken", ask2) until("a pulse for the second ask", func() bool { _, ok := has(watch.PulseBucket, pulseOf(askStream, ask2)); return ok }) - putStep(ask2, buildstep.StepPassed) + sayAsk(t, ctx, js, "step-ended", ask2) until("the pulse of the second ask deleted", func() bool { _, ok := has(watch.PulseBucket, pulseOf(askStream, ask2)); return !ok }) time.Sleep(6 * time.Second) // twice the bound and the marker time for _, k := range []string{stuckOf(watch.PresenceBucket, "n2"), stuckOf(askStream, ask2)} { @@ -387,7 +389,6 @@ func anAskRunningBeforeTheStartGetsItsPulse(t *testing.T) { // 1909 {Bucket: watch.PresenceBucket, History: 20}, {Bucket: watch.StuckBucket, History: 20}, {Bucket: watch.PulseBucket, LimitMarkerTTL: time.Second}, - {Bucket: watch.StepsBucket("mesh-build-go"), History: 5}, {Bucket: watch.AssignerWantsBucket, History: 20}, {Bucket: watch.HaveBucket, History: 20}, {Bucket: watch.FindingsBucket, History: 5}, @@ -401,14 +402,9 @@ func anAskRunningBeforeTheStartGetsItsPulse(t *testing.T) { // 1909 } serveOwners(t, ctx, js, ctl) findingsStream(t, ctx, js) + makeEvents(t, ctx, js) const id = "m0000009ffffff" - body, err := buildstep.EncodeStep(buildstep.Step{ID: id, Tool: "build_go_test", Node: "n1", State: buildstep.StepRunning, Started: time.Now().UTC()}) - if err != nil { - t.Fatal(err) - } - if _, err := kvs[watch.StepsBucket("mesh-build-go")].Put(ctx, "n1."+id, body); err != nil { - t.Fatal(err) - } + sayAsk(t, ctx, js, "ask-taken", id) broker := filepath.Join(dir, "broker") cred, _ := json.Marshal(process.Account{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 { @@ -475,7 +471,6 @@ func aWantIsMirroredAsAPulseAndTheHaveNamingItEndsIt(t *testing.T) { {Bucket: watch.PresenceBucket, History: 20}, {Bucket: watch.StuckBucket, History: 20}, {Bucket: watch.PulseBucket, LimitMarkerTTL: time.Second}, - {Bucket: watch.StepsBucket("mesh-build-go"), History: 5}, {Bucket: watch.AssignerWantsBucket, History: 20}, {Bucket: watch.HaveBucket, History: 20}, {Bucket: watch.FindingsBucket, History: 5}, @@ -489,6 +484,7 @@ func aWantIsMirroredAsAPulseAndTheHaveNamingItEndsIt(t *testing.T) { } serveOwners(t, ctx, js, ctl) findingsStream(t, ctx, js) + makeEvents(t, ctx, js) fpNew := meshcatalogclient.SourcePrefix2 + strings.Repeat("b", 64) fpOld := meshcatalogclient.SourcePrefix2 + strings.Repeat("a", 64) putWant := func(node, module, fp string) { 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 9e92c521..f3332391 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go @@ -19,7 +19,6 @@ import ( "time" "git.novox.be/novox/mesh-sdk/go/process" - "git.novox.be/novox/mesh-sdk/go/seatlist" "git.novox.be/novox/mesh-sdk/go/views" "github.com/nats-io/nats.go" @@ -27,7 +26,6 @@ import ( "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshwatchdogsclient" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein" - "git.novox.be/novox/mesh-delivery/sdk/buildstep" "mesh-delivery-watchman/internal/socket" "mesh-delivery-watchman/internal/watch" @@ -180,12 +178,6 @@ func follow(ctx context.Context, w jetstream.KeyWatcher, each func(jetstream.Key } } -// 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) { - w.Step(language, e.Key(), e.Value(), opOf(e)) -} - // createdOr is when an entry was written, or at when its picture does not say. mesh-sdk's views.Picture (the join's J3) // names each first-view key's written time as its Created, and the tool runner's state reads carry it (J2), so a want // that stood before the start waits from when it was written; at is left only for an owner whose answer names no time. @@ -342,37 +334,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro w.ReadWhole(watch.BuilderBuildsBucket) } } - // The build seats' steps: a step key `.` written is the ask taken (issue 1887 A8b). A seat whose build module - // keeps no steps bucket (yet) is said and left; the asks of its stream are mirrored all the same. - taken := map[string]jetstream.KeyWatcher{} - for _, lang := range buildstep.AskLanguages { - seat, err := buildstep.BuildSeat(lang) - if err != nil { - continue - } - // The build module is the holder of the build seat (design 47 G A-82, A-126): asked of the seat that places modules. - module, err := seatlist.HolderOf(calls.Call, seat) - if err != nil { - logger.Printf("the steps of %s cannot be read (%v): its asks are mirrored and never seen taken", seat, err) - continue - } - kv, err := js.KeyValue(ctx, watch.StepsBucket(module)) - if err != nil { - logger.Printf("the steps of %s cannot be read (%v): its asks are mirrored and never seen taken", module, err) - continue - } - lang := lang - sw, err := settle(ctx, kv, func(e jetstream.KeyValueEntry) { stepTaken(w, lang, e) }) - if err != nil { - logger.Printf("the steps of %s cannot be watched (%v): its asks are mirrored and never seen taken", module, err) - continue - } - taken[lang] = sw - if bucket, err := buildstep.AskStream(lang); err == nil { - w.ReadWhole(bucket) - } - } - // The events first, then the one request of the open connections, so no connection falls between them. onEvent := func(open bool) nats.MsgHandler { return func(m *nats.Msg) { @@ -535,10 +496,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro errs <- follow(ctx, buildSeen, func(e jetstream.KeyValueEntry) { w.Build(e.Key(), e.Value(), opOf(e), e.Created()) }) }() } - for lang, sw := range taken { - lang, sw := lang, sw - go func() { errs <- follow(ctx, sw, func(e jetstream.KeyValueEntry) { stepTaken(w, lang, e) }) }() - } go func() { for { select { diff --git a/modules/mesh-delivery-watchman/internal/watch/findings_test.go b/modules/mesh-delivery-watchman/internal/watch/findings_test.go index fea28393..dcdeb8cb 100644 --- a/modules/mesh-delivery-watchman/internal/watch/findings_test.go +++ b/modules/mesh-delivery-watchman/internal/watch/findings_test.go @@ -17,7 +17,6 @@ import ( "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein" - "git.novox.be/novox/mesh-delivery/sdk/buildstep" "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient/heard" "git.novox.be/novox/mesh-contracts/provisioning/go/clients/meshprovisionerclient" @@ -360,13 +359,13 @@ func TestTheWatchmansFindings(t *testing.T) { w.Have("ace.prov-y", nil, Delete) }, map[string]string{}}, {"WM4: a stuck key not moved is its finding, and its move deletes it", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { - w.Step("go", "n1.ask1", runningStep("ask1", t0), Put) + askProgress(w, "ask1", t0) expire(w, c, pulseKeyOf(t, checksKey, "ask1"), 120*time.Second) k := FindingKey(ScopeDelivery, stuckKeyOf(t, checksKey, "ask1"), KindStuck) if f := o.findings[k]; f.Kind != KindStuck || !strings.Contains(f.Summary, "the ask ask1 to end on n1") { t.Fatalf("findings = %v", o.findings) } - w.Step("go", "n1.ask1", endedStep("ask1", buildstep.StepPassed, t0), Put) + askEnded(w, "ask1") }, map[string]string{}}, {"WM2 S1: a node offline past its bound is silent, and back online it is not", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { w.Connect("node.ace") diff --git a/modules/mesh-delivery-watchman/internal/watch/mirror.go b/modules/mesh-delivery-watchman/internal/watch/mirror.go index 96e5a228..c2206791 100644 --- a/modules/mesh-delivery-watchman/internal/watch/mirror.go +++ b/modules/mesh-delivery-watchman/internal/watch/mirror.go @@ -1,6 +1,7 @@ package watch import ( + "encoding/json" "strings" "time" @@ -9,58 +10,91 @@ import ( "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshwatchdogsclient" "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshwantsclient" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" "git.novox.be/novox/mesh-delivery/sdk/buildstep" ) -// 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 (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 -// (buildstep.AskStream), under the ask's id, and it begins when the node started the step. +// A build ask taken by a node is a move (novox/hq ADR 0525 §4, issues 1887 A8b, 1908, 1909; design 47 G §483, §496, §602, +// A-136): the holder of the build seat says `ask-taken` when it takes any ask, `step-progress` once a minute while it runs +// it, and `step-ended` when it ends it. The pulse of the move is the pulse of the bucket named as the seat's stream +// (buildstep.AskStream), under the ask's id, and it begins at the latest of the ask's taking and its progress, so a step +// that runs long but says its progress raises nothing, and one whose holder stops is stuck within the progress bound. // -// **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). +// **An ask still waiting in the stream is a backlog, not a stall** (issue 1908): 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). -// 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 := buildstep.AskStream(language) - if err != nil { - w.say("no pulse for the step %s: %v", key, err) +// ProgressBound is how long a running ask may say nothing of its progress (design 47 G §75: five minutes). +const ProgressBound = 5 * time.Minute + +// languageOfSeat is the build language whose seat is seat, and whether it is a build seat. +func languageOfSeat(seat string) (string, bool) { + for _, lang := range buildstep.AskLanguages { + if s, err := buildstep.BuildSeat(lang); err == nil && s == seat { + return lang, true + } + } + return "", false +} + +// AskTaken hears a build seat's `ask-taken` (node: the subject's last token): the ask is taken, so its clock begins at the +// event's time. +func (w *Watchman) AskTaken(seat, node string, body []byte) { + var e nodebuildgo.AskTakenEvent + if err := json.Unmarshal(body, &e); err != nil || e.Ask == "" { + w.say("an ask-taken of %s does not read, left: %v", seat, err) return } - _, node, id, err := buildstep.ParseStepKey(key) + w.askClock(seat, node, e.Ask, e.At) +} + +// StepProgress hears a build seat's `step-progress`: the running ask says it lives, so its clock begins again at the +// event's time. +func (w *Watchman) StepProgress(seat, node string, body []byte) { + var e nodebuildgo.StepProgressEvent + if err := json.Unmarshal(body, &e); err != nil || e.Ask == "" { + w.say("a step-progress of %s does not read, left: %v", seat, err) + return + } + w.askClock(seat, node, e.Ask, e.At) +} + +// AskEnded hears a build seat's `step-ended`, whatever its asker: the ask's clock ends. +func (w *Watchman) AskEnded(seat string, body []byte) { + lang, ok := languageOfSeat(seat) + if !ok { + return + } + bucket, err := buildstep.AskStream(lang) if err != nil { return } + var e nodebuildgo.StepEndedEvent + if err := json.Unmarshal(body, &e); err != nil || e.Ask == "" { + w.say("a step-ended of %s does not read, left: %v", seat, err) + return + } w.mu.Lock() defer w.mu.Unlock() - switch op { - case Expired: - return - case Delete: - w.moved(bucket, id) + w.moved(bucket, e.Ask) +} + +func (w *Watchman) askClock(seat, node, ask string, at time.Time) { + lang, ok := languageOfSeat(seat) + if !ok { return } - w.stands(bucket, id) - step, err := buildstep.DecodeStep(value) + bucket, err := buildstep.AskStream(lang) if err != nil { - w.say("the step %s cannot be read, and is left: %v", key, err) + w.say("no pulse for the ask %s: %v", ask, err) return } - if step.State != buildstep.StepRunning { - w.moved(bucket, id) - return + w.mu.Lock() + defer w.mu.Unlock() + if at.IsZero() { + at = w.now() } - since := step.Started - if since.IsZero() { - since = w.now() - } - w.mirrorWithin(w.stepBound, bucket, id, describe("the ask %s to end on %s", id, node), node, since, 0) + w.mirrorWithin(w.progressBound, bucket, ask, describe("the ask %s to end on %s", ask, node), node, at, 0) } // A want is a move too (novox/hq ADR 0539 §19, issue 1906): the assigner writes `want..` in its wants bucket, diff --git a/modules/mesh-delivery-watchman/internal/watch/names.go b/modules/mesh-delivery-watchman/internal/watch/names.go index e9bf2353..c081279c 100644 --- a/modules/mesh-delivery-watchman/internal/watch/names.go +++ b/modules/mesh-delivery-watchman/internal/watch/names.go @@ -38,6 +38,3 @@ const ( FindingManifestWaits = "manifest-waits" FindingNodeAccountFallback = "node-account-fallback" ) - -// StepsBucket is a build module's state steps (reads: the build seats' steps, `seat:node-build-.steps`, held by the module that holds the seat). -func StepsBucket(module string) string { return module + "_steps" } diff --git a/modules/mesh-delivery-watchman/internal/watch/watch.go b/modules/mesh-delivery-watchman/internal/watch/watch.go index c42d0043..78d77652 100644 --- a/modules/mesh-delivery-watchman/internal/watch/watch.go +++ b/modules/mesh-delivery-watchman/internal/watch/watch.go @@ -105,6 +105,8 @@ type Watchman struct { tunnels map[string]string // stepBound is how long a build step may run once a node took the ask; a test shortens it. stepBound time.Duration + // progressBound is how long a running ask may say nothing of its progress; a test shortens it. + progressBound time.Duration // standing is, until the start ends (Reconcile), each mirrored key read standing, by the bucket a stuck key names it // under; readWhole the buckets whose keys were all read at this start (ReadWhole). reconciled is the start's end. standing map[string]map[string]bool @@ -127,7 +129,7 @@ type Watchman struct { // New is a watchman writing to out, reading the time from now. func New(out Out, now func() time.Time) *Watchman { return &Watchman{out: out, now: now, counts: map[string]int{}, presence: map[string]presence{}, - pulses: map[string]*pulse{}, stuck: map[string]meshwatchdogsclient.Stuck{}, wants: map[string]string{}, haves: map[string]string{}, held: map[string]bool{}, starts: map[string]startMark{}, bound: meshwatchdogsclient.PulseMoveBound, stepBound: StepBound, + pulses: map[string]*pulse{}, stuck: map[string]meshwatchdogsclient.Stuck{}, wants: map[string]string{}, haves: map[string]string{}, held: map[string]bool{}, starts: map[string]startMark{}, bound: meshwatchdogsclient.PulseMoveBound, stepBound: StepBound, progressBound: ProgressBound, findings: map[string][]byte{}, seededFindings: map[string]bool{}, signals: map[string]signalDue{}, busCreated: map[string]bool{}, takeInHeld: map[string]Finding{}, away: map[string]bool{}, haveState: map[string]nodeengine.HaveByNodeModuleValue{}, unhealthy: map[string]int{}, advised: map[string]int{}, advisedSince: map[string]time.Time{}, control: map[string]bool{}, failedSince: map[string]time.Time{}, provFailed: map[string]map[string]ProvReport{}, closures: map[string]map[string]time.Time{}, waiting: map[string]map[string]string{}, @@ -330,7 +332,7 @@ func (w *Watchman) Reconcile() { // (ReadWhole) and whose key was not read standing is written moved, which clears its finding, and said. A bucket that // could not be read closes nothing: its keys are not known to be gone. -// ReadWhole says every key of bucket (as a stuck key names it: an ask stream for a build seat's steps, the assigner's +// ReadWhole says every key of bucket (as a stuck key names it: the assigner's // wants) was read at this start; called once its watch holds what stands. func (w *Watchman) ReadWhole(bucket string) { w.mu.Lock() @@ -453,14 +455,14 @@ const StepBound = 30 * time.Minute func (w *Watchman) SetStepBound(d time.Duration) { w.mu.Lock() defer w.mu.Unlock() - w.stepBound = d + w.stepBound, w.progressBound = d, d } // SetBound sets how long a move, and a running step, may take; it is meshwatchdogsclient.PulseMoveBound unless a test shortens it. func (w *Watchman) SetBound(d time.Duration) { w.mu.Lock() defer w.mu.Unlock() - w.bound, w.stepBound = d, d + w.bound, w.stepBound, w.progressBound = d, d, d } // ttl is the time to live of a pulse for a move of the given age, and whether the move is stalled already diff --git a/modules/mesh-delivery-watchman/internal/watch/watch_test.go b/modules/mesh-delivery-watchman/internal/watch/watch_test.go index 740d3834..20473e2e 100644 --- a/modules/mesh-delivery-watchman/internal/watch/watch_test.go +++ b/modules/mesh-delivery-watchman/internal/watch/watch_test.go @@ -13,7 +13,6 @@ import ( "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshwatchdogsclient" "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshwantsclient" - "git.novox.be/novox/mesh-delivery/sdk/buildstep" ) type fakeOut struct { @@ -115,22 +114,23 @@ const ( presenceNS = 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 := buildstep.EncodeStep(buildstep.Step{ID: id, Tool: "build_go_test", Node: "n1", State: buildstep.StepRunning, Started: started}) +// askBody is the body of a build seat's step-progress, ask-taken or step-ended for an ask at a time. +func askBody(id string, at time.Time) []byte { + b, err := json.Marshal(map[string]any{"ask": id, "asker": "a1", "at": at, "phase": "build"}) if err != nil { panic(err) } return b } -func endedStep(id, state string, started time.Time) []byte { - b, err := buildstep.EncodeStep(buildstep.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 +// askTaken, askProgress and askEnded are what the holder of the Go build seat says on node n1 (design 47 G §496, §602). +func askTaken(w *Watchman, id string, at time.Time) { + w.AskTaken("node-build-go", "n1", askBody(id, at)) } +func askProgress(w *Watchman, id string, at time.Time) { + w.StepProgress("node-build-go", "n1", askBody(id, at)) +} +func askEnded(w *Watchman, id string) { w.AskEnded("node-build-go", askBody(id, t0)) } func pulseKeyOf(t *testing.T, bucket, key string) string { t.Helper() @@ -271,7 +271,7 @@ func rebuildTakesThePresenceOfWhoIsGone(t *testing.T) { // 1135/AC2 func TestRebuildPulsesFromCurrentKeys(t *testing.T) { // 1135/AC2 w, o, c := setup() c.t = t0.Add(30 * time.Second) - w.Step("go", "n1.askid1", runningStep("askid1", t0.Add(10*time.Second)), Put) + askProgress(w, "askid1", t0.Add(10*time.Second)) w.SeedPresence("ace", false, t0) w.Reconcile() if err := w.Rebuild(func([]byte) ([]byte, error) { return connzAnswer(), nil }); err != nil { @@ -290,7 +290,7 @@ func TestReconcileDropsPulsesNothingMirrorsAnyMore(t *testing.T) { // 1135/AC2 live := pulseKeyOf(t, checksKey, "askid1") w.SeedPulse(stale) w.SeedPulse(live) - w.Step("go", "n1.askid1", runningStep("askid1", t0), Put) + askProgress(w, "askid1", t0) w.Reconcile() if !slices.Equal(o.pulseDeletes, []string{stale}) { t.Fatalf("deleted = %v, want only %s", o.pulseDeletes, stale) @@ -303,7 +303,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.Step("go", "n1.askid1", runningStep("askid1", t0), Put) // the key found at start, last renewed 30s ago + askProgress(w, "askid1", t0) // 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]) @@ -321,7 +321,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.Step("go", "n1.askid1", runningStep("askid1", t0), Put) + askProgress(w, "askid1", t0) if len(o.pulses) != 0 { t.Fatalf("a pulse was written for a heartbeat 120s old: %v", o.pulses) } @@ -337,7 +337,7 @@ 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.Step("go", "n1.askid1", runningStep("askid1", t0), Put) + askProgress(w, "askid1", t0) w.SeedPresence("ace", false, t0) w.SeedPresence("part.mesh-delivery-view.ace", false, t0) w.Reconcile() @@ -357,13 +357,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.Step("go", "n1.askid1", runningStep("askid1", t0), Put) + askProgress(w, "askid1", t0) 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.Step("go", "n1.askid1", runningStep("askid1", c.t), Put) // the builder re-puts the key: same check, started long ago + askProgress(w, "askid1", c.t) // 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]) } @@ -372,7 +372,7 @@ func TestEveryHeartbeatRenewsThePulse(t *testing.T) { // 1135/AC3 } // The same revision heard twice changes nothing. o.ttls[k] = 0 - w.Step("go", "n1.askid1", runningStep("askid1", c.t), Put) + askProgress(w, "askid1", c.t) if o.ttls[k] != 0 { t.Fatal("one revision rewrote the pulse twice") } @@ -381,10 +381,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.Step("go", "n1.askid1", runningStep("askid1", t0), Put) + askProgress(w, "askid1", t0) w.Connect("node.ace") w.Disconnect("node.ace") - w.Step("go", "n1.askid1", nil, Delete) // the check ended + askEnded(w, "askid1") // the check ended w.Connect("node.ace") if len(o.pulses) != 0 { t.Fatalf("pulses left: %v", o.pulses) @@ -399,9 +399,9 @@ func TestMirroredMovesDeleteThePulse(t *testing.T) { // 1135/AC4 func TestACheckWhoseBuilderStopsIsStuckFromItsLastHeartbeat(t *testing.T) { // 1135/AC4 w, o, c := setup() - w.Step("go", "n1.askid1", runningStep("askid1", t0), Put) + askProgress(w, "askid1", t0) c.t = t0.Add(50 * time.Second) - w.Step("go", "n1.askid1", runningStep("askid1", c.t), Put) // the last heartbeat + askProgress(w, "askid1", c.t) // 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")] @@ -412,7 +412,7 @@ func TestACheckWhoseBuilderStopsIsStuckFromItsLastHeartbeat(t *testing.T) { // 1 func TestAStuckKeyIsWrittenMovedNeverDeleted(t *testing.T) { // 1135/AC4 w, o, c := setup() - w.Step("go", "n1.askid1", runningStep("askid1", t0), Put) + askProgress(w, "askid1", t0) pk := pulseKeyOf(t, checksKey, "askid1") c.t = t0.Add(120 * time.Second) w.Pulse(pk, Expired, c.t) @@ -421,13 +421,13 @@ func TestAStuckKeyIsWrittenMovedNeverDeleted(t *testing.T) { // 1135/AC4 t.Fatal("stuck written as moved at expiry") } c.t = t0.Add(300 * time.Second) - w.Step("go", "n1.askid1", nil, Delete) + askEnded(w, "askid1") s := o.stuck[sk] 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.Step("go", "n1.askid1", nil, Delete) + askEnded(w, "askid1") if o.stuckWrites != n { t.Fatal("a stuck key already moved was written again") } @@ -435,10 +435,10 @@ func TestAStuckKeyIsWrittenMovedNeverDeleted(t *testing.T) { // 1135/AC4 func TestAnAskEndedAfterStuckMovesIt(t *testing.T) { // 1135/AC4 w, o, c := setup() - w.Step("go", "n1.askid1", runningStep("askid1", t0), Put) + askProgress(w, "askid1", t0) c.t = t0.Add(120 * time.Second) w.Pulse(pulseKeyOf(t, checksKey, "askid1"), Expired, c.t) - w.Step("go", "n1.askid1", nil, Delete) // the ask ended at last + askEnded(w, "askid1") // 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") } @@ -467,7 +467,7 @@ func TestOnlyTheExpiryOfItsOwnPulseWritesStuck(t *testing.T) { // 1135/AC5 func expiryMarkerWritesStuck(t *testing.T) { // 1135/AC5 w, o, c := setup() - w.Step("go", "n1.askid1", runningStep("askid1", t0), Put) + askProgress(w, "askid1", t0) c.t = t0.Add(120 * time.Second) w.Pulse(pulseKeyOf(t, checksKey, "askid1"), Expired, c.t) s, ok := o.stuck[stuckKeyOf(t, checksKey, "askid1")] @@ -481,9 +481,9 @@ func expiryMarkerWritesStuck(t *testing.T) { // 1135/AC5 func ownDeleteOfAPulseWritesNothing(t *testing.T) { // 1135/AC5 w, o, c := setup() - w.Step("go", "n1.askid1", runningStep("askid1", t0), Put) + askProgress(w, "askid1", t0) pk := pulseKeyOf(t, checksKey, "askid1") - w.Step("go", "n1.askid1", nil, Delete) + askEnded(w, "askid1") 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) @@ -496,12 +496,12 @@ func TestTheBoundCanBeShortenedForATest(t *testing.T) { // 1135/AC8 w, o, c := setup() w.SetBound(2 * time.Second) c.t = t0.Add(500 * time.Millisecond) - w.Step("go", "n1.askid1", runningStep("askid1", t0), Put) + askProgress(w, "askid1", t0) 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.Step("go", "n1.askid2", runningStep("askid2", t0), Put) + askProgress(w, "askid2", t0) 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) } @@ -510,27 +510,25 @@ func TestTheBoundCanBeShortenedForATest(t *testing.T) { // 1135/AC8 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) + askProgress(w, "askid1", t0) if len(o.pulses) != 1 { t.Fatalf("pulses = %v, want the taken ask's", o.pulses) } - for _, state := range []string{buildstep.StepPassed, buildstep.StepFailed, buildstep.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 + w2, o2, c2 := setup() + c2.t = t0.Add(10 * time.Second) + askTaken(w2, "askid1", t0) + askEnded(w2, "askid1") + if len(o2.pulses) != 0 && o2.pulses[pulseKeyOf(t, checksKey, "askid1")].Key != "" { + t.Errorf("the pulse of an ended ask stands: %v", 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 + askProgress(w, "askid1", t0) // 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) @@ -540,35 +538,54 @@ func TestARunningAskFoundAtStartGetsItsPulse(t *testing.T) { // 1909 } } -func TestAPausedKeyOrAnotherNodesNonStepIsLeft(t *testing.T) { // 1908 +func TestAnEventThatDoesNotReadOrIsNoBuildSeatsIsLeft(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) + w.StepProgress("node-build-go", "n1", []byte(`{"v":99}`)) + w.StepProgress("node-build-go", "n1", askBody("", t0)) + w.AskTaken("node-build-go", "n1", []byte(`nope`)) + w.AskTaken("mesh-wants", "n1", askBody("askid1", t0)) if len(o.pulses) != 0 || len(o.stuck) != 0 { t.Errorf("pulses %v stuck %v", o.pulses, o.stuck) } } -func TestARunningStepOfTenMinutesUnderItsBoundIsNotStuck(t *testing.T) { // 1908 +func TestARunningStepThatSaysItsProgressIsNeverStuckAndOneWhoseHolderStopsIs(t *testing.T) { // design 47 G §496, §602 o, c := newOut(), &clock{t: t0} - w := New(o, c.now) // the step's own bound: 30 minutes - w.Step("go", "n1.askid1", runningStep("askid1", t0), Put) - c.t = t0.Add(10 * time.Minute) - // no heartbeat is written, and the server expires the pulse at its ttl: 30 minutes from the start, not 120 s - if len(o.stuck) != 0 { - t.Fatalf("a build of 10 minutes is stuck: %v", o.stuck) + w := New(o, c.now) + askTaken(w, "askid1", t0) + k := pulseKeyOf(t, checksKey, "askid1") + if got := o.ttls[k]; got != ProgressBound { + t.Errorf("ttl = %v, want the progress bound %v", got, ProgressBound) } - if got := o.ttls[pulseKeyOf(t, checksKey, "askid1")]; got != 30*time.Minute { - t.Errorf("ttl = %v, want the step's bound of 30m from its start", got) + // a 20-minute step: its progress each minute renews the clock + for m := 1; m <= 20; m++ { + c.t = t0.Add(time.Duration(m) * time.Minute) + askProgress(w, "askid1", c.t) + if o.ttls[k] != ProgressBound { + t.Fatalf("minute %d: ttl = %v, want the full bound again", m, o.ttls[k]) + } + } + if len(o.stuck) != 0 { + t.Fatalf("a step saying its progress is stuck: %v", o.stuck) + } + // its holder stops: the pulse expires, and the ask is stuck from its last progress + c.t = t0.Add(26 * time.Minute) + w.Pulse(k, Expired, c.t) + s, ok := o.stuck[stuckKeyOf(t, checksKey, "askid1")] + if !ok || s.Moved || !s.Since.Equal(t0.Add(20*time.Minute)) { + t.Fatalf("stuck = %+v, %v", s, ok) + } + askEnded(w, "askid1") + if !o.stuck[stuckKeyOf(t, checksKey, "askid1")].Moved { + t.Fatal("the end of the ask left it stuck") } } func TestARunningStepPastItsBoundIsStuck(t *testing.T) { // 1908 o, c := newOut(), &clock{t: t0} w := New(o, c.now) - w.Step("go", "n1.askid1", runningStep("askid1", t0), Put) - c.t = t0.Add(31 * time.Minute) + askProgress(w, "askid1", t0) + c.t = t0.Add(6 * time.Minute) w.Pulse(pulseKeyOf(t, checksKey, "askid1"), Expired, c.t) s, ok := o.stuck[stuckKeyOf(t, checksKey, "askid1")] if !ok || s.Moved || !s.Since.Equal(t0) { @@ -576,7 +593,7 @@ func TestARunningStepPastItsBoundIsStuck(t *testing.T) { // 1908 } // and one found past its bound at the watchman's start is stuck at once o2, c2 := newOut(), &clock{t: t0.Add(time.Hour)} - New(o2, c2.now).Step("go", "n1.askid1", runningStep("askid1", t0), Put) + askProgress(New(o2, c2.now), "askid1", t0) if _, ok := o2.stuck[stuckKeyOf(t, checksKey, "askid1")]; !ok || len(o2.pulses) != 0 { t.Errorf("stuck %v pulses %v", o2.stuck, o2.pulses) } @@ -683,9 +700,6 @@ func TestAStuckKeyWhoseKeyIsGoneIsClosedAtTheStart(t *testing.T) { closed bool late bool // read after the start's end (Awaits, ReadLate) }{ - {"an ask's step aged out of the steps bucket", checksKey, "askold1", func(*Watchman) {}, true, true, false}, - {"an ask's step still running", checksKey, "askold1", func(w *Watchman) { w.Step("go", "n1.askold1", runningStep("askold1", t0), Put) }, true, false, false}, - {"the steps bucket could not be read", checksKey, "askold1", func(*Watchman) {}, false, false, false}, {"a want deleted while nobody heard", AssignerWantsBucket, wantKey, func(*Watchman) {}, true, true, false}, {"a want still standing", AssignerWantsBucket, wantKey, func(w *Watchman) { w.Want(wantKey, wantValue(t, fpNew, false), Put, t0) }, true, false, false}, // DLV-M5 (ADR 0615): the wants are read from the assigner's first view, which may come after the start's end; what diff --git a/modules/mesh-delivery-watchman/manifest_test.go b/modules/mesh-delivery-watchman/manifest_test.go index 6fb90649..ce1c6227 100644 --- a/modules/mesh-delivery-watchman/manifest_test.go +++ b/modules/mesh-delivery-watchman/manifest_test.go @@ -13,8 +13,6 @@ import ( "testing" "mesh-delivery-watchman/internal/watch" - - "git.novox.be/novox/mesh-sdk/go/seatlist" ) // The controller's manifest reader is mesh-controller internal/catalogue (ParseManifest), an internal package of another @@ -104,9 +102,8 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6 } reads := slices.Clone(m.Reads) slices.Sort(reads) - // The build modules are no names (design 47 G A-82): the steps are the states of the holders of the build seats. - if !reflect.DeepEqual(reads, []string{"mesh-delivery-builder.builds", "mesh-host.health", "mesh-provisioner.provisions", "node-engine.have", - "seat:node-build-go.steps", "seat:node-build-image.steps", "seat:node-build-node.steps"}) { + // The build seats' steps are no read (design 47 G §483, A-136): the watchman times an ask from the seats' step-progress. + if !reflect.DeepEqual(reads, []string{"mesh-delivery-builder.builds", "mesh-host.health", "mesh-provisioner.provisions", "node-engine.have"}) { t.Errorf("reads = %v", reads) } // Design 47 G §368: the version is MAJOR.MINOR and nothing else. @@ -129,11 +126,11 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6 } // The join: its first view asked again by hand is a hand act, recorded through the seat (views.StatusTools). The waits the // indexer holds at the watchman's start are asked of its tool (ADR 0633 §5), a verb edge as the view's. - if want := []string{"mesh-delivery-assigner.assigner_read_wants", "mesh-delivery-indexer.indexer_pending", "mesh-delivery-indexer.indexer_read_modules", "seat:mesh-hand-acts.hand-act", "seat:mesh-module-assigner.assignments"}; !reflect.DeepEqual(m.Invokes, want) { + if want := []string{"mesh-delivery-assigner.assigner_read_wants", "mesh-delivery-indexer.indexer_pending", "mesh-delivery-indexer.indexer_read_modules", "seat:mesh-hand-acts.hand-act"}; !reflect.DeepEqual(m.Invokes, want) { t.Errorf("invokes = %v, want %v", m.Invokes, want) } // DLV-M5 its first views, 1585/T20 the indexer's take-in events, and DLV-W1 (ADR 0595 §4): the accepted verdicts and the bus module's created and refused, durably. - if want := []string{"mesh-delivery-assigner.views.wants", "mesh-delivery-indexer.take-in.pending", "mesh-delivery-indexer.take-in.taken", "mesh-delivery-indexer.views.modules", "mesh-catalog.manifest-accepted", "mesh-delivery.source-moved", "mesh-delivery.filling", "mesh-delivery.full", "mesh-delivery.room", "mesh-delivery.record-opened", "mesh-delivery.record-ended", "mesh-catalog.filling", "mesh-catalog.full", "mesh-catalog.room", "mesh-wants.filling", "mesh-wants.full", "mesh-wants.room", "mesh-grants.filling", "mesh-grants.full", "mesh-grants.room", "mesh-pipeline-runner.pipeline-started", "mesh-pipeline-runner.step-started", "mesh-pipeline-runner.step-ended", "mesh-pipeline-runner.wait-held", "mesh-pipeline-runner.wait-ended", "mesh-wants.wait-held", "mesh-wants.wait-ended", "mesh-grants.wait-held", "mesh-grants.wait-ended", "mesh-broker.created", "mesh-broker.refused", "mesh-conditions.findings-asked"}; !reflect.DeepEqual(m.Consumes, want) { + if want := []string{"mesh-delivery-assigner.views.wants", "mesh-delivery-indexer.take-in.pending", "mesh-delivery-indexer.take-in.taken", "mesh-delivery-indexer.views.modules", "mesh-catalog.manifest-accepted", "mesh-delivery.source-moved", "mesh-delivery.filling", "mesh-delivery.full", "mesh-delivery.room", "mesh-delivery.record-opened", "mesh-delivery.record-ended", "mesh-catalog.filling", "mesh-catalog.full", "mesh-catalog.room", "mesh-wants.filling", "mesh-wants.full", "mesh-wants.room", "mesh-grants.filling", "mesh-grants.full", "mesh-grants.room", "node-build-go.ask-taken", "node-build-go.step-progress", "node-build-go.step-ended", "node-build-image.ask-taken", "node-build-image.step-progress", "node-build-image.step-ended", "node-build-node.ask-taken", "node-build-node.step-progress", "node-build-node.step-ended", "mesh-pipeline-runner.pipeline-started", "mesh-pipeline-runner.step-started", "mesh-pipeline-runner.step-ended", "mesh-pipeline-runner.wait-held", "mesh-pipeline-runner.wait-ended", "mesh-wants.wait-held", "mesh-wants.wait-ended", "mesh-grants.wait-held", "mesh-grants.wait-ended", "mesh-broker.created", "mesh-broker.refused", "mesh-conditions.findings-asked"}; !reflect.DeepEqual(m.Consumes, want) { t.Errorf("consumes = %v, want %v", m.Consumes, want) } // 1952/WM1: the watchman holds the seat mesh-watchdogs at mesh scope, its first seat (novox/hq ADR 0564 §1, §4). @@ -171,16 +168,13 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6 return e } for name, row := range map[string]struct{ got, want string }{ - "module": {watch.WatchmanModule, m.Module}, - "presence": {watch.PresenceBucket, m.Module + "_" + m.State[0].Name}, - "pulse": {watch.PulseBucket, m.Module + "_" + m.State[1].Name}, - "stuck": {watch.StuckBucket, m.Module + "_" + m.State[2].Name}, - "have": {watch.HaveBucket, bucket(m.Reads, "node-engine.have")}, - "health": {watch.NodeHealthBucket, bucket(m.Reads, "mesh-host.health")}, - "builds": {watch.BuilderBuildsBucket, bucket(m.Reads, "mesh-delivery-builder.builds")}, - // The build modules are no names (design 47 G A-82): the steps are read as the states of the holders of the build seats. - "steps": {"seat:node-build-go.steps", entry(m.Reads, "seat:node-build-go.steps")}, - "holders": {"seat:" + seatlist.HolderSeat + "." + seatlist.HolderVerb, entry(m.Invokes, "seat:"+seatlist.HolderSeat+"."+seatlist.HolderVerb)}, + "module": {watch.WatchmanModule, m.Module}, + "presence": {watch.PresenceBucket, m.Module + "_" + m.State[0].Name}, + "pulse": {watch.PulseBucket, m.Module + "_" + m.State[1].Name}, + "stuck": {watch.StuckBucket, m.Module + "_" + m.State[2].Name}, + "have": {watch.HaveBucket, bucket(m.Reads, "node-engine.have")}, + "health": {watch.NodeHealthBucket, bucket(m.Reads, "mesh-host.health")}, + "builds": {watch.BuilderBuildsBucket, bucket(m.Reads, "mesh-delivery-builder.builds")}, "wants": {watch.AssignerWantsBucket, bucket(m.Consumes, "mesh-delivery-assigner.views.wants")}, "provisions": {watch.ProvisionsBucket, bucket(m.Reads, "mesh-provisioner.provisions")}, "modules": {watch.ModulesBucket, bucket(m.Consumes, "mesh-delivery-indexer.views.modules")}, diff --git a/modules/mesh-delivery-watchman/module.json b/modules/mesh-delivery-watchman/module.json index 48d7c590..a636cee8 100644 --- a/modules/mesh-delivery-watchman/module.json +++ b/modules/mesh-delivery-watchman/module.json @@ -19,17 +19,13 @@ "mesh-delivery-builder.builds", "mesh-host.health", "mesh-provisioner.provisions", - "node-engine.have", - "seat:node-build-go.steps", - "seat:node-build-image.steps", - "seat:node-build-node.steps" + "node-engine.have" ], "invokes": [ "mesh-delivery-assigner.assigner_read_wants", "mesh-delivery-indexer.indexer_pending", "mesh-delivery-indexer.indexer_read_modules", - "seat:mesh-hand-acts.hand-act", - "seat:mesh-module-assigner.assignments" + "seat:mesh-hand-acts.hand-act" ], "consumes": [ "mesh-delivery-assigner.views.wants", @@ -52,6 +48,15 @@ "mesh-grants.filling", "mesh-grants.full", "mesh-grants.room", + "node-build-go.ask-taken", + "node-build-go.step-progress", + "node-build-go.step-ended", + "node-build-image.ask-taken", + "node-build-image.step-progress", + "node-build-image.step-ended", + "node-build-node.ask-taken", + "node-build-node.step-progress", + "node-build-node.step-ended", "mesh-pipeline-runner.pipeline-started", "mesh-pipeline-runner.step-started", "mesh-pipeline-runner.step-ended",