Compare commits

..
Author SHA1 Message Date
jschoubben 9902a3f16c Name every group member's trunk in the composed verdict, the announced one when the seat names none (issue 1106) 2026-10-21 21:57:30 +02:00
mesh-tools f973cefd7f Merge pull request 'Ask the operator for a held move and a failed release, and send no image the store lacks (issues 1732, 1320, 1742)' (#428) from fix/1732-held-moves-asked into main 2026-10-21 19:07:51 +00:00
mesh-tools 98155e327b Merge pull request 'Take mesh-sdk at 07a6592: an item's text places a config value (issue 1658), and fit the unit level in its bound (issue 1763)' (#420) from fix/1656-xterm-config into main 2026-10-21 18:36:39 +00:00
jschoubben d6db334840 Offer no Release on a machine that also holds a person's own hold (issue 1732, review round 1)
mesh/repo-check head d6db334840a5 merged into trunk 98155e327b6d beside mesh-host#129@09c77d57 — pass: its merge-check.sh passed
mesh/merge-gate head d6db334840a5 merged into trunk 98155e327b6d — pass: group fix/1732-held-moves-asked composed: every machine composes with the change a…
mesh/delivery delivered
mesh/delivery-group group fix/1732-held-moves-asked delivered: every member is delivered
2026-10-21 20:15:58 +02:00
jschoubben 69d016bfdf Merge remote-tracking branch 'origin/main' into fix/1732-held-moves-asked 2026-10-21 20:15:13 +02:00
jschoubben a58b38cee3 Keep the missing image out of the condition's summary (issue 1742) 2026-10-21 20:06:10 +02:00
jschoubben 9d260412b3 Merge remote-tracking branch 'origin/main' into fix/1656-xterm-config
mesh/repo-check head 9d260412b330 merged into trunk 44033f4b6d72 beside mesh-catalog#416@b9dcf19c — pass: its merge-check.sh passed
mesh/merge-gate head 9d260412b330 merged into trunk 44033f4b6d72 — pass: group fix/1656-xterm-config composed: every machine composes with the change as it…
mesh/delivery-group group fix/1656-xterm-config failed: a member failed
mesh/delivery covered after its failure: a later walk that contains it was delivered: plan-1792609673527907956 (novox/mesh-controller at f973cefd into ma…
2026-10-21 19:59:59 +02:00
jschoubben 99610e9569 Cut the controller's slowest unit tests (a source scan that regexed every file, resolver stand-in timeouts, a wait bound) so mesh-controller's unit level fits its bound (issue 1763) 2026-10-21 19:59:58 +02:00
jschoubben dc06af1f5a Leave a module out of a send when the store lacks its image, and say it (issue 1742) 2026-10-21 19:43:36 +02:00
jschoubben b62fbeabbf Ask a held move and the builds held after a failed release, and say them on the plan's line (issue 1732, 1320) 2026-10-21 19:43:36 +02:00
jschoubben 373aefd14e Take mesh-sdk at 07a6592, which counts an item's text as a placement of a config value (issue 1658)
mesh/delivery-group group fix/1656-xterm-config checking: composed check out of date: novox/mesh-sdk@c8e35097e861 was judged against trunk 85b5a2687e6a, the tr…
mesh/merge-gate building on ace since 19:52
mesh/repo-check out of date: judged against trunk 22c9815f6522, the trunk is now 44033f4b6d72; judged again on a recheck or when it is merged
mesh/delivery superseded: a newer head of the same pull request
2026-10-21 19:43:32 +02:00
jschoubben f66458fe25 Merge remote-tracking branch 'origin/main' into fix/1656-xterm-config 2026-10-21 19:43:02 +02:00
jschoubben 99da867a98 Merge origin/main into fix/1656-xterm-config 2026-10-21 19:16:11 +02:00
jschoubben 69a67876c2 Take mesh-sdk at 6f63275, which counts an item's text as a placement of a config value (issue 1658)
mesh/delivery-group group fix/1656-xterm-config rejected: a member's own check failed
mesh/merge-gate waits for its group's composed check
mesh/repo-check head 69a67876c269 merged into trunk d96c9f56a3f1 beside mesh-catalog#416@f178aab6 — fail: its merge-check.sh failed: --- FAIL: TestModuleCh…
mesh/delivery superseded: a newer head of the same pull request
2026-10-21 17:03:35 +02:00
52 changed files with 1211 additions and 2245 deletions
+4
View File
@@ -645,6 +645,10 @@ func doesWords(act conditions.Action) string {
return "the delivery is stopped"
case act.Verb == broker.ControllerSeat+".plans" && act.Arguments["retry"] != "":
return "the failed builds are asked again"
case act.Verb == broker.ControllerSeat+".push" && act.Arguments["node"] != "":
return "the held builds are sent to " + act.Arguments["node"]
case act.Verb == broker.ControllerSeat+".upgrade" && act.Arguments["release-backlog"] != "":
return "the builds held after the failed delivery are released"
case strings.HasSuffix(act.Verb, ".restart"):
return "its service is restarted on " + act.Machine
}
+14 -2
View File
@@ -594,11 +594,13 @@ func checkedOf(result link.BuildResult) link.Checked {
// A delivery group's composed check (novox/hq ADR 0239): every head it judged, this one first.
if g := result.Checked; g.Group != "" {
c.Group = g.Group
// Every member is named with its trunk commit (novox/hq issue 1106): the seat's word, else the one the forge
// announced with the head, so the forge's merge tool never meets a member whose verdict names none.
c.Members = append(c.Members, link.CheckedMember{Owner: g.Owner, Repo: g.Repo, Number: g.Number,
Commit: result.Ref, Trunk: c.Trunk})
Commit: result.Ref, Trunk: firstTrunk(c.Trunk, g.Trunk)})
for _, m := range g.Members {
c.Members = append(c.Members, link.CheckedMember{Owner: m.Owner, Repo: m.Repo, Number: m.Number,
Commit: m.Ref, Trunk: memberTrunk(result.Check, m)})
Commit: m.Ref, Trunk: firstTrunk(memberTrunk(result.Check, m), m.Trunk)})
}
}
}
@@ -627,6 +629,16 @@ func checkedOf(result link.BuildResult) link.Checked {
return c
}
// firstTrunk is the first trunk commit named, as twelve characters; empty when none is.
func firstTrunk(named ...string) string {
for _, n := range named {
if n != "" {
return twelve(n)
}
}
return ""
}
// memberTrunk is the trunk commit the build seat merged one of a group's heads into (novox/hq ADR 0323 §6), empty
// when it said none.
func memberTrunk(outcome *link.CheckOutcome, m link.GroupMember) string {
+25
View File
@@ -250,6 +250,31 @@ func TestAVerdictNamesTheTrunkCommitItJudged(t *testing.T) {
}
}
// **A group's composed verdict names every member's trunk commit** (novox/hq issue 1106): a member the build seat named no
// trunk for (an SDK member whose base the seat did not merge, or a seat from before the rule) is named by the trunk the forge
// announced with its head, and the primary the same, so the merge tool never reads "names no trunk commit" on a member
// whose trunk did not move.
func TestAGroupVerdictNamesEveryMembersTrunkTheSDKMemberToo(t *testing.T) {
group := &link.CheckRequest{Owner: "novox", Repo: "mesh-catalog", Number: 7, Base: "main", Group: "feat/x",
Trunk: "dddddddddddd0000",
Members: []link.GroupMember{
{Owner: "novox", Repo: "mesh-sdk", Number: 8, Ref: "cccccccccccc0000", Base: "main", Trunk: "eeeeeeeeeeee0000"},
{Owner: "novox", Repo: "mesh-host", Number: 9, Ref: "bbbbbbbbbbbb0000", Base: "main", Trunk: "ffffffffffff0000"}}}
c := checkedOf(link.BuildResult{ID: "g", Ref: "aaaaaaaaaaaa0000", Checked: group, Check: &link.CheckOutcome{
Verdict: "pass", Head: "aaaaaaaaaaaa", Gate: &link.CheckLayer{Verdict: "pass"},
Members: []link.CheckedMember{{Owner: "novox", Repo: "mesh-host", Commit: "bbbbbbbbbbbb", Trunk: "222222222222"}}}})
want := []string{"dddddddddddd", "eeeeeeeeeeee", "222222222222"}
if len(c.Members) != 3 {
t.Fatalf("the composed check names its members %+v", c.Members)
}
for i, m := range c.Members {
if m.Trunk != want[i] {
t.Fatalf("member %d (%s/%s) is named with trunk %q, want %q (the seat's word first, else the announced one): %+v",
i, m.Owner, m.Repo, m.Trunk, want[i], c.Members)
}
}
}
// **The merge made for a check is never registered** (novox/hq ADR 0323 §2): a build of it — a commit on no branch
// of the forge — is refused, naming the trunk. The rule of ADR 0238 §1 stands unchanged; this holds it for the
// merge commit a check makes.
+5 -1
View File
@@ -239,7 +239,11 @@ func markUnresolvable(err error) error {
// still names anything foreseen is refused, said, and never sent.
func composeSend(ctx context.Context, open *stores, node string) (sendable, error) {
declared, _, err := composeSendValidated(ctx, open, node)
return declared, err
if err != nil {
return declared, err
}
// Never a declaration naming an image the store does not hold (novox/hq issue 1742): that module is left out.
return withoutMissingImages(ctx, open, node, declared), nil
}
// composeSendValidated is composeSend and what the node-engine's validator refuses of the declaration it composed, for the
+1 -1
View File
@@ -1204,7 +1204,7 @@ func askGroupCheck(ctx context.Context, open *stores, group string, members []or
return "", fmt.Errorf("%s cannot be cloned for the group's check: %w", m.ID, err)
}
request.Check.Members = append(request.Check.Members, link.GroupMember{Owner: mp.Owner, Repo: mp.Repo,
Number: mp.Number, Repository: url, Ref: mp.Commit, Paths: mp.Paths, Base: mp.Base})
Number: mp.Number, Repository: url, Ref: mp.Commit, Paths: mp.Paths, Base: mp.Base, Trunk: mp.BaseSHA})
}
seat := buildSeatHeld(ctx)
ask, err := askOverOn(seat)
+1 -1
View File
@@ -68,7 +68,7 @@ func TestTheForgeNotAnsweringInTimeIsNotAskedAgain(t *testing.T) {
// An ask that hangs past the bound is cut off at the bound, not at its own wait, and the failure still says it
// waited: the receive loop is held no longer than the bound (novox/hq issue 486).
func TestAnAskHangingPastTheBoundIsCutOffAtIt(t *testing.T) {
w := waitOut{bound: 200 * time.Millisecond, first: 10 * time.Millisecond, most: 50 * time.Millisecond,
w := waitOut{bound: 100 * time.Millisecond, first: 10 * time.Millisecond, most: 50 * time.Millisecond,
now: time.Now, sleep: sleepOrDone}
asked := 0
began := time.Now()
+198
View File
@@ -0,0 +1,198 @@
package main
import (
"context"
"fmt"
"sort"
"strings"
"github.com/novox/mesh-controller/internal/conditions"
"github.com/novox/mesh-controller/internal/inventory"
)
// A move held for a person is asked of the person (novox/hq issue 1732, issue 1320; ADR 0377, ADR 0259).
//
// **A wait only a person can end is said to that person at once.** ADR 0377 holds a recorded module's move that
// recreates a container or touches its data until a person pushes the machine. The walk said so in its note and
// ended "done", `status` named the machine behind, and nobody was asked: on 2026-10-21 mailu's new build waited on
// novox for 40 minutes behind a plan that read "done, 3 tier(s)", and on 2026-10-19 mssql waited on ace the same way.
// Row S24 raises one condition per machine that has such a move, for the operator, with the answer **Release**: the
// push of that machine, which the controller's own `push` verb performs on the operator's warrant (ADR 0259). It ends
// when the machine no longer waits, whoever pushed it.
//
// A hold a person chose is not asked again — a module they set to `record` with a why, and the bus, whose step has
// its own row (S17).
// kindMovesHeld is S24's kind: machine.<node>.moves-held.
const kindMovesHeld = "moves-held"
// heldFacts is one machine and the moves of recorded modules that wait there for a person.
type heldFacts struct {
node string
moves []heldMove
}
// gatherHeldMoves is, per machine that holds at least one move no person chose to hold, all its held moves, by machine.
func gatherHeldMoves(ctx context.Context, inv *inventory.Inventory) ([]heldFacts, error) {
nodes, err := inv.Nodes(ctx)
if err != nil {
return nil, fmt.Errorf("the machines cannot be read: %w", err)
}
f, err := readMoveFacts(ctx, inv)
if err != nil {
return nil, err
}
var out []heldFacts
for _, n := range nodes {
moves, err := heldMovesOn(ctx, inv, f, n.Name)
if err != nil {
return nil, err
}
h := heldFacts{node: n.Name, moves: moves}
if len(h.asked()) > 0 {
out = append(out, h)
}
}
return out, nil
}
// asked are the moves no person chose to hold, and chosen the ones a person did (a module set to record with a why, the bus).
func (h heldFacts) asked() []heldMove { return h.byChoice(false) }
func (h heldFacts) chosen() []heldMove { return h.byChoice(true) }
func (h heldFacts) byChoice(chosen bool) []heldMove {
var out []heldMove
for _, m := range h.moves {
if m.Chosen == chosen {
out = append(out, m)
}
}
return out
}
// modules are the held moves' modules, in order, of all of them.
func (h heldFacts) modules() []string { return moduleNames(h.moves) }
func moduleNames(moves []heldMove) []string {
out := make([]string, 0, len(moves))
for _, m := range moves {
out = append(out, m.Module)
}
return out
}
// releaseAction is the one answer a held move offers: the push of the machine, approved by the operator. The push
// names every module whose move waits there as one that may move its data: the operator's Approve is the word
// ADR 0217 asks for, and the ask says so. Offered only where no person's own hold is on the machine too: a push
// composes every recorded module at its current build, so it would send that move as well, which the ask never named
// (ADR 0259 §6), and hold the whole machine if that module's data would move.
func (h heldFacts) releaseAction() conditions.Action {
return conditions.Action{Label: "Release", Verb: "mesh-controller.push", Level: conditions.LevelApprove,
Arguments: map[string]string{"node": h.node, "move": strings.Join(moduleNames(h.asked()), ","), "why": "",
"cause": conditions.CauseOperatorAnswer}}
}
// heldWords are the plain words of a machine's held moves: which modules wait, why in general, and what an Approve does.
func heldWords(h heldFacts) (headline, explanation, needs string, resolved string) {
names := namesWords(moduleNames(h.asked()), 3)
headline = clipWords(conditions.Capital(names)+" waits for you on "+h.node, conditions.HeadlineMax)
recreates, data := false, false
for _, m := range h.asked() {
switch {
case strings.Contains(m.Why, "recreates"):
recreates = true
default:
data = true
}
}
why := "would change what it runs"
switch {
case recreates && data:
why = "would restart a container or touch its data"
case recreates:
why = "would restart one of its containers"
case data:
why = "would touch its files or data"
}
explanation = clipWords(fmt.Sprintf("The new version of %s %s on %s, so it does not go out with its delivery; "+
"it waits for you. Release sends %s everything it is owed, these among it, and lets their data move where it "+
"has to. Nothing is sent until you answer.", names, why, h.node, h.node), conditions.ExplanationMax)
needs = "release them " + FromMeshMCPServer
if chosen := h.chosen(); len(chosen) > 0 {
// A push of the machine would send the move a person chose to hold as well: no answer is offered here.
explanation = clipWords(fmt.Sprintf("The new version of %s %s on %s, so it does not go out with its delivery; "+
"it waits for you. %s is also held there by your own choice, and a push of %s would send that too, so "+
"nothing is offered here: the push is yours at the controller's terminal.",
names, why, h.node, conditions.Capital(namesWords(moduleNames(chosen), 3)), h.node), conditions.ExplanationMax)
needs = fmt.Sprintf("push %s at the controller's terminal, once you have decided about %s.", h.node,
namesWords(moduleNames(chosen), 3))
}
resolved = "Resolved: " + names + " is no longer waiting on " + h.node
return
}
// watchMovesHeld is S24: a machine that has moves of recorded modules waiting for a person, said to the operator with the
// answer that sends them. The condition ends when the machine no longer waits: it is no longer among the facts.
func watchMovesHeld(f *signalFacts) []conditions.Observation {
var out []conditions.Observation
for _, h := range f.held {
var said []string
for _, m := range h.moves {
said = append(said, fmt.Sprintf("%s %s → %s %s", m.Module, short(m.From), short(m.To), m.Why))
}
headline, explanation, needs, resolved := heldWords(h)
if c := h.chosen(); len(c) > 0 {
said = append(said, "held by a person's own choice: "+strings.Join(moduleNames(c), ", "))
}
var actions []conditions.Action
if len(h.chosen()) == 0 {
actions = []conditions.Action{h.releaseAction()}
}
out = append(out, conditions.Observation{Scope: conditions.ScopeMachine, ID: h.node, Kind: kindMovesHeld,
Machine: h.node, Severity: conditions.Warning, Resolver: conditions.ResolverOperator,
Summary: fmt.Sprintf("%s has %d move(s) of recorded modules waiting for a person's push: %s — `push %s` sends them",
h.node, len(h.moves), strings.Join(h.modules(), ", "), h.node),
Said: strings.Join(said, "; "),
Headline: headline, Explanation: explanation, Needs: needs, Resolved: resolved,
Actions: actions})
}
return out
}
// planHeld is, for a plan that ended or is running, the moves its modules carried that a machine still holds for a
// person: "module on machine", sorted. The plan's own record says the move waited when it was judged
// (PlanModule.Recorded); the machines say whether it still does, and for the build this plan carried, so a plan older
// than the build now held is not named.
func planHeld(p inventory.Plan, held []heldFacts) []string {
var out []string
for _, h := range held {
for _, m := range h.moves {
s := p.Modules[m.Module]
if s == nil || !strings.HasPrefix(s.Recorded[h.node], "waits:") || !sameCommit(s.Commit, m.To) {
continue
}
out = append(out, m.Module+" on "+h.node)
}
}
sort.Strings(out)
return out
}
// heldSaid is what a plan's line adds when it carries a move that a machine holds for a person (novox/hq issue 1732): a
// plan whose module waits for a push is not plainly "done".
func heldSaid(p inventory.Plan, held []heldFacts) string {
waiting := planHeld(p, held)
if len(waiting) == 0 {
return ""
}
return fmt.Sprintf(" — %d held for a person: %s (the operator is asked)", len(waiting), strings.Join(waiting, ", "))
}
// heldNow reads the machines' held moves for a plan's line, nil when they cannot be read: the line then reads as before.
func heldNow(ctx context.Context, inv *inventory.Inventory) []heldFacts {
held, err := gatherHeldMoves(ctx, inv)
if err != nil {
return nil // empty-on-error: a plan's line then reads as it did before; the watchdog (S24) says it cannot read
}
return held
}
@@ -0,0 +1,92 @@
//go:build integration
package main
import (
"context"
"strings"
"testing"
"time"
"github.com/novox/mesh-controller/internal/inventory"
)
func inventoryUpgrade(why string) inventory.Upgrade { return inventory.Upgrade{Why: why, By: "jochen"} }
func mustMoveFacts(t *testing.T, rm *recordedMesh) moveFacts {
t.Helper()
f, err := readMoveFacts(t.Context(), rm.open.inventory)
if err != nil {
t.Fatal(err)
}
return f
}
// 1732/AC8 (integration): a mesh running mailu at c1 on two machines, whose registered c2 recreates a container, is read
// as two machines each holding mailu's move for a person; a bundle-only c3 on top changes nothing of it; after a machine
// is sent c2 (as a push records it) it is no longer among them, and plans that carried c2 say nothing more of it.
func TestMachinesHoldingAMoveForAPersonAreReadFromTheRecords(t *testing.T) {
rm := aRecordedMesh(t, aMailu(true))
c2 := aMailu(true)
resourceOf(&c2, "imap")["image"] = "registry.invalid/imap@sha256:2"
rm.register(t, c2, "c2", time.Now().Add(-time.Minute))
ctx := t.Context()
held, err := gatherHeldMoves(ctx, rm.open.inventory)
if err != nil {
t.Fatal(err)
}
if len(held) != 2 || held[0].node != "anchor" || held[1].node != "laptop" || len(held[0].moves) != 1 ||
held[0].moves[0].Module != "mailu" || held[0].moves[0].To != "c2" || !strings.Contains(held[0].moves[0].Why, "recreates") {
t.Fatalf("held %+v", held)
}
if err := rm.open.inventory.RecordSent(ctx, nodeID(t, rm.open, "anchor"), "d-anchor-c2", map[string]string{"mailu": "c2"}); err != nil {
t.Fatal(err)
}
held, err = gatherHeldMoves(ctx, rm.open.inventory)
if err != nil || len(held) != 1 || held[0].node != "laptop" {
t.Fatalf("after anchor was sent c2: %+v (%v)", held, err)
}
}
// 1732/AC9 (integration): a module a person set to record is no ask: they chose the hold.
func TestAHoldAPersonChoseIsNotAsked(t *testing.T) {
rm := aRecordedMesh(t, aMailu(false))
c2 := aMailu(false)
c2.Bundles[0].Digest = "sha256:bbb"
rm.register(t, c2, "c2", time.Now().Add(-time.Minute))
if err := rm.open.inventory.SetUpgradeOf(t.Context(), "mailu", inventoryUpgrade("the migration week")); err != nil {
t.Fatal(err)
}
held, err := gatherHeldMoves(t.Context(), rm.open.inventory)
if err != nil || len(held) != 0 {
t.Fatalf("a machine holding only a person's own hold is asked: %+v (%v)", held, err)
}
if moves, _ := heldMovesOn(t.Context(), rm.open.inventory, mustMoveFacts(t, rm), "anchor"); len(moves) != 1 || !moves[0].Chosen {
t.Fatalf("status still names it, marked as chosen: %+v", moves)
}
}
// 1742/AC4 (integration): a registered build naming an image of the store that the store lacks is read as a missing
// image on each machine that runs the module; one the store holds is not; and what the store holds is not asked twice.
func TestARegisteredBuildNamingAnImageTheStoreLacksIsRead(t *testing.T) {
m := aMailu(true)
resourceOf(&m, "imap")["image"] = "artifact-store://mailu/imap@sha256:" + strings.Repeat("e", 64)
rm := aRecordedMesh(t, m)
asked := 0
lacking := func(_ context.Context, reference string) (bool, error) { asked++; return false, nil }
imagesHeld.Lock()
imagesHeld.held = map[string]bool{}
imagesHeld.Unlock()
got, err := gatherMissingImages(t.Context(), rm.open.inventory, lacking)
if err != nil || len(got) != 2 || got[0].Module != "mailu" || got[0].Node != "anchor" || got[1].Node != "laptop" {
t.Fatalf("missing %+v (%v)", got, err)
}
holding := func(_ context.Context, reference string) (bool, error) { asked++; return true, nil }
if got, err := gatherMissingImages(t.Context(), rm.open.inventory, holding); err != nil || len(got) != 0 {
t.Fatalf("an image the store holds is missing: %+v (%v)", got, err)
}
before := asked
if _, err := gatherMissingImages(t.Context(), rm.open.inventory, lacking); err != nil || asked != before {
t.Fatalf("an image the store was found to hold was asked about again (%d, %v)", asked-before, err)
}
}
+237
View File
@@ -0,0 +1,237 @@
package main
import (
"context"
"errors"
"strings"
"testing"
"time"
"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/inventory"
)
// **A move held for a person is asked of the person** (novox/hq issue 1732, issue 1320; ADR 0377, ADR 0259).
// aHeldMailu is novox holding mailu's new build because it would recreate a container: the walk's own words.
func aHeldMailu() heldFacts {
return heldFacts{node: "novox", moves: []heldMove{{Module: "mailu", From: "b2f5ebb2aaaa", To: "3259765dbbbb",
Why: "waits: recreates 1 of mailu's 12 container(s) (with a new image: automx)"}}}
}
func heldConditionOn(f *signalFacts) conditions.Condition {
o := watchMovesHeld(f)[0]
return conditions.Condition{Key: o.Key(), Kind: o.Kind, Severity: o.Severity, Headline: o.Headline,
Explanation: conditions.Verdict(o.Needs, o.Explanation), Needs: o.Needs, Actions: o.Actions}
}
// 1732/AC1: a machine holding a move of a recorded module for a person raises one condition for the operator, with
// Release as its only answer: the machine's push, approved, naming the modules whose moves wait there.
func TestAHeldMoveIsRaisedWithReleaseAsItsOnlyAction(t *testing.T) {
f := calm(time.Date(2026, 10, 21, 17, 30, 0, 0, time.UTC))
f.held = []heldFacts{aHeldMailu()}
got := watchMovesHeld(f)
if len(got) != 1 {
t.Fatalf("%+v", got)
}
o := got[0]
if o.Kind != "moves-held" || o.Scope != conditions.ScopeMachine || o.ID != "novox" || o.Severity != conditions.Warning ||
o.Resolver != conditions.ResolverOperator || o.Key() != "machine.novox.moves-held" {
t.Fatalf("kind, scope, subject, severity or resolver: %+v (%s)", o, o.Key())
}
if len(o.Actions) != 1 {
t.Fatalf("actions: %+v", o.Actions)
}
a := o.Actions[0]
if a.Label != "Release" || a.Verb != "mesh-controller.push" || a.Level != conditions.LevelApprove ||
a.Arguments["node"] != "novox" || a.Arguments["move"] != "mailu" || a.Arguments["cause"] != conditions.CauseOperatorAnswer {
t.Fatalf("the action is not the push of novox: %+v", a)
}
if _, takes := a.Arguments["why"]; !takes {
t.Fatal("the push takes a why, which the asker fills in with the operator's answer")
}
for _, want := range []string{"mailu", "novox"} {
if !strings.Contains(o.Headline+o.Explanation, want) {
t.Errorf("the words do not name %q: %q / %q", want, o.Headline, o.Explanation)
}
}
if !strings.Contains(o.Said, "recreates 1 of mailu's 12") {
t.Errorf("the evidence does not carry the walk's own words: %q", o.Said)
}
if w := (conditions.Observation{Headline: o.Headline, Explanation: o.Explanation, Needs: o.Needs, Resolved: o.Resolved}); w.Headline == "" || len(o.Headline) > conditions.HeadlineMax {
t.Errorf("headline: %q", o.Headline)
}
}
// 1732/AC2: a machine with nothing held raises nothing, so the condition ends the look after the move is sent.
func TestAMachineHoldingNothingRaisesNoHeldMoveCondition(t *testing.T) {
f := calm(time.Date(2026, 10, 21, 17, 30, 0, 0, time.UTC))
if got := watchMovesHeld(f); len(got) != 0 {
t.Fatalf("%+v", got)
}
}
// 1732/AC3: the condition is asked on a channel at the approve level in words that say what Release does, and the
// operator's warrant calls the controller's own push with the machine and the modules, once.
func TestAHeldMovesReleaseIsAskedAndAWarrantPushesTheMachine(t *testing.T) {
f := calm(time.Date(2026, 10, 21, 17, 30, 0, 0, time.UTC))
f.held = []heldFacts{aHeldMailu()}
c := heldConditionOn(f)
r := newAskerRig(t)
r.open = []conditions.Condition{c}
if err := r.a.reconcile(context.Background()); err != nil {
t.Fatal(err)
}
sent := r.asksSent(t)
if len(sent) != 1 || len(sent[0].Options) != 1 {
t.Fatalf("asked %+v", sent)
}
o := sent[0].Options[0]
if o.Label != "Release" || o.Level != asks.Approve || o.Does != "the held builds are sent to novox" {
t.Errorf("the option is %+v", o)
}
if err := sent[0].Check(r.now); err != nil {
t.Errorf("the ask is refused as made: %v", err)
}
w := r.warrantFor(t, c.Key, "Release")
answerWith(t, r, w)
answerWith(t, r, w)
if len(r.called) != 1 || !strings.HasPrefix(r.called[0], "mesh-controller.push@ ") ||
!strings.Contains(r.called[0], `"node":"novox"`) || !strings.Contains(r.called[0], `"move":"mailu"`) ||
!strings.Contains(r.called[0], `"cause":"`+conditions.CauseOperatorAnswer+`"`) ||
!strings.Contains(r.called[0], "chose Release") {
t.Fatalf("called %v", r.called)
}
}
// 1732/AC4: the controller may call its own push and upgrade verbs on a warrant: the grant names them.
func TestTheControllerMayActOnAWarrantForPushAndUpgrade(t *testing.T) {
for _, verb := range []string{"push", "upgrade"} {
granted := false
for _, v := range broker.VerbsTheControllerActsOnAWarrant {
granted = granted || (v.Seat == broker.ControllerSeat && v.Verb == verb)
}
if !granted {
t.Errorf("the controller may not act on a warrant for %s", verb)
}
}
}
func aDoneMailuPlan() inventory.Plan {
return inventory.Plan{ID: "plan-1792585970446304628", Repository: "novox/mesh-catalog", State: inventory.PlanDone,
Tiers: [][]string{{"a"}, {"b"}, {"mailu"}},
Modules: map[string]*inventory.PlanModule{"mailu": {State: "built", Commit: "3259765dbbbbcccc",
Recorded: map[string]string{"novox": "waits: recreates 1 of mailu's 12 container(s)"}}}}
}
// 1732/AC5: a done plan that carries a move a machine still holds for a person says so on its line; once the machine
// no longer holds it, or holds another build of the module, the line says nothing more.
func TestADonePlanSaysWhatItHoldsForAPerson(t *testing.T) {
p := aDoneMailuPlan()
line := planLine(p, time.Date(2026, 10, 21, 16, 0, 0, 0, time.UTC)) + heldSaid(p, []heldFacts{aHeldMailu()})
if !strings.Contains(line, "done, 3 tier(s)") || !strings.Contains(line, "1 held for a person: mailu on novox") {
t.Errorf("the line: %q", line)
}
if said := heldSaid(p, nil); said != "" {
t.Errorf("a plan whose moves were sent says %q", said)
}
other := aHeldMailu()
other.moves[0].To = "9999999ffff"
if said := heldSaid(p, []heldFacts{other}); said != "" {
t.Errorf("a plan older than the build now held says %q", said)
}
p.Modules["mailu"].Recorded = map[string]string{"novox": recordedRollsWhy}
if said := heldSaid(p, []heldFacts{aHeldMailu()}); said != "" {
t.Errorf("a move the plan sent says %q", said)
}
}
// 1732/AC6: the builds held after a failed release are asked of the operator: the held-release condition offers
// Release, which calls the controller's own `upgrade` with release-backlog and nothing else, once, on a warrant.
func TestTheBuildsHeldAfterAFailedReleaseAreAskedAndReleasedOnAWarrant(t *testing.T) {
saved := backlogNow
t.Cleanup(func() { backlogNow = saved })
backlogNow.held = "release-1792595552499982311 failed (failed on novox); what waits is released again by a person"
backlogNow.waiting = map[string][]inventory.CarriedMove{"novox": {{Module: "indexer"}, {Module: "nextcloud"}}}
got := backlogObservation()
if len(got) != 1 || len(got[0].Actions) != 1 {
t.Fatalf("raised %+v", got)
}
a := got[0].Actions[0]
if a.Label != "Release" || a.Verb != "mesh-controller.upgrade" || a.Level != conditions.LevelApprove ||
a.Arguments["release-backlog"] != "true" || a.Arguments["cause"] != conditions.CauseOperatorAnswer {
t.Fatalf("the action: %+v", a)
}
o := got[0]
c := conditions.Condition{Key: o.Key(), Kind: o.Kind, Severity: o.Severity, Headline: o.Headline,
Explanation: conditions.Verdict(o.Needs, o.Explanation), Needs: o.Needs, Actions: o.Actions}
r := newAskerRig(t)
r.open = []conditions.Condition{c}
if err := r.a.reconcile(context.Background()); err != nil {
t.Fatal(err)
}
sent := r.asksSent(t)
if len(sent) != 1 || len(sent[0].Options) != 1 || sent[0].Options[0].Does != "the builds held after the failed delivery are released" {
t.Fatalf("asked %+v", sent)
}
w := r.warrantFor(t, c.Key, "Release")
answerWith(t, r, w)
answerWith(t, r, w)
if len(r.called) != 1 || !strings.HasPrefix(r.called[0], "mesh-controller.upgrade@ ") ||
!strings.Contains(r.called[0], `"release-backlog":"true"`) || !strings.Contains(r.called[0], "chose Release") {
t.Fatalf("called %v", r.called)
}
}
// 1732/AC7: a plan whose send waits for a gate while the builds are held after a failed release reads as waiting for
// a person to release them, not "building ... LATE"; with nothing held it still reads as the wait it is.
func TestAPlanWaitingForAGateBehindAFailedReleaseIsNotLate(t *testing.T) {
saved := backlogNow
t.Cleanup(func() { backlogNow = saved })
now := time.Date(2026, 10, 21, 18, 50, 0, 0, time.UTC)
p := inventory.Plan{ID: "plan-1", Repository: "novox/mesh-sdk", Commit: "c0ffee00", State: inventory.PlanBuilding,
Tiers: [][]string{{"indexer"}}, TierEntered: now.Add(-100 * time.Minute), Created: now.Add(-110 * time.Minute),
Modules: map[string]*inventory.PlanModule{"indexer": {State: "built"}},
Note: refusalNote(0, errors.New("a build waits there for its gate: novox"))}
backlogNow.held = ""
if line := planLine(p, now); !strings.Contains(line, "a build waits there for its gate: novox") {
t.Errorf("the wait is not said: %q", line)
}
backlogNow.held = "release-1 failed (failed on novox); what waits is released again by a person"
line := planLine(p, now)
if strings.Contains(line, "LATE") || strings.Contains(line, "building") ||
!strings.Contains(line, "until a person releases it") {
t.Errorf("a plan held behind a failed release reads %q", line)
}
}
// 1732/AC10: a machine that also holds a move a person chose to hold gets no Release, since a push would send that move
// too; the condition names both and says the push is the controller's terminal. A machine with only asked moves keeps it.
func TestAMachineAlsoHoldingAPersonsOwnHoldIsOfferedNoRelease(t *testing.T) {
f := calm(time.Date(2026, 10, 21, 17, 30, 0, 0, time.UTC))
h := aHeldMailu()
h.moves = append(h.moves, heldMove{Module: "nftables", From: "n1aaaa", To: "n2bbbb",
Why: "waits: a person set it to record, and holds every move of it until they push", Chosen: true})
f.held = []heldFacts{h}
got := watchMovesHeld(f)
if len(got) != 1 || len(got[0].Actions) != 0 {
t.Fatalf("%+v", got)
}
o := got[0]
text := o.Headline + o.Explanation + o.Needs + o.Summary + o.Said
for _, want := range []string{"mailu", "nftables", "terminal", "push novox"} {
if !strings.Contains(text, want) {
t.Errorf("does not say %q: %q", want, text)
}
}
w := conditions.Words{Headline: o.Headline, Explanation: o.Explanation, Resolved: o.Resolved, Needs: o.Needs}
if why, ok := conditions.PlainWords(w, "novox"); !ok {
t.Errorf("not plain: %s", why)
}
f.held = []heldFacts{aHeldMailu()}
if got := watchMovesHeld(f); len(got) != 1 || len(got[0].Actions) != 1 || got[0].Actions[0].Arguments["move"] != "mailu" {
t.Fatalf("a machine with only asked moves: %+v", got)
}
}
+245
View File
@@ -0,0 +1,245 @@
package main
import (
"context"
"fmt"
"slices"
"sort"
"strings"
"sync"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"github.com/novox/mesh-controller/internal/artifacts"
"github.com/novox/mesh-controller/internal/conditions"
"github.com/novox/mesh-controller/internal/inventory"
)
// No declaration is sent that names an image the artifact store does not hold (novox/hq issue 1742).
//
// On 2026-10-21 at 17:13 a send to novox named `mailu/automx@sha256:c62150b8…`; the build record keeps that digest
// ("recent-build") and the store did not hold it. The node-engine removed the running container, then could not pull
// the replacement, and every later apply on novox failed for the next hour while the mail clients' autoconfiguration
// had no backend. The node-engine's half (pull first, replace after) is its own change; this is the controller's: **a
// module whose image the store lacks is left out of the declaration**, as a module whose setting nobody gave is
// (ADR 0163, rule 6) — the node-engine keeps what it has of it and touches none of its containers — and **it is said**,
// row S25: a condition naming the module, the machine and the image, until the store holds the image again (a new build
// of the module, or the copy put back).
//
// Judged only for an image the store's own address names, from the send's own resources; an image of another registry
// is no business of the store's. A store that cannot be asked judges nothing: a send is not refused for a question
// that did not answer.
// kindImageMissing is S25's kind: module.<module>.<machine>.image-missing.
const kindImageMissing = "image-missing"
// missingImage is a module held out of a machine's declaration because the store lacks an image it names.
type missingImage struct {
Node, Module, Image string
}
// storeHolds is whether the store holds the manifest an image reference at its address names.
type storeHolds func(ctx context.Context, reference string) (bool, error)
// imageInStore is the artifact-store reference of an image the store's address names, "" for any other.
func imageInStore(image, address string) string {
if address == "" || !strings.HasPrefix(image, address+"/") || !strings.Contains(image, "@sha256:") {
return ""
}
reference := catalogue.Recorded(image)
if _, ours := catalogue.InArtifactStore(reference); !ours {
return ""
}
return reference
}
// holdBackMissingImages leaves out of the declaration every module that names an image the store does not hold,
// and answers what it left out. The module's resources go, it joins LeftOut with the reason, and its build stays the
// one the machine was last sent (`was`).
func holdBackMissingImages(ctx context.Context, node string, s *sendable, address string, holds storeHolds,
was map[string]string) []missingImage {
lacks := map[string]bool{} // reference → the store does not hold it; absent: not asked, or no answer
asked := map[string]bool{}
lacking := map[string]string{} // module → the image
for _, r := range s.Resources {
if fmt.Sprint(r["type"]) != "container" {
continue
}
image, _ := r["image"].(string)
reference := imageInStore(image, address)
module := s.moduleOf[fmt.Sprint(r["id"])]
if reference == "" || module == "" || lacking[module] != "" {
continue
}
if !asked[reference] {
asked[reference] = true
held, err := holds(ctx, reference)
// empty-on-error: the store did not answer, so nothing is held back for a question unanswered
lacks[reference] = err == nil && !held
}
if lacks[reference] {
lacking[module] = image
}
}
if len(lacking) == 0 {
return nil
}
kept := s.Resources[:0:0]
for _, r := range s.Resources {
if _, out := lacking[s.moduleOf[fmt.Sprint(r["id"])]]; !out {
kept = append(kept, r)
}
}
s.Resources = kept
if s.leftOutWhy == nil {
s.leftOutWhy = map[string]string{}
}
var found []missingImage
for module, image := range lacking {
s.leftOutWhy[module] = "its image " + image + " is not in the artifact store"
if !slices.Contains(s.LeftOut, module) {
s.LeftOut = append(s.LeftOut, module)
}
if s.Builds != nil {
if prior, sent := was[module]; sent {
s.Builds[module] = prior
} else {
delete(s.Builds, module)
}
}
found = append(found, missingImage{Node: node, Module: module, Image: image})
}
sort.Strings(s.LeftOut)
sort.Slice(found, func(i, j int) bool { return found[i].Module < found[j].Module })
return found
}
// withoutMissingImages is composeSend's last step: the store asked about the images the declaration names, the modules
// whose image it lacks left out and kept in the ledger S25 reads. A store that cannot be found or asked leaves the
// declaration as composed.
func withoutMissingImages(ctx context.Context, open *stores, node string, s sendable) sendable {
inv := open.inventory
shelf, err := inv.Catalogue(ctx)
if err != nil {
return s
}
address, err := artifactStoreAddress(ctx, inv, shelf, node)
if err != nil || address == "" {
return s
}
sent, known, err := inv.SentBuilds(ctx, node)
if err != nil || !known {
sent = nil
}
store := artifacts.Store{Address: address}
found := holdBackMissingImages(ctx, node, &s, address, store.HoldsManifest, sent)
for _, m := range found {
fmt.Printf("%s: %s is left out of the declaration: its image %s is not in the artifact store\n", node, m.Module, m.Image)
}
return s
}
// imageFacts are the images S25 raises: a module whose registered build names an image the store lacks, on each machine
// that runs it.
type imageFacts struct {
missingImage
}
// imagesHeld remembers the images the store was found to hold: a stored image is named by its digest and stays what it
// is, so it is asked about once; one it lacks is asked again at every look.
var imagesHeld = struct {
sync.Mutex
held map[string]bool
}{held: map[string]bool{}}
// gatherMissingImages is every machine's module whose registered build — the one a send would carry — names an image of
// the artifact store that the store does not hold. Read from the catalogue and the store, never from what a send did:
// a push is a command of its own, and the serving controller's watchdog hears nothing of it.
func gatherMissingImages(ctx context.Context, inv *inventory.Inventory, holds storeHolds) ([]imageFacts, error) {
shelf, err := inv.Catalogue(ctx)
if err != nil {
return nil, fmt.Errorf("the catalogue cannot be read: %w", err)
}
var out []imageFacts
for _, module := range sortedKeysOf(modulesOf(shelf)) {
for _, r := range shelf[module].Resources {
if fmt.Sprint(r["type"]) != "container" {
continue
}
image, _ := r["image"].(string)
if _, ours := catalogue.InArtifactStore(image); !ours || !strings.Contains(image, "@sha256:") {
continue
}
imagesHeld.Lock()
known := imagesHeld.held[image]
imagesHeld.Unlock()
if !known {
held, err := holds(ctx, image)
if err != nil {
return nil, fmt.Errorf("the artifact store cannot be asked about %s: %w", image, err)
}
if held {
imagesHeld.Lock()
imagesHeld.held[image] = true
imagesHeld.Unlock()
continue
}
} else {
continue
}
running, err := inv.Running(ctx, module)
if err != nil {
return nil, err
}
for _, node := range running {
out = append(out, imageFacts{missingImage{Node: node, Module: module, Image: image}})
}
}
}
return out, nil
}
// modulesOf is the shelf's modules as a map of names, for sortedKeysOf.
func modulesOf(shelf map[string]catalogue.Manifest) map[string]string {
out := make(map[string]string, len(shelf))
for m := range shelf {
out[m] = m
}
return out
}
// watchMissingImages is S25: a module left out of a machine's declaration because the store lacks its image, said at
// once. The condition ends when the store holds the image, or a send names another.
func watchMissingImages(f *signalFacts) []conditions.Observation {
var out []conditions.Observation
for _, m := range f.images {
out = append(out, conditions.Observation{Scope: conditions.ScopeModule, ID: m.Module + "." + m.Node, Kind: kindImageMissing,
Machine: m.Node, Severity: conditions.Warning, Resolver: conditions.ResolverOperator,
Summary: fmt.Sprintf("%s is left out of every send to %s: its declaration names an image that the artifact store does "+
"not hold (the evidence names it); the running container is untouched", m.Module, m.Node),
Said: "image " + m.Image + " is not in the store",
Headline: clipWords(conditions.Capital(m.Module)+" is not updated on "+m.Node, conditions.HeadlineMax),
Explanation: clipWords(fmt.Sprintf("The new version of %s names a container image that the mesh's store does not "+
"hold, so it was not sent to %s. What runs there is untouched, and the rest of %s is sent as usual.",
m.Module, m.Node, m.Node), conditions.ExplanationMax),
Needs: fmt.Sprintf("build %s again; the details name the image.", m.Module),
Resolved: "Resolved: the store holds the image of " + m.Module + " again"})
}
return out
}
// gatherMissingImagesHere is gatherMissingImages asking the mesh's own artifact store; nothing to ask where the mesh has
// none on its network yet.
func gatherMissingImagesHere(ctx context.Context, inv *inventory.Inventory) ([]imageFacts, error) {
shelf, err := inv.Catalogue(ctx)
if err != nil {
return nil, fmt.Errorf("the catalogue cannot be read: %w", err)
}
address, err := artifactStoreAddress(ctx, inv, shelf, "")
if err != nil {
return nil, err
}
if address == "" {
return nil, nil
}
return gatherMissingImages(ctx, inv, artifacts.Store{Address: address}.HoldsManifest)
}
+125
View File
@@ -0,0 +1,125 @@
package main
import (
"context"
"errors"
"strings"
"testing"
"time"
"github.com/novox/mesh-controller/internal/conditions"
)
// **No declaration is sent that names an image the artifact store does not hold** (novox/hq issue 1742).
const storeAt = "novox.internal:5100"
// automx is the image the send of 2026-10-21 17:13 named and the store lacked; the others are held.
const (
automxImage = storeAt + "/mailu/automx@sha256:c62150b8e2cbcc18cb911b098c2877f928761a1e6251667f39ea18251ea35591"
imapImage = storeAt + "/mailu/imap@sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
)
func aMailSend() sendable {
return sendable{
Resources: []map[string]any{
{"id": "mailu.automx", "type": "container", "image": automxImage},
{"id": "mailu.automx-config", "type": "file", "path": "/etc/automx.conf"},
{"id": "mailu.imap", "type": "container", "image": imapImage},
{"id": "other.web", "type": "container", "image": "docker.io/library/nginx@sha256:" + strings.Repeat("b", 64)},
{"id": "dns.dnsmasq", "type": "container", "image": storeAt + "/dnsmasq/main@sha256:" + strings.Repeat("c", 64)},
},
moduleOf: map[string]string{"mailu.automx": "automx", "mailu.automx-config": "automx", "mailu.imap": "mailu",
"other.web": "other", "dns.dnsmasq": "dnsmasq"},
Builds: map[string]string{"automx": "b2f5ebb2new", "mailu": "m2", "other": "o1", "dnsmasq": "d1"},
}
}
func storeHolding(missing ...string) (storeHolds, *[]string) {
var asked []string
return func(_ context.Context, reference string) (bool, error) {
asked = append(asked, reference)
for _, m := range missing {
if strings.Contains(reference, m) {
return false, nil
}
}
return true, nil
}, &asked
}
// 1742/AC1: given a send naming automx's image, which the store does not hold, when the declaration is judged, then
// every resource of automx is out of it, automx is among the modules left out with the reason, its build is the one the
// machine was last sent, and every other module — an image of another registry among them — is as composed.
func TestAModuleWhoseImageTheStoreLacksIsLeftOutOfTheSend(t *testing.T) {
s := aMailSend()
holds, asked := storeHolding("mailu/automx@")
found := holdBackMissingImages(t.Context(), "novox", &s, storeAt, holds, map[string]string{"automx": "b2f5ebb2old"})
if len(found) != 1 || found[0].Module != "automx" || found[0].Node != "novox" || found[0].Image != automxImage {
t.Fatalf("found %+v", found)
}
var ids []string
for _, r := range s.Resources {
ids = append(ids, r["id"].(string))
}
if got := strings.Join(ids, ","); got != "mailu.imap,other.web,dns.dnsmasq" {
t.Errorf("the declaration keeps %s", got)
}
if len(s.LeftOut) != 1 || s.LeftOut[0] != "automx" || !strings.Contains(s.leftOutWhy["automx"], "not in the artifact store") {
t.Errorf("left out %v because %v", s.LeftOut, s.leftOutWhy)
}
if s.Builds["automx"] != "b2f5ebb2old" || s.Builds["mailu"] != "m2" || s.Builds["other"] != "o1" {
t.Errorf("builds %v", s.Builds)
}
for _, a := range *asked {
if strings.Contains(a, "docker.io") {
t.Errorf("the store was asked about an image of another registry: %s", a)
}
}
}
// 1742/AC2: given a module never sent to the machine, when its image is missing, then it has no build in the record of the
// send; given the store does not answer, then nothing is held back for a question unanswered.
func TestAnUnansweredStoreHoldsNothingBackAndANewModuleHasNoBuild(t *testing.T) {
s := aMailSend()
holds, _ := storeHolding("mailu/automx@")
found := holdBackMissingImages(t.Context(), "novox", &s, storeAt, holds, nil)
if len(found) != 1 {
t.Fatalf("%+v", found)
}
if _, has := s.Builds["automx"]; has {
t.Errorf("a module never sent is recorded as sent: %v", s.Builds)
}
s = aMailSend()
silent := func(context.Context, string) (bool, error) { return false, errors.New("registry: connection refused") }
if found := holdBackMissingImages(t.Context(), "novox", &s, storeAt, silent, nil); len(found) != 0 || len(s.LeftOut) != 0 ||
len(s.Resources) != 5 {
t.Errorf("an unanswered store held back %+v, left out %v", found, s.LeftOut)
}
}
// 1742/AC3: a module left out for its image raises a condition for the operator naming the module, the machine and the
// image in the evidence, plain words, and no answer to give; a mesh whose images are all held raises nothing.
func TestAMissingImageIsRaisedForTheOperator(t *testing.T) {
f := calm(time.Date(2026, 10, 21, 17, 40, 0, 0, time.UTC))
if got := watchMissingImages(f); len(got) != 0 {
t.Fatalf("%+v", got)
}
f.images = []imageFacts{{missingImage{Node: "novox", Module: "automx", Image: automxImage}}}
got := watchMissingImages(f)
if len(got) != 1 {
t.Fatalf("%+v", got)
}
o := got[0]
if o.Kind != "image-missing" || o.Key() != "module.automx.novox.image-missing" || o.Resolver != conditions.ResolverOperator ||
o.Severity != conditions.Warning || o.Machine != "novox" {
t.Fatalf("%+v (%s)", o, o.Key())
}
if !strings.Contains(o.Said, "c62150b8") || !strings.Contains(o.Headline+o.Explanation, "automx") {
t.Errorf("words: %q / %q / %q", o.Headline, o.Explanation, o.Said)
}
w := conditions.Words{Headline: o.Headline, Explanation: o.Explanation, Resolved: o.Resolved, Needs: o.Needs}
if why, ok := conditions.PlainWords(w, "novox"); !ok {
t.Errorf("not plain: %s", why)
}
}
+12
View File
@@ -457,6 +457,18 @@ var plainWordings = map[string]func(conditions.Observation) words{
Explanation: "A delivery did not finish. Nothing is tried again until you say so.",
Resolved: "Resolved: the delivery is no longer failed"}
}),
kindMovesHeld: worded(func(o conditions.Observation) words {
return words{Headline: "An update waits for you",
Explanation: "A new version of a module would restart a container or touch its data, so it waits for you.",
Resolved: "Resolved: the update is no longer waiting"}
}),
kindImageMissing: worded(func(o conditions.Observation) words {
return words{Headline: "A module is not updated",
Explanation: "The new version of a module names a container image that the mesh's store does not hold, so it " +
"was not sent. What runs is untouched.",
Needs: "build the module again; the details name the image.",
Resolved: "Resolved: the store holds the image again"}
}),
"ask-lost": worded(func(o conditions.Observation) words {
return words{Headline: "A build has no result",
Explanation: "A build was asked for and never answered. It is asked again with the next delivery.",
+1 -1
View File
@@ -260,7 +260,7 @@ func TestUpdatesHeldNameWhatWaitsAndWhereItIsReleased(t *testing.T) {
plainExample(t, got[0], "Updates of openrazer and sensors wait for your release",
"Needs you: release them from the mesh MCP server; this notification cannot do it. Updates of openrazer and "+
"sensors on g14 and shanks wait for a person to release them, because the last walk failed. "+
"They are not delivered until then, and stay held if you leave them.")
"They are not delivered until then, and stay held if you leave them.", "Release")
}
// **What held them is said in the glossary's words** (2026-10-08 review): the backlog is held after any
-1
View File
@@ -206,7 +206,6 @@ func servingPlanSaved(ctx context.Context, bus link.Bus, open *stores) func(*inv
return func(_ *inventory.Inventory, p inventory.Plan) {
sayPlanMoved(ctx, bus, open.inventory, p)
planKept(p)
keepBuildingKey(ctx, p)
if p.WalkSaid == inventory.WalkEnded {
go forgetChecksOfWalk(context.WithoutCancel(ctx), open.inventory, p)
}
@@ -31,7 +31,8 @@ func TestNoPlanIsWrittenButThroughSavePlansMoving(t *testing.T) {
return err
}
text := string(raw)
if write.MatchString(text) {
// The regular expression is the costly part of reading every source: only a file that names the table can match.
if strings.Contains(strings.ToLower(text), "release_plan") && write.MatchString(text) {
found = append(found, path+" writes the plan table")
}
// A plan record is written by the bucket's raw create and update, which only the plans' own file calls.
-238
View File
@@ -1,238 +0,0 @@
package main
import (
"context"
"encoding/json"
"errors"
"fmt"
"sync"
"sync/atomic"
"time"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"github.com/novox/mesh-controller/internal/broker"
"github.com/novox/mesh-controller/internal/inventory"
"github.com/novox/mesh-controller/internal/link"
)
// The plans building (novox/hq ADR 0483 clause 2, issue 1687).
//
// **A key per delivery plan in its build phase, so that the build seat takes no new check while a delivery builds.** The phase
// runs from the ask of the plan's first build until every build of its last tier is built, or the plan fails, stops or waits
// on the operator. It is derived from the plan alone (buildPhaseOf), so a controller that restarts finds the same answer. The
// key is put when the first build is asked (sayBuilding), kept current at every save of the plan (keepBuildingKey) and by a
// look every link.PlansBuildingRefresh while this controller holds the lease (keepingPlansBuilding), and deleted at the phase's
// end; the bucket's age removes the key of a controller that stopped, so checks are held at most that long.
//
// The layout is documented in internal/link/holds.go.
// buildPhaseOf says whether a plan is in its build phase: open, not waiting for the operator's word, with a build asked and a
// build of its tiers still to be built. A tier built and sent leaves the plan rolling while later tiers are still to be asked;
// it is in the phase still, which is what keeps a check from taking a place between two tiers.
func buildPhaseOf(p inventory.Plan) bool {
if !p.Open() || p.Waiting() || p.Release != nil {
return false
}
asked, unbuilt := false, false
for _, tier := range p.Tiers {
for _, m := range tier {
st := p.Modules[m]
if st != nil && st.State != "" {
asked = true
}
if st == nil || (st.State != "built" && st.State != planDeleted) {
unbuilt = true
}
}
}
return asked && unbuilt
}
// buildingSince is when the plan's first build was asked, or fallback when no module says.
func buildingSince(p inventory.Plan, fallback time.Time) time.Time {
since := time.Time{}
for _, st := range p.Modules {
if st != nil && st.AskedAt != nil && (since.IsZero() || st.AskedAt.Before(since)) {
since = *st.AskedAt
}
}
if since.IsZero() {
return fallback
}
return since
}
// buildingStore is the bucket of the plans building.
type buildingStore interface {
Put(ctx context.Context, plan string, value []byte) error
Delete(ctx context.Context, plan string) error
Keys(ctx context.Context) ([]string, error)
}
// syncBuilding makes the bucket say what the plans say: a key for each plan in its build phase, put again, and none for any
// other plan, deleted. It returns the faults, each once.
func syncBuilding(ctx context.Context, store buildingStore, plans []inventory.Plan, now time.Time) []error {
var faults []error
live := map[string]bool{}
for _, p := range plans {
if !buildPhaseOf(p) {
continue
}
live[p.ID] = true
if err := putBuilding(ctx, store, p.ID, buildingSince(p, now), now); err != nil {
faults = append(faults, err)
}
}
keys, err := store.Keys(ctx)
if err != nil {
return append(faults, err)
}
for _, key := range keys {
if live[key] {
continue
}
if err := store.Delete(ctx, key); err != nil {
faults = append(faults, fmt.Errorf("the key of %s could not be taken away: %w", key, err))
}
}
return faults
}
func putBuilding(ctx context.Context, store buildingStore, plan string, since, now time.Time) error {
body, _ := json.Marshal(link.PlanBuilding{Plan: plan, Since: since.UTC(), Seen: now.UTC()})
if err := store.Put(ctx, plan, body); err != nil {
return fmt.Errorf("the key of %s could not be put: %w", plan, err)
}
return nil
}
// busBuilding is the store on the controller's bucket.
type busBuilding struct {
conn *nats.Conn
mu sync.Mutex
kv jetstream.KeyValue
}
func (b *busBuilding) bucket(ctx context.Context) (jetstream.KeyValue, error) {
b.mu.Lock()
defer b.mu.Unlock()
if b.kv != nil {
return b.kv, nil
}
api, err := jetstream.New(b.conn)
if err != nil {
return nil, err
}
kv, err := api.KeyValue(ctx, broker.PlansBuildingBucket)
if err != nil {
return nil, fmt.Errorf("the plans building: %w", err)
}
b.kv = kv
return kv, nil
}
func (b *busBuilding) Put(ctx context.Context, plan string, value []byte) error {
kv, err := b.bucket(ctx)
if err != nil {
return err
}
_, err = kv.Put(ctx, plan, value)
return err
}
func (b *busBuilding) Delete(ctx context.Context, plan string) error {
kv, err := b.bucket(ctx)
if err != nil {
return err
}
if err := kv.Delete(ctx, plan); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
return err
}
return nil
}
func (b *busBuilding) Keys(ctx context.Context) ([]string, error) {
kv, err := b.bucket(ctx)
if err != nil {
return nil, err
}
keys, err := kv.Keys(ctx)
if errors.Is(err, jetstream.ErrNoKeysFound) {
return nil, nil
}
return keys, err
}
// buildingKeys is the store of the serving controller; nil in a process that is not it, which says nothing.
var buildingKeys atomic.Pointer[buildingStore]
func setBuildingStore(s buildingStore) {
if s == nil {
buildingKeys.Store(nil)
return
}
buildingKeys.Store(&s)
}
// sayBuilding puts a plan's key: its first build is being asked (or another of its builds). A key that could not be put is
// said; the look that follows puts it again.
func sayBuilding(ctx context.Context, plan string, since time.Time) {
s := buildingKeys.Load()
if s == nil {
return
}
if err := putBuilding(ctx, *s, plan, since, time.Now()); err != nil {
fmt.Printf("plans: %v; checks may take a place of %s until the next look\n", err, plan)
}
}
// keepBuildingKey is told each plan this controller keeps: its key put again while it builds and taken away once it does not.
func keepBuildingKey(ctx context.Context, p inventory.Plan) {
s := buildingKeys.Load()
if s == nil {
return
}
if buildPhaseOf(p) {
sayBuilding(ctx, p.ID, buildingSince(p, time.Now()))
return
}
if err := (*s).Delete(ctx, p.ID); err != nil {
fmt.Printf("plans: the key of %s could not be taken away; it ages out: %v\n", p.ID, err)
}
}
// keepingPlansBuilding looks at the plans every link.PlansBuildingRefresh for as long as this controller serves and holds the
// lease, so a key is put again before its age and the key of a plan that left the phase unseen is taken away.
func keepingPlansBuilding(ctx context.Context, inv *inventory.Inventory, conn *nats.Conn) {
store := &busBuilding{conn: conn}
setBuildingStore(store)
defer setBuildingStore(nil)
tick := time.NewTicker(broker.PlansBuildingRefresh)
defer tick.Stop()
failing := ""
for {
if epoch, err := theLease.epoch(ctx); err == nil && epoch != 0 {
why := ""
if plans, err := inv.OpenPlans(ctx); err != nil {
why = fmt.Sprintf("the plans cannot be read: %v", err)
} else if faults := syncBuilding(ctx, store, plans, time.Now()); len(faults) > 0 {
why = fmt.Sprint(faults)
}
if why != failing {
if why != "" {
fmt.Printf("plans: the keys of the plans building cannot be kept: %s\n", why)
} else {
fmt.Printf("plans: the keys of the plans building are kept again\n")
}
failing = why
}
}
select {
case <-ctx.Done():
return
case <-tick.C:
}
}
}
-175
View File
@@ -1,175 +0,0 @@
package main
import (
"context"
"encoding/json"
"sort"
"sync"
"testing"
"time"
"github.com/novox/mesh-controller/internal/inventory"
"github.com/novox/mesh-controller/internal/link"
)
// Issue 1687 (ADR 0483 clause 2): a delivery plan's key lives from the ask of its first build until every build of its last
// tier is built, or the plan fails, stops or waits on the operator.
type fakeBuildingStore struct {
mu sync.Mutex
keys map[string]link.PlanBuilding
puts map[string]int
}
func newFakeBuildingStore() *fakeBuildingStore {
return &fakeBuildingStore{keys: map[string]link.PlanBuilding{}, puts: map[string]int{}}
}
func (f *fakeBuildingStore) Put(_ context.Context, plan string, value []byte) error {
f.mu.Lock()
defer f.mu.Unlock()
var v link.PlanBuilding
if err := json.Unmarshal(value, &v); err != nil {
return err
}
f.keys[plan] = v
f.puts[plan]++
return nil
}
func (f *fakeBuildingStore) Delete(_ context.Context, plan string) error {
f.mu.Lock()
defer f.mu.Unlock()
delete(f.keys, plan)
return nil
}
func (f *fakeBuildingStore) Keys(context.Context) ([]string, error) {
f.mu.Lock()
defer f.mu.Unlock()
var out []string
for k := range f.keys {
out = append(out, k)
}
sort.Strings(out)
return out, nil
}
func aPlan(id, state string, tiers [][]string, states map[string]string) inventory.Plan {
p := inventory.Plan{ID: id, State: state, Tiers: tiers, Modules: map[string]*inventory.PlanModule{}}
now := time.Now().Add(-time.Minute)
for m, st := range states {
pm := &inventory.PlanModule{State: st}
if st != "" {
pm.AskedAt = &now
}
p.Modules[m] = pm
}
return p
}
func TestAPlanIsInItsBuildPhaseFromTheFirstAskToTheLastBuild(t *testing.T) {
tiers := [][]string{{"a", "b"}, {"c"}}
cases := []struct {
name string
plan inventory.Plan
want bool
}{
{"no build asked yet", aPlan("p", inventory.PlanBuilding, tiers, nil), false},
{"the first tier asked", aPlan("p", inventory.PlanBuilding, tiers, map[string]string{"a": "asked", "b": "asked"}), true},
// Between two tiers: the first built and sent (rolling), the second not asked yet.
{"between two tiers", aPlan("p", inventory.PlanRolling, tiers, map[string]string{"a": "built", "b": "built"}), true},
{"the last tier asked", aPlan("p", inventory.PlanBuilding, tiers, map[string]string{"a": "built", "b": "built", "c": "asked"}), true},
{"every build built", aPlan("p", inventory.PlanRolling, tiers, map[string]string{"a": "built", "b": "built", "c": "built"}), false},
// A module the plan deletes is done as far as its tiers go (release_plan.go), so it holds no key through the rollout.
{"a deleted module is done", aPlan("p", inventory.PlanRolling, tiers, map[string]string{"a": "built", "b": "built", "c": planDeleted}), false},
{"failed", aPlan("p", inventory.PlanFailed, tiers, map[string]string{"a": "failed", "b": "asked"}), false},
{"superseded", aPlan("p", inventory.PlanSuperseded, tiers, map[string]string{"a": "asked"}), false},
{"a batch whose window is open", aPlan("p", inventory.PlanAssembling, tiers, map[string]string{"a": "asked"}), false},
}
for _, c := range cases {
if got := buildPhaseOf(c.plan); got != c.want {
t.Errorf("%s: in the build phase = %v, want %v", c.name, got, c.want)
}
}
waiting := aPlan("p", inventory.PlanBuilding, tiers, map[string]string{"a": "asked"})
waiting.Delivery = &inventory.PlanDelivery{Awaits: "the operator's word"}
if buildPhaseOf(waiting) {
t.Error("a plan that waits on the operator is in its build phase")
}
}
// The key's life: put, refreshed, deleted at the phase's end; a plan that left the phase unseen loses its key at the next look.
func TestAPlanKeyIsPutRefreshedAndDeletedAtThePhasesEnd(t *testing.T) {
ctx := context.Background()
store := newFakeBuildingStore()
tiers := [][]string{{"a"}, {"b"}}
building := aPlan("plan-1", inventory.PlanBuilding, tiers, map[string]string{"a": "asked"})
if faults := syncBuilding(ctx, store, []inventory.Plan{building}, time.Now()); len(faults) > 0 {
t.Fatal(faults)
}
if _, ok := store.keys["plan-1"]; !ok {
t.Fatal("no key while the first build is asked")
}
first := store.keys["plan-1"]
// Refreshed: put again, the phase's start kept.
time.Sleep(2 * time.Millisecond)
syncBuilding(ctx, store, []inventory.Plan{building}, time.Now())
if store.puts["plan-1"] != 2 || !store.keys["plan-1"].Seen.After(first.Seen) || !store.keys["plan-1"].Since.Equal(first.Since) {
t.Errorf("the key was not refreshed with its start kept: %d puts, %+v then %+v", store.puts["plan-1"], first, store.keys["plan-1"])
}
// Between the tiers: still there.
between := aPlan("plan-1", inventory.PlanRolling, tiers, map[string]string{"a": "built"})
syncBuilding(ctx, store, []inventory.Plan{between}, time.Now())
if _, ok := store.keys["plan-1"]; !ok {
t.Error("the key was gone between two tiers")
}
// A key of another plan that is no longer in the phase.
store.keys["plan-0"] = link.PlanBuilding{Plan: "plan-0"}
done := aPlan("plan-1", inventory.PlanRolling, tiers, map[string]string{"a": "built", "b": "built"})
syncBuilding(ctx, store, []inventory.Plan{done}, time.Now())
if len(store.keys) != 0 {
t.Errorf("keys remain after the last build was built: %v", store.keys)
}
}
// keepBuildingKey, told of each save, puts the key while the plan builds and deletes it when the plan fails.
func TestASavedPlanPutsOrDeletesItsKey(t *testing.T) {
ctx := context.Background()
store := newFakeBuildingStore()
var s buildingStore = store
buildingKeys.Store(&s)
defer buildingKeys.Store(nil)
p := aPlan("plan-7", inventory.PlanBuilding, [][]string{{"a"}}, map[string]string{"a": "asked"})
keepBuildingKey(ctx, p)
if _, ok := store.keys["plan-7"]; !ok {
t.Fatal("the key was not put when the plan was saved building")
}
p.State = inventory.PlanFailed
keepBuildingKey(ctx, p)
if _, ok := store.keys["plan-7"]; ok {
t.Fatal("the key stayed after the plan failed")
}
}
// The plan's key is there before its first build is asked, so a check cannot take a place between the ask and the plan's save.
func TestThePlansKeyIsPutBeforeItsFirstBuildIsAsked(t *testing.T) {
store := newFakeBuildingStore()
var s buildingStore = store
buildingKeys.Store(&s)
defer buildingKeys.Store(nil)
p, byName := aTierOfThree()
keyThere := false
was := askABuild
askABuild = func(context.Context, buildSource, string, string) (string, error) {
_, keyThere = store.keys["plan-1"]
return "build-x", nil
}
t.Cleanup(func() { askABuild = was })
askModule(t.Context(), p, "a", byName)
if !keyThere {
t.Fatal("the plan's key was not there when its first build was asked")
}
}
+1 -1
View File
@@ -153,7 +153,7 @@ func planWaitCommand(ctx context.Context, inv *inventory.Inventory, id, untilAsk
line := planLineSaid(p, now, pause, bounds.of(p.Repository), func(module string) []string {
on, _ := inv.Running(ctx, module)
return on
}, waitsOf(ctx, inv, one))
}, waitsOf(ctx, inv, one)) + heldSaid(p, heldNow(ctx, inv))
doc := planWaitedDoc(p, planStatuses(one, now, pause, bounds)[0], until, came, now.Sub(began), line)
return json.NewEncoder(os.Stdout).Encode(doc)
}
-2
View File
@@ -284,8 +284,6 @@ func serve(ctx context.Context) (err error) {
go mirroringRecords(ctx, inv, bus.Conn, bus)
// And the bucket of the files that travel by digest, judged against what each machine's latest declaration names (ADR 0486 §3).
go sweepingDeclarationFiles(ctx, inv, bus.Conn)
// And the key of each plan in its build phase, which holds the build seat's checks while a delivery builds (ADR 0483 clause 2).
go keepingPlansBuilding(ctx, inv, bus.Conn)
// Every call carries the lease's epoch, and its record is written only under the lease (novox/hq
// to-be 45 §6).
link.Calls.UnderLease(func() (uint64, error) { return theLease.epoch(ctx) })
+6 -40
View File
@@ -91,9 +91,7 @@ type queueAnswer struct {
Dead int `json:"dead"`
MaxDeliver int `json:"max-deliver"`
Asks []queuedAnswer `json:"asks"`
// HeldFor are the delivery plans building, while which the waiting checks are held (ADR 0483 clause 5); empty when none.
HeldFor []string `json:"held-for,omitempty"`
Text string `json:"text"`
Text string `json:"text"`
}
// queuedAnswer is one ask as data. Never what the request carries beyond its source — not `held`, every
@@ -122,9 +120,6 @@ type queuedAnswer struct {
// Delivery is the how-manyth time an ask in flight was handed out; more than 1 is a restart
// (novox/hq issue 594).
Delivery int `json:"delivery,omitempty"`
// LetThrough is a check that waited past 30 minutes and is let through while a delivery builds: the next one, or the one
// that runs (ADR 0483 clause 5).
LetThrough bool `json:"let-through,omitempty"`
}
// aCommitRef is a ref that names a commit by its full hash (SHA-1 or SHA-256).
@@ -135,8 +130,7 @@ var aCommitRef = regexp.MustCompile(`^(?:[0-9a-f]{40}|[0-9a-f]{64})$`)
func queueAnswerOf(q link.Queue, now time.Time) queueAnswer {
waiting, running, dead := q.Of(link.AskWaiting), q.Of(link.AskInFlight), q.Of(link.AskDead)
out := queueAnswer{Seat: q.Seat, Waiting: len(waiting), InFlight: len(running), Dead: len(dead),
MaxDeliver: q.MaxDeliver, Asks: []queuedAnswer{}, HeldFor: q.Hold.Plans, Text: queueText(q, now)}
letNext, letRunning := q.LetThroughOf(now)
MaxDeliver: q.MaxDeliver, Asks: []queuedAnswer{}, Text: queueText(q, now)}
one := func(a link.QueuedAsk, position int) queuedAnswer {
kind := link.KindOf(a.Request)
since := a.AskedAt
@@ -149,7 +143,7 @@ func queueAnswerOf(q link.Queue, now time.Time) queueAnswer {
}
return queuedAnswer{ID: a.ID, State: a.State, Kind: kind, Repository: link.WithoutPassword(a.Repository), Path: a.Path,
Ref: a.Ref, Commit: commit, Machine: a.On, WasOn: a.Was, AskedAt: a.AskedAt, Since: since,
Position: position, Seq: a.Seq, Delivery: a.Delivery, LetThrough: a.ID != "" && (a.ID == letNext || a.ID == letRunning)}
Position: position, Seq: a.Seq, Delivery: a.Delivery}
}
for _, a := range running {
out.Asks = append(out.Asks, one(a, 0))
@@ -185,11 +179,6 @@ func queueText(q link.Queue, now time.Time) string {
}
return s
}
letNext, letRunning := q.LetThroughOf(now)
const passed = "waited past 30 minutes, let through"
if q.HoldFault != "" {
fmt.Fprintf(&b, "why checks wait cannot be read, so none is said held: %s\n", q.HoldFault)
}
if len(running) > 0 {
fmt.Fprintln(&b, "\nin flight:")
for _, a := range running {
@@ -204,27 +193,14 @@ func queueText(q link.Queue, now time.Time) string {
case a.Was != "":
where = fmt.Sprintf("handed back by %s, to be handed out again", a.Was)
}
note := ""
if a.ID != "" && a.ID == letRunning {
note = " — " + passed
}
fmt.Fprintf(&b, " %-26s %-8s %s — %s, %s (seq %d)%s\n", a.ID, link.KindOf(a.Request), what(a), where,
ago(a.AskedAt), a.Seq, note)
fmt.Fprintf(&b, " %-26s %-8s %s — %s, %s (seq %d)\n", a.ID, link.KindOf(a.Request), what(a), where,
ago(a.AskedAt), a.Seq)
}
}
if len(waiting) > 0 {
fmt.Fprintln(&b, "\nwaiting, in the order they are handed out:")
// **Why the checks wait** (ADR 0483 clause 5): above them, the plans they are held for.
if len(q.Hold.Plans) > 0 && hasCheck(waiting) {
fmt.Fprintf(&b, " checks are held while delivery %s builds; the oldest check that waited 30 minutes goes first, one at a time\n",
strings.Join(q.Hold.Plans, ", "))
}
for _, a := range waiting {
note := ""
if a.ID != "" && a.ID == letNext {
note = " — " + passed + " next"
}
fmt.Fprintf(&b, " %-26s %-8s %s — %s (seq %d)%s\n", a.ID, link.KindOf(a.Request), what(a), ago(a.AskedAt), a.Seq, note)
fmt.Fprintf(&b, " %-26s %-8s %s — %s (seq %d)\n", a.ID, link.KindOf(a.Request), what(a), ago(a.AskedAt), a.Seq)
}
}
if len(dead) > 0 {
@@ -242,16 +218,6 @@ func queueText(q link.Queue, now time.Time) string {
return b.String()
}
// hasCheck says a check is among the asks.
func hasCheck(asks []link.QueuedAsk) bool {
for _, a := range asks {
if a.Request.Check != nil {
return true
}
}
return false
}
// cancelCommand drops one waiting or dead ask.
func cancelCommand(ctx context.Context, args []string) error {
if len(args) != 1 {
-77
View File
@@ -1,77 +0,0 @@
package main
import (
"strings"
"testing"
"time"
"github.com/novox/mesh-controller/internal/link"
)
// Issue 1688 / 1652 AC5 (ADR 0483 clause 5): the queue names, above the waiting checks, the plan they are held for, and says of a
// check let through that it waited past 30 minutes.
func aHeldQueue(now time.Time) link.Queue {
check := func(id string, ago time.Duration, n int) link.QueuedAsk {
return link.QueuedAsk{Seq: uint64(n), ID: id, Repository: "ssh://forge/novox/r", AskedAt: now.Add(-ago), State: link.AskWaiting,
Request: link.BuildRequest{ID: id, Check: &link.CheckRequest{Owner: "novox", Repo: "r", Number: n}}}
}
return link.Queue{Seat: link.TheBuildMachine, Asks: []link.QueuedAsk{
check("build-young", 5*time.Minute, 1),
check("build-old", 45*time.Minute, 2),
check("build-older", 50*time.Minute, 3),
}, Hold: link.CheckHold{Plans: []string{"plan-42"}}}
}
func TestTheQueueNamesThePlanTheChecksAreHeldFor(t *testing.T) {
now := time.Date(2026, 10, 21, 12, 0, 0, 0, time.UTC)
text := queueText(aHeldQueue(now), now)
heldAt := strings.Index(text, "held while delivery plan-42 builds")
firstAt := strings.Index(text, "build-young")
if heldAt < 0 || firstAt < 0 || heldAt > firstAt {
t.Fatalf("the plan the checks are held for is not named above the waiting checks:\n%s", text)
}
// Nothing building: nothing held.
q := aHeldQueue(now)
q.Hold = link.CheckHold{}
if strings.Contains(queueText(q, now), "held while") {
t.Fatal("checks are said held with no delivery building")
}
}
func TestACheckLetThroughIsSaidToHaveWaitedPastThirtyMinutes(t *testing.T) {
now := time.Date(2026, 10, 21, 12, 0, 0, 0, time.UTC)
q := aHeldQueue(now)
text := queueText(q, now)
var saidOf []string
for _, l := range strings.Split(text, "\n") {
if strings.Contains(l, "waited past 30 minutes, let through") {
saidOf = append(saidOf, strings.Fields(l)[0])
}
}
if len(saidOf) != 1 || saidOf[0] != "build-older" {
t.Fatalf("the check let through next (the oldest) is not the one said: %v\n%s", saidOf, text)
}
answer := queueAnswerOf(q, now)
if len(answer.HeldFor) != 1 || answer.HeldFor[0] != "plan-42" {
t.Errorf("held-for = %v", answer.HeldFor)
}
for _, a := range answer.Asks {
if a.LetThrough != (a.ID == "build-older") {
t.Errorf("%s let-through = %v", a.ID, a.LetThrough)
}
}
// A check in flight under the let-through key is said too.
q.Asks = append(q.Asks, link.QueuedAsk{Seq: 9, ID: "build-running", State: link.AskInFlight, On: "ace", AskedAt: now.Add(-time.Hour),
Started: now.Add(-time.Minute), Request: link.BuildRequest{ID: "build-running", Check: &link.CheckRequest{Owner: "novox", Repo: "r", Number: 9}}})
q.Hold.LetThrough = &link.LetThrough{Builder: "ace"}
if !strings.Contains(queueText(q, now), "build-running") || !strings.Contains(queueText(q, now), "waited past 30 minutes, let through\n") {
t.Errorf("the check in flight let through is not said:\n%s", queueText(q, now))
}
// Younger than the bound: none is said.
young := aHeldQueue(now)
young.Asks = young.Asks[:1]
if strings.Contains(queueText(young, now), "waited past") {
t.Error("a check of 5 minutes is said to have waited past 30")
}
}
+28 -6
View File
@@ -181,15 +181,24 @@ func planModuleNames(p inventory.Plan) map[string]string {
return out
}
// recordedWaiting is, on one machine, every move of a recorded module that waits for a person, each said with why
// (ADR 0377 §6): what `status` names the machine behind for, with `push <node>` as the remedy. A move under a
// module's own `record` that recreates nothing and moves no data is not among them: it goes with its walk.
func recordedWaiting(ctx context.Context, inv *inventory.Inventory, f moveFacts, node string) ([]string, error) {
// heldMove is one move of a recorded module that waits for a person on one machine (ADR 0377 §6).
type heldMove struct {
Module string
From, To string
// Why is the walk's own word for it: "waits: " and the recreate words, the data change, or the hold.
Why string
// Chosen is whether a person's own choice holds it — a module they set to record, or the bus's step — which the
// operator is not asked about again: they chose that hold, and the bus has its own condition (S17).
Chosen bool
}
// heldMovesOn is, on one machine, every move of a recorded module that waits for a person, in module order.
func heldMovesOn(ctx context.Context, inv *inventory.Inventory, f moveFacts, node string) ([]heldMove, error) {
sent, known, err := inv.SentBuilds(ctx, node)
if err != nil || !known {
return nil, err
}
var out []string
var out []heldMove
for _, m := range sortedKeysOf(sent) {
was := sent[m]
now, held := f.current[m]
@@ -197,12 +206,25 @@ func recordedWaiting(ctx context.Context, inv *inventory.Inventory, f moveFacts,
continue
}
if v := f.recordedMove(m, was); !v.rolls {
out = append(out, fmt.Sprintf("%s %s → %s %s", m, short(was), short(now.Commit), v.said))
out = append(out, heldMove{Module: m, From: was, To: now.Commit, Why: v.said,
Chosen: now.From == catalogue.FromPerson || now.From == catalogue.FromBus})
}
}
return out, nil
}
// recordedWaiting is, on one machine, every move of a recorded module that waits for a person, each said with why
// (ADR 0377 §6): what `status` names the machine behind for, with `push <node>` as the remedy. A move under a
// module's own `record` that recreates nothing and moves no data is not among them: it goes with its walk.
func recordedWaiting(ctx context.Context, inv *inventory.Inventory, f moveFacts, node string) ([]string, error) {
moves, err := heldMovesOn(ctx, inv, f, node)
var out []string
for _, h := range moves {
out = append(out, fmt.Sprintf("%s %s → %s %s", h.Module, short(h.From), short(h.To), h.Why))
}
return out, err
}
// stringsKeys is a map of lists as a map of its keys, for sortedKeysOf.
func stringsKeys(m map[string][]string) map[string]string {
out := map[string]string{}
+4
View File
@@ -1371,6 +1371,10 @@ func backlogObservation() []conditions.Observation {
"them, `upgrade release-backlog --why …` releases them", n, strings.Join(machines, ", "), backlogNow.held)}
w := releaseHeldWords(modules, machines)
o.Headline, o.Explanation, o.Resolved, o.Needs = w.Headline, w.Explanation, w.Resolved, w.Needs
// Asked of the operator, who releases them with an Approve (novox/hq issue 1732): the controller's own
// `upgrade release-backlog`, which opens the release plan that walks them through the machines, each judged.
o.Actions = []conditions.Action{{Label: "Release", Verb: "mesh-controller.upgrade", Level: conditions.LevelApprove,
Arguments: map[string]string{"release-backlog": "true", "why": "", "cause": conditions.CauseOperatorAnswer}}}
return []conditions.Observation{o}
}
+9 -5
View File
@@ -577,9 +577,6 @@ func askModule(ctx context.Context, p *inventory.Plan, name string, byName map[s
}
// **The walk takes the check's build of the same key** (novox/hq ADR 0350 §2): its ask says so, with the pull
// requests it answers, whose checks name the part of a differing key.
// **The plan's key is there before its first build is** (ADR 0483 clause 2): a check taken between the ask and the plan's save
// would hold a place for its whole run.
sayBuilding(ctx, p.ID, now)
id, err := askABuild(withReuse(ctx, reuseWithVerdicts(ctx, p.ID, reuseAskOf(p), mergedHeadsOf(p))), source,
e.Source.Path, ref)
if err != nil {
@@ -2064,6 +2061,12 @@ func planLineSaid(p inventory.Plan, now time.Time, pause pauseView, bound time.D
// refused ("a build waits there for its gate: <machine> (...)") is kept there. "building" with LATE
// hid that reason from `plans`, `status` and the stalled condition (novox/hq issue 920).
what = p.Note
// And when the builds are held after a failed release, they wait for a person, not for a build, and the
// operator is asked to release them (novox/hq issue 1732): said so, and not late.
if strings.Contains(p.Note, errUngated.Error()) && backlogNow.held != "" {
what += " — held after a failed delivery until a person releases it (the operator is asked)"
late = ""
}
return fmt.Sprintf("%s · %s, held: %s, for %s%s", name, where, what, since, late)
}
return fmt.Sprintf("%s · %s, %s for %s%s", name, where, what, since, late)
@@ -2271,7 +2274,7 @@ func plansCommand(ctx context.Context, args []string) error {
bounds.of(p.Repository), func(module string) []string {
on, _ := inv.Running(ctx, module)
return on
}, waitsOf(ctx, inv, []inventory.Plan{p})))
}, waitsOf(ctx, inv, []inventory.Plan{p}))+heldSaid(p, heldNow(ctx, inv)))
if r := p.Release; r != nil {
// A release plan's walk (ADR 0236): machines done, the one judged, those to come.
fmt.Printf(" machines in order: %s; done: %s; skipped: %s\n", strings.Join(r.Order, ", "),
@@ -2466,11 +2469,12 @@ func plansCommand(ctx context.Context, args []string) error {
return on
}
waits := waitsOf(ctx, inv, plans)
held := heldNow(ctx, inv)
for _, p := range plans {
if p.Batch() {
continue
}
fmt.Printf("%-28s %s\n", p.ID, planLineSaid(p, now, pause, bounds.of(p.Repository), running, waits))
fmt.Printf("%-28s %s%s\n", p.ID, planLineSaid(p, now, pause, bounds.of(p.Repository), running, waits), heldSaid(p, held))
}
return nil
}
+1 -1
View File
@@ -152,7 +152,7 @@ func answersRightly(q dnsmessage.Question) *dnsmessage.Message {
func quickResolvers(t *testing.T, port string) {
t.Helper()
before, within, pause := resolverPort, resolverWithin, resolverPause
resolverPort, resolverWithin, resolverPause = port, 150*time.Millisecond, 10*time.Millisecond
resolverPort, resolverWithin, resolverPause = port, 60*time.Millisecond, 5*time.Millisecond
t.Cleanup(func() { resolverPort, resolverWithin, resolverPause = before, within, pause })
}
+16
View File
@@ -244,6 +244,22 @@ var signalsTable = []signalRow{
newest: func(f *signalFacts) time.Time {
return newestOf(f.failed, func(p failedPlanFacts) time.Time { return p.since })
}},
{Row: "S24", Signal: "a move of a recorded module that waits for a person's push is asked of the operator",
Emitter: "controller's plan", Trigger: "each machine with a move of a recorded module held for a person (novox/hq ADR 0377, issue 1732)",
Bound: "none: raised on the first look that finds the machine holding such a move, naming the modules and the " +
"machine, with Release (the machine's push) as the answer; ended when the machine no longer waits, whoever " +
"pushed it. A hold a person chose (a module set to record with a why, the bus) is not asked.",
Kind: kindMovesHeld, Severity: conditions.Warning, Phase: 3,
needs: func(f *signalFacts) error { return f.heldErr }, watch: watchMovesHeld,
newest: func(f *signalFacts) time.Time { return time.Time{} }},
{Row: "S25", Signal: "a module left out of a send because the artifact store lacks an image it names",
Emitter: "controller's send", Trigger: "each send whose declaration names an image of the store that the store does not hold (novox/hq issue 1742)",
Bound: "none: raised on the first look at a registered build that names such an image, naming the module, each " +
"machine that runs it and the image; every send leaves that module out; the store is asked again at each " +
"look and the condition ends when it holds the image",
Kind: kindImageMissing, Severity: conditions.Warning, Phase: 3,
needs: func(f *signalFacts) error { return f.imagesErr }, watch: watchMissingImages,
newest: func(f *signalFacts) time.Time { return time.Time{} }},
{Row: "S20", Signal: "a send refused for the assignment generation it was composed from",
Emitter: "node-engine", Trigger: "each refusal (novox/hq issue 234)",
Bound: "none: raised at the first refusal, naming its sender, the generation it came from and the one the " +
+13
View File
@@ -171,6 +171,19 @@ var suppressions = map[string]suppression{
since: f.now.Add(-time.Second)}}
},
},
// A machine holding a move of a recorded module for a person is asked at its first look (novox/hq issue 1732).
// Inside: nothing waits there.
"S24": {
inside: func(f *signalFacts) { f.held = nil },
past: func(f *signalFacts) { f.held = []heldFacts{aHeldMailu()} },
},
// A registered build naming an image the artifact store lacks is raised at the first look (novox/hq issue 1742).
"S25": {
inside: func(f *signalFacts) { f.images = nil },
past: func(f *signalFacts) {
f.images = []imageFacts{{missingImage{Node: "anchor", Module: "automx", Image: automxImage}}}
},
},
// A batch queued behind an open walk past that walk's bound, named.
"S19": {
inside: func(f *signalFacts) {
+8
View File
@@ -58,6 +58,12 @@ type signalFacts struct {
// failed are the delivery plans that failed and are asked of the operator (S23, novox/hq ADR 0449).
failed []failedPlanFacts
failedErr error
// held are the machines with moves of recorded modules waiting for a person's push (S24, novox/hq issue 1732).
held []heldFacts
heldErr error
// images are the modules left out of a send because the artifact store lacks an image they name (S25, issue 1742).
images []imageFacts
imagesErr error
// refusedSends are the sends a machine refused within the hour for the generation they were composed
// from, each naming its sender (S20, novox/hq issue 234).
refusedSends []inventory.Send
@@ -357,6 +363,8 @@ func (w *watchdogs) gather(ctx context.Context) *signalFacts {
f.batches, f.plansErr = gatherBatches(ctx, inv, now)
}
f.failed, f.failedErr = gatherFailedPlans(ctx, inv, now)
f.held, f.heldErr = gatherHeldMoves(ctx, inv)
f.images, f.imagesErr = gatherMissingImagesHere(ctx, inv)
f.loop, f.loopErr = w.gatherLoop()
f.asks, f.asksErr = w.gatherAsks(ctx, inv)
if open, err := w.keeper.Open(ctx); err != nil {
+1 -1
View File
@@ -3,7 +3,7 @@ module github.com/novox/mesh-controller
go 1.26.0
require (
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021153908-725590e06e75
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021173436-07a65927c362
github.com/jackc/pgx/v5 v5.10.0
github.com/nats-io/nats-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0
+2 -2
View File
@@ -1,7 +1,7 @@
git.novox.be/novox/mesh-host v0.0.0-20261018171808-9d9e3a46c4f5 h1:E5stfIeCN6uYgA9TCv+4K8JgaURg3zQppvUgJLaDMP4=
git.novox.be/novox/mesh-host v0.0.0-20261018171808-9d9e3a46c4f5/go.mod h1:4wyoCr6IIIIarEUuROThWDj7DAvJvIidqMjWcpSnOwE=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021153908-725590e06e75 h1:AcYfyJpYaUufhI93VG76+b001msXY4XX7kuRjnf33HY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021153908-725590e06e75/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021173436-07a65927c362 h1:Ud/YHmhlm0qdB45LyDvxq7l4x6Bg2Ye0siHK5JFE+kk=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021173436-07a65927c362/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-9
View File
@@ -46,7 +46,6 @@ func CheckStreams() []Stream {
Retention: RetentionWorkQueue,
MaxAge: 7 * 24 * 60 * 60,
MaxMsgsPerSubject: 1,
Direct: true,
Why: "the checks of pull requests marked ready, taken after a delivery build and before any other check " +
"(novox/hq ADR 0406 §3); one subject a pull request, the newest request kept, so a newer one replaces a " +
"request still waiting",
@@ -57,7 +56,6 @@ func CheckStreams() []Stream {
Retention: RetentionWorkQueue,
MaxAge: 7 * 24 * 60 * 60,
MaxMsgsPerSubject: 1,
Direct: true,
Why: "every other check, taken last and in the order asked (novox/hq ADR 0406 §3); one subject a pull " +
"request, the newest request kept; one left behind for a head already judged is dropped when taken",
},
@@ -100,13 +98,6 @@ func holderCheckGrants() []string {
"$KV."+CheckRunningBucket+".>",
"$JS.API.STREAM.INFO.KV_"+CheckRunningBucket,
"$JS.API.DIRECT.GET.KV_"+CheckRunningBucket+".>")
// And, to hold checks while a delivery builds (ADR 0483, issues 1687 and 1688): the bucket of the plans building, **read and
// never written** (the controller's alone, so no `$KV.` subject of it is granted), and the oldest waiting ask of each check
// queue, read by a direct get that takes nothing from the queue and so counts against no delivery of the ask.
pub = append(pub, readGrants(PlansBuildingBucket, "")...)
for _, w := range CheckWorkers() {
pub = append(pub, "$JS.API.DIRECT.GET."+w.Stream, "$JS.API.DIRECT.GET."+w.Stream+".>")
}
return pub
}
-34
View File
@@ -40,37 +40,3 @@ func TestTheBuildAgentReadsAndWatchesThePullRequestsKeys(t *testing.T) {
}
}
}
// Issue 1687 (ADR 0483 clause 2): a holder of the build seat reads the bucket of the plans building and writes none of it, and
// reads the oldest waiting ask of each check queue by a direct get, which takes nothing from the queue.
func TestTheBuildSeatsHoldersReadThePlansBuildingAndWriteNone(t *testing.T) {
seat := Seat{Name: TheBuildSeat, Scope: "node", Accepts: []string{"build"}, Emits: []string{"built", "started"}}
p := Principal{Kind: KindModule, Node: "one", Module: "build-agent", Holds: []Seat{seat}, PasswordHash: "x"}
perms, err := PermissionsFor(p)
if err != nil {
t.Fatal(err)
}
stream := "KV_" + PlansBuildingBucket
for _, s := range []string{
"$JS.API.STREAM.INFO." + stream,
"$JS.API.DIRECT.GET." + stream + ".>",
"$JS.API.CONSUMER.CREATE." + stream + ".>",
"$JS.API.DIRECT.GET.CHECKS_READY",
"$JS.API.DIRECT.GET.CHECKS_OTHER",
} {
has(t, perms.Publish, s)
}
for _, s := range perms.Publish {
if SubjectsOverlap(s, "$KV."+PlansBuildingBucket+".plan-1") {
t.Errorf("a holder of the build seat may write %s through %q", PlansBuildingBucket, s)
}
}
// And the controller alone writes it, as the writers table says.
for _, row := range WritersTable {
for _, s := range row.Subjects {
if s == "$KV."+PlansBuildingBucket+".>" && row.Writes(p, "$KV."+PlansBuildingBucket+".plan-1") {
t.Errorf("%q lets the build seat's holder write the plans building", row.State)
}
}
}
}
+2 -38
View File
@@ -63,26 +63,6 @@ var (
// repository: the controller reads the ask's warrant from the router and accepts only the operator's Approve of
// those heads. Written by the controller alone, from the announcement, when it holds the lease.
DeliveryMergesBucket = BucketName(ControllerSeat, "delivery-merges")
// PlansBuildingBucket holds one key per delivery plan in its build phase (novox/hq ADR 0483 clause 2, issue 1687): the key
// is `<plan id>` (`plan-<digits>`), its value the plan, the time its build phase began and the time of the last refresh.
// The controller (lease holder) puts it when the plan's first build is asked, puts it again about every thirty seconds
// while the plan builds, and deletes it when the phase ends (every build of the last tier built, or the plan fails, stops
// or waits on the operator); the bucket's age, PlansBuildingKeptFor, is how long a key outlives a controller that
// stopped. **Read by the build seat's holders and the checker, written by the controller alone**: a holder takes no new
// check while any key is there, but for one that waited past ADR 0483's bound. A reader lists the keys and reads none of
// their values to decide that (`link.BuildingPlans`).
PlansBuildingBucket = BucketName(ControllerSeat, PlansBuilding)
)
// PlansBuilding is the local name of the bucket of the plans building, as a reader names it in `reads`
// (`mesh-controller.plans-building`).
const PlansBuilding = "plans-building"
// PlansBuildingKeptFor is the age of a plan-building key: the controller puts it again at PlansBuildingRefresh, so a
// controller that stopped leaves checks held at most this long (ADR 0483 clause 2).
const (
PlansBuildingKeptFor = 2 * time.Minute
PlansBuildingRefresh = 30 * time.Second
)
// DeliveryMergesKeptFor is how long a merge mesh-delivery announced is kept: a week, as the events are.
@@ -224,14 +204,14 @@ func IsControllerBucket(bucket string) bool {
bucket == AssignmentsBucket || bucket == DeclaredConfigBucket || bucket == PredefinedBucket ||
bucket == MergeWindowBucket || bucket == RolloutSilentBucket || bucket == CheckRunningBucket ||
bucket == DeliveryMergesBucket || bucket == BuildsBucket || bucket == RolloutsBucket ||
bucket == PlansBuildingBucket || bucket == delivery.HaveBucket
bucket == delivery.HaveBucket
}
// ControllerBuckets are the controller's own buckets, in the order they are asserted.
func ControllerBuckets() []string {
return []string{LeaseBucket, CallsBucket, HandActsBucket, ConditionsBucket, ConditionHistoryBucket, AskedBucket,
AssignmentsBucket, DeclaredConfigBucket, PredefinedBucket, MergeWindowBucket, RolloutSilentBucket, CheckRunningBucket,
DeliveryMergesBucket, PlansBuildingBucket, BuildsBucket, RolloutsBucket, delivery.HaveBucket}
DeliveryMergesBucket, BuildsBucket, RolloutsBucket, delivery.HaveBucket}
}
// ControllerBucketsAsserter is what raising the controller's buckets needs of a connection.
@@ -374,22 +354,6 @@ func (j *JetStream) EnsureControllerBuckets() error {
}); err != nil {
return fmt.Errorf("asserting bucket %s: %w", CheckRunningBucket, err)
}
// **The plans building: keys that age** (ADR 0483 clause 2, issue 1687). One message a key, each refresh a new one, so the
// age of the bucket is how long a key outlives a controller that stopped refreshing it.
if _, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{
Bucket: PlansBuildingBucket,
Description: "the delivery plans in their build phase (novox/hq ADR 0483): `<plan id>` holds the plan and when its " +
"build phase began; put when the plan's first build is asked and about every thirty seconds, deleted when the " +
"phase ends, gone within two minutes of a controller that stopped; read by the build seat's holders, which " +
"take no new check while a key is there; written by the controller alone",
History: 1,
TTL: PlansBuildingKeptFor,
MaxValueSize: 4 << 10,
MaxBytes: 8 << 20,
Storage: jetstream.FileStorage,
}); err != nil {
return fmt.Errorf("asserting bucket %s: %w", PlansBuildingBucket, err)
}
// **The merges mesh-delivery announced: keys that age** (ADR 0420 §7, issue 1025). One message a key; a week.
if _, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{
Bucket: DeliveryMergesBucket,
+14 -7
View File
@@ -361,9 +361,11 @@ var VaultNodeKey = SeatVerb{Seat: "mesh-vault", Verb: "node-key"}
// 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
// controller's own `plans`. Named one by one; a node seat's on any machine, a mesh seat's flat.
// controller's own `plans`, a machine's held builds sent by its own `push` and the builds held after a failed release
// sent by its own `upgrade release-backlog` (novox/hq issue 1732). Named one by one; a node seat's on any machine, a
// mesh seat's flat.
var VerbsTheControllerActsOnAWarrant = []SeatVerb{{Seat: "node-service-manager", Verb: "restart"},
{Seat: ControllerSeat, Verb: "plans"}}
{Seat: ControllerSeat, Verb: "plans"}, {Seat: ControllerSeat, Verb: "push"}, {Seat: ControllerSeat, Verb: "upgrade"}}
// VerbsTheControllerAsksTheOperatorChannel are the operator channel's verbs the controller calls (novox/hq
// ADR 0380): `tell`, one plain message with no condition behind it — a module's own secret given, named,
@@ -387,6 +389,16 @@ type DegradableGrant struct {
// works. Every other grant stays required — missing one of those is a controller that comes up and is
// refused on the first thing it tries, with no fallback that covers it.
var DegradableGrants = []DegradableGrant{
{
Subject: "mesh.seat." + ControllerSeat + ".tool.push",
Reason: "refused: the ask of a held move (Release) is answered, and the refusal told, while the operator's own " +
"push at the terminal still sends the machine (issue 1732; ADR 0259)",
},
{
Subject: "mesh.seat." + ControllerSeat + ".tool.upgrade",
Reason: "refused: the ask of the builds held after a failed release is answered, and the refusal told, while the " +
"operator's own `upgrade release-backlog` at the terminal still releases them (issue 1732; ADR 0259)",
},
{
Subject: "mesh.seat." + VerbsTheControllerAsksTheOperatorChannel[0].Seat + ".tool." + VerbsTheControllerAsksTheOperatorChannel[0].Verb,
Reason: "refused: the controller raises the secret-given condition as before (ADR 0380 §9)",
@@ -412,11 +424,6 @@ var DegradableGrants = []DegradableGrant{
Reason: "refused: the controller asserts this bucket through the API and writes none of its keys — the build seat's " +
"holders do (ADR 0406 §3, issue 974)",
},
{
Subject: "$KV." + PlansBuildingBucket + ".>",
Reason: "refused: the controller then says no plan builds, and no check is held while a delivery builds — a check run " +
"wrongly costs a delivery a place, a check held wrongly costs a pull request its turn (ADR 0483 clause 2, issue 1687)",
},
{
Subject: "mesh.seat." + ControllerSeat + ".event.sent",
Reason: "refused: the controller says a tier sent to a machine only for outsiders, and a send that could not " +
-134
View File
@@ -1,134 +0,0 @@
//go:build integration
package broker
import (
"context"
"errors"
"os"
"path/filepath"
"testing"
"time"
"github.com/nats-io/nats-server/v2/server"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"golang.org/x/crypto/bcrypt"
)
// **A holder of the build seat reads the plans building and the server refuses its write** (novox/hq ADR 0483 clause 2, issue
// 1687): on a real server whose users are composed as the controller composes them, the controller makes the bucket and puts a
// plan's key; the build agent's credential lists the keys and gets one, and its put of a key is refused with a permissions
// violation. It also reads the first waiting ask of a check queue by a direct get, which takes nothing from the queue.
func TestNatsTheBuildSeatsHoldersReadThePlansBuildingAndTheServerRefusesTheirWrite(t *testing.T) { // 1687
hash, err := bcrypt.GenerateFromPassword([]byte("pw"), bcrypt.MinCost)
if err != nil {
t.Fatal(err)
}
seat := Seat{Name: TheBuildSeat, Scope: "node", Accepts: []string{"build"}, Emits: []string{"built", "started"}}
users, err := Users(Records{Nodes: []string{"alpha"}, Assigned: map[string][]Declared{
"alpha": {{Module: RuntimeModule}, {Module: "build-agent", Holds: []Seat{seat}}},
}})
if err != nil {
t.Fatal(err)
}
for i := range users {
users[i].PasswordHash = string(hash)
}
accounts, err := ComposeAccounts(users)
if err != nil {
t.Fatal(err)
}
conf := filepath.Join(t.TempDir(), "bus.conf")
if err := os.WriteFile(conf, []byte("listen: 127.0.0.1:-1\njetstream { store_dir: "+
`"`+t.TempDir()+`"`+" }\n"+accounts), 0o600); err != nil {
t.Fatal(err)
}
opts, err := server.ProcessConfigFile(conf)
if err != nil {
t.Fatalf("the composed accounts do not parse: %v", err)
}
opts.NoLog, opts.NoSigs = true, true
s, err := server.NewServer(opts)
if err != nil {
t.Fatal(err)
}
go s.Start()
if !s.ReadyForConnections(10 * time.Second) {
t.Fatal("the bus did not come up")
}
t.Cleanup(s.Shutdown)
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
connect := func(user string) (*nats.Conn, chan error) {
t.Helper()
refused := make(chan error, 8)
nc, err := nats.Connect(s.ClientURL(), nats.UserInfo(user, "pw"), nats.CustomInboxPrefix("_INBOX."+user),
nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) { refused <- err }))
if err != nil {
t.Fatalf("%s cannot connect: %v", user, err)
}
t.Cleanup(nc.Close)
return nc, refused
}
controller, _ := connect(ControllerName)
cj := OnConn(controller)
if err := AssertMeshStreams(cj); err != nil {
t.Fatalf("the controller cannot make the streams: %v", err)
}
if err := cj.EnsureControllerBuckets(); err != nil {
t.Fatalf("the controller cannot make its buckets: %v", err)
}
cjs, _ := jetstream.New(controller)
ckv, err := cjs.KeyValue(ctx, PlansBuildingBucket)
if err != nil {
t.Fatal(err)
}
if _, err := ckv.Put(ctx, "plan-1", []byte(`{"plan":"plan-1"}`)); err != nil {
t.Fatalf("the controller cannot put a plan's key: %v", err)
}
holder, refused := connect("alpha.build-agent")
hjs, err := jetstream.New(holder)
if err != nil {
t.Fatal(err)
}
kv, err := hjs.KeyValue(ctx, PlansBuildingBucket)
if err != nil {
t.Fatalf("a holder of the build seat cannot bind %s: %v", PlansBuildingBucket, err)
}
keys, err := kv.Keys(ctx)
if err != nil || len(keys) != 1 || keys[0] != "plan-1" {
t.Fatalf("the holder lists %v (%v), want plan-1", keys, err)
}
if e, err := kv.Get(ctx, "plan-1"); err != nil || e == nil {
t.Fatalf("the holder cannot get the key: %v", err)
}
// Its write is refused by the server.
if err := holder.Publish("$KV."+PlansBuildingBucket+".plan-2", []byte(`{}`)); err != nil {
t.Fatal(err)
}
_ = holder.Flush()
select {
case err := <-refused:
if !errors.Is(err, nats.ErrPermissionViolation) {
t.Errorf("the server said %v, want a permissions violation", err)
}
case <-time.After(3 * time.Second):
t.Error("the server did not refuse the holder's put of a plan's key")
}
if _, err := ckv.Get(ctx, "plan-2"); !errors.Is(err, jetstream.ErrKeyNotFound) {
t.Errorf("the refused put left a key: %v", err)
}
// And the direct get of a check queue's first waiting ask, which is granted.
reply, err := holder.RequestWithContext(ctx, "$JS.API.DIRECT.GET."+ChecksOtherStream, []byte(`{"seq":1,"next_by_subj":"`+
CheckSubjectPrefix(TheBuildSeat, false)+`>"}`))
if err != nil {
t.Fatalf("the holder's direct get of %s was not answered: %v", ChecksOtherStream, err)
}
if reply.Header.Get("Status") != "404" {
t.Errorf("an empty queue answered %q, want 404", reply.Header.Get("Status"))
}
}
+1 -1
View File
@@ -29,7 +29,7 @@ accounts {
]
users = [
{ user: "controller", password: "$2a$11$cccccccccccccccccccccc", permissions: {
publish: { allow: ["$JS.ACK.CONTROL.controller.>", "$JS.ACK.DEAD_LETTER_NOTICES.controller.>", "$JS.ACK.EVENTS.controller-merges.>", "$JS.ACK.EVENTS.controller.>", "$JS.API.>", "$KV.SEAT_MESH_BUILD_MACHINE_cancelled.>", "$KV.SEAT_NODE_BUILD_AGENT_cancelled.>", "$KV.mesh-controller_asked.>", "$KV.mesh-controller_assignments.>", "$KV.mesh-controller_builds.>", "$KV.mesh-controller_calls.>", "$KV.mesh-controller_checks-running.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_declared-config.>", "$KV.mesh-controller_delivery-merges.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$KV.mesh-controller_merge-window.>", "$KV.mesh-controller_plans-building.>", "$KV.mesh-controller_predefined.>", "$KV.mesh-controller_rollout-silent.>", "$KV.mesh-controller_rollouts.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "mesh.node.>", "mesh.seat.mesh-build-machine.accept.>", "mesh.seat.mesh-build-machine.tool.>", "mesh.seat.mesh-controller.event.applied", "mesh.seat.mesh-controller.event.assignment-changed", "mesh.seat.mesh-controller.event.built-before", "mesh.seat.mesh-controller.event.checked", "mesh.seat.mesh-controller.event.condition-changed", "mesh.seat.mesh-controller.event.condition-cleared", "mesh.seat.mesh-controller.event.condition-raised", "mesh.seat.mesh-controller.event.declared-config-changed", "mesh.seat.mesh-controller.event.doctor-heartbeat", "mesh.seat.mesh-controller.event.healer-acted", "mesh.seat.mesh-controller.event.plan-moved", "mesh.seat.mesh-controller.event.predefined-changed", "mesh.seat.mesh-controller.event.refused", "mesh.seat.mesh-controller.event.rolled-back", "mesh.seat.mesh-controller.event.secret-replaced", "mesh.seat.mesh-controller.event.sent", "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.check.>", "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.*", "mesh.seat.operator-channel.tool.tell"] }
publish: { allow: ["$JS.ACK.CONTROL.controller.>", "$JS.ACK.DEAD_LETTER_NOTICES.controller.>", "$JS.ACK.EVENTS.controller-merges.>", "$JS.ACK.EVENTS.controller.>", "$JS.API.>", "$KV.SEAT_MESH_BUILD_MACHINE_cancelled.>", "$KV.SEAT_NODE_BUILD_AGENT_cancelled.>", "$KV.mesh-controller_asked.>", "$KV.mesh-controller_assignments.>", "$KV.mesh-controller_builds.>", "$KV.mesh-controller_calls.>", "$KV.mesh-controller_checks-running.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_declared-config.>", "$KV.mesh-controller_delivery-merges.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$KV.mesh-controller_merge-window.>", "$KV.mesh-controller_predefined.>", "$KV.mesh-controller_rollout-silent.>", "$KV.mesh-controller_rollouts.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "mesh.node.>", "mesh.seat.mesh-build-machine.accept.>", "mesh.seat.mesh-build-machine.tool.>", "mesh.seat.mesh-controller.event.applied", "mesh.seat.mesh-controller.event.assignment-changed", "mesh.seat.mesh-controller.event.built-before", "mesh.seat.mesh-controller.event.checked", "mesh.seat.mesh-controller.event.condition-changed", "mesh.seat.mesh-controller.event.condition-cleared", "mesh.seat.mesh-controller.event.condition-raised", "mesh.seat.mesh-controller.event.declared-config-changed", "mesh.seat.mesh-controller.event.doctor-heartbeat", "mesh.seat.mesh-controller.event.healer-acted", "mesh.seat.mesh-controller.event.plan-moved", "mesh.seat.mesh-controller.event.predefined-changed", "mesh.seat.mesh-controller.event.refused", "mesh.seat.mesh-controller.event.rolled-back", "mesh.seat.mesh-controller.event.secret-replaced", "mesh.seat.mesh-controller.event.sent", "mesh.seat.mesh-controller.tool.plans", "mesh.seat.mesh-controller.tool.push", "mesh.seat.mesh-controller.tool.upgrade", "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.check.>", "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.*", "mesh.seat.operator-channel.tool.tell"] }
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" }
} }
-4
View File
@@ -177,10 +177,6 @@ var WritersTable = []WriterRow{
KeptIn: "key-value " + CheckRunningBucket + ", aged out when not renewed", Others: "the build seat's holders read it to drop a duplicate request",
Subjects: kvOf(CheckRunningBucket), Writes: holdsTheBuildSeat,
Shared: "every machine holding the build seat puts the key of the check it runs and deletes it when the check ends"},
{State: "a delivery plan in its build phase", Writer: "controller (lease holder), at the first build's ask and at each refresh",
KeptIn: "key-value " + PlansBuildingBucket + ", aged out when not refreshed",
Others: "the build seat's holders read it and take no new check while a key is there (ADR 0483)",
Subjects: kvOf(PlansBuildingBucket), Writes: isController},
{State: "a merge announced", Writer: "one announcer per forge (the hook, or the poll when the hook is absent — never both)",
KeptIn: "the bus", Others: "—", Subjects: []string{"mesh.mod.*.event.pull.merged"}, Writes: ownModule},
// A pull request's head, before it merges (novox/hq to-be 45 §9): the same one announcer.
-2
View File
@@ -29,8 +29,6 @@ var designRows = []string{
"builds and their outcomes",
"a pull request's check asked",
"a pull request's check running",
// The plans in their build phase, a bucket the build seat's holders read (ADR 0483 clause 2, issue 1687).
"a delivery plan in its build phase",
"a merge announced",
"a pull request's head announced",
"a pull request's merge check",
+24 -4
View File
@@ -59,15 +59,35 @@ func (r Registry) PublishImage(ctx context.Context, localTag, repository string)
}); err != nil {
return "", err
}
out, err := r.Run(ctx, "", "docker", "inspect", "--format", "{{index .RepoDigests 0}}", remote)
// Every digest the daemon lists for the image, one per line: the image may have been pushed under other names before
// (a check's scratch copy, another registry), and the first of them is not the one this push made (novox/hq issue
// 1742: a build recorded a digest the store never held).
out, err := r.Run(ctx, "", "docker", "inspect", "--format", "{{range .RepoDigests}}{{println .}}{{end}}", remote)
if err != nil {
return "", fmt.Errorf("pushed %s and cannot read back what it was pinned as: %w", remote, err)
}
pinned := strings.TrimSpace(out)
if !strings.Contains(pinned, "@sha256:") {
first, pinned := "", ""
for _, line := range strings.Split(out, "\n") {
line = strings.TrimSpace(line)
if line == "" {
continue
}
if first == "" {
first = line
}
if strings.HasPrefix(line, remote+"@sha256:") {
pinned = line
break
}
}
if pinned == "" {
if strings.Contains(first, "@sha256:") {
return "", fmt.Errorf("pushed %s, but the daemon lists no digest for that name (only %q): the digest of another "+
"repository is not what this registry serves", remote, first)
}
// A tag is not a pin. It can be made to point at something else, and this file is applied
// on machines with no mesh to ask about anything (novox/hq ADR 0006).
return "", fmt.Errorf("%s came back as %q, which is not pinned by digest", remote, pinned)
return "", fmt.Errorf("%s came back as %q, which is not pinned by digest", remote, first)
}
return pinned, nil
}
+29
View File
@@ -252,3 +252,32 @@ func TestAPushedImageComesBackPinned(t *testing.T) {
t.Fatalf("it did not tag and then push: %v", ran)
}
}
// 1742: an image pushed under other names before lists their digests too, and the digest recorded is the one of the name
// this push made — never the first listed, which may be another repository's.
func TestThePinnedDigestIsTheOneOfTheNamePushed(t *testing.T) {
want := "registry.invalid/shell/server@sha256:" + strings.Repeat("b", 64)
earlier := "registry.invalid/scratch/shell/server@sha256:" + strings.Repeat("a", 64)
r := Registry{Address: "registry.invalid", Run: func(
_ context.Context, _, name string, args ...string) (string, error) {
if name == "docker" && len(args) > 0 && args[0] == "inspect" {
return earlier + "\n" + want + "\n", nil
}
return "", nil
}}
got, err := r.PublishImage(context.Background(), "local", "shell/server")
if err != nil || got != want {
t.Fatalf("pinned %q (%v); want %q", got, err, want)
}
// Only another repository's digest is listed: the push is not taken for pinned.
r.Run = func(_ context.Context, _, name string, args ...string) (string, error) {
if name == "docker" && len(args) > 0 && args[0] == "inspect" {
return earlier + "\n", nil
}
return "", nil
}
if _, err := r.PublishImage(context.Background(), "local", "shell/server"); err == nil ||
!strings.Contains(err.Error(), "no digest for that name") {
t.Fatalf("a digest of another repository was taken for the push's own: %v", err)
}
}
+4
View File
@@ -292,6 +292,10 @@ type GroupMember struct {
// Base is the branch it merges into: the build seat merges the head into that trunk's head before the
// group is composed (novox/hq ADR 0323 §6). A build seat from before it composes the head alone.
Base string `json:"base,omitempty"`
// Trunk is the base branch's head the forge announced with the head (novox/hq issue 1106): the trunk commit the
// composed verdict names for this member when the build seat named none, so a member whose trunk did not move is
// never merged-refused for a verdict that names no trunk.
Trunk string `json:"trunk,omitempty"`
}
// Who judges a merge check's gate.
+62 -174
View File
@@ -6,7 +6,6 @@ import (
"errors"
"fmt"
"os"
"sync"
"time"
"github.com/nats-io/nats.go"
@@ -163,21 +162,11 @@ type natsMachine struct {
// build waits, then the checks of pull requests marked ready, then every other check (novox/hq ADR 0406 §3).
subs []*takeQueue
opts MachineOptions
// gate is what the loop asks of the bus to hold checks while a delivery builds; nil is the bus's (a test sets a fake).
gate checkGate
// said is the faults of the hold already said, letKey whether this holder created the let-through key for the check it
// took last (ADR 0483 clause 3).
said map[string]bool
letKey bool
// mu guards the subscriptions of subs, bound and let go of by the taking loop and closed by Close.
mu sync.Mutex
}
// takeQueue is one queue a machine pulls from.
type takeQueue struct {
sub *nats.Subscription
// pull is what the queue is fetched through: the subscription once bound (a test sets a fake).
pull puller
// kind is what its asks are: a build, a ready check, or another check.
kind string
// worker is the consumer bound; bound is false until a bind succeeded, for a queue the mesh may not have made yet.
@@ -185,11 +174,6 @@ type takeQueue struct {
bound bool
}
// puller is the fetch of a queue.
type puller interface {
Fetch(batch int, opts ...nats.PullOpt) ([]*nats.Msg, error)
}
// The kinds of queue a machine takes from, in the order it takes from them.
const (
queueBuilds = "builds"
@@ -211,9 +195,6 @@ type MachineOptions struct {
// the request in hand first and reads its drain after, so an ask of `busy` racing the fetch sees one or
// the other, never neither. Nil claims every one.
Claim func() bool
// LetThroughAfter is how long a check waits while a delivery builds before it is let through; zero is ADR 0483's thirty
// minutes. Set by a test.
LetThroughAfter time.Duration
}
// MachineOverNATS takes build work from the current build role.
@@ -241,8 +222,6 @@ const (
)
func (m *natsMachine) Close() {
m.mu.Lock()
defer m.mu.Unlock()
for _, q := range m.subs {
if q.sub != nil {
_ = q.sub.Unsubscribe()
@@ -265,9 +244,6 @@ func (m *natsMachine) Close() {
// **In this order** (novox/hq ADR 0406 §3): a delivery build, then a check of a pull request marked ready, then any
// other check, each queue tried without waiting; when all three are empty the first is waited on for a second at most
// and the three are tried again, so a delivery build is never much more than a second behind and nothing spins.
//
// **Unless a delivery builds** (novox/hq ADR 0483): while the controller's bucket of the plans building holds a key, no check queue
// is fetched from, but for the one check that waited LetThroughAfter, which goes first (see next).
func (m *natsMachine) Take(ctx context.Context, do func(context.Context, Build)) error {
worker, found := broker.HolderConsumerFor(m.on, "build-agent",
broker.DeclaredSeat{Name: m.seat, Accepts: []string{"build"}})
@@ -312,19 +288,6 @@ func (m *natsMachine) Take(ctx context.Context, do func(context.Context, Build))
}
continue
}
// The seat's own queue is bound again when a running ask made the machine let go of it (quiesce): a machine that cannot
// is not taking, which is said and tried again.
if !m.subs[0].bound {
if err := m.bind(m.subs[0]); err != nil {
fmt.Fprintf(os.Stderr, "this machine cannot bind %s again after a build, and tries again: %v\n", m.seat, err)
select {
case <-ctx.Done():
return nil
case <-time.After(pausedPoll):
}
continue
}
}
if time.Now().After(bindAgain) {
for _, q := range m.subs[1:] {
if !q.bound {
@@ -349,7 +312,7 @@ func (m *natsMachine) Take(ctx context.Context, do func(context.Context, Build))
// say why (2026-10-03, the first build agents). Asked again.
continue
default:
if m.subs[0].sub != nil && m.subs[0].sub.IsValid() {
if m.subs[0].sub.IsValid() {
// A transient fault in asking — a reconnect, a slow server — is asked past rather
// than ending the machine; one that outlasts the ack wait redelivers nothing lost.
time.Sleep(time.Second)
@@ -357,106 +320,67 @@ func (m *natsMachine) Take(ctx context.Context, do func(context.Context, Build))
}
return fmt.Errorf("the bus stopped delivering build work: %w", err)
}
m.handOver(ctx, msg, q, do)
// And the queues it let go of are bound again at once, not after the half minute.
bindAgain = time.Time{}
}
}
// handOver settles one fetched ask: dropped, handed back or built. **A check let through keeps its key for as long as it
// stays here** (ADR 0483 clause 3): refreshed while it runs and deleted when it ends, however it ends.
func (m *natsMachine) handOver(ctx context.Context, msg *nats.Msg, q *takeQueue, do func(context.Context, Build)) {
if m.letKey {
defer m.keepLetThrough(ctx, heartbeatWhileWorking)()
}
// **Paused while the pull was answered** (ADR 0219): "takes no new build" holds even for
// the one that arrived in that moment. Handed back at once for another holder — counted as
// a delivery, which a pause landing exactly then costs and nothing else does.
if m.opts.Paused != nil && m.opts.Paused() {
_ = msg.Nak()
return
}
var request BuildRequest
if err := json.Unmarshal(msg.Data, &request); err != nil {
// Unreadable: terminated rather than retried, because the next attempt reads the same
// bytes. Nobody waiting hears an answer, which is right — there was no request.
_ = msg.Term()
return
}
// **Cancelled in the moment it was fetched** (novox/hq ADR 0219): the controller wrote the
// id into the seat's cancelled set before deleting the ask, so one taken in between is
// ended here, terminated rather than redelivered, and its outcome said as failed — never
// built. A set that cannot be read is said and the ask built: a cancel is a person's
// exception, and a holder that refused every build while its grant was missing would
// stop the mesh building for it.
build := &natsBuild{request: request, msg: msg, on: m.on, js: m.js, seat: m.seat, delivery: deliveryOf(msg)}
if cancelled, err := IsCancelled(m.js.Conn(), m.seat, request.ID); err != nil {
fmt.Fprintf(os.Stderr, "could not read whether %s was cancelled, so it is built: %v\n", request.ID, err)
} else if cancelled {
endCancelled(ctx, build)
return
}
// **A check left behind is dropped when it is taken** (novox/hq ADR 0406 §3): a request on the third queue for a
// head and a trunk commit its pull request's key already says were judged is acknowledged and not run.
if q.kind == queueOther {
if why := m.judgedAlready(ctx, request); why != "" {
fmt.Fprintf(os.Stderr, "%s was left behind in the check queue and is dropped: %s\n", request.ID, why)
_ = msg.Ack()
return
}
}
// **Drained in the moment it was fetched** (novox/hq ADR 0335): handed back at once, as a pause
// landing then is.
if m.opts.Claim != nil && !m.opts.Claim() {
_ = msg.Nak()
return
}
// **A duplicate of a check another builder is running is dropped** (issue 1649): an ask redelivered to this machine
// because its first holder did not say, in time, that it had it. The running key names that builder and this same
// ask; running it a second time would spend a second place on one check.
if why := m.runningElsewhere(ctx, request); why != "" {
fmt.Fprintf(os.Stderr, "%s is handed out again while another builder runs it and is dropped: %s\n", request.ID, why)
_ = msg.Ack()
return
}
// A build outlives the acknowledgement window many times over; said while it runs,
// as the controller says it for its own long handlers, so the server neither hands
// the ask to a second machine nor counts the wait against its deliveries.
working := make(chan struct{})
go stillWorking(msg, working, heartbeatWhileWorking)
// And a check says it is running, to the other builders (ADR 0406 §3): its key put now and at each heartbeat,
// deleted when it ends, whatever the outcome.
stopRunning := keepRunning(ctx, m.js.Conn(), request, m.on, heartbeatWhileWorking)
// **And holds nothing else while it runs** (issue 1747): see quiesce.
m.quiesce(q)
do(ctx, build)
stopRunning()
close(working)
// **The queue it came from too, once the ask is settled**: a fetch that returned a late message leaves its pull request
// open, and the server can deliver one more into the subscription (issue 1747). Take binds every queue again first.
m.quiesce(nil)
}
// quiesce lets go of every queue but the one the running ask came from, for as long as it runs. **A builder never holds a
// message it does not start** (issue 1747). A message the server sends after a fetch gave up, within the few milliseconds its
// pull request still lived, lands in that queue's subscription and stays there until the next fetch of it: held, never
// acknowledged as in progress, for as long as the builder runs its current ask, five or ten minutes for a check, while the
// holders that could build it are all busy and a delivery waits behind it. The legacy pull subscription cannot be read
// without sending a new pull request, which could take one more, so it cannot be searched for such a message. It is dropped
// instead: the subscription is closed, a message in it or on its way is lost to this builder and the server hands it out again
// when its worker's wait for an acknowledgement ends (a minute), to any holder that is free, and the queue is bound again
// before the next fetch (Take). The queue the running ask came from keeps its subscription, which its acknowledgements use.
func (m *natsMachine) quiesce(keep *takeQueue) {
m.mu.Lock()
defer m.mu.Unlock()
for _, q := range m.subs {
if q == keep || !q.bound {
// **Paused while the pull was answered** (ADR 0219): "takes no new build" holds even for
// the one that arrived in that moment. Handed back at once for another holder — counted as
// a delivery, which a pause landing exactly then costs and nothing else does.
if m.opts.Paused != nil && m.opts.Paused() {
_ = msg.Nak()
continue
}
if q.sub != nil {
_ = q.sub.Unsubscribe()
var request BuildRequest
if err := json.Unmarshal(msg.Data, &request); err != nil {
// Unreadable: terminated rather than retried, because the next attempt reads the same
// bytes. Nobody waiting hears an answer, which is right — there was no request.
_ = msg.Term()
continue
}
q.sub, q.pull, q.bound = nil, nil, false
// **Cancelled in the moment it was fetched** (novox/hq ADR 0219): the controller wrote the
// id into the seat's cancelled set before deleting the ask, so one taken in between is
// ended here, terminated rather than redelivered, and its outcome said as failed — never
// built. A set that cannot be read is said and the ask built: a cancel is a person's
// exception, and a holder that refused every build while its grant was missing would
// stop the mesh building for it.
build := &natsBuild{request: request, msg: msg, on: m.on, js: m.js, seat: m.seat, delivery: deliveryOf(msg)}
if cancelled, err := IsCancelled(m.js.Conn(), m.seat, request.ID); err != nil {
fmt.Fprintf(os.Stderr, "could not read whether %s was cancelled, so it is built: %v\n", request.ID, err)
} else if cancelled {
endCancelled(ctx, build)
continue
}
// **A check left behind is dropped when it is taken** (novox/hq ADR 0406 §3): a request on the third queue for a
// head and a trunk commit its pull request's key already says were judged is acknowledged and not run.
if q.kind == queueOther {
if why := m.judgedAlready(ctx, request); why != "" {
fmt.Fprintf(os.Stderr, "%s was left behind in the check queue and is dropped: %s\n", request.ID, why)
_ = msg.Ack()
continue
}
}
// **Drained in the moment it was fetched** (novox/hq ADR 0335): handed back at once, as a pause
// landing then is.
if m.opts.Claim != nil && !m.opts.Claim() {
_ = msg.Nak()
continue
}
// **A duplicate of a check another builder is running is dropped** (issue 1649): an ask redelivered to this machine
// because its first holder did not say, in time, that it had it. The running key names that builder and this same
// ask; running it a second time would spend a second place on one check.
if why := m.runningElsewhere(ctx, request); why != "" {
fmt.Fprintf(os.Stderr, "%s is handed out again while another builder runs it and is dropped: %s\n", request.ID, why)
_ = msg.Ack()
continue
}
// A build outlives the acknowledgement window many times over; said while it runs,
// as the controller says it for its own long handlers, so the server neither hands
// the ask to a second machine nor counts the wait against its deliveries.
working := make(chan struct{})
go stillWorking(msg, working, heartbeatWhileWorking)
// And a check says it is running, to the other builders (ADR 0406 §3): its key put now and at each heartbeat,
// deleted when it ends, whatever the outcome.
stopRunning := keepRunning(ctx, m.js.Conn(), request, m.on, heartbeatWhileWorking)
do(ctx, build)
stopRunning()
close(working)
}
}
@@ -496,39 +420,25 @@ func duplicateOf(running CheckRunning, found bool, r BuildRequest, me string) st
// subscribing `…accept.build` against a consumer filtered on `…accept.>` is rejected even though it is narrower. Learned
// twice now, on two different consumers, which is why it is written down here.
func (m *natsMachine) bind(q *takeQueue) error {
m.mu.Lock()
defer m.mu.Unlock()
sub, err := m.js.Context().PullSubscribe(q.worker.Filters[0], q.worker.Name,
nats.Bind(q.worker.Stream, q.worker.Name), nats.ManualAck())
if err != nil {
return err
}
q.sub, q.pull, q.bound = sub, sub, true
q.sub, q.bound = sub, true
return nil
}
// next takes the next request in the order of the queues: each tried for a moment, and when all are empty the first waited
// on for a second. Nothing, with no error, when there was nothing.
func (m *natsMachine) next(ctx context.Context) (*nats.Msg, *takeQueue, error) {
// **While a delivery builds, only delivery builds are taken, but for one check that waited** (novox/hq ADR 0483): the
// bucket of the plans building is read before either check queue is fetched from, and a key there holds both. The check
// that waited LetThroughAfter goes first, ahead of the builds that wait; the rest of the holders take builds only.
if m.seat == broker.TheBuildSeat && len(m.subs) > 1 {
if plans := m.building(ctx); len(plans) > 0 {
if msg, q := m.letOneThrough(ctx); msg != nil {
fmt.Fprintf(os.Stderr, "a check that waited %s is let through while %s builds\n", m.letThroughAfter(), whyHeld(plans))
return msg, q, nil
}
return m.nextBuild(ctx)
}
}
var last error
for _, q := range m.subs {
if !q.bound {
continue
}
trying, end := context.WithTimeout(ctx, checkTakeWait)
fetched, err := q.pull.Fetch(1, nats.Context(trying))
fetched, err := q.sub.Fetch(1, nats.Context(trying))
end()
if len(fetched) > 0 {
return fetched[0], q, nil
@@ -544,18 +454,9 @@ func (m *natsMachine) next(ctx context.Context) (*nats.Msg, *takeQueue, error) {
if last != nil {
return nil, nil, last
}
return m.nextBuild(ctx)
}
// nextBuild waits for a second at most for a delivery build, the seat's own queue.
func (m *natsMachine) nextBuild(ctx context.Context) (*nats.Msg, *takeQueue, error) {
if !m.subs[0].bound {
// Let go of while an ask ran and not bound again yet (Take binds it before it asks): nothing to take from.
return nil, nil, nil
}
waiting, end := context.WithTimeout(ctx, takeIdleWait)
defer end()
fetched, err := m.subs[0].pull.Fetch(1, nats.Context(waiting))
fetched, err := m.subs[0].sub.Fetch(1, nats.Context(waiting))
if len(fetched) > 0 {
return fetched[0], m.subs[0], nil
}
@@ -718,16 +619,3 @@ func endCancelled(ctx context.Context, b *natsBuild) {
}
_ = b.msg.Term()
}
// boundKinds are the kinds of queue this machine holds a subscription to now.
func (m *natsMachine) boundKinds() []string {
m.mu.Lock()
defer m.mu.Unlock()
var out []string
for _, q := range m.subs {
if q.bound {
out = append(out, q.kind)
}
}
return out
}
+2 -6
View File
@@ -287,16 +287,12 @@ type CheckRunning struct {
}
// RunningKeyOf is the key a check is said running under: its pull request's tokens, or `group.<name>` for a group's
// composed check; `try.<id>` for a try, which no other request replaces.
// composed check; none for a try, which no other request replaces.
func RunningKeyOf(r BuildRequest) (string, bool) {
c := r.Check
switch {
case c == nil:
case c == nil, c.Try:
return "", false
case c.Try:
// A try says it runs under its own id, which no other request shares: so a reader of the bucket can tell a try that runs
// from a check left waiting for a redelivery (ADR 0483 clause 3).
return "try." + tokenOf(r.ID), true
case c.Group != "":
return "group." + tokenOf(c.Group), true
case c.Number <= 0:
+15 -5
View File
@@ -6,6 +6,7 @@ import (
"context"
"encoding/json"
"reflect"
"strings"
"testing"
"time"
@@ -594,12 +595,21 @@ func TestABusyMachineTakesNoWaitingAskWhateverItsPullsHold(t *testing.T) {
case <-time.After(10 * time.Second):
t.Fatal("builder a took nothing")
}
// **A busy machine holds no pull at all** (issues 1649 and 1747): the queues the running ask did not come from are let go of,
// so no message, a status message or a late delivery, can sit in one while it runs.
if kinds := a.boundKinds(); len(kinds) != 1 || kinds[0] != queueBuilds {
t.Fatalf("builder a holds %v while it builds; want only the queue its build came from", kinds)
var other *takeQueue
for _, q := range a.subs {
if q.kind == queueOther && q.bound {
other = q
}
}
if other == nil {
t.Fatal("builder a has no pull on the other checks")
}
// The queue is empty, so this request is answered at its expiry with a status message that lands in the pull.
reply := strings.TrimSuffix(other.sub.Subject, "*") + "late"
if err := js.Conn().PublishRequest("$JS.API.CONSUMER.MSG.NEXT."+other.worker.Stream+"."+other.worker.Name, reply,
[]byte(`{"batch":1,"expires":1000000}`)); err != nil {
t.Fatal(err)
}
other := &takeQueue{kind: queueOther, worker: broker.CheckWorkers()[1]}
time.Sleep(500 * time.Millisecond)
for i, id := range []string{"wait-1", "wait-2"} {
if err := asks.Ask(ctx, aPullCheck(id, 20+i, "head-x-000000000", "trunk-1-000000", false)); err != nil {
@@ -1,479 +0,0 @@
//go:build integration
package link
import (
"context"
"encoding/json"
"fmt"
"github.com/nats-io/nats.go"
"reflect"
"strings"
"sync"
"testing"
"time"
"github.com/nats-io/nats.go/jetstream"
"github.com/novox/mesh-controller/internal/broker"
)
// Checks wait while a delivery builds (novox/hq ADR 0483, issues 1652, 1687, 1688), against a real bus.
// aBusWithTheBuckets is the bus with the build role and the controller's buckets, the bucket of the checks running with an age
// of two seconds so a key a holder stopped refreshing is seen to expire.
func aBusWithTheBuckets(t *testing.T) (*broker.JetStream, jetstream.KeyValue) {
t.Helper()
js := aBusWithTheBuildRole(t)
api, err := jetstream.New(js.Conn())
if err != nil {
t.Fatal(err)
}
ctx := t.Context()
if _, err := api.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: broker.CheckRunningBucket, History: 1, TTL: 2 * time.Second}); err != nil {
t.Fatal(err)
}
plans, err := api.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: broker.PlansBuildingBucket, History: 1, TTL: broker.PlansBuildingKeptFor})
if err != nil {
t.Fatal(err)
}
return js, plans
}
func aCheckAskedAgo(ago time.Duration, number int, ready bool) BuildRequest {
r := aPullCheck(fmt.Sprintf("build-%d", time.Now().Add(-ago).UnixNano()), number, fmt.Sprintf("head-%d-0000000", number), "trunk-0-000000", ready)
return r
}
// 1652/AC6: a plan's builds asked in two tiers with a check waiting, the check is handed out only after the last tier's builds,
// and not in the gap between the tiers.
func TestNatsACheckIsHandedOutOnlyAfterTheLastTierOfAPlan(t *testing.T) {
js, plans := aBusWithTheBuckets(t)
ctx, stop := context.WithCancel(t.Context())
defer stop()
asks := BuildsOn(js, TheBuildMachine)
if _, err := plans.Put(ctx, "plan-1", []byte(`{"plan":"plan-1"}`)); err != nil {
t.Fatal(err)
}
if err := asks.Ask(ctx, aCheckAskedAgo(time.Minute, 7, true)); err != nil {
t.Fatal(err)
}
if err := asks.Ask(ctx, BuildRequest{ID: "tier-1-a", Repository: "/a"}); err != nil {
t.Fatal(err)
}
took := make(chan string, 16)
machine := MachineOverNATS(js, "anchor")
defer machine.Close()
go func() {
_ = machine.Take(ctx, func(ctx context.Context, work Build) {
took <- work.Request().ID
_ = work.Announce(ctx, BuildResult{ID: work.Request().ID, On: "anchor", Failed: "no"})
_ = work.Done()
})
}()
expect := func(want string) {
t.Helper()
select {
case got := <-took:
if got != want {
t.Fatalf("taken %s, want %s", got, want)
}
case <-time.After(15 * time.Second):
t.Fatalf("%s was not taken", want)
}
}
expect("tier-1-a")
// The gap between two tiers: the delivery's queue is empty, the plan's key is there, and the check is not taken.
select {
case got := <-took:
t.Fatalf("%s was taken in the gap between two tiers", got)
case <-time.After(3 * time.Second):
}
if err := asks.Ask(ctx, BuildRequest{ID: "tier-2-a", Repository: "/b"}); err != nil {
t.Fatal(err)
}
expect("tier-2-a")
select {
case got := <-took:
t.Fatalf("%s was taken before the plan's last build was built and its key deleted", got)
case <-time.After(2 * time.Second):
}
// The phase ends: the key is deleted, and the check is handed out.
if err := plans.Delete(ctx, "plan-1"); err != nil {
t.Fatal(err)
}
select {
case got := <-took:
if got == "tier-1-a" || got == "tier-2-a" {
t.Fatalf("a build was taken twice: %s", got)
}
case <-time.After(15 * time.Second):
t.Fatal("the check was not handed out once the plan's key was deleted")
}
}
// 1688: two holders racing for checks that waited past the bound: exactly one is let through at a time, by a create that fails
// while the key is there, and the key is deleted when its check ends, after which the next oldest goes.
func TestNatsTwoHoldersRacingForOldChecksLetExactlyOneThrough(t *testing.T) {
js, plans := aBusWithTheBuckets(t)
ctx, stop := context.WithCancel(t.Context())
defer stop()
// The buckets' age is two seconds here, so the keys are refreshed well within it.
was := heartbeatWhileWorking
heartbeatWhileWorking = 400 * time.Millisecond
t.Cleanup(func() { heartbeatWhileWorking = was })
asks := BuildsOn(js, TheBuildMachine)
if _, err := plans.Put(ctx, "plan-1", []byte(`{"plan":"plan-1"}`)); err != nil {
t.Fatal(err)
}
for i, ago := range []time.Duration{50 * time.Minute, 40 * time.Minute, 35 * time.Minute} {
if err := asks.Ask(ctx, aCheckAskedAgo(ago, i+1, false)); err != nil {
t.Fatal(err)
}
}
var mu sync.Mutex
running, most := 0, 0
started := make(chan string, 8)
release := make(chan struct{})
do := func(ctx context.Context, work Build) {
mu.Lock()
running++
most = max(most, running)
mu.Unlock()
started <- work.Request().Check.Repo + fmt.Sprint(work.Request().Check.Number)
<-release
mu.Lock()
running--
mu.Unlock()
_ = work.Announce(ctx, BuildResult{ID: work.Request().ID, On: "x", Failed: "no"})
_ = work.Done()
}
for _, on := range []string{"ace", "shanks"} {
machine := MachineOverNATS(js, on)
defer machine.Close()
go func() { _ = machine.Take(ctx, do) }()
}
var order []string
for range 3 {
select {
case id := <-started:
order = append(order, id)
case <-time.After(20 * time.Second):
t.Fatalf("started so far: %v", order)
}
// While it runs, no second check starts and the key is there.
select {
case id := <-started:
t.Fatalf("%s started beside a check let through", id)
case <-time.After(2500 * time.Millisecond):
}
kv := runningKV(t, js)
if _, err := kv.Get(ctx, LetThroughKey); err != nil {
t.Fatalf("the let-through key is not there while the check runs: %v", err)
}
release <- struct{}{}
// It is deleted when the check ends: the next check let through creates the key again at once, and a create fails
// while the key is there, so its start (the next round) is the proof; after the last one the key stays gone.
if len(order) == 3 {
deadline := time.Now().Add(10 * time.Second)
for {
if _, err := kv.Get(ctx, LetThroughKey); err != nil {
break
}
if time.Now().After(deadline) {
t.Fatal("the let-through key was not deleted when the last check ended")
}
time.Sleep(50 * time.Millisecond)
}
}
}
if want := []string{"a1", "a2", "a3"}; !reflect.DeepEqual(order, want) {
t.Fatalf("checks went %v, want the oldest first %v", order, want)
}
mu.Lock()
defer mu.Unlock()
if most != 1 {
t.Fatalf("%d checks ran at once while a delivery built", most)
}
}
func runningKV(t *testing.T, js *broker.JetStream) jetstream.KeyValue {
t.Helper()
api, err := jetstream.New(js.Conn())
if err != nil {
t.Fatal(err)
}
kv, err := api.KeyValue(t.Context(), broker.CheckRunningBucket)
if err != nil {
t.Fatal(err)
}
return kv
}
// 1688: the create of the let-through key succeeds for exactly one of holders asking at once, and a key its holder stopped
// refreshing expires, after which the next create succeeds.
func TestNatsOneCreateOfTheLetThroughKeySucceedsAndAStoppedHoldersKeyExpires(t *testing.T) {
js, _ := aBusWithTheBuckets(t)
gate := busGate{conn: js.Conn()}
ctx := t.Context()
var wg sync.WaitGroup
var mu sync.Mutex
won := 0
for i := range 8 {
wg.Add(1)
go func() {
defer wg.Done()
ok, err := gate.TakeLetThrough(ctx, fmt.Sprint("holder-", i))
if err != nil {
t.Error(err)
return
}
if ok {
mu.Lock()
won++
mu.Unlock()
}
}()
}
wg.Wait()
if won != 1 {
t.Fatalf("%d of 8 holders created the let-through key, want exactly one", won)
}
// The holder stopped: nothing refreshes the key, and the bucket's age (two seconds here) removes it.
deadline := time.Now().Add(15 * time.Second)
for {
ok, err := gate.TakeLetThrough(ctx, "next")
if err != nil {
t.Fatal(err)
}
if ok {
break
}
if time.Now().After(deadline) {
t.Fatal("the key of a stopped holder did not expire")
}
time.Sleep(250 * time.Millisecond)
}
// And a refreshed key does not expire.
for range 6 {
if err := gate.KeepLetThrough(ctx, "next"); err != nil {
t.Fatal(err)
}
time.Sleep(700 * time.Millisecond)
}
if ok, _ := gate.TakeLetThrough(ctx, "other"); ok {
t.Fatal("a refreshed key expired")
}
var v LetThrough
e, err := runningKV(t, js).Get(ctx, LetThroughKey)
if err != nil || json.Unmarshal(e.Value(), &v) != nil || v.Builder != "next" {
t.Fatalf("the key says %+v (%v)", v, err)
}
}
// 1687: the oldest waiting ask is read without being taken: no delivery is counted against it.
func TestNatsTheOldestWaitingAskIsReadWithoutTakingIt(t *testing.T) {
js, _ := aBusWithTheBuckets(t)
ctx := t.Context()
asks := BuildsOn(js, TheBuildMachine)
old := aCheckAskedAgo(45*time.Minute, 1, false)
if err := asks.Ask(ctx, old); err != nil {
t.Fatal(err)
}
gate := busGate{conn: js.Conn()}
q := &takeQueue{kind: queueOther, worker: broker.CheckWorkers()[1]}
at, found, err := gate.OldestWaiting(ctx, q)
if err != nil || !found {
t.Fatalf("the first waiting ask was not read: %v %v", found, err)
}
if want, _ := BuildAskedAt(old.ID); !at.Equal(want) {
t.Fatalf("read %v, asked %v", at, want)
}
for range 3 {
if _, _, err := gate.OldestWaiting(ctx, q); err != nil {
t.Fatal(err)
}
}
info, err := js.Context().ConsumerInfo(q.worker.Stream, q.worker.Name)
if err != nil {
t.Fatal(err)
}
if info.Delivered.Consumer != 0 || info.NumAckPending != 0 || info.NumPending != 1 {
t.Fatalf("reading the age took the ask: delivered %d, pending acks %d, waiting %d", info.Delivered.Consumer, info.NumAckPending, info.NumPending)
}
// And an empty queue reads as none.
if _, found, err := gate.OldestWaiting(ctx, &takeQueue{kind: queueReady, worker: broker.CheckWorkers()[0]}); err != nil || found {
t.Fatalf("an empty queue read as %v (%v)", found, err)
}
}
// Issue 1747: a message that reaches a builder's subscription outside a fetch is not held while the builder runs another ask. A
// stray check lands in the other queue's subscription of builder A, which then runs a long check; the server hands the stray
// out again when its wait for an acknowledgement ends (two seconds here), builder B runs it, and A, finished, does not run it a
// second time from the subscription it held.
func TestNatsAMessageDeliveredOutsideAFetchIsNotHeldWhileTheBuilderRunsAnotherAsk(t *testing.T) { // 1747
js := aBusWithTheBuildRole(t)
ctx, stop := context.WithCancel(t.Context())
defer stop()
for _, w := range broker.CheckWorkers() {
w.AckWaitSeconds = 2
if err := js.EnsureConsumer(w); err != nil {
t.Fatal(err)
}
}
// The long check says it is alive well within the ack wait of two seconds.
was := heartbeatWhileWorking
heartbeatWhileWorking = 400 * time.Millisecond
t.Cleanup(func() { heartbeatWhileWorking = was })
asks := BuildsOn(js, TheBuildMachine)
long := aPullCheck("build-"+fmt.Sprint(time.Now().UnixNano()), 1, "head-1-0000000", "trunk-0-000000", true)
stray := aPullCheck("build-"+fmt.Sprint(time.Now().UnixNano()+1), 2, "head-2-0000000", "trunk-0-000000", false)
for _, r := range []BuildRequest{long, stray} {
if err := asks.Ask(ctx, r); err != nil {
t.Fatal(err)
}
}
var mu sync.Mutex
runs := map[string][]string{} // ask id -> the machines that ran it
release := make(chan struct{})
started := make(chan string, 8)
do := func(on string) func(context.Context, Build) {
return func(ctx context.Context, work Build) {
id := work.Request().ID
mu.Lock()
runs[id] = append(runs[id], on)
mu.Unlock()
started <- on + ":" + id
if id == long.ID {
<-release
}
_ = work.Announce(ctx, BuildResult{ID: id, On: on, Failed: "no"})
_ = work.Done()
}
}
var a *natsMachine
strayLanded := make(chan error, 1)
a = MachineOverNATSWith(js, "ace", TheBuildMachine, MachineOptions{Claim: func() bool {
// The moment the long check is claimed: the server hands the other queue's waiting check to A's subscription there, as
// one that arrives after a fetch gave up does. Done once.
select {
case <-strayLanded:
return true
default:
}
other := a.subs[2]
inbox := strings.TrimSuffix(other.sub.Subject, "*") + "stray"
body, _ := json.Marshal(map[string]any{"batch": 1, "expires": int64(30 * time.Second)})
if err := js.Conn().PublishRequest("$JS.API.CONSUMER.MSG.NEXT."+other.worker.Stream+"."+other.worker.Name, inbox, body); err != nil {
strayLanded <- err
return true
}
_ = js.Conn().Flush()
for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(10 * time.Millisecond) {
if n, _, _ := other.sub.Pending(); n > 0 {
break
}
}
strayLanded <- nil
return true
}}).(*natsMachine)
defer a.Close()
go func() { _ = a.Take(ctx, do("ace")) }()
select {
case got := <-started:
if got != "ace:"+long.ID {
t.Fatalf("started %s, want the long check on ace first", got)
}
case <-time.After(15 * time.Second):
t.Fatal("the long check did not start")
}
if err := <-strayLanded; err != nil {
t.Fatal(err)
}
// B, free, runs the stray once the server hands it out again.
b := MachineOverNATS(js, "shanks")
defer b.Close()
go func() { _ = b.Take(ctx, do("shanks")) }()
select {
case got := <-started:
if got != "shanks:"+stray.ID {
t.Fatalf("started %s, want the stray on shanks", got)
}
case <-time.After(20 * time.Second):
t.Fatal("the stray was held by the builder that ran the long check and nobody ran it")
}
// A finishes, and does not run the stray from the subscription it held.
release <- struct{}{}
select {
case got := <-started:
t.Fatalf("%s started after the long check: the stray ran twice", got)
case <-time.After(6 * time.Second):
}
mu.Lock()
defer mu.Unlock()
if len(runs[stray.ID]) != 1 {
t.Fatalf("the stray ran on %v, want once", runs[stray.ID])
}
}
// Review of 1688: a young check pending redelivery (handed back by a Nak) stands ahead of a check older than 30 minutes in the
// same queue while a plan builds. The let-through fetch would return the young one first, so nothing is fetched: the young
// check's delivery count stays 1, it is not handed back every second until it is dead-lettered, and it is not run.
func TestNatsAYoungRedeliveryAheadOfAnOldCheckIsNeverFetchedByTheLetThrough(t *testing.T) {
js, plans := aBusWithTheBuckets(t)
ctx, stop := context.WithCancel(t.Context())
defer stop()
asks := BuildsOn(js, TheBuildMachine)
young := aCheckAskedAgo(time.Minute, 1, false)
if err := asks.Ask(ctx, young); err != nil {
t.Fatal(err)
}
// Someone took it and handed it back: it is pending redelivery.
worker := broker.CheckWorkers()[1]
sub, err := js.Context().PullSubscribe(worker.Filters[0], worker.Name, nats.Bind(worker.Stream, worker.Name), nats.ManualAck())
if err != nil {
t.Fatal(err)
}
got, err := sub.Fetch(1, nats.MaxWait(5*time.Second))
if err != nil || len(got) != 1 {
t.Fatalf("the young check was not fetched to be handed back: %v", err)
}
if err := got[0].Nak(); err != nil {
t.Fatal(err)
}
_ = sub.Unsubscribe()
old := aCheckAskedAgo(45*time.Minute, 2, false)
if err := asks.Ask(ctx, old); err != nil {
t.Fatal(err)
}
if _, err := plans.Put(ctx, "plan-1", []byte(`{"plan":"plan-1"}`)); err != nil {
t.Fatal(err)
}
started := make(chan string, 4)
machine := MachineOverNATS(js, "ace")
defer machine.Close()
go func() {
_ = machine.Take(ctx, func(ctx context.Context, work Build) {
started <- work.Request().ID
_ = work.Announce(ctx, BuildResult{ID: work.Request().ID, On: "ace", Failed: "no"})
_ = work.Done()
})
}()
select {
case id := <-started:
t.Fatalf("%s was run while the young redelivery stood ahead of it", id)
case <-time.After(6 * time.Second):
}
info, err := js.Context().ConsumerInfo(worker.Stream, worker.Name)
if err != nil {
t.Fatal(err)
}
if info.Delivered.Consumer != 1 || info.NumRedelivered != 0 {
t.Fatalf("the young check was handed out %d time(s), %d redelivered: the let-through fetched it", info.Delivered.Consumer, info.NumRedelivered)
}
}
-384
View File
@@ -1,384 +0,0 @@
package link
import (
"context"
"encoding/json"
"errors"
"fmt"
"os"
"strconv"
"strings"
"time"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"github.com/novox/mesh-controller/internal/broker"
)
// Checks wait while a delivery builds (novox/hq ADR 0483, issues 1652, 1687 and 1688).
//
// **The layout other parts read** (ADR 0484 clause 12 has the checker read the same key; a layout shared between modules
// belongs in mesh-sdk, and this one is not there yet):
//
// - bucket `mesh-controller_plans-building` (broker.PlansBuildingBucket): one key per delivery plan in its build phase, named
// by the plan's id (`plan-<digits>`), value a PlanBuilding as JSON. Written by the controller (lease holder) alone: put when
// the plan's first build is asked, put again about every thirty seconds, deleted when the phase ends; the bucket's age
// (two minutes) removes the key of a controller that stopped. A reader lists the keys and needs no value to decide.
// - key `let-through` in bucket `mesh-controller_checks-running` (broker.CheckRunningBucket), value a LetThrough as JSON:
// created by the holder that takes a check which waited LetThroughAfter, refreshed at each heartbeat of that check, deleted
// when it ends. At most one such check runs at a time on the build seat, because the create fails while the key is there.
// LetThroughAfter is how long a check waits, while a delivery builds, before it is let through (ADR 0483 clause 3).
const LetThroughAfter = 30 * time.Minute
// LetThroughKey is the key of the one check let through, in the bucket of the checks running. It has no dot, so it is no
// pull request's key (`<owner>.<repo>.<number>`) and no group's (`group.<name>`).
const LetThroughKey = "let-through"
// PlanBuilding is the value of a plan's key in the bucket of the plans building.
type PlanBuilding struct {
Plan string `json:"plan"`
Since time.Time `json:"since"`
Seen time.Time `json:"seen"`
}
// LetThrough is the value of the let-through key: the builder that runs the check and when it took it.
type LetThrough struct {
Builder string `json:"builder"`
Since time.Time `json:"since"`
}
// checkGate is what a taking loop asks of the bus to hold checks while a delivery builds; a variable of the machine so a test
// hands out against fakes.
type checkGate interface {
// Building is the ids of the plans in their build phase: the keys of the bucket. An error is a bucket that cannot be read.
Building(ctx context.Context) ([]string, error)
// OldestWaiting is when the first waiting ask of a check queue was asked, read without taking it.
OldestWaiting(ctx context.Context, q *takeQueue) (time.Time, bool, error)
// TakeLetThrough creates the let-through key for a builder: false when it is there (another builder runs a check let through).
TakeLetThrough(ctx context.Context, builder string) (bool, error)
// KeepLetThrough puts the key again, and DropLetThrough deletes it.
KeepLetThrough(ctx context.Context, builder string) error
DropLetThrough(ctx context.Context) error
}
// busGate is the gate over a connection.
type busGate struct{ conn *nats.Conn }
func (g busGate) Building(ctx context.Context) ([]string, error) {
return BuildingPlans(ctx, g.conn)
}
// BuildingPlans is the ids of the delivery plans in their build phase, from the controller's bucket. A reader that is not the
// build seat's (the checker, ADR 0484 clause 12) calls this. A bucket that cannot be read is an error; none is none building.
func BuildingPlans(ctx context.Context, conn *nats.Conn) ([]string, error) {
api, err := jetstream.New(conn)
if err != nil {
return nil, err
}
kv, err := api.KeyValue(ctx, broker.PlansBuildingBucket)
if err != nil {
return nil, fmt.Errorf("the plans building: %w", err)
}
keys, err := kv.Keys(ctx)
if errors.Is(err, jetstream.ErrNoKeysFound) {
return nil, nil
}
if err != nil {
return nil, fmt.Errorf("the plans building: %w", err)
}
return keys, nil
}
// OldestWaiting is when the check a fetch of this queue would return was asked, read without taking anything.
//
// **It sees redeliveries** (review of issue 1688). A work queue keeps only what is not acknowledged: what the worker handed out
// (sequence up to its delivered mark) is running or waiting to be handed out again, and a fetch returns those before anything
// new. An ask there that is not said running (a key of the checks running names its build) is such a redelivery. If one is
// younger than the bound, the queue's next fetch would return it, and a young check is never let through: the queue is not a
// candidate, and nothing is fetched, so nothing is handed back and no delivery is spent. Otherwise the candidate is the oldest
// of the redeliveries and the first ask after the delivered mark.
func (g busGate) OldestWaiting(ctx context.Context, q *takeQueue) (time.Time, bool, error) {
info, err := jsOf(g.conn).ConsumerInfo(q.worker.Stream, q.worker.Name, nats.Context(ctx))
if err != nil {
return time.Time{}, false, fmt.Errorf("the %s worker: %w", q.kind, err)
}
var oldest time.Time
note := func(at time.Time) {
if oldest.IsZero() || at.Before(oldest) {
oldest = at
}
}
for seq := uint64(1); seq <= info.Delivered.Stream; {
data, stored, got, err := g.directGet(ctx, q, seq)
if err != nil {
return time.Time{}, false, err
}
if got == 0 || got > info.Delivered.Stream {
break
}
seq = got + 1
var r BuildRequest
if json.Unmarshal(data, &r) != nil {
continue
}
running, ok, err := ReadRunning(ctx, g.conn, r)
if err != nil {
return time.Time{}, false, fmt.Errorf("the checks running: %w", err)
}
if ok && running.Build == r.ID {
continue // running now
}
at := askedAtOf(data, stored)
if time.Since(at) < LetThroughAfter {
return time.Time{}, false, nil // a young redelivery is next in line
}
note(at)
}
data, stored, got, err := g.directGet(ctx, q, info.Delivered.Stream+1)
if err != nil {
return time.Time{}, false, err
}
if got != 0 {
note(askedAtOf(data, stored))
}
return oldest, !oldest.IsZero(), nil
}
// directGet reads the first ask at or after a sequence of a check queue's stream, without taking it: its body, the time the
// stream stored it, and its sequence (0 when there is none).
func (g busGate) directGet(ctx context.Context, q *takeQueue, seq uint64) ([]byte, string, uint64, error) {
body, _ := json.Marshal(map[string]any{"seq": seq, "next_by_subj": q.worker.Filters[0]})
reply, err := g.conn.RequestWithContext(ctx, "$JS.API.DIRECT.GET."+q.worker.Stream, body)
if err != nil {
return nil, "", 0, fmt.Errorf("the asks of the %s queue: %w", q.kind, err)
}
if st := reply.Header.Get("Status"); st == "404" || st == "408" {
return nil, "", 0, nil
}
got, _ := strconv.ParseUint(reply.Header.Get("Nats-Sequence"), 10, 64)
return reply.Data, reply.Header.Get("Nats-Time-Stamp"), got, nil
}
// askedAtOf is when a request was asked: its id says (NewBuildID), the stream's own time otherwise, and a request that says
// neither is taken as asked now, so it is never let through on a guess.
func askedAtOf(data []byte, stored string) time.Time {
var r BuildRequest
if json.Unmarshal(data, &r) == nil {
if at, ok := BuildAskedAt(r.ID); ok {
return at
}
}
if at, err := time.Parse(time.RFC3339Nano, stored); err == nil {
return at
}
return time.Now()
}
func jsOf(conn *nats.Conn) nats.JetStreamContext {
js, _ := conn.JetStream()
return js
}
func (g busGate) TakeLetThrough(ctx context.Context, builder string) (bool, error) {
kv, err := runningBucket(ctx, g.conn)
if err != nil {
return false, err
}
body, _ := json.Marshal(LetThrough{Builder: builder, Since: time.Now().UTC()})
if _, err := kv.Create(ctx, LetThroughKey, body); err != nil {
if errors.Is(err, jetstream.ErrKeyExists) {
return false, nil
}
return false, err
}
return true, nil
}
func (g busGate) KeepLetThrough(ctx context.Context, builder string) error {
kv, err := runningBucket(ctx, g.conn)
if err != nil {
return err
}
body, _ := json.Marshal(LetThrough{Builder: builder, Since: time.Now().UTC()})
_, err = kv.Put(ctx, LetThroughKey, body)
return err
}
func (g busGate) DropLetThrough(ctx context.Context) error {
kv, err := runningBucket(ctx, g.conn)
if err != nil {
return err
}
if err := kv.Delete(ctx, LetThroughKey); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
return err
}
return nil
}
// CheckHold is what a reader of the queue (the controller's own verb) says of why checks wait: the plans building and the check
// let through now, if one.
type CheckHold struct {
Plans []string
LetThrough *LetThrough
}
// ReadCheckHold reads the plans building and the let-through key. A bucket that cannot be read is an error the caller says.
func ReadCheckHold(ctx context.Context, conn *nats.Conn) (CheckHold, error) {
var h CheckHold
plans, err := BuildingPlans(ctx, conn)
if err != nil {
return h, err
}
h.Plans = plans
kv, err := runningBucket(ctx, conn)
if err != nil {
return h, err
}
e, err := kv.Get(ctx, LetThroughKey)
if errors.Is(err, jetstream.ErrKeyNotFound) {
return h, nil
}
if err != nil {
return h, err
}
var l LetThrough
if json.Unmarshal(e.Value(), &l) == nil {
h.LetThrough = &l
}
return h, nil
}
// gateOf is the machine's gate: the one a test set, or the bus's.
func (m *natsMachine) gateOf() checkGate {
if m.gate != nil {
return m.gate
}
return busGate{conn: m.js.Conn()}
}
// sayOnce says a fault of the hold the first time it happens and again once it has recovered and happened anew, so a bucket
// that cannot be read is in the log once and not every second.
func (m *natsMachine) sayOnce(kind string, err error) {
if m.said == nil {
m.said = map[string]bool{}
}
if err == nil {
delete(m.said, kind)
return
}
if !m.said[kind] {
m.said[kind] = true
fmt.Fprintf(os.Stderr, "%s; read as empty, so no check is held for it: %v\n", kind, err)
}
}
// building says whether a delivery builds: any key in the bucket of the plans building. **A bucket that cannot be read is said
// and read as empty** (ADR 0483 clause 2): a check wrongly held costs a pull request its turn, a fault that held every check
// would stop the mesh checking.
func (m *natsMachine) building(ctx context.Context) []string {
plans, err := m.gateOf().Building(ctx)
m.sayOnce("the bucket of the plans building cannot be read", err)
if err != nil {
// empty-on-error: ADR 0483 clause 2 — a bucket that cannot be read is said (sayOnce) and read as empty, because a check wrongly held costs a pull request its turn and a fault that held every check would stop the mesh checking.
return nil
}
return plans
}
// letOneThrough takes the oldest waiting check of the two check queues once it waited LetThroughAfter, by the one let-through
// key: nil when none waited that long, when another holder runs the check let through, or when the check was taken by
// another holder in the moment. The caller owns the key while it returns a message and releases it with dropLetThrough.
func (m *natsMachine) letOneThrough(ctx context.Context) (*nats.Msg, *takeQueue) {
gate := m.gateOf()
var oldest *takeQueue
var oldestAt time.Time
for _, q := range m.subs[1:] {
if !q.bound {
continue
}
at, found, err := gate.OldestWaiting(ctx, q)
m.sayOnce("the first waiting check of the "+q.kind+" queue cannot be read", err)
if err != nil || !found {
continue
}
if oldest == nil || at.Before(oldestAt) {
oldest, oldestAt = q, at
}
}
if oldest == nil || time.Since(oldestAt) < m.letThroughAfter() {
return nil, nil
}
created, err := gate.TakeLetThrough(ctx, m.on)
m.sayOnce("the let-through key cannot be taken", err)
if err != nil || !created {
return nil, nil
}
trying, end := context.WithTimeout(ctx, checkTakeWait)
defer end()
fetched, err := oldest.pull.Fetch(1, nats.Context(trying))
if len(fetched) == 0 {
if err != nil && !errors.Is(err, context.DeadlineExceeded) && !errors.Is(err, nats.ErrTimeout) && !errors.Is(err, context.Canceled) {
m.sayOnce("the "+oldest.kind+" check queue cannot be taken from", err)
}
m.dropLetThrough(ctx)
return nil, nil
}
// Another holder took the one that was oldest in the moment between the read and the fetch: what came is a younger check,
// handed back and the key released, never run during a delivery.
if askedAtOf(fetched[0].Data, "").Add(m.letThroughAfter()).After(time.Now()) {
_ = fetched[0].NakWithDelay(time.Minute) // a delay keeps it off the queue of redeliveries a fetch reads first
m.dropLetThrough(ctx)
return nil, nil
}
m.letKey = true
return fetched[0], oldest
}
func (m *natsMachine) letThroughAfter() time.Duration {
if m.opts.LetThroughAfter > 0 {
return m.opts.LetThroughAfter
}
return LetThroughAfter
}
// dropLetThrough deletes the let-through key this holder created.
func (m *natsMachine) dropLetThrough(ctx context.Context) {
m.letKey = false
if err := m.gateOf().DropLetThrough(context.WithoutCancel(ctx)); err != nil {
fmt.Fprintf(os.Stderr, "could not take the let-through key away; it ages out: %v\n", err)
}
}
// keepLetThrough refreshes the key at every heartbeat while the check it was created for runs, and returns the end of that: it
// stops the refreshing and deletes the key.
func (m *natsMachine) keepLetThrough(ctx context.Context, every time.Duration) (end func()) {
done := make(chan struct{})
finished := make(chan struct{})
go func() {
defer close(finished)
tick := time.NewTicker(every)
defer tick.Stop()
for {
select {
case <-done:
return
case <-tick.C:
putting, stop := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
if err := m.gateOf().KeepLetThrough(putting, m.on); err != nil {
fmt.Fprintf(os.Stderr, "could not keep the let-through key: %v\n", err)
}
stop()
}
}
}()
return func() {
close(done)
<-finished
m.dropLetThrough(ctx)
}
}
// whyHeld words the plans a check is held for, for the log.
func whyHeld(plans []string) string { return strings.Join(plans, ", ") }
-307
View File
@@ -1,307 +0,0 @@
package link
import (
"context"
"encoding/json"
"errors"
"fmt"
"strings"
"sync"
"testing"
"time"
"github.com/nats-io/nats.go"
"github.com/novox/mesh-controller/internal/broker"
)
// Checks wait while a delivery builds (novox/hq ADR 0483, issues 1687 and 1688): the order the taking loop hands out in, against
// fake queues and a fake bus.
// fakeQueue is a queue the loop fetches from: the asks waiting, handed out in order, and how often it was fetched from.
type fakeQueue struct {
mu sync.Mutex
waiting []*nats.Msg
fetched int
}
func (f *fakeQueue) Fetch(int, ...nats.PullOpt) ([]*nats.Msg, error) {
f.mu.Lock()
defer f.mu.Unlock()
f.fetched++
if len(f.waiting) == 0 {
return nil, nats.ErrTimeout
}
m := f.waiting[0]
f.waiting = f.waiting[1:]
return []*nats.Msg{m}, nil
}
func (f *fakeQueue) fetches() int {
f.mu.Lock()
defer f.mu.Unlock()
return f.fetched
}
// askedAgo is a request asked this long ago, as the id says, for a check of this pull request.
func askedAgo(ago time.Duration, number int, ready bool) *nats.Msg {
id := fmt.Sprintf("build-%d", time.Now().Add(-ago).UnixNano())
body, _ := json.Marshal(BuildRequest{ID: id, Repository: "r", Check: &CheckRequest{Owner: "o", Repo: "r", Number: number, Ready: ready}})
return &nats.Msg{Data: body}
}
func buildAsk() *nats.Msg {
body, _ := json.Marshal(BuildRequest{ID: fmt.Sprintf("build-%d", time.Now().UnixNano()), Repository: "r"})
return &nats.Msg{Data: body}
}
// fakeGate is the bus as the loop sees it for the hold.
type fakeGate struct {
mu sync.Mutex
plans []string
plansErr error
letKey bool
taken int
kept int
dropped int
oldestFrom map[*takeQueue]*fakeQueue
}
func (g *fakeGate) Building(context.Context) ([]string, error) {
g.mu.Lock()
defer g.mu.Unlock()
return g.plans, g.plansErr
}
// OldestWaiting reads the first waiting ask of the fake queue without taking it.
func (g *fakeGate) OldestWaiting(_ context.Context, q *takeQueue) (time.Time, bool, error) {
f := g.oldestFrom[q]
f.mu.Lock()
defer f.mu.Unlock()
if len(f.waiting) == 0 {
return time.Time{}, false, nil
}
return askedAtOf(f.waiting[0].Data, ""), true, nil
}
func (g *fakeGate) TakeLetThrough(context.Context, string) (bool, error) {
g.mu.Lock()
defer g.mu.Unlock()
if g.letKey {
return false, nil
}
g.letKey = true
g.taken++
return true, nil
}
func (g *fakeGate) KeepLetThrough(context.Context, string) error {
g.mu.Lock()
defer g.mu.Unlock()
g.kept++
return nil
}
func (g *fakeGate) DropLetThrough(context.Context) error {
g.mu.Lock()
defer g.mu.Unlock()
g.letKey = false
g.dropped++
return nil
}
func (g *fakeGate) setPlans(plans ...string) {
g.mu.Lock()
defer g.mu.Unlock()
g.plans = plans
}
// aMachine is a loop on the build seat with a delivery queue and the two check queues, all fake.
func aMachine() (m *natsMachine, builds, ready, other *fakeQueue, gate *fakeGate) {
builds, ready, other = &fakeQueue{}, &fakeQueue{}, &fakeQueue{}
subs := []*takeQueue{
{kind: queueBuilds, pull: builds, bound: true},
{kind: queueReady, pull: ready, bound: true},
{kind: queueOther, pull: other, bound: true},
}
gate = &fakeGate{oldestFrom: map[*takeQueue]*fakeQueue{subs[1]: ready, subs[2]: other}}
m = &natsMachine{on: "ace", seat: broker.TheBuildSeat, subs: subs, gate: gate}
return m, builds, ready, other, gate
}
func nextAsk(t *testing.T, m *natsMachine) (*nats.Msg, *takeQueue) {
t.Helper()
msg, q, err := m.next(context.Background())
if err != nil {
t.Fatal(err)
}
return msg, q
}
// 1652/AC6, 1687: a waiting check is not taken while a plan key is there, also when the delivery's queue is empty for the moment
// (between two tiers).
func TestNoCheckIsTakenWhileAPlanBuilds(t *testing.T) {
m, builds, ready, other, gate := aMachine()
gate.setPlans("plan-1")
ready.waiting = []*nats.Msg{askedAgo(2*time.Minute, 1, true)}
other.waiting = []*nats.Msg{askedAgo(5*time.Minute, 2, false)}
// The delivery's queue is empty: between two tiers.
if msg, _ := nextAsk(t, m); msg != nil {
t.Fatalf("a check was taken while plan-1 builds: %s", msg.Data)
}
if ready.fetches() != 0 || other.fetches() != 0 {
t.Fatalf("the check queues were fetched from while plan-1 builds: ready %d, other %d", ready.fetches(), other.fetches())
}
if builds.fetches() == 0 {
t.Fatal("the delivery's queue was not looked at")
}
}
// A delivery build is still taken first, and the check waits behind it.
func TestADeliveryBuildIsTakenWhileAPlanBuilds(t *testing.T) {
m, builds, ready, _, gate := aMachine()
gate.setPlans("plan-1")
builds.waiting = []*nats.Msg{buildAsk()}
ready.waiting = []*nats.Msg{askedAgo(2*time.Minute, 1, true)}
msg, q := nextAsk(t, m)
if msg == nil || q.kind != queueBuilds {
t.Fatalf("the delivery build was not taken first: %v %v", msg, q)
}
if ready.fetches() != 0 {
t.Fatal("a check queue was fetched from")
}
}
// It is taken once the key is gone (deleted at the phase's end, or expired).
func TestACheckIsTakenOnceTheKeyIsGone(t *testing.T) {
m, _, ready, _, gate := aMachine()
gate.setPlans("plan-1")
ready.waiting = []*nats.Msg{askedAgo(2*time.Minute, 1, true)}
if msg, _ := nextAsk(t, m); msg != nil {
t.Fatal("taken while the key is there")
}
gate.setPlans()
msg, q := nextAsk(t, m)
if msg == nil || q.kind != queueReady {
t.Fatalf("the check was not taken once the key was gone: %v", q)
}
if m.letKey {
t.Fatal("a check taken without a delivery building took the let-through key")
}
}
// 1687: a bucket that cannot be read is said and read as empty, so the checks are taken.
func TestAnUnreadableBucketHoldsNoCheck(t *testing.T) {
m, _, ready, _, gate := aMachine()
gate.plansErr = errors.New("permissions violation")
gate.plans = []string{"plan-1"}
ready.waiting = []*nats.Msg{askedAgo(2*time.Minute, 1, true)}
msg, q := nextAsk(t, m)
if msg == nil || q.kind != queueReady {
t.Fatal("a check was held because the bucket could not be read")
}
if !m.said["the bucket of the plans building cannot be read"] {
t.Fatalf("the fault was not said: %v", m.said)
}
}
// 1688: a check asked 30 minutes ago is let through while a key is there; the oldest across both queues by ask time, ahead of a
// waiting delivery build; a younger one is never taken before it.
func TestTheOldestCheckPastTheBoundGoesFirstAcrossBothQueues(t *testing.T) {
m, builds, ready, other, gate := aMachine()
gate.setPlans("plan-1")
builds.waiting = []*nats.Msg{buildAsk()}
ready.waiting = []*nats.Msg{askedAgo(31*time.Minute, 1, true)}
other.waiting = []*nats.Msg{askedAgo(50*time.Minute, 2, false)}
msg, q := nextAsk(t, m)
if msg == nil || q.kind != queueOther || !strings.Contains(string(msg.Data), `"number":2`) {
t.Fatalf("the oldest check (50 minutes, other queue) was not let through first: %v", q)
}
if !m.letKey || gate.taken != 1 {
t.Fatalf("the let-through key was not created for it: held %v, created %d", m.letKey, gate.taken)
}
}
// 1688: several checks past the bound go one at a time, oldest first: while the key is there every other holder takes builds.
func TestChecksPastTheBoundGoOneAtATimeOldestFirst(t *testing.T) {
m, builds, ready, other, gate := aMachine()
gate.setPlans("plan-1")
ready.waiting = []*nats.Msg{askedAgo(33*time.Minute, 1, true), askedAgo(31*time.Minute, 3, true)}
other.waiting = []*nats.Msg{askedAgo(40*time.Minute, 2, false)}
var order []string
for range 3 {
msg, q := nextAsk(t, m)
if msg == nil {
t.Fatalf("no check let through after %v", order)
}
var r BuildRequest
_ = json.Unmarshal(msg.Data, &r)
order = append(order, fmt.Sprint(r.Check.Number, q.kind))
// While it runs another holder (the same fake gate, key held) takes builds only.
builds.waiting = []*nats.Msg{buildAsk()}
other2 := &natsMachine{on: "shanks", seat: broker.TheBuildSeat, subs: m.subs, gate: gate}
if next, nq := nextAsk(t, other2); next == nil || nq.kind != queueBuilds {
t.Fatalf("a second holder did not take a delivery build while the let-through key is held: %v", nq)
}
m.dropLetThrough(context.Background())
}
if want := []string{"2other", "1ready", "3ready"}; strings.Join(order, " ") != strings.Join(want, " ") {
t.Fatalf("checks went %v, want %v (oldest first, one at a time)", order, want)
}
}
// 1688: a check younger than the bound is never let through, and no key is created for it.
func TestACheckYoungerThanTheBoundIsHeld(t *testing.T) {
m, _, ready, _, gate := aMachine()
gate.setPlans("plan-1")
ready.waiting = []*nats.Msg{askedAgo(29*time.Minute, 1, true)}
if msg, _ := nextAsk(t, m); msg != nil {
t.Fatal("a check of 29 minutes was let through")
}
if gate.taken != 0 {
t.Fatal("a let-through key was created for a young check")
}
if ready.fetches() != 0 {
t.Fatal("the age was read by a fetch; it must be read without taking")
}
}
// 1688: the holder that lost the race for the key, or the check, takes builds only; and a younger check that arrives in the
// moment between the read and the fetch is handed back with the key released.
func TestAYoungerCheckFetchedInTheRaceIsHandedBackAndTheKeyReleased(t *testing.T) {
m, _, ready, _, gate := aMachine()
gate.setPlans("plan-1")
// The read sees the old check; before the fetch another holder takes it and a young one is first.
gate.oldestFrom[m.subs[1]] = &fakeQueue{waiting: []*nats.Msg{askedAgo(40*time.Minute, 1, true)}}
ready.waiting = []*nats.Msg{askedAgo(time.Minute, 2, true)}
if msg, _ := nextAsk(t, m); msg != nil {
t.Fatalf("a young check was run during a delivery: %s", msg.Data)
}
if gate.letKey || m.letKey {
t.Fatal("the let-through key was left held")
}
}
// The key of a check let through is refreshed while it runs and deleted when it ends.
func TestTheLetThroughKeyLivesAsLongAsTheCheckRuns(t *testing.T) {
m, _, ready, _, gate := aMachine()
gate.setPlans("plan-1")
ready.waiting = []*nats.Msg{askedAgo(40*time.Minute, 1, true)}
if msg, _ := nextAsk(t, m); msg == nil {
t.Fatal("not let through")
}
end := m.keepLetThrough(context.Background(), 20*time.Millisecond)
time.Sleep(120 * time.Millisecond)
gate.mu.Lock()
held, kept := gate.letKey, gate.kept
gate.mu.Unlock()
if !held || kept == 0 {
t.Errorf("while the check ran the key was held %v and refreshed %d times", held, kept)
}
end()
gate.mu.Lock()
defer gate.mu.Unlock()
if gate.letKey || m.letKey {
t.Error("the key was not deleted when the check ended")
}
}
-44
View File
@@ -88,40 +88,6 @@ type Queue struct {
AckPending int `json:"ack-pending"`
MaxDeliver int `json:"max-deliver"`
Asks []QueuedAsk `json:"asks"`
// Hold is why the checks wait: the delivery plans building and the check let through now (ADR 0483, issues 1687 and 1688);
// HoldFault says the bucket could not be read, in which case Hold is empty.
Hold CheckHold `json:"hold"`
HoldFault string `json:"hold-fault,omitempty"`
}
// LetThroughOf says which ask is let through under ADR 0483 clause 3: the check that waited LetThroughAfter or longer and is next
// (the oldest waiting check across the two check queues, while a delivery builds), and the check in flight that was let through
// (the let-through key names the machine that runs it). Empty when none.
func (q Queue) LetThroughOf(now time.Time) (next, running string) {
if len(q.Hold.Plans) == 0 && q.Hold.LetThrough == nil {
return "", ""
}
var oldest *QueuedAsk
for i := range q.Asks {
a := &q.Asks[i]
if a.State != AskWaiting || a.Request.Check == nil || a.AskedAt.IsZero() {
continue
}
if oldest == nil || a.AskedAt.Before(oldest.AskedAt) {
oldest = a
}
}
if oldest != nil && len(q.Hold.Plans) > 0 && now.Sub(oldest.AskedAt) >= LetThroughAfter {
next = oldest.ID
}
if q.Hold.LetThrough != nil {
for _, a := range q.Asks {
if a.State == AskInFlight && a.Request.Check != nil && a.On == q.Hold.LetThrough.Builder {
running = a.ID
}
}
}
return next, running
}
// Of is the asks in one state, in queue order.
@@ -353,16 +319,6 @@ func ReadQueue(ctx context.Context, js *broker.JetStream, seat string) (Queue, e
for _, r := range reads {
q.Asks = append(q.Asks, ClassifyQueue(r.asks, r.delivered, r.ackPending, heard)...)
}
// And why the checks wait, for the build seat that has them (ADR 0483 clause 5). A bucket that cannot be read is said by the
// queue, which then names no plan.
if seat == broker.TheBuildSeat {
hold, err := ReadCheckHold(ctx, js.Conn())
if err != nil {
q.HoldFault = err.Error()
} else {
q.Hold = hold
}
}
return q, nil
}
-54
View File
@@ -1,54 +0,0 @@
package link
import (
"testing"
"time"
"github.com/nats-io/nats.go"
)
// Issue 1747: a builder never holds a message it does not start. A message the server sends after a fetch gave up lands in that
// queue's subscription; while an ask runs, the queues it does not come from are let go of, so such a message is not held
// unheartbeated for the whole run, and the queues are bound again before the next fetch.
func TestARunningAskLetsGoOfTheQueuesItDidNotComeFrom(t *testing.T) {
m, builds, _, other, _ := aMachine()
// A delivery build reached the builds queue's subscription after its fetch gave up: it sits in the buffer.
builds.waiting = []*nats.Msg{buildAsk()}
took := m.subs[1] // the running ask came from the ready queue
m.quiesce(took)
if !took.bound || took.pull == nil {
t.Error("the queue the running ask came from was let go of; its acknowledgements need it")
}
for _, q := range []*takeQueue{m.subs[0], m.subs[2]} {
if q.bound || q.pull != nil || q.sub != nil {
t.Errorf("the %s queue is still held while an ask runs", q.kind)
}
}
// Nothing is fetched from a queue that was let go of: a message in it cannot be taken unheartbeated.
if msg, _, _ := m.next(t.Context()); msg != nil {
t.Fatalf("a message was taken from a queue that was let go of: %s", msg.Data)
}
if builds.fetches() != 0 || other.fetches() != 0 {
t.Errorf("fetched from a queue that was let go of: builds %d, other %d", builds.fetches(), other.fetches())
}
}
// Once the ask is settled the queue it came from is let go of too: a second message the server delivered into its subscription
// on a pull request still open is not run after the ask ends (review of 1747).
func TestTheQueueTheAskCameFromIsLetGoOfWhenTheAskEnds(t *testing.T) {
m, _, ready, _, _ := aMachine()
ready.waiting = []*nats.Msg{askedAgo(time.Minute, 2, true)}
m.quiesce(nil)
for _, q := range m.subs {
if q.bound || q.pull != nil {
t.Errorf("the %s queue is still held after the ask ended", q.kind)
}
}
m.subs[0].bound, m.subs[0].pull = true, &fakeQueue{}
if msg, _, _ := m.next(t.Context()); msg != nil || ready.fetches() != 0 {
t.Fatal("a message held in the kept queue's subscription was taken after the ask ended")
}
}
+1 -1
View File
@@ -9,7 +9,7 @@ a second place to clean besides the catalogue itself.
mesh-catalog e0bc919d4945bb5f13f971f74643a0b714cf45da modules/*/module.json, modules/nats/Dockerfile
mesh-host 9d9e3a46c4f5943cfffb413d962ce48b75e2fd39 examples/foundation-first-node.lock
mesh-sdk 725590e06e75c67b96ce58d015b3056c2d261249 conformance/events/module-event.json
mesh-sdk 07a65927c3620390e191cb6ccc97d3770f284526 conformance/events/module-event.json
To move them, from this repository's root, with the two repositories checked out beside it:
@@ -327,7 +327,7 @@ func settingKeysUsedBy(m Manifest) map[string]bool {
return used
}
// settingKeysAskedOutsideOwners is every key a module's files, units, contributions and served facts
// settingKeysAskedOutsideOwners is every key a module's files, units, items, 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{}
@@ -354,6 +354,10 @@ func settingKeysAskedOutsideOwners(m Manifest) map[string]bool {
}
}
}
// An item's text, filled from its template (WithTemplatesFrom), places a key as a file's does (issue 1658).
for _, it := range m.Items {
note(it.Content)
}
inValues := func(values map[string]any) {
for _, v := range values {
if s, ok := v.(string); ok {
+1 -1
View File
@@ -1,4 +1,4 @@
# git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021153908-725590e06e75
# git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021173436-07a65927c362
## explicit; go 1.26.0
git.novox.be/novox/mesh-sdk/go/asks
git.novox.be/novox/mesh-sdk/go/catalogue