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

Assigner: its own and the indexer's module names from their seats' clients, the teller's name beside the buckets, the delivery bound from motion, the provided modules from the overlay package, the build queue from buildstep (ADR 0634 §1, §4, §5; issue 1966)

Browse Source 
Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
This commit is contained in:
jschoubben
2026-11-05 16:06:38 +01:00
parent 463fa66aa1
commit 788ee56791
20 changed files with 78 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go
+20 -20
View File
@@ -16,6 +16,7 @@ import (
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"git.novox.be/novox/mesh-delivery/modules/mesh-grants/grantsstate"
"git.novox.be/novox/mesh-delivery/sdk/buildstep"
"io"
"maps"
"mesh-delivery-assigner/internal/buckets"
@@ -40,7 +41,6 @@ import (
"git.novox.be/novox/mesh-sdk/go/announce"
"git.novox.be/novox/mesh-sdk/go/buscompose"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/health"
"git.novox.be/novox/mesh-sdk/go/process"
"git.novox.be/novox/mesh-sdk/go/provisions"
@@ -122,7 +122,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 + "*", "mesh.mod." + delivery.IndexerModule + ".event.views.>"}, AckPolicy: jetstream.AckExplicitPolicy}); err != nil {
FilterSubjects: []string{enginereads.SubjectPrefix + "*", "mesh.mod." + meshcatalogclient.Module + ".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.
@@ -138,7 +138,7 @@ func startBed(t *testing.T, leaveOut string) *bed {
b.buckets[marksBucket] = kv
}
// The Go build seat's work queue, as the broker derives it from the seat's accepted kinds (delivery.AskStream).
seat, _ := delivery.AskStream(delivery.AskGo)
seat, _ := buildstep.AskStream(buildstep.AskGo)
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: seat, Subjects: []string{"mesh.seat.node-build-go.accept.>"},
Retention: jetstream.WorkQueuePolicy}); err != nil {
t.Fatal(err)
@@ -387,7 +387,7 @@ func TestWantsAreWrittenAndNoDifferenceIs(t *testing.T) {
firstViewBound, firstViewWait = 300*time.Millisecond, 50*time.Millisecond
t.Cleanup(func() { indexerLate, firstViewBound, firstViewWait = nil, 0, 0 })
for _, name := range append(append([]string{}, readBuckets...), watched...) {
if strings.HasPrefix(name, delivery.IndexerModule+"_") {
if strings.HasPrefix(name, meshcatalogclient.Module+"_") {
t.Errorf("the assigner reads the indexer's state %s itself", name)
}
}
@@ -427,10 +427,10 @@ func TestWantsAreWrittenAndNoDifferenceIs(t *testing.T) {
"seat-holders": {"n1.node-tools"},
"wants": {"n1.mesh-grants"},
})
graphKey := meshconditionsclient.Key(meshconditionsclient.ScopeModule, delivery.AssignerModule+"."+delivery.IndexerModule+"."+graph.State, views.UnansweredKind)
graphKey := meshconditionsclient.Key(meshconditionsclient.ScopeModule, mmac.Module+"."+meshcatalogclient.Module+"."+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")
return err == nil && strings.Contains(string(e.Value()), meshcatalogclient.Module+".graph")
}
for deadline := time.Now().Add(10 * time.Second); !graphRaised(); time.Sleep(50 * time.Millisecond) {
if time.Now().After(deadline) {
@@ -1092,7 +1092,7 @@ func (s bedStore) kv(state string) jetstream.KeyValue {
if strings.Contains(state, ".") {
return s.b.buckets[strings.Replace(state, ".", "_", 1)]
}
return s.b.buckets[delivery.AssignerModule+"_"+state]
return s.b.buckets[mmac.Module+"_"+state]
}
func (s bedStore) get(state, key string) ([]byte, uint64, bool, error) {
@@ -1127,7 +1127,7 @@ func (s bedStore) history(string, string) ([]stdio.StateRevision, error) { retur
// and declined, writes nothing and is withdrawn.
func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) {
b := startBed(t, "")
if _, err := b.js.CreateStream(b.ctx, jetstream.StreamConfig{Name: "MOD_" + delivery.AssignerModule + "_placement",
if _, err := b.js.CreateStream(b.ctx, jetstream.StreamConfig{Name: "MOD_" + mmac.Module + "_placement",
Subjects: placementSubjects()}); err != nil { // the seat's four events: the views family has a stream of its own
t.Fatal(err)
}
@@ -1137,7 +1137,7 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) {
t.Fatal(err)
}
t.Cleanup(nc.Close)
if _, err := nc.Subscribe(process.ModuleStreamSubject(delivery.AssignerModule, mma.EventPlacementChanged), func(m *nats.Msg) {
if _, err := nc.Subscribe(process.ModuleStreamSubject(mmac.Module, mma.EventPlacementChanged), func(m *nats.Msg) {
var e mma.PlacementEvent
if json.Unmarshal(m.Data, &e) == nil {
heard <- e
@@ -1379,8 +1379,8 @@ func (b *bed) viewsHold(t *testing.T, state, key string) {
t.Fatal(err)
}
for deadline := time.Now().Add(20 * time.Second); ; time.Sleep(50 * time.Millisecond) {
f := &views.Follower{Owner: delivery.AssignerModule, Tool: state, State: state, Ask: func(context.Context, string) (views.Answer, error) {
return views.FirstView(delivery.AssignerModule, state, src(kv), src(marks), "", time.Now())
f := &views.Follower{Owner: mmac.Module, Tool: state, State: state, Ask: func(context.Context, string) (views.Answer, error) {
return views.FirstView(mmac.Module, state, src(kv), src(marks), "", time.Now())
}}
if err := f.Start(b.ctx); err != nil {
t.Fatal(err)
@@ -1389,7 +1389,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 owner, _ := views.OwnerOfEvent(m.Subject); err != nil || owner != delivery.AssignerModule || views.StateOfEvent(m.Subject) != state {
if owner, _ := views.OwnerOfEvent(m.Subject); err != nil || owner != mmac.Module || views.StateOfEvent(m.Subject) != state {
continue
}
e, err := views.DecodeEvent(m.Data)
@@ -1447,7 +1447,7 @@ func (s bedSource) Get(key string) (*views.Entry, bool, error) {
func placementSubjects() []string {
var out []string
for _, e := range mma.Events() {
out = append(out, process.ModuleStreamSubject(delivery.AssignerModule, e))
out = append(out, process.ModuleStreamSubject(mmac.Module, e))
}
return out
}
@@ -1461,11 +1461,11 @@ var indexerLate 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: delivery.IndexerModule + "_" + views.MarksState, History: 1})
marks, err := b.js.CreateKeyValue(b.ctx, jetstream.KeyValueConfig{Bucket: meshcatalogclient.Module + "_" + views.MarksState, History: 1})
if err != nil {
t.Fatal(err)
}
em := views.Emitter{Owner: delivery.IndexerModule, Subject: process.Account{Node: "n1", Module: delivery.IndexerModule}.EventSubject,
em := views.Emitter{Owner: meshcatalogclient.Module, Subject: process.Account{Node: "n1", Module: meshcatalogclient.Module}.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})
@@ -1490,7 +1490,7 @@ func serveIndexer(t *testing.T, b *bed, nc *nats.Conn) {
After string `json:"after"`
}
_ = json.Unmarshal(m.Data, &args)
a, err := views.FirstView(delivery.IndexerModule, state, bedSource{b.ctx, kv}, bedSource{b.ctx, marks}, args.After, time.Now())
a, err := views.FirstView(meshcatalogclient.Module, 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()}
@@ -1550,7 +1550,7 @@ func firstViewsBeforeAnyNewFormState(t *testing.T, b *bed, granted map[string][]
outR, outW := io.Pipe()
t.Cleanup(func() { inW.Close(); outW.Close() })
go func() {
_ = stdio.ServeOn(delivery.AssignerModule, viewTools(func(k string) string { return env[k] }, state, time.Now), inR, outW)
_ = stdio.ServeOn(mmac.Module, viewTools(func(k string) string { return env[k] }, state, time.Now), inR, outW)
}()
answers := make(chan map[string]any, 16)
go func() {
@@ -1571,7 +1571,7 @@ func firstViewsBeforeAnyNewFormState(t *testing.T, b *bed, granted map[string][]
t.Fatal(err)
}
t.Cleanup(nc.Close)
sub, err := nc.Subscribe("mesh.mod."+delivery.AssignerModule+".call.*.*", func(msg *nats.Msg) {
sub, err := nc.Subscribe("mesh.mod."+mmac.Module+".call.*.*", func(msg *nats.Msg) {
one <- struct{}{}
defer func() { <-one }()
tokens := strings.Split(msg.Subject, ".")
@@ -1603,7 +1603,7 @@ func firstViewsBeforeAnyNewFormState(t *testing.T, b *bed, granted map[string][]
ask := func(user, tool string) (views.Answer, error) {
return views.AskVia(func(key string, body any) (json.RawMessage, error) {
data, _ := json.Marshal(body)
msg, err := nc.Request("mesh.mod."+delivery.AssignerModule+".call."+tool+"."+strings.ReplaceAll(user, ".", "~"), data, 5*time.Second)
msg, err := nc.Request("mesh.mod."+mmac.Module+".call."+tool+"."+strings.ReplaceAll(user, ".", "~"), data, 5*time.Second)
if err != nil {
return nil, err
}
@@ -1619,7 +1619,7 @@ func firstViewsBeforeAnyNewFormState(t *testing.T, b *bed, granted map[string][]
t.Errorf("%s's first view of %s before any new-form state: %v", user, tool, err)
case slices.Contains(allowed, user):
keys, _ := b.buckets[viewedStates[tool]].Keys(b.ctx)
if a.Owner != delivery.AssignerModule || a.State != tool || len(a.Values) != len(keys) {
if a.Owner != mmac.Module || a.State != tool || len(a.Values) != len(keys) {
t.Errorf("%s's first view of %s = %+v, the state holds %v", user, tool, a, keys)
}
case err == nil || !strings.Contains(err.Error(), "refused"):
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go
+4 -4
View File
@@ -10,6 +10,7 @@ import (
"context"
"errors"
"fmt"
mmac "git.novox.be/novox/mesh-contracts/placement/go/clients/meshmoduleassignerclient"
"io"
"log"
"mesh-delivery-assigner/internal/buckets"
@@ -26,7 +27,6 @@ import (
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
mma "git.novox.be/novox/mesh-contracts/placement/go/seats/meshmoduleassigner"
"git.novox.be/novox/mesh-sdk/go/announce"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/health"
"git.novox.be/novox/mesh-sdk/go/process"
"git.novox.be/novox/mesh-sdk/go/provisions"
@@ -85,7 +85,7 @@ var ownBuckets = []string{buckets.Asks, buckets.Differences,
const engineReadsState = "engine-reads"
// engineReadsBucket is that state's bucket.
const engineReadsBucket = delivery.AssignerModule + "_" + engineReadsState
const engineReadsBucket = mmac.Module + "_" + engineReadsState
// seatBuckets are the seat's states, new with it: until the controller has made them from the manifest and the bus grants
// them, the assigner starts without them and binds each once the bus says it was created (issue 1790's kind).
@@ -512,7 +512,7 @@ var PutsReopenEvery = time.Minute
// (ADR 0437 clause 9).
func sayVia(ctx context.Context, js jetstream.JetStream) func(oc.Condition) error {
return func(c oc.Condition) error {
subject, err := occ.ConditionSubject(delivery.AssignerModule)
subject, err := occ.ConditionSubject(mmac.Module)
if err != nil {
return err
}
@@ -529,7 +529,7 @@ func sayVia(ctx context.Context, js jetstream.JetStream) func(oc.Condition) erro
// consumer on EVENTS, and asks the node it names again; the hub also hands the indexer's views to the followers of
// indexerViews, and starts once both are registered.
func hearReadsChanged(ctx context.Context, js jetstream.JetStream, hub *process.EventHub, reads *enginereads.Keeper, findings assign.Findings, logf func(string, ...any)) {
account := process.Account{Node: thisNode, Module: delivery.AssignerModule}
account := process.Account{Node: thisNode, Module: mmac.Module}
w := notHeardWriter(findings, logf)
handle := func(m jetstream.Msg) {
if err := m.Ack(); err != nil {
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go
+5 -4
View File
@@ -10,6 +10,8 @@ import (
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks"
"git.novox.be/novox/mesh-delivery/modules/mesh-grants/grantsstate"
"git.novox.be/novox/mesh-networking/modules/mesh-wireguard/go/overlay"
"git.novox.be/novox/mesh-sdk/go/motion"
"maps"
"mesh-delivery-assigner/internal/buckets"
"slices"
@@ -18,7 +20,6 @@ import (
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"
"git.novox.be/novox/mesh-delivery/sdk/enginereads"
@@ -1008,7 +1009,7 @@ func (p *passer) provisionFindings(failed map[string]provisions.Provision) {
// MissingBound is how long a required provision may have no key before it is said: a module's delivery bound, the default
// when its manifest names none (ADR 0564 clause 13 (d)).
var MissingBound = delivery.DefaultDeliveryBound
var MissingBound = motion.DefaultDeliveryBound
// missingPulse is the pulse key of a provision with no key, `provision-missing.<node>.<module>.<provision>`.
func missingPulse(key string) string { return "provision-missing." + key }
@@ -1150,8 +1151,8 @@ func offered(manifests map[string]catalogue.Manifest) func(string) bool {
for _, m := range manifests {
add(m)
}
for _, module := range delivery.ProvidedModules() {
if m, _, err := delivery.ProvidedManifest(module); err == nil {
for _, module := range overlay.ProvidedModules() {
if m, _, err := overlay.ProvidedManifest(module); err == nil {
add(m)
}
}
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go
+1 -2
View File
@@ -31,7 +31,6 @@ import (
mma "git.novox.be/novox/mesh-contracts/placement/go/seats/meshmoduleassigner"
"git.novox.be/novox/mesh-sdk/go/busretry"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
"git.novox.be/novox/mesh-delivery/sdk/enginereads"
@@ -364,7 +363,7 @@ func putOfTheBus(t *testing.T) {
buildKV.values[bkey] = built
modules := mem()
modules.values["nats"] = mod
s := newPutTools(memStore{putsState: putsKV, builds.BuilderModule + ".builds": buildKV, delivery.IndexerModule + ".modules": modules})
s := newPutTools(memStore{putsState: putsKV, builds.BuilderModule + ".builds": buildKV, meshcatalogclient.Module + ".modules": modules})
s.now = func() time.Time { return now }
was := answerWithin
answerWithin = 0
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go
+1 -2
View File
@@ -21,7 +21,6 @@ import (
"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"
"mesh-delivery-assigner/internal/assign"
)
@@ -408,7 +407,7 @@ func (a *putActs) warrant(id, ask string) {
a.warranted[id] = true
a.mu.Unlock()
go func() {
if _, err := a.rec.RecordWarrant(delivery.TellerModule, ask); err != nil {
if _, err := a.rec.RecordWarrant(buckets.TellerModule, ask); err != nil {
a.logf("the put's warrant %s is not recorded in the hand-act log: %v", ask, err)
a.mu.Lock()
a.warrantNo[id] = fmt.Sprintf("the warrant %s is not recorded in the hand-act log: %v", ask, err)
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go
+3 -4
View File
@@ -26,7 +26,6 @@ import (
occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
mmac "git.novox.be/novox/mesh-contracts/placement/go/clients/meshmoduleassignerclient"
mma "git.novox.be/novox/mesh-contracts/placement/go/seats/meshmoduleassigner"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/process"
"mesh-delivery-assigner/internal/assign"
@@ -53,9 +52,9 @@ func streamPublisher(ctx context.Context, js jetstream.JetStream, node string) p
if err != nil {
return err
}
msg := nats.NewMsg(process.ModuleStreamSubject(delivery.AssignerModule, event))
msg := nats.NewMsg(process.ModuleStreamSubject(mmac.Module, event))
msg.Header.Set("x-event-id", id)
msg.Header.Set("x-source", delivery.AssignerModule)
msg.Header.Set("x-source", mmac.Module)
msg.Header.Set("x-node", node)
msg.Header.Set("x-time", time.Now().UTC().Format(time.RFC3339Nano))
msg.Header.Set("content-type", "application/json")
@@ -88,7 +87,7 @@ func (a *seatActs) record(warrant string) {
}
ask, _, _ := strings.Cut(warrant, "@")
go func() {
if taken, err := a.rec.RecordWarrant(delivery.TellerModule, ask); err != nil {
if taken, err := a.rec.RecordWarrant(buckets.TellerModule, ask); err != nil {
a.logf("the placement's warrant %s is not recorded in the hand-act log: %v", ask, err)
} else {
a.logf("the placement's warrant %s is recorded in the log of %s", ask, taken.Log)
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools.go
+1 -2
View File
@@ -15,7 +15,6 @@ import (
mmac "git.novox.be/novox/mesh-contracts/placement/go/clients/meshmoduleassignerclient"
mma "git.novox.be/novox/mesh-contracts/placement/go/seats/meshmoduleassigner"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/delivery"
"mesh-delivery-assigner/internal/assign"
plc "mesh-delivery-assigner/internal/placement"
@@ -43,7 +42,7 @@ func (runtimeStore) history(state, key string) ([]stdio.StateRevision, error) {
// The states the verbs read, as the runtime names them: the seat's own by their local names, the others'
// `<module>.<state>`.
const (
modulesState = delivery.IndexerModule + ".modules"
modulesState = meshcatalogclient.Module + ".modules"
haveState = "node-engine.have"
)
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go
+1 -2
View File
@@ -16,7 +16,6 @@ import (
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-delivery/sdk/enginereads"
)
@@ -77,7 +76,7 @@ func (s *putTools) version(module, version string) (string, error) {
v := strings.ToLower(strings.TrimSpace(version))
switch {
case v == "newest":
raw, _, found, err := s.st.get(delivery.IndexerModule+".modules", module)
raw, _, found, err := s.st.get(meshcatalogclient.ModulesRead, module)
if err != nil {
return "", fmt.Errorf("the indexer's modules cannot be read: %w", err)
}
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go
+14 -13
View File
@@ -3,6 +3,8 @@ package main
import (
"context"
"encoding/json"
mmac "git.novox.be/novox/mesh-contracts/placement/go/clients/meshmoduleassignerclient"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
"mesh-delivery-assigner/internal/buckets"
"slices"
"time"
@@ -12,7 +14,6 @@ import (
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient"
mma "git.novox.be/novox/mesh-contracts/placement/go/seats/meshmoduleassigner"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/health"
"git.novox.be/novox/mesh-sdk/go/process"
@@ -27,7 +28,7 @@ import (
)
// marksBucket is the assigner's state of what its views family said each state up to.
var marksBucket = delivery.AssignerModule + "_" + views.MarksState
var marksBucket = mmac.Module + "_" + views.MarksState
// viewedTool is one tool over one state of the assigner's.
type viewedTool struct{ tool, state, words string }
@@ -66,13 +67,13 @@ func viewTools(getenv func(string) string, state func(string) views.Source, now
marks := state(views.MarksState)
var out []stdio.Tool
for _, t := range firstViewTools {
out = append(out, views.FirstViewTool(delivery.AssignerModule, t.tool, t.state, t.words, views.ParseCallers(getenv(views.CallersEnv(t.tool))), state(t.state), marks, now))
out = append(out, views.FirstViewTool(mmac.Module, t.tool, t.state, t.words, views.ParseCallers(getenv(views.CallersEnv(t.tool))), state(t.state), marks, now))
}
for _, t := range readTools {
out = append(out, views.ReadTool(delivery.AssignerModule, t.tool, t.state, t.words, state(t.state), marks, now))
out = append(out, views.ReadTool(mmac.Module, 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), meshhandactsclient.Recorder{Ask: stdio.Ask}, now)...)
return append(out, views.StatusTools(mmac.Module, "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)} }
@@ -84,7 +85,7 @@ func startViews(ctx context.Context, js jetstream.JetStream, nc process.Subscrib
if findings.Put == nil {
return
}
if err := viewFindings(findings).Put(views.TooLargeFinding(delivery.AssignerModule, state, key, size, time.Now())); err != nil {
if err := viewFindings(findings).Put(views.TooLargeFinding(mmac.Module, state, key, size, time.Now())); err != nil {
logf("views: the finding of %s.%s too large is not written: %v", state, key, err)
}
}
@@ -109,7 +110,7 @@ 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.EventSubject, Publish: views.JetStreamPublish(js), Marks: mk, TooLarge: tooLarge, Logf: logf}
em := views.Emitter{Owner: mmac.Module, Subject: account.EventSubject, Publish: views.JetStreamPublish(js), Marks: mk, TooLarge: tooLarge, Logf: logf}
if findings.Put != nil && findings.Delete != nil {
em.Findings = viewFindings(findings)
}
@@ -122,7 +123,7 @@ func viewFindings(f assign.Findings) views.FindingWriter {
}
// firstViewsBucket is the assigner's own state of its first views (views.StatusState).
var firstViewsBucket = delivery.AssignerModule + "_" + views.StatusState
var firstViewsBucket = mmac.Module + "_" + views.StatusState
// indexerViews are the assigner's first views of the indexer's manifests, modules and module graph (ADR 0615 §1,
// ADR 0621 §2, ADR 0631; DLV-F2r): asked at the start through the indexer's tools, then followed on indexer-views, asking once more on a gap;
@@ -154,7 +155,7 @@ func (iv *indexerViews) start(ctx context.Context, w *world, wake func(), settle
if iv.findings.Put == nil {
return
}
f := views.UnansweredFinding(delivery.AssignerModule, s, time.Now())
f := views.UnansweredFinding(mmac.Module, s, time.Now())
var err error
if raised {
err = viewFindings(iv.findings).Put(f)
@@ -169,8 +170,8 @@ func (iv *indexerViews) start(ctx context.Context, w *world, wake func(), settle
var all []*views.Follower
for state, bucket := range indexerFirstViews {
bucket := bucket
tool := delivery.IndexerModule + "." + state
f := &views.Follower{Owner: delivery.IndexerModule, Tool: tool, State: state, Ask: views.AskVia(iv.call, tool), Logf: iv.logf,
tool := meshcatalogclient.Module + "." + state
f := &views.Follower{Owner: meshcatalogclient.Module, Tool: tool, State: state, Ask: views.AskVia(iv.call, tool), Logf: iv.logf,
Bound: firstViewBound, Wait: firstViewWait,
Raise: func(s views.Status) { finding(s, true) }, Clear: func(s views.Status) { finding(s, false) },
Changed: func(key string, value json.RawMessage, deleted bool, at time.Time) {
@@ -201,7 +202,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): the handler is
// registered on the module's hub, which the assigner starts once every handler is.
_ = notHeardWriter(iv.findings, iv.logf).Bind(process.EventsStream, "the indexer's manifests, modules and graph are the first views' alone until the assigner starts again", func() error {
return iv.hub.Handle("mesh.mod."+delivery.IndexerModule+".event."+views.EventPrefix+"*", views.Hear(ctx, followers, iv.logf))
return iv.hub.Handle("mesh.mod."+meshcatalogclient.Module+".event."+views.EventPrefix+"*", views.Hear(ctx, followers, iv.logf))
})
return len(all)
}
@@ -209,7 +210,7 @@ func (iv *indexerViews) start(ctx context.Context, w *world, wake func(), settle
// notHeardWriter writes the assigner's finding of a family it does not hear into its findings (heard.Writer); with no
// findings state it is said in the log alone.
func notHeardWriter(findings assign.Findings, logf func(string, ...any)) heard.Writer {
w := heard.Writer{Observer: delivery.AssignerModule, Logf: logf, Put: findings.Put}
w := heard.Writer{Observer: mmac.Module, Logf: logf, Put: findings.Put}
if findings.Delete != nil {
w.Delete = func(key string) error {
if keys, err := findings.Keys(); err == nil && !slices.Contains(keys, key) {
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/wants_tools.go
+1 -2
View File
@@ -18,7 +18,6 @@ import (
"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/health"
"mesh-delivery-assigner/internal/assign"
@@ -33,7 +32,7 @@ import (
// The states the verbs read, as the runtime names them.
const (
policiesState = buckets.PoliciesState
manifestsState = delivery.IndexerModule + ".manifests"
manifestsState = meshcatalogclient.Module + ".manifests"
// PolicyChangedEvent is the seat's event of a person's policy set or taken back.
PolicyChangedEvent = "policy-changed"
// busStepCause is the cause the bus's step is recorded under in the hand-act log.
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go
-5
View File
@@ -26,16 +26,11 @@ import (
mmac "git.novox.be/novox/mesh-contracts/placement/go/clients/meshmoduleassignerclient"
mma "git.novox.be/novox/mesh-contracts/placement/go/seats/meshmoduleassigner"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/delivery"
"mesh-delivery-assigner/internal/assign"
plc "mesh-delivery-assigner/internal/placement"
)
// noModuleWords is the indexer's reason for a module of a candidate whose directory the merge preview holds no manifest
// for (mesh-sdk delivery.CandidateNoneNoModule, which the indexer writes): the change removes the module.
const noModuleWords = delivery.CandidateNoneNoModule
// world is what the assigner watched, as it stands now: every value by bucket and key, and the manifests read by
// digest. The watches write it; a pass reads it.
type world struct {
modules/mesh-delivery-assigner/go.mod
+2
View File
@@ -67,3 +67,5 @@ require git.novox.be/novox/mesh-contracts/placement/go/clients/meshwantsclient v
require git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliverycheckerclient v0.0.0-00010101000000-000000000000
replace git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliverycheckerclient => ../../contracts/go/clients/meshdeliverycheckerclient
require git.novox.be/novox/mesh-networking/modules/mesh-wireguard/go v0.0.0-20261105000000-000000000000
modules/mesh-delivery-assigner/internal/assign/asks.go
+5 -5
View File
@@ -2,6 +2,7 @@ package assign
import (
"fmt"
mmac "git.novox.be/novox/mesh-contracts/placement/go/clients/meshmoduleassignerclient"
mwc "git.novox.be/novox/mesh-contracts/placement/go/clients/meshwantsclient"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient"
@@ -12,7 +13,6 @@ import (
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"
)
// HoldAsk is the ask about a module's fingerprint the assigner holds, at its key (asks.HoldAskKey): keep it held or
@@ -68,7 +68,7 @@ func RestoreAsk(key, node, module, fp, words string, now time.Time) asks.Ask {
// words, under the assigner's own name (ADR 0437 clause 9), keyed by the held fingerprint. needsYou is true for a module holding
// data, whose restore ask waits for the operator.
func HeldCondition(module, fp, node, words string, raised, needsYou bool) oc.Condition {
key := delivery.AssignerModule + ".held." + module + "." + strings.TrimPrefix(fp, meshcatalogclient.SourcePrefix2)
key := mmac.Module + ".held." + module + "." + strings.TrimPrefix(fp, meshcatalogclient.SourcePrefix2)
subject := oc.ConditionSubject{Scope: "module", ID: module, Machine: node}
if !raised {
return oc.Condition{Key: key, Subject: subject, State: occ.StateCleared}
@@ -91,7 +91,7 @@ func HeldCondition(module, fp, node, words string, raised, needsYou bool) oc.Con
// the node keeps its want. It names the node, the module, fp and the have's words, keyed by node and module, and is cleared
// (raised false) once the node no longer fails alone.
func NodeFailedCondition(node, module, fp string, f NodeFailure, raised bool) oc.Condition {
key := delivery.AssignerModule + ".failed-alone." + node + "." + module
key := mmac.Module + ".failed-alone." + node + "." + module
subject := oc.ConditionSubject{Scope: "module", ID: module, Machine: node}
if !raised {
return oc.Condition{Key: key, Subject: subject, State: occ.StateCleared,
@@ -114,7 +114,7 @@ func NodeFailedCondition(node, module, fp string, f NodeFailure, raised bool) oc
// naming the module, the node and the rule holding its first want back with the rule's words, under the assigner's own name keyed by
// node and module, and cleared (raised false) when the node has a want for it again.
func NeitherCondition(node, module string, h Holding, raised bool) oc.Condition {
key := delivery.AssignerModule + ".neither." + node + "." + module
key := mmac.Module + ".neither." + node + "." + module
subject := oc.ConditionSubject{Scope: "module", ID: module, Machine: node}
if !raised {
return oc.Condition{Key: key, Subject: subject, State: occ.StateCleared,
@@ -236,7 +236,7 @@ func BackupAsk(key, node, module, fp, digest, words string, asking int, now time
// BackupCondition is the condition raised when a backup before an apply fails and the module is held on the node (ADR 0539 §16),
// naming the node, the module and the backup's words; cleared under the same key.
func BackupCondition(node, module, fp, words string, raised bool) oc.Condition {
key := delivery.AssignerModule + ".backup." + node + "." + module + "." + strings.TrimPrefix(fp, meshcatalogclient.SourcePrefix2)
key := mmac.Module + ".backup." + node + "." + module + "." + strings.TrimPrefix(fp, meshcatalogclient.SourcePrefix2)
subject := oc.ConditionSubject{Scope: "module", ID: module, Machine: node}
if !raised {
return oc.Condition{Key: key, Subject: subject, State: occ.StateCleared}
modules/mesh-delivery-assigner/internal/assign/asks_test.go
+2 -2
View File
@@ -2,6 +2,7 @@ package assign
import (
"fmt"
mmac "git.novox.be/novox/mesh-contracts/placement/go/clients/meshmoduleassignerclient"
mwc "git.novox.be/novox/mesh-contracts/placement/go/clients/meshwantsclient"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient"
@@ -15,7 +16,6 @@ import (
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"
)
@@ -160,7 +160,7 @@ func saidConditionsAreFindings(t *testing.T) {
t.Fatalf("%s: %v", c.key, err)
}
got, ok := puts[c.key]
if !ok || got.Severity != health.Severity(c.raised.Severity) || got.Headline != c.raised.Headline || got.Observer != delivery.AssignerModule ||
if !ok || got.Severity != health.Severity(c.raised.Severity) || got.Headline != c.raised.Headline || got.Observer != mmac.Module ||
got.Subject.Machine != c.raised.Subject.Machine || said != 0 {
t.Fatalf("%s: finding %+v (written %v), said %d", c.key, got, puts, said)
}
modules/mesh-delivery-assigner/internal/assign/findings.go
+4 -4
View File
@@ -3,12 +3,12 @@ package assign
import (
"errors"
"fmt"
mmac "git.novox.be/novox/mesh-contracts/placement/go/clients/meshmoduleassignerclient"
"strings"
"time"
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"
)
@@ -20,13 +20,13 @@ import (
// assigner no longer says it straight (never both).
// FindingsBucket is the assigner's findings.
var FindingsBucket = health.FindingsBucket(delivery.AssignerModule)
var FindingsBucket = health.FindingsBucket(mmac.Module)
// FindingOf is a said condition as a finding: its key `<scope>.<id>.<kind>`, from the condition's own key
// `mesh-delivery-assigner.<kind>.<id…>` (the subject's id when the key names none), its words unchanged. raised is false for
// a clear, whose key is to be deleted; value is nil then.
func FindingOf(c oc.Condition, at time.Time) (key string, value []byte, raised bool, err error) {
rest, ok := strings.CutPrefix(c.Key, delivery.AssignerModule+".")
rest, ok := strings.CutPrefix(c.Key, mmac.Module+".")
if !ok || rest == "" {
return "", nil, false, fmt.Errorf("the condition %q is not the assigner's", c.Key)
}
@@ -56,7 +56,7 @@ func FindingOf(c oc.Condition, at time.Time) (key string, value []byte, raised b
value, err = health.EncodeFinding(health.Finding{Key: key, Kind: kind, Subject: health.Subject{Scope: scope, ID: id, Machine: c.Subject.Machine},
Severity: health.Severity(c.Severity), Summary: summary, Headline: c.Headline, Explanation: c.Explanation, Resolved: c.Resolved,
Needs: c.Needs, Actions: actions, Evidence: []health.Evidence{{At: at.UTC(), Said: summary}}, Source: "assigner." + kind,
Observer: delivery.AssignerModule, At: at.UTC()})
Observer: mmac.Module, At: at.UTC()})
return key, value, true, err
}
modules/mesh-delivery-assigner/internal/assign/placement.go
+3 -3
View File
@@ -1,18 +1,18 @@
package assign
import (
mmac "git.novox.be/novox/mesh-contracts/placement/go/clients/meshmoduleassignerclient"
"strings"
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"
)
// SeatWaitsCondition is the assigner's word on the seat mesh-placement while a state of the seat is not bound (ADR 0567
// §1): raised naming each, cleared when none is left.
func SeatWaitsCondition(unbound []string) oc.Condition {
key := delivery.AssignerModule + ".seat-states"
subject := oc.ConditionSubject{Scope: "module", ID: delivery.AssignerModule}
key := mmac.Module + ".seat-states"
subject := oc.ConditionSubject{Scope: "module", ID: mmac.Module}
if len(unbound) == 0 {
return oc.Condition{Key: key, Subject: subject, State: occ.StateCleared, Resolved: "the seat mesh-placement holds all its states."}
}
modules/mesh-delivery-assigner/internal/assign/policy.go
+5 -5
View File
@@ -3,6 +3,7 @@ package assign
import (
"encoding/json"
"fmt"
mmac "git.novox.be/novox/mesh-contracts/placement/go/clients/meshmoduleassignerclient"
"mesh-delivery-assigner/internal/buckets"
"sort"
"strings"
@@ -10,7 +11,6 @@ import (
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"
"git.novox.be/novox/mesh-sdk/go/provisions"
)
@@ -20,7 +20,7 @@ import (
// catalogue.Upgrade: a person's choice over the module's word, its irreplaceable data and the default, the bus as any module (ADR 0601 §1).
// PoliciesBucket is the assigner's state `policies`.
var PoliciesBucket = delivery.AssignerModule + "_" + buckets.PoliciesState
var PoliciesBucket = mmac.Module + "_" + buckets.PoliciesState
// DecodePolicies reads the policies state's values by key (the module); a value that cannot be read is left out, named.
func DecodePolicies(values map[string][]byte) (map[string]catalogue.PersonPolicy, []string) {
@@ -73,7 +73,7 @@ func ProvisionFailedFinding(key string, p provisions.Provision, at time.Time) ([
Severity: health.Warning, Summary: summary, Headline: clipHeadline(p.Consumer.Module + " waits for a provision that failed"),
Explanation: occ.NothingToDo + " " + summary + ". It clears once the provider provides it.",
Resolved: p.Provision + " for " + p.Consumer.Module + " is provided", Evidence: []health.Evidence{{At: at.UTC(), Said: p.Why}},
Source: "assigner." + KindProvisionFailed, Observer: delivery.AssignerModule, At: at.UTC()})
Source: "assigner." + KindProvisionFailed, Observer: mmac.Module, At: at.UTC()})
}
func also(provider, consumer string) []string {
@@ -105,7 +105,7 @@ func ProvisionMissingFinding(key string, bound time.Duration, at time.Time) ([]b
Severity: health.Warning, Summary: summary, Headline: clipHeadline(module + " on " + node + " lacks " + provision),
Explanation: occ.NothingToDo + " " + summary + ". It clears once the provision has its key.",
Resolved: provision + " for " + module + " on " + node + " has its key", Evidence: []health.Evidence{{At: at.UTC(), Said: summary}},
Source: "assigner." + KindProvisionMissing, Observer: delivery.AssignerModule, At: at.UTC()})
Source: "assigner." + KindProvisionMissing, Observer: mmac.Module, At: at.UTC()})
}
// KindUnplaced is, from the lease write, a node that runs but no placement names (ADR 0571 §6).
@@ -125,7 +125,7 @@ func UnplacedFinding(node string, at time.Time) ([]byte, error) {
Headline: clipHeadline(node + " has no placement"), Needs: "place it with mesh-placement.propose.",
Explanation: occ.NeedsYou + "place it with mesh-placement.propose. " + summary + ".",
Resolved: node + " is placed", Evidence: []health.Evidence{{At: at.UTC(), Said: summary}},
Source: "assigner." + KindUnplaced, Observer: delivery.AssignerModule, At: at.UTC()})
Source: "assigner." + KindUnplaced, Observer: mmac.Module, At: at.UTC()})
}
func clipHeadline(s string) string {
modules/mesh-delivery-assigner/internal/assign/seat.go
+1 -2
View File
@@ -13,7 +13,6 @@ import (
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
mmac "git.novox.be/novox/mesh-contracts/placement/go/clients/meshmoduleassignerclient"
mma "git.novox.be/novox/mesh-contracts/placement/go/seats/meshmoduleassigner"
"git.novox.be/novox/mesh-sdk/go/delivery"
plc "mesh-delivery-assigner/internal/placement"
)
@@ -178,7 +177,7 @@ const (
// CallClearedCondition clears the condition of a refused controller call a previous process raised, for its pair (node and
// module, or node and provision for a pin).
func CallClearedCondition(kind, node, module string) oc.Condition {
return oc.Condition{Key: delivery.AssignerModule + "." + kind + "." + node + "." + module,
return oc.Condition{Key: mmac.Module + "." + kind + "." + node + "." + module,
Subject: oc.ConditionSubject{Scope: "module", ID: module, Machine: node}, State: occ.StateCleared,
Resolved: "the assigner calls no verb of the controller any more: placement is the one source."}
}
modules/mesh-delivery-assigner/internal/buckets/buckets.go
+4 -1
View File
@@ -44,8 +44,11 @@ const (
// Source and Merges are the gate's.
Source = "mesh-delivery-gate_" + delivery.StateSource
Merges = "mesh-delivery-gate_" + delivery.StateMerges
// 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 = "mesh-delivery-teller_answers"
Answers = TellerModule + "_answers"
// Builds is the builder's.
Builds = builds.BuilderBuildsBucket
// Grants is the grants part's.
modules/mesh-delivery-assigner/internal/buckets/buckets_test.go
+1 -1
View File
@@ -13,7 +13,7 @@ func TestBucketsAreTheEarlierNames(t *testing.T) {
"Builds": {Builds, "mesh-delivery-builder_builds"}, "Grants": {Grants, "mesh-grants_grants"},
"Wants": {Wants, "mesh-delivery-assigner_wants"}, "Differences": {Differences, "mesh-delivery-assigner_differences"},
"Asks": {Asks, "mesh-delivery-assigner_asks"}, "Puts": {Puts, "mesh-delivery-assigner_puts"},
"Policies": {Policies, "mesh-delivery-assigner_policies"},
"Policies": {Policies, "mesh-delivery-assigner_policies"}, "TellerModule": {TellerModule, "mesh-delivery-teller"},
} {
if c[0] != c[1] {
t.Errorf("%s = %q, want %q", name, c[0], c[1])