Follow L2 under L3: the builder, checker, indexer and view read the seat mesh-placement's placement, the seat's Leased is the lease handed away, and the view keeps no read of the indexer's placement L2 removed (issue 774, ADR 0567 §8, ADR 0549 §3)
This commit is contained in:
@@ -86,7 +86,8 @@ var seatBuckets = []string{delivery.PlacementBucket, delivery.PlacementProposals
|
||||
// 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 compositions by the checker, its asks by the view
|
||||
// and the teller, its puts by whoever follows one (ADR 0546), and the seat mesh-placement's placement, proposals and
|
||||
// candidates by the node agents, the grants part, mesh-config and the view (ADR 0567 §9).
|
||||
// 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},
|
||||
@@ -108,7 +109,7 @@ type buckets map[string]jetstream.KeyValue
|
||||
// without it and takes no put further until it starts again (ADR 0546).
|
||||
func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any)) (buckets, error) {
|
||||
out := buckets{}
|
||||
// The controller's lease first (ADR 0549 §5, ADR 0561 §3): once it is handed away the catalogue's placement alone names
|
||||
// The controller's lease first (ADR 0549 §5, ADR 0561 §3): once it is handed away the seat's placement alone names
|
||||
// where modules run, and the controller's assignments are no bucket the assigner needs. Not granted or not there, the
|
||||
// controller holds it, as before.
|
||||
away, why := false, ""
|
||||
@@ -132,7 +133,7 @@ func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any
|
||||
}
|
||||
kv, err := js.KeyValue(ctx, name)
|
||||
if name == provider.AssignmentsBucket && err != nil && away {
|
||||
logf("the bucket %s cannot be opened (%v), and none is needed: %s, so the catalogue's placement alone is read", name, err, why)
|
||||
logf("the bucket %s cannot be opened (%v), and none is needed: %s, so the seat's placement alone is read", name, err, why)
|
||||
continue
|
||||
}
|
||||
if name == delivery.AssignerPutsBucket && err != nil {
|
||||
|
||||
@@ -82,7 +82,7 @@ func TestOpensWhatItsManifestNames(t *testing.T) {
|
||||
{delivery.PartGrantsBucket, jetstream.ErrBucketNotFound, true, nil, "the controller's grants are read"},
|
||||
{buscompose.LeaseBucket, errors.New("nats: permissions violation"), true, nil, "taken to hold its lease"},
|
||||
{provider.AssignmentsBucket, errors.New("nats: permissions violation"), false, []byte(`{"instance":"novox/1"}`), ""},
|
||||
{provider.AssignmentsBucket, errors.New("nats: permissions violation"), true, handed, "the catalogue's placement alone is read"},
|
||||
{provider.AssignmentsBucket, errors.New("nats: permissions violation"), true, handed, "the seat's placement alone is read"},
|
||||
} {
|
||||
var logged string
|
||||
kvs, err := open(context.Background(), refusingJS{missing: c.missing, err: c.err, lease: c.lease}, func(f string, a ...any) { logged += fmt.Sprintf(f, a...) })
|
||||
|
||||
@@ -136,6 +136,9 @@ func (m *memKV) Update(_ context.Context, key string, value []byte, revision uin
|
||||
type memStore map[string]*memKV
|
||||
|
||||
func (s memStore) get(state, key string) ([]byte, uint64, bool, error) {
|
||||
if s[state] == nil {
|
||||
return nil, 0, false, nil
|
||||
}
|
||||
v, ok := s[state].values[key]
|
||||
return v, uint64(len(v)), ok, nil
|
||||
}
|
||||
|
||||
@@ -661,7 +661,7 @@ func (p *passer) followWrite(prop delivery.PlacementProposal, answer delivery.An
|
||||
}
|
||||
p.seat.record(answer.Warrant)
|
||||
p.sayChanged(prop, answer, changes)
|
||||
if p.seat != nil && !assign.Leased && !prop.Seed {
|
||||
if p.seat != nil && !p.w.handedAway() && !prop.Seed {
|
||||
p.seat.callAll(assign.ControllerCalls(before, written, prop.ID, answer.Warrant))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -9,6 +9,7 @@ import (
|
||||
"time"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/buscompose"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
|
||||
@@ -214,7 +215,7 @@ func (s *seatTools) doPropose(args map[string]any, by string) (any, error) {
|
||||
}
|
||||
now := s.now()
|
||||
p, laid, findings := assign.Propose(assign.ProposeIn{ID: delivery.NewProposalID(now), By: by, Why: arg(args, "why"), At: now,
|
||||
Changes: changes, Current: current, Revisions: revs, Held: held, Recorded: recorded})
|
||||
Changes: changes, Current: current, Revisions: revs, Held: held, Recorded: recorded, Leased: s.leased()})
|
||||
if len(findings) > 0 {
|
||||
return nil, fmt.Errorf("the proposal is refused: %s", strings.Join(findings, "; "))
|
||||
}
|
||||
@@ -298,7 +299,7 @@ func (s *seatTools) doAdopt(args map[string]any, by string) (any, error) {
|
||||
now := s.now()
|
||||
p, laid, findings := assign.Propose(assign.ProposeIn{ID: delivery.NewProposalID(now), By: by, Why: arg(args, "why"), At: now,
|
||||
Changes: assign.SeedChanges(export), Current: map[string]delivery.Placement{}, Revisions: map[string]uint64{},
|
||||
Held: held, Recorded: recorded, Seed: true, ExportDigest: delivery.ExportDigest(export)})
|
||||
Held: held, Recorded: recorded, Seed: true, ExportDigest: delivery.ExportDigest(export), Leased: s.leased()})
|
||||
if len(findings) > 0 {
|
||||
return nil, fmt.Errorf("the seed is refused: %s", strings.Join(findings, "; "))
|
||||
}
|
||||
@@ -308,6 +309,15 @@ func (s *seatTools) doAdopt(args map[string]any, by string) (any, error) {
|
||||
return proposed{Proposal: p, Follow: followProposal(p.ID)}, nil
|
||||
}
|
||||
|
||||
// leased says the controller's lease is handed away (mesh-sdk buscompose.LeaseHandedAway, ADR 0561 §3; issue 774): the lease
|
||||
// write of ADR 0547 that ends keeping the controller in step (ADR 0567 §8). A lease not read, a controller's, expired or
|
||||
// unreadable, is not handed away.
|
||||
func (s *seatTools) leased() bool {
|
||||
raw, _, found, err := s.st.get(buscompose.ControllerLease, buscompose.ControllerLeaseKey)
|
||||
away, _ := buscompose.LeaseHandedAway(raw, found && raw != nil, err)
|
||||
return away
|
||||
}
|
||||
|
||||
// handActCall records a hand act in the controller's hand-act log.
|
||||
const handActCall = "seat:" + catalogue.ControllerSeatName + ".hand-act"
|
||||
|
||||
@@ -315,7 +325,7 @@ const handActCall = "seat:" + catalogue.ControllerSeatName + ".hand-act"
|
||||
// proposal naming every key at its revision; the assigner raises its authorising ask and, on its warrant, deletes the keys,
|
||||
// records the hand act and says placement-changed; every node returns to the controller's compositions.
|
||||
func (s *seatTools) doHandBack(args map[string]any, by string) (any, error) {
|
||||
if assign.Leased {
|
||||
if s.leased() {
|
||||
return nil, errors.New("placement is the mesh's alone since the lease write: there is no controller to hand it back to")
|
||||
}
|
||||
why := arg(args, "why")
|
||||
|
||||
@@ -15,10 +15,6 @@ import (
|
||||
// `placement`, changed only by a proposal (§2) the nodes it touches answer (§3), judged (§4) and written on the
|
||||
// operator's Approve at the revision it was laid over (§5). What here decides; the tools and the passes do.
|
||||
|
||||
// Leased says whether the lease write of ADR 0547 has happened, after which the controller's assignments are read no
|
||||
// more (§8). Not before L3: until then the holder keeps the controller in step and refuses what its verbs cannot take.
|
||||
var Leased = false
|
||||
|
||||
// ProposeIn is what a proposal is laid over: its id, who and why, its changes; every node's current placement and the
|
||||
// revision of its key; which modules the indexer holds and which nodes have a record; and, for the seed, the export's
|
||||
// digest.
|
||||
@@ -32,6 +28,10 @@ type ProposeIn struct {
|
||||
Recorded func(node string) bool
|
||||
Seed bool
|
||||
ExportDigest string
|
||||
// Leased says the lease write of ADR 0547 has happened (the controller's lease handed away, mesh-sdk
|
||||
// buscompose.LeaseHandedAway; issue 774), after which the controller's assignments are read no more (§8): until then
|
||||
// the holder keeps the controller in step and refuses what its verbs cannot take.
|
||||
Leased bool
|
||||
}
|
||||
|
||||
// Propose lays a proposal over the current keys (§2): the proposal (state answering, each changed node's base revision),
|
||||
@@ -58,7 +58,7 @@ func Propose(in ProposeIn) (delivery.PlacementProposal, map[string]*delivery.Pla
|
||||
findings = append(findings, fmt.Sprintf("change %d: the seed is adopt's alone, and never mixed with other changes", i+1))
|
||||
continue
|
||||
}
|
||||
if !Leased {
|
||||
if !in.Leased {
|
||||
if why := c.BeforeLease(); why != "" {
|
||||
findings = append(findings, fmt.Sprintf("change %d: %s", i+1, why))
|
||||
}
|
||||
|
||||
@@ -60,18 +60,18 @@ func TestThePlacementSeatDecides(t *testing.T) {
|
||||
if _, _, f := Propose(in(delivery.PlacementProposalChange{Op: delivery.OpConfig, Node: "anchor", Module: "web", Values: map[string]any{"puid": 1}})); len(f) != 0 {
|
||||
t.Errorf("config values before the lease write: %v", f)
|
||||
}
|
||||
Leased = true
|
||||
for _, c := range []delivery.PlacementProposalChange{
|
||||
{Op: delivery.OpPlace, Node: "laptop", Module: "web", Ports: map[string]int{"80": 8080}, Bindings: map[string]delivery.Provider{"db": {Node: "anchor", Module: "store"}}},
|
||||
{Op: delivery.OpConfig, Node: "anchor", Module: "web", Layer: map[string]any{"places": []any{}}},
|
||||
{Op: delivery.OpDecide, Node: "laptop", Field: "site", Value: json.RawMessage(`"attic"`)},
|
||||
{Op: delivery.OpHold, Node: "anchor", Module: "store", Claim: "node-shell", Scope: "node"},
|
||||
} {
|
||||
if _, _, f := Propose(in(c)); len(f) != 0 {
|
||||
leased := in(c)
|
||||
leased.Leased = true
|
||||
if _, _, f := Propose(leased); len(f) != 0 {
|
||||
t.Errorf("after the lease write %s is refused: %v", c.Op, f)
|
||||
}
|
||||
}
|
||||
Leased = false
|
||||
|
||||
// A key moved while the warrant wrote: what is left to write is the other nodes' changes, to be laid and asked again.
|
||||
if rest, ok := Rest(p, map[string]*delivery.Placement{"anchor": laid["anchor"]}); !ok || len(rest.Changes) != 1 || rest.Changes[0].Node != "laptop" ||
|
||||
|
||||
@@ -18,7 +18,7 @@ import (
|
||||
|
||||
// 1945 (a part exits at start for a bucket it works without): a bus that refuses every toolchains bucket leaves the
|
||||
// watch unstarted and the part running, said; a refused bucket it cannot work without is an error naming the bucket and
|
||||
// its grant. The gate over the catalogue's placement, and its refusal, are mesh-sdk's (buildrun.ToolchainGate,
|
||||
// its grant. The gate over the seat's placement, and its refusal, are mesh-sdk's (buildrun.ToolchainGate,
|
||||
// issue 774).
|
||||
func TestARefusedBucketIsSaidAndThePartGoesOn(t *testing.T) {
|
||||
set := &buildrun.ToolchainSet{}
|
||||
|
||||
@@ -93,7 +93,7 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
|
||||
// 1887/A4c: the builder chooses no node, so it reads neither presence nor have, invokes no build module's tool and
|
||||
// holds no capacity; it puts asks on the three build seats and reads their steps (ADR 0525 §1, §4, §5).
|
||||
if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-image.steps", "mesh-build-node.steps", "mesh-build-node.toolchains",
|
||||
"mesh-delivery-gate.source", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", "mesh-delivery-indexer.placement"}; !reflect.DeepEqual(sorted(m.Reads), want) {
|
||||
"mesh-delivery-assigner.placement", "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules"}; !reflect.DeepEqual(sorted(m.Reads), want) {
|
||||
t.Errorf("reads = %v, want %v", m.Reads, want)
|
||||
}
|
||||
if want := []string{"node-build-go", "node-build-image", "node-build-node"}; !reflect.DeepEqual(sorted(m.Uses), want) {
|
||||
|
||||
@@ -32,11 +32,11 @@
|
||||
"mesh-build-image.steps",
|
||||
"mesh-build-node.steps",
|
||||
"mesh-build-node.toolchains",
|
||||
"mesh-delivery-assigner.placement",
|
||||
"mesh-delivery-gate.source",
|
||||
"mesh-delivery-indexer.candidates",
|
||||
"mesh-delivery-indexer.manifests",
|
||||
"mesh-delivery-indexer.modules",
|
||||
"mesh-delivery-indexer.placement"
|
||||
"mesh-delivery-indexer.modules"
|
||||
],
|
||||
"uses": [
|
||||
"node-build-go",
|
||||
|
||||
@@ -17,7 +17,7 @@ import (
|
||||
|
||||
// 1945 (a part exits at start for a bucket it works without): a bus that refuses every toolchains bucket leaves the
|
||||
// watch unstarted and the part running, said; a refused bucket it cannot work without is an error naming the bucket and
|
||||
// its grant. The gate over the catalogue's placement, and its refusal, are mesh-sdk's (buildrun.ToolchainGate,
|
||||
// its grant. The gate over the seat's placement, and its refusal, are mesh-sdk's (buildrun.ToolchainGate,
|
||||
// issue 774).
|
||||
func TestARefusedBucketIsSaidAndThePartGoesOn(t *testing.T) {
|
||||
set := &buildrun.ToolchainSet{}
|
||||
|
||||
@@ -102,8 +102,8 @@ func TestManifestDeclarations(t *testing.T) { // 1678/AC1
|
||||
t.Errorf("state = %+v, want %+v", m.State, want)
|
||||
}
|
||||
if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-node.steps", "mesh-build-node.toolchains",
|
||||
"mesh-delivery-assigner.compositions", "mesh-delivery-builder.builds", "mesh-delivery-gate.source", "mesh-delivery-gate.turns", "mesh-delivery-indexer.candidates",
|
||||
"mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", "mesh-delivery-indexer.placement", "mesh-delivery-watchman.presence", "node-engine.have"}; !reflect.DeepEqual(sorted(m.Reads), want) {
|
||||
"mesh-delivery-assigner.compositions", "mesh-delivery-assigner.placement", "mesh-delivery-builder.builds", "mesh-delivery-gate.source", "mesh-delivery-gate.turns", "mesh-delivery-indexer.candidates",
|
||||
"mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", "mesh-delivery-watchman.presence", "node-engine.have"}; !reflect.DeepEqual(sorted(m.Reads), want) {
|
||||
t.Errorf("reads = %v, want %v", m.Reads, want)
|
||||
}
|
||||
// 1887/A5b: the checker reads nothing of the controller's plans-building (ADR 0483 superseded by ADR 0525 §7), calls no
|
||||
|
||||
@@ -35,13 +35,13 @@
|
||||
"mesh-build-node.steps",
|
||||
"mesh-build-node.toolchains",
|
||||
"mesh-delivery-assigner.compositions",
|
||||
"mesh-delivery-assigner.placement",
|
||||
"mesh-delivery-builder.builds",
|
||||
"mesh-delivery-gate.source",
|
||||
"mesh-delivery-gate.turns",
|
||||
"mesh-delivery-indexer.candidates",
|
||||
"mesh-delivery-indexer.manifests",
|
||||
"mesh-delivery-indexer.modules",
|
||||
"mesh-delivery-indexer.placement",
|
||||
"mesh-delivery-watchman.presence",
|
||||
"node-engine.have"
|
||||
],
|
||||
|
||||
@@ -621,7 +621,7 @@ func TestIndexerCountsOnlyThePlacedBuildNodesToolchains(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
// The indexer's placement, as the catalogue's files put it there (ADR 0549 §2): n1 places mesh-build-go, n2 nothing.
|
||||
// The seat mesh-placement's placement, as the assigner writes it (ADR 0567): n1 places mesh-build-go, n2 nothing.
|
||||
if placement, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.PlacementBucket, History: delivery.PlacementHistory}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
@@ -89,7 +89,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
|
||||
if want := []state{{Name: "candidates", History: 10}, {Name: "heard", History: 64}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}, {Name: "taken", History: 1}, {Name: "takeins", History: 1}}; !reflect.DeepEqual(m.State, want) {
|
||||
t.Errorf("state = %+v, want %+v", m.State, want)
|
||||
}
|
||||
if want := []string{"mesh-build-go.toolchains", "mesh-build-node.toolchains", "mesh-controller.builds", "mesh-delivery-checker.previews", "mesh-delivery-gate.source"}; !reflect.DeepEqual(sorted(m.Reads), want) {
|
||||
if want := []string{"mesh-build-go.toolchains", "mesh-build-node.toolchains", "mesh-controller.builds", "mesh-delivery-assigner.placement", "mesh-delivery-checker.previews", "mesh-delivery-gate.source"}; !reflect.DeepEqual(sorted(m.Reads), want) {
|
||||
t.Errorf("reads = %v, want %v", m.Reads, want)
|
||||
}
|
||||
// "Reads FORGE" is a consumes of the indexer's stream: the broker grants only buckets by reads (mesh-control
|
||||
|
||||
@@ -23,6 +23,7 @@
|
||||
"mesh-build-go.toolchains",
|
||||
"mesh-build-node.toolchains",
|
||||
"mesh-controller.builds",
|
||||
"mesh-delivery-assigner.placement",
|
||||
"mesh-delivery-checker.previews",
|
||||
"mesh-delivery-gate.source"
|
||||
],
|
||||
|
||||
@@ -68,7 +68,7 @@ func TestProcessWholeSeesTheBuckets(t *testing.T) { // 1134/AC8
|
||||
}
|
||||
return b
|
||||
}
|
||||
// The catalogue's placement (issue 774, ADR 0549 §3): x and y on a, nothing on b.
|
||||
// The seat's placement (issue 774, ADR 0549 §3): x and y on a, nothing on b.
|
||||
placeValue := func(node string, modules ...string) []byte {
|
||||
p := delivery.Placement{Node: node}
|
||||
for _, m := range modules {
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
// Command mesh-delivery-view is the operator's view of delivery (novox/hq ADR 0429 §3): it watches the have, the catalogue's placement
|
||||
// Command mesh-delivery-view is the operator's view of delivery (novox/hq ADR 0429 §3): it watches the have, the seat's placement
|
||||
// and builds buckets on its own bus credential, derives one row per module and node in memory, and serves them on a
|
||||
// local socket (for mesh-delivery-cli) and on a local web page. It starts, stops or retries nothing; the one thing it writes is the operator's answer to an ask that needs no proof of who
|
||||
// answered, under `<ask key>.view` in its bucket answers (ADR 0528 clause 6).
|
||||
|
||||
@@ -737,230 +737,3 @@ func (r Reader) Trusts() (TrustsAnswer, error) {
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// Placement is the indexer's placement as stored (ADR 0549 §2): one node's key with node, every node placing a module
|
||||
// with module (each node's key whole, so its pins, ports and bindings are beside the module), and every node's key with
|
||||
// neither. A key that does not read as a placement is listed as unreadable, never dropped.
|
||||
//
|
||||
// Beside it, candidates (issue 1126, ADR 0549 §4): each open catalogue pull request's placement of a node that differs from
|
||||
// the trunk's (the indexer's placement-candidates `<group>.<node>`), narrowed the same way, each with the node's answer from
|
||||
// its `<node>.node` have and whether that answer is for the candidate as it stands (its head and key revision).
|
||||
func (r Reader) Placement(node, module string) (map[string]any, error) {
|
||||
node, module = strings.TrimSpace(node), strings.TrimSpace(module)
|
||||
if r.PlacementState.Keys == nil || r.PlacementState.Get == nil {
|
||||
return nil, fmt.Errorf("the placement state cannot be read here")
|
||||
}
|
||||
candidates, err := r.PlacementCandidates(node, module)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if node != "" {
|
||||
key, err := delivery.PlacementKey(node)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
raw, err := r.PlacementState.Get(key)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if len(raw) == 0 && len(candidates) == 0 {
|
||||
return nil, fmt.Errorf("no placement of %s: the catalogue holds no placement file for it, or the indexer has read none yet", node)
|
||||
}
|
||||
out := map[string]json.RawMessage{}
|
||||
if len(raw) > 0 {
|
||||
out[node] = raw
|
||||
}
|
||||
return map[string]any{"placement": out, "candidates": candidates}, nil
|
||||
}
|
||||
keys, err := r.PlacementState.Keys()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
sort.Strings(keys)
|
||||
values, err := r.PlacementState.values(keys)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := map[string]json.RawMessage{}
|
||||
var unreadable []string
|
||||
for i, key := range keys {
|
||||
n, err := delivery.ParsePlacementKey(key)
|
||||
if err != nil {
|
||||
unreadable = append(unreadable, key)
|
||||
continue
|
||||
}
|
||||
p, err := delivery.DecodePlacement(values[i])
|
||||
if err != nil {
|
||||
unreadable = append(unreadable, key)
|
||||
continue
|
||||
}
|
||||
if module != "" && !slices.Contains(p.Assigned(), module) {
|
||||
continue
|
||||
}
|
||||
out[n] = values[i]
|
||||
}
|
||||
return map[string]any{"placement": out, "unreadable": unreadable, "candidates": candidates}, nil
|
||||
}
|
||||
|
||||
// The states of a node's answer to a placement candidate, as the view and the assigner's check read it.
|
||||
const (
|
||||
// AnswerNone: the node's have holds no answer for the candidate's group (a node asleep, or not composing yet).
|
||||
AnswerNone = "none"
|
||||
// AnswerStale: the node answered an earlier head or revision of the candidate.
|
||||
AnswerStale = "stale"
|
||||
)
|
||||
|
||||
// CandidateRow is one placement candidate and the node's answer to it.
|
||||
type CandidateRow struct {
|
||||
Group string `json:"group"`
|
||||
Node string `json:"node"`
|
||||
Pull delivery.Head `json:"pull"`
|
||||
Revision uint64 `json:"revision"`
|
||||
Placement json.RawMessage `json:"placement,omitempty"` // the candidate value as stored
|
||||
Removes bool `json:"removes,omitempty"` // the head removes the node's file
|
||||
Answered string `json:"answered"` // same, differs, refused, stale or none
|
||||
Answer *delivery.CandidateAnswer `json:"answer,omitempty"`
|
||||
}
|
||||
|
||||
// PlacementCandidates is every placement candidate, narrowed by node and by a module its placement names, with each
|
||||
// node's answer; beside each group's keys, a row for every other node placed at the group's head or on the trunk, whose
|
||||
// answer counts by the head alone (its file is unchanged, so it has no key, ADR 0559 §5, §7). A key that does not read is
|
||||
// a row naming only its key's group and node, answered none, never dropped.
|
||||
func (r Reader) PlacementCandidates(node, module string) ([]CandidateRow, error) {
|
||||
st := r.PlacementCandidatesState
|
||||
if st.Keys == nil || st.GetEntry == nil {
|
||||
return []CandidateRow{}, nil
|
||||
}
|
||||
keys, err := st.Keys()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("reading the placement candidates: %w", err)
|
||||
}
|
||||
sort.Strings(keys)
|
||||
// The trunk's placed nodes, each with its modules.
|
||||
trunk := map[string][]string{}
|
||||
if r.PlacementState.Keys != nil && r.PlacementState.Get != nil {
|
||||
if tkeys, err := r.PlacementState.Keys(); err == nil {
|
||||
for _, k := range tkeys {
|
||||
n, err := delivery.ParsePlacementKey(k)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
if raw, err := r.PlacementState.Get(k); err == nil && len(raw) > 0 {
|
||||
if p, err := delivery.DecodePlacement(raw); err == nil {
|
||||
trunk[n] = p.Assigned()
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
type group struct {
|
||||
pull delivery.Head
|
||||
keyed map[string]bool
|
||||
placed map[string]bool
|
||||
}
|
||||
groups := map[string]*group{}
|
||||
answers := map[string]map[string]delivery.CandidateAnswer{} // node, then group
|
||||
answerOf := func(n, g string) (delivery.CandidateAnswer, bool) {
|
||||
if _, read := answers[n]; !read {
|
||||
answers[n] = r.candidateAnswers(n)
|
||||
}
|
||||
a, ok := answers[n][g]
|
||||
return a, ok
|
||||
}
|
||||
rows := []CandidateRow{}
|
||||
for _, key := range keys {
|
||||
gname, n, err := delivery.ParsePlacementCandidateKey(key)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
raw, rev, err := st.GetEntry(key)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("reading %s: %w", key, err)
|
||||
}
|
||||
if len(raw) == 0 {
|
||||
continue
|
||||
}
|
||||
g := groups[gname]
|
||||
if g == nil {
|
||||
g = &group{keyed: map[string]bool{}, placed: map[string]bool{}}
|
||||
groups[gname] = g
|
||||
}
|
||||
g.keyed[n] = true
|
||||
row := CandidateRow{Group: gname, Node: n, Revision: rev, Placement: raw, Answered: AnswerNone}
|
||||
c, err := delivery.DecodePlacementCandidate(raw)
|
||||
if err == nil {
|
||||
g.pull = c.Pull
|
||||
if c.Placement != nil {
|
||||
g.placed[n] = true
|
||||
}
|
||||
row.Pull, row.Removes = c.Pull, c.Placement == nil
|
||||
if module != "" && (c.Placement == nil || !slices.Contains(c.Placement.Assigned(), module)) {
|
||||
continue
|
||||
}
|
||||
} else if module != "" {
|
||||
continue
|
||||
}
|
||||
if node != "" && n != node {
|
||||
continue
|
||||
}
|
||||
if a, ok := answerOf(n, gname); ok && err == nil {
|
||||
row.Answer = &a
|
||||
row.Answered = AnswerOf(c.Pull.Head, rev, true, a)
|
||||
}
|
||||
rows = append(rows, row)
|
||||
}
|
||||
for _, gname := range slices.Sorted(maps.Keys(groups)) {
|
||||
g := groups[gname]
|
||||
if g.pull.Head == "" {
|
||||
continue
|
||||
}
|
||||
others := map[string]bool{}
|
||||
for n := range trunk {
|
||||
others[n] = true
|
||||
}
|
||||
for n := range g.placed {
|
||||
others[n] = true
|
||||
}
|
||||
for _, n := range slices.Sorted(maps.Keys(others)) {
|
||||
if g.keyed[n] || (node != "" && n != node) || (module != "" && !slices.Contains(trunk[n], module)) {
|
||||
continue
|
||||
}
|
||||
row := CandidateRow{Group: gname, Node: n, Pull: g.pull, Answered: AnswerNone}
|
||||
if a, ok := answerOf(n, gname); ok {
|
||||
row.Answer = &a
|
||||
row.Answered = AnswerOf(g.pull.Head, 0, false, a)
|
||||
}
|
||||
rows = append(rows, row)
|
||||
}
|
||||
}
|
||||
return rows, nil
|
||||
}
|
||||
|
||||
// AnswerOf is the state of a node's answer a to its group's candidates at head: its verdict when it answers that head
|
||||
// and, for a node with a key of its own (keyed), that key's revision rev; AnswerStale otherwise.
|
||||
func AnswerOf(head string, rev uint64, keyed bool, a delivery.CandidateAnswer) string {
|
||||
if a.Head != head || (keyed && a.Revision != rev) {
|
||||
return AnswerStale
|
||||
}
|
||||
return a.Verdict
|
||||
}
|
||||
|
||||
// candidateAnswers is a node's answers to its candidates, by group, from its `<node>.node` have; none when it has none.
|
||||
func (r Reader) candidateAnswers(node string) map[string]delivery.CandidateAnswer {
|
||||
if r.HavesState.Get == nil {
|
||||
return nil
|
||||
}
|
||||
key, err := delivery.HaveNodeKey(node)
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
raw, err := r.HavesState.Get(key)
|
||||
if err != nil || len(raw) == 0 {
|
||||
return nil
|
||||
}
|
||||
h, err := delivery.DecodeHaveNode(raw)
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
return h.Candidates
|
||||
}
|
||||
|
||||
@@ -216,7 +216,7 @@ type UnreadableRow struct {
|
||||
// Model holds the rows' inputs in memory and nothing else: a restart fills it again from the buckets' last values.
|
||||
type Model struct {
|
||||
mu sync.Mutex
|
||||
placed *delivery.Placed // the catalogue's placement (the indexer's state, ADR 0549 §3)
|
||||
placed *delivery.Placed // the seat's placement (the assigner's state, ADR 0567, ADR 0549 §3)
|
||||
builds map[string]buildEntry // by <module>.build.<id>
|
||||
groups map[string]groupEntry // the indexer's candidates, by group
|
||||
modules map[string]moduleEntry // the indexer's modules, by module
|
||||
@@ -264,10 +264,10 @@ type stuckEntry struct {
|
||||
err string
|
||||
}
|
||||
|
||||
// pair is a module the catalogue's placement puts on a node.
|
||||
// pair is a module the seat's placement puts on a node.
|
||||
type pair struct{ Node, Module string }
|
||||
|
||||
// placedPairs is every module the catalogue's placement puts on a node, by node then module. The caller holds mu.
|
||||
// placedPairs is every module the seat's placement puts on a node, by node then module. The caller holds mu.
|
||||
func (m *Model) placedPairs() []pair {
|
||||
byNode := m.placed.ByNode()
|
||||
var out []pair
|
||||
@@ -529,7 +529,7 @@ func (m *Model) apply(e Entry) {
|
||||
st, err := delivery.DecodeStuck(e.Value)
|
||||
m.stuck[e.Key] = stuckEntry{value: st, err: errText(err)}
|
||||
case delivery.PlacementBucket:
|
||||
// A key that does not read places nothing on its node; the indexer's own read tool says why.
|
||||
// A key that does not read places nothing on its node; the seat's verb placement says why.
|
||||
_, _, _ = m.placed.Put(e.Key, e.Value, e.Deleted, e.At)
|
||||
case delivery.CandidatesBucket:
|
||||
if e.Deleted {
|
||||
|
||||
@@ -25,7 +25,7 @@ func haveE(t *testing.T, node, module string, v []byte, at time.Time) Entry {
|
||||
return Entry{Bucket: delivery.HaveBucket, Key: node + "." + module, Value: v, At: at}
|
||||
}
|
||||
|
||||
// place is the catalogue's placement of node, as the indexer keeps it (ADR 0549 §2): the modules placed there.
|
||||
// place is the seat's placement of node, as the assigner keeps it (ADR 0567): the modules placed there.
|
||||
func place(t *testing.T, node string, modules ...string) Entry {
|
||||
t.Helper()
|
||||
p := delivery.Placement{Node: node}
|
||||
@@ -101,7 +101,7 @@ func TestRowsPerAssignedModuleAndNode(t *testing.T) { // 1134/AC1
|
||||
}
|
||||
}
|
||||
|
||||
func TestLeftBehindIsWhatThePlacementDoesNotPut(t *testing.T) { // 1134/AC3, issue 774 (the catalogue's placement)
|
||||
func TestLeftBehindIsWhatThePlacementDoesNotPut(t *testing.T) { // 1134/AC3, issue 774 (the seat's placement)
|
||||
m := New()
|
||||
fill(m, base(t)...)
|
||||
fill(m, place(t, "b", "x")) // z taken off b
|
||||
|
||||
@@ -228,7 +228,7 @@ func TestARowReadsTheWantWhenTheAssignmentNamesNoBuild(t *testing.T) {
|
||||
// and a have nothing is assigned to keeps the module's name.
|
||||
func TestADottedModuleNameIsFoundAndKept(t *testing.T) {
|
||||
m := New()
|
||||
// The catalogue's placement refuses a dotted module name (mesh-sdk delivery.Placement), so such a module is never placed:
|
||||
// The seat's placement refuses a dotted module name (mesh-sdk delivery.Placement), so such a module is never placed:
|
||||
// its have is a row left behind under its own name.
|
||||
fill(m, place(t, "ace"), wantE(t, "ace", "novox.be", delivery.Want{FP: fpNew}), haveFP(t, "ace", "novox.be", fpNew, delivery.HealthOK),
|
||||
haveFP(t, "ace", "old.name", fpNew, delivery.HealthOK))
|
||||
|
||||
@@ -87,7 +87,7 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7
|
||||
if m.Module != "mesh-delivery-view" {
|
||||
t.Errorf("module = %q", m.Module)
|
||||
}
|
||||
want := []string{"mesh-build-go.pauses", "mesh-build-go.steps", "mesh-build-image.pauses", "mesh-build-image.steps", "mesh-build-node.pauses", "mesh-build-node.steps", "mesh-config.resolved", "mesh-controller.builds", "mesh-controller.composed", "mesh-delivery-assigner.asks", "mesh-delivery-assigner.differences", "mesh-delivery-assigner.wants", "mesh-delivery-builder.builds", "mesh-delivery-checker.tries", "mesh-delivery-gate.asks", "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.heard", "mesh-delivery-indexer.modules", "mesh-delivery-indexer.taken", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-delivery-watchman.stuck", "mesh-vault.signers", "node-engine.have"} // build_usage reads the build modules' steps and pauses, mesh-config's capacities and the checker's tries; 1168/AC1, 1241/AC4, 1345/AC3, 1297/B8, B9
|
||||
want := []string{"mesh-build-go.pauses", "mesh-build-go.steps", "mesh-build-image.pauses", "mesh-build-image.steps", "mesh-build-node.pauses", "mesh-build-node.steps", "mesh-config.resolved", "mesh-controller.builds", "mesh-controller.composed", "mesh-delivery-assigner.asks", "mesh-delivery-assigner.differences", "mesh-delivery-assigner.placement", "mesh-delivery-assigner.wants", "mesh-delivery-builder.builds", "mesh-delivery-checker.tries", "mesh-delivery-gate.asks", "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.heard", "mesh-delivery-indexer.modules", "mesh-delivery-indexer.taken", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-delivery-watchman.stuck", "mesh-vault.signers", "node-engine.have"} // build_usage reads the build modules' steps and pauses, mesh-config's capacities and the checker's tries; 1168/AC1, 1241/AC4, 1345/AC3, 1297/B8, B9
|
||||
got := slices.Clone(m.Reads)
|
||||
slices.Sort(got)
|
||||
if !reflect.DeepEqual(got, want) {
|
||||
|
||||
@@ -31,7 +31,6 @@
|
||||
{ "name": "reach", "history": 1 }
|
||||
],
|
||||
"reads": [
|
||||
"mesh-controller.builds",
|
||||
"mesh-build-go.pauses",
|
||||
"mesh-build-go.steps",
|
||||
"mesh-build-image.pauses",
|
||||
@@ -39,9 +38,11 @@
|
||||
"mesh-build-node.pauses",
|
||||
"mesh-build-node.steps",
|
||||
"mesh-config.resolved",
|
||||
"mesh-controller.builds",
|
||||
"mesh-controller.composed",
|
||||
"mesh-delivery-assigner.asks",
|
||||
"mesh-delivery-assigner.differences",
|
||||
"mesh-delivery-assigner.placement",
|
||||
"mesh-delivery-assigner.wants",
|
||||
"mesh-delivery-builder.builds",
|
||||
"mesh-delivery-checker.tries",
|
||||
|
||||
@@ -2059,7 +2059,7 @@ func TestGrants(t *testing.T) {
|
||||
if slices.Contains(m.Reads, grants.ReadAssignments) {
|
||||
reads[m.Module] = true
|
||||
if !allowed[m.Module] {
|
||||
t.Errorf("%s reads %s: read where modules run from the indexer's placement (ADR 0549 §3); the lease write "+
|
||||
t.Errorf("%s reads %s: read where modules run from the seat mesh-placement's placement (ADR 0549 §3); the lease write "+
|
||||
"is refused while a module reads the assignments (ADR 0549 §6)", m.Module, grants.ReadAssignments)
|
||||
}
|
||||
}
|
||||
@@ -2072,7 +2072,7 @@ func TestGrants(t *testing.T) {
|
||||
})
|
||||
|
||||
// 774 (the L3 review, ADR 0549 §3, §6): once the controller's lease is handed away the assignments are frozen and place
|
||||
// nothing; the catalogue's placement alone does. A module left only in the stale assignments loses its account and its
|
||||
// nothing; the seat's placement alone does. A module left only in the stale assignments loses its account and its
|
||||
// grant names no account of it; before the lease write the two are still unioned.
|
||||
t.Run("placement alone once the lease is handed away: a module only in the stale assignments has no account or grant of it", func(t *testing.T) {
|
||||
b := fixture()
|
||||
|
||||
@@ -107,7 +107,7 @@ type resolved struct {
|
||||
type Inputs struct {
|
||||
// Nodes is every machine seen: in a node agent's have or a standing assignment.
|
||||
Nodes []string
|
||||
// Assigned is each machine's standing assignments and what the catalogue's placement puts there, sorted: while both
|
||||
// Assigned is each machine's standing assignments and what the seat's placement puts there, sorted: while both
|
||||
// stand (ADR 0549 §5) a module either names keeps its membership.
|
||||
Assigned map[string][]string
|
||||
// Placed is each node's placement as the seat mesh-placement holds it (ADR 0567 §1); empty before any is read.
|
||||
@@ -290,7 +290,7 @@ func ReadInputs(r Reader, cache *Manifests) Inputs {
|
||||
away, _ := leaseHandedAway(r)
|
||||
if away && len(in.Placed) > 0 {
|
||||
in.Gaps = append(in.Gaps, Gap{Input: ReadAssignments, Why: "not used: the controller's lease is handed away, " +
|
||||
"so the catalogue's placement alone places modules (ADR 0549 §3, §6)"})
|
||||
"so the seat's placement alone places modules (ADR 0549 §3, §6)"})
|
||||
} else {
|
||||
for node, ms := range byAssignment {
|
||||
nodes[node] = true
|
||||
|
||||
Reference in New Issue
Block a user