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

Assigner: placements, proposals, candidates and seat holders are the contract's types and the assigner's own placement package, no longer go/delivery's (D-12; ADR 0634 §1; issue 1966)

Browse Source 
The judge keeps the node's answer as the verdict's body; pointer times of the contract are guarded. One ask test folded into a table to hold the unit count.

Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
This commit is contained in:
jschoubben
2026-11-05 15:31:50 +01:00
parent e75cb6e02e
commit 1266c1ada0
14 changed files with 455 additions and 285 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go
+44 -33
View File
@@ -23,6 +23,8 @@ 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"
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/announce"
"git.novox.be/novox/mesh-sdk/go/buscompose"
@@ -40,6 +42,7 @@ import (
"git.novox.be/novox/mesh-sdk/go/views"
"mesh-delivery-assigner/internal/assign"
plc "mesh-delivery-assigner/internal/placement"
)
// bed is an embedded bus with JetStream, every bucket the assigner reads and keeps, and the stream the mesh's events
@@ -274,7 +277,7 @@ func TestTheAssignerDoesNotStartWithoutABucketItReads(t *testing.T) {
if err != nil {
t.Fatal(err)
}
pl, _ := delivery.EncodePlacement(delivery.Placement{Node: "n1", Modules: []delivery.PlacedModule{{Module: "web"}}})
pl, _ := mmac.EncodePlacement(mma.Placement{Node: "n1", Modules: []mma.PlacedModule{{Module: "web"}}})
plkey, _ := delivery.LegacyPlacementKey("n1") // the form before, read beside the new (research 072 row 37)
if _, err := placement.Put(pb.ctx, plkey, pl); err != nil {
t.Fatal(err)
@@ -302,11 +305,11 @@ func (b *bed) place(t *testing.T, node, module string) {
if !slices.Contains(b.placed[node], module) {
b.placed[node] = append(b.placed[node], module)
}
pl := delivery.Placement{Node: node}
pl := mma.Placement{Node: node}
for _, m := range b.placed[node] {
pl.Modules = append(pl.Modules, delivery.PlacedModule{Module: m})
pl.Modules = append(pl.Modules, mma.PlacedModule{Module: m})
}
values, err := delivery.PlacementValues(pl)
values, err := mmac.PlacementValues(pl)
if err != nil {
t.Fatal(err)
}
@@ -479,7 +482,7 @@ func TestWantsAreWrittenAndNoDifferenceIs(t *testing.T) {
// assignments name it on. Each node then composes its own: its want names the build alone (Self, no composition), and
// its own want says it composes its own, which the node agent reads (mesh-host's TestNats...ComposesAndAppliesItsOwn).
for _, node := range []string{"ace", "g14", "shanks"} {
pl, err := delivery.EncodePlacement(delivery.Placement{Node: node, Modules: []delivery.PlacedModule{{Module: "web"}}})
pl, err := mmac.EncodePlacement(mma.Placement{Node: node, Modules: []mma.PlacedModule{{Module: "web"}}})
if err != nil {
t.Fatal(err)
}
@@ -1117,14 +1120,14 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) {
Subjects: placementSubjects()}); err != nil { // the seat's four events: the views family has a stream of its own
t.Fatal(err)
}
heard := make(chan delivery.PlacementEvent, 16)
heard := make(chan mma.PlacementEvent, 16)
nc, err := nats.Connect(b.srv.ClientURL())
if err != nil {
t.Fatal(err)
}
t.Cleanup(nc.Close)
if _, err := nc.Subscribe(process.ModuleStreamSubject(delivery.AssignerModule, delivery.PlacementChangedEvent), func(m *nats.Msg) {
var e delivery.PlacementEvent
var e mma.PlacementEvent
if json.Unmarshal(m.Data, &e) == nil {
heard <- e
}
@@ -1140,12 +1143,12 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) {
// place writes a node's placement as another writer would, in its node and module keys (research 072 row 37), and
// answers its node key's revision; legacy writes it in the form before, placement.<node>, alone.
place := func(node string, legacy bool, modules ...string) uint64 {
p := delivery.Placement{Node: node, Modules: []delivery.PlacedModule{}}
p := mma.Placement{Node: node, Modules: []mma.PlacedModule{}}
for _, m := range modules {
p.Modules = append(p.Modules, delivery.PlacedModule{Module: m})
p.Modules = append(p.Modules, mma.PlacedModule{Module: m})
}
values, _ := delivery.PlacementValues(p)
key, _ := delivery.PlacementNodeKey(node)
values, _ := mmac.PlacementValues(p)
key, _ := mmac.NodeKey(node)
if legacy {
key, _ = delivery.LegacyPlacementKey(node)
values = map[string][]byte{key: values[key]}
@@ -1184,7 +1187,7 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) {
}
}
verbs := newSeatTools(bedStore{b}, func(string, any) (json.RawMessage, error) { return json.RawMessage(`{"ok":true}`), nil })
propose := func() delivery.PlacementProposal {
propose := func() mma.Proposal {
t.Helper()
got, err := verbs.doPropose(map[string]any{"why": "move web", "changes": `[{"op":"take-off","node":"anchor","module":"web"},` +
`{"op":"place","node":"laptop","module":"web"}]`}, "operator@laptop")
@@ -1193,12 +1196,12 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) {
}
return got.(proposed).Proposal
}
proposalNow := func(id, what string, ok func(delivery.PlacementProposal) bool) delivery.PlacementProposal {
proposalNow := func(id, what string, ok func(mma.Proposal) bool) mma.Proposal {
t.Helper()
var last delivery.PlacementProposal
var last mma.Proposal
for deadline := time.Now().Add(15 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
if e, err := b.buckets[delivery.PlacementProposalsBucket].Get(b.ctx, id); err == nil {
if p, err := delivery.DecodePlacementProposal(e.Value()); err == nil {
if p, err := plc.DecodeProposal(e.Value()); err == nil {
if last = p; ok(p) {
return p
}
@@ -1208,10 +1211,10 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) {
t.Fatalf("the proposal %s never was %s: %+v", id, what, last)
return last
}
answerAll := func(p delivery.PlacementProposal) {
answerAll := func(p mma.Proposal) {
t.Helper()
for _, node := range p.Must {
key, _ := delivery.PlacementCandidateKey(p.ID, node)
key, _ := plc.CandidateKey(p.ID, node)
e, err := b.buckets[delivery.PlacementCandidatesBucket].Get(b.ctx, key)
if err != nil {
t.Fatalf("no candidate %s: %v", key, err)
@@ -1223,18 +1226,18 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) {
}
p := propose()
p = proposalNow(p.ID, "told who must answer", func(p delivery.PlacementProposal) bool { return len(p.Must) > 0 })
p = proposalNow(p.ID, "told who must answer", func(p mma.Proposal) bool { return len(p.Must) > 0 })
if !slices.Equal(p.Must, []string{"anchor", "laptop"}) {
t.Fatalf("must answer = %v, want both nodes", p.Must)
}
answerAll(p)
p = proposalNow(p.ID, "asked", func(p delivery.PlacementProposal) bool { return p.State == delivery.ProposalAsked })
p = proposalNow(p.ID, "asked", func(p mma.Proposal) bool { return p.State == plc.ProposalAsked })
akey, _ := delivery.PlacementAskKey(p.ID)
b.answer(t, akey, assign.OptionYes)
select {
case e := <-heard:
if e.Proposal.ID != p.ID || len(e.Nodes) != 2 || e.Nodes[0].Before != anchorRev || e.Nodes[1].Before != laptopRev ||
e.Nodes[0].After <= anchorRev || e.Warrant == nil || !slices.Equal(e.Nodes[0].Removed, []string{"web"}) {
if e.Proposal.ID != p.ID || len(e.Nodes) != 2 || changeIn(e.Nodes[0]).Before != anchorRev || changeIn(e.Nodes[1]).Before != laptopRev ||
changeIn(e.Nodes[0]).After <= anchorRev || e.Warrant == nil || !slices.Equal(changeIn(e.Nodes[0]).Removed, []string{"web"}) {
t.Errorf("placement-changed = %+v", e)
}
case <-time.After(15 * time.Second):
@@ -1248,13 +1251,13 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) {
values[key] = e.Value()
}
}
got, err := delivery.PlacementsOf(values)
if err != nil || !slices.Equal(got[node].Assigned(), want) {
t.Errorf("%s = %v, %v, want %v", node, got[node].Assigned(), err, want)
got, err := mmac.PlacementsOf(values)
if err != nil || !slices.Equal(mmac.Assigned(got[node]), want) {
t.Errorf("%s = %v, %v, want %v", node, mmac.Assigned(got[node]), err, want)
}
mirror, _ := delivery.DecodePlacement(values["placement."+node])
if !slices.Equal(mirror.Assigned(), want) {
t.Errorf("the mirror placement.%s = %v, want %v", node, mirror.Assigned(), want)
mirror, _ := mmac.DecodePlacement(values["placement."+node])
if !slices.Equal(mmac.Assigned(mirror), want) {
t.Errorf("the mirror placement.%s = %v, want %v", node, mmac.Assigned(mirror), want)
}
}
if b.hasKey(delivery.PlacementProposalsBucket, p.ID) {
@@ -1268,11 +1271,11 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) {
}
moved := q.(proposed).Proposal
place("laptop", false, "shell", "web") // another writer moved the key
relaid := proposalNow(moved.ID, "laid again", func(p delivery.PlacementProposal) bool {
relaid := proposalNow(moved.ID, "laid again", func(p mma.Proposal) bool {
return strings.Contains(p.Words, "laid again") && len(p.Must) > 0
})
answerAll(relaid)
proposalNow(moved.ID, "asked", func(p delivery.PlacementProposal) bool { return p.State == delivery.ProposalAsked })
proposalNow(moved.ID, "asked", func(p mma.Proposal) bool { return p.State == plc.ProposalAsked })
akey, _ = delivery.PlacementAskKey(moved.ID)
b.answer(t, akey, assign.OptionNo)
for deadline := time.Now().Add(15 * time.Second); b.hasKey(delivery.PlacementProposalsBucket, moved.ID); time.Sleep(50 * time.Millisecond) {
@@ -1280,7 +1283,7 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) {
t.Fatal("a declined proposal was not ended")
}
}
key, _ := delivery.PlacementNodeKey("laptop")
key, _ := mmac.NodeKey("laptop")
if e, _ := b.buckets[delivery.PlacementBucket].Get(b.ctx, key); e == nil || !strings.Contains(string(e.Value()), "web") {
t.Error("a declined proposal wrote its key")
}
@@ -1292,7 +1295,7 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) {
t.Fatal(err)
}
silent := q.(proposed).Proposal
decided := proposalNow(silent.ID, "decided without its silent nodes", func(p delivery.PlacementProposal) bool { return p.State == delivery.ProposalAsked })
decided := proposalNow(silent.ID, "decided without its silent nodes", func(p mma.Proposal) bool { return p.State == plc.ProposalAsked })
if decided.Verdict == nil || decided.Verdict.State != delivery.VerdictPass ||
!strings.Contains(strings.Join(decided.Verdict.Notes, "; "), "no answer from") || !strings.Contains(strings.Join(decided.Verdict.Notes, "; "), "laptop (offline)") {
t.Errorf("the silent node's verdict = %+v", decided.Verdict)
@@ -1311,7 +1314,7 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) {
}
select {
case e := <-heard:
if e.Proposal.ID != joined.(proposed).Proposal.ID || e.Warrant == nil || e.Warrant.Warrant != "enrol-ace-1@2026-10-28T20:00:00Z" || len(e.Nodes) != 1 || e.Nodes[0].Node != "ace" {
if e.Proposal.ID != joined.(proposed).Proposal.ID || e.Warrant == nil || e.Warrant.Warrant != "enrol-ace-1@2026-10-28T20:00:00Z" || len(e.Nodes) != 1 || changeIn(e.Nodes[0]).Node != "ace" {
t.Errorf("placement-changed for the joining node = %+v", e)
}
case <-time.After(15 * time.Second):
@@ -1334,7 +1337,7 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) {
}
select {
case e := <-heard:
if e.Warrant == nil || len(e.Nodes) != 1 || e.Nodes[0].Node != "ace" {
if e.Warrant == nil || len(e.Nodes) != 1 || changeIn(e.Nodes[0]).Node != "ace" {
t.Errorf("placement-changed for the removed node = %+v", e)
}
case <-time.After(15 * time.Second):
@@ -1605,3 +1608,11 @@ func firstViewsBeforeAnyNewFormState(t *testing.T, b *bed, granted map[string][]
}
}
}
// changeIn reads one entry of a placement-changed event's nodes.
func changeIn(body map[string]any) plc.Change {
raw, _ := json.Marshal(body)
var c plc.Change
_ = json.Unmarshal(raw, &c)
return c
}
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go
+3 -1
View File
@@ -19,6 +19,8 @@ import (
"github.com/nats-io/nats.go/jetstream"
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/catalogue"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/process"
@@ -109,7 +111,7 @@ func TestOpensWhatItsManifestNames(t *testing.T) {
for _, name := range append(append([]string{}, readBuckets...), ownBuckets...) {
kvs[name] = &watchKV{memKV: &memKV{values: map[string][]byte{}}}
}
placed, err := delivery.PlacementValues(delivery.Placement{Node: "ace"})
placed, err := mmac.PlacementValues(mma.Placement{Node: "ace"})
if err != nil {
t.Fatal(err)
}
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go
+2 -1
View File
@@ -18,6 +18,7 @@ import (
"git.novox.be/novox/mesh-delivery/sdk/enginereads"
"mesh-delivery-assigner/internal/assign"
plc "mesh-delivery-assigner/internal/placement"
)
// passer makes the assigner's passes: from what the world holds now, it writes each key whose value differs from the
@@ -1110,7 +1111,7 @@ func (p *passer) seatHolders(ctx context.Context) {
if len(placements) == 0 {
return // no placement read yet: nothing is known of who holds what, and nothing is deleted
}
want := delivery.SeatHoldersOf(placements)
want := plc.SeatHoldersOf(placements)
have := p.w.read(delivery.SeatHoldersBucket)
for key, sh := range want {
value, err := json.Marshal(sh)
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go
+10 -7
View File
@@ -15,6 +15,8 @@ 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"
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/busretry"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/delivery"
@@ -24,6 +26,7 @@ import (
"git.novox.be/novox/mesh-delivery/sdk/graph"
"mesh-delivery-assigner/internal/assign"
plc "mesh-delivery-assigner/internal/placement"
)
// memKV is the part of a bucket a pass writes: puts and deletes.
@@ -59,11 +62,11 @@ func newPasser(kvs buckets, at time.Time) *passer {
// (research 072 row 37).
func placeOn(t *testing.T, w *world, node string, modules ...string) {
t.Helper()
pl := delivery.Placement{Node: node}
pl := mma.Placement{Node: node}
for _, m := range modules {
pl.Modules = append(pl.Modules, delivery.PlacedModule{Module: m})
pl.Modules = append(pl.Modules, mma.PlacedModule{Module: m})
}
values, err := delivery.PlacementValues(pl)
values, err := mmac.PlacementValues(pl)
if err != nil {
t.Fatal(err)
}
@@ -985,7 +988,7 @@ func aPlacedNodesWantsNameTheBuildAloneAndItsYesComesFromItsHave(t *testing.T) {
mod, _ := delivery.EncodeModule(delivery.Module{FP2: fpTest, Repository: "novox/web", Commit: strings.Repeat("1", 40), Path: delivery.RootPath})
put(delivery.ModulesBucket, "web", mod) // the trunk's build is not done: rule 1 holds g14 at fpOld
placeOn(t, p.w, "g14", "web")
pl, _ := delivery.EncodePlacement(delivery.Placement{Node: "g14", Modules: []delivery.PlacedModule{{Module: "web"}}})
pl, _ := mmac.EncodePlacement(mma.Placement{Node: "g14", Modules: []mma.PlacedModule{{Module: "web"}}})
put(delivery.PlacementBucket, "placement.g14", pl)
on, _ := delivery.EncodePresence(delivery.Presence{Online: true})
put(delivery.PresenceBucket, "g14", on)
@@ -1070,9 +1073,9 @@ func TestAnApprovedPlacementCallsNoControllerVerb(t *testing.T) {
p.seat = newSeatActs(bus, func() {}, func(string, ...any) {})
var said []string
p.publish = func(event string, _ any, _ string) error { said = append(said, event); return nil }
prop := delivery.PlacementProposal{ID: "p-one", By: "operator@laptop", Why: "move web"}
change := delivery.ChangeOf(delivery.Placement{Node: "anchor"}, delivery.Placement{Node: "anchor", Modules: []delivery.PlacedModule{{Module: "web"}}})
p.followWrite(prop, delivery.Answer{}, []delivery.PlacementChange{change})
prop := mma.Proposal{ID: "p-one", By: "operator@laptop", Why: "move web"}
change := plc.ChangeOf(mma.Placement{Node: "anchor"}, mma.Placement{Node: "anchor", Modules: []mma.PlacedModule{{Module: "web"}}})
p.followWrite(prop, delivery.Answer{}, []plc.Change{change})
if len(bus.calls) != 0 || !slices.Equal(said, []string{delivery.PlacementChangedEvent}) {
t.Fatalf("calls = %d, said = %v", len(bus.calls), said)
}
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go
+88 -74
View File
@@ -17,11 +17,14 @@ 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"
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"
"mesh-delivery-assigner/internal/judge"
plc "mesh-delivery-assigner/internal/placement"
)
// The seat mesh-placement's holder in the assigner's process (novox/hq ADR 0567 §2-§8): each pass follows every open
@@ -136,20 +139,20 @@ func (p *passer) sayWaits(unbound []string) {
// seatPlacements is every node's placement and the revision of its node key (its legacy key's for a node not yet
// converted), as the watch holds them: the revision a proposal is laid over (§2, research 072 row 37).
func (p *passer) seatPlacements() (map[string]delivery.Placement, map[string]uint64) {
func (p *passer) seatPlacements() (map[string]mma.Placement, map[string]uint64) {
return p.w.placement.AllRevisions()
}
// candidatesOf is a proposal's candidates by node, with their keys' revisions and the newest write.
func (p *passer) candidatesOf(id string) (map[string]delivery.PlacementCandidate, map[string]uint64, time.Time) {
out, revs := map[string]delivery.PlacementCandidate{}, map[string]uint64{}
func (p *passer) candidatesOf(id string) (map[string]mma.CandidatesByProposalNodeValue, map[string]uint64, time.Time) {
out, revs := map[string]mma.CandidatesByProposalNodeValue{}, map[string]uint64{}
var newest time.Time
for key, value := range p.w.read(delivery.PlacementCandidatesBucket) {
pid, node, err := delivery.ParsePlacementCandidateKey(key)
pid, node, err := plc.ParseCandidateKey(key)
if err != nil || pid != id {
continue
}
c, err := delivery.DecodePlacementCandidate(value)
c, err := plc.DecodeCandidate(value)
if err != nil {
continue
}
@@ -200,7 +203,7 @@ func (p *passer) placementSeat(ctx context.Context) {
}
sort.Strings(ids)
for _, id := range ids {
prop, err := delivery.DecodePlacementProposal(raws[id])
prop, err := plc.DecodeProposal(raws[id])
if err != nil || prop.ID != id {
p.logf("the proposal %s cannot be read, so it is left: %v", id, err)
continue
@@ -237,7 +240,7 @@ func (p *passer) deletePulses(ctx context.Context, id string) {
}
// ask raises (or raises again) a proposal's authorising ask and its bound, and answers the proposal asked.
func (p *passer) ask(ctx context.Context, prop delivery.PlacementProposal, akey, brief string) (delivery.PlacementProposal, bool) {
func (p *passer) ask(ctx context.Context, prop mma.Proposal, akey, brief string) (mma.Proposal, bool) {
a, err := assign.PlacementAsk(akey, prop, brief, assign.PlacementAsking(prop, p.clock()), p.clock())
if err != nil {
p.logf("the ask about the proposal %s is not written: %v", prop.ID, err)
@@ -249,17 +252,17 @@ func (p *passer) ask(ctx context.Context, prop delivery.PlacementProposal, akey,
return prop, false
}
p.put(ctx, delivery.AssignerAsksBucket, akey, func() ([]byte, error) { return assign.EncodeAsk(akey, a) })
if pk, err := delivery.PulseAskKey(prop.ID); err == nil {
if pk, err := plc.PulseAskKey(prop.ID); err == nil {
p.pulse(ctx, pk, assign.ExpiresAt(placed).Sub(p.clock()))
}
prop.State, prop.Ask = delivery.ProposalAsked, placed.ID
prop.State, prop.Ask = plc.ProposalAsked, placed.ID
return prop, true
}
// stepProposal takes one proposal one step further.
func (p *passer) stepProposal(ctx context.Context, prop delivery.PlacementProposal, rev uint64, current map[string]delivery.Placement, revs map[string]uint64) {
if !p.saidProposed[prop.ID] && prop.State == delivery.ProposalAnswering && prop.Verdict == nil {
p.sayEvent(delivery.PlacementProposedEvent, delivery.PlacementEvent{Proposal: prop}, prop.ID)
func (p *passer) stepProposal(ctx context.Context, prop mma.Proposal, rev uint64, current map[string]mma.Placement, revs map[string]uint64) {
if !p.saidProposed[prop.ID] && prop.State == plc.ProposalAnswering && prop.Verdict == nil {
p.sayEvent(delivery.PlacementProposedEvent, mma.PlacementEvent{Proposal: prop}, prop.ID)
p.saidProposed[prop.ID] = true
}
if prop.Withdrawn != "" {
@@ -276,9 +279,9 @@ func (p *passer) stepProposal(ctx context.Context, prop delivery.PlacementPropos
return
}
akey, _ := delivery.PlacementAskKey(prop.ID)
again := !prop.Again.IsZero() && (prop.Verdict == nil || prop.Again.After(prop.Verdict.At))
if prop.State == delivery.ProposalAsked {
askPulse, _ := delivery.PulseAskKey(prop.ID)
again := prop.Again != nil && !prop.Again.IsZero() && (prop.Verdict == nil || prop.Again.After(prop.Verdict.At))
if prop.State == plc.ProposalAsked {
askPulse, _ := plc.PulseAskKey(prop.ID)
switch {
case p.answeredWith(akey, assign.OptionYes):
p.warranted(ctx, prop, rev, akey, current, revs)
@@ -305,12 +308,12 @@ func (p *passer) stepProposal(ctx context.Context, prop delivery.PlacementPropos
return // a candidate just written is not heard yet: its watch wakes the next pass
}
}
if written.After(prop.Written) {
prop.Written = written
if !written.IsZero() && (prop.Written == nil || written.After(*prop.Written)) {
prop.Written = &written
}
pulsing := map[string]bool{}
for _, node := range prop.Must {
if k, err := delivery.PulseAnswerKey(prop.ID, node); err == nil {
if k, err := plc.PulseAnswerKey(prop.ID, node); err == nil {
pulsing[node] = p.pulsing(k)
}
}
@@ -321,11 +324,11 @@ func (p *passer) stepProposal(ctx context.Context, prop delivery.PlacementPropos
prop.Verdict = &v
switch v.State {
case delivery.VerdictFail:
prop.State, prop.Ask = delivery.ProposalFailed, ""
case delivery.VerdictWaits:
prop.State, prop.Ask = delivery.ProposalAnswering, ""
prop.State, prop.Ask = plc.ProposalFailed, ""
case plc.VerdictWaits:
prop.State, prop.Ask = plc.ProposalAnswering, ""
case delivery.VerdictPass:
if prop.State != delivery.ProposalAsked || again {
if prop.State != plc.ProposalAsked || again {
var ok bool
if prop, ok = p.ask(ctx, prop, akey, judge.Brief(prop, v)); !ok {
return
@@ -336,21 +339,21 @@ func (p *passer) stepProposal(ctx context.Context, prop delivery.PlacementPropos
}
}
if p.writeProposal(ctx, prop, rev) {
p.sayEvent(delivery.PlacementJudgedEvent, delivery.PlacementEvent{Proposal: prop},
p.sayEvent(delivery.PlacementJudgedEvent, mma.PlacementEvent{Proposal: prop},
fmt.Sprintf("%s.judged.%d", prop.ID, v.At.UnixNano()))
}
}
// sameVerdict says two verdicts say the same, their time aside.
func sameVerdict(a, b delivery.PlacementVerdict) bool {
func sameVerdict(a, b mma.Verdict) bool {
a.At, b.At = time.Time{}, time.Time{}
return reflect.DeepEqual(a, b)
}
// moved says whether a key the proposal changes is no longer at the revision it was laid over.
func moved(prop delivery.PlacementProposal, revs map[string]uint64) bool {
func moved(prop mma.Proposal, revs map[string]uint64) bool {
for node, base := range prop.Base {
if revs[node] != base {
if revs[node] != uint64(base) {
return true
}
}
@@ -358,7 +361,7 @@ func moved(prop delivery.PlacementProposal, revs map[string]uint64) bool {
}
// intendedOf is, by proposal and the revisions it was laid over, what Intended computed, kept while this process runs.
func (p *passer) intendedOf(prop delivery.PlacementProposal, current map[string]delivery.Placement, cands map[string]delivery.PlacementCandidate) map[string]map[string]bool {
func (p *passer) intendedOf(prop mma.Proposal, current map[string]mma.Placement, cands map[string]mma.CandidatesByProposalNodeValue) map[string]map[string]bool {
key := fmt.Sprint(prop.ID, prop.Base)
if got, ok := p.intended[key]; ok {
return got
@@ -371,9 +374,9 @@ func (p *passer) intendedOf(prop delivery.PlacementProposal, current map[string]
// compute is what the proposal changes on each node by its placement, and the nodes that answer it: every node with a
// placement or a record (judge.Intended, ADR 0593 §8). The holder composes nothing: each node says what its compositions
// change in its answer.
func (p *passer) compute(prop delivery.PlacementProposal, current map[string]delivery.Placement, cands map[string]delivery.PlacementCandidate) (map[string]map[string]bool, []string) {
laid := map[string]*delivery.Placement{}
for _, node := range prop.Nodes() {
func (p *passer) compute(prop mma.Proposal, current map[string]mma.Placement, cands map[string]mma.CandidatesByProposalNodeValue) (map[string]map[string]bool, []string) {
laid := map[string]*mma.Placement{}
for _, node := range plc.Nodes(prop) {
if c, ok := cands[node]; ok && !c.Reached {
laid[node] = c.Placement
}
@@ -390,8 +393,8 @@ func (p *passer) compute(prop delivery.PlacementProposal, current map[string]del
// mustAnswer finds the nodes the proposal must hear from (§3), writes a reached candidate for each whose key it does not
// change, and writes them into the proposal.
func (p *passer) mustAnswer(ctx context.Context, prop delivery.PlacementProposal, rev uint64, current map[string]delivery.Placement, cands map[string]delivery.PlacementCandidate) {
for _, node := range prop.Nodes() {
func (p *passer) mustAnswer(ctx context.Context, prop mma.Proposal, rev uint64, current map[string]mma.Placement, cands map[string]mma.CandidatesByProposalNodeValue) {
for _, node := range plc.Nodes(prop) {
if _, ok := cands[node]; !ok {
return // the verbs write the candidates before the proposal: wait to hear them
}
@@ -402,18 +405,18 @@ func (p *passer) mustAnswer(ctx context.Context, prop delivery.PlacementProposal
if _, ok := cands[node]; ok {
continue
}
key, err := delivery.PlacementCandidateKey(prop.ID, node)
key, err := plc.CandidateKey(prop.ID, node)
if err != nil {
continue
}
c := delivery.PlacementCandidate{Proposal: prop.ID, Reached: true}
c := mma.CandidatesByProposalNodeValue{Proposal: prop.ID, Reached: true}
if pl, ok := current[node]; ok {
c.Placement = &pl
}
p.put(ctx, delivery.PlacementCandidatesBucket, key, func() ([]byte, error) { return delivery.EncodePlacementCandidate(c) })
p.put(ctx, delivery.PlacementCandidatesBucket, key, func() ([]byte, error) { return plc.EncodeCandidate(c) })
}
for _, node := range must {
if k, err := delivery.PulseAnswerKey(prop.ID, node); err == nil {
if k, err := plc.PulseAnswerKey(prop.ID, node); err == nil {
p.pulse(ctx, k, judge.AnswerBound)
}
}
@@ -423,7 +426,7 @@ func (p *passer) mustAnswer(ctx context.Context, prop delivery.PlacementProposal
// layAgain lays a proposal over the keys as they stand now (§5: a key moved): its candidates are written again, the
// nodes that must answer found again, and it is judged and asked again; a proposal that no longer lays ends.
func (p *passer) layAgain(ctx context.Context, prop delivery.PlacementProposal, rev uint64, current map[string]delivery.Placement, revs map[string]uint64, why string) {
func (p *passer) layAgain(ctx context.Context, prop mma.Proposal, rev uint64, current map[string]mma.Placement, revs map[string]uint64, why string) {
held, recorded := p.lays(ctx)
again, laid, findings := assign.Propose(assign.ProposeIn{ID: prop.ID, By: prop.By, Why: prop.Why, At: prop.At, Changes: prop.Changes,
Current: current, Revisions: revs, Held: held, Recorded: recorded})
@@ -433,12 +436,12 @@ func (p *passer) layAgain(ctx context.Context, prop delivery.PlacementProposal,
}
p.deleteCandidates(ctx, prop.ID)
p.deletePulses(ctx, prop.ID)
for _, node := range again.Nodes() {
key, _ := delivery.PlacementCandidateKey(prop.ID, node)
c := delivery.PlacementCandidate{Proposal: prop.ID, Placement: laid[node]}
p.put(ctx, delivery.PlacementCandidatesBucket, key, func() ([]byte, error) { return delivery.EncodePlacementCandidate(c) })
for _, node := range plc.Nodes(again) {
key, _ := plc.CandidateKey(prop.ID, node)
c := mma.CandidatesByProposalNodeValue{Proposal: prop.ID, Placement: laid[node]}
p.put(ctx, delivery.PlacementCandidatesBucket, key, func() ([]byte, error) { return plc.EncodeCandidate(c) })
}
again.Written, again.Words = p.clock().UTC(), why+": laid again over the keys as they stand, to be judged and asked again"
again.Written, again.Words = ptr(p.clock().UTC()), why+": laid again over the keys as they stand, to be judged and asked again"
p.writeProposal(ctx, again, rev)
}
@@ -457,7 +460,7 @@ func (p *passer) lays(ctx context.Context) (func(string) bool, func(string) bool
// warranted writes an approved proposal (§5): each changed key by compare-and-set at the revision it was laid over; then the warrant recorded, the candidates and the
// proposal deleted, and placement-changed said.
func (p *passer) warranted(ctx context.Context, prop delivery.PlacementProposal, rev uint64, akey string, current map[string]delivery.Placement, revs map[string]uint64) {
func (p *passer) warranted(ctx context.Context, prop mma.Proposal, rev uint64, akey string, current map[string]mma.Placement, revs map[string]uint64) {
if err := p.performs(akey, assign.PlacementAct(prop)); err != nil {
p.endProposal(ctx, prop, "its warrant does not bind these changes, so nothing is written: "+err.Error())
return
@@ -468,8 +471,8 @@ func (p *passer) warranted(ctx context.Context, prop delivery.PlacementProposal,
return
}
cands, _, _ := p.candidatesOf(prop.ID)
laid := map[string]*delivery.Placement{}
for _, node := range prop.Nodes() {
laid := map[string]*mma.Placement{}
for _, node := range plc.Nodes(prop) {
laid[node] = cands[node].Placement
}
writes, err := assign.Writes(prop, laid)
@@ -477,8 +480,8 @@ func (p *passer) warranted(ctx context.Context, prop delivery.PlacementProposal,
p.logf("the proposal %s cannot be written: %v", prop.ID, err)
return
}
var changes []delivery.PlacementChange
writtenLaid := map[string]*delivery.Placement{}
var changes []plc.Change
writtenLaid := map[string]*mma.Placement{}
after := maps.Clone(current)
afterRevs := maps.Clone(revs)
for _, w := range writes {
@@ -496,7 +499,7 @@ func (p *passer) warranted(ctx context.Context, prop delivery.PlacementProposal,
return
}
before := current[w.Node]
now := delivery.Placement{Node: w.Node}
now := mma.Placement{Node: w.Node}
if w.Value != nil {
now = *w.Value
after[w.Node] = *w.Value
@@ -504,7 +507,7 @@ func (p *passer) warranted(ctx context.Context, prop delivery.PlacementProposal,
delete(after, w.Node)
}
afterRevs[w.Node] = rev2
c := delivery.ChangeOf(before, now)
c := plc.ChangeOf(before, now)
c.Before, c.After = w.Base, rev2
changes = append(changes, c)
writtenLaid[w.Node] = w.Value
@@ -520,16 +523,16 @@ func (p *passer) warranted(ctx context.Context, prop delivery.PlacementProposal,
// was the issue's, so no node is waited on and no ask is raised. The key is created at the revision it was laid over (none: a
// node with a key is no joining node); the enrolment recorded the warrant in the hand-act log when it was given, so this does
// not record it again; placement-changed carries the warrant. A key written meanwhile ends the proposal, saying so.
func (p *passer) addNode(ctx context.Context, prop delivery.PlacementProposal, revs map[string]uint64) {
func (p *passer) addNode(ctx context.Context, prop mma.Proposal, revs map[string]uint64) {
current, _ := p.seatPlacements()
if moved(prop, revs) {
p.endProposal(ctx, prop, "a key it was laid over was written since: the enrolment proposes the node again")
return
}
laid, findings := delivery.LayChanges(current, prop.Changes)
laid, findings := plc.LayChanges(current, prop.Changes)
findings = append(findings, assign.ControllerRefusals(prop.Changes, laid)...)
node := prop.Changes[0].Node
if prop.Changes[0].Op == delivery.OpRemoveNode {
if prop.Changes[0].Op == plc.OpRemoveNode {
p.removeNode(ctx, prop, current, findings)
return
}
@@ -543,8 +546,8 @@ func (p *passer) addNode(ctx context.Context, prop delivery.PlacementProposal, r
p.endProposal(ctx, prop, "the node's key was not written: "+err.Error())
return
}
c := delivery.ChangeOf(delivery.Placement{Node: node}, value)
c.Before, c.After = prop.Base[node], rev
c := plc.ChangeOf(mma.Placement{Node: node}, value)
c.Before, c.After = uint64(prop.Base[node]), rev
p.deleteCandidates(ctx, prop.ID)
p.deletePulses(ctx, prop.ID)
_ = p.kvs[delivery.PlacementProposalsBucket].Delete(ctx, prop.ID)
@@ -552,13 +555,13 @@ func (p *passer) addNode(ctx context.Context, prop delivery.PlacementProposal, r
w := prop.Warrant
p.sayChanged(prop, delivery.Answer{Option: assign.OptionYes, Warrant: w.Warrant,
Words: fmt.Sprintf("the operator approved the machine %s at %s through the enrolment's ask %s", w.Node, w.Address, w.Ask)},
[]delivery.PlacementChange{c})
[]plc.Change{c})
}
// removeNode deletes the key of a joining node whose enrolment ended unconfirmed, on that enrolment's warrant (the operator's
// decision of 2026-11-02 01:05): no node answers and no ask is raised, as for its add-node; placement-changed carries the warrant.
// A node that runs modules or holds seats is no joining node and stays (LayChanges' finding ends the proposal).
func (p *passer) removeNode(ctx context.Context, prop delivery.PlacementProposal, current map[string]delivery.Placement, findings []string) {
func (p *passer) removeNode(ctx context.Context, prop mma.Proposal, current map[string]mma.Placement, findings []string) {
node := prop.Changes[0].Node
before, ok := current[node]
if len(findings) > 0 || !ok {
@@ -570,8 +573,8 @@ func (p *passer) removeNode(ctx context.Context, prop delivery.PlacementProposal
p.endProposal(ctx, prop, "the node's key was not deleted: "+err.Error())
return
}
c := delivery.ChangeOf(before, delivery.Placement{Node: node})
c.Before, c.After = prop.Base[node], rev
c := plc.ChangeOf(before, mma.Placement{Node: node})
c.Before, c.After = uint64(prop.Base[node]), rev
p.deleteCandidates(ctx, prop.ID)
p.deletePulses(ctx, prop.ID)
_ = p.kvs[delivery.PlacementProposalsBucket].Delete(ctx, prop.ID)
@@ -579,12 +582,12 @@ func (p *passer) removeNode(ctx context.Context, prop delivery.PlacementProposal
w := prop.Warrant
p.sayChanged(prop, delivery.Answer{Option: assign.OptionYes, Warrant: w.Warrant,
Words: fmt.Sprintf("the machine %s left placement: its enrolment (ask %s) ended before it confirmed", w.Node, w.Ask)},
[]delivery.PlacementChange{c})
[]plc.Change{c})
}
// followWrite follows the keys a warrant wrote: the warrant recorded and placement-changed said. No verb of the
// controller is called: the placement is the one source of where modules run (ADR 0567 §8, issue 774).
func (p *passer) followWrite(prop delivery.PlacementProposal, answer delivery.Answer, changes []delivery.PlacementChange) {
func (p *passer) followWrite(prop mma.Proposal, answer delivery.Answer, changes []plc.Change) {
if len(changes) == 0 {
return
}
@@ -592,7 +595,7 @@ func (p *passer) followWrite(prop delivery.PlacementProposal, answer delivery.An
p.sayChanged(prop, answer, changes)
}
func sortedNodes(m map[string]*delivery.Placement) []string {
func sortedNodes(m map[string]*mma.Placement) []string {
out := make([]string, 0, len(m))
for n := range m {
out = append(out, n)
@@ -602,7 +605,7 @@ func sortedNodes(m map[string]*delivery.Placement) []string {
}
// sayChanged says placement-changed for the keys written.
func (p *passer) sayChanged(prop delivery.PlacementProposal, answer delivery.Answer, changes []delivery.PlacementChange) {
func (p *passer) sayChanged(prop mma.Proposal, answer delivery.Answer, changes []plc.Change) {
if len(changes) == 0 {
return
}
@@ -612,14 +615,14 @@ func (p *passer) sayChanged(prop delivery.PlacementProposal, answer delivery.Ans
at = t
}
}
delivery.SortChanges(changes)
p.sayEvent(delivery.PlacementChangedEvent, delivery.PlacementEvent{Proposal: prop, Nodes: changes,
Warrant: &delivery.PlacementWarrant{Warrant: answer.Warrant, Words: answer.Words, At: at}}, prop.ID+".changed")
plc.SortChanges(changes)
p.sayEvent(delivery.PlacementChangedEvent, mma.PlacementEvent{Proposal: prop, Nodes: changeBodies(changes),
Warrant: &mma.PlacementEventWarrant{Warrant: answer.Warrant, Words: answer.Words, At: at}}, prop.ID+".changed")
}
// endProposal ends a proposal: its ask, its candidates and its key deleted, and placement-withdrawn said carrying it
// whole with why it ended.
func (p *passer) endProposal(ctx context.Context, prop delivery.PlacementProposal, why string) {
func (p *passer) endProposal(ctx context.Context, prop mma.Proposal, why string) {
if akey, err := delivery.PlacementAskKey(prop.ID); err == nil && p.w.has(delivery.AssignerAsksBucket, akey) {
if err := p.kvs[delivery.AssignerAsksBucket].Delete(ctx, akey); err == nil {
p.w.put(delivery.AssignerAsksBucket, akey, nil)
@@ -633,13 +636,13 @@ func (p *passer) endProposal(ctx context.Context, prop delivery.PlacementProposa
}
p.w.put(delivery.PlacementProposalsBucket, prop.ID, nil)
p.logf("the proposal %s ends: %s", prop.ID, why)
p.sayEvent(delivery.PlacementWithdrawnEvent, delivery.PlacementEvent{Proposal: prop, Ended: why}, prop.ID+".withdrawn")
p.sayEvent(delivery.PlacementWithdrawnEvent, mma.PlacementEvent{Proposal: prop, Ended: why}, prop.ID+".withdrawn")
}
// deleteCandidates deletes every candidate of a proposal.
func (p *passer) deleteCandidates(ctx context.Context, id string) {
for key := range p.w.read(delivery.PlacementCandidatesBucket) {
if pid, _, err := delivery.ParsePlacementCandidateKey(key); err == nil && pid == id {
if pid, _, err := plc.ParseCandidateKey(key); err == nil && pid == id {
if err := p.kvs[delivery.PlacementCandidatesBucket].Delete(ctx, key); err == nil {
p.w.put(delivery.PlacementCandidatesBucket, key, nil)
}
@@ -649,8 +652,8 @@ func (p *passer) deleteCandidates(ctx context.Context, id string) {
// writeProposal writes a proposal over the revision the pass read, by compare-and-set: a verb that marked it meanwhile
// (withdraw, again) wins, and the next pass reads that. It says whether it wrote.
func (p *passer) writeProposal(ctx context.Context, prop delivery.PlacementProposal, rev uint64) bool {
raw, err := delivery.EncodePlacementProposal(prop)
func (p *passer) writeProposal(ctx context.Context, prop mma.Proposal, rev uint64) bool {
raw, err := plc.EncodeProposal(prop)
if err != nil {
p.logf("the proposal %s is not written: %v", prop.ID, err)
return false
@@ -673,7 +676,7 @@ func (p *passer) writeProposal(ctx context.Context, prop delivery.PlacementPropo
}
// say says one of the seat's events, said in the journal when the bus refuses it.
func (p *passer) sayEvent(event string, body delivery.PlacementEvent, id string) {
func (p *passer) sayEvent(event string, body mma.PlacementEvent, id string) {
if p.publish == nil {
return
}
@@ -682,14 +685,14 @@ func (p *passer) sayEvent(event string, body delivery.PlacementEvent, id string)
}
}
// writeNode writes one node's placement after (nil: none) as its keys (mesh-sdk delivery.PlacementKeys.Writes; research
// writeNode writes one node's placement after (nil: none) as its keys (mesh-sdk mmac.Keys.Writes; research
// 072 row 37): the module keys that change, each carrying the value it replaces, then the node key, the commit, then the
// module keys no longer placed. It answers the node key's revision after (0: deleted). A write that fails up to the
// commit ends it with nothing changed for any reader, the module keys written before it read at their previous values;
// a delete after the commit that fails leaves a module key the node key does not name, which places nothing and is
// deleted at the node's next write. Each key written is taken into the world at once, so the next write of the pass
// reads it.
func (p *passer) writeNode(ctx context.Context, node string, after *delivery.Placement) (uint64, error) {
func (p *passer) writeNode(ctx context.Context, node string, after *mma.Placement) (uint64, error) {
writes, err := p.w.placement.Writes(node, after)
if err != nil {
return 0, err
@@ -714,7 +717,7 @@ func (p *passer) writeNode(ctx context.Context, node string, after *delivery.Pla
}
// writeKey makes one write of a placement key by compare-and-set and answers the key's revision after (0: deleted).
func writeKey(ctx context.Context, kv jetstream.KeyValue, w delivery.PlacementKeyWrite) (uint64, error) {
func writeKey(ctx context.Context, kv jetstream.KeyValue, w mmac.KeyWrite) (uint64, error) {
switch {
case w.Value == nil:
return 0, kv.Delete(ctx, w.Key, jetstream.LastRevision(w.Base))
@@ -751,3 +754,14 @@ func (p *passer) convertPlacement(ctx context.Context) {
}
p.convertedAll = !failed && err == nil
}
func ptr[T any](v T) *T { return &v }
// changeBodies are the changes as the seat's event carries them.
func changeBodies(changes []plc.Change) []map[string]any {
out := make([]map[string]any, 0, len(changes))
for _, c := range changes {
out = append(out, c.Body())
}
return out
}
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools.go
+63 -61
View File
@@ -4,16 +4,18 @@ import (
"encoding/json"
"errors"
"fmt"
"git.novox.be/novox/mesh-contracts/placement/go/seats/meshmoduleassigner"
"slices"
"sort"
"strings"
"time"
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"
)
// The seat mesh-placement's seven verbs (novox/hq ADR 0567 §10), served through the tool runner beside the put tools: a
@@ -55,22 +57,22 @@ func newSeatTools(st seatStore, ask func(string, any) (json.RawMessage, error))
// keyed is one placement key as a verb answers it.
type keyed struct {
Placement delivery.Placement `json:"placement"`
Revision uint64 `json:"revision"`
Placement mma.Placement `json:"placement"`
Revision uint64 `json:"revision"`
}
// placements reads every node's placement and the revision of its node key (its legacy key's before it is converted),
// assembled from its keys in either form (delivery.PlacementKeys; research 072 row 37).
func (s *seatTools) placements() (map[string]delivery.Placement, map[string]uint64, error) {
// assembled from its keys in either form (mmac.Keys; research 072 row 37).
func (s *seatTools) placements() (map[string]mma.Placement, map[string]uint64, error) {
keys, err := s.st.keys(delivery.PlacementState)
if err != nil {
return nil, nil, fmt.Errorf("placement cannot be listed: %w", err)
}
sort.Strings(keys)
held := delivery.NewPlacementKeys()
held := mmac.NewKeys()
var bad []string
for _, key := range keys {
if _, _, _, err := delivery.ParsePlacementKey(key); err != nil {
if _, _, _, err := mmac.ParseKey(key); err != nil {
continue
}
raw, rev, found, err := s.st.get(delivery.PlacementState, key)
@@ -123,30 +125,30 @@ func (s *seatTools) lays() (func(string) bool, func(string) bool, error) {
}
// proposal reads one open proposal.
func (s *seatTools) proposal(id string) (delivery.PlacementProposal, error) {
if err := delivery.CheckProposalID(id); err != nil {
return delivery.PlacementProposal{}, err
func (s *seatTools) proposal(id string) (mma.Proposal, error) {
if err := mmac.CheckProposalID(id); err != nil {
return mma.Proposal{}, err
}
raw, _, found, err := s.st.get(delivery.PlacementProposalsState, id)
if err != nil {
return delivery.PlacementProposal{}, fmt.Errorf("the proposal %s cannot be read: %w", id, err)
return mma.Proposal{}, fmt.Errorf("the proposal %s cannot be read: %w", id, err)
}
if !found {
return delivery.PlacementProposal{}, fmt.Errorf("no proposal %s is open: mesh-placement.proposals lists those that are", id)
return mma.Proposal{}, fmt.Errorf("no proposal %s is open: mesh-placement.proposals lists those that are", id)
}
return delivery.DecodePlacementProposal(raw)
return plc.DecodeProposal(raw)
}
// write writes a proposal laid by assign.Propose: each candidate, then the proposal, whose write the assigner's process
// takes as the event to compose and judge it.
func (s *seatTools) write(p delivery.PlacementProposal, laid map[string]*delivery.Placement) error {
p.Written = s.now().UTC()
for _, node := range p.Nodes() {
key, err := delivery.PlacementCandidateKey(p.ID, node)
func (s *seatTools) write(p mma.Proposal, laid map[string]*mma.Placement) error {
p.Written = ptr(s.now().UTC())
for _, node := range plc.Nodes(p) {
key, err := plc.CandidateKey(p.ID, node)
if err != nil {
return err
}
raw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Proposal: p.ID, Placement: laid[node]})
raw, err := plc.EncodeCandidate(mma.CandidatesByProposalNodeValue{Proposal: p.ID, Placement: laid[node]})
if err != nil {
return err
}
@@ -154,7 +156,7 @@ func (s *seatTools) write(p delivery.PlacementProposal, laid map[string]*deliver
return fmt.Errorf("the candidate %s is not written: %w", key, err)
}
}
raw, err := delivery.EncodePlacementProposal(p)
raw, err := plc.EncodeProposal(p)
if err != nil {
return err
}
@@ -166,8 +168,8 @@ func (s *seatTools) write(p delivery.PlacementProposal, laid map[string]*deliver
// proposed is what propose and adopt answer.
type proposed struct {
Proposal delivery.PlacementProposal `json:"proposal"`
Follow string `json:"follow"`
Proposal mma.Proposal `json:"proposal"`
Follow string `json:"follow"`
}
func followProposal(id string) string {
@@ -176,7 +178,7 @@ func followProposal(id string) string {
}
// changesOf reads propose's changes: a JSON array, given as one or as its text.
func changesOf(v any) ([]delivery.PlacementProposalChange, error) {
func changesOf(v any) ([]mma.ProposalChange, error) {
var raw []byte
switch c := v.(type) {
case string:
@@ -189,7 +191,7 @@ func changesOf(v any) ([]delivery.PlacementProposalChange, error) {
return nil, err
}
}
var out []delivery.PlacementProposalChange
var out []mma.ProposalChange
dec := json.NewDecoder(strings.NewReader(string(raw)))
dec.DisallowUnknownFields()
if err := dec.Decode(&out); err != nil {
@@ -205,7 +207,7 @@ func (s *seatTools) doPropose(args map[string]any, by string) (any, error) {
return nil, err
}
for _, c := range changes {
if c.Op == delivery.OpSeed {
if c.Op == plc.OpSeed {
return nil, errors.New("the seed is gone with adopt: propose takes place, take-off, pin, unpin, hold, unhold, decide and config")
}
}
@@ -222,9 +224,9 @@ func (s *seatTools) doPropose(args map[string]any, by string) (any, error) {
return nil, err
}
now := s.now()
p, laid, findings := assign.Propose(assign.ProposeIn{ID: delivery.NewProposalID(now), By: by, Why: arg(args, "why"), At: now,
p, laid, findings := assign.Propose(assign.ProposeIn{ID: plc.NewProposalID(now), By: by, Why: arg(args, "why"), At: now,
Changes: changes, Current: current, Revisions: revs, Held: held, Recorded: recorded, Warrant: warrant})
if refusal := delivery.AddNodeRefusal(p, moduleOfCaller(by)); refusal != "" {
if refusal := plc.AddNodeRefusal(p, moduleOfCaller(by), delivery.EnrolmentModule); refusal != "" {
findings = append(findings, refusal)
}
if len(findings) > 0 {
@@ -237,7 +239,7 @@ func (s *seatTools) doPropose(args map[string]any, by string) (any, error) {
}
// warrantOf reads propose's warrant, an enrolment's for an add-node: an object, given as one or as its text; none is none.
func warrantOf(v any) (*delivery.AddNodeWarrant, error) {
func warrantOf(v any) (*mma.Warrant, error) {
var raw []byte
switch w := v.(type) {
case nil:
@@ -253,7 +255,7 @@ func warrantOf(v any) (*delivery.AddNodeWarrant, error) {
return nil, err
}
}
var out delivery.AddNodeWarrant
var out mma.Warrant
dec := json.NewDecoder(strings.NewReader(string(raw)))
dec.DisallowUnknownFields()
if err := dec.Decode(&out); err != nil {
@@ -284,7 +286,7 @@ func (s *seatTools) doWithdraw(args map[string]any, by string) (any, error) {
return nil, err
}
p.Withdrawn = fmt.Sprintf("withdrawn by %s: %s", by, why)
raw, err := delivery.EncodePlacementProposal(p)
raw, err := plc.EncodeProposal(p)
if err != nil {
return nil, err
}
@@ -301,8 +303,8 @@ func (s *seatTools) doAgain(args map[string]any) (any, error) {
if err != nil {
return nil, err
}
p.Again = s.now().UTC()
raw, err := delivery.EncodePlacementProposal(p)
p.Again = ptr(s.now().UTC())
raw, err := plc.EncodeProposal(p)
if err != nil {
return nil, err
}
@@ -324,11 +326,11 @@ type placementAnswer struct {
// historyEntry is one change of a key in the state's revisions.
type historyEntry struct {
Node string `json:"node"`
Revision uint64 `json:"revision"`
At time.Time `json:"at"`
Change delivery.PlacementChange `json:"change"`
Deleted bool `json:"deleted,omitempty"`
Node string `json:"node"`
Revision uint64 `json:"revision"`
At time.Time `json:"at"`
Change plc.Change `json:"change"`
Deleted bool `json:"deleted,omitempty"`
}
// doPlacement is the verb placement (§10, §6): one node's key, one module's nodes, or every key; with history, the
@@ -347,7 +349,7 @@ func (s *seatTools) doPlacement(args map[string]any) (any, error) {
if node != "" && n != node {
continue
}
if module != "" && !strings.Contains(" "+strings.Join(p.Assigned(), " ")+" ", " "+module+" ") {
if module != "" && !strings.Contains(" "+strings.Join(mmac.Assigned(p), " ")+" ", " "+module+" ") {
continue
}
out.Keys[n] = keyed{Placement: p, Revision: revs[n]}
@@ -402,7 +404,7 @@ func (s *seatTools) doPlacement(args map[string]any) (any, error) {
// verdict with the differences and the answers it judged, its ask), its candidates by node with their revisions, and
// every node's answer now.
type openProposal struct {
Proposal delivery.PlacementProposal `json:"proposal"`
Proposal mma.Proposal `json:"proposal"`
Candidates map[string]uint64 `json:"candidates"`
Answers map[string]delivery.CandidateAnswer `json:"answers,omitempty"`
}
@@ -454,7 +456,7 @@ func (s *seatTools) doProposals(args map[string]any) (any, error) {
}
o := openProposal{Proposal: p, Candidates: map[string]uint64{}, Answers: answers[id]}
for _, k := range cands {
pid, node, err := delivery.ParsePlacementCandidateKey(k)
pid, node, err := plc.ParseCandidateKey(k)
if err != nil || pid != id {
continue
}
@@ -474,15 +476,15 @@ func (s *seatTools) doProposals(args map[string]any) (any, error) {
// only from a tool of that name), with the seat's own schemas.
func (s *seatTools) tools() []stdio.Tool {
run := map[string]func(map[string]any) (any, error){
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,
mma.VerbPlacement: s.doPlacement,
mma.VerbProposals: s.doProposals,
mma.VerbPropose: func(a map[string]any) (any, error) { return s.doPropose(a, stdio.CallerOf(a)) },
mma.VerbWithdraw: func(a map[string]any) (any, error) { return s.doWithdraw(a, stdio.CallerOf(a)) },
mma.VerbAgain: s.doAgain,
}
var out []stdio.Tool
for _, v := range meshmoduleassigner.Verbs() {
out = append(out, stdio.Tool{Name: meshmoduleassigner.Seat + "." + v, Run: run[v]})
for _, v := range mma.Verbs() {
out = append(out, stdio.Tool{Name: mma.Seat + "." + v, Run: run[v]})
}
return out
}
@@ -495,7 +497,7 @@ func (s *seatTools) seats(node string) (any, error) {
return nil, fmt.Errorf("seat-holders cannot be listed: %w", err)
}
sort.Strings(keys)
out := []delivery.SeatHolders{}
out := []mma.SeatHolders{}
for _, key := range keys {
raw, _, found, err := s.st.get(delivery.SeatHoldersState, key)
if err != nil {
@@ -504,11 +506,11 @@ func (s *seatTools) seats(node string) (any, error) {
if !found {
continue
}
var sh delivery.SeatHolders
var sh mma.SeatHolders
if json.Unmarshal(raw, &sh) != nil {
continue
}
if node != "" && sh.Node != node && !slices.ContainsFunc(sh.Holders, func(h delivery.SeatHolder) bool { return h.Node == node }) {
if node != "" && sh.Node != node && !slices.ContainsFunc(sh.Holders, func(h mma.SeatHoldersHoldersItem) bool { return h.Node == node }) {
continue
}
out = append(out, sh)
@@ -530,7 +532,7 @@ func (s *seatTools) nodeHistory(node string) ([]keyedRevision, error) {
if err != nil {
return nil, err
}
nodeKey, err := delivery.PlacementNodeKey(node)
nodeKey, err := mmac.NodeKey(node)
if err != nil {
return nil, err
}
@@ -549,43 +551,43 @@ func (s *seatTools) nodeHistory(node string) ([]keyedRevision, error) {
}
// sketch is a revision of a node's placement as far as its history says a change: each module's digest, and its pins
// and holdings, which change every module (delivery.ChangeOf).
// and holdings, which change every module (plc.ChangeOf).
type sketch struct {
modules []delivery.PlacementRef
modules []mma.PlacementByNodeNodeValueModulesItem
nodeWide string
}
// sketchOf reads one revision of a node key or a legacy key; a value that does not read places nothing.
func sketchOf(key string, value []byte) sketch {
_, _, kind, err := delivery.ParsePlacementKey(key)
_, _, kind, err := mmac.ParseKey(key)
if err != nil {
return sketch{}
}
wide := func(pins map[string]delivery.Provider, holds []delivery.Holding) string {
wide := func(pins map[string]mma.Provider, holds []mma.Holding) string {
raw, _ := json.Marshal([]any{pins, holds})
return string(raw)
}
if kind == delivery.PlacementNodeKind {
n, err := delivery.DecodePlacementNode(value)
if kind == mmac.NodeKind {
n, err := mmac.DecodeNode(value)
if err != nil {
return sketch{}
}
return sketch{modules: n.Modules, nodeWide: wide(n.Pins, n.Holds)}
}
p, err := delivery.DecodePlacement(value)
p, err := mmac.DecodePlacement(value)
if err != nil {
return sketch{}
}
out := sketch{nodeWide: wide(p.Pins, p.Holds)}
for _, m := range p.Modules {
out.modules = append(out.modules, delivery.PlacementRef{Module: m.Module, Digest: m.Digest()})
out.modules = append(out.modules, mma.PlacementByNodeNodeValueModulesItem{Module: m.Module, Digest: mmac.Digest(m)})
}
return out
}
// changeFrom is what changed from before to s on node, as delivery.ChangeOf says it.
func (s sketch) changeFrom(node string, before sketch) delivery.PlacementChange {
c := delivery.PlacementChange{Node: node}
// changeFrom is what changed from before to s on node, as plc.ChangeOf says it.
func (s sketch) changeFrom(node string, before sketch) plc.Change {
c := plc.Change{Node: node}
was := map[string]string{}
for _, r := range before.modules {
was[r.Module] = r.Digest
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go
+23 -20
View File
@@ -12,6 +12,8 @@ import (
"testing"
"time"
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/catalogue"
"git.novox.be/novox/mesh-sdk/go/delivery"
@@ -20,6 +22,7 @@ import (
"github.com/nats-io/nats.go/jetstream"
"mesh-delivery-assigner/internal/assign"
plc "mesh-delivery-assigner/internal/placement"
)
func (s memStore) del(state, key string) error {
@@ -39,7 +42,7 @@ func (s memStore) history(state, key string) ([]stdio.StateRevision, error) {
// or refuses naming every finding and writes nothing; withdraw and again mark it for the holder; placement and
// proposals read every state the seat keeps, and history names the stream for what the state no longer keeps.
func TestTheSeatsVerbs(t *testing.T) {
anchor := delivery.Placement{V: delivery.Version, Node: "anchor", Modules: []delivery.PlacedModule{{Module: "web"}}}
anchor := mma.Placement{V: delivery.Version, Node: "anchor", Modules: []mma.PlacedModule{{Module: "web"}}}
st := memStore{}
for _, state := range []string{delivery.PlacementState, delivery.PlacementProposalsState, delivery.PlacementCandidatesState, modulesState, haveState} {
st[state] = &memKV{values: map[string][]byte{}}
@@ -58,7 +61,7 @@ func TestTheSeatsVerbs(t *testing.T) {
})
s.now = func() time.Time { return time.Date(2026, 10, 28, 20, 0, 0, 0, time.UTC) }
raw, _ := delivery.EncodePlacement(anchor)
raw, _ := mmac.EncodePlacement(anchor)
st[delivery.PlacementState].values["placement.anchor"] = raw
// withdraw and again mark the proposal; the holder ends or judges it.
@@ -73,7 +76,7 @@ func TestTheSeatsVerbs(t *testing.T) {
if _, err := s.doAgain(map[string]any{"id": seed.ID}); err != nil {
t.Fatal(err)
}
if p, err := s.proposal(seed.ID); err != nil || !strings.Contains(p.Withdrawn, "not yet") || p.Again.IsZero() {
if p, err := s.proposal(seed.ID); err != nil || !strings.Contains(p.Withdrawn, "not yet") || p.Again == nil || p.Again.IsZero() {
t.Errorf("marked = %+v, %v", p, err)
}
if _, err := s.doWithdraw(map[string]any{"id": "p-none", "why": "x"}, "op"); err == nil {
@@ -96,7 +99,7 @@ func TestTheSeatsVerbs(t *testing.T) {
t.Fatal(err)
}
move := got.(proposed).Proposal
if move.By != "operator@laptop" || move.Base["anchor"] != uint64(len(raw)) || move.Base["laptop"] != 0 || len(st[delivery.PlacementCandidatesState].values) != 2 {
if move.By != "operator@laptop" || move.Base["anchor"] != int64(len(raw)) || move.Base["laptop"] != 0 || len(st[delivery.PlacementCandidatesState].values) != 2 {
t.Fatalf("propose = %+v", move)
}
@@ -161,7 +164,7 @@ func TestTheSeatsVerbs(t *testing.T) {
// A remove-node (the operator's decision of 2026-11-02 01:05) takes a joining node that never confirmed off placement: from
// mesh-enrolment, on the warrant naming that node, of a node running nothing, it is proposed with the node's key to delete;
// of a node running modules, or with another node's warrant, it is refused.
joiner, _ := delivery.EncodePlacement(delivery.Placement{V: delivery.Version, Node: "joiner", Decided: delivery.NodeDecided{Address: "10.9.0.7"}, Modules: []delivery.PlacedModule{}})
joiner, _ := mmac.EncodePlacement(mma.Placement{V: delivery.Version, Node: "joiner", Decided: mma.NodeDecided{Address: "10.9.0.7"}, Modules: []mma.PlacedModule{}})
st[delivery.PlacementState].values["placement.joiner"] = joiner
removeNode := func(node, warrant string) (any, error) {
return s.doPropose(map[string]any{"why": "enrolment ended", "warrant": warrant,
@@ -177,7 +180,7 @@ func TestTheSeatsVerbs(t *testing.T) {
}
if got, err := removeNode("joiner", `{"node":"joiner","address":"10.9.0.7","ask":"a","warrant":"a@t"}`); err != nil {
t.Errorf("a remove-node from mesh-enrolment: %v", err)
} else if rm := got.(proposed).Proposal; !slices.Equal(rm.Nodes(), []string{"joiner"}) || !slices.Equal(rm.RemovedNodes(), []string{"joiner"}) {
} else if rm := got.(proposed).Proposal; !slices.Equal(plc.Nodes(rm), []string{"joiner"}) || !slices.Equal(plc.RemovedNodes(rm), []string{"joiner"}) {
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).
@@ -188,7 +191,7 @@ func TestTheSeatsVerbs(t *testing.T) {
t.Run("1955/L7 (ADR 0571 §6): rule 7 reads the node's own composition; an unplaced node is said", func(t *testing.T) {
w := newWorld()
fp := delivery.SourcePrefix2 + strings.Repeat("2", 64)
pl, _ := delivery.EncodePlacement(delivery.Placement{V: delivery.Version, Node: "ace", Modules: []delivery.PlacedModule{{Module: "web"}}})
pl, _ := mmac.EncodePlacement(mma.Placement{V: delivery.Version, Node: "ace", Modules: []mma.PlacedModule{{Module: "web"}}})
pkey, _ := delivery.LegacyPlacementKey("ace")
w.putEntry(delivery.PlacementBucket, pkey, pl, 1, time.Now())
for node, digest := range map[string]string{"ace": "sha256:" + strings.Repeat("a", 64), "g14": ""} {
@@ -210,7 +213,7 @@ func TestTheSeatsVerbs(t *testing.T) {
if _, ok := findings[assign.UnplacedKey("g14")]; !ok || len(findings) != 1 {
t.Fatalf("findings = %v", findings)
}
pl2, _ := delivery.EncodePlacement(delivery.Placement{V: delivery.Version, Node: "g14", Modules: []delivery.PlacedModule{{Module: "web"}}})
pl2, _ := mmac.EncodePlacement(mma.Placement{V: delivery.Version, Node: "g14", Modules: []mma.PlacedModule{{Module: "web"}}})
pkey2, _ := delivery.LegacyPlacementKey("g14")
w.putEntry(delivery.PlacementBucket, pkey2, pl2, 1, time.Now())
p.unplacedFindings()
@@ -222,16 +225,16 @@ func TestTheSeatsVerbs(t *testing.T) {
writes := 0
kv := &countingKV{memKV: memKV{values: map[string][]byte{}}, writes: &writes}
p := &passer{w: newWorld(), logf: t.Logf, kvs: buckets{delivery.SeatHoldersBucket: kv}}
place := func(node string, holds ...delivery.Holding) {
pl, err := delivery.EncodePlacement(delivery.Placement{V: delivery.Version, Node: node, Modules: []delivery.PlacedModule{{Module: "web"}, {Module: "nats"}}, Holds: holds})
place := func(node string, holds ...mma.Holding) {
pl, err := mmac.EncodePlacement(mma.Placement{V: delivery.Version, Node: node, Modules: []mma.PlacedModule{{Module: "web"}, {Module: "nats"}}, Holds: holds})
if err != nil {
t.Fatal(err)
}
k, _ := delivery.LegacyPlacementKey(node)
p.w.putEntry(delivery.PlacementBucket, k, pl, 1, time.Now())
}
place("ace", delivery.Holding{Claim: "mesh-broker", Scope: "mesh", Module: "nats"}, delivery.Holding{Claim: "node-shell", Scope: "node", Module: "web"})
place("g14", delivery.Holding{Claim: "node-shell", Scope: "node", Module: "web"})
place("ace", mma.Holding{Claim: "mesh-broker", Scope: "mesh", Module: "nats"}, mma.Holding{Claim: "node-shell", Scope: "node", Module: "web"})
place("g14", mma.Holding{Claim: "node-shell", Scope: "node", Module: "web"})
ctx := context.Background()
p.seatHolders(ctx)
if writes != 3 || kv.values["mesh-broker"] == nil || kv.values["node-shell.ace"] == nil || kv.values["node-shell.g14"] == nil {
@@ -248,7 +251,7 @@ func TestTheSeatsVerbs(t *testing.T) {
}
st := memStore{delivery.SeatHoldersState: &memKV{values: kv.values}}
out, err := newSeatTools(st, nil).seats("ace")
if err != nil || len(out.(map[string]any)["seats"].([]delivery.SeatHolders)) != 2 {
if err != nil || len(out.(map[string]any)["seats"].([]mma.SeatHolders)) != 2 {
t.Errorf("seats on ace = %+v, %v", out, err)
}
})
@@ -261,8 +264,8 @@ func TestTheSeatsVerbs(t *testing.T) {
put := func(key string, value []byte, rev uint64) {
p.w.putEntry(delivery.PlacementBucket, key, value, rev, time.Now())
}
was := delivery.Placement{Node: "anchor", Modules: []delivery.PlacedModule{{Module: "web"}}}
raw, _ := delivery.EncodePlacement(was)
was := mma.Placement{Node: "anchor", Modules: []mma.PlacedModule{{Module: "web"}}}
raw, _ := mmac.EncodePlacement(was)
put("placement.anchor", raw, kv.write("placement.anchor", raw))
p.convertPlacement(ctx)
if strings.Join(kv.log, " ") != "create anchor.web create anchor.node" || !p.convertedAll ||
@@ -274,23 +277,23 @@ func TestTheSeatsVerbs(t *testing.T) {
t.Errorf("a converted node is converted again: %v", kv.log)
}
// A write: the module keys, the node key (its commit), the mirror, then the module key no longer placed.
after := delivery.Placement{Node: "anchor", Modules: []delivery.PlacedModule{{Module: "shell", Ports: map[string]int{"22": 2222}}}}
after := mma.Placement{Node: "anchor", Modules: []mma.PlacedModule{{Module: "shell", Ports: map[string]int64{"22": 2222}}}}
kv.log = nil
rev, err := p.writeNode(ctx, "anchor", &after)
if err != nil || strings.Join(kv.log, " ") != "create anchor.shell update anchor.node update placement.anchor delete anchor.web" || rev != kv.revs["anchor.node"] {
t.Fatalf("the write = %v, %d, %v", kv.log, rev, err)
}
if got, _ := p.seatPlacements(); !delivery.SamePlacement(got["anchor"], after) {
if got, _ := p.seatPlacements(); !mmac.SamePlacement(got["anchor"], after) {
t.Errorf("the write is not read back at once: %+v", got)
}
// A commit refused leaves every reader on what stood: the module key written before it carries the value it replaced.
kv.refuse, kv.log = "anchor.node", nil
moved := delivery.Placement{Node: "anchor", Modules: []delivery.PlacedModule{{Module: "shell", Ports: map[string]int{"22": 2223}}}}
moved := mma.Placement{Node: "anchor", Modules: []mma.PlacedModule{{Module: "shell", Ports: map[string]int64{"22": 2223}}}}
if _, err := p.writeNode(ctx, "anchor", &moved); err == nil || strings.Join(kv.log, " ") != "update anchor.shell" {
t.Fatalf("a refused commit = %v, %v", kv.log, err)
}
fresh, err := delivery.PlacementsOf(kv.values)
if err != nil || !delivery.SamePlacement(fresh["anchor"], after) {
fresh, err := mmac.PlacementsOf(kv.values)
if err != nil || !mmac.SamePlacement(fresh["anchor"], after) {
t.Errorf("a reader of the keys after a refused commit reads %+v, %v", fresh["anchor"], err)
}
kv.refuse = ""
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go
+8 -6
View File
@@ -11,6 +11,8 @@ import (
"github.com/nats-io/nats.go/jetstream"
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"
@@ -36,8 +38,8 @@ type world struct {
// trunkBuilt is, by module, its newest done trunk build in the builder's log (putBuilt).
trunkBuilt map[string]delivery.BuildLogEntry
// placement is the seat's state placement assembled node by node from its keys in either form (mesh-sdk
// delivery.PlacementKeys; research 072 row 37): the one reader of its keys in the assigner.
placement *delivery.PlacementKeys
// mmac.Keys; research 072 row 37): the one reader of its keys in the assigner.
placement *mmac.Keys
// sourceGroups is, by revision of the gate's source.main, the release (group) that moved it, from every revision its
// history keeps (watched with history): a build's source revision names the release that carries it (ADR 0601 §1).
sourceGroups map[uint64]string
@@ -58,7 +60,7 @@ type parsed struct {
func newWorld() *world {
return &world{values: map[string]map[string][]byte{}, manifests: map[string]parsed{}, healthy: map[string][]string{}, changes: map[string]uint64{},
meta: map[string]entryMeta{}, trunkBuilt: map[string]delivery.BuildLogEntry{}, placement: delivery.NewPlacementKeys(),
meta: map[string]entryMeta{}, trunkBuilt: map[string]delivery.BuildLogEntry{}, placement: mmac.NewKeys(),
sourceGroups: map[uint64]string{}}
}
@@ -262,13 +264,13 @@ func (w *world) assigned() map[string][]string {
func (w *world) placed() (map[string][]string, bool) {
out := map[string][]string{}
for node, p := range w.placements() {
out[node] = p.Assigned()
out[node] = mmac.Assigned(p)
}
return out, len(w.read(delivery.PlacementBucket)) > 0
}
// placements is every node's placement, assembled from its keys in either form (delivery.PlacementKeys).
func (w *world) placements() map[string]delivery.Placement {
// placements is every node's placement, assembled from its keys in either form (mmac.Keys).
func (w *world) placements() map[string]mma.Placement {
out, _ := w.placement.AllRevisions()
return out
}
modules/mesh-delivery-assigner/internal/assign/asks_test.go
+6 -1
View File
@@ -16,6 +16,11 @@ import (
// 1297/B9 (the assigner's half): the ask about a held fingerprint is a valid ask of the router and of mesh-sdk's layout at
// its key, of a kind the view may answer.
func TestTheHoldAskIsValid(t *testing.T) {
t.Run("hold", theHoldAskIsValid)
t.Run("restore", theRestoreAskIsValid)
}
func theHoldAskIsValid(t *testing.T) {
key, err := delivery.HoldAskKey("mail", fNew)
if err != nil {
t.Fatal(err)
@@ -62,7 +67,7 @@ func TestTheHoldAskIsValid(t *testing.T) {
// 1906/D3 (ADR 0539 §19): the ask about a data module held on a node is a valid ask of the router and of mesh-sdk's layout at
// its restore key, of a kind the view may answer; it offers restoring (losing the newer writes) and leaving it stopped.
func TestTheRestoreAskIsValid(t *testing.T) {
func theRestoreAskIsValid(t *testing.T) {
key, err := delivery.RestoreAskKey("ace", "mail", fNew)
if err != nil {
t.Fatal(err)
modules/mesh-delivery-assigner/internal/assign/seat.go
+38 -34
View File
@@ -9,7 +9,11 @@ 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"
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"
)
// The seat mesh-placement (novox/hq ADR 0567), held by the assigner beside its wants: placement is its state
@@ -22,13 +26,13 @@ import (
type ProposeIn struct {
ID, By, Why string
At time.Time
Changes []delivery.PlacementProposalChange
Current map[string]delivery.Placement
Changes []mma.ProposalChange
Current map[string]mma.Placement
Revisions map[string]uint64
Held func(module string) bool
Recorded func(node string) bool
// Warrant is the enrolment's warrant an add-node carries (ADR 0587 §7); only that change has one.
Warrant *delivery.AddNodeWarrant
Warrant *mma.Warrant
}
// Propose lays a proposal over the current keys (§2): the proposal (state answering, each changed node's base revision),
@@ -36,9 +40,9 @@ type ProposeIn struct {
// CheckPlacements refuses of the result laid over every placement (two modules on one host port, a provider not placed,
// a seat held once held twice, two nodes with one address, two hubs, a module the indexer does not hold, a node with no
// record). A key whose value the proposal leaves as it is changes nothing and is not laid.
func Propose(in ProposeIn) (delivery.PlacementProposal, map[string]*delivery.Placement, []string) {
p := delivery.PlacementProposal{ID: in.ID, By: strings.TrimSpace(in.By), Why: strings.TrimSpace(in.Why), At: in.At.UTC(),
Changes: in.Changes, Warrant: in.Warrant, Base: map[string]uint64{}, State: delivery.ProposalAnswering}
func Propose(in ProposeIn) (mma.Proposal, map[string]*mma.Placement, []string) {
p := mma.Proposal{ID: in.ID, By: strings.TrimSpace(in.By), Why: strings.TrimSpace(in.Why), At: in.At.UTC(),
Changes: in.Changes, Warrant: in.Warrant, Base: map[string]int64{}, State: plc.ProposalAnswering}
var findings []string
if p.By == "" {
findings = append(findings, "the runtime names no caller: a proposal is made by someone")
@@ -50,36 +54,36 @@ func Propose(in ProposeIn) (delivery.PlacementProposal, map[string]*delivery.Pla
findings = append(findings, "a proposal changes something: it names no change")
}
for i, c := range in.Changes {
if c.Op == delivery.OpSeed {
if c.Op == plc.OpSeed {
findings = append(findings, fmt.Sprintf("change %d: the seed is gone with adopt, the controller's export being gone", i+1))
}
}
laid, lay := delivery.LayChanges(in.Current, in.Changes)
laid, lay := plc.LayChanges(in.Current, in.Changes)
findings = append(findings, lay...)
findings = append(findings, ControllerRefusals(in.Changes, laid)...)
all := map[string]delivery.Placement{}
all := map[string]mma.Placement{}
for node, pl := range in.Current {
all[node] = pl
}
out := map[string]*delivery.Placement{}
out := map[string]*mma.Placement{}
for node, pl := range laid {
if before, ok := in.Current[node]; ok && delivery.SamePlacement(before, pl) {
if before, ok := in.Current[node]; ok && mmac.SamePlacement(before, pl) {
continue
}
pl := pl
all[node] = pl
out[node] = &pl
p.Base[node] = in.Revisions[node]
p.Base[node] = int64(in.Revisions[node])
}
// A remove-node deletes the key of a joining node that never confirmed (the operator's decision of 2026-11-02 01:05): its
// candidate is no placement, and the checks across nodes read the mesh without it. LayChanges said why one cannot go.
for _, node := range p.RemovedNodes() {
for _, node := range plc.RemovedNodes(p) {
if _, ok := in.Current[node]; !ok {
continue
}
delete(all, node)
out[node] = nil
p.Base[node] = in.Revisions[node]
p.Base[node] = int64(in.Revisions[node])
}
if len(findings) == 0 && len(out) == 0 {
findings = append(findings, "the changes leave every placement as it is")
@@ -95,7 +99,7 @@ func Propose(in ProposeIn) (delivery.PlacementProposal, map[string]*delivery.Pla
// this one caller is the exception to a node with no record (ADR 0587 §7, changing ADR 0567 §2).
joining := map[string]bool{}
for _, c := range in.Changes {
if c.Op == delivery.OpAddNode {
if c.Op == plc.OpAddNode {
joining[c.Node] = true
}
}
@@ -103,7 +107,7 @@ func Propose(in ProposeIn) (delivery.PlacementProposal, map[string]*delivery.Pla
was := recorded
recorded = func(node string) bool { return joining[node] || was(node) }
}
for _, r := range delivery.CheckPlacements(all, held, recorded) {
for _, r := range plc.CheckPlacements(all, held, recorded) {
findings = append(findings, r)
}
return p, out, findings
@@ -119,12 +123,12 @@ const ControllerEnded = "the controller ended by its unassignment on 2026-11-02;
// ControllerRefusals names every change, and every node as laid, that places the controller: by a place, by a seed's
// whole placement, or in what the changes lay. None when nothing does.
func ControllerRefusals(changes []delivery.PlacementProposalChange, laid map[string]delivery.Placement) []string {
func ControllerRefusals(changes []mma.ProposalChange, laid map[string]mma.Placement) []string {
var out []string
nodes := map[string]bool{}
for i, c := range changes {
places := c.Module == ControllerModule && c.Op != delivery.OpTakeOff
if c.Whole != nil && slices.Contains(c.Whole.Assigned(), ControllerModule) {
places := c.Module == ControllerModule && c.Op != plc.OpTakeOff
if c.Whole != nil && slices.Contains(mmac.Assigned(*c.Whole), ControllerModule) {
places = true
}
if places {
@@ -133,7 +137,7 @@ func ControllerRefusals(changes []delivery.PlacementProposalChange, laid map[str
}
}
for _, node := range slices.Sorted(maps.Keys(laid)) {
if !nodes[node] && slices.Contains(laid[node].Assigned(), ControllerModule) {
if !nodes[node] && slices.Contains(mmac.Assigned(laid[node]), ControllerModule) {
out = append(out, fmt.Sprintf("%s would run %s: %s", node, ControllerModule, ControllerEnded))
}
}
@@ -141,23 +145,23 @@ func ControllerRefusals(changes []delivery.PlacementProposalChange, laid map[str
}
// Write is one node a warrant writes (§5): the node, its node key `<node>.node` (the commit of the node's keys,
// mesh-sdk delivery.PlacementKeys.Writes), the revision of the node's placement it is written at (0: it has none, so
// mesh-sdk mmac.Keys.Writes), the revision of the node's placement it is written at (0: it has none, so
// it is created) and its value (nil: it is deleted).
type Write struct {
Node, Key string
Base uint64
Value *delivery.Placement
Value *mma.Placement
}
// Writes is every key a proposal writes, by node order.
func Writes(p delivery.PlacementProposal, laid map[string]*delivery.Placement) ([]Write, error) {
func Writes(p mma.Proposal, laid map[string]*mma.Placement) ([]Write, error) {
var out []Write
for _, node := range p.Nodes() {
key, err := delivery.PlacementNodeKey(node)
for _, node := range plc.Nodes(p) {
key, err := mmac.NodeKey(node)
if err != nil {
return nil, err
}
out = append(out, Write{Node: node, Key: key, Base: p.Base[node], Value: laid[node]})
out = append(out, Write{Node: node, Key: key, Base: uint64(p.Base[node]), Value: laid[node]})
}
return out, nil
}
@@ -178,24 +182,24 @@ func CallClearedCondition(kind, node, module string) oc.Condition {
}
// PlacementAct is the act an approved proposal's yes binds: writing exactly its changes at its base revisions.
func PlacementAct(p delivery.PlacementProposal) occ.Act {
return occ.Act{"placement": "write", "proposal": p.ID, "changes": delivery.ChangesDigest(p)}
func PlacementAct(p mma.Proposal) occ.Act {
return occ.Act{"placement": "write", "proposal": p.ID, "changes": plc.ChangesDigest(p)}
}
// PlacementAsking is a proposal's asking number: each judging that raises its ask again (again, a key moved) is
// another asking, under an id of its own.
func PlacementAsking(p delivery.PlacementProposal, at time.Time) int {
func PlacementAsking(p mma.Proposal, at time.Time) int {
return int(at.Unix())
}
// PlacementAsk is the authorising ask about a proposal (§5): an Approve that binds its act, so only a channel that
// proves who answered takes it, with the brief of the verdict by node (judge.Brief) as its explanation.
func PlacementAsk(key string, p delivery.PlacementProposal, brief string, asking int, now time.Time) (delivery.AssignerAsk, error) {
func PlacementAsk(key string, p mma.Proposal, brief string, asking int, now time.Time) (delivery.AssignerAsk, error) {
binds, err := occ.ActDigest(PlacementAct(p))
if err != nil {
return delivery.AssignerAsk{}, err
}
nodes := p.Nodes()
nodes := plc.Nodes(p)
headline := fmt.Sprintf("Change placement on %s?", strings.Join(nodes, ", "))
if n := len([]rune(headline)); n > occ.HeadlineLength {
headline = fmt.Sprintf("Change placement on %d machines?", len(nodes))
@@ -218,9 +222,9 @@ func PlacementAsk(key string, p delivery.PlacementProposal, brief string, asking
// Rest is what a proposal leaves to write once the keys of written are: its changes of the other nodes, to be laid again
// over the keys as the write left them (§5: a key moved while it was written). False when nothing is left.
func Rest(p delivery.PlacementProposal, written map[string]*delivery.Placement) (delivery.PlacementProposal, bool) {
func Rest(p mma.Proposal, written map[string]*mma.Placement) (mma.Proposal, bool) {
rest := p
rest.Changes, rest.Base = nil, map[string]uint64{}
rest.Changes, rest.Base = nil, map[string]int64{}
for _, c := range p.Changes {
if _, done := written[c.Node]; !done {
rest.Changes = append(rest.Changes, c)
@@ -231,6 +235,6 @@ func Rest(p delivery.PlacementProposal, written map[string]*delivery.Placement)
rest.Base[node] = base
}
}
rest.Must, rest.Verdict, rest.Ask, rest.State = nil, nil, "", delivery.ProposalAnswering
rest.Must, rest.Verdict, rest.Ask, rest.State = nil, nil, "", plc.ProposalAnswering
return rest, len(rest.Changes) > 0
}
modules/mesh-delivery-assigner/internal/assign/seat_test.go
+31 -27
View File
@@ -8,47 +8,51 @@ import (
"time"
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"
plc "mesh-delivery-assigner/internal/placement"
)
// ADR 0567 §2, §5, §7, §8: what the seat's holder decides of a proposal: every refusal named at once, a valid one laid
// with the base revision of each key it changes, the writes at those revisions, the controller's calls that keep its
// assignments in step, the authorising ask bound to the changes, and the export read as the controller answers it.
func TestThePlacementSeatDecides(t *testing.T) {
anchor := delivery.Placement{V: delivery.Version, Node: "anchor", Decided: delivery.NodeDecided{Hub: true, Address: "10.0.0.1"},
Modules: []delivery.PlacedModule{{Module: "store", Ports: map[string]int{"5432": 5432}}, {Module: "web"}}}
laptop := delivery.Placement{V: delivery.Version, Node: "laptop", Decided: delivery.NodeDecided{Address: "10.0.0.2"}, Modules: []delivery.PlacedModule{{Module: "shell"}}}
in := func(changes ...delivery.PlacementProposalChange) ProposeIn {
anchor := mma.Placement{V: delivery.Version, Node: "anchor", Decided: mma.NodeDecided{Hub: true, Address: "10.0.0.1"},
Modules: []mma.PlacedModule{{Module: "store", Ports: map[string]int64{"5432": 5432}}, {Module: "web"}}}
laptop := mma.Placement{V: delivery.Version, Node: "laptop", Decided: mma.NodeDecided{Address: "10.0.0.2"}, Modules: []mma.PlacedModule{{Module: "shell"}}}
in := func(changes ...mma.ProposalChange) ProposeIn {
return ProposeIn{ID: "p-one", By: "operator@laptop", Why: "move web", At: time.Unix(1, 0), Changes: changes,
Current: map[string]delivery.Placement{"anchor": anchor, "laptop": laptop}, Revisions: map[string]uint64{"anchor": 4, "laptop": 9},
Current: map[string]mma.Placement{"anchor": anchor, "laptop": laptop}, Revisions: map[string]uint64{"anchor": 4, "laptop": 9},
Held: func(m string) bool { return m != "ghost" },
Recorded: func(n string) bool { return n == "anchor" || n == "laptop" }}
}
move := []delivery.PlacementProposalChange{{Op: delivery.OpTakeOff, Node: "anchor", Module: "web"}, {Op: delivery.OpPlace, Node: "laptop", Module: "web"}}
move := []mma.ProposalChange{{Op: plc.OpTakeOff, Node: "anchor", Module: "web"}, {Op: plc.OpPlace, Node: "laptop", Module: "web"}}
p, laid, findings := Propose(in(move...))
if len(findings) != 0 || p.State != delivery.ProposalAnswering || p.Base["anchor"] != 4 || p.Base["laptop"] != 9 ||
!slices.Equal(laid["laptop"].Assigned(), []string{"shell", "web"}) || !slices.Equal(laid["anchor"].Assigned(), []string{"store"}) {
if len(findings) != 0 || p.State != plc.ProposalAnswering || p.Base["anchor"] != 4 || p.Base["laptop"] != 9 ||
!slices.Equal(mmac.Assigned(*laid["laptop"]), []string{"shell", "web"}) || !slices.Equal(mmac.Assigned(*laid["anchor"]), []string{"store"}) {
t.Fatalf("a move: %+v, %+v, %v", p, laid, findings)
}
if _, err := delivery.EncodePlacementProposal(p); err != nil {
if _, err := plc.EncodeProposal(p); err != nil {
t.Fatalf("the proposal cannot be written: %v", err)
}
for name, c := range map[string]struct {
in ProposeIn
want string
}{
"a module not held": {in(delivery.PlacementProposalChange{Op: delivery.OpPlace, Node: "laptop", Module: "ghost"}), "the indexer holds no module ghost"},
"a node with no record": {in(delivery.PlacementProposalChange{Op: delivery.OpPlace, Node: "stranger", Module: "web"}), "stranger is a node with no record"},
"nothing changes": {in(delivery.PlacementProposalChange{Op: delivery.OpPlace, Node: "anchor", Module: "web"}), "leave every placement as it is"},
"a module not held": {in(mma.ProposalChange{Op: plc.OpPlace, Node: "laptop", Module: "ghost"}), "the indexer holds no module ghost"},
"a node with no record": {in(mma.ProposalChange{Op: plc.OpPlace, Node: "stranger", Module: "web"}), "stranger is a node with no record"},
"nothing changes": {in(mma.ProposalChange{Op: plc.OpPlace, Node: "anchor", Module: "web"}), "leave every placement as it is"},
"no why": {func() ProposeIn { i := in(move...); i.Why = " "; return i }(), "says why"},
"no caller": {func() ProposeIn { i := in(move...); i.By = ""; return i }(), "names no caller"},
"a seed by hand": {in(delivery.PlacementProposalChange{Op: delivery.OpSeed, Node: "laptop", Whole: &laptop}), "gone with adopt"},
"a seed by hand": {in(mma.ProposalChange{Op: plc.OpSeed, Node: "laptop", Whole: &laptop}), "gone with adopt"},
// 1126: the controller ended by its unassignment; placed again it would be a second writer beside the parts that
// act without its lease.
"the controller placed again": {in(delivery.PlacementProposalChange{Op: delivery.OpPlace, Node: "laptop", Module: "mesh-controller"}), "ended by its unassignment"},
"the controller seeded whole": {in(delivery.PlacementProposalChange{Op: delivery.OpSeed, Node: "anchor", Whole: &delivery.Placement{V: delivery.Version, Node: "anchor",
Decided: anchor.Decided, Modules: []delivery.PlacedModule{{Module: "mesh-controller"}}}}), "two writers (issue 1126)"},
"a pin of nobody placed": {in(delivery.PlacementProposalChange{Op: delivery.OpPin, Node: "laptop", Provision: "db", Provider: &delivery.Provider{Node: "laptop", Module: "store"}}), "not placed there"},
"the controller placed again": {in(mma.ProposalChange{Op: plc.OpPlace, Node: "laptop", Module: "mesh-controller"}), "ended by its unassignment"},
"the controller seeded whole": {in(mma.ProposalChange{Op: plc.OpSeed, Node: "anchor", Whole: &mma.Placement{V: delivery.Version, Node: "anchor",
Decided: anchor.Decided, Modules: []mma.PlacedModule{{Module: "mesh-controller"}}}}), "two writers (issue 1126)"},
"a pin of nobody placed": {in(mma.ProposalChange{Op: plc.OpPin, Node: "laptop", Provision: "db", Provider: &mma.Provider{Node: "laptop", Module: "store"}}), "not placed there"},
} {
if _, _, f := Propose(c.in); !slices.ContainsFunc(f, func(s string) bool { return strings.Contains(s, c.want) }) {
t.Errorf("%s: findings %v, want one naming %q", name, f, c.want)
@@ -56,13 +60,13 @@ func TestThePlacementSeatDecides(t *testing.T) {
}
// Every kind of change is taken: the placement is the one source, and the controller's verbs limit nothing (ADR 0567 §8).
for _, c := range []delivery.PlacementProposalChange{
{Op: delivery.OpConfig, Node: "anchor", Module: "web", Values: map[string]any{"puid": 1}},
{Op: delivery.OpPlace, Node: "laptop", Module: "web", Ports: map[string]int{"80": 8080}, Bindings: map[string]delivery.Provider{"db": {Node: "anchor", Module: "store"}}},
{Op: delivery.OpConfig, Node: "anchor", Module: "web", Layer: map[string]any{"places": []any{}}},
{Op: delivery.OpConfig, Node: "anchor", Module: "web", MeshWide: map[string]any{"accesses": []any{}}},
{Op: delivery.OpDecide, Node: "laptop", Field: "site", Value: json.RawMessage(`"attic"`)},
{Op: delivery.OpHold, Node: "anchor", Module: "store", Claim: "node-shell", Scope: "node"},
for _, c := range []mma.ProposalChange{
{Op: plc.OpConfig, Node: "anchor", Module: "web", Values: map[string]any{"puid": 1}},
{Op: plc.OpPlace, Node: "laptop", Module: "web", Ports: map[string]int64{"80": 8080}, Bindings: map[string]mma.Provider{"db": {Node: "anchor", Module: "store"}}},
{Op: plc.OpConfig, Node: "anchor", Module: "web", Layer: map[string]any{"places": []any{}}},
{Op: plc.OpConfig, Node: "anchor", Module: "web", MeshWide: map[string]any{"accesses": []any{}}},
{Op: plc.OpDecide, Node: "laptop", Field: "site", Value: json.RawMessage(`"attic"`)},
{Op: plc.OpHold, Node: "anchor", Module: "store", Claim: "node-shell", Scope: "node"},
} {
if _, _, f := Propose(in(c)); len(f) != 0 {
t.Errorf("%s is refused: %v", c.Op, f)
@@ -70,8 +74,8 @@ func TestThePlacementSeatDecides(t *testing.T) {
}
// A key moved while the warrant wrote: what is left to write is the other nodes' changes, to be laid and asked again.
if rest, ok := Rest(p, map[string]*delivery.Placement{"anchor": laid["anchor"]}); !ok || len(rest.Changes) != 1 || rest.Changes[0].Node != "laptop" ||
len(rest.Base) != 1 || rest.State != delivery.ProposalAnswering || rest.Must != nil {
if rest, ok := Rest(p, map[string]*mma.Placement{"anchor": laid["anchor"]}); !ok || len(rest.Changes) != 1 || rest.Changes[0].Node != "laptop" ||
len(rest.Base) != 1 || rest.State != plc.ProposalAnswering || rest.Must != nil {
t.Errorf("the rest = %+v, %v", rest, ok)
}
if _, ok := Rest(p, laid); ok {
@@ -96,7 +100,7 @@ func TestThePlacementSeatDecides(t *testing.T) {
t.Fatalf("the ask = %+v, %v", a, err)
}
q := p
q.Changes = append(slices.Clone(p.Changes), delivery.PlacementProposalChange{Op: delivery.OpUnpin, Node: "laptop", Provision: "x"})
q.Changes = append(slices.Clone(p.Changes), mma.ProposalChange{Op: plc.OpUnpin, Node: "laptop", Provision: "x"})
if occ.Performs(mustAsk(t, a).Options[0], PlacementAct(q)) == nil {
t.Error("the ask's yes binds changes it was not asked about")
}
modules/mesh-delivery-assigner/internal/judge/judge.go
+34 -14
View File
@@ -15,7 +15,10 @@ import (
"time"
nec "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
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"
)
// AnswerBound is how long the holder waits for a node's answer after its candidate was written: the time to live of the
@@ -30,7 +33,7 @@ var AnswerBound = nec.AnswerBound
// node's set holds the modules the proposal adds, removes or whose placed entry differs on that node; a node with no
// key now gets an empty set (§7: the seed is answered as the same). What a node's compositions change beyond its own
// placement it says itself in its answer, marked in the brief as not computed.
func Intended(current map[string]delivery.Placement, laid map[string]*delivery.Placement, nodes []string) (intended map[string]map[string]bool, must []string) {
func Intended(current map[string]mma.Placement, laid map[string]*mma.Placement, nodes []string) (intended map[string]map[string]bool, must []string) {
intended = map[string]map[string]bool{}
all := map[string]bool{}
for _, n := range nodes {
@@ -48,11 +51,11 @@ func Intended(current map[string]delivery.Placement, laid map[string]*delivery.P
if !changesKey || !placed {
continue // untouched, or new to placement: every answer must be the same
}
var afterModules []delivery.PlacedModule
var afterModules []mma.PlacedModule
if after != nil {
afterModules = after.Modules
}
entries := map[string]delivery.PlacedModule{}
entries := map[string]mma.PlacedModule{}
for _, m := range before.Modules {
entries[m.Module] = m
}
@@ -80,10 +83,10 @@ func Intended(current map[string]delivery.Placement, laid map[string]*delivery.P
// on the bus) is named as giving no answer and the proposal decided on the answers it has (ADR 0593 §8).
// A difference fails nothing: one Intended computed is listed under Intended, one it did not under Unintended, both
// shown in the ask's brief. The seed alone fails on any difference.
func JudgeAnswers(p delivery.PlacementProposal, intended map[string]map[string]bool, answers map[string]*delivery.CandidateAnswer,
revisions map[string]uint64, online map[string]bool, pulsing map[string]bool, now time.Time) delivery.PlacementVerdict {
v := delivery.PlacementVerdict{At: now.UTC(), Intended: map[string][]string{}, Unintended: map[string][]string{}, Carried: map[string][]string{},
Answers: map[string]delivery.CandidateAnswer{}}
func JudgeAnswers(p mma.Proposal, intended map[string]map[string]bool, answers map[string]*delivery.CandidateAnswer,
revisions map[string]uint64, online map[string]bool, pulsing map[string]bool, now time.Time) mma.Verdict {
v := mma.Verdict{At: now.UTC(), Intended: map[string][]string{}, Unintended: map[string][]string{}, Carried: map[string][]string{},
Answers: map[string]map[string]any{}}
for node, set := range intended {
for m := range set {
v.Intended[node] = append(v.Intended[node], m)
@@ -92,7 +95,7 @@ func JudgeAnswers(p delivery.PlacementProposal, intended map[string]map[string]b
}
must := p.Must
if len(must) == 0 {
must = p.Nodes()
must = plc.Nodes(p)
}
for _, node := range must {
a := answers[node]
@@ -100,7 +103,7 @@ func JudgeAnswers(p delivery.PlacementProposal, intended map[string]map[string]b
v.Silent = append(v.Silent, node)
continue
}
v.Answers[node] = *a
v.Answers[node] = answerBody(*a)
if a.Refused != "" {
v.Fails = append(v.Fails, fmt.Sprintf("%s cannot compose the proposal: %s", node, a.Refused))
continue
@@ -167,7 +170,7 @@ func JudgeAnswers(p delivery.PlacementProposal, intended map[string]map[string]b
case len(v.Fails) > 0:
v.State = delivery.VerdictFail
case len(waiting) > 0:
v.State = delivery.VerdictWaits
v.State = plc.VerdictWaits
default:
v.State = delivery.VerdictPass
}
@@ -176,7 +179,7 @@ func JudgeAnswers(p delivery.PlacementProposal, intended map[string]map[string]b
// Brief is the verdict as the ask's brief shows it, by node (§5): what the proposal changes and what each node answered
// beside it, marked, and what stays held as the node holds it today, with the proposal's why.
func Brief(p delivery.PlacementProposal, v delivery.PlacementVerdict) string {
func Brief(p mma.Proposal, v mma.Verdict) string {
var b strings.Builder
fmt.Fprintf(&b, "%s (proposed by %s)", p.Why, p.By)
nodes := map[string]bool{}
@@ -186,11 +189,11 @@ func Brief(p delivery.PlacementProposal, v delivery.PlacementVerdict) string {
for n := range v.Unintended {
nodes[n] = true
}
for _, n := range p.Nodes() {
for _, n := range plc.Nodes(p) {
nodes[n] = true
}
for n, a := range v.Answers {
if len(a.Held) > 0 {
if len(answerOf(a).Held) > 0 {
nodes[n] = true
}
}
@@ -215,7 +218,7 @@ func Brief(p delivery.PlacementProposal, v delivery.PlacementVerdict) string {
if m := v.Carried[n]; len(m) > 0 {
fmt.Fprintf(&b, "; the seed carries: %s", strings.Join(m, "; "))
}
if held := v.Answers[n].Held; len(held) > 0 {
if held := answerOf(v.Answers[n]).Held; len(held) > 0 {
var names []string
for _, h := range held {
today := h.Today
@@ -330,3 +333,20 @@ func fieldName(running, composed, path string) string {
}
return path
}
// answerBody is a node's answer as the verdict carries it: the seat's contract keeps a verdict's answers as bodies, since
// the answer's layout is the node-engine's (ADR 0593 §7).
func answerBody(a delivery.CandidateAnswer) map[string]any {
raw, _ := json.Marshal(a)
var out map[string]any
_ = json.Unmarshal(raw, &out)
return out
}
// answerOf reads a verdict's answer body back; a body that does not fit is an answer with nothing in it.
func answerOf(body map[string]any) delivery.CandidateAnswer {
raw, _ := json.Marshal(body)
var out delivery.CandidateAnswer
_ = json.Unmarshal(raw, &out)
return out
}
modules/mesh-delivery-assigner/internal/judge/judge_test.go
+9 -6
View File
@@ -6,17 +6,20 @@ import (
"testing"
"time"
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"
)
// ADR 0567 §3, §4 (rows moved from the assigner's check of a pull request's placement, ADR 0559 §5): what a proposal
// changes by node, and the verdict of the answers: differences shown and failing nothing, a refusal and a silent node
// failing, a later answer judging again, the seed failing on any difference.
func TestJudge(t *testing.T) {
current := map[string]delivery.Placement{"anchor": {Node: "anchor", Modules: []delivery.PlacedModule{{Module: "mail"}}}, "server": {Node: "server"}}
laid := map[string]*delivery.Placement{
"anchor": {Node: "anchor", Modules: []delivery.PlacedModule{{Module: "mail", Ports: map[string]int{"25": 2525}}, {Module: "web"}}},
"laptop": {Node: "laptop", Modules: []delivery.PlacedModule{{Module: "web"}}}}
current := map[string]mma.Placement{"anchor": {Node: "anchor", Modules: []mma.PlacedModule{{Module: "mail"}}}, "server": {Node: "server"}}
laid := map[string]*mma.Placement{
"anchor": {Node: "anchor", Modules: []mma.PlacedModule{{Module: "mail", Ports: map[string]int64{"25": 2525}}, {Module: "web"}}},
"laptop": {Node: "laptop", Modules: []mma.PlacedModule{{Module: "web"}}}}
// ADR 0593 §8: every node answers; what the placement changes is read from the placement alone.
got, must := Intended(current, laid, []string{"anchor", "laptop", "server"})
if !got["anchor"]["mail"] || !got["anchor"]["web"] || len(got["laptop"]) != 0 || len(got["server"]) != 0 ||
@@ -25,7 +28,7 @@ func TestJudge(t *testing.T) {
}
at := time.Date(2026, 10, 27, 2, 0, 0, 0, time.UTC)
p := delivery.PlacementProposal{ID: "p-one", By: "op", Why: "move web", Written: at, Base: map[string]uint64{"anchor": 3, "laptop": 0},
p := mma.Proposal{ID: "p-one", By: "op", Why: "move web", Written: &at, Base: map[string]int64{"anchor": 3, "laptop": 0},
Must: []string{"anchor", "laptop"}}
revisions := map[string]uint64{"anchor": 7, "laptop": 8}
intended := map[string]map[string]bool{"anchor": {"web": true}}
@@ -56,7 +59,7 @@ func TestJudge(t *testing.T) {
Composed: `{"env":{"MESH_OWNED_PATHS":"{\"/etc/hosts\":\"hostname\",\"/etc/resolv.conf\":\"systemd-resolved (node-resolver)\"}","MESH_TOOL_MODULES":"… (cut at 8192 bytes)`}}}}}}, nil, delivery.VerdictPass, "", ""},
{"a refusal", false, map[string]*delivery.CandidateAnswer{"anchor": same(7), "laptop": {Proposal: "p-one", Revision: 8, Verdict: delivery.CandidateRefused,
Refusals: []delivery.ModuleRefused{{Module: "web", Why: "no own secret"}}}}, nil, delivery.VerdictFail, "laptop refuses", ""},
{"a node not answering, within its bound", false, map[string]*delivery.CandidateAnswer{"anchor": same(7)}, map[string]bool{"laptop": true}, delivery.VerdictWaits, "laptop (offline)", ""},
{"a node not answering, within its bound", false, map[string]*delivery.CandidateAnswer{"anchor": same(7)}, map[string]bool{"laptop": true}, plc.VerdictWaits, "laptop (offline)", ""},
{"an answer of an earlier revision, past its bound, is no answer and decided without it", false, map[string]*delivery.CandidateAnswer{"anchor": same(6), "laptop": same(8)}, nil, delivery.VerdictPass, "no answer from anchor", ""},
{"a module held as today passes, shown in the brief", false, map[string]*delivery.CandidateAnswer{"anchor": same(7), "laptop": {Proposal: "p-one", Revision: 8,
Verdict: delivery.CandidateSame, Held: []delivery.ModuleRefused{{Module: "spiegel", Why: "a required value is unset"}}}}, nil, delivery.VerdictPass, "", ""},
modules/mesh-delivery-assigner/internal/placement/candidate.go
+96
View File
@@ -0,0 +1,96 @@
package placement
import (
"encoding/json"
"fmt"
"strings"
client "git.novox.be/novox/mesh-contracts/placement/go/clients/meshmoduleassignerclient"
seat "git.novox.be/novox/mesh-contracts/placement/go/seats/meshmoduleassigner"
"git.novox.be/novox/mesh-sdk/go/keys"
)
// Placement candidates (novox/hq ADR 0567 §3): for each node a proposal must hear from, the holder writes one key
// `<proposal>.<node>` in its state `candidates`: the node's placement with the proposal laid over it for a node whose key
// it changes, or the node's current key for one whose composition it changes only through another node (Reached). The
// layouts are the seat's contract's (CandidatesByProposalNodeValue); what refuses a candidate is here.
// CandidateKey is the key of one node's candidate in one proposal: `<proposal>.<node>`, so a node watches `*.<node>` and a
// proposal's keys are `<proposal>.*`.
func CandidateKey(proposal, node string) (string, error) {
if err := client.CheckProposalID(proposal); err != nil {
return "", err
}
if err := keys.CheckName("node", node); err != nil {
return "", err
}
return seat.CandidatesByProposalNodeKey(proposal, node), nil
}
// ParseCandidateKey is the inverse of CandidateKey; it refuses any other key, naming it.
func ParseCandidateKey(key string) (proposal, node string, err error) {
proposal, node, ok := strings.Cut(key, ".")
if !ok || keys.CheckName("node", node) != nil || client.CheckProposalID(proposal) != nil {
return "", "", fmt.Errorf("placement: %q is not a placement candidate key, <proposal>.<node>", key)
}
return proposal, node, nil
}
func checkCandidate(c seat.CandidatesByProposalNodeValue) error {
if err := client.CheckProposalID(c.Proposal); err != nil {
return err
}
if c.Placement != nil {
return client.CheckPlacement(*c.Placement)
}
return nil
}
// EncodeCandidate is c as JSON at the keys' Version; it refuses what DecodeCandidate refuses.
func EncodeCandidate(c seat.CandidatesByProposalNodeValue) ([]byte, error) {
if err := checkCandidate(c); err != nil {
return nil, err
}
c.V = keys.Version
if c.Placement != nil {
p := *c.Placement
p.V = keys.Version
c.Placement = &p
}
return json.Marshal(c)
}
// DecodeCandidate reads a candidate; it refuses an unknown version, a proposal id that is not one, and a placement the
// client refuses.
func DecodeCandidate(data []byte) (seat.CandidatesByProposalNodeValue, error) {
if err := keys.CheckVersion(data); err != nil {
return seat.CandidatesByProposalNodeValue{}, err
}
c, err := seat.DecodeCandidatesByProposalNodeValue(data)
if err != nil {
return seat.CandidatesByProposalNodeValue{}, fmt.Errorf("placement: a candidate does not fit its shape: %w", err)
}
if err := checkCandidate(c); err != nil {
return seat.CandidatesByProposalNodeValue{}, err
}
return c, nil
}
// PulseAnswerKey is the key of the bound on one node's answer to a proposal: `answer.<proposal>.<node>`.
func PulseAnswerKey(proposal, node string) (string, error) {
if err := client.CheckProposalID(proposal); err != nil {
return "", err
}
if err := keys.CheckName("node", node); err != nil {
return "", err
}
return seat.PulsesByAnswerProposalNodeKey(proposal, node), nil
}
// PulseAskKey is the key of the bound on a proposal's raised ask: `ask.<proposal>`.
func PulseAskKey(proposal string) (string, error) {
if err := client.CheckProposalID(proposal); err != nil {
return "", err
}
return seat.PulsesByAskProposalKey(proposal), nil
}