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

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

@@ -91,7 +91,7 @@ public void Publish(Message message) {
 		}
 	}
 
-	sealed class NoOpStateHandler : IProjectionStateHandler {
+	class NoOpStateHandler : IProjectionStateHandler {
 		public void Load(string state) { }
 		public void LoadShared(string state) { }
 		public void Initialize() { }
@@ -100,7 +100,7 @@ public void InitializeShared() { }
 		public string GetStatePartition(CheckpointTag eventPosition, string category, ProjectionResolvedEvent data) =>
 			data.EventStreamId;
 
-		public bool ProcessEvent(string partition,
+		public virtual bool ProcessEvent(string partition,
 			CheckpointTag eventPosition,
 			string category,
 			ProjectionResolvedEvent @event,
@@ -136,6 +136,36 @@ public bool ProcessPartitionDeleted(string partition, CheckpointTag deletePositi
 		public void Dispose() { }
 	}
 
+	sealed class ThrowingStateHandler : NoOpStateHandler {
+		public override bool ProcessEvent(string partition,
+			CheckpointTag eventPosition,
+			string category,
+			ProjectionResolvedEvent @event,
+			out string newState,
+			out string newSharedState,
+			out EmittedEventEnvelope[] emittedEvents) {
+			throw new Exception("handler boom");
+		}
+	}
+
+	// Processes the first event normally, throws on the second. Used to fault a
+	// partition while a checkpoint marker can be in flight.
+	sealed class ThrowOnSecondEventStateHandler : NoOpStateHandler {
+		private int _seen;
+
+		public override bool ProcessEvent(string partition,
+			CheckpointTag eventPosition,
+			string category,
+			ProjectionResolvedEvent @event,
+			out string newState,
+			out string newSharedState,
+			out EmittedEventEnvelope[] emittedEvents) {
+			if (++_seen >= 2)
+				throw new Exception("handler boom");
+			return base.ProcessEvent(partition, eventPosition, category, @event, out newState, out newSharedState, out emittedEvents);
+		}
+	}
+
 	#endregion
 
 	[Test]
@@ -188,6 +218,86 @@ public async Task engine_handles_empty_read_stream() {
 		await Assert.That(engine.IsFaulted).IsFalse();
 	}
 
+	[Test]
+	public async Task engine_faults_when_partition_processor_throws() {
+		// DB-2159: a state-handler exception faults the partition processor's task, but
+		// nothing observed partition tasks while the read loop ran, so a continuous
+		// projection never faulted - it reported Running forever with no checkpoint, no
+		// persistence, and nothing logged. The infinite read strategy reproduces the
+		// continuous case; a finite stream would surface the fault via the end-of-read
+		// drain and mask the bug.
+		var publisher = new CapturingPublisher();
+		var user = new ClaimsPrincipal(new ClaimsIdentity());
+		var stateHandler = new ThrowingStateHandler();
+
+		var config = new ProjectionEngineV2Config {
+			ProjectionName = "faulting-test",
+			SourceDefinition = stateHandler.GetSourceDefinition(),
+			StateHandlerFactory = () => new ThrowingStateHandler(),
+			MaxPartitionStateCacheSize = 1000,
+			PartitionCount = 1,
+			CheckpointAfterMs = 0,
+			CheckpointHandledThreshold = 100,
+			CheckpointUnhandledBytesThreshold = long.MaxValue
+		};
+
+		var engine = new ProjectionEngineV2(config, new InfiniteReadStrategy(), new SystemClient(publisher), user);
+		engine.Start(new TFPos(0, 0));
+
+		var timeout = Task.Delay(TimeSpan.FromSeconds(5));
+		while (!engine.IsFaulted && !timeout.IsCompleted)
+			await Task.Delay(50);
+
+		// Assert before disposing: on a wedged engine DisposeAsync never returns (the
+		// read loop's final checkpoint marker blocks forever on the dead partition's
+		// full channel), so disposing first turns a failure into a hang.
+		await Assert.That(engine.IsFaulted).IsTrue();
+		// The fault reason names the projection, the handler, and the event position,
+		// and carries the handler's error message.
+		await Assert.That(engine.FaultException!.Message).Contains("failed to process an event");
+		await Assert.That(engine.FaultException!.Message).Contains("handler boom");
+
+		await engine.DisposeAsync();
+	}
+
+	[Test]
+	public async Task engine_faults_when_processor_throws_with_checkpoint_in_flight() {
+		// A checkpoint marker the dead partition never acks holds the coordinator's
+		// checkpoint lock forever, so the fault path must cancel the read loop's
+		// final marker injection rather than wait on the lock. Threshold 1 keeps
+		// markers in flight around the fault; the handler processes one event and
+		// throws on the second. The exact interleaving of marker injection and the
+		// fault race is timing-dependent, but every interleaving must fault - a
+		// wedge here reproduces DB-2159's narrowed variant.
+		var publisher = new CapturingPublisher();
+		var user = new ClaimsPrincipal(new ClaimsIdentity());
+		var stateHandler = new ThrowOnSecondEventStateHandler();
+
+		var config = new ProjectionEngineV2Config {
+			ProjectionName = "faulting-checkpoint-test",
+			SourceDefinition = stateHandler.GetSourceDefinition(),
+			StateHandlerFactory = () => new ThrowOnSecondEventStateHandler(),
+			MaxPartitionStateCacheSize = 1000,
+			PartitionCount = 1,
+			CheckpointAfterMs = 0,
+			CheckpointHandledThreshold = 1,
+			CheckpointUnhandledBytesThreshold = long.MaxValue
+		};
+
+		var engine = new ProjectionEngineV2(config, new InfiniteReadStrategy(), new SystemClient(publisher), user);
+		engine.Start(new TFPos(0, 0));
+
+		var timeout = Task.Delay(TimeSpan.FromSeconds(5));
+		while (!engine.IsFaulted && !timeout.IsCompleted)
+			await Task.Delay(50);
+
+		// Assert before disposing (see engine_faults_when_partition_processor_throws).
+		await Assert.That(engine.IsFaulted).IsTrue();
+		await Assert.That(engine.FaultException!.Message).Contains("handler boom");
+
+		await engine.DisposeAsync();
+	}
+
 	[Test]
 	public async Task engine_with_multiple_partitions_processes_events() {
 		var events = new CoreResolvedEvent[20];

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

@@ -0,0 +1,23 @@
+// Copyright (c) Kurrent, Inc and/or licensed to Kurrent, Inc under one or more agreements.
+// Kurrent, Inc licenses this file to you under the Kurrent License v1 (see LICENSE.md).
+
+#nullable enable
+
+using System;
+
+namespace KurrentDB.Projections.Core.Services.Processing.V2;
+
+/// <summary>
+/// An event-processing failure inside a partition processor. The message is the
+/// projection's fault reason as surfaced in stateReason: projection name, handler
+/// type, event position, and the handler's error. Same format as
+/// EventProcessingProjectionProcessingPhase.SafeProcessEventByHandler, so operators
+/// and tooling see one fault-reason shape regardless of engine version; keep in sync.
+/// </summary>
+public sealed class PartitionProcessingException : Exception {
+	public PartitionProcessingException(string projectionName, Type handlerType, string eventPosition, Exception inner)
+		: base(string.Format(
+			"The {0} projection failed to process an event.\r\nHandler: {1}\r\nEvent Position: {2}\r\n\r\nMessage:\r\n\r\n{3}",
+			projectionName, handlerType.Namespace + "." + handlerType.Name, eventPosition, inner.Message), inner) {
+	}
+}

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

@@ -52,40 +52,60 @@ public async Task Run(CancellationToken ct) {
 		}
 	}
 
+	// Wraps a state-handler invocation so a throw faults the projection with a
+	// debuggable fault reason: projection name, handler type, event position, and
+	// the handler's message. Only handler invocations are wrapped - infrastructure
+	// failures (state-stream reads, cache writes) propagate raw rather than
+	// blaming the user's handler.
+	private T InvokeHandler<T>(TFPos position, Func<T> invoke) {
+		try {
+			return invoke();
+		} catch (Exception ex) {
+			var tag = CheckpointTag.FromPosition(0, position.CommitPosition, position.PreparePosition);
+			throw new PartitionProcessingException(projectionName, stateHandler.GetType(), tag.ToString(), ex);
+		}
+	}
+
+	private void InvokeHandler(TFPos position, Action invoke) =>
+		InvokeHandler<object?>(position, () => {
+			invoke();
+			return null;
+		});
+
 	/// <summary>
 	/// Loads partition state into the state handler from cache, persisted result stream, or initializes fresh.
 	/// Returns true if the partition is new (not previously seen in this run or persisted).
 	/// </summary>
-	private async ValueTask<bool> LoadPartitionState(string partitionKey, CancellationToken ct) {
+	private async ValueTask<bool> LoadPartitionState(string partitionKey, TFPos position, CancellationToken ct) {
 		if (_stateCache.TryGet(partitionKey, out var cachedState)) {
 			// A null cached state means the handler explicitly set state to null (e.g. JS null).
 			// Load "null" so the handler gets JS null, not a fresh $init state.
-			stateHandler.Load(cachedState ?? "null");
+			InvokeHandler(position, () => stateHandler.Load(cachedState ?? "null"));
 			return false;
 		}
 
 		var persistedState = await loadPersistedState(partitionKey);
 		if (persistedState is not null) {
 			Log.Debug("Loaded persisted state for partition {Partition} in projection {Name}",
 				partitionKey, projectionName);
-			stateHandler.Load(persistedState);
+			InvokeHandler(position, () => stateHandler.Load(persistedState));
 			await _stateCache.Set(partitionKey, persistedState, ct);
 			return false;
 		}
 
-		stateHandler.Initialize();
+		InvokeHandler(position, stateHandler.Initialize);
 		return true;
 	}
 
 	// Loads the shared state into the state handler
-	private void LoadSharedState() {
+	private void LoadSharedState(TFPos position) {
 		if (!isBiState) return;
 
 		if (!_sharedStateInitialized) {
-			stateHandler.InitializeShared();
+			InvokeHandler(position, stateHandler.InitializeShared);
 			_sharedStateInitialized = true;
 		} else if (_sharedState != null) {
-			stateHandler.LoadShared(_sharedState);
+			InvokeHandler(position, () => stateHandler.LoadShared(_sharedState));
 		}
 	}
 
@@ -94,12 +114,15 @@ private async Task ProcessPartitionDeleted(PartitionEvent pe, CancellationToken
 
 		Log.Debug("Processing partition deleted partition={Partition}", partitionKey);
 
-		await LoadPartitionState(partitionKey, ct);
-		LoadSharedState();
+		await LoadPartitionState(partitionKey, pe.LogPosition, ct);
+		LoadSharedState(pe.LogPosition);
 
 		var checkpointTag = CheckpointTag.FromPosition(0, pe.LogPosition.CommitPosition, pe.LogPosition.PreparePosition);
 
-		var processed = stateHandler.ProcessPartitionDeleted(partitionKey, checkpointTag, out var newState);
+		var (processed, newState) = InvokeHandler(pe.LogPosition, () => {
+			var p = stateHandler.ProcessPartitionDeleted(partitionKey, checkpointTag, out var state);
+			return (p, state);
+		});
 
 		if (processed) {
 			await _stateCache.Set(partitionKey, newState, ct);
@@ -120,25 +143,31 @@ private async Task ProcessEvent(PartitionEvent pe, CancellationToken ct) {
 		Log.Verbose("Processing event stream={Stream} type={EventType} partition={Partition}",
 			projEvent.EventStreamId, projEvent.EventType, partitionKey);
 
-		var isNewPartition = await LoadPartitionState(partitionKey, ct);
-		LoadSharedState();
+		var isNewPartition = await LoadPartitionState(partitionKey, pe.LogPosition, ct);
+		LoadSharedState(pe.LogPosition);
 
 		var checkpointTag = CheckpointTag.FromPosition(0, pe.LogPosition.CommitPosition, pe.LogPosition.PreparePosition);
 
 		if (isNewPartition) {
-			stateHandler.ProcessPartitionCreated(partitionKey, checkpointTag, projEvent, out var createdEmittedEvents);
+			var createdEmittedEvents = InvokeHandler(pe.LogPosition, () => {
+				stateHandler.ProcessPartitionCreated(partitionKey, checkpointTag, projEvent, out var created);
+				return created;
+			});
 			if (emitEnabled)
 				_activeBuffer.AddEmittedEvents(createdEmittedEvents);
 		}
 
-		var processed = stateHandler.ProcessEvent(
-			partitionKey,
-			checkpointTag,
-			category: null, // todo: is this an important gap?
-			projEvent,
-			out var newState,
-			out var newSharedState,
-			out var emittedEvents);
+		var (processed, newState, newSharedState, emittedEvents) = InvokeHandler(pe.LogPosition, () => {
+			var p = stateHandler.ProcessEvent(
+				partitionKey,
+				checkpointTag,
+				category: null, // todo: is this an important gap?
+				projEvent,
+				out var state,
+				out var sharedState,
+				out var emitted);
+			return (p, state, sharedState, emitted);
+		});
 
 		if (processed) {
 			await _stateCache.Set(partitionKey, newState, ct);

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

@@ -4,6 +4,7 @@
 using System;
 using System.Collections.Generic;
 using System.Runtime.CompilerServices;
+using System.Runtime.ExceptionServices;
 using System.Security.Claims;
 using System.Text;
 using System.Threading;
@@ -144,13 +145,46 @@ private async Task Run(TFPos checkpoint, CancellationToken ct) {
 			partitionTasks[i] = Task.Run(() => processor.Run(partitionCt), partitionCt);
 		}
 
+		Exception partitionFault = null;
 		try {
-			await RunReadLoop(checkpoint, dispatcher, coordinator, ct);
+			// Race the read loop against the first partition exit: partition processors
+			// only complete early when they fault, and a continuous projection's read
+			// loop never completes, so awaiting the read loop alone left processor
+			// faults unobserved - the projection reported Running forever with no
+			// checkpoints, no persistence, and nothing logged (DB-2159).
+			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);
+					// Cancelling both tokens is what stops the read loop: every await
+					// in it is token-governed, including the final checkpoint marker
+					// (drain token), which must not wait on a checkpoint the dead
+					// partition will never ack. Completing the channels is what lets
+					// the sibling processors exit; doing it before awaiting the read
+					// loop is deliberate redundancy - if a future await in the read
+					// loop misses a token, channel closure still unblocks the writer
+					// instead of wedging the fault path again.
+					await readCts.CancelAsync();
+					await drainCts.CancelAsync();
+					dispatcher.Complete(partitionFault);
+					await readLoop.ConfigureAwait(ConfigureAwaitOptions.SuppressThrowing);
+					ExceptionDispatchInfo.Capture(partitionFault).Throw();
+				}
+			}
+
+			await readLoop;
 			dispatcher.Complete();
 		} catch (OperationCanceledException) when (ct.IsCancellationRequested) {
 			dispatcher.Complete();
 			throw;
-		} catch (Exception ex) {
+		} catch (Exception ex) when (!ReferenceEquals(ex, partitionFault)) {
 			Log.Error(ex, "ProjectionEngineV2 {Name} read loop failed", _config.ProjectionName);
 			dispatcher.Complete(ex);
 			throw;
@@ -160,6 +194,14 @@ private async Task Run(TFPos checkpoint, CancellationToken ct) {
 				await Task.WhenAll(partitionTasks);
 			} catch (OperationCanceledException) when (ct.IsCancellationRequested) {
 				// Expected on cancellation
+			} catch (Exception) when (partitionFault is not null) {
+				// The primary partition fault is already propagating; sibling
+				// processors failing during drain must not replace it. WhenAll only
+				// surfaces its first fault, so log distinct siblings off the tasks.
+				foreach (var task in partitionTasks) {
+					if (task.IsFaulted && !ReferenceEquals(task.Exception?.InnerException, partitionFault))
+						Log.Warning(task.Exception?.InnerException, "ProjectionEngineV2 {Name} sibling partition processor failed during drain", _config.ProjectionName);
+				}
 			}
 
 			// Dispose per-partition state handlers
@@ -170,7 +212,7 @@ private async Task Run(TFPos checkpoint, CancellationToken ct) {
 		}
 	}
 
-	private async Task RunReadLoop(TFPos checkpoint, PartitionDispatcher dispatcher, CheckpointCoordinator coordinator, CancellationToken ct) {
+	private async Task RunReadLoop(TFPos checkpoint, PartitionDispatcher dispatcher, CheckpointCoordinator coordinator, CancellationToken ct, CancellationToken drainCt) {
 		long eventsProcessed = 0;
 		long bytesProcessed = 0;
 		var lastCheckpointTime = Instant.Now;
@@ -263,9 +305,13 @@ private async Task RunReadLoop(TFPos checkpoint, PartitionDispatcher dispatcher,
 
 		// Inject a final checkpoint marker if the read position has advanced
 		// since the last checkpoint — covers both handled events and tails of
-		// filtered events that still moved the log position forward.
+		// filtered events that still moved the log position forward. The drain
+		// token (not ct) governs it: on graceful shutdown ct is already cancelled
+		// but the final checkpoint must still be written, while the partition-fault
+		// path cancels the drain so this cannot wait on a checkpoint the dead
+		// partition will never ack.
 		if (eventsProcessed > 0 || bytesProcessed > 0) {
-			await coordinator.InjectCheckpointMarker(lastLogPosition, CancellationToken.None);
+			await coordinator.InjectCheckpointMarker(lastLogPosition, drainCt);
 		}
 	}