fix: persistent subscription Nack and resubscribe overlap

src/Core/src/Eventuous.Subscriptions/EventSubscription.cs

@@ -34,6 +34,8 @@ public abstract class EventSubscription<T> : IMessageSubscription, IAsyncDisposa
 
     protected ulong Sequence;
 
+    int _resubscribing;
+
     protected EventSubscription(
             T                    options,
             ConsumePipe          consumePipe,
@@ -187,26 +189,39 @@ protected async ValueTask Handler(IMessageConsumeContext context) {
 
     protected abstract ValueTask Unsubscribe(CancellationToken cancellationToken);
 
+    /// <summary>
+    /// Stops the current subscription run (transport/runner) without affecting the overall lifecycle.
+    /// Override in derived classes to stop provider-specific transports.
+    /// </summary>
+    protected virtual ValueTask StopCurrentRun(CancellationToken cancellationToken) => default;
+
     [PublicAPI]
     [RequiresUnreferencedCode(AttrConstants.DynamicSerializationMessage)]
     [RequiresDynamicCode(AttrConstants.DynamicSerializationMessage)]
     protected virtual async Task Resubscribe(TimeSpan delay, CancellationToken cancellationToken) {
-        await Task.Delay(delay, cancellationToken).NoContext();
+        try {
+            // Stop the previous subscription run if it's still alive (e.g. when dropped from async worker path)
+            await StopCurrentRun(cancellationToken).NoContext();
 
-        while (IsRunning && IsDropped && !cancellationToken.IsCancellationRequested) {
-            try {
-                Log.SubscriptionResubscribing();
+            await Task.Delay(delay, cancellationToken).NoContext();
 
-                await Subscribe(cancellationToken).NoContext();
+            while (IsRunning && IsDropped && !cancellationToken.IsCancellationRequested) {
+                try {
+                    Log.SubscriptionResubscribing();
 
-                IsDropped = false;
-                _onSubscribed?.Invoke(Options.SubscriptionId);
+                    await Subscribe(cancellationToken).NoContext();
 
-                Log.SubscriptionResubscribed();
-            } catch (OperationCanceledException) { } catch (Exception e) {
-                Log.SubscriptionResubscribeFailed(e);
-                await Task.Delay(1000, cancellationToken).NoContext();
+                    IsDropped = false;
+                    _onSubscribed?.Invoke(Options.SubscriptionId);
+
+                    Log.SubscriptionResubscribed();
+                } catch (OperationCanceledException) { } catch (Exception e) {
+                    Log.SubscriptionResubscribeFailed(e);
+                    await Task.Delay(1000, cancellationToken).NoContext();
+                }
             }
+        } finally {
+            Interlocked.Exchange(ref _resubscribing, 0);
         }
     }
 
@@ -215,6 +230,9 @@ protected virtual async Task Resubscribe(TimeSpan delay, CancellationToken cance
     protected void Dropped(DropReason reason, Exception? exception) {
         if (!IsRunning) return;
 
+        // Prevent concurrent resubscribe attempts (e.g. NackOnAsyncWorker + transport drop callback)
+        if (Interlocked.CompareExchange(ref _resubscribing, 1, 0) != 0) return;
+
         Log.SubscriptionDropped(reason, exception);
 
         IsDropped = true;

src/Core/test/Eventuous.Tests.Subscriptions/ResubscribeOnHandlerFailureTests.cs

@@ -287,14 +287,17 @@ protected override ValueTask Subscribe(CancellationToken cancellationToken) {
             return default;
         }
 
-        protected override async ValueTask Unsubscribe(CancellationToken cancellationToken) {
+        protected override async ValueTask StopCurrentRun(CancellationToken cancellationToken) {
             if (_runner == null) return;
 
             await _runner.Stop(cancellationToken);
             _runner.Dispose();
             _runner = null;
         }
 
+        protected override ValueTask Unsubscribe(CancellationToken cancellationToken)
+            => StopCurrentRun(cancellationToken);
+
         async Task PollEvents(CancellationToken cancellationToken) {
             var checkpoint = await GetCheckpoint(cancellationToken);
             var start = (int)(checkpoint.Position ?? 0);

src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/KurrentDBCatchUpSubscriptionBase.cs

@@ -45,10 +45,17 @@
     /// Stops the subscription
     /// </summary>
     /// <param name="cancellationToken"></param>
+    protected override ValueTask StopCurrentRun(CancellationToken cancellationToken) {
+        Subscription?.Dispose();
+        Subscription = null;
+
+        return default;
+    }
+
     protected override async ValueTask Unsubscribe(CancellationToken cancellationToken) {
         try {
             Stopping.Cancel(false);
-            Subscription?.Dispose();
+            await StopCurrentRun(cancellationToken);
             await Task.Delay(100, cancellationToken);
         } catch (Exception) {
             // Nothing to see here

src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/PersistentSubscriptionBase.cs

@@ -187,11 +187,11 @@
 
         ctx.LogContext.MessageHandlingFailed(Options.SubscriptionId, ctx, exception);
 
-        if (Options.ThrowOnError) throw exception;
-
         var re           = ctx.Items.GetItem<ResolvedEvent>(ResolvedEventKey);
         var subscription = ctx.Items.GetItem<PersistentSubscription>(SubscriptionKey)!;
         await _handleEventProcessingFailure(Client, subscription, re, exception).NoContext();
+
+        if (Options.ThrowOnError) throw exception;
     }
 
     [RequiresDynamicCode(AttrConstants.DynamicSerializationMessage)]
@@ -229,13 +229,20 @@
     /// <returns></returns>
     protected abstract ulong GetContextStreamPosition(ResolvedEvent re);
 
+    protected override ValueTask StopCurrentRun(CancellationToken cancellationToken) {
+        _subscription?.Dispose();
+        _subscription = null;
+
+        return default;
+    }
+
     /// <summary>
     /// Unsubscribe from a persistent subscription
     /// </summary>
     /// <param name="cancellationToken"></param>
     protected override async ValueTask Unsubscribe(CancellationToken cancellationToken) {
         try {
-            _subscription?.Dispose();
+            await StopCurrentRun(cancellationToken).NoContext();
             Stopping.Cancel(false);
             await Task.Delay(100, cancellationToken);
         } catch (Exception) {

src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/PersistentSubscriptionNackTests.cs

@@ -0,0 +1,112 @@
+// Copyright (C) Eventuous HQ OÜ. All rights reserved
+// Licensed under the Apache License, Version 2.0.
+
+using Eventuous.KurrentDB.Subscriptions;
+using Eventuous.Producers;
+using Eventuous.Subscriptions;
+using Eventuous.Subscriptions.Context;
+using Eventuous.Subscriptions.Filters;
+using Eventuous.Tests.KurrentDB.Subscriptions.Fixtures;
+using Eventuous.Tests.Subscriptions.Base;
+using KurrentDB.Client;
+
+namespace Eventuous.Tests.KurrentDB.Subscriptions;
+
+/// <summary>
+/// Verifies that persistent subscriptions send Nack to KurrentDB when ThrowOnError is enabled,
+/// so that failed events are properly parked rather than silently lost. Regression test for #544.
+/// </summary>
+public class PersistentSubscriptionNackTests {
+    [Test]
+    [Category("Persistent subscription")]
+    [Retry(3)]
+    public async Task ShouldParkFailedEventWhenThrowOnErrorIsEnabled(CancellationToken cancellationToken) {
+        string? connectionString = null;
+        string? subscriptionId   = null;
+
+        var handler = new NackTestHandler();
+
+        var fixture = new PersistentSubscriptionFixture<
+            StreamPersistentSubscription,
+            StreamPersistentSubscriptionOptions,
+            NackTestHandler>(
+            handler,
+            (id, connString, stream, h, loggerFactory) => {
+                connectionString = connString;
+                subscriptionId   = id;
+
+                var settings = KurrentDBClientSettings.Create(connString);
+
+                return new(
+                    new KurrentDBClient(settings),
+                    new() {
+                        StreamName     = stream,
+                        SubscriptionId = id,
+                        ThrowOnError   = true,
+                        FailureHandler = (_, subscription, resolvedEvent, exception)
+                            => subscription.Nack(PersistentSubscriptionNakEventAction.Park, exception.Message, resolvedEvent)
+                    },
+                    new ConsumePipe().AddDefaultConsumer(h),
+                    loggerFactory
+                );
+            },
+            autoStart: false
+        );
+
+        await fixture.InitializeAsync();
+        await fixture.Start();
+
+        var testEvents = TestEvent.CreateMany(5);
+        await fixture.Producer.Produce(fixture.Stream, testEvents, new(), cancellationToken: cancellationToken);
+
+        // Wait for handler to fail once and then recover after resubscription
+        using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
+        cts.CancelAfter(TimeSpan.FromSeconds(15));
+
+        try {
+            while (!cts.Token.IsCancellationRequested) {
+                // After the first event is parked, the remaining 4 should be processed
+                if (handler is { HasFailed: true, SuccessCount: >= 4 }) break;
+
+                await Task.Delay(100, cts.Token);
+            }
+        } catch (OperationCanceledException) when (cts.Token.IsCancellationRequested) {
+            // Fall through to assertions
+        }
+
+        await fixture.Stop();
+
+        await Assert.That(handler.HasFailed).IsTrue();
+        await Assert.That(handler.SuccessCount).IsGreaterThanOrEqualTo(4);
+
+        // Verify that the failed event was actually parked in KurrentDB
+        var psClient = new KurrentDBPersistentSubscriptionsClient(KurrentDBClientSettings.Create(connectionString!));
+        var info     = await psClient.GetInfoToStreamAsync(fixture.Stream, subscriptionId!, cancellationToken: cancellationToken);
+
+        await Assert.That(info.Stats.ParkedMessageCount).IsGreaterThanOrEqualTo(1);
+
+        await fixture.DisposeAsync();
+    }
+}
+
+/// <summary>
+/// Handler that throws on the first event it sees, then succeeds on all subsequent events.
+/// Tracks both the failure and the number of successfully processed events.
+/// </summary>
+file class NackTestHandler : BaseEventHandler {
+    int _failCount;
+    int _successCount;
+
+    public bool HasFailed    => Volatile.Read(ref _failCount) > 0;
+    public int  SuccessCount => Volatile.Read(ref _successCount);
+
+    public override ValueTask<EventHandlingStatus> HandleEvent(IMessageConsumeContext context) {
+        if (Interlocked.CompareExchange(ref _failCount, 1, 0) == 0) {
+            throw new InvalidOperationException("Simulated handler failure for nack test");
+        }
+
+        Interlocked.Increment(ref _successCount);
+
+        return new(EventHandlingStatus.Success);
+    }
+}

src/Redis/src/Eventuous.Redis/Subscriptions/RedisSubscriptionBase.cs

@@ -44,14 +44,17 @@ protected override async ValueTask Subscribe(CancellationToken cancellationToken
         _runner = new TaskRunner(token => PollingQuery(position + 1, token)).Start();
     }
 
-    protected override async ValueTask Unsubscribe(CancellationToken cancellationToken) {
+    protected override async ValueTask StopCurrentRun(CancellationToken cancellationToken) {
         if (_runner == null) return;
 
         await _runner.Stop(cancellationToken);
         _runner.Dispose();
         _runner = null;
     }
 
+    protected override ValueTask Unsubscribe(CancellationToken cancellationToken)
+        => StopCurrentRun(cancellationToken);
+
     const string ContentType = "application/json";
 
     TaskRunner? _runner;

src/Relational/src/Eventuous.Sql.Base/Subscriptions/SqlSubscriptionBase.cs

@@ -230,14 +230,17 @@
     /// Stops the subscription.
     /// </summary>
     /// <param name="cancellationToken"></param>
-    protected override async ValueTask Unsubscribe(CancellationToken cancellationToken) {
+    protected override async ValueTask StopCurrentRun(CancellationToken cancellationToken) {
         if (_runner == null) return;
 
         await _runner.Stop(cancellationToken);
         _runner.Dispose();
         _runner = null;
     }
 
+    protected override ValueTask Unsubscribe(CancellationToken cancellationToken)
+        => StopCurrentRun(cancellationToken);
+
     /// <summary>
     /// This function is called before the subscription starts.
     /// </summary>