Compare commits

..
Author SHA1 Message Date
jschoubben ef10a00f6d Share one reading of the forge's address across take-ins
Each take-in made a fresh set of source forms (newSourceForms), so each
read the whole mesh once to learn the forge's address: about 120 reads
at once at 16:08, each one more load on a store that was already full.
The take-ins of an inventory now share one reading, made again once it
is a minute old so a moved forge is read within that bound. A base that
could not be read is not kept. The forms' read takes a lock, since the
take-ins that share it run side by side.

Issues 1945 and 1277.
2026-10-25 16:24:52 +01:00
32 changed files with 245 additions and 1638 deletions
-2
View File
@@ -399,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 {
+3 -4
View File
@@ -12,7 +12,6 @@ import (
"sort"
"strings"
"sync"
"sync/atomic"
"time"
"github.com/nats-io/nats.go"
@@ -28,7 +27,7 @@ import (
)
// askerFrom is the serving controller's asker; nil in any other process.
var askerFrom atomic.Pointer[asker]
var askerFrom *asker
// askWithin is how long a verb a warrant chose is given to answer.
const askWithin = time.Minute
@@ -574,8 +573,8 @@ func startAsking(ctx context.Context, open *stores, server *link.Server, conn *n
fmt.Printf("the operator's answers cannot be heard, so nothing is asked: %v\n", err)
return
}
askerFrom.Store(a)
goServing(ctx, func() { a.keep(ctx) })
askerFrom = a
go a.keep(ctx)
}
// servingAsker is the serving controller's asker, wired to the mesh: what it reads, publishes, performs and
-387
View File
@@ -1,387 +0,0 @@
package main
import (
"context"
"encoding/json"
"errors"
"fmt"
"strings"
"sync"
"time"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"github.com/novox/mesh-controller/internal/broker"
"github.com/novox/mesh-controller/internal/chain"
"github.com/novox/mesh-controller/internal/inventory"
)
// An assignment that stands issues its node's bus accounts and the module's grant (novox/hq ADR 0543, issue 1945; a
// step of research 072's grants: the memberships and credentials heard from placement, not from a send or a build).
//
// ADR 0539 clause 7 issues a module's credential and grant when its build is done. A module already built and assigned
// to one more node never has a build done again, so nothing issued that node's account: on 2026-10-25 at 17:19
// mesh-build-node and mesh-build-go were assigned to ace, composed for it, and the bus's user list never named
// ace.mesh-build-node nor ace.mesh-build-go; one runtime found no credential, the other was refused by the bus, both
// restarting every 5 s. The assign act composes the assigned node alone (composedAfterAct): never the node holding the
// bus, never the grant.
//
// So the write of an assignments key is the event. When a key `<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
// catalogue reads the catalogue once, for every assignment a look issues for.
catalogue func(ctx context.Context) (map[string]catalogue.Manifest, error)
// issue issues the module's accounts on the node that are not minted, from the catalogue read; issued says it
// minted one, line what it did.
issue func(ctx context.Context, shelf map[string]catalogue.Manifest, node, module string) (issued bool, line string, err error)
// busHolder is the node holding the bus, "" for none.
busHolder func(ctx context.Context) (string, error)
// compose writes the nodes' compositions; grants writes the grants issued.
compose func(ctx context.Context, nodes []string) error
grants func(ctx context.Context) error
}
// settle issues what the standing assignment of module on node lacks. fresh says the assignment just came to stand: the
// bus's node is composed and the grants written even when every account was minted before (a module assigned again
// keeps its account, and the bus's user list left it out while it was unassigned). It answers what it did, "" for
// nothing.
func (s assignmentAccountSteps) settle(ctx context.Context, node, module string) (string, error) {
if module == controllerModule {
// The controller's own module is issued at the terminal alone: its `broker` secret is the credential it is
// using (rollout.go), and it is assigned and unassigned there only (refusedMovingTheController).
return "", nil
}
if !s.holds(ctx) {
return "", errNotHolding
}
shelf, err := s.catalogue(ctx)
if err != nil {
return "", fmt.Errorf("the catalogue cannot be read: %w", err)
}
_, line, err := s.issue(ctx, shelf, node, module)
if err != nil {
return "", fmt.Errorf("its bus accounts on %s are not issued: %w", node, err)
}
said, err := s.cover(ctx)
prefix := module + " on " + node + ": "
if line != "" {
prefix += line + "; "
}
return prefix + said, err
}
// look is the start's look: the accounts every standing assignment lacks, issued one after another from one reading of
// the catalogue, then the bus's node composed and the grants written once, when it looked at any assignment, whatever
// it issued (an assignment that came to stand while no controller heard it may have its accounts and still lack its
// place in the bus's user list or its grant). It answers what it did.
func (s assignmentAccountSteps) look(ctx context.Context, standing []chain.Assignment) (string, error) {
var looked []chain.Assignment
for _, a := range standing {
if a.Module != controllerModule {
looked = append(looked, a)
}
}
if len(looked) == 0 {
return "", nil
}
if !s.holds(ctx) {
return "", errNotHolding
}
shelf, err := s.catalogue(ctx)
if err != nil {
return "", fmt.Errorf("the catalogue cannot be read: %w", err)
}
var lines []string
for _, a := range looked {
if ctx.Err() != nil {
return "", ctx.Err()
}
if _, line, err := s.issue(ctx, shelf, a.Node, a.Module); err != nil {
return "", fmt.Errorf("%s's bus accounts on %s are not issued: %w", a.Module, a.Node, err)
} else if line != "" {
lines = append(lines, a.Module+" on "+a.Node+": "+line)
}
}
said, err := s.cover(ctx)
lines = append(lines, fmt.Sprintf("%d standing assignment(s) looked at; %s", len(looked), said))
return strings.Join(lines, "\n"), err
}
// errNotHolding is a controller that does not hold the lease: it issues nothing, and tries again until it holds it.
var errNotHolding = errors.New("this controller does not hold the lease")
// cover composes the node holding the bus and writes the grants, and says so.
func (s assignmentAccountSteps) cover(ctx context.Context) (string, error) {
holder, err := s.busHolder(ctx)
if err != nil {
return "", fmt.Errorf("the node holding the bus cannot be read: %w", err)
}
if holder != "" {
if err := s.compose(ctx, []string{holder}); err != nil {
return "", fmt.Errorf("the bus's user list on %s is not composed: %w", holder, err)
}
}
if err := s.grants(ctx); err != nil {
return "", err
}
if holder != "" {
return "the bus's user list composed on " + holder + ", the grants written", nil
}
return "the grants written", nil
}
// assignmentStands says a value of the assignments bucket holds an assignment that stands: made (a module only pulled in
// by another names no time assigned), not ended and in no state (chain.Assignment.Stands).
func assignmentStands(a chain.Assignment) bool {
return !a.Assigned.IsZero() && a.Stands() && a.Node != "" && a.Module != ""
}
// assignmentTurns keeps which keys of the assignments bucket stood when last heard, to tell an assignment that comes to
// stand from a rewrite of one that stood.
type assignmentTurns struct {
mu sync.Mutex
stood map[string]bool
}
// heard records a value of key (nil for a delete or a purge) and answers whether it holds an assignment that just came
// to stand.
func (t *assignmentTurns) heard(key string, value []byte) (chain.Assignment, bool) {
var a chain.Assignment
stands := false
if value != nil && json.Unmarshal(value, &a) == nil {
stands = assignmentStands(a)
}
t.mu.Lock()
defer t.mu.Unlock()
if t.stood == nil {
t.stood = map[string]bool{}
}
was := t.stood[key]
t.stood[key] = stands
return a, stands && !was
}
// ownAccountOf is the username of the account a module speaks on the bus with on node, its own and no process's: the
// module's when it reads a `broker` own secret, else its provisioner client's when it runs one; false when it has none
// (a module whose accounts are its processes' alone, or one that does not speak on the bus).
func ownAccountOf(m catalogue.Manifest, node string) (string, bool) {
if _, reads := m.OwnSecrets["broker"]; reads {
return broker.Principal{Kind: broker.KindModule, Node: node, Module: m.Module}.Username(), true
}
if runsTheProvisionerClient(m) {
return broker.Principal{Kind: broker.KindProvisioner, Node: node, Module: m.Module}.Username(), true
}
return "", false
}
// issueAssignmentAccounts issues module's accounts on node that are not minted: its own (issueOnTheNewBus) and its
// processes' (issueMissingProcessAccounts). A module the catalogue does not hold, or one with nowhere to deliver an
// account (mayIssue), has nothing to issue; an account minted before is kept.
func issueAssignmentAccounts(ctx context.Context, inv *inventory.Inventory, shelf map[string]catalogue.Manifest,
node, module string) (bool, string, error) {
// Mint only what no row names: the assign act hears its own write and may issue at once (the review of issue 1945).
ctx = onlyIfAbsent(ctx)
m, known := shelf[module]
if !known || mayIssue(m) != nil {
return false, "", nil
}
if user, own := ownAccountOf(m, node); own {
_, minted, err := inv.BusUserHash(ctx, user)
if err != nil {
return false, "", fmt.Errorf("whether %s is minted cannot be read: %w", user, err)
}
if !minted {
busAddress, err := broker.BusAddress()
if err != nil {
return false, "", err
}
// Its processes' accounts too: issueOnTheNewBus issues every one of them with the module's own.
if err := issueOnTheNewBus(ctx, inv, m, node, busAddress); err != nil {
return false, "", err
}
return true, "its bus credential issued as " + user, nil
}
}
line, err := issueMissingProcessAccounts(ctx, inv, m, node)
if err != nil {
return false, "", err
}
return line != "", line, nil
}
// liveAssignmentAccountSteps are the steps on the controller's own stores and the bus: the composer's passes, as the
// done keys' take-ins use them, so a node asked for by both is composed once a pass.
func liveAssignmentAccountSteps(open *stores) assignmentAccountSteps {
inv := open.inventory
return assignmentAccountSteps{
holds: func(ctx context.Context) bool {
// The lease's gate, as every act's: a serving controller the bus does not grant the lease yet acts unleased.
_, err := theLease.epoch(ctx)
return err == nil
},
catalogue: func(ctx context.Context) (map[string]catalogue.Manifest, error) { return inv.Catalogue(ctx) },
issue: func(ctx context.Context, shelf map[string]catalogue.Manifest, node, module string) (bool, string, error) {
return issueAssignmentAccounts(ctx, inv, shelf, node, module)
},
busHolder: func(ctx context.Context) (string, error) {
holders, err := seatHolders(ctx, inv)
if err != nil {
return "", err
}
return holders[catalogue.BrokerSeat].Node, nil
},
compose: func(ctx context.Context, nodes []string) error {
return composerFor(open).compose(ctx, nodes, false)
},
grants: func(ctx context.Context) error {
return composerFor(open).compose(ctx, nil, true)
},
}
}
// assignmentAccountsRetry is how long a settle that failed waits before it tries again.
var assignmentAccountsRetry = 30 * time.Second
// untilSettled runs settle until it works or ctx ends: a failure, or a controller not holding the lease, is said and
// tried again every assignmentAccountsRetry, never dropped, since no later write of the key may come.
func untilSettled(ctx context.Context, what string, settle func(context.Context) (string, error)) {
for {
said, err := settle(ctx)
if err == nil {
if said != "" {
for _, line := range strings.Split(said, "\n") {
fmt.Printf("assignments: %s\n", line)
}
}
return
}
if ctx.Err() != nil {
return
}
fmt.Printf("assignments: %s: %v; tried again in %s\n", what, err, assignmentAccountsRetry)
select {
case <-ctx.Done():
return
case <-time.After(assignmentAccountsRetry):
}
}
}
// watchAssignmentAccounts settles every assignment of the assignments bucket that comes to stand, for as long as ctx: the
// bucket's current values first (the start's look, settled as not fresh), then each write. A watch that ends, or a
// bucket not there yet, is watched again a minute later.
func watchAssignmentAccounts(ctx context.Context, open *stores) {
steps := liveAssignmentAccountSteps(open)
goServing(ctx, func() {
runAssignmentWatches(ctx, open.inventory.WatchAssignments, func(a chain.Assignment) {
inBackground(ctx, func() {
untilSettled(ctx, a.Module+" on "+a.Node, func(ctx context.Context) (string, error) {
return steps.settle(ctx, a.Node, a.Module)
})
})
}, func(look []chain.Assignment) {
// The start's look, once in the process: one reading of the catalogue, one assignment after another,
// then one composition of the bus's node and the grants (issues 1945 and 1277).
inBackground(ctx, func() {
untilSettled(ctx, "the start's look", func(ctx context.Context) (string, error) {
return steps.look(ctx, look)
})
})
}, func() bool {
select {
case <-ctx.Done():
return false
case <-time.After(time.Minute):
return true
}
})
})
}
// runAssignmentWatches watches the assignments again each time a watch ends, until ctx ends or pause says stop: each
// assignment that comes to stand is handed to fresh, and the standing ones of the first watch that reads them all (its
// Caught mark) to look, once. A watch that ends, or fails to open, before the start's look ran is no start's look: the
// next one is, with what it hears from nothing (a fresh assignmentTurns), so no key it handed is dropped and no burst of
// fresh settles replaces the look (the review of issue 1945).
func runAssignmentWatches(ctx context.Context, watch func(context.Context) (<-chan chain.Entry, error),
fresh func(chain.Assignment), look func([]chain.Assignment), pause func() bool) {
turns := &assignmentTurns{}
first := true
for {
var atStart []chain.Assignment
if err := watchAssignmentsOnce(ctx, watch, turns, first, func(a chain.Assignment, isFresh bool) {
if !isFresh {
atStart = append(atStart, a)
return
}
fresh(a)
}, func() {
first = false
look(atStart)
atStart = nil
}); err != nil && ctx.Err() == nil {
fmt.Printf("assignments: the assignments are not heard, so a new assignment's bus accounts wait: %v\n", err)
}
if first {
turns = &assignmentTurns{} // the start's look has not run: the next watch hears every standing key anew
}
if ctx.Err() != nil || !pause() {
return
}
}
}
// watchAssignmentsOnce watches the assignments until the watch or ctx ends, handing each assignment that comes to stand
// to settle. On the process's first watch, the current values are the start's look (not fresh) and read is called once
// they are all handed (the chain's Caught mark); every other value, and every value of a watch opened again after one
// ended, is fresh: an assignment that came to stand while no watch heard it is settled whole, never as the start's look
// (the review of issue 1945).
func watchAssignmentsOnce(ctx context.Context, watch func(context.Context) (<-chan chain.Entry, error),
turns *assignmentTurns, first bool, settle func(a chain.Assignment, fresh bool), read func()) error {
ctx, stop := context.WithCancel(ctx)
defer stop()
entries, err := watch(ctx)
if err != nil {
return err
}
current := true
for {
select {
case <-ctx.Done():
return nil
case e, ok := <-entries:
if !ok {
if ctx.Err() != nil {
return nil
}
return fmt.Errorf("the watch ended")
}
if e.Caught {
if current {
current = false // the current values are read: what follows is written from now on
if first {
read()
}
}
continue
}
var value []byte
if !e.Deleted {
value = e.Value
}
if a, turned := turns.heard(e.Key, value); turned {
settle(a, !current || !first)
}
}
}
}
@@ -1,344 +0,0 @@
package main
import (
"context"
"encoding/json"
"errors"
"fmt"
"strings"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"github.com/novox/mesh-controller/internal/broker"
"github.com/novox/mesh-controller/internal/chain"
)
// An assignment that comes to stand issues its node's bus accounts and the module's grant (ADR 0543, issue 1945): on
// 2026-10-25 at 17:19 mesh-build-node and mesh-build-go, built long before, were assigned to ace, and nothing issued
// ace's accounts nor put them in the bus's user list.
func anAssignmentValue(t *testing.T, a chain.Assignment) []byte {
t.Helper()
raw, err := json.Marshal(a)
if err != nil {
t.Fatal(err)
}
return raw
}
// Only a write that makes the key stand turns it: a rewrite of a standing assignment (the provisioner's plan kept in its
// value), an ended one, a pending one or a module only pulled in by another does not. The process's first watch hands
// the assignments standing when it starts to the start's look (not fresh), says once that they are all read, and hands
// each assignment written afterwards as fresh; a watch opened again after one ended hands every standing assignment it
// has not seen stand as fresh and never calls the start's look (the review of issue 1945).
func TestAnAssignmentThatComesToStand(t *testing.T) {
at := time.Date(2026, 10, 25, 16, 19, 42, 0, time.UTC)
stands := chain.Assignment{Node: "ace", Module: "mesh-build-node", Assigned: at}
ended := stands
ended.Ended = at.Add(time.Hour)
pending := stands
pending.State = chain.StatePending
pulled := chain.Assignment{Node: "ace", Module: "mesh-build-node", RequiredBy: "mesh-build-go"}
replanned := stands
replanned.Provisioning = json.RawMessage(`{"plan":1}`)
type write struct {
value *chain.Assignment // nil for a delete
turning bool
}
for _, c := range []struct {
name string
writes []write
}{
{"a new assignment turns", []write{{&stands, true}}},
{"a rewrite of one that stands does not", []write{{&stands, true}, {&replanned, false}, {&stands, false}}},
{"an ended one does not", []write{{&ended, false}}},
{"assigned again after it ended turns again", []write{{&stands, true}, {&ended, false}, {&stands, true}}},
{"a pending one does not, and turns once made", []write{{&pending, false}, {&stands, true}}},
{"a module only pulled in does not", []write{{&pulled, false}}},
{"one deleted and written again turns again", []write{{&stands, true}, {nil, false}, {&stands, true}}},
} {
t.Run(c.name, func(t *testing.T) {
turns := &assignmentTurns{}
for i, w := range c.writes {
var value []byte
if w.value != nil {
value = anAssignmentValue(t, *w.value)
}
a, turned := turns.heard("ace.mesh-build-node", value)
if turned != w.turning {
t.Fatalf("write %d: turned %v, want %v", i, turned, w.turning)
}
if turned && (a.Node != "ace" || a.Module != "mesh-build-node") {
t.Fatalf("write %d: the assignment turned is %s on %s", i, a.Module, a.Node)
}
}
})
}
type handed struct {
key string
fresh bool
}
for _, first := range []bool{true, false} {
t.Run(fmt.Sprintf("the watch, first %v", first), func(t *testing.T) {
bucket := chain.NewMemory()
put := func(a chain.Assignment) {
t.Helper()
value := anAssignmentValue(t, a)
if e, held, _ := bucket.Get(t.Context(), a.Key()); held {
if _, err := bucket.Update(t.Context(), a.Key(), value, e.Revision); err != nil {
t.Fatal(err)
}
return
}
if _, err := bucket.Create(t.Context(), a.Key(), value); err != nil {
t.Fatal(err)
}
}
put(chain.Assignment{Node: "g14", Module: "mesh-build-go", Assigned: at})
got := make(chan handed, 8)
read := make(chan struct{}, 8)
ctx, stop := context.WithCancel(t.Context())
done := make(chan error, 1)
go func() {
done <- watchAssignmentsOnce(ctx, bucket.Watch, &assignmentTurns{}, first,
func(a chain.Assignment, fresh bool) { got <- handed{a.Key(), fresh} },
func() { read <- struct{}{} })
}()
next := func() handed {
t.Helper()
select {
case h := <-got:
return h
case <-time.After(5 * time.Second):
t.Fatal("nothing handed")
return handed{}
}
}
// Standing when the watch opened: the start's look on the first watch, fresh on any other.
if h := next(); h != (handed{"g14.mesh-build-go", !first}) {
t.Fatalf("the standing assignment was handed %+v", h)
}
put(chain.Assignment{Node: "g14", Module: "mesh-build-go", Assigned: at, Provisioning: json.RawMessage(`{"plan":2}`)})
put(chain.Assignment{Node: "ace", Module: "mesh-build-node", Assigned: at})
if h := next(); h != (handed{"ace.mesh-build-node", true}) {
t.Fatalf("handed %+v, want ace.mesh-build-node as fresh (and nothing for the rewrite before it)", h)
}
stop()
if err := <-done; err != nil {
t.Fatal(err)
}
if want := map[bool]int{true: 1, false: 0}[first]; len(read) != want {
t.Fatalf("the start's look was called %d time(s), want %d", len(read), want)
}
})
}
// A first watch that fails to open, or ends before its Caught mark, is no start's look: the next watch is, and hands
// every standing assignment to it once, none as fresh (the review of issue 1945: a burst of fresh settles, or keys
// handed and dropped).
for _, firstWatch := range []string{"fails to open", "ends before its Caught mark"} {
t.Run("a first watch that "+firstWatch, func(t *testing.T) {
bucket := chain.NewMemory()
standing := chain.Assignment{Node: "g14", Module: "mesh-build-go", Assigned: at}
if _, err := bucket.Create(t.Context(), standing.Key(), anAssignmentValue(t, standing)); err != nil {
t.Fatal(err)
}
watches := 0
watch := func(ctx context.Context) (<-chan chain.Entry, error) {
watches++
if watches > 1 {
return bucket.Watch(ctx)
}
if firstWatch == "fails to open" {
return nil, errors.New("the chain is not open yet")
}
cut := make(chan chain.Entry, 1)
cut <- chain.Entry{Key: standing.Key(), Value: anAssignmentValue(t, standing)}
close(cut)
return cut, nil
}
var looks [][]chain.Assignment
var fresh []chain.Assignment
ctx, stop := context.WithCancel(t.Context())
defer stop()
runAssignmentWatches(ctx, watch, func(a chain.Assignment) { fresh = append(fresh, a) },
func(look []chain.Assignment) { looks = append(looks, look); stop() },
func() bool { return ctx.Err() == nil })
if len(looks) != 1 || len(looks[0]) != 1 || looks[0][0].Key() != standing.Key() || len(fresh) != 0 {
t.Fatalf("after %d watch(es): looks %v, fresh %v; want one look at %s and nothing fresh",
watches, looks, fresh, standing.Key())
}
})
}
}
// settle issues what is missing and then puts it on the bus: the node holding the bus composed (its user list) and the
// grants written, even when every account was minted before (a module assigned again keeps its account). The start's
// look reads the catalogue once, issues for each standing assignment, and composes the bus's node and writes the grants
// once when it looked at any. Neither acts for a controller that does not hold the lease, nor for the controller's own
// module.
func TestSettlingAndTheStartsLook(t *testing.T) {
ace := func(m string) chain.Assignment { return chain.Assignment{Node: "ace", Module: m} }
cases := []struct {
name string
look []chain.Assignment // nil: settle mesh-build-node on ace
module string
notHolding bool
issued bool
issueErr, composeE error
holder string
wantIssues int
wantComposed []string
wantGrants bool
wantErr string
}{
{name: "an account not minted", module: "mesh-build-node", issued: true, holder: "novox", wantIssues: 1,
wantComposed: []string{"novox"}, wantGrants: true},
{name: "an account minted before: the bus and the grants still", module: "mesh-build-go", holder: "novox",
wantIssues: 1, wantComposed: []string{"novox"}, wantGrants: true},
{name: "no lease: nothing, and tried again", module: "mesh-build-node", notHolding: true, issued: true,
holder: "novox", wantErr: errNotHolding.Error()},
{name: "the controller's own module is the terminal's", module: controllerModule, issued: true, holder: "novox"},
{name: "no bus holder: the grants still", module: "mesh-build-node", issued: true, wantIssues: 1, wantGrants: true},
{name: "an issue that fails is the error, nothing composed", module: "mesh-build-node",
issueErr: errors.New("the bus cannot be reached"), holder: "novox", wantIssues: 1, wantErr: "the bus cannot be reached"},
{name: "a composition that fails is the error, no grant", module: "mesh-build-node", issued: true, holder: "novox",
composeE: errors.New("store refused"), wantIssues: 1, wantComposed: []string{"novox"}, wantErr: "store refused"},
{name: "the start's look: each issued, the bus and the grants once",
look: []chain.Assignment{ace("mesh-build-node"), ace(controllerModule), ace("mesh-build-go")}, issued: true,
holder: "novox", wantIssues: 2, wantComposed: []string{"novox"}, wantGrants: true},
{name: "the start's look with every account minted still covers once",
look: []chain.Assignment{ace("mesh-build-node"), ace("mesh-build-go")}, holder: "novox", wantIssues: 2,
wantComposed: []string{"novox"}, wantGrants: true},
{name: "the start's look at nothing does nothing", look: []chain.Assignment{ace(controllerModule)}, holder: "novox"},
}
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
var composed []string
issues, catalogues, grants := 0, 0, 0
steps := assignmentAccountSteps{
holds: func(context.Context) bool { return !c.notHolding },
catalogue: func(context.Context) (map[string]catalogue.Manifest, error) {
catalogues++
return map[string]catalogue.Manifest{}, nil
},
issue: func(context.Context, map[string]catalogue.Manifest, string, string) (bool, string, error) {
issues++
if c.issued {
return true, "its bus credential issued", c.issueErr
}
return false, "", c.issueErr
},
busHolder: func(context.Context) (string, error) { return c.holder, nil },
compose: func(_ context.Context, nodes []string) error {
composed = append(composed, nodes...)
return c.composeE
},
grants: func(context.Context) error { grants++; return nil },
}
var said string
var err error
if c.look != nil {
said, err = steps.look(t.Context(), c.look)
} else {
said, err = steps.settle(t.Context(), "ace", c.module)
}
if c.wantErr != "" {
if err == nil || !strings.Contains(err.Error(), c.wantErr) {
t.Fatalf("error %v, want one saying %q", err, c.wantErr)
}
} else if err != nil {
t.Fatal(err)
}
if issues != c.wantIssues || catalogues > 1 {
t.Fatalf("%d issue(s) from %d reading(s) of the catalogue, want %d from at most one", issues, catalogues, c.wantIssues)
}
if strings.Join(composed, ",") != strings.Join(c.wantComposed, ",") {
t.Fatalf("composed %v, want %v", composed, c.wantComposed)
}
if want := map[bool]int{true: 1}[c.wantGrants]; grants != want {
t.Fatalf("grants written %d time(s), want %d", grants, want)
}
if c.wantGrants && c.wantErr == "" && !strings.Contains(said, "the grants written") {
t.Fatalf("said %q, want it to name the grants", said)
}
})
}
}
// The accounts a module speaks with on a node: the module's own when it reads `broker`, its provisioner client's when
// it runs one alone, none when its accounts are its processes' or it does not speak on the bus. A process's account is
// held in the provisioner client's form, named for the process and carrying no claims, and may be issued only when
// the process declares its own secret broker-<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)
}
})
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)
}
}
})
}
+4
View File
@@ -9,6 +9,7 @@ import (
"regexp"
"slices"
"strings"
"sync"
"time"
"git.novox.be/novox/mesh-sdk/go/catalogue"
@@ -107,7 +108,10 @@ func sourceFormsRead(ctx context.Context, now func() time.Time,
f := &sourceForms{bases: map[string]string{}}
var failedAt time.Time // when the last read failed; zero when it did not
var failure error
var mu sync.Mutex // the forms are shared by take-ins side by side (sharedSourceForms): one read at a time
f.read = func(ctx context.Context, seat string, again bool) (string, error) {
mu.Lock()
defer mu.Unlock()
if b, known := f.bases[seat]; known {
return b, nil
}
+2 -6
View File
@@ -49,7 +49,7 @@ func attachTheChain(ctx context.Context, inv *inventory.Inventory, server *link.
// keys laid out before skips them, so this takes seconds; one that fails is tried again beside the serving, loudly.
if err := inv.LayBuildsOut(ctx, func(format string, args ...any) { fmt.Printf("chain: "+format+"\n", args...) }); err != nil {
fmt.Printf("chain: the builds are not laid out under their modules yet, and none is read until they are: %v\n", err)
goServing(ctx, func() { layBuildsOutSoon(ctx, inv) })
go layBuildsOutSoon(ctx, inv)
}
// The requests the move of delivery 1 found never handed over are not asked again: the controller asks no build
// (novox/hq ADR 0501; issue 1851, A10), and the builder builds every fingerprint the indexer writes.
@@ -80,11 +80,7 @@ func layBuildsOutSoon(ctx context.Context, inv *inventory.Inventory) {
Source: "chain", Summary: "the builds are not laid out under their modules, so none is read: "}
if err != nil {
fmt.Printf("chain: the builds are not laid out under their modules, and none is read until they are: %v\n", err)
if ctx.Err() != nil {
return // stopped: the next controller lays them out
}
// A cancelled layout is tried again and raises nothing: a cancellation is no fault of the chain (issue 1945).
if keeper := conditionsFrom; keeper != nil && !failed && !errors.Is(err, context.Canceled) {
if keeper := conditionsFrom; keeper != nil && !failed {
o.Summary += err.Error()
if _, oerr := keeper.Observe(ctx, o); oerr == nil {
failed = true
+8 -15
View File
@@ -166,12 +166,10 @@ func followDoneKey(ctx context.Context, open *stores, module string) {
err := takeOnceInSlot(ctx, open, module)
o := doneKeyObservation(module)
if err == nil {
// Cleared on every success, whoever raised it: a controller before a restart or another follower (issue
// 1945). A condition that is not open costs one read.
if keeper := conditionsFrom; keeper != nil {
if keeper := conditionsFrom; keeper != nil && failed {
_, _ = keeper.Clear(ctx, o.Key(), "composed and granted")
failed = false
}
failed = false
if _, again := doneKeysLookAgain.LoadAndDelete(module); again {
continue // a wake came while this one ran: once more, for the key written meanwhile
}
@@ -196,12 +194,7 @@ func followDoneKey(ctx context.Context, open *stores, module string) {
}
continue
}
if ctx.Err() != nil {
return // stopped: a take-in cut by the stop is no fault of the build, and the next controller takes it in
}
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).
if keeper := conditionsFrom; keeper != nil && !failed {
o.Summary += err.Error()
if _, oerr := keeper.Observe(ctx, o); oerr == nil {
failed = true
@@ -309,18 +302,18 @@ func watchDoneKeys(ctx context.Context, open *stores) {
return
}
if err := builderkeys.Watch(ctx, keys.JS, func(module string) {
inBackground(ctx, func() { takeDoneKey(ctx, open, module) })
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() {
go func() {
<-ctx.Done()
closeKeys()
})
}()
// And every module once now, after the watch is open so no write falls between the two.
inBackground(ctx, func() {
inBackground(func() {
doneKeysAtStart(ctx, catalogedModules(open), func(module string) { takeDoneKeyAtStart(ctx, open, module) })
})
}
@@ -391,7 +384,7 @@ func takeDoneKeyAtStart(ctx context.Context, open *stores, module string) {
_, again := doneKeysLookAgain.LoadAndDelete(module)
if err != nil || again {
// A failure, a wait or a wake meanwhile is followed by takeDoneKey, which says it and looks again.
inBackground(ctx, func() { takeDoneKey(ctx, open, module) })
inBackground(func() { takeDoneKey(ctx, open, module) })
}
}
@@ -1,83 +0,0 @@
package main
import (
"context"
"fmt"
"testing"
"time"
"github.com/novox/mesh-controller/internal/conditions"
)
// A take-in cut by a cancellation raises no condition (issue 1945: on 2026-10-25 thirty urgent fingerprint-not-composed
// conditions were live, each a take-in a stop or a restart cut with "context canceled"): when the controller stops, the
// next one takes it in; when a step of its own was cancelled, it is tried again.
func TestATakeInCutByACancellationRaisesNothing(t *testing.T) {
for _, tc := range []struct {
name string
stop bool // the controller's context ends, else only the step's error is a cancellation
}{{"the controller stops", true}, {"a step was cancelled", false}} {
t.Run(tc.name, func(t *testing.T) {
was, wasRetry, wasSay, wasKeeper := doneKeyTakeOnce, doneKeyRetry, doneKeysSay, conditionsFrom
t.Cleanup(func() {
doneKeyTakeOnce, doneKeyRetry, doneKeysSay, conditionsFrom = was, wasRetry, wasSay, wasKeeper
})
store := conditions.NewInMemory()
k := conditions.NewKeeper(t.Context(), conditions.Options{Store: store, History: store,
Teller: &conditions.Told{}})
t.Cleanup(func() { k.Close(context.Background()) })
conditionsFrom = k
doneKeyRetry = time.Millisecond
doneKeysSay = func(string) {}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
calls := 0
doneKeyTakeOnce = func(context.Context, *stores, string) error {
calls++
if tc.stop {
cancel() // stopped mid-way
return fmt.Errorf("the module cannot be composed: %w", context.Canceled)
}
if calls == 1 {
return fmt.Errorf("the module cannot be composed: %w", context.Canceled)
}
return nil
}
done := make(chan struct{})
go func() { takeDoneKey(ctx, nil, "cut-module"); close(done) }()
select {
case <-done:
case <-time.After(5 * time.Second):
t.Fatal("the take-in did not end")
}
if !tc.stop && calls != 2 {
t.Errorf("a take-in whose step was cancelled was tried %d time(s), want 2", calls)
}
if c, found, err := conditions.ReadOne(t.Context(), store, doneKeyObservation("cut-module").Key()); err != nil || found {
t.Fatalf("a take-in cut by a cancellation raised %+v (found %v, %v)", c, found, err)
}
})
}
}
// A take-in that succeeds clears the module's fingerprint-not-composed condition whoever raised it: a controller before a
// restart, or another follower (issue 1945: the operator read "a build needs a look" for take-ins that had since worked).
func TestASucceedingTakeInClearsAConditionAnotherRaised(t *testing.T) {
was, wasRetry, wasSay, wasKeeper := doneKeyTakeOnce, doneKeyRetry, doneKeysSay, conditionsFrom
t.Cleanup(func() { doneKeyTakeOnce, doneKeyRetry, doneKeysSay, conditionsFrom = was, wasRetry, wasSay, wasKeeper })
store := conditions.NewInMemory()
k := conditions.NewKeeper(t.Context(), conditions.Options{Store: store, History: store, Teller: &conditions.Told{}})
t.Cleanup(func() { k.Close(context.Background()) })
conditionsFrom = k
doneKeysSay = func(string) {}
o := doneKeyObservation("raised-before")
o.Summary += "the store refused the connection"
if _, err := k.Observe(t.Context(), o); err != nil {
t.Fatal(err)
}
doneKeyTakeOnce = func(context.Context, *stores, string) error { return nil }
takeDoneKey(t.Context(), nil, "raised-before")
if c, found, err := conditions.ReadOne(t.Context(), store, o.Key()); err != nil || found {
t.Fatalf("a take-in that succeeded left the condition another raised: %+v (found %v, %v)", c, found, err)
}
}
+3 -3
View File
@@ -134,7 +134,7 @@ func watchHavesForJudgement(ctx context.Context, open *stores) {
fmt.Printf("compositions: the haves are not watched: %v\n", err)
return
}
goServing(ctx, func() {
go func() {
defer bus.Close()
defer w.Stop()
for {
@@ -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)
}
@@ -165,5 +165,5 @@ func watchHavesForJudgement(ctx context.Context, open *stores) {
})
}
}
})
}()
}
@@ -608,11 +608,8 @@ func keepWhatServeSets(t *testing.T) {
t.Setenv(servedVar, "")
unheld, status, given, check, follow, keeper, hand := logUnheldChanges, statusFrom, givenEvents, checkEvents,
link.Calls.Follow, conditionsFrom, handActConn
doctor, asker := doctorFrom, askerFrom.Load()
t.Cleanup(func() {
logUnheldChanges, statusFrom, givenEvents, checkEvents, link.Calls.Follow, conditionsFrom, handActConn =
unheld, status, given, check, follow, keeper, hand
doctorFrom = doctor
askerFrom.Store(asker)
})
}
+1 -1
View File
@@ -49,7 +49,7 @@ func keeperOn(ctx context.Context, conn *nats.Conn, looks conditions.LookStore)
Say: func(format string, args ...any) { fmt.Fprintf(os.Stderr, format+"\n", args...) },
// What status leads with changed: composed again soon (a nudge outside the serving controller
// does nothing).
Changed: func() { statusFrom.nudge(); askerFrom.Load().nudge() },
Changed: func() { statusFrom.nudge(); askerFrom.nudge() },
// Written under the lease, carrying its epoch (novox/hq to-be 45 §6).
Epoch: func() (uint64, error) { return theLease.epoch(context.WithoutCancel(ctx)) }}), nil
}
-3
View File
@@ -225,9 +225,6 @@ func filesNotPutObservation(node, summary string) conditions.Observation {
// notePut records how putting a send's objects went: a failure counts, the second in a row raises the condition, a success
// clears the count and the condition.
func notePut(ctx context.Context, node string, put error) {
if errors.Is(put, context.Canceled) || ctx.Err() != nil {
return // a send cut by a cancellation or a stop says nothing of the bucket (issue 1945)
}
filesFailMu.Lock()
if put == nil {
delete(filesFailed, node)
@@ -415,7 +415,6 @@ func TestAPutThatFailsSendAfterSendRaisesAConditionAndClearsWhenItWorks(t *testi
was := conditionsFrom
conditionsFrom = k
t.Cleanup(func() { conditionsFrom = was })
forgetPutFailures(t, "cond-node")
key := filesNotPutObservation("cond-node", "").Key()
open := func() bool {
_, found, err := conditions.ReadOne(t.Context(), store, key)
@@ -443,36 +442,6 @@ func TestAPutThatFailsSendAfterSendRaisesAConditionAndClearsWhenItWorks(t *testi
}
}
// A send cut by a cancellation counts as no failure to put its files, however often (issue 1945: a cancellation is no fault
// of the bucket).
func TestAPutCutByACancellationRaisesNothing(t *testing.T) {
store := conditions.NewInMemory()
k := conditions.NewKeeper(t.Context(), conditions.Options{Store: store, History: store, Teller: &conditions.Told{}})
t.Cleanup(func() { k.Close(context.Background()) })
was := conditionsFrom
conditionsFrom = k
t.Cleanup(func() { conditionsFrom = was })
forgetPutFailures(t, "cut-node")
cut := fmt.Errorf("the files could not be put: %w", context.Canceled)
for range filesNotPutAfter + 1 {
notePut(t.Context(), "cut-node", cut)
}
if _, found, err := conditions.ReadOne(t.Context(), store, filesNotPutObservation("cut-node", "").Key()); err != nil || found {
t.Fatalf("sends cut by a cancellation raised the condition (found %v, %v)", found, err)
}
}
// forgetPutFailures starts the node's count of failed puts at nothing and leaves it so, as a run of the test again finds it.
func forgetPutFailures(t *testing.T, node string) {
forget := func() {
filesFailMu.Lock()
delete(filesFailed, node)
filesFailMu.Unlock()
}
forget()
t.Cleanup(forget)
}
// aBusThatSays is a bus that keeps the declarations published to it and, at each, asks whether the node's bucket is held.
type aBusThatSays struct {
link.Bus
@@ -87,29 +87,3 @@ func TestTheDeclaredPackagesMirrorNamesWhatEachNodesModulesDeclare(t *testing.T)
}
}
}
// 1945/AC-mirror-5: the digest the declared-packages mirror skips an unchanged pass by is made from a real reading of the
// stores, is the same over two readings of the same records, and moves when an assignment does.
func TestTheInputDigestOfTheStoresIsStable(t *testing.T) {
open := aMesh(t)
ctx := t.Context()
register(t, open, catalogue.Manifest{Module: "viewer", Version: "1", Resources: []map[string]any{
{"id": "pkg-curl", "type": "package", "package": "curl"}}})
read := func() string {
in, err := readComposeInput(ctx, open)
if err != nil {
t.Fatal(err)
}
return inputDigest(in)
}
first, second := read(), read()
if first == "" || first != second {
t.Fatalf("two readings of the same records gave the digests %q and %q", first, second)
}
if _, err := assign(ctx, open, "anchor", "viewer"); err != nil {
t.Fatalf("assign: %v", err)
}
if read() == first {
t.Errorf("an assignment left the digest of the stores' Input at %s", first)
}
}
+3 -3
View File
@@ -169,7 +169,7 @@ func watchHavesForGiven(ctx context.Context, open *stores) {
log.Printf("given values: the haves are not watched: %v", err)
return
}
goServing(ctx, func() {
go func() {
defer bus.Close()
defer w.Stop()
for {
@@ -195,8 +195,8 @@ 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}) })
}
}
})
}()
}
+39 -1
View File
@@ -13,6 +13,7 @@ import (
"path"
"sort"
"strings"
"sync"
"time"
"git.novox.be/novox/mesh-sdk/go/buildstep"
@@ -122,7 +123,7 @@ var predefinedAtTrunk = func(ctx context.Context, inv *inventory.Inventory) (pre
}
// formsOfSources reads how the sources are spelled on this mesh; a variable so a test names the forge's address.
var formsOfSources = newSourceForms
var formsOfSources = sharedSourceForms
// errNoCatalogueRepository says no module of the mesh is built from the catalogue's repository on the forge: not a wait,
// since no read of the forge makes one appear (issue 1902); the plan fails with these words.
@@ -632,3 +633,40 @@ func handCoversEvery(m catalogue.Manifest, k builderkeys.Key, hand map[string]st
}
return nil
}
// sharedSourceForms is the one reading of how the sources are spelled (the forge's address above all) that every take-in
// of the inventory shares, made again once it is older than sourceFormsKept, so a forge that moved is read within that
// bound. A fresh reading per take-in read the whole mesh once each, about 120 at once at 16:08 on 2026-10-25, each with
// pools of its own (issues 1945 and 1277). A base that could not be read is not kept: the reading reads it again.
func sharedSourceForms(ctx context.Context, inv *inventory.Inventory) *sourceForms {
sharedFormsMu.Lock()
defer sharedFormsMu.Unlock()
now := sourceFormsNow()
if kept, ok := sharedForms.Load(inv); ok {
if k := kept.(keptForms); now.Sub(k.at) < sourceFormsKept {
return k.forms
}
}
// Read with no caller's deadline: the reading outlives the take-in that made it.
forms := sourceFormsFor(context.WithoutCancel(ctx), inv)
sharedForms.Store(inv, keptForms{forms: forms, at: now})
return forms
}
// keptForms is a shared reading of the sources and when it was made.
type keptForms struct {
forms *sourceForms
at time.Time
}
// sourceFormsFor makes a fresh reading of the sources; a variable so a test counts them.
var sourceFormsFor = newSourceForms
var (
sharedForms sync.Map // *inventory.Inventory -> keptForms
sharedFormsMu sync.Mutex
sourceFormsNow = time.Now
)
// sourceFormsKept is how long the take-ins share one reading of the sources before it is made again.
const sourceFormsKept = time.Minute
+9 -31
View File
@@ -965,26 +965,6 @@ func processPrincipalFor(m catalogue.Manifest, processID string) (broker.Process
return broker.ProcessPrincipal{}, false
}
// mintOnlyIfAbsent marks a context whose issue mints only accounts no row names yet (onlyIfAbsent).
type mintOnlyIfAbsent struct{}
// onlyIfAbsent is ctx for an issue that must never replace a minted account: the assign act's and the assignments'
// watch's, which may run at once for one account (novox/hq ADR 0543, the review of issue 1945).
func onlyIfAbsent(ctx context.Context) context.Context {
return context.WithValue(ctx, mintOnlyIfAbsent{}, true)
}
// mintBusUser mints u's password. By default it replaces whatever stands (a rotation: `module issue`); under
// onlyIfAbsent it mints only when no row names u, and won false says another caller minted it first and seals it,
// so this one seals nothing.
func mintBusUser(ctx context.Context, inv *inventory.Inventory, u inventory.BusUser) (string, bool, error) {
if only, _ := ctx.Value(mintOnlyIfAbsent{}).(bool); only {
return inv.MintBusPasswordIfAbsent(ctx, u)
}
password, err := inv.MintBusPassword(ctx, u)
return password, err == nil, err
}
// issueOnTheNewBus gives an assigned module its credential on the bus being built.
//
// **Three things differ from a management call, and each is the point of the move.** The credential
@@ -1010,16 +990,14 @@ func issueOnTheNewBus(ctx context.Context, inv *inventory.Inventory, m catalogue
}
if _, reads := m.OwnSecrets["broker"]; reads {
user := broker.Principal{Kind: broker.KindModule, Node: node, Module: m.Module}.Username()
password, won, err := mintBusUser(ctx, inv, inventory.BusUser{
password, err := inv.MintBusPassword(ctx, inventory.BusUser{
Username: user, Kind: busKindOf(m.Module), Node: node, Module: m.Module,
})
if err != nil {
return err
}
if won {
if err := issueWith(ctx, inv, m, node, busAddress, known, reachable, user, password); err != nil {
return err
}
if err := issueWith(ctx, inv, m, node, busAddress, known, reachable, user, password); err != nil {
return err
}
}
// Every process with a bus declaration gets an account of its own (novox/hq ADR 0436 §1, §2), minted and
@@ -1043,11 +1021,11 @@ func issueOnTheNewBus(ctx context.Context, inv *inventory.Inventory, m catalogue
func issueProcess(ctx context.Context, inv *inventory.Inventory, m catalogue.Manifest, pr broker.ProcessPrincipal,
node, busAddress string, known broker.Broker, reachable string) error {
user := broker.Principal{Kind: broker.KindProcess, Node: node, Module: m.Module, Process: pr.Process}.Username()
password, won, err := mintBusUser(ctx, inv, inventory.BusUser{
password, err := inv.MintBusPassword(ctx, inventory.BusUser{
Username: user, Kind: inventory.BusProcess, Node: node, Module: m.Module,
})
if err != nil || !won {
return err // lost: the caller that minted it seals it
if err != nil {
return err
}
name := processSecretName(pr.Process)
held, err := heldProcessAccount(known, reachable, node, m.Module, user, password)
@@ -1113,11 +1091,11 @@ func runsTheProvisionerClient(m catalogue.Manifest) bool {
func issueProvisionerClient(ctx context.Context, inv *inventory.Inventory, m catalogue.Manifest, node string,
known broker.Broker, reachable string) error {
user := broker.Principal{Kind: broker.KindProvisioner, Node: node, Module: m.Module}.Username()
password, won, err := mintBusUser(ctx, inv, inventory.BusUser{
password, err := inv.MintBusPassword(ctx, inventory.BusUser{
Username: user, Kind: inventory.BusProvisioner, Node: node, Module: m.Module,
})
if err != nil || !won {
return err // lost: the caller that minted it seals it
if err != nil {
return err
}
held, err := json.Marshal(struct {
URL string `json:"url"`
+10 -16
View File
@@ -200,17 +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 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)
@@ -226,16 +223,16 @@ func watchNewModules(ctx context.Context, open *stores) {
fmt.Printf("new modules: the builder's keys are not watched: %v\n", err)
return
}
goServing(ctx, func() {
go func() {
<-ctx.Done()
closeKeys()
})
}()
}
// 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
@@ -253,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.
-11
View File
@@ -59,17 +59,6 @@ func dataWords(o conditions.Observation) (what, module string) {
}
var plainWordings = map[string]func(conditions.Observation) words{
// A module's new build whose compositions or grant the controller could not write (composed_keys.go).
kindNotComposed: worded(func(o conditions.Observation) words {
m := "a module"
if p := idPart(o, 0); o.Scope == conditions.ScopeBuild && p != "" {
m = p // the module: raised in the build's scope, its id the module's name
}
return words{Headline: "A new build of " + m + " cannot be delivered",
Explanation: fmt.Sprintf("A new build of %s was made, and the mesh could not prepare it for the machines, so "+
"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"}
}),
// Machines and their node-engine.
"silent": worded(func(o conditions.Observation) words {
m := machineOr(o, "a machine")
-2
View File
@@ -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
+58
View File
@@ -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) {
+14 -34
View File
@@ -125,15 +125,7 @@ func serve(ctx context.Context) (err error) {
defer theLease.release()
ctx, stopActing := context.WithCancel(ctx)
defer stopActing()
// Every goroutine started from here is counted, and ended and waited for before serve returns (serving_group.go):
// on the way out by error here, and at the end before anything else is undone.
ctx, running := withServingGroup(ctx)
stopRunning := func() {
stopActing()
running.stop()
}
defer stopRunning()
goServing(ctx, func() {
go func() {
select {
case <-lost:
stopActing()
@@ -143,7 +135,7 @@ func serve(ctx context.Context) (err error) {
// asks the controller after this one rather than have a command started and killed with this
// process (novox/hq issue 289).
handingOver.Store(true)
})
}()
defer func() {
select {
case <-lost:
@@ -170,17 +162,6 @@ func serve(ctx context.Context) (err error) {
if err := attachTheChain(ctx, inv, server); err != nil {
return err
}
bus, isNATS := server.Bus().(link.OverNATS)
if !isNATS {
return errors.New("the mesh's verbs are served over the bus, and this control plane is not on it")
}
// The condition store, opened before anything that raises or reads a condition starts beside the serving, so none
// of them meets conditionsFrom being written (issue 1945); closed after they all ended.
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
// somebody deleted, a mesh raised from a restored backup, or a bus whose data directory was
@@ -195,7 +176,7 @@ func serve(ctx context.Context) (err error) {
// And the pending assignments settled on a tick of their own (novox/hq ADR 0261): made once their module
// is registered, ended with why when its build will not register it, raised and cleared as conditions.
// Never by a read.
goServing(ctx, func() { settlingPending(ctx, open) })
go settlingPending(ctx, open)
// The durations the core's bounds are set from are kept a month (novox/hq to-be 45 Phase 0).
// And what the catalogue decided a build meant. The builder's own result is already handled
// above; this is the other half — the control plane is the only one of the three that knows
@@ -243,17 +224,19 @@ func serve(ctx context.Context) (err error) {
"Those answer the reason when called; everything else is served as usual\n",
catalogue.ControllerSeatName, strings.Join(behind, ", "))
}
bus, isNATS := server.Bus().(link.OverNATS)
if !isNATS {
return errors.New("the mesh's verbs are served over the bus, and this control plane is not on it")
}
// The hand-act log is counted for `status` on this connection rather than a new one a minute; no longer serving, the
// connection is closed, and a verb run after in this process dials its own (a served test's leftover failed the
// 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
@@ -264,18 +247,18 @@ func serve(ctx context.Context) (err error) {
return err
}
// Composed now and kept current, before the verb that answers from it is served.
goServing(ctx, func() { statusFrom.keep(ctx) })
go statusFrom.keep(ctx)
// The facts snapshot a merge check is fed (novox/hq to-be 45 §9): kept current while this
// controller holds the lease, read by the build seat from the artifact store.
goServing(ctx, func() { exportingFacts(ctx, open, bus.Conn.ConnectedServerVersion) })
go exportingFacts(ctx, open, bus.Conn.ConnectedServerVersion)
// The record mirrors (novox/hq ADR 0289, design 57 F3): the assignments and each registered module's
// declared config on the bus, written from the store only when a record changed, by the lease holder.
// The provisioner's plan of each handed module, as its machine was last sent it (design 57 P6), in the module's
// assignment.
inv.SetProvisioningPlans(inv.SentProvisioningPlans)
goServing(ctx, func() { mirroringRecords(ctx, open, inv, bus.Conn, bus) })
go mirroringRecords(ctx, open, inv, bus.Conn, bus)
// And the bucket of the files that travel by digest, judged against what each machine's latest declaration names (ADR 0486 §3).
goServing(ctx, func() { sweepingDeclarationFiles(ctx, inv, bus.Conn) })
go sweepingDeclarationFiles(ctx, inv, bus.Conn)
// And the modules the catalogue does not hold yet, registered from the builder's first done key of each, woken by
// the builder's and the indexer's writes (novox/hq ADR 0501 §6, issue 1851): nothing asks a first build.
watchNewModules(ctx, open)
@@ -307,7 +290,7 @@ func serve(ctx context.Context) (err error) {
// signals table, what the bus says about itself, and the self-check. A store that cannot be opened
// is said and the controller serves on: status then says the conditions cannot be read, and is
// not well — louder than not serving at all, and the push that repairs the bus still runs.
if stopWatching := watchTheMesh(ctx, open, server, bus, keeper); stopWatching != nil {
if stopWatching := watchTheMesh(ctx, open, server, bus); stopWatching != nil {
defer stopWatching()
}
stopServing, err := bus.ServeSeatTools(catalogue.ControllerSeatName, handlers, log.New(os.Stdout, "", log.LstdFlags))
@@ -322,7 +305,7 @@ func serve(ctx context.Context) (err error) {
}
defer stopCLI()
// Its read calls write no line each; how many each caller made is said once an hour (novox/hq issue 762).
goServing(ctx, func() { sayReadCalls(ctx) })
go sayReadCalls(ctx)
// And says so on the bus (novox/hq ADR 0197): what it serves, as the NATS services protocol asks.
stopAnnouncing, err := bus.Announce(seatAnnouncement(handlers), log.New(os.Stdout, "", log.LstdFlags))
if err != nil {
@@ -330,10 +313,7 @@ func serve(ctx context.Context) (err error) {
}
defer stopAnnouncing()
err = server.Serve(ctx)
// Ended and waited for before the verbs stop being served and the globals are put back, not after.
stopRunning()
return err
return server.Serve(ctx)
}
// OverlayCIDRVar is the range the mesh allocates node addresses from.
+23 -265
View File
@@ -2,15 +2,12 @@ package main
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"sort"
"time"
"git.novox.be/novox/mesh-sdk/go/compose"
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
@@ -27,8 +24,7 @@ import (
// so no one call site sees every change; what every change has in common is that it is committed to the
// store. So the serving controller reads the store every mirrorEvery and writes only what differs from what
// the bucket holds (ADR 0289 §1): a quiet mesh writes nothing, and a change rolled back is never read. A
// mirror is therefore at most mirrorEvery behind its store. The declared-packages mirror is the exception: it composes
// every node, so it passes on a change of what it composes from, not on this tick (keepingDeclaredPackages, issue 1945).
// mirror is therefore at most mirrorEvery behind its store.
// mirrorEvery is how often the mirrors are compared with the store: two small reads of the store.
const mirrorEvery = 10 * time.Second
@@ -49,31 +45,21 @@ func mirroringRecords(ctx context.Context, open *stores, inv *inventory.Inventor
return e
}
sources := map[string]func(context.Context) (map[string][]byte, error){
recordmirror.DeclaredConfig.Name: inv.DeclaredConfigMirror,
recordmirror.Predefined.Name: inv.PredefinedMirror,
recordmirror.DeclaredConfig.Name: inv.DeclaredConfigMirror,
recordmirror.Predefined.Name: inv.PredefinedMirror,
recordmirror.DeclaredPackages.Name: declaredPackagesMirror(open),
}
packages := keepingDeclaredPackages(ctx, open, api, announce, epoch, holds)
held := false
writers := map[string]*recordmirror.Writer{}
failing := map[string]string{}
tick := time.NewTicker(mirrorEvery)
defer tick.Stop()
events := &assignmentEvents{}
for {
now := holds()
if now && !held {
// The lease taken: the declared-packages mirror is brought to the stores at once, not at its next change.
packages.poke()
}
held = now
if now {
if holds() {
// The assignments are keys of the bus, written by the verbs that change them (novox/hq ADR 0408): this tick
// keeps the provisioner's plan in their values and says what changed. They are no mirror of the store.
keepingAssignments(ctx, inv, announce, epoch, events, failing)
for _, m := range recordmirror.Mirrors() {
if sources[m.Name] == nil {
continue // the declared packages, kept on a change (keepingDeclaredPackages)
}
why := mirrorOnce(ctx, api, m, writers, sources[m.Name], announce, epoch, holds)
if why != failing[m.Name] {
if why != "" {
@@ -193,263 +179,35 @@ func keepingAssignments(ctx context.Context, inv *inventory.Inventory, announce
events.known = now
}
// declaredPackagesMirror is the source of the declared-packages mirror (ADR 0507 §5) over a fresh reading of the stores.
// declaredPackagesMirror is the source of the declared-packages mirror (ADR 0507 §5): each node's composition, asked as a
// question that makes nothing, and the packages it declares with the modules that declare them. A node whose own set does
// not compose has no key, and any other failure to read is the pass's error, so a key is never purged for a store that
// could not be read.
func declaredPackagesMirror(open *stores) func(context.Context) (map[string][]byte, error) {
return func(ctx context.Context) (map[string][]byte, error) {
in, err := inputFor(ctx, open)
if err != nil {
return nil, err
}
return declaredPackagesFrom(ctx, in)
}
}
// declaredPackagesFrom is the declared-packages mirror over an Input: each node's composition, asked as a question that makes
// nothing, and the packages it declares with the modules that declare them. A node whose own set does not compose has no key,
// and any other failure to read is the pass's error, so a key is never purged for a store that could not be read.
func declaredPackagesFrom(_ context.Context, in *meshInput) (map[string][]byte, error) {
// Which module owns each resource is carried only for a node-engine that reads it, so every node is asked as one that does.
asked := in.Input
asked.Nodes = append([]compose.Node(nil), in.Nodes...)
for i := range asked.Nodes {
asked.Nodes[i].ReadsModules = true
}
out := map[string][]byte{}
for _, n := range in.sortedNodes() {
d, _, err := compose.ComposeIn(asked, n.Name, compose.Foreseeing)
if compose.IsUnresolvable(err) {
continue
// Which module owns each resource is carried only for a node-engine that reads it, so every node is asked as one that does.
asked := in.Input
asked.Nodes = append([]compose.Node(nil), in.Nodes...)
for i := range asked.Nodes {
asked.Nodes[i].ReadsModules = true
}
if err != nil {
return nil, fmt.Errorf("the packages %s declares cannot be read: %w", n.Name, err)
}
if out[n.Name], err = inventory.DeclaredPackagesValue(n.Name, d.Resources, d.ModuleOf); err != nil {
return nil, err
}
}
return out, nil
}
// **The declared-packages mirror passes on a change, not on a tick** (issue 1945 note 22). Its source composes every node in
// full, about 45 s on the control node; on the 10 s tick it never rested and held the controller at 130-175 % CPU. Every
// change that can move a node's packages is a change of its composition, which the controller writes into the bucket of the
// compositions' digests, and of the assignments, which the verbs write into theirs: a watch of those two buckets wakes the
// keeper. A pass reads the stores' Input once and composes only when the Input's digest differs from the one it last
// mirrored, so a wake that changed nothing it composes from costs one reading. One pass at a time; the wakes during a pass
// are one pass after it, at least packagesGap after the last began. A rare safety pass (packagesSafetyEvery) covers a
// change no watched key says, and with an unchanged digest composes nothing either.
const (
packagesSafetyEvery = 10 * time.Minute
packagesGap = 30 * time.Second
)
// packagesTiming is when the declared-packages mirror passes: the safety pass's period and the least time between the
// starts of two passes.
type packagesTiming struct{ safety, gap time.Duration }
// packagesKeeper keeps the declared-packages mirror: read is the Input, compose the mirror's values from it, sync writes
// those that differ from the bucket, holds says this controller holds the lease. Each a seam a test stands in for.
type packagesKeeper struct {
read func(ctx context.Context) (*meshInput, error)
compose func(ctx context.Context, in *meshInput) (map[string][]byte, error)
sync func(ctx context.Context, want map[string][]byte) error
holds func() bool
wake chan struct{}
// last is the digest of the Input last mirrored whole; empty after a failure or while the lease is not held, so the
// next pass composes.
last string
}
func newPackagesKeeper(read func(context.Context) (*meshInput, error),
compose func(context.Context, *meshInput) (map[string][]byte, error),
sync func(context.Context, map[string][]byte) error, holds func() bool) *packagesKeeper {
return &packagesKeeper{read: read, compose: compose, sync: sync, holds: holds, wake: make(chan struct{}, 1)}
}
// poke says something the mirror composes from may have changed; never blocks, and pokes before a pass are one.
func (k *packagesKeeper) poke() {
select {
case k.wake <- struct{}{}:
default:
}
}
// pass brings the mirror to the Input, composing only when the Input differs from the one last mirrored.
func (k *packagesKeeper) pass(ctx context.Context) error {
if !k.holds() {
k.last = ""
return nil
}
in, err := k.read(ctx)
if err != nil {
k.last = ""
return fmt.Errorf("the store cannot be read: %w", err)
}
digest := inputDigest(in)
if digest != "" && digest == k.last {
return nil
}
want, err := k.compose(ctx, in)
if err != nil {
k.last = ""
return err
}
if err := k.sync(ctx, want); err != nil {
k.last = ""
return err
}
k.last = digest
return nil
}
// run passes once at its start, then on each wake and each safety period, until ctx ends. said hears why a pass failed,
// and an empty line when one succeeds again.
func (k *packagesKeeper) run(ctx context.Context, at packagesTiming, said func(why string)) {
safety := time.NewTicker(at.safety)
defer safety.Stop()
failing := ""
var began time.Time
for {
// The wakes so far are this pass's: drained before it reads, so a change after is a pass of its own.
select {
case <-k.wake:
default:
}
began = time.Now()
why := ""
if err := k.pass(ctx); err != nil {
why = err.Error()
}
if why != failing {
said(why)
failing = why
}
select {
case <-ctx.Done():
return
case <-k.wake:
case <-safety.C:
}
if rest := at.gap - time.Since(began); rest > 0 {
select {
case <-ctx.Done():
return
case <-time.After(rest):
out := map[string][]byte{}
for _, n := range in.sortedNodes() {
d, _, err := compose.ComposeIn(asked, n.Name, compose.Foreseeing)
if compose.IsUnresolvable(err) {
continue
}
}
}
}
// inputDigest is a digest of everything the composition reads (the Input, and why a node or the bus is refused): the same
// for the same Input. Empty when it cannot be made, and then the pass composes.
func inputDigest(in *meshInput) string {
refused := map[string]string{}
for node, err := range in.refused {
if err != nil {
refused[node] = err.Error()
}
}
bus := ""
if in.busRefused != nil {
bus = in.busRefused.Error()
}
body, err := json.Marshal(struct {
Input compose.Input
Refused map[string]string
BusRefused string
}{in.Input, refused, bus})
if err != nil {
return ""
}
sum := sha256.Sum256(body)
return hex.EncodeToString(sum[:])
}
// keepingDeclaredPackages keeps the declared-packages mirror for as long as ctx: woken by the watch of the compositions'
// digests and of the assignments. It answers the keeper, whose poke a caller may also wake it with.
func keepingDeclaredPackages(ctx context.Context, open *stores, api jetstream.JetStream, announce recordmirror.Announcer,
epoch func() uint64, holds func() bool) *packagesKeeper {
m := recordmirror.DeclaredPackages
var w *recordmirror.Writer
read := func(ctx context.Context) (*meshInput, error) {
reading, cancel := context.WithTimeout(ctx, 2*time.Minute)
defer cancel()
return readComposeInput(reading, open)
}
sync := func(ctx context.Context, want map[string][]byte) error {
writing, cancel := context.WithTimeout(ctx, 30*time.Second)
defer cancel()
if w == nil {
kv, err := api.KeyValue(writing, m.Bucket)
if err != nil {
return fmt.Errorf("the bucket %s cannot be read: %v", m.Bucket, err)
return nil, fmt.Errorf("the packages %s declares cannot be read: %w", n.Name, err)
}
w = &recordmirror.Writer{Mirror: m, KV: kv, Announce: announce, Epoch: epoch, Holds: holds}
}
lost := w.LostEvents
r, err := w.Sync(writing, want)
if r.Written+r.Removed > 0 {
fmt.Printf("the %s mirror moved: %d written, %d removed\n", m.Name, r.Written, r.Removed)
}
if w.LostEvents > lost {
fmt.Printf("the %s mirror lost %d event(s): the bucket holds the change, its notice was refused twice\n",
m.Name, w.LostEvents-lost)
}
return err
}
k := newPackagesKeeper(read, declaredPackagesFrom, sync, holds)
for _, bucket := range []string{sdkdelivery.ControllerComposedBucket, broker.AssignmentsBucket} {
goServing(ctx, func() { wakeOnBucket(ctx, api, bucket, k.poke) })
}
goServing(ctx, func() {
k.run(ctx, packagesTiming{safety: packagesSafetyEvery, gap: packagesGap}, func(why string) {
if why != "" {
fmt.Printf("the %s mirror cannot be kept: %s\n", m.Name, why)
} else {
fmt.Printf("the %s mirror is kept again\n", m.Name)
}
})
})
return k
}
// wakeOnBucket calls wake on every write or delete of a key in the bucket, for as long as ctx: only the keys' metadata is
// watched. A bucket not there yet, or a watch that ends, is watched again a minute later; the safety pass covers the gap.
func wakeOnBucket(ctx context.Context, api jetstream.JetStream, bucket string, wake func()) {
for {
if err := watchBucketOnce(ctx, api, bucket, wake); err != nil && ctx.Err() == nil {
fmt.Printf("the declared-packages mirror does not hear %s: %v\n", bucket, err)
}
select {
case <-ctx.Done():
return
case <-time.After(time.Minute):
}
}
}
// watchBucketOnce watches the bucket until its watch ends or ctx does.
func watchBucketOnce(ctx context.Context, api jetstream.JetStream, bucket string, wake func()) error {
kv, err := api.KeyValue(ctx, bucket)
if err != nil {
return err
}
watch, err := kv.WatchAll(ctx, jetstream.UpdatesOnly(), jetstream.MetaOnly())
if err != nil {
return err
}
defer func() { _ = watch.Stop() }()
for {
select {
case <-ctx.Done():
return nil
case e, ok := <-watch.Updates():
if !ok {
return fmt.Errorf("the watch ended")
}
if e != nil {
wake()
if out[n.Name], err = inventory.DeclaredPackagesValue(n.Name, d.Resources, d.ModuleOf); err != nil {
return nil, err
}
}
return out, nil
}
}
-191
View File
@@ -1,191 +0,0 @@
package main
import (
"context"
"fmt"
"sync"
"sync/atomic"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/compose"
)
// The declared-packages mirror composes only when what it composes from changed (issue 1945 note 22: it composed every node
// in full every 10 s, a pass of about 45 s, so the controller on the control node never rested at 130-175 % CPU).
// scaled is a cadence of the mirror with a second taken as a millisecond, so a test's minute is 60 ms.
func scaled() packagesTiming {
return packagesTiming{safety: packagesSafetyEvery / 1000, gap: packagesGap / 1000}
}
const scaledMinute = 60 * time.Millisecond
// fakeMesh is the keeper's input as a test changes it: each node's assigned modules, and the compositions counted by node.
type fakeMesh struct {
mu sync.Mutex
assigned map[string][]string
composed atomic.Int64 // node compositions
synced atomic.Int64
lastSync map[string][]byte
// block, when set, holds the next composition until it is closed; entered is closed when that composition began.
block chan struct{}
entered chan struct{}
}
func newFakeMesh(nodes int) *fakeMesh {
f := &fakeMesh{assigned: map[string][]string{}}
for i := 0; i < nodes; i++ {
f.assigned[fmt.Sprintf("node%d", i)] = []string{"base"}
}
return f
}
func (f *fakeMesh) assign(node, module string) {
f.mu.Lock()
defer f.mu.Unlock()
f.assigned[node] = append(append([]string(nil), f.assigned[node]...), module)
}
func (f *fakeMesh) keeper() *packagesKeeper {
read := func(context.Context) (*meshInput, error) {
f.mu.Lock()
defer f.mu.Unlock()
in := compose.Input{Catalogue: map[string]catalogue.Manifest{"base": {Module: "base", Version: "1"}}}
for _, n := range nodesOfFake(f.assigned) {
in.Nodes = append(in.Nodes, compose.Node{Name: n, Assigned: append([]string(nil), f.assigned[n]...)})
}
return &meshInput{Input: in}, nil
}
composeAll := func(_ context.Context, in *meshInput) (map[string][]byte, error) {
f.mu.Lock()
block, entered := f.block, f.entered
f.block = nil
f.mu.Unlock()
if block != nil {
close(entered)
<-block
}
out := map[string][]byte{}
for _, n := range in.Nodes {
f.composed.Add(1)
out[n.Name] = []byte(fmt.Sprint(n.Assigned))
}
return out, nil
}
syncAll := func(_ context.Context, want map[string][]byte) error {
f.mu.Lock()
f.lastSync = want
f.mu.Unlock()
f.synced.Add(1)
return nil
}
return newPackagesKeeper(read, composeAll, syncAll, func() bool { return true })
}
func nodesOfFake(m map[string][]string) []string {
out := make([]string, 0, len(m))
for k := range m {
out = append(out, k)
}
for i := range out {
for j := i + 1; j < len(out); j++ {
if out[j] < out[i] {
out[i], out[j] = out[j], out[i]
}
}
}
return out
}
func runKeeper(t *testing.T, k *packagesKeeper) context.CancelFunc {
ctx, cancel := context.WithCancel(context.Background())
done := make(chan struct{})
go func() {
defer close(done)
k.run(ctx, scaled(), func(string) {})
}()
t.Cleanup(func() {
cancel()
<-done
})
return cancel
}
func eventually(t *testing.T, what string, cond func() bool) {
t.Helper()
deadline := time.Now().Add(time.Second)
for !cond() {
if time.Now().After(deadline) {
t.Fatalf("not within a second: %s", what)
}
time.Sleep(time.Millisecond)
}
}
// 1945/AC-mirror-1: given a mesh whose input does not change, when a minute passes after the mirror's first pass, then it
// composes no node (before: a pass every 10 s, each composing every node). The count per minute is said.
func TestTheDeclaredPackagesMirrorComposesNothingWithoutAChange(t *testing.T) {
const nodes = 5
f := newFakeMesh(nodes)
runKeeper(t, f.keeper())
eventually(t, "the first pass", func() bool { return f.synced.Load() >= 1 })
start := f.composed.Load()
time.Sleep(scaledMinute)
perMinute := f.composed.Load() - start
t.Logf("node compositions in a minute of an unchanged mesh of %d nodes: %d", nodes, perMinute)
if perMinute != 0 {
t.Errorf("an unchanged mesh of %d nodes was composed %d times in a minute; want 0", nodes, perMinute)
}
}
// 1945/AC-mirror-2: given a pass that is running, when fifty changes arrive during it, then exactly one more pass follows,
// never two at once and never one per change.
func TestTheDeclaredPackagesMirrorPassesOncePerBurst(t *testing.T) {
f := newFakeMesh(1)
release := make(chan struct{})
f.block, f.entered = release, make(chan struct{})
k := f.keeper()
runKeeper(t, k)
<-f.entered
for i := 0; i < 50; i++ {
f.assign("node0", fmt.Sprintf("m%d", i))
k.poke()
}
close(release)
eventually(t, "the pass after the burst", func() bool { return f.synced.Load() >= 2 })
time.Sleep(2*scaled().gap + 20*time.Millisecond)
if got := f.synced.Load(); got != 2 {
t.Errorf("a burst of 50 changes during a pass gave %d passes in all; want 2 (the running one and one after)", got)
}
}
// 1945/AC-mirror-3: given the mirror at rest, when an input changes and its event arrives, then the change is mirrored.
func TestTheDeclaredPackagesMirrorMirrorsAChange(t *testing.T) {
f := newFakeMesh(2)
k := f.keeper()
runKeeper(t, k)
eventually(t, "the first pass", func() bool { return f.synced.Load() >= 1 })
f.assign("node1", "viewer")
k.poke()
eventually(t, "the change mirrored", func() bool {
f.mu.Lock()
defer f.mu.Unlock()
return string(f.lastSync["node1"]) == "[base viewer]"
})
}
// 1945/AC-mirror-4: the digest of the input is the same for the same input and differs when an assignment differs.
func TestTheInputDigestFollowsTheInput(t *testing.T) {
of := func(assigned ...string) *meshInput {
return &meshInput{Input: compose.Input{Nodes: []compose.Node{{Name: "anchor", Assigned: assigned}}}}
}
a, b, c := inputDigest(of("base")), inputDigest(of("base")), inputDigest(of("base", "viewer"))
if a == "" || a != b {
t.Errorf("the same input gave the digests %q and %q", a, b)
}
if a == c {
t.Errorf("an added assignment left the digest at %q", a)
}
}
-60
View File
@@ -1,60 +0,0 @@
package main
import (
"context"
"sync"
)
// **Every goroutine the serving controller starts ends before serve returns** (issue 1945). A loop or a watch started
// beside the serving runs until serve's context ends, and serve's stop waits for it before it closes the stores, the
// bus and the condition keeper those loops read: a loop that outlived its serve met the next serve's globals, its
// closed stores and its bus, and in the test process that is the next test (a race on conditionsFrom, a package that
// passed every test and still exited failing). Where it is checked: the tests that run serve whole, under -race.
// servingGroup counts the goroutines one serve started. Once its stop began it starts nothing more: what would start is
// work for a controller that is stopping.
type servingGroup struct {
mu sync.Mutex
stopped bool
wg sync.WaitGroup
}
type servingGroupKey struct{}
// withServingGroup is ctx carrying a new group, which goServing under it counts.
func withServingGroup(ctx context.Context) (context.Context, *servingGroup) {
g := &servingGroup{}
return context.WithValue(ctx, servingGroupKey{}, g), g
}
// 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 {
g, _ := ctx.Value(servingGroupKey{}).(*servingGroup)
if g == nil {
go work()
return true
}
g.mu.Lock()
if g.stopped {
g.mu.Unlock()
return false
}
g.wg.Add(1)
g.mu.Unlock()
go func() {
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
// for nothing.
func (g *servingGroup) stop() {
g.mu.Lock()
g.stopped = true
g.mu.Unlock()
g.wg.Wait()
}
+52
View File
@@ -0,0 +1,52 @@
package main
import (
"context"
"sync"
"sync/atomic"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"github.com/novox/mesh-controller/internal/inventory"
)
// The take-ins share one reading of how the sources are spelled (the forge's address): a burst of 120 take-ins reads the
// mesh once, not once each (issues 1945 and 1277: each take-in's fresh read opened pools of its own at 16:08 on
// 2026-10-25). A reading older than sourceFormsKept is made again, so a moved forge is read.
func TestTakeInsShareOneReadingOfTheSources(t *testing.T) {
was, wasNow := sourceFormsFor, sourceFormsNow
t.Cleanup(func() { sourceFormsFor, sourceFormsNow = was, wasNow; sharedForms = sync.Map{} })
var made, worldReads atomic.Int64
sourceFormsFor = func(ctx context.Context, _ *inventory.Inventory) *sourceForms {
made.Add(1)
return sourceFormsRead(ctx, time.Now, func(context.Context) (catalogue.World, error) {
worldReads.Add(1)
return catalogue.World{}, nil
})
}
clock := time.Date(2026, 10, 25, 16, 8, 0, 0, time.UTC)
var clockMu sync.Mutex
sourceFormsNow = func() time.Time { clockMu.Lock(); defer clockMu.Unlock(); return clock }
inv := &inventory.Inventory{}
var wg sync.WaitGroup
for range 120 {
wg.Add(1)
go func() {
defer wg.Done()
f := formsOfSources(context.Background(), inv)
_, _ = f.read(context.Background(), gitSeat, true)
}()
}
wg.Wait()
if n := made.Load(); n != 1 {
t.Fatalf("made %d readings of the sources for 120 take-ins, want one", n)
}
clockMu.Lock()
clock = clock.Add(sourceFormsKept + time.Second)
clockMu.Unlock()
formsOfSources(context.Background(), inv)
if n := made.Load(); n != 2 {
t.Fatalf("a reading older than %s was kept (%d made)", sourceFormsKept, n)
}
}
+3 -3
View File
@@ -67,16 +67,16 @@ 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 {
return fmt.Errorf("following the states announced: %w", err)
}
goServing(ctx, func() {
go func() {
<-ctx.Done()
consuming.Stop()
})
}()
return nil
}
+1 -1
View File
@@ -210,7 +210,7 @@ func (l nudgingListener) Heard(ctx context.Context, report link.Report) (bool, e
// And, once the move of the secrets into the vault has started, what is left of this machine's (design 57's V3,
// issue 676): a machine asleep when it started is moved when it is next heard from. Off the report's path.
if err == nil && l.open != nil && report.Superseded == "" && report.Rekey == nil && report.Node != "" {
goServing(ctx, func() { moveWhenHeard(context.WithoutCancel(ctx), l.open, moveHeardOn(), report.Node) })
go moveWhenHeard(context.WithoutCancel(ctx), l.open, moveHeardOn(), report.Node)
}
return news, err
}
+11 -26
View File
@@ -545,9 +545,10 @@ func later(a, b time.Time) time.Time {
return b
}
// openServingConditions opens the condition store for the serving controller and makes it conditionsFrom. Nil when the store
// could not be opened, which is said: status then says the conditions cannot be read.
func openServingConditions(ctx context.Context, open *stores, bus link.OverNATS) *conditions.Keeper {
// watchTheMesh opens the condition store and starts the watchdogs, the bus's advisories and the
// self-check, for the serving controller; the returned function stops them. Nil when the store could
// not be opened, which is said.
func watchTheMesh(ctx context.Context, open *stores, server *link.Server, bus link.OverNATS) func() {
keeper, err := keeperOn(ctx, bus.Conn, storeLooks{open.inventory})
if err != nil {
fmt.Printf("the condition store could NOT be opened, so nothing that goes wrong is kept or said, and "+
@@ -555,22 +556,6 @@ func openServingConditions(ctx context.Context, open *stores, bus link.OverNATS)
return nil
}
conditionsFrom = keeper
return keeper
}
// closeConditions flushes what the keeper holds and closes it, once nothing that raises a condition runs.
func closeConditions(keeper *conditions.Keeper) {
flushing, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
keeper.Close(flushing)
}
// watchTheMesh starts the watchdogs, the bus's advisories and the self-check over the condition store's keeper, for the
// serving controller; the returned function stops them and waits for them to end. Nil when there is no keeper.
func watchTheMesh(ctx context.Context, open *stores, server *link.Server, bus link.OverNATS, keeper *conditions.Keeper) func() {
if keeper == nil {
return nil
}
logf := func(format string, args ...any) { fmt.Printf(format+"\n", args...) }
stopHearing, err := server.HearAdvisories(logf)
if err != nil {
@@ -583,26 +568,26 @@ func watchTheMesh(ctx context.Context, open *stores, server *link.Server, bus li
d := &doctor{open: open, js: server.JetStream(), keeper: keeper, teller: bus, watchdogs: w, host: host}
w.doctor = d
doctorFrom = d
// Its own group as well as serve's: its stop waits for what it started.
watching, stop := context.WithCancel(ctx)
watching, running := withServingGroup(watching)
goServing(watching, func() { w.keep(watching) })
goServing(watching, func() { d.keep(watching) })
go w.keep(watching)
go d.keep(watching)
// And the healers (novox/hq to-be 45 §7): what is open that a registered healer answers, repaired
// under the lease and the brake, every act said.
healers := newHealing(open, keeper, bus, server.JetStream())
goServing(watching, func() { healers.keep(watching) })
go healers.keep(watching)
// And the asker (novox/hq ADR 0259): what needs the operator and names its answers is asked of them,
// and the answer chosen is performed on its warrant.
startAsking(watching, open, server, bus.Conn, keeper)
goServing(watching, func() { forgettingOldHeals(watching, open.inventory) })
go forgettingOldHeals(watching, open.inventory)
fmt.Printf("watching the mesh: %d signal(s) every %s, %d probe(s) every %s; what is wrong is kept in %s "+
"and said as %s events\n", len(watchedRows()), watchEvery, len(runnableProbes()), doctorEvery,
broker.ConditionsBucket, conditions.Seat)
return func() {
stop()
running.stop()
stopHearing()
flushing, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
keeper.Close(flushing)
}
}
-31
View File
@@ -114,37 +114,6 @@ func (i *Inventory) writeBusUser(ctx context.Context, u BusUser, hash string) er
return nil
}
// MintBusPasswordIfAbsent mints u's password only when no row names u yet, in one statement: of two callers that both
// found the account missing, exactly one wins, and the other is told it lost (won false, no password) and must seal
// nothing, so the hash the bus holds is always the password the node is sealed (novox/hq ADR 0543, the review of
// issue 1945: the assign act and the assignments' watch once each minted, the later hash beside the earlier seal).
func (i *Inventory) MintBusPasswordIfAbsent(ctx context.Context, u BusUser) (string, bool, error) {
if u.Username == "" || u.Kind == "" {
return "", false, errors.New("a bus user needs a username and a kind")
}
raw := make([]byte, 32)
if _, err := rand.Read(raw); err != nil {
return "", false, fmt.Errorf("cannot generate a bus password: %w", err)
}
password := base64.RawURLEncoding.EncodeToString(raw)
hash, err := hashBusPassword(password)
if err != nil {
return "", false, fmt.Errorf("cannot hash a bus password: %w", err)
}
tag, err := i.store.Pool().Exec(ctx,
`insert into bus_user (username, kind, node, module, password_hash)
values ($1, $2, $3, $4, $5)
on conflict (username) do nothing`,
u.Username, u.Kind, u.Node, u.Module, string(hash))
if err != nil {
return "", false, fmt.Errorf("cannot record the bus user %s: %w", u.Username, err)
}
if tag.RowsAffected() == 0 {
return "", false, nil
}
return password, true, nil
}
func (i *Inventory) MintBusPassword(ctx context.Context, u BusUser) (string, error) {
if u.Username == "" || u.Kind == "" {
return "", errors.New("a bus user needs a username and a kind")
@@ -33,7 +33,7 @@ func TestABusPasswordIsReturnedOnceAndOnlyItsHashIsKept(t *testing.T) {
}
}
func TestMintingAgainRotatesAndMintingIfAbsentHasOneWinner(t *testing.T) {
func TestMintingAgainRotatesRatherThanAddsAUser(t *testing.T) {
inv := ForTest(t)
ctx := context.Background()
@@ -62,50 +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")
}
// **Minting only if absent never rotates, and of several at once exactly one wins** (ADR 0543, the review of issue
// 1945): the assign act and the assignments' watch may both find an account missing; the winner's password is the
// one the store's hash verifies, so the one it seals is the one the bus accepts, and a loser seals nothing.
if _, won, err := inv.MintBusPasswordIfAbsent(ctx, aBusUser("shop")); err != nil || won {
t.Fatalf("a mint-if-absent over a minted account won %v (%v): it would replace a sealed password", won, err)
}
const racers = 8
type minted struct {
password string
won bool
err error
}
results := make(chan minted, racers)
for range racers {
go func() {
p, won, err := inv.MintBusPasswordIfAbsent(ctx, aBusUser("racer"))
results <- minted{p, won, err}
}()
}
var winner string
wins := 0
for range racers {
r := <-results
if r.err != nil {
t.Fatal(r.err)
}
if r.won {
wins++
winner = r.password
} else if r.password != "" {
t.Fatal("a caller that lost was handed a password to seal")
}
}
if wins != 1 {
t.Fatalf("%d of %d concurrent mints won, want exactly one", wins, racers)
}
stored, _, err := inv.BusUserHash(ctx, "one.racer")
if err != nil {
t.Fatal(err)
}
if err := bcrypt.CompareHashAndPassword([]byte(stored), []byte(winner)); err != nil {
t.Fatal("the stored hash is not the winner's password: the node would be sealed a password the bus refuses")
}
}
func TestForgettingANodeTakesItsBusUsersWithIt(t *testing.T) {
-10
View File
@@ -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)
}