Pulsar support for retry and DLQ

src/Transports/Pulsar/Wolverine.Pulsar.Tests/DocumentationSamples.cs

@@ -37,6 +37,14 @@ public static async Task configure()
 
                 // And all the normal Wolverine options...
                 .Sequential();
+
+
+            // Listen for incoming messages from a Pulsar topic with a shared subscription and using RETRY and DLQ queues
+            opts.ListenToPulsarTopic("persistent://public/default/three")
+                .WithSharedSubscriptionType()
+                .DeadLetterQueueing(new DeadLetterTopic(DeadLetterTopicMode.Native))
+                .RetryLetterQueueing(new RetryLetterTopic([TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(3), TimeSpan.FromSeconds(5)]))
+                .Sequential();
         });
 
         #endregion

src/Transports/Pulsar/Wolverine.Pulsar.Tests/PulsarNativeReliabilityTests.cs

@@ -0,0 +1,212 @@
+using JasperFx.Core;
+using Microsoft.Extensions.Hosting;
+using Oakton;
+using Shouldly;
+using Wolverine.ComplianceTests;
+using Wolverine.ComplianceTests.Compliance;
+using Wolverine.ComplianceTests.Scheduling;
+using Wolverine.Tracking;
+using Xunit;
+
+namespace Wolverine.Pulsar.Tests;
+
+public class PulsarNativeReliabilityTests : /*TransportComplianceFixture,*/ IAsyncLifetime
+{
+    public IHost WolverineHost;
+
+    public PulsarNativeReliabilityTests()
+    {
+    }
+
+    private IHostBuilder ConfigureBuilder()
+    {
+        return Host.CreateDefaultBuilder()
+            .UseWolverine(opts =>
+            {
+
+                var topic = Guid.NewGuid().ToString();
+                var topicPath = $"persistent://public/default/compliance{topic}";
+
+                opts.UsePulsar(b => { });
+
+                opts.IncludeType<SRMessageHandlers>();
+
+                opts.PublishMessage<SRMessage1>()
+                    .ToPulsarTopic(topicPath);
+
+                opts.ListenToPulsarTopic(topicPath)
+                    .WithSharedSubscriptionType()
+                    .DeadLetterQueueing(DeadLetterTopic.DefaultNative)
+                    .RetryLetterQueueing(new RetryLetterTopic([TimeSpan.FromSeconds(4), TimeSpan.FromSeconds(2), TimeSpan.FromSeconds(3)]))
+                    //.ProcessInline();
+                    .BufferedInMemory();
+
+
+                var topicPath2 = $"persistent://public/default/no-retry-{topic}";
+                opts.IncludeType<SRMessageHandlers>();
+
+                opts.PublishMessage<SRMessage2>()
+                    .ToPulsarTopic(topicPath2);
+
+                opts.ListenToPulsarTopic(topicPath2)
+                    .WithSharedSubscriptionType()
+                    .DeadLetterQueueing(DeadLetterTopic.DefaultNative)
+                    .DisableRetryLetterQueueing()
+                    .ProcessInline();
+
+            });
+    }
+
+    public async Task InitializeAsync()
+    {
+        WolverineHost = ConfigureBuilder().Build();
+        await WolverineHost.StartAsync();
+    }
+
+    [Fact]
+    public async Task run_setup_with_simulated_exception_in_handler()
+    {
+        var session =  await WolverineHost.TrackActivity(TimeSpan.FromSeconds(1000))
+            //.WaitForMessageToBeReceivedAt<SRMessage1>(WolverineHost)
+            .DoNotAssertOnExceptionsDetected()
+            .IncludeExternalTransports()
+            .WaitForCondition(new WaitForDeadLetteredMessage<SRMessage1>())
+            .SendMessageAndWaitAsync(new SRMessage1());
+
+
+        session.Sent.AllMessages();
+        session.MovedToErrorQueue
+            .MessagesOf<SRMessage1>()
+            .Count()
+            .ShouldBe(1);
+
+        session.Received
+            .MessagesOf<SRMessage1>()
+            .Count()
+            .ShouldBe(4);
+
+        session.Requeued
+            .MessagesOf<SRMessage1>()
+            .Count()
+            .ShouldBe(3);
+
+
+        // TODO: I Guess the capture of the envelope headers occurs before we manipulate it
+        var firstRequeuedEnvelope = session.Requeued.Envelopes().First();
+        firstRequeuedEnvelope.ShouldSatisfyAllConditions(
+            () => firstRequeuedEnvelope.Attempts.ShouldBe(1),
+            () => firstRequeuedEnvelope.Headers.ContainsKey("DELAY_TIME").ShouldBeFalse()
+        );
+        var secondRequeuedEnvelope = session.Requeued.Envelopes().Skip(1).First();
+        secondRequeuedEnvelope.ShouldSatisfyAllConditions(
+            () => secondRequeuedEnvelope.Attempts.ShouldBe(2),
+            () => secondRequeuedEnvelope.Headers.ContainsKey("DELAY_TIME").ShouldBeTrue(),
+            () => secondRequeuedEnvelope.Headers["DELAY_TIME"].ShouldBe(TimeSpan.FromSeconds(4).TotalMilliseconds.ToString())
+        );
+
+        var thirdRequeuedEnvelope = session.Requeued.Envelopes().Skip(2).First();
+        thirdRequeuedEnvelope.ShouldSatisfyAllConditions(
+            () => thirdRequeuedEnvelope.Attempts.ShouldBe(3),
+            () => thirdRequeuedEnvelope.Headers.ContainsKey("DELAY_TIME").ShouldBeTrue(),
+            () => thirdRequeuedEnvelope.Headers["DELAY_TIME"].ShouldBe(TimeSpan.FromSeconds(2).TotalMilliseconds.ToString())
+        );
+
+
+        var dlqEnvelope = session.MovedToErrorQueue.Envelopes().First();
+        dlqEnvelope.ShouldSatisfyAllConditions(
+            () => dlqEnvelope.Headers.ContainsKey(PulsarEnvelopeConstants.Exception).ShouldBeTrue(),
+            () => dlqEnvelope.Headers[PulsarEnvelopeConstants.ReconsumeTimes].ShouldBe("3")
+        );
+
+    }
+
+    [Fact]
+    public async Task run_setup_with_simulated_exception_in_handler_only_native_dead_lettered_queue()
+    {
+        var session = await WolverineHost.TrackActivity(TimeSpan.FromSeconds(100))
+            .DoNotAssertOnExceptionsDetected()
+            .IncludeExternalTransports()
+            .WaitForCondition(new WaitForDeadLetteredMessage<SRMessage2>())
+            .SendMessageAndWaitAsync(new SRMessage2());
+
+
+        session.Sent.AllMessages();
+        session.MovedToErrorQueue
+            .MessagesOf<SRMessage2>()
+            .Count()
+            .ShouldBe(1);
+
+        session.Received
+            .MessagesOf<SRMessage2>()
+            .Count()
+            .ShouldBe(1);
+
+        session.Requeued
+            .MessagesOf<SRMessage2>()
+            .Count()
+            .ShouldBe(0);
+
+
+
+
+        var firstEnvelope = session.MovedToErrorQueue.Envelopes().First();
+        firstEnvelope.ShouldSatisfyAllConditions(
+            () => firstEnvelope.Headers.ContainsKey(PulsarEnvelopeConstants.Exception).ShouldBeTrue(),
+            () => firstEnvelope.Headers.ContainsKey(PulsarEnvelopeConstants.ReconsumeTimes).ShouldBeFalse()
+        );
+
+    }
+
+
+
+    public async Task DisposeAsync()
+    {
+        await WolverineHost.StopAsync();
+        WolverineHost.Dispose();
+    }
+
+
+}
+
+public class SRMessage1;
+public class SRMessage2;
+
+
+public class SRMessageHandlers
+{
+    public Task Handle(SRMessage1 message, IMessageContext context)
+    {
+        throw new InvalidOperationException("Simulated exception");
+    }
+
+    public Task Handle(SRMessage2 message, IMessageContext context)
+    {
+        throw new InvalidOperationException("Simulated exception");
+    }
+
+}
+
+
+
+public class WaitForDeadLetteredMessage<T> : ITrackedCondition
+{
+
+    private bool _found;
+
+    public WaitForDeadLetteredMessage()
+    {
+
+    }
+
+    public void Record(EnvelopeRecord record)
+    {
+        if (record.Envelope.Message is T && record.MessageEventType == MessageEventType.MovedToErrorQueue )
+           // && record.Envelope.Destination?.ToString().Contains(_dlqTopic) == true)
+        {
+            _found = true;
+        }
+    }
+
+    public bool IsCompleted() => _found;
+}
+

