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

Assigner: its views family is its ordinary events (views.<state> 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)

Browse Source 
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
This commit is contained in:
jschoubben
2026-11-05 15:08:25 +01:00
parent a1b22505e1
commit 2e8b4dcdcd
23 changed files with 107 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go
+10 -18
View File
@@ -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})
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go
+2 -2
View File
@@ -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"
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go
+1 -1
View File
@@ -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"
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go
+1 -1
View File
@@ -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"
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go
+2 -2
View File
@@ -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"
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go
+4 -4
View File
@@ -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
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go
+5 -5
View File
@@ -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
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools.go
+8 -14
View File
@@ -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 <seat>.<verb> (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
}
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go
+2 -3
View File
@@ -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)
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go
+17 -28
View File
@@ -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 {
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/wants_tools.go
+13 -11
View File
@@ -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 <seat>.<verb>, with the seat's own schemas.
// tools are the seat's verbs as the runtime serves them, each named <seat>.<verb>; 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
}
modules/mesh-delivery-assigner/go.mod
+6 -2
View File
@@ -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
modules/mesh-delivery-assigner/internal/assign/askjson.go
+2 -2
View File
@@ -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"
)
modules/mesh-delivery-assigner/internal/assign/asks.go
+2 -2
View File
@@ -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"
)
modules/mesh-delivery-assigner/internal/assign/asks_test.go
+2 -2
View File
@@ -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"
)
modules/mesh-delivery-assigner/internal/assign/findings.go
+2 -2
View File
@@ -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"
)
modules/mesh-delivery-assigner/internal/assign/placement.go
+2 -2
View File
@@ -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"
)
modules/mesh-delivery-assigner/internal/assign/policy.go
+1 -1
View File
@@ -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"
modules/mesh-delivery-assigner/internal/assign/put.go
+2 -2
View File
@@ -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"
)
modules/mesh-delivery-assigner/internal/assign/seat.go
+2 -2
View File
@@ -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"
)
modules/mesh-delivery-assigner/internal/assign/seat_test.go
+1 -1
View File
@@ -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"
)
modules/mesh-delivery-assigner/manifest_test.go
+11 -15
View File
@@ -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.<state>` 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)
}
}
modules/mesh-delivery-assigner/module.json
+9 -4
View File
@@ -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"