Compare commits

..
Author SHA1 Message Date
jschoubben 4e2b8f32ce Pin mesh-sdk a3718126, whose bus retry never asks again a call its callee answered (issue 774)
mesh/delivery closed: the pull request closed unmerged
mesh/delivery-group group fix/sweep-2026-10-30 rejected: composed together: the checker's verdict at the members' heads (9a9df8453781): g14
mesh/check failed: g14: novox/mesh-catalog#565: modules/mesh-vault/mesh-vault: a compiled program added in 2066021fca5f and removed later: it stays in…
2026-10-30 18:43:13 +01:00
jschoubben dd366f8a3f Pin the group's final mesh-sdk 206f794e with its captured conformance files, and read the engine's superseded word on a secret move (issue 774)
The captured SDK (testdata/beside, conformance/events) is taken again at the pinned commit, as CAPTURED
says. The controller's mirror of MoveSecretAnswer gains `superseded` (mesh-host b9c24a7): the vault holds
another value, the live one, and the machine's older value is not moved and not asked again. The field
test names it. A superseded row is still counted with the moved ones: counting it apart needs a field and
its words in nodeMove, more than the one line asked.
2026-10-30 18:37:56 +01:00
jschoubben 93d167f0e8 Say rightly which own secrets the core ask leaves out: broker-address and broker-management are no bus accounts (issue 774)
The comment of c421629f named the controller's broker-address and broker-management as process accounts.
They are not: no process of the controller declares a bus; broker-address is the bus's address, read by
broker.FromEnvironment, and broker-management the retired AMQP management login, read by nothing. The
filter never touched them; only the words were wrong.
2026-10-30 18:36:18 +01:00
jschoubben cf4ca55596 Judge apply-failing by the client's failing-since, so a failure said in new words is measured from when it began (issue 774)
mesh-provisioner-client's report now carries failing-since (mesh-catalog 3aae6c08), kept while a module
stays failed however its words change. S21 measures a failed placing from it, falling back to the
report's time from an older client; the hysteresis of 6ffcea4f (an open condition stays while any
failure stands) is kept. This is the thorough fix that commit named.

Research 072 piece: milestone 3 (health leaves the controller): the judgement reads the owner's state.
2026-10-30 18:35:37 +01:00
jschoubben 77c54445dc Pin the group's mesh-sdk ef1ec1c9 and capture its conformance files there, so the sweep builds on kvlimit and the named readers (issue 774)
The sweep's controller changes use mesh-sdk's go/kvlimit and the ReadsAnnouncedBy that grants named
readers. At ef1ec1c9 mesh-config is a named reader of the controller's lease (its take-over reads it,
ADR 0561 §3), so the read test now expects it granted and a module not named refused. The captured SDK
(testdata/beside, conformance/events) is taken again at the pinned commit, as CAPTURED says; its file
is unchanged.
2026-10-30 18:27:54 +01:00
jschoubben 8c3d372322 Compose no placed user from the node-engine's manifest in the controller's bus composer too, as mesh-sdk #136 does (issue 774)
mesh-sdk #136 (42988a7) keeps the node-engine's manifest out of every placed user in buscompose: where a
placement names it, no user, no process and nothing the tool runner carries; the node's account alone
writes the engine's state, and a second writer has the writers table refuse the whole list (met twice on
2026-10-30: mesh-host_acts, then mesh-host_health). The controller still composes the bus's users with
its own copy (internal/broker Users), which lacked the guard. The same two predicates, Composed and
CarriedByRunner, now guard it.

Read on the bus at 18:30: the bus follows the grants part's list (mesh-grants_user-list.sealed), so the
controller's copy is not what the bus loads; it still composes the bus module's files and the controller's
own grants, where a refusal would stop that composition.

Quick fix: the same guard here. Thorough fix: delete the controller's copy of the bus composer and use
buscompose (about 18,000 lines of internal/broker with its tests, its genesis list and its golden file).
Chose quick: the controller is being deprecated and the grants part already holds the list; the copy goes
with the controller's composition of the bus module, not in a sweep.

Research 072 piece: 03-domains, mesh-grants (membership and grants leave the controller): until the copy is
removed, it composes as the SDK does.
2026-10-30 18:23:42 +01:00
jschoubben 519434241a Queue one state per call for the bus, bounded far above a burst, so a burst of read calls no longer drops call records (issue 774)
At 17:44:53 the controller said 30 calls (tools, nodes, modules) were "kept in memory only: the bus is
not taking calls' records". No write had failed (no "could not be kept" line): a burst of about 200 read
calls in one second queued a begin and a finish each into a channel of 256 while the one writer put
them one by one, and the channel overflowed.

The queue now holds the latest state of each waiting call (a finish replaces its own waiting begin, as
only the last state is kept anyway), in first-queued order, bounded at 8192 calls; the writer takes the
whole queue at each wake. A burst now costs one write per call and drops nothing.

Quick fix: a larger channel. Thorough fix: coalesce by call, so the writes a burst asks are halved and
the bound is spent on calls, not states. Chose thorough; both are small.

Research 072 piece: §2's "read calls said" row: the calls bucket is the controller's own state on the
bus, kept from events (each call's change), never re-read.
2026-10-30 18:21:58 +01:00
jschoubben c421629f6e Ask the vault for a bus process's account only once it is issued, so the vault never makes a random value under an account's name (issue 774)
At 16:36:51 the vault handed own/novox/nats/broker-nats-tools to the controller as a plain random own
secret, and owes the bus-account hashes of it and of the controller's broker-address and
broker-management for good ("holds no account the controller hands (not JSON)"). The controller decided
that: its send asks the vault (`get`) for every own secret of a core module, a bus process's account
among them, before the account is issued; the vault, holding none, makes one.

The send now leaves out of that ask a process's account nobody issued (the same rule as its warning
line); the machine keeps the controller's stand-in until the account is issued and handed to the vault.
The vault side (not minting an account-shaped name, and replacing the random value already held) is
mesh-catalog's mesh-vault, with build-catalog.

Quick fix: this filter on the ask. Thorough fix: process accounts minted by the vault itself (ADR 0113 §6,
research 072 milestone 6), so no part asks for one before it exists. Chose quick: milestone 6 is not
started, and the filter removes the cause on the controller's side now.

Research 072 piece: milestone 6 (identity and bus accounts to the vault): until then the controller asks
the vault only for what exists.
2026-10-30 18:19:26 +01:00
jschoubben 07cecd287c Leave modules the mesh provides itself out of the lines that say a module lacks a fingerprint (issue 774)
mesh-wireguard is provided by the mesh (sdkdelivery.ProvidedByMesh): no repository builds it and no
fingerprint will ever name it, yet every composition said it composes "with the node's own: no
fingerprint of the new form names them", and the switch's list counted it "not known". Both now leave
such a module out; its resources still go with the node's own composition, as before.

Quick and thorough are one here: the SDK names the modules the mesh provides, and every line that judges
fingerprints filters by it.

Research 072 piece: §6 milestone 2 (the composer reacts; its words say only what a change can fix).
2026-10-30 18:19:26 +01:00
jschoubben e6ecc2d15e Put mesh-delivery's merge words through mesh-sdk's kvlimit, so a word too large for its bucket is refused once and never retried (issue 774)
The controller's other writer of the 16 KiB bucket of merges, the keeper of mesh-delivery's merge words,
called every put failure "not now". A word too large would be retried as the forge's merges were. It now
puts through kvlimit.Put: a value that does not fit is refused before it is put and returned as an answer
(the events consumer takes it, max-deliver 5 then a dead letter), and only another failure is retried.
It is the one write of the controller whose failure was retried as "not now" (searched: every Put,
Create and Update wrapped in ErrTryAgain).

Needs the group's mesh-sdk head (go/kvlimit, 7b02f63 or later).

Quick fix: none smaller. Thorough fix: every bucket write through kvlimit; this is the only one whose
refusal was retried, the others fail once and say so. No new test: kvlimit is tested in mesh-sdk, and an
integration test here would push the repository further over its budget (rules-are-weighed).

Research 072 piece: 03-domains (Control shrinks): the merge words stay the controller's only until the
registration of a moved module leaves it.
2026-10-30 18:16:09 +01:00
jschoubben d543ff7126 Judge each read for its reader with mesh-sdk's ReadsAnnouncedBy, so the parts named for the lease are granted it, and say a refusal once (issue 774)
2536 reads of mesh-controller.lease were refused on 2026-10-30 (mesh-vault, mesh-grants, the assigner,
artifact-store-tools, the watchman, mesh-config). The controller composed the bus's users with its own
copy of a reader-less judgement (internal/broker ReadsAnnounced), which knows no named readers: the lease
is announced readable by nobody and read by the parts named for it (ADR 0547 clause 9, ADR 0561 §3). The
copy is deleted; BusRecords judges each read for its module with buscompose.ReadsAnnouncedBy. Each
refusal was also printed on every composition, about 420 times a read; it is now said once per process.

mesh-config's read of the lease stays refused: it reads the lease for its take-over (hand-back ends at
the lease write, ADR 0561 §3) and is not among the named readers. That is mesh-sdk's NamedReaders,
routed to the lead.

Quick fix: mark the lease readable in the controller's announcement (tried and reverted: it would open
the lease to every module and contradict ADR 0561 §3). Thorough fix: one judgement, the SDK's, for every
composer. Chose thorough.

Research 072 piece: 03-domains, mesh-grants (bus membership and grants leave the controller): the
controller composes reads with the same SDK function the grants part uses.
2026-10-30 18:15:17 +01:00
jschoubben d63eb83ed3 Clear a condition whose check no longer exists, so a removed row or probe leaves no alarm open for good (issue 774)
About 35 conditions from 10-20 to 10-27 still showed as current. Read on the bus at 18:06: S23's three
failed plans (last observed 10-24) and D14's four stalled deliveries (last observed 10-24) come from a
signals row and a probe this controller no longer has; nothing reconciles their source, so nothing clears
them. The other old ones are observed every tick and are not stale by the controller: S9's dead letters
are still held, S15's hand acts are still kept, and DM's 26 stuck keys are the watchman's (it never writes
a key whose watched key aged out as moved) — routed to mesh-delivery.

Each watchdogs tick now clears every open condition whose source has a row's or a probe's form (S12, D4,
DW, D-root) and is in neither the signals table nor the probe registry, saying the check was removed.
Sources of other forms are left to their own word.

Quick fix: clear the seven by hand. Thorough fix: the controller judges its own retired checks on every
tick, so the next removed row or probe clears what it raised. Chose thorough: rows and probes are being
moved out of the controller one by one (ADR 0564), each removal would leave this again.

Research 072 piece: milestone 3 (health leaves the controller): the conditions keeper drops what no check
of its own observes, so the move to mesh-conditions starts from a bucket of live conditions.
2026-10-30 18:10:29 +01:00
jschoubben 6ffcea4f1d Keep apply-failing raised while any failure stands, so a failure said again in new words never clears it (issue 774)
machine.novox.apply-failing cleared at 16:56 while the vault refused every secret there: the provisioner
client says a failure again whenever its words change, with a new time, and S21 measured each module's
failure from that newest time, so every new wording hid the failure for one apply interval (raised again
17:01). The same holds for the node-engine's own failed apply, whose since restarts with new words.

The bound now gates raising only: once the condition is open (read from the keeper's open conditions on
the bus), it stays while any module's apply fails on that machine, and clears when none does.

Quick fix: this hysteresis on the open condition in the controller. Thorough fix: the provisioner client
carries the start of a failure run (failing-since) in its report, so every reader measures from the true
start; that is mesh-catalog's mesh-provisioner-client, routed to the lead. Chose the quick fix here,
because the check moves to mesh-watchdogs (ADR 0564) and the condition's own state already proves the run.

Research 072 piece: milestone 3 (health leaves the controller): the judgement reads bus state, the open
condition, rather than a report's time.
2026-10-30 18:08:09 +01:00
jschoubben 48d9f22fa5 Retire the controller's merges consumer and read merges from the forge bridge's stream, so a held merge can never again queue every merge or look like a deaf controller (issue 774)
The controller-merges consumer (no max-deliver, one at a time) held the de-spiegel merge e25e1beb from
10-29 09:39: keepForgeMerge put an announcement with its Paths into delivery-merges (16 KiB values), got
10054, retried "not now" forever, and 167 merges queued behind it. Every five-minute hold outran the deaf
bound, so core.controller.deaf flapped.

Merges are mesh-delivery's: the gate merges on the Approve and the forge bridge says each merge on its
stream FORGE (pull request, head, merge commit), kept a week. The controller's copy was a leftover, read
only at a moved module's registration (ADR 0420 §7, ADR 0432). That read now finds the merge on FORGE;
the consumer, its follow, its handler and its hold delays are removed, and every assertion removes the
retired consumer from the bus.

Kind fix for the watchdog: the event loop records when a message it held back is due again, and deafness
counts from then, so no hold longer than the bound (on any consumer) reads as deafness.

Quick fix: strip Paths from the kept announcement and add max-deliver with a dead letter. Thorough fix:
remove the duplicate duty and make the deaf watchdog hold-aware. Chose thorough: the controller is being
deprecated and the merges already have an owner; the hold-vs-deaf confusion is a kind, not one consumer.

