BusOptions.Headers: headers stamped on every published message

README.md

@@ -157,6 +157,14 @@ They travel in one `request-context-values` header and reach the consumer as `Se
 
 > **Never put a secret in a value.** Headers are readable by anything that can read the queue, and a message that fails for good keeps them in its dead-letter queue, which the error-queue API returns.
 
+Headers on every message, such as which deployment published it (`BusOptions.Headers`):
+
+```csharp
+services.AddBus(o => o.Headers["deployment"] = configuration["Deployment:Id"]);
+```
+
+A header the bus sets itself (`contract`, trace, user, `Id`...) keeps its value when a name clashes, and an empty value is left out. The same caution about secrets applies.
+
 ---
 
 ## Consuming Messages

SW.Bus.IntegrationTests/StaticHeaderTests.cs

@@ -0,0 +1,59 @@
+using System;
+using System.Text;
+using System.Threading.Tasks;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using RabbitMQ.Client;
+using SW.PrimitiveTypes;
+using Testcontainers.RabbitMq;
+
+namespace SW.Bus.IntegrationTests;
+
+public class HeaderedPlaced { public string Number { get; set; } = "SO-H"; }
+
+/// <summary>BusOptions.Headers go on every published message, without replacing a header the bus sets itself.</summary>
+[TestClass]
+public class StaticHeaderTests
+{
+    private static RabbitMqContainer _container = null!;
+
+    [ClassInitialize]
+    public static async Task ClassInitialize(TestContext _)
+    {
+        _container = Broker.Create();
+        await _container.StartAsync();
+    }
+
+    [ClassCleanup]
+    public static async Task ClassCleanup() => await _container.DisposeAsync();
+
+    [TestMethod]
+    public async Task Every_published_message_carries_the_configured_headers()
+    {
+        await using var harness = await BusHarness.StartAsync(_container.GetConnectionString(), o =>
+        {
+            o.Headers["deployment"] = "acme";
+            o.Headers["Id"] = "not-the-message-id";
+            o.Headers["empty"] = "";
+        });
+        await using var conn = await new ConnectionFactory { Uri = new Uri(_container.GetConnectionString()) }.CreateConnectionAsync();
+        await using var channel = await conn.CreateChannelAsync();
+        var queue = (await channel.QueueDeclareAsync()).QueueName;
+        await channel.QueueBindAsync(queue, harness.Options.ProcessExchange, nameof(HeaderedPlaced).ToLower());
+
+        await using (var scope = harness.Services.CreateAsyncScope())
+            await scope.ServiceProvider.GetRequiredService<IPublish>().Publish(new HeaderedPlaced());
+
+        BasicGetResult message = null;
+        for (var attempt = 0; attempt < 50 && message == null; attempt++)
+        {
+            message = await channel.BasicGetAsync(queue, autoAck: true);
+            if (message == null) await Task.Delay(100);
+        }
+        Assert.IsNotNull(message, "the message never arrived");
+        var headers = message.BasicProperties.Headers!;
+        Assert.AreEqual("acme", Encoding.UTF8.GetString((byte[])headers["deployment"]!));
+        Assert.AreEqual(message.BasicProperties.MessageId, Encoding.UTF8.GetString((byte[])headers["Id"]!), "the bus's own Id header wins");
+        Assert.IsFalse(headers.ContainsKey("empty"), "an empty value is left out");
+    }
+}

SW.Bus/BasicPublisher.cs

@@ -77,7 +77,7 @@
 
     /// <param name="valuesHeader">Written by <see cref="PublishedValues.ToHeader"/>, which also checks its size.</param>
     /// <param name="contract">The body's contract, "key@major" (<see cref="BusOptions.ContractHeaderName"/>).</param>
     public async Task Publish(string messageTypeName, byte[] message, string exchange, byte? priority = null,
         string valuesHeader = null, string contract = null)
     {
         var activity = BusDiagnostics.ActivitySource.StartActivity($"bus.publish {messageTypeName}", ActivityKind.Producer);
@@ -129,6 +129,11 @@
         props.Headers.Add(BusOptions.SourceNodeIdHeaderName,busOptions.NodeId);
         props.Headers.Add("Id", props.MessageId);
 
+        // Last, so a header the bus sets itself is never replaced (and Add above can't collide).
+        foreach (var (name, value) in busOptions.Headers)
+            if (!string.IsNullOrEmpty(value) && !props.Headers.Keys.Any(k => string.Equals(k, name, StringComparison.OrdinalIgnoreCase)))
+                props.Headers[name] = value;
+
         metrics.PublishStarted.Add(1);
         FireAndForget(new PublishStarted(
             DateTime.UtcNow,

SW.Bus/BusOptions.cs

@@ -194,6 +194,13 @@ public int MonitoringCacheSeconds
         /// </summary>
         public InvalidUserTokenBehavior OnInvalidUserToken { get; set; } = InvalidUserTokenBehavior.Fail;
 
+        /// <summary>
+        /// Headers stamped on every published message, such as which deployment published it. A header the bus
+        /// sets itself (contract, trace, user, correlation, Id...) keeps its value when a name here clashes with it.
+        /// Names are compared case-insensitively. Empty by default.
+        /// </summary>
+        public IDictionary<string, string> Headers { get; } = new Dictionary<string, string>(StringComparer.OrdinalIgnoreCase);
+
         /// <summary>
         /// Gets or sets the application name used in queue and exchange naming.
         /// When set, queues are named: {Exchange}.{ApplicationName}.{ConsumerName}.{MessageName}