src/Transports/Pulsar/Wolverine.Pulsar/DeadLetterTopic.cs

@@ -0,0 +1,73 @@
+
+
+namespace Wolverine.Pulsar;
+
+public enum DeadLetterTopicMode
+{
+    /// <summary>
+    /// Opt into using Pulsar's native dead letter topic approach. This is the default and recommended
+    /// </summary>
+    Native,
+
+    /// <summary>
+    /// Completely ignore Pulsar native dead letter topic in favor of Wolverine persistent dead letter queueing
+    /// </summary>
+    WolverineStorage
+}
+
+public class DeadLetterTopic
+{
+
+    public static DeadLetterTopic DefaultNative => new(DeadLetterTopicMode.Native);
+
+    private string? _topicName;
+
+    public DeadLetterTopicMode Mode { get; set; }
+
+    public DeadLetterTopic(DeadLetterTopicMode mode)
+    {
+        Mode = mode;
+    }
+
+    public DeadLetterTopic(string topicName, DeadLetterTopicMode mode)
+    {
+        _topicName = topicName;
+        Mode = mode;
+    }
+
+    public string TopicName
+    {
+        get => _topicName;
+        set => _topicName = value?? throw new ArgumentNullException(nameof(TopicName));
+    }
+
+    protected bool Equals(DeadLetterTopic other)
+    {
+        return _topicName == other._topicName;
+    }
+
+    public override bool Equals(object? obj)
+    {
+        if (ReferenceEquals(null, obj))
+        {
+            return false;
+        }
+
+        if (ReferenceEquals(this, obj))
+        {
+            return true;
+        }
+
+        if (obj.GetType() != this.GetType())
+        {
+            return false;
+        }
+
+        return Equals((DeadLetterTopic)obj);
+    }
+
+    public override int GetHashCode()
+    {
+        return _topicName.GetHashCode();
+    }
+}

