Compare commits
15
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
1f3fdf3111 | ||
|
|
f973cefd7f | ||
|
|
696300b9e2 | ||
|
|
98155e327b | ||
|
|
d6db334840 | ||
|
|
69d016bfdf | ||
|
|
a58b38cee3 | ||
|
|
9d260412b3 | ||
|
|
99610e9569 | ||
|
|
dc06af1f5a | ||
|
|
b62fbeabbf | ||
|
|
373aefd14e | ||
|
|
f66458fe25 | ||
|
|
99da867a98 | ||
|
|
69a67876c2 |
@@ -6,6 +6,8 @@ import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
)
|
||||
|
||||
@@ -73,14 +75,77 @@ func drainedTooLong(now time.Time) error {
|
||||
"within %s (novox/hq ADR 0335)", began.UTC().Format(time.RFC3339), drainTooLong)
|
||||
}
|
||||
|
||||
// inHand is true while this process has claimed a build and not yet released it. The mark in the temporary directory
|
||||
// is only believed while it is: a mark with no build of this process behind it is stale (a crash, a fault in an
|
||||
// earlier path) and is removed, because left alone it hands back every ask for ever (issue 1768).
|
||||
var inHand atomic.Bool
|
||||
|
||||
// markRetry is how long an agent that cannot mark a build in hand takes no work, and markFault when it last failed to.
|
||||
// Taking nothing is the point: an ask fetched and handed back at once, many times a second, spends all its deliveries.
|
||||
const markRetry = 30 * time.Second
|
||||
|
||||
var (
|
||||
markMu sync.Mutex
|
||||
markFault time.Time
|
||||
markSaid string
|
||||
)
|
||||
|
||||
// markFaulty says the agent could not mark a build in hand and has not yet been able to since: it takes no work. Once
|
||||
// markRetry has passed it tries the mark itself, with no ask in hand, so that an ask is fetched only when marking works.
|
||||
func markFaulty(now time.Time) bool {
|
||||
markMu.Lock()
|
||||
failed := markFault
|
||||
markMu.Unlock()
|
||||
if failed.IsZero() {
|
||||
return false
|
||||
}
|
||||
if now.Sub(failed) < markRetry {
|
||||
return true
|
||||
}
|
||||
f, err := os.OpenFile(inHandMark(), os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o644)
|
||||
if err != nil && errors.Is(err, os.ErrExist) && !inHand.Load() {
|
||||
_ = os.Remove(inHandMark())
|
||||
f, err = os.OpenFile(inHandMark(), os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o644)
|
||||
}
|
||||
if err != nil {
|
||||
markFailed(err)
|
||||
return true
|
||||
}
|
||||
_ = f.Close()
|
||||
_ = os.Remove(inHandMark())
|
||||
markMu.Lock()
|
||||
markFault, markSaid = time.Time{}, ""
|
||||
markMu.Unlock()
|
||||
fmt.Fprintf(os.Stderr, "the mark of a build in hand can be made again (%s): taking builds\n", inHandMark())
|
||||
return false
|
||||
}
|
||||
|
||||
// markFailed notes that a build could not be marked in hand, and says it when it is a new reason.
|
||||
func markFailed(err error) {
|
||||
markMu.Lock()
|
||||
defer markMu.Unlock()
|
||||
markFault = time.Now()
|
||||
if said := err.Error(); said != markSaid {
|
||||
markSaid = said
|
||||
fmt.Fprintf(os.Stderr, "cannot mark a build in hand (%s), so this agent takes no build for %s and tries again: %v\n",
|
||||
inHandMark(), markRetry, err)
|
||||
}
|
||||
}
|
||||
|
||||
// claim marks a request in hand, unless the agent is drained: then nothing is in hand, and the request goes
|
||||
// back to the queue.
|
||||
func claim() bool {
|
||||
if !inHand.Load() {
|
||||
// Nothing of this process is in hand, so a mark already there is stale.
|
||||
if err := os.Remove(inHandMark()); err == nil {
|
||||
fmt.Fprintf(os.Stderr, "removed the mark of a build in hand (%s): no build of this agent is in hand\n", inHandMark())
|
||||
}
|
||||
}
|
||||
// Made, never written over: one build at a time, and a mark already there is a build in hand.
|
||||
f, err := os.OpenFile(inHandMark(), os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o644)
|
||||
if err != nil {
|
||||
// Unmarked, an ask would say idle while this build runs: the build is not taken.
|
||||
fmt.Fprintf(os.Stderr, "cannot mark a build in hand (%s), so it is handed back: %v\n", inHandMark(), err)
|
||||
markFailed(err)
|
||||
return false
|
||||
}
|
||||
_, err = f.WriteString(time.Now().UTC().Format(time.RFC3339) + "\n")
|
||||
@@ -89,7 +154,7 @@ func claim() bool {
|
||||
}
|
||||
if err != nil {
|
||||
_ = os.Remove(inHandMark())
|
||||
fmt.Fprintf(os.Stderr, "cannot mark a build in hand (%s), so it is handed back: %v\n", inHandMark(), err)
|
||||
markFailed(err)
|
||||
return false
|
||||
}
|
||||
if drained() {
|
||||
@@ -97,6 +162,7 @@ func claim() bool {
|
||||
fmt.Fprintln(os.Stderr, "asked whether busy, so this agent takes no new build: handed back")
|
||||
return false
|
||||
}
|
||||
inHand.Store(true)
|
||||
return true
|
||||
}
|
||||
|
||||
@@ -107,6 +173,7 @@ func holding(what string) {
|
||||
|
||||
// released ends the build in hand.
|
||||
func released() {
|
||||
inHand.Store(false)
|
||||
if err := os.Remove(inHandMark()); err != nil && !errors.Is(err, os.ErrNotExist) {
|
||||
fmt.Fprintf(os.Stderr, "cannot remove the mark of the build in hand (%s): %v\n", inHandMark(), err)
|
||||
}
|
||||
|
||||
@@ -160,3 +160,64 @@ func TestAnUnknownCommandFails(t *testing.T) {
|
||||
t.Fatalf("help failed: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// 1768/AC2: a mark naming no build of this agent (left by a crash or a fault) is stale: the next ask removes it and is
|
||||
// run, never fetched and handed back in a loop.
|
||||
func TestAStaleInHandMarkIsRemovedAndTheAskRuns(t *testing.T) {
|
||||
t.Setenv("TMPDIR", t.TempDir())
|
||||
inHand.Store(false)
|
||||
if err := undrained(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.WriteFile(inHandMark(), []byte("2026-10-21T18:45:06Z\n"), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !claim() {
|
||||
t.Fatal("a stale in-hand mark made the agent hand back an ask")
|
||||
}
|
||||
// A live one — this agent's own build in hand — is never taken over.
|
||||
if claim() {
|
||||
t.Fatal("a second build was taken while one is in hand")
|
||||
}
|
||||
if _, err := busy(); err != nil {
|
||||
t.Fatalf("the build in hand was lost: %v", err)
|
||||
}
|
||||
released()
|
||||
}
|
||||
|
||||
// 1768/AC3: an agent that cannot mark a build in hand for another reason takes no work, instead of fetching asks and
|
||||
// handing them back; it takes work again once marking works.
|
||||
func TestAnAgentThatCannotMarkABuildTakesNoWorkUntilItCan(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
t.Setenv("TMPDIR", dir)
|
||||
inHand.Store(false)
|
||||
markFault, markSaid = time.Time{}, ""
|
||||
t.Cleanup(func() { markFault, markSaid = time.Time{}, "" })
|
||||
if err := undrained(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// The mark's place is a directory that cannot be removed or made over: a directory named as the mark.
|
||||
if err := os.MkdirAll(inHandMark()+"/x", 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if claim() {
|
||||
t.Fatal("an agent that cannot mark a build took it")
|
||||
}
|
||||
if !markFaulty(time.Now()) {
|
||||
t.Fatal("an agent that cannot mark a build still takes work")
|
||||
}
|
||||
// Still failing after the retry time: still takes nothing, and no ask was needed to find out.
|
||||
if !markFaulty(time.Now().Add(2 * markRetry)) {
|
||||
t.Fatal("an agent whose mark still cannot be made takes work")
|
||||
}
|
||||
if err := os.RemoveAll(inHandMark()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if markFaulty(time.Now().Add(4 * markRetry)) {
|
||||
t.Fatal("an agent whose mark can be made again takes no work")
|
||||
}
|
||||
if !claim() {
|
||||
t.Fatal("an agent that can mark takes no build")
|
||||
}
|
||||
released()
|
||||
}
|
||||
|
||||
@@ -184,7 +184,7 @@ func run() error {
|
||||
// Paused or drained, it takes nothing new (novox/hq ADR 0219, ADR 0335): a drained agent finishes the build
|
||||
// in hand and then says idle, and its container is replaced between builds.
|
||||
machine := link.MachineOverNATSWith(js, on, seat, link.MachineOptions{
|
||||
Paused: func() bool { return h.Paused() || drained() },
|
||||
Paused: func() bool { return h.Paused() || drained() || markFaulty(time.Now()) },
|
||||
Taking: func() { taking(seat) },
|
||||
Claim: claim,
|
||||
})
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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.",
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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:
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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) })
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
@@ -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{}
|
||||
|
||||
@@ -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}
|
||||
}
|
||||
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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 })
|
||||
}
|
||||
|
||||
|
||||
@@ -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 " +
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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=
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
@@ -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 " +
|
||||
|
||||
@@ -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
@@ -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" }
|
||||
} }
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,28 @@
|
||||
package link
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
)
|
||||
|
||||
// 1768/AC1: a request dropped as a duplicate of one another builder runs marks nothing in hand: the claim is not made,
|
||||
// so no mark is left behind to fail every later claim.
|
||||
func TestADuplicateDropIsJudgedBeforeTheClaimSoLeavesNoMark(t *testing.T) {
|
||||
claims := 0
|
||||
m := &natsMachine{
|
||||
opts: MachineOptions{Claim: func() bool { claims++; return true }},
|
||||
duplicate: func(context.Context, BuildRequest) string { return "ace is running it" },
|
||||
}
|
||||
if m.admit(context.Background(), &nats.Msg{}, BuildRequest{ID: "build-1"}) {
|
||||
t.Fatal("a duplicate was admitted")
|
||||
}
|
||||
if claims != 0 {
|
||||
t.Fatalf("a dropped duplicate was claimed %d times, which leaves the in-hand mark behind", claims)
|
||||
}
|
||||
m.duplicate = func(context.Context, BuildRequest) string { return "" }
|
||||
if !m.admit(context.Background(), &nats.Msg{}, BuildRequest{ID: "build-2"}) || claims != 1 {
|
||||
t.Fatalf("a request that is no duplicate was not claimed once: claims=%d", claims)
|
||||
}
|
||||
}
|
||||
+74
-168
@@ -6,7 +6,6 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
@@ -163,21 +162,13 @@ 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
|
||||
// duplicate stands in for runningElsewhere in tests; nil is the real one.
|
||||
duplicate func(context.Context, BuildRequest) string
|
||||
}
|
||||
|
||||
// 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 +176,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 +197,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 +224,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 +246,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 +290,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 +314,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,107 +322,84 @@ 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{}
|
||||
// **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
|
||||
}
|
||||
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
|
||||
}
|
||||
// **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
|
||||
}
|
||||
}
|
||||
if !m.admit(ctx, msg, request) {
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
// 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
|
||||
}
|
||||
// admit decides whether a fetched request is run here, and answers the message when it is not.
|
||||
//
|
||||
// **The duplicate is judged before the request is claimed** (issue 1768): a claim marks the build in hand, and a request
|
||||
// dropped after it left the mark behind, so every later claim failed and every ask was handed back at once until all
|
||||
// had spent their deliveries. A request dropped here has marked nothing.
|
||||
//
|
||||
// **Drained in the moment it was fetched** (novox/hq ADR 0335): handed back at once, as a pause landing then is.
|
||||
func (m *natsMachine) admit(ctx context.Context, msg *nats.Msg, request BuildRequest) bool {
|
||||
// **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 != "" {
|
||||
elsewhere := m.runningElsewhere
|
||||
if m.duplicate != nil {
|
||||
elsewhere = m.duplicate
|
||||
}
|
||||
if why := elsewhere(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
|
||||
return false
|
||||
}
|
||||
// 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 {
|
||||
continue
|
||||
}
|
||||
if q.sub != nil {
|
||||
_ = q.sub.Unsubscribe()
|
||||
}
|
||||
q.sub, q.pull, q.bound = nil, nil, false
|
||||
if m.opts.Claim != nil && !m.opts.Claim() {
|
||||
_ = msg.Nak()
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// runningElsewhere says why an ask need not run here, or nothing: the bucket of the checks running says another builder
|
||||
@@ -496,39 +438,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 +472,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 +637,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
|
||||
}
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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, ", ") }
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
Vendored
+1
-1
@@ -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 {
|
||||
|
||||
Vendored
+1
-1
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user