GH-4149: a Pulsar listener is not "started" until its subscription exists

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

@@ -0,0 +1,51 @@
+using Shouldly;
+using Wolverine.ComplianceTests;
+using Xunit;
+
+namespace Wolverine.Pulsar.Tests;
+
+// GH-4149. DotPulsar's IConsumerBuilder.Create() returns as soon as the consumer object exists -- the
+// Subscribe command goes to the broker on a background task. Wolverine did not wait for it, so
+// IHost.StartAsync() returned while the topic did not yet exist at the broker: the admin API answered
+// 404 for it at the instant start returned, five runs out of five.
+//
+// With SubscriptionInitialPosition defaulting to Latest, anything published into that window is not
+// delivered to the subscription and is not redeliverable -- on a brand-new topic there is no earlier
+// position to fall back to. It is silent message loss on first deployment, and it is what made the
+// Pulsar suite drop exactly the first message it published under parallel load.
+public class subscription_established_before_start_returns
+{
+    [Fact]
+    public async Task the_subscription_exists_at_the_broker_when_start_returns()
+    {
+        using var http = new HttpClient();
+
+        // Repeated because the defect is a race: on an idle machine the subscription often wins anyway.
+        // The assertion is on the broker's own view, not on message delivery, so it holds either way.
+        for (var attempt = 0; attempt < 3; attempt++)
+        {
+            var name = $"established-{Guid.NewGuid():N}";
+            var subscription = "sub-" + Guid.NewGuid().ToString("N");
+
+            using var host = await WolverineHost.ForAsync(opts =>
+            {
+                opts.UsePulsar(b => b.ServiceUrl(PulsarContainerFixture.ServiceUrl));
+                opts.ListenToPulsarTopic($"persistent://public/default/{name}")
+                    .SubscriptionName(subscription);
+            });
+
+            var response = await http.GetAsync(
+                $"{PulsarContainerFixture.HttpServiceUrl}/admin/v2/persistent/public/default/{name}/stats",
+                TestContext.Current.CancellationToken);
+
+            response.IsSuccessStatusCode.ShouldBeTrue(
+                $"attempt {attempt}: the topic did not exist at the broker when StartAsync returned " +
+                $"(admin API returned {(int)response.StatusCode})");
+
+            var stats = await response.Content.ReadAsStringAsync(TestContext.Current.CancellationToken);
+            stats.ShouldContain(subscription,
+                customMessage: $"attempt {attempt}: the topic existed but the subscription was not established " +
+                               "when StartAsync returned");
+        }
+    }
+}

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

@@ -32,7 +32,7 @@ internal class PulsarListener : IListener, ISupportDeadLetterQueue, ISupportNati
     // one at a time.
     private readonly BatchingChannel<Envelope>? _batching;
     private readonly Block<Envelope[]>? _batchFlush;
-    private readonly ILogger? _logger;
+    private readonly ILogger _logger;
     private readonly Schemas.IPulsarMessageCodec? _codec;
     private IProducer<ReadOnlySequence<byte>>? _retryLetterQueueProducer;
     private IProducer<ReadOnlySequence<byte>>? _dlqProducer;
