PR #5: Bug - Fix cascading checkpoint writes when a stopped flow re-enters TryStart on subsequent events

[AlexanderJohnston]

This image depicts modern Totem artwork in the style of a death spiral loop.

Problem

This bug has existed since ~2020 and is not part of the gRPC or net10 changes.

A single transient failure (e.g., a gRPC DeadlineExceeded timeout writing to EventStore) permanently kills a flow and then triggers a cascading death spiral that writes redundant stopped checkpoints for every subsequent event routed to that flow.

We observed this in production with ProjectClassifierTopic. A timeout at position #257138 caused the flow to correctly stop and persist its error state — but then every new event routed to the topic (positions #257138 through #325995) re-entered the same stop/write cycle, producing ~100 nested exception wrappers and ~100 unnecessary checkpoint appends.

The resulting log entry was a System.AggregateException containing a chain of "Flow is stopped at #N with this error:" messages nested roughly 100 levels deep, each wrapping the one before it, all the way back to the original timeout.

Cause

The issue is in FlowScope<T>.TryStart(). When a flow has no in-memory state (Flow == null) and receives a new event, ObservePointIfStarted calls TryStart() to read the checkpoint from the database. If the checkpoint indicates the flow is stopped, TryStart throws:

case FlowInfo.Stopped stopped:
    throw new Exception($"Flow is stopped at {stopped.Position} with this error: {stopped.Error}");

That exception propagates up to ObserveNextPoint's catch block, which calls Stop(). Stop() is designed for first-time failures — it persists the error state to a checkpoint and completes the lifetime task with an exception. But here, the error state is already persisted from the original failure. So Stop() redundantly:

1. Creates a new flow instance (since Flow is still null)
2. Sets the error on it (wrapping the already-wrapped error string)
3. Writes another checkpoint to EventStore (identical in meaning to the one already there)
4. Completes the lifetime with a new wrapping exception

The lifetime completing would normally end the while(Running) loop — and it does. But the damage per iteration is already done: one unnecessary checkpoint write and one layer of exception nesting. Since ObserveQueue processes the backlog of enqueued points before checking Running, every pending event triggers one full cycle before the loop exits.

The deeper issue is that TryStart treats "already stopped" as an observation-time error (by throwing), when it is actually a pre-existing terminal state that was already fully handled by the original Stop() call. The throw conflates discovering a stopped state with entering a stopped state.

Fix

Replace the throw in the FlowInfo.Stopped case with a direct lifetime completion and early return:

case FlowInfo.Stopped stopped:
    CompleteTask(new Exception(
        $"Flow {Key} stopped at position {stopped.Position}",
        new Exception(stopped.Error)));
    return;

CompleteTask sets the lifetime exception, which makes Running return false. The return exits TryStart back into ObservePointIfStarted, which checks Running before calling ObservePoint() and skips it. Control returns to ObserveNextPoint's try block (no exception, so the catch is not entered), then back to the while(Running) loop, which exits.

Reasoning For Fix

Before the fix:
An event dequeues
→ TryStart reads FlowInfo.Stopped
→ throws
→ Stop() writes a redundant checkpoint
→ CompleteTask sets Running = false
→ loop exits.
Any remaining events in _queue are abandoned with the FlowScope itself.

After the fix:
An event dequeues
→ TryStart reads FlowInfo.Stopped
→ CompleteTask sets Running = false
→ return skips back through ObservePointIfStarted
→ loop exits.
Same abandonment, just on the first event instead of the Nth.

In both cases, the queued events are never processed. _queue is an in-memory structure that dies with the FlowScope. The events themselves are still in the timeline stream and the resume projection still has them tracked. They aren't lost from EventStore, they're just unprocessed by this flow instance.

The difference is only in how many events get individually dequeued and fed through the Stop() path before the loop breaks. Before: all of them, each writing a checkpoint. After: one dequeue, zero checkpoint writes, immediate exit.

The difference is only in how many events get individually dequeued and fed through the Stop() path before the loop breaks. Before: all of them, each writing a checkpoint. After: one dequeue, zero checkpoint writes, immediate exit.

Recovery of those events is the same either way — it requires manual intervention to clear the stopped state (resetting the checkpoint stream or the projection's isStopped flag), at which point the resume projection would see isStopped = false and the flow would replay from its last good checkpoint.

What this preserves

This change is intentionally minimal and does not alter any of the existing design invariants:

Invariant	Before	After
Flow stops permanently on error	✅ Original Stop() persists the checkpoint	✅ Unchanged — the original Stop() still runs on first failure
Lifetime completes with exception	✅ Via Stop() → CompleteTask	✅ Via CompleteTask directly
Error is surfaced to callers	✅ Nested exception chain	✅ Single wrapping exception with original error as inner
Checkpoint written to EventStore	✅ Once (then N-1 redundant times)	✅ Exactly once
ObservePoint not called for stopped flows	✅ Running becomes false	✅ Running becomes false
Resume projection isStopped flag	✅ Set to true	✅ Set to true (from the original checkpoint, no change)

What this changes

• No redundant checkpoint writes. Previously, discovering a stopped flow wrote a new checkpoint per event. Now it writes zero — the original stop checkpoint is authoritative.
• No exception nesting cascade. The error message is a single level deep: "Flow {Key} stopped at position {Position}" wrapping the original error string. Previously it was N levels deep, one per event processed after the failure.
• Error message format. Changed from "Flow is stopped at {Position} with this error: {Error}" to "Flow {Key} stopped at position {Position}" with the original error as InnerException. This includes the flow key for diagnostics and uses structured exception nesting rather than string concatenation of the original error.

Scope

One file, one case branch, two lines. No new dependencies, no behavioral changes to healthy flows, no changes to the resume projection or checkpoint storage.