Pulsar: subscription initial position (Earliest/Latest) (#3178)

docs/guide/messaging/transports/pulsar.md

@@ -63,6 +63,30 @@ Depending on demand, the Pulsar transport will be enhanced to support convention
 topic routing later.
 ::: 
 
+## Subscription Initial Position
+
+When a Wolverine listener creates a **brand-new** Pulsar subscription, you can control where that subscription
+starts reading. This is the Pulsar analogue of the Kafka transport's `BeginAtEarliest()` / `BeginAtLatest()`:
+
+```csharp
+opts.ListenToPulsarTopic("persistent://public/default/orders")
+    // Replay from the earliest message still retained on the topic
+    .BeginAtEarliest();
+
+opts.ListenToPulsarTopic("persistent://public/default/notifications")
+    // Only consume messages published after the subscription is created (the default)
+    .BeginAtLatest();
+
+// Or set it explicitly with the DotPulsar enum:
+opts.ListenToPulsarTopic("persistent://public/default/audit")
+    .SubscriptionInitialPosition(SubscriptionInitialPosition.Earliest);
+```
+
+This only affects the **first** read of a not-yet-existing subscription. Once the subscription exists, Pulsar
+resumes from its committed cursor on every restart regardless of this setting, and `Earliest` can only replay
+messages that are still retained on the topic. The default is `SubscriptionInitialPosition.Latest`, matching
+DotPulsar's own default.
+
 ## Read Only Subscriptions <Badge type="tip" text="3.13" />
 
 As part of Wolverine's "Requeue" error handling action, the Pulsar transport tries to quietly create a matching sender

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

@@ -0,0 +1,166 @@
+using System.Collections.Concurrent;
+using DotPulsar;
+using JasperFx.Core;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Shouldly;
+using Wolverine.Configuration;
+using Wolverine.ComplianceTests;
+using Xunit;
+
+namespace Wolverine.Pulsar.Tests;
+
+// GH-3178: subscription initial position (Earliest/Latest) — the Pulsar analogue of Kafka's
+// BeginAtEarliest/BeginAtLatest cold-start control (#3146).
+public class subscription_initial_position
+{
+    private static PulsarEndpoint applyListener(Action<PulsarListenerConfiguration> configure)
+    {
+        var transport = new PulsarTransport();
+        var endpoint = new PulsarEndpoint("pulsar://persistent/public/default/events".ToUri(), transport);
+        var config = new PulsarListenerConfiguration(endpoint);
+        configure(config);
+        ((IDelayedEndpointConfiguration)config).Apply();
+        return endpoint;
+    }
+
+    // ---- config unit tests (no broker) ----
+
+    [Fact]
+    public void default_initial_position_is_latest()
+    {
+        var endpoint = new PulsarEndpoint("pulsar://persistent/public/default/events".ToUri(), new PulsarTransport());
+        endpoint.SubscriptionInitialPosition.ShouldBe(SubscriptionInitialPosition.Latest);
+    }
+
+    [Fact]
+    public void begin_at_earliest_sets_initial_position()
+    {
+        var endpoint = applyListener(c => c.BeginAtEarliest());
+        endpoint.SubscriptionInitialPosition.ShouldBe(SubscriptionInitialPosition.Earliest);
+    }
+
+    [Fact]
+    public void begin_at_latest_sets_initial_position()
+    {
+        var endpoint = applyListener(c => c.BeginAtLatest());
+        endpoint.SubscriptionInitialPosition.ShouldBe(SubscriptionInitialPosition.Latest);
+    }
+
+    [Fact]
+    public void subscription_initial_position_sets_value()
+    {
+        var endpoint = applyListener(c => c.SubscriptionInitialPosition(SubscriptionInitialPosition.Earliest));
+        endpoint.SubscriptionInitialPosition.ShouldBe(SubscriptionInitialPosition.Earliest);
+    }
+
+    // ---- end-to-end (Pulsar docker) ----
+
+    [Fact]
+    public async Task earliest_replays_messages_published_before_the_subscription_existed()
+    {
+        var topic = $"persistent://public/default/initpos-early-{Guid.NewGuid():N}";
+        var subscription = "sub-" + Guid.NewGuid().ToString("N");
+
+        using var publisher = await WolverineHost.ForAsync(opts =>
+        {
+            opts.UsePulsar(b => b.ServiceUrl(PulsarContainerFixture.ServiceUrl));
+            opts.PublishAllMessages().ToPulsarTopic(topic).SendInline();
+        });
+
+        // Publish BEFORE any consumer/subscription exists for this topic.
+        for (var i = 0; i < 4; i++)
+        {
+            await publisher.SendAsync(new InitialPositionMessage { Id = "pre-" + i });
+        }
+
+        using var consumer = await WolverineHost.ForAsync(opts =>
+        {
+            opts.UsePulsar(b => b.ServiceUrl(PulsarContainerFixture.ServiceUrl));
+            opts.ListenToPulsarTopic(topic)
+                .SubscriptionName(subscription)
+                .ProcessInline()
+                .BeginAtEarliest();
+            opts.Services.AddSingleton<InitialPositionSink>();
+            opts.Discovery.DisableConventionalDiscovery().IncludeType<InitialPositionHandler>();
+        });
+
+        var sink = consumer.Services.GetRequiredService<InitialPositionSink>();
+        await waitForCountAsync(sink.Received, 4);
+        sink.Received.OrderBy(x => x).ShouldBe(["pre-0", "pre-1", "pre-2", "pre-3"]);
+    }
+
+    [Fact]
+    public async Task latest_consumes_only_messages_published_after_the_subscription_exists()
+    {
+        var topic = $"persistent://public/default/initpos-late-{Guid.NewGuid():N}";
+        var subscription = "sub-" + Guid.NewGuid().ToString("N");
+
+        using var publisher = await WolverineHost.ForAsync(opts =>
+        {
+            opts.UsePulsar(b => b.ServiceUrl(PulsarContainerFixture.ServiceUrl));
+            opts.PublishAllMessages().ToPulsarTopic(topic).SendInline();
+        });
+
+        // Published before the Latest subscription exists -> must NOT be delivered.
+        for (var i = 0; i < 3; i++)
+        {
+            await publisher.SendAsync(new InitialPositionMessage { Id = "pre-" + i });
+        }
+
+        using var consumer = await WolverineHost.ForAsync(opts =>
+        {
+            opts.UsePulsar(b => b.ServiceUrl(PulsarContainerFixture.ServiceUrl));
+            opts.ListenToPulsarTopic(topic)
+                .SubscriptionName(subscription)
+                .ProcessInline()
+                .BeginAtLatest();
+            opts.Services.AddSingleton<InitialPositionSink>();
+            opts.Discovery.DisableConventionalDiscovery().IncludeType<InitialPositionHandler>();
+        });
+
+        // Give the subscription time to be established before publishing the "post" messages.
+        await Task.Delay(3.Seconds());
+
+        for (var i = 0; i < 3; i++)
+        {
+            await publisher.SendAsync(new InitialPositionMessage { Id = "post-" + i });
+        }
+
+        var sink = consumer.Services.GetRequiredService<InitialPositionSink>();
+        await waitForCountAsync(sink.Received, 3);
+
+        // Only the post-subscription messages, never the pre-subscription ones.
+        sink.Received.OrderBy(x => x).ShouldBe(["post-0", "post-1", "post-2"]);
+    }
+
+    private static async Task waitForCountAsync(ConcurrentBag<string> bag, int count, int timeoutMs = 30000)
+    {
+        var cutoff = DateTimeOffset.UtcNow.AddMilliseconds(timeoutMs);
+        while (DateTimeOffset.UtcNow < cutoff)
+        {
+            if (bag.Count >= count) return;
+            await Task.Delay(100);
+        }
+
+        throw new TimeoutException($"Only received {bag.Count} of {count} expected messages within {timeoutMs}ms");
+    }
+}
+
+public class InitialPositionMessage
+{
+    public string Id { get; set; } = string.Empty;
+}
+
+public class InitialPositionSink
+{
+    public ConcurrentBag<string> Received { get; } = new();
+}
+
+public class InitialPositionHandler
+{
+    public static void Handle(InitialPositionMessage message, InitialPositionSink sink)
+    {
+        sink.Received.Add(message.Id);
+    }
+}

src/Transports/Pulsar/Wolverine.Pulsar/PulsarEndpoint.cs

@@ -34,6 +34,16 @@ protected override PulsarEnvelopeMapper buildMapper(IWolverineRuntime runtime)
     public string? TopicName { get; private set; }
     public string SubscriptionName { get; internal set; } = "Wolverine";
     public SubscriptionType SubscriptionType { get; internal set; } = SubscriptionType.Exclusive;
+
+    /// <summary>
+    ///     Where a brand-new subscription starts consuming: <see cref="SubscriptionInitialPosition.Latest"/>
+    ///     (default — only messages published after the subscription is created) or
+    ///     <see cref="SubscriptionInitialPosition.Earliest"/> (replay from the start of the topic's
+    ///     retained backlog). Only applies on the first read of a not-yet-existing subscription.
+    /// </summary>
+    public SubscriptionInitialPosition SubscriptionInitialPosition { get; internal set; } =
+        SubscriptionInitialPosition.Latest;
+
     public bool EnableRequeue { get; internal set; } = true;
     public bool UnsubscribeOnClose { get; internal set; } = true;
 

src/Transports/Pulsar/Wolverine.Pulsar/PulsarListener.cs

@@ -52,6 +52,7 @@ public PulsarListener(IWolverineRuntime runtime, PulsarEndpoint endpoint, IRecei
         _consumer = transport.Client!.NewConsumer()
             .SubscriptionName(endpoint.SubscriptionName)
             .SubscriptionType(endpoint.SubscriptionType)
+            .InitialPosition(endpoint.SubscriptionInitialPosition)
             .Topic(endpoint.PulsarTopic())
             .Create();
 
@@ -131,6 +132,7 @@ private IConsumer<ReadOnlySequence<byte>> createRetryConsumer(PulsarEndpoint end
         return transport.Client!.NewConsumer()
             .SubscriptionName(endpoint.SubscriptionName)
             .SubscriptionType(endpoint.SubscriptionType)
+            .InitialPosition(endpoint.SubscriptionInitialPosition)
             .Topic(topicRetry!.ToString())
             .Create();
     }

src/Transports/Pulsar/Wolverine.Pulsar/PulsarTransportExtensions.cs

@@ -208,6 +208,44 @@ public PulsarListenerConfiguration SubscriptionType(SubscriptionType subscriptio
         return this;
     }
 
+    /// <summary>
+    /// Set where a brand-new subscription starts consuming. Only affects the first read of a
+    /// not-yet-existing subscription; once the subscription exists, Pulsar resumes from its
+    /// committed position regardless of this setting.
+    /// </summary>
+    /// <param name="initialPosition"></param>
+    /// <returns></returns>
+    public PulsarListenerConfiguration SubscriptionInitialPosition(SubscriptionInitialPosition initialPosition)
+    {
+        add(e => { e.SubscriptionInitialPosition = initialPosition; });
+        return this;
+    }
+
+    /// <summary>
+    /// Start a brand-new subscription at the earliest retained message in the topic (replay the
+    /// existing backlog). Pulsar analogue of the Kafka transport's <c>BeginAtEarliest()</c>. Only
+    /// applies on the first read of a not-yet-existing subscription.
+    /// </summary>
+    /// <returns></returns>
+    public PulsarListenerConfiguration BeginAtEarliest()
+    {
+        add(e => { e.SubscriptionInitialPosition = DotPulsar.SubscriptionInitialPosition.Earliest; });
+        return this;
+    }
+
+    /// <summary>
+    /// Start a brand-new subscription at the latest position so only messages published after the
+    /// subscription is created are consumed (DotPulsar's default). Pulsar analogue of the Kafka
+    /// transport's <c>BeginAtLatest()</c>. Only applies on the first read of a not-yet-existing
+    /// subscription.
+    /// </summary>
+    /// <returns></returns>
+    public PulsarListenerConfiguration BeginAtLatest()
+    {
+        add(e => { e.SubscriptionInitialPosition = DotPulsar.SubscriptionInitialPosition.Latest; });
+        return this;
+    }
+
     /// <summary>
     /// Override the Pulsar subscription type to  <see cref="DotPulsar.SubscriptionType.Failover"/> for just this topic
     /// </summary>