fix(schedule): bound per-node re-dispatch and surface a loud CapErr (#828)

pkg/orchestrator/dagrun.go

@@ -125,5 +125,10 @@ func (o *Orchestrator) runDAG(ctx context.Context) error {
 	sched.Seed(o.seedNodes())
 	sched.Run(ctx)
 
-	return errors.Join(o.finalize(), ctx.Err())
+	// sched.CapErr() is non-nil only if the scheduler's per-node re-dispatch cap
+	// tripped (Change A/#828): a non-converging emit was terminalized loudly
+	// instead of hanging. Fold it into the returned error so a #828 recurrence
+	// surfaces as an errored run naming the offending node — the scheduler writes
+	// no store and no controller changes.
+	return errors.Join(o.finalize(), ctx.Err(), sched.CapErr())
 }

pkg/schedule/schedule.go

@@ -23,7 +23,9 @@ package schedule
 
 import (
 	"context"
+	"fmt"
 	"slices"
+	"strings"
 	"sync"
 
 	"github.com/home-operations/flate/pkg/manifest"
@@ -87,6 +89,42 @@ const (
 	stateTerminal
 )
 
+// Change A (#828) — per-node re-dispatch cap. A schedulable node whose emitted
+// content is non-deterministic across concurrent re-runs feeds an unbounded
+// re-dispatch storm: Run never reaches its inFlight == 0 fixpoint and the pool
+// spawns dispatches without bound. The scheduler is a faithful AMPLIFIER of that
+// emit defect, not its source; the cap bounds the amplifier so a non-converging
+// emit fails LOUDLY (an errored Run naming the node, see CapErr) instead of
+// hanging silently.
+//
+// The ceiling is cumulative per node over one Run — never reset, because the
+// storm re-dispatches a terminal node, so a per-terminal reset would zero the
+// count every cycle and the cap would never fire. It scales with node count:
+//
+//	capLimit = capFloor + capPerNode*len(nodes)
+//
+// The legitimate per-node re-dispatch ceiling is itself bounded: 1 initial
+// dispatch + the <=6 re-runs a depth-5 dangling chain tolerates
+// (TestDanglingChainRunCountBounded) + <=2 drain-escalation re-queues + one
+// selector-ResourceSet re-expansion per distinct arrival (O(nodes)). capFloor
+// covers the constant terms with headroom; capPerNode covers the per-arrival
+// re-expansion with ~4x headroom. For a 90-node tree capLimit is ~392 —
+// comfortably above the ~100 legitimate ceiling and far below an unbounded
+// storm. The constants deliberately favor NEVER false-positive over fast
+// detection: a late-but-certain loud failure beats a false terminalize of a
+// converging DAG.
+const (
+	capFloor   = 32
+	capPerNode = 4
+)
+
+// capDiag records a node the re-dispatch cap terminalized and the cumulative
+// dispatch count it reached, for CapErr to report after Run returns.
+type capDiag struct {
+	id    NodeID
+	count int
+}
+
 type node struct {
 	id        NodeID
 	state     nodeState
@@ -100,6 +138,15 @@ type node struct {
 	// park on, so it must re-expand once the store has quiesced. Set from the
 	// scheduler's rerunAtDrain predicate after the node's first dispatch.
 	rerun bool
+	// dispatchCount is the cumulative number of times this node has been
+	// dispatched at the Run chokepoint over one Run, never reset (Change A/#828).
+	// When it would exceed capLimit the node is force-terminalized instead of
+	// dispatched, and capped is latched.
+	dispatchCount int
+	// capped is latched when the re-dispatch cap terminalizes this node; once
+	// set, no path (OnArrival re-admission, the fixpoint rerun sweep) may
+	// re-admit it, so the storm provably stops.
+	capped bool
 }
 
 // Scheduler is a re-entrant fixpoint reconcile driver. Construct with New,
@@ -127,6 +174,9 @@ type Scheduler struct {
 	// by the orchestrator (SetRerunAtDrain); evaluated off the hot path in the
 	// dispatch goroutine, never under mu.
 	rerunAtDrain func(NodeID) bool
+	// capExceeded records nodes the re-dispatch cap terminalized (Change A/#828),
+	// populated under mu at the dispatch chokepoint. CapErr reports them.
+	capExceeded []capDiag
 }
 
 // SetRerunAtDrain installs the predicate that decides whether a node re-runs at
@@ -190,6 +240,23 @@ func (s *Scheduler) Run(ctx context.Context) {
 			if n == nil || n.state != stateRunnable {
 				continue
 			}
+			// Change A (#828): count every dispatch at this single chokepoint —
+			// the one site all enqueue paths (OnArrival, complete()'s
+			// rerunRequested re-queue, unparkLocked, requeueRerunLocked) funnel
+			// through. Past the cumulative cap, do NOT dispatch: force the node
+			// terminal, record the diagnostic, and wake anything parked on it
+			// (mirroring the normal terminal path in complete()). This converts a
+			// non-converging re-dispatch storm from a silent hang into a loud,
+			// attributable failure surfaced via CapErr.
+			n.dispatchCount++
+			if n.dispatchCount > s.capLimitLocked() {
+				n.capped = true
+				n.state = stateTerminal
+				n.blockedOn = nil
+				s.capExceeded = append(s.capExceeded, capDiag{id: id, count: n.dispatchCount})
+				s.wakeWaitersLocked(id)
+				continue
+			}
 			n.state = stateRunning
 			n.rerunRequested = false
 			n.blockedOn = nil
@@ -247,6 +314,32 @@ func (s *Scheduler) Run(ctx context.Context) {
 	s.tasks.BlockTillDone()
 }
 
+// capLimitLocked returns the per-node cumulative re-dispatch ceiling for the
+// current node set: capFloor + capPerNode*len(nodes). It scales with node count
+// because the legitimate re-dispatch ceiling does (per-arrival selector
+// re-expansion is O(nodes)). Caller holds mu.
+func (s *Scheduler) capLimitLocked() int {
+	return capFloor + capPerNode*len(s.nodes)
+}
+
+// CapErr returns a non-nil error naming every node the re-dispatch cap
+// terminalized (Change A/#828) and the cumulative dispatch count each reached,
+// or nil if the cap never tripped. Its contract is to be read after Run returns;
+// it takes mu so a race-detector-clean mid-Run read is also safe.
+func (s *Scheduler) CapErr() error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if len(s.capExceeded) == 0 {
+		return nil
+	}
+	parts := make([]string, 0, len(s.capExceeded))
+	for _, d := range s.capExceeded {
+		parts = append(parts, fmt.Sprintf("%s (re-dispatched %d times)", d.id.String(), d.count))
+	}
+	return fmt.Errorf("schedule: re-dispatch cap exceeded for %d node(s): %s",
+		len(s.capExceeded), strings.Join(parts, ", "))
+}
+
 // complete records the result of one Dispatch. Runs on the worker goroutine;
 // acquires mu. Touches ONLY scheduler state — never the store or the pool.
 func (s *Scheduler) complete(id NodeID, out Outcome, blocked []NodeID, rerun bool) {
@@ -382,9 +475,14 @@ func (s *Scheduler) OnArrival(id NodeID, schedulable bool) {
 		switch n.state {
 		case stateTerminal:
 			// Content changed (Refire reset, or a parent re-emitted a mutated
-			// spec): re-run so the new content is reconciled.
-			n.state = stateRunnable
-			s.runq = append(s.runq, id)
+			// spec): re-run so the new content is reconciled — UNLESS the
+			// re-dispatch cap already terminalized this node (Change A/#828), in
+			// which case a later content-changed arrival must not re-admit it and
+			// reopen the storm.
+			if !n.capped {
+				n.state = stateRunnable
+				s.runq = append(s.runq, id)
+			}
 		case stateRunning:
 			n.rerunRequested = true
 		}
@@ -455,7 +553,10 @@ func (s *Scheduler) requeueAllParkedLocked() {
 func (s *Scheduler) requeueRerunLocked() bool {
 	var due []*node
 	for _, n := range s.nodes {
-		if n.state == stateTerminal && n.rerun {
+		// !capped: the re-dispatch cap (Change A/#828) terminalized this node, so
+		// the fixpoint rerun sweep must not re-admit it either — the chokepoint
+		// would only re-cap it, and no path may reopen the storm.
+		if n.state == stateTerminal && n.rerun && !n.capped {
 			due = append(due, n)
 		}
 	}

pkg/schedule/schedule_test.go

@@ -2,7 +2,9 @@ package schedule
 
 import (
 	"context"
+	"fmt"
 	"runtime"
+	"strings"
 	"sync"
 	"testing"
 
@@ -373,3 +375,137 @@ func TestDanglingChainRunCountBounded(t *testing.T) {
 		}
 	}
 }
+
+func TestSchedulerCapFiresOnPathologicalNode(t *testing.T) {
+	// Change A (#828), GC-828-R1 observability: a node re-admitted without bound
+	// must trip the per-node re-dispatch cap — Run RETURNS (does not hang) and
+	// CapErr() names the capped node and its count. Drive the storm through the
+	// REAL re-dispatch seam: pump OnArrival(x, schedulable=true) on a node that
+	// terminalizes each cycle, so every arrival re-admits it via the
+	// stateTerminal -> stateRunnable transition (the #828 path). A SEPARATE gated
+	// keepalive holds inFlight > 0 so Run cannot reach its fixpoint and settle
+	// between pumps (which would pass the test vacuously); the storming node x is
+	// NOT gated (build-lane note OI1).
+	f := newFake(map[NodeID][]NodeID{
+		id("x"):         nil, // terminalizes Ready on every dispatch
+		id("keepalive"): nil,
+	})
+	gate := make(chan struct{})
+	f.gate[id("keepalive")] = gate
+
+	ts := task.NewBounded(8)
+	s := New(ts, f)
+	s.Seed([]NodeID{id("x"), id("keepalive")})
+	done := make(chan struct{})
+	go func() { s.Run(context.Background()); close(done) }()
+
+	// x has been dispatched once (its initial seed dispatch); keepalive's task
+	// was launched in the same frontier pass, so inFlight is held > 0.
+	waitUntil(t, func() bool { return f.runCount(id("x")) >= 1 })
+
+	// Pump the storm past capLimit. CapErr() is mu-guarded, so this concurrent
+	// read is race-safe (the -race run on pkg/schedule is the backstop); break as
+	// soon as the cap trips. OnArrival on an already-capped node is refused, so
+	// runCount plateaus — CapErr is the termination signal here.
+	tripped := false
+	for range 5_000_000 {
+		if s.CapErr() != nil {
+			tripped = true
+			break
+		}
+		s.OnArrival(id("x"), true)
+		runtime.Gosched()
+	}
+	close(gate) // release keepalive so Run can settle
+	<-done
+
+	if !tripped {
+		t.Fatal("re-dispatch cap never tripped while pumping the storm")
+	}
+	err := s.CapErr()
+	if err == nil {
+		t.Fatal("CapErr() is nil after Run returned; want the capped node named")
+	}
+	if !strings.Contains(err.Error(), id("x").String()) {
+		t.Fatalf("CapErr() = %q; want it to name the capped node %q", err, id("x").String())
+	}
+	// The cap trips at capLimit+1 dispatches, so x was dispatched EXACTLY capLimit
+	// times (cumulative, never reset) for the 2-node set.
+	wantDispatches := capFloor + capPerNode*2
+	if rc := f.runCount(id("x")); rc != wantDispatches {
+		t.Fatalf("x dispatched %d times; want capLimit=%d (cap trips on the next attempt)", rc, wantDispatches)
+	}
+}
+
+func TestSchedulerCapNoFalsePositiveOnConvergingDAG(t *testing.T) {
+	// Change A (#828), GC-828-R3 false-positive guard: a legitimately-converging
+	// large DAG must NOT trip the cap. capLimit = capFloor + capPerNode*len(nodes)
+	// scales far above the legitimate per-node re-dispatch ceiling (<=6
+	// dangling-chain re-runs + bounded drain escalation + one selector
+	// re-expansion per arrival), so CapErr() stays nil. Exercise both legitimate
+	// re-dispatch sources at once: several depth-5 dangling chains that drain with
+	// bounded re-runs, and a selector rerun node re-expanding on distinct arrivals.
+	g := map[NodeID][]NodeID{}
+	const chains, depth = 10, 5
+	var chainNodes []string
+	for c := range chains {
+		names := make([]string, depth)
+		for d := range depth {
+			names[d] = fmt.Sprintf("c%d-n%d", c, d)
+		}
+		for i, n := range names {
+			if i+1 < len(names) {
+				g[id(n)] = []NodeID{id(names[i+1])}
+			} else {
+				g[id(n)] = []NodeID{id("absent")} // leaf blocks on an absent dep -> drains
+			}
+			chainNodes = append(chainNodes, n)
+		}
+	}
+	g[id("rs")] = nil
+	g[id("keepalive")] = nil
+	f := newFake(g)
+	f.drainRerun[id("rs")] = true
+
+	gate := make(chan struct{})
+	f.gate[id("keepalive")] = gate
+
+	ts := task.NewBounded(8)
+	s := New(ts, f)
+	s.SetRerunAtDrain(func(id NodeID) bool { return f.drainRerun[id] })
+	seeds := make([]NodeID, 0, len(g))
+	for k := range g {
+		seeds = append(seeds, k)
+	}
+	s.Seed(seeds)
+	done := make(chan struct{})
+	go func() { s.Run(context.Background()); close(done) }()
+
+	// rs terminalizes on its first dispatch; wait for it, then dirty the store
+	// with several distinct data arrivals (buffered while keepalive holds the
+	// pool non-idle). Releasing the gate lets the chains drain and rs re-expand
+	// against the now-quiesced store.
+	waitUntil(t, func() bool { return f.runCount(id("rs")) >= 1 })
+	for i := range 5 {
+		s.OnArrival(NodeID{Kind: manifest.KindConfigMap, Namespace: "ns", Name: fmt.Sprintf("late-%d", i)}, false)
+	}
+	close(gate)
+	<-done
+
+	if err := s.CapErr(); err != nil {
+		t.Fatalf("converging DAG tripped the re-dispatch cap: %v", err)
+	}
+	// Every chain node drains to terminal-Failed with a bounded re-run count, far
+	// below capLimit = capFloor + capPerNode*len(nodes).
+	capLimit := capFloor + capPerNode*len(g)
+	for _, n := range chainNodes {
+		assertFailed(t, f, n)
+		if rc := f.runCount(id(n)); rc >= capLimit {
+			t.Fatalf("%s ran %d times; a converging node must stay far below capLimit=%d", n, rc, capLimit)
+		}
+	}
+	assertReady(t, f, "rs")
+	if rc := f.runCount(id("rs")); rc < 2 {
+		t.Fatalf("rerun node ran %d times; want >=2 (initial + at least one re-expansion)", rc)
+	}
+}