diff --git a/pkg/orchestrator/dagrun.go b/pkg/orchestrator/dagrun.go index c76c1052..d73f3a9f 100644 --- a/pkg/orchestrator/dagrun.go +++ b/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()) } diff --git a/pkg/schedule/schedule.go b/pkg/schedule/schedule.go index 9de49496..1c2b5404 100644 --- a/pkg/schedule/schedule.go +++ b/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) } } diff --git a/pkg/schedule/schedule_test.go b/pkg/schedule/schedule_test.go index 2cdc9293..e957e5d1 100644 --- a/pkg/schedule/schedule_test.go +++ b/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) + } +}