Compare commits

..
Author SHA1 Message Date
jochen 3647928668 Judge an upstream pin's host like any name, and harden the upstream unpack (review 35, hq issue 345)
mesh/delivery-group group fix/345-reminder-test rejected: a member's own check failed
mesh/merge-gate pass: builds build-agent, mesh-controller → ace, g14, novox, shanks; no bus step; every machine composes with the change as it did without …
mesh/repo-check fail: its merge-check.sh failed: --- FAIL: TestTheInstallersFirstUserListIsWhatTheControllerWouldCompose (0.64s)
mesh/delivery stopped: the pull request closed unmerged
The url was skipped by the installation check, so an installation's host or
an address passed; it is judged whole now, the vendor's host declared in the
artifact's names-on-purpose, and an address or a reserved name as the host
is refused. Only the version's dotted-quad reading is skipped. The unpack
caps its entries, drains xz before waiting for it, reads a pax global
header and a nested within's parents, and a redirect off https is not
followed.
2026-10-12 13:25:18 +02:00
jochen 9d108feab8 Let an archive be an upstream release pinned by URL and sha256, mirrored by the build seat (hq issue 345, ADR 0316)
mesh/merge-gate pass: builds build-agent, mesh-controller → ace, g14, novox, shanks; no bus step; every machine composes with the change as it did without …
mesh/repo-check pass: its merge-check.sh passed
mesh/delivery-group group fix/345-reminder-test rejected: a member's own check failed
mesh/delivery superseded: a newer head of the same pull request
A mail client's release is about 80 MB per version, too large to keep in a
repository at every upgrade, and an add-on whose licence forbids copying it
to third parties may not be in a repository at all. An archive artifact may
now name url, sha256 and version (and file or within) instead of from: the
build seat fetches it once, refuses it when its digest is not the pin's and
names both, keeps it in its workspace under the digest (a vendor that serves
only its newest version cannot break a later build of the same pin), and
packs it like any archive, so a node gets the mesh's own archive by digest
and never asks the vendor. The build seat's image carries xz, which Go's
standard library does not read. The installation check takes the vendor's
address and version of such a pin for the world's, as a registry is an
image's.
2026-10-12 12:57:30 +02:00
50 changed files with 1141 additions and 3279 deletions
+3 -2
View File
@@ -16,7 +16,8 @@ RUN CGO_ENABLED=0 go build -trimpath -ldflags '-s -w' -o /mesh-builder ./cmd/mes
FROM ${ALPINE_BASE}
# git to clone what it is asked to build, and the docker client to build and push it. The daemon
# is the machine's, reached through its socket — a build machine shares the runtime it was given
# rather than running one inside itself.
RUN apk add --no-cache git docker-cli
# rather than running one inside itself. xz to read an upstream release an archive is pinned to
# (novox/hq ADR 0316): Go's standard library reads gzip and bzip2 and not xz.
RUN apk add --no-cache git docker-cli xz
COPY --from=build /mesh-builder /usr/local/bin/mesh-builder
ENTRYPOINT ["/usr/local/bin/mesh-builder"]
-9
View File
@@ -213,8 +213,6 @@ type asker struct {
saidNoRouter bool
saidNoGrant bool
// saidGrantCaveat is whether a grant held with a caveat was said, so it is said once.
saidGrantCaveat bool
mu sync.Mutex
nudged chan struct{}
@@ -319,13 +317,6 @@ func (a *asker) reconcile(ctx context.Context) error {
}
return a.sayUnasked(ctx, unasked, "the bus does not hold the controller's grant to ask yet: "+why)
}
if why != "" && !a.saidGrantCaveat {
// Held, with a caveat worth one line (an older bus module cannot confirm the reload).
a.logf("the controller asks: %s", why)
a.saidGrantCaveat = true
} else if why == "" {
a.saidGrantCaveat = false
}
a.saidNoGrant = false
}
channels := ""
-205
View File
@@ -12,7 +12,6 @@ import (
"git.novox.be/novox/mesh-sdk/go/asks"
"github.com/novox/mesh-controller/internal/broker"
"github.com/novox/mesh-controller/internal/conditions"
"github.com/novox/mesh-controller/internal/link"
)
@@ -809,207 +808,3 @@ func TestARefusalIsAskedAgainAndTheUndeliveredConditionClearsOnceTaken(t *testin
t.Errorf("the explanation says its verdict twice: %q", e)
}
}
// hq issue 353, the review: the bus holds the controller's grant to ask only when both facts hold — the mesh
// composes the controller's own grant, and the bus's module says it is in force on the running server — and
// each fact not known is not held, said with why.
func TestTheGrantIsHeldOnlyWhenComposedAndInForceOnTheBus(t *testing.T) {
yes := func(context.Context) (bool, string, error) { return true, "", nil }
no := func(why string) grantJudge {
return func(context.Context) (bool, string, error) { return false, why, nil }
}
broken := func(context.Context) (bool, string, error) { return false, "", errors.New("the store is away") }
for _, c := range []struct {
name string
composed, inForce grantJudge
held bool
says string
}{
{"composed and in force", yes, yes, true, ""},
{"not composed", no("no router is assigned"), yes, false, "no router is assigned"},
{"composed, written, not reloaded", yes, no("the server has not reloaded"), false, "has not reloaded"},
{"the composition unknown", broken, yes, false, "could not be worked out: the store is away"},
{"the bus unknown", yes, broken, false, "could not be worked out: the store is away"},
{"nothing judges the bus", yes, nil, false, "not judged"},
} {
held, why, err := grantHeldBy(c.composed, c.inForce, time.Now)(context.Background())
if err != nil || held != c.held || !strings.Contains(why, c.says) {
t.Errorf("%s: held %v (%q, %v)", c.name, held, why, err)
}
}
// Remembered for grantLookEvery, then judged again.
now := time.Date(2026, 10, 9, 18, 0, 0, 0, time.UTC)
inForce := false
judge := grantHeldBy(yes, func(context.Context) (bool, string, error) { return inForce, "not yet", nil },
func() time.Time { return now })
if held, _, _ := judge(context.Background()); held {
t.Fatal("held before the bus enforces it")
}
inForce = true
if held, _, _ := judge(context.Background()); held {
t.Error("judged again inside grantLookEvery")
}
now = now.Add(grantLookEvery)
if held, _, _ := judge(context.Background()); !held {
t.Error("not judged again after grantLookEvery")
}
}
// The controller's own grant, as composed: present while a router that takes asks is assigned, absent
// otherwise — whatever else in the user list changes.
func TestTheControllersOwnGrantToAskIsReadFromTheComposition(t *testing.T) {
router := broker.Declared{Module: "messenger", Holds: []broker.Seat{{Name: broker.AsksSeat, Scope: "mesh",
Accepts: []string{"ask", "cancel"}, Emits: []string{"decided"}, ByCaller: []string{"ask", "cancel", "decided"}}}}
other := broker.Declared{Module: "dunst"}
with := broker.Records{Nodes: []string{"anchor"}, Assigned: map[string][]broker.Declared{"anchor": {router, other}}}
without := broker.Records{Nodes: []string{"anchor"}, Assigned: map[string][]broker.Declared{"anchor": {other}}}
if may, err := controllerMayAsk(with); err != nil || !may {
t.Errorf("with a router assigned: %v %v", may, err)
}
if may, err := controllerMayAsk(without); err != nil || may {
t.Errorf("with no router assigned: %v %v", may, err)
}
}
// The bus module's answer, with the record's word on the list sent as the check only where the answer does
// not settle it (reviews 22 and 31 of PR 188); each way it is not held names its own remedy.
func TestTheBusModulesAnswerNamesTheRemedyForEachCase(t *testing.T) {
const (
reloaded = `{"allowed":true,"in_force":true,"file_written":"2026-10-09T15:55:56Z","server_loaded":"2026-10-09T15:56:05Z"}`
notLoaded = `{"allowed":true,"in_force":false,"why":"the user list on the machine allows it, and the server has not reloaded since it was written","file_written":"2026-10-09T15:55:56Z","server_loaded":"2026-10-09T15:50:00Z"}`
noVarz = `{"allowed":true,"in_force":false,"why":"the server's last configuration load could not be read: refused"}`
noStat = `{"allowed":true,"in_force":false,"why":"when the bus's user list was written could not be read inside mesh-broker-nats: exit 1"}`
notAllowed = `{"allowed":false,"in_force":false,"why":"the user list on the machine does not allow it"}`
)
current := busListState{Machine: "anchor"}
behind := busListState{Machine: "anchor", Behind: true}
stepped := busListState{Machine: "anchor", Behind: true, BusStep: "sending anchor would replace the bus; `bus upgrade --why …`"}
uncomposed := busListState{Machine: "anchor", Behind: true, Uncomposed: errors.New("module foo requires a provision nobody provides")}
for _, c := range []struct {
name, answer string
askErr error
list busListState
listErr error
held bool
says, not string
}{
{"in force", reloaded, nil, behind, nil, true, "", "push"},
{"written, not reloaded", notLoaded, nil, current, nil, false, "within about 5 s", "push"},
{"in force not known: varz", noVarz, nil, current, nil, false, "is not known: the server's last configuration load", "wait"},
{"in force not known: stat", noStat, nil, current, nil, false, "is not known: when the bus's user list was written", "wait"},
{"not allowed, list behind", notAllowed, nil, behind, nil, false, "`push anchor` carries it", "5 s"},
{"not allowed, bus step pending", notAllowed, nil, stepped, nil, false, "`bus upgrade", "`push anchor`"},
{"not allowed, list uncomposed", notAllowed, nil, uncomposed, nil, false, "`plan anchor`", "`push anchor`"},
{"older module, allowed, list current", `{"allowed":true}`, nil, current, nil, true, "not confirmed", "push"},
{"older module, allowed, list behind", `{"allowed":true}`, nil, behind, nil, true, "not confirmed", "push"},
{"older module, allowed, list unknown", `{"allowed":true}`, nil, current, errors.New("the store is away"), true, "`bus upgrade`", "push"},
{"older module, not allowed, list behind", `{"allowed":false}`, nil, behind, nil, false, "`push anchor` carries it", "5 s"},
{"module cannot answer, list current", "", errors.New("no responders"), current, nil, true, "could not answer", "push"},
{"module cannot answer, list behind", "", errors.New("no responders"), behind, nil, false, "`push anchor` carries it", ""},
{"module cannot answer, list unknown", "", errors.New("no responders"), current, errors.New("the store is away"), false, "could not be worked out", ""},
{"answer unreadable, list behind", `not json`, nil, behind, nil, false, "could not be read", ""},
} {
judge := busGrantBy(
func(context.Context) (string, json.RawMessage, error) {
return "anchor", json.RawMessage(c.answer), c.askErr
},
func(context.Context) (busListState, error) { return c.list, c.listErr })
held, why, _ := judgeGrantOnce(judge)
if held != c.held || !strings.Contains(why, c.says) || (c.not != "" && strings.Contains(why, c.not)) {
t.Errorf("%s: held %v (%q)", c.name, held, why)
}
}
none := busGrantBy(func(context.Context) (string, json.RawMessage, error) { return "", nil, nil }, nil)
if held, why, _ := judgeGrantOnce(none); held || !strings.Contains(why, "no module holds the bus") {
t.Errorf("no bus: held %v (%q)", held, why)
}
}
// judgeGrantOnce is the whole judgement with the composed grant held, as the asker makes it.
func judgeGrantOnce(bus grantJudge) (bool, string, error) {
yes := func(context.Context) (bool, string, error) { return true, "", nil }
return grantHeldBy(yes, bus, time.Now)(context.Background())
}
// The orders reviews 22 and 31 of PR 188 found: this controller live before the bus module that answers
// in_force, with the bus's machine behind on its list — and, after mesh-catalog #166 merged, unpushable until a
// person's bus upgrade. An older module that allows the ask asks (once, with the caveat said); a new module that
// has not reloaded names the wait, not a push; one that cannot say whether it reloaded names what is not known.
func TestAnOlderBusModuleNeitherSilencesTheAsksNorNamesTheWrongRemedy(t *testing.T) {
for _, c := range []struct {
name, answer string
list busListState
asks bool
says, not string
}{
{"older module, list current", `{"allowed":true}`, busListState{Machine: "anchor"}, true, "not confirmed", ""},
{"older module, list behind, bus step pending", `{"allowed":true}`,
busListState{Machine: "anchor", Behind: true, BusStep: "a planned step"}, true, "not confirmed", ""},
{"new module, not reloaded",
`{"allowed":true,"in_force":false,"why":"not reloaded","file_written":"2026-10-09T15:55:56Z","server_loaded":"2026-10-09T15:50:00Z"}`,
busListState{Machine: "anchor"}, false, "within about 5 s", "push"},
{"new module, reload not known", `{"allowed":true,"in_force":false,"why":"the server's last configuration load could not be read: refused"}`,
busListState{Machine: "anchor"}, false, "is not known", "wait"},
} {
t.Run(c.name, func(t *testing.T) {
r := newAskerRig(t)
var said []string
r.a.logf = func(f string, a ...any) { said = append(said, fmt.Sprintf(f, a...)) }
var raised [][]conditions.Observation
r.a.raise = func(_ context.Context, obs []conditions.Observation) error {
raised = append(raised, obs)
return nil
}
yes := func(context.Context) (bool, string, error) { return true, "", nil }
r.a.grantHeld = grantHeldBy(yes, busGrantBy(
func(context.Context) (string, json.RawMessage, error) {
return "anchor", json.RawMessage(c.answer), nil
},
func(context.Context) (busListState, error) { return c.list, nil }),
func() time.Time { return r.now })
r.open = []conditions.Condition{heldCondition()}
_ = r.a.reconcile(context.Background())
_ = r.a.reconcile(context.Background())
sent := r.asksSent(t)
if c.asks {
if len(sent) != 1 || sent[0].About != heldCondition().Key {
t.Fatalf("not asked: %+v", sent)
}
n := 0
for _, s := range said {
if strings.Contains(s, c.says) {
n++
}
}
if n != 1 {
t.Errorf("the caveat said %d times: %q", n, said)
}
return
}
if len(sent) != 0 {
t.Fatalf("asked while the bus does not hold the grant: %+v", sent)
}
if len(raised) == 0 || len(raised[len(raised)-1]) != 1 ||
raised[len(raised)-1][0].Kind != "asks-undelivered" {
t.Fatalf("not said as undelivered: %+v", raised)
}
summary := raised[len(raised)-1][0].Summary
if !strings.Contains(summary, c.says) || (c.not != "" && strings.Contains(summary, c.not)) {
t.Errorf("the condition names the wrong remedy: %q", summary)
}
})
}
}
// The serving asker is wired to the real judgement: with records and a bus it cannot read, the grant is not
// held, and so nothing is asked (a stub saying yes, or no judgement at all, would ask).
func TestTheServingAskerJudgesTheGrantForReal(t *testing.T) {
a := servingAsker(&stores{}, nil, nil, nil)
if a.grantHeld == nil {
t.Fatal("the serving asker does not judge the bus's grant")
}
held, why, err := a.grantHeld(context.Background())
if err != nil || held || !strings.Contains(why, "could not be worked out") {
t.Errorf("held %v (%q, %v) with nothing to read", held, why, err)
}
}
+21 -271
View File
@@ -235,27 +235,12 @@ func routerHereIn(inv *inventory.Inventory) func(ctx context.Context) (bool, err
}
}
// grantHeldIn says whether the bus holds the controller's grant to ask (novox/hq issue 353), judged on the two
// facts that make it so, and failing closed on each:
//
// 1. **composed**: the user list the mesh composes now grants the controller itself to publish its ask
// (controllerMayAsk) — the controller's own grant, not whether any user of the list changed;
// 2. **in force**: the bus's own module, on the machine holding the bus, says the user list there allows it
// and the server reloaded after that list was written (the nats module's `nats_user_can`, `in_force`) — a
// list sent is not a list the bus enforces until it reloads.
//
// Anything that cannot be worked out is not held, with why, so the conditions that need the operator are said
// as undelivered rather than asked into a refusal.
func grantHeldIn(open *stores, conn *nats.Conn) func(ctx context.Context) (bool, string, error) {
return grantHeldBy(composedGrantIn(open), busGrantIn(open, conn), time.Now)
}
// grantJudge is one of the two facts: whether it holds, and why not.
type grantJudge func(ctx context.Context) (bool, string, error)
// grantHeldBy is the judgement over the two facts, remembered for grantLookEvery: composing the list resolves
// the bus's machine whole, and asking the bus's module is a call.
func grantHeldBy(composed, inForce grantJudge, now func() time.Time) func(ctx context.Context) (bool, string, error) {
// grantHeldIn says whether the bus holds the controller's grant to ask (novox/hq issue 353): the user list the
// machine holding the bus was last sent is the one the mesh composes now (brokerBehind, the same judgement a
// push makes to send that machine first). While it is behind, the controller's ask is refused by the bus,
// whatever the record says of the router, so nothing is asked and the operator is told to push that machine.
// Judged at most every grantLookEvery: composing the list resolves the bus's machine whole.
func grantHeldIn(open *stores) func(ctx context.Context) (bool, string, error) {
var mu sync.Mutex
var at time.Time
var held bool
@@ -263,251 +248,22 @@ func grantHeldBy(composed, inForce grantJudge, now func() time.Time) func(ctx co
return func(ctx context.Context) (bool, string, error) {
mu.Lock()
defer mu.Unlock()
if !at.IsZero() && now().Sub(at) < grantLookEvery {
if !at.IsZero() && time.Since(at) < grantLookEvery {
return held, why, nil
}
held, why = judgeGrant(ctx, composed, inForce)
at = now()
return held, why, nil
}
}
// judgeGrant is held only when both facts hold; an error is a fact not known, never a yes. A fact that holds
// with a caveat (a reload not confirmed) gives that caveat as the why of a yes.
func judgeGrant(ctx context.Context, composed, inForce grantJudge) (bool, string) {
var caveat []string
for _, fact := range []struct {
judge grantJudge
what string
}{{composed, "whether the mesh composes the controller a grant to ask"}, {inForce, "whether the bus enforces it"}} {
if fact.judge == nil {
return false, fact.what + " is not judged here"
}
ok, why, err := fact.judge(ctx)
if err != nil {
return false, fact.what + " could not be worked out: " + err.Error()
}
if !ok {
return false, why
}
if why != "" {
caveat = append(caveat, why)
}
}
return true, strings.Join(caveat, "; ")
}
// controllerMayAsk says whether the controller's own grant, as composed from these records, lets it publish its
// ask (ADR 0259 §3: composed while a router that takes asks under its asker's name is assigned).
func controllerMayAsk(records broker.Records) (bool, error) {
users, err := broker.Users(records)
if err != nil {
return false, err
}
for _, u := range users {
if u.Kind != broker.KindController {
continue
}
perms, err := broker.PermissionsFor(u)
if err != nil {
return false, err
}
return broker.MayPublish(perms, asks.AskSubject(askerName)), nil
}
return false, errors.New("the composition holds no controller")
}
// composedGrantIn is the first fact, from the mesh's records.
func composedGrantIn(open *stores) grantJudge {
return func(ctx context.Context) (bool, string, error) {
if open == nil || open.inventory == nil {
return false, "", errors.New("the mesh's records are not open")
}
records, err := open.inventory.BusRecords(ctx)
machine, behind, err := brokerBehind(ctx, open, nil)
if err != nil {
return false, "", err
}
may, err := controllerMayAsk(records)
if err != nil || may {
return may, "", err
at, held, why = time.Now(), !behind, ""
if behind {
why = fmt.Sprintf("the bus's user list on %s is behind what the mesh composes, so the bus has not been "+
"given the controller's grant to ask; `push %s` carries it", machine, machine)
}
return false, "the mesh composes the controller no grant to ask: no router that takes asks under its " +
"asker's name is assigned", nil
return held, why, nil
}
}
// busGrantIn is the second fact, asked of the bus's own module on the machine holding the bus, with the check
// main made before it (novox/hq issue 353, PR 186: is the user list sent to the bus's machine the one composed
// now) kept for the cases the module's word does not settle.
func busGrantIn(open *stores, conn *nats.Conn) grantJudge {
ask := func(ctx context.Context) (string, json.RawMessage, error) {
if open == nil || open.inventory == nil || conn == nil {
return "", nil, errors.New("the mesh's records or the bus are not open")
}
holders, err := seatHolders(ctx, open.inventory)
if err != nil {
return "", nil, err
}
h, held := holders[theBrokerSeat]
if !held || h.Node == "" || h.Module == "" {
return "", nil, nil
}
a, err := link.AskModuleToolOn(ctx, conn, h.Module, busUserCanTool, h.Node, map[string]any{
"user": broker.ControllerName, "action": "publish", "subject": asks.AskSubject(askerName)}, 15*time.Second)
if err != nil {
return h.Node, nil, err
}
if a.Error != "" {
return h.Node, nil, errors.New(a.Error)
}
return h.Node, a.Result, nil
}
list := func(ctx context.Context) (busListState, error) {
if open == nil || open.inventory == nil {
return busListState{}, errors.New("the mesh's records are not open")
}
s, err := brokerListState(ctx, open, nil)
if err != nil || s.Machine == "" || !s.Behind {
return s, err
}
step, err := busHeld(ctx, open.inventory, []string{s.Machine})
if err != nil {
return busListState{}, err
}
s.BusStep = step[s.Machine]
return s, nil
}
return busGrantBy(ask, list)
}
// busListState is the user list on the bus's machine as the controller's record knows it: whether the list
// sent there is behind the one composed now, why it could not be composed, and the bus step a send there waits
// on (that machine cannot be pushed until a person's `bus upgrade`).
type busListState struct {
Machine string
Behind bool
Uncomposed error
BusStep string
}
// remedy is the one thing that sends the bus's machine the list composed now: the plan's own reason when it
// cannot be composed, the bus upgrade while sends there wait on it, and otherwise a push of that machine.
func (s busListState) remedy(node string) string {
if s.Machine != "" {
node = s.Machine
}
switch {
case s.Uncomposed != nil:
return fmt.Sprintf("the user list for %s cannot be composed (%v); `plan %s` says what to fix first",
node, s.Uncomposed, node)
case s.BusStep != "":
return fmt.Sprintf("sends to %s wait on the bus's planned step, so a push will not carry it: %s", node, s.BusStep)
}
return fmt.Sprintf("`push %s` carries it", node)
}
// busGrantBy judges the bus's word, and names for each way it is not held the one thing that makes it held:
//
// - **in force**: held;
// - **allowed, from a module older than `in_force`**: held, with the caveat that the reload is not confirmed.
// The file allows the publish, and the bus reloads within about 5 s of a change to it, so the worst case is
// one ask refused in those seconds, which is asked again (issue 369). The list check is not consulted:
// while a bus step is pending the bus's machine cannot be pushed, so any change to the composed list would
// read as behind for good and silence every ask until a person's bus upgrade (the review of PR 188);
// - **allowed, written, not reloaded** (the module gives both times): wait about 5 s;
// - **allowed, in force not known** (the module could not read a time): what is not known, said;
// - **not allowed, or the module cannot answer**: the list check decides — the remedy that sends the list
// when it is behind (busListState.remedy), and, for a module that cannot answer, held while the list sent
// is current, as main judged it.
//
// ask answers the machine holding the bus and its module's answer ("" when nothing holds the bus); list is the
// record's word on the list sent to that machine.
func busGrantBy(ask func(ctx context.Context) (string, json.RawMessage, error),
list func(ctx context.Context) (busListState, error)) grantJudge {
return func(ctx context.Context) (bool, string, error) {
node, raw, askErr := ask(ctx)
if askErr == nil && node == "" {
return false, "no module holds the bus, so nothing says what it enforces", nil
}
var answer busAnswer
if askErr == nil {
if answer, askErr = readBusAnswer(raw); askErr != nil {
askErr = fmt.Errorf("its answer could not be read: %w", askErr)
}
}
if askErr != nil {
if node == "" {
node = "the bus's machine"
}
s, err := list(ctx)
if err != nil {
return false, "", fmt.Errorf("the bus module on %s could not answer (%v), and whether the list sent "+
"there is the one composed now could not be worked out: %w", node, askErr, err)
}
if s.Behind {
return false, fmt.Sprintf("the bus module on %s could not answer (%v), and the user list sent there is "+
"behind what the mesh composes; %s", node, askErr, s.remedy(node)), nil
}
return true, fmt.Sprintf("the bus module on %s could not answer (%v); judged by the user list sent there, "+
"which is the one composed now", node, askErr), nil
}
notAllowed := func(older string) (bool, string, error) {
s, err := list(ctx)
if err != nil {
return false, "", fmt.Errorf("the user list on %s does not allow the controller to ask%s, and whether "+
"the list sent there is the one composed now could not be worked out: %w", node, older, err)
}
if s.Behind {
return false, fmt.Sprintf("the user list on %s does not allow the controller to ask%s: it is behind "+
"what the mesh composes; %s", node, older, s.remedy(node)), nil
}
return false, fmt.Sprintf("the user list on %s does not allow the controller to ask%s, though the list "+
"sent there is the one composed now; %s again", node, older, s.remedy(node)), nil
}
switch {
case answer.InForce == nil && answer.Allowed:
return true, fmt.Sprintf("the user list on %s allows the controller to ask, but its reload is not "+
"confirmed: the bus module there is older than the answer whether a grant is in force, and the bus "+
"upgrade (`bus upgrade`) carries one that confirms it", node), nil
case answer.InForce == nil:
return notAllowed(" (the bus module there is older than the answer whether a grant is in force)")
case !answer.Allowed:
return notAllowed("")
case *answer.InForce:
return true, "", nil
case answer.FileWritten != "" && answer.ServerLoaded != "":
return false, fmt.Sprintf("the bus on %s has the controller's grant to ask written (%s) but has not "+
"reloaded since (last load %s); it reloads in place within about 5 s, so nothing is to be done but wait",
node, answer.FileWritten, answer.ServerLoaded), nil
}
why := answer.Why
if why == "" {
why = "it gives no reason"
}
return false, fmt.Sprintf("the user list on %s allows the controller to ask, but whether the bus enforces it "+
"is not known: %s", node, why), nil
}
}
// busUserCanTool is the bus module's answer to whether a user may do something, and whether it is in force.
const busUserCanTool = "nats_user_can"
// busAnswer is the bus module's answer to busUserCanTool; InForce is nil from a module older than that field,
// and FileWritten and ServerLoaded are given only when it could read both times (mesh-catalog's nats InForce).
type busAnswer struct {
Allowed bool `json:"allowed"`
InForce *bool `json:"in_force"`
Why string `json:"why"`
FileWritten string `json:"file_written"`
ServerLoaded string `json:"server_loaded"`
}
func readBusAnswer(raw json.RawMessage) (busAnswer, error) {
var answer busAnswer
err := json.Unmarshal(raw, &answer)
return answer, err
}
// grantLookEvery is how often the bus's user list is judged against the one its machine was last sent.
const grantLookEvery = 30 * time.Second
@@ -546,19 +302,7 @@ func startAsking(ctx context.Context, open *stores, server *link.Server, conn *n
fmt.Printf("the operator cannot be asked: %v\n", err)
return
}
a := servingAsker(open, conn, js, keeper)
if err := server.Decides(a); err != nil {
fmt.Printf("the operator's answers cannot be heard, so nothing is asked: %v\n", err)
return
}
askerFrom = a
go a.keep(ctx)
}
// servingAsker is the serving controller's asker, wired to the mesh: what it reads, publishes, performs and
// records, and the judgements it asks by — the bus's grant to ask among them (novox/hq issue 353).
func servingAsker(open *stores, conn *nats.Conn, js jetstream.JetStream, keeper *conditions.Keeper) *asker {
return &asker{
a := &asker{
open: keeper.Open,
silence: func(ctx context.Context, key string, d time.Duration, by, why string) error {
_, err := keeper.Silence(ctx, key, d, by, why)
@@ -578,7 +322,7 @@ func servingAsker(open *stores, conn *nats.Conn, js jetstream.JetStream, keeper
},
routerRecord: routerRecordOf(conn, open.inventory),
routerHere: routerHereIn(open.inventory),
grantHeld: grantHeldIn(open, conn),
grantHeld: grantHeldIn(open),
channels: channelsIn(open.inventory),
raise: func(ctx context.Context, obs []conditions.Observation) error {
return keeper.Reconcile(ctx, sourceAsker, obs)
@@ -586,4 +330,10 @@ func servingAsker(open *stores, conn *nats.Conn, js jetstream.JetStream, keeper
now: time.Now,
logf: func(format string, args ...any) { fmt.Printf(format+"\n", args...) },
}
if err := server.Decides(a); err != nil {
fmt.Printf("the operator's answers cannot be heard, so nothing is asked: %v\n", err)
return
}
askerFrom = a
go a.keep(ctx)
}
+17 -232
View File
@@ -37,11 +37,6 @@ import (
// own commit, the newest first, one walk each, until one is delivered; the older ones are then answered
// by that one.
//
// **A merge waits only for a batch it shares something with** (novox/hq ADR 0304 decision 1, ADR 0306 decisions
// 2 and 3, issue 525): it joins a batch it is related to, and one that is related to none and whose delivery
// group has no member still to merge is a batch whose window is closed at once (batchToJoin). Walks stay one at
// a time: such a batch is cut when no walk is open, and queued behind the one that is otherwise.
//
// A merge on the controller's own path (a module whose walk waits for nobody's word) never shares a batch with
// one that waits for mesh-delivery's: each kind has a batch of its own, an own-path batch is cut first and folds
// no waiting catalogue walk, and a catalogue walk let go while another walk runs starts once it ended — so no
@@ -178,12 +173,6 @@ func (f following) hearMerge(ctx context.Context, m link.SourceMoved, now time.T
fmt.Printf("%s merged into %s (%.8s); nothing the mesh holds reads it\n", repository, m.Base, m.Commit)
return nil
}
// **Its delivery group's word, before the plans are held** (ADR 0306 decision 2): a slow answer holds up no walk.
own := ownPath(touches)
waits, err := groupWait(ctx, inv, m, deliverySeatHeld(entries), own)
if err != nil {
return notNow(err)
}
release, err := inv.HoldPlans(ctx, true)
if err != nil {
return notNow(err)
@@ -209,6 +198,7 @@ func (f following) hearMerge(ctx context.Context, m link.SourceMoved, now time.T
// **A merge on the controller's own path never shares a batch with one that waits for the delivery's
// word** (decided during the build, 2026-10-10): batched together, the catalogue's deliveries would start
// with the controller's and skip their turn. Each kind has a batch of its own.
own := ownPath(touches)
if p, ok, err := answeredByALaterMerge(ctx, inv, heard, touches, own, deliverySeatHeld(entries)); err != nil {
return notNow(err)
} else if ok {
@@ -228,7 +218,7 @@ func (f following) hearMerge(ctx context.Context, m link.SourceMoved, now time.T
return cutBatchesHeld(ctx, f.open, now)
}
batch, why, err := batchToJoin(ctx, inv, heard, entries, read, now, own, waits)
batch, err := theOpenBatch(ctx, inv, heard, now, own)
if err != nil {
return notNow(err)
}
@@ -239,7 +229,7 @@ func (f following) hearMerge(ctx context.Context, m link.SourceMoved, now time.T
if err := keepBatch(ctx, inv, &batch, now, ""); err != nil {
return notNow(err)
}
fmt.Printf("%s merged into %s (%.8s); %s, it joins %s — %s\n", repository, m.Base, m.Commit, why, batch.ID,
fmt.Printf("%s merged into %s (%.8s); it joins %s — %s\n", repository, m.Base, m.Commit, batch.ID,
batchWords(batch, now))
return cutBatchesHeld(ctx, f.open, now)
}
@@ -330,211 +320,22 @@ func ownPath(touches []inventory.Entry) bool {
return false
}
// **A merge waits only for a batch it shares something with** (novox/hq ADR 0304 decision 1, as ADR 0306
// decisions 2 and 3 read it). batchToJoin is the batch a merge heard now joins, and why in words:
//
// - an assembling or queued batch of its kind it is related to (related.go): it joins it and waits for the
// window, which its coming restarts. Related to two or more, the newer are folded into the oldest, which
// answers all their merges: a merge never waits in two windows, and what it shares with each walks in one.
// - none, and its delivery group has a member still to merge, or mesh-delivery could not say (groupWait, waits
// names why): a new batch, which waits for the window, beside the ones it shares nothing with; its words say
// why.
// - none, and no member of its group is still to merge: a new batch whose window is closed at once. Cut at
// once means only that: walks stay one at a time (ADR 0276 decision 4), so with a walk open the batch is
// queued behind it as any closed batch is.
func batchToJoin(ctx context.Context, inv *inventory.Inventory, heard inventory.BatchedMerge, entries []inventory.Entry,
read map[string][]inventory.ReadRepository, now time.Time, own bool, waits string) (inventory.Plan, string, error) {
edges, err := inv.Dependencies(ctx)
if err != nil {
return inventory.Plan{}, "", err
}
joins, whys, _, err := relatedBatches(ctx, inv, []inventory.BatchedMerge{heard}, own, "", entries, read, edges)
if err != nil {
return inventory.Plan{}, "", err
}
if len(joins) > 0 {
into := joins[0]
why := "related to " + into.ID + " (" + whys[0] + ")"
if len(joins) > 1 {
plans, moves, err := foldInto(ctx, inv, &into, joins[1:])
if err != nil {
return inventory.Plan{}, "", err
}
if err := inv.SavePlansMoving(ctx, plans, moves); err != nil {
return inventory.Plan{}, "", err
}
var folded []string
for i, b := range joins[1:] {
folded = append(folded, b.ID+" ("+whys[i+1]+")")
}
why += ", and to " + strings.Join(folded, ", ") + ", folded into it"
}
into.Delivery.Batch.CutAtOnce = false
return into, why, nil
}
if waits != "" {
// Its own window, beside the batches it shares nothing with: waiting is for its group's members, which
// are related to it by their head and join it (ADR 0306 decision 2).
b, err := newBatch(ctx, inv, heard, now, own, false, waits)
return b, waits + ", it waits for the window in a batch of its own", err
}
b, err := newBatch(ctx, inv, heard, now, own, true, "")
return b, "it shares nothing with a batch assembling and no member of a group is still to merge: it is cut at once", err
}
// relatedBatches are the batches not yet cut of a kind, oldest first, parted into those related to merges (with
// the first reason of each) and the others. Each batch is judged by what its cut would move (ADR 0306 decision
// 3): its merges' moves as they were kept and as they read now — the catalogue may have moved since a merge was
// heard — widened to their dependents, and, for a catalogue batch, every walk waiting for its word, which the
// cut folds in (cutBatch's carry). A merge kept with no moves leaves its batch unknown, related to everything.
// except is a walk not to count as folded: the one being stopped.
func relatedBatches(ctx context.Context, inv *inventory.Inventory, merges []inventory.BatchedMerge, own bool,
except string, entries []inventory.Entry, read map[string][]inventory.ReadRepository,
edges []inventory.Edge) (joins []inventory.Plan, whys []string, others []inventory.Plan, err error) {
// theOpenBatch is the batch a merge heard now joins: the one of its kind not yet cut, or a new one.
func theOpenBatch(ctx context.Context, inv *inventory.Inventory, first inventory.BatchedMerge, now time.Time, own bool) (inventory.Plan, error) {
batches, err := inv.Batches(ctx)
if err != nil {
return nil, nil, nil, err
return inventory.Plan{}, err
}
var kind []inventory.Plan
for _, b := range batches {
if b.OwnPath() == own {
kind = append(kind, b)
return b, nil
}
}
if len(kind) == 0 {
return nil, nil, nil, nil
}
carry := newFootprint()
if !own {
// An own-path batch folds no waiting walk (cutBatchesHeld); a catalogue batch may fold every one.
plans, err := inv.OpenPlans(ctx)
if err != nil {
return nil, nil, nil, err
}
for _, p := range plans {
if p.Release != nil || !p.Waiting() || p.ID == except {
continue
}
kept, err := inv.MergesOf(ctx, p.ID)
if err != nil {
return nil, nil, nil, err
}
carry = unionFootprint(carry, footprintOfWalk(p, kept))
}
}
mine := footprintAsCut(merges, entries, read, edges)
for _, b := range kind {
held, err := inv.MergesOf(ctx, b.ID)
if err != nil {
return nil, nil, nil, err
}
if b.Delivery == nil {
b.Delivery = &inventory.PlanDelivery{}
}
if b.Delivery.Batch == nil {
b.Delivery.Batch = &inventory.PlanBatch{Own: own}
}
if ok, why := related(mine, unionFootprint(footprintAsCut(held, entries, read, edges), carry), edges); ok {
joins, whys = append(joins, b), append(whys, why)
} else {
others = append(others, b)
}
}
return joins, whys, others, nil
}
// footprintAsCut is a batch's footprint as a cut now would read it: footprintOfBatch, widened by the modules each
// merge moves as the catalogue reads now (planBatch judges each merge again at the cut, so what it moves may
// have changed since it was heard). Both are kept: when in doubt, related.
func footprintAsCut(merges []inventory.BatchedMerge, entries []inventory.Entry,
read map[string][]inventory.ReadRepository, edges []inventory.Edge) footprint {
f := footprintOfBatch(merges, edges)
var moved []string
for _, m := range merges {
s := announcedOf(m).SourceMoved
if s.Owner == "" {
continue
}
s.MergedAt = "" // news, whatever a cut marked seen since (as planBatch reads it)
for _, e := range wouldMove(s, entries, read) {
moved = append(moved, e.Manifest.Module)
}
}
for _, name := range reachableFrom(moved, edges) {
f.modules[name] = true
}
return f
}
// unionFootprint is what two changes reach together.
func unionFootprint(a, b footprint) footprint {
f := newFootprint()
for _, from := range []footprint{a, b} {
for k := range from.repos {
f.repos[k] = true
}
for k := range from.modules {
f.modules[k] = true
}
for k := range from.heads {
f.heads[k] = true
}
}
f.unknown = a.unknown
if f.unknown == "" {
f.unknown = b.unknown
}
return f
}
// foldInto folds batches into the oldest one related to a merge: the plans to write (into first, so no merge
// names a plan the store does not hold) and the merges to move, for one SavePlansMoving. The folded batches name
// the batch that took them over; a stop's closed window and its withheld modules (ADR 0299) go with their merges.
func foldInto(ctx context.Context, inv *inventory.Inventory, into *inventory.Plan, from []inventory.Plan) ([]*inventory.Plan,
[]inventory.MergeMove, error) {
if into.Delivery == nil {
into.Delivery = &inventory.PlanDelivery{}
}
if into.Delivery.Batch == nil {
into.Delivery.Batch = &inventory.PlanBatch{Own: into.OwnPath()}
}
plans := []*inventory.Plan{into}
var moves []inventory.MergeMove
for i := range from {
f := from[i]
merges, err := inv.MergesOf(ctx, f.ID)
if err != nil {
return nil, nil, err
}
for _, m := range merges {
moves = append(moves, inventory.MergeMove{Repository: m.Repository, Commit: m.Commit, Plan: into.ID, Alone: m.Alone})
}
if f.Delivery == nil {
f.Delivery = &inventory.PlanDelivery{}
}
if w := f.Delivery.Batch; w != nil {
into.Delivery.Batch.Closed = into.Delivery.Batch.Closed || w.Closed
into.Delivery.Batch.Withheld = union(into.Delivery.Batch.Withheld, w.Withheld)
sort.Strings(into.Delivery.Batch.Withheld)
}
f.State = inventory.PlanSuperseded
f.Delivery.Batch = nil
f.Delivery.TakenOverBy = into.ID
f.Note = fmt.Sprintf("folded into %s before it was cut: a merge related to both joined it, and %s answers "+
"its merges", into.ID, into.ID)
plans = append(plans, &f)
}
return plans, moves, nil
}
// newBatch is a new batch of a kind, kept before its first merge names it: a merge never names a plan the store
// does not hold. cutAtOnce closes its window at once; waits says why it waits for its group, when it does.
func newBatch(ctx context.Context, inv *inventory.Inventory, first inventory.BatchedMerge, now time.Time, own,
cutAtOnce bool, waits string) (inventory.Plan, error) {
b := inventory.Plan{ID: fmt.Sprintf("plan-%d", now.UnixNano()), Repository: first.Repository, Branch: first.Branch,
Commit: first.Commit, Merged: first.Merged, Created: now, State: inventory.PlanAssembling,
Tiers: [][]string{}, Modules: map[string]*inventory.PlanModule{},
Delivery: &inventory.PlanDelivery{Batch: &inventory.PlanBatch{Own: own, CutAtOnce: cutAtOnce, Waits: waits}}}
Delivery: &inventory.PlanDelivery{Batch: &inventory.PlanBatch{Own: own}}}
// Kept before its first merge names it: a merge never names a plan the store does not hold.
if err := inv.SavePlan(ctx, &b); err != nil {
return inventory.Plan{}, err
}
@@ -566,16 +367,13 @@ func keepBatch(ctx context.Context, inv *inventory.Inventory, b *inventory.Plan,
}
closes, latest := windowOf(merges, b.Created, window, atMost)
b.Delivery.Merges = named
var closed, cutAtOnce bool
var closed bool
var withheld []string
var waits string
if was := b.Delivery.Batch; was != nil {
closed, withheld = was.Closed, was.Withheld // a stop's, which no look takes back (novox/hq ADR 0299)
cutAtOnce = was.CutAtOnce // the merge's, until another joins it (ADR 0304 decision 1)
waits = was.Waits // its group's, as said when it was made (ADR 0306 decision 2)
}
b.Delivery.Batch = &inventory.PlanBatch{ClosesAt: closes, AtMost: latest, Behind: behind, Own: b.OwnPath(),
Closed: closed, Withheld: withheld, CutAtOnce: cutAtOnce, Waits: waits}
Closed: closed, Withheld: withheld}
b.State = inventory.PlanAssembling
if behind != "" && windowClosed(*b, now) {
b.State = inventory.PlanQueued
@@ -604,14 +402,13 @@ func windowOf(merges []inventory.BatchedMerge, created time.Time, window, atMost
return last.Add(window), created.Add(atMost)
}
// windowClosed says a batch's window is closed: closed by a stop, closed at once for a merge that shares
// nothing (ADR 0304 decision 1), no merge for the window's length, or its maximum reached.
// windowClosed says a batch's window is closed: no merge for the window's length, or its maximum reached.
func windowClosed(b inventory.Plan, now time.Time) bool {
if b.Delivery == nil || b.Delivery.Batch == nil {
return true
}
w := b.Delivery.Batch
return w.Closed || w.CutAtOnce || !now.Before(w.ClosesAt) || !now.Before(w.AtMost)
return w.Closed || !now.Before(w.ClosesAt) || !now.Before(w.AtMost)
}
// carriedOf is a set of merges as a walk carries them: the latest merge of each repository's branch, and
@@ -820,9 +617,8 @@ func cutBatchesHeld(ctx context.Context, open *stores, now time.Time) error {
return err
}
// The oldest batch whose window closed is cut, an own-path one before a catalogue one (it goes first, and
// the catalogue batch queues behind it rather than being cut and overtaken). Several of a kind may be open
// since a merge that shares nothing is cut at once (ADR 0304 decision 1): walks stay one at a time, so one
// is cut per look, and once its walk started the others are queued behind it.
// the catalogue batch queues behind it rather than being cut and overtaken); one of each kind at most is
// open (theOpenBatch).
sort.SliceStable(batches, func(i, j int) bool { return batches[i].OwnPath() && !batches[j].OwnPath() })
for i := range batches {
b := &batches[i]
@@ -1090,9 +886,6 @@ func walkTimesAtCut(batch, walk inventory.Plan, entries []inventory.Entry, now t
if !w.AtMost.IsZero() && w.AtMost.Before(closed) {
closed = w.AtMost
}
if w.CutAtOnce && !batch.Created.IsZero() {
closed = batch.Created // closed when it was made (ADR 0304 decision 1)
}
if !closed.IsZero() {
if closed.After(now) {
closed = now
@@ -1349,10 +1142,6 @@ func batchWords(b inventory.Plan, now time.Time) string {
if b.State == inventory.PlanQueued {
return fmt.Sprintf("queued behind %s; grouped: %s; plan not yet calculated", w.Behind, grouped)
}
if w.CutAtOnce {
return fmt.Sprintf("cut at once, sharing nothing with a batch assembling; grouped: %s; plan not yet calculated",
grouped)
}
left := w.ClosesAt.Sub(now)
if w.AtMost.Before(w.ClosesAt) {
left = w.AtMost.Sub(now)
@@ -1360,12 +1149,8 @@ func batchWords(b inventory.Plan, now time.Time) string {
if left < 0 {
left = 0
}
waits := ""
if w.Waits != "" {
waits = " — " + w.Waits
}
return fmt.Sprintf("assembling: %s left (at the latest %s)%s; grouped: %s; plan not yet calculated",
secondsWords(left), w.AtMost.Local().Format("15:04:05"), waits, grouped)
return fmt.Sprintf("assembling: %s left (at the latest %s); grouped: %s; plan not yet calculated",
secondsWords(left), w.AtMost.Local().Format("15:04:05"), grouped)
}
// groupedWords is a batch's commits, each with the earlier merges it answers.
+13 -77
View File
@@ -34,7 +34,6 @@ func windowed(t *testing.T) *stores {
t.Fatal(err)
}
t.Setenv(mergeWindowFileVar, settings)
noGroups(t)
if err := inv.RegisterModule(ctx, catalogue.Manifest{Module: "mesh-controller", Version: "1"},
inventory.Source{Repository: "novox/mesh-controller", Seat: "git", Ref: "main", BuiltFrom: "c0",
Head: "c0"}); err != nil {
@@ -59,70 +58,19 @@ func windowed(t *testing.T) *stores {
// t0 is the moment a test's clock starts: merges are dated by it, and heard after it.
var t0 = time.Now().UTC().Truncate(time.Second)
// catalogueMerge is a merge of the catalogue changing one module's directory, made at a moment, from a pull
// request's head branch of its own, as the forge announces every merge. Whether its head names a delivery group
// with a member still to merge is mesh-delivery's word, which windowed stubs as no group at all (noGroups); a test
// whose merges wait for the window as a group's says so (groupsWaiting).
// catalogueMerge is a merge of the catalogue changing one module's directory, made at a moment.
func catalogueMerge(commit, module string, made time.Time) link.SourceMoved {
return link.SourceMoved{Owner: "novox", Repo: "mesh-catalog", Base: "main", Head: "pr/" + commit, Commit: commit,
return link.SourceMoved{Owner: "novox", Repo: "mesh-catalog", Base: "main", Commit: commit,
MergedAt: made.Format(time.RFC3339Nano), Paths: []string{"modules/" + module + "/module.json"},
ModuleDirs: []string{"modules/" + module}, ModuleDirsSaid: true}
}
// repoMerge is a merge of a repository of one module's own, from a head branch of its own (catalogueMerge).
// repoMerge is a merge of a repository of one module's own.
func repoMerge(repo, commit string, made time.Time) link.SourceMoved {
return link.SourceMoved{Owner: "novox", Repo: repo, Base: "main", Head: "pr/" + commit, Commit: commit,
return link.SourceMoved{Owner: "novox", Repo: repo, Base: "main", Commit: commit,
MergedAt: made.Format(time.RFC3339Nano), Paths: []string{"main.go"}}
}
// noGroups stubs mesh-delivery's word on groups for a test: no head names a group.
func noGroups(t *testing.T) {
t.Helper()
was := deliveryGroupOf
deliveryGroupOf = func(context.Context, bool, string) ([]groupMember, bool, error) { return nil, false, nil }
t.Cleanup(func() { deliveryGroupOf = was })
}
// groupsWaiting stubs mesh-delivery's word on groups for a test: each head named is a group with a member of another
// repository whose pull request is open (novox/elsewhere, ready), so a merge from it waits for the window (ADR 0306
// decision 2); every other head names no group.
func groupsWaiting(t *testing.T, heads ...string) {
t.Helper()
was := deliveryGroupOf
deliveryGroupOf = func(_ context.Context, _ bool, head string) ([]groupMember, bool, error) {
if slices.Contains(heads, head) {
return []groupMember{{Repository: "novox/elsewhere", State: "ready"}}, true, nil
}
return nil, false, nil
}
t.Cleanup(func() { deliveryGroupOf = was })
}
// everyHeadWaits stubs mesh-delivery's word on groups for a test whose merges each wait for the window as they did
// before novox/hq ADR 0304: every head is a group with a member of another repository still to merge.
func everyHeadWaits(t *testing.T) {
t.Helper()
was := deliveryGroupOf
deliveryGroupOf = func(context.Context, bool, string) ([]groupMember, bool, error) {
return []groupMember{{Repository: "novox/elsewhere", State: "ready"}}, true, nil
}
t.Cleanup(func() { deliveryGroupOf = was })
}
// fromHead is a merge from a head branch a test names: merges from one head are members of one delivery group,
// related by it (novox/hq ADR 0304 decision 1), so they share a batch whatever else they share.
func fromHead(m link.SourceMoved, head string) link.SourceMoved {
m.Head = head
return m
}
// lone is a merge that names no head: known to be no delivery group's, it waits for no window unless it is
// related to a batch assembling (ADR 0304 decision 1).
func lone(m link.SourceMoved) link.SourceMoved {
m.Head = ""
return m
}
// hear is a merge heard at a moment.
func hear(t *testing.T, open *stores, m link.SourceMoved, now time.Time) {
t.Helper()
@@ -160,7 +108,6 @@ func walks(t *testing.T, open *stores) (walks, batches []inventory.Plan) {
// later commit, which names both merges — the earlier carried by the later.
func TestTwoMergesSecondsApartAreOneWalkAtTheLaterCommit(t *testing.T) {
open := windowed(t)
everyHeadWaits(t)
asked := asksWithPaths(t)
claude := catalogueMerge("553b7191claude", "app", t0)
dunst := catalogueMerge("48bda475dunst", "notes", t0.Add(17*time.Second))
@@ -205,10 +152,9 @@ func TestTwoMergesSecondsApartAreOneWalkAtTheLaterCommit(t *testing.T) {
// Merges of three repositories in one window are one walk, one commit for each.
func TestThreeRepositoriesInOneWindowAreOneWalk(t *testing.T) {
open := windowed(t)
groupsWaiting(t, "feat/three")
asksRecorded(t)
for i, r := range []string{"one", "two", "three"} {
hear(t, open, fromHead(repoMerge(r, "c-"+r, t0.Add(time.Duration(i)*20*time.Second)), "feat/three"),
hear(t, open, repoMerge(r, "c-"+r, t0.Add(time.Duration(i)*20*time.Second)),
t0.Add(time.Duration(i)*20*time.Second+time.Second))
}
cutAt(t, open, t0.Add(41*time.Second+90*time.Second))
@@ -241,10 +187,9 @@ func TestThreeRepositoriesInOneWindowAreOneWalk(t *testing.T) {
// closes at most ten minutes after its first merge however busy.
func TestAMergeRestartsTheWindowAndTheMaximumClosesIt(t *testing.T) {
open := windowed(t)
groupsWaiting(t, "feat/both", "feat/busy")
asksWithPaths(t)
hear(t, open, fromHead(repoMerge("one", "c1", t0), "feat/both"), t0)
hear(t, open, fromHead(repoMerge("two", "c2", t0.Add(80*time.Second)), "feat/both"), t0.Add(80*time.Second))
hear(t, open, repoMerge("one", "c1", t0), t0)
hear(t, open, repoMerge("two", "c2", t0.Add(80*time.Second)), t0.Add(80*time.Second))
cutAt(t, open, t0.Add(100*time.Second))
if ws, _ := walks(t, open); len(ws) != 0 {
t.Fatal("the window closed 90 seconds after its first merge, not after its last")
@@ -255,13 +200,12 @@ func TestAMergeRestartsTheWindowAndTheMaximumClosesIt(t *testing.T) {
}
busy := windowed(t)
groupsWaiting(t, "feat/both", "feat/busy")
asksWithPaths(t)
var last time.Time
for i := 0; i < 12; i++ {
last = t0.Add(time.Duration(i) * time.Minute)
repo := []string{"one", "two", "three"}[i%3]
hear(t, busy, fromHead(repoMerge(repo, "c"+string(rune('a'+i)), last), "feat/busy"), last)
hear(t, busy, repoMerge(repo, "c"+string(rune('a'+i)), last), last)
if ws, _ := walks(t, busy); len(ws) > 0 {
if i != 10 {
t.Fatalf("a busy window closed at merge %d (minute %d), not at its maximum", i, i)
@@ -279,11 +223,10 @@ func TestAMergeRestartsTheWindowAndTheMaximumClosesIt(t *testing.T) {
// what is grouped, and that the plan is not yet calculated. It says so on the bus as plan-moved, too.
func TestTheAssemblingBatchIsShown(t *testing.T) {
open := windowed(t)
groupsWaiting(t, "pr/553b7191claude")
asksWithPaths(t)
var said []inventory.Plan
was := inventory.PlanSaved
inventory.PlanSaved = func(_ *inventory.Inventory, p inventory.Plan) { said = append(said, p) }
inventory.PlanSaved = func(p inventory.Plan) { said = append(said, p) }
t.Cleanup(func() { inventory.PlanSaved = was })
now := time.Now().UTC()
claude := catalogueMerge("553b7191claude", "app", now.Add(-20*time.Second))
@@ -292,7 +235,7 @@ func TestTheAssemblingBatchIsShown(t *testing.T) {
dunst.Number, dunst.Title = 175, "dunst: the font the operator chose"
hear(t, open, claude, now.Add(-19*time.Second))
hear(t, open, dunst, now.Add(-time.Second))
hear(t, open, fromHead(repoMerge("one", "a6bc0931one", now), dunst.Head), now)
hear(t, open, repoMerge("one", "a6bc0931one", now), now)
out := captured(t, func() error { return plansCommand(t.Context(), nil) })
lines := strings.Split(out, "\n")
first := lines[0]
@@ -417,7 +360,6 @@ func TestAWaitingWalkIsFoldedIntoTheNextBatch(t *testing.T) {
// moves — named at once on a walk done — and joins the next batch when it does not.
func TestALateMergeIsAnsweredByTheWalkOfTheLaterOne(t *testing.T) {
open := windowed(t)
everyHeadWaits(t)
asksWithPaths(t)
ctx := t.Context()
later := catalogueMerge("48bda475dunst", "notes", t0.Add(17*time.Second))
@@ -449,7 +391,6 @@ func TestALateMergeIsAnsweredByTheWalkOfTheLaterOne(t *testing.T) {
// first; the first delivered answers the older ones, and the search stops.
func TestAFailedWalkWalksItsEarlierMergesAlone(t *testing.T) {
open := windowed(t)
everyHeadWaits(t)
asked := asksWithPaths(t)
ctx := t.Context()
// Made before the cut marks their modules seen, as merges are: walked again, they are news all the same.
@@ -518,7 +459,6 @@ func TestAFailedWalkWalksItsEarlierMergesAlone(t *testing.T) {
func TestAGroupsOrderBecomesTiersInsideTheWalk(t *testing.T) {
for _, grouped := range []bool{true, false} {
open := windowed(t)
everyHeadWaits(t)
asksWithPaths(t)
ctx := t.Context()
for _, m := range []string{"mesh-host"} {
@@ -556,10 +496,9 @@ func TestAGroupsOrderBecomesTiersInsideTheWalk(t *testing.T) {
// store, a merge handed over again is not doubled, and the batch is cut when its window closes.
func TestABatchSurvivesARestart(t *testing.T) {
open := windowed(t)
groupsWaiting(t, "feat/both")
asksWithPaths(t)
hear(t, open, fromHead(repoMerge("one", "c1", t0), "feat/both"), t0)
hear(t, open, fromHead(repoMerge("two", "c2", t0.Add(30*time.Second)), "feat/both"), t0.Add(30*time.Second))
hear(t, open, repoMerge("one", "c1", t0), t0)
hear(t, open, repoMerge("two", "c2", t0.Add(30*time.Second)), t0.Add(30*time.Second))
// A new controller: nothing of the first one's but the store.
again, err := openStores(t.Context())
@@ -567,7 +506,7 @@ func TestABatchSurvivesARestart(t *testing.T) {
t.Fatal(err)
}
t.Cleanup(again.Close)
hear(t, again, fromHead(repoMerge("one", "c1", t0), "feat/both"), t0.Add(60*time.Second)) // the bus hands it over again
hear(t, again, repoMerge("one", "c1", t0), t0.Add(60*time.Second)) // the bus hands it over again
cutAt(t, again, t0.Add(119*time.Second))
if ws, bs := walks(t, again); len(ws) != 0 || len(bs) != 1 || len(bs[0].Delivery.Merges) != 2 {
t.Fatalf("the restarted controller's batch: %d walk(s), %+v", len(ws), bs)
@@ -624,7 +563,6 @@ func TestAWaitingWalkNamesTheMergesItAnswers(t *testing.T) {
// nothing, and the merge joins the next batch.
func TestALateMergeOfAOneModuleRepositoryIsNamed(t *testing.T) {
open := windowed(t)
everyHeadWaits(t)
asksWithPaths(t)
ctx := t.Context()
c2 := repoMerge("one", "c2", t0.Add(-50*time.Second))
@@ -778,7 +716,6 @@ func TestTheCatchUpKeepsWhatACutMadeHistory(t *testing.T) {
// still waits for the word, so no catalogue delivery skips its turn.
func TestAMergeOnTheControllersPathNeverSharesABatch(t *testing.T) {
open := windowed(t)
everyHeadWaits(t)
asked := asksWithPaths(t)
ctx := t.Context()
if err := open.inventory.RegisterModule(ctx, catalogue.Manifest{Module: "mesh-delivery", Version: "1",
@@ -923,7 +860,6 @@ func sameMerge(a, b inventory.PlanMerge) bool {
// its delivery would skip its turn. It joins the next catalogue batch.
func TestALateCatalogueMergeIsNotAnsweredByAnOwnPathWalk(t *testing.T) {
open := windowed(t)
everyHeadWaits(t)
asksWithPaths(t)
ctx := t.Context()
for _, m := range []struct {
-4
View File
@@ -48,10 +48,6 @@ func gatherBatches(ctx context.Context, inv *inventory.Inventory, now time.Time)
if f.atMost.Before(f.closed) {
f.closed = f.atMost
}
if w.Batch.CutAtOnce && !b.Created.IsZero() {
// Its window closed when it was made, which is its latest too (ADR 0304 decision 1).
f.atMost, f.closed = b.Created, b.Created
}
}
if f.behind != "" {
if walk, err := inv.PlanByID(ctx, f.behind); err == nil {
+29 -105
View File
@@ -10,7 +10,6 @@ import (
"slices"
"sort"
"strings"
"sync"
"time"
"github.com/nats-io/nats.go"
@@ -241,7 +240,7 @@ func stopWalkWithout(ctx context.Context, inv *inventory.Inventory, id, by, why
}
return p, said, inv.SavePlan(ctx, &p)
}
target, folded, foldMoves, err := batchToWalkAgain(ctx, inv, p.ID, again, now)
target, err := batchToWalkAgain(ctx, inv, again, now)
if err != nil {
return p, said, fmt.Errorf("%s is not stopped: %w", p.ID, err)
}
@@ -262,10 +261,8 @@ func stopWalkWithout(ctx context.Context, inv *inventory.Inventory, id, by, why
if len(said.Kept) > 0 {
p.Note += "; " + keptWords(said.Kept)
}
// The batch first, so no merge names a plan the store does not hold; all of it, or nothing (decision 7) — the
// batches folded into it too.
if err := inv.SavePlansMoving(ctx, append([]*inventory.Plan{&target, &p}, folded...),
append(moves, foldMoves...)); err != nil {
// The batch first, so no merge names a plan the store does not hold; all of it, or nothing (decision 7).
if err := inv.SavePlansMoving(ctx, []*inventory.Plan{&target, &p}, moves); err != nil {
return p, said, fmt.Errorf("%s is not stopped, and none of its merges moved: %w", p.ID, err)
}
// What the batch holds and names now: written again by the cutter's next look should this fail.
@@ -320,42 +317,36 @@ func stopInBatch(ctx context.Context, inv *inventory.Inventory, b *inventory.Pla
return nil
}
// batchToWalkAgain is the batch a stopped walk's merges join (novox/hq ADR 0299): a batch of their kind not yet cut
// that they are related to (ADR 0304 decision 1, ADR 0306 decision 3), or a new one, not yet written. Related to
// two or more, the newer are folded into the oldest: the plans and merge moves that fold them, to be written in
// the stop's one act. The kind is read from what the merges moved; the stopped walk itself is no walk a cut folds.
func batchToWalkAgain(ctx context.Context, inv *inventory.Inventory, stopping string, again []inventory.BatchedMerge,
now time.Time) (inventory.Plan, []*inventory.Plan, []inventory.MergeMove, error) {
// batchToWalkAgain is the batch a stopped walk's merges join: the open batch of their kind, or a new one, not yet
// written. The kind is read from what the merges moved.
func batchToWalkAgain(ctx context.Context, inv *inventory.Inventory, again []inventory.BatchedMerge,
now time.Time) (inventory.Plan, error) {
own := false
for _, n := range movesOf(again) {
if _, on := onTheControllersPath[n]; on {
own = true
}
}
entries, read, edges, err := theGraph(ctx, inv)
batches, err := inv.Batches(ctx)
if err != nil {
return inventory.Plan{}, nil, nil, err
return inventory.Plan{}, err
}
joins, _, _, err := relatedBatches(ctx, inv, again, own, stopping, entries, read, edges)
if err != nil {
return inventory.Plan{}, nil, nil, err
}
if len(joins) > 0 {
into := joins[0]
if len(joins) == 1 {
return into, nil, nil, nil
for _, b := range batches {
if b.OwnPath() == own {
if b.Delivery == nil {
b.Delivery = &inventory.PlanDelivery{}
}
if b.Delivery.Batch == nil {
b.Delivery.Batch = &inventory.PlanBatch{Own: own}
}
return b, nil
}
plans, moves, err := foldInto(ctx, inv, &into, joins[1:])
if err != nil {
return inventory.Plan{}, nil, nil, err
}
return into, plans[1:], moves, nil
}
first := again[0]
return inventory.Plan{ID: fmt.Sprintf("plan-%d", now.UnixNano()), Repository: first.Repository, Branch: first.Branch,
Commit: first.Commit, Merged: first.Merged, Created: now, State: inventory.PlanAssembling,
Tiers: [][]string{}, Modules: map[string]*inventory.PlanModule{},
Delivery: &inventory.PlanDelivery{Batch: &inventory.PlanBatch{Own: own}}}, nil, nil, nil
Delivery: &inventory.PlanDelivery{Batch: &inventory.PlanBatch{Own: own}}}, nil
}
// leaveOut splits a walk's merges by the merges named to leave out, each `<repository>@<commit>` (a commit's
@@ -949,7 +940,7 @@ func deliveryCommand(ctx context.Context, args []string) error {
if *group == "" && len(members) == 1 {
// One head, checked again as the forge's announcement would have it (a recheck): the controller's
// own path for a pull request, run because the delivery's owner asked.
if err := sayingOnTheBus(ctx, inv, func() error { return (following{open}).PullUpdated(ctx, members[0].pull()) }); err != nil {
if err := sayingOnTheBus(ctx, func() error { return (following{open}).PullUpdated(ctx, members[0].pull()) }); err != nil {
return err
}
return answer(map[string]any{"rechecked": members[0].ID})
@@ -967,7 +958,7 @@ func deliveryCommand(ctx context.Context, args []string) error {
return errors.New("delivery go <plan> [--by <who>] [--why <text>]")
}
var p inventory.Plan
if err := sayingOnTheBus(ctx, inv, func() (err error) {
if err := sayingOnTheBus(ctx, func() (err error) {
p, err = letGo(ctx, inv, positionals[0], *by, strings.TrimSpace(*why))
return err
}); err != nil {
@@ -982,7 +973,7 @@ func deliveryCommand(ctx context.Context, args []string) error {
}
var p inventory.Plan
var said stopAnswer
if err := sayingOnTheBus(ctx, inv, func() (err error) {
if err := sayingOnTheBus(ctx, func() (err error) {
p, said, err = stopWalkWithout(ctx, inv, positionals[0], *by, strings.TrimSpace(*why), splitList(*without),
time.Now().UTC())
return err
@@ -1224,19 +1215,15 @@ func shelfOf(entries []inventory.Entry) map[string]catalogue.Manifest {
// delivery it walks reads its steps from. Never in the way of the walk: said beside it, and a save that could
// not be said is logged — the delivery's owner, which also asks for the walks it follows, finds it by
// comparison. Records arriving out of order are told apart by the plan's revision.
//
// On the save that starts the walk and the one that ends it, the walk also says what it touches, its bound and,
// at the end, its outcome (novox/hq ADR 0308 decision 1, walk_said.go), read from facts; nil facts say only what
// the walk recorded.
func sayPlanMoved(ctx context.Context, bus link.Bus, facts walkFacts, p inventory.Plan) {
go publishPlanMoved(ctx, bus, facts, p)
func sayPlanMoved(ctx context.Context, bus link.Bus, p inventory.Plan) {
go publishPlanMoved(ctx, bus, p)
}
func publishPlanMoved(ctx context.Context, bus link.Bus, facts walkFacts, p inventory.Plan) {
func publishPlanMoved(ctx context.Context, bus link.Bus, p inventory.Plan) {
// Its phases so far (novox/hq ADR 0282 decision 6): the rest's reports come after the walk ends, and are
// read by whoever asks for the walk (`delivery walks`).
p.Phases = p.WalkPhases(time.Now(), nil)
body, err := json.Marshal(planMovedBody{Plan: p, walkSaid: sayWalk(ctx, facts, p)})
body, err := json.Marshal(p)
if err != nil {
return
}
@@ -1251,7 +1238,7 @@ func publishPlanMoved(ctx context.Context, bus link.Bus, facts walkFacts, p inve
// controller's, or a connection of the command's own — a verb runs as a command of its own, and what it kept
// would otherwise be said by nobody until the delivery's owner read the walks back. Without a bus the command
// still runs; what it did is found by comparison.
func sayingOnTheBus(ctx context.Context, facts walkFacts, f func() error) error {
func sayingOnTheBus(ctx context.Context, f func() error) error {
if checkEvents != nil {
return f()
}
@@ -1262,10 +1249,9 @@ func sayingOnTheBus(ctx context.Context, facts walkFacts, f func() error) error
return err
}
bus := link.OverNATS{Conn: conn, JS: js}
was := inventory.PlanSaved
checkEvents = bus
inventory.PlanSaved = func(_ *inventory.Inventory, p inventory.Plan) { publishPlanMoved(ctx, bus, facts, p) }
defer func() { checkEvents, inventory.PlanSaved = nil, was }()
inventory.PlanSaved = func(p inventory.Plan) { publishPlanMoved(ctx, bus, p) }
defer func() { checkEvents, inventory.PlanSaved = nil, nil }()
ran = true
return f()
})
@@ -1275,65 +1261,3 @@ func sayingOnTheBus(ctx context.Context, facts walkFacts, f func() error) error
}
return err
}
// sayingWhatAVerbKept is PlanSaved for a process that is not the serving controller: each walk the verb keeps is
// said as `plan-moved` before the save returns, with the facts of the inventory that kept it, on the bus the verb
// reaches. It reaches the bus on the first save and never for a verb that keeps no walk. When the bus cannot be
// reached, a save that started or ended a walk says so on the verb's error output: its mark is kept and no later
// save says it, so the router holds what the walk touched until its bound, or not at all for a start. done lets
// the bus go and takes the hook back.
func sayingWhatAVerbKept(ctx context.Context) (said func(*inventory.Inventory, inventory.Plan), done func()) {
var (
mu sync.Mutex
reached bool
bus link.Bus
closing func()
failed error
)
said = func(inv *inventory.Inventory, p inventory.Plan) {
mu.Lock()
defer mu.Unlock()
if !reached {
reached = true
if handActConn != nil {
bus, failed = overNATS(handActConn)
} else if js, err := aBus(); err != nil {
failed = err
} else if bus, failed = overNATS(js.Conn()); failed == nil {
closing = js.Close
} else {
js.Close()
}
}
if bus == nil {
switch p.WalkSaid {
case inventory.WalkStarted:
fmt.Fprintf(os.Stderr, "%s: its start is kept and could NOT be said as plan-moved (%v): the router "+
"does not hold what it touches\n", p.ID, failed)
case inventory.WalkEnded:
fmt.Fprintf(os.Stderr, "%s: its end is kept and could NOT be said as plan-moved (%v): the router holds "+
"what it touched until its bound\n", p.ID, failed)
}
return
}
publishPlanMoved(ctx, bus, inv, p)
}
done = func() {
mu.Lock()
defer mu.Unlock()
inventory.PlanSaved = nil
if closing != nil {
closing()
}
}
return said, done
}
// overNATS is the bus a connection carries, with its JetStream.
func overNATS(conn *nats.Conn) (link.Bus, error) {
js, err := conn.JetStream()
if err != nil {
return nil, err
}
return link.OverNATS{Conn: conn, JS: js}, nil
}
+4 -8
View File
@@ -47,14 +47,10 @@ type stalledLine struct {
// issue 443): a time inside a finished sentence cannot be said again in another zone. A mesh-delivery from before
// says none, and Waiting is said as it reads.
Checks []waitingCheck `json:"checks,omitempty"`
// Held is the state of the delivery of a line of the state `unanswered`, ready or rejected, when one is held
// (novox/hq issues 441, 451): its check answered once, so a recheck is the step. Empty for a head with no
// delivery, and from a mesh-delivery that does not say it yet; the words are never read for it.
Held string `json:"held,omitempty"`
For string `json:"for"`
Bound string `json:"bound"`
H2 string `json:"h2"`
Says string `json:"says"`
For string `json:"for"`
Bound string `json:"bound"`
H2 string `json:"h2"`
Says string `json:"says"`
}
// waitingCheck is one merge check a stalled line waits on, as mesh-delivery gives it: its context, its state —
@@ -119,13 +119,6 @@ func TestTheBusMachineIsBehindByItsUserListAlone(t *testing.T) {
if userListBehind("", "") {
t.Fatal("a machine sent no list reads as behind")
}
// A list that could not be composed is behind: not known is never "not behind" (the review of hq issue 353).
if !listBehind("", errors.New("the plan cannot be worked out"), digestOf([]byte(list))) {
t.Fatal("a list nobody could compose reads as current")
}
if listBehind(list, nil, digestOf([]byte(list))) || !listBehind(list, nil, "") {
t.Fatal("listBehind does not read the digest as userListBehind does")
}
}
// The machine holding the bus goes first: its declaration carries the user list the new grants are
-158
View File
@@ -1,158 +0,0 @@
package main
import (
"context"
"encoding/json"
"errors"
"fmt"
"strings"
"time"
"github.com/novox/mesh-controller/internal/catalogue"
"github.com/novox/mesh-controller/internal/inventory"
"github.com/novox/mesh-controller/internal/link"
)
// **A delivery group still assembles** (novox/hq ADR 0306 decision 2): a merge whose delivery group has a member
// not yet merged waits for the merge window, whether or not a batch is assembling, so the group's members walk in
// one batch, in their order (ADR 0276 decision 5). Which groups exist and where each member stands is
// mesh-delivery's record (ADR 0239): a group is the pull requests of two repositories or more sharing a head
// branch name, which is its id. The controller asks it, through its `groups` verb, when a merge from a head is
// heard; a member whose pull request is open (proposed, checked, ready or rejected) is still to merge. When
// mesh-delivery does not answer within groupAskWithin, or answers something unreadable, the merge waits: the safe
// error is to wait, and its batch says why.
// groupAskWithin is how long mesh-delivery is given to say a merge's group, while the merge is being heard.
var groupAskWithin = 3 * time.Second
// openPullStates are the states of a delivery whose pull request is open and not merged (mesh-delivery's table):
// a member in one of them is still to merge. Closed unmerged is stopped, final, and waits for nothing.
var openPullStates = map[string]bool{"proposed": true, "checked": true, "ready": true, "rejected": true}
// groupMember is one member of a delivery group as mesh-delivery's `groups` says it: `<owner/repo>@<commit> (state)`.
type groupMember struct {
Repository, State string
}
// deliveryGroupOf is what mesh-delivery says of the group a head names: its members, and whether the head names a
// group at all; an error when it cannot be asked or did not answer. A seam a test replaces.
var deliveryGroupOf = func(ctx context.Context, held bool, head string) ([]groupMember, bool, error) {
if !held {
return nil, false, fmt.Errorf("no holder of %s is on record to say it", catalogue.DeliverySeat)
}
serving := servingBus.Load()
if serving == nil {
return nil, false, errors.New("this controller is not on the bus")
}
asking, cancel := context.WithTimeout(ctx, groupAskWithin)
defer cancel()
raw, err := askDeliveryOwner(asking, serving.Conn(), "groups", map[string]any{})
if err != nil {
if errors.Is(err, link.ErrNothingServes) {
return nil, false, fmt.Errorf("nothing serves %s.groups", catalogue.DeliverySeat)
}
return nil, false, err
}
var lines []struct {
ID string `json:"id"`
Members []string `json:"members"`
}
if err := json.Unmarshal(raw, &lines); err != nil {
return nil, false, fmt.Errorf("%s.groups answered something unreadable: %w", catalogue.DeliverySeat, err)
}
for _, l := range lines {
if l.ID != head {
continue
}
var members []groupMember
for _, s := range l.Members {
members = append(members, memberOf(s))
}
return members, true, nil
}
return nil, false, nil
}
// memberOf reads one member as `groups` says it: `novox/mesh-sdk@d41d255f28ad (held)`.
func memberOf(s string) groupMember {
id, state := strings.TrimSpace(s), ""
if i := strings.LastIndex(id, " ("); i >= 0 && strings.HasSuffix(id, ")") {
id, state = id[:i], id[i+2:len(id)-1]
}
repository, _, _ := strings.Cut(id, "@")
return groupMember{Repository: repository, State: state}
}
// groupWait says whether a merge heard now waits for the window for its delivery group, and why in words; nothing
// when it names no head, its head names no group, or every other member is merged or closed. own is the merge's
// kind (ownPath).
//
// A member of another repository whose merge this controller already keeps in a batch not yet cut of the same
// kind is merged, whatever mesh-delivery says yet (it marks a member merged once its walk opened): the merge joins
// that batch by their head. One kept in a batch of the other kind keeps the merge waiting, whatever mesh-delivery
// says (review of #250): kinds never share a batch, so the merge's batch is a new one, and cut at once it would
// walk before that member's batch is cut and break the group's order (ADR 0276 decision 5). Waiting, the two
// batches are cut in the order their windows close, an own-path one first at one look, as cutBatchesHeld cuts
// every pair of kinds.
//
// Asked before the plans are held: a slow answer holds up no walk.
func groupWait(ctx context.Context, inv *inventory.Inventory, m link.SourceMoved, held, own bool) (string, error) {
if m.Head == "" {
return "", nil
}
same, other, err := keptFromHead(ctx, inv, m.Head, own)
if err != nil {
return "", err
}
if len(other) > 0 {
return "group " + m.Head + " has a member waiting in a batch of the other kind: " + strings.Join(other, ", "), nil
}
members, isGroup, err := deliveryGroupOf(ctx, held, m.Head)
if err != nil {
return fmt.Sprintf("%s did not say whether group %s has a member still to merge (%v): it waits for the "+
"window, the safe side", catalogue.DeliverySeat, m.Head, err), nil
}
if !isGroup {
return "", nil
}
var open []string
for _, g := range members {
if strings.EqualFold(g.Repository, m.Owner+"/"+m.Repo) || same[strings.ToLower(g.Repository)] ||
!openPullStates[g.State] {
continue
}
open = append(open, g.Repository+" ("+g.State+")")
}
if len(open) == 0 {
return "", nil
}
return "group " + m.Head + " has a member still to merge: " + strings.Join(open, ", "), nil
}
// keptFromHead are the merges from a head that batches not yet cut hold: the repositories (lower case) of those in
// a batch of the kind own says, and, in words, each of the other kind with its batch.
func keptFromHead(ctx context.Context, inv *inventory.Inventory, head string, own bool) (map[string]bool, []string, error) {
batches, err := inv.Batches(ctx)
if err != nil {
return nil, nil, err
}
same := map[string]bool{}
var other []string
for _, b := range batches {
merges, err := inv.MergesOf(ctx, b.ID)
if err != nil {
return nil, nil, err
}
for _, k := range merges {
if announcedOf(k).Head != head {
continue
}
if b.OwnPath() == own {
same[strings.ToLower(k.Repository)] = true
} else {
other = append(other, fmt.Sprintf("%s@%s in %s", k.Repository, short(k.Commit), b.ID))
}
}
}
return same, other, nil
}
+5 -11
View File
@@ -123,12 +123,11 @@ func TestDeliveryStopTakesTheMergesToLeaveOut(t *testing.T) {
// that builds the other repository's module alone.
func TestAStopWithoutAMergeWalksTheOthersAgain(t *testing.T) {
open := windowed(t)
groupsWaiting(t, "pr/cat1aaaaaaaa")
endsItsLine(t)
asksRecorded(t)
ctx := t.Context()
cat := catalogueMerge("cat1aaaaaaaa", "app", t0)
one := fromHead(repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second)), cat.Head)
one := repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second))
hear(t, open, cat, t0.Add(time.Second))
hear(t, open, one, t0.Add(6*time.Second))
cutAt(t, open, t0.Add(2*time.Minute))
@@ -182,13 +181,12 @@ func TestAStopWithoutAMergeWalksTheOthersAgain(t *testing.T) {
// ended. A merge the walk does not answer is refused, and nothing ends.
func TestAStopWithoutAMergeEndsTheMergesThatContainIt(t *testing.T) {
open := windowed(t)
groupsWaiting(t, "pr/cat1aaaaaaaa")
endsItsLine(t)
asksRecorded(t)
ctx := t.Context()
first := catalogueMerge("cat1aaaaaaaa", "app", t0)
later := catalogueMerge("cat2cccccccc", "notes", t0.Add(10*time.Second))
one := fromHead(repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second)), first.Head)
one := repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second))
hear(t, open, first, t0.Add(time.Second))
hear(t, open, one, t0.Add(6*time.Second))
hear(t, open, later, t0.Add(11*time.Second))
@@ -240,12 +238,11 @@ func TestAStopWithoutAMergeEndsTheMergesThatContainIt(t *testing.T) {
// (review of #232): its window closes, and its walk is at the newer commit, answering both.
func TestAStopWalksTheOthersAgainInTheOpenBatch(t *testing.T) {
open := windowed(t)
groupsWaiting(t, "pr/cat1aaaaaaaa")
endsItsLine(t)
asksRecorded(t)
ctx := t.Context()
cat := catalogueMerge("cat1aaaaaaaa", "app", t0)
one1 := fromHead(repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second)), cat.Head)
one1 := repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second))
hear(t, open, cat, t0.Add(time.Second))
hear(t, open, one1, t0.Add(6*time.Second))
cutAt(t, open, t0.Add(2*time.Minute))
@@ -293,12 +290,11 @@ func TestAStopWalksTheOthersAgainInTheOpenBatch(t *testing.T) {
// that walked its merges again is refused a retry, saying why.
func TestAStopNamesWhatStaysOnTheMachinesAndWhatStillCarriesIt(t *testing.T) {
open := windowed(t)
groupsWaiting(t, "pr/cat1aaaaaaaa")
endsItsLine(t)
asksRecorded(t)
ctx := t.Context()
cat := catalogueMerge("cat1aaaaaaaa", "app", t0)
one1 := fromHead(repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second)), cat.Head)
one1 := repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second))
hear(t, open, cat, t0.Add(time.Second))
hear(t, open, one1, t0.Add(6*time.Second))
cutAt(t, open, t0.Add(2*time.Minute))
@@ -339,12 +335,11 @@ func TestAStopNamesWhatStaysOnTheMachinesAndWhatStillCarriesIt(t *testing.T) {
// with the others and withholds the stopped merge's modules; with every merge left out, the batch is stopped.
func TestAStopWithoutAMergeLeavesItOutOfABatch(t *testing.T) {
open := windowed(t)
everyHeadWaits(t)
endsItsLine(t)
asksRecorded(t)
ctx := t.Context()
cat := catalogueMerge("cat1aaaaaaaa", "app", t0)
one := fromHead(repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second)), cat.Head)
one := repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second))
hear(t, open, cat, t0.Add(time.Second))
hear(t, open, one, t0.Add(6*time.Second))
_, bs := walks(t, open)
@@ -387,7 +382,6 @@ func TestAStopWithoutAMergeLeavesItOutOfABatch(t *testing.T) {
// else since it was read refuses all of it, and no merge moves, no plan is written.
func TestSavingPlansAndMovingMergesIsOneAct(t *testing.T) {
open := windowed(t)
everyHeadWaits(t)
endsItsLine(t)
ctx := t.Context()
hear(t, open, catalogueMerge("cat1aaaaaaaa", "app", t0), t0.Add(time.Second))
-398
View File
@@ -1,398 +0,0 @@
package main
import (
"context"
"errors"
"strings"
"testing"
"time"
"github.com/novox/mesh-controller/internal/catalogue"
"github.com/novox/mesh-controller/internal/inventory"
)
// The tests of novox/hq issue 525's window skip: a merge waits for the merge window only when a batch assembling
// holds a merge related to it, or its delivery group may have a member still to merge (ADR 0304 decision 1, ADR
// 0306 decisions 2 and 3). Walks stay one at a time.
// A merge from no head that shares nothing, with no walk open, is cut into its walk when it is heard: no window.
func TestALoneMergeSharingNothingIsCutAtOnce(t *testing.T) {
open := windowed(t)
asksWithPaths(t)
hear(t, open, lone(repoMerge("one", "c1", t0)), t0)
ws, bs := walks(t, open)
if len(ws) != 1 || len(bs) != 0 || !ws[0].Open() || ws[0].Commit != "c1" {
t.Fatalf("a lone merge sharing nothing: %d walk(s) %+v, %d batch(es)", len(ws), ws, len(bs))
}
if ws[0].Times == nil || ws[0].Times.WindowClosed == nil || !ws[0].Times.WindowClosed.Equal(t0) {
t.Fatalf("its window closed at %+v; want when it was heard, %s", ws[0].Times, t0)
}
}
// A merge related to the batch assembling joins it and waits for the window, which its coming restarts: by its
// repository, by a module in common (a dependent of what the batch moves), by an edge of the module graph between
// their modules (one that widens nothing), or by its delivery group's head.
func TestARelatedMergeJoinsTheBatchAndWaitsForTheWindow(t *testing.T) {
ctx := t.Context()
for _, c := range []string{"same repository", "a module in common", "an edge of the graph", "the same group head"} {
t.Run(c, func(t *testing.T) {
open := windowed(t)
groupsWaiting(t, "pr/c1")
asksWithPaths(t)
inv := open.inventory
first := repoMerge("one", "c1", t0)
second := lone(repoMerge("two", "c2", t0.Add(10*time.Second)))
why := map[string]string{"same repository": "both change novox/one", "a module in common": "both move two",
"an edge of the graph": "two depends on builder (built-by)",
"the same group head": "both are members of group " + first.Head}[c]
switch c {
case "same repository":
second = lone(repoMerge("one", "c2", t0.Add(10*time.Second)))
case "a module in common":
// two is built on one: a merge of one moves two too.
if err := inv.RegisterModule(ctx, catalogue.Manifest{Module: "two", Version: "1",
Build: &catalogue.Build{On: []catalogue.BuildsOn{{Arg: "BASE", Module: "one"}}}},
inventory.Source{Repository: "novox/two", Seat: "git", Ref: "main", BuiltFrom: "c0", Head: "c0"}); err != nil {
t.Fatal(err)
}
case "an edge of the graph":
// two is built by the build machine: an edge that orders and widens nothing.
if err := inv.RegisterModule(ctx, catalogue.Manifest{Module: "builder", Version: "1",
Claims: []catalogue.Claim{{Name: "mesh-build-machine", Scope: catalogue.ScopeMesh}}},
inventory.Source{Repository: "novox/builder", Seat: "git", Ref: "main", BuiltFrom: "c0", Head: "c0"}); err != nil {
t.Fatal(err)
}
first = repoMerge("builder", "c1", t0)
case "the same group head":
second = fromHead(repoMerge("two", "c2", t0.Add(10*time.Second)), first.Head)
}
hear(t, open, first, t0)
said := captured(t, func() error {
return (following{open: open}).hearMerge(ctx, second, t0.Add(10*time.Second))
})
if !strings.Contains(said, "("+why+")") {
t.Fatalf("the merge was not said related by %q: %s", why, said)
}
ws, bs := walks(t, open)
if len(ws) != 0 || len(bs) != 1 || len(bs[0].Delivery.Merges) != 2 {
t.Fatalf("the related merge: %d walk(s), batches %+v", len(ws), bs)
}
if bs[0].Delivery.Batch.CutAtOnce {
t.Fatal("a batch a related merge joined is cut at once")
}
cutAt(t, open, t0.Add(99*time.Second))
if ws, _ := walks(t, open); len(ws) != 0 {
t.Fatalf("cut before the window the related merge restarted closed: %+v", ws)
}
cutAt(t, open, t0.Add(100*time.Second))
ws, bs = walks(t, open)
if len(ws) != 1 || len(bs) != 0 || len(ws[0].Delivery.Merges) != 2 {
t.Fatalf("after the window: %d walk(s), %d batch(es)", len(ws), len(bs))
}
})
}
}
// A merge from a head whose delivery group has a member still to merge waits for the window, even with no batch
// assembling, and says why; its sibling joins the same batch by their head (ADR 0306 decision 2). A merge from a
// head that names no group shares nothing and is cut at once, as one from no head is.
func TestAMergeOfAGroupWaitsEvenAlone(t *testing.T) {
open := windowed(t)
asksWithPaths(t)
groupsWaiting(t, "feat/x")
hear(t, open, fromHead(repoMerge("one", "c1", t0), "feat/x"), t0)
ws, bs := walks(t, open)
if len(ws) != 0 || len(bs) != 1 || bs[0].Delivery.Batch.CutAtOnce || windowClosed(bs[0], t0) {
t.Fatalf("a group's merge alone: %d walk(s), batches %+v", len(ws), bs)
}
if words := batchWords(bs[0], t0); !strings.HasPrefix(words, "assembling: 90 s left") ||
!strings.Contains(words, "group feat/x has a member still to merge: novox/elsewhere (ready)") {
t.Fatalf("it reads %q", words)
}
hear(t, open, fromHead(repoMerge("two", "c2", t0.Add(5*time.Second)), "feat/x"), t0.Add(5*time.Second))
if _, bs := walks(t, open); len(bs) != 1 || len(bs[0].Delivery.Merges) != 2 {
t.Fatalf("the group's second member did not join the first: %+v", bs)
}
cutAt(t, open, t0.Add(94*time.Second))
if ws, _ := walks(t, open); len(ws) != 0 {
t.Fatalf("cut before its window closed: %+v", ws)
}
cutAt(t, open, t0.Add(95*time.Second))
ws, bs = walks(t, open)
if len(ws) != 1 || len(bs) != 0 || ws[0].CommitOf("novox/one") != "c1" || ws[0].CommitOf("novox/two") != "c2" {
t.Fatalf("at its window: %d walk(s) %+v, %d batch(es)", len(ws), ws, len(bs))
}
}
// A merge from a head that names no group, sharing nothing, with no walk open, is cut at once.
func TestAHeadedMergeInNoGroupIsCutAtOnce(t *testing.T) {
open := windowed(t)
asksWithPaths(t)
hear(t, open, fromHead(repoMerge("one", "c1", t0), "fix/alone"), t0)
if ws, bs := walks(t, open); len(ws) != 1 || len(bs) != 0 || !ws[0].Open() {
t.Fatalf("a headed merge in no group: %d walk(s), %d batch(es)", len(ws), len(bs))
}
}
// When mesh-delivery cannot say a head's group, the merge waits for the window, the safe side, and its batch says
// why.
func TestAGroupNobodyCanSayWaits(t *testing.T) {
open := windowed(t)
asksWithPaths(t)
was := deliveryGroupOf
deliveryGroupOf = func(context.Context, bool, string) ([]groupMember, bool, error) {
return nil, false, errors.New("nothing serves mesh-delivery.groups")
}
t.Cleanup(func() { deliveryGroupOf = was })
hear(t, open, fromHead(repoMerge("one", "c1", t0), "feat/x"), t0)
ws, bs := walks(t, open)
if len(ws) != 0 || len(bs) != 1 || bs[0].Delivery.Batch.CutAtOnce {
t.Fatalf("a merge whose group nobody could say: %d walk(s), batches %+v", len(ws), bs)
}
if words := batchWords(bs[0], t0); !strings.Contains(words, "mesh-delivery did not say whether group feat/x has a "+
"member still to merge (nothing serves mesh-delivery.groups): it waits for the window, the safe side") {
t.Fatalf("it reads %q", words)
}
}
// A member counts as still to merge only while its pull request is open, and not once this controller keeps its
// merge: a group whose other members are merged or closed makes nobody wait.
func TestAGroupsMergedOrClosedMembersMakeNobodyWait(t *testing.T) {
open := windowed(t)
asksWithPaths(t)
ctx := t.Context()
members := []groupMember{{Repository: "novox/one", State: "ready"}, {Repository: "novox/two", State: "published"},
{Repository: "novox/three", State: "stopped"}}
was := deliveryGroupOf
deliveryGroupOf = func(context.Context, bool, string) ([]groupMember, bool, error) { return members, true, nil }
t.Cleanup(func() { deliveryGroupOf = was })
m := fromHead(repoMerge("one", "c1", t0), "feat/x")
if why, err := groupWait(ctx, open.inventory, m, true, false); err != nil || why != "" {
t.Fatalf("the group's other members are merged or closed, yet: %q %v", why, err)
}
members = append(members, groupMember{Repository: "novox/four", State: "checked"})
if why, _ := groupWait(ctx, open.inventory, m, true, false); !strings.Contains(why, "novox/four (checked)") {
t.Fatalf("an open member: %q", why)
}
// novox/four's merge is kept in a batch already: merged, whatever mesh-delivery says yet.
hear(t, open, fromHead(repoMerge("two", "c4", t0), "feat/x"), t0)
members[3].Repository = "novox/two"
if why, _ := groupWait(ctx, open.inventory, m, true, false); why != "" {
t.Fatalf("a member this controller keeps the merge of still waits: %q", why)
}
if got := memberOf("novox/mesh-sdk@d41d255f28ad (held)"); got != (groupMember{Repository: "novox/mesh-sdk", State: "held"}) {
t.Fatalf("a member reads %+v", got)
}
}
// A merge related to two batches assembling folds the newer into the older, which answers every merge of both
// and names nothing twice; the folded one names the batch that took it over.
func TestAMergeRelatedToTwoBatchesFoldsThem(t *testing.T) {
open := windowed(t)
groupsWaiting(t, "feat/a", "feat/b")
asksWithPaths(t)
ctx := t.Context()
hear(t, open, fromHead(repoMerge("one", "c1", t0), "feat/a"), t0)
hear(t, open, fromHead(repoMerge("two", "c2", t0.Add(5*time.Second)), "feat/b"), t0.Add(5*time.Second))
_, bs := walks(t, open)
if len(bs) != 2 {
t.Fatalf("two merges sharing nothing: %d batch(es)", len(bs))
}
older, newer := bs[1], bs[0] // newest first
// Of one's repository and from two's head: related to both.
hear(t, open, fromHead(repoMerge("one", "c3", t0.Add(10*time.Second)), "feat/b"), t0.Add(10*time.Second))
_, bs = walks(t, open)
if len(bs) != 1 || bs[0].ID != older.ID || len(bs[0].Delivery.Merges) != 3 {
t.Fatalf("after a merge related to both: batches %+v; want %s answering three merges", bs, older.ID)
}
folded, err := open.inventory.PlanByID(ctx, newer.ID)
if err != nil || folded.State != inventory.PlanSuperseded || folded.Delivery == nil ||
folded.Delivery.TakenOverBy != older.ID {
t.Fatalf("the newer batch reads %s %+v (%v)", folded.State, folded.Delivery, err)
}
if left, _ := open.inventory.MergesOf(ctx, newer.ID); len(left) != 0 {
t.Fatalf("the folded batch still answers %+v", left)
}
cutAt(t, open, t0.Add(100*time.Second))
ws, bs := walks(t, open)
var cut []inventory.Plan
for _, w := range ws {
if w.State != inventory.PlanSuperseded {
cut = append(cut, w)
}
}
if len(cut) != 1 || len(bs) != 0 || cut[0].ID != older.ID || cut[0].CommitOf("novox/one") != "c3" ||
cut[0].CommitOf("novox/two") != "c2" {
t.Fatalf("the folded batch's walk: %+v, %d batch(es)", cut, len(bs))
}
}
// With a walk open, a batch cut at once still waits for it, said queued behind it (walks stay one at a time, ADR
// 0276 decision 4); it is cut once that walk ended, with no window to wait for.
func TestACutAtOnceBatchQueuesBehindAnOpenWalk(t *testing.T) {
open := windowed(t)
groupsWaiting(t, "pr/c1")
asksWithPaths(t)
hear(t, open, repoMerge("one", "c1", t0), t0)
cutAt(t, open, t0.Add(90*time.Second))
ws, _ := walks(t, open)
if len(ws) != 1 || !ws[0].Open() {
t.Fatalf("no walk open: %+v", ws)
}
first := ws[0]
heard := t0.Add(100 * time.Second)
hear(t, open, lone(repoMerge("two", "c2", heard)), heard)
ws, bs := walks(t, open)
if len(ws) != 1 || len(bs) != 1 || bs[0].State != inventory.PlanQueued || bs[0].Delivery.Batch.Behind != first.ID ||
!bs[0].Delivery.Batch.CutAtOnce {
t.Fatalf("a lone merge during a walk: %d walk(s), batches %+v", len(ws), bs)
}
if !strings.HasPrefix(batchWords(bs[0], heard), "queued behind "+first.ID) {
t.Fatalf("it reads %q", batchWords(bs[0], heard))
}
first.State = inventory.PlanDone
if err := open.inventory.SavePlan(t.Context(), &first); err != nil {
t.Fatal(err)
}
cutAt(t, open, heard.Add(time.Second))
ws, bs = walks(t, open)
if len(ws) != 2 || len(bs) != 0 || ws[0].Commit != "c2" {
t.Fatalf("the queued batch was not cut once the walk ended: %+v %+v", ws, bs)
}
if ws[0].Times == nil || ws[0].Times.WindowClosed == nil || !ws[0].Times.WindowClosed.Equal(heard) {
t.Fatalf("its window closed at %+v; want when its merge was heard, %s", ws[0].Times, heard)
}
}
// A stopped walk's other merges are walked again in a batch they are related to, and a batch of their own when
// they share nothing with the one assembling (ADR 0299, ADR 0304 decision 1): never in a batch of another
// change, whose window the stop would close.
func TestAStopWalksTheOthersAgainInABatchOfTheirOwnWhenTheyShareNothing(t *testing.T) {
open := windowed(t)
groupsWaiting(t, "pr/cat1aaaaaaaa")
endsItsLine(t)
asksRecorded(t)
ctx := t.Context()
cat := catalogueMerge("cat1aaaaaaaa", "app", t0)
one := fromHead(repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second)), cat.Head)
hear(t, open, cat, t0.Add(time.Second))
hear(t, open, one, t0.Add(6*time.Second))
cutAt(t, open, t0.Add(2*time.Minute))
ws, _ := walks(t, open)
walk := ws[0]
hear(t, open, repoMerge("two", "two1cccccccc", t0.Add(3*time.Minute)), t0.Add(3*time.Minute))
_, bs := walks(t, open)
if len(bs) != 1 {
t.Fatalf("the merge heard while the walk ran is in %d batches", len(bs))
}
other := bs[0]
_, said, err := stopWalkWithout(ctx, open.inventory, walk.ID, "mesh-delivery for jochen", "its group stopped",
[]string{"novox/mesh-catalog@" + cat.Commit}, t0.Add(3*time.Minute+10*time.Second))
if err != nil {
t.Fatal(err)
}
if said.WalkedAgainBy == "" || said.WalkedAgainBy == other.ID {
t.Fatalf("walked again by %q; want a batch of its own, not %s", said.WalkedAgainBy, other.ID)
}
b, _ := open.inventory.PlanByID(ctx, other.ID)
if b.Delivery.Batch == nil || b.Delivery.Batch.Closed || len(b.Delivery.Merges) != 1 {
t.Fatalf("the other change's batch reads %+v; want it untouched by the stop", b.Delivery)
}
}
// A catalogue batch's cut folds in every walk waiting for its delivery's word (cutBatch's carry), so what that
// walk moves is the batch's too: a lone merge related to the waiting walk joins the batch assembling and waits for
// its window, rather than being cut at once as sharing nothing with what the batch's own merges move.
func TestALoneMergeRelatedToAWalkTheCutFoldsJoinsTheBatch(t *testing.T) {
open := windowed(t)
groupsWaiting(t, "pr/c3")
asksWithPaths(t)
ctx := t.Context()
if err := open.inventory.RegisterModule(ctx, catalogue.Manifest{Module: "mesh-delivery", Version: "1",
Claims: []catalogue.Claim{{Name: catalogue.DeliverySeat, Scope: catalogue.ScopeMesh}}},
inventory.Source{Repository: "novox/mesh-catalog", Seat: "git", Path: "modules/mesh-delivery", Ref: "main",
BuiltFrom: "c0", Head: "c0"}); err != nil {
t.Fatal(err)
}
if _, err := open.inventory.Assign(ctx, "anchor", "mesh-delivery"); err != nil {
t.Fatal(err)
}
hear(t, open, repoMerge("one", "c1", t0), t0)
cutAt(t, open, t0.Add(90*time.Second))
if ws, _ := walks(t, open); len(ws) != 1 || !ws[0].Waiting() {
t.Fatalf("the walk does not wait for its word: %+v", ws)
}
hear(t, open, repoMerge("three", "c3", t0.Add(2*time.Minute)), t0.Add(2*time.Minute))
hear(t, open, lone(repoMerge("one", "c2", t0.Add(2*time.Minute+5*time.Second))), t0.Add(2*time.Minute+5*time.Second))
_, bs := walks(t, open)
if len(bs) != 1 || len(bs[0].Delivery.Merges) != 2 || bs[0].Delivery.Batch.CutAtOnce {
t.Fatalf("a lone merge related to the waiting walk: batches %+v", bs)
}
}
// A delivery group with a member of each kind (review of #250): a merge on the controller's own path and a
// catalogue merge never share a batch, so the later member cannot join the earlier one's. Kept in a batch of the
// other kind, the earlier member keeps the later one waiting, whatever mesh-delivery says yet — never cut at once
// to walk before it. The two batches are then cut in the order their windows close (cutBatchesHeld: the oldest
// closed batch first, an own-path one first at one look, and the other queued behind a walk that started): the
// earlier member's window closes first, so it walks first, in both orders.
func TestAGroupAcrossBothKindsKeepsItsOrder(t *testing.T) {
for _, ownFirst := range []bool{true, false} {
name := "catalogue member first"
if ownFirst {
name = "controller member first"
}
t.Run(name, func(t *testing.T) {
open := windowed(t)
asksWithPaths(t)
ctx := t.Context()
was := deliveryGroupOf
// mesh-delivery still says both members ready: it marks one merged only once its walk opened.
deliveryGroupOf = func(context.Context, bool, string) ([]groupMember, bool, error) {
return []groupMember{{Repository: "novox/mesh-controller", State: "ready"},
{Repository: "novox/mesh-catalog", State: "ready"}}, true, nil
}
t.Cleanup(func() { deliveryGroupOf = was })
own := fromHead(repoMerge("mesh-controller", "k1", t0), "feat/both-kinds")
cat := fromHead(catalogueMerge("cat1aaaaaaaa", "app", t0), "feat/both-kinds")
first, second := cat, own
if ownFirst {
first, second = own, cat
}
second.MergedAt = t0.Add(10 * time.Second).Format(time.RFC3339Nano)
hear(t, open, first, t0)
hear(t, open, second, t0.Add(10*time.Second))
ws, bs := walks(t, open)
if len(ws) != 0 || len(bs) != 2 {
t.Fatalf("both members heard: %d walk(s), %d batch(es); want two batches, nothing cut", len(ws), len(bs))
}
later := bs[0] // newest first
if later.Delivery.Batch.CutAtOnce || !strings.Contains(later.Delivery.Batch.Waits,
"has a member waiting in a batch of the other kind: novox/"+first.Repo+"@") {
t.Fatalf("the later member's batch reads %+v; want it waiting for the earlier member", later.Delivery.Batch)
}
// The earlier member's window closes first: its batch alone is cut.
cutAt(t, open, t0.Add(90*time.Second))
ws, bs = walks(t, open)
if len(ws) != 1 || !ws[0].Open() || ws[0].CommitOf("novox/"+first.Repo) != first.Commit || len(bs) != 1 ||
bs[0].ID != later.ID {
t.Fatalf("at the earlier window: walks %+v, batches %+v; want the earlier member alone cut", ws, bs)
}
earlier := ws[0]
// The later member's window closes while the earlier walk runs: it queues behind it.
cutAt(t, open, t0.Add(100*time.Second))
if _, bs := walks(t, open); len(bs) != 1 || bs[0].State != inventory.PlanQueued || bs[0].Delivery.Batch.Behind != earlier.ID {
t.Fatalf("the later member's batch does not queue behind the earlier walk: %+v", bs)
}
earlier.State = inventory.PlanDone
if err := open.inventory.SavePlan(ctx, &earlier); err != nil {
t.Fatal(err)
}
cutAt(t, open, t0.Add(101*time.Second))
got, err := open.inventory.PlanByID(ctx, later.ID)
if err != nil || got.Batch() || got.CommitOf("novox/"+second.Repo) != second.Commit {
t.Fatalf("the later member was not walked after the earlier: %s %v", got.State, err)
}
})
}
}
-9
View File
@@ -63,15 +63,6 @@ func run() error {
// 45 §6), so the next controller takes it at once rather than after its age.
defer theLease.release()
// Every walk a verb keeps is said on the bus, whichever verb kept it (novox/hq ADR 0308 decision 1): the save
// that starts or ends a walk keeps its mark, and no later save says it again. The serving controller says its
// own saves.
if args[0] != "serve" {
said, done := sayingWhatAVerbKept(ctx)
inventory.PlanSaved = said
defer done()
}
switch args[0] {
case "build":
return buildCommand(ctx, args[1:])
+5 -20
View File
@@ -885,11 +885,6 @@ func stalledWords(l stalledLine, o conditions.Observation) (headline, explanatio
// within its bound (novox/hq issue 438): mesh-delivery holds no delivery of it, so there is nothing to stop, release
// or close, and no action is offered. The operator's acts are a new commit, which asks the check again, or a look
// at the build queue, where a check that never ran may still wait.
//
// A head whose delivery is held in ready or rejected is said the same way (novox/hq issue 441), and mesh-delivery
// names that state as data (Held). Its check answered once, so the build queue is the wrong place to look: the step
// is a recheck of the delivery, named by its id, which asks the check again (novox/hq issue 451). It is advised, not
// offered as an action, since the controller is not granted the recheck.
func unansweredWords(l stalledLine, o conditions.Observation) (headline, explanation, resolved, needs string,
actions []conditions.Action) {
repository, _, _ := strings.Cut(l.ID, "@")
@@ -904,23 +899,13 @@ func unansweredWords(l stalledLine, o conditions.Observation) (headline, explana
if d, err := time.ParseDuration(l.Bound); err == nil {
limit = ", past its limit of " + humanDuration(d)
}
check := "merge check"
which := "Its merge check"
if said := uncheckedWaitWords(l); len(said) > 0 {
check = "merge check (" + strings.Join(said, ", ") + ")"
which = "Its merge check (" + strings.Join(said, ", ") + ")"
}
which := "Its " + check
switch l.Held {
case "ready", "rejected":
which = "Its delivery is " + l.Held + ", but on the forge its " + check
if o.Resolver == conditions.ResolverOperator {
needs = "recheck its delivery, which asks its check again — mesh-delivery.recheck with id " + l.ID +
" and why — or push a new commit to its branch, " + FromMeshMCPServer
}
default:
if o.Resolver == conditions.ResolverOperator {
needs = "push a new commit to its branch, which asks the check again, or see whether its check still " +
"waits in the build queue: mesh-controller.queue."
}
if o.Resolver == conditions.ResolverOperator {
needs = "push a new commit to its branch, which asks the check again, or see whether its check still waits " +
"in the build queue: mesh-controller.queue."
}
return fmt.Sprintf("%s's merge check has not answered %s", pull, long),
fmt.Sprintf("%s is open on a branch that requires the merge check. %s was asked and has not answered %s%s. "+
-46
View File
@@ -452,49 +452,3 @@ func TestAnUnansweredMergeChecksTimeIsSaidInLocalTime(t *testing.T) {
t.Fatalf("a pending check without its time reads %q", o.Explanation)
}
}
// **An unanswered merge check on a ready or rejected delivery is told to recheck it** (novox/hq issue 451):
// mesh-delivery says such a line too (issue 441) and names the delivery's state as data, `held`. That check already
// answered once, so the build queue is the wrong place to look; a recheck of the delivery, named by its id, asks the
// check again. A line with no delivery held keeps the advice of issue 438, and so does a line from a mesh-delivery
// that does not say `held` yet: the controller never guesses the state from the words.
func TestAnUnansweredMergeCheckOnAReadyOrRejectedDeliveryOffersARecheck(t *testing.T) {
line := func(held string) stalledLine {
var l stalledLine
raw := `{"id":"novox/mesh-controller@c11222026a3b","number":212,"state":"unanswered",` +
`"waiting":["mesh/merge-gate pending since 2026-10-11T00:11:39Z"],"for":"1h1m0s","bound":"1h0m0s",` +
`"h2":"none: its delivery is ` + held + `, a state the operator moves — the operator's",` +
`"says":"novox/mesh-controller#212 is open on main, which requires the merge check, and its delivery is ` +
held + ` here, but on the forge its check has not answered: mesh/merge-gate pending since ` +
`2026-10-11T00:11:39Z. The pull request waits on that status and nothing asks it again. A recheck of the ` +
`delivery asks its check again, as does a new commit on its branch"`
if held != "" {
raw += `,"held":"` + held + `"`
}
if err := json.Unmarshal([]byte(raw+`}`), &l); err != nil {
t.Fatal(err)
}
return l
}
for _, held := range []string{"ready", "rejected"} {
o := stalledObservations([]stalledLine{line(held)})[0]
t.Logf("%s: %s\nneeds: %s", held, o.Explanation, o.Needs)
if !strings.Contains(o.Needs, "mesh-delivery.recheck") || !strings.Contains(o.Needs, "novox/mesh-controller@c11222026a3b") {
t.Fatalf("a %s delivery's unanswered check is not offered its recheck: %q", held, o.Needs)
}
if strings.Contains(o.Needs, "build queue") || strings.Contains(o.Needs, "mesh-controller.queue") {
t.Fatalf("a %s delivery's unanswered check, which answered once, is sent to the build queue: %q", held, o.Needs)
}
if !strings.Contains(o.Explanation, "Its delivery is "+held) {
t.Fatalf("a %s delivery's explanation does not say its state: %q", held, o.Explanation)
}
if len(o.Actions) != 0 {
t.Fatalf("it offers an action the controller is not granted: %+v", o.Actions)
}
}
// No delivery held, or a mesh-delivery that does not say: the advice of issue 438, and no recheck.
o := stalledObservations([]stalledLine{line("")})[0]
if !strings.Contains(o.Needs, "in the build queue: mesh-controller.queue") || strings.Contains(o.Needs, "recheck") {
t.Fatalf("a line with no delivery held says to %q", o.Needs)
}
}
-12
View File
@@ -251,16 +251,6 @@ func resumed(p *inventory.Plan, why string) {
if p.State == inventory.PlanFailed && len(failedIn(*p)) == 0 {
p.State = inventory.PlanBuilding
p.Note = why
goesOnAgain(p)
}
}
// goesOnAgain clears a stop from a walk that walks again: a stop said why the walk ended, and a walk resumed has
// not ended. Kept, it would end a retried walk that fails again as stopped, not failed (novox/hq ADR 0308
// decision 1), and keep the search for the merge that brought the failure from running (searchAfterFailures).
func goesOnAgain(p *inventory.Plan) {
if p.Delivery != nil {
p.Delivery.Stopped, p.Delivery.StoppedWhy = "", ""
}
}
@@ -470,7 +460,6 @@ func retryRollouts(ctx context.Context, open *stores, p *inventory.Plan) (string
said = append(said, m+" to "+strings.Join(again, ", "))
}
p.State = inventory.PlanRolling
goesOnAgain(p)
p.Note = fmt.Sprintf("tier %d retried by hand; sent %s first again", p.Tier, strings.Join(said, "; "))
if err := open.inventory.SavePlan(ctx, p); err != nil {
return "", err
@@ -536,7 +525,6 @@ func retryTierWhole(ctx context.Context, open *stores, p *inventory.Plan, again
}
if p.State == inventory.PlanFailed && len(failedIn(*p)) == 0 {
p.State = inventory.PlanBuilding
goesOnAgain(p)
p.Note = fmt.Sprintf("tier %d retried by hand: %s asked again; %s sent again with the tier", p.Tier,
strings.Join(again, ", "), orNone(strings.Join(resent, ", ")))
}
+1 -1
View File
@@ -591,7 +591,7 @@ func proposeCommand(ctx context.Context, module, node, valuesArg string, clear,
return err
},
routerHere: routerHereIn(open.inventory),
grantHeld: grantHeldIn(open, conn),
grantHeld: grantHeldIn(open),
routerRecord: func(ctx context.Context, id string) (string, error) {
bucket, err := asksRecords(ctx, open.inventory)
if err != nil || bucket == "" {
+14 -30
View File
@@ -249,7 +249,7 @@ func serve(ctx context.Context) (err error) {
// (novox/hq to-be 45 §9).
checkEvents = bus
// And every walk this controller keeps in a new state, for the delivery it walks (novox/hq ADR 0239).
inventory.PlanSaved = func(_ *inventory.Inventory, p inventory.Plan) { sayPlanMoved(ctx, bus, open.inventory, p) }
inventory.PlanSaved = func(p inventory.Plan) { sayPlanMoved(ctx, bus, p) }
if err := server.Checks(following{open}); err != nil {
return err
}
@@ -1084,58 +1084,42 @@ func brokerFirst(names []string, holder string, behind bool) []string {
// send anywhere, and a module running there always put it in its first wave. A digest of the list
// last sent is kept for this (ADR 0043: the list is composed on each push, never kept itself).
func brokerBehind(ctx context.Context, open *stores, names []string) (string, bool, error) {
s, err := brokerListState(ctx, open, names)
return s.Machine, s.Behind, err
}
// brokerListState is brokerBehind with why the list could not be composed, when it could not: the plan's own
// reason, which a remedy names rather than a push that would fail on it (the review of hq issue 353).
func brokerListState(ctx context.Context, open *stores, names []string) (busListState, error) {
inv := open.inventory
holders, err := seatHolders(ctx, inv)
if err != nil {
return busListState{}, err
return "", false, err
}
h, held := holders[theBrokerSeat]
if !held || h.Node == "" {
return busListState{}, nil
return "", false, nil
}
shelf, err := inv.Catalogue(ctx)
if err != nil {
return busListState{}, err
return "", false, err
}
if m, known := shelf[h.Module]; !known || m.BusUsers == "" {
// A holder that is sent no user list carries no grant: nothing to send first.
return busListState{}, nil
return "", false, nil
}
for _, n := range names {
if n == h.Node {
return busListState{Machine: h.Node}, nil
return h.Node, false, nil
}
}
// **What cannot be worked out is behind** (the review of hq issue 353). The list it would be sent cannot be
// composed, so whether it carries the grants a send relies on is not known; it is taken as behind, so a push
// sends that machine first and says why it cannot (`plan` says it too), and a rollout that needs the list
// refuses rather than sending code whose grants the bus may not hold. It said "not behind" until then: a
// send went ahead as if the bus held a list nobody could compose.
list, composeErr := "", error(nil)
plan, _, err := planFor(ctx, open, h.Node)
if err != nil {
composeErr = err
} else if list, _, err = busUserList(ctx, inv, plan.Modules); err != nil {
composeErr = err
// It cannot be worked out: sending it would refuse the whole send, and `plan` says why.
return h.Node, false, nil
}
list, _, err := busUserList(ctx, inv, plan.Modules)
if err != nil {
return h.Node, false, nil
}
sent, err := inv.SentBusUsers(ctx, h.Node)
if err != nil {
return busListState{}, err
return "", false, err
}
return busListState{Machine: h.Node, Behind: listBehind(list, composeErr, sent), Uncomposed: composeErr}, nil
}
// listBehind is whether the bus's machine is behind: the list composed now is not the one last sent, or it
// could not be composed at all — not known is behind, never "not behind".
func listBehind(list string, composeErr error, sentDigest string) bool {
return composeErr != nil || userListBehind(list, sentDigest)
return h.Node, userListBehind(list, sent), nil
}
// userListBehind is whether the user list composed now is not the one last sent, by its digest. An
+1 -1
View File
@@ -80,7 +80,7 @@ func TestAVerbsOwnProcessNamesItsConnection(t *testing.T) {
// A silence acts, so it is still its own process; it dials, and finds no such condition.
_, _ = handlers["conditions"](context.Background(),
json.RawMessage(`{"silence":"bus.nothing.here","for":"1h","why":"a test"}`))
names := closedNames(t, bus, "mesh-controller verb conditions")
names := closedNames(t, bus)
found := false
for _, n := range names {
found = found || n == "mesh-controller verb conditions"
+7 -12
View File
@@ -812,19 +812,14 @@ func backlogCommand(ctx context.Context, sub string, args []string) error {
}
defer open.Close()
if sub == "release-backlog" {
// Said on the bus (novox/hq ADR 0308 decision 1): the release plan starts walking here, and its start is
// said by the verb that opens it.
var p *inventory.Plan
if err := sayingOnTheBus(ctx, open.inventory, func() error {
release, err := open.inventory.HoldPlans(ctx, true)
if err != nil {
return err
}
defer release()
why.record(ctx, "upgrade release-backlog", nil)
p, err = releaseBacklog(ctx, open, link.Caller())
release, err := open.inventory.HoldPlans(ctx, true)
if err != nil {
return err
}); err != nil {
}
defer release()
why.record(ctx, "upgrade release-backlog", nil)
p, err := releaseBacklog(ctx, open, link.Caller())
if err != nil {
return err
}
if p == nil {
+23 -43
View File
@@ -1559,7 +1559,7 @@ func plansCommand(ctx context.Context, args []string) error {
if len(positionals) == 2 && positionals[0] == "go" {
why.record(ctx, "plans go", positionals[1:])
var p inventory.Plan
if err := sayingOnTheBus(ctx, inv, func() (err error) {
if err := sayingOnTheBus(ctx, func() (err error) {
p, err = letGo(ctx, inv, positionals[1], "a person ("+whoAsked()+")", strings.TrimSpace(*why.why))
return err
}); err != nil {
@@ -1571,13 +1571,8 @@ func plansCommand(ctx context.Context, args []string) error {
}
// `retry` (novox/hq ADR 0219): a failed plan's failed builds asked again, and the plan goes on.
if len(positionals) == 2 && positionals[0] == "retry" {
// Said on the bus (novox/hq ADR 0308 decision 1): a retried walk starts again, and its start is said
// by the verb that starts it.
var said string
if err := sayingOnTheBus(ctx, inv, func() (err error) {
said, err = retryPlan(ctx, open, positionals[1])
return err
}); err != nil {
said, err := retryPlan(ctx, open, positionals[1])
if err != nil {
return err
}
fmt.Println(said)
@@ -1591,29 +1586,26 @@ func plansCommand(ctx context.Context, args []string) error {
if positionals[0] == "close" {
how = "closed"
}
// Said on the bus (novox/hq ADR 0308 decision 1): the walk's end, outcome stopped, is said by the verb
// that ends it, so what it touched is not held until its bound.
var p inventory.Plan
if err := sayingOnTheBus(ctx, inv, func() error {
release, err := inv.HoldPlans(ctx, true)
if err != nil {
return err
}
defer release()
if p, err = inv.PlanByID(ctx, positionals[1]); err != nil {
return err
}
if !p.Open() {
return fmt.Errorf("%s is already %s", p.ID, p.State)
}
why.record(ctx, "plans "+positionals[0], positionals[1:])
stopByHand(&p, how, "a person ("+whoAsked()+")", strings.TrimSpace(*why.why))
sayUnsent(&p, func(m string) bool {
u, err := inv.UpgradeOf(ctx, m)
return err == nil && u.RollOut
})
return inv.SavePlan(ctx, &p)
}); err != nil {
release, err := inv.HoldPlans(ctx, true)
if err != nil {
return err
}
defer release()
p, err := inv.PlanByID(ctx, positionals[1])
if err != nil {
return err
}
if !p.Open() {
return fmt.Errorf("%s is already %s", p.ID, p.State)
}
why.record(ctx, "plans "+positionals[0], positionals[1:])
p.State = inventory.PlanFailed
p.Note = how + " by hand at tier " + fmt.Sprint(p.Tier) + ": " + strings.TrimSpace(*why.why)
sayUnsent(&p, func(m string) bool {
u, err := inv.UpgradeOf(ctx, m)
return err == nil && u.RollOut
})
if err := inv.SavePlan(ctx, &p); err != nil {
return err
}
fmt.Printf("%s %s at tier %d of %d; what was asked still builds and registers, nothing further is asked\n",
@@ -1932,15 +1924,3 @@ func forgetDeleted(ctx context.Context, inv *inventory.Inventory, module string,
"`module forget %s`: %v\n", p.ID, module, module, firstLine(err.Error()))
}
}
// stopByHand ends a walk a person stopped or closed (`plans stop|close`): failed, with who stopped it and why
// kept as the delivery's owner keeps its own stop, so the walk ends with outcome stopped, not failed (novox/hq
// ADR 0308 decision 1).
func stopByHand(p *inventory.Plan, how, by, why string) {
p.State = inventory.PlanFailed
p.Note = how + " by hand at tier " + fmt.Sprint(p.Tier) + ": " + why
if p.Delivery == nil {
p.Delivery = &inventory.PlanDelivery{}
}
p.Delivery.Stopped, p.Delivery.StoppedWhy = by, why
}
+3 -11
View File
@@ -84,12 +84,8 @@ func asAProcess(t *testing.T, url string) {
t.Setenv(broker.CertificateVar, "")
}
// closedNames are the names of the connections the bus saw closed, waited for by want's own name: the
// bus can register another connection's close first (the test's own setup connection, say), so returning
// on the first non-empty list names whichever closed first, not necessarily the one being waited for
// (novox/hq issue 543). closedNames instead waits until want itself appears among the closed connections,
// or the deadline passes, and returns whatever names it last saw either way.
func closedNames(t *testing.T, bus *server.Server, want string) []string {
// closedNames are the names of the connections the bus saw closed.
func closedNames(t *testing.T, bus *server.Server) []string {
t.Helper()
deadline := time.Now().Add(5 * time.Second)
var names []string
@@ -99,14 +95,10 @@ func closedNames(t *testing.T, bus *server.Server, want string) []string {
t.Fatal(err)
}
names = names[:0]
found := false
for _, c := range connz.Conns {
names = append(names, c.Name)
if c.Name == want {
found = true
}
}
if found {
if len(names) > 0 {
return names
}
time.Sleep(50 * time.Millisecond)
@@ -15,8 +15,7 @@ import (
func serviceManagerHolder() catalogue.Manifest {
return catalogue.Manifest{Module: "systemd", Version: "1",
Claims: []catalogue.Claim{{Name: catalogue.ServiceManagerSeat, Scope: catalogue.ScopeNode,
Serves: []string{"units", "status", "start", "stop", "restart", "enable", "disable", "journal",
"reset-failed", "wanted-by", "unlink-dangling"}}},
Serves: []string{"units", "status", "start", "stop", "restart", "enable", "disable", "journal"}}},
Resources: []map[string]any{{"id": "systemd", "type": "package", "package": "systemd"}}}
}
@@ -344,58 +344,3 @@ func TestAnyKeyOfAModuleWithATrustedMergeableFileIsRefusedThroughAVerb(t *testin
t.Fatalf("a file marked untrusted through the verb: %v", err)
}
}
// Who owns a module's directory is said at the terminal alone (novox/hq issue 279): a directory's owner may name
// the settings the module runs as, root chowns the folder to them at the next push, and through the settings verb
// any caller could have set puid to 0 or to its own uid. The key is refused through every verb route even where a
// file nothing trusts also asks for it, and taken at the terminal; the layer a refused call meets is unchanged.
func TestADirectoryOwnersKeyIsRefusedThroughAVerb(t *testing.T) {
open := aMesh(t)
ctx := t.Context()
register(t, open, catalogue.Manifest{Module: "sonarr", Version: "1",
Resources: []map[string]any{
{"id": "recycle-bin", "type": "directory", "mode": "0755", "owner": "${setting:puid}:${setting:pgid}"},
{"id": "identity", "type": "file", "path": "/etc/sonarr.env", "mode": "0644", "trusted": false,
"content": "PUID=${setting:puid}\nPGID=${setting:pgid}\nX=${setting:x}\n"}}})
if err := atTheTerminal(t, "settings", "set", "sonarr", `{"puid":1001,"pgid":2000,"x":0}`,
"--node", "anchor"); err != nil {
t.Fatalf("the owner at the terminal: %v", err)
}
if _, err := assign(ctx, open, "anchor", "sonarr"); err != nil {
t.Fatal(err)
}
layer := func() string {
t.Helper()
values, _, err := open.inventory.Layer(ctx, "anchor", "sonarr")
if err != nil {
t.Fatal(err)
}
raw, _ := json.Marshal(values)
return string(raw)
}
kept := layer()
refused := func(what string, err error) {
t.Helper()
if err == nil || !strings.Contains(err.Error(), "controller's terminal") {
t.Fatalf("%s: %v", what, err)
}
}
refused("puid 0 through the verb", throughVerb(t, "settings", map[string]any{"module": "sonarr",
"node": "anchor", "values": `{"puid":0,"pgid":2000,"x":0}`}))
refused("pgid through the verb, mesh-wide", throughVerb(t, "settings", map[string]any{"module": "sonarr",
"values": `{"pgid":0}`}))
refused("cleared through the verb", throughVerb(t, "settings", map[string]any{"module": "sonarr",
"node": "anchor", "clear": "true"}))
if err := throughVerb(t, "command", map[string]any{
"command": `settings set sonarr '{"puid":0,"pgid":2000,"x":0}' --node anchor`}); err == nil {
t.Fatal("the command verb set the owner")
}
if got := layer(); got != kept {
t.Fatalf("a refused call changed the layer: %s, was %s", got, kept)
}
// Another key, the owner kept, is still the verb's.
if err := throughVerb(t, "settings", map[string]any{"module": "sonarr", "node": "anchor",
"values": `{"puid":1001,"pgid":2000,"x":1}`}); err != nil {
t.Fatalf("another key through the verb, the owner kept: %v", err)
}
}
-334
View File
@@ -1,334 +0,0 @@
package main
import (
"context"
"fmt"
"os"
"slices"
"sort"
"time"
"github.com/novox/mesh-controller/internal/broker"
"github.com/novox/mesh-controller/internal/catalogue"
"github.com/novox/mesh-controller/internal/inventory"
)
// What a walk touches, its bound and its outcome (novox/hq ADR 0308 decision 1, decision 5): said in
// `plan-moved` on the save that starts the walk and the save that ends it, beside the walk whole, for the holder
// of the operator's channel to hold what the walk touches while it runs and tell its result once.
//
// **What a walk touches** is every module it moves on every machine it sends it to, and every module a moved
// tool runner (node-tools) or node-engine (mesh-host) carries on that machine: what restarts there. At the start
// that is what the walk will send — each module of its tiers whose policy rolls it out, on every machine running
// it; at the end it is what it reached — the machines each module was sent to, first and rest, and what a gated
// send carried with it.
//
// **Its bound** is twice the largest delivery budget among the walk's deliveries (deliveryBudgets: a core
// module's ten minutes, a leaf module's budget in force), and fifteen minutes, not doubled, for a walk none of
// whose deliveries has a budget: a release plan (builds waiting for a person's gate), a walk whose every module's
// policy records rather than rolls out (held for a person, ADR 0282 decision 1), or a walk kept without a class.
// walkNoBudgetBound is the bound of a walk with no budget (ADR 0308 decision 5): not doubled.
const walkNoBudgetBound = 15 * time.Minute
// budgetNone is the budget class of a walk none of whose deliveries has a budget.
const budgetNone = "none"
// walkSaid is what `plan-moved` adds on a walk's start and end, beside the walk whole.
type walkSaid struct {
Walk string `json:"walk"`
Phase string `json:"phase"`
// Touches is, per machine, the modules the walk moves or reaches there and what a moved tool runner or
// node-engine carries there; never null.
Touches []walkTouch `json:"touches"`
// Outcome is on the end only: delivered, failed or stopped.
Outcome string `json:"outcome,omitempty"`
// BoundSeconds and BudgetClass are said on both, read from the walk's class at the start.
BoundSeconds int `json:"bound_seconds"`
BudgetClass string `json:"budget_class"`
// StartedAt is when the walk was said started, on both; EndedAt when it was said ended, on the end: the
// bound runs from StartedAt, whenever the event is heard.
StartedAt *time.Time `json:"started_at,omitempty"`
EndedAt *time.Time `json:"ended_at,omitempty"`
}
// walkTouch is what a walk touches on one machine.
type walkTouch struct {
Machine string `json:"machine"`
Modules []string `json:"modules"`
// Carried names those of Modules that are there only because a moved tool runner or node-engine carries
// them: what restarts with it, not what the walk builds.
Carried []string `json:"carried,omitempty"`
}
// planMovedBody is `plan-moved`'s body: the walk whole, as before, and on its start and end what it touches.
type planMovedBody struct {
inventory.Plan
*walkSaid
}
// walkFacts is what the touches are read from: the controller's inventory.
type walkFacts interface {
Running(ctx context.Context, module string) ([]string, error)
UpgradeOf(ctx context.Context, module string) (inventory.Upgrade, error)
Assigned(ctx context.Context, node string) ([]string, error)
SentBuilds(ctx context.Context, node string) (map[string]string, bool, error)
Catalogue(ctx context.Context) (map[string]catalogue.Manifest, error)
}
// sayWalk is what `plan-moved` adds to the save that started or ended the walk; nil on every other save. A fact
// that cannot be read is said on the controller's log and left out, never in the way of the event: a walk said
// with fewer touches is held less, never longer.
func sayWalk(ctx context.Context, facts walkFacts, p inventory.Plan) *walkSaid {
if p.WalkSaid == "" {
return nil
}
w := &walkSaid{Walk: p.ID, Phase: p.WalkSaid}
if p.Times != nil {
w.StartedAt = p.Times.Started
if p.WalkSaid == inventory.WalkEnded {
w.EndedAt, w.Outcome = p.Times.Ended, p.Outcome()
}
}
read := readWalkFacts(ctx, facts, p)
w.BudgetClass = walkBudgetClass(p, read.rollsOut)
w.BoundSeconds = int(walkBound(w.BudgetClass) / time.Second)
var moved map[string][]string
if p.WalkSaid == inventory.WalkStarted {
moved = walkSends(p, read.running, read.rollsOut)
} else {
moved = walkReached(p, read.running)
}
var shelf map[string]catalogue.Manifest
if len(moved[broker.RuntimeModule]) > 0 {
// Which modules the tool runner serves is the catalogue's: read only when the walk moves it.
shelf = read.shelf()
}
w.Touches = touchesOf(moved, read.onMachine(moved), shelf)
return w
}
// walkBudgetClass is the largest budget class among the walk's deliveries: its class read at the cut when one of
// its modules rolls out, and none when it has no budget.
func walkBudgetClass(p inventory.Plan, rollsOut func(string) bool) string {
if p.Release != nil || p.Times == nil {
return budgetNone
}
if _, budgeted := deliveryBudgets[p.Times.Class]; !budgeted {
return budgetNone
}
for _, tier := range p.Tiers {
for _, m := range tier {
if rollsOut(m) {
return p.Times.Class
}
}
}
return budgetNone
}
// walkBound is a walk's bound by its budget class: twice its budget, or fifteen minutes with none.
func walkBound(class string) time.Duration {
if b, ok := deliveryBudgets[class]; ok {
return 2 * b
}
return walkNoBudgetBound
}
// walkSends is what a walk will send at its start, per module the machines it goes to: each module of its tiers
// whose policy rolls it out, on every machine running it (for a release plan, those of its order).
func walkSends(p inventory.Plan, running func(string) []string, rollsOut func(string) bool) map[string][]string {
out := map[string][]string{}
for _, tier := range p.Tiers {
for _, m := range tier {
if s := p.Modules[m]; s != nil && s.State == planDeleted {
continue
}
if p.Release == nil && !rollsOut(m) {
continue
}
for _, n := range running(m) {
if p.Release != nil && !slices.Contains(p.Release.Order, n) {
continue
}
out[m] = appendNew(out[m], n)
}
}
}
return out
}
// walkReached is what a walk sent by its end, per module the machines it reached: its first machines, its rest,
// and every module a gated send carried with it. A release plan reached the machines it walked: those done and
// the one being judged.
func walkReached(p inventory.Plan, running func(string) []string) map[string][]string {
out := map[string][]string{}
carried := func(g *inventory.PlanGate) {
if g == nil {
return
}
for _, c := range g.Carried {
out[c.Module] = appendNew(out[c.Module], c.Node)
}
}
if r := p.Release; r != nil {
reached := append([]string{}, r.Done...)
if r.Gate != nil {
reached = appendNew(reached, r.Gate.Machines...)
carried(r.Gate)
}
for _, tier := range p.Tiers {
for _, m := range tier {
for _, n := range running(m) {
if slices.Contains(reached, n) {
out[m] = appendNew(out[m], n)
}
}
}
}
return out
}
for m, s := range p.Modules {
if s == nil {
continue
}
out[m] = appendNew(out[m], s.First...)
for n := range s.Rest {
out[m] = appendNew(out[m], n)
}
carried(s.Gate)
if len(out[m]) == 0 {
delete(out, m)
}
}
return out
}
// touchesOf turns what a walk moves, per module, into what it touches, per machine: the modules it moves there,
// and, where it moves the tool runner, every module there whose tools it serves, and where it moves the
// node-engine, every module there. Sorted by machine and module.
func touchesOf(moved map[string][]string, on map[string][]string, shelf map[string]catalogue.Manifest) []walkTouch {
byMachine := map[string][]string{}
for m, machines := range moved {
for _, n := range machines {
byMachine[n] = appendNew(byMachine[n], m)
}
}
out := make([]walkTouch, 0, len(byMachine))
for n, modules := range byMachine {
t := walkTouch{Machine: n, Modules: modules}
engine, runner := slices.Contains(modules, hostModule), slices.Contains(modules, broker.RuntimeModule)
if engine || runner {
for _, m := range on[n] {
if slices.Contains(t.Modules, m) || (!engine && !servesTools(shelf[m])) {
continue
}
t.Modules = append(t.Modules, m)
t.Carried = append(t.Carried, m)
}
}
sort.Strings(t.Modules)
sort.Strings(t.Carried)
out = append(out, t)
}
sort.Slice(out, func(i, j int) bool { return out[i].Machine < out[j].Machine })
return out
}
// servesTools says the tool runner serves something of this module: a bundle it loads entrypoints from, or the
// module's own tools.
func servesTools(m catalogue.Manifest) bool {
if len(m.Tools) > 0 {
return true
}
for _, b := range m.Bundles {
if len(b.Loads) > 0 {
return true
}
}
return false
}
// walkRead is the facts a walk's touches are worked out from, read once per event.
type walkRead struct {
running func(string) []string
rollsOut func(string) bool
onMachine func(moved map[string][]string) map[string][]string
shelf func() map[string]catalogue.Manifest
}
// readWalkFacts reads what the walk's touches need. Without facts (a test, a command with no store), nothing runs
// anywhere and every module rolls out: the touches are then what the walk itself recorded.
func readWalkFacts(ctx context.Context, facts walkFacts, p inventory.Plan) walkRead {
cantRead := func(what string, err error) {
fmt.Fprintf(os.Stderr, "%s: %s cannot be read for what the walk touches: %v\n", p.ID, what, err)
}
r := walkRead{
running: func(string) []string { return nil },
rollsOut: func(string) bool { return true },
onMachine: func(map[string][]string) map[string][]string { return nil },
shelf: func() map[string]catalogue.Manifest { return nil },
}
if facts == nil {
return r
}
running := map[string][]string{}
r.running = func(m string) []string {
if on, read := running[m]; read {
return on
}
on, err := facts.Running(ctx, m)
if err != nil {
cantRead("where "+m+" runs", err)
}
running[m] = on
return on
}
rolls := map[string]bool{}
r.rollsOut = func(m string) bool {
if yes, read := rolls[m]; read {
return yes
}
u, err := facts.UpgradeOf(ctx, m)
if err != nil {
cantRead(m+"'s policy", err)
}
// A policy that cannot be read is taken to roll out: the walk is then held more, never less.
rolls[m] = err != nil || u.RollOut
return rolls[m]
}
r.onMachine = func(moved map[string][]string) map[string][]string {
out := map[string][]string{}
for _, core := range []string{hostModule, broker.RuntimeModule} {
for _, n := range moved[core] {
if _, read := out[n]; read {
continue
}
var there []string
if sent, known, err := facts.SentBuilds(ctx, n); err != nil {
cantRead("what "+n+" was sent", err)
} else if known {
for m := range sent {
there = append(there, m)
}
}
if assigned, err := facts.Assigned(ctx, n); err != nil {
cantRead("what "+n+" is assigned", err)
} else {
there = appendNew(there, assigned...)
}
sort.Strings(there)
out[n] = there
}
}
return out
}
r.shelf = func() map[string]catalogue.Manifest {
shelf, err := facts.Catalogue(ctx)
if err != nil {
cantRead("the catalogue", err)
}
return shelf
}
return r
}
-514
View File
@@ -1,514 +0,0 @@
package main
import (
"context"
"encoding/json"
"os"
"reflect"
"testing"
"time"
"github.com/nats-io/nats.go"
"github.com/novox/mesh-controller/internal/broker"
"github.com/novox/mesh-controller/internal/catalogue"
"github.com/novox/mesh-controller/internal/inventory"
"github.com/novox/mesh-controller/internal/link"
"github.com/novox/mesh-controller/internal/testbus"
)
// fakeWalkFacts is the inventory as far as a walk's touches read it.
type fakeWalkFacts struct {
running map[string][]string
records map[string]bool // modules whose policy records rather than rolls out
sent map[string]map[string]string
assigned map[string][]string
shelf map[string]catalogue.Manifest
}
func (f fakeWalkFacts) Running(_ context.Context, m string) ([]string, error) {
return f.running[m], nil
}
func (f fakeWalkFacts) UpgradeOf(_ context.Context, m string) (inventory.Upgrade, error) {
return inventory.Upgrade{RollOut: !f.records[m]}, nil
}
func (f fakeWalkFacts) Assigned(_ context.Context, n string) ([]string, error) {
return f.assigned[n], nil
}
func (f fakeWalkFacts) SentBuilds(_ context.Context, n string) (map[string]string, bool, error) {
s, ok := f.sent[n]
return s, ok, nil
}
func (f fakeWalkFacts) Catalogue(context.Context) (map[string]catalogue.Manifest, error) {
return f.shelf, nil
}
// theMesh is four machines: the tool runner on each, messenger and mesh-issues on novox, gitea on ace.
func theMesh() fakeWalkFacts {
return fakeWalkFacts{
running: map[string][]string{
broker.RuntimeModule: {"ace", "g14", "novox", "shanks"},
"messenger": {"novox"},
"gitea": {"ace"},
"plex": {"ace"},
},
sent: map[string]map[string]string{
"novox": {broker.RuntimeModule: "c0", "messenger": "c1", "mesh-issues": "c2", "postgres": "c3"},
"ace": {broker.RuntimeModule: "c0", "gitea": "c4"},
},
assigned: map[string][]string{"g14": {broker.RuntimeModule, "desktop"}, "shanks": {broker.RuntimeModule}},
shelf: map[string]catalogue.Manifest{
"messenger": {Module: "messenger", Tools: []string{"notify"}},
"mesh-issues": {Module: "mesh-issues", Bundles: []catalogue.Bundle{{Name: "tools", Loads: []string{"main"}}}},
"postgres": {Module: "postgres"},
"gitea": {Module: "gitea", Tools: []string{"gitea_api"}},
"desktop": {Module: "desktop", Tools: []string{"notify_send"}},
},
}
}
func walkOfTiers(class string, tiers ...[]string) inventory.Plan {
cut := time.Date(2026, 10, 12, 11, 0, 0, 0, time.UTC)
started := cut.Add(time.Second)
p := inventory.Plan{ID: "plan-308", State: inventory.PlanBuilding, Tiers: tiers,
Modules: map[string]*inventory.PlanModule{}, Times: &inventory.PlanTimes{Cut: &cut, Class: class, Started: &started}}
for _, tier := range tiers {
for _, m := range tier {
p.Modules[m] = &inventory.PlanModule{}
}
}
p.WalkSaid = inventory.WalkStarted
return p
}
// novox/hq ADR 0308 How it is checked §1: the start names the modules the walk moves and the machines it sends
// them to, and every module a moved tool runner carries there.
func TestAWalksStartNamesWhatItTouchesWithWhatTheToolRunnerCarries(t *testing.T) {
p := walkOfTiers(inventory.ClassCore, []string{broker.RuntimeModule}, []string{"messenger"})
w := sayWalk(t.Context(), theMesh(), p)
if w == nil || w.Walk != "plan-308" || w.Phase != inventory.WalkStarted || w.Outcome != "" {
t.Fatalf("the start is said: %+v", w)
}
want := []walkTouch{
{Machine: "ace", Modules: []string{"gitea", broker.RuntimeModule}, Carried: []string{"gitea"}},
{Machine: "g14", Modules: []string{"desktop", broker.RuntimeModule}, Carried: []string{"desktop"}},
// postgres serves no tools: the tool runner does not carry it.
{Machine: "novox", Modules: []string{"mesh-issues", "messenger", broker.RuntimeModule}, Carried: []string{"mesh-issues"}},
{Machine: "shanks", Modules: []string{broker.RuntimeModule}},
}
if !reflect.DeepEqual(w.Touches, want) {
t.Fatalf("touches:\n got %+v\nwant %+v", w.Touches, want)
}
if w.BudgetClass != inventory.ClassCore || w.BoundSeconds != 1200 || w.StartedAt == nil {
t.Fatalf("a core walk is bound at 20 minutes: %+v", w)
}
}
// A moved node-engine carries every module on its machine; a module whose policy records is not sent, and not
// touched.
func TestAMovedNodeEngineCarriesEverythingOnItsMachine(t *testing.T) {
f := theMesh()
f.running[hostModule] = []string{"novox"}
f.records = map[string]bool{"plex": true}
p := walkOfTiers(inventory.ClassCore, []string{hostModule, "plex"})
w := sayWalk(t.Context(), f, p)
want := []walkTouch{{Machine: "novox",
Modules: []string{"mesh-host", "mesh-issues", "messenger", "node-tools", "postgres"},
Carried: []string{"mesh-issues", "messenger", "node-tools", "postgres"}}}
if !reflect.DeepEqual(w.Touches, want) {
t.Fatalf("touches:\n got %+v\nwant %+v", w.Touches, want)
}
}
// The end names what the walk reached — its first machines, its rest, what a gated send carried — and its
// outcome; the modules a reached tool runner carries are touched there too.
func TestAWalksEndNamesWhatItReachedAndItsOutcome(t *testing.T) {
p := walkOfTiers(inventory.ClassCore, []string{broker.RuntimeModule}, []string{"messenger"})
ended := p.Times.Started.Add(4 * time.Minute)
p.Times.Ended = &ended
p.State, p.WalkSaid = inventory.PlanDone, inventory.WalkEnded
p.Modules[broker.RuntimeModule] = &inventory.PlanModule{First: []string{"shanks"},
Rest: map[string]inventory.SentDeclaration{"novox": {}},
Gate: &inventory.PlanGate{Carried: []inventory.CarriedMove{{Module: "gitea", Node: "shanks"}}}}
// messenger was built and found unchanged: no move, not reached.
p.Modules["messenger"] = &inventory.PlanModule{Why: "no move"}
w := sayWalk(t.Context(), theMesh(), p)
if w.Phase != inventory.WalkEnded || w.Outcome != inventory.WalkDelivered || w.EndedAt == nil {
t.Fatalf("the end is said with its outcome: %+v", w)
}
want := []walkTouch{
{Machine: "novox", Modules: []string{"mesh-issues", "messenger", broker.RuntimeModule},
Carried: []string{"mesh-issues", "messenger"}},
{Machine: "shanks", Modules: []string{"gitea", broker.RuntimeModule}},
}
if !reflect.DeepEqual(w.Touches, want) {
t.Fatalf("touches:\n got %+v\nwant %+v", w.Touches, want)
}
for outcome, end := range map[string]func(*inventory.Plan){
inventory.WalkFailed: func(p *inventory.Plan) { p.State = inventory.PlanFailed },
inventory.WalkStopped: func(p *inventory.Plan) {
p.State, p.Delivery = inventory.PlanFailed, &inventory.PlanDelivery{Stopped: "a person"}
},
} {
q := p
end(&q)
if got := sayWalk(t.Context(), theMesh(), q).Outcome; got != outcome {
t.Errorf("outcome %q, want %q", got, outcome)
}
}
}
// novox/hq ADR 0308 How it is checked §1 and decision 5: the bound is twice the largest budget in force among
// the walk's deliveries — 20 minutes for a core walk, twice the leaf budget for a leaf walk — and 15 minutes, not
// doubled, for a walk none of whose deliveries has a budget.
func TestAWalksBoundIsTwiceItsBudgetOrFifteenMinutesWithNone(t *testing.T) {
f := theMesh()
f.records = map[string]bool{"plex": true}
release := walkOfTiers("", []string{"gitea"})
release.Times, release.Release = &inventory.PlanTimes{}, &inventory.PlanRelease{Order: []string{"ace"}}
for name, c := range map[string]struct {
p inventory.Plan
class string
bound int
}{
"core": {walkOfTiers(inventory.ClassCore, []string{"messenger"}), inventory.ClassCore, 1200},
"leaf": {walkOfTiers(inventory.ClassLeaf, []string{"messenger"}), inventory.ClassLeaf, int(2 * deliveryBudgets[inventory.ClassLeaf] / time.Second)},
"held for a person": {walkOfTiers(inventory.ClassLeaf, []string{"plex"}), budgetNone, 900},
"a release": {release, budgetNone, 900},
"kept without a class": {walkOfTiers("", []string{"messenger"}), budgetNone, 900},
} {
w := sayWalk(t.Context(), f, c.p)
if w.BudgetClass != c.class || w.BoundSeconds != c.bound {
t.Errorf("%s: %s bound %ds, want %s %ds", name, w.BudgetClass, w.BoundSeconds, c.class, c.bound)
}
}
// The leaf budget in force is three minutes (ADR 0304 decision 4): a leaf walk is bound at six.
if deliveryBudgets[inventory.ClassLeaf] == 3*time.Minute {
if w := sayWalk(t.Context(), f, walkOfTiers(inventory.ClassLeaf, []string{"messenger"})); w.BoundSeconds != 360 {
t.Errorf("a leaf walk at a three-minute budget is bound at 360 s: %d", w.BoundSeconds)
}
}
}
// plan-moved's body: the walk whole as before, and on its start the walk's fields beside it; on any other save
// none of them.
func TestPlanMovedCarriesTheWalkWholeAndOnItsStartWhatItTouches(t *testing.T) {
p := walkOfTiers(inventory.ClassLeaf, []string{"messenger"})
body, err := json.Marshal(planMovedBody{Plan: p, walkSaid: sayWalk(t.Context(), theMesh(), p)})
if err != nil {
t.Fatal(err)
}
var said map[string]any
if err := json.Unmarshal(body, &said); err != nil {
t.Fatal(err)
}
for _, k := range []string{"id", "state", "tiers", "modules", "times", "walk", "phase", "touches", "bound_seconds", "budget_class", "started_at"} {
if _, ok := said[k]; !ok {
t.Errorf("plan-moved lacks %q: %s", k, body)
}
}
if _, ok := said["outcome"]; ok {
t.Errorf("a start says no outcome: %s", body)
}
if said["walk"] != "plan-308" || said["phase"] != "started" || said["budget_class"] != "leaf" {
t.Errorf("the walk's fields: %s", body)
}
p.WalkSaid = ""
body, _ = json.Marshal(planMovedBody{Plan: p, walkSaid: sayWalk(t.Context(), theMesh(), p)})
said = map[string]any{}
_ = json.Unmarshal(body, &said)
for _, k := range []string{"walk", "phase", "touches", "bound_seconds"} {
if _, ok := said[k]; ok {
t.Errorf("a save neither starting nor ending the walk says %q: %s", k, body)
}
}
// A walk that touches nothing says an empty list, never null.
q := walkOfTiers(inventory.ClassLeaf, []string{"nobody-runs-this"})
body, _ = json.Marshal(planMovedBody{Plan: q, walkSaid: sayWalk(t.Context(), theMesh(), q)})
said = map[string]any{}
_ = json.Unmarshal(body, &said)
if touches, ok := said["touches"].([]any); !ok || len(touches) != 0 {
t.Errorf("touches nothing: %s", body)
}
}
// Live, with the store and a bus: a walk let go by the delivery's verb is said started with what it touches and
// its bound, and stopped by the verb is said ended, stopped (novox/hq ADR 0308 How it is checked §1).
func TestAWalkLetGoAndStoppedByVerbsIsSaidStartedAndEnded(t *testing.T) {
open := aMesh(t)
ctx := t.Context()
next := planMovedHeard(t)
register(t, open, catalogue.Manifest{Module: "app", Version: "1"})
if _, err := assign(ctx, open, "anchor", "app"); err != nil {
t.Fatal(err)
}
cut := time.Now().UTC()
waiting := inventory.Plan{ID: "plan-308-live", Repository: "novox/mesh-catalog", Branch: "main", Commit: "c8c8c8c8",
Created: cut, State: inventory.PlanBuilding, Tiers: [][]string{{"app"}},
Modules: map[string]*inventory.PlanModule{"app": {}},
Delivery: &inventory.PlanDelivery{Awaits: catalogue.DeliverySeat},
Times: &inventory.PlanTimes{Cut: &cut, Class: inventory.ClassLeaf}}
if err := open.inventory.SavePlan(ctx, &waiting); err != nil {
t.Fatal(err)
}
if err := deliveryCommand(ctx, []string{"go", waiting.ID, "--why", "its turn"}); err != nil {
t.Fatal(err)
}
started := next(inventory.WalkStarted)
touches, _ := json.Marshal(started["touches"])
if started["walk"] != waiting.ID || string(touches) != `[{"machine":"anchor","modules":["app"]}]` ||
started["budget_class"] != inventory.ClassLeaf ||
started["bound_seconds"] != float64(2*deliveryBudgets[inventory.ClassLeaf]/time.Second) {
t.Fatalf("the start: %v", started)
}
if err := deliveryCommand(ctx, []string{"stop", waiting.ID, "--why", "a test", "--by", "a person"}); err != nil {
t.Fatal(err)
}
ended := next(inventory.WalkEnded)
touches, _ = json.Marshal(ended["touches"])
if ended["walk"] != waiting.ID || ended["outcome"] != inventory.WalkStopped || string(touches) != `[]` {
t.Fatalf("the end, which sent nothing: %v", ended)
}
}
// planMovedHeard raises a bus of the test's own, points the verbs' hand-act connection at it, and returns what
// waits for the next plan-moved said with the given phase, failing the test when none comes within 5 s.
func planMovedHeard(t *testing.T) func(phase string) map[string]any {
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)
}
before := handActConn
handActConn = js.Conn()
t.Cleanup(func() { handActConn = before })
heard := make(chan *nats.Msg, 64)
sub, err := js.Conn().ChanSubscribe(link.SeatEventSubject(link.MeshControllerSeat, link.KeyPlanMoved), heard)
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = sub.Unsubscribe() })
return func(phase string) map[string]any {
t.Helper()
deadline := time.After(5 * time.Second)
for {
select {
case m := <-heard:
var said map[string]any
if err := json.Unmarshal(m.Data, &said); err != nil {
t.Fatal(err)
}
if said["phase"] == phase {
return said
}
case <-deadline:
t.Fatalf("no walk was said %s", phase)
}
}
}
}
// Live, with the store and a bus: a walk a person stops with `plans stop` is said ended, outcome stopped, by the
// verb itself, and is kept stopped by that person; `plans close` ends one the same way (novox/hq ADR 0308
// decision 1, review 15 on mesh-controller#247).
func TestAWalkStoppedOrClosedByAPersonIsSaidEndedStopped(t *testing.T) {
open := aMesh(t)
ctx := t.Context()
next := planMovedHeard(t)
register(t, open, catalogue.Manifest{Module: "app", Version: "1"})
if _, err := assign(ctx, open, "anchor", "app"); err != nil {
t.Fatal(err)
}
for _, verb := range []string{"stop", "close"} {
cut := time.Now().UTC()
walk := inventory.Plan{ID: "plan-308-" + verb, Repository: "novox/mesh-catalog", Branch: "main",
Commit: "c8c8c8c8", Created: cut, State: inventory.PlanBuilding, Tiers: [][]string{{"app"}},
Modules: map[string]*inventory.PlanModule{"app": {}},
Times: &inventory.PlanTimes{Cut: &cut, Started: &cut, Class: inventory.ClassLeaf}}
if err := open.inventory.SavePlan(ctx, &walk); err != nil {
t.Fatal(err)
}
if err := plansCommand(ctx, []string{verb, walk.ID, "--why", "a test"}); err != nil {
t.Fatal(err)
}
ended := next(inventory.WalkEnded)
if ended["walk"] != walk.ID || ended["outcome"] != inventory.WalkStopped {
t.Fatalf("plans %s: the end is %v", verb, ended)
}
p, err := open.inventory.PlanByID(ctx, walk.ID)
if err != nil {
t.Fatal(err)
}
if p.Outcome() != inventory.WalkStopped || p.Delivery == nil || p.Delivery.StoppedWhy != "a test" {
t.Fatalf("plans %s kept the walk %s, %+v", verb, p.Outcome(), p.Delivery)
}
}
}
// Live, with the store and a bus: a failed walk retried with `plans retry` starts again, and the verb says its
// start (novox/hq ADR 0308 decision 1, review 15 on mesh-controller#247).
func TestAWalkRetriedByAPersonIsSaidStartedAgain(t *testing.T) {
open := aMesh(t)
ctx := t.Context()
asksRecorded(t)
next := planMovedHeard(t)
twoTiers(t, open)
before := time.Now().UTC().Add(-time.Hour)
ended := before.Add(time.Minute)
failed := inventory.Plan{ID: "plan-308-retry", Repository: "novox/a", Branch: "main", Commit: "c0ffee",
Created: before, State: inventory.PlanFailed, Tier: 0, Tiers: [][]string{{"a"}, {"b"}},
Note: "a failed to build in tier 0",
Modules: map[string]*inventory.PlanModule{"a": {State: "failed", AskedAt: &before, Build: "build-1",
Why: link.KilledByHand}},
Times: &inventory.PlanTimes{Cut: &before, Started: &before, Ended: &ended, Class: inventory.ClassLeaf}}
if err := open.inventory.SavePlan(ctx, &failed); err != nil {
t.Fatal(err)
}
if err := plansCommand(ctx, []string{"retry", failed.ID}); err != nil {
t.Fatal(err)
}
started := next(inventory.WalkStarted)
if started["walk"] != failed.ID || started["budget_class"] != inventory.ClassLeaf {
t.Fatalf("the retried start: %v", started)
}
}
// Live, with the store and a bus: the release plan `upgrade release-backlog` opens is said started by the verb
// (novox/hq ADR 0308 decision 1, review 15 on mesh-controller#247).
func TestABacklogReleasedByAPersonIsSaidStarted(t *testing.T) {
b := aBacklog(t)
next := planMovedHeard(t)
if err := backlogCommand(t.Context(), "release-backlog", []string{"--why", "a test"}); err != nil {
t.Fatal(err)
}
started := next(inventory.WalkStarted)
if p := b.release(t); started["walk"] != p.ID {
t.Fatalf("the release's start: %v", started)
}
}
// asTyped runs a verb as the controller's command line runs it, through run, as a person's terminal would.
func asTyped(t *testing.T, args ...string) error {
t.Helper()
was := os.Args
os.Args = append([]string{"mesh-controller"}, args...)
t.Cleanup(func() { os.Args = was })
err := run()
// run gives back the lease as the process would end; the test's process goes on, and takes it again.
theLease.mu.Lock()
theLease.held, theLease.kv, theLease.noBus = nil, nil, false
theLease.mu.Unlock()
return err
}
// aStartedWalkAsking keeps a live walk, said started, whose first tier asked a as build.
func aStartedWalkAsking(t *testing.T, open *stores, id, build string, asked time.Time) inventory.Plan {
t.Helper()
walk := inventory.Plan{ID: id, Repository: "novox/a", Branch: "main", Commit: "c0ffee", Created: asked,
State: inventory.PlanBuilding, Tiers: [][]string{{"a"}, {"b"}},
Modules: map[string]*inventory.PlanModule{"a": {State: "asked", AskedAt: &asked, Build: build}},
Times: &inventory.PlanTimes{Cut: &asked, Started: &asked, Class: inventory.ClassLeaf}}
if err := open.inventory.SavePlan(t.Context(), &walk); err != nil {
t.Fatal(err)
}
return walk
}
// Live, with the store and a bus: `cancel <build>` and `clear` fail the walk that asked the build, and the verb
// says its end, failed (novox/hq ADR 0308 decision 1, review 21 on mesh-controller#247).
func TestAWalkEndedByCancelOrClearIsSaidEnded(t *testing.T) {
_, ask := aBuildQueue(t)
open := aMesh(t)
next := planMovedHeard(t)
twoTiers(t, open)
for _, verb := range [][]string{{"cancel"}, {"clear"}} {
id := link.NewBuildID(time.Now())
ask(id, "https://forge.example/novox/a.git")
asked, _ := link.BuildAskedAt(id)
walk := aStartedWalkAsking(t, open, "plan-308-"+verb[0], id, asked)
args := verb
if verb[0] == "cancel" {
args = append(args, id)
}
if err := asTyped(t, args...); err != nil {
t.Fatal(err)
}
ended := next(inventory.WalkEnded)
if ended["walk"] != walk.ID || ended["outcome"] != inventory.WalkFailed {
t.Fatalf("%s: the end is %v", verb[0], ended)
}
// Ended, and so the next walk may be cut on the same repository.
p, _ := open.inventory.PlanByID(t.Context(), walk.ID)
if p.State != inventory.PlanFailed {
t.Fatalf("%s left the walk %s", verb[0], p.State)
}
}
}
// Live, with the store and a bus: `rebuild <module>` joining a failed walk sets it building again, and the verb
// says its start (novox/hq ADR 0308 decision 1, review 21 on mesh-controller#247).
func TestAWalkRestartedByARebuildIsSaidStarted(t *testing.T) {
open := aMesh(t)
asksRecorded(t)
next := planMovedHeard(t)
twoTiers(t, open)
before := time.Now().UTC().Add(-time.Hour)
ended := before.Add(time.Minute)
failed := inventory.Plan{ID: "plan-308-rebuild", Repository: "novox/a", Branch: "main", Commit: "c0ffee",
Created: before, State: inventory.PlanFailed, Tiers: [][]string{{"a"}, {"b"}},
Modules: map[string]*inventory.PlanModule{"a": {State: "failed", AskedAt: &before, Build: "build-1",
Why: link.CancelledByHand}},
Times: &inventory.PlanTimes{Cut: &before, Started: &before, Ended: &ended, Class: inventory.ClassLeaf}}
if err := open.inventory.SavePlan(t.Context(), &failed); err != nil {
t.Fatal(err)
}
if err := asTyped(t, "rebuild", "a"); err != nil {
t.Fatal(err)
}
if started := next(inventory.WalkStarted); started["walk"] != failed.ID {
t.Fatalf("the rebuilt walk's start: %v", started)
}
}
// Live, with the store: a walk a person stopped and then retried, by `plans retry` or by `rebuild`, walks again
// without the stop, so when it fails again it ends failed, not stopped, and the search for the merge that
// brought the failure is not kept from it (novox/hq ADR 0308 decision 1, review 21 on mesh-controller#247).
func TestAWalkStoppedByHandAndRetriedEndsFailedWhenItFailsAgain(t *testing.T) {
open := aMesh(t)
ctx := t.Context()
asksRecorded(t)
twoTiers(t, open)
for _, again := range [][]string{{"plans", "retry"}, {"rebuild", "a"}} {
before := time.Now().UTC().Add(-time.Hour)
walk := aStartedWalkAsking(t, open, "plan-308-again-"+again[0], "build-1", before)
if err := asTyped(t, "plans", "stop", walk.ID, "--why", "a test"); err != nil {
t.Fatal(err)
}
// The build asked before the stop fails after it.
p, _ := open.inventory.PlanByID(ctx, walk.ID)
p.Modules["a"].State, p.Modules["a"].Why = "failed", "it broke"
if err := open.inventory.SavePlan(ctx, &p); err != nil {
t.Fatal(err)
}
args := again
if again[0] == "plans" {
args = append(append([]string(nil), again...), walk.ID)
}
if err := asTyped(t, args...); err != nil {
t.Fatal(err)
}
p, _ = open.inventory.PlanByID(ctx, walk.ID)
if p.State != inventory.PlanBuilding || (p.Delivery != nil && p.Delivery.Stopped != "") {
t.Fatalf("%s: the walk is %s and keeps its stop: %+v", again[0], p.State, p.Delivery)
}
asking, _ := link.BuildAskedAt(p.Modules["a"].Build)
planBuilt(ctx, open, "a", "c0ffee", "it broke again", asking, p.Modules["a"].Build)
if p, _ = open.inventory.PlanByID(ctx, walk.ID); p.State != inventory.PlanFailed || p.Outcome() != inventory.WalkFailed {
t.Fatalf("%s: failing again, the walk is %s and ends %s", again[0], p.State, p.Outcome())
}
}
}
+1 -4
View File
@@ -262,12 +262,9 @@ func MaySubscribe(perms Permissions, subject string) bool {
// delivery held past its bound, and calls them on the operator's warrant, with its why.
//
// And, since novox/hq ADR 0282, `times`: the self-check reads the delivery times to say a delivery over its budget.
//
// And, since novox/hq ADR 0306, `groups`: a merge heard from a head waits for the window while its delivery group
// has a member still to merge.
var VerbsTheControllerAsksTheDeliveryOwner = []SeatVerb{{Seat: "mesh-delivery", Verb: "stalled"},
{Seat: "mesh-delivery", Verb: "close"}, {Seat: "mesh-delivery", Verb: "release"}, {Seat: "mesh-delivery", Verb: "stop"},
{Seat: "mesh-delivery", Verb: "times"}, {Seat: "mesh-delivery", Verb: "groups"}}
{Seat: "mesh-delivery", Verb: "times"}}
// VerbsTheControllerActsOnAWarrant are the other seat verbs the controller calls when the operator's warrant
// chooses them (novox/hq ADR 0259): a machine's service restarted, and a walk started or stopped through the
+1 -1
View File
@@ -24,7 +24,7 @@ accounts {
jetstream: enabled
users = [
{ 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_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.*"] }
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.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"] }
allow_responses: { max: 1, ttl: "1m" }
} }
+13 -2
View File
@@ -971,8 +971,19 @@ func one(ctx context.Context, run Runner, publish Publisher,
return catalogue.Built{Name: a.Name, Kind: a.Kind, Reference: reference}, nil
case catalogue.ArtifactArchive:
src.ownHas(cleanEntry(a.From) + "/**")
body, err := pack(filepath.Join(tree, a.From))
var body []byte
var err error
if a.URL != "" {
// An upstream file pinned by URL and sha256 (novox/hq ADR 0316): its build source is the pin, which
// is in module.json, always part of it.
body, err = upstreamArchive(ctx, workspace, a, say)
if err != nil {
return catalogue.Built{}, fmt.Errorf("%s: %s: %w", module, a.Name, err)
}
} else {
src.ownHas(cleanEntry(a.From) + "/**")
body, err = pack(filepath.Join(tree, a.From))
}
if err != nil {
return catalogue.Built{}, fmt.Errorf("%s: packing %s failed: %w", module, a.Name, err)
}
+326
View File
@@ -0,0 +1,326 @@
package builder
import (
"archive/tar"
"bufio"
"bytes"
"compress/bzip2"
"compress/gzip"
"context"
"crypto/sha256"
"encoding/hex"
"errors"
"fmt"
"io"
"net/http"
"os"
"os/exec"
"path"
"path/filepath"
"strings"
"time"
"github.com/novox/mesh-controller/internal/catalogue"
)
// An archive whose source is somebody else's published file, pinned by URL and sha256 (novox/hq ADR 0316
// §2, §6, to-be 56 §7): a mail client's release, an add-on's xpi.
//
// **Fetched by the build seat, checked against the pin before anything is read from it, and packed as any
// archive is packed.** What a machine receives is the mesh's own archive from the artifact store, by digest,
// exactly as for a directory of the repository; no machine ever asks the vendor. A file whose bytes are not
// the pin's fails the build and names both digests, because the address serving something else is the one
// thing the pin exists to catch.
//
// **Kept in the build seat's workspace under its sha256**, so it is fetched once: a vendor that serves only
// its newest version at one address (Owl) would otherwise break every later build of the same pin.
// maxUpstream is the largest upstream file fetched, and the largest a release may unpack to: the node-engine
// unpacks no archive larger than this either (mesh-host internal/apply/archive.go, maxArchive).
const maxUpstream = 512 << 20
// maxEntries is the most entries an upstream archive may hold: millions of empty ones would fill the build
// seat's inodes and the node's. Betterbird's release holds 64.
const maxEntries = 100_000
// noDowngrade refuses a redirect off https: the digest still guards the bytes, and a fetch in the clear says
// to whoever is on the path what the build seat fetches.
func noDowngrade(req *http.Request, via []*http.Request) error {
if req.URL.Scheme != "https" {
return fmt.Errorf("%s redirected to %s, which is not https", via[0].URL, req.URL)
}
if len(via) >= 10 {
return errors.New("more than 10 redirects")
}
return nil
}
// upstreamClient fetches upstream files: a variable so a test serves its own.
var upstreamClient = &http.Client{Timeout: 15 * time.Minute}
// upstreamDir is where the build seat keeps the upstream files it fetched, by sha256.
func upstreamDir(workspace string) string { return filepath.Join(workspace, "upstream") }
// upstreamArchive is the packed archive of an artifact pinned by URL: the file fetched (or kept from an
// earlier build) and checked, then laid out as the artifact says and packed.
func upstreamArchive(ctx context.Context, workspace string, a catalogue.Artifact,
say func(step, format string, args ...any)) ([]byte, error) {
file, err := upstreamFile(ctx, workspace, a, say)
if err != nil {
return nil, err
}
laid := filepath.Join(workspace, "upstream-"+a.Name)
if err := os.RemoveAll(laid); err != nil {
return nil, err
}
defer os.RemoveAll(laid)
if err := os.MkdirAll(laid, 0o755); err != nil {
return nil, err
}
if a.File != "" {
body, err := os.ReadFile(file)
if err != nil {
return nil, err
}
if err := os.WriteFile(filepath.Join(laid, a.File), body, 0o644); err != nil {
return nil, err
}
} else if err := unpackUpstream(ctx, file, laid, a.Within); err != nil {
return nil, fmt.Errorf("%s (%s): %w", a.URL, a.Version, err)
}
return pack(laid)
}
// upstreamFile is the path of the pinned file in the workspace: kept from an earlier build when its bytes
// are still the pin's, else fetched now and kept only when they are.
func upstreamFile(ctx context.Context, workspace string, a catalogue.Artifact,
say func(step, format string, args ...any)) (string, error) {
want := strings.ToLower(a.SHA256)
dir := upstreamDir(workspace)
kept := filepath.Join(dir, want)
if got, err := sumOf(kept); err == nil {
if got == want {
say("upstream", "%s %s: kept from an earlier build (sha256:%s)", a.Name, a.Version, short(want))
return kept, nil
}
// Changed on disk since it was kept: fetched again rather than trusted.
os.Remove(kept)
}
if err := os.MkdirAll(dir, 0o755); err != nil {
return "", err
}
say("upstream", "fetching %s (%s)", a.URL, a.Version)
request, err := http.NewRequestWithContext(ctx, http.MethodGet, a.URL, nil)
if err != nil {
return "", err
}
client := *upstreamClient
client.CheckRedirect = noDowngrade
response, err := client.Do(request)
if err != nil {
return "", fmt.Errorf("cannot fetch %s: %w", a.URL, err)
}
defer response.Body.Close()
if response.StatusCode != http.StatusOK {
return "", fmt.Errorf("%s answered %s", a.URL, response.Status)
}
part, err := os.CreateTemp(dir, want+".*.part")
if err != nil {
return "", err
}
defer os.Remove(part.Name())
hash := sha256.New()
n, err := io.Copy(io.MultiWriter(part, hash), io.LimitReader(response.Body, maxUpstream+1))
if cerr := part.Close(); err == nil {
err = cerr
}
if err != nil {
return "", fmt.Errorf("fetching %s: %w", a.URL, err)
}
if n > maxUpstream {
return "", fmt.Errorf("%s is larger than %d bytes, more than a machine unpacks; nothing was kept", a.URL, maxUpstream)
}
got := hex.EncodeToString(hash.Sum(nil))
if got != want {
return "", fmt.Errorf("%s was pinned as sha256:%s and what it served is sha256:%s (%d bytes); nothing "+
"was kept. The address serves another file than the one reviewed: compare it, then move the pin "+
"in a reviewed change (novox/hq ADR 0316)", a.URL, want, got, n)
}
if err := os.Rename(part.Name(), kept); err != nil {
return "", err
}
say("upstream", "%s %s: %d bytes, sha256:%s as pinned", a.Name, a.Version, n, short(want))
return kept, nil
}
// sumOf is the sha256 of a file, in hexadecimal.
func sumOf(file string) (string, error) {
f, err := os.Open(file)
if err != nil {
return "", err
}
defer f.Close()
hash := sha256.New()
if _, err := io.Copy(hash, f); err != nil {
return "", err
}
return hex.EncodeToString(hash.Sum(nil)), nil
}
// unpackUpstream unpacks a tar archive, compressed with gzip, bzip2 or xz (told by its first bytes), into a
// directory: of its entries only those within `within` (all, when empty), with that prefix taken off.
//
// **Only files and directories, and nothing outside.** A link, a device or an entry naming a path outside
// the archive, or outside `within`, refuses the build by name: the node-engine unpacks only files and
// directories, and an archive quietly unpacked without what it held would be a release that is not the one
// pinned.
func unpackUpstream(ctx context.Context, file, into, within string) error {
f, err := os.Open(file)
if err != nil {
return err
}
defer f.Close()
buffered := bufio.NewReader(f)
magic, _ := buffered.Peek(6)
var stream io.Reader
var wait func() error
switch {
case bytes.HasPrefix(magic, []byte{0x1f, 0x8b}):
zipped, err := gzip.NewReader(buffered)
if err != nil {
return err
}
defer zipped.Close()
stream = zipped
case bytes.HasPrefix(magic, []byte("BZh")):
stream = bzip2.NewReader(buffered)
case bytes.HasPrefix(magic, []byte{0xfd, '7', 'z', 'X', 'Z', 0x00}):
// Go's standard library reads no xz; the build seat's image carries xz (cmd/mesh-builder/Dockerfile).
if _, err := exec.LookPath("xz"); err != nil {
return errors.New("it is compressed with xz, and this build seat has no xz to read it")
}
cmd := exec.CommandContext(ctx, "xz", "-dc")
cmd.Stdin = buffered
var stderr bytes.Buffer
cmd.Stderr = &stderr
out, err := cmd.StdoutPipe()
if err != nil {
return err
}
if err := cmd.Start(); err != nil {
return err
}
stream = out
wait = func() error {
if err := cmd.Wait(); err != nil {
return fmt.Errorf("xz could not read it: %v %s", err, strings.TrimSpace(stderr.String()))
}
return nil
}
defer func() {
if wait != nil {
_ = cmd.Process.Kill()
_ = cmd.Wait()
}
}()
default:
return errors.New("it is not a tar archive compressed with gzip, bzip2 or xz; a single file is kept " +
"whole by naming it in `file`")
}
if err := untar(stream, into, within); err != nil {
return err
}
if wait != nil {
// Whatever follows the tar's end is read before xz is waited for: xz blocks on a full pipe otherwise.
if _, err := io.Copy(io.Discard, stream); err != nil {
return err
}
w := wait
wait = nil
return w()
}
return nil
}
// untar writes a tar stream's files and directories within `within` into a directory.
func untar(stream io.Reader, into, within string) error {
reader := tar.NewReader(stream)
prefix := ""
if within != "" {
prefix = path.Clean(within) + "/"
}
var total int64
files, entries := 0, 0
for {
header, err := reader.Next()
if err == io.EOF {
break
}
if err != nil {
return err
}
if entries++; entries > maxEntries {
return fmt.Errorf("it holds more than %d entries", maxEntries)
}
if header.Typeflag == tar.TypeXGlobalHeader {
// A pax global header (git archive writes one) names no file.
continue
}
name := path.Clean(strings.TrimPrefix(header.Name, "./"))
if path.IsAbs(name) || name == ".." || strings.HasPrefix(name, "../") {
return fmt.Errorf("%s names a path outside the archive", header.Name)
}
if name == "." || name+"/" == prefix ||
(header.Typeflag == tar.TypeDir && strings.HasPrefix(prefix, name+"/")) {
// The archive's root, the directory taken, or one of its parents.
continue
}
if prefix != "" {
if !strings.HasPrefix(name, prefix) {
return fmt.Errorf("%s is outside %s, the directory the artifact is taken from", header.Name, within)
}
name = strings.TrimPrefix(name, prefix)
}
target := filepath.Join(into, filepath.FromSlash(name))
switch header.Typeflag {
case tar.TypeDir:
if err := os.MkdirAll(target, 0o755); err != nil {
return err
}
case tar.TypeReg:
total += header.Size
if total > maxUpstream {
return fmt.Errorf("it unpacks to more than %d bytes, more than a machine unpacks", maxUpstream)
}
if err := os.MkdirAll(filepath.Dir(target), 0o755); err != nil {
return err
}
mode := os.FileMode(0o644)
if header.Mode&0o111 != 0 {
mode = 0o755
}
out, err := os.OpenFile(target, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, mode)
if err != nil {
return err
}
if _, err := io.Copy(out, io.LimitReader(reader, header.Size)); err != nil {
out.Close()
return err
}
if err := out.Close(); err != nil {
return err
}
files++
default:
return fmt.Errorf("%s is a %c, and an archive carries only files and directories", header.Name,
header.Typeflag)
}
}
if files == 0 {
if within != "" {
return fmt.Errorf("it holds no file within %s", within)
}
return errors.New("it holds no file")
}
return nil
}
+385
View File
@@ -0,0 +1,385 @@
package builder
import (
"archive/tar"
"bytes"
"compress/gzip"
"context"
"crypto/sha256"
"encoding/hex"
"fmt"
"io"
"net"
"net/http"
"net/http/httptest"
"os"
"os/exec"
"path/filepath"
"sort"
"strings"
"testing"
"time"
"github.com/novox/mesh-controller/internal/catalogue"
)
// An archive pinned by URL and sha256 (novox/hq ADR 0316 §2, §6; How it is checked: "an upstream artifact
// whose download does not match its sha256 fails the build and names both digests").
type tarEntry struct {
name, body string
mode int64
kind byte
}
func tarOf(t *testing.T, entries []tarEntry) []byte {
t.Helper()
var raw bytes.Buffer
w := tar.NewWriter(&raw)
for _, e := range entries {
kind := e.kind
if kind == 0 {
kind = tar.TypeReg
}
h := &tar.Header{Name: e.name, Mode: e.mode, Typeflag: kind, Size: int64(len(e.body))}
if kind == tar.TypeSymlink {
h.Linkname, h.Size = "/usr/share/hunspell", 0
}
if kind == tar.TypeDir {
h.Size = 0
}
if err := w.WriteHeader(h); err != nil {
t.Fatal(err)
}
if kind == tar.TypeReg {
if _, err := w.Write([]byte(e.body)); err != nil {
t.Fatal(err)
}
}
}
if err := w.Close(); err != nil {
t.Fatal(err)
}
return raw.Bytes()
}
func gzipped(t *testing.T, body []byte) []byte {
t.Helper()
var out bytes.Buffer
z := gzip.NewWriter(&out)
if _, err := z.Write(body); err != nil {
t.Fatal(err)
}
if err := z.Close(); err != nil {
t.Fatal(err)
}
return out.Bytes()
}
func sumHex(b []byte) string {
s := sha256.Sum256(b)
return hex.EncodeToString(s[:])
}
// vendor serves files by path over https and counts what it was asked.
type vendor struct {
files map[string][]byte
asked int
}
func serveVendor(t *testing.T, v *vendor) string {
t.Helper()
server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
v.asked++
body, ok := v.files[r.URL.Path]
if !ok {
http.NotFound(w, r)
return
}
w.Write(body)
}))
t.Cleanup(server.Close)
before := upstreamClient
// Under a public name, as a manifest must say it: example.com, which the test server's certificate
// names, dialled at the server.
transport := server.Client().Transport.(*http.Transport).Clone()
addr := server.Listener.Addr().String()
transport.DialContext = func(ctx context.Context, network, _ string) (net.Conn, error) {
return (&net.Dialer{}).DialContext(ctx, network, addr)
}
upstreamClient = &http.Client{Transport: transport}
t.Cleanup(func() { upstreamClient = before })
return "https://example.com"
}
// entriesOf lists a packed archive: each file with its mode and content.
func entriesOf(t *testing.T, packed []byte) map[string]string {
t.Helper()
z, err := gzip.NewReader(bytes.NewReader(packed))
if err != nil {
t.Fatal(err)
}
r := tar.NewReader(z)
out := map[string]string{}
for {
h, err := r.Next()
if err == io.EOF {
return out
}
if err != nil {
t.Fatal(err)
}
body, _ := io.ReadAll(r)
out[h.Name] = string(body) + " " + os.FileMode(h.Mode).String()
}
}
func release(t *testing.T) []byte {
return gzipped(t, tarOf(t, []tarEntry{
{name: "betterbird/", kind: tar.TypeDir, mode: 0o755},
{name: "betterbird/application.ini", body: "[App]\nVersion=153.4.0\n", mode: 0o644},
{name: "betterbird/betterbird", body: "#!/bin/sh\n", mode: 0o755},
{name: "betterbird/defaults/pref/channel-prefs.js", body: "pref(\"app.update.channel\", \"default\");\n", mode: 0o644},
}))
}
func pinnedManifest(base, sum string) string {
return `{"module":"betterbird","version":"1",
"build":{"artifacts":[
{"name":"archive","kind":"archive","url":"` + base + `/release.tar.gz","sha256":"` + sum + `","version":"153.4.0esr-bb10","within":"betterbird"}]},
"resources":[{"id":"installation","type":"archive","path":"/opt/betterbird","artifact":"archive"}]}`
}
func TestAnArchivePinnedByURLIsFetchedCheckedAndPackedWithoutItsOwnDirectory(t *testing.T) {
body := release(t)
v := &vendor{files: map[string][]byte{"/release.tar.gz": body}}
base := serveVendor(t, v)
r, workspace := aRepository(t, pinnedManifest(base, sumHex(body)), nil)
got, err := Build(context.Background(), r.run, r, "https://forge.invalid/catalogue.git", "", "", workspace, nil,
Npmrc{}, GitCredential{}, nil)
if err != nil {
t.Fatal(err)
}
digest, _ := got.Manifest.Resources[0]["digest"].(string)
if !strings.HasPrefix(digest, "sha256:") || r.archives["betterbird/archive"] != digest {
t.Fatalf("the release was not published and pinned as the module's archive: %v, published %v",
got.Manifest.Resources[0], r.archives)
}
packed, err := upstreamArchive(context.Background(), workspace, catalogue.Artifact{Name: "archive", Kind: "archive",
URL: base + "/release.tar.gz", SHA256: sumHex(body), Version: "153.4.0esr-bb10", Within: "betterbird"},
func(string, string, ...any) {})
if err != nil {
t.Fatal(err)
}
if "sha256:"+sumHex(packed) != digest {
t.Fatalf("a second packing of one pin gave another digest: %s and %s", digest, sumHex(packed))
}
entries := entriesOf(t, packed)
want := map[string]string{
"application.ini": "[App]\nVersion=153.4.0\n -rw-r--r--",
"betterbird": "#!/bin/sh\n -rwxr-xr-x",
"defaults/pref/channel-prefs.js": "pref(\"app.update.channel\", \"default\");\n -rw-r--r--",
}
if len(entries) != len(want) {
t.Fatalf("the archive holds %v, want %v", entries, want)
}
for name, w := range want {
if entries[name] != w {
t.Errorf("%s is %q in the archive, want %q", name, entries[name], w)
}
}
}
func TestAnUpstreamFileIsFetchedOnceAndKeptByItsDigest(t *testing.T) {
// The vendor of an add-on serves only its newest version at one address: a pin that was fetched once
// must still build after the address moved on.
body := release(t)
v := &vendor{files: map[string][]byte{"/release.tar.gz": body}}
base := serveVendor(t, v)
r, workspace := aRepository(t, pinnedManifest(base, sumHex(body)), nil)
first, err := Build(context.Background(), r.run, r, "https://forge.invalid/catalogue.git", "", "", workspace, nil,
Npmrc{}, GitCredential{}, nil)
if err != nil {
t.Fatal(err)
}
v.files["/release.tar.gz"] = []byte("the vendor's next release")
second, err := Build(context.Background(), r.run, r, "https://forge.invalid/catalogue.git", "", "", workspace, nil,
Npmrc{}, GitCredential{}, nil)
if err != nil {
t.Fatalf("a pin fetched before failed to build once the address served something else: %v", err)
}
if v.asked != 1 {
t.Fatalf("the vendor was asked %d times for one pin, want once", v.asked)
}
if first.Manifest.Resources[0]["digest"] != second.Manifest.Resources[0]["digest"] {
t.Fatal("two builds of one pin gave two digests")
}
}
func TestAnUpstreamFileWhoseDownloadDoesNotMatchItsPinFailsAndNamesBothDigests(t *testing.T) {
pinned := release(t)
served := gzipped(t, tarOf(t, []tarEntry{{name: "betterbird/betterbird", body: "not what was reviewed", mode: 0o755}}))
v := &vendor{files: map[string][]byte{"/release.tar.gz": served}}
base := serveVendor(t, v)
r, workspace := aRepository(t, pinnedManifest(base, sumHex(pinned)), nil)
_, err := Build(context.Background(), r.run, r, "https://forge.invalid/catalogue.git", "", "", workspace, nil,
Npmrc{}, GitCredential{}, nil)
if err == nil {
t.Fatal("a download that is not the pinned file was built")
}
for _, digest := range []string{sumHex(pinned), sumHex(served)} {
if !strings.Contains(err.Error(), "sha256:"+digest) {
t.Errorf("the refusal does not name sha256:%s: %v", digest, err)
}
}
if len(r.archives) != 0 {
t.Fatalf("something was published: %v", r.archives)
}
kept, _ := filepath.Glob(filepath.Join(upstreamDir(workspace), "*"))
if len(kept) != 0 {
t.Fatalf("the refused download was kept: %v", kept)
}
}
func TestAnUpstreamFileThatIsNotAnArchiveIsKeptWholeUnderItsName(t *testing.T) {
xpi := []byte("PK\x03\x04 an add-on")
v := &vendor{files: map[string][]byte{"/owl.xpi": xpi}}
base := serveVendor(t, v)
packed, err := upstreamArchive(context.Background(), t.TempDir(), catalogue.Artifact{Name: "owl", Kind: "archive",
URL: base + "/owl.xpi", SHA256: sumHex(xpi), Version: "1.5.6.1", File: "owl.xpi"}, func(string, string, ...any) {})
if err != nil {
t.Fatal(err)
}
entries := entriesOf(t, packed)
if len(entries) != 1 || entries["owl.xpi"] != string(xpi)+" -rw-r--r--" {
t.Fatalf("the add-on was not kept whole as owl.xpi: %v", entries)
}
}
func TestAnUpstreamArchiveHoldingALinkOrAnEntryOutsideItsDirectoryIsRefused(t *testing.T) {
cases := map[string][]tarEntry{
// The community package links the dictionaries; a release carrying a link is not one a machine unpacks.
"is a 2": {{name: "betterbird/betterbird", body: "x", mode: 0o755},
{name: "betterbird/dictionaries", kind: tar.TypeSymlink}},
"outside betterbird": {{name: "betterbird/betterbird", body: "x", mode: 0o755},
{name: "README", body: "x", mode: 0o644}},
"outside the archive": {{name: "../etc/passwd", body: "x", mode: 0o644}},
"no file within": {{name: "betterbird/", kind: tar.TypeDir, mode: 0o755}},
}
for want, entries := range cases {
body := gzipped(t, tarOf(t, entries))
v := &vendor{files: map[string][]byte{"/r.tar.gz": body}}
base := serveVendor(t, v)
_, err := upstreamArchive(context.Background(), t.TempDir(), catalogue.Artifact{Name: "archive", Kind: "archive",
URL: base + "/r.tar.gz", SHA256: sumHex(body), Version: "1", Within: "betterbird"}, func(string, string, ...any) {})
if err == nil || !strings.Contains(err.Error(), want) {
t.Errorf("an archive whose entry is %q was not refused for it: %v", want, err)
}
}
}
func TestAnUpstreamArchiveCompressedWithXzIsRead(t *testing.T) {
// Betterbird publishes its Linux release as .tar.xz; the build seat's image carries xz for it.
if _, err := exec.LookPath("xz"); err != nil {
t.Skip("NOT TESTED here: no xz on this machine (the build seat's image carries it)")
}
plain := tarOf(t, []tarEntry{{name: "betterbird/betterbird", body: "#!/bin/sh\n", mode: 0o755}})
cmd := exec.Command("xz", "-zc")
cmd.Stdin = bytes.NewReader(plain)
compressed, err := cmd.Output()
if err != nil {
t.Fatal(err)
}
v := &vendor{files: map[string][]byte{"/r.tar.xz": compressed}}
base := serveVendor(t, v)
packed, err := upstreamArchive(context.Background(), t.TempDir(), catalogue.Artifact{Name: "archive", Kind: "archive",
URL: base + "/r.tar.xz", SHA256: sumHex(compressed), Version: "1", Within: "betterbird"}, func(string, string, ...any) {})
if err != nil {
t.Fatal(err)
}
var names []string
for n := range entriesOf(t, packed) {
names = append(names, n)
}
sort.Strings(names)
if strings.Join(names, ",") != "betterbird" {
t.Fatalf("the xz release unpacked to %v", names)
}
}
func TestAnUpstreamTarWithAGlobalHeaderNestedWithinAndTrailingDataIsRead(t *testing.T) {
// git archive writes a pax global header; a release may unpack under a/b with a/ listed first; xz may be
// followed by bytes after the tar's end, which must be read before xz is waited for.
var raw bytes.Buffer
w := tar.NewWriter(&raw)
w.WriteHeader(&tar.Header{Typeflag: tar.TypeXGlobalHeader, Name: "pax_global_header", PAXRecords: map[string]string{"comment": "abc"}})
w.WriteHeader(&tar.Header{Name: "a/", Typeflag: tar.TypeDir, Mode: 0o755})
w.WriteHeader(&tar.Header{Name: "a/b/", Typeflag: tar.TypeDir, Mode: 0o755})
w.WriteHeader(&tar.Header{Name: "a/b/f", Typeflag: tar.TypeReg, Mode: 0o644, Size: 1})
w.Write([]byte("x"))
w.Close()
body := gzipped(t, raw.Bytes())
v := &vendor{files: map[string][]byte{"/r.tar.gz": body}}
base := serveVendor(t, v)
packed, err := upstreamArchive(context.Background(), t.TempDir(), catalogue.Artifact{Name: "a", Kind: "archive",
URL: base + "/r.tar.gz", SHA256: sumHex(body), Version: "1", Within: "a/b"}, func(string, string, ...any) {})
if err != nil {
t.Fatal(err)
}
if e := entriesOf(t, packed); len(e) != 1 || e["f"] == "" {
t.Fatalf("%v", e)
}
if _, err := exec.LookPath("xz"); err != nil {
t.Skip("NOT TESTED here: no xz")
}
// A megabyte after the tar's end: more than a pipe holds.
trailing := append(append([]byte{}, raw.Bytes()...), bytes.Repeat([]byte{0}, 1<<20)...)
cmd := exec.Command("xz", "-zc")
cmd.Stdin = bytes.NewReader(trailing)
compressed, err := cmd.Output()
if err != nil {
t.Fatal(err)
}
v.files["/r.tar.xz"] = compressed
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
if _, err := upstreamArchive(ctx, t.TempDir(), catalogue.Artifact{Name: "a", Kind: "archive", URL: base + "/r.tar.xz",
SHA256: sumHex(compressed), Version: "1", Within: "a/b"}, func(string, string, ...any) {}); err != nil {
t.Fatal(err)
}
}
func TestAnUpstreamArchiveWithTooManyEntriesIsRefused(t *testing.T) {
var raw bytes.Buffer
w := tar.NewWriter(&raw)
for i := 0; i <= maxEntries; i++ {
w.WriteHeader(&tar.Header{Name: fmt.Sprintf("d%d/", i), Typeflag: tar.TypeDir, Mode: 0o755})
}
w.Close()
body := gzipped(t, raw.Bytes())
v := &vendor{files: map[string][]byte{"/r.tar.gz": body}}
base := serveVendor(t, v)
_, err := upstreamArchive(context.Background(), t.TempDir(), catalogue.Artifact{Name: "a", Kind: "archive",
URL: base + "/r.tar.gz", SHA256: sumHex(body), Version: "1"}, func(string, string, ...any) {})
if err == nil || !strings.Contains(err.Error(), "entries") {
t.Fatalf("%v", err)
}
}
func TestARedirectOffHttpsIsNotFollowed(t *testing.T) {
plain := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Write([]byte("x")) }))
t.Cleanup(plain.Close)
secure := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
http.Redirect(w, r, plain.URL+"/r", http.StatusFound)
}))
t.Cleanup(secure.Close)
before := upstreamClient
upstreamClient = secure.Client()
t.Cleanup(func() { upstreamClient = before })
_, err := upstreamFile(context.Background(), t.TempDir(), catalogue.Artifact{Name: "a", URL: secure.URL + "/r",
SHA256: sumHex([]byte("x")), Version: "1", File: "a"}, func(string, string, ...any) {})
if err == nil || !strings.Contains(err.Error(), "not https") {
t.Fatalf("%v", err)
}
}
+83 -1
View File
@@ -2,7 +2,10 @@ package catalogue
import (
"fmt"
"net"
"net/url"
"path"
"regexp"
"sort"
"strings"
)
@@ -312,7 +315,7 @@ func (b *Build) problems(module string) []string {
module, a.Name, a.System, spokenSystems()))
}
} else {
if a.From == "" {
if a.From == "" && a.URL == "" {
problems = append(problems, fmt.Sprintf(
"%s: %q says nothing about what it is built from", module, a.Name))
}
@@ -350,6 +353,85 @@ func (b *Build) problems(module string) []string {
problems = append(problems, fmt.Sprintf(
"%s: %q mirrors %q, which names no tag or digest", module, a.Name, a.From))
}
problems = append(problems, upstreamFileProblems(module, a)...)
}
return problems
}
// reservedSuffixes are names no public vendor publishes under: an installation's own and documentation's.
var reservedSuffixes = []string{".internal", ".local", ".lan", ".home", ".corp", ".localhost", ".home.arpa",
".test", ".invalid", ".example", ".intranet", ".private"}
// notAPublicHost says why a host is not a public vendor's name, or "".
func notAPublicHost(host string) string {
h := strings.ToLower(strings.TrimSuffix(host, "."))
if net.ParseIP(h) != nil {
return "an address rather than a name"
}
if !strings.Contains(h, ".") {
return "a name with no domain"
}
for _, s := range reservedSuffixes {
if strings.HasSuffix(h, s) {
return "a name under " + s + ", which is nobody's public domain"
}
}
return ""
}
// sha256Hex is a pinned sha256 as a manifest writes it: 64 lower-case hexadecimal characters.
var sha256Hex = regexp.MustCompile(`^[0-9a-f]{64}$`)
// upstreamFileProblems is what is wrong with an archive pinned by URL (novox/hq ADR 0316 §2, §6): it is an
// archive, fetched over https, pinned by a sha256 and a version, and it names no directory of the repository
// besides; a file name or a directory within is a plain path that stays where it is put.
func upstreamFileProblems(module string, a Artifact) []string {
pinned := a.URL != "" || a.SHA256 != "" || a.Version != "" || a.File != "" || a.Within != ""
if !pinned {
return nil
}
var problems []string
say := func(format string, args ...any) {
problems = append(problems, fmt.Sprintf("%s: %q ", module, a.Name)+fmt.Sprintf(format, args...))
}
if a.Kind != ArtifactArchive {
say("is a %q and names an upstream file (url, sha256, version, file, within): only an archive is "+
"mirrored from one (novox/hq ADR 0316)", a.Kind)
return problems
}
if a.URL == "" {
say("names a sha256, a version, a file or a directory within, and no url to fetch it from")
return problems
}
if a.From != "" {
say("is built from %q and fetched from %s: an archive is one or the other", a.From, a.URL)
}
if u, err := url.Parse(a.URL); err != nil || u.Scheme != "https" || u.Host == "" {
say("is fetched from %q, and an upstream file is fetched over https from a named host", a.URL)
} else if why := notAPublicHost(u.Hostname()); why != "" {
say("is fetched from %s, %s: an upstream file is somebody else's release, published under a public name", a.URL, why)
}
if !sha256Hex.MatchString(a.SHA256) {
say("is fetched from %s without a pinned sha256 (64 lower-case hexadecimal characters): what that "+
"address serves changes under the mesh, and only the digest says it is what was reviewed", a.URL)
}
if strings.TrimSpace(a.Version) == "" {
say("pins %s and says no version, so a person reading the pin cannot tell which release it is", a.URL)
}
if a.File != "" && a.Within != "" {
say("names a file (%q) and a directory within (%q): a file is kept whole, a directory is taken out "+
"of an archive", a.File, a.Within)
}
for field, p := range map[string]string{"file": a.File, "within": a.Within} {
if p == "" {
continue
}
clean := path.Clean(p)
if clean != p || strings.HasPrefix(p, "/") || clean == "." || clean == ".." || strings.HasPrefix(clean, "../") ||
(field == "file" && strings.Contains(p, "/")) {
say("names %s %q, which is not a plain relative %s", field, p, map[string]string{
"file": "file name", "within": "directory"}[field])
}
}
return problems
}
+7 -1
View File
@@ -71,6 +71,9 @@ var hostname = regexp.MustCompile(
`(?:be|nl|de|fr|uk|eu|com|net|org|io|dev|app|cloud|site|online|me|co|ch|at|lu|` +
`internal|example|tld|test|invalid)`)
// upstreamVersion is where an artifact pinned by URL says which release it is (Artifact.Version).
var upstreamVersion = regexp.MustCompile(`^build\.artifacts\[[0-9]+\]\.version$`)
// address is a dotted quad.
var address = regexp.MustCompile(`(?:[0-9]{1,3}\.){3}[0-9]{1,3}`)
@@ -120,7 +123,10 @@ func InstallationProblems(m Manifest) []string {
problems = append(problems, fmt.Sprintf("%s names %s at %s", m.Module, name, at))
}
for _, ip := range addressesIn(value) {
if meant[ip] {
// An upstream pin's version may have four dotted numbers (Owl 1.5.6.1), which is no address; a name
// in it is still judged above. Its URL is judged whole: the vendor's host goes through the artifact's
// names-on-purpose like any name a definition means (novox/hq ADR 0316).
if meant[ip] || upstreamVersion.MatchString(at) {
continue
}
problems = append(problems, fmt.Sprintf("%s names the public address %s at %s", m.Module, ip, at))
+29 -1
View File
@@ -976,6 +976,35 @@ type Artifact struct {
// success, and was invisible to the launcher (novox/hq 04-ISSUES/142).
Binary string `json:"binary,omitempty"`
// URL is where an archive's upstream file is fetched from, for an archive that is a release
// somebody else published rather than a directory of this repository (novox/hq ADR 0316 §2, §6).
// With it come SHA256 and Version, and From stays empty.
//
// **Pinned by the digest of what the vendor published, mirrored by the build seat.** A release of
// about 80 MB copied into a repository is kept for ever in its history at every upgrade, and an
// add-on whose licence forbids copying it to third parties may not be in a repository at all. So the
// manifest names the file, the build seat fetches it, refuses it when its sha256 is not the pin's and
// names both digests, and packs it as any archive is packed: what reaches a machine is the mesh's own
// archive in the artifact store, fetched by digest like every other, and a machine never asks the
// vendor. Upgrading is changing the pin, reviewed like any other change.
URL string `json:"url,omitempty"`
// SHA256 is the sha256 of the file at URL, as 64 lower-case hexadecimal characters.
SHA256 string `json:"sha256,omitempty"`
// Version is what the vendor calls the release pinned, said where a person reads the pin; the build
// does not read it.
Version string `json:"version,omitempty"`
// File is the name an upstream file that is not an archive is kept under: the archive built from it
// holds that one file (an add-on's xpi). Empty means the file is a tar archive (gzip, bzip2 or xz)
// whose files are the artifact.
File string `json:"file,omitempty"`
// Within is the directory inside an upstream tar archive whose content is the artifact, for a
// release that unpacks into a directory of its own name (`betterbird/`). An entry outside it refuses
// the build. Empty means the archive's root.
Within string `json:"within,omitempty"`
// NamesOnPurpose are the names this artifact means to name, each with why: the vendor's host an upstream
// file is fetched from, which the installation check judges like any name (InstallationProblems).
NamesOnPurpose map[string]string `json:"names-on-purpose,omitempty"`
// Entrypoints are the compiled files a tool host should load from this module, relative to the
// bundle's root.
//
@@ -1829,7 +1858,6 @@ func ParseManifest(raw []byte) (Manifest, error) {
problems = append(problems, invokeProblems(m)...)
problems = append(problems, replacesProblems(m)...)
problems = append(problems, UnitSettingProblems(m)...)
problems = append(problems, OwnerSettingProblems(m)...)
problems = append(problems, endpointNameProblems(m)...)
problems = append(problems, RouteProblems(m)...)
for _, port := range m.Guards {
+19 -34
View File
@@ -68,10 +68,9 @@ func TestFailedIsAnOptionalVerbOfTheServiceManager(t *testing.T) {
holder := func(serves ...string) Manifest {
return Manifest{Module: "systemd", Version: "1", Claims: []Claim{{Name: ServiceManagerSeat, Scope: ScopeNode, Serves: serves}}}
}
eight := []string{"units", "status", "start", "stop", "restart", "enable", "disable", "journal",
"reset-failed", "wanted-by", "unlink-dangling"}
eight := []string{"units", "status", "start", "stop", "restart", "enable", "disable", "journal"}
if err := CanHold(holder(eight...), seat); err != nil {
t.Fatalf("a holder without failed is refused: %v", err)
t.Fatalf("today's holder, without failed, is refused: %v", err)
}
if err := CanHold(holder(append(eight, "failed")...), seat); err != nil {
t.Fatalf("a holder serving failed is refused: %v", err)
@@ -82,7 +81,7 @@ func TestFailedIsAnOptionalVerbOfTheServiceManager(t *testing.T) {
if err := CanHold(holder(eight[1:]...), seat); err == nil || !strings.Contains(err.Error(), "does not serve units") {
t.Fatalf("a holder missing a required verb was accepted: %v", err)
}
optional := map[string]bool{"failed": true}
optional := map[string]bool{"failed": true, "reset-failed": true, "wanted-by": true, "unlink-dangling": true}
for _, v := range seat.Serves {
if v.Optional != optional[v.Name] {
t.Errorf("%s optional: %v", v.Name, v.Optional)
@@ -90,35 +89,26 @@ func TestFailedIsAnOptionalVerbOfTheServiceManager(t *testing.T) {
}
}
// **`reset-failed`, `wanted-by` and `unlink-dangling` are required** (novox/hq issue 332, ADR 0246 step 3):
// every holder serves them, so a holder serving only the nine verbs before them is refused by the verb it
// lacks; a holder serving all twelve holds the seat; each replaces the shell command it names; and read
// back from the store's row they stay required.
func TestResetFailedWantedByAndUnlinkDanglingAreRequiredVerbsOfTheServiceManager(t *testing.T) {
// **`reset-failed` and `wanted-by` join the seat optional** (novox/hq issue 332, ADR 0246 step 1): the
// holder running today, serving the nine verbs, still holds the seat; a holder serving all eleven is not
// refused; each replaces the shell command it names; and read back from the store's row they stay optional.
func TestResetFailedAndWantedByAreOptionalVerbsOfTheServiceManager(t *testing.T) {
defer UseSeats(DefaultSeats())
nine := []string{"units", "status", "start", "stop", "restart", "enable", "disable", "journal", "failed"}
all := append(slices.Clone(nine), "reset-failed", "wanted-by", "unlink-dangling")
holder := func(serves ...string) Manifest {
return Manifest{Module: "systemd", Version: "1", Claims: []Claim{{Name: ServiceManagerSeat, Scope: ScopeNode, Serves: serves}}}
}
refusedWithout := func(seat Seat) {
t.Helper()
for _, verb := range []string{"reset-failed", "wanted-by", "unlink-dangling"} {
serves := slices.DeleteFunc(slices.Clone(all), func(v string) bool { return v == verb })
if err := CanHold(holder(serves...), seat); err == nil || !strings.Contains(err.Error(), "does not serve "+verb) {
t.Errorf("a holder not serving %s holds the seat: %v", verb, err)
}
}
}
seat, _ := SeatNamed(ServiceManagerSeat)
if err := CanHold(holder(nine...), seat); err == nil {
t.Fatal("a holder serving only the nine verbs before issue 332 holds the seat")
if err := CanHold(holder(nine...), seat); err != nil {
t.Fatalf("today's holder, without the two verbs, is refused: %v", err)
}
refusedWithout(seat)
if err := CanHold(holder(all...), seat); err != nil {
t.Fatalf("a holder serving all twelve is refused: %v", err)
if err := CanHold(holder(append(nine, "reset-failed", "wanted-by")...), seat); err != nil {
t.Fatalf("a holder serving both is refused: %v", err)
}
if err := CanHold(holder(append(slices.Clone(all), "why-started")...), seat); err == nil || !strings.Contains(err.Error(), "does not promise") {
if err := CanHold(holder(append(nine, "reset-failed", "wanted-by", "unlink-dangling")...), seat); err != nil {
t.Fatalf("a holder serving unlink-dangling too is refused: %v", err)
}
if err := CanHold(holder(append(nine, "why-started")...), seat); err == nil || !strings.Contains(err.Error(), "does not promise") {
t.Fatalf("a verb the seat does not promise was accepted: %v", err)
}
replaces := map[string]string{"reset-failed": "systemctl reset-failed", "wanted-by": "systemctl list-dependencies --reverse",
@@ -128,9 +118,6 @@ func TestResetFailedWantedByAndUnlinkDanglingAreRequiredVerbsOfTheServiceManager
if !ok {
continue
}
if v.Optional {
t.Errorf("%s is still optional", v.Name)
}
if !slices.Contains(v.Replaces, want) {
t.Errorf("%s does not say it replaces %q: %v", v.Name, want, v.Replaces)
}
@@ -161,9 +148,8 @@ func TestResetFailedWantedByAndUnlinkDanglingAreRequiredVerbsOfTheServiceManager
}
UseSeats(rows)
seat, _ = SeatNamed(ServiceManagerSeat)
refusedWithout(seat)
if err := CanHold(holder(all...), seat); err != nil {
t.Fatalf("read back from the row, a holder serving all twelve is refused: %v", err)
if err := CanHold(holder(nine...), seat); err != nil {
t.Fatalf("read back from the row, the two verbs are required: %v", err)
}
}
@@ -183,9 +169,8 @@ func TestAnOptionalVerbStaysOptionalInASetReadFromTheStore(t *testing.T) {
UseSeats(rows)
seat, _ := SeatNamed(ServiceManagerSeat)
holder := Manifest{Module: "systemd", Version: "1", Claims: []Claim{{Name: ServiceManagerSeat, Scope: ScopeNode,
Serves: []string{"units", "status", "start", "stop", "restart", "enable", "disable", "journal",
"reset-failed", "wanted-by", "unlink-dangling"}}}}
Serves: []string{"units", "status", "start", "stop", "restart", "enable", "disable", "journal"}}}}
if err := CanHold(holder, seat); err != nil {
t.Fatalf("a set read from rows refuses a holder without the optional failed: %v", err)
t.Fatalf("a set read from rows refuses today's holder: %v", err)
}
}
+7 -7
View File
@@ -678,13 +678,13 @@ func serviceManagerVerbs() []Verb {
Replaces: []string{"systemctl --failed"}},
// **Clearing a failed record, and finding what starts a unit** (novox/hq issue 332): a unit whose
// file is gone stays failed in its manager until the record is reset, and what keeps asking for it
// is a dependency or an enable link no verb could show. Required since every holder serves them
// (ADR 0246 step 3).
{Name: "reset-failed", Description: "Clear one unit's failed record in its manager — " +
// is a dependency or an enable link no verb could show. Both optional while their holders catch up
// (ADR 0246 step 1); a later change requires them.
{Name: "reset-failed", Optional: true, Description: "Clear one unit's failed record in its manager — " +
"its failed state and its start-limit count — and answer its state after. It removes no file and " +
"starts nothing.",
Input: scoped(unit, []string{"unit"}), Replaces: []string{"systemctl reset-failed"}},
{Name: "wanted-by", Description: "What wants, requires or triggers one unit, read-only: the " +
{Name: "wanted-by", Optional: true, Description: "What wants, requires or triggers one unit, read-only: the " +
"manager's reverse dependencies (WantedBy, RequiredBy, UpheldBy, BoundBy, TriggeredBy and the rest) and " +
"every enable link naming it in the scope's configuration directories (*.wants, *.requires, *.upholds), " +
"each with where it points and whether it dangles — the account's own directory included in user scope.",
@@ -692,9 +692,9 @@ func serviceManagerVerbs() []Verb {
Replaces: []string{"systemctl list-dependencies --reverse", "systemctl show -p WantedBy",
"ls ~/.config/systemd/user/*.wants"}},
// **A dangling enable link removed on the person's word** (novox/hq issue 332): `systemctl disable`
// leaves a link whose unit file is gone, so the manager keeps asking for the unit. Required since
// every holder serves it (ADR 0246 step 3).
{Name: "unlink-dangling", Description: "Remove the dangling enable links named for one " +
// leaves a link whose unit file is gone, so the manager keeps asking for the unit. Optional until its
// holders serve it (ADR 0246 step 1).
{Name: "unlink-dangling", Optional: true, Description: "Remove the dangling enable links named for one " +
"unit — links in a *.wants, *.requires or *.upholds directory of the account's or the machine's own " +
"configuration whose target does not exist, which `disable` leaves once the unit's file is gone — on " +
"the person's word: `why` is required. Each is recorded (its path, its target, the reason) before it is " +
-181
View File
@@ -1,181 +0,0 @@
package catalogue
import (
"strings"
"testing"
)
// A declared directory may be owned by the account its module runs as (novox/hq issue 279): a media
// manager runs as the puid and pgid its assignment sets per node, and a folder it must write was owned
// by the manifest's literal, so the app refused it on a machine whose identity was not the image's.
func recycler() Manifest {
return Manifest{Module: "sonarr",
Resources: []map[string]any{
{"id": "recycle-bin", "type": "directory", "path": "/srv/sonarr/recycle", "mode": "0755",
"owner": "${setting:puid}:${setting:pgid}"},
{"id": "identity", "type": "file", "path": "/srv/sonarr/identity.env", "mode": "0644",
"content": "PUID=${setting:puid}\nPGID=${setting:pgid}\n"},
},
}
}
func copied(r map[string]any) map[string]any {
out := map[string]any{}
for k, v := range r {
out[k] = v
}
return out
}
// The node's identity reaches the directory's owner, the node's layer over the mesh's, a number given
// as a string or as a JSON number alike.
func TestASettingOwnsADirectory(t *testing.T) {
m := recycler()
layers := []Layer{
{From: "the mesh", Values: map[string]any{"puid": "1000", "pgid": "1000"}},
{From: "this node", Values: map[string]any{"puid": float64(1001), "pgid": "2000"}},
}
r := copied(m.Resources[0])
if err := settingInto(r, layers, m.Module); err != nil {
t.Fatal(err)
}
if r["owner"] != "1001:2000" {
t.Fatalf("owned by %q, not 1001:2000", r["owner"])
}
user := copied(m.Resources[0])
user["owner"] = "${setting:puid}"
if err := settingInto(user, layers, m.Module); err != nil || user["owner"] != "1001" {
t.Fatalf("user alone: %v, %v", user["owner"], err)
}
}
// Composed for a machine, the way a push writes it: the directory reaches the node-engine owned by the
// number the node's settings name, never by the placeholder.
func TestAComposedMachineGetsTheOwnerTheSettingNames(t *testing.T) {
r := anAdoptedAnchor()
r.Modules = append(r.Modules, recycler())
with := anchorRendering(false)
with.Settings["sonarr"] = []Layer{{From: "anchor", Values: map[string]any{"puid": "1001", "pgid": "2000"}}}
composed, err := r.Compose(with)
if err != nil {
t.Fatal(err)
}
if why, left := composed.LeftOut["sonarr"]; left {
t.Fatalf("left out: %s", why)
}
got := byID(composed.Resources)["sonarr.recycle-bin"]
if got["owner"] != "1001:2000" {
t.Fatalf("composed owned by %v, not 1001:2000", got["owner"])
}
}
// Only a whole number owns a directory: an account's name, text, a sign, a fraction or a value past the
// kernel's ids is refused by name, and the directory is left as it was. So is a key nothing sets.
func TestAnOwnerSettingThatIsNoNumberIsRefused(t *testing.T) {
m := recycler()
for _, bad := range []any{"abc", "root", "", "-1", "+5", "1000:1000", "10 00", "1000\nx", float64(1.5),
float64(-3), "4294967295", "99999999999", true} {
r := copied(m.Resources[0])
err := settingInto(r, []Layer{{From: "ace", Values: map[string]any{"puid": bad, "pgid": "1000"}}}, m.Module)
if err == nil || !strings.Contains(err.Error(), `"puid"`) || !strings.Contains(err.Error(), "whole number") {
t.Errorf("%v: %v", bad, err)
}
if r["owner"] != m.Resources[0]["owner"] {
t.Errorf("%v: the owner was changed on refusal: %v", bad, r["owner"])
}
}
r := copied(m.Resources[0])
if err := settingInto(r, nil, m.Module); err == nil || !strings.Contains(err.Error(), "${setting:puid}") {
t.Errorf("nothing set: %v", err)
}
if err := JudgeSettings(m, []Layer{{From: "ace", Values: map[string]any{"puid": "media", "pgid": "1000"}}}, false); err == nil ||
!strings.Contains(err.Error(), "whole number") {
t.Errorf("a stored name is not refused when judged: %v", err)
}
}
// Where the manifest is read: an owner holding a setting is a whole placeholder or a whole number on
// each side of the colon, and every key is one the module asks for elsewhere or declares a default for.
func TestAnOwnerSettingIsJudgedWhereTheManifestIsRead(t *testing.T) {
ok := []string{"${setting:puid}:${setting:pgid}", "${setting:puid}", "${setting:puid}:1000", "1000:${setting:pgid}",
"1000:1000", "media:media"}
bad := []string{
"${setting:puid}x",
"x${setting:puid}",
"${setting:puid}:${setting:pgid}:1",
"${setting:puid}${setting:pgid}",
"media:${setting:pgid}",
"${setting:puid}:",
"${setting:other}:${setting:pgid}",
}
for _, owner := range append(ok, bad...) {
m := recycler()
m.Resources[0] = copied(m.Resources[0])
m.Resources[0]["owner"] = owner
err := parsed(t, m)
refused := err != nil && strings.Contains(err.Error(), "owner")
want := false
for _, b := range bad {
want = want || b == owner
}
if refused != want {
t.Errorf("%s: refused %v, want %v (%v)", owner, refused, want, err)
}
}
// A key with a default is declared, though nothing else asks for it.
m := recycler()
m.Resources = m.Resources[:1]
m.Resources[0] = copied(m.Resources[0])
m.Resources[0]["owner"] = "${setting:workers}"
m.Settings = map[string]SettingDeclaration{"workers": {Kind: KindPreference, Default: float64(4), Why: "a test"}}
if err := parsed(t, m); err != nil && strings.Contains(err.Error(), "owner") {
t.Errorf("a declared key was refused: %v", err)
}
// A file's owner takes no setting: the sweep says so.
f := recycler()
f.Resources[1] = copied(f.Resources[1])
f.Resources[1]["owner"] = "${setting:puid}"
if err := parsed(t, f); err == nil || !strings.Contains(err.Error(), "not filled in this field") {
t.Errorf("a file's owner setting passed: %v", err)
}
}
// A key a directory's owner asks for is a destination, so setting it is not called stray.
func TestASettingAnOwnerAsksForIsNotStray(t *testing.T) {
m := recycler()
m.Resources = m.Resources[:1]
stray := strings.Join(UnusedSettings(m, []Layer{{From: "ace", Values: map[string]any{"puid": "1001", "pgid": "2000"}}}), "; ")
if strings.Contains(stray, "puid") || strings.Contains(stray, "pgid") {
t.Errorf("called stray: %s", stray)
}
}
// An owner is written as the number parsed: "0001" and "00" reach the node-engine as 1 and 0, never as text.
// Uid 0 is allowed, because a key an owner asks for is the terminal's alone.
func TestAnOwnerSettingIsWrittenAsTheNumber(t *testing.T) {
m := recycler()
r := copied(m.Resources[0])
if err := settingInto(r, []Layer{{From: "ace", Values: map[string]any{"puid": "0001", "pgid": "00"}}}, m.Module); err != nil {
t.Fatal(err)
}
if r["owner"] != "1:0" {
t.Fatalf("owned by %q, not 1:0", r["owner"])
}
}
// Every key a directory's owner asks for is the controller's terminal's alone (novox/hq issue 279, issue 339):
// root chowns the folder to it, so it is terminal-only whatever else asks for it — a file nothing trusts, a
// default, or nothing at all.
func TestAnOwnersKeyIsTheTerminals(t *testing.T) {
m := recycler()
m.Resources[1] = copied(m.Resources[1])
m.Resources[1]["trusted"] = false
m.Settings = map[string]SettingDeclaration{"pgid": {Kind: KindPreference, Default: "1000", Why: "a test"}}
keys := strings.Join(TerminalKeys(m), ",")
for _, k := range []string{"puid", "pgid"} {
if !strings.Contains(","+keys+",", ","+k+",") {
t.Errorf("%s is not the terminal's: %s", k, keys)
}
}
}
+3 -125
View File
@@ -4,7 +4,6 @@ import (
"fmt"
"regexp"
"sort"
"strconv"
"strings"
)
@@ -16,8 +15,7 @@ import (
// (novox/hq issues 122, 134). ADR 0112 names the operator as one of the four providers; this is the
// operator answering.
//
// `${setting:<key>}` in a file's content, in a service's unit name and in a directory's owner (a uid or
// gid only, issue 279) is filled from the module's settings layers — the mesh's,
// `${setting:<key>}` in a file's content, and in a service's unit name, is filled from the module's settings layers — the mesh's,
// then this node's — the same layers a mergeable JSON file and a contribution already take, so
// `settings set <module>` is the one place a person's values go. **Refused when no layer sets it**,
// naming the key and the remedy: a definition that carried a default for a mail domain would be
@@ -61,9 +59,6 @@ func settingInto(resource map[string]any, layers []Layer, module string) error {
if fmt.Sprint(resource["type"]) == "service" {
return settingIntoUnit(resource, layers, module)
}
if fmt.Sprint(resource["type"]) == "directory" {
return settingIntoOwner(resource, layers, module)
}
if fmt.Sprint(resource["type"]) != "file" {
return nil
}
@@ -148,107 +143,6 @@ func settingIntoUnit(resource map[string]any, layers []Layer, module string) err
return nil
}
// A declared directory may be owned by the account its module runs as (novox/hq issue 279).
//
// A containerised module runs as the identity its assignment sets per node — its puid and pgid
// settings, written into the environment file its image reads (ADR 0112, issue 153) — and a directory
// its app must write, beyond the one its image re-owns at start, was owned by whatever the manifest
// spelt: right on a machine whose identity matched the image's default, refused by the app on one whose
// did not. So a directory's `owner` may name the settings that identity is: `${setting:<key>}` for the
// user alone, `${setting:<user>}:${setting:<group>}` for both, filled from the same layers at
// composition.
//
// **Whole numbers only.** An owner a setting fills is a uid or a gid — never an account's name, never
// text — so no value can put anything into the ownership the node-engine applies but a number; one that
// is not is refused at composition, naming the key, and the directory is left as it was. The shape is
// judged where the manifest is read (OwnerSettingProblems): each side of the colon a whole placeholder
// or a whole number, and every key one the module itself asks for elsewhere — in a file, a unit, a
// contribution or a served fact — or declares a default for, so the owner follows the identity its
// process is told it runs as, never a key that reaches nothing else.
// ownerSide is one side of an owner a setting may fill: a whole placeholder or a whole number.
var ownerSide = `(?:\$\{setting:[a-z0-9][a-z0-9_.-]*\}|[0-9]+)`
// ownerWithSetting is the only shape an owner holding a setting may have: user, or user:group.
var ownerWithSetting = regexp.MustCompile(`^` + ownerSide + `(?::` + ownerSide + `)?$`)
// wholeID is a value as a uid or a gid, and whether it is one: a whole number below the kernel's "no id"
// (2^32-1). Uid 0 is one: a key an owner asks for is the controller's terminal's alone (TerminalKeys).
func wholeID(v string) (uint64, bool) {
if v == "" || strings.TrimLeft(v, "0123456789") != "" {
return 0, false
}
n, err := strconv.ParseUint(v, 10, 32)
return n, err == nil && n < 1<<32-1
}
// OwnerSettingProblems refuses, where a manifest is read, a directory whose owner carries a setting
// in any shape but a whole placeholder or number on each side of the colon, or names a key the module
// asks for nowhere else and declares no default for.
func OwnerSettingProblems(m Manifest) []string {
var problems []string
asked := settingKeysAskedOutsideOwners(m)
for k := range m.Settings {
asked[k] = true
}
for _, r := range m.Resources {
owner, _ := r["owner"].(string)
if !strings.Contains(owner, "${setting:") {
continue
}
if fmt.Sprint(r["type"]) != "directory" {
// Said by the placeholder sweep: only a directory's owner takes a setting.
continue
}
if !ownerWithSetting.MatchString(owner) {
problems = append(problems, fmt.Sprintf("%s: the directory %v's owner %q carries a setting in a shape "+
"the mesh does not fill; an owner names the account by number only, each side of the colon a whole "+
"${setting:key} or a whole number (${setting:puid}:${setting:pgid}) (novox/hq issue 279)",
m.Module, r["id"], owner))
continue
}
for _, key := range settingsUsed(owner) {
if !asked[key] {
problems = append(problems, fmt.Sprintf("%s: the directory %v's owner says ${setting:%s}, and the "+
"module asks for %q nowhere else and declares no default for it: an owner follows the identity "+
"the module's process is told it runs as (its identity file's ${setting:%s}), never a key of its "+
"own (novox/hq issue 279)", m.Module, r["id"], key, key, key))
}
}
}
return problems
}
// settingIntoOwner fills ${setting:…} in a directory's owner, refusing — with the key and why, the
// resource left as it was — a key nothing sets and a value that is not a whole number.
func settingIntoOwner(resource map[string]any, layers []Layer, module string) error {
owner, ok := resource["owner"].(string)
if !ok {
return nil
}
for _, key := range settingsUsed(owner) {
value, set := settingValue(layers, key)
if !set {
return &UnsetSettingError{Module: module, Setting: key, said: fmt.Sprintf(
"%s has a directory %v whose owner says ${setting:%s}, and nothing sets %q for it — an operator's "+
"value is the assignment's, never the definition's (novox/hq ADR 0112): "+
"`settings set %s <file>` with {%q: …}%s",
module, resource["id"], key, key, module, key, orNoSettings(layers))}
}
v := plainly(value)
id, whole := wholeID(v)
if !whole {
return fmt.Errorf("%s: the setting %q is %q, which cannot own the directory %v: an owner a setting "+
"names is a uid or a gid, a whole number, never an account's name or other text (novox/hq issue 279)",
module, key, v, resource["id"])
}
// Written as the number parsed, so "0001" reaches the node-engine as 1, never as text.
owner = strings.ReplaceAll(owner, "${setting:"+key+"}", strconv.FormatUint(id, 10))
}
resource["owner"] = owner
return nil
}
func settingValue(layers []Layer, key string) (any, bool) {
var value any
set := false
@@ -277,25 +171,9 @@ func orNoSettings(layers []Layer) string {
return "; set today: " + strings.Join(keys, ", ")
}
// settingKeysUsedBy is every key a module's files, units, directory owners, contributions and served
// facts ask for, so a setting that lands in one is not called stray.
// settingKeysUsedBy is every key a module's files, contributions and served facts ask for, so a
// setting that lands in one is not called stray.
func settingKeysUsedBy(m Manifest) map[string]bool {
used := settingKeysAskedOutsideOwners(m)
for _, r := range m.Resources {
if fmt.Sprint(r["type"]) == "directory" {
if owner, ok := r["owner"].(string); ok {
for _, k := range settingsUsed(owner) {
used[k] = true
}
}
}
}
return used
}
// settingKeysAskedOutsideOwners is every key a module's files, units, contributions and served facts
// ask for: what a directory's owner may follow (OwnerSettingProblems).
func settingKeysAskedOutsideOwners(m Manifest) map[string]bool {
used := map[string]bool{}
note := func(s string) {
for _, k := range settingsUsed(s) {
+2 -12
View File
@@ -33,9 +33,6 @@ import (
// 4. **A module's whole layer, when it has a mergeable file not marked `"trusted": false`** (novox/hq issue 340,
// TrustedMergeable). A mergeable file takes any key a layer sets, not only those its content names, so no list
// of keys covers it: an empty runtime configuration a provider reads would take a `url` of the caller's.
// 5. **Every setting a directory's owner asks for** (novox/hq issue 279): root chowns the directory to the number
// at the next push, so through a verb any caller could hand itself — or uid 0 — a module's folder. Whatever
// else asks for the key, a file marked `"trusted": false` included, the owner makes it the terminal's.
//
// Derived from the manifest, never listed by hand, so a provider or a trusted file added tomorrow is covered.
@@ -45,8 +42,8 @@ import (
const TrustedField = "trusted"
// TerminalKeys are the settings keys of a module that are set at the controller's terminal alone: places and
// accesses, every key its provisions serve and every setting a served value asks for, every setting a file
// asks for unless it says `"trusted": false`, and every setting a directory's owner asks for. Places and accesses first, then the rest sorted.
// accesses, every key its provisions serve and every setting a served value asks for, and every setting a file
// asks for unless it says `"trusted": false`. Places and accesses first, then the rest sorted.
func TerminalKeys(m Manifest) []string {
keys := map[string]bool{}
for _, served := range m.Serves {
@@ -60,13 +57,6 @@ func TerminalKeys(m Manifest) []string {
}
}
for _, r := range m.Resources {
if fmt.Sprint(r["type"]) == "directory" {
owner, _ := r["owner"].(string)
for _, asked := range settingsUsed(owner) {
keys[asked] = true
}
continue
}
if fmt.Sprint(r["type"]) != "file" {
continue
}
+1 -1
View File
@@ -56,7 +56,7 @@ type filler struct {
// seatInto, machineInto and contributionsInto. A pass that comes to read another field adds it here,
// or the sweep refuses the placeholder it would have filled.
var fillers = []filler{
{namespace: "setting", patterns: []*regexp.Regexp{settingRef}, fields: map[string][]string{"file": {"content"}, "service": {"unit"}, "directory": {"owner"}}},
{namespace: "setting", patterns: []*regexp.Regexp{settingRef}, fields: map[string][]string{"file": {"content"}, "service": {"unit"}}},
{namespace: "dir", patterns: []*regexp.Regexp{dirRef}, fields: map[string][]string{"*": {"path", "content", "volumes", "env", "env-file"}}},
{namespace: "access", patterns: []*regexp.Regexp{accessRef}, fields: map[string][]string{"*": {"path", "content", "volumes", "env", "env-file"}}},
{namespace: "secret", patterns: []*regexp.Regexp{placeholder}, fields: map[string][]string{"file": {"content"}},
+101
View File
@@ -0,0 +1,101 @@
package catalogue
import (
"strings"
"testing"
)
// An archive pinned by URL and sha256 (novox/hq ADR 0316 §2, §6): the vendor's file, mirrored by the build
// seat, never a copy in the repository.
const pinnedSum = "38dd24890a66b4cf9a45bc24e15192e00c26b929e1bb8df7974c16a4539129e2"
func upstreamManifest(artifact string) string {
return `{"module":"a","version":"1","build":{"artifacts":[` + artifact + `]}}`
}
func TestAnArchivePinnedByURLAndSha256IsAccepted(t *testing.T) {
for _, a := range []string{
`{"name":"archive","kind":"archive","url":"https://vendor.example.org/release-1.tar.xz","sha256":"` + pinnedSum +
`","version":"1.0","within":"release"}`,
`{"name":"addon","kind":"archive","url":"https://vendor.example.org/addon.xpi","sha256":"` + pinnedSum +
`","version":"1.5.6.1","file":"addon.xpi"}`,
} {
if _, err := ParseManifest([]byte(upstreamManifest(a))); err != nil {
t.Errorf("%s was refused: %v", a, err)
}
}
}
func TestAnUpstreamFileThatIsNotPinnedIsRefused(t *testing.T) {
cases := map[string]string{
// No digest: what the address serves changes under the mesh.
`{"name":"x","kind":"archive","url":"https://vendor.invalid/r.tar.xz","version":"1"}`: "without a pinned sha256",
`{"name":"x","kind":"archive","url":"https://vendor.invalid/r.tar.xz","sha256":"ABC","version":"1"}`: "without a pinned sha256",
// No version: a person reading the pin cannot tell which release.
`{"name":"x","kind":"archive","url":"https://vendor.invalid/r.tar.xz","sha256":"` + pinnedSum + `"}`: "says no version",
// Not https.
`{"name":"x","kind":"archive","url":"http://vendor.invalid/r.tar.xz","sha256":"` + pinnedSum + `","version":"1"}`: "over https",
// Both a directory of the repository and a URL.
`{"name":"x","kind":"archive","from":"files","url":"https://vendor.invalid/r.tar.xz","sha256":"` + pinnedSum +
`","version":"1"}`: "one or the other",
// A pin without a URL.
`{"name":"x","kind":"archive","from":"files","sha256":"` + pinnedSum + `"}`: "no url",
// Only an archive is mirrored from a file.
`{"name":"x","kind":"image","from":"Dockerfile","url":"https://vendor.invalid/r.tar.xz","sha256":"` + pinnedSum +
`","version":"1"}`: "only an archive",
// A file name that is a path, a directory within that leaves the archive, and both at once.
`{"name":"x","kind":"archive","url":"https://vendor.invalid/a.xpi","sha256":"` + pinnedSum +
`","version":"1","file":"../a.xpi"}`: "not a plain relative file name",
`{"name":"x","kind":"archive","url":"https://vendor.invalid/r.tar.xz","sha256":"` + pinnedSum +
`","version":"1","within":"../up"}`: "not a plain relative directory",
`{"name":"x","kind":"archive","url":"https://vendor.invalid/r.tar.xz","sha256":"` + pinnedSum +
`","version":"1","within":"r","file":"r.tar.xz"}`: "a file is kept whole",
}
for artifact, want := range cases {
_, err := ParseManifest([]byte(upstreamManifest(artifact)))
if err == nil {
t.Errorf("%s was accepted", artifact)
continue
}
if !strings.Contains(err.Error(), want) {
t.Errorf("%s was refused for another reason than %q: %v", artifact, want, err)
}
}
}
func TestAnUpstreamFilesVendorIsNamedOnPurposeAndAnInstallationsIsRefused(t *testing.T) {
// The vendor's host is judged like any name, and passes through names-on-purpose; a version of four dotted
// numbers is not an address, and a name in it is still judged.
m, err := ParseManifest([]byte(upstreamManifest(`{"name":"owl","kind":"archive","url":"https://vendor.be/owl/owl.xpi","sha256":"` +
pinnedSum + `","version":"1.5.6.1","file":"owl.xpi","names-on-purpose":{"vendor.be":"Owl's vendor publishes it there"}}`)))
if err != nil {
t.Fatal(err)
}
if problems := InstallationProblems(m); len(problems) != 0 {
t.Fatalf("the declared vendor and the version were taken for an installation: %v", problems)
}
m.Build.Artifacts[0].NamesOnPurpose = nil
if problems := InstallationProblems(m); len(problems) != 1 || !strings.Contains(problems[0], "vendor.be") {
t.Fatalf("an undeclared vendor host is judged: %v", problems)
}
m.Build.Artifacts[0].Version = "git.novox.be 10.0.0.1"
if problems := InstallationProblems(m); len(problems) != 2 {
t.Fatalf("a name in the version is judged: %v", problems)
}
m.Build.Artifacts[0].URL, m.Build.Artifacts[0].Version = "https://mail.novox.be/owl.xpi", "1"
if problems := InstallationProblems(m); len(problems) != 1 || !strings.Contains(problems[0], "mail.novox.be") {
t.Fatalf("an installation's host in the url is judged: %v", problems)
}
}
func TestAnUpstreamFileFromAnAddressOrAReservedNameIsRefused(t *testing.T) {
for _, u := range []string{"https://git.novox.internal/r.tar.xz", "https://10.1.2.3/r.tar.xz",
"https://203.0.113.9/r.tar.xz", "https://[2001:db8::1]/r.tar.xz", "https://nas.lan/r.tar.xz", "https://vendor/r.tar.xz"} {
_, err := ParseManifest([]byte(upstreamManifest(`{"name":"x","kind":"archive","url":"` + u + `","sha256":"` +
pinnedSum + `","version":"1"}`)))
if err == nil || !strings.Contains(err.Error(), "published under a public name") {
t.Errorf("%s: %v", u, err)
}
}
}
+8 -40
View File
@@ -61,10 +61,6 @@ type Plan struct {
// Phases is the walk's time from its merge, phase by phase (ADR 0282 decision 6): never kept, worked out
// from the walk's own moments by whoever says the walk (`delivery walks`, `plan-moved`).
Phases *WalkPhases `json:"phases,omitempty"`
// WalkSaid is, on the save that started or ended the walk, which of the two it was (novox/hq ADR 0308
// decision 1): set by the save, never kept, read by whoever says the save (`plan-moved`). Empty on every
// other save.
WalkSaid string `json:"-"`
}
// PlanTimes are a walk's own moments beside its modules' (novox/hq ADR 0282 decision 6).
@@ -77,10 +73,6 @@ type PlanTimes struct {
// Class is the walk's module class, read at its cut: core when it moves a module on the controller's own
// path or the mesh's resolver, leaf otherwise (ADR 0282 decision 1).
Class string `json:"class,omitempty"`
// Started is when the walk was said started, and Ended when it was said ended (novox/hq ADR 0308 decision
// 1): each said once, on the save that marked it.
Started *time.Time `json:"started,omitempty"`
Ended *time.Time `json:"ended,omitempty"`
}
// PlanCommit is one repository's commit a walk carries: the latest merge of its branch in the batch.
@@ -125,14 +117,6 @@ type PlanBatch struct {
// the batch's walk does not build them.
Closed bool `json:"closed,omitempty"`
Withheld []string `json:"withheld,omitempty"`
// CutAtOnce says the batch's window closed when it was made: its merge shares nothing with a batch
// assembling and has no delivery group member still to merge, so it waits for no window (novox/hq ADR 0304
// decision 1, ADR 0306 decisions 2 and 3). A merge that joins it later opens its window again. It is not
// Closed, which is a stop's.
CutAtOnce bool `json:"cut_at_once,omitempty"`
// Waits says why a batch made for one merge waits for the window: its delivery group has a member still to
// merge, or mesh-delivery could not say whether it has (novox/hq ADR 0306 decision 2).
Waits string `json:"waits,omitempty"`
}
// OwnPath says the record is a batch of merges on the controller's own path.
@@ -199,8 +183,8 @@ type PlanDelivery struct {
Go *time.Time `json:"go,omitempty"`
By string `json:"by,omitempty"`
Why string `json:"why,omitempty"`
// Stopped is who ended the walk, through the delivery's owner or by hand (`plans stop|close`), and StoppedWhy
// why: the walk is failed, and the delivery reads it as stopped, not as a build that failed.
// Stopped is who ended the walk through the delivery's owner, and StoppedWhy why: the walk is failed,
// and the delivery reads it as stopped, not as a build that failed.
Stopped string `json:"stopped,omitempty"`
StoppedWhy string `json:"stopped_why,omitempty"`
// Merges is every merge the walk or batch answers (novox/hq ADR 0276): its own commits, and the earlier
@@ -252,12 +236,10 @@ func (p Plan) Waiting() bool {
return p.Delivery != nil && p.Delivery.Awaits != "" && p.Delivery.Go == nil
}
// PlanSaved is told every plan this process kept, after it is kept, with the inventory that kept it (novox/hq
// ADR 0239): the serving controller says it on the bus as `plan-moved`, and so does a command (the controller's
// run installs it for every verb, novox/hq ADR 0308 decision 1), so the save that starts or ends a walk is said
// whichever process made it. Nil says nothing; whoever follows a walk also asks for it, and finds an unsaid save
// by comparison.
var PlanSaved func(*Inventory, Plan)
// PlanSaved is told every plan this process kept, after it is kept (novox/hq ADR 0239): the serving
// controller says it on the bus as `plan-moved`. Nil in a command, which says nothing; whoever follows a
// walk also asks for it, so a save made by a command is found by comparison.
var PlanSaved func(Plan)
// ErrPlanMoved is a save against a plan written by somebody else since it was read.
var ErrPlanMoved = errors.New("the plan was written by somebody else since it was read")
@@ -469,7 +451,7 @@ type MergeMove struct {
// leaves a merge answered by two plans, nor a walk stopped whose merges were not moved). A plan written by
// somebody else since it was read refuses the whole act with ErrPlanMoved. Each plan is said (PlanSaved) once
// the transaction is committed, in order.
func (i *Inventory) SavePlansMoving(ctx context.Context, plans []*Plan, moves []MergeMove) (err error) {
func (i *Inventory) SavePlansMoving(ctx context.Context, plans []*Plan, moves []MergeMove) error {
epoch, err := i.actingEpoch(ctx)
if err != nil {
if len(plans) > 0 {
@@ -490,20 +472,6 @@ func (i *Inventory) SavePlansMoving(ctx context.Context, plans []*Plan, moves []
entered time.Time
}
done := make([]written, len(plans))
// **The save that starts or ends a walk is marked** (novox/hq ADR 0308 decision 1), in the same write, and
// unmarked should the write not be made, so the next save marks it again.
marked := time.Now()
undo := make([]func(), 0, len(plans))
defer func() {
if err != nil {
for _, u := range undo {
u()
}
}
}()
for _, p := range plans {
undo = append(undo, p.markWalk(marked))
}
for n, p := range plans {
revision, entered, err := writePlan(ctx, tx, p, epoch)
if err != nil {
@@ -529,7 +497,7 @@ func (i *Inventory) SavePlansMoving(ctx context.Context, plans []*Plan, moves []
p.Epoch = 0
}
if PlanSaved != nil {
PlanSaved(i, *p)
PlanSaved(*p)
}
}
return nil
+3 -4
View File
@@ -202,15 +202,14 @@ func TestASeatsVerbGainsTheArgumentsTheBinaryNames(t *testing.T) {
t.Fatal("failed was not added to the row")
}
// The optional mark is not stored; the working set read from the rows takes it from the compiled seat,
// so a holder that does not serve the optional failed still holds the seat.
// so today's holder, which does not serve failed, still holds the seat.
catalogue.UseSeats(seats)
defer catalogue.UseSeats(catalogue.DefaultSeats())
live, _ := catalogue.SeatNamed(catalogue.ServiceManagerSeat)
holder := catalogue.Manifest{Module: "systemd", Version: "1", Claims: []catalogue.Claim{{Name: catalogue.ServiceManagerSeat,
Scope: catalogue.ScopeNode, Serves: []string{"units", "status", "start", "stop", "restart", "enable", "disable", "journal",
"reset-failed", "wanted-by", "unlink-dangling"}}}}
Scope: catalogue.ScopeNode, Serves: []string{"units", "status", "start", "stop", "restart", "enable", "disable", "journal"}}}}
if err := catalogue.CanHold(holder, live); err != nil {
t.Fatalf("the seat read from the store refuses a holder without the optional failed: %v", err)
t.Fatalf("the seat read from the store refuses today's holder: %v", err)
}
// Seeding again changes nothing more.
before := verbs["journal"]
-81
View File
@@ -1,81 +0,0 @@
package inventory
import "time"
// A walk says when it starts and when it ends (novox/hq ADR 0308 decision 1): the save that starts it and the
// save that ends it are marked, so `plan-moved` names what the walk touches, its bound and its outcome on those
// two saves, and on no other. The marks are kept in the walk's times, so a controller replaced mid-walk (a core
// walk replaces it) neither says a start again nor misses the end.
// The moments of a walk `plan-moved` names.
const (
WalkStarted = "started"
WalkEnded = "ended"
)
// The outcomes of an ended walk (ADR 0308 decision 1).
const (
WalkDelivered = "delivered"
WalkFailed = "failed"
WalkStopped = "stopped"
)
// Live says the walk is walking: cut, not waiting for its delivery's word, and not ended.
func (p Plan) Live() bool { return p.Open() && !p.Batch() && !p.Waiting() }
// Ended says the walk is in a state it does not leave by itself: done, failed or superseded.
func (p Plan) Ended() bool {
return p.State == PlanDone || p.State == PlanFailed || p.State == PlanSuperseded
}
// Outcome is how an ended walk ended: delivered when done; stopped when a person or the delivery's owner stopped
// it, or a newer walk took it over; failed otherwise. Empty for a walk not ended.
func (p Plan) Outcome() string {
switch {
case p.State == PlanDone:
return WalkDelivered
case p.State == PlanSuperseded:
return WalkStopped
case p.State == PlanFailed && p.Delivery != nil && p.Delivery.Stopped != "":
return WalkStopped
case p.State == PlanFailed:
return WalkFailed
}
return ""
}
// markWalk marks the save that starts or ends the walk, in its times and in WalkSaid, and returns what undoes
// the mark should the save not be written. A walk starts on the first save that finds it live — when it is cut
// and needs no word, or when it is let go — which is before its first tier is asked, and so before anything is
// sent: what it restarts is held from its first send on. It starts again when a failed walk is retried. It ends
// on the first save that finds it ended after it started; one that ends without ever starting (a batch that
// moved nothing, a walk stopped while it waited for its word) touched nothing and says neither. A walk kept
// before its times were (ADR 0282 decision 6) is not marked; a release plan is, its times holding only these.
func (p *Plan) markWalk(now time.Time) (undo func()) {
p.WalkSaid = ""
if p.Batch() || (p.Times == nil && p.Release == nil) {
return func() {}
}
var was *PlanTimes
if p.Times != nil {
kept := *p.Times
was = &kept
}
times := PlanTimes{}
if p.Times != nil {
times = *p.Times
}
at := now.UTC()
switch {
case p.Live() && (times.Started == nil || times.Ended != nil):
times.Started, times.Ended = &at, nil
p.WalkSaid = WalkStarted
case p.Ended() && times.Started != nil && times.Ended == nil:
times.Ended = &at
p.WalkSaid = WalkEnded
default:
return func() {}
}
p.Times = &times
return func() { p.Times, p.WalkSaid = was, "" }
}
-130
View File
@@ -1,130 +0,0 @@
package inventory
import (
"testing"
"time"
)
// A walk is said started on the save that finds it live, ended on the save that finds it ended after it started,
// and neither on any other save (novox/hq ADR 0308 decision 1).
func TestAWalkIsSaidStartedOnceAndEndedOnce(t *testing.T) {
now := time.Date(2026, 10, 12, 11, 0, 0, 0, time.UTC)
cut := now.Add(-time.Second)
p := Plan{ID: "plan-1", State: PlanBuilding, Tiers: [][]string{{"messenger"}},
Modules: map[string]*PlanModule{"messenger": {}}, Times: &PlanTimes{Cut: &cut, Class: ClassLeaf}}
p.markWalk(now)
if p.WalkSaid != WalkStarted || p.Times.Started == nil || !p.Times.Started.Equal(now) {
t.Fatalf("a live walk's first save starts it: %q %+v", p.WalkSaid, p.Times)
}
p.State = PlanRolling
p.markWalk(now.Add(time.Minute))
if p.WalkSaid != "" || !p.Times.Started.Equal(now) {
t.Fatalf("a later save says nothing and keeps the start: %q %+v", p.WalkSaid, p.Times)
}
p.State = PlanDone
p.markWalk(now.Add(2 * time.Minute))
if p.WalkSaid != WalkEnded || p.Times.Ended == nil || p.Outcome() != WalkDelivered {
t.Fatalf("the save that finds it done ends it, delivered: %q %+v %q", p.WalkSaid, p.Times, p.Outcome())
}
p.markWalk(now.Add(3 * time.Minute))
if p.WalkSaid != "" {
t.Fatalf("an ended walk saved again says nothing: %q", p.WalkSaid)
}
// Retried, it walks again: a new start.
p.State = PlanBuilding
p.markWalk(now.Add(4 * time.Minute))
if p.WalkSaid != WalkStarted || p.Times.Ended != nil || !p.Times.Started.Equal(now.Add(4*time.Minute)) {
t.Fatalf("a retried walk starts again: %q %+v", p.WalkSaid, p.Times)
}
}
// What is never said: a batch, a walk waiting for its word, a walk that ends without having started, and a walk
// kept before its times were. A release plan is said.
func TestWhatIsNotAWalkStartIsNotSaid(t *testing.T) {
now := time.Now()
for name, p := range map[string]Plan{
"a batch": {State: PlanAssembling, Times: &PlanTimes{}},
"waiting for a word": {State: PlanBuilding, Times: &PlanTimes{}, Delivery: &PlanDelivery{Awaits: "mesh-delivery"}},
"ended, never begun": {State: PlanDone, Times: &PlanTimes{}},
"kept before times": {State: PlanBuilding},
"stopped while waits": {State: PlanFailed, Times: &PlanTimes{}, Delivery: &PlanDelivery{Awaits: "mesh-delivery", Stopped: "x"}},
} {
p.markWalk(now)
if p.WalkSaid != "" {
t.Errorf("%s: said %q", name, p.WalkSaid)
}
}
r := Plan{State: PlanRolling, Release: &PlanRelease{Order: []string{"ace"}}}
r.markWalk(now)
if r.WalkSaid != WalkStarted || r.Times == nil || r.Times.Started == nil || r.Times.Class != "" {
t.Fatalf("a release plan is said started, its times holding only that: %q %+v", r.WalkSaid, r.Times)
}
// Let go, the walk starts.
go1 := now
w := Plan{State: PlanBuilding, Times: &PlanTimes{}, Delivery: &PlanDelivery{Awaits: "mesh-delivery", Go: &go1}}
w.markWalk(now)
if w.WalkSaid != WalkStarted {
t.Fatalf("a walk let go starts: %q", w.WalkSaid)
}
}
// An ended walk's outcome: delivered, failed, or stopped by a person or the delivery's owner, or by a newer walk.
func TestAWalksOutcome(t *testing.T) {
for want, p := range map[string]Plan{
WalkDelivered: {State: PlanDone},
WalkFailed: {State: PlanFailed},
WalkStopped: {State: PlanFailed, Delivery: &PlanDelivery{Stopped: "mesh-delivery"}},
"": {State: PlanRolling},
} {
if got := p.Outcome(); got != want {
t.Errorf("%s: outcome %q, want %q", p.State, got, want)
}
}
if (Plan{State: PlanSuperseded}).Outcome() != WalkStopped {
t.Error("a superseded walk is stopped")
}
}
// The marks are kept with the walk, so a controller replaced mid-walk neither says its start again nor misses
// its end; and a save refused leaves the walk unmarked, so the save that is made says it.
func TestAWalksStartIsKeptAndARefusedSaveLeavesItUnsaid(t *testing.T) {
inv := ForTest(t)
ctx := t.Context()
cut := time.Now().UTC().Add(-time.Minute)
p := Plan{ID: "plan-308", Repository: "novox/mesh-catalog", Commit: "abc", Created: cut, State: PlanBuilding,
Tiers: [][]string{{"messenger"}}, Modules: map[string]*PlanModule{"messenger": {}},
Times: &PlanTimes{Cut: &cut, Class: ClassLeaf}}
var said []string
was := PlanSaved
PlanSaved = func(_ *Inventory, p Plan) { said = append(said, p.WalkSaid) }
t.Cleanup(func() { PlanSaved = was })
if err := inv.SavePlan(ctx, &p); err != nil {
t.Fatal(err)
}
back, err := inv.PlanByID(ctx, p.ID)
if err != nil || back.Times == nil || back.Times.Started == nil || back.Times.Class != ClassLeaf {
t.Fatalf("the start is kept with the walk: %v %+v", err, back.Times)
}
// Read back by another controller and saved: not said again.
back.State = PlanRolling
if err := inv.SavePlan(ctx, &back); err != nil {
t.Fatal(err)
}
// A stale copy ending it is refused, and unmarked.
stale := p
stale.Revision = 1
stale.State = PlanDone
if err := inv.SavePlan(ctx, &stale); err == nil {
t.Fatal("a stale save is refused")
}
if stale.WalkSaid != "" || stale.Times.Ended != nil {
t.Fatalf("a refused save is unmarked: %q %+v", stale.WalkSaid, stale.Times)
}
back.State = PlanDone
if err := inv.SavePlan(ctx, &back); err != nil {
t.Fatal(err)
}
if len(said) != 3 || said[0] != WalkStarted || said[1] != "" || said[2] != WalkEnded {
t.Fatalf("said: %q, want started, nothing, ended", said)
}
}
+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.
mesh-catalog b9de001833b1b61b297182b8e3bcdae1cbdeace9 modules/*/module.json, modules/nats/Dockerfile
mesh-host a76baa4b47f07a8b2bab065603414530bd02b53a examples/foundation-first-node-nats.lock
mesh-host bd5cc6980419c1bd4824be6d58dfebd2381a9de3 examples/foundation-first-node-nats.lock
mesh-sdk 33917f91ac3cc213b879b32f3c70e2aa70395f98 conformance/events/module-event.json
To move them, from this repository's root, with the two repositories checked out beside it:
@@ -161,7 +161,7 @@
"type": "file",
"path": "/var/lib/mesh-bus-conf/accounts.conf",
"mode": "0600",
"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_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.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"
},
{
"id": "broker",
@@ -31,13 +31,6 @@ package apply
// 5. **A mount source that is a refused directory or a refused access is refused with it**: the container
// would otherwise bind the very path the engine would not place, and the runtime creates a missing one
// as root.
// 6. **A file is judged as a directory is** (novox/hq issue 495): never one of the machine's own roots, nor
// in a tree nothing is placed in (rule 2), and never below a directory or an access refused here. A file
// or an archive below a refused path is refused with it, whichever module declares it and in whatever
// order. Rules 3 and 4 are a directory's owner and do not reach a file: a file below /etc is the mesh's
// ordinary configuration, and one below a home is judged by home_links.go and agent_home_guard.go.
// A refused directory also gates the rest of its module, as a step that did not complete does
// (apply.go's `gated`): what a module declares after its directory was written for that directory.
//
// Each is a failed resource with its reason in the node's report; nothing is touched. Paths are judged as
// declared and again with every link in them resolved, so a link at /srv/x pointing at /etc places nothing.
@@ -407,7 +400,6 @@ const (
placingDirectory placing = iota
placingAccess
placingMount
placingFile
)
// refusePath says why a path is not placed or mounted; nil when it may be. module is the resource's module,
@@ -474,7 +466,7 @@ func refuseOne(path string, kind placing, module, owner string) *PlacementRefuse
for _, part := range strings.Split(path, "/") {
ssh = ssh || part == ".ssh"
}
if ssh && (kind == placingAccess || kind == placingMount) {
if ssh && kind != placingDirectory {
return &PlacementRefusedError{Path: path, Why: "it is an account's .ssh, which holds its keys and who may " +
"log in as it, and is never an access or a mount"}
}
@@ -510,12 +502,8 @@ func moduleOfID(id string) string {
return module
}
// refusedPlaces judges every directory, access, file and archive of a declaration before anything is applied,
// and answers each refusal by path: what is refused is refused again as a container's mount source.
//
// Directories and accesses first, so a file or an archive below one of them is refused with it whatever order
// they are declared in (novox/hq issue 495): the engine refused Docker's data root as a directory and then wrote
// a file into it, because nothing judged the file's path.
// refusedPlaces judges every directory and access of a declaration before anything is applied, and answers
// each refusal by path: what is refused is refused again as a container's mount source.
func refusedPlaces(resources []declaration.Resource) map[string]error {
out := map[string]error{}
for _, r := range resources {
@@ -532,60 +520,9 @@ func refusedPlaces(resources []declaration.Resource) map[string]error {
out[filepath.Clean(r.Target())] = err
}
}
trees := make(map[string]error, len(out))
for path, err := range out {
trees[path] = err
}
for _, r := range resources {
var err error
switch res := r.(type) {
case *declaration.File:
if err = belowRefused(res.Path, trees); err == nil {
err = refusePath(res.Path, placingFile, moduleOfID(res.ID), "")
}
case *declaration.Archive:
// Only below a refused path: where an archive may unpack is otherwise the archive's own judgement
// (archive.go), and the engine's own module unpacks its versions into its own tree.
err = belowRefused(res.Path, trees)
default:
continue
}
if err != nil {
out[filepath.Clean(r.Target())] = err
}
}
return out
}
// belowRefused says why a path below a refused directory or access is refused with it; nil when it is below
// none. Judged as declared and with its links resolved, as refusePath judges.
func belowRefused(path string, refused map[string]error) error {
clean := filepath.Clean(path)
if !filepath.IsAbs(clean) {
return nil
}
for _, p := range []string{clean, resolved(clean)} {
// The deepest refused tree it is below, so the same declaration always gives the same reason.
var tree string
for t := range refused {
if below(p, t) && len(t) > len(tree) {
tree = t
}
}
if tree == "" {
continue
}
why := refused[tree]
reason := why.Error()
var refusal *PlacementRefusedError
if errors.As(why, &refusal) {
reason = refusal.Why
}
return &PlacementRefusedError{Path: clean, Why: "it is below " + tree + ", which is refused: " + reason}
}
return nil
}
// refuseMounts says why a container's mounts are not made; nil when they may be.
func refuseMounts(c *declaration.Container, refused map[string]error) error {
for _, v := range c.Volumes {
-9
View File
@@ -14,12 +14,3 @@
./internal/inventory TestAnIssuedValueIsSealedAgainAfterAKeyChangeOrAnAcceptance
./internal/inventory TestAConditionsLooksAreKeptBesideTheBucket
./cmd/mesh-controller TestTheConditionsVerbReadsTheLooksTheBucketNoLongerHolds
./internal/inventory TestAWalksStartIsKeptAndARefusedSaveLeavesItUnsaid
./cmd/mesh-controller TestAWalkLetGoAndStoppedByVerbsIsSaidStartedAndEnded
./cmd/mesh-controller TestAWalkStoppedOrClosedByAPersonIsSaidEndedStopped
./cmd/mesh-controller TestAWalkRetriedByAPersonIsSaidStartedAgain
./cmd/mesh-controller TestABacklogReleasedByAPersonIsSaidStarted
./cmd/mesh-controller TestAWalkEndedByCancelOrClearIsSaidEnded
./cmd/mesh-controller TestAWalkRestartedByARebuildIsSaidStarted
./cmd/mesh-controller TestAWalkStoppedByHandAndRetriedEndsFailedWhenItFailsAgain
./cmd/mesh-controller TestADirectoryOwnersKeyIsRefusedThroughAVerb