From ec7e672c1f166584dbc730332d2176524760d397 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 23:47:59 +0100 Subject: [PATCH] B2 (watchman): the wants are followed at the holder of the seat mesh-wants, the indexer's tools at the holder of mesh-catalog MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit FollowSeat(mesh-wants, wants) replaces the follow by the assigner's module; the pulse and stuck keys name the holder's real bucket taken at each use (WantsAt, AwaitsWants), a start never waiting for it and the seeded keys of any `_wants` kept until the late read (A-321). indexer_pending and indexer_verdicts are asked of the holder of mesh-catalog on each call. The one catch-all of the seats' events is one handler per seat (A-322), source-moved taken from the branches follower (AdoptEvent). The manifest consumes the seat's views and the seat-holders view. Builds: design 47 G A-210, A-318, A-321, A-322, §428 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../cmd/mesh-delivery-watchman/consumes.go | 31 +++++++-- .../integration_test.go | 55 ++++++++++++++-- .../cmd/mesh-delivery-watchman/main.go | 54 ++++++++++++---- .../cmd/mesh-delivery-watchman/main_test.go | 12 +++- .../internal/watch/mirror.go | 21 +++--- .../internal/watch/names.go | 7 +- .../internal/watch/verdicts.go | 5 +- .../internal/watch/watch.go | 64 ++++++++++++++++--- .../internal/watch/watch_test.go | 23 +++++-- .../mesh-delivery-watchman/manifest_test.go | 10 +-- modules/mesh-delivery-watchman/module.json | 6 +- 11 files changed, 228 insertions(+), 60 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 624de9cc..fd878ccc 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go @@ -10,6 +10,8 @@ import ( "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/nodebuildgo" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/nodebuildimage" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/nodebuildnode" "slices" "strings" "sync" @@ -34,8 +36,11 @@ import ( // Consumed event names, as the consumes name them (the emitter is read from the subject's emitter token). const ( eventAccepted = "manifest-accepted" - eventCreated = "created" - eventRefused = "refused" + // eventManifestRefused is the seat mesh-catalog's manifest-refused; meshGrantsSeat the seat of the store whose filling, full, room and waits it hears. + eventManifestRefused = "manifest-refused" + meshGrantsSeat = "mesh-grants" + eventCreated = "created" + eventRefused = "refused" ) // consumed hands one consumed event to the watchman, by its subject: manifest-accepted, and the seat mesh-broker's created @@ -85,15 +90,31 @@ func hearConsumed(ctx context.Context, js jetstream.JetStream, cred process.Acco // one taker; and the binding asks show again at each bind of the hub's consumer, as its Register does. brokerEvents := []string{seatdef.EventSubject(meshbroker.Seat, "recreated"), seatdef.EventSubject(meshbroker.Seat, "dropped")} hub.OnBound(func(time.Time) { binding.Start() }) - if err := hub.Handle("mesh.seat.*.event.>", func(m jetstream.Msg) { + handleSeat := func(m jetstream.Msg) { if slices.Contains(brokerEvents, m.Subject()) { binding.Hear(m) return } consumed(w, m.Subject(), m.Data(), logf) _ = m.Ack() - }); err != nil { - w.NotHeard("EVENTS", "the seats' events cannot be handled", err) + } + // One handler per seat it consumes, since a message has one taker (design 47 G A-322): the events the branches follower of the + // catalog half adopts (the seat mesh-catalog's registration-changed, the seat mesh-delivery's source-moved) are the follower's, and + // the watchman takes source-moved from it (AdoptEvent, main.go), never from a second handler. + var patterns []string + for _, seat := range []string{meshwants.Seat, meshGrantsSeat, meshpipelinerunner.Seat, nodeengine.Seat, meshbroker.Seat, nodebuildgo.Seat, nodebuildimage.Seat, nodebuildnode.Seat} { + patterns = append(patterns, "mesh.seat."+seat+".event.>") + } + for _, e := range []string{meshdelivery.EventRecordOpened, meshdelivery.EventRecordEnded, meshdelivery.EventFilling, meshdelivery.EventFull, meshdelivery.EventRoom} { + patterns = append(patterns, seatdef.EventSubject(meshdelivery.Seat, e)) + } + for _, e := range []string{eventAccepted, eventManifestRefused, meshdelivery.EventFilling, meshdelivery.EventFull, meshdelivery.EventRoom} { + patterns = append(patterns, seatdef.EventSubject(meshcatalog.Seat, e)) + } + for _, pattern := range patterns { + if err := hub.Handle(pattern, handleSeat); err != nil { + w.NotHeard("EVENTS", "the seats' events cannot be handled", err) + } } var mu sync.Mutex opened := map[string]bool{} 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 794afb0b..b73bd86f 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 @@ -21,6 +21,8 @@ import ( "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshwatchdogsclient" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshwantsclient" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" "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" @@ -67,7 +69,16 @@ accounts { // families (DLV-M5, ADR 0615): the watchman (n1) follows both, never their buckets. func serveOwners(t *testing.T, ctx context.Context, js jetstream.JetStream, nc *nats.Conn) { t.Helper() - viewstest.Serve(t, ctx, js, nc, watch.AssignerModule, wantsState, watch.AssignerModule+".assigner_read_wants", "n1_"+watchmanModule) + // The wants are followed in the seat form, at the module the seat-holders view of the catalog half's claimant names for the seat + // mesh-wants (design 47 G §717, A-321): the stand-in holder's real bucket is `_wants`, which the watchman names its keys by. + viewstest.Serve(t, ctx, js, nc, testAssigner, meshwants.StateWants, testAssigner+"."+meshwants.StateWants, "n1_"+watchmanModule) + holders := viewstest.Serve(t, ctx, js, nc, testAssigner, "seat-holders", testAssigner+".seat-holders", "n1_"+watchmanModule) + for _, seat := range []string{"mesh-module-assigner", meshwants.Seat} { + if _, err := holders.Put(ctx, seat, []byte(`{"seat":"`+seat+`","holders":[{"module":"`+testAssigner+`","node":"n1"}]}`)); err != nil { + t.Fatal(err) + } + } + serveCatalogHalf(t, nc) } @@ -171,7 +182,7 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8 {Bucket: own.Presence, History: 20}, {Bucket: own.Stuck, History: 20}, {Bucket: own.Pulse, LimitMarkerTTL: time.Second}, - {Bucket: watch.AssignerWantsBucket, History: 20}, + {Bucket: testWantsBucket, History: 20}, {Bucket: (watchmanModule + "_findings"), History: 5}, {Bucket: firstViewsBucket, History: 1}, } { @@ -419,7 +430,7 @@ func anAskRunningBeforeTheStartGetsItsPulse(t *testing.T) { // 1909 {Bucket: own.Presence, History: 20}, {Bucket: own.Stuck, History: 20}, {Bucket: own.Pulse, LimitMarkerTTL: time.Second}, - {Bucket: watch.AssignerWantsBucket, History: 20}, + {Bucket: testWantsBucket, History: 20}, {Bucket: (watchmanModule + "_findings"), History: 5}, {Bucket: firstViewsBucket, History: 1}, } { @@ -500,7 +511,7 @@ func aWantIsMirroredAsAPulseAndTheHaveNamingItEndsIt(t *testing.T) { {Bucket: own.Presence, History: 20}, {Bucket: own.Stuck, History: 20}, {Bucket: own.Pulse, LimitMarkerTTL: time.Second}, - {Bucket: watch.AssignerWantsBucket, History: 20}, + {Bucket: testWantsBucket, History: 20}, {Bucket: (watchmanModule + "_findings"), History: 5}, {Bucket: firstViewsBucket, History: 1}, } { @@ -522,7 +533,7 @@ func aWantIsMirroredAsAPulseAndTheHaveNamingItEndsIt(t *testing.T) { t.Fatal(err) } key, _ := meshwantsclient.WantKey(node, module) - if _, err := kvs[watch.AssignerWantsBucket].Put(ctx, key, body); err != nil { + if _, err := kvs[testWantsBucket].Put(ctx, key, body); err != nil { t.Fatal(err) } } @@ -548,7 +559,7 @@ func aWantIsMirroredAsAPulseAndTheHaveNamingItEndsIt(t *testing.T) { t.Cleanup(func() { stop(); <-done }) pulse := func(node, module string) string { wk, _ := meshwantsclient.WantKey(node, module) - pk, err := meshwatchdogsclient.PulseKey(watch.AssignerWantsBucket, wk) + pk, err := meshwatchdogsclient.PulseKey(testWantsBucket, wk) if err != nil { t.Fatal(err) } @@ -581,9 +592,39 @@ func aWantIsMirroredAsAPulseAndTheHaveNamingItEndsIt(t *testing.T) { } for _, m := range []string{"web"} { wk, _ := meshwantsclient.WantKey("n2", m) - sk, _ := meshwatchdogsclient.StuckKey(watch.AssignerWantsBucket, wk) + sk, _ := meshwatchdogsclient.StuckKey(testWantsBucket, wk) 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) } } } + +// testAssigner is the stand-in holder of the seats mesh-wants and mesh-module-assigner; testWantsBucket the wants' bucket at it. +const ( + testAssigner = "acme-assigner" + testWantsBucket = testAssigner + "_" + meshwants.StateWants +) + +// serveCatalogHalf answers the catalog half's first views (design 47 G §614): the gate's one source branch, the stand-in +// registered on it, and its trunk revision whose entry claims mesh-module-assigner. +func serveCatalogHalf(t *testing.T, nc *nats.Conn) { + t.Helper() + respond := func(a any) nats.MsgHandler { + raw, _ := json.Marshal(map[string]any{"result": a, "node": "n0"}) + return func(m *nats.Msg) { _ = m.Respond(raw) } + } + revision := meshcatalog.RevisionAnswer{Revision: &meshcatalog.CatalogRevision{Revision: 1, Kind: meshcatalogclient.KindTrunk}, + Entries: []meshcatalog.CatalogEntry{{Module: testAssigner, ManifestDigest: "d1", Manifest: map[string]any{"module": testAssigner, + "claims": []any{map[string]any{"name": "mesh-module-assigner", "scope": "mesh"}}}}}} + registrations := meshcatalog.RegistrationsAnswer{Registrations: []meshcatalog.Registration{{Module: testAssigner, Version: 1, Branches: []string{"release"}}}} + for subject, handle := range map[string]nats.MsgHandler{"mesh.seat.mesh-catalog.call.revision": respond(revision), + "mesh.seat.mesh-catalog.call.registrations": respond(registrations), "mesh.seat.mesh-delivery.call.source": respond(map[string]any{"branches": []string{"release"}})} { + for _, s := range []string{subject, subject + ".>"} { + sub, err := nc.Subscribe(s, handle) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = sub.Unsubscribe() }) + } + } +} 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 efc8e1bc..a39e2c55 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go @@ -10,6 +10,10 @@ import ( "encoding/json" "errors" "fmt" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" "io" "log" "os" @@ -143,7 +147,7 @@ func settle(ctx context.Context, kv jetstream.KeyValue, each func(jetstream.KeyV // readLate reads an owner's state the watchman follows by its first view (pic), on a goroutine of its own once that is // answered, so an owner that is down never holds the start: every key it holds now, then the start's end for its bucket // (watch.ReadLate), then each change until ctx ends. -func readLate(ctx context.Context, w *watch.Watchman, pic *views.Picture, bucket string, each func(jetstream.KeyValueEntry)) error { +func readLate(ctx context.Context, w *watch.Watchman, pic *views.Picture, bucket func() string, each func(jetstream.KeyValueEntry)) error { select { case <-ctx.Done(): return nil @@ -156,7 +160,7 @@ func readLate(ctx context.Context, w *watch.Watchman, pic *views.Picture, bucket } return err } - w.ReadLate(bucket) + w.ReadLate(bucket()) return follow(ctx, sw, each) } @@ -189,11 +193,6 @@ func createdOr(e jetstream.KeyValueEntry, at time.Time) time.Time { return at } -const ( - wantsState = "wants" - // The owners' views families, as their manifests' `streams` name them. -) - // run is the program: it returns when ctx ends or the bus cannot be used. func run(ctx context.Context, getenv func(string) string, logOut io.Writer) error { logger := log.New(logOut, "mesh-delivery-watchman: ", log.LstdFlags) @@ -257,15 +256,31 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro defer held.w.Close() reader := &views.ProcessReader{Module: cred.Module, Call: calls.Call, Logf: logger.Printf, Hub: hub, Findings: func() views.WireWriter { return held.w }} - wants, err := reader.Follow(watch.AssignerModule, wantsState, watch.AssignerModule+".assigner_read_wants") + // The catalog half's claimant says each seat's holder (design 47 G §717, A-298): the indexer's tools are asked of the holder of + // the seat mesh-catalog when asked. + branches, err := meshcatalogclient.FollowBranchesProcess(reader, calls.Call) + if err != nil { + return err + } + branches.Logf = logger.Printf + holders := branches.SeatHolders(ctx) + // The assigner's wants are followed in the seat form, at the module holding the seat mesh-wants (design 47 G §717, A-318, A-321): + // the bucket a pulse or a stuck key names is that holder's real bucket, taken at each use, and a start never waits for it. + wants, err := reader.FollowSeat(meshwants.Seat, meshwants.StateWants, holders) if err != nil { return err } 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) + // The seat mesh-delivery's source-moved is the branches follower's event (one taker, design 47 G A-322): the watchman takes it from the + // follower's subscription, beside the follower's handler. + if err := reader.AdoptEvent(meshdelivery.Seat, meshdelivery.EventSourceMoved, func(_ context.Context, data []byte) error { w.SourceMoved(data); return nil }); err != nil { + return err + } // 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) + w.WantsAt(func() string { return reader.HolderBucket(meshwants.Seat, meshwants.StateWants) }) + w.AwaitsWants() // A test shortens how long a move may take; the mesh never sets it. if v := strings.TrimSpace(getenv("MESH_DELIVERY_PULSE_BOUND")); v != "" { d, err := time.ParseDuration(v) @@ -382,8 +397,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if calls, err := process.Connect(cred, nats.Name(cred.Node+"."+cred.Module+".calls")); err != nil { logger.Printf("take-in: no connection to ask %s (%v), so a wait the indexer held before this start is said only once a take-in event of it is heard", pendingTool, err) } else { - seedTakeIn(w, calls.Call, logger.Printf) - seedVerdicts(w, calls.Call, logger.Printf) + atCatalog := atHolder(calls.Call, holders, meshcatalog.Seat) // the indexer holds the seat mesh-catalog + seedTakeIn(w, atCatalog, logger.Printf) + seedVerdicts(w, atCatalog, logger.Printf) calls.Close() } // The take-in events are the indexer's ordinary events (its emits), heard on the shared events stream through this @@ -418,7 +434,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro hearConsumed(ctx, js, cred, hub, w, binding, logger.Printf) defer hub.Stop() go func() { - errs <- readLate(ctx, w, wants, watch.AssignerWantsBucket, func(e jetstream.KeyValueEntry) { + errs <- readLate(ctx, w, wants, func() string { return reader.HolderBucket(meshwants.Seat, meshwants.StateWants) }, func(e jetstream.KeyValueEntry) { w.Want(e.Key(), e.Value(), opOf(e), createdOr(e, time.Now())) }) }() @@ -462,7 +478,19 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro const eventsStream = "EVENTS" // pendingTool is the indexer's tool that answers the waits it holds (ADR 0633 §5), as the view asks it. -const pendingTool = watch.IndexerModule + ".indexer_pending" +const pendingTool = "indexer_pending" + +// atHolder asks a tool of the first module that holds the seat when it is asked (design 47 G A-210, §428), as the view's tools do: the +// holder comes from the seat-holders view of the catalog half's claimant, found for each call, and none is an error that says so. +func atHolder(call func(key string, body any) (json.RawMessage, error), holders func(seat string) []string, seat string) func(key string, body any) (json.RawMessage, error) { + return func(tool string, body any) (json.RawMessage, error) { + h := holders(seat) + if len(h) == 0 { + return nil, fmt.Errorf("the holder of %s is not found, so %s is not asked", seat, tool) + } + return call(h[0]+"."+tool, body) + } +} // seedTakeIn asks the indexer the waits it holds now and starts each (Watchman.SeedTakeIn). An indexer that does not // answer is said: the waits are then known only from the events heard after. 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 1f684d63..75f1ff81 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 @@ -5,6 +5,8 @@ import ( "encoding/json" "errors" "fmt" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" "io" "strings" "testing" @@ -137,6 +139,12 @@ func TestRunNeedsItsCredential(t *testing.T) { // 1135/AC6 } w.TakeIn(takein.TakeInPendingEvent, body) seedTakeIn(w, call, func(string, ...any) {}) + var asked string + seedTakeIn(w, atHolder(func(key string, _ any) (json.RawMessage, error) { asked = key; return nil, errors.New("no answer") }, + func(seat string) []string { return []string{"acme-indexer"} }, meshcatalog.Seat), func(string, ...any) {}) + if asked != "acme-indexer."+pendingTool { // the tool is asked of the holder of the seat mesh-catalog, found when asked + t.Errorf("%s: asked %q, want the holder's tool", name, asked) + } if !out.findings[old.Key()] || out.pulseDeletes != 0 { t.Errorf("%s: finding stands %v, signals ended %d: want the previous run's finding and the held wait kept", name, out.findings[old.Key()], out.pulseDeletes) } @@ -147,10 +155,10 @@ func TestRunNeedsItsCredential(t *testing.T) { // 1135/AC6 // the time the owner's bus wrote the key, never the watchman's start. written := time.Date(2026, 11, 4, 9, 0, 0, 0, time.UTC) r := &views.ProcessReader{Module: watchmanModule, Call: func(string, any) (json.RawMessage, error) { - return json.Marshal(views.Answer{V: views.Version, Owner: watch.AssignerModule, State: wantsState, Revision: 3, Created: written, + return json.Marshal(views.Answer{V: views.Version, Owner: "acme-assigner", State: meshwants.StateWants, Revision: 3, Created: written, At: written.Add(time.Hour), Values: map[string]json.RawMessage{"ace.mail": json.RawMessage(`{}`)}, Written: map[string]time.Time{"ace.mail": written}}) }} - p, err := r.Follow(watch.AssignerModule, wantsState, watch.AssignerModule+".assigner_read_wants") + p, err := r.Follow("acme-assigner", meshwants.StateWants, "acme-assigner.wants") if err != nil { t.Fatal(err) } diff --git a/modules/mesh-delivery-watchman/internal/watch/mirror.go b/modules/mesh-delivery-watchman/internal/watch/mirror.go index a45fc04f..afd564e6 100644 --- a/modules/mesh-delivery-watchman/internal/watch/mirror.go +++ b/modules/mesh-delivery-watchman/internal/watch/mirror.go @@ -112,16 +112,19 @@ func (w *Watchman) Want(key string, value []byte, op Op, at time.Time) { } w.mu.Lock() defer w.mu.Unlock() + if w.wantsBucket() == "" { + return // no holder of the seat mesh-wants is followed: the key names no bucket (A-321) + } switch op { case Expired: return case Delete: delete(w.wants, key) delete(w.held, key) - w.moved(AssignerWantsBucket, key) + w.moved(w.wantsBucket(), key) return } - w.stands(AssignerWantsBucket, key) + w.stands(w.wantsBucket(), key) want, err := meshwantsclient.DecodeWant(value) if err != nil { w.say("the want %s cannot be read, and is left: %v", key, err) @@ -130,7 +133,7 @@ func (w *Watchman) Want(key string, value []byte, op Op, at time.Time) { if want.Removed { delete(w.wants, key) delete(w.held, key) - w.moved(AssignerWantsBucket, key) + w.moved(w.wantsBucket(), key) return } w.wants[key] = want.Fp @@ -141,26 +144,26 @@ func (w *Watchman) Want(key string, value []byte, op Op, at time.Time) { at = w.now() } w.held[key] = true - w.mirrorKept(AssignerWantsBucket, key, + w.mirrorKept(w.wantsBucket(), key, describe("the hold on %s on %s at %s (rule %d) to end, bound %s: %s", module, node, shortFP(want.Fp), want.Rule, w.stepBound, want.Words), "held|"+want.Fp+"|"+want.Newest, at) return } if w.held[key] { delete(w.held, key) - if pk, err := meshwatchdogsclient.PulseKey(AssignerWantsBucket, key); err == nil { + if pk, err := meshwatchdogsclient.PulseKey(w.wantsBucket(), key); err == nil { delete(w.starts, pk) } - w.stuckMoved(AssignerWantsBucket, key) // the hold ended: its finding is cleared + w.stuckMoved(w.wantsBucket(), key) // the hold ended: its finding is cleared } if w.have(node, module) == want.Fp { - w.moved(AssignerWantsBucket, key) + w.moved(w.wantsBucket(), key) return } if at.IsZero() { at = w.now() } - w.mirrorWithin(w.stepBound, AssignerWantsBucket, key, + w.mirrorWithin(w.stepBound, w.wantsBucket(), key, describe("the node agent of %s to run %s at %s", node, module, shortFP(want.Fp)), want.Fp, at, 0) } @@ -188,7 +191,7 @@ func (w *Watchman) EngineRunning(node, module string, body []byte) { return } if fp, ok := w.wants[wk]; ok && fp == r.Fingerprint { - w.moved(AssignerWantsBucket, wk) + w.moved(w.wantsBucket(), wk) } } diff --git a/modules/mesh-delivery-watchman/internal/watch/names.go b/modules/mesh-delivery-watchman/internal/watch/names.go index d26efc7c..a1d78974 100644 --- a/modules/mesh-delivery-watchman/internal/watch/names.go +++ b/modules/mesh-delivery-watchman/internal/watch/names.go @@ -2,20 +2,21 @@ package watch import ( meshwatchdogs "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshwatchdogs" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" ) // The holders and buckets the watchman names, taken from its own manifest (module.json: its name, its `reads`, its // states and the tools it invokes) and the seats' state names; a unit row checks them against it (TestManifestStateAndReads). // No contract names a holder (novox/hq ADR 0625 §4, ADR 0634 §1). const ( - AssignerModule = "mesh-delivery-assigner" // ProvisionsBucket is the seat mesh-provisioner's state provisions (reads: mesh-provisioner.provisions). ProvisionsBucket = "mesh-provisioner_provisions" // 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" - // AssignerWantsBucket is the assigner's state wants, followed by its views family (consumes: mesh-delivery-assigner.views.wants). - AssignerWantsBucket = AssignerModule + "_wants" + // WantsState is the state wants of the seat mesh-wants, followed in the seat form (consumes: seat:mesh-wants.views.wants); the + // bucket a pulse or stuck key of it names is the holder's real bucket `_wants`, taken at each use (WantsAt, A-321). + WantsState = meshwants.StateWants ) // OwnBuckets are the watchman's own states of the seat mesh-watchdogs, `_`. diff --git a/modules/mesh-delivery-watchman/internal/watch/verdicts.go b/modules/mesh-delivery-watchman/internal/watch/verdicts.go index dfdf8a58..d0f7d2e9 100644 --- a/modules/mesh-delivery-watchman/internal/watch/verdicts.go +++ b/modules/mesh-delivery-watchman/internal/watch/verdicts.go @@ -23,8 +23,9 @@ const KindManifestOutgrown = FindingManifestOutgrown // `manifest-outgrown` or, with none, clears it and any refusal. The watchman seeds both from the indexer's verdicts at its start // (VerdictsTool). `manifest-waits` is no longer written: every revision is judged by a validator built from its own heads (§16). -// VerdictsTool is the indexer's tool that lists its verdicts, which seeds the findings at the start. -const VerdictsTool = IndexerModule + ".indexer_verdicts" +// VerdictsTool is the indexer's tool that lists its verdicts, which seeds the findings at the start; it is asked of the holder of +// the seat mesh-catalog when it is asked (design 47 G A-210, §428), never of a module named here. +const VerdictsTool = "indexer_verdicts" // VerdictRow is one verdict as the tool lists it (the fields the watchman reads). type VerdictRow struct { diff --git a/modules/mesh-delivery-watchman/internal/watch/watch.go b/modules/mesh-delivery-watchman/internal/watch/watch.go index 08aab116..43b15745 100644 --- a/modules/mesh-delivery-watchman/internal/watch/watch.go +++ b/modules/mesh-delivery-watchman/internal/watch/watch.go @@ -5,6 +5,7 @@ package watch import ( "fmt" + "strings" "sync" "time" @@ -106,6 +107,10 @@ type Watchman struct { // pulses, stuck keys and findings wait for its own end (ReadLate), so a restart while its owner is down neither drops // nor raises again what it holds. awaited map[string]bool + // wantsAt names the bucket of the assigner's wants as the holder of the seat mesh-wants keeps it, "" while none is followed + // (WantsAt, design 47 G A-321); awaitsWants keeps the seeded entries of any `_wants` bucket until the late read. + wantsAt func() string + awaitsWants bool // heads is, by watched branch, the head still waiting for its pipeline (pipelines.go). heads map[string]head // pendingBranch is, by key, the watched branch of each open take-in wait that names one (takein.go, design 47 G §497). @@ -313,7 +318,7 @@ func (w *Watchman) Reconcile() { w.mu.Lock() defer w.mu.Unlock() for pk, p := range w.pulses { - if p.seeded && !w.awaited[p.bucket] { + if p.seeded && !w.awaits(p.bucket) { w.deletePulse(pk) } } @@ -338,7 +343,7 @@ func (w *Watchman) ReadWhole(bucket string) { // stands records a mirrored key read standing during the start. func (w *Watchman) stands(bucket, key string) { - if w.reconciled && !w.awaited[bucket] { + if w.reconciled && !w.awaits(bucket) { return } if w.standing[bucket] == nil { @@ -349,7 +354,7 @@ func (w *Watchman) stands(bucket, key string) { func (w *Watchman) closeGone() { for sk, s := range w.stuck { - if s.Moved || w.awaited[s.Bucket] || !w.readWhole[s.Bucket] || w.standing[s.Bucket][s.Key] { + if s.Moved || w.awaits(s.Bucket) || !w.readWhole[s.Bucket] || w.standing[s.Bucket][s.Key] { continue } w.closeStuck(sk, s) @@ -358,6 +363,9 @@ func (w *Watchman) closeGone() { for b := range w.awaited { standing[b] = w.standing[b] } + if b := w.wantsBucket(); w.awaitsWants && b != "" { + standing[b] = w.standing[b] + } w.standing, w.readWhole, w.reconciled = standing, map[string]bool{}, true } @@ -383,27 +391,67 @@ func (w *Watchman) Awaits(bucket string) { func (w *Watchman) ReadLate(bucket string) { w.mu.Lock() defer w.mu.Unlock() - if !w.awaited[bucket] { + if !w.awaits(bucket) { return } + // The wants' late read ends the start for every `_wants` bucket a previous run named (A-321): the keys of a holder that + // is gone are not standing in the one followed now. + of := func(b string) bool { + return b == bucket || (w.awaitsWants && isWantsBucket(bucket) && isWantsBucket(b)) + } for pk, p := range w.pulses { - if p.seeded && p.bucket == bucket { + if p.seeded && of(p.bucket) { w.deletePulse(pk) } } for sk, s := range w.stuck { - if !s.Moved && s.Bucket == bucket && !w.standing[bucket][s.Key] { + if !s.Moved && of(s.Bucket) && !w.standing[bucket][s.Key] { w.closeStuck(sk, s) } } for key := range w.seededFindings { - if f, err := DecodeFinding(w.findings[key]); err == nil && f.Source == bucket { + if f, err := DecodeFinding(w.findings[key]); err == nil && of(f.Source) { delete(w.seededFindings, key) w.unfind(key) } } delete(w.awaited, bucket) delete(w.standing, bucket) + if w.awaitsWants && isWantsBucket(bucket) { + w.awaitsWants = false + } +} + +// WantsAt says how the bucket of the assigner's wants is named, the holder's real bucket `_wants` at each use ("" while the +// seat mesh-wants has no followed holder; design 47 G A-321, A-303). +func (w *Watchman) WantsAt(bucket func() string) { + w.mu.Lock() + defer w.mu.Unlock() + w.wantsAt = bucket +} + +// AwaitsWants awaits the wants as Awaits does a bucket (design 47 G A-321): a start never waits for the holder, and the seeded +// pulses, stuck keys and findings of any `_wants` bucket are kept until the late read of the resolved one. Called before +// Reconcile. +func (w *Watchman) AwaitsWants() { + w.mu.Lock() + defer w.mu.Unlock() + w.awaitsWants = true +} + +func isWantsBucket(b string) bool { return strings.HasSuffix(b, "_"+WantsState) } + +// awaits says bucket is read only after the start's end (Awaits, WantsAt). +func (w *Watchman) awaits(bucket string) bool { + return w.awaited[bucket] || (w.awaitsWants && isWantsBucket(bucket)) +} + +// wantsBucket is the bucket of the wants now, "" while the seat mesh-wants has no followed holder. +func (w *Watchman) wantsBucket() string { + if w.wantsAt == nil { + return "" + } + return w.wantsAt() } // reconcileFindings ends the start for the findings: one read at the start that what the watchman read since did not @@ -415,7 +463,7 @@ func (w *Watchman) reconcileFindings() { kept[due.Finding.Key()] = true } for key := range w.seededFindings { - if f, err := DecodeFinding(w.findings[key]); err == nil && w.awaited[f.Source] { + if f, err := DecodeFinding(w.findings[key]); err == nil && w.awaits(f.Source) { continue // ended by ReadLate } delete(w.seededFindings, key) diff --git a/modules/mesh-delivery-watchman/internal/watch/watch_test.go b/modules/mesh-delivery-watchman/internal/watch/watch_test.go index c45bc463..9550a9bf 100644 --- a/modules/mesh-delivery-watchman/internal/watch/watch_test.go +++ b/modules/mesh-delivery-watchman/internal/watch/watch_test.go @@ -96,6 +96,7 @@ func (c *clock) now() time.Time { return c.t } func setup() (*Watchman, *fakeOut, *clock) { o, c := newOut(), &clock{t: t0} w := New(testModule, o, c.now) + w.WantsAt(testWants) 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 } @@ -551,6 +552,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(testModule, o, c.now) + w.WantsAt(testWants) askTaken(w, "askid1", t0) k := pulseKeyOf(t, checksKey, "askid1") if got := o.ttls[k]; got != ProgressBound { @@ -583,6 +585,7 @@ func TestARunningStepThatSaysItsProgressIsNeverStuckAndOneWhoseHolderStopsIs(t * func TestARunningStepPastItsBoundIsStuck(t *testing.T) { // 1908 o, c := newOut(), &clock{t: t0} w := New(testModule, o, c.now) + w.WantsAt(testWants) askProgress(w, "askid1", t0) c.t = t0.Add(6 * time.Minute) w.Pulse(pulseKeyOf(t, checksKey, "askid1"), Expired, c.t) @@ -592,7 +595,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(testModule, o2, c2.now), "askid1", t0) + askProgress(newWith(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) } @@ -664,6 +667,7 @@ func TestAHaveThatCameBeforeTheWantEndsItAtOnceAndARemovedWantHasNoPulse(t *test func TestAWantNotAppliedByItsBoundIsStuckAndAKeyDeletedEndsIt(t *testing.T) { o, c := newOut(), &clock{t: t0} w := New(testModule, o, c.now) + w.WantsAt(testWants) 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) @@ -691,7 +695,7 @@ func TestAStuckKeyWhoseKeyIsGoneIsClosedAtTheStart(t *testing.T) { standing func(w *Watchman) // what the start reads standing readWhole bool closed bool - late bool // read after the start's end (Awaits, ReadLate) + late bool // read after the start's end (AwaitsWants, ReadLate) }{ {"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}, @@ -699,6 +703,9 @@ func TestAStuckKeyWhoseKeyIsGoneIsClosedAtTheStart(t *testing.T) { // they hold waits for it, and is then reconciled as at the start. {"a want deleted while nobody heard, read from a late first view", AssignerWantsBucket, wantKey, func(*Watchman) {}, true, true, true}, {"a want still standing, read from a late first view", AssignerWantsBucket, wantKey, func(w *Watchman) { w.Want(wantKey, wantValue(t, fpNew, false), Put, t0) }, true, false, true}, + // A-321: the holder is not followed at the start, so an earlier run's `_wants` keys wait for the late read of the + // bucket resolved after it, and a holder that is gone left keys that no longer stand. + {"a want of an earlier holder, kept at the start and closed by the late read", "old-assigner_" + WantsState, wantKey, func(*Watchman) {}, true, true, true}, } { t.Run(row.name, func(t *testing.T) { w, o, c := setup() @@ -720,7 +727,7 @@ func TestAStuckKeyWhoseKeyIsGoneIsClosedAtTheStart(t *testing.T) { fk := FindingKey(ScopeModule, "gone", KindManifestRefused) o.findings[fk], _ = DecodeFinding(fv) w.SeedFinding(fk, fv) - w.Awaits(row.bucket) + w.AwaitsWants() w.Reconcile() _, open := o.findings[FindingKey(ScopeDelivery, sk, KindStuck)] _, seeded := o.findings[fk] @@ -728,7 +735,7 @@ func TestAStuckKeyWhoseKeyIsGoneIsClosedAtTheStart(t *testing.T) { t.Fatalf("the start's end took what an awaited bucket holds: moved %v, stuck open %v, finding %v, pulses deleted %v", o.stuck[sk].Moved, open, seeded, o.pulseDeletes) } row.standing(w) - w.ReadLate(row.bucket) + w.ReadLate(AssignerWantsBucket) // the bucket the holder followed now keeps if _, seeded := o.findings[fk]; seeded || !slices.Contains(o.pulseDeletes, pk) { t.Errorf("the late read kept the finding (%v) or the pulse (deleted %v) nothing made again", seeded, o.pulseDeletes) } @@ -756,3 +763,11 @@ func TestABusUserIsSaidByItsModuleWithoutAModulesNameInTheCode(t *testing.T) { } } } + +// AssignerWantsBucket is the bucket of the wants as the tests' holder of the seat mesh-wants keeps it: `_wants`, taken at each +// use through WantsAt, never a constant of the code (design 47 G A-321). +const AssignerWantsBucket = "acme-assigner_" + WantsState + +func testWants() string { return AssignerWantsBucket } + +func newWith(w *Watchman) *Watchman { w.WantsAt(testWants); return w } diff --git a/modules/mesh-delivery-watchman/manifest_test.go b/modules/mesh-delivery-watchman/manifest_test.go index f9f4ba2a..0bb88173 100644 --- a/modules/mesh-delivery-watchman/manifest_test.go +++ b/modules/mesh-delivery-watchman/manifest_test.go @@ -3,6 +3,7 @@ package manifest import ( "bytes" "encoding/json" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" "io/fs" "os" "path/filepath" @@ -126,11 +127,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_verdicts", "seat:mesh-hand-acts.hand-act", "seat:mesh-conditions.conditions", "seat:mesh-broker.show"}; !reflect.DeepEqual(m.Invokes, want) { + if want := []string{"mesh-delivery-indexer.indexer_pending", "mesh-delivery-indexer.indexer_verdicts", "seat:mesh-hand-acts.hand-act", "seat:mesh-conditions.conditions", "seat:mesh-broker.show"}; !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-catalog.manifest-accepted", "mesh-catalog.manifest-refused", "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-delivery-builder.wait-held", "mesh-delivery-builder.wait-ended", "node-engine.wait-held", "node-engine.wait-ended", "node-engine.running", "mesh-broker.created", "mesh-broker.refused", "mesh-broker.recreated", "mesh-broker.dropped", "mesh-conditions.findings-asked"}; !reflect.DeepEqual(m.Consumes, want) { + if want := []string{"seat:mesh-wants.views.wants", "seat:mesh-module-assigner.views.seat-holders", "mesh-catalog.registration-changed", "mesh-catalog.catalog-revision", "mesh-delivery-indexer.take-in.pending", "mesh-delivery-indexer.take-in.taken", "mesh-catalog.manifest-accepted", "mesh-catalog.manifest-refused", "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-delivery-builder.wait-held", "mesh-delivery-builder.wait-ended", "node-engine.wait-held", "node-engine.wait-ended", "node-engine.running", "mesh-broker.created", "mesh-broker.refused", "mesh-broker.recreated", "mesh-broker.dropped", "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). @@ -173,10 +174,9 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6 "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")}, + "wants": {meshwants.Seat + "_" + watch.WantsState, strings.TrimPrefix(bucket(m.Consumes, "seat:mesh-wants.views.wants"), "seat:")}, "provisions": {watch.ProvisionsBucket, bucket(m.Reads, "mesh-provisioner.provisions")}, - "assigner": {watch.AssignerModule + ".assigner_read_wants", entry(m.Invokes, "mesh-delivery-assigner.assigner_read_wants")}, - "indexer": {watch.IndexerModule + ".indexer_pending", entry(m.Invokes, "mesh-delivery-indexer.indexer_pending")}, + "indexer": {"mesh-delivery-indexer.indexer_pending", entry(m.Invokes, "mesh-delivery-indexer.indexer_pending")}, } { if row.got != row.want { t.Errorf("the name for %s is %q, its manifest's is %q", name, row.got, row.want) diff --git a/modules/mesh-delivery-watchman/module.json b/modules/mesh-delivery-watchman/module.json index c474595e..f95c7740 100644 --- a/modules/mesh-delivery-watchman/module.json +++ b/modules/mesh-delivery-watchman/module.json @@ -19,7 +19,6 @@ "mesh-provisioner.provisions" ], "invokes": [ - "mesh-delivery-assigner.assigner_read_wants", "mesh-delivery-indexer.indexer_pending", "mesh-delivery-indexer.indexer_verdicts", "seat:mesh-hand-acts.hand-act", @@ -27,7 +26,10 @@ "seat:mesh-broker.show" ], "consumes": [ - "mesh-delivery-assigner.views.wants", + "seat:mesh-wants.views.wants", + "seat:mesh-module-assigner.views.seat-holders", + "mesh-catalog.registration-changed", + "mesh-catalog.catalog-revision", "mesh-delivery-indexer.take-in.pending", "mesh-delivery-indexer.take-in.taken", "mesh-catalog.manifest-accepted",