diff --git a/cmd/mesh-host/busy_test.go b/cmd/mesh-host/busy_test.go new file mode 100644 index 0000000..b1edaea --- /dev/null +++ b/cmd/mesh-host/busy_test.go @@ -0,0 +1,66 @@ +package main + +import ( + "context" + "sync/atomic" + "testing" + "time" + + "github.com/novox/mesh-host/internal/link" +) + +// A container waiting because it said busy is news when it starts waiting and when it is replaced, never on +// its output alone (novox/hq ADR 0335): the walk waiting on this machine starts its gate at the report that +// says it replaced, and a reconcile is otherwise silent. +func TestABusyWaitIsSaidWhenItStartsAndWhenItEnds(t *testing.T) { + quiet := link.Report{Outward: []string{"eth0"}} + waiting := link.Report{Outward: []string{"eth0"}, Busy: []link.Busy{{ID: "build-agent.server", + Container: "mesh-build-agent", State: "waiting", Said: "building a"}}} + still := link.Report{Outward: []string{"eth0"}, Busy: []link.Busy{{ID: "build-agent.server", + Container: "mesh-build-agent", State: "waiting", Said: "building b, a minute on"}}} + replaced := link.Report{Outward: []string{"eth0"}, Busy: []link.Busy{{ID: "build-agent.server", + Container: "mesh-build-agent", State: "replaced-when-idle"}}} + if adoptionFingerprint(waiting) == adoptionFingerprint(quiet) { + t.Error("a container starting to wait is not news") + } + if adoptionFingerprint(still) != adoptionFingerprint(waiting) { + t.Error("a wait whose output changed reads as news on every ask") + } + if adoptionFingerprint(replaced) == adoptionFingerprint(waiting) { + t.Error("a container replaced after it waited is not news") + } + if !worthSaying(link.Report{Busy: replaced.Busy}) { + t.Error("a replacement after a wait is not worth saying") + } +} + +// While a container waits, a pass is asked for every so often, not at the reconcile's five minutes; with +// nothing waiting, none (novox/hq ADR 0335 §2). +func TestAWaitIsAskedAgainSoonerThanTheReconcile(t *testing.T) { + if BusyAskEvery >= ReconcileEvery || BusyAskEvery > time.Minute { + t.Fatalf("a waiting container is asked every %s", BusyAskEvery) + } + var reconciles atomic.Int32 + queue := &link.Queue{Reconcile: func(context.Context) (link.Report, bool) { + reconciles.Add(1) + return link.Report{}, false + }} + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go queue.Run(ctx) + busyWaiting.Store(false) + t.Cleanup(func() { busyWaiting.Store(false) }) + go busyPasses(ctx, queue, 10*time.Millisecond) + time.Sleep(100 * time.Millisecond) + if n := reconciles.Load(); n != 0 { + t.Fatalf("%d passes were asked with nothing waiting", n) + } + busyWaiting.Store(true) + deadline := time.Now().Add(2 * time.Second) + for reconciles.Load() == 0 && time.Now().Before(deadline) { + time.Sleep(10 * time.Millisecond) + } + if reconciles.Load() == 0 { + t.Fatal("no pass was asked while a container waits") + } +} diff --git a/cmd/mesh-host/main.go b/cmd/mesh-host/main.go index 83b5473..8c79ab3 100644 --- a/cmd/mesh-host/main.go +++ b/cmd/mesh-host/main.go @@ -24,6 +24,7 @@ import ( "sort" "strings" "sync" + "sync/atomic" "syscall" "text/tabwriter" "time" @@ -1259,6 +1260,7 @@ func runLink(ctx context.Context, opts options) error { queue.ReconcileDue() } go holdTheMachine(ctx, queue) + go busyPasses(ctx, queue, BusyAskEvery) // And whether what it runs stays up, looked at on its own clock and said when it changes (novox/hq // ADR 0240) — between applies, which is when a container crash-loops. if judging != nil { @@ -1397,6 +1399,12 @@ func adoptionFingerprint(r link.Report) string { // it could only speak when a declaration arrived, and a declaration cannot be composed until it // has spoken: a machine waiting for a push that is waiting for the machine. parts := []string{"firewall=" + r.Firewall, "outward=" + strings.Join(r.Outward, ",")} + // And a container waiting because it said busy, or replaced after it waited (novox/hq ADR 0335): the walk + // waiting on this machine starts its gate at the report that says it replaced. Not its output, which + // may change on every ask. + for _, b := range r.Busy { + parts = append(parts, "busy "+b.ID+"="+b.State) + } for _, h := range r.Held { parts = append(parts, "held "+h.ID+"="+h.Changed) } @@ -1736,7 +1744,7 @@ func healthAsReported(st liveness.Statement, ns *network.Statement) *link.Health // LibraryContract: this engine reads a resource's declared `health` and judges it (ADR 0240 Phase B), judges a // user's declared `root` (novox/hq ADR 0266), and reads a declaration's identities, groups and libraries // (novox/hq ADR 0328), which is what tells the controller it may be sent any of those fields. - h := &link.Health{Contract: link.OwnGroupContract, At: st.At.UTC(), Resources: []link.ResourceHealth{}} + h := &link.Health{Contract: link.BusyContract, At: st.At.UTC(), Resources: []link.ResourceHealth{}} for _, r := range st.Resources { h.Resources = append(h.Resources, link.ResourceHealth{Module: r.Module, Resource: r.ID, Kind: r.Kind, Target: r.Target, State: r.State, Reason: r.Reason, Since: r.Since.UTC(), Streak: r.Streak, @@ -1868,6 +1876,32 @@ func holdTheMachine(ctx context.Context, queue *link.Queue) { } } +// busyWaiting is whether the last apply left a container running because it said busy (novox/hq ADR 0335). +var busyWaiting atomic.Bool + +// BusyAskEvery is how often a container waiting because it said busy is asked again (novox/hq ADR 0335 §2): +// each ask is a pass of its own, so a build that ends is followed by its replacement within this, not +// within the reconcile's five minutes — and the bound of 15 minutes is met within this of its end. +const BusyAskEvery = 20 * time.Second + +// busyPasses asks for a pass every so often while a container waits because it said busy. **It asks; it +// does not apply**, as holdTheMachine: the queue's worker does, when its turn comes, so a pass never holds +// the machine for the wait, and a delivery waiting is that pass's apply. +func busyPasses(ctx context.Context, queue *link.Queue, every time.Duration) { + ticker := time.NewTicker(every) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + if busyWaiting.Load() { + queue.ReconcileDue() + } + } + } +} + // reconcileKept is the reconcile's act, as the queue's worker runs it: hold the machine to what it // was last told, read once it is this act's turn, and say what did not go. // @@ -1919,7 +1953,7 @@ func worthSaying(report link.Report) bool { } return len(report.Held) > 0 || report.Firewall != "" || len(report.Outward) > 0 || len(report.Filters) > 0 || report.FoundFirewall != nil || len(report.Windows) > 0 || - len(report.Rollbacks) > 0 + len(report.Rollbacks) > 0 || len(report.Busy) > 0 } // announceOr is what the apply writes its detail with, given what the caller has to say things with. @@ -2168,6 +2202,14 @@ func applyAndKeepHeld(ctx context.Context, opts options, raw []byte, signed *sto for _, w := range outcome.Windows { report.Windows = append(report.Windows, link.Window{Step: w.Step, Holds: w.Holds, Since: w.Opened, Until: w.Until}) } + // And a container left running because it said busy, or replaced after it waited (novox/hq ADR 0335): + // the walk waiting on this machine reads it. While one waits, this machine asks it again sooner than its + // reconcile (busyPasses). + for _, b := range outcome.Busy { + report.Busy = append(report.Busy, link.Busy{ID: b.ID, Container: b.Container, State: b.State, + Since: b.Since, Until: b.Until, Said: b.Said}) + } + busyWaiting.Store(outcome.Waiting()) // And what runs here that nobody asked for (novox/hq ADR 0163). if strays, err := apply.Strays(ctx, apply.ExecRunner, updated); err != nil { fmt.Fprintf(os.Stderr, "mesh-host: applied, and could not list what else runs here: %v\n", err) diff --git a/cmd/mesh-host/units_test.go b/cmd/mesh-host/units_test.go index 0084497..61e98a0 100644 --- a/cmd/mesh-host/units_test.go +++ b/cmd/mesh-host/units_test.go @@ -87,10 +87,10 @@ func TestTheStatementSaysAnAccountJudgedForRoot(t *testing.T) { Groups: []string{"openrazer"}}, State: accounts.Healthy, Since: at}, }} h := withAccounts(healthAsReported(liveness.Statement{At: at}, nil), as) - // At RootContract or later: an engine that also reads an identity's runner says OwnGroupContract (novox/hq ADR - // 0334), one above ADR 0328's LibraryContract. - if h.Contract < link.RootContract || link.RootContract != 3 || h.Contract != link.OwnGroupContract || - link.OwnGroupContract != link.LibraryContract+1 { + // At RootContract or later: an engine that also reads a container's busy says BusyContract (novox/hq ADR + // 0335), one above ADR 0334's OwnGroupContract. + if h.Contract < link.RootContract || link.RootContract != 3 || h.Contract != link.BusyContract || + link.BusyContract != link.OwnGroupContract+1 { t.Fatalf("contract %d", h.Contract) } got := map[string]string{} diff --git a/internal/apply/apply.go b/internal/apply/apply.go index dbad871..3695a09 100644 --- a/internal/apply/apply.go +++ b/internal/apply/apply.go @@ -92,6 +92,11 @@ type Outcome struct { // reads is, for a container, the digest of each file it was created reading, by path — so // the next apply can say which one changed (novox/hq 04-ISSUES/103). reads map[string]string + // busySince is, for a container left running because it said busy, when the first apply that would + // have replaced it ran (novox/hq ADR 0335): kept in its record, so the bound holds across passes and + // across a restart of the node-engine. busy is what the report says of it. + busySince *time.Time + busy *BusyWait } // Report is what an apply did, in the order it did it. @@ -108,6 +113,20 @@ type Report struct { // 224): a machine whose server is stopped at 03:31 because its collector is running is // working, not broken, and the report is where that difference is said. Windows []Window `json:"windows,omitempty"` + // Busy is every container this apply left running because it said busy, or replaced after it had + // waited (novox/hq ADR 0335): a machine whose walk waits for it says why, and how it ended. + Busy []BusyWait `json:"busy,omitempty"` +} + +// Waiting says whether a container of this apply waits to be replaced because it is busy: what makes the +// node-engine ask again sooner than its reconcile (novox/hq ADR 0335 §2). +func (r Report) Waiting() bool { + for _, b := range r.Busy { + if b.State == BusyWaiting { + return true + } + } + return false } // Changed reports whether anything about the machine actually moved. An apply that changed @@ -118,7 +137,10 @@ func (r Report) Changed() bool { // whose account's manager is not running, which nothing moved either (novox/hq ADR 0177); // held still is a container a maintenance window stopped, which the apply did not touch // (novox/hq issue 224). - if o.Action != "unchanged" && o.Action != "held" && o.Action != "waiting" && o.Action != heldStill { + // waits busy is a container left running because it said busy (novox/hq ADR 0335), which nothing + // moved either. + if o.Action != "unchanged" && o.Action != "held" && o.Action != "waiting" && o.Action != heldStill && + o.Action != waitsBusy { return true } } @@ -987,6 +1009,7 @@ func ApplyMindingWindows( KeptMode: keptMode, KeptOwner: keptOwner, Reads: outcome.reads, + BusySince: outcome.busySince, Stateless: outcome.stateless, Scope: outcome.scope, User: outcome.user, @@ -1046,7 +1069,10 @@ func ApplyMindingWindows( Action: "forgotten", Detail: detail}) log(fmt.Sprintf(" forgotten %s (%s): %s", from.ID, from.Target, detail)) } - if outcome.Action == heldStill { + if outcome.busy != nil { + report.Busy = append(report.Busy, *outcome.busy) + } + if outcome.Action == heldStill || outcome.Action == waitsBusy { // Not changed: nothing about it moved, so nothing that restarts on it restarts. log(fmt.Sprintf(" %s %s (%s): %s", outcome.Action, outcome.ID, outcome.Target, outcome.Detail)) } else if outcome.Action != "unchanged" { @@ -2707,6 +2733,15 @@ func applyContainer(ctx context.Context, r *declaration.Container, run Runner, out.Action = "unchanged" return out, nil case existed: + // **A container that says it is busy is asked before it is replaced** (novox/hq ADR 0335 §2): left + // running while it says busy, within the bound, and replaced on the first pass it says idle. Only a + // running one: a stopped container holds no work in hand. + if r.Busy != nil && before.Running { + if waits := askBeforeReplacing(ctx, r, run, cri, previous, &out); waits { + out.reads = wasReading + return out, nil + } + } if _, err := run(ctx, cri, "rm", "-f", r.Name); err != nil { return out, fmt.Errorf("replacing container %s: %w", r.Name, err) } @@ -2807,6 +2842,9 @@ func applyContainer(ctx context.Context, r *declaration.Container, run Runner, default: out.Detail = "replaced; a container's configuration is fixed when it is created" } + if out.busy != nil { + out.Detail += "; " + out.busy.Words() + } } return out, nil } diff --git a/internal/apply/busy.go b/internal/apply/busy.go new file mode 100644 index 0000000..0c80bfb --- /dev/null +++ b/internal/apply/busy.go @@ -0,0 +1,139 @@ +package apply + +import ( + "context" + "fmt" + "strings" + "time" + "unicode/utf8" + + "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). +// +// **Asked, never waited on.** The apply asks the container once, and while it says busy leaves it running +// as it is and goes on with everything else of the declaration: the machine's lock is held for one ask, +// never for the wait. The wait is carried between passes in the container's record (store.Applied's +// BusySince), so the next pass — the node-engine asks again sooner than its reconcile while anything waits +// — asks again, and so does a node-engine started again in between: the bound runs from the first apply +// that would have replaced the container, whoever applied it. + +// waitsBusy is the action of a container left running because it said busy (novox/hq ADR 0335 §2). Its own +// word rather than "unchanged": its declaration moved, and the machine is not yet what was declared. +const waitsBusy = "waits-busy" + +// What a report says of a busy container (novox/hq ADR 0335 §2, §3). +const ( + // BusyWaiting is a container left running because it said busy, waiting to be replaced. + BusyWaiting = "waiting" + // BusyReplacedIdle is a container replaced on the first pass it said idle, after it had waited. + BusyReplacedIdle = "replaced-when-idle" + // BusyReplacedAtBound is a container still busy at the bound, replaced anyway. + BusyReplacedAtBound = "replaced-at-bound" +) + +// BusyWait is one container a busy command held, as a report says it. +type BusyWait struct { + // ID is the resource's id (`.`), Container its runtime name. + ID string `json:"id"` + Container string `json:"container"` + // State is BusyWaiting, BusyReplacedIdle or BusyReplacedAtBound. + State string `json:"state"` + // Since is when the first apply that would have replaced it ran; Until when the bound replaces it. + Since time.Time `json:"since"` + Until time.Time `json:"until"` + // Said is the last output its busy command gave. + Said string `json:"said,omitempty"` +} + +// Words is the wait in a sentence, for an outcome's detail and the console. +func (b BusyWait) Words() string { + since := b.Since.UTC().Format(time.RFC3339) + said := "" + if b.Said != "" { + said = "; its busy command last said: " + b.Said + } + switch b.State { + case BusyReplacedAtBound: + return fmt.Sprintf("replaced while still busy: it said busy since %s and the bound of %s ran out "+ + "(novox/hq ADR 0335 §3)%s", since, declaration.BusyBound, said) + case BusyReplacedIdle: + return fmt.Sprintf("replaced once it said idle, after waiting since %s (novox/hq ADR 0335 §2)%s", since, said) + } + return fmt.Sprintf("busy: left running and replaced on the first pass it says idle, or at %s at the latest; "+ + "waiting since %s (novox/hq ADR 0335 §2)%s", b.Until.UTC().Format(time.RFC3339), since, said) +} + +// busyClock is the clock the bound is read on. A test moves it. +var busyClock = time.Now + +// busySaidMost is how much of a busy command's output a report carries: its last lines, enough to say what +// it held. +const busySaidMost = 400 + +// askBeforeReplacing asks a running container that declares busy whether it may be replaced now, and says +// in out what came of it. True when it waits: out is then the whole outcome of this pass. False when the +// caller replaces it — idle, or past the bound — with out carrying the wait it ended, if there was one. +func askBeforeReplacing(ctx context.Context, r *declaration.Container, run Runner, cri string, + previous store.Applied, out *Outcome) bool { + now := busyClock().UTC() + busy, said := askBusy(ctx, r, run, cri) + var since time.Time + waited := previous.BusySince != nil && previous.Target == r.Name + if waited { + since = previous.BusySince.UTC() + } + wait := BusyWait{ID: r.ID, Container: r.Name, Since: since, Said: said} + switch { + case !busy && !waited: + // Idle at the first ask: replaced at once, as a container that declares nothing is. + return false + case !busy: + wait.State, wait.Until = BusyReplacedIdle, since.Add(declaration.BusyBound) + out.busy = &wait + return false + } + if !waited { + since = now + wait.Since = now + } + wait.Until = since.Add(declaration.BusyBound) + if !now.Before(wait.Until) { + wait.State = BusyReplacedAtBound + out.busy = &wait + return false + } + wait.State = BusyWaiting + out.Action, out.Detail = waitsBusy, wait.Words() + out.busySince, out.busy = &since, &wait + return true +} + +// askBusy runs a container's busy command inside it, by its shell, as an exec health check is run: busy is +// an exit of 0 and nothing else. An ask that could not be made, or did not answer within its timeout, is +// not busy: the bound is what keeps a container whose command never answers from being held. +func askBusy(ctx context.Context, r *declaration.Container, run Runner, cri string) (bool, string) { + ask, cancel := context.WithTimeout(ctx, r.Busy.Within()) + defer cancel() + output, err := run(ask, cri, "exec", r.Name, "/bin/sh", "-c", r.Busy.Command) + said := strings.TrimSpace(output) + if err != nil && said == "" { + said = err.Error() + } + return err == nil, lastOf(said, busySaidMost) +} + +// lastOf is the end of a text, at most n bytes, on one line. +func lastOf(s string, n int) string { + s = strings.Join(strings.Fields(s), " ") + if len(s) <= n { + return s + } + from := len(s) - n + for from < len(s) && !utf8.RuneStart(s[from]) { + from++ + } + return "…" + s[from:] +} diff --git a/internal/apply/busy_test.go b/internal/apply/busy_test.go new file mode 100644 index 0000000..5f5c25d --- /dev/null +++ b/internal/apply/busy_test.go @@ -0,0 +1,326 @@ +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) + } + } +} diff --git a/internal/declaration/busy.go b/internal/declaration/busy.go new file mode 100644 index 0000000..3b73485 --- /dev/null +++ b/internal/declaration/busy.go @@ -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 +} diff --git a/internal/declaration/declaration.go b/internal/declaration/declaration.go index 4e54fc7..a265afa 100644 --- a/internal/declaration/declaration.go +++ b/internal/declaration/declaration.go @@ -1239,6 +1239,11 @@ type Container struct { // 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 @@ -1318,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") } diff --git a/internal/link/messages.go b/internal/link/messages.go index 08b237c..0dc192e 100644 --- a/internal/link/messages.go +++ b/internal/link/messages.go @@ -174,6 +174,14 @@ type Report struct { // broken. Windows []Window `json:"windows,omitempty"` + // Busy is every container the apply that reports left running because it said busy, and every one it + // replaced after it had waited, with how (novox/hq ADR 0335): `waiting`, `replaced-when-idle` or + // `replaced-at-bound`, since when, until when at the latest, and the last output of its busy command. + // A walk waiting on this machine for a module whose container is listed `waiting` waits for the report + // that no longer lists it so: its first-machine gate starts then (ADR 0335 §4). Absent from an engine + // older than BusyContract, which never waits. + Busy []Busy `json:"busy,omitempty"` + // Strays is what runs on the machine that the mesh neither wrote nor holds (novox/hq ADR // 0163): containers nobody declared and nobody holds, the ones a cutover leaves behind. Strays []Stray `json:"strays,omitempty"` @@ -280,6 +288,13 @@ const LibraryContract = 6 // before it, its presence is what tells the controller this host may be sent `runner` on an identity. const OwnGroupContract = 7 +// BusyContract is the statement of a host that also reads a container's `busy` (novox/hq ADR 0335): it asks a +// container that declares it before replacing it, leaves it running while it says busy, within a bound of +// 15 minutes, and says the wait in its report's `busy`. Like the fields before it, its presence is what tells +// the controller this host may be sent `busy` on a container; an older host refuses the whole declaration +// for it. +const BusyContract = 8 + // Health is one statement of every long-running resource's state on this machine (to-be 48 §4). Said in // every report, as an event on each change, and again every minute while anything is not healthy — so a // lost statement is not a lost fault. @@ -731,6 +746,17 @@ type Window struct { Until time.Time `json:"until"` } +// Busy is one container a busy command held (novox/hq ADR 0335), as a report says it. ID is the resource's +// id, `.`. +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"` +} + // A Stray is a container the mesh neither wrote nor holds (ADR 0163). type Stray struct { Kind string `json:"kind"` diff --git a/internal/store/store.go b/internal/store/store.go index 5bdd2a4..3726606 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -155,6 +155,11 @@ type Applied struct { // container's spec label; this is the same information kept per file, so that when the spec // no longer matches the host can say WHICH file changed rather than only that something did. Reads map[string]string `json:"reads,omitempty"` + + // BusySince is, for a container left running because it said busy, when the first apply that would + // have replaced it ran (novox/hq ADR 0335 §3): kept here, beside the machine's state, so the bound runs + // from then across every pass and across a restart of the node-engine. Absent while nothing waits. + BusySince *time.Time `json:"busy_since,omitempty"` } // Into is what a file written into held before the mesh's keys, or before the mesh's block.