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

Assigner: the seat's state names, events, buckets and history are the contract's and the placement package's (D-12; ADR 0634 §1; issue 1966)

Browse Source 
Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
This commit is contained in:
jschoubben
2026-11-05 15:33:01 +01:00
parent 1266c1ada0
commit a3cb3e4a9b
13 changed files with 175 additions and 143 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go
+24 -24
View File
@@ -90,7 +90,7 @@ func startBed(t *testing.T, leaveOut string) *bed {
continue
}
cfg := jetstream.KeyValueConfig{Bucket: name, History: 5}
if name == delivery.PlacementPulsesBucket {
if name == plc.PulsesBucket {
cfg.History, cfg.LimitMarkerTTL = 1, time.Minute // a time to live per key, as the controller makes it (key-ttl)
}
kv, err := js.CreateKeyValue(ctx, cfg)
@@ -266,19 +266,19 @@ func TestTheAssignerDoesNotStartWithoutABucketItReads(t *testing.T) {
// Without the seat's placement state it starts all the same, and once the bucket is made (the controller makes
// it from the manifest) it is read with no restart: a node it names gets its own want, the node composing its own.
pb := startBed(t, delivery.PlacementBucket)
pb := startBed(t, plc.PlacementBucket)
pdone := pb.start(t)
select {
case err := <-pdone:
t.Fatalf("without %s the assigner stopped: %v", delivery.PlacementBucket, err)
t.Fatalf("without %s the assigner stopped: %v", plc.PlacementBucket, err)
case <-time.After(2 * time.Second):
}
placement, err := pb.js.CreateKeyValue(pb.ctx, jetstream.KeyValueConfig{Bucket: delivery.PlacementBucket, History: 5})
placement, err := pb.js.CreateKeyValue(pb.ctx, jetstream.KeyValueConfig{Bucket: plc.PlacementBucket, History: 5})
if err != nil {
t.Fatal(err)
}
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)
plkey, _ := mmac.LegacyKey("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)
}
@@ -314,7 +314,7 @@ func (b *bed) place(t *testing.T, node, module string) {
t.Fatal(err)
}
for key, raw := range values {
b.putJSON(t, delivery.PlacementBucket, key, raw)
b.putJSON(t, plc.PlacementBucket, key, raw)
}
}
@@ -486,8 +486,8 @@ func TestWantsAreWrittenAndNoDifferenceIs(t *testing.T) {
if err != nil {
t.Fatal(err)
}
pkey, _ := delivery.LegacyPlacementKey(node)
b.putJSON(t, delivery.PlacementBucket, pkey, pl)
pkey, _ := mmac.LegacyKey(node)
b.putJSON(t, plc.PlacementBucket, pkey, pl)
}
if w := b.wantNow(t, "ace", "web", "ace's want once placed: the build alone, composed by the node", func(w delivery.Want) bool { return w.Self }); w.Composed != "" || w.FP != fNew {
t.Errorf("ace's want once placed = %+v", w)
@@ -1126,7 +1126,7 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) {
t.Fatal(err)
}
t.Cleanup(nc.Close)
if _, err := nc.Subscribe(process.ModuleStreamSubject(delivery.AssignerModule, delivery.PlacementChangedEvent), func(m *nats.Msg) {
if _, err := nc.Subscribe(process.ModuleStreamSubject(delivery.AssignerModule, mma.EventPlacementChanged), func(m *nats.Msg) {
var e mma.PlacementEvent
if json.Unmarshal(m.Data, &e) == nil {
heard <- e
@@ -1150,18 +1150,18 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) {
values, _ := mmac.PlacementValues(p)
key, _ := mmac.NodeKey(node)
if legacy {
key, _ = delivery.LegacyPlacementKey(node)
key, _ = mmac.LegacyKey(node)
values = map[string][]byte{key: values[key]}
}
for _, k := range slices.Sorted(maps.Keys(values)) {
if k == key {
continue
}
if _, err := b.buckets[delivery.PlacementBucket].Put(b.ctx, k, values[k]); err != nil {
if _, err := b.buckets[plc.PlacementBucket].Put(b.ctx, k, values[k]); err != nil {
t.Fatal(err)
}
}
rev, err := b.buckets[delivery.PlacementBucket].Put(b.ctx, key, values[key])
rev, err := b.buckets[plc.PlacementBucket].Put(b.ctx, key, values[key])
if err != nil {
t.Fatal(err)
}
@@ -1176,9 +1176,9 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) {
// The holder converts anchor, held in the form before alone, to its node and module keys once at its start.
var anchorRev uint64
for deadline := time.Now().Add(15 * time.Second); anchorRev == 0; time.Sleep(50 * time.Millisecond) {
if e, err := b.buckets[delivery.PlacementBucket].Get(b.ctx, "anchor.node"); err == nil {
if e, err := b.buckets[plc.PlacementBucket].Get(b.ctx, "anchor.node"); err == nil {
anchorRev = e.Revision()
if _, err := b.buckets[delivery.PlacementBucket].Get(b.ctx, "anchor.web"); err != nil {
if _, err := b.buckets[plc.PlacementBucket].Get(b.ctx, "anchor.web"); err != nil {
t.Fatalf("anchor converted without its module key: %v", err)
}
}
@@ -1200,7 +1200,7 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) {
t.Helper()
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 e, err := b.buckets[plc.ProposalsBucket].Get(b.ctx, id); err == nil {
if p, err := plc.DecodeProposal(e.Value()); err == nil {
if last = p; ok(p) {
return p
@@ -1215,7 +1215,7 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) {
t.Helper()
for _, node := range p.Must {
key, _ := plc.CandidateKey(p.ID, node)
e, err := b.buckets[delivery.PlacementCandidatesBucket].Get(b.ctx, key)
e, err := b.buckets[plc.CandidatesBucket].Get(b.ctx, key)
if err != nil {
t.Fatalf("no candidate %s: %v", key, err)
}
@@ -1247,7 +1247,7 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) {
// The node key names the modules, each module key carries its own, and the mirror of the form before follows.
values := map[string][]byte{}
for _, key := range []string{node + ".node", "placement." + node, node + ".web", node + ".shell"} {
if e, err := b.buckets[delivery.PlacementBucket].Get(b.ctx, key); err == nil {
if e, err := b.buckets[plc.PlacementBucket].Get(b.ctx, key); err == nil {
values[key] = e.Value()
}
}
@@ -1260,7 +1260,7 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) {
t.Errorf("the mirror placement.%s = %v, want %v", node, mmac.Assigned(mirror), want)
}
}
if b.hasKey(delivery.PlacementProposalsBucket, p.ID) {
if b.hasKey(plc.ProposalsBucket, p.ID) {
t.Error("an applied proposal stays in the state")
}
@@ -1278,13 +1278,13 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) {
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) {
for deadline := time.Now().Add(15 * time.Second); b.hasKey(plc.ProposalsBucket, moved.ID); time.Sleep(50 * time.Millisecond) {
if time.Now().After(deadline) {
t.Fatal("a declined proposal was not ended")
}
}
key, _ := mmac.NodeKey("laptop")
if e, _ := b.buckets[delivery.PlacementBucket].Get(b.ctx, key); e == nil || !strings.Contains(string(e.Value()), "web") {
if e, _ := b.buckets[plc.PlacementBucket].Get(b.ctx, key); e == nil || !strings.Contains(string(e.Value()), "web") {
t.Error("a declined proposal wrote its key")
}
@@ -1320,10 +1320,10 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) {
case <-time.After(15 * time.Second):
t.Fatal("no placement-changed heard for the joining node")
}
if ace, err := b.buckets[delivery.PlacementBucket].Get(b.ctx, "ace.node"); err != nil || !strings.Contains(string(ace.Value()), "10.9.0.2") {
if ace, err := b.buckets[plc.PlacementBucket].Get(b.ctx, "ace.node"); err != nil || !strings.Contains(string(ace.Value()), "10.9.0.2") {
t.Errorf("the joining node's key = %v, %v", ace, err)
}
if akey, _ := delivery.PlacementAskKey(joined.(proposed).Proposal.ID); b.hasKey(delivery.AssignerAsksBucket, akey) || b.hasKey(delivery.PlacementProposalsBucket, joined.(proposed).Proposal.ID) {
if akey, _ := delivery.PlacementAskKey(joined.(proposed).Proposal.ID); b.hasKey(delivery.AssignerAsksBucket, akey) || b.hasKey(plc.ProposalsBucket, joined.(proposed).Proposal.ID) {
t.Error("an ask was raised for the joining node, or its proposal stays open")
}
// Its enrolment ends before it confirms: a remove-node on the same warrant deletes its key at once, with no ask, and
@@ -1343,7 +1343,7 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) {
case <-time.After(15 * time.Second):
t.Fatal("no placement-changed heard for the removed node")
}
if _, err := b.buckets[delivery.PlacementBucket].Get(b.ctx, "ace.node"); err == nil {
if _, err := b.buckets[plc.PlacementBucket].Get(b.ctx, "ace.node"); err == nil {
t.Error("the removed node's key stays")
}
}
@@ -1426,7 +1426,7 @@ func (s bedSource) Get(key string) (*views.Entry, bool, error) {
// placementSubjects are the subjects of the seat's four events on the assigner's stream placement.
func placementSubjects() []string {
var out []string
for _, e := range delivery.PlacementEvents {
for _, e := range mma.Events() {
out = append(out, process.ModuleStreamSubject(delivery.AssignerModule, e))
}
return out
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go
+19 -17
View File
@@ -23,6 +23,7 @@ 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"
mma "git.novox.be/novox/mesh-contracts/placement/go/seats/meshmoduleassigner"
"git.novox.be/novox/mesh-sdk/go/announce"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/health"
@@ -33,6 +34,7 @@ import (
"git.novox.be/novox/mesh-sdk/go/views"
"mesh-delivery-assigner/internal/assign"
plc "mesh-delivery-assigner/internal/placement"
)
func main() {
@@ -73,8 +75,8 @@ var readFamilies = []string{delivery.ComposedRead}
// ownBuckets are the assigner's own states, as its manifest declares them: the seat mesh-placement's three among them
// (ADR 0567 §1), named after the assigner as every module state is.
var ownBuckets = []string{delivery.AssignerAsksBucket, delivery.AssignerDifferencesBucket,
delivery.AssignerPutsBucket, delivery.AssignerWantsBucket, delivery.PlacementBucket, delivery.PlacementProposalsBucket,
delivery.PlacementCandidatesBucket, delivery.PlacementPulsesBucket, assign.FindingsBucket, assign.PoliciesBucket, delivery.SeatHoldersBucket,
delivery.AssignerPutsBucket, delivery.AssignerWantsBucket, plc.PlacementBucket, plc.ProposalsBucket,
plc.CandidatesBucket, plc.PulsesBucket, assign.FindingsBucket, assign.PoliciesBucket, plc.SeatHoldersBucket,
engineReadsBucket}
// engineReadsState is the assigner's state of what each node's engine states it reads (ADR 0618 §3), one key `<node>`, as
@@ -86,8 +88,8 @@ const engineReadsBucket = delivery.AssignerModule + "_" + engineReadsState
// seatBuckets are the seat's states, new with it: until the controller has made them from the manifest and the bus grants
// them, the assigner starts without them and binds each once the bus says it was created (issue 1790's kind).
var seatBuckets = []string{delivery.PlacementBucket, delivery.PlacementProposalsBucket, delivery.PlacementCandidatesBucket,
delivery.PlacementPulsesBucket}
var seatBuckets = []string{plc.PlacementBucket, plc.ProposalsBucket, plc.CandidatesBucket,
plc.PulsesBucket}
// keptStates is what the assigner keeps and who may read it (ADR 0495 §2-§3): its wants by the watchman and the view
// (and the node agent from release 6), its differences by the view, its asks by the view
@@ -95,17 +97,17 @@ var seatBuckets = []string{delivery.PlacementBucket, delivery.PlacementProposals
// candidates by the node agents, the grants part, mesh-config and the view (ADR 0567 §9), and its placement by
// the builder, the checker and the indexer, through mesh-sdk's toolchain gate (issue 774, ADR 0549 §3).
var keptStates = []announce.State{
{Name: delivery.PlacementState, Readable: true},
{Name: delivery.PlacementProposalsState, Readable: true},
{Name: delivery.PlacementCandidatesState, Readable: true},
{Name: delivery.PlacementPulsesState, Readable: true},
{Name: mma.StatePlacement, Readable: true},
{Name: mma.StateProposals, Readable: true},
{Name: mma.StateCandidates, Readable: true},
{Name: mma.StatePulses, Readable: true},
{Name: "asks", Readable: true},
{Name: "differences", Readable: true},
{Name: "puts", Readable: true},
{Name: "wants", Readable: true},
{Name: health.FindingsState, Readable: true},
{Name: delivery.AssignerPoliciesState, Readable: true},
{Name: delivery.SeatHoldersState, Readable: true},
{Name: mma.StateSeatHolders, Readable: true},
{Name: views.MarksState, Readable: true}, // what its views family said each state up to (views.go)
{Name: views.StatusState, Readable: true}, // its first views of the indexer's states (views.go)
{Name: engineReadsState, Readable: true},
@@ -131,7 +133,7 @@ func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any
logf("the bucket %s cannot be opened (%v): it is bound once the bus says it was made", name, err)
continue
}
if name == delivery.SeatHoldersBucket && err != nil {
if name == plc.SeatHoldersBucket && err != nil {
// The seat's state, new with ADR 0571 §4: bound once the bus says it was made; until then nobody is told who
// holds which seat here, and the controller's seats still answer.
logf("the bucket %s cannot be opened (%v): it is bound once the bus says it was made", name, err)
@@ -184,7 +186,7 @@ func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any
// states and those new with a delivery of their own, bound once made, and the two that choose the grants' source.
func startsWithout(name string) bool {
switch name {
case delivery.SeatHoldersBucket, provisions.ProvisionsBucket, assign.PoliciesBucket, assign.FindingsBucket,
case plc.SeatHoldersBucket, provisions.ProvisionsBucket, assign.PoliciesBucket, assign.FindingsBucket,
delivery.AssignerPutsBucket, delivery.SourceBucket, delivery.MergesBucket, engineReadsBucket:
return true
}
@@ -269,10 +271,10 @@ var watched = []string{
delivery.PartGrantsBucket,
delivery.AnswersBucket,
delivery.BuilderBuildsBucket,
delivery.PlacementBucket,
delivery.PlacementProposalsBucket,
delivery.PlacementCandidatesBucket,
delivery.PlacementPulsesBucket,
plc.PlacementBucket,
plc.ProposalsBucket,
plc.CandidatesBucket,
plc.PulsesBucket,
delivery.PresenceBucket,
delivery.HaveBucket,
delivery.AssignerAsksBucket,
@@ -280,7 +282,7 @@ var watched = []string{
delivery.AssignerWantsBucket,
assign.PoliciesBucket,
provisions.ProvisionsBucket,
delivery.SeatHoldersBucket,
plc.SeatHoldersBucket,
delivery.SourceBucket,
delivery.MergesBucket,
engineReadsBucket,
@@ -425,7 +427,7 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
}
}
// And the two that choose where grants are read from, with the same delivery as the grants part that writes one.
for _, name := range []string{delivery.PartGrantsBucket, assign.PoliciesBucket, provisions.ProvisionsBucket, delivery.SeatHoldersBucket} {
for _, name := range []string{delivery.PartGrantsBucket, assign.PoliciesBucket, provisions.ProvisionsBucket, plc.SeatHoldersBucket} {
if kvs[name] == nil && js != nil {
name := name
process.AwaitBucket(ctx, js, nc, name, "until then the bucket is not read",
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go
+2 -1
View File
@@ -2,6 +2,7 @@ package main
import (
"mesh-delivery-assigner/internal/assign"
plc "mesh-delivery-assigner/internal/placement"
"context"
"errors"
@@ -116,7 +117,7 @@ func TestOpensWhatItsManifestNames(t *testing.T) {
t.Fatal(err)
}
for key, raw := range placed {
kv := kvs[delivery.PlacementBucket].(*watchKV)
kv := kvs[plc.PlacementBucket].(*watchKV)
kv.entries = append(kv.entries, kvEntry{key: key, value: raw})
}
kvs[delivery.AssignerPutsBucket].(*watchKV).closed = true
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go
+6 -6
View File
@@ -1012,7 +1012,7 @@ func missingPulse(key string) string { return "provision-missing." + key }
// `provider.<node>.<module>.<provision>.provision-missing` is written; one keyed again ends both. The pulse's expiry, heard on
// the watch, is the pass that writes it: no timer.
func (p *passer) provisionsMissing(ctx context.Context, missing map[string]bool) {
pulses := p.kvs[delivery.PlacementPulsesBucket]
pulses := p.kvs[plc.PulsesBucket]
if p.missing == nil {
p.missing = map[string]bool{}
}
@@ -1048,7 +1048,7 @@ func (p *passer) provisionsMissing(ctx context.Context, missing map[string]bool)
delete(p.missing, key)
if pulses != nil && p.pulsing(missingPulse(key)) {
if err := pulses.Delete(ctx, missingPulse(key)); err == nil {
p.w.put(delivery.PlacementPulsesBucket, missingPulse(key), nil)
p.w.put(plc.PulsesBucket, missingPulse(key), nil)
}
}
if p.missingSaid[key] && p.findings.Delete != nil {
@@ -1103,7 +1103,7 @@ func (p *passer) unplacedFindings() {
// placement holds any more deleted. A pass follows each placement change and each manifest the indexer takes in, so no
// timer writes it.
func (p *passer) seatHolders(ctx context.Context) {
kv := p.kvs[delivery.SeatHoldersBucket]
kv := p.kvs[plc.SeatHoldersBucket]
if kv == nil {
return
}
@@ -1112,13 +1112,13 @@ func (p *passer) seatHolders(ctx context.Context) {
return // no placement read yet: nothing is known of who holds what, and nothing is deleted
}
want := plc.SeatHoldersOf(placements)
have := p.w.read(delivery.SeatHoldersBucket)
have := p.w.read(plc.SeatHoldersBucket)
for key, sh := range want {
value, err := json.Marshal(sh)
if err != nil || string(have[key]) == string(value) {
continue
}
p.put(ctx, delivery.SeatHoldersBucket, key, func() ([]byte, error) { return value, nil })
p.put(ctx, plc.SeatHoldersBucket, key, func() ([]byte, error) { return value, nil })
}
for key := range have {
if _, ok := want[key]; ok {
@@ -1128,7 +1128,7 @@ func (p *passer) seatHolders(ctx context.Context) {
p.logf("the seat holders %s are not deleted: %v", key, err)
continue
}
p.w.put(delivery.SeatHoldersBucket, key, nil)
p.w.put(plc.SeatHoldersBucket, key, nil)
}
}
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go
+3 -3
View File
@@ -71,7 +71,7 @@ func placeOn(t *testing.T, w *world, node string, modules ...string) {
t.Fatal(err)
}
for key, raw := range values {
w.put(delivery.PlacementBucket, key, raw)
w.put(plc.PlacementBucket, key, raw)
}
}
@@ -989,7 +989,7 @@ func aPlacedNodesWantsNameTheBuildAloneAndItsYesComesFromItsHave(t *testing.T) {
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, _ := mmac.EncodePlacement(mma.Placement{Node: "g14", Modules: []mma.PlacedModule{{Module: "web"}}})
put(delivery.PlacementBucket, "placement.g14", pl)
put(plc.PlacementBucket, "placement.g14", pl)
on, _ := delivery.EncodePresence(delivery.Presence{Online: true})
put(delivery.PresenceBucket, "g14", on)
have, _ := delivery.EncodeHave(delivery.Have{FP: fpOld, Composed: was, Phase: delivery.PhaseRunning, Health: delivery.HealthOK})
@@ -1076,7 +1076,7 @@ func TestAnApprovedPlacementCallsNoControllerVerb(t *testing.T) {
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}) {
if len(bus.calls) != 0 || !slices.Equal(said, []string{mma.EventPlacementChanged}) {
t.Fatalf("calls = %d, said = %v", len(bus.calls), said)
}
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go
+31 -31
View File
@@ -147,7 +147,7 @@ func (p *passer) seatPlacements() (map[string]mma.Placement, 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) {
for key, value := range p.w.read(plc.CandidatesBucket) {
pid, node, err := plc.ParseCandidateKey(key)
if err != nil || pid != id {
continue
@@ -156,7 +156,7 @@ func (p *passer) candidatesOf(id string) (map[string]mma.CandidatesByProposalNod
if err != nil {
continue
}
meta := p.w.metaOf(delivery.PlacementCandidatesBucket, key)
meta := p.w.metaOf(plc.CandidatesBucket, key)
out[node], revs[node] = c, meta.revision
if meta.at.After(newest) {
newest = meta.at
@@ -197,7 +197,7 @@ func (p *passer) placementSeat(ctx context.Context) {
p.convertPlacement(ctx)
current, revs := p.seatPlacements()
ids := make([]string, 0)
raws := p.w.read(delivery.PlacementProposalsBucket)
raws := p.w.read(plc.ProposalsBucket)
for id := range raws {
ids = append(ids, id)
}
@@ -208,32 +208,32 @@ func (p *passer) placementSeat(ctx context.Context) {
p.logf("the proposal %s cannot be read, so it is left: %v", id, err)
continue
}
p.stepProposal(ctx, prop, p.w.metaOf(delivery.PlacementProposalsBucket, id).revision, current, revs)
p.stepProposal(ctx, prop, p.w.metaOf(plc.ProposalsBucket, id).revision, current, revs)
}
}
// pulse writes a bound's key with its time to live (§4): its expiry, heard on the watch as the server's delete marker,
// is the event that ends the wait. Written again over one that stands, so its time starts anew.
func (p *passer) pulse(ctx context.Context, key string, ttl time.Duration) {
kv := p.kvs[delivery.PlacementPulsesBucket]
kv := p.kvs[plc.PulsesBucket]
_ = kv.Delete(ctx, key)
rev, err := kv.Create(ctx, key, []byte(`{}`), jetstream.KeyTTL(ttl))
if err != nil {
p.logf("the bound %s is not written, so its wait ends at once: %v", key, err)
return
}
p.w.putEntry(delivery.PlacementPulsesBucket, key, []byte(`{}`), rev, p.clock())
p.w.putEntry(plc.PulsesBucket, key, []byte(`{}`), rev, p.clock())
}
// pulsing says whether a bound's key stands.
func (p *passer) pulsing(key string) bool { return p.w.has(delivery.PlacementPulsesBucket, key) }
func (p *passer) pulsing(key string) bool { return p.w.has(plc.PulsesBucket, key) }
// deletePulses deletes every bound of a proposal.
func (p *passer) deletePulses(ctx context.Context, id string) {
for key := range p.w.read(delivery.PlacementPulsesBucket) {
for key := range p.w.read(plc.PulsesBucket) {
if strings.HasSuffix(key, "."+id) || strings.Contains(key, "."+id+".") {
if err := p.kvs[delivery.PlacementPulsesBucket].Delete(ctx, key); err == nil {
p.w.put(delivery.PlacementPulsesBucket, key, nil)
if err := p.kvs[plc.PulsesBucket].Delete(ctx, key); err == nil {
p.w.put(plc.PulsesBucket, key, nil)
}
}
}
@@ -262,7 +262,7 @@ func (p *passer) ask(ctx context.Context, prop mma.Proposal, akey, brief string)
// stepProposal takes one proposal one step further.
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.sayEvent(mma.EventPlacementProposed, mma.PlacementEvent{Proposal: prop}, prop.ID)
p.saidProposed[prop.ID] = true
}
if prop.Withdrawn != "" {
@@ -339,7 +339,7 @@ func (p *passer) stepProposal(ctx context.Context, prop mma.Proposal, rev uint64
}
}
if p.writeProposal(ctx, prop, rev) {
p.sayEvent(delivery.PlacementJudgedEvent, mma.PlacementEvent{Proposal: prop},
p.sayEvent(mma.EventPlacementJudged, mma.PlacementEvent{Proposal: prop},
fmt.Sprintf("%s.judged.%d", prop.ID, v.At.UnixNano()))
}
}
@@ -413,7 +413,7 @@ func (p *passer) mustAnswer(ctx context.Context, prop mma.Proposal, rev uint64,
if pl, ok := current[node]; ok {
c.Placement = &pl
}
p.put(ctx, delivery.PlacementCandidatesBucket, key, func() ([]byte, error) { return plc.EncodeCandidate(c) })
p.put(ctx, plc.CandidatesBucket, key, func() ([]byte, error) { return plc.EncodeCandidate(c) })
}
for _, node := range must {
if k, err := plc.PulseAnswerKey(prop.ID, node); err == nil {
@@ -439,7 +439,7 @@ func (p *passer) layAgain(ctx context.Context, prop mma.Proposal, rev uint64, cu
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) })
p.put(ctx, plc.CandidatesBucket, key, func() ([]byte, error) { return plc.EncodeCandidate(c) })
}
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)
@@ -514,8 +514,8 @@ func (p *passer) warranted(ctx context.Context, prop mma.Proposal, rev uint64, a
}
p.deleteCandidates(ctx, prop.ID)
p.deletePulses(ctx, prop.ID)
_ = p.kvs[delivery.PlacementProposalsBucket].Delete(ctx, prop.ID)
p.w.put(delivery.PlacementProposalsBucket, prop.ID, nil)
_ = p.kvs[plc.ProposalsBucket].Delete(ctx, prop.ID)
p.w.put(plc.ProposalsBucket, prop.ID, nil)
p.followWrite(prop, answer, changes)
}
@@ -550,8 +550,8 @@ func (p *passer) addNode(ctx context.Context, prop mma.Proposal, revs map[string
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)
p.w.put(delivery.PlacementProposalsBucket, prop.ID, nil)
_ = p.kvs[plc.ProposalsBucket].Delete(ctx, prop.ID)
p.w.put(plc.ProposalsBucket, prop.ID, nil)
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)},
@@ -577,8 +577,8 @@ func (p *passer) removeNode(ctx context.Context, prop mma.Proposal, current map[
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)
p.w.put(delivery.PlacementProposalsBucket, prop.ID, nil)
_ = p.kvs[plc.ProposalsBucket].Delete(ctx, prop.ID)
p.w.put(plc.ProposalsBucket, prop.ID, nil)
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)},
@@ -616,7 +616,7 @@ func (p *passer) sayChanged(prop mma.Proposal, answer delivery.Answer, changes [
}
}
plc.SortChanges(changes)
p.sayEvent(delivery.PlacementChangedEvent, mma.PlacementEvent{Proposal: prop, Nodes: changeBodies(changes),
p.sayEvent(mma.EventPlacementChanged, mma.PlacementEvent{Proposal: prop, Nodes: changeBodies(changes),
Warrant: &mma.PlacementEventWarrant{Warrant: answer.Warrant, Words: answer.Words, At: at}}, prop.ID+".changed")
}
@@ -630,21 +630,21 @@ func (p *passer) endProposal(ctx context.Context, prop mma.Proposal, why string)
}
p.deleteCandidates(ctx, prop.ID)
p.deletePulses(ctx, prop.ID)
if err := p.kvs[delivery.PlacementProposalsBucket].Delete(ctx, prop.ID); err != nil && !errors.Is(err, context.Canceled) {
if err := p.kvs[plc.ProposalsBucket].Delete(ctx, prop.ID); err != nil && !errors.Is(err, context.Canceled) {
p.logf("the proposal %s is not deleted, and is ended at the next pass: %v", prop.ID, err)
return
}
p.w.put(delivery.PlacementProposalsBucket, prop.ID, nil)
p.w.put(plc.ProposalsBucket, prop.ID, nil)
p.logf("the proposal %s ends: %s", prop.ID, why)
p.sayEvent(delivery.PlacementWithdrawnEvent, mma.PlacementEvent{Proposal: prop, Ended: why}, prop.ID+".withdrawn")
p.sayEvent(mma.EventPlacementWithdrawn, 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) {
for key := range p.w.read(plc.CandidatesBucket) {
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)
if err := p.kvs[plc.CandidatesBucket].Delete(ctx, key); err == nil {
p.w.put(plc.CandidatesBucket, key, nil)
}
}
}
@@ -658,7 +658,7 @@ func (p *passer) writeProposal(ctx context.Context, prop mma.Proposal, rev uint6
p.logf("the proposal %s is not written: %v", prop.ID, err)
return false
}
kv := p.kvs[delivery.PlacementProposalsBucket]
kv := p.kvs[plc.ProposalsBucket]
var at uint64
if rev == 0 {
at, err = kv.Create(ctx, prop.ID, raw)
@@ -671,7 +671,7 @@ func (p *passer) writeProposal(ctx context.Context, prop mma.Proposal, rev uint6
}
return false
}
p.w.putEntry(delivery.PlacementProposalsBucket, prop.ID, raw, at, p.clock())
p.w.putEntry(plc.ProposalsBucket, prop.ID, raw, at, p.clock())
return true
}
@@ -697,7 +697,7 @@ func (p *passer) writeNode(ctx context.Context, node string, after *mma.Placemen
if err != nil {
return 0, err
}
kv := p.kvs[delivery.PlacementBucket]
kv := p.kvs[plc.PlacementBucket]
committed, rev := false, uint64(0)
for _, w := range writes {
got, err := writeKey(ctx, kv, w)
@@ -708,7 +708,7 @@ func (p *passer) writeNode(ctx context.Context, node string, after *mma.Placemen
}
return 0, fmt.Errorf("placement %s at revision %d: %w", w.Key, w.Base, err)
}
p.w.putEntry(delivery.PlacementBucket, w.Key, w.Value, got, p.clock())
p.w.putEntry(plc.PlacementBucket, w.Key, w.Value, got, p.clock())
if w.Commit {
committed, rev = true, got
}
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools.go
+16 -16
View File
@@ -64,7 +64,7 @@ type keyed struct {
// 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 (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)
keys, err := s.st.keys(mma.StatePlacement)
if err != nil {
return nil, nil, fmt.Errorf("placement cannot be listed: %w", err)
}
@@ -75,7 +75,7 @@ func (s *seatTools) placements() (map[string]mma.Placement, map[string]uint64, e
if _, _, _, err := mmac.ParseKey(key); err != nil {
continue
}
raw, rev, found, err := s.st.get(delivery.PlacementState, key)
raw, rev, found, err := s.st.get(mma.StatePlacement, key)
if err != nil {
return nil, nil, fmt.Errorf("placement key %s cannot be read: %w", key, err)
}
@@ -129,7 +129,7 @@ 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)
raw, _, found, err := s.st.get(mma.StateProposals, id)
if err != nil {
return mma.Proposal{}, fmt.Errorf("the proposal %s cannot be read: %w", id, err)
}
@@ -152,7 +152,7 @@ func (s *seatTools) write(p mma.Proposal, laid map[string]*mma.Placement) error
if err != nil {
return err
}
if err := s.st.put(delivery.PlacementCandidatesState, key, raw); err != nil {
if err := s.st.put(mma.StateCandidates, key, raw); err != nil {
return fmt.Errorf("the candidate %s is not written: %w", key, err)
}
}
@@ -160,7 +160,7 @@ func (s *seatTools) write(p mma.Proposal, laid map[string]*mma.Placement) error
if err != nil {
return err
}
if err := s.st.put(delivery.PlacementProposalsState, p.ID, raw); err != nil {
if err := s.st.put(mma.StateProposals, p.ID, raw); err != nil {
return fmt.Errorf("the proposal %s is not written: %w", p.ID, err)
}
return nil
@@ -290,7 +290,7 @@ func (s *seatTools) doWithdraw(args map[string]any, by string) (any, error) {
if err != nil {
return nil, err
}
if err := s.st.put(delivery.PlacementProposalsState, p.ID, raw); err != nil {
if err := s.st.put(mma.StateProposals, p.ID, raw); err != nil {
return nil, fmt.Errorf("the proposal %s is not withdrawn: %w", p.ID, err)
}
return map[string]any{"proposal": p.ID, "withdrawn": p.Withdrawn}, nil
@@ -308,7 +308,7 @@ func (s *seatTools) doAgain(args map[string]any) (any, error) {
if err != nil {
return nil, err
}
if err := s.st.put(delivery.PlacementProposalsState, p.ID, raw); err != nil {
if err := s.st.put(mma.StateProposals, p.ID, raw); err != nil {
return nil, fmt.Errorf("the proposal %s is not asked again: %w", p.ID, err)
}
return map[string]any{"proposal": p.ID, "again": p.Again, "follow": followProposal(p.ID)}, nil
@@ -344,7 +344,7 @@ func (s *seatTools) doPlacement(args map[string]any) (any, error) {
return nil, err
}
node, module := arg(args, "node"), arg(args, "module")
out := placementAnswer{Keys: map[string]keyed{}, Count: len(all), History: delivery.PlacementHistory}
out := placementAnswer{Keys: map[string]keyed{}, Count: len(all), History: plc.PlacementHistory}
for n, p := range all {
if node != "" && n != node {
continue
@@ -396,7 +396,7 @@ func (s *seatTools) doPlacement(args map[string]any) (any, error) {
}
}
out.Older = fmt.Sprintf("the state keeps %d revisions of each key; an older change is in its placement-changed event on "+
"the assigner's stream %s, kept %d days", delivery.PlacementHistory, delivery.PlacementStream, delivery.PlacementStreamDays)
"the assigner's stream %s, kept %d days", plc.PlacementHistory, plc.Stream, plc.StreamDays)
return out, nil
}
@@ -411,11 +411,11 @@ type openProposal struct {
// doProposals is the verb proposals (§10).
func (s *seatTools) doProposals(args map[string]any) (any, error) {
ids, err := s.st.keys(delivery.PlacementProposalsState)
ids, err := s.st.keys(mma.StateProposals)
if err != nil {
return nil, fmt.Errorf("the proposals cannot be listed: %w", err)
}
cands, err := s.st.keys(delivery.PlacementCandidatesState)
cands, err := s.st.keys(mma.StateCandidates)
if err != nil {
return nil, fmt.Errorf("the candidates cannot be listed: %w", err)
}
@@ -460,7 +460,7 @@ func (s *seatTools) doProposals(args map[string]any) (any, error) {
if err != nil || pid != id {
continue
}
if _, rev, found, err := s.st.get(delivery.PlacementCandidatesState, k); err == nil && found {
if _, rev, found, err := s.st.get(mma.StateCandidates, k); err == nil && found {
o.Candidates[node] = rev
}
}
@@ -492,14 +492,14 @@ func (s *seatTools) tools() []stdio.Tool {
// seats is the verb placement with seats (ADR 0571 §4): who holds each seat and on which node, as the state seat-holders
// holds it; with node, the keys of seats held on that node.
func (s *seatTools) seats(node string) (any, error) {
keys, err := s.st.keys(delivery.SeatHoldersState)
keys, err := s.st.keys(mma.StateSeatHolders)
if err != nil {
return nil, fmt.Errorf("seat-holders cannot be listed: %w", err)
}
sort.Strings(keys)
out := []mma.SeatHolders{}
for _, key := range keys {
raw, _, found, err := s.st.get(delivery.SeatHoldersState, key)
raw, _, found, err := s.st.get(mma.StateSeatHolders, key)
if err != nil {
return nil, fmt.Errorf("seat-holders %s cannot be read: %w", key, err)
}
@@ -528,7 +528,7 @@ type keyedRevision struct {
// `placement.<node>` before it was converted, then its node key's `<node>.node`, whose every revision is one write of
// the node (research 072 row 37: the node key is a write's commit, so its history is the node's placement history).
func (s *seatTools) nodeHistory(node string) ([]keyedRevision, error) {
legacy, err := delivery.LegacyPlacementKey(node)
legacy, err := mmac.LegacyKey(node)
if err != nil {
return nil, err
}
@@ -538,7 +538,7 @@ func (s *seatTools) nodeHistory(node string) ([]keyedRevision, error) {
}
var out []keyedRevision
for _, key := range []string{legacy, nodeKey} {
revisions, err := s.st.history(delivery.PlacementState, key)
revisions, err := s.st.history(mma.StatePlacement, key)
if err != nil {
return nil, fmt.Errorf("the history of %s cannot be read: %w", key, err)
}
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go
+24 -24
View File
@@ -44,7 +44,7 @@ func (s memStore) history(state, key string) ([]stdio.StateRevision, error) {
func TestTheSeatsVerbs(t *testing.T) {
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} {
for _, state := range []string{mma.StatePlacement, mma.StateProposals, mma.StateCandidates, modulesState, haveState} {
st[state] = &memKV{values: map[string][]byte{}}
}
for _, m := range []string{"web", "shell"} {
@@ -62,7 +62,7 @@ func TestTheSeatsVerbs(t *testing.T) {
s.now = func() time.Time { return time.Date(2026, 10, 28, 20, 0, 0, 0, time.UTC) }
raw, _ := mmac.EncodePlacement(anchor)
st[delivery.PlacementState].values["placement.anchor"] = raw
st[mma.StatePlacement].values["placement.anchor"] = raw
// withdraw and again mark the proposal; the holder ends or judges it.
got, err := s.doPropose(map[string]any{"why": "hold", "changes": `[{"op":"hold","node":"anchor","module":"web","claim":"node-shell","scope":"node"}]`}, "op")
@@ -82,12 +82,12 @@ func TestTheSeatsVerbs(t *testing.T) {
if _, err := s.doWithdraw(map[string]any{"id": "p-none", "why": "x"}, "op"); err == nil {
t.Error("an unknown proposal was withdrawn")
}
delete(st[delivery.PlacementProposalsState].values, seed.ID) // the holder ended it
st[delivery.PlacementCandidatesState].values = map[string][]byte{}
delete(st[mma.StateProposals].values, seed.ID) // the holder ended it
st[mma.StateCandidates].values = map[string][]byte{}
// propose: refused whole, naming every finding; a valid move laid over the keys at their revisions.
if _, err := s.doPropose(map[string]any{"why": "x", "changes": `[{"op":"place","node":"anchor","module":"ghost"},{"op":"hold","node":"anchor","module":"web","claim":"x"}]`}, "op"); err == nil ||
!strings.Contains(err.Error(), "ghost") || len(st[delivery.PlacementProposalsState].values) != 0 {
!strings.Contains(err.Error(), "ghost") || len(st[mma.StateProposals].values) != 0 {
t.Errorf("a refused proposal: %v", err)
}
if _, err := s.doPropose(map[string]any{"why": "x", "changes": `[{"op":"swap","node":"anchor"}]`}, "op"); err == nil {
@@ -99,17 +99,17 @@ func TestTheSeatsVerbs(t *testing.T) {
t.Fatal(err)
}
move := got.(proposed).Proposal
if move.By != "operator@laptop" || move.Base["anchor"] != int64(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[mma.StateCandidates].values) != 2 {
t.Fatalf("propose = %+v", move)
}
// placement and proposals read every state the seat keeps (§10); history names the stream for what is older.
reads := map[string]bool{}
pa, err := s.doPlacement(map[string]any{"module": "web", "history": "true"})
if a := pa.(placementAnswer); err != nil || !slices.Equal(a.Nodes, []string{"anchor"}) || len(a.Changes) != 1 || !strings.Contains(a.Older, delivery.PlacementStream) {
if a := pa.(placementAnswer); err != nil || !slices.Equal(a.Nodes, []string{"anchor"}) || len(a.Changes) != 1 || !strings.Contains(a.Older, plc.Stream) {
t.Errorf("placement of web = %+v, %v", pa, err)
} else {
reads[delivery.PlacementState], reads["stream "+delivery.PlacementStream] = true, true
reads[mma.StatePlacement], reads["stream "+plc.Stream] = true, true
}
if _, err := s.doPlacement(map[string]any{"node": "laptop"}); err == nil {
t.Error("a node with no key was answered")
@@ -118,14 +118,14 @@ func TestTheSeatsVerbs(t *testing.T) {
if list := ps.(map[string]any)["proposals"].([]openProposal); err != nil || len(list) != 1 || len(list[0].Candidates) != 2 {
t.Errorf("proposals = %+v, %v", ps, err)
} else {
reads[delivery.PlacementProposalsState], reads[delivery.PlacementCandidatesState] = true, true
reads[mma.StateProposals], reads[mma.StateCandidates] = true, true
}
raw0, err := readManifest()
if err != nil {
t.Fatal(err)
}
for _, state := range raw0.State {
if slices.Contains([]string{delivery.PlacementState, delivery.PlacementProposalsState, delivery.PlacementCandidatesState}, state.Name) && !reads[state.Name] {
if slices.Contains([]string{mma.StatePlacement, mma.StateProposals, mma.StateCandidates}, state.Name) && !reads[state.Name] {
t.Errorf("no verb reads the seat's state %s", state.Name)
}
}
@@ -165,7 +165,7 @@ func TestTheSeatsVerbs(t *testing.T) {
// 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, _ := 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
st[mma.StatePlacement].values["placement.joiner"] = joiner
removeNode := func(node, warrant string) (any, error) {
return s.doPropose(map[string]any{"why": "enrolment ended", "warrant": warrant,
"changes": []any{map[string]any{"op": "remove-node", "node": node}}}, "novox.mesh-enrolment")
@@ -192,8 +192,8 @@ func TestTheSeatsVerbs(t *testing.T) {
w := newWorld()
fp := delivery.SourcePrefix2 + strings.Repeat("2", 64)
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())
pkey, _ := mmac.LegacyKey("ace")
w.putEntry(plc.PlacementBucket, pkey, pl, 1, time.Now())
for node, digest := range map[string]string{"ace": "sha256:" + strings.Repeat("a", 64), "g14": ""} {
hn, _ := delivery.EncodeHaveNode(delivery.HaveNode{Composed: digest})
hk, _ := delivery.HaveNodeKey(node)
@@ -214,8 +214,8 @@ func TestTheSeatsVerbs(t *testing.T) {
t.Fatalf("findings = %v", findings)
}
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())
pkey2, _ := mmac.LegacyKey("g14")
w.putEntry(plc.PlacementBucket, pkey2, pl2, 1, time.Now())
p.unplacedFindings()
if len(findings) != 0 {
t.Errorf("placed again: findings %v", findings)
@@ -224,14 +224,14 @@ func TestTheSeatsVerbs(t *testing.T) {
t.Run("1955/L7 (ADR 0571 §4): seat-holders is written from placement, only the keys a change moves, and answered by placement seats", func(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}}
p := &passer{w: newWorld(), logf: t.Logf, kvs: buckets{plc.SeatHoldersBucket: kv}}
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())
k, _ := mmac.LegacyKey(node)
p.w.putEntry(plc.PlacementBucket, k, pl, 1, time.Now())
}
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"})
@@ -249,7 +249,7 @@ func TestTheSeatsVerbs(t *testing.T) {
if writes != 3 || kv.values["node-shell.g14"] != nil || kv.values["node-shell.ace"] == nil {
t.Errorf("after g14's change: %d writes, %v", writes, kv.values)
}
st := memStore{delivery.SeatHoldersState: &memKV{values: kv.values}}
st := memStore{mma.StateSeatHolders: &memKV{values: kv.values}}
out, err := newSeatTools(st, nil).seats("ace")
if err != nil || len(out.(map[string]any)["seats"].([]mma.SeatHolders)) != 2 {
t.Errorf("seats on ace = %+v, %v", out, err)
@@ -259,10 +259,10 @@ func TestTheSeatsVerbs(t *testing.T) {
ctx := context.Background()
kv := &revKV{values: map[string][]byte{}, revs: map[string]uint64{}}
var logs []string
p := &passer{w: newWorld(), kvs: buckets{delivery.PlacementBucket: kv}, now: time.Now,
p := &passer{w: newWorld(), kvs: buckets{plc.PlacementBucket: kv}, now: time.Now,
logf: func(f string, a ...any) { logs = append(logs, fmt.Sprintf(f, a...)) }}
put := func(key string, value []byte, rev uint64) {
p.w.putEntry(delivery.PlacementBucket, key, value, rev, time.Now())
p.w.putEntry(plc.PlacementBucket, key, value, rev, time.Now())
}
was := mma.Placement{Node: "anchor", Modules: []mma.PlacedModule{{Module: "web"}}}
raw, _ := mmac.EncodePlacement(was)
@@ -298,7 +298,7 @@ func TestTheSeatsVerbs(t *testing.T) {
}
kv.refuse = ""
// The verb's history reads the form before, then the node key.
st := memStore{delivery.PlacementState: &memKV{values: kv.values}}
st := memStore{mma.StatePlacement: &memKV{values: kv.values}}
hs := newSeatTools(st, nil)
hs.st = historyStore{memStore: st, revisions: map[string][]stdio.StateRevision{
"placement.anchor": {{Key: "placement.anchor", Value: raw, Revision: 1, Op: "put"}},
@@ -342,7 +342,7 @@ func TestTheSeatsVerbs(t *testing.T) {
}
// A required provision with no key: pulsed with the bound first, said once the pulse expired, ended once keyed.
ctx := context.Background()
p.kvs[delivery.PlacementPulsesBucket] = &memKV{values: map[string][]byte{}}
p.kvs[plc.PulsesBucket] = &memKV{values: map[string][]byte{}}
gone := map[string]bool{"ace.photos.s3": true}
p.provisionsMissing(ctx, gone)
if !p.pulsing("provision-missing.ace.photos.s3") || len(findings) != 0 {
@@ -352,7 +352,7 @@ func TestTheSeatsVerbs(t *testing.T) {
if len(findings) != 0 {
t.Fatal("said before its bound")
}
p.w.put(delivery.PlacementPulsesBucket, "provision-missing.ace.photos.s3", nil) // the server expired it
p.w.put(plc.PulsesBucket, "provision-missing.ace.photos.s3", nil) // the server expired it
p.provisionsMissing(ctx, gone)
if _, ok := findings[assign.ProvisionMissingKey("ace.photos.s3")]; !ok {
t.Fatalf("past its bound: findings %v", findings)
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go
+13 -11
View File
@@ -9,6 +9,7 @@ import (
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient"
mma "git.novox.be/novox/mesh-contracts/placement/go/seats/meshmoduleassigner"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/health"
@@ -21,6 +22,7 @@ import (
"git.novox.be/novox/mesh-delivery/sdk/heard"
"mesh-delivery-assigner/internal/assign"
plc "mesh-delivery-assigner/internal/placement"
)
// marksBucket is the assigner's state of what its views family said each state up to.
@@ -32,15 +34,15 @@ type viewedTool struct{ tool, state, words string }
// firstViewTools are the assigner's first-view tools (ADR 0621 §1): each answers only its callers list
// (views.CallersEnv), which the manifest writes.
var firstViewTools = []viewedTool{
{"placement", delivery.PlacementState, "where each module is placed: the seat mesh-placement's placement, by module."},
{"seat-holders", delivery.SeatHoldersState, "who holds which seat on which node."},
{"placement", mma.StatePlacement, "where each module is placed: the seat mesh-placement's placement, by module."},
{"seat-holders", mma.StateSeatHolders, "who holds which seat on which node."},
{"wants", "wants", "what each node should run: want.<node>.<module>, its fingerprint and the rule that gave it."},
}
// readTools are the assigner's ordinary read tools, one per state its readers read (design 25 § 12.2).
var readTools = []viewedTool{
{"assigner_read_placement", delivery.PlacementState, "where each module is placed, by module."},
{"assigner_read_seat_holders", delivery.SeatHoldersState, "who holds which seat on which node."},
{"assigner_read_placement", mma.StatePlacement, "where each module is placed, by module."},
{"assigner_read_seat_holders", mma.StateSeatHolders, "who holds which seat on which node."},
{"assigner_read_wants", "wants", "what each node should run: want.<node>.<module>."},
{"assigner_read_findings", health.FindingsState, "the assigner's findings, which the condition keeper raises and clears."},
{"assigner_read_asks", "asks", "the asks the assigner put to a person, and their answers."},
@@ -49,12 +51,12 @@ var readTools = []viewedTool{
// viewedStates are the states the views family says, each by its bucket: the six its readers read.
var viewedStates = map[string]string{
delivery.PlacementState: delivery.PlacementBucket,
delivery.SeatHoldersState: delivery.SeatHoldersBucket,
"wants": delivery.AssignerWantsBucket,
health.FindingsState: assign.FindingsBucket,
"asks": delivery.AssignerAsksBucket,
"differences": delivery.AssignerDifferencesBucket,
mma.StatePlacement: plc.PlacementBucket,
mma.StateSeatHolders: plc.SeatHoldersBucket,
"wants": delivery.AssignerWantsBucket,
health.FindingsState: assign.FindingsBucket,
"asks": delivery.AssignerAsksBucket,
"differences": delivery.AssignerDifferencesBucket,
}
// viewTools are the assigner's first-view and read tools, reading only its own states, so they answer as soon as its
@@ -87,7 +89,7 @@ func startViews(ctx context.Context, js jetstream.JetStream, nc process.Subscrib
}
process.AwaitBucket(ctx, js, nc, marksBucket, "no change is said on the views family, so every reader waits on its first view", logf).Ready(func(mk jetstream.KeyValue) {
em := viewsEmitter(js, account, mk, findings, tooLarge, logf)
for _, state := range []string{"asks", "differences", health.FindingsState, delivery.PlacementState, delivery.SeatHoldersState, "wants"} {
for _, state := range []string{"asks", "differences", health.FindingsState, mma.StatePlacement, mma.StateSeatHolders, "wants"} {
state := state
start := func(kv jetstream.KeyValue) {
if err := em.Start(ctx, state, kv, errs); err != nil {
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go
+5 -4
View File
@@ -17,6 +17,7 @@ import (
"git.novox.be/novox/mesh-sdk/go/delivery"
"mesh-delivery-assigner/internal/assign"
plc "mesh-delivery-assigner/internal/placement"
)
// noModuleWords is the indexer's reason for a module of a candidate whose directory the merge preview holds no manifest
@@ -74,7 +75,7 @@ func (w *world) putEntry(bucket, key string, value []byte, revision uint64, at t
return
}
w.meta[bucket+"|"+key] = entryMeta{revision: revision, at: at}
if bucket == delivery.PlacementBucket {
if bucket == plc.PlacementBucket {
_, _ = w.placement.Put(key, value, revision, false) // put took it already; this is its revision
}
if bucket == delivery.SourceBucket && key == delivery.SourceMainKey {
@@ -134,14 +135,14 @@ func (w *world) put(bucket, key string, value []byte) {
w.changes[bucket]++ // the bucket's generation (bucketGen), a delete included
if value == nil {
delete(w.values[bucket], key)
if bucket == delivery.PlacementBucket {
if bucket == plc.PlacementBucket {
_, _ = w.placement.Put(key, nil, 0, true)
}
return
}
w.values[bucket][key] = value
w.changes[bucket+"|"+key]++
if bucket == delivery.PlacementBucket {
if bucket == plc.PlacementBucket {
// A key or value that does not read places nothing new on its node; the indexer's read tool says it.
_, _ = w.placement.Put(key, value, 0, false)
}
@@ -266,7 +267,7 @@ func (w *world) placed() (map[string][]string, bool) {
for node, p := range w.placements() {
out[node] = mmac.Assigned(p)
}
return out, len(w.read(delivery.PlacementBucket)) > 0
return out, len(w.read(plc.PlacementBucket)) > 0
}
// placements is every node's placement, assembled from its keys in either form (mmac.Keys).
modules/mesh-delivery-assigner/internal/assign/seat_test.go
+2 -2
View File
@@ -83,10 +83,10 @@ func TestThePlacementSeatDecides(t *testing.T) {
}
// The seat says it waits while one of its states is not bound, naming it, and clears once all are.
for _, unbound := range [][]string{{delivery.PlacementPulsesBucket}, nil} {
for _, unbound := range [][]string{{plc.PulsesBucket}, nil} {
c := SeatWaitsCondition(unbound)
if _, err := occ.EncodeCondition(c); err != nil || (len(unbound) > 0) != (c.State == occ.StateRaised) ||
(len(unbound) > 0 && !strings.Contains(c.Explanation, "the seat waits for its states: "+delivery.PlacementPulsesBucket)) {
(len(unbound) > 0 && !strings.Contains(c.Explanation, "the seat waits for its states: "+plc.PulsesBucket)) {
t.Errorf("waiting for %v: %+v, %v", unbound, c, err)
}
}
modules/mesh-delivery-assigner/internal/placement/state.go
+25
View File
@@ -0,0 +1,25 @@
package placement
import seat "git.novox.be/novox/mesh-contracts/placement/go/seats/meshmoduleassigner"
// Module is the module that holds the seat mesh-module-assigner: the states of the seat are named after it, as every
// module state is (a bucket is `<module>_<state>`), and its manifest declares them.
const Module = "mesh-delivery-assigner"
// The buckets of the seat's states, which only the holder writes (ADR 0567 §1).
const (
PlacementBucket = Module + "_" + seat.StatePlacement
ProposalsBucket = Module + "_" + seat.StateProposals
CandidatesBucket = Module + "_" + seat.StateCandidates
PulsesBucket = Module + "_" + seat.StatePulses
SeatHoldersBucket = Module + "_" + seat.StateSeatHolders
)
const (
// PlacementHistory is the revisions the placement bucket keeps per key: the bus's most (§6).
PlacementHistory = 64
// Stream is the holder's stream of the seat's events, kept StreamDays (§6), and declared as the holder's data so the
// mesh's backups keep it.
Stream = "placement"
StreamDays = 30
)
modules/mesh-delivery-assigner/manifest_test.go
+5 -4
View File
@@ -9,9 +9,10 @@ import (
"testing"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/views"
plc "mesh-delivery-assigner/internal/placement"
)
func load(t *testing.T) catalogue.Manifest {
@@ -102,9 +103,9 @@ func TestManifest(t *testing.T) {
if len(m.Streams) != 1 {
t.Errorf("streams = %+v, want placement alone", m.Streams)
}
if m.Streams[0].Name != delivery.PlacementStream || m.Streams[0].MaxAgeDays != delivery.PlacementStreamDays ||
!slices.Equal(slices.Sorted(slices.Values(m.Streams[0].Events)), slices.Sorted(slices.Values(delivery.PlacementEvents))) {
t.Errorf("streams = %+v, want placement carrying the seat's four events for %d days", m.Streams, delivery.PlacementStreamDays)
if m.Streams[0].Name != plc.Stream || m.Streams[0].MaxAgeDays != plc.StreamDays ||
!slices.Equal(slices.Sorted(slices.Values(m.Streams[0].Events)), slices.Sorted(slices.Values(meshmoduleassigner.Events()))) {
t.Errorf("streams = %+v, want placement carrying the seat's four events for %d days", m.Streams, plc.StreamDays)
}
// ADR 0618 §4: and it hears each node's engine say its reads changed.
if !slices.Equal(m.Consumes, []string{"mesh-delivery-indexer.views.graph", "mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules", "node-engine.manifest-reads-changed"}) {