From 3aec7f9b02c942192fbf4dd29ab3ff5e8032cfb7 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 29 Oct 2026 22:34:08 +0100 Subject: [PATCH] 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) --- .../cmd/mesh-delivery-assigner/main.go | 30 +- .../cmd/mesh-delivery-assigner/main_test.go | 4 +- .../cmd/mesh-delivery-assigner/pass.go | 64 +++- .../cmd/mesh-delivery-assigner/puts.go | 5 +- .../mesh-delivery-assigner/seat_tools_test.go | 114 +++++++ .../cmd/mesh-delivery-assigner/tools.go | 7 +- .../cmd/mesh-delivery-assigner/wants_tools.go | 300 ++++++++++++++++++ .../internal/assign/findings.go | 1 + .../internal/assign/policy.go | 135 ++++++++ .../internal/assign/wants.go | 13 +- .../internal/assign/wants_test.go | 6 + .../mesh-delivery-assigner/manifest_test.go | 15 +- modules/mesh-delivery-assigner/module.json | 12 +- 13 files changed, 686 insertions(+), 20 deletions(-) create mode 100644 modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/wants_tools.go create mode 100644 modules/mesh-delivery-assigner/internal/assign/policy.go diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go index 23a76f4..934df8f 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -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) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go index f7a91d4..d2a1a83 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go @@ -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" { diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go index 24d22c5..facdff2 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -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 +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go index 26be2fa..2def647 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go @@ -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 diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go index 89c25dd..cba607c 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go @@ -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. diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go index 0e47bb4..74a5180 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go @@ -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. diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/wants_tools.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/wants_tools.go new file mode 100644 index 0000000..945a959 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/wants_tools.go @@ -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 ., 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 +} diff --git a/modules/mesh-delivery-assigner/internal/assign/findings.go b/modules/mesh-delivery-assigner/internal/assign/findings.go index cd0baaf..d6e2ea9 100644 --- a/modules/mesh-delivery-assigner/internal/assign/findings.go +++ b/modules/mesh-delivery-assigner/internal/assign/findings.go @@ -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 } diff --git a/modules/mesh-delivery-assigner/internal/assign/policy.go b/modules/mesh-delivery-assigner/internal/assign/policy.go new file mode 100644 index 0000000..287f8df --- /dev/null +++ b/modules/mesh-delivery-assigner/internal/assign/policy.go @@ -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]) +} diff --git a/modules/mesh-delivery-assigner/internal/assign/wants.go b/modules/mesh-delivery-assigner/internal/assign/wants.go index 6d54820..d839b93 100644 --- a/modules/mesh-delivery-assigner/internal/assign/wants.go +++ b/modules/mesh-delivery-assigner/internal/assign/wants.go @@ -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 } diff --git a/modules/mesh-delivery-assigner/internal/assign/wants_test.go b/modules/mesh-delivery-assigner/internal/assign/wants_test.go index 8d75165..3a6bb0c 100644 --- a/modules/mesh-delivery-assigner/internal/assign/wants_test.go +++ b/modules/mesh-delivery-assigner/internal/assign/wants_test.go @@ -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 diff --git a/modules/mesh-delivery-assigner/manifest_test.go b/modules/mesh-delivery-assigner/manifest_test.go index 3964db8..6f0c3ca 100644 --- a/modules/mesh-delivery-assigner/manifest_test.go +++ b/modules/mesh-delivery-assigner/manifest_test.go @@ -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) } diff --git a/modules/mesh-delivery-assigner/module.json b/modules/mesh-delivery-assigner/module.json index 267eb62..74542b6 100644 --- a/modules/mesh-delivery-assigner/module.json +++ b/modules/mesh-delivery-assigner/module.json @@ -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": [ {