From ca5f74eadba92a36321cf52cc925493f2ac31b1b Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 17:37:03 +0100 Subject: [PATCH] Issue a node's bus accounts and grant when its assignment comes to stand (issue 1945, ADR 0543) ADR 0539 clause 7 issues a module's credential and grant only when its build is done. A module built long before and assigned to one more node never has a build done again: on 2026-10-25 at 17:19 mesh-build-node and mesh-build-go were assigned to ace and composed for it, the bus's user list never named ace's accounts, and both runtimes restarted every 5 s (one with no credential, one refused by the bus). The assign act composes the assigned node alone, never the node holding the bus, and writes no grant. So a write of the assignments bucket is the event: when a key comes to stand (new, or ended or pending before), the lease holder issues the module's accounts on that node that are not minted (never re-issued), composes the node holding the bus and writes the grants, through the composer the done keys' take-ins use. A rewrite of a standing assignment does nothing; the start's look settles every standing one, one after another. A step of research 072's grants (memberships and credentials heard from placement). --- cmd/mesh-controller/assignment_accounts.go | 299 ++++++++++++++++++ .../assignment_accounts_test.go | 251 +++++++++++++++ cmd/mesh-controller/new_modules.go | 3 + internal/inventory/chain_assignments.go | 10 + 4 files changed, 563 insertions(+) create mode 100644 cmd/mesh-controller/assignment_accounts.go create mode 100644 cmd/mesh-controller/assignment_accounts_test.go diff --git a/cmd/mesh-controller/assignment_accounts.go b/cmd/mesh-controller/assignment_accounts.go new file mode 100644 index 00000000..531900f3 --- /dev/null +++ b/cmd/mesh-controller/assignment_accounts.go @@ -0,0 +1,299 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + "sync" + "time" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + + "github.com/novox/mesh-controller/internal/broker" + "github.com/novox/mesh-controller/internal/chain" + "github.com/novox/mesh-controller/internal/inventory" +) + +// An assignment that stands issues its node's bus accounts and the module's grant (novox/hq ADR 0543, issue 1945; a +// step of research 072's grants: the memberships and credentials heard from placement, not from a send or a build). +// +// ADR 0539 clause 7 issues a module's credential and grant when its build is done. A module already built and assigned +// to one more node never has a build done again, so nothing issued that node's account: on 2026-10-25 at 17:19 +// mesh-build-node and mesh-build-go were assigned to ace, composed for it, and the bus's user list never named +// ace.mesh-build-node nor ace.mesh-build-go; one runtime found no credential, the other was refused by the bus, both +// restarting every 5 s. The assign act composes the assigned node alone (composedAfterAct): never the node holding the +// bus, never the grant. +// +// So the write of an assignments key is the event. When a key `.` of the assignments bucket comes to stand +// (a new key, or one that was ended or pending), the lease holder issues the module's accounts on that node that are not +// minted (idempotent: a minted account is never issued again, a rotation is on purpose), composes the node holding the +// bus so its user list carries them, and writes the grants. A rewrite of an assignment that already stood (the +// provisioner's plan kept in its value) does nothing. A starting controller looks at every standing assignment once: +// it issues what is missing and, when it issued anything, composes the bus's node and writes the grants. + +// assignmentAccountSteps are what an assignment that comes to stand does, each a step a test may stand in for. +type assignmentAccountSteps struct { + // holds says this controller holds the lease: only its holder issues. + holds func(ctx context.Context) bool + // issue issues the module's accounts on the node that are not minted; issued says it minted one, line what it did. + issue func(ctx context.Context, node, module string) (issued bool, line string, err error) + // busHolder is the node holding the bus, "" for none. + busHolder func(ctx context.Context) (string, error) + // compose writes the nodes' compositions; grants writes the grants issued. + compose func(ctx context.Context, nodes []string) error + grants func(ctx context.Context) error +} + +// settle issues what the standing assignment of module on node lacks. fresh says the assignment just came to stand: the +// bus's node is composed and the grants written even when every account was minted before (a module assigned again +// keeps its account, and the bus's user list left it out while it was unassigned). It answers what it did, "" for +// nothing. +func (s assignmentAccountSteps) settle(ctx context.Context, node, module string, fresh bool) (string, error) { + if !s.holds(ctx) || module == controllerModule { + // The controller's own module is issued at the terminal alone: its `broker` secret is the credential it is + // using (rollout.go), and it is assigned and unassigned there only (refusedMovingTheController). + return "", nil + } + issued, line, err := s.issue(ctx, node, module) + if err != nil { + return "", fmt.Errorf("its bus accounts on %s are not issued: %w", node, err) + } + if !issued && !fresh { + return "", nil + } + holder, err := s.busHolder(ctx) + if err != nil { + return line, fmt.Errorf("the node holding the bus cannot be read: %w", err) + } + if holder != "" { + if err := s.compose(ctx, []string{holder}); err != nil { + return line, fmt.Errorf("the bus's user list on %s is not composed: %w", holder, err) + } + } + if err := s.grants(ctx); err != nil { + return line, err + } + said := module + " on " + node + ": " + if line != "" { + said += line + "; " + } + if holder != "" { + said += "the bus's user list composed on " + holder + ", " + } + return said + "its grant written", nil +} + +// assignmentStands says a value of the assignments bucket holds an assignment that stands: made (a module only pulled in +// by another names no time assigned), not ended and in no state (chain.Assignment.Stands). +func assignmentStands(a chain.Assignment) bool { + return !a.Assigned.IsZero() && a.Stands() && a.Node != "" && a.Module != "" +} + +// assignmentTurns keeps which keys of the assignments bucket stood when last heard, to tell an assignment that comes to +// stand from a rewrite of one that stood. +type assignmentTurns struct { + mu sync.Mutex + stood map[string]bool +} + +// heard records a value of key (nil for a delete or a purge) and answers whether it holds an assignment that just came +// to stand. +func (t *assignmentTurns) heard(key string, value []byte) (chain.Assignment, bool) { + var a chain.Assignment + stands := false + if value != nil && json.Unmarshal(value, &a) == nil { + stands = assignmentStands(a) + } + t.mu.Lock() + defer t.mu.Unlock() + if t.stood == nil { + t.stood = map[string]bool{} + } + was := t.stood[key] + t.stood[key] = stands + return a, stands && !was +} + +// ownAccountOf is the username of the account a module speaks on the bus with on node, its own and no process's: the +// module's when it reads a `broker` own secret, else its provisioner client's when it runs one; false when it has none +// (a module whose accounts are its processes' alone, or one that does not speak on the bus). +func ownAccountOf(m catalogue.Manifest, node string) (string, bool) { + if _, reads := m.OwnSecrets["broker"]; reads { + return broker.Principal{Kind: broker.KindModule, Node: node, Module: m.Module}.Username(), true + } + if runsTheProvisionerClient(m) { + return broker.Principal{Kind: broker.KindProvisioner, Node: node, Module: m.Module}.Username(), true + } + return "", false +} + +// issueAssignmentAccounts issues module's accounts on node that are not minted: its own (issueOnTheNewBus) and its +// processes' (issueMissingProcessAccounts). A module the catalogue does not hold, or one with nowhere to deliver an +// account (mayIssue), has nothing to issue; an account minted before is kept. +func issueAssignmentAccounts(ctx context.Context, inv *inventory.Inventory, node, module string) (bool, string, error) { + shelf, err := inv.Catalogue(ctx) + if err != nil { + return false, "", fmt.Errorf("the catalogue cannot be read: %w", err) + } + m, known := shelf[module] + if !known || mayIssue(m) != nil { + return false, "", nil + } + if user, own := ownAccountOf(m, node); own { + _, minted, err := inv.BusUserHash(ctx, user) + if err != nil { + return false, "", fmt.Errorf("whether %s is minted cannot be read: %w", user, err) + } + if !minted { + busAddress, err := broker.BusAddress() + if err != nil { + return false, "", err + } + // Its processes' accounts too: issueOnTheNewBus issues every one of them with the module's own. + if err := issueOnTheNewBus(ctx, inv, m, node, busAddress); err != nil { + return false, "", err + } + return true, "its bus credential issued as " + user, nil + } + } + line, err := issueMissingProcessAccounts(ctx, inv, m, node) + if err != nil { + return false, "", err + } + return line != "", line, nil +} + +// liveAssignmentAccountSteps are the steps on the controller's own stores and the bus: the composer's passes, as the +// done keys' take-ins use them, so a node asked for by both is composed once a pass. +func liveAssignmentAccountSteps(open *stores) assignmentAccountSteps { + inv := open.inventory + return assignmentAccountSteps{ + holds: func(ctx context.Context) bool { + epoch, err := theLease.epoch(ctx) + return err == nil && epoch > 0 + }, + issue: func(ctx context.Context, node, module string) (bool, string, error) { + return issueAssignmentAccounts(ctx, inv, node, module) + }, + busHolder: func(ctx context.Context) (string, error) { + holders, err := seatHolders(ctx, inv) + if err != nil { + return "", err + } + return holders[catalogue.BrokerSeat].Node, nil + }, + compose: func(ctx context.Context, nodes []string) error { + return composerFor(open).compose(ctx, nodes, false) + }, + grants: func(ctx context.Context) error { + return composerFor(open).compose(ctx, nil, true) + }, + } +} + +// assignmentAccountsRetry is how long a settle that failed waits before it tries again. +var assignmentAccountsRetry = 30 * time.Second + +// settleAssignment settles one assignment until it works or ctx ends: a failure is said and tried again every +// assignmentAccountsRetry, never dropped, since no later write of the key may come. +func settleAssignment(ctx context.Context, steps assignmentAccountSteps, node, module string, fresh bool) { + for { + said, err := steps.settle(ctx, node, module, fresh) + if err == nil { + if said != "" { + fmt.Printf("assignments: %s\n", said) + } + return + } + if ctx.Err() != nil { + return + } + fmt.Printf("assignments: %s on %s: %v; tried again in %s\n", module, node, err, assignmentAccountsRetry) + select { + case <-ctx.Done(): + return + case <-time.After(assignmentAccountsRetry): + } + } +} + +// watchAssignmentAccounts settles every assignment of the assignments bucket that comes to stand, for as long as ctx: the +// bucket's current values first (the start's look, settled as not fresh), then each write. A watch that ends, or a +// bucket not there yet, is watched again a minute later. +func watchAssignmentAccounts(ctx context.Context, open *stores) { + steps := liveAssignmentAccountSteps(open) + turns := &assignmentTurns{} + goServing(ctx, func() { + for { + var atStart []chain.Assignment + if err := watchAssignmentsOnce(ctx, open.inventory.WatchAssignments, turns, func(a chain.Assignment, fresh bool) { + if !fresh { + atStart = append(atStart, a) + return + } + inBackground(ctx, func() { settleAssignment(ctx, steps, a.Node, a.Module, true) }) + }, func() { + // The start's look, one assignment after another: a reading of the catalogue each, never all at once + // (issues 1945 and 1277: a burst of take-ins once filled the store). + look := atStart + atStart = nil + inBackground(ctx, func() { + for _, a := range look { + if ctx.Err() != nil { + return + } + settleAssignment(ctx, steps, a.Node, a.Module, false) + } + }) + }); err != nil && ctx.Err() == nil { + fmt.Printf("assignments: the assignments are not heard, so a new assignment's bus accounts wait: %v\n", err) + } + select { + case <-ctx.Done(): + return + case <-time.After(time.Minute): + } + } + }) +} + +// watchAssignmentsOnce watches the assignments until the watch or ctx ends, handing each assignment that comes to stand +// to settle: fresh for one written after the current values were read, not fresh for one of them, and calling read once +// the current values are all handed (the chain's Caught mark). +func watchAssignmentsOnce(ctx context.Context, watch func(context.Context) (<-chan chain.Entry, error), + turns *assignmentTurns, settle func(a chain.Assignment, fresh bool), read func()) error { + ctx, stop := context.WithCancel(ctx) + defer stop() + entries, err := watch(ctx) + if err != nil { + return err + } + current := true + for { + select { + case <-ctx.Done(): + return nil + case e, ok := <-entries: + if !ok { + if ctx.Err() != nil { + return nil + } + return fmt.Errorf("the watch ended") + } + if e.Caught { + if current { + current = false // the current values are read: what follows is written from now on + read() + } + continue + } + var value []byte + if !e.Deleted { + value = e.Value + } + if a, turned := turns.heard(e.Key, value); turned { + settle(a, !current) + } + } + } +} diff --git a/cmd/mesh-controller/assignment_accounts_test.go b/cmd/mesh-controller/assignment_accounts_test.go new file mode 100644 index 00000000..80f2cf66 --- /dev/null +++ b/cmd/mesh-controller/assignment_accounts_test.go @@ -0,0 +1,251 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "github.com/novox/mesh-controller/internal/broker" + "github.com/novox/mesh-controller/internal/chain" +) + +// An assignment that comes to stand issues its node's bus accounts and the module's grant (ADR 0543, issue 1945): on +// 2026-10-25 at 17:19 mesh-build-node and mesh-build-go, built long before, were assigned to ace, and nothing issued +// ace's accounts nor put them in the bus's user list. + +func anAssignmentValue(t *testing.T, a chain.Assignment) []byte { + t.Helper() + raw, err := json.Marshal(a) + if err != nil { + t.Fatal(err) + } + return raw +} + +// Only a write that makes the key stand turns it: a rewrite of a standing assignment (the provisioner's plan kept in its +// value), an ended one, a pending one or a module only pulled in by another does not. +func TestAnAssignmentTurnsWhenItComesToStand(t *testing.T) { + at := time.Date(2026, 10, 25, 16, 19, 42, 0, time.UTC) + stands := chain.Assignment{Node: "ace", Module: "mesh-build-node", Assigned: at} + ended := stands + ended.Ended = at.Add(time.Hour) + pending := stands + pending.State = chain.StatePending + pulled := chain.Assignment{Node: "ace", Module: "mesh-build-node", RequiredBy: "mesh-build-go"} + replanned := stands + replanned.Provisioning = json.RawMessage(`{"plan":1}`) + + type write struct { + value *chain.Assignment // nil for a delete + turning bool + } + cases := []struct { + name string + writes []write + }{ + {"a new assignment turns", []write{{&stands, true}}}, + {"a rewrite of one that stands does not", []write{{&stands, true}, {&replanned, false}, {&stands, false}}}, + {"an ended one does not", []write{{&ended, false}}}, + {"assigned again after it ended turns again", []write{{&stands, true}, {&ended, false}, {&stands, true}}}, + {"a pending one does not, and turns once made", []write{{&pending, false}, {&stands, true}}}, + {"a module only pulled in does not", []write{{&pulled, false}}}, + {"one deleted and written again turns again", []write{{&stands, true}, {nil, false}, {&stands, true}}}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + turns := &assignmentTurns{} + for i, w := range c.writes { + var value []byte + if w.value != nil { + value = anAssignmentValue(t, *w.value) + } + a, turned := turns.heard("ace.mesh-build-node", value) + if turned != w.turning { + t.Fatalf("write %d: turned %v, want %v", i, turned, w.turning) + } + if turned && (a.Node != "ace" || a.Module != "mesh-build-node") { + t.Fatalf("write %d: the assignment turned is %s on %s", i, a.Module, a.Node) + } + } + }) + } +} + +// settle issues what is missing and then puts it on the bus: the node holding the bus composed (its user list) and the +// grants written; for an assignment that just came to stand even when its accounts were minted before (a module +// assigned again keeps its account), and never by a controller that does not hold the lease. +func TestSettlingAnAssignmentIssuesComposesTheBusAndGrants(t *testing.T) { + cases := []struct { + name string + module string + holds, fresh bool + issued bool + issueErr, composeE error + holder string + wantComposed []string + wantGrants bool + wantErr string + }{ + {name: "a new assignment whose account is not minted", module: "mesh-build-node", holds: true, fresh: true, + issued: true, holder: "novox", wantComposed: []string{"novox"}, wantGrants: true}, + {name: "a new assignment whose account was minted before", module: "mesh-build-go", holds: true, fresh: true, + holder: "novox", wantComposed: []string{"novox"}, wantGrants: true}, + {name: "the start's look issuing a missing account", module: "mesh-build-node", holds: true, issued: true, + holder: "novox", wantComposed: []string{"novox"}, wantGrants: true}, + {name: "the start's look with every account minted does nothing", module: "mesh-build-go", holds: true, + holder: "novox"}, + {name: "no lease, nothing", module: "mesh-build-node", fresh: true, issued: true, holder: "novox"}, + {name: "the controller's own module is the terminal's", module: controllerModule, holds: true, fresh: true, + issued: true, holder: "novox"}, + {name: "no bus holder: the grants still", module: "mesh-build-node", holds: true, fresh: true, issued: true, + wantGrants: true}, + {name: "an issue that fails is the error, nothing composed", module: "mesh-build-node", holds: true, fresh: true, + issueErr: errors.New("the bus cannot be reached"), holder: "novox", wantErr: "the bus cannot be reached"}, + {name: "a composition that fails is the error, no grant", module: "mesh-build-node", holds: true, fresh: true, + issued: true, holder: "novox", composeE: errors.New("store refused"), wantComposed: []string{"novox"}, + wantErr: "store refused"}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + var composed []string + issuedFor, grants := "", false + steps := assignmentAccountSteps{ + holds: func(context.Context) bool { return c.holds }, + issue: func(_ context.Context, node, module string) (bool, string, error) { + issuedFor = module + "@" + node + if c.issued { + return true, "its bus credential issued", c.issueErr + } + return false, "", c.issueErr + }, + busHolder: func(context.Context) (string, error) { return c.holder, nil }, + compose: func(_ context.Context, nodes []string) error { + composed = append(composed, nodes...) + return c.composeE + }, + grants: func(context.Context) error { grants = true; return nil }, + } + said, err := steps.settle(t.Context(), "ace", c.module, c.fresh) + if c.wantErr != "" { + if err == nil || !strings.Contains(err.Error(), c.wantErr) { + t.Fatalf("error %v, want one saying %q", err, c.wantErr) + } + } else if err != nil { + t.Fatal(err) + } + if strings.Join(composed, ",") != strings.Join(c.wantComposed, ",") { + t.Fatalf("composed %v, want %v", composed, c.wantComposed) + } + if grants != c.wantGrants { + t.Fatalf("grants written %v, want %v", grants, c.wantGrants) + } + skipped := !c.holds || c.module == controllerModule + if skipped != (issuedFor == "") { + t.Fatalf("issue asked for %q (lease held %v, module %s)", issuedFor, c.holds, c.module) + } + if c.wantGrants && c.wantErr == "" && !strings.Contains(said, "its grant written") { + t.Fatalf("said %q, want it to name the grant", said) + } + }) + } +} + +// The account a module speaks with on a node: the module's own when it reads `broker`, its provisioner client's when it +// runs one alone, none when its accounts are its processes' or it does not speak on the bus. +func TestAModulesOwnAccountOnANode(t *testing.T) { + client := aProvisioner() + delete(client.OwnSecrets, "broker") + cases := []struct { + name string + m catalogue.Manifest + user string + own bool + }{ + {"a module reading broker", catalogue.Manifest{Module: "mesh-build-node", + OwnSecrets: catalogue.OwnSecrets{"broker": {Path: "/var/lib/mesh-build-node/broker"}}}, + broker.Principal{Kind: broker.KindModule, Node: "ace", Module: "mesh-build-node"}.Username(), true}, + {"a module reading broker and running the client", aProvisioner(), + broker.Principal{Kind: broker.KindModule, Node: "ace", Module: "mesh-provisioner"}.Username(), true}, + {"the provisioner client alone", client, + broker.Principal{Kind: broker.KindProvisioner, Node: "ace", Module: "mesh-provisioner"}.Username(), true}, + {"no broker, no client", catalogue.Manifest{Module: "hello-web"}, "", false}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + user, own := ownAccountOf(c.m, "ace") + if user != c.user || own != c.own { + t.Fatalf("got %q %v, want %q %v", user, own, c.user, c.own) + } + }) + } +} + +// The watch hands the assignments standing when it starts to the start's look (not fresh), says once that they are all +// read, and hands each assignment written afterwards as fresh; a rewrite of one that stands is handed nothing. +func TestTheWatchHandsTheStartsLookThenEachNewAssignment(t *testing.T) { + at := time.Date(2026, 10, 25, 16, 19, 42, 0, time.UTC) + bucket := chain.NewMemory() + put := func(a chain.Assignment) { + t.Helper() + value := anAssignmentValue(t, a) + if e, held, _ := bucket.Get(t.Context(), a.Key()); held { + if _, err := bucket.Update(t.Context(), a.Key(), value, e.Revision); err != nil { + t.Fatal(err) + } + return + } + if _, err := bucket.Create(t.Context(), a.Key(), value); err != nil { + t.Fatal(err) + } + } + put(chain.Assignment{Node: "g14", Module: "mesh-build-go", Assigned: at}) + + type handed struct { + key string + fresh bool + } + got := make(chan handed, 8) + read := make(chan struct{}, 8) + ctx, stop := context.WithCancel(t.Context()) + done := make(chan error, 1) + go func() { + done <- watchAssignmentsOnce(ctx, bucket.Watch, &assignmentTurns{}, + func(a chain.Assignment, fresh bool) { got <- handed{a.Key(), fresh} }, + func() { read <- struct{}{} }) + }() + next := func() handed { + t.Helper() + select { + case h := <-got: + return h + case <-time.After(5 * time.Second): + t.Fatal("nothing handed") + return handed{} + } + } + if h := next(); h != (handed{"g14.mesh-build-go", false}) { + t.Fatalf("the start's look handed %+v", h) + } + select { + case <-read: + case <-time.After(5 * time.Second): + t.Fatal("the current values were never said read") + } + replanned := chain.Assignment{Node: "g14", Module: "mesh-build-go", Assigned: at, Provisioning: json.RawMessage(`{"plan":2}`)} + put(replanned) + put(chain.Assignment{Node: "ace", Module: "mesh-build-node", Assigned: at}) + if h := next(); h != (handed{"ace.mesh-build-node", true}) { + t.Fatalf("handed %+v, want ace.mesh-build-node as fresh (and nothing for the rewrite before it)", h) + } + stop() + if err := <-done; err != nil { + t.Fatal(err) + } + if len(read) != 0 { + t.Fatal("the current values were said read more than once") + } +} diff --git a/cmd/mesh-controller/new_modules.go b/cmd/mesh-controller/new_modules.go index 098d9404..7b014605 100644 --- a/cmd/mesh-controller/new_modules.go +++ b/cmd/mesh-controller/new_modules.go @@ -208,6 +208,9 @@ func watchNewModules(ctx context.Context, open *stores) { // composed, the replaced controller composed, and from now on each new fingerprint registered, composed and granted. inBackground(ctx, func() { composeAtStart(ctx, open) }) watchDoneKeys(ctx, open) + // And each assignment that comes to stand issues its node's bus accounts and the module's grant (ADR 0543): a module + // built long before is assigned to one more node with no build done again (issue 1945, assignment_accounts.go). + watchAssignmentAccounts(ctx, open) watchHavesForJudgement(ctx, open) // And the keep set keeps the builds every want names (ADR 0539 clause 1), with no plan left to name them. open.inventory.SetWanted(wantedOnTheBus) diff --git a/internal/inventory/chain_assignments.go b/internal/inventory/chain_assignments.go index 1d56aae6..652df44a 100644 --- a/internal/inventory/chain_assignments.go +++ b/internal/inventory/chain_assignments.go @@ -361,3 +361,13 @@ func (i *Inventory) AssignmentValue(ctx context.Context, node, module string) (c k, found, err := c.Assignments.Get(ctx, assignmentKey(node, module)) return k.Value, found, err } + +// WatchAssignments is the assignments bucket's raw entries, as the chain's watch gives them: every current value, a Caught +// mark, then each write (novox/hq ADR 0543: an assignment that comes to stand issues its node's bus accounts). +func (i *Inventory) WatchAssignments(ctx context.Context) (<-chan chain.Entry, error) { + c, err := i.theChain(ctx) + if err != nil { + return nil, err + } + return c.Assignments.Watch(ctx) +}