diff --git a/.github/workflows/integration-tests.yml b/.github/workflows/integration-tests.yml new file mode 100644 index 0000000..efafd7b --- /dev/null +++ b/.github/workflows/integration-tests.yml @@ -0,0 +1,32 @@ +name: Integration tests +run-name: integration-tests-${{ github.ref_name }}-${{ github.run_number }} + +# The publish workflow runs no tests (run-tests: false), so nothing else exercises the bus against a +# real broker. These start RabbitMQ in Docker through Testcontainers, which GitHub's Ubuntu runners have. +on: + pull_request: + branches: [main] + push: + branches: [main] + workflow_dispatch: + +permissions: + contents: read + +concurrency: + group: integration-tests-${{ github.ref }} + cancel-in-progress: true + +jobs: + integration-tests: + runs-on: ubuntu-latest + timeout-minutes: 30 + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-dotnet@v4 + with: + dotnet-version: '10.0.x' + + - name: Test + run: dotnet test SW.Bus.IntegrationTests/SW.Bus.IntegrationTests.csproj --logger "console;verbosity=normal" diff --git a/README.md b/README.md index 9b24149..2ec9e63 100644 --- a/README.md +++ b/README.md @@ -903,6 +903,7 @@ services.AddBus(config => // ── Publishing and shutdown ─────────────────────────────────────────── config.PublisherConfirms = true; // a publish waits for the broker's confirm + config.PublishTimeout = TimeSpan.FromSeconds(15); // a publish the broker hasn't taken by then throws config.DrainTimeout = TimeSpan.FromSeconds(30); // wait for running messages on stop // ── Broadcast listeners ─────────────────────────────────────────────── @@ -1022,7 +1023,7 @@ dotnet test SW.Bus.IntegrationTests/SW.Bus.IntegrationTests.csproj **`ConcurrencyTests`**: a queue runs as many messages at once as its concurrency; at 1 it runs them one at a time in order; concurrency comes from the consumer options, then `AddQueueOption`, then `DefaultConsumerConcurrency`, and prefetch caps it; a refresh moves a consumer to a new concurrency without processing a message twice; a refresh applies a new prefetch; concurrent publishes from one scope and from many scopes all arrive. -**`ConnectionTests`**: a publish against a stopped broker throws, and publishing resumes when it returns; a broker restart leaves exactly one consumer per queue (read through `IConsumerReader`) and on the node queue; a host whose broker is down at boot starts, reports not connected, and connects when the broker returns. +**`ConnectionTests`**: a publish against a stopped broker throws, and publishing resumes when it returns; a publish the broker blocks (a memory alarm) fails after `PublishTimeout`, and publishing resumes when the alarm clears; the consumer connection is named after the application; a broker restart leaves exactly one consumer per queue (read through `IConsumerReader`) and on the node queue; a host whose broker is down at boot starts, reports not connected, and connects when the broker returns. **`DrainAndRetryTests`**: stopping lets running messages finish and acknowledge, and returns the ones not started; a message still running at the drain timeout returns to its queue; a failing message is retried its retry count and then moved to the bad queue with the same headers as before. @@ -1077,8 +1078,14 @@ With `BusOptions.PublisherConfirms` (default true), `IPublish`, `IPublishWithVal - Each DI scope publishes on a channel of its own, opened on the scope's first publish and closed with the scope. The channel is safe to publish on from several threads of the scope. - A publish waits about one broker round trip (plus the broker's disk write for persistent messages) for its confirm. Publishes in parallel (for example `Task.WhenAll`) share that wait. -- The client adds an `x-dotnet-pub-seq-no` header to each confirmed message, its own tracking number. It is not copied to the dead-letter copy of a failed message. +- The client adds an `x-dotnet-pub-seq-no` header to each confirmed message, its own tracking number. The original's is not copied to the dead-letter copy of a failed message; the copy carries the number of its own confirmed publish. - The publisher connection and the consumer connection are separate, as before, so the broker's flow control on a busy publisher never holds back consumer acknowledgements. +- A publish waits at most `BusOptions.PublishTimeout` (default 15 s) to open its channel, send, and be confirmed, then throws a `BusException`. A broker short of memory or disk blocks every connection that publishes until the alarm clears; without the timeout, each publishing request would wait that long. A publish that timed out may or may not have reached the broker. `TimeSpan.Zero` waits as long as it takes. +- A channel's close is bounded too, so a scope that ends while the broker blocks publishers is not held either. + +### Dead-lettering + +A message that fails its last retry is copied to its bad queue on a confirmed channel of the publisher connection, and only then acknowledged. If the broker does not confirm the copy (it is down, or blocking publishers past `PublishTimeout`), the message is rejected to its retry queue instead, and comes back after its retry delay to be dead-lettered again. A failed message is therefore never lost between the two, though a crash between the copy and the acknowledgement can leave it in the bad queue twice. ### Connecting and recovery diff --git a/SW.Bus.IntegrationTests/Broker.cs b/SW.Bus.IntegrationTests/Broker.cs index 1204cba..0985b23 100644 --- a/SW.Bus.IntegrationTests/Broker.cs +++ b/SW.Bus.IntegrationTests/Broker.cs @@ -33,6 +33,29 @@ public static async Task StartApp(RabbitMqContainer container) await Exec(container, "rabbitmqctl", "await_startup"); } + /// + /// Raises a memory alarm, as a broker short of memory does: it blocks every connection that publishes, + /// and stops reading from it, until the alarm clears. Consumers are not blocked. + /// + public static async Task BlockPublishers(RabbitMqContainer container) => + await Exec(container, "rabbitmqctl", "set_vm_memory_high_watermark", "absolute", "1"); + + public static async Task UnblockPublishers(RabbitMqContainer container) => + await Exec(container, "rabbitmqctl", "set_vm_memory_high_watermark", "0.4"); + + /// The client-provided names of the broker's connections, read through the management API. + public static async Task ConnectionNames(RabbitMqContainer container) + { + using var all = await Management(container, "/api/connections"); + if (all == null) return []; + return all.RootElement.EnumerateArray() + .Select(c => c.TryGetProperty("client_properties", out var p) && p.TryGetProperty("connection_name", out var n) + ? n.GetString() + : null) + .Where(n => n != null) + .ToArray()!; + } + private static async Task Exec(RabbitMqContainer container, params string[] command) { var result = await container.ExecAsync(command); diff --git a/SW.Bus.IntegrationTests/ConnectionTests.cs b/SW.Bus.IntegrationTests/ConnectionTests.cs index 7e0c2ac..978a5ce 100644 --- a/SW.Bus.IntegrationTests/ConnectionTests.cs +++ b/SW.Bus.IntegrationTests/ConnectionTests.cs @@ -1,4 +1,6 @@ using System; +using System.Collections.Generic; +using System.Diagnostics; using System.Linq; using System.Threading.Tasks; using Microsoft.Extensions.DependencyInjection; @@ -130,4 +132,68 @@ public async Task A_host_whose_broker_is_down_at_boot_starts_reports_not_connect await ConcurrencyTests.Publish(harness, "late", 1); Assert.IsTrue(await Broker.Eventually(() => plan.Log("late").Completed.Count == 1), "The message never arrived."); } + + [TestMethod] + public async Task A_publish_the_broker_blocks_fails_after_the_publish_timeout_and_publishing_resumes() + { + var plan = new WorkPlan(); + plan.Types["blocked"] = new ConsumerOptions(); + await using var harness = await BusHarness.StartAsync(_container.GetConnectionString(), + configureBus: o => o.PublishTimeout = TimeSpan.FromSeconds(2), + configureServices: s => s.AddSingleton(plan)); + + await ConcurrencyTests.Publish(harness, "blocked", 1); + Assert.IsTrue(await Broker.Eventually(() => plan.Log("blocked").Completed.Count == 1), "The first publish never arrived."); + + await Broker.BlockPublishers(_container); + try + { + await using var scope = harness.Services.CreateAsyncScope(); + var elapsed = Stopwatch.StartNew(); + var publish = scope.ServiceProvider.GetRequiredService().Publish("blocked", "held"); + var finished = await Task.WhenAny(publish, Task.Delay(TimeSpan.FromSeconds(30))); + elapsed.Stop(); + + Assert.AreSame(publish, finished, "A publish to a broker blocking publishers waited past its timeout."); + Assert.IsTrue(publish.IsFaulted, "A publish the broker never took returned as if it was sent."); + var chain = new List(); + for (Exception? e = publish.Exception; e != null; e = e.InnerException) chain.Add(e); + Assert.IsTrue(chain.OfType().Any(), $"Expected a BusException, got: {publish.Exception}"); + Assert.IsTrue(elapsed.Elapsed < TimeSpan.FromSeconds(10), $"The publish took {elapsed.Elapsed}."); + } + finally + { + await Broker.UnblockPublishers(_container); + } + + Assert.IsTrue(await Broker.Eventually(async () => + { + try + { + await ConcurrencyTests.Publish(harness, "blocked", 1); + return true; + } + catch + { + return false; + } + }, 60), "Publishing never resumed after the alarm cleared."); + } + + [TestMethod] + public async Task The_consumer_connection_is_named_after_the_application() + { + var plan = new WorkPlan(); + plan.Types["named"] = new ConsumerOptions(); + await using var harness = await BusHarness.StartAsync(_container.GetConnectionString(), + configureBus: o => o.ManagementUrl = Broker.ManagementUrl(_container), + configureServices: s => s.AddSingleton(plan)); + + await ConcurrencyTests.Publish(harness, "named", 1); + Assert.IsTrue(await Broker.Eventually(() => plan.Log("named").Completed.Count == 1), "The message never arrived."); + + var names = await Broker.ConnectionNames(_container); + CollectionAssert.Contains(names, "SW.Bus.IntegrationTests Consumer", string.Join(", ", names)); + CollectionAssert.DoesNotContain(names, "SW.Bus Consumer"); + } } diff --git a/SW.Bus.IntegrationTests/DrainAndRetryTests.cs b/SW.Bus.IntegrationTests/DrainAndRetryTests.cs index 8f83a65..6606962 100644 --- a/SW.Bus.IntegrationTests/DrainAndRetryTests.cs +++ b/SW.Bus.IntegrationTests/DrainAndRetryTests.cs @@ -95,7 +95,9 @@ public async Task A_failing_message_is_retried_its_retry_count_then_moved_to_the Assert.IsTrue(headers.ContainsKey("exception1")); Assert.IsTrue(headers.ContainsKey("source-node-id")); Assert.IsFalse(headers.ContainsKey("x-death")); - Assert.IsFalse(headers.ContainsKey(Constants.PublishSequenceNumberHeader)); + // The copy is published with confirms, so the client stamps it with the tracking number of that + // publish (the original's number is dropped before copying). + Assert.IsTrue(headers.ContainsKey(Constants.PublishSequenceNumberHeader)); Assert.AreEqual(DeliveryModes.Persistent, message.BasicProperties.DeliveryMode); } } diff --git a/SW.Bus/BasicPublisher.cs b/SW.Bus/BasicPublisher.cs index 266483b..c75e7a7 100644 --- a/SW.Bus/BasicPublisher.cs +++ b/SW.Bus/BasicPublisher.cs @@ -125,10 +125,30 @@ public async Task Publish(string messageTypeName, byte[] message, string exchang try { - var publishChannel = await GetChannelAsync(); - // With confirms on, this completes when the broker has the message, and throws when the - // broker refuses it or the connection is lost before it answers. - await publishChannel.BasicPublishAsync(exchange, messageTypeName.ToLower(), false, props, message); + using (var timeout = busOptions.PublishTimeout > TimeSpan.Zero + ? new CancellationTokenSource(busOptions.PublishTimeout) + : null) + { + var cancellationToken = timeout?.Token ?? CancellationToken.None; + try + { + var publishChannel = await GetChannelAsync(cancellationToken); + // With confirms on, this completes when the broker has the message, and throws when the + // broker refuses it or the connection is lost before it answers. + await publishChannel.BasicPublishAsync(exchange, messageTypeName.ToLower(), false, props, message, + cancellationToken); + } + catch (OperationCanceledException) when (timeout?.IsCancellationRequested == true) + { + // The channel may hold a half-sent message or a confirm still owed; the next publish of + // this scope opens a fresh one instead. + await DisposeChannelAsync(); + throw new BusException( + $"The broker did not take '{messageTypeName}' within {busOptions.PublishTimeout.TotalSeconds:0.#} s; " + + "it may be blocking publishers (a memory or disk alarm). The broker may or may not have the message."); + } + } + stopwatch.Stop(); activity?.SetTag("messaging.system", "rabbitmq"); activity?.SetTag("messaging.destination.name", exchange); @@ -190,12 +210,12 @@ public async Task Publish(string messageTypeName, byte[] message, string exchang } } - private async Task GetChannelAsync() + private async Task GetChannelAsync(CancellationToken cancellationToken) { var current = channel; if (current is { IsClosed: false }) return current; - await channelLock.WaitAsync(); + await channelLock.WaitAsync(cancellationToken); try { // Closed by the broker (a channel error) or never opened: open one. A channel that is @@ -203,10 +223,10 @@ private async Task GetChannelAsync() if (channel is { IsClosed: false }) return channel; if (channel != null) await DisposeChannelAsync(); - var connection = await busConnection.GetPublisherConnectionAsync(); + var connection = await busConnection.GetPublisherConnectionAsync(cancellationToken); channel = await connection.CreateChannelAsync(new CreateChannelOptions( publisherConfirmationsEnabled: busOptions.PublisherConfirms, - publisherConfirmationTrackingEnabled: busOptions.PublisherConfirms)); + publisherConfirmationTrackingEnabled: busOptions.PublisherConfirms), cancellationToken); return channel; } finally @@ -220,18 +240,32 @@ private async Task DisposeChannelAsync() var closing = channel; channel = null; if (closing == null) return; - try - { - if (closing.IsOpen) await closing.CloseAsync(); - } - catch (Exception ex) - { - Trace.TraceWarning("Failed to close a publisher channel: {0}", ex.Message); - } - finally + await CloseBoundedAsync(closing); + } + + /// + /// Closes a channel, waiting at most . A broker that blocks + /// publishers stops reading the connection, so an unbounded close would hold the scope (an HTTP + /// request, a consumed message) until the alarm clears; past the bound it finishes in the background. + /// + internal static async Task CloseBoundedAsync(IChannel closing) + { + var close = Task.Run(async () => { - await closing.DisposeAsync(); - } + try + { + if (closing.IsOpen) await closing.CloseAsync(); + } + catch (Exception ex) + { + Trace.TraceWarning("Failed to close a publisher channel: {0}", ex.Message); + } + finally + { + await closing.DisposeAsync(); + } + }); + await Task.WhenAny(close, Task.Delay(BusConnection.CloseTimeout)); } public async ValueTask DisposeAsync() => await DisposeChannelAsync(); diff --git a/SW.Bus/BusOptions.cs b/SW.Bus/BusOptions.cs index b4b6624..b110970 100644 --- a/SW.Bus/BusOptions.cs +++ b/SW.Bus/BusOptions.cs @@ -229,6 +229,15 @@ public int MonitoringCacheSeconds /// public bool PublisherConfirms { get; set; } = true; + /// + /// Gets or sets how long a publish may wait for the broker before it fails: to open its channel, to send, + /// and, with , for the confirm. A broker that blocks publishers (a memory or + /// disk alarm) otherwise holds every publishing request until the alarm clears. + /// A publish that times out throws; the broker may or may not have the message. + /// Default: 15 seconds. waits as long as it takes. + /// + public TimeSpan PublishTimeout { get; set; } = TimeSpan.FromSeconds(15); + /// /// Gets or sets how long stopping the bus waits for messages already being processed to finish /// and be acknowledged, after the consumers stop receiving new ones. Messages still running diff --git a/SW.Bus/ConsumerRunner.cs b/SW.Bus/ConsumerRunner.cs index 0ad6e69..b4e3fed 100644 --- a/SW.Bus/ConsumerRunner.cs +++ b/SW.Bus/ConsumerRunner.cs @@ -5,6 +5,7 @@ using System.Reflection; using System.Text; using System.Text.Json; +using System.Threading; using System.Threading.Tasks; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; @@ -24,19 +25,27 @@ internal class ConsumerRunner private readonly BusMetrics metrics; private readonly ILogger logger; + private readonly BusConnection busConnection; + + // Dead-letter copies go out on a confirmed channel of the publisher connection, so a failed + // message is acknowledged only once the broker has its copy. + private readonly SemaphoreSlim deadLetterChannelLock = new(1, 1); + private IChannel deadLetterChannel; public ConsumerRunner( IServiceProvider sp, BusOptions busOptions, IOperationalEventPublisher operationalEventPublisher, BusMetrics metrics, - ILogger logger) + ILogger logger, + BusConnection busConnection) { this.sp = sp; this.busOptions = busOptions; this.operationalEventPublisher = operationalEventPublisher; this.metrics = metrics; this.logger = logger; + this.busConnection = busConnection; } internal async Task Run(BasicDeliverEventArgs ea, ConsumerDefinition consumerDefinition, IChannel model) @@ -92,7 +101,7 @@ internal async Task Run(BasicDeliverEventArgs ea, ConsumerDefinition consumerDef try { - using var scope = sp.CreateScope(); + await using var scope = sp.CreateAsyncScope(); TryBuildBusRequestContext(scope.ServiceProvider,consumerDefinition.MessageTypeName, ea.BasicProperties, remainingRetryCount); var body = ea.Body; @@ -189,11 +198,12 @@ Total retries configured {consumerDefinition.RetryCount}. } else { - await model.BasicAckAsync(ea.DeliveryTag, false); logger.LogError(ex, @$"Failed to process message '{consumerDefinition.MessageTypeName}', in '{consumerDefinition.ServiceType.Name}'. Message {message}, Total retries {consumerDefinition.RetryCount}"); - await PublishBad(model, ea.Body, ea.BasicProperties, busOptions.DeadLetterExchange, consumerDefinition.BadRoutingKey, ex); + if (!await TryPublishBad(model, ea, busOptions.DeadLetterExchange, consumerDefinition.BadRoutingKey, ex)) + return; + await model.BasicAckAsync(ea.DeliveryTag, false); metrics.DeadLetterMoved.Add(1); FireAndForget(new MessageMovedToDeadLetter( DateTime.UtcNow, @@ -251,7 +261,7 @@ internal async Task RunNodeMessage(BasicDeliverEventArgs ea, IChannel model, object svc = null; try { - using var scope = sp.CreateScope(); + await using var scope = sp.CreateAsyncScope(); var body = ea.Body; message = Encoding.UTF8.GetString(body.ToArray()); @@ -301,15 +311,17 @@ Total retries configured {busOptions.ListenRetryCount}. } else { - await model.BasicAckAsync(ea.DeliveryTag, false); logger.LogError(ex, @$"Failed to process message '{listenerDefinition?.MessageTypeName} ?? ?? message', in '{listenerDefinition?.ServiceType.Name ?? "reloading"}'. Message {message}, Total retries {busOptions.ListenRetryCount}"); - await PublishBad(model, ea.Body, ea.BasicProperties,busOptions.NodeDeadLetterExchange, busOptions.NodeBadRoutingKey, ex); - metrics.DeadLetterMoved.Add(1); - await RunOnFail(svc, failMethod, ex, message); + if (await TryPublishBad(model, ea, busOptions.NodeDeadLetterExchange, busOptions.NodeBadRoutingKey, ex)) + { + await model.BasicAckAsync(ea.DeliveryTag, false); + metrics.DeadLetterMoved.Add(1); + await RunOnFail(svc, failMethod, ex, message); + } } FireAndForget(new MessageProcessingFailed( @@ -459,7 +471,55 @@ List ReadPublishedValues(IReadOnlyBasicProperties basicProperties, } - private async Task PublishBad(IChannel model, ReadOnlyMemory body, IReadOnlyBasicProperties messageProps, + /// + /// Publishes the dead-letter copy and reports whether the broker confirmed it. When it did not, the + /// original is rejected instead of acknowledged: it goes to the retry queue and comes back after + /// RetryAfter to be dead-lettered again, rather than being lost. + /// + private async Task TryPublishBad(IChannel model, BasicDeliverEventArgs ea, string exchange, + string routingKey, Exception ex) + { + try + { + await PublishBad(ea.Body, ea.BasicProperties, exchange, routingKey, ex); + return true; + } + catch (Exception badEx) + { + logger.LogError(badEx, + "Could not move a failed message to {RoutingKey}; it goes back to its retry queue instead.", routingKey); + await model.BasicRejectAsync(ea.DeliveryTag, false); + return false; + } + } + + private async Task DeadLetterChannelAsync(CancellationToken cancellationToken) + { + var current = deadLetterChannel; + if (current is { IsClosed: false }) return current; + + await deadLetterChannelLock.WaitAsync(cancellationToken); + try + { + if (deadLetterChannel is { IsClosed: false }) return deadLetterChannel; + if (deadLetterChannel != null) + { + try { await deadLetterChannel.DisposeAsync(); } + catch (Exception e) { logger.LogDebug(e, "Failed to dispose a closed dead-letter channel."); } + } + + var connection = await busConnection.GetPublisherConnectionAsync(cancellationToken); + deadLetterChannel = await connection.CreateChannelAsync(new CreateChannelOptions( + publisherConfirmationsEnabled: true, publisherConfirmationTrackingEnabled: true), cancellationToken); + return deadLetterChannel; + } + finally + { + deadLetterChannelLock.Release(); + } + } + + private async Task PublishBad(ReadOnlyMemory body, IReadOnlyBasicProperties messageProps, string exchange, string routingKey, Exception ex) { const string exception = "exception"; @@ -478,7 +538,25 @@ private async Task PublishBad(IChannel model, ReadOnlyMemory body, IReadOn props.Headers.Add($"{exception}{totalBad}", SerializeException(ex)); props.DeliveryMode = DeliveryModes.Persistent; - await model.BasicPublishAsync(exchange, routingKey, false, props, body); + + using var timeout = busOptions.PublishTimeout > TimeSpan.Zero + ? new CancellationTokenSource(busOptions.PublishTimeout) + : null; + var cancellationToken = timeout?.Token ?? CancellationToken.None; + IChannel channel = null; + try + { + channel = await DeadLetterChannelAsync(cancellationToken); + await channel.BasicPublishAsync(exchange, routingKey, false, props, body, cancellationToken); + } + catch (OperationCanceledException) when (timeout?.IsCancellationRequested == true) + { + // The channel may owe a confirm or hold a half-sent message: the next dead letter opens a fresh one. + if (channel != null && ReferenceEquals(Interlocked.CompareExchange(ref deadLetterChannel, null, channel), channel)) + await BasicPublisher.CloseBoundedAsync(channel); + throw new BusException( + $"The broker did not take the dead-letter copy for '{routingKey}' within {busOptions.PublishTimeout.TotalSeconds:0.#} s."); + } } // some exceptions (thrown ones carry a TargetSite) are not serializable; the message is already diff --git a/SW.Bus/IServiceCollectionExtensions.cs b/SW.Bus/IServiceCollectionExtensions.cs index d8f350c..f124e6b 100644 --- a/SW.Bus/IServiceCollectionExtensions.cs +++ b/SW.Bus/IServiceCollectionExtensions.cs @@ -109,7 +109,9 @@ public static IServiceCollection AddBus(this IServiceCollection services, Action services.AddSingleton(new BusConnectionNames { Publisher = $"{Assembly.GetCallingAssembly().GetName().Name} Publisher", - Consumer = $"{typeof(IServiceCollectionExtensions).Assembly.GetName().Name} Consumer" + // Named after the application, like the publisher, so the broker's connection list tells + // services apart; every service's consumer connection used to be "SW.Bus Consumer". + Consumer = $"{Assembly.GetCallingAssembly().GetName().Name} Consumer" }); services.AddSingleton(); services.AddSingleton(sp => sp.GetRequiredService());