src/Transports/Pulsar/Wolverine.Pulsar/ErrorHandling/PulsarNativeContinuationSource.cs

@@ -0,0 +1,21 @@
+using Wolverine.ErrorHandling;
+using Wolverine.Runtime;
+
+namespace Wolverine.Pulsar.ErrorHandling;
+
+public class PulsarNativeContinuationSource : IContinuationSource
+{
+    public string Description { get; }
+
+    public IContinuation Build(Exception ex, Envelope envelope)
+    {
+        // Only handle Pulsar envelopes/listeners
+        if (envelope.Listener is PulsarListener)
+        {
+            return new PulsarNativeResiliencyContinuation(ex);
+        }
+
+        // Return null to let the next continuation source handle it
+        return null;
+    }
+}

src/Transports/Pulsar/Wolverine.Pulsar/ErrorHandling/PulsarNativeResiliencyContinuation.cs

@@ -0,0 +1,45 @@
+using System.Diagnostics;
+using Wolverine.ErrorHandling;
+using Wolverine.Runtime;
+
+namespace Wolverine.Pulsar.ErrorHandling;
+
+public class PulsarNativeResiliencyContinuation : IContinuation
+{
+    private readonly Exception _exception;
+
+    public PulsarNativeResiliencyContinuation(Exception exception)
+    {
+        _exception = exception;
+    }
+
+    public async ValueTask ExecuteAsync(IEnvelopeLifecycle lifecycle, IWolverineRuntime runtime, DateTimeOffset now, Activity? activity)
+    {
+        var context = lifecycle as MessageContext;
+        var envelope = context?.Envelope;
+
+        if (envelope?.Listener is PulsarListener listener)
+        {
+            if (listener.NativeRetryLetterQueueEnabled && listener.RetryLetterTopic!.Retry.Count >= envelope.Attempts)
+            {
+                // Use native retry if enabled and attempts are within bounds
+                //await listener.MoveToScheduledUntilAsync(envelope, now);
+                await new ScheduledRetryContinuation(listener.RetryLetterTopic!.Retry[envelope.Attempts - 1])
+                    .ExecuteAsync(lifecycle, runtime, now, activity);
+                return;
+            }
+
+            if (listener.NativeDeadLetterQueueEnabled)
+            {
+                await new MoveToErrorQueue(_exception)
+                    .ExecuteAsync(lifecycle, runtime, now, activity);
+                //await listener.MoveToErrorsAsync(envelope, _exception);
+            }
+
+            return;
+        }
+
+        // Fall back to MoveToErrorQueue for other listener types
+        await new MoveToErrorQueue(_exception).ExecuteAsync(lifecycle, runtime, now, activity);
+    }
+}

src/Transports/Pulsar/Wolverine.Pulsar/ErrorHandling/PulsarNativeResiliencyPolicy.cs

@@ -0,0 +1,22 @@
+using Wolverine.Configuration;
+using Wolverine.ErrorHandling;
+using Wolverine.ErrorHandling.Matches;
+
+namespace Wolverine.Pulsar.ErrorHandling;
+
+public class PulsarNativeResiliencyPolicy : IWolverinePolicy
+{
+    public void Apply(WolverineOptions options)
+    {
+        var rule = new FailureRule(new AlwaysMatches());
+
+        rule.AddSlot(new PulsarNativeContinuationSource());
+
+        // Set the same source as the InfiniteSource to handle all other attempts
+        rule.InfiniteSource = new PulsarNativeContinuationSource();
+
+        // Add this rule to the global failure collection
+        // This ensures it will be checked before other rules
+        options.Policies.Failures.Add(rule);
+    }
+}