Release v1.5.0 (ADR-0038, ADR-0039)

CHANGELOG.md

@@ -9,6 +9,25 @@ The envelope wire format is versioned separately by `meta.schema_version`
 
 ## [Unreleased]
 
+## [1.5.0] - 2026-11-03
+
+### Added
+
+- **Dead-letter observer (ADR-0039 section 5.1).** New `SqsConsumerOptions.DeadLetterObserver` (`BabelQueue.IDeadLetterObserver`
+  from `BabelQueue.Core`, e.g. `MetricsDeadLetterObserver`, which counts `babelqueue.dlq.messages`). It is called for each
+  record the consumer **wrote** to `DeadLetterQueueUrl`, after the send is confirmed, at-least-once (if the `DeleteMessage` that
+  follows fails the message is redelivered and written again, so the observer is called again; a FIFO DLQ deduplicates the
+  record but not the call): an unknown URN under the `dead_letter`
+  strategy (`reason` `unknown_urn`, `Queue` = `meta.queue`), a poison body (`poison`) and a raw schema-version-gated body
+  (empty `reason`). `DeadLetterQueue` is the DLQ's name. It is not called when the send fails, when the message is dropped
+  (`delete`), when no DLQ URL is set, or for exhausted retries, which stay with the queue's native `RedrivePolicy`. An exception
+  the observer throws is caught and logged and never changes the settle outcome. `null` (the default) changes nothing.
+  Needs a `BabelQueue.Core` release that ships `IDeadLetterObserver`; no wire change (`schema_version: 1`).
+
+### Changed
+
+- `BabelQueue.Core` dependency raised from `1.13.0` to `1.14.0` (`IDeadLetterObserver`).
+
 ## [1.4.0] - 2026-10-26
 
 Implements ADR-0037 decisions 1 and 2. The envelope stays at `schema_version: 1`; every change is a consumer-side settle

src/BabelQueue.Sqs/BabelQueue.Sqs.csproj

@@ -16,7 +16,7 @@
 
     <!-- NuGet package metadata -->
     <PackageId>BabelQueue.Sqs</PackageId>
-    <Version>1.4.0</Version>
+    <Version>1.5.0</Version>
     <Authors>Muhammet Şafak</Authors>
     <Company>BabelQueue</Company>
     <Description>Amazon SQS transport for BabelQueue — a canonical-envelope publisher and a URN-routed consumer over the AWS SDK for .NET, on the framework-agnostic core.</Description>
@@ -33,7 +33,7 @@
 
   <ItemGroup>
     <PackageReference Include="AWSSDK.SQS" Version="4.0.100.15" />
-    <PackageReference Include="BabelQueue.Core" Version="1.13.0" />
+    <PackageReference Include="BabelQueue.Core" Version="1.14.0" />
   </ItemGroup>
 
   <ItemGroup>

src/BabelQueue.Sqs/SqsConsumer.cs

@@ -139,7 +139,7 @@ private async Task HandleAsync(Message message, CancellationToken cancellationTo
                 $"Rejected an SQS message: unsupported {SchemaVersionAttribute} attribute '{declaredVersion}' "
                 + $"(supported: {EnvelopeCodec.SchemaVersion}); body not decoded.");
             var empty = new Envelope(null, null, null, null, 0, null);
-            await SettleTerminalAsync(message, () => message.Body ?? string.Empty, empty, rejected, cancellationToken)
+            await SettleTerminalAsync(message, () => message.Body ?? string.Empty, string.Empty, empty, rejected, cancellationToken)
                 .ConfigureAwait(false);
             return;
         }
@@ -161,6 +161,7 @@ await SettleTerminalAsync(
                     DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(),
                     SqsPublisher.QueueName(_queueUrl),
                     envelope.Attempts)),
+                "poison",
                 envelope,
                 cause,
                 cancellationToken).ConfigureAwait(false);
