Compare commits

..
108 changed files with 688 additions and 1855 deletions
+5 -22
View File
@@ -30,11 +30,6 @@ 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. // 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 // 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 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 { func watchApplyFailing(f *signalFacts) []conditions.Observation {
heard := map[string]bool{} heard := map[string]bool{}
for _, m := range f.machines { for _, m := range f.machines {
@@ -45,7 +40,7 @@ func watchApplyFailing(f *signalFacts) []conditions.Observation {
var out []conditions.Observation var out []conditions.Observation
said := map[string]bool{} said := map[string]bool{}
for _, d := range f.doing { for _, d := range f.doing {
if !d.Wrong() || d.Since == nil || !heard[d.Node] || (f.now.Sub(*d.Since) <= applyEvery && !f.applyFailingOpen[d.Node]) { if !d.Wrong() || d.Since == nil || f.now.Sub(*d.Since) <= applyEvery || !heard[d.Node] {
continue continue
} }
out = append(out, applyFailing(d, f.now)) out = append(out, applyFailing(d, f.now))
@@ -56,8 +51,7 @@ func watchApplyFailing(f *signalFacts) []conditions.Observation {
// on a machine whose node-engine applied what it was sent. // on a machine whose node-engine applied what it was sent.
failed := map[string][]inventory.ProvReport{} failed := map[string][]inventory.ProvReport{}
for _, r := range f.provFailed { for _, r := range f.provFailed {
if r.Status != inventory.ProvFailed || !heard[r.Node] || said[r.Node] || if r.Status != inventory.ProvFailed || f.now.Sub(r.At) <= applyEvery || !heard[r.Node] || said[r.Node] {
(f.now.Sub(r.Since()) <= applyEvery && !f.applyFailingOpen[r.Node]) {
continue continue
} }
failed[r.Node] = append(failed[r.Node], r) failed[r.Node] = append(failed[r.Node], r)
@@ -68,28 +62,17 @@ func watchApplyFailing(f *signalFacts) []conditions.Observation {
return out 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 // provisioningFailing is one machine's S21 condition for the modules whose files its provisioner client failed to
// place. // place.
func provisioningFailing(m string, failed []inventory.ProvReport, now time.Time) conditions.Observation { 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 }) sort.Slice(failed, func(a, b int) bool { return failed[a].Module < failed[b].Module })
var names, words []string var names, words []string
since := failed[0].Since() since := failed[0].At
for _, r := range failed { for _, r := range failed {
names = append(names, r.Module) names = append(names, r.Module)
words = append(words, r.Module+": "+oneLine(orNoReason(r.Reason))) words = append(words, r.Module+": "+oneLine(orNoReason(r.Reason)))
if r.Since().Before(since) { if r.At.Before(since) {
since = r.Since() since = r.At
} }
} }
o := conditions.Observation{Scope: conditions.ScopeMachine, ID: m, Kind: kindApplyFailing, Machine: m, o := conditions.Observation{Scope: conditions.ScopeMachine, ID: m, Kind: kindApplyFailing, Machine: m,
+2 -8
View File
@@ -56,15 +56,9 @@ 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 catalogue this // The real catalogue passes the command, the way it passes the test that used to be the only check.
// 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) { func TestModuleCheckPassesTheCatalogue(t *testing.T) {
root := filepath.Join(beside.CapturedDir(t, "mesh-catalog"), "modules") root := beside.Catalogue(t)
paths, err := manifestsUnder(root) paths, err := manifestsUnder(root)
if err != nil || len(paths) == 0 { if err != nil || len(paths) == 0 {
t.Fatalf("no manifests under %s: %v", root, err) t.Fatalf("no manifests under %s: %v", root, 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 { if n.Stated && n.Contract >= compose.VaultBoxContract {
boxes, err := coreBoxesFor(ctx, inv, node, resolved.Plan.Modules, func(m catalogue.Manifest) bool { boxes, err := coreBoxesFor(ctx, inv, node, resolved.Plan.Modules, func(m catalogue.Manifest) bool {
return handing.Handing(m) return handing.Handing(m)
}, true, in.BusMinted) }, true)
if err != nil { if err != nil {
return nil, err return nil, err
} }
+2 -2
View File
@@ -124,11 +124,11 @@ func composeModulesByNode(ctx context.Context, open *stores, nodes []string) (ma
errsOf[node] = append(errsOf[node], fmt.Errorf("%s: %w", node, err)) errsOf[node] = append(errsOf[node], fmt.Errorf("%s: %w", node, err))
return return
} }
if moved := builtByTheMesh(moved); len(moved) > 0 { if len(moved) > 0 {
fmt.Printf("compositions: %s's %s compose with the node's own: no fingerprint of the new form names them\n", 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, ", ")) node, strings.Join(moved, ", "))
} }
if left := builtByTheMesh(left); len(left) > 0 { if len(left) > 0 {
fmt.Printf("compositions: %s's %s are left out of the node's own: the controller knows no fingerprint of the "+ 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", "new form for them, and the node's want or have names their own composition (issue 1945)\n",
node, strings.Join(left, ", ")) node, strings.Join(left, ", "))
+4 -12
View File
@@ -69,15 +69,7 @@ type coreBoxes struct {
// coreModulesOf is every module of the plan on the core's list whose own secrets this machine's declaration carries, // 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 // 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. // 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{} out := map[string][]string{}
for _, m := range modules { for _, m := range modules {
if !catalogue.CarriesVaultSealed(m.Module) || (handed != nil && handed(m)) { if !catalogue.CarriesVaultSealed(m.Module) || (handed != nil && handed(m)) {
@@ -85,7 +77,7 @@ func coreModulesOf(modules []catalogue.Manifest, handed func(catalogue.Manifest)
} }
var names []string var names []string
for name, own := range m.OwnSecrets.Plain() { for name, own := range m.OwnSecrets.Plain() {
if own.IssuedBy != catalogue.IssuedOutside && processAccountWarning(m, node, name, minted) == "" { if own.IssuedBy != catalogue.IssuedOutside {
names = append(names, name) names = append(names, name)
} }
} }
@@ -216,8 +208,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. // 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, func coreBoxesFor(ctx context.Context, inv *inventory.Inventory, node string, modules []catalogue.Manifest,
handed func(catalogue.Manifest) bool, allocating bool, minted []string) (coreBoxes, error) { handed func(catalogue.Manifest) bool, allocating bool) (coreBoxes, error) {
core := coreModulesOf(modules, handed, node, minted) core := coreModulesOf(modules, handed)
if len(core) == 0 { if len(core) == 0 {
if allocating { if allocating {
settleVaultSilent(ctx, node) settleVaultSilent(ctx, node)
+1 -28
View File
@@ -124,35 +124,8 @@ func TestTheCoresModulesAreTheListsWithTheirOwnSecrets(t *testing.T) {
{Module: "postgres", OwnSecrets: catalogue.OwnSecrets{"superuser": {Path: "/c"}}}, {Module: "postgres", OwnSecrets: catalogue.OwnSecrets{"superuser": {Path: "/c"}}},
{Module: "mesh-provisioner", OwnSecrets: catalogue.OwnSecrets{"broker": {Path: "/d"}}}, {Module: "mesh-provisioner", OwnSecrets: catalogue.OwnSecrets{"broker": {Path: "/d"}}},
} }
got := coreModulesOf(modules, func(m catalogue.Manifest) bool { return m.Module == "mesh-provisioner" }, "anchor", nil) got := coreModulesOf(modules, func(m catalogue.Manifest) bool { return m.Module == "mesh-provisioner" })
if want := map[string][]string{"nats": {"broker"}}; !reflect.DeepEqual(got, want) { if want := map[string][]string{"nats": {"broker"}}; !reflect.DeepEqual(got, want) {
t.Fatalf("got %v, want %v", 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)
}
}
}
@@ -1,31 +0,0 @@
//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
@@ -1,106 +0,0 @@
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 return p
} }
objects := write("objects", `{"module":"objects","version":"1", objects := write("objects", `{"module":"objects","version":"1",
"provides":[{"name":"s3-bucket","scope":"mesh","identity":{"max":20,"in":"an S3 access key"},"credential":{"form":"pair"}}], "provides":[{"name":"s3-bucket","scope":"mesh","identity":{"max":20,"in":"an S3 access key"}}],
"receives":{"s3-bucket":"/var/lib/mesh/objects/mesh.json"}}`) "receives":{"s3-bucket":"/var/lib/mesh/objects/mesh.json"}}`)
resolver := write("resolver", `{"module":"resolver","version":"1", resolver := write("resolver", `{"module":"resolver","version":"1",
"provides":[{"name":"wildcard-resolution","scope":"mesh","identity":false,"credential":{"form":"none"}}]}`) "provides":[{"name":"wildcard-resolution","scope":"mesh","identity":false}]}`)
album := write("photoalbum", `{"module":"photoalbum","version":"1","requires":["s3-bucket"]}`) album := write("photoalbum", `{"module":"photoalbum","version":"1","requires":["s3-bucket"]}`)
nm := write("networkmanager", `{"module":"networkmanager","version":"1","requires":["wildcard-resolution"]}`) nm := write("networkmanager", `{"module":"networkmanager","version":"1","requires":["wildcard-resolution"]}`)
+2 -7
View File
@@ -182,9 +182,6 @@ func TestANeedIsIssuedToItsHolderOnlyWhereTheProviderIsMovedAndItsSoftwareIssues
offer := catalogue.Offer{Name: "gitea-read", Scope: catalogue.ScopeMesh} offer := catalogue.Offer{Name: "gitea-read", Scope: catalogue.ScopeMesh}
if issued { if issued {
offer.Credential = &catalogue.OfferCredential{IssuedBy: catalogue.IssuedBySoftware, Tool: "gitea_agent_token"} 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}} return catalogue.Manifest{Module: "gitea", Version: "1", Moved: moved, Provides: []catalogue.Offer{offer}}
} }
@@ -219,10 +216,8 @@ func TestANeedIsIssuedToItsHolderOnlyWhereTheProviderIsMovedAndItsSoftwareIssues
// marked moved; another module is not. // marked moved; another module is not.
func TestTheModuleHoldingTheVaultSeatCountsAsMovedForItsConsumers(t *testing.T) { func TestTheModuleHoldingTheVaultSeatCountsAsMovedForItsConsumers(t *testing.T) {
vault := catalogue.Manifest{Module: "mesh-vault", Version: "1", Claims: []catalogue.Claim{{Name: "mesh-vault", Scope: catalogue.ScopeMesh}}, 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, 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}}}
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 { 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) t.Errorf("the vault's consumer is not answered by a moved provider: %+v", got)
} }
+1 -54
View File
@@ -21,13 +21,8 @@ func holdingOne(names ...string) func(broker.Consumer) (uint64, error) {
} }
func deafObservations(now time.Time, held func(broker.Consumer) (uint64, error)) (string, int) { 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) f := calm(now)
loop, err := readLoop(now.Add(-loopDeafAfter-time.Minute), due, loopConsumers(), held) loop, err := readLoop(now.Add(-loopDeafAfter-time.Minute), loopConsumers(), held)
if err != nil { if err != nil {
panic(err) panic(err)
} }
@@ -70,51 +65,3 @@ 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
}
+42 -98
View File
@@ -16,8 +16,6 @@ import (
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/catalogue" "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/broker"
"github.com/novox/mesh-controller/internal/inventory" "github.com/novox/mesh-controller/internal/inventory"
"github.com/novox/mesh-controller/internal/link" "github.com/novox/mesh-controller/internal/link"
@@ -181,12 +179,7 @@ func (k keepingDeliveryMerges) KeepDeliveryMerge(ctx context.Context, emitter st
if err != nil { if err != nil {
return fmt.Errorf("%w: the bucket of mesh-delivery's merges: %w", link.ErrTryAgain, err) return fmt.Errorf("%w: the bucket of mesh-delivery's merges: %w", link.ErrTryAgain, err)
} }
// Through kvlimit (sweep 2026-10-30, issue 774): a word too large for the bucket is refused before it is put, and if _, err := kv.Put(ctx, deliveryMergeKey(d.Owner, d.Repo, d.Number, d.HeadSHA), body); err != nil {
// 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) return fmt.Errorf("%w: %w", link.ErrTryAgain, err)
} }
if k.kept != nil { if k.kept != nil {
@@ -350,113 +343,64 @@ func claimantsOf(entries []inventory.Entry, seat string) []string {
return out return out
} }
// keptForgeMerge is the forge's word on the merge of a repository at a commit: the pull request and the head it merged, // forgeMergeKey is where the forge's announcement of a merge is kept, by the merge commit: the one thing a build of a merge
// read from the forge bridge's own stream, FORGE, where mesh-delivery-forge-bridge says every merge as // names, from which the pull request and its head are found (novox/hq ADR 0420 §7).
// `forge.<owner>.<name>.merged` and keeps it a week (novox/hq ADR 0437). False when the stream holds no merge at that func forgeMergeKey(owner, repo, commit string) string {
// commit. A variable so a test can hand it one. return fmt.Sprintf("merge/%s/%s/%s", owner, repo, commit)
// }
// **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 // keepForgeMerge keeps the forge's announcement of a merge in the bucket of the merges, a week at most as that bucket
// bucket's value size held that consumer for a day and queued every merge behind it, and the deaf watchdog read the // keeps them. The controller plans nothing from a merge any more (ADR 0539 §17); this is all it keeps of one.
// queue as a deaf controller. The forge bridge's stream is the record of the merges (research 072, 03-domains: the var keepForgeMerge = func(ctx context.Context, m link.SourceMoved) error {
// delivery parts own what a merge is); the controller reads it where it needs one, at a moved module's registration. 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.
var keptForgeMerge = func(ctx context.Context, owner, repo, commit string) (keptMerge, bool, error) { var keptForgeMerge = func(ctx context.Context, owner, repo, commit string) (keptMerge, bool, error) {
serving := servingBus.Load() serving := servingBus.Load()
if serving == nil { if serving == nil {
return keptMerge{}, false, errors.New("only the serving controller reads the forge's merges") return keptMerge{}, false, errors.New("only the serving controller keeps what the forge announced")
} }
api, err := jetstream.New(serving.Conn()) api, err := jetstream.New(serving.Conn())
if err != nil { if err != nil {
return keptMerge{}, false, err return keptMerge{}, false, err
} }
stream, err := api.Stream(ctx, forgeStream) kv, err := api.KeyValue(ctx, broker.DeliveryMergesBucket)
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 { if err != nil {
return keptMerge{}, false, err return keptMerge{}, false, err
} }
// Read for the four fields a registration needs, not checked whole (sdkdelivery.DecodeForge): a merged message a e, err := kv.Get(ctx, forgeMergeKey(owner, repo, commit))
// newer bridge writes, or one the SDK's check refuses for a part the controller never reads, still names its pull if errors.Is(err, jetstream.ErrKeyNotFound) {
// 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 return keptMerge{}, false, nil
} }
if err != nil { if err != nil {
return keptMerge{}, false, err return keptMerge{}, false, err
} }
if k, ok := match(data); ok { var k keptMerge
return k, true, nil if err := json.Unmarshal(e.Value(), &k); err != nil {
return keptMerge{}, false, err
} }
for from := uint64(1); from < lastSeq; { return k, true, nil
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. // keptMerge is a merge as the record keeps it: the forge's announcement.
+10 -2
View File
@@ -99,9 +99,17 @@ func TestAModuleBuiltFromANewRepositoryIsRegisteredOnlyOnTheOperatorsApproveOnTh
return err == nil && kept return err == nil && kept
}) })
// The forge bridge's merge of that pull request on its stream, and the build it asked for. // The forge's merge of that pull request, as the controller heard it, and the build it asked for.
const commit = "c0ffee0123456789" const commit = "c0ffee0123456789"
sayForgeMerge(ctx, t, js, "novox/mesh-host", 7, a.merge.HeadSHA, commit) 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)
}
id := fmt.Sprintf("build-%d", time.Now().Add(time.Hour).UnixNano()) 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", 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 { Path: "modules/sudo", Ref: "main", Commit: commit, For: "merge"}); err != nil {
+10 -2
View File
@@ -100,9 +100,17 @@ func TestANewModuleFromANewRepositoryIsRegisteredOnlyOnTheOperatorsApproveOnTheB
return err == nil && kept return err == nil && kept
}) })
// The forge bridge's merge of that pull request on its stream, and the build it asked for. // The forge's merge of that pull request, as the controller heard it, and the build it asked for.
const commit = "c0ffee0123456789" const commit = "c0ffee0123456789"
sayForgeMerge(ctx, t, js, "novox/mesh-delivery", 7, a.merge.HeadSHA, commit) 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)
}
id := fmt.Sprintf("build-%d", time.Now().Add(time.Hour).UnixNano()) 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", 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 { Path: "modules/mesh-delivery-view", Ref: "main", Commit: commit, For: "merge"}); err != nil {
@@ -11,6 +11,7 @@ import (
"git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/catalogue"
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/novox/mesh-controller/internal/inventory" "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 // 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
@@ -32,6 +33,12 @@ func TestAMergeAddingAModuleEndsRegisteredFromItsKeyAndAsksNoBuild(t *testing.T)
watchNewModules(watching, open) watchNewModules(watching, open)
const commit = "3da80a4b00000000" 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. // The indexer writes the module's trunk value; the builder builds it, then its key is done.
fp := fpOf('7') fp := fpOf('7')
manifest := b.indexed(t, "systemd-resolved", "1", "novox/mesh-catalog", "modules/systemd-resolved", commit, fp) manifest := b.indexed(t, "systemd-resolved", "1", "novox/mesh-catalog", "modules/systemd-resolved", commit, fp)
-42
View File
@@ -7,7 +7,6 @@ import (
"time" "time"
"git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/catalogue"
"github.com/novox/mesh-controller/internal/conditions"
"github.com/novox/mesh-controller/internal/inventory" "github.com/novox/mesh-controller/internal/inventory"
) )
@@ -73,44 +72,3 @@ func TestAFailedPlacingRaisesApplyFailing(t *testing.T) {
t.Fatalf("a failed placing is not raised as anchor's apply-failing: %+v", got) 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)
}
}
}
-54
View File
@@ -1,54 +0,0 @@
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
}
@@ -1,31 +0,0 @@
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)
}
}
}
+1 -5
View File
@@ -284,17 +284,13 @@ func watchLoop(f *signalFacts) []conditions.Observation {
if since.IsZero() { if since.IsZero() {
since = f.started 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 { if f.now.Sub(since) <= loopDeafAfter {
return nil return nil
} }
return []conditions.Observation{{Scope: conditions.ScopeCore, ID: "controller", Kind: "controller-deaf", return []conditions.Observation{{Scope: conditions.ScopeCore, ID: "controller", Kind: "controller-deaf",
Token: "deaf", Machine: f.host, Severity: conditions.Urgent, 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 "+ Summary: fmt.Sprintf("the controller's event loop has taken nothing for %s while its consumers hold %d "+
"message(s): reports and events are not being acted on", ago(f.now.Sub(since)), f.loop.pending), "message(s): reports, builds and merges 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))}} Said: fmt.Sprintf("%d pending (%s), last taken %s", f.loop.pending, f.loop.where, since.UTC().Format(time.RFC3339))}}
} }
+1 -14
View File
@@ -160,7 +160,7 @@ func switchList(ctx context.Context, inv *inventory.Inventory, keys switchReads)
var out []switchLine var out []switchLine
for _, e := range entries { for _, e := range entries {
m := e.Manifest.Module m := e.Manifest.Module
if e.Provided || e.Source.Repository == "" || sdkdelivery.ProvidedByMesh(m) { if e.Provided || e.Source.Repository == "" {
continue continue
} }
trunk, err := keys.ModuleAt(ctx, m, "") trunk, err := keys.ModuleAt(ctx, m, "")
@@ -202,16 +202,3 @@ 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 // empty-on-error: not an error; the mesh holds no build of that commit on record, which the line says
return nil, nil 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,6 +93,15 @@ func (f following) Announceable(ctx context.Context) ([]link.Announcement, error
return out, nil 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). // actingOnMerges keeps one merge acted on at a time (novox/hq issue 266).
var actingOnMerges sync.Mutex var actingOnMerges sync.Mutex
+4 -14
View File
@@ -58,11 +58,6 @@ type signalFacts struct {
// provFailed is every module whose files a machine's provisioner client last said it failed to place (S21, // 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. // novox/hq ADR 0362 §3): that machine's failed apply.
provFailed []inventory.ProvReport 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 loop loopFacts
loopErr error loopErr error
@@ -152,9 +147,7 @@ type machineFacts struct {
func (m machineFacts) asleep() bool { return m.power.Away() } func (m machineFacts) asleep() bool { return m.power.Away() }
type loopFacts struct { 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 pending uint64
where string where string
} }
@@ -259,7 +252,6 @@ func (w *watchdogs) see(running context.Context, f *signalFacts) {
problems = append(problems, "S8: "+err.Error()) problems = append(problems, "S8: "+err.Error())
} }
} }
problems = append(problems, clearRetiredChecks(running, w.keeper, f.retiredChecks)...)
if err := w.keeper.EndSilences(running); err != nil { if err := w.keeper.EndSilences(running); err != nil {
problems = append(problems, err.Error()) problems = append(problems, err.Error())
} }
@@ -323,8 +315,6 @@ func (w *watchdogs) gather(ctx context.Context) *signalFacts {
} else { } else {
f.standings = providerStandings(open) f.standings = providerStandings(open)
f.providerWords = providerConditions(open) f.providerWords = providerConditions(open)
f.applyFailingOpen = applyFailingOpen(open)
f.retiredChecks = retiredCheckConditions(open, liveChecks())
} }
f.advisories = link.Advisories.Since(now.Add(-advisoryQuiet)) f.advisories = link.Advisories.Since(now.Add(-advisoryQuiet))
f.lostConsumers, f.advisoriesErr = w.lostConsumers(ctx, f.advisories) f.lostConsumers, f.advisoriesErr = w.lostConsumers(ctx, f.advisories)
@@ -468,7 +458,7 @@ func (w *watchdogs) gatherLoop() (loopFacts, error) {
if w.js == nil { if w.js == nil {
return loopFacts{took: link.Loop.Last()}, errors.New("this controller is not on the bus") return loopFacts{took: link.Loop.Last()}, errors.New("this controller is not on the bus")
} }
return readLoop(link.Loop.Last(), link.Loop.Due(), loopConsumers(), func(c broker.Consumer) (uint64, error) { return readLoop(link.Loop.Last(), loopConsumers(), func(c broker.Consumer) (uint64, error) {
info, err := w.js.Context().ConsumerInfo(c.Stream, c.Name) info, err := w.js.Context().ConsumerInfo(c.Stream, c.Name)
if err != nil { if err != nil {
return 0, err return 0, err
@@ -478,8 +468,8 @@ func (w *watchdogs) gatherLoop() (loopFacts, error) {
} }
// readLoop sums what the given consumers hold, as held reads each. // readLoop sums what the given consumers hold, as held reads each.
func readLoop(took, due time.Time, consumers []broker.Consumer, held func(broker.Consumer) (uint64, error)) (loopFacts, error) { func readLoop(took time.Time, consumers []broker.Consumer, held func(broker.Consumer) (uint64, error)) (loopFacts, error) {
f := loopFacts{took: took, due: due} f := loopFacts{took: took}
var where []string var where []string
for _, c := range consumers { for _, c := range consumers {
n, err := held(c) n, err := held(c)
+1 -1
View File
@@ -3,7 +3,7 @@ module github.com/novox/mesh-controller
go 1.26.0 go 1.26.0
require ( require (
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030174137-a37181261c24 git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030100935-c2ec8b6b2fde
github.com/jackc/pgx/v5 v5.10.0 github.com/jackc/pgx/v5 v5.10.0
github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0 github.com/nats-io/nats.go v1.54.0
+6 -2
View File
@@ -1,7 +1,11 @@
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 h1:E5stfIeCN6uYgA9TCv+4K8JgaURg3zQppvUgJLaDMP4=
git.novox.be/novox/mesh-host v0.0.0-20261018171808-9d9e3a46c4f5/go.mod h1:4wyoCr6IIIIarEUuROThWDj7DAvJvIidqMjWcpSnOwE= 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.20261030174137-a37181261c24 h1:s0XXZtR7CFPBbnaDygWnE4/Q827iEFt0YfEzujk4Q9g= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029233535-78c9ec881771 h1:Tt16HJk8JZRce1i3JzzDSNdCWOrv7dw8FQ/bckNQ5Ig=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030174137-a37181261c24/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029233535-78c9ec881771/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030095816-530d8d3007dc h1:Wdjl2gomooHswJ5qpkwZPq/ureZgXVw6FMCsgzPJhPo=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030095816-530d8d3007dc/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030100935-c2ec8b6b2fde h1:eKXSllHCS31GgMazfvZ7BeiLoXIUibnUmzxWVkj+Vcw=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030100935-c2ec8b6b2fde/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 h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= 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= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+2 -11
View File
@@ -53,16 +53,6 @@ func Dir(t testing.TB, repository string) string {
t.Logf("judged against %s as cloned beside this check", dir) t.Logf("judged against %s as cloned beside this check", dir)
return 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) from := filepath.Join(Captured(), repository)
if _, err := os.Stat(from); err != nil { if _, err := os.Stat(from); err != nil {
t.Fatalf("no captured copy of %s at %s: %v", repository, from, err) t.Fatalf("no captured copy of %s at %s: %v", repository, from, err)
@@ -91,7 +81,8 @@ func CapturedDir(t testing.TB, repository string) string {
if err != nil { if err != nil {
t.Fatalf("the captured copy of %s could not be laid out: %v", repository, err) 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)", repository) 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)
return dir return dir
} }
+17
View File
@@ -41,6 +41,23 @@ func ControllerAnnouncement(node string) announce.States {
return s 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 // 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. // readers' `reads` as before, until they announce.
func NotAnnounced(r Records) []string { func NotAnnounced(r Records) []string {
+17
View File
@@ -7,6 +7,23 @@ import (
"git.novox.be/novox/mesh-sdk/go/announce" "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 // 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 // declared packages the pacman module reads among them, ADR 0507 §5), the chain's buckets, the checks running and the
// plans building. // plans building.
-39
View File
@@ -1,39 +0,0 @@
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)
}
}
}
+22 -34
View File
@@ -339,36 +339,13 @@ func Overlaps() []string {
// ack subject is derived from (nats.go: `$JS.ACK.<stream>.controller.>`). // ack subject is derived from (nats.go: `$JS.ACK.<stream>.controller.>`).
const ControllerName = "controller" const ControllerName = "controller"
// MergesConsumerName was the controller's durable consumer of the forge's merged event on EVENTS (novox/hq ADR 0407 // MergesConsumerName is the controller's durable consumer of the forge's merged event on EVENTS, apart from the
// decision 3, issue 973), with no max-deliver and one message at a time. **Retired** (sweep 2026-10-30, issue 774): // events consumer (novox/hq ADR 0407 decision 3, issue 973). A merge is never dropped, so this consumer has no
// the forge bridge's stream FORGE is the record of the merges, and the controller reads it where it needs a merge. On // max-deliver, which would send a merge held five times to DEAD_LETTERS to wait for a person, and no healer
// 2026-10-29 one announcement too large for the controller's bucket of merges was held on it for over a day, with // resets it, which would drop the merges it holds. A consumer of its own because both of those are properties of
// 163 merges queued behind it, and the deaf watchdog read the queue as a deaf controller. RetiredConsumers removes it. // a consumer, not of a subject.
const MergesConsumerName = "controller-merges" 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 // ControllerSeat is the role the control plane holds, and ControllerStates are the facts it states
// under it (novox/hq ADR 0134). // under it (novox/hq ADR 0134).
// //
@@ -536,6 +513,23 @@ func ControllerConsumers() []Consumer {
Why: "the controller is the single consumer of what nodes say; explicit ack and no " + 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", "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, Name: ControllerName,
Stream: "EVENTS", Stream: "EVENTS",
@@ -587,9 +581,6 @@ func NoticesConsumer() Consumer {
// one on DEAD_LETTER_NOTICES, which the keeper of dead letters asserts and retries by itself, so a fault // 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). // there never stops the controller serving (novox/hq issue 330).
func AssertServingConsumers(e Ensurer) error { func AssertServingConsumers(e Ensurer) error {
if err := removeRetired(e); err != nil {
return err
}
for _, c := range MeshConsumers() { for _, c := range MeshConsumers() {
if c.Stream == DeadLetterNoticesStream { if c.Stream == DeadLetterNoticesStream {
continue continue
@@ -612,9 +603,6 @@ type Ensurer interface {
// After the streams, necessarily: a consumer on a stream that does not exist is refused, and the // 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. // refusal names the stream rather than the order.
func AssertMeshConsumers(e Ensurer) error { func AssertMeshConsumers(e Ensurer) error {
if err := removeRetired(e); err != nil {
return err
}
for _, c := range MeshConsumers() { for _, c := range MeshConsumers() {
if err := e.EnsureConsumer(c); err != nil { if err := e.EnsureConsumer(c); err != nil {
return fmt.Errorf("asserting consumer %s on %s: %w", c.Name, c.Stream, err) return fmt.Errorf("asserting consumer %s on %s: %w", c.Name, c.Stream, err)
+1 -18
View File
@@ -181,11 +181,6 @@ func Users(r Records) ([]Principal, error) {
} }
} }
for _, d := range r.Assigned[node] { 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 // **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 // 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 // itself has an account, and is never carried by the machine's runtime, which runs as the
@@ -232,7 +227,7 @@ func Users(r Records) ([]Principal, error) {
if runtimeHere { if runtimeHere {
var carried []Declared var carried []Declared
for _, d := range r.Assigned[node] { for _, d := range r.Assigned[node] {
if CarriedByRunner(d) { if d.RunsAs == "" && runnerServes(d) {
// Never a restricted stream for what the tool runner carries (novox/hq ADR 0376 §1). // Never a restricted stream for what the tool runner carries (novox/hq ADR 0376 §1).
d.Hears = ComposedHears(d) d.Hears = ComposedHears(d)
carried = append(carried, d) carried = append(carried, d)
@@ -459,15 +454,3 @@ func declaredStreamsOf(r Records) []string {
sort.Strings(out) sort.Strings(out)
return 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) }
+10 -34
View File
@@ -6,9 +6,7 @@ import (
"git.novox.be/novox/mesh-sdk/go/announce" "git.novox.be/novox/mesh-sdk/go/announce"
"slices" "slices"
"strings" "strings"
"sync"
"git.novox.be/novox/mesh-sdk/go/buscompose"
"git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/catalogue"
"github.com/novox/mesh-controller/internal/broker" "github.com/novox/mesh-controller/internal/broker"
) )
@@ -121,18 +119,22 @@ 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 // 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 // before. Announcements that cannot be read are said, and every read is granted as before: the bus's users are
// composed either way. // 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 { if Announcements != nil {
announced, err := Announcements(ctx) announced, err := Announcements(ctx)
if err != nil { 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) fmt.Printf("the states the modules announced cannot be read, so every read is granted from `reads` as before: %v\n", err)
} else { } else {
out.Announced = announced out.Announced = announced
judgeReads(out.Assigned, announced, sayRefusedOnce) 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)
}
}
}
} }
} }
@@ -613,29 +615,3 @@ func hearsOf(m catalogue.Manifest, declared map[string]catalogue.Manifest, node
} }
return out, touches, onEvents 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
@@ -1,44 +0,0 @@
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)
}
}
}
-11
View File
@@ -26,17 +26,6 @@ type ProvReport struct {
WaitsFor []string `json:"waits-for,omitempty"` WaitsFor []string `json:"waits-for,omitempty"`
Templates string `json:"templates,omitempty"` Templates string `json:"templates,omitempty"`
At time.Time `json:"at"` 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. // The words a client's report says.
+2 -3
View File
@@ -127,10 +127,9 @@ var (
// begin. // begin.
var PlacementKinds = []string{"dir", "access", "port", "seat", "machine", "backup"} var PlacementKinds = []string{"dir", "access", "port", "seat", "machine", "backup"}
// credentialForms are the forms a provider's offer may declare for its credential (issue 768); "none" is an offer with // credentialForms are the forms a provider's offer may declare for its credential (issue 768).
// no credential, which the release's catalogue states on every such offer (issue 774).
var credentialForms = []string{catalogue.CredentialFormShared, catalogue.CredentialFormPair, var credentialForms = []string{catalogue.CredentialFormShared, catalogue.CredentialFormPair,
catalogue.CredentialFormIssued, catalogue.CredentialFormGiven, catalogue.CredentialFormNone} catalogue.CredentialFormIssued, catalogue.CredentialFormGiven}
var ( var (
digestShape = regexp.MustCompile(`^sha256:[0-9a-f]{64}$`) digestShape = regexp.MustCompile(`^sha256:[0-9a-f]{64}$`)
+2 -4
View File
@@ -316,16 +316,14 @@ func TestAPlanNamesTheCredentialFormOfEachProvidersOffer(t *testing.T) {
p := aPlan() p := aPlan()
p.Provisions = []ProvisionedCredential{ p.Provisions = []ProvisionedCredential{
{Provision: "postgres-database", Provider: "novox", ProviderModule: "postgres", CredentialForm: "pair"}, {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}) values, err := AssignmentValues(assigned, map[string]Provisioning{"g14.dunst": p})
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
if want := `{"provision":"postgres-database","provider":"novox","provider-module":"postgres",` + if want := `{"provision":"postgres-database","provider":"novox","provider-module":"postgres",` +
`"credential-form":"pair"},{"provision":"sonarr-api","provider":"ace","provider-module":"sonarr",` + `"credential-form":"pair"},{"provision":"sonarr-api","provider":"ace","provider-module":"sonarr",` +
`"credential-form":"shared"},{"provision":"artifact-store","provider":"novox","provider-module":"distribution",` + `"credential-form":"shared"}`; !strings.Contains(string(values["g14.dunst"]), want) {
`"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) t.Errorf("the plan's value is %s, want it to carry %s", values["g14.dunst"], want)
} }
bad := aPlan() bad := aPlan()
+36 -98
View File
@@ -127,13 +127,9 @@ type CallLog struct {
// keeper keeps every call beyond this process, when Durably was given one; writes go through // 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. // one goroutine, in order, so a call's last state is the one kept.
keeper CallKeeper keeper CallKeeper
holder string holder string
// queued are the calls' states waiting for the keeper, one per call, in the order each call was first queued, under writes chan Call
// 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 logger *log.Logger
lost int // writes the keeper could not take, said once each lost int // writes the keeper could not take, said once each
keepErr error keepErr error
@@ -169,8 +165,8 @@ const keepTries = 5
func (l *CallLog) Durably(ctx context.Context, keeper CallKeeper, holder string, logger *log.Logger) error { func (l *CallLog) Durably(ctx context.Context, keeper CallKeeper, holder string, logger *log.Logger) error {
l.mu.Lock() l.mu.Lock()
l.keeper, l.holder, l.logger = keeper, holder, logger l.keeper, l.holder, l.logger = keeper, holder, logger
if l.woken == nil { if l.writes == nil {
l.woken = make(chan struct{}, 1) l.writes = make(chan Call, 256)
go l.keepWrites() go l.keepWrites()
} }
l.mu.Unlock() l.mu.Unlock()
@@ -207,109 +203,51 @@ func orSomebody(s string) string {
return s 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 // 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. // that is not taking writes, and that is said rather than waited on.
func (l *CallLog) keep(c Call) { func (l *CallLog) keep(c Call) {
if l.woken == nil { if l.writes == 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 return
} }
select { select {
case l.woken <- struct{}{}: case l.writes <- c:
default: 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() { func (l *CallLog) keepWrites() {
for range l.woken { for c := range l.writes {
l.qmu.Lock() l.mu.Lock()
waiting := l.queued.take() gate := l.epoch
l.qmu.Unlock() l.mu.Unlock()
for _, c := range waiting { if gate != nil {
l.keepOne(c) 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)
}
// keepOne writes one call's state through the keeper, under the lease. continue
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
var err error for try := 0; try < keepTries; try++ {
for try := 0; try < keepTries; try++ { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) err = l.keeper.Keep(ctx, c)
err = l.keeper.Keep(ctx, c) cancel()
cancel() if err == nil {
if err == nil { break
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)
} }
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)
} }
} }
+4 -32
View File
@@ -254,7 +254,8 @@ 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. // 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) { func TestAJoinTokenIsShownToItsCallerAndNotKept(t *testing.T) {
l, a := NewCallLog(), newAnswers(t) l, a := NewCallLog(), newAnswers(t)
l.woken = make(chan struct{}, 1) // queued for the keeper, and nothing takes them: read below writes := make(chan Call, 4)
l.writes = writes
l.serveCall("mesh-controller", "token", json.RawMessage(`{"new":"laptop"}`), "_INBOX.x.1", 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 }, func(context.Context, json.RawMessage) (any, error) { return "token for laptop: s3cret-join", nil },
a.respond, nil) a.respond, nil)
@@ -265,7 +266,8 @@ func TestAJoinTokenIsShownToItsCallerAndNotKept(t *testing.T) {
if len(recent) != 1 || strings.Contains(string(recent[0].Answer), "s3cret-join") { if len(recent) != 1 || strings.Contains(string(recent[0].Answer), "s3cret-join") {
t.Fatalf("the token was kept in memory: %+v", recent) t.Fatalf("the token was kept in memory: %+v", recent)
} }
for _, c := range l.queued.take() { close(writes)
for c := range writes {
if strings.Contains(string(c.Answer), "s3cret-join") { if strings.Contains(string(c.Answer), "s3cret-join") {
t.Fatalf("the token was sent to be kept on the bus: %s", c.Answer) t.Fatalf("the token was sent to be kept on the bus: %s", c.Answer)
} }
@@ -274,33 +276,3 @@ func TestAJoinTokenIsShownToItsCallerAndNotKept(t *testing.T) {
// A verb whose handler panics answers an error, and the controller serves the next call (review of // 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). // 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")
}
}
+3 -2
View File
@@ -30,8 +30,9 @@ var Contracts = map[string]Contract{
KindBuilt: {Unordered: "a build's outcome on the retired build seat's subjects: dropped, nothing takes a build in " + 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 " + "from the bus any more (R4g, ADR 0501 §1); the subjects stay followed while the installer's carried user list " +
"names them"}, "names them"},
KindSourceMoved: {Unordered: "the forge's merge on EVENTS: no longer followed, dropped if one arrives; the forge " + KindSourceMoved: {Ordered: "by the merge's commit against what was built from it: a merge already acted on " +
"bridge's stream FORGE is the record of the merges, read at a moved module's registration (sweep 2026-10-30)"}, "or older than the last look is history, not a second plan (issues 250, 266)",
Tests: []string{"TestAMergeOlderThanTheLastLookIsHistory", "TestAMergeMatchesTheSourcesBuiltFromIt"}},
KindToolsHeartbeat: {Unordered: "a word that the node tools are there, as a machine's heartbeat"}, 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 " + 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 " + "than the one kept for that machine is refused, so an event arriving after a newer report does not undo it " +
+4 -2
View File
@@ -125,7 +125,8 @@ func askCLI(t *testing.T, conn *nats.Conn, node string, asked CLIAsked) (CLIAnsw
// bus, where `calls` answers anyone who may call the seat. // bus, where `calls` answers anyone who may call the seat.
func TestAMeshCLIRecordKeepsNeitherItsLineNorItsAnswerOnTheBus(t *testing.T) { func TestAMeshCLIRecordKeepsNeitherItsLineNorItsAnswerOnTheBus(t *testing.T) {
l, a := NewCallLog(), newAnswers(t) l, a := NewCallLog(), newAnswers(t)
l.woken = make(chan struct{}, 1) // queued for the keeper, and nothing takes them: read below writes := make(chan Call, 4)
l.writes = writes
asked, _ := json.Marshal(CLIAsked{Line: []string{"settings", "set", "x", `{"password":"s3cret"}`}, Account: "op"}) asked, _ := json.Marshal(CLIAsked{Line: []string{"settings", "set", "x", `{"password":"s3cret"}`}, Account: "op"})
l.serveCall(CLISeat, "laptop", asked, "_INBOX.node.laptop.abcdefghijklmnopqrstuv", l.serveCall(CLISeat, "laptop", asked, "_INBOX.node.laptop.abcdefghijklmnopqrstuv",
func(context.Context, json.RawMessage) (any, error) { func(context.Context, json.RawMessage) (any, error) {
@@ -133,7 +134,8 @@ func TestAMeshCLIRecordKeepsNeitherItsLineNorItsAnswerOnTheBus(t *testing.T) {
}, },
a.respond, nil) a.respond, nil)
_ = a.only() _ = a.only()
for _, c := range l.queued.take() { close(writes)
for c := range writes {
if carries(c.Args, "s3cret") || carries(c.Answer, "s3cret-join") { if carries(c.Args, "s3cret") || carries(c.Answer, "s3cret-join") {
t.Fatalf("the bus was sent %s / %s", c.Args, c.Answer) t.Fatalf("the bus was sent %s / %s", c.Args, c.Answer)
} }
-3
View File
@@ -52,9 +52,6 @@ type MoveSecretAnswer struct {
Secret string `json:"secret,omitempty"` Secret string `json:"secret,omitempty"`
Generation int `json:"generation,omitempty"` Generation int `json:"generation,omitempty"`
Unchanged bool `json:"unchanged,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. // MoveSecretWithin is how long the controller waits for one move: the engine asks the vault twice.
+2 -3
View File
@@ -30,9 +30,8 @@ func TestTheMovesAskAndAnswerKeepTheirFieldNames(t *testing.T) {
"provision sealed sealed_to" { "provision sealed sealed_to" {
t.Fatalf("the ask's fields are %q", got) t.Fatalf("the ask's fields are %q", got)
} }
answer := MoveSecretAnswer{Said: "s", Refused: "r", Secret: "own/a/b/c", Generation: 1, Unchanged: true, 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 unchanged" {
if got := keys(answer); got != "generation refused said secret superseded unchanged" {
t.Fatalf("the answer's fields are %q", got) t.Fatalf("the answer's fields are %q", got)
} }
if broker.AskMoveSecretSubject("laptop") != "mesh.node.laptop.ask.move-secret" || if broker.AskMoveSecretSubject("laptop") != "mesh.node.laptop.ask.move-secret" ||
+20 -6
View File
@@ -62,7 +62,7 @@ func Nats(js *broker.JetStream) Inbound {
// whatever was asked for — and not at all when nothing was. // whatever was asked for — and not at all when nothing was.
func (n *natsInbound) Also(kind string) error { func (n *natsInbound) Also(kind string) error {
switch kind { switch kind {
case KindModuleMoved, KindCatchUp, KindProvisioner, KindPullUpdated, KindDecided, case KindModuleMoved, KindCatchUp, KindSourceMoved, KindProvisioner, KindPullUpdated, KindDecided,
KindDeliveryMerged: KindDeliveryMerged:
n.follows[kind] = true n.follows[kind] = true
return nil return nil
@@ -145,6 +145,24 @@ func (n *natsInbound) Receive(ctx context.Context, act func(context.Context, Con
defer func() { _ = followed.Unsubscribe() }() 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 // A connection that dropped is said, not discovered. A controller whose bus connection is gone
// is a mesh where nothing can be told anything. // is a mesh where nothing can be told anything.
gone := make(chan error, 1) gone := make(chan error, 1)
@@ -405,11 +423,7 @@ func (m *natsControl) Hold(after time.Duration) error {
if _, already := m.on.since[m.seq]; !already { if _, already := m.on.since[m.seq]; !already {
m.on.since[m.seq] = time.Now() m.on.since[m.seq] = time.Now()
} }
if err := m.msg.NakWithDelay(after); err != nil { return m.msg.NakWithDelay(after)
return err
}
Loop.HeldUntil(time.Now().Add(after))
return nil
} }
func (m *natsControl) forget() { func (m *natsControl) forget() {
@@ -394,3 +394,35 @@ func TestNatsAReportThatIsNotNewsIsNotStated(t *testing.T) {
case <-time.After(3 * time.Second): 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,6 +2,7 @@ package link
import ( import (
"context" "context"
"errors"
"io" "io"
"log" "log"
"sync" "sync"
@@ -198,3 +199,46 @@ func (s slowly) Heard(context.Context, Report) (bool, error) { s.work(); return
type sameAgain struct{} type sameAgain struct{}
func (sameAgain) Heard(context.Context, Report) (bool, error) { return false, nil } 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])
}
}
}
+61 -3
View File
@@ -49,6 +49,8 @@ type Upgrader interface {
// stop every upgrade behind it — except the store unreachable for the moment, which is asked // stop every upgrade behind it — except the store unreachable for the moment, which is asked
// again for a bounded time (novox/hq issue 083). // again for a bounded time (novox/hq issue 083).
Upgraded(ctx context.Context, u Upgraded) error 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 // Checker is what the controller does when the forge says a pull request's head moved: ask for it to be
@@ -103,6 +105,9 @@ const GiveUpAfter = 2 * time.Minute
// Follows says what to do about upgrades, and asks for them to be delivered. // Follows says what to do about upgrades, and asks for them to be delivered.
func (s *Server) Follows(u Upgrader) error { 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 { if err := s.inbound.Also(KindModuleMoved); err != nil {
return err return err
} }
@@ -220,9 +225,7 @@ func (s *Server) act(ctx context.Context, m Control) {
case KindModuleMoved: case KindModuleMoved:
s.moved(ctx, m) s.moved(ctx, m)
case KindSourceMoved: case KindSourceMoved:
// The forge's merges are no longer followed (sweep 2026-10-30): the forge bridge's stream is their record, read s.sourceMoved(ctx, m)
// where a merge is needed (keptForgeMerge). One that arrives on an old consumer is let go of.
_ = m.Drop()
case KindCatchUp: case KindCatchUp:
s.catchingUp(ctx, m) s.catchingUp(ctx, m)
case KindProvisioner: case KindProvisioner:
@@ -518,6 +521,61 @@ func short(commit string) string {
return commit 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 // 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. // asked is said here, and the next push to the pull request asks again.
func (s *Server) pullUpdated(ctx context.Context, m Control) { func (s *Server) pullUpdated(ctx context.Context, m Control) {
+60
View File
@@ -0,0 +1,60 @@
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)
}
}
}
+60 -1
View File
@@ -17,7 +17,8 @@ func (r recordsWith) Built(context.Context, BuildResult) error { return r.err }
type upgradesWith struct{ err error } type upgradesWith struct{ err error }
func (u upgradesWith) Upgraded(context.Context, Upgraded) error { return u.err } func (u upgradesWith) Upgraded(context.Context, Upgraded) error { return u.err }
func (u upgradesWith) SourceMoved(context.Context, SourceMoved) error { return u.err }
type replaysWith struct{ err error } type replaysWith struct{ err error }
@@ -68,3 +69,61 @@ 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,26 +85,6 @@ func nodeOfToolsAlive(subject string) (string, bool) {
type LoopActivity struct { type LoopActivity struct {
mu sync.Mutex mu sync.Mutex
took time.Time 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. // Loop is this process's event loop.
+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 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. a second place to clean besides the catalogue itself.
mesh-catalog f3a57dcc56fb6c198095cafd11d0decff9f720c4 modules/*/module.json, modules/*/files, modules/*/templates, predefined.json mesh-catalog 3a1017f5594217092a654dd1e2d72226766bc16a modules/*/module.json, modules/*/files, modules/*/templates, predefined.json
mesh-host 9d9e3a46c4f5943cfffb413d962ce48b75e2fd39 examples/foundation-first-node.lock mesh-host 9d9e3a46c4f5943cfffb413d962ce48b75e2fd39 examples/foundation-first-node.lock
mesh-sdk a37181261c24a15b8092c06e11cce63e0e5f8ade conformance/events/module-event.json mesh-sdk c2ec8b6b2fdece8dd76a9d2d1f29d3ae4579f174 conformance/events/module-event.json
To move them, from this repository's root, with the two repositories checked out beside it: To move them, from this repository's root, with the two repositories checked out beside it:
@@ -1 +0,0 @@
${seat:mesh-artifact-store:reach}
@@ -1,52 +1,26 @@
{ {
"module": "artifact-store-tools", "module": "artifact-store-tools",
"version": "1", "version": "1",
"claims": [
{
"name": "mesh-artifact-sweep",
"scope": "mesh",
"serves": [
"artifacts",
"held",
"collect",
"mirrors",
"images"
]
}
],
"requires": [
"artifact-store"
],
"invokes": [ "invokes": [
"seat:mesh-hand-acts.hand-act", "seat:mesh-controller.artifacts",
"seat:mesh-controller.collect",
"mesh-catalog.catalog_modules", "mesh-catalog.catalog_modules",
"mesh-catalog.catalog_module" "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": [ "tools": [
"artifact_store_repositories", "artifact_store_repositories",
"artifact_store_usage", "artifact_store_usage",
"artifact_store_references", "artifact_store_references",
"artifact_store_collect" "artifact_store_collect"
], ],
"config": { "config": {
"older-store-names": { "older-store-names": {
"type": "string", "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).", "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.:-]+)*", "pattern": "none|[A-Za-z0-9.:-]+( [A-Za-z0-9.:-]+)*",
"default": "none" "default": "none"
} }
}, },
"resources": [ "resources": [
{ {
"id": "state", "id": "state",
@@ -62,14 +36,6 @@
"mode": "0600", "mode": "0600",
"trusted": false, "trusted": false,
"template": "files/store-names" "template": "files/store-names"
},
{
"id": "store-address",
"type": "file",
"path": "${dir:state}/store-address",
"mode": "0600",
"trusted": false,
"template": "files/store-address"
} }
], ],
"build": { "build": {
@@ -86,8 +52,7 @@
], ],
"env": { "env": {
"MESH_ARTIFACT_STORE_CONTAINER": "mesh-registry", "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,10 +8,7 @@
"provides": [ "provides": [
{ {
"name": "rootless-image-build", "name": "rootless-image-build",
"reach": "machine", "reach": "machine"
"credential": {
"form": "none"
}
} }
], ],
"resources": [ "resources": [
@@ -7,9 +7,7 @@
"disk_filesystems", "disk_filesystems",
"disk_folders", "disk_folders",
"disk_devices", "disk_devices",
"cpu_top", "cpu_top"
"disk_readings",
"disk_measure"
], ],
"build": { "build": {
"artifacts": [ "artifacts": [
@@ -25,16 +23,5 @@
] ]
} }
] ]
}, }
"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,10 +4,7 @@
"provides": [ "provides": [
{ {
"name": "artifact-store", "name": "artifact-store",
"scope": "mesh", "scope": "mesh"
"credential": {
"form": "none"
}
} }
], ],
"claims": [ "claims": [
@@ -1,15 +1,14 @@
# Written by the mesh (module dnsmasq, novox/hq issue 348). Replaced on every push. # 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 # The resolver binds its addresses once, at start (bind-interfaces), so it starts only once this
# machine's private address is there: a start that finds it missing fails after at most two seconds and # machine's private address is there: it waits up to 80 seconds for it, and a start that found it
# is tried again every ten seconds (Restart=on-failure, no start limit) until the tunnel has brought it. # missing is tried again every ten seconds rather than left failed.
# 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] [Unit]
After=network-online.target
Wants=network-online.target
StartLimitIntervalSec=0 StartLimitIntervalSec=0
[Service] [Service]
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' 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'
Restart=on-failure Restart=on-failure
RestartSec=10s RestartSec=10s
@@ -9,10 +9,7 @@
{ {
"name": "wildcard-resolution", "name": "wildcard-resolution",
"scope": "mesh", "scope": "mesh",
"identity": false, "identity": false
"credential": {
"form": "none"
}
} }
], ],
"requires": [ "requires": [
@@ -7,7 +7,7 @@
"privileged" "privileged"
], ],
"invokes": [ "invokes": [
"seat:mesh-artifact-sweep.images" "seat:mesh-controller.images"
], ],
"claims": [ "claims": [
{ {
@@ -40,7 +40,6 @@
], ],
"emits": [ "emits": [
"repo.created", "repo.created",
"repo.writable",
"issue.opened", "issue.opened",
"pull.merged", "pull.merged",
"pull.updated", "pull.updated",
@@ -182,12 +181,6 @@
"trusted": true, "trusted": true,
"default": {} "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": { "main-mergers": {
"type": "list", "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.", "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.",
@@ -266,8 +259,7 @@
"env": { "env": {
"DB_TYPE": "postgres", "DB_TYPE": "postgres",
"USER_UID": "1000", "USER_UID": "1000",
"USER_GID": "1000", "USER_GID": "1000"
"SSH_MAX_STARTUPS": "100:30:200"
}, },
"env-file": [ "env-file": [
"${dir:state}/server.env" "${dir:state}/server.env"
@@ -324,8 +316,7 @@
"takes-config": [ "takes-config": [
"url", "url",
"keep-main-rules", "keep-main-rules",
"main-mergers", "main-mergers"
"check-gates-merge"
] ]
}, },
{ {
@@ -400,7 +391,6 @@
"provides": [ "provides": [
{ {
"name": "npm-package-registry", "name": "npm-package-registry",
"credential": {"form": "pair"},
"scope": "mesh", "scope": "mesh",
"identity": { "identity": {
"max": 40, "max": 40,
@@ -409,7 +399,6 @@
}, },
{ {
"name": "git", "name": "git",
"credential": {"form": "none"},
"scope": "mesh" "scope": "mesh"
}, },
{ {
@@ -422,7 +411,6 @@
}, },
{ {
"name": "forge-account", "name": "forge-account",
"credential": {"form": "none"},
"scope": "mesh" "scope": "mesh"
} }
], ],
@@ -549,7 +537,6 @@
"MESH_GITEA_ADMIN_PASSWORD_FILE": "${dir:state}/admin.secret", "MESH_GITEA_ADMIN_PASSWORD_FILE": "${dir:state}/admin.secret",
"MESH_GITEA_ORIGINS_FILE": "${dir:mesh-state}/origins", "MESH_GITEA_ORIGINS_FILE": "${dir:mesh-state}/origins",
"MESH_GITEA_SSH_PORT": "222", "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" "MESH_FORGE_HOST_FILE": "/etc/gitea/forge-host"
} }
} }
@@ -567,14 +554,11 @@
"announced" "announced"
], ],
"reads": [ "reads": [
"mesh-provisioner.plans", "mesh-controller.assignments",
"mesh-provisioner.provisions",
"ssh-client.keys", "ssh-client.keys",
"mesh-build-go.keys", "mesh-build-go.keys",
"mesh-build-node.keys", "mesh-build-node.keys",
"mesh-build-image.keys", "mesh-build-image.keys",
"mesh-delivery-checker.owed", "mesh-delivery-checker.owed"
"mesh-delivery-indexer.modules",
"mesh-delivery-gate.source"
] ]
} }
@@ -7,10 +7,7 @@
"provides": [ "provides": [
{ {
"name": "go-toolchain", "name": "go-toolchain",
"reach": "machine", "reach": "machine"
"credential": {
"form": "none"
}
} }
], ],
"resources": [ "resources": [
@@ -7,9 +7,6 @@
"scope": "mesh", "scope": "mesh",
"identity": { "identity": {
"in": "an InfluxDB v1 authorization" "in": "an InfluxDB v1 authorization"
},
"credential": {
"form": "pair"
} }
} }
], ],
@@ -1,4 +0,0 @@
# 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,25 +100,6 @@
"id": "linux-lts-headers", "id": "linux-lts-headers",
"type": "package", "type": "package",
"package": "linux-lts-headers" "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": { "build": {
@@ -9,7 +9,6 @@
"provides": [ "provides": [
{ {
"name": "oidc-client", "name": "oidc-client",
"credential": {"form": "pair"},
"scope": "mesh", "scope": "mesh",
"identity": { "identity": {
"max": 255, "max": 255,
@@ -241,7 +240,6 @@
"bound" "bound"
], ],
"reads": [ "reads": [
"mesh-provisioner.plans", "mesh-controller.assignments"
"mesh-provisioner.provisions"
] ]
} }
@@ -665,14 +665,7 @@
"path": "${dir:mesh-state}/config.json", "path": "${dir:mesh-state}/config.json",
"mode": "0600", "mode": "0600",
"template": "files/config.json", "template": "files/config.json",
"takes-config": [ "takes-config": ["domain", "proxy-address", "sitename", "website", "url", "container"],
"domain",
"proxy-address",
"sitename",
"website",
"url",
"container"
],
"merge": "json" "merge": "json"
}, },
{ {
@@ -765,9 +758,6 @@
"identity": { "identity": {
"max": 64, "max": 64,
"in": "a mailbox's local part" "in": "a mailbox's local part"
},
"credential": {
"form": "pair"
} }
} }
], ],
@@ -1,5 +1,3 @@
{ {
"capacity": 0, "capacity": "shanks=3,ace=2,g14=0,novox=0"
"cores-per-step": 6,
"build-hours": ""
} }
@@ -10,7 +10,7 @@
"state": [ "state": [
"toolchains", "toolchains",
"keys", "keys",
{"name": "steps", "ttl-seconds": 172800}, "pauses" "steps"
], ],
"claims": [ "claims": [
{ {
@@ -20,27 +20,14 @@
} }
], ],
"reads": [ "reads": [
"mesh-delivery-checker.kept", "mesh-delivery-checker.kept"
"mesh-delivery-indexer.manifests"
], ],
"config": { "config": {
"capacity": { "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", "type": "string",
"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.", "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": "|([01][0-9]|2[0-3]):[0-5][0-9]-([01][0-9]|2[0-3]):[0-5][0-9]", "pattern": "|[a-z0-9_-]+=[0-9]+(,[a-z0-9_-]+=[0-9]+)*",
"default": "" "default": "shanks=3,ace=2,g14=0,novox=0"
},
"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": [ "tools": [
@@ -51,7 +38,6 @@
"build_go_release", "build_go_release",
"build_go_step", "build_go_step",
"build_go_steps", "build_go_steps",
"build_go_step_end",
"build_go_toolchains", "build_go_toolchains",
"build_go_status", "build_go_status",
"build_go_health" "build_go_health"
@@ -119,7 +105,7 @@
"path": "${dir:state}/settings.json", "path": "${dir:state}/settings.json",
"mode": "0600", "mode": "0600",
"owner": "mesh-build-go", "owner": "mesh-build-go",
"takes-config": ["capacity", "cores-per-step", "build-hours"], "takes-config": ["capacity"],
"merge": "json", "merge": "json",
"template": "files/settings.json" "template": "files/settings.json"
}, },
@@ -1,5 +1,3 @@
{ {
"capacity": 0, "capacity": "shanks=3,ace=2,g14=0,novox=0"
"cores-per-step": 2,
"build-hours": ""
} }
@@ -9,7 +9,7 @@
], ],
"state": [ "state": [
"keys", "keys",
{"name": "steps", "ttl-seconds": 172800}, "pauses" "steps"
], ],
"claims": [ "claims": [
{ {
@@ -20,22 +20,10 @@
], ],
"config": { "config": {
"capacity": { "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", "type": "string",
"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.", "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": "|([01][0-9]|2[0-3]):[0-5][0-9]-([01][0-9]|2[0-3]):[0-5][0-9]", "pattern": "|[a-z0-9_-]+=[0-9]+(,[a-z0-9_-]+=[0-9]+)*",
"default": "" "default": "shanks=3,ace=2,g14=0,novox=0"
},
"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": [ "tools": [
@@ -45,7 +33,6 @@
"build_image_release", "build_image_release",
"build_image_step", "build_image_step",
"build_image_steps", "build_image_steps",
"build_image_step_end",
"build_image_status", "build_image_status",
"build_image_health" "build_image_health"
], ],
@@ -90,7 +77,7 @@
"path": "${dir:state}/settings.json", "path": "${dir:state}/settings.json",
"mode": "0600", "mode": "0600",
"owner": "mesh-build-image", "owner": "mesh-build-image",
"takes-config": ["capacity", "cores-per-step", "build-hours"], "takes-config": ["capacity"],
"merge": "json", "merge": "json",
"template": "files/settings.json" "template": "files/settings.json"
}, },
@@ -1,5 +1,3 @@
{ {
"capacity": 0, "capacity": "shanks=3,ace=2,g14=0,novox=0"
"cores-per-step": 2,
"build-hours": ""
} }
@@ -10,7 +10,7 @@
"state": [ "state": [
"toolchains", "toolchains",
"keys", "keys",
{"name": "steps", "ttl-seconds": 172800}, "pauses" "steps"
], ],
"claims": [ "claims": [
{ {
@@ -21,22 +21,10 @@
], ],
"config": { "config": {
"capacity": { "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", "type": "string",
"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.", "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": "|([01][0-9]|2[0-3]):[0-5][0-9]-([01][0-9]|2[0-3]):[0-5][0-9]", "pattern": "|[a-z0-9_-]+=[0-9]+(,[a-z0-9_-]+=[0-9]+)*",
"default": "" "default": "shanks=3,ace=2,g14=0,novox=0"
},
"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": [ "tools": [
@@ -46,7 +34,6 @@
"build_node_release", "build_node_release",
"build_node_step", "build_node_step",
"build_node_steps", "build_node_steps",
"build_node_step_end",
"build_node_toolchains", "build_node_toolchains",
"build_node_status", "build_node_status",
"build_node_health" "build_node_health"
@@ -112,7 +99,7 @@
"path": "${dir:state}/settings.json", "path": "${dir:state}/settings.json",
"mode": "0600", "mode": "0600",
"owner": "mesh-build-node", "owner": "mesh-build-node",
"takes-config": ["capacity", "cores-per-step", "build-hours"], "takes-config": ["capacity"],
"merge": "json", "merge": "json",
"template": "files/settings.json" "template": "files/settings.json"
}, },
@@ -25,7 +25,9 @@
"secrets": { "secrets": {
"postgres-database": "${dir:state}/database.secret" "postgres-database": "${dir:state}/database.secret"
}, },
"consumes": [], "consumes": [
"mesh-controller.built-before"
],
"emits": [ "emits": [
"registered", "registered",
"upgraded", "upgraded",
@@ -90,12 +92,5 @@
"from": "files" "from": "files"
} }
] ]
}, }
"invokes": [
"mesh-delivery-builder.builder_builds_since"
],
"reads": [
"mesh-delivery-builder.builds",
"mesh-delivery-indexer.manifests"
]
} }
@@ -1,87 +0,0 @@
{
"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"
]
}
]
}
}
@@ -7,10 +7,7 @@
"name": "mesh-config", "name": "mesh-config",
"scope": "mesh", "scope": "mesh",
"serves": [ "serves": [
"config", "config"
"compare",
"take-over",
"hand-back"
] ]
} }
], ],
@@ -26,30 +23,14 @@
"gitea.forge.pull-request.closed" "gitea.forge.pull-request.closed"
], ],
"invokes": [ "invokes": [
"seat:mesh-hand-acts.hand-act",
"seat:mesh-hand-acts.warranted",
"seat:mesh-controller.hand-act", "seat:mesh-controller.hand-act",
"seat:mesh-controller.warranted", "seat:mesh-controller.warranted",
"gitea.gitea_get_pull_request" "gitea.gitea_get_pull_request"
], ],
"reads": [ "reads": [
"mesh-delivery-assigner.placement", "mesh-controller.assignments",
"mesh-delivery-assigner.wants",
"mesh-delivery-builder.builds",
"mesh-delivery-indexer.modules",
"mesh-delivery-indexer.manifests",
"mesh-controller.predefined",
"mesh-controller.declared-config", "mesh-controller.declared-config",
"mesh-delivery-indexer.predefined", "mesh-controller.predefined"
"mesh-delivery-indexer.config-candidates",
"mesh-controller.lease"
],
"tools": [
"config_compare",
"config_take_over",
"config_hand_back",
"config_declared",
"config_resync"
], ],
"state": [ "state": [
"values", "values",
@@ -57,9 +38,7 @@
"predefined", "predefined",
"proposals", "proposals",
"pulls", "pulls",
"kept", "kept"
"declared",
"takeover"
], ],
"logs": [ "logs": [
"history" "history"
@@ -1,5 +1,4 @@
{ {
"refusal-stall-floor-minutes": 10, "refusal-stall-floor-minutes": 10,
"author-new-head-minutes": 120, "author-new-head-minutes": 120
"check-gates-merge": true
} }
@@ -14,8 +14,7 @@
"table", "table",
"stalled", "stalled",
"recheck", "recheck",
"close", "close"
"settle"
] ]
} }
], ],
@@ -27,17 +26,13 @@
"gitea.forge.pull-request.edited", "gitea.forge.pull-request.edited",
"gitea.forge.pull-request.closed", "gitea.forge.pull-request.closed",
"mesh-delivery-forge-bridge.forge.**", "mesh-delivery-forge-bridge.forge.**",
"mesh-config.config-changed", "mesh-config.config-changed"
"gitea.repo.writable"
], ],
"emits": [ "emits": [
"transition", "transition",
"group", "group",
"merged" "merged"
], ],
"uses": [
"operator-channel"
],
"invokes": [ "invokes": [
"gitea.gitea_commit_statuses", "gitea.gitea_commit_statuses",
"gitea.gitea_contains", "gitea.gitea_contains",
@@ -62,23 +57,14 @@
{ {
"name": "source", "name": "source",
"history": 64 "history": 64
},
"conditions",
{
"name": "turns",
"history": 5
} }
], ],
"reads": [ "reads": [
"mesh-delivery-teller.answers", "mesh-delivery-teller.answers",
"mesh-delivery-checker.verdicts", "mesh-delivery-checker.verdicts"
"mesh-delivery-view.reach"
], ],
"tools": [ "tools": [
"delivery_hand_now", "delivery_status"
"delivery_merge_again",
"delivery_status",
"delivery_turns"
], ],
"config": { "config": {
"refusal-stall-floor-minutes": { "refusal-stall-floor-minutes": {
@@ -96,12 +82,6 @@
"maximum": 10080, "maximum": 10080,
"trusted": false, "trusted": false,
"default": 120 "default": 120
},
"check-gates-merge": {
"type": "boolean",
"description": "Whether mesh/check gates the merge. True (the default): the operator's Approve is asked, and a merge handed, only on a passing verdict judged against the trunk's head, and a trunk's move puts a verdict out of date (ADR 0447, ADR 0505 §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\"; developers run the unit and integration levels locally before a pull request): a pull request reviewed at its head is ready, and its Approve is asked once the check at its heads (a group: of all its heads together) has a verdict, a pass or a failure said in the ask, or after 60 minutes with none, said so (the operator, 2026-10-27 23:54: \"if you don't want me to approve it, don't send the approval request\"); an approved merge waits for no verdict; no merge is refused for a check judged on an older trunk; a later verdict is shown on the pull request and the delivery, and asks a pending Approve again in its words; a group still merges all its members in one pass. Set it false together with the gitea module's check-gates-merge, which lets its merge tool merge an approved head without a passing mesh/check.",
"trusted": false,
"default": true
} }
}, },
"resources": [ "resources": [
@@ -119,8 +99,7 @@
"mode": "0600", "mode": "0600",
"takes-config": [ "takes-config": [
"refusal-stall-floor-minutes", "refusal-stall-floor-minutes",
"author-new-head-minutes", "author-new-head-minutes"
"check-gates-merge"
], ],
"merge": "json", "merge": "json",
"template": "files/settings.json" "template": "files/settings.json"
@@ -1,65 +0,0 @@
{
"module": "mesh-hand-acts",
"version": "1",
"runs-as": "mesh-hand-acts",
"claims": [
{
"name": "mesh-hand-acts",
"scope": "mesh",
"serves": [
"hand-act",
"hand-acts",
"warranted"
]
}
],
"reads": [
"mesh-controller.hand-acts",
"messenger.asks"
],
"logs": [
{
"name": "acts",
"max-age-days": 400
}
],
"state": [
"warrants"
],
"own-secrets": {
"broker": "${dir:state}/broker"
},
"secrets-owner": "mesh-hand-acts",
"resources": [
{
"id": "account",
"type": "user",
"name": "mesh-hand-acts",
"shell": "/usr/bin/nologin",
"home": "/var/lib/mesh-hand-acts"
},
{
"id": "state",
"type": "directory",
"mode": "0700",
"owner": "mesh-hand-acts",
"place": ".",
"unknown": "program"
}
],
"build": {
"artifacts": [
{
"name": "tools",
"kind": "bundle",
"language": "go",
"system": "arch",
"from": "cmd/mesh-hand-acts",
"binary": "mesh-hand-acts",
"loads": [
"mesh-hand-acts"
]
}
]
}
}
@@ -1,82 +0,0 @@
{
"module": "mesh-healers",
"version": "1",
"runs-as": "mesh-healers",
"claims": [
{
"name": "mesh-healers",
"scope": "mesh",
"serves": [
"healers",
"heal",
"drill"
]
}
],
"consumes": [
"mesh-conditions.condition-raised",
"mesh-conditions.condition-changed",
"mesh-hand-acts.act-recorded"
],
"invokes": [
"mesh-grants.grants_assert",
"seat:mesh-hand-acts.hand-act",
"seat:mesh-controller.drill"
],
"reads": [
"mesh-controller.lease",
"mesh-conditions.conditions",
"mesh-controller.hand-acts"
],
"state": [
{
"name": "attempts",
"key-ttl": true
},
{
"name": "would",
"key-ttl": true
},
"findings",
{
"name": "wakes",
"key-ttl": true
}
],
"own-secrets": {
"broker": "${dir:state}/broker"
},
"secrets-owner": "mesh-healers",
"resources": [
{
"id": "account",
"type": "user",
"name": "mesh-healers",
"shell": "/usr/bin/nologin",
"home": "/var/lib/mesh-healers"
},
{
"id": "state",
"type": "directory",
"mode": "0700",
"owner": "mesh-healers",
"place": ".",
"unknown": "program"
}
],
"build": {
"artifacts": [
{
"name": "tools",
"kind": "bundle",
"language": "go",
"system": "arch",
"from": "cmd/mesh-healers",
"binary": "mesh-healers",
"loads": [
"mesh-healers"
]
}
]
}
}
@@ -62,13 +62,10 @@
"mesh-catalog modules/mesh-build-node", "mesh-catalog modules/mesh-build-node",
"mesh-catalog modules/mesh-catalog", "mesh-catalog modules/mesh-catalog",
"mesh-catalog modules/mesh-cli", "mesh-catalog modules/mesh-cli",
"mesh-catalog modules/mesh-conditions",
"mesh-catalog modules/mesh-config", "mesh-catalog modules/mesh-config",
"mesh-catalog modules/mesh-delivery", "mesh-catalog modules/mesh-delivery",
"mesh-catalog modules/mesh-delivery-gate", "mesh-catalog modules/mesh-delivery-gate",
"mesh-catalog modules/mesh-files", "mesh-catalog modules/mesh-files",
"mesh-catalog modules/mesh-hand-acts",
"mesh-catalog modules/mesh-healers",
"mesh-catalog modules/mesh-issues", "mesh-catalog modules/mesh-issues",
"mesh-catalog modules/mesh-knowledge", "mesh-catalog modules/mesh-knowledge",
"mesh-catalog modules/mesh-mail", "mesh-catalog modules/mesh-mail",
@@ -78,7 +75,6 @@
"mesh-catalog modules/mesh-provisioner", "mesh-catalog modules/mesh-provisioner",
"mesh-catalog modules/mesh-provisioner-client", "mesh-catalog modules/mesh-provisioner-client",
"mesh-catalog modules/mesh-reads", "mesh-catalog modules/mesh-reads",
"mesh-catalog modules/mesh-self-check",
"mesh-catalog modules/mesh-vault", "mesh-catalog modules/mesh-vault",
"mesh-catalog modules/mesh-watcher", "mesh-catalog modules/mesh-watcher",
"mesh-catalog modules/messenger", "mesh-catalog modules/messenger",
@@ -13,17 +13,12 @@
"move", "move",
"note", "note",
"link", "link",
"milestones",
"milestone-set",
"milestone-move",
"import", "import",
"groom", "groom",
"set", "set",
"tracking", "tracking",
"reserve", "reserve",
"fix-times", "fix-times"
"learn-fixes",
"changes"
] ]
} }
], ],
@@ -38,17 +33,12 @@
"move", "move",
"note", "note",
"link", "link",
"milestones",
"milestone-set",
"milestone-move",
"import", "import",
"groom", "groom",
"set", "set",
"tracking", "tracking",
"reserve", "reserve",
"fix-times", "fix-times"
"learn-fixes",
"changes"
] ]
} }
], ],
@@ -62,9 +52,7 @@
"triaged", "triaged",
"ready", "ready",
"ready-to-verify", "ready-to-verify",
"reserved", "reserved"
"milestone-set",
"milestone-moved"
], ],
"consumes": [ "consumes": [
"mesh-issues.opened", "mesh-issues.opened",
@@ -78,7 +66,6 @@
"seat:operator-channel.notify", "seat:operator-channel.notify",
"gitea.gitea_pull_reviews", "gitea.gitea_pull_reviews",
"gitea.gitea_contains", "gitea.gitea_contains",
"gitea.gitea_get_pull_request",
"records.records_status", "records.records_status",
"records.records_list", "records.records_list",
"records.records_read", "records.records_read",
@@ -94,11 +81,7 @@
}, },
"counters", "counters",
"grooming", "grooming",
"reservations", "reservations"
{
"name": "milestones",
"history": 5
}
], ],
"config": { "config": {
"first-number": { "first-number": {
@@ -174,13 +157,10 @@
"mesh-catalog modules/mesh-build-node", "mesh-catalog modules/mesh-build-node",
"mesh-catalog modules/mesh-catalog", "mesh-catalog modules/mesh-catalog",
"mesh-catalog modules/mesh-cli", "mesh-catalog modules/mesh-cli",
"mesh-catalog modules/mesh-conditions",
"mesh-catalog modules/mesh-config", "mesh-catalog modules/mesh-config",
"mesh-catalog modules/mesh-delivery", "mesh-catalog modules/mesh-delivery",
"mesh-catalog modules/mesh-delivery-gate", "mesh-catalog modules/mesh-delivery-gate",
"mesh-catalog modules/mesh-files", "mesh-catalog modules/mesh-files",
"mesh-catalog modules/mesh-hand-acts",
"mesh-catalog modules/mesh-healers",
"mesh-catalog modules/mesh-issues", "mesh-catalog modules/mesh-issues",
"mesh-catalog modules/mesh-knowledge", "mesh-catalog modules/mesh-knowledge",
"mesh-catalog modules/mesh-mail", "mesh-catalog modules/mesh-mail",
@@ -190,7 +170,6 @@
"mesh-catalog modules/mesh-provisioner", "mesh-catalog modules/mesh-provisioner",
"mesh-catalog modules/mesh-provisioner-client", "mesh-catalog modules/mesh-provisioner-client",
"mesh-catalog modules/mesh-reads", "mesh-catalog modules/mesh-reads",
"mesh-catalog modules/mesh-self-check",
"mesh-catalog modules/mesh-vault", "mesh-catalog modules/mesh-vault",
"mesh-catalog modules/mesh-watcher", "mesh-catalog modules/mesh-watcher",
"mesh-catalog modules/messenger", "mesh-catalog modules/messenger",
@@ -32,8 +32,7 @@
"propose", "propose",
"import", "import",
"return", "return",
"withdraw", "withdraw"
"progress"
] ]
} }
], ],
@@ -53,15 +52,13 @@
"propose", "propose",
"import", "import",
"return", "return",
"withdraw", "withdraw"
"progress"
] ]
} }
], ],
"tools": [ "tools": [
"knowledge_health", "knowledge_health",
"knowledge_highest", "knowledge_highest"
"knowledge_marks"
], ],
"emits": [ "emits": [
"reserved", "reserved",
@@ -84,14 +81,7 @@
"records.records_status", "records.records_status",
"records.records_list", "records.records_list",
"records.records_read", "records.records_read",
"seat:operator-channel.notify", "seat:operator-channel.notify"
"seat:issue-tracker.list",
"seat:issue-tracker.get",
"seat:issue-tracker.milestones",
"seat:issue-tracker.changes",
"seat:mesh-delivery.deliveries",
"seat:mesh-delivery.show",
"seat:mesh-delivery.groups"
], ],
"config": { "config": {
"judge-model": { "judge-model": {
@@ -11,36 +11,11 @@
] ]
} }
], ],
"uses": [
"operator-channel"
],
"reads": [
"mesh-delivery-assigner.placement",
"mesh-delivery-assigner.wants",
"mesh-delivery-indexer.modules",
"mesh-delivery-indexer.manifests",
"mesh-delivery-builder.builds",
"node-engine.have",
"mesh-config.resolved",
"gitea.bound",
"keycloak.bound",
"postgres.bound",
"route-adapter.bound",
"mesh-vault.bound"
],
"tools": [
"provisioner_provisions",
"provisioner_plans",
"provisioner_compose",
"provisioner_decide"
],
"state": [ "state": [
"registry", "registry",
"needs", "needs",
"applied", "applied",
"collect", "collect"
"provisions",
"plans"
], ],
"own-secrets": { "own-secrets": {
"broker": "${dir:state}/broker" "broker": "${dir:state}/broker"
@@ -1,87 +0,0 @@
{
"module": "mesh-self-check",
"version": "1",
"runs-as": "mesh-self-check",
"claims": [
{
"name": "mesh-self-check",
"scope": "mesh",
"serves": [
"doctor",
"probes"
]
}
],
"invokes": [
"nats.nats_streams",
"nats.nats_closed_connections",
"seat:mesh-watchdogs.signals",
"seat:node-intrusion-prevention.banned"
],
"reads": [
"mesh-delivery-assigner.placement",
"mesh-delivery-assigner.seat-holders",
"mesh-delivery-watchman.presence",
"mesh-host.health",
"disk-load.readings",
"restic.good",
"mesh-delivery-indexer.modules",
"mesh-delivery-indexer.manifests"
],
"state": [
"findings",
"verdict",
{
"name": "verdicts",
"ttl-seconds": 604800
}
],
"own-secrets": {
"broker": "${dir:state}/broker"
},
"secrets-owner": "mesh-self-check",
"resources": [
{
"id": "account",
"type": "user",
"name": "mesh-self-check",
"shell": "/usr/bin/nologin",
"home": "/var/lib/mesh-self-check"
},
{
"id": "state",
"type": "directory",
"mode": "0700",
"owner": "mesh-self-check",
"place": ".",
"unknown": "program"
}
],
"build": {
"artifacts": [
{
"name": "tools",
"kind": "bundle",
"language": "go",
"system": "arch",
"from": "cmd/mesh-self-check",
"binary": "mesh-self-check",
"loads": [
"mesh-self-check"
]
}
]
},
"config": {
"probes": {
"type": "list",
"description": "The ids of the probes that run, each judging one thing of the mesh (mesh-self-check.probes lists them with what each asserts). A probe not listed runs nothing and raises nothing, and what it raised before is cleared. D8 and DA are in the code and off by default.",
"trusted": true,
"default": [
"D13",
"D-fill"
],
"runtime": true
}
}
}
@@ -5,9 +5,6 @@
"provides": [ "provides": [
{ {
"name": "secret", "name": "secret",
"credential": {
"form": "pair"
},
"scope": "mesh", "scope": "mesh",
"identity": { "identity": {
"in": "a vault entry" "in": "a vault entry"
@@ -17,27 +14,10 @@
"emits": [ "emits": [
"provisioned", "provisioned",
"rotated", "rotated",
"secret-changed", "secret-changed"
"account-hashed",
"own-replaced",
"account-issued",
"signed"
], ],
"reads": [ "reads": [
"node-engine.have", "mesh-controller.assignments"
"mesh-grants.accounts",
"mesh-controller.lease",
"mesh-delivery-assigner.placement",
"mesh-provisioner.plans",
"mesh-provisioner.provisions"
],
"state": [
{
"name": "bus-accounts",
"history": 2
},
"signers",
"bound"
], ],
"receives": { "receives": {
"secret": "${dir:grants}/mesh.json" "secret": "${dir:grants}/mesh.json"
@@ -155,8 +135,7 @@
"MESH_VAULT_LEDGER": "${dir:ledger}", "MESH_VAULT_LEDGER": "${dir:ledger}",
"MESH_VAULT_ROOT": "${dir:root}", "MESH_VAULT_ROOT": "${dir:root}",
"MESH_VAULT_STORE": "${dir:store}", "MESH_VAULT_STORE": "${dir:store}",
"MESH_VAULT_STATE": "${dir:state}", "MESH_VAULT_STATE": "${dir:state}"
"MESH_VAULT_SECRET_ASK_CALLERS": "node-tools"
} }
} }
] ]
@@ -166,10 +145,6 @@
"secret_holders", "secret_holders",
"secret_verify", "secret_verify",
"secret_handed", "secret_handed",
"secret_own",
"account_hash",
"account_adopt",
"account_retire",
"secret_asks", "secret_asks",
"secret_vault_export", "secret_vault_export",
"provisioner_retirement", "provisioner_retirement",
@@ -177,12 +152,7 @@
"provisioner_retire_reject", "provisioner_retire_reject",
"provisioner_delete", "provisioner_delete",
"vault_status", "vault_status",
"vault_health", "vault_health"
"own_drift",
"own_take_switch",
"vault_signers",
"vault_signed",
"signers_reassert"
], ],
"claims": [ "claims": [
{ {
@@ -196,10 +166,7 @@
"rotate", "rotate",
"move-in", "move-in",
"node-key", "node-key",
"move-asks", "move-asks"
"own",
"issue-account",
"sign"
] ]
} }
] ]
@@ -3,15 +3,11 @@
"version": "1", "version": "1",
"slug": "watch", "slug": "watch",
"consumes": [ "consumes": [
"mesh-controller.doctor-heartbeat", "mesh-controller.doctor-heartbeat"
"mesh-self-check.checked"
],
"reads": [
"mesh-conditions.takeover",
"mesh-delivery-assigner.seat-holders"
], ],
"invokes": [ "invokes": [
"mesh-watcher.watcher_ping" "mesh-watcher.watcher_ping",
"seat:mesh-controller.seats"
], ],
"own-secrets": { "own-secrets": {
"telegram-token": "${dir:state}/telegram-token" "telegram-token": "${dir:state}/telegram-token"
@@ -43,9 +39,7 @@
"type": "file", "type": "file",
"path": "${dir:state}/settings.json", "path": "${dir:state}/settings.json",
"mode": "0600", "mode": "0600",
"takes-config": [ "takes-config": ["telegram-chat-id"],
"telegram-chat-id"
],
"merge": "json", "merge": "json",
"template": "files/settings.json" "template": "files/settings.json"
} }
@@ -77,9 +77,6 @@
"mesh-controller.condition-raised", "mesh-controller.condition-raised",
"mesh-controller.condition-changed", "mesh-controller.condition-changed",
"mesh-controller.condition-cleared", "mesh-controller.condition-cleared",
"mesh-conditions.condition-raised",
"mesh-conditions.condition-changed",
"mesh-conditions.condition-cleared",
"mesh-delivery-view.moving", "mesh-delivery-view.moving",
"mesh-delivery-view.settled", "mesh-delivery-view.settled",
"mesh-delivery-view.stuck", "mesh-delivery-view.stuck",
@@ -93,13 +90,9 @@
], ],
"invokes": [ "invokes": [
"seat:mesh-controller.conditions", "seat:mesh-controller.conditions",
"seat:mesh-conditions.conditions", "seat:mesh-controller.root-free",
"seat:node-engine.root-free",
"seat:issue-tracker.tracking" "seat:issue-tracker.tracking"
], ],
"reads": [
"mesh-conditions.takeover"
],
"state": [ "state": [
"open", "open",
"sent", "sent",
@@ -12,9 +12,6 @@
"identity": { "identity": {
"max": 20, "max": 20,
"in": "an S3 access key" "in": "an S3 access key"
},
"credential": {
"form": "pair"
} }
} }
], ],
@@ -39,8 +36,8 @@
] ]
} }
], ],
"reads": [ "invokes": [
"mesh-delivery-assigner.seat-holders" "seat:mesh-controller.seats"
], ],
"contributes": { "contributes": {
"route": { "route": {
@@ -12,9 +12,6 @@
"identity": { "identity": {
"max": 63, "max": 63,
"in": "a MongoDB database name" "in": "a MongoDB database name"
},
"credential": {
"form": "pair"
} }
} }
], ],
@@ -8,9 +8,6 @@
"scope": "mesh", "scope": "mesh",
"identity": { "identity": {
"in": "a Mosquitto client and topic prefix" "in": "a Mosquitto client and topic prefix"
},
"credential": {
"form": "pair"
} }
} }
], ],
@@ -5,9 +5,11 @@
"nfs-share" "nfs-share"
], ],
"reads": [ "reads": [
"disk-load.readings",
"nfs-server.exports" "nfs-server.exports"
], ],
"invokes": [
"seat:mesh-controller.data"
],
"capabilities": [ "capabilities": [
"package-manager", "package-manager",
"service-manager" "service-manager"
@@ -36,20 +38,20 @@
"per-machine": true "per-machine": true
} }
], ],
"config": { "config": {
"shares": { "shares": {
"type": "string", "type": "string",
"description": "The shares of another machine this machine mounts: space-separated name=mountpoint, each optionally :ro, or none (novox/hq ADR 0263 rule 5).", "description": "The shares of another machine this machine mounts: space-separated name=mountpoint, each optionally :ro, or none (novox/hq ADR 0263 rule 5).",
"trusted": true, "trusted": true,
"default": "none" "default": "none"
},
"sources": {
"type": "string",
"description": "The occasional sources this machine mounts: space-separated name=smb://[<user>@]<host>/<share>@<mountpoint> or name=disk:<uuid>@<mountpoint>, each optionally :ro, or none (novox/hq ADR 0263 rule 6); an SMB source's username is in its entry, its password the secret smb-password-<name> (novox/hq ADR 0283).",
"trusted": true,
"default": "none"
}
}, },
"sources": {
"type": "string",
"description": "The occasional sources this machine mounts: space-separated name=smb://[<user>@]<host>/<share>@<mountpoint> or name=disk:<uuid>@<mountpoint>, each optionally :ro, or none (novox/hq ADR 0263 rule 6); an SMB source's username is in its entry, its password the secret smb-password-<name> (novox/hq ADR 0283).",
"trusted": true,
"default": "none"
}
},
"own-secrets": { "own-secrets": {
"smb-password-*": { "smb-password-*": {
"path": "${dir:state}/smb-password-*.secret", "path": "${dir:state}/smb-password-*.secret",
@@ -12,9 +12,6 @@
"identity": { "identity": {
"max": 128, "max": 128,
"in": "a SQL Server login and database name" "in": "a SQL Server login and database name"
},
"credential": {
"form": "pair"
} }
} }
], ],
@@ -8,10 +8,7 @@
"provides": [ "provides": [
{ {
"name": "mesh-bus", "name": "mesh-bus",
"scope": "mesh", "scope": "mesh"
"credential": {
"form": "none"
}
} }
], ],
"claims": [ "claims": [
@@ -22,16 +19,12 @@
], ],
"bus-users": "/var/lib/nats-module/conf/accounts.conf", "bus-users": "/var/lib/nats-module/conf/accounts.conf",
"own-secrets": { "own-secrets": {
"broker": "${dir:mesh-state}/broker", "broker": "${dir:mesh-state}/broker"
"broker-nats-tools": "${dir:tools-secret}/broker-nats-tools"
}, },
"capabilities": [ "capabilities": [
"container-runtime" "container-runtime"
], ],
"emits": [ "emits": [],
"consumer-removed",
"bucket-keys-removed"
],
"consumes": [], "consumes": [],
"listens": [ "listens": [
{ {
@@ -52,9 +45,7 @@
"nats_buckets", "nats_buckets",
"nats_users", "nats_users",
"nats_user_can", "nats_user_can",
"nats_snapshot_verify", "nats_snapshot_verify"
"nats_consumer_remove",
"nats_bucket_remove"
], ],
"guards": [ "guards": [
8222 8222
@@ -95,13 +86,6 @@
"mode": "0700", "mode": "0700",
"unknown": "program" "unknown": "program"
}, },
{
"id": "tools-secret",
"type": "directory",
"path": "/var/lib/mesh/nats-tools",
"mode": "0711",
"unknown": "program"
},
{ {
"id": "snapshots", "id": "snapshots",
"type": "directory", "type": "directory",
@@ -206,18 +190,11 @@
"loads": [ "loads": [
"nats-tools" "nats-tools"
], ],
"bus": {
"removes": [
"DEAD_LETTER_NOTICES.controller",
"mesh-host_acts"
]
},
"env": { "env": {
"MESH_NATS_MONITOR": "http://127.0.0.1:8222", "MESH_NATS_MONITOR": "http://127.0.0.1:8222",
"MESH_NATS_CONTAINER": "mesh-broker-nats", "MESH_NATS_CONTAINER": "mesh-broker-nats",
"MESH_NATS_USERS_FILE": "/etc/nats/accounts.conf", "MESH_NATS_USERS_FILE": "/etc/nats/accounts.conf",
"MESH_NATS_SNAPSHOTS_DIR": "${dir:snapshots}", "MESH_NATS_SNAPSHOTS_DIR": "${dir:snapshots}"
"NATS_TOOLS_CREDENTIAL": "${dir:tools-secret}/broker-nats-tools"
} }
}, },
{ {
@@ -72,7 +72,7 @@
}, },
"user": { "user": {
"type": "string", "type": "string",
"description": "The Nextcloud account the module's tools act as; empty is mesh-admin, the account the module makes and keeps for itself. Any other account is used as given and never repaired.", "description": "The Nextcloud account the module's tools act as; empty is admin.",
"trusted": true, "trusted": true,
"default": "" "default": ""
} }
@@ -13,10 +13,7 @@
{ {
"name": "nfs-share", "name": "nfs-share",
"scope": "mesh", "scope": "mesh",
"identity": false, "identity": false
"credential": {
"form": "none"
}
} }
], ],
"data": { "data": {
@@ -48,24 +45,26 @@
} }
], ],
"reads": [ "reads": [
"disk-load.readings",
"mesh-delivery-assigner.seat-holders",
"mounts.shares" "mounts.shares"
], ],
"config": { "invokes": [
"shares": { "seat:mesh-controller.seats",
"type": "string", "seat:mesh-controller.data"
"description": "The folders this machine exports, as space-separated name=path; the operator's, with no default (novox/hq ADR 0263).", ],
"trusted": true, "config": {
"required": true "shares": {
"type": "string",
"description": "The folders this machine exports, as space-separated name=path; the operator's, with no default (novox/hq ADR 0263).",
"trusted": true,
"required": true
},
"grants": {
"type": "string",
"description": "Which nodes may mount which share, and how: share=node:rw|ro[,node:rw|ro], entries separated by spaces, or none to export to no node whatever a node wants (novox/hq ADR 0263).",
"trusted": true,
"default": "none"
}
}, },
"grants": {
"type": "string",
"description": "Which nodes may mount which share, and how: share=node:rw|ro[,node:rw|ro], entries separated by spaces, or none to export to no node whatever a node wants (novox/hq ADR 0263).",
"trusted": true,
"default": "none"
}
},
"facts": { "facts": {
"machines": { "machines": {
"path": "/etc/nfs-server/machines", "path": "/etc/nfs-server/machines",
@@ -7,10 +7,7 @@
"provides": [ "provides": [
{ {
"name": "node-toolchain", "name": "node-toolchain",
"reach": "machine", "reach": "machine"
"credential": {
"form": "none"
}
} }
], ],
"resources": [ "resources": [
@@ -7,10 +7,7 @@
"provides": [ "provides": [
{ {
"name": "model-access", "name": "model-access",
"scope": "node", "scope": "node"
"credential": {
"form": "none"
}
} }
], ],
"listens": [ "listens": [
@@ -9,7 +9,6 @@
"provides": [ "provides": [
{ {
"name": "postgres-database", "name": "postgres-database",
"credential": {"form": "pair"},
"scope": "mesh", "scope": "mesh",
"identity": { "identity": {
"max": 63, "max": 63,
@@ -196,7 +195,6 @@
"bound" "bound"
], ],
"reads": [ "reads": [
"mesh-provisioner.plans", "mesh-controller.assignments"
"mesh-provisioner.provisions"
] ]
} }
@@ -314,48 +314,6 @@
"mode": "0644", "mode": "0644",
"template": "templates/systemd/system/suspend-then-hibernate.target.d/mesh-power.conf" "template": "templates/systemd/system/suspend-then-hibernate.target.d/mesh-power.conf"
}, },
{
"id": "reboot-drop-ins",
"type": "directory",
"path": "/etc/systemd/system/reboot.target.d",
"mode": "0755",
"unknown": "program"
},
{
"id": "reboot-bound",
"type": "file",
"path": "/etc/systemd/system/reboot.target.d/mesh-power.conf",
"mode": "0644",
"template": "templates/systemd/system/reboot.target.d/mesh-power.conf"
},
{
"id": "poweroff-drop-ins",
"type": "directory",
"path": "/etc/systemd/system/poweroff.target.d",
"mode": "0755",
"unknown": "program"
},
{
"id": "poweroff-bound",
"type": "file",
"path": "/etc/systemd/system/poweroff.target.d/mesh-power.conf",
"mode": "0644",
"template": "templates/systemd/system/poweroff.target.d/mesh-power.conf"
},
{
"id": "halt-drop-ins",
"type": "directory",
"path": "/etc/systemd/system/halt.target.d",
"mode": "0755",
"unknown": "program"
},
{
"id": "halt-bound",
"type": "file",
"path": "/etc/systemd/system/halt.target.d/mesh-power.conf",
"mode": "0644",
"template": "templates/systemd/system/halt.target.d/mesh-power.conf"
},
{ {
"id": "udev-rule", "id": "udev-rule",
"type": "file", "type": "file",
@@ -1,8 +0,0 @@
# Written by the mesh (module power, novox/hq ADR 0211; issue 1126). Replaced on every push.
#
# A shutdown is bounded at ten minutes, not systemd's thirty. A unit started again during the shutdown
# (on ace, 2026-10-28, the bus and the authorization manager) is never stopped a second time, and the
# shutdown then waits on this job's timeout with nothing on the console. The target keeps its own
# forcing action: the last phase still ends every process, unmounts and syncs.
[Unit]
JobTimeoutSec=10min
@@ -1,8 +0,0 @@
# Written by the mesh (module power, novox/hq ADR 0211; issue 1126). Replaced on every push.
#
# A shutdown is bounded at ten minutes, not systemd's thirty. A unit started again during the shutdown
# (on ace, 2026-10-28, the bus and the authorization manager) is never stopped a second time, and the
# shutdown then waits on this job's timeout with nothing on the console. The target keeps its own
# forcing action: the last phase still ends every process, unmounts and syncs.
[Unit]
JobTimeoutSec=10min
@@ -1,8 +0,0 @@
# Written by the mesh (module power, novox/hq ADR 0211; issue 1126). Replaced on every push.
#
# A shutdown is bounded at ten minutes, not systemd's thirty. A unit started again during the shutdown
# (on ace, 2026-10-28, the bus and the authorization manager) is never stopped a second time, and the
# shutdown then waits on this job's timeout with nothing on the console. The target keeps its own
# forcing action: the last phase still ends every process, unmounts and syncs.
[Unit]
JobTimeoutSec=10min

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