Compare commits

...
Author SHA1 Message Date
jschoubben 3e39ccf958 Let only a watch that read every standing assignment be the start's look (issue 1945, ADR 0543, review)
The flag that made the first watch the start's look was cleared after every
watch, even one that failed to open or ended before its Caught mark: the next
watch then handed every standing assignment as fresh, about 300 settles at once
with a catalogue read each, or dropped the keys the cut watch had handed. The
flag is now cleared by the start's look itself, and a watch cut before it opens
the next with nothing heard. Two rows of the existing table test cover it.
Research 072: the grants piece (memberships and credentials heard from placement).
2026-10-25 18:07:21 +01:00
jschoubben 9ff47f6aaa Mint an assignment's account once, whoever finds it missing first (issue 1945, ADR 0543, review)
The assign act and the assignments' watch both checked whether an account was
minted and then minted it, and minting overwrites the hash: interleaved, the
bus kept one caller's hash and the node was sealed the other's password, the
crash loop this branch fixes. Both now mint through one statement that inserts
the row only when none stands; the caller whose insert wins seals, the other
seals nothing. issueMissingProcessAccounts mints the same way.

A watch opened again after one ended handed the standing assignments as the
start's look, so one that came to stand in the gap got no bus composition or
grant when its account already stood: only the process's first watch is the
start's look now. The start's look reads the catalogue once and ends with one
composition of the bus's node and the grants; a controller not holding the
lease is told so and tries again rather than dropping the key.

Tests folded to stay within the budget: the assignment tests are three table
tests, the process account and mayIssue tests folded into one of them, and
the concurrent mint is a case of the inventory's rotation test.
Research 072: the grants piece (memberships and credentials heard from placement).
2026-10-25 17:51:06 +01:00
jschoubben ca5f74eadb Issue a node's bus accounts and grant when its assignment comes to stand (issue 1945, ADR 0543)
ADR 0539 clause 7 issues a module's credential and grant only when its build is done.
A module built long before and assigned to one more node never has a build done again:
on 2026-10-25 at 17:19 mesh-build-node and mesh-build-go were assigned to ace and composed
for it, the bus's user list never named ace's accounts, and both runtimes restarted every
5 s (one with no credential, one refused by the bus). The assign act composes the assigned
node alone, never the node holding the bus, and writes no grant.

