Compare commits

..
Author SHA1 Message Date
mesh-admin 2959c81435 Merge pull request 'Link a build's bases into its repository before pushing it (issue 694)' (#318) from fix/694-a-push-holds-its-bases into main 2026-10-14 01:19:24 +00:00
jochen 8933864a49 Link a build's bases into its repository before pushing it (issue 694)
mesh/merge-gate head 8933864a49d0 merged into trunk df9e5728a925 — pass: builds build-agent, mesh-controller → ace, g14, novox, shanks; no bus step; every …
mesh/repo-check head 8933864a49d0 merged into trunk df9e5728a925 — pass: its merge-check.sh passed
mesh/delivery delivered
A runtime on containerd's image store, the default on a fresh docker 29,
need not hold the compressed layers of a base it built FROM: a layer it
had unpacked already is never fetched. Its push then fails on that layer
although the registry holds it under the base's mirror. A push reads
locally only what the target repository lacks, so mount the base's blobs
there first.
2026-10-14 02:59:53 +02:00
mesh-admin df9e5728a9 Merge pull request 'Try a pushed branch on the build seat before its pull request (hq issue 686)' (#314) from fix/686-a-branch-is-tested-before-its-pull-request into main 2026-10-14 00:57:46 +00:00
mesh-admin 2d18f438c5 Merge pull request 'A no-move tier waits for no new report, so its walk asks the next tier (issue 648)' (#309) from fix/648-a-no-move-tier-waits-for-nothing into main 2026-10-14 00:32:47 +00:00
jschoubben 3ce5a0239b Keep a try off the forge and bound how many run (hq issue 686)
mesh/merge-gate head 3ce5a0239bb1 merged into trunk 2d18f438c516 — pass: builds build-agent, mesh-controller → ace, g14, novox, shanks; no bus step; every …
mesh/repo-check head 3ce5a0239bb1 merged into trunk 2d18f438c516 — pass: THE CHANGE ALTERS ITS OWN CHECK (merge-check.sh): main's version judged it; the ch…
mesh/delivery delivered
Review of #314: a try's start named its commit, which the forge's holder
matches to an open pull request's head and marks pending. Its start now
names no repository or commit. One try per commit and four in all, a try
never ages ahead of deliveries or pull request checks, and its id is
checked before it is a bus subject.
2026-10-14 02:29:28 +02:00
jochen 70885ab3c4 Share one prefix between the no-move Why's writer and its reader (review of #309, issue 648)
mesh/delivery delivered
mesh/merge-gate head 70885ab3c4a3 merged into trunk ca3f43c400ba — pass: builds mesh-controller → novox; no bus step; every machine composes with the chang…
mesh/repo-check head 70885ab3c4a3 merged into trunk ca3f43c400ba — pass: its merge-check.sh passed
2026-10-14 02:22:25 +02:00
jschoubben 65508d9df9 Try a pushed branch on the build seat before its pull request (hq issue 686)
mesh/merge-gate head 65508d9df9f4 merged into trunk ca3f43c400ba — pass: builds build-agent, mesh-controller → ace, g14, novox, shanks; no bus step; every …
mesh/repo-check head 65508d9df9f4 merged into trunk ca3f43c400ba — pass: THE CHANGE ALTERS ITS OWN CHECK (merge-check.sh): main's version judged it; the ch…
mesh/delivery superseded: a newer head of the same pull request
Store tests skip without a store and no agent may start one, so each failure
cost a 15-30 minute review round. The try verb asks the build seat for the
repository's merge-check.sh on the ref merged into the trunk, as a check in
the build queue, and reads its outcome from the build's log: no status on
the forge, nothing registered, nothing built for a merge.
2026-10-14 02:20:37 +02:00
jschoubben 8123ca5654 Test that a branch is tried before its pull request (hq issue 686)
A builder's store tests first ran in the pull request's merge check, a review
round late. These fail on main: a try asks the repository's own check alone,
its outcome sets no status and registers nothing, an unknown ref is refused
before anything is asked, and the answer carries the failing lines.
2026-10-14 02:20:37 +02:00
jochen d9b86376d3 A no-move tier waits for no new report, so its walk asks the next tier (issue 648)
A build made from the source every machine running it runs is marked sent
and sent nothing, yet the tier's wait asked each machine for a report after
that marking. None follows, and the walk waited for ever. Each machine is
now judged by whether it answered its last send, the one whose builds made
the build no move.
2026-10-14 02:18:53 +02:00
mesh-admin ca3f43c400 Merge pull request 'The rest go at once after the first passes, and a gate survives a new controller (hq ADR 0353)' (#312) from fix/the-rest-in-parallel-and-gates-survive-a-new-controller into main 2026-10-13 23:43:02 +00:00
jschoubben b482122d66 Say a read error of a silent rest, and keep the node-engine one at a time in a kept step (hq ADR 0353)
mesh/delivery delivered
mesh/merge-gate head b482122d6646 merged into trunk 1601cb17432e — pass: builds build-agent, mesh-controller → ace, g14, novox, shanks; no bus step; every …
mesh/repo-check head b482122d6646 merged into trunk 1601cb17432e — pass: its merge-check.sh passed
An empty answer on a read error would clear release-silent unseen.
2026-10-14 01:26:00 +02:00
jschoubben 2f5cc146fe Keep a walk's sends to what was kept, and say a silent rest (hq ADR 0353 review)
mesh/merge-gate head 2f5cc146fe46 merged into trunk 1601cb17432e — pass: builds build-agent, mesh-controller → ace, g14, novox, shanks; no bus step; every …
mesh/repo-check head 2f5cc146fe46 merged into trunk 1601cb17432e — fail: its merge-check.sh failed: --- FAIL: TestAFirstMachineThatFailsStillStopsTheWalk (…
mesh/delivery superseded: a newer head of the same pull request
A release's own gates held its other sends back for ever; an adopted send
lost what the machine ran before; a kept send could grow a build no gate
passed; a take-over matched by commit alone; a rest that went silent ended
done. The node-engine and the resolver still go one machine at a time, since
a fault that cuts a machine off shows only as silence.
2026-10-14 01:18:51 +02:00
jschoubben c2ea9b4c51 Send the rest at once after the first passes, and resume a stored gate (hq ADR 0353)
mesh/repo-check head c2ea9b4c5187 merged into trunk 1601cb17432e — fail: its merge-check.sh failed: --- FAIL: TestAReleaseSendsTheRestAtOnceAfterTheFirstPa…
mesh/delivery superseded: a newer head of the same pull request
mesh/merge-gate head c2ea9b4c5187 merged into trunk 1601cb17432e — pass: builds build-agent, mesh-controller → ace, g14, novox, shanks; no bus step; every …
A release now sends every other machine in one step once its first machine
passed, each judged by its own gate; a failing one is put back without
stopping the others, a silent one is left. A send is kept before it is made,
and a machine already sent what a gate is for is not sent again: the gate
counts from that send and takes over another walk's readings of the same
build there.
2026-10-14 01:06:49 +02:00
jschoubben b03d257f81 Test that the rest go at once after the first passes and a gate survives a new controller (hq ADR 0353)
The release walked four machines one after another in eleven minutes, and a
new controller re-sent and re-judged a build its machine already ran.
2026-10-14 01:06:49 +02:00
55 changed files with 2508 additions and 1691 deletions
-6
View File
@@ -1,6 +0,0 @@
# Release 57: config and provisioning
This branch carries the work of hq design 57 (ADR 0348, ADR 0349) in this repository. Each package
merges here through its own reviewed pull request; the branch reaches main as one pull request, in one
delivery group with the branch of the same name in mesh-controller, mesh-host, mesh-sdk, mesh-catalog
and hq, after a rehearsal in the lab. This file is deleted by the last commit before that pull request.
+25 -1
View File
@@ -332,6 +332,12 @@ func answer(ctx context.Context, publisher builder.Publisher, on, workspace stri
result.Check.Members = append(result.Check.Members, link.CheckedMember{Owner: m.Owner, Repo: m.Repo, result.Check.Members = append(result.Check.Members, link.CheckedMember{Owner: m.Owner, Repo: m.Repo,
Commit: m.Head, Trunk: m.Trunk}) Commit: m.Head, Trunk: m.Trunk})
} }
// A branch tried before its pull request reads its outcome from this log, whole (novox/hq issue 686).
if request.Check.Try {
if line, err := json.Marshal(triedOf(v)); err == nil {
say(link.StepTried, string(line))
}
}
} }
} else if err == nil { } else if err == nil {
// The package-registry credential is a build input, so it is resolved before the clone: a // The package-registry credential is a build input, so it is resolved before the clone: a
@@ -457,7 +463,8 @@ func checkSpecOf(request link.BuildRequest) builder.CheckSpec {
spec := builder.CheckSpec{ID: request.ID, Repository: request.Repository, Ref: request.Ref, spec := builder.CheckSpec{ID: request.ID, Repository: request.Repository, Ref: request.Ref,
Owner: c.Owner, Repo: c.Repo, Number: c.Number, Paths: c.Paths, Beside: map[string]builder.Beside{}, Owner: c.Owner, Repo: c.Repo, Number: c.Number, Paths: c.Paths, Beside: map[string]builder.Beside{},
Modules: c.Modules, New: c.New, Manifests: c.Manifests, Judge: c.Judge, Base: c.Base, Modules: c.Modules, New: c.New, Manifests: c.Manifests, Judge: c.Judge, Base: c.Base,
Toolchain: builder.ToolchainOf(request.Held), Toolchains: builder.ToolchainsOf(request.Held)} Toolchain: builder.ToolchainOf(request.Held), Toolchains: builder.ToolchainsOf(request.Held),
Tests: c.Tests, StoreOnly: c.StoreOnly}
for dir, b := range c.Beside { for dir, b := range c.Beside {
spec.Beside[dir] = builder.Beside{Repository: b.Repository, Ref: b.Ref} spec.Beside[dir] = builder.Beside{Repository: b.Repository, Ref: b.Ref}
} }
@@ -472,6 +479,23 @@ func checkSpecOf(request link.BuildRequest) builder.CheckSpec {
return spec return spec
} }
// triedOf is a try's outcome as its log line says it (novox/hq issue 686): the repository's own check's verdict,
// or the check's where that did not run — a head that does not merge — and the lines naming what failed.
func triedOf(v builder.CheckVerdict) link.Tried {
t := link.Tried{Verdict: v.Verdict, Summary: v.Summary, Took: v.Took.Round(time.Second).String(),
Head: v.Head, Trunk: v.Trunk}
if v.Repo != nil && (v.Gate == nil || v.Gate.Verdict == "pass") {
t.Verdict, t.Summary = v.Repo.Verdict, v.Repo.Summary
}
if v.Repo != nil && v.Repo.Failed != "" {
t.Failed = strings.Split(v.Repo.Failed, "\n")
}
if t.Verdict == "" {
t.Verdict = "error"
}
return t
}
// checkBuiltOf is a check's builds for its merge as the outcome carries them (novox/hq ADR 0350 §1). // checkBuiltOf is a check's builds for its merge as the outcome carries them (novox/hq ADR 0350 §1).
func checkBuiltOf(built []builder.CheckBuilt) []link.CheckBuilt { func checkBuiltOf(built []builder.CheckBuilt) []link.CheckBuilt {
var out []link.CheckBuilt var out []link.CheckBuilt
+31
View File
@@ -0,0 +1,31 @@
package main
import (
"testing"
"time"
"github.com/novox/mesh-controller/internal/builder"
)
// A try's outcome, as the build seat says it in the build's log (novox/hq issue 686): the repository's own
// check's verdict, how long it took, and each line naming what failed.
func TestATrysOutcomeNamesWhatFailed(t *testing.T) {
tried := triedOf(builder.CheckVerdict{Verdict: "pass", Summary: "the change touches no module", Took: 303 * time.Second,
Head: "c0ffee", Trunk: "ca3f43c", Gate: &builder.Layer{Verdict: "pass"},
Repo: &builder.Layer{Verdict: "fail", Summary: "its merge-check.sh failed: TestX",
Failed: "--- FAIL: TestX (0.01s)\n x_test.go:12: got 1, want 2"}})
if tried.Verdict != "fail" || tried.Took != "5m3s" || len(tried.Failed) != 2 || tried.Trunk != "ca3f43c" ||
tried.Summary != "its merge-check.sh failed: TestX" {
t.Fatalf("a failed try: %+v", tried)
}
// A head that does not merge fails both layers: the check's own word.
conflict := triedOf(builder.CheckVerdict{Verdict: "fail", Summary: "does not merge", Gate: &builder.Layer{Verdict: "fail"},
Repo: &builder.Layer{Verdict: "fail", Summary: "does not merge"}})
if conflict.Verdict != "fail" || conflict.Summary != "does not merge" {
t.Fatalf("a try that does not merge: %+v", conflict)
}
// No verdict is an error, never a pass.
if none := triedOf(builder.CheckVerdict{}); none.Verdict != "error" {
t.Fatalf("a try with no verdict: %+v", none)
}
}
+12
View File
@@ -1443,6 +1443,12 @@ func probeGates(ctx context.Context, d *doctor) ([]conditions.Observation, error
} }
// A release held after a failed one, while builds still wait for a gate (ADR 0236). // A release held after a failed one, while builds still wait for a gate (ADR 0236).
out = append(out, backlogObservation()...) out = append(out, backlogObservation()...)
// And a rest that went silent (novox/hq ADR 0353 §4).
silent, err := silentObservation(ctx, d.open.inventory)
if err != nil {
return nil, err
}
out = append(out, silent...)
return sortedFound(dedupeObservations(out)), nil return sortedFound(dedupeObservations(out)), nil
} }
@@ -1532,6 +1538,12 @@ func gateLine(g *inventory.PlanGate) string {
case inventory.GateFailed: case inventory.GateFailed:
what = "FAILED" what = "FAILED"
} }
switch {
case g.Verdict == "" && g.Pending:
what = "kept, not yet sent" // novox/hq ADR 0353 §8
case g.Verdict == "" && g.Silent:
what = "silent on its send, left" // ADR 0353 §4
}
if g.Verdict == "" && g.Waiting() { if g.Verdict == "" && g.Waiting() {
what = "waits for a busy container on " + g.Busy.Machine what = "waits for a busy container on " + g.Busy.Machine
} }
+1 -1
View File
@@ -151,7 +151,7 @@ func heldMachines(ctx context.Context, open *stores, names []string) (map[string
for _, mv := range f.moves(node, modules, same, known, false) { for _, mv := range f.moves(node, modules, same, known, false) {
if planStillToSend(plans, mv.Module, node) == "" { if planStillToSend(plans, mv.Module, node) == "" {
why = append(why, fmt.Sprintf("%s would move from %s to %s, which has passed no gate yet — a release "+ why = append(why, fmt.Sprintf("%s would move from %s to %s, which has passed no gate yet — a release "+
"plan sends it, one machine at a time, judged", mv.Module, buildName(mv.From), buildName(mv.To))) "plan sends it, one machine first and then the rest at once, judged", mv.Module, buildName(mv.From), buildName(mv.To)))
} }
} }
if len(why) > 0 { if len(why) > 0 {
+87
View File
@@ -0,0 +1,87 @@
package main
import (
"encoding/json"
"strings"
"testing"
"time"
"github.com/novox/mesh-controller/internal/catalogue"
"github.com/novox/mesh-controller/internal/inventory"
)
// A tier whose module is no move — made from the source every machine running it runs, so nothing is
// sent — is not waited on for a report after its marking (novox/hq issue 648). Its machines answered the
// send that last carried that build before the walk got there, and nothing will make them report again:
// the walk asks the next tier in the same step, rather than "waiting for app on anchor, laptop" for ever.
func TestANoMoveTierWaitsForNoNewReport(t *testing.T) {
judgeLeavesAsTheCore(t)
g := aGateMesh(t)
ctx := t.Context()
inv := g.open.inventory
// app is the build machine, so a module of the next tier is built by it: the tier gates the next on app
// running on its machines.
for _, m := range []catalogue.Manifest{
{Module: "app", Version: "c2", Claims: []catalogue.Claim{{Name: "node-build-agent", Scope: catalogue.ScopeNode}}},
{Module: "web", Version: "w1"},
} {
if err := inv.RegisterModule(ctx, m, inventory.Source{Repository: "novox/mesh-catalog", Seat: "git",
Path: "modules/" + m.Module, BuiltFrom: m.Version, Head: m.Version, Asked: time.Now().Add(-time.Minute)}); err != nil {
t.Fatal(err)
}
}
// What anchor and laptop were last sent (c1) was made from the source the plan's build of c2 is.
for _, b := range []inventory.Build{
{ID: "build-1s", Module: "app", Commit: "c1", SourceFingerprint: "src1:app",
Asked: time.Now().Add(-30 * time.Second), At: time.Now().Add(-30 * time.Second)},
{ID: "build-2s", Module: "app", Commit: "c2", SourceFingerprint: "src1:app", Asked: time.Now(), At: time.Now()},
} {
b.Manifest, _ = json.Marshal(catalogue.Manifest{Module: "app", Version: "1"})
if err := inv.RecordBuild(ctx, b); err != nil {
t.Fatal(err)
}
}
// Each machine answered that send, applied, before the walk's tier was built.
for _, n := range []string{"anchor", "laptop"} {
if _, err := inv.RecordDoing(ctx, nodeID(t, g.open, n), inventory.Doing{Node: n,
Outcome: inventory.OutcomeApplied, Declared: "d-" + n + "-c1", Applied: 1,
At: time.Now().Add(-10 * time.Second)}); err != nil {
t.Fatal(err)
}
}
built := time.Now().UTC()
p := g.plan(t)
p.Tiers = [][]string{{"app"}, {"web"}}
p.Modules["app"].Build, p.Modules["app"].BuiltAt = "build-2s", &built
if err := inv.SavePlan(ctx, &p); err != nil {
t.Fatal(err)
}
advancePlans(ctx, g.open)
if len(g.sent) != 0 {
t.Fatalf("a no-move build was sent: %v", g.sent)
}
p = g.plan(t)
if s := p.Modules["app"]; s == nil || !strings.HasPrefix(s.Why, "no move") {
t.Fatalf("the build was not marked no move: %+v", s)
}
if p.Tier < 1 || strings.Contains(p.Note, "waiting for app") {
t.Fatalf("the walk waits on a no-move tier for a report nothing will cause: tier %d, %s %q", p.Tier, p.State, p.Note)
}
// A machine that has not answered its last send is still waited on: no move is not "never sent".
p.Tier, p.State, p.Note = 0, inventory.PlanRolling, ""
if err := inv.RecordSent(ctx, nodeID(t, g.open, "laptop"), "d-laptop-later", map[string]string{"app": "c1"}); err != nil {
t.Fatal(err)
}
if err := inv.SavePlan(ctx, &p); err != nil {
t.Fatal(err)
}
advancePlans(ctx, g.open)
p = g.plan(t)
if p.Tier != 0 || !strings.Contains(p.Note, "waiting for app on laptop") {
t.Fatalf("a machine that has not answered its last send was not waited on: tier %d, %q", p.Tier, p.Note)
}
}
+9
View File
@@ -106,6 +106,8 @@ func run() error {
return killCommand(ctx, args[1:]) return killCommand(ctx, args[1:])
case "forget-checks": case "forget-checks":
return forgetChecksCommand(ctx, args[1:]) return forgetChecksCommand(ctx, args[1:])
case "try":
return tryCommand(ctx, args[1:])
case "pause", "resume": case "pause", "resume":
return pauseCommand(ctx, args[0], args[1:]) return pauseCommand(ctx, args[0], args[1:])
case "collection": case "collection":
@@ -381,6 +383,13 @@ func (b builds) Built(ctx context.Context, result link.BuildResult) error {
// **A merge check builds nothing** (novox/hq to-be 45 §9): its verdict is said, and nothing of it is // **A merge check builds nothing** (novox/hq to-be 45 §9): its verdict is said, and nothing of it is
// recorded or registered. // recorded or registered.
if result.Check != nil || result.Checked != nil { if result.Check != nil || result.Checked != nil {
if isTry(result) {
// A branch tried before its pull request (novox/hq issue 686): its asker reads the outcome from the
// build's log. Never said as `checked`, so the forge's holder sets no status.
fmt.Printf("%s: a try of %s at %.8s, read with `try --id`: said to nobody, nothing taken in\n", result.ID,
result.Repository, result.Ref)
return nil
}
checked(ctx, result) checked(ctx, result)
return nil return nil
} }
+16
View File
@@ -456,6 +456,9 @@ var plainWordings = map[string]func(conditions.Observation) words{
"release-held": worded(func(o conditions.Observation) words { "release-held": worded(func(o conditions.Observation) words {
return releaseHeldWords(nil, o.Also) return releaseHeldWords(nil, o.Also)
}), }),
"release-silent": worded(func(o conditions.Observation) words {
return releaseSilentWords(o.Also)
}),
"facts-stale": worded(func(o conditions.Observation) words { "facts-stale": worded(func(o conditions.Observation) words {
return words{Headline: "Merge checks use outdated facts", return words{Headline: "Merge checks use outdated facts",
Explanation: "The facts a merge check judges a change against are out of date; they are gathered again " + Explanation: "The facts a merge check judges a change against are out of date; they are gathered again " +
@@ -841,6 +844,19 @@ func releaseHeldWords(modules, machines []string) words {
Resolved: "Resolved: the held updates are released"} Resolved: "Resolved: the held updates are released"}
} }
// releaseSilentWords are the plain words of machines that said nothing on an update they were sent (ADR 0353 §4).
func releaseSilentWords(machines []string) words {
where := "Some machines"
if len(machines) > 0 {
where = conditions.Capital(namesWords(machines, 4))
}
return words{Headline: "Machines did not answer an update",
Needs: "check that they are awake and on the mesh; the next walk sends them what waits.",
Explanation: where + " were sent module updates and said nothing about them within fifteen minutes, so " +
"they were left as they were. What they run is not known to be the update.",
Resolved: "Resolved: the next walk reached them"}
}
// reloginNeeds is what an account waiting for its groups needs (ADR 0252). // reloginNeeds is what an account waiting for its groups needs (ADR 0252).
func reloginNeeds(node string) string { func reloginNeeds(node string) string {
return fmt.Sprintf("log out of %s completely and log in again, or restart it.", node) return fmt.Sprintf("log out of %s completely and log in again, or restart it.", node)
-3
View File
@@ -264,9 +264,6 @@ func serve(ctx context.Context) (err error) {
// The facts snapshot a merge check is fed (novox/hq to-be 45 §9): kept current while this // The facts snapshot a merge check is fed (novox/hq to-be 45 §9): kept current while this
// controller holds the lease, read by the build seat from the artifact store. // controller holds the lease, read by the build seat from the artifact store.
go exportingFacts(ctx, open, bus.Conn.ConnectedServerVersion) go exportingFacts(ctx, open, bus.Conn.ConnectedServerVersion)
// The record mirrors (novox/hq ADR 0289, design 57 F3): the assignments and each registered module's
// declared config on the bus, written from the store only when a record changed, by the lease holder.
go mirroringRecords(ctx, inv, bus.Conn, bus)
// Every call carries the lease's epoch, and its record is written only under the lease (novox/hq // Every call carries the lease's epoch, and its record is written only under the lease (novox/hq
// to-be 45 §6). // to-be 45 §6).
link.Calls.UnderLease(func() (uint64, error) { return theLease.epoch(ctx) }) link.Calls.UnderLease(func() (uint64, error) { return theLease.epoch(ctx) })
-105
View File
@@ -1,105 +0,0 @@
package main
import (
"context"
"fmt"
"time"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"github.com/novox/mesh-controller/internal/inventory"
"github.com/novox/mesh-controller/internal/recordmirror"
)
// The controller's record mirrors on the bus (novox/hq ADR 0289, design 54, design 57 F3): the assignments,
// and each registered module's declared config, which mesh-config resolves a module's config values from.
//
// **Built from the store, on a tick of their own, by the lease holder alone.** A record changes through a
// verb that runs as a command of its own, a reconcile, a merge's registration or a pending assignment made,
// so no one call site sees every change; what every change has in common is that it is committed to the
// store. So the serving controller reads the store every mirrorEvery and writes only what differs from what
// the bucket holds (ADR 0289 §1): a quiet mesh writes nothing, and a change rolled back is never read. A
// mirror is therefore at most mirrorEvery behind its store.
// mirrorEvery is how often the mirrors are compared with the store: two small reads of the store.
const mirrorEvery = 10 * time.Second
// mirroringRecords keeps the mirrors current for as long as this controller serves: ctx ends when it stops.
func mirroringRecords(ctx context.Context, inv *inventory.Inventory, conn *nats.Conn, announce recordmirror.Announcer) {
api, err := jetstream.New(conn)
if err != nil {
fmt.Printf("the record mirrors are not kept: %v\n", err)
return
}
holds := func() bool {
epoch, err := theLease.epoch(ctx)
return err == nil && epoch > 0
}
epoch := func() uint64 {
e, _ := theLease.epoch(ctx)
return e
}
sources := map[string]func(context.Context) (map[string][]byte, error){
recordmirror.Assignments.Name: inv.AssignmentsMirror,
recordmirror.DeclaredConfig.Name: inv.DeclaredConfigMirror,
}
writers := map[string]*recordmirror.Writer{}
failing := map[string]string{}
tick := time.NewTicker(mirrorEvery)
defer tick.Stop()
for {
if holds() {
for _, m := range recordmirror.Mirrors() {
why := mirrorOnce(ctx, api, m, writers, sources[m.Name], announce, epoch, holds)
if why != failing[m.Name] {
if why != "" {
fmt.Printf("the %s mirror cannot be kept: %s\n", m.Name, why)
} else {
fmt.Printf("the %s mirror is kept again\n", m.Name)
}
failing[m.Name] = why
}
}
}
select {
case <-ctx.Done():
return
case <-tick.C:
}
}
}
// mirrorOnce brings one mirror to what the store holds, and answers why it could not; empty when it could.
func mirrorOnce(ctx context.Context, api jetstream.JetStream, m recordmirror.Mirror, writers map[string]*recordmirror.Writer,
source func(context.Context) (map[string][]byte, error), announce recordmirror.Announcer, epoch func() uint64,
holds func() bool) string {
reading, cancel := context.WithTimeout(ctx, 30*time.Second)
defer cancel()
w := writers[m.Name]
if w == nil {
kv, err := api.KeyValue(reading, m.Bucket)
if err != nil {
return fmt.Sprintf("the bucket %s cannot be read: %v", m.Bucket, err)
}
w = &recordmirror.Writer{Mirror: m, KV: kv, Announce: announce, Epoch: epoch, Holds: holds}
writers[m.Name] = w
}
want, err := source(reading)
if err != nil {
return fmt.Sprintf("the store cannot be read: %v", err)
}
lost := w.LostEvents
r, err := w.Sync(reading, want)
if r.Written+r.Removed > 0 {
fmt.Printf("the %s mirror moved: %d written, %d removed\n", m.Name, r.Written, r.Removed)
}
if w.LostEvents > lost {
fmt.Printf("the %s mirror lost %d event(s): the bucket holds the change, its notice was refused twice\n",
m.Name, w.LostEvents-lost)
}
if err != nil {
return err.Error()
}
return ""
}
+537 -42
View File
@@ -39,8 +39,8 @@ import (
// from the same manifest is no move. // from the same manifest is no move.
// //
// **The release plan** is what walks the waiting moves through: whenever moves wait for a gate that no // **The release plan** is what walks the waiting moves through: whenever moves wait for a gate that no
// started plan is walking, the mesh opens one — every such machine, one at a time, the control node last, // started plan is walking, the mesh opens one — every such machine, the first sent and judged, then the rest
// each sent and judged before the next. A release plan that fails stops the next one opening on its own: // at once, each judged by its own gate (novox/hq ADR 0353, in place of one at a time). A release plan that fails stops the next one opening on its own:
// a person releases it again (`upgrade release-backlog --why`), after the condition says why. // a person releases it again (`upgrade release-backlog --why`), after the condition says why.
// sendScope is what a send may carry that no gate has seen. // sendScope is what a send may carry that no gate has seen.
@@ -81,6 +81,15 @@ type moveFacts struct {
srcs map[string]map[string]string srcs map[string]map[string]string
passed map[string]map[string]bool passed map[string]map[string]bool
plans []inventory.Plan plans []inventory.Plan
// self is the release walking these moves, whose own gates are no other walk's (novox/hq ADR 0353).
self string
}
type walkingKey struct{}
// withWalking says a release plan is the one sending: its own gates never hold its own sends back.
func withWalking(ctx context.Context, id string) context.Context {
return context.WithValue(ctx, walkingKey{}, id)
} }
func readMoveFacts(ctx context.Context, inv *inventory.Inventory) (moveFacts, error) { func readMoveFacts(ctx context.Context, inv *inventory.Inventory) (moveFacts, error) {
@@ -99,6 +108,7 @@ func readMoveFacts(ctx context.Context, inv *inventory.Inventory) (moveFacts, er
return f, err return f, err
} }
f.plans, err = inv.OpenPlans(ctx) f.plans, err = inv.OpenPlans(ctx)
f.self, _ = ctx.Value(walkingKey{}).(string)
return f, err return f, err
} }
@@ -193,17 +203,24 @@ func (f moveFacts) moves(node string, modules []string, sent map[string]string,
// feet. A release keeps no module records: its open gate's carried moves are its walk. // feet. A release keeps no module records: its open gate's carried moves are its walk.
func (f moveFacts) walkedBy(module, node, to string) string { func (f moveFacts) walkedBy(module, node, to string) string {
for _, p := range f.plans { for _, p := range f.plans {
if !p.Open() { if !p.Open() || (p.Release != nil && p.ID == f.self) {
continue continue
} }
if p.Release != nil { if p.Release != nil {
g := p.Release.Gate // The first machine's gate, and every gate of the rest still judging (novox/hq ADR 0353).
if g == nil || g.Verdict != "" { gates := []*inventory.PlanGate{p.Release.Gate}
continue for _, g := range p.Release.Gates {
gates = append(gates, g)
} }
for _, c := range g.Carried { for _, g := range gates {
if c.Module == module && !(slices.Contains(g.Machines, node) && sameCommit(c.To, to)) { // A send kept and not yet made walks nothing yet (ADR 0353 §8).
return p.ID if g == nil || g.Verdict != "" || g.Silent || g.Pending {
continue
}
for _, c := range g.Carried {
if c.Module == module && !(slices.Contains(g.Machines, node) && sameCommit(c.To, to)) {
return p.ID
}
} }
} }
continue continue
@@ -303,8 +320,8 @@ func ungatedIn(ctx context.Context, open *stores, names []string, addedHolder st
} }
} }
if len(refused) > 0 { if len(refused) > 0 {
return nil, fmt.Errorf("%w: %s — a walk sends them, one machine at a time, each judged (novox/hq ADR "+ return nil, fmt.Errorf("%w: %s — a walk sends them, one machine first and then the rest at once, each judged (novox/hq ADR "+
"0236); `upgrade backlog` lists them", errUngated, strings.Join(refused, "; ")) "0236, ADR 0353); `upgrade backlog` lists them", errUngated, strings.Join(refused, "; "))
} }
return kept, nil return kept, nil
} }
@@ -320,19 +337,35 @@ func ungatedIn(ctx context.Context, open *stores, names []string, addedHolder st
// And the grants step first, when the send changes what the bus's user list must say (novox/hq issue 490): // And the grants step first, when the send changes what the bus's user list must say (novox/hq issue 490):
// answered for the gate to keep, nil when there was none. // answered for the gate to keep, nil when there was none.
func gatedSend(ctx context.Context, open *stores, node string, owns []inventory.CarriedMove) ([]inventory.CarriedMove, []string, *inventory.GrantsStep, error) { func gatedSend(ctx context.Context, open *stores, node string, owns []inventory.CarriedMove) ([]inventory.CarriedMove, []string, *inventory.GrantsStep, error) {
inv := open.inventory moves, err := gatedMoves(ctx, open, node, owns)
f, err := readMoveFacts(ctx, inv) if err != nil || len(moves) == 0 {
return nil, nil, nil, err
}
sayRecreations(ctx, open, moves)
sent, grants, err := sendJudged(ctx, open, node)
if err != nil { if err != nil {
return nil, nil, nil, err return nil, nil, nil, err
} }
return moves, sent, grants, nil
}
// gatedMoves is what a gated send to one machine would carry, with the build each moves to, made of nothing:
// gatedSend's reading, which a walk also keeps before it sends (novox/hq ADR 0353 §8). Refused the same way
// while a plan that has started walks one of those builds elsewhere.
func gatedMoves(ctx context.Context, open *stores, node string, owns []inventory.CarriedMove) ([]inventory.CarriedMove, error) {
inv := open.inventory
f, err := readMoveFacts(ctx, inv)
if err != nil {
return nil, err
}
moves, err := machineMoves(ctx, open, f, node, true) moves, err := machineMoves(ctx, open, f, node, true)
if err != nil { if err != nil {
return nil, nil, nil, err return nil, err
} }
// A move the send keeps is not made, and is no move for the gate to judge (novox/hq issue 201): a core module // A move the send keeps is not made, and is no move for the gate to judge (novox/hq issue 201): a core module
// kept over an older record judged as moving back to it would read the gate as superseded by its own send. // kept over an older record judged as moving back to it would read the gate as superseded by its own send.
if moves, err = withoutKept(ctx, open, node, moves); err != nil { if moves, err = withoutKept(ctx, open, node, moves); err != nil {
return nil, nil, nil, err return nil, err
} }
own := func(module string) bool { own := func(module string) bool {
return slices.ContainsFunc(owns, func(o inventory.CarriedMove) bool { return o.Module == module }) return slices.ContainsFunc(owns, func(o inventory.CarriedMove) bool { return o.Module == module })
@@ -342,7 +375,7 @@ func gatedSend(ctx context.Context, open *stores, node string, owns []inventory.
continue continue
} }
if id := f.walkedBy(mv.Module, node, mv.To); id != "" { if id := f.walkedBy(mv.Module, node, mv.To); id != "" {
return nil, nil, nil, fmt.Errorf("%w: %s's build %s waits on %s, which %s is walking", errWalkedElsewhere, return nil, fmt.Errorf("%w: %s's build %s waits on %s, which %s is walking", errWalkedElsewhere,
mv.Module, short(mv.To), node, id) mv.Module, short(mv.To), node, id)
} }
} }
@@ -350,7 +383,7 @@ func gatedSend(ctx context.Context, open *stores, node string, owns []inventory.
// plan, or a release, on this machine is not crossed with a newer build; this send waits for its gate. // plan, or a release, on this machine is not crossed with a newer build; this send waits for its gate.
for _, o := range owns { for _, o := range owns {
if id := f.walkedBy(o.Module, node, o.To); id != "" { if id := f.walkedBy(o.Module, node, o.To); id != "" {
return nil, nil, nil, fmt.Errorf("%w: %s's build %s waits on %s, which %s is walking", errWalkedElsewhere, return nil, fmt.Errorf("%w: %s's build %s waits on %s, which %s is walking", errWalkedElsewhere,
o.Module, short(o.To), node, id) o.Module, short(o.To), node, id)
} }
} }
@@ -363,30 +396,26 @@ func gatedSend(ctx context.Context, open *stores, node string, owns []inventory.
moves[i].Build = o.Build moves[i].Build = o.Build
} }
} }
if len(moves) == 0 && len(owns) == 0 { if len(moves) == 0 {
return nil, nil, nil, nil return nil, nil
} }
for i := range moves { for i := range moves {
if moves[i].Build == "" { if moves[i].Build == "" {
if moves[i].Build, err = inv.BuildOf(ctx, moves[i].Module, moves[i].To); err != nil { if moves[i].Build, err = inv.BuildOf(ctx, moves[i].Module, moves[i].To); err != nil {
return nil, nil, nil, err return nil, err
} }
} }
} }
sort.Slice(moves, func(i, j int) bool { return moves[i].Module < moves[j].Module }) sort.Slice(moves, func(i, j int) bool { return moves[i].Module < moves[j].Module })
sayRecreations(ctx, open, moves) return moves, nil
sent, grants, err := sendJudged(ctx, open, node)
if err != nil {
return nil, nil, nil, err
}
return moves, sent, grants, nil
} }
// passCarried keeps a pass as the verdict of every build the gate judged beside its own module. // passCarried keeps a pass as the verdict of every build the gate judged beside its own module.
func passCarried(ctx context.Context, open *stores, p *inventory.Plan, g *inventory.PlanGate, except string) { func passCarried(ctx context.Context, open *stores, p *inventory.Plan, g *inventory.PlanGate, except string) {
seen := map[string]bool{} seen := map[string]bool{}
for _, c := range g.Carried { for _, c := range g.Carried {
if c.Module == except || c.Build == "" || seen[c.Build] { // A build put back with another machine's failure is not this gate's to pass (novox/hq ADR 0353 §3).
if c.Module == except || c.Build == "" || seen[c.Build] || slices.Contains(g.Returned, c.Module) {
continue continue
} }
seen[c.Build] = true seen[c.Build] = true
@@ -695,7 +724,7 @@ func releaseBacklog(ctx context.Context, open *stores, by string) (*inventory.Pl
p := inventory.Plan{ID: fmt.Sprintf("release-%d", now.UnixNano()), Repository: releaseRepository, p := inventory.Plan{ID: fmt.Sprintf("release-%d", now.UnixNano()), Repository: releaseRepository,
Created: now, State: inventory.PlanRolling, Tiers: [][]string{names}, Modules: map[string]*inventory.PlanModule{}, Created: now, State: inventory.PlanRolling, Tiers: [][]string{names}, Modules: map[string]*inventory.PlanModule{},
Release: &inventory.PlanRelease{Order: order, By: by}, Release: &inventory.PlanRelease{Order: order, By: by},
Note: fmt.Sprintf("%d build(s) wait for a gate on %s; one machine at a time, each judged", len(names), Note: fmt.Sprintf("%d build(s) wait for a gate on %s; one machine first, then the rest at once, each judged", len(names),
strings.Join(order, ", "))} strings.Join(order, ", "))}
if err := inv.SavePlan(ctx, &p); err != nil { if err := inv.SavePlan(ctx, &p); err != nil {
return nil, err return nil, err
@@ -704,14 +733,23 @@ func releaseBacklog(ctx context.Context, open *stores, by string) (*inventory.Pl
return &p, nil return &p, nil
} }
// advanceRelease takes one step of a release plan: the next machine sent everything waiting there under a // advanceRelease takes one step of a release plan: the first machine sent everything waiting there under a
// gate, or the machine being judged judged once more; the plan done when every machine is. // gate and judged; once it passed, every other machine sent at once, each under its own gate (novox/hq ADR
// 0353 §1), and judged; the plan done when every machine is.
//
// **A send is kept before it is made** (ADR 0353 §8): the step that chooses a machine keeps its gate pending with
// what it will carry, and the next step makes the send. A controller replaced between the send and keeping it
// finds the gate pending, reads the machine's last send, and neither sends again nor leaves the machine unjudged.
func advanceRelease(ctx context.Context, open *stores, p *inventory.Plan) (bool, error) { func advanceRelease(ctx context.Context, open *stores, p *inventory.Plan) (bool, error) {
r := p.Release r := p.Release
if r.Settings != nil { if r.Settings != nil {
return advanceSettingsWalk(ctx, open, p) return advanceSettingsWalk(ctx, open, p)
} }
ctx = withWalking(ctx, p.ID)
now := time.Now().UTC() now := time.Now().UTC()
if r.Rest {
return advanceRest(ctx, open, p, now)
}
if r.Gate == nil { if r.Gate == nil {
heard, err := releaseHeard(ctx, open) heard, err := releaseHeard(ctx, open)
if err != nil { if err != nil {
@@ -724,7 +762,7 @@ func advanceRelease(ctx context.Context, open *stores, p *inventory.Plan) (bool,
r.Next++ r.Next++
continue continue
} }
moves, sent, grants, err := gatedSend(ctx, open, node, nil) moves, err := gatedMoves(ctx, open, node, nil)
if errors.Is(err, errWalkedElsewhere) { if errors.Is(err, errWalkedElsewhere) {
note := fmt.Sprintf("waiting before %s: %v", node, err) note := fmt.Sprintf("waiting before %s: %v", node, err)
changed := p.Note != note changed := p.Note != note
@@ -732,23 +770,15 @@ func advanceRelease(ctx context.Context, open *stores, p *inventory.Plan) (bool,
return changed, nil return changed, nil
} }
if err != nil { if err != nil {
return false, fmt.Errorf("sending %s what waits there: %w", node, err) return false, fmt.Errorf("reading what waits on %s: %w", node, err)
} }
if len(moves) == 0 { if len(moves) == 0 {
r.Done = append(r.Done, node) r.Done = append(r.Done, node)
r.Next++ r.Next++
continue continue
} }
r.Gate = &inventory.PlanGate{Machines: sent, Since: &now, Carried: moves, Sent: sentNow(ctx, open.inventory, sent), r.Gate = &inventory.PlanGate{Machines: []string{node}, Carried: moves, Pending: true}
Grants: grants} p.Note = fmt.Sprintf("sending %s %d build(s) that wait for a gate, first", node, len(moves))
p.Note = fmt.Sprintf("sent %s %d build(s) that waited for a gate; judging them there", node, len(moves))
if said := grantsSaid(grants); said != "" {
p.Note += "; " + said
}
if said := recreationsSaid(moves); said != "" {
p.Note += "; " + said
}
fmt.Printf("%s: %s\n", p.ID, p.Note)
return true, nil return true, nil
} }
p.State, p.Tier = inventory.PlanDone, len(p.Tiers) p.State, p.Tier = inventory.PlanDone, len(p.Tiers)
@@ -759,6 +789,19 @@ func advanceRelease(ctx context.Context, open *stores, p *inventory.Plan) (bool,
return true, nil return true, nil
} }
g := r.Gate g := r.Gate
if g.Pending {
node := firstOf(g.Machines)
out, err := sendKept(ctx, open, p, node, g, now)
if err != nil || out == keptWaiting {
return out == keptWaiting, err
}
if out == keptNothing {
r.Gate = nil
r.Done = append(r.Done, node)
r.Next++
}
return true, nil
}
verdict, err := judgeMoves(ctx, open, g, nil, now) verdict, err := judgeMoves(ctx, open, g, nil, now)
if err != nil { if err != nil {
return false, err return false, err
@@ -777,6 +820,11 @@ func advanceRelease(ctx context.Context, open *stores, p *inventory.Plan) (bool,
r.Done = append(r.Done, firstOf(g.Machines)) r.Done = append(r.Done, firstOf(g.Machines))
r.Next++ r.Next++
r.Gate = nil r.Gate = nil
// **Then the rest at once** (novox/hq ADR 0353 §1): every other machine heard from, each judged by its own
// gate, in place of one machine at a time (ADR 0236 §22).
r.Rest = true
p.Note = fmt.Sprintf("passed on %s; the rest are sent at once, each judged", firstOf(g.Machines))
fmt.Printf("%s: %s\n", p.ID, p.Note)
return true, nil return true, nil
case inventory.GateSuperseded: case inventory.GateSuperseded:
// Another send moved a judged module on the judged machine (novox/hq issue 352): no verdict on what // Another send moved a judged module on the judged machine (novox/hq issue 352): no verdict on what
@@ -797,6 +845,453 @@ func advanceRelease(ctx context.Context, open *stores, p *inventory.Plan) (bool,
return true, nil return true, nil
} }
// What sendKept did with a kept send.
const (
keptSent = iota // sent now
keptAlready // the machine was already sent it: judged from that send, nothing sent
keptNothing // nothing waits there any more: no gate
keptWaiting // another walk judges a build it would carry: it waits
)
// sendKept makes a send a gate kept before it was made (novox/hq ADR 0353 §8) — unless the machine's last send
// already carries what the gate is for, made before a controller was replaced and before it was kept: then
// nothing is sent, and the gate judges from that send (§6).
func sendKept(ctx context.Context, open *stores, p *inventory.Plan, node string, g *inventory.PlanGate,
now time.Time) (int, error) {
inv := open.inventory
moves, sent, grants, err := gatedSend(ctx, open, node, nil)
if errors.Is(err, errWalkedElsewhere) {
p.Note = fmt.Sprintf("waiting before %s: %v", node, err)
return keptWaiting, nil
}
if err != nil {
return keptSent, fmt.Errorf("sending %s what waits there: %w", node, err)
}
if len(moves) > 0 {
g.Machines, g.Carried, g.Since, g.Grants, g.Pending = sent, moves, &now, grants, false
g.Sent = sentNow(ctx, inv, sent)
p.Note = fmt.Sprintf("sent %s %d build(s) that waited for a gate; judging them there", node, len(moves))
if said := grantsSaid(grants); said != "" {
p.Note += "; " + said
}
if said := recreationsSaid(moves); said != "" {
p.Note += "; " + said
}
fmt.Printf("%s: %s\n", p.ID, p.Note)
return keptSent, nil
}
if !alreadySent(ctx, inv, node, g.Carried) {
return keptNothing, nil
}
since := now
if s, found, err := inv.SentTo(ctx, node); err == nil && found {
if !s.At.IsZero() {
since = s.At
}
g.Sent = map[string]inventory.SentDeclaration{node: s}
}
g.Since, g.Pending = &since, false
p.Note = fmt.Sprintf("%s was already sent the %d build(s) kept for it, at %s: judged from that send, nothing sent again",
node, len(g.Carried), since.Local().Format("15:04:05"))
fmt.Printf("%s: %s\n", p.ID, p.Note)
return keptAlready, nil
}
// alreadySent says a machine's last send carries every build of the moves: a send made and not kept.
func alreadySent(ctx context.Context, inv *inventory.Inventory, node string, moves []inventory.CarriedMove) bool {
if len(moves) == 0 {
return false
}
sent, known, err := inv.SentBuilds(ctx, node)
if err != nil || !known {
return false
}
for _, mv := range moves {
if !sameCommit(sent[mv.Module], mv.To) {
return false
}
}
return true
}
// advanceRest takes one step of a release plan's rest (novox/hq ADR 0353): the sends kept for a step made, every
// machine of the step judged by its own gate, one that fails put back by itself while the others go on to their
// verdicts, and the next step — or the end — once every gate of the step has its verdict or is silent.
func advanceRest(ctx context.Context, open *stores, p *inventory.Plan, now time.Time) (bool, error) {
r := p.Release
// The sends kept for this step, in the plan's order: the control node, last in it, last (§1).
var pending []string
for _, n := range r.Order {
if g := r.Gates[n]; g != nil && g.Pending {
pending = append(pending, n)
}
}
if len(pending) > 0 {
// **A kept send carries only what was kept** (ADR 0353 §2): a build that passed no gate and came to wait on
// a machine since its step was chosen could reach a second machine in it; the step is chosen again.
if grew, err := keptStepGrew(ctx, open, r, pending); err != nil {
return false, err
} else if grew != "" {
for _, n := range pending {
delete(r.Gates, n)
}
opened, waiting, err := openRestStep(ctx, open, p)
if err != nil {
return true, err
}
p.Note = "chosen again: " + grew
if !opened && waiting != "" {
p.Note += "; " + waiting
}
return true, nil
}
var sent, already, waiting []string
for _, node := range pending {
out, err := sendKept(ctx, open, p, node, r.Gates[node], now)
if err != nil {
return true, err
}
switch out {
case keptSent:
sent = append(sent, node)
case keptAlready:
already = append(already, node)
case keptNothing:
delete(r.Gates, node)
r.Done = append(r.Done, node)
case keptWaiting:
waiting = append(waiting, node)
}
}
var said []string
if len(sent) > 0 {
said = append(said, "sent "+strings.Join(sent, ", ")+" at once, each judged there")
}
if len(already) > 0 {
said = append(said, strings.Join(already, ", ")+" already sent it: judged from that send")
}
if len(waiting) > 0 {
said = append(said, "waiting before "+strings.Join(waiting, ", ")+": another walk judges a build it carries")
}
if len(said) > 0 {
p.Note = strings.Join(said, "; ")
}
return len(sent)+len(already) > 0 || len(waiting) < len(pending), nil
}
reports, err := open.inventory.LastReports(ctx)
if err != nil {
return false, err
}
byNode := map[string]inventory.Reported{}
for _, rep := range reports {
byNode[rep.Node] = rep
}
changed, noteWas := false, p.Note
var notes, judging []string
batched, back := batchingRollbacks(ctx)
for _, node := range r.Order {
g := r.Gates[node]
if g == nil || g.Verdict != "" || g.Silent {
continue
}
// What another machine's failure already put back is not this gate's to judge (ADR 0353 §3).
if len(g.Carried) > 0 && len(g.Returned) >= len(g.Carried) {
decide(g, inventory.GateFailed, "what it was sent was put back with another machine's failure", now)
r.Failed = append(r.Failed, node)
changed = true
continue
}
// **A machine that says nothing on its send is left silent, not failed** (ADR 0353 §4): it is not judged
// until it reports, and past ApplySilentAfter it is listed and left.
if !reportedOnSend(byNode[node], node, g) {
if now.Sub(*g.Since) > inventory.ApplySilentAfter {
g.Silent = true
r.Silent = append(r.Silent, node)
changed = true
notes = append(notes, fmt.Sprintf("%s said nothing on its send within %s: left silent", node,
inventory.ApplySilentAfter))
continue
}
judging = append(judging, node+": not yet reported on its send")
continue
}
verdict, err := judgeMoves(ctx, open, g, nil, now)
if err != nil {
return changed, err
}
switch verdict {
case "":
if putBackBroken(ctx, open, p, g, "", nil) {
changed = true
notes = append(notes, p.Note)
}
judging = append(judging, node+": "+gateLine(g))
case inventory.GatePassed:
passCarried(ctx, open, p, g, "")
r.Done = append(r.Done, node)
changed = true
case inventory.GateSuperseded:
// Another send moved a judged module there (novox/hq issue 352): nothing judged, nothing put back on
// it; what still waits is released again.
r.Done = append(r.Done, node)
changed = true
notes = append(notes, fmt.Sprintf("superseded on %s: %s", node, g.Why))
default:
// **Put back by itself** (ADR 0353 §3): what was found wanting goes back at once, and every other
// machine of the step goes on to its own verdict.
stateWas := p.State
p.Note = ""
failCarried(batched, open, p, g, "")
p.State = stateWas
r.Failed = append(r.Failed, node)
changed = true
notes = append(notes, fmt.Sprintf("failed its gate on %s: %s — %s", node, g.Why, p.Note))
returnedElsewhere(ctx, open, r, node, g)
}
}
p.Note = ""
sendRollbacks(ctx, open, p, back)
if p.Note != "" {
notes = append(notes, strings.TrimPrefix(p.Note, "; "))
}
if len(judging) > 0 {
note := strings.Join(append(notes, "judging "+strings.Join(judging, "; ")), "; ")
p.Note = note
return changed || noteWas != note, nil
}
// Every gate of the step has its verdict or is silent: the next step, unless a machine failed (§3).
if len(r.Failed) == 0 {
opened, waiting, err := openRestStep(ctx, open, p)
if err != nil {
return changed, err
}
if opened {
return true, nil
}
if waiting != "" {
note := strings.Join(append(notes, waiting), "; ")
p.Note = note
return changed || noteWas != note, nil
}
}
p.Tier = len(p.Tiers)
end := fmt.Sprintf("released on %s", orNone(strings.Join(r.Done, ", ")))
if len(r.Silent) > 0 {
end += "; silent on their send, left: " + strings.Join(r.Silent, ", ")
}
if len(r.Skipped) > 0 {
end += "; not heard from, left as they were: " + strings.Join(r.Skipped, ", ")
}
p.State = inventory.PlanDone
if len(r.Failed) > 0 {
p.State = inventory.PlanFailed
end = "failed on " + strings.Join(r.Failed, ", ") + "; " + end
} else if len(r.Silent) > 0 && len(r.Gates) == len(r.Silent) {
// **A rest that said nothing is no success** (ADR 0353 §4): failure is loud; the condition says it.
p.State = inventory.PlanFailed
end = "no machine of the rest reported on its send; " + end
}
p.Note = strings.Join(append(notes, end), "; ")
fmt.Printf("%s: %s\n", p.ID, p.Note)
return true, nil
}
// keptStepGrew says why a step's kept sends would now carry a build no gate passed that they were not kept with;
// empty when none would.
func keptStepGrew(ctx context.Context, open *stores, r *inventory.PlanRelease, pending []string) (string, error) {
f, err := readMoveFacts(ctx, open.inventory)
if err != nil {
return "", err
}
shelf, err := open.inventory.Catalogue(ctx)
if err != nil {
return "", err
}
for _, node := range pending {
moves, err := gatedMoves(ctx, open, node, nil)
if errors.Is(err, errWalkedElsewhere) {
continue // sendKept says it waits
}
if err != nil {
return "", err
}
kept := r.Gates[node].Carried
for _, mv := range moves {
// The node-engine and the resolver keep one machine at a time, passed or not (ADR 0353 §9).
if (f.gated(mv.Module, mv.To) && !oneAtATime(mv.Module, shelf[mv.Module])) || slices.ContainsFunc(kept, func(c inventory.CarriedMove) bool {
return c.Module == mv.Module && sameCommit(c.To, mv.To)
}) {
continue
}
return fmt.Sprintf("%s's build %s, which this step did not keep for it, came to wait on %s", mv.Module, short(mv.To), node), nil
}
}
return "", nil
}
// reportedOnSend says a machine's last report is about the send its gate judges (novox/hq issue 352): the
// declaration kept on the gate, or one sequenced after it; a gate that kept none reads the send made last.
func reportedOnSend(r inventory.Reported, node string, g *inventory.PlanGate) bool {
if r.At == nil {
return false
}
if s, kept := g.Sent[node]; kept {
return s.ReportsOn(r)
}
return r.Current
}
// returnedElsewhere marks, on every other gate of the rest still judging, the builds a failed machine's
// rollback put back wherever they were sent (ADR 0236 §5): no longer theirs to judge (ADR 0353 §3).
func returnedElsewhere(ctx context.Context, open *stores, r *inventory.PlanRelease, failed string, g *inventory.PlanGate) {
for _, c := range g.Carried {
if c.Build == "" {
continue
}
if marked, err := open.inventory.GateFailed(ctx, c.Build); err != nil || !marked {
continue
}
for node, other := range r.Gates {
if node == failed || other.Verdict != "" || slices.Contains(other.Returned, c.Module) {
continue
}
for _, oc := range other.Carried {
if oc.Module == c.Module && sameCommit(oc.To, c.To) {
other.Returned = append(other.Returned, c.Module)
}
}
}
}
}
// openRestStep chooses the next step of a release plan's rest and keeps its sends (novox/hq ADR 0353 §1, §2):
// every machine left that is heard from and has something waiting, except that a build no gate has passed goes
// to one machine of the step only. Answers whether a step was kept, or why the walk waits.
func openRestStep(ctx context.Context, open *stores, p *inventory.Plan) (bool, string, error) {
r := p.Release
heard, err := releaseHeard(ctx, open)
if err != nil {
return false, "", err
}
f, err := readMoveFacts(ctx, open.inventory)
if err != nil {
return false, "", err
}
waiting := map[string][]inventory.CarriedMove{}
var nodes, walked []string
for _, node := range r.Order[min(r.Next, len(r.Order)):] {
if r.Gates[node] != nil || slices.Contains(r.Done, node) || slices.Contains(r.Skipped, node) ||
slices.Contains(r.Failed, node) || slices.Contains(r.Silent, node) {
continue
}
if !heard[node] {
r.Skipped = append(r.Skipped, node)
continue
}
moves, err := gatedMoves(ctx, open, node, nil)
if errors.Is(err, errWalkedElsewhere) {
walked = append(walked, fmt.Sprintf("%s (%v)", node, err))
continue
}
if err != nil {
return false, "", fmt.Errorf("reading what waits on %s: %w", node, err)
}
if len(moves) == 0 {
r.Done = append(r.Done, node)
continue
}
waiting[node] = moves
nodes = append(nodes, node)
}
shelf, err := open.inventory.Catalogue(ctx)
if err != nil {
return false, "", err
}
step := restStep(nodes, waiting, f.gated, func(module string) bool { return oneAtATime(module, shelf[module]) })
if len(step) == 0 {
if len(walked) > 0 {
return false, "waiting before " + strings.Join(walked, "; "), nil
}
return false, "", nil
}
if r.Gates == nil {
r.Gates = map[string]*inventory.PlanGate{}
}
for _, node := range step {
r.Gates[node] = &inventory.PlanGate{Machines: []string{node}, Carried: waiting[node], Pending: true}
}
p.Note = fmt.Sprintf("sending %s at once, each judged there", strings.Join(step, ", "))
if left := len(nodes) - len(step); left > 0 {
p.Note += fmt.Sprintf("; %d machine(s) wait for a build that passed no gate yet to pass on one", left)
}
return true, "", nil
}
// oneAtATime is a module whose builds go to the rest one machine at a time (novox/hq ADR 0353 §9): the
// node-engine, which applies every send, and a module holding the mesh's resolver, which names the bus.
func oneAtATime(module string, m catalogue.Manifest) bool {
if module == hostModule {
return true
}
return slices.ContainsFunc(m.Claims, func(c catalogue.Claim) bool { return c.Name == resolverSeat })
}
// restStep is the machines of one step of the rest (novox/hq ADR 0353 §2), in the order given: each machine, unless
// a build it would carry has passed no gate and an earlier machine of the step already carries it — that build's
// first machine. Pure.
//
// **The node-engine and the mesh's resolver still go one machine at a time** (ADR 0353 §9): a step carries a
// build of either to one machine only, judged before the next, since a fault that cuts a machine off the bus
// shows only as its silence and no rollback reaches it.
func restStep(nodes []string, waiting map[string][]inventory.CarriedMove, passed func(module, commit string) bool,
alone func(module string) bool) []string {
first := map[string]string{}
var out []string
for _, n := range nodes {
var unpassed []string
for _, mv := range waiting[n] {
if !passed(mv.Module, mv.To) || alone(mv.Module) {
unpassed = append(unpassed, mv.Module+"@"+mv.To)
}
}
if slices.ContainsFunc(unpassed, func(k string) bool { _, taken := first[k]; return taken }) {
continue
}
for _, k := range unpassed {
first[k] = n
}
out = append(out, n)
}
return out
}
// silentObservation is what the gate's probe says of the newest release whose rest went silent (novox/hq ADR 0353
// §4): machines that never reported on their send, left as they were — loud, never a success said by nothing.
func silentObservation(ctx context.Context, inv *inventory.Inventory) ([]conditions.Observation, error) {
recent, err := inv.RecentPlans(ctx, 20)
if err != nil {
return nil, err
}
for _, p := range recent {
if p.Release == nil || p.Release.Settings != nil || p.Open() {
continue
}
if len(p.Release.Silent) == 0 {
return nil, nil
}
machines := append([]string(nil), p.Release.Silent...)
sort.Strings(machines)
o := conditions.Observation{Scope: conditions.ScopeMesh, ID: "release", Token: "silent", Kind: "release-silent",
Severity: conditions.Warning, Resolver: conditions.ResolverOperator, Also: machines,
Summary: fmt.Sprintf("%s: %s said nothing on the send of the builds that waited for a gate within %s, and were "+
"left as they were", p.ID, strings.Join(machines, ", "), inventory.ApplySilentAfter)}
w := releaseSilentWords(machines)
o.Headline, o.Explanation, o.Resolved, o.Needs = w.Headline, w.Explanation, w.Resolved, w.Needs
return []conditions.Observation{o}, nil
}
return nil, nil
}
// backlogObservation is what the gate's probe says of a release held after a failure. // backlogObservation is what the gate's probe says of a release held after a failure.
func backlogObservation() []conditions.Observation { func backlogObservation() []conditions.Observation {
if backlogNow.held == "" || len(backlogNow.waiting) == 0 { if backlogNow.held == "" || len(backlogNow.waiting) == 0 {
+183 -10
View File
@@ -251,6 +251,20 @@ func gates(p inventory.Plan, edges []inventory.Edge, rollsOut func(string) bool)
return out return out
} }
// noMoveWhy is what a plan says of a module of its tier whose build was made from the source every machine
// running it runs (novox/hq issue 280): nothing is sent for it.
// noMovePrefix opens it, and is what noMove reads: one prefix for the writer and the reader.
const (
noMovePrefix = "no move"
noMoveWhy = noMovePrefix + ": made from the source every machine running it runs"
)
// noMove is whether the plan marked the module no move and sent it nothing. Read from what the plan says,
// so a plan kept before the mark had a name of its own is read the same (novox/hq issue 648).
func noMove(s *inventory.PlanModule) bool {
return s != nil && s.FirstAt == nil && s.Gate == nil && strings.HasPrefix(s.Why, noMovePrefix)
}
// applied says whether every machine running the module has reported since the module was built. // applied says whether every machine running the module has reported since the module was built.
// appliedEach is applied with a moment of its own for each machine: the reports that count are the ones // appliedEach is applied with a moment of its own for each machine: the reports that count are the ones
// after that machine was sent the build (novox/hq issue 256). // after that machine was sent the build (novox/hq issue 256).
@@ -595,7 +609,7 @@ func advancePlans(ctx context.Context, open *stores) {
return return
} }
defer release() defer release()
// What waits for a gate, released one machine at a time (ADR 0236). // What waits for a gate, released one machine first, then the rest at once (ADR 0236, ADR 0353).
if _, err := releaseBacklog(ctx, open, ""); err != nil { if _, err := releaseBacklog(ctx, open, ""); err != nil {
fmt.Printf("plans: what waits for a gate could not be looked at: %v\n", err) fmt.Printf("plans: what waits for a gate could not be looked at: %v\n", err)
} }
@@ -805,7 +819,7 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan,
} else if same { } else if same {
now := time.Now().UTC() now := time.Now().UTC()
state.SentAt = &now state.SentAt = &now
state.Why = "no move: made from the source every machine running it runs" state.Why = noMoveWhy
fmt.Printf("%s: %s built from the source every machine running it runs: no move, nothing sent\n", p.ID, m) fmt.Printf("%s: %s built from the source every machine running it runs: no move, nothing sent\n", p.ID, m)
continue continue
} }
@@ -995,6 +1009,25 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan,
if state.BuiltAt != nil { if state.BuiltAt != nil {
since = *state.BuiltAt since = *state.BuiltAt
} }
// **A build that is no move is waited on by the send that last carried it** (novox/hq issue 648).
// Nothing was sent for it, so no report follows its marking or its build: a machine that does not
// happen to report again held the walk for ever. Each machine runs it once it answered its last
// send — the one whose builds made it no move — whenever that was.
if noMove(state) {
byNode := map[string]inventory.Reported{}
for _, r := range reports {
byNode[r.Node] = r
}
if ok, on := appliedEach(m, func(node string) time.Time {
if s := byNode[node].Sent; s != nil {
return *s
}
return time.Time{}
}, running, reports); !ok {
waiting = append(waiting, fmt.Sprintf("%s on %s", m, strings.Join(on, ", ")))
}
continue
}
// **Each machine from its own send** (novox/hq issue 256). With one machine first (ADR 0218) // **Each machine from its own send** (novox/hq issue 256). With one machine first (ADR 0218)
// a module is sent twice — the first machine, then the rest — and SentAt is the second. // a module is sent twice — the first machine, then the rest — and SentAt is the second.
// Asked of every machine, the first machine's report, made between the two sends, read as // Asked of every machine, the first machine's report, made between the two sends, read as
@@ -1056,10 +1089,36 @@ func firstSend(ctx context.Context, open *stores, p *inventory.Plan, node string
runningOf map[string][]string) error { runningOf map[string][]string) error {
inv := open.inventory inv := open.inventory
before, beforeKnown, _ := inv.SentBuilds(ctx, node) before, beforeKnown, _ := inv.SentBuilds(ctx, node)
// **A machine already sent what this send is for is not sent again** (novox/hq ADR 0353 §6): sent by this plan
// before a controller was replaced and before it was kept, or by another walk. Judged from that send, and from
// the readings of a gate already judging it there.
// Only a send the machine reported it applied is one it has; one it never reported on is sent, as before
// (issue 280: what it ran before is then that build, and a failure leaves it as it was).
if beforeKnown && slices.IndexFunc(modules, func(m string) bool { return !sameCommit(before[m], p.Modules[m].Commit) }) < 0 &&
appliedLastSend(ctx, inv, node) {
if others, err := gatedMoves(ctx, open, node, nil); err == nil && len(others) == 0 {
adoptSend(ctx, open, p, node, modules, before, runningOf)
return nil
}
}
// **What the machine ran before is kept before the send is made** (novox/hq ADR 0353 §6): a controller replaced
// between the send and keeping it reads it here, and a failure still puts the machine back.
kept := false
for _, m := range modules {
s := p.Modules[m]
if beforeKnown && (s.Previous == "" || !sameCommit(before[m], s.Commit)) && s.Previous != before[m] {
s.Previous, kept = before[m], true
}
}
if kept {
if err := inv.SavePlan(ctx, p); err != nil {
return fmt.Errorf("what %s ran before is not kept, so nothing is sent: %w", node, err)
}
}
owns := make([]inventory.CarriedMove, 0, len(modules)) owns := make([]inventory.CarriedMove, 0, len(modules))
for _, m := range modules { for _, m := range modules {
s := p.Modules[m] s := p.Modules[m]
owns = append(owns, inventory.CarriedMove{Module: m, Node: node, From: before[m], To: s.Commit, Build: s.Build}) owns = append(owns, inventory.CarriedMove{Module: m, Node: node, From: s.Previous, To: s.Commit, Build: s.Build})
} }
// A gated send (ADR 0236): everything waiting on the machine goes with the tier, and the gate judges // A gated send (ADR 0236): everything waiting on the machine goes with the tier, and the gate judges
// all of it there. // all of it there.
@@ -1074,10 +1133,7 @@ func firstSend(ctx context.Context, open *stores, p *inventory.Plan, node string
sentWhat := sentNow(ctx, inv, sent) sentWhat := sentNow(ctx, inv, sent)
for _, m := range modules { for _, m := range modules {
s := p.Modules[m] s := p.Modules[m]
// What the first machine ran before: what a failed gate puts back (ADR 0236). // What the first machine ran before (ADR 0236), kept above before the send.
if beforeKnown {
s.Previous = before[m]
}
s.First, s.FirstAt = sent, &now s.First, s.FirstAt = sent, &now
s.Gate = &inventory.PlanGate{Component: coreComponent(m), Machines: firstRunning(sent, runningOf[m]), s.Gate = &inventory.PlanGate{Component: coreComponent(m), Machines: firstRunning(sent, runningOf[m]),
From: s.Previous, To: s.Commit, Since: &now, Sent: sentWhat} From: s.Previous, To: s.Commit, Since: &now, Sent: sentWhat}
@@ -1106,6 +1162,113 @@ func firstSend(ctx context.Context, open *stores, p *inventory.Plan, node string
return nil return nil
} }
// adoptSend records a first send the machine already had (novox/hq ADR 0353 §6, §7): nothing is sent; the gate counts
// from the machine's last send, and takes over the readings of an open gate of another walk judging the same build
// there — whose carried move also says what the machine ran before, for a rollback.
func adoptSend(ctx context.Context, open *stores, p *inventory.Plan, node string, modules []string,
before map[string]string, runningOf map[string][]string) {
inv := open.inventory
since := time.Now().UTC()
last, found, err := inv.SentTo(ctx, node)
if err == nil && found && !last.At.IsZero() {
since = last.At
}
lead := modules[0]
other, id, from := judgingThere(ctx, inv, p.ID, lead, node, p.Modules[lead].Commit, p.Modules[lead].Build)
for _, m := range modules {
s := p.Modules[m]
// What it ran before: kept by this walk before its send was made (§6); else what the other walk's send
// moved it from; else the build itself, so a failure leaves it as it was (issue 280).
switch {
case s.Previous != "":
case m == lead && from != "":
s.Previous = from
default:
s.Previous = before[m]
}
at := since
s.First, s.FirstAt = []string{node}, &at
s.Gate = &inventory.PlanGate{Component: coreComponent(m), Machines: firstRunning([]string{node}, runningOf[m]),
From: s.Previous, To: s.Commit, Since: &at}
if found {
s.Gate.Sent = map[string]inventory.SentDeclaration{node: last}
}
s.GatedBy = ""
if m == lead {
s.Gate.TakeReadings(other, modules, id)
} else {
s.GatedBy = lead
}
}
p.State = inventory.PlanRolling
p.Note = fmt.Sprintf("tier %d built; %s already runs %s's build: nothing sent again, judged from that send",
p.Tier, node, strings.Join(modules, ", "))
if id != "" {
p.Note += ", going on from the readings of " + id
}
fmt.Printf("%s: %s\n", p.ID, p.Note)
}
// appliedLastSend says a machine reported it applied the declaration it was sent last.
func appliedLastSend(ctx context.Context, inv *inventory.Inventory, node string) bool {
reports, err := inv.LastReports(ctx)
if err != nil {
return false
}
for _, r := range reports {
if r.Node == node {
return r.Current && r.At != nil && r.Outcome == inventory.OutcomeApplied
}
}
return false
}
// judgingThere is an open gate of another walk judging a module's build on a machine (novox/hq ADR 0353 §7), the
// walk's id, and what the machine ran before; nil when none does.
func judgingThere(ctx context.Context, inv *inventory.Inventory, self, module, node, commit, build string) (*inventory.PlanGate, string, string) {
plans, err := inv.OpenPlans(ctx)
if err != nil {
return nil, "", ""
}
// The same build, by its id where both name one, else by its commit.
same := func(b, c string) bool {
if b != "" && build != "" {
return b == build
}
return sameCommit(c, commit)
}
usable := func(g *inventory.PlanGate) bool {
return g != nil && !g.Pending && g.Since != nil && (g.Verdict == "" || g.Verdict == inventory.GatePassed) &&
slices.Contains(g.Machines, node)
}
for _, q := range plans {
if q.ID == self {
continue
}
if r := q.Release; r != nil {
gates := []*inventory.PlanGate{r.Gate}
for _, g := range r.Gates {
gates = append(gates, g)
}
for _, g := range gates {
if !usable(g) {
continue
}
for _, c := range g.Carried {
if c.Module == module && c.Node == node && same(c.Build, c.To) {
return g, q.ID, c.From
}
}
}
continue
}
if s := q.Modules[module]; s != nil && usable(s.Gate) && same(s.Build, s.Commit) {
return s.Gate, q.ID, s.Previous
}
}
return nil, "", ""
}
// passedWith keeps, on a module sent with others, the verdict of the gate that judged the send: the gate // passedWith keeps, on a module sent with others, the verdict of the gate that judged the send: the gate
// on the first of them passed, and its pass was kept for every build it carried (passCarried). // on the first of them passed, and its pass was kept for every build it carried (passCarried).
func passedWith(module string, s *inventory.PlanModule, lead string, g *inventory.PlanGate) { func passedWith(module string, s *inventory.PlanModule, lead string, g *inventory.PlanGate) {
@@ -1594,9 +1757,19 @@ func plansCommand(ctx context.Context, args []string) error {
// A release plan's walk (ADR 0236): machines done, the one judged, those to come. // A release plan's walk (ADR 0236): machines done, the one judged, those to come.
fmt.Printf(" machines in order: %s; done: %s; skipped: %s\n", strings.Join(r.Order, ", "), fmt.Printf(" machines in order: %s; done: %s; skipped: %s\n", strings.Join(r.Order, ", "),
orNone(strings.Join(r.Done, ", ")), orNone(strings.Join(r.Skipped, ", "))) orNone(strings.Join(r.Done, ", ")), orNone(strings.Join(r.Skipped, ", ")))
if r.Gate != nil { if len(r.Failed)+len(r.Silent) > 0 {
fmt.Printf(" %s\n", gateLine(r.Gate)) fmt.Printf(" failed: %s; silent: %s\n", orNone(strings.Join(r.Failed, ", ")), orNone(strings.Join(r.Silent, ", ")))
for _, c := range r.Gate.Carried { }
gates := []*inventory.PlanGate{r.Gate}
for _, n := range r.Order {
gates = append(gates, r.Gates[n])
}
for _, g := range gates {
if g == nil {
continue
}
fmt.Printf(" %s\n", gateLine(g))
for _, c := range g.Carried {
fmt.Printf(" %-22s %s → %s\n", c.Module, short(c.From), short(c.To)) fmt.Printf(" %-22s %s → %s\n", c.Module, short(c.From), short(c.To))
if c.Recreates != "" { if c.Recreates != "" {
fmt.Printf(" %-22s %s\n", "", c.Recreates) fmt.Printf(" %-22s %s\n", "", c.Recreates)
+372
View File
@@ -0,0 +1,372 @@
package main
import (
"context"
"slices"
"strings"
"testing"
"time"
"github.com/novox/mesh-controller/internal/catalogue"
"github.com/novox/mesh-controller/internal/inventory"
)
// novox/hq ADR 0353: after its first machine passes, a build goes to every other machine at once, each judged by
// its own gate and a failing one put back by itself; and a gate's readings outlive the controller that took them —
// a machine already sent what a send is for is not sent again, and its judging goes on from what was read.
//
// Seen 2026-10-14: release-1791930194184991192 sent ace, g14, shanks and the control node one after another in
// eleven minutes; the new controller its last send started sent the control node again the controller build it
// had just applied, and judged it from its first reading.
// aBacklogOnThree is the backlog fixture with a third machine, desk, running app at c1 like the others.
func aBacklogOnThree(t *testing.T) *backlogMesh {
t.Helper()
b := aBacklog(t)
ctx := t.Context()
inv := b.open.inventory
if _, err := inv.AddNode(ctx, "desk"); err != nil {
t.Fatal(err)
}
if _, err := inv.Assign(ctx, "desk", "app"); err != nil {
t.Fatal(err)
}
if err := inv.RecordSent(ctx, nodeID(t, b.open, "desk"), "d-desk", map[string]string{"app": "c1"}); err != nil {
t.Fatal(err)
}
wasHeard := releaseHeard
releaseHeard = func(context.Context, *stores) (map[string]bool, error) {
return map[string]bool{"anchor": true, "laptop": true, "desk": true}, nil
}
t.Cleanup(func() { releaseHeard = wasHeard })
return b
}
// events records every send and every judging, in order.
func events(t *testing.T) *[]string {
t.Helper()
var said []string
wasSend, wasGather := sendRollout, gatherGateFacts
sendRollout = func(ctx context.Context, open *stores, names []string) ([]string, error) {
said = append(said, "send "+strings.Join(names, ","))
return wasSend(ctx, open, names)
}
gatherGateFacts = func(ctx context.Context, open *stores, component string) (gateFacts, error) {
said = append(said, "judge")
return wasGather(ctx, open, component)
}
t.Cleanup(func() { sendRollout, gatherGateFacts = wasSend, wasGather })
return &said
}
// Once the first machine passed, the rest are both sent before either is judged (ADR 0353 §1); before, the second
// was judged to its verdict before the third was sent.
func TestAReleaseSendsTheRestAtOnceAfterTheFirstPasses(t *testing.T) {
judgeLeavesAsTheCore(t)
b := aBacklogOnThree(t)
ctx := t.Context()
said := events(t)
gateEvery = 0
for i := 0; i < 12; i++ {
advancePlans(ctx, b.open)
}
p := b.release(t)
if p.State != inventory.PlanDone || len(p.Release.Done) != 3 {
t.Fatalf("the release is %s: %s %+v", p.State, p.Note, p.Release)
}
desk, laptop := slices.Index(*said, "send desk"), slices.Index(*said, "send laptop")
if desk < 0 || laptop < 0 || slices.Index(*said, "send anchor") > min(desk, laptop) {
t.Fatalf("sends and judgings: %v", *said)
}
if slices.Contains((*said)[min(desk, laptop):max(desk, laptop)], "judge") {
t.Fatalf("the rest were sent one at a time, each judged before the next: %v", *said)
}
if v, found, err := b.open.inventory.GateOf(ctx, "build-late-c2"); err != nil || !found || v.Verdict != inventory.GatePassed {
t.Fatalf("late's pass on laptop was not kept: %+v %v %v", v, found, err)
}
}
// A machine of the rest whose gate fails is put back at once, and the other machine of the step goes on to its own
// verdict (ADR 0353 §3); before, the walk stopped at the failure and the machine after it was never sent.
func TestARestMachineThatFailsIsPutBackAloneAndTheOthersAreJudgedOn(t *testing.T) {
judgeLeavesAsTheCore(t)
b := aBacklogOnThree(t)
ctx := t.Context()
inv := b.open.inventory
gateEvery = 0
b.broken["desk"] = true // desk fails what it is sent in the rest's step
advancePlans(ctx, b.open)
for i := 0; i < 12 && b.release(t).Open(); i++ {
advancePlans(ctx, b.open)
}
p := b.release(t)
if p.State != inventory.PlanFailed || !slices.Equal(p.Release.Failed, []string{"desk"}) {
t.Fatalf("the release is %s: %s %+v", p.State, p.Note, p.Release)
}
if !slices.ContainsFunc(b.sent, func(s []string) bool { return slices.Equal(s, []string{"laptop"}) }) {
t.Fatalf("laptop was never sent: %v", b.sent)
}
if !slices.Contains(p.Release.Done, "laptop") {
t.Fatalf("laptop has no verdict of its own: %+v", p.Release)
}
// late waited on laptop alone and was healthy there: its pass is kept, whatever desk did.
if v, found, err := inv.GateOf(ctx, "build-late-c2"); err != nil || !found || v.Verdict != inventory.GatePassed {
t.Fatalf("late's pass on laptop was not kept: %+v %v %v", v, found, err)
}
// app failed on desk: marked, and put back wherever it was sent (ADR 0236).
if failed, _ := inv.GateFailed(ctx, "build-app-c2"); !failed {
t.Fatal("app's build is not marked failed")
}
if sent, _, _ := inv.SentBuilds(ctx, "desk"); sent["app"] != "c1" {
t.Fatalf("desk was not put back: %v", sent)
}
}
// A walk whose first machine already runs its build under another walk's gate sends nothing and goes on from that
// gate's readings (ADR 0353 §7); before, it sent the machine again and judged from its first reading.
func TestAWalkTakesOverTheReadingsOfAGateJudgingTheSameBuildThere(t *testing.T) {
judgeLeavesAsTheCore(t)
b := aBacklog(t)
ctx := t.Context()
inv := b.open.inventory
advancePlans(ctx, b.open) // the release sends anchor app c2, and reads it once
release := b.release(t)
judging := release.Release.Gate
if judging == nil || judging.Pending || judging.Passes == 0 {
t.Fatalf("the release is not judging anchor: %+v", judging)
}
built := time.Now().UTC()
plan := inventory.Plan{ID: "plan-app", Repository: "novox/mesh-catalog", Branch: "main", Commit: "c2",
Created: built, State: inventory.PlanBuilding, Tiers: [][]string{{"app"}},
Modules: map[string]*inventory.PlanModule{"app": {State: "built", BuiltAt: &built, Commit: "c2",
Build: "build-app-c2"}}}
if err := inv.SavePlan(ctx, &plan); err != nil {
t.Fatal(err)
}
advancePlans(ctx, b.open)
if len(b.sent) != 1 {
t.Fatalf("anchor was sent again what it runs: %v", b.sent)
}
p, err := inv.PlanByID(ctx, "plan-app")
if err != nil {
t.Fatal(err)
}
g := p.Modules["app"].Gate
if g == nil || g.Since == nil || !g.Since.Equal(*judging.Since) || g.Passes < judging.Passes || g.TookOver != release.ID {
t.Fatalf("the walk's gate starts from zero: %+v (the release's: %+v)", g, judging)
}
if p.Modules["app"].Previous != "c1" {
t.Fatalf("what anchor ran before is %q, not c1", p.Modules["app"].Previous)
}
}
// A walk's first send that was made and not kept — the controller replaced between the two — is not made again
// by the controller that takes over: the gate counts from that send (ADR 0353 §5, §6).
func TestAFirstSendMadeAndNotKeptIsNotSentAgain(t *testing.T) {
judgeLeavesAsTheCore(t)
g := aGateMesh(t)
ctx := t.Context()
// The old controller sends anchor app c2 and is replaced before it keeps the send.
made := sendRollout
sendRollout = func(ctx context.Context, open *stores, names []string) ([]string, error) {
if _, err := made(ctx, open, names); err != nil {
return nil, err
}
return nil, context.Canceled
}
advancePlans(ctx, g.open)
sendRollout = made
if len(g.sent) != 1 || g.plan(t).Modules["app"].FirstAt != nil {
t.Fatalf("the fixture's send was not made, or was kept: %v", g.sent)
}
last, _, err := g.open.inventory.SentTo(ctx, "anchor")
if err != nil {
t.Fatal(err)
}
advancePlans(ctx, g.open)
if len(g.sent) != 1 {
t.Fatalf("the send was made again: %v", g.sent)
}
s := g.plan(t).Modules["app"]
if s.FirstAt == nil || s.Gate == nil || s.Gate.Since == nil || !s.Gate.Since.Equal(last.At) {
t.Fatalf("the gate does not count from the send made: %+v %+v", s, s.Gate)
}
// What anchor ran before was kept before the send, so a failure still puts it back (§5).
if s.Previous != "c1" || s.Gate.From != "c1" {
t.Fatalf("what anchor ran before is %q (gate from %q), not c1", s.Previous, s.Gate.From)
}
}
// A release's send kept before it was made, and made, is judged by the controller that takes over without being
// made again (ADR 0353 §8).
func TestAReleaseSendKeptAndMadeIsNotMadeAgain(t *testing.T) {
judgeLeavesAsTheCore(t)
b := aBacklog(t)
ctx := t.Context()
inv := b.open.inventory
p, err := releaseBacklog(ctx, b.open, "")
if err != nil || p == nil {
t.Fatalf("no release opened: %v", err)
}
if _, err := advanceRelease(ctx, b.open, p); err != nil || p.Release.Gate == nil || !p.Release.Gate.Pending {
t.Fatalf("the send was not kept before it was made: %+v %v", p.Release, err)
}
if err := inv.SavePlan(ctx, p); err != nil {
t.Fatal(err)
}
// The send is made, and the controller is replaced before it keeps it.
if _, err := sendRollout(ctx, b.open, []string{"anchor"}); err != nil {
t.Fatal(err)
}
advancePlans(ctx, b.open)
if len(b.sent) != 1 {
t.Fatalf("anchor was sent again: %v", b.sent)
}
g := b.release(t).Release.Gate
if g == nil || g.Pending || g.Since == nil || len(g.Carried) == 0 {
t.Fatalf("anchor is not judged on the send made: %+v", g)
}
}
// A build no gate has passed goes to one machine of a step, its first; a machine that would be its second waits
// for the next step, and one whose every build passed goes now (ADR 0353 §2).
func TestABuildThatPassedNoGateGoesToOneMachineOfAStep(t *testing.T) {
waiting := map[string][]inventory.CarriedMove{
"desk": {{Module: "app", To: "c2"}, {Module: "new", To: "n1"}},
"laptop": {{Module: "app", To: "c2"}, {Module: "new", To: "n1"}},
"shanks": {{Module: "app", To: "c2"}},
"novox": {{Module: "app", To: "c2"}, {Module: "ctl", To: "k1"}},
}
passed := func(module, commit string) bool { return module == "app" && commit == "c2" }
never := func(string) bool { return false }
step := restStep([]string{"desk", "laptop", "shanks", "novox"}, waiting, passed, never)
if !slices.Equal(step, []string{"desk", "shanks", "novox"}) {
t.Fatalf("the step is %v: laptop would be new's second machine before it passed", step)
}
if step := restStep([]string{"laptop"}, waiting, func(string, string) bool { return true }, never); !slices.Equal(step, []string{"laptop"}) {
t.Fatalf("a machine whose every build passed waits: %v", step)
}
}
// The node-engine and the mesh's resolver go to the rest one machine at a time, even passed (ADR 0353 §9): a step
// carries either to one machine only, while what else passed goes everywhere at once.
func TestTheNodeEngineAndTheResolverGoToTheRestOneAtATime(t *testing.T) {
waiting := map[string][]inventory.CarriedMove{
"desk": {{Module: hostModule, To: "h2"}},
"laptop": {{Module: hostModule, To: "h2"}, {Module: "app", To: "c2"}},
"shanks": {{Module: "app", To: "c2"}},
"novox": {{Module: "dns", To: "d2"}},
"ace": {{Module: "dns", To: "d2"}},
}
resolver := catalogue.Manifest{Module: "dns", Claims: []catalogue.Claim{{Name: resolverSeat}}}
alone := func(m string) bool {
if m == "dns" {
return oneAtATime(m, resolver)
}
return oneAtATime(m, catalogue.Manifest{Module: m})
}
step := restStep([]string{"desk", "laptop", "shanks", "novox", "ace"}, waiting,
func(string, string) bool { return true }, alone)
if !slices.Equal(step, []string{"desk", "shanks", "novox"}) {
t.Fatalf("the step is %v: the node-engine and the resolver each reach one machine of it", step)
}
if oneAtATime("app", catalogue.Manifest{Module: "app"}) {
t.Fatal("a leaf module is held to one machine at a time")
}
}
// A first machine that fails still stops the walk (ADR 0236 §5, kept by ADR 0353 §3): the rest are never sent.
func TestAFirstMachineThatFailsStillStopsTheWalk(t *testing.T) {
b := aBacklogOnThree(t)
ctx := t.Context()
gateEvery = 0
b.broken["anchor"] = true
advancePlans(ctx, b.open)
for i := 0; i < 6 && b.release(t).Open(); i++ {
advancePlans(ctx, b.open)
}
if p := b.release(t); p.State != inventory.PlanFailed || p.Release.Rest {
t.Fatalf("the release is %s: %s %+v", p.State, p.Note, p.Release)
}
for _, s := range b.sent {
if !slices.Equal(s, []string{"anchor"}) {
t.Fatalf("sent %v: the rest after a failed first machine", b.sent)
}
}
}
// The control node is sent last in a step of the rest (ADR 0353 §1): the machine holding the controller is
// ordered after every other.
func TestTheControlNodeIsSentLastInAStep(t *testing.T) {
judgeLeavesAsTheCore(t)
b := aBacklogOnThree(t)
ctx := t.Context()
register(t, b.open, catalogue.Manifest{Module: catalogue.ControllerSeatName, Version: "1"})
if _, err := b.open.inventory.Assign(ctx, "anchor", catalogue.ControllerSeatName); err != nil {
t.Fatal(err)
}
said := events(t)
gateEvery = 0
for i := 0; i < 12 && (len(b.sent) == 0 || b.release(t).Open()); i++ {
advancePlans(ctx, b.open)
}
if p := b.release(t); !slices.Equal(p.Release.Order, []string{"desk", "laptop", "anchor"}) {
t.Fatalf("the order is %v", p.Release.Order)
}
laptop, anchor := slices.Index(*said, "send laptop"), slices.Index(*said, "send anchor")
if laptop < 0 || anchor < laptop || slices.Contains((*said)[laptop:anchor], "judge") {
t.Fatalf("the control node is not sent last in the rest's step: %v", *said)
}
}
// A machine of the rest that says nothing on its send is left silent after fifteen minutes: not failed, not put
// back; a rest that went silent whole ends failed, said as a condition (ADR 0353 §4).
func TestARestMachineSilentPastItsBoundIsLeftAndSaid(t *testing.T) {
judgeLeavesAsTheCore(t)
b := aBacklog(t)
ctx := t.Context()
inv := b.open.inventory
reporting := sendRollout
sendRollout = func(ctx context.Context, open *stores, names []string) ([]string, error) {
if slices.Equal(names, []string{"laptop"}) { // laptop sleeps: sent, and says nothing
b.sent = append(b.sent, names)
current, _ := open.inventory.CurrentBuilds(ctx)
return names, open.inventory.RecordSent(ctx, nodeID(t, open, "laptop"), "d-laptop-asleep",
map[string]string{"app": current["app"].Commit, "late": current["late"].Commit, "same": current["same"].Commit})
}
return reporting(ctx, open, names)
}
t.Cleanup(func() { sendRollout = reporting })
gateEvery = 0
for i := 0; i < 8; i++ {
advancePlans(ctx, b.open)
}
p := b.release(t)
g := p.Release.Gates["laptop"]
if !p.Open() || g == nil || g.Verdict != "" || g.Silent {
t.Fatalf("laptop is judged before fifteen minutes: %s %s %+v", p.State, p.Note, g)
}
since := time.Now().Add(-inventory.ApplySilentAfter - time.Minute)
g.Since = &since
if err := inv.SavePlan(ctx, &p); err != nil {
t.Fatal(err)
}
advancePlans(ctx, b.open)
p = b.release(t)
if p.State != inventory.PlanFailed || !slices.Equal(p.Release.Silent, []string{"laptop"}) || len(p.Release.Failed) > 0 {
t.Fatalf("the release is %s: %s %+v", p.State, p.Note, p.Release)
}
if failed, _ := inv.GateFailed(ctx, "build-late-c2"); failed {
t.Fatal("a silent machine's build was marked failed")
}
if sent, _, _ := inv.SentBuilds(ctx, "laptop"); sent["late"] != "c2" {
t.Fatalf("laptop was put back: %v", sent)
}
obs, err := silentObservation(ctx, inv)
if err != nil {
t.Fatal(err)
}
if len(obs) != 1 || obs[0].Kind != "release-silent" || !slices.Equal(obs[0].Also, []string{"laptop"}) {
t.Fatalf("the silence is not said: %+v", obs)
}
}
+28
View File
@@ -457,6 +457,28 @@ func (a *verbArguments) commandLine() ([]string, error) {
return nil, err return nil, err
} }
return []string{"forget-checks", str("owner"), str("repo"), str("number")}, nil return []string{"forget-checks", str("owner"), str("repo"), str("number")}, nil
case "try":
// A branch tried before its pull request (novox/hq issue 686): asked, or read by the id it answered.
if id := str("id"); id != "" {
return []string{"try", "--id", id}, nil
}
if a.given["repository"] == "" && a.given["ref"] == "" {
// Neither shape named: the command says its usage, which names both, as `build` does.
return []string{"try"}, nil
}
if err := need("repository", "ref"); err != nil {
return nil, err
}
argv := []string{"try", str("repository"), str("ref")}
for _, flag := range []string{"base", "tests"} {
if v := str(flag); v != "" {
argv = append(argv, "--"+flag, v)
}
}
if on("store-only") {
argv = append(argv, "--store-only")
}
return argv, nil
case "cancel", "kill": case "cancel", "kill":
if err := need("id"); err != nil { if err := need("id"); err != nil {
return nil, err return nil, err
@@ -1018,6 +1040,8 @@ var jsonVerbs = map[string]bool{"status": true, "seats": true, "plan": true, "co
"delivery": true, "delivery": true,
// The build queue, each ask with its position among the waiting (novox/hq issue 455). // The build queue, each ask with its position among the waiting (novox/hq issue 455).
"queue": true, "queue": true,
// A branch tried before its pull request (novox/hq issue 686).
"try": true,
// A plan followed (novox/hq issue 229): `plans wait` answers JSON; its other forms say text, and so answer no data. // A plan followed (novox/hq issue 229): `plans wait` answers JSON; its other forms say text, and so answer no data.
"plans": true} "plans": true}
@@ -1308,6 +1332,10 @@ func seatToolHandlers() (map[string]link.ToolHandler, []string, error) {
link.AnswersWithin(ctx, bound+planWaitStart) link.AnswersWithin(ctx, bound+planWaitStart)
} }
} }
if verb == "try" && len(argv) > 1 && argv[1] == "--id" {
// Reading a try holds its answer for the verdict, up to its bound (novox/hq issue 686, 229).
link.AnswersWithin(ctx, tryWait+planWaitStart)
}
if answer, read := readHere(ctx, argv); read { if answer, read := readHere(ctx, argv); read {
return answer, nil return answer, nil
} }
+353
View File
@@ -0,0 +1,353 @@
package main
import (
"context"
"encoding/json"
"errors"
"flag"
"fmt"
"net/url"
"os"
"regexp"
"strings"
"time"
"github.com/nats-io/nats.go"
"github.com/novox/mesh-controller/internal/broker"
"github.com/novox/mesh-controller/internal/link"
)
// A branch is tried before its pull request (novox/hq issue 686).
//
// A repository's store tests skip without a store, and no agent may start a database on its machine. So a
// builder pushed, a reviewer read, and only the pull request's merge check ran those tests for the first time:
// each failure there cost a round of fifteen to thirty minutes (mesh-controller #283, #310, #312).
//
// `try <owner>/<repository> <ref>` asks the build seat for the repository's own check of a pushed branch or
// commit, as the merge check runs it (internal/builder/check.go): the ref merged into the trunk's head, the
// trunk's merge-check.sh, each part in its toolchain's container, against a throwaway store and bus. It is a
// check and nothing more:
//
// - **no pull request, so no status**: its outcome is never said as `checked`, which is what the forge's
// holder sets statuses from (isTry, in Built);
// - **nothing registered, nothing built**: it names no module of the graph, so the merge gate does not run and
// no build is made for a merge (novox/hq ADR 0350's scratch builds are a pull request's check's alone);
// - **a check in the build queue**: after deliveries (link.KindOf, novox/hq issue 558), and bounded by the
// check's own bound (checkTimeout);
// - **the code a pull request's check would run anyway**: only a repository of the mesh, whose every pull
// request the build seat already checks, in containers that hold no runtime socket.
//
// A check outlasts a seat verb's one answer, so the ask answers its id at once and `try --id <id>` reads what
// came of it from the build's log, where the build seat says it whole (link.StepTried): holding its answer up
// to tryWait for a verdict, like `plans wait` (novox/hq issue 229). Nothing is kept beside the log.
// tryWait is how long reading a try holds its answer for the verdict. A variable so a test need not wait.
var tryWait = 20 * time.Second
// A ref as the forge is asked for it — a branch or a commit, never an option — and a `go test -run` pattern.
var (
tryRef = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._/-]*$`)
tryTests = regexp.MustCompile(`^[A-Za-z0-9_^$|().*+?/\[\]-]+$`)
)
// tryAnswer is what `try` answers, asking or reading.
type tryAnswer struct {
ID string `json:"id"`
// State is asked (just now), waiting (nothing said of it yet: in the build queue), running, or the verdict:
// pass, warning (the repository declares no merge-check.sh), fail, or error (it could not run).
State string `json:"state"`
Said string `json:"said"`
Repository string `json:"repository,omitempty"`
Ref string `json:"ref,omitempty"`
Commit string `json:"commit,omitempty"`
Base string `json:"base,omitempty"`
Trunk string `json:"trunk,omitempty"`
On string `json:"on,omitempty"`
Took string `json:"took,omitempty"`
Failed []string `json:"failed,omitempty"`
// Last is the last thing the build seat said of one still running.
Last string `json:"last,omitempty"`
}
// isTry says a check's outcome is a try's: marked so, or — echoed by a build seat from before the mark — a
// check of no pull request and no group, which is nobody's status either.
func isTry(result link.BuildResult) bool {
c := result.Checked
return c != nil && (c.Try || (c.Number == 0 && c.Group == ""))
}
// asATry makes a check's ask a try's: the repository's own script alone, for no pull request.
func asATry(request link.BuildRequest, tests string, storeOnly bool) link.BuildRequest {
c := *request.Check
c.Number, c.Try, c.Tests, c.StoreOnly = 0, true, tests, storeOnly
c.Paths, c.Modules, c.New, c.Dependents, c.Manifests, c.Builds = nil, nil, nil, nil, nil, nil
c.Judge, c.Plan, c.Group, c.Members = "", nil, "", nil
request.Check = &c
return request
}
// maxTries is how many tries may wait or run at once, mesh-wide: each holds a build machine for minutes.
const maxTries = 4
// triesIn reads the build queue for a new try of one commit: the id of a try of that commit already waiting or
// running — the answer to a second ask — and whether there is room for another.
func triesIn(q link.Queue, owner, repo, commit string) (same string, room bool) {
n := 0
for _, a := range q.Asks {
if !link.IsTry(a.Request) || (a.State != link.AskWaiting && a.State != link.AskInFlight) {
continue
}
n++
c := a.Request.Check
if same == "" && strings.EqualFold(c.Owner, owner) && strings.EqualFold(c.Repo, repo) && a.Request.Ref == commit {
same = a.ID
}
}
return same, n < maxTries
}
// forgeCommit asks the forge, through its module's tools on the bus, which commit a branch or commit of a
// repository is, and the repository's default branch. A variable so a test needs no forge.
var forgeCommit = func(ctx context.Context, owner, repo, ref string) (commit, trunk string, err error) {
js, err := aBus()
if err != nil {
return "", "", err
}
defer js.Close()
bus := link.OverNATS{Conn: js.Conn()}
ask := func(path string, into any) error {
raw, _ := json.Marshal(map[string]any{"path": path})
answer, err := link.Ask(ctx, bus, "gitea", "gitea_api", raw, 30*time.Second)
if err != nil {
return err
}
if answer.Error != "" {
return errors.New(answer.Error)
}
return json.Unmarshal(answer.Result, into)
}
var commits struct {
Result []struct {
SHA string `json:"sha"`
} `json:"result"`
}
if err := ask("/repos/"+owner+"/"+repo+"/commits?limit=1&stat=false&verification=false&files=false&sha="+
url.QueryEscape(ref), &commits); err != nil {
return "", "", err
}
if len(commits.Result) == 0 || commits.Result[0].SHA == "" {
return "", "", errors.New("the forge named no commit for it")
}
var found struct {
Result struct {
DefaultBranch string `json:"default_branch"`
} `json:"result"`
}
if err := ask("/repos/"+owner+"/"+repo, &found); err != nil {
return "", "", err
}
return commits.Result[0].SHA, found.Result.DefaultBranch, nil
}
// tryCommand is `try <owner>/<repository> <ref> [--base <branch>] [--tests <pattern>] [--store-only]`, and
// `try --id <id>` to read what came of one.
func tryCommand(ctx context.Context, args []string) error {
set := flag.NewFlagSet("try", flag.ContinueOnError)
base := set.String("base", "", "the branch it is merged into (the repository's default branch when none)")
tests := set.String("tests", "", "only the tests this `go test -run` pattern names")
storeOnly := set.Bool("store-only", false, "only the repository's store tests")
id := set.String("id", "", "a try's id: what came of it")
positionals, err := parseAround(set, args)
if err != nil {
return err
}
usage := errors.New("try <owner>/<repository> <branch or commit> [--base <branch>] [--tests <pattern>] " +
"[--store-only], or try --id <id>")
if *id != "" {
if len(positionals) != 0 {
return usage
}
// Before it is a token of a bus subject: an id, never a pattern.
if !link.SafeBuildID(*id) {
return fmt.Errorf("%q is not a try's id", *id)
}
return tryRead(ctx, *id)
}
if len(positionals) != 2 {
return usage
}
owner, repo, isPath := strings.Cut(positionals[0], "/")
ref := positionals[1]
switch {
case !isPath || !tryRef.MatchString(owner) || !tryRef.MatchString(repo) || strings.Contains(repo, "/"):
return fmt.Errorf("%q is not a repository on the forge, owner/name; nothing was asked", positionals[0])
case !tryRef.MatchString(ref):
return fmt.Errorf("%q is not a branch or a commit; nothing was asked", ref)
case *base != "" && !tryRef.MatchString(*base):
return fmt.Errorf("%q is not a branch; nothing was asked", *base)
case *tests != "" && !tryTests.MatchString(*tests):
return fmt.Errorf("%q is not a `go test -run` pattern of test names; nothing was asked", *tests)
}
// **An unknown ref is refused here, loudly** — not sent to a machine to fail at its clone minutes later.
commit, trunk, err := forgeCommit(ctx, owner, repo, ref)
if err != nil {
return fmt.Errorf("the forge holds no branch or commit %s in %s/%s — is it pushed? — so nothing was asked: %w",
ref, owner, repo, err)
}
if *base != "" {
trunk = *base
}
if trunk == "" {
return fmt.Errorf("the forge named no default branch for %s/%s, and no --base says what %s is merged into; "+
"nothing was asked", owner, repo, ref)
}
open, err := openStores(ctx)
if err != nil {
return err
}
defer open.Close()
entries, read, edges, err := theGraph(ctx, open.inventory)
if err != nil {
return err
}
p := link.PullUpdated{Owner: owner, Repo: repo, Base: trunk, Commit: commit}
// Whose repository it is, as a pull request's check reckons it: only the mesh's is tried, the code the
// build seat runs for every pull request already.
mesh := pullScope(p, entries, read, edges)
if !mesh.Mesh {
return fmt.Errorf("%s/%s is no repository of the mesh — no module is built from it and its owner is not "+
"the core's — and only the mesh's are tried on the build seat; nothing was asked", owner, repo)
}
request, err := checkRequestFor(ctx, open, p, checkScope{Mesh: true, From: mesh.From}, entries)
if err != nil {
return err
}
request = asATry(request, *tests, *storeOnly)
seat := buildSeatHeld(ctx)
// **Bounded in number** (novox/hq issue 686): one try of a commit at a time, and maxTries in all.
js, err := dialTheBus()
if err != nil {
return err
}
queue, err := link.ReadQueue(ctx, js, seat)
js.Close()
if err != nil {
return err
}
same, room := triesIn(queue, owner, repo, commit)
if same != "" {
return printTry(tryAnswer{ID: same, State: "asked", Repository: owner + "/" + repo, Ref: ref, Commit: commit,
On: seat, Said: fmt.Sprintf("a try of %s/%s at %.12s already waits or runs, as %s: nothing new was asked. "+
"Read it with try id=%s", owner, repo, commit, same, same)})
}
if !room {
return fmt.Errorf("%d tries already wait or run on the build seat, the most at once (`queue` lists them); "+
"nothing was asked — read one of yours, or ask again when one has ended", maxTries)
}
ask, err := askOverOn(seat)
if err != nil {
return err
}
defer ask.Close()
if err := ask.Ask(ctx, request); err != nil {
return err
}
return printTry(tryAnswer{ID: request.ID, State: "asked", Repository: owner + "/" + repo, Ref: ref, Commit: commit,
Base: trunk, On: seat, Said: fmt.Sprintf("asked %s to try %s/%s at %.12s (%s) merged into %s: its own "+
"merge-check.sh, as a check in the build queue, bounded at %s; no status is set and nothing is registered. "+
"Read it with try id=%s — a full run of this kind takes minutes", seat, owner, repo, commit, ref, trunk,
checkTimeout, request.ID)})
}
// tryRead answers what came of a try, holding the answer up to tryWait for its verdict.
func tryRead(ctx context.Context, id string) error {
until := time.Now().Add(tryWait)
for {
lines, err := buildLines(ctx, id)
if err != nil {
return err
}
a := tryAnswerOf(id, lines)
if (a.State != "waiting" && a.State != "running") || !time.Now().Add(2*time.Second).Before(until) {
return printTry(a)
}
if err := sleepOrDone(ctx, 2*time.Second); err != nil {
return err
}
}
}
// tryAnswerOf reads a try's answer from what the build seat said of it: its outcome under StepTried; `failed`
// for one that could not run — an error, never a pass; else how far it is.
func tryAnswerOf(id string, lines []link.BuildLine) tryAnswer {
a := tryAnswer{ID: id, State: "waiting", Said: "nothing is said of " + id + " yet: it waits in the build queue " +
"(`queue` says where), or no try has that id; ask again"}
for _, l := range lines {
switch l.Step {
case link.StepTried:
var t link.Tried
if err := json.Unmarshal([]byte(l.Message), &t); err != nil || t.Verdict == "" {
return tryAnswer{ID: id, State: "error", Said: "the build seat's word on it cannot be read: " + l.Message}
}
return tryAnswer{ID: id, State: t.Verdict, Said: t.Summary, Took: t.Took, Commit: t.Head, Trunk: t.Trunk,
Failed: t.Failed}
case "failed":
return tryAnswer{ID: id, State: "error", Said: "it could not run: " + l.Message}
case "checked":
// A check that is no try, or a build seat from before the verb: its verdict is not a try's.
return tryAnswer{ID: id, State: "error", Said: "the build seat checked it and said no try's outcome — " +
"it predates the verb (novox/hq issue 686), or " + id + " is no try: " + l.Message + "; `builds` with " +
"the id says every line"}
}
a.State, a.Last = "running", l.Message
a.Said = "still running; ask again. `builds` with the id says every line so far"
}
return a
}
func printTry(a tryAnswer) error {
fmt.Fprintln(os.Stderr, a.Said)
for _, l := range a.Failed {
fmt.Fprintln(os.Stderr, " "+l)
}
out, err := json.MarshalIndent(a, "", " ")
if err != nil {
return err
}
fmt.Println(string(out))
return nil
}
// buildLines is everything the build seat said of one build, as the bus keeps it, in order.
func buildLines(ctx context.Context, id string) ([]link.BuildLine, error) {
js, err := aBus()
if err != nil {
return nil, fmt.Errorf("cannot reach the bus to read a build's log: %w", err)
}
defer js.Close()
subject := link.BuildLogOf("*", id)
sub, err := js.Context().PullSubscribe(subject, "",
nats.BindStream(broker.EventsStream), nats.DeliverAll(), nats.AckNone())
if err != nil {
return nil, fmt.Errorf("cannot read %s from the bus: %w", subject, err)
}
defer func() { _ = sub.Unsubscribe() }()
var lines []link.BuildLine
for {
batch, err := sub.Fetch(200, nats.MaxWait(2*time.Second))
if err != nil && !errors.Is(err, nats.ErrTimeout) && !errors.Is(err, context.DeadlineExceeded) {
return nil, fmt.Errorf("reading a build's log: %w", err)
}
for _, msg := range batch {
var line link.BuildLine
if json.Unmarshal(msg.Data, &line) == nil {
lines = append(lines, line)
}
}
if len(batch) < 200 {
return lines, nil
}
}
}
+194
View File
@@ -0,0 +1,194 @@
package main
import (
"context"
"encoding/json"
"errors"
"reflect"
"strings"
"testing"
"github.com/novox/mesh-controller/internal/catalogue"
"github.com/novox/mesh-controller/internal/link"
)
// A branch is tried before its pull request (novox/hq issue 686): the repository's own check asked for a pushed
// branch or commit, with no pull request.
// What a try asks is a check — after deliveries in the build queue, bounded as one — of the repository's own
// script alone: no module of the graph, so no gate; nothing built for a merge, so no reusable image.
func TestATryAsksACheckOfTheRepositorysOwnScriptAlone(t *testing.T) {
request := link.BuildRequest{ID: "build-1", Ref: "c0ffee", Check: &link.CheckRequest{Owner: "novox", Repo: "a",
Number: 7, Base: "main", Modules: []string{"a"}, New: []string{"b"}, Manifests: []string{"module.json"},
Dependents: []string{"c"}, Builds: []link.CheckBuild{{Module: "a"}}, Judge: link.JudgeSelf,
Plan: &link.ChangePlan{}}}
tried := asATry(request, "^TestStore", true)
c := tried.Check
if !c.Try || c.Number != 0 || c.Tests != "^TestStore" || !c.StoreOnly {
t.Fatalf("not marked a try, or it kept a pull request's number: %+v", c)
}
if len(c.Modules)+len(c.New)+len(c.Manifests)+len(c.Dependents)+len(c.Builds) != 0 || c.Judge != "" || c.Plan != nil {
t.Fatalf("a try runs the gate or builds for a merge: %+v", c)
}
if c.Base != "main" || c.Owner != "novox" || tried.Ref != "c0ffee" {
t.Fatalf("a try lost what it merges into what: %+v", tried)
}
if link.KindOf(tried) != link.KindCheck {
t.Fatal("a try does not count as a check in the build queue")
}
}
// A try's outcome is heard and left there: no `checked` is said, so the forge's holder sets no status, and
// nothing is recorded or registered (the handler holds no store here, and would fail reaching for one).
func TestATrysOutcomeSetsNoStatusAndRegistersNothing(t *testing.T) {
said := &saidOnTheBus{}
checkEvents = said
defer func() { checkEvents = nil }()
for name, result := range map[string]link.BuildResult{
"a try": {ID: "build-1", Ref: "c0ffee", Checked: &link.CheckRequest{Owner: "novox", Repo: "a", Try: true},
Check: &link.CheckOutcome{Verdict: "pass", RepoCheck: &link.CheckLayer{Verdict: "fail", Summary: "failed"}}},
"a try that could not run": {ID: "build-2", Checked: &link.CheckRequest{Owner: "novox", Repo: "a", Try: true},
Failed: "cannot clone"},
// A build seat from before the mark echoes the ask without it: a check of no pull request and no
// group is still nobody's status.
"a try echoed by an older build seat": {ID: "build-3", Checked: &link.CheckRequest{Owner: "novox", Repo: "a"},
Check: &link.CheckOutcome{Verdict: "pass"}},
} {
if err := (builds{}).Built(context.Background(), result); err != nil {
t.Fatalf("%s: %v", name, err)
}
}
if len(said.events) != 0 {
t.Fatalf("a try's outcome was said for the forge: %v", said.events)
}
// A pull request's check is still said.
_ = (builds{}).Built(context.Background(), link.BuildResult{ID: "build-4", Checked: &link.CheckRequest{
Owner: "novox", Repo: "a", Number: 7}, Check: &link.CheckOutcome{Verdict: "pass"}})
if len(said.events) != 1 {
t.Fatalf("a pull request's check was said %d time(s)", len(said.events))
}
}
// saidOnTheBus keeps the seat events said, and nothing else.
type saidOnTheBus struct {
link.Bus
events []string
}
func (s *saidOnTheBus) PublishSeatEvent(_ context.Context, seat, event string, _ []byte) error {
s.events = append(s.events, seat+" "+event)
return nil
}
// A ref the forge does not hold is refused before anything is asked, naming it.
func TestATryOfAnUnknownRefIsRefusedLoudly(t *testing.T) {
was := forgeCommit
defer func() { forgeCommit = was }()
forgeCommit = func(_ context.Context, owner, repo, ref string) (string, string, error) {
return "", "", errors.New("404 object does not exist [id: " + ref + "]")
}
err := tryCommand(context.Background(), []string{"novox/mesh-controller", "fix/no-such-branch"})
if err == nil {
t.Fatal("a try of a ref the forge does not hold was asked")
}
for _, want := range []string{"fix/no-such-branch", "novox/mesh-controller", "nothing was asked", "404"} {
if !strings.Contains(err.Error(), want) {
t.Errorf("the refusal does not say %q: %v", want, err)
}
}
for _, args := range [][]string{{"mesh-controller", "main"}, {"novox/mesh-controller"}, {"novox/a", "--upload-pack=x"},
{"novox/a", "main", "--tests", "x; rm -rf /"}, {"novox/a", "main", "--id", "build-1"}} {
if err := tryCommand(context.Background(), args); err == nil {
t.Errorf("try %v was taken", args)
}
}
}
// The answer of a try carries the lines that name what failed (novox/hq issue 460's shape) and how long it took.
func TestATrysAnswerCarriesTheFailingLines(t *testing.T) {
tried, _ := json.Marshal(link.Tried{Verdict: "fail", Summary: "its merge-check.sh failed: TestX", Took: "5m3s",
Head: "c0ffee", Trunk: "ca3f43c", Failed: []string{"--- FAIL: TestX (0.01s)", " x_test.go:12: got 1, want 2",
"FAIL\tgithub.com/novox/a/cmd\t3.1s"}})
lines := []link.BuildLine{{Seq: 1, Step: "check", Message: "novox/a#0 at c0ffee"},
{Seq: 2, Step: "output", Message: "ok github.com/novox/a/internal 1.2s"},
{Seq: 3, Step: link.StepTried, Message: string(tried)}, {Seq: 4, Step: "checked", Message: "pass: x"}}
a := tryAnswerOf("build-1", lines)
if a.State != "fail" || a.Took != "5m3s" || len(a.Failed) != 3 || a.Failed[0] != "--- FAIL: TestX (0.01s)" ||
a.Trunk != "ca3f43c" || a.ID != "build-1" {
t.Fatalf("the answer lost the verdict, the time or the failing lines: %+v", a)
}
// Not answered yet: said as running, with how far it is — never as a pass.
if a := tryAnswerOf("build-1", lines[:2]); a.State != "running" || a.Last == "" {
t.Fatalf("a try still running: %+v", a)
}
if a := tryAnswerOf("build-1", nil); a.State != "waiting" {
t.Fatalf("a try nothing was said of: %+v", a)
}
// One that could not run is an error with the build seat's word, never a pass.
if a := tryAnswerOf("build-1", append(lines[:1:1], link.BuildLine{Seq: 2, Step: "failed",
Message: "cannot clone"})); a.State != "error" || !strings.Contains(a.Said, "cannot clone") {
t.Fatalf("a try that could not run: %+v", a)
}
}
// One try of a commit at a time, and four in all (novox/hq issue 686): a second ask is answered the first's id,
// a fifth is refused; a try that ended, and a pull request's check, count for nothing.
func TestTriesAreBoundedInNumber(t *testing.T) {
ask := func(id, repo, ref, state string, try bool) link.QueuedAsk {
return link.QueuedAsk{ID: id, State: state, Request: link.BuildRequest{ID: id, Ref: ref,
Check: &link.CheckRequest{Owner: "novox", Repo: repo, Try: try}}}
}
q := link.Queue{Asks: []link.QueuedAsk{
ask("t1", "a", "c1", link.AskInFlight, true), ask("t2", "a", "c2", link.AskWaiting, true),
ask("t3", "b", "c1", link.AskDead, true), ask("p1", "a", "c9", link.AskWaiting, false)}}
if same, room := triesIn(q, "novox", "a", "c2"); same != "t2" || !room {
t.Fatalf("a second try of a commit: %q %v", same, room)
}
if same, room := triesIn(q, "novox", "b", "c1"); same != "" || !room {
t.Fatalf("a dead try, or another repository's, was taken for this one: %q %v", same, room)
}
if same, _ := triesIn(q, "novox", "a", "c9"); same != "" {
t.Fatalf("a pull request's check was taken for a try: %q", same)
}
q.Asks = append(q.Asks, ask("t4", "c", "c3", link.AskWaiting, true), ask("t5", "d", "c4", link.AskWaiting, true))
if _, room := triesIn(q, "novox", "e", "c5"); room {
t.Fatal("a fifth try was let in")
}
// An id is checked before it is a token of a bus subject.
for _, id := range []string{"x.>", "*", "a b"} {
if err := tryCommand(context.Background(), []string{"--id", id}); err == nil ||
!strings.Contains(err.Error(), "not a try's id") {
t.Errorf("try --id %q: %v", id, err)
}
}
}
// The verb runs its command, and is declared with the shell habit it replaces (novox/hq ADR 0245).
func TestTheTryVerbRunsItsCommand(t *testing.T) {
for _, c := range []struct {
args map[string]any
want []string
}{
{map[string]any{"repository": "novox/a", "ref": "fix/x"}, []string{"try", "novox/a", "fix/x"}},
{map[string]any{"repository": "novox/a", "ref": "fix/x", "base": "main", "tests": "^TestX$", "store-only": "true"},
[]string{"try", "novox/a", "fix/x", "--base", "main", "--tests", "^TestX$", "--store-only"}},
{map[string]any{"id": "build-1"}, []string{"try", "--id", "build-1"}},
} {
got, err := argvFor("try", c.args)
if err != nil || !reflect.DeepEqual(got, c.want) {
t.Errorf("try %v: %v %v, want %v", c.args, got, err, c.want)
}
}
if _, err := argvFor("try", map[string]any{"repository": "novox/a"}); err == nil {
t.Error("a try of no ref was taken")
}
for _, v := range catalogue.ControllerVerbs {
if v.Name == "try" {
if len(v.Replaces) == 0 {
t.Error("try names no shell habit it replaces")
}
return
}
}
t.Error("try is not a verb of the controller seat")
}
+1 -1
View File
@@ -79,7 +79,7 @@ func (f following) Upgraded(ctx context.Context, u link.Upgraded) error {
// `rebuild`, a `replay` registered — was sent here one machine after another without any judging; it // `rebuild`, a `replay` registered — was sent here one machine after another without any judging; it
// now waits for a gate like any other build, and the release plan walks it through the machines, one // now waits for a gate like any other build, and the release plan walks it through the machines, one
// at a time, each judged. // at a time, each judged.
fmt.Printf("%s moved to %s outside a plan; a release plan sends it to %s, one machine at a time, each judged "+ fmt.Printf("%s moved to %s outside a plan; a release plan sends it to %s, one machine first and then the rest at once, each judged "+
"(`upgrade backlog` lists what waits)\n", u.Module, shortCommit(u.Commit), readableList(on)) "(`upgrade backlog` lists what waits)\n", u.Module, shortCommit(u.Commit), readableList(on))
return nil return nil
} }
+6
View File
@@ -177,6 +177,12 @@ func walkReached(p inventory.Plan, running func(string) []string) map[string][]s
reached = appendNew(reached, r.Gate.Machines...) reached = appendNew(reached, r.Gate.Machines...)
carried(r.Gate) carried(r.Gate)
} }
for _, g := range r.Gates {
if g != nil && !g.Pending {
reached = appendNew(reached, g.Machines...)
carried(g)
}
}
for _, tier := range p.Tiers { for _, tier := range p.Tiers {
for _, m := range tier { for _, m := range tier {
for _, n := range running(m) { for _, n := range running(m) {
+2 -58
View File
@@ -40,39 +40,6 @@ var (
AskedBucket = BucketName(ControllerSeat, "asked") AskedBucket = BucketName(ControllerSeat, "asked")
) )
// The controller's record mirrors (novox/hq ADR 0289, design 54, design 57 package F3): a bucket per kind of
// record, one key per record, written from the store by the lease holder only when the record changed, with
// one event on the controller's seat per write. Read by whatever names them in `reads` as
// `mesh-controller.<mirror>` (ADR 0201 §1), written by nobody else. **No secret in either** (ADR 0201): an
// assignment's value names its module and node; a module's declared config is what its manifest, which is
// public, declares.
//
// AssignmentsBucket holds which module is assigned to which node, keyed `<node>.<module>`; and
// DeclaredConfigBucket each registered module's config declarations as its manifest holds them, keyed by
// the module: what mesh-config resolves a config value from (ADR 0348 clauses 2 and 4).
var (
AssignmentsBucket = BucketName(ControllerSeat, MirrorAssignments)
DeclaredConfigBucket = BucketName(ControllerSeat, MirrorDeclaredConfig)
)
// The mirrors' local names, as a reader names them in `reads` after `mesh-controller.`.
const (
MirrorAssignments = "assignments"
MirrorDeclaredConfig = "declared-config"
)
// ControllerMirrors are the local names of the controller's record mirrors, which a module may read.
func ControllerMirrors() []string { return []string{MirrorAssignments, MirrorDeclaredConfig} }
// The mirrors' caps (design 54 §5): one value per key, no age, 64 KiB a value, 64 MiB a bucket.
const (
MirrorValueBytes = 64 << 10
MirrorBytes = 64 << 20
// MirrorMarkerTTL is the longest a purge's marker is kept (design 54 §2): the bucket allows a
// per-message age for it, and the writer gives each marker a day.
MirrorMarkerTTL = 24 * time.Hour
)
// AskedKeptFor is how long an ask is kept after it was made: a month, as the router keeps its own. // AskedKeptFor is how long an ask is kept after it was made: a month, as the router keeps its own.
const AskedKeptFor = 30 * 24 * time.Hour const AskedKeptFor = 30 * 24 * time.Hour
@@ -99,14 +66,12 @@ const (
// IsControllerBucket says a bucket is the controller's own, not a module's state nothing declares. // IsControllerBucket says a bucket is the controller's own, not a module's state nothing declares.
func IsControllerBucket(bucket string) bool { func IsControllerBucket(bucket string) bool {
return bucket == CallsBucket || bucket == HandActsBucket || bucket == ConditionsBucket || return bucket == CallsBucket || bucket == HandActsBucket || bucket == ConditionsBucket ||
bucket == ConditionHistoryBucket || bucket == LeaseBucket || bucket == AskedBucket || bucket == ConditionHistoryBucket || bucket == LeaseBucket || bucket == AskedBucket
bucket == AssignmentsBucket || bucket == DeclaredConfigBucket
} }
// ControllerBuckets are the controller's own buckets, in the order they are asserted. // ControllerBuckets are the controller's own buckets, in the order they are asserted.
func ControllerBuckets() []string { func ControllerBuckets() []string {
return []string{LeaseBucket, CallsBucket, HandActsBucket, ConditionsBucket, ConditionHistoryBucket, AskedBucket, return []string{LeaseBucket, CallsBucket, HandActsBucket, ConditionsBucket, ConditionHistoryBucket, AskedBucket}
AssignmentsBucket, DeclaredConfigBucket}
} }
// ControllerBucketsAsserter is what raising the controller's buckets needs of a connection. // ControllerBucketsAsserter is what raising the controller's buckets needs of a connection.
@@ -203,27 +168,6 @@ func (j *JetStream) EnsureControllerBuckets() error {
}); err != nil { }); err != nil {
return fmt.Errorf("asserting bucket %s: %w", AskedBucket, err) return fmt.Errorf("asserting bucket %s: %w", AskedBucket, err)
} }
// **The record mirrors: no age** (design 54 §5). A key is removed when its record is, by a purge whose
// marker lives a day; a key that expired would be a record the bus forgot while the store still holds it.
for _, m := range []struct{ bucket, why string }{
{AssignmentsBucket, "which module is assigned to which node, keyed <node>.<module> (novox/hq ADR 0289, " +
"design 54): written by the controller alone from its store when an assignment changes, read through `reads`"},
{DeclaredConfigBucket, "each registered module's config declarations as its manifest holds them, keyed by " +
"the module (novox/hq ADR 0348, design 57 F3): written by the controller alone from its store when a " +
"registration changes them, read through `reads`"},
} {
if _, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{
Bucket: m.bucket,
Description: m.why,
History: 1,
MaxValueSize: MirrorValueBytes,
MaxBytes: MirrorBytes,
Storage: jetstream.FileStorage,
LimitMarkerTTL: MirrorMarkerTTL,
}); err != nil {
return fmt.Errorf("asserting bucket %s: %w", m.bucket, err)
}
}
return nil return nil
} }
+1 -83
View File
@@ -21,8 +21,7 @@ func TestTheControllerMayWriteEveryBucketItWrites(t *testing.T) {
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
want := []string{"$KV." + CallsBucket + ".>", "$KV." + HandActsBucket + ".>", want := []string{"$KV." + CallsBucket + ".>", "$KV." + HandActsBucket + ".>"}
"$KV." + AssignmentsBucket + ".>", "$KV." + DeclaredConfigBucket + ".>"}
for _, seat := range seatsTheControllerAsks { for _, seat := range seatsTheControllerAsks {
if hasCancelledSet(seat) { if hasCancelledSet(seat) {
want = append(want, "$KV."+CancelledSetName(seat)+".>") want = append(want, "$KV."+CancelledSetName(seat)+".>")
@@ -38,34 +37,6 @@ func TestTheControllerMayWriteEveryBucketItWrites(t *testing.T) {
} }
} }
// **A module reads a record mirror through `reads`, and never writes it** (ADR 0201 §1, design 54 §5,
// design 57 F3): granted the direct reads and watches of the bucket, and refused at composition a grant to
// write it, whose writer is the controller.
func TestAModuleReadsTheControllersMirrorsAndWritesNone(t *testing.T) {
reader := Principal{Kind: KindModule, Node: "one", Module: "mesh-config", PasswordHash: "x",
Reads: []string{"mesh-controller." + MirrorAssignments, "mesh-controller." + MirrorDeclaredConfig}}
p, err := PermissionsFor(reader)
if err != nil {
t.Fatal(err)
}
for _, bucket := range []string{AssignmentsBucket, DeclaredConfigBucket} {
for _, subject := range []string{"$JS.API.STREAM.INFO.KV_" + bucket, "$JS.API.DIRECT.GET.KV_" + bucket + ".>",
"$JS.API.CONSUMER.CREATE.KV_" + bucket + ".>"} {
if !slices.Contains(p.Publish, subject) {
t.Errorf("a module reading %s may not publish %s", bucket, subject)
}
}
for _, subject := range p.Publish {
if SubjectsOverlap(subject, "$KV."+bucket+".>") {
t.Errorf("a module reading %s may write it through %s", bucket, subject)
}
}
if err := CheckWriters(reader, []string{"$KV." + bucket + ".>"}); err == nil {
t.Errorf("a module granted a write of %s would compose", bucket)
}
}
}
// **A machine's node tools may say they are there, as that machine and no other** (novox/hq to-be 45 // **A machine's node tools may say they are there, as that machine and no other** (novox/hq to-be 45
// S11), and the controller may hear the bus's advisories and ask who answers — read-only, named. // S11), and the controller may hear the bus's advisories and ask who answers — read-only, named.
func TestTheWatchedSignalsMayBeSaidAndHeard(t *testing.T) { func TestTheWatchedSignalsMayBeSaidAndHeard(t *testing.T) {
@@ -125,56 +96,3 @@ func TestWhatTheControllerAskedIsBounded(t *testing.T) {
t.Errorf("history %d, age %s, bytes %d", status.History(), status.TTL(), info.Config.MaxBytes) t.Errorf("history %d, age %s, bytes %d", status.History(), status.TTL(), info.Config.MaxBytes)
} }
} }
// **A listed mirror is no undeclared bucket** (ADR 0356 §3, ADR 0201 §16): the controller's list of its mirrors
// stands for their declaration, so the report of state nothing declares never names one.
func TestAListedMirrorIsNotReportedAsUndeclared(t *testing.T) {
names := []string{"gone_state"}
for _, m := range ControllerMirrors() {
names = append(names, BucketName(ControllerSeat, m))
}
undeclared, err := RaiseBuckets(fakeBuckets{names: names}, nil)
if err != nil {
t.Fatal(err)
}
if !slices.Equal(undeclared, []string{"gone_state"}) {
t.Fatalf("the report of undeclared state names %v", undeclared)
}
}
// **The mirrors' list and the writers table agree** (ADR 0356 §3, to-be 45 §1): every listed mirror is a row of
// the table whose one writer is the controller, on its bucket alone; and every row that writes a bucket of the
// controller's named as a mirror is on the list.
func TestTheMirrorsListAndTheWritersTableAgree(t *testing.T) {
controller := Principal{Kind: KindController}
module := Principal{Kind: KindModule, Module: ControllerSeat, Node: "one"}
rowed := map[string]bool{}
for _, row := range WritersTable {
if !strings.HasSuffix(row.State, " mirror") {
continue
}
for _, subject := range row.Subjects {
bucket := strings.TrimSuffix(strings.TrimPrefix(subject, "$KV."), ".>")
rowed[bucket] = true
if row.Writes == nil || !row.Writes(controller, subject) || row.Writes(module, subject) {
t.Errorf("%q is not written by the controller alone", row.State)
}
}
}
listed := map[string]bool{}
for _, m := range ControllerMirrors() {
bucket := BucketName(ControllerSeat, m)
listed[bucket] = true
if !rowed[bucket] {
t.Errorf("the mirror %s is listed and no row of the writers table writes it", bucket)
}
if !IsControllerBucket(bucket) || !slices.Contains(ControllerBuckets(), bucket) {
t.Errorf("the mirror %s is listed and not asserted as the controller's own", bucket)
}
}
for bucket := range rowed {
if !listed[bucket] {
t.Errorf("the writers table writes the mirror %s, which the list does not name", bucket)
}
}
}
+2 -6
View File
@@ -172,9 +172,7 @@ func TestRaisingLogsReportsWhatNothingDeclares(t *testing.T) {
// **No path of the mesh deletes or purges a log or a bucket, or recreates one** (novox/hq ADR 0297 §2, // **No path of the mesh deletes or purges a log or a bucket, or recreates one** (novox/hq ADR 0297 §2,
// ADR 0201 §15–16): no code of this repository outside its tests calls the client's stream or bucket // ADR 0201 §15–16): no code of this repository outside its tests calls the client's stream or bucket
// delete, or purges a stream, but for the controller lease's own stream, which purges its own history // delete, or purges a stream, but for the controller lease's own stream, which purges its own history
// below a sequence (internal/lease), and the controller's record mirrors, which purge the key of a record // below a sequence (internal/lease). A new call is a decision, not a refactor.
// the store no longer holds (internal/recordmirror; novox/hq ADR 0289, design 54 §2). A new call is a decision,
// not a refactor.
func TestNoPathDeletesALogOrABucket(t *testing.T) { func TestNoPathDeletesALogOrABucket(t *testing.T) {
removers := map[string]bool{"DeleteStream": true, "DeleteKeyValue": true, "PurgeStream": true, removers := map[string]bool{"DeleteStream": true, "DeleteKeyValue": true, "PurgeStream": true,
"DeleteObjectStore": true} "DeleteObjectStore": true}
@@ -199,9 +197,7 @@ func TestNoPathDeletesALogOrABucket(t *testing.T) {
return true return true
} }
name := sel.Sel.Name name := sel.Sel.Name
decided := strings.Contains(filepath.ToSlash(path), "internal/lease/") || if removers[name] || (name == "Purge" && !strings.Contains(filepath.ToSlash(path), "internal/lease/")) {
strings.Contains(filepath.ToSlash(path), "internal/recordmirror/")
if removers[name] || (name == "Purge" && !decided) {
found = append(found, path+": "+name) found = append(found, path+": "+name)
} }
return true return true
-7
View File
@@ -34,8 +34,6 @@ func TestTheFactsTheGrantPermitsAreTheFactsTheMeshStates(t *testing.T) {
states = append(states, link.KeyChecked) states = append(states, link.KeyChecked)
// And a walk kept in a new state, for the delivery it walks (novox/hq ADR 0239). // And a walk kept in a new state, for the delivery it walks (novox/hq ADR 0239).
states = append(states, link.KeyPlanMoved) states = append(states, link.KeyPlanMoved)
// And a key of a record mirror written or removed (novox/hq ADR 0289, design 54 §3; design 57 F3).
states = append(states, link.KeyAssignmentChanged, link.KeyDeclaredConfigChanged)
for _, event := range states { for _, event := range states {
if !slices.Contains(broker.ControllerStates, event) { if !slices.Contains(broker.ControllerStates, event) {
t.Errorf("the mesh states %q and its account may not publish it", event) t.Errorf("the mesh states %q and its account may not publish it", event)
@@ -44,11 +42,6 @@ func TestTheFactsTheGrantPermitsAreTheFactsTheMeshStates(t *testing.T) {
if len(broker.ControllerStates) != len(states) { if len(broker.ControllerStates) != len(states) {
t.Errorf("the grant permits %v, which is more than the mesh states", broker.ControllerStates) t.Errorf("the grant permits %v, which is more than the mesh states", broker.ControllerStates)
} }
// The record mirrors a module may read are the ones the bus holds (design 54 §5, design 57 F3).
if !slices.Equal(catalogue.ControllerMirrors, broker.ControllerMirrors()) {
t.Errorf("the catalogue lets a module read the controller's mirrors %v and the bus holds %v",
catalogue.ControllerMirrors, broker.ControllerMirrors())
}
// **And the seat says it.** A seat carries the protocol of its role (novox/hq ADR 0129), so the // **And the seat says it.** A seat carries the protocol of its role (novox/hq ADR 0129), so the
// facts the control plane states are the seat's `emits` — which is what lets anything else declare // facts the control plane states are the seat's `emits` — which is what lets anything else declare
// that it consumes them, and what the subject-agreement check reads to know they have an owner. // that it consumes them, and what the subject-agreement check reads to know they have an owner.
+1 -4
View File
@@ -313,10 +313,7 @@ var ControllerStates = []string{"applied", "refused", "built-before",
// the pull request's status, and anybody else may read. // the pull request's status, and anybody else may read.
"checked", "checked",
// And a walk kept in a new state (novox/hq ADR 0239): what the delivery it walks reads its steps from. // And a walk kept in a new state (novox/hq ADR 0239): what the delivery it walks reads its steps from.
"plan-moved", "plan-moved"}
// And a key of a record mirror written or removed (novox/hq ADR 0289, design 54 §3): the assignments and
// each registered module's declared config (design 57 F3).
"assignment-changed", "declared-config-changed"}
// BusAdvisories are what the bus server says about the mesh's own account that the controller // BusAdvisories are what the bus server says about the mesh's own account that the controller
// reads (novox/hq to-be 45 §3, S9): a durable consumer that handed a message over as often as it // reads (novox/hq to-be 45 §3, S9): a durable consumer that handed a message over as often as it
+1 -1
View File
@@ -24,7 +24,7 @@ accounts {
jetstream: enabled jetstream: enabled
users = [ users = [
{ user: "controller", password: "$2a$11$cccccccccccccccccccccc", permissions: { { user: "controller", password: "$2a$11$cccccccccccccccccccccc", permissions: {
publish: { allow: ["$JS.ACK.CONTROL.controller.>", "$JS.ACK.DEAD_LETTER_NOTICES.controller.>", "$JS.ACK.EVENTS.controller.>", "$JS.API.>", "$KV.SEAT_MESH_BUILD_MACHINE_cancelled.>", "$KV.SEAT_NODE_BUILD_AGENT_cancelled.>", "$KV.mesh-controller_asked.>", "$KV.mesh-controller_assignments.>", "$KV.mesh-controller_calls.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_declared-config.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "mesh.node.>", "mesh.seat.mesh-build-machine.accept.>", "mesh.seat.mesh-build-machine.tool.>", "mesh.seat.mesh-controller.event.applied", "mesh.seat.mesh-controller.event.assignment-changed", "mesh.seat.mesh-controller.event.built-before", "mesh.seat.mesh-controller.event.checked", "mesh.seat.mesh-controller.event.condition-changed", "mesh.seat.mesh-controller.event.condition-cleared", "mesh.seat.mesh-controller.event.condition-raised", "mesh.seat.mesh-controller.event.declared-config-changed", "mesh.seat.mesh-controller.event.doctor-heartbeat", "mesh.seat.mesh-controller.event.healer-acted", "mesh.seat.mesh-controller.event.plan-moved", "mesh.seat.mesh-controller.event.refused", "mesh.seat.mesh-controller.event.rolled-back", "mesh.seat.mesh-controller.event.secret-replaced", "mesh.seat.mesh-controller.tool.plans", "mesh.seat.mesh-delivery.tool.close", "mesh.seat.mesh-delivery.tool.groups", "mesh.seat.mesh-delivery.tool.release", "mesh.seat.mesh-delivery.tool.stalled", "mesh.seat.mesh-delivery.tool.stop", "mesh.seat.mesh-delivery.tool.times", "mesh.seat.node-backup.tool.backed-up.*", "mesh.seat.node-backup.tool.now.*", "mesh.seat.node-build-agent.accept.>", "mesh.seat.node-build-agent.tool.>", "mesh.seat.node-intrusion-prevention.tool.banned.*", "mesh.seat.node-launcher.tool.secret.*", "mesh.seat.node-service-manager.tool.restart.*"] } publish: { allow: ["$JS.ACK.CONTROL.controller.>", "$JS.ACK.DEAD_LETTER_NOTICES.controller.>", "$JS.ACK.EVENTS.controller.>", "$JS.API.>", "$KV.SEAT_MESH_BUILD_MACHINE_cancelled.>", "$KV.SEAT_NODE_BUILD_AGENT_cancelled.>", "$KV.mesh-controller_asked.>", "$KV.mesh-controller_calls.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "mesh.node.>", "mesh.seat.mesh-build-machine.accept.>", "mesh.seat.mesh-build-machine.tool.>", "mesh.seat.mesh-controller.event.applied", "mesh.seat.mesh-controller.event.built-before", "mesh.seat.mesh-controller.event.checked", "mesh.seat.mesh-controller.event.condition-changed", "mesh.seat.mesh-controller.event.condition-cleared", "mesh.seat.mesh-controller.event.condition-raised", "mesh.seat.mesh-controller.event.doctor-heartbeat", "mesh.seat.mesh-controller.event.healer-acted", "mesh.seat.mesh-controller.event.plan-moved", "mesh.seat.mesh-controller.event.refused", "mesh.seat.mesh-controller.event.rolled-back", "mesh.seat.mesh-controller.event.secret-replaced", "mesh.seat.mesh-controller.tool.plans", "mesh.seat.mesh-delivery.tool.close", "mesh.seat.mesh-delivery.tool.groups", "mesh.seat.mesh-delivery.tool.release", "mesh.seat.mesh-delivery.tool.stalled", "mesh.seat.mesh-delivery.tool.stop", "mesh.seat.mesh-delivery.tool.times", "mesh.seat.node-backup.tool.backed-up.*", "mesh.seat.node-backup.tool.now.*", "mesh.seat.node-build-agent.accept.>", "mesh.seat.node-build-agent.tool.>", "mesh.seat.node-intrusion-prevention.tool.banned.*", "mesh.seat.node-launcher.tool.secret.*", "mesh.seat.node-service-manager.tool.restart.*"] }
subscribe: { allow: ["$JS.API.>", "$JS.EVENT.ADVISORY.CONSUMER.DELETED.>", "$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.>", "$SRV.INFO", "$SRV.INFO.mesh-controller", "$SRV.INFO.mesh-controller.>", "$SRV.PING", "$SRV.PING.mesh-controller", "$SRV.PING.mesh-controller.>", "$SRV.STATS", "$SRV.STATS.mesh-controller", "$SRV.STATS.mesh-controller.>", "_DELIVER.controller", "_DELIVER.controller.>", "_INBOX.controller.>", "mesh.control.>", "mesh.mod.*.event.provisioner.failing", "mesh.mod.*.event.provisioner.recovered", "mesh.mod.*.event.provisioner.retirement", "mesh.mod.gitea.event.pull.merged", "mesh.mod.gitea.event.pull.updated", "mesh.mod.mesh-catalog.event.catching-up", "mesh.mod.mesh-catalog.event.upgraded", "mesh.seat.mesh-build-machine.event.built", "mesh.seat.mesh-controller.tool.>", "mesh.seat.node-build-agent.event.built", "mesh.seat.operator-channel.event.decided.mesh-controller"] } subscribe: { allow: ["$JS.API.>", "$JS.EVENT.ADVISORY.CONSUMER.DELETED.>", "$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.>", "$SRV.INFO", "$SRV.INFO.mesh-controller", "$SRV.INFO.mesh-controller.>", "$SRV.PING", "$SRV.PING.mesh-controller", "$SRV.PING.mesh-controller.>", "$SRV.STATS", "$SRV.STATS.mesh-controller", "$SRV.STATS.mesh-controller.>", "_DELIVER.controller", "_DELIVER.controller.>", "_INBOX.controller.>", "mesh.control.>", "mesh.mod.*.event.provisioner.failing", "mesh.mod.*.event.provisioner.recovered", "mesh.mod.*.event.provisioner.retirement", "mesh.mod.gitea.event.pull.merged", "mesh.mod.gitea.event.pull.updated", "mesh.mod.mesh-catalog.event.catching-up", "mesh.mod.mesh-catalog.event.upgraded", "mesh.seat.mesh-build-machine.event.built", "mesh.seat.mesh-controller.tool.>", "mesh.seat.node-build-agent.event.built", "mesh.seat.operator-channel.event.decided.mesh-controller"] }
allow_responses: { max: 1, ttl: "1m" } allow_responses: { max: 1, ttl: "1m" }
} } } }
-9
View File
@@ -158,15 +158,6 @@ var WritersTable = []WriterRow{
{State: "a message a consumer gave up on", Writer: "controller", KeptIn: "the bus, the stream " + DeadLettersStream, {State: "a message a consumer gave up on", Writer: "controller", KeptIn: "the bus, the stream " + DeadLettersStream,
Others: "read, delivered again or dropped through the controller's dead-letters verb", Others: "read, delivered again or dropped through the controller's dead-letters verb",
Subjects: []string{deadLetterPrefix + ">", againPrefix + ">"}, Writes: isController}, Subjects: []string{deadLetterPrefix + ">", againPrefix + ">"}, Writes: isController},
// The record mirrors (novox/hq ADR 0289 §9, design 54 §2): rows of this table, so a grant that lets
// anything else write them is refused at composition. Built so far: the assignments and each registered
// module's declared config (design 57 F3).
{State: "the assignments mirror", Writer: "controller (lease holder), from its store, only on a change",
KeptIn: "key-value " + AssignmentsBucket, Others: "read through `reads` as mesh-controller." + MirrorAssignments,
Subjects: kvOf(AssignmentsBucket), Writes: isController},
{State: "the declared-config mirror", Writer: "controller (lease holder), from its store, only on a change",
KeptIn: "key-value " + DeclaredConfigBucket, Others: "read through `reads` as mesh-controller." + MirrorDeclaredConfig,
Subjects: kvOf(DeclaredConfigBucket), Writes: isController},
} }
// CheckWriters refuses a grant that lets a principal publish on a subject the writers table gives // CheckWriters refuses a grant that lets a principal publish on a subject the writers table gives
-3
View File
@@ -34,9 +34,6 @@ var designRows = []string{
"the operator-channel's open messages", "the operator-channel's open messages",
"the facts snapshot", "the facts snapshot",
"a message a consumer gave up on", "a message a consumer gave up on",
// The record mirrors, rows of to-be 45 §1's table by design 54 §2 (ADR 0289 §9).
"the assignments mirror",
"the declared-config mirror",
} }
func TestTheWritersTableIsTheDesigns(t *testing.T) { func TestTheWritersTableIsTheDesigns(t *testing.T) {
+23
View File
@@ -1076,6 +1076,16 @@ func one(ctx context.Context, run Runner, publish Publisher,
return catalogue.Built{}, fmt.Errorf("%s: building %s failed: %w", module, a.Name, err) return catalogue.Built{}, fmt.Errorf("%s: building %s failed: %w", module, a.Name, err)
} }
say("image", "built, publishing") say("image", "built, publishing")
// Its bases linked into its repository first, so the push needs none of their blobs from this
// machine's store, which may not hold them (novox/hq issue 694). Said and not fatal: a push
// from a store that holds them works without, and one that fails says which blob it lacked.
if lender, can := publish.(BaseLender); can {
if n, err := lender.LendBases(ctx, module+"/"+a.Name, buildArgValues(args)); err != nil {
say("image", "its bases could not all be linked into %s/%s before the push: %v", module, a.Name, err)
} else if n > 0 {
say("image", "%d blob(s) of its bases linked into %s/%s before the push", n, module, a.Name)
}
}
reference, err := publish.PublishImage(ctx, local, module+"/"+a.Name) reference, err := publish.PublishImage(ctx, local, module+"/"+a.Name)
if err != nil { if err != nil {
return catalogue.Built{}, err return catalogue.Built{}, err
@@ -1427,6 +1437,19 @@ func standingOn(ctx context.Context, manifest catalogue.Manifest, held map[strin
return args, resolved, nil return args, resolved, nil
} }
// buildArgValues is what each --build-arg of a build was handed: the bases, as this mesh holds them.
func buildArgValues(args []string) []string {
var values []string
for i := 0; i+1 < len(args); i += 2 {
if args[i] == "--build-arg" {
if _, value, has := strings.Cut(args[i+1], "="); has {
values = append(values, value)
}
}
}
return values
}
// compile runs a module's own code through its toolchain, and says where the result is. // compile runs a module's own code through its toolchain, and says where the result is.
// //
// **In the module's own directory, under the path the toolchain expects.** A module is compiled // **In the module's own directory, under the path the toolchain expects.** A module is compiled
+23
View File
@@ -103,6 +103,23 @@ type CheckSpec struct {
// for the walk to take. None for a group's composed check, whose members' own checks build. // for the walk to take. None for a group's composed check, whose members' own checks build.
Builds []CheckBuildSpec Builds []CheckBuildSpec
BuildWith *CheckBuilding BuildWith *CheckBuilding
// Tests and StoreOnly narrow what the repository's own script runs, for a branch tried before its pull
// request (novox/hq issue 686): handed to the script as EnvTests and EnvStoreOnly, which it may read.
Tests string
StoreOnly bool
}
// selection is what a check hands the repository's script to narrow what it runs (novox/hq issue 686): nothing
// for a pull request's check, which always runs whole.
func selection(spec CheckSpec) []string {
var env []string
if spec.Tests != "" {
env = append(env, EnvTests+"="+spec.Tests)
}
if spec.StoreOnly {
env = append(env, EnvStoreOnly+"=1")
}
return env
} }
// GroupHead is one other head of a delivery group's composed check. // GroupHead is one other head of a delivery group's composed check.
@@ -193,6 +210,11 @@ const (
// EnvGroup is a delivery group's other heads, as JSON, for the gate (novox/hq ADR 0239); empty for a // EnvGroup is a delivery group's other heads, as JSON, for the gate (novox/hq ADR 0239); empty for a
// pull request checked alone. // pull request checked alone.
EnvGroup = "MESH_CHECK_GROUP" EnvGroup = "MESH_CHECK_GROUP"
// EnvTests is a `go test -run` pattern and EnvStoreOnly is 1 when a branch tried before its pull request
// asks for only those tests, or only the store tests (novox/hq issue 686). Never set for a pull request's
// check. A script that reads neither runs whole.
EnvTests = "MESH_CHECK_TESTS"
EnvStoreOnly = "MESH_CHECK_STORE_ONLY"
) )
// CheckTimeout bounds one check; a check that runs past it is an error, not a pass. // CheckTimeout bounds one check; a check that runs past it is an error, not a pass.
@@ -515,6 +537,7 @@ func Check(ctx context.Context, run Runner, spec CheckSpec, workspace, registry
EnvModules + "=" + strings.Join(append(append([]string{}, spec.Modules...), spec.New...), ","), EnvModules + "=" + strings.Join(append(append([]string{}, spec.Modules...), spec.New...), ","),
EnvGroup + "=" + groupFile, EnvGroup + "=" + groupFile,
"GOCACHE=" + filepath.Join(workspace, "go-cache"), "GOMODCACHE=" + filepath.Join(workspace, "go-modules")} "GOCACHE=" + filepath.Join(workspace, "go-cache"), "GOMODCACHE=" + filepath.Join(workspace, "go-modules")}
env = append(env, selection(spec)...)
v := CheckVerdict{} v := CheckVerdict{}
modules := append(append([]string{}, spec.Modules...), prefixed("new:", spec.New)...) modules := append(append([]string{}, spec.Modules...), prefixed("new:", spec.New)...)
+12
View File
@@ -590,3 +590,15 @@ func TestARepositoryInTwoLanguagesIsCheckedInBoth(t *testing.T) {
t.Errorf("a part in a toolchain the mesh does not hold answered %+v — never a pass", layer) t.Errorf("a part in a toolchain the mesh does not hold answered %+v — never a pass", layer)
} }
} }
// A branch tried before its pull request may narrow what the repository's script runs (novox/hq issue 686); a
// pull request's check is handed nothing, and runs whole.
func TestOnlyATrysSelectionReachesTheScript(t *testing.T) {
if env := selection(CheckSpec{Number: 7}); len(env) != 0 {
t.Fatalf("a pull request's check was handed a selection: %v", env)
}
env := selection(CheckSpec{Tests: "^TestX$", StoreOnly: true})
if len(env) != 2 || env[0] != "MESH_CHECK_TESTS=^TestX$" || env[1] != "MESH_CHECK_STORE_ONLY=1" {
t.Fatalf("a try's selection: %v", env)
}
}
+178
View File
@@ -0,0 +1,178 @@
package builder
import (
"context"
"encoding/json"
"fmt"
"io"
"net/http"
"runtime"
"strings"
)
// **A built image's bases are linked into its repository before it is pushed** (novox/hq issue 694).
//
// A push reads every blob of the image out of this machine's store, unless the repository it goes to
// already holds that blob: the runtime asks the registry first, blob by blob, and reads locally only
// what the registry lacks. A runtime keeping its images in containerd's store — the default for a
// fresh install from docker 29 on — does not always hold the compressed layers of the base an image
// was built FROM: a layer whose unpacked form it already had is never fetched. The push of such an
// image then fails on a layer this machine never had ("not all of them are available locally"), while
// the very bytes sit in this registry, under the repository the base was mirrored into.
//
// So the base's blobs are mounted into the image's repository first. A mount moves no bytes, and once
// the repository holds a blob the push asks this machine for nothing of it — whichever store the
// runtime keeps. A runtime that holds every layer (the classic store) is not changed by this: its push
// finds the same blobs already there.
// BaseLender links what a build's bases hold into the repository its image is about to be pushed to.
type BaseLender interface {
// LendBases mounts every blob of each base this registry holds — for the platform this machine
// builds — into `repository`, and answers how many it mounted.
LendBases(ctx context.Context, repository string, bases []string) (int, error)
}
// platformed is an index's entry: a manifest and the platform it is for.
type platformed struct {
descriptor
Platform *struct {
OS string `json:"os"`
Architecture string `json:"architecture"`
} `json:"platform"`
}
// LendBases mounts each base's blobs into the repository. A base this registry does not hold under its
// repository — named by tag, or held somewhere else — lends nothing and is no error: the push reads it
// from this machine's store as before.
func (r Registry) LendBases(ctx context.Context, repository string, bases []string) (int, error) {
mounted := 0
for _, base := range bases {
name, digest, pinned := strings.Cut(base, "@")
if !pinned || !strings.HasPrefix(digest, "sha256:") {
continue
}
where, err := parseReference(name + "@" + digest)
if err != nil {
continue
}
from := where.repository
if from == repository {
continue
}
blobs, held, err := r.blobsOf(ctx, from, digest)
if err != nil {
return mounted, fmt.Errorf("reading %s: %w", base, err)
}
if !held {
continue
}
for _, blob := range blobs {
did, err := r.mount(ctx, repository, from, blob)
if err != nil {
return mounted, fmt.Errorf("linking %s of %s into %s: %w", blob, base, repository, err)
}
if did {
mounted++
}
}
}
return mounted, nil
}
// blobsOf is the config and layers of the manifest under repository@digest — for an index, of each
// manifest it names for this machine's platform. False when the registry holds no such manifest there.
func (r Registry) blobsOf(ctx context.Context, repository, digest string) ([]string, bool, error) {
body, held, err := r.manifest(ctx, repository, digest)
if err != nil || !held {
return nil, held, err
}
var document struct {
Manifests []platformed `json:"manifests"`
Config *descriptor `json:"config"`
Layers []descriptor `json:"layers"`
}
if err := json.Unmarshal(body, &document); err != nil {
return nil, false, fmt.Errorf("%s@%s is not a manifest: %w", repository, digest, err)
}
var blobs []string
if document.Config != nil {
blobs = append(blobs, document.Config.Digest)
}
for _, l := range document.Layers {
blobs = append(blobs, l.Digest)
}
for _, m := range document.Manifests {
if m.Platform == nil || m.Platform.OS != runtime.GOOS || m.Platform.Architecture != runtime.GOARCH {
continue
}
more, held, err := r.blobsOf(ctx, repository, m.Digest)
if err != nil {
return nil, false, err
}
if held {
blobs = append(blobs, more...)
}
}
return blobs, true, nil
}
// manifest reads one manifest document from this registry; false when it is not there.
func (r Registry) manifest(ctx context.Context, repository, digest string) ([]byte, bool, error) {
request, err := http.NewRequestWithContext(ctx, http.MethodGet,
"http://"+r.Address+"/v2/"+repository+"/manifests/"+digest, nil)
if err != nil {
return nil, false, err
}
request.Header.Set("Accept", manifestAccept)
response, err := r.client().Do(request)
if err != nil {
return nil, false, fmt.Errorf("cannot reach the registry at %s: %w", r.Address, err)
}
defer response.Body.Close()
if response.StatusCode == http.StatusNotFound {
return nil, false, nil
}
if response.StatusCode != http.StatusOK {
return nil, false, fmt.Errorf("%s/%s@%s: %s", r.Address, repository, digest, response.Status)
}
body, err := io.ReadAll(io.LimitReader(response.Body, 4<<20))
return body, err == nil, err
}
// mount links one blob from one repository into another; false when the repository already held it.
func (r Registry) mount(ctx context.Context, repository, from, digest string) (bool, error) {
base := "http://" + r.Address + "/v2/" + repository
if there, err := r.has(ctx, base+"/blobs/"+digest); err != nil {
return false, err
} else if there {
return false, nil
}
start, err := http.NewRequestWithContext(ctx, http.MethodPost,
base+"/blobs/uploads/?mount="+digest+"&from="+from, nil)
if err != nil {
return false, err
}
begun, err := r.client().Do(start)
if err != nil {
return false, fmt.Errorf("cannot reach the registry at %s: %w", r.Address, err)
}
begun.Body.Close()
if begun.StatusCode == http.StatusCreated {
return true, nil
}
if begun.StatusCode == http.StatusAccepted {
// Not mounted, and an upload begun in its place: let go of it, so nothing is left half-open.
if location := begun.Header.Get("Location"); location != "" {
if strings.HasPrefix(location, "/") {
location = "http://" + r.Address + location
}
if cancel, err := http.NewRequestWithContext(ctx, http.MethodDelete, location, nil); err == nil {
if done, err := r.client().Do(cancel); err == nil {
done.Body.Close()
}
}
}
return false, fmt.Errorf("%s did not mount it from %s", r.Address, from)
}
return false, fmt.Errorf("%s answered %s when asked to mount it from %s", r.Address, begun.Status, from)
}
+140
View File
@@ -0,0 +1,140 @@
package builder
import (
"context"
"encoding/json"
"net/http/httptest"
"runtime"
"strings"
"testing"
)
// A built image's bases are linked into its repository before it is pushed (novox/hq issue 694): a
// runtime keeping containerd's image store may not hold a base layer it never had to fetch, and its push
// asks this machine only for what the repository lacks.
// aBaseForThisMachine puts an index into the registry under `repository`: one manifest for this
// machine's platform, one for another, and answers the index's digest and each manifest's blobs.
func aBaseForThisMachine(t *testing.T, dst *aRegistryOfRepositories, repository string) (string, []string, []string) {
t.Helper()
put := func(arch string) (string, []string) {
config := []byte(`{"architecture":"` + arch + `"}`)
layer := []byte(arch + " layer bytes")
for _, b := range [][]byte{config, layer} {
dst.blobs[digestOf(b)] = b
dst.link(repository, digestOf(b))
}
m, _ := json.Marshal(map[string]any{
"schemaVersion": 2, "mediaType": mediaManifestOCI,
"config": map[string]any{"mediaType": "application/vnd.oci.image.config.v1+json", "digest": digestOf(config), "size": len(config)},
"layers": []map[string]any{{"mediaType": "application/vnd.oci.image.layer.v1.tar+gzip", "digest": digestOf(layer), "size": len(layer)}},
})
dst.manifests[repository+"@"+digestOf(m)] = m
return digestOf(m), []string{digestOf(config), digestOf(layer)}
}
other := "s390x"
if runtime.GOARCH == other {
other = "amd64"
}
mine, mineBlobs := put(runtime.GOARCH)
theirs, theirBlobs := put(other)
index, _ := json.Marshal(map[string]any{
"schemaVersion": 2, "mediaType": mediaIndexOCI,
"manifests": []map[string]any{
{"mediaType": mediaManifestOCI, "digest": mine, "size": 1, "platform": map[string]string{"os": runtime.GOOS, "architecture": runtime.GOARCH}},
{"mediaType": mediaManifestOCI, "digest": theirs, "size": 1, "platform": map[string]string{"os": "linux", "architecture": other}},
},
})
dst.manifests[repository+"@"+digestOf(index)] = index
return digestOf(index), mineBlobs, theirBlobs
}
func TestABuildsBaseIsLinkedIntoItsRepositoryBeforeThePush(t *testing.T) {
dst := newRegistryOfRepositories()
server := httptest.NewServer(dst.handler())
defer server.Close()
address := strings.TrimPrefix(server.URL, "http://")
r := Registry{Address: address}
const from = "upstream/docker.io/library/golang"
index, mine, theirs := aBaseForThisMachine(t, dst, from)
bases := []string{
address + "/" + from + "@" + index,
// Not bases this registry holds: a package version, a tag, an image held nowhere here.
"@novox/mesh-sdk@1.2.3",
"golang:1-alpine",
"elsewhere.invalid/thing@sha256:" + strings.Repeat("e", 64),
}
n, err := r.LendBases(context.Background(), "mesh-tools-go/build", bases)
if err != nil {
t.Fatal(err)
}
// Every blob of this machine's platform is in the image's repository, so a push asks this machine's
// store for none of them; another platform's are not, since nothing built here stands on them.
for _, blob := range mine {
if !dst.links["mesh-tools-go/build"][blob] {
t.Fatalf("the base's blob %s is not in the image's repository", blob)
}
}
for _, blob := range theirs {
if dst.links["mesh-tools-go/build"][blob] {
t.Fatalf("another platform's blob %s was linked", blob)
}
}
if n != len(mine) || dst.mounts != len(mine) || dst.uploads != 0 {
t.Fatalf("%d linked, %d mounts, %d uploads; want %d mounts and no bytes moved", n, dst.mounts, dst.uploads, len(mine))
}
// Asked again, nothing is mounted twice.
if n, err := r.LendBases(context.Background(), "mesh-tools-go/build", bases); err != nil || n != 0 {
t.Fatalf("a second lending mounted %d: %v", n, err)
}
}
// A registry that will not mount is said, not hidden: the push is still tried, from the store.
func TestABaseTheRegistryWillNotMountIsAnError(t *testing.T) {
dst := newRegistryOfRepositories()
dst.noMount = true
server := httptest.NewServer(dst.handler())
defer server.Close()
address := strings.TrimPrefix(server.URL, "http://")
index, _, _ := aBaseForThisMachine(t, dst, "upstream/thing")
_, err := Registry{Address: address}.LendBases(context.Background(), "hello/server",
[]string{address + "/upstream/thing@" + index})
if err == nil || !strings.Contains(err.Error(), "did not mount") {
t.Fatalf("a refused mount passed silently: %v", err)
}
}
// lending records what a build lent, and when, against the push.
type lending struct {
*mirroring
order []string
}
func (l *lending) LendBases(_ context.Context, repository string, bases []string) (int, error) {
l.order = append(l.order, "lend "+repository+" "+strings.Join(bases, " "))
return len(bases), nil
}
func (l *lending) PublishImage(ctx context.Context, localTag, repository string) (string, error) {
l.order = append(l.order, "push "+repository)
return l.mirroring.PublishImage(ctx, localTag, repository)
}
func TestAnImageIsPushedOnlyAfterItsBasesAreLent(t *testing.T) {
r, workspace := aRepository(t, anImage, map[string]string{"modules/bus/Dockerfile": "ARG BASE\nFROM ${BASE}"})
r.contents["modules/bus/module.json"] = anImage
l := &lending{mirroring: &mirroring{recorded: r, at: "127.0.0.1:5000"}}
if _, err := Build(context.Background(), r.run, l, "https://forge.invalid/catalogue.git", "modules/bus", "", workspace,
nil, Npmrc{}, GitCredential{}, nil); err != nil {
t.Fatal(err)
}
want := []string{
"lend bus/server 127.0.0.1:5000/upstream/docker.io/vendor/server@sha256:" + strings.Repeat("1", 64),
"push bus/server",
}
if strings.Join(l.order, "\n") != strings.Join(want, "\n") {
t.Fatalf("the build did:\n%s\nwant:\n%s", strings.Join(l.order, "\n"), strings.Join(want, "\n"))
}
}
+7
View File
@@ -257,6 +257,13 @@ func (s scratchPublisher) PublishArchive(ctx context.Context, repository string,
return s.to.PublishArchive(ctx, ScratchPrefix+repository, body, digest) return s.to.PublishArchive(ctx, ScratchPrefix+repository, body, digest)
} }
func (s scratchPublisher) LendBases(ctx context.Context, repository string, bases []string) (int, error) {
if l, can := s.to.(BaseLender); can {
return l.LendBases(ctx, ScratchPrefix+repository, bases)
}
return 0, nil
}
func (s scratchPublisher) MirrorImage(ctx context.Context, from, repository string) (string, error) { func (s scratchPublisher) MirrorImage(ctx context.Context, from, repository string) (string, error) {
m, can := s.to.(Mirrorer) m, can := s.to.(Mirrorer)
if !can { if !can {
+1 -3
View File
@@ -107,9 +107,7 @@ var defaultSeats = append([]Seat{
// A pull request's merge check, judged (novox/hq to-be 45 §9). // A pull request's merge check, judged (novox/hq to-be 45 §9).
"checked", "checked",
// A walk kept in a new state, for the delivery it walks (novox/hq ADR 0239). // A walk kept in a new state, for the delivery it walks (novox/hq ADR 0239).
"plan-moved", "plan-moved"},
// A key of a record mirror written or removed (novox/hq ADR 0289, design 54 §3; design 57 F3).
"assignment-changed", "declared-config-changed"},
Serves: ControllerVerbs}, Serves: ControllerVerbs},
// The store's first verbs (novox/hq ADR 0159): the smallest set that makes the store askable, // The store's first verbs (novox/hq ADR 0159): the smallest set that makes the store askable,
// served by whichever module holds the seat with tools of these names. // served by whichever module holds the seat with tools of these names.
-13
View File
@@ -124,14 +124,6 @@ func StateProblems(m Manifest) []string {
return problems return problems
} }
// ControllerMirrors are the controller's record mirrors a module may read, each `mesh-controller.<name>`
// (novox/hq ADR 0289, design 54 §5, design 57 F3). **The controller keeps them, not its manifest's `state`**:
// they are asserted with the controller's own buckets, with options of their own, and written by the
// controller's account alone (the writers table); declared as a module's state, they would be asserted a
// second way and granted to the module's own principal as well. Named here so a read of one is a read of
// kept state. The bus's list is broker.ControllerMirrors; a test keeps the two agreeing.
var ControllerMirrors = []string{"assignments", "declared-config"}
// StateReadsNothingDeclares is every read across a catalogue whose owner is present and declares no // StateReadsNothingDeclares is every read across a catalogue whose owner is present and declares no
// such state. An absent owner says nothing — a module may be installed long before the one whose // such state. An absent owner says nothing — a module may be installed long before the one whose
// state it reads, as a consumer may before its emitter (design 32 §1). // state it reads, as a consumer may before its emitter (design 32 §1).
@@ -144,11 +136,6 @@ func StateReadsNothingDeclares(manifests []Manifest) []string {
} }
declared[m.Module] = own declared[m.Module] = own
} }
if own, present := declared[ControllerSeatName]; present {
for _, name := range ControllerMirrors {
own[name] = true
}
}
var problems []string var problems []string
for _, m := range manifests { for _, m := range manifests {
for _, r := range m.Reads { for _, r := range m.Reads {
-15
View File
@@ -67,21 +67,6 @@ func TestAReadNamesStateItsOwnerKeeps(t *testing.T) {
} }
} }
// **The controller's record mirrors are kept state** (design 54 §5, design 57 F3): a module may read
// `mesh-controller.assignments` and `mesh-controller.declared-config` although the controller's manifest
// declares no `state`; a read of anything else of the controller's is still named.
func TestAReadOfTheControllersMirrorsIsAReadOfKeptState(t *testing.T) {
controller := Manifest{Module: ControllerSeatName}
reader := Manifest{Module: "mesh-config", Reads: []string{"mesh-controller.assignments", "mesh-controller.declared-config"}}
if p := StateReadsNothingDeclares([]Manifest{controller, reader}); len(p) != 0 {
t.Fatalf("a read of the controller's mirrors was refused: %v", p)
}
other := Manifest{Module: "other", Reads: []string{"mesh-controller.calls"}}
if p := StateReadsNothingDeclares([]Manifest{controller, other}); len(p) != 1 {
t.Fatalf("a read of the controller's own calls was not named: %v", p)
}
}
// **Across the whole catalogue**: every state name is local, and every read whose owner is present // **Across the whole catalogue**: every state name is local, and every read whose owner is present
// names state that owner keeps. // names state that owner keeps.
func TestEveryManifestsStateIsLocalAndEveryReadIsKept(t *testing.T) { func TestEveryManifestsStateIsLocalAndEveryReadIsKept(t *testing.T) {
+22
View File
@@ -362,6 +362,28 @@ var ControllerVerbs = []Verb{
"register": "\"true\" to register what it builds", "register": "\"true\" to register what it builds",
"older": "\"true\", with register: even though a newer build of the module is registered", "older": "\"true\", with register: even though a newer build of the module is registered",
}, []string{"id"}, "register", "older")}, }, []string{"id"}, "register", "older")},
// novox/hq issue 686: a builder ran a branch's store tests for the first time in its pull request's merge check.
{Name: "try", Description: "Try a pushed branch or commit before its pull request (novox/hq issue 686): the build " +
"seat merges it into the repository's trunk and runs the repository's own merge-check.sh on the merge, in the " +
"toolchain containers and against the throwaway store and bus a pull request's merge check gets. Answers at " +
"once with an id; asked again with that id it waits up to 20 s and answers pass, fail or error, how long it " +
"took and the lines naming what failed — or that it is still waiting or running, to ask again. Sets no " +
"status on the forge, registers and builds nothing, runs no merge gate, counts as a check in the build queue " +
"(after deliveries and after every pull request's check, however long it waits) and is bounded as one. One " +
"try of a commit at a time — a second ask answers the first's id — and at most four waiting or running in " +
"all: a fifth is refused. Only for a repository of the mesh; the ref must be on the forge.",
Input: schema(map[string]string{
"repository": "the repository on the forge, owner/name; with ref",
"ref": "the branch or commit to try, as pushed to the forge",
"base": "the branch it is merged into (the repository's default branch when none is named). The " +
"merge-check.sh that runs is that branch's, never the ref's own: the same trust as a pull request's base",
"tests": "only the tests this `go test -run` pattern names, where the repository's merge-check.sh reads " +
"MESH_CHECK_TESTS; a script that does not runs whole",
"store-only": "\"true\": only the repository's store tests, where its merge-check.sh reads " +
"MESH_CHECK_STORE_ONLY",
"id": "instead of a repository: the id a try answered, to read what came of it",
}, nil, "store-only"),
Replaces: []string{"sh merge-check.sh", "docker run postgres"}},
// novox/hq ADR 0350 §5: asked by the delivery's owner when the forge says a pull request closed unmerged. // novox/hq ADR 0350 §5: asked by the delivery's owner when the forge says a pull request closed unmerged.
{Name: "forget-checks", Description: "Let go of every build a pull request's merge checks made for its merge in " + {Name: "forget-checks", Description: "Let go of every build a pull request's merge checks made for its merge in " +
"the artifact store's scratch namespace (novox/hq ADR 0350 §5): asked when it closed unmerged; a merged one's go " + "the artifact store's scratch namespace (novox/hq ADR 0350 §5): asked when it closed unmerged; a merged one's go " +
-122
View File
@@ -1,122 +0,0 @@
package inventory
import (
"bytes"
"context"
"encoding/json"
"fmt"
"time"
)
// The values of the controller's record mirrors, built from the store (novox/hq ADR 0289, design 54 §1,
// design 57 package F3).
//
// **Each value is built from a named list of fields, and nothing else** (design 54 §1): no epoch, no time
// of writing and no time of the last look, so any lease holder builds the same bytes from the same record
// and a pass over unchanged records writes nothing (ADR 0289 §1, §4). The encoding is fixed: the fields in
// their struct's order, a declaration's maps sorted.
// AssignmentMirrored is one assignment as the assignments mirror holds it: the module, the node, and since
// when it is assigned there. Design 54 §1 also lists whether it is pending, the builds sent and running and
// the seats and provisions it holds there; those join with issue 419, which builds the rest of design 54.
type AssignmentMirrored struct {
Node string `json:"node"`
Module string `json:"module"`
Assigned time.Time `json:"assigned"`
}
// DeclaredConfigMirrored is one registered module's config declarations as the declared-config mirror holds
// them: the module, and each key its manifest's `settings` declares with that declaration's listed fields.
// An empty object when the module declares none: registered, declaring nothing.
type DeclaredConfigMirrored struct {
Module string `json:"module"`
Settings map[string]DeclaredMirrored `json:"settings"`
}
// DeclaredMirrored is one key's declaration as the mirror holds it: **its listed fields and no other**
// (ADR 0289 §12). Today a declaration is kind, default and why (catalogue.SettingDeclaration); a field the
// manifest gains reaches the bus only once it is added here, deliberately, and a test fails until it is
// listed or refused (ADR 0348 clause 4's fields arrive with package K1). A default is never a secret: the
// catalogue refuses a default for a key whose name says secret.
type DeclaredMirrored struct {
Kind string `json:"kind"`
Default json.RawMessage `json:"default"`
Why string `json:"why"`
}
// DeclaredMirroredFields are the fields of a declaration the mirror carries, by their manifest names.
var DeclaredMirroredFields = []string{"kind", "default", "why"}
// AssignmentsMirror is every assignment's mirrored value, keyed `<node>.<module>`.
func (i *Inventory) AssignmentsMirror(ctx context.Context) (map[string][]byte, error) {
rows, err := i.store.Pool().Query(ctx,
`select n.name, a.module, a.assigned from assignment a join node n on n.id = a.node`)
if err != nil {
return nil, err
}
defer rows.Close()
out := map[string][]byte{}
for rows.Next() {
var a AssignmentMirrored
if err := rows.Scan(&a.Node, &a.Module, &a.Assigned); err != nil {
return nil, err
}
a.Assigned = a.Assigned.UTC()
value, err := json.Marshal(a)
if err != nil {
return nil, err
}
out[a.Node+"."+a.Module] = value
}
return out, rows.Err()
}
// DeclaredConfigMirror is every registered module's mirrored config declarations, keyed by the module.
func (i *Inventory) DeclaredConfigMirror(ctx context.Context) (map[string][]byte, error) {
rows, err := i.store.Pool().Query(ctx,
`select name, coalesce(manifest->'settings', '{}'::jsonb) from module`)
if err != nil {
return nil, err
}
defer rows.Close()
out := map[string][]byte{}
for rows.Next() {
var (
d DeclaredConfigMirrored
raw []byte
)
if err := rows.Scan(&d.Module, &raw); err != nil {
return nil, err
}
var declared map[string]DeclaredMirrored
if err := json.Unmarshal(raw, &declared); err != nil {
return nil, fmt.Errorf("the declarations of %s cannot be read: %w", d.Module, err)
}
d.Settings = map[string]DeclaredMirrored{}
for key, decl := range declared {
if len(decl.Default) > 0 {
if decl.Default, err = canonical(decl.Default); err != nil {
return nil, err
}
}
d.Settings[key] = decl
}
value, err := json.Marshal(d)
if err != nil {
return nil, err
}
out[d.Module] = value
}
return out, rows.Err()
}
// canonical is a JSON document encoded the one way: objects' keys sorted, numbers as written.
func canonical(raw []byte) ([]byte, error) {
dec := json.NewDecoder(bytes.NewReader(raw))
dec.UseNumber()
var v any
if err := dec.Decode(&v); err != nil {
return nil, err
}
return json.Marshal(v)
}
+70 -2
View File
@@ -5,6 +5,7 @@ import (
"encoding/json" "encoding/json"
"errors" "errors"
"fmt" "fmt"
"slices"
"strings" "strings"
"time" "time"
@@ -391,6 +392,65 @@ type PlanGate struct {
// reports its module's container left running because it said busy, the judging waits and has not // reports its module's container left running because it said busy, the judging waits and has not
// started; once it reports the container replaced, Since is moved to then. Nil when nothing waited. // started; once it reports the container replaced, Since is moved to then. Nil when nothing waited.
Busy *GateBusy `json:"busy,omitempty"` Busy *GateBusy `json:"busy,omitempty"`
// Pending says the send this gate judges is kept and not yet made (novox/hq ADR 0353 §8): a controller that
// takes over reads the machine's last send before making it, and never sends it twice.
Pending bool `json:"pending,omitempty"`
// Silent says the machine reported nothing on its send within ApplySilentAfter (novox/hq ADR 0353 §4): left,
// neither failed nor put back.
Silent bool `json:"silent,omitempty"`
// TookOver names the gate whose readings this one took over (novox/hq ADR 0353 §7): another walk's judging of
// the same build on the same machine, which this gate goes on from instead of starting from zero.
TookOver string `json:"took_over,omitempty"`
}
// TakeReadings makes this gate go on from another's readings of the same build on the same machine (novox/hq ADR
// 0353 §7): when the judging began, when each machine applied, the passes counted and when, and what each found.
// A module's own count is taken only for the modules this gate judges.
func (g *PlanGate) TakeReadings(from *PlanGate, modules []string, id string) {
if from == nil {
return
}
if from.Since != nil {
since := *from.Since
g.Since = &since
}
for n, s := range from.Sent {
if slices.Contains(g.Machines, n) {
if g.Sent == nil {
g.Sent = map[string]SentDeclaration{}
}
g.Sent[n] = s
}
}
for n, at := range from.Applied {
if slices.Contains(g.Machines, n) {
if g.Applied == nil {
g.Applied = map[string]time.Time{}
}
g.Applied[n] = at
}
}
g.Passes, g.Faulted, g.HealthySeen, g.Last = from.Passes, from.Faulted, from.HealthySeen, from.Last
if from.LastPass != nil {
at := *from.LastPass
g.LastPass = &at
}
for _, m := range modules {
if n, counted := from.Healthy[m]; counted {
if g.Healthy == nil {
g.Healthy = map[string]int{}
}
g.Healthy[m] = n
}
if at, counted := from.HealthyAt[m]; counted {
if g.HealthyAt == nil {
g.HealthyAt = map[string]time.Time{}
}
g.HealthyAt[m] = at
}
}
g.Readings = append([]GateReading(nil), from.Readings...)
g.TookOver = id
} }
// GateBusy is a first-node gate's wait for a busy container (novox/hq ADR 0335 §4). // GateBusy is a first-node gate's wait for a busy container (novox/hq ADR 0335 §4).
@@ -477,9 +537,17 @@ type CarriedMove struct {
type PlanRelease struct { type PlanRelease struct {
Order []string `json:"order"` Order []string `json:"order"`
Next int `json:"next"` Next int `json:"next"`
// Gate is the machine being judged; nil between machines. // Gate is the first machine being judged; nil between machines.
Gate *PlanGate `json:"gate,omitempty"` Gate *PlanGate `json:"gate,omitempty"`
Done []string `json:"done,omitempty"` // Rest says the first machine passed (novox/hq ADR 0353 §1): every other machine is sent at once, each judged by
// its own gate in Gates, by machine — a build no gate passed going to one machine of a step only (§2).
Rest bool `json:"rest,omitempty"`
Gates map[string]*PlanGate `json:"gates,omitempty"`
// Failed are the machines of the rest whose gate failed, and Silent those that said nothing on their send
// within ApplySilentAfter (ADR 0353 §3 and §4).
Failed []string `json:"failed,omitempty"`
Silent []string `json:"silent,omitempty"`
Done []string `json:"done,omitempty"`
// Skipped are the machines not heard from when their turn came, left as they were. // Skipped are the machines not heard from when their turn came, left as they were.
Skipped []string `json:"skipped,omitempty"` Skipped []string `json:"skipped,omitempty"`
// By is the person who released it, empty when the mesh did. // By is the person who released it, empty when the mesh did.
-41
View File
@@ -1,41 +0,0 @@
package inventory
import (
"reflect"
"slices"
"strings"
"testing"
"github.com/novox/mesh-controller/internal/catalogue"
)
// jsonFields is the names a struct's fields are encoded under.
func jsonFields(t reflect.Type) []string {
var out []string
for i := 0; i < t.NumField(); i++ {
name, _, _ := strings.Cut(t.Field(i).Tag.Get("json"), ",")
if name != "" && name != "-" {
out = append(out, name)
}
}
return out
}
// **A declaration reaches the bus with its listed fields and no other** (ADR 0289 §12): the mirror's value
// carries exactly DeclaredMirroredFields, and every field a manifest's declaration has is on that list. A
// field a declaration gains (package K1 widens it, ADR 0348 clause 4) fails here until it is listed — read,
// judged not a secret and carried — so nothing reaches the declared-config mirror unreviewed.
func TestTheDeclaredConfigMirrorCarriesOnlyItsListedFields(t *testing.T) {
if got := jsonFields(reflect.TypeOf(DeclaredMirrored{})); !slices.Equal(got, DeclaredMirroredFields) {
t.Errorf("the mirrored declaration carries %v, its list is %v", got, DeclaredMirroredFields)
}
for _, field := range jsonFields(reflect.TypeOf(catalogue.SettingDeclaration{})) {
if !slices.Contains(DeclaredMirroredFields, field) {
t.Errorf("a declaration has the field %q, which the declared-config mirror does not list: list it "+
"in DeclaredMirrored (and say why it is no secret), or say here why it stays off the bus", field)
}
}
if got := jsonFields(reflect.TypeOf(DeclaredConfigMirrored{})); !slices.Equal(got, []string{"module", "settings"}) {
t.Errorf("the mirrored module carries %v", got)
}
}
+24
View File
@@ -164,6 +164,30 @@ type CheckRequest struct {
// each, from the merge it judged, into the scratch namespace once it has judged and not failed, for the walk to // each, from the merge it judged, into the scratch namespace once it has judged and not failed, for the walk to
// take. A build seat from before it builds nothing, and the walk builds as before. // take. A build seat from before it builds nothing, and the walk builds as before.
Builds []CheckBuild `json:"builds,omitempty"` Builds []CheckBuild `json:"builds,omitempty"`
// Try marks a branch tried before its pull request (novox/hq issue 686): the repository's own check for a
// pushed branch or commit, asked by a builder through the controller's `try` verb. Its outcome is read from
// the build's log (StepTried) and said to nobody: no status on the forge, nothing recorded or registered,
// nothing built for a merge. Tests narrows what the repository's script runs to the tests a pattern names,
// and StoreOnly to its store tests; a script that reads neither runs whole.
Try bool `json:"try,omitempty"`
Tests string `json:"tests,omitempty"`
StoreOnly bool `json:"store-only,omitempty"`
}
// StepTried is the step a build seat says a try's outcome under, in the build's log: one line, a Tried as JSON
// (novox/hq issue 686).
const StepTried = "tried"
// Tried is what came of a branch tried before its pull request (novox/hq issue 686): the verdict of the
// repository's own check for the ref merged into the trunk, how long it took, and the lines that name what
// failed, picked from the whole of its output (issue 460).
type Tried struct {
Verdict string `json:"verdict"`
Summary string `json:"summary"`
Took string `json:"took"`
Head string `json:"head,omitempty"`
Trunk string `json:"trunk,omitempty"`
Failed []string `json:"failed,omitempty"`
} }
// CheckBuild is one module a check builds for its merge: from the merge it made when Merged — a module built from // CheckBuild is one module a check builds for its merge: from the merge it made when Merged — a module built from
+19
View File
@@ -96,8 +96,27 @@ type BuildStart struct {
// an earlier taker left unfinished — most often a build agent replaced mid-build — and started // an earlier taker left unfinished — most often a build agent replaced mid-build — and started
// again from the beginning (novox/hq issue 594). 0 from a machine that predates it. // again from the beginning (novox/hq issue 594). 0 from a machine that predates it.
Delivery int `json:"delivery,omitempty"` Delivery int `json:"delivery,omitempty"`
// Try marks the start of a branch tried before its pull request (novox/hq issue 686), which names no
// repository and no ref (startOf).
Try bool `json:"try,omitempty"`
} }
// startOf is what a build machine says of the ask it took. **A try's start names no repository and no commit**
// (novox/hq issue 686): the forge's holder writes mesh/merge-gate pending on an open pull request whose head is
// the commit a start names, and a try of that head is no check of the pull request — nothing on the forge moves
// for a try. Its id, machine and time are all the queue reads of a start.
func startOf(r BuildRequest, on string, delivery int, at time.Time) BuildStart {
s := BuildStart{ID: r.ID, Repository: r.Repository, Path: r.Path, Ref: r.Ref, On: on,
At: at.UTC().Format(time.RFC3339Nano), Delivery: delivery}
if IsTry(r) {
s.Repository, s.Path, s.Ref, s.Try = "", "", "", true
}
return s
}
// IsTry says an ask is a branch tried before its pull request (novox/hq issue 686).
func IsTry(r BuildRequest) bool { return r.Check != nil && r.Check.Try }
// Restarted is how a build handed out again says so (novox/hq issue 594), in its log and in its check's // Restarted is how a build handed out again says so (novox/hq issue 594), in its log and in its check's
// report, so a second run never reads as the first. Neutral on why: a holder replaced mid-build, a paused // report, so a second run never reads as the first. Neutral on why: a holder replaced mid-build, a paused
// holder handing an ask back as it arrived, and a bus that held its acknowledgements past the wait all // holder handing an ask back as it arrived, and a bus that held its acknowledgements past the wait all
+1 -4
View File
@@ -401,10 +401,7 @@ func (b *natsBuild) Done() error { return b.msg.Ack() }
// Began publishes that this machine has taken the build, into the stream like the outcome, so a // Began publishes that this machine has taken the build, into the stream like the outcome, so a
// reader that asks afterwards sees when it started as well as how it ended. // reader that asks afterwards sees when it started as well as how it ended.
func (b *natsBuild) Began(ctx context.Context) error { func (b *natsBuild) Began(ctx context.Context) error {
body, err := json.Marshal(BuildStart{ body, err := json.Marshal(startOf(b.request, b.on, b.delivery, time.Now()))
ID: b.request.ID, Repository: b.request.Repository, Path: b.request.Path,
Ref: b.request.Ref, On: b.on, At: time.Now().UTC().Format(time.RFC3339Nano), Delivery: b.delivery,
})
if err != nil { if err != nil {
return err return err
} }
-5
View File
@@ -81,11 +81,6 @@ const (
// steps from. Said after every save of the serving controller; a delivery's holder also asks for the // steps from. Said after every save of the serving controller; a delivery's holder also asks for the
// walks it follows, so a save made elsewhere is found by comparison, never lost. // walks it follows, so a save made elsewhere is found by comparison, never lost.
KeyPlanMoved = "plan-moved" KeyPlanMoved = "plan-moved"
// KeyAssignmentChanged and KeyDeclaredConfigChanged: a key of the controller's assignments or declared-config
// mirror was written or removed (novox/hq ADR 0289, design 54 §3, design 57 F3): which key, how and at which
// revision of the bucket — never the value, which the bucket holds.
KeyAssignmentChanged = "assignment-changed"
KeyDeclaredConfigChanged = "declared-config-changed"
) )
// Applied is what a machine now runs, as the mesh states it. // Applied is what a machine now runs, as the mesh states it.
+3
View File
@@ -427,6 +427,9 @@ func ReadCheckBuilds(ctx context.Context, js *broker.JetStream, seat string, pul
// safeKey is an id that can be a key, and a subject token, as it is: a build id, never a pattern. // safeKey is an id that can be a key, and a subject token, as it is: a build id, never a pattern.
var safeKey = regexp.MustCompile(`^[A-Za-z0-9_-]+$`) var safeKey = regexp.MustCompile(`^[A-Za-z0-9_-]+$`)
// SafeBuildID says an id can be a subject token as it is: a build's id, never a pattern.
func SafeBuildID(id string) bool { return safeKey.MatchString(id) }
// Cancelled is what the controller writes for an ask it cancelled. // Cancelled is what the controller writes for an ask it cancelled.
type Cancelled struct { type Cancelled struct {
At string `json:"at"` At string `json:"at"`
+9 -2
View File
@@ -26,6 +26,8 @@ import (
// - a check that has waited CheckWaitsAtMost goes before every delivery asked after it, so no check // - a check that has waited CheckWaitsAtMost goes before every delivery asked after it, so no check
// waits for ever behind a stream of deliveries; // waits for ever behind a stream of deliveries;
// - within a kind, first come, first served, by when it was asked; // - within a kind, first come, first served, by when it was asked;
// - a branch tried before its pull request goes after everything else, however long it has waited: it is
// never promoted ahead of a delivery, and never ahead of a pull request's check (novox/hq issue 686);
// - an ask whose kind cannot be told counts as a check: a delivery taken late breaks a budget, a check // - an ask whose kind cannot be told counts as a check: a delivery taken late breaks a budget, a check
// taken late breaks nothing. // taken late breaks nothing.
// //
@@ -64,8 +66,12 @@ func (a QueuedAsk) askedOrStored() time.Time {
// QueueOrder is the order the waiting asks are to be handed out in, now: the deliveries and every check // QueueOrder is the order the waiting asks are to be handed out in, now: the deliveries and every check
// that has waited CheckWaitsAtMost, by when each was asked, then the other checks, by when each was asked. // that has waited CheckWaitsAtMost, by when each was asked, then the other checks, by when each was asked.
func QueueOrder(waiting []QueuedAsk, now time.Time) []QueuedAsk { func QueueOrder(waiting []QueuedAsk, now time.Time) []QueuedAsk {
var first, then []QueuedAsk var first, then, tries []QueuedAsk
for _, a := range waiting { for _, a := range waiting {
if IsTry(a.Request) {
tries = append(tries, a)
continue
}
if KindOf(a.Request) == KindCheck && now.Sub(a.askedOrStored()) < CheckWaitsAtMost { if KindOf(a.Request) == KindCheck && now.Sub(a.askedOrStored()) < CheckWaitsAtMost {
then = append(then, a) then = append(then, a)
continue continue
@@ -82,7 +88,8 @@ func QueueOrder(waiting []QueuedAsk, now time.Time) []QueuedAsk {
} }
byAsking(first) byAsking(first)
byAsking(then) byAsking(then)
return append(first, then...) byAsking(tries)
return append(append(first, then...), tries...)
} }
// toMove is the asks to move to the tail, in this order, for the queue to stand as order says: everything // toMove is the asks to move to the tail, in this order, for the queue to stand as order says: everything
+55
View File
@@ -0,0 +1,55 @@
package link
import (
"encoding/json"
"reflect"
"strings"
"testing"
"time"
)
// The start of a try names no repository and no commit (novox/hq issue 686): the forge's holder writes
// mesh/merge-gate pending on the open pull request whose head a start names, and nothing on the forge moves
// for a try. A pull request's check still names both.
func TestATrysStartNamesNoCommitTheForgeCouldMatch(t *testing.T) {
sha := strings.Repeat("c0ffee", 7)[:40]
at := time.Date(2026, 10, 14, 0, 0, 0, 0, time.UTC)
try := BuildRequest{ID: "build-1", Repository: "https://forge/novox/a.git", Ref: sha,
Check: &CheckRequest{Owner: "novox", Repo: "a", Try: true}}
body, _ := json.Marshal(startOf(try, "ace", 1, at))
if strings.Contains(string(body), sha) || strings.Contains(string(body), "novox/a") {
t.Fatalf("a try's start names what the forge matches a pull request by: %s", body)
}
if s := startOf(try, "ace", 2, at); s.ID != "build-1" || s.On != "ace" || !s.Try || s.Delivery != 2 || s.At == "" {
t.Fatalf("a try's start lost what the queue reads: %+v", s)
}
pull := try
pull.Check = &CheckRequest{Owner: "novox", Repo: "a", Number: 7}
if s := startOf(pull, "ace", 1, at); s.Ref != sha || s.Repository == "" || s.Try {
t.Fatalf("a pull request's check's start: %+v", s)
}
}
// A try stays behind every delivery and every pull request's check, however long it has waited: the ten
// minutes that put a check ahead of later deliveries never promote a try (novox/hq issue 686).
func TestATryIsNeverHandedOutBeforeADeliveryOrAPullRequestsCheck(t *testing.T) {
now := time.Date(2026, 10, 12, 12, 0, 0, 0, time.UTC)
try := func(seq uint64, id string, asked time.Time) QueuedAsk {
a := anAsk(seq, id, asked, true)
a.Request.Check.Try = true
return a
}
waiting := []QueuedAsk{
try(1, "try-old", now.Add(-40*time.Minute)),
try(2, "try-young", now.Add(-time.Minute)),
anAsk(3, "check-young", now.Add(-2*time.Minute), true),
anAsk(4, "delivery", now.Add(-time.Minute), false),
}
want := []string{"delivery", "check-young", "try-old", "try-young"}
if got := idsOf(QueueOrder(waiting, now)); !reflect.DeepEqual(got, want) {
t.Fatalf("handed out as %v, want %v", got, want)
}
if !SafeBuildID("build-20261014T000000-1a2b") || SafeBuildID("x.>") || SafeBuildID("*") || SafeBuildID("") {
t.Fatal("an id that is a pattern passes for a build's id")
}
}
-344
View File
@@ -1,344 +0,0 @@
// Package recordmirror writes the controller's record mirrors (novox/hq ADR 0289, design 54): a bucket per kind
// of record, one key per record, written from the store only when the record changed, and one event on the
// controller's seat per write.
//
// **The store stays the record.** A mirror is built from what the store committed, so a change rolled back
// is never written. Every pass reads what the bucket holds and compares it byte for byte with what the store
// says: a pass over unchanged records writes nothing and says nothing (ADR 0289 §1), and a value somebody
// else wrote behind the writer's back (a former lease holder) differs from the store and is put right on
// the next pass. Each write names the revision it read (compare-and-set), so a write in between is noticed,
// never overwritten blind.
//
// **Only the lease holder writes** (ADR 0289 §9, design 54 §2): a pass begins by asking whether this
// controller holds the lease, and asks again before every write; without it, it writes nothing.
//
// **The bucket is the authority, the event a notice** (design 54 §3): the event is sent after the write
// succeeded, so there is no event for a value the bucket does not hold. A refused event is tried once more
// and then counted as lost; a watcher that must not miss a change watches the bucket.
//
// What this package does not do yet is design 54 §4's audit (probe D17 and its healer H6), which compares a
// mirror with its store behind the writer's back; novox/hq issue 419 carries it.
package recordmirror
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"regexp"
"sort"
"time"
"github.com/nats-io/nats.go/jetstream"
"github.com/novox/mesh-controller/internal/broker"
"github.com/novox/mesh-controller/internal/link"
)
// A Mirror is one kind of record the controller keeps on the bus: its local name, its bucket and the event
// each write raises on the controller's seat.
type Mirror struct {
Name string
Bucket string
Event string
}
// The mirrors this package writes (novox/hq design 57, package F3). The assignments mirror is design 54's,
// keyed `<node>.<module>`; the declared config is each registered module's config declarations as its
// manifest holds them, keyed by the module, which mesh-config resolves a module's config values from
// (ADR 0348 clauses 2 and 4).
var (
Assignments = Mirror{Name: broker.MirrorAssignments, Bucket: broker.AssignmentsBucket,
Event: link.KeyAssignmentChanged}
DeclaredConfig = Mirror{Name: broker.MirrorDeclaredConfig, Bucket: broker.DeclaredConfigBucket,
Event: link.KeyDeclaredConfigChanged}
)
// Mirrors is every mirror this package writes, in the order they are written.
func Mirrors() []Mirror { return []Mirror{Assignments, DeclaredConfig} }
// The changes an event says.
const (
Added = "added"
Changed = "changed"
Removed = "removed"
)
// Event is what one write says on the controller's seat (design 54 §3): which mirror and key, what kind of
// change, the names of the top-level fields that changed, the bucket's revision of the write, the lease's
// epoch and when. Never the value: a watcher reads it from the bucket at that revision or later.
type Event struct {
Mirror string `json:"mirror"`
Key string `json:"key"`
Change string `json:"change"`
Fields []string `json:"fields,omitempty"`
Revision uint64 `json:"revision"`
Epoch uint64 `json:"epoch,omitempty"`
At time.Time `json:"at"`
}
// Announcer says an event on a seat; link.OverNATS is one, and so is link.Bus.
type Announcer interface {
PublishSeatEvent(ctx context.Context, seat, event string, body []byte) error
}
// MarkerTTL is how long the marker a purge leaves is kept (design 54 §2): a day, so the markers of removed
// records do not pile up in a bucket that has no age.
const MarkerTTL = 24 * time.Hour
// ErrNotHeld is a pass refused because this controller does not hold the lease: it wrote nothing.
var ErrNotHeld = errors.New("this controller does not hold the lease, so it writes no mirror")
// Writer writes one mirror. Not safe for use by two goroutines at once: one controller's one loop owns it.
type Writer struct {
Mirror Mirror
KV jetstream.KeyValue
Announce Announcer
// Epoch is the lease's epoch the events carry; nil carries none.
Epoch func() uint64
// Holds says this controller holds the lease. Asked before every pass and every write; a writer
// without it writes nothing (design 54 §2). Required: nil holds nothing.
Holds func() bool
// Now is the clock the events are dated by; nil is time.Now.
Now func() time.Time
known map[string]kept
// LostEvents counts the events refused twice since the writer was made.
LostEvents int
}
type kept struct {
value []byte
revision uint64
}
// Result is what one pass did: keys written, keys removed, and keys refused with why.
type Result struct {
Written int
Removed int
Refused int
}
// Sync makes the bucket hold want, one key per record: writes each value that differs from what the key
// holds, purges each key no record is left for, and raises one event per write. A key the bus cannot hold
// is refused alone, and the others are written; the error then names it.
func (w *Writer) Sync(ctx context.Context, want map[string][]byte) (Result, error) {
var r Result
if !w.holds() {
return r, ErrNotHeld
}
// What the bucket holds now, every pass: a value written behind this writer's back differs from the
// store's and is written again below.
if err := w.fill(ctx); err != nil {
return r, fmt.Errorf("reading the %s mirror: %w", w.Mirror.Name, err)
}
var problems []error
keys := make([]string, 0, len(want))
for k := range want {
keys = append(keys, k)
}
sort.Strings(keys)
for _, key := range keys {
value := want[key]
before, had := w.known[key]
if had && bytes.Equal(before.value, value) {
continue
}
if err := validKey(key); err != nil {
r.Refused++
problems = append(problems, err)
continue
}
if !w.holds() {
return r, errors.Join(append(problems, ErrNotHeld)...)
}
revision, wrote, err := w.put(ctx, key, value, before, had)
if err != nil {
problems = append(problems, fmt.Errorf("writing %s: %w", key, err))
continue
}
w.known[key] = kept{value: append([]byte(nil), value...), revision: revision}
if !wrote {
continue
}
r.Written++
change, fields := Added, []string(nil)
if had {
change, fields = Changed, changedFields(before.value, value)
}
w.say(ctx, key, change, fields, revision)
}
gone := make([]string, 0)
for key := range w.known {
if _, ok := want[key]; !ok {
gone = append(gone, key)
}
}
sort.Strings(gone)
for _, key := range gone {
k := w.known[key]
if !w.holds() {
return r, errors.Join(append(problems, ErrNotHeld)...)
}
err := w.KV.Purge(ctx, key, jetstream.LastRevision(k.revision), jetstream.PurgeTTL(MarkerTTL))
if refusedByRevision(err) {
// Read it again: purged already, or moved by another writer.
if e, gerr := w.KV.Get(ctx, key); errors.Is(gerr, jetstream.ErrKeyNotFound) {
delete(w.known, key)
continue
} else if gerr == nil && w.holds() {
err = w.KV.Purge(ctx, key, jetstream.LastRevision(e.Revision()), jetstream.PurgeTTL(MarkerTTL))
}
}
if err != nil {
problems = append(problems, fmt.Errorf("removing %s: %w", key, err))
continue
}
delete(w.known, key)
r.Removed++
revision := uint64(0)
if status, err := w.KV.Status(ctx); err == nil {
// The purge's own sequence is the stream's last one only when nothing wrote in between, which no
// other writer may; said as the bucket's revision after it.
if s, ok := status.(interface{ StreamInfo() *jetstream.StreamInfo }); ok {
revision = s.StreamInfo().State.LastSeq
}
}
w.say(ctx, key, Removed, nil, revision)
}
return r, errors.Join(problems...)
}
// put writes one value by compare-and-set and answers its revision, and whether this call wrote it (false
// when the bucket already held it).
func (w *Writer) put(ctx context.Context, key string, value []byte, before kept, had bool) (uint64, bool, error) {
var (
revision uint64
err error
)
if had {
revision, err = w.KV.Update(ctx, key, value, before.revision)
} else {
revision, err = w.KV.Create(ctx, key, value)
}
if !refusedByRevision(err) {
return revision, err == nil, err
}
// Refused: another writer in between, or this writer's own write whose acknowledgement was lost.
e, gerr := w.KV.Get(ctx, key)
switch {
case errors.Is(gerr, jetstream.ErrKeyNotFound):
if !w.holds() {
return 0, false, errors.New("the key moved and this controller no longer holds the lease")
}
revision, err = w.KV.Create(ctx, key, value)
return revision, err == nil, err
case gerr != nil:
return 0, false, gerr
case bytes.Equal(e.Value(), value):
return e.Revision(), false, nil
case !w.holds():
return 0, false, errors.New("the key moved and this controller no longer holds the lease")
}
revision, err = w.KV.Update(ctx, key, value, e.Revision())
return revision, err == nil, err
}
// say raises one write's event, tried once more when refused and then counted as lost.
func (w *Writer) say(ctx context.Context, key, change string, fields []string, revision uint64) {
if w.Announce == nil {
return
}
e := Event{Mirror: w.Mirror.Name, Key: key, Change: change, Fields: fields, Revision: revision, At: w.now().UTC()}
if w.Epoch != nil {
e.Epoch = w.Epoch()
}
body, err := json.Marshal(e)
if err != nil {
w.LostEvents++
return
}
for try := 0; try < 2; try++ {
if err = w.Announce.PublishSeatEvent(ctx, broker.ControllerSeat, w.Mirror.Event, body); err == nil {
return
}
}
w.LostEvents++
}
// fill reads what the bucket holds now, so a controller started again writes only what differs.
func (w *Writer) fill(ctx context.Context) error {
w.known = map[string]kept{}
watch, err := w.KV.WatchAll(ctx, jetstream.IgnoreDeletes())
if err != nil {
return err
}
defer func() { _ = watch.Stop() }()
for {
select {
case <-ctx.Done():
return ctx.Err()
case e := <-watch.Updates():
if e == nil {
return nil
}
w.known[e.Key()] = kept{value: append([]byte(nil), e.Value()...), revision: e.Revision()}
}
}
}
func (w *Writer) holds() bool { return w.Holds != nil && w.Holds() }
func (w *Writer) now() time.Time {
if w.Now == nil {
return time.Now()
}
return w.Now()
}
// refusedByRevision is the server refusing a compare-and-set: the key is not at the revision named.
func refusedByRevision(err error) bool {
if err == nil {
return false
}
if errors.Is(err, jetstream.ErrKeyExists) {
return true
}
var apiErr *jetstream.APIError
return errors.As(err, &apiErr) && apiErr.ErrorCode == jetstream.JSErrCodeStreamWrongLastSequence
}
// keyChars is what a bucket's key may be made of.
var keyChars = regexp.MustCompile(`^[-/_=.a-zA-Z0-9]+$`)
// validKey refuses a key the bus cannot hold, naming it.
func validKey(key string) error {
if key == "" || !keyChars.MatchString(key) || key[0] == '.' || key[len(key)-1] == '.' {
return fmt.Errorf("%q cannot be a key of a bucket: letters, digits, '-', '_', '=', '/' and '.', not at "+
"either end", key)
}
return nil
}
// changedFields names the top-level fields whose values differ between two encoded objects, sorted; nil
// when either is not an object.
func changedFields(before, after []byte) []string {
var a, b map[string]json.RawMessage
if json.Unmarshal(before, &a) != nil || json.Unmarshal(after, &b) != nil || a == nil || b == nil {
return nil
}
var out []string
for k, v := range b {
if old, ok := a[k]; !ok || !bytes.Equal(old, v) {
out = append(out, k)
}
}
for k := range a {
if _, ok := b[k]; !ok {
out = append(out, k)
}
}
sort.Strings(out)
return out
}
-178
View File
@@ -1,178 +0,0 @@
package recordmirror
import (
"encoding/json"
"strings"
"testing"
"github.com/novox/mesh-controller/internal/broker"
"github.com/novox/mesh-controller/internal/catalogue"
"github.com/novox/mesh-controller/internal/inventory"
)
// **Design 57's F3, end to end** (novox/hq issue 652): from the controller's store to a real bus. An
// assignment changes exactly one key of the assignments mirror and raises exactly one `assignment-changed`;
// a registration changes exactly one key of the declared-config mirror and raises exactly one
// `declared-config-changed`; and the same records written again change nothing and say nothing — a module
// registered again unchanged included, although the store's registration time moved.
func TestNatsAnAssignmentAndARegistrationEachChangeOneKeyAndRaiseOneEvent(t *testing.T) {
inv := inventory.ForTest(t)
ctx := t.Context()
js, api := aMirrorBus(t)
assignments := aWriter(t, js, api, Assignments)
declared := aWriter(t, js, api, DeclaredConfig)
heardAssignments := listen(t, js.Conn(), "mesh.seat."+broker.ControllerSeat+".event."+Assignments.Event)
heardDeclared := listen(t, js.Conn(), "mesh.seat."+broker.ControllerSeat+".event."+DeclaredConfig.Event)
pass := func() (Result, Result) {
t.Helper()
a, err := inv.AssignmentsMirror(ctx)
if err != nil {
t.Fatal(err)
}
ra, err := assignments.Sync(ctx, a)
if err != nil {
t.Fatal(err)
}
d, err := inv.DeclaredConfigMirror(ctx)
if err != nil {
t.Fatal(err)
}
rd, err := declared.Sync(ctx, d)
if err != nil {
t.Fatal(err)
}
return ra, rd
}
if _, err := inv.AddNode(ctx, "anchor"); err != nil {
t.Fatal(err)
}
ui := catalogue.Manifest{Module: "ui", Version: "1", Settings: map[string]catalogue.SettingDeclaration{
"font-size": {Kind: catalogue.KindPreference, Default: 14, Why: "readable on the desktop"}}}
if err := inv.RegisterModule(ctx, ui, inventory.Source{}); err != nil {
t.Fatal(err)
}
pass()
assignmentWrites, declaredWrites := writes(t, api, Assignments.Bucket), writes(t, api, DeclaredConfig.Bucket)
assignmentsSaid, declaredSaid := len(settle(heardAssignments, 0)), len(settle(heardDeclared, 1))
// **A registration**: a module new to the catalogue.
if err := inv.RegisterModule(ctx, catalogue.Manifest{Module: "notes", Version: "1"}, inventory.Source{}); err != nil {
t.Fatal(err)
}
ra, rd := pass()
if ra != (Result{}) || rd != (Result{Written: 1}) {
t.Errorf("a registration wrote assignments %+v and declared config %+v", ra, rd)
}
if got := writes(t, api, DeclaredConfig.Bucket) - declaredWrites; got != 1 {
t.Errorf("a registration changed %d keys of the declared config", got)
}
if got := writes(t, api, Assignments.Bucket) - assignmentWrites; got != 0 {
t.Errorf("a registration changed %d keys of the assignments", got)
}
said := settle(heardDeclared, declaredSaid+1)[declaredSaid:]
if len(said) != 1 || said[0].Key != "notes" || said[0].Change != Added {
t.Errorf("a registration raised %+v", said)
}
declaredWrites, declaredSaid = writes(t, api, DeclaredConfig.Bucket), declaredSaid+len(said)
// **An assignment.**
if _, err := inv.Assign(ctx, "anchor", "ui"); err != nil {
t.Fatal(err)
}
ra, rd = pass()
if ra != (Result{Written: 1}) || rd != (Result{}) {
t.Errorf("an assignment wrote assignments %+v and declared config %+v", ra, rd)
}
if got := writes(t, api, Assignments.Bucket) - assignmentWrites; got != 1 {
t.Errorf("an assignment changed %d keys of the assignments", got)
}
said = settle(heardAssignments, assignmentsSaid+1)[assignmentsSaid:]
if len(said) != 1 || said[0].Key != "anchor.ui" || said[0].Change != Added {
t.Errorf("an assignment raised %+v", said)
}
assignmentWrites, assignmentsSaid = writes(t, api, Assignments.Bucket), assignmentsSaid+len(said)
e, err := assignments.KV.Get(ctx, "anchor.ui")
if err != nil {
t.Fatal(err)
}
var value inventory.AssignmentMirrored
if err := json.Unmarshal(e.Value(), &value); err != nil || value.Node != "anchor" || value.Module != "ui" ||
value.Assigned.IsZero() {
t.Errorf("the assignment is mirrored as %s (%v)", e.Value(), err)
}
// **The same records written again change nothing**: an assignment made again, and a module registered
// again with the manifest it had.
if _, err := inv.Assign(ctx, "anchor", "ui"); err != nil {
t.Fatal(err)
}
if err := inv.RegisterModule(ctx, ui, inventory.Source{}); err != nil {
t.Fatal(err)
}
if ra, rd := pass(); ra != (Result{}) || rd != (Result{}) {
t.Errorf("unchanged records wrote assignments %+v and declared config %+v", ra, rd)
}
if writes(t, api, Assignments.Bucket) != assignmentWrites || writes(t, api, DeclaredConfig.Bucket) != declaredWrites {
t.Error("unchanged records changed a key")
}
if a, d := settle(heardAssignments, assignmentsSaid+1), settle(heardDeclared, declaredSaid+1); len(a) != assignmentsSaid ||
len(d) != declaredSaid {
t.Errorf("unchanged records raised %+v and %+v", a[assignmentsSaid:], d[declaredSaid:])
}
// **A declaration changed by a registration**: one key, one event, naming the field that moved.
ui.Version = "2"
ui.Settings["font-size"] = catalogue.SettingDeclaration{Kind: catalogue.KindPreference, Default: 16,
Why: "readable on the desktop"}
if err := inv.RegisterModule(ctx, ui, inventory.Source{}); err != nil {
t.Fatal(err)
}
if ra, rd := pass(); ra != (Result{}) || rd != (Result{Written: 1}) {
t.Errorf("a changed declaration wrote assignments %+v and declared config %+v", ra, rd)
}
said = settle(heardDeclared, declaredSaid+1)[declaredSaid:]
if len(said) != 1 || said[0].Key != "ui" || said[0].Change != Changed || strings.Join(said[0].Fields, ",") != "settings" {
t.Errorf("a changed declaration raised %+v", said)
}
got, err := declared.KV.Get(ctx, "ui")
if err != nil {
t.Fatal(err)
}
if want := `{"module":"ui","settings":{"font-size":{"kind":"preference","default":16,"why":"readable on the desktop"}}}`; string(got.Value()) != want {
t.Errorf("the declaration is mirrored as %s, want %s", got.Value(), want)
}
// **An unassignment** removes its key, said once.
if err := inv.Unassign(ctx, "anchor", "ui"); err != nil {
t.Fatal(err)
}
if ra, _ := pass(); ra != (Result{Removed: 1}) {
t.Errorf("an unassignment did %+v", ra)
}
said = settle(heardAssignments, assignmentsSaid+1)[assignmentsSaid:]
if len(said) != 1 || said[0].Key != "anchor.ui" || said[0].Change != Removed {
t.Errorf("an unassignment raised %+v", said)
}
// **A module no longer registered** has its key removed, said once as `removed` (ADR 0356 §4).
if err := inv.ForgetModule(ctx, "notes"); err != nil {
t.Fatal(err)
}
declaredWrites = writes(t, api, DeclaredConfig.Bucket)
declaredSaid = len(heardDeclared.all())
if _, rd := pass(); rd != (Result{Removed: 1}) {
t.Errorf("a module forgotten did %+v", rd)
}
if got := writes(t, api, DeclaredConfig.Bucket) - declaredWrites; got != 1 {
t.Errorf("a module forgotten changed %d keys of the declared config", got)
}
if _, err := declared.KV.Get(ctx, "notes"); err == nil {
t.Error("a module forgotten keeps its key")
}
said = settle(heardDeclared, declaredSaid+1)[declaredSaid:]
if len(said) != 1 || said[0].Key != "notes" || said[0].Change != Removed {
t.Errorf("a module forgotten raised %+v", said)
}
}
-356
View File
@@ -1,356 +0,0 @@
package recordmirror
import (
"encoding/json"
"errors"
"slices"
"sync"
"testing"
"time"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"github.com/novox/mesh-controller/internal/broker"
"github.com/novox/mesh-controller/internal/link"
"github.com/novox/mesh-controller/internal/testbus"
)
// A record mirror against a real bus (novox/hq design 54 §2, §3; ADR 0289): a change of a record changes
// exactly one key and raises exactly one event; a pass over unchanged records writes nothing and says
// nothing; a record gone is a purge said as `removed`.
// aMirrorBus is a bus of this test alone with the mesh's streams and the controller's buckets, as the
// controller asserts them at its start, and a counter of the writes made to one bucket.
func aMirrorBus(t *testing.T) (*broker.JetStream, jetstream.JetStream) {
t.Helper()
js, err := broker.Dial(testbus.URL(t))
if err != nil {
t.Fatal(err)
}
t.Cleanup(js.Close)
if err := broker.AssertMeshStreams(js); err != nil {
t.Fatal(err)
}
if err := js.EnsureControllerBuckets(); err != nil {
t.Fatal(err)
}
api, err := jetstream.New(js.Conn())
if err != nil {
t.Fatal(err)
}
return js, api
}
// heard counts what is said on one subject, each body kept.
type heard struct {
mu sync.Mutex
bodies []Event
}
func (h *heard) all() []Event {
h.mu.Lock()
defer h.mu.Unlock()
return append([]Event(nil), h.bodies...)
}
func listen(t *testing.T, conn *nats.Conn, subject string) *heard {
t.Helper()
h := &heard{}
sub, err := conn.Subscribe(subject, func(m *nats.Msg) {
var e Event
if err := json.Unmarshal(m.Data, &e); err != nil {
t.Errorf("an event that is not one: %s", m.Data)
return
}
h.mu.Lock()
h.bodies = append(h.bodies, e)
h.mu.Unlock()
})
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = sub.Unsubscribe() })
if err := conn.Flush(); err != nil {
t.Fatal(err)
}
return h
}
// writes is how many messages the bucket's stream has taken: every put and every purge is one.
func writes(t *testing.T, api jetstream.JetStream, bucket string) uint64 {
t.Helper()
s, err := api.Stream(t.Context(), "KV_"+bucket)
if err != nil {
t.Fatal(err)
}
info, err := s.Info(t.Context())
if err != nil {
t.Fatal(err)
}
return info.State.LastSeq
}
// settle waits until n events are heard or a second passed, and answers what was heard.
func settle(h *heard, n int) []Event {
deadline := time.Now().Add(time.Second)
for time.Now().Before(deadline) && len(h.all()) < n {
time.Sleep(10 * time.Millisecond)
}
// And a moment more, so an event too many is heard too.
time.Sleep(100 * time.Millisecond)
return h.all()
}
func aWriter(t *testing.T, js *broker.JetStream, api jetstream.JetStream, m Mirror) *Writer {
t.Helper()
kv, err := api.KeyValue(t.Context(), m.Bucket)
if err != nil {
t.Fatal(err)
}
return &Writer{Mirror: m, KV: kv, Announce: link.OverNATS{Conn: js.Conn(), JS: js.Context()},
Epoch: func() uint64 { return 7 }, Holds: func() bool { return true }}
}
func TestNatsAMirrorChangesOneKeyAndRaisesOneEventPerChange(t *testing.T) {
for _, m := range []Mirror{Assignments, DeclaredConfig} {
t.Run(m.Name, func(t *testing.T) {
js, api := aMirrorBus(t)
events := listen(t, js.Conn(), "mesh.seat."+broker.ControllerSeat+".event."+m.Event)
w := aWriter(t, js, api, m)
ctx := t.Context()
first := map[string][]byte{"anchor.gitea": []byte(`{"a":1}`)}
if _, err := w.Sync(ctx, first); err != nil {
t.Fatal(err)
}
before := writes(t, api, m.Bucket)
heardBefore := len(settle(events, 1))
// One record changes, one is added: two keys, two events, nothing else.
second := map[string][]byte{"anchor.gitea": []byte(`{"a":2}`), "anchor.forgejo": []byte(`{"a":1}`)}
r, err := w.Sync(ctx, second)
if err != nil {
t.Fatal(err)
}
if r.Written != 2 || r.Removed != 0 {
t.Errorf("a pass over one changed and one new record wrote %+v", r)
}
if got := writes(t, api, m.Bucket) - before; got != 2 {
t.Errorf("the bucket took %d writes for two changed records", got)
}
said := settle(events, heardBefore+2)[heardBefore:]
if len(said) != 2 {
t.Fatalf("two changes raised %d events: %+v", len(said), said)
}
changes := map[string]Event{}
for _, e := range said {
changes[e.Key] = e
}
if e := changes["anchor.gitea"]; e.Change != Changed || !slices.Equal(e.Fields, []string{"a"}) ||
e.Mirror != m.Name || e.Epoch != 7 || e.Revision == 0 {
t.Errorf("the change of anchor.gitea was said as %+v", e)
}
if e := changes["anchor.forgejo"]; e.Change != Added {
t.Errorf("the new anchor.forgejo was said as %+v", e)
}
// **An unchanged pass writes nothing**, and says nothing (ADR 0289 §1).
heardBefore = len(events.all())
before = writes(t, api, m.Bucket)
r, err = w.Sync(ctx, second)
if err != nil {
t.Fatal(err)
}
if r != (Result{}) {
t.Errorf("a pass over unchanged records did %+v", r)
}
if got := writes(t, api, m.Bucket) - before; got != 0 {
t.Errorf("a pass over unchanged records wrote the bucket %d times", got)
}
if said := settle(events, heardBefore+1)[heardBefore:]; len(said) != 0 {
t.Errorf("a pass over unchanged records raised %+v", said)
}
// **A record gone is a purge**, one write and one `removed`.
before = writes(t, api, m.Bucket)
r, err = w.Sync(ctx, map[string][]byte{"anchor.gitea": []byte(`{"a":2}`)})
if err != nil {
t.Fatal(err)
}
if r.Removed != 1 || r.Written != 0 {
t.Errorf("a record gone did %+v", r)
}
if got := writes(t, api, m.Bucket) - before; got != 1 {
t.Errorf("a record gone wrote the bucket %d times", got)
}
if _, err := w.KV.Get(ctx, "anchor.forgejo"); !errors.Is(err, jetstream.ErrKeyNotFound) {
t.Errorf("the key of a record gone is still read: %v", err)
}
said = settle(events, heardBefore+1)[heardBefore:]
if len(said) != 1 || said[0].Change != Removed || said[0].Key != "anchor.forgejo" {
t.Errorf("a record gone was said as %+v", said)
}
})
}
}
// **A controller started again writes nothing for what the bucket already holds** (design 54 §2: the writer's
// memory is filled from the bucket), and a key nobody holds a record for is removed.
func TestNatsAWriterStartedAgainWritesOnlyWhatDiffers(t *testing.T) {
js, api := aMirrorBus(t)
ctx := t.Context()
first := aWriter(t, js, api, Assignments)
held := map[string][]byte{"anchor.gitea": []byte(`{"a":1}`), "anchor.forgejo": []byte(`{"a":1}`)}
if _, err := first.Sync(ctx, held); err != nil {
t.Fatal(err)
}
before := writes(t, api, Assignments.Bucket)
again := aWriter(t, js, api, Assignments)
r, err := again.Sync(ctx, held)
if err != nil {
t.Fatal(err)
}
if r != (Result{}) || writes(t, api, Assignments.Bucket) != before {
t.Errorf("a writer started again over unchanged records did %+v", r)
}
}
// **Another writer in between is no lost value** (design 54 §2, compare-and-set): a write refused because the
// key moved reads it again, takes the revision when it already holds the value, and otherwise writes again.
func TestNatsAWriteRefusedByCompareAndSetReadsAgain(t *testing.T) {
js, api := aMirrorBus(t)
ctx := t.Context()
w := aWriter(t, js, api, Assignments)
if _, err := w.Sync(ctx, map[string][]byte{"anchor.gitea": []byte(`{"a":1}`)}); err != nil {
t.Fatal(err)
}
// Behind its back, as a former lease holder would.
if _, err := w.KV.Put(ctx, "anchor.gitea", []byte(`{"a":9}`)); err != nil {
t.Fatal(err)
}
r, err := w.Sync(ctx, map[string][]byte{"anchor.gitea": []byte(`{"a":2}`)})
if err != nil {
t.Fatal(err)
}
if r.Written != 1 {
t.Errorf("a refused write was not made again: %+v", r)
}
e, err := w.KV.Get(ctx, "anchor.gitea")
if err != nil || string(e.Value()) != `{"a":2}` {
t.Fatalf("the bucket holds %v, %v", e, err)
}
}
// **A value written behind the writer's back is put right on the next pass** (ADR 0289 §10): a former lease
// holder's write differs from the store, and the store's value is written again, said as one change.
func TestNatsAForeignWriteIsCorrectedOnTheNextPass(t *testing.T) {
js, api := aMirrorBus(t)
ctx := t.Context()
events := listen(t, js.Conn(), "mesh.seat."+broker.ControllerSeat+".event."+Assignments.Event)
w := aWriter(t, js, api, Assignments)
store := map[string][]byte{"anchor.gitea": []byte(`{"a":1}`)}
if _, err := w.Sync(ctx, store); err != nil {
t.Fatal(err)
}
heardBefore := len(settle(events, 1))
if _, err := w.KV.Put(ctx, "anchor.gitea", []byte(`{"a":9}`)); err != nil {
t.Fatal(err)
}
r, err := w.Sync(ctx, store)
if err != nil {
t.Fatal(err)
}
if r != (Result{Written: 1}) {
t.Errorf("a pass after a foreign write did %+v", r)
}
e, err := w.KV.Get(ctx, "anchor.gitea")
if err != nil || string(e.Value()) != `{"a":1}` {
t.Fatalf("the bucket still holds %v, %v", e, err)
}
if said := settle(events, heardBefore+1)[heardBefore:]; len(said) != 1 || said[0].Change != Changed {
t.Errorf("the correction was said as %+v", said)
}
// And put right, it stays: the next pass writes nothing.
if r, err := w.Sync(ctx, store); err != nil || r != (Result{}) {
t.Errorf("a pass after the correction did %+v, %v", r, err)
}
}
// **A writer without the lease writes nothing** (ADR 0289 §9, design 54 §2): not a new key, not a change, not
// a removal, and it says so.
func TestNatsAWriterWithoutTheLeaseWritesNothing(t *testing.T) {
js, api := aMirrorBus(t)
ctx := t.Context()
held := aWriter(t, js, api, Assignments)
if _, err := held.Sync(ctx, map[string][]byte{"anchor.gitea": []byte(`{"a":1}`)}); err != nil {
t.Fatal(err)
}
before := writes(t, api, Assignments.Bucket)
for _, holds := range []func() bool{nil, func() bool { return false }} {
w := aWriter(t, js, api, Assignments)
w.Holds = holds
r, err := w.Sync(ctx, map[string][]byte{"anchor.gitea": []byte(`{"a":2}`), "anchor.forgejo": []byte(`{"a":1}`)})
if !errors.Is(err, ErrNotHeld) || r != (Result{}) {
t.Errorf("a writer without the lease did %+v, %v", r, err)
}
r, err = w.Sync(ctx, map[string][]byte{})
if !errors.Is(err, ErrNotHeld) || r != (Result{}) {
t.Errorf("a writer without the lease removed: %+v, %v", r, err)
}
}
// And one that loses it mid-pass stops before its next write.
asked := 0
w := aWriter(t, js, api, Assignments)
w.Holds = func() bool { asked++; return asked <= 2 }
r, err := w.Sync(ctx, map[string][]byte{"anchor.a": []byte(`{}`), "anchor.b": []byte(`{}`), "anchor.gitea": []byte(`{"a":1}`)})
if !errors.Is(err, ErrNotHeld) || r.Written != 1 {
t.Errorf("a writer that lost the lease mid-pass did %+v, %v", r, err)
}
if got := writes(t, api, Assignments.Bucket) - before; got != 1 {
t.Errorf("writers without the lease wrote the bucket %d times", got)
}
}
// **A removal asks for the lease too** (ADR 0289 §9): a writer that held the lease when its pass began and lost
// it before its first removal purges nothing, and says it does not hold the lease.
func TestNatsARemovalWithoutTheLeaseRemovesNothing(t *testing.T) {
js, api := aMirrorBus(t)
ctx := t.Context()
held := aWriter(t, js, api, Assignments)
if _, err := held.Sync(ctx, map[string][]byte{"anchor.gitea": []byte(`{"a":1}`)}); err != nil {
t.Fatal(err)
}
before := writes(t, api, Assignments.Bucket)
asked := 0
w := aWriter(t, js, api, Assignments)
w.Holds = func() bool { asked++; return asked == 1 }
r, err := w.Sync(ctx, map[string][]byte{})
if !errors.Is(err, ErrNotHeld) || r != (Result{}) || asked != 2 {
t.Errorf("a removal without the lease did %+v, %v (the lease asked %d times)", r, err, asked)
}
if got := writes(t, api, Assignments.Bucket) - before; got != 0 {
t.Errorf("a removal without the lease wrote the bucket %d times", got)
}
if _, err := w.KV.Get(ctx, "anchor.gitea"); err != nil {
t.Errorf("the key was removed without the lease: %v", err)
}
}
// **A key the bus cannot hold is refused alone**, and the others are written.
func TestAKeyTheBusCannotHoldIsRefusedAlone(t *testing.T) {
if validKey("anchor.gitea") != nil || validKey("bad key") == nil || validKey(".lead") == nil ||
validKey("") == nil || validKey("trail.") == nil {
t.Error("the keys a bucket may hold are not told apart")
}
}
func TestTheFieldsThatChangedAreNamed(t *testing.T) {
got := changedFields([]byte(`{"a":1,"b":{"x":1},"c":"same"}`), []byte(`{"a":2,"b":{"x":1},"d":true,"c":"same"}`))
if !slices.Equal(got, []string{"a", "d"}) {
t.Errorf("the fields that changed are %v", got)
}
if got := changedFields(nil, []byte(`{"a":1}`)); got != nil {
t.Errorf("an added key names fields %v", got)
}
}
-181
View File
@@ -1,181 +0,0 @@
package recordmirror
import (
"go/ast"
"go/parser"
"go/token"
"io/fs"
"os"
"path/filepath"
"slices"
"strconv"
"strings"
"testing"
"github.com/novox/mesh-controller/internal/broker"
)
// mayNameAMirror are the files of the controller that may name a record mirror's bucket outside their tests:
// where the buckets are defined, asserted and granted (internal/broker), the list a reader names in `reads`
// (internal/catalogue/state.go), and each mirror's own writer (this package and the loop that runs it). The
// audit of a mirror against its store (ADR 0289 §4, issue 419) joins this list when it is built.
var mayNameAMirror = []string{
"cmd/mesh-controller/record_mirrors.go",
"internal/broker/controller_buckets.go",
"internal/broker/writers.go",
"internal/catalogue/state.go",
}
// The packages whose names stand for a mirror, by import path, and the names in each.
const (
brokerPath = "github.com/novox/mesh-controller/internal/broker"
recordmirrorPath = "github.com/novox/mesh-controller/internal/recordmirror"
)
var mirrorNames = map[string][]string{
brokerPath: {"AssignmentsBucket", "DeclaredConfigBucket", "MirrorAssignments", "MirrorDeclaredConfig",
"ControllerMirrors"},
recordmirrorPath: {"Assignments", "DeclaredConfig", "Mirrors"},
}
// **Nothing in the controller reads a mirror to decide anything** (ADR 0289 §19 as ADR 0356 §2 changes it): no
// file of the controller's packages outside its tests names a mirror's bucket but the audit and a mirror's own
// writer. A reader elsewhere fails here, naming the file and what it named.
//
// **What it catches**: a name that stands for a mirror, qualified by the package it comes from under any
// local name the file imports it as (`broker.AssignmentsBucket`, `b.AssignmentsBucket`); the same names bare
// in a file of package broker or package recordmirror; and a string literal holding a mirror's bucket name.
// **What it does not catch**: a bucket name built at run time, such as `BucketName(ControllerSeat,
// "assignments")` or literals concatenated into one, and a mirror reached through a value passed in from an
// allowed file. Those are deliberate evasions rather than ordinary code; a reviewer judges them.
func TestOnlyTheAuditAndAMirrorsWriterReadAMirror(t *testing.T) {
var buckets []string
for _, m := range broker.ControllerMirrors() {
buckets = append(buckets, broker.BucketName(broker.ControllerSeat, m))
}
root := filepath.Join("..", "..")
found := mirrorReaders(t, root, buckets)
if len(found) > 0 {
t.Fatalf("a package of the controller reads a record mirror, which only the audit and the mirror's "+
"writer may (ADR 0289 §19, ADR 0356 §2):\n %s", strings.Join(found, "\n "))
}
}
// **The reader test sees what it says it sees**: in a planted tree, each planted reader is named, and an allowed
// file and a selector on a value are not.
func TestTheReaderTestFindsPlantedReaders(t *testing.T) {
root := t.TempDir()
plant := func(path, src string) {
full := filepath.Join(root, path)
if err := os.MkdirAll(filepath.Dir(full), 0o755); err != nil {
t.Fatal(err)
}
if err := os.WriteFile(full, []byte(src), 0o644); err != nil {
t.Fatal(err)
}
}
plant("cmd/mesh-controller/qualified.go", "package main\n\nimport \""+brokerPath+"\"\n\nvar a = broker.AssignmentsBucket\n")
plant("internal/x/aliased.go", "package x\n\nimport b \""+brokerPath+"\"\n\nvar a = b.DeclaredConfigBucket\n")
plant("internal/broker/zz_reader.go", "package broker\n\nfunc zzR() string { return AssignmentsBucket }\n")
plant("internal/inventory/literal.go", "package inventory\n\nvar a = \"mesh-controller_declared-config\"\n")
plant("internal/y/writer.go", "package y\n\nimport \""+recordmirrorPath+"\"\n\nvar a = recordmirror.Assignments\n")
// Allowed: a mirror's definition and its writer.
plant("internal/broker/controller_buckets.go", "package broker\n\nvar AssignmentsBucket = \"x\"\n")
plant("internal/broker/other.go", "package broker\n\nfunc other() string { return \"nothing\" }\n")
// A selector on a value (v.AssignmentsBucket) is no use of the mirror, but declaring a field of a mirror's
// name in package broker is named, on the safe side; and a bare use inside a call under a selector is found.
plant("internal/broker/field.go", "package broker\n\ntype s struct{ AssignmentsBucket string }\n\nfunc f(v s) string { return v.AssignmentsBucket }\n")
plant("internal/broker/nested.go", "package broker\n\nfunc g() int { return len(h(DeclaredConfigBucket).x) }\n")
found := mirrorReaders(t, root, []string{"mesh-controller_assignments", "mesh-controller_declared-config"})
want := []string{
"cmd/mesh-controller/qualified.go: broker.AssignmentsBucket",
"internal/broker/field.go: AssignmentsBucket",
"internal/broker/nested.go: DeclaredConfigBucket",
"internal/broker/zz_reader.go: AssignmentsBucket",
"internal/inventory/literal.go: \"mesh-controller_declared-config\"",
"internal/x/aliased.go: b.DeclaredConfigBucket",
"internal/y/writer.go: recordmirror.Assignments",
}
slices.Sort(found)
if !slices.Equal(found, want) {
t.Errorf("the planted readers were found as\n %s\nwant\n %s", strings.Join(found, "\n "),
strings.Join(want, "\n "))
}
}
// mirrorReaders is every place under root's cmd and internal, outside tests and the allowed files, that names a
// mirror.
func mirrorReaders(t *testing.T, root string, buckets []string) []string {
t.Helper()
var found []string
for _, dir := range []string{"cmd", "internal"} {
err := filepath.WalkDir(filepath.Join(root, dir), func(path string, e fs.DirEntry, err error) error {
if err != nil || e.IsDir() || !strings.HasSuffix(path, ".go") || strings.HasSuffix(path, "_test.go") {
return err
}
rel, err := filepath.Rel(root, path)
if err != nil {
return err
}
rel = filepath.ToSlash(rel)
if strings.HasPrefix(rel, "internal/recordmirror/") || slices.Contains(mayNameAMirror, rel) {
return nil
}
f, err := parser.ParseFile(token.NewFileSet(), path, nil, 0)
if err != nil {
return err
}
// Each import's local name, mapped to the package it stands for.
imported := map[string]string{}
for _, imp := range f.Imports {
p, _ := strconv.Unquote(imp.Path.Value)
if _, ok := mirrorNames[p]; !ok {
continue
}
local := filepath.Base(p)
if imp.Name != nil {
local = imp.Name.Name
}
imported[local] = p
}
// A file of the package that defines the names may use them bare.
var bare []string
switch f.Name.Name {
case "broker":
bare = mirrorNames[brokerPath]
case "recordmirror":
bare = mirrorNames[recordmirrorPath]
}
ast.Inspect(f, func(n ast.Node) bool {
switch n := n.(type) {
case *ast.SelectorExpr:
pkg, ok := n.X.(*ast.Ident)
if !ok {
return true
}
if p, imp := imported[pkg.Name]; imp && slices.Contains(mirrorNames[p], n.Sel.Name) {
found = append(found, rel+": "+pkg.Name+"."+n.Sel.Name)
}
// Neither a selector's own name nor its package is a bare use.
return false
case *ast.Ident:
if slices.Contains(bare, n.Name) {
found = append(found, rel+": "+n.Name)
}
case *ast.BasicLit:
for _, b := range buckets {
if n.Kind == token.STRING && strings.Contains(n.Value, b) {
found = append(found, rel+": "+n.Value)
}
}
}
return true
})
return nil
})
if err != nil {
t.Fatal(err)
}
}
return found
}
+19 -4
View File
@@ -54,11 +54,26 @@ if [ -n "$unformatted" ]; then
fi fi
CGO_ENABLED=0 go vet ./... CGO_ENABLED=0 go vet ./...
if command -v gcc >/dev/null 2>&1; then suite() {
CGO_ENABLED=1 go test -race -count=1 -timeout 30m ./... if command -v gcc >/dev/null 2>&1; then
CGO_ENABLED=1 go test -race -count=1 -timeout 30m "$@"
else
echo "NOT RACE-CHECKED: the toolchain holds no C compiler; the suite runs without the race detector"
CGO_ENABLED=0 go test -count=1 -timeout 30m "$@"
fi
}
# **A branch tried before its pull request may ask for a part** (novox/hq issue 686): the controller's `try` verb
# hands MESH_CHECK_TESTS, a `go test -run` pattern, or MESH_CHECK_STORE_ONLY=1. A pull request's merge check is
# handed neither and always runs whole; a part that passes is said as a part, never as the check.
if [ -n "${MESH_CHECK_TESTS:-}" ]; then
echo "TRIED IN PART: only the tests matching $MESH_CHECK_TESTS, in every package"
suite -run "$MESH_CHECK_TESTS" ./...
exit 0
fi
if [ "${MESH_CHECK_STORE_ONLY:-}" = 1 ]; then
echo "TRIED IN PART: only the store tests listed in testdata/store-tests"
else else
echo "NOT RACE-CHECKED: the toolchain holds no C compiler; the suite runs without the race detector" suite ./...
CGO_ENABLED=0 go test -count=1 -timeout 30m ./...
fi fi
# **The store tests listed ran** (novox/hq issue 459): a test that needs the store skips without one, and a skip # **The store tests listed ran** (novox/hq issue 459): a test that needs the store skips without one, and a skip
+1
View File
@@ -77,6 +77,7 @@
"replay", "replay",
"kill", "kill",
"forget-checks", "forget-checks",
"try",
"pause", "pause",
"resume", "resume",
"hand-act", "hand-act",
+1 -1
View File
@@ -8,7 +8,7 @@ private addresses. That is acceptable while mesh-controller is private; if it ev
a second place to clean besides the catalogue itself. a second place to clean besides the catalogue itself.
mesh-catalog e0bc919d4945bb5f13f971f74643a0b714cf45da modules/*/module.json, modules/nats/Dockerfile mesh-catalog e0bc919d4945bb5f13f971f74643a0b714cf45da modules/*/module.json, modules/nats/Dockerfile
mesh-host c8920622516548978d92e734f3991a262f98a742 examples/foundation-first-node.lock mesh-host 7a41e120635b3f7b6cb94cca8b3fc4093e4e6110 examples/foundation-first-node.lock
mesh-sdk 5933eb757d443293631be2bed27a96288c2348d8 conformance/events/module-event.json mesh-sdk 5933eb757d443293631be2bed27a96288c2348d8 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:
+29 -79
View File
@@ -28,12 +28,11 @@
// image, and this file is applied on a machine with no mesh to ask about anything (ADR 0006). The // image, and this file is applied on a machine with no mesh to ask about anything (ADR 0006). The
// registry prefix is the lab's; the digests are upstream's. // registry prefix is the lab's; the digests are upstream's.
// //
// **The store is raised where the `postgres` module keeps it** (novox/hq ADR 0338): the container // The store's data is a NAMED VOLUME, not a directory on the machine. A directory the host
// `postgres`, on the module's pgvector image, its data in the directory /var/lib/mesh-store owned by // creates is owned by root, and the database runs as somebody else inside the container — so it
// the image's postgres user, its superuser password read from the module's own-secret file — so the // could not write, and the container crash-looped. A named volume lets the image set up its own
// store step adopts it as it runs. Pinned upstream by the module's own digest, not through the lab's // ownership, and outlives the container, which is what you want for the thing holding the mesh's
// registry prefix, because the two must be the same reference (mesh-host internal/bootstrap holds // state.
// them to each other against the module's manifest).
{ {
"declaration": 1, "declaration": 1,
@@ -84,28 +83,17 @@
"boot": "enabled", "boot": "enabled",
"restart-on": ["base-filter"] "restart-on": ["base-filter"]
}, },
// The store's data is a directory owned by the image's postgres user (999), made before the
// container starts: a directory the runtime made would be root's, and the database could not
// write it. Measured on the module's image 2026-10-13: it initialises a new store on a directory
// made so (novox/hq ADR 0338 §5).
{
"id": "store-data",
"type": "directory",
"path": "/var/lib/mesh-store",
"mode": "0700",
"owner": "999:70"
},
{ {
"id": "store", "id": "store",
"type": "container", "type": "container",
"name": "postgres", "name": "mesh-store",
"image": "pgvector/pgvector@sha256:cf134a767f474095eeba57e0117be8e568e011a63f33fbf252f14c9b760f8e6f", "image": "192.0.2.250:5000/postgres@sha256:7abf537131b66ed5af448d90653abf1679b0c7e9a1f07efdd4c3108a401b259a",
"env": { "env": {
"POSTGRES_PASSWORD": "bootstrap", "POSTGRES_PASSWORD": "bootstrap",
"PGDATA": "/var/lib/postgresql/data/pgdata" "PGDATA": "/var/lib/postgresql/data/pgdata"
}, },
"ports": ["5432:5432"], "ports": ["5432:5432"],
"volumes": ["/var/lib/mesh-store:/var/lib/postgresql/data"] "volumes": ["mesh-store-data:/var/lib/postgresql/data"]
}, },
// Over TCP, not the socket. While the store initialises it runs a temporary server on the // Over TCP, not the socket. While the store initialises it runs a temporary server on the
// socket ONLY, then stops it and starts the real one — so a socket check passes, the action // socket ONLY, then stops it and starts the real one — so a socket check passes, the action
@@ -114,21 +102,21 @@
{ {
"id": "store-ready", "id": "store-ready",
"type": "action", "type": "action",
"in": "postgres", "in": "mesh-store",
"command": ["sh", "-c", "for i in $(seq 1 180); do pg_isready -h 127.0.0.1 -U postgres >/dev/null 2>&1 && exit 0; sleep 1; done; echo 'the store did not answer within 180s; its own last words follow'; pg_isready -h 127.0.0.1 -U postgres; tail -n 20 /var/lib/postgresql/data/log/*.log 2>/dev/null; exit 1"], "command": ["sh", "-c", "for i in $(seq 1 180); do pg_isready -h 127.0.0.1 -U postgres >/dev/null 2>&1 && exit 0; sleep 1; done; echo 'the store did not answer within 180s; its own last words follow'; pg_isready -h 127.0.0.1 -U postgres; tail -n 20 /var/lib/postgresql/data/log/*.log 2>/dev/null; exit 1"],
"verify": ["pg_isready", "-h", "127.0.0.1", "-U", "postgres"] "verify": ["pg_isready", "-h", "127.0.0.1", "-U", "postgres"]
}, },
{ {
"id": "inventory-database", "id": "inventory-database",
"type": "action", "type": "action",
"in": "postgres", "in": "mesh-store",
"command": ["sh", "-c", "psql -U postgres -c 'CREATE DATABASE inventory'"], "command": ["sh", "-c", "psql -U postgres -c 'CREATE DATABASE inventory'"],
"verify": ["sh", "-c", "psql -U postgres -lqt | cut -d'|' -f1 | grep -qw inventory"] "verify": ["sh", "-c", "psql -U postgres -lqt | cut -d'|' -f1 | grep -qw inventory"]
}, },
{ {
"id": "identity-database", "id": "identity-database",
"type": "action", "type": "action",
"in": "postgres", "in": "mesh-store",
"command": ["sh", "-c", "psql -U postgres -c 'CREATE DATABASE identity'"], "command": ["sh", "-c", "psql -U postgres -c 'CREATE DATABASE identity'"],
"verify": ["sh", "-c", "psql -U postgres -lqt | cut -d'|' -f1 | grep -qw identity"] "verify": ["sh", "-c", "psql -U postgres -lqt | cut -d'|' -f1 | grep -qw identity"]
}, },
@@ -138,44 +126,20 @@
{ {
"id": "licences-database", "id": "licences-database",
"type": "action", "type": "action",
"in": "postgres", "in": "mesh-store",
"command": ["sh", "-c", "psql -U postgres -c 'CREATE DATABASE licences'"], "command": ["sh", "-c", "psql -U postgres -c 'CREATE DATABASE licences'"],
"verify": ["sh", "-c", "psql -U postgres -lqt | cut -d'|' -f1 | grep -qw licences"] "verify": ["sh", "-c", "psql -U postgres -lqt | cut -d'|' -f1 | grep -qw licences"]
}, },
{ {
"id": "context-schemas", "id": "context-schemas",
"type": "action", "type": "action",
"command": ["docker", "run", "--rm", "--network", "container:postgres", "command": ["docker", "run", "--rm", "--network", "container:mesh-store",
"-e", "MESH_STORE_INVENTORY=postgres://postgres:bootstrap@127.0.0.1:5432/inventory?sslmode=disable", "-e", "MESH_STORE_INVENTORY=postgres://postgres:bootstrap@127.0.0.1:5432/inventory?sslmode=disable",
"-e", "MESH_STORE_IDENTITY=postgres://postgres:bootstrap@127.0.0.1:5432/identity?sslmode=disable", "-e", "MESH_STORE_IDENTITY=postgres://postgres:bootstrap@127.0.0.1:5432/identity?sslmode=disable",
"-e", "MESH_STORE_LICENCES=postgres://postgres:bootstrap@127.0.0.1:5432/licences?sslmode=disable", "-e", "MESH_STORE_LICENCES=postgres://postgres:bootstrap@127.0.0.1:5432/licences?sslmode=disable",
"192.0.2.250:5000/mesh-controller@sha256:c67db38439ff0aee242b467486765467bb95801f52175fc5727cc4e437338ace", "192.0.2.250:5000/mesh-controller@sha256:c67db38439ff0aee242b467486765467bb95801f52175fc5727cc4e437338ace",
"migrate"], "migrate"],
"verify": ["sh", "-c", "docker exec postgres psql -U postgres -d inventory -tAc \"select to_regclass('public.node')\" | grep -qx node && docker exec postgres psql -U postgres -d identity -tAc \"select to_regclass('public.signing_key')\" | grep -qx signing_key && docker exec postgres psql -U postgres -d licences -tAc \"select to_regclass('public.licence')\" | grep -qx licence"] "verify": ["sh", "-c", "docker exec mesh-store psql -U postgres -d inventory -tAc \"select to_regclass('public.node')\" | grep -qx node && docker exec mesh-store psql -U postgres -d identity -tAc \"select to_regclass('public.signing_key')\" | grep -qx signing_key && docker exec mesh-store psql -U postgres -d licences -tAc \"select to_regclass('public.licence')\" | grep -qx licence"]
},
// **The bus is raised where the `nats` module keeps it** (novox/hq ADR 0333 §1): the container
// `mesh-broker-nats`, its configuration and user list in /var/lib/nats-module/conf, its streams in
// /var/lib/mesh-broker-nats and its certificate in /var/lib/mesh-broker-tls — the names and paths the
// catalogue's module declares. So adopting it as that module at the broker step keeps it: the
// module's built image takes the place of the upstream one below, once, on the same streams, users
// and certificate, and nothing else moves. mesh-host's internal/bootstrap holds the two to each other
// against the module's own manifest (busLayout), at the broker step and in its tests.
//
// Directories on the machine, not volumes the runtime names: the module mounts these paths, and the
// control plane, once it runs as the mesh's process, reads the certificate at its path on the machine.
// The certificate's directory is readable by everyone because the certificate is public and the control
// plane runs as nobody; its key is written 0600 by the program that makes it.
{
"id": "bus-tls-dir",
"type": "directory",
"path": "/var/lib/mesh-broker-tls",
"mode": "0755"
},
{
"id": "bus-data-dir",
"type": "directory",
"path": "/var/lib/mesh-broker-nats",
"mode": "0700"
}, },
{ {
"id": "bus-certificate", "id": "bus-certificate",
@@ -187,63 +151,49 @@
// it, and asking nothing of the image it writes into. Self-signed on purpose: a host pins // it, and asking nothing of the image it writes into. Self-signed on purpose: a host pins
// this server's exact certificate (novox/hq ADR 0004), and at this moment there is no mesh // this server's exact certificate (novox/hq ADR 0004), and at this moment there is no mesh
// to ask an authority of. // to ask an authority of.
// `--user 0:0` because the directory is root's and this image runs as nobody, which is right // `--user 0:0` because the volume is root's and this image runs as nobody, which is right
// for the long-running control plane and wrong for a one-shot writing into a fresh directory. // for the long-running control plane and wrong for a one-shot writing into a fresh volume.
"command": ["docker", "run", "--rm", "--user", "0:0", "-v", "/var/lib/mesh-broker-tls:/tls", "command": ["docker", "run", "--rm", "--user", "0:0", "-v", "mesh-broker-tls:/tls",
"192.0.2.250:5000/mesh-controller@sha256:c67db38439ff0aee242b467486765467bb95801f52175fc5727cc4e437338ace", "192.0.2.250:5000/mesh-controller@sha256:c67db38439ff0aee242b467486765467bb95801f52175fc5727cc4e437338ace",
"broker", "certificate", "--into", "/tls"], "broker", "certificate", "--into", "/tls"],
"verify": ["docker", "run", "--rm", "--user", "0:0", "-v", "/var/lib/mesh-broker-tls:/tls", "verify": ["docker", "run", "--rm", "--user", "0:0", "-v", "mesh-broker-tls:/tls",
"192.0.2.250:5000/mesh-controller@sha256:c67db38439ff0aee242b467486765467bb95801f52175fc5727cc4e437338ace", "192.0.2.250:5000/mesh-controller@sha256:c67db38439ff0aee242b467486765467bb95801f52175fc5727cc4e437338ace",
"broker", "certificate", "--check", "--into", "/tls"] "broker", "certificate", "--check", "--into", "/tls"]
}, },
// **Seeds, written only where nothing is** (create-once). The bus's configuration and user list are
// the bundle's until the bus is adopted, and the module's from then on (ADR 0333 §4): an installer
// run again after the adoption applies this bundle again, and an ordinary file would put the
// bundle's one user back over every user the mesh composed, and leave the node unable to reach its
// bus. Until the adoption the installer places the composed list over the seed (enrol.go).
{ {
"id": "bus-conf-dir", "id": "bus-conf-dir",
"type": "directory", "type": "directory",
"path": "/var/lib/nats-module/conf", "path": "/var/lib/mesh-bus-conf",
"mode": "0700" "mode": "0700"
}, },
{ {
"id": "bus-conf", "id": "bus-conf",
"type": "file", "type": "file",
"path": "/var/lib/nats-module/conf/nats.conf", "path": "/var/lib/mesh-bus-conf/nats.conf",
"mode": "0644", "mode": "0644",
"create-once": true, "content": "port: 4222\nhttp: 127.0.0.1:8222\n\ntls {\n cert_file: \"/tls/tls.crt\"\n key_file: \"/tls/tls.key\"\n}\n\njetstream {\n store_dir: \"/data\"\n}\n\ninclude accounts.conf\n"
"content": "port: 4222\nhttp: 0.0.0.0:8222\n\ntls {\n cert_file: \"/tls/tls.crt\"\n key_file: \"/tls/tls.key\"\n}\n\njetstream {\n store_dir: \"/data\"\n}\n\ninclude accounts.conf\n"
}, },
{ {
"id": "bus-accounts", "id": "bus-accounts",
"type": "file", "type": "file",
"path": "/var/lib/nats-module/conf/accounts.conf", "path": "/var/lib/mesh-bus-conf/accounts.conf",
"mode": "0600", "mode": "0600",
"create-once": true, "content": "// The first user list, carried by the installer because at genesis there is no mesh to\n// compose one. A bootstrap credential, rotated with the store's and replaced by the\n// controller's own composition from its first start onward.\naccounts {\n MESH {\n jetstream: enabled\n users = [\n { user: \"controller\", password: \"$2a$10$AHqJgOifIVbU41KmATiMhuXFs8xa7Wl2HuN4UVBCXdN2jIQzjqApy\", permissions: {\n publish: { allow: [\"$JS.ACK.CONTROL.controller.>\", \"$JS.ACK.EVENTS.controller.>\", \"$JS.ACK.DEAD_LETTER_NOTICES.controller.>\", \"$JS.API.>\", \"$KV.mesh-controller_asked.>\", \"$KV.mesh-controller_calls.>\", \"$KV.mesh-controller_hand-acts.>\", \"$KV.mesh-controller_conditions.>\", \"$KV.mesh-controller_condition-history.>\", \"$KV.mesh-controller_lease.>\", \"$KV.SEAT_MESH_BUILD_MACHINE_cancelled.>\", \"$KV.SEAT_NODE_BUILD_AGENT_cancelled.>\", \"_INBOX.enrol.>\", \"mesh.assignment.>\", \"mesh.mod.*.tool.>\", \"mesh.node.>\", \"mesh.seat.mesh-build-machine.accept.>\", \"mesh.seat.node-build-agent.accept.>\", \"mesh.seat.mesh-build-machine.tool.>\", \"mesh.seat.node-build-agent.tool.>\", \"mesh.seat.mesh-controller.event.applied\", \"mesh.seat.mesh-controller.event.built-before\", \"mesh.seat.mesh-controller.event.refused\", \"mesh.seat.mesh-controller.event.rolled-back\", \"mesh.seat.mesh-controller.event.checked\", \"mesh.seat.mesh-controller.event.plan-moved\", \"mesh.seat.mesh-controller.event.condition-raised\", \"mesh.seat.mesh-controller.event.condition-changed\", \"mesh.seat.mesh-controller.event.condition-cleared\", \"mesh.seat.mesh-controller.event.doctor-heartbeat\", \"mesh.seat.mesh-controller.event.secret-replaced\", \"mesh.seat.mesh-controller.event.healer-acted\", \"$SRV.INFO\", \"mesh.seat.node-intrusion-prevention.tool.banned.*\", \"mesh.seat.node-backup.tool.backed-up.*\", \"mesh.seat.node-backup.tool.now.*\", \"mesh.seat.mesh-delivery.tool.stalled\", \"mesh.seat.mesh-delivery.tool.close\", \"mesh.seat.node-launcher.tool.secret.*\", \"mesh.seat.mesh-delivery.tool.release\", \"mesh.seat.mesh-delivery.tool.stop\", \"mesh.seat.mesh-delivery.tool.times\", \"mesh.seat.mesh-delivery.tool.groups\", \"mesh.seat.node-service-manager.tool.restart.*\", \"mesh.seat.mesh-controller.tool.plans\", \"mesh.events.dead.>\", \"mesh.again.>\"] }\n subscribe: { allow: [\"$JS.API.>\", \"$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.>\", \"$JS.EVENT.ADVISORY.CONSUMER.DELETED.>\", \"_DELIVER.controller\", \"_DELIVER.controller.>\", \"_INBOX.controller.>\", \"mesh.control.>\", \"mesh.mod.*.event.provisioner.failing\", \"mesh.mod.*.event.provisioner.recovered\", \"mesh.mod.*.event.provisioner.retirement\", \"mesh.mod.gitea.event.pull.merged\", \"mesh.mod.gitea.event.pull.updated\", \"mesh.mod.mesh-catalog.event.catching-up\", \"mesh.mod.mesh-catalog.event.upgraded\", \"mesh.seat.mesh-build-machine.event.built\", \"mesh.seat.node-build-agent.event.built\", \"mesh.seat.operator-channel.event.decided.mesh-controller\", \"mesh.seat.mesh-controller.tool.>\", \"$SRV.PING\", \"$SRV.INFO\", \"$SRV.PING.mesh-controller\", \"$SRV.PING.mesh-controller.>\", \"$SRV.INFO.mesh-controller\", \"$SRV.INFO.mesh-controller.>\", \"$SRV.STATS\", \"$SRV.STATS.mesh-controller\", \"$SRV.STATS.mesh-controller.>\"] }\n allow_responses: { max: 1, ttl: \"1m\" }\n } }\n ]\n }\n}\n"
"content": "// The first user list, carried by the installer because at genesis there is no mesh to\n// compose one. A bootstrap credential, rotated with the store's and replaced by the\n// controller's own composition from its first start onward.\naccounts {\n MESH {\n jetstream: enabled\n users = [\n { user: \"controller\", password: \"$2a$10$AHqJgOifIVbU41KmATiMhuXFs8xa7Wl2HuN4UVBCXdN2jIQzjqApy\", permissions: {\n publish: { allow: [\"$JS.ACK.CONTROL.controller.>\", \"$JS.ACK.EVENTS.controller.>\", \"$JS.ACK.DEAD_LETTER_NOTICES.controller.>\", \"$JS.API.>\", \"$KV.mesh-controller_asked.>\", \"$KV.mesh-controller_assignments.>\", \"$KV.mesh-controller_calls.>\", \"$KV.mesh-controller_hand-acts.>\", \"$KV.mesh-controller_conditions.>\", \"$KV.mesh-controller_condition-history.>\", \"$KV.mesh-controller_declared-config.>\", \"$KV.mesh-controller_lease.>\", \"$KV.SEAT_MESH_BUILD_MACHINE_cancelled.>\", \"$KV.SEAT_NODE_BUILD_AGENT_cancelled.>\", \"_INBOX.enrol.>\", \"mesh.assignment.>\", \"mesh.mod.*.tool.>\", \"mesh.node.>\", \"mesh.seat.mesh-build-machine.accept.>\", \"mesh.seat.node-build-agent.accept.>\", \"mesh.seat.mesh-build-machine.tool.>\", \"mesh.seat.node-build-agent.tool.>\", \"mesh.seat.mesh-controller.event.applied\", \"mesh.seat.mesh-controller.event.built-before\", \"mesh.seat.mesh-controller.event.refused\", \"mesh.seat.mesh-controller.event.rolled-back\", \"mesh.seat.mesh-controller.event.checked\", \"mesh.seat.mesh-controller.event.plan-moved\", \"mesh.seat.mesh-controller.event.assignment-changed\", \"mesh.seat.mesh-controller.event.declared-config-changed\", \"mesh.seat.mesh-controller.event.condition-raised\", \"mesh.seat.mesh-controller.event.condition-changed\", \"mesh.seat.mesh-controller.event.condition-cleared\", \"mesh.seat.mesh-controller.event.doctor-heartbeat\", \"mesh.seat.mesh-controller.event.secret-replaced\", \"mesh.seat.mesh-controller.event.healer-acted\", \"$SRV.INFO\", \"mesh.seat.node-intrusion-prevention.tool.banned.*\", \"mesh.seat.node-backup.tool.backed-up.*\", \"mesh.seat.node-backup.tool.now.*\", \"mesh.seat.mesh-delivery.tool.stalled\", \"mesh.seat.mesh-delivery.tool.close\", \"mesh.seat.node-launcher.tool.secret.*\", \"mesh.seat.mesh-delivery.tool.release\", \"mesh.seat.mesh-delivery.tool.stop\", \"mesh.seat.mesh-delivery.tool.times\", \"mesh.seat.mesh-delivery.tool.groups\", \"mesh.seat.node-service-manager.tool.restart.*\", \"mesh.seat.mesh-controller.tool.plans\", \"mesh.events.dead.>\", \"mesh.again.>\"] }\n subscribe: { allow: [\"$JS.API.>\", \"$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.>\", \"$JS.EVENT.ADVISORY.CONSUMER.DELETED.>\", \"_DELIVER.controller\", \"_DELIVER.controller.>\", \"_INBOX.controller.>\", \"mesh.control.>\", \"mesh.mod.*.event.provisioner.failing\", \"mesh.mod.*.event.provisioner.recovered\", \"mesh.mod.*.event.provisioner.retirement\", \"mesh.mod.gitea.event.pull.merged\", \"mesh.mod.gitea.event.pull.updated\", \"mesh.mod.mesh-catalog.event.catching-up\", \"mesh.mod.mesh-catalog.event.upgraded\", \"mesh.seat.mesh-build-machine.event.built\", \"mesh.seat.node-build-agent.event.built\", \"mesh.seat.operator-channel.event.decided.mesh-controller\", \"mesh.seat.mesh-controller.tool.>\", \"$SRV.PING\", \"$SRV.INFO\", \"$SRV.PING.mesh-controller\", \"$SRV.PING.mesh-controller.>\", \"$SRV.INFO.mesh-controller\", \"$SRV.INFO.mesh-controller.>\", \"$SRV.STATS\", \"$SRV.STATS.mesh-controller\", \"$SRV.STATS.mesh-controller.>\"] }\n allow_responses: { max: 1, ttl: \"1m\" }\n } }\n ]\n }\n}\n"
}, },
// The upstream server, until the broker step adopts it: the `nats` module's own image, built by the
// mesh, takes its place there once, on these same directories (ADR 0333 §3), as the bus's planned
// step (ADR 0236). The release is the one the module's image is built on (its NATS_BASE, 2.11.17),
// pinned upstream by that same digest: the adoption changes no server release, so it can be undone
// by putting the old image back and is said reversible. The published port is the
// node's bus port, 5671 unless given, which every enrolment token names; the installer gives the
// module the setting that maps its declared 4222 to it (ADR 0333 §2).
{ {
"id": "broker", "id": "broker",
"type": "container", "type": "container",
"name": "mesh-broker-nats", "name": "mesh-broker",
"image": "nats@sha256:e4bf19f15fd3218814a4e3c9e0064e1334bd8aa20d5984b9f1a0afd084f8cc00", "image": "192.0.2.250:5000/nats@sha256:b83efabe3e7def1e0a4a31ec6e078999bb17c80363f881df35edc70fcb6bb927",
"ports": ["5671:4222", "127.0.0.1:8222:8222"], "ports": ["5671:4222", "127.0.0.1:8222:8222"],
"volumes": ["/var/lib/mesh-broker-nats:/data", "/var/lib/nats-module/conf:/etc/nats:ro", "/var/lib/mesh-broker-tls:/tls:ro"], "volumes": ["mesh-broker-data:/data", "mesh-broker-tls:/tls:ro", "/var/lib/mesh-bus-conf:/etc/nats:ro"],
"args": ["-c", "/etc/nats/nats.conf", "-js"] "args": ["-c", "/etc/nats/nats.conf", "-js"]
}, },
{ {
"id": "broker-ready", "id": "broker-ready",
"type": "action", "type": "action",
"command": ["sh", "-c", "for i in $(seq 1 60); do docker run --rm --network host --entrypoint sh nats@sha256:e4bf19f15fd3218814a4e3c9e0064e1334bd8aa20d5984b9f1a0afd084f8cc00 -c 'nc -z 127.0.0.1 5671' >/dev/null 2>&1 && exit 0; sleep 1; done; exit 1"], "command": ["sh", "-c", "for i in $(seq 1 60); do docker run --rm --network host --entrypoint sh 192.0.2.250:5000/nats@sha256:b83efabe3e7def1e0a4a31ec6e078999bb17c80363f881df35edc70fcb6bb927 -c 'nc -z 127.0.0.1 5671' >/dev/null 2>&1 && exit 0; sleep 1; done; exit 1"],
"verify": ["sh", "-c", "docker run --rm --network host --entrypoint sh nats@sha256:e4bf19f15fd3218814a4e3c9e0064e1334bd8aa20d5984b9f1a0afd084f8cc00 -c 'nc -z 127.0.0.1 5671'"] "verify": ["sh", "-c", "docker run --rm --network host --entrypoint sh 192.0.2.250:5000/nats@sha256:b83efabe3e7def1e0a4a31ec6e078999bb17c80363f881df35edc70fcb6bb927 -c 'nc -z 127.0.0.1 5671'"]
}, },
{ {
"id": "control-plane", "id": "control-plane",
@@ -252,7 +202,7 @@
"image": "192.0.2.250:5000/mesh-controller@sha256:c67db38439ff0aee242b467486765467bb95801f52175fc5727cc4e437338ace", "image": "192.0.2.250:5000/mesh-controller@sha256:c67db38439ff0aee242b467486765467bb95801f52175fc5727cc4e437338ace",
"network": "host", "network": "host",
"args": ["serve"], "args": ["serve"],
"volumes": ["/var/lib/mesh-broker-tls:/broker-tls:ro"], "volumes": ["mesh-broker-tls:/broker-tls:ro"],
"env": { "env": {
"MESH_STORE_INVENTORY": "postgres://postgres:bootstrap@127.0.0.1:5432/inventory?sslmode=disable", "MESH_STORE_INVENTORY": "postgres://postgres:bootstrap@127.0.0.1:5432/inventory?sslmode=disable",
"MESH_STORE_IDENTITY": "postgres://postgres:bootstrap@127.0.0.1:5432/identity?sslmode=disable", "MESH_STORE_IDENTITY": "postgres://postgres:bootstrap@127.0.0.1:5432/identity?sslmode=disable",
+9 -1
View File
@@ -49,7 +49,6 @@
./internal/inventory TestACredentialIssuedOutsideIsNeverMade ./internal/inventory TestACredentialIssuedOutsideIsNeverMade
./internal/inventory TestOneAcceptForAProviderIsSealedToEveryConsumer ./internal/inventory TestOneAcceptForAProviderIsSealedToEveryConsumer
./internal/inventory TestASharedOwnSecretIssuedOutsideIsNeverMade ./internal/inventory TestASharedOwnSecretIssuedOutsideIsNeverMade
./internal/recordmirror TestNatsAnAssignmentAndARegistrationEachChangeOneKeyAndRaiseOneEvent
./cmd/mesh-controller TestOneAcceptForAProviderIsPlacedForEveryConsumer ./cmd/mesh-controller TestOneAcceptForAProviderIsPlacedForEveryConsumer
./cmd/mesh-controller TestReplay164 ./cmd/mesh-controller TestReplay164
./cmd/mesh-controller TestAnApprovedLayerOpensTheWalkThatSendsIt ./cmd/mesh-controller TestAnApprovedLayerOpensTheWalkThatSendsIt
@@ -62,3 +61,12 @@
./cmd/mesh-controller TestAProviderElsewhereServesItsOwnComposedName ./cmd/mesh-controller TestAProviderElsewhereServesItsOwnComposedName
./cmd/mesh-controller TestAReusedCheckBuildIsRegisteredByItsDigestAndGatedAsABuiltOne ./cmd/mesh-controller TestAReusedCheckBuildIsRegisteredByItsDigestAndGatedAsABuiltOne
./internal/inventory TestAReportOfABusyContainerIsKeptAndReadBack ./internal/inventory TestAReportOfABusyContainerIsKeptAndReadBack
./cmd/mesh-controller TestAReleaseSendsTheRestAtOnceAfterTheFirstPasses
./cmd/mesh-controller TestARestMachineThatFailsIsPutBackAloneAndTheOthersAreJudgedOn
./cmd/mesh-controller TestAWalkTakesOverTheReadingsOfAGateJudgingTheSameBuildThere
./cmd/mesh-controller TestAFirstSendMadeAndNotKeptIsNotSentAgain
./cmd/mesh-controller TestAReleaseSendKeptAndMadeIsNotMadeAgain
./cmd/mesh-controller TestAFirstMachineThatFailsStillStopsTheWalk
./cmd/mesh-controller TestTheControlNodeIsSentLastInAStep
./cmd/mesh-controller TestARestMachineSilentPastItsBoundIsLeftAndSaid
./cmd/mesh-controller TestANoMoveTierWaitsForNoNewReport