Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b1546ed28a | ||
|
|
56d6313262 | ||
|
|
7f3ff32361 | ||
|
|
21ac8fd858 | ||
|
|
2819915e8e | ||
|
|
7892019d9e | ||
|
|
e63af0df17 | ||
|
|
961f2e82ea | ||
|
|
e727589e48 | ||
|
|
2f157148c6 | ||
|
|
13be75eb65 | ||
|
|
13711fb5b9 | ||
|
|
fc573f951a | ||
|
|
1ded7dff34 | ||
|
|
50c55a773e | ||
|
|
c6e234455f | ||
|
|
45a1783ba9 | ||
|
|
b538397055 | ||
|
|
dd10889209 | ||
|
|
286257d802 | ||
|
|
67d54d9d09 | ||
|
|
7953d2889d |
@@ -0,0 +1,101 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// **The build agent says whether it is busy, and being asked drains it** (novox/hq ADR 0335 §1, §5). Its
|
||||
// container declares `busy` as `mesh-builder busy`: the node-engine runs it inside the container before it
|
||||
// replaces it. It exits 0 while a build is in hand and 1 when none is; and from the first ask the agent takes
|
||||
// no new build request, so its answer ends once the build in hand ends — a replaced builder finishes the check
|
||||
// it runs, instead of the build queue handing it out again from the start (issue 594).
|
||||
//
|
||||
// Two marks in the container's own temporary directory, as the ready mark (ready.go): a new container starts
|
||||
// without them, and a restarted one removes them before it takes work again.
|
||||
//
|
||||
// - in hand: written by the agent when it claims a request, before it reads whether it is drained; removed
|
||||
// when the build ends.
|
||||
// - drained: written by the ask, before it reads whether a build is in hand.
|
||||
//
|
||||
// Each side writes its own mark before it reads the other's, so an ask racing a fetch sees the build in hand,
|
||||
// or the agent sees the drain and hands the request back; never neither.
|
||||
|
||||
func inHandMark() string { return filepath.Join(os.TempDir(), "mesh-builder.in-hand") }
|
||||
func drainMark() string { return filepath.Join(os.TempDir(), "mesh-builder.drained") }
|
||||
|
||||
// drained says the agent was asked whether it is busy: it takes no new build request.
|
||||
func drained() bool {
|
||||
_, err := os.Stat(drainMark())
|
||||
return err == nil
|
||||
}
|
||||
|
||||
// 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 {
|
||||
// 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)
|
||||
return false
|
||||
}
|
||||
_, err = f.WriteString(time.Now().UTC().Format(time.RFC3339) + "\n")
|
||||
if closeErr := f.Close(); err == nil {
|
||||
err = closeErr
|
||||
}
|
||||
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)
|
||||
return false
|
||||
}
|
||||
if drained() {
|
||||
_ = os.Remove(inHandMark())
|
||||
fmt.Fprintln(os.Stderr, "asked whether busy, so this agent takes no new build: handed back")
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// holding says what is in hand, for the ask to print: written once the request is read.
|
||||
func holding(what string) {
|
||||
_ = os.WriteFile(inHandMark(), []byte(time.Now().UTC().Format(time.RFC3339)+" "+what+"\n"), 0o644)
|
||||
}
|
||||
|
||||
// released ends the build in hand.
|
||||
func released() {
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
// undrained clears both marks at the start of a holder: a container started again is asked afresh.
|
||||
func undrained() error {
|
||||
for _, mark := range []string{drainMark(), inHandMark()} {
|
||||
if err := os.Remove(mark); err != nil && !errors.Is(err, os.ErrNotExist) {
|
||||
return fmt.Errorf("cannot remove %s: %w", mark, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// busy is the ask: it drains the agent, then says whether a build is in hand. Nil — exit 0 — while one is.
|
||||
func busy() (string, error) {
|
||||
if err := os.WriteFile(drainMark(), []byte(time.Now().UTC().Format(time.RFC3339)+"\n"), 0o644); err != nil {
|
||||
// Not drained, it could take the next build: said, and answered busy only for what is in hand now.
|
||||
fmt.Fprintf(os.Stderr, "cannot mark this agent drained (%s): %v\n", drainMark(), err)
|
||||
}
|
||||
held, err := os.ReadFile(inHandMark())
|
||||
if err != nil {
|
||||
return "", errors.New("idle: no build in hand, and this agent takes no new one")
|
||||
}
|
||||
since, what, _ := strings.Cut(strings.TrimSpace(string(held)), " ")
|
||||
if what == "" {
|
||||
what = "a build"
|
||||
}
|
||||
return fmt.Sprintf("busy: %s, in hand since %s; taking no new build", what, since), nil
|
||||
}
|
||||
@@ -0,0 +1,77 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// `mesh-builder busy` answers 0 while a build is in hand and non-zero when idle, and after its first ask the
|
||||
// agent takes no new build request (novox/hq ADR 0335 §5).
|
||||
func TestBusyAnswersWhileABuildIsInHandAndDrainsTheAgent(t *testing.T) {
|
||||
t.Setenv("TMPDIR", t.TempDir())
|
||||
if err := undrained(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if drained() {
|
||||
t.Fatal("an agent never asked is drained")
|
||||
}
|
||||
// A build is claimed and runs.
|
||||
if !claim() {
|
||||
t.Fatal("an agent never asked refused a build")
|
||||
}
|
||||
holding("novox/mesh-controller (check-280)")
|
||||
said, err := busy()
|
||||
if err != nil || !strings.Contains(said, "novox/mesh-controller (check-280)") {
|
||||
t.Fatalf("an agent with a build in hand did not answer busy with what it holds: %q %v", said, err)
|
||||
}
|
||||
// From that ask on it takes nothing new: the fetch is not made, and one fetched is handed back.
|
||||
if !drained() {
|
||||
t.Fatal("the ask did not drain the agent")
|
||||
}
|
||||
if claim() {
|
||||
t.Fatal("a drained agent took a new build")
|
||||
}
|
||||
// The build in hand is still in hand after the refused claim, and busy goes on saying so.
|
||||
if _, err := busy(); err != nil {
|
||||
t.Fatalf("a refused claim ended the build in hand: %v", err)
|
||||
}
|
||||
// It ends: idle, and still drained.
|
||||
released()
|
||||
if _, err := busy(); err == nil {
|
||||
t.Fatal("an agent with nothing in hand answered busy")
|
||||
}
|
||||
if claim() {
|
||||
t.Fatal("a drained agent took a build after its last one ended")
|
||||
}
|
||||
// Started again, it is asked afresh.
|
||||
if err := undrained(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if drained() || !claim() {
|
||||
t.Fatal("a holder started again is still drained")
|
||||
}
|
||||
released()
|
||||
}
|
||||
|
||||
// An ask racing a fetch sees the build in hand, or the agent sees the drain and hands the request back: never
|
||||
// neither, which would replace the container under a build just taken.
|
||||
func TestAnAskRacingAFetchNeverMissesTheBuild(t *testing.T) {
|
||||
t.Setenv("TMPDIR", t.TempDir())
|
||||
for i := 0; i < 200; i++ {
|
||||
if err := undrained(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var claimed bool
|
||||
var askErr error
|
||||
var wg sync.WaitGroup
|
||||
wg.Add(2)
|
||||
go func() { defer wg.Done(); claimed = claim() }()
|
||||
go func() { defer wg.Done(); _, askErr = busy() }()
|
||||
wg.Wait()
|
||||
if claimed && askErr != nil {
|
||||
t.Fatalf("round %d: the agent took a build and the ask said idle", i)
|
||||
}
|
||||
released()
|
||||
}
|
||||
}
|
||||
@@ -66,6 +66,17 @@ broker to take work from or a registry to publish into:
|
||||
|
||||
Without --registry the artifacts stay in this machine's container runtime, named by the
|
||||
digest of their own configuration. The result is printed as JSON.
|
||||
|
||||
mesh-builder ready
|
||||
|
||||
exits 0 once the agent running in this container is bound to its build work and pulls it, and
|
||||
1 before: its container's health check.
|
||||
|
||||
mesh-builder busy
|
||||
|
||||
exits 0 while the agent running in this container has a build in hand, and 1 when it has none;
|
||||
from the first ask the agent takes no new build: its container's busy command, asked before the
|
||||
container is replaced (novox/hq ADR 0335).
|
||||
`
|
||||
|
||||
func run() error {
|
||||
@@ -76,12 +87,30 @@ func run() error {
|
||||
return nil
|
||||
case "build":
|
||||
return buildOnce(context.Background(), os.Args[2:])
|
||||
case "ready":
|
||||
// The container's health check (novox/hq issue 544): exits non-zero until the agent takes work.
|
||||
return ready()
|
||||
case "busy":
|
||||
// The container's busy command (novox/hq ADR 0335): drains the agent, and exits non-zero once
|
||||
// nothing is in hand.
|
||||
said, err := busy()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
fmt.Println(said)
|
||||
return nil
|
||||
default:
|
||||
fmt.Print(usage)
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
if err := notReady(); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := undrained(); err != nil {
|
||||
return err
|
||||
}
|
||||
credential, err := brokerFrom()
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -134,7 +163,13 @@ func run() error {
|
||||
// **What an earlier holder here left goes before anything is taken** (novox/hq issue 306): a holder
|
||||
// recreated mid-check left the check's store and bus running, and the ask went to another machine.
|
||||
h.removeLeftBehind()
|
||||
machine := link.MachineOverNATSWith(js, on, seat, link.MachineOptions{Paused: h.Paused})
|
||||
// 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() },
|
||||
Taking: func() { taking(seat) },
|
||||
Claim: claim,
|
||||
})
|
||||
defer machine.Close()
|
||||
|
||||
// The seat's verbs, on this machine's subjects. Only the seat that declares them: the retired
|
||||
@@ -160,6 +195,8 @@ func run() error {
|
||||
publisher := builder.Registry{Address: registry, Run: builder.Command}
|
||||
|
||||
return machine.Take(ctx, func(ctx context.Context, work link.Build) {
|
||||
defer released()
|
||||
holding(work.Request().Repository + " (" + work.Request().ID + ")")
|
||||
answer(ctx, publisher, on, workspace, work, h)
|
||||
})
|
||||
}
|
||||
|
||||
@@ -0,0 +1,44 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
)
|
||||
|
||||
// **The build agent says it is ready** (novox/hq issue 544). It listens on nothing, so no http or tcp
|
||||
// check can see it work; it writes a mark once it is bound to its worker and pulls build work, and its
|
||||
// container's health check runs `mesh-builder ready`, which passes only while the mark is there. Without
|
||||
// it the node-engine could judge the agent only alive, and every walk that moved it waited out the whole
|
||||
// grace on each machine.
|
||||
//
|
||||
// The mark lives in the container's own temporary directory, never in the workspace the machine keeps:
|
||||
// a new container starts without it, and a restarted one removes it before it binds again.
|
||||
func readyMark() string { return filepath.Join(os.TempDir(), "mesh-builder.ready") }
|
||||
|
||||
// notReady removes the mark: a holder starting is not taking work until it says so again.
|
||||
func notReady() error {
|
||||
if err := os.Remove(readyMark()); err != nil && !errors.Is(err, os.ErrNotExist) {
|
||||
return fmt.Errorf("cannot remove the mark that this agent takes build work (%s): %w", readyMark(), err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// taking writes the mark. A mark that cannot be written is said, and the agent goes on building: its
|
||||
// check then says it is not ready, which is the truth the gate should hear.
|
||||
func taking(seat string) {
|
||||
if err := os.WriteFile(readyMark(), []byte(seat+"\n"), 0o644); err != nil {
|
||||
fmt.Fprintf(os.Stderr, "cannot mark that this agent takes build work (%s): %v\n", readyMark(), err)
|
||||
return
|
||||
}
|
||||
fmt.Fprintf(os.Stderr, "taking build work from %s\n", seat)
|
||||
}
|
||||
|
||||
// ready is the health check: nil when this container's agent is bound and pulling build work.
|
||||
func ready() error {
|
||||
if _, err := os.Stat(readyMark()); err != nil {
|
||||
return fmt.Errorf("not taking build work yet: %s is not there", readyMark())
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,32 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"os"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// The health check passes only between the agent saying it takes work and its next start (novox/hq
|
||||
// issue 544).
|
||||
func TestTheAgentIsReadyOnlyOnceItTakesWork(t *testing.T) {
|
||||
t.Setenv("TMPDIR", t.TempDir())
|
||||
if err := ready(); err == nil {
|
||||
t.Fatal("an agent that never took work is ready")
|
||||
}
|
||||
taking("mesh-build-machine")
|
||||
if err := ready(); err != nil {
|
||||
t.Fatalf("an agent taking work is not ready: %v", err)
|
||||
}
|
||||
// Started again, it is not ready until it binds again.
|
||||
if err := notReady(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := ready(); err == nil {
|
||||
t.Fatal("an agent starting again is ready before it takes work")
|
||||
}
|
||||
if err := notReady(); err != nil {
|
||||
t.Fatalf("removing a mark that is not there failed: %v", err)
|
||||
}
|
||||
if _, err := os.Stat(readyMark()); err == nil {
|
||||
t.Fatal("the mark is still there")
|
||||
}
|
||||
}
|
||||
@@ -16,6 +16,7 @@ func TestBuilderDiagnosticsStayOffStdout(t *testing.T) {
|
||||
"version)": true, // --version
|
||||
`"stopping")`: true, // the loop.s shutdown line
|
||||
"usage)": true, // --help text, for a human
|
||||
"said)": true, // busy: the answer its node-engine reads (novox/hq ADR 0335)
|
||||
}
|
||||
for _, file := range []string{"once.go", "main.go"} {
|
||||
src, err := os.ReadFile(file)
|
||||
|
||||
@@ -64,6 +64,11 @@ type actor struct {
|
||||
// the highest epoch issued, and what was said of it; zero when it was not (S12).
|
||||
reset time.Time
|
||||
resetSaid string
|
||||
// renewEvery and poll, when set, are how often the serving controller renews the lease it holds and
|
||||
// looks again at one another holds, in place of the lease's own: a test's, so two controllers hand the
|
||||
// lease over in a second rather than in a renewal of five (novox/hq issue 591). A renewal is given
|
||||
// the lease's own RenewEvery to answer however often it is made, so a slow bus does not lose it.
|
||||
renewEvery, poll time.Duration
|
||||
}
|
||||
|
||||
// theLease is this process's standing to the lease.
|
||||
@@ -223,7 +228,7 @@ func (a *actor) serveUnderTheLease(ctx context.Context, inv *inventory.Inventory
|
||||
}
|
||||
say := func(format string, args ...any) { fmt.Printf(format+"\n", args...) }
|
||||
l, err := lease.Open(ctx, api, broker.LeaseBucket, lease.Options{Holder: holderOf(instance),
|
||||
Floor: inv.HighestEpoch, Say: say, Health: controllerHealth, Moved: func(was, floor uint64) {
|
||||
RenewEvery: a.renewEvery, RenewWithin: lease.RenewEvery, Poll: a.poll, Floor: inv.HighestEpoch, Say: say, Health: controllerHealth, Moved: func(was, floor uint64) {
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
a.reset = time.Now()
|
||||
|
||||
@@ -0,0 +1,84 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/inventory"
|
||||
)
|
||||
|
||||
// A first-node gate waits for a busy container (novox/hq ADR 0335 §4).
|
||||
//
|
||||
// When a machine reports, on what the gate sent it, that a container of a judged module was left running
|
||||
// because it said busy, the judging has not started: the build is not yet what runs there. The walk says it
|
||||
// waits for a busy container on that machine, counts no pass and no bound, and starts the judging — its
|
||||
// Since — at the report that no longer lists the container waiting. The wait is the node-engine's, bounded
|
||||
// at 15 minutes; the controller's own bound (busyGateBound past the engine's) is for an engine that stopped
|
||||
// reporting while it waited, so no walk waits for ever. An engine older than the field never lists one, and
|
||||
// its gate starts as before.
|
||||
|
||||
// busyGateSlack is how long past the engine's bound the controller still waits for the report that says the
|
||||
// container replaced: a pass of the engine (its asks every 20 seconds) and the report's way here.
|
||||
var busyGateSlack = 5 * time.Minute
|
||||
|
||||
// busyWait decides, before a judging, whether the gate waits for a busy container on a machine it judges:
|
||||
// true while it waits, the gate saying so. False when nothing waits, or no longer: a wait that ended moves the
|
||||
// gate's Since to its end. Pure but for the gate it is handed.
|
||||
func busyWait(g *inventory.PlanGate, pairs []judged, f gateFacts, now time.Time) bool {
|
||||
if g.Busy != nil && g.Busy.Ended != nil {
|
||||
return false
|
||||
}
|
||||
for _, j := range pairs {
|
||||
b, waits := busyOn(j, f)
|
||||
if !waits {
|
||||
continue
|
||||
}
|
||||
if g.Busy == nil {
|
||||
g.Busy = &inventory.GateBusy{Machine: j.node, Module: j.module, Container: b.Container, Since: b.Since}
|
||||
}
|
||||
g.Busy.Until, g.Busy.Said = b.Until, b.Said
|
||||
if now.After(g.Busy.Until.Add(busyGateSlack)) {
|
||||
// The engine said it would have replaced it by now: the judging starts, and finds the module not
|
||||
// yet healthy while the machine still lists it waiting (busyStill), until the gate's own bound.
|
||||
end := now
|
||||
g.Busy.Ended, g.Busy.PastBound = &end, true
|
||||
g.Since, g.Passes, g.LastPass, g.Healthy, g.HealthyAt = &end, 0, nil, nil, nil
|
||||
return false
|
||||
}
|
||||
g.Passes, g.LastPass, g.Last, g.Failing = 0, nil, g.Busy.Words(), nil
|
||||
g.Read(now, false, g.Busy.Words())
|
||||
return true
|
||||
}
|
||||
if g.Busy == nil {
|
||||
return false
|
||||
}
|
||||
// Replaced: the judging starts at the report that said so.
|
||||
end := now
|
||||
if r, said := f.reports[g.Busy.Machine]; said && r.At != nil && r.At.After(g.Busy.Since) && r.At.Before(now) {
|
||||
end = r.At.UTC()
|
||||
}
|
||||
g.Busy.Ended = &end
|
||||
g.Since, g.Passes, g.LastPass, g.Last, g.Healthy, g.HealthyAt = &end, 0, nil, "", nil, nil
|
||||
return false
|
||||
}
|
||||
|
||||
// busyOn is the container of a judged module its machine last reported, on what the gate sent it, as left
|
||||
// running because it said busy.
|
||||
func busyOn(j judged, f gateFacts) (inventory.BusyWait, bool) {
|
||||
r, said := f.reports[j.node]
|
||||
if !said || r.At == nil || !f.reportedOn(j.node, r) {
|
||||
return inventory.BusyWait{}, false
|
||||
}
|
||||
return inventory.WaitingFor(r.Busy, j.module)
|
||||
}
|
||||
|
||||
// busyStill is a judged module whose container its machine still lists waiting once the gate stopped waiting
|
||||
// for it: not yet what was sent, so not healthy, whatever its health says of the container still running.
|
||||
func busyStill(j judged, f gateFacts) (string, bool) {
|
||||
b, waits := busyOn(j, f)
|
||||
if !waits {
|
||||
return "", false
|
||||
}
|
||||
return fmt.Sprintf("%s still runs its busy container %s, past its node-engine's bound of %s",
|
||||
j.node, b.Container, b.Until.UTC().Format(time.RFC3339)), true
|
||||
}
|
||||
@@ -0,0 +1,111 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/inventory"
|
||||
)
|
||||
|
||||
// A walk whose module's container waits as busy on its first machine says so, and starts that machine's
|
||||
// first-machine gate only once the machine reports the container replaced (novox/hq ADR 0335 §4).
|
||||
func TestAGateWaitsForABusyContainerAndStartsWhenItIsReplaced(t *testing.T) {
|
||||
sent := time.Date(2026, 10, 13, 1, 0, 0, 0, time.UTC)
|
||||
g := &inventory.PlanGate{Machines: []string{"novox"}, Since: &sent,
|
||||
Sent: map[string]inventory.SentDeclaration{"novox": {Digest: "d1", Sequence: 7}}}
|
||||
pairs := []judged{{module: "build-agent", node: "novox"}, {module: "mesh-controller", node: "novox"}}
|
||||
reported := sent.Add(20 * time.Second)
|
||||
waiting := []inventory.BusyWait{{ID: "build-agent.server", Container: "mesh-build-agent",
|
||||
State: inventory.BusyWaiting, Since: reported, Until: reported.Add(15 * time.Minute),
|
||||
Said: "busy: novox/mesh-controller (check of #280), in hand since 00:58:00Z"}}
|
||||
facts := func(at time.Time, busy []inventory.BusyWait) gateFacts {
|
||||
// As judgeMoves hands it: the report held against the gate's own send.
|
||||
return gateFacts{sent: g.Sent, reports: map[string]inventory.Reported{"novox": {Node: "novox",
|
||||
Outcome: inventory.OutcomeApplied, At: &at, Declared: "d1", ReportedSequence: 7, Busy: busy}}}
|
||||
}
|
||||
|
||||
now := reported.Add(time.Minute)
|
||||
if !busyWait(g, pairs, facts(reported, waiting), now) {
|
||||
t.Fatal("a gate whose module's container waits as busy judged it")
|
||||
}
|
||||
if !g.Since.Equal(sent) || g.Busy == nil || g.Busy.Machine != "novox" || g.Busy.Ended != nil {
|
||||
t.Fatalf("the gate does not keep the wait: %+v %+v", g.Since, g.Busy)
|
||||
}
|
||||
for _, said := range []string{gateLine(g), g.Last} {
|
||||
if !strings.Contains(said, "waits for a busy container on novox") {
|
||||
t.Errorf("the walk does not say it waits for a busy container there: %s", said)
|
||||
}
|
||||
}
|
||||
if !strings.Contains(g.Last, "check of #280") {
|
||||
t.Errorf("the wait does not say what the container holds: %s", g.Last)
|
||||
}
|
||||
|
||||
// Ten minutes on, still busy, within the engine's bound: still waiting, nothing judged, no bound run.
|
||||
now = reported.Add(10 * time.Minute)
|
||||
if !busyWait(g, pairs, facts(reported.Add(9*time.Minute), waiting), now) || g.Passes != 0 {
|
||||
t.Fatalf("the gate stopped waiting within the bound: %+v", g)
|
||||
}
|
||||
|
||||
// The machine reports it replaced: the judging starts at that report.
|
||||
replacedAt := reported.Add(12 * time.Minute)
|
||||
now = replacedAt.Add(5 * time.Second)
|
||||
replaced := []inventory.BusyWait{{ID: "build-agent.server", Container: "mesh-build-agent",
|
||||
State: "replaced-when-idle", Since: reported, Until: reported.Add(15 * time.Minute)}}
|
||||
if busyWait(g, pairs, facts(replacedAt, replaced), now) {
|
||||
t.Fatal("the gate still waits after the container was replaced")
|
||||
}
|
||||
if g.Since == nil || !g.Since.Equal(replacedAt) || g.Busy.Ended == nil || !g.Busy.Ended.Equal(replacedAt) {
|
||||
t.Fatalf("the gate does not start at the report that said it replaced: since %v, %+v", g.Since, g.Busy)
|
||||
}
|
||||
if strings.Contains(gateLine(g), "waits for a busy container") {
|
||||
t.Errorf("a gate judging still says it waits: %s", gateLine(g))
|
||||
}
|
||||
// And a later report listing it waiting again does not hold a gate that started.
|
||||
if busyWait(g, pairs, facts(now, waiting), now.Add(time.Minute)) {
|
||||
t.Error("a gate that started waits again")
|
||||
}
|
||||
}
|
||||
|
||||
// An engine that stops reporting while it waits holds no walk for ever: past the engine's bound and the slack,
|
||||
// the judging starts, and finds the module not yet healthy while its container is still listed waiting.
|
||||
func TestAGateWaitsNoLongerThanTheEnginesBound(t *testing.T) {
|
||||
sent := time.Date(2026, 10, 13, 1, 0, 0, 0, time.UTC)
|
||||
g := &inventory.PlanGate{Machines: []string{"novox"}, Since: &sent}
|
||||
pairs := []judged{{module: "build-agent", node: "novox"}}
|
||||
waiting := []inventory.BusyWait{{ID: "build-agent.server", Container: "mesh-build-agent",
|
||||
State: inventory.BusyWaiting, Since: sent, Until: sent.Add(15 * time.Minute)}}
|
||||
f := gateFacts{reports: map[string]inventory.Reported{"novox": {Node: "novox", Outcome: inventory.OutcomeApplied,
|
||||
At: &sent, Current: true, Busy: waiting}}}
|
||||
now := sent.Add(15*time.Minute + busyGateSlack + time.Second)
|
||||
if busyWait(g, pairs, f, now) {
|
||||
t.Fatal("a gate waited past the engine's bound and the slack")
|
||||
}
|
||||
if !g.Busy.PastBound || !g.Since.Equal(now) {
|
||||
t.Fatalf("the gate does not start, said past the bound: %+v %v", g.Busy, g.Since)
|
||||
}
|
||||
if why, still := busyStill(pairs[0], f); !still || !strings.Contains(why, "past its node-engine's bound") {
|
||||
t.Fatalf("a module still running its busy container reads as what was sent: %q", why)
|
||||
}
|
||||
}
|
||||
|
||||
// A report from an engine older than the field lists nothing: the gate starts at the send, as before; and a
|
||||
// report not about the gate's send says nothing of it.
|
||||
func TestAGateOfAnEngineThatNeverWaitsStartsAsBefore(t *testing.T) {
|
||||
sent := time.Date(2026, 10, 13, 1, 0, 0, 0, time.UTC)
|
||||
g := &inventory.PlanGate{Machines: []string{"ace"}, Since: &sent,
|
||||
Sent: map[string]inventory.SentDeclaration{"ace": {Digest: "d2", Sequence: 9}}}
|
||||
pairs := []judged{{module: "build-agent", node: "ace"}}
|
||||
at := sent.Add(time.Minute)
|
||||
older := gateFacts{sent: g.Sent, reports: map[string]inventory.Reported{"ace": {Node: "ace", Outcome: inventory.OutcomeApplied,
|
||||
At: &at, Declared: "d2", ReportedSequence: 9}}}
|
||||
if busyWait(g, pairs, older, at) || g.Busy != nil || !g.Since.Equal(sent) {
|
||||
t.Fatalf("a gate on an engine that never waits did not start at the send: %+v %v", g.Busy, g.Since)
|
||||
}
|
||||
elsewhere := gateFacts{sent: g.Sent, reports: map[string]inventory.Reported{"ace": {Node: "ace", Outcome: inventory.OutcomeApplied,
|
||||
At: &at, Declared: "d1", ReportedSequence: 8, Busy: []inventory.BusyWait{{ID: "build-agent.server",
|
||||
State: inventory.BusyWaiting, Until: at.Add(time.Hour)}}}}}
|
||||
if busyWait(g, pairs, elsewhere, at) {
|
||||
t.Fatal("a wait reported about an earlier declaration held the gate")
|
||||
}
|
||||
}
|
||||
@@ -1042,13 +1042,13 @@ func deliveryCommand(ctx context.Context, args []string) error {
|
||||
return fmt.Errorf("delivery %s: plan, order, check, go, stop or walks", sub)
|
||||
}
|
||||
|
||||
// appliedFrom answers a machine's first report after a send from the controller's `apply` durations; a lookup
|
||||
// that fails is a report not read, which leaves the walk's end unknown rather than wrong.
|
||||
// appliedFrom answers a machine's own report of the send a walk made to it from the controller's `apply` durations
|
||||
// (novox/hq issue 582); a lookup that fails is a report not read, which leaves the walk's end unknown rather than wrong.
|
||||
func appliedFrom(ctx context.Context, inv *inventory.Inventory) inventory.AppliedLookup {
|
||||
return func(node string, sent time.Time) (inventory.AppliedReport, bool) {
|
||||
r, ok, err := inv.FirstAppliedAfter(ctx, node, sent)
|
||||
return func(node string, send inventory.SentDeclaration, walkSent time.Time) (inventory.AppliedReport, bool) {
|
||||
r, ok, err := inv.FirstAppliedAfter(ctx, node, send, walkSent)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "the report of %s after %s could not be read: %v\n", node, sent.Format(time.RFC3339), err)
|
||||
fmt.Fprintf(os.Stderr, "the report of %s after %s could not be read: %v\n", node, walkSent.Format(time.RFC3339), err)
|
||||
return inventory.AppliedReport{}, false
|
||||
}
|
||||
return r, ok
|
||||
|
||||
@@ -544,6 +544,11 @@ func judgeMoves(ctx context.Context, open *stores, g *inventory.PlanGate, pairs
|
||||
facts.sentBuilds[j.node] = builds
|
||||
}
|
||||
}
|
||||
// **A busy container on a judged machine holds the judging before it starts** (novox/hq ADR 0335 §4): the
|
||||
// walk says it waits there, and its gate starts at the report that says the container replaced.
|
||||
if busyWait(g, pairs, facts, now) {
|
||||
return "", nil
|
||||
}
|
||||
// **What is wrong with a machine itself is the machine's** (novox/hq issue 281): read once for each
|
||||
// machine judged, apart from what is wrong with a module there, and never pinned on the module the
|
||||
// gate happens to be kept on.
|
||||
@@ -568,6 +573,9 @@ func judgeMoves(ctx context.Context, open *stores, g *inventory.PlanGate, pairs
|
||||
for _, j := range pairs {
|
||||
w := words[j.node]
|
||||
h, said := judgeHealth(j.module, coreComponent(j.module), shelf[j.module], j.node, *g.Since, w.facts)
|
||||
if why, still := busyStill(j, facts); still && (h == healthGood || h == healthPerson) {
|
||||
h, said = healthNotYet, why
|
||||
}
|
||||
if h == healthGood || h == healthPerson {
|
||||
if on, named := w.on[j.module]; named {
|
||||
h, said = healthNotYet, on
|
||||
@@ -1372,6 +1380,9 @@ func gateLine(g *inventory.PlanGate) string {
|
||||
case inventory.GateFailed:
|
||||
what = "FAILED"
|
||||
}
|
||||
if g.Verdict == "" && g.Waiting() {
|
||||
what = "waits for a busy container on " + g.Busy.Machine
|
||||
}
|
||||
line := fmt.Sprintf("gate on %s: %s", strings.Join(g.Machines, ", "), what)
|
||||
if g.Component != "" {
|
||||
line += " (core: " + g.Component + ")"
|
||||
|
||||
@@ -55,8 +55,14 @@ func TestTwoControllersOneActs(t *testing.T) {
|
||||
inv := inventory.ForTest(t)
|
||||
ctx := t.Context()
|
||||
|
||||
// Each renews five times and looks again ten times as often as a serving controller does, so the
|
||||
// handover and the loss below are seen in a second, not in a renewal of five (novox/hq issue 591): what
|
||||
// is checked is the order of events, which does not change with the interval. Each renewal still has a
|
||||
// serving controller's five seconds to answer (acting.go), so a slow bus does not lose A the lease.
|
||||
const renewEvery, poll = lease.RenewEvery / 5, lease.Poll / 10
|
||||
|
||||
// Controller A takes the lease.
|
||||
a := &actor{}
|
||||
a := &actor{renewEvery: renewEvery, poll: poll}
|
||||
aCtx, stopA := context.WithCancel(ctx)
|
||||
defer stopA()
|
||||
lostA, err := a.serveUnderTheLease(aCtx, inv, url)
|
||||
@@ -69,7 +75,7 @@ func TestTwoControllersOneActs(t *testing.T) {
|
||||
}
|
||||
|
||||
// Controller B starts while A holds it, and waits — acting on nothing meanwhile.
|
||||
b := &actor{}
|
||||
b := &actor{renewEvery: renewEvery, poll: poll}
|
||||
bCtx, stopB := context.WithCancel(ctx)
|
||||
defer stopB()
|
||||
tookB := make(chan (<-chan struct{}), 1)
|
||||
@@ -82,10 +88,11 @@ func TestTwoControllersOneActs(t *testing.T) {
|
||||
}
|
||||
tookB <- lost
|
||||
}()
|
||||
// B looks fifteen times, across one of A's renewals at least.
|
||||
select {
|
||||
case <-tookB:
|
||||
t.Fatal("B took the lease while A held it")
|
||||
case <-time.After(3 * time.Second):
|
||||
case <-time.After(15 * poll):
|
||||
}
|
||||
if _, err := b.epoch(ctx); !errors.Is(err, lease.ErrNotHeld) {
|
||||
t.Fatalf("B, waiting, may act: %v", err)
|
||||
@@ -125,7 +132,7 @@ func TestTwoControllersOneActs(t *testing.T) {
|
||||
}
|
||||
select {
|
||||
case <-lostB:
|
||||
case <-time.After(2 * lease.RenewEvery):
|
||||
case <-time.After(2 * renewEvery):
|
||||
t.Fatal("B was not told it lost the lease")
|
||||
}
|
||||
if _, err := b.epoch(ctx); !errors.Is(err, lease.ErrNotHeld) {
|
||||
|
||||
@@ -0,0 +1,135 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/inventory"
|
||||
)
|
||||
|
||||
// machineLines is what a machine said it is made of (novox/hq issue 160), as `node` shows it: every fact in
|
||||
// one block with its age, the architecture and kernel the profile always carried among them, and each fact
|
||||
// the machine could not determine said as unknown with its why — never as a zero, and never left out.
|
||||
func machineLines(a inventory.MachineAccount, had bool, now time.Time) []string {
|
||||
if !had {
|
||||
return []string{" what it is made of: never said — this machine has not reported a profile"}
|
||||
}
|
||||
m := a.Machine
|
||||
if m == nil {
|
||||
return []string{
|
||||
" what it is made of: not said — its node-engine is older than the one that says it",
|
||||
" " + factLine("architecture", orUnsaid(a.Architecture)),
|
||||
" " + factLine("kernel", orUnsaid(a.Kernel)),
|
||||
}
|
||||
}
|
||||
age := roughly(now.Sub(m.ObservedAt)) + " ago"
|
||||
if m.ObservedAt.IsZero() {
|
||||
// Said with no time: its age is not known, and is never read as two thousand years.
|
||||
age = "at a time it did not say (age unknown)"
|
||||
}
|
||||
out := []string{fmt.Sprintf(" what it is made of, as it said %s:", age)}
|
||||
add := func(name, value string) { out = append(out, " "+factLine(name, value)) }
|
||||
unknown := map[string]string{}
|
||||
for k, v := range m.Unknown {
|
||||
unknown[k] = v
|
||||
}
|
||||
// said is the value, or the fact's own reason it is unknown; a fact neither said nor unknown is a
|
||||
// node-engine that broke its own rule, and is said so rather than shown blank.
|
||||
said := func(fact, value string) string {
|
||||
if value != "" {
|
||||
delete(unknown, fact)
|
||||
return value
|
||||
}
|
||||
if why, ok := unknown[fact]; ok {
|
||||
delete(unknown, fact)
|
||||
return "unknown: " + why
|
||||
}
|
||||
return "unknown: the machine said neither it nor why"
|
||||
}
|
||||
|
||||
add("architecture", orUnsaid(a.Architecture))
|
||||
release := said("kernel release", m.KernelRelease)
|
||||
if m.KernelRelease == "" {
|
||||
release = "(release " + release + ")"
|
||||
}
|
||||
add("kernel", orUnsaid(a.Kernel)+" "+release)
|
||||
dist := said("distribution", m.Distribution)
|
||||
if m.Distribution != "" {
|
||||
dist += " " + said("version", m.Version)
|
||||
}
|
||||
add("distribution", dist)
|
||||
add("processor", said("processor", m.Processor))
|
||||
add("cores", said("cores", countWords(m.Cores)))
|
||||
add("threads", said("threads", countWords(m.Threads)))
|
||||
add("memory", said("memory", bytesWords(m.MemoryBytes)))
|
||||
add("virtual", said("virtual or physical", m.Virtual))
|
||||
up := ""
|
||||
if m.BootedAt != nil {
|
||||
up = fmt.Sprintf("%s, since %s", roughly(now.Sub(*m.BootedAt)), m.BootedAt.Local().Format("2006-01-02 15:04"))
|
||||
}
|
||||
add("up", said("uptime", up))
|
||||
add("timezone", said("timezone", m.Timezone))
|
||||
if len(m.Filesystems) == 0 {
|
||||
add("filesystems", said("filesystems", ""))
|
||||
} else {
|
||||
for i, f := range m.Filesystems {
|
||||
name := ""
|
||||
if i == 0 {
|
||||
name = "filesystems"
|
||||
}
|
||||
add(name, fmt.Sprintf("%s %s, %s free (%s %s)", f.Mount, bytesWords(f.SizeBytes),
|
||||
bytesWords(f.FreeBytes), f.Type, f.Device))
|
||||
}
|
||||
if why, ok := unknown["filesystems"]; ok {
|
||||
add("", "and some unknown: "+why)
|
||||
delete(unknown, "filesystems")
|
||||
}
|
||||
}
|
||||
add("public address", said("public address", ""))
|
||||
// Any other fact a newer node-engine named unknown, so nothing it said is dropped.
|
||||
rest := make([]string, 0, len(unknown))
|
||||
for k := range unknown {
|
||||
rest = append(rest, k)
|
||||
}
|
||||
sort.Strings(rest)
|
||||
for _, k := range rest {
|
||||
add(k, "unknown: "+unknown[k])
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func factLine(name, value string) string { return fmt.Sprintf("%-15s %s", name, value) }
|
||||
|
||||
func orUnsaid(s string) string {
|
||||
if strings.TrimSpace(s) == "" {
|
||||
return "not said"
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
// countWords is a number a machine said, or nothing when it said none (its unknown says why).
|
||||
func countWords(n int) string {
|
||||
if n <= 0 {
|
||||
return ""
|
||||
}
|
||||
return fmt.Sprint(n)
|
||||
}
|
||||
|
||||
// bytesWords is an amount of bytes a person reads, in binary units; nothing when none was said.
|
||||
func bytesWords(n int64) string {
|
||||
if n <= 0 {
|
||||
return ""
|
||||
}
|
||||
units := []string{"B", "KiB", "MiB", "GiB", "TiB", "PiB"}
|
||||
v, u := float64(n), 0
|
||||
for v >= 1024 && u < len(units)-1 {
|
||||
v /= 1024
|
||||
u++
|
||||
}
|
||||
if u == 0 {
|
||||
return fmt.Sprintf("%d B", n)
|
||||
}
|
||||
return fmt.Sprintf("%.1f %s", v, units[u])
|
||||
}
|
||||
@@ -0,0 +1,88 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/inventory"
|
||||
)
|
||||
|
||||
// Defends novox/hq issue 160: `node` shows every fact a machine said it is made of, in one block with its
|
||||
// age, the architecture and kernel among them; and a fact it could not determine as unknown with why.
|
||||
func TestNodeShowsWhatAMachineIsMadeOfWithItsAge(t *testing.T) {
|
||||
now := time.Date(2026, 10, 12, 15, 0, 0, 0, time.UTC)
|
||||
booted := now.Add(-75 * time.Hour)
|
||||
a := inventory.MachineAccount{Architecture: "amd64", Kernel: "linux", Machine: &inventory.MadeOf{
|
||||
ObservedAt: now.Add(-3 * time.Hour), KernelRelease: "7.2.8-arch1-2", Distribution: "Arch Linux",
|
||||
Version: "rolling", Processor: "AMD Ryzen 9 8945HS", Cores: 8, Threads: 16, MemoryBytes: 32891494400,
|
||||
Virtual: "physical", BootedAt: &booted, Timezone: "Europe/Brussels",
|
||||
Filesystems: []inventory.MountedFilesystem{{Mount: "/", Device: "/dev/nvme0n1p3", Type: "ext4",
|
||||
SizeBytes: 987978522624, FreeBytes: 31830110208}},
|
||||
Unknown: map[string]string{"public address": "how it learns it is not decided yet"},
|
||||
}}
|
||||
got := strings.Join(machineLines(a, true, now), "\n")
|
||||
t.Log("\n" + got)
|
||||
for _, want := range []string{
|
||||
"what it is made of, as it said 3h ago:",
|
||||
"architecture amd64",
|
||||
"kernel linux 7.2.8-arch1-2",
|
||||
"distribution Arch Linux rolling",
|
||||
"processor AMD Ryzen 9 8945HS",
|
||||
"cores 8",
|
||||
"threads 16",
|
||||
"memory 30.6 GiB",
|
||||
"virtual physical",
|
||||
"up 3d, since",
|
||||
"timezone Europe/Brussels",
|
||||
"filesystems / 920.1 GiB, 29.6 GiB free (ext4 /dev/nvme0n1p3)",
|
||||
"public address unknown: how it learns it is not decided yet",
|
||||
} {
|
||||
if !strings.Contains(got, want) {
|
||||
t.Errorf("missing %q", want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestAFactAMachineCouldNotReadIsShownUnknownNeverZero(t *testing.T) {
|
||||
now := time.Date(2026, 10, 12, 15, 0, 0, 0, time.UTC)
|
||||
a := inventory.MachineAccount{Architecture: "arm64", Kernel: "linux", Machine: &inventory.MadeOf{
|
||||
ObservedAt: now, Unknown: map[string]string{"memory": "MemTotal not found in /proc/meminfo",
|
||||
"cores": "/proc/cpuinfo says no core id", "kernel release": "permission denied"}}}
|
||||
got := strings.Join(machineLines(a, true, now), "\n")
|
||||
t.Log("\n" + got)
|
||||
for _, want := range []string{
|
||||
"memory unknown: MemTotal not found in /proc/meminfo",
|
||||
"cores unknown: /proc/cpuinfo says no core id",
|
||||
"kernel linux (release unknown: permission denied)",
|
||||
// Neither said nor named unknown: the node-engine broke its rule, and that is shown.
|
||||
"timezone unknown: the machine said neither it nor why",
|
||||
} {
|
||||
if !strings.Contains(got, want) {
|
||||
t.Errorf("missing %q", want)
|
||||
}
|
||||
}
|
||||
if strings.Contains(got, " 0 ") || strings.Contains(got, "0 B") {
|
||||
t.Error("a fact the machine could not read was shown as a zero")
|
||||
}
|
||||
}
|
||||
|
||||
func TestAnOlderNodeEngineStillShowsArchitectureAndKernel(t *testing.T) {
|
||||
got := strings.Join(machineLines(inventory.MachineAccount{Architecture: "amd64", Kernel: "linux"}, true,
|
||||
time.Now()), "\n")
|
||||
if !strings.Contains(got, "older than the one that says it") || !strings.Contains(got, "architecture amd64") ||
|
||||
!strings.Contains(got, "kernel linux") {
|
||||
t.Errorf("%s", got)
|
||||
}
|
||||
if got := strings.Join(machineLines(inventory.MachineAccount{}, false, time.Now()), "\n"); !strings.Contains(got, "never said") {
|
||||
t.Errorf("%s", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAnAccountWithoutItsTimeSaysAgeUnknown(t *testing.T) {
|
||||
got := machineLines(inventory.MachineAccount{Architecture: "amd64", Kernel: "linux",
|
||||
Machine: &inventory.MadeOf{KernelRelease: "7.2"}}, true, time.Now())[0]
|
||||
if !strings.Contains(got, "age unknown") || strings.Contains(got, "d ago") {
|
||||
t.Errorf("%s", got)
|
||||
}
|
||||
}
|
||||
@@ -875,6 +875,17 @@ func showNode(ctx context.Context, stored *stores, name string) error {
|
||||
}
|
||||
}
|
||||
|
||||
// What it is made of, in one block with its age (novox/hq issue 160): collected and shown in the same
|
||||
// breath, so no fact the machine says is kept and read by nothing.
|
||||
if account, had, err := inv.MachineOf(ctx, name); err != nil {
|
||||
fmt.Printf("\n what it is made of could NOT be read: %v\n", err)
|
||||
} else {
|
||||
fmt.Println()
|
||||
for _, line := range machineLines(account, had, time.Now()) {
|
||||
fmt.Println(line)
|
||||
}
|
||||
}
|
||||
|
||||
held, err := inv.Profile(ctx, name)
|
||||
if err != nil {
|
||||
return err
|
||||
|
||||
@@ -975,9 +975,14 @@ func renderingFor(ctx context.Context, open *stores, node string,
|
||||
if err != nil {
|
||||
return catalogue.Rendering{}, inventory.Node{}, err
|
||||
}
|
||||
readsBusy, err := engineReadsBusy(ctx, inv, node)
|
||||
if err != nil {
|
||||
return catalogue.Rendering{}, inventory.Node{}, err
|
||||
}
|
||||
return catalogue.Rendering{
|
||||
ReadsHealth: readsHealth,
|
||||
JudgesRoot: judgesRoot,
|
||||
ReadsBusy: readsBusy,
|
||||
BusMembership: memberships[node],
|
||||
Settings: settings, Generators: gens, Grants: grants, Needed: needed, Foreseen: foreseen, Ports: ports,
|
||||
Certificate: certificate, Authority: authority, Mesh: private, Names: names,
|
||||
@@ -1010,6 +1015,16 @@ func engineJudgesRoot(ctx context.Context, inv *inventory.Inventory, node string
|
||||
return had && stated.Contract >= link.RootContract, nil
|
||||
}
|
||||
|
||||
// engineReadsBusy says whether a machine's node-engine reads a container's `busy` (novox/hq ADR 0335), by its
|
||||
// own newest statement, for the same reason as engineReadsHealth: an older engine parses strictly.
|
||||
func engineReadsBusy(ctx context.Context, inv *inventory.Inventory, node string) (bool, error) {
|
||||
stated, had, err := inv.HealthOf(ctx, node)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
return had && stated.Contract >= link.BusyContract, nil
|
||||
}
|
||||
|
||||
// zonesInTheMesh is every zone a module in the mesh declares, where the mesh placed it (novox/hq ADR
|
||||
// 0199): the zone settled from that node's settings, the node's private address, the port the
|
||||
// answering listen is published on there.
|
||||
|
||||
@@ -30,8 +30,9 @@ var maxPlanWait = link.LongestAnswer - planWaitStart
|
||||
// answered that the call is still running, and `calls` with its id says what came of it.
|
||||
const planWaitStart = 5 * time.Second
|
||||
|
||||
// planWaitEvery is how often a wait reads the plan again.
|
||||
const planWaitEvery = time.Second
|
||||
// planWaitEvery is how often a wait reads the plan again. A variable so a test reads it more often
|
||||
// (novox/hq issue 591).
|
||||
var planWaitEvery = time.Second
|
||||
|
||||
// The things a wait waits until.
|
||||
var planWaitUntil = []string{"moved", "tier", "done"}
|
||||
|
||||
@@ -138,6 +138,12 @@ func TestThePlansVerbComposesAWait(t *testing.T) {
|
||||
func TestAPlanFollowedFromTheStoreAnswersWhenItMoves(t *testing.T) {
|
||||
open := aMesh(t)
|
||||
ctx := t.Context()
|
||||
// Read every 50ms rather than every second, so the wait answers as soon as the plan moves and the bound
|
||||
// below passes in a fraction of a second (novox/hq issue 591); the move still waits 1.5s, long enough
|
||||
// for the wait to have read the plan before it.
|
||||
was := planWaitEvery
|
||||
planWaitEvery = 50 * time.Millisecond
|
||||
t.Cleanup(func() { planWaitEvery = was })
|
||||
now := time.Now().UTC()
|
||||
walk := inventory.Plan{ID: "plan-229", Repository: "novox/mesh-catalog", Branch: "main", Commit: "c229c229",
|
||||
Created: now, State: inventory.PlanBuilding, Tiers: [][]string{{"a"}, {"b"}},
|
||||
@@ -169,11 +175,11 @@ func TestAPlanFollowedFromTheStoreAnswersWhenItMoves(t *testing.T) {
|
||||
if got.Came != "tier" || got.Plan.Tier != 1 || got.Plan.State != inventory.PlanRolling {
|
||||
t.Fatalf("the wait answered %+v", got)
|
||||
}
|
||||
if waited := time.Since(began); waited > 10*time.Second {
|
||||
if waited := time.Since(began); waited > 3*time.Second {
|
||||
t.Errorf("it answered after %s, not when the plan moved", waited)
|
||||
}
|
||||
out = printed(t, func() error {
|
||||
return plansCommand(ctx, []string{"wait", walk.ID, "--bound", "2"})
|
||||
return plansCommand(ctx, []string{"wait", walk.ID, "--bound", "300ms"})
|
||||
})
|
||||
if err := json.Unmarshal([]byte(out), &got); err != nil || got.Came != "bound" || got.Until != "moved" {
|
||||
t.Fatalf("a plan that did not move answered %+v (%v)", got, err)
|
||||
|
||||
@@ -14,7 +14,6 @@ import (
|
||||
"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"
|
||||
@@ -81,6 +80,30 @@ func listQueue(ctx context.Context, args []string, w io.Writer) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// keepingTheQueueInOrder runs the build queue's passes while this controller acts (novox/hq issue 558):
|
||||
// the one process that puts the queue in order, so no two passes pause and resume it over each other.
|
||||
// The build seat is read from the catalogue at most every half minute.
|
||||
func keepingTheQueueInOrder(ctx context.Context, open *stores, js *broker.JetStream) {
|
||||
if js == nil {
|
||||
return
|
||||
}
|
||||
seat, readAt := link.TheBuildMachine, time.Time{}
|
||||
seatNow := func() string {
|
||||
if time.Since(readAt) < watchEvery {
|
||||
return seat
|
||||
}
|
||||
if entries, err := open.inventory.Catalogued(ctx); err == nil {
|
||||
seat, readAt = buildSeatAmong(entries), time.Now()
|
||||
} else {
|
||||
fmt.Printf("the build seat is read as %s: the catalogue cannot be read: %v\n", seat, err)
|
||||
}
|
||||
return seat
|
||||
}
|
||||
link.KeepingInOrder(ctx, js, seatNow, link.Holding, func(format string, args ...any) {
|
||||
fmt.Printf(format+"\n", args...)
|
||||
})
|
||||
}
|
||||
|
||||
// queueAnswer is the queue as data (novox/hq issue 455): what a caller reads to say how far back an ask
|
||||
// waits — "waiting in the build queue behind 2" — which the table alone could not tell it. Text is the
|
||||
// table, written from the same read: the verb answers it as its output and the rest as its answer, so the
|
||||
@@ -100,7 +123,8 @@ type queueAnswer struct {
|
||||
type queuedAnswer struct {
|
||||
ID string `json:"id"`
|
||||
State string `json:"state"`
|
||||
// Kind is "check" for a pull request's merge check, "build" for anything else.
|
||||
// Kind is "delivery" for a build a walk waits for, "check" for a pull request's check and any ask
|
||||
// whose kind cannot be told (link.KindOf, novox/hq issue 558).
|
||||
Kind string `json:"kind"`
|
||||
Repository string `json:"repository"`
|
||||
Path string `json:"path,omitempty"`
|
||||
@@ -129,10 +153,7 @@ func queueAnswerOf(q link.Queue, now time.Time) queueAnswer {
|
||||
out := queueAnswer{Seat: q.Seat, Waiting: len(waiting), InFlight: len(running), Dead: len(dead),
|
||||
MaxDeliver: q.MaxDeliver, Asks: []queuedAnswer{}, Text: queueText(q, now)}
|
||||
one := func(a link.QueuedAsk, position int) queuedAnswer {
|
||||
kind := "build"
|
||||
if a.Request.Check != nil {
|
||||
kind = "check"
|
||||
}
|
||||
kind := link.KindOf(a.Request)
|
||||
since := a.AskedAt
|
||||
if a.State == link.AskInFlight && a.On != "" && !a.Started.IsZero() {
|
||||
since = a.Started
|
||||
@@ -189,19 +210,20 @@ 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)
|
||||
}
|
||||
fmt.Fprintf(&b, " %-26s %s — %s, %s (seq %d)\n", a.ID, what(a), where, ago(a.AskedAt), a.Seq)
|
||||
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:")
|
||||
fmt.Fprintln(&b, "\nwaiting, in the order they are handed out:")
|
||||
for _, a := range waiting {
|
||||
fmt.Fprintf(&b, " %-26s %s — %s (seq %d)\n", a.ID, what(a), ago(a.AskedAt), a.Seq)
|
||||
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 {
|
||||
fmt.Fprintf(&b, "\ndead — handed out as often as the worker allows (%d) and never settled, still held:\n", q.MaxDeliver)
|
||||
for _, a := range dead {
|
||||
fmt.Fprintf(&b, " %-26s %s — %s (seq %d)\n", a.ID, what(a), ago(a.AskedAt), a.Seq)
|
||||
fmt.Fprintf(&b, " %-26s %-8s %s — %s (seq %d)\n", a.ID, link.KindOf(a.Request), what(a), ago(a.AskedAt), a.Seq)
|
||||
}
|
||||
}
|
||||
switch {
|
||||
@@ -275,6 +297,15 @@ func cancelAsk(ctx context.Context, js *broker.JetStream, open *stores, seat str
|
||||
}
|
||||
}
|
||||
worker, _ := broker.HolderConsumerFor("", "", broker.DeclaredSeat{Name: seat, Accepts: []string{"build"}})
|
||||
// **Where it waits now** (novox/hq issue 558): moved into order since it was read, a waiting ask has a
|
||||
// later sequence, and the worker handing out what was behind its old one is not it being taken.
|
||||
if ask.State == link.AskWaiting {
|
||||
if q, err := link.ReadQueue(ctx, js, seat); err == nil {
|
||||
if now, ok := q.Find(ask.ID); ok && now.State == link.AskWaiting {
|
||||
ask.Seq = now.Seq
|
||||
}
|
||||
}
|
||||
}
|
||||
switch ask.State {
|
||||
case link.AskWaiting:
|
||||
if taken, err := takenSince(js, worker, ask.Seq); err != nil {
|
||||
@@ -296,14 +327,25 @@ func cancelAsk(ctx context.Context, js *broker.JetStream, open *stores, seat str
|
||||
"ends it there", ask.ID, started, ask.ID)
|
||||
}
|
||||
}
|
||||
if err := js.Context().DeleteMsg(worker.Stream, ask.Seq); err != nil && !errors.Is(err, nats.ErrMsgNotFound) &&
|
||||
!errors.Is(err, jetstream.ErrMsgNotFound) && !strings.Contains(err.Error(), "no message found") {
|
||||
// Found by its id when it was moved into order since it was read (novox/hq issue 558).
|
||||
removed, err := link.DeleteAsk(ctx, js, seat, ask)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("%s is marked cancelled and could not be deleted from the queue (seq %d): %w — a "+
|
||||
"holder taking it ends it as cancelled", ask.ID, ask.Seq, err)
|
||||
}
|
||||
recordCancelled(ctx, open, ask)
|
||||
return fmt.Sprintf("cancelled %s (%s, %s): deleted from the %s queue and recorded failed, %s — a plan "+
|
||||
"that asked for it fails with that", ask.ID, ask.Repository, ask.State, seat, link.CancelledByHand), nil
|
||||
return cancelSaid(ask, seat, removed), nil
|
||||
}
|
||||
|
||||
// cancelSaid is what a cancel says: deleted from the queue only when it was.
|
||||
func cancelSaid(ask link.QueuedAsk, seat string, removed uint64) string {
|
||||
where := fmt.Sprintf("deleted from the %s queue", seat)
|
||||
if removed == 0 {
|
||||
where = fmt.Sprintf("no longer in the %s queue to delete, and marked cancelled, so a holder taking it "+
|
||||
"ends it", seat)
|
||||
}
|
||||
return fmt.Sprintf("cancelled %s (%s, %s): %s and recorded failed, %s — a plan "+
|
||||
"that asked for it fails with that", ask.ID, ask.Repository, ask.State, where, link.CancelledByHand)
|
||||
}
|
||||
|
||||
// holderLooks is how long a cancel waits for a holder that took the ask before the mark to say it
|
||||
|
||||
@@ -24,7 +24,7 @@ func aQueueOfEachKind(now time.Time) link.Queue {
|
||||
State: link.AskInFlight, On: "ace", Started: started,
|
||||
Request: link.BuildRequest{ID: "build-running", Check: &link.CheckRequest{Owner: "novox", Repo: "mesh-host"}}},
|
||||
{Seq: 3, ID: "build-first", Repository: "ssh://forge/novox/mesh-catalog", Ref: "main", AskedAt: asked,
|
||||
State: link.AskWaiting},
|
||||
State: link.AskWaiting, Request: link.BuildRequest{ID: "build-first", Repository: "ssh://forge/novox/mesh-catalog"}},
|
||||
{Seq: 4, ID: "build-second", Repository: "ssh://forge/novox/mesh-catalog", Path: "modules/gitea", Ref: aCommit,
|
||||
AskedAt: now.Add(-time.Minute), State: link.AskWaiting,
|
||||
Request: link.BuildRequest{ID: "build-second", Check: &link.CheckRequest{Owner: "novox", Repo: "mesh-catalog"}}},
|
||||
@@ -51,7 +51,7 @@ func TestTheQueueVerbAnswersEachAskAsData(t *testing.T) {
|
||||
t.Errorf("in flight: %+v", running)
|
||||
}
|
||||
first, second := byID["build-first"], byID["build-second"]
|
||||
if first.Position != 1 || first.Kind != "build" || first.Commit != "" || first.Ref != "main" || first.Machine != "" ||
|
||||
if first.Position != 1 || first.Kind != "delivery" || first.Commit != "" || first.Ref != "main" || first.Machine != "" ||
|
||||
!first.Since.Equal(now.Add(-5*time.Minute)) {
|
||||
t.Errorf("first waiting: %+v", first)
|
||||
}
|
||||
@@ -62,6 +62,13 @@ func TestTheQueueVerbAnswersEachAskAsData(t *testing.T) {
|
||||
if dead := byID["build-dead"]; dead.Position != 0 || dead.State != link.AskDead {
|
||||
t.Errorf("dead: %+v", dead)
|
||||
}
|
||||
// And the table says each ask's kind too (novox/hq issue 558).
|
||||
for _, line := range []string{"build-running check ", "build-first delivery ",
|
||||
"build-second check ", "build-dead check "} {
|
||||
if !strings.Contains(got.Text, line) {
|
||||
t.Errorf("the table does not say %q:\n%s", line, got.Text)
|
||||
}
|
||||
}
|
||||
// In the table's order: in flight, then the waiting by position, then the dead.
|
||||
var order []string
|
||||
for _, a := range got.Asks {
|
||||
@@ -135,3 +142,16 @@ func TestTheQueueNeverShowsARepositorysPassword(t *testing.T) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// A cancel says it deleted the ask only when it did: moved into order and gone since, it says so (novox/hq
|
||||
// issue 558).
|
||||
func TestACancelSaysDeletedOnlyWhenItDeleted(t *testing.T) {
|
||||
ask := link.QueuedAsk{ID: "build-1", Repository: "novox/x", State: link.AskWaiting, Seq: 4}
|
||||
if said := cancelSaid(ask, link.TheBuildMachine, 9); !strings.Contains(said, "deleted from the") {
|
||||
t.Errorf("removed, it says: %s", said)
|
||||
}
|
||||
if said := cancelSaid(ask, link.TheBuildMachine, 0); strings.Contains(said, "deleted from") ||
|
||||
!strings.Contains(said, "no longer in the") {
|
||||
t.Errorf("gone already, it says: %s", said)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,90 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/catalogue"
|
||||
"github.com/novox/mesh-controller/internal/inventory"
|
||||
)
|
||||
|
||||
// TestReplay201ThePlansSend replays novox/hq issue 201 (2026-10-02): two merges a minute apart made two plans; the
|
||||
// controller's plan rolled the newer controller onto the control node, and the node-engine's plan then sent that
|
||||
// machine the controller build before it, whose binary refused the seat row the newer one had widened. Here the
|
||||
// control node, anchor, was last sent the newer controller c2 while the mesh's record of the controller is the
|
||||
// build before it, c1, which passed its gate. The outcome asserted: a send that is no person's push — composed as a
|
||||
// plan's send composes it (keepingRecorded, what sendKeeps gives every send through sendToEach) — carries the
|
||||
// controller at c2, the build the machine was last sent, and not the older record; a person's push still carries
|
||||
// the record. Written with only what the controller had before its fix (mesh-controller #273) — the inventory's
|
||||
// build, gate, assignment and send records, keepingRecorded, planFor and declarationFor — so it is laid over the
|
||||
// commit before; the fix's own TestReplay201 reads what the fix added. Needs the store (MESH_TEST_POSTGRES), and
|
||||
// is named in testdata/store-tests.
|
||||
func TestReplay201ThePlansSend(t *testing.T) {
|
||||
open := aMesh(t)
|
||||
ctx := t.Context()
|
||||
inv := open.inventory
|
||||
const older, newer = "c1aaaaaaaaaa", "c2bbbbbbbbbb"
|
||||
old, now := time.Now().Add(-2*time.Hour), time.Now().Add(-time.Minute)
|
||||
record := func(commit string, asked time.Time) inventory.Build {
|
||||
t.Helper()
|
||||
manifest, _ := json.Marshal(catalogue.Manifest{Module: "mesh-controller", Version: "1"})
|
||||
b := inventory.Build{ID: "build-mesh-controller-" + commit, Module: "mesh-controller", Commit: commit,
|
||||
Repository: "novox/mesh-controller", Manifest: manifest, Asked: asked, At: asked,
|
||||
Made: []inventory.Artifact{{Name: "x", Kind: "bundle", Reference: "sha256:" + commit + "0000000000000000"}}}
|
||||
if err := inv.RecordBuild(ctx, b); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return b
|
||||
}
|
||||
|
||||
// The older controller rolled out and passed its gate; the newer one was asked after it and registered.
|
||||
before := record(older, old)
|
||||
if err := inv.RecordGate(ctx, inventory.GateVerdict{Build: before.ID, Module: "mesh-controller", Commit: older,
|
||||
Machines: []string{"anchor"}, Verdict: inventory.GatePassed}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
record(newer, now)
|
||||
if err := inv.RegisterModule(ctx, catalogue.Manifest{Module: "mesh-controller", Version: "1"}, inventory.Source{
|
||||
Repository: "novox/mesh-controller", BuiltFrom: newer, Head: newer, Asked: now}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := inv.Assign(ctx, "anchor", "mesh-controller"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// The controller's plan sent anchor the newer controller.
|
||||
if err := inv.RecordSent(ctx, nodeID(t, open, "anchor"), "d-anchor", map[string]string{"mesh-controller": newer}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// And the mesh's record of the controller is the build before it — whatever made it so.
|
||||
if err := inv.RestoreModule(ctx, before); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// The controller each send to anchor carries: a plan's send keeps recorded builds (keepingRecorded, what
|
||||
// sendKeeps gives every send but a person's push); a person's push composes the mesh's record.
|
||||
controllerSent := func(sends string, keeping bool) string {
|
||||
t.Helper()
|
||||
c := ctx
|
||||
if keeping {
|
||||
c = keepingRecorded(ctx)
|
||||
}
|
||||
plan, settings, err := planFor(c, open, "anchor")
|
||||
if err != nil {
|
||||
t.Fatalf("%s to anchor did not compose: %v", sends, err)
|
||||
}
|
||||
declared, err := declarationFor(c, open, "anchor", plan, settings)
|
||||
if err != nil {
|
||||
t.Fatalf("%s to anchor did not compose: %v", sends, err)
|
||||
}
|
||||
return declared.Builds["mesh-controller"]
|
||||
}
|
||||
|
||||
if got := controllerSent("a plan's send", true); got != newer {
|
||||
t.Fatalf("a plan's send carried anchor the controller %s, older than the %s anchor was last sent: the "+
|
||||
"older binary would read the row its successor widened", short(got), short(newer))
|
||||
}
|
||||
if got := controllerSent("a person's push", false); got != older {
|
||||
t.Fatalf("a person's push carried the controller %s, not the mesh's record %s", short(got), short(older))
|
||||
}
|
||||
}
|
||||
@@ -26,6 +26,14 @@ import (
|
||||
// shape fails the suite, naming its source. The keeper would say such a summary in words at run time;
|
||||
// this is where the producer is made to say it rightly in the first place.
|
||||
func TestMain(m *testing.M) {
|
||||
// **A command line this suite runs ends when it is done** (novox/hq issue 591). Many tests run this test
|
||||
// binary as a verb's command line, and a binary built with the race detector sleeps a second before it
|
||||
// exits (GORACE's atexit_sleep_ms), which was more than the rest of each such test. The children still
|
||||
// run under the race detector and still fail on a race they find; they only stop waiting at their end.
|
||||
// Set before any child starts, and only where nobody chose GORACE.
|
||||
if _, chosen := os.LookupEnv("GORACE"); !chosen {
|
||||
_ = os.Setenv("GORACE", "atexit_sleep_ms=0")
|
||||
}
|
||||
// The process a mesh-cli test runs as a command line: it says the verb and the caller it was given, and
|
||||
// ends (meshcli_test.go).
|
||||
// The process a mesh-cli test runs as `secret accept`: it reads its value exactly as `secret accept` does
|
||||
|
||||
@@ -739,6 +739,9 @@ func watchTheMesh(ctx context.Context, open *stores, server *link.Server, bus li
|
||||
// and the answer chosen is performed on its warrant.
|
||||
startAsking(watching, open, server, bus.Conn, keeper)
|
||||
go forgettingOldHeals(watching, open.inventory)
|
||||
// And the build queue's order (novox/hq issue 558): a check that has waited its while goes ahead of the
|
||||
// deliveries asked after it even when nothing new is asked.
|
||||
go keepingTheQueueInOrder(watching, open, server.JetStream())
|
||||
fmt.Printf("watching the mesh: %d signal(s) every %s, %d probe(s) every %s; what is wrong is kept in %s "+
|
||||
"and said as %s events\n", len(watchedRows()), watchEvery, len(runnableProbes()), doctorEvery,
|
||||
broker.ConditionsBucket, conditions.Seat)
|
||||
|
||||
@@ -36,4 +36,4 @@ require (
|
||||
// committed. Every build (the build agent's `go build`, the Dockerfile) compiles from vendor/ and
|
||||
// fetches nothing; go refuses to build when vendor/ and this file disagree, so a pin moved without
|
||||
// `go mod vendor` fails loudly, at once, everywhere.
|
||||
replace github.com/novox/mesh-host => git.novox.be/novox/mesh-host v0.0.0-20261011092230-20d5af9b2d5f
|
||||
replace github.com/novox/mesh-host => git.novox.be/novox/mesh-host v0.0.0-20261012224132-bb60e7b584d8
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
git.novox.be/novox/mesh-host v0.0.0-20261011092230-20d5af9b2d5f h1:8N5OW2mdTNIck2pe4EciTYX5NsrPsHrTLENGNIWYNTU=
|
||||
git.novox.be/novox/mesh-host v0.0.0-20261011092230-20d5af9b2d5f/go.mod h1:tcTK4LMs1d6JpZUwy3hSHMFG/MIuDr/XFs7/nbeaW/c=
|
||||
git.novox.be/novox/mesh-host v0.0.0-20261012224132-bb60e7b584d8 h1:a+SH9NroLc3FkNYcbg0wjMEWGjWp6SCl32oVH2jxh9s=
|
||||
git.novox.be/novox/mesh-host v0.0.0-20261012224132-bb60e7b584d8/go.mod h1:tcTK4LMs1d6JpZUwy3hSHMFG/MIuDr/XFs7/nbeaW/c=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261010191001-33917f91ac3c h1:l5onJwoIeH8yE/PjVlEeoPyXBsHp2NQiWLllz2fwX7s=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261010191001-33917f91ac3c/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
|
||||
|
||||
@@ -0,0 +1,124 @@
|
||||
package catalogue
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// A long-running container may say it is busy (novox/hq ADR 0335 §1).
|
||||
//
|
||||
// **Beside its health, on the container**: a command run inside it, by its shell, as an exec health check
|
||||
// is, that exits 0 while the container holds work a replacement would lose and non-zero when it is idle;
|
||||
// and how long one ask may take. Being asked is also its notice that it is about to be replaced: from the
|
||||
// first ask it takes no new work. The node-engine asks before it replaces the container, leaves it running
|
||||
// while it says busy, and replaces it on the first pass it says idle, or at 15 minutes — the engine's bound,
|
||||
// never the module's.
|
||||
//
|
||||
// Sent only to a node-engine that reads it (link.BusyContract): an older, strict engine refuses a field it
|
||||
// does not know, whole, and to it the container is replaced at once, as before.
|
||||
|
||||
// BusyField is the resource field.
|
||||
const BusyField = "busy"
|
||||
|
||||
// The bounds of one ask (ADR 0335), the node-engine's own, held here near the author.
|
||||
const (
|
||||
BusyTimeoutDefault = 10 * time.Second
|
||||
BusyTimeoutCeiling = time.Minute
|
||||
)
|
||||
|
||||
// Busy is one container's busy, read.
|
||||
type Busy struct {
|
||||
Command string
|
||||
Timeout time.Duration
|
||||
}
|
||||
|
||||
// ReadBusy reads a resource's `busy` field, its default applied. False when it carries none.
|
||||
func ReadBusy(r map[string]any) (Busy, bool, []string) {
|
||||
raw, present := r[BusyField]
|
||||
if !present {
|
||||
return Busy{}, false, nil
|
||||
}
|
||||
id := fmt.Sprint(r["id"])
|
||||
fields, ok := raw.(map[string]any)
|
||||
if !ok {
|
||||
return Busy{}, true, []string{fmt.Sprintf("%s: busy is %T; it is an object with a command", id, raw)}
|
||||
}
|
||||
var problems []string
|
||||
keys := make([]string, 0, len(fields))
|
||||
for k := range fields {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
sort.Strings(keys)
|
||||
for _, k := range keys {
|
||||
if k != "command" && k != "timeout" {
|
||||
problems = append(problems, fmt.Sprintf("%s: its busy says %q; busy has a command and a timeout", id, k))
|
||||
}
|
||||
}
|
||||
b := Busy{Timeout: BusyTimeoutDefault}
|
||||
if v, has := fields["command"]; has {
|
||||
s, ok := v.(string)
|
||||
if !ok {
|
||||
problems = append(problems, fmt.Sprintf("%s: its busy's command is %T; it is text", id, v))
|
||||
}
|
||||
b.Command = strings.TrimSpace(s)
|
||||
}
|
||||
if v, has := fields["timeout"]; has {
|
||||
s, _ := v.(string)
|
||||
d, err := time.ParseDuration(strings.TrimSpace(s))
|
||||
switch {
|
||||
case err != nil:
|
||||
problems = append(problems, fmt.Sprintf("%s: its busy's timeout is %v; it is a duration such as \"10s\"", id, v))
|
||||
case d <= 0 || d > BusyTimeoutCeiling:
|
||||
problems = append(problems, fmt.Sprintf("%s: an ask of its busy takes more than nothing and at most %s, not %s",
|
||||
id, BusyTimeoutCeiling, d))
|
||||
default:
|
||||
b.Timeout = d
|
||||
}
|
||||
}
|
||||
return b, true, problems
|
||||
}
|
||||
|
||||
// busyProblems is everything wrong with a manifest's `busy` fields, in the manifest's words: a command on
|
||||
// one line, a timeout within its bounds, and only on a container that stays up.
|
||||
func busyProblems(m Manifest) []string {
|
||||
var problems []string
|
||||
for _, r := range m.Resources {
|
||||
b, has, read := ReadBusy(r)
|
||||
if !has {
|
||||
continue
|
||||
}
|
||||
where := m.Module + ": " + fmt.Sprint(r["id"])
|
||||
for _, p := range read {
|
||||
problems = append(problems, m.Module+": "+p)
|
||||
}
|
||||
if fmt.Sprint(r["type"]) != "container" || !LongRunning(r) {
|
||||
problems = append(problems, fmt.Sprintf("%s declares busy and is not a container that stays up: busy is "+
|
||||
"asked of a running container before it is replaced (ADR 0335)", where))
|
||||
}
|
||||
if b.Command == "" {
|
||||
problems = append(problems, where+"'s busy names no command: one run inside the container, exiting 0 "+
|
||||
"while it is busy")
|
||||
}
|
||||
if strings.ContainsAny(b.Command, "\n\r") {
|
||||
problems = append(problems, where+"'s busy command is one line")
|
||||
}
|
||||
}
|
||||
return problems
|
||||
}
|
||||
|
||||
// busyInto composes a container's `busy` into the node-engine's words, or takes it away: for an engine that
|
||||
// reads it, the command and its timeout written out; for one that does not, the field is not sent, and the
|
||||
// container is replaced at once, as before.
|
||||
func busyInto(resource map[string]any, with Rendering) {
|
||||
if _, has := resource[BusyField]; !has {
|
||||
return
|
||||
}
|
||||
if !with.ReadsBusy {
|
||||
delete(resource, BusyField)
|
||||
return
|
||||
}
|
||||
b, _, _ := ReadBusy(resource)
|
||||
resource[BusyField] = map[string]any{"command": b.Command, "timeout": b.Timeout.String()}
|
||||
}
|
||||
@@ -0,0 +1,94 @@
|
||||
package catalogue
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/novox/mesh-host/validate"
|
||||
)
|
||||
|
||||
// A long-running container may say it is busy (novox/hq ADR 0335 §1): `module check` holds the field to its
|
||||
// shape near its author, and it is sent only to a node-engine that reads it, in words the engine's own
|
||||
// validator accepts.
|
||||
|
||||
func busyManifest(t *testing.T, on string, busy any) []byte {
|
||||
t.Helper()
|
||||
extra := map[string]any{"id": "agent", "type": "container", "name": "app-agent",
|
||||
"image": "registry.example/app" + healthDigest}
|
||||
var m map[string]any
|
||||
if err := json.Unmarshal(healthManifest(t, "", nil, extra), &m); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, r := range m["resources"].([]any) {
|
||||
if r := r.(map[string]any); r["id"] == on {
|
||||
r[BusyField] = busy
|
||||
}
|
||||
}
|
||||
out, err := json.Marshal(m)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func TestABusyIsHeldToItsShape(t *testing.T) {
|
||||
if _, err := ParseManifest(busyManifest(t, "agent", map[string]any{"command": "mesh-builder busy"})); err != nil {
|
||||
t.Fatalf("a well-formed busy was refused: %v", err)
|
||||
}
|
||||
if _, err := ParseManifest(busyManifest(t, "agent", map[string]any{"command": "x", "timeout": "30s"})); err != nil {
|
||||
t.Fatalf("a busy with its timeout was refused: %v", err)
|
||||
}
|
||||
for name, c := range map[string]struct {
|
||||
on string
|
||||
busy any
|
||||
says string
|
||||
}{
|
||||
"text": {"agent", "mesh-builder busy", "it is an object"},
|
||||
"no command": {"agent", map[string]any{"timeout": "5s"}, "names no command"},
|
||||
"two lines": {"agent", map[string]any{"command": "a\nb"}, "one line"},
|
||||
"a long timeout": {"agent", map[string]any{"command": "x", "timeout": "5m"}, "at most 1m0s"},
|
||||
"an unknown key": {"agent", map[string]any{"command": "x", "every": "1m"}, `says "every"`},
|
||||
"on a step": {"seed", map[string]any{"command": "x"}, "not a container that stays up"},
|
||||
"on a service": {"daemon", map[string]any{"command": "x"}, "not a container that stays up"},
|
||||
} {
|
||||
_, err := ParseManifest(busyManifest(t, c.on, c.busy))
|
||||
if err == nil || !strings.Contains(err.Error(), c.says) {
|
||||
t.Errorf("%s: want a refusal saying %q, got %v", name, c.says, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestABusyIsSentOnlyToAnEngineThatReadsIt(t *testing.T) {
|
||||
m, err := ParseManifest(busyManifest(t, "agent", map[string]any{"command": "mesh-builder busy"}))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
compose := func(with Rendering) []map[string]any {
|
||||
out, err := Resolution{Node: "laptop", Modules: []Manifest{m}}.Declaration(with)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return out
|
||||
}
|
||||
older := compose(Rendering{})
|
||||
if b, sent := older[indexOfID(older, "app.agent")][BusyField]; sent {
|
||||
t.Fatalf("busy was sent to an engine that does not read it: %v", b)
|
||||
}
|
||||
newer := compose(Rendering{ReadsBusy: true})
|
||||
got, _ := newer[indexOfID(newer, "app.agent")][BusyField].(map[string]any)
|
||||
if got["command"] != "mesh-builder busy" || got["timeout"] != "10s" {
|
||||
t.Fatalf("busy composed as %v", got)
|
||||
}
|
||||
// And the engine's own validator takes the container with it.
|
||||
body, err := json.Marshal(map[string]any{"declaration": validate.Version, "resources": newer})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if problems := validate.Declaration(body); len(problems) > 0 {
|
||||
t.Fatalf("the node-engine's validator refuses the composed busy: %v", problems)
|
||||
}
|
||||
if _, has := m.Resources[len(m.Resources)-1][BusyField].(map[string]any)["timeout"]; has {
|
||||
t.Error("composing wrote the timeout into the catalogue's own manifest")
|
||||
}
|
||||
}
|
||||
@@ -246,6 +246,11 @@ type Rendering struct {
|
||||
// statement's contract is link.RootContract or later). To an older, strict engine the field is not
|
||||
// sent, and the account it names is not judged — which the self-check says, as not judged.
|
||||
JudgesRoot bool
|
||||
|
||||
// ReadsBusy says this machine's node-engine reads a container's `busy` (novox/hq ADR 0335: its statement's
|
||||
// contract is link.BusyContract or later). To an older, strict engine the field is not sent, and the
|
||||
// container is replaced at once, as before.
|
||||
ReadsBusy bool
|
||||
}
|
||||
|
||||
// RootField is a user resource's field saying the account must never become root without a person
|
||||
@@ -1051,6 +1056,8 @@ func (r Resolution) compose(with Rendering, owner map[string]string,
|
||||
// How it is ready, in the node-engine's words: its endpoint as the port this machine
|
||||
// published it on — or not sent at all to an engine older than the field (ADR 0240).
|
||||
healthInto(copied, m, with)
|
||||
// And whether it is busy (novox/hq ADR 0335): sent only to an engine that asks it.
|
||||
busyInto(copied, with)
|
||||
// And a user's `root` (novox/hq ADR 0266): sent only when it asserts something, to an engine
|
||||
// that judges it.
|
||||
rootInto(copied, with)
|
||||
|
||||
@@ -1693,6 +1693,8 @@ func ParseManifest(raw []byte) (Manifest, error) {
|
||||
problems = append(problems, LogProblems(m)...)
|
||||
// And the defaults it gives its settings (setting_defaults.go, novox/hq ADR 0262).
|
||||
problems = append(problems, SettingProblems(m)...)
|
||||
// And which setting keys each mergeable file takes (setting_takes.go, novox/hq ADR 0326).
|
||||
problems = append(problems, TakenSettingsProblems(m)...)
|
||||
wellFormed := true
|
||||
for _, c := range m.Claims {
|
||||
if !name.MatchString(c.Name) {
|
||||
@@ -2240,6 +2242,8 @@ func ParseManifest(raw []byte) (Manifest, error) {
|
||||
// How each long-running resource is ready (novox/hq ADR 0240 rule 2): said near its author, in the
|
||||
// words the node-engine would refuse it in far away.
|
||||
problems = append(problems, healthProblems(m)...)
|
||||
// And whether a container is busy (novox/hq ADR 0335), likewise.
|
||||
problems = append(problems, busyProblems(m)...)
|
||||
if len(problems) > 0 {
|
||||
sort.Strings(problems)
|
||||
return Manifest{}, fmt.Errorf("this manifest cannot be used:\n - %s",
|
||||
|
||||
@@ -0,0 +1,176 @@
|
||||
package catalogue
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"strings"
|
||||
"unicode"
|
||||
)
|
||||
|
||||
// A mergeable file names the setting keys it takes (novox/hq ADR 0326, issue 168).
|
||||
//
|
||||
// Before this, every key of a module's layers merged into every one of its mergeable files: the
|
||||
// mesh's own words, a key meant for a `${setting:<key>}` in another file, a key for a second
|
||||
// configuration file. So a module could not own two configurable files, and every receiver had to
|
||||
// ignore what it did not know. ADR 0164 §11 decided that a setting names the files that read it; this
|
||||
// is that relation, stated from the file's side, so the operator's own values stay declared nowhere
|
||||
// (ADR 0262 §2).
|
||||
|
||||
// TakesSettingsField is the key a mergeable file carries to name the top-level setting keys it takes.
|
||||
const TakesSettingsField = "takes-settings"
|
||||
|
||||
// takenKeys is the keys a resource names in takes-settings, and whether it names any list at all. An
|
||||
// empty list is a list: that file takes no setting.
|
||||
func takenKeys(resource map[string]any) (map[string]bool, bool) {
|
||||
raw, said := resource[TakesSettingsField]
|
||||
if !said {
|
||||
return nil, false
|
||||
}
|
||||
keys := map[string]bool{}
|
||||
for _, k := range stringsOf(raw) {
|
||||
keys[k] = true
|
||||
}
|
||||
return keys, true
|
||||
}
|
||||
|
||||
// onlyTaken is the layers with every key the file does not name left out. The file's own defaults and
|
||||
// what the mesh reads itself are not touched here: settle already keeps both out of a file.
|
||||
func onlyTaken(layers []Layer, taken map[string]bool) []Layer {
|
||||
kept := make([]Layer, 0, len(layers))
|
||||
for _, layer := range layers {
|
||||
values := map[string]any{}
|
||||
for key, value := range layer.Values {
|
||||
if taken[key] {
|
||||
values[key] = value
|
||||
}
|
||||
}
|
||||
kept = append(kept, Layer{From: layer.From, Values: values, Default: layer.Default})
|
||||
}
|
||||
return kept
|
||||
}
|
||||
|
||||
// namesItsKeys says whether a module's mergeable files name the keys they take (ADR 0326 §2): true
|
||||
// only when it has one and every one carries takes-settings. A module with a silent mergeable file
|
||||
// keeps today's form (§3), where every key reaches every mergeable file.
|
||||
func namesItsKeys(m Manifest) bool {
|
||||
merging := 0
|
||||
for _, r := range m.Resources {
|
||||
if how, _ := r["merge"].(string); how == "" {
|
||||
continue
|
||||
}
|
||||
merging++
|
||||
if _, said := r[TakesSettingsField]; !said {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return merging > 0
|
||||
}
|
||||
|
||||
// TakenSettingsProblems is every way a definition's takes-settings is wrong (ADR 0326 §2 and §4), in
|
||||
// its own words: on a file that does not merge, not a list of keys, a key named twice, one the mesh
|
||||
// reads itself, one the file keeps for itself, or one mergeable file naming its keys beside another
|
||||
// that names none.
|
||||
func TakenSettingsProblems(m Manifest) []string {
|
||||
// Not yet here: refusing a module that declares a setting and owns a mergeable file naming no keys
|
||||
// (ADR 0326 §3). betterbird is one, and the catalogue cannot name its keys before this controller
|
||||
// runs, so that refusal follows once the catalogue does (novox/hq issue 574).
|
||||
return takenSettingsProblems(m)
|
||||
}
|
||||
|
||||
// takenSettingsProblems is every rule of ADR 0326 §2 and §4, judged where a definition is checked and
|
||||
// where a stored one is composed.
|
||||
func takenSettingsProblems(m Manifest) []string {
|
||||
var problems []string
|
||||
var naming, silent []string
|
||||
for _, r := range m.Resources {
|
||||
id := fmt.Sprint(r["id"])
|
||||
how, _ := r["merge"].(string)
|
||||
raw, said := r[TakesSettingsField]
|
||||
if how == "" {
|
||||
if said {
|
||||
problems = append(problems, fmt.Sprintf("%s's %s says %s and does not merge: only a mergeable "+
|
||||
"file takes settings by key (novox/hq ADR 0326)", m.Module, id, TakesSettingsField))
|
||||
}
|
||||
continue
|
||||
}
|
||||
if !said {
|
||||
silent = append(silent, id)
|
||||
continue
|
||||
}
|
||||
naming = append(naming, id)
|
||||
list, isList := raw.([]any)
|
||||
if !isList {
|
||||
problems = append(problems, fmt.Sprintf("%s's %s: %s is a list of setting keys, and this is %T",
|
||||
m.Module, id, TakesSettingsField, raw))
|
||||
continue
|
||||
}
|
||||
protected := map[string]bool{}
|
||||
for _, k := range stringsOf(r["protected"]) {
|
||||
protected[k] = true
|
||||
}
|
||||
seen := map[string]bool{}
|
||||
for _, item := range list {
|
||||
key, isString := item.(string)
|
||||
switch {
|
||||
case !isString || !usableKey(key):
|
||||
problems = append(problems, fmt.Sprintf("%s's %s: %v in %s is not a usable setting key",
|
||||
m.Module, id, item, TakesSettingsField))
|
||||
case seen[key]:
|
||||
problems = append(problems, fmt.Sprintf("%s's %s names %q twice in %s",
|
||||
m.Module, id, key, TakesSettingsField))
|
||||
case meshWords[key]:
|
||||
problems = append(problems, fmt.Sprintf("%s's %s names %q in %s, and the mesh reads %q itself "+
|
||||
"(novox/hq ADR 0164 §14)", m.Module, id, key, TakesSettingsField, key))
|
||||
case protected[key]:
|
||||
problems = append(problems, fmt.Sprintf("%s's %s names %q in %s and keeps it as protected: a "+
|
||||
"key is the file's or the setting's, never both", m.Module, id, key, TakesSettingsField))
|
||||
}
|
||||
if isString {
|
||||
seen[key] = true
|
||||
}
|
||||
}
|
||||
}
|
||||
if len(naming) > 0 {
|
||||
for _, id := range silent {
|
||||
problems = append(problems, fmt.Sprintf("%s's %s merges and names no %s while %s does: once one "+
|
||||
"mergeable file of a module names the keys it takes, every one does, an empty list for none "+
|
||||
"(novox/hq ADR 0326)", m.Module, id, TakesSettingsField, strings.Join(naming, ", ")))
|
||||
}
|
||||
}
|
||||
return problems
|
||||
}
|
||||
|
||||
// usableKey is a key a file can be said to take: not empty, and no space, line end or control
|
||||
// character. Not the lower-case rule of a `${setting:<key>}`: a file's own keys are its software's
|
||||
// spelling, such as `timeZone`.
|
||||
func usableKey(key string) bool {
|
||||
if key == "" {
|
||||
return false
|
||||
}
|
||||
for _, r := range key {
|
||||
if unicode.IsSpace(r) || unicode.IsControl(r) || !unicode.IsPrint(r) {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// takenSaid is what a stray key's message adds for a module whose files name their keys: which keys
|
||||
// each file takes, so the person sees where the key would have had to be named.
|
||||
func takenSaid(m Manifest) string {
|
||||
if !namesItsKeys(m) {
|
||||
return ""
|
||||
}
|
||||
var files []string
|
||||
for _, r := range m.Resources {
|
||||
if how, _ := r["merge"].(string); how == "" {
|
||||
continue
|
||||
}
|
||||
keys := stringsOf(r[TakesSettingsField])
|
||||
if len(keys) == 0 {
|
||||
files = append(files, fmt.Sprintf("%v takes none", r["id"]))
|
||||
continue
|
||||
}
|
||||
files = append(files, fmt.Sprintf("%v takes %s", r["id"], strings.Join(keys, ", ")))
|
||||
}
|
||||
return " (" + strings.Join(files, "; ") + ")"
|
||||
}
|
||||
@@ -0,0 +1,196 @@
|
||||
package catalogue
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// twoFiles is a module that owns two configurable files, which issue 168 said no module could:
|
||||
// every key of every layer reached both.
|
||||
func twoFiles(app, sidecar []any) Manifest {
|
||||
appFile := map[string]any{"id": "app", "type": "file", "path": "/etc/app.json", "merge": MergeJSON,
|
||||
"content": `{"server":{"port":8080},"search":{"safe":1}}`}
|
||||
sideFile := map[string]any{"id": "sidecar", "type": "file", "path": "/etc/sidecar.json", "merge": MergeJSON,
|
||||
"content": `{"url":"http://127.0.0.1:8080"}`}
|
||||
if app != nil {
|
||||
appFile[TakesSettingsField] = app
|
||||
}
|
||||
if sidecar != nil {
|
||||
sideFile[TakesSettingsField] = sidecar
|
||||
}
|
||||
return Manifest{Module: "search", Resources: []map[string]any{
|
||||
appFile, sideFile,
|
||||
{"id": "env", "type": "file", "path": "/etc/search.env", "content": "SITE=${setting:site-name}\n"},
|
||||
}}
|
||||
}
|
||||
|
||||
func settledContent(t *testing.T, resource map[string]any, layers []Layer) map[string]any {
|
||||
t.Helper()
|
||||
out, err := ApplySettings(resource, layers)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, kept := out[TakesSettingsField]; kept {
|
||||
t.Errorf("%v still carries %s; the host knows no such field", out["id"], TakesSettingsField)
|
||||
}
|
||||
var parsed map[string]any
|
||||
if err := json.Unmarshal([]byte(out["content"].(string)), &parsed); err != nil {
|
||||
t.Fatalf("%v is not JSON after settling: %v", out["id"], err)
|
||||
}
|
||||
return parsed
|
||||
}
|
||||
|
||||
func TestEachMergeableFileTakesOnlyTheKeysItNames(t *testing.T) {
|
||||
// novox/hq ADR 0326 §1, issue 168: two configurable files, each taking its own keys. A key for
|
||||
// the other file, for a ${setting:…} elsewhere or for the mesh itself lands in neither.
|
||||
m := twoFiles([]any{"server", "search"}, []any{"url"})
|
||||
layers := []Layer{
|
||||
{From: "the mesh", Values: map[string]any{
|
||||
"server": map[string]any{"base_url": "https://search.example.tld/"},
|
||||
"url": "http://127.0.0.1:9090", "site-name": "Search",
|
||||
"endpoints": map[string]any{"web": map[string]any{"reach": "public"}}}},
|
||||
{From: "node-a", Values: map[string]any{"search": map[string]any{"safe": 0.0}}},
|
||||
}
|
||||
|
||||
app := settledContent(t, m.Resources[0], layers)
|
||||
for _, stray := range []string{"url", "site-name", "endpoints"} {
|
||||
if _, in := app[stray]; in {
|
||||
t.Errorf("app took %q, which it does not name: %v", stray, app)
|
||||
}
|
||||
}
|
||||
server, _ := app["server"].(map[string]any)
|
||||
if server["base_url"] != "https://search.example.tld/" || server["port"] != 8080.0 {
|
||||
t.Errorf("app's server block is %v; the setting should merge into the file's own", server)
|
||||
}
|
||||
if search, _ := app["search"].(map[string]any); search["safe"] != 0.0 {
|
||||
t.Errorf("app's search block is %v; the node's layer should win", search)
|
||||
}
|
||||
|
||||
side := settledContent(t, m.Resources[1], layers)
|
||||
if side["url"] != "http://127.0.0.1:9090" {
|
||||
t.Errorf("sidecar's url is %v; it names url", side["url"])
|
||||
}
|
||||
if len(side) != 1 {
|
||||
t.Errorf("sidecar took keys it does not name: %v", side)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAnEmptyListTakesNoSetting(t *testing.T) {
|
||||
// ADR 0164 §12: a file that names no setting takes none.
|
||||
m := twoFiles([]any{"server"}, []any{})
|
||||
side := settledContent(t, m.Resources[1], []Layer{{From: "the mesh", Values: map[string]any{
|
||||
"server": map[string]any{"port": 1.0}, "url": "x"}}})
|
||||
if len(side) != 1 || side["url"] != "http://127.0.0.1:8080" {
|
||||
t.Errorf("a file taking none was changed: %v", side)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAFileThatNamesNothingKeepsTodaysForm(t *testing.T) {
|
||||
// ADR 0326 §3, ADR 0164 §15: a module none of whose files names keys is left as it was.
|
||||
m := twoFiles(nil, nil)
|
||||
app := settledContent(t, m.Resources[0], []Layer{{From: "the mesh", Values: map[string]any{"url": "x"}}})
|
||||
if app["url"] != "x" {
|
||||
t.Errorf("a silent file should still take every key until its module names them: %v", app)
|
||||
}
|
||||
if unused := UnusedSettings(m, []Layer{{From: "the mesh", Values: map[string]any{"anything": 1}}}); len(unused) != 0 {
|
||||
t.Errorf("today's form has nothing stray, and said: %v", unused)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAKeyNoFileNamesIsStray(t *testing.T) {
|
||||
// ADR 0326 §5: once the files name their keys, a key none of them names — and nothing else
|
||||
// takes — is stray, and is said with what each file takes.
|
||||
m := twoFiles([]any{"server", "search"}, []any{"url"})
|
||||
unused := UnusedSettings(m, []Layer{{From: "the mesh", Values: map[string]any{
|
||||
"server": 1, "url": "x", "site-name": "Search", "ui": map[string]any{"theme": "dark"}}}})
|
||||
if len(unused) != 1 || !strings.Contains(unused[0], `"ui"`) {
|
||||
t.Fatalf("only ui reaches nothing; named: %v", unused)
|
||||
}
|
||||
if !strings.Contains(unused[0], "app takes server, search") || !strings.Contains(unused[0], "sidecar takes url") {
|
||||
t.Errorf("the stray key's message does not say what each file takes: %s", unused[0])
|
||||
}
|
||||
}
|
||||
|
||||
func TestOneNamingFileMakesEveryMergeableFileName(t *testing.T) {
|
||||
// ADR 0326 §2: a module with one naming and one silent mergeable file is refused, naming the
|
||||
// silent one — the silent one would otherwise still take every key.
|
||||
m := twoFiles([]any{"server"}, nil)
|
||||
problems := TakenSettingsProblems(m)
|
||||
if len(problems) != 1 || !strings.Contains(problems[0], "sidecar merges and names no takes-settings") {
|
||||
t.Fatalf("problems: %v", problems)
|
||||
}
|
||||
if err := JudgeSettings(m, nil, false); err == nil || !strings.Contains(err.Error(), "sidecar") {
|
||||
t.Errorf("composing it should refuse it too, and said: %v", err)
|
||||
}
|
||||
raw, err := json.Marshal(map[string]any{"module": m.Module, "version": "1", "resources": m.Resources})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := ParseManifest(raw); err == nil || !strings.Contains(err.Error(), "sidecar merges and names no") {
|
||||
t.Errorf("the module check does not say it: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTakesSettingsIsRefusedWhereItCannotMeanAnything(t *testing.T) {
|
||||
// ADR 0326 §4.
|
||||
cases := map[string]struct {
|
||||
resource map[string]any
|
||||
says string
|
||||
}{
|
||||
"on a file that does not merge": {
|
||||
map[string]any{"id": "plain", "type": "file", "path": "/x", "content": "x",
|
||||
TakesSettingsField: []any{"a"}},
|
||||
"does not merge"},
|
||||
"not a list": {
|
||||
map[string]any{"id": "conf", "type": "file", "path": "/x", "merge": MergeJSON, "content": "{}",
|
||||
TakesSettingsField: "server"},
|
||||
"is a list of setting keys"},
|
||||
"a mesh word": {
|
||||
map[string]any{"id": "conf", "type": "file", "path": "/x", "merge": MergeJSON, "content": "{}",
|
||||
TakesSettingsField: []any{"endpoints"}},
|
||||
"the mesh reads"},
|
||||
"a protected key": {
|
||||
map[string]any{"id": "conf", "type": "file", "path": "/x", "merge": MergeJSON, "content": "{}",
|
||||
"protected": []any{"secret"}, TakesSettingsField: []any{"secret"}},
|
||||
"keeps it as protected"},
|
||||
"named twice": {
|
||||
map[string]any{"id": "conf", "type": "file", "path": "/x", "merge": MergeJSON, "content": "{}",
|
||||
TakesSettingsField: []any{"a", "a"}},
|
||||
"twice"},
|
||||
"not a key": {
|
||||
map[string]any{"id": "conf", "type": "file", "path": "/x", "merge": MergeJSON, "content": "{}",
|
||||
TakesSettingsField: []any{"two words"}},
|
||||
"not a usable setting key"},
|
||||
}
|
||||
for name, c := range cases {
|
||||
problems := TakenSettingsProblems(Manifest{Module: "m", Resources: []map[string]any{c.resource}})
|
||||
if !containsText(problems, c.says) {
|
||||
t.Errorf("%s: want a problem saying %q, got %v", name, c.says, problems)
|
||||
}
|
||||
}
|
||||
// And a file that does not merge is refused where it is settled, not handed to the host.
|
||||
plain := map[string]any{"id": "plain", "type": "file", "path": "/x", "content": "x",
|
||||
TakesSettingsField: []any{"a"}}
|
||||
if _, err := ApplySettings(plain, nil); err == nil {
|
||||
t.Error("a file that does not merge and names keys was settled")
|
||||
}
|
||||
}
|
||||
|
||||
func TestASoftwareSpelledKeyIsUsable(t *testing.T) {
|
||||
// A file's keys are its software's: nodered's timeZone is not lower case and is still a key.
|
||||
m := Manifest{Module: "flows", Resources: []map[string]any{{"id": "settings", "type": "file",
|
||||
"path": "/x", "merge": MergeJSON, "content": "{}", TakesSettingsField: []any{"timeZone"}}}}
|
||||
if problems := TakenSettingsProblems(m); len(problems) != 0 {
|
||||
t.Errorf("timeZone was refused: %v", problems)
|
||||
}
|
||||
}
|
||||
|
||||
func containsText(list []string, text string) bool {
|
||||
for _, s := range list {
|
||||
if strings.Contains(s, text) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
@@ -46,11 +46,24 @@ type Layer struct {
|
||||
// rather than by repeating everything the rest already say.
|
||||
func ApplySettings(resource map[string]any, layers []Layer) (map[string]any, error) {
|
||||
how, _ := resource["merge"].(string)
|
||||
taken, names := takenKeys(resource)
|
||||
if how == "" {
|
||||
if names {
|
||||
// Refused by the module check too (TakenSettingsProblems); said here as well, for a
|
||||
// definition stored by an older binary, rather than handed to the host, which knows no
|
||||
// such field.
|
||||
return nil, fmt.Errorf("%v says %s and does not merge: only a mergeable file takes "+
|
||||
"settings by key (novox/hq ADR 0326)", resource["id"], TakesSettingsField)
|
||||
}
|
||||
// Not a mergeable file. Settings for it are ignored rather than silently doing nothing
|
||||
// somewhere else — see UnusedSettings, which is what says so.
|
||||
return resource, nil
|
||||
}
|
||||
if names {
|
||||
// **The file takes the keys it names, and no other** (novox/hq ADR 0326, issue 168): a key
|
||||
// for another file, for a `${setting:<key>}` or for the mesh itself does not land here.
|
||||
layers = onlyTaken(layers, taken)
|
||||
}
|
||||
if how != MergeJSON {
|
||||
return nil, fmt.Errorf(
|
||||
"%v says it merges as %q, and this control plane knows how to merge %q",
|
||||
@@ -91,6 +104,7 @@ func ApplySettings(resource map[string]any, layers []Layer) (map[string]any, err
|
||||
delete(out, "merge")
|
||||
delete(out, "protected")
|
||||
delete(out, TrustedField)
|
||||
delete(out, TakesSettingsField)
|
||||
return out, nil
|
||||
}
|
||||
|
||||
@@ -324,8 +338,11 @@ func sortedKeysAny(m map[string]any) []string {
|
||||
// (novox/hq 04-ISSUES/173); and the mesh's own words — `expose`, `ports`, `reach`, `endpoints` —
|
||||
// are read by the mesh. A key none of those takes is stray, and is said so rather than dropped.
|
||||
func UnusedSettings(m Manifest, layers []Layer) []string {
|
||||
named := namesItsKeys(m)
|
||||
for _, r := range m.Resources {
|
||||
if how, _ := r["merge"].(string); how != "" {
|
||||
if how, _ := r["merge"].(string); how != "" && !named {
|
||||
// Today's form (novox/hq ADR 0326 §3, ADR 0164 §15): a mergeable file that names no
|
||||
// keys takes every key, so nothing is stray.
|
||||
return nil
|
||||
}
|
||||
}
|
||||
@@ -337,6 +354,13 @@ func UnusedSettings(m Manifest, layers []Layer) []string {
|
||||
}
|
||||
|
||||
lands := settingKeysUsedBy(m)
|
||||
// A mergeable file takes the keys it names (novox/hq ADR 0326 §5).
|
||||
for _, r := range m.Resources {
|
||||
taken, _ := takenKeys(r)
|
||||
for key := range taken {
|
||||
lands[key] = true
|
||||
}
|
||||
}
|
||||
// A key the module gives a default reaches what asks for it (novox/hq ADR 0262): setting it
|
||||
// overrides the default, which is how a running module gains a setting without a gap between.
|
||||
for key := range m.Settings {
|
||||
@@ -401,8 +425,8 @@ func UnusedSettings(m Manifest, layers []Layer) []string {
|
||||
}
|
||||
unused = append(unused, fmt.Sprintf(
|
||||
"%s sets %q, and %s has no file that merges it, asks for no ${setting:%s}, and "+
|
||||
"declares no %q in what it contributes or serves",
|
||||
layer.From, key, m.Module, key, key))
|
||||
"declares no %q in what it contributes or serves%s",
|
||||
layer.From, key, m.Module, key, key, takenSaid(m)))
|
||||
}
|
||||
}
|
||||
sort.Strings(unused)
|
||||
@@ -512,6 +536,11 @@ func JudgeSettings(m Manifest, layers []Layer, adopted bool) error {
|
||||
if err := settingsHoldOneLine(m.Module, layers); err != nil {
|
||||
return err
|
||||
}
|
||||
// Which keys each mergeable file takes is the definition's to say right (novox/hq ADR 0326);
|
||||
// judged here too for a definition stored by an older binary.
|
||||
if problems := takenSettingsProblems(m); len(problems) > 0 {
|
||||
return fmt.Errorf("%s", problems[0])
|
||||
}
|
||||
// With no layers too: a definition may ask for a setting nobody made — an access placed by
|
||||
// nobody, a file's ${setting:…} nothing sets — and that is the same statement, missing.
|
||||
if _, err := GivenPorts(m, layers); err != nil {
|
||||
|
||||
@@ -29,14 +29,13 @@ func ForTest(t *testing.T) *Identity {
|
||||
if len(name) > 60 {
|
||||
name = name[:60]
|
||||
}
|
||||
conn, err := pgx.Connect(t.Context(), admin)
|
||||
// Created already migrated, a copy of a template migrated once (novox/hq issue 591); the Migrate
|
||||
// below then finds nothing pending, which it does only if the copy's record matches every migration.
|
||||
migrations, err := Migrations()
|
||||
if err != nil {
|
||||
t.Fatalf("cannot reach the test PostgreSQL: %v", err)
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := conn.Exec(t.Context(), "create database "+name); err != nil {
|
||||
t.Fatalf("cannot create %s: %v", name, err)
|
||||
}
|
||||
conn.Close(t.Context())
|
||||
store.CreateMigratedForTest(t, admin, name, migrations)
|
||||
|
||||
cut := strings.LastIndex(admin, "/")
|
||||
t.Setenv(store.Variable(Name), admin[:cut]+"/"+name+"?sslmode=disable")
|
||||
@@ -57,10 +56,6 @@ func ForTest(t *testing.T) *Identity {
|
||||
if err := ident.Ready(t.Context(), 20*time.Second); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
migrations, err := Migrations()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := ident.store.Migrate(t.Context(), migrations); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
"encoding/base64"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sync/atomic"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"golang.org/x/crypto/bcrypt"
|
||||
@@ -23,6 +24,25 @@ import (
|
||||
// access change would blank every module's password if the mesh kept nothing (design 25 §4, and the
|
||||
// migration beside this).
|
||||
|
||||
// busPasswordCostForTest, when set, is the bcrypt cost a bus password is hashed at instead of the
|
||||
// library's default. Only ForTest sets it, to bcrypt's least (novox/hq issue 591): at the default
|
||||
// cost one hash takes most of a second under the race detector, and a hash at the least cost is the
|
||||
// same kind of hash, checked by the same comparison and by the bus the same way.
|
||||
var busPasswordCostForTest atomic.Int32
|
||||
|
||||
// busPasswordCost is the bcrypt cost a bus password is hashed at: the library's default, but in a test.
|
||||
func busPasswordCost() int {
|
||||
if c := busPasswordCostForTest.Load(); c != 0 {
|
||||
return int(c)
|
||||
}
|
||||
return bcrypt.DefaultCost
|
||||
}
|
||||
|
||||
// hashBusPassword is the hash a bus password is kept as, minted, recorded or seeded alike.
|
||||
func hashBusPassword(password string) ([]byte, error) {
|
||||
return bcrypt.GenerateFromPassword([]byte(password), busPasswordCost())
|
||||
}
|
||||
|
||||
// BusUser is one user of the bus, as the mesh records it.
|
||||
type BusUser struct {
|
||||
Username string
|
||||
@@ -67,7 +87,7 @@ func (i *Inventory) RecordBusPassword(ctx context.Context, u BusUser, password s
|
||||
if password == "" {
|
||||
return errors.New("a bus user needs a password")
|
||||
}
|
||||
hash, err := bcrypt.GenerateFromPassword([]byte(password), bcrypt.DefaultCost)
|
||||
hash, err := hashBusPassword(password)
|
||||
if err != nil {
|
||||
return fmt.Errorf("cannot hash a bus password: %w", err)
|
||||
}
|
||||
@@ -101,7 +121,7 @@ func (i *Inventory) MintBusPassword(ctx context.Context, u BusUser) (string, err
|
||||
// The cost the server will pay on every connection. Left at the library's default rather than
|
||||
// raised: a node reconnecting after a network blip pays it, and the mesh's own links reconnect
|
||||
// far more often than a person logs in anywhere.
|
||||
hash, err := bcrypt.GenerateFromPassword([]byte(password), bcrypt.DefaultCost)
|
||||
hash, err := hashBusPassword(password)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("cannot hash a bus password: %w", err)
|
||||
}
|
||||
@@ -183,7 +203,7 @@ func (i *Inventory) SeedBusUser(ctx context.Context, u BusUser, password string)
|
||||
if u.Username == "" || u.Kind == "" || password == "" {
|
||||
return errors.New("a bus user needs a username, a kind and the credential it is using")
|
||||
}
|
||||
hash, err := bcrypt.GenerateFromPassword([]byte(password), bcrypt.DefaultCost)
|
||||
hash, err := hashBusPassword(password)
|
||||
if err != nil {
|
||||
return fmt.Errorf("cannot hash a bus password: %w", err)
|
||||
}
|
||||
|
||||
@@ -158,3 +158,17 @@ func TestACredentialTheMeshDidNotMintIsRecordedOnceAndNotOverwritten(t *testing.
|
||||
t.Fatal("a restart put the bootstrap credential back over a rotated one")
|
||||
}
|
||||
}
|
||||
|
||||
// Outside a test a bus password is hashed at bcrypt's default cost: only ForTest lowers it (novox/hq issue 591).
|
||||
func TestABusPasswordIsHashedAtTheDefaultCostOutsideATest(t *testing.T) {
|
||||
was := busPasswordCostForTest.Load()
|
||||
t.Cleanup(func() { busPasswordCostForTest.Store(was) })
|
||||
busPasswordCostForTest.Store(0)
|
||||
hash, err := hashBusPassword("a password")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got, err := bcrypt.Cost(hash); err != nil || got != bcrypt.DefaultCost {
|
||||
t.Fatalf("a bus password is hashed at cost %d (%v), not bcrypt's default %d", got, err, bcrypt.DefaultCost)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,142 @@
|
||||
package inventory
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// A walk whose first machine waited for a busy container counts the wait as a phase of its own, between the
|
||||
// send to the first machine and the judging, and its delivery time holds it (novox/hq ADR 0335 §4): the
|
||||
// recorded walk again, its judging started only at 17:54:07, a minute after the send, when the machine said
|
||||
// the container replaced.
|
||||
func TestABusyWaitIsAPhaseOfTheDeliveryTime(t *testing.T) {
|
||||
raw := strings.Replace(recordedWalk, `"gate": {"machines": ["novox"], "since": "2026-10-10T17:53:07.287136827Z",`,
|
||||
`"gate": {"machines": ["novox"], "since": "2026-10-10T17:54:07.287136827Z",
|
||||
"busy": {"machine": "novox", "module": "mesh-delivery", "container": "mesh-delivery",
|
||||
"since": "2026-10-10T17:53:07Z", "until": "2026-10-10T18:08:07Z", "ended": "2026-10-10T17:54:07.287136827Z"},`, 1)
|
||||
w := walkOf(t, raw).WalkPhases(time.Date(2026, 10, 10, 18, 0, 0, 0, time.UTC), nil)
|
||||
sumsUp(t, w)
|
||||
if w.TotalMS != 282894 {
|
||||
t.Fatalf("the wait changed the delivery's total: %dms", w.TotalMS)
|
||||
}
|
||||
busy := phase(w, PhaseBusy, 0)
|
||||
if busy.State != PhaseMeasured || busy.TookMS != 60000 {
|
||||
t.Fatalf("the busy wait is not a measured phase of a minute: %+v", busy)
|
||||
}
|
||||
if judged := phase(w, PhaseJudge, 0); judged.TookMS != 148607-60000 {
|
||||
t.Fatalf("the judging still counts from the send, not from the replacement: %+v", judged)
|
||||
}
|
||||
var names []string
|
||||
for _, ph := range w.Phases {
|
||||
names = append(names, ph.Name)
|
||||
}
|
||||
if !strings.Contains(strings.Join(names, " "), PhaseSend+" "+PhaseBusy+" "+PhaseJudge) {
|
||||
t.Fatalf("the busy wait is not between the send and the judging: %v", names)
|
||||
}
|
||||
}
|
||||
|
||||
// While the wait goes on, the walk says it waits for a busy container there; a walk where nothing waited has no
|
||||
// such phase.
|
||||
func TestAnOpenBusyWaitIsSaid(t *testing.T) {
|
||||
raw := strings.Replace(recordedWalk, `"gate": {"machines": ["novox"], "since": "2026-10-10T17:53:07.287136827Z", "passes": 3,
|
||||
"verdict": "passed", "judged_at": "2026-10-10T17:55:35.894985053Z"}`,
|
||||
`"gate": {"machines": ["novox"], "since": "2026-10-10T17:53:07.287136827Z",
|
||||
"busy": {"machine": "novox", "module": "mesh-delivery", "container": "mesh-delivery",
|
||||
"since": "2026-10-10T17:53:07Z", "until": "2026-10-10T18:08:07Z"}}`, 1)
|
||||
raw = strings.Replace(raw, `"state": "done"`, `"state": "rolling"`, 1)
|
||||
p := walkOf(t, raw)
|
||||
if !p.Modules["mesh-delivery"].Gate.Waiting() {
|
||||
t.Fatal("a gate whose busy wait has not ended does not say it waits")
|
||||
}
|
||||
w := p.WalkPhases(time.Date(2026, 10, 10, 17, 58, 0, 0, time.UTC), nil)
|
||||
busy := phase(w, PhaseBusy, 0)
|
||||
if busy.State != PhaseUnknown || !strings.Contains(busy.Said, "waits for a busy container on novox") {
|
||||
t.Fatalf("an open busy wait is not said: %+v", busy)
|
||||
}
|
||||
if plain := walkOf(t, recordedWalk).WalkPhases(time.Date(2026, 10, 10, 18, 0, 0, 0, time.UTC), nil); phase(plain, PhaseBusy, 0).Name != "" {
|
||||
t.Fatalf("a walk where nothing waited has a busy phase: %+v", plain.Phases)
|
||||
}
|
||||
}
|
||||
|
||||
// A container a report lists waiting is its module's by its id; its output alone is not a new account.
|
||||
func TestABusyContainerIsItsModules(t *testing.T) {
|
||||
list := []BusyWait{{ID: "build-agent.server", Container: "mesh-build-agent", State: BusyWaiting, Said: "a"},
|
||||
{ID: "web.server", State: "replaced-when-idle"}}
|
||||
if b, ok := WaitingFor(list, "build-agent"); !ok || b.Container != "mesh-build-agent" {
|
||||
t.Fatalf("build-agent's container is not found waiting: %+v", b)
|
||||
}
|
||||
if _, ok := WaitingFor(list, "web"); ok {
|
||||
t.Error("a container replaced reads as waiting")
|
||||
}
|
||||
if _, ok := WaitingFor(list, "build"); ok {
|
||||
t.Error("another module's container is found by a prefix of its name")
|
||||
}
|
||||
again := []BusyWait{{ID: "build-agent.server", State: BusyWaiting, Said: "b, a minute on"}, list[1]}
|
||||
if !sameBusy(list, again) {
|
||||
t.Error("a wait whose output changed is a new account")
|
||||
}
|
||||
if sameBusy(list, list[1:]) || !(Doing{Busy: list}).Waiting() || (Doing{Busy: list[1:]}).Waiting() {
|
||||
t.Error("a wait that ended is not a new account")
|
||||
}
|
||||
}
|
||||
|
||||
// A machine's report of a busy container is kept with its last account and read back beside it, and its
|
||||
// starting and its end are news; its output alone is not (novox/hq ADR 0335). Needs the store.
|
||||
func TestAReportOfABusyContainerIsKeptAndReadBack(t *testing.T) {
|
||||
inv, node := aNodeWithModules(t)
|
||||
ctx := t.Context()
|
||||
record, err := inv.NodeByName(ctx, node)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
since := time.Date(2026, 10, 13, 1, 0, 0, 0, time.UTC)
|
||||
waiting := []BusyWait{{ID: "build-agent.server", Container: "mesh-build-agent", State: BusyWaiting,
|
||||
Since: since, Until: since.Add(15 * time.Minute), Said: "busy: a check"}}
|
||||
keep := func(busy []BusyWait) bool {
|
||||
t.Helper()
|
||||
news, err := inv.RecordOrderedDoing(ctx, record.ID, Doing{Outcome: OutcomeApplied, Declared: "d1", Applied: 1,
|
||||
Busy: busy}, ReportOrder{Epoch: 1, Sequence: 1}, func(ReportOrder) bool { return false })
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return news
|
||||
}
|
||||
read := func() []BusyWait {
|
||||
t.Helper()
|
||||
reports, err := inv.LastReports(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, r := range reports {
|
||||
if r.Node == node {
|
||||
return r.Busy
|
||||
}
|
||||
}
|
||||
t.Fatal("the machine's report is not read back")
|
||||
return nil
|
||||
}
|
||||
keep(nil)
|
||||
if got := read(); len(got) != 0 {
|
||||
t.Fatalf("a report with nothing busy reads back %+v", got)
|
||||
}
|
||||
if !keep(waiting) {
|
||||
t.Error("a container starting to wait is not news")
|
||||
}
|
||||
got := read()
|
||||
if b, ok := WaitingFor(got, "build-agent"); !ok || !b.Since.Equal(since) || b.Said != "busy: a check" {
|
||||
t.Fatalf("the wait is not read back: %+v", got)
|
||||
}
|
||||
again := []BusyWait{waiting[0]}
|
||||
again[0].Said = "busy: a check, a minute on"
|
||||
if keep(again) {
|
||||
t.Error("a wait whose output changed is news")
|
||||
}
|
||||
if !keep([]BusyWait{{ID: "build-agent.server", Container: "mesh-build-agent", State: "replaced-when-idle",
|
||||
Since: since, Until: since.Add(15 * time.Minute)}}) {
|
||||
t.Error("a container replaced after it waited is not news")
|
||||
}
|
||||
if _, ok := WaitingFor(read(), "build-agent"); ok {
|
||||
t.Error("a container replaced still reads as waiting")
|
||||
}
|
||||
}
|
||||
@@ -9,6 +9,7 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"golang.org/x/crypto/bcrypt"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/store"
|
||||
)
|
||||
@@ -30,14 +31,14 @@ func ForTest(t *testing.T) *Inventory {
|
||||
name = name[:60]
|
||||
}
|
||||
|
||||
conn, err := pgx.Connect(t.Context(), admin)
|
||||
// Created already migrated, a copy of a template migrated once (novox/hq issue 591); the Migrate
|
||||
// below then finds nothing pending, which it does only if the copy's record matches every migration.
|
||||
migrations, err := Migrations()
|
||||
if err != nil {
|
||||
t.Fatalf("cannot reach the test PostgreSQL: %v", err)
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := conn.Exec(t.Context(), "create database "+name); err != nil {
|
||||
t.Fatalf("cannot create %s: %v", name, err)
|
||||
}
|
||||
conn.Close(t.Context())
|
||||
store.CreateMigratedForTest(t, admin, name, migrations)
|
||||
busPasswordCostForTest.Store(int32(bcrypt.MinCost))
|
||||
|
||||
cut := strings.LastIndex(admin, "/")
|
||||
t.Setenv(store.Variable(Name), admin[:cut]+"/"+name+"?sslmode=disable")
|
||||
@@ -59,10 +60,6 @@ func ForTest(t *testing.T) *Inventory {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
migrations, err := Migrations()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := inv.store.Migrate(t.Context(), migrations); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,74 @@
|
||||
package inventory
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
)
|
||||
|
||||
// MachineAccount is what a machine said it is made of (novox/hq issue 160), from the profile it last
|
||||
// reported: the architecture and kernel family the profile always carried, and the facts a node-engine that
|
||||
// knows how says beside them. Read by `node`, so what a machine says of itself is visible in one place —
|
||||
// collected and shown in the same breath, not collected and read by nothing (issue 160's pattern).
|
||||
type MachineAccount struct {
|
||||
Architecture string `json:"architecture"`
|
||||
Kernel string `json:"kernel"`
|
||||
// Machine is nil when the node-engine that reported is older than the one that says it.
|
||||
Machine *MadeOf `json:"machine"`
|
||||
}
|
||||
|
||||
// MadeOf is the node-engine's account of what it is made of, as it put it on the wire. A fact it could
|
||||
// not determine is absent here and named in Unknown with why — never a zero.
|
||||
type MadeOf struct {
|
||||
ObservedAt time.Time `json:"observed_at"`
|
||||
KernelRelease string `json:"kernel_release"`
|
||||
Distribution string `json:"distribution"`
|
||||
Version string `json:"version"`
|
||||
Processor string `json:"processor"`
|
||||
Cores int `json:"cores"`
|
||||
Threads int `json:"threads"`
|
||||
MemoryBytes int64 `json:"memory_bytes"`
|
||||
Virtual string `json:"virtual"`
|
||||
BootedAt *time.Time `json:"booted_at"`
|
||||
Timezone string `json:"timezone"`
|
||||
Filesystems []MountedFilesystem `json:"filesystems"`
|
||||
Unknown map[string]string `json:"unknown"`
|
||||
}
|
||||
|
||||
// MountedFilesystem is one local filesystem a machine has mounted.
|
||||
type MountedFilesystem struct {
|
||||
Mount string `json:"mount"`
|
||||
Device string `json:"device"`
|
||||
Type string `json:"type"`
|
||||
SizeBytes int64 `json:"size_bytes"`
|
||||
FreeBytes int64 `json:"free_bytes"`
|
||||
}
|
||||
|
||||
// MachineOf is what a node last said it is made of. had is false for a node that has never reported a
|
||||
// profile at all, which is a different answer from one whose node-engine does not say the facts yet.
|
||||
func (i *Inventory) MachineOf(ctx context.Context, nodeName string) (account MachineAccount, had bool, err error) {
|
||||
var raw []byte
|
||||
err = i.store.Pool().QueryRow(ctx, `select profile from node where name = $1`, nodeName).Scan(&raw)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return MachineAccount{}, false, fmt.Errorf("%w: %s", ErrNoSuchNode, nodeName)
|
||||
}
|
||||
if err != nil {
|
||||
return MachineAccount{}, false, err
|
||||
}
|
||||
return machineFrom(raw)
|
||||
}
|
||||
|
||||
func machineFrom(raw []byte) (MachineAccount, bool, error) {
|
||||
if len(raw) == 0 {
|
||||
return MachineAccount{}, false, nil
|
||||
}
|
||||
var account MachineAccount
|
||||
if err := json.Unmarshal(raw, &account); err != nil {
|
||||
return MachineAccount{}, false, err
|
||||
}
|
||||
return account, true, nil
|
||||
}
|
||||
@@ -0,0 +1,42 @@
|
||||
package inventory
|
||||
|
||||
import (
|
||||
"testing"
|
||||
)
|
||||
|
||||
// The node-engine's words for a machine's facts (novox/hq issue 160), as mesh-host's profile.Machine puts
|
||||
// them on the wire. A rename on either side fails here rather than as a blank on a real machine.
|
||||
func TestAMachinesAccountIsReadInTheNodeEnginesWords(t *testing.T) {
|
||||
raw := []byte(`{"architecture":"amd64","kernel":"linux","capabilities":[],
|
||||
"machine":{"observed_at":"2026-10-12T15:00:00Z","kernel_release":"7.2.8-arch1-2",
|
||||
"distribution":"Arch Linux","version":"rolling","processor":"AMD Ryzen 9","cores":8,"threads":16,
|
||||
"memory_bytes":32891494400,"virtual":"physical","booted_at":"2026-10-09T08:50:34Z",
|
||||
"timezone":"Europe/Brussels",
|
||||
"filesystems":[{"mount":"/","device":"/dev/nvme0n1p3","type":"ext4","size_bytes":100,"free_bytes":40}],
|
||||
"unknown":{"public address":"not decided"}}}`)
|
||||
a, had, err := machineFrom(raw)
|
||||
if err != nil || !had {
|
||||
t.Fatalf("had %v, err %v", had, err)
|
||||
}
|
||||
m := a.Machine
|
||||
if a.Architecture != "amd64" || a.Kernel != "linux" || m == nil {
|
||||
t.Fatalf("%+v", a)
|
||||
}
|
||||
if m.KernelRelease != "7.2.8-arch1-2" || m.Distribution != "Arch Linux" || m.Version != "rolling" ||
|
||||
m.Processor != "AMD Ryzen 9" || m.Cores != 8 || m.Threads != 16 || m.MemoryBytes != 32891494400 ||
|
||||
m.Virtual != "physical" || m.BootedAt == nil || m.Timezone != "Europe/Brussels" ||
|
||||
len(m.Filesystems) != 1 || m.Filesystems[0].FreeBytes != 40 || m.Unknown["public address"] == "" ||
|
||||
m.ObservedAt.IsZero() {
|
||||
t.Errorf("%+v", m)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAProfileWithoutTheFactsIsAnOlderNodeEngineNotNothing(t *testing.T) {
|
||||
a, had, err := machineFrom([]byte(`{"architecture":"amd64","kernel":"linux","capabilities":[]}`))
|
||||
if err != nil || !had || a.Machine != nil || a.Architecture != "amd64" {
|
||||
t.Errorf("%+v had %v err %v", a, had, err)
|
||||
}
|
||||
if _, had, _ := machineFrom(nil); had {
|
||||
t.Error("a node that never reported read as having reported")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,11 @@
|
||||
-- A machine's report says which container it left running because it said busy, and which it replaced
|
||||
-- after it waited (novox/hq ADR 0335, issue 594).
|
||||
--
|
||||
-- A send that replaced the build agent killed the check it ran, and the queue started it again elsewhere.
|
||||
-- A container may now declare `busy`; the node-engine asks it before replacing it, leaves it running while it
|
||||
-- says busy, within 15 minutes, and says so in its report's `busy`: [{"id", "container", "state", "since",
|
||||
-- "until", "said"}], state `waiting`, `replaced-when-idle` or `replaced-at-bound`. Kept with the machine's
|
||||
-- last account, so a walk's first-machine gate waits while its module's container is listed waiting — across
|
||||
-- a restart of the controller, which a merge into mesh-controller makes in the same walk that moves the build
|
||||
-- agent. Empty from an engine that never waits.
|
||||
alter table node_report add column busy jsonb not null default '[]';
|
||||
+97
-13
@@ -10,6 +10,7 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"regexp"
|
||||
"slices"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
@@ -846,6 +847,46 @@ type Doing struct {
|
||||
// machine doing what it was told.
|
||||
Since *time.Time
|
||||
Times int
|
||||
|
||||
// Busy is every container the apply left running because it said busy, and every one it replaced after
|
||||
// it waited (novox/hq ADR 0335).
|
||||
Busy []BusyWait
|
||||
}
|
||||
|
||||
// BusyWait is one container a busy command held, as the machine reported it (novox/hq ADR 0335): its
|
||||
// resource id (`<module>.<resource>`), its name, its state (link.BusyWaiting, BusyReplacedIdle or
|
||||
// BusyReplacedAtBound), since when it waits, until when at the latest, and the last output of its command.
|
||||
type BusyWait struct {
|
||||
ID string `json:"id"`
|
||||
Container string `json:"container"`
|
||||
State string `json:"state"`
|
||||
Since time.Time `json:"since"`
|
||||
Until time.Time `json:"until"`
|
||||
Said string `json:"said,omitempty"`
|
||||
}
|
||||
|
||||
// BusyWaiting is the state of a container left running because it said busy (link.BusyWaiting).
|
||||
const BusyWaiting = "waiting"
|
||||
|
||||
// WaitingFor is the container of a module that waits because it said busy, and whether one does: a
|
||||
// resource whose id is `<module>.<resource>`.
|
||||
func WaitingFor(busy []BusyWait, module string) (BusyWait, bool) {
|
||||
for _, b := range busy {
|
||||
if b.State == BusyWaiting && strings.HasPrefix(b.ID, module+".") {
|
||||
return b, true
|
||||
}
|
||||
}
|
||||
return BusyWait{}, false
|
||||
}
|
||||
|
||||
// Waiting says whether any container waits because it said busy.
|
||||
func (d Doing) Waiting() bool {
|
||||
for _, b := range d.Busy {
|
||||
if b.State == BusyWaiting {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// FailedResource is one thing a node could not do.
|
||||
@@ -873,6 +914,20 @@ func (d Doing) Stuck() bool { return d.Wrong() && d.Times >= StuckAfter && d.Sin
|
||||
// the same failed resources BY ID. Not by the host's words: an error that carries a duration, a
|
||||
// counter or a temporary path would read as new on every report, and the resource looping on it —
|
||||
// which is what stuck is for — would never be said to be (novox/hq 04-ISSUES/065).
|
||||
// sameBusy says two reports hold the same containers in the same state of a busy wait: its output, which
|
||||
// may change on every ask, is not compared.
|
||||
func sameBusy(a, b []BusyWait) bool {
|
||||
key := func(list []BusyWait) []string {
|
||||
out := make([]string, 0, len(list))
|
||||
for _, w := range list {
|
||||
out = append(out, w.ID+"="+w.State)
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out
|
||||
}
|
||||
return slices.Equal(key(a), key(b))
|
||||
}
|
||||
|
||||
func sameFailure(a, b Doing) bool {
|
||||
if a.Outcome != b.Outcome || a.Refused != b.Refused || len(a.Failed) != len(b.Failed) {
|
||||
return false
|
||||
@@ -937,13 +992,20 @@ func recordDoing(ctx context.Context, q queries, node string, d Doing) (news boo
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
if d.Busy == nil {
|
||||
d.Busy = []BusyWait{}
|
||||
}
|
||||
busy, err := json.Marshal(d.Busy)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
var before Doing
|
||||
var beforeFailed []byte
|
||||
var beforeFailed, beforeBusy []byte
|
||||
found := q.QueryRow(ctx,
|
||||
`select outcome, refused, failed, failing_since, failures, coalesce(declared,'')
|
||||
`select outcome, refused, failed, failing_since, failures, coalesce(declared,''), busy
|
||||
from node_report where node = $1`,
|
||||
node).Scan(&before.Outcome, &before.Refused, &beforeFailed, &before.Since, &before.Times,
|
||||
&before.Declared)
|
||||
&before.Declared, &beforeBusy)
|
||||
switch {
|
||||
case errors.Is(found, pgx.ErrNoRows):
|
||||
news = true
|
||||
@@ -953,7 +1015,13 @@ func recordDoing(ctx context.Context, q queries, node string, d Doing) (news boo
|
||||
if err := json.Unmarshal(beforeFailed, &before.Failed); err != nil {
|
||||
return false, err
|
||||
}
|
||||
news = before.Outcome != d.Outcome || before.Declared != d.Declared || !sameFailure(before, d)
|
||||
if err := json.Unmarshal(beforeBusy, &before.Busy); err != nil {
|
||||
return false, err
|
||||
}
|
||||
// A container that starts or ends a wait because it said busy is news (novox/hq ADR 0335): the walk
|
||||
// waiting on this machine is told by it.
|
||||
news = before.Outcome != d.Outcome || before.Declared != d.Declared || !sameFailure(before, d) ||
|
||||
!sameBusy(before.Busy, d.Busy)
|
||||
}
|
||||
var since *time.Time
|
||||
times := 0
|
||||
@@ -966,13 +1034,13 @@ func recordDoing(ctx context.Context, q queries, node string, d Doing) (news boo
|
||||
}
|
||||
_, err = q.Exec(ctx,
|
||||
`insert into node_report (node, outcome, refused, failed, applied, at, declared,
|
||||
failing_since, failures)
|
||||
values ($1, $2, $3, $4, $5, now(), $6, $7, $8)
|
||||
failing_since, failures, busy)
|
||||
values ($1, $2, $3, $4, $5, now(), $6, $7, $8, $9)
|
||||
on conflict (node) do update set outcome = excluded.outcome, refused = excluded.refused,
|
||||
failed = excluded.failed, applied = excluded.applied, at = excluded.at,
|
||||
declared = excluded.declared,
|
||||
failing_since = excluded.failing_since, failures = excluded.failures`,
|
||||
node, d.Outcome, d.Refused, failed, d.Applied, d.Declared, since, times)
|
||||
failing_since = excluded.failing_since, failures = excluded.failures, busy = excluded.busy`,
|
||||
node, d.Outcome, d.Refused, failed, d.Applied, d.Declared, since, times, busy)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
@@ -1059,14 +1127,21 @@ type Reported struct {
|
||||
// gate holds against the send it made, rather than against the send made last (novox/hq issue 352).
|
||||
Declared string
|
||||
ReportedSequence int64
|
||||
// Busy is what the last report said of containers held by their busy command (novox/hq ADR 0335).
|
||||
Busy []BusyWait
|
||||
}
|
||||
|
||||
// SentDeclaration is what one send carried to a machine, as a gate keeps it: the declaration's digest and
|
||||
// its sequence (novox/hq issue 352). A report about this declaration, or about one sequenced after it, is a
|
||||
// report on what the gate sent — whatever the machine was sent since.
|
||||
//
|
||||
// At is when the controller sent it to that machine: the node's own send time, which is where that machine's
|
||||
// `apply` duration starts. A walk measures its rest from it, not from its own record of the send, made only after
|
||||
// every machine of the rest was sent (novox/hq issue 582). Zero for a send kept before it was recorded.
|
||||
type SentDeclaration struct {
|
||||
Digest string `json:"digest"`
|
||||
Sequence int64 `json:"sequence,omitempty"`
|
||||
Digest string `json:"digest"`
|
||||
Sequence int64 `json:"sequence,omitempty"`
|
||||
At time.Time `json:"at,omitzero"`
|
||||
}
|
||||
|
||||
// ReportsOn says a report is about this send: the declaration itself; one the same machine was sequenced
|
||||
@@ -1085,7 +1160,8 @@ func (s SentDeclaration) ReportsOn(r Reported) bool {
|
||||
func (i *Inventory) SentTo(ctx context.Context, name string) (SentDeclaration, bool, error) {
|
||||
var digest *string
|
||||
var seq *int64
|
||||
err := i.store.Pool().QueryRow(ctx, `select sent, sequence from node where name = $1`, name).Scan(&digest, &seq)
|
||||
var at *time.Time
|
||||
err := i.store.Pool().QueryRow(ctx, `select sent, sequence, sent_at from node where name = $1`, name).Scan(&digest, &seq, &at)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return SentDeclaration{}, false, fmt.Errorf("%w: %s", ErrNoSuchNode, name)
|
||||
}
|
||||
@@ -1096,6 +1172,9 @@ func (i *Inventory) SentTo(ctx context.Context, name string) (SentDeclaration, b
|
||||
if seq != nil {
|
||||
s.Sequence = *seq
|
||||
}
|
||||
if at != nil {
|
||||
s.At = at.UTC()
|
||||
}
|
||||
return s, true, nil
|
||||
}
|
||||
|
||||
@@ -1104,7 +1183,7 @@ func (i *Inventory) LastReports(ctx context.Context) ([]Reported, error) {
|
||||
rows, err := i.store.Pool().Query(ctx,
|
||||
`select n.name, coalesce(r.outcome, ''), r.at, n.sent_at,
|
||||
r.declared is not null and r.declared <> '' and r.declared = n.sent,
|
||||
coalesce(r.declared, ''), coalesce(r.reported_sequence, 0)
|
||||
coalesce(r.declared, ''), coalesce(r.reported_sequence, 0), coalesce(r.busy, '[]')
|
||||
from node n left join node_report r on r.node = n.id
|
||||
order by n.name`)
|
||||
if err != nil {
|
||||
@@ -1114,7 +1193,12 @@ func (i *Inventory) LastReports(ctx context.Context) ([]Reported, error) {
|
||||
var out []Reported
|
||||
for rows.Next() {
|
||||
var r Reported
|
||||
if err := rows.Scan(&r.Node, &r.Outcome, &r.At, &r.Sent, &r.Current, &r.Declared, &r.ReportedSequence); err != nil {
|
||||
var busy []byte
|
||||
if err := rows.Scan(&r.Node, &r.Outcome, &r.At, &r.Sent, &r.Current, &r.Declared, &r.ReportedSequence,
|
||||
&busy); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := json.Unmarshal(busy, &r.Busy); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, r)
|
||||
|
||||
@@ -367,6 +367,39 @@ type PlanGate struct {
|
||||
// alone to the machine holding the bus, every build there kept, so what the send newly grants is on the
|
||||
// bus before the gate judges it. Nil when the list did not change, or the gate's machine holds the bus.
|
||||
Grants *GrantsStep `json:"grants,omitempty"`
|
||||
// Busy is the wait for a busy container on a judged machine (novox/hq ADR 0335 §4): while the machine
|
||||
// reports its module's container left running because it said busy, the judging waits and has not
|
||||
// started; once it reports the container replaced, Since is moved to then. Nil when nothing waited.
|
||||
Busy *GateBusy `json:"busy,omitempty"`
|
||||
}
|
||||
|
||||
// GateBusy is a first-node gate's wait for a busy container (novox/hq ADR 0335 §4).
|
||||
type GateBusy struct {
|
||||
Machine string `json:"machine"`
|
||||
Module string `json:"module"`
|
||||
Container string `json:"container"`
|
||||
// Since is when the machine's first apply that would have replaced it ran, Until when its node-engine
|
||||
// replaces it at the latest; Ended when the machine reported it replaced and the judging began — or when
|
||||
// the controller stopped waiting past the engine's bound, which PastBound says.
|
||||
Since time.Time `json:"since"`
|
||||
Until time.Time `json:"until"`
|
||||
Ended *time.Time `json:"ended,omitempty"`
|
||||
PastBound bool `json:"past_bound,omitempty"`
|
||||
// Said is the last output of its busy command, as the machine reported it.
|
||||
Said string `json:"said,omitempty"`
|
||||
}
|
||||
|
||||
// Waiting says the gate waits for a busy container and has not started.
|
||||
func (g *PlanGate) Waiting() bool { return g != nil && g.Busy != nil && g.Busy.Ended == nil }
|
||||
|
||||
// Words is the wait in a line, as the walk says it.
|
||||
func (b *GateBusy) Words() string {
|
||||
said := ""
|
||||
if b.Said != "" {
|
||||
said = "; its busy command says: " + b.Said
|
||||
}
|
||||
return fmt.Sprintf("waits for a busy container on %s: %s holds work, and is replaced once it says idle, by %s "+
|
||||
"at the latest (novox/hq ADR 0335)%s", b.Machine, b.Container, b.Until.UTC().Format(time.RFC3339), said)
|
||||
}
|
||||
|
||||
// GrantsStep is the bus's user list sent to the machine holding the bus ahead of a gated send (novox/hq
|
||||
|
||||
@@ -2,13 +2,10 @@ package inventory
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
)
|
||||
|
||||
// Where a walk's time went (novox/hq ADR 0282 decision 6, issue 382): from its merge to every machine of its
|
||||
@@ -29,6 +26,7 @@ const (
|
||||
PhaseBetween = "between-tiers" // one tier's end to the next tier's ask
|
||||
PhaseBuild = "build" // a tier asked to its last module built
|
||||
PhaseSend = "send-first" // built to sent to the first machines
|
||||
PhaseBusy = "busy-wait" // sent first to a busy container there replaced, when one waited (ADR 0335)
|
||||
PhaseJudge = "judgement" // sent first to the first-node gate's last verdict: its readings
|
||||
PhaseRest = "send-rest" // judged to sent to the rest
|
||||
PhaseApply = "apply" // the last send to every machine of the rest reporting the build applied
|
||||
@@ -59,11 +57,14 @@ type WalkPhases struct {
|
||||
TotalMS int64 `json:"total_ms,omitempty"`
|
||||
Total string `json:"total,omitempty"`
|
||||
// UnknownMS is the time within the walk no phase could be measured over.
|
||||
UnknownMS int64 `json:"unknown_ms,omitempty"`
|
||||
Phases []WalkPhase `json:"phases"`
|
||||
Silent []string `json:"silent,omitempty"`
|
||||
Said string `json:"said"`
|
||||
Merges []MergeStart `json:"merges,omitempty"`
|
||||
UnknownMS int64 `json:"unknown_ms,omitempty"`
|
||||
Phases []WalkPhase `json:"phases"`
|
||||
Silent []string `json:"silent,omitempty"`
|
||||
// Rest is each machine of the rest's own report of the send that carried the walk's build there (novox/hq ADR
|
||||
// 0282 decision 6, issue 582): when it was sent, when it said, and what.
|
||||
Rest []RestReport `json:"rest,omitempty"`
|
||||
Said string `json:"said"`
|
||||
Merges []MergeStart `json:"merges,omitempty"`
|
||||
}
|
||||
|
||||
// MergeStart is one merge the walk answers and when it was made: where that merge's delivery time starts.
|
||||
@@ -92,8 +93,34 @@ type AppliedReport struct {
|
||||
Outcome string
|
||||
}
|
||||
|
||||
// AppliedLookup answers a machine's first report after a send to it; false when it has not reported.
|
||||
type AppliedLookup func(node string, sent time.Time) (AppliedReport, bool)
|
||||
// AppliedLookup answers a machine's report of the send the walk made to it: send is what the walk kept of it, and
|
||||
// walkSent the walk's own record of its send to the rest; false when it has not reported.
|
||||
type AppliedLookup func(node string, send SentDeclaration, walkSent time.Time) (AppliedReport, bool)
|
||||
|
||||
// RestReport is one machine of the rest and its own report of a send of the walk (novox/hq issue 582).
|
||||
type RestReport struct {
|
||||
Node string `json:"node"`
|
||||
Modules []string `json:"modules"`
|
||||
// Sent is when that machine was sent, by its own send time where the walk kept it; else the walk's record of
|
||||
// its send to the rest, and SentSaid says so.
|
||||
Sent time.Time `json:"sent"`
|
||||
SentSaid string `json:"sent_said,omitempty"`
|
||||
Sequence int64 `json:"sequence,omitempty"`
|
||||
Reported *time.Time `json:"reported,omitempty"`
|
||||
Took string `json:"took,omitempty"`
|
||||
Outcome string `json:"outcome,omitempty"`
|
||||
// State is applied, failed, waiting (not yet heard, within ApplySilentAfter) or silent (not heard within it:
|
||||
// listed, never moving the walk's end).
|
||||
State string `json:"state"`
|
||||
}
|
||||
|
||||
// The states of a machine of the rest.
|
||||
const (
|
||||
RestApplied = "applied"
|
||||
RestFailed = "failed"
|
||||
RestWaiting = "waiting"
|
||||
RestSilent = "silent"
|
||||
)
|
||||
|
||||
// point is one moment of the walk, ending the phase named.
|
||||
type point struct {
|
||||
@@ -274,8 +301,11 @@ func (p Plan) points(now time.Time, applied AppliedLookup, w *WalkPhases) []poin
|
||||
if t > p.Tier || (t == p.Tier && p.Tier < len(p.Tiers) && !askedAnyOf(p, tier)) {
|
||||
break
|
||||
}
|
||||
var asked, built, first, judged, rest *time.Time
|
||||
var asked, built, first, judged, rest, busyEnded *time.Time
|
||||
allBuilt, anyFirst, allJudged, allRest := true, false, true, true
|
||||
// A busy container on a first machine holds the judging until it is replaced (novox/hq ADR 0335 §4): a
|
||||
// phase of its own, only in a tier where one waited, counted in the delivery time and never exempted.
|
||||
anyBusy, allBusyEnded, busyOn := false, true, ""
|
||||
for _, m := range tier {
|
||||
s := p.Modules[m]
|
||||
if s == nil {
|
||||
@@ -293,6 +323,14 @@ func (p Plan) points(now time.Time, applied AppliedLookup, w *WalkPhases) []poin
|
||||
if s.FirstAt != nil {
|
||||
anyFirst = true
|
||||
first = earliest(first, s.FirstAt)
|
||||
if s.Gate != nil && s.Gate.Busy != nil {
|
||||
anyBusy = true
|
||||
if s.Gate.Busy.Ended == nil {
|
||||
allBusyEnded, busyOn = false, s.Gate.Busy.Machine
|
||||
} else {
|
||||
busyEnded = latest(busyEnded, s.Gate.Busy.Ended)
|
||||
}
|
||||
}
|
||||
if s.Gate != nil {
|
||||
if s.Gate.JudgedAt == nil {
|
||||
allJudged = false
|
||||
@@ -307,8 +345,8 @@ func (p Plan) points(now time.Time, applied AppliedLookup, w *WalkPhases) []poin
|
||||
}
|
||||
} else {
|
||||
rest = latest(rest, s.SentAt)
|
||||
for node := range s.Rest {
|
||||
lastSends = append(lastSends, restSend{module: m, node: node, at: *s.SentAt})
|
||||
for node, what := range s.Rest {
|
||||
lastSends = append(lastSends, restSend{module: m, node: node, walkSent: *s.SentAt, what: what})
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -323,7 +361,15 @@ func (p Plan) points(now time.Time, applied AppliedLookup, w *WalkPhases) []poin
|
||||
if !allJudged {
|
||||
judged = nil
|
||||
}
|
||||
out = append(out, point{phase: PhaseSend, tier: t, at: first}, point{phase: PhaseJudge, tier: t, at: judged})
|
||||
out = append(out, point{phase: PhaseSend, tier: t, at: first})
|
||||
if anyBusy {
|
||||
said := ""
|
||||
if !allBusyEnded {
|
||||
busyEnded, said = nil, "waits for a busy container on "+busyOn
|
||||
}
|
||||
out = append(out, point{phase: PhaseBusy, tier: t, at: busyEnded, said: said})
|
||||
}
|
||||
out = append(out, point{phase: PhaseJudge, tier: t, at: judged})
|
||||
} else {
|
||||
out = append(out, point{phase: PhaseSend, tier: t, none: true, said: "no first machine: nothing to judge"},
|
||||
point{phase: PhaseJudge, tier: t, none: true, said: "no first machine: nothing to judge"})
|
||||
@@ -347,22 +393,73 @@ func (p Plan) points(now time.Time, applied AppliedLookup, w *WalkPhases) []poin
|
||||
if applied == nil {
|
||||
return append(out, point{phase: PhaseApply, tier: -1, said: "the rest's reports were not read"})
|
||||
}
|
||||
var end *time.Time
|
||||
var waiting, failed []string
|
||||
// Each machine of the rest, by its own report of the send that carried the build there (novox/hq issue 582):
|
||||
// never a report of a send made after it, which another walk's send may be.
|
||||
reports := map[string]*RestReport{}
|
||||
var order []string
|
||||
for _, s := range lastSends {
|
||||
r, ok := applied(s.node, s.at)
|
||||
switch {
|
||||
case !ok && now.Sub(s.at) > ApplySilentAfter, ok && r.At.Sub(s.at) > ApplySilentAfter:
|
||||
w.Silent = appendOnce(w.Silent, s.node)
|
||||
case !ok:
|
||||
waiting = appendOnce(waiting, s.node)
|
||||
case r.Outcome != OutcomeApplied:
|
||||
failed = appendOnce(failed, s.node+" ("+r.Outcome+")")
|
||||
default:
|
||||
at := r.At
|
||||
end = latest(end, &at)
|
||||
at, said := s.walkSent, ""
|
||||
if !s.what.At.IsZero() {
|
||||
at = s.what.At
|
||||
} else {
|
||||
said = "the walk's record of its send to the rest: this machine's own send time was not kept"
|
||||
}
|
||||
key := s.node + "\x00" + s.what.Digest + "\x00" + at.Format(time.RFC3339Nano)
|
||||
if r := reports[key]; r != nil {
|
||||
r.Modules = appendOnce(r.Modules, s.module)
|
||||
continue
|
||||
}
|
||||
rr := &RestReport{Node: s.node, Modules: []string{s.module}, Sent: at, SentSaid: said, Sequence: s.what.Sequence}
|
||||
if r, ok := applied(s.node, s.what, s.walkSent); ok {
|
||||
got := r.At
|
||||
rr.Reported, rr.Outcome, rr.Took = &got, r.Outcome, words(r.At.Sub(at))
|
||||
}
|
||||
reports[key] = rr
|
||||
order = append(order, key)
|
||||
}
|
||||
// A machine sent twice by the walk (one tier's rest, then the next's) that said nothing of the first send has
|
||||
// said it of the second, which carries the first's builds too: both are this walk's.
|
||||
for _, k := range order {
|
||||
rr := reports[k]
|
||||
if rr.Reported != nil {
|
||||
continue
|
||||
}
|
||||
for _, l := range order {
|
||||
later := reports[l]
|
||||
if later.Node == rr.Node && later.Sent.After(rr.Sent) && later.Reported != nil &&
|
||||
(rr.Reported == nil || later.Reported.Before(*rr.Reported)) {
|
||||
got := *later.Reported
|
||||
rr.Reported, rr.Outcome, rr.Took = &got, later.Outcome, words(got.Sub(rr.Sent))
|
||||
}
|
||||
}
|
||||
}
|
||||
var end *time.Time
|
||||
var waiting, failed []string
|
||||
for _, k := range order {
|
||||
rr := reports[k]
|
||||
switch {
|
||||
case rr.Reported == nil && now.Sub(rr.Sent) > ApplySilentAfter, rr.Reported != nil && rr.Reported.Sub(rr.Sent) > ApplySilentAfter:
|
||||
rr.State = RestSilent
|
||||
w.Silent = appendOnce(w.Silent, rr.Node)
|
||||
case rr.Reported == nil:
|
||||
rr.State = RestWaiting
|
||||
waiting = appendOnce(waiting, rr.Node)
|
||||
case rr.Outcome != OutcomeApplied:
|
||||
rr.State = RestFailed
|
||||
failed = appendOnce(failed, rr.Node+" ("+rr.Outcome+")")
|
||||
default:
|
||||
rr.State = RestApplied
|
||||
end = latest(end, rr.Reported)
|
||||
}
|
||||
sort.Strings(rr.Modules)
|
||||
w.Rest = append(w.Rest, *rr)
|
||||
}
|
||||
sort.SliceStable(w.Rest, func(a, b int) bool {
|
||||
if w.Rest[a].Node != w.Rest[b].Node {
|
||||
return w.Rest[a].Node < w.Rest[b].Node
|
||||
}
|
||||
return w.Rest[a].Sent.Before(w.Rest[b].Sent)
|
||||
})
|
||||
switch {
|
||||
case len(failed) > 0:
|
||||
return append(out, point{phase: PhaseApply, tier: -1, said: "not applied on " + strings.Join(failed, ", ")})
|
||||
@@ -393,7 +490,8 @@ func (p Plan) points(now time.Time, applied AppliedLookup, w *WalkPhases) []poin
|
||||
|
||||
type restSend struct {
|
||||
module, node string
|
||||
at time.Time
|
||||
walkSent time.Time
|
||||
what SentDeclaration
|
||||
}
|
||||
|
||||
func askedAnyOf(p Plan, tier []string) bool {
|
||||
@@ -451,28 +549,79 @@ func words(d time.Duration) string {
|
||||
return d.Round(100 * time.Millisecond).String()
|
||||
}
|
||||
|
||||
// FirstAppliedAfter is a machine's first report of a send made at or after a walk's send to it, within
|
||||
// ApplySilentAfter of it — the controller's `apply` durations, which measure every send to its first report (to-be
|
||||
// 45 Phase 0). A declaration sent later carries the build too, so its report counts; one sent past the bound is a
|
||||
// machine that slept, listed as silent and never moving the walk's end (ADR 0282 decision 1).
|
||||
func (i *Inventory) FirstAppliedAfter(ctx context.Context, node string, sent time.Time) (AppliedReport, bool, error) {
|
||||
var started time.Time
|
||||
var ms int64
|
||||
var outcome string
|
||||
err := i.store.Pool().QueryRow(ctx,
|
||||
`select started, took_ms, detail from duration
|
||||
// FirstAppliedAfter is a machine's report of the send a walk made to it: the controller's `apply` durations, which
|
||||
// measure every send to its first report (to-be 45 Phase 0), each row starting at the machine's own send time and
|
||||
// naming the declaration it was about. send is what the walk kept of that send; walkSent its own record of the
|
||||
// send to the rest, made after every machine of the rest was sent (novox/hq issue 582). FirstReportOf chooses.
|
||||
func (i *Inventory) FirstAppliedAfter(ctx context.Context, node string, send SentDeclaration, walkSent time.Time) (AppliedReport, bool, error) {
|
||||
from, to := applyRowsBetween(send, walkSent)
|
||||
rows, err := i.store.Pool().Query(ctx,
|
||||
`select ref, started, took_ms, detail from duration
|
||||
where kind = $1 and subject = $2 and started >= $3 and started < $4
|
||||
order by started limit 1`,
|
||||
DurationApply, node, sent.Add(-appliedSlack), sent.Add(ApplySilentAfter)).Scan(&started, &ms, &outcome)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return AppliedReport{}, false, nil
|
||||
}
|
||||
order by started`,
|
||||
DurationApply, node, from, to)
|
||||
if err != nil {
|
||||
return AppliedReport{}, false, err
|
||||
}
|
||||
return AppliedReport{At: started.Add(time.Duration(ms) * time.Millisecond).UTC(), Outcome: outcome}, true, nil
|
||||
defer rows.Close()
|
||||
var ds []Duration
|
||||
for rows.Next() {
|
||||
d := Duration{Kind: DurationApply, Subject: node, Node: node}
|
||||
var ms int64
|
||||
if err := rows.Scan(&d.Ref, &d.Started, &ms, &d.Detail); err != nil {
|
||||
return AppliedReport{}, false, err
|
||||
}
|
||||
d.Took = time.Duration(ms) * time.Millisecond
|
||||
ds = append(ds, d)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return AppliedReport{}, false, err
|
||||
}
|
||||
r, ok := FirstReportOf(ds, send, walkSent)
|
||||
return r, ok, nil
|
||||
}
|
||||
|
||||
// appliedSlack is how much earlier than a walk's record of its send the machine's own record of it may be:
|
||||
// the send is recorded on the machine first, then on the walk.
|
||||
// applyRowsBetween is the span of a machine's `apply` durations that may hold its report of a walk's send.
|
||||
func applyRowsBetween(send SentDeclaration, walkSent time.Time) (time.Time, time.Time) {
|
||||
switch {
|
||||
case !send.At.IsZero():
|
||||
return send.At.Add(-time.Millisecond), send.At.Add(time.Millisecond)
|
||||
case send.Digest != "":
|
||||
return walkSent.Add(-restSendBound), walkSent.Add(ApplySilentAfter)
|
||||
}
|
||||
return walkSent.Add(-appliedSlack), walkSent.Add(ApplySilentAfter)
|
||||
}
|
||||
|
||||
// FirstReportOf picks, from a machine's `apply` durations ordered by their start, its report of the send a walk made
|
||||
// to it (novox/hq issue 582):
|
||||
// - with the machine's own send time kept (every walk since issue 582): the duration of that very send, and of the
|
||||
// declaration sent where the walk kept its digest. A report of a later send is never this walk's: a send made
|
||||
// later, by another walk, would end this walk at that walk's send.
|
||||
// - with only the declaration's digest kept (a walk kept before): the first duration about that declaration
|
||||
// within restSendBound before the walk's record of its send, the machine sent first in the rest included.
|
||||
// - with neither (a walk kept before ADR 0282's rest): the first duration started no earlier than appliedSlack
|
||||
// before the walk's record of its send, as it was read before.
|
||||
//
|
||||
// A send past ApplySilentAfter is never read: its machine slept, and is listed silent (ADR 0282 decision 1).
|
||||
func FirstReportOf(ds []Duration, send SentDeclaration, walkSent time.Time) (AppliedReport, bool) {
|
||||
from, to := applyRowsBetween(send, walkSent)
|
||||
for _, d := range ds {
|
||||
if d.Started.Before(from) || !d.Started.Before(to) {
|
||||
continue
|
||||
}
|
||||
if send.Digest != "" && !strings.HasPrefix(d.Ref, send.Digest+"@") {
|
||||
continue
|
||||
}
|
||||
return AppliedReport{At: d.Started.Add(d.Took).UTC(), Outcome: d.Detail}, true
|
||||
}
|
||||
return AppliedReport{}, false
|
||||
}
|
||||
|
||||
// appliedSlack is how much earlier than a walk's record of its send the machine's own record of it may be, for a
|
||||
// walk that kept neither the machine's send time nor what it sent.
|
||||
const appliedSlack = 2 * time.Second
|
||||
|
||||
// restSendBound is how long before a walk's record of its send to the rest a machine of the rest may have been
|
||||
// sent, for a walk that kept what it sent and not when: the send to the rest goes machine by machine, and took
|
||||
// 19 to 28 seconds on 2026-10-12 (novox/hq issue 582).
|
||||
const restSendBound = 5 * time.Minute
|
||||
|
||||
@@ -0,0 +1,188 @@
|
||||
package inventory
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// novox/hq issue 582: the walk's own record of its send to the rest is made after every machine of the rest was
|
||||
// sent, and that send took 19 to 28 s on 2026-10-12. The machine sent first reported 5 s after its own send, long
|
||||
// before the walk's record; read from the walk's record, its report was missed and the walk ended at another walk's
|
||||
// send to it, minutes later. A walk of one tier and three machines of the rest, the send to the rest taking 20 s.
|
||||
type sendRest582 struct {
|
||||
judged, walkSent time.Time
|
||||
walk Plan
|
||||
rows map[string][]Duration
|
||||
}
|
||||
|
||||
func walk582(t *testing.T) sendRest582 {
|
||||
t.Helper()
|
||||
judged := time.Date(2026, 10, 12, 17, 45, 35, 137000000, time.UTC)
|
||||
ownSend := map[string]time.Duration{"g14": 500 * time.Millisecond, "novox": 10 * time.Second, "shanks": 20 * time.Second}
|
||||
walkSent := judged.Add(20200 * time.Millisecond)
|
||||
merged := judged.Add(-10 * time.Minute)
|
||||
built, first := judged.Add(-3*time.Minute), judged.Add(-2*time.Minute)
|
||||
rest := map[string]SentDeclaration{}
|
||||
rows := map[string][]Duration{}
|
||||
took := map[string]time.Duration{"g14": 5 * time.Second, "novox": 13 * time.Second, "shanks": 15 * time.Second}
|
||||
for i, n := range []string{"g14", "novox", "shanks"} {
|
||||
at := judged.Add(ownSend[n])
|
||||
digest := "walk-" + n
|
||||
rest[n] = SentDeclaration{Digest: digest, Sequence: int64(647 + i), At: at}
|
||||
rows[n] = append(rows[n],
|
||||
// This walk's send, and its report.
|
||||
Duration{Kind: DurationApply, Subject: n, Node: n, Ref: digest + "@" + at.Format("2006-01-02T15:04:05.000000"),
|
||||
Started: at, Took: took[n], Detail: OutcomeApplied},
|
||||
// The next walk's send to the same machine, nine minutes later, and its report.
|
||||
Duration{Kind: DurationApply, Subject: n, Node: n, Ref: "next-" + n + "@x",
|
||||
Started: judged.Add(9*time.Minute + ownSend[n]), Took: 4 * time.Second, Detail: OutcomeApplied})
|
||||
}
|
||||
p := Plan{ID: "plan-582", State: PlanDone, Tier: 1, Tiers: [][]string{{"build-agent"}}, Created: merged,
|
||||
Delivery: &PlanDelivery{Merges: []PlanMerge{{Repository: "novox/mesh-controller", Commit: "c", Merged: merged}}},
|
||||
Times: &PlanTimes{Cut: &merged, Class: ClassCore},
|
||||
Modules: map[string]*PlanModule{"build-agent": {State: "built", AskedAt: &merged, BuiltAt: &built,
|
||||
FirstAt: &first, SentAt: &walkSent, Gate: &PlanGate{JudgedAt: &judged}, Rest: rest}}}
|
||||
return sendRest582{judged: judged, walkSent: walkSent, walk: p, rows: rows}
|
||||
}
|
||||
|
||||
// lookup answers from the rows as the controller's durations would.
|
||||
func (w sendRest582) lookup(node string, send SentDeclaration, walkSent time.Time) (AppliedReport, bool) {
|
||||
return FirstReportOf(w.rows[node], send, walkSent)
|
||||
}
|
||||
|
||||
// The apply phase ends at the last machine's own report of this walk's send — shanks, sent at the end of the 20 s
|
||||
// and reporting 15 s later — and never at the next walk's send; every machine's report is shown.
|
||||
func TestTheApplyPhaseEndsAtEachMachinesOwnReport(t *testing.T) {
|
||||
c := walk582(t)
|
||||
w := c.walk.WalkPhases(c.judged.Add(time.Hour), c.lookup)
|
||||
sumsUp(t, w)
|
||||
want := c.judged.Add(35 * time.Second) // shanks: sent 20 s after the judgement, reported 15 s later
|
||||
if !w.End.Equal(want) {
|
||||
t.Fatalf("the walk ends at %s, want %s: its rest %+v", w.End, want, w.Rest)
|
||||
}
|
||||
if got := phase(w, PhaseApply, -1); got.State != PhaseMeasured || got.TookMS != want.Sub(c.walkSent).Milliseconds() {
|
||||
t.Fatalf("apply: %+v", got)
|
||||
}
|
||||
if len(w.Rest) != 3 || len(w.Silent) != 0 {
|
||||
t.Fatalf("every machine of the rest is shown with its report: %+v, silent %v", w.Rest, w.Silent)
|
||||
}
|
||||
g14 := w.Rest[0]
|
||||
if g14.Node != "g14" || g14.State != RestApplied || g14.Reported == nil ||
|
||||
!g14.Reported.Equal(c.judged.Add(5500*time.Millisecond)) || g14.Took != "5s" || g14.Sequence != 647 ||
|
||||
!g14.Sent.Equal(c.judged.Add(500*time.Millisecond)) || g14.SentSaid != "" {
|
||||
t.Fatalf("g14, sent first and reporting 5 s after its own send: %+v", g14)
|
||||
}
|
||||
}
|
||||
|
||||
// A walk kept before its machines' own send times were: matched by the declaration it sent, within restSendBound
|
||||
// of its record of the send, so the machine sent first is still found; a walk that kept neither is read as before.
|
||||
func TestAnOlderWalkIsReadByWhatItSent(t *testing.T) {
|
||||
c := walk582(t)
|
||||
for n, s := range c.walk.Modules["build-agent"].Rest {
|
||||
s.At = time.Time{}
|
||||
c.walk.Modules["build-agent"].Rest[n] = s
|
||||
}
|
||||
w := c.walk.WalkPhases(c.judged.Add(time.Hour), c.lookup)
|
||||
sumsUp(t, w)
|
||||
if !w.End.Equal(c.judged.Add(35 * time.Second)) {
|
||||
t.Fatalf("by the declaration sent: end %s, rest %+v", w.End, w.Rest)
|
||||
}
|
||||
if !strings.Contains(w.Rest[0].SentSaid, "walk's record") || !w.Rest[0].Sent.Equal(c.walkSent) {
|
||||
t.Fatalf("a send time not kept is said: %+v", w.Rest[0])
|
||||
}
|
||||
// Nothing of the send kept: the first report no earlier than appliedSlack before the walk's record, as before.
|
||||
if r, ok := FirstReportOf(c.rows["shanks"], SentDeclaration{}, c.walkSent); !ok || !r.At.Equal(c.judged.Add(35*time.Second)) {
|
||||
t.Fatalf("a walk that kept nothing of its send: %+v %v", r, ok)
|
||||
}
|
||||
}
|
||||
|
||||
// A machine that said nothing of this walk's send is waiting, then silent past ApplySilentAfter: the report of a
|
||||
// later send to it is never taken as this walk's, and never moves the walk's end.
|
||||
func TestALaterSendsReportIsNeverThisWalks(t *testing.T) {
|
||||
c := walk582(t)
|
||||
c.rows["g14"] = c.rows["g14"][1:] // g14 slept through this walk's send; the next walk's send it reported
|
||||
w := c.walk.WalkPhases(c.judged.Add(5*time.Minute), c.lookup)
|
||||
if w.End != nil || !strings.Contains(phase(w, PhaseApply, -1).Said, "waiting for g14") {
|
||||
t.Fatalf("g14 not heard from yet: end %v, apply %+v", w.End, phase(w, PhaseApply, -1))
|
||||
}
|
||||
w = c.walk.WalkPhases(c.judged.Add(time.Hour), c.lookup)
|
||||
sumsUp(t, w)
|
||||
if len(w.Silent) != 1 || w.Silent[0] != "g14" || !w.End.Equal(c.judged.Add(35*time.Second)) {
|
||||
t.Fatalf("g14 silent, the end at shanks' report: silent %v, end %s", w.Silent, w.End)
|
||||
}
|
||||
if w.Rest[0].State != RestSilent || w.Rest[0].Reported != nil {
|
||||
t.Fatalf("g14's later report taken as this walk's: %+v", w.Rest[0])
|
||||
}
|
||||
}
|
||||
|
||||
// A machine sent by two tiers of one walk that said nothing of the first send has said it of the second, which
|
||||
// carries the first tier's build too.
|
||||
func TestAMachineSentTwiceByOneWalkIsHeardByItsLaterSend(t *testing.T) {
|
||||
c := walk582(t)
|
||||
second := c.judged.Add(4 * time.Minute)
|
||||
secondSent := second.Add(time.Second)
|
||||
c.walk.Tiers = append(c.walk.Tiers, []string{"route-proxy"})
|
||||
c.walk.Tier = 2
|
||||
asked, built := c.judged.Add(time.Minute), c.judged.Add(2*time.Minute)
|
||||
first := built.Add(time.Second)
|
||||
c.walk.Modules["route-proxy"] = &PlanModule{State: "built", AskedAt: &asked, BuiltAt: &built, FirstAt: &first,
|
||||
SentAt: &secondSent, Gate: &PlanGate{JudgedAt: &second},
|
||||
Rest: map[string]SentDeclaration{"novox": {Digest: "walk-novox-2", Sequence: 1554, At: second.Add(500 * time.Millisecond)}}}
|
||||
c.rows["novox"] = []Duration{{Kind: DurationApply, Subject: "novox", Ref: "walk-novox-2@x",
|
||||
Started: second.Add(500 * time.Millisecond), Took: 14 * time.Second, Detail: OutcomeApplied}}
|
||||
w := c.walk.WalkPhases(c.judged.Add(time.Hour), c.lookup)
|
||||
sumsUp(t, w)
|
||||
want := second.Add(14500 * time.Millisecond)
|
||||
if !w.End.Equal(want) || len(w.Silent) != 0 {
|
||||
t.Fatalf("end %s (want %s), silent %v, rest %+v", w.End, want, w.Silent, w.Rest)
|
||||
}
|
||||
for _, r := range w.Rest {
|
||||
if r.Node == "novox" && (r.State != RestApplied || !r.Reported.Equal(want)) {
|
||||
t.Fatalf("novox, heard by its later send of this walk: %+v", r)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// The store: a machine's own send time is read with what it was sent, and its `apply` duration is found by it —
|
||||
// the send and the duration start at one moment, the node's sent_at.
|
||||
func TestAWalksRestIsReadFromEachMachinesOwnSend(t *testing.T) {
|
||||
inv := ForTest(t)
|
||||
ctx := t.Context()
|
||||
record, err := inv.AddNode(ctx, "g14")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := inv.NextSequence(ctx, record.ID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := inv.RecordSent(ctx, record.ID, "d-582", map[string]string{"build-agent": "c1"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
sent, found, err := inv.SentTo(ctx, "g14")
|
||||
if err != nil || !found || sent.At.IsZero() || sent.Digest != "d-582" {
|
||||
t.Fatalf("the machine's own send time: %+v %v %v", sent, found, err)
|
||||
}
|
||||
if _, ok, err := inv.FirstAppliedAfter(ctx, "g14", sent, sent.At.Add(20*time.Second)); err != nil || ok {
|
||||
t.Fatalf("no report yet: %v %v", ok, err)
|
||||
}
|
||||
if err := inv.RecordApplyDuration(ctx, "g14", "d-582", OutcomeApplied); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// The walk's record of its send, 20 s after the machine's own: still found, by the machine's own send.
|
||||
r, ok, err := inv.FirstAppliedAfter(ctx, "g14", sent, sent.At.Add(20*time.Second))
|
||||
if err != nil || !ok || r.Outcome != OutcomeApplied || r.At.Before(sent.At) {
|
||||
t.Fatalf("the machine's own report of its send: %+v %v %v", r, ok, err)
|
||||
}
|
||||
// A later send, reported: never this walk's.
|
||||
if err := inv.RecordSent(ctx, record.ID, "d-next", map[string]string{"build-agent": "c2"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := inv.RecordApplyDuration(ctx, "g14", "d-next", OutcomeApplied); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
again, _, _ := inv.FirstAppliedAfter(ctx, "g14", sent, sent.At.Add(20*time.Second))
|
||||
if !again.At.Equal(r.At) {
|
||||
t.Fatalf("a later send's report read as this walk's: %+v, was %+v", again, r)
|
||||
}
|
||||
}
|
||||
@@ -116,7 +116,7 @@ func TestAWalkOfTwoTiersEndsAtItsRestsLastReport(t *testing.T) {
|
||||
}}
|
||||
reports := map[string]AppliedReport{"ace": {At: *at("05:12"), Outcome: OutcomeApplied},
|
||||
"shanks": {At: *at("08:15.25"), Outcome: OutcomeApplied}}
|
||||
w := p.WalkPhases(*at("30:00"), func(node string, _ time.Time) (AppliedReport, bool) {
|
||||
w := p.WalkPhases(*at("30:00"), func(node string, _ SentDeclaration, _ time.Time) (AppliedReport, bool) {
|
||||
r, ok := reports[node]
|
||||
return r, ok
|
||||
})
|
||||
@@ -135,7 +135,7 @@ func TestAWalkOfTwoTiersEndsAtItsRestsLastReport(t *testing.T) {
|
||||
}
|
||||
// A report not yet in: the walk has no end, and its apply is unknown — never zero.
|
||||
delete(reports, "shanks")
|
||||
w = p.WalkPhases(*at("10:00"), func(node string, _ time.Time) (AppliedReport, bool) {
|
||||
w = p.WalkPhases(*at("10:00"), func(node string, _ SentDeclaration, _ time.Time) (AppliedReport, bool) {
|
||||
r, ok := reports[node]
|
||||
return r, ok
|
||||
})
|
||||
@@ -146,7 +146,7 @@ func TestAWalkOfTwoTiersEndsAtItsRestsLastReport(t *testing.T) {
|
||||
t.Errorf("apply while shanks has not reported: %+v", got)
|
||||
}
|
||||
// Silent past the bound: listed, not waited for.
|
||||
w = p.WalkPhases(at("08:00.5").Add(ApplySilentAfter+time.Second), func(node string, _ time.Time) (AppliedReport, bool) {
|
||||
w = p.WalkPhases(at("08:00.5").Add(ApplySilentAfter+time.Second), func(node string, _ SentDeclaration, _ time.Time) (AppliedReport, bool) {
|
||||
r, ok := reports[node]
|
||||
return r, ok
|
||||
})
|
||||
@@ -156,7 +156,7 @@ func TestAWalkOfTwoTiersEndsAtItsRestsLastReport(t *testing.T) {
|
||||
}
|
||||
// A failed report: the walk has no end, said.
|
||||
reports["shanks"] = AppliedReport{At: *at("08:10"), Outcome: OutcomeFailed}
|
||||
w = p.WalkPhases(*at("30:00"), func(node string, _ time.Time) (AppliedReport, bool) {
|
||||
w = p.WalkPhases(*at("30:00"), func(node string, _ SentDeclaration, _ time.Time) (AppliedReport, bool) {
|
||||
r, ok := reports[node]
|
||||
return r, ok
|
||||
})
|
||||
@@ -247,7 +247,7 @@ func TestAWalksMomentsAreKeptAndItsRestsReportRead(t *testing.T) {
|
||||
t.Fatal("a merge's time was not kept")
|
||||
}
|
||||
sent := time.Now().UTC().Add(-time.Minute).Truncate(time.Millisecond)
|
||||
if _, ok, err := inv.FirstAppliedAfter(ctx, "ace", sent); err != nil || ok {
|
||||
if _, ok, err := inv.FirstAppliedAfter(ctx, "ace", SentDeclaration{}, sent); err != nil || ok {
|
||||
t.Fatalf("no report yet: %v %v", ok, err)
|
||||
}
|
||||
for _, d := range []Duration{
|
||||
@@ -258,7 +258,7 @@ func TestAWalksMomentsAreKeptAndItsRestsReportRead(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
r, ok, err := inv.FirstAppliedAfter(ctx, "ace", sent)
|
||||
r, ok, err := inv.FirstAppliedAfter(ctx, "ace", SentDeclaration{}, sent)
|
||||
if err != nil || !ok || r.Outcome != OutcomeApplied || !r.At.Equal(sent.Add(11*time.Second)) {
|
||||
t.Fatalf("the report after the send: %+v %v %v", r, ok, err)
|
||||
}
|
||||
@@ -281,7 +281,7 @@ func TestALateReportIsSilentNotTheEnd(t *testing.T) {
|
||||
p := walkOf(t, recordedWalk)
|
||||
p.Modules["mesh-delivery"].SentAt = &sent
|
||||
p.Modules["mesh-delivery"].Rest = map[string]SentDeclaration{"laptop": {Digest: "d"}, "server": {Digest: "e"}}
|
||||
w := p.WalkPhases(sent.Add(3*time.Hour), func(node string, _ time.Time) (AppliedReport, bool) {
|
||||
w := p.WalkPhases(sent.Add(3*time.Hour), func(node string, _ SentDeclaration, _ time.Time) (AppliedReport, bool) {
|
||||
if node == "server" {
|
||||
return AppliedReport{At: sent.Add(20 * time.Second), Outcome: OutcomeApplied}, true
|
||||
}
|
||||
@@ -296,7 +296,7 @@ func TestALateReportIsSilentNotTheEnd(t *testing.T) {
|
||||
Started: sent.Add(time.Hour), Took: time.Second, Detail: OutcomeApplied}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, ok, err := inv.FirstAppliedAfter(ctx, "laptop", sent); err != nil || ok {
|
||||
if _, ok, err := inv.FirstAppliedAfter(ctx, "laptop", SentDeclaration{}, sent); err != nil || ok {
|
||||
t.Fatalf("a send past the bound read as the walk's: %v %v", ok, err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -75,6 +75,10 @@ type Options struct {
|
||||
Holder Holder
|
||||
// RenewEvery, Margin and Poll default to the package's.
|
||||
RenewEvery, Margin, Poll time.Duration
|
||||
// RenewWithin is how long one renewal may take before it counts as failed: RenewEvery when zero. Apart
|
||||
// from it so a test that renews often still gives a slow bus a serving controller's time to answer
|
||||
// (novox/hq issue 591).
|
||||
RenewWithin time.Duration
|
||||
// Floor is the highest epoch this mesh has issued, from the controller's own store. **An epoch is
|
||||
// never issued twice**: a lease bucket raised again from nothing — a bus whose data was replaced —
|
||||
// starts its revisions over, and every machine that heard epoch 57 would refuse the next controller
|
||||
@@ -138,6 +142,9 @@ func Open(ctx context.Context, js jetstream.JetStream, bucket string, o Options)
|
||||
if o.RenewEvery <= 0 {
|
||||
o.RenewEvery = RenewEvery
|
||||
}
|
||||
if o.RenewWithin <= 0 {
|
||||
o.RenewWithin = o.RenewEvery
|
||||
}
|
||||
if o.Margin <= 0 {
|
||||
o.Margin = Margin
|
||||
}
|
||||
@@ -406,7 +413,7 @@ func (l *Lease) Renew(ctx context.Context) error {
|
||||
if err != nil {
|
||||
return l.lose(err)
|
||||
}
|
||||
renewing, cancel := context.WithTimeout(ctx, l.o.RenewEvery)
|
||||
renewing, cancel := context.WithTimeout(ctx, l.o.RenewWithin)
|
||||
defer cancel()
|
||||
next, err := l.kv.Update(renewing, Key, value, revision)
|
||||
if err != nil {
|
||||
|
||||
@@ -32,14 +32,13 @@ func ForTest(t *testing.T) *Licences {
|
||||
name = name[:60]
|
||||
}
|
||||
|
||||
conn, err := pgx.Connect(t.Context(), admin)
|
||||
// Created already migrated, a copy of a template migrated once (novox/hq issue 591); the Migrate
|
||||
// below then finds nothing pending, which it does only if the copy's record matches every migration.
|
||||
migrations, err := Migrations()
|
||||
if err != nil {
|
||||
t.Fatalf("cannot reach the test PostgreSQL: %v", err)
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := conn.Exec(t.Context(), "create database "+name); err != nil {
|
||||
t.Fatalf("cannot create %s: %v", name, err)
|
||||
}
|
||||
conn.Close(t.Context())
|
||||
store.CreateMigratedForTest(t, admin, name, migrations)
|
||||
|
||||
cut := strings.LastIndex(admin, "/")
|
||||
t.Setenv(store.Variable(Name), admin[:cut]+"/"+name+"?sslmode=disable")
|
||||
@@ -61,10 +60,6 @@ func ForTest(t *testing.T) *Licences {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
migrations, err := Migrations()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := held.store.Migrate(t.Context(), migrations); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
@@ -79,6 +79,8 @@ func (b *natsBuilds) Ask(ctx context.Context, request BuildRequest) error {
|
||||
if _, err := b.js.Context().Publish(BuildWorkOf(b.role()), body, nats.Context(publish)); err != nil {
|
||||
return fmt.Errorf("cannot submit a build: %w", err)
|
||||
}
|
||||
// The acting controller puts it in order (novox/hq issue 558); this only says an ask joined.
|
||||
signalAsked()
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -113,6 +115,8 @@ func (b *natsBuilds) Submit(ctx context.Context, request BuildRequest,
|
||||
if _, err := b.js.Context().Publish(BuildWorkOf(b.role()), body, nats.Context(publish)); err != nil {
|
||||
return BuildResult{}, fmt.Errorf("%w: cannot submit a build: %w", ErrNotHandedOver, err)
|
||||
}
|
||||
// The acting controller puts it in order (novox/hq issue 558); this only says an ask joined.
|
||||
signalAsked()
|
||||
|
||||
waiting, cancelWait := context.WithTimeout(ctx, wait)
|
||||
defer cancelWait()
|
||||
@@ -167,6 +171,15 @@ type MachineOptions struct {
|
||||
// Paused is asked before every fetch: while it says so, nothing new is taken, and a build
|
||||
// already running finishes. Nil is never paused.
|
||||
Paused func() bool
|
||||
// Taking is called once, when this machine is bound to its worker and pulls work: from then on it
|
||||
// can take a build, which is what its container's health check reads (novox/hq issue 544). Nil is
|
||||
// told nothing.
|
||||
Taking func()
|
||||
// Claim is asked once a request is fetched, just before it is handed over: false hands it back to the
|
||||
// queue at once, for another holder. A holder being drained says so here (novox/hq ADR 0335): it marks
|
||||
// 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
|
||||
}
|
||||
|
||||
// MachineOverNATS takes build work from the current build role.
|
||||
@@ -232,6 +245,9 @@ func (m *natsMachine) Take(ctx context.Context, do func(context.Context, Build))
|
||||
"the mesh's to answer, not this machine's", m.seat, err)
|
||||
}
|
||||
m.sub = sub
|
||||
if m.opts.Taking != nil {
|
||||
m.opts.Taking()
|
||||
}
|
||||
|
||||
for {
|
||||
if ctx.Err() != nil {
|
||||
@@ -306,6 +322,12 @@ func (m *natsMachine) Take(ctx context.Context, do func(context.Context, Build))
|
||||
endCancelled(ctx, build)
|
||||
continue
|
||||
}
|
||||
// **Drained in the moment it was fetched** (novox/hq ADR 0335): handed back at once, as a pause
|
||||
// landing then is.
|
||||
if m.opts.Claim != nil && !m.opts.Claim() {
|
||||
_ = msg.Nak()
|
||||
continue
|
||||
}
|
||||
// A 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.
|
||||
|
||||
@@ -368,3 +368,36 @@ func TestNatsAMachineOnTheRetiredBuildRoleTakesThatRolesAsks(t *testing.T) {
|
||||
t.Errorf("the retired role's holder did not answer: %+v", result)
|
||||
}
|
||||
}
|
||||
|
||||
// A holder says once that it takes work, and only once it is bound to its worker (novox/hq issue 544):
|
||||
// its container's health check reads that, so a holder that could not bind is never said ready.
|
||||
func TestNatsAHolderSaysItTakesWorkOnceItIsBound(t *testing.T) {
|
||||
js := aBusWithTheBuildRole(t)
|
||||
ctx, stop := context.WithCancel(context.Background())
|
||||
defer stop()
|
||||
said := make(chan struct{}, 4)
|
||||
machine := MachineOverNATSWith(js, "anchor", TheBuildMachine, MachineOptions{Taking: func() { said <- struct{}{} }})
|
||||
defer machine.Close()
|
||||
go func() { _ = machine.Take(ctx, func(context.Context, Build) {}) }()
|
||||
select {
|
||||
case <-said:
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("a holder bound to its worker never said it takes work")
|
||||
}
|
||||
time.Sleep(300 * time.Millisecond)
|
||||
if len(said) != 0 {
|
||||
t.Fatalf("a holder said it takes work %d more time(s)", len(said))
|
||||
}
|
||||
|
||||
// A role the mesh raised no worker for cannot be bound, and its holder says nothing.
|
||||
bounded, end := context.WithTimeout(ctx, 10*time.Second)
|
||||
defer end()
|
||||
other := MachineOverNATSWith(js, "anchor", "mesh-no-such-seat", MachineOptions{Taking: func() { said <- struct{}{} }})
|
||||
defer other.Close()
|
||||
if err := other.Take(bounded, func(context.Context, Build) {}); err == nil {
|
||||
t.Fatal("a machine with no worker took work")
|
||||
}
|
||||
if len(said) != 0 {
|
||||
t.Fatal("a machine that could not bind said it takes work")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -11,6 +11,7 @@ import (
|
||||
"fmt"
|
||||
"log"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/broker"
|
||||
@@ -418,6 +419,12 @@ func (e Enrolment) Heard(ctx context.Context, report Report) (news bool, err err
|
||||
for id, why := range report.Failed {
|
||||
doing.Failed = append(doing.Failed, inventory.FailedResource{ID: id, Error: why})
|
||||
}
|
||||
// And a container left running because it said busy, or replaced after it waited (novox/hq ADR 0335):
|
||||
// kept with the account, for the walk whose first-machine gate waits on it.
|
||||
for _, b := range report.Busy {
|
||||
doing.Busy = append(doing.Busy, inventory.BusyWait{ID: b.ID, Container: b.Container, State: b.State,
|
||||
Since: b.Since, Until: b.Until, Said: b.Said})
|
||||
}
|
||||
// Ordered, so two readings of one failure are the same reading.
|
||||
sort.Slice(doing.Failed, func(i, j int) bool { return doing.Failed[i].ID < doing.Failed[j].ID })
|
||||
|
||||
@@ -492,7 +499,14 @@ func (e Enrolment) Heard(ctx context.Context, report Report) (news bool, err err
|
||||
}
|
||||
// And how long the machine took, from the send to this first account of it (novox/hq to-be 45
|
||||
// Phase 0): what the sent-not-reported bound will be set from. Said if lost, never a failure.
|
||||
if err := e.Inventory.RecordApplyDuration(ctx, report.Node, report.Declared, doing.Outcome); err != nil {
|
||||
//
|
||||
// **Not while a container waits because it said busy** (novox/hq ADR 0335 §4): the machine is not yet
|
||||
// running what it was sent, and the account that says the container replaced is the one that ends the
|
||||
// apply — so a walk's rest counts the wait in its delivery time, never exempted.
|
||||
if doing.Waiting() {
|
||||
log.Printf("%s applied declaration %s and leaves a busy container running until it says idle: %s",
|
||||
report.Node, short(report.Declared), busyWords(doing.Busy))
|
||||
} else if err := e.Inventory.RecordApplyDuration(ctx, report.Node, report.Declared, doing.Outcome); err != nil {
|
||||
log.Printf("how long %s took to apply could not be recorded: %v", report.Node, err)
|
||||
}
|
||||
|
||||
@@ -504,3 +518,15 @@ func (e Enrolment) Heard(ctx context.Context, report Report) (news bool, err err
|
||||
}
|
||||
return news, e.Inventory.RecordOwned(ctx, node.ID, report.Applied)
|
||||
}
|
||||
|
||||
// busyWords is every container waiting because it said busy, in a line (novox/hq ADR 0335).
|
||||
func busyWords(busy []inventory.BusyWait) string {
|
||||
var out []string
|
||||
for _, b := range busy {
|
||||
if b.State == inventory.BusyWaiting {
|
||||
out = append(out, fmt.Sprintf("%s since %s, at the latest until %s", b.Container,
|
||||
b.Since.UTC().Format(time.RFC3339), b.Until.UTC().Format(time.RFC3339)))
|
||||
}
|
||||
}
|
||||
return strings.Join(out, "; ")
|
||||
}
|
||||
|
||||
@@ -272,6 +272,12 @@ type Report struct {
|
||||
// `witness` and `not-reversible` are sent only to a machine whose report carries it.
|
||||
Witness int `json:"witness,omitempty"`
|
||||
|
||||
// Busy is every container the reporting apply left running because it said busy, and every one it
|
||||
// replaced after it had waited (novox/hq ADR 0335; mesh-host internal/link/messages.go): a walk's
|
||||
// first-machine gate waits while its module's container is listed waiting, and starts at the report
|
||||
// that no longer lists it so. Absent from an engine older than BusyContract, which never waits.
|
||||
Busy []Busy `json:"busy,omitempty"`
|
||||
|
||||
// Health is the machine's word on every long-running resource it runs for a module (novox/hq ADR
|
||||
// 0240, to-be 48 §4; mesh-host's internal/liveness): its state, since when, its failing streak and
|
||||
// the restarts its node-engine counted. **Absent from an engine older than the judging**, which is
|
||||
@@ -436,6 +442,31 @@ const ReadinessContract = 2
|
||||
// parses strictly and would refuse the whole declaration for it.
|
||||
const RootContract = 3
|
||||
|
||||
// BusyContract is the statement of an engine that also reads a container's `busy` (novox/hq ADR 0335): it asks
|
||||
// a container that declares it before replacing it and says the wait in its report. Only to such an engine is
|
||||
// the field sent: an older one parses strictly and would refuse the whole declaration for it. The same number as
|
||||
// mesh-host internal/link/messages.go's BusyContract.
|
||||
const BusyContract = 8
|
||||
|
||||
// The states of a busy container in a report (novox/hq ADR 0335 §2, §3).
|
||||
const (
|
||||
BusyWaiting = "waiting"
|
||||
BusyReplacedIdle = "replaced-when-idle"
|
||||
BusyReplacedAtBound = "replaced-at-bound"
|
||||
)
|
||||
|
||||
// Busy is one container a busy command held, as a node-engine reports it (novox/hq ADR 0335). ID is the
|
||||
// resource's id, `<module>.<resource>`; Since when the first apply that would have replaced it ran, Until when
|
||||
// the engine's bound replaces it at the latest, Said the last output of its busy command.
|
||||
type Busy struct {
|
||||
ID string `json:"id"`
|
||||
Container string `json:"container"`
|
||||
State string `json:"state"`
|
||||
Since time.Time `json:"since"`
|
||||
Until time.Time `json:"until"`
|
||||
Said string `json:"said,omitempty"`
|
||||
}
|
||||
|
||||
// ReasonRoot starts the reason of an account verdict that found a way to root (ADR 0266); the node-engine's
|
||||
// own words (mesh-host internal/accounts ReasonRoot).
|
||||
const ReasonRoot = "can become root without a person"
|
||||
|
||||
+40
-15
@@ -62,7 +62,11 @@ type QueuedAsk struct {
|
||||
Path string `json:"path,omitempty"`
|
||||
Ref string `json:"ref,omitempty"`
|
||||
AskedAt time.Time `json:"asked-at,omitempty"`
|
||||
State string `json:"state"`
|
||||
// Stored is when the stream first took it, a move into order kept (novox/hq issue 558).
|
||||
Stored time.Time `json:"stored,omitempty"`
|
||||
State string `json:"state"`
|
||||
// movedFrom is the MovedFromHeader it carries: the sequence of the message a move copied it from.
|
||||
movedFrom string
|
||||
// On and Started are the holder building an in-flight ask and when it started, from its started
|
||||
// event. Was is the holder that started an in-flight ask and is no longer building it — handed
|
||||
// back, to be handed out again — with Started its start there.
|
||||
@@ -236,15 +240,7 @@ func ReadQueue(ctx context.Context, js *broker.JetStream, seat string) (Queue, e
|
||||
if err != nil {
|
||||
return q, fmt.Errorf("reading the %s work queue: %w", seat, err)
|
||||
}
|
||||
ask := QueuedAsk{Seq: msg.Sequence}
|
||||
if json.Unmarshal(msg.Data, &ask.Request) == nil {
|
||||
r := ask.Request
|
||||
ask.ID, ask.Repository, ask.Path, ask.Ref = r.ID, r.Repository, r.Path, r.Ref
|
||||
if at, ok := BuildAskedAt(r.ID); ok {
|
||||
ask.AskedAt = at
|
||||
}
|
||||
}
|
||||
asks = append(asks, ask)
|
||||
asks = append(asks, askOf(msg))
|
||||
seq = msg.Sequence + 1
|
||||
}
|
||||
|
||||
@@ -269,6 +265,26 @@ func ReadQueue(ctx context.Context, js *broker.JetStream, seat string) (Queue, e
|
||||
return q, nil
|
||||
}
|
||||
|
||||
// askOf is one ask as the stream holds it: what its request says, and when the stream took it.
|
||||
func askOf(msg *jetstream.RawStreamMsg) QueuedAsk {
|
||||
ask := QueuedAsk{Seq: msg.Sequence, Stored: msg.Time}
|
||||
// Moved into order, it keeps when the stream first took it (novox/hq issue 558).
|
||||
if at, err := time.Parse(time.RFC3339Nano, msg.Header.Get(QueuedAtHeader)); err == nil {
|
||||
ask.Stored = at
|
||||
}
|
||||
if json.Unmarshal(msg.Data, &ask.Request) == nil {
|
||||
r := ask.Request
|
||||
ask.ID, ask.Repository, ask.Path, ask.Ref = r.ID, r.Repository, r.Path, r.Ref
|
||||
if at, ok := BuildAskedAt(r.ID); ok {
|
||||
ask.AskedAt = at
|
||||
}
|
||||
} else {
|
||||
// Nothing read is nothing kept: a half-read request would pass for a delivery (novox/hq issue 558).
|
||||
ask.Request = BuildRequest{}
|
||||
}
|
||||
return ask
|
||||
}
|
||||
|
||||
// WithoutPassword is a repository as a listing may say it: a URL's password, should one ever be written
|
||||
// into it, left out (novox/hq issue 455). Only for what is shown: the ask itself keeps what it was given,
|
||||
// which a cancel's outcome and a rebuild's match are read from.
|
||||
@@ -305,11 +321,23 @@ func ReadBuildEvents(ctx context.Context, js *broker.JetStream, seat string, sin
|
||||
return heard, fmt.Errorf("cannot read %s from the bus: %w", subject, err)
|
||||
}
|
||||
defer func() { _ = sub.Unsubscribe() }()
|
||||
for {
|
||||
batch, err := sub.Fetch(500, nats.MaxWait(2*time.Second))
|
||||
// **Read what is there, then stop** (novox/hq issue 591): the consumer says how many events it holds
|
||||
// from the moment asked, and the reading ends once they are read. A fetch for a full batch waits out
|
||||
// its whole wait when fewer are there, so every reading of the queue took two seconds and more.
|
||||
info, err := sub.ConsumerInfo()
|
||||
if err != nil {
|
||||
return heard, fmt.Errorf("cannot read %s from the bus: %w", subject, err)
|
||||
}
|
||||
for remaining := info.NumPending; remaining > 0; {
|
||||
batch, err := sub.Fetch(int(min(remaining, 500)), nats.MaxWait(2*time.Second))
|
||||
if err != nil && !errors.Is(err, nats.ErrTimeout) && !errors.Is(err, context.DeadlineExceeded) {
|
||||
return heard, fmt.Errorf("reading %s: %w", subject, err)
|
||||
}
|
||||
// None in the whole wait: what was counted is gone from the stream since (its limits), not coming.
|
||||
if len(batch) == 0 {
|
||||
break
|
||||
}
|
||||
remaining -= uint64(len(batch))
|
||||
for _, msg := range batch {
|
||||
switch msg.Subject {
|
||||
case BuildStartedOf(seat):
|
||||
@@ -326,9 +354,6 @@ func ReadBuildEvents(ctx context.Context, js *broker.JetStream, seat string, sin
|
||||
}
|
||||
}
|
||||
}
|
||||
if len(batch) < 500 {
|
||||
break
|
||||
}
|
||||
if ctx.Err() != nil {
|
||||
return heard, ctx.Err()
|
||||
}
|
||||
|
||||
@@ -0,0 +1,426 @@
|
||||
package link
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/broker"
|
||||
)
|
||||
|
||||
// The order the build queue hands asks out in (novox/hq issue 558).
|
||||
//
|
||||
// **A build a walk waits for goes before a pull request's check.** The queue was first come, first served,
|
||||
// and a merged change's delivery build waited minutes behind checks while its walk, and every walk after
|
||||
// it, waited on the build. So:
|
||||
//
|
||||
// - a delivery is handed out before any check;
|
||||
// - a check that has waited CheckWaitsAtMost goes before every delivery asked after it, so no check
|
||||
// waits for ever behind a stream of deliveries;
|
||||
// - within a kind, first come, first served, by when it was asked;
|
||||
// - an ask whose kind cannot be told counts as a check: a delivery taken late breaks a budget, a check
|
||||
// taken late breaks nothing.
|
||||
//
|
||||
// **Kept in the stream, never beside it.** The queue is the seat's work queue, which the server hands out
|
||||
// in stream order to whichever holder pulls; nothing else holds the order, so a restarted controller
|
||||
// finds it as it was left. An ask that must go behind another is moved to the tail — deleted from where
|
||||
// it was and published again, its body unchanged, under the same id — while the worker is paused, so no
|
||||
// holder takes it in the moment it moves. Only waiting asks move: what a holder took is never touched.
|
||||
|
||||
// The kinds of ask, as the queue says them.
|
||||
const (
|
||||
KindDelivery = "delivery"
|
||||
KindCheck = "check"
|
||||
)
|
||||
|
||||
// CheckWaitsAtMost is how long a check waits before it goes ahead of the deliveries asked after it.
|
||||
var CheckWaitsAtMost = 10 * time.Minute
|
||||
|
||||
// KindOf is an ask's kind: a check for a pull request's check — a recheck the merge tool asks included —
|
||||
// a dry run, or an ask nothing could be read from; a delivery for any other build.
|
||||
func KindOf(r BuildRequest) string {
|
||||
if r.ID == "" || r.Check != nil || r.DryRun {
|
||||
return KindCheck
|
||||
}
|
||||
return KindDelivery
|
||||
}
|
||||
|
||||
// askedOrStored is when an ask was asked, as its id says, or else when the queue took it.
|
||||
func (a QueuedAsk) askedOrStored() time.Time {
|
||||
if !a.AskedAt.IsZero() {
|
||||
return a.AskedAt
|
||||
}
|
||||
return a.Stored
|
||||
}
|
||||
|
||||
// QueueOrder is the order the waiting asks are to be handed out in, now: the deliveries and every check
|
||||
// that has waited CheckWaitsAtMost, by when each was asked, then the other checks, by when each was asked.
|
||||
func QueueOrder(waiting []QueuedAsk, now time.Time) []QueuedAsk {
|
||||
var first, then []QueuedAsk
|
||||
for _, a := range waiting {
|
||||
if KindOf(a.Request) == KindCheck && now.Sub(a.askedOrStored()) < CheckWaitsAtMost {
|
||||
then = append(then, a)
|
||||
continue
|
||||
}
|
||||
first = append(first, a)
|
||||
}
|
||||
byAsking := func(s []QueuedAsk) {
|
||||
sort.SliceStable(s, func(i, j int) bool {
|
||||
if ti, tj := s[i].askedOrStored(), s[j].askedOrStored(); !ti.Equal(tj) {
|
||||
return ti.Before(tj)
|
||||
}
|
||||
return s[i].Seq < s[j].Seq
|
||||
})
|
||||
}
|
||||
byAsking(first)
|
||||
byAsking(then)
|
||||
return append(first, then...)
|
||||
}
|
||||
|
||||
// toMove is the asks to move to the tail, in this order, for the queue to stand as order says: everything
|
||||
// after the longest start of order that the queue already holds in that order.
|
||||
func toMove(queue, order []QueuedAsk) []QueuedAsk {
|
||||
at := make(map[uint64]int, len(queue))
|
||||
for i, a := range queue {
|
||||
at[a.Seq] = i
|
||||
}
|
||||
kept, last := 0, -1
|
||||
for kept < len(order) {
|
||||
i, ok := at[order[kept].Seq]
|
||||
if !ok || i < last {
|
||||
break
|
||||
}
|
||||
last = i
|
||||
kept++
|
||||
}
|
||||
return order[kept:]
|
||||
}
|
||||
|
||||
// orderPasses keeps one pass at a time. Only the acting controller's loop runs one (KeepingInOrder), so
|
||||
// there is one; the lock holds a test's two at once to that too.
|
||||
var orderPasses sync.Mutex
|
||||
|
||||
// pausedFor bounds a pause the pass could not end itself: a controller that stops mid-pass leaves the
|
||||
// worker paused this long at most. moveMargin is the part of it no move starts in: a pass stops moving
|
||||
// with at least this left, and the next pass moves the rest, so no move runs past the pause.
|
||||
const (
|
||||
pausedFor = 20 * time.Second
|
||||
moveMargin = 5 * time.Second
|
||||
publishWaits = 3 * time.Second
|
||||
)
|
||||
|
||||
// QueuedAtHeader carries, on a moved ask, when the stream first took it: a move keeps an ask's place in
|
||||
// time, its id unread or not.
|
||||
const QueuedAtHeader = "Mesh-Queued-At"
|
||||
|
||||
// MovedFromHeader names, on a move's copy, the sequence of the message it is a copy of: with QueuedAtHeader,
|
||||
// what proves a second message of one id is a move's copy and not an ask delivered again (ADR 0293), whose
|
||||
// headers are its original's — and so never name that original's own sequence.
|
||||
const MovedFromHeader = "Mesh-Moved-From"
|
||||
|
||||
// Pass is what one pass did: the asks it moved, and each second copy it removed, in words.
|
||||
type Pass struct {
|
||||
Moved []string
|
||||
Removed []string
|
||||
}
|
||||
|
||||
// asked is the signal an ask sends the acting controller's pass loop, in this process. An ask asked in
|
||||
// another process — a verb run as its own command — is seen by the loop as the queue's last sequence moving.
|
||||
var asked = make(chan struct{}, 1)
|
||||
|
||||
// signalAsked tells the pass loop an ask joined the queue. Never blocks, never runs a pass.
|
||||
func signalAsked() {
|
||||
select {
|
||||
case asked <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
// PutInOrder moves the seat's waiting asks into the order QueueOrder says, and says each it moved. A queue
|
||||
// already in order is read and left alone; the worker is paused only while asks move. Run by the acting
|
||||
// controller alone (KeepingInOrder): two passes in two processes would each resume what the other paused.
|
||||
func PutInOrder(ctx context.Context, js *broker.JetStream, seat string) (Pass, error) {
|
||||
var pass Pass
|
||||
orderPasses.Lock()
|
||||
defer orderPasses.Unlock()
|
||||
api, err := jetstream.New(js.Conn())
|
||||
if err != nil {
|
||||
return pass, err
|
||||
}
|
||||
worker, _ := broker.HolderConsumerFor("", "", broker.DeclaredSeat{Name: seat, Accepts: []string{"build"}})
|
||||
waiting, extra, delivered, err := readWaiting(ctx, api, worker, seat)
|
||||
if err != nil || (len(extra) == 0 && len(toMove(waiting, QueueOrder(waiting, time.Now()))) == 0) {
|
||||
return pass, err
|
||||
}
|
||||
// Paused, then read again: what the worker handed out up to the pause is not waiting any more.
|
||||
pauseEnds := time.Now().Add(pausedFor)
|
||||
if delivered >= 0 {
|
||||
if _, err := api.PauseConsumer(ctx, worker.Stream, worker.Name, pauseEnds); err != nil {
|
||||
return pass, fmt.Errorf("the %s worker could not be paused to put its queue in order, so it is left "+
|
||||
"first come, first served: %w", seat, err)
|
||||
}
|
||||
defer func() {
|
||||
resuming, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
|
||||
defer cancel()
|
||||
if _, err := api.ResumeConsumer(resuming, worker.Stream, worker.Name); err != nil {
|
||||
fmt.Printf("the %s worker could not be resumed after its queue was put in order; it resumes by "+
|
||||
"itself by %s: %v\n", seat, pauseEnds.Format(time.RFC3339), err)
|
||||
}
|
||||
}()
|
||||
if waiting, extra, _, err = readWaiting(ctx, api, worker, seat); err != nil {
|
||||
return pass, err
|
||||
}
|
||||
}
|
||||
stream, err := api.Stream(ctx, worker.Stream)
|
||||
if err != nil {
|
||||
return pass, err
|
||||
}
|
||||
// A second copy of one ask — a move that published and stopped before it deleted — goes first.
|
||||
for _, a := range extra {
|
||||
if err := stream.DeleteMsg(ctx, a.Seq); err != nil && !errors.Is(err, jetstream.ErrMsgNotFound) {
|
||||
return pass, fmt.Errorf("a second copy of %s (seq %d) could not be removed: %w", a.ID, a.Seq, err)
|
||||
}
|
||||
pass.Removed = append(pass.Removed, fmt.Sprintf("%s at seq %d, a move's second copy of the message at seq %s",
|
||||
a.ID, a.Seq, a.movedFrom))
|
||||
}
|
||||
for _, a := range toMove(waiting, QueueOrder(waiting, time.Now())) {
|
||||
if delivered >= 0 && time.Until(pauseEnds) < moveMargin {
|
||||
// The rest waits for the next pass, under a pause of its own.
|
||||
break
|
||||
}
|
||||
ok, err := moveToTheTail(ctx, js, stream, worker.Name, seat, a)
|
||||
if err != nil {
|
||||
return pass, err
|
||||
}
|
||||
if ok {
|
||||
pass.Moved = append(pass.Moved, a.ID)
|
||||
}
|
||||
}
|
||||
return pass, nil
|
||||
}
|
||||
|
||||
// readWaiting is every ask the worker has not yet handed out, in queue order, but each second copy of one
|
||||
// ask; the second copies, to remove; and the last sequence the worker handed out, -1 when there is no
|
||||
// worker, so nothing is handed out to pause.
|
||||
//
|
||||
// **A second copy is proven, never guessed from an id**: a waiting message whose MovedFromHeader names an
|
||||
// earlier message of the same request id still in the queue, and whose QueuedAtHeader is that message's
|
||||
// first-queued time — a move that published its copy and stopped before it deleted the original. The
|
||||
// original stays, waiting or handed out (a holder's then, and the copy would build it twice). Any other
|
||||
// second message of one id — an ask a dead letter delivered again beside its original (ADR 0293) — is a
|
||||
// real ask and stays.
|
||||
func readWaiting(ctx context.Context, api jetstream.JetStream, worker broker.Consumer, seat string) (
|
||||
waiting, extra []QueuedAsk, delivered int64, err error) {
|
||||
stream, err := api.Stream(ctx, worker.Stream)
|
||||
if err != nil {
|
||||
return nil, nil, 0, fmt.Errorf("the %s work queue (%s) cannot be read: %w", seat, worker.Stream, err)
|
||||
}
|
||||
delivered = -1
|
||||
if consumer, err := stream.Consumer(ctx, worker.Name); err == nil {
|
||||
info, err := consumer.Info(ctx)
|
||||
if err != nil {
|
||||
return nil, nil, 0, fmt.Errorf("the worker of %s cannot be read: %w", seat, err)
|
||||
}
|
||||
delivered = int64(info.Delivered.Stream)
|
||||
} else if !errors.Is(err, jetstream.ErrConsumerNotFound) {
|
||||
return nil, nil, 0, fmt.Errorf("the worker of %s cannot be read: %w", seat, err)
|
||||
}
|
||||
// From the first: the stream holds only what is unsettled, so this is a handful of reads.
|
||||
// Every message by sequence, with its id and first-queued time, to prove a copy against.
|
||||
type held struct {
|
||||
id string
|
||||
stored time.Time
|
||||
}
|
||||
bySeq := map[string]held{}
|
||||
var seq uint64 = 1
|
||||
for n := 0; n < 10000; n++ {
|
||||
msg, err := stream.GetMsg(ctx, seq, jetstream.WithGetMsgSubject(BuildWorkOf(seat)))
|
||||
if errors.Is(err, jetstream.ErrMsgNotFound) {
|
||||
break
|
||||
}
|
||||
if err != nil {
|
||||
return nil, nil, 0, fmt.Errorf("reading the %s work queue: %w", seat, err)
|
||||
}
|
||||
seq = msg.Sequence + 1
|
||||
a := askOf(msg)
|
||||
a.movedFrom = msg.Header.Get(MovedFromHeader)
|
||||
from, known := bySeq[a.movedFrom]
|
||||
again := a.ID != "" && a.movedFrom != "" && known && from.id == a.ID && from.stored.Equal(a.Stored)
|
||||
bySeq[strconv.FormatUint(a.Seq, 10)] = held{id: a.ID, stored: a.Stored}
|
||||
switch {
|
||||
case int64(a.Seq) <= delivered:
|
||||
case again:
|
||||
extra = append(extra, a)
|
||||
default:
|
||||
waiting = append(waiting, a)
|
||||
}
|
||||
}
|
||||
return waiting, extra, delivered, nil
|
||||
}
|
||||
|
||||
// moveToTheTail publishes the ask's body again at the queue's tail and then deletes it where it was. **The
|
||||
// copy first**: a publish that fails deletes nothing, and a pass that stops between the two leaves two
|
||||
// copies, which the next pass sees by their one id and makes one (readWaiting). A move never loses an ask.
|
||||
// Looked at once more just before: one the worker has handed out since is a holder's now and stays. False
|
||||
// when it was not moved.
|
||||
func moveToTheTail(ctx context.Context, js *broker.JetStream, stream jetstream.Stream, workerName, seat string,
|
||||
a QueuedAsk) (bool, error) {
|
||||
if consumer, err := stream.Consumer(ctx, workerName); err == nil {
|
||||
info, err := consumer.Info(ctx)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
if info.Delivered.Stream >= a.Seq {
|
||||
return false, nil
|
||||
}
|
||||
}
|
||||
msg, err := stream.GetMsg(ctx, a.Seq)
|
||||
if errors.Is(err, jetstream.ErrMsgNotFound) {
|
||||
return false, nil
|
||||
}
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
again := &nats.Msg{Subject: msg.Subject, Data: msg.Data, Header: nats.Header{}}
|
||||
for k, v := range msg.Header {
|
||||
// The server's own headers stay behind: a message id would be taken for a duplicate and dropped.
|
||||
if !strings.HasPrefix(k, "Nats-") {
|
||||
again.Header[k] = v
|
||||
}
|
||||
}
|
||||
// Its place in time goes with it: when the stream first took it, as askOf read it; and the message it
|
||||
// is a copy of, which proves it one.
|
||||
again.Header.Set(QueuedAtHeader, askOf(msg).Stored.UTC().Format(time.RFC3339Nano))
|
||||
again.Header.Set(MovedFromHeader, strconv.FormatUint(a.Seq, 10))
|
||||
// Inside the margin, so the publish ends while the worker is still paused.
|
||||
publishing, cancel := context.WithTimeout(ctx, publishWaits)
|
||||
ack, err := js.Context().PublishMsg(again, nats.Context(publishing))
|
||||
cancel()
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("%s (seq %d) could not be put behind, and stays where it is: %w", a.ID, a.Seq, err)
|
||||
}
|
||||
// **Handed out meanwhile** — a bus slow enough that the pause ended: the original is a holder's, and the
|
||||
// copy goes instead, so it is built once.
|
||||
if consumer, err := stream.Consumer(ctx, workerName); err == nil {
|
||||
info, err := consumer.Info(ctx)
|
||||
if err != nil || info.Delivered.Stream >= a.Seq {
|
||||
if derr := stream.DeleteMsg(ctx, ack.Sequence); derr != nil && !errors.Is(derr, jetstream.ErrMsgNotFound) {
|
||||
return false, fmt.Errorf("%s was handed out while it moved and its copy (seq %d) could not be "+
|
||||
"removed; the next pass removes it: %w", a.ID, ack.Sequence, derr)
|
||||
}
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("whether %s was handed out while it moved cannot be read, so its copy was "+
|
||||
"removed and it stays where it was: %w", a.ID, err)
|
||||
}
|
||||
if info.Delivered.Stream >= ack.Sequence {
|
||||
return false, fmt.Errorf("%s was handed out while it moved, and its copy too: it may be built twice", a.ID)
|
||||
}
|
||||
return false, nil
|
||||
}
|
||||
}
|
||||
if err := stream.DeleteMsg(ctx, a.Seq); err != nil && !errors.Is(err, jetstream.ErrMsgNotFound) {
|
||||
return false, fmt.Errorf("%s was copied behind and its original (seq %d) not removed; the next pass "+
|
||||
"removes the second copy: %w", a.ID, a.Seq, err)
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// KeepingInOrder runs the passes, on the acting controller alone, until ctx ends: when an ask in this
|
||||
// process says so, when the queue's last sequence moved — an ask from another process — and every
|
||||
// passEvery, for a check that comes of age with nothing asked. seat says the build seat to order; acting
|
||||
// whether this controller acts (Holding); said what a pass did.
|
||||
func KeepingInOrder(ctx context.Context, js *broker.JetStream, seat func() string, acting func() bool,
|
||||
said func(format string, args ...any)) {
|
||||
tick := time.NewTicker(orderLooks)
|
||||
defer tick.Stop()
|
||||
var lastSeq uint64
|
||||
var lastPass time.Time
|
||||
for {
|
||||
signalled := false
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-asked:
|
||||
signalled = true
|
||||
case <-tick.C:
|
||||
}
|
||||
if !acting() {
|
||||
continue
|
||||
}
|
||||
name := seat()
|
||||
worker, _ := broker.HolderConsumerFor("", "", broker.DeclaredSeat{Name: name, Accepts: []string{"build"}})
|
||||
info, err := js.Context().StreamInfo(worker.Stream)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
if !signalled && info.State.LastSeq == lastSeq && time.Since(lastPass) < passEvery {
|
||||
continue
|
||||
}
|
||||
lastSeq, lastPass = info.State.LastSeq, time.Now()
|
||||
passing, cancel := context.WithTimeout(ctx, pausedFor+10*time.Second)
|
||||
pass, err := PutInOrder(passing, js, name)
|
||||
cancel()
|
||||
for _, r := range pass.Removed {
|
||||
said("the %s queue: removed %s", name, r)
|
||||
}
|
||||
if len(pass.Moved) > 0 {
|
||||
said("the %s queue put in order: %s moved behind what goes first", name, strings.Join(pass.Moved, ", "))
|
||||
}
|
||||
if err != nil {
|
||||
said("the %s queue could not be put in order: %v", name, err)
|
||||
}
|
||||
if info, err := js.Context().StreamInfo(worker.Stream); err == nil {
|
||||
// Its own moves are not an ask.
|
||||
lastSeq = info.State.LastSeq
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// orderLooks is how often the loop looks whether an ask joined from another process; passEvery how often
|
||||
// it passes with nothing asked.
|
||||
var (
|
||||
orderLooks = 2 * time.Second
|
||||
passEvery = 30 * time.Second
|
||||
)
|
||||
|
||||
// DeleteAsk removes a waiting or dead ask from the queue: at its sequence, or — moved into order since it
|
||||
// was read — the copy carrying its id. Says the sequence it removed, 0 when neither was there any more.
|
||||
func DeleteAsk(ctx context.Context, js *broker.JetStream, seat string, ask QueuedAsk) (uint64, error) {
|
||||
worker, _ := broker.HolderConsumerFor("", "", broker.DeclaredSeat{Name: seat, Accepts: []string{"build"}})
|
||||
gone := func(err error) bool {
|
||||
return errors.Is(err, nats.ErrMsgNotFound) || errors.Is(err, jetstream.ErrMsgNotFound) ||
|
||||
(err != nil && strings.Contains(err.Error(), "no message found"))
|
||||
}
|
||||
err := js.Context().DeleteMsg(worker.Stream, ask.Seq)
|
||||
if err == nil {
|
||||
return ask.Seq, nil
|
||||
}
|
||||
if !gone(err) {
|
||||
return 0, err
|
||||
}
|
||||
if ask.ID == "" {
|
||||
return 0, nil
|
||||
}
|
||||
q, err := ReadQueue(ctx, js, seat)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
for _, a := range q.Asks {
|
||||
if a.ID == ask.ID && a.Seq != ask.Seq && a.State != AskInFlight {
|
||||
if err := js.Context().DeleteMsg(worker.Stream, a.Seq); err != nil && !gone(err) {
|
||||
return 0, err
|
||||
} else if err == nil {
|
||||
return a.Seq, nil
|
||||
}
|
||||
}
|
||||
}
|
||||
return 0, nil
|
||||
}
|
||||
@@ -0,0 +1,441 @@
|
||||
package link
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"reflect"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/broker"
|
||||
)
|
||||
|
||||
// The order the build queue hands asks out in (novox/hq issue 558): a build a walk waits for before any
|
||||
// pull request's check, a check that has waited ten minutes before every delivery asked after it, and
|
||||
// first come, first served within a kind.
|
||||
|
||||
func anAsk(seq uint64, id string, asked time.Time, check bool) QueuedAsk {
|
||||
r := BuildRequest{ID: id, Repository: "/r"}
|
||||
if check {
|
||||
r.Check = &CheckRequest{Owner: "novox", Repo: "r"}
|
||||
}
|
||||
return QueuedAsk{Seq: seq, ID: id, AskedAt: asked, Request: r, State: AskWaiting}
|
||||
}
|
||||
|
||||
func idsOf(asks []QueuedAsk) []string {
|
||||
out := []string{}
|
||||
for _, a := range asks {
|
||||
out = append(out, a.ID)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func TestADeliveryIsHandedOutBeforeTheChecksAskedBeforeIt(t *testing.T) {
|
||||
now := time.Date(2026, 10, 12, 12, 0, 0, 0, time.UTC)
|
||||
waiting := []QueuedAsk{
|
||||
anAsk(1, "check-a", now.Add(-3*time.Minute), true),
|
||||
anAsk(2, "check-b", now.Add(-2*time.Minute), true),
|
||||
anAsk(3, "delivery-d", now.Add(-time.Minute), false),
|
||||
}
|
||||
want := []string{"delivery-d", "check-a", "check-b"}
|
||||
if got := idsOf(QueueOrder(waiting, now)); !reflect.DeepEqual(got, want) {
|
||||
t.Fatalf("handed out as %v, want %v", got, want)
|
||||
}
|
||||
// Only the checks ahead of it move behind it; the delivery stays where it is.
|
||||
if got := idsOf(toMove(waiting, QueueOrder(waiting, now))); !reflect.DeepEqual(got, []string{"check-a", "check-b"}) {
|
||||
t.Fatalf("moved %v", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestACheckThatWaitedTenMinutesGoesBeforeALaterDelivery(t *testing.T) {
|
||||
now := time.Date(2026, 10, 12, 12, 0, 0, 0, time.UTC)
|
||||
// As the queue stands after a delivery put the young checks behind it, ten minutes ago and since.
|
||||
waiting := []QueuedAsk{
|
||||
anAsk(5, "delivery-early", now.Add(-12*time.Minute), false),
|
||||
anAsk(6, "delivery-later", now.Add(-time.Minute), false),
|
||||
anAsk(7, "check-old", now.Add(-11*time.Minute), true),
|
||||
anAsk(8, "check-young", now.Add(-9*time.Minute), true),
|
||||
}
|
||||
want := []string{"delivery-early", "check-old", "delivery-later", "check-young"}
|
||||
if got := idsOf(QueueOrder(waiting, now)); !reflect.DeepEqual(got, want) {
|
||||
t.Fatalf("handed out as %v, want %v", got, want)
|
||||
}
|
||||
if got := idsOf(toMove(waiting, QueueOrder(waiting, now))); !reflect.DeepEqual(got, []string{"delivery-later", "check-young"}) {
|
||||
t.Fatalf("moved %v", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAnAskWhoseKindCannotBeToldIsACheck(t *testing.T) {
|
||||
for _, r := range []BuildRequest{
|
||||
{}, // unreadable: nothing was read from it
|
||||
{ID: "build-1", Check: &CheckRequest{}}, // a pull request's check, or a recheck the merge tool asked
|
||||
{ID: "build-2", DryRun: true}, // looked at, never delivered
|
||||
} {
|
||||
if got := KindOf(r); got != KindCheck {
|
||||
t.Errorf("%+v is a %s", r, got)
|
||||
}
|
||||
}
|
||||
if got := KindOf(BuildRequest{ID: "build-3", Repository: "/r"}); got != KindDelivery {
|
||||
t.Errorf("a build is a %s", got)
|
||||
}
|
||||
}
|
||||
|
||||
// Against a real server: two checks wait, a delivery is asked, and the next free holder takes the delivery.
|
||||
func TestNatsADeliveryIsTakenBeforeTheChecksQueuedAheadOfIt(t *testing.T) {
|
||||
js := aBusWithTheBuildRole(t)
|
||||
ctx, stop := context.WithCancel(context.Background())
|
||||
defer stop()
|
||||
asks := BuildsOn(js, TheBuildMachine)
|
||||
now := time.Now()
|
||||
for i, r := range []BuildRequest{
|
||||
{ID: NewBuildID(now.Add(-3 * time.Minute)), Repository: "/a", Check: &CheckRequest{Owner: "novox", Repo: "a"}},
|
||||
{ID: NewBuildID(now.Add(-2 * time.Minute)), Repository: "/b", Check: &CheckRequest{Owner: "novox", Repo: "b"}},
|
||||
{ID: NewBuildID(now), Repository: "/d"},
|
||||
} {
|
||||
if i == 2 {
|
||||
// And one nobody can read, whose kind cannot be told: it waits with the checks.
|
||||
if _, err := js.Context().Publish(BuildWork(), []byte("not an ask")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if err := asks.Ask(ctx, r); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
// The acting controller's loop puts it in order, told by the asks.
|
||||
go KeepingInOrder(ctx, js, func() string { return TheBuildMachine }, func() bool { return true },
|
||||
func(string, ...any) {})
|
||||
time.Sleep(500 * time.Millisecond)
|
||||
took := make(chan string, 4)
|
||||
machine := MachineOverNATS(js, "anchor")
|
||||
defer machine.Close()
|
||||
go func() {
|
||||
_ = machine.Take(ctx, func(ctx context.Context, work Build) {
|
||||
took <- work.Request().Repository
|
||||
_ = work.Announce(ctx, BuildResult{ID: work.Request().ID, On: "anchor", Failed: "no"})
|
||||
_ = work.Done()
|
||||
})
|
||||
}()
|
||||
var order []string
|
||||
for len(order) < 3 {
|
||||
select {
|
||||
case r := <-took:
|
||||
order = append(order, r)
|
||||
case <-time.After(10 * time.Second):
|
||||
t.Fatalf("taken so far: %v", order)
|
||||
}
|
||||
}
|
||||
if want := []string{"/d", "/a", "/b"}; !reflect.DeepEqual(order, want) {
|
||||
t.Fatalf("taken as %v, want %v", order, want)
|
||||
}
|
||||
}
|
||||
|
||||
// --- the moves, against a real server ------------------------------------------------------------
|
||||
|
||||
func publishAsk(t *testing.T, js *broker.JetStream, r BuildRequest) {
|
||||
t.Helper()
|
||||
body, _ := json.Marshal(r)
|
||||
if _, err := js.Context().Publish(BuildWork(), body); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func aCheck(asked time.Time, repository string) BuildRequest {
|
||||
return BuildRequest{ID: NewBuildID(asked), Repository: repository, Check: &CheckRequest{Owner: "novox", Repo: "r"}}
|
||||
}
|
||||
|
||||
func aDelivery(asked time.Time, repository string) BuildRequest {
|
||||
return BuildRequest{ID: NewBuildID(asked), Repository: repository}
|
||||
}
|
||||
|
||||
// queued is the stream's asks, in queue order.
|
||||
func queued(t *testing.T, js *broker.JetStream) []QueuedAsk {
|
||||
t.Helper()
|
||||
q, err := ReadQueue(context.Background(), js, TheBuildMachine)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return q.Asks
|
||||
}
|
||||
|
||||
// takeAll runs a holder that records how often it was handed each repository.
|
||||
func takeAll(t *testing.T, ctx context.Context, js *broker.JetStream) (chan string, func() map[string]int) {
|
||||
t.Helper()
|
||||
took := make(chan string, 100)
|
||||
var mu sync.Mutex
|
||||
counts := map[string]int{}
|
||||
machine := MachineOverNATS(js, "anchor")
|
||||
t.Cleanup(machine.Close)
|
||||
go func() {
|
||||
_ = machine.Take(ctx, func(ctx context.Context, work Build) {
|
||||
mu.Lock()
|
||||
counts[work.Request().Repository]++
|
||||
mu.Unlock()
|
||||
took <- work.Request().Repository
|
||||
_ = work.Announce(ctx, BuildResult{ID: work.Request().ID, On: "anchor", Failed: "no"})
|
||||
_ = work.Done()
|
||||
})
|
||||
}()
|
||||
return took, func() map[string]int {
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
out := map[string]int{}
|
||||
for k, v := range counts {
|
||||
out[k] = v
|
||||
}
|
||||
return out
|
||||
}
|
||||
}
|
||||
|
||||
// A move that published its copy and stopped before deleting the original leaves two copies of one ask;
|
||||
// the next pass makes them one, and a copy of one a holder already took is removed, never built twice.
|
||||
func TestNatsAMoveCutShortLeavesOneCopyAfterTheNextPass(t *testing.T) {
|
||||
js := aBusWithTheBuildRole(t)
|
||||
ctx := t.Context()
|
||||
now := time.Now()
|
||||
check := aCheck(now.Add(-2*time.Minute), "/c")
|
||||
publishAsk(t, js, check)
|
||||
publishAsk(t, js, aDelivery(now.Add(-time.Minute), "/d"))
|
||||
moveCopyOf(t, js, check.ID) // the copy, published; the original never deleted
|
||||
if pass, err := PutInOrder(ctx, js, TheBuildMachine); err != nil || len(pass.Removed) != 1 {
|
||||
t.Fatalf("removed %v: %v", pass.Removed, err)
|
||||
}
|
||||
var got []string
|
||||
for _, a := range queued(t, js) {
|
||||
got = append(got, a.Repository)
|
||||
}
|
||||
if want := []string{"/d", "/c"}; !reflect.DeepEqual(got, want) {
|
||||
t.Fatalf("after the pass the queue holds %v, want %v", got, want)
|
||||
}
|
||||
|
||||
// And one a holder took before the copy was published: the copy goes, the holder's stays.
|
||||
taken := aCheck(now, "/taken")
|
||||
publishAsk(t, js, taken)
|
||||
worker, _ := broker.HolderConsumerFor("", "", broker.DeclaredSeat{Name: TheBuildMachine, Accepts: []string{"build"}})
|
||||
sub, err := js.Context().PullSubscribe(worker.Filters[0], worker.Name, nats.Bind(worker.Stream, worker.Name), nats.ManualAck())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer func() { _ = sub.Unsubscribe() }()
|
||||
// Take everything waiting up to and including /taken, and keep /taken unsettled.
|
||||
for {
|
||||
msgs, err := sub.Fetch(1, nats.MaxWait(2*time.Second))
|
||||
if err != nil || len(msgs) == 0 {
|
||||
t.Fatalf("nothing to take: %v", err)
|
||||
}
|
||||
var r BuildRequest
|
||||
_ = json.Unmarshal(msgs[0].Data, &r)
|
||||
if r.Repository == "/taken" {
|
||||
break
|
||||
}
|
||||
_ = msgs[0].Ack()
|
||||
}
|
||||
moveCopyOf(t, js, taken.ID)
|
||||
if _, err := PutInOrder(ctx, js, TheBuildMachine); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
n := 0
|
||||
for _, a := range queued(t, js) {
|
||||
if a.ID == taken.ID {
|
||||
n++
|
||||
}
|
||||
}
|
||||
if n != 1 {
|
||||
t.Fatalf("%d copies of the ask a holder took are in the queue, want 1", n)
|
||||
}
|
||||
}
|
||||
|
||||
// moveCopyOf publishes what a move publishes for the ask with this id, as a pass stopped before its delete
|
||||
// would leave it.
|
||||
func moveCopyOf(t *testing.T, js *broker.JetStream, id string) {
|
||||
t.Helper()
|
||||
for _, a := range queued(t, js) {
|
||||
if a.ID != id {
|
||||
continue
|
||||
}
|
||||
body, _ := json.Marshal(a.Request)
|
||||
msg := &nats.Msg{Subject: BuildWork(), Data: body, Header: nats.Header{}}
|
||||
msg.Header.Set(QueuedAtHeader, a.Stored.UTC().Format(time.RFC3339Nano))
|
||||
msg.Header.Set(MovedFromHeader, fmt.Sprint(a.Seq))
|
||||
if _, err := js.Context().PublishMsg(msg); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return
|
||||
}
|
||||
t.Fatalf("%s is not queued", id)
|
||||
}
|
||||
|
||||
// A dead letter delivered again whose original stayed beside it (ADR 0293: its removal failed) is a real
|
||||
// ask with the same id, and is never removed as a second copy — even when the original was itself a move's
|
||||
// copy, whose headers the delivery carries.
|
||||
func TestNatsAnAskDeliveredAgainBesideItsOriginalStays(t *testing.T) {
|
||||
js := aBusWithTheBuildRole(t)
|
||||
ctx := t.Context()
|
||||
ask := aCheck(time.Now(), "/again")
|
||||
publishAsk(t, js, aCheck(time.Now().Add(-time.Minute), "/before"))
|
||||
publishAsk(t, js, ask)
|
||||
publishAsk(t, js, aDelivery(time.Now(), "/d"))
|
||||
if _, err := PutInOrder(ctx, js, TheBuildMachine); err != nil { // /again becomes a move's copy
|
||||
t.Fatal(err)
|
||||
}
|
||||
worker, _ := broker.HolderConsumerFor("", "", broker.DeclaredSeat{Name: TheBuildMachine, Accepts: []string{"build"}})
|
||||
sub, err := js.Context().PullSubscribe(worker.Filters[0], worker.Name, nats.Bind(worker.Stream, worker.Name), nats.ManualAck())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer func() { _ = sub.Unsubscribe() }()
|
||||
var original *nats.Msg
|
||||
for original == nil {
|
||||
msgs, err := sub.Fetch(1, nats.MaxWait(2*time.Second))
|
||||
if err != nil || len(msgs) == 0 {
|
||||
t.Fatalf("nothing to take: %v", err)
|
||||
}
|
||||
var r BuildRequest
|
||||
_ = json.Unmarshal(msgs[0].Data, &r)
|
||||
if r.ID == ask.ID {
|
||||
original = msgs[0] // given up on, never settled: dead, and kept
|
||||
continue
|
||||
}
|
||||
_ = msgs[0].Ack()
|
||||
}
|
||||
// Delivered again as DeliverAgain does: its headers, and its own.
|
||||
again := &nats.Msg{Subject: BuildWork(), Data: original.Data, Header: nats.Header{}}
|
||||
for k, v := range original.Header {
|
||||
again.Header[k] = v
|
||||
}
|
||||
again.Header.Set(AgainHeader, "7")
|
||||
if _, err := js.Context().PublishMsg(again); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
pass, err := PutInOrder(ctx, js, TheBuildMachine)
|
||||
if err != nil || len(pass.Removed) != 0 {
|
||||
t.Fatalf("removed %v: %v", pass.Removed, err)
|
||||
}
|
||||
n := 0
|
||||
for _, a := range queued(t, js) {
|
||||
if a.ID == ask.ID {
|
||||
n++
|
||||
}
|
||||
}
|
||||
if n != 2 {
|
||||
t.Fatalf("%d messages of the ask delivered again are queued, want its original and the delivery", n)
|
||||
}
|
||||
}
|
||||
|
||||
// Two passes at once: nothing is built twice, and nothing is lost.
|
||||
func TestNatsTwoPassesAtOnceBuildNothingTwice(t *testing.T) {
|
||||
js := aBusWithTheBuildRole(t)
|
||||
ctx, stop := context.WithCancel(t.Context())
|
||||
defer stop()
|
||||
took, counts := takeAll(t, ctx, js)
|
||||
now := time.Now()
|
||||
const asks = 24
|
||||
var passes sync.WaitGroup
|
||||
for p := 0; p < 2; p++ {
|
||||
passes.Add(1)
|
||||
go func() {
|
||||
defer passes.Done()
|
||||
for i := 0; i < 15; i++ {
|
||||
_, _ = PutInOrder(ctx, js, TheBuildMachine)
|
||||
}
|
||||
}()
|
||||
}
|
||||
for i := 0; i < asks; i++ {
|
||||
repository := fmt.Sprintf("/%02d", i)
|
||||
if i%3 == 2 {
|
||||
publishAsk(t, js, aDelivery(now.Add(time.Duration(i)*time.Millisecond), repository))
|
||||
} else {
|
||||
publishAsk(t, js, aCheck(now.Add(time.Duration(i)*time.Millisecond), repository))
|
||||
}
|
||||
}
|
||||
passes.Wait()
|
||||
for got := 0; got < asks; got++ {
|
||||
select {
|
||||
case <-took:
|
||||
case <-time.After(15 * time.Second):
|
||||
t.Fatalf("taken %d of %d: %v", got, asks, counts())
|
||||
}
|
||||
}
|
||||
time.Sleep(time.Second)
|
||||
c := counts()
|
||||
for i := 0; i < asks; i++ {
|
||||
if n := c[fmt.Sprintf("/%02d", i)]; n != 1 {
|
||||
t.Errorf("/%02d was built %d times", i, n)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// A check that has waited ten minutes goes before a delivery asked after it, even one queued ahead of it.
|
||||
func TestNatsACheckThatWaitedTenMinutesIsTakenBeforeALaterDelivery(t *testing.T) {
|
||||
js := aBusWithTheBuildRole(t)
|
||||
ctx, stop := context.WithCancel(t.Context())
|
||||
defer stop()
|
||||
now := time.Now()
|
||||
publishAsk(t, js, aDelivery(now.Add(-time.Minute), "/delivery"))
|
||||
publishAsk(t, js, aCheck(now.Add(-11*time.Minute), "/old-check"))
|
||||
if _, err := PutInOrder(ctx, js, TheBuildMachine); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
took, _ := takeAll(t, ctx, js)
|
||||
var order []string
|
||||
for len(order) < 2 {
|
||||
select {
|
||||
case r := <-took:
|
||||
order = append(order, r)
|
||||
case <-time.After(10 * time.Second):
|
||||
t.Fatalf("taken so far: %v", order)
|
||||
}
|
||||
}
|
||||
if want := []string{"/old-check", "/delivery"}; !reflect.DeepEqual(order, want) {
|
||||
t.Fatalf("taken as %v, want %v", order, want)
|
||||
}
|
||||
}
|
||||
|
||||
// A cancel that read an ask before it moved deletes the moved copy, found by its id, and says when
|
||||
// there was nothing left to delete.
|
||||
func TestNatsACancelRacingAMoveDeletesTheMovedCopy(t *testing.T) {
|
||||
js := aBusWithTheBuildRole(t)
|
||||
ctx := t.Context()
|
||||
now := time.Now()
|
||||
publishAsk(t, js, aCheck(now.Add(-time.Minute), "/c"))
|
||||
read := queued(t, js)[0]
|
||||
publishAsk(t, js, aDelivery(now, "/d"))
|
||||
if pass, err := PutInOrder(ctx, js, TheBuildMachine); err != nil || len(pass.Moved) != 1 {
|
||||
t.Fatalf("moved %v: %v", pass.Moved, err)
|
||||
}
|
||||
removed, err := DeleteAsk(ctx, js, TheBuildMachine, read)
|
||||
if err != nil || removed == 0 || removed == read.Seq {
|
||||
t.Fatalf("removed seq %d (read at %d): %v", removed, read.Seq, err)
|
||||
}
|
||||
for _, a := range queued(t, js) {
|
||||
if a.ID == read.ID {
|
||||
t.Fatalf("the cancelled ask is still queued at seq %d", a.Seq)
|
||||
}
|
||||
}
|
||||
if removed, err := DeleteAsk(ctx, js, TheBuildMachine, read); err != nil || removed != 0 {
|
||||
t.Fatalf("an ask gone already is said removed at %d: %v", removed, err)
|
||||
}
|
||||
}
|
||||
|
||||
// An ask nothing can be read from keeps its place in time when it moves.
|
||||
func TestNatsAnUnreadableAskKeepsItsAgeWhenItMoves(t *testing.T) {
|
||||
js := aBusWithTheBuildRole(t)
|
||||
ctx := t.Context()
|
||||
if _, err := js.Context().Publish(BuildWork(), []byte("not an ask")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
first := queued(t, js)[0].Stored
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
publishAsk(t, js, aDelivery(time.Now(), "/d"))
|
||||
if pass, err := PutInOrder(ctx, js, TheBuildMachine); err != nil || len(pass.Moved) != 1 {
|
||||
t.Fatalf("moved %v: %v", pass.Moved, err)
|
||||
}
|
||||
asks := queued(t, js)
|
||||
if len(asks) != 2 || asks[1].ID != "" || !asks[1].Stored.Equal(first) {
|
||||
t.Fatalf("after the move the unreadable ask was taken at %v, first at %v: %+v", asks[1].Stored, first, asks)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,138 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/binary"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgconn"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
// CreateMigratedForTest creates the database name on the test PostgreSQL that admin reaches, holding
|
||||
// every one of migrations already applied and recorded, as a fresh database migrated by Migrate would.
|
||||
//
|
||||
// **A copy of a template, migrated once** (novox/hq issue 591): a store test used to create an empty
|
||||
// database and run every migration into it, some 90 for the inventory, and with hundreds of store tests
|
||||
// that alone was most of the controller's check. The template is migrated by Migrate itself, once for
|
||||
// each set of migrations, and every test's database is copied from it by PostgreSQL. The caller still
|
||||
// runs Migrate on the copy, which finds nothing pending only if the copy's record matches the migrations
|
||||
// exactly, so a template that disagreed with them would fail the test rather than pass it unseen.
|
||||
//
|
||||
// The template is named after a digest of the migrations, so a changed or added migration is a new
|
||||
// template, never an old one reused. Packages tested at once in separate processes share one: it is
|
||||
// built under an advisory lock, and a template a crashed build left half done is not marked a template
|
||||
// and is built again. It takes a *testing.T, so nothing that is not a test can call it.
|
||||
func CreateMigratedForTest(t *testing.T, admin, name string, migrations []Migration) {
|
||||
t.Helper()
|
||||
template, err := templateFor(admin, migrations)
|
||||
if err != nil {
|
||||
t.Fatalf("cannot raise the migrated template: %v", err)
|
||||
}
|
||||
conn, err := pgx.Connect(t.Context(), admin)
|
||||
if err != nil {
|
||||
t.Fatalf("cannot reach the test PostgreSQL: %v", err)
|
||||
}
|
||||
defer conn.Close(context.Background())
|
||||
// A copy is refused while anything is connected to its template; the template allows no connection
|
||||
// once built, so only the builder's last connection closing can still be in the way.
|
||||
for deadline := time.Now().Add(10 * time.Second); ; {
|
||||
_, err = conn.Exec(t.Context(), "create database "+name+" template "+template)
|
||||
var pgErr *pgconn.PgError
|
||||
if err == nil || !errors.As(err, &pgErr) || pgErr.Code != "55006" || time.Now().After(deadline) {
|
||||
break
|
||||
}
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
}
|
||||
if err != nil {
|
||||
t.Fatalf("cannot create %s from %s: %v", name, template, err)
|
||||
}
|
||||
}
|
||||
|
||||
// templates are the templates this process has raised or found, by name.
|
||||
var templates sync.Map // name → *templateOnce
|
||||
|
||||
type templateOnce struct {
|
||||
once sync.Once
|
||||
err error
|
||||
}
|
||||
|
||||
// templateFor is the name of the template holding migrations, raised if no process has yet.
|
||||
func templateFor(admin string, migrations []Migration) (string, error) {
|
||||
digest := sha256.New()
|
||||
for _, m := range migrations {
|
||||
fmt.Fprintf(digest, "%d %s %s\n", m.Number, m.Name, m.Checksum)
|
||||
}
|
||||
sum := digest.Sum(nil)
|
||||
name := "tpl_" + hex.EncodeToString(sum[:12])
|
||||
held, _ := templates.LoadOrStore(name, &templateOnce{})
|
||||
once := held.(*templateOnce)
|
||||
once.once.Do(func() {
|
||||
// Not the asking test's context: the template outlives that test, and is raised once for all.
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
|
||||
defer cancel()
|
||||
once.err = raiseTemplate(ctx, admin, name, int64(binary.BigEndian.Uint64(sum[12:20])), migrations)
|
||||
})
|
||||
return name, once.err
|
||||
}
|
||||
|
||||
// raiseTemplate builds the template name under the advisory lock key, unless it stands built.
|
||||
func raiseTemplate(ctx context.Context, admin, name string, key int64, migrations []Migration) error {
|
||||
conn, err := pgx.Connect(ctx, admin)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer conn.Close(context.Background())
|
||||
if _, err := conn.Exec(ctx, `select pg_advisory_lock($1)`, key); err != nil {
|
||||
return err
|
||||
}
|
||||
defer func() { _, _ = conn.Exec(context.Background(), `select pg_advisory_unlock($1)`, key) }()
|
||||
|
||||
var built bool
|
||||
if err := conn.QueryRow(ctx, `select exists (select 1 from pg_database where datname = $1 and datistemplate)`,
|
||||
name).Scan(&built); err != nil {
|
||||
return err
|
||||
}
|
||||
if built {
|
||||
return nil
|
||||
}
|
||||
if _, err := conn.Exec(ctx, "drop database if exists "+name+" with (force)"); err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := conn.Exec(ctx, "create database "+name); err != nil {
|
||||
return err
|
||||
}
|
||||
cut := strings.LastIndex(admin, "/")
|
||||
rest := ""
|
||||
if q := strings.Index(admin[cut:], "?"); q >= 0 {
|
||||
rest = admin[cut+q:]
|
||||
}
|
||||
pool, err := pgxpool.New(ctx, admin[:cut]+"/"+name+rest)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
_, err = (&Store{context: "template", pool: pool}).Migrate(ctx, migrations)
|
||||
pool.Close()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
// Marked a template only once whole, and closed to connections, so a copy is never refused for one.
|
||||
for _, sql := range []string{
|
||||
"alter database " + name + " with allow_connections false",
|
||||
"select pg_terminate_backend(pid) from pg_stat_activity where datname = '" + name + "'",
|
||||
"alter database " + name + " with is_template true",
|
||||
} {
|
||||
if _, err := conn.Exec(ctx, sql); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
Vendored
+3
@@ -35,3 +35,6 @@
|
||||
./cmd/mesh-controller TestARollbackWhoseSendWasRefusedIsCompletedByTheNextSend
|
||||
./cmd/mesh-controller TestAGateJudgesTheOtherMovesBesideAKeptCore
|
||||
./cmd/mesh-controller TestReplay401
|
||||
./cmd/mesh-controller TestReplay201ThePlansSend
|
||||
./internal/inventory TestAWalksRestIsReadFromEachMachinesOwnSend
|
||||
./internal/inventory TestAReportOfABusyContainerIsKeptAndReadBack
|
||||
|
||||
+87
@@ -0,0 +1,87 @@
|
||||
package declaration
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Busy is how a long-running container says it holds work that would be lost if it were stopped now
|
||||
// (novox/hq ADR 0335 §1): a command run inside it, by its shell, as an exec health check is, that exits 0
|
||||
// while it is busy and non-zero when it is idle. Being asked is also the container's notice that it is
|
||||
// about to be replaced: from the first ask it takes no new work, so its answer ends once the work in hand
|
||||
// ends.
|
||||
//
|
||||
// **Not part of what the container is.** Busy is not in the container's spec: a changed command changes
|
||||
// when the container is replaced, never whether — a container is not recreated for its busy command alone.
|
||||
type Busy struct {
|
||||
Command string `json:"command"`
|
||||
// Timeout is how long one ask may take; empty is BusyTimeoutDefault.
|
||||
Timeout string `json:"timeout,omitempty"`
|
||||
}
|
||||
|
||||
// The bounds of an ask (novox/hq ADR 0335).
|
||||
const (
|
||||
BusyTimeoutDefault = 10 * time.Second
|
||||
// BusyTimeoutCeiling is the longest one ask may take: an ask runs inside an apply, and an apply waits
|
||||
// for nothing longer than a look.
|
||||
BusyTimeoutCeiling = time.Minute
|
||||
// BusyBound is how long a container that says busy is left running after the first apply that would
|
||||
// have replaced it (ADR 0335 §3). The node-engine's, not the module's: no module holds its own
|
||||
// replacement for ever.
|
||||
BusyBound = 15 * time.Minute
|
||||
)
|
||||
|
||||
// UnmarshalJSON reads a busy strictly, as everything a declaration carries is read.
|
||||
func (b *Busy) UnmarshalJSON(raw []byte) error {
|
||||
type plain Busy
|
||||
var p plain
|
||||
dec := json.NewDecoder(bytes.NewReader(raw))
|
||||
dec.DisallowUnknownFields()
|
||||
if err := dec.Decode(&p); err != nil {
|
||||
return fmt.Errorf("busy: %w", err)
|
||||
}
|
||||
*b = Busy(p)
|
||||
return nil
|
||||
}
|
||||
|
||||
// Within is how long one ask may take. Validated on arrival, so an unreadable timeout is the default.
|
||||
func (b *Busy) Within() time.Duration {
|
||||
if d, err := time.ParseDuration(b.Timeout); err == nil && d > 0 {
|
||||
return d
|
||||
}
|
||||
return BusyTimeoutDefault
|
||||
}
|
||||
|
||||
// problems holds a container's busy to its shape: a one-line command, a timeout within its bounds, and only
|
||||
// on a container that stays up.
|
||||
func (b *Busy) problems(where string, longRunning bool) []string {
|
||||
if b == nil {
|
||||
return nil
|
||||
}
|
||||
var problems []string
|
||||
say := func(format string, args ...any) {
|
||||
problems = append(problems, where+": "+fmt.Sprintf(format, args...))
|
||||
}
|
||||
if !longRunning {
|
||||
say("busy is asked of a container that stays up; a step or a scheduled run is never replaced mid-run")
|
||||
}
|
||||
if strings.TrimSpace(b.Command) == "" {
|
||||
say("busy needs a command")
|
||||
}
|
||||
if strings.ContainsAny(b.Command, "\n\r") {
|
||||
say("busy's command is one line")
|
||||
}
|
||||
if b.Timeout != "" {
|
||||
d, err := time.ParseDuration(b.Timeout)
|
||||
switch {
|
||||
case err != nil:
|
||||
say("busy's timeout is a duration, not %q", b.Timeout)
|
||||
case d <= 0 || d > BusyTimeoutCeiling:
|
||||
say("an ask of busy takes more than nothing and at most %s, not %s", BusyTimeoutCeiling, d)
|
||||
}
|
||||
}
|
||||
return problems
|
||||
}
|
||||
+354
-5
@@ -12,6 +12,7 @@ import (
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"path/filepath"
|
||||
"reflect"
|
||||
"regexp"
|
||||
"slices"
|
||||
@@ -132,7 +133,13 @@ func (d *Directory) validate(where string, _ bool) []string {
|
||||
if d.Path == "" {
|
||||
problems = append(problems, where+": a directory needs a path")
|
||||
}
|
||||
return append(problems, checkMode(where, d.Mode)...)
|
||||
// A directory's mode may carry the setgid bit (2xxx), so what is made beneath it keeps its group (novox/hq ADR
|
||||
// 0334 §1); never setuid or sticky, which mean nothing the mesh asks of a directory it owns.
|
||||
mode := d.Mode
|
||||
if len(mode) == 4 && mode[0] == '2' {
|
||||
mode = "0" + mode[1:]
|
||||
}
|
||||
return append(problems, checkMode(where, mode)...)
|
||||
}
|
||||
|
||||
// File is a file with literal content. The host renders nothing.
|
||||
@@ -456,14 +463,51 @@ const (
|
||||
// **The distinction 04-ISSUES/036 and 026 turn on.** A `directory` resource is the mesh's own —
|
||||
// it creates it, sets its owner and mode, and removes it when empty ([ADR 0030](novox/hq)). A
|
||||
// media library, a download spool is the operator's: it existed before the mesh, several modules
|
||||
// read and write it at once, and the mesh must not create, chown, reconcile or remove it. The
|
||||
// host confirms it is there and mounts it; nothing else.
|
||||
// read and write it at once, and the mesh must not create, chown, chmod or remove it. The host
|
||||
// confirms it is there and mounts it; nothing else. Library data is a library's (novox/hq ADR 0328),
|
||||
// not an access.
|
||||
type Access struct {
|
||||
ID string `json:"id"`
|
||||
Type Type `json:"type"`
|
||||
Path string `json:"path"`
|
||||
// Mode is how this module reaches the path: read or read-write. Absent narrows to read.
|
||||
Mode string `json:"mode,omitempty"`
|
||||
|
||||
// Users and Taken are what a controller speaking ADR 0325 sends (the identities that use the folder, and
|
||||
// that the operator's warrant was carried out): accepted, so its declaration is not refused whole, and not
|
||||
// acted on since ADR 0328 superseded ADR 0325 — no ACL entry is kept.
|
||||
Users []AccessUser `json:"users,omitempty"`
|
||||
Taken bool `json:"taken,omitempty"`
|
||||
|
||||
// Grant is what a controller older than ADR 0325 still sends (ADR 0312's group grant, superseded): accepted
|
||||
// so its declaration is not refused whole, and never acted on — no account is put in a group for data (§5).
|
||||
Grant json.RawMessage `json:"grant,omitempty"`
|
||||
}
|
||||
|
||||
// The uses an access's user has (ADR 0051 §3), and where it runs (ADR 0325 §4).
|
||||
const (
|
||||
FromContainer = "container"
|
||||
FromToolRunner = "tool-runner"
|
||||
)
|
||||
|
||||
// AccessUser is one identity that uses an accessed folder: its name and uid as the controller allocated them, its
|
||||
// use, and where it runs. A tool runner's account may come with no uid (0 or absent): the engine reads it by name.
|
||||
type AccessUser struct {
|
||||
Identity string `json:"identity"`
|
||||
UID int `json:"uid,omitempty"`
|
||||
Use string `json:"use"`
|
||||
From string `json:"from"`
|
||||
}
|
||||
|
||||
// UnmarshalJSON reads a user as strictly as a resource is read.
|
||||
func (u *AccessUser) UnmarshalJSON(raw []byte) error {
|
||||
type plain AccessUser
|
||||
var p plain
|
||||
if err := strictDecode(raw, &p); err != nil {
|
||||
return fmt.Errorf("an access's user: %w", err)
|
||||
}
|
||||
*u = AccessUser(p)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (a *Access) Identity() string { return a.ID }
|
||||
@@ -482,9 +526,43 @@ func (a *Access) validate(where string, _ bool) []string {
|
||||
problems = append(problems, fmt.Sprintf(
|
||||
"%s: an access is %q or %q, not %q", where, AccessRead, AccessReadWrite, a.Mode))
|
||||
}
|
||||
for _, u := range a.Users {
|
||||
switch {
|
||||
case u.Identity == "" || !accountName.MatchString(u.Identity):
|
||||
problems = append(problems, fmt.Sprintf("%s: a user of an access is an account's name, and %q is not one",
|
||||
where, u.Identity))
|
||||
case u.UID < 0 || (u.UID == 0 && u.From != FromToolRunner):
|
||||
// Never root (novox/hq ADR 0325 §11): an entry for uid 0 would be a right root has anyway, and a
|
||||
// number below it is no account. A tool runner's account alone may come without its number: the
|
||||
// controller cannot know the operator's account's, and the engine reads it by name
|
||||
// (apply.ResolveRunners), refusing one that is root's.
|
||||
problems = append(problems, fmt.Sprintf("%s: the user %s has uid %d, and a module's identity is never "+
|
||||
"root or below it (novox/hq ADR 0325 §11)", where, u.Identity, u.UID))
|
||||
}
|
||||
if u.Use != AccessRead && u.Use != AccessReadWrite {
|
||||
problems = append(problems, fmt.Sprintf("%s: the user %s uses it %q, and a use is %q or %q", where,
|
||||
u.Identity, u.Use, AccessRead, AccessReadWrite))
|
||||
}
|
||||
if u.Use == AccessReadWrite && a.Mode != AccessReadWrite {
|
||||
problems = append(problems, fmt.Sprintf("%s: the user %s writes an access declared %s", where,
|
||||
u.Identity, orRead(a.Mode)))
|
||||
}
|
||||
if u.From != FromContainer && u.From != FromToolRunner {
|
||||
problems = append(problems, fmt.Sprintf("%s: the user %s runs from %q, and that is %q or %q", where,
|
||||
u.Identity, u.From, FromContainer, FromToolRunner))
|
||||
}
|
||||
}
|
||||
return problems
|
||||
}
|
||||
|
||||
// orRead is an access's mode as said: absent narrows to read.
|
||||
func orRead(mode string) string {
|
||||
if mode == "" {
|
||||
return AccessRead
|
||||
}
|
||||
return mode
|
||||
}
|
||||
|
||||
func (u *User) Identity() string { return u.ID }
|
||||
func (u *User) Kind() Type { return TypeUser }
|
||||
func (u *User) Target() string { return u.Name }
|
||||
@@ -1160,6 +1238,83 @@ type Container struct {
|
||||
// Health is how this resource is ready (novox/hq ADR 0240 rule 2, Phase B): one kind and its
|
||||
// timing, judged by the node-engine beside liveness. Absent: judged alive or not, and nothing more.
|
||||
Health *Health `json:"health,omitempty"`
|
||||
|
||||
// Busy is how this container says it holds work a replacement would lose (novox/hq ADR 0335): asked
|
||||
// before it is replaced, and while it says busy it is left running, within the bound. Absent: never
|
||||
// busy, and replaced at once, as before.
|
||||
Busy *Busy `json:"busy,omitempty"`
|
||||
|
||||
// RunAs is who the container runs as, resolved by the controller from the module's identity and the
|
||||
// image's declared interface (novox/hq ADR 0325 §2, §3): told by environment (the image reads PUID and
|
||||
// PGID, say) or by the runtime's own user option. Part of the spec, so a change recreates it. Absent: the
|
||||
// container runs as its image says. Its key is `identity`, the manifest's word; the field is not, because
|
||||
// a resource's Identity is its id.
|
||||
RunAs *ContainerIdentity `json:"identity,omitempty"`
|
||||
}
|
||||
|
||||
// The ways an image is told who to run as (novox/hq ADR 0325 §2).
|
||||
const (
|
||||
IdentityByEnv = "env"
|
||||
IdentityByRun = "run"
|
||||
)
|
||||
|
||||
// ContainerIdentity is who a container runs as, and how its image is told.
|
||||
type ContainerIdentity struct {
|
||||
// Mode is IdentityByEnv or IdentityByRun.
|
||||
Mode string `json:"mode"`
|
||||
// UserEnv and GroupEnv are the variables the image reads the user's and the group's number from, for
|
||||
// IdentityByEnv: "PUID", "PGID".
|
||||
UserEnv string `json:"user_env,omitempty"`
|
||||
GroupEnv string `json:"group_env,omitempty"`
|
||||
UID int `json:"uid"`
|
||||
GID int `json:"gid"`
|
||||
}
|
||||
|
||||
// UnmarshalJSON reads an identity as strictly as a resource is read.
|
||||
func (i *ContainerIdentity) UnmarshalJSON(raw []byte) error {
|
||||
type plain ContainerIdentity
|
||||
var p plain
|
||||
if err := strictDecode(raw, &p); err != nil {
|
||||
return fmt.Errorf("a container's identity: %w", err)
|
||||
}
|
||||
*i = ContainerIdentity(p)
|
||||
return nil
|
||||
}
|
||||
|
||||
// envName is a variable's name as a runtime takes it.
|
||||
var envName = regexp.MustCompile(`^[A-Za-z_][A-Za-z0-9_]*$`)
|
||||
|
||||
func (i *ContainerIdentity) problems(where string, env map[string]string) []string {
|
||||
if i == nil {
|
||||
return nil
|
||||
}
|
||||
var problems []string
|
||||
switch i.Mode {
|
||||
case IdentityByEnv:
|
||||
if !envName.MatchString(i.UserEnv) || (i.GroupEnv != "" && !envName.MatchString(i.GroupEnv)) {
|
||||
problems = append(problems, fmt.Sprintf("%s: an identity told by environment names the variables "+
|
||||
"it is told in, and %q and %q are not both names", where, i.UserEnv, i.GroupEnv))
|
||||
}
|
||||
for _, name := range []string{i.UserEnv, i.GroupEnv} {
|
||||
if _, set := env[name]; set && name != "" {
|
||||
problems = append(problems, fmt.Sprintf("%s: env sets %s, which the identity sets; one value "+
|
||||
"for one variable", where, name))
|
||||
}
|
||||
}
|
||||
case IdentityByRun:
|
||||
if i.UserEnv != "" || i.GroupEnv != "" {
|
||||
problems = append(problems, where+": an identity told by the runtime's user option names no variable")
|
||||
}
|
||||
default:
|
||||
problems = append(problems, fmt.Sprintf("%s: an identity is told by %q or %q, not %q", where,
|
||||
IdentityByEnv, IdentityByRun, i.Mode))
|
||||
}
|
||||
// Never root (novox/hq ADR 0325 §11): refused here, so nothing of the declaration is made or started.
|
||||
if i.UID <= 0 || i.GID <= 0 {
|
||||
problems = append(problems, fmt.Sprintf("%s: the identity is uid %d, gid %d, and a module's identity is "+
|
||||
"never root, nor a number below it (novox/hq ADR 0325 §11)", where, i.UID, i.GID))
|
||||
}
|
||||
return problems
|
||||
}
|
||||
|
||||
func (c *Container) Identity() string { return c.ID }
|
||||
@@ -1168,6 +1323,7 @@ func (c *Container) Target() string { return c.Name }
|
||||
|
||||
func (c *Container) validate(where string, _ bool) []string {
|
||||
problems := c.Health.problems(where, true, !c.RunOnce && c.Schedule == "")
|
||||
problems = append(problems, c.Busy.problems(where, !c.RunOnce && c.Schedule == "")...)
|
||||
if c.Name == "" {
|
||||
problems = append(problems, where+": a container needs a name")
|
||||
}
|
||||
@@ -1242,6 +1398,7 @@ func (c *Container) validate(where string, _ bool) []string {
|
||||
problems = append(problems, where+": networks is for a container that keeps running; a step "+
|
||||
"runs and exits, and joins nothing afterwards")
|
||||
}
|
||||
problems = append(problems, c.RunAs.problems(where+": container "+c.Name, c.Env)...)
|
||||
return append(problems, checkImage(where, c.Image)...)
|
||||
}
|
||||
|
||||
@@ -1392,6 +1549,185 @@ type Declaration struct {
|
||||
// the generation of what it puts back, which may be older than what the machine applied, and is
|
||||
// not refused for it.
|
||||
PutBack bool
|
||||
|
||||
// Identities is the identity of every module assigned to this machine (novox/hq ADR 0328 §4): an account
|
||||
// and a group `mesh-<module>` with the numbers the controller allocated once for the module, the same on
|
||||
// every machine. The engine makes each before any resource is applied, and removes what it made of one no
|
||||
// longer declared. Absent is a controller older than ADR 0325.
|
||||
Identities []Identity
|
||||
|
||||
// Groups is the group of every library this machine is reached by (novox/hq ADR 0328 §3): `mesh-<library>`
|
||||
// with the number the controller allocated once, mesh-wide. The engine makes each, and removes what it made of
|
||||
// one no longer declared. Absent is a controller older than ADR 0328.
|
||||
Groups []Group
|
||||
|
||||
// Libraries is every library this machine is reached by (ADR 0328 §1, §6, §9): its root here, its group, whether
|
||||
// this is its home, whether the operator's warrant to take it was carried out, the sub-paths its modules ask for,
|
||||
// and the tool runner's account to put in its group (§11). Absent is a controller older than ADR 0328.
|
||||
Libraries []Library
|
||||
}
|
||||
|
||||
// Identity is one module's identity, as the controller allocated it (ADR 0328 §4): its user `mesh-<module>` with
|
||||
// uid, its own group of the same name and number, its primary group gid (its own, or its first library's group),
|
||||
// and the groups of its other libraries as added groups.
|
||||
type Identity struct {
|
||||
Module string `json:"module"`
|
||||
Name string `json:"name"`
|
||||
UID int `json:"uid"`
|
||||
GID int `json:"gid"`
|
||||
Groups []int `json:"groups,omitempty"`
|
||||
// Runner is the tool runner's account, put in the module's own group (novox/hq ADR 0334 §2) when its tools run
|
||||
// on this machine. Absent for a module that declares no tools here, and from a controller older than ADR 0334.
|
||||
Runner string `json:"runner,omitempty"`
|
||||
}
|
||||
|
||||
// Group is one library's group, as the controller allocated it (ADR 0328 §3).
|
||||
type Group struct {
|
||||
Name string `json:"name"`
|
||||
GID int `json:"gid"`
|
||||
}
|
||||
|
||||
// GroupPrefix starts every library group's name, as IdentityPrefix starts every module's.
|
||||
const GroupPrefix = "mesh-"
|
||||
|
||||
// LibraryAccessPrefix starts the access a library's wait and its take act name: `library:<name>`.
|
||||
const LibraryAccessPrefix = "library:"
|
||||
|
||||
// Library is one library as this machine sees it (ADR 0328).
|
||||
type Library struct {
|
||||
Library string `json:"library"`
|
||||
// Root is the library's root on this machine: on its home the data itself, elsewhere the mount point the mounts
|
||||
// module mounts it at.
|
||||
Root string `json:"root"`
|
||||
Group string `json:"group"`
|
||||
GID int `json:"gid"`
|
||||
// Home is that the library's data lives on this machine: only there does the engine keep it.
|
||||
Home bool `json:"home"`
|
||||
// Taken is the controller saying the operator's warrant to take it was carried out (§7).
|
||||
Taken bool `json:"taken,omitempty"`
|
||||
// Paths are the sub-paths its modules ask for, relative to the root.
|
||||
Paths []string `json:"paths,omitempty"`
|
||||
// Runner is the tool runner's account, put in the library's group (§11). Absent when no module's tools use it.
|
||||
Runner string `json:"runner,omitempty"`
|
||||
}
|
||||
|
||||
// checkLibraries judges the groups and libraries on arrival: every library's group declared, named for it, never
|
||||
// root's number; paths that stay below the root; a runner that is an account's name; and no two roots that nest
|
||||
// (§13).
|
||||
func checkLibraries(groups []Group, libs []Library) []string {
|
||||
var problems []string
|
||||
byName := map[string]int{}
|
||||
numbers := map[int]string{}
|
||||
for _, g := range groups {
|
||||
where := fmt.Sprintf("group %q", g.Name)
|
||||
if !strings.HasPrefix(g.Name, GroupPrefix) || !accountName.MatchString(g.Name) {
|
||||
problems = append(problems, fmt.Sprintf("%s: a library's group is named %s<library>", where, GroupPrefix))
|
||||
}
|
||||
if g.GID <= 0 {
|
||||
problems = append(problems, fmt.Sprintf("%s: gid %d, and a library's group is never root, nor a number "+
|
||||
"below it (novox/hq ADR 0328 §15); nothing of this declaration was made", where, g.GID))
|
||||
}
|
||||
if _, twice := byName[g.Name]; twice {
|
||||
problems = append(problems, where+": declared twice")
|
||||
}
|
||||
if other, used := numbers[g.GID]; used {
|
||||
problems = append(problems, fmt.Sprintf("%s: gid %d is %s's too", where, g.GID, other))
|
||||
}
|
||||
byName[g.Name], numbers[g.GID] = g.GID, g.Name
|
||||
}
|
||||
seen := map[string]bool{}
|
||||
for _, l := range libs {
|
||||
where := fmt.Sprintf("library %q", l.Library)
|
||||
switch {
|
||||
case l.Library == "" || strings.ContainsAny(l.Library, "/ \t\n"):
|
||||
problems = append(problems, where+": a library is named by one word")
|
||||
case seen[l.Library]:
|
||||
problems = append(problems, where+": declared twice")
|
||||
}
|
||||
seen[l.Library] = true
|
||||
if !filepath.IsAbs(l.Root) || filepath.Clean(l.Root) != l.Root || l.Root == "/" {
|
||||
problems = append(problems, fmt.Sprintf("%s: its root %q is not an absolute, clean path below /", where,
|
||||
l.Root))
|
||||
}
|
||||
if gid, ok := byName[l.Group]; !ok || gid != l.GID {
|
||||
problems = append(problems, fmt.Sprintf("%s: its group %s (gid %d) is not one this declaration makes",
|
||||
where, l.Group, l.GID))
|
||||
}
|
||||
if l.Group != GroupPrefix+l.Library {
|
||||
problems = append(problems, fmt.Sprintf("%s: its group is %s%s, and %s is not", where, GroupPrefix,
|
||||
l.Library, l.Group))
|
||||
}
|
||||
for _, p := range l.Paths {
|
||||
if p == "" || filepath.IsAbs(p) || filepath.Clean(p) != p || p == ".." || strings.HasPrefix(p, "../") {
|
||||
problems = append(problems, fmt.Sprintf("%s: the path %q is not a clean path below its root", where, p))
|
||||
}
|
||||
}
|
||||
if l.Runner != "" && !accountName.MatchString(l.Runner) {
|
||||
problems = append(problems, fmt.Sprintf("%s: its runner %q is not an account's name", where, l.Runner))
|
||||
}
|
||||
}
|
||||
for i, a := range libs {
|
||||
for _, b := range libs[i+1:] {
|
||||
if a.Root != "" && b.Root != "" && (a.Root == b.Root || strings.HasPrefix(a.Root, b.Root+"/") ||
|
||||
strings.HasPrefix(b.Root, a.Root+"/")) {
|
||||
problems = append(problems, fmt.Sprintf("library %q: its root and library %q's nest, and libraries "+
|
||||
"never overlap (novox/hq ADR 0328 §13)", a.Library, b.Library))
|
||||
}
|
||||
}
|
||||
}
|
||||
return problems
|
||||
}
|
||||
|
||||
// IdentityPrefix starts every module identity's name.
|
||||
const IdentityPrefix = "mesh-"
|
||||
|
||||
// checkIdentities judges the identities on arrival: each named for its module, never root, and no name or number
|
||||
// twice — two modules with one account would write as each other.
|
||||
func checkIdentities(ids []Identity, groups []Group) []string {
|
||||
var problems []string
|
||||
names, uids := map[string]bool{}, map[int]string{}
|
||||
libraryGroups := map[int]bool{}
|
||||
for _, g := range groups {
|
||||
libraryGroups[g.GID] = true
|
||||
}
|
||||
for _, id := range ids {
|
||||
where := fmt.Sprintf("identity %q", id.Name)
|
||||
if id.Module == "" || id.Name != IdentityPrefix+id.Module || !accountName.MatchString(id.Name) {
|
||||
problems = append(problems, fmt.Sprintf("%s: a module's identity is named %s<module>, and this one is "+
|
||||
"for the module %q", where, IdentityPrefix, id.Module))
|
||||
}
|
||||
if id.UID <= 0 || id.GID <= 0 {
|
||||
problems = append(problems, fmt.Sprintf("%s: uid %d, gid %d, and a module's identity is never root, "+
|
||||
"nor a number below it (novox/hq ADR 0328 §15); nothing of this declaration was made", where,
|
||||
id.UID, id.GID))
|
||||
}
|
||||
if names[id.Name] {
|
||||
problems = append(problems, where+": declared twice")
|
||||
}
|
||||
if other, used := uids[id.UID]; used {
|
||||
problems = append(problems, fmt.Sprintf("%s: uid %d is %s's too", where, id.UID, other))
|
||||
}
|
||||
// Its primary group is its own (gid == uid) or a library's this declaration makes (ADR 0328 §4); so is
|
||||
// every added group. A module's own number is never a library group's.
|
||||
if id.GID != id.UID && !libraryGroups[id.GID] {
|
||||
problems = append(problems, fmt.Sprintf("%s: its primary group gid %d is neither its own (%d) nor a "+
|
||||
"library's this declaration makes", where, id.GID, id.UID))
|
||||
}
|
||||
if libraryGroups[id.UID] {
|
||||
problems = append(problems, fmt.Sprintf("%s: uid %d is a library group's number too", where, id.UID))
|
||||
}
|
||||
for _, g := range id.Groups {
|
||||
if !libraryGroups[g] {
|
||||
problems = append(problems, fmt.Sprintf("%s: its added group gid %d is not a library's this "+
|
||||
"declaration makes", where, g))
|
||||
}
|
||||
}
|
||||
if id.Runner != "" && !accountName.MatchString(id.Runner) {
|
||||
problems = append(problems, fmt.Sprintf("%s: its runner %q is not an account's name", where, id.Runner))
|
||||
}
|
||||
names[id.Name], uids[id.UID] = true, id.Name
|
||||
}
|
||||
return problems
|
||||
}
|
||||
|
||||
// LeftOutModuleOf says which left-out module a recorded resource belongs to, if any: its id is the
|
||||
@@ -1561,6 +1897,13 @@ type envelope struct {
|
||||
// sends them only to a host whose reports say `reads_generation`.
|
||||
Generation int64 `json:"generation,omitempty"`
|
||||
PutBack bool `json:"put_back,omitempty"`
|
||||
// Identities is optional on the wire too (novox/hq ADR 0325 §3, ADR 0328 §4). **An older host refuses this key**; a
|
||||
// controller sends it only to a host whose health says link.IdentityContract.
|
||||
Identities []Identity `json:"identities,omitempty"`
|
||||
// Groups and Libraries are optional too (novox/hq ADR 0328). **An older host refuses these keys**; a controller
|
||||
// sends them only to a host whose health says link.LibraryContract.
|
||||
Groups []Group `json:"groups,omitempty"`
|
||||
Libraries []Library `json:"libraries,omitempty"`
|
||||
}
|
||||
|
||||
func parse(raw []byte, allowActions bool) (*Declaration, error) {
|
||||
@@ -1578,8 +1921,10 @@ func parse(raw []byte, allowActions bool) (*Declaration, error) {
|
||||
}
|
||||
|
||||
d := &Declaration{Version: env.Version, For: env.For, Adoption: env.Adoption, Sequence: env.Sequence,
|
||||
Epoch: env.Epoch, LeftOut: env.LeftOut, Generation: env.Generation, PutBack: env.PutBack}
|
||||
var problems []string
|
||||
Epoch: env.Epoch, LeftOut: env.LeftOut, Generation: env.Generation, PutBack: env.PutBack,
|
||||
Identities: env.Identities, Groups: env.Groups, Libraries: env.Libraries}
|
||||
problems := checkIdentities(env.Identities, env.Groups)
|
||||
problems = append(problems, checkLibraries(env.Groups, env.Libraries)...)
|
||||
if env.Generation < 0 {
|
||||
// Below zero would read as "none claimed" and pass every refusal of an older generation.
|
||||
problems = append(problems, fmt.Sprintf("an assignment generation below zero (%d) is not one the "+
|
||||
@@ -1732,6 +2077,10 @@ func checkWhileStopped(resources []Resource) []string {
|
||||
return problems
|
||||
}
|
||||
|
||||
// accountName is what an identity's or an access user's name may be: the portable account name the user tools
|
||||
// take, optionally ending in `$`, never starting with a dash — it is handed to them as an argument.
|
||||
var accountName = regexp.MustCompile(`^[A-Za-z0-9_.][A-Za-z0-9_.-]*\$?$`)
|
||||
|
||||
func strictDecode(raw []byte, into any) error {
|
||||
// DisallowUnknownFields is the whole point rather than strictness for its own sake: a
|
||||
// field the host does not know is a thing the control plane believes it asked for.
|
||||
|
||||
Vendored
+2
-2
@@ -78,7 +78,7 @@ github.com/nats-io/nkeys
|
||||
# github.com/nats-io/nuid v1.0.1
|
||||
## explicit
|
||||
github.com/nats-io/nuid
|
||||
# github.com/novox/mesh-host v0.0.0 => git.novox.be/novox/mesh-host v0.0.0-20261011092230-20d5af9b2d5f
|
||||
# github.com/novox/mesh-host v0.0.0 => git.novox.be/novox/mesh-host v0.0.0-20261012224132-bb60e7b584d8
|
||||
## explicit; go 1.26.0
|
||||
github.com/novox/mesh-host/internal/declaration
|
||||
github.com/novox/mesh-host/rootsearch
|
||||
@@ -138,4 +138,4 @@ golang.org/x/text/width
|
||||
# golang.org/x/time v0.15.0
|
||||
## explicit; go 1.25.0
|
||||
golang.org/x/time/rate
|
||||
# github.com/novox/mesh-host => git.novox.be/novox/mesh-host v0.0.0-20261011092230-20d5af9b2d5f
|
||||
# github.com/novox/mesh-host => git.novox.be/novox/mesh-host v0.0.0-20261012224132-bb60e7b584d8
|
||||
|
||||
Reference in New Issue
Block a user