@@ -215,10 +216,16 @@ await GuardAsync(
     /// message deleted only after the write succeeded (a failed write leaves it undeleted: unsettled, reported); with no
     /// DLQ it is handed to the native mechanism -- not deleted, not released -- so visibility expiry redelivers it and the
     /// queue's <c>RedrivePolicy</c> applies. The rejection is reported after the settle.
+    /// <para>
+    /// After the DLQ write is confirmed the optional <see cref="SqsConsumerOptions.DeadLetterObserver"/> is told with
+    /// <paramref name="reason"/> (empty for a raw version-gated body, <c>poison</c> for a poison record). The native hand-off
+    /// and a drop are not SDK dead-letters and are not reported (ADR-0039 section 5.1).
+    /// </para>
     /// </summary>
     private async Task SettleTerminalAsync(
         Message message,
         Func<string> record,
+        string reason,
         Envelope envelope,
         BabelQueueException rejected,
         CancellationToken cancellationToken)
@@ -233,7 +240,7 @@ await GuardAsync(() => DeleteAsync(message, cancellationToken), envelope, messag
             await GuardAsync(
                 async () =>
                 {
-                    await WriteDeadLetterRecordAsync(message, record(), cancellationToken).ConfigureAwait(false);
+                    await WriteDeadLetterRecordAsync(message, record(), reason, cancellationToken).ConfigureAwait(false);
                     await DeleteAsync(message, cancellationToken).ConfigureAwait(false);
                 },
                 envelope,
@@ -251,8 +258,12 @@ await GuardAsync(
         _options.OnError?.Invoke(rejected, envelope, message);
     }
 
-    /// <summary>Writes <paramref name="body"/> as the body of a new message on the DLQ (FIFO: group = queue name, dedupe = SQS message id).</summary>
-    private async Task WriteDeadLetterRecordAsync(Message message, string body, CancellationToken cancellationToken)
+    /// <summary>
+    /// Writes <paramref name="body"/> as the body of a new message on the DLQ (FIFO: group = queue name, dedupe = SQS message id),
+    /// then reports it to the dead-letter observer. A failed send throws before the observer is called. If the delete that
+    /// follows fails the message is redelivered and written again, so the observer is called again (at-least-once).
+    /// </summary>
+    private async Task WriteDeadLetterRecordAsync(Message message, string body, string reason, CancellationToken cancellationToken)
     {
         var request = new SendMessageRequest { QueueUrl = _options.DeadLetterQueueUrl, MessageBody = body };
         if (_options.DeadLetterQueueUrl!.EndsWith(".fifo", StringComparison.Ordinal))
@@ -262,6 +273,11 @@ private async Task WriteDeadLetterRecordAsync(Message message, string body, Canc
         }
 
         await _client.SendMessageAsync(request, cancellationToken).ConfigureAwait(false);
+        if (_options.DeadLetterObserver is { } observer)
+        {
+            observer.NotifySafely(new DeadLetterEvent(
+                SqsPublisher.QueueName(_queueUrl), SqsPublisher.QueueName(_options.DeadLetterQueueUrl), reason));
+        }
     }
 
     private static readonly JsonWriterOptions PoisonWriterOptions = new()
@@ -404,6 +420,35 @@ private async Task DeadLetterAsync(Envelope envelope, CancellationToken cancella
         }
 
         await _client.SendMessageAsync(request, cancellationToken).ConfigureAwait(false);
+        if (_options.DeadLetterObserver is { } observer)
+        {
+            observer.NotifySafely(new DeadLetterEvent(
+                IsBlankQueue(envelope.Meta?.Queue) ? SqsPublisher.QueueName(_queueUrl) : envelope.Meta!.Queue!,
+                SqsPublisher.QueueName(_options.DeadLetterQueueUrl),
+                "unknown_urn"));
+        }
+    }
+
+    /// <summary>
+    /// Blank in the shared cross-SDK sense (ADR-0031): empty or only ASCII whitespace (space, <c>\t</c>, <c>\n</c>, U+000B,
+    /// <c>\f</c>, <c>\r</c>) -- not <see cref="string.IsNullOrWhiteSpace"/>, which also counts Unicode spaces.
+    /// </summary>
+    private static bool IsBlankQueue(string? value)
+    {
+        if (value is null)
+        {
+            return true;
+        }
+
+        foreach (var c in value)
+        {
+            if (c != ' ' && (c < '\t' || c > '\r'))
+            {
+                return false;
+            }
+        }
+
+        return true;
     }
 
     /// <summary>

src/BabelQueue.Sqs/SqsConsumerOptions.cs

@@ -102,4 +102,19 @@ public int RetryBackoffSeconds
     /// <c>MessageDeduplicationId</c> = the SQS message id (not <c>meta.queue</c> / <c>meta.id</c>: there is no usable envelope).
     /// </summary>
     public string? DeadLetterQueueUrl { get; set; }
+
+    /// <summary>
+    /// Optional hook called for each dead-letter record this consumer <b>wrote</b> to <see cref="DeadLetterQueueUrl"/>,
+    /// after the send is confirmed (ADR-0039 section 5.1), <b>at-least-once</b>: if the <c>DeleteMessage</c> that follows
+    /// fails, the message is redelivered and written again, so the observer is called again. A FIFO DLQ deduplicates the
+    /// record (same <c>MessageDeduplicationId</c>) but not the call, so the count can exceed the records in the DLQ.
+    /// Reported cases: an unknown URN under the <c>dead_letter</c> strategy
+    /// (<c>reason</c> <c>unknown_urn</c>), a poison body (<c>poison</c>) and a raw schema-version-gated body (empty
+    /// <c>reason</c>). It is not called when the send fails, when the message is dropped (<c>delete</c>) or left to the
+    /// queue's native <c>RedrivePolicy</c> (also what exhausted retries use), or when no DLQ URL is configured.
+    /// An exception it throws is caught and logged by <see cref="DeadLetterObservers.NotifySafely"/> and never changes
+    /// the settle outcome. <c>null</c> (the default) changes nothing; <see cref="Tracing.MetricsDeadLetterObserver"/>
+    /// counts <c>babelqueue.dlq.messages</c>.
+    /// </summary>
+    public IDeadLetterObserver? DeadLetterObserver { get; set; }
 }

tests/BabelQueue.Sqs.Tests/SqsConsumerTests.cs

@@ -943,4 +943,220 @@ public void UnrecognisedUnknownUrnStrategyIsRejected(string strategy)
         Assert.Throws<ArgumentException>(
             () => new SqsConsumer(new Mock<IAmazonSQS>().Object, Url, new Dictionary<string, BabelHandler>(), options));
     }
