diff --git a/src/Transports/Pulsar/Wolverine.Pulsar.Tests/subscription_established_before_start_returns.cs b/src/Transports/Pulsar/Wolverine.Pulsar.Tests/subscription_established_before_start_returns.cs new file mode 100644 index 000000000..8f1f8f50e --- /dev/null +++ b/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"); + } + } +} diff --git a/src/Transports/Pulsar/Wolverine.Pulsar/PulsarListener.cs b/src/Transports/Pulsar/Wolverine.Pulsar/PulsarListener.cs index 8e78c812e..d38ee0cab 100644 --- a/src/Transports/Pulsar/Wolverine.Pulsar/PulsarListener.cs +++ b/src/Transports/Pulsar/Wolverine.Pulsar/PulsarListener.cs @@ -32,7 +32,7 @@ internal class PulsarListener : IListener, ISupportDeadLetterQueue, ISupportNati // one at a time. private readonly BatchingChannel? _batching; private readonly Block? _batchFlush; - private readonly ILogger? _logger; + private readonly ILogger _logger; private readonly Schemas.IPulsarMessageCodec? _codec; private IProducer>? _retryLetterQueueProducer; private IProducer>? _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(); // 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(); _batchFlush = new Block((batch, _) => deliverBatchAsync(batch)); _batching = new BatchingChannel(TimeSpan.FromMilliseconds(5), _batchFlush, endpoint.MaximumMessagesToReceive); @@ -211,6 +211,73 @@ public PulsarListener(IWolverineRuntime runtime, PulsarEndpoint endpoint, IRecei } } + /// + /// GH-4149. Block until this listener's consumers have actually subscribed at the broker. + /// + /// DotPulsar's IConsumerBuilder.Create() 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 IHost.StartAsync() returned with the listener reporting + /// started while the topic did not yet exist at the broker — measured directly: the admin API + /// answered 404 for the topic at the instant start returned, on five runs out of five. + /// + /// Because defaults to + /// , 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. + /// + /// A consumer leaves once the broker has + /// acknowledged the subscribe, so that transition is the signal. Waiting for any state + /// other than Disconnected rather than for Active specifically matters for Failover subscriptions, + /// where a standby consumer is legitimately established but Inactive. + /// + /// 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. + /// + internal async Task WaitForSubscriptionAsync(TimeSpan timeout) + { + await waitForConsumerAsync(_consumer, timeout); + await waitForConsumerAsync(_retryConsumer, timeout); + } + + private async Task waitForConsumerAsync(IConsumer>? 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); } } diff --git a/src/Transports/Pulsar/Wolverine.Pulsar/PulsarTransport.cs b/src/Transports/Pulsar/Wolverine.Pulsar/PulsarTransport.cs index 71d0d5c5a..cceef2c6d 100644 --- a/src/Transports/Pulsar/Wolverine.Pulsar/PulsarTransport.cs +++ b/src/Transports/Pulsar/Wolverine.Pulsar/PulsarTransport.cs @@ -15,6 +15,13 @@ public class PulsarTransport : TransportBase, IAsyncDisposable { public const string ProtocolName = "pulsar"; + /// + /// 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. + /// + internal static readonly TimeSpan SubscriptionEstablishmentTimeout = TimeSpan.FromSeconds(10); + private readonly LightweightCache _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 () branch is /// preserved with the same per-tenant treatment. Modeled on RabbitMqTransport.BuildListenerAsync. /// - internal ValueTask BuildListenerAsync(PulsarEndpoint endpoint, IReceiver receiver, + internal async ValueTask BuildListenerAsync(PulsarEndpoint endpoint, IReceiver receiver, IWolverineRuntime runtime) { if (Tenants.Any() && endpoint.TenancyBehavior == TenancyBehavior.TenantAware) @@ -280,10 +287,31 @@ internal ValueTask BuildListenerAsync(PulsarEndpoint endpoint, IRecei compound.Inner.Add(buildSingleListener(endpoint, wrapped, tenant.Transport, runtime)); } - return ValueTask.FromResult(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; + } + + /// + /// GH-4149. Do not report a listener started until its subscription actually exists at the broker. + /// See 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. + /// + 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,