Assigner: hold the seat mesh-wants, keep a person's upgrade policy and judge moves by it, say a policy split with the controller as a finding, and answer the bus step, so the policies and the bus step leave the controller (issue 1953 D4-D6)

This commit is contained in:
2026-10-29 22:34:08 +01:00
parent 2488d839ab
commit 3aec7f9b02
13 changed files with 686 additions and 20 deletions
@@ -85,7 +85,7 @@ const goBuildModule = "mesh-build-go"
// (ADR 0567 §1), named after the assigner as every module state is.
var ownBuckets = []string{delivery.AssignerAsksBucket, delivery.AssignerCompositionsBucket, delivery.AssignerDifferencesBucket,
delivery.AssignerPutsBucket, delivery.AssignerWantsBucket, delivery.PlacementBucket, delivery.PlacementProposalsBucket,
delivery.PlacementCandidatesBucket, delivery.PlacementPulsesBucket, assign.FindingsBucket}
delivery.PlacementCandidatesBucket, delivery.PlacementPulsesBucket, assign.FindingsBucket, assign.PoliciesBucket}
// 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).
@@ -108,6 +108,7 @@ var keptStates = []announce.State{
{Name: "puts", Readable: true},
{Name: "wants", Readable: true},
{Name: health.FindingsState, Readable: true},
{Name: delivery.AssignerPoliciesState, Readable: true},
}
// buckets is every bucket the assigner opened, by name.
@@ -156,6 +157,12 @@ 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 == assign.PoliciesBucket && err != nil {
// The seat mesh-wants' state, new with it (ADR 0565 §7): bound once the bus says it was made; until then no
// person's policy is read, so the controller's beside each composition holds, as before.
logf("the bucket %s cannot be opened (%v): it is bound once the bus says it was made", name, err)
continue
}
if (name == assign.FindingsBucket || name == takeoverBucket) && err != nil {
// New with health leaving the controller (issue 1952 WM6): without them the conditions are said straight to the
// router and kept as no finding, as before.
@@ -210,7 +217,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
}
thisNode = account.Node
logger.Printf("started on the account %s, reading %d buckets", account.User, len(kvs))
return follow(ctx, js, conn, kvs, findingsOf(ctx, kvs, logger.Printf).Say(sayVia(ctx, js)), emitting{said}, logger)
findings := findingsOf(ctx, kvs, logger.Printf)
return follow(ctx, js, conn, kvs, findings.Say(sayVia(ctx, js)), findings, emitting{said}, logger)
}
// findingsOf is where the assigner's said conditions go as findings (issue 1952 WM6): its findings bucket, and the
@@ -222,6 +230,13 @@ func findingsOf(ctx context.Context, kvs buckets, logf func(string, ...any)) ass
if kv := kvs[assign.FindingsBucket]; kv != nil {
f.Put = func(key string, value []byte) error { _, err := kv.Put(ctx, key, value); return err }
f.Delete = func(key string) error { return kv.Delete(ctx, key) }
f.Keys = func() ([]string, error) {
keys, err := kv.Keys(ctx)
if errors.Is(err, jetstream.ErrNoKeysFound) {
return nil, nil
}
return keys, err
}
}
var keeper atomic.Bool
f.ToKeeper = keeper.Load
@@ -297,11 +312,12 @@ var watched = []string{
delivery.AssignerCompositionsBucket,
delivery.AssignerDifferencesBucket,
delivery.AssignerWantsBucket,
assign.PoliciesBucket,
}
// follow watches every bucket a pass reads and makes a pass at start, once every watch handed what stands, and after
// each change; changes that come during a pass make one more. It returns when ctx ends or a watch ends.
func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, kvs buckets, say func(asks.Condition) error, bus caller, logger *log.Logger) error {
func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, kvs buckets, say func(asks.Condition) error, findings assign.Findings, bus caller, logger *log.Logger) error {
w := newWorld()
dirty := make(chan struct{}, 1)
wake := func() {
@@ -393,7 +409,7 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
}
started := 0
for _, name := range watched {
if kvs[name] == nil && (slices.Contains(seatBuckets, name) || sourceOfGrants(name) || name == provider.AssignmentsBucket) {
if kvs[name] == nil && (slices.Contains(seatBuckets, name) || sourceOfGrants(name) || name == provider.AssignmentsBucket || name == assign.PoliciesBucket) {
continue // not on the bus at this start (open said so)
}
started++
@@ -423,10 +439,10 @@ 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, buscompose.LeaseBucket} {
for _, name := range []string{delivery.PartGrantsBucket, buscompose.LeaseBucket, assign.PoliciesBucket} {
if kvs[name] == nil && js != nil {
name := name
process.AwaitBucket(ctx, js, nc, name, "the controller's grants are read",
process.AwaitBucket(ctx, js, nc, name, "until then the bucket is not read",
logger.Printf).Ready(func(kv jetstream.KeyValue) { placementAppeared <- appeared{name, kv} })
}
}
@@ -447,7 +463,7 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
case <-settled:
}
}
p := &passer{kvs: kvs, w: w, logf: logger.Printf, say: say, said: map[string]bool{}, answeredAt: map[string]uint64{}}
p := &passer{kvs: kvs, w: w, logf: logger.Printf, say: say, findings: findings, said: map[string]bool{}, answeredAt: map[string]uint64{}}
p.asks = newComposeAsks(ctx, buildrun.Asker{Publisher: buildrun.JetStreamAsks{JS: js}, Waiter: buildrun.JetStreamAsks{JS: js}}, wake, logger.Printf)
if bus != nil && kvs[delivery.AssignerPutsBucket] != nil {
p.acts = newPutActs(bus, wake, logger.Printf, w.handedAway)
@@ -1,6 +1,8 @@
package main
import (
"mesh-delivery-assigner/internal/assign"
"context"
"errors"
"fmt"
@@ -110,7 +112,7 @@ func TestOpensWhatItsManifestNames(t *testing.T) {
kvs[delivery.ControllerComposedBucket].(*watchKV).entries = []kvEntry{{key: ckey, value: cd}}
kvs[delivery.AssignerPutsBucket].(*watchKV).closed = true
kvs[delivery.AssignerWantsBucket].(*watchKV).wrote = wrote
go follow(ctx, nil, nil, kvs, func(asks.Condition) error { return nil }, nil, log.New(io.Discard, "", 0))
go follow(ctx, nil, nil, kvs, func(asks.Condition) error { return nil }, assign.Findings{}, nil, log.New(io.Discard, "", 0))
select {
case key := <-wrote:
if key != "want.ace.node" {
@@ -27,6 +27,10 @@ type passer struct {
// pass says a condition once and a restart says it again (the router keeps one per key).
say func(asks.Condition) error
said map[string]bool
// findings is the assigner's findings bucket (its said conditions and policy-split, ADR 0564 §2, ADR 0565 §8); splits the
// modules whose policy-split finding stands, nil until the first pass, which deletes any a previous process left.
findings assign.Findings
splits map[string]bool
// answeredAt is, by the id of a backup ask, how many values the have had when its answer was written into the want.
answeredAt map[string]uint64
asks *composeAsks
@@ -129,6 +133,8 @@ func (p *passer) wants(ctx context.Context) {
}
}
wanted := p.w.wantsOf()
chosen, _ := assign.DecodePolicies(p.w.read(assign.PoliciesBucket))
defer p.policySplits(manifests, chosen)
frozen := p.sayPlacement(p.withCalls(p.w.placementDiffers()))
placed := p.w.placedNodes()
puts := p.putsByModule()
@@ -154,7 +160,7 @@ func (p *passer) wants(ctx context.Context) {
for name := range names {
fp := trunk[name].FP2
name := name
m := assign.Module{Name: name, FP: fp, Manifest: manifests[name], Nodes: nodesOf[name], Now: time.Now(), Puts: puts[name],
m := assign.Module{Name: name, FP: fp, Manifest: manifests[name], Chosen: chosenOf(chosen, name), Nodes: nodesOf[name], Now: time.Now(), Puts: puts[name],
Self: func(node string) bool { return placed[node] },
Composed: func(node, fp string) (delivery.ComposedDigest, bool) {
if placed[node] {
@@ -932,3 +938,59 @@ func (p *passer) yeses(ctx context.Context, module string, m assign.Module, befo
}
}
}
// chosenOf is a person's choice for module, nil when none.
func chosenOf(chosen map[string]catalogue.PersonPolicy, module string) *catalogue.PersonPolicy {
if c, ok := chosen[module]; ok {
return &c
}
return nil
}
// policySplits writes the finding policy-split for each module whose policy at the assigner differs from one the controller
// writes beside a composition of it, and deletes it once they agree (ADR 0565 §8): written when a split begins, never again
// while it stands. The first pass after a start deletes every one a previous process left that no longer splits.
func (p *passer) policySplits(manifests map[string]catalogue.Manifest, chosen map[string]catalogue.PersonPolicy) {
if p.findings.Put == nil {
return
}
now := time.Now
if p.now != nil {
now = p.now
}
split := map[string]bool{}
for _, s := range assign.Splits(manifests, chosen, assign.ControllerPolicies(p.w.read(delivery.ControllerComposedBucket))) {
split[s.Module] = true
if p.splits[s.Module] {
continue
}
value, err := assign.PolicySplitFinding(s, now())
if err == nil {
err = p.findings.Put(assign.PolicySplitKey(s.Module), value)
}
if err != nil {
p.logf("the policy split of %s is not written: %v", s.Module, err)
delete(split, s.Module)
}
}
gone := p.splits
if gone == nil && p.findings.Keys != nil {
gone = map[string]bool{}
if keys, err := p.findings.Keys(); err == nil {
for _, k := range keys {
if m, ok := strings.CutSuffix(strings.TrimPrefix(k, "module."), "."+assign.KindPolicySplit); ok && strings.HasPrefix(k, "module.") {
gone[m] = true
}
}
}
}
for m := range gone {
if !split[m] {
if err := p.findings.Delete(assign.PolicySplitKey(m)); err != nil {
p.logf("the policy split of %s is not deleted: %v", m, err)
split[m] = true // tried again at the next pass
}
}
}
p.splits = split
}
@@ -186,7 +186,10 @@ func (p *passer) putWorld(node, module string, put delivery.Put, manifest catalo
w.Composed, w.Granted = p.composedFor(node, module, put.FP)
if d := w.Composed; d != nil {
policy := d.Policy
if policy == "" {
chosen, _ := assign.DecodePolicies(p.w.read(assign.PoliciesBucket))
if c, ok := chosen[module]; ok {
policy, _, _, _ = catalogue.Upgrade(manifest, &c) // a person's choice at the assigner (ADR 0565 §7)
} else if policy == "" {
policy, _, _ = catalogue.DerivedUpgrade(manifest)
}
w.Recorded = policy == catalogue.PolicyRecord
@@ -12,6 +12,8 @@ import (
"git.novox.be/novox/mesh-sdk/go/buscompose"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/delivery"
"mesh-delivery-assigner/internal/assign"
)
func (s memStore) del(state, key string) error {
@@ -155,6 +157,118 @@ func TestTheSeatsVerbs(t *testing.T) {
if tools := s.tools(); len(tools) != 7 || tools[2].Name != "mesh-placement.propose" || tools[2].Run == nil || tools[2].Input == nil {
t.Errorf("the seat's verbs as served: %+v", tools)
}
t.Run("1953/D4-D6: the seat mesh-wants' upgrade and bus", wantsVerbs)
}
// wantsVerbs (novox/hq ADR 0565 §7, §8, §9): upgrade answers each module's policy and its source; a person's choice is
// written with who and why, recorded as a hand act in the seat's log and said as policy-changed, and default takes it back;
// the bus is refused; compare lists the modules the controller writes another policy for; the process writes policy-split
// for a split and deletes it once they agree; bus names the node behind the indexer's newest.
func wantsVerbs(t *testing.T) {
fpOld, fpNew := delivery.SourcePrefix2+strings.Repeat("1", 64), delivery.SourcePrefix2+strings.Repeat("2", 64)
st := memStore{}
for _, state := range []string{modulesState, manifestsState, composedState, policiesState, haveState, putsState, "findings"} {
st[state] = &memKV{values: map[string][]byte{}}
}
manifest := func(name, extra string) {
raw := `{"module":"` + name + `","version":"1"` + extra + `}`
digest := delivery.ManifestDigest([]byte(raw))
st[manifestsState].values[digest], _ = delivery.EncodeManifest(delivery.Manifest{V: delivery.Version, Module: name, JSON: raw})
st[modulesState].values[name], _ = delivery.EncodeModule(delivery.Module{V: delivery.Version, FP2: fpNew, Manifest: digest, Repository: "novox/x", Commit: "c"})
}
manifest("web", "")
manifest("photos", `,"upgrade":{"policy":"record","why":"a person checks the albums"}`)
manifest("nats", `,"provides":[{"name":"mesh-bus","scope":"mesh"}]`)
for node, fp := range map[string]string{"anchor": fpOld, "laptop": fpNew} {
st[haveState].values[node+".nats"], _ = delivery.EncodeHave(delivery.Have{FP: fp, Phase: delivery.PhaseRunning, Health: delivery.HealthOK})
}
composed := func(module, policy string) {
key, _ := delivery.ComposedKey("anchor", module, fpNew)
st[composedState].values[key], _ = json.Marshal(delivery.ComposedDigest{V: delivery.Version, Digest: "sha256:" + strings.Repeat("c", 64),
Judgement: delivery.JudgedNeither, Policy: policy, PolicyFrom: catalogue.FromPerson, PolicyWhy: "chosen at the controller"})
}
composed("web", catalogue.PolicyRecord)
composed("photos", catalogue.PolicyRecord)
var asked []string
var said []string
w := newWantsTools(st, func(key string, body any) (json.RawMessage, error) {
asked = append(asked, key)
return json.RawMessage(`{"ok":true}`), nil
}, func(event, id string, body any) error { said = append(said, event+" "+id); return nil })
policyOf := func(module string) policyRow {
out, err := w.doUpgrade(map[string]any{"module": module}, "op")
if err != nil {
t.Fatal(err)
}
return out.(map[string]any)["modules"].([]policyRow)[0]
}
if p := policyOf("photos"); p.Policy != catalogue.PolicyRecord || p.From != catalogue.FromModule {
t.Errorf("photos = %+v", p)
}
if p := policyOf("nats"); p.Policy != catalogue.PolicyRecord || p.From != catalogue.FromBus {
t.Errorf("nats = %+v", p)
}
// compare: web rolls at the assigner and records at the controller.
out, _ := w.doUpgrade(map[string]any{"compare": "true"}, "op")
if splits := out.(map[string]any)["splits"].([]assign.Split); len(splits) != 1 || splits[0].Module != "web" || out.(map[string]any)["findings"] == nil {
t.Errorf("splits = %+v", splits)
}
// The process says the split as its finding, and deletes it once they agree.
findings := map[string][]byte{}
p := &passer{w: newWorld(), logf: func(string, ...any) {}, findings: assign.Findings{
Put: func(k string, v []byte) error { findings[k] = v; return nil },
Delete: func(k string) error { delete(findings, k); return nil },
Keys: func() ([]string, error) { return []string{"module.gone.policy-split"}, nil }}}
for k, v := range st[composedState].values {
p.w.putEntry(delivery.ControllerComposedBucket, k, v, 1, time.Now())
}
man := map[string]catalogue.Manifest{}
for _, name := range []string{"web", "photos"} {
raw, _, _, _ := st.get(modulesState, name)
mod, _ := delivery.DecodeModule(raw)
mraw, _, _, _ := st.get(manifestsState, mod.Manifest)
kept, _ := delivery.DecodeManifest(mraw)
man[name], _ = catalogue.ParseManifest([]byte(kept.JSON))
}
findings["module.gone.policy-split"] = []byte("{}")
p.policySplits(man, nil)
if _, ok := findings[assign.PolicySplitKey("web")]; !ok || len(findings) != 1 {
t.Errorf("findings = %v, want web's split alone (a stale one deleted)", findings)
}
// a person's record for web: written, recorded, said; the split is gone.
if _, err := w.doUpgrade(map[string]any{"module": "web", "policy": "record", "why": "the operator checks it"}, "person.jochen"); err != nil {
t.Fatal(err)
}
chosen, _ := assign.DecodePolicies(st[policiesState].values)
if c := chosen["web"]; c.Policy != catalogue.PolicyRecord || c.By != "person.jochen" || !slices.Contains(asked, "seat:mesh-hand-acts.hand-act") ||
len(said) != 1 || !strings.HasPrefix(said[0], "policy-changed policy-changed.web.") {
t.Errorf("chosen %+v, asked %v, said %v", c, asked, said)
}
p.policySplits(man, chosen)
if len(findings) != 0 {
t.Errorf("findings after the choice = %v", findings)
}
if _, err := w.doUpgrade(map[string]any{"module": "nats", "policy": "roll", "why": "x"}, "op"); err == nil {
t.Error("the bus was given a policy")
}
if _, err := w.doUpgrade(map[string]any{"module": "web", "policy": "roll"}, "op"); err == nil {
t.Error("a policy without why was set")
}
if _, err := w.doUpgrade(map[string]any{"module": "web", "policy": "default", "why": "back to the module"}, "op"); err != nil || len(st[policiesState].values) != 0 {
t.Errorf("default left %v (%v)", st[policiesState].values, err)
}
// bus: the node behind the indexer's newest.
b, err := w.doBus(nil)
if err != nil {
t.Fatal(err)
}
nodes := b.(map[string]any)["nodes"].([]busNode)
if b.(map[string]any)["module"] != "nats" || len(nodes) != 2 || !nodes[0].Behind || nodes[1].Behind || !slices.Contains(asked, "seat:mesh-hand-acts.hand-acts") {
t.Errorf("bus = %+v, asked %v", b, asked)
}
if tools := w.tools(); len(tools) != 2 || tools[0].Name != "mesh-wants.upgrade" || tools[1].Run == nil {
t.Errorf("the seat's verbs as served: %+v", tools)
}
}
// readManifest is the module's manifest, read as the controller reads it.
@@ -351,7 +351,12 @@ func serveTools() error {
time.Sleep(wait)
}
}()
return stdio.Serve("", append(s.tools(), newSeatTools(runtimeStore{}, stdio.Ask).tools()...))
tools := append(s.tools(), newSeatTools(runtimeStore{}, stdio.Ask).tools()...)
tools = append(tools, newWantsTools(runtimeStore{}, stdio.Ask, func(e, id string, b any) error {
// The seat's event, said on the seat: a claimed seat's event is never the module's own.
return stdio.SeatPublish(wantsEventSubject(e), b, id)
}).tools()...)
return stdio.Serve("", tools)
}
// shortFP is a fingerprint's first twelve hex digits.
@@ -0,0 +1,300 @@
package main
import (
"encoding/json"
"errors"
"fmt"
"sort"
"strconv"
"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"
"git.novox.be/novox/mesh-sdk/go/handacts"
"git.novox.be/novox/mesh-sdk/go/health"
"mesh-delivery-assigner/internal/assign"
)
// The seat mesh-wants' verbs (novox/hq ADR 0565 §7, §8, §9, §15; issue 1953 D4-D6), served through the tool runner beside the
// put tools and mesh-placement's verbs. `upgrade` reads and sets a person's upgrade policy, kept in the state `policies`
// (each set recorded as a hand act and said as policy-changed), and with compare lists every module whose policy differs from
// the one the controller writes beside its compositions; `bus` says what a bus upgrade would do. The assigner's process reads
// `policies` when it judges a move and writes the finding policy-split while a split stands.
// The states the verbs read, as the runtime names them.
const (
policiesState = delivery.AssignerPoliciesState
manifestsState = delivery.IndexerModule + ".manifests"
composedState = "mesh-controller.composed"
// PolicyChangedEvent is the seat's event of a person's policy set or taken back.
PolicyChangedEvent = "policy-changed"
// busStepCause is the cause the bus's step is recorded under in the hand-act log.
busStepCause = "bus-upgrade"
)
// wantsTools serves the seat's verbs over a store, asking seats through ask and saying events through emit.
type wantsTools struct {
st seatStore
ask func(key string, body any) (json.RawMessage, error)
emit func(event, id string, body any) error
now func() time.Time
}
// wantsEventSubject is the subject of one of the seat's events.
func wantsEventSubject(event string) string {
return "mesh.seat." + catalogue.WantsSeat + ".event." + event
}
func newWantsTools(st seatStore, ask func(string, any) (json.RawMessage, error), emit func(event, id string, body any) error) *wantsTools {
return &wantsTools{st: st, ask: ask, emit: emit, now: time.Now}
}
// leased says the controller's lease is handed away.
func (s *wantsTools) leased() bool {
raw, _, found, err := s.st.get(buscompose.ControllerLease, buscompose.ControllerLeaseKey)
away, _ := buscompose.LeaseHandedAway(raw, found && raw != nil, err)
return away
}
// values reads every key of a state.
func (s *wantsTools) values(state string) (map[string][]byte, error) {
keys, err := s.st.keys(state)
if err != nil {
return nil, fmt.Errorf("%s cannot be listed: %w", state, err)
}
out := map[string][]byte{}
for _, k := range keys {
raw, _, found, err := s.st.get(state, k)
if err != nil {
return nil, fmt.Errorf("%s %s cannot be read: %w", state, k, err)
}
if found {
out[k] = raw
}
}
return out, nil
}
// manifests is every module the indexer holds, with its trunk manifest and fingerprint.
func (s *wantsTools) manifests() (map[string]catalogue.Manifest, map[string]delivery.Module, error) {
mods, err := s.values(modulesState)
if err != nil {
return nil, nil, err
}
out, held := map[string]catalogue.Manifest{}, map[string]delivery.Module{}
for name, raw := range mods {
m, err := delivery.DecodeModule(raw)
if err != nil {
continue
}
held[name] = m
if m.Manifest == "" {
continue
}
mraw, _, found, err := s.st.get(manifestsState, m.Manifest)
if err != nil || !found {
continue
}
kept, err := delivery.DecodeManifest(mraw)
if err != nil {
continue
}
if man, err := catalogue.ParseManifest([]byte(kept.JSON)); err == nil {
out[name] = man
}
}
return out, held, nil
}
// policyRow is one module's policy as upgrade answers it.
type policyRow struct {
Module string `json:"module"`
Policy string `json:"policy"`
From string `json:"from"`
Why string `json:"why,omitempty"`
By string `json:"by,omitempty"`
}
// doUpgrade is the verb upgrade: read every module's policy or one's; with policy and why set a person's choice, or take it
// back with default; with compare list the splits.
func (s *wantsTools) doUpgrade(args map[string]any, by string) (any, error) {
manifests, _, err := s.manifests()
if err != nil {
return nil, err
}
raw, err := s.values(policiesState)
if err != nil {
return nil, err
}
chosen, unreadable := assign.DecodePolicies(raw)
if arg(args, "compare") == "true" {
composed, err := s.values(composedState)
if err != nil {
return nil, err
}
splits := assign.Splits(manifests, chosen, assign.ControllerPolicies(composed))
if splits == nil {
splits = []assign.Split{}
}
out := map[string]any{"splits": splits, "empty": len(splits) == 0}
// The assigner's findings, its said conditions and its policy splits as the condition keeper reads them (§8, §15).
if found, err := s.values(health.FindingsState); err != nil {
out["findings-unread"] = err.Error()
} else {
keys := []string{}
for k := range found {
keys = append(keys, k)
}
sort.Strings(keys)
out["findings"] = keys
}
return out, nil
}
module, policy := arg(args, "module"), arg(args, "policy")
if policy != "" {
return s.setPolicy(module, policy, arg(args, "why"), by, manifests)
}
var rows []policyRow
for name, m := range manifests {
if module != "" && name != module {
continue
}
p, from, why, who := assign.PolicyOf(m, chosen)
rows = append(rows, policyRow{Module: name, Policy: p, From: from, Why: why, By: who})
}
if module != "" && len(rows) == 0 {
return nil, fmt.Errorf("the indexer holds no manifest of %s", module)
}
sort.Slice(rows, func(i, j int) bool { return rows[i].Module < rows[j].Module })
return map[string]any{"modules": rows, "unreadable": unreadable}, nil
}
// setPolicy writes a person's choice of a module's policy, or deletes it for default, records the act in the hand-act log
// and says policy-changed. The bus is never rolled: a choice for it is refused.
func (s *wantsTools) setPolicy(module, policy, why, by string, manifests map[string]catalogue.Manifest) (any, error) {
if module == "" {
return nil, errors.New("a policy is set for one module: name it")
}
m, ok := manifests[module]
if !ok {
return nil, fmt.Errorf("the indexer holds no manifest of %s", module)
}
if catalogue.ProvidesBus(m) {
return nil, fmt.Errorf("%s is the bus, which is never rolled whoever says so (novox/hq ADR 0236): its step is the put on the operator's warrant", module)
}
if why == "" {
return nil, errors.New("a person's policy says why")
}
if by == "" || by == stdio.Unproven {
return nil, errors.New("a person's policy names who chose it, and the runtime proved no caller")
}
what := fmt.Sprintf("mesh-wants.upgrade %s %s", module, policy)
if policy == "default" {
if err := s.st.del(policiesState, module); err != nil {
return nil, fmt.Errorf("the choice for %s cannot be taken back: %w", module, err)
}
} else {
p := catalogue.PersonPolicy{V: catalogue.PersonPolicyVersion, Module: module, Policy: policy, Why: why, By: by, At: s.now().UTC()}
if err := p.Check(); err != nil {
return nil, err
}
raw, _ := json.Marshal(p)
if err := s.st.put(policiesState, module, raw); err != nil {
return nil, fmt.Errorf("the choice for %s cannot be written: %w", module, err)
}
}
rec := handacts.Recorder{Ask: s.ask, LeaseAway: s.leased}
taken, rerr := rec.Record(handacts.Request{What: what, Why: why, Cause: "upgrade-policy", Verb: "mesh-wants.upgrade"})
recorded := taken.Log
if rerr != nil {
recorded = "not recorded: " + rerr.Error()
}
body := map[string]any{"module": module, "policy": policy, "why": why, "by": by}
said := ""
if s.emit != nil {
id := PolicyChangedEvent + "." + module + "." + strconv.FormatInt(s.now().UnixNano(), 10)
if err := s.emit(PolicyChangedEvent, id, body); err != nil {
said = "not said: " + err.Error()
}
}
return map[string]any{"set": body, "hand-act": recorded, "event": said}, nil
}
// busNode is one node's bus as bus answers it.
type busNode struct {
Node string `json:"node"`
FP string `json:"fp"`
Behind bool `json:"behind"`
}
// doBus is the verb bus: the fingerprint of the bus module each node runs against the indexer's newest, the nodes behind,
// the last bus steps of the hand-act log and the assigner's puts of the bus. Read-only.
func (s *wantsTools) doBus(map[string]any) (any, error) {
manifests, held, err := s.manifests()
if err != nil {
return nil, err
}
bus := ""
for name, m := range manifests {
if catalogue.ProvidesBus(m) {
bus = name
}
}
if bus == "" {
return nil, errors.New("the indexer holds no module that provides the bus")
}
newest := held[bus].FP2
haves, err := s.values(haveState)
if err != nil {
return nil, err
}
var nodes []busNode
for key, raw := range haves {
node, module, isNode, err := delivery.ParseHaveKey(key)
if err != nil || isNode || module != bus {
continue
}
h, err := delivery.DecodeHave(raw)
if err != nil || h.Phase == delivery.PhaseRemoved {
continue
}
nodes = append(nodes, busNode{Node: node, FP: h.FP, Behind: newest != "" && h.FP != newest})
}
sort.Slice(nodes, func(i, j int) bool { return nodes[i].Node < nodes[j].Node })
out := map[string]any{"module": bus, "newest": newest, "nodes": nodes}
if steps, err := s.ask("seat:"+handacts.Seat+".hand-acts", map[string]any{"cause": busStepCause, "days": 90}); err != nil {
out["steps-unread"] = err.Error()
} else {
out["steps"] = steps
}
puts, err := s.values(putsState)
if err != nil {
out["puts-unread"] = err.Error()
} else {
mine := map[string]json.RawMessage{}
for k, v := range puts {
if _, module, err := delivery.ParsePutKey(k); err == nil && module == bus {
mine[k] = v
}
}
out["puts"] = mine
}
return out, nil
}
// tools are the seat's verbs as the runtime serves them, each named <seat>.<verb>, with the seat's own schemas.
func (s *wantsTools) tools() []stdio.Tool {
seat, _ := catalogue.SeatNamed(catalogue.WantsSeat)
run := map[string]func(map[string]any) (any, error){
"upgrade": func(a map[string]any) (any, error) { return s.doUpgrade(a, stdio.CallerOf(a)) },
"bus": s.doBus,
}
var out []stdio.Tool
for _, v := range seat.Serves {
out = append(out, stdio.Tool{Name: catalogue.WantsSeat + "." + v.Name, Description: v.Description, Input: v.Input, Run: run[v.Name]})
}
return out
}
@@ -64,6 +64,7 @@ func FindingOf(c asks.Condition, at time.Time) (key string, value []byte, raised
type Findings struct {
Put func(key string, value []byte) error
Delete func(key string) error
Keys func() ([]string, error)
ToKeeper func() bool
Now func() time.Time
}
@@ -0,0 +1,135 @@
package assign
import (
"encoding/json"
"fmt"
"sort"
"strings"
"time"
"git.novox.be/novox/mesh-sdk/go/asks"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/health"
)
// A person's upgrade policy is the assigner's (novox/hq ADR 0565 §7, §8, issue 1953 D4, D5): kept in its state `policies`, one
// key per module, set and taken back through `mesh-wants.upgrade`. Every reader takes one derivation, mesh-sdk's
// catalogue.Upgrade: a person's choice over the module's word, its irreplaceable data and the default, the bus never rolled.
// Until the lease write the controller still writes a policy beside each composition's digest; where it differs from the
// assigner's, the assigner says so as its finding `policy-split`, naming mesh-wants.upgrade as the remedy.
// PoliciesBucket is the assigner's state `policies`.
var PoliciesBucket = delivery.AssignerModule + "_" + delivery.AssignerPoliciesState
// DecodePolicies reads the policies state's values by key (the module); a value that cannot be read is left out, named.
func DecodePolicies(values map[string][]byte) (map[string]catalogue.PersonPolicy, []string) {
out := map[string]catalogue.PersonPolicy{}
var bad []string
for key, raw := range values {
var p catalogue.PersonPolicy
if err := json.Unmarshal(raw, &p); err != nil || p.V != catalogue.PersonPolicyVersion || p.Check() != nil || p.Module != key {
bad = append(bad, key)
continue
}
out[key] = p
}
sort.Strings(bad)
return out, bad
}
// PolicyOf is a module's policy and where it comes from, with the person's choice for it when there is one.
func PolicyOf(m catalogue.Manifest, chosen map[string]catalogue.PersonPolicy) (policy, from, why, by string) {
var c *catalogue.PersonPolicy
if p, ok := chosen[m.Module]; ok {
c = &p
}
return catalogue.Upgrade(m, c)
}
// ControllerPolicies is, by module, every policy the controller writes beside a composition's digest of it.
func ControllerPolicies(composed map[string][]byte) map[string][]string {
out := map[string][]string{}
for key, raw := range composed {
_, module, _, isNode, err := delivery.ParseComposedKey(key)
if err != nil || isNode {
continue
}
d, err := delivery.DecodeComposedDigest(raw)
if err != nil || d.Policy == "" {
continue
}
if !contains(out[module], d.Policy) {
out[module] = append(out[module], d.Policy)
sort.Strings(out[module])
}
}
return out
}
func contains(list []string, s string) bool {
for _, v := range list {
if v == s {
return true
}
}
return false
}
// Split is a module whose policy at the assigner differs from one the controller writes beside a composition of it.
type Split struct {
Module string `json:"module"`
Policy string `json:"policy"`
From string `json:"from"`
Controller []string `json:"controller"`
}
// Splits is every module of manifests the controller writes a policy for that differs from the assigner's, sorted.
func Splits(manifests map[string]catalogue.Manifest, chosen map[string]catalogue.PersonPolicy, controller map[string][]string) []Split {
var out []Split
for module, theirs := range controller {
m, ok := manifests[module]
if !ok {
continue
}
policy, from, _, _ := PolicyOf(m, chosen)
for _, p := range theirs {
if p != policy {
out = append(out, Split{Module: module, Policy: policy, From: from, Controller: theirs})
break
}
}
}
sort.Slice(out, func(i, j int) bool { return out[i].Module < out[j].Module })
return out
}
// KindPolicySplit is the assigner's finding of a split policy (ADR 0565 §8).
const KindPolicySplit = "policy-split"
// PolicySplitKey is a module's policy-split finding's key.
func PolicySplitKey(module string) string {
return health.FindingKey(health.ScopeModule, module, KindPolicySplit)
}
// PolicySplitFinding is the finding of one split, in mesh-sdk's layout.
func PolicySplitFinding(s Split, at time.Time) ([]byte, error) {
summary := fmt.Sprintf("%s's upgrade policy is %s at the assigner (%s) and %s at the controller", s.Module, s.Policy, s.From,
strings.Join(s.Controller, " and "))
return health.EncodeFinding(health.Finding{Key: PolicySplitKey(s.Module), Kind: KindPolicySplit,
Subject: health.Subject{Scope: health.ScopeModule, ID: s.Module}, Severity: health.Warning, Summary: summary,
Headline: clipHeadline(s.Module + "'s upgrade policy is split"),
Needs: "set it with mesh-wants.upgrade, so the assigner and the controller agree",
Explanation: asks.NeedsYou + "set it with mesh-wants.upgrade, so the assigner and the controller agree." + " The controller still " +
"writes a policy beside each composition until it hands its lease away; a choice made at its upgrade verb after the copy " +
"is not the assigner's.",
Resolved: s.Module + "'s upgrade policy agrees again", Evidence: []health.Evidence{{At: at.UTC(), Said: summary}},
Source: "assigner." + KindPolicySplit, Observer: delivery.AssignerModule, At: at.UTC()})
}
func clipHeadline(s string) string {
if len([]rune(s)) <= 60 {
return s
}
return string([]rune(s)[:60])
}
@@ -29,7 +29,9 @@ type Module struct {
// 1945, live on 2026-10-25 at 20:04: ace's mesh-build-go was wanted with no account on the bus). Nil: the grant alone.
GrantedOn func(node string) bool
Manifest catalogue.Manifest
Nodes []string
// Chosen is a person's choice of the module's upgrade policy (the assigner's `policies`, ADR 0565 §7); nil: none.
Chosen *catalogue.PersonPolicy
Nodes []string
// Composed answers the controller's composition of the module for a node at a fingerprint (mesh-controller_composed): its
// digest and the judgement of the move to it, and whether it exists. Nil: none exists (ADR 0539 §1, §6).
Composed func(node, fp string) (delivery.ComposedDigest, bool)
@@ -639,9 +641,14 @@ func (m Module) busBar(before delivery.Want, had bool, running string) (prev str
return prev, catalogue.ProvidesBus(m.Manifest) && prev != m.FP
}
// recorded says whether the module is of policy record: as the controller holds it beside the composition's digest, the
// operator's own choice included (ADR 0539 §14); on a composition made by hand, which carries none, as the manifest derives it.
// recorded says whether the module is of policy record: a person's choice at the assigner over everything but the bus (ADR
// 0565 §7, catalogue.Upgrade); for a module with none, as the controller holds it beside the composition's digest until the
// lease write (ADR 0539 §14); on a composition made by hand, which carries none, as the manifest derives it.
func (m Module) recorded(d delivery.ComposedDigest) bool {
if m.Chosen != nil {
policy, _, _, _ := catalogue.Upgrade(m.Manifest, m.Chosen)
return policy == catalogue.PolicyRecord
}
if d.Policy != "" {
return d.Policy == delivery.PolicyRecord
}
@@ -802,6 +802,12 @@ func TestTheRecordPolicyIsTheOneTheControllerHolds(t *testing.T) {
if out := Wants(m, n, wanted, nil); len(out.Record) == 0 {
t.Errorf("a hand-made composition fell through the manifest's record policy: %+v", out)
}
// 1953/D4 (ADR 0565 §7): a person's choice at the assigner is over the controller's beside the composition.
with(delivery.PolicyRecord, "person")
m.Chosen = &catalogue.PersonPolicy{V: catalogue.PersonPolicyVersion, Module: m.Name, Policy: catalogue.PolicyRoll, Why: "w", By: "op"}
if out := Wants(m, n, wanted, nil); len(out.Record) != 0 || out.Wants["ace"].FP != fNew {
t.Errorf("a person's roll at the assigner was not honoured over the controller's record: %+v", out)
}
}
// Review round 1, item 2 (ADR 0539 §15): a node that runs the bus with no readable want compares the new fingerprint with the one
@@ -44,7 +44,7 @@ func TestManifest(t *testing.T) {
states[s.Name] = s.History
}
if want := map[string]int{"asks": 5, "compositions": 5, "differences": 64, "puts": 20, "wants": 20,
"placement": 64, "proposals": 5, "candidates": 5, "pulses": 0, "findings": 5}; !reflect.DeepEqual(states, want) {
"placement": 64, "proposals": 5, "candidates": 5, "pulses": 0, "findings": 5, "policies": 20}; !reflect.DeepEqual(states, want) {
t.Errorf("state = %v, want %v", states, want)
}
if _, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 {
@@ -63,9 +63,14 @@ func TestManifest(t *testing.T) {
// ADR 0567 §1, §6-§10: it holds the seat mesh-placement, serving its seven verbs; keeps the seat's events on its stream
// placement for 30 days; calls the controller's export, its four verbs that keep its assignments in step and its
// hand-act log; and reads mesh-config's resolved values to compose every node a proposal touches.
if len(m.Claims) != 1 || m.Claims[0].Name != catalogue.PlacementSeat || m.Claims[0].At() != catalogue.ScopeMesh ||
!slices.Equal(m.Claims[0].Serves, []string{"placement", "proposals", "propose", "withdraw", "again", "adopt", "hand-back"}) {
t.Errorf("the assigner claims %+v, want mesh-placement with its seven verbs", m.Claims)
// ADR 0565 §7, §9 (issue 1953 D4): and the seat mesh-wants, serving upgrade and bus.
if len(m.Claims) != 2 || m.Claims[0].Name != catalogue.PlacementSeat || m.Claims[0].At() != catalogue.ScopeMesh ||
!slices.Equal(m.Claims[0].Serves, []string{"placement", "proposals", "propose", "withdraw", "again", "adopt", "hand-back"}) ||
m.Claims[1].Name != catalogue.WantsSeat || m.Claims[1].At() != catalogue.ScopeMesh || !slices.Equal(m.Claims[1].Serves, []string{"upgrade", "bus"}) {
t.Errorf("the assigner claims %+v, want mesh-placement with its seven verbs and mesh-wants with its two", m.Claims)
}
if seat, _ := catalogue.SeatNamed(catalogue.WantsSeat); catalogue.CanHold(m, seat) != nil {
t.Errorf("the assigner cannot hold mesh-wants: %v", catalogue.CanHold(m, seat))
}
if seat, _ := catalogue.SeatNamed(catalogue.PlacementSeat); catalogue.CanHold(m, seat) != nil {
t.Errorf("the assigner cannot hold the seat: %v", catalogue.CanHold(m, seat))
@@ -79,7 +84,7 @@ func TestManifest(t *testing.T) {
}
if want := []string{"seat:mesh-controller.assign", "seat:mesh-controller.hand-act", "seat:mesh-controller.pin",
"seat:mesh-controller.placement-export", "seat:mesh-controller.unassign", "seat:mesh-controller.unpin",
"seat:mesh-controller.warranted", "seat:mesh-hand-acts.hand-act", "seat:mesh-hand-acts.warranted", // ADR 0565 §11: the seat, the controller's only while it has no holder
"seat:mesh-controller.warranted", "seat:mesh-hand-acts.hand-act", "seat:mesh-hand-acts.hand-acts", "seat:mesh-hand-acts.warranted", // ADR 0565 §9, §11: the seat, the controller's only while it has no holder
"seat:node-backup.backed-up", "seat:node-backup.now"}; !slices.Equal(slices.Sorted(slices.Values(m.Invokes)), want) {
t.Errorf("invokes = %v, want %v", m.Invokes, want)
}
+11 -1
View File
@@ -15,6 +15,14 @@
"adopt",
"hand-back"
]
},
{
"name": "mesh-wants",
"scope": "mesh",
"serves": [
"upgrade",
"bus"
]
}
],
"reads": [
@@ -48,6 +56,7 @@
"seat:mesh-controller.unpin",
"seat:mesh-controller.warranted",
"seat:mesh-hand-acts.hand-act",
"seat:mesh-hand-acts.hand-acts",
"seat:mesh-hand-acts.warranted",
"seat:node-backup.backed-up",
"seat:node-backup.now"
@@ -70,7 +79,8 @@
{ "name": "differences", "history": 64 },
{ "name": "puts", "history": 20 },
{ "name": "wants", "history": 20 },
{ "name": "findings", "history": 5 }
{ "name": "findings", "history": 5 },
{ "name": "policies", "history": 20 }
],
"streams": [
{