Files
mesh-host/internal/apply/busy_test.go
T
jschoubben 695e022991 Keep a replacement owed after a busy wait, and clamp a wait from a later clock
A restart-on trigger that is no file counts as changed only in its own pass,
so after a busy wait the container read unchanged and was never replaced,
while the report stopped listing it (review of #98).
2026-10-13 23:59:58 +02:00

434 lines
17 KiB
Go

package apply
import (
"context"
"errors"
"path/filepath"
"slices"
"strings"
"sync"
"testing"
"time"
"github.com/novox/mesh-host/internal/declaration"
"github.com/novox/mesh-host/internal/store"
)
// A container that says it is busy is replaced once it is idle, within a bound (novox/hq ADR 0335). The
// machine of these tests runs two containers, a build agent that declares `busy` and a web server that
// does not; every call that replaces a container is counted.
type aBusyMachine struct {
mu sync.Mutex
// containers is each container's name and the spec it was created with; running whether it runs.
containers map[string]string
busy bool // what the agent's busy command answers
said string // what it prints
asked int
calls []string
}
func newBusyMachine() *aBusyMachine {
return &aBusyMachine{containers: map[string]string{}}
}
func (m *aBusyMachine) run(_ context.Context, _ string, args ...string) (string, error) {
m.mu.Lock()
defer m.mu.Unlock()
switch args[0] {
case "info", "image":
return "27.0\n", nil
case "container": // inspect
name := args[len(args)-1]
spec, there := m.containers[name]
if !there {
return "", errors.New("no such container")
}
return "true\t" + spec, nil
case "exec":
m.asked++
m.calls = append(m.calls, "ask "+args[1])
if strings.Join(args[2:], " ") != "/bin/sh -c mesh-builder busy" {
return "", errors.New("asked something else: " + strings.Join(args[2:], " "))
}
if m.busy {
return m.said + "\n", nil
}
return m.said + "\n", errors.New("exited 1")
case "rm":
name := args[len(args)-1]
m.calls = append(m.calls, "rm "+name)
delete(m.containers, name)
case "run":
name := args[slices.Index(args, "--name")+1]
m.calls = append(m.calls, "create "+name)
for i, a := range args {
if a == "--label" && strings.HasPrefix(args[i+1], specLabel+"=") {
m.containers[name] = strings.TrimPrefix(args[i+1], specLabel+"=")
}
}
}
return "", nil
}
func (m *aBusyMachine) mark() int {
m.mu.Lock()
defer m.mu.Unlock()
return len(m.calls)
}
// since is every call since mark that replaced a container, or asked one.
func (m *aBusyMachine) since(mark int) []string {
m.mu.Lock()
defer m.mu.Unlock()
return append([]string(nil), m.calls[mark:]...)
}
func (m *aBusyMachine) answer(busy bool, said string) {
m.mu.Lock()
defer m.mu.Unlock()
m.busy, m.said = busy, said
}
// twoContainers is the declaration at a version: v moves the agent's image tag through its env, w the web's.
func twoContainers(t *testing.T, agent, web string) *declaration.Declaration {
t.Helper()
return parse(t, `{"declaration":1,"resources":[
{"id":"build-agent.server","type":"container","name":"mesh-build-agent","image":"`+pinned+`",
"env":{"V":"`+agent+`"},"busy":{"command":"mesh-builder busy","timeout":"5s"}},
{"id":"web.server","type":"container","name":"web","image":"`+pinned+`","env":{"V":"`+web+`"}}
]}`)
}
// atClock moves the bound's clock for one test.
func atClock(t *testing.T, at *time.Time) {
t.Helper()
was := busyClock
busyClock = func() time.Time { return *at }
t.Cleanup(func() { busyClock = was })
}
func busyOf(r Report, id string) (BusyWait, bool) {
for _, b := range r.Busy {
if b.ID == id {
return b, true
}
}
return BusyWait{}, false
}
func applyBusy(t *testing.T, m *aBusyMachine, d *declaration.Declaration, known store.State) (Report, store.State) {
t.Helper()
report, state, err := Apply(context.Background(), archHost(t), d, known, store.OriginDeclared, m.run, nil, nil)
if err != nil {
t.Fatalf("the apply failed: %v", err)
}
return report, state
}
// The decision's first check: an apply that would replace a container whose busy command exits 0 leaves it
// running, applies the rest, and replaces it on the first pass the command exits non-zero.
func TestABusyContainerIsLeftRunningAndReplacedOnTheFirstPassItSaysIdle(t *testing.T) {
now := time.Date(2026, 10, 13, 1, 0, 0, 0, time.UTC)
atClock(t, &now)
m := newBusyMachine()
_, state := applyBusy(t, m, twoContainers(t, "1", "1"), store.State{})
// A merge moves both: the agent holds a build, the web server holds nothing.
m.answer(true, "building novox/mesh-controller (check of #280) since 00:58")
mark := m.mark()
report, state := applyBusy(t, m, twoContainers(t, "2", "2"), state)
if got := m.since(mark); !slices.Equal(got, []string{"ask mesh-build-agent", "rm web", "create web"}) {
t.Fatalf("a busy agent was not left running while the rest applied: %v", got)
}
o := outcomeOf(report, "build-agent.server")
if o.Action != waitsBusy {
t.Fatalf("a busy container was reported %q, want %q: %+v", o.Action, waitsBusy, o)
}
for _, says := range []string{"busy", "first pass it says idle", "check of #280", "ADR 0335"} {
if !strings.Contains(o.Detail, says) {
t.Errorf("the outcome does not say %q: %s", says, o.Detail)
}
}
if (Report{Outcomes: []Outcome{o}}).Changed() {
t.Errorf("a container left running reads as a change: %+v", o)
}
b, said := busyOf(report, "build-agent.server")
if !said || b.State != BusyWaiting || !b.Since.Equal(now) || !b.Until.Equal(now.Add(15*time.Minute)) ||
b.Container != "mesh-build-agent" || !strings.Contains(b.Said, "check of #280") {
t.Fatalf("the report does not say the agent waits, since when and until when: %+v", report.Busy)
}
if !report.Waiting() {
t.Error("a report with a container waiting does not say so")
}
if rec, _ := state.Find("build-agent.server"); rec.BusySince == nil || !rec.BusySince.Equal(now) {
t.Fatalf("the wait is not kept in the container's record: %+v", rec)
}
// Asked again a minute later, still busy: still left running, the wait counted from the first apply.
now = now.Add(time.Minute)
mark = m.mark()
report, state = applyBusy(t, m, twoContainers(t, "2", "2"), state)
if got := m.since(mark); !slices.Equal(got, []string{"ask mesh-build-agent"}) {
t.Fatalf("a second pass while busy did more than ask: %v", got)
}
if b, _ := busyOf(report, "build-agent.server"); b.State != BusyWaiting || !b.Since.Equal(now.Add(-time.Minute)) {
t.Fatalf("the second pass does not keep the wait from the first apply: %+v", report.Busy)
}
// The build ends: the next pass replaces it, and says it waited.
now = now.Add(time.Minute)
m.answer(false, "idle")
mark = m.mark()
report, state = applyBusy(t, m, twoContainers(t, "2", "2"), state)
if got := m.since(mark); !slices.Equal(got, []string{"ask mesh-build-agent", "rm mesh-build-agent", "create mesh-build-agent"}) {
t.Fatalf("the agent was not replaced on the first pass it said idle: %v", got)
}
if o := outcomeOf(report, "build-agent.server"); o.Action != "updated" || !strings.Contains(o.Detail, "once it said idle") {
t.Errorf("the replacement does not say it waited: %+v", o)
}
if b, _ := busyOf(report, "build-agent.server"); b.State != BusyReplacedIdle {
t.Errorf("the report does not say the agent was replaced when idle: %+v", report.Busy)
}
if report.Waiting() {
t.Error("a report with nothing waiting says something waits")
}
if rec, _ := state.Find("build-agent.server"); rec.BusySince != nil {
t.Errorf("the wait is still kept after the replacement: %+v", rec)
}
// And the pass after that has nothing to say about it.
report, _ = applyBusy(t, m, twoContainers(t, "2", "2"), state)
if len(report.Busy) != 0 {
t.Errorf("a container replaced is still said: %+v", report.Busy)
}
}
// An idle container is replaced at once, as one that declares nothing is: asked once, nothing waits.
func TestAnIdleContainerIsReplacedAtOnce(t *testing.T) {
m := newBusyMachine()
_, state := applyBusy(t, m, twoContainers(t, "1", "1"), store.State{})
m.answer(false, "")
mark := m.mark()
report, _ := applyBusy(t, m, twoContainers(t, "2", "1"), state)
if got := m.since(mark); !slices.Equal(got, []string{"ask mesh-build-agent", "rm mesh-build-agent", "create mesh-build-agent"}) {
t.Fatalf("an idle agent was not replaced at once: %v", got)
}
if len(report.Busy) != 0 {
t.Errorf("an idle container replaced at once is said as a wait: %+v", report.Busy)
}
// And a container whose declaration did not move is never asked.
mark = m.mark()
applyBusy(t, m, twoContainers(t, "2", "1"), state)
if got := m.since(mark); slices.Contains(got, "ask mesh-build-agent") {
t.Errorf("a container nothing replaces was asked: %v", got)
}
}
// Past the bound it is replaced anyway, the report carrying the command's last output — and the bound holds
// across a restart of the node-engine: the wait is read back from the state on disk.
func TestABusyContainerIsReplacedAtTheBoundEvenAfterARestart(t *testing.T) {
now := time.Date(2026, 10, 13, 1, 0, 0, 0, time.UTC)
atClock(t, &now)
m := newBusyMachine()
_, state := applyBusy(t, m, twoContainers(t, "1", "1"), store.State{})
m.answer(true, "building novox/mesh-host (a check that never ends)")
_, state = applyBusy(t, m, twoContainers(t, "2", "1"), state)
// The node-engine stops and starts again: what it knows is what it saved.
path := filepath.Join(t.TempDir(), "state.json")
if err := store.Save(path, state); err != nil {
t.Fatal(err)
}
reloaded, err := store.Load(path)
if err != nil {
t.Fatal(err)
}
now = now.Add(14 * time.Minute)
mark := m.mark()
report, reloaded := applyBusy(t, m, twoContainers(t, "2", "1"), reloaded)
if got := m.since(mark); slices.Contains(got, "rm mesh-build-agent") {
t.Fatalf("a container busy within the bound was replaced after a restart: %v", got)
}
if b, _ := busyOf(report, "build-agent.server"); b.State != BusyWaiting || !b.Since.Equal(now.Add(-14*time.Minute)) {
t.Fatalf("a restart started the wait again: %+v", report.Busy)
}
now = now.Add(time.Minute)
mark = m.mark()
report, _ = applyBusy(t, m, twoContainers(t, "2", "1"), reloaded)
if got := m.since(mark); !slices.Equal(got, []string{"ask mesh-build-agent", "rm mesh-build-agent", "create mesh-build-agent"}) {
t.Fatalf("a container still busy at the bound was not replaced: %v", got)
}
b, _ := busyOf(report, "build-agent.server")
if b.State != BusyReplacedAtBound || !strings.Contains(b.Said, "never ends") {
t.Fatalf("the report does not say it replaced a busy container, with its last output: %+v", report.Busy)
}
if o := outcomeOf(report, "build-agent.server"); !strings.Contains(o.Detail, "still busy") ||
!strings.Contains(o.Detail, "never ends") {
t.Errorf("the outcome does not say a busy container was replaced, with what it said: %s", o.Detail)
}
}
// While one container waits, a send that changes another module on the same machine is applied at once: the
// apply returns after one ask, and holds nothing for the wait.
func TestASendToAnotherModuleIsAppliedWhileOneWaits(t *testing.T) {
now := time.Date(2026, 10, 13, 1, 0, 0, 0, time.UTC)
atClock(t, &now)
m := newBusyMachine()
_, state := applyBusy(t, m, twoContainers(t, "1", "1"), store.State{})
m.answer(true, "building")
_, state = applyBusy(t, m, twoContainers(t, "2", "1"), state)
// The next send moves only the web server.
mark := m.mark()
done := make(chan struct{})
var report Report
go func() {
defer close(done)
report, _ = applyBusy(t, m, twoContainers(t, "2", "2"), state)
}()
select {
case <-done:
case <-time.After(5 * time.Second):
t.Fatal("an apply while a container waits did not return")
}
got := m.since(mark)
if !slices.Contains(got, "create web") || slices.Contains(got, "rm mesh-build-agent") {
t.Fatalf("the other module's send was not applied at once beside the wait: %v", got)
}
if o := outcomeOf(report, "web.server"); o.Action != "updated" {
t.Errorf("the web server was reported %q: %+v", o.Action, o)
}
}
// A container's busy is held to its shape where it arrives.
func TestBusyIsReadAndHeldToItsShape(t *testing.T) {
d := twoContainers(t, "1", "1")
c := d.Resources[0].(*declaration.Container)
if c.Busy == nil || c.Busy.Command != "mesh-builder busy" || c.Busy.Within() != 5*time.Second {
t.Fatalf("busy was not read: %+v", c.Busy)
}
for name, raw := range map[string]string{
"no command": `"busy":{"command":" "}`,
"a long timeout": `"busy":{"command":"x","timeout":"5m"}`,
"not a duration": `"busy":{"command":"x","timeout":"soon"}`,
"an unknown key": `"busy":{"command":"x","every":"1m"}`,
"on a step": `"run-once":true,"busy":{"command":"x"}`,
} {
_, err := declaration.Parse([]byte(`{"declaration":1,"resources":[{"id":"a.b","type":"container",` +
`"name":"b","image":"` + pinned + `",` + raw + `}]}`))
if err == nil {
t.Errorf("%s: a busy of that shape was accepted", name)
}
}
}
// A replacement owed to a restart-on trigger is still owed after the wait: a trigger that is no file or
// directory (a service, an identity's or a library's change) is in no container's spec and counts as changed only
// in the pass it changed, so the container must not read unchanged at the next pass and drop from the report as if
// replaced (novox/hq ADR 0335).
func TestAReplacementOwedToARestartTriggerIsMadeOnceIdle(t *testing.T) {
now := time.Date(2026, 10, 13, 1, 0, 0, 0, time.UTC)
atClock(t, &now)
d := parse(t, `{"declaration":1,"resources":[
{"id":"build-agent.server","type":"container","name":"mesh-build-agent","image":"`+pinned+`",
"restart-on":["identity:build-agent"],"busy":{"command":"mesh-builder busy"}}
]}`)
r := d.Resources[0].(*declaration.Container)
in := inputs{declares: map[string]string{}}
m := newBusyMachine()
m.containers["mesh-build-agent"] = containerSpec(r, in)
ctx := context.Background()
// The trigger moved this pass, and the agent holds a build: left running.
m.answer(true, "building")
out, err := applyContainer(ctx, r, m.run, map[string]bool{"identity:build-agent": true}, in, store.Applied{})
if err != nil || out.Action != waitsBusy || out.busySince == nil {
t.Fatalf("a busy container was not left running for its trigger: %+v %v", out, err)
}
record := store.Applied{ID: r.ID, Target: r.Name, BusySince: out.busySince}
// The next pass: nothing triggers now, and the replacement is still owed.
now = now.Add(time.Minute)
m.answer(false, "idle")
mark := m.mark()
out, err = applyContainer(ctx, r, m.run, map[string]bool{}, in, record)
if err != nil {
t.Fatal(err)
}
if got := m.since(mark); !slices.Equal(got, []string{"ask mesh-build-agent", "rm mesh-build-agent", "create mesh-build-agent"}) {
t.Fatalf("the replacement owed to the trigger was dropped: %v (%+v)", got, out)
}
if out.busy == nil || out.busy.State != BusyReplacedIdle || out.busySince != nil {
t.Fatalf("the replacement does not say it was made once idle: %+v", out)
}
}
// A wait kept from a clock that later went back is never longer than the bound from now.
func TestAWaitFromAClockThatWentBackIsClampedToNow(t *testing.T) {
now := time.Date(2026, 10, 13, 1, 0, 0, 0, time.UTC)
atClock(t, &now)
m := newBusyMachine()
_, state := applyBusy(t, m, twoContainers(t, "1", "1"), store.State{})
m.answer(true, "building")
_, state = applyBusy(t, m, twoContainers(t, "2", "1"), state)
now = now.Add(-time.Hour) // the clock went back an hour
report, _ := applyBusy(t, m, twoContainers(t, "2", "1"), state)
b, _ := busyOf(report, "build-agent.server")
if b.State != BusyWaiting || b.Since.After(now) || !b.Until.Equal(now.Add(15*time.Minute)) {
t.Fatalf("a wait from a later clock is not clamped to now: %+v (now %s)", b, now)
}
}
// An ask that fails — no shell in the image, a timeout, a paused container — is not busy: replaced at once.
func TestAnAskThatFailsIsIdle(t *testing.T) {
m := newBusyMachine()
_, state := applyBusy(t, m, twoContainers(t, "1", "1"), store.State{})
failing := func(ctx context.Context, name string, args ...string) (string, error) {
if args[0] == "exec" {
m.mu.Lock()
m.calls = append(m.calls, "ask "+args[1])
m.mu.Unlock()
return "", errors.New("OCI runtime exec failed: exec: \"/bin/sh\": no such file or directory")
}
return m.run(ctx, name, args...)
}
mark := m.mark()
report, _, err := Apply(context.Background(), archHost(t), twoContainers(t, "2", "1"), state,
store.OriginDeclared, failing, nil, nil)
if err != nil {
t.Fatal(err)
}
if got := m.since(mark); !slices.Equal(got, []string{"ask mesh-build-agent", "rm mesh-build-agent", "create mesh-build-agent"}) {
t.Fatalf("an ask that failed held the container: %v", got)
}
if len(report.Busy) != 0 {
t.Errorf("an ask that failed is said as a wait: %+v", report.Busy)
}
}
// A stopped container holds no work: it is replaced without being asked.
func TestAStoppedContainerIsNotAsked(t *testing.T) {
m := newBusyMachine()
_, state := applyBusy(t, m, twoContainers(t, "1", "1"), store.State{})
m.answer(true, "would say busy")
stopped := func(ctx context.Context, name string, args ...string) (string, error) {
if args[0] == "container" && args[len(args)-1] == "mesh-build-agent" {
out, err := m.run(ctx, name, args...)
return strings.Replace(out, "true\t", "false\t", 1), err
}
return m.run(ctx, name, args...)
}
mark := m.mark()
if _, _, err := Apply(context.Background(), archHost(t), twoContainers(t, "2", "1"), state,
store.OriginDeclared, stopped, nil, nil); err != nil && !strings.Contains(err.Error(), "not running") {
t.Fatal(err)
}
got := m.since(mark)
if slices.Contains(got, "ask mesh-build-agent") || !slices.Contains(got, "rm mesh-build-agent") {
t.Fatalf("a stopped container was asked, or not replaced: %v", got)
}
}