From e8d1bf96be5620648eb989a7d87d187ae85cf3a3 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 19:09:46 +0100 Subject: [PATCH] Names N3 (watchman): the watchman names its own buckets from its credential's module Its states presence, pulse and stuck and its first views were named from the constant WatchmanModule; they are now `_` (watch.BucketsOf), and the watchman's stuck keys of presence take its own presence bucket's name. Builds: design 47 G A-296 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../integration_test.go | 53 +++++++++++-------- .../cmd/mesh-delivery-watchman/main.go | 13 ++--- .../cmd/mesh-delivery-watchman/main_test.go | 10 ++-- .../internal/watch/names.go | 14 +++-- .../internal/watch/presence.go | 4 +- .../internal/watch/signals.go | 2 +- .../internal/watch/watch.go | 7 +-- .../internal/watch/watch_test.go | 13 ++--- .../mesh-delivery-watchman/manifest_test.go | 9 ++-- 9 files changed, 68 insertions(+), 57 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 39028cea..f63192ef 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 @@ -24,6 +24,7 @@ import ( "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep" "git.novox.be/novox/mesh-sdk/go/buscompose" "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/views" "git.novox.be/novox/mesh-sdk/go/views/viewstest" "mesh-delivery-watchman/internal/deadletters" @@ -92,6 +93,12 @@ func sayAsk(t *testing.T, ctx context.Context, js jetstream.JetStream, event, id } } +// own and firstViewsBucket are the beds' watchman's own buckets, named from the module its credential names (A-296). +var ( + own = watch.BucketsOf(watch.WatchmanModule) + firstViewsBucket = watch.WatchmanModule + "_" + views.StatusState +) + // findingEventsSubject is the subject the stand-in seat mesh-broker's show answers for the watchman's finding-events. const findingEventsSubject = "mesh.bus." + watch.WatchmanModule + ".finding-events" @@ -161,9 +168,9 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8 defer cancel() kvs := map[string]jetstream.KeyValue{} for _, c := range []jetstream.KeyValueConfig{ - {Bucket: watch.PresenceBucket, History: 20}, - {Bucket: watch.StuckBucket, History: 20}, - {Bucket: watch.PulseBucket, LimitMarkerTTL: time.Second}, + {Bucket: own.Presence, History: 20}, + {Bucket: own.Stuck, History: 20}, + {Bucket: own.Pulse, LimitMarkerTTL: time.Second}, {Bucket: watch.AssignerWantsBucket, History: 20}, {Bucket: watch.FindingsBucket, History: 5}, {Bucket: firstViewsBucket, History: 1}, @@ -224,7 +231,7 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8 return e, err == nil } online := func(key string) (bool, bool) { - e, ok := has(watch.PresenceBucket, key) + e, ok := has(own.Presence, key) if !ok { return false, false } @@ -267,10 +274,10 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8 askStream, _ := buildstep.AskStream(buildstep.AskGo) const ask1, ask2 = "m0000001aaaaaa", "m0000002bbbbbb" 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 }) + until("a pulse for the ask a node took", func() bool { _, ok := has(own.Pulse, pulseOf(askStream, ask1)); return ok }) var stuck meshwatchdogsclient.Stuck until("a stuck key for the ask no node ended", func() bool { - e, ok := has(watch.StuckBucket, stuckOf(askStream, ask1)) + e, ok := has(own.Stuck, stuckOf(askStream, ask1)) if !ok { return false } @@ -283,7 +290,7 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8 } // The node n1, gone offline and never back, is stuck as well. until("a stuck key for node n1 that did not come back", func() bool { - _, ok := has(watch.StuckBucket, stuckOf(watch.PresenceBucket, "n1")) + _, ok := has(own.Stuck, stuckOf(own.Presence, "n1")) return ok }) @@ -291,18 +298,18 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8 n2 := connect("node.n2") until("node n2 online", func() bool { on, ok := online("n2"); return ok && on }) n2.Close() - until("a pulse for node n2 gone offline", func() bool { _, ok := has(watch.PulseBucket, pulseOf(watch.PresenceBucket, "n2")); return ok }) + until("a pulse for node n2 gone offline", func() bool { _, ok := has(own.Pulse, pulseOf(own.Presence, "n2")); return ok }) connect("node.n2") 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 }) + until("the pulse of node n2 deleted", func() bool { _, ok := has(own.Pulse, pulseOf(own.Presence, "n2")); return !ok }) // An ask a node takes at once is the same: no stuck key, however long the bound passes. 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 }) + until("a pulse for the second ask", func() bool { _, ok := has(own.Pulse, pulseOf(askStream, ask2)); return ok }) 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 }) + until("the pulse of the second ask deleted", func() bool { _, ok := has(own.Pulse, 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)} { - if _, ok := has(watch.StuckBucket, k); ok { + for _, k := range []string{stuckOf(own.Presence, "n2"), stuckOf(askStream, ask2)} { + if _, ok := has(own.Stuck, k); ok { t.Errorf("a stuck key %s was written for a move that happened before the bound", k) } } @@ -409,9 +416,9 @@ func anAskRunningBeforeTheStartGetsItsPulse(t *testing.T) { // 1909 defer cancel() kvs := map[string]jetstream.KeyValue{} for _, c := range []jetstream.KeyValueConfig{ - {Bucket: watch.PresenceBucket, History: 20}, - {Bucket: watch.StuckBucket, History: 20}, - {Bucket: watch.PulseBucket, LimitMarkerTTL: time.Second}, + {Bucket: own.Presence, History: 20}, + {Bucket: own.Stuck, History: 20}, + {Bucket: own.Pulse, LimitMarkerTTL: time.Second}, {Bucket: watch.AssignerWantsBucket, History: 20}, {Bucket: watch.FindingsBucket, History: 5}, {Bucket: firstViewsBucket, History: 1}, @@ -443,7 +450,7 @@ func anAskRunningBeforeTheStartGetsItsPulse(t *testing.T) { // 1909 t.Fatal(err) } for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { - if _, err := kvs[watch.PulseBucket].Get(ctx, pk); err == nil { + if _, err := kvs[own.Pulse].Get(ctx, pk); err == nil { return } } @@ -490,9 +497,9 @@ func aWantIsMirroredAsAPulseAndTheHaveNamingItEndsIt(t *testing.T) { defer cancel() kvs := map[string]jetstream.KeyValue{} for _, c := range []jetstream.KeyValueConfig{ - {Bucket: watch.PresenceBucket, History: 20}, - {Bucket: watch.StuckBucket, History: 20}, - {Bucket: watch.PulseBucket, LimitMarkerTTL: time.Second}, + {Bucket: own.Presence, History: 20}, + {Bucket: own.Stuck, History: 20}, + {Bucket: own.Pulse, LimitMarkerTTL: time.Second}, {Bucket: watch.AssignerWantsBucket, History: 20}, {Bucket: watch.FindingsBucket, History: 5}, {Bucket: firstViewsBucket, History: 1}, @@ -556,11 +563,11 @@ func aWantIsMirroredAsAPulseAndTheHaveNamingItEndsIt(t *testing.T) { } t.Fatalf("never saw: %s", what) } - has := func(key string) bool { _, err := kvs[watch.PulseBucket].Get(ctx, key); return err == nil } + has := func(key string) bool { _, err := kvs[own.Pulse].Get(ctx, key); return err == nil } until("a pulse for the want that stood before the start", func() bool { return has(pulse("n2", "web")) }) putWant("n2", "mail", fpNew) until("a pulse for a want written after the start", func() bool { return has(pulse("n2", "mail")) }) - e, err := kvs[watch.PulseBucket].Get(ctx, pulse("n2", "web")) + e, err := kvs[own.Pulse].Get(ctx, pulse("n2", "web")) if err != nil { t.Fatal(err) } @@ -575,7 +582,7 @@ func aWantIsMirroredAsAPulseAndTheHaveNamingItEndsIt(t *testing.T) { for _, m := range []string{"web"} { wk, _ := meshwantsclient.WantKey("n2", m) sk, _ := meshwatchdogsclient.StuckKey(watch.AssignerWantsBucket, wk) - if _, err := kvs[watch.StuckBucket].Get(ctx, sk); err == nil { + if _, err := kvs[own.Stuck].Get(ctx, sk); err == nil { t.Errorf("a stuck key was written for the want of %s that was applied", m) } } 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 dbd28216..f8fbeb4c 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go @@ -189,11 +189,6 @@ func createdOr(e jetstream.KeyValueEntry, at time.Time) time.Time { return at } -// The watchman's own states of its first views (views.StatusState); its findings of a first view unanswered past its -// bound go to its findings bucket with the others (ADR 0615 §6). The owners' states it follows are the assigner's wants -// and the indexer's modules (DLV-M5). -var firstViewsBucket = watch.WatchmanModule + "_" + views.StatusState - const ( wantsState = "wants" // The owners' views families, as their manifests' `streams` name them. @@ -222,11 +217,13 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // Its own presence, pulses and stuck rows are what it keeps: each stops the start, naming its grant. The assigner's wants // and the node-engine's haves only make the pulses of a move: without the haves the watchman watches presence and the // build seats' steps, and says that moves are not watched (issue 1945). + // They are named from its credential's module (design 47 G A-296). + own := watch.BucketsOf(cred.Module) var kvs struct{ presence, pulse, stuck jetstream.KeyValue } for _, b := range []struct { name string into *jetstream.KeyValue - }{{watch.PresenceBucket, &kvs.presence}, {watch.PulseBucket, &kvs.pulse}, {watch.StuckBucket, &kvs.stuck}} { + }{{own.Presence, &kvs.presence}, {own.Pulse, &kvs.pulse}, {own.Stuck, &kvs.stuck}} { if *b.into, err = process.EssentialBucket(ctx, js, "watchman", b.name, "say who is online or what stopped moving"); err != nil { return err } @@ -265,7 +262,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return err } - w := watch.New(&kvOut{ctx: ctx, presence: kvs.presence, pulse: kvs.pulse, stuk: kvs.stuck, findings: held, js: js, cred: cred}, time.Now) + w := watch.New(cred.Module, &kvOut{ctx: ctx, presence: kvs.presence, pulse: kvs.pulse, stuk: kvs.stuck, findings: held, js: js, cred: cred}, time.Now) w.Logf(logger.Printf) // What the wants and the modules hold is ended by their own late read (watch.ReadLate), never by the start's end. w.Awaits(watch.AssignerWantsBucket) @@ -416,7 +413,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro errs := make(chan error, 16) // The first views of the wants and the modules, asked now and read each on its own goroutine once answered (ADR 0615). - reader.Start(ctx, js, cred, process.AwaitBucket(ctx, js, conn, firstViewsBucket, "watchman_first_views reads nothing and watchman_first_view_again is not heard", logger.Printf)) + reader.Start(ctx, js, cred, process.AwaitBucket(ctx, js, conn, cred.Module+"_"+views.StatusState, "watchman_first_views reads nothing and watchman_first_view_again is not heard", logger.Printf)) // Every part has registered: the hub pulls the module's durable, and the watchman's other consumers open. hearConsumed(ctx, js, cred, hub, w, binding, logger.Printf) defer hub.Stop() diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go index 7d24b481..d6f6358a 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go @@ -54,7 +54,7 @@ func TestRunNeedsItsCredential(t *testing.T) { // 1135/AC6 }) t.Run("DLV-W1: the consumed events reach the watchman by their subjects", func(t *testing.T) { out := &findingsOut{findings: map[string]bool{}} - w := watch.New(out, time.Now) + w := watch.New(watch.WatchmanModule, out, time.Now) say := func(string, ...any) {} // The bus module's own stream is its history, never heard here: only the seat's subject (ruling 13:0x). consumed(w, "mesh.stream.nats.event.refused", []byte(`{"module":"m","local":"l","why":"no room"}`), say) @@ -77,7 +77,7 @@ func TestRunNeedsItsCredential(t *testing.T) { // 1135/AC6 }) t.Run("design 47 G §500: the indexer's manifest-refused reaches the watchman by its subject", func(t *testing.T) { out := &findingsOut{findings: map[string]bool{}} - w := watch.New(out, time.Now) + w := watch.New(watch.WatchmanModule, out, time.Now) consumed(w, "mesh.seat.mesh-catalog.event.manifest-refused", []byte(`{"module":"m","digest":"d","problems":["x"]}`), func(string, ...any) {}) if !out.findings[watch.FindingKey(watch.ScopeModule, "m", watch.KindManifestRefused)] { t.Fatalf("a refused manifest reached %v", out.findings) @@ -85,7 +85,7 @@ func TestRunNeedsItsCredential(t *testing.T) { // 1135/AC6 }) t.Run("design 47 G §499, §541: a node-engine's wait-held and wait-ended reach the watchman by their subjects", func(t *testing.T) { out := &findingsOut{findings: map[string]bool{}} - w := watch.New(out, time.Now) + w := watch.New(watch.WatchmanModule, out, time.Now) say := func(string, ...any) {} body := fmt.Sprintf(`{"step":"compose","reason":"entry-missing","since":%q}`, time.Now().Add(-2*time.Hour).UTC().Format(time.RFC3339)) consumed(w, "mesh.seat.node-engine.event.wait-held.ace.web", []byte(body), say) @@ -99,7 +99,7 @@ func TestRunNeedsItsCredential(t *testing.T) { // 1135/AC6 }) t.Run("design 47 G A-192: a module's wait-held on its stream, naming a part, reaches the watchman", func(t *testing.T) { out := &findingsOut{findings: map[string]bool{}} - w := watch.New(out, time.Now) + w := watch.New(watch.WatchmanModule, out, time.Now) body := fmt.Sprintf(`{"step":"building","part":"mesh-sdk@v0.1.14","module":"web","fingerprint":"src2:f","bound":"1h0m0s","since":%q}`, time.Now().Add(-2*time.Hour).UTC().Format(time.RFC3339)) consumed(w, "mesh.stream.mesh-delivery-builder.event.wait-held.web.src2:f", []byte(body), func(string, ...any) {}) if len(out.findings) != 1 { @@ -120,7 +120,7 @@ func TestRunNeedsItsCredential(t *testing.T) { // 1135/AC6 "unreadable": func(string, any) (json.RawMessage, error) { return json.RawMessage(`{"waits":`), nil }, } { out := &findingsOut{findings: map[string]bool{}} - w := watch.New(out, time.Now) + w := watch.New(watch.WatchmanModule, out, time.Now) const key = "m.00000000000000000042" old := watch.Finding{Scope: watch.ScopeDelivery, ID: key, Kind: watch.KindTakeInWaits, Severity: watch.Warning, Source: "take-in", Since: time.Now().Add(-3 * time.Hour), Summary: "held", Said: "held", Headline: "held", Explanation: "held", Needs: "a take-in"} diff --git a/modules/mesh-delivery-watchman/internal/watch/names.go b/modules/mesh-delivery-watchman/internal/watch/names.go index 67060d94..c891be91 100644 --- a/modules/mesh-delivery-watchman/internal/watch/names.go +++ b/modules/mesh-delivery-watchman/internal/watch/names.go @@ -15,15 +15,19 @@ const ( // IndexerModule is the holder of the seat mesh-catalog, named from this module's manifest (consumes and invokes of mesh-delivery-indexer). IndexerModule = "mesh-delivery-indexer" - // The watchman's own states of the seat mesh-watchdogs. - PresenceBucket = WatchmanModule + "_" + meshwatchdogs.StatePresence - PulseBucket = WatchmanModule + "_" + meshwatchdogs.StatePulse - StuckBucket = WatchmanModule + "_" + meshwatchdogs.StateStuck - // AssignerWantsBucket is the assigner's state wants, followed by its views family (consumes: mesh-delivery-assigner.views.wants). AssignerWantsBucket = AssignerModule + "_wants" ) +// OwnBuckets are the watchman's own states of the seat mesh-watchdogs, `_`. +type OwnBuckets struct{ Presence, Pulse, Stuck string } + +// BucketsOf names the watchman's own buckets from module, its credential's, never a constant (design 47 G A-296). +func BucketsOf(module string) OwnBuckets { + return OwnBuckets{Presence: module + "_" + meshwatchdogs.StatePresence, Pulse: module + "_" + meshwatchdogs.StatePulse, + Stuck: module + "_" + meshwatchdogs.StateStuck} +} + // The findings a value of another part is raised as, named by their words. const ( FindingManifestOutgrown = "manifest-outgrown" diff --git a/modules/mesh-delivery-watchman/internal/watch/presence.go b/modules/mesh-delivery-watchman/internal/watch/presence.go index b9881a31..263e1d18 100644 --- a/modules/mesh-delivery-watchman/internal/watch/presence.go +++ b/modules/mesh-delivery-watchman/internal/watch/presence.go @@ -92,7 +92,7 @@ func (w *Watchman) settle(key string) { defer w.applyCheck(node) // S21 is judged only while the machine is heard } if online { - w.moved(PresenceBucket, key) + w.moved(w.own.Presence, key) return } w.ensureOffline(key) @@ -101,7 +101,7 @@ func (w *Watchman) settle(key string) { // ensureOffline mirrors the return of a node or part that is offline. func (w *Watchman) ensureOffline(key string) { p := w.presence[key] - w.mirror(PresenceBucket, key, waitedOf(key), "", p.since, 0) + w.mirror(w.own.Presence, key, waitedOf(key), "", p.since, 0) } // SeedPresence tells the watchman a presence key as it stands in its bucket at start, with the time it was written. diff --git a/modules/mesh-delivery-watchman/internal/watch/signals.go b/modules/mesh-delivery-watchman/internal/watch/signals.go index 115687ed..9391c456 100644 --- a/modules/mesh-delivery-watchman/internal/watch/signals.go +++ b/modules/mesh-delivery-watchman/internal/watch/signals.go @@ -235,7 +235,7 @@ func (w *Watchman) Power(node, state string) { // moved is S1, the machine silent, unless it said it is away; any other key not moved is `delivery..stuck`; a key // moved ends its finding. func (w *Watchman) stuckFinding(sk string, s meshwatchdogsclient.Stuck) { - if s.Bucket == PresenceBucket { + if s.Bucket == w.own.Presence { if node, part, _, err := meshwatchdogsclient.ParsePresenceKey(s.Key); err == nil && part == "" && node != "" { key := FindingKey(ScopeMachine, node, KindSilent) if s.Moved { diff --git a/modules/mesh-delivery-watchman/internal/watch/watch.go b/modules/mesh-delivery-watchman/internal/watch/watch.go index 4213a57f..d17e328e 100644 --- a/modules/mesh-delivery-watchman/internal/watch/watch.go +++ b/modules/mesh-delivery-watchman/internal/watch/watch.go @@ -56,6 +56,7 @@ type pulse struct { // safe to call from several watches at once. type Watchman struct { mu sync.Mutex + own OwnBuckets // its own buckets, named from its credential's module (A-296) out Out now func() time.Time logf func(format string, args ...any) @@ -120,9 +121,9 @@ type Watchman struct { opened map[string]Finding } -// 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{}, +// New is the watchman of module (its credential's), writing to out, reading the time from now. +func New(module string, out Out, now func() time.Time) *Watchman { + return &Watchman{own: BucketsOf(module), 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, progressBound: ProgressBound, findings: map[string][]byte{}, seededFindings: map[string]bool{}, signals: map[string]signalDue{}, busCreated: map[string]bool{}, takeInHeld: map[string]Finding{}, pendingBranch: map[string]string{}, takenEntries: map[string][]string{}, recordMilestone: map[string]string{}, away: map[string]bool{}, advised: map[string]int{}, advisedSince: map[string]time.Time{}, diff --git a/modules/mesh-delivery-watchman/internal/watch/watch_test.go b/modules/mesh-delivery-watchman/internal/watch/watch_test.go index 591d1748..c45bc463 100644 --- a/modules/mesh-delivery-watchman/internal/watch/watch_test.go +++ b/modules/mesh-delivery-watchman/internal/watch/watch_test.go @@ -95,7 +95,7 @@ func (c *clock) now() time.Time { return c.t } func setup() (*Watchman, *fakeOut, *clock) { o, c := newOut(), &clock{t: t0} - w := New(o, c.now) + w := New(testModule, o, c.now) w.SetStepBound(meshwatchdogsclient.PulseMoveBound) // the older tests move a step as they move any key: stuck at the pulse's bound return w, o, c } @@ -109,7 +109,8 @@ func checkValue(started time.Time, head string) []byte { const ( checksKey = "SEAT_NODE_BUILD_GO" - presenceNS = PresenceBucket + testModule = "a-watchman" + presenceNS = testModule + "_presence" ) // askBody is the body of a build seat's step-progress, ask-taken or step-ended for an ask at a time. @@ -549,7 +550,7 @@ func TestAnEventThatDoesNotReadOrIsNoBuildSeatsIsLeft(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) + w := New(testModule, o, c.now) askTaken(w, "askid1", t0) k := pulseKeyOf(t, checksKey, "askid1") if got := o.ttls[k]; got != ProgressBound { @@ -581,7 +582,7 @@ func TestARunningStepThatSaysItsProgressIsNeverStuckAndOneWhoseHolderStopsIs(t * func TestARunningStepPastItsBoundIsStuck(t *testing.T) { // 1908 o, c := newOut(), &clock{t: t0} - w := New(o, c.now) + w := New(testModule, o, c.now) askProgress(w, "askid1", t0) c.t = t0.Add(6 * time.Minute) w.Pulse(pulseKeyOf(t, checksKey, "askid1"), Expired, c.t) @@ -591,7 +592,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)} - askProgress(New(o2, c2.now), "askid1", t0) + askProgress(New(testModule, 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) } @@ -662,7 +663,7 @@ func TestAHaveThatCameBeforeTheWantEndsItAtOnceAndARemovedWantHasNoPulse(t *test func TestAWantNotAppliedByItsBoundIsStuckAndAKeyDeletedEndsIt(t *testing.T) { o, c := newOut(), &clock{t: t0} - w := New(o, c.now) + w := New(testModule, o, c.now) w.Want("want.ace.web", wantValue(t, fpNew, false), Put, t0) pk := pulseKeyOf(t, AssignerWantsBucket, "want.ace.web") c.t = t0.Add(31 * time.Minute) diff --git a/modules/mesh-delivery-watchman/manifest_test.go b/modules/mesh-delivery-watchman/manifest_test.go index c7056d48..1bbbfc0b 100644 --- a/modules/mesh-delivery-watchman/manifest_test.go +++ b/modules/mesh-delivery-watchman/manifest_test.go @@ -168,10 +168,11 @@ 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}, + "module": {watch.WatchmanModule, m.Module}, + // A-296: its own buckets are named from its credential's module, each a state its manifest keeps. + "presence": {watch.BucketsOf("a-watchman").Presence, "a-watchman_" + m.State[0].Name}, + "pulse": {watch.BucketsOf("a-watchman").Pulse, "a-watchman_" + m.State[1].Name}, + "stuck": {watch.BucketsOf("a-watchman").Stuck, "a-watchman_" + m.State[2].Name}, "wants": {watch.AssignerWantsBucket, bucket(m.Consumes, "mesh-delivery-assigner.views.wants")}, "provisions": {watch.ProvisionsBucket, bucket(m.Reads, "mesh-provisioner.provisions")}, "assigner": {watch.AssignerModule + ".assigner_read_wants", entry(m.Invokes, "mesh-delivery-assigner.assigner_read_wants")},