Research 072 piece: 03-domains (Control shrinks to placement and composition; the delivery parts own
what a merge is) and milestone 3 (health signals stop sharing the controller's loop assumptions).
2026-10-30 18:05:46 +01:00
jschoubben a4426190b0 Merge pull request 'Accept the credential form none in a provisioner's plan, so design 47's release delivers (issue 774)' (#483) from fix/774-none-form into main
Reviewed-on: #483
2026-10-30 11:23:41 +00:00
jschoubben 2154ea8e3c Accept the credential form none in a provisioner's plan, so the release's catalogue, which states it on every offer without a credential, no longer has every plan refused and every module held (issue 774)
mesh/check checking 2154ea8e3c4e on a4426190b0c7
mesh/delivery merged: source.main moved to novox/mesh-controller@a4426190b0c7
2026-10-30 12:18:36 +01:00
mesh-tools 812c528479 Merge pull request 'Pin the release's mesh-sdk in the controller, so it reads design 47's catalog until its lease is handed away (issue 774)' (#482) from feat/774-l3-reland into main 2026-10-30 11:01:57 +00:00
jschoubben 8b32e04e2f Judge the controller's module check against the catalogue it is released with, captured at the group's catalogue head f3a57dcc, so a group that brings a word the trunk's catalogue lacks does not fail on the trunk (issue 774)
mesh/check checking 8b32e04e2f8b on 6fe9efa71219 | novox/mesh-controller#482 in group feat/774-l3-reland
mesh/delivery merged: source.main moved to novox/mesh-controller@812c528479e1
mesh/delivery-group group feat/774-l3-reland merged: every member is merged
The release's fourth check failed TestModuleCheckPassesTheCatalogue with 27 problems. The checker clones the catalogue
beside a controller's tests at its trunk, even in a group holding a catalogue pull request (mesh-delivery checker
internal/check/beside.go besideFor, ADR 0484 clause 4). The release's sdk refuses an offer stating no credential form,
and the trunk's catalogue states none until the release's catalogue merges.

- Quick fix (not chosen): recapture alone, which the test never reads while MESH_CHECK_BESIDE is set.
- Chosen: the test reads the captured copy, now at the group's catalogue f3a57dcc, through beside.CapturedDir. Every other
  beside test still reads the clone. The check itself is unchanged: every rule, every manifest. Moving CAPTURED to the
  catalogue a change is released with is part of that change.

Run as the checker runs it (catalogue at main, mesh-host at main, mesh-sdk at the pin, mesh-controller's trunk
beside): the whole unit and integration suite passes, 1706 tests, 8 skipped. The test also passes with the catalogue at
f3a57dcc beside, and with nothing beside.
2026-10-30 10:38:29 +01:00
jschoubben 79f44386d2 Merge main (delivery 1, the parse-only sdk) into L3's re-land, and pin mesh-sdk 2da0de7, the release's final head (issue 774)
mesh/check failed: unit: github.com/novox/mesh-controller/cmd/mesh-controller.TestModuleCheckPassesTheCatalogue: the catalogue does not pass its own c…
mesh/delivery-group group feat/774-l3-reland rejected: composed together: the checker's verdict at the members' heads (8bad7efb3e7a): unit: github.com/novox/me…
mesh/delivery closed: a newer head of the same pull request
The conflicts were go.mod, go.sum and testdata/beside/CAPTURED alone, taken from the release. 2da0de7 is 6e4a301's code;
its conformance events are unchanged.
2026-10-30 01:13:41 +01:00
mesh-tools 6fe9efa712 Merge pull request 'Pin the parse-only mesh-sdk in the controller, so it reads design 47's manifests (issue 774)' (#481) from fix/774-new-words-first into main 2026-10-30 00:10:12 +00:00
jschoubben d96542cd25 Pin the parse-only mesh-sdk 78c9ec8 in place of 6e4a301, so main's controller reads the release's new words and means nothing new by them (issue 774)
mesh/check pass d96542cd2503 on e04846e9fa49 | unit 9174 run, 546 kept, 51 skipped | integration 9952 run, 659 kept, 60 skipped | over bound: unit, in…
mesh/delivery merged: source.main moved to novox/mesh-controller@6fe9efa71219
mesh/delivery-group group fix/774-new-words-first merged: every member is merged
78c9ec8 is sdk main 307b623 (tree of main's pin dbcaa29) with one parse-only commit: an offer with no form still means
pair. The fixture changes the full sdk forced are reverted to main's; the sdk's conformance events are captured at
78c9ec8 and unchanged. TestModuleCheckPassesTheCatalogue passes, on the captured copy and on mesh-catalog main 96b129e5
cloned beside. Whole unit and integration suites: 1706 tests, 8 skipped for a registry, a container, a refusing bus or a
lab's run.
2026-10-30 00:42:04 +01:00
jschoubben 80dc6980b7 Pin mesh-sdk 6e4a301 on main's controller, so it reads the release's new manifest words before the release uses them (issue 774)
No code of ours changes; the sdk's conformance events are unchanged and captured at 6e4a301. The fixtures of
identity_bound_test and issued_test say the credential form their rows meant (pair, none), since the sdk no longer reads
a missing form as pair (6b8c694).

NOT GREEN: TestModuleCheckPassesTheCatalogue fails. The sdk refuses every offer that states no credential form, and
main's catalogue states none (27 offers). Recapturing cannot fix it: this pin and main's catalogue cannot both stand.
2026-10-30 00:30:05 +01:00
jschoubben 80df1b6361 Pin mesh-sdk 6e4a301, the release's final sdk (issue 774)
The sdk's conformance events are unchanged at 6e4a301 and captured so. The whole integration suite passes at this head
against a local store and bus: 1706 tests, 8 skipped for a registry, a container, a refusing bus or a lab's run.
2026-10-30 00:20:10 +01:00
jschoubben 816cf96459 Pin mesh-sdk 007b6fd, the release's final sdk, so the controller reads the release's manifests and composes as the group does (issue 774)
The sdk now refuses an offer that says no credential form (6b8c694, ADR 0570). So:
- the catalogue copy in testdata/beside is captured again at the release branch's catalogue 8ae8e88d, whose offers each
  say one, and the sdk's conformance events at 007b6fd, as CAPTURED asks;
- the fixtures of identity_bound_test and issued_test say the form their rows meant: pair for a login of each
  consumer's own, none for a keyless provision. No behaviour of the controller changed.
2026-10-30 00:07:04 +01:00
jschoubben 3f05df5d1a Pin mesh-sdk 2e5cc4f, so the controller reads the release's new manifest words (issue 774)
The controller on dbcaa29 parses manifests strictly and refuses `bus` on an artifact and `max-age-days` on a log, both
in the re-land group's catalogue. The SDK's conformance events are captured again at the pinned commit, as
testdata/beside/CAPTURED asks; they are unchanged.
2026-10-29 22:19:39 +01:00
mesh-tools e04846e9fa Merge pull request 'Pin mesh-sdk dbcaa29, so the controller composes with the group fix/1960-no-hold-spread's SDK (issue 1960)' (#480) from fix/1960-no-hold-spread into main 2026-10-29 15:35:31 +00:00
jschoubben 020c6a99b0 Pin mesh-sdk dbcaa29 and capture its conformance events there, so the controller composes with the group fix/1960-no-hold-spread's SDK (issue 1960)
mesh/check pass 020c6a99b035 on 5a9766d4a47a | unit 3318 run, 1951 kept, 38 skipped | integration 3994 run, 2125 kept, 38 skipped | over bound: unit, …
mesh/delivery merged: source.main moved to novox/mesh-controller@e04846e9fa49
mesh/delivery-group group fix/1960-no-hold-spread merged: every member is merged
2026-10-29 16:18:13 +01:00
mesh-tools 5a9766d4a4 Merge pull request 'Pin mesh-sdk 38dbe1d, so the controller knows the seat mesh-placement (issue 1126, delivery 2 of 2)' (#479) from feat/1126-placement-seat into main 2026-10-28 18:33:05 +00:00
jschoubben f46a08772f Pin mesh-sdk 38dbe1d, the group's head, so the controller knows the seat mesh-placement its holder claims (ADR 0567, issue 1126)
mesh/check failed: integration: meshbuildgo/cmd/mesh-build-go.TestAComposeAskIsBuiltOnceAndAnsweredOnTheBus: the compose ask mvsg85b34d25b4 never ende…
mesh/delivery merged: source.main moved to novox/mesh-controller@5a9766d4a47a
mesh/delivery-group group feat/1126-placement-seat merged: every member is merged
2026-10-28 18:49:02 +01:00
jschoubben 1c2adfdca9 Build with the group's mesh-sdk, so the controller knows the seat mesh-placement its holder claims (ADR 0567, issue 1126) 2026-10-28 18:14:58 +01:00
mesh-tools 0fbb5ef359 Merge pull request 'The controller gives its lease back at the stop signal and says a takeover once, so a controller delivery raises no urgent alarm (issue 906, ADR 0558)' (#478) from fix/906-lease-release into main 2026-10-27 00:07:46 +00:00
jschoubben 8f84731896 Name the epoch given back in the lease's delete marker, so only a holder's own release counts as a handover (issue 906, ADR 0558 §3, review round 1)
mesh/delivery merged: source.main moved to novox/mesh-controller@0fbb5ef35991
mesh/check being carried to 8f847318963c on 0fbb5ef35991 | from 2dafbe60e939
A plain delete marker on the lease key is also written by a command's own
lease and by a take given back over the floor; counted as a handover, it
would record a crashed holder's epoch released and say nothing. Release now
writes the marker itself, compare-and-set on the key's revision, with the
header Mesh-Lease-Released naming its epoch, and the next holder ends only
that epoch as released. A takeover whose predecessor cannot be read is now
raised as an expiry, not only printed.
2026-10-27 00:55:30 +01:00
jschoubben fbaa463a38 Give the controller lease back at the stop signal and say a takeover once, so a controller delivery raises no urgent alarm (issue 906, ADR 0558)
The 2026-10-26 replacements showed the lease was given back, but last: the
shutdown took 10 s, the gate had closed at half the key's age, and the release
recorded the epoch released only through the gate, so the next holder found
it expired and S12 said it, urgent, for an hour.

Now the stop signal gives the key back first and records the epoch released
whenever the delete succeeded; a candidate reads the key's delete marker on
the bus to tell a handover from an expiry; and a real takeover (lost or
expired) is raised and cleared once by the controller that took it, instead
of an hour of S12 observation.

Research 072 piece moved: the lease's takeover read from its own key on the
bus and said as an event (03-domains, mesh-controller row; 04-findings 2,
the watchdogs' age reading).
Quick fix weighed: drop the gate check from the release record only; it
leaves the release last in the shutdown and the hour-long alarm for a real
takeover.
2026-10-27 00:49:13 +01:00
mesh-tools 2dafbe60e9 Merge pull request 'Broker tests read a consumer's pending count through pendingOf wherever they expect a count after a publish (issue 1126)' (#477) from fix/1126-tests-wait-for-consumer-counts into main 2026-10-26 22:57:59 +00:00
jschoubben 8d58afea6c Broker tests read a consumer's pending count through pendingOf, which waits up to 10 s, wherever they expect a count after a publish (issue 1126)
mesh/delivery merged: source.main moved to novox/mesh-controller@2dafbe60e939
mesh/check being carried to 8d58afea6cb8 on 2dafbe60e939 | from b611f7375a47
The server counts a consumer's pending after it acknowledges the publish, so a read at once failed the try of
main on a loaded ace (TestAResetOfTheEventsConsumerLeavesTheMergesAndTheirConsumerAlone, 2026-10-26 22:24 UTC).
Fixed for the kind, not the one test: all six reads of a positive count go through the package's own pendingOf,
whose deadline was 1 s.

Research 'The controller, NATS-native': none; this changes tests only, not a part of the controller.
2026-10-26 23:28:03 +01:00
mesh-tools b611f7375a Merge pull request 'Pin mesh-sdk main, whose seats promise own and settle, so the catalogue passes the module check again (issue 1126)' (#476) from fix/1126-controller-pins-sdk-main into main 2026-10-26 22:08:19 +00:00
jschoubben 253bbcea2d Pin mesh-sdk main f178323, whose delivery seat promises settle beside the vault's own, so the catalogue check passes against mesh-catalog main f0f7b673 (issue 1126)
mesh/delivery merged: source.main moved to novox/mesh-controller@b611f7375a47
mesh/check checking 253bbcea2da8 on b611f7375a47
Moves no piece of research 'The controller, NATS-native': a repin of the seat catalogue, no controller code changed.
2026-10-26 23:03:47 +01:00
mesh-tools 2c6382346f Merge pull request 'Pin the mesh-sdk whose vault seat promises own, so the catalogue passes the module check again (issue 1126)' (#475) from fix/1126-vault-seat-own into main 2026-10-26 22:00:46 +00:00
jschoubben b8cafccaf6 Pin mesh-sdk at e78e881, whose vault seat promises own, so the catalogue check passes against mesh-catalog main (issue 1126)
mesh/delivery merged: source.main moved to novox/mesh-controller@2c6382346f48
mesh/delivery-group group fix/1126-vault-seat-own merged: every member is merged
mesh/check failed: unit: github.com/novox/mesh-controller/cmd/mesh-controller.TestModuleCheckPassesTheCatalogue: the catalogue does not pass its own c…
Moves no piece of research 'The controller, NATS-native': a repin for a seat verb list, no controller code changed.
2026-10-26 22:40:34 +01:00
mesh-tools 82d212a2a1 Merge pull request 'The controller answers each node's placement from its own store, the one verb of its exit (issue 1126)' (#474) from feat/1126-nodes-compose into main 2026-10-26 21:11:21 +00:00
jschoubben df8ebf6a77 Pin mesh-sdk 9796444 (l2-next: feat/1126-nodes-compose merged with sdk main #101, the standing-test race fixes) (issue 1126)
mesh/delivery merged: source.main moved to novox/mesh-controller@82d212a2a1ff
mesh/delivery-group group feat/1126-nodes-compose merged: every member is merged
mesh/check checking df8ebf6a77c1 on 82d212a2a1ff | novox/mesh-controller#474 in group feat/1126-nodes-compose
2026-10-26 22:05:41 +01:00
jschoubben dc414fbfef Pin mesh-sdk 5f45175 (l2-next: feat/1126-nodes-compose merged with sdk main) after merging main (issue 1126) 2026-10-26 21:59:21 +01:00
jschoubben fd891688ac Pin mesh-sdk feat/1126-nodes-compose 03dce12, the group's head: the checker refuses a pin that is neither trunk nor the group's head (issue 1126)
mesh/delivery closed: a newer head of the same pull request
mesh/delivery-group group feat/1126-nodes-compose checking: 4 of 6 member(s) ready
mesh/check checking fd891688acdf on 2ecfbb3a29cb | novox/mesh-controller#474 in group feat/1126-nodes-compose
2026-10-26 20:35:49 +01:00
jschoubben 9d8a5a7ede Review of L2, round 2: the placement export is checked within an existing integration test over the same fresh mesh, so it adds no test of its own (the test budget; issue 1126)
mesh/delivery closed: a newer head of the same pull request
mesh/check checking 9d8a5a7ede90 on 2ecfbb3a29cb | novox/mesh-controller#474 in group feat/1126-nodes-compose
2026-10-26 20:23:51 +01:00
jschoubben 4e9e5c27d4 Review of L2: the export's test reads each printed placement back as the file the seed writes, at its version (S1); pin mesh-sdk 3968849, CAPTURED taken again (issue 1126)
mesh/check failed: unit: ..test: ✔ a transient empty list for four passes retires nothing; five stable passes retire, data kept (290.408814ms) | 4e9e5…
mesh/delivery closed: a newer head of the same pull request
2026-10-26 20:12:49 +01:00
jschoubben 7c8a2e03dd Pin mesh-sdk feat/1126-nodes-compose 9713c30: placement carries a settings-form module's values (issue 1126)
mesh/check failed: ace: novox/mesh-catalog#505 does not merge into the trunk: files in conflict: .gitignore | 7c8a2e03dd8e on 2ecfbb3a29cb | unit 1245…
mesh/delivery closed: a newer head of the same pull request
2026-10-26 19:40:43 +01:00
jschoubben 89f52a6504 Pin mesh-sdk 53c2def, the group's head, and take the SDK's conformance copy again at it (issue 1126)
mesh/delivery closed: a newer head of the same pull request
mesh/check failed: unit: git.novox.be/novox/mesh-catalog/checks/nobinaries.TestTheCatalogueHoldsNoCompiledProgram: modules/mesh-config/cmd/mesh-config…
2026-10-26 19:28:39 +01:00
jschoubben fb2f5a2962 The controller answers each node's placement from its own store, read-only, for the seed of the catalogue's placement files: the one verb of its exit (issue 1126, ADR 0555 §6; the operator 2026-10-26 19:06:40) 2026-10-26 19:16:09 +01:00
mesh-tools 2ecfbb3a29 Merge pull request 'The controller takes mesh-sdk a7a1fbd3, so its composer gives build modules the node's CPU share (issue 1806, ADR 0553/0554 change B)' (#473) from fix/controller-sdk-cpu-share into main 2026-10-26 15:41:03 +00:00
jschoubben 9ad8715871 Bump mesh-sdk to a7a1fbd3 so the build seats' runtimes are composed to yield (issue 1806)
mesh/delivery merged: source.main moved to novox/mesh-controller@2ecfbb3a29cb
mesh/check pass 9ad8715871ea on 2ecfbb3a29cb | carried from e1a55f099b4e | unit 1023 run, 171 kept, 2 skipped | integration 1701 run, 214 kept, 9 skip…
The controller is the only composer, and its pin (10b6e202) predates mesh-sdk #95: without the bump the
build runtimes never get yields {priority idle, cpu-share} or MESH_BUILD_CPU_SHARE (ADR 0553, ADR 0554,
change B). The operator said yes to this one change of the frozen controller on 2026-10-26 15:54.

A composition at both pins over every node's assignment differs only in mesh-build-go, mesh-build-node and
mesh-build-image (the yields and the share word); every other composition is byte-identical. The SDK's
conformance capture moves with the pin, as testdata/beside/CAPTURED asks.

Research 072 (the controller, NATS-native): no piece moved; a pin bump the operator allowed.
2026-10-26 16:34:55 +01:00
mesh-tools e1a55f099b Merge pull request 'Seat users pulled by kind may read their own seat's dead letters (issue 734)' (#472) from fix/734-lost-ask into main 2026-10-26 13:40:43 +00:00
jschoubben ccc7ba334c broker: DeadLetterReads grants only what a pulled ordered consumer needs, as mesh-sdk buscompose (issue 734, review round 1)
mesh/delivery merged: source.main moved to novox/mesh-controller@e1a55f099b4e
mesh/delivery-group group fix/734-lost-ask merged: every member is merged
mesh/check checking ccc7ba334c82 on 08a261ac1a6d | novox/mesh-controller#472 in group fix/734-lost-ask
NATS-native research piece: none moved; the copy stays in step with mesh-sdk until mesh-grants composes the list.
2026-10-26 14:17:00 +01:00
jschoubben 65d5ed3367 broker: a user of a seat pulled by kind reads its seat's dead letters, as mesh-sdk buscompose composes it (issue 734)
The checker hears from DEAD_LETTERS which of its asks a build seat gave up on, and ends the try or asks the check again;
the controller composes today's user list, so its copy of the composer carries the grant too. NATS-native research piece:
none moved; the composition itself moves to mesh-grants (ADR 0547), and this keeps the copy in step with mesh-sdk.
2026-10-26 13:51:35 +01:00
mesh-tools 08a261ac1a Merge pull request 'Declarations test: the indexer stand-in answers the trunk value at the trunk's commit alone (issue 1945)' (#471) from fix/declarations-test-stand-in into main 2026-10-26 12:30:18 +00:00
jschoubben 23226384e5 Answer the indexer's trunk value at the trunk's commit alone in the declarations test, so its take-in records mesh-host's build (issue 1945)
mesh/delivery merged: source.main moved to novox/mesh-controller@08a261ac1a6d
mesh/check being carried to 23226384e529 on 08a261ac1a6d | from 350f8ebdfe5f
The stub answered the trunk for every commit, so since registering on the
fingerprint (666d1b68, #465) mesh-host, registered at an older commit, looked
already at the trunk's fingerprint: the take-in only moved its commit, no build
was recorded, and the fingerprints never named the new node-engine. The real
indexer holds no value at a commit it never indexed.

(cherry picked from commit 779533764b)
2026-10-26 13:29:04 +01:00
mesh-tools 350f8ebdfe Merge pull request 'Broker: the bus node may follow the grants part's sealed user list and the grants part may write it, the operator's second one-grant exception (issue 1945)' (#470) from feat/1945-user-list-switch into main 2026-10-26 12:21:44 +00:00
jschoubben 09d2cc5056 Broker: the bus node may follow the grants part's sealed user list and the grants part may write it, the operator's second one-grant exception (issue 1945, ADR 0547 clauses 7-8)
mesh/delivery merged: source.main moved to novox/mesh-controller@350f8ebdfe5f
mesh/delivery-group group feat/1945-user-list-switch merged: every member is merged
mesh/check checking 09d2cc5056ef on 47e64f0284a9 | novox/mesh-controller#470 in group feat/1945-user-list-switch
2026-10-26 13:03:15 +01:00
mesh-tools 47e64f0284 Merge pull request 'Broker: the grants part may create a placed module's declared buckets, logs and streams, creation only (issue 1790; the operator's one-grant exception)' (#469) from fix/1790-grants-create-buckets into main 2026-10-26 08:17:19 +00:00
jschoubben 6927b02d4d Grant the grants part creating the streams manifests declare, so a new module state no longer waits for a controller start
mesh/delivery merged: source.main moved to novox/mesh-controller@47e64f0284a9
mesh/delivery-group group fix/1790-grants-create-buckets merged: every member is merged
mesh/check checking 6927b02d4d58 on fda4859a95b8 | novox/mesh-controller#469 in group fix/1790-grants-create-buckets
The operator's one more grant in the frozen controller's composer (2026-10-26 08:54), beside the
memberships grant of 22:31 (ADR 0547 clause 1 and 5): mesh-grants' principal may publish
$JS.API.STREAM.CREATE.<stream> and $JS.API.STREAM.INFO.<stream> for each bucket
(KV_<module>_<state>), log (LOG_<module>_<log>) and module stream a placed module's manifest
declares, each by its own name, and nothing wider. The writers table lets it create, never update
or delete. Byte-equal with mesh-sdk's buscompose (fix/1790-grants-create-buckets), whose expected
lists this code wrote; the same test file holds both copies. Nothing else changes, and the mesh-sdk
pin stays.

Research 072: moves the grants piece (the grants part split out of the controller) one step, the
creation of a module's declared bus objects; the controller's raise at start stays until it loses
its lease.

Issue: 1790
2026-10-26 09:02:44 +01:00
jschoubben fda4859a95 Merge pull request 'The bus composer grants mesh-grants publishing memberships, nothing more (ADR 0547, the operator's one grant of 2026-10-25 22:31)' (#468) from fix/1945-memberships into main
Reviewed-on: #468
2026-10-25 22:44:04 +00:00
jschoubben a81717e53c Seal the seeded account in the adapter-input test, the second fixture of its kind (issue 1945, PR #467)
mesh/delivery merged: source.main moved to novox/mesh-controller@fda4859a95b8
mesh/delivery-group group fix/1945-memberships merged: every member is merged
mesh/check checking a81717e53cf4 on 7cd4d18bd54e | novox/mesh-controller#468 in group fix/1945-memberships
The kind: a test that seeds a bus account without sealing it, which an account counted as issued only once sealed
(ADR 0543 §2) no longer counts. The whole integration level, run locally, found two: the addresses test (fixed
before) and TestTheAdaptersInputCarriesTheOwnSecretsTheBoxesTheLicencesAndTheBus. Test only.
2026-10-25 23:30:20 +01:00
jschoubben d01839c29c Grant the grants part reading every membership; seal the seeded account in the addresses test (ADR 0547 §5, review; issue 1945)
mesh/delivery-group group fix/1945-memberships rejected: composed together: the checker's verdict at the members' heads (2253ba8bbe9b): integration: github.com…
mesh/delivery closed: a newer head of the same pull request
mesh/check failed: integration: github.com/novox/mesh-controller/cmd/mesh-controller.TestTheAdaptersInputCarriesTheOwnSecretsTheBoxesTheLicencesAndThe…
The grants part reads each module's last membership before it publishes, so it is granted the direct get on every
membership subject beside the publish (review of mesh-grants, B1).

The test TestTheControlPlaneIsToldWhereTheNodePutTheStoreAndTheBroker failed on main since PR #467: it seeds
anchor.mesh-controller with no seal, and an account counts as issued only once sealed (ADR 0543 §2). The rule is right;
the fixture is fixed, sealing the seeded credential the controller already uses. Test only.
2026-10-25 22:51:25 +01:00
jschoubben f67d15536b Grant the grants part publishing the modules' memberships, and nothing more (ADR 0547 §5, issue 1945)
No membership is published since release 6 (issue 1945 notes 42, 47), so a seat-holding module placed now cannot
serve its seat or pull asks (mesh-build-go on ace, 2026-10-25). The operator decided at 22:31 that the grants part
publishes them from the first milestone: its principal, module mesh-grants, may publish mesh.assignment.*.* and no
stream definition. The SDK's copy of this composer gets the same line, so the two stay byte-equal.

Research 072 (the controller, NATS-native): this moves the memberships of its Bus row to the grants part.
2026-10-25 22:32:35 +01:00
136 changed files with 2650 additions and 1023 deletions
+42 -6
View File
@@ -13,6 +13,7 @@ import (
"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"
"github.com/novox/mesh-controller/internal/lease"
"github.com/novox/mesh-controller/internal/link"
@@ -72,6 +73,14 @@ type actor struct {
// lease over in a second rather than in a renewal of five (novox/hq issue 591). A renewal is given
// the lease's own RenewEvery to answer however often it is made, so a slow bus does not lose it.
renewEvery, poll time.Duration
// releasing is held for the whole of a release, so a second caller — the stop signal's and serve's own
// deferred one — waits until the lease is given back and its epoch recorded, never past it to the store's
// close (novox/hq issue 906).
releasing sync.Mutex
// keeper and inv are where a takeover is said (sayTakeover): set once the serving controller's condition
// store opens; nil before, and in a command.
keeper *conditions.Keeper
inv *inventory.Inventory
}
// theLease is this process's standing to the lease.
@@ -174,8 +183,11 @@ func (a *actor) forACommand(ctx context.Context) (uint64, error) {
return holder.Epoch, nil
}
// release gives back what this process holds, at its end.
// release gives back what this process holds, at its end — or at the stop signal, before the rest of the
// process is taken down (serveUnderTheLease). A second call waits for the first to finish.
func (a *actor) release() {
a.releasing.Lock()
defer a.releasing.Unlock()
a.mu.Lock()
closing := a.close
a.close = nil
@@ -266,20 +278,40 @@ func (a *actor) serveUnderTheLease(ctx context.Context, inv *inventory.Inventory
}
a.mu.Lock()
a.close = func() {
if held, err := l.Epoch(); err == nil {
// Given back first, then recorded: an epoch is recorded released only when its key is gone, and
// whatever the gate says by then — a shutdown longer than half the key's age closed it, and the epoch
// was left open for the next holder to find "expired" (novox/hq issue 906, 2026-10-26).
if given := l.Release(context.Background()); given != 0 {
ending, cancel := context.WithTimeout(context.Background(), 5*time.Second)
if err := inv.EndEpoch(ending, held, inventory.EpochReleased); err != nil {
say("how epoch %d ended could not be recorded: %v", held, err)
if err := inv.EndEpoch(ending, given, inventory.EpochReleased); err != nil {
say("how epoch %d ended could not be recorded: %v", given, err)
}
cancel()
}
l.Release(context.Background())
js.Close()
}
a.mu.Unlock()
// **The stop signal gives the lease back first** (novox/hq issue 906): the moment ctx ends, before the rest
// of the process is taken down, the gate closes and the key goes, so the next controller takes the lease
// at once and the epoch is recorded released. Waiting for the process's own end left the key unrenewed for
// as long as the shutdown took — ten seconds on 2026-10-26, past the gate's half-age.
go func() {
<-ctx.Done()
if a.holds(l) {
say("stopping: the controller lease is given back first")
}
a.release()
}()
return gone, nil
}
// holds is whether this process holds l now.
func (a *actor) holds(l *lease.Lease) bool {
a.mu.Lock()
defer a.mu.Unlock()
return a.held == l
}
// took is the lease taken: recorded, earlier epochs nobody gave back ended as expired, kept.
func (a *actor) took(ctx context.Context, l *lease.Lease, inv *inventory.Inventory, epoch uint64, gone chan struct{}) {
a.mu.Lock()
@@ -288,15 +320,19 @@ func (a *actor) took(ctx context.Context, l *lease.Lease, inv *inventory.Invento
h := holderOf(instance)
recording, cancel := context.WithTimeout(ctx, 10*time.Second)
expired, err := inv.TookEpoch(recording, inventory.Epoch{Epoch: epoch, Instance: h.Instance, Host: h.Host,
Build: h.Build, Taken: time.Now()})
Build: h.Build, Taken: time.Now()}, l.GivenBack())
cancel()
if err != nil {
fmt.Printf("epoch %d could not be recorded as taken, so a stale refusal from it will not name it: %v\n", epoch, err)
}
if given := l.GivenBack(); given != 0 {
fmt.Printf("the lease was given back by the controller of epoch %d: taken at once, at epoch %d\n", given, epoch)
}
for _, e := range expired {
fmt.Printf("the controller of epoch %d (%s) stopped renewing the lease without giving it back: it is "+
"taken over at epoch %d\n", e.Epoch, e.Instance, epoch)
}
a.sayTakeover(ctx, epoch)
go l.Keep(ctx)
go func() {
<-l.Lost()
@@ -78,6 +78,8 @@ func TestTheRegistryTrustAndEveryImageFollowThePortTheNodeGaveTheStore(t *testin
func TestTheControlPlaneIsToldWhereTheNodePutTheStoreAndTheBroker(t *testing.T) {
open := aMesh(t)
// Issue 1126: the placement export over this fresh mesh, before anything below changes it.
placementExportIsEachNodesPlacementFromTheStore(t, open)
ctx := t.Context()
raw, err := os.ReadFile("../../module.json")
if err != nil {
@@ -111,6 +113,11 @@ func TestTheControlPlaneIsToldWhereTheNodePutTheStoreAndTheBroker(t *testing.T)
Kind: inventory.BusController, Node: "anchor", Module: "mesh-controller"}, "bootstrap"); err != nil {
t.Fatal(err)
}
// Seeded with the credential the controller already uses, so it is sealed: an account counts as issued only once
// sealed (ADR 0543 §2, PR #467), and genesis's seed is the credential in use.
if err := open.inventory.MarkBusAccountSealed(ctx, "anchor.mesh-controller"); err != nil {
t.Fatal(err)
}
if _, err := assign(ctx, open, "anchor", "mesh-controller"); err != nil {
t.Fatal(err)
}
+22 -5
View File
@@ -30,6 +30,11 @@ const packageManagerKey = "the package manager ("
// says a re-apply only when how it went changed, so the same failure is one report however long it lasts.
// Names the refusal's or the failure's words; cleared by the clean apply the node-engine then says. Not while
// the machine is silent, asleep or not known: S1 says silent, and asleep is not lost.
//
// **The bound gates raising, never keeping** (sweep 2026-10-30, issue 774). A report says a failure again whenever its
// words change, with a new time: on novox the vault's refusal of every secret replaced a bus refusal at 16:56, and the
// condition cleared for five minutes while every module still failed. Once raised, it stays raised while any failure
// stands on that machine, judged by the open condition the keeper holds on the bus, not by a report's time.
func watchApplyFailing(f *signalFacts) []conditions.Observation {
heard := map[string]bool{}
for _, m := range f.machines {
@@ -40,7 +45,7 @@ func watchApplyFailing(f *signalFacts) []conditions.Observation {
var out []conditions.Observation
said := map[string]bool{}
for _, d := range f.doing {
if !d.Wrong() || d.Since == nil || f.now.Sub(*d.Since) <= applyEvery || !heard[d.Node] {
if !d.Wrong() || d.Since == nil || !heard[d.Node] || (f.now.Sub(*d.Since) <= applyEvery && !f.applyFailingOpen[d.Node]) {
continue
}
out = append(out, applyFailing(d, f.now))
@@ -51,7 +56,8 @@ func watchApplyFailing(f *signalFacts) []conditions.Observation {
// on a machine whose node-engine applied what it was sent.
failed := map[string][]inventory.ProvReport{}
for _, r := range f.provFailed {
if r.Status != inventory.ProvFailed || f.now.Sub(r.At) <= applyEvery || !heard[r.Node] || said[r.Node] {
if r.Status != inventory.ProvFailed || !heard[r.Node] || said[r.Node] ||
(f.now.Sub(r.Since()) <= applyEvery && !f.applyFailingOpen[r.Node]) {
continue
}
failed[r.Node] = append(failed[r.Node], r)
@@ -62,17 +68,28 @@ func watchApplyFailing(f *signalFacts) []conditions.Observation {
return out
}
// applyFailingOpen are the machines whose S21 condition is open among the open conditions.
func applyFailingOpen(open []conditions.Condition) map[string]bool {
out := map[string]bool{}
for _, c := range open {
if c.Kind == kindApplyFailing && c.Subject.Machine != "" {
out[c.Subject.Machine] = true
}
}
return out
}
// provisioningFailing is one machine's S21 condition for the modules whose files its provisioner client failed to
// place.
func provisioningFailing(m string, failed []inventory.ProvReport, now time.Time) conditions.Observation {
sort.Slice(failed, func(a, b int) bool { return failed[a].Module < failed[b].Module })
var names, words []string
since := failed[0].At
since := failed[0].Since()
for _, r := range failed {
names = append(names, r.Module)
words = append(words, r.Module+": "+oneLine(orNoReason(r.Reason)))
if r.At.Before(since) {
since = r.At
if r.Since().Before(since) {
since = r.Since()
}
}
o := conditions.Observation{Scope: conditions.ScopeMachine, ID: m, Kind: kindApplyFailing, Machine: m,
+8 -2
View File
@@ -56,9 +56,15 @@ func TestModuleCheckJudgesBetweenTheManifestsGiven(t *testing.T) {
}
}
// The real catalogue passes the command, the way it passes the test that used to be the only check.
// The real catalogue passes the command, the way it passes the test that used to be the only check: the catalogue this
// controller is released with, captured in testdata/beside at the commit CAPTURED names. Not the clone a merge check puts
// beside, which is the catalogue's trunk even in a delivery group holding a catalogue pull request (ADR 0484 clause 4):
// a controller whose check reads a word that group's catalogue brings would fail on the trunk's catalogue before the
// group merges both (issue 774: the release's sdk refuses an offer that states no credential form, and the catalogue's
// trunk states none until the release's catalogue merges). Moving CAPTURED to the catalogue a change is released with is
// part of that change.
func TestModuleCheckPassesTheCatalogue(t *testing.T) {
root := beside.Catalogue(t)
root := filepath.Join(beside.CapturedDir(t, "mesh-catalog"), "modules")
paths, err := manifestsUnder(root)
if err != nil || len(paths) == 0 {
t.Fatalf("no manifests under %s: %v", root, err)
@@ -217,6 +217,11 @@ func TestTheAdaptersInputCarriesTheOwnSecretsTheBoxesTheLicencesAndTheBus(t *tes
Module: "objects"}, "a-password"); err != nil {
t.Fatal(err)
}
// Minted and sealed to its node: an account counts as issued, and is named among the minted, only once sealed
// (ADR 0543 §2, PR #467).
if err := inv.MarkBusAccountSealed(ctx, "anchor.objects"); err != nil {
t.Fatal(err)
}
if err := inv.PutBusMembership(ctx, "laptop", "sealed-membership"); err != nil {
t.Fatal(err)
}
+1 -1
View File
@@ -250,7 +250,7 @@ func prepareSend(ctx context.Context, open *stores, node string) ([]string, erro
if n.Stated && n.Contract >= compose.VaultBoxContract {
boxes, err := coreBoxesFor(ctx, inv, node, resolved.Plan.Modules, func(m catalogue.Manifest) bool {
return handing.Handing(m)
}, true)
}, true, in.BusMinted)
if err != nil {
return nil, err
}
@@ -13,6 +13,7 @@ import (
"github.com/nats-io/nats.go/jetstream"
"github.com/novox/mesh-controller/internal/broker"
"github.com/novox/mesh-controller/internal/builderkeys"
"github.com/novox/mesh-controller/internal/identity"
"github.com/novox/mesh-controller/internal/link"
"github.com/novox/mesh-controller/internal/testbus"
@@ -76,13 +77,18 @@ func aComposingMesh(t *testing.T) *composingMesh {
return m
}
// trunkKeys are a done key and the indexer's trunk value naming it.
// trunkKeys are a done key and the indexer's trunk value naming it. The indexer holds that value at the trunk's commit
// alone, as the real one does: answering it for any commit made a module registered at an older commit look already
// registered at the trunk's fingerprint, so the take-in only moved its commit and recorded no build (issue 1945).
type trunkKeys struct {
keyBuckets
trunk sdkdelivery.Module
}
func (k trunkKeys) ModuleAt(context.Context, string, string) (sdkdelivery.Module, error) {
func (k trunkKeys) ModuleAt(_ context.Context, _ string, commit string) (sdkdelivery.Module, error) {
if commit != "" && !sameCommitID(commit, k.trunk.Commit) {
return sdkdelivery.Module{}, builderkeys.ErrNoValue
}
return k.trunk, nil
}
+2 -11
View File
@@ -88,8 +88,6 @@ func composeModulesByNode(ctx context.Context, open *stores, nodes []string) (ma
}
return u.Policy(), u.From, u.Why
}
// The memberships of each node's modules are issued by its composition (memberships.go): read once for the pass.
issuing, issuingErr := membershipsFor(ctx, open, link.OverNATS{Conn: bus.Conn(), JS: bus.Context()})
sort.Strings(nodes)
out := map[string][]string{}
errsOf := map[string][]error{}
@@ -117,13 +115,6 @@ func composeModulesByNode(ctx context.Context, open *stores, nodes []string) (ma
errsOf[node] = append(errsOf[node], fmt.Errorf("%s: %w", node, err))
return
}
// Its modules' memberships, before its compositions (issue 249): a failure is the node's, and its compositions
// are still written; the take-in that asked tries the node again.
if issuingErr != nil {
errsOf[node] = append(errsOf[node], fmt.Errorf("%s: %w", node, issuingErr))
} else if err := issuing.issue(ctx, node, declared); err != nil {
errsOf[node] = append(errsOf[node], fmt.Errorf("%s: %w", node, err))
}
// The provisioner's plans of the modules the node's composition hands to it, for the assignments mirror.
if err := open.inventory.RecordComposedPlans(ctx, node, plansKept(declared)); err != nil {
errsOf[node] = append(errsOf[node], fmt.Errorf("%s: its provisioner's plans: %w", node, err))
@@ -133,11 +124,11 @@ func composeModulesByNode(ctx context.Context, open *stores, nodes []string) (ma
errsOf[node] = append(errsOf[node], fmt.Errorf("%s: %w", node, err))
return
}
if len(moved) > 0 {
if moved := builtByTheMesh(moved); len(moved) > 0 {
fmt.Printf("compositions: %s's %s compose with the node's own: no fingerprint of the new form names them\n",
node, strings.Join(moved, ", "))
}
if len(left) > 0 {
if left := builtByTheMesh(left); len(left) > 0 {
fmt.Printf("compositions: %s's %s are left out of the node's own: the controller knows no fingerprint of the "+
"new form for them, and the node's want or have names their own composition (issue 1945)\n",
node, strings.Join(left, ", "))
+12 -4
View File
@@ -69,7 +69,15 @@ type coreBoxes struct {
// coreModulesOf is every module of the plan on the core's list whose own secrets this machine's declaration carries,
// with their plain own secret names, sorted: one handed to the provisioner has its secrets from the vault through the
// client, and one issued outside the mesh is never the vault's to seal here.
func coreModulesOf(modules []catalogue.Manifest, handed func(catalogue.Manifest) bool) map[string][]string {
//
// **Nor a process's bus account nobody issued yet** (sweep 2026-10-30, issue 774). Asked with `get`, the vault holds
// none and makes a random value under the account's name: own/novox/nats/broker-nats-tools at 16:36:51. Until the
// controller issues the account and hands it to the vault, the machine has the controller's stand-in (prepareSend), as
// every module below the vault's contract does. (The controller's own broker-address and broker-management are no
// accounts: the bus's address, which broker.FromEnvironment reads, and the retired AMQP management login; no process
// declares them, so this leaves them asked as before.)
func coreModulesOf(modules []catalogue.Manifest, handed func(catalogue.Manifest) bool, node string,
minted []string) map[string][]string {
out := map[string][]string{}
for _, m := range modules {
if !catalogue.CarriesVaultSealed(m.Module) || (handed != nil && handed(m)) {
@@ -77,7 +85,7 @@ func coreModulesOf(modules []catalogue.Manifest, handed func(catalogue.Manifest)
}
var names []string
for name, own := range m.OwnSecrets.Plain() {
if own.IssuedBy != catalogue.IssuedOutside {
if own.IssuedBy != catalogue.IssuedOutside && processAccountWarning(m, node, name, minted) == "" {
names = append(names, name)
}
}
@@ -208,8 +216,8 @@ var coreVault = func(ctx context.Context) coreAsk {
// coreBoxesFor is coreBoxesOf for one machine, read from and kept in the store, and said on a send.
func coreBoxesFor(ctx context.Context, inv *inventory.Inventory, node string, modules []catalogue.Manifest,
handed func(catalogue.Manifest) bool, allocating bool) (coreBoxes, error) {
core := coreModulesOf(modules, handed)
handed func(catalogue.Manifest) bool, allocating bool, minted []string) (coreBoxes, error) {
core := coreModulesOf(modules, handed, node, minted)
if len(core) == 0 {
if allocating {
settleVaultSilent(ctx, node)
+28 -1
View File
@@ -124,8 +124,35 @@ func TestTheCoresModulesAreTheListsWithTheirOwnSecrets(t *testing.T) {
{Module: "postgres", OwnSecrets: catalogue.OwnSecrets{"superuser": {Path: "/c"}}},
{Module: "mesh-provisioner", OwnSecrets: catalogue.OwnSecrets{"broker": {Path: "/d"}}},
}
got := coreModulesOf(modules, func(m catalogue.Manifest) bool { return m.Module == "mesh-provisioner" })
got := coreModulesOf(modules, func(m catalogue.Manifest) bool { return m.Module == "mesh-provisioner" }, "anchor", nil)
if want := map[string][]string{"nats": {"broker"}}; !reflect.DeepEqual(got, want) {
t.Fatalf("got %v, want %v", got, want)
}
}
// Sweep 2026-10-30 (issue 774): a bus process's account is asked of the vault only once it is issued; before, the vault
// would make a random value under the account's name.
func TestAnUnissuedProcessAccountIsNotAskedOfTheVault(t *testing.T) {
nats, err := catalogue.ParseManifest([]byte(`{"module":"nats","version":"1","emits":["x"],
"own-secrets":{"broker":{"path":"/a"},"broker-nats-tools":{"path":"/b","owner":"nats-tools"}},
"resources":[{"id":"account","type":"user","name":"nats-tools"},
{"id":"nats-tools","type":"process","run":["./nats-tools"],"user":"nats-tools","bus":{"emits":["x"]}}]}`))
if err != nil {
t.Fatal(err)
}
if len(nats.ProcessBuses()) == 0 {
t.Fatal("the fixture declares no bus process")
}
for name, c := range map[string]struct {
minted []string
want []string
}{
"not issued": {nil, []string{"broker"}},
"issued": {[]string{processUser("anchor", "nats", "nats-tools")}, []string{"broker", "broker-nats-tools"}},
} {
got := coreModulesOf([]catalogue.Manifest{nats}, nil, "anchor", c.minted)
if !reflect.DeepEqual(got["nats"], c.want) {
t.Errorf("%s: asks %v, want %v", name, got["nats"], c.want)
}
}
}
@@ -0,0 +1,31 @@
//go:build integration
package main
import (
"context"
"testing"
"github.com/nats-io/nats.go/jetstream"
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/process"
"github.com/novox/mesh-controller/internal/broker"
)
// sayForgeMerge is the forge bridge saying a merge on its stream FORGE, which it makes when missing.
func sayForgeMerge(ctx context.Context, t *testing.T, js *broker.JetStream, repo string, number int, head, merge string) {
t.Helper()
api, err := jetstream.New(js.Conn())
if err != nil {
t.Fatal(err)
}
if _, err := api.CreateOrUpdateStream(ctx, jetstream.StreamConfig{Name: forgeStream,
Subjects: []string{process.ModuleStreamSubject(sdkdelivery.ForgeBridgeModule, "forge.>")}}); err != nil {
t.Fatal(err)
}
subject, body := forgeMerged(t, repo, number, head, merge)
if _, err := api.Publish(ctx, subject, body); err != nil {
t.Fatal(err)
}
}
+106
View File
@@ -0,0 +1,106 @@
package main
import (
"context"
"encoding/json"
"strings"
"testing"
"github.com/nats-io/nats.go/jetstream"
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
)
// fakeForge is the forge bridge's stream as forgeMergeAt reads it: messages by sequence, each on a subject.
type fakeForge struct {
subjects []string
data [][]byte
}
func (f fakeForge) last(_ context.Context, subject string) (uint64, []byte, error) {
for i := len(f.subjects) - 1; i >= 0; i-- {
if f.subjects[i] == subject {
return uint64(i + 1), f.data[i], nil
}
}
return 0, nil, jetstream.ErrMsgNotFound
}
func (f fakeForge) next(_ context.Context, subject string, from uint64) (uint64, []byte, error) {
for i := int(from) - 1; i < len(f.subjects); i++ {
if i >= 0 && f.subjects[i] == subject {
return uint64(i + 1), f.data[i], nil
}
}
return 0, nil, jetstream.ErrMsgNotFound
}
func forgeMerged(t *testing.T, repo string, number int, head, merge string) (string, []byte) {
t.Helper()
owner, name, _ := strings.Cut(repo, "/")
subject, err := forgeMergeSubject(owner, name)
if err != nil {
t.Fatal(err)
}
body, err := json.Marshal(sdkdelivery.Forge{V: 1, Kind: sdkdelivery.ForgeMerged, Repository: repo, Branch: "main",
Number: number, Head: head, Merge: merge, Root: "t0", Trees: map[string]string{".": "t1"}})
if err != nil {
t.Fatal(err)
}
return subject, body
}
// **A moved module's merge is read from the forge bridge's stream** (sweep 2026-10-30, issue 774): the controller keeps
// no copy of the forge's merges, so its registration finds the pull request and head of a merge commit on FORGE, the
// newest first and then every merge the stream keeps; a merge of another repository or commit is never taken for it.
func TestAMergeIsFoundOnTheForgeBridgesStream(t *testing.T) {
var f fakeForge
add := func(repo string, number int, head, merge string) {
s, b := forgeMerged(t, repo, number, head, merge)
f.subjects, f.data = append(f.subjects, s), append(f.data, b)
}
add("novox/mesh-host", 6, "h6", "c6c6c6c6c6")
add("novox/mesh-catalog", 9, "h9", "c7c7c7c7c7")
add("novox/mesh-host", 7, "h7", "c7c7c7c7c7")
f.subjects, f.data = append(f.subjects, f.subjects[0]), append(f.data, []byte("not a forge message"))
add("novox/mesh-host", 8, "h8", "c8c8c8c8c8")
for name, c := range map[string]struct {
repo, commit string
number int
found bool
}{
"the newest merge": {"novox/mesh-host", "c8c8c8c8c8", 8, true},
"an older merge, found by scanning": {"novox/mesh-host", "c6c6c6c6c6", 6, true},
"a short commit": {"novox/mesh-host", "c7c7c7c", 7, true},
"a commit of another repository": {"novox/mesh-catalog", "c6c6c6c6c6", 0, false},
"a repository with no merge": {"novox/hq", "c6c6c6c6c6", 0, false},
} {
owner, name2, _ := strings.Cut(c.repo, "/")
k, found, err := forgeMergeAt(t.Context(), f, owner, name2, c.commit)
if err != nil {
t.Fatalf("%s: %v", name, err)
}
if found != c.found || k.Number != c.number {
t.Errorf("%s: found %v #%d, want %v #%d", name, found, k.Number, c.found, c.number)
}
if found && (k.Owner != owner || k.Repo != name2 || k.HeadSHA == "") {
t.Errorf("%s: the merge reads as %+v", name, k.SourceMoved)
}
}
}
// Sweep 2026-10-30 (issue 774): a module the mesh provides itself is never named as lacking a fingerprint.
func TestAModuleTheMeshProvidesIsNotSaidToLackAFingerprint(t *testing.T) {
for _, c := range []struct {
in, want []string
}{
{[]string{"mesh-wireguard"}, nil},
{[]string{"claude-code", "mesh-wireguard", "power"}, []string{"claude-code", "power"}},
{nil, nil},
} {
if got := builtByTheMesh(c.in); strings.Join(got, ",") != strings.Join(c.want, ",") {
t.Errorf("%v: %v, want %v", c.in, got, c.want)
}
}
}
+2 -2
View File
@@ -23,10 +23,10 @@ func TestModuleCheckRefusesAnIdentityThatOverflowsWhatItRequires(t *testing.T) {
return p
}
objects := write("objects", `{"module":"objects","version":"1",
"provides":[{"name":"s3-bucket","scope":"mesh","identity":{"max":20,"in":"an S3 access key"}}],
"provides":[{"name":"s3-bucket","scope":"mesh","identity":{"max":20,"in":"an S3 access key"},"credential":{"form":"pair"}}],
"receives":{"s3-bucket":"/var/lib/mesh/objects/mesh.json"}}`)
resolver := write("resolver", `{"module":"resolver","version":"1",
"provides":[{"name":"wildcard-resolution","scope":"mesh","identity":false}]}`)
"provides":[{"name":"wildcard-resolution","scope":"mesh","identity":false,"credential":{"form":"none"}}]}`)
album := write("photoalbum", `{"module":"photoalbum","version":"1","requires":["s3-bucket"]}`)
nm := write("networkmanager", `{"module":"networkmanager","version":"1","requires":["wildcard-resolution"]}`)
+7 -2
View File
@@ -182,6 +182,9 @@ func TestANeedIsIssuedToItsHolderOnlyWhereTheProviderIsMovedAndItsSoftwareIssues
offer := catalogue.Offer{Name: "gitea-read", Scope: catalogue.ScopeMesh}
if issued {
offer.Credential = &catalogue.OfferCredential{IssuedBy: catalogue.IssuedBySoftware, Tool: "gitea_agent_token"}
} else {
// A login of each consumer's own, said: an offer naming no form has no credential (mesh-sdk 6b8c694).
offer.Credential = &catalogue.OfferCredential{Form: catalogue.CredentialFormPair}
}
return catalogue.Manifest{Module: "gitea", Version: "1", Moved: moved, Provides: []catalogue.Offer{offer}}
}
@@ -216,8 +219,10 @@ func TestANeedIsIssuedToItsHolderOnlyWhereTheProviderIsMovedAndItsSoftwareIssues
// marked moved; another module is not.
func TestTheModuleHoldingTheVaultSeatCountsAsMovedForItsConsumers(t *testing.T) {
vault := catalogue.Manifest{Module: "mesh-vault", Version: "1", Claims: []catalogue.Claim{{Name: "mesh-vault", Scope: catalogue.ScopeMesh}},
Provides: []catalogue.Offer{{Name: "secret", Scope: catalogue.ScopeMesh}}}
other := catalogue.Manifest{Module: "mesh-vault", Version: "1", Provides: []catalogue.Offer{{Name: "secret", Scope: catalogue.ScopeMesh}}}
Provides: []catalogue.Offer{{Name: "secret", Scope: catalogue.ScopeMesh,
Credential: &catalogue.OfferCredential{Form: catalogue.CredentialFormPair}}}}
other := catalogue.Manifest{Module: "mesh-vault", Version: "1", Provides: []catalogue.Offer{{Name: "secret", Scope: catalogue.ScopeMesh,
Credential: &catalogue.OfferCredential{Form: catalogue.CredentialFormPair}}}}
if got := needOf(t, map[string]catalogue.Manifest{"mesh-vault": vault}, "mesh-vault", "gitea", "secret"); !got.IssuedToHolder || !got.ProviderMoved {
t.Errorf("the vault's consumer is not answered by a moved provider: %+v", got)
}
+15 -2
View File
@@ -89,15 +89,28 @@ func TestTwoControllersOneActs(t *testing.T) {
t.Fatalf("B, waiting, may act: %v", err)
}
// A hands over, as a controller being replaced does: B takes the lease at once, at a higher epoch.
// A is stopped, as its service manager stops a controller being replaced (novox/hq issue 906): the stop
// alone gives the lease back, before anything else of A is taken down, and B takes it within a second, at
// a higher epoch — no wait for the key's age, and no condition said.
stopA()
a.release()
stopped := time.Now()
var lostB <-chan struct{}
select {
case lostB = <-tookB:
case <-time.After(10 * time.Second):
t.Fatal("B did not take the lease A gave back")
}
if waited := time.Since(stopped); waited > time.Second {
t.Fatalf("B took the lease %s after A was stopped: the stop did not give it back first", waited)
}
a.release() // what serve's own end does after: waits for the release under way, gives nothing twice
conditionsOfB := conditions.NewInMemory()
keeperOfB := conditions.NewKeeper(ctx, conditions.Options{Store: conditionsOfB, History: conditionsOfB})
b.watchTakeovers(ctx, inv, keeperOfB)
keeperOfB.Close(ctx)
if said, err := conditionsOfB.Since(ctx, stopped.Add(-time.Minute)); err != nil || len(said) != 0 {
t.Fatalf("a lease given back is said as a condition: %+v, %v", said, err)
}
select {
case <-lostA:
default:
+54 -1
View File
@@ -21,8 +21,13 @@ func holdingOne(names ...string) func(broker.Consumer) (uint64, error) {
}
func deafObservations(now time.Time, held func(broker.Consumer) (uint64, error)) (string, int) {
return deafHeld(now, time.Time{}, held)
}
// deafHeld is deafObservations with the loop's last hold due at due.
func deafHeld(now, due time.Time, held func(broker.Consumer) (uint64, error)) (string, int) {
f := calm(now)
loop, err := readLoop(now.Add(-loopDeafAfter-time.Minute), loopConsumers(), held)
loop, err := readLoop(now.Add(-loopDeafAfter-time.Minute), due, loopConsumers(), held)
if err != nil {
panic(err)
}
@@ -65,3 +70,51 @@ func TestDeafLookReadsExactlyTheControllersOwnConsumers(t *testing.T) {
}
}
}
// Sweep 2026-10-30 (issue 774): a message the loop held back with a delay longer than the deaf bound is not a deaf
// controller until it is due and still not taken; the merges consumer's five-minute holds flapped the alarm this way.
func TestAMessageHeldBackIsNotDeafUntilItIsDue(t *testing.T) {
now := time.Now()
held := holdingOne("CONTROL/" + broker.ControllerName)
for name, c := range map[string]struct {
due time.Time
deaf int
}{
"nothing held": {time.Time{}, 1},
"held, due in three minutes": {now.Add(3 * time.Minute), 0},
"held, due a moment ago": {now.Add(-time.Second), 0},
"held, due past the deaf bound, unsent": {now.Add(-loopDeafAfter - time.Second), 1},
} {
if _, n := deafHeld(now, c.due, held); n != c.deaf {
t.Errorf("%s: %d deaf observation(s), want %d", name, n, c.deaf)
}
}
}
// Sweep 2026-10-30 (issue 774): the controller holds no consumer of the forge's merges any more, and removes the one
// it held at every assertion.
func TestTheMergesConsumerIsRetired(t *testing.T) {
for _, c := range broker.MeshConsumers() {
if c.Name == broker.MergesConsumerName {
t.Fatalf("the controller still holds %s on %s", c.Name, c.Stream)
}
}
r := &removed{}
if err := broker.AssertMeshConsumers(r); err != nil {
t.Fatal(err)
}
if !r.gone[broker.EventsStream+"/"+broker.MergesConsumerName] {
t.Fatalf("the retired merges consumer is not removed: removed %v", r.gone)
}
}
type removed struct{ gone map[string]bool }
func (*removed) EnsureConsumer(broker.Consumer) error { return nil }
func (r *removed) RemoveConsumer(stream, name string) error {
if r.gone == nil {
r.gone = map[string]bool{}
}
r.gone[stream+"/"+name] = true
return nil
}
-186
View File
@@ -1,186 +0,0 @@
package main
import (
"context"
"encoding/json"
"errors"
"fmt"
"reflect"
"slices"
"sort"
"strings"
"time"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"github.com/nats-io/nats.go"
"github.com/novox/mesh-controller/internal/broker"
"github.com/novox/mesh-controller/internal/link"
)
// **A node's composition issues its modules' memberships** (novox/hq ADR 0160, ADR 0201; issue 249 met again under
// issue 1945).
//
// A membership is what an assignment serves and reaches: its subjects, its seats, and the state, logs, streams and waits
// its manifest declares, published last-per-subject on the AssignmentsStream, where the tool runner and every bus process
// read and follow their own. A push issued them before it sent a declaration; release 6 removed the push and nothing
// issued them after, so a module taken in with a new state, read or stream was granted it on the bus and refused it by its
// runtime, which keeps a module to the membership last issued (2026-10-25: mesh-delivery-gate's `conditions`, on novox
// from 20:33). Composing a node is now where its memberships are issued: from the same records the bus's grants are
// composed from (so what a runtime serves and what its account may are one composition), with the private network's
// addresses and what each module receives from the node's composition, before its compositions are written (the code
// arriving first was refused its own state, issue 249).
//
// Only a membership that differs from the one last issued is published, and only over that one, so a pass that changes
// nothing re-serves nothing and a newer one issued meanwhile is not overwritten (issues 930, 934).
// membershipPass is what one composer pass issues every node's memberships from, read once for the pass.
type membershipPass struct {
records broker.Records
where broker.Placements
bus link.OverNATS
}
// membershipsFor reads the records, asserts the memberships' stream and every declared bucket and log (a membership
// names them, and a module whose state does not exist fails its first read), and judges which machines are root-free
// (ADR 0259 §8: a channel's verified sender is composed only from one) when a seat promises one, as a push did.
func membershipsFor(ctx context.Context, open *stores, bus link.OverNATS) (*membershipPass, error) {
records, err := open.inventory.BusRecords(ctx)
if err != nil {
return nil, fmt.Errorf("the memberships are not issued: %w", err)
}
raiser := broker.OnConn(bus.Conn)
for _, s := range broker.MeshStreams() {
if s.Name != broker.AssignmentsStream {
continue
}
if err := raiser.EnsureStream(s); err != nil {
return nil, fmt.Errorf("the memberships are not issued: their stream: %w", err)
}
}
buckets, err := open.inventory.DeclaredBuckets(ctx)
if err != nil {
return nil, fmt.Errorf("the memberships are not issued: the modules' state: %w", err)
}
if _, err := broker.RaiseBuckets(raiser, buckets); err != nil {
return nil, fmt.Errorf("the memberships are not issued: the modules' state could not be asserted: %w", err)
}
logs, err := open.inventory.DeclaredLogs(ctx)
if err != nil {
return nil, fmt.Errorf("the memberships are not issued: the modules' logs: %w", err)
}
if _, err := broker.RaiseLogs(raiser, logs); err != nil {
return nil, fmt.Errorf("the memberships are not issued: the modules' logs could not be asserted: %w", err)
}
var whys map[string]string
if promisesVerifiedSender(records) {
records.RootFree, whys = rootVerdictsOf(judgeRootFree(ctx, newRootReader(ctx, open.inventory, bus.Conn),
records.Nodes, time.Now()))
}
where := broker.PlacementsOf(records, records.Interchangeable)
for _, line := range droppedVerifiedSenders(records, where, whys) {
fmt.Println(line)
}
return &membershipPass{records: records, where: where, bus: bus}, nil
}
// promisesVerifiedSender says a seat some module holds promises a verified sender: only then is the root-free judgement,
// a discovery on the bus, worth its seconds. Pure.
func promisesVerifiedSender(records broker.Records) bool {
for _, declared := range records.Assigned {
for _, d := range declared {
for _, s := range d.Holds {
if slices.Contains(s.Capabilities, "verified-sender") {
return true
}
}
}
}
return false
}
// issue issues node's memberships on the bus, from its composition's addresses and received contributions.
func (p *membershipPass) issue(ctx context.Context, node string, declared sendable) error {
issued, err := issueComposedMemberships(node, p.records, p.where, declared.Mesh, declared.Received,
func(node, module string) ([]byte, uint64, error) {
m, err := p.bus.JS.GetLastMsg(broker.AssignmentsStream, broker.MembershipSubject(node, module), nats.Context(ctx))
if err != nil {
return nil, 0, err
}
return m.Data, m.Sequence, nil
},
func(node, module string, body []byte, seq uint64) error {
return p.bus.PublishMembershipAfter(ctx, node, module, body, seq)
})
if len(issued) > 0 {
fmt.Printf("memberships: %s: issued %s\n", node, strings.Join(issued, ", "))
}
return err
}
// issueComposedMemberships composes the membership of every module assigned to node and publishes each one that differs
// from the one last issued, over that one (seq 0 when none was: only if none was issued meanwhile). It answers the
// modules it issued; every failure is tried past and joined in the error.
func issueComposedMemberships(node string, records broker.Records, where broker.Placements, mesh []string,
received map[string]map[string][]catalogue.Contribution,
last func(node, module string) ([]byte, uint64, error),
publish func(node, module string, body []byte, seq uint64) error) ([]string, error) {
var issued []string
var errs []error
for _, d := range records.Assigned[node] {
body, err := composedMembership(node, d, where, mesh, received[d.Module])
if err != nil {
errs = append(errs, fmt.Errorf("%s on %s: its membership: %w", d.Module, node, err))
continue
}
was, seq, err := last(node, d.Module)
switch {
case errors.Is(err, nats.ErrMsgNotFound):
was, seq = nil, 0
case err != nil:
errs = append(errs, fmt.Errorf("%s on %s: its membership as last issued could not be read: %w", d.Module, node, err))
continue
}
if sameMembership(was, body) {
continue
}
if err := publish(node, d.Module, body, seq); err != nil {
errs = append(errs, fmt.Errorf("%s on %s: its membership could not be issued: %w", d.Module, node, err))
continue
}
issued = append(issued, d.Module)
}
sort.Strings(issued)
return issued, errors.Join(errs...)
}
// composedMembership is one assignment's membership as a node's composition issues it: MembershipFor, with the private
// network's addresses and what the module receives, by requirement (ADR 0167).
func composedMembership(node string, d broker.Declared, where broker.Placements, mesh []string,
received map[string][]catalogue.Contribution) ([]byte, error) {
m := broker.MembershipFor(node, d, where)
m.Mesh = mesh
for requirement, given := range received {
raw, err := json.Marshal(given)
if err != nil {
return nil, err
}
if m.Receives == nil {
m.Receives = map[string]json.RawMessage{}
}
m.Receives[requirement] = raw
}
return json.Marshal(m)
}
// sameMembership says two memberships say the same, whatever the order of their fields. Pure.
func sameMembership(was, now []byte) bool {
if was == nil {
return false
}
var a, b any
if json.Unmarshal(was, &a) != nil || json.Unmarshal(now, &b) != nil {
return false
}
return reflect.DeepEqual(a, b)
}
+98 -42
View File
@@ -16,6 +16,8 @@ import (
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/kvlimit"
"git.novox.be/novox/mesh-sdk/go/process"
"github.com/novox/mesh-controller/internal/broker"
"github.com/novox/mesh-controller/internal/inventory"
"github.com/novox/mesh-controller/internal/link"
@@ -179,7 +181,12 @@ func (k keepingDeliveryMerges) KeepDeliveryMerge(ctx context.Context, emitter st
if err != nil {
return fmt.Errorf("%w: the bucket of mesh-delivery's merges: %w", link.ErrTryAgain, err)
}
if _, err := kv.Put(ctx, deliveryMergeKey(d.Owner, d.Repo, d.Number, d.HeadSHA), body); err != nil {
// Through kvlimit (sweep 2026-10-30, issue 774): a word too large for the bucket is refused before it is put, and
// is an answer, never "not now" — retried, it held the forge's merges for a day (10054 on de-spiegel's merge).
if _, err := kvlimit.Put(ctx, kv, deliveryMergeKey(d.Owner, d.Repo, d.Number, d.HeadSHA), body); err != nil {
if errors.Is(err, kvlimit.ErrTooLarge) {
return err
}
return fmt.Errorf("%w: %w", link.ErrTryAgain, err)
}
if k.kept != nil {
@@ -343,64 +350,113 @@ func claimantsOf(entries []inventory.Entry, seat string) []string {
return out
}
// forgeMergeKey is where the forge's announcement of a merge is kept, by the merge commit: the one thing a build of a merge
// names, from which the pull request and its head are found (novox/hq ADR 0420 §7).
func forgeMergeKey(owner, repo, commit string) string {
return fmt.Sprintf("merge/%s/%s/%s", owner, repo, commit)
}
// keepForgeMerge keeps the forge's announcement of a merge in the bucket of the merges, a week at most as that bucket
// keeps them. The controller plans nothing from a merge any more (ADR 0539 §17); this is all it keeps of one.
var keepForgeMerge = func(ctx context.Context, m link.SourceMoved) error {
serving := servingBus.Load()
if serving == nil {
return fmt.Errorf("%w: the bus is not held", link.ErrTryAgain)
}
api, err := jetstream.New(serving.Conn())
if err != nil {
return fmt.Errorf("%w: %w", link.ErrTryAgain, err)
}
kv, err := api.KeyValue(ctx, broker.DeliveryMergesBucket)
if err != nil {
return fmt.Errorf("%w: the bucket of the merges: %w", link.ErrTryAgain, err)
}
body, err := json.Marshal(keptMerge{SourceMoved: m})
if err != nil {
return err
}
if _, err := kv.Put(ctx, forgeMergeKey(m.Owner, m.Repo, m.Commit), body); err != nil {
return fmt.Errorf("%w: %w", link.ErrTryAgain, err)
}
return nil
}
// keptForgeMerge is the forge's announcement of a merge, by its commit; false when none was kept. A variable so a test can
// hand it one.
// keptForgeMerge is the forge's word on the merge of a repository at a commit: the pull request and the head it merged,
// read from the forge bridge's own stream, FORGE, where mesh-delivery-forge-bridge says every merge as
// `forge.<owner>.<name>.merged` and keeps it a week (novox/hq ADR 0437). False when the stream holds no merge at that
// commit. A variable so a test can hand it one.
//
// **The controller keeps no copy of the forge's merges any more** (sweep 2026-10-30, issue 774). It once kept each in
// its bucket of merges off a consumer of its own on EVENTS that never gave up a message; one announcement over the
// bucket's value size held that consumer for a day and queued every merge behind it, and the deaf watchdog read the
// queue as a deaf controller. The forge bridge's stream is the record of the merges (research 072, 03-domains: the
// delivery parts own what a merge is); the controller reads it where it needs one, at a moved module's registration.
var keptForgeMerge = func(ctx context.Context, owner, repo, commit string) (keptMerge, bool, error) {
serving := servingBus.Load()
if serving == nil {
return keptMerge{}, false, errors.New("only the serving controller keeps what the forge announced")
return keptMerge{}, false, errors.New("only the serving controller reads the forge's merges")
}
api, err := jetstream.New(serving.Conn())
if err != nil {
return keptMerge{}, false, err
}
kv, err := api.KeyValue(ctx, broker.DeliveryMergesBucket)
stream, err := api.Stream(ctx, forgeStream)
if err != nil {
return keptMerge{}, false, fmt.Errorf("the forge bridge's stream %s cannot be read: %w", forgeStream, err)
}
return forgeMergeAt(ctx, streamMerges{stream}, owner, repo, commit)
}
// forgeStream is the forge bridge's stream as the bus names it (its declared name, forge, in capitals).
var forgeStream = catalogue.StreamBusName(sdkdelivery.ForgeStream)
// forgeMergeSubject is the subject of the forge bridge's merged message for a repository.
func forgeMergeSubject(owner, repo string) (string, error) {
event, err := sdkdelivery.ForgeEvent(owner+"/"+repo, sdkdelivery.ForgeMerged)
if err != nil {
return "", err
}
return process.ModuleStreamSubject(sdkdelivery.ForgeBridgeModule, event), nil
}
// mergeMessages is what forgeMergeAt reads of a stream: the last message on a subject, and the first at or after a
// sequence. Each answers jetstream.ErrMsgNotFound when there is none.
type mergeMessages interface {
last(ctx context.Context, subject string) (uint64, []byte, error)
next(ctx context.Context, subject string, from uint64) (uint64, []byte, error)
}
type streamMerges struct{ s jetstream.Stream }
func (m streamMerges) last(ctx context.Context, subject string) (uint64, []byte, error) {
msg, err := m.s.GetLastMsgForSubject(ctx, subject)
if err != nil {
return 0, nil, err
}
return msg.Sequence, msg.Data, nil
}
func (m streamMerges) next(ctx context.Context, subject string, from uint64) (uint64, []byte, error) {
msg, err := m.s.GetMsg(ctx, from, jetstream.WithGetMsgSubject(subject))
if err != nil {
return 0, nil, err
}
return msg.Sequence, msg.Data, nil
}
// forgeMergeAt finds the merged message of owner/repo at commit: the newest first, which is the merge a take-in
// follows nearly always, then every one the stream keeps, oldest first. A message that does not read as a forge
// message is passed over.
func forgeMergeAt(ctx context.Context, msgs mergeMessages, owner, repo, commit string) (keptMerge, bool, error) {
subject, err := forgeMergeSubject(owner, repo)
if err != nil {
return keptMerge{}, false, err
}
e, err := kv.Get(ctx, forgeMergeKey(owner, repo, commit))
if errors.Is(err, jetstream.ErrKeyNotFound) {
// Read for the four fields a registration needs, not checked whole (sdkdelivery.DecodeForge): a merged message a
// newer bridge writes, or one the SDK's check refuses for a part the controller never reads, still names its pull
// request, head and commit.
match := func(data []byte) (keptMerge, bool) {
var f sdkdelivery.Forge
if json.Unmarshal(data, &f) != nil || f.Kind != sdkdelivery.ForgeMerged || f.Number <= 0 || f.Head == "" ||
!sameCommitID(f.Merge, commit) {
return keptMerge{}, false
}
return keptMerge{SourceMoved: link.SourceMoved{Owner: owner, Repo: repo, Base: f.Branch, Number: f.Number,
HeadSHA: f.Head, Commit: f.Merge}}, true
}
lastSeq, data, err := msgs.last(ctx, subject)
if errors.Is(err, jetstream.ErrMsgNotFound) {
return keptMerge{}, false, nil
}
if err != nil {
return keptMerge{}, false, err
}
var k keptMerge
if err := json.Unmarshal(e.Value(), &k); err != nil {
return keptMerge{}, false, err
if k, ok := match(data); ok {
return k, true, nil
}
return k, true, nil
for from := uint64(1); from < lastSeq; {
seq, data, err := msgs.next(ctx, subject, from)
if errors.Is(err, jetstream.ErrMsgNotFound) || (err == nil && seq >= lastSeq) {
return keptMerge{}, false, nil
}
if err != nil {
return keptMerge{}, false, err
}
if k, ok := match(data); ok {
return k, true, nil
}
from = seq + 1
}
return keptMerge{}, false, nil
}
// keptMerge is a merge as the record keeps it: the forge's announcement.
+2 -10
View File
@@ -99,17 +99,9 @@ func TestAModuleBuiltFromANewRepositoryIsRegisteredOnlyOnTheOperatorsApproveOnTh
return err == nil && kept
})
// The forge's merge of that pull request, as the controller heard it, and the build it asked for.
// The forge bridge's merge of that pull request on its stream, and the build it asked for.
const commit = "c0ffee0123456789"
event, _ := json.Marshal(map[string]any{"owner": "novox", "repo": "mesh-host", "base": "main", "number": 7,
"head_sha": a.merge.HeadSHA, "merge_commit_sha": commit})
var forgeMerge link.SourceMoved
if err := json.Unmarshal(event, &forgeMerge); err != nil {
t.Fatal(err)
}
if err := keepForgeMerge(ctx, forgeMerge); err != nil {
t.Fatal(err)
}
sayForgeMerge(ctx, t, js, "novox/mesh-host", 7, a.merge.HeadSHA, commit)
id := fmt.Sprintf("build-%d", time.Now().Add(time.Hour).UnixNano())
if err := inv.RecordBuildRequest(ctx, inventory.BuildRequest{ID: id, Repository: "novox/mesh-host", Seat: "git",
Path: "modules/sudo", Ref: "main", Commit: commit, For: "merge"}); err != nil {
+2 -10
View File
@@ -100,17 +100,9 @@ func TestANewModuleFromANewRepositoryIsRegisteredOnlyOnTheOperatorsApproveOnTheB
return err == nil && kept
})
// The forge's merge of that pull request, as the controller heard it, and the build it asked for.
// The forge bridge's merge of that pull request on its stream, and the build it asked for.
const commit = "c0ffee0123456789"
event, _ := json.Marshal(map[string]any{"owner": "novox", "repo": "mesh-delivery", "base": "main", "number": 7,
"head_sha": a.merge.HeadSHA, "merge_commit_sha": commit})
var forgeMerge link.SourceMoved
if err := json.Unmarshal(event, &forgeMerge); err != nil {
t.Fatal(err)
}
if err := keepForgeMerge(ctx, forgeMerge); err != nil {
t.Fatal(err)
}
sayForgeMerge(ctx, t, js, "novox/mesh-delivery", 7, a.merge.HeadSHA, commit)
id := fmt.Sprintf("build-%d", time.Now().Add(time.Hour).UnixNano())
if err := inv.RecordBuildRequest(ctx, inventory.BuildRequest{ID: id, Repository: "novox/mesh-delivery", Seat: "git",
Path: "modules/mesh-delivery-view", Ref: "main", Commit: commit, For: "merge"}); err != nil {
@@ -11,7 +11,6 @@ import (
"git.novox.be/novox/mesh-sdk/go/catalogue"
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/novox/mesh-controller/internal/inventory"
"github.com/novox/mesh-controller/internal/link"
)
// ADR 0501 §6 (issue 1851, A5): the live case of 2026-10-07 (novox/hq issue 300), on keys. A merge of the catalogue adds
@@ -33,12 +32,6 @@ func TestAMergeAddingAModuleEndsRegisteredFromItsKeyAndAsksNoBuild(t *testing.T)
watchNewModules(watching, open)
const commit = "3da80a4b00000000"
m := link.SourceMoved{Owner: "novox", Repo: "mesh-catalog", Base: "main", Commit: commit,
Paths: []string{"modules/systemd-resolved/module.json", "modules/systemd-resolved/cmd/main.go"},
ModuleDirs: []string{"modules/systemd-resolved"}, ModuleDirsSaid: true}
if err := (following{open: open}).SourceMoved(ctx, m); err != nil {
t.Fatal(err)
}
// The indexer writes the module's trunk value; the builder builds it, then its key is done.
fp := fpOf('7')
manifest := b.indexed(t, "systemd-resolved", "1", "novox/mesh-catalog", "modules/systemd-resolved", commit, fp)
+6 -1
View File
@@ -133,9 +133,14 @@ func nodeCommand(ctx context.Context, args []string) error {
// refuse (novox/hq issue 358). Read-only on the node and here, so a verb may run it too.
return nodeJudgeDirectories(ctx, open, args[1:])
case "placement-export":
// Each node's placement as the catalogue's placement file says it (novox/hq ADR 0555 §6): read-only, so a verb
// may run it.
return placementExport(ctx, open, args[1:])
default:
return fmt.Errorf("node has no %q; it has add, list, show, public-domain, account, agent-account, hand-over, "+
"setuid-search and judge-directories", args[0])
"setuid-search, judge-directories and placement-export", args[0])
}
}
+50
View File
@@ -0,0 +1,50 @@
package main
import (
"context"
"errors"
"sort"
"git.novox.be/novox/mesh-sdk/go/compose/frombus"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// placementExport prints each node's placement as the catalogue's placement file says it (novox/hq ADR 0555 §6, issue
// 1126): what a person decided of the node and where its modules run, read by this controller's own code from its own
// store, as JSON keyed by node. The one verb the controller gains for its exit (the operator, 2026-10-26 19:06:40, ADR
// 0547 §1's one exception besides §5's grant): read once by the seed of the placement files, whose pull request merges
// only when its composition check composes every node equal to what it runs. It reads the same Input every composition
// of this controller reads (readMeshAndNodes, readSettings) and writes nothing.
//
// It moves a piece of the research "The controller, NATS-native": the controller's last node data (node rows, placement,
// ports, pins, bindings, config placement) leaves it for the catalogue, before its exit at ADR 0547's third milestone.
func placementExport(ctx context.Context, open *stores, args []string) error {
if len(args) > 1 {
return errors.New("node placement-export [<name>]")
}
in, _, err := readMeshAndNodes(ctx, open)
if err != nil {
return err
}
if _, err := readSettings(ctx, open, &in); err != nil {
return err
}
var names []string
if len(args) == 1 {
names = []string{args[0]}
} else {
for _, n := range in.Nodes {
names = append(names, n.Name)
}
sort.Strings(names)
}
out := make(map[string]delivery.Placement, len(names))
for _, name := range names {
p, err := frombus.Export(in, name)
if err != nil {
return err
}
out[name] = p
}
return printJSON(out)
}
@@ -0,0 +1,71 @@
//go:build integration
package main
import (
"encoding/json"
"strings"
"testing"
"git.novox.be/novox/mesh-sdk/go/compose/frombus"
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/overlay"
)
// Issue 1126 (novox/hq ADR 0555 §6, the operator's one export verb of 2026-10-26 19:06:40): `node placement-export`
// answers, from the real store, each node's placement as the catalogue's placement file says it: what a person decided
// of the node (its endpoint, site, hub and address on the private network) and its modules in the order the composition
// reads them, exactly frombus.Export of the input this controller composes from; one node when named; a node the store
// lacks is refused. Checked within TestTheControlPlaneIsToldWhereTheNodePutTheStoreAndTheBroker, over the same fresh
// mesh, so the export adds no integration test of its own (the test budget).
func placementExportIsEachNodesPlacementFromTheStore(t *testing.T, open *stores) {
t.Helper()
ctx := t.Context()
in, _, err := readMeshAndNodes(ctx, open)
if err != nil {
t.Fatal(err)
}
if _, err := readSettings(ctx, open, &in); err != nil {
t.Fatal(err)
}
var all map[string]sdkdelivery.Placement
if err := json.Unmarshal([]byte(stdoutOf(t, func() error { return placementExport(ctx, open, nil) })), &all); err != nil {
t.Fatal(err)
}
if len(all) != 2 {
t.Fatalf("exported %d nodes, want anchor and laptop", len(all))
}
for i, name := range []string{"anchor", "laptop"} {
got := all[name]
d := got.Decided
if d.Endpoint != name+".example:51820" || d.Site != "here" || d.Hub != (name == "anchor") || d.Address != []string{"10.77.0.1", "10.77.0.2"}[i] {
t.Errorf("%s's decided facts are %+v", name, d)
}
if a := got.Assigned(); len(a) == 0 || !strings.Contains(strings.Join(a, ","), overlay.Name) {
t.Errorf("%s's modules are %v, want the private network among them", name, a)
}
// What it prints is the file the seed writes: at the placement file's version, read back by its reader.
if raw, _ := json.Marshal(got); got.V != sdkdelivery.Version {
t.Errorf("%s is exported at version %d: %s", name, got.V, raw)
} else if _, err := sdkdelivery.DecodePlacement(raw); err != nil {
t.Errorf("%s's export does not read as a placement file: %v", name, err)
}
want, err := frombus.Export(in, name)
if err != nil {
t.Fatal(err)
}
if !sdkdelivery.SamePlacement(got, want) {
t.Errorf("%s's export differs from frombus.Export of the controller's own input:\n%+v\n%+v", name, got, want)
}
}
var one map[string]sdkdelivery.Placement
if err := json.Unmarshal([]byte(stdoutOf(t, func() error { return placementExport(ctx, open, []string{"laptop"}) })), &one); err != nil {
t.Fatal(err)
}
if _, ok := one["laptop"]; !ok || len(one) != 1 {
t.Errorf("one node named exported %v", one)
}
if err := placementExport(ctx, open, []string{"stranger"}); err == nil {
t.Error("a node the store lacks was exported")
}
}
+42
View File
@@ -7,6 +7,7 @@ import (
"time"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"github.com/novox/mesh-controller/internal/conditions"
"github.com/novox/mesh-controller/internal/inventory"
)
@@ -72,3 +73,44 @@ func TestAFailedPlacingRaisesApplyFailing(t *testing.T) {
t.Fatalf("a failed placing is not raised as anchor's apply-failing: %+v", got)
}
}
// Sweep 2026-10-30 (issue 774): a failure said again in new words carries a new time, and the condition it raised stays
// raised while it stands; only a machine with no open condition waits out the bound, and a cleared failure clears it.
func TestApplyFailingStaysRaisedWhileAnyFailureStands(t *testing.T) {
now := time.Date(2026, 10, 30, 16, 57, 0, 0, time.UTC)
since := now.Add(-time.Minute)
for name, c := range map[string]struct {
open bool
prov string
doing bool
raised bool
since time.Duration // the client's failing-since, before now; zero when it says none
}{
"a fresh placing failure, nothing open": {false, inventory.ProvFailed, false, false, 0},
"a fresh placing failure, condition open": {true, inventory.ProvFailed, false, true, 0},
"placing applied again, condition open": {true, inventory.ProvApplied, false, false, 0},
"a fresh failed apply, condition open": {true, "", true, true, 0},
"a fresh failed apply, nothing open": {false, "", true, false, 0},
"new words of a failure standing ten minutes, nothing open": {false, inventory.ProvFailed, false, true, 10 * time.Minute},
} {
f := calm(now)
if c.open {
f.applyFailingOpen = applyFailingOpen([]conditions.Condition{{Kind: kindApplyFailing,
Subject: conditions.Subject{Scope: conditions.ScopeMachine, ID: "anchor", Machine: "anchor"}}})
}
if c.prov != "" {
f.provFailed = []inventory.ProvReport{{Node: "anchor", Module: "postgres", Status: c.prov,
Reason: "the vault refused it", At: now.Add(-time.Minute)}}
if c.since > 0 {
f.provFailed[0].FailingSince = now.Add(-c.since)
}
}
if c.doing {
f.doing = []inventory.Doing{{Node: "anchor", Outcome: inventory.OutcomeRefused, Refused: "no", At: since,
Since: &since}}
}
if got := watchApplyFailing(f); (len(got) == 1) != c.raised || len(got) > 1 {
t.Errorf("%s: %d observation(s), raised want %v", name, len(got), c.raised)
}
}
}
+3
View File
@@ -180,6 +180,9 @@ func serve(ctx context.Context) (err error) {
if keeper != nil {
defer closeConditions(keeper)
defer stopRunning() // on an error return, the goroutines end before the keeper closes
// The takeover of the lease said as it happened, raised and cleared (novox/hq issue 906).
theLease.watchTakeovers(ctx, inv, keeper)
defer theLease.watchTakeovers(ctx, nil, nil) // the keeper closes after this: nothing said into it then
}
// The bus's own objects, asserted on every start. **Not created once at genesis**: a stream
@@ -108,83 +108,14 @@ func TestADroppedVerifiedSenderIsSaidWithItsReason(t *testing.T) {
}
}
// **A node's composition issues every membership that changed** (issue 249 met again under issue 1945): release 6 removed
// the push that issued memberships, and mesh-delivery-gate taken in with a new state `conditions` was granted its bucket and
// refused it by the tool runner, which reads the membership last issued (2026-10-25, novox). The self-check's reissue
// leaves one never issued alone: the composition issues it.
func TestAComposedNodeIssuesEveryMembershipThatChanged(t *testing.T) {
gate := func(states ...string) broker.Declared {
d := broker.Declared{Module: "mesh-delivery-gate"}
for _, s := range states {
d.State = append(d.State, broker.Bucket{Module: "mesh-delivery-gate", Name: s})
}
return d
}
records := func(d broker.Declared) broker.Records {
return broker.Records{Nodes: []string{"novox"}, Assigned: map[string][]broker.Declared{"novox": {d}}}
}
issuedAs := func(d broker.Declared, mesh []string) []byte {
body, err := composedMembership("novox", d, broker.PlacementsOf(records(d), nil), mesh, nil)
if err != nil {
t.Fatal(err)
}
return body
}
newer := []byte(`{"node":"novox","module":"mesh-delivery-gate","newer":true}`)
for _, c := range []struct {
name string
before []byte // the membership last issued, nil for none
between bool // another issued meanwhile, after the read
issued []string
states []string // what the bus holds after, by state name; nil when it must hold newer
err string
}{
{name: "never issued is issued", issued: []string{"mesh-delivery-gate"}, states: []string{"asks", "conditions"}},
{name: "a new state is issued", before: issuedAs(gate("asks"), []string{"10.0.0.1"}),
issued: []string{"mesh-delivery-gate"}, states: []string{"asks", "conditions"}},
{name: "unchanged is not issued again", before: issuedAs(gate("asks", "conditions"), []string{"10.0.0.1"}),
states: []string{"asks", "conditions"}},
{name: "a newer one meanwhile is not overwritten", before: issuedAs(gate("asks"), []string{"10.0.0.1"}),
between: true, err: "wrong last sequence"},
} {
t.Run(c.name, func(t *testing.T) {
r := records(gate("asks", "conditions"))
where := broker.PlacementsOf(r, nil)
bus := &fakeBus{last: map[string][]byte{}, published: map[string][]byte{}, seq: map[string]uint64{}}
if c.before != nil {
bus.last["novox/mesh-delivery-gate"], bus.seq["novox/mesh-delivery-gate"] = c.before, 1
}
if c.between {
bus.between = func() { bus.last["novox/mesh-delivery-gate"], bus.seq["novox/mesh-delivery-gate"] = newer, 2 }
}
if n := reissueVerifiedSenders(r, where, bus.get, bus.put, func(string, ...any) {}); n != 0 {
t.Fatalf("the self-check's reissue issued %d", n)
}
issued, err := issueComposedMemberships("novox", r, where, []string{"10.0.0.1"}, nil, bus.get, bus.put)
if c.err == "" && err != nil || c.err != "" && (err == nil || !strings.Contains(err.Error(), c.err)) {
t.Fatalf("error %v, want %q", err, c.err)
}
if strings.Join(issued, ",") != strings.Join(c.issued, ",") {
t.Fatalf("issued %v, want %v", issued, c.issued)
}
if c.states == nil {
if string(bus.last["novox/mesh-delivery-gate"]) != string(newer) {
t.Fatalf("the newer membership was overwritten: %s", bus.last["novox/mesh-delivery-gate"])
}
return
}
var m broker.Membership
if err := json.Unmarshal(bus.last["novox/mesh-delivery-gate"], &m); err != nil {
t.Fatal(err)
}
var names []string
for _, s := range m.State {
names = append(names, s.Name)
}
if strings.Join(names, ",") != strings.Join(c.states, ",") || strings.Join(m.Mesh, ",") != "10.0.0.1" {
t.Fatalf("the bus holds states %v and mesh %v, want %v", names, m.Mesh, c.states)
}
})
func TestAMembershipNeverIssuedIsLeftToItsOwnSend(t *testing.T) {
records, _ := verifiedSenderRecords(map[string]bool{"novox": true})
bus := &fakeBus{last: map[string][]byte{}, published: map[string][]byte{}, seq: map[string]uint64{}}
var said []string
n := reissueVerifiedSenders(records, broker.PlacementsOf(records, nil), bus.get, bus.put,
func(f string, a ...any) { said = append(said, f) })
if n != 0 || len(said) != 0 {
t.Fatalf("issued %d, said %v", n, said)
}
}
+54
View File
@@ -0,0 +1,54 @@
package main
import (
"context"
"fmt"
"regexp"
"github.com/novox/mesh-controller/internal/conditions"
)
// **A condition whose check no longer exists is cleared** (sweep 2026-10-30, issue 774). A signals row or a self-check
// probe raises and clears its conditions by reconciling what it sees; one removed from the controller reconciles
// nothing again, so what it raised stayed open for good and read as current: S23's failed plans (the plans went with
// ADR 0539) and D14's stalled deliveries, from 2026-10-20 on. Each watchdogs tick clears every open condition whose
// source is a row's or a probe's name that neither the signals table nor the probe registry holds any more, saying so.
// A source of another form (an event, a module's health, a verb) is never judged here: its own word clears it.
// checkSource is the form of a row's (S1) or a probe's (D4, DW, D-root) name.
var checkSource = regexp.MustCompile(`^(S[0-9]+|D[0-9]+|D[A-Z]+|D-[a-z]+)$`)
// liveChecks are the sources this controller still reconciles: its signals rows and its self-check's probes.
func liveChecks() map[string]bool {
out := map[string]bool{}
for _, r := range signalsTable {
out[r.Row] = true
}
for _, p := range probeRegistry {
out[p.ID] = true
}
return out
}
// retiredCheckConditions are the open conditions raised by a row or a probe the controller no longer has.
func retiredCheckConditions(open []conditions.Condition, live map[string]bool) []conditions.Condition {
var out []conditions.Condition
for _, c := range open {
if checkSource.MatchString(c.Source) && !live[c.Source] {
out = append(out, c)
}
}
return out
}
// clearRetiredChecks clears each of them, saying why.
func clearRetiredChecks(ctx context.Context, k *conditions.Keeper, retired []conditions.Condition) []string {
var problems []string
for _, c := range retired {
why := fmt.Sprintf("%s, which raised it, is no longer a check of the controller", c.Source)
if _, err := k.ClearSaying(ctx, c.Key, why, "No longer checked: the check that raised this was removed"); err != nil {
problems = append(problems, c.Key+": "+err.Error())
}
}
return problems
}
@@ -0,0 +1,31 @@
package main
import (
"testing"
"github.com/novox/mesh-controller/internal/conditions"
)
// Sweep 2026-10-30 (issue 774): an open condition raised by a row or a probe the controller no longer has is cleared;
// one raised by a check it still has, or by a source of another form, is left to its own word.
func TestAConditionWhoseCheckIsGoneIsCleared(t *testing.T) {
live := liveChecks()
for source, gone := range map[string]bool{
"S23": true,
"D14": true,
"S21": false,
"S9": false,
"D-root": false,
"DM": false,
"DB": false,
"D4": false,
"health": false,
"provisioner.failing": false,
"done keys": false,
} {
got := retiredCheckConditions([]conditions.Condition{{Key: "k." + source, Source: source}}, live)
if (len(got) == 1) != gone {
t.Errorf("%s: judged retired %v, want %v", source, len(got) == 1, gone)
}
}
}
+8 -1
View File
@@ -349,6 +349,11 @@ func (a *verbArguments) commandLine() ([]string, error) {
return nil, err
}
return []string{"node", "judge-directories", str("node")}, nil
case "placement-export":
if node := str("node"); node != "" {
return []string{"node", "placement-export", node}, nil
}
return []string{"node", "placement-export"}, nil
case "modules":
return []string{"module", "list", "--json"}, nil
case "seats":
@@ -1400,7 +1405,9 @@ func seatAnnouncement(handlers map[string]link.ToolHandler) micro.Info {
// nodeReads are the `node` subcommands a verb may run: the ones that only read.
var nodeReads = map[string]bool{"list": true, "show": true,
// Read-only on the node too: the engine judges its directories and changes nothing (novox/hq issue 358).
"judge-directories": true}
"judge-directories": true,
// Each node's placement as the catalogue's placement file says it, read from the store (novox/hq ADR 0555 §6).
"placement-export": true}
// flagsOnly says a command line's rest names no subcommand: empty, or beginning with a flag. For a command
// with no subcommands every word is a flag, its value or a name it reads.
+118 -16
View File
@@ -1,6 +1,7 @@
package main
import (
"context"
"fmt"
"sort"
"strings"
@@ -147,9 +148,10 @@ var signalsTable = []signalRow{
return newestOf(f.machines, func(m machineFacts) time.Time { return m.toolsHeard })
}},
{Row: "S12", Signal: "the controller lease renewed", Emitter: "controller", Trigger: "every 5 s",
Bound: "15 s (the key's age); a holder that lost the lease, or stopped renewing and was taken over, and a " +
"lease bucket found raised again from nothing, are said for an hour after; a controller serving without " +
"the lease, for as long as it does",
Bound: "15 s (the key's age); a holder that lost the lease or stopped renewing it is said once by the " +
"controller that took it over, raised and cleared as it takes it, and for an hour after while none " +
"holds it; a lease given back is never said; a lease bucket found raised again from nothing is said " +
"for an hour after; a controller serving without the lease, for as long as it does",
Kind: "lease-lost", Severity: conditions.Urgent, Phase: 2,
needs: func(f *signalFacts) error { return f.leaseErr }, watch: watchLease,
newest: func(f *signalFacts) time.Time { return f.lease.renewed }},
@@ -282,13 +284,17 @@ func watchLoop(f *signalFacts) []conditions.Observation {
if since.IsZero() {
since = f.started
}
// A message the loop held back waits by the loop's own choice until it is due: deafness is counted from then.
if f.loop.due.After(since) {
since = f.loop.due
}
if f.now.Sub(since) <= loopDeafAfter {
return nil
}
return []conditions.Observation{{Scope: conditions.ScopeCore, ID: "controller", Kind: "controller-deaf",
Token: "deaf", Machine: f.host, Severity: conditions.Urgent,
Summary: fmt.Sprintf("the controller's event loop has taken nothing for %s while its consumers hold %d "+
"message(s): reports, builds and merges are not being acted on", ago(f.now.Sub(since)), f.loop.pending),
"message(s): reports and events are not being acted on", ago(f.now.Sub(since)), f.loop.pending),
Said: fmt.Sprintf("%d pending (%s), last taken %s", f.loop.pending, f.loop.where, since.UTC().Format(time.RFC3339))}}
}
@@ -644,10 +650,29 @@ func watchLease(f *signalFacts) []conditions.Observation {
"does not know it lost it",
Said: l.resetSaid})
}
// An epoch below the one this controller holds was taken over by it: said once by the takeover itself,
// raised and cleared (sayTakeover), not observed for an hour after (novox/hq issue 906). One that ended
// while this controller does not hold the lease — serving without it — is still said for the hour.
var ended []inventory.Epoch
for _, e := range l.ended {
if l.held && e.Epoch < l.epoch {
continue
}
ended = append(ended, e)
}
if o, ok := lostObservation(ended, l.epoch, f.now); ok {
out = append(out, o)
}
return out
}
// lostObservation is the lease-lost condition of the epochs that lost the lease or let it expire within the
// hour, named by the newest; false when none did. An epoch given back is no loss.
func lostObservation(ended []inventory.Epoch, acting uint64, now time.Time) (conditions.Observation, bool) {
var lost []string
newest := inventory.Epoch{}
for _, e := range l.ended {
if e.Ended == nil || e.How == inventory.EpochReleased || f.now.Sub(*e.Ended) > advisoryQuiet {
for _, e := range ended {
if e.Ended == nil || e.How == inventory.EpochReleased || now.Sub(*e.Ended) > advisoryQuiet {
continue
}
lost = append(lost, fmt.Sprintf("epoch %d (%s) %s at %s", e.Epoch, e.Instance, e.How,
@@ -656,18 +681,95 @@ func watchLease(f *signalFacts) []conditions.Observation {
newest = e
}
}
if len(lost) > 0 {
how := "lost it: another holder wrote its key, or no renewal was made within the key's age"
if newest.How == inventory.EpochExpired {
how = "stopped renewing it without giving it back, and was taken over"
if len(lost) == 0 {
return conditions.Observation{}, false
}
how := "lost it: another holder wrote its key, or no renewal was made within the key's age"
if newest.How == inventory.EpochExpired {
how = "stopped renewing it without giving it back, and was taken over"
}
return conditions.Observation{Scope: conditions.ScopeCore, ID: "controller.lease", Token: "lost",
Kind: "lease-lost", Machine: newest.Host, Severity: conditions.Urgent, Source: "S12",
Summary: fmt.Sprintf("the controller of epoch %d (%s) %s; the controller of epoch %d acts now", newest.Epoch,
newest.Instance, how, acting),
Said: strings.Join(lost, "; ")}, true
}
// takeover is the condition a takeover says: the epoch just before the one taken, when it lost the lease or
// let it expire; false when it gave it back, or none is recorded.
func takeover(ended []inventory.Epoch, taken uint64, now time.Time) (conditions.Observation, bool) {
var before *inventory.Epoch
for i, e := range ended {
if e.Epoch < taken && (before == nil || e.Epoch > before.Epoch) {
before = &ended[i]
}
out = append(out, conditions.Observation{Scope: conditions.ScopeCore, ID: "controller.lease", Token: "lost",
Kind: "lease-lost", Machine: newest.Host, Severity: conditions.Urgent,
Summary: fmt.Sprintf("the controller of epoch %d (%s) %s; the controller of epoch %d acts now", newest.Epoch,
newest.Instance, how, l.epoch),
Said: strings.Join(lost, "; ")})
}
if before == nil {
return conditions.Observation{}, false
}
return lostObservation([]inventory.Epoch{*before}, taken, now)
}
// sayTakeover is the takeover said as an event (novox/hq issue 906): when the controller before this one lost
// the lease or let it expire, the lease-lost condition is raised and cleared at once by this controller, which
// holds the lease — the operator reads that it happened and that it is over, not an urgent alarm for an hour.
// A lease given back says nothing. Said once the condition store is open; a take before that is said then.
func (a *actor) sayTakeover(ctx context.Context, epoch uint64) {
a.mu.Lock()
keeper, inv := a.keeper, a.inv
a.mu.Unlock()
if keeper == nil || inv == nil || epoch == 0 {
return
}
now := time.Now()
reading, cancel := context.WithTimeout(ctx, 10*time.Second)
defer cancel()
ended, err := inv.EpochsSince(reading, now.Add(-advisoryQuiet))
if err != nil {
// Failure is loud: S12 no longer observes an epoch below the one held, so an unread takeover is said as
// one that was not a handover — raised, and left for S12 to clear (ADR 0558 §3).
fmt.Printf("how the epoch before %d ended cannot be read, so its takeover is said as an expiry: %v\n", epoch, err)
o := conditions.Observation{Scope: conditions.ScopeCore, ID: "controller.lease", Token: "lost",
Kind: "lease-lost", Severity: conditions.Urgent, Source: "S12",
Summary: fmt.Sprintf("the controller of epoch %d took the lease, and how the controller before it ended "+
"cannot be read: it is said as one that stopped renewing without giving it back", epoch),
Said: err.Error()}
if _, err := keeper.Observe(reading, o); err != nil {
fmt.Printf("the takeover at epoch %d could not be said: %v\n", epoch, err)
}
return
}
sayTakeoverOf(reading, keeper, ended, epoch, now)
}
// sayTakeoverOf raises and clears the takeover's condition over the epochs read.
func sayTakeoverOf(ctx context.Context, keeper *conditions.Keeper, ended []inventory.Epoch, epoch uint64, now time.Time) {
o, ok := takeover(ended, epoch, now)
if !ok {
return
}
if _, err := keeper.Observe(ctx, o); err != nil {
fmt.Printf("the takeover at epoch %d could not be said: %v\n", epoch, err)
return
}
if _, err := keeper.Clear(ctx, o.Key(), fmt.Sprintf("taken over: the controller of epoch %d holds the lease", epoch)); err != nil {
fmt.Printf("the takeover at epoch %d was said and could not be cleared: %v\n", epoch, err)
}
}
// watchTakeovers gives the serving controller's condition store to the lease, and says the takeover of the
// epoch it holds, if it took one over before the store opened.
func (a *actor) watchTakeovers(ctx context.Context, inv *inventory.Inventory, keeper *conditions.Keeper) {
a.mu.Lock()
a.keeper, a.inv = keeper, inv
held := a.held
a.mu.Unlock()
if held == nil {
return
}
if epoch, err := held.Epoch(); err == nil {
a.sayTakeover(ctx, epoch)
}
return out
}
// handActsWithin is how far back a repeated cause counts (S15).
+32 -1
View File
@@ -491,13 +491,41 @@ func TestAStaleWriterIsNamedByItsEpoch(t *testing.T) {
}
}
// **The lease lost is said for an hour, and serving without it for as long as it lasts** (S12).
// **The lease lost is said for an hour while nobody took it over, a takeover is said once as it happens —
// raised and cleared by the controller that took it — a lease given back is never said, and serving without
// it for as long as it lasts** (S12, novox/hq issue 906).
func TestALeaseLostOrMissingIsSaid(t *testing.T) {
now := time.Now()
f := calm(now)
expired := now.Add(-59 * time.Minute)
f.lease.ended = []inventory.Epoch{{Epoch: 41, Instance: "controller@anchor pid 7", Host: "anchor",
Ended: &expired, How: inventory.EpochExpired}}
// Held at epoch 57, above it: the takeover said it, and no hour of observation follows.
if got := watchLease(f); len(got) != 0 {
t.Fatalf("an epoch this controller took over is observed after the takeover: %+v", got)
}
store := conditions.NewInMemory()
k := conditions.NewKeeper(t.Context(), conditions.Options{Store: store, History: store, Now: func() time.Time { return now }})
sayTakeoverOf(t.Context(), k, f.lease.ended, 57, now)
if open, err := k.Open(t.Context()); err != nil || len(open) != 0 {
t.Fatalf("the takeover is left open: %+v, %v", open, err)
}
k.Close(t.Context()) // its transitions written to the history
history, err := store.Since(t.Context(), now.Add(-time.Minute))
if err != nil {
t.Fatal(err)
}
var changes []string
for _, e := range history {
if e.Key == "core.controller.lease.lost" {
changes = append(changes, e.Change)
}
}
if !slices.Equal(changes, []string{"raised", "cleared"}) {
t.Fatalf("the takeover of an expired lease is not raised and cleared: %v", changes)
}
// Not holding the lease, the loss is said for the hour.
f.lease.held = false
got := watchLease(f)
if len(got) != 1 || got[0].Key() != "core.controller.lease.lost" || !strings.Contains(got[0].Summary, "epoch 41") ||
got[0].Severity != conditions.Urgent {
@@ -512,6 +540,9 @@ func TestALeaseLostOrMissingIsSaid(t *testing.T) {
if got := watchLease(f); len(got) != 0 {
t.Fatalf("a lease given back is said as lost: %+v", got)
}
if _, said := takeover(f.lease.ended, 57, now); said {
t.Fatal("the takeover of a lease given back is said")
}
f.lease = leaseFacts{held: true, epoch: 501, renewed: now, reset: now.Add(-time.Minute), resetSaid: "moved past 500"}
if got := watchLease(f); len(got) != 1 || got[0].Key() != "core.controller.lease.reset" {
t.Fatalf("a lease bucket raised again from nothing was not said: %+v", got)
+14 -1
View File
@@ -160,7 +160,7 @@ func switchList(ctx context.Context, inv *inventory.Inventory, keys switchReads)
var out []switchLine
for _, e := range entries {
m := e.Manifest.Module
if e.Provided || e.Source.Repository == "" {
if e.Provided || e.Source.Repository == "" || sdkdelivery.ProvidedByMesh(m) {
continue
}
trunk, err := keys.ModuleAt(ctx, m, "")
@@ -202,3 +202,16 @@ func heldArtifacts(ctx context.Context, inv *inventory.Inventory, module, builtF
// empty-on-error: not an error; the mesh holds no build of that commit on record, which the line says
return nil, nil
}
// builtByTheMesh are the modules of a list that a build of the mesh's makes: without the ones the mesh provides itself
// (sdkdelivery.ProvidedByMesh, the private network), which no fingerprint ever names, so a line saying one has none
// says nothing (sweep 2026-10-30, issue 774: "mesh-wireguard ... no fingerprint of the new form" on every composition).
func builtByTheMesh(modules []string) []string {
var out []string
for _, m := range modules {
if !sdkdelivery.ProvidedByMesh(m) {
out = append(out, m)
}
}
return out
}
-9
View File
@@ -93,15 +93,6 @@ func (f following) Announceable(ctx context.Context) ([]link.Announcement, error
return out, nil
}
// SourceMoved is the forge announcing a merge. It is put into the open batch (novox/hq ADR 0276), which
// becomes one walk when its merge window closes: hearMerge, and cutBatchesHeld in batches.go.
func (f following) SourceMoved(ctx context.Context, m link.SourceMoved) error {
// A merge plans nothing here any more: mesh-delivery merges on the Approve, the indexer writes the fingerprints and
// the builder builds them (ADR 0539 §5, §17). The controller keeps the announcement, for the build of a merge that
// moves a module between repositories to be shown the operator's Approve (ADR 0420 §7).
return keepForgeMerge(ctx, m)
}
// actingOnMerges keeps one merge acted on at a time (novox/hq issue 266).
var actingOnMerges sync.Mutex
+14 -4
View File
@@ -58,6 +58,11 @@ type signalFacts struct {
// provFailed is every module whose files a machine's provisioner client last said it failed to place (S21,
// novox/hq ADR 0362 §3): that machine's failed apply.
provFailed []inventory.ProvReport
// applyFailingOpen are the machines whose S21 condition is open: kept raised while any failure stands, whatever
// the newest report's time (sweep 2026-10-30).
applyFailingOpen map[string]bool
// retiredChecks are the open conditions raised by a row or a probe the controller no longer has (sweep 2026-10-30).
retiredChecks []conditions.Condition
loop loopFacts
loopErr error
@@ -147,7 +152,9 @@ type machineFacts struct {
func (m machineFacts) asleep() bool { return m.power.Away() }
type loopFacts struct {
took time.Time
took time.Time
// due is when a message the loop held back is handed over again (link.Loop.Due): the loop is not deaf before it.
due time.Time
pending uint64
where string
}
@@ -252,6 +259,7 @@ func (w *watchdogs) see(running context.Context, f *signalFacts) {
problems = append(problems, "S8: "+err.Error())
}
}
problems = append(problems, clearRetiredChecks(running, w.keeper, f.retiredChecks)...)
if err := w.keeper.EndSilences(running); err != nil {
problems = append(problems, err.Error())
}
@@ -315,6 +323,8 @@ func (w *watchdogs) gather(ctx context.Context) *signalFacts {
} else {
f.standings = providerStandings(open)
f.providerWords = providerConditions(open)
f.applyFailingOpen = applyFailingOpen(open)
f.retiredChecks = retiredCheckConditions(open, liveChecks())
}
f.advisories = link.Advisories.Since(now.Add(-advisoryQuiet))
f.lostConsumers, f.advisoriesErr = w.lostConsumers(ctx, f.advisories)
@@ -458,7 +468,7 @@ func (w *watchdogs) gatherLoop() (loopFacts, error) {
if w.js == nil {
return loopFacts{took: link.Loop.Last()}, errors.New("this controller is not on the bus")
}
return readLoop(link.Loop.Last(), loopConsumers(), func(c broker.Consumer) (uint64, error) {
return readLoop(link.Loop.Last(), link.Loop.Due(), loopConsumers(), func(c broker.Consumer) (uint64, error) {
info, err := w.js.Context().ConsumerInfo(c.Stream, c.Name)
if err != nil {
return 0, err
@@ -468,8 +478,8 @@ func (w *watchdogs) gatherLoop() (loopFacts, error) {
}
// readLoop sums what the given consumers hold, as held reads each.
func readLoop(took time.Time, consumers []broker.Consumer, held func(broker.Consumer) (uint64, error)) (loopFacts, error) {
f := loopFacts{took: took}
func readLoop(took, due time.Time, consumers []broker.Consumer, held func(broker.Consumer) (uint64, error)) (loopFacts, error) {
f := loopFacts{took: took, due: due}
var where []string
for _, c := range consumers {
n, err := held(c)
+1 -1
View File
@@ -3,7 +3,7 @@ module github.com/novox/mesh-controller
go 1.26.0
require (
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024234636-10b6e202320f
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030174137-a37181261c24
github.com/jackc/pgx/v5 v5.10.0
github.com/nats-io/nats-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0
+2 -2
View File
@@ -1,7 +1,7 @@
git.novox.be/novox/mesh-host v0.0.0-20261018171808-9d9e3a46c4f5 h1:E5stfIeCN6uYgA9TCv+4K8JgaURg3zQppvUgJLaDMP4=
git.novox.be/novox/mesh-host v0.0.0-20261018171808-9d9e3a46c4f5/go.mod h1:4wyoCr6IIIIarEUuROThWDj7DAvJvIidqMjWcpSnOwE=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024234636-10b6e202320f h1:lUEyNPPF3w4Q0LIgcUUD6ofkKVhCF0n7dVjMoixQmXo=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024234636-10b6e202320f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030174137-a37181261c24 h1:s0XXZtR7CFPBbnaDygWnE4/Q827iEFt0YfEzujk4Q9g=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030174137-a37181261c24/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+11 -2
View File
@@ -53,6 +53,16 @@ func Dir(t testing.TB, repository string) string {
t.Logf("judged against %s as cloned beside this check", dir)
return dir
}
t.Logf("no %s beside this check: a merge check judges %s against its own clone", Env, repository)
return CapturedDir(t, repository)
}
// CapturedDir is the copy of the named repository captured in testdata/beside, laid out as Dir lays it out, whatever
// MESH_CHECK_BESIDE says: for a test that judges this repository against the commit it was released with, which a
// merge check does not clone (it clones the catalogue at its trunk even in a delivery group holding a catalogue pull
// request, ADR 0484 clause 4).
func CapturedDir(t testing.TB, repository string) string {
t.Helper()
from := filepath.Join(Captured(), repository)
if _, err := os.Stat(from); err != nil {
t.Fatalf("no captured copy of %s at %s: %v", repository, from, err)
@@ -81,8 +91,7 @@ func Dir(t testing.TB, repository string) string {
if err != nil {
t.Fatalf("the captured copy of %s could not be laid out: %v", repository, err)
}
t.Logf("judged against the copy of %s captured in testdata/beside (see CAPTURED); a merge check "+
"judges it against the repository's own clone", repository)
t.Logf("judged against the copy of %s captured in testdata/beside (see CAPTURED)", repository)
return dir
}
-17
View File
@@ -41,23 +41,6 @@ func ControllerAnnouncement(node string) announce.States {
return s
}
// ReadsAnnounced splits a reader's `reads` (`<owner>.<name>`) by what each owner announced: granted when the owner
// announced the state readable, or announced nothing yet (the transition); refused when the owner announced and did
// not announce that state readable.
func ReadsAnnounced(reads []string, announced map[string]announce.States) (granted, refused []string) {
for _, r := range reads {
owner, name, ok := strings.Cut(r, ".")
s, said := announced[owner]
switch {
case !ok || !said || s.Readable(name):
granted = append(granted, r)
default:
refused = append(refused, r)
}
}
return granted, refused
}
// NotAnnounced are the owners some module reads the state of and that have announced nothing yet, sorted: read from the
// readers' `reads` as before, until they announce.
func NotAnnounced(r Records) []string {
-17
View File
@@ -7,23 +7,6 @@ import (
"git.novox.be/novox/mesh-sdk/go/announce"
)
// D3 (ADR 0495 clause 3, ADR 0513 §3; issue 1790): a reader's `reads` is granted where the owner announced the state
// readable, refused where it announced and did not, and granted as before where the owner announced nothing yet.
func TestAReadIsGrantedFromItsOwnersAnnouncement(t *testing.T) {
announced := map[string]announce.States{
"owner": {Module: "owner", States: []announce.State{{Name: "open", Readable: true}, {Name: "closed"}}},
}
granted, refused := ReadsAnnounced([]string{"owner.open", "owner.closed", "owner.unsaid", "quiet.any"}, announced)
if !slices.Equal(granted, []string{"owner.open", "quiet.any"}) || !slices.Equal(refused, []string{"owner.closed", "owner.unsaid"}) {
t.Fatalf("granted %v, refused %v", granted, refused)
}
r := Records{Announced: announced, Assigned: map[string][]Declared{"one": {{Module: "reader", Reads: granted,
ReadsRefused: refused}}}}
if got := NotAnnounced(r); !slices.Equal(got, []string{"quiet"}) {
t.Fatalf("not yet announced: %v", got)
}
}
// D1 (ADR 0495 clause 2): the controller announces every bucket of its own, and readable exactly the record mirrors (the
// declared packages the pacman module reads among them, ADR 0507 §5), the chain's buckets, the checks running and the
// plans building.
+9 -7
View File
@@ -46,9 +46,8 @@ func TestAConsumerMadeFromNowNeverReplaysTheStreamsHistory(t *testing.T) {
t.Fatalf("a consumer made from now holds %d of the stream's past (policy %v)", info.NumPending, info.Config.DeliverPolicy)
}
_, _ = js.js.Publish("history.merged", []byte("new"))
time.Sleep(100 * time.Millisecond)
if info, _ = js.js.ConsumerInfo(stream, "fresh"); info.NumPending != 1 {
t.Fatalf("a new announcement is not pending: %d", info.NumPending)
if got := pendingOf(js, stream, "fresh", 1); got != 1 {
t.Fatalf("a new announcement is not pending: %d", got)
}
// Asserted again, it keeps where it is.
if err := js.EnsureConsumer(fresh); err != nil {
@@ -61,8 +60,8 @@ func TestAConsumerMadeFromNowNeverReplaysTheStreamsHistory(t *testing.T) {
if err := js.EnsureConsumer(old); err != nil {
t.Fatal(err)
}
if info, _ = js.js.ConsumerInfo(stream, "old"); info.NumPending != 6 {
t.Fatalf("the default should replay all six: %d", info.NumPending)
if got := pendingOf(js, stream, "old", 6); got != 6 {
t.Fatalf("the default should replay all six: %d", got)
}
before, after, err := js.ResetConsumer(stream, "old")
if err != nil {
@@ -89,10 +88,13 @@ func TestAConsumerMadeFromNowNeverReplaysTheStreamsHistory(t *testing.T) {
}
}
// pendingOf polls a consumer's pending count until it is want or a second has passed, and returns what it saw last.
// pendingOf polls a consumer's pending count until it is want or ten seconds have passed, and returns what it saw
// last. The server counts a consumer's pending after it acknowledges the publish, not before, so a test that reads
// the count at once fails on a loaded build node (a try of main on ace, 2026-10-26): every test that expects a
// count after a publish reads it through here.
func pendingOf(js *JetStream, stream, name string, want uint64) uint64 {
var got uint64
for deadline := time.Now().Add(time.Second); ; time.Sleep(20 * time.Millisecond) {
for deadline := time.Now().Add(10 * time.Second); ; time.Sleep(20 * time.Millisecond) {
if info, err := js.js.ConsumerInfo(stream, name); err == nil {
got = info.NumPending
}
@@ -73,15 +73,15 @@ func TestAResetOfTheEventsConsumerLeavesTheMergesAndTheirConsumerAlone(t *testin
if _, err := js.Context().Publish(ControllerFollows[0], []byte(`{}`)); err != nil {
t.Fatal(err)
}
merges, err := js.Context().ConsumerInfo(EventsStream, MergesConsumerName)
if err != nil || merges.NumPending != 1 {
t.Fatalf("the merges consumer holds %v (%v); want the merge", merges, err)
if got := pendingOf(js, EventsStream, MergesConsumerName, 1); got != 1 {
t.Fatalf("the merges consumer holds %d pending; want the merge", got)
}
if _, _, err := js.ResetConsumer(EventsStream, ControllerName); err != nil {
t.Fatal(err)
}
merges, err = js.Context().ConsumerInfo(EventsStream, MergesConsumerName)
if err != nil || merges.NumPending != 1 || merges.Delivered.Stream != 0 {
pending := pendingOf(js, EventsStream, MergesConsumerName, 1)
merges, err := js.Context().ConsumerInfo(EventsStream, MergesConsumerName)
if err != nil || pending != 1 || merges.Delivered.Stream != 0 {
t.Fatalf("after the reset the merges consumer holds %+v (%v); want its merge left alone", merges, err)
}
}
@@ -107,9 +107,8 @@ func TestAMergeTheOldConsumerHadNotAcknowledgedIsTheMergesConsumers(t *testing.T
if err := AssertMeshConsumers(js); err != nil {
t.Fatal(err)
}
merges, err := js.Context().ConsumerInfo(EventsStream, MergesConsumerName)
if err != nil || merges.NumPending != 1 {
t.Fatalf("the merges consumer holds %v (%v); want the merge the old consumer never acknowledged", merges, err)
if got := pendingOf(js, EventsStream, MergesConsumerName, 1); got != 1 {
t.Fatalf("the merges consumer holds %d pending; want the merge the old consumer never acknowledged", got)
}
}
+106
View File
@@ -0,0 +1,106 @@
package broker
import "testing"
// Defends issue 1790 and novox/hq ADR 0547 §5 (the operator, 2026-10-26 08:54): the grants part may create the stream
// of each bucket, log and stream a placed module's manifest declares, by its own name, and read it back; nothing
// wider, nothing updated or deleted, and no other principal so. Held byte-equal in the controller's copy. And ADR 0547
// clause 7 (issue 1945, the operator's second exception): the grants part writes its own object bucket of sealed user
// lists, and the node-engine of the node holding mesh-broker alone reads its one key and an object of it by name.
func TestTheGrantsPartCreatesOnlyDeclaredStreams(t *testing.T) {
records := Records{
Nodes: []string{"novox", "ace"},
Assigned: map[string][]Declared{"ace": {{Module: "plex"}}, "novox": {
{Module: "nats", SnapshotsTheBus: true},
{Module: "mesh-delivery-builder",
State: []Bucket{{Module: "mesh-delivery-builder", Name: "steps", History: 5}},
Logs: []Log{{Module: "mesh-delivery-builder", Name: "builds"}},
Streams: []ModuleStream{{Module: "mesh-delivery-builder", Name: "BUILDASKS", Events: []string{"ask"}}}},
{Module: GrantsModule, State: []Bucket{{Module: GrantsModule, Name: "assertions"}}},
}},
}
users, err := Users(records)
if err != nil {
t.Fatal(err)
}
perms := map[string]Permissions{}
for _, u := range users {
if u.Kind == KindNode {
p, err := PermissionsFor(u)
if err != nil {
t.Fatalf("%s: %v", u.Username(), err)
}
perms[u.Username()] = p
continue
}
if u.Kind != KindModule {
continue
}
p, err := PermissionsFor(u)
if err != nil {
t.Fatalf("%s: %v", u.Username(), err)
}
perms[u.Module] = p
}
for _, row := range []struct {
who, subject string
may bool
}{
{GrantsModule, "$JS.API.STREAM.CREATE.KV_mesh-delivery-builder_steps", true},
{GrantsModule, "$JS.API.STREAM.INFO.KV_mesh-delivery-builder_steps", true},
{GrantsModule, "$JS.API.STREAM.CREATE.LOG_mesh-delivery-builder_builds", true},
{GrantsModule, "$JS.API.STREAM.CREATE.BUILDASKS", true},
{GrantsModule, "$JS.API.STREAM.CREATE.KV_mesh-grants_assertions", true},
{GrantsModule, "$JS.API.STREAM.UPDATE.KV_mesh-delivery-builder_steps", false},
{GrantsModule, "$JS.API.STREAM.DELETE.KV_mesh-delivery-builder_steps", false},
{GrantsModule, "$JS.API.STREAM.PURGE.KV_mesh-delivery-builder_steps", false},
{GrantsModule, "$JS.API.STREAM.CREATE.KV_mesh-delivery-builder_undeclared", false},
{GrantsModule, "$JS.API.STREAM.CREATE.EVENTS", false},
{GrantsModule, "$JS.API.STREAM.CREATE.ASSIGNMENTS", false},
{GrantsModule, "$JS.API.STREAM.INFO.EVENTS", false},
{"mesh-delivery-builder", "$JS.API.STREAM.CREATE.KV_mesh-delivery-builder_steps", false},
// The sealed user list (ADR 0547 clause 7): the grants part writes its own bucket and no other.
{GrantsModule, "$JS.API.STREAM.CREATE.OBJ_mesh-grants-user-list", true},
{GrantsModule, "$O.mesh-grants-user-list.C.abc", true},
{GrantsModule, "$JS.API.STREAM.PURGE.OBJ_mesh-grants-user-list", true},
{GrantsModule, "$O.declaration-files-novox.C.abc", false},
{GrantsModule, "$JS.API.STREAM.DELETE.OBJ_mesh-grants-user-list", false},
{"mesh-delivery-builder", "$O.mesh-grants-user-list.C.abc", false},
// The bus node's engine reads the one key and an object by name; nothing of the bucket's list, no write.
{"node.novox", "$JS.API.DIRECT.GET.KV_mesh-grants_user-list.$KV.mesh-grants_user-list.sealed", true},
{"node.novox", "$JS.API.CONSUMER.CREATE.KV_mesh-grants_user-list.abc.$KV.mesh-grants_user-list.sealed", true},
{"node.novox", "$JS.API.DIRECT.GET.OBJ_mesh-grants-user-list.$O.mesh-grants-user-list.M.abc", true},
{"node.novox", "$JS.API.CONSUMER.CREATE.OBJ_mesh-grants-user-list.abc.$O.mesh-grants-user-list.C.abc", true},
{"node.novox", "$JS.API.DIRECT.GET.KV_mesh-grants_user-list.$KV.mesh-grants_user-list.list", false},
{"node.novox", "$JS.API.STREAM.INFO.OBJ_mesh-grants-user-list", false},
{"node.novox", "$JS.API.STREAM.MSG.GET.OBJ_mesh-grants-user-list", false},
{"node.novox", "$O.mesh-grants-user-list.C.abc", false},
{"node.ace", "$JS.API.DIRECT.GET.KV_mesh-grants_user-list.$KV.mesh-grants_user-list.sealed", false},
{"node.ace", "$JS.API.DIRECT.GET.OBJ_mesh-grants-user-list.$O.mesh-grants-user-list.M.abc", false},
} {
if got := MayPublish(perms[row.who], row.subject); got != row.may {
t.Errorf("%s may publish %s: %v, want %v", row.who, row.subject, got, row.may)
}
}
// A name that is no plain token is never granted, and a module other than the grants part is never granted one.
if got := CreateStreamGrants([]string{"*", ">", "KV_a.b", "KV_a_b"}); len(got) != 2 {
t.Errorf("grants for wildcard names: %v", got)
}
other, err := PermissionsFor(Principal{Kind: KindModule, Node: "novox", Module: "shop", CreatesStreams: []string{"KV_shop_x"}})
if err != nil {
t.Fatal(err)
}
if MayPublish(other, "$JS.API.STREAM.CREATE.KV_shop_x") {
t.Error("a module other than the grants part was granted a create")
}
// The writers table refuses the grants part an update, and a create of a stream it was not given.
g := Principal{Kind: KindModule, Node: "novox", Module: GrantsModule, CreatesStreams: []string{"KV_a_b"}}
if err := CheckWriters(g, []string{"$JS.API.STREAM.CREATE.KV_a_b"}); err != nil {
t.Errorf("the declared create refused: %v", err)
}
for _, refused := range []string{"$JS.API.STREAM.UPDATE.KV_a_b", "$JS.API.STREAM.CREATE.KV_a_c", "$JS.API.STREAM.CREATE.*"} {
if CheckWriters(g, []string{refused}) == nil {
t.Errorf("%s was not refused", refused)
}
}
}
+39
View File
@@ -0,0 +1,39 @@
package broker
import "testing"
// Sweep 2026-10-30 (issue 774), the guard of mesh-sdk #136 in the controller's copy: where a placement names the
// node-engine's manifest, it adds no user, no process and nothing the tool runner carries; a module beside it does.
func TestTheNodeEnginesManifestComposesNoPlacedUser(t *testing.T) {
engine := Declared{Module: EngineModule, Serves: []string{"health"}, State: []Bucket{{Name: "health"}},
Processes: []ProcessPrincipal{{Process: "apply"}}}
beside := Declared{Module: "power", Serves: []string{"power_draw"}}
for name, c := range map[string]struct {
assigned []Declared
module bool
}{
"with the runner": {[]Declared{{Module: RuntimeModule, NoAccount: true}, engine, beside}, false},
"without the runner": {[]Declared{engine, beside}, true},
} {
users, err := Users(Records{Nodes: []string{"anchor"}, Assigned: map[string][]Declared{"anchor": c.assigned}})
if err != nil {
t.Fatalf("%s: %v", name, err)
}
besideSeen := false
for _, u := range users {
if u.Module == EngineModule && (u.Kind == KindModule || u.Kind == KindProcess) {
t.Errorf("%s: the node-engine's manifest composed a %v user", name, u.Kind)
}
besideSeen = besideSeen || u.Module == beside.Module
for _, carried := range u.Carries {
if carried.Module == EngineModule {
t.Errorf("%s: the tool runner carries the node-engine's manifest", name)
}
besideSeen = besideSeen || carried.Module == beside.Module
}
}
if !besideSeen {
t.Errorf("%s: the module beside it is composed nowhere", name)
}
}
}
+4 -1
View File
@@ -213,8 +213,11 @@ func TestNoPathDeletesALogOrABucket(t *testing.T) {
// Only the writers table names it, as a subject no one but the controller may publish.
// And the operator's release 6 hand path, whose object store purges a replaced object's chunks in a node's
// object bucket (novox/hq ADR 0539 clause 10): a grant, never a call of this repository.
// And the grants part's own object bucket of sealed user lists, whose put purges a replaced object's chunks
// (novox/hq ADR 0547 clause 7): a grant, never a call of this repository.
if !strings.HasSuffix(filepath.ToSlash(path), "internal/broker/writers.go") &&
!strings.HasSuffix(filepath.ToSlash(path), "internal/broker/composed.go") {
!strings.HasSuffix(filepath.ToSlash(path), "internal/broker/composed.go") &&
!strings.HasSuffix(filepath.ToSlash(path), "internal/broker/userlist_switch.go") {
found = append(found, path+": "+lit)
}
}
+11
View File
@@ -72,6 +72,17 @@ func TestAnAccountMayReadItsOwnMembershipAndNoOthers(t *testing.T) {
has(t, perms.Subscribe, "mesh.assignment.anchor.postgres")
has(t, perms.Publish, "$JS.API.DIRECT.GET.ASSIGNMENTS.mesh.assignment.anchor.postgres")
hasNot(t, perms.Subscribe, "mesh.assignment.>")
hasNot(t, perms.Publish, "mesh.assignment.*.*")
// The grants part alone publishes memberships (ADR 0547 §5), and defines no stream.
grants, err := PermissionsFor(Principal{Kind: KindModule, Node: "anchor", Module: GrantsModule, PasswordHash: "x"})
if err != nil {
t.Fatal(err)
}
has(t, grants.Publish, "mesh.assignment.*.*")
has(t, grants.Publish, "$JS.API.DIRECT.GET.ASSIGNMENTS.mesh.assignment.*.*")
hasNot(t, perms.Publish, "$JS.API.DIRECT.GET.ASSIGNMENTS.mesh.assignment.*.*")
hasNot(t, grants.Publish, "$JS.API.STREAM.CREATE.>")
hasNot(t, grants.Publish, "mesh.assignment.>")
}
// The runtime arriving on a machine changes nothing about what each module is issued (to-be 38 WP2):
+2 -3
View File
@@ -52,9 +52,8 @@ func TestAModuleStreamOnARealBus(t *testing.T) {
t.Fatalf("publishing %s's mail: %v", account, err)
}
}
time.Sleep(50 * time.Millisecond)
if info, _ := js.js.ConsumerInfo("MAIL", c.Name); info.NumPending != 1 {
t.Fatalf("the reader granted `mesh` has %d pending, want 1", info.NumPending)
if got := pendingOf(js, "MAIL", c.Name, 1); got != 1 {
t.Fatalf("the reader granted `mesh` has %d pending, want 1", got)
}
narrowed := c
narrowed.Filters = []string{"mesh.stream.mesh-mail.event.mail.mesh.message.changed"}
+42
View File
@@ -214,9 +214,17 @@ type Principal struct {
// StatesItsKeyToTheVault is a machine principal whose node-engine may ask the vault's `node-key` in its own name
// (issue 786, novox/hq ADR 0372): set while the vault's holder serves it. That one verb, and nothing else new.
StatesItsKeyToTheVault bool
// FollowsGrantsUserList is the machine principal of the node holding mesh-broker: its node-engine reads the grants
// part's sealed user list, one key and an object by name (novox/hq ADR 0547 clause 7, GrantsUserListFollows).
FollowsGrantsUserList bool
// BoundBuckets are the providers' buckets of entries, `<module>_bound`, a provisioner client reads its own
// machine's keys of (mesh-sdk go/provider, issue 679): every bucket named bound of a module the mesh places.
BoundBuckets []string
// CreatesStreams are the streams the grants part's principal alone may create (novox/hq ADR 0547 §5; the operator,
// 2026-10-26 08:54): the stream of every bucket, log and stream a manifest of a module placed anywhere declares,
// by name. Created only, never updated or deleted: a new module state is on the bus when the grants part hears its
// manifest, not at the controller's next start (issue 1790).
CreatesStreams []string
// PasswordHash is the bcrypt hash the mesh minted. The plaintext is sealed to the principal
// and never appears here: this file is written to a node's disk and read by a server, and a
@@ -908,6 +916,11 @@ func PermissionsFor(p Principal) (Permissions, error) {
if p.StatesItsKeyToTheVault {
pub = append(pub, "mesh.seat."+VaultNodeKey.Seat+"."+CallKind+"."+VaultNodeKey.Verb+"."+CallerToken(p.Username()))
}
// And, on the node holding mesh-broker, the grants part's sealed user list it follows from take-over (novox/hq ADR
// 0547 clause 7): one key, and an object by its name.
if p.FollowsGrantsUserList {
pub = append(pub, GrantsUserListFollows()...)
}
case KindModule:
// 0. The watchman alone hears who connects and disconnects, and asks who is connected (ADR 0431 §2).
@@ -915,6 +928,17 @@ func PermissionsFor(p Principal) (Permissions, error) {
sub = append(sub, BusConnectSubject, BusDisconnectSubject)
pub = append(pub, BusConnzSubject)
}
// The grants part alone publishes the modules' memberships into the assignments stream (novox/hq ADR 0547 §5;
// the operator, 2026-10-25 22:31): one per subject, `mesh.assignment.<node>.<module>`, reading the last one of each
// first so it publishes only what changed, and no stream it defines.
if p.Module == GrantsModule {
pub = append(pub, "mesh.assignment.*.*", "$JS.API.DIRECT.GET."+AssignmentsStream+".mesh.assignment.*.*")
// And creating each stream a manifest declares, by its own name (the operator, 2026-10-26 08:54; issue
// 1790): the create and the read of that one stream, never an update, a delete or a stream not declared.
pub = append(pub, CreateStreamGrants(p.CreatesStreams)...)
// And its own object bucket, where it writes the user list sealed to the bus node (ADR 0547 clause 7).
pub = append(pub, GrantsUserListWrites()...)
}
// The view alone reads the build seats' workers for the backlog of asks (issue 1887 A8a).
if p.Module == ViewModule {
pub = append(pub, BuildBacklogReads()...)
@@ -1581,6 +1605,24 @@ func ComposeAccounts(principals []Principal) (string, error) {
// from them. No other principal is granted the subjects below, and nothing is a wildcard over them.
const WatchmanModule = "mesh-delivery-watchman"
// CreateStreamGrants is what creating each named stream needs, and nothing wider (novox/hq ADR 0547 §5; the operator,
// 2026-10-26 08:54): its create, and its information to read back what it holds. Each by its own name: a name that
// is not one plain subject token is left out, since a wildcard there would be every stream.
func CreateStreamGrants(streams []string) []string {
var out []string
for _, s := range streams {
if !safeSubject.MatchString(s) {
continue
}
out = append(out, "$JS.API.STREAM.CREATE."+s, "$JS.API.STREAM.INFO."+s)
}
return out
}
// GrantsModule is the one module whose principal publishes the modules' memberships (novox/hq ADR 0547 §5): the grants
// part, which composes them from the manifests and placement. It defines no stream and is granted no other subject here.
const GrantsModule = "mesh-grants"
// The bus's own events, imported into MESH from the default system account `$SYS` under subjects of MESH's own
// (novox/hq ADR 0431 §1, issue 1147): the server says an account's connections opening and closing only to the system
// account, and the list of an account's connections only to a request made there. `$SYS` holds no user, so nobody
+20 -2
View File
@@ -24,8 +24,9 @@ import (
// subjects cover it, so what travels there — a code typed by the operator — is never persisted. A kinded
// holder asks with its own kind; the modules that watch the seat answer.
//
// And one read: **a holder's records**, a bucket the seat names, read by each user under its own name only
// (`$KV.<bucket>.<module>.>`), so an asker reads the state of its own asks and no other asker's.
// And two reads: **a holder's records**, a bucket the seat names, read by each user under its own name only
// (`$KV.<bucket>.<module>.>`), so an asker reads the state of its own asks and no other asker's; and on a seat pulled by
// kind, **what its workers gave up on**, the copies DEAD_LETTERS keeps of the seat's stream (DeadLetterReads, issue 734).
// BuildAskMaxDeliver bounds the deliveries of a build seat's ask (issue 1907): mesh-sdk delivery.AskMaxDeliver, written
// here as the broker mirrors the catalogue rather than importing it.
@@ -129,6 +130,9 @@ func SeatTrafficOf(module string, holds, uses, watches []Seat) SeatTraffic {
}
t.Records = append(t.Records, "$JS.API.DIRECT.GET.KV_"+b+".$KV."+b+"."+module+".>")
}
if s.PulledByKind && len(s.Accepts) > 0 && safeSubject.MatchString(s.Name) {
t.Records = append(t.Records, DeadLetterReads(seatStreamName(s.Name))...)
}
}
for _, w := range watches {
if w.Kinded {
@@ -151,6 +155,20 @@ func SeatTrafficOf(module string, holds, uses, watches []Seat) SeatTraffic {
return t
}
// DeadLetterReads is what a user of a seat pulled by kind is granted to hear which of the asks it put the seat's workers
// gave up on (novox/hq issue 734): an ordered consumer on DEAD_LETTERS filtered to the copies the controller keeps of that
// seat's stream (DeadLetterSubject(stream, *)), pulled, and its deletion. A copy carries the ask's own message id
// (Mesh-Dead-Msg-Id), so the asker ends the step it waits for instead of waiting for ever; it reads the seat's dead
// letters and no other stream's, and changes none.
func DeadLetterReads(stream string) []string {
return []string{
"$JS.API.STREAM.INFO." + DeadLettersStream,
"$JS.API.CONSUMER.CREATE." + DeadLettersStream + ".*." + DeadLetterSubject(stream, ">"),
"$JS.API.CONSUMER.MSG.NEXT." + DeadLettersStream + ".*",
"$JS.API.CONSUMER.DELETE." + DeadLettersStream + ".*",
}
}
// seatTrafficGrants is the bus permissions seat traffic needs: the subjects themselves, and the JetStream API a
// worker is pulled and acknowledged through and a record is read through.
func seatTrafficGrants(t SeatTraffic) (pub, sub []string) {
+13
View File
@@ -403,6 +403,19 @@ func TestTheBuilderMayAskABuildSeatAndASessionMayNot(t *testing.T) {
t.Errorf("the builder may not publish %s", s)
}
}
// Issue 734: the asker hears which of its asks the seat's workers gave up on, from DEAD_LETTERS, and no other stream's.
for subject, want := range map[string]bool{
"$JS.API.CONSUMER.CREATE.DEAD_LETTERS.x1.mesh.events.dead.SEAT_NODE_BUILD_GO.>": true,
"$JS.API.CONSUMER.DELETE.DEAD_LETTERS.x1": true,
"$JS.API.CONSUMER.MSG.NEXT.DEAD_LETTERS.x1": true,
"$JS.API.CONSUMER.CREATE.DEAD_LETTERS.x1.mesh.events.dead.EVENTS.>": false,
"$JS.API.STREAM.PURGE.DEAD_LETTERS": false,
"$JS.FC.DEAD_LETTERS.x1": false,
} {
if got := allowed(builder.Publish, subject); got != want {
t.Errorf("the builder may publish %s: %v, want %v", subject, got, want)
}
}
session := perms(t, Principal{Kind: KindPerson, Module: "jochen", Invokes: []string{"mesh-controller.status"}})
if allowed(session.Publish, "mesh.seat.node-build-go.accept.delivery") {
t.Error("a session's account may put an ask")
+34 -22
View File
@@ -339,13 +339,36 @@ func Overlaps() []string {
// ack subject is derived from (nats.go: `$JS.ACK.<stream>.controller.>`).
const ControllerName = "controller"
// MergesConsumerName is the controller's durable consumer of the forge's merged event on EVENTS, apart from the
// events consumer (novox/hq ADR 0407 decision 3, issue 973). A merge is never dropped, so this consumer has no
// max-deliver, which would send a merge held five times to DEAD_LETTERS to wait for a person, and no healer
// resets it, which would drop the merges it holds. A consumer of its own because both of those are properties of
// a consumer, not of a subject.
// MergesConsumerName was the controller's durable consumer of the forge's merged event on EVENTS (novox/hq ADR 0407
// decision 3, issue 973), with no max-deliver and one message at a time. **Retired** (sweep 2026-10-30, issue 774):
// the forge bridge's stream FORGE is the record of the merges, and the controller reads it where it needs a merge. On
// 2026-10-29 one announcement too large for the controller's bucket of merges was held on it for over a day, with
// 163 merges queued behind it, and the deaf watchdog read the queue as a deaf controller. RetiredConsumers removes it.
const MergesConsumerName = "controller-merges"
// RetiredConsumers are durable consumers the controller once held and holds no more: each assertion removes them,
// so a retired consumer cannot keep a queue nobody takes from (research 072, 03-domains: the controller shrinks).
var RetiredConsumers = []Consumer{{Name: MergesConsumerName, Stream: EventsStream}}
// Remover is the part of a JetStream connection that removes a retired consumer.
type Remover interface {
RemoveConsumer(stream, name string) error
}
// removeRetired removes every retired consumer, when the connection can.
func removeRetired(e Ensurer) error {
r, ok := e.(Remover)
if !ok {
return nil
}
for _, c := range RetiredConsumers {
if err := r.RemoveConsumer(c.Stream, c.Name); err != nil {
return fmt.Errorf("removing the retired consumer %s on %s: %w", c.Name, c.Stream, err)
}
}
return nil
}
// ControllerSeat is the role the control plane holds, and ControllerStates are the facts it states
// under it (novox/hq ADR 0134).
//
@@ -513,23 +536,6 @@ func ControllerConsumers() []Consumer {
Why: "the controller is the single consumer of what nodes say; explicit ack and no " +
"max-deliver, because the store window's bound is the controller's own",
},
// The forge's merges (novox/hq ADR 0407 decision 3, issue 973): never dropped, so no max-deliver and no
// reset; a merge the controller cannot record yet is held with a growing delay until it can. **Asserted before
// the events consumer**, which stops filtering merges: the order of this list is the order of assertion.
{
Name: MergesConsumerName,
Stream: "EVENTS",
// Within what the controller may subscribe already (`_DELIVER.controller.>`).
DeliverSubject: "_DELIVER." + ControllerName + ".merges",
Filters: []string{ControllerFollows[3]},
Push: true,
AckWaitSeconds: 30,
MaxAckPending: 1,
FromNow: true,
TakesOverFrom: ControllerName,
Why: "the forge's merges, one at a time; no max-deliver and never reset, because a merge is never " +
"dropped and no person releases one",
},
{
Name: ControllerName,
Stream: "EVENTS",
@@ -581,6 +587,9 @@ func NoticesConsumer() Consumer {
// one on DEAD_LETTER_NOTICES, which the keeper of dead letters asserts and retries by itself, so a fault
// there never stops the controller serving (novox/hq issue 330).
func AssertServingConsumers(e Ensurer) error {
if err := removeRetired(e); err != nil {
return err
}
for _, c := range MeshConsumers() {
if c.Stream == DeadLetterNoticesStream {
continue
@@ -603,6 +612,9 @@ type Ensurer interface {
// After the streams, necessarily: a consumer on a stream that does not exist is refused, and the
// refusal names the stream rather than the order.
func AssertMeshConsumers(e Ensurer) error {
if err := removeRetired(e); err != nil {
return err
}
for _, c := range MeshConsumers() {
if err := e.EnsureConsumer(c); err != nil {
return fmt.Errorf("asserting consumer %s on %s: %w", c.Name, c.Stream, err)
+55
View File
@@ -0,0 +1,55 @@
package broker
// **The grants part's user list, sealed to the bus node** (novox/hq ADR 0547 clause 7, issue 1945): the operator's
// second exception to the frozen controller (the first, memberships, 2026-10-25 22:31). The controller's copy of the bus
// composer grants what mesh-sdk's copy grants, byte for byte, and nothing else: the node-engine of the node holding
// mesh-broker reads the grants part's one key `user-list.sealed`, watched, and an object of its bucket by name; the
// grants part writes that bucket. Without it the bus node could never read the first list the grants part writes, since
// until take-over the list the bus loads is this one (mesh-sdk go/buscompose/userlist_switch.go keeps the other copy).
// The grants part's user list on the bus (mesh-sdk go/delivery userlist.go names them).
const (
GrantsUserListBucket = "mesh-grants_user-list"
GrantsUserListKey = "sealed"
GrantsUserListObjects = "mesh-grants-user-list"
GrantsUserListObjectStream = "OBJ_" + GrantsUserListObjects
)
// GrantsUserListFollows is what the node-engine of the node holding mesh-broker may publish to follow the grants part's
// list: the one key read and watched, and an object of the bucket by its name (as ADR 0486 §2 reads a declaration's
// files: never the object stream's info, which lists every name, never a message by sequence).
func GrantsUserListFollows() []string {
kv := "KV_" + GrantsUserListBucket
key := "$KV." + GrantsUserListBucket + "." + GrantsUserListKey
obj, s := GrantsUserListObjects, GrantsUserListObjectStream
return []string{
"$JS.API.DIRECT.GET." + kv + "." + key,
"$JS.API.CONSUMER.CREATE." + kv + ".*." + key,
"$JS.API.CONSUMER.DELETE." + kv + ".*",
"$JS.API.STREAM.INFO." + kv,
"$JS.FC." + kv + ".>",
"$JS.API.DIRECT.GET." + s + ".$O." + obj + ".M.*",
"$JS.API.CONSUMER.CREATE." + s + ".*.$O." + obj + ".C.*",
"$JS.API.CONSUMER.DELETE." + s + ".*",
"$JS.FC." + s + ".>",
}
}
// GrantsUserListWrites is what the grants part may publish to keep its object bucket.
func GrantsUserListWrites() []string {
obj, s := GrantsUserListObjects, GrantsUserListObjectStream
return []string{
"$JS.API.STREAM.CREATE." + s,
"$JS.API.STREAM.INFO." + s,
"$JS.API.STREAM.PURGE." + s,
"$JS.API.DIRECT.GET." + s + ".>",
"$JS.API.STREAM.MSG.GET." + s,
"$O." + obj + ".>",
}
}
// grantsOwnObjects is the grants part writing its own object bucket, and nothing of another.
func grantsOwnObjects(p Principal, pattern string) bool {
return p.Kind == KindModule && p.Module == GrantsModule &&
(pattern == "$O."+GrantsUserListObjects+".>" || pattern == "$JS.API.STREAM.CREATE."+GrantsUserListObjectStream)
}
+55 -2
View File
@@ -147,14 +147,19 @@ func Users(r Records) ([]Principal, error) {
for _, node := range sortedCopy(r.Nodes) {
witness := false
var checks []string
bus := false
for _, d := range r.Assigned[node] {
if d.Module == controllerModule {
witness = true
}
if d.SnapshotsTheBus {
bus = true
}
checks = append(checks, d.Checks...)
}
out = append(out, Principal{Kind: KindNode, Node: node, WitnessesController: witness, Checks: checks,
MovesIntoTheVault: servesMoveIn(r), StatesItsKeyToTheVault: servesVerb(r, VaultNodeKey)})
MovesIntoTheVault: servesMoveIn(r), StatesItsKeyToTheVault: servesVerb(r, VaultNodeKey),
FollowsGrantsUserList: bus})
// **And the machine's provisioner client, on an account of its own** (novox/hq ADR 0349 clause 11): one
// per machine that runs it, delivered to the module running it.
for _, d := range r.Assigned[node] {
@@ -176,6 +181,11 @@ func Users(r Records) ([]Principal, error) {
}
}
for _, d := range r.Assigned[node] {
// **The node-engine's manifest is no placed module's** (novox/hq ADR 0563 §2, §3; sweep 2026-10-30, issue
// 774): the guard mesh-sdk #136 put in buscompose, kept here while the controller composes with its own copy.
if !Composed(d) {
continue
}
// **A process account is its own, delivered in its own secret** (broker-<process id>, novox/hq
// ADR 0436 §1, §2), never the module's `broker` — so it is composed whether or not the module
// itself has an account, and is never carried by the machine's runtime, which runs as the
@@ -207,6 +217,11 @@ func Users(r Records) ([]Principal, error) {
PerMachine: perMachineNames(d.State), KeyedReads: d.KeyedReads,
Logs: logNames(d.Logs), Streams: d.Streams, Hears: ComposedHears(d), Queues: d.Queues,
})
// **The grants part creates what a manifest declares** (novox/hq ADR 0547 §5; the operator, 2026-10-26
// 08:54): every placed module's buckets, logs and streams, each by its stream's name.
if d.Module == GrantsModule {
out[len(out)-1].CreatesStreams = declaredStreamsOf(r)
}
// **The vault's holder asks the desks and the issuing providers itself** (issue 723), under the
// account its own runtime runs on (ADR 0259 §8's runs-as): the one module granted it.
if holdsSeat(d, VaultSeat) {
@@ -217,7 +232,7 @@ func Users(r Records) ([]Principal, error) {
if runtimeHere {
var carried []Declared
for _, d := range r.Assigned[node] {
if d.RunsAs == "" && runnerServes(d) {
if CarriedByRunner(d) {
// Never a restricted stream for what the tool runner carries (novox/hq ADR 0376 §1).
d.Hears = ComposedHears(d)
carried = append(carried, d)
@@ -418,3 +433,41 @@ func boundBucketsOf(r Records) []string {
func runnerServes(d Declared) bool {
return d.Module == RuntimeModule || d.NoAccount || d.Loads || len(d.Serves) > 0 || len(d.Holds) > 0 || len(d.Waits) > 0 || len(d.Checks) > 0
}
// declaredStreamsOf is the stream of every bucket, log and stream a module placed anywhere declares, once each, in name
// order (novox/hq ADR 0547 §5, issue 1790): `KV_<module>_<state>`, `LOG_<module>_<log>` and each module stream's own
// name. What the grants part may create, and nothing beside.
func declaredStreamsOf(r Records) []string {
seen := map[string]bool{}
for _, mods := range r.Assigned {
for _, d := range mods {
for _, b := range d.State {
seen["KV_"+b.Bucket()] = true
}
for _, l := range d.Logs {
seen[l.Stream()] = true
}
for _, s := range d.Streams {
seen[s.Name] = true
}
}
}
out := make([]string, 0, len(seen))
for s := range seen {
out = append(out, s)
}
sort.Strings(out)
return out
}
// EngineModule is the node-engine's manifest's module name (mesh-sdk buscompose.EngineModule).
const EngineModule = "mesh-host"
// Composed says a placed module is composed as one: its own user, its processes and what the runner carries of it. The
// node-engine's manifest is not, though a placement may name it: each node's engine speaks as the node's account, which
// alone writes the engine's state (ADR 0563 §2, §3); granted to another account it is a second writer, and the writers
// table refuses the whole list (issue 774: mesh-host_acts, then mesh-host_health, 2026-10-30). mesh-sdk buscompose.Composed.
func Composed(d Declared) bool { return d.Module != EngineModule }
// CarriedByRunner says the machine's tool runner carries this placed module (mesh-sdk buscompose.CarriedByRunner).
func CarriedByRunner(d Declared) bool { return Composed(d) && d.RunsAs == "" && runnerServes(d) }
+20 -2
View File
@@ -37,6 +37,20 @@ type WriterRow struct {
// isController, and the other writers' tests, are who a row's writer is as a principal.
func isController(p Principal, _ string) bool { return p.Kind == KindController }
// createsDeclared is the controller, or the grants part creating one stream a manifest declares, by its own name: the
// create alone, never an update or a delete (novox/hq ADR 0547 §5; the operator, 2026-10-26 08:54).
func createsDeclared(p Principal, pattern string) bool {
if isController(p, pattern) {
return true
}
name, ok := strings.CutPrefix(pattern, "$JS.API.STREAM.CREATE.")
if ok && p.Kind == KindModule && p.Module == GrantsModule && name == GrantsUserListObjectStream {
return true
}
return ok && p.Kind == KindModule && p.Module == GrantsModule && slices.Contains(p.CreatesStreams, name) &&
safeSubject.MatchString(name)
}
// ownMachine is a node writing a subject whose third token is its own name, and no wildcard there.
func ownMachine(p Principal, pattern string) bool {
tokens := strings.Split(pattern, ".")
@@ -170,7 +184,8 @@ var WritersTable = []WriterRow{
// bucket's stream (ADR 0201), which defines nothing the mesh keeps.
Subjects: []string{"$JS.API.STREAM.CREATE.>", "$JS.API.STREAM.UPDATE.>", "$JS.API.STREAM.DELETE.>",
"$JS.API.CONSUMER.DURABLE.CREATE.>"},
Writes: isController},
Writes: createsDeclared,
Shared: "the grants part creates a stream a manifest declares, by its name, and never updates or deletes one (novox/hq ADR 0547 §5; the operator, 2026-10-26 08:54)"},
{State: "builds and their outcomes", Writer: "the build seat's holder", KeptIn: "its own state",
Others: "the controller asks",
Subjects: []string{"mesh.seat.node-build-agent.event.built", "mesh.seat.mesh-build-machine.event.built"},
@@ -237,7 +252,10 @@ var WritersTable = []WriterRow{
KeptIn: "the object bucket " + DeclarationFilesBucket + "-<node> of each node, an object removed a day after the " +
"node's latest declaration no longer names it",
Others: "a node-engine reads an object of its own bucket by its name and may neither list the bucket nor write it",
Subjects: []string{"$O.*.>"}, Writes: isControllerOrTheHandPath},
Subjects: []string{"$O.*.>"}, Writes: func(p Principal, pattern string) bool {
return isControllerOrTheHandPath(p, pattern) || grantsOwnObjects(p, pattern)
},
Shared: "the grants part writes its own object bucket " + GrantsUserListObjects + ", the user list sealed to the bus node, and no other (novox/hq ADR 0547 clause 7)"},
{State: "the declared-config mirror", Writer: "controller (lease holder), from its store, only on a change",
KeptIn: "key-value " + DeclaredConfigBucket, Others: "read through `reads` as mesh-controller." + MirrorDeclaredConfig,
Subjects: kvOf(DeclaredConfigBucket), Writes: isController},
+34 -10
View File
@@ -6,7 +6,9 @@ import (
"git.novox.be/novox/mesh-sdk/go/announce"
"slices"
"strings"
"sync"
"git.novox.be/novox/mesh-sdk/go/buscompose"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"github.com/novox/mesh-controller/internal/broker"
)
@@ -119,22 +121,18 @@ func (i *Inventory) BusRecords(ctx context.Context) (broker.Records, error) {
// announcement does not make readable is granted nothing, and said; an owner that announced nothing yet is read as
// before. Announcements that cannot be read are said, and every read is granted as before: the bus's users are
// composed either way.
//
// **Judged for its reader, by the SDK's judgement** (sweep 2026-10-30, issue 774): mesh-sdk buscompose's
// ReadsAnnouncedBy knows the named readers (ADR 0547 clause 9, ADR 0561 §3), so the lease is granted to the parts
// named for it and refused to every other. The controller's own reader-less copy refused every named reader: 2536
// refused reads of mesh-controller.lease on 2026-10-30, each said on every composition.
if Announcements != nil {
announced, err := Announcements(ctx)
if err != nil {
fmt.Printf("the states the modules announced cannot be read, so every read is granted from `reads` as before: %v\n", err)
} else {
out.Announced = announced
for node, ds := range out.Assigned {
for k := range ds {
granted, refused := broker.ReadsAnnounced(ds[k].Reads, announced)
ds[k].Reads, ds[k].ReadsRefused = granted, refused
for _, r := range refused {
fmt.Printf("%s on %s reads %s, which its owner does not announce readable: not granted\n",
ds[k].Module, node, r)
}
}
}
judgeReads(out.Assigned, announced, sayRefusedOnce)
}
}
@@ -615,3 +613,29 @@ func hearsOf(m catalogue.Manifest, declared map[string]catalogue.Manifest, node
}
return out, touches, onEvents
}
// refusalsSaid are the refused reads already said by this process, by `<module>@<node> <read>`: a refusal is said once,
// when first composed, not on every composition (about 420 times a read on 2026-10-30). `status` names them all.
var refusalsSaid sync.Map
// sayRefusedOnce says a refused read the first time this process composes it.
func sayRefusedOnce(module, node, read string) {
if _, said := refusalsSaid.LoadOrStore(module+"@"+node+" "+read, true); said {
return
}
fmt.Printf("%s on %s reads %s, which its owner does not make readable by it: not granted\n", module, node, read)
}
// judgeReads splits each assigned module's reads into granted and refused, for that module, by what the owners announced
// and the states' named readers (buscompose.ReadsAnnouncedBy), and says each refusal.
func judgeReads(assigned map[string][]broker.Declared, announced map[string]announce.States, say func(module, node, read string)) {
for node, ds := range assigned {
for k := range ds {
granted, refused := buscompose.ReadsAnnouncedBy(ds[k].Module, ds[k].Reads, announced)
ds[k].Reads, ds[k].ReadsRefused = granted, refused
for _, r := range refused {
say(ds[k].Module, node, r)
}
}
}
}
+44
View File
@@ -0,0 +1,44 @@
package inventory
import (
"slices"
"testing"
"git.novox.be/novox/mesh-sdk/go/announce"
"github.com/novox/mesh-controller/internal/broker"
)
// Sweep 2026-10-30 (issue 774): a read is judged for its reader. The controller's lease, which its announcement marks
// readable by nobody, is granted to the parts named for it (ADR 0561 §3) and refused to every other (mesh-config named since mesh-sdk ef1ec1c9); a state its owner
// announced readable is granted; one it keeps and did not is refused; an owner that announced nothing is read as before.
// Each refusal is said, and a refusal already said is not said again.
func TestAReadIsJudgedForItsReader(t *testing.T) {
announced := map[string]announce.States{
"mesh-controller": broker.ControllerAnnouncement("novox"),
"owner": {Module: "owner", States: []announce.State{{Name: "open", Readable: true}, {Name: "closed"}}},
}
for name, c := range map[string]struct {
reader, read string
granted bool
}{
"the lease, by the vault": {"mesh-vault", "mesh-controller.lease", true},
"the lease, by the grants part": {"mesh-grants", "mesh-controller.lease", true},
"the lease, by mesh-config": {"mesh-config", "mesh-controller.lease", true},
"the lease, by a module not named": {"restic", "mesh-controller.lease", false},
"a readable state": {"reader", "owner.open", true},
"a state kept and not readable": {"reader", "owner.closed", false},
"an owner that announced nothing": {"reader", "quiet.any", true},
"a mirror the controller marks open": {"pacman", "mesh-controller.declared-packages", true},
} {
assigned := map[string][]broker.Declared{"novox": {{Module: c.reader, Reads: []string{c.read}}}}
var said []string
judgeReads(assigned, announced, func(module, node, read string) { said = append(said, read) })
d := assigned["novox"][0]
if slices.Contains(d.Reads, c.read) != c.granted || slices.Contains(d.ReadsRefused, c.read) == c.granted {
t.Errorf("%s: granted %v, refused %v", name, d.Reads, d.ReadsRefused)
}
if (len(said) == 1) == c.granted {
t.Errorf("%s: said %v", name, said)
}
}
}
+12 -3
View File
@@ -67,14 +67,23 @@ func (i *Inventory) HighestEpoch(ctx context.Context) (uint64, error) {
}
// TookEpoch records an epoch taken, and ends every earlier one still open as found expired: the lease
// was free to take, so whoever held it last neither holds it nor said it let go. Answers the epochs it
// ended that way, newest first.
func (i *Inventory) TookEpoch(ctx context.Context, e Epoch) ([]Epoch, error) {
// was free to take, so whoever held it last neither holds it nor said it let go. givenBack is the epoch the
// bus says gave the key back (the lease's own delete marker names it, novox/hq issue 906, ADR 0558 §3): that
// epoch, if still open, is ended as released — its holder gave the key back and had not yet recorded it — and
// every other as expired; zero is none. Answers the epochs it ended as expired, newest first.
func (i *Inventory) TookEpoch(ctx context.Context, e Epoch, givenBack uint64) ([]Epoch, error) {
tx, err := i.store.Pool().Begin(ctx)
if err != nil {
return nil, err
}
defer func() { _ = tx.Rollback(ctx) }()
if givenBack != 0 && givenBack < e.Epoch {
if _, err := tx.Exec(ctx,
`update controller_epoch set ended = now(), how = $2 where epoch = $1 and ended is null`,
int64(givenBack), EpochReleased); err != nil {
return nil, err
}
}
rows, err := tx.Query(ctx,
`update controller_epoch set ended = now(), how = $2
where ended is null and epoch < $1
+11
View File
@@ -26,6 +26,17 @@ type ProvReport struct {
WaitsFor []string `json:"waits-for,omitempty"`
Templates string `json:"templates,omitempty"`
At time.Time `json:"at"`
// FailingSince is, on a failed report, when the module's standing failure began, kept by the client while only its
// words change (mesh-catalog 3aae6c08); zero from an older client.
FailingSince time.Time `json:"failing-since,omitzero"`
}
// Since is when a failed report's failure began: FailingSince when the client says it, else At.
func (r ProvReport) Since() time.Time {
if !r.FailingSince.IsZero() {
return r.FailingSince
}
return r.At
}
// The words a client's report says.
+3 -2
View File
@@ -127,9 +127,10 @@ var (
// begin.
var PlacementKinds = []string{"dir", "access", "port", "seat", "machine", "backup"}
// credentialForms are the forms a provider's offer may declare for its credential (issue 768).
// credentialForms are the forms a provider's offer may declare for its credential (issue 768); "none" is an offer with
// no credential, which the release's catalogue states on every such offer (issue 774).
var credentialForms = []string{catalogue.CredentialFormShared, catalogue.CredentialFormPair,
catalogue.CredentialFormIssued, catalogue.CredentialFormGiven}
catalogue.CredentialFormIssued, catalogue.CredentialFormGiven, catalogue.CredentialFormNone}
var (
digestShape = regexp.MustCompile(`^sha256:[0-9a-f]{64}$`)
+4 -2
View File
@@ -316,14 +316,16 @@ func TestAPlanNamesTheCredentialFormOfEachProvidersOffer(t *testing.T) {
p := aPlan()
p.Provisions = []ProvisionedCredential{
{Provision: "postgres-database", Provider: "novox", ProviderModule: "postgres", CredentialForm: "pair"},
{Provision: "sonarr-api", Provider: "ace", ProviderModule: "sonarr", CredentialForm: "shared"}}
{Provision: "sonarr-api", Provider: "ace", ProviderModule: "sonarr", CredentialForm: "shared"},
{Provision: "artifact-store", Provider: "novox", ProviderModule: "distribution", CredentialForm: "none"}}
values, err := AssignmentValues(assigned, map[string]Provisioning{"g14.dunst": p})
if err != nil {
t.Fatal(err)
}
if want := `{"provision":"postgres-database","provider":"novox","provider-module":"postgres",` +
`"credential-form":"pair"},{"provision":"sonarr-api","provider":"ace","provider-module":"sonarr",` +
`"credential-form":"shared"}`; !strings.Contains(string(values["g14.dunst"]), want) {
`"credential-form":"shared"},{"provision":"artifact-store","provider":"novox","provider-module":"distribution",` +
`"credential-form":"none"}`; !strings.Contains(string(values["g14.dunst"]), want) {
t.Errorf("the plan's value is %s, want it to carry %s", values["g14.dunst"], want)
}
bad := aPlan()
+71 -9
View File
@@ -34,9 +34,11 @@ import (
"encoding/json"
"errors"
"fmt"
"strconv"
"sync"
"time"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
)
@@ -115,6 +117,7 @@ type Options struct {
// Lease is one instance's hold, or its wait for one.
type Lease struct {
kv jetstream.KeyValue
js jetstream.JetStream
stream jetstream.Stream
ttl time.Duration
o Options
@@ -139,6 +142,9 @@ type Lease struct {
lostWhy error
lost chan struct{}
taken time.Time
// givenBack is the epoch whose holder gave back the key this instance took — the epoch its delete marker
// names on the bus — and zero when the key was left to expire, or freed by anything else (novox/hq issue 906).
givenBack uint64
// update writes the key at a revision: the bucket's own, or a test's bus that does not answer.
update func(ctx context.Context, key string, value []byte, last uint64) (uint64, error)
}
@@ -184,7 +190,7 @@ func Open(ctx context.Context, js jetstream.JetStream, bucket string, o Options)
return nil, fmt.Errorf("a lease renewed every %s with a margin of %s does not fit a key that lives %s",
o.RenewEvery, o.Margin, status.TTL())
}
return &Lease{kv: kv, stream: stream, ttl: status.TTL(), o: o, lost: make(chan struct{}), update: kv.Update}, nil
return &Lease{kv: kv, js: js, stream: stream, ttl: status.TTL(), o: o, lost: make(chan struct{}), update: kv.Update}, nil
}
// TTL is how long the key lives unrenewed.
@@ -271,6 +277,7 @@ func (l *Lease) TryTake(ctx context.Context) (uint64, error) {
} else if found {
return 0, &heldBy{h}
}
givenBack := l.wasGivenBack(ctx)
if err := l.aboveTheFloor(ctx); err != nil {
return 0, err
}
@@ -307,13 +314,51 @@ func (l *Lease) TryTake(ctx context.Context) (uint64, error) {
"back to be taken again", revision, floor)
}
l.mu.Lock()
l.held, l.epoch, l.revision, l.taken = true, revision, revision, now
l.held, l.epoch, l.revision, l.taken, l.givenBack = true, revision, revision, now, givenBack
l.renewedAt(anchor)
l.mu.Unlock()
l.o.Say("took the controller lease at epoch %d (%s)", revision, h.Instance)
return revision, nil
}
// ReleasedHeader is the header of the delete marker a holder's Release writes, naming the epoch it gave back:
// the one mark that says a handover. A delete marker without it — a command's own lease given back, a take
// given back over the floor — names no epoch, so it says nothing about the holder before (novox/hq issue 906).
const ReleasedHeader = "Mesh-Lease-Released"
// wasGivenBack is the epoch the free key was given back by: the epoch its last message on the bus — the delete
// marker a holder's Release writes — names. Zero when the key expired (the bucket's age removed every message),
// when the marker names no epoch, and when it cannot be read: a takeover nobody can prove was a handover is
// said as one that was not (novox/hq issue 906, ADR 0558 §3).
func (l *Lease) wasGivenBack(ctx context.Context) uint64 {
if l.stream == nil {
return 0
}
last, err := l.stream.GetLastMsgForSubject(ctx, l.subject())
if err != nil || last == nil {
return 0
}
if op := last.Header.Get("KV-Operation"); op != "DEL" && op != "PURGE" {
return 0
}
epoch, err := strconv.ParseUint(last.Header.Get(ReleasedHeader), 10, 64)
if err != nil {
return 0
}
return epoch
}
// subject is the lease key's subject in its bucket's stream.
func (l *Lease) subject() string { return "$KV." + l.kv.Bucket() + "." + Key }
// GivenBack is the epoch whose holder gave back the key this instance took, zero when it was not given back:
// what the takeover was, read from the bus (novox/hq issue 906).
func (l *Lease) GivenBack() uint64 {
l.mu.Lock()
defer l.mu.Unlock()
return l.givenBack
}
// floor is the highest epoch issued, zero without a floor.
func (l *Lease) floor(ctx context.Context) (uint64, error) {
if l.o.Floor == nil {
@@ -588,9 +633,11 @@ func (l *Lease) lose(why error) error {
return l.lostWhy
}
// Release gives the lease back, so the next candidate takes it at once rather than after its age.
// Only the key this instance last wrote is deleted: one that moved is somebody else's.
func (l *Lease) Release(ctx context.Context) {
// Release gives the lease back, so the next candidate takes it at once rather than after its age, and answers
// the epoch it gave back — zero when it gave nothing back: it held none, had lost it, or the bus would not
// delete the key. Only the key this instance last wrote is deleted: one that moved is somebody else's. The
// gate closes before the key goes, so nothing this instance starts after overlaps its successor.
func (l *Lease) Release(ctx context.Context) uint64 {
l.mu.Lock()
held, revision, epoch := l.held && l.lostWhy == nil, l.revision, l.epoch
l.held = false
@@ -600,11 +647,11 @@ func (l *Lease) Release(ctx context.Context) {
}
l.mu.Unlock()
if !held {
return
return 0
}
releasing, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
defer cancel()
err := l.kv.Delete(releasing, Key, jetstream.LastRevision(revision))
err := l.giveBack(releasing, epoch, revision)
if err != nil && isWrongLast(err) {
// A renewal was in flight as this began and landed first: the key moved, and is still this
// instance's if it still names it at this epoch. Deleted at the revision it has now.
@@ -612,14 +659,29 @@ func (l *Lease) Release(ctx context.Context) {
var h Holder
if json.Unmarshal(entry.Value(), &h) == nil && h.Instance == l.o.Holder.Instance &&
(h.Epoch == epoch || h.Epoch == 0 && entry.Revision() == epoch) {
err = l.kv.Delete(releasing, Key, jetstream.LastRevision(entry.Revision()))
err = l.giveBack(releasing, epoch, entry.Revision())
}
}
}
if err != nil {
l.o.Say("the controller lease (epoch %d) could not be given back, so the next controller waits for "+
"it to expire: %v", epoch, err)
return
return 0
}
l.o.Say("gave the controller lease back (epoch %d)", epoch)
return epoch
}
// giveBack deletes the key at the revision this instance last wrote, as the bucket's own delete does — a
// marker with KV-Operation DEL, compare-and-set on the key's last revision — and names the epoch given back in
// the marker, so the next holder can tell this handover from any other delete (ReleasedHeader).
func (l *Lease) giveBack(ctx context.Context, epoch, revision uint64) error {
if l.js == nil {
return l.kv.Delete(ctx, Key, jetstream.LastRevision(revision))
}
msg := nats.NewMsg(l.subject())
msg.Header.Set("KV-Operation", "DEL")
msg.Header.Set(ReleasedHeader, strconv.FormatUint(epoch, 10))
_, err := l.js.PublishMsg(ctx, msg, jetstream.WithExpectLastSequencePerSubject(revision))
return err
}
+17 -3
View File
@@ -133,7 +133,12 @@ func TestTheSecondControllerWaitsAndTakesAHigherEpochOnHandover(t *testing.T) {
// a gives it back: b takes it at once, not after the key's age, at a higher epoch.
stop()
handedOver := time.Now()
a.Release(t.Context())
if given := a.Release(t.Context()); given != epochA {
t.Fatalf("a's release answers epoch %d given back, want %d", given, epochA)
}
if again := a.Release(t.Context()); again != 0 {
t.Fatalf("a second release gives epoch %d back again", again)
}
select {
case epochB := <-took:
if epochB <= epochA {
@@ -142,6 +147,9 @@ func TestTheSecondControllerWaitsAndTakesAHigherEpochOnHandover(t *testing.T) {
if waited := time.Since(handedOver); waited > testTTL {
t.Fatalf("b took the lease %s after a gave it back: it waited out the key's age", waited)
}
if given := b.GivenBack(); given != epochA {
t.Fatalf("b reads from the bus that epoch %d gave the key back, want a's %d (novox/hq issue 906)", given, epochA)
}
case <-time.After(2 * testTTL):
t.Fatal("b never took the lease a gave back")
}
@@ -182,6 +190,9 @@ func TestAHolderThatStopsRenewingStopsActingBeforeItIsTakenOver(t *testing.T) {
if epochB <= epochA {
t.Fatalf("b took epoch %d after a's %d", epochB, epochA)
}
if given := b.GivenBack(); given != 0 {
t.Fatalf("b reads a key that expired as given back by epoch %d (novox/hq issue 906)", given)
}
// And a, the moment it tries to renew, knows it lost: the key moved.
if err := a.Renew(t.Context()); err == nil {
t.Fatal("a renewed a lease b holds")
@@ -226,8 +237,11 @@ func TestARenewalRefusedIsALossAtOnce(t *testing.T) {
if _, err := a.Epoch(); !errors.Is(err, ErrNotHeld) {
t.Fatalf("a acts after its renewal was refused: %v", err)
}
// And giving back a lease it lost deletes nothing of the one who holds it now.
a.Release(t.Context())
// And giving back a lease it lost deletes nothing of the one who holds it now, and answers no epoch given
// back, so none is recorded released (novox/hq issue 906).
if given := a.Release(t.Context()); given != 0 {
t.Fatalf("a says it gave back epoch %d of a lease it lost", given)
}
if h, found, err := Current(t.Context(), kv); err != nil || !found || h.Instance != "intruder" {
t.Fatalf("after a gave back what it lost, the key holds %+v (%v, %v)", h, found, err)
}
+98 -36
View File
@@ -127,9 +127,13 @@ type CallLog struct {
// keeper keeps every call beyond this process, when Durably was given one; writes go through
// one goroutine, in order, so a call's last state is the one kept.
keeper CallKeeper
holder string
writes chan Call
keeper CallKeeper
holder string
// queued are the calls' states waiting for the keeper, one per call, in the order each call was first queued, under
// qmu (keep is called with mu held); woken is told when one is queued (callQueue).
qmu sync.Mutex
queued callQueue
woken chan struct{}
logger *log.Logger
lost int // writes the keeper could not take, said once each
keepErr error
@@ -165,8 +169,8 @@ const keepTries = 5
func (l *CallLog) Durably(ctx context.Context, keeper CallKeeper, holder string, logger *log.Logger) error {
l.mu.Lock()
l.keeper, l.holder, l.logger = keeper, holder, logger
if l.writes == nil {
l.writes = make(chan Call, 256)
if l.woken == nil {
l.woken = make(chan struct{}, 1)
go l.keepWrites()
}
l.mu.Unlock()
@@ -203,51 +207,109 @@ func orSomebody(s string) string {
return s
}
// callQueueBound is how many calls may wait for the keeper at once: far beyond a burst of reads (about 200 calls in one
// second on 2026-10-30), and a bound, so a keeper that takes nothing for long is said rather than grown into.
const callQueueBound = 8192
// callQueue holds the latest state of each call waiting for the keeper. **One state per call** (sweep 2026-10-30, issue
// 774): a call's later state replaces its earlier one still waiting, since only the last is kept anyway; a burst of
// read calls queued begin and finish each, and a channel of 256 dropped 30 of them in one second.
type callQueue struct {
order []string
byID map[string]Call
}
// put queues c, replacing a waiting state of the same call; false when the queue is full.
func (q *callQueue) put(c Call) bool {
if q.byID == nil {
q.byID = map[string]Call{}
}
if _, waiting := q.byID[c.ID]; waiting {
q.byID[c.ID] = c
return true
}
if len(q.order) >= callQueueBound {
return false
}
q.order = append(q.order, c.ID)
q.byID[c.ID] = c
return true
}
// take is every waiting state, in order, and empties the queue.
func (q *callQueue) take() []Call {
out := make([]Call, 0, len(q.order))
for _, id := range q.order {
out = append(out, q.byID[id])
}
q.order, q.byID = nil, nil
return out
}
// keep queues one call's state for the keeper. Never blocks a call: a queue that is full is a keeper
// that is not taking writes, and that is said rather than waited on.
func (l *CallLog) keep(c Call) {
if l.writes == nil {
if l.woken == nil {
return
}
l.qmu.Lock()
queued := l.queued.put(c)
if !queued {
l.lost++
}
lost := l.lost
l.qmu.Unlock()
if !queued {
if l.logger != nil {
l.logger.Printf("%s (%s.%s) is kept in memory only: %d calls wait for the bus to take their records "+
"(%d not kept)", c.ID, c.Seat, c.Verb, callQueueBound, lost)
}
return
}
select {
case l.writes <- c:
case l.woken <- struct{}{}:
default:
l.lost++
if l.logger != nil {
l.logger.Printf("%s (%s.%s) is kept in memory only: the bus is not taking calls' records (%d not kept)",
c.ID, c.Seat, c.Verb, l.lost)
}
}
}
func (l *CallLog) keepWrites() {
for c := range l.writes {
l.mu.Lock()
gate := l.epoch
l.mu.Unlock()
if gate != nil {
if _, err := gate(); err != nil {
if l.logger != nil {
l.logger.Printf("%s (%s.%s, %s) is not kept on the bus: %v — the controller holding the lease "+
"marks it abandoned", c.ID, c.Seat, c.Verb, c.State, err)
}
continue
for range l.woken {
l.qmu.Lock()
waiting := l.queued.take()
l.qmu.Unlock()
for _, c := range waiting {
l.keepOne(c)
}
}
}
// keepOne writes one call's state through the keeper, under the lease.
func (l *CallLog) keepOne(c Call) {
l.mu.Lock()
gate := l.epoch
l.mu.Unlock()
if gate != nil {
if _, err := gate(); err != nil {
if l.logger != nil {
l.logger.Printf("%s (%s.%s, %s) is not kept on the bus: %v — the controller holding the lease "+
"marks it abandoned", c.ID, c.Seat, c.Verb, c.State, err)
}
return
}
var err error
for try := 0; try < keepTries; try++ {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
err = l.keeper.Keep(ctx, c)
cancel()
if err == nil {
break
}
time.Sleep(time.Duration(try+1) * time.Second)
}
if err != nil && l.logger != nil {
l.logger.Printf("%s (%s.%s, %s) could not be kept on the bus after %d tries: %v — `calls` "+
"answers it from memory until this controller stops", c.ID, c.Seat, c.Verb, c.State, keepTries, err)
}
var err error
for try := 0; try < keepTries; try++ {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
err = l.keeper.Keep(ctx, c)
cancel()
if err == nil {
break
}
time.Sleep(time.Duration(try+1) * time.Second)
}
if err != nil && l.logger != nil {
l.logger.Printf("%s (%s.%s, %s) could not be kept on the bus after %d tries: %v — `calls` "+
"answers it from memory until this controller stops", c.ID, c.Seat, c.Verb, c.State, keepTries, err)
}
}
+32 -4
View File
@@ -254,8 +254,7 @@ func TestAHandoverRefusalIsMarkedRetryable(t *testing.T) {
// who may call the seat, and a token is the one-time right to become a machine of the mesh.
func TestAJoinTokenIsShownToItsCallerAndNotKept(t *testing.T) {
l, a := NewCallLog(), newAnswers(t)
writes := make(chan Call, 4)
l.writes = writes
l.woken = make(chan struct{}, 1) // queued for the keeper, and nothing takes them: read below
l.serveCall("mesh-controller", "token", json.RawMessage(`{"new":"laptop"}`), "_INBOX.x.1",
func(context.Context, json.RawMessage) (any, error) { return "token for laptop: s3cret-join", nil },
a.respond, nil)
@@ -266,8 +265,7 @@ func TestAJoinTokenIsShownToItsCallerAndNotKept(t *testing.T) {
if len(recent) != 1 || strings.Contains(string(recent[0].Answer), "s3cret-join") {
t.Fatalf("the token was kept in memory: %+v", recent)
}
close(writes)
for c := range writes {
for _, c := range l.queued.take() {
if strings.Contains(string(c.Answer), "s3cret-join") {
t.Fatalf("the token was sent to be kept on the bus: %s", c.Answer)
}
@@ -276,3 +274,33 @@ func TestAJoinTokenIsShownToItsCallerAndNotKept(t *testing.T) {
// A verb whose handler panics answers an error, and the controller serves the next call (review of
// novox/hq issue 327: read verbs are answered in the serving process now).
// Sweep 2026-10-30 (issue 774): a burst of calls never loses a record while the keeper is slow: each call waits once,
// with its latest state, and every call of a burst far over the old queue of 256 is kept with its last state.
func TestABurstOfCallsIsKeptWithEachCallsLastState(t *testing.T) {
var q callQueue
for i := range 2000 {
id := fmt.Sprintf("call-%d", i)
if !q.put(Call{ID: id, State: CallRunning}) || !q.put(Call{ID: id, State: CallAnswered}) {
t.Fatalf("%s was refused by a queue of %d", id, callQueueBound)
}
}
got := q.take()
if len(got) != 2000 {
t.Fatalf("%d calls wait, want 2000", len(got))
}
for i, c := range got {
if c.ID != fmt.Sprintf("call-%d", i) || c.State != CallAnswered {
t.Fatalf("call %d waits as %+v", i, c)
}
}
if len(q.take()) != 0 {
t.Fatal("taken calls wait again")
}
for i := range callQueueBound {
q.put(Call{ID: fmt.Sprint(i)})
}
if q.put(Call{ID: "one too many"}) {
t.Fatal("the queue grew past its bound")
}
}
+2 -3
View File
@@ -30,9 +30,8 @@ var Contracts = map[string]Contract{
KindBuilt: {Unordered: "a build's outcome on the retired build seat's subjects: dropped, nothing takes a build in " +
"from the bus any more (R4g, ADR 0501 §1); the subjects stay followed while the installer's carried user list " +
"names them"},
KindSourceMoved: {Ordered: "by the merge's commit against what was built from it: a merge already acted on " +
"or older than the last look is history, not a second plan (issues 250, 266)",
Tests: []string{"TestAMergeOlderThanTheLastLookIsHistory", "TestAMergeMatchesTheSourcesBuiltFromIt"}},
KindSourceMoved: {Unordered: "the forge's merge on EVENTS: no longer followed, dropped if one arrives; the forge " +
"bridge's stream FORGE is the record of the merges, read at a moved module's registration (sweep 2026-10-30)"},
KindToolsHeartbeat: {Unordered: "a word that the node tools are there, as a machine's heartbeat"},
KindHealth: {Ordered: "by the time the node-engine looked (`at`, on the machine's clock): a statement older " +
"than the one kept for that machine is refused, so an event arriving after a newer report does not undo it " +
+2 -4
View File
@@ -125,8 +125,7 @@ func askCLI(t *testing.T, conn *nats.Conn, node string, asked CLIAsked) (CLIAnsw
// bus, where `calls` answers anyone who may call the seat.
func TestAMeshCLIRecordKeepsNeitherItsLineNorItsAnswerOnTheBus(t *testing.T) {
l, a := NewCallLog(), newAnswers(t)
writes := make(chan Call, 4)
l.writes = writes
l.woken = make(chan struct{}, 1) // queued for the keeper, and nothing takes them: read below
asked, _ := json.Marshal(CLIAsked{Line: []string{"settings", "set", "x", `{"password":"s3cret"}`}, Account: "op"})
l.serveCall(CLISeat, "laptop", asked, "_INBOX.node.laptop.abcdefghijklmnopqrstuv",
func(context.Context, json.RawMessage) (any, error) {
@@ -134,8 +133,7 @@ func TestAMeshCLIRecordKeepsNeitherItsLineNorItsAnswerOnTheBus(t *testing.T) {
},
a.respond, nil)
_ = a.only()
close(writes)
for c := range writes {
for _, c := range l.queued.take() {
if carries(c.Args, "s3cret") || carries(c.Answer, "s3cret-join") {
t.Fatalf("the bus was sent %s / %s", c.Args, c.Answer)
}
+3
View File
@@ -52,6 +52,9 @@ type MoveSecretAnswer struct {
Secret string `json:"secret,omitempty"`
Generation int `json:"generation,omitempty"`
Unchanged bool `json:"unchanged,omitempty"`
// Superseded says the vault holds another value of the secret, which is the live one, and this machine's older value
// was not moved and is not asked again (mesh-host b9c24a7, sweep 2026-10-30).
Superseded bool `json:"superseded,omitempty"`
}
// MoveSecretWithin is how long the controller waits for one move: the engine asks the vault twice.
+3 -2
View File
@@ -30,8 +30,9 @@ func TestTheMovesAskAndAnswerKeepTheirFieldNames(t *testing.T) {
"provision sealed sealed_to" {
t.Fatalf("the ask's fields are %q", got)
}
answer := MoveSecretAnswer{Said: "s", Refused: "r", Secret: "own/a/b/c", Generation: 1, Unchanged: true}
if got := keys(answer); got != "generation refused said secret unchanged" {
answer := MoveSecretAnswer{Said: "s", Refused: "r", Secret: "own/a/b/c", Generation: 1, Unchanged: true,
Superseded: true}
if got := keys(answer); got != "generation refused said secret superseded unchanged" {
t.Fatalf("the answer's fields are %q", got)
}
if broker.AskMoveSecretSubject("laptop") != "mesh.node.laptop.ask.move-secret" ||
+6 -20
View File
@@ -62,7 +62,7 @@ func Nats(js *broker.JetStream) Inbound {
// whatever was asked for — and not at all when nothing was.
func (n *natsInbound) Also(kind string) error {
switch kind {
case KindModuleMoved, KindCatchUp, KindSourceMoved, KindProvisioner, KindPullUpdated, KindDecided,
case KindModuleMoved, KindCatchUp, KindProvisioner, KindPullUpdated, KindDecided,
KindDeliveryMerged:
n.follows[kind] = true
return nil
@@ -145,24 +145,6 @@ func (n *natsInbound) Receive(ctx context.Context, act func(context.Context, Con
defer func() { _ = followed.Unsubscribe() }()
}
// And the forge's merges, off a consumer of their own: no max-deliver and no reset, so a merge held for the
// plans waits for them however long (novox/hq ADR 0407 decision 3). Bound to the same channel: one loop.
if n.follows[KindSourceMoved] {
if events == nil {
events = make(chan *nats.Msg, Prefetch)
}
merges, err := standingBy(ctx, log.Default(), "EVENTS", func() (*nats.Subscription, error) {
return js.ChanSubscribe(MergedSubject, events, nats.Bind("EVENTS", broker.MergesConsumerName))
})
if err != nil {
return fmt.Errorf("subscribing to the forge's merges: %w", err)
}
if merges == nil {
return nil
}
defer func() { _ = merges.Unsubscribe() }()
}
// A connection that dropped is said, not discovered. A controller whose bus connection is gone
// is a mesh where nothing can be told anything.
gone := make(chan error, 1)
@@ -423,7 +405,11 @@ func (m *natsControl) Hold(after time.Duration) error {
if _, already := m.on.since[m.seq]; !already {
m.on.since[m.seq] = time.Now()
}
return m.msg.NakWithDelay(after)
if err := m.msg.NakWithDelay(after); err != nil {
return err
}
Loop.HeldUntil(time.Now().Add(after))
return nil
}
func (m *natsControl) forget() {
@@ -394,35 +394,3 @@ func TestNatsAReportThatIsNotNewsIsNotStated(t *testing.T) {
case <-time.After(3 * time.Second):
}
}
func TestNatsAMergeHeldMoreThanFiveTimesIsStillHandedOverAndRecorded(t *testing.T) {
first, most := MergeHeldFirst, MergeHeldMost
MergeHeldFirst, MergeHeldMost = 5*time.Millisecond, 40*time.Millisecond
t.Cleanup(func() { MergeHeldFirst, MergeHeldMost = first, most })
js := aBus(t)
up := &manyTimesBusy{busy: 8}
s := &Server{inbound: Nats(js), bus: OverNATS{Conn: js.Conn(), JS: js.Context()}, log: quiet()}
if err := s.Follows(up); err != nil {
t.Fatal(err)
}
ctx, stop := context.WithCancel(context.Background())
defer stop()
go func() { _ = s.Serve(ctx) }()
eventually(t, "the merges consumer being made", func() bool {
_, err := js.Context().ConsumerInfo("EVENTS", broker.MergesConsumerName)
return err == nil
})
body, _ := json.Marshal(SourceMoved{Owner: "novox", Repo: "one", Base: "main", Commit: "abcdef0123"})
if _, err := js.Context().Publish(MergedSubject, body); err != nil {
t.Fatal(err)
}
eventually(t, "the ninth hand-over recording the merge", func() bool { return up.called() == 9 })
eventually(t, "the merge being acknowledged", func() bool {
info, err := js.Context().ConsumerInfo("EVENTS", broker.MergesConsumerName)
return err == nil && info.NumAckPending == 0 && info.NumPending == 0
})
if info, err := js.Context().StreamInfo(broker.DeadLettersStream); err == nil && info.State.Msgs != 0 {
t.Fatalf("%d message(s) in dead letters: a merge was given up on", info.State.Msgs)
}
}
-44
View File
@@ -2,7 +2,6 @@ package link
import (
"context"
"errors"
"io"
"log"
"sync"
@@ -199,46 +198,3 @@ func (s slowly) Heard(context.Context, Report) (bool, error) { s.work(); return
type sameAgain struct{}
func (sameAgain) Heard(context.Context, Report) (bool, error) { return false, nil }
// manyTimesBusy is an upgrader whose plans are busy for its first merges, as another controller's hold is.
type manyTimesBusy struct {
toldAbout
busy int
calls int
}
func (u *manyTimesBusy) SourceMoved(context.Context, SourceMoved) error {
u.mu.Lock()
defer u.mu.Unlock()
u.calls++
if u.calls <= u.busy {
return errors.Join(ErrTryAgain, errors.New("another controller is working the plans"))
}
return nil
}
func (u *manyTimesBusy) called() int {
u.mu.Lock()
defer u.mu.Unlock()
return u.calls
}
// **A merge held more than five times is still handed over and recorded** (issue 973, novox/hq ADR 0407 decision
// 3): the merges have a durable consumer of their own with no max-deliver, so they never go to DEAD_LETTERS to wait
// for a person; the delay grows, and is capped.
// The delay a held merge waits grows from the first to the most and stays there.
func TestAHeldMergesDelayGrowsAndIsCapped(t *testing.T) {
var got []time.Duration
for _, held := range []time.Duration{0, 4 * time.Second, 10 * time.Second, 25 * time.Second, 70 * time.Second,
200 * time.Second, 20 * time.Minute} {
got = append(got, MergeRedeliverAfter(held))
}
want := []time.Duration{5 * time.Second, 5 * time.Second, 10 * time.Second, 25 * time.Second, 70 * time.Second,
200 * time.Second, 5 * time.Minute}
for i := range want {
if got[i] != want[i] {
t.Errorf("held %d: waits %v; want %v", i, got[i], want[i])
}
}
}
+3 -61
View File
@@ -49,8 +49,6 @@ type Upgrader interface {
// stop every upgrade behind it — except the store unreachable for the moment, which is asked
// again for a bounded time (novox/hq issue 083).
Upgraded(ctx context.Context, u Upgraded) error
// SourceMoved is a merge on the forge: build what that source produces, bases first.
SourceMoved(ctx context.Context, m SourceMoved) error
}
// Checker is what the controller does when the forge says a pull request's head moved: ask for it to be
@@ -105,9 +103,6 @@ const GiveUpAfter = 2 * time.Minute
// Follows says what to do about upgrades, and asks for them to be delivered.
func (s *Server) Follows(u Upgrader) error {
if err := s.inbound.Also(KindSourceMoved); err != nil {
return err
}
if err := s.inbound.Also(KindModuleMoved); err != nil {
return err
}
@@ -225,7 +220,9 @@ func (s *Server) act(ctx context.Context, m Control) {
case KindModuleMoved:
s.moved(ctx, m)
case KindSourceMoved:
s.sourceMoved(ctx, m)
// The forge's merges are no longer followed (sweep 2026-10-30): the forge bridge's stream is their record, read
// where a merge is needed (keptForgeMerge). One that arrives on an old consumer is let go of.
_ = m.Drop()
case KindCatchUp:
s.catchingUp(ctx, m)
case KindProvisioner:
@@ -521,61 +518,6 @@ func short(commit string) string {
return commit
}
// MergeHeldFirst is how long a merge the controller could not record yet waits before the bus hands it over
// again, and MergeHeldMost the longest it ever waits: the delay grows with how long the merge has been held, from
// the first to the most (novox/hq ADR 0407 decision 3). Variables so a test need not wait.
var (
MergeHeldFirst = 5 * time.Second
MergeHeldMost = 5 * time.Minute
)
// MergeRedeliverAfter is the delay for a merge held for the time given: that time, so the delay about doubles
// with each hold, between MergeHeldFirst and MergeHeldMost.
func MergeRedeliverAfter(heldFor time.Duration) time.Duration {
return min(max(heldFor, MergeHeldFirst), MergeHeldMost)
}
// sourceMoved acts on the forge's announcement of a merge. **A merge is never dropped** (issue 921, novox/hq ADR
// 0407 decision 3): one the controller could not record for now (the plans held by another controller, the store
// away: ErrTryAgain) is held and handed over again until it is recorded; the controller giving up on it would
// lose a change for good. Anything else is an answer and is taken: a build that fails is reported by the build
// itself, and re-delivering the merge would only re-fail it.
func (s *Server) sourceMoved(ctx context.Context, m Control) {
var moved SourceMoved
if err := json.Unmarshal(m.Body(), &moved); err != nil {
s.log.Printf("a merge announcement could not be read: %v", err)
_ = m.Took()
return
}
if moved.Commit == "" || moved.Repo == "" {
s.log.Printf("a merge announcement named no repository or no commit; ignored")
_ = m.Took()
return
}
err := s.upgrader.SourceMoved(ctx, moved)
// Left unsettled when the controller is ending, whatever error that gave (the plans' hold returns a bare
// context.Canceled): the bus hands it to whatever consumes next, and taking it would lose the merge.
if err != nil && ctx.Err() != nil {
return
}
if err != nil && errors.Is(err, ErrTryAgain) {
// Also left unsettled when the bus will not hold it.
delay := MergeRedeliverAfter(m.HeldFor())
if held := m.Hold(delay); held != nil {
s.log.Printf("%s/%s moved to %.8s and could not be held (%v): %v; left for the bus to hand over again",
moved.Owner, moved.Repo, moved.Commit, held, err)
return
}
s.log.Printf("%s/%s moved to %.8s and is held: %v; handed over again in %s", moved.Owner, moved.Repo,
moved.Commit, err, delay)
return
}
if err != nil {
s.log.Printf("%s/%s moved to %.8s and the mesh could not act on it: %v", moved.Owner, moved.Repo, moved.Commit, err)
}
_ = m.Took()
}
// pullUpdated asks for a pull request's merge check. Taken whatever happens: a check that could not be
// asked is said here, and the next push to the pull request asks again.
func (s *Server) pullUpdated(ctx context.Context, m Control) {
-60
View File
@@ -1,60 +0,0 @@
package link
import (
"context"
"encoding/json"
"errors"
"io"
"log"
"testing"
"time"
)
// heldControl is a merge announcement that records how it was settled.
type heldControl struct {
body []byte
took, held, drops int
}
func (c *heldControl) Kind() string { return "source-moved" }
func (c *heldControl) Body() []byte { return c.body }
func (c *heldControl) Subject() string { return "" }
func (c *heldControl) Redelivered() bool { return false }
func (c *heldControl) HeldFor() time.Duration { return 0 }
func (c *heldControl) Answer(context.Context, []byte) error { return nil }
func (c *heldControl) Took() error { c.took++; return nil }
func (c *heldControl) Hold(time.Duration) error { c.held++; return nil }
func (c *heldControl) Drop() error { c.drops++; return nil }
type sourceMover struct {
Upgrader
answer error
}
func (s sourceMover) SourceMoved(context.Context, SourceMoved) error { return s.answer }
// **A merge is acknowledged only after the controller recorded it** (novox/hq ADR 0407 decision 3, issue 1092): an
// answer of "not now" — the plan it belongs to could not be saved, the bus away, the lease lost — leaves the announcement
// unacknowledged, held to be handed over again; only an answer that is not "not now" settles it.
func TestAMergeTheControllerCouldNotRecordIsHeldAndNotAcknowledged(t *testing.T) {
body, err := json.Marshal(SourceMoved{Owner: "novox", Repo: "mesh-catalog", Base: "main", Commit: "c1"})
if err != nil {
t.Fatal(err)
}
s := &Server{log: log.New(io.Discard, "", 0)}
for name, tc := range map[string]struct {
answer error
took, held int
}{
"recorded": {nil, 1, 0},
"not now": {errors.Join(ErrTryAgain, errors.New("the plan is not saved")), 0, 1},
"an answer, not a wait": {errors.New("the mesh could not act on it"), 1, 0},
} {
s.upgrader = sourceMover{answer: tc.answer}
m := &heldControl{body: body}
s.sourceMoved(context.Background(), m)
if m.took != tc.took || m.held != tc.held || m.drops != 0 {
t.Errorf("%s: acknowledged %d times and held %d times, want %d and %d", name, m.took, m.held, tc.took, tc.held)
}
}
}
+1 -60
View File
@@ -17,8 +17,7 @@ func (r recordsWith) Built(context.Context, BuildResult) error { return r.err }
type upgradesWith struct{ err error }
func (u upgradesWith) Upgraded(context.Context, Upgraded) error { return u.err }
func (u upgradesWith) SourceMoved(context.Context, SourceMoved) error { return u.err }
func (u upgradesWith) Upgraded(context.Context, Upgraded) error { return u.err }
type replaysWith struct{ err error }
@@ -69,61 +68,3 @@ func TestACatchUpWaitsOutARestartingStore(t *testing.T) {
}
}
}
// A merge the controller could not record for now (the plans held by another controller, the store away) is held
// and handed over again, never taken: the bus keeps it until it is recorded (issue 921). Any other failure is an
// answer and is taken, as before.
func TestAMergeThatCouldNotBeRecordedNowIsHeldNotDropped(t *testing.T) {
merged := SourceMoved{Owner: "novox", Repo: "one", Base: "main", Commit: "abcdef0123"}
for _, c := range []struct {
what string
err error
want func(*settled) bool
}{
{"the plans held by another controller", errors.Join(ErrTryAgain, errors.New("another controller is working the plans")),
func(s *settled) bool { return s.unsettled() }},
{"the store away", errors.Join(ErrTryAgain, restarting), func(s *settled) bool { return s.unsettled() }},
{"cannot act", errors.New("no such module"), func(s *settled) bool { return s.acked && !s.nacked }},
{"kept", nil, func(s *settled) bool { return s.acked && !s.nacked }},
} {
s, in := serving()
s.upgrader = upgradesWith{err: c.err}
to := &settled{}
held := in.sends(t, to, KindSourceMoved, merged).(*fakeControl)
s.act(context.Background(), held)
if !c.want(to) {
t.Errorf("%s: a merge was settled as %+v", c.what, *to)
}
if c.err != nil && errors.Is(c.err, ErrTryAgain) && len(in.held) != 1 {
t.Errorf("%s: the merge was not held to be handed over again", c.what)
}
}
}
// A merge being held when the controller is replaced (its context ended) or when the bus will not hold it is left
// unsettled, never taken: the bus hands it to whatever consumes next (review of issue 921).
func TestAMergeIsLeftUnsettledWhenTheControllerEndsOrTheHoldFails(t *testing.T) {
merged := SourceMoved{Owner: "novox", Repo: "one", Base: "main", Commit: "abcdef0123"}
try := errors.Join(ErrTryAgain, errors.New("another controller is working the plans"))
// The plans' hold returns a bare context.Canceled when the controller is replaced, not ErrTryAgain.
s, in := serving()
s.upgrader = upgradesWith{err: context.Canceled}
to := &settled{}
ended, end := context.WithCancel(context.Background())
end()
s.act(ended, in.sends(t, to, KindSourceMoved, merged))
if !to.unsettled() {
t.Errorf("a merge was settled as %+v when the controller's context ended", *to)
}
s, in = serving()
s.upgrader = upgradesWith{err: try}
to = &settled{}
m := in.sends(t, to, KindSourceMoved, merged).(*fakeControl)
m.holdFails = errors.New("the bus will not hold it")
s.act(context.Background(), m)
if !to.unsettled() {
t.Errorf("a merge was settled as %+v when it could not be held", *to)
}
}
+20
View File
@@ -85,6 +85,26 @@ func nodeOfToolsAlive(subject string) (string, bool) {
type LoopActivity struct {
mu sync.Mutex
took time.Time
// due is when the latest message the loop held back with a delay is handed over again: until then the loop waits
// by its own choice, which is not deafness (sweep 2026-10-30: a merge held five minutes at a time read as a deaf
// controller every time its delay outran the deaf bound).
due time.Time
}
// HeldUntil records that the loop handed a message back to be delivered again at due.
func (l *LoopActivity) HeldUntil(due time.Time) {
l.mu.Lock()
defer l.mu.Unlock()
if due.After(l.due) {
l.due = due
}
}
// Due is when the latest message the loop held is handed over again; zero when it held none.
func (l *LoopActivity) Due() time.Time {
l.mu.Lock()
defer l.mu.Unlock()
return l.due
}
// Loop is this process's event loop.
+1
View File
@@ -82,6 +82,7 @@
"dead-letters",
"root-free",
"judge-directories",
"placement-export",
"data",
"build",
"artifacts",
+2 -2
View File
@@ -7,9 +7,9 @@ The copy carries the catalogue's private details: domains, first names in module
private addresses. That is acceptable while mesh-controller is private; if it ever goes public, this copy is
a second place to clean besides the catalogue itself.
mesh-catalog 3a1017f5594217092a654dd1e2d72226766bc16a modules/*/module.json, modules/*/files, modules/*/templates, predefined.json
mesh-catalog f3a57dcc56fb6c198095cafd11d0decff9f720c4 modules/*/module.json, modules/*/files, modules/*/templates, predefined.json
mesh-host 9d9e3a46c4f5943cfffb413d962ce48b75e2fd39 examples/foundation-first-node.lock
mesh-sdk 10b6e202320f38301c67a6822b0734e0b870ea30 conformance/events/module-event.json
mesh-sdk a37181261c24a15b8092c06e11cce63e0e5f8ade conformance/events/module-event.json
To move them, from this repository's root, with the two repositories checked out beside it:
@@ -0,0 +1 @@
${seat:mesh-artifact-store:reach}
@@ -1,26 +1,52 @@
{
"module": "artifact-store-tools",
"version": "1",
"claims": [
{
"name": "mesh-artifact-sweep",
"scope": "mesh",
"serves": [
"artifacts",
"held",
"collect",
"mirrors",
"images"
]
}
],
"requires": [
"artifact-store"
],
"invokes": [
"seat:mesh-controller.artifacts",
"seat:mesh-controller.collect",
"seat:mesh-hand-acts.hand-act",
"mesh-catalog.catalog_modules",
"mesh-catalog.catalog_module"
],
"reads": [
"mesh-delivery-assigner.wants",
"mesh-delivery-builder.builds",
"mesh-delivery-builder.build-log",
"mesh-delivery-indexer.manifests",
"node-engine.have",
"mesh-controller.lease"
],
"state": [
"findings"
],
"tools": [
"artifact_store_repositories",
"artifact_store_usage",
"artifact_store_references",
"artifact_store_collect"
],
"config": {
"older-store-names": {
"type": "string",
"description": "The names the artifact store had before, as host names or host:port separated by spaces, or none when it had no other. artifact_store_references reads every definition under these as well as under the store's current name, so an image a definition pins under one is said to be named by a definition (novox/hq issue 323).",
"pattern": "none|[A-Za-z0-9.:-]+( [A-Za-z0-9.:-]+)*",
"default": "none"
}
},
"config": {
"older-store-names": {
"type": "string",
"description": "The names the artifact store had before, as host names or host:port separated by spaces, or none when it had no other. artifact_store_references reads every definition under these as well as under the store's current name, so an image a definition pins under one is said to be named by a definition (novox/hq issue 323).",
"pattern": "none|[A-Za-z0-9.:-]+( [A-Za-z0-9.:-]+)*",
"default": "none"
}
},
"resources": [
{
"id": "state",
@@ -36,6 +62,14 @@
"mode": "0600",
"trusted": false,
"template": "files/store-names"
},
{
"id": "store-address",
"type": "file",
"path": "${dir:state}/store-address",
"mode": "0600",
"trusted": false,
"template": "files/store-address"
}
],
"build": {
@@ -52,7 +86,8 @@
],
"env": {
"MESH_ARTIFACT_STORE_CONTAINER": "mesh-registry",
"MESH_ARTIFACT_STORE_NAMES_FILE": "${dir:state}/store-names"
"MESH_ARTIFACT_STORE_NAMES_FILE": "${dir:state}/store-names",
"MESH_ARTIFACT_STORE_ADDRESS_FILE": "${dir:state}/store-address"
}
},
{
@@ -8,7 +8,10 @@
"provides": [
{
"name": "rootless-image-build",
"reach": "machine"
"reach": "machine",
"credential": {
"form": "none"
}
}
],
"resources": [
@@ -7,7 +7,9 @@
"disk_filesystems",
"disk_folders",
"disk_devices",
"cpu_top"
"cpu_top",
"disk_readings",
"disk_measure"
],
"build": {
"artifacts": [
@@ -23,5 +25,16 @@
]
}
]
}
},
"state": [
"readings"
],
"reads": [
"mesh-config.resolved",
"mesh-delivery-assigner.placement",
"mesh-delivery-builder.builds",
"mesh-delivery-indexer.manifests",
"mesh-delivery-indexer.modules",
"node-engine.have"
]
}
@@ -4,7 +4,10 @@
"provides": [
{
"name": "artifact-store",
"scope": "mesh"
"scope": "mesh",
"credential": {
"form": "none"
}
}
],
"claims": [
@@ -1,14 +1,15 @@
# Written by the mesh (module dnsmasq, novox/hq issue 348). Replaced on every push.
#
# The resolver binds its addresses once, at start (bind-interfaces), so it starts only once this
# machine's private address is there: it waits up to 80 seconds for it, and a start that found it
# missing is tried again every ten seconds rather than left failed.
# machine's private address is there: a start that finds it missing fails after at most two seconds and
# is tried again every ten seconds (Restart=on-failure, no start limit) until the tunnel has brought it.
# It does not order itself After= or Wants= network-online.target, and its wait is short: the packaged
# unit is Before= network-online.target and nss-lookup.target, and the tunnel is ordered after them, so
# a long wait would hold the tunnel back and a drop-in ordering would be a cycle (issue 1126).
[Unit]
After=network-online.target
Wants=network-online.target
StartLimitIntervalSec=0
[Service]
ExecStartPre=/usr/bin/sh -c 'for i in $(seq 1 80); do ip -4 -o addr show to ${machine:address} | grep -q . && exit 0; sleep 1; done; echo "${machine:address} is not on this machine" >&2; exit 1'
ExecStartPre=/usr/bin/sh -c 'for i in $(seq 1 2); do ip -4 -o addr show to ${machine:address} | grep -q . && exit 0; sleep 1; done; echo "${machine:address} is not on this machine" >&2; exit 1'
Restart=on-failure
RestartSec=10s
@@ -9,7 +9,10 @@
{
"name": "wildcard-resolution",
"scope": "mesh",
"identity": false
"identity": false,
"credential": {
"form": "none"
}
}
],
"requires": [
@@ -7,7 +7,7 @@
"privileged"
],
"invokes": [
"seat:mesh-controller.images"
"seat:mesh-artifact-sweep.images"
],
"claims": [
{
@@ -40,6 +40,7 @@
],
"emits": [
"repo.created",
"repo.writable",
"issue.opened",
"pull.merged",
"pull.updated",
@@ -181,6 +182,12 @@
"trusted": true,
"default": {}
},
"check-gates-merge": {
"type": "boolean",
"description": "Whether gitea_merge_pull_request judges mesh/check before it merges. True (the default): a head merges only on a passing mesh/check judged against the trunk's head (ADR 0447, ADR 0534 §4). False (the operator, 2026-10-25 22:56: \"run the check locally only, but not as a gate, it loses us too much time\"): a head the operator's Approve names in approved_heads merges whatever its mesh/check says (pending, failed, missing or judged on an older trunk); a head that moved, a branch no rule guards and a merge git refuses are still refused. Set together with mesh-delivery-gate's check-gates-merge.",
"trusted": true,
"default": true
},
"main-mergers": {
"type": "list",
"description": "The forge logins that may merge into main beside the tools' account, on every repository of the organisation (ADR 0534 §2): the operator's own, so the operator can always merge by hand at the forge. The tools' account is always on main's merge whitelist; no agent's account belongs here.",
@@ -259,7 +266,8 @@
"env": {
"DB_TYPE": "postgres",
"USER_UID": "1000",
"USER_GID": "1000"
"USER_GID": "1000",
"SSH_MAX_STARTUPS": "100:30:200"
},
"env-file": [
"${dir:state}/server.env"
@@ -316,7 +324,8 @@
"takes-config": [
"url",
"keep-main-rules",
"main-mergers"
"main-mergers",
"check-gates-merge"
]
},
{
@@ -391,6 +400,7 @@
"provides": [
{
"name": "npm-package-registry",
"credential": {"form": "pair"},
"scope": "mesh",
"identity": {
"max": 40,
@@ -399,6 +409,7 @@
},
{
"name": "git",
"credential": {"form": "none"},
"scope": "mesh"
},
{
@@ -411,6 +422,7 @@
},
{
"name": "forge-account",
"credential": {"form": "none"},
"scope": "mesh"
}
],
@@ -537,6 +549,7 @@
"MESH_GITEA_ADMIN_PASSWORD_FILE": "${dir:state}/admin.secret",
"MESH_GITEA_ORIGINS_FILE": "${dir:mesh-state}/origins",
"MESH_GITEA_SSH_PORT": "222",
"MESH_GITEA_HOST_KEY_FILE": "${dir:data}/ssh/ssh_host_ed25519_key.pub",
"MESH_FORGE_HOST_FILE": "/etc/gitea/forge-host"
}
}
@@ -554,11 +567,14 @@
"announced"
],
"reads": [
"mesh-controller.assignments",
"mesh-provisioner.plans",
"mesh-provisioner.provisions",
"ssh-client.keys",
"mesh-build-go.keys",
"mesh-build-node.keys",
"mesh-build-image.keys",
"mesh-delivery-checker.owed"
"mesh-delivery-checker.owed",
"mesh-delivery-indexer.modules",
"mesh-delivery-gate.source"
]
}
@@ -7,7 +7,10 @@
"provides": [
{
"name": "go-toolchain",
"reach": "machine"
"reach": "machine",
"credential": {
"form": "none"
}
}
],
"resources": [
@@ -7,6 +7,9 @@
"scope": "mesh",
"identity": {
"in": "an InfluxDB v1 authorization"
},
"credential": {
"form": "pair"
}
}
],
@@ -0,0 +1,4 @@
# Written by the mesh's kernel module (systemd-modules-load reads it at boot).
# tun: buildah's network for a RUN line (pasta) and every other user-space tunnel open /dev/net/tun.
# Loaded at boot, so an upgrade that removes the running kernel's modules cannot keep it out later.
tun
@@ -100,6 +100,25 @@
"id": "linux-lts-headers",
"type": "package",
"package": "linux-lts-headers"
},
{
"id": "modules-load",
"type": "file",
"path": "/etc/modules-load.d/mesh-kernel.conf",
"mode": "0644",
"template": "files/modules-load.conf"
},
{
"id": "modules-keep",
"type": "package",
"package": "kernel-modules-hook"
},
{
"id": "modules-cleanup",
"type": "service",
"unit": "linux-modules-cleanup.service",
"state": "stopped",
"boot": "enabled"
}
],
"build": {
@@ -9,6 +9,7 @@
"provides": [
{
"name": "oidc-client",
"credential": {"form": "pair"},
"scope": "mesh",
"identity": {
"max": 255,
@@ -240,6 +241,7 @@
"bound"
],
"reads": [
"mesh-controller.assignments"
"mesh-provisioner.plans",
"mesh-provisioner.provisions"
]
}
@@ -665,7 +665,14 @@
"path": "${dir:mesh-state}/config.json",
"mode": "0600",
"template": "files/config.json",
"takes-config": ["domain", "proxy-address", "sitename", "website", "url", "container"],
"takes-config": [
"domain",
"proxy-address",
"sitename",
"website",
"url",
"container"
],
"merge": "json"
},
{
@@ -758,6 +765,9 @@
"identity": {
"max": 64,
"in": "a mailbox's local part"
},
"credential": {
"form": "pair"
}
}
],
@@ -1,3 +1,5 @@
{
"capacity": "shanks=3,ace=2,g14=0,novox=0"
"capacity": 0,
"cores-per-step": 6,
"build-hours": ""
}
@@ -10,7 +10,7 @@
"state": [
"toolchains",
"keys",
"steps"
{"name": "steps", "ttl-seconds": 172800}, "pauses"
],
"claims": [
{
@@ -20,14 +20,27 @@
}
],
"reads": [
"mesh-delivery-checker.kept"
"mesh-delivery-checker.kept",
"mesh-delivery-indexer.manifests"
],
"config": {
"capacity": {
"type": "number",
"description": "How many asks of the Go build seat this module runs at once on this node (novox/hq ADR 0525 §5); 0 pulls nothing. It is set per node in the mesh's configuration; a node with no value builds nothing.",
"minimum": 0,
"default": 0
},
"build-hours": {
"type": "string",
"description": "How many asks of the Go build seat this module runs at once on each node, as node=asks pairs separated by commas, such as shanks=3,ace=2 (novox/hq ADR 0525 §5); a node set to 0 pulls nothing, a node not named runs one at a time. The control-node and the laptop build nothing.",
"pattern": "|[a-z0-9_-]+=[0-9]+(,[a-z0-9_-]+=[0-9]+)*",
"default": "shanks=3,ace=2,g14=0,novox=0"
"description": "When this node may take this module's build asks: one window HH:MM-HH:MM in the node's local time (an end before its start crosses midnight, 18:00-06:00). Outside it the node takes no new step and finishes the ones it runs; an empty value builds at any hour. It is set per node in the mesh's configuration.",
"pattern": "|([01][0-9]|2[0-3]):[0-5][0-9]-([01][0-9]|2[0-3]):[0-5][0-9]",
"default": ""
},
"cores-per-step": {
"type": "number",
"description": "How many processors one Go step of this module needs (a full check of a large repository takes what it is given: three at about 4 cores each ran past the 30-minute step bound on 2026-10-26). The capacity above is an upper bound: on each node the module runs at most floor(the cores of its build slice's CPU quota / this) steps at once, the quota read live from its cgroup (issue 1806), so a lowered build CPU share never starves its steps. Above 0.",
"minimum": 0.1,
"default": 6
}
},
"tools": [
@@ -38,6 +51,7 @@
"build_go_release",
"build_go_step",
"build_go_steps",
"build_go_step_end",
"build_go_toolchains",
"build_go_status",
"build_go_health"
@@ -105,7 +119,7 @@
"path": "${dir:state}/settings.json",
"mode": "0600",
"owner": "mesh-build-go",
"takes-config": ["capacity"],
"takes-config": ["capacity", "cores-per-step", "build-hours"],
"merge": "json",
"template": "files/settings.json"
},
@@ -1,3 +1,5 @@
{
"capacity": "shanks=3,ace=2,g14=0,novox=0"
"capacity": 0,
"cores-per-step": 2,
"build-hours": ""
}
@@ -9,7 +9,7 @@
],
"state": [
"keys",
"steps"
{"name": "steps", "ttl-seconds": 172800}, "pauses"
],
"claims": [
{
@@ -20,10 +20,22 @@
],
"config": {
"capacity": {
"type": "number",
"description": "How many asks of the image build seat this module runs at once on this node (novox/hq ADR 0525 §5); 0 pulls nothing. It is set per node in the mesh's configuration; a node with no value builds nothing.",
"minimum": 0,
"default": 0
},
"build-hours": {
"type": "string",
"description": "How many asks of the image build seat this module runs at once on each node, as node=asks pairs separated by commas, such as shanks=3,ace=2 (novox/hq ADR 0525 §5); a node set to 0 pulls nothing, a node not named runs one at a time. The control-node and the laptop build nothing.",
"pattern": "|[a-z0-9_-]+=[0-9]+(,[a-z0-9_-]+=[0-9]+)*",
"default": "shanks=3,ace=2,g14=0,novox=0"
"description": "When this node may take this module's build asks: one window HH:MM-HH:MM in the node's local time (an end before its start crosses midnight, 18:00-06:00). Outside it the node takes no new step and finishes the ones it runs; an empty value builds at any hour. It is set per node in the mesh's configuration.",
"pattern": "|([01][0-9]|2[0-3]):[0-5][0-9]-([01][0-9]|2[0-3]):[0-5][0-9]",
"default": ""
},
"cores-per-step": {
"type": "number",
"description": "How many processors one image step of this module needs (buildah, whose RUN lines are mostly fetches). The capacity above is an upper bound: on each node the module runs at most floor(the cores of its build slice's CPU quota / this) steps at once, the quota read live from its cgroup (issue 1806), so a lowered build CPU share never starves its steps. Above 0.",
"minimum": 0.1,
"default": 2
}
},
"tools": [
@@ -33,6 +45,7 @@
"build_image_release",
"build_image_step",
"build_image_steps",
"build_image_step_end",
"build_image_status",
"build_image_health"
],
@@ -77,7 +90,7 @@
"path": "${dir:state}/settings.json",
"mode": "0600",
"owner": "mesh-build-image",
"takes-config": ["capacity"],
"takes-config": ["capacity", "cores-per-step", "build-hours"],
"merge": "json",
"template": "files/settings.json"
},
@@ -1,3 +1,5 @@
{
"capacity": "shanks=3,ace=2,g14=0,novox=0"
"capacity": 0,
"cores-per-step": 2,
"build-hours": ""
}
@@ -10,7 +10,7 @@
"state": [
"toolchains",
"keys",
"steps"
{"name": "steps", "ttl-seconds": 172800}, "pauses"
],
"claims": [
{
@@ -21,10 +21,22 @@
],
"config": {
"capacity": {
"type": "number",
"description": "How many asks of the Node build seat this module runs at once on this node (novox/hq ADR 0525 §5); 0 pulls nothing. It is set per node in the mesh's configuration; a node with no value builds nothing.",
"minimum": 0,
"default": 0
},
"build-hours": {
"type": "string",
"description": "How many asks of the Node build seat this module runs at once on each node, as node=asks pairs separated by commas, such as shanks=3,ace=2 (novox/hq ADR 0525 §5); a node set to 0 pulls nothing, a node not named runs one at a time. The control-node and the laptop build nothing.",
"pattern": "|[a-z0-9_-]+=[0-9]+(,[a-z0-9_-]+=[0-9]+)*",
"default": "shanks=3,ace=2,g14=0,novox=0"
"description": "When this node may take this module's build asks: one window HH:MM-HH:MM in the node's local time (an end before its start crosses midnight, 18:00-06:00). Outside it the node takes no new step and finishes the ones it runs; an empty value builds at any hour. It is set per node in the mesh's configuration.",
"pattern": "|([01][0-9]|2[0-3]):[0-5][0-9]-([01][0-9]|2[0-3]):[0-5][0-9]",
"default": ""
},
"cores-per-step": {
"type": "number",
"description": "How many processors one Node step of this module needs (tsc, esbuild and a test runner; a median step takes 15 s). The capacity above is an upper bound: on each node the module runs at most floor(the cores of its build slice's CPU quota / this) steps at once, the quota read live from its cgroup (issue 1806), so a lowered build CPU share never starves its steps. Above 0.",
"minimum": 0.1,
"default": 2
}
},
"tools": [
@@ -34,6 +46,7 @@
"build_node_release",
"build_node_step",
"build_node_steps",
"build_node_step_end",
"build_node_toolchains",
"build_node_status",
"build_node_health"
@@ -99,7 +112,7 @@
"path": "${dir:state}/settings.json",
"mode": "0600",
"owner": "mesh-build-node",
"takes-config": ["capacity"],
"takes-config": ["capacity", "cores-per-step", "build-hours"],
"merge": "json",
"template": "files/settings.json"
},
@@ -25,9 +25,7 @@
"secrets": {
"postgres-database": "${dir:state}/database.secret"
},
"consumes": [
"mesh-controller.built-before"
],
"consumes": [],
"emits": [
"registered",
"upgraded",
@@ -92,5 +90,12 @@
"from": "files"
}
]
}
},
"invokes": [
"mesh-delivery-builder.builder_builds_since"
],
"reads": [
"mesh-delivery-builder.builds",
"mesh-delivery-indexer.manifests"
]
}
@@ -0,0 +1,87 @@
{
"module": "mesh-conditions",
"version": "1",
"runs-as": "mesh-conditions",
"claims": [
{
"name": "mesh-conditions",
"scope": "mesh",
"serves": [
"conditions",
"silence",
"compare",
"take-over",
"hand-back",
"resync"
]
}
],
"uses": [
"operator-channel"
],
"reads": [
"mesh-delivery-watchman.findings",
"mesh-self-check.findings",
"mesh-healers.findings",
"mesh-delivery-assigner.findings",
"messenger.findings",
"mesh-provisioner.findings",
"artifact-store-tools.findings",
"mesh-healers.attempts",
"mesh-controller.conditions",
"mesh-controller.lease",
"mesh-delivery-assigner.wants",
"node-engine.have",
"mesh-delivery-assigner.placement"
],
"state": [
"conditions",
{
"name": "history",
"ttl-seconds": 7776000
},
"takeover",
"held",
{
"name": "bounds",
"key-ttl": true
},
"asked"
],
"own-secrets": {
"broker": "${dir:state}/broker"
},
"secrets-owner": "mesh-conditions",
"resources": [
{
"id": "account",
"type": "user",
"name": "mesh-conditions",
"shell": "/usr/bin/nologin",
"home": "/var/lib/mesh-conditions"
},
{
"id": "state",
"type": "directory",
"mode": "0700",
"owner": "mesh-conditions",
"place": ".",
"unknown": "program"
}
],
"build": {
"artifacts": [
{
"name": "tools",
"kind": "bundle",
"language": "go",
"system": "arch",
"from": "cmd/mesh-conditions",
"binary": "mesh-conditions",
"loads": [
"mesh-conditions"
]
}
]
}
}

Some files were not shown because too many files have changed in this diff Show More