From 2e8b4dcdcd896360ab0e4a7cf2f76f8200e29340 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 5 Nov 2026 15:08:25 +0100 Subject: [PATCH] Assigner: its views family is its ordinary events (views. among its emits, no stream of its own), the indexer's followed on the shared events stream, its findings through the views' writer; the seat mesh-module-assigner and mesh-wants served by the contracts' verb names, described by the tool runner from the seat list; operator-channel, placement and hand-act names from the contracts and clients (D-25 owner half, D-12, D-24; issue 1966) The assigner's two consumers of its durable on the shared stream still compete; they join the module's one hub. Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i --- .../integration_test.go | 28 +++++------- .../cmd/mesh-delivery-assigner/main.go | 4 +- .../cmd/mesh-delivery-assigner/main_test.go | 2 +- .../cmd/mesh-delivery-assigner/pass.go | 2 +- .../cmd/mesh-delivery-assigner/pass_test.go | 4 +- .../cmd/mesh-delivery-assigner/puts.go | 8 ++-- .../cmd/mesh-delivery-assigner/seat.go | 10 ++--- .../cmd/mesh-delivery-assigner/seat_tools.go | 22 ++++----- .../mesh-delivery-assigner/seat_tools_test.go | 5 +-- .../cmd/mesh-delivery-assigner/views.go | 45 +++++++------------ .../cmd/mesh-delivery-assigner/wants_tools.go | 24 +++++----- modules/mesh-delivery-assigner/go.mod | 8 +++- .../internal/assign/askjson.go | 4 +- .../internal/assign/asks.go | 4 +- .../internal/assign/asks_test.go | 4 +- .../internal/assign/findings.go | 4 +- .../internal/assign/placement.go | 4 +- .../internal/assign/policy.go | 2 +- .../internal/assign/put.go | 4 +- .../internal/assign/seat.go | 4 +- .../internal/assign/seat_test.go | 2 +- .../mesh-delivery-assigner/manifest_test.go | 26 +++++------ modules/mesh-delivery-assigner/module.json | 13 ++++-- 23 files changed, 107 insertions(+), 126 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 525135b3..b2d6f4e4 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 @@ -21,8 +21,8 @@ import ( "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" - occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient" - oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel" + occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" + oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/announce" "git.novox.be/novox/mesh-sdk/go/buscompose" @@ -34,6 +34,7 @@ import ( "mesh-delivery-assigner/internal/judge" "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" + "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" "git.novox.be/novox/mesh-delivery/sdk/enginereads" "git.novox.be/novox/mesh-delivery/sdk/graph" "git.novox.be/novox/mesh-sdk/go/views" @@ -107,7 +108,7 @@ 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 + "*"}, AckPolicy: jetstream.AckExplicitPolicy}); err != nil { + FilterSubjects: []string{enginereads.SubjectPrefix + "*", "mesh.mod." + delivery.IndexerModule + ".event.views.>"}, AckPolicy: jetstream.AckExplicitPolicy}); err != nil { t.Fatal(err) } // The operator-channel seat's stream of conditions, which the assigner says its conditions on. @@ -116,10 +117,7 @@ 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 assigner's views family and its marks (ADR 0615 §5, ADR 0619 §2), on its old sections. - if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "ASSIGNER_VIEWS", Subjects: []string{"mesh.stream.mesh-delivery-assigner.event.views.>"}}); err != nil { - t.Fatal(err) - } + // The assigner's views family is its ordinary events on EVENTS (ADR 0635 §5); its marks are its state. if kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: marksBucket, History: 1}); err != nil { t.Fatal(err) } else { @@ -415,7 +413,7 @@ func TestWantsAreWrittenAndNoDifferenceIs(t *testing.T) { "seat-holders": {"n1.node-tools"}, "wants": {"n1.mesh-grants"}, }) - graphKey, _, _ := views.UnansweredFinding(delivery.AssignerModule, views.Status{Owner: delivery.IndexerModule, State: graph.State}, time.Now()) + graphKey := meshconditionsclient.Key(meshconditionsclient.ScopeModule, delivery.AssignerModule+"."+delivery.IndexerModule+"."+graph.State, views.UnansweredKind) graphRaised := func() bool { e, err := b.buckets[assign.FindingsBucket].Get(b.ctx, graphKey) return err == nil && strings.Contains(string(e.Value()), delivery.IndexerModule+".graph") @@ -1353,7 +1351,7 @@ func (b *bed) viewsHold(t *testing.T, state, key string) { t.Helper() kv, marks := b.buckets[viewedStates[state]], b.buckets[marksBucket] src := func(kv jetstream.KeyValue) views.Source { return bedSource{b.ctx, kv} } - st, err := b.js.Stream(b.ctx, "ASSIGNER_VIEWS") + st, err := b.js.Stream(b.ctx, "EVENTS") if err != nil { t.Fatal(err) } @@ -1368,7 +1366,7 @@ func (b *bed) viewsHold(t *testing.T, state, key string) { gap := false for seq := info.State.FirstSeq; seq > 0 && seq <= info.State.LastSeq; seq++ { m, err := st.GetMsg(b.ctx, seq) - if err != nil || views.StateOfEvent(m.Subject) != state { + if owner, _ := views.OwnerOfEvent(m.Subject); err != nil || owner != delivery.AssignerModule || views.StateOfEvent(m.Subject) != state { continue } e, err := views.DecodeEvent(m.Data) @@ -1436,7 +1434,7 @@ func placementSubjects() []string { var indexerLate chan struct{} // serveIndexer is the indexer as the owner of its manifests, modules and module graph (ADR 0615, ADR 0621, ADR 0631): its -// buckets, which a test fills as the indexer writes them; its views family said by the views emitter onto INDEXER_VIEWS, +// buckets, which a test fills as the indexer writes them; its views family said by the views emitter as ordinary events on EVENTS, // 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() @@ -1444,13 +1442,7 @@ func serveIndexer(t *testing.T, b *bed, nc *nats.Conn) { if err != nil { t.Fatal(err) } - if _, err := b.js.CreateStream(b.ctx, jetstream.StreamConfig{Name: "INDEXER_VIEWS", Subjects: []string{"mesh.stream.mesh-delivery-indexer.event.views.>"}}); err != nil { - t.Fatal(err) - } - if _, err := b.js.CreateOrUpdateConsumer(b.ctx, "INDEXER_VIEWS", jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-assigner", AckPolicy: jetstream.AckExplicitPolicy}); err != nil { - t.Fatal(err) - } - em := views.Emitter{Owner: delivery.IndexerModule, Subject: func(e string) string { return process.ModuleStreamSubject(delivery.IndexerModule, e) }, + em := views.Emitter{Owner: delivery.IndexerModule, Subject: process.Account{Node: "n1", Module: delivery.IndexerModule}.EventSubject, Publish: views.JetStreamPublish(b.js), Marks: marks} for state, bucket := range indexerFirstViews { kv, err := b.js.CreateKeyValue(b.ctx, jetstream.KeyValueConfig{Bucket: bucket, History: 5}) 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 9e558938..384b382e 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -21,8 +21,8 @@ import ( "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" - occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient" - oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel" + occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" + oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" "git.novox.be/novox/mesh-sdk/go/announce" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/health" diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go index 584b29b2..1741bc0c 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go @@ -18,7 +18,7 @@ import ( "github.com/nats-io/nats.go/jetstream" - oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel" + oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/process" diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go index 2fb7907d..4d7e54dd 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -10,7 +10,7 @@ import ( "strings" "time" - oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel" + oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/provisions" diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go index 6ad7309d..f1b0dbfa 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go @@ -13,8 +13,8 @@ import ( "github.com/nats-io/nats.go/jetstream" - occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient" - oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel" + occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" + oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" "git.novox.be/novox/mesh-sdk/go/busretry" "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go index a89fb78a..63ad8026 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go @@ -10,10 +10,10 @@ import ( "sync" "time" - occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient" + "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" + occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" - "git.novox.be/novox/mesh-sdk/go/handacts" "mesh-delivery-assigner/internal/assign" ) @@ -48,7 +48,7 @@ type putActs struct { snapNo map[string]string warranted map[string]bool // rec records the put's warrant in the hand-act log (ADR 0565 §11). - rec handacts.Recorder + rec meshhandactsclient.Recorder warrantNo map[string]string said map[string]string // snapshot takes the bus's snapshot on node; a variable so a test takes none. @@ -56,7 +56,7 @@ type putActs struct { } func newPutActs(bus caller, wake func(), logf func(string, ...any)) *putActs { - a := &putActs{rec: handacts.Recorder{Ask: bus.Call}, bus: bus, wake: wake, logf: logf, snapping: map[string]bool{}, snapped: map[string]string{}, snapNo: map[string]string{}, warranted: map[string]bool{}, warrantNo: map[string]string{}, + a := &putActs{rec: meshhandactsclient.Recorder{Ask: bus.Call}, bus: bus, wake: wake, logf: logf, snapping: map[string]bool{}, snapped: map[string]string{}, snapNo: map[string]string{}, warranted: map[string]bool{}, warrantNo: map[string]string{}, said: map[string]string{}} a.snapshot = a.snapshotTheBus return a diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go index 77ffeb92..96f14228 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go @@ -15,9 +15,9 @@ import ( "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" - occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient" + "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" + occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" "git.novox.be/novox/mesh-sdk/go/delivery" - "git.novox.be/novox/mesh-sdk/go/handacts" "git.novox.be/novox/mesh-sdk/go/process" "mesh-delivery-assigner/internal/assign" @@ -62,16 +62,16 @@ type seatActs struct { wake func() logf func(string, ...any) // rec records the seat's acts in the hand-act log (ADR 0565 §11). - rec handacts.Recorder + rec meshhandactsclient.Recorder } func newSeatActs(bus caller, wake func(), logf func(string, ...any)) *seatActs { - return &seatActs{bus: bus, wake: wake, logf: logf, rec: handacts.Recorder{Ask: bus.Call}} + return &seatActs{bus: bus, wake: wake, logf: logf, rec: meshhandactsclient.Recorder{Ask: bus.Call}} } // record records once, in the hand-act log, that placement was written on the operator's warrant (§5): the router's // record of the teller's ask says who chose and how (ADR 0274). The log is the seat mesh-hand-acts' from its delivery, and -// the controller's while that seat has no holder (ADR 0565 §11, handacts.Recorder). +// the controller's while that seat has no holder (ADR 0565 §11, meshhandactsclient.Recorder). func (a *seatActs) record(warrant string) { if a == nil || warrant == "" { return diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools.go index 030cea1d..5168d686 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools.go @@ -4,13 +4,13 @@ import ( "encoding/json" "errors" "fmt" + "git.novox.be/novox/mesh-contracts/placement/go/seats/meshmoduleassigner" "slices" "sort" "strings" "time" stdio "git.novox.be/novox/mesh-sdk/go" - "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" "mesh-delivery-assigner/internal/assign" @@ -473,22 +473,16 @@ func (s *seatTools) doProposals(args map[string]any) (any, error) { // tools are the seat's verbs as the runtime serves them, each named . (the tool runner serves a seat's verb // only from a tool of that name), with the seat's own schemas. func (s *seatTools) tools() []stdio.Tool { - seat, _ := catalogue.SeatNamed(catalogue.PlacementSeat) - input := map[string]map[string]any{} - description := map[string]string{} - for _, v := range seat.Serves { - input[v.Name], description[v.Name] = v.Input, v.Description - } run := map[string]func(map[string]any) (any, error){ - catalogue.PlacementVerbPlacement: s.doPlacement, - catalogue.PlacementVerbProposals: s.doProposals, - catalogue.PlacementVerbPropose: func(a map[string]any) (any, error) { return s.doPropose(a, stdio.CallerOf(a)) }, - catalogue.PlacementVerbWithdraw: func(a map[string]any) (any, error) { return s.doWithdraw(a, stdio.CallerOf(a)) }, - catalogue.PlacementVerbAgain: s.doAgain, + meshmoduleassigner.VerbPlacement: s.doPlacement, + meshmoduleassigner.VerbProposals: s.doProposals, + meshmoduleassigner.VerbPropose: func(a map[string]any) (any, error) { return s.doPropose(a, stdio.CallerOf(a)) }, + meshmoduleassigner.VerbWithdraw: func(a map[string]any) (any, error) { return s.doWithdraw(a, stdio.CallerOf(a)) }, + meshmoduleassigner.VerbAgain: s.doAgain, } var out []stdio.Tool - for _, v := range seat.Serves { - out = append(out, stdio.Tool{Name: catalogue.PlacementSeat + "." + v.Name, Description: description[v.Name], Input: input[v.Name], Run: run[v.Name]}) + for _, v := range meshmoduleassigner.Verbs() { + out = append(out, stdio.Tool{Name: meshmoduleassigner.Seat + "." + v, Run: run[v]}) } return out } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go index 0506e404..2efb95dc 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go @@ -127,8 +127,7 @@ func TestTheSeatsVerbs(t *testing.T) { } } for _, stream := range raw0.Streams { - // The views family is no seat's stream: the readers of the assigner's states follow it (ADR 0615 §5). - if stream.Name != viewsStream && !reads["stream "+stream.Name] { + if !reads["stream "+stream.Name] { t.Errorf("no verb reads the seat's stream %s", stream.Name) } } @@ -182,7 +181,7 @@ func TestTheSeatsVerbs(t *testing.T) { t.Errorf("the remove-node proposed %+v", rm) } // 1126/B8: the seat serves its five verbs; adopt and hand-back went with the controller (ADR 0590). - if tools := s.tools(); len(tools) != 5 || tools[2].Name != "mesh-placement.propose" || tools[2].Run == nil || tools[2].Input == nil { + if tools := s.tools(); len(tools) != 5 || tools[2].Name != "mesh-module-assigner.propose" || tools[2].Run == nil { t.Errorf("the seat's verbs as served: %+v", tools) } t.Run("1953/D4-D6: the seat mesh-wants' upgrade and bus", wantsVerbs) 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 ca7ae4f1..9de44e35 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go @@ -9,28 +9,21 @@ import ( "github.com/nats-io/nats.go/jetstream" + "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" stdio "git.novox.be/novox/mesh-sdk/go" - "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" - "git.novox.be/novox/mesh-sdk/go/handacts" "git.novox.be/novox/mesh-sdk/go/health" "git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-sdk/go/views" + sdkfindings "git.novox.be/novox/mesh-delivery/sdk/findings" "git.novox.be/novox/mesh-delivery/sdk/graph" "git.novox.be/novox/mesh-delivery/sdk/heard" "mesh-delivery-assigner/internal/assign" ) -// viewsStream is the assigner's views family on its old streams section, beside placement (ADR 0619 §2, ADR 0615 §5), -// as its manifest's `streams` names it; indexerViewsStream is the indexer's, as the indexer's manifest names it. -const ( - viewsStream = "assigner-views" - indexerViewsStream = "indexer-views" -) - // marksBucket is the assigner's state of what its views family said each state up to. var marksBucket = delivery.AssignerModule + "_" + views.MarksState @@ -77,7 +70,7 @@ func viewTools(getenv func(string) string, state func(string) views.Source, now out = append(out, views.ReadTool(delivery.AssignerModule, t.tool, t.state, t.words, state(t.state), marks, now)) } // Its own first views, of the indexer's manifests, modules and graph (DLV-F3r, ADR 0631 §2): read, and asked again by hand. - return append(out, views.StatusTools(delivery.AssignerModule, "assigner_first_views", "assigner_first_view_again", stdio.State(views.StatusState), handacts.Recorder{Ask: stdio.Ask}, now)...) + return append(out, views.StatusTools(delivery.AssignerModule, "assigner_first_views", "assigner_first_view_again", stdio.State(views.StatusState), meshhandactsclient.Recorder{Ask: stdio.Ask}, now)...) } func keptSource(name string) views.Source { return views.KeptSource{State: stdio.State(name)} } @@ -89,11 +82,7 @@ func startViews(ctx context.Context, js jetstream.JetStream, nc process.Subscrib if findings.Put == nil { return } - k, v, err := views.TooLargeFinding(delivery.AssignerModule, state, key, size, time.Now()) - if err == nil { - err = findings.Put(k, v) - } - if err != nil { + if err := viewFindings(findings).Put(views.TooLargeFinding(delivery.AssignerModule, state, key, size, time.Now())); err != nil { logf("views: the finding of %s.%s too large is not written: %v", state, key, err) } } @@ -118,18 +107,17 @@ func startViews(ctx context.Context, js jetstream.JetStream, nc process.Subscrib // viewsEmitter is the assigner's emitter of its views family: a change not said past views.Bound is its finding // view-unsaid, in its findings (review round 2); with no findings state it is said in the log only. func viewsEmitter(js jetstream.JetStream, account process.Account, mk jetstream.KeyValue, findings assign.Findings, tooLarge func(string, string, int), logf func(string, ...any)) views.Emitter { - em := views.Emitter{Owner: delivery.AssignerModule, Subject: account.StreamSubject, Publish: views.JetStreamPublish(js), Marks: mk, TooLarge: tooLarge, Logf: logf} + em := views.Emitter{Owner: delivery.AssignerModule, Subject: account.EventSubject, Publish: views.JetStreamPublish(js), Marks: mk, TooLarge: tooLarge, Logf: logf} if findings.Put != nil && findings.Delete != nil { - em.Findings = assignerFindings{findings} + em.Findings = viewFindings(findings) } return em } -// assignerFindings is the assigner's findings as views.Emitter's Findings. -type assignerFindings struct{ f assign.Findings } - -func (a assignerFindings) Put(key string, value []byte) error { return a.f.Put(key, value) } -func (a assignerFindings) Delete(key string) error { return a.f.Delete(key) } +// viewFindings is the assigner's findings as the views' FindingWriter, in the seat mesh-conditions' layout. +func viewFindings(f assign.Findings) views.FindingWriter { + return sdkfindings.WriterOver(f.Put, f.Delete) +} // firstViewsBucket is the assigner's own state of its first views (views.StatusState). var firstViewsBucket = delivery.AssignerModule + "_" + views.StatusState @@ -163,11 +151,12 @@ func (iv *indexerViews) start(ctx context.Context, w *world, wake func(), settle if iv.findings.Put == nil { return } - key, value, err := views.UnansweredFinding(delivery.AssignerModule, s, time.Now()) - if err == nil && raised { - err = iv.findings.Put(key, value) - } else if err == nil { - err = iv.findings.Delete(key) + f := views.UnansweredFinding(delivery.AssignerModule, s, time.Now()) + var err error + if raised { + err = viewFindings(iv.findings).Put(f) + } else { + err = viewFindings(iv.findings).Delete(f) } if err != nil { iv.logf("views: the finding of the first view %s is not written: %v", s.Tool, err) @@ -207,7 +196,7 @@ func (iv *indexerViews) start(ctx context.Context, w *world, wake func(), settle }) } // Not heard, the indexer's views family is the finding family-not-heard until it binds (ADR 0615 §5). - stream := catalogue.StreamBusName(indexerViewsStream) + stream := process.EventsStream _ = notHeardWriter(iv.findings, iv.logf).Bind(stream, "the indexer's manifests, modules and graph are the first views' alone until the assigner starts again", func() error { cons, err := iv.account.EventConsumer(ctx, iv.js, stream) if err != nil { diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/wants_tools.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/wants_tools.go index d0c4092c..6a8925ba 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/wants_tools.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/wants_tools.go @@ -4,14 +4,16 @@ import ( "encoding/json" "errors" "fmt" + meshhandacts "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshhandacts" + "git.novox.be/novox/mesh-contracts/placement/go/seats/meshwants" "sort" "strconv" "time" + "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" - "git.novox.be/novox/mesh-sdk/go/handacts" "git.novox.be/novox/mesh-sdk/go/health" "mesh-delivery-assigner/internal/assign" @@ -47,7 +49,7 @@ type wantsTools struct { // wantsEventSubject is the subject of one of the seat's events. func wantsEventSubject(event string) string { - return "mesh.seat." + catalogue.WantsSeat + ".event." + event + return "mesh.seat." + meshwants.Seat + ".event." + event } func newWantsTools(st seatStore, ask func(string, any) (json.RawMessage, error), emit func(event, id string, body any) error) *wantsTools { @@ -226,8 +228,8 @@ func (s *wantsTools) setPolicy(module, policy, why, by string, manifests map[str return nil, fmt.Errorf("the choice for %s cannot be written: %w", module, err) } } - rec := handacts.Recorder{Ask: s.ask} - taken, rerr := rec.Record(handacts.Request{What: what, Why: why, Cause: "upgrade-policy", Verb: "mesh-wants.upgrade"}) + rec := meshhandactsclient.Recorder{Ask: s.ask} + taken, rerr := rec.Record(meshhandactsclient.Request{What: what, Why: why, Cause: "upgrade-policy", Verb: "mesh-wants.upgrade"}) recorded := taken.Log if rerr != nil { recorded = "not recorded: " + rerr.Error() @@ -285,7 +287,7 @@ func (s *wantsTools) doBus(map[string]any) (any, error) { } sort.Slice(nodes, func(i, j int) bool { return nodes[i].Node < nodes[j].Node }) out := map[string]any{"module": bus, "newest": newest, "nodes": nodes} - if steps, err := s.ask("seat:"+handacts.Seat+".hand-acts", map[string]any{"cause": busStepCause, "days": 90}); err != nil { + if steps, err := s.ask("seat:"+meshhandacts.Seat+".hand-acts", map[string]any{"cause": busStepCause, "days": 90}); err != nil { out["steps-unread"] = err.Error() } else { out["steps"] = steps @@ -305,16 +307,16 @@ func (s *wantsTools) doBus(map[string]any) (any, error) { return out, nil } -// tools are the seat's verbs as the runtime serves them, each named ., with the seat's own schemas. +// tools are the seat's verbs as the runtime serves them, each named .; the tool runner describes them with the +// seat's definition from the seat list (ADR 0635 §1), so the tools carry none. func (s *wantsTools) tools() []stdio.Tool { - seat, _ := catalogue.SeatNamed(catalogue.WantsSeat) run := map[string]func(map[string]any) (any, error){ - "upgrade": func(a map[string]any) (any, error) { return s.doUpgrade(a, stdio.CallerOf(a)) }, - "bus": s.doBus, + meshwants.VerbUpgrade: func(a map[string]any) (any, error) { return s.doUpgrade(a, stdio.CallerOf(a)) }, + meshwants.VerbBus: s.doBus, } var out []stdio.Tool - for _, v := range seat.Serves { - out = append(out, stdio.Tool{Name: catalogue.WantsSeat + "." + v.Name, Description: v.Description, Input: v.Input, Run: run[v.Name]}) + for _, v := range meshwants.Verbs() { + out = append(out, stdio.Tool{Name: meshwants.Seat + "." + v, Run: run[v]}) } return out } diff --git a/modules/mesh-delivery-assigner/go.mod b/modules/mesh-delivery-assigner/go.mod index 83b11eac..d04550c6 100644 --- a/modules/mesh-delivery-assigner/go.mod +++ b/modules/mesh-delivery-assigner/go.mod @@ -3,6 +3,10 @@ module mesh-delivery-assigner go 1.26.0 require ( + git.novox.be/novox/mesh-contracts/placement/go/seats/meshmoduleassigner v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-contracts/placement/go/seats/meshwants v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshhandacts v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261104174359-05b3e2e1b94f github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 @@ -26,9 +30,9 @@ require git.novox.be/novox/mesh-delivery/sdk v0.0.0-00010101000000-000000000000 require ( git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient v0.0.0-20261104174344-ae5266e77639 - git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient v0.0.0-20261104174344-ae5266e77639 + git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient v0.0.0-20261104174344-ae5266e77639 git.novox.be/novox/mesh-contracts/go/seats/nodeengine v0.0.0-20261104174344-ae5266e77639 - git.novox.be/novox/mesh-contracts/go/seats/operatorchannel v0.0.0-20261104174344-ae5266e77639 + git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel v0.0.0-20261104174344-ae5266e77639 ) replace git.novox.be/novox/mesh-delivery/sdk => ../../sdk diff --git a/modules/mesh-delivery-assigner/internal/assign/askjson.go b/modules/mesh-delivery-assigner/internal/assign/askjson.go index ce80db6d..d3b0c594 100644 --- a/modules/mesh-delivery-assigner/internal/assign/askjson.go +++ b/modules/mesh-delivery-assigner/internal/assign/askjson.go @@ -5,8 +5,8 @@ import ( "fmt" "time" - occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient" - oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel" + occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" + oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" "git.novox.be/novox/mesh-sdk/go/delivery" ) diff --git a/modules/mesh-delivery-assigner/internal/assign/asks.go b/modules/mesh-delivery-assigner/internal/assign/asks.go index c7fd22d6..fa286477 100644 --- a/modules/mesh-delivery-assigner/internal/assign/asks.go +++ b/modules/mesh-delivery-assigner/internal/assign/asks.go @@ -6,8 +6,8 @@ import ( "strings" "time" - occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient" - oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel" + occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" + oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" "git.novox.be/novox/mesh-sdk/go/delivery" ) diff --git a/modules/mesh-delivery-assigner/internal/assign/asks_test.go b/modules/mesh-delivery-assigner/internal/assign/asks_test.go index 0c5b5f20..28aec5cb 100644 --- a/modules/mesh-delivery-assigner/internal/assign/asks_test.go +++ b/modules/mesh-delivery-assigner/internal/assign/asks_test.go @@ -7,8 +7,8 @@ import ( "testing" "time" - occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient" - oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel" + occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" + oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/health" ) diff --git a/modules/mesh-delivery-assigner/internal/assign/findings.go b/modules/mesh-delivery-assigner/internal/assign/findings.go index bc20d30f..254c38e8 100644 --- a/modules/mesh-delivery-assigner/internal/assign/findings.go +++ b/modules/mesh-delivery-assigner/internal/assign/findings.go @@ -6,8 +6,8 @@ import ( "strings" "time" - occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient" - oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel" + occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" + oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/health" ) diff --git a/modules/mesh-delivery-assigner/internal/assign/placement.go b/modules/mesh-delivery-assigner/internal/assign/placement.go index d6dc84d5..d0f920da 100644 --- a/modules/mesh-delivery-assigner/internal/assign/placement.go +++ b/modules/mesh-delivery-assigner/internal/assign/placement.go @@ -3,8 +3,8 @@ package assign import ( "strings" - occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient" - oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel" + occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" + oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" "git.novox.be/novox/mesh-sdk/go/delivery" ) diff --git a/modules/mesh-delivery-assigner/internal/assign/policy.go b/modules/mesh-delivery-assigner/internal/assign/policy.go index a4ce3cd1..2723ac92 100644 --- a/modules/mesh-delivery-assigner/internal/assign/policy.go +++ b/modules/mesh-delivery-assigner/internal/assign/policy.go @@ -7,7 +7,7 @@ import ( "strings" "time" - occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient" + occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/health" diff --git a/modules/mesh-delivery-assigner/internal/assign/put.go b/modules/mesh-delivery-assigner/internal/assign/put.go index 2c99034f..06a36f47 100644 --- a/modules/mesh-delivery-assigner/internal/assign/put.go +++ b/modules/mesh-delivery-assigner/internal/assign/put.go @@ -4,8 +4,8 @@ import ( "fmt" "time" - occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient" - oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel" + occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" + oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" "git.novox.be/novox/mesh-sdk/go/delivery" ) diff --git a/modules/mesh-delivery-assigner/internal/assign/seat.go b/modules/mesh-delivery-assigner/internal/assign/seat.go index 3dd39970..2c6099a2 100644 --- a/modules/mesh-delivery-assigner/internal/assign/seat.go +++ b/modules/mesh-delivery-assigner/internal/assign/seat.go @@ -7,8 +7,8 @@ import ( "strings" "time" - occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient" - oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel" + occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" + oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" "git.novox.be/novox/mesh-sdk/go/delivery" ) diff --git a/modules/mesh-delivery-assigner/internal/assign/seat_test.go b/modules/mesh-delivery-assigner/internal/assign/seat_test.go index 8e6c9d71..92e54691 100644 --- a/modules/mesh-delivery-assigner/internal/assign/seat_test.go +++ b/modules/mesh-delivery-assigner/internal/assign/seat_test.go @@ -7,7 +7,7 @@ import ( "testing" "time" - occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient" + occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" "git.novox.be/novox/mesh-sdk/go/delivery" ) diff --git a/modules/mesh-delivery-assigner/manifest_test.go b/modules/mesh-delivery-assigner/manifest_test.go index 0b616aa5..48de9857 100644 --- a/modules/mesh-delivery-assigner/manifest_test.go +++ b/modules/mesh-delivery-assigner/manifest_test.go @@ -1,6 +1,8 @@ package manifest import ( + "git.novox.be/novox/mesh-contracts/placement/go/seats/meshmoduleassigner" + "git.novox.be/novox/mesh-contracts/placement/go/seats/meshwants" "os" "reflect" "slices" @@ -71,17 +73,12 @@ func TestManifest(t *testing.T) { // controller, ADR 0590); keeps the seat's events on its stream placement for 30 days; calls the hand-act log; and reads // mesh-config's resolved values to compose every node a proposal touches. // ADR 0565 §7, §9 (issue 1953 D4): and the seat mesh-wants, serving upgrade and bus. - if len(m.Claims) != 2 || m.Claims[0].Name != catalogue.PlacementSeat || m.Claims[0].At() != catalogue.ScopeMesh || + if len(m.Claims) != 2 || m.Claims[0].Name != meshmoduleassigner.Seat || m.Claims[0].At() != catalogue.ScopeMesh || !slices.Equal(m.Claims[0].Serves, []string{"placement", "proposals", "propose", "withdraw", "again"}) || - m.Claims[1].Name != catalogue.WantsSeat || m.Claims[1].At() != catalogue.ScopeMesh || !slices.Equal(m.Claims[1].Serves, []string{"upgrade", "bus"}) { - t.Errorf("the assigner claims %+v, want mesh-placement with its five verbs and mesh-wants with its two", m.Claims) - } - if seat, _ := catalogue.SeatNamed(catalogue.WantsSeat); catalogue.CanHold(m, seat) != nil { - t.Errorf("the assigner cannot hold mesh-wants: %v", catalogue.CanHold(m, seat)) - } - if seat, _ := catalogue.SeatNamed(catalogue.PlacementSeat); catalogue.CanHold(m, seat) != nil { - t.Errorf("the assigner cannot hold the seat: %v", catalogue.CanHold(m, seat)) + m.Claims[1].Name != meshwants.Seat || m.Claims[1].At() != catalogue.ScopeMesh || !slices.Equal(m.Claims[1].Serves, []string{"upgrade", "bus"}) { + t.Errorf("the assigner claims %+v, want mesh-module-assigner with its five verbs and mesh-wants with its two", m.Claims) } + // Whether the seats' definitions let it hold them is the manifest validator's (the judge, at the seat list). // ADR 0621 §1: the first-view tools placement, seat-holders and wants and design 25 § 12.2's six read tools, of the // program's bundle, which runs in a runtime of its own on the assigner's account; ADR 0618 §3: and assigner_engine_reads, // which lists and asks again what each node's engine reads. @@ -101,10 +98,9 @@ func TestManifest(t *testing.T) { t.Errorf("%s callers = %q, want %q", tool, env[views.CallersEnv(tool)], list) } } - // ADR 0619 §2, ADR 0615 §5: its views family beside placement on its old streams section. - if len(m.Streams) != 2 || m.Streams[1].Name != "assigner-views" || !slices.Equal(m.Streams[1].Events, []string{views.EventPattern}) || - m.Streams[1].MaxAgeDays != views.StreamDays || m.Streams[1].MaxMiB != views.StreamMiB || m.Streams[1].Retention != "" { - t.Errorf("streams = %+v, want placement and assigner-views", m.Streams) + // ADR 0635 §5: its views family is its ordinary events, `views.` among its emits, on no stream of its own. + if len(m.Streams) != 1 { + t.Errorf("streams = %+v, want placement alone", m.Streams) } if m.Streams[0].Name != delivery.PlacementStream || m.Streams[0].MaxAgeDays != delivery.PlacementStreamDays || !slices.Equal(slices.Sorted(slices.Values(m.Streams[0].Events)), slices.Sorted(slices.Values(delivery.PlacementEvents))) { @@ -120,7 +116,7 @@ func TestManifest(t *testing.T) { "seat:node-engine.manifest-reads"}; !slices.Equal(slices.Sorted(slices.Values(m.Invokes)), want) { // ADR 0618 §3 t.Errorf("invokes = %v, want %v", m.Invokes, want) } - if !slices.Equal(m.Emits, []string{"put"}) { - t.Errorf("emits = %v, want put", m.Emits) + if want := []string{"put", "views.asks", "views.differences", "views.findings", "views.placement", "views.seat-holders", "views.wants"}; !slices.Equal(m.Emits, want) { + t.Errorf("emits = %v, want put and its six views", m.Emits) } } diff --git a/modules/mesh-delivery-assigner/module.json b/modules/mesh-delivery-assigner/module.json index 97a163ae..8a2ecdb2 100644 --- a/modules/mesh-delivery-assigner/module.json +++ b/modules/mesh-delivery-assigner/module.json @@ -5,7 +5,7 @@ "runs-as": "mesh-delivery-assigner", "claims": [ { - "name": "mesh-placement", + "name": "mesh-module-assigner", "scope": "mesh", "serves": [ "placement", @@ -56,7 +56,13 @@ "seat:node-engine.manifest-reads" ], "emits": [ - "put" + "put", + "views.asks", + "views.differences", + "views.findings", + "views.placement", + "views.seat-holders", + "views.wants" ], "tools": [ "assigner_put", @@ -101,8 +107,7 @@ "placement-withdrawn" ], "max-age-days": 30 - }, - { "name": "assigner-views", "events": ["views.**"], "max-age-days": 7, "max-mib": 256 } + } ], "own-secrets": { "broker": "${dir:state}/broker"