Merge pull request '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) from fix/1945-take-in-consumer into main

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