Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
 jschoubben
|
b980ba3631 | ||
|
jschoubben
|
e092f01e54 | ||
|
jschoubben
|
b6379b9e86 |
@@ -292,22 +292,6 @@ func unassign(ctx context.Context, open *stores, node string, modules ...string)
|
||||
answer += "\n " + strings.TrimSpace(line)
|
||||
}
|
||||
}
|
||||
// Its bus accounts on the node go with it once it no longer resolves there (a module another requires keeps running
|
||||
// and keeps them), so an assignment again issues them afresh (issue 1945). The node's own provisioner account stays.
|
||||
if after.err == nil {
|
||||
for _, module := range modules {
|
||||
if slices.Contains(after.modules, module) {
|
||||
continue
|
||||
}
|
||||
forgot, err := open.inventory.ForgetBusAccountsOf(ctx, node, module, false)
|
||||
switch {
|
||||
case err != nil:
|
||||
answer += fmt.Sprintf("\n its bus accounts on %s are not forgotten: %v", node, err)
|
||||
case len(forgot) > 0:
|
||||
answer += fmt.Sprintf("\n its bus accounts on %s forgotten: %s", node, strings.Join(forgot, ", "))
|
||||
}
|
||||
}
|
||||
}
|
||||
// What it leaves behind that is irreplaceable is kept and retired, never removed (novox/hq ADR 0233).
|
||||
for _, line := range keptOnUnassign(ctx, open.inventory, node, modules) {
|
||||
answer += "\n " + line
|
||||
@@ -415,8 +399,6 @@ 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 {
|
||||
@@ -437,9 +419,9 @@ func issueOnAssign(ctx context.Context, open *stores, node, module string) strin
|
||||
// A module running the provisioner client alone is minted the client's account (issue 917).
|
||||
user = broker.Principal{Kind: broker.KindProvisioner, Node: node, Module: module}.Username()
|
||||
}
|
||||
if _, sealed, err := inv.BusAccountIssued(ctx, user); err != nil {
|
||||
if _, minted, err := inv.BusUserHash(ctx, user); err != nil {
|
||||
return ""
|
||||
} else if sealed { // a row whose credential is not sealed to the node is not issued (issue 1945)
|
||||
} else if minted {
|
||||
// Its own account stands and is kept; a process it gained is not issued yet (issue 1226).
|
||||
return processAccountsOnAssign(ctx, inv, m, node)
|
||||
}
|
||||
|
||||
@@ -1,489 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"maps"
|
||||
"slices"
|
||||
"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 `<node>.<module>` 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
|
||||
// forgetStale forgets every module's accounts on a node where the module no longer resolves, and answers a line for
|
||||
// each (issue 1945: rows of an unassignment before this release, marked sealed by migration 0108).
|
||||
forgetStale func(ctx context.Context) ([]string, error)
|
||||
// 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 forgot []string
|
||||
if len(standing) > 0 && s.forgetStale != nil {
|
||||
// Only from a reading of the assignments that holds some: an empty or failed one would forget every account.
|
||||
if !s.holds(ctx) {
|
||||
return "", errNotHolding
|
||||
}
|
||||
var err error
|
||||
if forgot, err = s.forgetStale(ctx); err != nil {
|
||||
return "", fmt.Errorf("the accounts of modules no longer on their nodes are not forgotten: %w", err)
|
||||
}
|
||||
}
|
||||
var looked []chain.Assignment
|
||||
for _, a := range standing {
|
||||
if a.Module != controllerModule {
|
||||
looked = append(looked, a)
|
||||
}
|
||||
}
|
||||
if len(looked) == 0 {
|
||||
return strings.Join(forgot, "\n"), 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(append(forgot, 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 _, own := ownAccountOf(m, node); own {
|
||||
minted, err := inv.BusMinted(ctx) // sealed accounts alone: a row with no sealed credential is not issued
|
||||
if err != nil {
|
||||
return false, "", fmt.Errorf("the issued accounts cannot be read: %w", err)
|
||||
}
|
||||
if missingAccounts(m, node, minted) {
|
||||
busAddress, err := broker.BusAddress()
|
||||
if err != nil {
|
||||
return false, "", err
|
||||
}
|
||||
// Its processes' and provisioner client's accounts too: issueOnTheNewBus issues each that is not issued.
|
||||
ctx, got := withIssued(ctx)
|
||||
if err := issueOnTheNewBus(ctx, inv, m, node, busAddress); err != nil {
|
||||
return false, "", err
|
||||
}
|
||||
if issued := got.list(); len(issued) > 0 {
|
||||
return true, "issued and sealed to " + node + ": " + strings.Join(issued, ", "), nil
|
||||
}
|
||||
return false, "", nil
|
||||
}
|
||||
return false, "its bus accounts on " + node + " stood issued and sealed", nil
|
||||
}
|
||||
line, err := issueMissingProcessAccounts(ctx, inv, m, node)
|
||||
if err != nil {
|
||||
return false, "", err
|
||||
}
|
||||
return line != "", line, nil
|
||||
}
|
||||
|
||||
// missingAccounts says some account of m on node is not issued: its own, a process's, or its provisioner client's.
|
||||
func missingAccounts(m catalogue.Manifest, node string, minted []string) bool {
|
||||
if user, own := ownAccountOf(m, node); own && !slices.Contains(minted, user) {
|
||||
return true
|
||||
}
|
||||
if runsTheProvisionerClient(m) &&
|
||||
!slices.Contains(minted, broker.Principal{Kind: broker.KindProvisioner, Node: node, Module: m.Module}.Username()) {
|
||||
return true
|
||||
}
|
||||
return len(processAccountsMissing(m, node, minted)) > 0
|
||||
}
|
||||
|
||||
// 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
|
||||
},
|
||||
forgetStale: func(ctx context.Context) ([]string, error) { return forgetStaleAccounts(ctx, open) },
|
||||
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)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// staleAccounts is, by node, each module some account of which stands on a node where the module no longer resolves:
|
||||
// the accounts of a module's own, its processes' and its provisioner client's. A node whose resolution is unknown is
|
||||
// left alone.
|
||||
func staleAccounts(users map[string]inventory.BusUser, resolved map[string]map[string]bool) map[string][]string {
|
||||
seen := map[string]map[string]bool{}
|
||||
for _, u := range users {
|
||||
switch u.Kind {
|
||||
case inventory.BusModule, inventory.BusNodeTools, inventory.BusProcess, inventory.BusProvisioner:
|
||||
default:
|
||||
continue
|
||||
}
|
||||
on, known := resolved[u.Node]
|
||||
if u.Node == "" || u.Module == "" || !known || on[u.Module] {
|
||||
continue
|
||||
}
|
||||
if seen[u.Node] == nil {
|
||||
seen[u.Node] = map[string]bool{}
|
||||
}
|
||||
seen[u.Node][u.Module] = true
|
||||
}
|
||||
out := map[string][]string{}
|
||||
for node, modules := range seen {
|
||||
out[node] = slices.Sorted(maps.Keys(modules))
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// forgetStaleAccounts forgets the accounts of every module on a node where it no longer resolves (assigned or
|
||||
// required), from each node's own resolution; a node that does not resolve is left alone.
|
||||
func forgetStaleAccounts(ctx context.Context, open *stores) ([]string, error) {
|
||||
inv := open.inventory
|
||||
users, err := inv.BusUsers(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ctx = withInputCache(ctx) // one reading of the mesh for every node resolved
|
||||
resolved := map[string]map[string]bool{}
|
||||
for _, u := range users {
|
||||
if _, done := resolved[u.Node]; done || u.Node == "" {
|
||||
continue
|
||||
}
|
||||
plan, _, err := resolveNode(ctx, open, u.Node)
|
||||
if err != nil {
|
||||
continue // unknown: nothing of this node is forgotten
|
||||
}
|
||||
on := map[string]bool{}
|
||||
for _, m := range plan.Modules {
|
||||
on[m.Module] = true
|
||||
}
|
||||
resolved[u.Node] = on
|
||||
}
|
||||
var lines []string
|
||||
stale := staleAccounts(users, resolved)
|
||||
for _, node := range slices.Sorted(maps.Keys(stale)) {
|
||||
for _, module := range stale[node] {
|
||||
forgot, err := inv.ForgetBusAccountsOf(ctx, node, module, true)
|
||||
if err != nil {
|
||||
return lines, err
|
||||
}
|
||||
if len(forgot) > 0 {
|
||||
lines = append(lines, fmt.Sprintf("%s no longer runs on %s: its bus accounts forgotten: %s", module, node,
|
||||
strings.Join(forgot, ", ")))
|
||||
}
|
||||
}
|
||||
}
|
||||
return lines, nil
|
||||
}
|
||||
@@ -1,452 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"reflect"
|
||||
"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"
|
||||
"github.com/novox/mesh-controller/internal/inventory"
|
||||
)
|
||||
|
||||
// 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
|
||||
stale []string // what forgetting the accounts of modules no longer on their nodes says
|
||||
}{
|
||||
{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"},
|
||||
{name: "the start's look forgets the accounts of modules no longer on their nodes first",
|
||||
look: []chain.Assignment{ace("mesh-build-node")}, holder: "novox", wantIssues: 1, wantComposed: []string{"novox"},
|
||||
wantGrants: true, stale: []string{"mesh-build-go no longer runs on ace: its bus accounts forgotten: ace.mesh-build-go"}},
|
||||
}
|
||||
for _, c := range cases {
|
||||
t.Run(c.name, func(t *testing.T) {
|
||||
var composed []string
|
||||
issues, catalogues, grants := 0, 0, 0
|
||||
forgets := 0
|
||||
steps := assignmentAccountSteps{
|
||||
holds: func(context.Context) bool { return !c.notHolding },
|
||||
forgetStale: func(context.Context) ([]string, error) {
|
||||
forgets++
|
||||
return c.stale, nil
|
||||
},
|
||||
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 want := map[bool]int{true: 1}[len(c.look) > 0 && !c.notHolding]; forgets != want {
|
||||
t.Fatalf("forgot stale accounts %d time(s), want %d (only from a reading that holds assignments)", forgets, want)
|
||||
}
|
||||
for _, line := range c.stale {
|
||||
if !strings.Contains(said, line) {
|
||||
t.Fatalf("said %q, want it to say %q", said, line)
|
||||
}
|
||||
}
|
||||
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-<process id> (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)
|
||||
}
|
||||
})
|
||||
|
||||
// An issue mints and seals an account only when it is not issued: no row, or a row with no sealed credential (issue
|
||||
// 1945, live on 2026-10-25 at 20:04: ace's mesh-build-go had a row from 17:19 and no credential), under the lock on
|
||||
// the account; a seal that fails leaves it unsealed for the next issue. A rotation always mints.
|
||||
for _, c := range []struct {
|
||||
name string
|
||||
only bool
|
||||
minted, sealed bool
|
||||
sealErr error
|
||||
want string
|
||||
}{
|
||||
{"no row: minted unsealed, sealed, marked", true, false, false, nil, "lock mint-unsealed seal mark release"},
|
||||
{"a row sealed: nothing", true, true, true, nil, "lock release"},
|
||||
{"a row with no sealed credential: minted and sealed again", true, true, false, nil, "lock mint-unsealed seal mark release"},
|
||||
{"a seal that fails: not marked", true, false, false, errors.New("the vault did not take it"), "lock mint-unsealed seal release"},
|
||||
{"a rotation mints whatever stands", false, true, true, nil, "mint seal"},
|
||||
} {
|
||||
t.Run("issuing an account: "+c.name, func(t *testing.T) {
|
||||
acts := &anAccountStore{minted: c.minted, sealed: c.sealed}
|
||||
ctx, got := withIssued(t.Context())
|
||||
if c.only {
|
||||
ctx = onlyIfAbsent(ctx)
|
||||
}
|
||||
err := issueAccount(ctx, acts, inventory.BusUser{Username: "ace.mesh-build-go", Kind: inventory.BusModule},
|
||||
func(password string) error {
|
||||
acts.did("seal")
|
||||
if password == "" {
|
||||
t.Error("sealed no password")
|
||||
}
|
||||
return c.sealErr
|
||||
})
|
||||
if (err != nil) != (c.sealErr != nil) {
|
||||
t.Fatalf("error %v, want %v", err, c.sealErr)
|
||||
}
|
||||
if strings.Join(acts.acts, " ") != c.want {
|
||||
t.Fatalf("acts %q, want %q", strings.Join(acts.acts, " "), c.want)
|
||||
}
|
||||
if issued := len(got.list()) == 1; issued != (c.sealErr == nil && strings.Contains(c.want, "seal")) {
|
||||
t.Fatalf("said issued %v", got.list())
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// The accounts of a module that no longer resolves on its node are stale, its provisioner client's among them; a
|
||||
// module still resolved (assigned or required), an account of another kind, and a node whose resolution is unknown
|
||||
// are left alone (issue 1945: ace's mesh-build-go rows from 17:19, marked sealed by migration 0108).
|
||||
t.Run("stale accounts", func(t *testing.T) {
|
||||
u := func(kind, node, module string) inventory.BusUser {
|
||||
return inventory.BusUser{Username: node + "." + module + "." + kind, Kind: kind, Node: node, Module: module}
|
||||
}
|
||||
users := map[string]inventory.BusUser{}
|
||||
for _, b := range []inventory.BusUser{u(inventory.BusModule, "ace", "mesh-build-go"),
|
||||
u(inventory.BusProcess, "ace", "mesh-build-go"), u(inventory.BusProvisioner, "ace", "mesh-provisioner"),
|
||||
u(inventory.BusModule, "ace", "mesh-build-node"), u(inventory.BusNode, "ace", ""),
|
||||
u(inventory.BusModule, "g14", "mesh-build-go")} {
|
||||
users[b.Username] = b
|
||||
}
|
||||
got := staleAccounts(users, map[string]map[string]bool{"ace": {"mesh-build-node": true}})
|
||||
if want := map[string][]string{"ace": {"mesh-build-go", "mesh-provisioner"}}; !reflect.DeepEqual(got, want) {
|
||||
t.Fatalf("stale %v, want %v", 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-<id> 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)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// anAccountStore is an accountIssuer that writes down what was asked of it.
|
||||
type anAccountStore struct {
|
||||
minted, sealed bool
|
||||
acts []string
|
||||
}
|
||||
|
||||
func (a *anAccountStore) did(act string) { a.acts = append(a.acts, act) }
|
||||
func (a *anAccountStore) LockBusAccount(context.Context, string) (func(), error) {
|
||||
a.did("lock")
|
||||
return func() { a.did("release") }, nil
|
||||
}
|
||||
func (a *anAccountStore) BusAccountIssued(context.Context, string) (bool, bool, error) {
|
||||
return a.minted, a.sealed, nil
|
||||
}
|
||||
func (a *anAccountStore) MintBusPassword(context.Context, inventory.BusUser) (string, error) {
|
||||
a.did("mint")
|
||||
return "a-password", nil
|
||||
}
|
||||
func (a *anAccountStore) MintBusPasswordUnsealed(context.Context, inventory.BusUser) (string, error) {
|
||||
a.did("mint-unsealed")
|
||||
return "a-password", nil
|
||||
}
|
||||
func (a *anAccountStore) MarkBusAccountSealed(context.Context, string) error {
|
||||
a.did("mark")
|
||||
return nil
|
||||
}
|
||||
@@ -273,7 +273,4 @@ 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
|
||||
}
|
||||
|
||||
@@ -326,13 +326,9 @@ func restOfTheMesh(ctx context.Context, open *stores, shelf map[string]catalogue
|
||||
return world, nil
|
||||
}
|
||||
|
||||
// restOfTheMeshOf is restOfTheMesh for a caller that holds the inventory alone: on the serving controller's stores where it
|
||||
// lends them (sharedStores), never a pool per call; elsewhere (a verb run as its own process) the other stores the Input
|
||||
// reads are opened for the call and let go of again.
|
||||
// restOfTheMeshOf is restOfTheMesh for a caller that holds the inventory alone: the other stores the Input reads are opened for
|
||||
// the call and let go of again.
|
||||
func restOfTheMeshOf(ctx context.Context, inv *inventory.Inventory, shelf map[string]catalogue.Manifest, exclude string) (catalogue.World, error) {
|
||||
if shared, ok := sharedStoresOf(inv); ok {
|
||||
return restOfTheMeshOn(ctx, shared, shelf, exclude)
|
||||
}
|
||||
open := &stores{inventory: inv}
|
||||
defer func() {
|
||||
if open.licences != nil {
|
||||
@@ -342,12 +338,9 @@ func restOfTheMeshOf(ctx context.Context, inv *inventory.Inventory, shelf map[st
|
||||
open.identity.Close()
|
||||
}
|
||||
}()
|
||||
return restOfTheMeshOn(ctx, open, shelf, exclude)
|
||||
return restOfTheMesh(ctx, open, shelf, exclude)
|
||||
}
|
||||
|
||||
// restOfTheMeshOn is restOfTheMesh; a variable so a test reads which stores it is handed.
|
||||
var restOfTheMeshOn = restOfTheMesh
|
||||
|
||||
// sayFound says the lines a composition found while composing, once each: a machine left out of the rest of the mesh, a seat
|
||||
// that two claimants could take, the bus's users left out for want of a credential.
|
||||
func sayFound(ctx context.Context, d compose.Declared) {
|
||||
|
||||
@@ -102,10 +102,8 @@ 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")
|
||||
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 := liveDoneKeySteps(m.open, trunkKeys{keyBuckets: keys, trunk: trunk}).take(ctx, "mesh-host"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := m.js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: sdkdelivery.HaveBucket, History: sdkdelivery.HaveHistory}); err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -142,35 +140,4 @@ func TestTheDeclarationsStreamGoesOnceEveryNodeRunsTheNewEngine(t *testing.T) {
|
||||
if !declarationsRetiredNow(ctx, m.open.inventory) {
|
||||
t.Errorf("a raise after the retirement would assert the stream again")
|
||||
}
|
||||
|
||||
// Issue 1945: a node-engine other than the registered one landing on laptop (a newer build on its way) leaves the
|
||||
// declarations retired, so what the health check, the probes and healer H3 expect of the bus names neither the
|
||||
// stream nor a node's consumer on it, and asserting the bus's objects again does not make the stream anew.
|
||||
have("laptop", fpOf('7'))
|
||||
if !declarationsRetiredNow(ctx, m.open.inventory) {
|
||||
t.Fatalf("with laptop on another node-engine after the stream went, the declarations are no longer retired")
|
||||
}
|
||||
streams, consumers, err := expectedBusObjects(ctx, m.open.inventory)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, s := range streams {
|
||||
if s.Name == broker.DeclarationsStream {
|
||||
t.Errorf("the bus's health check and probe D7 still expect the stream %s", s.Name)
|
||||
}
|
||||
}
|
||||
for _, c := range consumers {
|
||||
if c.Stream == broker.DeclarationsStream {
|
||||
t.Errorf("the bus's health check and probe D6 still expect %s on %s", c.Name, c.Stream)
|
||||
}
|
||||
}
|
||||
bus, err := broker.Dial(m.conn.ConnectedUrl())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(bus.Close)
|
||||
_ = assertOnSend(ctx, m.open.inventory, bus, " ")
|
||||
if _, err := m.js.Stream(ctx, broker.DeclarationsStream); !errors.Is(err, jetstream.ErrStreamNotFound) {
|
||||
t.Errorf("healer H3's assertion of the bus's objects made the stream anew: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4,8 +4,6 @@ import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"maps"
|
||||
"slices"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
@@ -14,285 +12,210 @@ 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, 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.<module>.<fp>`. 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.
|
||||
// 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.<module>.<fp>`. 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).
|
||||
|
||||
// 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)
|
||||
// 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)
|
||||
// trunk is the indexer's trunk value of the module.
|
||||
trunk func(ctx context.Context, module string) (sdkdelivery.Module, error)
|
||||
// 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).
|
||||
// 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.
|
||||
takeIn func(ctx context.Context, module string, trunk sdkdelivery.Module) 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.
|
||||
// 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)
|
||||
busHolder func(ctx context.Context) (string, 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)
|
||||
// compose writes the nodes' compositions; grants writes the grants issued.
|
||||
compose func(ctx context.Context, nodes []string) error
|
||||
grants func(ctx context.Context) error
|
||||
}
|
||||
|
||||
// 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
|
||||
}
|
||||
|
||||
// verdict is what becomes of a module's delivered keys.
|
||||
type verdict int
|
||||
|
||||
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
|
||||
)
|
||||
|
||||
// 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
|
||||
}
|
||||
|
||||
// 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)
|
||||
}
|
||||
again[k.module] = again[k.module] || k.again
|
||||
// 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
|
||||
}
|
||||
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})
|
||||
}
|
||||
return out, 0
|
||||
trunk, err := s.trunk(ctx, module)
|
||||
if errors.Is(err, builderkeys.ErrNoValue) {
|
||||
return "", nil
|
||||
}
|
||||
held, err := s.catalogue(ctx)
|
||||
if err != nil {
|
||||
return all(refused, fmt.Errorf("the catalogue cannot be read: %w", err))
|
||||
return "", err
|
||||
}
|
||||
type toCompose struct {
|
||||
at int
|
||||
nodes []string
|
||||
mark string
|
||||
said string
|
||||
if !strings.HasPrefix(trunk.FP2, sdkdelivery.SourcePrefix2) {
|
||||
return "", nil
|
||||
}
|
||||
var need []toCompose
|
||||
for _, m := range order {
|
||||
r := settledKey{module: m}
|
||||
nodes, isHeld := held[m]
|
||||
if !isHeld {
|
||||
out = append(out, r)
|
||||
continue
|
||||
}
|
||||
trunk, err := s.trunk(ctx, m)
|
||||
if errors.Is(err, builderkeys.ErrNoValue) {
|
||||
out = append(out, r)
|
||||
continue
|
||||
}
|
||||
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
|
||||
}
|
||||
}
|
||||
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)
|
||||
at, err := s.registeredAt(ctx, module, trunk.Commit)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
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
|
||||
said := ""
|
||||
if !at {
|
||||
if err := s.takeIn(ctx, module, trunk); err != nil {
|
||||
return "", err
|
||||
}
|
||||
return out, 0
|
||||
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 refuseNeed(fmt.Errorf("the node holding the bus cannot be read: %w", err))
|
||||
return said, 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)
|
||||
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
|
||||
}
|
||||
|
||||
func inList(list []string, s string) bool {
|
||||
for _, v := range list {
|
||||
if v == s {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// 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) }
|
||||
)
|
||||
|
||||
// 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
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 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 := doneKeyTakeOnce(ctx, open, module)
|
||||
o := doneKeyObservation(module)
|
||||
if err == nil {
|
||||
if keeper := conditionsFrom; keeper != nil && failed {
|
||||
_, _ = keeper.Clear(ctx, o.Key(), "composed and granted")
|
||||
failed = false
|
||||
}
|
||||
}
|
||||
}
|
||||
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 _, again := doneKeysLookAgain.LoadAndDelete(module); again {
|
||||
continue // a wake came while this one ran: once more, for the key written meanwhile
|
||||
}
|
||||
return
|
||||
}
|
||||
if grantsErr != nil {
|
||||
errs = append(errs, grantsErr)
|
||||
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))
|
||||
}
|
||||
r := &out[n.at]
|
||||
if len(errs) > 0 {
|
||||
r.verdict, r.err = refused, errors.Join(errs...)
|
||||
if errors.Is(err, errKeyWaits) {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-time.After(doneKeyRetry):
|
||||
}
|
||||
continue
|
||||
}
|
||||
memo[r.module] = n.mark
|
||||
r.taken = true
|
||||
r.said = n.said + "; composed for " + strings.Join(n.nodes, ", ") + ", its grant written"
|
||||
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
|
||||
}
|
||||
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
|
||||
}
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-time.After(doneKeyRetry):
|
||||
}
|
||||
}
|
||||
return out, passes
|
||||
}
|
||||
|
||||
// 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.
|
||||
// doneKeyRetry is how long a failed take-in of a done key waits before it tries again.
|
||||
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"
|
||||
@@ -303,37 +226,40 @@ func doneKeyObservation(module string) conditions.Observation {
|
||||
Summary: module + "'s new fingerprint is not composed or granted, so no want can name it: "}
|
||||
}
|
||||
|
||||
// liveTakeInSteps are the steps on the controller's own stores and the bus.
|
||||
func liveTakeInSteps(open *stores, keys registerKeys, js jetstream.JetStream) takeInSteps {
|
||||
// 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 {
|
||||
inv := open.inventory
|
||||
return takeInSteps{
|
||||
catalogue: func(ctx context.Context) (map[string][]string, error) {
|
||||
entries, err := inv.Catalogued(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
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
|
||||
}
|
||||
out := make(map[string][]string, len(entries))
|
||||
for _, e := range entries {
|
||||
out[e.Manifest.Module] = e.On
|
||||
}
|
||||
return out, nil
|
||||
return err == nil, err
|
||||
},
|
||||
trunk: func(ctx context.Context, module string) (sdkdelivery.Module, error) {
|
||||
return keys.ModuleAt(ctx, module, "")
|
||||
},
|
||||
registered: func(ctx context.Context, module string, trunk sdkdelivery.Module) (bool, bool, error) {
|
||||
registeredAt: func(ctx context.Context, module, commit string) (bool, error) {
|
||||
src, err := inv.SourceOf(ctx, module)
|
||||
if err != nil {
|
||||
return false, false, err
|
||||
return false, err
|
||||
}
|
||||
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)
|
||||
return sameCommitID(src.BuiltFrom, commit), nil
|
||||
},
|
||||
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}})
|
||||
@@ -342,8 +268,17 @@ func liveTakeInSteps(open *stores, keys registerKeys, js jetstream.JetStream) ta
|
||||
}
|
||||
return err
|
||||
},
|
||||
taken: func(ctx context.Context, module, fp string, nodes []string) (bool, error) {
|
||||
return composedAndGranted(ctx, js, module, fp, nodes)
|
||||
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
|
||||
},
|
||||
busHolder: func(ctx context.Context) (string, error) {
|
||||
holders, err := seatHolders(ctx, inv)
|
||||
@@ -354,64 +289,107 @@ func liveTakeInSteps(open *stores, keys registerKeys, js jetstream.JetStream) ta
|
||||
},
|
||||
// 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, grants bool) (map[string]error, error, error) {
|
||||
return composerFor(open).composeEach(ctx, nodes, grants)
|
||||
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)
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// 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 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
|
||||
}
|
||||
}
|
||||
at, err := keys.ModuleAt(ctx, module, commit)
|
||||
// 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 {
|
||||
return "", err
|
||||
fmt.Printf("done keys: the builder's keys are not watched: %v\n", err)
|
||||
return
|
||||
}
|
||||
return at.FP2, nil
|
||||
if err := builderkeys.Watch(ctx, keys.JS, func(module string) {
|
||||
inBackground(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(func() {
|
||||
doneKeysAtStart(ctx, catalogedModules(open), func(module string) { takeDoneKeyAtStart(ctx, open, module) })
|
||||
})
|
||||
}
|
||||
|
||||
// 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)
|
||||
// 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 {
|
||||
return false, err
|
||||
fmt.Printf("done keys: the modules are not looked at once at the start: %v\n", err)
|
||||
return
|
||||
}
|
||||
key, err := sdkdelivery.ControllerGrantKey(module, fp)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
if _, err := grants.Get(ctx, key); err != nil {
|
||||
if errors.Is(err, jetstream.ErrKeyNotFound) {
|
||||
return false, nil
|
||||
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
|
||||
}
|
||||
return false, err
|
||||
sem <- struct{}{}
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer func() { <-sem; wg.Done() }()
|
||||
take(m)
|
||||
}()
|
||||
}
|
||||
digests, err := js.KeyValue(ctx, sdkdelivery.ControllerComposedBucket)
|
||||
if err != nil {
|
||||
return false, err
|
||||
wg.Wait()
|
||||
}
|
||||
|
||||
// startTakeIns is how many take-ins the start's look runs at once.
|
||||
const startTakeIns = 16
|
||||
|
||||
// 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
|
||||
}
|
||||
for _, node := range nodes {
|
||||
key, err := sdkdelivery.ComposedKey(node, module, fp)
|
||||
doneKeysLookAgain.Delete(module) // what this look reads covers every wake before it
|
||||
err := doneKeyTakeOnce(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(func() { takeDoneKey(ctx, open, module) })
|
||||
}
|
||||
}
|
||||
|
||||
// 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)
|
||||
if err != nil {
|
||||
return false, err
|
||||
return nil, err
|
||||
}
|
||||
if _, err := digests.Get(ctx, key); err != nil {
|
||||
if errors.Is(err, jetstream.ErrKeyNotFound) {
|
||||
return false, nil
|
||||
}
|
||||
return false, err
|
||||
out := make([]string, 0, len(entries))
|
||||
for _, e := range entries {
|
||||
out = append(out, e.Manifest.Module)
|
||||
}
|
||||
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
|
||||
|
||||
@@ -0,0 +1,61 @@
|
||||
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)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,34 @@
|
||||
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)
|
||||
}
|
||||
}
|
||||
@@ -4,305 +4,84 @@ 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"
|
||||
)
|
||||
|
||||
// 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 {
|
||||
// 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 == "" {
|
||||
return sdkdelivery.Module{}, builderkeys.ErrNoValue
|
||||
}
|
||||
return sdkdelivery.Module{Commit: commit, FP2: fp(module)}, nil
|
||||
return sdkdelivery.Module{Commit: commit, FP2: fp}, 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)
|
||||
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
|
||||
},
|
||||
takeIn: func(_ context.Context, module string, trunk sdkdelivery.Module) error {
|
||||
m.acts = append(m.acts, "take in "+module+" at "+trunk.Commit)
|
||||
return m.takeInErr[module]
|
||||
},
|
||||
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
|
||||
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
|
||||
},
|
||||
grants: func(context.Context) error { *acts = append(*acts, "grants"); return nil },
|
||||
}
|
||||
}
|
||||
|
||||
// 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
|
||||
// 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)
|
||||
}
|
||||
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 got := strings.Join(acts, "; "); got != "take in mesh-mail at new; compose ace,g14; compose novox; grants" {
|
||||
t.Errorf("the acts were %q", got)
|
||||
}
|
||||
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)
|
||||
}
|
||||
}
|
||||
})
|
||||
if !strings.Contains(said, "registered from the builder's done key") {
|
||||
t.Errorf("said %q", said)
|
||||
}
|
||||
|
||||
// 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)
|
||||
}
|
||||
})
|
||||
|
||||
// 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
|
||||
acts = nil
|
||||
if _, err := stepsFor(&acts, "new", "new", fp, []string{"novox"}, "novox").take(context.Background(), "mesh-mail"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
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},
|
||||
}},
|
||||
if got := strings.Join(acts, "; "); got != "compose novox; grants" {
|
||||
t.Errorf("a module registered at its trunk already: %q", got)
|
||||
}
|
||||
|
||||
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"),
|
||||
} {
|
||||
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)
|
||||
}
|
||||
}
|
||||
})
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -336,3 +115,15 @@ 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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,184 @@
|
||||
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)
|
||||
}
|
||||
}
|
||||
@@ -88,8 +88,6 @@ func composeModulesByNode(ctx context.Context, open *stores, nodes []string) (ma
|
||||
}
|
||||
return u.Policy(), u.From, u.Why
|
||||
}
|
||||
// The memberships of each node's modules are issued by its composition (memberships.go): read once for the pass.
|
||||
issuing, issuingErr := membershipsFor(ctx, open, link.OverNATS{Conn: bus.Conn(), JS: bus.Context()})
|
||||
sort.Strings(nodes)
|
||||
out := map[string][]string{}
|
||||
errsOf := map[string][]error{}
|
||||
@@ -117,13 +115,6 @@ func composeModulesByNode(ctx context.Context, open *stores, nodes []string) (ma
|
||||
errsOf[node] = append(errsOf[node], fmt.Errorf("%s: %w", node, err))
|
||||
return
|
||||
}
|
||||
// Its modules' memberships, before its compositions (issue 249): a failure is the node's, and its compositions
|
||||
// are still written; the take-in that asked tries the node again.
|
||||
if issuingErr != nil {
|
||||
errsOf[node] = append(errsOf[node], fmt.Errorf("%s: %w", node, issuingErr))
|
||||
} else if err := issuing.issue(ctx, node, declared); err != nil {
|
||||
errsOf[node] = append(errsOf[node], fmt.Errorf("%s: %w", node, err))
|
||||
}
|
||||
// The provisioner's plans of the modules the node's composition hands to it, for the assignments mirror.
|
||||
if err := open.inventory.RecordComposedPlans(ctx, node, plansKept(declared)); err != nil {
|
||||
errsOf[node] = append(errsOf[node], fmt.Errorf("%s: its provisioner's plans: %w", node, err))
|
||||
|
||||
@@ -152,7 +152,7 @@ func watchHavesForJudgement(ctx context.Context, open *stores) {
|
||||
if err != nil || isNode {
|
||||
continue
|
||||
}
|
||||
inBackground(ctx, func() {
|
||||
inBackground(func() {
|
||||
if err := rejudgeOn(ctx, open, []string{node}, module); err != nil {
|
||||
fmt.Printf("compositions: %s on %s is not judged again: %v\n", module, node, err)
|
||||
}
|
||||
|
||||
@@ -135,18 +135,9 @@ func declarationsDone(nodes []string, runs map[string]string, current string) bo
|
||||
return true
|
||||
}
|
||||
|
||||
// declarationsRetiredOn reads whether the declarations are done on the mesh now: their stream gone from the bus, or
|
||||
// every node's have of the node-engine against the fingerprint it is registered at.
|
||||
//
|
||||
// **A stream that is gone stays gone** (issue 1945). Release 6 sends no declaration (ADR 0539 clauses 5 and 11), so a
|
||||
// stream made again would hold no node's last declaration and help no node; and the done test alone is not lasting:
|
||||
// a node-engine newer than the one registered landing, or a node asleep on an older have, makes it false again after
|
||||
// the start deleted the stream, and then the raise, the bus's health check (H-bus), the probes (D6, D7) and healer H3
|
||||
// expected the stream and every node's consumer again, raised them missing and made them anew (2026-10-25 14:52).
|
||||
// declarationsRetiredOn reads whether the declarations are done on the mesh now: every node's have of the node-engine
|
||||
// against the fingerprint it is registered at.
|
||||
func declarationsRetiredOn(ctx context.Context, inv *inventory.Inventory, js jetstream.JetStream) (bool, error) {
|
||||
if _, err := js.Stream(ctx, broker.DeclarationsStream); errors.Is(err, jetstream.ErrStreamNotFound) {
|
||||
return true, nil
|
||||
}
|
||||
fps, err := currentFingerprints(ctx, inv)
|
||||
if err != nil {
|
||||
return false, err
|
||||
@@ -171,8 +162,7 @@ func declarationsRetiredOn(ctx context.Context, inv *inventory.Inventory, js jet
|
||||
}
|
||||
|
||||
// retireDeclarations deletes the declarations' stream, with every node's consumer on it, once every node runs the new
|
||||
// node-engine, and says it once; a stream already gone is said by nobody, and is done. It answers whether the
|
||||
// declarations are done.
|
||||
// node-engine, and says it once; a stream already gone is said by nobody. It answers whether the declarations are done.
|
||||
func retireDeclarations(ctx context.Context, open *stores) (bool, error) {
|
||||
if _, err := theLease.epoch(ctx); err != nil {
|
||||
return false, err
|
||||
@@ -316,33 +306,26 @@ func composeReplacedController(ctx context.Context, open *stores) error {
|
||||
// replacedComposition is the controller's composition on r's node at r's fingerprint: in composed with r's manifest
|
||||
// in place of the controller's registered one.
|
||||
func replacedComposition(in compose.Input, r replacedController) (sdkdelivery.Composed, error) {
|
||||
return keptComposition(in, r.node, controllerModule, r.fp, r.manifest)
|
||||
}
|
||||
|
||||
// keptComposition is module's composition on node at fp: in composed with manifest, that fingerprint's own, in place of
|
||||
// the module's registered one, in the reading mode (nothing made). in is not changed.
|
||||
func keptComposition(in compose.Input, node, module, fp string, manifest catalogue.Manifest) (sdkdelivery.Composed, error) {
|
||||
shelf := make(map[string]catalogue.Manifest, len(in.Catalogue))
|
||||
for k, v := range in.Catalogue {
|
||||
shelf[k] = v
|
||||
}
|
||||
shelf[module] = manifest
|
||||
shelf[controllerModule] = r.manifest
|
||||
in.Catalogue = shelf
|
||||
all, _, err := compose.Compositions(in, node, map[string]string{module: fp}, compose.Reading)
|
||||
all, _, err := compose.Compositions(in, r.node, map[string]string{controllerModule: r.fp}, compose.Reading)
|
||||
if err != nil {
|
||||
return sdkdelivery.Composed{}, err
|
||||
}
|
||||
key, _ := sdkdelivery.ComposedKey(node, module, fp)
|
||||
key, _ := sdkdelivery.ComposedKey(r.node, controllerModule, r.fp)
|
||||
c, ok := all[key]
|
||||
if !ok {
|
||||
return sdkdelivery.Composed{}, fmt.Errorf("%s composes no %s with the manifest of %s", node, module, short(fp))
|
||||
return sdkdelivery.Composed{}, fmt.Errorf("%s composes no %s with the replaced build's manifest", r.node, controllerModule)
|
||||
}
|
||||
return c, nil
|
||||
}
|
||||
|
||||
// declarationsRetiredNow says the declarations are done on the mesh now, for a raise of the bus's objects and for what
|
||||
// the health check, the probes and the healers expect of the bus; a bus or a store it cannot read is not done, so a
|
||||
// stream still there is asserted as before.
|
||||
// declarationsRetiredNow says the declarations are done on the mesh now, for a raise of the bus's objects; a bus or a
|
||||
// store it cannot read is not done, so the stream is asserted as before.
|
||||
func declarationsRetiredNow(ctx context.Context, inv *inventory.Inventory) bool {
|
||||
bus, err := aBus()
|
||||
if err != nil {
|
||||
|
||||
@@ -3,8 +3,6 @@ package main
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
@@ -47,97 +45,51 @@ func TestTheFirstStartDeletesThePlansBucketsAndSaysEachOnce(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// An earlier fingerprint is composed from its own build's manifest with today's inputs: the controller this one
|
||||
// replaced (1945/AC6), and every kept composition of a module whose config changed (ADR 0545: on 2026-10-25 a want back
|
||||
// to an earlier mesh-build-go on g14 applied the capacity of 0 its kept composition was made with). The composition
|
||||
// holds what that manifest declares and today's config, the kept keys are those of the touched modules at a
|
||||
// fingerprint other than the registered one, the build of a fingerprint is its newest that worked, and a write of
|
||||
// mesh-config's resolved values names the node and module it touches.
|
||||
func TestAnEarlierFingerprintIsComposedFromItsOwnManifest(t *testing.T) {
|
||||
fp := func(c string) string { return sdkdelivery.SourcePrefix2 + strings.Repeat(c, 64) }
|
||||
t.Run("from its own manifest, with today's inputs", func(t *testing.T) {
|
||||
manifest := func(version string) catalogue.Manifest {
|
||||
return catalogue.Manifest{Module: controllerModule, Version: version,
|
||||
Settings: map[string]catalogue.SettingDeclaration{
|
||||
"capacity": {Kind: catalogue.KindPreference, Default: float64(0), Why: "how many builds at once"}},
|
||||
Resources: []map[string]any{{"id": "settings", "type": "file", "path": "/etc/mesh-controller/settings",
|
||||
"content": version + " ${setting:capacity}\n"}}}
|
||||
}
|
||||
now, before := manifest("six"), manifest("five")
|
||||
in := compose.Input{Catalogue: map[string]catalogue.Manifest{controllerModule: now},
|
||||
Nodes: []compose.Node{{Name: "anchor", Assigned: []string{controllerModule}, Stated: true,
|
||||
Settings: map[string][]catalogue.Layer{controllerModule: {{From: "mesh", Values: map[string]any{"capacity": float64(4)}}}}}}}
|
||||
for name, compose := range map[string]func() (sdkdelivery.Composed, error){
|
||||
"the replaced controller": func() (sdkdelivery.Composed, error) {
|
||||
return replacedComposition(in, replacedController{node: "anchor", fp: fp("5"), manifest: before})
|
||||
},
|
||||
"a kept composition": func() (sdkdelivery.Composed, error) {
|
||||
return keptComposition(in, "anchor", controllerModule, fp("5"), before)
|
||||
},
|
||||
} {
|
||||
c, err := compose()
|
||||
if err != nil {
|
||||
t.Fatalf("%s: %v", name, err)
|
||||
}
|
||||
if c.Module != controllerModule || c.FP != fp("5") || c.Node != "anchor" || len(c.Resources) != 1 {
|
||||
t.Fatalf("%s: composed %s at %s on %s with %d resources", name, c.Module, c.FP, c.Node, len(c.Resources))
|
||||
}
|
||||
var f struct{ Content string }
|
||||
_ = json.Unmarshal(c.Resources[0], &f)
|
||||
if f.Content != "five 4\n" {
|
||||
t.Errorf("%s holds %q, want its own manifest's five with today's capacity 4", name, f.Content)
|
||||
}
|
||||
}
|
||||
if in.Catalogue[controllerModule].Version != "six" {
|
||||
t.Errorf("composing an earlier fingerprint changed the Input's catalogue")
|
||||
}
|
||||
})
|
||||
// 1945/AC6 (unit half): the controller this one replaced is composed on its node from its own build's manifest at its
|
||||
// own fingerprint: the composition holds what that manifest declares, not what the controller registered now declares.
|
||||
func TestTheReplacedControllerIsComposedFromItsOwnManifest(t *testing.T) {
|
||||
now := catalogue.Manifest{Module: controllerModule, Version: "6", Resources: []map[string]any{
|
||||
{"id": "settings", "type": "file", "path": "/etc/mesh-controller/settings", "content": "six\n"}}}
|
||||
before := catalogue.Manifest{Module: controllerModule, Version: "5", Resources: []map[string]any{
|
||||
{"id": "settings", "type": "file", "path": "/etc/mesh-controller/settings", "content": "five\n"}}}
|
||||
in := compose.Input{Catalogue: map[string]catalogue.Manifest{controllerModule: now},
|
||||
Nodes: []compose.Node{{Name: "anchor", Assigned: []string{controllerModule}, Stated: true}}}
|
||||
fp := sdkdelivery.SourcePrefix2 + strings.Repeat("5", 64)
|
||||
c, err := replacedComposition(in, replacedController{node: "anchor", fp: fp, manifest: before})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if c.Module != controllerModule || c.FP != fp || c.Node != "anchor" || len(c.Resources) != 1 {
|
||||
t.Fatalf("composed %s at %s on %s with %d resources", c.Module, c.FP, c.Node, len(c.Resources))
|
||||
}
|
||||
var f struct{ Content string }
|
||||
_ = json.Unmarshal(c.Resources[0], &f)
|
||||
if f.Content != "five\n" {
|
||||
t.Errorf("the replaced controller's composition holds %q, its own manifest says five", f.Content)
|
||||
}
|
||||
if in.Catalogue[controllerModule].Version != "6" {
|
||||
t.Errorf("composing the replaced controller changed the Input's catalogue")
|
||||
}
|
||||
}
|
||||
|
||||
t.Run("which build, which keys, which touch", func(t *testing.T) {
|
||||
m := []byte(`{"module":"mesh-controller","version":"5"}`)
|
||||
builds := []inventory.Build{
|
||||
{ID: "c6", SourceFingerprint: fp("6"), Manifest: m},
|
||||
{ID: "c5-failed", SourceFingerprint: fp("5"), Failed: "no", Manifest: m},
|
||||
{ID: "c5", SourceFingerprint: fp("5"), Manifest: m},
|
||||
{ID: "c4", SourceFingerprint: fp("4"), Manifest: m},
|
||||
}
|
||||
if b, ok := replacedBuild(builds, fp("6")); !ok || b.ID != "c5" {
|
||||
t.Errorf("the replaced build is %+v (%v), want c5", b, ok)
|
||||
}
|
||||
if _, ok := replacedBuild(builds[:1], fp("6")); ok {
|
||||
t.Errorf("a controller with no earlier build replaced one")
|
||||
}
|
||||
if b, ok := buildAt(builds, fp("5")); !ok || b.ID != "c5" {
|
||||
t.Errorf("the build at 5 is %+v (%v), want c5, the one that worked", b, ok)
|
||||
}
|
||||
if _, ok := buildAt(builds, fp("3")); ok {
|
||||
t.Errorf("a fingerprint with no build has one")
|
||||
}
|
||||
key := func(node, module, f string) string { k, _ := sdkdelivery.ComposedKey(node, module, f); return k }
|
||||
keys := []string{key("g14", "mesh-build-go", fp("2")), key("g14", "mesh-build-go", fp("1")),
|
||||
key("g14", "mesh-build-go", fp("3")), key("g14", "hello-web", fp("1")), key("ace", "mesh-build-go", fp("1"))}
|
||||
registered := map[string]string{"mesh-build-go": fp("3"), "hello-web": fp("9")}
|
||||
got := keptToFollow(keys, "g14", map[string]bool{"mesh-build-go": true}, registered)
|
||||
if want := []keptKey{{"mesh-build-go", fp("1")}, {"mesh-build-go", fp("2")}}; !slices.Equal(got, want) {
|
||||
t.Errorf("kept to follow %v, want %v", got, want)
|
||||
}
|
||||
if got := keptToFollow(keys, "g14", nil, registered); len(got) != 3 {
|
||||
t.Errorf("every module's kept compositions: %v", got)
|
||||
}
|
||||
for k, want := range map[string][2]string{"g14.mesh-build-go.capacity": {"g14", "mesh-build-go"},
|
||||
"g14.mesh-build-go.limits.cpu": {"g14", "mesh-build-go"}, "g14.mesh-build-go": {}, "": {}} {
|
||||
node, module, ok := configTouched(k)
|
||||
if [2]string{node, module} != want || ok != (want[0] != "") {
|
||||
t.Errorf("%q touches %s %s (%v), want %v", k, node, module, ok, want)
|
||||
}
|
||||
}
|
||||
// A value set before the consumer was made reaches the kept compositions: it reads each key once at creation.
|
||||
if c := configFollowConfig(); c.DeliverPolicy != jetstream.DeliverLastPerSubjectPolicy ||
|
||||
c.AckPolicy != jetstream.AckExplicitPolicy {
|
||||
t.Errorf("the config consumer delivers %v, acknowledged %v; want each key's last value, acknowledged",
|
||||
c.DeliverPolicy, c.AckPolicy)
|
||||
}
|
||||
})
|
||||
// 1945/AC6 (unit half): the controller replaced is the newest earlier build of the controller that worked and names a
|
||||
// fingerprint of the new form, not the one registered now.
|
||||
func TestTheReplacedControllerIsItsNewestEarlierBuild(t *testing.T) {
|
||||
now, before, older := sdkdelivery.SourcePrefix2+strings.Repeat("6", 64), sdkdelivery.SourcePrefix2+strings.Repeat("5", 64),
|
||||
sdkdelivery.SourcePrefix2+strings.Repeat("4", 64)
|
||||
m := []byte(`{"module":"mesh-controller","version":"5"}`)
|
||||
builds := []inventory.Build{
|
||||
{ID: "c6", SourceFingerprint: now, Manifest: m},
|
||||
{ID: "c5-failed", SourceFingerprint: before, Failed: "no", Manifest: m},
|
||||
{ID: "c5", SourceFingerprint: before, Manifest: m},
|
||||
{ID: "c4", SourceFingerprint: older, Manifest: m},
|
||||
}
|
||||
if b, ok := replacedBuild(builds, now); !ok || b.ID != "c5" {
|
||||
t.Errorf("the replaced build is %+v (%v), want c5", b, ok)
|
||||
}
|
||||
if _, ok := replacedBuild(builds[:1], now); ok {
|
||||
t.Errorf("a controller with no earlier build replaced one")
|
||||
}
|
||||
}
|
||||
|
||||
// 1945/AC14 (unit half): the declarations' stream stays while one node's have names another node-engine than the one
|
||||
@@ -158,28 +110,3 @@ func TestTheDeclarationsGoOnlyOnceEveryNodeRunsTheNewEngine(t *testing.T) {
|
||||
t.Errorf("done with no fingerprint registered, or no node")
|
||||
}
|
||||
}
|
||||
|
||||
// streamsGone is a bus that holds none of the streams asked for; every other part of it is unread.
|
||||
type streamsGone struct{ jetstream.JetStream }
|
||||
|
||||
func (streamsGone) Stream(context.Context, string) (jetstream.Stream, error) {
|
||||
return nil, jetstream.ErrStreamNotFound
|
||||
}
|
||||
|
||||
// Issue 1945 (the alarm of a stream the start deleted): once the declarations' stream is gone from the bus, the
|
||||
// declarations are retired whatever the nodes run now, so no raise, health check, probe or healer expects the stream
|
||||
// or a node's consumer on it again. A node-engine newer than the registered one, or a node asleep on an older have,
|
||||
// made the done test false after the stream went, and the bus's objects named it missing again (2026-10-25 14:52).
|
||||
func TestTheDeclarationsStayRetiredOnceTheirStreamIsGone(t *testing.T) {
|
||||
retired, err := func() (retired bool, err error) {
|
||||
defer func() {
|
||||
if p := recover(); p != nil {
|
||||
err = fmt.Errorf("it read what the nodes run although the stream is gone: %v", p)
|
||||
}
|
||||
}()
|
||||
return declarationsRetiredOn(t.Context(), nil, streamsGone{})
|
||||
}()
|
||||
if err != nil || !retired {
|
||||
t.Fatalf("with no declarations' stream on the bus: retired %v (%v), want retired", retired, err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -29,13 +29,7 @@ type composer struct {
|
||||
type waiter struct {
|
||||
nodes []string
|
||||
grants bool
|
||||
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
|
||||
done chan error
|
||||
}
|
||||
|
||||
func newComposer(compose func(context.Context, []string) map[string]error, grants func(context.Context) error) *composer {
|
||||
@@ -45,23 +39,7 @@ 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 {
|
||||
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)
|
||||
done := make(chan error, 1)
|
||||
c.mu.Lock()
|
||||
for _, n := range nodes {
|
||||
c.pending[n] = true
|
||||
@@ -74,10 +52,10 @@ func (c *composer) composeEach(ctx context.Context, nodes []string, grants bool)
|
||||
default:
|
||||
}
|
||||
select {
|
||||
case r := <-done:
|
||||
return r.failed, r.grants, nil
|
||||
case err := <-done:
|
||||
return err
|
||||
case <-ctx.Done():
|
||||
return nil, nil, ctx.Err()
|
||||
return ctx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
@@ -112,16 +90,16 @@ func (c *composer) run(ctx context.Context) {
|
||||
grantsErr = c.writeGrants(ctx)
|
||||
}
|
||||
for _, w := range waiters {
|
||||
r := passResult{failed: map[string]error{}}
|
||||
var errs []error
|
||||
for _, n := range w.nodes {
|
||||
if err := failed[n]; err != nil {
|
||||
r.failed[n] = err
|
||||
errs = append(errs, err)
|
||||
}
|
||||
}
|
||||
if w.grants {
|
||||
r.grants = grantsErr
|
||||
if w.grants && grantsErr != nil {
|
||||
errs = append(errs, grantsErr)
|
||||
}
|
||||
w.done <- r
|
||||
w.done <- errors.Join(errs...)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -195,7 +195,7 @@ func watchHavesForGiven(ctx context.Context, open *stores) {
|
||||
if at.IsZero() {
|
||||
continue
|
||||
}
|
||||
inBackground(ctx, func() { replaceGiven(ctx, open, node, map[string]time.Time{module: at}) })
|
||||
inBackground(func() { replaceGiven(ctx, open, node, map[string]time.Time{module: at}) })
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
@@ -23,13 +23,10 @@ import (
|
||||
// writes `<module>.<fingerprint hex>` then, naming the accounts it composed for the module, never removed. A module one
|
||||
// of whose accounts has no password yet (the list leaves it out) has no grant until a later send carries it.
|
||||
|
||||
// grantsFrom is the grant of every module at the fingerprint current names for it, naming its accounts that have a
|
||||
// sealed password in hashes; a module whose current build names no fingerprint of the new form has none. A module one
|
||||
// of whose accounts has none is granted its sealed accounts alone, and when it has none, its grant is dropped (drop):
|
||||
// an older grant naming a node whose account is gone would let the assigner want it (issue 1945, rule 6 per node), and
|
||||
// an empty grant names every node. principals are the bus's users as broker.Users composes them.
|
||||
func grantsFrom(principals []broker.Principal, hashes map[string]string, current map[string]string) (
|
||||
map[string]sdkdelivery.ControllerGrant, []string) {
|
||||
// grantsFrom is the grant of every module whose accounts all have a password in hashes, at the fingerprint current names
|
||||
// for it; a module whose current build names no fingerprint of the new form has none. principals are the bus's users
|
||||
// as broker.Users composes them.
|
||||
func grantsFrom(principals []broker.Principal, hashes map[string]string, current map[string]string) map[string]sdkdelivery.ControllerGrant {
|
||||
accounts, missing := map[string][]string{}, map[string]bool{}
|
||||
for _, p := range principals {
|
||||
if p.Module == "" || (p.Kind != broker.KindModule && p.Kind != broker.KindProcess) {
|
||||
@@ -42,9 +39,8 @@ func grantsFrom(principals []broker.Principal, hashes map[string]string, current
|
||||
accounts[p.Module] = append(accounts[p.Module], p.Username())
|
||||
}
|
||||
out := map[string]sdkdelivery.ControllerGrant{}
|
||||
var drop []string
|
||||
for module, fp := range current {
|
||||
if !strings.HasPrefix(fp, sdkdelivery.SourcePrefix2) {
|
||||
if missing[module] || !strings.HasPrefix(fp, sdkdelivery.SourcePrefix2) {
|
||||
continue
|
||||
}
|
||||
key, err := sdkdelivery.ControllerGrantKey(module, fp)
|
||||
@@ -52,15 +48,10 @@ func grantsFrom(principals []broker.Principal, hashes map[string]string, current
|
||||
continue
|
||||
}
|
||||
names := accounts[module]
|
||||
if missing[module] && len(names) == 0 {
|
||||
drop = append(drop, key)
|
||||
continue
|
||||
}
|
||||
sort.Strings(names)
|
||||
out[key] = sdkdelivery.ControllerGrant{Accounts: names}
|
||||
}
|
||||
sort.Strings(drop)
|
||||
return out, drop
|
||||
return out
|
||||
}
|
||||
|
||||
// currentFingerprints is, by module, the fingerprint of the build each registered module is registered at: its
|
||||
@@ -257,27 +248,13 @@ func grantsIssued(ctx context.Context, inv *inventory.Inventory, js jetstream.Je
|
||||
}
|
||||
hashes := make(map[string]string, len(kept))
|
||||
for name, u := range kept {
|
||||
// A grant names an account whose credential is sealed to its node, never one only minted: the assigner wants a
|
||||
// node only once the grant names its account (issue 1945, rule 6 per node).
|
||||
if u.Sealed {
|
||||
hashes[name] = u.PasswordHash
|
||||
}
|
||||
hashes[name] = u.PasswordHash
|
||||
}
|
||||
current, err := currentFingerprints(ctx, inv)
|
||||
if err != nil {
|
||||
return fmt.Errorf("the grants issued are not written: %w", err)
|
||||
}
|
||||
grants, drop := grantsFrom(principals, hashes, current)
|
||||
wrote, err := keepGrants(ctx, kv, grants)
|
||||
for _, key := range drop {
|
||||
if _, gerr := kv.Get(ctx, key); gerr == nil {
|
||||
if derr := kv.Delete(ctx, key); derr != nil {
|
||||
err = errors.Join(err, fmt.Errorf("%s: %w", key, derr))
|
||||
} else {
|
||||
fmt.Printf("%sthe grant %s dropped: none of its accounts is sealed to its node\n", indent, key)
|
||||
}
|
||||
}
|
||||
}
|
||||
wrote, err := keepGrants(ctx, kv, grantsFrom(principals, hashes, current))
|
||||
if wrote > 0 {
|
||||
fmt.Printf("%s%d grant(s) issued written for the assigner\n", indent, wrote)
|
||||
}
|
||||
|
||||
@@ -23,25 +23,12 @@ func TestTheGrantsAListIssues(t *testing.T) {
|
||||
y := broker.Principal{Kind: broker.KindModule, Node: "ace", Module: "mesh-y"}
|
||||
node := broker.Principal{Kind: broker.KindNode, Node: "novox"}
|
||||
hashes := map[string]string{x.Username(): "h1", xp.Username(): "h2", node.Username(): "h3"}
|
||||
// mesh-z is sealed on novox and not on ace (issue 1945: an older grant must not name ace): its grant names novox alone.
|
||||
zn := broker.Principal{Kind: broker.KindModule, Node: "novox", Module: "mesh-z"}
|
||||
za := broker.Principal{Kind: broker.KindModule, Node: "ace", Module: "mesh-z"}
|
||||
hashes[zn.Username()] = "h4"
|
||||
got, drop := grantsFrom([]broker.Principal{x, xp, y, node, zn, za}, hashes,
|
||||
map[string]string{"mesh-x": grantFP('a'), "mesh-y": grantFP('b'), "web": grantFP('c'), "old": "src1:abc",
|
||||
"mesh-z": grantFP('d')})
|
||||
got := grantsFrom([]broker.Principal{x, xp, y, node}, hashes,
|
||||
map[string]string{"mesh-x": grantFP('a'), "mesh-y": grantFP('b'), "web": grantFP('c'), "old": "src1:abc"})
|
||||
xKey, _ := sdkdelivery.ControllerGrantKey("mesh-x", grantFP('a'))
|
||||
yKey, _ := sdkdelivery.ControllerGrantKey("mesh-y", grantFP('b'))
|
||||
zKey, _ := sdkdelivery.ControllerGrantKey("mesh-z", grantFP('d'))
|
||||
webKey, _ := sdkdelivery.ControllerGrantKey("web", grantFP('c'))
|
||||
if len(got) != 3 {
|
||||
t.Fatalf("grants = %v, want mesh-x's, mesh-z's and web's", got)
|
||||
}
|
||||
if g := got[zKey]; len(g.Accounts) != 1 || g.Accounts[0] != zn.Username() {
|
||||
t.Errorf("mesh-z's grant = %+v, want its sealed account alone", g)
|
||||
}
|
||||
if len(drop) != 1 || drop[0] != yKey {
|
||||
t.Errorf("dropped %v, want mesh-y's grant, none of whose accounts is sealed", drop)
|
||||
if len(got) != 2 {
|
||||
t.Fatalf("grants = %v, want mesh-x's and web's", got)
|
||||
}
|
||||
if g := got[xKey]; len(g.Accounts) != 2 || g.Accounts[0] != x.Username() && g.Accounts[1] != x.Username() {
|
||||
t.Errorf("mesh-x's grant = %+v, want both its accounts", g)
|
||||
|
||||
@@ -1,352 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"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/conditions"
|
||||
"github.com/novox/mesh-controller/internal/inventory"
|
||||
)
|
||||
|
||||
// **A kept composition follows its inputs** (novox/hq ADR 0545, issue 1945). The controller composes a module only at
|
||||
// the fingerprint it is registered at, and keeps the compositions of up to five fingerprints (ADR 0539 §4) so that
|
||||
// going back is one want. A kept one was never composed again, so it carried the config of the moment it was made: on
|
||||
// 2026-10-25 a want back to an earlier mesh-build-go on g14 applied its capacity of 0, the value from before it was
|
||||
// set. Now a write of mesh-config's resolved values for a module on a node, heard from that bucket, composes the node
|
||||
// and then every kept fingerprint of that module on that node again, each from its own build's manifest with today's
|
||||
// inputs, writing only those whose composition differs (writeComposed). The writes are read from a durable consumer,
|
||||
// acknowledged once handled, so a change made while no controller ran is delivered to the next one, and a start
|
||||
// composes again only what was written since the last write handled, never every kept composition.
|
||||
|
||||
// configTouched is the node and module a key of mesh-config_resolved (`<node>.<module>.<key>`) names.
|
||||
func configTouched(key string) (node, module string, ok bool) {
|
||||
parts := strings.SplitN(key, ".", 3)
|
||||
if len(parts) != 3 || parts[0] == "" || parts[1] == "" {
|
||||
return "", "", false
|
||||
}
|
||||
return parts[0], parts[1], true
|
||||
}
|
||||
|
||||
// keptKey is one kept composition to compose again.
|
||||
type keptKey struct{ module, fp string }
|
||||
|
||||
// keptToFollow is, of a node's composition keys, each of modules (every module when nil) at a fingerprint other than
|
||||
// the one it is registered at, sorted: the kept compositions. The node's own composition and other nodes' are not.
|
||||
func keptToFollow(keys []string, node string, modules map[string]bool, registered map[string]string) []keptKey {
|
||||
var out []keptKey
|
||||
for _, key := range keys {
|
||||
n, module, fp, isNode, err := sdkdelivery.ParseComposedKey(key)
|
||||
if err != nil || isNode || n != node || (modules != nil && !modules[module]) {
|
||||
continue
|
||||
}
|
||||
if fp == registered[module] || !strings.HasPrefix(fp, sdkdelivery.SourcePrefix2) {
|
||||
continue
|
||||
}
|
||||
out = append(out, keptKey{module, fp})
|
||||
}
|
||||
sort.Slice(out, func(i, j int) bool {
|
||||
if out[i].module != out[j].module {
|
||||
return out[i].module < out[j].module
|
||||
}
|
||||
return out[i].fp < out[j].fp
|
||||
})
|
||||
return out
|
||||
}
|
||||
|
||||
// buildAt is, of a module's builds newest first, the newest that worked at fp and kept its manifest.
|
||||
func buildAt(builds []inventory.Build, fp string) (inventory.Build, bool) {
|
||||
for _, b := range builds {
|
||||
if b.Worked() && b.SourceFingerprint == fp && len(b.Manifest) > 0 {
|
||||
return b, true
|
||||
}
|
||||
}
|
||||
return inventory.Build{}, false
|
||||
}
|
||||
|
||||
// followKept composes again, on node, every kept composition of modules (every module when nil) with today's inputs,
|
||||
// and writes those that differ. It answers the keys written; a fingerprint whose build's manifest is gone is said and
|
||||
// left as it is.
|
||||
func followKept(ctx context.Context, open *stores, node string, modules map[string]bool) ([]string, error) {
|
||||
if _, err := theLease.epoch(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
bus, err := aBus()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer bus.Close()
|
||||
js, err := jetstream.New(bus.Conn())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
inv := open.inventory
|
||||
registered, err := currentFingerprints(ctx, inv)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
at, err := openComposedPlace(ctx, js, bus.Conn(), node)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
lister, err := at.bucket.ListKeys(ctx)
|
||||
if err != nil {
|
||||
if errors.Is(err, jetstream.ErrNoKeysFound) {
|
||||
return nil, nil // empty-on-error: a node with no compositions keeps none to compose again
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
var keys []string
|
||||
for k := range lister.Keys() {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
kept := keptToFollow(keys, node, modules, registered)
|
||||
if len(kept) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
sign, err := controllerSigner(ctx, open)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
upgrades, err := inv.Upgrades(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
at.sign = sign
|
||||
at.policy = func(module string) (string, string, string) {
|
||||
u, ok := upgrades[module]
|
||||
if !ok {
|
||||
return "", "", ""
|
||||
}
|
||||
return u.Policy(), u.From, u.Why
|
||||
}
|
||||
in, err := inputFor(ctx, open)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var wrote []string
|
||||
var errs []error
|
||||
builds := map[string][]inventory.Build{}
|
||||
for _, k := range kept {
|
||||
if ctx.Err() != nil {
|
||||
return wrote, errors.Join(append(errs, ctx.Err())...) // stopped: the rest is the next controller's
|
||||
}
|
||||
if _, read := builds[k.module]; !read {
|
||||
if builds[k.module], err = inv.Builds(ctx, k.module, 20); err != nil {
|
||||
errs = append(errs, fmt.Errorf("%s's builds: %w", k.module, err))
|
||||
continue
|
||||
}
|
||||
}
|
||||
b, found := buildAt(builds[k.module], k.fp)
|
||||
if !found {
|
||||
fmt.Printf("compositions: %s's kept %s on %s has no build whose manifest is kept, and is not composed again\n",
|
||||
k.module, short(k.fp), node)
|
||||
continue
|
||||
}
|
||||
var m catalogue.Manifest
|
||||
if err := json.Unmarshal(b.Manifest, &m); err != nil {
|
||||
errs = append(errs, fmt.Errorf("the manifest of %s's build %s: %w", k.module, b.ID, err))
|
||||
continue
|
||||
}
|
||||
c, err := keptComposition(in.Input, node, k.module, k.fp, m)
|
||||
if err != nil {
|
||||
errs = append(errs, fmt.Errorf("%s at %s on %s: %w", k.module, short(k.fp), node, err))
|
||||
continue
|
||||
}
|
||||
key, _ := sdkdelivery.ComposedKey(node, k.module, k.fp)
|
||||
// One write on the node's keys at a time in this process, held only for the write: composing in memory holds
|
||||
// nothing, so a sweep never keeps a take-in from the node (the review of ADR 0545).
|
||||
unlock := lockComposing(node)
|
||||
w, err := writeComposed(ctx, at, key, c)
|
||||
unlock()
|
||||
if err != nil {
|
||||
errs = append(errs, err)
|
||||
continue
|
||||
}
|
||||
if w {
|
||||
wrote = append(wrote, key)
|
||||
}
|
||||
}
|
||||
if len(wrote) > 0 {
|
||||
fmt.Printf("compositions: %s: kept compositions composed again with today's inputs: %s\n", node, strings.Join(wrote, ", "))
|
||||
}
|
||||
return wrote, errors.Join(errs...)
|
||||
}
|
||||
|
||||
// configFollowGather is how long the config consumer waits after a write for more before it composes: mesh-config writes
|
||||
// a module's keys one by one.
|
||||
var configFollowGather = time.Second
|
||||
|
||||
// configFollowConsumer is the controller's durable consumer of mesh-config's resolved values (ADR 0545 §1, §3).
|
||||
const configFollowConsumer = broker.ConfigFollowConsumer
|
||||
|
||||
// configFollowConfig is that consumer: each resolved key's last value once when it is made, then each write,
|
||||
// acknowledged after the write is handled, so what a stop cut is delivered to the next controller. The first start of
|
||||
// the release that makes it composes again the kept compositions of every module with config, through the same gather,
|
||||
// lock and acknowledgement: mesh-config never writes an unchanged value again, so a value set before this release (g14's
|
||||
// mesh-build-go capacity of 2, against kept compositions carrying 0) would otherwise never reach them. Its deliver
|
||||
// policy is set before its first delivery; an existing durable's cannot be changed.
|
||||
func configFollowConfig() jetstream.ConsumerConfig {
|
||||
return jetstream.ConsumerConfig{Durable: configFollowConsumer, FilterSubject: "$KV." + broker.ConfigResolvedBucket + ".>",
|
||||
DeliverPolicy: jetstream.DeliverLastPerSubjectPolicy, AckPolicy: jetstream.AckExplicitPolicy, AckWait: takeInAckWait,
|
||||
MaxAckPending: takeInPending, Description: "the controller's following of config into kept compositions (novox/hq ADR 0545)"}
|
||||
}
|
||||
|
||||
// kindConfigNotFollowed is the condition of a config consumer that cannot be read: a change of config composes nothing
|
||||
// until it is.
|
||||
const kindConfigNotFollowed = "config-not-followed"
|
||||
|
||||
func configNotFollowedObservation(err error) conditions.Observation {
|
||||
return conditions.Observation{Scope: conditions.ScopeCore, ID: "config-follow", Token: kindConfigNotFollowed,
|
||||
Kind: kindConfigNotFollowed, Severity: conditions.Urgent, Resolver: conditions.ResolverOperator, Source: "config",
|
||||
Summary: fmt.Sprintf("mesh-config's resolved values are not heard, so a change of config composes nothing: %v", err)}
|
||||
}
|
||||
|
||||
// watchConfigFollow reads the writes of mesh-config's resolved values from the controller's durable consumer and, for
|
||||
// each node and module they touch, composes the node and the module's kept compositions there again, until ctx ends
|
||||
// (ADR 0545). A consumer that cannot be made or read raises config-not-followed and is tried again a minute later; it is
|
||||
// cleared once it is read.
|
||||
func watchConfigFollow(ctx context.Context, open *stores) {
|
||||
goServing(ctx, func() {
|
||||
for {
|
||||
err := followConfigOnce(ctx, open)
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
fmt.Printf("config: %v; tried again in a minute\n", err)
|
||||
if keeper := conditionsFrom; keeper != nil {
|
||||
_, _ = keeper.Observe(ctx, configNotFollowedObservation(err))
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-time.After(time.Minute):
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// followConfigOnce reads the consumer until it fails or ctx ends.
|
||||
func followConfigOnce(ctx context.Context, open *stores) error {
|
||||
bus, err := aBus()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer bus.Close()
|
||||
js, err := jetstream.New(bus.Conn())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
c, err := js.CreateOrUpdateConsumer(ctx, "KV_"+broker.ConfigResolvedBucket, configFollowConfig())
|
||||
if err != nil {
|
||||
return fmt.Errorf("the consumer of %s: %w", broker.ConfigResolvedBucket, err)
|
||||
}
|
||||
it, err := c.Messages(jetstream.PullMaxMessages(startTakeIns))
|
||||
if err != nil {
|
||||
return fmt.Errorf("the consumer of %s: %w", broker.ConfigResolvedBucket, err)
|
||||
}
|
||||
defer it.Stop()
|
||||
if keeper := conditionsFrom; keeper != nil {
|
||||
_, _ = keeper.Clear(ctx, configNotFollowedObservation(nil).Key(), "mesh-config's resolved values are heard")
|
||||
}
|
||||
type next struct {
|
||||
m jetstream.Msg
|
||||
err error
|
||||
}
|
||||
got := make(chan next)
|
||||
go func() {
|
||||
for {
|
||||
m, err := it.Next()
|
||||
select {
|
||||
case got <- next{m, err}:
|
||||
case <-ctx.Done():
|
||||
if m != nil {
|
||||
_ = m.Nak()
|
||||
}
|
||||
return
|
||||
}
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
var held []jetstream.Msg
|
||||
touched := map[string]map[string]bool{}
|
||||
var gather <-chan time.Time
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
for _, m := range held {
|
||||
_ = m.Nak() // delivered at once to the next controller
|
||||
}
|
||||
return nil
|
||||
case n := <-got:
|
||||
if n.err != nil {
|
||||
for _, m := range held {
|
||||
_ = m.Nak()
|
||||
}
|
||||
return fmt.Errorf("the consumer of %s stopped: %w", broker.ConfigResolvedBucket, n.err)
|
||||
}
|
||||
held = append(held, n.m)
|
||||
key := strings.TrimPrefix(n.m.Subject(), "$KV."+broker.ConfigResolvedBucket+".")
|
||||
if node, module, ok := configTouched(key); ok {
|
||||
if touched[node] == nil {
|
||||
touched[node] = map[string]bool{}
|
||||
}
|
||||
touched[node][module] = true
|
||||
}
|
||||
if gather == nil {
|
||||
gather = time.After(configFollowGather)
|
||||
}
|
||||
case <-gather:
|
||||
gather = nil
|
||||
now, msgs := touched, held
|
||||
touched, held = map[string]map[string]bool{}, nil
|
||||
forgetResolved() // the next composition reads the values just written, not a read of the last 5 s
|
||||
failed := false
|
||||
for node, modules := range now {
|
||||
if ctx.Err() != nil {
|
||||
failed = true // stopped: the writes are refused below and delivered to the next controller
|
||||
break
|
||||
}
|
||||
if err := composerFor(open).compose(ctx, []string{node}, false); err != nil {
|
||||
fmt.Printf("config: %s is not composed after its config changed: %v\n", node, err)
|
||||
failed = true
|
||||
}
|
||||
if _, err := followKept(ctx, open, node, modules); err != nil {
|
||||
fmt.Printf("config: %s's kept compositions are not composed again: %v\n", node, err)
|
||||
failed = true
|
||||
}
|
||||
}
|
||||
for _, m := range msgs {
|
||||
if failed && ctx.Err() != nil {
|
||||
_ = m.Nak() // stopped: delivered at once to the next controller
|
||||
} else if failed {
|
||||
_ = m.NakWithDelay(doneKeyRetry) // tried again: the bus delivers the writes again
|
||||
} else {
|
||||
acking, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
|
||||
if err := m.DoubleAck(acking); err != nil {
|
||||
fmt.Printf("config: the bus did not take the acknowledgement of %s: %v\n", m.Subject(), err)
|
||||
}
|
||||
cancel()
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// forgetResolved drops the last read of mesh-config's resolved values.
|
||||
func forgetResolved() {
|
||||
theResolved.mu.Lock()
|
||||
theResolved.entries = nil
|
||||
theResolved.mu.Unlock()
|
||||
}
|
||||
@@ -39,11 +39,6 @@ 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")
|
||||
|
||||
@@ -303,15 +298,12 @@ 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 waitsForAWrite("%s at %.8s has no fingerprint to take a build by: %s", t.Module, t.Trunk.Commit, why)
|
||||
return waits("%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 {
|
||||
@@ -327,9 +319,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 waitsForAWrite("%s", k.Says())
|
||||
return waits("%s", k.Says())
|
||||
case len(k.Unreferenced()) > 0:
|
||||
return waitsForAWrite("%s", k.Says())
|
||||
return waits("%s", k.Says())
|
||||
}
|
||||
src, err := keys.Manifest(ctx, k.Build.Manifest)
|
||||
if err != nil {
|
||||
|
||||
@@ -1,186 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"reflect"
|
||||
"slices"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"github.com/nats-io/nats.go"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/broker"
|
||||
"github.com/novox/mesh-controller/internal/link"
|
||||
)
|
||||
|
||||
// **A node's composition issues its modules' memberships** (novox/hq ADR 0160, ADR 0201; issue 249 met again under
|
||||
// issue 1945).
|
||||
//
|
||||
// A membership is what an assignment serves and reaches: its subjects, its seats, and the state, logs, streams and waits
|
||||
// its manifest declares, published last-per-subject on the AssignmentsStream, where the tool runner and every bus process
|
||||
// read and follow their own. A push issued them before it sent a declaration; release 6 removed the push and nothing
|
||||
// issued them after, so a module taken in with a new state, read or stream was granted it on the bus and refused it by its
|
||||
// runtime, which keeps a module to the membership last issued (2026-10-25: mesh-delivery-gate's `conditions`, on novox
|
||||
// from 20:33). Composing a node is now where its memberships are issued: from the same records the bus's grants are
|
||||
// composed from (so what a runtime serves and what its account may are one composition), with the private network's
|
||||
// addresses and what each module receives from the node's composition, before its compositions are written (the code
|
||||
// arriving first was refused its own state, issue 249).
|
||||
//
|
||||
// Only a membership that differs from the one last issued is published, and only over that one, so a pass that changes
|
||||
// nothing re-serves nothing and a newer one issued meanwhile is not overwritten (issues 930, 934).
|
||||
|
||||
// membershipPass is what one composer pass issues every node's memberships from, read once for the pass.
|
||||
type membershipPass struct {
|
||||
records broker.Records
|
||||
where broker.Placements
|
||||
bus link.OverNATS
|
||||
}
|
||||
|
||||
// membershipsFor reads the records, asserts the memberships' stream and every declared bucket and log (a membership
|
||||
// names them, and a module whose state does not exist fails its first read), and judges which machines are root-free
|
||||
// (ADR 0259 §8: a channel's verified sender is composed only from one) when a seat promises one, as a push did.
|
||||
func membershipsFor(ctx context.Context, open *stores, bus link.OverNATS) (*membershipPass, error) {
|
||||
records, err := open.inventory.BusRecords(ctx)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the memberships are not issued: %w", err)
|
||||
}
|
||||
raiser := broker.OnConn(bus.Conn)
|
||||
for _, s := range broker.MeshStreams() {
|
||||
if s.Name != broker.AssignmentsStream {
|
||||
continue
|
||||
}
|
||||
if err := raiser.EnsureStream(s); err != nil {
|
||||
return nil, fmt.Errorf("the memberships are not issued: their stream: %w", err)
|
||||
}
|
||||
}
|
||||
buckets, err := open.inventory.DeclaredBuckets(ctx)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the memberships are not issued: the modules' state: %w", err)
|
||||
}
|
||||
if _, err := broker.RaiseBuckets(raiser, buckets); err != nil {
|
||||
return nil, fmt.Errorf("the memberships are not issued: the modules' state could not be asserted: %w", err)
|
||||
}
|
||||
logs, err := open.inventory.DeclaredLogs(ctx)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the memberships are not issued: the modules' logs: %w", err)
|
||||
}
|
||||
if _, err := broker.RaiseLogs(raiser, logs); err != nil {
|
||||
return nil, fmt.Errorf("the memberships are not issued: the modules' logs could not be asserted: %w", err)
|
||||
}
|
||||
var whys map[string]string
|
||||
if promisesVerifiedSender(records) {
|
||||
records.RootFree, whys = rootVerdictsOf(judgeRootFree(ctx, newRootReader(ctx, open.inventory, bus.Conn),
|
||||
records.Nodes, time.Now()))
|
||||
}
|
||||
where := broker.PlacementsOf(records, records.Interchangeable)
|
||||
for _, line := range droppedVerifiedSenders(records, where, whys) {
|
||||
fmt.Println(line)
|
||||
}
|
||||
return &membershipPass{records: records, where: where, bus: bus}, nil
|
||||
}
|
||||
|
||||
// promisesVerifiedSender says a seat some module holds promises a verified sender: only then is the root-free judgement,
|
||||
// a discovery on the bus, worth its seconds. Pure.
|
||||
func promisesVerifiedSender(records broker.Records) bool {
|
||||
for _, declared := range records.Assigned {
|
||||
for _, d := range declared {
|
||||
for _, s := range d.Holds {
|
||||
if slices.Contains(s.Capabilities, "verified-sender") {
|
||||
return true
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// issue issues node's memberships on the bus, from its composition's addresses and received contributions.
|
||||
func (p *membershipPass) issue(ctx context.Context, node string, declared sendable) error {
|
||||
issued, err := issueComposedMemberships(node, p.records, p.where, declared.Mesh, declared.Received,
|
||||
func(node, module string) ([]byte, uint64, error) {
|
||||
m, err := p.bus.JS.GetLastMsg(broker.AssignmentsStream, broker.MembershipSubject(node, module), nats.Context(ctx))
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
return m.Data, m.Sequence, nil
|
||||
},
|
||||
func(node, module string, body []byte, seq uint64) error {
|
||||
return p.bus.PublishMembershipAfter(ctx, node, module, body, seq)
|
||||
})
|
||||
if len(issued) > 0 {
|
||||
fmt.Printf("memberships: %s: issued %s\n", node, strings.Join(issued, ", "))
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
// issueComposedMemberships composes the membership of every module assigned to node and publishes each one that differs
|
||||
// from the one last issued, over that one (seq 0 when none was: only if none was issued meanwhile). It answers the
|
||||
// modules it issued; every failure is tried past and joined in the error.
|
||||
func issueComposedMemberships(node string, records broker.Records, where broker.Placements, mesh []string,
|
||||
received map[string]map[string][]catalogue.Contribution,
|
||||
last func(node, module string) ([]byte, uint64, error),
|
||||
publish func(node, module string, body []byte, seq uint64) error) ([]string, error) {
|
||||
var issued []string
|
||||
var errs []error
|
||||
for _, d := range records.Assigned[node] {
|
||||
body, err := composedMembership(node, d, where, mesh, received[d.Module])
|
||||
if err != nil {
|
||||
errs = append(errs, fmt.Errorf("%s on %s: its membership: %w", d.Module, node, err))
|
||||
continue
|
||||
}
|
||||
was, seq, err := last(node, d.Module)
|
||||
switch {
|
||||
case errors.Is(err, nats.ErrMsgNotFound):
|
||||
was, seq = nil, 0
|
||||
case err != nil:
|
||||
errs = append(errs, fmt.Errorf("%s on %s: its membership as last issued could not be read: %w", d.Module, node, err))
|
||||
continue
|
||||
}
|
||||
if sameMembership(was, body) {
|
||||
continue
|
||||
}
|
||||
if err := publish(node, d.Module, body, seq); err != nil {
|
||||
errs = append(errs, fmt.Errorf("%s on %s: its membership could not be issued: %w", d.Module, node, err))
|
||||
continue
|
||||
}
|
||||
issued = append(issued, d.Module)
|
||||
}
|
||||
sort.Strings(issued)
|
||||
return issued, errors.Join(errs...)
|
||||
}
|
||||
|
||||
// composedMembership is one assignment's membership as a node's composition issues it: MembershipFor, with the private
|
||||
// network's addresses and what the module receives, by requirement (ADR 0167).
|
||||
func composedMembership(node string, d broker.Declared, where broker.Placements, mesh []string,
|
||||
received map[string][]catalogue.Contribution) ([]byte, error) {
|
||||
m := broker.MembershipFor(node, d, where)
|
||||
m.Mesh = mesh
|
||||
for requirement, given := range received {
|
||||
raw, err := json.Marshal(given)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if m.Receives == nil {
|
||||
m.Receives = map[string]json.RawMessage{}
|
||||
}
|
||||
m.Receives[requirement] = raw
|
||||
}
|
||||
return json.Marshal(m)
|
||||
}
|
||||
|
||||
// sameMembership says two memberships say the same, whatever the order of their fields. Pure.
|
||||
func sameMembership(was, now []byte) bool {
|
||||
if was == nil {
|
||||
return false
|
||||
}
|
||||
var a, b any
|
||||
if json.Unmarshal(was, &a) != nil || json.Unmarshal(now, &b) != nil {
|
||||
return false
|
||||
}
|
||||
return reflect.DeepEqual(a, b)
|
||||
}
|
||||
+18
-118
@@ -9,10 +9,8 @@ import (
|
||||
"net"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"slices"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
@@ -967,101 +965,6 @@ func processPrincipalFor(m catalogue.Manifest, processID string) (broker.Process
|
||||
return broker.ProcessPrincipal{}, false
|
||||
}
|
||||
|
||||
// mintOnlyIfAbsent marks a context whose issue mints only accounts that are not issued (onlyIfAbsent).
|
||||
type mintOnlyIfAbsent struct{}
|
||||
|
||||
// onlyIfAbsent is ctx for an issue that never replaces an issued account: the assign act's, the assignments' watch's
|
||||
// and a registration's, which may run at once for one account (novox/hq ADR 0543, issue 1945).
|
||||
func onlyIfAbsent(ctx context.Context) context.Context {
|
||||
return context.WithValue(ctx, mintOnlyIfAbsent{}, true)
|
||||
}
|
||||
|
||||
// issuedNow collects the accounts an issue minted and sealed, for the line that says it (withIssued).
|
||||
type issuedNow struct {
|
||||
mu sync.Mutex
|
||||
users []string
|
||||
}
|
||||
|
||||
type issuedNowKey struct{}
|
||||
|
||||
// withIssued is ctx collecting the accounts its issues mint and seal.
|
||||
func withIssued(ctx context.Context) (context.Context, *issuedNow) {
|
||||
got := &issuedNow{}
|
||||
return context.WithValue(ctx, issuedNowKey{}, got), got
|
||||
}
|
||||
|
||||
func (g *issuedNow) list() []string {
|
||||
if g == nil {
|
||||
return nil
|
||||
}
|
||||
g.mu.Lock()
|
||||
defer g.mu.Unlock()
|
||||
return slices.Clone(g.users)
|
||||
}
|
||||
|
||||
// accountIssuer is what issuing one account needs of the store.
|
||||
type accountIssuer interface {
|
||||
LockBusAccount(ctx context.Context, username string) (func(), error)
|
||||
BusAccountIssued(ctx context.Context, username string) (minted, sealed bool, err error)
|
||||
MintBusPassword(ctx context.Context, u inventory.BusUser) (string, error)
|
||||
MintBusPasswordUnsealed(ctx context.Context, u inventory.BusUser) (string, error)
|
||||
MarkBusAccountSealed(ctx context.Context, username string) error
|
||||
}
|
||||
|
||||
// issueLockedFor is the longest an issue holds an account's lock: a seal cut by it leaves the account unsealed, issued
|
||||
// again by the next issue.
|
||||
const issueLockedFor = 2 * time.Minute
|
||||
|
||||
// issueAccount mints u's password and seals it to its node with seal. By default it replaces whatever stands (a
|
||||
// rotation: `module issue`). Under onlyIfAbsent it issues only an account that is not issued, a row with no sealed
|
||||
// credential counting as none (issue 1945, live on 2026-10-25 at 20:04: a row whose credential the node no longer held
|
||||
// left ace's mesh-build-go with no credential), under the store's lock on the account, so two issuers never interleave
|
||||
// their look, mint and seal: the hash the bus holds is always the password sealed. A seal that fails leaves the
|
||||
// account unsealed, issued again by the next issue.
|
||||
func issueAccount(ctx context.Context, accounts accountIssuer, u inventory.BusUser, seal func(password string) error) error {
|
||||
only, _ := ctx.Value(mintOnlyIfAbsent{}).(bool)
|
||||
if !only {
|
||||
password, err := accounts.MintBusPassword(ctx, u)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := seal(password); err != nil {
|
||||
return err
|
||||
}
|
||||
} else {
|
||||
// A deadline on the locked section: it holds a connection of the store's pool across the vault and the bus.
|
||||
var cancel context.CancelFunc
|
||||
ctx, cancel = context.WithTimeout(ctx, issueLockedFor)
|
||||
defer cancel()
|
||||
release, err := accounts.LockBusAccount(ctx, u.Username)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer release()
|
||||
if minted, sealed, err := accounts.BusAccountIssued(ctx, u.Username); err != nil {
|
||||
return err
|
||||
} else if minted && sealed {
|
||||
return nil
|
||||
}
|
||||
password, err := accounts.MintBusPasswordUnsealed(ctx, u)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := seal(password); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := accounts.MarkBusAccountSealed(ctx, u.Username); err != nil {
|
||||
return fmt.Errorf("%s is sealed and not marked so, and is issued again by the next issue: %w", u.Username, err)
|
||||
}
|
||||
}
|
||||
if got, _ := ctx.Value(issuedNowKey{}).(*issuedNow); got != nil {
|
||||
got.mu.Lock()
|
||||
got.users = append(got.users, u.Username)
|
||||
got.mu.Unlock()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// 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
|
||||
@@ -1087,10 +990,13 @@ 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()
|
||||
if err := issueAccount(ctx, inv, inventory.BusUser{Username: user, Kind: busKindOf(m.Module), Node: node,
|
||||
Module: m.Module}, func(password string) error {
|
||||
return issueWith(ctx, inv, m, node, busAddress, known, reachable, user, password)
|
||||
}); err != nil {
|
||||
password, err := inv.MintBusPassword(ctx, 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
|
||||
}
|
||||
}
|
||||
@@ -1115,15 +1021,12 @@ 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()
|
||||
return issueAccount(ctx, inv, inventory.BusUser{Username: user, Kind: inventory.BusProcess, Node: node, Module: m.Module},
|
||||
func(password string) error {
|
||||
return sealProcess(ctx, inv, m, pr, node, busAddress, known, reachable, user, password)
|
||||
})
|
||||
}
|
||||
|
||||
// sealProcess seals a process's minted account to its node, hands it to the vault and makes its work-queue consumers.
|
||||
func sealProcess(ctx context.Context, inv *inventory.Inventory, m catalogue.Manifest, pr broker.ProcessPrincipal,
|
||||
node, busAddress string, known broker.Broker, reachable, user, password string) error {
|
||||
password, err := inv.MintBusPassword(ctx, inventory.BusUser{
|
||||
Username: user, Kind: inventory.BusProcess, Node: node, Module: m.Module,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
name := processSecretName(pr.Process)
|
||||
held, err := heldProcessAccount(known, reachable, node, m.Module, user, password)
|
||||
if err != nil {
|
||||
@@ -1188,15 +1091,12 @@ 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()
|
||||
return issueAccount(ctx, inv, inventory.BusUser{Username: user, Kind: inventory.BusProvisioner, Node: node,
|
||||
Module: m.Module}, func(password string) error {
|
||||
return sealProvisionerClient(ctx, inv, m, node, known, reachable, user, password)
|
||||
password, err := inv.MintBusPassword(ctx, inventory.BusUser{
|
||||
Username: user, Kind: inventory.BusProvisioner, Node: node, Module: m.Module,
|
||||
})
|
||||
}
|
||||
|
||||
// sealProvisionerClient seals the provisioner client's minted account to its machine and hands it to the vault.
|
||||
func sealProvisionerClient(ctx context.Context, inv *inventory.Inventory, m catalogue.Manifest, node string,
|
||||
known broker.Broker, reachable, user, password string) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
held, err := json.Marshal(struct {
|
||||
URL string `json:"url"`
|
||||
Fingerprint string `json:"fingerprint,omitempty"`
|
||||
|
||||
@@ -200,20 +200,14 @@ func newModuleObservations(now time.Time) []conditions.Observation {
|
||||
// watchNewModules wakes the take-in of new modules for each module the builder or the indexer writes, until ctx ends, and
|
||||
// looks at every module once now. A bucket not there yet is said: the look at start and a first merge's word still work.
|
||||
func watchNewModules(ctx context.Context, open *stores) {
|
||||
inBackground(ctx, func() {
|
||||
inBackground(func() {
|
||||
takeNewModules(ctx, open)
|
||||
})
|
||||
// And, beside it, the start's pass of release 6 and the take-in of each registered module's new done key (novox/hq
|
||||
// ADR 0539 clauses 5, 7, 10 and 13; composed_start.go, composed_keys.go): the plans' buckets deleted, every node
|
||||
// composed, the replaced controller composed, and from now on each new fingerprint registered, composed and granted.
|
||||
inBackground(ctx, func() { composeAtStart(ctx, open) })
|
||||
inBackground(func() { composeAtStart(ctx, open) })
|
||||
watchDoneKeys(ctx, open)
|
||||
// And a change of config composes the node and the touched modules' kept compositions again, heard from
|
||||
// mesh-config's resolved values (ADR 0545, kept_compositions.go).
|
||||
watchConfigFollow(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)
|
||||
@@ -238,7 +232,7 @@ func watchNewModules(ctx context.Context, open *stores) {
|
||||
// wakeNewModule takes module in when the catalogue does not hold it: never on the watch's own goroutine, which must keep
|
||||
// draining.
|
||||
func wakeNewModule(ctx context.Context, open *stores, module string) {
|
||||
inBackground(ctx, func() {
|
||||
inBackground(func() {
|
||||
entries, err := open.inventory.Catalogued(ctx)
|
||||
if err != nil {
|
||||
return
|
||||
@@ -256,17 +250,14 @@ func wakeNewModule(ctx context.Context, open *stores, module string) {
|
||||
// ends only once the work its mesh started has.
|
||||
var background sync.WaitGroup
|
||||
|
||||
// inBackground runs work beside the caller, counted in background and, under a serving controller, in its serving group,
|
||||
// so serve's stop waits for it and a serve that is stopping starts none (serving_group.go). A variable so a test of what a
|
||||
// walk does around a take-in may run it in the caller's turn and need not wait for it.
|
||||
var inBackground = func(ctx context.Context, work func()) {
|
||||
// inBackground runs work beside the caller, counted in background. A variable so a test of what a walk does around a
|
||||
// take-in may run it in the caller's turn and need not wait for it.
|
||||
var inBackground = func(work func()) {
|
||||
background.Add(1)
|
||||
if !goServing(ctx, func() {
|
||||
go func() {
|
||||
defer background.Done()
|
||||
work()
|
||||
}) {
|
||||
background.Done()
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// newModulesOf are the modules the indexer holds in a repository: what a first merge's word wakes.
|
||||
|
||||
@@ -70,27 +70,6 @@ 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"}
|
||||
}),
|
||||
// Config not followed into the compositions (kept_compositions.go, novox/hq ADR 0545).
|
||||
kindConfigNotFollowed: worded(func(o conditions.Observation) words {
|
||||
return words{Headline: "Changes of settings are not reaching the machines",
|
||||
Explanation: "The mesh could not read the settings as they change, so a changed setting is not prepared for " +
|
||||
"the machines until it can. It tries again every minute; the details say why.",
|
||||
Resolved: "changes of settings reach the machines again"}
|
||||
}),
|
||||
// Machines and their node-engine.
|
||||
"silent": worded(func(o conditions.Observation) words {
|
||||
m := machineOr(o, "a machine")
|
||||
|
||||
@@ -72,8 +72,6 @@ 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
|
||||
@@ -94,18 +92,15 @@ func issueMissingProcessAccounts(ctx context.Context, inv *inventory.Inventory,
|
||||
if err != nil {
|
||||
busAddress = ""
|
||||
}
|
||||
ctx, got := withIssued(ctx)
|
||||
var ids []string
|
||||
for _, pr := range missing {
|
||||
if err := issueProcess(ctx, inv, m, pr, node, busAddress, known, reachable); err != nil {
|
||||
return "", err
|
||||
}
|
||||
}
|
||||
issued := got.list()
|
||||
if len(issued) == 0 {
|
||||
return "", nil // another issuer sealed them meanwhile
|
||||
ids = append(ids, pr.Process)
|
||||
}
|
||||
return fmt.Sprintf("its process account(s) %s issued and sealed to %s, and arrive with the push",
|
||||
strings.Join(issued, ", "), node), nil
|
||||
strings.Join(ids, ", "), node), nil
|
||||
}
|
||||
|
||||
// issueGainedProcessAccounts runs after a module's new definition is registered: on each machine assigned it, the
|
||||
|
||||
@@ -1,9 +1,12 @@
|
||||
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):
|
||||
@@ -22,6 +25,61 @@ 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-<process id> 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-<id> 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-<id> 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) {
|
||||
|
||||
@@ -59,9 +59,6 @@ func serve(ctx context.Context) (err error) {
|
||||
}
|
||||
defer open.Close()
|
||||
inv := open.inventory
|
||||
// Lent to every reader in this process that holds only the inventory, so none opens pools of its own per call
|
||||
// (issues 1945 and 1277: about 120 take-ins at once filled novox's store at 16:08 on 2026-10-25).
|
||||
defer shareStores(open)()
|
||||
// **What this build reads of the store's schema, on record** (novox/hq issue 352): the highest migration
|
||||
// it carries, by its version, so a gate that would put this build back later knows it reads the store
|
||||
// as it is then. A build that does not know its version records nothing, and is never put back.
|
||||
@@ -78,11 +75,11 @@ func serve(ctx context.Context) (err error) {
|
||||
"recorded, and a gate will never put it back\n", RunningBuildVar)
|
||||
}
|
||||
|
||||
// The controller's own identity store, the one its readers share (closed with open), not a second pool.
|
||||
ident, err := open.Identity(ctx)
|
||||
ident, err := openIdentity(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer ident.Close()
|
||||
|
||||
// Established at start rather than on first use. A control plane that cannot sign is one
|
||||
// whose declarations every node correctly refuses, and that should be a startup failure
|
||||
@@ -179,7 +176,6 @@ func serve(ctx context.Context) (err error) {
|
||||
keeper := openServingConditions(ctx, open, bus)
|
||||
if keeper != nil {
|
||||
defer closeConditions(keeper)
|
||||
defer stopRunning() // on an error return, the goroutines end before the keeper closes
|
||||
}
|
||||
|
||||
// The bus's own objects, asserted on every start. **Not created once at genesis**: a stream
|
||||
@@ -248,12 +244,10 @@ func serve(ctx context.Context) (err error) {
|
||||
// next test's `conditions` with "connection closed", issue 1945).
|
||||
handActConn = bus.Conn
|
||||
defer func() { handActConn = nil }()
|
||||
defer stopRunning() // on an error return, the goroutines end before handActConn is put back
|
||||
// And everything else this controller does on the bus for a moment (novox/hq issue 327).
|
||||
servingBus.Store(server.JetStream())
|
||||
// No longer serving: nothing is lent, and dead-letters says it is not read here (novox/hq issue 330).
|
||||
defer servingBus.Store(nil)
|
||||
defer stopRunning() // on an error return, the goroutines end before the bus is taken back
|
||||
// And says when it replaced a value given by hand (novox/hq ADR 0228).
|
||||
givenEvents = bus
|
||||
// And a pull request's merge check, asked when the forge announces its head and said when judged
|
||||
|
||||
@@ -111,9 +111,6 @@ 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.
|
||||
@@ -269,7 +266,6 @@ 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})
|
||||
|
||||
@@ -108,83 +108,14 @@ func TestADroppedVerifiedSenderIsSaidWithItsReason(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// **A node's composition issues every membership that changed** (issue 249 met again under issue 1945): release 6 removed
|
||||
// the push that issued memberships, and mesh-delivery-gate taken in with a new state `conditions` was granted its bucket and
|
||||
// refused it by the tool runner, which reads the membership last issued (2026-10-25, novox). The self-check's reissue
|
||||
// leaves one never issued alone: the composition issues it.
|
||||
func TestAComposedNodeIssuesEveryMembershipThatChanged(t *testing.T) {
|
||||
gate := func(states ...string) broker.Declared {
|
||||
d := broker.Declared{Module: "mesh-delivery-gate"}
|
||||
for _, s := range states {
|
||||
d.State = append(d.State, broker.Bucket{Module: "mesh-delivery-gate", Name: s})
|
||||
}
|
||||
return d
|
||||
}
|
||||
records := func(d broker.Declared) broker.Records {
|
||||
return broker.Records{Nodes: []string{"novox"}, Assigned: map[string][]broker.Declared{"novox": {d}}}
|
||||
}
|
||||
issuedAs := func(d broker.Declared, mesh []string) []byte {
|
||||
body, err := composedMembership("novox", d, broker.PlacementsOf(records(d), nil), mesh, nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return body
|
||||
}
|
||||
newer := []byte(`{"node":"novox","module":"mesh-delivery-gate","newer":true}`)
|
||||
for _, c := range []struct {
|
||||
name string
|
||||
before []byte // the membership last issued, nil for none
|
||||
between bool // another issued meanwhile, after the read
|
||||
issued []string
|
||||
states []string // what the bus holds after, by state name; nil when it must hold newer
|
||||
err string
|
||||
}{
|
||||
{name: "never issued is issued", issued: []string{"mesh-delivery-gate"}, states: []string{"asks", "conditions"}},
|
||||
{name: "a new state is issued", before: issuedAs(gate("asks"), []string{"10.0.0.1"}),
|
||||
issued: []string{"mesh-delivery-gate"}, states: []string{"asks", "conditions"}},
|
||||
{name: "unchanged is not issued again", before: issuedAs(gate("asks", "conditions"), []string{"10.0.0.1"}),
|
||||
states: []string{"asks", "conditions"}},
|
||||
{name: "a newer one meanwhile is not overwritten", before: issuedAs(gate("asks"), []string{"10.0.0.1"}),
|
||||
between: true, err: "wrong last sequence"},
|
||||
} {
|
||||
t.Run(c.name, func(t *testing.T) {
|
||||
r := records(gate("asks", "conditions"))
|
||||
where := broker.PlacementsOf(r, nil)
|
||||
bus := &fakeBus{last: map[string][]byte{}, published: map[string][]byte{}, seq: map[string]uint64{}}
|
||||
if c.before != nil {
|
||||
bus.last["novox/mesh-delivery-gate"], bus.seq["novox/mesh-delivery-gate"] = c.before, 1
|
||||
}
|
||||
if c.between {
|
||||
bus.between = func() { bus.last["novox/mesh-delivery-gate"], bus.seq["novox/mesh-delivery-gate"] = newer, 2 }
|
||||
}
|
||||
if n := reissueVerifiedSenders(r, where, bus.get, bus.put, func(string, ...any) {}); n != 0 {
|
||||
t.Fatalf("the self-check's reissue issued %d", n)
|
||||
}
|
||||
issued, err := issueComposedMemberships("novox", r, where, []string{"10.0.0.1"}, nil, bus.get, bus.put)
|
||||
if c.err == "" && err != nil || c.err != "" && (err == nil || !strings.Contains(err.Error(), c.err)) {
|
||||
t.Fatalf("error %v, want %q", err, c.err)
|
||||
}
|
||||
if strings.Join(issued, ",") != strings.Join(c.issued, ",") {
|
||||
t.Fatalf("issued %v, want %v", issued, c.issued)
|
||||
}
|
||||
if c.states == nil {
|
||||
if string(bus.last["novox/mesh-delivery-gate"]) != string(newer) {
|
||||
t.Fatalf("the newer membership was overwritten: %s", bus.last["novox/mesh-delivery-gate"])
|
||||
}
|
||||
return
|
||||
}
|
||||
var m broker.Membership
|
||||
if err := json.Unmarshal(bus.last["novox/mesh-delivery-gate"], &m); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var names []string
|
||||
for _, s := range m.State {
|
||||
names = append(names, s.Name)
|
||||
}
|
||||
if strings.Join(names, ",") != strings.Join(c.states, ",") || strings.Join(m.Mesh, ",") != "10.0.0.1" {
|
||||
t.Fatalf("the bus holds states %v and mesh %v, want %v", names, m.Mesh, c.states)
|
||||
}
|
||||
})
|
||||
func TestAMembershipNeverIssuedIsLeftToItsOwnSend(t *testing.T) {
|
||||
records, _ := verifiedSenderRecords(map[string]bool{"novox": true})
|
||||
bus := &fakeBus{last: map[string][]byte{}, published: map[string][]byte{}, seq: map[string]uint64{}}
|
||||
var said []string
|
||||
n := reissueVerifiedSenders(records, broker.PlacementsOf(records, nil), bus.get, bus.put,
|
||||
func(f string, a ...any) { said = append(said, f) })
|
||||
if n != 0 || len(said) != 0 {
|
||||
t.Fatalf("issued %d, said %v", n, said)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -29,17 +29,17 @@ func withServingGroup(ctx context.Context) (context.Context, *servingGroup) {
|
||||
|
||||
// goServing runs work beside the caller, counted in the group ctx carries, so serve's stop waits for it. Under a group
|
||||
// whose stop began it runs nothing; under a context with no group (a part run alone, by a test or a verb) it is a plain
|
||||
// goroutine, as before. It says whether it started work.
|
||||
func goServing(ctx context.Context, work func()) bool {
|
||||
// goroutine, as before.
|
||||
func goServing(ctx context.Context, work func()) {
|
||||
g, _ := ctx.Value(servingGroupKey{}).(*servingGroup)
|
||||
if g == nil {
|
||||
go work()
|
||||
return true
|
||||
return
|
||||
}
|
||||
g.mu.Lock()
|
||||
if g.stopped {
|
||||
g.mu.Unlock()
|
||||
return false
|
||||
return
|
||||
}
|
||||
g.wg.Add(1)
|
||||
g.mu.Unlock()
|
||||
@@ -47,7 +47,6 @@ func goServing(ctx context.Context, work func()) bool {
|
||||
defer g.wg.Done()
|
||||
work()
|
||||
}()
|
||||
return true
|
||||
}
|
||||
|
||||
// stop starts nothing more and waits for what was started; the caller ends their context first. A second stop waits
|
||||
|
||||
@@ -1,86 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"github.com/novox/mesh-controller/internal/identity"
|
||||
"github.com/novox/mesh-controller/internal/inventory"
|
||||
"github.com/novox/mesh-controller/internal/licences"
|
||||
)
|
||||
|
||||
// The serving controller's reads of the rest of the mesh open no pool per call: the licences and identity stores are the
|
||||
// controller's own, opened once and shared. On 2026-10-25 at 16:08 each take-in opened two pools of its own
|
||||
// (restOfTheMeshOf), about 120 at once, and novox's store refused every new connection (issues 1945 and 1277).
|
||||
func TestRestOfTheMeshOfOpensNoPoolPerCall(t *testing.T) {
|
||||
wasIdent, wasLic, wasRest := identityOpener, licencesOpener, restOfTheMeshOn
|
||||
t.Cleanup(func() { identityOpener, licencesOpener, restOfTheMeshOn = wasIdent, wasLic, wasRest })
|
||||
var identities, licencesOpened atomic.Int64
|
||||
identityOpener = func(context.Context) (*identity.Identity, error) {
|
||||
identities.Add(1)
|
||||
return &identity.Identity{}, nil
|
||||
}
|
||||
licencesOpener = func(context.Context) (*licences.Licences, error) {
|
||||
licencesOpened.Add(1)
|
||||
return &licences.Licences{}, nil
|
||||
}
|
||||
// The read of the Input reaches both stores, as readComposeInput does.
|
||||
restOfTheMeshOn = func(ctx context.Context, open *stores, _ map[string]catalogue.Manifest, _ string) (catalogue.World, error) {
|
||||
if _, err := open.Licences(ctx); err != nil {
|
||||
return catalogue.World{}, err
|
||||
}
|
||||
if _, err := open.Identity(ctx); err != nil {
|
||||
return catalogue.World{}, err
|
||||
}
|
||||
return catalogue.World{}, nil
|
||||
}
|
||||
inv := &inventory.Inventory{}
|
||||
serving := &stores{inventory: inv}
|
||||
unshare := shareStores(serving)
|
||||
t.Cleanup(unshare)
|
||||
|
||||
var wg sync.WaitGroup
|
||||
for range 120 {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
if _, err := restOfTheMeshOf(context.Background(), inv, nil, ""); err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
if n := licencesOpened.Load(); n != 1 {
|
||||
t.Fatalf("opened the licences store %d times for 120 reads, want once", n)
|
||||
}
|
||||
if n := identities.Load(); n != 1 {
|
||||
t.Fatalf("opened the identity store %d times for 120 reads, want once", n)
|
||||
}
|
||||
if serving.licences == nil || serving.identity == nil {
|
||||
t.Fatal("the reads did not use the serving controller's stores")
|
||||
}
|
||||
}
|
||||
|
||||
// A process that shares no stores (a verb run as its own short process) still opens them for the call and lets go.
|
||||
func TestRestOfTheMeshOfOpensItsOwnWhereNoneAreShared(t *testing.T) {
|
||||
wasIdent, wasLic, wasRest := identityOpener, licencesOpener, restOfTheMeshOn
|
||||
t.Cleanup(func() { identityOpener, licencesOpener, restOfTheMeshOn = wasIdent, wasLic, wasRest })
|
||||
var opened atomic.Int64
|
||||
licencesOpener = func(context.Context) (*licences.Licences, error) { opened.Add(1); return &licences.Licences{}, nil }
|
||||
restOfTheMeshOn = func(ctx context.Context, open *stores, _ map[string]catalogue.Manifest, _ string) (catalogue.World, error) {
|
||||
_, err := open.Licences(ctx)
|
||||
return catalogue.World{}, err
|
||||
}
|
||||
inv := &inventory.Inventory{}
|
||||
for range 2 {
|
||||
if _, err := restOfTheMeshOf(context.Background(), inv, nil, ""); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if n := opened.Load(); n != 2 {
|
||||
t.Fatalf("opened %d times, want once per call", n)
|
||||
}
|
||||
}
|
||||
@@ -67,7 +67,7 @@ func keepAnnouncements(ctx context.Context, js *broker.JetStream, open *stores)
|
||||
return
|
||||
}
|
||||
if changed {
|
||||
inBackground(ctx, func() { composeAnnouncedReads(context.WithoutCancel(ctx), open) })
|
||||
inBackground(func() { composeAnnouncedReads(context.WithoutCancel(ctx), open) })
|
||||
}
|
||||
})
|
||||
if err != nil {
|
||||
|
||||
@@ -151,10 +151,6 @@ 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
|
||||
@@ -475,13 +471,6 @@ 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 {
|
||||
|
||||
@@ -4,7 +4,6 @@ import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
@@ -164,13 +163,6 @@ func openLicences(ctx context.Context) (*licences.Licences, error) {
|
||||
return held, nil
|
||||
}
|
||||
|
||||
// identityOpener and licencesOpener open a store the first time a set of stores needs it; variables so a test counts
|
||||
// the pools opened.
|
||||
var (
|
||||
identityOpener = openIdentity
|
||||
licencesOpener = openLicences
|
||||
)
|
||||
|
||||
// stores is what one command has open.
|
||||
//
|
||||
// **Opened once, not once per machine.** Working out what a machine should be reaches the identity
|
||||
@@ -184,9 +176,6 @@ type stores struct {
|
||||
inventory *inventory.Inventory
|
||||
identity *identity.Identity
|
||||
licences *licences.Licences
|
||||
// mu keeps one opening of each store at a time: a serving controller's stores are shared by readers that run side
|
||||
// by side (sharedStores).
|
||||
mu sync.Mutex
|
||||
}
|
||||
|
||||
// open connects to the inventory, which every command that touches the mesh needs.
|
||||
@@ -200,12 +189,10 @@ func openStores(ctx context.Context) (*stores, error) {
|
||||
|
||||
// Identity is this control plane's own identity context, opened if it has not been.
|
||||
func (h *stores) Identity(ctx context.Context) (*identity.Identity, error) {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
if h.identity != nil {
|
||||
return h.identity, nil
|
||||
}
|
||||
opened, err := identityOpener(ctx)
|
||||
opened, err := openIdentity(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -215,12 +202,10 @@ func (h *stores) Identity(ctx context.Context) (*identity.Identity, error) {
|
||||
|
||||
// Licences is the context holding model access, opened if it has not been.
|
||||
func (h *stores) Licences(ctx context.Context) (*licences.Licences, error) {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
if h.licences != nil {
|
||||
return h.licences, nil
|
||||
}
|
||||
opened, err := licencesOpener(ctx)
|
||||
opened, err := openLicences(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -241,22 +226,3 @@ func (h *stores) Close() {
|
||||
h.inventory.Close()
|
||||
}
|
||||
}
|
||||
|
||||
// sharedStores are the stores a serving controller lends, by its inventory, to every reader in the process that holds
|
||||
// only the inventory (restOfTheMeshOf): one pool of each store for the controller, never one per call (issues 1945 and
|
||||
// 1277, 2026-10-25 16:08: about 120 take-ins opened two pools each and filled novox's store).
|
||||
var sharedStores sync.Map // *inventory.Inventory -> *stores
|
||||
|
||||
// shareStores lends open by its inventory until the function returned is called.
|
||||
func shareStores(open *stores) func() {
|
||||
sharedStores.Store(open.inventory, open)
|
||||
return func() { sharedStores.CompareAndDelete(open.inventory, open) }
|
||||
}
|
||||
|
||||
// sharedStoresOf are the stores lent for inv, if any are.
|
||||
func sharedStoresOf(inv *inventory.Inventory) (*stores, bool) {
|
||||
if open, ok := sharedStores.Load(inv); ok {
|
||||
return open.(*stores), true
|
||||
}
|
||||
return nil, false
|
||||
}
|
||||
|
||||
@@ -1,452 +0,0 @@
|
||||
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")
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -512,9 +512,7 @@ func (w *watchdogs) lostConsumers(ctx context.Context, heard []link.Advisory) (m
|
||||
for _, a := range deleted {
|
||||
_, err := w.js.Context().ConsumerInfo(a.Stream, a.Consumer)
|
||||
switch {
|
||||
// A consumer whose stream is gone is gone with it: the retired declarations' stream took every node's
|
||||
// consumer, and its deletion said each as deleted (issue 1945).
|
||||
case errors.Is(err, nats.ErrConsumerNotFound) || errors.Is(err, nats.ErrStreamNotFound):
|
||||
case errors.Is(err, nats.ErrConsumerNotFound):
|
||||
out[a.Stream+"."+a.Consumer] = want[a.Stream+"."+a.Consumer]
|
||||
case err != nil:
|
||||
return nil, fmt.Errorf("whether %s exists cannot be read: %w", link.ConsumerInWords(a.Stream, a.Consumer), err)
|
||||
|
||||
@@ -165,8 +165,7 @@ func HandPathWrites(nodes []string) ([]string, error) {
|
||||
// DeclarationsStream is the stream of each node's last declaration, `mesh.node.<node>.declare` (issue 107). Release 6
|
||||
// sends none (ADR 0539 clauses 5 and 11): the stream is kept, with each node's consumer, only so a node away during the
|
||||
// switch finds its last declaration and with it the new node-engine; the controller deletes it once every node's have
|
||||
// names the new node-engine, and from then on neither asserts it nor any node's consumer on it; nor makes it again once
|
||||
// it is gone from the bus for any reason, since a stream made anew would hold no node's last declaration (issue 1945).
|
||||
// names the new node-engine, and from then on neither asserts it nor any node's consumer on it.
|
||||
const DeclarationsStream = "NODES"
|
||||
|
||||
// RaiseAfterDeclarations is Raise once the declarations' stream is retired: the mesh's streams and consumers but the
|
||||
@@ -182,15 +181,3 @@ 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}
|
||||
|
||||
// ConfigFollowConsumer is the durable consumer the controller follows mesh-config's resolved values with, on that
|
||||
// bucket's stream (novox/hq ADR 0545).
|
||||
const ConfigFollowConsumer = "mesh-controller-config-follow"
|
||||
|
||||
@@ -432,16 +432,6 @@ 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 " +
|
||||
@@ -717,11 +707,6 @@ 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
|
||||
@@ -1273,11 +1258,6 @@ func PermissionsFor(p Principal) (Permissions, error) {
|
||||
pub = append(pub, "mesh.seat."+v.Seat+"."+CallKind+"."+v.Verb+"."+CallerToken(p.Username()))
|
||||
}
|
||||
}
|
||||
// And its consumer of mesh-config's resolved values, acknowledged once a change is composed (novox/hq ADR 0545):
|
||||
// where mesh-config serves, as its verb, so genesis grants nothing of mesh-config.
|
||||
if len(p.AsksTheConfig) > 0 {
|
||||
pub = append(pub, "$JS.ACK.KV_"+ConfigResolvedBucket+"."+ConfigFollowConsumer+".>")
|
||||
}
|
||||
}
|
||||
|
||||
sort.Strings(pub)
|
||||
|
||||
+1
-1
@@ -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.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"] }
|
||||
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"] }
|
||||
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" }
|
||||
} }
|
||||
|
||||
@@ -7,7 +7,6 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"golang.org/x/crypto/bcrypt"
|
||||
@@ -53,8 +52,6 @@ type BusUser struct {
|
||||
// PasswordHash is what the composed file carries. The plaintext is returned once, by Mint, and
|
||||
// then exists only where it was sealed.
|
||||
PasswordHash string
|
||||
// Sealed says its credential reached its node: an issue's mint is not sealed until then (ADR 0543 §2).
|
||||
Sealed bool
|
||||
}
|
||||
|
||||
// The kinds of bus user the mesh records. The same words the composer uses, so a row and a
|
||||
@@ -103,126 +100,21 @@ func (i *Inventory) RecordBusPassword(ctx context.Context, u BusUser, password s
|
||||
return i.writeBusUser(ctx, u, string(hash))
|
||||
}
|
||||
|
||||
// writeBusUser is the row, whoever chose the password, taken as sealed.
|
||||
// writeBusUser is the row, whoever chose the password.
|
||||
func (i *Inventory) writeBusUser(ctx context.Context, u BusUser, hash string) error {
|
||||
return i.writeBusUserSealed(ctx, u, hash, true)
|
||||
}
|
||||
|
||||
// writeBusUserSealed is the row, sealed now or not yet.
|
||||
func (i *Inventory) writeBusUserSealed(ctx context.Context, u BusUser, hash string, sealed bool) error {
|
||||
if _, err := i.store.Pool().Exec(ctx,
|
||||
`insert into bus_user (username, kind, node, module, password_hash, sealed_at)
|
||||
values ($1, $2, $3, $4, $5, case when $6 then now() end)
|
||||
`insert into bus_user (username, kind, node, module, password_hash)
|
||||
values ($1, $2, $3, $4, $5)
|
||||
on conflict (username) do update
|
||||
set kind = excluded.kind, node = excluded.node, module = excluded.module,
|
||||
password_hash = excluded.password_hash, minted_at = now(), sealed_at = excluded.sealed_at`,
|
||||
u.Username, u.Kind, u.Node, u.Module, hash, sealed); err != nil {
|
||||
password_hash = excluded.password_hash, minted_at = now()`,
|
||||
u.Username, u.Kind, u.Node, u.Module, hash); err != nil {
|
||||
return fmt.Errorf("cannot record the bus user %s: %w", u.Username, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// LockBusAccount holds the store's lock on one account until the answered release is called, so two issuers of it
|
||||
// (the assign act, the assignments' watch, a registration's process accounts) never interleave their look, mint and
|
||||
// seal: the hash the bus holds is always the password sealed to the node (novox/hq ADR 0543 §2, issue 1945).
|
||||
func (i *Inventory) LockBusAccount(ctx context.Context, username string) (func(), error) {
|
||||
conn, err := i.store.Pool().Acquire(ctx)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the lock on %s cannot be taken: %w", username, err)
|
||||
}
|
||||
if _, err := conn.Exec(ctx, `select pg_advisory_lock(hashtext($1))`, "bus_user/"+username); err != nil {
|
||||
conn.Release()
|
||||
return nil, fmt.Errorf("the lock on %s cannot be taken: %w", username, err)
|
||||
}
|
||||
return func() {
|
||||
_, _ = conn.Exec(context.WithoutCancel(ctx), `select pg_advisory_unlock(hashtext($1))`, "bus_user/"+username)
|
||||
conn.Release()
|
||||
}, nil
|
||||
}
|
||||
|
||||
// ForgetBusAccountsOf forgets the bus accounts of module on node, and the credentials sealed to the node for them (its
|
||||
// `broker`, each process's `broker-<id>`, and with nodeOwned its provisioner client's): a module that no longer resolves
|
||||
// on a node keeps no account there, so an assignment again mints and seals afresh, never trusting a row whose credential
|
||||
// the node no longer holds (issue 1945, live on 2026-10-25 at 20:04: ace's mesh-build-go rows from 17:19 outlived its
|
||||
// unassignment, nothing was issued again, and the module found no credential). The provisioner client's account is the
|
||||
// node's (`provisioner.<node>`), forgotten only with nodeOwned. Each account's lock is held across its delete, so an issue
|
||||
// in flight never writes a row back after it. It answers the accounts forgotten.
|
||||
func (i *Inventory) ForgetBusAccountsOf(ctx context.Context, node, module string, nodeOwned bool) ([]string, error) {
|
||||
if node == "" || module == "" {
|
||||
return nil, errors.New("forgetting the bus accounts of no node or no module would forget others' accounts")
|
||||
}
|
||||
rows, err := i.store.Pool().Query(ctx,
|
||||
`select username from bus_user where node = $1 and module = $2 and kind <> $3 and ($4 or kind <> $5)
|
||||
order by username`, node, module, BusController, nodeOwned, BusProvisioner)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var names []string
|
||||
for rows.Next() {
|
||||
var n string
|
||||
if err := rows.Scan(&n); err != nil {
|
||||
rows.Close()
|
||||
return nil, err
|
||||
}
|
||||
names = append(names, n)
|
||||
}
|
||||
rows.Close()
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for _, n := range names { // in order, one lock at a time per issuer: no two hold each other's
|
||||
release, err := i.LockBusAccount(ctx, n)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer release()
|
||||
}
|
||||
tx, err := i.store.Pool().Begin(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer func() { _ = tx.Rollback(context.WithoutCancel(ctx)) }()
|
||||
if _, err := tx.Exec(ctx, `delete from bus_user where username = any($1)`, names); err != nil {
|
||||
return nil, fmt.Errorf("the bus accounts of %s on %s are not forgotten: %w", module, node, err)
|
||||
}
|
||||
if _, err := tx.Exec(ctx,
|
||||
`delete from module_secret s using node n
|
||||
where s.node = n.id and n.name = $1 and s.module = $2
|
||||
and (s.name = 'broker' or s.name like 'broker-%' or ($3 and s.name = 'provisioner-broker'))`,
|
||||
node, module, nodeOwned); err != nil {
|
||||
return nil, fmt.Errorf("the bus credentials of %s on %s are not forgotten: %w", module, node, err)
|
||||
}
|
||||
return names, tx.Commit(ctx)
|
||||
}
|
||||
|
||||
// MintBusPasswordUnsealed mints u's password, replacing whatever stands, with no seal yet: the account is not issued
|
||||
// until MarkBusAccountSealed says its credential reached the node (ADR 0543 §2).
|
||||
func (i *Inventory) MintBusPasswordUnsealed(ctx context.Context, u BusUser) (string, error) {
|
||||
return i.mintBusPassword(ctx, u, false)
|
||||
}
|
||||
|
||||
// MarkBusAccountSealed says the credential last minted for username is sealed to its node.
|
||||
func (i *Inventory) MarkBusAccountSealed(ctx context.Context, username string) error {
|
||||
_, err := i.store.Pool().Exec(ctx, `update bus_user set sealed_at = now() where username = $1`, username)
|
||||
return err
|
||||
}
|
||||
|
||||
// BusAccountIssued says whether an account is minted, and whether its credential is sealed to its node.
|
||||
func (i *Inventory) BusAccountIssued(ctx context.Context, username string) (minted, sealed bool, err error) {
|
||||
var at *time.Time
|
||||
err = i.store.Pool().QueryRow(ctx, `select sealed_at from bus_user where username = $1`, username).Scan(&at)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return false, false, nil
|
||||
}
|
||||
return err == nil, err == nil && at != nil, err
|
||||
}
|
||||
|
||||
func (i *Inventory) MintBusPassword(ctx context.Context, u BusUser) (string, error) {
|
||||
return i.mintBusPassword(ctx, u, true)
|
||||
}
|
||||
|
||||
// mintBusPassword mints, marked sealed when the caller seals it itself in the same act (every mint but an issue's).
|
||||
func (i *Inventory) mintBusPassword(ctx context.Context, u BusUser, sealed bool) (string, error) {
|
||||
if u.Username == "" || u.Kind == "" {
|
||||
return "", errors.New("a bus user needs a username and a kind")
|
||||
}
|
||||
@@ -240,7 +132,7 @@ func (i *Inventory) mintBusPassword(ctx context.Context, u BusUser, sealed bool)
|
||||
return "", fmt.Errorf("cannot hash a bus password: %w", err)
|
||||
}
|
||||
|
||||
if err := i.writeBusUserSealed(ctx, u, string(hash), sealed); err != nil {
|
||||
if err := i.writeBusUser(ctx, u, string(hash)); err != nil {
|
||||
return "", err
|
||||
}
|
||||
return password, nil
|
||||
@@ -253,7 +145,7 @@ func (i *Inventory) mintBusPassword(ctx context.Context, u BusUser, sealed bool)
|
||||
// password, and the composer refuses it rather than writing a user anybody is.
|
||||
func (i *Inventory) BusUsers(ctx context.Context) (map[string]BusUser, error) {
|
||||
rows, err := i.store.Pool().Query(ctx,
|
||||
`select username, kind, node, module, password_hash, sealed_at is not null from bus_user order by username`)
|
||||
`select username, kind, node, module, password_hash from bus_user order by username`)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -261,7 +153,7 @@ func (i *Inventory) BusUsers(ctx context.Context) (map[string]BusUser, error) {
|
||||
out := map[string]BusUser{}
|
||||
for rows.Next() {
|
||||
var u BusUser
|
||||
if err := rows.Scan(&u.Username, &u.Kind, &u.Node, &u.Module, &u.PasswordHash, &u.Sealed); err != nil {
|
||||
if err := rows.Scan(&u.Username, &u.Kind, &u.Node, &u.Module, &u.PasswordHash); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out[u.Username] = u
|
||||
|
||||
@@ -4,9 +4,7 @@ package inventory
|
||||
|
||||
import (
|
||||
"context"
|
||||
"slices"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"golang.org/x/crypto/bcrypt"
|
||||
)
|
||||
@@ -35,7 +33,7 @@ func TestABusPasswordIsReturnedOnceAndOnlyItsHashIsKept(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestMintingAgainRotatesAndMintingIfAbsentHasOneWinner(t *testing.T) {
|
||||
func TestMintingAgainRotatesRatherThanAddsAUser(t *testing.T) {
|
||||
inv := ForTest(t)
|
||||
ctx := context.Background()
|
||||
|
||||
@@ -64,56 +62,6 @@ func TestMintingAgainRotatesAndMintingIfAbsentHasOneWinner(t *testing.T) {
|
||||
if bcrypt.CompareHashAndPassword([]byte(hash), []byte(first)) == nil {
|
||||
t.Fatal("the previous password still verifies, so a rotation revoked nothing")
|
||||
}
|
||||
|
||||
// **An issue's mint is not issued until its seal is marked, and forgotten with its assignment** (ADR 0543 §2, issue
|
||||
// 1945, live on 2026-10-25 at 20:04): a row whose credential the node does not hold is not issued.
|
||||
if minted, sealed, err := inv.BusAccountIssued(ctx, "one.shop"); err != nil || !minted || !sealed {
|
||||
t.Fatalf("a mint that seals in its own act: minted %v sealed %v (%v)", minted, sealed, err)
|
||||
}
|
||||
if _, err := inv.MintBusPasswordUnsealed(ctx, aBusUser("racer")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if minted, sealed, err := inv.BusAccountIssued(ctx, "one.racer"); err != nil || !minted || sealed {
|
||||
t.Fatalf("an issue's mint before its seal: minted %v sealed %v (%v)", minted, sealed, err)
|
||||
}
|
||||
if names, err := inv.BusMinted(ctx); err != nil || slices.Contains(names, "one.racer") {
|
||||
t.Fatalf("an account not sealed counts as minted: %v (%v)", names, err)
|
||||
}
|
||||
if err := inv.MarkBusAccountSealed(ctx, "one.racer"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, sealed, _ := inv.BusAccountIssued(ctx, "one.racer"); !sealed {
|
||||
t.Fatal("an account marked sealed is not")
|
||||
}
|
||||
// The lock on an account holds a second issuer until the first lets go.
|
||||
release, err := inv.LockBusAccount(ctx, "one.racer")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
waiting := make(chan struct{})
|
||||
go func() {
|
||||
again, err := inv.LockBusAccount(ctx, "one.racer")
|
||||
if err == nil {
|
||||
again()
|
||||
}
|
||||
close(waiting)
|
||||
}()
|
||||
select {
|
||||
case <-waiting:
|
||||
t.Fatal("a second issuer took the lock the first held")
|
||||
case <-time.After(200 * time.Millisecond):
|
||||
}
|
||||
release()
|
||||
<-waiting
|
||||
if _, err := inv.ForgetBusAccountsOf(ctx, "one", "racer", true); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if minted, _, _ := inv.BusAccountIssued(ctx, "one.racer"); minted {
|
||||
t.Fatal("an unassigned module's account was kept")
|
||||
}
|
||||
if minted, _, _ := inv.BusAccountIssued(ctx, "one.shop"); !minted {
|
||||
t.Fatal("forgetting one module's accounts forgot another's")
|
||||
}
|
||||
}
|
||||
|
||||
func TestForgettingANodeTakesItsBusUsersWithIt(t *testing.T) {
|
||||
|
||||
@@ -210,21 +210,6 @@ 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)
|
||||
|
||||
@@ -361,13 +361,3 @@ 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)
|
||||
}
|
||||
|
||||
@@ -1,8 +0,0 @@
|
||||
-- A bus account says when the credential minted for it was sealed to its node (novox/hq ADR 0543 §2, issue 1945). The
|
||||
-- assign act and the assignments' watch issue an account only when it is not issued: a row with no sealed credential
|
||||
-- is not issued. On 2026-10-25 at 20:04 ace's mesh-build-go had a row minted at 17:19 whose credential the node no
|
||||
-- longer held; minting nothing, the mesh sealed nothing, and the module crash-looped for its missing credential. Null
|
||||
-- while an issue mints and seals; set once the credential is handed over. Every account standing before this migration
|
||||
-- is taken as sealed: each works or worked, and its own issue sealed it.
|
||||
alter table bus_user add column sealed_at timestamptz;
|
||||
update bus_user set sealed_at = minted_at;
|
||||
@@ -226,10 +226,8 @@ func (i *Inventory) BusMinted(ctx context.Context) ([]string, error) {
|
||||
return nil, err
|
||||
}
|
||||
out := make([]string, 0, len(kept))
|
||||
for name, u := range kept {
|
||||
if u.Sealed { // an account whose credential is not sealed to its node is not issued (ADR 0543 §2)
|
||||
out = append(out, name)
|
||||
}
|
||||
for name := range kept {
|
||||
out = append(out, name)
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out, nil
|
||||
|
||||
@@ -1,82 +0,0 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgconn"
|
||||
)
|
||||
|
||||
// A full store (SQLSTATE 53300, "too many clients already") is not asked again every 250 ms: Ready backs off,
|
||||
// doubling its wait up to a bound, and says once that the store is full (issues 1945 and 1277, 2026-10-25 16:08: the
|
||||
// controller's retries kept novox's store full).
|
||||
func TestReadyBacksOffOnAFullStore(t *testing.T) {
|
||||
wasPing, wasWait, wasSay := pingStore, readyWait, readySay
|
||||
t.Cleanup(func() { pingStore, readyWait, readySay = wasPing, wasWait, wasSay })
|
||||
asked := 0
|
||||
pingStore = func(context.Context, *Store) error {
|
||||
asked++
|
||||
if asked <= 8 {
|
||||
return &pgconn.PgError{Code: "53300", Message: "sorry, too many clients already"}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
var waits []time.Duration
|
||||
readyWait = func(_ context.Context, d time.Duration) error { waits = append(waits, d); return nil }
|
||||
var said []string
|
||||
readySay = func(line string) { said = append(said, line) }
|
||||
|
||||
if err := (&Store{context: "example"}).Ready(context.Background(), time.Hour); err != nil {
|
||||
t.Fatalf("Ready: %v", err)
|
||||
}
|
||||
if len(waits) != 8 {
|
||||
t.Fatalf("waited %d times, want 8: %v", len(waits), waits)
|
||||
}
|
||||
for i := 1; i < len(waits); i++ {
|
||||
if waits[i] < waits[i-1] {
|
||||
t.Fatalf("wait %d (%s) is shorter than the one before (%s): %v", i, waits[i], waits[i-1], waits)
|
||||
}
|
||||
}
|
||||
if waits[1] <= waits[0] {
|
||||
t.Fatalf("Ready does not back off on a full store: %v", waits)
|
||||
}
|
||||
if most := waits[len(waits)-1]; most > fullStoreMostWait {
|
||||
t.Fatalf("the wait grew past its bound %s: %v", fullStoreMostWait, waits)
|
||||
}
|
||||
full := 0
|
||||
for _, l := range said {
|
||||
if strings.Contains(l, "full") {
|
||||
full++
|
||||
}
|
||||
}
|
||||
if full != 1 {
|
||||
t.Fatalf("said that the store is full %d times, want once: %q", full, said)
|
||||
}
|
||||
}
|
||||
|
||||
// A store that does not answer for another reason (not up yet) is still asked again every 250 ms.
|
||||
func TestReadyWaitsShortlyOnAStoreNotUp(t *testing.T) {
|
||||
wasPing, wasWait, wasSay := pingStore, readyWait, readySay
|
||||
t.Cleanup(func() { pingStore, readyWait, readySay = wasPing, wasWait, wasSay })
|
||||
asked := 0
|
||||
pingStore = func(context.Context, *Store) error {
|
||||
asked++
|
||||
if asked <= 3 {
|
||||
return context.DeadlineExceeded
|
||||
}
|
||||
return nil
|
||||
}
|
||||
var waits []time.Duration
|
||||
readyWait = func(_ context.Context, d time.Duration) error { waits = append(waits, d); return nil }
|
||||
readySay = func(string) {}
|
||||
if err := (&Store{context: "example"}).Ready(context.Background(), time.Hour); err != nil {
|
||||
t.Fatalf("Ready: %v", err)
|
||||
}
|
||||
for _, w := range waits {
|
||||
if w != 250*time.Millisecond {
|
||||
t.Fatalf("waits %v, want 250ms each", waits)
|
||||
}
|
||||
}
|
||||
}
|
||||
+6
-47
@@ -24,7 +24,6 @@ import (
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgconn"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/envfile"
|
||||
@@ -214,7 +213,7 @@ func (s *Store) Pool() *pgxpool.Pool { return s.pool }
|
||||
|
||||
// Close releases the connections.
|
||||
func (s *Store) Close() {
|
||||
if s != nil && s.pool != nil {
|
||||
if s.pool != nil {
|
||||
s.pool.Close()
|
||||
}
|
||||
}
|
||||
@@ -228,10 +227,8 @@ func (s *Store) Close() {
|
||||
func (s *Store) Ready(ctx context.Context, within time.Duration) error {
|
||||
deadline := time.Now().Add(within)
|
||||
var last error
|
||||
wait := 250 * time.Millisecond
|
||||
saidFull := false
|
||||
for {
|
||||
err := pingStore(ctx, s)
|
||||
err := s.pool.Ping(ctx)
|
||||
if err == nil {
|
||||
return nil
|
||||
}
|
||||
@@ -243,48 +240,10 @@ func (s *Store) Ready(ctx context.Context, within time.Duration) error {
|
||||
return fmt.Errorf(
|
||||
"the %s store did not answer within %s: %w", s.context, within, last)
|
||||
}
|
||||
// A full store (SQLSTATE 53300) is not asked again every 250 ms: each ask is one more connection it refuses
|
||||
// while it is full. The wait doubles up to fullStoreMostWait, and the journal says once why (issues 1945 and
|
||||
// 1277, 2026-10-25 16:08: the controller's own retries kept the store full).
|
||||
if isFull(err) {
|
||||
if !saidFull {
|
||||
readySay(fmt.Sprintf("the %s store is full (too many clients already); asked again at a growing "+
|
||||
"interval, at most every %s, for at most %s in all", s.context, fullStoreMostWait, within))
|
||||
saidFull = true
|
||||
}
|
||||
} else {
|
||||
wait = 250 * time.Millisecond
|
||||
}
|
||||
if err := readyWait(ctx, wait); err != nil {
|
||||
return errors.Join(err, last)
|
||||
}
|
||||
if isFull(err) {
|
||||
wait = min(wait*2, fullStoreMostWait)
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return errors.Join(ctx.Err(), last)
|
||||
case <-time.After(250 * time.Millisecond):
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// pingStore asks the store once whether it answers; a variable so a test needs no database.
|
||||
var pingStore = func(ctx context.Context, s *Store) error { return s.pool.Ping(ctx) }
|
||||
|
||||
// readyWait waits d or until ctx ends; a variable so a test waits in no real time.
|
||||
var readyWait = func(ctx context.Context, d time.Duration) error {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case <-time.After(d):
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// readySay says a line of Ready's wait in the journal; a variable so a test reads it.
|
||||
// isFull says whether err is the store refusing a connection because it is full (SQLSTATE 53300).
|
||||
func isFull(err error) bool {
|
||||
var pg *pgconn.PgError
|
||||
return errors.As(err, &pg) && pg.Code == "53300"
|
||||
}
|
||||
|
||||
// fullStoreMostWait bounds the wait between two asks of a full store.
|
||||
const fullStoreMostWait = 5 * time.Second
|
||||
|
||||
var readySay = func(line string) { fmt.Fprintln(os.Stderr, line) }
|
||||
|
||||
Reference in New Issue
Block a user