Count a livelock from its first stalled cycle

.agent/docs/runtime.md

@@ -143,12 +143,17 @@ a no-op.
    `ErrLivelockDetected`. After the natural stop (a livelocked cycle activated something) and last
    (a real error is always the better explanation).
    - A cycle is *stalled* when every activated component returned `ErrWaitKeepingInputs` **and**
-     the mesh's pending signal count did not change. Both halves are needed: the first alone flags a
-     component legitimately accumulating; the second alone flags a busy but idempotent mesh.
+     the mesh's pending signal count (after activation) equals the count the previous drain left.
+     Both halves are needed: the first alone flags a component a hook or its own activation
+     function feeds or consumes; the second alone flags a busy but idempotent mesh. The baseline is
+     taken after the drain, so `n` is exact: a stall that starts right after a productive cycle
+     stops the run `n` cycles later, not `n+1`.
    - `n` stalled cycles in a row end the run. The error names each waiting component with its empty
-     and non-empty input ports, plus a count of components that never activated.
-   - This is exact, not a guess: waiting components are not drained, so a stalled cycle moves no
-     signals and the next cycle is identical. Waiters that *drop* inputs never count as stalled:
+     and non-empty input ports, plus a count of components with no input signals.
+   - Waiting components are not drained, so a stalled cycle moves no signals, and the next cycle
+     is identical **as long as activation depends only on inputs**. A waiter that proceeds because
+     of its `State()`, the clock or the outside world is a false positive; so is a hook that swaps
+     signals one for one. Waiters that *drop* inputs never count as stalled:
      next cycle they have no input (a natural stop). The pending count alone cannot tell, because
      it is taken before the drain clears the dropped inputs.
 

config.go

