Publish timeout, confirmed dead-letter before ack, tests in CI

.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"

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
 

SW.Bus.IntegrationTests/Broker.cs

@@ -33,6 +33,29 @@ public static async Task StartApp(RabbitMqContainer container)
         await Exec(container, "rabbitmqctl", "await_startup");
     }
 
+    /// <summary>
+    /// 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.
+    /// </summary>
+    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");
+
+    /// <summary>The client-provided names of the broker's connections, read through the management API.</summary>
+    public static async Task<string[]> 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);

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<IPublish>().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<Exception>();
+            for (Exception? e = publish.Exception; e != null; e = e.InnerException) chain.Add(e);
+            Assert.IsTrue(chain.OfType<BusException>().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");
+    }
 }

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);
     }
 }

SW.Bus/BasicPublisher.cs

@@ -55,7 +55,7 @@
     }
 
     /// <param name="valuesHeader">Written by <see cref="PublishedValues.ToHeader"/>, which also checks its size.</param>
     public async Task Publish(string messageTypeName, byte[] message, string exchange, byte? priority = null,
         string valuesHeader = null)
     {
         var activity = BusDiagnostics.ActivitySource.StartActivity($"bus.publish {messageTypeName}", ActivityKind.Producer);
@@ -125,10 +125,30 @@
 
         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,23 +210,23 @@
         }
     }
 
-    private async Task<IChannel> GetChannelAsync()
+    private async Task<IChannel> 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
             // recovering with its connection is not closed and is kept.
             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 @@
         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);
+    }
+
+    /// <summary>
+    /// Closes a channel, waiting at most <see cref="BusConnection.CloseTimeout"/>. 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.
+    /// </summary>
+    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();

SW.Bus/BusOptions.cs

@@ -229,6 +229,15 @@ public int MonitoringCacheSeconds
         /// </summary>
         public bool PublisherConfirms { get; set; } = true;
 
+        /// <summary>
+        /// Gets or sets how long a publish may wait for the broker before it fails: to open its channel, to send,
+        /// and, with <see cref="PublisherConfirms"/>, 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. <see cref="TimeSpan.Zero"/> waits as long as it takes.
+        /// </summary>
+        public TimeSpan PublishTimeout { get; set; } = TimeSpan.FromSeconds(15);
+
         /// <summary>
         /// 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