[DB-2159] Fault V2 projections on partition processor errors

[George-Payne]

a V2 projection whose event processing throws never faults. The partition processor's task dies unobserved - partition tasks are only awaited after the read loop, which never completes for a continuous projection, and IsFaulted only watches the read-loop task - so the projection reports Running forever: no checkpoint is written, no state or emitted events are persisted (the checkpoint write needs every partition to ack), nothing is logged, and DisposeAsync hangs on the dead partition's full channel. The identical query under V1 faults immediately.

• Commit 1 adds two deliberately failing tests: a plain processor throw, and a throw with a checkpoint marker in flight (which also pins the coordinator's checkpoint lock). CI is red at that commit by design; green at HEAD.
• ProjectionEngineV2 races the read loop against the first partition exit: on a processor fault it logs, cancels the read loop, completes the channels, and faults the engine with the processor's exception - the existing CoreProjectionV2 poll then publishes Faulted with the reason. The read loop's final checkpoint marker is governed by a drain token: live on graceful shutdown (the engine token is already cancelled there but the final checkpoint must still be written), cancelled on the fault path so it cannot wait on a channel the dead partition stopped reading or a checkpoint lock it will never release. Sibling processor failures during drain are logged without replacing the primary fault.
• PartitionProcessor wraps state-handler invocations in a new PartitionProcessingException so the fault reason surfaced as stateReason is debuggable: projection name, handler type, event position, and the handler's message - the same format V1 reports, so operators and tooling see one fault-reason shape regardless of engine version. Infrastructure failures (persisted-state reads, cache writes) propagate raw rather than blaming the user's handler.

The fault-message format string is duplicated from EventProcessingProjectionProcessingPhase rather than extracted to a shared location, to keep this change out of V1 code.

[linear-code]

DB-2159

[qodo-code-review]

PR Summary by Qodo

Fault V2 projections when partition processors throw (DB-2159)

🐞 Bug fix 🧪 Tests 🕐 40+ Minutes

AI Description

• Fault ProjectionEngineV2 when any partition processor task exits faulted during continuous reads.
• Cancel read-loop draining and complete dispatch channels to avoid checkpoint/DisposeAsync
 deadlocks.
• Wrap handler exceptions with a V1-compatible fault reason and add regression tests.

Diagram

graph TD
  Engine["ProjectionEngineV2"] --> Read["Read loop"] --> Ch[("Partition channels")] --> Proc["PartitionProcessor"]
  Proc --> Handler["State handler"]
  Proc --> Coord["CheckpointCoordinator"] --> Client["SystemClient"]
  Proc -. "exception" .-> Fault["Projection Faulted"]
  Read -. "cancel + stop drain" .-> Fault

Loading

High-Level Assessment

The following are alternative approaches to this PR:

1. Attach fault continuations to partition tasks

• ➕ Simpler control flow than racing read loop vs partitions
• ➕ Can fault engine immediately without additional CTS plumbing
• ➖ Still needs explicit handling to unblock read-loop drain/checkpoint injection
• ➖ Continuation ordering/exception propagation can be harder to reason about than a single race

2. Propagate partition faults through dispatcher/coordinator as a dedicated message

• ➕ Keeps fault signaling within the existing channel-based orchestration
• ➕ Could centralize shutdown behavior (complete channels, cancel drain) in one place
• ➖ More invasive API changes across dispatcher/coordinator
• ➖ Harder to preserve the original exception as the primary fault without extra plumbing

Recommendation: The chosen approach (race read-loop completion against first partition exit, then explicitly cancel read and drain + complete channels) is the most direct fix for continuous projections and addresses the real deadlock surfaces (final checkpoint marker + bounded channels). Alternatives still require the same drain/unblock mechanics and would likely be more invasive or less explicit.

Files changed (4) +234 / -28

Enhancement (1) +23 / -0

PartitionProcessingException.csAdd PartitionProcessingException for debuggable, V1-shaped fault reasons +23/-0

Add PartitionProcessingException for debuggable, V1-shaped fault reasons

• Adds a dedicated exception type that formats the fault reason to match the V1 event-processing fault message (projection name, handler type, event position, and handler message). Intended to be surfaced as the projection stateReason for operator/tooling consistency.

src/KurrentDB.Projections.V2/Services/Processing/V2/PartitionProcessingException.cs

Bug fix (2) +99 / -26

PartitionProcessor.csWrap handler invocations to rethrow as PartitionProcessingException +50/-21

Wrap handler invocations to rethrow as PartitionProcessingException

• Introduces InvokeHandler helpers that wrap state-handler calls (Load/Initialize/Process* methods) and rethrow handler exceptions with projection/handler/position context. Infrastructure failures (e.g., state-stream reads, cache writes) remain unwrapped to avoid misattribution.

src/KurrentDB.Projections.V2/Services/Processing/V2/PartitionProcessor.cs

ProjectionEngineV2.csObserve partition processor failures and prevent drain-time deadlocks +49/-5

Observe partition processor failures and prevent drain-time deadlocks

• Changes engine Run() to race the read loop against the first partition task exit; on a partition fault it logs the failure, cancels the read loop, cancels a new drain token to stop final checkpoint marker injection, completes dispatch channels, and faults the engine with the original exception. RunReadLoop now uses the drain token (not CancellationToken.None) for its final checkpoint marker, ensuring graceful shutdown still checkpoints while the fault path cannot block forever.

src/KurrentDB.Projections.V2/Services/Processing/V2/ProjectionEngineV2.cs

Tests (1) +112 / -2

ProjectionEngineV2LifecycleTests.csAdd regression tests for V2 engine faulting on partition processor exceptions +112/-2

Add regression tests for V2 engine faulting on partition processor exceptions

• Introduces throwing state handlers and two tests that reproduce DB-2159 for continuous projections: a direct handler throw and a throw while checkpoint markers can be in flight. Tests assert the engine faults and surfaces a meaningful fault message without hanging on DisposeAsync.

src/KurrentDB.Projections.V2.Tests/Unit/ProjectionEngineV2LifecycleTests.cs

[qodo-code-review]

Code Review by Qodo

🐞 Bugs (1) 📘 Rule violations (0) 📎 Requirement gaps (0) 🎨 UX issues (0) 🔗 Cross-repo conflicts (0) 📜 Skill insights (0)

1. Complete before readloop stops 🐞 Bug ☼ Reliability

Description

In the partition-fault path, ProjectionEngineV2 calls dispatcher.Complete(partitionFault) before
awaiting the read loop to finish, even though the read loop is the channel writer. This makes
read-loop writes fail via channel-closure (the code comment explicitly calls out
ChannelClosedException) rather than cleanly via the already-cancelled write token, adding
timing-dependent failure behavior during fault shutdown.

Code

src/KurrentDB.Projections.V2/Services/Processing/V2/ProjectionEngineV2.cs[R155-176]

+			using var readCts = CancellationTokenSource.CreateLinkedTokenSource(ct);
+			using var drainCts = new CancellationTokenSource();
+			var readLoop = RunReadLoop(checkpoint, dispatcher, coordinator, readCts.Token, drainCts.Token);
+			var firstPartitionExit = Task.WhenAny(partitionTasks);
+
+			var winner = await Task.WhenAny(readLoop, firstPartitionExit);
+			if (winner == firstPartitionExit) {
+				var exited = await firstPartitionExit;
+				if (exited.IsFaulted) {
+					partitionFault = exited.Exception!.InnerException ?? exited.Exception;
+					Log.Error(partitionFault, "ProjectionEngineV2 {Name} partition processor failed", _config.ProjectionName);
+					// Unblock the read loop before awaiting it: cancel its final
+					// checkpoint marker via the drain token (a marker the dead
+					// partition never acked holds the coordinator's checkpoint lock
+					// forever), and complete the channels so pending and subsequent
+					// writes throw ChannelClosedException instead of blocking on a
+					// full channel.
+					await readCts.CancelAsync();
+					await drainCts.CancelAsync();
+					dispatcher.Complete(partitionFault);
+					await readLoop.ConfigureAwait(ConfigureAwaitOptions.SuppressThrowing);
+					ExceptionDispatchInfo.Capture(partitionFault).Throw();

Evidence

The fault path completes channels before awaiting the read loop, and the code comment indicates this
forces writes to throw ChannelClosedException. The dispatcher’s write APIs used by the read loop all
perform Writer.WriteAsync(..., ct), so cancelling the read-loop CTS already unblocks those writes
without needing early completion.

src/KurrentDB.Projections.V2/Services/Processing/V2/ProjectionEngineV2.cs[148-177]
src/KurrentDB.Projections.V2/Services/Processing/V2/PartitionDispatcher.cs[47-76]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

## Issue description
On partition fault, `ProjectionEngineV2.Run` completes the dispatcher channels (`dispatcher.Complete(partitionFault)`) before awaiting the read loop task. Since the read loop performs `WriteAsync` calls into these channels, closing them early can turn in-flight writes into `ChannelClosedException` behavior (as noted in the code comment) rather than the intended cancellation-driven unwinding.

## Issue Context
All dispatcher write operations already accept a `CancellationToken` and the read loop passes its cancelable token to those writes, so cancelling `readCts` is sufficient to unblock a blocked `WriteAsync` without requiring early channel completion.

## Fix Focus Areas
- src/KurrentDB.Projections.V2/Services/Processing/V2/ProjectionEngineV2.cs[148-180]
- src/KurrentDB.Projections.V2/Services/Processing/V2/PartitionDispatcher.cs[47-76]

## Suggested fix
Reorder the fault path to:
1) cancel `readCts` and `drainCts`,
2) await `readLoop` completion (still using `SuppressThrowing`),
3) then call `dispatcher.Complete(...)` to stop partitions.

This keeps shutdown deterministic (no concurrent close-vs-write) while still ensuring partitions stop and the engine faults with the primary partition exception.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---