diff --git a/cmd/mesh-controller/acts.go b/cmd/mesh-controller/acts.go index 3215b1aa..8de675b5 100644 --- a/cmd/mesh-controller/acts.go +++ b/cmd/mesh-controller/acts.go @@ -399,6 +399,8 @@ func blockedElsewhere(ctx context.Context, open *stores, except string) string { // purpose, never by re-assigning); and when the bus cannot be reached from here, the line names the // verb and the push that would refuse the module until it is run — never a silent placeholder (novox/hq issue 203). func issueOnAssign(ctx context.Context, open *stores, node, module string) string { + // Mint only what no row names: the assignments' watch hears the same write and may issue at once (ADR 0543). + ctx = onlyIfAbsent(ctx) inv := open.inventory shelf, err := inv.Catalogue(ctx) if err != nil { diff --git a/cmd/mesh-controller/assignment_accounts.go b/cmd/mesh-controller/assignment_accounts.go new file mode 100644 index 00000000..8c6ba982 --- /dev/null +++ b/cmd/mesh-controller/assignment_accounts.go @@ -0,0 +1,387 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + "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, +// from one reading of the catalogue: it issues what is missing, then composes the bus's node and writes the grants +// once. The assign act issues too; both mint only an account no row names, in one statement, so exactly one of them +// mints and seals it (mintBusUser, the review of issue 1945). + +// 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 + // catalogue reads the catalogue once, for every assignment a look issues for. + catalogue func(ctx context.Context) (map[string]catalogue.Manifest, error) + // issue issues the module's accounts on the node that are not minted, from the catalogue read; issued says it + // minted one, line what it did. + issue func(ctx context.Context, shelf map[string]catalogue.Manifest, 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) (string, error) { + if 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 + } + if !s.holds(ctx) { + return "", errNotHolding + } + shelf, err := s.catalogue(ctx) + if err != nil { + return "", fmt.Errorf("the catalogue cannot be read: %w", err) + } + _, line, err := s.issue(ctx, shelf, node, module) + if err != nil { + return "", fmt.Errorf("its bus accounts on %s are not issued: %w", node, err) + } + said, err := s.cover(ctx) + prefix := module + " on " + node + ": " + if line != "" { + prefix += line + "; " + } + return prefix + said, err +} + +// look is the start's look: the accounts every standing assignment lacks, issued one after another from one reading of +// the catalogue, then the bus's node composed and the grants written once, when it looked at any assignment, whatever +// it issued (an assignment that came to stand while no controller heard it may have its accounts and still lack its +// place in the bus's user list or its grant). It answers what it did. +func (s assignmentAccountSteps) look(ctx context.Context, standing []chain.Assignment) (string, error) { + var looked []chain.Assignment + for _, a := range standing { + if a.Module != controllerModule { + looked = append(looked, a) + } + } + if len(looked) == 0 { + return "", nil + } + if !s.holds(ctx) { + return "", errNotHolding + } + shelf, err := s.catalogue(ctx) + if err != nil { + return "", fmt.Errorf("the catalogue cannot be read: %w", err) + } + var lines []string + for _, a := range looked { + if ctx.Err() != nil { + return "", ctx.Err() + } + if _, line, err := s.issue(ctx, shelf, a.Node, a.Module); err != nil { + return "", fmt.Errorf("%s's bus accounts on %s are not issued: %w", a.Module, a.Node, err) + } else if line != "" { + lines = append(lines, a.Module+" on "+a.Node+": "+line) + } + } + said, err := s.cover(ctx) + lines = append(lines, fmt.Sprintf("%d standing assignment(s) looked at; %s", len(looked), said)) + return strings.Join(lines, "\n"), err +} + +// errNotHolding is a controller that does not hold the lease: it issues nothing, and tries again until it holds it. +var errNotHolding = errors.New("this controller does not hold the lease") + +// cover composes the node holding the bus and writes the grants, and says so. +func (s assignmentAccountSteps) cover(ctx context.Context) (string, error) { + holder, err := s.busHolder(ctx) + if err != nil { + return "", fmt.Errorf("the node holding the bus cannot be read: %w", err) + } + if holder != "" { + if err := s.compose(ctx, []string{holder}); err != nil { + return "", fmt.Errorf("the bus's user list on %s is not composed: %w", holder, err) + } + } + if err := s.grants(ctx); err != nil { + return "", err + } + if holder != "" { + return "the bus's user list composed on " + holder + ", the grants written", nil + } + return "the grants 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, shelf map[string]catalogue.Manifest, + node, module string) (bool, string, error) { + // Mint only what no row names: the assign act hears its own write and may issue at once (the review of issue 1945). + ctx = onlyIfAbsent(ctx) + 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 { + // The lease's gate, as every act's: a serving controller the bus does not grant the lease yet acts unleased. + _, err := theLease.epoch(ctx) + return err == nil + }, + catalogue: func(ctx context.Context) (map[string]catalogue.Manifest, error) { return inv.Catalogue(ctx) }, + issue: func(ctx context.Context, shelf map[string]catalogue.Manifest, node, module string) (bool, string, error) { + return issueAssignmentAccounts(ctx, inv, shelf, 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 + +// untilSettled runs settle until it works or ctx ends: a failure, or a controller not holding the lease, is said and +// tried again every assignmentAccountsRetry, never dropped, since no later write of the key may come. +func untilSettled(ctx context.Context, what string, settle func(context.Context) (string, error)) { + for { + said, err := settle(ctx) + if err == nil { + if said != "" { + for _, line := range strings.Split(said, "\n") { + fmt.Printf("assignments: %s\n", line) + } + } + return + } + if ctx.Err() != nil { + return + } + fmt.Printf("assignments: %s: %v; tried again in %s\n", what, 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) + goServing(ctx, func() { + runAssignmentWatches(ctx, open.inventory.WatchAssignments, func(a chain.Assignment) { + inBackground(ctx, func() { + untilSettled(ctx, a.Module+" on "+a.Node, func(ctx context.Context) (string, error) { + return steps.settle(ctx, a.Node, a.Module) + }) + }) + }, func(look []chain.Assignment) { + // The start's look, once in the process: one reading of the catalogue, one assignment after another, + // then one composition of the bus's node and the grants (issues 1945 and 1277). + inBackground(ctx, func() { + untilSettled(ctx, "the start's look", func(ctx context.Context) (string, error) { + return steps.look(ctx, look) + }) + }) + }, func() bool { + select { + case <-ctx.Done(): + return false + case <-time.After(time.Minute): + return true + } + }) + }) +} + +// runAssignmentWatches watches the assignments again each time a watch ends, until ctx ends or pause says stop: each +// assignment that comes to stand is handed to fresh, and the standing ones of the first watch that reads them all (its +// Caught mark) to look, once. A watch that ends, or fails to open, before the start's look ran is no start's look: the +// next one is, with what it hears from nothing (a fresh assignmentTurns), so no key it handed is dropped and no burst of +// fresh settles replaces the look (the review of issue 1945). +func runAssignmentWatches(ctx context.Context, watch func(context.Context) (<-chan chain.Entry, error), + fresh func(chain.Assignment), look func([]chain.Assignment), pause func() bool) { + turns := &assignmentTurns{} + first := true + for { + var atStart []chain.Assignment + if err := watchAssignmentsOnce(ctx, watch, turns, first, func(a chain.Assignment, isFresh bool) { + if !isFresh { + atStart = append(atStart, a) + return + } + fresh(a) + }, func() { + first = false + look(atStart) + atStart = nil + }); err != nil && ctx.Err() == nil { + fmt.Printf("assignments: the assignments are not heard, so a new assignment's bus accounts wait: %v\n", err) + } + if first { + turns = &assignmentTurns{} // the start's look has not run: the next watch hears every standing key anew + } + if ctx.Err() != nil || !pause() { + return + } + } +} + +// watchAssignmentsOnce watches the assignments until the watch or ctx ends, handing each assignment that comes to stand +// to settle. On the process's first watch, the current values are the start's look (not fresh) and read is called once +// they are all handed (the chain's Caught mark); every other value, and every value of a watch opened again after one +// ended, is fresh: an assignment that came to stand while no watch heard it is settled whole, never as the start's look +// (the review of issue 1945). +func watchAssignmentsOnce(ctx context.Context, watch func(context.Context) (<-chan chain.Entry, error), + turns *assignmentTurns, first bool, 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 + if first { + read() + } + } + continue + } + var value []byte + if !e.Deleted { + value = e.Value + } + if a, turned := turns.heard(e.Key, value); turned { + settle(a, !current || !first) + } + } + } +} diff --git a/cmd/mesh-controller/assignment_accounts_test.go b/cmd/mesh-controller/assignment_accounts_test.go new file mode 100644 index 00000000..277e8c8d --- /dev/null +++ b/cmd/mesh-controller/assignment_accounts_test.go @@ -0,0 +1,344 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "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. The process's first 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 watch opened again after one ended hands every standing assignment it +// has not seen stand as fresh and never calls the start's look (the review of issue 1945). +func TestAnAssignmentThatComesToStand(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 + } + for _, c := range []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}}}, + } { + 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) + } + } + }) + } + + type handed struct { + key string + fresh bool + } + for _, first := range []bool{true, false} { + t.Run(fmt.Sprintf("the watch, first %v", first), func(t *testing.T) { + 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}) + 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{}, first, + 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{} + } + } + // Standing when the watch opened: the start's look on the first watch, fresh on any other. + if h := next(); h != (handed{"g14.mesh-build-go", !first}) { + t.Fatalf("the standing assignment was handed %+v", h) + } + put(chain.Assignment{Node: "g14", Module: "mesh-build-go", Assigned: at, Provisioning: json.RawMessage(`{"plan":2}`)}) + 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 want := map[bool]int{true: 1, false: 0}[first]; len(read) != want { + t.Fatalf("the start's look was called %d time(s), want %d", len(read), want) + } + }) + } + // A first watch that fails to open, or ends before its Caught mark, is no start's look: the next watch is, and hands + // every standing assignment to it once, none as fresh (the review of issue 1945: a burst of fresh settles, or keys + // handed and dropped). + for _, firstWatch := range []string{"fails to open", "ends before its Caught mark"} { + t.Run("a first watch that "+firstWatch, func(t *testing.T) { + bucket := chain.NewMemory() + standing := chain.Assignment{Node: "g14", Module: "mesh-build-go", Assigned: at} + if _, err := bucket.Create(t.Context(), standing.Key(), anAssignmentValue(t, standing)); err != nil { + t.Fatal(err) + } + watches := 0 + watch := func(ctx context.Context) (<-chan chain.Entry, error) { + watches++ + if watches > 1 { + return bucket.Watch(ctx) + } + if firstWatch == "fails to open" { + return nil, errors.New("the chain is not open yet") + } + cut := make(chan chain.Entry, 1) + cut <- chain.Entry{Key: standing.Key(), Value: anAssignmentValue(t, standing)} + close(cut) + return cut, nil + } + var looks [][]chain.Assignment + var fresh []chain.Assignment + ctx, stop := context.WithCancel(t.Context()) + defer stop() + runAssignmentWatches(ctx, watch, func(a chain.Assignment) { fresh = append(fresh, a) }, + func(look []chain.Assignment) { looks = append(looks, look); stop() }, + func() bool { return ctx.Err() == nil }) + if len(looks) != 1 || len(looks[0]) != 1 || looks[0][0].Key() != standing.Key() || len(fresh) != 0 { + t.Fatalf("after %d watch(es): looks %v, fresh %v; want one look at %s and nothing fresh", + watches, looks, fresh, standing.Key()) + } + }) + } +} + +// 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, even when every account was minted before (a module assigned again keeps its account). The start's +// look reads the catalogue once, issues for each standing assignment, and composes the bus's node and writes the grants +// once when it looked at any. Neither acts for a controller that does not hold the lease, nor for the controller's own +// module. +func TestSettlingAndTheStartsLook(t *testing.T) { + ace := func(m string) chain.Assignment { return chain.Assignment{Node: "ace", Module: m} } + cases := []struct { + name string + look []chain.Assignment // nil: settle mesh-build-node on ace + module string + notHolding bool + issued bool + issueErr, composeE error + holder string + wantIssues int + wantComposed []string + wantGrants bool + wantErr string + }{ + {name: "an account not minted", module: "mesh-build-node", issued: true, holder: "novox", wantIssues: 1, + wantComposed: []string{"novox"}, wantGrants: true}, + {name: "an account minted before: the bus and the grants still", module: "mesh-build-go", holder: "novox", + wantIssues: 1, wantComposed: []string{"novox"}, wantGrants: true}, + {name: "no lease: nothing, and tried again", module: "mesh-build-node", notHolding: true, issued: true, + holder: "novox", wantErr: errNotHolding.Error()}, + {name: "the controller's own module is the terminal's", module: controllerModule, issued: true, holder: "novox"}, + {name: "no bus holder: the grants still", module: "mesh-build-node", issued: true, wantIssues: 1, wantGrants: true}, + {name: "an issue that fails is the error, nothing composed", module: "mesh-build-node", + issueErr: errors.New("the bus cannot be reached"), holder: "novox", wantIssues: 1, wantErr: "the bus cannot be reached"}, + {name: "a composition that fails is the error, no grant", module: "mesh-build-node", issued: true, holder: "novox", + composeE: errors.New("store refused"), wantIssues: 1, wantComposed: []string{"novox"}, wantErr: "store refused"}, + {name: "the start's look: each issued, the bus and the grants once", + look: []chain.Assignment{ace("mesh-build-node"), ace(controllerModule), ace("mesh-build-go")}, issued: true, + holder: "novox", wantIssues: 2, wantComposed: []string{"novox"}, wantGrants: true}, + {name: "the start's look with every account minted still covers once", + look: []chain.Assignment{ace("mesh-build-node"), ace("mesh-build-go")}, holder: "novox", wantIssues: 2, + wantComposed: []string{"novox"}, wantGrants: true}, + {name: "the start's look at nothing does nothing", look: []chain.Assignment{ace(controllerModule)}, holder: "novox"}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + var composed []string + issues, catalogues, grants := 0, 0, 0 + steps := assignmentAccountSteps{ + holds: func(context.Context) bool { return !c.notHolding }, + catalogue: func(context.Context) (map[string]catalogue.Manifest, error) { + catalogues++ + return map[string]catalogue.Manifest{}, nil + }, + issue: func(context.Context, map[string]catalogue.Manifest, string, string) (bool, string, error) { + issues++ + 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++; return nil }, + } + var said string + var err error + if c.look != nil { + said, err = steps.look(t.Context(), c.look) + } else { + said, err = steps.settle(t.Context(), "ace", c.module) + } + 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 issues != c.wantIssues || catalogues > 1 { + t.Fatalf("%d issue(s) from %d reading(s) of the catalogue, want %d from at most one", issues, catalogues, c.wantIssues) + } + if strings.Join(composed, ",") != strings.Join(c.wantComposed, ",") { + t.Fatalf("composed %v, want %v", composed, c.wantComposed) + } + if want := map[bool]int{true: 1}[c.wantGrants]; grants != want { + t.Fatalf("grants written %d time(s), want %d", grants, want) + } + if c.wantGrants && c.wantErr == "" && !strings.Contains(said, "the grants written") { + t.Fatalf("said %q, want it to name the grants", said) + } + }) + } +} + +// The accounts 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. A process's account is +// held in the provisioner client's form, named for the process and carrying no claims, and may be issued only when +// the process declares its own secret broker- (ADR 0436 §2; issue 1181). +func TestTheAccountsAModuleSpeaksWithOnANode(t *testing.T) { + client := aProvisioner() + delete(client.OwnSecrets, "broker") + for _, c := range []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}, + } { + 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) + } + }) + } + + t.Run("a process's account", func(t *testing.T) { + user := broker.Principal{Kind: broker.KindProcess, Node: "novox", Module: "mesh-mail", Process: "send"}.Username() + if user != "novox.mesh-mail.send" { + t.Fatalf("the process's username: %q", user) + } + held, err := heldProcessAccount(broker.Broker{Fingerprint: "abc"}, "novox.example:4222", "novox", "mesh-mail", + user, "a-password") + if err != nil { + t.Fatal(err) + } + var got map[string]any + if err := json.Unmarshal(held, &got); err != nil { + t.Fatal(err) + } + want := map[string]any{"url": "nats://novox.example:4222", "fingerprint": "abc", "node": "novox", + "module": "mesh-mail", "user": "novox.mesh-mail.send", "password": "a-password"} + for k, v := range want { + if got[k] != v { + t.Errorf("%s: %v, want %v", k, got[k], v) + } + } + if len(got) != len(want) { + t.Errorf("the held account carries %v; want exactly %v (never claims — a process holds no seat)", got, want) + } + }) + + t.Run("a process may be issued only with its own broker secret", func(t *testing.T) { + // mesh-mail's real shape: no `broker` of its own, its process declaring bus (review round 1 of issue 1181). + if err := mayIssue(mailSenderWithProcess()); err != nil { + t.Fatalf("a module with no broker of its own, whose process declares bus, was refused: %v", err) + } + m := mailSenderWithProcess() + delete(m.OwnSecrets, "broker-send") + err := mayIssue(m) + if err == nil { + t.Fatal("a process with no broker- own secret was issued an account") + } + for _, want := range []string{"mesh-mail", "send", "broker-send", "own-secrets"} { + if !strings.Contains(err.Error(), want) { + t.Errorf("the refusal does not say %q: %v", want, err) + } + } + }) +} diff --git a/cmd/mesh-controller/build.go b/cmd/mesh-controller/build.go index f4c78b0d..310c9d91 100644 --- a/cmd/mesh-controller/build.go +++ b/cmd/mesh-controller/build.go @@ -273,4 +273,7 @@ type answers struct { // applied, is not well: an assignment somebody made is not made yet, or will not be. pending []inventory.PendingAssignment pendingUnread string + // takeIn is the take-in's backlog of done keys on the bus (novox/hq ADR 0544 §6), nil where it cannot be read + // (a process with no bus, or a consumer not made yet). + takeIn *takeInBacklog } diff --git a/cmd/mesh-controller/composed_integration_test.go b/cmd/mesh-controller/composed_integration_test.go index 5c87f775..28d7cc41 100644 --- a/cmd/mesh-controller/composed_integration_test.go +++ b/cmd/mesh-controller/composed_integration_test.go @@ -102,8 +102,10 @@ func TestTheDeclarationsStreamGoesOnceEveryNodeRunsTheNewEngine(t *testing.T) { // The new node-engine registered at its fingerprint, from its done key. n6 := fpOf('6') keys, trunk := aKeyFor("mesh-host", "6", n6, "novox/mesh-host", "", "f00d0123456789ab") - if _, err := liveDoneKeySteps(m.open, trunkKeys{keyBuckets: keys, trunk: trunk}).take(ctx, "mesh-host"); err != nil { - t.Fatal(err) + settled, _ := liveTakeInSteps(m.open, trunkKeys{keyBuckets: keys, trunk: trunk}, m.js).batch(ctx, + []delivered{{module: "mesh-host", again: true}}, map[string]string{}) + if len(settled) != 1 || settled[0].err != nil || settled[0].verdict != acknowledged { + t.Fatalf("the take-in of mesh-host: %+v", settled) } if _, err := m.js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: sdkdelivery.HaveBucket, History: sdkdelivery.HaveHistory}); err != nil { t.Fatal(err) diff --git a/cmd/mesh-controller/composed_keys.go b/cmd/mesh-controller/composed_keys.go index 2b2ac625..d83ec114 100644 --- a/cmd/mesh-controller/composed_keys.go +++ b/cmd/mesh-controller/composed_keys.go @@ -4,6 +4,8 @@ import ( "context" "errors" "fmt" + "maps" + "slices" "sort" "strings" "sync" @@ -12,212 +14,285 @@ import ( "git.novox.be/novox/mesh-sdk/go/catalogue" sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" "github.com/nats-io/nats.go/jetstream" + "github.com/novox/mesh-controller/internal/broker" "github.com/novox/mesh-controller/internal/builderkeys" "github.com/novox/mesh-controller/internal/conditions" "github.com/novox/mesh-controller/internal/inventory" ) // A new fingerprint of a registered module, from its done key to its compositions (novox/hq ADR 0539 clauses 5 and 7, -// ADR 0501 §2). With no delivery plan left to take a module's done key in, a write of the builder's or the indexer's -// keys does it: the controller takes in the done key of the fingerprint the indexer's trunk value names (registers the -// module at it, as a plan's take-in did), composes every node the module is assigned to (which makes what a module's -// new manifest needs, a credential among it, as a send made it), composes the node holding the bus (its user list now -// carries the module's permissions at the new fingerprint: a new revision of the bus module's composition alone, -// written into its want at once by the assigner), and writes `grants..`. The assigner's sixth and seventh -// rules then let a want name the fingerprint. **No timer**: woken by the keys, and a starting controller looks at every -// module once (doneKeysAtStart). +// ADR 0501 §2, ADR 0544). With no delivery plan left to take a module's done key in, a write of the builder's or the +// indexer's keys does it: the controller takes in the done key of the fingerprint the indexer's trunk value names +// (registers the module at it, as a plan's take-in did), composes every node the module is assigned to (which makes what +// a module's new manifest needs, a credential among it), composes the node holding the bus (its user list now carries +// the module's permissions at the new fingerprint), and writes `grants..`. The assigner's sixth and seventh +// rules then let a want name the fingerprint. +// +// **A durable pull consumer, not a watch** (ADR 0544; research 072, milestones 2 and 4). The writes are read from two +// durable consumers (take_in_consumer.go): the bus keeps what is not taken in yet, so a restart takes in only what the +// controller before it left unacknowledged, and the start's look is the consumers' own catch-up. A batch of keys asks the +// composer twice whatever its size, and a key with nothing new composes nothing. -// doneKeySteps are what taking a module's done key in does, each a step a test may stand in for. -type doneKeySteps struct { - // held says the catalogue holds the module (a new one is the new-module take-in's, new_modules.go). - held func(ctx context.Context, module string) (bool, error) +// takeInSteps are what taking a batch of done keys in does, each a step a test may stand in for. +type takeInSteps struct { + // catalogue is every module the catalogue holds, with the nodes it is assigned to: read once a batch. A module it + // does not hold is the new-module take-in's (new_modules.go). + catalogue func(ctx context.Context) (map[string][]string, error) // trunk is the indexer's trunk value of the module. trunk func(ctx context.Context, module string) (sdkdelivery.Module, error) - // registeredAt says the module is registered at the commit. - registeredAt func(ctx context.Context, module, commit string) (bool, error) - // takeIn registers the module from the builder's done key of the trunk's fingerprint (takeInKey); errKeyWaits when - // the key is not done yet. + // registered says whether the module is registered at the trunk's commit, and whether at its fingerprint: a commit + // that moves while the fingerprint stays (a catalogue merge that leaves the module's sources alone) is no new build. + registered func(ctx context.Context, module string, trunk sdkdelivery.Module) (atCommit, atFP bool, err error) + // recordCommit records the module's registration at the trunk's commit, its fingerprint unchanged: nothing composed. + recordCommit func(ctx context.Context, module, commit string) error + // takeIn registers the module from the builder's done key of the trunk's fingerprint (takeInKey). takeIn func(ctx context.Context, module string, trunk sdkdelivery.Module) error - // nodesOf are the nodes the module is assigned to; busHolder the node holding the bus, "" for none. - nodesOf func(ctx context.Context, module string) ([]string, error) + // taken says the module's compositions at fp exist for every node and its grant is written: nothing to compose. + taken func(ctx context.Context, module, fp string, nodes []string) (bool, 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 + // compose composes the nodes, and writes the grants after when grants: each node's error, and the grants'. + compose func(ctx context.Context, nodes []string, grants bool) (map[string]error, error, error) } -// take takes module's done key in and composes what it changed. It answers what it did in a line, "" when there was -// nothing to do. -func (s doneKeySteps) take(ctx context.Context, module string) (string, error) { - held, err := s.held(ctx, module) - if err != nil || !held { - return "", err - } - trunk, err := s.trunk(ctx, module) - if errors.Is(err, builderkeys.ErrNoValue) { - return "", nil - } - if err != nil { - return "", err - } - if !strings.HasPrefix(trunk.FP2, sdkdelivery.SourcePrefix2) { - return "", nil - } - at, err := s.registeredAt(ctx, module, trunk.Commit) - if err != nil { - return "", err - } - said := "" - if !at { - if err := s.takeIn(ctx, module, trunk); err != nil { - return "", err - } - said = fmt.Sprintf("%s registered from the builder's done key %s (%s)", module, short(trunk.FP2), short(trunk.Commit)) - } - nodes, err := s.nodesOf(ctx, module) - if err != nil { - return said, err - } - if len(nodes) == 0 { - return said, nil - } - // The module's nodes first: composing them makes what the new manifest needs, its credentials among it. - if err := s.compose(ctx, nodes); err != nil { - return said, err - } - // Then the bus's user list, which carries the module's permissions at its new fingerprint. - holder, err := s.busHolder(ctx) - if err != nil { - return said, err - } - if holder != "" && !inList(nodes, holder) { - if err := s.compose(ctx, []string{holder}); err != nil { - return said, err - } - } - // Then the grant, which with the composition lets a want name the fingerprint (rules 6 and 7). - if err := s.grants(ctx); err != nil { - return said, err - } - if said == "" { - said = fmt.Sprintf("%s at %s", module, short(trunk.FP2)) - } - return said + "; composed for " + strings.Join(nodes, ", ") + ", its grant written", nil +// delivered is one key a consumer delivered: its module, and whether it was delivered before (a take-in that may have +// stopped half-way, so it is taken in whole). +type delivered struct { + module string + again bool + // deliveries is how often the bus delivered it, this time included: what a refusal's delay grows with. + deliveries uint64 } -func inList(list []string, s string) bool { - for _, v := range list { - if v == s { - return true - } - } - return false -} +// verdict is what becomes of a module's delivered keys. +type verdict int -// doneKeyTakeOnce is one take-in; a variable so a test may stand in for it. doneKeysSay says a line of the take-ins. -var ( - doneKeyTakeOnce = takeDoneKeyOnce - doneKeysSay = func(s string) { fmt.Println(s) } +const ( + // acknowledged: taken in, nothing new, a wait a later write ends, or a build key that failed. + acknowledged verdict = iota + // refused: delivered again by the bus after doneKeyRetry (a wait on a read, or a failure). + refused ) -// doneKeysBeforeLetGo is called just before a follower lets go of its module; a variable so a test can wake the module -// at that moment. Nil does nothing. -var doneKeysBeforeLetGo func(module string) - -// doneKeysLookAgain marks a module woken while a take-in of it ran: that take-in goes round once more. -var doneKeysLookAgain sync.Map - -// doneKeysMu keeps one take-in of a module's done key at a time in this process, per module. -var doneKeysMu sync.Map - -// takeDoneKey is a woken module's take-in on the live stores and buckets, said. A take-in that fails (its compositions -// or its grant not written) is tried again every doneKeyRetry while it fails, loudly: printed each time, and an urgent -// condition raised by the first until it has worked, as the chain's own moves are (layBuildsOutSoon). A key not done -// yet is no failure: it is said once for each reason it waits for and looked at again every doneKeyRetry. -func takeDoneKey(ctx context.Context, open *stores, module string) { - mu, _ := doneKeysMu.LoadOrStore(module, &sync.Mutex{}) - for { - if !mu.(*sync.Mutex).TryLock() { - // A take-in of the module already follows it: it is told to look again once it is done, so a newer key - // written meanwhile is never lost (the review of issue 1945). - doneKeysLookAgain.Store(module, true) - return - } - followDoneKey(ctx, open, module) - if doneKeysBeforeLetGo != nil { - doneKeysBeforeLetGo(module) - } - mu.(*sync.Mutex).Unlock() - // A wake between this follower's last look and its letting go failed its TryLock and only marked the module: - // looked at again now that it is let go, and taken once more (the review of issue 1945). - if _, again := doneKeysLookAgain.LoadAndDelete(module); !again || ctx.Err() != nil { - return - } - } +// settledKey is what a batch did with one module. +type settledKey struct { + module string + verdict verdict + // taken: registered, composed and granted now (not nothing new); its condition is cleared. + taken bool + // nothingNew: registered and composed before, so nothing was done. + nothingNew bool + // waits is why it waits, "" when it does not. + waits string + // err is a failure: raised as the module's condition. + err error + // said is the line said of it, "" for none. + said string } -// followDoneKey takes a module's done key in, under the module's lock, until it is taken in or ctx ends: a failure or a -// wait is tried again every doneKeyRetry. -func followDoneKey(ctx context.Context, open *stores, module string) { - failed := false - saidWait := "" - for { - doneKeysLookAgain.Delete(module) // what this look reads covers every wake before it - err := takeOnceInSlot(ctx, open, module) - o := doneKeyObservation(module) - if err == nil { - // Cleared on every success, whoever raised it: a controller before a restart or another follower (issue - // 1945). A condition that is not open costs one read. - if keeper := conditionsFrom; keeper != nil { - _, _ = keeper.Clear(ctx, o.Key(), "composed and granted") - } - failed = false - if _, again := doneKeysLookAgain.LoadAndDelete(module); again { - continue // a wake came while this one ran: once more, for the key written meanwhile - } - return +// batch takes the delivered keys in, each module once: it answers one settledKey per module in the order first +// delivered, and the number of composer passes it asked. memo is, per module, the trunk (fingerprint and commit) this +// process took in last; a module at the same trunk delivered for the first time has nothing new. +func (s takeInSteps) batch(ctx context.Context, keys []delivered, memo map[string]string) ([]settledKey, int) { + var order []string + again := map[string]bool{} + for _, k := range keys { + if _, seen := again[k.module]; !seen { + order = append(order, k.module) } - if errors.Is(err, errKeyWaits) { - // A take-in that waits is said once for each reason, and tried again every doneKeyRetry: the key write that - // would wake it may never come (issue 1945, live on 2026-10-25: mesh-host's take-in waited unsaid). - if why := err.Error(); why != saidWait { - doneKeysSay(fmt.Sprintf("done keys: %s waits: %s; looked at again every %s", module, - strings.TrimPrefix(why, errKeyWaits.Error()+": "), doneKeyRetry)) - saidWait = why - } - } else { - doneKeysSay(fmt.Sprintf("done keys: %s: %v; tried again in %s", module, err, doneKeyRetry)) + again[k.module] = again[k.module] || k.again + } + out := make([]settledKey, 0, len(order)) + all := func(v verdict, err error) ([]settledKey, int) { + for _, m := range order { + out = append(out, settledKey{module: m, verdict: v, err: err}) } - if errors.Is(err, errKeyWaits) { - select { - case <-ctx.Done(): - return - case <-time.After(doneKeyRetry): - } + return out, 0 + } + held, err := s.catalogue(ctx) + if err != nil { + return all(refused, fmt.Errorf("the catalogue cannot be read: %w", err)) + } + type toCompose struct { + at int + nodes []string + mark string + said string + } + var need []toCompose + for _, m := range order { + r := settledKey{module: m} + nodes, isHeld := held[m] + if !isHeld { + out = append(out, r) continue } - if ctx.Err() != nil { - return // stopped: a take-in cut by the stop is no fault of the build, and the next controller takes it in + trunk, err := s.trunk(ctx, m) + if errors.Is(err, builderkeys.ErrNoValue) { + out = append(out, r) + continue } - if keeper := conditionsFrom; keeper != nil && !failed && !errors.Is(err, context.Canceled) { - // A take-in whose own step was cancelled is tried again, and raises nothing: a cancellation is no fault - // of the build (issue 1945: 30 urgent conditions live, each a take-in cut by a restart). - o.Summary += err.Error() - if _, oerr := keeper.Observe(ctx, o); oerr == nil { - failed = true + if err == nil && !strings.HasPrefix(trunk.FP2, sdkdelivery.SourcePrefix2) { + // The indexer's next write names one: said once, as every wait for a write (ADR 0544 §4). + why := trunk.None2 + if why == "" { + why = "the indexer wrote no fingerprint of the new form" + } + r.waits = fmt.Sprintf("%v: %s at %.8s has no fingerprint to take a build by: %s", errWaitsForAWrite, m, + trunk.Commit, why) + out = append(out, r) + continue + } + if err != nil { + r.verdict, r.err = refused, err + out = append(out, r) + continue + } + mark := trunk.FP2 + "@" + trunk.Commit + if !again[m] && memo[m] == mark { + r.nothingNew = true + out = append(out, r) + continue + } + at, atFP, err := s.registered(ctx, m, trunk) + if err == nil && !at && atFP { + // Only the commit moved (issue 1945: every catalogue merge took about 200 modules in again, 25 minutes): the + // new commit is recorded, and the module is judged as one registered, composed only if something is missing. + if err = s.recordCommit(ctx, m, trunk.Commit); err == nil { + at = true } } - select { - case <-ctx.Done(): - return - case <-time.After(doneKeyRetry): + if err != nil { + r.verdict, r.err = refused, err + out = append(out, r) + continue + } + said := "" + if !at { + if err := s.takeIn(ctx, m, trunk); err != nil { + switch { + case errors.Is(err, errWaitsForAWrite): + r.waits = err.Error() + case errors.Is(err, errKeyWaits): + r.verdict, r.waits = refused, err.Error() + case errors.Is(err, errKeyFailed): + r.err = err // acknowledged: only a new build ends it, and its key is delivered then + default: + r.verdict, r.err = refused, err + } + out = append(out, r) + continue + } + said = fmt.Sprintf("%s registered from the builder's done key %s (%s)", m, short(trunk.FP2), short(trunk.Commit)) + } else if !again[m] { + // Registered at its trunk already: composed and granted by a take-in before this one, or at this + // controller's start, unless that take-in stopped half-way, which leaves a composition or the grant missing. + if done, err := s.taken(ctx, m, trunk.FP2, nodes); err == nil && done { + memo[m] = mark + r.nothingNew = true + out = append(out, r) + continue + } + } + if len(nodes) == 0 { + memo[m] = mark + r.taken, r.said = true, said + out = append(out, r) + continue + } + if said == "" { + said = fmt.Sprintf("%s at %s", m, short(trunk.FP2)) + } + need = append(need, toCompose{at: len(out), nodes: nodes, mark: mark, said: said}) + out = append(out, r) + } + if len(need) == 0 { + return out, 0 + } + refuseNeed := func(err error) ([]settledKey, int) { + for _, n := range need { + out[n.at].verdict, out[n.at].err = refused, err + } + return out, 0 + } + holder, err := s.busHolder(ctx) + if err != nil { + return refuseNeed(fmt.Errorf("the node holding the bus cannot be read: %w", err)) + } + // The modules' nodes first, in one pass: composing them makes what their new manifests need, credentials among it. + var union []string + for _, n := range need { + for _, node := range n.nodes { + if node != holder && !slices.Contains(union, node) { + union = append(union, node) + } } } + sort.Strings(union) + passes := 0 + failed := map[string]error{} + if len(union) > 0 { + f, _, err := s.compose(ctx, union, false) + passes++ + if err != nil { + return refuseNeed(err) + } + maps.Copy(failed, f) + } + // Then the bus's user list, which carries the modules' permissions at their new fingerprints, and the grants, which + // with the compositions let a want name each fingerprint (rules 6 and 7). + var bus []string + if holder != "" { + bus = []string{holder} + } + f, grantsErr, err := s.compose(ctx, bus, true) + passes++ + if err != nil { + return refuseNeed(err) + } + maps.Copy(failed, f) + for _, n := range need { + var errs []error + for _, node := range append(slices.Clone(n.nodes), bus...) { + if e := failed[node]; e != nil && !slices.ContainsFunc(errs, func(x error) bool { return x == e }) { + errs = append(errs, e) + } + } + if grantsErr != nil { + errs = append(errs, grantsErr) + } + r := &out[n.at] + if len(errs) > 0 { + r.verdict, r.err = refused, errors.Join(errs...) + continue + } + memo[r.module] = n.mark + r.taken = true + r.said = n.said + "; composed for " + strings.Join(n.nodes, ", ") + ", its grant written" + } + return out, passes } -// doneKeyRetry is how long a failed take-in of a done key waits before it tries again. +// doneKeysSay says a line of the take-ins; a variable so a test may hear it. +var doneKeysSay = func(s string) { fmt.Println(s) } + +// doneKeyRetry is how long the bus waits before it delivers again a key whose take-in was refused the first time; each +// refusal after doubles it, up to doneKeyRetryMost, so a key refused for long (a manifest word this controller does not +// read yet, a node whose composition keeps failing) costs a take-in every few minutes, not every half minute. var doneKeyRetry = 30 * time.Second +// doneKeyRetryMost is the longest a refused key waits before it is delivered again. +const doneKeyRetryMost = 10 * time.Minute + +// refusalDelay is how long the bus waits before it delivers again a key refused on its deliveries-th delivery. +func refusalDelay(deliveries uint64) time.Duration { + d := doneKeyRetry + for i := uint64(1); i < deliveries && d < doneKeyRetryMost; i++ { + d *= 2 + } + return min(d, doneKeyRetryMost) +} + // kindNotComposed is the condition of a module's new fingerprint whose compositions or grant are not written: until // they are, no want can name it. const kindNotComposed = "fingerprint-not-composed" @@ -228,40 +303,37 @@ func doneKeyObservation(module string) conditions.Observation { Summary: module + "'s new fingerprint is not composed or granted, so no want can name it: "} } -// takeDoneKeyOnce is one take-in of a module's done key. -func takeDoneKeyOnce(ctx context.Context, open *stores, module string) error { - keys, closeKeys, err := openBuilderKeys() - if err != nil { - return fmt.Errorf("the builder's keys cannot be read: %w", err) - } - defer closeKeys() - said, err := liveDoneKeySteps(open, keys).take(ctx, module) - if err == nil && said != "" { - fmt.Printf("done keys: %s\n", said) - } - return err -} - -// liveDoneKeySteps are the steps on the controller's own stores and the bus. -func liveDoneKeySteps(open *stores, keys registerKeys) doneKeySteps { +// liveTakeInSteps are the steps on the controller's own stores and the bus. +func liveTakeInSteps(open *stores, keys registerKeys, js jetstream.JetStream) takeInSteps { inv := open.inventory - return doneKeySteps{ - held: func(ctx context.Context, module string) (bool, error) { - _, err := inv.SourceOf(ctx, module) - if errors.Is(err, inventory.ErrNoSuchModule) { - return false, nil + return takeInSteps{ + catalogue: func(ctx context.Context) (map[string][]string, error) { + entries, err := inv.Catalogued(ctx) + if err != nil { + return nil, err } - return err == nil, err + out := make(map[string][]string, len(entries)) + for _, e := range entries { + out[e.Manifest.Module] = e.On + } + return out, nil }, trunk: func(ctx context.Context, module string) (sdkdelivery.Module, error) { return keys.ModuleAt(ctx, module, "") }, - registeredAt: func(ctx context.Context, module, commit string) (bool, error) { + registered: func(ctx context.Context, module string, trunk sdkdelivery.Module) (bool, bool, error) { src, err := inv.SourceOf(ctx, module) if err != nil { - return false, err + return false, false, err } - return sameCommitID(src.BuiltFrom, commit), nil + if sameCommitID(src.BuiltFrom, trunk.Commit) { + return true, true, nil + } + fp, err := registeredFingerprint(ctx, inv, keys, module, src.BuiltFrom) + return false, err == nil && fp != "" && fp == trunk.FP2, nil + }, + recordCommit: func(ctx context.Context, module, commit string) error { + return inv.RecordBuiltFrom(ctx, module, commit) }, takeIn: func(ctx context.Context, module string, trunk sdkdelivery.Module) error { _, _, err := takeInKey(ctx, inv, keys, keyTake{Module: module, Trunk: trunk, Ask: registrationAsk{Merge: trunk.Commit}}) @@ -270,17 +342,8 @@ func liveDoneKeySteps(open *stores, keys registerKeys) doneKeySteps { } return err }, - nodesOf: func(ctx context.Context, module string) ([]string, error) { - entries, err := inv.Catalogued(ctx) - if err != nil { - return nil, err - } - for _, e := range entries { - if e.Manifest.Module == module { - return e.On, nil - } - } - return nil, nil + taken: func(ctx context.Context, module, fp string, nodes []string) (bool, error) { + return composedAndGranted(ctx, js, module, fp, nodes) }, busHolder: func(ctx context.Context) (string, error) { holders, err := seatHolders(ctx, inv) @@ -291,123 +354,64 @@ func liveDoneKeySteps(open *stores, keys registerKeys) doneKeySteps { }, // The composer's, never the take-in's own: one pass at a time composes each node asked for once, and writes the // grants once (issue 1945, composer.go). - 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) + compose: func(ctx context.Context, nodes []string, grants bool) (map[string]error, error, error) { + return composerFor(open).composeEach(ctx, nodes, grants) }, } } -// watchDoneKeys wakes the take-in of a module's done key for each module the builder or the indexer writes, until ctx -// ends: never on the watch's own goroutine. -func watchDoneKeys(ctx context.Context, open *stores) { - keys, closeKeys, err := openBuilderKeys() - if err != nil { - fmt.Printf("done keys: the builder's keys are not watched: %v\n", err) - return +// registeredFingerprint is the fingerprint a module is registered at, as currentFingerprints reads it: its commit's +// newest build's, or the indexer's value at the commit when the commit has no build of its own; "" when neither says. +func registeredFingerprint(ctx context.Context, inv *inventory.Inventory, keys registerKeys, module, commit string) (string, error) { + if commit == "" { + return "", nil } - if err := builderkeys.Watch(ctx, keys.JS, func(module string) { - inBackground(ctx, func() { takeDoneKey(ctx, open, module) }) - }); err != nil { - closeKeys() - fmt.Printf("done keys: the builder's keys are not watched: %v\n", err) - return - } - goServing(ctx, func() { - <-ctx.Done() - closeKeys() - }) - // And every module once now, after the watch is open so no write falls between the two. - inBackground(ctx, func() { - doneKeysAtStart(ctx, catalogedModules(open), func(module string) { takeDoneKeyAtStart(ctx, open, module) }) - }) -} - -// doneKeysAtStart takes in the done key of every catalogued module once, one after another: the start's look the watch of -// new writes cannot give (it hears updates only). A done key written while no controller served, or whose take-in the -// controller this one replaced was stopped in, is taken here; without it the module stays in the node's own composition -// with no credential until its keys are written again (issue 1945, live on 2026-10-25 at 02:34). A take-in with nothing -// to do does nothing; a catalogue that cannot be read is said. -func doneKeysAtStart(ctx context.Context, modules func(context.Context) ([]string, error), take func(module string)) { - names, err := modules(ctx) - if err != nil { - fmt.Printf("done keys: the modules are not looked at once at the start: %v\n", err) - return - } - sort.Strings(names) - // Side by side, a few at a time: their compositions meet in the composer's passes, each node composed once per pass - // (issue 1945), so the start's look is not one module's composition after another. - sem := make(chan struct{}, startTakeIns) - var wg sync.WaitGroup - for _, m := range names { - if ctx.Err() != nil { - break + if id, err := inv.BuildOf(ctx, module, commit); err == nil && id != "" { + if b, found, err := inv.BuildByID(ctx, id); err == nil && found && b.SourceFingerprint != "" { + return b.SourceFingerprint, nil } - sem <- struct{}{} - wg.Add(1) - go func() { - defer func() { <-sem; wg.Done() }() - take(m) - }() } - wg.Wait() + at, err := keys.ModuleAt(ctx, module, commit) + if err != nil { + return "", err + } + return at.FP2, nil } -// startTakeIns is how many take-ins the start's look runs at once, and how many run at once in all. -const startTakeIns = 16 - -// takeInSlots are the take-ins that may run at once in this process, the watch's and the start's look's together: a -// burst of done keys (about 120 at 16:07:36 on 2026-10-25) once started a take-in each, all at once, and their reads -// filled the store until it refused every connection (issues 1945 and 1277). -var takeInSlots = make(chan struct{}, startTakeIns) - -// takeOnceInSlot is one take-in once a slot is free; a take-in that waits for its next try holds none. -func takeOnceInSlot(ctx context.Context, open *stores, module string) error { - select { - case takeInSlots <- struct{}{}: - case <-ctx.Done(): - return ctx.Err() +// composedAndGranted says the bus holds module's grant at fp and its composition at fp for every node: a few reads of +// keys, never a composition. +func composedAndGranted(ctx context.Context, js jetstream.JetStream, module, fp string, nodes []string) (bool, error) { + grants, err := js.KeyValue(ctx, broker.GrantsBucket) + if err != nil { + return false, err } - defer func() { <-takeInSlots }() - return doneKeyTakeOnce(ctx, open, module) -} - -// takeDoneKeyAtStart is one take-in of the start's look, under the module's lock; one that fails is handed to takeDoneKey, -// which tries again every doneKeyRetry, loudly, so a failing module never holds up the look at the others. -func takeDoneKeyAtStart(ctx context.Context, open *stores, module string) { - mu, _ := doneKeysMu.LoadOrStore(module, &sync.Mutex{}) - if !mu.(*sync.Mutex).TryLock() { - // A take-in already follows the module: told to look again, it takes what the start would have (issue 1945). - doneKeysLookAgain.Store(module, true) - return + key, err := sdkdelivery.ControllerGrantKey(module, fp) + if err != nil { + return false, err } - doneKeysLookAgain.Delete(module) // what this look reads covers every wake before it - err := takeOnceInSlot(ctx, open, module) - mu.(*sync.Mutex).Unlock() - // Read after letting go, as every holder of a module's lock does: a wake during this look only marked the module - // (the review of issue 1945). - _, again := doneKeysLookAgain.LoadAndDelete(module) - if err != nil || again { - // A failure, a wait or a wake meanwhile is followed by takeDoneKey, which says it and looks again. - inBackground(ctx, func() { takeDoneKey(ctx, open, module) }) + if _, err := grants.Get(ctx, key); err != nil { + if errors.Is(err, jetstream.ErrKeyNotFound) { + return false, nil + } + return false, err } -} - -// catalogedModules are the names of every module the catalogue holds. -func catalogedModules(open *stores) func(context.Context) ([]string, error) { - return func(ctx context.Context) ([]string, error) { - entries, err := open.inventory.Catalogued(ctx) + digests, err := js.KeyValue(ctx, sdkdelivery.ControllerComposedBucket) + if err != nil { + return false, err + } + for _, node := range nodes { + key, err := sdkdelivery.ComposedKey(node, module, fp) if err != nil { - return nil, err + return false, err } - out := make([]string, 0, len(entries)) - for _, e := range entries { - out = append(out, e.Manifest.Module) + if _, err := digests.Get(ctx, key); err != nil { + if errors.Is(err, jetstream.ErrKeyNotFound) { + return false, nil + } + return false, err } - return out, nil } + return true, nil } // doneKeysComposers are the composers of the done keys' take-ins in this process, one for each set of stores (a test diff --git a/cmd/mesh-controller/composed_keys_cancel_test.go b/cmd/mesh-controller/composed_keys_cancel_test.go deleted file mode 100644 index 0e34904c..00000000 --- a/cmd/mesh-controller/composed_keys_cancel_test.go +++ /dev/null @@ -1,83 +0,0 @@ -package main - -import ( - "context" - "fmt" - "testing" - "time" - - "github.com/novox/mesh-controller/internal/conditions" -) - -// A take-in cut by a cancellation raises no condition (issue 1945: on 2026-10-25 thirty urgent fingerprint-not-composed -// conditions were live, each a take-in a stop or a restart cut with "context canceled"): when the controller stops, the -// next one takes it in; when a step of its own was cancelled, it is tried again. -func TestATakeInCutByACancellationRaisesNothing(t *testing.T) { - for _, tc := range []struct { - name string - stop bool // the controller's context ends, else only the step's error is a cancellation - }{{"the controller stops", true}, {"a step was cancelled", false}} { - t.Run(tc.name, func(t *testing.T) { - was, wasRetry, wasSay, wasKeeper := doneKeyTakeOnce, doneKeyRetry, doneKeysSay, conditionsFrom - t.Cleanup(func() { - doneKeyTakeOnce, doneKeyRetry, doneKeysSay, conditionsFrom = was, wasRetry, wasSay, wasKeeper - }) - store := conditions.NewInMemory() - k := conditions.NewKeeper(t.Context(), conditions.Options{Store: store, History: store, - Teller: &conditions.Told{}}) - t.Cleanup(func() { k.Close(context.Background()) }) - conditionsFrom = k - doneKeyRetry = time.Millisecond - doneKeysSay = func(string) {} - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - calls := 0 - doneKeyTakeOnce = func(context.Context, *stores, string) error { - calls++ - if tc.stop { - cancel() // stopped mid-way - return fmt.Errorf("the module cannot be composed: %w", context.Canceled) - } - if calls == 1 { - return fmt.Errorf("the module cannot be composed: %w", context.Canceled) - } - return nil - } - done := make(chan struct{}) - go func() { takeDoneKey(ctx, nil, "cut-module"); close(done) }() - select { - case <-done: - case <-time.After(5 * time.Second): - t.Fatal("the take-in did not end") - } - if !tc.stop && calls != 2 { - t.Errorf("a take-in whose step was cancelled was tried %d time(s), want 2", calls) - } - if c, found, err := conditions.ReadOne(t.Context(), store, doneKeyObservation("cut-module").Key()); err != nil || found { - t.Fatalf("a take-in cut by a cancellation raised %+v (found %v, %v)", c, found, err) - } - }) - } -} - -// A take-in that succeeds clears the module's fingerprint-not-composed condition whoever raised it: a controller before a -// restart, or another follower (issue 1945: the operator read "a build needs a look" for take-ins that had since worked). -func TestASucceedingTakeInClearsAConditionAnotherRaised(t *testing.T) { - was, wasRetry, wasSay, wasKeeper := doneKeyTakeOnce, doneKeyRetry, doneKeysSay, conditionsFrom - t.Cleanup(func() { doneKeyTakeOnce, doneKeyRetry, doneKeysSay, conditionsFrom = was, wasRetry, wasSay, wasKeeper }) - store := conditions.NewInMemory() - k := conditions.NewKeeper(t.Context(), conditions.Options{Store: store, History: store, Teller: &conditions.Told{}}) - t.Cleanup(func() { k.Close(context.Background()) }) - conditionsFrom = k - doneKeysSay = func(string) {} - o := doneKeyObservation("raised-before") - o.Summary += "the store refused the connection" - if _, err := k.Observe(t.Context(), o); err != nil { - t.Fatal(err) - } - doneKeyTakeOnce = func(context.Context, *stores, string) error { return nil } - takeDoneKey(t.Context(), nil, "raised-before") - if c, found, err := conditions.ReadOne(t.Context(), store, o.Key()); err != nil || found { - t.Fatalf("a take-in that succeeded left the condition another raised: %+v (found %v, %v)", c, found, err) - } -} diff --git a/cmd/mesh-controller/composed_keys_cap_test.go b/cmd/mesh-controller/composed_keys_cap_test.go deleted file mode 100644 index 54303253..00000000 --- a/cmd/mesh-controller/composed_keys_cap_test.go +++ /dev/null @@ -1,58 +0,0 @@ -package main - -import ( - "context" - "fmt" - "sync" - "sync/atomic" - "testing" - "time" -) - -// A burst of done keys runs at most startTakeIns take-ins at once, the watch's and the start's look together: at -// 16:07:36 on 2026-10-25 the builder rewrote about 120 done keys at once, each woke its own take-in, and each opened -// pools of its own until novox's store refused every connection (issues 1945 and 1277). -func TestABurstOfDoneKeysRunsAtMostSixteenTakeInsAtOnce(t *testing.T) { - was := doneKeyTakeOnce - t.Cleanup(func() { doneKeyTakeOnce = was }) - var running, most atomic.Int64 - var taken atomic.Int64 - doneKeyTakeOnce = func(context.Context, *stores, string) error { - now := running.Add(1) - for { - m := most.Load() - if now <= m || most.CompareAndSwap(m, now) { - break - } - } - time.Sleep(2 * time.Millisecond) - running.Add(-1) - taken.Add(1) - return nil - } - ctx := context.Background() - var wg sync.WaitGroup - // The watch's wakes: one take-in each, in the background, as watchDoneKeys starts them. - for i := range 120 { - wg.Add(1) - go func() { defer wg.Done(); takeDoneKey(ctx, nil, fmt.Sprintf("burst-%03d", i)) }() - } - // And the start's look at the same moment, over other modules. - wg.Add(1) - go func() { - defer wg.Done() - var start []string - for i := range 40 { - start = append(start, fmt.Sprintf("start-%03d", i)) - } - doneKeysAtStart(ctx, func(context.Context) ([]string, error) { return start, nil }, - func(m string) { takeDoneKeyAtStart(ctx, nil, m) }) - }() - wg.Wait() - if got := taken.Load(); got != 160 { - t.Fatalf("took in %d modules, want 160", got) - } - if m := most.Load(); m > startTakeIns { - t.Fatalf("ran %d take-ins at once, want at most %d", m, startTakeIns) - } -} diff --git a/cmd/mesh-controller/composed_keys_start_test.go b/cmd/mesh-controller/composed_keys_start_test.go deleted file mode 100644 index 5dece4d8..00000000 --- a/cmd/mesh-controller/composed_keys_start_test.go +++ /dev/null @@ -1,34 +0,0 @@ -package main - -import ( - "context" - "errors" - "reflect" - "sort" - "sync" - "testing" -) - -// A starting controller takes in the done key of every catalogued module once, beside the watch of new writes: a done key -// written while no controller served, or whose take-in the controller it replaced was stopped in, is otherwise never -// taken, and its module stays in the node's own composition with no credential (issue 1945, live on 2026-10-25 at 02:34: -// the controller replaced mid take-in left mesh-provisioner-client, gitea, postgres and many more unregistered at their -// new fingerprints, and the new one, watching updates only, never looked). -func TestAStartingControllerTakesEveryModulesDoneKeyOnce(t *testing.T) { - var took []string - var mu sync.Mutex - doneKeysAtStart(context.Background(), func(context.Context) ([]string, error) { - return []string{"mesh-provisioner-client", "gitea", "postgres"}, nil - }, func(module string) { mu.Lock(); took = append(took, module); mu.Unlock() }) - sort.Strings(took) - if want := []string{"gitea", "mesh-provisioner-client", "postgres"}; !reflect.DeepEqual(took, want) { - t.Fatalf("took %v, want %v", took, want) - } - // A catalogue that cannot be read takes nothing and is said, never fatal. - took = nil - doneKeysAtStart(context.Background(), func(context.Context) ([]string, error) { return nil, errors.New("down") }, - func(module string) { took = append(took, module) }) - if len(took) != 0 { - t.Fatalf("took %v from a catalogue that could not be read", took) - } -} diff --git a/cmd/mesh-controller/composed_keys_test.go b/cmd/mesh-controller/composed_keys_test.go index b336ba9f..9e133ce5 100644 --- a/cmd/mesh-controller/composed_keys_test.go +++ b/cmd/mesh-controller/composed_keys_test.go @@ -4,84 +4,305 @@ import ( "context" "encoding/json" "errors" + "fmt" + "slices" "strings" "testing" + "time" sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" + "github.com/nats-io/nats.go/jetstream" "github.com/novox/mesh-controller/internal/builderkeys" ) -// stepsFor are doneKeySteps over a module registered at registered, whose trunk value names fp at commit, assigned to -// nodes, the bus held on holder; every act is written to acts in order. -func stepsFor(acts *[]string, registered, commit, fp string, nodes []string, holder string) doneKeySteps { - return doneKeySteps{ - held: func(_ context.Context, module string) (bool, error) { return module == "mesh-mail", nil }, - trunk: func(context.Context, string) (sdkdelivery.Module, error) { - if fp == "" { +// aTakeIn is a mesh for the take-in's steps: per module its registered commit, its trunk (commit, fingerprint), its +// nodes, whether its compositions and grant exist, and what its take-in answers; every act is written to acts. +type aTakeIn struct { + registered map[string]string + sameFP map[string]bool // registered at another commit, at the trunk's fingerprint + trunks map[string]string // module -> commit; no entry: no trunk value + nodes map[string][]string + taken map[string]bool + takeInErr map[string]error + composeErr map[string]error + grantsErr error + holder string + acts []string +} + +func (m *aTakeIn) steps() takeInSteps { + fp := func(module string) string { return sdkdelivery.SourcePrefix2 + strings.Repeat(module[:1], 64) } + return takeInSteps{ + catalogue: func(context.Context) (map[string][]string, error) { + m.acts = append(m.acts, "catalogue") + out := map[string][]string{} + for k, v := range m.nodes { + out[k] = v + } + return out, nil + }, + trunk: func(_ context.Context, module string) (sdkdelivery.Module, error) { + commit, ok := m.trunks[module] + if !ok { return sdkdelivery.Module{}, builderkeys.ErrNoValue } - return sdkdelivery.Module{Commit: commit, FP2: fp}, nil + return sdkdelivery.Module{Commit: commit, FP2: fp(module)}, nil + }, + registered: func(_ context.Context, module string, trunk sdkdelivery.Module) (bool, bool, error) { + if m.registered[module] == trunk.Commit { + return true, true, nil + } + return false, m.sameFP[module], nil + }, + recordCommit: func(_ context.Context, module, commit string) error { + m.acts = append(m.acts, "record "+module+" at "+commit) + return nil }, - registeredAt: func(_ context.Context, _ string, c string) (bool, error) { return c == registered, nil }, takeIn: func(_ context.Context, module string, trunk sdkdelivery.Module) error { - *acts = append(*acts, "take in "+module+" at "+trunk.Commit) - return nil + m.acts = append(m.acts, "take in "+module+" at "+trunk.Commit) + return m.takeInErr[module] }, - nodesOf: func(context.Context, string) ([]string, error) { return nodes, nil }, - busHolder: func(context.Context) (string, error) { return holder, nil }, - compose: func(_ context.Context, ns []string) error { - *acts = append(*acts, "compose "+strings.Join(ns, ",")) - return nil + taken: func(_ context.Context, module, _ string, _ []string) (bool, error) { return m.taken[module], nil }, + busHolder: func(context.Context) (string, error) { return m.holder, nil }, + compose: func(_ context.Context, nodes []string, grants bool) (map[string]error, error, error) { + act := "compose " + strings.Join(nodes, ",") + if grants { + act += " and grants" + } + m.acts = append(m.acts, act) + failed := map[string]error{} + for _, n := range nodes { + if e := m.composeErr[n]; e != nil { + failed[n] = e + } + } + if grants { + return failed, m.grantsErr, nil + } + return failed, nil, nil }, - grants: func(context.Context) error { *acts = append(*acts, "grants"); return nil }, } } -// 1945/AC3 (unit half): a done key of a new fingerprint of an assigned module is taken in, then the module's nodes are -// composed, then the node holding the bus (its user list), then the grants: the grant and the composition exist -// before the assigner's rules let a want name the fingerprint. A module registered at the trunk's commit already is -// composed and granted without a second take-in; one assigned nowhere, unregistered, or with no trunk value is left -// alone. -func TestADoneKeyIsTakenInComposedAndGranted(t *testing.T) { - fp := sdkdelivery.SourcePrefix2 + strings.Repeat("c", 64) - var acts []string - said, err := stepsFor(&acts, "old", "new", fp, []string{"ace", "g14"}, "novox").take(context.Background(), "mesh-mail") - if err != nil { - t.Fatal(err) +// A batch of done keys takes each module in once and asks the composer at most twice, whatever its size: once for the +// nodes of every module with something new, then once for the node holding the bus with the grants (ADR 0544 §5). A +// module registered at its trunk whose compositions and grant exist, or that this process took in at the same trunk, +// composes nothing (§3): the old take-in composed it three times over, every key and every start (issue 1945, 17:36 on +// 2026-10-25). A key delivered again is taken in whole. A wait for a write is acknowledged, a wait on a read and a +// failure refused, a failed build key acknowledged with its condition (§4). +func TestATakeInBatch(t *testing.T) { + waitsForAWrite := fmt.Errorf("%w: the key is building", errWaitsForAWrite) + waitsOnARead := fmt.Errorf("%w: the store is away", errKeyWaits) + keyFailed := fmt.Errorf("%w: the build failed", errKeyFailed) + type want struct { + verdict verdict + taken, nothingNew bool + waits, err bool } - if got := strings.Join(acts, "; "); got != "take in mesh-mail at new; compose ace,g14; compose novox; grants" { - t.Errorf("the acts were %q", got) + cases := []struct { + name string + mesh aTakeIn + keys []delivered + memo map[string]string + acts []string + passes int + want map[string]want + }{ + {name: "a new fingerprint: taken in, its nodes, then the bus's node and the grants", + mesh: aTakeIn{registered: map[string]string{"mail": "old"}, trunks: map[string]string{"mail": "new"}, + nodes: map[string][]string{"mail": {"ace", "g14", "novox"}}, holder: "novox"}, + keys: []delivered{{module: "mail"}}, + acts: []string{"catalogue", "take in mail at new", "compose ace,g14", "compose novox and grants"}, passes: 2, + want: map[string]want{"mail": {taken: true}}}, + {name: "registered, composed and granted: nothing composed (fails on the old take-in)", + mesh: aTakeIn{registered: map[string]string{"mail": "new", "nats": "n"}, trunks: map[string]string{"mail": "new", "nats": "n"}, + nodes: map[string][]string{"mail": {"ace"}, "nats": {"novox"}}, taken: map[string]bool{"mail": true, "nats": true}, + holder: "novox"}, + keys: []delivered{{module: "mail"}, {module: "nats"}, {module: "mail"}}, + acts: []string{"catalogue"}, + want: map[string]want{"mail": {nothingNew: true}, "nats": {nothingNew: true}}}, + {name: "a catalogue merge that moved the commit and not the fingerprint: recorded, nothing composed", + mesh: aTakeIn{registered: map[string]string{"mail": "a85d9037"}, sameFP: map[string]bool{"mail": true}, + trunks: map[string]string{"mail": "2df05030"}, nodes: map[string][]string{"mail": {"ace"}}, + taken: map[string]bool{"mail": true}, holder: "novox"}, + keys: []delivered{{module: "mail"}}, + acts: []string{"catalogue", "record mail at 2df05030"}, want: map[string]want{"mail": {nothingNew: true}}}, + {name: "taken in by this process at the same trunk: nothing new, not even a look at its keys", + mesh: aTakeIn{registered: map[string]string{"mail": "new"}, trunks: map[string]string{"mail": "new"}, + nodes: map[string][]string{"mail": {"ace"}}, holder: "novox"}, + keys: []delivered{{module: "mail"}}, + memo: map[string]string{"mail": sdkdelivery.SourcePrefix2 + strings.Repeat("m", 64) + "@new"}, + acts: []string{"catalogue"}, want: map[string]want{"mail": {nothingNew: true}}}, + {name: "delivered again: taken in whole, even when it looks done", + mesh: aTakeIn{registered: map[string]string{"mail": "new"}, trunks: map[string]string{"mail": "new"}, + nodes: map[string][]string{"mail": {"ace"}}, taken: map[string]bool{"mail": true}, holder: "novox"}, + keys: []delivered{{module: "mail", again: true}}, + acts: []string{"catalogue", "compose ace", "compose novox and grants"}, passes: 2, + want: map[string]want{"mail": {taken: true}}}, + {name: "a take-in that stopped half-way: composed", + mesh: aTakeIn{registered: map[string]string{"mail": "new"}, trunks: map[string]string{"mail": "new"}, + nodes: map[string][]string{"mail": {"ace"}}, holder: "novox"}, + keys: []delivered{{module: "mail"}}, + acts: []string{"catalogue", "compose ace", "compose novox and grants"}, passes: 2, + want: map[string]want{"mail": {taken: true}}}, + {name: "three modules, one pass for their nodes and one for the bus", + mesh: aTakeIn{registered: map[string]string{}, trunks: map[string]string{"mail": "1", "web": "2", "bus": "3"}, + nodes: map[string][]string{"mail": {"ace", "novox"}, "web": {"g14", "ace"}, "bus": {"novox"}}, holder: "novox"}, + keys: []delivered{{module: "mail"}, {module: "web"}, {module: "bus"}}, + acts: []string{"catalogue", "take in mail at 1", "take in web at 2", "take in bus at 3", "compose ace,g14", + "compose novox and grants"}, passes: 2, + want: map[string]want{"mail": {taken: true}, "web": {taken: true}, "bus": {taken: true}}}, + {name: "assigned nowhere, held by no catalogue, no trunk value: nothing composed", + mesh: aTakeIn{registered: map[string]string{}, trunks: map[string]string{"mail": "1", "odd": "2"}, + nodes: map[string][]string{"mail": nil, "none": {"ace"}}, holder: "novox"}, + keys: []delivered{{module: "mail"}, {module: "odd"}, {module: "none"}}, + acts: []string{"catalogue", "take in mail at 1"}, + want: map[string]want{"mail": {taken: true}, "odd": {}, "none": {}}}, + {name: "a grant not written refuses the key, never swallowed", + mesh: aTakeIn{registered: map[string]string{}, trunks: map[string]string{"mail": "1"}, + nodes: map[string][]string{"mail": {"ace"}}, holder: "novox", grantsErr: errors.New("the bucket mesh-controller_grants: not found")}, + keys: []delivered{{module: "mail"}}, + acts: []string{"catalogue", "take in mail at 1", "compose ace", "compose novox and grants"}, passes: 2, + want: map[string]want{"mail": {verdict: refused, err: true}}}, + {name: "a node that fails refuses only its modules", + mesh: aTakeIn{registered: map[string]string{}, trunks: map[string]string{"mail": "1", "web": "2"}, + nodes: map[string][]string{"mail": {"ace"}, "web": {"g14"}}, holder: "novox", + composeErr: map[string]error{"ace": errors.New("store refused")}}, + keys: []delivered{{module: "mail"}, {module: "web"}}, + acts: []string{"catalogue", "take in mail at 1", "take in web at 2", "compose ace,g14", "compose novox and grants"}, + passes: 2, want: map[string]want{"mail": {verdict: refused, err: true}, "web": {taken: true}}}, + {name: "waits and failures: a write's wait acknowledged, a read's refused, a failed key acknowledged", + mesh: aTakeIn{registered: map[string]string{}, trunks: map[string]string{"a": "1", "b": "2", "c": "3"}, + nodes: map[string][]string{"a": {"ace"}, "b": {"ace"}, "c": {"ace"}}, holder: "novox", + takeInErr: map[string]error{"a": waitsForAWrite, "b": waitsOnARead, "c": keyFailed}}, + keys: []delivered{{module: "a"}, {module: "b"}, {module: "c"}}, + acts: []string{"catalogue", "take in a at 1", "take in b at 2", "take in c at 3"}, + want: map[string]want{"a": {waits: true}, "b": {verdict: refused, waits: true}, "c": {err: true}}}, } - if !strings.Contains(said, "registered from the builder's done key") { - t.Errorf("said %q", said) + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + memo := c.memo + if memo == nil { + memo = map[string]string{} + } + settled, passes := c.mesh.steps().batch(t.Context(), c.keys, memo) + if !slices.Equal(c.mesh.acts, c.acts) { + t.Fatalf("the acts were\n %q\nwant\n %q", c.mesh.acts, c.acts) + } + if passes != c.passes { + t.Fatalf("%d composer pass(es), want %d", passes, c.passes) + } + if len(settled) != len(c.want) { + t.Fatalf("settled %d module(s), want %d: %+v", len(settled), len(c.want), settled) + } + for _, r := range settled { + w := c.want[r.module] + got := want{verdict: r.verdict, taken: r.taken, nothingNew: r.nothingNew, waits: r.waits != "", err: r.err != nil} + if got != w { + t.Errorf("%s: %+v, want %+v (%v)", r.module, got, w, r.err) + } + if r.taken && memo[r.module] == "" { + t.Errorf("%s was taken in and not remembered", r.module) + } + } + }) } - acts = nil - if _, err := stepsFor(&acts, "new", "new", fp, []string{"novox"}, "novox").take(context.Background(), "mesh-mail"); err != nil { - t.Fatal(err) - } - if got := strings.Join(acts, "; "); got != "compose novox; grants" { - t.Errorf("a module registered at its trunk already: %q", got) - } + // And each delivered key is acknowledged or refused by its module's verdict, the refused with the retry's delay. + t.Run("acknowledged and refused", func(t *testing.T) { + mesh := aTakeIn{registered: map[string]string{}, trunks: map[string]string{"a": "1", "b": "2"}, + nodes: map[string][]string{"a": {"ace"}, "b": {"ace"}}, takeInErr: map[string]error{"b": waitsOnARead}} + var acked, naked []string + msg := func(module string) takeInMsg { + return takeInMsg{delivered: delivered{module: module, deliveries: 3}, ack: func() { acked = append(acked, module) }, + nak: func(d time.Duration) { + if d != refusalDelay(3) { + t.Errorf("refused with a delay of %s", d) + } + naked = append(naked, module) + }} + } + state := &takeInState{memo: map[string]string{}, raised: map[string]bool{}, saidWait: map[string]string{}} + said := state.take(t.Context(), mesh.steps(), []takeInMsg{msg("a"), msg("b"), msg("a")}) + if !slices.Equal(acked, []string{"a", "a"}) || !slices.Equal(naked, []string{"b"}) { + t.Fatalf("acknowledged %v and refused %v", acked, naked) + } + if !strings.Contains(said, "1 taken in") || !strings.Contains(said, "1 waiting") { + t.Fatalf("said %q", said) + } + }) - for name, s := range map[string]doneKeySteps{ - "assigned nowhere": stepsFor(&acts, "old", "new", fp, nil, "novox"), - "no trunk value": stepsFor(&acts, "old", "new", "", []string{"ace"}, "novox"), + // A key refused for long holds no slot the others need: each consumer leaves far more keys unacknowledged than a + // batch takes, so 8 keys refused on each stop no ninth (the review of ADR 0544: with 8 a consumer, they stopped it + // for good), and a refusal waits longer each time, up to ten minutes. + t.Run("refused keys stop no other", func(t *testing.T) { + for _, bucket := range []string{sdkdelivery.ModulesBucket, sdkdelivery.BuilderBuildsBucket} { + if c := takeInConfig(bucket); c.MaxAckPending < 8+startTakeIns || c.AckPolicy != jetstream.AckExplicitPolicy { + t.Errorf("%s: %d unacknowledged at most (%v): 8 refused keys and a batch do not fit", bucket, + c.MaxAckPending, c.AckPolicy) + } + } + was := time.Duration(0) + for n := uint64(1); n <= 8; n++ { + d := refusalDelay(n) + if d < was || d > doneKeyRetryMost || (n == 1 && d != doneKeyRetry) { + t.Fatalf("the %d. refusal waits %s (the one before %s)", n, d, was) + } + was = d + } + if was != doneKeyRetryMost { + t.Fatalf("the 8. refusal waits %s, want %s", was, doneKeyRetryMost) + } + }) +} + +// The take-in stalls when it holds work (keys not delivered yet, or a batch handed and not settled) and settles none +// for five minutes; refused keys waiting for their delay are no work, so a quiet mesh is no stall. The first look +// always says what it found, so a stall raised before a restart is cleared once this controller moves (ADR 0544 §6 +// and its second review). +func TestATakeInStallsWhenItHoldsWorkAndSettlesNone(t *testing.T) { + at := time.Date(2026, 10, 25, 17, 36, 0, 0, time.UTC) + type reading struct { + after time.Duration + pending, taking uint64 + handed, settled uint64 + stalled, changed bool + } + for _, c := range []struct { + name string + readings []reading + }{ + {"keys not delivered and nothing settled", []reading{ + {0, 190, 8, 50, 50, false, true}, // the first look always says what it found + {4 * time.Minute, 190, 8, 50, 50, false, false}, + {5 * time.Minute, 190, 8, 50, 50, true, true}, + {6 * time.Minute, 190, 8, 50, 50, true, false}, + {7 * time.Minute, 180, 8, 60, 60, false, true}, + {20 * time.Minute, 0, 0, 60, 60, false, false}, + }}, + {"refused keys waiting alone are no stall", []reading{ + {0, 0, 3, 40, 40, false, true}, + {30 * time.Minute, 0, 3, 40, 40, false, false}, + }}, + {"every key settled, one of them refused by the bus, nothing pending: no stall", []reading{ + {0, 0, 1, 3, 3, false, true}, + {6 * time.Minute, 0, 1, 3, 3, false, false}, + }}, + {"a batch handed and never settled is a stall", []reading{ + {0, 0, 16, 56, 40, false, true}, + {5 * time.Minute, 0, 16, 56, 40, true, true}, + }}, } { - acts = nil - if _, err := s.take(context.Background(), "mesh-mail"); err != nil { - t.Fatal(err) - } - if name == "assigned nowhere" && strings.Join(acts, "; ") != "take in mesh-mail at new" { - t.Errorf("%s: %v", name, acts) - } - if name == "no trunk value" && len(acts) != 0 { - t.Errorf("%s: %v", name, acts) - } - } - acts = nil - if _, err := stepsFor(&acts, "old", "new", fp, []string{"ace"}, "novox").take(context.Background(), "unknown"); err != nil || len(acts) != 0 { - t.Errorf("a module the catalogue does not hold: %v %v", acts, err) + t.Run(c.name, func(t *testing.T) { + w := &stallWatch{} + for i, r := range c.readings { + stalled, changed := w.look(takeInBacklog{Pending: r.pending, Taking: r.taking}, r.handed, r.settled, + at.Add(r.after)) + if stalled != r.stalled || changed != r.changed { + t.Fatalf("reading %d: stalled %v changed %v, want %v %v", i, stalled, changed, r.stalled, r.changed) + } + } + }) } } @@ -115,15 +336,3 @@ func TestAChangedUserListIsANewRevisionOfTheBusModulesCompositionAlone(t *testin } } } - -// 1945 (the review's D1): a grant that is not written is the take-in's error, never swallowed: with no send left, -// nothing else would write it, and no want could name the fingerprint. -func TestAGrantNotWrittenIsTheTakeInsError(t *testing.T) { - fp := sdkdelivery.SourcePrefix2 + strings.Repeat("c", 64) - var acts []string - s := stepsFor(&acts, "old", "new", fp, []string{"ace"}, "novox") - s.grants = func(context.Context) error { return errors.New("the bucket mesh-controller_grants: not found") } - if _, err := s.take(context.Background(), "mesh-mail"); err == nil || !strings.Contains(err.Error(), "grants") { - t.Errorf("a grant not written: %v", err) - } -} diff --git a/cmd/mesh-controller/composed_keys_waits_test.go b/cmd/mesh-controller/composed_keys_waits_test.go deleted file mode 100644 index 839f9b4f..00000000 --- a/cmd/mesh-controller/composed_keys_waits_test.go +++ /dev/null @@ -1,184 +0,0 @@ -package main - -import ( - "context" - "fmt" - "strings" - "sync" - "testing" - "time" -) - -// A take-in that waits is said once for each reason it waits for, and tried again every doneKeyRetry until it is taken -// in: never left for a key write that may not come (issue 1945, live on 2026-10-25: mesh-host's take-in at 26bd waited, -// said nothing, and nothing woke it until another write). -func TestAWaitingTakeInIsSaidOnceAndTriedAgain(t *testing.T) { - was, wasRetry, wasSay := doneKeyTakeOnce, doneKeyRetry, doneKeysSay - t.Cleanup(func() { doneKeyTakeOnce, doneKeyRetry, doneKeysSay = was, wasRetry, wasSay }) - doneKeyRetry = 5 * time.Millisecond - var mu sync.Mutex - var said []string - doneKeysSay = func(s string) { mu.Lock(); said = append(said, s); mu.Unlock() } - calls := 0 - doneKeyTakeOnce = func(context.Context, *stores, string) error { - calls++ - switch { - case calls <= 2: - return fmt.Errorf("%w: the artifact store could not hold mesh-host's artifact", errKeyWaits) - case calls == 3: - return fmt.Errorf("%w: mesh-host's key cannot be read", errKeyWaits) - } - return nil - } - done := make(chan struct{}) - go func() { takeDoneKey(context.Background(), nil, "mesh-host"); close(done) }() - select { - case <-done: - case <-time.After(5 * time.Second): - t.Fatal("a waiting take-in was not tried again until it was taken in") - } - if calls != 4 { - t.Fatalf("tried %d times, want 4", calls) - } - mu.Lock() - defer mu.Unlock() - waits := 0 - for _, s := range said { - if strings.Contains(s, "mesh-host waits") { - waits++ - } - } - if waits != 2 { - t.Fatalf("said %q: want each of the two reasons once", said) - } -} - -// A wake while a take-in of the module already follows it adds no second one: the follower looks again itself. -func TestAWakeWhileATakeInFollowsAddsNone(t *testing.T) { - was, wasRetry, wasSay := doneKeyTakeOnce, doneKeyRetry, doneKeysSay - t.Cleanup(func() { doneKeyTakeOnce, doneKeyRetry, doneKeysSay = was, wasRetry, wasSay }) - doneKeyRetry = 20 * time.Millisecond - doneKeysSay = func(string) {} - var mu sync.Mutex - calls := 0 - release := make(chan struct{}) - doneKeyTakeOnce = func(context.Context, *stores, string) error { - mu.Lock() - calls++ - mu.Unlock() - select { - case <-release: - return nil - default: - return fmt.Errorf("%w: the key says building", errKeyWaits) - } - } - follower := make(chan struct{}) - go func() { takeDoneKey(context.Background(), nil, "board"); close(follower) }() - time.Sleep(5 * time.Millisecond) - returned := make(chan struct{}) - go func() { takeDoneKey(context.Background(), nil, "board"); close(returned) }() - select { - case <-returned: - case <-time.After(time.Second): - t.Fatal("a second wake waited for the follower") - } - close(release) - <-follower -} - -// A wake while a take-in of the module follows it is not lost: the follower goes round once more after it succeeds, so -// a newer build's key written meanwhile is taken in (the review of issue 1945). -func TestAWakeDuringATakeInIsTakenAfterIt(t *testing.T) { - was, wasRetry, wasSay := doneKeyTakeOnce, doneKeyRetry, doneKeysSay - t.Cleanup(func() { doneKeyTakeOnce, doneKeyRetry, doneKeysSay = was, wasRetry, wasSay }) - doneKeyRetry = 20 * time.Millisecond - doneKeysSay = func(string) {} - var mu sync.Mutex - calls := 0 - inFirst := make(chan struct{}) - release := make(chan struct{}) - doneKeyTakeOnce = func(context.Context, *stores, string) error { - mu.Lock() - calls++ - n := calls - mu.Unlock() - if n == 1 { - close(inFirst) - <-release - } - return nil - } - follower := make(chan struct{}) - go func() { takeDoneKey(context.Background(), nil, "mesh-host"); close(follower) }() - <-inFirst - takeDoneKey(context.Background(), nil, "mesh-host") // the newer key's wake, while the first runs - close(release) - <-follower - mu.Lock() - defer mu.Unlock() - if calls != 2 { - t.Fatalf("taken %d times: the wake during the take-in was lost", calls) - } -} - -// A wake that comes after the follower's last look but before it lets go of the module is not lost: the follower looks -// at the mark again once it has let go, and takes the module once more (the review of issue 1945). -func TestAWakeJustBeforeTheFollowerLetsGoIsTaken(t *testing.T) { - was, wasRetry, wasSay, wasHook := doneKeyTakeOnce, doneKeyRetry, doneKeysSay, doneKeysBeforeLetGo - t.Cleanup(func() { - doneKeyTakeOnce, doneKeyRetry, doneKeysSay, doneKeysBeforeLetGo = was, wasRetry, wasSay, wasHook - }) - doneKeyRetry = 10 * time.Millisecond - doneKeysSay = func(string) {} - var mu sync.Mutex - calls := 0 - doneKeyTakeOnce = func(context.Context, *stores, string) error { - mu.Lock() - calls++ - mu.Unlock() - return nil - } - woke := false - doneKeysBeforeLetGo = func(module string) { - if !woke { - woke = true - takeDoneKey(context.Background(), nil, module) // fails its TryLock and marks the module - } - } - takeDoneKey(context.Background(), nil, "mesh-host") - mu.Lock() - defer mu.Unlock() - if calls != 2 { - t.Fatalf("taken %d times: the wake just before the follower let go was lost", calls) - } -} - -// A wake during the start's take-in of a module is not lost: the start pass reads the mark once it lets go of the module, -// and hands the module to takeDoneKey, which takes it once more (the review of issue 1945: the third dropped wake of its -// kind). -func TestAWakeDuringTheStartsTakeInIsTaken(t *testing.T) { - was, wasRetry, wasSay := doneKeyTakeOnce, doneKeyRetry, doneKeysSay - t.Cleanup(func() { doneKeyTakeOnce, doneKeyRetry, doneKeysSay = was, wasRetry, wasSay }) - doneKeyRetry = 10 * time.Millisecond - doneKeysSay = func(string) {} - var mu sync.Mutex - calls := 0 - doneKeyTakeOnce = func(context.Context, *stores, string) error { - mu.Lock() - calls++ - n := calls - mu.Unlock() - if n == 1 { - takeDoneKey(context.Background(), nil, "mesh-host") // the newer key's wake, during the start's take-in - } - return nil - } - takeDoneKeyAtStart(context.Background(), nil, "mesh-host") - background.Wait() - mu.Lock() - defer mu.Unlock() - if calls != 2 { - t.Fatalf("taken %d times: the wake during the start's take-in was lost", calls) - } -} diff --git a/cmd/mesh-controller/composer.go b/cmd/mesh-controller/composer.go index c6637eea..025321cb 100644 --- a/cmd/mesh-controller/composer.go +++ b/cmd/mesh-controller/composer.go @@ -29,7 +29,13 @@ type composer struct { type waiter struct { nodes []string grants bool - done chan error + done chan passResult +} + +// passResult is a waiter's answer: the error of each of its nodes that failed, and of the grants when it asked for them. +type passResult struct { + failed map[string]error + grants error } func newComposer(compose func(context.Context, []string) map[string]error, grants func(context.Context) error) *composer { @@ -39,7 +45,23 @@ func newComposer(compose func(context.Context, []string) map[string]error, grant // compose asks for the nodes to be composed, and the grants written after when grants is true, and waits for the pass // that does it, with the errors of its own nodes alone. func (c *composer) compose(ctx context.Context, nodes []string, grants bool) error { - done := make(chan error, 1) + failed, grantsErr, err := c.composeEach(ctx, nodes, grants) + if err != nil { + return err + } + var errs []error + for _, n := range nodes { + if e := failed[n]; e != nil { + errs = append(errs, e) + } + } + return errors.Join(append(errs, grantsErr)...) +} + +// composeEach is compose answering each node's error apart, and the grants' (ADR 0544 §5: a batch of take-ins asks one +// pass and settles each module by its own nodes); err is ctx's when it ends first. +func (c *composer) composeEach(ctx context.Context, nodes []string, grants bool) (map[string]error, error, error) { + done := make(chan passResult, 1) c.mu.Lock() for _, n := range nodes { c.pending[n] = true @@ -52,10 +74,10 @@ func (c *composer) compose(ctx context.Context, nodes []string, grants bool) err default: } select { - case err := <-done: - return err + case r := <-done: + return r.failed, r.grants, nil case <-ctx.Done(): - return ctx.Err() + return nil, nil, ctx.Err() } } @@ -90,16 +112,16 @@ func (c *composer) run(ctx context.Context) { grantsErr = c.writeGrants(ctx) } for _, w := range waiters { - var errs []error + r := passResult{failed: map[string]error{}} for _, n := range w.nodes { if err := failed[n]; err != nil { - errs = append(errs, err) + r.failed[n] = err } } - if w.grants && grantsErr != nil { - errs = append(errs, grantsErr) + if w.grants { + r.grants = grantsErr } - w.done <- errors.Join(errs...) + w.done <- r } } } diff --git a/cmd/mesh-controller/key_takein.go b/cmd/mesh-controller/key_takein.go index 3f00bbc2..abb3a88b 100644 --- a/cmd/mesh-controller/key_takein.go +++ b/cmd/mesh-controller/key_takein.go @@ -39,6 +39,11 @@ import ( // errKeyWaits says a take-in from a key waits: for the builder, a reference, the store or the forge. Never a failure. var errKeyWaits = errors.New("waits") +// errWaitsForAWrite is a take-in that waits for the builder or the indexer to write again (the key not done, its +// artifacts not referenced, no fingerprint of the new form): that write wakes it, so nothing looks again before it +// (ADR 0544 §4). +var errWaitsForAWrite = fmt.Errorf("%w for the builder's or the indexer's next write", errKeyWaits) + // errKeyFailed says the module's key failed: the builder's own words follow. var errKeyFailed = errors.New("the builder's key failed") @@ -298,12 +303,15 @@ func buildOfKey(ctx context.Context, keys keyReads, t keyTake, kc keyContext) ( waits := func(format string, a ...any) (inventory.Build, catalogue.Manifest, error) { return inventory.Build{}, catalogue.Manifest{}, fmt.Errorf("%w: "+format, append([]any{errKeyWaits}, a...)...) } + waitsForAWrite := func(format string, a ...any) (inventory.Build, catalogue.Manifest, error) { + return inventory.Build{}, catalogue.Manifest{}, fmt.Errorf("%w: "+format, append([]any{errWaitsForAWrite}, a...)...) + } if t.Trunk.FP2 == "" && t.Hand == nil { why := t.Trunk.None2 if why == "" { why = "the indexer wrote no fingerprint of the new form" } - return waits("%s at %.8s has no fingerprint to take a build by: %s", t.Module, t.Trunk.Commit, why) + return waitsForAWrite("%s at %.8s has no fingerprint to take a build by: %s", t.Module, t.Trunk.Commit, why) } k, err := keys.Build(ctx, t.Module, t.Trunk.FP2) if t.Hand != nil { @@ -319,9 +327,9 @@ func buildOfKey(ctx context.Context, keys keyReads, t keyTake, kc keyContext) ( case k.Failed(): return inventory.Build{}, catalogue.Manifest{}, fmt.Errorf("%w: %s", errKeyFailed, k.Says()) case !k.Done(): - return waits("%s", k.Says()) + return waitsForAWrite("%s", k.Says()) case len(k.Unreferenced()) > 0: - return waits("%s", k.Says()) + return waitsForAWrite("%s", k.Says()) } src, err := keys.Manifest(ctx, k.Build.Manifest) if err != nil { diff --git a/cmd/mesh-controller/modules.go b/cmd/mesh-controller/modules.go index 12b84e9d..c690dd34 100644 --- a/cmd/mesh-controller/modules.go +++ b/cmd/mesh-controller/modules.go @@ -965,6 +965,26 @@ func processPrincipalFor(m catalogue.Manifest, processID string) (broker.Process return broker.ProcessPrincipal{}, false } +// mintOnlyIfAbsent marks a context whose issue mints only accounts no row names yet (onlyIfAbsent). +type mintOnlyIfAbsent struct{} + +// onlyIfAbsent is ctx for an issue that must never replace a minted account: the assign act's and the assignments' +// watch's, which may run at once for one account (novox/hq ADR 0543, the review of issue 1945). +func onlyIfAbsent(ctx context.Context) context.Context { + return context.WithValue(ctx, mintOnlyIfAbsent{}, true) +} + +// mintBusUser mints u's password. By default it replaces whatever stands (a rotation: `module issue`); under +// onlyIfAbsent it mints only when no row names u, and won false says another caller minted it first and seals it, +// so this one seals nothing. +func mintBusUser(ctx context.Context, inv *inventory.Inventory, u inventory.BusUser) (string, bool, error) { + if only, _ := ctx.Value(mintOnlyIfAbsent{}).(bool); only { + return inv.MintBusPasswordIfAbsent(ctx, u) + } + password, err := inv.MintBusPassword(ctx, u) + return password, err == nil, err +} + // issueOnTheNewBus gives an assigned module its credential on the bus being built. // // **Three things differ from a management call, and each is the point of the move.** The credential @@ -990,14 +1010,16 @@ func issueOnTheNewBus(ctx context.Context, inv *inventory.Inventory, m catalogue } if _, reads := m.OwnSecrets["broker"]; reads { user := broker.Principal{Kind: broker.KindModule, Node: node, Module: m.Module}.Username() - password, err := inv.MintBusPassword(ctx, inventory.BusUser{ + password, won, err := mintBusUser(ctx, inv, inventory.BusUser{ Username: user, Kind: busKindOf(m.Module), Node: node, Module: m.Module, }) if err != nil { return err } - if err := issueWith(ctx, inv, m, node, busAddress, known, reachable, user, password); err != nil { - return err + if won { + if err := issueWith(ctx, inv, m, node, busAddress, known, reachable, user, password); err != nil { + return err + } } } // Every process with a bus declaration gets an account of its own (novox/hq ADR 0436 §1, §2), minted and @@ -1021,11 +1043,11 @@ func issueOnTheNewBus(ctx context.Context, inv *inventory.Inventory, m catalogue func issueProcess(ctx context.Context, inv *inventory.Inventory, m catalogue.Manifest, pr broker.ProcessPrincipal, node, busAddress string, known broker.Broker, reachable string) error { user := broker.Principal{Kind: broker.KindProcess, Node: node, Module: m.Module, Process: pr.Process}.Username() - password, err := inv.MintBusPassword(ctx, inventory.BusUser{ + password, won, err := mintBusUser(ctx, inv, inventory.BusUser{ Username: user, Kind: inventory.BusProcess, Node: node, Module: m.Module, }) - if err != nil { - return err + if err != nil || !won { + return err // lost: the caller that minted it seals it } name := processSecretName(pr.Process) held, err := heldProcessAccount(known, reachable, node, m.Module, user, password) @@ -1091,11 +1113,11 @@ func runsTheProvisionerClient(m catalogue.Manifest) bool { func issueProvisionerClient(ctx context.Context, inv *inventory.Inventory, m catalogue.Manifest, node string, known broker.Broker, reachable string) error { user := broker.Principal{Kind: broker.KindProvisioner, Node: node, Module: m.Module}.Username() - password, err := inv.MintBusPassword(ctx, inventory.BusUser{ + password, won, err := mintBusUser(ctx, inv, inventory.BusUser{ Username: user, Kind: inventory.BusProvisioner, Node: node, Module: m.Module, }) - if err != nil { - return err + if err != nil || !won { + return err // lost: the caller that minted it seals it } held, err := json.Marshal(struct { URL string `json:"url"` 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/cmd/mesh-controller/plain_words.go b/cmd/mesh-controller/plain_words.go index 2b1fb1b0..c1ebdf54 100644 --- a/cmd/mesh-controller/plain_words.go +++ b/cmd/mesh-controller/plain_words.go @@ -70,6 +70,20 @@ var plainWordings = map[string]func(conditions.Observation) words{ "none can be given it yet. It is tried again every half minute; the details say what fails.", m), Resolved: "the new build of " + m + " can be delivered"} }), + // The take-in of done keys whose backlog stood still (take_in_consumer.go, novox/hq ADR 0544 §6). + kindTakeInStalled: worded(func(o conditions.Observation) words { + return words{Headline: "New builds are not being prepared for the machines", + Explanation: "The mesh prepares each new build for the machines as it is made, and has prepared none for " + + "five minutes while builds wait. Until it moves, no new build reaches a machine; the details say how many wait.", + Resolved: "new builds are being prepared again"} + }), + // A bucket of done keys whose take-in consumer could not be made (take_in_consumer.go, novox/hq ADR 0544 §6). + kindTakeInNotConsumed: worded(func(o conditions.Observation) words { + return words{Headline: "Some new builds are not being prepared", + Explanation: "The mesh could not start reading one of the two places new builds are announced in, so the " + + "builds announced there wait until the controller starts again. The details say which and why.", + Resolved: "the mesh reads where new builds are announced again"} + }), // Machines and their node-engine. "silent": worded(func(o conditions.Observation) words { m := machineOr(o, "a machine") diff --git a/cmd/mesh-controller/process_accounts.go b/cmd/mesh-controller/process_accounts.go index 43bbc84d..2032697f 100644 --- a/cmd/mesh-controller/process_accounts.go +++ b/cmd/mesh-controller/process_accounts.go @@ -72,6 +72,8 @@ func issueMissingProcessAccounts(ctx context.Context, inv *inventory.Inventory, if len(m.ProcessBuses()) == 0 || mayIssue(m) != nil { return "", nil } + // Only the missing: never a second password over one another caller minted between this look and the mint. + ctx = onlyIfAbsent(ctx) minted, err := inv.BusMinted(ctx) if err != nil { return "", err diff --git a/cmd/mesh-controller/process_issue_test.go b/cmd/mesh-controller/process_issue_test.go index 1d49ad62..1e128b89 100644 --- a/cmd/mesh-controller/process_issue_test.go +++ b/cmd/mesh-controller/process_issue_test.go @@ -1,12 +1,9 @@ package main import ( - "encoding/json" - "strings" "testing" "git.novox.be/novox/mesh-sdk/go/catalogue" - "github.com/novox/mesh-controller/internal/broker" ) // mailSenderWithProcess is mesh-mail's real manifest shape (novox/hq ADR 0436 §7, §8, the lead's fixture): @@ -25,61 +22,6 @@ func mailSenderWithProcess() catalogue.Manifest { return m } -// review round 1, finding 1 (1181): a module declaring no `broker` of its own at all — mesh-mail's real -// shape, tools served by the tool runner — still passes mayIssue on its process's bus declaration alone. -func TestAModuleWithNoBrokerAndOnlyAProcessMayBeIssued(t *testing.T) { - if err := mayIssue(mailSenderWithProcess()); err != nil { - t.Fatalf("a module with no broker of its own, whose process declares bus, was refused: %v", err) - } -} - -// 1181/AC1: a process with a bus declaration and no own secret broker- cannot be issued — the -// same refusal mayIssue already gives a module with no `broker` (novox/hq ADR 0436 §2). -func TestAProcessWithNoBrokerSecretCannotBeIssued(t *testing.T) { - m := mailSenderWithProcess() - delete(m.OwnSecrets, "broker-send") - err := mayIssue(m) - if err == nil { - t.Fatal("a process with no broker- own secret was issued an account") - } - for _, want := range []string{"mesh-mail", "send", "broker-send", "own-secrets"} { - if !strings.Contains(err.Error(), want) { - t.Errorf("the refusal does not say %q: %v", want, err) - } - } - if err := mayIssue(mailSenderWithProcess()); err != nil { - t.Errorf("a process declaring its own broker- secret was refused: %v", err) - } -} - -// 1181/AC1: the account held for a process is the provisioner client's form — {url, fingerprint, node, -// module, user, password} — named for the process, never the module's own account. -func TestAProcesssHeldAccountIsTheProvisionersForm(t *testing.T) { - user := broker.Principal{Kind: broker.KindProcess, Node: "novox", Module: "mesh-mail", Process: "send"}.Username() - if user != "novox.mesh-mail.send" { - t.Fatalf("the process's username: %q", user) - } - held, err := heldProcessAccount(broker.Broker{Fingerprint: "abc"}, "novox.example:4222", "novox", "mesh-mail", - user, "a-password") - if err != nil { - t.Fatal(err) - } - var got map[string]any - if err := json.Unmarshal(held, &got); err != nil { - t.Fatal(err) - } - want := map[string]any{"url": "nats://novox.example:4222", "fingerprint": "abc", "node": "novox", - "module": "mesh-mail", "user": "novox.mesh-mail.send", "password": "a-password"} - for k, v := range want { - if got[k] != v { - t.Errorf("%s: %v, want %v", k, got[k], v) - } - } - if len(got) != len(want) { - t.Errorf("the held account carries %v; want exactly %v (never claims — a process holds no seat)", got, want) - } -} - // 1181/AC1: composed, the process's own secret belongs to its own account at 0600 (novox/hq ADR 0436 §1, §2; // the owner mechanism is issue 1170's, proven generically there — this is mesh-mail's own shape). func TestAProcessesOwnSecretIsOwnedByItsOwnAccount(t *testing.T) { diff --git a/cmd/mesh-controller/readable.go b/cmd/mesh-controller/readable.go index c71d9b21..3b6e53c1 100644 --- a/cmd/mesh-controller/readable.go +++ b/cmd/mesh-controller/readable.go @@ -111,6 +111,9 @@ type meshStatus struct { // why they could not be read. Pending []pendingStatus `json:"pending,omitempty"` PendingUnread string `json:"pendingUnread,omitempty"` + // TakeIn is the take-in's backlog of done keys: keys not delivered yet and keys being taken in (novox/hq ADR 0544 + // §6). Absent where it cannot be read. + TakeIn *takeInBacklog `json:"take-in,omitempty"` } // pendingStatus is one pending assignment as status says it. @@ -266,6 +269,7 @@ func statusAsJSON(asked answers) ([]byte, error) { out.Failing = providerStandings(asked.conditions) out.Overflowing = asked.overflowing out.PendingUnread = asked.pendingUnread + out.TakeIn = asked.takeIn for _, p := range asked.pending { out.Pending = append(out.Pending, pendingStatus{Node: p.Node, Module: p.Module, State: p.State, Build: p.Build, Repository: p.Repository, Path: p.Path, Since: p.Since, Settled: p.Settled, Note: p.Note}) diff --git a/cmd/mesh-controller/status.go b/cmd/mesh-controller/status.go index d2e940f2..a9f86272 100644 --- a/cmd/mesh-controller/status.go +++ b/cmd/mesh-controller/status.go @@ -151,6 +151,10 @@ func printStatus(asked answers) error { } printPending(asked.pending, asked.pendingUnread) + if b := asked.takeIn; b != nil && b.Total() > 0 { + fmt.Printf("the take-in of done keys has %d key(s) not taken in yet: %d waiting to be delivered, %d being taken in\n\n", + b.Total(), b.Pending, b.Taking) + } if len(behind) > 0 { var names []string @@ -471,6 +475,13 @@ func theThreeQuestions(ctx context.Context, open *stores) (answers, error) { out.pendingUnread = err.Error() err = nil } + // And the take-in's backlog of done keys, read from its consumers on the bus (novox/hq ADR 0544 §6): where there is + // a bus; a stall is a condition of its own, so this is said and judges nothing. + if _, onBus := broker.BusAddress(); onBus == nil { + if b, berr := readTakeInBacklog(ctx); berr == nil { + out.takeIn = &b + } + } // And how many repairs were done by hand this week (novox/hq to-be 45 §7) — where there is a bus // to read the log from; a process with none has no log to count. if _, onBus := broker.BusAddress(); onBus == nil { diff --git a/cmd/mesh-controller/take_in_consumer.go b/cmd/mesh-controller/take_in_consumer.go new file mode 100644 index 00000000..8ae7f4b7 --- /dev/null +++ b/cmd/mesh-controller/take_in_consumer.go @@ -0,0 +1,452 @@ +package main + +import ( + "context" + "errors" + "fmt" + "strings" + "sync" + "sync/atomic" + "time" + + sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" + "github.com/nats-io/nats.go/jetstream" + "github.com/novox/mesh-controller/internal/broker" + "github.com/novox/mesh-controller/internal/conditions" +) + +// The take-in's two durable pull consumers (novox/hq ADR 0544; research 072, milestones 2 and 4). On the indexer's +// modules bucket the consumer delivers each module's last trunk value when it is made, then each write: the start's look +// of a controller whose consumer is new, and nothing at a restart but what the controller before it left +// unacknowledged. On the builder's builds bucket it delivers each write from when it is made. A key is acknowledged once +// its module is taken in, has nothing new, or waits for a write that will be delivered; any other wait or failure is +// refused with a delay that grows with each refusal, and the bus delivers it again. +// +// **The bound on the store is the one take-in loop**, a batch of at most startTakeIns keys at a time (issues 1945 and +// 1277: a burst of take-ins once filled the store until it refused every connection). The consumers' own limit on +// unacknowledged keys is far above it, because a refused key stays unacknowledged until its delay ends: with a limit of +// a batch, a handful of keys refused for long (a manifest word this controller does not read yet) would stop a consumer +// for good, the build that reads the word among what it holds back (the review of ADR 0544). + +// startTakeIns is the most a batch takes: the take-ins that may wait on the store at once. +const startTakeIns = 16 + +// takeInPending is how many keys each consumer may leave unacknowledged: delivered and being taken in, or refused and +// waiting for their delay. Far above a batch, so refused keys never stop the others. +const takeInPending = 1000 + +// takeInBuffered is how many delivered keys each consumer's reader holds ahead of the loop. +const takeInBuffered = startTakeIns / 2 + +// takeInAckWait is how long a delivered key may stay unacknowledged before the bus delivers it again: longer than a batch +// takes, its two composer passes among it. +const takeInAckWait = 10 * time.Minute + +// takeInGather is how long a batch waits for more keys after its first, so a burst is one batch. +var takeInGather = 200 * time.Millisecond + +// takeInStall is how long a backlog may stand without a key delivered before it is a condition (ADR 0544 §6). +var takeInStall = 5 * time.Minute + +// takeInMsg is one delivered key, as the take-in uses it. +type takeInMsg struct { + delivered + ack func() + nak func(delay time.Duration) +} + +// takeInKeyOf is the module a delivered key names, false for a delete or a purge, which takes nothing in. +func takeInKeyOf(bucket string, m jetstream.Msg) (delivered, bool) { + if op := m.Headers().Get("KV-Operation"); op == "DEL" || op == "PURGE" { + return delivered{}, false + } + key := strings.TrimPrefix(m.Subject(), "$KV."+bucket+".") + module, _, _ := strings.Cut(key, ".") + if module == "" { + return delivered{}, false + } + var deliveries uint64 = 1 + if meta, err := m.Metadata(); err == nil { + deliveries = meta.NumDelivered + } + return delivered{module: module, again: deliveries > 1, deliveries: deliveries}, true +} + +// takeInConsumer is one bucket's consumer of the take-in. +type takeInConsumer struct { + bucket string + consumer jetstream.Consumer +} + +// takeInConfig is the take-in's consumer on bucket's stream. +func takeInConfig(bucket string) jetstream.ConsumerConfig { + deliver := jetstream.DeliverNewPolicy + if bucket == sdkdelivery.ModulesBucket { + deliver = jetstream.DeliverLastPerSubjectPolicy // the start's look: every module's trunk value once + } + return jetstream.ConsumerConfig{ + Durable: broker.TakeInConsumer, FilterSubject: "$KV." + bucket + ".>", DeliverPolicy: deliver, + AckPolicy: jetstream.AckExplicitPolicy, AckWait: takeInAckWait, MaxAckPending: takeInPending, + Description: "the controller's take-in of done keys (novox/hq ADR 0544)", + } +} + +// takeInConsumers makes or finds the take-in's consumer on each bucket's stream; one that cannot be made is said, and +// the other is still read. +func takeInConsumers(ctx context.Context, js jetstream.JetStream) []takeInConsumer { + var out []takeInConsumer + for _, bucket := range broker.TakeInBuckets { + c, err := js.CreateOrUpdateConsumer(ctx, "KV_"+bucket, takeInConfig(bucket)) + o := takeInNotConsumedObservation(bucket, err) + if err != nil { + // Loud: a consumer whose deliver policy differs cannot be updated, and nothing of its bucket is taken in. + doneKeysSay("done keys: " + o.Summary) + if keeper := conditionsFrom; keeper != nil { + _, _ = keeper.Observe(ctx, o) + } + continue + } + // Made: a condition a controller before this one raised for it is cleared (a restart that works clears it). + if keeper := conditionsFrom; keeper != nil { + _, _ = keeper.Clear(ctx, o.Key(), "the take-in's consumer on "+bucket+" is made") + } + out = append(out, takeInConsumer{bucket: bucket, consumer: c}) + } + return out +} + +// watchDoneKeys takes in the done keys the two consumers deliver, batch after batch, until ctx ends, and watches their +// backlog. A consumer that cannot be made is said: its bucket is then not taken in until the next start. +func watchDoneKeys(ctx context.Context, open *stores) { + keys, closeKeys, err := openBuilderKeys() + if err != nil { + fmt.Printf("done keys: the builder's keys are not taken in: %v\n", err) + return + } + consumers := takeInConsumers(ctx, keys.JS) + if len(consumers) == 0 { + closeKeys() + return + } + in := make(chan takeInMsg) + // moving counts the keys handed to the loop and the keys it settled (acknowledged or refused, each as the bus took + // it): what tells a take-in that moves from one that holds work and does nothing (the stall watch, ADR 0544 §6). + moving := &takeInMoving{} + read := make([]jetstream.Consumer, 0, len(consumers)) + // The connection closes only once the readers and the loop have ended, so the refusals they send at the stop reach + // the bus (the third review of ADR 0544). + var users sync.WaitGroup + for _, c := range consumers { + bucket := c.bucket + read = append(read, c.consumer) + it, err := c.consumer.Messages(jetstream.PullMaxMessages(takeInBuffered)) + if err != nil { + fmt.Printf("done keys: the take-in's consumer on %s is not read: %v\n", bucket, err) + continue + } + goServing(ctx, func() { + <-ctx.Done() + it.Drain() // the keys it holds are handed out and refused at once below, not lost to their ack wait + }) + users.Add(1) + goServing(ctx, func() { + defer users.Done() + for { + m, err := it.Next() + if err != nil { + if ctx.Err() == nil && !errors.Is(err, jetstream.ErrMsgIteratorClosed) { + fmt.Printf("done keys: the take-in's consumer on %s stopped: %v\n", bucket, err) + } + return + } + d, ok := takeInKeyOf(bucket, m) + if !ok { + _ = m.Ack() + continue + } + // Settled is the loop's attempt, whatever the bus answers: an acknowledgement it refuses (at a first start, + // until the bus node is composed) is delivered again and settled again, and never counts as work held + // (the third review of ADR 0544). Acknowledged synchronously and past the stop, so a refusal is said. + msg := takeInMsg{delivered: d, ack: func() { + acking, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second) + defer cancel() + if err := m.DoubleAck(acking); err != nil { + doneKeysSay(fmt.Sprintf("done keys: the bus did not take the acknowledgement of %s: %v", d.module, err)) + } + moving.settled.Add(1) + }, + nak: func(delay time.Duration) { + _ = m.NakWithDelay(delay) + moving.settled.Add(1) + }} + if ctx.Err() != nil { + _ = m.Nak() + continue // drained: each key it held, refused at once + } + select { + case in <- msg: + moving.handed.Add(1) + case <-ctx.Done(): + _ = m.Nak() // delivered again at once to the controller after this one, not after the ack wait + continue + } + } + }) + } + backlog := func(ctx context.Context) (takeInBacklog, error) { return backlogOf(ctx, read) } + steps := liveTakeInSteps(open, keys, keys.JS) + users.Add(1) + goServing(ctx, func() { + defer users.Done() + takeInLoop(ctx, in, steps, backlog) + }) + goServing(ctx, func() { watchTakeInStall(ctx, backlog, moving) }) + goServing(ctx, func() { + <-ctx.Done() + users.Wait() + closeKeys() + }) +} + +// takeInLoop gathers delivered keys into batches and takes each batch in, until ctx ends. +func takeInLoop(ctx context.Context, in <-chan takeInMsg, steps takeInSteps, + backlog func(context.Context) (takeInBacklog, error)) { + t := &takeInState{memo: map[string]string{}, raised: map[string]bool{}, saidWait: map[string]string{}} + for { + var batch []takeInMsg + select { + case <-ctx.Done(): + return + case m := <-in: + batch = append(batch, m) + } + gather := time.NewTimer(takeInGather) + more: + for len(batch) < startTakeIns { + select { + case m := <-in: + batch = append(batch, m) + case <-gather.C: + break more + case <-ctx.Done(): + gather.Stop() + for _, m := range batch { + m.nak(0) + } + return + } + } + gather.Stop() + said := t.take(ctx, steps, batch) + if b, err := backlog(ctx); err == nil { + said += fmt.Sprintf("; backlog %d", b.Total()) + } + doneKeysSay("done keys: " + said) + } +} + +// takeInState is what the take-in keeps between batches in this process: the trunk each module was last taken in at, +// the modules whose condition it raised, and the wait last said of each. +type takeInState struct { + memo map[string]string + raised map[string]bool + saidWait map[string]string +} + +// take takes one batch in and acknowledges or refuses each key, raising and clearing the modules' conditions. It answers +// the batch in a line. +func (t *takeInState) take(ctx context.Context, steps takeInSteps, batch []takeInMsg) string { + keys := make([]delivered, len(batch)) + for i, m := range batch { + keys[i] = m.delivered + } + settled, passes := steps.batch(ctx, keys, t.memo) + verdicts := make(map[string]verdict, len(settled)) + var taken, nothing, waiting, failed int + for _, r := range settled { + verdicts[r.module] = r.verdict + o := doneKeyObservation(r.module) + switch { + case r.err != nil: + failed++ + doneKeysSay(fmt.Sprintf("done keys: %s: %v", r.module, r.err)) + if keeper := conditionsFrom; keeper != nil && !t.raised[r.module] && ctx.Err() == nil && + !errors.Is(r.err, context.Canceled) { + o.Summary += r.err.Error() + if _, err := keeper.Observe(ctx, o); err == nil { + t.raised[r.module] = true + } + } + case r.waits != "": + waiting++ + if t.saidWait[r.module] != r.waits { + doneKeysSay(fmt.Sprintf("done keys: %s %s", r.module, r.waits)) + t.saidWait[r.module] = r.waits + } + case r.taken: + taken++ + delete(t.saidWait, r.module) + if r.said != "" { + doneKeysSay("done keys: " + r.said) + } + // Cleared on every take-in, whoever raised it: a controller before a restart, whose key was delivered again. + if keeper := conditionsFrom; keeper != nil { + if _, err := keeper.Clear(ctx, o.Key(), "composed and granted"); err == nil { + delete(t.raised, r.module) + } + } + case r.nothingNew: + nothing++ + } + } + for _, m := range batch { + if verdicts[m.module] == refused { + if ctx.Err() != nil { + m.nak(0) // cut by the stop: the next controller takes it at once + } else { + m.nak(refusalDelay(m.deliveries)) + } + } else { + m.ack() + } + } + return fmt.Sprintf("%d key(s) of %d module(s): %d taken in, %d with nothing new, %d waiting, %d failed; %d composer pass(es)", + len(batch), len(settled), taken, nothing, waiting, failed, passes) +} + +// takeInBacklog is the take-in's backlog on the bus: keys not delivered yet, and keys delivered and not acknowledged +// (being taken in, or refused and waiting for their delay). +type takeInBacklog struct { + Pending uint64 `json:"pending"` + Taking uint64 `json:"taking"` +} + +// Total is every key the take-in has not acknowledged. +func (b takeInBacklog) Total() uint64 { return b.Pending + b.Taking } + +// backlogOf reads the consumers' own counts from the bus. +func backlogOf(ctx context.Context, consumers []jetstream.Consumer) (takeInBacklog, error) { + var b takeInBacklog + for _, c := range consumers { + info, err := c.Info(ctx) + if err != nil { + return takeInBacklog{}, err + } + b.Pending += info.NumPending + b.Taking += uint64(info.NumAckPending) + } + return b, nil +} + +// readTakeInBacklog is the take-in's backlog for a reader in any process (`status`): the consumers asked by name. +func readTakeInBacklog(ctx context.Context) (takeInBacklog, error) { + keys, closeKeys, err := openBuilderKeys() + if err != nil { + return takeInBacklog{}, err + } + defer closeKeys() + var consumers []jetstream.Consumer + for _, bucket := range broker.TakeInBuckets { + c, err := keys.JS.Consumer(ctx, "KV_"+bucket, broker.TakeInConsumer) + if err != nil { + return takeInBacklog{}, fmt.Errorf("the take-in's consumer on %s: %w", bucket, err) + } + consumers = append(consumers, c) + } + return backlogOf(ctx, consumers) +} + +// kindTakeInStalled is the condition of a take-in that holds work and settles none (ADR 0544 §6); kindTakeInNotConsumed +// that of a bucket whose consumer could not be made. +const ( + kindTakeInStalled = "take-in-stalled" + kindTakeInNotConsumed = "take-in-not-consumed" +) + +func takeInStalledObservation(b takeInBacklog, since time.Time) conditions.Observation { + return conditions.Observation{Scope: conditions.ScopeCore, ID: "take-in", Token: kindTakeInStalled, + Kind: kindTakeInStalled, Severity: conditions.Urgent, Resolver: conditions.ResolverOperator, Source: "done keys", + Summary: fmt.Sprintf("the take-in of done keys has %d key(s) waiting and has settled none since %s", + b.Total(), since.UTC().Format(time.RFC3339))} +} + +// takeInNotConsumedObservation is the condition of bucket's consumer not made, with its own id, so it is raised and +// cleared apart from a stall and from the other bucket's. +func takeInNotConsumedObservation(bucket string, err error) conditions.Observation { + summary := "the take-in's consumer on " + bucket + " is made" + if err != nil { + summary = fmt.Sprintf("the take-in's consumer on %s is not made, so its keys are not taken in until the next "+ + "start: %v", bucket, err) + } + return conditions.Observation{Scope: conditions.ScopeCore, ID: "take-in-" + bucket, Token: kindTakeInNotConsumed, + Kind: kindTakeInNotConsumed, Severity: conditions.Urgent, Resolver: conditions.ResolverOperator, + Source: "done keys", Summary: summary} +} + +// takeInMoving counts the keys handed to the take-in loop and the keys it settled. +type takeInMoving struct { + handed, settled atomic.Uint64 +} + +// stallWatch judges each look: stalled once there was work to do (keys not delivered yet, or keys handed to the loop +// and not settled) and nothing settled for takeInStall. A refused key waiting for its delay is no work of the loop's: +// with nothing else, a quiet mesh is not a stall; a loop that hangs on its batch is (the review of ADR 0544). Its first +// look always says it changed, so a condition a controller before this one raised is cleared once this one moves. +type stallWatch struct { + mu sync.Mutex + settled uint64 + movedAt time.Time + stalled bool + known bool +} + +// look takes a reading at now: the backlog, and how many keys were handed and settled; it answers whether the take-in +// stalls and whether that changed (always on the first look). +func (w *stallWatch) look(b takeInBacklog, handed, settled uint64, now time.Time) (stalled, changed bool) { + w.mu.Lock() + defer w.mu.Unlock() + work := b.Pending > 0 || handed > settled + if !work || settled != w.settled || w.movedAt.IsZero() { + w.movedAt = now + } + w.settled = settled + nowStalled := work && now.Sub(w.movedAt) >= takeInStall + changed = !w.known || nowStalled != w.stalled + w.stalled, w.known = nowStalled, true + return nowStalled, changed +} + +// watchTakeInStall reads the backlog every minute and raises take-in-stalled while the take-in holds work and settles +// none, clearing it when it moves, and at its first look when it does not stall: a watchdog's timer, as research 072 §2 +// keeps them until the health part leaves the controller. +func watchTakeInStall(ctx context.Context, backlog func(context.Context) (takeInBacklog, error), moving *takeInMoving) { + w := &stallWatch{} + tick := time.NewTicker(time.Minute) + defer tick.Stop() + for { + select { + case <-ctx.Done(): + return + case <-tick.C: + } + b, err := backlog(ctx) + if err != nil { + continue + } + stalled, changed := w.look(b, moving.handed.Load(), moving.settled.Load(), time.Now()) + keeper := conditionsFrom + if !changed || keeper == nil { + continue + } + w.mu.Lock() + since := w.movedAt + w.mu.Unlock() + o := takeInStalledObservation(b, since) + if stalled { + doneKeysSay("done keys: " + o.Summary) + _, _ = keeper.Observe(ctx, o) + } else { + _, _ = keeper.Clear(ctx, o.Key(), "the take-in moves") + } + } +} diff --git a/internal/broker/composed.go b/internal/broker/composed.go index 8856186e..b950e058 100644 --- a/internal/broker/composed.go +++ b/internal/broker/composed.go @@ -182,3 +182,11 @@ func RaiseAfterDeclarations(r Raiser) error { } return AssertMeshConsumers(r) } + +// TakeInConsumer is the durable consumer the controller takes done keys in from, on the stream of each of TakeInBuckets +// (novox/hq ADR 0544 §1). +const TakeInConsumer = "mesh-controller-take-in" + +// TakeInBuckets are the buckets the take-in consumes: the indexer's modules (each module's trunk value) and the builder's +// builds (each build's key). +var TakeInBuckets = []string{delivery.ModulesBucket, delivery.BuilderBuildsBucket} diff --git a/internal/broker/nats.go b/internal/broker/nats.go index 2add5851..734bffc1 100644 --- a/internal/broker/nats.go +++ b/internal/broker/nats.go @@ -432,6 +432,16 @@ var DegradableGrants = []DegradableGrant{ Reason: "carried only at genesis: the controller composes its own permissions into the bus module's composition, " + "and writes a composition's digest only after the bus applied it (novox/hq ADR 0539 clauses 2 and 7)", }, + { + Subject: "$JS.ACK.KV_" + delivery.ModulesBucket + "." + TakeInConsumer + ".>", + Reason: "refused: an acknowledgement refused leaves the key delivered again after its wait, and the take-in's first " + + "batch with something new composes the bus's node, which grants it (novox/hq ADR 0544 §1)", + }, + { + Subject: "$JS.ACK.KV_" + delivery.BuilderBuildsBucket + "." + TakeInConsumer + ".>", + Reason: "refused: an acknowledgement refused leaves the key delivered again after its wait, and the take-in's first " + + "batch with something new composes the bus's node, which grants it (novox/hq ADR 0544 §1)", + }, { Subject: "$KV." + GrantsBucket + ".>", Reason: "carried only at genesis: the controller composes its own permissions and pushes them to the bus at its " + @@ -707,6 +717,11 @@ func PermissionsFor(p Principal) (Permissions, error) { // ever, refused by the list it already has. sub = append(sub, ControllerFollows...) pub = append(pub, "$JS.ACK.EVENTS."+ControllerName+".>", "$JS.ACK.EVENTS."+MergesConsumerName+".>") + // And the take-in's two durable consumers on the indexer's and the builder's buckets (novox/hq ADR 0544 §1): + // acknowledged after each key is taken in, and no other consumer of those streams. + for _, bucket := range TakeInBuckets { + pub = append(pub, "$JS.ACK.KV_"+bucket+"."+TakeInConsumer+".>") + } // **Where an enrolment's answer goes**, and `allow_responses` does not cover it. That // permits one reply to the reply subject of a message the user received — and a message a diff --git a/internal/broker/testdata/composed.conf b/internal/broker/testdata/composed.conf index 0cc3eab2..7560fd8c 100644 --- a/internal/broker/testdata/composed.conf +++ b/internal/broker/testdata/composed.conf @@ -29,7 +29,7 @@ accounts { ] users = [ { user: "controller", password: "$2a$11$cccccccccccccccccccccc", permissions: { - publish: { allow: ["$JS.ACK.CONTROL.controller.>", "$JS.ACK.DEAD_LETTER_NOTICES.controller.>", "$JS.ACK.EVENTS.controller-merges.>", "$JS.ACK.EVENTS.controller.>", "$JS.API.>", "$KV.mesh-controller_asked.>", "$KV.mesh-controller_assignments.>", "$KV.mesh-controller_builds.>", "$KV.mesh-controller_calls.>", "$KV.mesh-controller_composed.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_declared-config.>", "$KV.mesh-controller_declared-packages.>", "$KV.mesh-controller_delivery-merges.>", "$KV.mesh-controller_grants.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$KV.mesh-controller_predefined.>", "$KV.mesh-controller_rollouts.>", "$KV.mesh-controller_states-announced.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "mesh.mod.mesh-controller.event.states", "mesh.node.>", "mesh.seat.mesh-controller.event.applied", "mesh.seat.mesh-controller.event.artifacts-gone", "mesh.seat.mesh-controller.event.assignment-changed", "mesh.seat.mesh-controller.event.bucket-deleted", "mesh.seat.mesh-controller.event.built-before", "mesh.seat.mesh-controller.event.checked", "mesh.seat.mesh-controller.event.condition-changed", "mesh.seat.mesh-controller.event.condition-cleared", "mesh.seat.mesh-controller.event.condition-raised", "mesh.seat.mesh-controller.event.declared-config-changed", "mesh.seat.mesh-controller.event.doctor-heartbeat", "mesh.seat.mesh-controller.event.healer-acted", "mesh.seat.mesh-controller.event.predefined-changed", "mesh.seat.mesh-controller.event.refused", "mesh.seat.mesh-controller.event.rolled-back", "mesh.seat.mesh-controller.event.secret-replaced", "mesh.seat.mesh-controller.tool.plans", "mesh.seat.mesh-controller.tool.push", "mesh.seat.mesh-controller.tool.upgrade", "mesh.seat.mesh-delivery.tool.close", "mesh.seat.mesh-delivery.tool.groups", "mesh.seat.mesh-delivery.tool.release", "mesh.seat.mesh-delivery.tool.stalled", "mesh.seat.mesh-delivery.tool.stop", "mesh.seat.mesh-delivery.tool.times", "mesh.seat.node-backup.tool.backed-up.*", "mesh.seat.node-backup.tool.now.*", "mesh.seat.node-intrusion-prevention.tool.banned.*", "mesh.seat.node-launcher.tool.secret.*", "mesh.seat.node-service-manager.tool.restart.*", "mesh.seat.operator-channel.tool.tell"] } + publish: { allow: ["$JS.ACK.CONTROL.controller.>", "$JS.ACK.DEAD_LETTER_NOTICES.controller.>", "$JS.ACK.EVENTS.controller-merges.>", "$JS.ACK.EVENTS.controller.>", "$JS.ACK.KV_mesh-delivery-builder_builds.mesh-controller-take-in.>", "$JS.ACK.KV_mesh-delivery-indexer_modules.mesh-controller-take-in.>", "$JS.API.>", "$KV.mesh-controller_asked.>", "$KV.mesh-controller_assignments.>", "$KV.mesh-controller_builds.>", "$KV.mesh-controller_calls.>", "$KV.mesh-controller_composed.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_declared-config.>", "$KV.mesh-controller_declared-packages.>", "$KV.mesh-controller_delivery-merges.>", "$KV.mesh-controller_grants.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$KV.mesh-controller_predefined.>", "$KV.mesh-controller_rollouts.>", "$KV.mesh-controller_states-announced.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "mesh.mod.mesh-controller.event.states", "mesh.node.>", "mesh.seat.mesh-controller.event.applied", "mesh.seat.mesh-controller.event.artifacts-gone", "mesh.seat.mesh-controller.event.assignment-changed", "mesh.seat.mesh-controller.event.bucket-deleted", "mesh.seat.mesh-controller.event.built-before", "mesh.seat.mesh-controller.event.checked", "mesh.seat.mesh-controller.event.condition-changed", "mesh.seat.mesh-controller.event.condition-cleared", "mesh.seat.mesh-controller.event.condition-raised", "mesh.seat.mesh-controller.event.declared-config-changed", "mesh.seat.mesh-controller.event.doctor-heartbeat", "mesh.seat.mesh-controller.event.healer-acted", "mesh.seat.mesh-controller.event.predefined-changed", "mesh.seat.mesh-controller.event.refused", "mesh.seat.mesh-controller.event.rolled-back", "mesh.seat.mesh-controller.event.secret-replaced", "mesh.seat.mesh-controller.tool.plans", "mesh.seat.mesh-controller.tool.push", "mesh.seat.mesh-controller.tool.upgrade", "mesh.seat.mesh-delivery.tool.close", "mesh.seat.mesh-delivery.tool.groups", "mesh.seat.mesh-delivery.tool.release", "mesh.seat.mesh-delivery.tool.stalled", "mesh.seat.mesh-delivery.tool.stop", "mesh.seat.mesh-delivery.tool.times", "mesh.seat.node-backup.tool.backed-up.*", "mesh.seat.node-backup.tool.now.*", "mesh.seat.node-intrusion-prevention.tool.banned.*", "mesh.seat.node-launcher.tool.secret.*", "mesh.seat.node-service-manager.tool.restart.*", "mesh.seat.operator-channel.tool.tell"] } subscribe: { allow: ["$JS.API.>", "$JS.EVENT.ADVISORY.CONSUMER.DELETED.>", "$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.>", "$SRV.INFO", "$SRV.INFO.mesh-controller", "$SRV.INFO.mesh-controller.>", "$SRV.PING", "$SRV.PING.mesh-controller", "$SRV.PING.mesh-controller.>", "$SRV.STATS", "$SRV.STATS.mesh-controller", "$SRV.STATS.mesh-controller.>", "_DELIVER.controller", "_DELIVER.controller.>", "_INBOX.controller.>", "mesh.control.>", "mesh.mod.*.event.provisioner.failing", "mesh.mod.*.event.provisioner.recovered", "mesh.mod.*.event.provisioner.retirement", "mesh.mod.gitea.event.pull.merged", "mesh.mod.gitea.event.pull.updated", "mesh.mod.mesh-catalog.event.catching-up", "mesh.mod.mesh-catalog.event.upgraded", "mesh.seat.mesh-build-machine.event.built", "mesh.seat.mesh-controller.tool.>", "mesh.seat.node-build-agent.event.built", "mesh.seat.operator-channel.event.decided.mesh-controller"] } allow_responses: { max: 1, ttl: "1m" } } } diff --git a/internal/inventory/bususers.go b/internal/inventory/bususers.go index 388e6aaf..ee5c3be1 100644 --- a/internal/inventory/bususers.go +++ b/internal/inventory/bususers.go @@ -114,6 +114,37 @@ func (i *Inventory) writeBusUser(ctx context.Context, u BusUser, hash string) er return nil } +// MintBusPasswordIfAbsent mints u's password only when no row names u yet, in one statement: of two callers that both +// found the account missing, exactly one wins, and the other is told it lost (won false, no password) and must seal +// nothing, so the hash the bus holds is always the password the node is sealed (novox/hq ADR 0543, the review of +// issue 1945: the assign act and the assignments' watch once each minted, the later hash beside the earlier seal). +func (i *Inventory) MintBusPasswordIfAbsent(ctx context.Context, u BusUser) (string, bool, error) { + if u.Username == "" || u.Kind == "" { + return "", false, errors.New("a bus user needs a username and a kind") + } + raw := make([]byte, 32) + if _, err := rand.Read(raw); err != nil { + return "", false, fmt.Errorf("cannot generate a bus password: %w", err) + } + password := base64.RawURLEncoding.EncodeToString(raw) + hash, err := hashBusPassword(password) + if err != nil { + return "", false, fmt.Errorf("cannot hash a bus password: %w", err) + } + tag, err := i.store.Pool().Exec(ctx, + `insert into bus_user (username, kind, node, module, password_hash) + values ($1, $2, $3, $4, $5) + on conflict (username) do nothing`, + u.Username, u.Kind, u.Node, u.Module, string(hash)) + if err != nil { + return "", false, fmt.Errorf("cannot record the bus user %s: %w", u.Username, err) + } + if tag.RowsAffected() == 0 { + return "", false, nil + } + return password, true, nil +} + func (i *Inventory) MintBusPassword(ctx context.Context, u BusUser) (string, error) { if u.Username == "" || u.Kind == "" { return "", errors.New("a bus user needs a username and a kind") diff --git a/internal/inventory/bususers_integration_test.go b/internal/inventory/bususers_integration_test.go index 390a8ac4..e04a2786 100644 --- a/internal/inventory/bususers_integration_test.go +++ b/internal/inventory/bususers_integration_test.go @@ -33,7 +33,7 @@ func TestABusPasswordIsReturnedOnceAndOnlyItsHashIsKept(t *testing.T) { } } -func TestMintingAgainRotatesRatherThanAddsAUser(t *testing.T) { +func TestMintingAgainRotatesAndMintingIfAbsentHasOneWinner(t *testing.T) { inv := ForTest(t) ctx := context.Background() @@ -62,6 +62,50 @@ func TestMintingAgainRotatesRatherThanAddsAUser(t *testing.T) { if bcrypt.CompareHashAndPassword([]byte(hash), []byte(first)) == nil { t.Fatal("the previous password still verifies, so a rotation revoked nothing") } + + // **Minting only if absent never rotates, and of several at once exactly one wins** (ADR 0543, the review of issue + // 1945): the assign act and the assignments' watch may both find an account missing; the winner's password is the + // one the store's hash verifies, so the one it seals is the one the bus accepts, and a loser seals nothing. + if _, won, err := inv.MintBusPasswordIfAbsent(ctx, aBusUser("shop")); err != nil || won { + t.Fatalf("a mint-if-absent over a minted account won %v (%v): it would replace a sealed password", won, err) + } + const racers = 8 + type minted struct { + password string + won bool + err error + } + results := make(chan minted, racers) + for range racers { + go func() { + p, won, err := inv.MintBusPasswordIfAbsent(ctx, aBusUser("racer")) + results <- minted{p, won, err} + }() + } + var winner string + wins := 0 + for range racers { + r := <-results + if r.err != nil { + t.Fatal(r.err) + } + if r.won { + wins++ + winner = r.password + } else if r.password != "" { + t.Fatal("a caller that lost was handed a password to seal") + } + } + if wins != 1 { + t.Fatalf("%d of %d concurrent mints won, want exactly one", wins, racers) + } + stored, _, err := inv.BusUserHash(ctx, "one.racer") + if err != nil { + t.Fatal(err) + } + if err := bcrypt.CompareHashAndPassword([]byte(stored), []byte(winner)); err != nil { + t.Fatal("the stored hash is not the winner's password: the node would be sealed a password the bus refuses") + } } func TestForgettingANodeTakesItsBusUsersWithIt(t *testing.T) { diff --git a/internal/inventory/catalogue.go b/internal/inventory/catalogue.go index c4aa198a..cebcd655 100644 --- a/internal/inventory/catalogue.go +++ b/internal/inventory/catalogue.go @@ -210,6 +210,21 @@ func (i *Inventory) hasModule(ctx context.Context, name string) (bool, error) { // This is the whole of noticing. Nothing here builds anything — it writes down that the two // halves differ, which is what makes *is this current?* answerable without building, and what // makes a module that nobody rebuilt visible rather than silent. +// RecordBuiltFrom records that a registered module stands at commit, its manifest and fingerprint unchanged: a catalogue +// merge that left its sources alone (novox/hq ADR 0544 §3). The head moves with it, as a registration moves it; the +// module's builds and manifest stay as they are. +func (i *Inventory) RecordBuiltFrom(ctx context.Context, module, commit string) error { + tag, err := i.store.Pool().Exec(ctx, + `update module set built_from = $2, source_head = $2 where name = $1`, module, commit) + if err != nil { + return fmt.Errorf("cannot record %s at %s: %w", module, commit, err) + } + if tag.RowsAffected() == 0 { + return fmt.Errorf("%w: %s", ErrNoSuchModule, module) + } + return nil +} + func (i *Inventory) SourceMoved(ctx context.Context, module, head string) error { tag, err := i.store.Pool().Exec(ctx, `update module set source_head = $2, source_seen = now() where name = $1`, module, head) 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) +}