@@ -51,6 +51,7 @@ public PulsarListener(IWolverineRuntime runtime, PulsarEndpoint endpoint, IRecei
         _receiver = receiver ?? throw new ArgumentNullException(nameof(receiver));
         _cancellation = cancellation;
         _codec = endpoint.MessageCodec;
+        _logger = runtime.LoggerFactory.CreateLogger<PulsarListener>();
 
         // GH-4047. Belt and braces with the bootstrap check in PulsarEndpoint.validateModeConfiguration(): that one
         // runs over every *listening* endpoint at startup, this one covers any path that builds a listener without
@@ -145,7 +146,6 @@ public PulsarListener(IWolverineRuntime runtime, PulsarEndpoint endpoint, IRecei
 
         if (endpoint.Mode == EndpointMode.Durable && endpoint.MaximumMessagesToReceive > 1)
         {
-            _logger = runtime.LoggerFactory.CreateLogger<PulsarListener>();
             _batchFlush = new Block<Envelope[]>((batch, _) => deliverBatchAsync(batch));
             _batching = new BatchingChannel<Envelope>(TimeSpan.FromMilliseconds(5), _batchFlush,
                 endpoint.MaximumMessagesToReceive);
@@ -211,6 +211,73 @@ public PulsarListener(IWolverineRuntime runtime, PulsarEndpoint endpoint, IRecei
         }
     }
 
+    /// <summary>
+    ///     GH-4149. Block until this listener's consumers have actually subscribed at the broker.
+    ///
+    ///     <para>DotPulsar's <c>IConsumerBuilder.Create()</c> returns as soon as the consumer object
+    ///     exists; the Subscribe command travels to the broker on a background task. Wolverine's listener
+    ///     startup did not wait for it, so <c>IHost.StartAsync()</c> returned with the listener reporting
+    ///     started while the topic did not yet exist at the broker — measured directly: the admin API
+    ///     answered <c>404</c> for the topic at the instant start returned, on five runs out of five.</para>
+    ///
+    ///     <para>Because <see cref="PulsarEndpoint.SubscriptionInitialPosition" /> defaults to
+    ///     <see cref="SubscriptionInitialPosition.Latest" />, anything published into that window is not
+    ///     delivered to the subscription at all. It is silently dropped: no error, no redelivery, and on a
+    ///     brand-new topic no earlier position to fall back to. That is a real message-loss window on
+    ///     first deployment of a service that publishes to a topic it also listens to, and it is what makes
+    ///     the Pulsar suite drop exactly the first message it publishes under parallel load.</para>
+    ///
+    ///     <para>A consumer leaves <see cref="ConsumerState.Disconnected" /> once the broker has
+    ///     acknowledged the subscribe, so that transition is the signal. Waiting for <em>any</em> state
+    ///     other than Disconnected rather than for Active specifically matters for Failover subscriptions,
+    ///     where a standby consumer is legitimately established but Inactive.</para>
+    ///
+    ///     <para>Bounded, and a timeout is logged rather than thrown: a broker that is slow or briefly
+    ///     unreachable at startup must not stop the host from coming up. The listener still works once
+    ///     DotPulsar connects — the wait closes the ordering race, it does not add a hard dependency.</para>
+    /// </summary>
+    internal async Task WaitForSubscriptionAsync(TimeSpan timeout)
+    {
+        await waitForConsumerAsync(_consumer, timeout);
+        await waitForConsumerAsync(_retryConsumer, timeout);
+    }
+
+    private async Task waitForConsumerAsync(IConsumer<ReadOnlySequence<byte>>? consumer, TimeSpan timeout)
+    {
+        if (consumer == null)
+        {
+            return;
+        }
+
+        // NOTE: deliberately the no-delay overload. DotPulsar's StateChangedFrom(state, TimeSpan, ct)
+        // takes a *settle* delay, not a timeout -- it waits the full TimeSpan and only then reports the
+        // state, so using it here added the whole budget to every single host start (measured: 10,030ms
+        // per start against a broker that had gone Active in milliseconds). The timeout has to be ours.
+        using var timeoutSource = CancellationTokenSource.CreateLinkedTokenSource(_cancellation);
+        timeoutSource.CancelAfter(timeout);
+
+        try
+        {
+            await consumer.StateChangedFrom(ConsumerState.Disconnected, timeoutSource.Token);
+        }
+        catch (OperationCanceledException) when (_cancellation.IsCancellationRequested)
+        {
+            // Shutting down before the subscription was ever established; nothing to report.
+        }
+        catch (OperationCanceledException)
+        {
+            _logger.LogWarning(
+                "The Pulsar consumer for topic {Topic} at {Address} had not subscribed after {Timeout}; starting the listener anyway. Messages published to this topic before the subscription is established may not be delivered to it.",
+                consumer.Topic, Address, timeout);
+        }
+        catch (Exception e)
+        {
+            _logger.LogWarning(e,
+                "Error waiting for the Pulsar subscription on topic {Topic} at {Address} to be established; starting the listener anyway",
+                consumer.Topic, Address);
+        }
+    }
+
     private void trySetupNativeResiliency(PulsarEndpoint endpoint, PulsarTransport transport)
     {
         if (!NativeRetryLetterQueueEnabled && !NativeDeadLetterQueueEnabled)
@@ -282,7 +349,7 @@ private async Task deliverBatchAsync(Envelope[] batch)
         }
         catch (Exception e)
         {
-            _logger?.LogError(e,
+            _logger.LogError(e,
                 "Failure receiving a batch of {Count} Pulsar messages at {Address}, deferring them for redelivery",
                 batch.Length, Address);
 
@@ -294,7 +361,7 @@ private async Task deliverBatchAsync(Envelope[] batch)
                 }
                 catch (Exception deferException)
                 {
-                    _logger?.LogError(deferException, "Failure deferring Pulsar message for envelope {EnvelopeId}", envelope.Id);
+                    _logger.LogError(deferException, "Failure deferring Pulsar message for envelope {EnvelopeId}", envelope.Id);
                 }
             }
         }
@@ -366,7 +433,7 @@ public async ValueTask DisposeAsync()
             }
             catch (Exception e)
             {
-                _logger?.LogDebug(e, "Error flushing the pending Pulsar receive batch at {Address}", Address);
+                _logger.LogDebug(e, "Error flushing the pending Pulsar receive batch at {Address}", Address);
             }
         }
 