@@ -34,7 +34,9 @@ type config struct {
 	// LivelockThreshold is how many consecutive stalled cycles end the run with
 	// ErrLivelockDetected. A cycle is stalled when every component that activated
 	// was waiting and keeping its inputs and the mesh's pending signals did not
-	// change, which means the next cycle would be identical to this one.
+	// change, which means the next cycle would be identical to this one, as long
+	// as activation depends only on inputs: a waiter that proceeds because of its
+	// State, the clock or the outside world needs a higher threshold or none.
 	//
 	// The default of 2 rather than 1 keeps the detector conservative.
 	// 0 disables detection (use WithoutLivelockDetection to say so explicitly).

docs/wiki/401.-Scheduling-rules.md

@@ -119,19 +119,25 @@ livelock detected: no progress for 2 consecutive cycles, stopped at cycle #2. Ev
 that activated is waiting for inputs and no signals moved, so every following cycle would be
 identical to this one
   "joiner" is waiting (keeping inputs): empty input ports [b], holding signals on [a]
-  1 other component(s) never activated: no signals ever reached them
+  1 other component(s) have no input signals
 ```
 
 Here `empty input ports [b]` is the answer: nothing was piped to `b`.
 
 A cycle is **stalled** when both are true:
 
 - every component that activated was waiting for inputs, and
-- the number of signals on input ports did not change.
+- the number of signals on input ports is what the previous cycle left.
 
 A component that collects input over many cycles is never flagged, because its signal count keeps
 changing.
 
+Detection assumes a waiting component decides from its inputs alone. A component that waits keeping
+its inputs and then proceeds because of something else (a counter in its `State()`, the clock, an
+external service) is stopped as livelocked once it has waited for the threshold. Raise the threshold
+or turn detection off for such a mesh. A hook that swaps a waiting component's signals for the same
+number of new ones is likewise not seen as progress.
+
 ```go
 fmesh.New("mesh", fmesh.WithLivelockThreshold(5))   // stop after 5 stalled cycles (default 2)
 fmesh.New("mesh", fmesh.WithoutLivelockDetection()) // never stop for this reason

fmesh.go

@@ -300,23 +300,22 @@ func (fm *FMesh) countPendingSignals() int {
 // updates the stall bookkeeping. Called once per cycle.
 //
 // A cycle is stalled when every component that activated was waiting and keeping
-// its inputs *and* the pending signal count is unchanged — both halves are
-// needed: the first alone flags legitimate input accumulation, the second alone
-// flags a busy-but-idempotent mesh. A waiter that drops its inputs never stalls:
-// the count is taken before the drain clears them, so it would look unchanged.
+// its inputs *and* the pending signal count is what the previous drain left —
+// both halves are needed: the first alone flags inputs a hook or an activation
+// function changed, the second alone flags a busy-but-idempotent mesh. A waiter
+// that drops its inputs never stalls: the count is taken before the drain clears
+// them, so it would look unchanged.
 func (fm *FMesh) detectLivelock(lastCycle *cycle.Cycle) bool {
 	if fm.config.LivelockThreshold <= 0 {
 		return false
 	}
 
-	pending := fm.countPendingSignals()
 	allKeeping := lastCycle.HasActivatedComponents() && lastCycle.ActivationResults().Every(component.WantsToKeepInputs)
-	if allKeeping && pending == fm.pendingSignals {
+	if allKeeping && fm.countPendingSignals() == fm.pendingSignals {
 		fm.stalledCycles++
 	} else {
 		fm.stalledCycles = 0
 	}
-	fm.pendingSignals = pending
 
 	return fm.stalledCycles >= fm.config.LivelockThreshold
 }
@@ -348,14 +347,13 @@ func (fm *FMesh) livelockError(lastCycle *cycle.Cycle) error {
 	var detail strings.Builder
 	starved, named, waiting := 0, 0, 0
 	for _, c := range fm.Components().AllOrdered() {
-		activationResult := lastCycle.ActivationResults().ByName(c.Name())
-		if activationResult == nil || !component.IsWaitingForInput(activationResult) {
-			// A component with no input is not recorded at all. Counting them is
-			// worth the line: in a mutual wait only the component that happens to
-			// hold the signal shows up above, and the one starving it is invisible.
-			if activationResult == nil {
-				starved++
-			}
+		// In a stalled cycle every recorded result is a wait keeping inputs, so a
+		// component is either waiting or had no input and was not recorded.
+		if lastCycle.ActivationResults().ByName(c.Name()) == nil {
+			// Counting them is worth the line: in a mutual wait only the component
+			// that happens to hold the signal shows up above, and the one starving
+			// it is invisible.
+			starved++
 			continue
 		}
 
@@ -374,20 +372,16 @@ func (fm *FMesh) livelockError(lastCycle *cycle.Cycle) error {
 		}
 		named++
 
-		mode := "dropping inputs"
-		if component.WantsToKeepInputs(activationResult) {
-			mode = "keeping inputs"
-		}
-		fmt.Fprintf(&detail, "\n  %q is waiting (%s): empty input ports %v, holding signals on %v",
-			c.Name(), mode, empty, holding)
+		fmt.Fprintf(&detail, "\n  %q is waiting (keeping inputs): empty input ports %v, holding signals on %v",
+			c.Name(), empty, holding)
 	}
 
 	if waiting > named {
 		fmt.Fprintf(&detail, "\n  ...and %d more waiting component(s)", waiting-named)
 	}
 
 	if starved > 0 {
-		fmt.Fprintf(&detail, "\n  %d other component(s) never activated: no signals ever reached them", starved)
+		fmt.Fprintf(&detail, "\n  %d other component(s) have no input signals", starved)
 	}
 
 	return fmt.Errorf("%w: no progress for %d consecutive cycles, stopped at cycle #%d. "+
@@ -459,6 +453,12 @@ func (fm *FMesh) Run(ctx context.Context) (ri *RuntimeInfo, runErr error) {
 		if err := fm.drainComponents(ctx); err != nil {
 			return ri, err
 		}
+
+		// The baseline the next livelock check compares against: taken after the
+		// drain, so the cycle signals were just delivered into counts in full.
+		if fm.config.LivelockThreshold > 0 {
+			fm.pendingSignals = fm.countPendingSignals()
+		}
 	}
 }
 

integration_tests/livelock/detection_test.go

@@ -8,6 +8,7 @@ package livelock
 
 import (
 	"context"
+	"fmt"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -53,7 +54,7 @@ func TestLivelock_MutualWait(t *testing.T) {
 	assert.Contains(t, err.Error(), `"alpha" is waiting (keeping inputs)`)
 	assert.Contains(t, err.Error(), "empty input ports []")
 	assert.Contains(t, err.Error(), "holding signals on [in]")
-	assert.Contains(t, err.Error(), "1 other component(s) never activated",
+	assert.Contains(t, err.Error(), "1 other component(s) have no input signals",
 		"the component starving the other one is invisible otherwise")
 	t.Logf("reported as:\n%v", err)
 }
@@ -228,3 +229,163 @@ func TestLivelock_NamesAFewWaitersAndCountsTheRest(t *testing.T) {
 	assert.NotContains(t, err.Error(), `"w6" is waiting`)
 	assert.Contains(t, err.Error(), "...and 2 more waiting component(s)")
 }
+
+// emitter forwards one signal per activation, however many it received.
+func emitter(name string) *component.Component {
+	return testutil.MustComponent(name,
+		component.WithInputs("in"),
+		component.WithOutputs("out"),
+		component.WithActivationFunc(func(_ context.Context, this *component.Component) error {
+			return this.OutputByName("out").PutSignals(signal.New(1))
+		}))
+}
+
+func TestLivelock_ThresholdIsExact(t *testing.T) {
+	for _, threshold := range []int{1, 2, 3} {
+		t.Run(fmt.Sprintf("stalled from the first cycle, threshold %d", threshold), func(t *testing.T) {
+			stuck := waiter("stuck", component.ErrWaitKeepingInputs)
+			fm, err := fmesh.New("seeded", fmesh.WithLivelockThreshold(threshold))
+			require.NoError(t, err)
+			require.NoError(t, fm.AddComponents(stuck))
+			require.NoError(t, stuck.InputByName("in").PutSignals(signal.New(1)))
+
+			ri, err := fm.Run(context.Background())
+
+			require.ErrorIs(t, err, fmesh.ErrLivelockDetected)
+			assert.Equal(t, threshold, ri.Cycles.Len())
+		})
+
+		t.Run(fmt.Sprintf("stalled after a productive cycle, threshold %d", threshold), func(t *testing.T) {
+			// The source turns three signals into one, so the pending count
+			// differs from the one the previous cycle left before its drain:
+			// the stall must still be counted from its first cycle.
+			src, stuck := emitter("src"), waiter("stuck", component.ErrWaitKeepingInputs)
+			fm, err := fmesh.New("after-progress", fmesh.WithLivelockThreshold(threshold))
+			require.NoError(t, err)
+			require.NoError(t, fm.AddComponents(src, stuck))
+			require.NoError(t, src.OutputByName("out").PipeTo(stuck.InputByName("in")))
+			require.NoError(t, src.InputByName("in").PutSignals(signal.New(1), signal.New(2), signal.New(3)))
+
+			ri, err := fm.Run(context.Background())
+
+			require.ErrorIs(t, err, fmesh.ErrLivelockDetected)
+			assert.Equal(t, 1+threshold, ri.Cycles.Len())
+			assert.Contains(t, err.Error(), "1 other component(s) have no input signals",
+				"src activated in cycle 1, so it must not be described as never activated")
+		})
+	}
+}
+
+func TestLivelock_ConsumingWaiterIsProgress(t *testing.T) {
+	// A waiter that keeps its inputs but takes one signal per cycle is working
+	// through a backlog, not stuck.
+	consumer := testutil.MustComponent("consumer",
+		component.WithInputs("in"),
+		component.WithOutputs("out"),
+		component.WithActivationFunc(func(ctx context.Context, this *component.Component) error {
+			in := this.InputByName("in")
+			signals := in.Signals()
+			if signals.Len() == 1 {
+				return this.OutputByName("out").PutSignals(signal.New("drained"))
+			}
+			first := signals.First()
+			if err := in.Clear(ctx); err != nil {
+				return err
+			}
+			if err := in.PutSignalGroups(signals.Filter(func(s *signal.Signal) bool { return s != first })); err != nil {
+				return err
+			}
+			return component.ErrWaitKeepingInputs
+		}))
+
+	fm, err := fmesh.New("backlog", fmesh.WithLivelockThreshold(1))
+	require.NoError(t, err)
+	require.NoError(t, fm.AddComponents(consumer))
+	require.NoError(t, consumer.InputByName("in").PutSignals(signal.New(1), signal.New(2), signal.New(3), signal.New(4)))
+
+	_, err = fm.Run(context.Background())
+
+	require.NoError(t, err)
+	assert.True(t, consumer.OutputByName("out").HasSignals())
+}
+
+func TestLivelock_StatefulWaiterIsReported(t *testing.T) {
+	// The documented limit: detection assumes a waiter decides from its inputs
+	// alone. One that proceeds on its own state looks stuck while it waits.
+	newMesh := func(t *testing.T, opts ...fmesh.Option) (*fmesh.FMesh, *component.Component) {
+		t.Helper()
+		patient := testutil.MustComponent("patient",
+			component.WithInputs("in"),
+			component.WithOutputs("out"),
+			component.WithActivationFunc(func(_ context.Context, this *component.Component) error {
+				waited, _ := this.State().GetOrDefault("waited", 0).(int)
+				if waited < 3 {
+					this.State().Set("waited", waited+1)
+					return component.ErrWaitKeepingInputs
+				}
+				return this.OutputByName("out").PutSignals(signal.New("done"))
+			}))
+		fm, err := fmesh.New("patient", opts...)
+		require.NoError(t, err)
+		require.NoError(t, fm.AddComponents(patient))
+		require.NoError(t, patient.InputByName("in").PutSignals(signal.New(1)))
+		return fm, patient
+	}
+
+	fm, _ := newMesh(t)
+	_, err := fm.Run(context.Background())
+	require.ErrorIs(t, err, fmesh.ErrLivelockDetected)
+
+	fm, patient := newMesh(t, fmesh.WithoutLivelockDetection())
+	_, err = fm.Run(context.Background())
+	require.NoError(t, err)
+	assert.True(t, patient.OutputByName("out").HasSignals())
+}
+
+func TestLivelock_AnyOtherActivationIsNotAStall(t *testing.T) {
+	// A stuck waiter next to a component that keeps working: the mesh as a
+	// whole is busy, so only the cycle limit stops it.
+	stuck, busy := waiter("stuck", component.ErrWaitKeepingInputs), emitter("busy")
+	require.NoError(t, busy.LoopbackPipe("out", "in"))
+
+	fm, err := fmesh.New("busy", fmesh.WithLivelockThreshold(1), fmesh.WithCyclesLimit(20))
+	require.NoError(t, err)
+	require.NoError(t, fm.AddComponents(stuck, busy))
+	require.NoError(t, stuck.InputByName("in").PutSignals(signal.New(1)))
+	require.NoError(t, busy.InputByName("in").PutSignals(signal.New(1)))
+
+	_, err = fm.Run(context.Background())
+
+	require.ErrorIs(t, err, fmesh.ErrReachedMaxAllowedCycles)
+	require.NotErrorIs(t, err, fmesh.ErrLivelockDetected)
+}
+
+func TestLivelock_CyclesLimitWinsATie(t *testing.T) {
+	stuck := waiter("stuck", component.ErrWaitKeepingInputs)
+	fm, err := fmesh.New("tie", fmesh.WithCyclesLimit(2))
+	require.NoError(t, err)
+	require.NoError(t, fm.AddComponents(stuck))
+	require.NoError(t, stuck.InputByName("in").PutSignals(signal.New(1)))
+
+	_, err = fm.Run(context.Background())
+
+	require.ErrorIs(t, err, fmesh.ErrReachedMaxAllowedCycles)
+	require.NotErrorIs(t, err, fmesh.ErrLivelockDetected)
+}
+
+func TestLivelock_EachRunCountsFromZero(t *testing.T) {
+	// Inputs survive a failed run, so a rerun stalls again, but it must take
+	// the full threshold again rather than inherit the previous run's count.
+	const threshold = 3
+	stuck := waiter("stuck", component.ErrWaitKeepingInputs)
+	fm, err := fmesh.New("rerun", fmesh.WithLivelockThreshold(threshold))
+	require.NoError(t, err)
+	require.NoError(t, fm.AddComponents(stuck))
+	require.NoError(t, stuck.InputByName("in").PutSignals(signal.New(1)))
+
+	for run := 1; run <= 2; run++ {
+		ri, err := fm.Run(context.Background())
+		require.ErrorIs(t, err, fmesh.ErrLivelockDetected)
+		assert.Equal(t, threshold, ri.Cycles.Len(), "run %d", run)
+	}
+}