From a3cb3e4a9b83dc483187b3b0765c20b8faf33ce1 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 5 Nov 2026 15:33:01 +0100 Subject: [PATCH] =?UTF-8?q?Assigner:=20the=20seat's=20state=20names,=20eve?= =?UTF-8?q?nts,=20buckets=20and=20history=20are=20the=20contract's=20and?= =?UTF-8?q?=20the=20placement=20package's=20(D-12;=20ADR=200634=20=C2=A71;?= =?UTF-8?q?=20issue=201966)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i --- .../integration_test.go | 48 +++++++------- .../cmd/mesh-delivery-assigner/main.go | 36 ++++++----- .../cmd/mesh-delivery-assigner/main_test.go | 3 +- .../cmd/mesh-delivery-assigner/pass.go | 12 ++-- .../cmd/mesh-delivery-assigner/pass_test.go | 6 +- .../cmd/mesh-delivery-assigner/seat.go | 62 +++++++++---------- .../cmd/mesh-delivery-assigner/seat_tools.go | 32 +++++----- .../mesh-delivery-assigner/seat_tools_test.go | 48 +++++++------- .../cmd/mesh-delivery-assigner/views.go | 24 +++---- .../cmd/mesh-delivery-assigner/world.go | 9 +-- .../internal/assign/seat_test.go | 4 +- .../internal/placement/state.go | 25 ++++++++ .../mesh-delivery-assigner/manifest_test.go | 9 +-- 13 files changed, 175 insertions(+), 143 deletions(-) create mode 100644 modules/mesh-delivery-assigner/internal/placement/state.go diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go index 5f91bab9..9dbfa675 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go @@ -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 diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go index 12895cd8..fb6eb36e 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -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 ``, 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", diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go index ad676c6e..622d27cc 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go @@ -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 diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go index 749f092e..1bef6193 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -1012,7 +1012,7 @@ func missingPulse(key string) string { return "provision-missing." + key } // `provider....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) } } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go index 7e4a19d6..e00948ea 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go @@ -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) } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go index 8d46ae2b..f8e6dbdc 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go @@ -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 } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools.go index 8765a256..75dcafcf 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools.go @@ -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.` before it was converted, then its node key's `.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) } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go index 40763a8d..7b94537e 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go @@ -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) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go index 9906624e..37c1b42f 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go @@ -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.., 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..."}, {"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 { diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go index 3d2b7ff9..dc5f602d 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go @@ -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). diff --git a/modules/mesh-delivery-assigner/internal/assign/seat_test.go b/modules/mesh-delivery-assigner/internal/assign/seat_test.go index c78cca93..9dac1640 100644 --- a/modules/mesh-delivery-assigner/internal/assign/seat_test.go +++ b/modules/mesh-delivery-assigner/internal/assign/seat_test.go @@ -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) } } diff --git a/modules/mesh-delivery-assigner/internal/placement/state.go b/modules/mesh-delivery-assigner/internal/placement/state.go new file mode 100644 index 00000000..bc776ee6 --- /dev/null +++ b/modules/mesh-delivery-assigner/internal/placement/state.go @@ -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 `_`), 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 +) diff --git a/modules/mesh-delivery-assigner/manifest_test.go b/modules/mesh-delivery-assigner/manifest_test.go index 48de9857..ccb18490 100644 --- a/modules/mesh-delivery-assigner/manifest_test.go +++ b/modules/mesh-delivery-assigner/manifest_test.go @@ -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"}) {