So a write of the assignments bucket is the event: when a key comes to stand (new, or
ended or pending before), the lease holder issues the module's accounts on that node that
are not minted (never re-issued), composes the node holding the bus and writes the grants,
through the composer the done keys' take-ins use. A rewrite of a standing assignment does
nothing; the start's look settles every standing one, one after another. A step of
research 072's grants (memberships and credentials heard from placement).
2026-10-25 17:37:03 +01:00
jschoubben c3db00ceac Merge pull request 'Compose the packages mirror only on a change, end every serve goroutine before serve returns, and raise no alarm for a cancellation (issue 1945)' (#464) from fix/r6-mirror-on-change-2 into main
Reviewed-on: #464
2026-10-25 16:20:55 +00:00
jschoubben 8e02c4726e Clear a module's fingerprint-not-composed on every successful take-in, and stop serve's goroutines before an error return undoes anything (issue 1945)
mesh/check pass 8e02c4726eb4 on c3db00ceac25 | unit 981 run, 171 kept, 2 skipped | integration 1641 run, 232 kept, 7 skipped | over bound: integration…
mesh/delivery held for a person: merged without a passing check: only a person decides what follows
A success cleared the condition only when the same follower had raised it, so
one raised before a restart or by another follower stayed open after the take-in
worked, and the operator read that a build needed a look. Clearing a condition
that is not open costs one read. On an error return, serve now ends its
goroutines before the keeper closes and before handActConn and the bus are put
back (the review of fix 12).
2026-10-25 17:09:35 +01:00
jschoubben cc5c9ae888 Count the take-ins and every other background work in the serving group, so serve's stop waits for them too (issue 1945)
inBackground ran the take-ins of done keys (now in #463's slots), the start's
look, the given values' replacement and the announced reads beside serve without
serve waiting for them: a take-in could outlive its controller as the loops did.
It now takes the context and is counted in its serving group as well, and a
serve that is stopping starts none.
2026-10-25 16:36:49 +01:00
jschoubben c7e31640cf Raise no condition for a take-in, a layout or a put cut by a cancellation (issue 1945)
Thirty urgent fingerprint-not-composed conditions were live on 2026-10-25, each a
done key's take-in a stop or a restart cut with "context canceled". A
cancellation is no fault of the build, the chain or the bucket: a take-in or a
layout whose controller stops returns and the next one does it; one whose own
step was cancelled is tried again without a condition; a send cut by a
cancellation counts as no failure to put its files.

The put test now starts and leaves the node's failure count at nothing, so a
second run of it (-count=2) finds no count left by the first.
2026-10-25 16:30:58 +01:00
jschoubben c75d726205 End every goroutine serve starts before serve returns, so no loop outlives its controller (issue 1945)
The try of the on-change mirror failed with every test passed: goroutines a
served test started outlived its serve and met the next test's globals and its
closed stores. Under -race the served tests show it as a race on conditionsFrom
(watchTheMesh writing it while settlingPending, already running, read it), and
a take-in cut by the stop raised fingerprint-not-composed, a kind with no plain
words, which fails the package after its PASS line.

- serve counts every goroutine it and its parts start (serving_group.go) and
  ends and waits for them before anything else is undone: the lease's watch,
  the pending settler, the status summary, the facts, the mirrors and the new
  declared-packages keeper with its two bucket watches, the declaration sweep,
  the read-call counter, the watches' closers, the have watches, the vault move
  on a report, and the watchdogs, doctor, healers and asker under their own
  group.
- The condition store opens before anything that reads conditionsFrom starts;
  askerFrom is an atomic pointer, read from the keeper's goroutines.
- A done key's take-in cut by the stop raises nothing, and fingerprint-not-composed
  has plain words.
2026-10-25 16:30:58 +01:00
jschoubben a4cd5ba375 Compose the declared-packages mirror only when what it composes from changed (issue 1945)
The mirror composed every node in full on the record mirrors' 10 s tick; a
pass takes about 45 s on the control node, so the controller never rested
(130-175 % CPU, issue 1945 note 22). It now passes on a watch of the
compositions' digests and of the assignments, one pass at a time with the
wakes of a pass coalesced into one after it (at least 30 s apart), skips the
composition when the stores' Input digest is unchanged, and keeps a safety
pass every 10 minutes that composes nothing when nothing changed.
2026-10-25 16:30:58 +01:00
jschoubben 62802cd966 Merge pull request 'Share the controller's stores and take in at most 16 done keys at once, so a burst of builds cannot fill the store (issue 1277, issue 1945)' (#463) from fix/store-pool-flood into main
Reviewed-on: #463
2026-10-25 15:29:58 +00:00
jschoubben e9a885cda9 Share the controller's stores with its take-ins and cap take-ins at 16
mesh/check checking e9a885cda925 on 62802cd96602
mesh/delivery held for a person: merged without a passing check: only a person decides what follows
At 16:07:36 the builder rewrote about 120 done keys at once. Each woke a
take-in of its own with no limit, and each take-in read the rest of the
mesh through restOfTheMeshOf, which opened a licences and an identity
pool per call. novox's store ran out of connections at 16:08 and refused
everyone, the controller's retries keeping it full.

- The serving controller lends its stores by inventory (shareStores);
  restOfTheMeshOf reads on them instead of opening pools per call, and a
  verb run as its own process still opens its own. stores opens each
  store once under a lock, since readers now share it. serve uses the
  shared identity store instead of a second pool.
- The watch's take-ins and the start's look share one cap of 16 take-ins
  at once (takeInSlots); a take-in waiting for its next try holds none.
- store.Ready backs off on a full store (SQLSTATE 53300), doubling its
  wait up to 5 s, and says once that the store is full.

Issues 1945 and 1277.
2026-10-25 16:23:55 +01:00
mesh-tools 2814262ac0 Merge pull request 'Keep the declarations retired once their stream is gone, so the bus is not called unhealthy for a stream release 6 deleted (issue 1945)' (#462) from fix/r6-nodes-stream-gone into main 2026-10-25 14:25:11 +00:00
jschoubben 0a536d7b6d Keep the declarations retired once their stream is gone, so nothing expects, asserts or makes NODES again (issue 1945)
mesh/check pass 0a536d7b6d79 on 2bcd85d2df4c | unit 966 run, 172 kept, 2 skipped | integration 1622 run, 236 kept, 7 skipped | over bound: integration…
mesh/delivery merged: source.main moved to novox/mesh-controller@2814262ac0fa
A node-engine newer than the registered one, or a node asleep on an older
have, made the done test false after the start deleted NODES; the bus's
health check, probes D6 and D7 and healer H3 then expected the stream and
every node's consumer again, raised them urgent and made them anew. Release
6 sends no declaration, so a stream made again helps no node: a gone stream
is retired. The advisory watchdog also reads a consumer whose stream is gone
as gone instead of failing.
2026-10-25 15:01:09 +01:00
mesh-tools 2bcd85d2df Merge pull request 'Never put a module whose want names its own composition into the node's own, and read a re-registered module's fingerprint from the indexer (issue 1945)' (#461) from fix/r6-own-composition-dupes into main 2026-10-25 13:37:34 +00:00
jschoubben ac34bce49f Take a module's fingerprint from the indexer when its commit has no build, and never compose a module its want names into the node's own (issue 1945)
mesh/check pass ac34bce49f58 on 2bcd85d2df4c | unit 966 run, 171 kept, 2 skipped | integration 1625 run, 232 kept, 7 skipped | over bound: integration…
mesh/delivery merged: source.main moved to novox/mesh-controller@2bcd85d2df4c
A catalogue merge that leaves a module's sources alone registers it at the new
commit from the same done key, whose build record stands at its first commit,
so the commit had no build and the module no fingerprint of the new form: it
went into the node's own composition while its want named its own, and every
node declared its files twice. currentFingerprints now reads the indexer's
trunk value at the commit when no build points at it (1945/AC15), and a module
a want or have names by its own composition is left out of the node's own
(1945/AC16).
2026-10-25 14:10:51 +01:00
mesh-tools af2b0844a1 Merge pull request 'Compose each machine once per pass, so take-ins of done builds stop colliding and every new build gets its composition and credential (issue 1945)' (#460) from fix/r6-compose-once into main 2026-10-25 12:30:32 +00:00
jschoubben 3b4cb69e84 Take again a replacement of given values asked while one ran on the node (issue 1945)
mesh/check pass 3b4cb69e84dd on af2b0844a1fc | unit 963 run, 171 kept, 2 skipped | integration 1622 run, 232 kept, 7 skipped | over bound: integration…
mesh/delivery merged: source.main moved to novox/mesh-controller@af2b0844a1fc
2026-10-25 13:02:49 +01:00
jschoubben c47a590463 The start pass reads a module's look-again mark after letting go, and hands it on to takeDoneKey (issue 1945) 2026-10-25 13:00:43 +01:00
jschoubben f35aa35541 Look at a module's mark again after letting go of it, so a wake just before is taken (issue 1945) 2026-10-25 12:55:33 +01:00
jschoubben 8114cee2af Give each caller of a pass its own nodes' errors and write the grants whenever asked; a wake during a take-in makes it look once more (issue 1945)
mesh/delivery closed: a newer head of the same pull request
mesh/check pass 8114cee2af47 on 0fff3efd4e0b | unit 959 run, 172 kept, 2 skipped | integration 1615 run, 236 kept, 7 skipped | over bound: integration…
2026-10-25 12:44:28 +01:00
jschoubben baa8f58473 Compose each node once per pass in one composer for the done keys' take-ins, one composition of a node at a time, and the start's look side by side (issue 1945) 2026-10-25 12:39:59 +01:00
jschoubben 0470869fbb Keep the digest of the composition the bucket holds, following another writer's composition written in between (issue 1945) 2026-10-25 12:36:00 +01:00
jschoubben 8a3c0cd022 Say each wait of a done key's take-in once with its reason and look again every 30 s, one follower per module (issue 1945) 2026-10-25 12:35:42 +01:00
52 changed files with 3242 additions and 291 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
// verb and the push that would refuse the module until it is run — never a silent placeholder (novox/hq issue 203).
func issueOnAssign(ctx context.Context, open *stores, node, module string) string {
// Mint only what no row names: the assignments' watch hears the same write and may issue at once (ADR 0543).
ctx = onlyIfAbsent(ctx)
inv := open.inventory
shelf, err := inv.Catalogue(ctx)
if err != nil {
+4 -3
View File
@@ -12,6 +12,7 @@ import (
"sort"
"strings"
"sync"
"sync/atomic"
"time"
"github.com/nats-io/nats.go"
@@ -27,7 +28,7 @@ import (
)
// askerFrom is the serving controller's asker; nil in any other process.
var askerFrom *asker
var askerFrom atomic.Pointer[asker]
// askWithin is how long a verb a warrant chose is given to answer.
const askWithin = time.Minute
@@ -573,8 +574,8 @@ func startAsking(ctx context.Context, open *stores, server *link.Server, conn *n
fmt.Printf("the operator's answers cannot be heard, so nothing is asked: %v\n", err)
return
}
askerFrom = a
go a.keep(ctx)
askerFrom.Store(a)
goServing(ctx, func() { a.keep(ctx) })
}
// servingAsker is the serving controller's asker, wired to the mesh: what it reads, publishes, performs and
+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)
}
}
})
}
+6 -2
View File
@@ -49,7 +49,7 @@ func attachTheChain(ctx context.Context, inv *inventory.Inventory, server *link.
// keys laid out before skips them, so this takes seconds; one that fails is tried again beside the serving, loudly.
if err := inv.LayBuildsOut(ctx, func(format string, args ...any) { fmt.Printf("chain: "+format+"\n", args...) }); err != nil {
fmt.Printf("chain: the builds are not laid out under their modules yet, and none is read until they are: %v\n", err)
go layBuildsOutSoon(ctx, inv)
goServing(ctx, func() { layBuildsOutSoon(ctx, inv) })
}
// The requests the move of delivery 1 found never handed over are not asked again: the controller asks no build
// (novox/hq ADR 0501; issue 1851, A10), and the builder builds every fingerprint the indexer writes.
@@ -80,7 +80,11 @@ func layBuildsOutSoon(ctx context.Context, inv *inventory.Inventory) {
Source: "chain", Summary: "the builds are not laid out under their modules, so none is read: "}
if err != nil {
fmt.Printf("chain: the builds are not laid out under their modules, and none is read until they are: %v\n", err)
if keeper := conditionsFrom; keeper != nil && !failed {
if ctx.Err() != nil {
return // stopped: the next controller lays them out
}
// A cancelled layout is tried again and raises nothing: a cancellation is no fault of the chain (issue 1945).
if keeper := conditionsFrom; keeper != nil && !failed && !errors.Is(err, context.Canceled) {
o.Summary += err.Error()
if _, oerr := keeper.Observe(ctx, o); oerr == nil {
failed = true
+10 -3
View File
@@ -326,9 +326,13 @@ func restOfTheMesh(ctx context.Context, open *stores, shelf map[string]catalogue
return world, nil
}
// restOfTheMeshOf is restOfTheMesh for a caller that holds the inventory alone: the other stores the Input reads are opened for
// the call and let go of again.
// restOfTheMeshOf is restOfTheMesh for a caller that holds the inventory alone: on the serving controller's stores where it
// lends them (sharedStores), never a pool per call; elsewhere (a verb run as its own process) the other stores the Input
// reads are opened for the call and let go of again.
func restOfTheMeshOf(ctx context.Context, inv *inventory.Inventory, shelf map[string]catalogue.Manifest, exclude string) (catalogue.World, error) {
if shared, ok := sharedStoresOf(inv); ok {
return restOfTheMeshOn(ctx, shared, shelf, exclude)
}
open := &stores{inventory: inv}
defer func() {
if open.licences != nil {
@@ -338,9 +342,12 @@ func restOfTheMeshOf(ctx context.Context, inv *inventory.Inventory, shelf map[st
open.identity.Close()
}
}()
return restOfTheMesh(ctx, open, shelf, exclude)
return restOfTheMeshOn(ctx, open, shelf, exclude)
}
// restOfTheMeshOn is restOfTheMesh; a variable so a test reads which stores it is handed.
var restOfTheMeshOn = restOfTheMesh
// sayFound says the lines a composition found while composing, once each: a machine left out of the rest of the mesh, a seat
// that two claimants could take, the bus's users left out for want of a credential.
func sayFound(ctx context.Context, d compose.Declared) {
@@ -0,0 +1,78 @@
package main
import (
"context"
"encoding/json"
"testing"
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/nats-io/nats.go/jetstream"
)
// raceKV is a digests bucket in which another writer writes its composition and digest just before this writer reads
// the digest key: what two take-ins composing one node at once do.
type raceKV struct {
*fakeComposedKV
before func()
}
func (r *raceKV) Get(ctx context.Context, key string) (jetstream.KeyValueEntry, error) {
if r.before != nil {
b := r.before
r.before = nil
b()
}
return r.fakeComposedKV.Get(ctx, key)
}
// The digest kept beside a composition is always the digest of the composition the bucket holds: when another writer
// writes its composition and its digest between this writer's composition and its digest, this writer's digest must not
// win, or the assigner's want names a composition the bucket no longer holds and the node applies nothing (issue 1945,
// live on 2026-10-25: shanks' node-tools composition had digest 575a5c96, its want named a36e63c8).
func TestTheDigestKeptIsTheDigestOfTheCompositionHeld(t *testing.T) {
at, _, _, _ := aPlace(t, "shanks")
key, _ := sdkdelivery.ComposedKey("shanks", "node-tools", fp2Of('a'))
file := func(content string) json.RawMessage {
return json.RawMessage(`{"id":"node-tools.f","type":"file","module":"node-tools","path":"/etc/nt","mode":"0644","content":"` +
content + `"}`)
}
ours := sdkdelivery.Composed{Node: "shanks", Module: "node-tools", FP: fp2Of('a'), Resources: []json.RawMessage{file("ours")}}
theirs := sdkdelivery.Composed{Node: "shanks", Module: "node-tools", FP: fp2Of('a'), Resources: []json.RawMessage{file("theirs")}}
race := &raceKV{fakeComposedKV: at.digests.(*fakeComposedKV)}
at.digests = race
race.before = func() {
// The other writer: its composition over ours, then its digest.
other := at
other.digests = race.fakeComposedKV
if _, err := writeComposed(context.Background(), other, key, theirs); err != nil {
t.Fatal(err)
}
}
if _, err := writeComposed(context.Background(), at, key, ours); err != nil {
t.Fatal(err)
}
held, _, err := readComposed(context.Background(), at.bucket, key)
if err != nil {
t.Fatal(err)
}
heldDigest, _ := sdkdelivery.ComposedDigestOf(held)
e, err := race.fakeComposedKV.Get(context.Background(), key)
if err != nil {
t.Fatal(err)
}
kept, err := sdkdelivery.DecodeComposedDigest(e.Value())
if err != nil {
t.Fatal(err)
}
if kept.Digest != heldDigest {
t.Fatalf("the digest kept is %s and the composition held has %s", kept.Digest, heldDigest)
}
}
func fp2Of(c byte) string {
b := make([]byte, 64)
for i := range b {
b[i] = c
}
return sdkdelivery.SourcePrefix2 + string(b)
}
@@ -140,4 +140,35 @@ func TestTheDeclarationsStreamGoesOnceEveryNodeRunsTheNewEngine(t *testing.T) {
if !declarationsRetiredNow(ctx, m.open.inventory) {
t.Errorf("a raise after the retirement would assert the stream again")
}
// Issue 1945: a node-engine other than the registered one landing on laptop (a newer build on its way) leaves the
// declarations retired, so what the health check, the probes and healer H3 expect of the bus names neither the
// stream nor a node's consumer on it, and asserting the bus's objects again does not make the stream anew.
have("laptop", fpOf('7'))
if !declarationsRetiredNow(ctx, m.open.inventory) {
t.Fatalf("with laptop on another node-engine after the stream went, the declarations are no longer retired")
}
streams, consumers, err := expectedBusObjects(ctx, m.open.inventory)
if err != nil {
t.Fatal(err)
}
for _, s := range streams {
if s.Name == broker.DeclarationsStream {
t.Errorf("the bus's health check and probe D7 still expect the stream %s", s.Name)
}
}
for _, c := range consumers {
if c.Stream == broker.DeclarationsStream {
t.Errorf("the bus's health check and probe D6 still expect %s on %s", c.Name, c.Stream)
}
}
bus, err := broker.Dial(m.conn.ConnectedUrl())
if err != nil {
t.Fatal(err)
}
t.Cleanup(bus.Close)
_ = assertOnSend(ctx, m.open.inventory, bus, " ")
if _, err := m.js.Stream(ctx, broker.DeclarationsStream); !errors.Is(err, jetstream.ErrStreamNotFound) {
t.Errorf("healer H3's assertion of the bus's objects made the stream anew: %v", err)
}
}
+157 -31
View File
@@ -114,29 +114,94 @@ func inList(list []string, s string) bool {
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: its next write wakes this again.
// 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{})
mu.(*sync.Mutex).Lock()
defer mu.(*sync.Mutex).Unlock()
failed := false
for {
err := takeDoneKeyOnce(ctx, open, module)
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 || errors.Is(err, errKeyWaits) {
if keeper := conditionsFrom; keeper != nil && failed {
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
}
fmt.Printf("done keys: %s: %v; tried again in %s\n", module, err, doneKeyRetry)
if keeper := conditionsFrom; keeper != nil && !failed {
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):
}
continue
}
if ctx.Err() != nil {
return // stopped: a take-in cut by the stop is no fault of the build, and the next controller takes it in
}
if keeper := conditionsFrom; keeper != nil && !failed && !errors.Is(err, context.Canceled) {
// A take-in whose own step was cancelled is tried again, and raises nothing: a cancellation is no fault
// of the build (issue 1945: 30 urgent conditions live, each a take-in cut by a restart).
o.Summary += err.Error()
if _, oerr := keeper.Observe(ctx, o); oerr == nil {
failed = true
@@ -224,21 +289,13 @@ func liveDoneKeySteps(open *stores, keys registerKeys) doneKeySteps {
}
return holders[catalogue.BrokerSeat].Node, nil
},
// The composer's, never the take-in's own: one pass at a time composes each node asked for once, and writes the
// grants once (issue 1945, composer.go).
compose: func(ctx context.Context, nodes []string) error {
_, err := composeModules(ctx, open, nodes)
return err
return composerFor(open).compose(ctx, nodes, false)
},
grants: func(ctx context.Context) error {
bus, err := aBus()
if err != nil {
return err
}
defer bus.Close()
js, err := jetstream.New(bus.Conn())
if err != nil {
return err
}
return grantsIssued(ctx, inv, js, "done keys: ")
return composerFor(open).compose(ctx, nil, true)
},
}
}
@@ -252,18 +309,18 @@ func watchDoneKeys(ctx context.Context, open *stores) {
return
}
if err := builderkeys.Watch(ctx, keys.JS, func(module string) {
inBackground(func() { takeDoneKey(ctx, open, module) })
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
}
go func() {
goServing(ctx, func() {
<-ctx.Done()
closeKeys()
}()
})
// And every module once now, after the watch is open so no write falls between the two.
inBackground(func() {
inBackground(ctx, func() {
doneKeysAtStart(ctx, catalogedModules(open), func(module string) { takeDoneKeyAtStart(ctx, open, module) })
})
}
@@ -280,23 +337,61 @@ func doneKeysAtStart(ctx context.Context, modules func(context.Context) ([]strin
return
}
sort.Strings(names)
// Side by side, a few at a time: their compositions meet in the composer's passes, each node composed once per pass
// (issue 1945), so the start's look is not one module's composition after another.
sem := make(chan struct{}, startTakeIns)
var wg sync.WaitGroup
for _, m := range names {
if ctx.Err() != nil {
return
break
}
take(m)
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{})
mu.(*sync.Mutex).Lock()
err := takeDoneKeyOnce(ctx, open, module)
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()
if err != nil && !errors.Is(err, errKeyWaits) {
inBackground(func() { takeDoneKey(ctx, open, module) })
// 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) })
}
}
@@ -314,3 +409,34 @@ func catalogedModules(open *stores) func(context.Context) ([]string, error) {
return out, nil
}
}
// doneKeysComposers are the composers of the done keys' take-ins in this process, one for each set of stores (a test
// runs several meshes), each started at its first use and run for as long as the process serves.
var doneKeysComposers sync.Map
// composerFor is the composer of the stores, its passes composing on them and writing the grants on the bus.
func composerFor(open *stores) *composer {
if c, ok := doneKeysComposers.Load(open); ok {
return c.(*composer)
}
c := newComposer(func(ctx context.Context, nodes []string) map[string]error {
_, failed := composeModulesByNode(ctx, open, nodes)
return failed
}, func(ctx context.Context) error {
bus, err := aBus()
if err != nil {
return err
}
defer bus.Close()
js, err := jetstream.New(bus.Conn())
if err != nil {
return err
}
return grantsIssued(ctx, open.inventory, js, "done keys: ")
})
if was, loaded := doneKeysComposers.LoadOrStore(open, c); loaded {
return was.(*composer)
}
go c.run(context.Background())
return c
}
@@ -0,0 +1,83 @@
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)
}
}
@@ -0,0 +1,58 @@
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)
}
}
@@ -5,6 +5,7 @@ import (
"errors"
"reflect"
"sort"
"sync"
"testing"
)
@@ -15,9 +16,10 @@ import (
// 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) { took = append(took, module) })
}, 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)
+1 -1
View File
@@ -99,7 +99,7 @@ func TestAChangedUserListIsANewRevisionOfTheBusModulesCompositionAlone(t *testin
d["resources"] = append(d["resources"].([]any), map[string]any{"id": "nats.bus-users", "type": "file",
"module": "nats", "path": "/var/lib/mesh/nats/users.conf", "content": users, "mode": "0600"})
raw, _ := json.Marshal(d)
comps, _, err := compositionsOf("g14", raw, fps, nil, nil)
comps, _, _, err := compositionsOf("g14", raw, fps, nil, nil, nil)
if err != nil {
t.Fatal(err)
}
@@ -0,0 +1,184 @@
package main
import (
"context"
"fmt"
"strings"
"sync"
"testing"
"time"
)
// A take-in that waits is said once for each reason it waits for, and tried again every doneKeyRetry until it is taken
// in: never left for a key write that may not come (issue 1945, live on 2026-10-25: mesh-host's take-in at 26bd waited,
// said nothing, and nothing woke it until another write).
func TestAWaitingTakeInIsSaidOnceAndTriedAgain(t *testing.T) {
was, wasRetry, wasSay := doneKeyTakeOnce, doneKeyRetry, doneKeysSay
t.Cleanup(func() { doneKeyTakeOnce, doneKeyRetry, doneKeysSay = was, wasRetry, wasSay })
doneKeyRetry = 5 * time.Millisecond
var mu sync.Mutex
var said []string
doneKeysSay = func(s string) { mu.Lock(); said = append(said, s); mu.Unlock() }
calls := 0
doneKeyTakeOnce = func(context.Context, *stores, string) error {
calls++
switch {
case calls <= 2:
return fmt.Errorf("%w: the artifact store could not hold mesh-host's artifact", errKeyWaits)
case calls == 3:
return fmt.Errorf("%w: mesh-host's key cannot be read", errKeyWaits)
}
return nil
}
done := make(chan struct{})
go func() { takeDoneKey(context.Background(), nil, "mesh-host"); close(done) }()
select {
case <-done:
case <-time.After(5 * time.Second):
t.Fatal("a waiting take-in was not tried again until it was taken in")
}
if calls != 4 {
t.Fatalf("tried %d times, want 4", calls)
}
mu.Lock()
defer mu.Unlock()
waits := 0
for _, s := range said {
if strings.Contains(s, "mesh-host waits") {
waits++
}
}
if waits != 2 {
t.Fatalf("said %q: want each of the two reasons once", said)
}
}
// A wake while a take-in of the module already follows it adds no second one: the follower looks again itself.
func TestAWakeWhileATakeInFollowsAddsNone(t *testing.T) {
was, wasRetry, wasSay := doneKeyTakeOnce, doneKeyRetry, doneKeysSay
t.Cleanup(func() { doneKeyTakeOnce, doneKeyRetry, doneKeysSay = was, wasRetry, wasSay })
doneKeyRetry = 20 * time.Millisecond
doneKeysSay = func(string) {}
var mu sync.Mutex
calls := 0
release := make(chan struct{})
doneKeyTakeOnce = func(context.Context, *stores, string) error {
mu.Lock()
calls++
mu.Unlock()
select {
case <-release:
return nil
default:
return fmt.Errorf("%w: the key says building", errKeyWaits)
}
}
follower := make(chan struct{})
go func() { takeDoneKey(context.Background(), nil, "board"); close(follower) }()
time.Sleep(5 * time.Millisecond)
returned := make(chan struct{})
go func() { takeDoneKey(context.Background(), nil, "board"); close(returned) }()
select {
case <-returned:
case <-time.After(time.Second):
t.Fatal("a second wake waited for the follower")
}
close(release)
<-follower
}
// A wake while a take-in of the module follows it is not lost: the follower goes round once more after it succeeds, so
// a newer build's key written meanwhile is taken in (the review of issue 1945).
func TestAWakeDuringATakeInIsTakenAfterIt(t *testing.T) {
was, wasRetry, wasSay := doneKeyTakeOnce, doneKeyRetry, doneKeysSay
t.Cleanup(func() { doneKeyTakeOnce, doneKeyRetry, doneKeysSay = was, wasRetry, wasSay })
doneKeyRetry = 20 * time.Millisecond
doneKeysSay = func(string) {}
var mu sync.Mutex
calls := 0
inFirst := make(chan struct{})
release := make(chan struct{})
doneKeyTakeOnce = func(context.Context, *stores, string) error {
mu.Lock()
calls++
n := calls
mu.Unlock()
if n == 1 {
close(inFirst)
<-release
}
return nil
}
follower := make(chan struct{})
go func() { takeDoneKey(context.Background(), nil, "mesh-host"); close(follower) }()
<-inFirst
takeDoneKey(context.Background(), nil, "mesh-host") // the newer key's wake, while the first runs
close(release)
<-follower
mu.Lock()
defer mu.Unlock()
if calls != 2 {
t.Fatalf("taken %d times: the wake during the take-in was lost", calls)
}
}
// A wake that comes after the follower's last look but before it lets go of the module is not lost: the follower looks
// at the mark again once it has let go, and takes the module once more (the review of issue 1945).
func TestAWakeJustBeforeTheFollowerLetsGoIsTaken(t *testing.T) {
was, wasRetry, wasSay, wasHook := doneKeyTakeOnce, doneKeyRetry, doneKeysSay, doneKeysBeforeLetGo
t.Cleanup(func() {
doneKeyTakeOnce, doneKeyRetry, doneKeysSay, doneKeysBeforeLetGo = was, wasRetry, wasSay, wasHook
})
doneKeyRetry = 10 * time.Millisecond
doneKeysSay = func(string) {}
var mu sync.Mutex
calls := 0
doneKeyTakeOnce = func(context.Context, *stores, string) error {
mu.Lock()
calls++
mu.Unlock()
return nil
}
woke := false
doneKeysBeforeLetGo = func(module string) {
if !woke {
woke = true
takeDoneKey(context.Background(), nil, module) // fails its TryLock and marks the module
}
}
takeDoneKey(context.Background(), nil, "mesh-host")
mu.Lock()
defer mu.Unlock()
if calls != 2 {
t.Fatalf("taken %d times: the wake just before the follower let go was lost", calls)
}
}
// A wake during the start's take-in of a module is not lost: the start pass reads the mark once it lets go of the module,
// and hands the module to takeDoneKey, which takes it once more (the review of issue 1945: the third dropped wake of its
// kind).
func TestAWakeDuringTheStartsTakeInIsTaken(t *testing.T) {
was, wasRetry, wasSay := doneKeyTakeOnce, doneKeyRetry, doneKeysSay
t.Cleanup(func() { doneKeyTakeOnce, doneKeyRetry, doneKeysSay = was, wasRetry, wasSay })
doneKeyRetry = 10 * time.Millisecond
doneKeysSay = func(string) {}
var mu sync.Mutex
calls := 0
doneKeyTakeOnce = func(context.Context, *stores, string) error {
mu.Lock()
calls++
n := calls
mu.Unlock()
if n == 1 {
takeDoneKey(context.Background(), nil, "mesh-host") // the newer key's wake, during the start's take-in
}
return nil
}
takeDoneKeyAtStart(context.Background(), nil, "mesh-host")
background.Wait()
mu.Lock()
defer mu.Unlock()
if calls != 2 {
t.Fatalf("taken %d times: the wake during the start's take-in was lost", calls)
}
}
+131 -56
View File
@@ -26,29 +26,52 @@ import (
// every refusal is named in the error. It answers the keys it wrote, by node. Under the lease only: a controller that
// does not hold it writes nothing and says so.
func composeModules(ctx context.Context, open *stores, nodes []string) (map[string][]string, error) {
out, byNode := composeModulesByNode(ctx, open, nodes)
names := make([]string, 0, len(byNode))
for n := range byNode {
names = append(names, n)
}
sort.Strings(names)
var errs []error
for _, n := range names {
errs = append(errs, byNode[n])
}
return out, errors.Join(errs...)
}
// composeModulesByNode is composeModules with each node's error apart: a failure before any node is composed is every
// node's (the composer gives each caller only the errors of the nodes it asked for, issue 1945).
func composeModulesByNode(ctx context.Context, open *stores, nodes []string) (map[string][]string, map[string]error) {
all := func(err error) (map[string][]string, map[string]error) {
byNode := map[string]error{}
for _, n := range nodes {
byNode[n] = err
}
return nil, byNode
}
if _, err := theLease.epoch(ctx); err != nil {
return nil, fmt.Errorf("nothing is composed: %w", err)
return all(fmt.Errorf("nothing is composed: %w", err))
}
bus, err := aBus()
if err != nil {
return nil, fmt.Errorf("nothing is composed: the bus: %w", err)
return all(fmt.Errorf("nothing is composed: the bus: %w", err))
}
defer bus.Close()
js, err := jetstream.New(bus.Conn())
if err != nil {
return nil, err
return all(err)
}
sign, err := controllerSigner(ctx, open)
if err != nil {
return nil, fmt.Errorf("nothing is composed: the controller's signing key: %w", err)
return all(fmt.Errorf("nothing is composed: the controller's signing key: %w", err))
}
fps, err := currentFingerprints(ctx, open.inventory)
if err != nil {
return nil, err
return all(err)
}
entries, err := open.inventory.Catalogued(ctx)
if err != nil {
return nil, err
return all(err)
}
manifests := make(map[string]catalogue.Manifest, len(entries))
for _, e := range entries {
@@ -56,7 +79,7 @@ func composeModules(ctx context.Context, open *stores, nodes []string) (map[stri
}
upgrades, err := open.inventory.Upgrades(ctx)
if err != nil {
return nil, err
return all(err)
}
policy := func(module string) (string, string, string) {
u, ok := upgrades[module]
@@ -67,59 +90,101 @@ func composeModules(ctx context.Context, open *stores, nodes []string) (map[stri
}
sort.Strings(nodes)
out := map[string][]string{}
var errs []error
errsOf := map[string][]error{}
for _, node := range nodes {
at, err := openComposedPlace(ctx, js, bus.Conn(), node)
if err != nil {
errs = append(errs, fmt.Errorf("%s: %w", node, err))
continue
}
at.sign, at.policy = sign, policy
if composeTrace != nil {
traceInput(ctx, open, node)
}
declared, err := composeSend(ctx, open, node)
if err != nil {
errs = append(errs, fmt.Errorf("%s: %w", node, err))
continue
}
body, err := compositionBody(declared)
if err != nil {
errs = append(errs, fmt.Errorf("%s: %w", node, err))
continue
}
// The provisioner's plans of the modules the node's composition hands to it, for the assignments mirror.
if err := open.inventory.RecordComposedPlans(ctx, node, plansKept(declared)); err != nil {
errs = append(errs, fmt.Errorf("%s: its provisioner's plans: %w", node, err))
}
comps, moved, err := compositionsOf(node, body, fps, manifests, declared.modules)
if err != nil {
errs = append(errs, fmt.Errorf("%s: %w", node, err))
continue
}
if len(moved) > 0 {
fmt.Printf("compositions: %s's %s compose with the node's own: no fingerprint of the new form names them\n",
node, strings.Join(moved, ", "))
}
if composeTrace != nil {
traceComposed(ctx, at, node, comps)
}
wrote, err := writeNodeCompositions(ctx, at, comps)
if len(wrote) > 0 {
out[node] = wrote
fmt.Printf("compositions: %s: wrote %s\n", node, strings.Join(wrote, ", "))
}
// And where a data-keeping provision bound each consumer: what the next resolution keeps it at (novox/hq ADR 0232).
if len(declared.Bindings) > 0 {
if err := open.inventory.RecordBindings(ctx, node, declared.Bindings); err != nil {
errs = append(errs, fmt.Errorf("%s: where its consumers are bound to their data: %w", node, err))
// One composition of a node at a time in this process: two at once collided on its keys' revisions (issue 1945).
unlock := lockComposing(node)
func() {
defer unlock()
at, err := openComposedPlace(ctx, js, bus.Conn(), node)
if err != nil {
errsOf[node] = append(errsOf[node], fmt.Errorf("%s: %w", node, err))
return
}
}
if err != nil {
errs = append(errs, fmt.Errorf("%s: %w", node, err))
at.sign, at.policy = sign, policy
if composeTrace != nil {
traceInput(ctx, open, node)
}
declared, err := composeSend(ctx, open, node)
if err != nil {
errsOf[node] = append(errsOf[node], fmt.Errorf("%s: %w", node, err))
return
}
body, err := compositionBody(declared)
if err != nil {
errsOf[node] = append(errsOf[node], fmt.Errorf("%s: %w", node, err))
return
}
// The provisioner's plans of the modules the node's composition hands to it, for the assignments mirror.
if err := open.inventory.RecordComposedPlans(ctx, node, plansKept(declared)); err != nil {
errsOf[node] = append(errsOf[node], fmt.Errorf("%s: its provisioner's plans: %w", node, err))
}
comps, moved, left, err := compositionsOf(node, body, fps, manifests, declared.modules, namedByOwnComposition(ctx, at))
if err != nil {
errsOf[node] = append(errsOf[node], fmt.Errorf("%s: %w", node, err))
return
}
if len(moved) > 0 {
fmt.Printf("compositions: %s's %s compose with the node's own: no fingerprint of the new form names them\n",
node, strings.Join(moved, ", "))
}
if len(left) > 0 {
fmt.Printf("compositions: %s's %s are left out of the node's own: the controller knows no fingerprint of the "+
"new form for them, and the node's want or have names their own composition (issue 1945)\n",
node, strings.Join(left, ", "))
}
if composeTrace != nil {
traceComposed(ctx, at, node, comps)
}
wrote, err := writeNodeCompositions(ctx, at, comps)
if len(wrote) > 0 {
out[node] = wrote
fmt.Printf("compositions: %s: wrote %s\n", node, strings.Join(wrote, ", "))
}
// And where a data-keeping provision bound each consumer: what the next resolution keeps it at (novox/hq ADR 0232).
if len(declared.Bindings) > 0 {
if err := open.inventory.RecordBindings(ctx, node, declared.Bindings); err != nil {
errsOf[node] = append(errsOf[node], fmt.Errorf("%s: where its consumers are bound to their data: %w", node, err))
}
}
if err != nil {
errsOf[node] = append(errsOf[node], fmt.Errorf("%s: %w", node, err))
}
}()
}
byNode := map[string]error{}
for node, errs := range errsOf {
if len(errs) > 0 {
byNode[node] = errors.Join(errs...)
}
}
return out, errors.Join(errs...)
return out, byNode
}
// namedByOwnComposition says a module's want or have on the node names a composition of its own: a fingerprint of the
// new form (issue 1945).
func namedByOwnComposition(ctx context.Context, at composedPlace) func(module string) (bool, error) {
return func(module string) (bool, error) {
if at.wanted != nil {
fp, err := at.wanted(ctx, module)
if err != nil {
return false, err
}
if strings.HasPrefix(fp, sdkdelivery.SourcePrefix2) {
return true, nil
}
}
if at.ran != nil {
h, found, err := at.ran(ctx, module)
if err != nil {
return false, err
}
if found && strings.HasPrefix(h.FP, sdkdelivery.SourcePrefix2) {
return true, nil
}
}
return false, nil
}
}
// composeAndSay is composeModules for an act that has nothing to do with its answer but say it: what each former send
@@ -370,3 +435,13 @@ func compositionBody(declared sendable) ([]byte, error) {
declared.fileReferences = false
return declared.Body()
}
// composingLocks hold one composition of a node at a time in this process (issue 1945).
var composingLocks sync.Map
func lockComposing(node string) func() {
l, _ := composingLocks.LoadOrStore(node, &sync.Mutex{})
mu := l.(*sync.Mutex)
mu.Lock()
return mu.Unlock
}
+3 -3
View File
@@ -26,7 +26,7 @@ func TestALargeFileOfACompositionHasItsObjectInTheBucket(t *testing.T) {
if err != nil {
t.Fatal(err)
}
comps, _, err := compositionsOf("novox", body, nil, nil, nil)
comps, _, _, err := compositionsOf("novox", body, nil, nil, nil, nil)
if err != nil {
t.Fatal(err)
}
@@ -50,7 +50,7 @@ func TestACompositionNamingAnObjectTheBucketLacksIsRefused(t *testing.T) {
if err != nil {
t.Fatal(err)
}
comps, _, err := compositionsOf("novox", body, nil, nil, nil)
comps, _, _, err := compositionsOf("novox", body, nil, nil, nil, nil)
if err != nil {
t.Fatal(err)
}
@@ -78,7 +78,7 @@ func TestEveryFilledFileStaysFilledInTheCompositions(t *testing.T) {
for _, m := range []string{"power", "claude-code", "node-env"} {
delete(fps, m)
}
comps, _, err := compositionsOf(node, body, fps, nil, nil)
comps, _, _, err := compositionsOf(node, body, fps, nil, nil, nil)
if err != nil {
t.Fatalf("%s: %v", node, err)
}
+3 -3
View File
@@ -134,7 +134,7 @@ func watchHavesForJudgement(ctx context.Context, open *stores) {
fmt.Printf("compositions: the haves are not watched: %v\n", err)
return
}
go func() {
goServing(ctx, func() {
defer bus.Close()
defer w.Stop()
for {
@@ -152,7 +152,7 @@ func watchHavesForJudgement(ctx context.Context, open *stores) {
if err != nil || isNode {
continue
}
inBackground(func() {
inBackground(ctx, func() {
if err := rejudgeOn(ctx, open, []string{node}, module); err != nil {
fmt.Printf("compositions: %s on %s is not judged again: %v\n", module, node, err)
}
@@ -165,5 +165,5 @@ func watchHavesForJudgement(ctx context.Context, open *stores) {
})
}
}
}()
})
}
@@ -608,8 +608,11 @@ func keepWhatServeSets(t *testing.T) {
t.Setenv(servedVar, "")
unheld, status, given, check, follow, keeper, hand := logUnheldChanges, statusFrom, givenEvents, checkEvents,
link.Calls.Follow, conditionsFrom, handActConn
doctor, asker := doctorFrom, askerFrom.Load()
t.Cleanup(func() {
logUnheldChanges, statusFrom, givenEvents, checkEvents, link.Calls.Follow, conditionsFrom, handActConn =
unheld, status, given, check, follow, keeper, hand
doctorFrom = doctor
askerFrom.Store(asker)
})
}
+16 -5
View File
@@ -135,9 +135,18 @@ func declarationsDone(nodes []string, runs map[string]string, current string) bo
return true
}
// declarationsRetiredOn reads whether the declarations are done on the mesh now: every node's have of the node-engine
// against the fingerprint it is registered at.
// declarationsRetiredOn reads whether the declarations are done on the mesh now: their stream gone from the bus, or
// every node's have of the node-engine against the fingerprint it is registered at.
//
// **A stream that is gone stays gone** (issue 1945). Release 6 sends no declaration (ADR 0539 clauses 5 and 11), so a
// stream made again would hold no node's last declaration and help no node; and the done test alone is not lasting:
// a node-engine newer than the one registered landing, or a node asleep on an older have, makes it false again after
// the start deleted the stream, and then the raise, the bus's health check (H-bus), the probes (D6, D7) and healer H3
// expected the stream and every node's consumer again, raised them missing and made them anew (2026-10-25 14:52).
func declarationsRetiredOn(ctx context.Context, inv *inventory.Inventory, js jetstream.JetStream) (bool, error) {
if _, err := js.Stream(ctx, broker.DeclarationsStream); errors.Is(err, jetstream.ErrStreamNotFound) {
return true, nil
}
fps, err := currentFingerprints(ctx, inv)
if err != nil {
return false, err
@@ -162,7 +171,8 @@ func declarationsRetiredOn(ctx context.Context, inv *inventory.Inventory, js jet
}
// retireDeclarations deletes the declarations' stream, with every node's consumer on it, once every node runs the new
// node-engine, and says it once; a stream already gone is said by nobody. It answers whether the declarations are done.
// node-engine, and says it once; a stream already gone is said by nobody, and is done. It answers whether the
// declarations are done.
func retireDeclarations(ctx context.Context, open *stores) (bool, error) {
if _, err := theLease.epoch(ctx); err != nil {
return false, err
@@ -324,8 +334,9 @@ func replacedComposition(in compose.Input, r replacedController) (sdkdelivery.Co
return c, nil
}
// declarationsRetiredNow says the declarations are done on the mesh now, for a raise of the bus's objects; a bus or a
// store it cannot read is not done, so the stream is asserted as before.
// declarationsRetiredNow says the declarations are done on the mesh now, for a raise of the bus's objects and for what
// the health check, the probes and the healers expect of the bus; a bus or a store it cannot read is not done, so a
// stream still there is asserted as before.
func declarationsRetiredNow(ctx context.Context, inv *inventory.Inventory) bool {
bus, err := aBus()
if err != nil {
@@ -3,6 +3,7 @@ package main
import (
"context"
"encoding/json"
"fmt"
"strings"
"testing"
@@ -110,3 +111,28 @@ func TestTheDeclarationsGoOnlyOnceEveryNodeRunsTheNewEngine(t *testing.T) {
t.Errorf("done with no fingerprint registered, or no node")
}
}
// streamsGone is a bus that holds none of the streams asked for; every other part of it is unread.
type streamsGone struct{ jetstream.JetStream }
func (streamsGone) Stream(context.Context, string) (jetstream.Stream, error) {
return nil, jetstream.ErrStreamNotFound
}
// Issue 1945 (the alarm of a stream the start deleted): once the declarations' stream is gone from the bus, the
// declarations are retired whatever the nodes run now, so no raise, health check, probe or healer expects the stream
// or a node's consumer on it again. A node-engine newer than the registered one, or a node asleep on an older have,
// made the done test false after the stream went, and the bus's objects named it missing again (2026-10-25 14:52).
func TestTheDeclarationsStayRetiredOnceTheirStreamIsGone(t *testing.T) {
retired, err := func() (retired bool, err error) {
defer func() {
if p := recover(); p != nil {
err = fmt.Errorf("it read what the nodes run although the stream is gone: %v", p)
}
}()
return declarationsRetiredOn(t.Context(), nil, streamsGone{})
}()
if err != nil || !retired {
t.Fatalf("with no declarations' stream on the bus: retired %v (%v), want retired", retired, err)
}
}
+72 -23
View File
@@ -67,8 +67,15 @@ type composedPlace struct {
// fingerprint fps gives it, the node's own with its adoption and provisioner, and every resource of a module fps names
// no fingerprint for in the node's own, its module taken off it and named in moved. Data says the module's manifest
// declares data.
func compositionsOf(node string, body []byte, fps map[string]string, manifests map[string]catalogue.Manifest, runs []string) (
out map[string]sdkdelivery.Composed, moved []string, err error) {
//
// **A module a want or a have names by its own composition is never in the node's own** (issue 1945): the node-engine
// runs that module's composition, and the same resources in the node's own were declared twice, which it refuses
// whole ("2 resources declare X", every node's apply failing from 2026-10-25 12:43). named says, for a module fps names
// no fingerprint of the new form for, that the node's want or have names a fingerprint of the new form; such a module's
// resources and filled paths are left out of this pass (its own composition, as written before, stays what the node
// runs) and it is named in left.
func compositionsOf(node string, body []byte, fps map[string]string, manifests map[string]catalogue.Manifest, runs []string,
named func(module string) (bool, error)) (out map[string]sdkdelivery.Composed, moved, left []string, err error) {
var envelope struct {
Resources []map[string]json.RawMessage `json:"resources"`
@@ -77,9 +84,9 @@ func compositionsOf(node string, body []byte, fps map[string]string, manifests m
Provisioner string `json:"provisioner_account"`
}
if err := json.Unmarshal(body, &envelope); err != nil {
return nil, nil, fmt.Errorf("%s's declaration: %w", node, err)
return nil, nil, nil, fmt.Errorf("%s's declaration: %w", node, err)
}
inModule := map[string]bool{}
inModule, ownComposition := map[string]bool{}, map[string]bool{}
resources := make([]json.RawMessage, 0, len(envelope.Resources))
for _, r := range envelope.Resources {
var module string
@@ -87,15 +94,30 @@ func compositionsOf(node string, body []byte, fps map[string]string, manifests m
_ = json.Unmarshal(raw, &module)
}
if module != "" && !strings.HasPrefix(fps[module], sdkdelivery.SourcePrefix2) {
delete(r, "module")
if !inModule[module] {
inModule[module] = true
moved = append(moved, module)
if !inModule[module] && !ownComposition[module] {
isNamed := false
if named != nil {
if isNamed, err = named(module); err != nil {
return nil, nil, nil, fmt.Errorf("whether %s's want or have names %s's own composition: %w",
node, module, err)
}
}
if isNamed {
ownComposition[module] = true
left = append(left, module)
} else {
inModule[module] = true
moved = append(moved, module)
}
}
if ownComposition[module] {
continue
}
delete(r, "module")
}
raw, err := json.Marshal(r)
if err != nil {
return nil, nil, err
return nil, nil, nil, err
}
resources = append(resources, raw)
}
@@ -106,22 +128,26 @@ func compositionsOf(node string, body []byte, fps map[string]string, manifests m
// and shanks).
var filled, ownFilled []string
for _, p := range envelope.Filled {
if filledByMoved(envelope.Resources, p, inModule) {
switch {
case filledByMoved(envelope.Resources, p, ownComposition):
// The file of a module whose own composition the node runs: that composition names it, not this pass.
case filledByMoved(envelope.Resources, p, inModule):
ownFilled = append(ownFilled, p)
} else {
default:
filled = append(filled, p)
}
}
sort.Strings(moved)
sort.Strings(left)
out, err = sdkdelivery.SplitComposed(node, resources, fps, filled,
sdkdelivery.NodeOwn{Adoption: envelope.Adoption, Provisioner: envelope.Provisioner})
if err != nil {
return nil, nil, err
return nil, nil, nil, err
}
if len(ownFilled) > 0 {
own, err := sdkdelivery.ComposedNodeKey(node)
if err != nil {
return nil, nil, err
return nil, nil, nil, err
}
c := out[own]
c.Filled = append(c.Filled, ownFilled...)
@@ -149,7 +175,7 @@ func compositionsOf(node string, body []byte, fps map[string]string, manifests m
out[key] = c
}
}
return out, moved, nil
return out, moved, left, nil
}
// filledByMoved says a filled path is a file of a module whose resources were moved into the node's own composition.
@@ -218,16 +244,39 @@ func writeComposed(ctx context.Context, at composedPlace, key string, c sdkdeliv
}
wrote = true
}
judged, err := judgeMove(ctx, at, key, c, digest)
if composeTrace != nil {
composeTrace(fmt.Sprintf("%s: %s: written %v, judged %s %q user-list-only %v against %.19s, err %v", at.node, key, wrote,
judged.Judgement, judged.Words, judged.UserListOnly, judged.Against, err))
// **The digest kept is the digest of the composition the bucket holds** (issue 1945): another writer may write its
// composition and its digest between this one's composition and its digest, and the digest written last must not be
// of a composition the bucket no longer holds, or the assigner's want names it and the node applies nothing. After
// its digest, this writer reads the composition again and, when another holds the key, keeps that one's digest.
written := false
for follows := 0; ; follows++ {
judged, err := judgeMove(ctx, at, key, c, digest)
if composeTrace != nil {
composeTrace(fmt.Sprintf("%s: %s: written %v, judged %s %q user-list-only %v against %.19s, err %v", at.node, key,
wrote, judged.Judgement, judged.Words, judged.UserListOnly, judged.Against, err))
}
if err != nil {
return wrote || written, err
}
w, err := keepDigest(ctx, at.digests, key, judged)
written = written || w
if err != nil {
return wrote || written, err
}
now, rev, err := readComposed(ctx, at.bucket, key)
if err != nil {
return wrote || written, err
}
nowDigest, err := sdkdelivery.ComposedDigestOf(now)
if rev == 0 || err != nil || nowDigest == digest {
return wrote || written, nil
}
if follows >= 3 {
return wrote || written, fmt.Errorf("%s moved under this writer %d times; its digest is kept by the next pass",
key, follows+1)
}
c, digest = now, nowDigest
}
if err != nil {
return wrote, err
}
written, err := keepDigest(ctx, at.digests, key, judged)
return wrote || written, err
}
// objectsHeld refuses a composition that names a file by a digest whose object the node's bucket does not hold: a node
+54 -3
View File
@@ -218,7 +218,7 @@ func TestTheCompositionsOfANodeAreItsDeclaration(t *testing.T) {
Resources []json.RawMessage `json:"resources"`
}
_ = json.Unmarshal(body, &d)
comps, moved, err := compositionsOf(node, body, fps, nil, nil)
comps, moved, _, err := compositionsOf(node, body, fps, nil, nil, nil)
if err != nil {
t.Fatalf("%s: %v", node, err)
}
@@ -253,7 +253,7 @@ func TestTheCompositionsOfANodeAreItsDeclaration(t *testing.T) {
// unchanged it writes nothing; a changed input writes a new revision of that key alone, by compare-and-set.
func TestACompositionIsWrittenOnceAndAgainOnlyWhenItChanges(t *testing.T) {
body, fps := capturedBody(t, "g14")
comps, _, err := compositionsOf("g14", body, fps, nil, nil)
comps, _, _, err := compositionsOf("g14", body, fps, nil, nil, nil)
if err != nil {
t.Fatal(err)
}
@@ -342,7 +342,7 @@ func TestCompositionsOutsideTheFiveNewestAreDeleted(t *testing.T) {
func TestAModuleWithNoResourceOfItsOwnStillHasItsComposition(t *testing.T) {
body := []byte(`{"declaration":1,"resources":[]}`)
fp := sdkdelivery.SourcePrefix2 + strings.Repeat("b", 64)
comps, _, err := compositionsOf("laptop", body, map[string]string{"board": fp}, nil, []string{"board", "legacy"})
comps, _, _, err := compositionsOf("laptop", body, map[string]string{"board": fp}, nil, []string{"board", "legacy"}, nil)
if err != nil {
t.Fatal(err)
}
@@ -360,3 +360,54 @@ func TestAModuleWithNoResourceOfItsOwnStillHasItsComposition(t *testing.T) {
}
}
}
// 1945/AC16 (unit): a module the controller knows no fingerprint of the new form for, whose want or have on the node
// names its own composition, is never in the node's own composition, neither its resources nor its filled paths, and is
// named as left; a module neither names still goes with the node's own, its filled path with it. (Live from 2026-10-25
// 12:43: zsh, power, restic and more were in both, and every node refused "2 resources declare X".)
func TestAModuleItsWantNamesIsNeverInTheNodesOwnComposition(t *testing.T) {
body := []byte(`{"declaration":1,"resources":[
{"id":"zsh.rc","type":"file","module":"zsh","path":"/etc/zsh/zshrc"},
{"id":"power.unit","type":"file","module":"power","path":"/etc/power.conf"},
{"id":"mesh-wireguard.conf","type":"file","module":"mesh-wireguard","path":"/etc/wireguard/mesh0.conf"}],
"filled_files":["/etc/zsh/zshrc","/etc/power.conf","/etc/wireguard/mesh0.conf"]}`)
asked := map[string]int{}
named := func(module string) (bool, error) {
asked[module]++
return module == "zsh" || module == "power", nil
}
comps, moved, left, err := compositionsOf("ace", body, map[string]string{}, nil, nil, named)
if err != nil {
t.Fatal(err)
}
if strings.Join(left, ",") != "power,zsh" || strings.Join(moved, ",") != "mesh-wireguard" {
t.Errorf("left %v, moved %v; want power,zsh left and mesh-wireguard moved", left, moved)
}
if asked["zsh"] != 1 || asked["power"] != 1 {
t.Errorf("asked each module once: %v", asked)
}
own, _ := sdkdelivery.ComposedNodeKey("ace")
for key, c := range comps {
for _, id := range resourceIDs(t, c.Resources) {
if strings.HasPrefix(id, "zsh.") || strings.HasPrefix(id, "power.") {
t.Errorf("%s holds %s, which the module's own composition declares", key, id)
}
}
for _, p := range c.Filled {
if p == "/etc/zsh/zshrc" || p == "/etc/power.conf" {
t.Errorf("%s names %s as filled, which the module's own composition names", key, p)
}
}
}
if ids := strings.Join(resourceIDs(t, comps[own].Resources), " "); ids != "mesh-wireguard.conf" {
t.Errorf("the node's own composition holds %q, want mesh-wireguard.conf alone", ids)
}
if f := strings.Join(comps[own].Filled, " "); f != "/etc/wireguard/mesh0.conf" {
t.Errorf("the node's own composition names %q as filled, want mesh-wireguard's file alone", f)
}
if _, _, _, err := compositionsOf("ace", body, map[string]string{}, nil, nil,
func(string) (bool, error) { return false, fmt.Errorf("the wants bucket is gone") }); err == nil {
t.Errorf("a want that cannot be read composed the node anyway")
}
}
+105
View File
@@ -0,0 +1,105 @@
package main
import (
"context"
"errors"
"sort"
"sync"
)
// **One composer, one pass at a time, each node once per pass** (issue 1945). A done key's take-in registers its module
// and asks the composer for its nodes; the composer gathers every node asked for while a pass runs and composes them in
// the next pass, in one composition of each node, then writes the grants once. Before, each take-in of a burst composed
// every node of its module itself, at once: they collided on the compositions' revisions ("wrong last sequence … key
// revision mismatch"), retried every 30 s, and the take-in that mattered waited behind them (2026-10-25 at 11:18 and
// 12:32). A caller waits for the pass that began after it asked, and hears that pass's error.
type composer struct {
composeNodes func(ctx context.Context, nodes []string) map[string]error
writeGrants func(ctx context.Context) error
mu sync.Mutex
pending map[string]bool
grants bool
waiters []waiter
wake chan struct{}
}
// waiter is a caller waiting for its pass: the nodes it asked for, whether it asked for the grants, and where its answer
// goes.
type waiter struct {
nodes []string
grants bool
done chan error
}
func newComposer(compose func(context.Context, []string) map[string]error, grants func(context.Context) error) *composer {
return &composer{composeNodes: compose, writeGrants: grants, pending: map[string]bool{}, wake: make(chan struct{}, 1)}
}
// compose asks for the nodes to be composed, and the grants written after when grants is true, and waits for the pass
// that does it, with the errors of its own nodes alone.
func (c *composer) compose(ctx context.Context, nodes []string, grants bool) error {
done := make(chan error, 1)
c.mu.Lock()
for _, n := range nodes {
c.pending[n] = true
}
c.grants = c.grants || grants
c.waiters = append(c.waiters, waiter{nodes: nodes, grants: grants, done: done})
c.mu.Unlock()
select {
case c.wake <- struct{}{}:
default:
}
select {
case err := <-done:
return err
case <-ctx.Done():
return ctx.Err()
}
}
// run makes the passes until ctx ends: each takes everything asked for since the last began.
func (c *composer) run(ctx context.Context) {
for {
select {
case <-ctx.Done():
return
case <-c.wake:
}
c.mu.Lock()
nodes := make([]string, 0, len(c.pending))
for n := range c.pending {
nodes = append(nodes, n)
}
grants, waiters := c.grants, c.waiters
c.pending, c.grants, c.waiters = map[string]bool{}, false, nil
c.mu.Unlock()
if len(waiters) == 0 {
continue
}
sort.Strings(nodes)
var failed map[string]error
if len(nodes) > 0 {
failed = c.composeNodes(ctx, nodes)
}
// The grants whenever a caller asked for them, whatever a node's composition said: a node that failed fails its
// own callers alone (issue 1945).
var grantsErr error
if grants {
grantsErr = c.writeGrants(ctx)
}
for _, w := range waiters {
var errs []error
for _, n := range w.nodes {
if err := failed[n]; err != nil {
errs = append(errs, err)
}
}
if w.grants && grantsErr != nil {
errs = append(errs, grantsErr)
}
w.done <- errors.Join(errs...)
}
}
}
+143
View File
@@ -0,0 +1,143 @@
package main
import (
"context"
"slices"
"sort"
"sync"
"testing"
"time"
)
// A burst of done keys composes each node once per pass and writes the grants once, never two passes at a time: the
// take-ins of a burst each composed every node of their module at once, collided on the compositions' revisions and
// retried every 30 s, while the one that mattered (mesh-host's) waited (issue 1945, live on 2026-10-25 at 11:18 and
// 12:32).
func TestABurstComposesEachNodeOncePerPass(t *testing.T) {
var mu sync.Mutex
var passes [][]string
grants := 0
running, maxRunning := 0, 0
release := make(chan struct{})
first := true
c := newComposer(func(_ context.Context, nodes []string) map[string]error {
mu.Lock()
running++
maxRunning = max(maxRunning, running)
passes = append(passes, slices.Clone(nodes))
wait := first
first = false
mu.Unlock()
if wait {
<-release // the first pass runs while the rest of the burst arrives
}
mu.Lock()
running--
mu.Unlock()
return nil
}, func(context.Context) error {
mu.Lock()
grants++
mu.Unlock()
return nil
})
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go c.run(ctx)
var wg sync.WaitGroup
wg.Add(1)
go func() { defer wg.Done(); _ = c.compose(ctx, []string{"ace"}, true) }()
time.Sleep(20 * time.Millisecond)
for _, nodes := range [][]string{{"ace", "g14"}, {"g14"}, {"shanks", "ace"}, {"novox"}, {"ace"}} {
wg.Add(1)
go func() { defer wg.Done(); _ = c.compose(ctx, nodes, true) }()
}
time.Sleep(20 * time.Millisecond)
close(release)
wg.Wait()
mu.Lock()
defer mu.Unlock()
if maxRunning != 1 {
t.Fatalf("%d passes ran at once", maxRunning)
}
if len(passes) != 2 {
t.Fatalf("passes %v: want the first and one more for the rest of the burst", passes)
}
second := slices.Clone(passes[1])
sort.Strings(second)
if !slices.Equal(second, []string{"ace", "g14", "novox", "shanks"}) {
t.Fatalf("the second pass composed %v: each node once", passes[1])
}
if grants != 2 {
t.Fatalf("the grants were written %d times for two passes", grants)
}
}
// A caller hears its own pass's failure, and only a pass that began after it asked counts as its own.
func TestACallerHearsThePassThatComposedItsNodes(t *testing.T) {
calls := 0
c := newComposer(func(_ context.Context, nodes []string) map[string]error {
calls++
if calls == 1 {
return map[string]error{"ace": context.DeadlineExceeded}
}
return nil
}, func(context.Context) error { return nil })
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go c.run(ctx)
if err := c.compose(ctx, []string{"ace"}, false); err == nil {
t.Fatal("a failed pass was not heard")
}
if err := c.compose(ctx, []string{"ace"}, false); err != nil {
t.Fatalf("a good pass said %v", err)
}
}
// One node that fails its composition fails only the callers that asked for it, and the grants are written whenever a
// caller asked for them (the review of issue 1945: one joined error failed every take-in of the pass and stopped their
// grants).
func TestANodeThatFailsFailsOnlyItsCallers(t *testing.T) {
grants := 0
release := make(chan struct{})
first := true
c := newComposer(func(_ context.Context, nodes []string) map[string]error {
if first {
first = false
<-release
}
out := map[string]error{}
for _, n := range nodes {
if n == "broken" {
out[n] = context.DeadlineExceeded
}
}
return out
}, func(context.Context) error { grants++; return nil })
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go c.run(ctx)
go func() { _ = c.compose(ctx, []string{"warm"}, false) }()
time.Sleep(20 * time.Millisecond)
errs := make(chan error, 2)
go func() { errs <- c.compose(ctx, []string{"ace"}, true) }()
go func() { errs <- c.compose(ctx, []string{"broken"}, false) }()
time.Sleep(20 * time.Millisecond)
close(release)
var failed, fine int
for range 2 {
if err := <-errs; err != nil {
failed++
} else {
fine++
}
}
if failed != 1 || fine != 1 {
t.Fatalf("failed %d, fine %d: only the caller of the broken node fails", failed, fine)
}
if grants != 1 {
t.Fatalf("the grants were written %d times; a caller asked for them", grants)
}
}
+1 -1
View File
@@ -49,7 +49,7 @@ func keeperOn(ctx context.Context, conn *nats.Conn, looks conditions.LookStore)
Say: func(format string, args ...any) { fmt.Fprintf(os.Stderr, format+"\n", args...) },
// What status leads with changed: composed again soon (a nudge outside the serving controller
// does nothing).
Changed: func() { statusFrom.nudge(); askerFrom.nudge() },
Changed: func() { statusFrom.nudge(); askerFrom.Load().nudge() },
// Written under the lease, carrying its epoch (novox/hq to-be 45 §6).
Epoch: func() (uint64, error) { return theLease.epoch(context.WithoutCancel(ctx)) }}), nil
}
+3
View File
@@ -225,6 +225,9 @@ func filesNotPutObservation(node, summary string) conditions.Observation {
// notePut records how putting a send's objects went: a failure counts, the second in a row raises the condition, a success
// clears the count and the condition.
func notePut(ctx context.Context, node string, put error) {
if errors.Is(put, context.Canceled) || ctx.Err() != nil {
return // a send cut by a cancellation or a stop says nothing of the bucket (issue 1945)
}
filesFailMu.Lock()
if put == nil {
delete(filesFailed, node)
@@ -415,6 +415,7 @@ func TestAPutThatFailsSendAfterSendRaisesAConditionAndClearsWhenItWorks(t *testi
was := conditionsFrom
conditionsFrom = k
t.Cleanup(func() { conditionsFrom = was })
forgetPutFailures(t, "cond-node")
key := filesNotPutObservation("cond-node", "").Key()
open := func() bool {
_, found, err := conditions.ReadOne(t.Context(), store, key)
@@ -442,6 +443,36 @@ func TestAPutThatFailsSendAfterSendRaisesAConditionAndClearsWhenItWorks(t *testi
}
}
// A send cut by a cancellation counts as no failure to put its files, however often (issue 1945: a cancellation is no fault
// of the bucket).
func TestAPutCutByACancellationRaisesNothing(t *testing.T) {
store := conditions.NewInMemory()
k := conditions.NewKeeper(t.Context(), conditions.Options{Store: store, History: store, Teller: &conditions.Told{}})
t.Cleanup(func() { k.Close(context.Background()) })
was := conditionsFrom
conditionsFrom = k
t.Cleanup(func() { conditionsFrom = was })
forgetPutFailures(t, "cut-node")
cut := fmt.Errorf("the files could not be put: %w", context.Canceled)
for range filesNotPutAfter + 1 {
notePut(t.Context(), "cut-node", cut)
}
if _, found, err := conditions.ReadOne(t.Context(), store, filesNotPutObservation("cut-node", "").Key()); err != nil || found {
t.Fatalf("sends cut by a cancellation raised the condition (found %v, %v)", found, err)
}
}
// forgetPutFailures starts the node's count of failed puts at nothing and leaves it so, as a run of the test again finds it.
func forgetPutFailures(t *testing.T, node string) {
forget := func() {
filesFailMu.Lock()
delete(filesFailed, node)
filesFailMu.Unlock()
}
forget()
t.Cleanup(forget)
}
// aBusThatSays is a bus that keeps the declarations published to it and, at each, asks whether the node's bucket is held.
type aBusThatSays struct {
link.Bus
@@ -87,3 +87,29 @@ func TestTheDeclaredPackagesMirrorNamesWhatEachNodesModulesDeclare(t *testing.T)
}
}
}
// 1945/AC-mirror-5: the digest the declared-packages mirror skips an unchanged pass by is made from a real reading of the
// stores, is the same over two readings of the same records, and moves when an assignment does.
func TestTheInputDigestOfTheStoresIsStable(t *testing.T) {
open := aMesh(t)
ctx := t.Context()
register(t, open, catalogue.Manifest{Module: "viewer", Version: "1", Resources: []map[string]any{
{"id": "pkg-curl", "type": "package", "package": "curl"}}})
read := func() string {
in, err := readComposeInput(ctx, open)
if err != nil {
t.Fatal(err)
}
return inputDigest(in)
}
first, second := read(), read()
if first == "" || first != second {
t.Fatalf("two readings of the same records gave the digests %q and %q", first, second)
}
if _, err := assign(ctx, open, "anchor", "viewer"); err != nil {
t.Fatalf("assign: %v", err)
}
if read() == first {
t.Errorf("an assignment left the digest of the stores' Input at %s", first)
}
}
@@ -0,0 +1,93 @@
package main
import (
"context"
"errors"
"fmt"
"testing"
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/novox/mesh-controller/internal/builderkeys"
)
// 1945/AC15 (unit): a module registered at a commit no build record points at (a catalogue merge re-indexed it with
// the fingerprint it had, and the take-in registered it from the done key whose record stands at its first commit)
// has, as its current fingerprint, the one the indexer's trunk value at that commit names; a module whose commit has a
// build keeps that build's; a module the indexer names nothing of the new form for has none; an indexer error is the
// answer's, never a module silently left out.
func TestACommitWithNoBuildOfItsOwnTakesTheIndexersFingerprint(t *testing.T) {
zsh, board := grantFP('a'), grantFP('b')
commits := map[string]string{"zsh": "68f7431d98d7", "board": "2167b95a1111", "legacy": "abcdef012345", "held": ""}
built := func(_ context.Context, module, commit string) (string, bool, error) {
if module == "board" {
return board, true, nil
}
return "", false, nil
}
indexed := func(_ context.Context, module, commit string) (string, error) {
switch module {
case "zsh":
return zsh, nil
case "board":
t.Errorf("the indexer was asked for board, whose commit has a build")
case "held":
t.Errorf("the indexer was asked for a module with no commit")
}
return "", nil
}
got, err := fingerprintsAt(context.Background(), commits, built, indexed)
if err != nil {
t.Fatal(err)
}
if got["zsh"] != zsh || got["board"] != board || len(got) != 2 {
t.Errorf("got %v, want zsh at %s and board at %s alone", got, zsh, board)
}
broken := func(context.Context, string, string) (string, error) { return "", errors.New("the bus is gone") }
if _, err := fingerprintsAt(context.Background(), commits, built, broken); err == nil {
t.Errorf("an indexer error was not the answer's")
}
}
// fakeTrunk is the indexer's trunk values of one module: now, and its history by commit.
type fakeTrunk struct {
now *sdkdelivery.Module
history map[string]sdkdelivery.Module
asked []string
}
func (f *fakeTrunk) ModuleAt(_ context.Context, module, commit string) (sdkdelivery.Module, error) {
f.asked = append(f.asked, commit)
if commit == "" {
if f.now == nil {
return sdkdelivery.Module{}, fmt.Errorf("%w (%s)", builderkeys.ErrNoValue, module)
}
return *f.now, nil
}
m, ok := f.history[commit]
if !ok {
return sdkdelivery.Module{}, fmt.Errorf("%w (%s at %s)", builderkeys.ErrNoValue, module, commit)
}
return m, nil
}
// 1945/AC15 (unit): the indexer's fingerprint at a commit is read from the value it holds now when that is the
// commit's (one read), else from the key's history at the commit; none held is no fingerprint.
func TestTheIndexersFingerprintAtACommit(t *testing.T) {
now := sdkdelivery.Module{Commit: "68f7431d98d7a1279cba3a615b6e9f22c3295783", FP2: grantFP('a')}
f := &fakeTrunk{now: &now}
if fp, err := indexedFingerprintAt(context.Background(), f, "zsh", "68f7431d98d7a1279cba3a615b6e9f22c3295783"); err != nil || fp != grantFP('a') || len(f.asked) != 1 {
t.Errorf("now's commit: %q (%v), asked %v", fp, err, f.asked)
}
f = &fakeTrunk{now: &now, history: map[string]sdkdelivery.Module{"2167b95a": {Commit: "2167b95a", FP2: grantFP('c')}}}
if fp, err := indexedFingerprintAt(context.Background(), f, "zsh", "2167b95a"); err != nil || fp != grantFP('c') {
t.Errorf("an older commit: %q (%v)", fp, err)
}
if fp, err := indexedFingerprintAt(context.Background(), f, "zsh", "99999999"); err != nil || fp != "" {
t.Errorf("a commit the indexer never wrote: %q (%v)", fp, err)
}
if fp, err := indexedFingerprintAt(context.Background(), &fakeTrunk{}, "zsh", "68f7431d"); err != nil || fp != "" {
t.Errorf("no value at all: %q (%v)", fp, err)
}
}
+54 -5
View File
@@ -53,12 +53,61 @@ func startedOn(h sdkdelivery.Have, written time.Time) time.Time {
const givenWhy = "a value given by hand lives only until its module's first good start under the mesh (novox/hq ADR 0228)"
// replaceGiven replaces what a module's good start makes due, composes the machines, and says so.
// replaceGiven replaces what a module's good start makes due, composes the machines, and says so. One at a time for a
// node: a call that finds one running leaves what it was asked for as a mark, and the one running takes it once it is
// done, so a start heard meanwhile is never lost (issue 1945, the kind of the dropped wake).
func replaceGiven(ctx context.Context, open *stores, node string, started map[string]time.Time) {
if _, busy := replacing.LoadOrStore(node, true); busy {
markGivenAgain(node, started)
return
}
defer replacing.Delete(node)
for {
replaceGivenOnce(ctx, open, node, started)
replacing.Delete(node)
// Read after letting go: a call meanwhile only marked the node.
pending := takeGivenMark(node)
if pending == nil || ctx.Err() != nil {
return
}
if _, busy := replacing.LoadOrStore(node, true); busy {
markGivenAgain(node, pending) // the one running now takes it
return
}
started = pending
}
}
// givenAgain is, by node, the starts asked for while a replacement ran there, merged.
var givenAgain = struct {
sync.Mutex
by map[string]map[string]time.Time
}{by: map[string]map[string]time.Time{}}
func markGivenAgain(node string, started map[string]time.Time) {
givenAgain.Lock()
defer givenAgain.Unlock()
m := givenAgain.by[node]
if m == nil {
m = map[string]time.Time{}
givenAgain.by[node] = m
}
for module, at := range started {
if at.After(m[module]) {
m[module] = at
}
}
}
func takeGivenMark(node string) map[string]time.Time {
givenAgain.Lock()
defer givenAgain.Unlock()
m := givenAgain.by[node]
delete(givenAgain.by, node)
return m
}
// replaceGivenOnce is one replacement on a node; a variable so a test can stand in for it.
var replaceGivenOnce = func(ctx context.Context, open *stores, node string, started map[string]time.Time) {
replaced, err := open.inventory.ReplaceGivenAfterStart(ctx, node, started)
if err != nil {
log.Printf("a value given by hand on %s could not be replaced after its module started: %v", node, err)
@@ -120,7 +169,7 @@ func watchHavesForGiven(ctx context.Context, open *stores) {
log.Printf("given values: the haves are not watched: %v", err)
return
}
go func() {
goServing(ctx, func() {
defer bus.Close()
defer w.Stop()
for {
@@ -146,8 +195,8 @@ func watchHavesForGiven(ctx context.Context, open *stores) {
if at.IsZero() {
continue
}
inBackground(func() { replaceGiven(ctx, open, node, map[string]time.Time{module: at}) })
inBackground(ctx, func() { replaceGiven(ctx, open, node, map[string]time.Time{module: at}) })
}
}
}()
})
}
+34
View File
@@ -0,0 +1,34 @@
package main
import (
"context"
"sync"
"testing"
"time"
)
// A start heard while a replacement of given values runs on the node is not lost: it is marked, and the replacement
// running takes it once it is done (issue 1945, the kind of the dropped wake: replaceGiven returned at once when busy).
func TestAStartHeardDuringAReplacementIsTaken(t *testing.T) {
was := replaceGivenOnce
t.Cleanup(func() { replaceGivenOnce = was })
var mu sync.Mutex
var asked []map[string]time.Time
at := time.Date(2026, 10, 25, 13, 0, 0, 0, time.UTC)
replaceGivenOnce = func(ctx context.Context, open *stores, node string, started map[string]time.Time) {
mu.Lock()
asked = append(asked, started)
first := len(asked) == 1
mu.Unlock()
if first {
// Another module starts on the node while this one is being replaced.
replaceGiven(ctx, open, node, map[string]time.Time{"redis": at})
}
}
replaceGiven(context.Background(), nil, "ace", map[string]time.Time{"postgres": at})
mu.Lock()
defer mu.Unlock()
if len(asked) != 2 || !asked[1]["redis"].Equal(at) {
t.Fatalf("asked %v: the start heard during the replacement was lost", asked)
}
}
+108 -8
View File
@@ -11,6 +11,7 @@ import (
"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/inventory"
)
@@ -55,29 +56,128 @@ func grantsFrom(principals []broker.Principal, hashes map[string]string, current
// currentFingerprints is, by module, the fingerprint of the build each registered module is registered at: its
// manifest's commit (built_from), that commit's newest successful build, and the fingerprint that build names.
//
// **A commit with no build of its own asks the indexer** (issue 1945, live from 2026-10-25 12:43). A catalogue merge
// that leaves a module's sources alone re-indexes it at the merge's commit with the fingerprint it had: the take-in
// registers the module at the new commit from the same done key, whose build record (one per key and revision) already
// stands at the commit it was first taken in at, so the new commit has no build pointing at it. Read only by the
// commit, such a module had no fingerprint of the new form: it was composed into the node's own composition while its
// want named its own, and every node declared its files twice. The indexer's trunk value at the commit names the
// fingerprint the module was registered from, so that is its fingerprint.
func currentFingerprints(ctx context.Context, inv *inventory.Inventory) (map[string]string, error) {
builds, err := inv.CurrentBuilds(ctx)
if err != nil {
return nil, err
}
out := map[string]string{}
commits := make(map[string]string, len(builds))
for module, b := range builds {
if b.Commit == "" {
commits[module] = b.Commit
}
var indexed func(ctx context.Context, module, commit string) (string, error)
var closeIndexed func()
defer func() {
if closeIndexed != nil {
closeIndexed()
}
}()
return fingerprintsAt(ctx, commits,
func(ctx context.Context, module, commit string) (string, bool, error) {
id, err := inv.BuildOf(ctx, module, commit)
if err != nil || id == "" {
return "", false, nil
}
build, found, err := inv.BuildByID(ctx, id)
if err != nil || !found {
return "", false, nil
}
return build.SourceFingerprint, true, nil
},
func(ctx context.Context, module, commit string) (string, error) {
if indexed == nil {
read, done, err := openIndexedFingerprints()
if err != nil {
return "", err
}
indexed, closeIndexed = read, done
}
return indexed(ctx, module, commit)
})
}
// fingerprintsAt is, by module, the fingerprint of the build registered at its commit (built), or else the fingerprint
// of the new form the indexer's trunk value at that commit names (indexed). A module with no commit (held without a
// source) has none, and a module neither knows has none. An error of the indexer's is the answer's: a module left out
// for it would be composed into the node's own composition beside its own (issue 1945).
func fingerprintsAt(ctx context.Context, commits map[string]string,
built func(ctx context.Context, module, commit string) (string, bool, error),
indexed func(ctx context.Context, module, commit string) (string, error)) (map[string]string, error) {
modules := make([]string, 0, len(commits))
for module := range commits {
modules = append(modules, module)
}
sort.Strings(modules)
out := map[string]string{}
for _, module := range modules {
commit := commits[module]
if commit == "" {
continue
}
id, err := inv.BuildOf(ctx, module, b.Commit)
if err != nil || id == "" {
fp, found, err := built(ctx, module, commit)
if err != nil {
return nil, err
}
if found {
out[module] = fp
continue
}
build, found, err := inv.BuildByID(ctx, id)
if err != nil || !found {
continue
fp, err = indexed(ctx, module, commit)
if err != nil {
return nil, fmt.Errorf("the fingerprint %s is registered at (%.8s): %w", module, commit, err)
}
if strings.HasPrefix(fp, sdkdelivery.SourcePrefix2) {
out[module] = fp
}
out[module] = build.SourceFingerprint
}
return out, nil
}
// openIndexedFingerprints opens the indexer's trunk values: what reads the fingerprint of the new form a module's value
// at a commit names ("" when it holds none for that commit), and what closes the bus connection it opened.
var openIndexedFingerprints = func() (func(ctx context.Context, module, commit string) (string, error), func(), error) {
keys, closeKeys, err := openBuilderKeys()
if err != nil {
return nil, nil, fmt.Errorf("the indexer's trunk values: %w", err)
}
return func(ctx context.Context, module, commit string) (string, error) {
return indexedFingerprintAt(ctx, keys, module, commit)
}, closeKeys, nil
}
// indexedFingerprintAt is the fingerprint of the new form the indexer's value of module at commit names: the value it
// holds now when that is the commit's (one read), else the newest of the key's history at the commit; "" when none.
func indexedFingerprintAt(ctx context.Context, keys interface {
ModuleAt(ctx context.Context, module, commit string) (sdkdelivery.Module, error)
}, module, commit string) (string, error) {
m, err := keys.ModuleAt(ctx, module, "")
if err == nil && sameCommitID(m.Commit, commit) {
return m.FP2, nil
}
if err != nil && !errors.Is(err, builderkeys.ErrNoValue) {
return "", err
}
if errors.Is(err, builderkeys.ErrNoValue) {
return "", nil
}
m, err = keys.ModuleAt(ctx, module, commit)
if errors.Is(err, builderkeys.ErrNoValue) {
return "", nil
}
if err != nil {
return "", err
}
return m.FP2, nil
}
// grantsKV is what writing the grants needs of their bucket.
type grantsKV interface {
Get(ctx context.Context, key string) (jetstream.KeyValueEntry, error)
+31 -9
View File
@@ -965,6 +965,26 @@ func processPrincipalFor(m catalogue.Manifest, processID string) (broker.Process
return broker.ProcessPrincipal{}, false
}
// mintOnlyIfAbsent marks a context whose issue mints only accounts no row names yet (onlyIfAbsent).
type mintOnlyIfAbsent struct{}
// onlyIfAbsent is ctx for an issue that must never replace a minted account: the assign act's and the assignments'
// watch's, which may run at once for one account (novox/hq ADR 0543, the review of issue 1945).
func onlyIfAbsent(ctx context.Context) context.Context {
return context.WithValue(ctx, mintOnlyIfAbsent{}, true)
}
// mintBusUser mints u's password. By default it replaces whatever stands (a rotation: `module issue`); under
// onlyIfAbsent it mints only when no row names u, and won false says another caller minted it first and seals it,
// so this one seals nothing.
func mintBusUser(ctx context.Context, inv *inventory.Inventory, u inventory.BusUser) (string, bool, error) {
if only, _ := ctx.Value(mintOnlyIfAbsent{}).(bool); only {
return inv.MintBusPasswordIfAbsent(ctx, u)
}
password, err := inv.MintBusPassword(ctx, u)
return password, err == nil, err
}
// issueOnTheNewBus gives an assigned module its credential on the bus being built.
//
// **Three things differ from a management call, and each is the point of the move.** The credential
@@ -990,14 +1010,16 @@ func issueOnTheNewBus(ctx context.Context, inv *inventory.Inventory, m catalogue
}
if _, reads := m.OwnSecrets["broker"]; reads {
user := broker.Principal{Kind: broker.KindModule, Node: node, Module: m.Module}.Username()
password, err := inv.MintBusPassword(ctx, inventory.BusUser{
password, won, err := mintBusUser(ctx, inv, inventory.BusUser{
Username: user, Kind: busKindOf(m.Module), Node: node, Module: m.Module,
})
if err != nil {
return err
}
if err := issueWith(ctx, inv, m, node, busAddress, known, reachable, user, password); err != nil {
return err
if won {
if err := issueWith(ctx, inv, m, node, busAddress, known, reachable, user, password); err != nil {
return err
}
}
}
// Every process with a bus declaration gets an account of its own (novox/hq ADR 0436 §1, §2), minted and
@@ -1021,11 +1043,11 @@ func issueOnTheNewBus(ctx context.Context, inv *inventory.Inventory, m catalogue
func issueProcess(ctx context.Context, inv *inventory.Inventory, m catalogue.Manifest, pr broker.ProcessPrincipal,
node, busAddress string, known broker.Broker, reachable string) error {
user := broker.Principal{Kind: broker.KindProcess, Node: node, Module: m.Module, Process: pr.Process}.Username()
password, err := inv.MintBusPassword(ctx, inventory.BusUser{
password, won, err := mintBusUser(ctx, inv, inventory.BusUser{
Username: user, Kind: inventory.BusProcess, Node: node, Module: m.Module,
})
if err != nil {
return err
if err != nil || !won {
return err // lost: the caller that minted it seals it
}
name := processSecretName(pr.Process)
held, err := heldProcessAccount(known, reachable, node, m.Module, user, password)
@@ -1091,11 +1113,11 @@ func runsTheProvisionerClient(m catalogue.Manifest) bool {
func issueProvisionerClient(ctx context.Context, inv *inventory.Inventory, m catalogue.Manifest, node string,
known broker.Broker, reachable string) error {
user := broker.Principal{Kind: broker.KindProvisioner, Node: node, Module: m.Module}.Username()
password, err := inv.MintBusPassword(ctx, inventory.BusUser{
password, won, err := mintBusUser(ctx, inv, inventory.BusUser{
Username: user, Kind: inventory.BusProvisioner, Node: node, Module: m.Module,
})
if err != nil {
return err
if err != nil || !won {
return err // lost: the caller that minted it seals it
}
held, err := json.Marshal(struct {
URL string `json:"url"`
+16 -10
View File
@@ -200,14 +200,17 @@ func newModuleObservations(now time.Time) []conditions.Observation {
// watchNewModules wakes the take-in of new modules for each module the builder or the indexer writes, until ctx ends, and
// looks at every module once now. A bucket not there yet is said: the look at start and a first merge's word still work.
func watchNewModules(ctx context.Context, open *stores) {
inBackground(func() {
inBackground(ctx, func() {
takeNewModules(ctx, open)
})
// And, beside it, the start's pass of release 6 and the take-in of each registered module's new done key (novox/hq
// ADR 0539 clauses 5, 7, 10 and 13; composed_start.go, composed_keys.go): the plans' buckets deleted, every node
// composed, the replaced controller composed, and from now on each new fingerprint registered, composed and granted.
inBackground(func() { composeAtStart(ctx, open) })
inBackground(ctx, func() { composeAtStart(ctx, open) })
watchDoneKeys(ctx, open)
// And each assignment that comes to stand issues its node's bus accounts and the module's grant (ADR 0543): a module
// built long before is assigned to one more node with no build done again (issue 1945, assignment_accounts.go).
watchAssignmentAccounts(ctx, open)
watchHavesForJudgement(ctx, open)
// And the keep set keeps the builds every want names (ADR 0539 clause 1), with no plan left to name them.
open.inventory.SetWanted(wantedOnTheBus)
@@ -223,16 +226,16 @@ func watchNewModules(ctx context.Context, open *stores) {
fmt.Printf("new modules: the builder's keys are not watched: %v\n", err)
return
}
go func() {
goServing(ctx, func() {
<-ctx.Done()
closeKeys()
}()
})
}
// wakeNewModule takes module in when the catalogue does not hold it: never on the watch's own goroutine, which must keep
// draining.
func wakeNewModule(ctx context.Context, open *stores, module string) {
inBackground(func() {
inBackground(ctx, func() {
entries, err := open.inventory.Catalogued(ctx)
if err != nil {
return
@@ -250,14 +253,17 @@ func wakeNewModule(ctx context.Context, open *stores, module string) {
// ends only once the work its mesh started has.
var background sync.WaitGroup
// inBackground runs work beside the caller, counted in background. A variable so a test of what a walk does around a
// take-in may run it in the caller's turn and need not wait for it.
var inBackground = func(work func()) {
// inBackground runs work beside the caller, counted in background and, under a serving controller, in its serving group,
// so serve's stop waits for it and a serve that is stopping starts none (serving_group.go). A variable so a test of what a
// walk does around a take-in may run it in the caller's turn and need not wait for it.
var inBackground = func(ctx context.Context, work func()) {
background.Add(1)
go func() {
if !goServing(ctx, func() {
defer background.Done()
work()
}()
}) {
background.Done()
}
}
// newModulesOf are the modules the indexer holds in a repository: what a first merge's word wakes.
+11
View File
@@ -59,6 +59,17 @@ func dataWords(o conditions.Observation) (what, module string) {
}
var plainWordings = map[string]func(conditions.Observation) words{
// A module's new build whose compositions or grant the controller could not write (composed_keys.go).
kindNotComposed: worded(func(o conditions.Observation) words {
m := "a module"
if p := idPart(o, 0); o.Scope == conditions.ScopeBuild && p != "" {
m = p // the module: raised in the build's scope, its id the module's name
}
return words{Headline: "A new build of " + m + " cannot be delivered",
Explanation: fmt.Sprintf("A new build of %s was made, and the mesh could not prepare it for the machines, so "+
"none can be given it yet. It is tried again every half minute; the details say what fails.", m),
Resolved: "the new build of " + m + " can be delivered"}
}),
// Machines and their node-engine.
"silent": worded(func(o conditions.Observation) words {
m := machineOr(o, "a machine")
+2
View File
@@ -72,6 +72,8 @@ func issueMissingProcessAccounts(ctx context.Context, inv *inventory.Inventory,
if len(m.ProcessBuses()) == 0 || mayIssue(m) != nil {
return "", nil
}
// Only the missing: never a second password over one another caller minted between this look and the mint.
ctx = onlyIfAbsent(ctx)
minted, err := inv.BusMinted(ctx)
if err != nil {
return "", err
-58
View File
@@ -1,12 +1,9 @@
package main
import (
"encoding/json"
"strings"
"testing"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"github.com/novox/mesh-controller/internal/broker"
)
// mailSenderWithProcess is mesh-mail's real manifest shape (novox/hq ADR 0436 §7, §8, the lead's fixture):
@@ -25,61 +22,6 @@ func mailSenderWithProcess() catalogue.Manifest {
return m
}
// review round 1, finding 1 (1181): a module declaring no `broker` of its own at all — mesh-mail's real
// shape, tools served by the tool runner — still passes mayIssue on its process's bus declaration alone.
func TestAModuleWithNoBrokerAndOnlyAProcessMayBeIssued(t *testing.T) {
if err := mayIssue(mailSenderWithProcess()); err != nil {
t.Fatalf("a module with no broker of its own, whose process declares bus, was refused: %v", err)
}
}
// 1181/AC1: a process with a bus declaration and no own secret broker-<process id> cannot be issued — the
// same refusal mayIssue already gives a module with no `broker` (novox/hq ADR 0436 §2).
func TestAProcessWithNoBrokerSecretCannotBeIssued(t *testing.T) {
m := mailSenderWithProcess()
delete(m.OwnSecrets, "broker-send")
err := mayIssue(m)
if err == nil {
t.Fatal("a process with no broker-<id> own secret was issued an account")
}
for _, want := range []string{"mesh-mail", "send", "broker-send", "own-secrets"} {
if !strings.Contains(err.Error(), want) {
t.Errorf("the refusal does not say %q: %v", want, err)
}
}
if err := mayIssue(mailSenderWithProcess()); err != nil {
t.Errorf("a process declaring its own broker-<id> secret was refused: %v", err)
}
}
// 1181/AC1: the account held for a process is the provisioner client's form — {url, fingerprint, node,
// module, user, password} — named for the process, never the module's own account.
func TestAProcesssHeldAccountIsTheProvisionersForm(t *testing.T) {
user := broker.Principal{Kind: broker.KindProcess, Node: "novox", Module: "mesh-mail", Process: "send"}.Username()
if user != "novox.mesh-mail.send" {
t.Fatalf("the process's username: %q", user)
}
held, err := heldProcessAccount(broker.Broker{Fingerprint: "abc"}, "novox.example:4222", "novox", "mesh-mail",
user, "a-password")
if err != nil {
t.Fatal(err)
}
var got map[string]any
if err := json.Unmarshal(held, &got); err != nil {
t.Fatal(err)
}
want := map[string]any{"url": "nats://novox.example:4222", "fingerprint": "abc", "node": "novox",
"module": "mesh-mail", "user": "novox.mesh-mail.send", "password": "a-password"}
for k, v := range want {
if got[k] != v {
t.Errorf("%s: %v, want %v", k, got[k], v)
}
}
if len(got) != len(want) {
t.Errorf("the held account carries %v; want exactly %v (never claims — a process holds no seat)", got, want)
}
}
// 1181/AC1: composed, the process's own secret belongs to its own account at 0600 (novox/hq ADR 0436 §1, §2;
// the owner mechanism is issue 1170's, proven generically there — this is mesh-mail's own shape).
func TestAProcessesOwnSecretIsOwnedByItsOwnAccount(t *testing.T) {
+39 -16
View File
@@ -59,6 +59,9 @@ func serve(ctx context.Context) (err error) {
}
defer open.Close()
inv := open.inventory
// Lent to every reader in this process that holds only the inventory, so none opens pools of its own per call
// (issues 1945 and 1277: about 120 take-ins at once filled novox's store at 16:08 on 2026-10-25).
defer shareStores(open)()
// **What this build reads of the store's schema, on record** (novox/hq issue 352): the highest migration
// it carries, by its version, so a gate that would put this build back later knows it reads the store
// as it is then. A build that does not know its version records nothing, and is never put back.
@@ -75,11 +78,11 @@ func serve(ctx context.Context) (err error) {
"recorded, and a gate will never put it back\n", RunningBuildVar)
}
ident, err := openIdentity(ctx)
// The controller's own identity store, the one its readers share (closed with open), not a second pool.
ident, err := open.Identity(ctx)
if err != nil {
return err
}
defer ident.Close()
// Established at start rather than on first use. A control plane that cannot sign is one
// whose declarations every node correctly refuses, and that should be a startup failure
@@ -122,7 +125,15 @@ func serve(ctx context.Context) (err error) {
defer theLease.release()
ctx, stopActing := context.WithCancel(ctx)
defer stopActing()
go func() {
// Every goroutine started from here is counted, and ended and waited for before serve returns (serving_group.go):
// on the way out by error here, and at the end before anything else is undone.
ctx, running := withServingGroup(ctx)
stopRunning := func() {
stopActing()
running.stop()
}
defer stopRunning()
goServing(ctx, func() {
select {
case <-lost:
stopActing()
@@ -132,7 +143,7 @@ func serve(ctx context.Context) (err error) {
// asks the controller after this one rather than have a command started and killed with this
// process (novox/hq issue 289).
handingOver.Store(true)
}()
})
defer func() {
select {
case <-lost:
@@ -159,6 +170,17 @@ func serve(ctx context.Context) (err error) {
if err := attachTheChain(ctx, inv, server); err != nil {
return err
}
bus, isNATS := server.Bus().(link.OverNATS)
if !isNATS {
return errors.New("the mesh's verbs are served over the bus, and this control plane is not on it")
}
// The condition store, opened before anything that raises or reads a condition starts beside the serving, so none
// of them meets conditionsFrom being written (issue 1945); closed after they all ended.
keeper := openServingConditions(ctx, open, bus)
if keeper != nil {
defer closeConditions(keeper)
defer stopRunning() // on an error return, the goroutines end before the keeper closes
}
// The bus's own objects, asserted on every start. **Not created once at genesis**: a stream
// somebody deleted, a mesh raised from a restored backup, or a bus whose data directory was
@@ -173,7 +195,7 @@ func serve(ctx context.Context) (err error) {
// And the pending assignments settled on a tick of their own (novox/hq ADR 0261): made once their module
// is registered, ended with why when its build will not register it, raised and cleared as conditions.
// Never by a read.
go settlingPending(ctx, open)
goServing(ctx, func() { settlingPending(ctx, open) })
// The durations the core's bounds are set from are kept a month (novox/hq to-be 45 Phase 0).
// And what the catalogue decided a build meant. The builder's own result is already handled
// above; this is the other half — the control plane is the only one of the three that knows
@@ -221,19 +243,17 @@ func serve(ctx context.Context) (err error) {
"Those answer the reason when called; everything else is served as usual\n",
catalogue.ControllerSeatName, strings.Join(behind, ", "))
}
bus, isNATS := server.Bus().(link.OverNATS)
if !isNATS {
return errors.New("the mesh's verbs are served over the bus, and this control plane is not on it")
}
// The hand-act log is counted for `status` on this connection rather than a new one a minute; no longer serving, the
// connection is closed, and a verb run after in this process dials its own (a served test's leftover failed the
// next test's `conditions` with "connection closed", issue 1945).
handActConn = bus.Conn
defer func() { handActConn = nil }()
defer stopRunning() // on an error return, the goroutines end before handActConn is put back
// And everything else this controller does on the bus for a moment (novox/hq issue 327).
servingBus.Store(server.JetStream())
// No longer serving: nothing is lent, and dead-letters says it is not read here (novox/hq issue 330).
defer servingBus.Store(nil)
defer stopRunning() // on an error return, the goroutines end before the bus is taken back
// And says when it replaced a value given by hand (novox/hq ADR 0228).
givenEvents = bus
// And a pull request's merge check, asked when the forge announces its head and said when judged
@@ -244,18 +264,18 @@ func serve(ctx context.Context) (err error) {
return err
}
// Composed now and kept current, before the verb that answers from it is served.
go statusFrom.keep(ctx)
goServing(ctx, func() { statusFrom.keep(ctx) })
// The facts snapshot a merge check is fed (novox/hq to-be 45 §9): kept current while this
// controller holds the lease, read by the build seat from the artifact store.
go exportingFacts(ctx, open, bus.Conn.ConnectedServerVersion)
goServing(ctx, func() { exportingFacts(ctx, open, bus.Conn.ConnectedServerVersion) })
// The record mirrors (novox/hq ADR 0289, design 57 F3): the assignments and each registered module's
// declared config on the bus, written from the store only when a record changed, by the lease holder.
// The provisioner's plan of each handed module, as its machine was last sent it (design 57 P6), in the module's
// assignment.
inv.SetProvisioningPlans(inv.SentProvisioningPlans)
go mirroringRecords(ctx, open, inv, bus.Conn, bus)
goServing(ctx, func() { mirroringRecords(ctx, open, inv, bus.Conn, bus) })
// And the bucket of the files that travel by digest, judged against what each machine's latest declaration names (ADR 0486 §3).
go sweepingDeclarationFiles(ctx, inv, bus.Conn)
goServing(ctx, func() { sweepingDeclarationFiles(ctx, inv, bus.Conn) })
// And the modules the catalogue does not hold yet, registered from the builder's first done key of each, woken by
// the builder's and the indexer's writes (novox/hq ADR 0501 §6, issue 1851): nothing asks a first build.
watchNewModules(ctx, open)
@@ -287,7 +307,7 @@ func serve(ctx context.Context) (err error) {
// signals table, what the bus says about itself, and the self-check. A store that cannot be opened
// is said and the controller serves on: status then says the conditions cannot be read, and is
// not well — louder than not serving at all, and the push that repairs the bus still runs.
if stopWatching := watchTheMesh(ctx, open, server, bus); stopWatching != nil {
if stopWatching := watchTheMesh(ctx, open, server, bus, keeper); stopWatching != nil {
defer stopWatching()
}
stopServing, err := bus.ServeSeatTools(catalogue.ControllerSeatName, handlers, log.New(os.Stdout, "", log.LstdFlags))
@@ -302,7 +322,7 @@ func serve(ctx context.Context) (err error) {
}
defer stopCLI()
// Its read calls write no line each; how many each caller made is said once an hour (novox/hq issue 762).
go sayReadCalls(ctx)
goServing(ctx, func() { sayReadCalls(ctx) })
// And says so on the bus (novox/hq ADR 0197): what it serves, as the NATS services protocol asks.
stopAnnouncing, err := bus.Announce(seatAnnouncement(handlers), log.New(os.Stdout, "", log.LstdFlags))
if err != nil {
@@ -310,7 +330,10 @@ func serve(ctx context.Context) (err error) {
}
defer stopAnnouncing()
return server.Serve(ctx)
err = server.Serve(ctx)
// Ended and waited for before the verbs stop being served and the globals are put back, not after.
stopRunning()
return err
}
// OverlayCIDRVar is the range the mesh allocates node addresses from.
+266 -24
View File
@@ -2,12 +2,15 @@ package main
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"sort"
"time"
"git.novox.be/novox/mesh-sdk/go/compose"
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
@@ -24,7 +27,8 @@ import (
// so no one call site sees every change; what every change has in common is that it is committed to the
// store. So the serving controller reads the store every mirrorEvery and writes only what differs from what
// the bucket holds (ADR 0289 §1): a quiet mesh writes nothing, and a change rolled back is never read. A
// mirror is therefore at most mirrorEvery behind its store.
// mirror is therefore at most mirrorEvery behind its store. The declared-packages mirror is the exception: it composes
// every node, so it passes on a change of what it composes from, not on this tick (keepingDeclaredPackages, issue 1945).
// mirrorEvery is how often the mirrors are compared with the store: two small reads of the store.
const mirrorEvery = 10 * time.Second
@@ -45,21 +49,31 @@ func mirroringRecords(ctx context.Context, open *stores, inv *inventory.Inventor
return e
}
sources := map[string]func(context.Context) (map[string][]byte, error){
recordmirror.DeclaredConfig.Name: inv.DeclaredConfigMirror,
recordmirror.Predefined.Name: inv.PredefinedMirror,
recordmirror.DeclaredPackages.Name: declaredPackagesMirror(open),
recordmirror.DeclaredConfig.Name: inv.DeclaredConfigMirror,
recordmirror.Predefined.Name: inv.PredefinedMirror,
}
packages := keepingDeclaredPackages(ctx, open, api, announce, epoch, holds)
held := false
writers := map[string]*recordmirror.Writer{}
failing := map[string]string{}
tick := time.NewTicker(mirrorEvery)
defer tick.Stop()
events := &assignmentEvents{}
for {
if holds() {
now := holds()
if now && !held {
// The lease taken: the declared-packages mirror is brought to the stores at once, not at its next change.
packages.poke()
}
held = now
if now {
// The assignments are keys of the bus, written by the verbs that change them (novox/hq ADR 0408): this tick
// keeps the provisioner's plan in their values and says what changed. They are no mirror of the store.
keepingAssignments(ctx, inv, announce, epoch, events, failing)
for _, m := range recordmirror.Mirrors() {
if sources[m.Name] == nil {
continue // the declared packages, kept on a change (keepingDeclaredPackages)
}
why := mirrorOnce(ctx, api, m, writers, sources[m.Name], announce, epoch, holds)
if why != failing[m.Name] {
if why != "" {
@@ -179,35 +193,263 @@ func keepingAssignments(ctx context.Context, inv *inventory.Inventory, announce
events.known = now
}
// declaredPackagesMirror is the source of the declared-packages mirror (ADR 0507 §5): each node's composition, asked as a
// question that makes nothing, and the packages it declares with the modules that declare them. A node whose own set does
// not compose has no key, and any other failure to read is the pass's error, so a key is never purged for a store that
// could not be read.
// declaredPackagesMirror is the source of the declared-packages mirror (ADR 0507 §5) over a fresh reading of the stores.
func declaredPackagesMirror(open *stores) func(context.Context) (map[string][]byte, error) {
return func(ctx context.Context) (map[string][]byte, error) {
in, err := inputFor(ctx, open)
if err != nil {
return nil, err
}
// Which module owns each resource is carried only for a node-engine that reads it, so every node is asked as one that does.
asked := in.Input
asked.Nodes = append([]compose.Node(nil), in.Nodes...)
for i := range asked.Nodes {
asked.Nodes[i].ReadsModules = true
}
out := map[string][]byte{}
for _, n := range in.sortedNodes() {
d, _, err := compose.ComposeIn(asked, n.Name, compose.Foreseeing)
if compose.IsUnresolvable(err) {
continue
return declaredPackagesFrom(ctx, in)
}
}
// declaredPackagesFrom is the declared-packages mirror over an Input: each node's composition, asked as a question that makes
// nothing, and the packages it declares with the modules that declare them. A node whose own set does not compose has no key,
// and any other failure to read is the pass's error, so a key is never purged for a store that could not be read.
func declaredPackagesFrom(_ context.Context, in *meshInput) (map[string][]byte, error) {
// Which module owns each resource is carried only for a node-engine that reads it, so every node is asked as one that does.
asked := in.Input
asked.Nodes = append([]compose.Node(nil), in.Nodes...)
for i := range asked.Nodes {
asked.Nodes[i].ReadsModules = true
}
out := map[string][]byte{}
for _, n := range in.sortedNodes() {
d, _, err := compose.ComposeIn(asked, n.Name, compose.Foreseeing)
if compose.IsUnresolvable(err) {
continue
}
if err != nil {
return nil, fmt.Errorf("the packages %s declares cannot be read: %w", n.Name, err)
}
if out[n.Name], err = inventory.DeclaredPackagesValue(n.Name, d.Resources, d.ModuleOf); err != nil {
return nil, err
}
}
return out, nil
}
// **The declared-packages mirror passes on a change, not on a tick** (issue 1945 note 22). Its source composes every node in
// full, about 45 s on the control node; on the 10 s tick it never rested and held the controller at 130-175 % CPU. Every
// change that can move a node's packages is a change of its composition, which the controller writes into the bucket of the
// compositions' digests, and of the assignments, which the verbs write into theirs: a watch of those two buckets wakes the
// keeper. A pass reads the stores' Input once and composes only when the Input's digest differs from the one it last
// mirrored, so a wake that changed nothing it composes from costs one reading. One pass at a time; the wakes during a pass
// are one pass after it, at least packagesGap after the last began. A rare safety pass (packagesSafetyEvery) covers a
// change no watched key says, and with an unchanged digest composes nothing either.
const (
packagesSafetyEvery = 10 * time.Minute
packagesGap = 30 * time.Second
)
// packagesTiming is when the declared-packages mirror passes: the safety pass's period and the least time between the
// starts of two passes.
type packagesTiming struct{ safety, gap time.Duration }
// packagesKeeper keeps the declared-packages mirror: read is the Input, compose the mirror's values from it, sync writes
// those that differ from the bucket, holds says this controller holds the lease. Each a seam a test stands in for.
type packagesKeeper struct {
read func(ctx context.Context) (*meshInput, error)
compose func(ctx context.Context, in *meshInput) (map[string][]byte, error)
sync func(ctx context.Context, want map[string][]byte) error
holds func() bool
wake chan struct{}
// last is the digest of the Input last mirrored whole; empty after a failure or while the lease is not held, so the
// next pass composes.
last string
}
func newPackagesKeeper(read func(context.Context) (*meshInput, error),
compose func(context.Context, *meshInput) (map[string][]byte, error),
sync func(context.Context, map[string][]byte) error, holds func() bool) *packagesKeeper {
return &packagesKeeper{read: read, compose: compose, sync: sync, holds: holds, wake: make(chan struct{}, 1)}
}
// poke says something the mirror composes from may have changed; never blocks, and pokes before a pass are one.
func (k *packagesKeeper) poke() {
select {
case k.wake <- struct{}{}:
default:
}
}
// pass brings the mirror to the Input, composing only when the Input differs from the one last mirrored.
func (k *packagesKeeper) pass(ctx context.Context) error {
if !k.holds() {
k.last = ""
return nil
}
in, err := k.read(ctx)
if err != nil {
k.last = ""
return fmt.Errorf("the store cannot be read: %w", err)
}
digest := inputDigest(in)
if digest != "" && digest == k.last {
return nil
}
want, err := k.compose(ctx, in)
if err != nil {
k.last = ""
return err
}
if err := k.sync(ctx, want); err != nil {
k.last = ""
return err
}
k.last = digest
return nil
}
// run passes once at its start, then on each wake and each safety period, until ctx ends. said hears why a pass failed,
// and an empty line when one succeeds again.
func (k *packagesKeeper) run(ctx context.Context, at packagesTiming, said func(why string)) {
safety := time.NewTicker(at.safety)
defer safety.Stop()
failing := ""
var began time.Time
for {
// The wakes so far are this pass's: drained before it reads, so a change after is a pass of its own.
select {
case <-k.wake:
default:
}
began = time.Now()
why := ""
if err := k.pass(ctx); err != nil {
why = err.Error()
}
if why != failing {
said(why)
failing = why
}
select {
case <-ctx.Done():
return
case <-k.wake:
case <-safety.C:
}
if rest := at.gap - time.Since(began); rest > 0 {
select {
case <-ctx.Done():
return
case <-time.After(rest):
}
}
}
}
// inputDigest is a digest of everything the composition reads (the Input, and why a node or the bus is refused): the same
// for the same Input. Empty when it cannot be made, and then the pass composes.
func inputDigest(in *meshInput) string {
refused := map[string]string{}
for node, err := range in.refused {
if err != nil {
refused[node] = err.Error()
}
}
bus := ""
if in.busRefused != nil {
bus = in.busRefused.Error()
}
body, err := json.Marshal(struct {
Input compose.Input
Refused map[string]string
BusRefused string
}{in.Input, refused, bus})
if err != nil {
return ""
}
sum := sha256.Sum256(body)
return hex.EncodeToString(sum[:])
}
// keepingDeclaredPackages keeps the declared-packages mirror for as long as ctx: woken by the watch of the compositions'
// digests and of the assignments. It answers the keeper, whose poke a caller may also wake it with.
func keepingDeclaredPackages(ctx context.Context, open *stores, api jetstream.JetStream, announce recordmirror.Announcer,
epoch func() uint64, holds func() bool) *packagesKeeper {
m := recordmirror.DeclaredPackages
var w *recordmirror.Writer
read := func(ctx context.Context) (*meshInput, error) {
reading, cancel := context.WithTimeout(ctx, 2*time.Minute)
defer cancel()
return readComposeInput(reading, open)
}
sync := func(ctx context.Context, want map[string][]byte) error {
writing, cancel := context.WithTimeout(ctx, 30*time.Second)
defer cancel()
if w == nil {
kv, err := api.KeyValue(writing, m.Bucket)
if err != nil {
return nil, fmt.Errorf("the packages %s declares cannot be read: %w", n.Name, err)
return fmt.Errorf("the bucket %s cannot be read: %v", m.Bucket, err)
}
w = &recordmirror.Writer{Mirror: m, KV: kv, Announce: announce, Epoch: epoch, Holds: holds}
}
lost := w.LostEvents
r, err := w.Sync(writing, want)
if r.Written+r.Removed > 0 {
fmt.Printf("the %s mirror moved: %d written, %d removed\n", m.Name, r.Written, r.Removed)
}
if w.LostEvents > lost {
fmt.Printf("the %s mirror lost %d event(s): the bucket holds the change, its notice was refused twice\n",
m.Name, w.LostEvents-lost)
}
return err
}
k := newPackagesKeeper(read, declaredPackagesFrom, sync, holds)
for _, bucket := range []string{sdkdelivery.ControllerComposedBucket, broker.AssignmentsBucket} {
goServing(ctx, func() { wakeOnBucket(ctx, api, bucket, k.poke) })
}
goServing(ctx, func() {
k.run(ctx, packagesTiming{safety: packagesSafetyEvery, gap: packagesGap}, func(why string) {
if why != "" {
fmt.Printf("the %s mirror cannot be kept: %s\n", m.Name, why)
} else {
fmt.Printf("the %s mirror is kept again\n", m.Name)
}
})
})
return k
}
// wakeOnBucket calls wake on every write or delete of a key in the bucket, for as long as ctx: only the keys' metadata is
// watched. A bucket not there yet, or a watch that ends, is watched again a minute later; the safety pass covers the gap.
func wakeOnBucket(ctx context.Context, api jetstream.JetStream, bucket string, wake func()) {
for {
if err := watchBucketOnce(ctx, api, bucket, wake); err != nil && ctx.Err() == nil {
fmt.Printf("the declared-packages mirror does not hear %s: %v\n", bucket, err)
}
select {
case <-ctx.Done():
return
case <-time.After(time.Minute):
}
}
}
// watchBucketOnce watches the bucket until its watch ends or ctx does.
func watchBucketOnce(ctx context.Context, api jetstream.JetStream, bucket string, wake func()) error {
kv, err := api.KeyValue(ctx, bucket)
if err != nil {
return err
}
watch, err := kv.WatchAll(ctx, jetstream.UpdatesOnly(), jetstream.MetaOnly())
if err != nil {
return err
}
defer func() { _ = watch.Stop() }()
for {
select {
case <-ctx.Done():
return nil
case e, ok := <-watch.Updates():
if !ok {
return fmt.Errorf("the watch ended")
}
if out[n.Name], err = inventory.DeclaredPackagesValue(n.Name, d.Resources, d.ModuleOf); err != nil {
return nil, err
if e != nil {
wake()
}
}
return out, nil
}
}
+191
View File
@@ -0,0 +1,191 @@
package main
import (
"context"
"fmt"
"sync"
"sync/atomic"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/compose"
)
// The declared-packages mirror composes only when what it composes from changed (issue 1945 note 22: it composed every node
// in full every 10 s, a pass of about 45 s, so the controller on the control node never rested at 130-175 % CPU).
// scaled is a cadence of the mirror with a second taken as a millisecond, so a test's minute is 60 ms.
func scaled() packagesTiming {
return packagesTiming{safety: packagesSafetyEvery / 1000, gap: packagesGap / 1000}
}
const scaledMinute = 60 * time.Millisecond
// fakeMesh is the keeper's input as a test changes it: each node's assigned modules, and the compositions counted by node.
type fakeMesh struct {
mu sync.Mutex
assigned map[string][]string
composed atomic.Int64 // node compositions
synced atomic.Int64
lastSync map[string][]byte
// block, when set, holds the next composition until it is closed; entered is closed when that composition began.
block chan struct{}
entered chan struct{}
}
func newFakeMesh(nodes int) *fakeMesh {
f := &fakeMesh{assigned: map[string][]string{}}
for i := 0; i < nodes; i++ {
f.assigned[fmt.Sprintf("node%d", i)] = []string{"base"}
}
return f
}
func (f *fakeMesh) assign(node, module string) {
f.mu.Lock()
defer f.mu.Unlock()
f.assigned[node] = append(append([]string(nil), f.assigned[node]...), module)
}
func (f *fakeMesh) keeper() *packagesKeeper {
read := func(context.Context) (*meshInput, error) {
f.mu.Lock()
defer f.mu.Unlock()
in := compose.Input{Catalogue: map[string]catalogue.Manifest{"base": {Module: "base", Version: "1"}}}
for _, n := range nodesOfFake(f.assigned) {
in.Nodes = append(in.Nodes, compose.Node{Name: n, Assigned: append([]string(nil), f.assigned[n]...)})
}
return &meshInput{Input: in}, nil
}
composeAll := func(_ context.Context, in *meshInput) (map[string][]byte, error) {
f.mu.Lock()
block, entered := f.block, f.entered
f.block = nil
f.mu.Unlock()
if block != nil {
close(entered)
<-block
}
out := map[string][]byte{}
for _, n := range in.Nodes {
f.composed.Add(1)
out[n.Name] = []byte(fmt.Sprint(n.Assigned))
}
return out, nil
}
syncAll := func(_ context.Context, want map[string][]byte) error {
f.mu.Lock()
f.lastSync = want
f.mu.Unlock()
f.synced.Add(1)
return nil
}
return newPackagesKeeper(read, composeAll, syncAll, func() bool { return true })
}
func nodesOfFake(m map[string][]string) []string {
out := make([]string, 0, len(m))
for k := range m {
out = append(out, k)
}
for i := range out {
for j := i + 1; j < len(out); j++ {
if out[j] < out[i] {
out[i], out[j] = out[j], out[i]
}
}
}
return out
}
func runKeeper(t *testing.T, k *packagesKeeper) context.CancelFunc {
ctx, cancel := context.WithCancel(context.Background())
done := make(chan struct{})
go func() {
defer close(done)
k.run(ctx, scaled(), func(string) {})
}()
t.Cleanup(func() {
cancel()
<-done
})
return cancel
}
func eventually(t *testing.T, what string, cond func() bool) {
t.Helper()
deadline := time.Now().Add(time.Second)
for !cond() {
if time.Now().After(deadline) {
t.Fatalf("not within a second: %s", what)
}
time.Sleep(time.Millisecond)
}
}
// 1945/AC-mirror-1: given a mesh whose input does not change, when a minute passes after the mirror's first pass, then it
// composes no node (before: a pass every 10 s, each composing every node). The count per minute is said.
func TestTheDeclaredPackagesMirrorComposesNothingWithoutAChange(t *testing.T) {
const nodes = 5
f := newFakeMesh(nodes)
runKeeper(t, f.keeper())
eventually(t, "the first pass", func() bool { return f.synced.Load() >= 1 })
start := f.composed.Load()
time.Sleep(scaledMinute)
perMinute := f.composed.Load() - start
t.Logf("node compositions in a minute of an unchanged mesh of %d nodes: %d", nodes, perMinute)
if perMinute != 0 {
t.Errorf("an unchanged mesh of %d nodes was composed %d times in a minute; want 0", nodes, perMinute)
}
}
// 1945/AC-mirror-2: given a pass that is running, when fifty changes arrive during it, then exactly one more pass follows,
// never two at once and never one per change.
func TestTheDeclaredPackagesMirrorPassesOncePerBurst(t *testing.T) {
f := newFakeMesh(1)
release := make(chan struct{})
f.block, f.entered = release, make(chan struct{})
k := f.keeper()
runKeeper(t, k)
<-f.entered
for i := 0; i < 50; i++ {
f.assign("node0", fmt.Sprintf("m%d", i))
k.poke()
}
close(release)
eventually(t, "the pass after the burst", func() bool { return f.synced.Load() >= 2 })
time.Sleep(2*scaled().gap + 20*time.Millisecond)
if got := f.synced.Load(); got != 2 {
t.Errorf("a burst of 50 changes during a pass gave %d passes in all; want 2 (the running one and one after)", got)
}
}
// 1945/AC-mirror-3: given the mirror at rest, when an input changes and its event arrives, then the change is mirrored.
func TestTheDeclaredPackagesMirrorMirrorsAChange(t *testing.T) {
f := newFakeMesh(2)
k := f.keeper()
runKeeper(t, k)
eventually(t, "the first pass", func() bool { return f.synced.Load() >= 1 })
f.assign("node1", "viewer")
k.poke()
eventually(t, "the change mirrored", func() bool {
f.mu.Lock()
defer f.mu.Unlock()
return string(f.lastSync["node1"]) == "[base viewer]"
})
}
// 1945/AC-mirror-4: the digest of the input is the same for the same input and differs when an assignment differs.
func TestTheInputDigestFollowsTheInput(t *testing.T) {
of := func(assigned ...string) *meshInput {
return &meshInput{Input: compose.Input{Nodes: []compose.Node{{Name: "anchor", Assigned: assigned}}}}
}
a, b, c := inputDigest(of("base")), inputDigest(of("base")), inputDigest(of("base", "viewer"))
if a == "" || a != b {
t.Errorf("the same input gave the digests %q and %q", a, b)
}
if a == c {
t.Errorf("an added assignment left the digest at %q", a)
}
}
+60
View File
@@ -0,0 +1,60 @@
package main
import (
"context"
"sync"
)
// **Every goroutine the serving controller starts ends before serve returns** (issue 1945). A loop or a watch started
// beside the serving runs until serve's context ends, and serve's stop waits for it before it closes the stores, the
// bus and the condition keeper those loops read: a loop that outlived its serve met the next serve's globals, its
// closed stores and its bus, and in the test process that is the next test (a race on conditionsFrom, a package that
// passed every test and still exited failing). Where it is checked: the tests that run serve whole, under -race.
// servingGroup counts the goroutines one serve started. Once its stop began it starts nothing more: what would start is
// work for a controller that is stopping.
type servingGroup struct {
mu sync.Mutex
stopped bool
wg sync.WaitGroup
}
type servingGroupKey struct{}
// withServingGroup is ctx carrying a new group, which goServing under it counts.
func withServingGroup(ctx context.Context) (context.Context, *servingGroup) {
g := &servingGroup{}
return context.WithValue(ctx, servingGroupKey{}, g), g
}
// goServing runs work beside the caller, counted in the group ctx carries, so serve's stop waits for it. Under a group
// whose stop began it runs nothing; under a context with no group (a part run alone, by a test or a verb) it is a plain
// goroutine, as before. It says whether it started work.
func goServing(ctx context.Context, work func()) bool {
g, _ := ctx.Value(servingGroupKey{}).(*servingGroup)
if g == nil {
go work()
return true
}
g.mu.Lock()
if g.stopped {
g.mu.Unlock()
return false
}
g.wg.Add(1)
g.mu.Unlock()
go func() {
defer g.wg.Done()
work()
}()
return true
}
// stop starts nothing more and waits for what was started; the caller ends their context first. A second stop waits
// for nothing.
func (g *servingGroup) stop() {
g.mu.Lock()
g.stopped = true
g.mu.Unlock()
g.wg.Wait()
}
+86
View File
@@ -0,0 +1,86 @@
package main
import (
"context"
"sync"
"sync/atomic"
"testing"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"github.com/novox/mesh-controller/internal/identity"
"github.com/novox/mesh-controller/internal/inventory"
"github.com/novox/mesh-controller/internal/licences"
)
// The serving controller's reads of the rest of the mesh open no pool per call: the licences and identity stores are the
// controller's own, opened once and shared. On 2026-10-25 at 16:08 each take-in opened two pools of its own
// (restOfTheMeshOf), about 120 at once, and novox's store refused every new connection (issues 1945 and 1277).
func TestRestOfTheMeshOfOpensNoPoolPerCall(t *testing.T) {
wasIdent, wasLic, wasRest := identityOpener, licencesOpener, restOfTheMeshOn
t.Cleanup(func() { identityOpener, licencesOpener, restOfTheMeshOn = wasIdent, wasLic, wasRest })
var identities, licencesOpened atomic.Int64
identityOpener = func(context.Context) (*identity.Identity, error) {
identities.Add(1)
return &identity.Identity{}, nil
}
licencesOpener = func(context.Context) (*licences.Licences, error) {
licencesOpened.Add(1)
return &licences.Licences{}, nil
}
// The read of the Input reaches both stores, as readComposeInput does.
restOfTheMeshOn = func(ctx context.Context, open *stores, _ map[string]catalogue.Manifest, _ string) (catalogue.World, error) {
if _, err := open.Licences(ctx); err != nil {
return catalogue.World{}, err
}
if _, err := open.Identity(ctx); err != nil {
return catalogue.World{}, err
}
return catalogue.World{}, nil
}
inv := &inventory.Inventory{}
serving := &stores{inventory: inv}
unshare := shareStores(serving)
t.Cleanup(unshare)
var wg sync.WaitGroup
for range 120 {
wg.Add(1)
go func() {
defer wg.Done()
if _, err := restOfTheMeshOf(context.Background(), inv, nil, ""); err != nil {
t.Error(err)
}
}()
}
wg.Wait()
if n := licencesOpened.Load(); n != 1 {
t.Fatalf("opened the licences store %d times for 120 reads, want once", n)
}
if n := identities.Load(); n != 1 {
t.Fatalf("opened the identity store %d times for 120 reads, want once", n)
}
if serving.licences == nil || serving.identity == nil {
t.Fatal("the reads did not use the serving controller's stores")
}
}
// A process that shares no stores (a verb run as its own short process) still opens them for the call and lets go.
func TestRestOfTheMeshOfOpensItsOwnWhereNoneAreShared(t *testing.T) {
wasIdent, wasLic, wasRest := identityOpener, licencesOpener, restOfTheMeshOn
t.Cleanup(func() { identityOpener, licencesOpener, restOfTheMeshOn = wasIdent, wasLic, wasRest })
var opened atomic.Int64
licencesOpener = func(context.Context) (*licences.Licences, error) { opened.Add(1); return &licences.Licences{}, nil }
restOfTheMeshOn = func(ctx context.Context, open *stores, _ map[string]catalogue.Manifest, _ string) (catalogue.World, error) {
_, err := open.Licences(ctx)
return catalogue.World{}, err
}
inv := &inventory.Inventory{}
for range 2 {
if _, err := restOfTheMeshOf(context.Background(), inv, nil, ""); err != nil {
t.Fatal(err)
}
}
if n := opened.Load(); n != 2 {
t.Fatalf("opened %d times, want once per call", n)
}
}
+3 -3
View File
@@ -67,16 +67,16 @@ func keepAnnouncements(ctx context.Context, js *broker.JetStream, open *stores)
return
}
if changed {
inBackground(func() { composeAnnouncedReads(context.WithoutCancel(ctx), open) })
inBackground(ctx, func() { composeAnnouncedReads(context.WithoutCancel(ctx), open) })
}
})
if err != nil {
return fmt.Errorf("following the states announced: %w", err)
}
go func() {
goServing(ctx, func() {
<-ctx.Done()
consuming.Stop()
}()
})
return nil
}
+1 -1
View File
@@ -210,7 +210,7 @@ func (l nudgingListener) Heard(ctx context.Context, report link.Report) (bool, e
// And, once the move of the secrets into the vault has started, what is left of this machine's (design 57's V3,
// issue 676): a machine asleep when it started is moved when it is next heard from. Off the report's path.
if err == nil && l.open != nil && report.Superseded == "" && report.Rekey == nil && report.Node != "" {
go moveWhenHeard(context.WithoutCancel(ctx), l.open, moveHeardOn(), report.Node)
goServing(ctx, func() { moveWhenHeard(context.WithoutCancel(ctx), l.open, moveHeardOn(), report.Node) })
}
return news, err
}
+36 -2
View File
@@ -4,6 +4,7 @@ import (
"context"
"fmt"
"os"
"sync"
"time"
"git.novox.be/novox/mesh-sdk/go/catalogue"
@@ -163,6 +164,13 @@ func openLicences(ctx context.Context) (*licences.Licences, error) {
return held, nil
}
// identityOpener and licencesOpener open a store the first time a set of stores needs it; variables so a test counts
// the pools opened.
var (
identityOpener = openIdentity
licencesOpener = openLicences
)
// stores is what one command has open.
//
// **Opened once, not once per machine.** Working out what a machine should be reaches the identity
@@ -176,6 +184,9 @@ type stores struct {
inventory *inventory.Inventory
identity *identity.Identity
licences *licences.Licences
// mu keeps one opening of each store at a time: a serving controller's stores are shared by readers that run side
// by side (sharedStores).
mu sync.Mutex
}
// open connects to the inventory, which every command that touches the mesh needs.
@@ -189,10 +200,12 @@ func openStores(ctx context.Context) (*stores, error) {
// Identity is this control plane's own identity context, opened if it has not been.
func (h *stores) Identity(ctx context.Context) (*identity.Identity, error) {
h.mu.Lock()
defer h.mu.Unlock()
if h.identity != nil {
return h.identity, nil
}
opened, err := openIdentity(ctx)
opened, err := identityOpener(ctx)
if err != nil {
return nil, err
}
@@ -202,10 +215,12 @@ func (h *stores) Identity(ctx context.Context) (*identity.Identity, error) {
// Licences is the context holding model access, opened if it has not been.
func (h *stores) Licences(ctx context.Context) (*licences.Licences, error) {
h.mu.Lock()
defer h.mu.Unlock()
if h.licences != nil {
return h.licences, nil
}
opened, err := openLicences(ctx)
opened, err := licencesOpener(ctx)
if err != nil {
return nil, err
}
@@ -226,3 +241,22 @@ func (h *stores) Close() {
h.inventory.Close()
}
}
// sharedStores are the stores a serving controller lends, by its inventory, to every reader in the process that holds
// only the inventory (restOfTheMeshOf): one pool of each store for the controller, never one per call (issues 1945 and
// 1277, 2026-10-25 16:08: about 120 take-ins opened two pools each and filled novox's store).
var sharedStores sync.Map // *inventory.Inventory -> *stores
// shareStores lends open by its inventory until the function returned is called.
func shareStores(open *stores) func() {
sharedStores.Store(open.inventory, open)
return func() { sharedStores.CompareAndDelete(open.inventory, open) }
}
// sharedStoresOf are the stores lent for inv, if any are.
func sharedStoresOf(inv *inventory.Inventory) (*stores, bool) {
if open, ok := sharedStores.Load(inv); ok {
return open.(*stores), true
}
return nil, false
}
+29 -12
View File
@@ -512,7 +512,9 @@ func (w *watchdogs) lostConsumers(ctx context.Context, heard []link.Advisory) (m
for _, a := range deleted {
_, err := w.js.Context().ConsumerInfo(a.Stream, a.Consumer)
switch {
case errors.Is(err, nats.ErrConsumerNotFound):
// A consumer whose stream is gone is gone with it: the retired declarations' stream took every node's
// consumer, and its deletion said each as deleted (issue 1945).
case errors.Is(err, nats.ErrConsumerNotFound) || errors.Is(err, nats.ErrStreamNotFound):
out[a.Stream+"."+a.Consumer] = want[a.Stream+"."+a.Consumer]
case err != nil:
return nil, fmt.Errorf("whether %s exists cannot be read: %w", link.ConsumerInWords(a.Stream, a.Consumer), err)
@@ -543,10 +545,9 @@ func later(a, b time.Time) time.Time {
return b
}
// watchTheMesh opens the condition store and starts the watchdogs, the bus's advisories and the
// self-check, for the serving controller; the returned function stops them. Nil when the store could
// not be opened, which is said.
func watchTheMesh(ctx context.Context, open *stores, server *link.Server, bus link.OverNATS) func() {
// openServingConditions opens the condition store for the serving controller and makes it conditionsFrom. Nil when the store
// could not be opened, which is said: status then says the conditions cannot be read.
func openServingConditions(ctx context.Context, open *stores, bus link.OverNATS) *conditions.Keeper {
keeper, err := keeperOn(ctx, bus.Conn, storeLooks{open.inventory})
if err != nil {
fmt.Printf("the condition store could NOT be opened, so nothing that goes wrong is kept or said, and "+
@@ -554,6 +555,22 @@ func watchTheMesh(ctx context.Context, open *stores, server *link.Server, bus li
return nil
}
conditionsFrom = keeper
return keeper
}
// closeConditions flushes what the keeper holds and closes it, once nothing that raises a condition runs.
func closeConditions(keeper *conditions.Keeper) {
flushing, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
keeper.Close(flushing)
}
// watchTheMesh starts the watchdogs, the bus's advisories and the self-check over the condition store's keeper, for the
// serving controller; the returned function stops them and waits for them to end. Nil when there is no keeper.
func watchTheMesh(ctx context.Context, open *stores, server *link.Server, bus link.OverNATS, keeper *conditions.Keeper) func() {
if keeper == nil {
return nil
}
logf := func(format string, args ...any) { fmt.Printf(format+"\n", args...) }
stopHearing, err := server.HearAdvisories(logf)
if err != nil {
@@ -566,26 +583,26 @@ func watchTheMesh(ctx context.Context, open *stores, server *link.Server, bus li
d := &doctor{open: open, js: server.JetStream(), keeper: keeper, teller: bus, watchdogs: w, host: host}
w.doctor = d
doctorFrom = d
// Its own group as well as serve's: its stop waits for what it started.
watching, stop := context.WithCancel(ctx)
go w.keep(watching)
go d.keep(watching)
watching, running := withServingGroup(watching)
goServing(watching, func() { w.keep(watching) })
goServing(watching, func() { d.keep(watching) })
// And the healers (novox/hq to-be 45 §7): what is open that a registered healer answers, repaired
// under the lease and the brake, every act said.
healers := newHealing(open, keeper, bus, server.JetStream())
go healers.keep(watching)
goServing(watching, func() { healers.keep(watching) })
// And the asker (novox/hq ADR 0259): what needs the operator and names its answers is asked of them,
// and the answer chosen is performed on its warrant.
startAsking(watching, open, server, bus.Conn, keeper)
go forgettingOldHeals(watching, open.inventory)
goServing(watching, func() { forgettingOldHeals(watching, open.inventory) })
fmt.Printf("watching the mesh: %d signal(s) every %s, %d probe(s) every %s; what is wrong is kept in %s "+
"and said as %s events\n", len(watchedRows()), watchEvery, len(runnableProbes()), doctorEvery,
broker.ConditionsBucket, conditions.Seat)
return func() {
stop()
running.stop()
stopHearing()
flushing, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
keeper.Close(flushing)
}
}
+2 -1
View File
@@ -165,7 +165,8 @@ func HandPathWrites(nodes []string) ([]string, error) {
// DeclarationsStream is the stream of each node's last declaration, `mesh.node.<node>.declare` (issue 107). Release 6
// sends none (ADR 0539 clauses 5 and 11): the stream is kept, with each node's consumer, only so a node away during the
// switch finds its last declaration and with it the new node-engine; the controller deletes it once every node's have
// names the new node-engine, and from then on neither asserts it nor any node's consumer on it.
// names the new node-engine, and from then on neither asserts it nor any node's consumer on it; nor makes it again once
// it is gone from the bus for any reason, since a stream made anew would hold no node's last declaration (issue 1945).
const DeclarationsStream = "NODES"
// RaiseAfterDeclarations is Raise once the declarations' stream is retired: the mesh's streams and consumers but the
+31
View File
@@ -114,6 +114,37 @@ func (i *Inventory) writeBusUser(ctx context.Context, u BusUser, hash string) er
return nil
}
// MintBusPasswordIfAbsent mints u's password only when no row names u yet, in one statement: of two callers that both
// found the account missing, exactly one wins, and the other is told it lost (won false, no password) and must seal
// nothing, so the hash the bus holds is always the password the node is sealed (novox/hq ADR 0543, the review of
// issue 1945: the assign act and the assignments' watch once each minted, the later hash beside the earlier seal).
func (i *Inventory) MintBusPasswordIfAbsent(ctx context.Context, u BusUser) (string, bool, error) {
if u.Username == "" || u.Kind == "" {
return "", false, errors.New("a bus user needs a username and a kind")
}
raw := make([]byte, 32)
if _, err := rand.Read(raw); err != nil {
return "", false, fmt.Errorf("cannot generate a bus password: %w", err)
}
password := base64.RawURLEncoding.EncodeToString(raw)
hash, err := hashBusPassword(password)
if err != nil {
return "", false, fmt.Errorf("cannot hash a bus password: %w", err)
}
tag, err := i.store.Pool().Exec(ctx,
`insert into bus_user (username, kind, node, module, password_hash)
values ($1, $2, $3, $4, $5)
on conflict (username) do nothing`,
u.Username, u.Kind, u.Node, u.Module, string(hash))
if err != nil {
return "", false, fmt.Errorf("cannot record the bus user %s: %w", u.Username, err)
}
if tag.RowsAffected() == 0 {
return "", false, nil
}
return password, true, nil
}
func (i *Inventory) MintBusPassword(ctx context.Context, u BusUser) (string, error) {
if u.Username == "" || u.Kind == "" {
return "", errors.New("a bus user needs a username and a kind")
@@ -33,7 +33,7 @@ func TestABusPasswordIsReturnedOnceAndOnlyItsHashIsKept(t *testing.T) {
}
}
func TestMintingAgainRotatesRatherThanAddsAUser(t *testing.T) {
func TestMintingAgainRotatesAndMintingIfAbsentHasOneWinner(t *testing.T) {
inv := ForTest(t)
ctx := context.Background()
@@ -62,6 +62,50 @@ func TestMintingAgainRotatesRatherThanAddsAUser(t *testing.T) {
if bcrypt.CompareHashAndPassword([]byte(hash), []byte(first)) == nil {
t.Fatal("the previous password still verifies, so a rotation revoked nothing")
}
// **Minting only if absent never rotates, and of several at once exactly one wins** (ADR 0543, the review of issue
// 1945): the assign act and the assignments' watch may both find an account missing; the winner's password is the
// one the store's hash verifies, so the one it seals is the one the bus accepts, and a loser seals nothing.
if _, won, err := inv.MintBusPasswordIfAbsent(ctx, aBusUser("shop")); err != nil || won {
t.Fatalf("a mint-if-absent over a minted account won %v (%v): it would replace a sealed password", won, err)
}
const racers = 8
type minted struct {
password string
won bool
err error
}
results := make(chan minted, racers)
for range racers {
go func() {
p, won, err := inv.MintBusPasswordIfAbsent(ctx, aBusUser("racer"))
results <- minted{p, won, err}
}()
}
var winner string
wins := 0
for range racers {
r := <-results
if r.err != nil {
t.Fatal(r.err)
}
if r.won {
wins++
winner = r.password
} else if r.password != "" {
t.Fatal("a caller that lost was handed a password to seal")
}
}
if wins != 1 {
t.Fatalf("%d of %d concurrent mints won, want exactly one", wins, racers)
}
stored, _, err := inv.BusUserHash(ctx, "one.racer")
if err != nil {
t.Fatal(err)
}
if err := bcrypt.CompareHashAndPassword([]byte(stored), []byte(winner)); err != nil {
t.Fatal("the stored hash is not the winner's password: the node would be sealed a password the bus refuses")
}
}
func TestForgettingANodeTakesItsBusUsersWithIt(t *testing.T) {
+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))
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)
}
+82
View File
@@ -0,0 +1,82 @@
package store
import (
"context"
"strings"
"testing"
"time"
"github.com/jackc/pgx/v5/pgconn"
)
// A full store (SQLSTATE 53300, "too many clients already") is not asked again every 250 ms: Ready backs off,
// doubling its wait up to a bound, and says once that the store is full (issues 1945 and 1277, 2026-10-25 16:08: the
// controller's retries kept novox's store full).
func TestReadyBacksOffOnAFullStore(t *testing.T) {
wasPing, wasWait, wasSay := pingStore, readyWait, readySay
t.Cleanup(func() { pingStore, readyWait, readySay = wasPing, wasWait, wasSay })
asked := 0
pingStore = func(context.Context, *Store) error {
asked++
if asked <= 8 {
return &pgconn.PgError{Code: "53300", Message: "sorry, too many clients already"}
}
return nil
}
var waits []time.Duration
readyWait = func(_ context.Context, d time.Duration) error { waits = append(waits, d); return nil }
var said []string
readySay = func(line string) { said = append(said, line) }
if err := (&Store{context: "example"}).Ready(context.Background(), time.Hour); err != nil {
t.Fatalf("Ready: %v", err)
}
if len(waits) != 8 {
t.Fatalf("waited %d times, want 8: %v", len(waits), waits)
}
for i := 1; i < len(waits); i++ {
if waits[i] < waits[i-1] {
t.Fatalf("wait %d (%s) is shorter than the one before (%s): %v", i, waits[i], waits[i-1], waits)
}
}
if waits[1] <= waits[0] {
t.Fatalf("Ready does not back off on a full store: %v", waits)
}
if most := waits[len(waits)-1]; most > fullStoreMostWait {
t.Fatalf("the wait grew past its bound %s: %v", fullStoreMostWait, waits)
}
full := 0
for _, l := range said {
if strings.Contains(l, "full") {
full++
}
}
if full != 1 {
t.Fatalf("said that the store is full %d times, want once: %q", full, said)
}
}
// A store that does not answer for another reason (not up yet) is still asked again every 250 ms.
func TestReadyWaitsShortlyOnAStoreNotUp(t *testing.T) {
wasPing, wasWait, wasSay := pingStore, readyWait, readySay
t.Cleanup(func() { pingStore, readyWait, readySay = wasPing, wasWait, wasSay })
asked := 0
pingStore = func(context.Context, *Store) error {
asked++
if asked <= 3 {
return context.DeadlineExceeded
}
return nil
}
var waits []time.Duration
readyWait = func(_ context.Context, d time.Duration) error { waits = append(waits, d); return nil }
readySay = func(string) {}
if err := (&Store{context: "example"}).Ready(context.Background(), time.Hour); err != nil {
t.Fatalf("Ready: %v", err)
}
for _, w := range waits {
if w != 250*time.Millisecond {
t.Fatalf("waits %v, want 250ms each", waits)
}
}
}
+47 -6
View File
@@ -24,6 +24,7 @@ import (
"strings"
"time"
"github.com/jackc/pgx/v5/pgconn"
"github.com/jackc/pgx/v5/pgxpool"
"github.com/novox/mesh-controller/internal/envfile"
@@ -213,7 +214,7 @@ func (s *Store) Pool() *pgxpool.Pool { return s.pool }
// Close releases the connections.
func (s *Store) Close() {
if s.pool != nil {
if s != nil && s.pool != nil {
s.pool.Close()
}
}
@@ -227,8 +228,10 @@ func (s *Store) Close() {
func (s *Store) Ready(ctx context.Context, within time.Duration) error {
deadline := time.Now().Add(within)
var last error
wait := 250 * time.Millisecond
saidFull := false
for {
err := s.pool.Ping(ctx)
err := pingStore(ctx, s)
if err == nil {
return nil
}
@@ -240,10 +243,48 @@ func (s *Store) Ready(ctx context.Context, within time.Duration) error {
return fmt.Errorf(
"the %s store did not answer within %s: %w", s.context, within, last)
}
select {
case <-ctx.Done():
return errors.Join(ctx.Err(), last)
case <-time.After(250 * time.Millisecond):
// A full store (SQLSTATE 53300) is not asked again every 250 ms: each ask is one more connection it refuses
// while it is full. The wait doubles up to fullStoreMostWait, and the journal says once why (issues 1945 and
// 1277, 2026-10-25 16:08: the controller's own retries kept the store full).
if isFull(err) {
if !saidFull {
readySay(fmt.Sprintf("the %s store is full (too many clients already); asked again at a growing "+
"interval, at most every %s, for at most %s in all", s.context, fullStoreMostWait, within))
saidFull = true
}
} else {
wait = 250 * time.Millisecond
}
if err := readyWait(ctx, wait); err != nil {
return errors.Join(err, last)
}
if isFull(err) {
wait = min(wait*2, fullStoreMostWait)
}
}
}
// pingStore asks the store once whether it answers; a variable so a test needs no database.
var pingStore = func(ctx context.Context, s *Store) error { return s.pool.Ping(ctx) }
// readyWait waits d or until ctx ends; a variable so a test waits in no real time.
var readyWait = func(ctx context.Context, d time.Duration) error {
select {
case <-ctx.Done():
return ctx.Err()
case <-time.After(d):
return nil
}
}
// readySay says a line of Ready's wait in the journal; a variable so a test reads it.
// isFull says whether err is the store refusing a connection because it is full (SQLSTATE 53300).
func isFull(err error) bool {
var pg *pgconn.PgError
return errors.As(err, &pg) && pg.Code == "53300"
}
// fullStoreMostWait bounds the wait between two asks of a full store.
const fullStoreMostWait = 5 * time.Second
var readySay = func(line string) { fmt.Fprintln(os.Stderr, line) }