novox/mesh-delivery
10
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Teller answers: the assigner follows the teller's views.answers, never its bucket

Browse Source 
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
This commit is contained in:
jschoubben
2026-11-12 06:10:47 +01:00
parent 298f26bb4c
commit d9dcb3ef36
6 changed files with 64 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go
+30 -9
View File
@@ -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()}
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go
+13 -4
View File
@@ -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
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go
+9
View File
@@ -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
}
modules/mesh-delivery-assigner/internal/buckets/buckets.go
+5 -2
View File
@@ -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, `<owner>_<state>`.
AnswersState = "answers"
AnswersReadTool = "teller_read_answers"
Answers = TellerModule + "_" + AnswersState
// Grants is the grants part's.
Grants = grantsstate.PartGrantsBucket
)
modules/mesh-delivery-assigner/manifest_test.go
+5 -3
View File
@@ -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, `<module>.<state>` as `<module>_<state>`
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
modules/mesh-delivery-assigner/module.json
+2 -1
View File
@@ -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",