Leave a busy container running until it says idle, within 15 minutes

A send that replaced the build agent killed the check it ran, and the queue
started it again elsewhere (issue 594). A container may now declare busy; the
node-engine asks it before replacing it, applies everything else, says the
wait, and replaces it on the first idle pass or at the bound, kept across
restarts (novox/hq ADR 0335).

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")
+	}
+}

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)

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{}