Compare commits

..
Author SHA1 Message Date
jschoubben 7cd4d18bd5 Merge pull request 'A bus account counts as issued only once sealed, stale accounts are forgotten, and grants name sealed accounts only (issue 1945 note 35)' (#467) from fix/1945-credential-when-row-without-seal into main
Reviewed-on: #467
2026-10-25 19:31:42 +00:00
jschoubben fbf14b68db Forget the accounts of modules no longer on their nodes at the start's look, and grant sealed accounts alone (issue 1945, ADR 0543, review)
mesh/check checking fbf14b68dbab on 40034be479f8
mesh/delivery held for a person: merged without a passing check: only a person decides what follows
mesh/delivery-group group fix/1945-credential-when-row-without-seal held: a member is held: novox/mesh-controller@fbf14b68dbab (merged without a passing check:…
Migration 0108 marks every standing account sealed, ace's stale mesh-build-go
rows from 17:19 among them, and the forgetting ran only at a later unassign:
the next assignment to ace would have issued nothing again.

- The start's look first forgets, from each node's own resolution (assigned
  and required), the accounts of every module that no longer resolves on its
  node, and says each; only from a reading that holds assignments, and a node
  whose resolution is unknown is left alone.
- Unassign forgets a module's accounts only once it no longer resolves on the
  node (a module another requires keeps running), and never the node's own
  provisioner account; the forgetting moved from the inventory to the act.
- Forgetting holds each account's lock across its delete, so an issue in flight
  never writes a row back after it.
- A module whose accounts are not all sealed is granted its sealed accounts
  alone, and its grant is dropped when none is: an older grant naming ace no
  longer lets the assigner want it.
- An issue's locked section has a deadline of two minutes, since it holds a
  connection of the store's pool across the vault and the bus.
Tests: cases in the existing tables.
Research 072: the grants piece (memberships and credentials heard from placement).
2026-10-25 20:27:30 +01:00
jschoubben 79ac62ffbc Issue an account whose credential is not sealed, and forget an unassigned module's accounts (issue 1945, ADR 0543)
Met a second time live on 2026-10-25 at 20:04 (note 35): mesh-build-go was
assigned to ace again, and its account rows, minted at 17:19, had outlived the
unassignment. The issue minted only when no row stood, so it minted and sealed
nothing; the node had no credential and the bus no user, and the module
crash-looped 20 times. The watch still said "the grants written".

- A bus account now says when its credential was sealed to its node (migration
  0108; every account standing before is taken as sealed). An issue under the
  assign act, the assignments' watch or a registration mints only an account
  that is not issued, a row with no seal counting as none, under the store's
  lock on the account across its look, mint and seal, and marks it sealed once
  sealed. A seal that fails leaves it unsealed for the next issue.
- Unassigning a module forgets its bus accounts and their credentials on that
  node, so an assignment again issues them afresh.
- Only sealed accounts count as minted for a composition and are named in a
  grant, so the assigner, which now wants a node only once the grant names its
  account (mesh-delivery, same branch), holds the node until it is.
- The assignment's log line says which accounts were issued and sealed, or that
  they stood issued and sealed.

The process-only issue drops the atomic insert of the review of 9ff47f6a for
the lock, which also covers the seal. Tests: cases in the existing tables.
Research 072: the grants piece (memberships and credentials heard from placement).
2026-10-25 20:17:46 +01:00
jschoubben 40034be479 Merge pull request 'Kept compositions follow config changes from a durable consumer, so going back to an earlier build keeps today's config (issue 1945, ADR 0545)' (#466) from fix/1945-kept-compositions-follow-config into main
Reviewed-on: #466
2026-10-25 18:37:57 +00:00
jschoubben 045f25bd13 Merge pull request 'Take new builds in from a durable queue on the bus, register on the fingerprint, and issue a node's bus accounts when its assignment stands (issue 1945, ADR 0543, ADR 0544)' (#465) from fix/1945-take-in-consumer into main
Reviewed-on: #465
2026-10-25 18:06:47 +00:00
jschoubben ff3cc47027 Let the kept compositions' pass end at the stop (issue 1945, ADR 0545, review 3)
mesh/check failed: integration: github.com/novox/mesh-controller/cmd/mesh-controller.TestTheDeclarationsStreamGoesOnceEveryNodeRunsTheNewEngine: with …
mesh/delivery held for a person: merged without a passing check: only a person decides what follows
mesh/delivery-group group fix/1945-kept-compositions-follow-config held: a member is held: novox/mesh-controller@ff3cc47027e0 (merged without a passing check: …
followKept went on composing every kept key, and the config consumer every
touched node, after the controller was told to stop. Both now check the stop
before each key and each node; the writes of an unfinished batch are refused
at once, so the next controller takes them.
Research 072: milestone 2, the composer reacting to the change of an input.
2026-10-25 18:43:38 +01:00
jschoubben 31a2bcdb3c Read each resolved config key once when the consumer is made, so values set before this release reach the kept compositions (issue 1945, ADR 0545, review 3)
The config consumer delivered new writes only, and mesh-config never writes an
unchanged value again: g14's mesh-build-go capacity of 2, set before this
release, would never reach its kept compositions carrying 0. A newly made
durable now reads each key's last value once, through the same gather,
per-write lock and acknowledgement; its deliver policy is set before its first
delivery. Acknowledgements are synchronous and sent past the stop, and one the
bus refuses is said by its key.
Research 072: milestone 2, the composer reacting to the change of an input.
2026-10-25 18:38:45 +01:00
jschoubben 2df47d7b1c Merge branch 'fix/1945-take-in-consumer' into fix/1945-kept-compositions-follow-config 2026-10-25 18:30:53 +01:00
jschoubben 666d1b682e Register on the fingerprint, not the commit; count the take-in's attempts as settled (issue 1945, ADR 0544, review 3)
mesh/check checking 666d1b682e7f on 045f25bd1353
mesh/delivery held for a person: merged without a passing check: only a person decides what follows
registeredAt compared commits, so a catalogue merge that moved only the commit
(2df05030 at 18:02, a85d9037 before it) took every catalogue module in again
and composed it: about 200 take-ins, some 25 minutes, on every mesh-catalog
merge, which held nextcloud for an hour. A module whose trunk fingerprint is
the one it is registered at now has the new commit recorded (built_from and
source_head, as a registration moves them) and composes nothing unless its
composition or grant is missing. Readers of the registered commit: the
fingerprints read the commit's build or, with none, the indexer's value at it,
which names the same fingerprint; status, modules, the board and the facts
read it as the source's commit, which it is; the upgrade path counts it as
built from that commit, which it is.

The stall watch counted only settles the bus accepted, so each refused
acknowledgement at a first start left work held for good: it now counts the
loop's every attempt. The acknowledgement and refusals at the stop are sent
past the stop, and the connection closes only after the readers and the loop
have ended.
Research 072: the take-ins on a bounded pull consumer.
2026-10-25 18:30:07 +01:00
jschoubben e40b1e70db Follow config from a durable consumer, compose again only what changed since, and hold the node only for a write (issue 1945, ADR 0545, review)
The start composed every kept composition of every module on every node, each a
whole node composed in memory under the node's lock: 10 to 50 minutes at a
start, holding take-ins back. Now the writes of mesh-config_resolved are read
from the durable consumer mesh-controller-config-follow, acknowledged once the
change is composed, so a restart composes again only what was written since
the last change handled, and a first start nothing it did not hear. The node's
lock is taken for each write alone. A consumer that cannot be made or read
raises config-not-followed and is tried again a minute later, cleared once it
reads. The controller's account is granted that consumer's acknowledgements
where mesh-config serves, as its verb is.
Research 072: milestone 2, the composer reacting to the change of an input.
2026-10-25 18:27:43 +01:00
jschoubben 6b116586fc Merge branch 'fix/1945-take-in-consumer' into fix/1945-kept-compositions-follow-config 2026-10-25 18:21:58 +01:00
jschoubben bf20d9d8cf Judge the take-in's stall by keys settled with work to do, and clear its alarms after a restart (issue 1945, ADR 0544, review 2)
A refused key stays unacknowledged during its delay, so one waiting key on a
quiet mesh raised take-in-stalled, cleared at the next ack and raised again.
The watch now counts the keys the loop settled (acknowledged or refused, as
the bus took them) and calls it a stall only with work to do, keys not
delivered yet or a batch handed and not settled, and nothing settled for 5
minutes: a hung loop stays loud. Its first look always says what it found, so
a stall raised before a restart is cleared once this controller moves. A
consumer not made raises take-in-not-consumed under its own id per bucket,
cleared when that consumer is made. Acks are synchronous, so only those the
bus took count; on stop the iterator is drained and what it held, and a batch
cut by the stop, are refused with no delay.
Research 072: the take-ins on a bounded pull consumer.
2026-10-25 18:21:52 +01:00
jschoubben 81e335e42d Say why a node with no compositions answers none to compose again (ADR 0545, the empty-on-error rule) 2026-10-25 18:15:22 +01:00
jschoubben 2407bfb9ff Compose kept compositions again when a module's config changes (issue 1945, ADR 0545)
The controller composed a module only at the fingerprint it is registered at,
so each of the up to five compositions kept for going back carried the config
of the moment it was made: a want back to an earlier mesh-build-go on g14 put
its capacity back to 0. Nothing composed a node when mesh-config's resolved
values changed, either.

Now a write of mesh-config_resolved (<node>.<module>.<key>), heard from that
bucket's watch and gathered for a second, drops the controller's 5 s reading
of the values, composes the node, and composes every kept fingerprint of the
touched modules on that node again from its own build's manifest with today's
inputs, writing only the compositions whose digest differs. A starting
controller does it once for every node, for a change no controller heard.
The two replaced-controller tests are one table test with the kept cases.
Research 072: milestone 2, the composer reacting to the change of an input.
2026-10-25 18:14:03 +01:00
jschoubben 850f4b4d26 Keep refused keys from holding the take-in's slots, and judge its stall by acknowledgements (issue 1945, ADR 0544, review)
A refused key stays unacknowledged until its delay ends, so with 8 a consumer
a handful refused for long (a manifest word this controller does not read
yet, a node that keeps failing) stopped that consumer for good, the build that
reads the word among what it held back. Each consumer now leaves up to 1000
unacknowledged; the store's bound is the one loop's batch of 16. A refusal
waits 30 s, doubling each time up to 10 minutes.

The stall watch judged movement by deliveries, which rise with every
redelivery: it now counts the keys this process acknowledged. A consumer that
cannot be made raises take-in-stalled instead of a log line. On stop the
gathered batch and a key in hand are refused with no delay, so the next
controller has them at once. A trunk value with no fingerprint of the new form
is said once, as every wait for a write.
Research 072: the take-ins on a bounded pull consumer.
2026-10-25 18:10:35 +01:00
jschoubben e7673e3bd3 Merge branch 'fix/1945-credential-on-assignment' into fix/1945-take-in-consumer 2026-10-25 18:07:30 +01:00
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 94bb74966f Bound the take-in's two consumers to 16 keys in flight together, 8 each (ADR 0544 §1, its review)
Two consumers of 16 each allowed 32 take-ins on the store at once, twice
issue 1945's bound. A batch takes up to 16. Research 072: the take-ins on a
bounded pull consumer.
2026-10-25 18:01:18 +01:00
jschoubben bfb271a91b Take done keys in from a durable pull consumer, a batch at a time (issue 1945, ADR 0544)
The take-in watched the builder's and the indexer's buckets, started a follower
per module that retried on a 30 s timer, and at start took every catalogued
module in alphabetically: on 2026-10-25 a restart took in about 2 modules a
minute, about 190 followers queued behind 16 slots, and every take-in composed
two or three times even with nothing new, so a new build waited 30 minutes and
more for its credential.

Now two durable consumers named mesh-controller-take-in read the indexer's
modules bucket (every module's last trunk value when the consumer is made, then
each write) and the builder's builds bucket (each write), at most 16 keys
unacknowledged each. The bus keeps what is not taken in: the start's look is the
consumers' own catch-up, and a restart costs only what was left unacknowledged.
A batch reads the catalogue once, takes each module in once, and asks the
composer twice: once for its modules' nodes, once for the bus's node with the
grants. A module registered at its trunk whose compositions and grant exist, or
that this process took in at the same trunk, composes nothing; a key delivered
again is taken in whole. A wait for the builder's or indexer's next write is
acknowledged, a wait on a read or a failure refused with a 30 s delay. The
backlog is in status, and a backlog standing still for 5 minutes raises
take-in-stalled. The controller's account is granted the two consumers' acks.

The follower tests go with the followers: twelve tests become three.
Research 072: the take-ins on a bounded pull consumer (milestone 4, done keys
alone) and the start's look composing only what changed (milestone 2).
2026-10-25 17:59:20 +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/delivery held for a person: merged without a passing check: only a person decides what follows
mesh/check pass 8e02c4726eb4 on c3db00ceac25 | unit 981 run, 171 kept, 2 skipped | integration 1641 run, 232 kept, 7 skipped | over bound: integration…
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
45 changed files with 3329 additions and 843 deletions
+20 -2
View File
@@ -292,6 +292,22 @@ func unassign(ctx context.Context, open *stores, node string, modules ...string)
answer += "\n " + strings.TrimSpace(line) answer += "\n " + strings.TrimSpace(line)
} }
} }
// Its bus accounts on the node go with it once it no longer resolves there (a module another requires keeps running
// and keeps them), so an assignment again issues them afresh (issue 1945). The node's own provisioner account stays.
if after.err == nil {
for _, module := range modules {
if slices.Contains(after.modules, module) {
continue
}
forgot, err := open.inventory.ForgetBusAccountsOf(ctx, node, module, false)
switch {
case err != nil:
answer += fmt.Sprintf("\n its bus accounts on %s are not forgotten: %v", node, err)
case len(forgot) > 0:
answer += fmt.Sprintf("\n its bus accounts on %s forgotten: %s", node, strings.Join(forgot, ", "))
}
}
}
// What it leaves behind that is irreplaceable is kept and retired, never removed (novox/hq ADR 0233). // What it leaves behind that is irreplaceable is kept and retired, never removed (novox/hq ADR 0233).
for _, line := range keptOnUnassign(ctx, open.inventory, node, modules) { for _, line := range keptOnUnassign(ctx, open.inventory, node, modules) {
answer += "\n " + line answer += "\n " + line
@@ -399,6 +415,8 @@ func blockedElsewhere(ctx context.Context, open *stores, except string) string {
// purpose, never by re-assigning); and when the bus cannot be reached from here, the line names the // purpose, never by re-assigning); and when the bus cannot be reached from here, the line names the
// verb and the push that would refuse the module until it is run — never a silent placeholder (novox/hq issue 203). // verb and the push that would refuse the module until it is run — never a silent placeholder (novox/hq issue 203).
func issueOnAssign(ctx context.Context, open *stores, node, module string) string { func issueOnAssign(ctx context.Context, open *stores, node, module string) string {
// Mint only what no row names: the assignments' watch hears the same write and may issue at once (ADR 0543).
ctx = onlyIfAbsent(ctx)
inv := open.inventory inv := open.inventory
shelf, err := inv.Catalogue(ctx) shelf, err := inv.Catalogue(ctx)
if err != nil { if err != nil {
@@ -419,9 +437,9 @@ func issueOnAssign(ctx context.Context, open *stores, node, module string) strin
// A module running the provisioner client alone is minted the client's account (issue 917). // A module running the provisioner client alone is minted the client's account (issue 917).
user = broker.Principal{Kind: broker.KindProvisioner, Node: node, Module: module}.Username() user = broker.Principal{Kind: broker.KindProvisioner, Node: node, Module: module}.Username()
} }
if _, minted, err := inv.BusUserHash(ctx, user); err != nil { if _, sealed, err := inv.BusAccountIssued(ctx, user); err != nil {
return "" return ""
} else if minted { } else if sealed { // a row whose credential is not sealed to the node is not issued (issue 1945)
// Its own account stands and is kept; a process it gained is not issued yet (issue 1226). // Its own account stands and is kept; a process it gained is not issued yet (issue 1226).
return processAccountsOnAssign(ctx, inv, m, node) return processAccountsOnAssign(ctx, inv, m, node)
} }
+489
View File
@@ -0,0 +1,489 @@
package main
import (
"context"
"encoding/json"
"errors"
"fmt"
"maps"
"slices"
"strings"
"sync"
"time"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"github.com/novox/mesh-controller/internal/broker"
"github.com/novox/mesh-controller/internal/chain"
"github.com/novox/mesh-controller/internal/inventory"
)
// An assignment that stands issues its node's bus accounts and the module's grant (novox/hq ADR 0543, issue 1945; a
// step of research 072's grants: the memberships and credentials heard from placement, not from a send or a build).
//
// ADR 0539 clause 7 issues a module's credential and grant when its build is done. A module already built and assigned
// to one more node never has a build done again, so nothing issued that node's account: on 2026-10-25 at 17:19
// mesh-build-node and mesh-build-go were assigned to ace, composed for it, and the bus's user list never named
// ace.mesh-build-node nor ace.mesh-build-go; one runtime found no credential, the other was refused by the bus, both
// restarting every 5 s. The assign act composes the assigned node alone (composedAfterAct): never the node holding the
// bus, never the grant.
//
// So the write of an assignments key is the event. When a key `<node>.<module>` of the assignments bucket comes to stand
// (a new key, or one that was ended or pending), the lease holder issues the module's accounts on that node that are not
// minted (idempotent: a minted account is never issued again, a rotation is on purpose), composes the node holding the
// bus so its user list carries them, and writes the grants. A rewrite of an assignment that already stood (the
// provisioner's plan kept in its value) does nothing. A starting controller looks at every standing assignment once,
// from one reading of the catalogue: it issues what is missing, then composes the bus's node and writes the grants
// once. The assign act issues too; both mint only an account no row names, in one statement, so exactly one of them
// mints and seals it (mintBusUser, the review of issue 1945).
// assignmentAccountSteps are what an assignment that comes to stand does, each a step a test may stand in for.
type assignmentAccountSteps struct {
// holds says this controller holds the lease: only its holder issues.
holds func(ctx context.Context) bool
// forgetStale forgets every module's accounts on a node where the module no longer resolves, and answers a line for
// each (issue 1945: rows of an unassignment before this release, marked sealed by migration 0108).
forgetStale func(ctx context.Context) ([]string, error)
// catalogue reads the catalogue once, for every assignment a look issues for.
catalogue func(ctx context.Context) (map[string]catalogue.Manifest, error)
// issue issues the module's accounts on the node that are not minted, from the catalogue read; issued says it
// minted one, line what it did.
issue func(ctx context.Context, shelf map[string]catalogue.Manifest, node, module string) (issued bool, line string, err error)
// busHolder is the node holding the bus, "" for none.
busHolder func(ctx context.Context) (string, error)
// compose writes the nodes' compositions; grants writes the grants issued.
compose func(ctx context.Context, nodes []string) error
grants func(ctx context.Context) error
}
// settle issues what the standing assignment of module on node lacks. fresh says the assignment just came to stand: the
// bus's node is composed and the grants written even when every account was minted before (a module assigned again
// keeps its account, and the bus's user list left it out while it was unassigned). It answers what it did, "" for
// nothing.
func (s assignmentAccountSteps) settle(ctx context.Context, node, module string) (string, error) {
if module == controllerModule {
// The controller's own module is issued at the terminal alone: its `broker` secret is the credential it is
// using (rollout.go), and it is assigned and unassigned there only (refusedMovingTheController).
return "", nil
}
if !s.holds(ctx) {
return "", errNotHolding
}
shelf, err := s.catalogue(ctx)
if err != nil {
return "", fmt.Errorf("the catalogue cannot be read: %w", err)
}
_, line, err := s.issue(ctx, shelf, node, module)
if err != nil {
return "", fmt.Errorf("its bus accounts on %s are not issued: %w", node, err)
}
said, err := s.cover(ctx)
prefix := module + " on " + node + ": "
if line != "" {
prefix += line + "; "
}
return prefix + said, err
}
// look is the start's look: the accounts every standing assignment lacks, issued one after another from one reading of
// the catalogue, then the bus's node composed and the grants written once, when it looked at any assignment, whatever
// it issued (an assignment that came to stand while no controller heard it may have its accounts and still lack its
// place in the bus's user list or its grant). It answers what it did.
func (s assignmentAccountSteps) look(ctx context.Context, standing []chain.Assignment) (string, error) {
var forgot []string
if len(standing) > 0 && s.forgetStale != nil {
// Only from a reading of the assignments that holds some: an empty or failed one would forget every account.
if !s.holds(ctx) {
return "", errNotHolding
}
var err error
if forgot, err = s.forgetStale(ctx); err != nil {
return "", fmt.Errorf("the accounts of modules no longer on their nodes are not forgotten: %w", err)
}
}
var looked []chain.Assignment
for _, a := range standing {
if a.Module != controllerModule {
looked = append(looked, a)
}
}
if len(looked) == 0 {
return strings.Join(forgot, "\n"), nil
}
if !s.holds(ctx) {
return "", errNotHolding
}
shelf, err := s.catalogue(ctx)
if err != nil {
return "", fmt.Errorf("the catalogue cannot be read: %w", err)
}
var lines []string
for _, a := range looked {
if ctx.Err() != nil {
return "", ctx.Err()
}
if _, line, err := s.issue(ctx, shelf, a.Node, a.Module); err != nil {
return "", fmt.Errorf("%s's bus accounts on %s are not issued: %w", a.Module, a.Node, err)
} else if line != "" {
lines = append(lines, a.Module+" on "+a.Node+": "+line)
}
}
said, err := s.cover(ctx)
lines = append(append(forgot, lines...), fmt.Sprintf("%d standing assignment(s) looked at; %s", len(looked), said))
return strings.Join(lines, "\n"), err
}
// errNotHolding is a controller that does not hold the lease: it issues nothing, and tries again until it holds it.
var errNotHolding = errors.New("this controller does not hold the lease")
// cover composes the node holding the bus and writes the grants, and says so.
func (s assignmentAccountSteps) cover(ctx context.Context) (string, error) {
holder, err := s.busHolder(ctx)
if err != nil {
return "", fmt.Errorf("the node holding the bus cannot be read: %w", err)
}
if holder != "" {
if err := s.compose(ctx, []string{holder}); err != nil {
return "", fmt.Errorf("the bus's user list on %s is not composed: %w", holder, err)
}
}
if err := s.grants(ctx); err != nil {
return "", err
}
if holder != "" {
return "the bus's user list composed on " + holder + ", the grants written", nil
}
return "the grants written", nil
}
// assignmentStands says a value of the assignments bucket holds an assignment that stands: made (a module only pulled in
// by another names no time assigned), not ended and in no state (chain.Assignment.Stands).
func assignmentStands(a chain.Assignment) bool {
return !a.Assigned.IsZero() && a.Stands() && a.Node != "" && a.Module != ""
}
// assignmentTurns keeps which keys of the assignments bucket stood when last heard, to tell an assignment that comes to
// stand from a rewrite of one that stood.
type assignmentTurns struct {
mu sync.Mutex
stood map[string]bool
}
// heard records a value of key (nil for a delete or a purge) and answers whether it holds an assignment that just came
// to stand.
func (t *assignmentTurns) heard(key string, value []byte) (chain.Assignment, bool) {
var a chain.Assignment
stands := false
if value != nil && json.Unmarshal(value, &a) == nil {
stands = assignmentStands(a)
}
t.mu.Lock()
defer t.mu.Unlock()
if t.stood == nil {
t.stood = map[string]bool{}
}
was := t.stood[key]
t.stood[key] = stands
return a, stands && !was
}
// ownAccountOf is the username of the account a module speaks on the bus with on node, its own and no process's: the
// module's when it reads a `broker` own secret, else its provisioner client's when it runs one; false when it has none
// (a module whose accounts are its processes' alone, or one that does not speak on the bus).
func ownAccountOf(m catalogue.Manifest, node string) (string, bool) {
if _, reads := m.OwnSecrets["broker"]; reads {
return broker.Principal{Kind: broker.KindModule, Node: node, Module: m.Module}.Username(), true
}
if runsTheProvisionerClient(m) {
return broker.Principal{Kind: broker.KindProvisioner, Node: node, Module: m.Module}.Username(), true
}
return "", false
}
// issueAssignmentAccounts issues module's accounts on node that are not minted: its own (issueOnTheNewBus) and its
// processes' (issueMissingProcessAccounts). A module the catalogue does not hold, or one with nowhere to deliver an
// account (mayIssue), has nothing to issue; an account minted before is kept.
func issueAssignmentAccounts(ctx context.Context, inv *inventory.Inventory, shelf map[string]catalogue.Manifest,
node, module string) (bool, string, error) {
// Mint only what no row names: the assign act hears its own write and may issue at once (the review of issue 1945).
ctx = onlyIfAbsent(ctx)
m, known := shelf[module]
if !known || mayIssue(m) != nil {
return false, "", nil
}
if _, own := ownAccountOf(m, node); own {
minted, err := inv.BusMinted(ctx) // sealed accounts alone: a row with no sealed credential is not issued
if err != nil {
return false, "", fmt.Errorf("the issued accounts cannot be read: %w", err)
}
if missingAccounts(m, node, minted) {
busAddress, err := broker.BusAddress()
if err != nil {
return false, "", err
}
// Its processes' and provisioner client's accounts too: issueOnTheNewBus issues each that is not issued.
ctx, got := withIssued(ctx)
if err := issueOnTheNewBus(ctx, inv, m, node, busAddress); err != nil {
return false, "", err
}
if issued := got.list(); len(issued) > 0 {
return true, "issued and sealed to " + node + ": " + strings.Join(issued, ", "), nil
}
return false, "", nil
}
return false, "its bus accounts on " + node + " stood issued and sealed", nil
}
line, err := issueMissingProcessAccounts(ctx, inv, m, node)
if err != nil {
return false, "", err
}
return line != "", line, nil
}
// missingAccounts says some account of m on node is not issued: its own, a process's, or its provisioner client's.
func missingAccounts(m catalogue.Manifest, node string, minted []string) bool {
if user, own := ownAccountOf(m, node); own && !slices.Contains(minted, user) {
return true
}
if runsTheProvisionerClient(m) &&
!slices.Contains(minted, broker.Principal{Kind: broker.KindProvisioner, Node: node, Module: m.Module}.Username()) {
return true
}
return len(processAccountsMissing(m, node, minted)) > 0
}
// liveAssignmentAccountSteps are the steps on the controller's own stores and the bus: the composer's passes, as the
// done keys' take-ins use them, so a node asked for by both is composed once a pass.
func liveAssignmentAccountSteps(open *stores) assignmentAccountSteps {
inv := open.inventory
return assignmentAccountSteps{
holds: func(ctx context.Context) bool {
// The lease's gate, as every act's: a serving controller the bus does not grant the lease yet acts unleased.
_, err := theLease.epoch(ctx)
return err == nil
},
forgetStale: func(ctx context.Context) ([]string, error) { return forgetStaleAccounts(ctx, open) },
catalogue: func(ctx context.Context) (map[string]catalogue.Manifest, error) { return inv.Catalogue(ctx) },
issue: func(ctx context.Context, shelf map[string]catalogue.Manifest, node, module string) (bool, string, error) {
return issueAssignmentAccounts(ctx, inv, shelf, node, module)
},
busHolder: func(ctx context.Context) (string, error) {
holders, err := seatHolders(ctx, inv)
if err != nil {
return "", err
}
return holders[catalogue.BrokerSeat].Node, nil
},
compose: func(ctx context.Context, nodes []string) error {
return composerFor(open).compose(ctx, nodes, false)
},
grants: func(ctx context.Context) error {
return composerFor(open).compose(ctx, nil, true)
},
}
}
// assignmentAccountsRetry is how long a settle that failed waits before it tries again.
var assignmentAccountsRetry = 30 * time.Second
// untilSettled runs settle until it works or ctx ends: a failure, or a controller not holding the lease, is said and
// tried again every assignmentAccountsRetry, never dropped, since no later write of the key may come.
func untilSettled(ctx context.Context, what string, settle func(context.Context) (string, error)) {
for {
said, err := settle(ctx)
if err == nil {
if said != "" {
for _, line := range strings.Split(said, "\n") {
fmt.Printf("assignments: %s\n", line)
}
}
return
}
if ctx.Err() != nil {
return
}
fmt.Printf("assignments: %s: %v; tried again in %s\n", what, err, assignmentAccountsRetry)
select {
case <-ctx.Done():
return
case <-time.After(assignmentAccountsRetry):
}
}
}
// watchAssignmentAccounts settles every assignment of the assignments bucket that comes to stand, for as long as ctx: the
// bucket's current values first (the start's look, settled as not fresh), then each write. A watch that ends, or a
// bucket not there yet, is watched again a minute later.
func watchAssignmentAccounts(ctx context.Context, open *stores) {
steps := liveAssignmentAccountSteps(open)
goServing(ctx, func() {
runAssignmentWatches(ctx, open.inventory.WatchAssignments, func(a chain.Assignment) {
inBackground(ctx, func() {
untilSettled(ctx, a.Module+" on "+a.Node, func(ctx context.Context) (string, error) {
return steps.settle(ctx, a.Node, a.Module)
})
})
}, func(look []chain.Assignment) {
// The start's look, once in the process: one reading of the catalogue, one assignment after another,
// then one composition of the bus's node and the grants (issues 1945 and 1277).
inBackground(ctx, func() {
untilSettled(ctx, "the start's look", func(ctx context.Context) (string, error) {
return steps.look(ctx, look)
})
})
}, func() bool {
select {
case <-ctx.Done():
return false
case <-time.After(time.Minute):
return true
}
})
})
}
// runAssignmentWatches watches the assignments again each time a watch ends, until ctx ends or pause says stop: each
// assignment that comes to stand is handed to fresh, and the standing ones of the first watch that reads them all (its
// Caught mark) to look, once. A watch that ends, or fails to open, before the start's look ran is no start's look: the
// next one is, with what it hears from nothing (a fresh assignmentTurns), so no key it handed is dropped and no burst of
// fresh settles replaces the look (the review of issue 1945).
func runAssignmentWatches(ctx context.Context, watch func(context.Context) (<-chan chain.Entry, error),
fresh func(chain.Assignment), look func([]chain.Assignment), pause func() bool) {
turns := &assignmentTurns{}
first := true
for {
var atStart []chain.Assignment
if err := watchAssignmentsOnce(ctx, watch, turns, first, func(a chain.Assignment, isFresh bool) {
if !isFresh {
atStart = append(atStart, a)
return
}
fresh(a)
}, func() {
first = false
look(atStart)
atStart = nil
}); err != nil && ctx.Err() == nil {
fmt.Printf("assignments: the assignments are not heard, so a new assignment's bus accounts wait: %v\n", err)
}
if first {
turns = &assignmentTurns{} // the start's look has not run: the next watch hears every standing key anew
}
if ctx.Err() != nil || !pause() {
return
}
}
}
// watchAssignmentsOnce watches the assignments until the watch or ctx ends, handing each assignment that comes to stand
// to settle. On the process's first watch, the current values are the start's look (not fresh) and read is called once
// they are all handed (the chain's Caught mark); every other value, and every value of a watch opened again after one
// ended, is fresh: an assignment that came to stand while no watch heard it is settled whole, never as the start's look
// (the review of issue 1945).
func watchAssignmentsOnce(ctx context.Context, watch func(context.Context) (<-chan chain.Entry, error),
turns *assignmentTurns, first bool, settle func(a chain.Assignment, fresh bool), read func()) error {
ctx, stop := context.WithCancel(ctx)
defer stop()
entries, err := watch(ctx)
if err != nil {
return err
}
current := true
for {
select {
case <-ctx.Done():
return nil
case e, ok := <-entries:
if !ok {
if ctx.Err() != nil {
return nil
}
return fmt.Errorf("the watch ended")
}
if e.Caught {
if current {
current = false // the current values are read: what follows is written from now on
if first {
read()
}
}
continue
}
var value []byte
if !e.Deleted {
value = e.Value
}
if a, turned := turns.heard(e.Key, value); turned {
settle(a, !current || !first)
}
}
}
}
// staleAccounts is, by node, each module some account of which stands on a node where the module no longer resolves:
// the accounts of a module's own, its processes' and its provisioner client's. A node whose resolution is unknown is
// left alone.
func staleAccounts(users map[string]inventory.BusUser, resolved map[string]map[string]bool) map[string][]string {
seen := map[string]map[string]bool{}
for _, u := range users {
switch u.Kind {
case inventory.BusModule, inventory.BusNodeTools, inventory.BusProcess, inventory.BusProvisioner:
default:
continue
}
on, known := resolved[u.Node]
if u.Node == "" || u.Module == "" || !known || on[u.Module] {
continue
}
if seen[u.Node] == nil {
seen[u.Node] = map[string]bool{}
}
seen[u.Node][u.Module] = true
}
out := map[string][]string{}
for node, modules := range seen {
out[node] = slices.Sorted(maps.Keys(modules))
}
return out
}
// forgetStaleAccounts forgets the accounts of every module on a node where it no longer resolves (assigned or
// required), from each node's own resolution; a node that does not resolve is left alone.
func forgetStaleAccounts(ctx context.Context, open *stores) ([]string, error) {
inv := open.inventory
users, err := inv.BusUsers(ctx)
if err != nil {
return nil, err
}
ctx = withInputCache(ctx) // one reading of the mesh for every node resolved
resolved := map[string]map[string]bool{}
for _, u := range users {
if _, done := resolved[u.Node]; done || u.Node == "" {
continue
}
plan, _, err := resolveNode(ctx, open, u.Node)
if err != nil {
continue // unknown: nothing of this node is forgotten
}
on := map[string]bool{}
for _, m := range plan.Modules {
on[m.Module] = true
}
resolved[u.Node] = on
}
var lines []string
stale := staleAccounts(users, resolved)
for _, node := range slices.Sorted(maps.Keys(stale)) {
for _, module := range stale[node] {
forgot, err := inv.ForgetBusAccountsOf(ctx, node, module, true)
if err != nil {
return lines, err
}
if len(forgot) > 0 {
lines = append(lines, fmt.Sprintf("%s no longer runs on %s: its bus accounts forgotten: %s", module, node,
strings.Join(forgot, ", ")))
}
}
}
return lines, nil
}
@@ -0,0 +1,452 @@
package main
import (
"context"
"encoding/json"
"errors"
"fmt"
"reflect"
"strings"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"github.com/novox/mesh-controller/internal/broker"
"github.com/novox/mesh-controller/internal/chain"
"github.com/novox/mesh-controller/internal/inventory"
)
// An assignment that comes to stand issues its node's bus accounts and the module's grant (ADR 0543, issue 1945): on
// 2026-10-25 at 17:19 mesh-build-node and mesh-build-go, built long before, were assigned to ace, and nothing issued
// ace's accounts nor put them in the bus's user list.
func anAssignmentValue(t *testing.T, a chain.Assignment) []byte {
t.Helper()
raw, err := json.Marshal(a)
if err != nil {
t.Fatal(err)
}
return raw
}
// Only a write that makes the key stand turns it: a rewrite of a standing assignment (the provisioner's plan kept in its
// value), an ended one, a pending one or a module only pulled in by another does not. The process's first watch hands
// the assignments standing when it starts to the start's look (not fresh), says once that they are all read, and hands
// each assignment written afterwards as fresh; a watch opened again after one ended hands every standing assignment it
// has not seen stand as fresh and never calls the start's look (the review of issue 1945).
func TestAnAssignmentThatComesToStand(t *testing.T) {
at := time.Date(2026, 10, 25, 16, 19, 42, 0, time.UTC)
stands := chain.Assignment{Node: "ace", Module: "mesh-build-node", Assigned: at}
ended := stands
ended.Ended = at.Add(time.Hour)
pending := stands
pending.State = chain.StatePending
pulled := chain.Assignment{Node: "ace", Module: "mesh-build-node", RequiredBy: "mesh-build-go"}
replanned := stands
replanned.Provisioning = json.RawMessage(`{"plan":1}`)
type write struct {
value *chain.Assignment // nil for a delete
turning bool
}
for _, c := range []struct {
name string
writes []write
}{
{"a new assignment turns", []write{{&stands, true}}},
{"a rewrite of one that stands does not", []write{{&stands, true}, {&replanned, false}, {&stands, false}}},
{"an ended one does not", []write{{&ended, false}}},
{"assigned again after it ended turns again", []write{{&stands, true}, {&ended, false}, {&stands, true}}},
{"a pending one does not, and turns once made", []write{{&pending, false}, {&stands, true}}},
{"a module only pulled in does not", []write{{&pulled, false}}},
{"one deleted and written again turns again", []write{{&stands, true}, {nil, false}, {&stands, true}}},
} {
t.Run(c.name, func(t *testing.T) {
turns := &assignmentTurns{}
for i, w := range c.writes {
var value []byte
if w.value != nil {
value = anAssignmentValue(t, *w.value)
}
a, turned := turns.heard("ace.mesh-build-node", value)
if turned != w.turning {
t.Fatalf("write %d: turned %v, want %v", i, turned, w.turning)
}
if turned && (a.Node != "ace" || a.Module != "mesh-build-node") {
t.Fatalf("write %d: the assignment turned is %s on %s", i, a.Module, a.Node)
}
}
})
}
type handed struct {
key string
fresh bool
}
for _, first := range []bool{true, false} {
t.Run(fmt.Sprintf("the watch, first %v", first), func(t *testing.T) {
bucket := chain.NewMemory()
put := func(a chain.Assignment) {
t.Helper()
value := anAssignmentValue(t, a)
if e, held, _ := bucket.Get(t.Context(), a.Key()); held {
if _, err := bucket.Update(t.Context(), a.Key(), value, e.Revision); err != nil {
t.Fatal(err)
}
return
}
if _, err := bucket.Create(t.Context(), a.Key(), value); err != nil {
t.Fatal(err)
}
}
put(chain.Assignment{Node: "g14", Module: "mesh-build-go", Assigned: at})
got := make(chan handed, 8)
read := make(chan struct{}, 8)
ctx, stop := context.WithCancel(t.Context())
done := make(chan error, 1)
go func() {
done <- watchAssignmentsOnce(ctx, bucket.Watch, &assignmentTurns{}, first,
func(a chain.Assignment, fresh bool) { got <- handed{a.Key(), fresh} },
func() { read <- struct{}{} })
}()
next := func() handed {
t.Helper()
select {
case h := <-got:
return h
case <-time.After(5 * time.Second):
t.Fatal("nothing handed")
return handed{}
}
}
// Standing when the watch opened: the start's look on the first watch, fresh on any other.
if h := next(); h != (handed{"g14.mesh-build-go", !first}) {
t.Fatalf("the standing assignment was handed %+v", h)
}
put(chain.Assignment{Node: "g14", Module: "mesh-build-go", Assigned: at, Provisioning: json.RawMessage(`{"plan":2}`)})
put(chain.Assignment{Node: "ace", Module: "mesh-build-node", Assigned: at})
if h := next(); h != (handed{"ace.mesh-build-node", true}) {
t.Fatalf("handed %+v, want ace.mesh-build-node as fresh (and nothing for the rewrite before it)", h)
}
stop()
if err := <-done; err != nil {
t.Fatal(err)
}
if want := map[bool]int{true: 1, false: 0}[first]; len(read) != want {
t.Fatalf("the start's look was called %d time(s), want %d", len(read), want)
}
})
}
// A first watch that fails to open, or ends before its Caught mark, is no start's look: the next watch is, and hands
// every standing assignment to it once, none as fresh (the review of issue 1945: a burst of fresh settles, or keys
// handed and dropped).
for _, firstWatch := range []string{"fails to open", "ends before its Caught mark"} {
t.Run("a first watch that "+firstWatch, func(t *testing.T) {
bucket := chain.NewMemory()
standing := chain.Assignment{Node: "g14", Module: "mesh-build-go", Assigned: at}
if _, err := bucket.Create(t.Context(), standing.Key(), anAssignmentValue(t, standing)); err != nil {
t.Fatal(err)
}
watches := 0
watch := func(ctx context.Context) (<-chan chain.Entry, error) {
watches++
if watches > 1 {
return bucket.Watch(ctx)
}
if firstWatch == "fails to open" {
return nil, errors.New("the chain is not open yet")
}
cut := make(chan chain.Entry, 1)
cut <- chain.Entry{Key: standing.Key(), Value: anAssignmentValue(t, standing)}
close(cut)
return cut, nil
}
var looks [][]chain.Assignment
var fresh []chain.Assignment
ctx, stop := context.WithCancel(t.Context())
defer stop()
runAssignmentWatches(ctx, watch, func(a chain.Assignment) { fresh = append(fresh, a) },
func(look []chain.Assignment) { looks = append(looks, look); stop() },
func() bool { return ctx.Err() == nil })
if len(looks) != 1 || len(looks[0]) != 1 || looks[0][0].Key() != standing.Key() || len(fresh) != 0 {
t.Fatalf("after %d watch(es): looks %v, fresh %v; want one look at %s and nothing fresh",
watches, looks, fresh, standing.Key())
}
})
}
}
// settle issues what is missing and then puts it on the bus: the node holding the bus composed (its user list) and the
// grants written, even when every account was minted before (a module assigned again keeps its account). The start's
// look reads the catalogue once, issues for each standing assignment, and composes the bus's node and writes the grants
// once when it looked at any. Neither acts for a controller that does not hold the lease, nor for the controller's own
// module.
func TestSettlingAndTheStartsLook(t *testing.T) {
ace := func(m string) chain.Assignment { return chain.Assignment{Node: "ace", Module: m} }
cases := []struct {
name string
look []chain.Assignment // nil: settle mesh-build-node on ace
module string
notHolding bool
issued bool
issueErr, composeE error
holder string
wantIssues int
wantComposed []string
wantGrants bool
wantErr string
stale []string // what forgetting the accounts of modules no longer on their nodes says
}{
{name: "an account not minted", module: "mesh-build-node", issued: true, holder: "novox", wantIssues: 1,
wantComposed: []string{"novox"}, wantGrants: true},
{name: "an account minted before: the bus and the grants still", module: "mesh-build-go", holder: "novox",
wantIssues: 1, wantComposed: []string{"novox"}, wantGrants: true},
{name: "no lease: nothing, and tried again", module: "mesh-build-node", notHolding: true, issued: true,
holder: "novox", wantErr: errNotHolding.Error()},
{name: "the controller's own module is the terminal's", module: controllerModule, issued: true, holder: "novox"},
{name: "no bus holder: the grants still", module: "mesh-build-node", issued: true, wantIssues: 1, wantGrants: true},
{name: "an issue that fails is the error, nothing composed", module: "mesh-build-node",
issueErr: errors.New("the bus cannot be reached"), holder: "novox", wantIssues: 1, wantErr: "the bus cannot be reached"},
{name: "a composition that fails is the error, no grant", module: "mesh-build-node", issued: true, holder: "novox",
composeE: errors.New("store refused"), wantIssues: 1, wantComposed: []string{"novox"}, wantErr: "store refused"},
{name: "the start's look: each issued, the bus and the grants once",
look: []chain.Assignment{ace("mesh-build-node"), ace(controllerModule), ace("mesh-build-go")}, issued: true,
holder: "novox", wantIssues: 2, wantComposed: []string{"novox"}, wantGrants: true},
{name: "the start's look with every account minted still covers once",
look: []chain.Assignment{ace("mesh-build-node"), ace("mesh-build-go")}, holder: "novox", wantIssues: 2,
wantComposed: []string{"novox"}, wantGrants: true},
{name: "the start's look at nothing does nothing", look: []chain.Assignment{ace(controllerModule)}, holder: "novox"},
{name: "the start's look forgets the accounts of modules no longer on their nodes first",
look: []chain.Assignment{ace("mesh-build-node")}, holder: "novox", wantIssues: 1, wantComposed: []string{"novox"},
wantGrants: true, stale: []string{"mesh-build-go no longer runs on ace: its bus accounts forgotten: ace.mesh-build-go"}},
}
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
var composed []string
issues, catalogues, grants := 0, 0, 0
forgets := 0
steps := assignmentAccountSteps{
holds: func(context.Context) bool { return !c.notHolding },
forgetStale: func(context.Context) ([]string, error) {
forgets++
return c.stale, nil
},
catalogue: func(context.Context) (map[string]catalogue.Manifest, error) {
catalogues++
return map[string]catalogue.Manifest{}, nil
},
issue: func(context.Context, map[string]catalogue.Manifest, string, string) (bool, string, error) {
issues++
if c.issued {
return true, "its bus credential issued", c.issueErr
}
return false, "", c.issueErr
},
busHolder: func(context.Context) (string, error) { return c.holder, nil },
compose: func(_ context.Context, nodes []string) error {
composed = append(composed, nodes...)
return c.composeE
},
grants: func(context.Context) error { grants++; return nil },
}
var said string
var err error
if c.look != nil {
said, err = steps.look(t.Context(), c.look)
} else {
said, err = steps.settle(t.Context(), "ace", c.module)
}
if c.wantErr != "" {
if err == nil || !strings.Contains(err.Error(), c.wantErr) {
t.Fatalf("error %v, want one saying %q", err, c.wantErr)
}
} else if err != nil {
t.Fatal(err)
}
if issues != c.wantIssues || catalogues > 1 {
t.Fatalf("%d issue(s) from %d reading(s) of the catalogue, want %d from at most one", issues, catalogues, c.wantIssues)
}
if strings.Join(composed, ",") != strings.Join(c.wantComposed, ",") {
t.Fatalf("composed %v, want %v", composed, c.wantComposed)
}
if want := map[bool]int{true: 1}[c.wantGrants]; grants != want {
t.Fatalf("grants written %d time(s), want %d", grants, want)
}
if want := map[bool]int{true: 1}[len(c.look) > 0 && !c.notHolding]; forgets != want {
t.Fatalf("forgot stale accounts %d time(s), want %d (only from a reading that holds assignments)", forgets, want)
}
for _, line := range c.stale {
if !strings.Contains(said, line) {
t.Fatalf("said %q, want it to say %q", said, line)
}
}
if c.wantGrants && c.wantErr == "" && !strings.Contains(said, "the grants written") {
t.Fatalf("said %q, want it to name the grants", said)
}
})
}
}
// The accounts a module speaks with on a node: the module's own when it reads `broker`, its provisioner client's when
// it runs one alone, none when its accounts are its processes' or it does not speak on the bus. A process's account is
// held in the provisioner client's form, named for the process and carrying no claims, and may be issued only when
// the process declares its own secret broker-<process id> (ADR 0436 §2; issue 1181).
func TestTheAccountsAModuleSpeaksWithOnANode(t *testing.T) {
client := aProvisioner()
delete(client.OwnSecrets, "broker")
for _, c := range []struct {
name string
m catalogue.Manifest
user string
own bool
}{
{"a module reading broker", catalogue.Manifest{Module: "mesh-build-node",
OwnSecrets: catalogue.OwnSecrets{"broker": {Path: "/var/lib/mesh-build-node/broker"}}},
broker.Principal{Kind: broker.KindModule, Node: "ace", Module: "mesh-build-node"}.Username(), true},
{"a module reading broker and running the client", aProvisioner(),
broker.Principal{Kind: broker.KindModule, Node: "ace", Module: "mesh-provisioner"}.Username(), true},
{"the provisioner client alone", client,
broker.Principal{Kind: broker.KindProvisioner, Node: "ace", Module: "mesh-provisioner"}.Username(), true},
{"no broker, no client", catalogue.Manifest{Module: "hello-web"}, "", false},
} {
t.Run(c.name, func(t *testing.T) {
user, own := ownAccountOf(c.m, "ace")
if user != c.user || own != c.own {
t.Fatalf("got %q %v, want %q %v", user, own, c.user, c.own)
}
})
}
t.Run("a process's account", func(t *testing.T) {
user := broker.Principal{Kind: broker.KindProcess, Node: "novox", Module: "mesh-mail", Process: "send"}.Username()
if user != "novox.mesh-mail.send" {
t.Fatalf("the process's username: %q", user)
}
held, err := heldProcessAccount(broker.Broker{Fingerprint: "abc"}, "novox.example:4222", "novox", "mesh-mail",
user, "a-password")
if err != nil {
t.Fatal(err)
}
var got map[string]any
if err := json.Unmarshal(held, &got); err != nil {
t.Fatal(err)
}
want := map[string]any{"url": "nats://novox.example:4222", "fingerprint": "abc", "node": "novox",
"module": "mesh-mail", "user": "novox.mesh-mail.send", "password": "a-password"}
for k, v := range want {
if got[k] != v {
t.Errorf("%s: %v, want %v", k, got[k], v)
}
}
if len(got) != len(want) {
t.Errorf("the held account carries %v; want exactly %v (never claims — a process holds no seat)", got, want)
}
})
// An issue mints and seals an account only when it is not issued: no row, or a row with no sealed credential (issue
// 1945, live on 2026-10-25 at 20:04: ace's mesh-build-go had a row from 17:19 and no credential), under the lock on
// the account; a seal that fails leaves it unsealed for the next issue. A rotation always mints.
for _, c := range []struct {
name string
only bool
minted, sealed bool
sealErr error
want string
}{
{"no row: minted unsealed, sealed, marked", true, false, false, nil, "lock mint-unsealed seal mark release"},
{"a row sealed: nothing", true, true, true, nil, "lock release"},
{"a row with no sealed credential: minted and sealed again", true, true, false, nil, "lock mint-unsealed seal mark release"},
{"a seal that fails: not marked", true, false, false, errors.New("the vault did not take it"), "lock mint-unsealed seal release"},
{"a rotation mints whatever stands", false, true, true, nil, "mint seal"},
} {
t.Run("issuing an account: "+c.name, func(t *testing.T) {
acts := &anAccountStore{minted: c.minted, sealed: c.sealed}
ctx, got := withIssued(t.Context())
if c.only {
ctx = onlyIfAbsent(ctx)
}
err := issueAccount(ctx, acts, inventory.BusUser{Username: "ace.mesh-build-go", Kind: inventory.BusModule},
func(password string) error {
acts.did("seal")
if password == "" {
t.Error("sealed no password")
}
return c.sealErr
})
if (err != nil) != (c.sealErr != nil) {
t.Fatalf("error %v, want %v", err, c.sealErr)
}
if strings.Join(acts.acts, " ") != c.want {
t.Fatalf("acts %q, want %q", strings.Join(acts.acts, " "), c.want)
}
if issued := len(got.list()) == 1; issued != (c.sealErr == nil && strings.Contains(c.want, "seal")) {
t.Fatalf("said issued %v", got.list())
}
})
}
// The accounts of a module that no longer resolves on its node are stale, its provisioner client's among them; a
// module still resolved (assigned or required), an account of another kind, and a node whose resolution is unknown
// are left alone (issue 1945: ace's mesh-build-go rows from 17:19, marked sealed by migration 0108).
t.Run("stale accounts", func(t *testing.T) {
u := func(kind, node, module string) inventory.BusUser {
return inventory.BusUser{Username: node + "." + module + "." + kind, Kind: kind, Node: node, Module: module}
}
users := map[string]inventory.BusUser{}
for _, b := range []inventory.BusUser{u(inventory.BusModule, "ace", "mesh-build-go"),
u(inventory.BusProcess, "ace", "mesh-build-go"), u(inventory.BusProvisioner, "ace", "mesh-provisioner"),
u(inventory.BusModule, "ace", "mesh-build-node"), u(inventory.BusNode, "ace", ""),
u(inventory.BusModule, "g14", "mesh-build-go")} {
users[b.Username] = b
}
got := staleAccounts(users, map[string]map[string]bool{"ace": {"mesh-build-node": true}})
if want := map[string][]string{"ace": {"mesh-build-go", "mesh-provisioner"}}; !reflect.DeepEqual(got, want) {
t.Fatalf("stale %v, want %v", got, want)
}
})
t.Run("a process may be issued only with its own broker secret", func(t *testing.T) {
// mesh-mail's real shape: no `broker` of its own, its process declaring bus (review round 1 of issue 1181).
if err := mayIssue(mailSenderWithProcess()); err != nil {
t.Fatalf("a module with no broker of its own, whose process declares bus, was refused: %v", err)
}
m := mailSenderWithProcess()
delete(m.OwnSecrets, "broker-send")
err := mayIssue(m)
if err == nil {
t.Fatal("a process with no broker-<id> own secret was issued an account")
}
for _, want := range []string{"mesh-mail", "send", "broker-send", "own-secrets"} {
if !strings.Contains(err.Error(), want) {
t.Errorf("the refusal does not say %q: %v", want, err)
}
}
})
}
// anAccountStore is an accountIssuer that writes down what was asked of it.
type anAccountStore struct {
minted, sealed bool
acts []string
}
func (a *anAccountStore) did(act string) { a.acts = append(a.acts, act) }
func (a *anAccountStore) LockBusAccount(context.Context, string) (func(), error) {
a.did("lock")
return func() { a.did("release") }, nil
}
func (a *anAccountStore) BusAccountIssued(context.Context, string) (bool, bool, error) {
return a.minted, a.sealed, nil
}
func (a *anAccountStore) MintBusPassword(context.Context, inventory.BusUser) (string, error) {
a.did("mint")
return "a-password", nil
}
func (a *anAccountStore) MintBusPasswordUnsealed(context.Context, inventory.BusUser) (string, error) {
a.did("mint-unsealed")
return "a-password", nil
}
func (a *anAccountStore) MarkBusAccountSealed(context.Context, string) error {
a.did("mark")
return nil
}
+3
View File
@@ -273,4 +273,7 @@ type answers struct {
// applied, is not well: an assignment somebody made is not made yet, or will not be. // applied, is not well: an assignment somebody made is not made yet, or will not be.
pending []inventory.PendingAssignment pending []inventory.PendingAssignment
pendingUnread string pendingUnread string
// takeIn is the take-in's backlog of done keys on the bus (novox/hq ADR 0544 §6), nil where it cannot be read
// (a process with no bus, or a consumer not made yet).
takeIn *takeInBacklog
} }
+10 -3
View File
@@ -326,9 +326,13 @@ func restOfTheMesh(ctx context.Context, open *stores, shelf map[string]catalogue
return world, nil return world, nil
} }
// restOfTheMeshOf is restOfTheMesh for a caller that holds the inventory alone: the other stores the Input reads are opened for // restOfTheMeshOf is restOfTheMesh for a caller that holds the inventory alone: on the serving controller's stores where it
// the call and let go of again. // 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) { 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} open := &stores{inventory: inv}
defer func() { defer func() {
if open.licences != nil { if open.licences != nil {
@@ -338,9 +342,12 @@ func restOfTheMeshOf(ctx context.Context, inv *inventory.Inventory, shelf map[st
open.identity.Close() 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 // 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. // that two claimants could take, the bus's users left out for want of a credential.
func sayFound(ctx context.Context, d compose.Declared) { func sayFound(ctx context.Context, d compose.Declared) {
@@ -102,8 +102,10 @@ func TestTheDeclarationsStreamGoesOnceEveryNodeRunsTheNewEngine(t *testing.T) {
// The new node-engine registered at its fingerprint, from its done key. // The new node-engine registered at its fingerprint, from its done key.
n6 := fpOf('6') n6 := fpOf('6')
keys, trunk := aKeyFor("mesh-host", "6", n6, "novox/mesh-host", "", "f00d0123456789ab") keys, trunk := aKeyFor("mesh-host", "6", n6, "novox/mesh-host", "", "f00d0123456789ab")
if _, err := liveDoneKeySteps(m.open, trunkKeys{keyBuckets: keys, trunk: trunk}).take(ctx, "mesh-host"); err != nil { settled, _ := liveTakeInSteps(m.open, trunkKeys{keyBuckets: keys, trunk: trunk}, m.js).batch(ctx,
t.Fatal(err) []delivered{{module: "mesh-host", again: true}}, map[string]string{})
if len(settled) != 1 || settled[0].err != nil || settled[0].verdict != acknowledged {
t.Fatalf("the take-in of mesh-host: %+v", settled)
} }
if _, err := m.js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: sdkdelivery.HaveBucket, History: sdkdelivery.HaveHistory}); err != nil { if _, err := m.js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: sdkdelivery.HaveBucket, History: sdkdelivery.HaveHistory}); err != nil {
t.Fatal(err) t.Fatal(err)
@@ -140,4 +142,35 @@ func TestTheDeclarationsStreamGoesOnceEveryNodeRunsTheNewEngine(t *testing.T) {
if !declarationsRetiredNow(ctx, m.open.inventory) { if !declarationsRetiredNow(ctx, m.open.inventory) {
t.Errorf("a raise after the retirement would assert the stream again") 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)
}
} }
+315 -293
View File
@@ -4,6 +4,8 @@ import (
"context" "context"
"errors" "errors"
"fmt" "fmt"
"maps"
"slices"
"sort" "sort"
"strings" "strings"
"sync" "sync"
@@ -12,210 +14,285 @@ import (
"git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/catalogue"
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/nats-io/nats.go/jetstream" "github.com/nats-io/nats.go/jetstream"
"github.com/novox/mesh-controller/internal/broker"
"github.com/novox/mesh-controller/internal/builderkeys" "github.com/novox/mesh-controller/internal/builderkeys"
"github.com/novox/mesh-controller/internal/conditions" "github.com/novox/mesh-controller/internal/conditions"
"github.com/novox/mesh-controller/internal/inventory" "github.com/novox/mesh-controller/internal/inventory"
) )
// A new fingerprint of a registered module, from its done key to its compositions (novox/hq ADR 0539 clauses 5 and 7, // A new fingerprint of a registered module, from its done key to its compositions (novox/hq ADR 0539 clauses 5 and 7,
// ADR 0501 §2). With no delivery plan left to take a module's done key in, a write of the builder's or the indexer's // ADR 0501 §2, ADR 0544). With no delivery plan left to take a module's done key in, a write of the builder's or the
// keys does it: the controller takes in the done key of the fingerprint the indexer's trunk value names (registers the // indexer's keys does it: the controller takes in the done key of the fingerprint the indexer's trunk value names
// module at it, as a plan's take-in did), composes every node the module is assigned to (which makes what a module's // (registers the module at it, as a plan's take-in did), composes every node the module is assigned to (which makes what
// new manifest needs, a credential among it, as a send made it), composes the node holding the bus (its user list now // a module's new manifest needs, a credential among it), composes the node holding the bus (its user list now carries
// carries the module's permissions at the new fingerprint: a new revision of the bus module's composition alone, // the module's permissions at the new fingerprint), and writes `grants.<module>.<fp>`. The assigner's sixth and seventh
// written into its want at once by the assigner), and writes `grants.<module>.<fp>`. The assigner's sixth and seventh // rules then let a want name the fingerprint.
// rules then let a want name the fingerprint. **No timer**: woken by the keys, and a starting controller looks at every //
// module once (doneKeysAtStart). // **A durable pull consumer, not a watch** (ADR 0544; research 072, milestones 2 and 4). The writes are read from two
// durable consumers (take_in_consumer.go): the bus keeps what is not taken in yet, so a restart takes in only what the
// controller before it left unacknowledged, and the start's look is the consumers' own catch-up. A batch of keys asks the
// composer twice whatever its size, and a key with nothing new composes nothing.
// doneKeySteps are what taking a module's done key in does, each a step a test may stand in for. // takeInSteps are what taking a batch of done keys in does, each a step a test may stand in for.
type doneKeySteps struct { type takeInSteps struct {
// held says the catalogue holds the module (a new one is the new-module take-in's, new_modules.go). // catalogue is every module the catalogue holds, with the nodes it is assigned to: read once a batch. A module it
held func(ctx context.Context, module string) (bool, error) // does not hold is the new-module take-in's (new_modules.go).
catalogue func(ctx context.Context) (map[string][]string, error)
// trunk is the indexer's trunk value of the module. // trunk is the indexer's trunk value of the module.
trunk func(ctx context.Context, module string) (sdkdelivery.Module, error) trunk func(ctx context.Context, module string) (sdkdelivery.Module, error)
// registeredAt says the module is registered at the commit. // registered says whether the module is registered at the trunk's commit, and whether at its fingerprint: a commit
registeredAt func(ctx context.Context, module, commit string) (bool, error) // that moves while the fingerprint stays (a catalogue merge that leaves the module's sources alone) is no new build.
// takeIn registers the module from the builder's done key of the trunk's fingerprint (takeInKey); errKeyWaits when registered func(ctx context.Context, module string, trunk sdkdelivery.Module) (atCommit, atFP bool, err error)
// the key is not done yet. // recordCommit records the module's registration at the trunk's commit, its fingerprint unchanged: nothing composed.
recordCommit func(ctx context.Context, module, commit string) error
// takeIn registers the module from the builder's done key of the trunk's fingerprint (takeInKey).
takeIn func(ctx context.Context, module string, trunk sdkdelivery.Module) error takeIn func(ctx context.Context, module string, trunk sdkdelivery.Module) error
// nodesOf are the nodes the module is assigned to; busHolder the node holding the bus, "" for none. // taken says the module's compositions at fp exist for every node and its grant is written: nothing to compose.
nodesOf func(ctx context.Context, module string) ([]string, error) taken func(ctx context.Context, module, fp string, nodes []string) (bool, error)
// busHolder is the node holding the bus, "" for none.
busHolder func(ctx context.Context) (string, error) busHolder func(ctx context.Context) (string, error)
// compose writes the nodes' compositions; grants writes the grants issued. // compose composes the nodes, and writes the grants after when grants: each node's error, and the grants'.
compose func(ctx context.Context, nodes []string) error compose func(ctx context.Context, nodes []string, grants bool) (map[string]error, error, error)
grants func(ctx context.Context) error
} }
// take takes module's done key in and composes what it changed. It answers what it did in a line, "" when there was // delivered is one key a consumer delivered: its module, and whether it was delivered before (a take-in that may have
// nothing to do. // stopped half-way, so it is taken in whole).
func (s doneKeySteps) take(ctx context.Context, module string) (string, error) { type delivered struct {
held, err := s.held(ctx, module) module string
if err != nil || !held { again bool
return "", err // deliveries is how often the bus delivered it, this time included: what a refusal's delay grows with.
} deliveries uint64
trunk, err := s.trunk(ctx, module)
if errors.Is(err, builderkeys.ErrNoValue) {
return "", nil
}
if err != nil {
return "", err
}
if !strings.HasPrefix(trunk.FP2, sdkdelivery.SourcePrefix2) {
return "", nil
}
at, err := s.registeredAt(ctx, module, trunk.Commit)
if err != nil {
return "", err
}
said := ""
if !at {
if err := s.takeIn(ctx, module, trunk); err != nil {
return "", err
}
said = fmt.Sprintf("%s registered from the builder's done key %s (%s)", module, short(trunk.FP2), short(trunk.Commit))
}
nodes, err := s.nodesOf(ctx, module)
if err != nil {
return said, err
}
if len(nodes) == 0 {
return said, nil
}
// The module's nodes first: composing them makes what the new manifest needs, its credentials among it.
if err := s.compose(ctx, nodes); err != nil {
return said, err
}
// Then the bus's user list, which carries the module's permissions at its new fingerprint.
holder, err := s.busHolder(ctx)
if err != nil {
return said, err
}
if holder != "" && !inList(nodes, holder) {
if err := s.compose(ctx, []string{holder}); err != nil {
return said, err
}
}
// Then the grant, which with the composition lets a want name the fingerprint (rules 6 and 7).
if err := s.grants(ctx); err != nil {
return said, err
}
if said == "" {
said = fmt.Sprintf("%s at %s", module, short(trunk.FP2))
}
return said + "; composed for " + strings.Join(nodes, ", ") + ", its grant written", nil
} }
func inList(list []string, s string) bool { // verdict is what becomes of a module's delivered keys.
for _, v := range list { type verdict int
if v == s {
return true
}
}
return false
}
// doneKeyTakeOnce is one take-in; a variable so a test may stand in for it. doneKeysSay says a line of the take-ins. const (
var ( // acknowledged: taken in, nothing new, a wait a later write ends, or a build key that failed.
doneKeyTakeOnce = takeDoneKeyOnce acknowledged verdict = iota
doneKeysSay = func(s string) { fmt.Println(s) } // refused: delivered again by the bus after doneKeyRetry (a wait on a read, or a failure).
refused
) )
// doneKeysBeforeLetGo is called just before a follower lets go of its module; a variable so a test can wake the module // settledKey is what a batch did with one module.
// at that moment. Nil does nothing. type settledKey struct {
var doneKeysBeforeLetGo func(module string) module string
verdict verdict
// doneKeysLookAgain marks a module woken while a take-in of it ran: that take-in goes round once more. // taken: registered, composed and granted now (not nothing new); its condition is cleared.
var doneKeysLookAgain sync.Map taken bool
// nothingNew: registered and composed before, so nothing was done.
// doneKeysMu keeps one take-in of a module's done key at a time in this process, per module. nothingNew bool
var doneKeysMu sync.Map // waits is why it waits, "" when it does not.
waits string
// takeDoneKey is a woken module's take-in on the live stores and buckets, said. A take-in that fails (its compositions // err is a failure: raised as the module's condition.
// or its grant not written) is tried again every doneKeyRetry while it fails, loudly: printed each time, and an urgent err error
// condition raised by the first until it has worked, as the chain's own moves are (layBuildsOutSoon). A key not done // said is the line said of it, "" for none.
// yet is no failure: it is said once for each reason it waits for and looked at again every doneKeyRetry. said string
func takeDoneKey(ctx context.Context, open *stores, module string) {
mu, _ := doneKeysMu.LoadOrStore(module, &sync.Mutex{})
for {
if !mu.(*sync.Mutex).TryLock() {
// A take-in of the module already follows it: it is told to look again once it is done, so a newer key
// written meanwhile is never lost (the review of issue 1945).
doneKeysLookAgain.Store(module, true)
return
}
followDoneKey(ctx, open, module)
if doneKeysBeforeLetGo != nil {
doneKeysBeforeLetGo(module)
}
mu.(*sync.Mutex).Unlock()
// A wake between this follower's last look and its letting go failed its TryLock and only marked the module:
// looked at again now that it is let go, and taken once more (the review of issue 1945).
if _, again := doneKeysLookAgain.LoadAndDelete(module); !again || ctx.Err() != nil {
return
}
}
} }
// followDoneKey takes a module's done key in, under the module's lock, until it is taken in or ctx ends: a failure or a // batch takes the delivered keys in, each module once: it answers one settledKey per module in the order first
// wait is tried again every doneKeyRetry. // delivered, and the number of composer passes it asked. memo is, per module, the trunk (fingerprint and commit) this
func followDoneKey(ctx context.Context, open *stores, module string) { // process took in last; a module at the same trunk delivered for the first time has nothing new.
failed := false func (s takeInSteps) batch(ctx context.Context, keys []delivered, memo map[string]string) ([]settledKey, int) {
saidWait := "" var order []string
for { again := map[string]bool{}
doneKeysLookAgain.Delete(module) // what this look reads covers every wake before it for _, k := range keys {
err := doneKeyTakeOnce(ctx, open, module) if _, seen := again[k.module]; !seen {
o := doneKeyObservation(module) order = append(order, k.module)
if err == nil {
if keeper := conditionsFrom; keeper != nil && failed {
_, _ = keeper.Clear(ctx, o.Key(), "composed and granted")
failed = false
}
if _, again := doneKeysLookAgain.LoadAndDelete(module); again {
continue // a wake came while this one ran: once more, for the key written meanwhile
}
return
} }
if errors.Is(err, errKeyWaits) { again[k.module] = again[k.module] || k.again
// 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). out := make([]settledKey, 0, len(order))
if why := err.Error(); why != saidWait { all := func(v verdict, err error) ([]settledKey, int) {
doneKeysSay(fmt.Sprintf("done keys: %s waits: %s; looked at again every %s", module, for _, m := range order {
strings.TrimPrefix(why, errKeyWaits.Error()+": "), doneKeyRetry)) out = append(out, settledKey{module: m, verdict: v, err: err})
saidWait = why
}
} else {
doneKeysSay(fmt.Sprintf("done keys: %s: %v; tried again in %s", module, err, doneKeyRetry))
} }
if errors.Is(err, errKeyWaits) { return out, 0
select { }
case <-ctx.Done(): held, err := s.catalogue(ctx)
return if err != nil {
case <-time.After(doneKeyRetry): return all(refused, fmt.Errorf("the catalogue cannot be read: %w", err))
} }
type toCompose struct {
at int
nodes []string
mark string
said string
}
var need []toCompose
for _, m := range order {
r := settledKey{module: m}
nodes, isHeld := held[m]
if !isHeld {
out = append(out, r)
continue continue
} }
if ctx.Err() != nil { trunk, err := s.trunk(ctx, m)
return // stopped: a take-in cut by the stop is no fault of the build, and the next controller takes it in if errors.Is(err, builderkeys.ErrNoValue) {
out = append(out, r)
continue
} }
if keeper := conditionsFrom; keeper != nil && !failed && !errors.Is(err, context.Canceled) { if err == nil && !strings.HasPrefix(trunk.FP2, sdkdelivery.SourcePrefix2) {
// A take-in whose own step was cancelled is tried again, and raises nothing: a cancellation is no fault // The indexer's next write names one: said once, as every wait for a write (ADR 0544 §4).
// of the build (issue 1945: 30 urgent conditions live, each a take-in cut by a restart). why := trunk.None2
o.Summary += err.Error() if why == "" {
if _, oerr := keeper.Observe(ctx, o); oerr == nil { why = "the indexer wrote no fingerprint of the new form"
failed = true }
r.waits = fmt.Sprintf("%v: %s at %.8s has no fingerprint to take a build by: %s", errWaitsForAWrite, m,
trunk.Commit, why)
out = append(out, r)
continue
}
if err != nil {
r.verdict, r.err = refused, err
out = append(out, r)
continue
}
mark := trunk.FP2 + "@" + trunk.Commit
if !again[m] && memo[m] == mark {
r.nothingNew = true
out = append(out, r)
continue
}
at, atFP, err := s.registered(ctx, m, trunk)
if err == nil && !at && atFP {
// Only the commit moved (issue 1945: every catalogue merge took about 200 modules in again, 25 minutes): the
// new commit is recorded, and the module is judged as one registered, composed only if something is missing.
if err = s.recordCommit(ctx, m, trunk.Commit); err == nil {
at = true
} }
} }
select { if err != nil {
case <-ctx.Done(): r.verdict, r.err = refused, err
return out = append(out, r)
case <-time.After(doneKeyRetry): continue
}
said := ""
if !at {
if err := s.takeIn(ctx, m, trunk); err != nil {
switch {
case errors.Is(err, errWaitsForAWrite):
r.waits = err.Error()
case errors.Is(err, errKeyWaits):
r.verdict, r.waits = refused, err.Error()
case errors.Is(err, errKeyFailed):
r.err = err // acknowledged: only a new build ends it, and its key is delivered then
default:
r.verdict, r.err = refused, err
}
out = append(out, r)
continue
}
said = fmt.Sprintf("%s registered from the builder's done key %s (%s)", m, short(trunk.FP2), short(trunk.Commit))
} else if !again[m] {
// Registered at its trunk already: composed and granted by a take-in before this one, or at this
// controller's start, unless that take-in stopped half-way, which leaves a composition or the grant missing.
if done, err := s.taken(ctx, m, trunk.FP2, nodes); err == nil && done {
memo[m] = mark
r.nothingNew = true
out = append(out, r)
continue
}
}
if len(nodes) == 0 {
memo[m] = mark
r.taken, r.said = true, said
out = append(out, r)
continue
}
if said == "" {
said = fmt.Sprintf("%s at %s", m, short(trunk.FP2))
}
need = append(need, toCompose{at: len(out), nodes: nodes, mark: mark, said: said})
out = append(out, r)
}
if len(need) == 0 {
return out, 0
}
refuseNeed := func(err error) ([]settledKey, int) {
for _, n := range need {
out[n.at].verdict, out[n.at].err = refused, err
}
return out, 0
}
holder, err := s.busHolder(ctx)
if err != nil {
return refuseNeed(fmt.Errorf("the node holding the bus cannot be read: %w", err))
}
// The modules' nodes first, in one pass: composing them makes what their new manifests need, credentials among it.
var union []string
for _, n := range need {
for _, node := range n.nodes {
if node != holder && !slices.Contains(union, node) {
union = append(union, node)
}
} }
} }
sort.Strings(union)
passes := 0
failed := map[string]error{}
if len(union) > 0 {
f, _, err := s.compose(ctx, union, false)
passes++
if err != nil {
return refuseNeed(err)
}
maps.Copy(failed, f)
}
// Then the bus's user list, which carries the modules' permissions at their new fingerprints, and the grants, which
// with the compositions let a want name each fingerprint (rules 6 and 7).
var bus []string
if holder != "" {
bus = []string{holder}
}
f, grantsErr, err := s.compose(ctx, bus, true)
passes++
if err != nil {
return refuseNeed(err)
}
maps.Copy(failed, f)
for _, n := range need {
var errs []error
for _, node := range append(slices.Clone(n.nodes), bus...) {
if e := failed[node]; e != nil && !slices.ContainsFunc(errs, func(x error) bool { return x == e }) {
errs = append(errs, e)
}
}
if grantsErr != nil {
errs = append(errs, grantsErr)
}
r := &out[n.at]
if len(errs) > 0 {
r.verdict, r.err = refused, errors.Join(errs...)
continue
}
memo[r.module] = n.mark
r.taken = true
r.said = n.said + "; composed for " + strings.Join(n.nodes, ", ") + ", its grant written"
}
return out, passes
} }
// doneKeyRetry is how long a failed take-in of a done key waits before it tries again. // doneKeysSay says a line of the take-ins; a variable so a test may hear it.
var doneKeysSay = func(s string) { fmt.Println(s) }
// doneKeyRetry is how long the bus waits before it delivers again a key whose take-in was refused the first time; each
// refusal after doubles it, up to doneKeyRetryMost, so a key refused for long (a manifest word this controller does not
// read yet, a node whose composition keeps failing) costs a take-in every few minutes, not every half minute.
var doneKeyRetry = 30 * time.Second var doneKeyRetry = 30 * time.Second
// doneKeyRetryMost is the longest a refused key waits before it is delivered again.
const doneKeyRetryMost = 10 * time.Minute
// refusalDelay is how long the bus waits before it delivers again a key refused on its deliveries-th delivery.
func refusalDelay(deliveries uint64) time.Duration {
d := doneKeyRetry
for i := uint64(1); i < deliveries && d < doneKeyRetryMost; i++ {
d *= 2
}
return min(d, doneKeyRetryMost)
}
// kindNotComposed is the condition of a module's new fingerprint whose compositions or grant are not written: until // kindNotComposed is the condition of a module's new fingerprint whose compositions or grant are not written: until
// they are, no want can name it. // they are, no want can name it.
const kindNotComposed = "fingerprint-not-composed" const kindNotComposed = "fingerprint-not-composed"
@@ -226,40 +303,37 @@ func doneKeyObservation(module string) conditions.Observation {
Summary: module + "'s new fingerprint is not composed or granted, so no want can name it: "} Summary: module + "'s new fingerprint is not composed or granted, so no want can name it: "}
} }
// takeDoneKeyOnce is one take-in of a module's done key. // liveTakeInSteps are the steps on the controller's own stores and the bus.
func takeDoneKeyOnce(ctx context.Context, open *stores, module string) error { func liveTakeInSteps(open *stores, keys registerKeys, js jetstream.JetStream) takeInSteps {
keys, closeKeys, err := openBuilderKeys()
if err != nil {
return fmt.Errorf("the builder's keys cannot be read: %w", err)
}
defer closeKeys()
said, err := liveDoneKeySteps(open, keys).take(ctx, module)
if err == nil && said != "" {
fmt.Printf("done keys: %s\n", said)
}
return err
}
// liveDoneKeySteps are the steps on the controller's own stores and the bus.
func liveDoneKeySteps(open *stores, keys registerKeys) doneKeySteps {
inv := open.inventory inv := open.inventory
return doneKeySteps{ return takeInSteps{
held: func(ctx context.Context, module string) (bool, error) { catalogue: func(ctx context.Context) (map[string][]string, error) {
_, err := inv.SourceOf(ctx, module) entries, err := inv.Catalogued(ctx)
if errors.Is(err, inventory.ErrNoSuchModule) { if err != nil {
return false, nil return nil, err
} }
return err == nil, err out := make(map[string][]string, len(entries))
for _, e := range entries {
out[e.Manifest.Module] = e.On
}
return out, nil
}, },
trunk: func(ctx context.Context, module string) (sdkdelivery.Module, error) { trunk: func(ctx context.Context, module string) (sdkdelivery.Module, error) {
return keys.ModuleAt(ctx, module, "") return keys.ModuleAt(ctx, module, "")
}, },
registeredAt: func(ctx context.Context, module, commit string) (bool, error) { registered: func(ctx context.Context, module string, trunk sdkdelivery.Module) (bool, bool, error) {
src, err := inv.SourceOf(ctx, module) src, err := inv.SourceOf(ctx, module)
if err != nil { if err != nil {
return false, err return false, false, err
} }
return sameCommitID(src.BuiltFrom, commit), nil if sameCommitID(src.BuiltFrom, trunk.Commit) {
return true, true, nil
}
fp, err := registeredFingerprint(ctx, inv, keys, module, src.BuiltFrom)
return false, err == nil && fp != "" && fp == trunk.FP2, nil
},
recordCommit: func(ctx context.Context, module, commit string) error {
return inv.RecordBuiltFrom(ctx, module, commit)
}, },
takeIn: func(ctx context.Context, module string, trunk sdkdelivery.Module) error { takeIn: func(ctx context.Context, module string, trunk sdkdelivery.Module) error {
_, _, err := takeInKey(ctx, inv, keys, keyTake{Module: module, Trunk: trunk, Ask: registrationAsk{Merge: trunk.Commit}}) _, _, err := takeInKey(ctx, inv, keys, keyTake{Module: module, Trunk: trunk, Ask: registrationAsk{Merge: trunk.Commit}})
@@ -268,17 +342,8 @@ func liveDoneKeySteps(open *stores, keys registerKeys) doneKeySteps {
} }
return err return err
}, },
nodesOf: func(ctx context.Context, module string) ([]string, error) { taken: func(ctx context.Context, module, fp string, nodes []string) (bool, error) {
entries, err := inv.Catalogued(ctx) return composedAndGranted(ctx, js, module, fp, nodes)
if err != nil {
return nil, err
}
for _, e := range entries {
if e.Manifest.Module == module {
return e.On, nil
}
}
return nil, nil
}, },
busHolder: func(ctx context.Context) (string, error) { busHolder: func(ctx context.Context) (string, error) {
holders, err := seatHolders(ctx, inv) holders, err := seatHolders(ctx, inv)
@@ -289,107 +354,64 @@ func liveDoneKeySteps(open *stores, keys registerKeys) doneKeySteps {
}, },
// The composer's, never the take-in's own: one pass at a time composes each node asked for once, and writes the // The composer's, never the take-in's own: one pass at a time composes each node asked for once, and writes the
// grants once (issue 1945, composer.go). // grants once (issue 1945, composer.go).
compose: func(ctx context.Context, nodes []string) error { compose: func(ctx context.Context, nodes []string, grants bool) (map[string]error, error, error) {
return composerFor(open).compose(ctx, nodes, false) return composerFor(open).composeEach(ctx, nodes, grants)
},
grants: func(ctx context.Context) error {
return composerFor(open).compose(ctx, nil, true)
}, },
} }
} }
// watchDoneKeys wakes the take-in of a module's done key for each module the builder or the indexer writes, until ctx // registeredFingerprint is the fingerprint a module is registered at, as currentFingerprints reads it: its commit's
// ends: never on the watch's own goroutine. // newest build's, or the indexer's value at the commit when the commit has no build of its own; "" when neither says.
func watchDoneKeys(ctx context.Context, open *stores) { func registeredFingerprint(ctx context.Context, inv *inventory.Inventory, keys registerKeys, module, commit string) (string, error) {
keys, closeKeys, err := openBuilderKeys() if commit == "" {
if err != nil { return "", nil
fmt.Printf("done keys: the builder's keys are not watched: %v\n", err)
return
} }
if err := builderkeys.Watch(ctx, keys.JS, func(module string) { if id, err := inv.BuildOf(ctx, module, commit); err == nil && id != "" {
inBackground(func() { takeDoneKey(ctx, open, module) }) if b, found, err := inv.BuildByID(ctx, id); err == nil && found && b.SourceFingerprint != "" {
}); err != nil { return b.SourceFingerprint, nil
closeKeys()
fmt.Printf("done keys: the builder's keys are not watched: %v\n", err)
return
}
goServing(ctx, func() {
<-ctx.Done()
closeKeys()
})
// And every module once now, after the watch is open so no write falls between the two.
inBackground(func() {
doneKeysAtStart(ctx, catalogedModules(open), func(module string) { takeDoneKeyAtStart(ctx, open, module) })
})
}
// doneKeysAtStart takes in the done key of every catalogued module once, one after another: the start's look the watch of
// new writes cannot give (it hears updates only). A done key written while no controller served, or whose take-in the
// controller this one replaced was stopped in, is taken here; without it the module stays in the node's own composition
// with no credential until its keys are written again (issue 1945, live on 2026-10-25 at 02:34). A take-in with nothing
// to do does nothing; a catalogue that cannot be read is said.
func doneKeysAtStart(ctx context.Context, modules func(context.Context) ([]string, error), take func(module string)) {
names, err := modules(ctx)
if err != nil {
fmt.Printf("done keys: the modules are not looked at once at the start: %v\n", err)
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 {
break
} }
sem <- struct{}{}
wg.Add(1)
go func() {
defer func() { <-sem; wg.Done() }()
take(m)
}()
} }
wg.Wait() at, err := keys.ModuleAt(ctx, module, commit)
if err != nil {
return "", err
}
return at.FP2, nil
} }
// startTakeIns is how many take-ins the start's look runs at once. // composedAndGranted says the bus holds module's grant at fp and its composition at fp for every node: a few reads of
const startTakeIns = 16 // keys, never a composition.
func composedAndGranted(ctx context.Context, js jetstream.JetStream, module, fp string, nodes []string) (bool, error) {
// takeDoneKeyAtStart is one take-in of the start's look, under the module's lock; one that fails is handed to takeDoneKey, grants, err := js.KeyValue(ctx, broker.GrantsBucket)
// which tries again every doneKeyRetry, loudly, so a failing module never holds up the look at the others. if err != nil {
func takeDoneKeyAtStart(ctx context.Context, open *stores, module string) { return false, err
mu, _ := doneKeysMu.LoadOrStore(module, &sync.Mutex{})
if !mu.(*sync.Mutex).TryLock() {
// A take-in already follows the module: told to look again, it takes what the start would have (issue 1945).
doneKeysLookAgain.Store(module, true)
return
} }
doneKeysLookAgain.Delete(module) // what this look reads covers every wake before it key, err := sdkdelivery.ControllerGrantKey(module, fp)
err := doneKeyTakeOnce(ctx, open, module) if err != nil {
mu.(*sync.Mutex).Unlock() return false, err
// Read after letting go, as every holder of a module's lock does: a wake during this look only marked the module
// (the review of issue 1945).
_, again := doneKeysLookAgain.LoadAndDelete(module)
if err != nil || again {
// A failure, a wait or a wake meanwhile is followed by takeDoneKey, which says it and looks again.
inBackground(func() { takeDoneKey(ctx, open, module) })
} }
} if _, err := grants.Get(ctx, key); err != nil {
if errors.Is(err, jetstream.ErrKeyNotFound) {
// catalogedModules are the names of every module the catalogue holds. return false, nil
func catalogedModules(open *stores) func(context.Context) ([]string, error) { }
return func(ctx context.Context) ([]string, error) { return false, err
entries, err := open.inventory.Catalogued(ctx) }
digests, err := js.KeyValue(ctx, sdkdelivery.ControllerComposedBucket)
if err != nil {
return false, err
}
for _, node := range nodes {
key, err := sdkdelivery.ComposedKey(node, module, fp)
if err != nil { if err != nil {
return nil, err return false, err
} }
out := make([]string, 0, len(entries)) if _, err := digests.Get(ctx, key); err != nil {
for _, e := range entries { if errors.Is(err, jetstream.ErrKeyNotFound) {
out = append(out, e.Manifest.Module) return false, nil
}
return false, err
} }
return out, nil
} }
return true, nil
} }
// doneKeysComposers are the composers of the done keys' take-ins in this process, one for each set of stores (a test // doneKeysComposers are the composers of the done keys' take-ins in this process, one for each set of stores (a test
@@ -1,61 +0,0 @@
package main
import (
"context"
"fmt"
"testing"
"time"
"github.com/novox/mesh-controller/internal/conditions"
)
// A take-in cut by a cancellation raises no condition (issue 1945: on 2026-10-25 thirty urgent fingerprint-not-composed
// conditions were live, each a take-in a stop or a restart cut with "context canceled"): when the controller stops, the
// next one takes it in; when a step of its own was cancelled, it is tried again.
func TestATakeInCutByACancellationRaisesNothing(t *testing.T) {
for _, tc := range []struct {
name string
stop bool // the controller's context ends, else only the step's error is a cancellation
}{{"the controller stops", true}, {"a step was cancelled", false}} {
t.Run(tc.name, func(t *testing.T) {
was, wasRetry, wasSay, wasKeeper := doneKeyTakeOnce, doneKeyRetry, doneKeysSay, conditionsFrom
t.Cleanup(func() {
doneKeyTakeOnce, doneKeyRetry, doneKeysSay, conditionsFrom = was, wasRetry, wasSay, wasKeeper
})
store := conditions.NewInMemory()
k := conditions.NewKeeper(t.Context(), conditions.Options{Store: store, History: store,
Teller: &conditions.Told{}})
t.Cleanup(func() { k.Close(context.Background()) })
conditionsFrom = k
doneKeyRetry = time.Millisecond
doneKeysSay = func(string) {}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
calls := 0
doneKeyTakeOnce = func(context.Context, *stores, string) error {
calls++
if tc.stop {
cancel() // stopped mid-way
return fmt.Errorf("the module cannot be composed: %w", context.Canceled)
}
if calls == 1 {
return fmt.Errorf("the module cannot be composed: %w", context.Canceled)
}
return nil
}
done := make(chan struct{})
go func() { takeDoneKey(ctx, nil, "cut-module"); close(done) }()
select {
case <-done:
case <-time.After(5 * time.Second):
t.Fatal("the take-in did not end")
}
if !tc.stop && calls != 2 {
t.Errorf("a take-in whose step was cancelled was tried %d time(s), want 2", calls)
}
if c, found, err := conditions.ReadOne(t.Context(), store, doneKeyObservation("cut-module").Key()); err != nil || found {
t.Fatalf("a take-in cut by a cancellation raised %+v (found %v, %v)", c, found, err)
}
})
}
}
@@ -1,34 +0,0 @@
package main
import (
"context"
"errors"
"reflect"
"sort"
"sync"
"testing"
)
// A starting controller takes in the done key of every catalogued module once, beside the watch of new writes: a done key
// written while no controller served, or whose take-in the controller it replaced was stopped in, is otherwise never
// taken, and its module stays in the node's own composition with no credential (issue 1945, live on 2026-10-25 at 02:34:
// the controller replaced mid take-in left mesh-provisioner-client, gitea, postgres and many more unregistered at their
// new fingerprints, and the new one, watching updates only, never looked).
func TestAStartingControllerTakesEveryModulesDoneKeyOnce(t *testing.T) {
var took []string
var mu sync.Mutex
doneKeysAtStart(context.Background(), func(context.Context) ([]string, error) {
return []string{"mesh-provisioner-client", "gitea", "postgres"}, nil
}, func(module string) { mu.Lock(); took = append(took, module); mu.Unlock() })
sort.Strings(took)
if want := []string{"gitea", "mesh-provisioner-client", "postgres"}; !reflect.DeepEqual(took, want) {
t.Fatalf("took %v, want %v", took, want)
}
// A catalogue that cannot be read takes nothing and is said, never fatal.
took = nil
doneKeysAtStart(context.Background(), func(context.Context) ([]string, error) { return nil, errors.New("down") },
func(module string) { took = append(took, module) })
if len(took) != 0 {
t.Fatalf("took %v from a catalogue that could not be read", took)
}
}
+277 -68
View File
@@ -4,84 +4,305 @@ import (
"context" "context"
"encoding/json" "encoding/json"
"errors" "errors"
"fmt"
"slices"
"strings" "strings"
"testing" "testing"
"time"
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/nats-io/nats.go/jetstream"
"github.com/novox/mesh-controller/internal/builderkeys" "github.com/novox/mesh-controller/internal/builderkeys"
) )
// stepsFor are doneKeySteps over a module registered at registered, whose trunk value names fp at commit, assigned to // aTakeIn is a mesh for the take-in's steps: per module its registered commit, its trunk (commit, fingerprint), its
// nodes, the bus held on holder; every act is written to acts in order. // nodes, whether its compositions and grant exist, and what its take-in answers; every act is written to acts.
func stepsFor(acts *[]string, registered, commit, fp string, nodes []string, holder string) doneKeySteps { type aTakeIn struct {
return doneKeySteps{ registered map[string]string
held: func(_ context.Context, module string) (bool, error) { return module == "mesh-mail", nil }, sameFP map[string]bool // registered at another commit, at the trunk's fingerprint
trunk: func(context.Context, string) (sdkdelivery.Module, error) { trunks map[string]string // module -> commit; no entry: no trunk value
if fp == "" { nodes map[string][]string
taken map[string]bool
takeInErr map[string]error
composeErr map[string]error
grantsErr error
holder string
acts []string
}
func (m *aTakeIn) steps() takeInSteps {
fp := func(module string) string { return sdkdelivery.SourcePrefix2 + strings.Repeat(module[:1], 64) }
return takeInSteps{
catalogue: func(context.Context) (map[string][]string, error) {
m.acts = append(m.acts, "catalogue")
out := map[string][]string{}
for k, v := range m.nodes {
out[k] = v
}
return out, nil
},
trunk: func(_ context.Context, module string) (sdkdelivery.Module, error) {
commit, ok := m.trunks[module]
if !ok {
return sdkdelivery.Module{}, builderkeys.ErrNoValue return sdkdelivery.Module{}, builderkeys.ErrNoValue
} }
return sdkdelivery.Module{Commit: commit, FP2: fp}, nil return sdkdelivery.Module{Commit: commit, FP2: fp(module)}, nil
},
registered: func(_ context.Context, module string, trunk sdkdelivery.Module) (bool, bool, error) {
if m.registered[module] == trunk.Commit {
return true, true, nil
}
return false, m.sameFP[module], nil
},
recordCommit: func(_ context.Context, module, commit string) error {
m.acts = append(m.acts, "record "+module+" at "+commit)
return nil
}, },
registeredAt: func(_ context.Context, _ string, c string) (bool, error) { return c == registered, nil },
takeIn: func(_ context.Context, module string, trunk sdkdelivery.Module) error { takeIn: func(_ context.Context, module string, trunk sdkdelivery.Module) error {
*acts = append(*acts, "take in "+module+" at "+trunk.Commit) m.acts = append(m.acts, "take in "+module+" at "+trunk.Commit)
return nil return m.takeInErr[module]
}, },
nodesOf: func(context.Context, string) ([]string, error) { return nodes, nil }, taken: func(_ context.Context, module, _ string, _ []string) (bool, error) { return m.taken[module], nil },
busHolder: func(context.Context) (string, error) { return holder, nil }, busHolder: func(context.Context) (string, error) { return m.holder, nil },
compose: func(_ context.Context, ns []string) error { compose: func(_ context.Context, nodes []string, grants bool) (map[string]error, error, error) {
*acts = append(*acts, "compose "+strings.Join(ns, ",")) act := "compose " + strings.Join(nodes, ",")
return nil if grants {
act += " and grants"
}
m.acts = append(m.acts, act)
failed := map[string]error{}
for _, n := range nodes {
if e := m.composeErr[n]; e != nil {
failed[n] = e
}
}
if grants {
return failed, m.grantsErr, nil
}
return failed, nil, nil
}, },
grants: func(context.Context) error { *acts = append(*acts, "grants"); return nil },
} }
} }
// 1945/AC3 (unit half): a done key of a new fingerprint of an assigned module is taken in, then the module's nodes are // A batch of done keys takes each module in once and asks the composer at most twice, whatever its size: once for the
// composed, then the node holding the bus (its user list), then the grants: the grant and the composition exist // nodes of every module with something new, then once for the node holding the bus with the grants (ADR 0544 §5). A
// before the assigner's rules let a want name the fingerprint. A module registered at the trunk's commit already is // module registered at its trunk whose compositions and grant exist, or that this process took in at the same trunk,
// composed and granted without a second take-in; one assigned nowhere, unregistered, or with no trunk value is left // composes nothing (§3): the old take-in composed it three times over, every key and every start (issue 1945, 17:36 on
// alone. // 2026-10-25). A key delivered again is taken in whole. A wait for a write is acknowledged, a wait on a read and a
func TestADoneKeyIsTakenInComposedAndGranted(t *testing.T) { // failure refused, a failed build key acknowledged with its condition (§4).
fp := sdkdelivery.SourcePrefix2 + strings.Repeat("c", 64) func TestATakeInBatch(t *testing.T) {
var acts []string waitsForAWrite := fmt.Errorf("%w: the key is building", errWaitsForAWrite)
said, err := stepsFor(&acts, "old", "new", fp, []string{"ace", "g14"}, "novox").take(context.Background(), "mesh-mail") waitsOnARead := fmt.Errorf("%w: the store is away", errKeyWaits)
if err != nil { keyFailed := fmt.Errorf("%w: the build failed", errKeyFailed)
t.Fatal(err) type want struct {
verdict verdict
taken, nothingNew bool
waits, err bool
} }
if got := strings.Join(acts, "; "); got != "take in mesh-mail at new; compose ace,g14; compose novox; grants" { cases := []struct {
t.Errorf("the acts were %q", got) name string
mesh aTakeIn
keys []delivered
memo map[string]string
acts []string
passes int
want map[string]want
}{
{name: "a new fingerprint: taken in, its nodes, then the bus's node and the grants",
mesh: aTakeIn{registered: map[string]string{"mail": "old"}, trunks: map[string]string{"mail": "new"},
nodes: map[string][]string{"mail": {"ace", "g14", "novox"}}, holder: "novox"},
keys: []delivered{{module: "mail"}},
acts: []string{"catalogue", "take in mail at new", "compose ace,g14", "compose novox and grants"}, passes: 2,
want: map[string]want{"mail": {taken: true}}},
{name: "registered, composed and granted: nothing composed (fails on the old take-in)",
mesh: aTakeIn{registered: map[string]string{"mail": "new", "nats": "n"}, trunks: map[string]string{"mail": "new", "nats": "n"},
nodes: map[string][]string{"mail": {"ace"}, "nats": {"novox"}}, taken: map[string]bool{"mail": true, "nats": true},
holder: "novox"},
keys: []delivered{{module: "mail"}, {module: "nats"}, {module: "mail"}},
acts: []string{"catalogue"},
want: map[string]want{"mail": {nothingNew: true}, "nats": {nothingNew: true}}},
{name: "a catalogue merge that moved the commit and not the fingerprint: recorded, nothing composed",
mesh: aTakeIn{registered: map[string]string{"mail": "a85d9037"}, sameFP: map[string]bool{"mail": true},
trunks: map[string]string{"mail": "2df05030"}, nodes: map[string][]string{"mail": {"ace"}},
taken: map[string]bool{"mail": true}, holder: "novox"},
keys: []delivered{{module: "mail"}},
acts: []string{"catalogue", "record mail at 2df05030"}, want: map[string]want{"mail": {nothingNew: true}}},
{name: "taken in by this process at the same trunk: nothing new, not even a look at its keys",
mesh: aTakeIn{registered: map[string]string{"mail": "new"}, trunks: map[string]string{"mail": "new"},
nodes: map[string][]string{"mail": {"ace"}}, holder: "novox"},
keys: []delivered{{module: "mail"}},
memo: map[string]string{"mail": sdkdelivery.SourcePrefix2 + strings.Repeat("m", 64) + "@new"},
acts: []string{"catalogue"}, want: map[string]want{"mail": {nothingNew: true}}},
{name: "delivered again: taken in whole, even when it looks done",
mesh: aTakeIn{registered: map[string]string{"mail": "new"}, trunks: map[string]string{"mail": "new"},
nodes: map[string][]string{"mail": {"ace"}}, taken: map[string]bool{"mail": true}, holder: "novox"},
keys: []delivered{{module: "mail", again: true}},
acts: []string{"catalogue", "compose ace", "compose novox and grants"}, passes: 2,
want: map[string]want{"mail": {taken: true}}},
{name: "a take-in that stopped half-way: composed",
mesh: aTakeIn{registered: map[string]string{"mail": "new"}, trunks: map[string]string{"mail": "new"},
nodes: map[string][]string{"mail": {"ace"}}, holder: "novox"},
keys: []delivered{{module: "mail"}},
acts: []string{"catalogue", "compose ace", "compose novox and grants"}, passes: 2,
want: map[string]want{"mail": {taken: true}}},
{name: "three modules, one pass for their nodes and one for the bus",
mesh: aTakeIn{registered: map[string]string{}, trunks: map[string]string{"mail": "1", "web": "2", "bus": "3"},
nodes: map[string][]string{"mail": {"ace", "novox"}, "web": {"g14", "ace"}, "bus": {"novox"}}, holder: "novox"},
keys: []delivered{{module: "mail"}, {module: "web"}, {module: "bus"}},
acts: []string{"catalogue", "take in mail at 1", "take in web at 2", "take in bus at 3", "compose ace,g14",
"compose novox and grants"}, passes: 2,
want: map[string]want{"mail": {taken: true}, "web": {taken: true}, "bus": {taken: true}}},
{name: "assigned nowhere, held by no catalogue, no trunk value: nothing composed",
mesh: aTakeIn{registered: map[string]string{}, trunks: map[string]string{"mail": "1", "odd": "2"},
nodes: map[string][]string{"mail": nil, "none": {"ace"}}, holder: "novox"},
keys: []delivered{{module: "mail"}, {module: "odd"}, {module: "none"}},
acts: []string{"catalogue", "take in mail at 1"},
want: map[string]want{"mail": {taken: true}, "odd": {}, "none": {}}},
{name: "a grant not written refuses the key, never swallowed",
mesh: aTakeIn{registered: map[string]string{}, trunks: map[string]string{"mail": "1"},
nodes: map[string][]string{"mail": {"ace"}}, holder: "novox", grantsErr: errors.New("the bucket mesh-controller_grants: not found")},
keys: []delivered{{module: "mail"}},
acts: []string{"catalogue", "take in mail at 1", "compose ace", "compose novox and grants"}, passes: 2,
want: map[string]want{"mail": {verdict: refused, err: true}}},
{name: "a node that fails refuses only its modules",
mesh: aTakeIn{registered: map[string]string{}, trunks: map[string]string{"mail": "1", "web": "2"},
nodes: map[string][]string{"mail": {"ace"}, "web": {"g14"}}, holder: "novox",
composeErr: map[string]error{"ace": errors.New("store refused")}},
keys: []delivered{{module: "mail"}, {module: "web"}},
acts: []string{"catalogue", "take in mail at 1", "take in web at 2", "compose ace,g14", "compose novox and grants"},
passes: 2, want: map[string]want{"mail": {verdict: refused, err: true}, "web": {taken: true}}},
{name: "waits and failures: a write's wait acknowledged, a read's refused, a failed key acknowledged",
mesh: aTakeIn{registered: map[string]string{}, trunks: map[string]string{"a": "1", "b": "2", "c": "3"},
nodes: map[string][]string{"a": {"ace"}, "b": {"ace"}, "c": {"ace"}}, holder: "novox",
takeInErr: map[string]error{"a": waitsForAWrite, "b": waitsOnARead, "c": keyFailed}},
keys: []delivered{{module: "a"}, {module: "b"}, {module: "c"}},
acts: []string{"catalogue", "take in a at 1", "take in b at 2", "take in c at 3"},
want: map[string]want{"a": {waits: true}, "b": {verdict: refused, waits: true}, "c": {err: true}}},
} }
if !strings.Contains(said, "registered from the builder's done key") { for _, c := range cases {
t.Errorf("said %q", said) t.Run(c.name, func(t *testing.T) {
memo := c.memo
if memo == nil {
memo = map[string]string{}
}
settled, passes := c.mesh.steps().batch(t.Context(), c.keys, memo)
if !slices.Equal(c.mesh.acts, c.acts) {
t.Fatalf("the acts were\n %q\nwant\n %q", c.mesh.acts, c.acts)
}
if passes != c.passes {
t.Fatalf("%d composer pass(es), want %d", passes, c.passes)
}
if len(settled) != len(c.want) {
t.Fatalf("settled %d module(s), want %d: %+v", len(settled), len(c.want), settled)
}
for _, r := range settled {
w := c.want[r.module]
got := want{verdict: r.verdict, taken: r.taken, nothingNew: r.nothingNew, waits: r.waits != "", err: r.err != nil}
if got != w {
t.Errorf("%s: %+v, want %+v (%v)", r.module, got, w, r.err)
}
if r.taken && memo[r.module] == "" {
t.Errorf("%s was taken in and not remembered", r.module)
}
}
})
} }
acts = nil // And each delivered key is acknowledged or refused by its module's verdict, the refused with the retry's delay.
if _, err := stepsFor(&acts, "new", "new", fp, []string{"novox"}, "novox").take(context.Background(), "mesh-mail"); err != nil { t.Run("acknowledged and refused", func(t *testing.T) {
t.Fatal(err) mesh := aTakeIn{registered: map[string]string{}, trunks: map[string]string{"a": "1", "b": "2"},
} nodes: map[string][]string{"a": {"ace"}, "b": {"ace"}}, takeInErr: map[string]error{"b": waitsOnARead}}
if got := strings.Join(acts, "; "); got != "compose novox; grants" { var acked, naked []string
t.Errorf("a module registered at its trunk already: %q", got) msg := func(module string) takeInMsg {
} return takeInMsg{delivered: delivered{module: module, deliveries: 3}, ack: func() { acked = append(acked, module) },
nak: func(d time.Duration) {
if d != refusalDelay(3) {
t.Errorf("refused with a delay of %s", d)
}
naked = append(naked, module)
}}
}
state := &takeInState{memo: map[string]string{}, raised: map[string]bool{}, saidWait: map[string]string{}}
said := state.take(t.Context(), mesh.steps(), []takeInMsg{msg("a"), msg("b"), msg("a")})
if !slices.Equal(acked, []string{"a", "a"}) || !slices.Equal(naked, []string{"b"}) {
t.Fatalf("acknowledged %v and refused %v", acked, naked)
}
if !strings.Contains(said, "1 taken in") || !strings.Contains(said, "1 waiting") {
t.Fatalf("said %q", said)
}
})
for name, s := range map[string]doneKeySteps{ // A key refused for long holds no slot the others need: each consumer leaves far more keys unacknowledged than a
"assigned nowhere": stepsFor(&acts, "old", "new", fp, nil, "novox"), // batch takes, so 8 keys refused on each stop no ninth (the review of ADR 0544: with 8 a consumer, they stopped it
"no trunk value": stepsFor(&acts, "old", "new", "", []string{"ace"}, "novox"), // for good), and a refusal waits longer each time, up to ten minutes.
t.Run("refused keys stop no other", func(t *testing.T) {
for _, bucket := range []string{sdkdelivery.ModulesBucket, sdkdelivery.BuilderBuildsBucket} {
if c := takeInConfig(bucket); c.MaxAckPending < 8+startTakeIns || c.AckPolicy != jetstream.AckExplicitPolicy {
t.Errorf("%s: %d unacknowledged at most (%v): 8 refused keys and a batch do not fit", bucket,
c.MaxAckPending, c.AckPolicy)
}
}
was := time.Duration(0)
for n := uint64(1); n <= 8; n++ {
d := refusalDelay(n)
if d < was || d > doneKeyRetryMost || (n == 1 && d != doneKeyRetry) {
t.Fatalf("the %d. refusal waits %s (the one before %s)", n, d, was)
}
was = d
}
if was != doneKeyRetryMost {
t.Fatalf("the 8. refusal waits %s, want %s", was, doneKeyRetryMost)
}
})
}
// The take-in stalls when it holds work (keys not delivered yet, or a batch handed and not settled) and settles none
// for five minutes; refused keys waiting for their delay are no work, so a quiet mesh is no stall. The first look
// always says what it found, so a stall raised before a restart is cleared once this controller moves (ADR 0544 §6
// and its second review).
func TestATakeInStallsWhenItHoldsWorkAndSettlesNone(t *testing.T) {
at := time.Date(2026, 10, 25, 17, 36, 0, 0, time.UTC)
type reading struct {
after time.Duration
pending, taking uint64
handed, settled uint64
stalled, changed bool
}
for _, c := range []struct {
name string
readings []reading
}{
{"keys not delivered and nothing settled", []reading{
{0, 190, 8, 50, 50, false, true}, // the first look always says what it found
{4 * time.Minute, 190, 8, 50, 50, false, false},
{5 * time.Minute, 190, 8, 50, 50, true, true},
{6 * time.Minute, 190, 8, 50, 50, true, false},
{7 * time.Minute, 180, 8, 60, 60, false, true},
{20 * time.Minute, 0, 0, 60, 60, false, false},
}},
{"refused keys waiting alone are no stall", []reading{
{0, 0, 3, 40, 40, false, true},
{30 * time.Minute, 0, 3, 40, 40, false, false},
}},
{"every key settled, one of them refused by the bus, nothing pending: no stall", []reading{
{0, 0, 1, 3, 3, false, true},
{6 * time.Minute, 0, 1, 3, 3, false, false},
}},
{"a batch handed and never settled is a stall", []reading{
{0, 0, 16, 56, 40, false, true},
{5 * time.Minute, 0, 16, 56, 40, true, true},
}},
} { } {
acts = nil t.Run(c.name, func(t *testing.T) {
if _, err := s.take(context.Background(), "mesh-mail"); err != nil { w := &stallWatch{}
t.Fatal(err) for i, r := range c.readings {
} stalled, changed := w.look(takeInBacklog{Pending: r.pending, Taking: r.taking}, r.handed, r.settled,
if name == "assigned nowhere" && strings.Join(acts, "; ") != "take in mesh-mail at new" { at.Add(r.after))
t.Errorf("%s: %v", name, acts) if stalled != r.stalled || changed != r.changed {
} t.Fatalf("reading %d: stalled %v changed %v, want %v %v", i, stalled, changed, r.stalled, r.changed)
if name == "no trunk value" && len(acts) != 0 { }
t.Errorf("%s: %v", name, acts) }
} })
}
acts = nil
if _, err := stepsFor(&acts, "old", "new", fp, []string{"ace"}, "novox").take(context.Background(), "unknown"); err != nil || len(acts) != 0 {
t.Errorf("a module the catalogue does not hold: %v %v", acts, err)
} }
} }
@@ -115,15 +336,3 @@ func TestAChangedUserListIsANewRevisionOfTheBusModulesCompositionAlone(t *testin
} }
} }
} }
// 1945 (the review's D1): a grant that is not written is the take-in's error, never swallowed: with no send left,
// nothing else would write it, and no want could name the fingerprint.
func TestAGrantNotWrittenIsTheTakeInsError(t *testing.T) {
fp := sdkdelivery.SourcePrefix2 + strings.Repeat("c", 64)
var acts []string
s := stepsFor(&acts, "old", "new", fp, []string{"ace"}, "novox")
s.grants = func(context.Context) error { return errors.New("the bucket mesh-controller_grants: not found") }
if _, err := s.take(context.Background(), "mesh-mail"); err == nil || !strings.Contains(err.Error(), "grants") {
t.Errorf("a grant not written: %v", err)
}
}
@@ -1,184 +0,0 @@
package main
import (
"context"
"fmt"
"strings"
"sync"
"testing"
"time"
)
// A take-in that waits is said once for each reason it waits for, and tried again every doneKeyRetry until it is taken
// in: never left for a key write that may not come (issue 1945, live on 2026-10-25: mesh-host's take-in at 26bd waited,
// said nothing, and nothing woke it until another write).
func TestAWaitingTakeInIsSaidOnceAndTriedAgain(t *testing.T) {
was, wasRetry, wasSay := doneKeyTakeOnce, doneKeyRetry, doneKeysSay
t.Cleanup(func() { doneKeyTakeOnce, doneKeyRetry, doneKeysSay = was, wasRetry, wasSay })
doneKeyRetry = 5 * time.Millisecond
var mu sync.Mutex
var said []string
doneKeysSay = func(s string) { mu.Lock(); said = append(said, s); mu.Unlock() }
calls := 0
doneKeyTakeOnce = func(context.Context, *stores, string) error {
calls++
switch {
case calls <= 2:
return fmt.Errorf("%w: the artifact store could not hold mesh-host's artifact", errKeyWaits)
case calls == 3:
return fmt.Errorf("%w: mesh-host's key cannot be read", errKeyWaits)
}
return nil
}
done := make(chan struct{})
go func() { takeDoneKey(context.Background(), nil, "mesh-host"); close(done) }()
select {
case <-done:
case <-time.After(5 * time.Second):
t.Fatal("a waiting take-in was not tried again until it was taken in")
}
if calls != 4 {
t.Fatalf("tried %d times, want 4", calls)
}
mu.Lock()
defer mu.Unlock()
waits := 0
for _, s := range said {
if strings.Contains(s, "mesh-host waits") {
waits++
}
}
if waits != 2 {
t.Fatalf("said %q: want each of the two reasons once", said)
}
}
// A wake while a take-in of the module already follows it adds no second one: the follower looks again itself.
func TestAWakeWhileATakeInFollowsAddsNone(t *testing.T) {
was, wasRetry, wasSay := doneKeyTakeOnce, doneKeyRetry, doneKeysSay
t.Cleanup(func() { doneKeyTakeOnce, doneKeyRetry, doneKeysSay = was, wasRetry, wasSay })
doneKeyRetry = 20 * time.Millisecond
doneKeysSay = func(string) {}
var mu sync.Mutex
calls := 0
release := make(chan struct{})
doneKeyTakeOnce = func(context.Context, *stores, string) error {
mu.Lock()
calls++
mu.Unlock()
select {
case <-release:
return nil
default:
return fmt.Errorf("%w: the key says building", errKeyWaits)
}
}
follower := make(chan struct{})
go func() { takeDoneKey(context.Background(), nil, "board"); close(follower) }()
time.Sleep(5 * time.Millisecond)
returned := make(chan struct{})
go func() { takeDoneKey(context.Background(), nil, "board"); close(returned) }()
select {
case <-returned:
case <-time.After(time.Second):
t.Fatal("a second wake waited for the follower")
}
close(release)
<-follower
}
// A wake while a take-in of the module follows it is not lost: the follower goes round once more after it succeeds, so
// a newer build's key written meanwhile is taken in (the review of issue 1945).
func TestAWakeDuringATakeInIsTakenAfterIt(t *testing.T) {
was, wasRetry, wasSay := doneKeyTakeOnce, doneKeyRetry, doneKeysSay
t.Cleanup(func() { doneKeyTakeOnce, doneKeyRetry, doneKeysSay = was, wasRetry, wasSay })
doneKeyRetry = 20 * time.Millisecond
doneKeysSay = func(string) {}
var mu sync.Mutex
calls := 0
inFirst := make(chan struct{})
release := make(chan struct{})
doneKeyTakeOnce = func(context.Context, *stores, string) error {
mu.Lock()
calls++
n := calls
mu.Unlock()
if n == 1 {
close(inFirst)
<-release
}
return nil
}
follower := make(chan struct{})
go func() { takeDoneKey(context.Background(), nil, "mesh-host"); close(follower) }()
<-inFirst
takeDoneKey(context.Background(), nil, "mesh-host") // the newer key's wake, while the first runs
close(release)
<-follower
mu.Lock()
defer mu.Unlock()
if calls != 2 {
t.Fatalf("taken %d times: the wake during the take-in was lost", calls)
}
}
// A wake that comes after the follower's last look but before it lets go of the module is not lost: the follower looks
// at the mark again once it has let go, and takes the module once more (the review of issue 1945).
func TestAWakeJustBeforeTheFollowerLetsGoIsTaken(t *testing.T) {
was, wasRetry, wasSay, wasHook := doneKeyTakeOnce, doneKeyRetry, doneKeysSay, doneKeysBeforeLetGo
t.Cleanup(func() {
doneKeyTakeOnce, doneKeyRetry, doneKeysSay, doneKeysBeforeLetGo = was, wasRetry, wasSay, wasHook
})
doneKeyRetry = 10 * time.Millisecond
doneKeysSay = func(string) {}
var mu sync.Mutex
calls := 0
doneKeyTakeOnce = func(context.Context, *stores, string) error {
mu.Lock()
calls++
mu.Unlock()
return nil
}
woke := false
doneKeysBeforeLetGo = func(module string) {
if !woke {
woke = true
takeDoneKey(context.Background(), nil, module) // fails its TryLock and marks the module
}
}
takeDoneKey(context.Background(), nil, "mesh-host")
mu.Lock()
defer mu.Unlock()
if calls != 2 {
t.Fatalf("taken %d times: the wake just before the follower let go was lost", calls)
}
}
// A wake during the start's take-in of a module is not lost: the start pass reads the mark once it lets go of the module,
// and hands the module to takeDoneKey, which takes it once more (the review of issue 1945: the third dropped wake of its
// kind).
func TestAWakeDuringTheStartsTakeInIsTaken(t *testing.T) {
was, wasRetry, wasSay := doneKeyTakeOnce, doneKeyRetry, doneKeysSay
t.Cleanup(func() { doneKeyTakeOnce, doneKeyRetry, doneKeysSay = was, wasRetry, wasSay })
doneKeyRetry = 10 * time.Millisecond
doneKeysSay = func(string) {}
var mu sync.Mutex
calls := 0
doneKeyTakeOnce = func(context.Context, *stores, string) error {
mu.Lock()
calls++
n := calls
mu.Unlock()
if n == 1 {
takeDoneKey(context.Background(), nil, "mesh-host") // the newer key's wake, during the start's take-in
}
return nil
}
takeDoneKeyAtStart(context.Background(), nil, "mesh-host")
background.Wait()
mu.Lock()
defer mu.Unlock()
if calls != 2 {
t.Fatalf("taken %d times: the wake during the start's take-in was lost", calls)
}
}
+1 -1
View File
@@ -152,7 +152,7 @@ func watchHavesForJudgement(ctx context.Context, open *stores) {
if err != nil || isNode { if err != nil || isNode {
continue continue
} }
inBackground(func() { inBackground(ctx, func() {
if err := rejudgeOn(ctx, open, []string{node}, module); err != nil { 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) fmt.Printf("compositions: %s on %s is not judged again: %v\n", module, node, err)
} }
+26 -9
View File
@@ -135,9 +135,18 @@ func declarationsDone(nodes []string, runs map[string]string, current string) bo
return true return true
} }
// declarationsRetiredOn reads whether the declarations are done on the mesh now: every node's have of the node-engine // declarationsRetiredOn reads whether the declarations are done on the mesh now: their stream gone from the bus, or
// against the fingerprint it is registered at. // 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) { 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) fps, err := currentFingerprints(ctx, inv)
if err != nil { if err != nil {
return false, err 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 // 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) { func retireDeclarations(ctx context.Context, open *stores) (bool, error) {
if _, err := theLease.epoch(ctx); err != nil { if _, err := theLease.epoch(ctx); err != nil {
return false, err return false, err
@@ -306,26 +316,33 @@ func composeReplacedController(ctx context.Context, open *stores) error {
// replacedComposition is the controller's composition on r's node at r's fingerprint: in composed with r's manifest // replacedComposition is the controller's composition on r's node at r's fingerprint: in composed with r's manifest
// in place of the controller's registered one. // in place of the controller's registered one.
func replacedComposition(in compose.Input, r replacedController) (sdkdelivery.Composed, error) { func replacedComposition(in compose.Input, r replacedController) (sdkdelivery.Composed, error) {
return keptComposition(in, r.node, controllerModule, r.fp, r.manifest)
}
// keptComposition is module's composition on node at fp: in composed with manifest, that fingerprint's own, in place of
// the module's registered one, in the reading mode (nothing made). in is not changed.
func keptComposition(in compose.Input, node, module, fp string, manifest catalogue.Manifest) (sdkdelivery.Composed, error) {
shelf := make(map[string]catalogue.Manifest, len(in.Catalogue)) shelf := make(map[string]catalogue.Manifest, len(in.Catalogue))
for k, v := range in.Catalogue { for k, v := range in.Catalogue {
shelf[k] = v shelf[k] = v
} }
shelf[controllerModule] = r.manifest shelf[module] = manifest
in.Catalogue = shelf in.Catalogue = shelf
all, _, err := compose.Compositions(in, r.node, map[string]string{controllerModule: r.fp}, compose.Reading) all, _, err := compose.Compositions(in, node, map[string]string{module: fp}, compose.Reading)
if err != nil { if err != nil {
return sdkdelivery.Composed{}, err return sdkdelivery.Composed{}, err
} }
key, _ := sdkdelivery.ComposedKey(r.node, controllerModule, r.fp) key, _ := sdkdelivery.ComposedKey(node, module, fp)
c, ok := all[key] c, ok := all[key]
if !ok { if !ok {
return sdkdelivery.Composed{}, fmt.Errorf("%s composes no %s with the replaced build's manifest", r.node, controllerModule) return sdkdelivery.Composed{}, fmt.Errorf("%s composes no %s with the manifest of %s", node, module, short(fp))
} }
return c, nil 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 // declarationsRetiredNow says the declarations are done on the mesh now, for a raise of the bus's objects and for what
// store it cannot read is not done, so the stream is asserted as before. // 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 { func declarationsRetiredNow(ctx context.Context, inv *inventory.Inventory) bool {
bus, err := aBus() bus, err := aBus()
if err != nil { if err != nil {
+117 -44
View File
@@ -3,6 +3,8 @@ package main
import ( import (
"context" "context"
"encoding/json" "encoding/json"
"fmt"
"slices"
"strings" "strings"
"testing" "testing"
@@ -45,51 +47,97 @@ func TestTheFirstStartDeletesThePlansBucketsAndSaysEachOnce(t *testing.T) {
} }
} }
// 1945/AC6 (unit half): the controller this one replaced is composed on its node from its own build's manifest at its // An earlier fingerprint is composed from its own build's manifest with today's inputs: the controller this one
// own fingerprint: the composition holds what that manifest declares, not what the controller registered now declares. // replaced (1945/AC6), and every kept composition of a module whose config changed (ADR 0545: on 2026-10-25 a want back
func TestTheReplacedControllerIsComposedFromItsOwnManifest(t *testing.T) { // to an earlier mesh-build-go on g14 applied the capacity of 0 its kept composition was made with). The composition
now := catalogue.Manifest{Module: controllerModule, Version: "6", Resources: []map[string]any{ // holds what that manifest declares and today's config, the kept keys are those of the touched modules at a
{"id": "settings", "type": "file", "path": "/etc/mesh-controller/settings", "content": "six\n"}}} // fingerprint other than the registered one, the build of a fingerprint is its newest that worked, and a write of
before := catalogue.Manifest{Module: controllerModule, Version: "5", Resources: []map[string]any{ // mesh-config's resolved values names the node and module it touches.
{"id": "settings", "type": "file", "path": "/etc/mesh-controller/settings", "content": "five\n"}}} func TestAnEarlierFingerprintIsComposedFromItsOwnManifest(t *testing.T) {
in := compose.Input{Catalogue: map[string]catalogue.Manifest{controllerModule: now}, fp := func(c string) string { return sdkdelivery.SourcePrefix2 + strings.Repeat(c, 64) }
Nodes: []compose.Node{{Name: "anchor", Assigned: []string{controllerModule}, Stated: true}}} t.Run("from its own manifest, with today's inputs", func(t *testing.T) {
fp := sdkdelivery.SourcePrefix2 + strings.Repeat("5", 64) manifest := func(version string) catalogue.Manifest {
c, err := replacedComposition(in, replacedController{node: "anchor", fp: fp, manifest: before}) return catalogue.Manifest{Module: controllerModule, Version: version,
if err != nil { Settings: map[string]catalogue.SettingDeclaration{
t.Fatal(err) "capacity": {Kind: catalogue.KindPreference, Default: float64(0), Why: "how many builds at once"}},
} Resources: []map[string]any{{"id": "settings", "type": "file", "path": "/etc/mesh-controller/settings",
if c.Module != controllerModule || c.FP != fp || c.Node != "anchor" || len(c.Resources) != 1 { "content": version + " ${setting:capacity}\n"}}}
t.Fatalf("composed %s at %s on %s with %d resources", c.Module, c.FP, c.Node, len(c.Resources)) }
} now, before := manifest("six"), manifest("five")
var f struct{ Content string } in := compose.Input{Catalogue: map[string]catalogue.Manifest{controllerModule: now},
_ = json.Unmarshal(c.Resources[0], &f) Nodes: []compose.Node{{Name: "anchor", Assigned: []string{controllerModule}, Stated: true,
if f.Content != "five\n" { Settings: map[string][]catalogue.Layer{controllerModule: {{From: "mesh", Values: map[string]any{"capacity": float64(4)}}}}}}}
t.Errorf("the replaced controller's composition holds %q, its own manifest says five", f.Content) for name, compose := range map[string]func() (sdkdelivery.Composed, error){
} "the replaced controller": func() (sdkdelivery.Composed, error) {
if in.Catalogue[controllerModule].Version != "6" { return replacedComposition(in, replacedController{node: "anchor", fp: fp("5"), manifest: before})
t.Errorf("composing the replaced controller changed the Input's catalogue") },
} "a kept composition": func() (sdkdelivery.Composed, error) {
} return keptComposition(in, "anchor", controllerModule, fp("5"), before)
},
} {
c, err := compose()
if err != nil {
t.Fatalf("%s: %v", name, err)
}
if c.Module != controllerModule || c.FP != fp("5") || c.Node != "anchor" || len(c.Resources) != 1 {
t.Fatalf("%s: composed %s at %s on %s with %d resources", name, c.Module, c.FP, c.Node, len(c.Resources))
}
var f struct{ Content string }
_ = json.Unmarshal(c.Resources[0], &f)
if f.Content != "five 4\n" {
t.Errorf("%s holds %q, want its own manifest's five with today's capacity 4", name, f.Content)
}
}
if in.Catalogue[controllerModule].Version != "six" {
t.Errorf("composing an earlier fingerprint changed the Input's catalogue")
}
})
// 1945/AC6 (unit half): the controller replaced is the newest earlier build of the controller that worked and names a t.Run("which build, which keys, which touch", func(t *testing.T) {
// fingerprint of the new form, not the one registered now. m := []byte(`{"module":"mesh-controller","version":"5"}`)
func TestTheReplacedControllerIsItsNewestEarlierBuild(t *testing.T) { builds := []inventory.Build{
now, before, older := sdkdelivery.SourcePrefix2+strings.Repeat("6", 64), sdkdelivery.SourcePrefix2+strings.Repeat("5", 64), {ID: "c6", SourceFingerprint: fp("6"), Manifest: m},
sdkdelivery.SourcePrefix2+strings.Repeat("4", 64) {ID: "c5-failed", SourceFingerprint: fp("5"), Failed: "no", Manifest: m},
m := []byte(`{"module":"mesh-controller","version":"5"}`) {ID: "c5", SourceFingerprint: fp("5"), Manifest: m},
builds := []inventory.Build{ {ID: "c4", SourceFingerprint: fp("4"), Manifest: m},
{ID: "c6", SourceFingerprint: now, Manifest: m}, }
{ID: "c5-failed", SourceFingerprint: before, Failed: "no", Manifest: m}, if b, ok := replacedBuild(builds, fp("6")); !ok || b.ID != "c5" {
{ID: "c5", SourceFingerprint: before, Manifest: m}, t.Errorf("the replaced build is %+v (%v), want c5", b, ok)
{ID: "c4", SourceFingerprint: older, Manifest: m}, }
} if _, ok := replacedBuild(builds[:1], fp("6")); ok {
if b, ok := replacedBuild(builds, now); !ok || b.ID != "c5" { t.Errorf("a controller with no earlier build replaced one")
t.Errorf("the replaced build is %+v (%v), want c5", b, ok) }
} if b, ok := buildAt(builds, fp("5")); !ok || b.ID != "c5" {
if _, ok := replacedBuild(builds[:1], now); ok { t.Errorf("the build at 5 is %+v (%v), want c5, the one that worked", b, ok)
t.Errorf("a controller with no earlier build replaced one") }
} if _, ok := buildAt(builds, fp("3")); ok {
t.Errorf("a fingerprint with no build has one")
}
key := func(node, module, f string) string { k, _ := sdkdelivery.ComposedKey(node, module, f); return k }
keys := []string{key("g14", "mesh-build-go", fp("2")), key("g14", "mesh-build-go", fp("1")),
key("g14", "mesh-build-go", fp("3")), key("g14", "hello-web", fp("1")), key("ace", "mesh-build-go", fp("1"))}
registered := map[string]string{"mesh-build-go": fp("3"), "hello-web": fp("9")}
got := keptToFollow(keys, "g14", map[string]bool{"mesh-build-go": true}, registered)
if want := []keptKey{{"mesh-build-go", fp("1")}, {"mesh-build-go", fp("2")}}; !slices.Equal(got, want) {
t.Errorf("kept to follow %v, want %v", got, want)
}
if got := keptToFollow(keys, "g14", nil, registered); len(got) != 3 {
t.Errorf("every module's kept compositions: %v", got)
}
for k, want := range map[string][2]string{"g14.mesh-build-go.capacity": {"g14", "mesh-build-go"},
"g14.mesh-build-go.limits.cpu": {"g14", "mesh-build-go"}, "g14.mesh-build-go": {}, "": {}} {
node, module, ok := configTouched(k)
if [2]string{node, module} != want || ok != (want[0] != "") {
t.Errorf("%q touches %s %s (%v), want %v", k, node, module, ok, want)
}
}
// A value set before the consumer was made reaches the kept compositions: it reads each key once at creation.
if c := configFollowConfig(); c.DeliverPolicy != jetstream.DeliverLastPerSubjectPolicy ||
c.AckPolicy != jetstream.AckExplicitPolicy {
t.Errorf("the config consumer delivers %v, acknowledged %v; want each key's last value, acknowledged",
c.DeliverPolicy, c.AckPolicy)
}
})
} }
// 1945/AC14 (unit half): the declarations' stream stays while one node's have names another node-engine than the one // 1945/AC14 (unit half): the declarations' stream stays while one node's have names another node-engine than the one
@@ -110,3 +158,28 @@ func TestTheDeclarationsGoOnlyOnceEveryNodeRunsTheNewEngine(t *testing.T) {
t.Errorf("done with no fingerprint registered, or no node") 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)
}
}
+32 -10
View File
@@ -29,7 +29,13 @@ type composer struct {
type waiter struct { type waiter struct {
nodes []string nodes []string
grants bool grants bool
done chan error done chan passResult
}
// passResult is a waiter's answer: the error of each of its nodes that failed, and of the grants when it asked for them.
type passResult struct {
failed map[string]error
grants error
} }
func newComposer(compose func(context.Context, []string) map[string]error, grants func(context.Context) error) *composer { func newComposer(compose func(context.Context, []string) map[string]error, grants func(context.Context) error) *composer {
@@ -39,7 +45,23 @@ func newComposer(compose func(context.Context, []string) map[string]error, grant
// compose asks for the nodes to be composed, and the grants written after when grants is true, and waits for the pass // compose asks for the nodes to be composed, and the grants written after when grants is true, and waits for the pass
// that does it, with the errors of its own nodes alone. // that does it, with the errors of its own nodes alone.
func (c *composer) compose(ctx context.Context, nodes []string, grants bool) error { func (c *composer) compose(ctx context.Context, nodes []string, grants bool) error {
done := make(chan error, 1) failed, grantsErr, err := c.composeEach(ctx, nodes, grants)
if err != nil {
return err
}
var errs []error
for _, n := range nodes {
if e := failed[n]; e != nil {
errs = append(errs, e)
}
}
return errors.Join(append(errs, grantsErr)...)
}
// composeEach is compose answering each node's error apart, and the grants' (ADR 0544 §5: a batch of take-ins asks one
// pass and settles each module by its own nodes); err is ctx's when it ends first.
func (c *composer) composeEach(ctx context.Context, nodes []string, grants bool) (map[string]error, error, error) {
done := make(chan passResult, 1)
c.mu.Lock() c.mu.Lock()
for _, n := range nodes { for _, n := range nodes {
c.pending[n] = true c.pending[n] = true
@@ -52,10 +74,10 @@ func (c *composer) compose(ctx context.Context, nodes []string, grants bool) err
default: default:
} }
select { select {
case err := <-done: case r := <-done:
return err return r.failed, r.grants, nil
case <-ctx.Done(): case <-ctx.Done():
return ctx.Err() return nil, nil, ctx.Err()
} }
} }
@@ -90,16 +112,16 @@ func (c *composer) run(ctx context.Context) {
grantsErr = c.writeGrants(ctx) grantsErr = c.writeGrants(ctx)
} }
for _, w := range waiters { for _, w := range waiters {
var errs []error r := passResult{failed: map[string]error{}}
for _, n := range w.nodes { for _, n := range w.nodes {
if err := failed[n]; err != nil { if err := failed[n]; err != nil {
errs = append(errs, err) r.failed[n] = err
} }
} }
if w.grants && grantsErr != nil { if w.grants {
errs = append(errs, grantsErr) r.grants = grantsErr
} }
w.done <- errors.Join(errs...) w.done <- r
} }
} }
} }
+1 -1
View File
@@ -195,7 +195,7 @@ func watchHavesForGiven(ctx context.Context, open *stores) {
if at.IsZero() { if at.IsZero() {
continue 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}) })
} }
} }
}) })
+31 -8
View File
@@ -23,10 +23,13 @@ import (
// writes `<module>.<fingerprint hex>` then, naming the accounts it composed for the module, never removed. A module one // writes `<module>.<fingerprint hex>` then, naming the accounts it composed for the module, never removed. A module one
// of whose accounts has no password yet (the list leaves it out) has no grant until a later send carries it. // of whose accounts has no password yet (the list leaves it out) has no grant until a later send carries it.
// grantsFrom is the grant of every module whose accounts all have a password in hashes, at the fingerprint current names // grantsFrom is the grant of every module at the fingerprint current names for it, naming its accounts that have a
// for it; a module whose current build names no fingerprint of the new form has none. principals are the bus's users // sealed password in hashes; a module whose current build names no fingerprint of the new form has none. A module one
// as broker.Users composes them. // of whose accounts has none is granted its sealed accounts alone, and when it has none, its grant is dropped (drop):
func grantsFrom(principals []broker.Principal, hashes map[string]string, current map[string]string) map[string]sdkdelivery.ControllerGrant { // an older grant naming a node whose account is gone would let the assigner want it (issue 1945, rule 6 per node), and
// an empty grant names every node. principals are the bus's users as broker.Users composes them.
func grantsFrom(principals []broker.Principal, hashes map[string]string, current map[string]string) (
map[string]sdkdelivery.ControllerGrant, []string) {
accounts, missing := map[string][]string{}, map[string]bool{} accounts, missing := map[string][]string{}, map[string]bool{}
for _, p := range principals { for _, p := range principals {
if p.Module == "" || (p.Kind != broker.KindModule && p.Kind != broker.KindProcess) { if p.Module == "" || (p.Kind != broker.KindModule && p.Kind != broker.KindProcess) {
@@ -39,8 +42,9 @@ func grantsFrom(principals []broker.Principal, hashes map[string]string, current
accounts[p.Module] = append(accounts[p.Module], p.Username()) accounts[p.Module] = append(accounts[p.Module], p.Username())
} }
out := map[string]sdkdelivery.ControllerGrant{} out := map[string]sdkdelivery.ControllerGrant{}
var drop []string
for module, fp := range current { for module, fp := range current {
if missing[module] || !strings.HasPrefix(fp, sdkdelivery.SourcePrefix2) { if !strings.HasPrefix(fp, sdkdelivery.SourcePrefix2) {
continue continue
} }
key, err := sdkdelivery.ControllerGrantKey(module, fp) key, err := sdkdelivery.ControllerGrantKey(module, fp)
@@ -48,10 +52,15 @@ func grantsFrom(principals []broker.Principal, hashes map[string]string, current
continue continue
} }
names := accounts[module] names := accounts[module]
if missing[module] && len(names) == 0 {
drop = append(drop, key)
continue
}
sort.Strings(names) sort.Strings(names)
out[key] = sdkdelivery.ControllerGrant{Accounts: names} out[key] = sdkdelivery.ControllerGrant{Accounts: names}
} }
return out sort.Strings(drop)
return out, drop
} }
// currentFingerprints is, by module, the fingerprint of the build each registered module is registered at: its // currentFingerprints is, by module, the fingerprint of the build each registered module is registered at: its
@@ -248,13 +257,27 @@ func grantsIssued(ctx context.Context, inv *inventory.Inventory, js jetstream.Je
} }
hashes := make(map[string]string, len(kept)) hashes := make(map[string]string, len(kept))
for name, u := range kept { for name, u := range kept {
hashes[name] = u.PasswordHash // A grant names an account whose credential is sealed to its node, never one only minted: the assigner wants a
// node only once the grant names its account (issue 1945, rule 6 per node).
if u.Sealed {
hashes[name] = u.PasswordHash
}
} }
current, err := currentFingerprints(ctx, inv) current, err := currentFingerprints(ctx, inv)
if err != nil { if err != nil {
return fmt.Errorf("the grants issued are not written: %w", err) return fmt.Errorf("the grants issued are not written: %w", err)
} }
wrote, err := keepGrants(ctx, kv, grantsFrom(principals, hashes, current)) grants, drop := grantsFrom(principals, hashes, current)
wrote, err := keepGrants(ctx, kv, grants)
for _, key := range drop {
if _, gerr := kv.Get(ctx, key); gerr == nil {
if derr := kv.Delete(ctx, key); derr != nil {
err = errors.Join(err, fmt.Errorf("%s: %w", key, derr))
} else {
fmt.Printf("%sthe grant %s dropped: none of its accounts is sealed to its node\n", indent, key)
}
}
}
if wrote > 0 { if wrote > 0 {
fmt.Printf("%s%d grant(s) issued written for the assigner\n", indent, wrote) fmt.Printf("%s%d grant(s) issued written for the assigner\n", indent, wrote)
} }
+17 -4
View File
@@ -23,12 +23,25 @@ func TestTheGrantsAListIssues(t *testing.T) {
y := broker.Principal{Kind: broker.KindModule, Node: "ace", Module: "mesh-y"} y := broker.Principal{Kind: broker.KindModule, Node: "ace", Module: "mesh-y"}
node := broker.Principal{Kind: broker.KindNode, Node: "novox"} node := broker.Principal{Kind: broker.KindNode, Node: "novox"}
hashes := map[string]string{x.Username(): "h1", xp.Username(): "h2", node.Username(): "h3"} hashes := map[string]string{x.Username(): "h1", xp.Username(): "h2", node.Username(): "h3"}
got := grantsFrom([]broker.Principal{x, xp, y, node}, hashes, // mesh-z is sealed on novox and not on ace (issue 1945: an older grant must not name ace): its grant names novox alone.
map[string]string{"mesh-x": grantFP('a'), "mesh-y": grantFP('b'), "web": grantFP('c'), "old": "src1:abc"}) zn := broker.Principal{Kind: broker.KindModule, Node: "novox", Module: "mesh-z"}
za := broker.Principal{Kind: broker.KindModule, Node: "ace", Module: "mesh-z"}
hashes[zn.Username()] = "h4"
got, drop := grantsFrom([]broker.Principal{x, xp, y, node, zn, za}, hashes,
map[string]string{"mesh-x": grantFP('a'), "mesh-y": grantFP('b'), "web": grantFP('c'), "old": "src1:abc",
"mesh-z": grantFP('d')})
xKey, _ := sdkdelivery.ControllerGrantKey("mesh-x", grantFP('a')) xKey, _ := sdkdelivery.ControllerGrantKey("mesh-x", grantFP('a'))
yKey, _ := sdkdelivery.ControllerGrantKey("mesh-y", grantFP('b'))
zKey, _ := sdkdelivery.ControllerGrantKey("mesh-z", grantFP('d'))
webKey, _ := sdkdelivery.ControllerGrantKey("web", grantFP('c')) webKey, _ := sdkdelivery.ControllerGrantKey("web", grantFP('c'))
if len(got) != 2 { if len(got) != 3 {
t.Fatalf("grants = %v, want mesh-x's and web's", got) t.Fatalf("grants = %v, want mesh-x's, mesh-z's and web's", got)
}
if g := got[zKey]; len(g.Accounts) != 1 || g.Accounts[0] != zn.Username() {
t.Errorf("mesh-z's grant = %+v, want its sealed account alone", g)
}
if len(drop) != 1 || drop[0] != yKey {
t.Errorf("dropped %v, want mesh-y's grant, none of whose accounts is sealed", drop)
} }
if g := got[xKey]; len(g.Accounts) != 2 || g.Accounts[0] != x.Username() && g.Accounts[1] != x.Username() { if g := got[xKey]; len(g.Accounts) != 2 || g.Accounts[0] != x.Username() && g.Accounts[1] != x.Username() {
t.Errorf("mesh-x's grant = %+v, want both its accounts", g) t.Errorf("mesh-x's grant = %+v, want both its accounts", g)
+352
View File
@@ -0,0 +1,352 @@
package main
import (
"context"
"encoding/json"
"errors"
"fmt"
"sort"
"strings"
"time"
"git.novox.be/novox/mesh-sdk/go/catalogue"
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/nats-io/nats.go/jetstream"
"github.com/novox/mesh-controller/internal/broker"
"github.com/novox/mesh-controller/internal/conditions"
"github.com/novox/mesh-controller/internal/inventory"
)
// **A kept composition follows its inputs** (novox/hq ADR 0545, issue 1945). The controller composes a module only at
// the fingerprint it is registered at, and keeps the compositions of up to five fingerprints (ADR 0539 §4) so that
// going back is one want. A kept one was never composed again, so it carried the config of the moment it was made: on
// 2026-10-25 a want back to an earlier mesh-build-go on g14 applied its capacity of 0, the value from before it was
// set. Now a write of mesh-config's resolved values for a module on a node, heard from that bucket, composes the node
// and then every kept fingerprint of that module on that node again, each from its own build's manifest with today's
// inputs, writing only those whose composition differs (writeComposed). The writes are read from a durable consumer,
// acknowledged once handled, so a change made while no controller ran is delivered to the next one, and a start
// composes again only what was written since the last write handled, never every kept composition.
// configTouched is the node and module a key of mesh-config_resolved (`<node>.<module>.<key>`) names.
func configTouched(key string) (node, module string, ok bool) {
parts := strings.SplitN(key, ".", 3)
if len(parts) != 3 || parts[0] == "" || parts[1] == "" {
return "", "", false
}
return parts[0], parts[1], true
}
// keptKey is one kept composition to compose again.
type keptKey struct{ module, fp string }
// keptToFollow is, of a node's composition keys, each of modules (every module when nil) at a fingerprint other than
// the one it is registered at, sorted: the kept compositions. The node's own composition and other nodes' are not.
func keptToFollow(keys []string, node string, modules map[string]bool, registered map[string]string) []keptKey {
var out []keptKey
for _, key := range keys {
n, module, fp, isNode, err := sdkdelivery.ParseComposedKey(key)
if err != nil || isNode || n != node || (modules != nil && !modules[module]) {
continue
}
if fp == registered[module] || !strings.HasPrefix(fp, sdkdelivery.SourcePrefix2) {
continue
}
out = append(out, keptKey{module, fp})
}
sort.Slice(out, func(i, j int) bool {
if out[i].module != out[j].module {
return out[i].module < out[j].module
}
return out[i].fp < out[j].fp
})
return out
}
// buildAt is, of a module's builds newest first, the newest that worked at fp and kept its manifest.
func buildAt(builds []inventory.Build, fp string) (inventory.Build, bool) {
for _, b := range builds {
if b.Worked() && b.SourceFingerprint == fp && len(b.Manifest) > 0 {
return b, true
}
}
return inventory.Build{}, false
}
// followKept composes again, on node, every kept composition of modules (every module when nil) with today's inputs,
// and writes those that differ. It answers the keys written; a fingerprint whose build's manifest is gone is said and
// left as it is.
func followKept(ctx context.Context, open *stores, node string, modules map[string]bool) ([]string, error) {
if _, err := theLease.epoch(ctx); err != nil {
return nil, err
}
bus, err := aBus()
if err != nil {
return nil, err
}
defer bus.Close()
js, err := jetstream.New(bus.Conn())
if err != nil {
return nil, err
}
inv := open.inventory
registered, err := currentFingerprints(ctx, inv)
if err != nil {
return nil, err
}
at, err := openComposedPlace(ctx, js, bus.Conn(), node)
if err != nil {
return nil, err
}
lister, err := at.bucket.ListKeys(ctx)
if err != nil {
if errors.Is(err, jetstream.ErrNoKeysFound) {
return nil, nil // empty-on-error: a node with no compositions keeps none to compose again
}
return nil, err
}
var keys []string
for k := range lister.Keys() {
keys = append(keys, k)
}
kept := keptToFollow(keys, node, modules, registered)
if len(kept) == 0 {
return nil, nil
}
sign, err := controllerSigner(ctx, open)
if err != nil {
return nil, err
}
upgrades, err := inv.Upgrades(ctx)
if err != nil {
return nil, err
}
at.sign = sign
at.policy = func(module string) (string, string, string) {
u, ok := upgrades[module]
if !ok {
return "", "", ""
}
return u.Policy(), u.From, u.Why
}
in, err := inputFor(ctx, open)
if err != nil {
return nil, err
}
var wrote []string
var errs []error
builds := map[string][]inventory.Build{}
for _, k := range kept {
if ctx.Err() != nil {
return wrote, errors.Join(append(errs, ctx.Err())...) // stopped: the rest is the next controller's
}
if _, read := builds[k.module]; !read {
if builds[k.module], err = inv.Builds(ctx, k.module, 20); err != nil {
errs = append(errs, fmt.Errorf("%s's builds: %w", k.module, err))
continue
}
}
b, found := buildAt(builds[k.module], k.fp)
if !found {
fmt.Printf("compositions: %s's kept %s on %s has no build whose manifest is kept, and is not composed again\n",
k.module, short(k.fp), node)
continue
}
var m catalogue.Manifest
if err := json.Unmarshal(b.Manifest, &m); err != nil {
errs = append(errs, fmt.Errorf("the manifest of %s's build %s: %w", k.module, b.ID, err))
continue
}
c, err := keptComposition(in.Input, node, k.module, k.fp, m)
if err != nil {
errs = append(errs, fmt.Errorf("%s at %s on %s: %w", k.module, short(k.fp), node, err))
continue
}
key, _ := sdkdelivery.ComposedKey(node, k.module, k.fp)
// One write on the node's keys at a time in this process, held only for the write: composing in memory holds
// nothing, so a sweep never keeps a take-in from the node (the review of ADR 0545).
unlock := lockComposing(node)
w, err := writeComposed(ctx, at, key, c)
unlock()
if err != nil {
errs = append(errs, err)
continue
}
if w {
wrote = append(wrote, key)
}
}
if len(wrote) > 0 {
fmt.Printf("compositions: %s: kept compositions composed again with today's inputs: %s\n", node, strings.Join(wrote, ", "))
}
return wrote, errors.Join(errs...)
}
// configFollowGather is how long the config consumer waits after a write for more before it composes: mesh-config writes
// a module's keys one by one.
var configFollowGather = time.Second
// configFollowConsumer is the controller's durable consumer of mesh-config's resolved values (ADR 0545 §1, §3).
const configFollowConsumer = broker.ConfigFollowConsumer
// configFollowConfig is that consumer: each resolved key's last value once when it is made, then each write,
// acknowledged after the write is handled, so what a stop cut is delivered to the next controller. The first start of
// the release that makes it composes again the kept compositions of every module with config, through the same gather,
// lock and acknowledgement: mesh-config never writes an unchanged value again, so a value set before this release (g14's
// mesh-build-go capacity of 2, against kept compositions carrying 0) would otherwise never reach them. Its deliver
// policy is set before its first delivery; an existing durable's cannot be changed.
func configFollowConfig() jetstream.ConsumerConfig {
return jetstream.ConsumerConfig{Durable: configFollowConsumer, FilterSubject: "$KV." + broker.ConfigResolvedBucket + ".>",
DeliverPolicy: jetstream.DeliverLastPerSubjectPolicy, AckPolicy: jetstream.AckExplicitPolicy, AckWait: takeInAckWait,
MaxAckPending: takeInPending, Description: "the controller's following of config into kept compositions (novox/hq ADR 0545)"}
}
// kindConfigNotFollowed is the condition of a config consumer that cannot be read: a change of config composes nothing
// until it is.
const kindConfigNotFollowed = "config-not-followed"
func configNotFollowedObservation(err error) conditions.Observation {
return conditions.Observation{Scope: conditions.ScopeCore, ID: "config-follow", Token: kindConfigNotFollowed,
Kind: kindConfigNotFollowed, Severity: conditions.Urgent, Resolver: conditions.ResolverOperator, Source: "config",
Summary: fmt.Sprintf("mesh-config's resolved values are not heard, so a change of config composes nothing: %v", err)}
}
// watchConfigFollow reads the writes of mesh-config's resolved values from the controller's durable consumer and, for
// each node and module they touch, composes the node and the module's kept compositions there again, until ctx ends
// (ADR 0545). A consumer that cannot be made or read raises config-not-followed and is tried again a minute later; it is
// cleared once it is read.
func watchConfigFollow(ctx context.Context, open *stores) {
goServing(ctx, func() {
for {
err := followConfigOnce(ctx, open)
if ctx.Err() != nil {
return
}
fmt.Printf("config: %v; tried again in a minute\n", err)
if keeper := conditionsFrom; keeper != nil {
_, _ = keeper.Observe(ctx, configNotFollowedObservation(err))
}
select {
case <-ctx.Done():
return
case <-time.After(time.Minute):
}
}
})
}
// followConfigOnce reads the consumer until it fails or ctx ends.
func followConfigOnce(ctx context.Context, open *stores) error {
bus, err := aBus()
if err != nil {
return err
}
defer bus.Close()
js, err := jetstream.New(bus.Conn())
if err != nil {
return err
}
c, err := js.CreateOrUpdateConsumer(ctx, "KV_"+broker.ConfigResolvedBucket, configFollowConfig())
if err != nil {
return fmt.Errorf("the consumer of %s: %w", broker.ConfigResolvedBucket, err)
}
it, err := c.Messages(jetstream.PullMaxMessages(startTakeIns))
if err != nil {
return fmt.Errorf("the consumer of %s: %w", broker.ConfigResolvedBucket, err)
}
defer it.Stop()
if keeper := conditionsFrom; keeper != nil {
_, _ = keeper.Clear(ctx, configNotFollowedObservation(nil).Key(), "mesh-config's resolved values are heard")
}
type next struct {
m jetstream.Msg
err error
}
got := make(chan next)
go func() {
for {
m, err := it.Next()
select {
case got <- next{m, err}:
case <-ctx.Done():
if m != nil {
_ = m.Nak()
}
return
}
if err != nil {
return
}
}
}()
var held []jetstream.Msg
touched := map[string]map[string]bool{}
var gather <-chan time.Time
for {
select {
case <-ctx.Done():
for _, m := range held {
_ = m.Nak() // delivered at once to the next controller
}
return nil
case n := <-got:
if n.err != nil {
for _, m := range held {
_ = m.Nak()
}
return fmt.Errorf("the consumer of %s stopped: %w", broker.ConfigResolvedBucket, n.err)
}
held = append(held, n.m)
key := strings.TrimPrefix(n.m.Subject(), "$KV."+broker.ConfigResolvedBucket+".")
if node, module, ok := configTouched(key); ok {
if touched[node] == nil {
touched[node] = map[string]bool{}
}
touched[node][module] = true
}
if gather == nil {
gather = time.After(configFollowGather)
}
case <-gather:
gather = nil
now, msgs := touched, held
touched, held = map[string]map[string]bool{}, nil
forgetResolved() // the next composition reads the values just written, not a read of the last 5 s
failed := false
for node, modules := range now {
if ctx.Err() != nil {
failed = true // stopped: the writes are refused below and delivered to the next controller
break
}
if err := composerFor(open).compose(ctx, []string{node}, false); err != nil {
fmt.Printf("config: %s is not composed after its config changed: %v\n", node, err)
failed = true
}
if _, err := followKept(ctx, open, node, modules); err != nil {
fmt.Printf("config: %s's kept compositions are not composed again: %v\n", node, err)
failed = true
}
}
for _, m := range msgs {
if failed && ctx.Err() != nil {
_ = m.Nak() // stopped: delivered at once to the next controller
} else if failed {
_ = m.NakWithDelay(doneKeyRetry) // tried again: the bus delivers the writes again
} else {
acking, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
if err := m.DoubleAck(acking); err != nil {
fmt.Printf("config: the bus did not take the acknowledgement of %s: %v\n", m.Subject(), err)
}
cancel()
}
}
}
}
}
// forgetResolved drops the last read of mesh-config's resolved values.
func forgetResolved() {
theResolved.mu.Lock()
theResolved.entries = nil
theResolved.mu.Unlock()
}
+11 -3
View File
@@ -39,6 +39,11 @@ import (
// errKeyWaits says a take-in from a key waits: for the builder, a reference, the store or the forge. Never a failure. // errKeyWaits says a take-in from a key waits: for the builder, a reference, the store or the forge. Never a failure.
var errKeyWaits = errors.New("waits") var errKeyWaits = errors.New("waits")
// errWaitsForAWrite is a take-in that waits for the builder or the indexer to write again (the key not done, its
// artifacts not referenced, no fingerprint of the new form): that write wakes it, so nothing looks again before it
// (ADR 0544 §4).
var errWaitsForAWrite = fmt.Errorf("%w for the builder's or the indexer's next write", errKeyWaits)
// errKeyFailed says the module's key failed: the builder's own words follow. // errKeyFailed says the module's key failed: the builder's own words follow.
var errKeyFailed = errors.New("the builder's key failed") var errKeyFailed = errors.New("the builder's key failed")
@@ -298,12 +303,15 @@ func buildOfKey(ctx context.Context, keys keyReads, t keyTake, kc keyContext) (
waits := func(format string, a ...any) (inventory.Build, catalogue.Manifest, error) { waits := func(format string, a ...any) (inventory.Build, catalogue.Manifest, error) {
return inventory.Build{}, catalogue.Manifest{}, fmt.Errorf("%w: "+format, append([]any{errKeyWaits}, a...)...) return inventory.Build{}, catalogue.Manifest{}, fmt.Errorf("%w: "+format, append([]any{errKeyWaits}, a...)...)
} }
waitsForAWrite := func(format string, a ...any) (inventory.Build, catalogue.Manifest, error) {
return inventory.Build{}, catalogue.Manifest{}, fmt.Errorf("%w: "+format, append([]any{errWaitsForAWrite}, a...)...)
}
if t.Trunk.FP2 == "" && t.Hand == nil { if t.Trunk.FP2 == "" && t.Hand == nil {
why := t.Trunk.None2 why := t.Trunk.None2
if why == "" { if why == "" {
why = "the indexer wrote no fingerprint of the new form" why = "the indexer wrote no fingerprint of the new form"
} }
return waits("%s at %.8s has no fingerprint to take a build by: %s", t.Module, t.Trunk.Commit, why) return waitsForAWrite("%s at %.8s has no fingerprint to take a build by: %s", t.Module, t.Trunk.Commit, why)
} }
k, err := keys.Build(ctx, t.Module, t.Trunk.FP2) k, err := keys.Build(ctx, t.Module, t.Trunk.FP2)
if t.Hand != nil { if t.Hand != nil {
@@ -319,9 +327,9 @@ func buildOfKey(ctx context.Context, keys keyReads, t keyTake, kc keyContext) (
case k.Failed(): case k.Failed():
return inventory.Build{}, catalogue.Manifest{}, fmt.Errorf("%w: %s", errKeyFailed, k.Says()) return inventory.Build{}, catalogue.Manifest{}, fmt.Errorf("%w: %s", errKeyFailed, k.Says())
case !k.Done(): case !k.Done():
return waits("%s", k.Says()) return waitsForAWrite("%s", k.Says())
case len(k.Unreferenced()) > 0: case len(k.Unreferenced()) > 0:
return waits("%s", k.Says()) return waitsForAWrite("%s", k.Says())
} }
src, err := keys.Manifest(ctx, k.Build.Manifest) src, err := keys.Manifest(ctx, k.Build.Manifest)
if err != nil { if err != nil {
+118 -18
View File
@@ -9,8 +9,10 @@ import (
"net" "net"
"os" "os"
"path/filepath" "path/filepath"
"slices"
"sort" "sort"
"strings" "strings"
"sync"
"time" "time"
"git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/catalogue"
@@ -965,6 +967,101 @@ func processPrincipalFor(m catalogue.Manifest, processID string) (broker.Process
return broker.ProcessPrincipal{}, false return broker.ProcessPrincipal{}, false
} }
// mintOnlyIfAbsent marks a context whose issue mints only accounts that are not issued (onlyIfAbsent).
type mintOnlyIfAbsent struct{}
// onlyIfAbsent is ctx for an issue that never replaces an issued account: the assign act's, the assignments' watch's
// and a registration's, which may run at once for one account (novox/hq ADR 0543, issue 1945).
func onlyIfAbsent(ctx context.Context) context.Context {
return context.WithValue(ctx, mintOnlyIfAbsent{}, true)
}
// issuedNow collects the accounts an issue minted and sealed, for the line that says it (withIssued).
type issuedNow struct {
mu sync.Mutex
users []string
}
type issuedNowKey struct{}
// withIssued is ctx collecting the accounts its issues mint and seal.
func withIssued(ctx context.Context) (context.Context, *issuedNow) {
got := &issuedNow{}
return context.WithValue(ctx, issuedNowKey{}, got), got
}
func (g *issuedNow) list() []string {
if g == nil {
return nil
}
g.mu.Lock()
defer g.mu.Unlock()
return slices.Clone(g.users)
}
// accountIssuer is what issuing one account needs of the store.
type accountIssuer interface {
LockBusAccount(ctx context.Context, username string) (func(), error)
BusAccountIssued(ctx context.Context, username string) (minted, sealed bool, err error)
MintBusPassword(ctx context.Context, u inventory.BusUser) (string, error)
MintBusPasswordUnsealed(ctx context.Context, u inventory.BusUser) (string, error)
MarkBusAccountSealed(ctx context.Context, username string) error
}
// issueLockedFor is the longest an issue holds an account's lock: a seal cut by it leaves the account unsealed, issued
// again by the next issue.
const issueLockedFor = 2 * time.Minute
// issueAccount mints u's password and seals it to its node with seal. By default it replaces whatever stands (a
// rotation: `module issue`). Under onlyIfAbsent it issues only an account that is not issued, a row with no sealed
// credential counting as none (issue 1945, live on 2026-10-25 at 20:04: a row whose credential the node no longer held
// left ace's mesh-build-go with no credential), under the store's lock on the account, so two issuers never interleave
// their look, mint and seal: the hash the bus holds is always the password sealed. A seal that fails leaves the
// account unsealed, issued again by the next issue.
func issueAccount(ctx context.Context, accounts accountIssuer, u inventory.BusUser, seal func(password string) error) error {
only, _ := ctx.Value(mintOnlyIfAbsent{}).(bool)
if !only {
password, err := accounts.MintBusPassword(ctx, u)
if err != nil {
return err
}
if err := seal(password); err != nil {
return err
}
} else {
// A deadline on the locked section: it holds a connection of the store's pool across the vault and the bus.
var cancel context.CancelFunc
ctx, cancel = context.WithTimeout(ctx, issueLockedFor)
defer cancel()
release, err := accounts.LockBusAccount(ctx, u.Username)
if err != nil {
return err
}
defer release()
if minted, sealed, err := accounts.BusAccountIssued(ctx, u.Username); err != nil {
return err
} else if minted && sealed {
return nil
}
password, err := accounts.MintBusPasswordUnsealed(ctx, u)
if err != nil {
return err
}
if err := seal(password); err != nil {
return err
}
if err := accounts.MarkBusAccountSealed(ctx, u.Username); err != nil {
return fmt.Errorf("%s is sealed and not marked so, and is issued again by the next issue: %w", u.Username, err)
}
}
if got, _ := ctx.Value(issuedNowKey{}).(*issuedNow); got != nil {
got.mu.Lock()
got.users = append(got.users, u.Username)
got.mu.Unlock()
}
return nil
}
// issueOnTheNewBus gives an assigned module its credential on the bus being built. // issueOnTheNewBus gives an assigned module its credential on the bus being built.
// //
// **Three things differ from a management call, and each is the point of the move.** The credential // **Three things differ from a management call, and each is the point of the move.** The credential
@@ -990,13 +1087,10 @@ func issueOnTheNewBus(ctx context.Context, inv *inventory.Inventory, m catalogue
} }
if _, reads := m.OwnSecrets["broker"]; reads { if _, reads := m.OwnSecrets["broker"]; reads {
user := broker.Principal{Kind: broker.KindModule, Node: node, Module: m.Module}.Username() user := broker.Principal{Kind: broker.KindModule, Node: node, Module: m.Module}.Username()
password, err := inv.MintBusPassword(ctx, inventory.BusUser{ if err := issueAccount(ctx, inv, inventory.BusUser{Username: user, Kind: busKindOf(m.Module), Node: node,
Username: user, Kind: busKindOf(m.Module), Node: node, Module: m.Module, Module: m.Module}, func(password string) error {
}) return issueWith(ctx, inv, m, node, busAddress, known, reachable, user, password)
if err != nil { }); err != nil {
return err
}
if err := issueWith(ctx, inv, m, node, busAddress, known, reachable, user, password); err != nil {
return err return err
} }
} }
@@ -1021,12 +1115,15 @@ func issueOnTheNewBus(ctx context.Context, inv *inventory.Inventory, m catalogue
func issueProcess(ctx context.Context, inv *inventory.Inventory, m catalogue.Manifest, pr broker.ProcessPrincipal, func issueProcess(ctx context.Context, inv *inventory.Inventory, m catalogue.Manifest, pr broker.ProcessPrincipal,
node, busAddress string, known broker.Broker, reachable string) error { node, busAddress string, known broker.Broker, reachable string) error {
user := broker.Principal{Kind: broker.KindProcess, Node: node, Module: m.Module, Process: pr.Process}.Username() user := broker.Principal{Kind: broker.KindProcess, Node: node, Module: m.Module, Process: pr.Process}.Username()
password, err := inv.MintBusPassword(ctx, inventory.BusUser{ return issueAccount(ctx, inv, inventory.BusUser{Username: user, Kind: inventory.BusProcess, Node: node, Module: m.Module},
Username: user, Kind: inventory.BusProcess, Node: node, Module: m.Module, func(password string) error {
}) return sealProcess(ctx, inv, m, pr, node, busAddress, known, reachable, user, password)
if err != nil { })
return err }
}
// sealProcess seals a process's minted account to its node, hands it to the vault and makes its work-queue consumers.
func sealProcess(ctx context.Context, inv *inventory.Inventory, m catalogue.Manifest, pr broker.ProcessPrincipal,
node, busAddress string, known broker.Broker, reachable, user, password string) error {
name := processSecretName(pr.Process) name := processSecretName(pr.Process)
held, err := heldProcessAccount(known, reachable, node, m.Module, user, password) held, err := heldProcessAccount(known, reachable, node, m.Module, user, password)
if err != nil { if err != nil {
@@ -1091,12 +1188,15 @@ func runsTheProvisionerClient(m catalogue.Manifest) bool {
func issueProvisionerClient(ctx context.Context, inv *inventory.Inventory, m catalogue.Manifest, node string, func issueProvisionerClient(ctx context.Context, inv *inventory.Inventory, m catalogue.Manifest, node string,
known broker.Broker, reachable string) error { known broker.Broker, reachable string) error {
user := broker.Principal{Kind: broker.KindProvisioner, Node: node, Module: m.Module}.Username() user := broker.Principal{Kind: broker.KindProvisioner, Node: node, Module: m.Module}.Username()
password, err := inv.MintBusPassword(ctx, inventory.BusUser{ return issueAccount(ctx, inv, inventory.BusUser{Username: user, Kind: inventory.BusProvisioner, Node: node,
Username: user, Kind: inventory.BusProvisioner, Node: node, Module: m.Module, Module: m.Module}, func(password string) error {
return sealProvisionerClient(ctx, inv, m, node, known, reachable, user, password)
}) })
if err != nil { }
return err
} // sealProvisionerClient seals the provisioner client's minted account to its machine and hands it to the vault.
func sealProvisionerClient(ctx context.Context, inv *inventory.Inventory, m catalogue.Manifest, node string,
known broker.Broker, reachable, user, password string) error {
held, err := json.Marshal(struct { held, err := json.Marshal(struct {
URL string `json:"url"` URL string `json:"url"`
Fingerprint string `json:"fingerprint,omitempty"` Fingerprint string `json:"fingerprint,omitempty"`
+17 -8
View File
@@ -200,14 +200,20 @@ 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 // 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. // 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) { func watchNewModules(ctx context.Context, open *stores) {
inBackground(func() { inBackground(ctx, func() {
takeNewModules(ctx, open) 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 // 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 // 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. // 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) watchDoneKeys(ctx, open)
// And a change of config composes the node and the touched modules' kept compositions again, heard from
// mesh-config's resolved values (ADR 0545, kept_compositions.go).
watchConfigFollow(ctx, open)
// And each assignment that comes to stand issues its node's bus accounts and the module's grant (ADR 0543): a module
// built long before is assigned to one more node with no build done again (issue 1945, assignment_accounts.go).
watchAssignmentAccounts(ctx, open)
watchHavesForJudgement(ctx, open) watchHavesForJudgement(ctx, open)
// And the keep set keeps the builds every want names (ADR 0539 clause 1), with no plan left to name them. // And the keep set keeps the builds every want names (ADR 0539 clause 1), with no plan left to name them.
open.inventory.SetWanted(wantedOnTheBus) open.inventory.SetWanted(wantedOnTheBus)
@@ -232,7 +238,7 @@ func watchNewModules(ctx context.Context, open *stores) {
// wakeNewModule takes module in when the catalogue does not hold it: never on the watch's own goroutine, which must keep // wakeNewModule takes module in when the catalogue does not hold it: never on the watch's own goroutine, which must keep
// draining. // draining.
func wakeNewModule(ctx context.Context, open *stores, module string) { func wakeNewModule(ctx context.Context, open *stores, module string) {
inBackground(func() { inBackground(ctx, func() {
entries, err := open.inventory.Catalogued(ctx) entries, err := open.inventory.Catalogued(ctx)
if err != nil { if err != nil {
return return
@@ -250,14 +256,17 @@ func wakeNewModule(ctx context.Context, open *stores, module string) {
// ends only once the work its mesh started has. // ends only once the work its mesh started has.
var background sync.WaitGroup 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 // inBackground runs work beside the caller, counted in background and, under a serving controller, in its serving group,
// take-in may run it in the caller's turn and need not wait for it. // 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
var inBackground = func(work func()) { // 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) background.Add(1)
go func() { if !goServing(ctx, func() {
defer background.Done() defer background.Done()
work() work()
}() }) {
background.Done()
}
} }
// newModulesOf are the modules the indexer holds in a repository: what a first merge's word wakes. // newModulesOf are the modules the indexer holds in a repository: what a first merge's word wakes.
+21
View File
@@ -70,6 +70,27 @@ var plainWordings = map[string]func(conditions.Observation) words{
"none can be given it yet. It is tried again every half minute; the details say what fails.", m), "none can be given it yet. It is tried again every half minute; the details say what fails.", m),
Resolved: "the new build of " + m + " can be delivered"} Resolved: "the new build of " + m + " can be delivered"}
}), }),
// The take-in of done keys whose backlog stood still (take_in_consumer.go, novox/hq ADR 0544 §6).
kindTakeInStalled: worded(func(o conditions.Observation) words {
return words{Headline: "New builds are not being prepared for the machines",
Explanation: "The mesh prepares each new build for the machines as it is made, and has prepared none for " +
"five minutes while builds wait. Until it moves, no new build reaches a machine; the details say how many wait.",
Resolved: "new builds are being prepared again"}
}),
// A bucket of done keys whose take-in consumer could not be made (take_in_consumer.go, novox/hq ADR 0544 §6).
kindTakeInNotConsumed: worded(func(o conditions.Observation) words {
return words{Headline: "Some new builds are not being prepared",
Explanation: "The mesh could not start reading one of the two places new builds are announced in, so the " +
"builds announced there wait until the controller starts again. The details say which and why.",
Resolved: "the mesh reads where new builds are announced again"}
}),
// Config not followed into the compositions (kept_compositions.go, novox/hq ADR 0545).
kindConfigNotFollowed: worded(func(o conditions.Observation) words {
return words{Headline: "Changes of settings are not reaching the machines",
Explanation: "The mesh could not read the settings as they change, so a changed setting is not prepared for " +
"the machines until it can. It tries again every minute; the details say why.",
Resolved: "changes of settings reach the machines again"}
}),
// Machines and their node-engine. // Machines and their node-engine.
"silent": worded(func(o conditions.Observation) words { "silent": worded(func(o conditions.Observation) words {
m := machineOr(o, "a machine") m := machineOr(o, "a machine")
+8 -3
View File
@@ -72,6 +72,8 @@ func issueMissingProcessAccounts(ctx context.Context, inv *inventory.Inventory,
if len(m.ProcessBuses()) == 0 || mayIssue(m) != nil { if len(m.ProcessBuses()) == 0 || mayIssue(m) != nil {
return "", nil return "", nil
} }
// Only the missing: never a second password over one another caller minted between this look and the mint.
ctx = onlyIfAbsent(ctx)
minted, err := inv.BusMinted(ctx) minted, err := inv.BusMinted(ctx)
if err != nil { if err != nil {
return "", err return "", err
@@ -92,15 +94,18 @@ func issueMissingProcessAccounts(ctx context.Context, inv *inventory.Inventory,
if err != nil { if err != nil {
busAddress = "" busAddress = ""
} }
var ids []string ctx, got := withIssued(ctx)
for _, pr := range missing { for _, pr := range missing {
if err := issueProcess(ctx, inv, m, pr, node, busAddress, known, reachable); err != nil { if err := issueProcess(ctx, inv, m, pr, node, busAddress, known, reachable); err != nil {
return "", err return "", err
} }
ids = append(ids, pr.Process) }
issued := got.list()
if len(issued) == 0 {
return "", nil // another issuer sealed them meanwhile
} }
return fmt.Sprintf("its process account(s) %s issued and sealed to %s, and arrive with the push", return fmt.Sprintf("its process account(s) %s issued and sealed to %s, and arrive with the push",
strings.Join(ids, ", "), node), nil strings.Join(issued, ", "), node), nil
} }
// issueGainedProcessAccounts runs after a module's new definition is registered: on each machine assigned it, the // issueGainedProcessAccounts runs after a module's new definition is registered: on each machine assigned it, the
-58
View File
@@ -1,12 +1,9 @@
package main package main
import ( import (
"encoding/json"
"strings"
"testing" "testing"
"git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/catalogue"
"github.com/novox/mesh-controller/internal/broker"
) )
// mailSenderWithProcess is mesh-mail's real manifest shape (novox/hq ADR 0436 §7, §8, the lead's fixture): // mailSenderWithProcess is mesh-mail's real manifest shape (novox/hq ADR 0436 §7, §8, the lead's fixture):
@@ -25,61 +22,6 @@ func mailSenderWithProcess() catalogue.Manifest {
return m return m
} }
// review round 1, finding 1 (1181): a module declaring no `broker` of its own at all — mesh-mail's real
// shape, tools served by the tool runner — still passes mayIssue on its process's bus declaration alone.
func TestAModuleWithNoBrokerAndOnlyAProcessMayBeIssued(t *testing.T) {
if err := mayIssue(mailSenderWithProcess()); err != nil {
t.Fatalf("a module with no broker of its own, whose process declares bus, was refused: %v", err)
}
}
// 1181/AC1: a process with a bus declaration and no own secret broker-<process id> cannot be issued — the
// same refusal mayIssue already gives a module with no `broker` (novox/hq ADR 0436 §2).
func TestAProcessWithNoBrokerSecretCannotBeIssued(t *testing.T) {
m := mailSenderWithProcess()
delete(m.OwnSecrets, "broker-send")
err := mayIssue(m)
if err == nil {
t.Fatal("a process with no broker-<id> own secret was issued an account")
}
for _, want := range []string{"mesh-mail", "send", "broker-send", "own-secrets"} {
if !strings.Contains(err.Error(), want) {
t.Errorf("the refusal does not say %q: %v", want, err)
}
}
if err := mayIssue(mailSenderWithProcess()); err != nil {
t.Errorf("a process declaring its own broker-<id> secret was refused: %v", err)
}
}
// 1181/AC1: the account held for a process is the provisioner client's form — {url, fingerprint, node,
// module, user, password} — named for the process, never the module's own account.
func TestAProcesssHeldAccountIsTheProvisionersForm(t *testing.T) {
user := broker.Principal{Kind: broker.KindProcess, Node: "novox", Module: "mesh-mail", Process: "send"}.Username()
if user != "novox.mesh-mail.send" {
t.Fatalf("the process's username: %q", user)
}
held, err := heldProcessAccount(broker.Broker{Fingerprint: "abc"}, "novox.example:4222", "novox", "mesh-mail",
user, "a-password")
if err != nil {
t.Fatal(err)
}
var got map[string]any
if err := json.Unmarshal(held, &got); err != nil {
t.Fatal(err)
}
want := map[string]any{"url": "nats://novox.example:4222", "fingerprint": "abc", "node": "novox",
"module": "mesh-mail", "user": "novox.mesh-mail.send", "password": "a-password"}
for k, v := range want {
if got[k] != v {
t.Errorf("%s: %v, want %v", k, got[k], v)
}
}
if len(got) != len(want) {
t.Errorf("the held account carries %v; want exactly %v (never claims — a process holds no seat)", got, want)
}
}
// 1181/AC1: composed, the process's own secret belongs to its own account at 0600 (novox/hq ADR 0436 §1, §2; // 1181/AC1: composed, the process's own secret belongs to its own account at 0600 (novox/hq ADR 0436 §1, §2;
// the owner mechanism is issue 1170's, proven generically there — this is mesh-mail's own shape). // the owner mechanism is issue 1170's, proven generically there — this is mesh-mail's own shape).
func TestAProcessesOwnSecretIsOwnedByItsOwnAccount(t *testing.T) { func TestAProcessesOwnSecretIsOwnedByItsOwnAccount(t *testing.T) {
+8 -2
View File
@@ -59,6 +59,9 @@ func serve(ctx context.Context) (err error) {
} }
defer open.Close() defer open.Close()
inv := open.inventory 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 // **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 // 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. // 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) "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 { if err != nil {
return err return err
} }
defer ident.Close()
// Established at start rather than on first use. A control plane that cannot sign is one // 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 // whose declarations every node correctly refuses, and that should be a startup failure
@@ -176,6 +179,7 @@ func serve(ctx context.Context) (err error) {
keeper := openServingConditions(ctx, open, bus) keeper := openServingConditions(ctx, open, bus)
if keeper != nil { if keeper != nil {
defer closeConditions(keeper) 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 // The bus's own objects, asserted on every start. **Not created once at genesis**: a stream
@@ -244,10 +248,12 @@ func serve(ctx context.Context) (err error) {
// next test's `conditions` with "connection closed", issue 1945). // next test's `conditions` with "connection closed", issue 1945).
handActConn = bus.Conn handActConn = bus.Conn
defer func() { handActConn = nil }() 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). // And everything else this controller does on the bus for a moment (novox/hq issue 327).
servingBus.Store(server.JetStream()) servingBus.Store(server.JetStream())
// No longer serving: nothing is lent, and dead-letters says it is not read here (novox/hq issue 330). // No longer serving: nothing is lent, and dead-letters says it is not read here (novox/hq issue 330).
defer servingBus.Store(nil) 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). // And says when it replaced a value given by hand (novox/hq ADR 0228).
givenEvents = bus givenEvents = bus
// And a pull request's merge check, asked when the forge announces its head and said when judged // And a pull request's merge check, asked when the forge announces its head and said when judged
+4
View File
@@ -111,6 +111,9 @@ type meshStatus struct {
// why they could not be read. // why they could not be read.
Pending []pendingStatus `json:"pending,omitempty"` Pending []pendingStatus `json:"pending,omitempty"`
PendingUnread string `json:"pendingUnread,omitempty"` PendingUnread string `json:"pendingUnread,omitempty"`
// TakeIn is the take-in's backlog of done keys: keys not delivered yet and keys being taken in (novox/hq ADR 0544
// §6). Absent where it cannot be read.
TakeIn *takeInBacklog `json:"take-in,omitempty"`
} }
// pendingStatus is one pending assignment as status says it. // pendingStatus is one pending assignment as status says it.
@@ -266,6 +269,7 @@ func statusAsJSON(asked answers) ([]byte, error) {
out.Failing = providerStandings(asked.conditions) out.Failing = providerStandings(asked.conditions)
out.Overflowing = asked.overflowing out.Overflowing = asked.overflowing
out.PendingUnread = asked.pendingUnread out.PendingUnread = asked.pendingUnread
out.TakeIn = asked.takeIn
for _, p := range asked.pending { for _, p := range asked.pending {
out.Pending = append(out.Pending, pendingStatus{Node: p.Node, Module: p.Module, State: p.State, out.Pending = append(out.Pending, pendingStatus{Node: p.Node, Module: p.Module, State: p.State,
Build: p.Build, Repository: p.Repository, Path: p.Path, Since: p.Since, Settled: p.Settled, Note: p.Note}) Build: p.Build, Repository: p.Repository, Path: p.Path, Since: p.Since, Settled: p.Settled, Note: p.Note})
+5 -4
View File
@@ -29,17 +29,17 @@ func withServingGroup(ctx context.Context) (context.Context, *servingGroup) {
// goServing runs work beside the caller, counted in the group ctx carries, so serve's stop waits for it. Under a group // 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 // 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. // goroutine, as before. It says whether it started work.
func goServing(ctx context.Context, work func()) { func goServing(ctx context.Context, work func()) bool {
g, _ := ctx.Value(servingGroupKey{}).(*servingGroup) g, _ := ctx.Value(servingGroupKey{}).(*servingGroup)
if g == nil { if g == nil {
go work() go work()
return return true
} }
g.mu.Lock() g.mu.Lock()
if g.stopped { if g.stopped {
g.mu.Unlock() g.mu.Unlock()
return return false
} }
g.wg.Add(1) g.wg.Add(1)
g.mu.Unlock() g.mu.Unlock()
@@ -47,6 +47,7 @@ func goServing(ctx context.Context, work func()) {
defer g.wg.Done() defer g.wg.Done()
work() work()
}() }()
return true
} }
// stop starts nothing more and waits for what was started; the caller ends their context first. A second stop waits // stop starts nothing more and waits for what was started; the caller ends their context first. A second stop waits
+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)
}
}
+1 -1
View File
@@ -67,7 +67,7 @@ func keepAnnouncements(ctx context.Context, js *broker.JetStream, open *stores)
return return
} }
if changed { if changed {
inBackground(func() { composeAnnouncedReads(context.WithoutCancel(ctx), open) }) inBackground(ctx, func() { composeAnnouncedReads(context.WithoutCancel(ctx), open) })
} }
}) })
if err != nil { if err != nil {
+11
View File
@@ -151,6 +151,10 @@ func printStatus(asked answers) error {
} }
printPending(asked.pending, asked.pendingUnread) printPending(asked.pending, asked.pendingUnread)
if b := asked.takeIn; b != nil && b.Total() > 0 {
fmt.Printf("the take-in of done keys has %d key(s) not taken in yet: %d waiting to be delivered, %d being taken in\n\n",
b.Total(), b.Pending, b.Taking)
}
if len(behind) > 0 { if len(behind) > 0 {
var names []string var names []string
@@ -471,6 +475,13 @@ func theThreeQuestions(ctx context.Context, open *stores) (answers, error) {
out.pendingUnread = err.Error() out.pendingUnread = err.Error()
err = nil err = nil
} }
// And the take-in's backlog of done keys, read from its consumers on the bus (novox/hq ADR 0544 §6): where there is
// a bus; a stall is a condition of its own, so this is said and judges nothing.
if _, onBus := broker.BusAddress(); onBus == nil {
if b, berr := readTakeInBacklog(ctx); berr == nil {
out.takeIn = &b
}
}
// And how many repairs were done by hand this week (novox/hq to-be 45 §7) — where there is a bus // And how many repairs were done by hand this week (novox/hq to-be 45 §7) — where there is a bus
// to read the log from; a process with none has no log to count. // to read the log from; a process with none has no log to count.
if _, onBus := broker.BusAddress(); onBus == nil { if _, onBus := broker.BusAddress(); onBus == nil {
+36 -2
View File
@@ -4,6 +4,7 @@ import (
"context" "context"
"fmt" "fmt"
"os" "os"
"sync"
"time" "time"
"git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/catalogue"
@@ -163,6 +164,13 @@ func openLicences(ctx context.Context) (*licences.Licences, error) {
return held, nil 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. // stores is what one command has open.
// //
// **Opened once, not once per machine.** Working out what a machine should be reaches the identity // **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 inventory *inventory.Inventory
identity *identity.Identity identity *identity.Identity
licences *licences.Licences 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. // 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. // 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) { func (h *stores) Identity(ctx context.Context) (*identity.Identity, error) {
h.mu.Lock()
defer h.mu.Unlock()
if h.identity != nil { if h.identity != nil {
return h.identity, nil return h.identity, nil
} }
opened, err := openIdentity(ctx) opened, err := identityOpener(ctx)
if err != nil { if err != nil {
return nil, err 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. // Licences is the context holding model access, opened if it has not been.
func (h *stores) Licences(ctx context.Context) (*licences.Licences, error) { func (h *stores) Licences(ctx context.Context) (*licences.Licences, error) {
h.mu.Lock()
defer h.mu.Unlock()
if h.licences != nil { if h.licences != nil {
return h.licences, nil return h.licences, nil
} }
opened, err := openLicences(ctx) opened, err := licencesOpener(ctx)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@@ -226,3 +241,22 @@ func (h *stores) Close() {
h.inventory.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
}
+452
View File
@@ -0,0 +1,452 @@
package main
import (
"context"
"errors"
"fmt"
"strings"
"sync"
"sync/atomic"
"time"
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/nats-io/nats.go/jetstream"
"github.com/novox/mesh-controller/internal/broker"
"github.com/novox/mesh-controller/internal/conditions"
)
// The take-in's two durable pull consumers (novox/hq ADR 0544; research 072, milestones 2 and 4). On the indexer's
// modules bucket the consumer delivers each module's last trunk value when it is made, then each write: the start's look
// of a controller whose consumer is new, and nothing at a restart but what the controller before it left
// unacknowledged. On the builder's builds bucket it delivers each write from when it is made. A key is acknowledged once
// its module is taken in, has nothing new, or waits for a write that will be delivered; any other wait or failure is
// refused with a delay that grows with each refusal, and the bus delivers it again.
//
// **The bound on the store is the one take-in loop**, a batch of at most startTakeIns keys at a time (issues 1945 and
// 1277: a burst of take-ins once filled the store until it refused every connection). The consumers' own limit on
// unacknowledged keys is far above it, because a refused key stays unacknowledged until its delay ends: with a limit of
// a batch, a handful of keys refused for long (a manifest word this controller does not read yet) would stop a consumer
// for good, the build that reads the word among what it holds back (the review of ADR 0544).
// startTakeIns is the most a batch takes: the take-ins that may wait on the store at once.
const startTakeIns = 16
// takeInPending is how many keys each consumer may leave unacknowledged: delivered and being taken in, or refused and
// waiting for their delay. Far above a batch, so refused keys never stop the others.
const takeInPending = 1000
// takeInBuffered is how many delivered keys each consumer's reader holds ahead of the loop.
const takeInBuffered = startTakeIns / 2
// takeInAckWait is how long a delivered key may stay unacknowledged before the bus delivers it again: longer than a batch
// takes, its two composer passes among it.
const takeInAckWait = 10 * time.Minute
// takeInGather is how long a batch waits for more keys after its first, so a burst is one batch.
var takeInGather = 200 * time.Millisecond
// takeInStall is how long a backlog may stand without a key delivered before it is a condition (ADR 0544 §6).
var takeInStall = 5 * time.Minute
// takeInMsg is one delivered key, as the take-in uses it.
type takeInMsg struct {
delivered
ack func()
nak func(delay time.Duration)
}
// takeInKeyOf is the module a delivered key names, false for a delete or a purge, which takes nothing in.
func takeInKeyOf(bucket string, m jetstream.Msg) (delivered, bool) {
if op := m.Headers().Get("KV-Operation"); op == "DEL" || op == "PURGE" {
return delivered{}, false
}
key := strings.TrimPrefix(m.Subject(), "$KV."+bucket+".")
module, _, _ := strings.Cut(key, ".")
if module == "" {
return delivered{}, false
}
var deliveries uint64 = 1
if meta, err := m.Metadata(); err == nil {
deliveries = meta.NumDelivered
}
return delivered{module: module, again: deliveries > 1, deliveries: deliveries}, true
}
// takeInConsumer is one bucket's consumer of the take-in.
type takeInConsumer struct {
bucket string
consumer jetstream.Consumer
}
// takeInConfig is the take-in's consumer on bucket's stream.
func takeInConfig(bucket string) jetstream.ConsumerConfig {
deliver := jetstream.DeliverNewPolicy
if bucket == sdkdelivery.ModulesBucket {
deliver = jetstream.DeliverLastPerSubjectPolicy // the start's look: every module's trunk value once
}
return jetstream.ConsumerConfig{
Durable: broker.TakeInConsumer, FilterSubject: "$KV." + bucket + ".>", DeliverPolicy: deliver,
AckPolicy: jetstream.AckExplicitPolicy, AckWait: takeInAckWait, MaxAckPending: takeInPending,
Description: "the controller's take-in of done keys (novox/hq ADR 0544)",
}
}
// takeInConsumers makes or finds the take-in's consumer on each bucket's stream; one that cannot be made is said, and
// the other is still read.
func takeInConsumers(ctx context.Context, js jetstream.JetStream) []takeInConsumer {
var out []takeInConsumer
for _, bucket := range broker.TakeInBuckets {
c, err := js.CreateOrUpdateConsumer(ctx, "KV_"+bucket, takeInConfig(bucket))
o := takeInNotConsumedObservation(bucket, err)
if err != nil {
// Loud: a consumer whose deliver policy differs cannot be updated, and nothing of its bucket is taken in.
doneKeysSay("done keys: " + o.Summary)
if keeper := conditionsFrom; keeper != nil {
_, _ = keeper.Observe(ctx, o)
}
continue
}
// Made: a condition a controller before this one raised for it is cleared (a restart that works clears it).
if keeper := conditionsFrom; keeper != nil {
_, _ = keeper.Clear(ctx, o.Key(), "the take-in's consumer on "+bucket+" is made")
}
out = append(out, takeInConsumer{bucket: bucket, consumer: c})
}
return out
}
// watchDoneKeys takes in the done keys the two consumers deliver, batch after batch, until ctx ends, and watches their
// backlog. A consumer that cannot be made is said: its bucket is then not taken in until the next start.
func watchDoneKeys(ctx context.Context, open *stores) {
keys, closeKeys, err := openBuilderKeys()
if err != nil {
fmt.Printf("done keys: the builder's keys are not taken in: %v\n", err)
return
}
consumers := takeInConsumers(ctx, keys.JS)
if len(consumers) == 0 {
closeKeys()
return
}
in := make(chan takeInMsg)
// moving counts the keys handed to the loop and the keys it settled (acknowledged or refused, each as the bus took
// it): what tells a take-in that moves from one that holds work and does nothing (the stall watch, ADR 0544 §6).
moving := &takeInMoving{}
read := make([]jetstream.Consumer, 0, len(consumers))
// The connection closes only once the readers and the loop have ended, so the refusals they send at the stop reach
// the bus (the third review of ADR 0544).
var users sync.WaitGroup
for _, c := range consumers {
bucket := c.bucket
read = append(read, c.consumer)
it, err := c.consumer.Messages(jetstream.PullMaxMessages(takeInBuffered))
if err != nil {
fmt.Printf("done keys: the take-in's consumer on %s is not read: %v\n", bucket, err)
continue
}
goServing(ctx, func() {
<-ctx.Done()
it.Drain() // the keys it holds are handed out and refused at once below, not lost to their ack wait
})
users.Add(1)
goServing(ctx, func() {
defer users.Done()
for {
m, err := it.Next()
if err != nil {
if ctx.Err() == nil && !errors.Is(err, jetstream.ErrMsgIteratorClosed) {
fmt.Printf("done keys: the take-in's consumer on %s stopped: %v\n", bucket, err)
}
return
}
d, ok := takeInKeyOf(bucket, m)
if !ok {
_ = m.Ack()
continue
}
// Settled is the loop's attempt, whatever the bus answers: an acknowledgement it refuses (at a first start,
// until the bus node is composed) is delivered again and settled again, and never counts as work held
// (the third review of ADR 0544). Acknowledged synchronously and past the stop, so a refusal is said.
msg := takeInMsg{delivered: d, ack: func() {
acking, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
defer cancel()
if err := m.DoubleAck(acking); err != nil {
doneKeysSay(fmt.Sprintf("done keys: the bus did not take the acknowledgement of %s: %v", d.module, err))
}
moving.settled.Add(1)
},
nak: func(delay time.Duration) {
_ = m.NakWithDelay(delay)
moving.settled.Add(1)
}}
if ctx.Err() != nil {
_ = m.Nak()
continue // drained: each key it held, refused at once
}
select {
case in <- msg:
moving.handed.Add(1)
case <-ctx.Done():
_ = m.Nak() // delivered again at once to the controller after this one, not after the ack wait
continue
}
}
})
}
backlog := func(ctx context.Context) (takeInBacklog, error) { return backlogOf(ctx, read) }
steps := liveTakeInSteps(open, keys, keys.JS)
users.Add(1)
goServing(ctx, func() {
defer users.Done()
takeInLoop(ctx, in, steps, backlog)
})
goServing(ctx, func() { watchTakeInStall(ctx, backlog, moving) })
goServing(ctx, func() {
<-ctx.Done()
users.Wait()
closeKeys()
})
}
// takeInLoop gathers delivered keys into batches and takes each batch in, until ctx ends.
func takeInLoop(ctx context.Context, in <-chan takeInMsg, steps takeInSteps,
backlog func(context.Context) (takeInBacklog, error)) {
t := &takeInState{memo: map[string]string{}, raised: map[string]bool{}, saidWait: map[string]string{}}
for {
var batch []takeInMsg
select {
case <-ctx.Done():
return
case m := <-in:
batch = append(batch, m)
}
gather := time.NewTimer(takeInGather)
more:
for len(batch) < startTakeIns {
select {
case m := <-in:
batch = append(batch, m)
case <-gather.C:
break more
case <-ctx.Done():
gather.Stop()
for _, m := range batch {
m.nak(0)
}
return
}
}
gather.Stop()
said := t.take(ctx, steps, batch)
if b, err := backlog(ctx); err == nil {
said += fmt.Sprintf("; backlog %d", b.Total())
}
doneKeysSay("done keys: " + said)
}
}
// takeInState is what the take-in keeps between batches in this process: the trunk each module was last taken in at,
// the modules whose condition it raised, and the wait last said of each.
type takeInState struct {
memo map[string]string
raised map[string]bool
saidWait map[string]string
}
// take takes one batch in and acknowledges or refuses each key, raising and clearing the modules' conditions. It answers
// the batch in a line.
func (t *takeInState) take(ctx context.Context, steps takeInSteps, batch []takeInMsg) string {
keys := make([]delivered, len(batch))
for i, m := range batch {
keys[i] = m.delivered
}
settled, passes := steps.batch(ctx, keys, t.memo)
verdicts := make(map[string]verdict, len(settled))
var taken, nothing, waiting, failed int
for _, r := range settled {
verdicts[r.module] = r.verdict
o := doneKeyObservation(r.module)
switch {
case r.err != nil:
failed++
doneKeysSay(fmt.Sprintf("done keys: %s: %v", r.module, r.err))
if keeper := conditionsFrom; keeper != nil && !t.raised[r.module] && ctx.Err() == nil &&
!errors.Is(r.err, context.Canceled) {
o.Summary += r.err.Error()
if _, err := keeper.Observe(ctx, o); err == nil {
t.raised[r.module] = true
}
}
case r.waits != "":
waiting++
if t.saidWait[r.module] != r.waits {
doneKeysSay(fmt.Sprintf("done keys: %s %s", r.module, r.waits))
t.saidWait[r.module] = r.waits
}
case r.taken:
taken++
delete(t.saidWait, r.module)
if r.said != "" {
doneKeysSay("done keys: " + r.said)
}
// Cleared on every take-in, whoever raised it: a controller before a restart, whose key was delivered again.
if keeper := conditionsFrom; keeper != nil {
if _, err := keeper.Clear(ctx, o.Key(), "composed and granted"); err == nil {
delete(t.raised, r.module)
}
}
case r.nothingNew:
nothing++
}
}
for _, m := range batch {
if verdicts[m.module] == refused {
if ctx.Err() != nil {
m.nak(0) // cut by the stop: the next controller takes it at once
} else {
m.nak(refusalDelay(m.deliveries))
}
} else {
m.ack()
}
}
return fmt.Sprintf("%d key(s) of %d module(s): %d taken in, %d with nothing new, %d waiting, %d failed; %d composer pass(es)",
len(batch), len(settled), taken, nothing, waiting, failed, passes)
}
// takeInBacklog is the take-in's backlog on the bus: keys not delivered yet, and keys delivered and not acknowledged
// (being taken in, or refused and waiting for their delay).
type takeInBacklog struct {
Pending uint64 `json:"pending"`
Taking uint64 `json:"taking"`
}
// Total is every key the take-in has not acknowledged.
func (b takeInBacklog) Total() uint64 { return b.Pending + b.Taking }
// backlogOf reads the consumers' own counts from the bus.
func backlogOf(ctx context.Context, consumers []jetstream.Consumer) (takeInBacklog, error) {
var b takeInBacklog
for _, c := range consumers {
info, err := c.Info(ctx)
if err != nil {
return takeInBacklog{}, err
}
b.Pending += info.NumPending
b.Taking += uint64(info.NumAckPending)
}
return b, nil
}
// readTakeInBacklog is the take-in's backlog for a reader in any process (`status`): the consumers asked by name.
func readTakeInBacklog(ctx context.Context) (takeInBacklog, error) {
keys, closeKeys, err := openBuilderKeys()
if err != nil {
return takeInBacklog{}, err
}
defer closeKeys()
var consumers []jetstream.Consumer
for _, bucket := range broker.TakeInBuckets {
c, err := keys.JS.Consumer(ctx, "KV_"+bucket, broker.TakeInConsumer)
if err != nil {
return takeInBacklog{}, fmt.Errorf("the take-in's consumer on %s: %w", bucket, err)
}
consumers = append(consumers, c)
}
return backlogOf(ctx, consumers)
}
// kindTakeInStalled is the condition of a take-in that holds work and settles none (ADR 0544 §6); kindTakeInNotConsumed
// that of a bucket whose consumer could not be made.
const (
kindTakeInStalled = "take-in-stalled"
kindTakeInNotConsumed = "take-in-not-consumed"
)
func takeInStalledObservation(b takeInBacklog, since time.Time) conditions.Observation {
return conditions.Observation{Scope: conditions.ScopeCore, ID: "take-in", Token: kindTakeInStalled,
Kind: kindTakeInStalled, Severity: conditions.Urgent, Resolver: conditions.ResolverOperator, Source: "done keys",
Summary: fmt.Sprintf("the take-in of done keys has %d key(s) waiting and has settled none since %s",
b.Total(), since.UTC().Format(time.RFC3339))}
}
// takeInNotConsumedObservation is the condition of bucket's consumer not made, with its own id, so it is raised and
// cleared apart from a stall and from the other bucket's.
func takeInNotConsumedObservation(bucket string, err error) conditions.Observation {
summary := "the take-in's consumer on " + bucket + " is made"
if err != nil {
summary = fmt.Sprintf("the take-in's consumer on %s is not made, so its keys are not taken in until the next "+
"start: %v", bucket, err)
}
return conditions.Observation{Scope: conditions.ScopeCore, ID: "take-in-" + bucket, Token: kindTakeInNotConsumed,
Kind: kindTakeInNotConsumed, Severity: conditions.Urgent, Resolver: conditions.ResolverOperator,
Source: "done keys", Summary: summary}
}
// takeInMoving counts the keys handed to the take-in loop and the keys it settled.
type takeInMoving struct {
handed, settled atomic.Uint64
}
// stallWatch judges each look: stalled once there was work to do (keys not delivered yet, or keys handed to the loop
// and not settled) and nothing settled for takeInStall. A refused key waiting for its delay is no work of the loop's:
// with nothing else, a quiet mesh is not a stall; a loop that hangs on its batch is (the review of ADR 0544). Its first
// look always says it changed, so a condition a controller before this one raised is cleared once this one moves.
type stallWatch struct {
mu sync.Mutex
settled uint64
movedAt time.Time
stalled bool
known bool
}
// look takes a reading at now: the backlog, and how many keys were handed and settled; it answers whether the take-in
// stalls and whether that changed (always on the first look).
func (w *stallWatch) look(b takeInBacklog, handed, settled uint64, now time.Time) (stalled, changed bool) {
w.mu.Lock()
defer w.mu.Unlock()
work := b.Pending > 0 || handed > settled
if !work || settled != w.settled || w.movedAt.IsZero() {
w.movedAt = now
}
w.settled = settled
nowStalled := work && now.Sub(w.movedAt) >= takeInStall
changed = !w.known || nowStalled != w.stalled
w.stalled, w.known = nowStalled, true
return nowStalled, changed
}
// watchTakeInStall reads the backlog every minute and raises take-in-stalled while the take-in holds work and settles
// none, clearing it when it moves, and at its first look when it does not stall: a watchdog's timer, as research 072 §2
// keeps them until the health part leaves the controller.
func watchTakeInStall(ctx context.Context, backlog func(context.Context) (takeInBacklog, error), moving *takeInMoving) {
w := &stallWatch{}
tick := time.NewTicker(time.Minute)
defer tick.Stop()
for {
select {
case <-ctx.Done():
return
case <-tick.C:
}
b, err := backlog(ctx)
if err != nil {
continue
}
stalled, changed := w.look(b, moving.handed.Load(), moving.settled.Load(), time.Now())
keeper := conditionsFrom
if !changed || keeper == nil {
continue
}
w.mu.Lock()
since := w.movedAt
w.mu.Unlock()
o := takeInStalledObservation(b, since)
if stalled {
doneKeysSay("done keys: " + o.Summary)
_, _ = keeper.Observe(ctx, o)
} else {
_, _ = keeper.Clear(ctx, o.Key(), "the take-in moves")
}
}
}
+3 -1
View File
@@ -512,7 +512,9 @@ func (w *watchdogs) lostConsumers(ctx context.Context, heard []link.Advisory) (m
for _, a := range deleted { for _, a := range deleted {
_, err := w.js.Context().ConsumerInfo(a.Stream, a.Consumer) _, err := w.js.Context().ConsumerInfo(a.Stream, a.Consumer)
switch { 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] out[a.Stream+"."+a.Consumer] = want[a.Stream+"."+a.Consumer]
case err != nil: case err != nil:
return nil, fmt.Errorf("whether %s exists cannot be read: %w", link.ConsumerInWords(a.Stream, a.Consumer), err) return nil, fmt.Errorf("whether %s exists cannot be read: %w", link.ConsumerInWords(a.Stream, a.Consumer), err)
+14 -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 // 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 // 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 // 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" const DeclarationsStream = "NODES"
// RaiseAfterDeclarations is Raise once the declarations' stream is retired: the mesh's streams and consumers but the // RaiseAfterDeclarations is Raise once the declarations' stream is retired: the mesh's streams and consumers but the
@@ -181,3 +182,15 @@ func RaiseAfterDeclarations(r Raiser) error {
} }
return AssertMeshConsumers(r) return AssertMeshConsumers(r)
} }
// TakeInConsumer is the durable consumer the controller takes done keys in from, on the stream of each of TakeInBuckets
// (novox/hq ADR 0544 §1).
const TakeInConsumer = "mesh-controller-take-in"
// TakeInBuckets are the buckets the take-in consumes: the indexer's modules (each module's trunk value) and the builder's
// builds (each build's key).
var TakeInBuckets = []string{delivery.ModulesBucket, delivery.BuilderBuildsBucket}
// ConfigFollowConsumer is the durable consumer the controller follows mesh-config's resolved values with, on that
// bucket's stream (novox/hq ADR 0545).
const ConfigFollowConsumer = "mesh-controller-config-follow"
+20
View File
@@ -432,6 +432,16 @@ var DegradableGrants = []DegradableGrant{
Reason: "carried only at genesis: the controller composes its own permissions into the bus module's composition, " + Reason: "carried only at genesis: the controller composes its own permissions into the bus module's composition, " +
"and writes a composition's digest only after the bus applied it (novox/hq ADR 0539 clauses 2 and 7)", "and writes a composition's digest only after the bus applied it (novox/hq ADR 0539 clauses 2 and 7)",
}, },
{
Subject: "$JS.ACK.KV_" + delivery.ModulesBucket + "." + TakeInConsumer + ".>",
Reason: "refused: an acknowledgement refused leaves the key delivered again after its wait, and the take-in's first " +
"batch with something new composes the bus's node, which grants it (novox/hq ADR 0544 §1)",
},
{
Subject: "$JS.ACK.KV_" + delivery.BuilderBuildsBucket + "." + TakeInConsumer + ".>",
Reason: "refused: an acknowledgement refused leaves the key delivered again after its wait, and the take-in's first " +
"batch with something new composes the bus's node, which grants it (novox/hq ADR 0544 §1)",
},
{ {
Subject: "$KV." + GrantsBucket + ".>", Subject: "$KV." + GrantsBucket + ".>",
Reason: "carried only at genesis: the controller composes its own permissions and pushes them to the bus at its " + Reason: "carried only at genesis: the controller composes its own permissions and pushes them to the bus at its " +
@@ -707,6 +717,11 @@ func PermissionsFor(p Principal) (Permissions, error) {
// ever, refused by the list it already has. // ever, refused by the list it already has.
sub = append(sub, ControllerFollows...) sub = append(sub, ControllerFollows...)
pub = append(pub, "$JS.ACK.EVENTS."+ControllerName+".>", "$JS.ACK.EVENTS."+MergesConsumerName+".>") pub = append(pub, "$JS.ACK.EVENTS."+ControllerName+".>", "$JS.ACK.EVENTS."+MergesConsumerName+".>")
// And the take-in's two durable consumers on the indexer's and the builder's buckets (novox/hq ADR 0544 §1):
// acknowledged after each key is taken in, and no other consumer of those streams.
for _, bucket := range TakeInBuckets {
pub = append(pub, "$JS.ACK.KV_"+bucket+"."+TakeInConsumer+".>")
}
// **Where an enrolment's answer goes**, and `allow_responses` does not cover it. That // **Where an enrolment's answer goes**, and `allow_responses` does not cover it. That
// permits one reply to the reply subject of a message the user received — and a message a // permits one reply to the reply subject of a message the user received — and a message a
@@ -1258,6 +1273,11 @@ func PermissionsFor(p Principal) (Permissions, error) {
pub = append(pub, "mesh.seat."+v.Seat+"."+CallKind+"."+v.Verb+"."+CallerToken(p.Username())) pub = append(pub, "mesh.seat."+v.Seat+"."+CallKind+"."+v.Verb+"."+CallerToken(p.Username()))
} }
} }
// And its consumer of mesh-config's resolved values, acknowledged once a change is composed (novox/hq ADR 0545):
// where mesh-config serves, as its verb, so genesis grants nothing of mesh-config.
if len(p.AsksTheConfig) > 0 {
pub = append(pub, "$JS.ACK.KV_"+ConfigResolvedBucket+"."+ConfigFollowConsumer+".>")
}
} }
sort.Strings(pub) sort.Strings(pub)
+1 -1
View File
@@ -29,7 +29,7 @@ accounts {
] ]
users = [ users = [
{ user: "controller", password: "$2a$11$cccccccccccccccccccccc", permissions: { { user: "controller", password: "$2a$11$cccccccccccccccccccccc", permissions: {
publish: { allow: ["$JS.ACK.CONTROL.controller.>", "$JS.ACK.DEAD_LETTER_NOTICES.controller.>", "$JS.ACK.EVENTS.controller-merges.>", "$JS.ACK.EVENTS.controller.>", "$JS.API.>", "$KV.mesh-controller_asked.>", "$KV.mesh-controller_assignments.>", "$KV.mesh-controller_builds.>", "$KV.mesh-controller_calls.>", "$KV.mesh-controller_composed.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_declared-config.>", "$KV.mesh-controller_declared-packages.>", "$KV.mesh-controller_delivery-merges.>", "$KV.mesh-controller_grants.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$KV.mesh-controller_predefined.>", "$KV.mesh-controller_rollouts.>", "$KV.mesh-controller_states-announced.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "mesh.mod.mesh-controller.event.states", "mesh.node.>", "mesh.seat.mesh-controller.event.applied", "mesh.seat.mesh-controller.event.artifacts-gone", "mesh.seat.mesh-controller.event.assignment-changed", "mesh.seat.mesh-controller.event.bucket-deleted", "mesh.seat.mesh-controller.event.built-before", "mesh.seat.mesh-controller.event.checked", "mesh.seat.mesh-controller.event.condition-changed", "mesh.seat.mesh-controller.event.condition-cleared", "mesh.seat.mesh-controller.event.condition-raised", "mesh.seat.mesh-controller.event.declared-config-changed", "mesh.seat.mesh-controller.event.doctor-heartbeat", "mesh.seat.mesh-controller.event.healer-acted", "mesh.seat.mesh-controller.event.predefined-changed", "mesh.seat.mesh-controller.event.refused", "mesh.seat.mesh-controller.event.rolled-back", "mesh.seat.mesh-controller.event.secret-replaced", "mesh.seat.mesh-controller.tool.plans", "mesh.seat.mesh-controller.tool.push", "mesh.seat.mesh-controller.tool.upgrade", "mesh.seat.mesh-delivery.tool.close", "mesh.seat.mesh-delivery.tool.groups", "mesh.seat.mesh-delivery.tool.release", "mesh.seat.mesh-delivery.tool.stalled", "mesh.seat.mesh-delivery.tool.stop", "mesh.seat.mesh-delivery.tool.times", "mesh.seat.node-backup.tool.backed-up.*", "mesh.seat.node-backup.tool.now.*", "mesh.seat.node-intrusion-prevention.tool.banned.*", "mesh.seat.node-launcher.tool.secret.*", "mesh.seat.node-service-manager.tool.restart.*", "mesh.seat.operator-channel.tool.tell"] } publish: { allow: ["$JS.ACK.CONTROL.controller.>", "$JS.ACK.DEAD_LETTER_NOTICES.controller.>", "$JS.ACK.EVENTS.controller-merges.>", "$JS.ACK.EVENTS.controller.>", "$JS.ACK.KV_mesh-delivery-builder_builds.mesh-controller-take-in.>", "$JS.ACK.KV_mesh-delivery-indexer_modules.mesh-controller-take-in.>", "$JS.API.>", "$KV.mesh-controller_asked.>", "$KV.mesh-controller_assignments.>", "$KV.mesh-controller_builds.>", "$KV.mesh-controller_calls.>", "$KV.mesh-controller_composed.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_declared-config.>", "$KV.mesh-controller_declared-packages.>", "$KV.mesh-controller_delivery-merges.>", "$KV.mesh-controller_grants.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$KV.mesh-controller_predefined.>", "$KV.mesh-controller_rollouts.>", "$KV.mesh-controller_states-announced.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "mesh.mod.mesh-controller.event.states", "mesh.node.>", "mesh.seat.mesh-controller.event.applied", "mesh.seat.mesh-controller.event.artifacts-gone", "mesh.seat.mesh-controller.event.assignment-changed", "mesh.seat.mesh-controller.event.bucket-deleted", "mesh.seat.mesh-controller.event.built-before", "mesh.seat.mesh-controller.event.checked", "mesh.seat.mesh-controller.event.condition-changed", "mesh.seat.mesh-controller.event.condition-cleared", "mesh.seat.mesh-controller.event.condition-raised", "mesh.seat.mesh-controller.event.declared-config-changed", "mesh.seat.mesh-controller.event.doctor-heartbeat", "mesh.seat.mesh-controller.event.healer-acted", "mesh.seat.mesh-controller.event.predefined-changed", "mesh.seat.mesh-controller.event.refused", "mesh.seat.mesh-controller.event.rolled-back", "mesh.seat.mesh-controller.event.secret-replaced", "mesh.seat.mesh-controller.tool.plans", "mesh.seat.mesh-controller.tool.push", "mesh.seat.mesh-controller.tool.upgrade", "mesh.seat.mesh-delivery.tool.close", "mesh.seat.mesh-delivery.tool.groups", "mesh.seat.mesh-delivery.tool.release", "mesh.seat.mesh-delivery.tool.stalled", "mesh.seat.mesh-delivery.tool.stop", "mesh.seat.mesh-delivery.tool.times", "mesh.seat.node-backup.tool.backed-up.*", "mesh.seat.node-backup.tool.now.*", "mesh.seat.node-intrusion-prevention.tool.banned.*", "mesh.seat.node-launcher.tool.secret.*", "mesh.seat.node-service-manager.tool.restart.*", "mesh.seat.operator-channel.tool.tell"] }
subscribe: { allow: ["$JS.API.>", "$JS.EVENT.ADVISORY.CONSUMER.DELETED.>", "$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.>", "$SRV.INFO", "$SRV.INFO.mesh-controller", "$SRV.INFO.mesh-controller.>", "$SRV.PING", "$SRV.PING.mesh-controller", "$SRV.PING.mesh-controller.>", "$SRV.STATS", "$SRV.STATS.mesh-controller", "$SRV.STATS.mesh-controller.>", "_DELIVER.controller", "_DELIVER.controller.>", "_INBOX.controller.>", "mesh.control.>", "mesh.mod.*.event.provisioner.failing", "mesh.mod.*.event.provisioner.recovered", "mesh.mod.*.event.provisioner.retirement", "mesh.mod.gitea.event.pull.merged", "mesh.mod.gitea.event.pull.updated", "mesh.mod.mesh-catalog.event.catching-up", "mesh.mod.mesh-catalog.event.upgraded", "mesh.seat.mesh-build-machine.event.built", "mesh.seat.mesh-controller.tool.>", "mesh.seat.node-build-agent.event.built", "mesh.seat.operator-channel.event.decided.mesh-controller"] } subscribe: { allow: ["$JS.API.>", "$JS.EVENT.ADVISORY.CONSUMER.DELETED.>", "$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.>", "$SRV.INFO", "$SRV.INFO.mesh-controller", "$SRV.INFO.mesh-controller.>", "$SRV.PING", "$SRV.PING.mesh-controller", "$SRV.PING.mesh-controller.>", "$SRV.STATS", "$SRV.STATS.mesh-controller", "$SRV.STATS.mesh-controller.>", "_DELIVER.controller", "_DELIVER.controller.>", "_INBOX.controller.>", "mesh.control.>", "mesh.mod.*.event.provisioner.failing", "mesh.mod.*.event.provisioner.recovered", "mesh.mod.*.event.provisioner.retirement", "mesh.mod.gitea.event.pull.merged", "mesh.mod.gitea.event.pull.updated", "mesh.mod.mesh-catalog.event.catching-up", "mesh.mod.mesh-catalog.event.upgraded", "mesh.seat.mesh-build-machine.event.built", "mesh.seat.mesh-controller.tool.>", "mesh.seat.node-build-agent.event.built", "mesh.seat.operator-channel.event.decided.mesh-controller"] }
allow_responses: { max: 1, ttl: "1m" } allow_responses: { max: 1, ttl: "1m" }
} } } }
+116 -8
View File
@@ -7,6 +7,7 @@ import (
"errors" "errors"
"fmt" "fmt"
"sync/atomic" "sync/atomic"
"time"
"github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5"
"golang.org/x/crypto/bcrypt" "golang.org/x/crypto/bcrypt"
@@ -52,6 +53,8 @@ type BusUser struct {
// PasswordHash is what the composed file carries. The plaintext is returned once, by Mint, and // PasswordHash is what the composed file carries. The plaintext is returned once, by Mint, and
// then exists only where it was sealed. // then exists only where it was sealed.
PasswordHash string PasswordHash string
// Sealed says its credential reached its node: an issue's mint is not sealed until then (ADR 0543 §2).
Sealed bool
} }
// The kinds of bus user the mesh records. The same words the composer uses, so a row and a // The kinds of bus user the mesh records. The same words the composer uses, so a row and a
@@ -100,21 +103,126 @@ func (i *Inventory) RecordBusPassword(ctx context.Context, u BusUser, password s
return i.writeBusUser(ctx, u, string(hash)) return i.writeBusUser(ctx, u, string(hash))
} }
// writeBusUser is the row, whoever chose the password. // writeBusUser is the row, whoever chose the password, taken as sealed.
func (i *Inventory) writeBusUser(ctx context.Context, u BusUser, hash string) error { func (i *Inventory) writeBusUser(ctx context.Context, u BusUser, hash string) error {
return i.writeBusUserSealed(ctx, u, hash, true)
}
// writeBusUserSealed is the row, sealed now or not yet.
func (i *Inventory) writeBusUserSealed(ctx context.Context, u BusUser, hash string, sealed bool) error {
if _, err := i.store.Pool().Exec(ctx, if _, err := i.store.Pool().Exec(ctx,
`insert into bus_user (username, kind, node, module, password_hash) `insert into bus_user (username, kind, node, module, password_hash, sealed_at)
values ($1, $2, $3, $4, $5) values ($1, $2, $3, $4, $5, case when $6 then now() end)
on conflict (username) do update on conflict (username) do update
set kind = excluded.kind, node = excluded.node, module = excluded.module, set kind = excluded.kind, node = excluded.node, module = excluded.module,
password_hash = excluded.password_hash, minted_at = now()`, password_hash = excluded.password_hash, minted_at = now(), sealed_at = excluded.sealed_at`,
u.Username, u.Kind, u.Node, u.Module, hash); err != nil { u.Username, u.Kind, u.Node, u.Module, hash, sealed); err != nil {
return fmt.Errorf("cannot record the bus user %s: %w", u.Username, err) return fmt.Errorf("cannot record the bus user %s: %w", u.Username, err)
} }
return nil return nil
} }
// LockBusAccount holds the store's lock on one account until the answered release is called, so two issuers of it
// (the assign act, the assignments' watch, a registration's process accounts) never interleave their look, mint and
// seal: the hash the bus holds is always the password sealed to the node (novox/hq ADR 0543 §2, issue 1945).
func (i *Inventory) LockBusAccount(ctx context.Context, username string) (func(), error) {
conn, err := i.store.Pool().Acquire(ctx)
if err != nil {
return nil, fmt.Errorf("the lock on %s cannot be taken: %w", username, err)
}
if _, err := conn.Exec(ctx, `select pg_advisory_lock(hashtext($1))`, "bus_user/"+username); err != nil {
conn.Release()
return nil, fmt.Errorf("the lock on %s cannot be taken: %w", username, err)
}
return func() {
_, _ = conn.Exec(context.WithoutCancel(ctx), `select pg_advisory_unlock(hashtext($1))`, "bus_user/"+username)
conn.Release()
}, nil
}
// ForgetBusAccountsOf forgets the bus accounts of module on node, and the credentials sealed to the node for them (its
// `broker`, each process's `broker-<id>`, and with nodeOwned its provisioner client's): a module that no longer resolves
// on a node keeps no account there, so an assignment again mints and seals afresh, never trusting a row whose credential
// the node no longer holds (issue 1945, live on 2026-10-25 at 20:04: ace's mesh-build-go rows from 17:19 outlived its
// unassignment, nothing was issued again, and the module found no credential). The provisioner client's account is the
// node's (`provisioner.<node>`), forgotten only with nodeOwned. Each account's lock is held across its delete, so an issue
// in flight never writes a row back after it. It answers the accounts forgotten.
func (i *Inventory) ForgetBusAccountsOf(ctx context.Context, node, module string, nodeOwned bool) ([]string, error) {
if node == "" || module == "" {
return nil, errors.New("forgetting the bus accounts of no node or no module would forget others' accounts")
}
rows, err := i.store.Pool().Query(ctx,
`select username from bus_user where node = $1 and module = $2 and kind <> $3 and ($4 or kind <> $5)
order by username`, node, module, BusController, nodeOwned, BusProvisioner)
if err != nil {
return nil, err
}
var names []string
for rows.Next() {
var n string
if err := rows.Scan(&n); err != nil {
rows.Close()
return nil, err
}
names = append(names, n)
}
rows.Close()
if err := rows.Err(); err != nil {
return nil, err
}
for _, n := range names { // in order, one lock at a time per issuer: no two hold each other's
release, err := i.LockBusAccount(ctx, n)
if err != nil {
return nil, err
}
defer release()
}
tx, err := i.store.Pool().Begin(ctx)
if err != nil {
return nil, err
}
defer func() { _ = tx.Rollback(context.WithoutCancel(ctx)) }()
if _, err := tx.Exec(ctx, `delete from bus_user where username = any($1)`, names); err != nil {
return nil, fmt.Errorf("the bus accounts of %s on %s are not forgotten: %w", module, node, err)
}
if _, err := tx.Exec(ctx,
`delete from module_secret s using node n
where s.node = n.id and n.name = $1 and s.module = $2
and (s.name = 'broker' or s.name like 'broker-%' or ($3 and s.name = 'provisioner-broker'))`,
node, module, nodeOwned); err != nil {
return nil, fmt.Errorf("the bus credentials of %s on %s are not forgotten: %w", module, node, err)
}
return names, tx.Commit(ctx)
}
// MintBusPasswordUnsealed mints u's password, replacing whatever stands, with no seal yet: the account is not issued
// until MarkBusAccountSealed says its credential reached the node (ADR 0543 §2).
func (i *Inventory) MintBusPasswordUnsealed(ctx context.Context, u BusUser) (string, error) {
return i.mintBusPassword(ctx, u, false)
}
// MarkBusAccountSealed says the credential last minted for username is sealed to its node.
func (i *Inventory) MarkBusAccountSealed(ctx context.Context, username string) error {
_, err := i.store.Pool().Exec(ctx, `update bus_user set sealed_at = now() where username = $1`, username)
return err
}
// BusAccountIssued says whether an account is minted, and whether its credential is sealed to its node.
func (i *Inventory) BusAccountIssued(ctx context.Context, username string) (minted, sealed bool, err error) {
var at *time.Time
err = i.store.Pool().QueryRow(ctx, `select sealed_at from bus_user where username = $1`, username).Scan(&at)
if errors.Is(err, pgx.ErrNoRows) {
return false, false, nil
}
return err == nil, err == nil && at != nil, err
}
func (i *Inventory) MintBusPassword(ctx context.Context, u BusUser) (string, error) { func (i *Inventory) MintBusPassword(ctx context.Context, u BusUser) (string, error) {
return i.mintBusPassword(ctx, u, true)
}
// mintBusPassword mints, marked sealed when the caller seals it itself in the same act (every mint but an issue's).
func (i *Inventory) mintBusPassword(ctx context.Context, u BusUser, sealed bool) (string, error) {
if u.Username == "" || u.Kind == "" { if u.Username == "" || u.Kind == "" {
return "", errors.New("a bus user needs a username and a kind") return "", errors.New("a bus user needs a username and a kind")
} }
@@ -132,7 +240,7 @@ func (i *Inventory) MintBusPassword(ctx context.Context, u BusUser) (string, err
return "", fmt.Errorf("cannot hash a bus password: %w", err) return "", fmt.Errorf("cannot hash a bus password: %w", err)
} }
if err := i.writeBusUser(ctx, u, string(hash)); err != nil { if err := i.writeBusUserSealed(ctx, u, string(hash), sealed); err != nil {
return "", err return "", err
} }
return password, nil return password, nil
@@ -145,7 +253,7 @@ func (i *Inventory) MintBusPassword(ctx context.Context, u BusUser) (string, err
// password, and the composer refuses it rather than writing a user anybody is. // password, and the composer refuses it rather than writing a user anybody is.
func (i *Inventory) BusUsers(ctx context.Context) (map[string]BusUser, error) { func (i *Inventory) BusUsers(ctx context.Context) (map[string]BusUser, error) {
rows, err := i.store.Pool().Query(ctx, rows, err := i.store.Pool().Query(ctx,
`select username, kind, node, module, password_hash from bus_user order by username`) `select username, kind, node, module, password_hash, sealed_at is not null from bus_user order by username`)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@@ -153,7 +261,7 @@ func (i *Inventory) BusUsers(ctx context.Context) (map[string]BusUser, error) {
out := map[string]BusUser{} out := map[string]BusUser{}
for rows.Next() { for rows.Next() {
var u BusUser var u BusUser
if err := rows.Scan(&u.Username, &u.Kind, &u.Node, &u.Module, &u.PasswordHash); err != nil { if err := rows.Scan(&u.Username, &u.Kind, &u.Node, &u.Module, &u.PasswordHash, &u.Sealed); err != nil {
return nil, err return nil, err
} }
out[u.Username] = u out[u.Username] = u
@@ -4,7 +4,9 @@ package inventory
import ( import (
"context" "context"
"slices"
"testing" "testing"
"time"
"golang.org/x/crypto/bcrypt" "golang.org/x/crypto/bcrypt"
) )
@@ -33,7 +35,7 @@ func TestABusPasswordIsReturnedOnceAndOnlyItsHashIsKept(t *testing.T) {
} }
} }
func TestMintingAgainRotatesRatherThanAddsAUser(t *testing.T) { func TestMintingAgainRotatesAndMintingIfAbsentHasOneWinner(t *testing.T) {
inv := ForTest(t) inv := ForTest(t)
ctx := context.Background() ctx := context.Background()
@@ -62,6 +64,56 @@ func TestMintingAgainRotatesRatherThanAddsAUser(t *testing.T) {
if bcrypt.CompareHashAndPassword([]byte(hash), []byte(first)) == nil { if bcrypt.CompareHashAndPassword([]byte(hash), []byte(first)) == nil {
t.Fatal("the previous password still verifies, so a rotation revoked nothing") t.Fatal("the previous password still verifies, so a rotation revoked nothing")
} }
// **An issue's mint is not issued until its seal is marked, and forgotten with its assignment** (ADR 0543 §2, issue
// 1945, live on 2026-10-25 at 20:04): a row whose credential the node does not hold is not issued.
if minted, sealed, err := inv.BusAccountIssued(ctx, "one.shop"); err != nil || !minted || !sealed {
t.Fatalf("a mint that seals in its own act: minted %v sealed %v (%v)", minted, sealed, err)
}
if _, err := inv.MintBusPasswordUnsealed(ctx, aBusUser("racer")); err != nil {
t.Fatal(err)
}
if minted, sealed, err := inv.BusAccountIssued(ctx, "one.racer"); err != nil || !minted || sealed {
t.Fatalf("an issue's mint before its seal: minted %v sealed %v (%v)", minted, sealed, err)
}
if names, err := inv.BusMinted(ctx); err != nil || slices.Contains(names, "one.racer") {
t.Fatalf("an account not sealed counts as minted: %v (%v)", names, err)
}
if err := inv.MarkBusAccountSealed(ctx, "one.racer"); err != nil {
t.Fatal(err)
}
if _, sealed, _ := inv.BusAccountIssued(ctx, "one.racer"); !sealed {
t.Fatal("an account marked sealed is not")
}
// The lock on an account holds a second issuer until the first lets go.
release, err := inv.LockBusAccount(ctx, "one.racer")
if err != nil {
t.Fatal(err)
}
waiting := make(chan struct{})
go func() {
again, err := inv.LockBusAccount(ctx, "one.racer")
if err == nil {
again()
}
close(waiting)
}()
select {
case <-waiting:
t.Fatal("a second issuer took the lock the first held")
case <-time.After(200 * time.Millisecond):
}
release()
<-waiting
if _, err := inv.ForgetBusAccountsOf(ctx, "one", "racer", true); err != nil {
t.Fatal(err)
}
if minted, _, _ := inv.BusAccountIssued(ctx, "one.racer"); minted {
t.Fatal("an unassigned module's account was kept")
}
if minted, _, _ := inv.BusAccountIssued(ctx, "one.shop"); !minted {
t.Fatal("forgetting one module's accounts forgot another's")
}
} }
func TestForgettingANodeTakesItsBusUsersWithIt(t *testing.T) { func TestForgettingANodeTakesItsBusUsersWithIt(t *testing.T) {
+15
View File
@@ -210,6 +210,21 @@ func (i *Inventory) hasModule(ctx context.Context, name string) (bool, error) {
// This is the whole of noticing. Nothing here builds anything — it writes down that the two // This is the whole of noticing. Nothing here builds anything — it writes down that the two
// halves differ, which is what makes *is this current?* answerable without building, and what // halves differ, which is what makes *is this current?* answerable without building, and what
// makes a module that nobody rebuilt visible rather than silent. // makes a module that nobody rebuilt visible rather than silent.
// RecordBuiltFrom records that a registered module stands at commit, its manifest and fingerprint unchanged: a catalogue
// merge that left its sources alone (novox/hq ADR 0544 §3). The head moves with it, as a registration moves it; the
// module's builds and manifest stay as they are.
func (i *Inventory) RecordBuiltFrom(ctx context.Context, module, commit string) error {
tag, err := i.store.Pool().Exec(ctx,
`update module set built_from = $2, source_head = $2 where name = $1`, module, commit)
if err != nil {
return fmt.Errorf("cannot record %s at %s: %w", module, commit, err)
}
if tag.RowsAffected() == 0 {
return fmt.Errorf("%w: %s", ErrNoSuchModule, module)
}
return nil
}
func (i *Inventory) SourceMoved(ctx context.Context, module, head string) error { func (i *Inventory) SourceMoved(ctx context.Context, module, head string) error {
tag, err := i.store.Pool().Exec(ctx, tag, err := i.store.Pool().Exec(ctx,
`update module set source_head = $2, source_seen = now() where name = $1`, module, head) `update module set source_head = $2, source_seen = now() where name = $1`, module, head)
+10
View File
@@ -361,3 +361,13 @@ func (i *Inventory) AssignmentValue(ctx context.Context, node, module string) (c
k, found, err := c.Assignments.Get(ctx, assignmentKey(node, module)) k, found, err := c.Assignments.Get(ctx, assignmentKey(node, module))
return k.Value, found, err return k.Value, found, err
} }
// WatchAssignments is the assignments bucket's raw entries, as the chain's watch gives them: every current value, a Caught
// mark, then each write (novox/hq ADR 0543: an assignment that comes to stand issues its node's bus accounts).
func (i *Inventory) WatchAssignments(ctx context.Context) (<-chan chain.Entry, error) {
c, err := i.theChain(ctx)
if err != nil {
return nil, err
}
return c.Assignments.Watch(ctx)
}
@@ -0,0 +1,8 @@
-- A bus account says when the credential minted for it was sealed to its node (novox/hq ADR 0543 §2, issue 1945). The
-- assign act and the assignments' watch issue an account only when it is not issued: a row with no sealed credential
-- is not issued. On 2026-10-25 at 20:04 ace's mesh-build-go had a row minted at 17:19 whose credential the node no
-- longer held; minting nothing, the mesh sealed nothing, and the module crash-looped for its missing credential. Null
-- while an issue mints and seals; set once the credential is handed over. Every account standing before this migration
-- is taken as sealed: each works or worked, and its own issue sealed it.
alter table bus_user add column sealed_at timestamptz;
update bus_user set sealed_at = minted_at;
+4 -2
View File
@@ -226,8 +226,10 @@ func (i *Inventory) BusMinted(ctx context.Context) ([]string, error) {
return nil, err return nil, err
} }
out := make([]string, 0, len(kept)) out := make([]string, 0, len(kept))
for name := range kept { for name, u := range kept {
out = append(out, name) if u.Sealed { // an account whose credential is not sealed to its node is not issued (ADR 0543 §2)
out = append(out, name)
}
} }
sort.Strings(out) sort.Strings(out)
return out, nil return out, nil
+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" "strings"
"time" "time"
"github.com/jackc/pgx/v5/pgconn"
"github.com/jackc/pgx/v5/pgxpool" "github.com/jackc/pgx/v5/pgxpool"
"github.com/novox/mesh-controller/internal/envfile" "github.com/novox/mesh-controller/internal/envfile"
@@ -213,7 +214,7 @@ func (s *Store) Pool() *pgxpool.Pool { return s.pool }
// Close releases the connections. // Close releases the connections.
func (s *Store) Close() { func (s *Store) Close() {
if s.pool != nil { if s != nil && s.pool != nil {
s.pool.Close() s.pool.Close()
} }
} }
@@ -227,8 +228,10 @@ func (s *Store) Close() {
func (s *Store) Ready(ctx context.Context, within time.Duration) error { func (s *Store) Ready(ctx context.Context, within time.Duration) error {
deadline := time.Now().Add(within) deadline := time.Now().Add(within)
var last error var last error
wait := 250 * time.Millisecond
saidFull := false
for { for {
err := s.pool.Ping(ctx) err := pingStore(ctx, s)
if err == nil { if err == nil {
return nil return nil
} }
@@ -240,10 +243,48 @@ func (s *Store) Ready(ctx context.Context, within time.Duration) error {
return fmt.Errorf( return fmt.Errorf(
"the %s store did not answer within %s: %w", s.context, within, last) "the %s store did not answer within %s: %w", s.context, within, last)
} }
select { // A full store (SQLSTATE 53300) is not asked again every 250 ms: each ask is one more connection it refuses
case <-ctx.Done(): // while it is full. The wait doubles up to fullStoreMostWait, and the journal says once why (issues 1945 and
return errors.Join(ctx.Err(), last) // 1277, 2026-10-25 16:08: the controller's own retries kept the store full).
case <-time.After(250 * time.Millisecond): 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) }