diff --git a/src/KurrentDB.Projections.V2.Tests/Unit/ProjectionEngineV2LifecycleTests.cs b/src/KurrentDB.Projections.V2.Tests/Unit/ProjectionEngineV2LifecycleTests.cs index fd6519e0523..93dfff232dd 100644 --- a/src/KurrentDB.Projections.V2.Tests/Unit/ProjectionEngineV2LifecycleTests.cs +++ b/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]; diff --git a/src/KurrentDB.Projections.V2/Services/Processing/V2/PartitionProcessingException.cs b/src/KurrentDB.Projections.V2/Services/Processing/V2/PartitionProcessingException.cs new file mode 100644 index 00000000000..78c8834e4df --- /dev/null +++ b/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; + +/// +/// 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. +/// +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) { + } +} diff --git a/src/KurrentDB.Projections.V2/Services/Processing/V2/PartitionProcessor.cs b/src/KurrentDB.Projections.V2/Services/Processing/V2/PartitionProcessor.cs index 968490ff16b..6806b85961e 100644 --- a/src/KurrentDB.Projections.V2/Services/Processing/V2/PartitionProcessor.cs +++ b/src/KurrentDB.Projections.V2/Services/Processing/V2/PartitionProcessor.cs @@ -52,15 +52,35 @@ 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(TFPos position, Func 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(position, () => { + invoke(); + return null; + }); + /// /// 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). /// - private async ValueTask LoadPartitionState(string partitionKey, CancellationToken ct) { + private async ValueTask 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; } @@ -68,24 +88,24 @@ private async ValueTask LoadPartitionState(string partitionKey, Cancellati 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); diff --git a/src/KurrentDB.Projections.V2/Services/Processing/V2/ProjectionEngineV2.cs b/src/KurrentDB.Projections.V2/Services/Processing/V2/ProjectionEngineV2.cs index 183cf404a2b..176ebce3e22 100644 --- a/src/KurrentDB.Projections.V2/Services/Processing/V2/ProjectionEngineV2.cs +++ b/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); } }