Take new builds in from a durable queue on the bus, register on the fingerprint, and issue a node's bus accounts when its assignment stands (issue 1945, ADR 0543, ADR 0544) #465

Merged
jschoubben merged 9 commits from fix/1945-take-in-consumer into main 2026-10-25 18:06:47 +00:00
8 changed files with 389 additions and 223 deletions
Showing only changes of commit 9ff47f6aaa - Show all commits
+2
View File
@@ -399,6 +399,8 @@ func blockedElsewhere(ctx context.Context, open *stores, except string) string {
// purpose, never by re-assigning); and when the bus cannot be reached from here, the line names the
// verb and the push that would refuse the module until it is run — never a silent placeholder (novox/hq issue 203).
func issueOnAssign(ctx context.Context, open *stores, node, module string) string {
// Mint only what no row names: the assignments' watch hears the same write and may issue at once (ADR 0543).
ctx = onlyIfAbsent(ctx)
inv := open.inventory
shelf, err := inv.Catalogue(ctx)
if err != nil {
+114 -47
View File
@@ -3,7 +3,9 @@ package main
import (
"context"
"encoding/json"
"errors"
"fmt"
"strings"
"sync"
"time"
@@ -28,15 +30,20 @@ import (
// (a new key, or one that was ended or pending), the lease holder issues the module's accounts on that node that are not
// minted (idempotent: a minted account is never issued again, a rotation is on purpose), composes the node holding the
// bus so its user list carries them, and writes the grants. A rewrite of an assignment that already stood (the
// provisioner's plan kept in its value) does nothing. A starting controller looks at every standing assignment once:
// it issues what is missing and, when it issued anything, composes the bus's node and writes the grants.
// 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
// issue issues the module's accounts on the node that are not minted; issued says it minted one, line what it did.
issue func(ctx context.Context, node, module string) (issued bool, line string, err error)
// 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.
@@ -48,39 +55,89 @@ type assignmentAccountSteps struct {
// bus's node is composed and the grants written even when every account was minted before (a module assigned again
// keeps its account, and the bus's user list left it out while it was unassigned). It answers what it did, "" for
// nothing.
func (s assignmentAccountSteps) settle(ctx context.Context, node, module string, fresh bool) (string, error) {
if !s.holds(ctx) || module == controllerModule {
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
}
issued, line, err := s.issue(ctx, node, module)
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)
}
if !issued && !fresh {
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 line, fmt.Errorf("the node holding the bus cannot be read: %w", err)
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 line, fmt.Errorf("the bus's user list on %s is not composed: %w", holder, err)
return "", fmt.Errorf("the bus's user list on %s is not composed: %w", holder, err)
}
}
if err := s.grants(ctx); err != nil {
return line, err
}
said := module + " on " + node + ": "
if line != "" {
said += line + "; "
return "", err
}
if holder != "" {
said += "the bus's user list composed on " + holder + ", "
return "the bus's user list composed on " + holder + ", the grants written", nil
}
return said + "its grant 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
@@ -130,11 +187,10 @@ func ownAccountOf(m catalogue.Manifest, node string) (string, bool) {
// issueAssignmentAccounts issues module's accounts on node that are not minted: its own (issueOnTheNewBus) and its
// processes' (issueMissingProcessAccounts). A module the catalogue does not hold, or one with nowhere to deliver an
// account (mayIssue), has nothing to issue; an account minted before is kept.
func issueAssignmentAccounts(ctx context.Context, inv *inventory.Inventory, node, module string) (bool, string, error) {
shelf, err := inv.Catalogue(ctx)
if err != nil {
return false, "", fmt.Errorf("the catalogue cannot be read: %w", err)
}
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
@@ -169,11 +225,13 @@ func liveAssignmentAccountSteps(open *stores) assignmentAccountSteps {
inv := open.inventory
return assignmentAccountSteps{
holds: func(ctx context.Context) bool {
epoch, err := theLease.epoch(ctx)
return err == nil && epoch > 0
// 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
},
issue: func(ctx context.Context, node, module string) (bool, string, error) {
return issueAssignmentAccounts(ctx, inv, node, module)
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)
@@ -194,21 +252,23 @@ func liveAssignmentAccountSteps(open *stores) assignmentAccountSteps {
// assignmentAccountsRetry is how long a settle that failed waits before it tries again.
var assignmentAccountsRetry = 30 * time.Second
// settleAssignment settles one assignment until it works or ctx ends: a failure is said and tried again every
// assignmentAccountsRetry, never dropped, since no later write of the key may come.
func settleAssignment(ctx context.Context, steps assignmentAccountSteps, node, module string, fresh bool) {
// 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 := steps.settle(ctx, node, module, fresh)
said, err := settle(ctx)
if err == nil {
if said != "" {
fmt.Printf("assignments: %s\n", said)
for _, line := range strings.Split(said, "\n") {
fmt.Printf("assignments: %s\n", line)
}
}
return
}
if ctx.Err() != nil {
return
}
fmt.Printf("assignments: %s on %s: %v; tried again in %s\n", module, node, err, assignmentAccountsRetry)
fmt.Printf("assignments: %s: %v; tried again in %s\n", what, err, assignmentAccountsRetry)
select {
case <-ctx.Done():
return
@@ -224,30 +284,33 @@ func watchAssignmentAccounts(ctx context.Context, open *stores) {
steps := liveAssignmentAccountSteps(open)
turns := &assignmentTurns{}
goServing(ctx, func() {
first := true
for {
var atStart []chain.Assignment
if err := watchAssignmentsOnce(ctx, open.inventory.WatchAssignments, turns, func(a chain.Assignment, fresh bool) {
if err := watchAssignmentsOnce(ctx, open.inventory.WatchAssignments, turns, first, func(a chain.Assignment, fresh bool) {
if !fresh {
atStart = append(atStart, a)
return
}
inBackground(ctx, func() { settleAssignment(ctx, steps, a.Node, a.Module, true) })
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() {
// The start's look, one assignment after another: a reading of the catalogue each, never all at once
// (issues 1945 and 1277: a burst of take-ins once filled the store).
// 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).
look := atStart
atStart = nil
inBackground(ctx, func() {
for _, a := range look {
if ctx.Err() != nil {
return
}
settleAssignment(ctx, steps, a.Node, a.Module, false)
}
untilSettled(ctx, "the start's look", func(ctx context.Context) (string, error) {
return steps.look(ctx, look)
})
})
}); err != nil && ctx.Err() == nil {
fmt.Printf("assignments: the assignments are not heard, so a new assignment's bus accounts wait: %v\n", err)
}
first = false
select {
case <-ctx.Done():
return
@@ -258,10 +321,12 @@ func watchAssignmentAccounts(ctx context.Context, open *stores) {
}
// watchAssignmentsOnce watches the assignments until the watch or ctx ends, handing each assignment that comes to stand
// to settle: fresh for one written after the current values were read, not fresh for one of them, and calling read once
// the current values are all handed (the chain's Caught mark).
// 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, settle func(a chain.Assignment, fresh bool), read func()) 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)
@@ -283,7 +348,9 @@ func watchAssignmentsOnce(ctx context.Context, watch func(context.Context) (<-ch
if e.Caught {
if current {
current = false // the current values are read: what follows is written from now on
read()
if first {
read()
}
}
continue
}
@@ -292,7 +359,7 @@ func watchAssignmentsOnce(ctx context.Context, watch func(context.Context) (<-ch
value = e.Value
}
if a, turned := turns.heard(e.Key, value); turned {
settle(a, !current)
settle(a, !current || !first)
}
}
}
+164 -108
View File
@@ -4,6 +4,7 @@ import (
"context"
"encoding/json"
"errors"
"fmt"
"strings"
"testing"
"time"
@@ -27,8 +28,11 @@ func anAssignmentValue(t *testing.T, a chain.Assignment) []byte {
}
// Only a write that makes the key stand turns it: a rewrite of a standing assignment (the provisioner's plan kept in its
// value), an ended one, a pending one or a module only pulled in by another does not.
func TestAnAssignmentTurnsWhenItComesToStand(t *testing.T) {
// 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
@@ -43,7 +47,7 @@ func TestAnAssignmentTurnsWhenItComesToStand(t *testing.T) {
value *chain.Assignment // nil for a delete
turning bool
}
cases := []struct {
for _, c := range []struct {
name string
writes []write
}{
@@ -54,8 +58,7 @@ func TestAnAssignmentTurnsWhenItComesToStand(t *testing.T) {
{"a pending one does not, and turns once made", []write{{&pending, false}, {&stands, true}}},
{"a module only pulled in does not", []write{{&pulled, false}}},
{"one deleted and written again turns again", []write{{&stands, true}, {nil, false}, {&stands, true}}},
}
for _, c := range cases {
} {
t.Run(c.name, func(t *testing.T) {
turns := &assignmentTurns{}
for i, w := range c.writes {
@@ -73,50 +76,119 @@ func TestAnAssignmentTurnsWhenItComesToStand(t *testing.T) {
}
})
}
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)
}
})
}
}
// settle issues what is missing and then puts it on the bus: the node holding the bus composed (its user list) and the
// grants written; for an assignment that just came to stand even when its accounts were minted before (a module
// assigned again keeps its account), and never by a controller that does not hold the lease.
func TestSettlingAnAssignmentIssuesComposesTheBusAndGrants(t *testing.T) {
// 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
holds, fresh bool
notHolding bool
issued bool
issueErr, composeE error
holder string
wantIssues int
wantComposed []string
wantGrants bool
wantErr string
}{
{name: "a new assignment whose account is not minted", module: "mesh-build-node", holds: true, fresh: true,
issued: true, holder: "novox", wantComposed: []string{"novox"}, wantGrants: true},
{name: "a new assignment whose account was minted before", module: "mesh-build-go", holds: true, fresh: true,
holder: "novox", wantComposed: []string{"novox"}, wantGrants: true},
{name: "the start's look issuing a missing account", module: "mesh-build-node", holds: true, issued: true,
holder: "novox", wantComposed: []string{"novox"}, wantGrants: true},
{name: "the start's look with every account minted does nothing", module: "mesh-build-go", holds: true,
holder: "novox"},
{name: "no lease, nothing", module: "mesh-build-node", fresh: true, issued: true, holder: "novox"},
{name: "the controller's own module is the terminal's", module: controllerModule, holds: true, fresh: true,
issued: true, holder: "novox"},
{name: "no bus holder: the grants still", module: "mesh-build-node", holds: true, fresh: true, issued: true,
wantGrants: true},
{name: "an issue that fails is the error, nothing composed", module: "mesh-build-node", holds: true, fresh: true,
issueErr: errors.New("the bus cannot be reached"), holder: "novox", wantErr: "the bus cannot be reached"},
{name: "a composition that fails is the error, no grant", module: "mesh-build-node", holds: true, fresh: true,
issued: true, holder: "novox", composeE: errors.New("store refused"), wantComposed: []string{"novox"},
wantErr: "store refused"},
{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
issuedFor, grants := "", false
issues, catalogues, grants := 0, 0, 0
steps := assignmentAccountSteps{
holds: func(context.Context) bool { return c.holds },
issue: func(_ context.Context, node, module string) (bool, string, error) {
issuedFor = module + "@" + node
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
}
@@ -127,9 +199,15 @@ func TestSettlingAnAssignmentIssuesComposesTheBusAndGrants(t *testing.T) {
composed = append(composed, nodes...)
return c.composeE
},
grants: func(context.Context) error { grants = true; return nil },
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)
}
said, err := steps.settle(t.Context(), "ace", c.module, c.fresh)
if c.wantErr != "" {
if err == nil || !strings.Contains(err.Error(), c.wantErr) {
t.Fatalf("error %v, want one saying %q", err, c.wantErr)
@@ -137,29 +215,30 @@ func TestSettlingAnAssignmentIssuesComposesTheBusAndGrants(t *testing.T) {
} 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 grants != c.wantGrants {
t.Fatalf("grants written %v, want %v", grants, c.wantGrants)
if want := map[bool]int{true: 1}[c.wantGrants]; grants != want {
t.Fatalf("grants written %d time(s), want %d", grants, want)
}
skipped := !c.holds || c.module == controllerModule
if skipped != (issuedFor == "") {
t.Fatalf("issue asked for %q (lease held %v, module %s)", issuedFor, c.holds, c.module)
}
if c.wantGrants && c.wantErr == "" && !strings.Contains(said, "its grant written") {
t.Fatalf("said %q, want it to name the grant", said)
if c.wantGrants && c.wantErr == "" && !strings.Contains(said, "the grants written") {
t.Fatalf("said %q, want it to name the grants", said)
}
})
}
}
// The account a module speaks with on a node: the module's own when it reads `broker`, its provisioner client's when it
// runs one alone, none when its accounts are its processes' or it does not speak on the bus.
func TestAModulesOwnAccountOnANode(t *testing.T) {
// 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")
cases := []struct {
for _, c := range []struct {
name string
m catalogue.Manifest
user string
@@ -173,8 +252,7 @@ func TestAModulesOwnAccountOnANode(t *testing.T) {
{"the provisioner client alone", client,
broker.Principal{Kind: broker.KindProvisioner, Node: "ace", Module: "mesh-provisioner"}.Username(), true},
{"no broker, no client", catalogue.Manifest{Module: "hello-web"}, "", false},
}
for _, c := range cases {
} {
t.Run(c.name, func(t *testing.T) {
user, own := ownAccountOf(c.m, "ace")
if user != c.user || own != c.own {
@@ -182,70 +260,48 @@ func TestAModulesOwnAccountOnANode(t *testing.T) {
}
})
}
}
// The watch hands the assignments standing when it starts to the start's look (not fresh), says once that they are all
// read, and hands each assignment written afterwards as fresh; a rewrite of one that stands is handed nothing.
func TestTheWatchHandsTheStartsLookThenEachNewAssignment(t *testing.T) {
at := time.Date(2026, 10, 25, 16, 19, 42, 0, time.UTC)
bucket := chain.NewMemory()
put := func(a chain.Assignment) {
t.Helper()
value := anAssignmentValue(t, a)
if e, held, _ := bucket.Get(t.Context(), a.Key()); held {
if _, err := bucket.Update(t.Context(), a.Key(), value, e.Revision); err != nil {
t.Fatal(err)
}
return
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)
}
if _, err := bucket.Create(t.Context(), a.Key(), value); err != nil {
held, err := heldProcessAccount(broker.Broker{Fingerprint: "abc"}, "novox.example:4222", "novox", "mesh-mail",
user, "a-password")
if err != nil {
t.Fatal(err)
}
}
put(chain.Assignment{Node: "g14", Module: "mesh-build-go", Assigned: at})
type handed struct {
key string
fresh bool
}
got := make(chan handed, 8)
read := make(chan struct{}, 8)
ctx, stop := context.WithCancel(t.Context())
done := make(chan error, 1)
go func() {
done <- watchAssignmentsOnce(ctx, bucket.Watch, &assignmentTurns{},
func(a chain.Assignment, fresh bool) { got <- handed{a.Key(), fresh} },
func() { read <- struct{}{} })
}()
next := func() handed {
t.Helper()
select {
case h := <-got:
return h
case <-time.After(5 * time.Second):
t.Fatal("nothing handed")
return handed{}
var got map[string]any
if err := json.Unmarshal(held, &got); err != nil {
t.Fatal(err)
}
}
if h := next(); h != (handed{"g14.mesh-build-go", false}) {
t.Fatalf("the start's look handed %+v", h)
}
select {
case <-read:
case <-time.After(5 * time.Second):
t.Fatal("the current values were never said read")
}
replanned := chain.Assignment{Node: "g14", Module: "mesh-build-go", Assigned: at, Provisioning: json.RawMessage(`{"plan":2}`)}
put(replanned)
put(chain.Assignment{Node: "ace", Module: "mesh-build-node", Assigned: at})
if h := next(); h != (handed{"ace.mesh-build-node", true}) {
t.Fatalf("handed %+v, want ace.mesh-build-node as fresh (and nothing for the rewrite before it)", h)
}
stop()
if err := <-done; err != nil {
t.Fatal(err)
}
if len(read) != 0 {
t.Fatal("the current values were said read more than once")
}
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)
}
}
})
}
+31 -9
View File
@@ -965,6 +965,26 @@ func processPrincipalFor(m catalogue.Manifest, processID string) (broker.Process
return broker.ProcessPrincipal{}, false
}
// mintOnlyIfAbsent marks a context whose issue mints only accounts no row names yet (onlyIfAbsent).
type mintOnlyIfAbsent struct{}
// onlyIfAbsent is ctx for an issue that must never replace a minted account: the assign act's and the assignments'
// watch's, which may run at once for one account (novox/hq ADR 0543, the review of issue 1945).
func onlyIfAbsent(ctx context.Context) context.Context {
return context.WithValue(ctx, mintOnlyIfAbsent{}, true)
}
// mintBusUser mints u's password. By default it replaces whatever stands (a rotation: `module issue`); under
// onlyIfAbsent it mints only when no row names u, and won false says another caller minted it first and seals it,
// so this one seals nothing.
func mintBusUser(ctx context.Context, inv *inventory.Inventory, u inventory.BusUser) (string, bool, error) {
if only, _ := ctx.Value(mintOnlyIfAbsent{}).(bool); only {
return inv.MintBusPasswordIfAbsent(ctx, u)
}
password, err := inv.MintBusPassword(ctx, u)
return password, err == nil, err
}
// issueOnTheNewBus gives an assigned module its credential on the bus being built.
//
// **Three things differ from a management call, and each is the point of the move.** The credential
@@ -990,14 +1010,16 @@ func issueOnTheNewBus(ctx context.Context, inv *inventory.Inventory, m catalogue
}
if _, reads := m.OwnSecrets["broker"]; reads {
user := broker.Principal{Kind: broker.KindModule, Node: node, Module: m.Module}.Username()
password, err := inv.MintBusPassword(ctx, inventory.BusUser{
password, won, err := mintBusUser(ctx, inv, inventory.BusUser{
Username: user, Kind: busKindOf(m.Module), Node: node, Module: m.Module,
})
if err != nil {
return err
}
if err := issueWith(ctx, inv, m, node, busAddress, known, reachable, user, password); err != nil {
return err
if won {
if err := issueWith(ctx, inv, m, node, busAddress, known, reachable, user, password); err != nil {
return err
}
}
}
// Every process with a bus declaration gets an account of its own (novox/hq ADR 0436 §1, §2), minted and
@@ -1021,11 +1043,11 @@ func issueOnTheNewBus(ctx context.Context, inv *inventory.Inventory, m catalogue
func issueProcess(ctx context.Context, inv *inventory.Inventory, m catalogue.Manifest, pr broker.ProcessPrincipal,
node, busAddress string, known broker.Broker, reachable string) error {
user := broker.Principal{Kind: broker.KindProcess, Node: node, Module: m.Module, Process: pr.Process}.Username()
password, err := inv.MintBusPassword(ctx, inventory.BusUser{
password, won, err := mintBusUser(ctx, inv, inventory.BusUser{
Username: user, Kind: inventory.BusProcess, Node: node, Module: m.Module,
})
if err != nil {
return err
if err != nil || !won {
return err // lost: the caller that minted it seals it
}
name := processSecretName(pr.Process)
held, err := heldProcessAccount(known, reachable, node, m.Module, user, password)
@@ -1091,11 +1113,11 @@ func runsTheProvisionerClient(m catalogue.Manifest) bool {
func issueProvisionerClient(ctx context.Context, inv *inventory.Inventory, m catalogue.Manifest, node string,
known broker.Broker, reachable string) error {
user := broker.Principal{Kind: broker.KindProvisioner, Node: node, Module: m.Module}.Username()
password, err := inv.MintBusPassword(ctx, inventory.BusUser{
password, won, err := mintBusUser(ctx, inv, inventory.BusUser{
Username: user, Kind: inventory.BusProvisioner, Node: node, Module: m.Module,
})
if err != nil {
return err
if err != nil || !won {
return err // lost: the caller that minted it seals it
}
held, err := json.Marshal(struct {
URL string `json:"url"`
+2
View File
@@ -72,6 +72,8 @@ func issueMissingProcessAccounts(ctx context.Context, inv *inventory.Inventory,
if len(m.ProcessBuses()) == 0 || mayIssue(m) != nil {
return "", nil
}
// Only the missing: never a second password over one another caller minted between this look and the mint.
ctx = onlyIfAbsent(ctx)
minted, err := inv.BusMinted(ctx)
if err != nil {
return "", err
-58
View File
@@ -1,12 +1,9 @@
package main
import (
"encoding/json"
"strings"
"testing"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"github.com/novox/mesh-controller/internal/broker"
)
// mailSenderWithProcess is mesh-mail's real manifest shape (novox/hq ADR 0436 §7, §8, the lead's fixture):
@@ -25,61 +22,6 @@ func mailSenderWithProcess() catalogue.Manifest {
return m
}
// review round 1, finding 1 (1181): a module declaring no `broker` of its own at all — mesh-mail's real
// shape, tools served by the tool runner — still passes mayIssue on its process's bus declaration alone.
func TestAModuleWithNoBrokerAndOnlyAProcessMayBeIssued(t *testing.T) {
if err := mayIssue(mailSenderWithProcess()); err != nil {
t.Fatalf("a module with no broker of its own, whose process declares bus, was refused: %v", err)
}
}
// 1181/AC1: a process with a bus declaration and no own secret broker-<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) {
+31
View File
@@ -114,6 +114,37 @@ func (i *Inventory) writeBusUser(ctx context.Context, u BusUser, hash string) er
return nil
}
// MintBusPasswordIfAbsent mints u's password only when no row names u yet, in one statement: of two callers that both
// found the account missing, exactly one wins, and the other is told it lost (won false, no password) and must seal
// nothing, so the hash the bus holds is always the password the node is sealed (novox/hq ADR 0543, the review of
// issue 1945: the assign act and the assignments' watch once each minted, the later hash beside the earlier seal).
func (i *Inventory) MintBusPasswordIfAbsent(ctx context.Context, u BusUser) (string, bool, error) {
if u.Username == "" || u.Kind == "" {
return "", false, errors.New("a bus user needs a username and a kind")
}
raw := make([]byte, 32)
if _, err := rand.Read(raw); err != nil {
return "", false, fmt.Errorf("cannot generate a bus password: %w", err)
}
password := base64.RawURLEncoding.EncodeToString(raw)
hash, err := hashBusPassword(password)
if err != nil {
return "", false, fmt.Errorf("cannot hash a bus password: %w", err)
}
tag, err := i.store.Pool().Exec(ctx,
`insert into bus_user (username, kind, node, module, password_hash)
values ($1, $2, $3, $4, $5)
on conflict (username) do nothing`,
u.Username, u.Kind, u.Node, u.Module, string(hash))
if err != nil {
return "", false, fmt.Errorf("cannot record the bus user %s: %w", u.Username, err)
}
if tag.RowsAffected() == 0 {
return "", false, nil
}
return password, true, nil
}
func (i *Inventory) MintBusPassword(ctx context.Context, u BusUser) (string, error) {
if u.Username == "" || u.Kind == "" {
return "", errors.New("a bus user needs a username and a kind")
@@ -33,7 +33,7 @@ func TestABusPasswordIsReturnedOnceAndOnlyItsHashIsKept(t *testing.T) {
}
}
func TestMintingAgainRotatesRatherThanAddsAUser(t *testing.T) {
func TestMintingAgainRotatesAndMintingIfAbsentHasOneWinner(t *testing.T) {
inv := ForTest(t)
ctx := context.Background()
@@ -62,6 +62,50 @@ func TestMintingAgainRotatesRatherThanAddsAUser(t *testing.T) {
if bcrypt.CompareHashAndPassword([]byte(hash), []byte(first)) == nil {
t.Fatal("the previous password still verifies, so a rotation revoked nothing")
}
// **Minting only if absent never rotates, and of several at once exactly one wins** (ADR 0543, the review of issue
// 1945): the assign act and the assignments' watch may both find an account missing; the winner's password is the
// one the store's hash verifies, so the one it seals is the one the bus accepts, and a loser seals nothing.
if _, won, err := inv.MintBusPasswordIfAbsent(ctx, aBusUser("shop")); err != nil || won {
t.Fatalf("a mint-if-absent over a minted account won %v (%v): it would replace a sealed password", won, err)
}
const racers = 8
type minted struct {
password string
won bool
err error
}
results := make(chan minted, racers)
for range racers {
go func() {
p, won, err := inv.MintBusPasswordIfAbsent(ctx, aBusUser("racer"))
results <- minted{p, won, err}
}()
}
var winner string
wins := 0
for range racers {
r := <-results
if r.err != nil {
t.Fatal(r.err)
}
if r.won {
wins++
winner = r.password
} else if r.password != "" {
t.Fatal("a caller that lost was handed a password to seal")
}
}
if wins != 1 {
t.Fatalf("%d of %d concurrent mints won, want exactly one", wins, racers)
}
stored, _, err := inv.BusUserHash(ctx, "one.racer")
if err != nil {
t.Fatal(err)
}
if err := bcrypt.CompareHashAndPassword([]byte(stored), []byte(winner)); err != nil {
t.Fatal("the stored hash is not the winner's password: the node would be sealed a password the bus refuses")
}
}
func TestForgettingANodeTakesItsBusUsersWithIt(t *testing.T) {