+
+    // ---- ADR-0039 section 5.1: dead-letter observer ----
+
+    private sealed class RecordingObserver : IDeadLetterObserver
+    {
+        public List<DeadLetterEvent> Events { get; } = new();
+
+        public bool Throw { get; init; }
+
+        public void OnDeadLetter(DeadLetterEvent deadLetterEvent)
+        {
+            Events.Add(deadLetterEvent);
+            if (Throw)
+            {
+                throw new InvalidOperationException("observer failure");
+            }
+        }
+    }
+
+    [Fact]
+    public async Task ObserverIsCalledOnceForAVersionGatedRawBodyWithAnEmptyReason()
+    {
+        var observer = new RecordingObserver();
+        var (mock, _) = MockWithDlq(WithSchemaVersion(Envelope(), "2"));
+        var options = new SqsConsumerOptions { DeadLetterQueueUrl = Dlq, DeadLetterObserver = observer };
+
+        await new SqsConsumer(mock.Object, Url, new Dictionary<string, BabelHandler>(), options).PollAsync();
+
+        Assert.Equal(new DeadLetterEvent("orders", "orders.dlq", string.Empty), Assert.Single(observer.Events));
+    }
+
+    [Fact]
+    public async Task ObserverIsCalledOnceForAPoisonBodyWithTheCanonicalReason()
+    {
+        var observer = new RecordingObserver();
+        var (mock, _) = MockWithDlq(Seed("{\"not\":\"an envelope\"}", 1));
+        var options = new SqsConsumerOptions { DeadLetterQueueUrl = Dlq, DeadLetterObserver = observer };
+
+        await new SqsConsumer(mock.Object, Url, new Dictionary<string, BabelHandler>(), options).PollAsync();
+
+        Assert.Equal(new DeadLetterEvent("orders", "orders.dlq", "poison"), Assert.Single(observer.Events));
+    }
+
+    [Fact]
+    public async Task ObserverIsCalledOnceForAnUnknownUrnDeadLetterWithTheEnvelopeQueue()
+    {
+        var observer = new RecordingObserver();
+        var (mock, _) = MockWithDlq(Seed(Envelope(), 1));
+        var options = new SqsConsumerOptions
+        {
+            UnknownUrnStrategy = UnknownUrnStrategy.DeadLetter,
+            DeadLetterQueueUrl = Dlq,
+            DeadLetterObserver = observer,
+        };
+
+        await new SqsConsumer(mock.Object, Url, new Dictionary<string, BabelHandler>(), options).PollAsync();
+
+        Assert.Equal(new DeadLetterEvent("orders", "orders.dlq", "unknown_urn"), Assert.Single(observer.Events));
+    }
+
+    private static string EnvelopeFor(string queue)
+        => Envelope().Replace("\"queue\":\"orders\"", "\"queue\":\"" + queue + "\"", StringComparison.Ordinal);
+
+    [Fact]
+    public async Task UnknownUrnEventCarriesTheEnvelopeQueueNotTheConsumerQueue()
+    {
+        var body = EnvelopeFor("billing");
+        Assert.Contains("\"queue\":\"billing\"", body, StringComparison.Ordinal);
+        var observer = new RecordingObserver();
+        var (mock, _) = MockWithDlq(Seed(body, 1));
+        var options = new SqsConsumerOptions
+        {
+            UnknownUrnStrategy = UnknownUrnStrategy.DeadLetter,
+            DeadLetterQueueUrl = Dlq,
+            DeadLetterObserver = observer,
+        };
+
+        await new SqsConsumer(mock.Object, Url, new Dictionary<string, BabelHandler>(), options).PollAsync();
+
+        Assert.Equal(new DeadLetterEvent("billing", "orders.dlq", "unknown_urn"), Assert.Single(observer.Events));
+    }
+
+    [Theory]
+    [InlineData("")]
+    [InlineData("  ")]
+    [InlineData("\\t\\n")]
+    public async Task UnknownUrnEventFallsBackToTheConsumerQueueWhenMetaQueueIsBlank(string blank)
+    {
+        var body = EnvelopeFor(blank);
+        var observer = new RecordingObserver();
+        var (mock, _) = MockWithDlq(Seed(body, 1));
+        var options = new SqsConsumerOptions
+        {
+            UnknownUrnStrategy = UnknownUrnStrategy.DeadLetter,
+            DeadLetterQueueUrl = Dlq,
+            DeadLetterObserver = observer,
+        };
+
+        await new SqsConsumer(mock.Object, Url, new Dictionary<string, BabelHandler>(), options).PollAsync();
+
+        Assert.Equal("orders", Assert.Single(observer.Events).Queue);
+    }
+
+    [Fact]
+    public async Task PoisonAndGatedEventsCarryTheConsumerQueueEvenWhenTheBodyNamesAnother()
+    {
+        var observer = new RecordingObserver();
+        var options = new SqsConsumerOptions { DeadLetterQueueUrl = Dlq, DeadLetterObserver = observer };
+
+        // Poison: carries meta.queue "billing" but has no job/urn identity.
+        var (poison, _) = MockWithDlq(Seed("{\"meta\":{\"queue\":\"billing\"}}", 1));
+        await new SqsConsumer(poison.Object, Url, new Dictionary<string, BabelHandler>(), options).PollAsync();
+
+        // Version-gated: a valid envelope for "billing" behind an unsupported bq-schema-version attribute.
+        var (gated, _) = MockWithDlq(WithSchemaVersion(EnvelopeFor("billing"), "2"));
+        await new SqsConsumer(gated.Object, Url, new Dictionary<string, BabelHandler>(), options).PollAsync();
+
+        Assert.Equal(
+            new[] { new DeadLetterEvent("orders", "orders.dlq", "poison"), new DeadLetterEvent("orders", "orders.dlq", string.Empty) },
+            observer.Events);
+    }
+
+    [Fact]
+    public async Task ObserverIsNotCalledWhenTheDlqWriteFails()
+    {
+        foreach (var message in new[] { WithSchemaVersion(Envelope(), "2"), Seed("{\"not\":\"an envelope\"}", 1), Seed(Envelope(), 1) })
+        {
+            var observer = new RecordingObserver();
+            var (mock, _) = MockWithDlq(message, new AmazonSQSException("dlq down"));
+            var options = new SqsConsumerOptions
+            {
+                UnknownUrnStrategy = UnknownUrnStrategy.DeadLetter,
+                DeadLetterQueueUrl = Dlq,
+                DeadLetterObserver = observer,
+                OnError = (_, _, _) => { },
+            };
+
+            await new SqsConsumer(mock.Object, Url, new Dictionary<string, BabelHandler>(), options).PollAsync();
+
+            Assert.Empty(observer.Events);
+            mock.Verify(c => c.DeleteMessageAsync(It.IsAny<DeleteMessageRequest>(), It.IsAny<CancellationToken>()), Times.Never);
+        }
+    }
+
+    [Fact]
+    public async Task ObserverIsNotCalledWhenNothingIsWrittenToTheDlq()
+    {
+        var observer = new RecordingObserver();
+
+        // No DLQ URL: poison and gated are left to the native RedrivePolicy; the dead_letter strategy degrades to delete.
+        foreach (var message in new[] { WithSchemaVersion(Envelope(), "2"), Seed("{\"not\":\"an envelope\"}", 1), Seed(Envelope(), 1) })
+        {
+            var (mock, sent) = MockWithDlq(message);
+            var options = new SqsConsumerOptions { UnknownUrnStrategy = UnknownUrnStrategy.DeadLetter, DeadLetterObserver = observer };
+            await new SqsConsumer(mock.Object, Url, new Dictionary<string, BabelHandler>(), options).PollAsync();
+            Assert.Empty(sent);
+        }
+
+        // Delete strategy with a DLQ configured: dropped, not dead-lettered.
+        foreach (var message in new[] { WithSchemaVersion(Envelope(), "2"), Seed("{\"not\":\"an envelope\"}", 1) })
+        {
+            var (mock, sent) = MockWithDlq(message);
+            var options = new SqsConsumerOptions
+            {
+                UnknownUrnStrategy = UnknownUrnStrategy.Delete,
+                DeadLetterQueueUrl = Dlq,
+                DeadLetterObserver = observer,
+            };
+            await new SqsConsumer(mock.Object, Url, new Dictionary<string, BabelHandler>(), options).PollAsync();
+            Assert.Empty(sent);
+        }
+
+        Assert.Empty(observer.Events);
+    }
+
+    [Fact]
+    public async Task ThrowingObserverDoesNotChangeTheSettleOutcome()
+    {
+        var observer = new RecordingObserver { Throw = true };
+        var (mock, sent) = MockWithDlq(Seed("{\"not\":\"an envelope\"}", 1));
+        var reported = new List<Exception>();
+        var options = new SqsConsumerOptions
+        {
+            DeadLetterQueueUrl = Dlq,
+            DeadLetterObserver = observer,
+            OnError = (e, _, _) => reported.Add(e),
+        };
+
+        await new SqsConsumer(mock.Object, Url, new Dictionary<string, BabelHandler>(), options).PollAsync();
+
+        Assert.Single(observer.Events);
+        Assert.Single(sent);
+        mock.Verify(c => c.DeleteMessageAsync(It.Is<DeleteMessageRequest>(r => r.ReceiptHandle == "rh-1"), It.IsAny<CancellationToken>()), Times.Once);
+        Assert.IsType<BabelQueueException>(Assert.Single(reported));
+    }
+
+    [Fact]
+    public async Task ThrowingObserverOnAnUnknownUrnDeadLetterStillDeletesTheMessage()
+    {
+        var observer = new RecordingObserver { Throw = true };
+        var (mock, sent) = MockWithDlq(Seed(Envelope(), 1));
+        var errors = 0;
+        var options = new SqsConsumerOptions
+        {
+            UnknownUrnStrategy = UnknownUrnStrategy.DeadLetter,
+            DeadLetterQueueUrl = Dlq,
+            DeadLetterObserver = observer,
+            OnError = (_, _, _) => errors++,
+        };
+
+        await new SqsConsumer(mock.Object, Url, new Dictionary<string, BabelHandler>(), options).PollAsync();
+
+        Assert.Single(sent);
+        Assert.Equal(0, errors);
+        mock.Verify(c => c.DeleteMessageAsync(It.IsAny<DeleteMessageRequest>(), It.IsAny<CancellationToken>()), Times.Once);
+    }
 }