From d9dcb3ef36503d40d96f3d092a6892afc3604ccf Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 06:10:47 +0100 Subject: [PATCH] Teller answers: the assigner follows the teller's views.answers, never its bucket The assigner's read of mesh-delivery-teller_answers goes: its process reader follows the teller's answers by their first view (teller_read_answers) and the views family views.answers, and the pass watches that picture as it watched the bucket, without gating the first pass on it. Builds: design 47 G A-175, A-178 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../integration_test.go | 39 ++++++++++++++----- .../cmd/mesh-delivery-assigner/main.go | 17 ++++++-- .../cmd/mesh-delivery-assigner/views.go | 9 +++++ .../internal/buckets/buckets.go | 7 +++- .../mesh-delivery-assigner/manifest_test.go | 8 ++-- modules/mesh-delivery-assigner/module.json | 3 +- 6 files changed, 64 insertions(+), 19 deletions(-) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go index 849d0215..1606a79c 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go @@ -204,7 +204,8 @@ func startBed(t *testing.T, leaveOut string) *bed { // event names the engine's machine last, so the filter is the event with every machine (mesh-grants buscompose // machineEvent; the grants part's own test holds the derivation). if _, err := js.CreateOrUpdateConsumer(ctx, "EVENTS", jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-assigner", - FilterSubjects: []string{enginereads.SubjectPrefix + "*", "mesh.mod." + buckets.IndexerModule + ".event.views.>", meshcatalog.SubjectCatalogRevisionAll}, AckPolicy: jetstream.AckExplicitPolicy}); err != nil { + FilterSubjects: []string{enginereads.SubjectPrefix + "*", "mesh.mod." + buckets.IndexerModule + ".event.views.>", + "mesh.mod." + buckets.TellerModule + ".event.views.>", meshcatalog.SubjectCatalogRevisionAll}, AckPolicy: jetstream.AckExplicitPolicy}); err != nil { t.Fatal(err) } // The operator-channel seat's stream of conditions, which the assigner says its conditions on. @@ -213,6 +214,8 @@ func startBed(t *testing.T, leaveOut string) *bed { } // The indexer as the owner of its manifests and modules (DLV-F2r), with the real owner's side. serveIndexer(t, b, nc) + // The teller as the owner of its answers (A-175). + serveTeller(t, b, nc) // The seat mesh-catalog's catalog half (design 47 G §194), which the assigner follows per branch. serveCatalog(t, b, nc) // The assigner's views family is its ordinary events on EVENTS (ADR 0635 §5); its marks are its state. @@ -1708,13 +1711,32 @@ var catalogLate chan struct{} // read by the assigner's durable consumer; and its first-view tools of the three answered from the buckets at their marks. func serveIndexer(t *testing.T, b *bed, nc *nats.Conn) { t.Helper() - marks, err := b.js.CreateKeyValue(b.ctx, jetstream.KeyValueConfig{Bucket: buckets.IndexerModule + "_" + views.MarksState, History: 1}) + tools := map[string][2]string{} + for state, bucket := range indexerFirstViews { + tools[state] = [2]string{state, bucket} + } + serveOwner(t, b, nc, buckets.IndexerModule, tools, catalogLate) +} + +// serveTeller is the teller as the owner of its answers (design 47 G A-175, A-178): its bucket, which a test fills as the +// teller writes it, said by its views family, and its read tool teller_read_answers answering the first view. +func serveTeller(t *testing.T, b *bed, nc *nats.Conn) { + t.Helper() + serveOwner(t, b, nc, buckets.TellerModule, map[string][2]string{buckets.AnswersReadTool: {buckets.AnswersState, buckets.Answers}}, nil) +} + +// serveOwner serves an owner's states as the owner does: each state's bucket, its views family on EVENTS, and the first-view +// tool of each (by tool, its state and bucket) answered from the bucket at its mark. While late is open every ask fails. +func serveOwner(t *testing.T, b *bed, nc *nats.Conn, owner string, tools map[string][2]string, late chan struct{}) { + t.Helper() + marks, err := b.js.CreateKeyValue(b.ctx, jetstream.KeyValueConfig{Bucket: owner + "_" + views.MarksState, History: 1}) if err != nil { t.Fatal(err) } - em := views.Emitter{Owner: buckets.IndexerModule, Subject: process.Account{Node: "n1", Module: buckets.IndexerModule}.EventSubject, + em := views.Emitter{Owner: owner, Subject: process.Account{Node: "n1", Module: owner}.EventSubject, Publish: views.JetStreamPublish(b.js), Marks: marks} - for state, bucket := range indexerFirstViews { + for tool, sb := range tools { + state, bucket := sb[0], sb[1] kv, err := b.js.CreateKeyValue(b.ctx, jetstream.KeyValueConfig{Bucket: bucket, History: 5}) if err != nil { t.Fatal(err) @@ -1723,13 +1745,12 @@ func serveIndexer(t *testing.T, b *bed, nc *nats.Conn) { if err := em.Start(b.ctx, state, kv, nil); err != nil { t.Fatal(err) } - late := catalogLate - sub, err := nc.Subscribe("mesh.mod.mesh-delivery-indexer.call."+state+".>", func(m *nats.Msg) { - if late != nil { // its runtime comes up after the assigner started (DLV-F5r): until then every ask fails, as on a down indexer + sub, err := nc.Subscribe("mesh.mod."+owner+".call."+tool+".>", func(m *nats.Msg) { + if late != nil { // its runtime comes up after the assigner started (DLV-F5r): until then every ask fails, as on a down owner select { case <-late: default: - _ = m.Respond([]byte(`{"error":"the indexer's runtime is not up"}`)) + _ = m.Respond([]byte(`{"error":"the owner's runtime is not up"}`)) return } } @@ -1737,7 +1758,7 @@ func serveIndexer(t *testing.T, b *bed, nc *nats.Conn) { After string `json:"after"` } _ = json.Unmarshal(m.Data, &args) - a, err := views.FirstView(buckets.IndexerModule, state, bedSource{b.ctx, kv}, bedSource{b.ctx, marks}, args.After, time.Now()) + a, err := views.FirstView(owner, state, bedSource{b.ctx, kv}, bedSource{b.ctx, marks}, args.After, time.Now()) out := map[string]any{"result": a, "node": "n0"} if err != nil { out = map[string]any{"error": err.Error()} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go index fc61ab40..976b5bfe 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -62,7 +62,6 @@ func main() { // holds the two lists equal), so the controller's broker grants exactly these. var readBuckets = []string{ buckets.Grants, - buckets.Answers, buckets.Presence, buckets.Have, buckets.Provisions, @@ -324,11 +323,13 @@ type emitting struct{ *process.Conn } func (e emitting) Emit(eventType string, body any) error { return e.Conn.Emit(eventType, body) } +// optionalEnds is what stops when the watch of a bucket that never gates the first pass ends. +var optionalEnds = map[string]string{buckets.Puts: "no put is taken further", buckets.Answers: "no ask is answered"} + // watched are the buckets whose every value a pass reads; the manifests are read by digest when a pass needs one, and // watched only for keys, so a manifest that arrives wakes a composition that waits for it. var watched = []string{ buckets.Grants, - buckets.Answers, plc.AssignmentsBucket, plc.ProposalsBucket, plc.CandidatesBucket, @@ -378,7 +379,7 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, return fmt.Errorf("the bucket %s cannot be watched: %w", name, err) } // The puts' watch never stops the wants (ADR 0546): when it ends, the puts are no longer heard, said in the journal. - optional := name == buckets.Puts + optional := name == buckets.Puts || name == buckets.Answers go func() { defer kw.Stop() for { @@ -389,7 +390,7 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, case e, ok := <-kw.Updates(): if !ok { if optional { - logger.Printf("the watch of %s ended: no put is taken further until the assigner starts again", name) + logger.Printf("the watch of %s ended: %s until the assigner starts again", name, optionalEnds[name]) return } errs <- fmt.Errorf("the watch of %s ended", name) @@ -478,6 +479,14 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, if iv != nil { iv.hub = hub started += iv.start(ctx, w, wake, settled) + // The teller's answers' picture is watched as a bucket is, but never gates the first pass: a teller that does not + // answer its first view only leaves the asks unanswered, said by the reader's finding (A-175). + if iv.answers != nil { + kvs[buckets.Answers] = iv.answers + if err := watch(buckets.Answers); err != nil { + logger.Printf("%v: no ask is answered until the assigner starts again", err) + } + } } // The seat's states arrive with the same delivery as this assigner (ADR 0567 §1): one not on the bus at this start is // bound once the bus says it was created, and watched from then on, with no restart (issue 1790's kind). Until then no diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go index 77d5144b..8391bf3f 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go @@ -122,6 +122,9 @@ type catalogViews struct { hub *process.EventHub h *halves + // answers is the picture of the teller's answers (design 47 G A-175, A-178): its first view through teller_read_answers, + // then its views family; nil when it is not followed. + answers *views.Picture } // firstViewBound and firstViewWait are each first view's bound before its condition is raised, and its first wait @@ -139,6 +142,12 @@ func (iv *catalogViews) start(ctx context.Context, w *world, wake func(), settle if err := iv.h.follow(); err != nil { iv.logf("the catalog half is not followed, so no module is placed until the assigner starts again: %v", err) } + // The teller's answers to the assigner's asks, followed as every owner's state is (A-175), never its bucket. + if pic, err := reader.Follow(buckets.TellerModule, buckets.AnswersState, buckets.TellerModule+"."+buckets.AnswersReadTool); err != nil { + iv.logf("the teller's answers are not followed, so no ask is answered until the assigner starts again: %v", err) + } else { + iv.answers = pic + } reader.Start(ctx, iv.js, iv.account, iv.status) return 1 } diff --git a/modules/mesh-delivery-assigner/internal/buckets/buckets.go b/modules/mesh-delivery-assigner/internal/buckets/buckets.go index 9f6842bc..4711f9b1 100644 --- a/modules/mesh-delivery-assigner/internal/buckets/buckets.go +++ b/modules/mesh-delivery-assigner/internal/buckets/buckets.go @@ -63,8 +63,11 @@ const ( // TellerModule is the module whose hand the operator's answers pass: the one that holds each ask's answer, and the // name the hand-act log records a warrant under. Held equal to the manifest's reads by its test. TellerModule = "mesh-delivery-teller" - // Answers holds the teller's answer to each ask. - Answers = TellerModule + "_answers" + // AnswersState is the teller's state of its answer to each ask, followed through its first view by AnswersReadTool and its + // views family `views.answers` (design 47 G A-175, A-178), never its bucket; Answers is the picture's name, `_`. + AnswersState = "answers" + AnswersReadTool = "teller_read_answers" + Answers = TellerModule + "_" + AnswersState // Grants is the grants part's. Grants = grantsstate.PartGrantsBucket ) diff --git a/modules/mesh-delivery-assigner/manifest_test.go b/modules/mesh-delivery-assigner/manifest_test.go index fcd4d943..ac4a2561 100644 --- a/modules/mesh-delivery-assigner/manifest_test.go +++ b/modules/mesh-delivery-assigner/manifest_test.go @@ -48,13 +48,14 @@ func TestManifest(t *testing.T) { // on, ADR 0614 §2, §3) are read through its first views and views family, none of the indexer's states by reads. // A-181: what was built is heard from the builder's stream of ends, so the builder's builds bucket is no read. // §431, §507: the gate's source is read through its verb `source` and its event `source-moved`, and no `merges` is read. - want := []string{"mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-grants.grants", "mesh-host.composed", + // A-175, A-178: the teller's answers are followed through its first view and its views family, never its bucket. + want := []string{"mesh-delivery-watchman.presence", "mesh-grants.grants", "mesh-host.composed", "mesh-provisioner.provisions", "node-engine.have"} if !reflect.DeepEqual(reads, want) { t.Errorf("reads = %v, want exactly %v", reads, want) } // the buckets the program names are the manifest's reads, `.` as `_` - named := []string{buckets.Answers, buckets.Presence, buckets.Grants, + named := []string{buckets.Presence, buckets.Grants, buckets.ComposedRead, "mesh-provisioner_provisions", buckets.Have} slices.Sort(named) for i, r := range want { @@ -131,10 +132,11 @@ func TestManifest(t *testing.T) { } // ADR 0618 §4: and it hears each node's engine say its reads changed; ADR 0636 §4 (issue 1966): and the seat list's two events; // issue 1966 note 151: and the teller's ask-ended, its subject the one the assigner hears. - if !slices.Equal(m.Consumes, []string{"mesh-conditions.findings-asked", "mesh-conditions.condition-raised", "mesh-conditions.condition-changed", "mesh-conditions.condition-cleared", "mesh-catalog.catalog-revision", "mesh-catalog.registration-changed", "mesh-delivery.source-moved", "mesh-catalog.version-released", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", builds.BuilderModule + "." + builds.BuildsStreamEvents, buckets.TellerModule + "." + tellerevents.AskEnded, "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", "node-engine.manifest-reads-changed", "node-engine.running", "mesh-delivery.pinned", "mesh-delivery.freed", "mesh-delivery.heads-replaced", "mesh-delivery.pins-stated"}) { + if !slices.Equal(m.Consumes, []string{"mesh-conditions.findings-asked", "mesh-conditions.condition-raised", "mesh-conditions.condition-changed", "mesh-conditions.condition-cleared", "mesh-catalog.catalog-revision", "mesh-catalog.registration-changed", "mesh-delivery.source-moved", "mesh-catalog.version-released", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", builds.BuilderModule + "." + builds.BuildsStreamEvents, buckets.TellerModule + "." + tellerevents.AskEnded, buckets.TellerModule + ".views." + buckets.AnswersState, "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", "node-engine.manifest-reads-changed", "node-engine.running", "mesh-delivery.pinned", "mesh-delivery.freed", "mesh-delivery.heads-replaced", "mesh-delivery.pins-stated"}) { t.Errorf("consumes = %v, want the catalog half's revisions (design 47 G §195), the teller's ask-ended, the engines' reads changed and what they run", m.Consumes) } if want := []string{"mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", + buckets.TellerModule + "." + buckets.AnswersReadTool, // A-175, A-178: the teller's answers' first view "seat:mesh-catalog.registrations", "seat:mesh-catalog.revision", "seat:mesh-catalog.seat-definitions", // ADR 0636 §4: the seat list "seat:mesh-conditions.conditions", // design 47 G §706: the conditions read at each judging of an outside wait "seat:mesh-delivery.source", // §507: the branches the gate keeps a source revision of, before the first registrations diff --git a/modules/mesh-delivery-assigner/module.json b/modules/mesh-delivery-assigner/module.json index c909550f..36762452 100644 --- a/modules/mesh-delivery-assigner/module.json +++ b/modules/mesh-delivery-assigner/module.json @@ -30,7 +30,6 @@ } ], "reads": [ - "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-grants.grants", "mesh-host.composed", @@ -53,6 +52,7 @@ "mesh-catalog.seat-definition-removed", "mesh-delivery-builder.build.ended.**", "mesh-delivery-teller.ask-ended", + "mesh-delivery-teller.views.answers", "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", "node-engine.manifest-reads-changed", @@ -65,6 +65,7 @@ "invokes": [ "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", + "mesh-delivery-teller.teller_read_answers", "seat:mesh-catalog.registrations", "seat:mesh-catalog.revision", "seat:mesh-catalog.seat-definitions",