Mail and bus hardening: restricted streams, recorded moves that restart nothing, desk prompts, abandoned calls (issues 727, 728, 785, 789, 711) #344

Merged
mesh-tools merged 17 commits from feat/mail-hardening into main 2026-10-15 14:34:25 +00:00
16 changed files with 1003 additions and 8 deletions
Showing only changes of commit 0d1bb035cb - Show all commits
+3
View File
@@ -820,6 +820,9 @@ type answers struct {
sources map[string]inventory.Source
// waiting is every machine not running what the mesh would send it.
waiting []inventory.Machine
// recordedWaits is, per machine, every move of a recorded module that waits there for a person (novox/hq ADR
// 0377 §6): the machine is behind for these, and `push <node>` sends them.
recordedWaits map[string][]string
// reported is every machine's last word beside when it was last sent a declaration — the
// pair that answers "has it caught up", which waiting alone cannot (the sent digest is
// recorded at send, not at apply).
+5
View File
@@ -73,6 +73,11 @@ func changePlanOf(repository, base, head string, r mergeReach, entries []invento
"(issue 280)", busUpgradeNeeded, name, busUpgradeVerb, strings.Join(e.On, ", ")))
case name == "nats" || catalogue.ProvidesBus(e.Manifest):
p.Steps = append(p.Steps, "a planned bus step: the bus is upgraded by `bus upgrade`, never by an ordinary send")
case waits && len(e.On) > 0 && (u.From == catalogue.FromModule || u.From == catalogue.FromData):
// Judged per machine by the walk, against the build each was last sent (novox/hq ADR 0377).
p.Steps = append(p.Steps, fmt.Sprintf("%s waits for a person: its policy records (%s) — unless on a machine "+
"its move recreates none of its containers and moves none of its data, which the walk judges and then sends",
name, orNone(u.From)))
case waits && len(e.On) > 0:
p.Steps = append(p.Steps, fmt.Sprintf("%s waits for a person: its policy records (%s)", name,
orNone(u.From)))
+2 -1
View File
@@ -642,7 +642,8 @@ func planStale(ctx context.Context, inv *inventory.Inventory, p inventory.Plan)
if err != nil {
return "", "", err
}
if u.RollOut {
// A recorded module whose move goes out on a machine is still the walk's to send (novox/hq ADR 0377).
if u.RollOut || len(wentOut(s.Recorded)) > 0 {
return "", "", nil
}
}
+17 -1
View File
@@ -146,7 +146,23 @@ func heldMachines(ctx context.Context, open *stores, names []string) (map[string
same[m] = current[m].Commit
}
}
why := heldBack(node, modules, same, known, current, plans)
// A move under a module's own `record` that recreates nothing and moves no data goes as a module that rolls
// out goes (novox/hq ADR 0377): a walk's to send, not a person's.
here, copied := current, false
for m, was := range same {
if now, held := current[m]; held && now.OwnRecord() && !f.identical(m, was, now.Commit) && f.rolls(m, was) {
if !copied {
copied = true
here = make(map[string]inventory.CurrentBuild, len(current))
for k, v := range current {
here[k] = v
}
}
now.RollOut = true
here[m] = now
}
}
why := heldBack(node, modules, same, known, here, plans)
// And a build no gate has seen is not carried by a send that does not judge it (ADR 0236).
for _, mv := range f.moves(node, modules, same, known, false) {
if planStillToSend(plans, mv.Module, node) == "" {
+6
View File
@@ -158,6 +158,12 @@ func recordedKept(ctx context.Context, open *stores, node string) (map[string]st
if f.identical(m, was, now.Commit) {
continue
}
// **Under its own `record`, a move that recreates nothing and moves no data is not kept** (novox/hq ADR 0377
// §1): it goes as a module that rolls out goes. A person's `record`, the bus, and a move that cannot be
// judged are kept as before (§4, §5).
if !every && f.rolls(m, was) {
continue
}
out[m] = was
}
// **And the controller and the node-engine at the build the machine was last sent, over an older record**
+213
View File
@@ -0,0 +1,213 @@
package main
import (
"context"
"fmt"
"github.com/novox/mesh-controller/internal/catalogue"
"github.com/novox/mesh-controller/internal/inventory"
)
// Under a module's own `record`, a move that recreates no container and moves no data rolls out (novox/hq ADR
// 0377, changing ADR 0242 §1 and ADR 0360 §1).
//
// A module records because its manifest says so, because it keeps irreplaceable data, because a person set it, or
// because it provides the bus. The first two hold for a reason the manifest shows — a change that recreates mail's
// containers interrupts everyone's mail, a new database image may upgrade its databases in place — and that hold
// also held the moves that interrupt nothing: on 2026-10-15 the database server moved with only its tools bundle
// changed, its walk said it "recreates none of its containers", and ace stayed behind until a person pushed.
//
// So **a move under a module's own `record` is judged per machine, against the build that machine was last sent**
// (§2): when it recreates none of the module's containers and moves none of its data (catalogue.Holds, §3) it
// goes as a module that rolls out goes — a walk sends it, and every other send treats it as a move waiting for its
// walk — and otherwise it is kept at the build the machine runs, as before. The module's policy stays `record`. A
// move that cannot be judged waits (§4); a person's `record` and the bus hold every move (§5); and each walk says
// per machine which way a recorded module went (§6).
// recordedRollsWhy is what a walk says of a recorded module's move that went out (ADR 0377 §6).
const recordedRollsWhy = "went out: it recreates none of its containers and moves none of its data (novox/hq ADR 0377)"
// recordVerdict is one machine's move of a recorded module: whether it rolls out, and what the walk says of it.
type recordVerdict struct {
rolls bool
// said is what the walk says: recordedRollsWhy, or "waits: " and why — the recreate words, the data change
// named, the person's or the bus's hold, or that it could not be judged.
said string
}
// recordedMove judges a module's move on a machine from the build it was last sent (`was`) to the one it would be
// sent now. A module that rolls out rolls; one whose policy is a person's or the bus's waits, said; one under its
// own `record` rolls when the move recreates nothing and moves no data, and waits when it does or cannot be judged.
func (f moveFacts) recordedMove(module, was string) recordVerdict {
now := f.current[module]
switch {
case now.RollOut:
return recordVerdict{rolls: true}
case now.From == catalogue.FromPerson:
return recordVerdict{said: "waits: a person set it to record, and holds every move of it until they push (novox/hq ADR 0377 §5)"}
case now.From == catalogue.FromBus:
return recordVerdict{said: "waits: it provides the bus, which moves only by the bus step a person starts (novox/hq ADR 0377 §5)"}
case !now.OwnRecord():
return recordVerdict{said: "waits: its policy records"}
case f.judge == nil:
return recordVerdict{said: "waits: it could not be judged — the build records were not read (novox/hq ADR 0377 §4)"}
}
return f.judge(module, was, now.Commit)
}
// rolls is whether a module's move on a machine from `was` goes as a move that rolls out (ADR 0377 §1).
func (f moveFacts) rolls(module, was string) bool {
return f.recordedMove(module, was).rolls
}
// recordJudge is the judging of moves under a module's own `record`, reading the build records once per move and
// the catalogue once (ADR 0377 §2–§4).
func recordJudge(ctx context.Context, inv *inventory.Inventory) func(module, was, now string) recordVerdict {
judged := map[string]recordVerdict{}
var shelf map[string]catalogue.Manifest
var shelfErr error
shelfRead := false
return func(module, was, now string) recordVerdict {
key := module + "\x00" + was + "\x00" + now
if v, seen := judged[key]; seen {
return v
}
v := func() recordVerdict {
cannot := func(why string) recordVerdict {
return recordVerdict{said: "waits: it could not be judged — " + why + " (novox/hq ADR 0377 §4)"}
}
if was == "" {
return cannot("what the machine was last sent of it is not known")
}
from, found, err := inv.ManifestAt(ctx, module, was)
if err != nil {
return cannot(fmt.Sprintf("its build %s, which the machine was last sent, cannot be read: %v", short(was), err))
}
if !found {
return cannot(fmt.Sprintf("its build %s, which the machine was last sent, is not in the build records", short(was)))
}
if !shelfRead {
shelf, shelfErr = inv.Catalogue(ctx)
shelfRead = true
}
if shelfErr != nil {
return cannot(fmt.Sprintf("the build it would be sent cannot be read: %v", shelfErr))
}
to, held := shelf[module]
if !held {
return cannot("the build it would be sent is not in the catalogue")
}
if why := catalogue.Holds(from, to); why != "" {
return recordVerdict{said: "waits: " + why}
}
return recordVerdict{rolls: true, said: recordedRollsWhy}
}()
judged[key] = v
return v
}
}
// judgeRecordedInWalk judges, for a walk carrying a module whose policy records, each machine running it whose
// build would move (ADR 0377 §2), keeps on the plan which way each went (§6), says it on the controller's log, and
// answers the machines its move goes out to. Judged once per walk: a walk read again answers what it kept.
func judgeRecordedInWalk(ctx context.Context, inv *inventory.Inventory, p *inventory.Plan, module string,
state *inventory.PlanModule) ([]string, error) {
if state.Recorded != nil {
return wentOut(state.Recorded), nil
}
running, err := inv.Running(ctx, module)
if err != nil {
return nil, err
}
f, err := readMoveFacts(ctx, inv)
if err != nil {
return nil, err
}
now := f.current[module]
said := map[string]string{}
for _, n := range running {
sent, known, err := inv.SentBuilds(ctx, n)
if err != nil {
return nil, err
}
was, carried := sent[module]
if known && carried && f.identical(module, was, now.Commit) {
continue // no move there
}
v := f.recordedMove(module, was)
if !known {
v = recordVerdict{said: "waits: it could not be judged — what " + n + " was last sent is not known (novox/hq ADR 0377 §4)"}
}
said[n] = v.said
fmt.Printf("%s: %s on %s %s\n", p.ID, module, n, v.said)
}
if len(said) > 0 {
state.Recorded = said
}
return wentOut(said), nil
}
// wentOut is the machines a recorded module's move went out to, sorted.
func wentOut(recorded map[string]string) []string {
var out []string
for _, n := range sortedKeysOf(recorded) {
if recorded[n] == recordedRollsWhy {
out = append(out, n)
}
}
return out
}
// recordedSaid is, for a walk's note at its end, which way each recorded module it carried went on each machine
// (ADR 0377 §6).
func recordedSaid(p inventory.Plan) []string {
var out []string
for _, m := range sortedKeysOf(planModuleNames(p)) {
s := p.Modules[m]
for _, n := range sortedKeysOf(s.Recorded) {
out = append(out, fmt.Sprintf("%s on %s %s", m, n, s.Recorded[n]))
}
}
return out
}
func planModuleNames(p inventory.Plan) map[string]string {
out := map[string]string{}
for m, s := range p.Modules {
if s != nil && len(s.Recorded) > 0 {
out[m] = m
}
}
return out
}
// recordedWaiting is, on one machine, every move of a recorded module that waits for a person, each said with why
// (ADR 0377 §6): what `status` names the machine behind for, with `push <node>` as the remedy. A move under a
// module's own `record` that recreates nothing and moves no data is not among them: it goes with its walk.
func recordedWaiting(ctx context.Context, inv *inventory.Inventory, f moveFacts, node string) ([]string, error) {
sent, known, err := inv.SentBuilds(ctx, node)
if err != nil || !known {
return nil, err
}
var out []string
for _, m := range sortedKeysOf(sent) {
was := sent[m]
now, held := f.current[m]
if !held || now.RollOut || now.Commit == "" || was == "" || f.identical(m, was, now.Commit) {
continue
}
if v := f.recordedMove(m, was); !v.rolls {
out = append(out, fmt.Sprintf("%s %s → %s %s", m, short(was), short(now.Commit), v.said))
}
}
return out, nil
}
// stringsKeys is a map of lists as a map of its keys, for sortedKeysOf.
func stringsKeys(m map[string][]string) map[string]string {
out := map[string]string{}
for k := range m {
out[k] = k
}
return out
}
+361
View File
@@ -0,0 +1,361 @@
package main
import (
"context"
"encoding/json"
"fmt"
"slices"
"strings"
"testing"
"time"
"github.com/novox/mesh-controller/internal/catalogue"
"github.com/novox/mesh-controller/internal/inventory"
)
// Under a module's own `record`, a move that recreates no container and moves no data rolls out (novox/hq ADR
// 0377). Seen 2026-10-15: the database server moved with only its tools bundle changed, its walk said it
// "recreates none of its containers", and ace stayed behind until a person pushed.
// aMailu is a mail module as the catalogue declares it, small: it records because its manifest says so, keeps its
// mailboxes as data, mounts its environment file, and serves its tools from a bundle.
func aMailu(policy bool) catalogue.Manifest {
raw := `{"module":"mailu","version":"1","tools":["mailu_status"],
"data":{"own":[{"id":"mail","path":"${dir:data-mail}","class":"valuable","why":"every mailbox"}]},
"bundles":[{"name":"tools","source":"store/mailu-tools@sha256:aaa","digest":"sha256:aaa"}],
"resources":[
{"id":"state","type":"directory"},
{"id":"mesh-state","type":"directory"},
{"id":"data-mail","type":"directory"},
{"id":"config-env","type":"file","path":"${dir:state}/mailu.env","content":"A=1"},
{"id":"runtime-config","type":"file","path":"${dir:mesh-state}/config.json","content":"{}"},
{"id":"imap","type":"container","name":"mailu-imap","image":"registry.invalid/imap@sha256:1",
"volumes":["${dir:data-mail}:/mail"],"env-file":["${dir:state}/mailu.env"]}]}`
var m catalogue.Manifest
if err := json.Unmarshal([]byte(raw), &m); err != nil {
panic(err)
}
if policy {
m.Upgrade = &catalogue.UpgradePolicy{Policy: catalogue.PolicyRecord, Why: "people's mail: a person chooses the moment"}
}
return m
}
func resourceOf(m *catalogue.Manifest, id string) map[string]any {
for _, r := range m.Resources {
if r["id"] == id {
return r
}
}
panic("no resource " + id)
}
// recordedMesh is a mesh running mailu at build c1 on anchor and laptop, as each was last sent.
type recordedMesh struct {
open *stores
sent [][]string
}
func aRecordedMesh(t *testing.T, c1 catalogue.Manifest) *recordedMesh {
t.Helper()
open := aMesh(t)
ctx := t.Context()
inv := open.inventory
rm := &recordedMesh{open: open}
rm.register(t, c1, "c1", time.Now().Add(-2*time.Hour))
for _, n := range []string{"anchor", "laptop"} {
if _, err := inv.Assign(ctx, n, c1.Module); err != nil {
t.Fatal(err)
}
if err := inv.RecordSent(ctx, nodeID(t, open, n), "d-"+n+"-c1", map[string]string{c1.Module: "c1"}); err != nil {
t.Fatal(err)
}
}
return rm
}
// register keeps a build of a module and makes it the mesh's.
func (rm *recordedMesh) register(t *testing.T, m catalogue.Manifest, commit string, at time.Time) {
t.Helper()
ctx := t.Context()
raw, _ := json.Marshal(m)
if err := rm.open.inventory.RecordBuild(ctx, inventory.Build{ID: "build-" + m.Module + "-" + commit, Module: m.Module,
Commit: commit, Repository: "novox/mesh-catalog", Path: "modules/" + m.Module, Manifest: raw, Asked: at, At: at}); err != nil {
t.Fatal(err)
}
if err := rm.open.inventory.RegisterModule(ctx, m, inventory.Source{Repository: "novox/mesh-catalog", Seat: "git",
Path: "modules/" + m.Module, BuiltFrom: commit, Head: commit, Asked: at}); err != nil {
t.Fatal(err)
}
}
// keptOn is what an ordinary send — a walk's, a healer's — keeps on a machine: module → the build it keeps.
func (rm *recordedMesh) keptOn(t *testing.T, node string) map[string]string {
t.Helper()
ctx := t.Context()
under, err := sendKeeps(ctx, rm.open.inventory)
if err != nil {
t.Fatal(err)
}
kept, err := recordedKept(under, rm.open, node)
if err != nil {
t.Fatal(err)
}
return kept
}
// walk opens a merge's walk of mailu's c2, built, and walks it with every send composed as an ordinary send
// composes it — a recorded module kept where its move waits — and applied at once.
func (rm *recordedMesh) walk(t *testing.T) inventory.Plan {
t.Helper()
ctx := t.Context()
inv := rm.open.inventory
wasSend := sendRollout
n := 0
sendRollout = func(ctx context.Context, open *stores, names []string) ([]string, error) {
rm.sent = append(rm.sent, append([]string(nil), names...))
current, err := open.inventory.CurrentBuilds(ctx)
if err != nil {
return nil, err
}
for _, node := range names {
kept := rm.keptOn(t, node)
builds := map[string]string{"mailu": current["mailu"].Commit}
if was, held := kept["mailu"]; held {
builds["mailu"] = was
}
n++
digest := fmt.Sprintf("d-%s-%d", node, n)
if err := inv.RecordSent(ctx, nodeID(t, open, node), digest, builds); err != nil {
return nil, err
}
if _, err := inv.RecordDoing(ctx, nodeID(t, open, node), inventory.Doing{Node: node,
Outcome: inventory.OutcomeApplied, Declared: digest, Applied: 1, At: time.Now()}); err != nil {
return nil, err
}
}
return names, nil
}
t.Cleanup(func() { sendRollout = wasSend })
built := time.Now().UTC()
plan := inventory.Plan{ID: "plan-mailu", Repository: "novox/mesh-catalog", Branch: "main", Commit: "c2",
Created: built, State: inventory.PlanBuilding, Tiers: [][]string{{"mailu"}},
Modules: map[string]*inventory.PlanModule{"mailu": {State: "built", BuiltAt: &built, Commit: "c2", Build: "build-mailu-c2"}}}
if err := inv.SavePlan(ctx, &plan); err != nil {
t.Fatal(err)
}
for i := 0; i < 8; i++ {
advancePlans(ctx, rm.open)
}
p, err := inv.PlanByID(ctx, "plan-mailu")
if err != nil {
t.Fatal(err)
}
return p
}
func (rm *recordedMesh) sentBuild(t *testing.T, node string) string {
t.Helper()
sent, _, err := rm.open.inventory.SentBuilds(t.Context(), node)
if err != nil {
t.Fatal(err)
}
return sent["mailu"]
}
// recordedWaitsOn is what `status` names a machine behind for, waiting for a person (ADR 0377 §6).
func (rm *recordedMesh) recordedWaitsOn(t *testing.T, node string) []string {
t.Helper()
f, err := readMoveFacts(t.Context(), rm.open.inventory)
if err != nil {
t.Fatal(err)
}
held, err := recordedWaiting(t.Context(), rm.open.inventory, f, node)
if err != nil {
t.Fatal(err)
}
return held
}
// §1, §6: a module that records because its manifest says so, whose move changes only its tools bundle, goes out
// with its walk to every machine, without a push; the walk says per machine that it went out and why; the policy
// still reads record; and status names no machine behind for it.
func TestARecordedModulesBundleOnlyMoveRollsOutInItsWalk(t *testing.T) {
c1 := aMailu(true)
rm := aRecordedMesh(t, c1)
c2 := aMailu(true)
c2.Bundles[0].Digest, c2.Bundles[0].Source = "sha256:bbb", "store/mailu-tools@sha256:bbb"
c2.Tools = append(c2.Tools, "mailu_queue")
rm.register(t, c2, "c2", time.Now().Add(-time.Minute))
for _, n := range []string{"anchor", "laptop"} {
if _, kept := rm.keptOn(t, n)["mailu"]; kept {
t.Fatalf("an ordinary send to %s keeps mailu at c1 for a move that recreates nothing", n)
}
}
p := rm.walk(t)
if p.State != inventory.PlanDone {
t.Fatalf("the walk is %s: %s", p.State, p.Note)
}
for _, n := range []string{"anchor", "laptop"} {
if got := rm.sentBuild(t, n); got != "c2" {
t.Fatalf("%s was sent mailu %s; want c2, with its walk", n, got)
}
if said := p.Modules["mailu"].Recorded[n]; said != recordedRollsWhy {
t.Fatalf("the walk says of mailu on %s %q", n, said)
}
if held := rm.recordedWaitsOn(t, n); len(held) != 0 {
t.Fatalf("status names %s behind for %v", n, held)
}
}
if !strings.Contains(p.Note, "mailu on anchor went out") || !strings.Contains(p.Note, "mailu on laptop went out") {
t.Fatalf("the walk's note does not say which way mailu went: %q", p.Note)
}
if u, err := rm.open.inventory.UpgradeOf(t.Context(), "mailu"); err != nil || u.Policy() != catalogue.PolicyRecord {
t.Fatalf("mailu's policy is %q (%v); want record still", u.Policy(), err)
}
}
// §1, §2, §6: a move that recreates a container — a new image, or a changed declaration — still waits for a push;
// the walk says per machine that it waits, with the recreate words; and status names the machine behind, with
// `push <node>` as the remedy. A bundle-only change on top of it stays held: it is judged against what the machine
// was last sent.
func TestARecordedModulesMoveThatRecreatesAContainerWaits(t *testing.T) {
for name, change := range map[string]func(m *catalogue.Manifest){
"image": func(m *catalogue.Manifest) { resourceOf(m, "imap")["image"] = "registry.invalid/imap@sha256:2" },
"spec": func(m *catalogue.Manifest) { resourceOf(m, "imap")["health"] = map[string]any{"kind": "runtime"} },
} {
t.Run(name, func(t *testing.T) {
rm := aRecordedMesh(t, aMailu(true))
c2 := aMailu(true)
change(&c2)
c2.Bundles[0].Digest = "sha256:bbb"
rm.register(t, c2, "c2", time.Now().Add(-time.Minute))
p := rm.walk(t)
for _, n := range []string{"anchor", "laptop"} {
if got := rm.sentBuild(t, n); got != "c1" {
t.Fatalf("%s was sent mailu %s; want c1 kept, waiting for a push", n, got)
}
if said := p.Modules["mailu"].Recorded[n]; !strings.HasPrefix(said, "waits: recreates 1 of mailu's 1") {
t.Fatalf("the walk says of mailu on %s %q", n, said)
}
if held := rm.recordedWaitsOn(t, n); len(held) != 1 || !strings.Contains(held[0], "recreates") {
t.Fatalf("status names %s behind for %v", n, held)
}
}
// §2: the next build changes only the bundle, and the machines were last sent c1: still held.
c3 := c2
c3.Bundles = []catalogue.Bundle{{Name: "tools", Source: "store/mailu-tools@sha256:ccc", Digest: "sha256:ccc"}}
rm.register(t, c3, "c3", time.Now())
if kept := rm.keptOn(t, "anchor"); kept["mailu"] != "c1" {
t.Fatalf("a bundle-only build on top of a held container change is not held: %v", kept)
}
})
}
}
// §3: a move touching a file a container mounts, a directory holding declared data, the data declaration or a
// provider waits.
func TestARecordedModulesMoveThatTouchesDataOrAProviderWaits(t *testing.T) {
for name, c := range map[string]struct {
change func(m *catalogue.Manifest)
says string
}{
"a mounted file": {func(m *catalogue.Manifest) { resourceOf(m, "config-env")["content"] = "A=2" }, "which its container imap mounts"},
"a data directory": {func(m *catalogue.Manifest) { resourceOf(m, "data-mail")["mode"] = "0700" }, "data-mail"},
"the data declaration": {func(m *catalogue.Manifest) {
m.Data = &catalogue.Data{Own: []catalogue.DataItem{{ID: "mail", Path: "${dir:data-mail}", Class: "irreplaceable", Why: "x"}}}
}, "data declaration"},
"a provider": {func(m *catalogue.Manifest) { m.Requires = append(m.Requires, "postgres-database") }, "what it requires"},
} {
t.Run(name, func(t *testing.T) {
rm := aRecordedMesh(t, aMailu(true))
c2 := aMailu(true)
c.change(&c2)
rm.register(t, c2, "c2", time.Now().Add(-time.Minute))
kept := rm.keptOn(t, "anchor")
if kept["mailu"] != "c1" {
t.Fatalf("an ordinary send does not keep mailu at c1: %v", kept)
}
held := rm.recordedWaitsOn(t, "anchor")
if len(held) != 1 || !strings.Contains(held[0], c.says) {
t.Fatalf("status says %v; want it to name %q", held, c.says)
}
})
}
}
// §5: a module a person set to record, and the bus, wait on a bundle-only move.
func TestAPersonsRecordAndTheBusHoldEveryMove(t *testing.T) {
t.Run("a person sets record", func(t *testing.T) {
rm := aRecordedMesh(t, aMailu(false))
c2 := aMailu(false)
c2.Bundles[0].Digest = "sha256:bbb"
rm.register(t, c2, "c2", time.Now().Add(-time.Minute))
if err := rm.open.inventory.SetUpgradeOf(t.Context(), "mailu", inventory.Upgrade{Why: "the migration week", By: "jochen"}); err != nil {
t.Fatal(err)
}
p := rm.walk(t)
if got := rm.sentBuild(t, "anchor"); got != "c1" {
t.Fatalf("a person's record was walked past: anchor was sent %s", got)
}
if said := p.Modules["mailu"].Recorded["anchor"]; !strings.Contains(said, "a person set it to record") {
t.Fatalf("the walk says %q", said)
}
if held := rm.recordedWaitsOn(t, "anchor"); len(held) != 1 {
t.Fatalf("status names anchor behind for %v", held)
}
})
t.Run("the bus", func(t *testing.T) {
bus := aMailu(false)
bus.Module = "nats"
bus.Provides = []catalogue.Offer{{Name: "mesh-bus"}}
bus.Data = nil
rm := aRecordedMesh(t, bus)
c2 := bus
c2.Bundles = []catalogue.Bundle{{Name: "tools", Source: "store/nats-tools@sha256:bbb", Digest: "sha256:bbb"}}
rm.register(t, c2, "c2", time.Now().Add(-time.Minute))
if kept := rm.keptOn(t, "anchor"); kept["nats"] != "c1" {
t.Fatalf("the bus moved by an ordinary send: %v", kept)
}
f, err := readMoveFacts(t.Context(), rm.open.inventory)
if err != nil {
t.Fatal(err)
}
if v := f.recordedMove("nats", "c1"); v.rolls || !strings.Contains(v.said, "bus") {
t.Fatalf("the bus's move: %+v", v)
}
})
}
// §4: a move whose earlier build is not in the build records cannot be judged, waits, and is said so.
func TestARecordedMoveThatCannotBeJudgedWaits(t *testing.T) {
rm := aRecordedMesh(t, aMailu(true))
c2 := aMailu(true)
c2.Bundles[0].Digest = "sha256:bbb"
rm.register(t, c2, "c2", time.Now().Add(-time.Minute))
// anchor was last sent a build the records do not hold.
if err := rm.open.inventory.RecordSent(t.Context(), nodeID(t, rm.open, "anchor"), "d-anchor-c0",
map[string]string{"mailu": "c0"}); err != nil {
t.Fatal(err)
}
f, err := readMoveFacts(t.Context(), rm.open.inventory)
if err != nil {
t.Fatal(err)
}
if v := f.recordedMove("mailu", "c0"); v.rolls || !strings.Contains(v.said, "could not be judged") ||
!strings.Contains(v.said, "not in the build records") {
t.Fatalf("an unjudgeable move: %+v", v)
}
if _, _, err := planFor(keepingRecorded(t.Context()), rm.open, "anchor"); err == nil || !strings.Contains(err.Error(), "push anchor") {
t.Fatalf("a send that cannot keep the unjudged build did not refuse with its remedy: %v", err)
}
// laptop, judged from c1, goes; anchor waits.
p := rm.walk(t)
if said := p.Modules["mailu"].Recorded["anchor"]; !strings.Contains(said, "could not be judged") {
t.Fatalf("the walk says of anchor %q", said)
}
if !slices.Contains(wentOut(p.Modules["mailu"].Recorded), "laptop") || rm.sentBuild(t, "laptop") != "c2" {
t.Fatalf("laptop, judged from c1, did not go: %+v", p.Modules["mailu"].Recorded)
}
}
+7 -2
View File
@@ -83,6 +83,9 @@ type moveFacts struct {
plans []inventory.Plan
// self is the release walking these moves, whose own gates are no other walk's (novox/hq ADR 0353).
self string
// judge judges a move under a module's own `record` (novox/hq ADR 0377): nil when the facts were not read,
// and every such move then waits.
judge func(module, was, now string) recordVerdict
}
type walkingKey struct{}
@@ -109,6 +112,7 @@ func readMoveFacts(ctx context.Context, inv *inventory.Inventory) (moveFacts, er
}
f.plans, err = inv.OpenPlans(ctx)
f.self, _ = ctx.Value(walkingKey{}).(string)
f.judge = recordJudge(ctx, inv)
return f, err
}
@@ -172,7 +176,8 @@ func inRun(run map[string]bool, commit string) bool {
}
// moves is what a machine's next send would move that no gate has seen: modules whose policy rolls out
// (a recorded one is a person's push), that the machine was sent before, moving to a build not identical
// (a recorded one is a person's push, unless its move under its own record recreates nothing and moves no data —
// novox/hq ADR 0377), that the machine was sent before, moving to a build not identical
// to the one it runs and that has passed no gate. A machine whose last send's builds are not known names
// none: the cascade holds it whole (ADR 0221).
//
@@ -185,7 +190,7 @@ func (f moveFacts) moves(node string, modules []string, sent map[string]string,
for _, m := range modules {
now := f.current[m]
was, carried := sent[m]
if !now.RollOut || !carried || f.identical(m, was, now.Commit) || (!all && f.gated(m, now.Commit)) {
if !carried || f.identical(m, was, now.Commit) || (!all && f.gated(m, now.Commit)) || !f.rolls(m, was) {
continue
}
out = append(out, inventory.CarriedMove{Module: m, Node: node, From: was, To: now.Commit})
+27 -1
View File
@@ -865,13 +865,32 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan,
together := map[string]bool{}
for _, m := range tier {
state := p.Modules[m]
if state == nil || state.SentAt != nil || state.State == planDeleted || !rollsOut(m) {
if state == nil || state.SentAt != nil || state.State == planDeleted {
continue
}
// **A recorded module goes where its move recreates nothing and moves no data** (novox/hq ADR 0377): judged
// per machine against the build it was last sent, said per machine, and sent only there.
var goes map[string]bool
if !rollsOut(m) {
out, err := judgeRecordedInWalk(ctx, inv, p, m, state)
if err != nil {
return false, err
}
if len(out) == 0 {
continue
}
goes = map[string]bool{}
for _, n := range out {
goes[n] = true
}
}
running, err := inv.Running(ctx, m)
if err != nil {
return false, err
}
if goes != nil {
running = slices.DeleteFunc(running, func(n string) bool { return !goes[n] })
}
runningOf[m] = running
// **A build whose source is unchanged is never a move** (novox/hq issue 280): made from the source
// of the build every machine running it was sent, it is registered with that build's artifacts —
@@ -1150,6 +1169,7 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan,
said = append(said, fmt.Sprintf("said nothing on its send within %s: %s", inventory.ApplySilentAfter,
strings.Join(silent, ", ")))
}
said = append(said, recordedSaid(*p)...)
end = strings.Join(said, "; ")
// **The new controller's user list follows its walk** (ADR 0236 §34, read without the gate since ADR 0360):
// once a walk that moved the controller ends, the bus's machine is sent the list it composes, if behind.
@@ -1955,6 +1975,12 @@ func plansCommand(ctx context.Context, args []string) error {
case s != nil && s.Gate != nil:
fmt.Printf(" %-22s %s\n", "", gateLine(s.Gate))
}
// Which way a recorded module went on each machine (novox/hq ADR 0377 §6).
if s != nil {
for _, n := range sortedKeysOf(s.Recorded) {
fmt.Printf(" %-22s on %s %s\n", "", n, s.Recorded[n])
}
}
// What the send to its first machine did to its containers (ADR 0245).
if s != nil && s.Gate != nil {
for _, c := range s.Gate.Carried {
+27
View File
@@ -205,6 +205,10 @@ func printStatus(asked answers) error {
fmt.Printf("%d machine(s) have never been sent anything:\n %s\n",
len(never), strings.Join(never, ", "))
}
for _, n := range sortedKeysOf(stringsKeys(asked.recordedWaits)) {
fmt.Printf(" %s is behind, waiting for a person: %s — `push %s` sends it\n", n,
strings.Join(asked.recordedWaits[n], "; "), n)
}
fmt.Printf("\n `push --behind` sends them\n\n")
}
@@ -540,6 +544,29 @@ func theThreeQuestions(ctx context.Context, open *stores) (answers, error) {
if err != nil {
return answers{}, err
}
// And, of those, what waits for a person's push (novox/hq ADR 0377 §6): a recorded module's move that went out
// with its walk raises nothing here.
if len(out.waiting) > 0 {
f, err := readMoveFacts(ctx, inv)
if err != nil {
return answers{}, err
}
for _, m := range out.waiting {
if m.Never {
continue
}
held, err := recordedWaiting(ctx, inv, f, m.Node)
if err != nil {
return answers{}, err
}
if len(held) > 0 {
if out.recordedWaits == nil {
out.recordedWaits = map[string][]string{}
}
out.recordedWaits[m.Node] = held
}
}
}
out.reported, err = inv.LastReports(ctx)
if err != nil {
return answers{}, err
+96
View File
@@ -0,0 +1,96 @@
package catalogue
import (
"encoding/json"
"strings"
"testing"
)
// novox/hq ADR 0377 §3: a move under a module's own `record` rolls out when it recreates none of its containers
// and moves none of its data, judged from the two builds' manifests.
func TestWhatAMoveUnderRecordHoldsFor(t *testing.T) {
base := func() Manifest {
raw := `{"module":"mailu","tools":["mailu_status"],
"requires":["postgres-database"],
"data":{"own":[{"id":"mail","path":"${dir:data-mail}","class":"valuable"}]},
"bundles":[{"name":"tools","source":"store/mailu-tools@sha256:aaa","digest":"sha256:aaa"}],
"resources":[
{"id":"state","type":"directory"},
{"id":"mesh-state","type":"directory"},
{"id":"data-mail","type":"directory"},
{"id":"config-env","type":"file","path":"${dir:state}/mailu.env","content":"A=1"},
{"id":"runtime-config","type":"file","path":"${dir:mesh-state}/config.json","content":"{}"},
{"id":"inbox-note","type":"file","path":"${dir:data-mail}/README","content":"x"},
{"id":"imap","type":"container","image":"registry/imap@sha256:1","volumes":["${dir:data-mail}:/mail"],
"env-file":["${dir:state}/mailu.env"]},
{"id":"watcher","type":"process","command":"watch"}]}`
var m Manifest
if err := json.Unmarshal([]byte(raw), &m); err != nil {
t.Fatal(err)
}
return m
}
resource := func(m *Manifest, id string) map[string]any {
for _, r := range m.Resources {
if r["id"] == id {
return r
}
}
t.Fatalf("no resource %s", id)
return nil
}
cases := []struct {
name string
change func(m *Manifest)
holds string // empty: it rolls out
}{
{"nothing", func(*Manifest) {}, ""},
{"a tools bundle", func(m *Manifest) {
m.Bundles[0].Digest, m.Bundles[0].Source = "sha256:bbb", "store/mailu-tools@sha256:bbb"
}, ""},
{"the tool list", func(m *Manifest) { m.Tools = append(m.Tools, "mailu_queue") }, ""},
{"a file no container mounts", func(m *Manifest) { resource(m, "runtime-config")["content"] = `{"a":1}` }, ""},
{"a container's image", func(m *Manifest) { resource(m, "imap")["image"] = "registry/imap@sha256:2" }, "recreates 1 of mailu's 1"},
{"a container's declaration", func(m *Manifest) { resource(m, "imap")["health"] = map[string]any{"kind": "runtime"} }, "recreates 1"},
{"a container added", func(m *Manifest) {
m.Resources = append(m.Resources, map[string]any{"id": "smtp", "type": "container", "image": "x"})
}, "recreates"},
{"a mounted file", func(m *Manifest) { resource(m, "config-env")["content"] = "A=2" }, "which its container imap mounts"},
{"a file within declared data", func(m *Manifest) { resource(m, "inbox-note")["content"] = "y" }, "which its container imap mounts"},
{"a directory holding declared data", func(m *Manifest) { resource(m, "data-mail")["mode"] = "0700" }, "data-mail"},
{"the data declaration", func(m *Manifest) { m.Data.Own[0].Class = ClassIrreplaceable }, "data declaration"},
{"a provider", func(m *Manifest) { m.Requires = append(m.Requires, "s3") }, "what it requires"},
{"what it provides", func(m *Manifest) { m.Provides = append(m.Provides, Offer{Name: "smtp"}) }, "what it provides"},
{"a process of a module with data", func(m *Manifest) { resource(m, "watcher")["command"] = "watch2" }, "long-running process watcher"},
}
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
from, to := base(), base()
c.change(&to)
got := Holds(from, to)
switch {
case c.holds == "" && got != "":
t.Fatalf("held: %s", got)
case c.holds != "" && !strings.Contains(got, c.holds):
t.Fatalf("holds %q; want it to say %q", got, c.holds)
}
})
}
// A data directory nothing mounts still holds its data.
from, to := base(), base()
from.Data.Own = append(from.Data.Own, DataItem{ID: "archive", Path: "${dir:mesh-state}/archive", Class: "valuable"})
to.Data = &Data{Own: append([]DataItem(nil), from.Data.Own...)}
resource(&to, "runtime-config")["path"] = "${dir:mesh-state}/archive/config.json"
resource(&from, "runtime-config")["path"] = "${dir:mesh-state}/archive/config.json"
resource(&to, "runtime-config")["content"] = "new"
if got := Holds(from, to); !strings.Contains(got, "lies within its data archive") {
t.Fatalf("a file within data nothing mounts: %q", got)
}
// A process of a module with no data of its own may change.
from, to = base(), base()
from.Data, to.Data = nil, nil
resource(&to, "watcher")["command"] = "watch2"
if got := Holds(from, to); got != "" {
t.Fatalf("a process of a module with no data held: %s", got)
}
}
+215
View File
@@ -109,3 +109,218 @@ func sameDeclaration(a, b any) bool {
_ = json.Unmarshal(rb, &nb)
return reflect.DeepEqual(na, nb)
}
// What a move under a module's own `record` may do without a person (novox/hq ADR 0377).
//
// A module records because its manifest says so or because it keeps irreplaceable data: mail, whose containers
// recreated together interrupt everyone's mail, and the database server, whose restart drops every consumer. That
// hold also held the moves that interrupt nothing — on 2026-10-15 the database server moved with only its tools
// bundle changed, its walk said it "recreates none of its containers", and the machine stayed behind until a
// person pushed. So such a move rolls out like any other, and only a move that recreates a container or touches
// data still waits.
// Holds is why a move from one build's manifest to another's still waits for a person under a module's own
// `record` (ADR 0377 §3), empty when it recreates none of its containers and moves none of its data: no container
// changed, added or gone; no file or directory changed, added or gone that a container mounts, or that holds or
// lies within a path of the module's declared data; no change to its `data` declaration, nor to what it provides,
// requires or contributes; and, for a module that declares data of its own, no change to a long-running process.
// Its tools bundles, a file nothing mounts that holds none of its data, and manifest fields that recreate
// nothing — its tool list — may differ.
func Holds(from, to Manifest) string {
module := to.Module
if module == "" {
module = from.Module
}
if said := Recreates(from, to).Say(module); said != "" {
return said
}
if !sameDeclaration(from.Data, to.Data) {
return "it changes " + module + "'s data declaration"
}
for _, f := range []struct {
what string
was, s any
}{
{"what it provides", from.Provides, to.Provides},
{"what it requires", from.Requires, to.Requires},
{"which of its requirements are optional", from.OptionalRequires, to.OptionalRequires},
{"what it contributes to a provider", from.Contributes, to.Contributes},
{"what it contributes to a provider", from.ContributesMany, to.ContributesMany},
// A setting is filled into the files the module writes: a changed default changes what a container may
// mount without the file's own declaration changing.
{"its settings, which the files it writes are filled from", from.Settings, to.Settings},
} {
if !sameDeclaration(emptyAsNil(f.was), emptyAsNil(f.s)) {
return "it changes " + f.what
}
}
mounted := append(mountSources(from), mountSources(to)...)
data := append(dataPaths(from), dataPaths(to)...)
keepsData := (from.Data != nil && len(from.Data.Own) > 0) || (to.Data != nil && len(to.Data.Own) > 0)
before, after := resourcesByID(from), resourcesByID(to)
var changed []string
for id, r := range after {
if was, held := before[id]; !held || !sameDeclaration(was, r) {
changed = append(changed, id)
}
}
for id := range before {
if _, kept := after[id]; !kept {
changed = append(changed, id)
}
}
sort.Strings(changed)
dirs := absoluteDirectories(from, to)
for _, key := range changed {
for _, r := range []map[string]any{before[key], after[key]} {
if r == nil {
continue
}
kind, id := fmt.Sprint(r["type"]), fmt.Sprint(r["id"])
switch kind {
case "file", "directory":
for _, p := range pathsOf(r, dirs) {
for _, m := range mounted {
if pathsMeet(p, expandDir(m.source, dirs)) {
return fmt.Sprintf("it changes the %s %s, which its container %s mounts", kind, id, m.container)
}
}
for _, d := range data {
if pathsMeet(p, expandDir(d.path, dirs)) {
return fmt.Sprintf("it changes the %s %s, which holds or lies within its data %s", kind, id, d.id)
}
}
}
case "process", "service":
if keepsData {
return fmt.Sprintf("it changes its long-running %s %s, and it keeps data of its own", kind, id)
}
}
}
}
return ""
}
type mountSource struct{ container, source string }
// mountSources is every path a container of the module reads from its machine: its volumes' sources and its
// environment files.
func mountSources(m Manifest) []mountSource {
var out []mountSource
for _, r := range m.Resources {
if fmt.Sprint(r["type"]) != "container" {
continue
}
id := fmt.Sprint(r["id"])
vols, _ := r["volumes"].([]any)
for _, v := range vols {
s, _ := v.(string)
if mount := volumeMount.FindStringSubmatch(s); mount != nil {
out = append(out, mountSource{id, mount[1]})
}
}
switch files := r["env-file"].(type) {
case string:
out = append(out, mountSource{id, files})
case []any:
for _, f := range files {
if s, ok := f.(string); ok {
out = append(out, mountSource{id, s})
}
}
}
}
return out
}
type dataPath struct{ id, path string }
func dataPaths(m Manifest) []dataPath {
if m.Data == nil {
return nil
}
var out []dataPath
for _, it := range m.Data.Own {
out = append(out, dataPath{it.ID, it.Path})
}
return out
}
func resourcesByID(m Manifest) map[string]map[string]any {
out := map[string]map[string]any{}
for _, r := range m.Resources {
out[fmt.Sprint(r["type"])+":"+fmt.Sprint(r["id"])] = r
}
return out
}
// absoluteDirectories is, per directory id, the absolute path a directory of either build states.
func absoluteDirectories(ms ...Manifest) map[string]string {
out := map[string]string{}
for _, m := range ms {
for _, r := range m.Resources {
if fmt.Sprint(r["type"]) != "directory" {
continue
}
if p, ok := r["path"].(string); ok && strings.HasPrefix(p, "/") {
out[fmt.Sprint(r["id"])] = strings.TrimRight(p, "/")
}
}
}
return out
}
// pathsOf is every way a file or directory resource is named: its path, and a directory's `${dir:<id>}`.
func pathsOf(r map[string]any, dirs map[string]string) []string {
var out []string
if p, ok := r["path"].(string); ok && p != "" {
out = append(out, expandDir(p, dirs))
}
if fmt.Sprint(r["type"]) == "directory" {
out = append(out, expandDir("${dir:"+fmt.Sprint(r["id"])+"}", dirs))
}
return out
}
// expandDir writes a leading `${dir:<id>}` as the directory's absolute path when one is stated, so a path named
// either way is one path.
func expandDir(p string, dirs map[string]string) string {
p = strings.TrimRight(p, "/")
if ref := dirPlain.FindStringSubmatch(p); ref != nil {
if abs, ok := dirs[ref[1]]; ok {
return abs
}
return p
}
if strings.HasPrefix(p, "${dir:") {
end := strings.Index(p, "}")
if abs, ok := dirs[p[len("${dir:"):end]]; ok {
return abs + p[end+1:]
}
}
return p
}
// pathsMeet is whether one path is the other or lies within it.
func pathsMeet(a, b string) bool {
if a == "" || b == "" {
return false
}
return a == b || strings.HasPrefix(a, b+"/") || strings.HasPrefix(b, a+"/")
}
// emptyAsNil reads an empty list or map as nothing, so a field written empty is no change from one left out.
func emptyAsNil(v any) any {
rv := reflect.ValueOf(v)
switch rv.Kind() {
case reflect.Slice, reflect.Map:
if rv.Len() == 0 {
return nil
}
case reflect.Ptr:
if rv.IsNil() {
return nil
}
}
return v
}
+12 -1
View File
@@ -1476,6 +1476,16 @@ type CurrentBuild struct {
Commit string
// RollOut is the module's upgrade policy, as Upgrade.RollOut.
RollOut bool
// From is where the policy came from, as Upgrade.From: a move under a module's own `record` — its manifest's or
// its irreplaceable data's — rolls out when it recreates nothing and moves no data (novox/hq ADR 0377), one under
// a person's or the bus's never does.
From string
}
// OwnRecord is whether a module records because its manifest says so or because it keeps irreplaceable data
// (novox/hq ADR 0377 §1): the `record` whose moves that recreate nothing roll out.
func (b CurrentBuild) OwnRecord() bool {
return !b.RollOut && (b.From == catalogue.FromModule || b.From == catalogue.FromData)
}
// CurrentBuilds is every module's current build and upgrade policy, in one read: what a send records
@@ -1497,7 +1507,8 @@ func (i *Inventory) CurrentBuilds(ctx context.Context) (map[string]CurrentBuild,
if err := rows.Scan(&name, &b.Commit, &chosen, &together, &why, &by, &manifest); err != nil {
return nil, err
}
b.RollOut = upgradeFrom(chosen, together, why, by, manifest).RollOut
u := upgradeFrom(chosen, together, why, by, manifest)
b.RollOut, b.From = u.RollOut, u.From
out[name] = b
}
return out, rows.Err()
+5
View File
@@ -313,6 +313,11 @@ type PlanModule struct {
// Rest is, per machine of the rest, the declaration the send after the first-node gate carried there (novox/hq ADR
// 0282 decision 6): the machine's first report of it, applied, is when the build runs there.
Rest map[string]SentDeclaration `json:"rest,omitempty"`
// Recorded is, for a module whose policy records, per machine running it with a move, which way the walk took it
// (novox/hq ADR 0377 §6): "went out: …" when the move recreates none of its containers and moves none of its
// data, or "waits: …" with the recreate words, the data change named, the person's or the bus's hold, or why it
// could not be judged.
Recorded map[string]string `json:"recorded,omitempty"`
}
// PlanGate is one module's rollout record at its gate (to-be 45 §8): the component, the first machine,
+2 -2
View File
@@ -73,10 +73,10 @@ func TestTheCurrentBuildsAreTheCatalogues(t *testing.T) {
t.Fatal(err)
}
// Rolled out by default, one machine first and gated (novox/hq ADR 0236).
if got := current["resolver"]; got != (CurrentBuild{Commit: "c1", RollOut: true}) {
if got := current["resolver"]; got != (CurrentBuild{Commit: "c1", RollOut: true, From: catalogue.FromDefault}) {
t.Errorf("resolver is at %+v", got)
}
if got := current["by-hand"]; got != (CurrentBuild{RollOut: true}) {
if got := current["by-hand"]; got != (CurrentBuild{RollOut: true, From: catalogue.FromPerson}) {
t.Errorf("a module with no source is at %+v", got)
}
}
+5
View File
@@ -74,3 +74,8 @@
./cmd/mesh-controller TestOneProviderDownAndThreeConsumersFailingAreOneCondition
./cmd/mesh-controller TestAConsumerFailingBesideAHealthyProviderIsItsOwn
./internal/inventory TestWhatRunsIsTheRegisteredBuildNotANewerOnePutBack
./cmd/mesh-controller TestARecordedModulesBundleOnlyMoveRollsOutInItsWalk
./cmd/mesh-controller TestARecordedModulesMoveThatRecreatesAContainerWaits
./cmd/mesh-controller TestARecordedModulesMoveThatTouchesDataOrAProviderWaits
./cmd/mesh-controller TestAPersonsRecordAndTheBusHoldEveryMove
./cmd/mesh-controller TestARecordedMoveThatCannotBeJudgedWaits