The node-engine follows the module-assignment seat: assignments, not placement; the seat's packages move to mesh-contracts and mesh-delivery (issue 1966)
The state, events and types of the seat mesh-module-assigner say assignments; the legacy node key form is gone from the reads; the dead seat-verb list of the retired name is deleted. The placement guard and the placed-file sense stay. Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
This commit is contained in:
@@ -8,13 +8,13 @@ import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient"
|
||||
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
|
||||
"git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner"
|
||||
nodeengine "git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
|
||||
"git.novox.be/novox/mesh-contracts/placement/go/clients/meshmoduleassignerclient"
|
||||
"git.novox.be/novox/mesh-contracts/placement/go/seats/meshmoduleassigner"
|
||||
"git.novox.be/novox/mesh-contracts/placement/go/seats/meshwants"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
|
||||
"git.novox.be/novox/mesh-sdk/go/declaration"
|
||||
"github.com/novox/mesh-host/internal/buckets"
|
||||
@@ -197,13 +197,13 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
okey, _ := meshcatalogclient.ManifestsKey(odigest)
|
||||
b.kv[buckets.Manifests+"/"+okey] = oman
|
||||
omod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: meshcatalogclient.SourcePrefix2 + strings.Repeat("c", 64), Manifest: odigest, Repository: "novox/o", Commit: "c"})
|
||||
place, _ := meshmoduleassignerclient.EncodePlacement(meshmoduleassigner.Placement{Node: "n1", Decided: meshmoduleassigner.NodeDecided{Hub: true, Address: "10.77.0.1"},
|
||||
place := assignedValues(t, meshmoduleassigner.Assignments{Node: "n1", Decided: meshmoduleassigner.NodeDecided{Hub: true, Address: "10.77.0.1"},
|
||||
Modules: []meshmoduleassigner.PlacedModule{{Module: "lemurs"}, {Module: "oldie"}}})
|
||||
have, _ := nodeengineclient.EncodeHaveNode(nodeengine.HaveByNodeNodeValue{Record: &nodeengine.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}})
|
||||
mod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fp, Manifest: digest, Repository: "novox/x", Commit: "c"})
|
||||
want, _ := bucketstest.Want(meshwants.WantsByWantNodeModuleValue{Fp: fp, Self: true})
|
||||
nodeWant, _ := bucketstest.NodeWant(meshwants.WantsByWantNodeNodeValue{Self: true})
|
||||
b.put(t, buckets.Placement, "placement.n1", place)
|
||||
b.putAssigned(t, place)
|
||||
b.put(t, buckets.Have, "n1.node", have)
|
||||
oldHave, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Build: "build-1", Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK})
|
||||
b.put(t, buckets.Have, "n1.oldie", oldHave)
|
||||
@@ -211,7 +211,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
b.put(t, buckets.Modules, "oldie", omod)
|
||||
b.put(t, buckets.Wants, "want.n1.lemurs", want)
|
||||
b.put(t, buckets.Wants, "want.n1.node", nodeWant)
|
||||
for _, bucket := range []string{buckets.Wants, buckets.Placement, buckets.Have, buckets.Modules, ResolvedBucket, builds.BuilderBuildsBucket} {
|
||||
for _, bucket := range []string{buckets.Wants, buckets.Assignments, buckets.Have, buckets.Modules, ResolvedBucket, builds.BuilderBuildsBucket} {
|
||||
b.send(bucket, agent.Update{Caught: true})
|
||||
}
|
||||
got := map[string]bool{}
|
||||
@@ -324,8 +324,8 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
for k, v := range c.wants {
|
||||
c2.wants[k] = v
|
||||
}
|
||||
for k, v := range c.placements {
|
||||
c2.placements[k] = v
|
||||
for k, v := range c.assignments {
|
||||
c2.assignments[k] = v
|
||||
}
|
||||
for k, v := range c.records {
|
||||
c2.records[k] = v
|
||||
@@ -441,9 +441,9 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
c3.modules[m] = meshcatalog.ModulesByModuleValue{Fp2: mfp, Manifest: dig, Repository: "novox/" + m, Commit: "c"}
|
||||
c3.wants[m] = meshwants.WantsByWantNodeModuleValue{Fp: mfp, Self: true}
|
||||
}
|
||||
p3 := c3.placements["n1"]
|
||||
p3 := c3.assignments["n1"]
|
||||
p3.Modules = append(append([]meshmoduleassigner.PlacedModule{}, p3.Modules...), meshmoduleassigner.PlacedModule{Module: "proxy"}, meshmoduleassigner.PlacedModule{Module: "de-spiegel"})
|
||||
c3.placements["n1"] = p3
|
||||
c3.assignments["n1"] = p3
|
||||
r3 := c3.records["n1"]
|
||||
r3.Capabilities = []string{"systemd", "container-runtime"}
|
||||
c3.records["n1"] = r3
|
||||
@@ -495,9 +495,9 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
// The indexer's verdict on the captured manifest (ADR 0578): the engine judges nothing, it takes this.
|
||||
Verdict: meshcatalogclient.VerdictRefused, Problems: []string{`bar item "shows" is none of them`}}
|
||||
c4.wants["power"] = meshwants.WantsByWantNodeModuleValue{Fp: powerFP, Self: true}
|
||||
p4 := c4.placements["n1"]
|
||||
p4 := c4.assignments["n1"]
|
||||
p4.Modules = append(append([]meshmoduleassigner.PlacedModule{}, p4.Modules...), meshmoduleassigner.PlacedModule{Module: "power"})
|
||||
c4.placements["n1"] = p4
|
||||
c4.assignments["n1"] = p4
|
||||
handed4 := map[string]agent.Update{}
|
||||
for _, u := range c4.compose(ctx) {
|
||||
handed4[u.Key] = u
|
||||
@@ -513,10 +513,10 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
// what runs nor the kept declaration: with nothing running (the kept read retired, Running empty) the node's own
|
||||
// still carries its resources, so retiring the read removes nothing of it.
|
||||
c5, told5 := fresh(env.Get)
|
||||
p5 := c5.placements["n1"]
|
||||
p5 := c5.assignments["n1"]
|
||||
p5.Modules = append(append([]meshmoduleassigner.PlacedModule{}, p5.Modules...), meshmoduleassigner.PlacedModule{Module: overlay.Name})
|
||||
p5.Decided.Endpoint = "n1.example:51820"
|
||||
c5.placements["n1"] = p5
|
||||
c5.assignments["n1"] = p5
|
||||
c5.inOwn[overlay.Name] = true
|
||||
nodeKey5, _ := compositions.NodeKey("n1")
|
||||
var own5 string
|
||||
@@ -597,12 +597,12 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
// secrets. A key written again with what it held, or a want whose words moved and not its build, makes none.
|
||||
worded, _ := bucketstest.Want(meshwants.WantsByWantNodeModuleValue{Fp: fp, Self: true, Words: "lemurs is held at aaaaaaaaaaaa"})
|
||||
elsewhere, _ := bucketstest.Want(meshwants.WantsByWantNodeModuleValue{Fp: meshcatalogclient.SourcePrefix2 + strings.Repeat("e", 64), Self: true})
|
||||
// Research 072 row 37: the placement converted to one key per module and node moves nothing composed; a write in
|
||||
// keys makes its pass at its node key, the commit, and not at a module key written before it nor at the mirror.
|
||||
placeBase := meshmoduleassigner.Placement{Node: "n1", Decided: meshmoduleassigner.NodeDecided{Hub: true, Address: "10.77.0.1"},
|
||||
// Research 072 row 37: the assignments kept as one key per module and node move nothing composed when written again; a
|
||||
// write in keys makes its pass at its node key, the commit, and not at a module key written before it.
|
||||
placeBase := meshmoduleassigner.Assignments{Node: "n1", Decided: meshmoduleassigner.NodeDecided{Hub: true, Address: "10.77.0.1"},
|
||||
Modules: []meshmoduleassigner.PlacedModule{{Module: "lemurs"}, {Module: "oldie"}}}
|
||||
placeKeys := meshmoduleassignerclient.NewKeys()
|
||||
keyRows := func(after meshmoduleassigner.Placement, what string) []struct {
|
||||
keyRows := func(after meshmoduleassigner.Assignments, what string) []struct {
|
||||
name, bucket, key string
|
||||
value []byte
|
||||
pass bool
|
||||
@@ -622,23 +622,25 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
name, bucket, key string
|
||||
value []byte
|
||||
pass bool
|
||||
}{what + ": " + w.Key, buckets.Placement, w.Key, w.Value, w.Commit && what != "converted"})
|
||||
}{what + ": " + w.Key, buckets.Assignments, w.Key, w.Value, w.Commit})
|
||||
}
|
||||
return rows
|
||||
}
|
||||
_, _ = placeKeys.Put("placement.n1", place, 1, false)
|
||||
for k, v := range place {
|
||||
_, _ = placeKeys.Put(k, v, 1, false)
|
||||
}
|
||||
lemursMoved := placeBase
|
||||
lemursMoved.Modules = []meshmoduleassigner.PlacedModule{{Module: "lemurs", Values: map[string]any{"greeting": "hoi"}}, {Module: "oldie"}}
|
||||
var placementRows []struct {
|
||||
var assignmentRows []struct {
|
||||
name, bucket, key string
|
||||
value []byte
|
||||
pass bool
|
||||
}
|
||||
for _, step := range []struct {
|
||||
what string
|
||||
p meshmoduleassigner.Placement
|
||||
}{{"converted", placeBase}, {"lemurs moved", lemursMoved}, {"lemurs back", placeBase}} {
|
||||
placementRows = append(placementRows, keyRows(step.p, step.what)...)
|
||||
p meshmoduleassigner.Assignments
|
||||
}{{"lemurs moved", lemursMoved}, {"lemurs back", placeBase}} {
|
||||
assignmentRows = append(assignmentRows, keyRows(step.p, step.what)...)
|
||||
}
|
||||
for _, row := range append([]struct {
|
||||
name, bucket, key string
|
||||
@@ -648,11 +650,11 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
{"lemurs's want again", buckets.Wants, "want.n1.lemurs", want, false},
|
||||
{"the node's own want again", buckets.Wants, "want.n1.node", nodeWant, false},
|
||||
{"lemurs's want with other words", buckets.Wants, "want.n1.lemurs", worded, false},
|
||||
{"the placement again", buckets.Placement, "placement.n1", place, false},
|
||||
{"the assignment again", buckets.Assignments, "n1.node", place["n1.node"], false},
|
||||
{"lemurs's module again", buckets.Modules, "lemurs", mod, false},
|
||||
{"lemurs's want at another build", buckets.Wants, "want.n1.lemurs", elsewhere, true},
|
||||
{"lemurs's want back at its build", buckets.Wants, "want.n1.lemurs", want, true},
|
||||
}, placementRows...) {
|
||||
}, assignmentRows...) {
|
||||
before := passes()
|
||||
b.put(t, row.bucket, row.key, row.value)
|
||||
if !row.pass {
|
||||
@@ -698,18 +700,18 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
tplace, _ := meshmoduleassignerclient.EncodePlacement(meshmoduleassigner.Placement{Node: "n1", Decided: meshmoduleassigner.NodeDecided{Hub: true, Address: "10.77.0.1"},
|
||||
tplace := assignedValues(t, meshmoduleassigner.Assignments{Node: "n1", Decided: meshmoduleassigner.NodeDecided{Hub: true, Address: "10.77.0.1"},
|
||||
Modules: []meshmoduleassigner.PlacedModule{{Module: "tool"}}})
|
||||
tmod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: next2, Manifest: tdigest, Repository: "novox/t", Commit: "c"})
|
||||
twant, _ := bucketstest.Want(meshwants.WantsByWantNodeModuleValue{Fp: next2, Self: true})
|
||||
building, _ := builds.EncodeBuilderBuild(builds.BuilderBuild{State: "building"})
|
||||
bkey, _ := builds.BuilderBuildKey("tool", next2)
|
||||
tb.kv[builds.BuilderBuildsBucket+"/"+bkey] = building
|
||||
tb.put(t, buckets.Placement, "placement.n1", tplace)
|
||||
tb.putAssigned(t, tplace)
|
||||
tb.put(t, buckets.Have, "n1.node", have)
|
||||
tb.put(t, buckets.Modules, "tool", tmod)
|
||||
tb.put(t, buckets.Wants, "want.n1.tool", twant)
|
||||
for _, bucket := range []string{buckets.Wants, buckets.Placement, buckets.Have, buckets.Modules,
|
||||
for _, bucket := range []string{buckets.Wants, buckets.Assignments, buckets.Have, buckets.Modules,
|
||||
ResolvedBucket, builds.BuilderBuildsBucket} {
|
||||
tb.send(bucket, agent.Update{Caught: true})
|
||||
}
|
||||
@@ -777,18 +779,19 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
b.kv[buckets.Manifests+"/"+kkey] = kman
|
||||
b.mu.Unlock()
|
||||
kmod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fp, Manifest: kdigest, Repository: "novox/k", Commit: "c"})
|
||||
both, _ := meshmoduleassignerclient.EncodePlacement(meshmoduleassigner.Placement{Node: "n1", Decided: meshmoduleassigner.NodeDecided{Hub: true, Address: "10.77.0.1"},
|
||||
both := assignedValues(t, meshmoduleassigner.Assignments{Node: "n1", Decided: meshmoduleassigner.NodeDecided{Hub: true, Address: "10.77.0.1"},
|
||||
Modules: []meshmoduleassigner.PlacedModule{{Module: "lemurs"}, {Module: "oldie"}, {Module: "keeper"}}})
|
||||
b.put(t, buckets.Modules, "keeper", kmod)
|
||||
// n1 is held in keys per module and node since the rows above, which a legacy key no longer moves.
|
||||
bothPlaced, _ := meshmoduleassignerclient.DecodePlacement(both)
|
||||
bothWrites, err := placeKeys.Writes("n1", &bothPlaced)
|
||||
// n1 is held in keys per module and node, which the writes below move.
|
||||
bothPlaced, _ := meshmoduleassignerclient.AssignmentsOf(both)
|
||||
bothNode := bothPlaced["n1"]
|
||||
bothWrites, err := placeKeys.Writes("n1", &bothNode)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, w := range bothWrites {
|
||||
if w.Value != nil {
|
||||
b.put(t, buckets.Placement, w.Key, w.Value)
|
||||
b.put(t, buckets.Assignments, w.Key, w.Value)
|
||||
}
|
||||
}
|
||||
b.put(t, buckets.Wants, "want.n1.keeper", want)
|
||||
@@ -889,8 +892,8 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
for k, v := range c.wants {
|
||||
again.wants[k] = v
|
||||
}
|
||||
for k, v := range c.placements {
|
||||
again.placements[k] = v
|
||||
for k, v := range c.assignments {
|
||||
again.assignments[k] = v
|
||||
}
|
||||
for k, v := range c.records {
|
||||
again.records[k] = v
|
||||
@@ -912,7 +915,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
// Then the catalogue no longer places the node: it composes nothing of its own, and owes no pass either, so
|
||||
// no timer is armed again (rev-seed-same-3, round 1).
|
||||
again.mu.Lock()
|
||||
clear(again.placements)
|
||||
clear(again.assignments)
|
||||
again.mu.Unlock()
|
||||
if again.compose(ctx); again.owed.Load() {
|
||||
t.Errorf("%s: an unplaced node still owes a pass", row.name)
|
||||
@@ -1034,17 +1037,17 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
if asked, _ := json.Marshal(sent["credentials"]); !strings.Contains(string(asked), `"role":"consumer"`) || !strings.Contains(string(asked), `"role":"provider"`) {
|
||||
t.Errorf("the vault was asked %s", asked)
|
||||
}
|
||||
// ADR 0559 §4: the node answers placement candidates in shadow, placed or not, applying nothing and handing no
|
||||
// composition: the same; a module the group adds, its secret not yet issued; a module removed; a placement it cannot
|
||||
// ADR 0559 §4: the node answers assignment candidates in shadow, placed or not, applying nothing and handing no
|
||||
// composition: the same; a module the group adds, its secret not yet issued; a module removed; a assignment it cannot
|
||||
// compose; a group whose keys went is no longer answered. What runs is never composed from a candidate: the
|
||||
// Composer does not read the state.
|
||||
for _, in := range New(env).inputs() {
|
||||
if in.bucket == buckets.PlacementCandidates {
|
||||
t.Fatal("the node's own composer reads placement candidates")
|
||||
if in.bucket == buckets.ProposalCandidates {
|
||||
t.Fatal("the node's own composer reads assignment candidates")
|
||||
}
|
||||
}
|
||||
sb := newPatternBus()
|
||||
trunk := meshmoduleassigner.Placement{Node: "n1", Decided: meshmoduleassigner.NodeDecided{Hub: true, Address: "10.77.0.1"}, Modules: []meshmoduleassigner.PlacedModule{{Module: "lemurs"}}}
|
||||
trunk := meshmoduleassigner.Assignments{Node: "n1", Decided: meshmoduleassigner.NodeDecided{Hub: true, Address: "10.77.0.1"}, Modules: []meshmoduleassigner.PlacedModule{{Module: "lemurs"}}}
|
||||
mods := map[string]meshcatalog.ModulesByModuleValue{"lemurs": {Fp2: fp, Manifest: digest, Repository: "novox/x", Commit: "c"},
|
||||
"keeper": {Fp2: fp, Manifest: kdigest, Repository: "novox/k", Commit: "c"}}
|
||||
readOnly := &fakeBus{kv: map[string][]byte{buckets.Manifests + "/" + key: man, buckets.Manifests + "/" + kkey: kman}}
|
||||
@@ -1058,7 +1061,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
return json.RawMessage(`{"node":"n1","key":"sk","own":[]}`), nil // nothing issued for keeper: it is not placed here
|
||||
}
|
||||
// What n1 runs: lemurs and its own composition, as its composer composes them from the trunk.
|
||||
running, _, _, err := frombus.Compose(ctx, fixedReader{placements: map[string]meshmoduleassigner.Placement{"n1": trunk},
|
||||
running, _, _, err := frombus.Compose(ctx, fixedReader{assignments: map[string]meshmoduleassigner.Assignments{"n1": trunk},
|
||||
records: map[string]nodeengine.NodeRecord{"n1": {Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}},
|
||||
modules: mods, manifests: map[string][]byte{digest: []byte(lemurs), kdigest: []byte(keeper)}}, frombus.Mesh{}, frombus.Scope{Node: "n1"}, map[string]string{"lemurs": fp})
|
||||
if err != nil {
|
||||
@@ -1071,7 +1074,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
}
|
||||
return out
|
||||
}
|
||||
// The controller's compositions at n1's wants, in n1's composition bucket: what a head deleting its placement
|
||||
// The controller's compositions at n1's wants, in n1's composition bucket: what a head deleting its assignment
|
||||
// returns it to (the seed's revert); here the ones it runs.
|
||||
composedBucket, _ := compositions.BucketFor("n1")
|
||||
for key, c := range running {
|
||||
@@ -1089,15 +1092,15 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
defer scancel()
|
||||
// Review of L2b, round 1, finding 2: the candidates' state is not there when the node starts (a delivery before the
|
||||
// indexer made it): the shadow tries again, and answers once it appears.
|
||||
sb.missing(buckets.PlacementCandidates, true)
|
||||
sb.missing(buckets.ProposalCandidates, true)
|
||||
kept := make(chan struct{})
|
||||
go func() { sh.Keep(sctx, func(int) time.Duration { return 5 * time.Millisecond }); close(kept) }()
|
||||
time.Sleep(30 * time.Millisecond)
|
||||
sb.missing(buckets.PlacementCandidates, false)
|
||||
sb.missing(buckets.ProposalCandidates, false)
|
||||
sb.made(6)
|
||||
rec, _ := nodeengineclient.EncodeHaveNode(nodeengine.HaveByNodeNodeValue{Record: &nodeengine.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}})
|
||||
tp, _ := meshmoduleassignerclient.EncodePlacement(trunk)
|
||||
sb.put(buckets.Placement, "placement.n1", tp, 1)
|
||||
tp := assignedValues(t, trunk)
|
||||
sb.putAssigned(tp, 1)
|
||||
sb.put(buckets.Have, "n1.node", rec, 1)
|
||||
for name, m := range mods {
|
||||
raw, _ := meshcatalogclient.EncodeModule(m)
|
||||
@@ -1124,33 +1127,33 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
for _, m := range modules {
|
||||
p.Modules = append(p.Modules, meshmoduleassigner.PlacedModule{Module: m})
|
||||
}
|
||||
raw, err := bucketstest.Candidate(meshmoduleassigner.CandidatesByProposalNodeValue{Proposal: group, Placement: &p})
|
||||
raw, err := bucketstest.Candidate(meshmoduleassigner.CandidatesByProposalNodeValue{Proposal: group, Assignments: &p})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
k := meshmoduleassigner.CandidatesByProposalNodeKey(group, "n1")
|
||||
sb.put(buckets.PlacementCandidates, k, raw, rev)
|
||||
sb.put(buckets.ProposalCandidates, k, raw, rev)
|
||||
}
|
||||
candidate("same", 3, "lemurs")
|
||||
candidate("adds", 4, "lemurs", "keeper")
|
||||
candidate("drops", 5)
|
||||
candidate("ghost", 6, "lemurs", "ghost")
|
||||
// A head deleting n1's placement file (the seed's revert, mesh-catalog #526): n1 returns to the controller's
|
||||
// A head deleting n1's assignment file (the seed's revert, mesh-catalog #526): n1 returns to the controller's
|
||||
// compositions, answered as such, not as every module removed.
|
||||
if raw, err := bucketstest.Candidate(meshmoduleassigner.CandidatesByProposalNodeValue{Proposal: "deletes"}); err != nil {
|
||||
t.Fatal(err)
|
||||
} else {
|
||||
k := meshmoduleassigner.CandidatesByProposalNodeKey("deletes", "n1")
|
||||
sb.put(buckets.PlacementCandidates, k, raw, 7)
|
||||
sb.put(buckets.ProposalCandidates, k, raw, 7)
|
||||
}
|
||||
// A proposal that changes another node's key and reaches n1's composition (ADR 0567 §3): n1's candidate holds no
|
||||
// placement of its own change (Reached, nil), and n1 answers against its current placement, never as one whose key
|
||||
// assignment of its own change (Reached, nil), and n1 answers against its current assignment, never as one whose key
|
||||
// the proposal takes away.
|
||||
if raw, err := bucketstest.Candidate(meshmoduleassigner.CandidatesByProposalNodeValue{Proposal: "reach", Reached: true}); err != nil {
|
||||
t.Fatal(err)
|
||||
} else {
|
||||
k := meshmoduleassigner.CandidatesByProposalNodeKey("reach", "n1")
|
||||
sb.put(buckets.PlacementCandidates, k, raw, 8)
|
||||
sb.put(buckets.ProposalCandidates, k, raw, 8)
|
||||
}
|
||||
sb.caught()
|
||||
var answers map[string]nodeengine.CandidateAnswer
|
||||
@@ -1199,7 +1202,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
return a.Verdict == nodeengine.CandidateAnswerVerdictSame && a.Returns == "" && a.Revision == 8 && len(a.Removed) == 0 &&
|
||||
fmt.Sprint(a.Same) == "[lemurs node]"
|
||||
},
|
||||
// Issue 1126, A7: a candidate deleting the node's placement is refused, never answered as a return to the
|
||||
// Issue 1126, A7: a candidate deleting the node's assignment is refused, never answered as a return to the
|
||||
// controller's compositions: no composition made elsewhere is applied, so the node would compose nothing.
|
||||
"deletes": func(a nodeengine.CandidateAnswer) bool {
|
||||
return a.Verdict == nodeengine.CandidateAnswerVerdictRefused && a.Returns == "" && a.Revision == 7 &&
|
||||
@@ -1218,7 +1221,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
case <-time.After(100 * time.Millisecond):
|
||||
}
|
||||
// Issue 1126, 2026-10-28 (shanks answered probe/1126-seed-7 at revision 84 every 13-17 s): a candidate revision is
|
||||
// answered once. A module's have whose health or words alone move (each look rewrites them), and a placement, a
|
||||
// answered once. A module's have whose health or words alone move (each look rewrites them), and a assignment, a
|
||||
// module or a config value written again with the value it held, answer nothing again; a new revision answers its
|
||||
// group alone, and what runs moving answers every group once.
|
||||
answering := func() int {
|
||||
@@ -1226,7 +1229,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
defer readOnly.mu.Unlock()
|
||||
n := 0
|
||||
for _, l := range readOnly.said {
|
||||
if strings.HasPrefix(l, "answering placement candidate ") {
|
||||
if strings.HasPrefix(l, "answering assignment candidate ") {
|
||||
n++
|
||||
}
|
||||
}
|
||||
@@ -1255,7 +1258,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
settled := answering()
|
||||
lemursHave("its tool lemurs_health: no answer within 10s", 11)
|
||||
lemursHave("its tool lemurs_health: no answer within 20s", 12)
|
||||
sb.put(buckets.Placement, "placement.n1", tp, 2)
|
||||
sb.putAssigned(tp, 2)
|
||||
lraw, _ := meshcatalogclient.EncodeModule(mods["lemurs"])
|
||||
sb.put(buckets.Modules, "lemurs", lraw, 2)
|
||||
drain()
|
||||
@@ -1279,7 +1282,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
}
|
||||
return ""
|
||||
}
|
||||
if got := lastSaid("answering placement candidate same "); got != "answering placement candidate same (revision 8; woken by the candidate)" {
|
||||
if got := lastSaid("answering assignment candidate same "); got != "answering assignment candidate same (revision 8; woken by the candidate)" {
|
||||
t.Errorf("a new revision's answer said %q", got)
|
||||
}
|
||||
settled = answering()
|
||||
@@ -1290,7 +1293,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
if got := answering(); got != settled+len(answers) {
|
||||
t.Errorf("what runs moved, and %d answer(s) were composed again for %d group(s)", got-settled, len(answers))
|
||||
}
|
||||
if got := lastSaid("answering placement candidate drops "); got != "answering placement candidate drops (revision 5; woken by haves)" {
|
||||
if got := lastSaid("answering assignment candidate drops "); got != "answering assignment candidate drops (revision 5; woken by haves)" {
|
||||
t.Errorf("an answer woken by what runs moving said %q", got)
|
||||
}
|
||||
// A read the bus did not answer is no answer to keep: the group is answered again after a bound, though nothing moved.
|
||||
@@ -1305,41 +1308,41 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
return a["adds"].Revision == 9 && a["adds"].Verdict == nodeengine.CandidateAnswerVerdictDiffers
|
||||
})
|
||||
drain()
|
||||
if got := lastSaid("answering placement candidate adds "); got != "answering placement candidate adds (revision 9; woken by a retry after a read the bus did not answer)" {
|
||||
if got := lastSaid("answering assignment candidate adds "); got != "answering assignment candidate adds (revision 9; woken by a retry after a read the bus did not answer)" {
|
||||
t.Errorf("an answer composed again after the bus did not answer said %q", got)
|
||||
}
|
||||
k := meshmoduleassigner.CandidatesByProposalNodeKey("same", "n1")
|
||||
sb.del(buckets.PlacementCandidates, k)
|
||||
sb.del(buckets.ProposalCandidates, k)
|
||||
waitFor("the deleted group's answer stays", func(a map[string]nodeengine.CandidateAnswer) bool { _, ok := a["same"]; return !ok && len(a) == 5 })
|
||||
|
||||
// Review of L2b, round 2: a watch ends while the link stays; a group's key is deleted before the shadow watches
|
||||
// again. The new run reads afresh, so that group is no longer answered.
|
||||
sb.missing(buckets.PlacementCandidates, true)
|
||||
sb.missing(buckets.ProposalCandidates, true)
|
||||
sb.end(buckets.Modules)
|
||||
dropsKey := meshmoduleassigner.CandidatesByProposalNodeKey("drops", "n1")
|
||||
sb.del(buckets.PlacementCandidates, dropsKey)
|
||||
sb.del(buckets.ProposalCandidates, dropsKey)
|
||||
time.Sleep(30 * time.Millisecond)
|
||||
sb.missing(buckets.PlacementCandidates, false)
|
||||
sb.missing(buckets.ProposalCandidates, false)
|
||||
waitFor("a group deleted while the shadow was down is still answered", func(a map[string]nodeengine.CandidateAnswer) bool {
|
||||
_, ok := a["drops"]
|
||||
return !ok && len(a) == 4
|
||||
})
|
||||
|
||||
// Issue 1126, A7 AC1: a placement that cannot be read composes nothing and says why; the node never applies a
|
||||
// composition made elsewhere instead. (A placement read without the node's key composes nothing either: above.)
|
||||
// Issue 1126, A7 AC1: a assignment that cannot be read composes nothing and says why; the node never applies a
|
||||
// composition made elsewhere instead. (A assignment read without the node's key composes nothing either: above.)
|
||||
{
|
||||
refused := newFakeBus()
|
||||
env := refused.env()
|
||||
watch := env.Watch
|
||||
env.Watch = func(ctx context.Context, bucket, pattern string) (<-chan agent.Update, error) {
|
||||
if bucket == buckets.Placement {
|
||||
if bucket == buckets.Assignments {
|
||||
return nil, fmt.Errorf("nats: Permissions Violation for Subscription to %s", bucket)
|
||||
}
|
||||
return watch(ctx, bucket, pattern)
|
||||
}
|
||||
out, err := New(env).Watch(ctx, "")
|
||||
if out != nil || err == nil || !strings.Contains(err.Error(), buckets.Placement) || !strings.Contains(err.Error(), "cannot be watched") {
|
||||
t.Errorf("an unreadable placement: composed %v, said %v", out != nil, err)
|
||||
if out != nil || err == nil || !strings.Contains(err.Error(), buckets.Assignments) || !strings.Contains(err.Error(), "cannot be watched") {
|
||||
t.Errorf("an unreadable assignment: composed %v, said %v", out != nil, err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1359,20 +1362,20 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
defer ncancel()
|
||||
go func() { _ = nsh.Run(nctx) }()
|
||||
nb.made(6)
|
||||
nb.put(buckets.Placement, "placement.n1", tp, 1)
|
||||
nb.putAssigned(tp, 1)
|
||||
nb.put(buckets.Have, "n1.node", rec, 1)
|
||||
for name, m := range mods {
|
||||
raw, _ := meshcatalogclient.EncodeModule(m)
|
||||
nb.put(buckets.Modules, name, raw, 1)
|
||||
}
|
||||
sameRaw, _ := bucketstest.Candidate(meshmoduleassigner.CandidatesByProposalNodeValue{Proposal: "same", Placement: &trunk})
|
||||
sameRaw, _ := bucketstest.Candidate(meshmoduleassigner.CandidatesByProposalNodeValue{Proposal: "same", Assignments: &trunk})
|
||||
sameKey := meshmoduleassigner.CandidatesByProposalNodeKey("same", "n1")
|
||||
nb.put(buckets.PlacementCandidates, sameKey, sameRaw, 3)
|
||||
nb.put(buckets.ProposalCandidates, sameKey, sameRaw, 3)
|
||||
// Issue 1126, T3: an older group (a closed pull request's candidate, rev 2) is answered after the newest.
|
||||
oldRaw, _ := bucketstest.Candidate(meshmoduleassigner.CandidatesByProposalNodeValue{Proposal: "older", Placement: &trunk})
|
||||
oldRaw, _ := bucketstest.Candidate(meshmoduleassigner.CandidatesByProposalNodeValue{Proposal: "older", Assignments: &trunk})
|
||||
oldKey := meshmoduleassigner.CandidatesByProposalNodeKey("older", "n1")
|
||||
nb.put(buckets.PlacementCandidates, oldKey, oldRaw, 2)
|
||||
for _, bucket := range []string{buckets.PlacementCandidates, buckets.Placement, buckets.Have, buckets.Modules, builds.BuilderBuildsBucket} {
|
||||
nb.put(buckets.ProposalCandidates, oldKey, oldRaw, 2)
|
||||
for _, bucket := range []string{buckets.ProposalCandidates, buckets.Assignments, buckets.Have, buckets.Modules, builds.BuilderBuildsBucket} {
|
||||
nb.send(bucket, agent.Update{Caught: true})
|
||||
}
|
||||
select {
|
||||
@@ -1384,7 +1387,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
t.Fatal("a shadow not ready after its wait answered nothing")
|
||||
}
|
||||
nmu.Lock()
|
||||
if !strings.Contains(strings.Join(nsaid, "\n"), "this node answers no placement candidate yet: the node has not read "+ResolvedBucket) {
|
||||
if !strings.Contains(strings.Join(nsaid, "\n"), "this node answers no assignment candidate yet: the node has not read "+ResolvedBucket) {
|
||||
t.Errorf("a shadow not ready said %q", nsaid)
|
||||
}
|
||||
nmu.Unlock()
|
||||
@@ -1408,13 +1411,13 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
// Issue 1126, T2: an answer says when it starts and when it ends (and, every minute between, what it waits on).
|
||||
nmu.Lock()
|
||||
said := strings.Join(nsaid, "\n")
|
||||
for _, line := range []string{"the answer to same took", "answering placement candidate same (revision 3; woken by its first answer since the shadow began)",
|
||||
"answered placement candidate same in", "answering placement candidate older (revision 2;"} {
|
||||
for _, line := range []string{"the answer to same took", "answering assignment candidate same (revision 3; woken by its first answer since the shadow began)",
|
||||
"answered assignment candidate same in", "answering assignment candidate older (revision 2;"} {
|
||||
if !strings.Contains(said, line) {
|
||||
t.Errorf("%q was not said: %q", line, nsaid)
|
||||
}
|
||||
}
|
||||
if strings.Index(said, "answering placement candidate same") > strings.Index(said, "answering placement candidate older") {
|
||||
if strings.Index(said, "answering assignment candidate same") > strings.Index(said, "answering assignment candidate older") {
|
||||
t.Errorf("the older group was answered first: %q", nsaid)
|
||||
}
|
||||
nmu.Unlock()
|
||||
@@ -1454,7 +1457,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
// trunk's.
|
||||
inOwnRunning := []compositions.Composed{{Node: "n1", Module: "lemurs", FP: fp}}
|
||||
other := "src2:" + strings.Repeat("9", 64)
|
||||
gotFps := shadowBuilds(inOwnRunning, meshmoduleassigner.Placement{Node: "n1", Modules: []meshmoduleassigner.PlacedModule{{Module: "lemurs"}, {Module: "nats"}, {Module: "keeper"}}},
|
||||
gotFps := shadowBuilds(inOwnRunning, meshmoduleassigner.Assignments{Node: "n1", Modules: []meshmoduleassigner.PlacedModule{{Module: "lemurs"}, {Module: "nats"}, {Module: "keeper"}}},
|
||||
map[string]meshcatalog.ModulesByModuleValue{"lemurs": {Fp2: other}, "nats": {Fp2: other}, "keeper": {Fp2: other}}, []string{"nats"})
|
||||
if fmt.Sprint(gotFps) != fmt.Sprint(map[string]string{"lemurs": fp, "keeper": other}) {
|
||||
t.Errorf("shadow builds %v; want lemurs at what runs, keeper at the trunk's, nats at none (folded)", gotFps)
|
||||
@@ -1519,12 +1522,12 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
cc.manifests[cdigest] = []byte(withCert)
|
||||
cc.modules = map[string]meshcatalog.ModulesByModuleValue{"lemurs": {Manifest: digest}, "web": {Manifest: cdigest}}
|
||||
cc.records["n1"] = nodeengine.NodeRecord{Node: "n1", ServingKey: base64.StdEncoding.EncodeToString(make([]byte, 32))}
|
||||
onlyLemurs := meshmoduleassigner.Placement{Node: "n1", Modules: []meshmoduleassigner.PlacedModule{{Module: "lemurs"}}}
|
||||
onlyLemurs := meshmoduleassigner.Assignments{Node: "n1", Modules: []meshmoduleassigner.PlacedModule{{Module: "lemurs"}}}
|
||||
cc.certifyIfWanted(context.Background(), reader{c: cc}, onlyLemurs)
|
||||
if certAsks != 0 {
|
||||
t.Errorf("no module asks for a certificate, and the vault was asked %d times", certAsks)
|
||||
}
|
||||
withWeb := meshmoduleassigner.Placement{Node: "n1", Modules: []meshmoduleassigner.PlacedModule{{Module: "lemurs"}, {Module: "web"}}}
|
||||
withWeb := meshmoduleassigner.Assignments{Node: "n1", Modules: []meshmoduleassigner.PlacedModule{{Module: "lemurs"}, {Module: "web"}}}
|
||||
cc.certifyIfWanted(context.Background(), reader{c: cc}, withWeb)
|
||||
cc.certifyIfWanted(context.Background(), reader{c: cc}, withWeb)
|
||||
if certAsks != 1 {
|
||||
@@ -1665,14 +1668,14 @@ func (b *patternBus) caught() {
|
||||
|
||||
// fixedReader is frombus's Reader over values held, for composing what a node runs.
|
||||
type fixedReader struct {
|
||||
placements map[string]meshmoduleassigner.Placement
|
||||
records map[string]nodeengine.NodeRecord
|
||||
modules map[string]meshcatalog.ModulesByModuleValue
|
||||
manifests map[string][]byte
|
||||
assignments map[string]meshmoduleassigner.Assignments
|
||||
records map[string]nodeengine.NodeRecord
|
||||
modules map[string]meshcatalog.ModulesByModuleValue
|
||||
manifests map[string][]byte
|
||||
}
|
||||
|
||||
func (r fixedReader) Placements(context.Context) (map[string]meshmoduleassigner.Placement, error) {
|
||||
return r.placements, nil
|
||||
func (r fixedReader) Assignments(context.Context) (map[string]meshmoduleassigner.Assignments, error) {
|
||||
return r.assignments, nil
|
||||
}
|
||||
func (r fixedReader) Records(context.Context) (map[string]nodeengine.NodeRecord, error) {
|
||||
return r.records, nil
|
||||
@@ -1791,18 +1794,18 @@ func TestAnUnreadSeatListHoldsWhatIsNewAndHandsItAtTheRead(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
place, _ := meshmoduleassignerclient.EncodePlacement(meshmoduleassigner.Placement{Node: "n1", Decided: meshmoduleassigner.NodeDecided{Hub: true, Address: "10.77.0.1"},
|
||||
place := assignedValues(t, meshmoduleassigner.Assignments{Node: "n1", Decided: meshmoduleassigner.NodeDecided{Hub: true, Address: "10.77.0.1"},
|
||||
Modules: []meshmoduleassigner.PlacedModule{{Module: "lemurs"}}})
|
||||
have, _ := nodeengineclient.EncodeHaveNode(nodeengine.HaveByNodeNodeValue{Record: &nodeengine.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}})
|
||||
mod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fp, Manifest: digest, Repository: "novox/x", Commit: "c"})
|
||||
want, _ := bucketstest.Want(meshwants.WantsByWantNodeModuleValue{Fp: fp, Self: true})
|
||||
nodeWant, _ := bucketstest.NodeWant(meshwants.WantsByWantNodeNodeValue{Self: true})
|
||||
b.put(t, buckets.Placement, "placement.n1", place)
|
||||
b.putAssigned(t, place)
|
||||
b.put(t, buckets.Have, "n1.node", have)
|
||||
b.put(t, buckets.Modules, "lemurs", mod)
|
||||
b.put(t, buckets.Wants, "want.n1.lemurs", want)
|
||||
b.put(t, buckets.Wants, "want.n1.node", nodeWant)
|
||||
for _, bucket := range []string{buckets.Wants, buckets.Placement, buckets.Have, buckets.Modules, ResolvedBucket, builds.BuilderBuildsBucket} {
|
||||
for _, bucket := range []string{buckets.Wants, buckets.Assignments, buckets.Have, buckets.Modules, ResolvedBucket, builds.BuilderBuildsBucket} {
|
||||
b.send(bucket, agent.Update{Caught: true})
|
||||
}
|
||||
handed := map[string]bool{}
|
||||
@@ -1872,18 +1875,18 @@ func TestAHeldOwnCompositionThatWasHandedIsNotDropped(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
place, _ := meshmoduleassignerclient.EncodePlacement(meshmoduleassigner.Placement{Node: "n1", Decided: meshmoduleassigner.NodeDecided{Hub: true, Address: "10.77.0.1"},
|
||||
place := assignedValues(t, meshmoduleassigner.Assignments{Node: "n1", Decided: meshmoduleassigner.NodeDecided{Hub: true, Address: "10.77.0.1"},
|
||||
Modules: []meshmoduleassigner.PlacedModule{{Module: "lemurs"}}})
|
||||
have, _ := nodeengineclient.EncodeHaveNode(nodeengine.HaveByNodeNodeValue{Record: &nodeengine.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}})
|
||||
mod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fp, Manifest: digest, Repository: "novox/x", Commit: "c"})
|
||||
want, _ := bucketstest.Want(meshwants.WantsByWantNodeModuleValue{Fp: fp, Self: true})
|
||||
nodeWant, _ := bucketstest.NodeWant(meshwants.WantsByWantNodeNodeValue{Self: true})
|
||||
b.put(t, buckets.Placement, "placement.n1", place)
|
||||
b.putAssigned(t, place)
|
||||
b.put(t, buckets.Have, "n1.node", have)
|
||||
b.put(t, buckets.Modules, "lemurs", mod)
|
||||
b.put(t, buckets.Wants, "want.n1.lemurs", want)
|
||||
b.put(t, buckets.Wants, "want.n1.node", nodeWant)
|
||||
for _, bucket := range []string{buckets.Wants, buckets.Placement, buckets.Have, buckets.Modules, ResolvedBucket, builds.BuilderBuildsBucket} {
|
||||
for _, bucket := range []string{buckets.Wants, buckets.Assignments, buckets.Have, buckets.Modules, ResolvedBucket, builds.BuilderBuildsBucket} {
|
||||
b.send(bucket, agent.Update{Caught: true})
|
||||
}
|
||||
// The list is read: the node's own composition is handed. Then the reader cannot act again (it restarted and has
|
||||
@@ -1999,13 +2002,13 @@ func TestTheHolderOfTheCatalogIsStartedFromTheOwnCompositionBeforeTheListIsReadA
|
||||
b.put(t, buckets.Have, hk, hv)
|
||||
placed = append(placed, meshmoduleassigner.PlacedModule{Module: m})
|
||||
}
|
||||
place, _ := meshmoduleassignerclient.EncodePlacement(meshmoduleassigner.Placement{Node: "n1", Decided: meshmoduleassigner.NodeDecided{Hub: true, Address: "10.77.0.1"}, Modules: placed})
|
||||
place := assignedValues(t, meshmoduleassigner.Assignments{Node: "n1", Decided: meshmoduleassigner.NodeDecided{Hub: true, Address: "10.77.0.1"}, Modules: placed})
|
||||
have, _ := nodeengineclient.EncodeHaveNode(nodeengine.HaveByNodeNodeValue{Record: &nodeengine.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}})
|
||||
nodeWant, _ := bucketstest.NodeWant(meshwants.WantsByWantNodeNodeValue{Self: true})
|
||||
b.put(t, buckets.Placement, "placement.n1", place)
|
||||
b.putAssigned(t, place)
|
||||
b.put(t, buckets.Have, "n1.node", have)
|
||||
b.put(t, buckets.Wants, "want.n1.node", nodeWant)
|
||||
for _, bucket := range []string{buckets.Wants, buckets.Placement, buckets.Have, buckets.Modules, ResolvedBucket, builds.BuilderBuildsBucket} {
|
||||
for _, bucket := range []string{buckets.Wants, buckets.Assignments, buckets.Have, buckets.Modules, ResolvedBucket, builds.BuilderBuildsBucket} {
|
||||
b.send(bucket, agent.Update{Caught: true})
|
||||
}
|
||||
servicesIn := func(u agent.Update) []string {
|
||||
@@ -2056,3 +2059,28 @@ func TestACompositionHoldsWhenTheSeatListReaderCouldNotStart(t *testing.T) {
|
||||
t.Errorf("held %v, comps %v: a reader that did not start must hold the node's own composition", held, comps)
|
||||
}
|
||||
}
|
||||
|
||||
// assignedValues is every key the assigner writes for p: the node's, `<node>.node`, and each module's, `<node>.<module>`.
|
||||
func assignedValues(t *testing.T, p meshmoduleassigner.Assignments) map[string][]byte {
|
||||
t.Helper()
|
||||
vals, err := meshmoduleassignerclient.AssignmentsValues(p)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return vals
|
||||
}
|
||||
|
||||
// putAssigned writes the keys of one node's assignments into the assignments state.
|
||||
func (b *fakeBus) putAssigned(t *testing.T, vals map[string][]byte) {
|
||||
t.Helper()
|
||||
for _, k := range slices.Sorted(maps.Keys(vals)) {
|
||||
b.put(t, buckets.Assignments, k, vals[k])
|
||||
}
|
||||
}
|
||||
|
||||
// putAssigned sends the keys of one node's assignments, each at the revision.
|
||||
func (b *patternBus) putAssigned(vals map[string][]byte, rev uint64) {
|
||||
for _, k := range slices.Sorted(maps.Keys(vals)) {
|
||||
b.put(buckets.Assignments, k, vals[k], rev)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user