src/Transports/Pulsar/Wolverine.Pulsar/PulsarTransport.cs

@@ -15,6 +15,13 @@ public class PulsarTransport : TransportBase<PulsarEndpoint>, IAsyncDisposable
 {
     public const string ProtocolName = "pulsar";
 
+    /// <summary>
+    ///     GH-4149. How long listener startup waits for a consumer to actually subscribe at the broker
+    ///     before giving up and starting anyway. Generous enough to cover topic auto-creation on a cold
+    ///     broker, bounded so an unreachable broker cannot stop the host from coming up.
+    /// </summary>
+    internal static readonly TimeSpan SubscriptionEstablishmentTimeout = TimeSpan.FromSeconds(10);
+
     private readonly LightweightCache<Uri, PulsarEndpoint> _endpoints;
 
     public PulsarTransport() : this(ProtocolName)
@@ -266,7 +273,7 @@ internal ISender BuildSender(PulsarEndpoint endpoint, IWolverineRuntime runtime)
     /// with the tenant id they were consumed under. The hot-tail (<see cref="PulsarReaderListener"/>) branch is
     /// preserved with the same per-tenant treatment. Modeled on <c>RabbitMqTransport.BuildListenerAsync</c>.
     /// </summary>
-    internal ValueTask<IListener> BuildListenerAsync(PulsarEndpoint endpoint, IReceiver receiver,
+    internal async ValueTask<IListener> BuildListenerAsync(PulsarEndpoint endpoint, IReceiver receiver,
         IWolverineRuntime runtime)
     {
         if (Tenants.Any() && endpoint.TenancyBehavior == TenancyBehavior.TenantAware)
@@ -280,10 +287,31 @@ internal ValueTask<IListener> BuildListenerAsync(PulsarEndpoint endpoint, IRecei
                 compound.Inner.Add(buildSingleListener(endpoint, wrapped, tenant.Transport, runtime));
             }
 
-            return ValueTask.FromResult<IListener>(compound);
+            foreach (var inner in compound.Inner)
+            {
+                await waitForSubscriptionAsync(inner);
+            }
+
+            return compound;
         }
 
-        return ValueTask.FromResult(buildSingleListener(endpoint, receiver, this, runtime));
+        var listener = buildSingleListener(endpoint, receiver, this, runtime);
+        await waitForSubscriptionAsync(listener);
+
+        return listener;
+    }
+
+    /// <summary>
+    ///     GH-4149. Do not report a listener started until its subscription actually exists at the broker.
+    ///     See <see cref="PulsarListener.WaitForSubscriptionAsync" /> for why this matters: with the default
+    ///     Latest initial position, anything published before the subscription is established is silently
+    ///     dropped rather than delivered or redelivered.
+    /// </summary>
+    private static ValueTask waitForSubscriptionAsync(IListener listener)
+    {
+        return listener is PulsarListener pulsar
+            ? new ValueTask(pulsar.WaitForSubscriptionAsync(SubscriptionEstablishmentTimeout))
+            : ValueTask.CompletedTask;
     }
 
     private static IListener buildSingleListener(PulsarEndpoint endpoint, IReceiver receiver,