From b6974e1ffee942a1ec9b7ea9eed800a248db011f Mon Sep 17 00:00:00 2001 From: brenpike Date: Fri, 25 Sep 2026 15:28:39 -0600 Subject: [PATCH 01/11] fix(messagebrokers): gate only the message the delivery admitted (#534) --- .../Reliability/Inbox/InboxBehavior.cs | 14 ++- .../Reliability/Inbox/InboxDeliveryEntry.cs | 26 +++++ .../Inbox/UsingInboxBehavior/WhenHandling.cs | 101 ++++++++++++++---- 3 files changed, 122 insertions(+), 19 deletions(-) create mode 100644 src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/Inbox/InboxDeliveryEntry.cs diff --git a/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/Inbox/InboxBehavior.cs b/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/Inbox/InboxBehavior.cs index b12c2afc..e357444b 100644 --- a/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/Inbox/InboxBehavior.cs +++ b/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/Inbox/InboxBehavior.cs @@ -22,7 +22,19 @@ public InboxBehavior(IBrokeredMessageInbox brokeredMessageInbox, ILogger)}"); - if (messageHandlerContext is IMessageBrokerContext messageBrokerContext) + + // INVARIANT: only the Delivery Entry - the message this delivery admitted first - is received via the + // inbox; every other command dispatched on the same broker context (nested in-process dispatch + // inherits it) goes straight to next(), because the entry's handler already holds the delivery's + // message id (ADR-0041). Pinned by WhenHandling: restoring the bare `is IMessageBrokerContext` gate + // reddens MustInvokeTheNestedHandlerWhenAGatedHandlersOwnDispatchReEntersTheBehavior, + // MustReceiveViaInboxOnlyTheMessageTheDeliveryAdmitted and + // MustInvokeTheHandlerOfASiblingDispatchedAfterTheAdmittedMessageCompleted; binding on a first-call flag + // instead of the instance reddens MustReceiveViaInboxAgainWhenTheAdmittedMessageIsRedispatchedAfterItsHandlerFailed. + // A handler that re-dispatches the received INSTANCE itself is still gated, and so skipped while its + // own receipt holds the id; no test pins that. + if (messageHandlerContext is IMessageBrokerContext messageBrokerContext + && messageBrokerContext.Container.GetOrNew().Admits(message)) { return _brokeredMessageInbox.ReceiveViaInbox(message, messageBrokerContext, () => next()); } diff --git a/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/Inbox/InboxDeliveryEntry.cs b/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/Inbox/InboxDeliveryEntry.cs new file mode 100644 index 00000000..7d05fb37 --- /dev/null +++ b/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/Inbox/InboxDeliveryEntry.cs @@ -0,0 +1,26 @@ +namespace Chatter.MessageBrokers.Reliability.Inbox +{ + /// + /// The Delivery Entry: the message a delivery admitted into the command pipeline first. It lives in the + /// delivery's , so it is scoped to one + /// and therefore to one delivery. + /// + internal sealed class InboxDeliveryEntry + { + // INVARIANT: no lock. The entry lives in the delivery's ContextContainer, which is unsynchronized and must + // never be used from two threads at once (ADR-0011), and a delivery's nested dispatches are awaited inside + // its handler, so the first Admits call has bound the entry before any other call on this context runs. + // No test pins it. + private object _admittedMessage; + + /// + /// Binds the entry to on the first call; thereafter reports whether + /// is that same instance. + /// + public bool Admits(object message) + { + _admittedMessage ??= message; + return ReferenceEquals(_admittedMessage, message); + } + } +} diff --git a/src/Chatter.MessageBrokers/tests/Reliability/Inbox/UsingInboxBehavior/WhenHandling.cs b/src/Chatter.MessageBrokers/tests/Reliability/Inbox/UsingInboxBehavior/WhenHandling.cs index 6efe6295..385d0450 100644 --- a/src/Chatter.MessageBrokers/tests/Reliability/Inbox/UsingInboxBehavior/WhenHandling.cs +++ b/src/Chatter.MessageBrokers/tests/Reliability/Inbox/UsingInboxBehavior/WhenHandling.cs @@ -44,7 +44,7 @@ public void MustThrowArgumentNullExceptionWhenLoggerIsNull() public async Task MustReceiveViaInboxWhenContextIsMessageBrokerContext() { var message = new FakeCommand(); - var brokerContext = new Mock(); + var brokerContext = CreateBrokerContext(); await _sut.Handle(message, brokerContext.Object, () => Task.CompletedTask); @@ -55,7 +55,7 @@ public async Task MustReceiveViaInboxWhenContextIsMessageBrokerContext() public async Task MustWrapNextInTheReceiverDelegate() { var invoked = false; - var brokerContext = new Mock(); + var brokerContext = CreateBrokerContext(); _inbox.Setup(i => i.ReceiveViaInbox(It.IsAny(), It.IsAny(), It.IsAny>())) .Returns>((_, __, messageReceiver) => messageReceiver()); @@ -80,28 +80,93 @@ public async Task MustInvokeNextAndSkipInboxWhenContextIsNotMessageBrokerContext public async Task MustNotInvokeNextOnDuplicateMessageId() { // A real InMemoryBrokeredMessageInbox proves the behavior dedupes already-processed - // message ids end to end rather than asserting only on a mock interaction. - var bodyConverter = new Mock(); - bodyConverter.SetupGet(c => c.ContentType).Returns("application/json"); - var inbox = new InMemoryBrokeredMessageInbox(new Mock>().Object, - new ReliabilityOptions - { - InMemoryInboxDeduplicationWindowInMinutes = 60, - InMemoryInboxMaxEntries = 200000 - }); - var sut = new InboxBehavior(inbox, _logger); - - var inbound = new InboundBrokeredMessage("id-1", new byte[] { 1 }, new Dictionary(), "receiver-path", bodyConverter.Object); - var brokerContext = new Mock(); - brokerContext.SetupGet(c => c.BrokeredMessage).Returns(inbound); + // message ids end to end rather than asserting only on a mock interaction. A redelivery is a + // SECOND context carrying the same message id, because every receiver builds a fresh + // MessageBrokerContext per delivery. + var sut = new InboxBehavior(CreateInMemoryInbox(), _logger); var firstInvoked = false; - await sut.Handle(new FakeCommand(), brokerContext.Object, () => { firstInvoked = true; return Task.CompletedTask; }); + await sut.Handle(new FakeCommand(), CreateBrokerContext("id-1").Object, () => { firstInvoked = true; return Task.CompletedTask; }); firstInvoked.Should().BeTrue(); var secondInvoked = false; - await sut.Handle(new FakeCommand(), brokerContext.Object, () => { secondInvoked = true; return Task.CompletedTask; }); + await sut.Handle(new FakeCommand(), CreateBrokerContext("id-1").Object, () => { secondInvoked = true; return Task.CompletedTask; }); secondInvoked.Should().BeFalse(); } + + [Fact] + public async Task MustInvokeTheNestedHandlerWhenAGatedHandlersOwnDispatchReEntersTheBehavior() + { + var sut = new InboxBehavior(CreateInMemoryInbox(), _logger); + var brokerContext = CreateBrokerContext("id-1"); + var nestedInvoked = false; + + await sut.Handle(new FakeCommand(), brokerContext.Object, + () => sut.Handle(new FakeCommand(), brokerContext.Object, () => { nestedInvoked = true; return Task.CompletedTask; })); + + nestedInvoked.Should().BeTrue(); + } + + [Fact] + public async Task MustReceiveViaInboxOnlyTheMessageTheDeliveryAdmitted() + { + var admitted = new FakeCommand(); + var brokerContext = CreateBrokerContext(); + _inbox.Setup(i => i.ReceiveViaInbox(It.IsAny(), It.IsAny(), It.IsAny>())) + .Returns>((_, __, messageReceiver) => messageReceiver()); + + await _sut.Handle(admitted, brokerContext.Object, + () => _sut.Handle(new FakeCommand(), brokerContext.Object, () => Task.CompletedTask)); + + _inbox.Verify(i => i.ReceiveViaInbox(It.IsAny(), It.IsAny(), It.IsAny>()), Times.Once); + _inbox.Verify(i => i.ReceiveViaInbox(admitted, brokerContext.Object, It.IsAny>()), Times.Once); + } + + [Fact] + public async Task MustInvokeTheHandlerOfASiblingDispatchedAfterTheAdmittedMessageCompleted() + { + var sut = new InboxBehavior(CreateInMemoryInbox(), _logger); + var brokerContext = CreateBrokerContext("id-1"); + await sut.Handle(new FakeCommand(), brokerContext.Object, () => Task.CompletedTask); + + var siblingInvoked = false; + await sut.Handle(new FakeCommand(), brokerContext.Object, () => { siblingInvoked = true; return Task.CompletedTask; }); + + siblingInvoked.Should().BeTrue(); + } + + [Fact] + public async Task MustReceiveViaInboxAgainWhenTheAdmittedMessageIsRedispatchedAfterItsHandlerFailed() + { + var admitted = new FakeCommand(); + var brokerContext = CreateBrokerContext(); + _inbox.Setup(i => i.ReceiveViaInbox(It.IsAny(), It.IsAny(), It.IsAny>())) + .Returns>((_, __, messageReceiver) => messageReceiver()); + + await FluentActions.Awaiting(() => _sut.Handle(admitted, brokerContext.Object, () => throw new InvalidOperationException("handler failed"))) + .Should().ThrowAsync(); + await _sut.Handle(admitted, brokerContext.Object, () => Task.CompletedTask); + + _inbox.Verify(i => i.ReceiveViaInbox(admitted, brokerContext.Object, It.IsAny>()), Times.Exactly(2)); + } + + private static Mock CreateBrokerContext(string messageId = "id-1") + { + var bodyConverter = new Mock(); + bodyConverter.SetupGet(c => c.ContentType).Returns("application/json"); + var inbound = new InboundBrokeredMessage(messageId, new byte[] { 1 }, new Dictionary(), "receiver-path", bodyConverter.Object); + var brokerContext = new Mock(); + brokerContext.SetupGet(c => c.BrokeredMessage).Returns(inbound); + brokerContext.SetupGet(c => c.Container).Returns(new ContextContainer()); + return brokerContext; + } + + private static InMemoryBrokeredMessageInbox CreateInMemoryInbox() + => new InMemoryBrokeredMessageInbox(new Mock>().Object, + new ReliabilityOptions + { + InMemoryInboxDeduplicationWindowInMinutes = 60, + InMemoryInboxMaxEntries = 200000 + }); } } From 5eddba1dfa9ba57f0ef2ad6da3ea785e597a6d49 Mon Sep 17 00:00:00 2001 From: brenpike Date: Fri, 25 Sep 2026 15:28:39 -0600 Subject: [PATCH 02/11] docs(messagebrokers): record the inbox Delivery Entry in ADR-0041 (#534) --- ...-every-command-that-carries-its-context.md | 210 ++++++++++++++++++ src/Chatter.MessageBrokers/CONTEXT.md | 6 +- src/Chatter.MessageBrokers/src/README.md | 6 +- 3 files changed, 218 insertions(+), 4 deletions(-) create mode 100644 docs/adr/0041-the-inbox-gates-the-message-the-delivery-admitted-not-every-command-that-carries-its-context.md diff --git a/docs/adr/0041-the-inbox-gates-the-message-the-delivery-admitted-not-every-command-that-carries-its-context.md b/docs/adr/0041-the-inbox-gates-the-message-the-delivery-admitted-not-every-command-that-carries-its-context.md new file mode 100644 index 00000000..a6c0dfdd --- /dev/null +++ b/docs/adr/0041-the-inbox-gates-the-message-the-delivery-admitted-not-every-command-that-carries-its-context.md @@ -0,0 +1,210 @@ +--- +status: accepted +date: 2026-09-25 +--- + +# The Inbox gates the message the delivery admitted, not every command that carries its context + +The Inbox deduplicates a delivery. `InboxBehavior` now gates only the Delivery Entry: the message instance +the delivery admitted into the Command Pipeline. Every other message dispatched on that delivery's context goes straight +to `next()`. Before this change the behaviour gated any Command whose context was an `IMessageBrokerContext`, and a +Command dispatched in-process from inside a received handler carries that same context. This ADR records what that +did on each reliability tier, the options considered, the one that was chosen, and what it leaves as it was. + +Issue #534. + +## Context + +**The gate keyed on the context's type.** At master `77318ff`, `InboxBehavior.Handle` +(`src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/Inbox/InboxBehavior.cs:22-31`) sent the message +through `IBrokeredMessageInbox.ReceiveViaInbox` whenever `messageHandlerContext is IMessageBrokerContext` (`:25`), and +called `next()` directly otherwise. Nothing else was checked. + +**Nested dispatch hands the delivery's context on.** `context.InMemory()` +(`src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Context/MessageHandlerContextExtensions.cs:34-35`) returns an +`InMemoryDispatcher` over the handler's own context, and `InMemoryDispatcher.Dispatch` +(`src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Sending/InMemoryDispatcher.cs:14-22`) dispatches the new +message with that same context (`:18`). A Command dispatched this way from inside a received handler therefore reached +`InboxBehavior` with the outer delivery's `IMessageBrokerContext`, and so with the outer delivery's message id. To the +Inbox it looked like the delivery arriving a second time. + +**What each tier did with it.** Each tier's `ReceiveViaInbox` answered "has this message id been received?" for the +nested Command, and the answer was the outer delivery's own claim. + +- **In-memory** (`src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/Inbox/InMemoryBrokeredMessageInbox.cs:116-131`): + the outer delivery's reservation was still held, so `TryReserve` failed and the method returned without invoking + the nested handler. The only record was a `Trace` log line (`:129`). The nested Command's work was silently dropped. +- **Entity Framework (relational)** (`src/Chatter.MessageBrokers.Reliability.EntityFramework/src/Chatter.MessageBrokers.Reliability.EntityFramework/BrokeredMessageInbox.cs:120-131`): + the outer delivery's claim carries no timestamp, so it read as "a claim no handler completed", was claimed again in + place and the nested handler ran. That nested call then stamped the row as handled (`:154-158`) while the outer + handler was still running. A second nested Command in the same delivery found that stamp, read it as handled within + the Deduplication Window (`:114-118`, `IsHandledWithinTheDeduplicationWindow` at `:263-274`) and was skipped with + only an `Information` log line. Every stamp and claim sits in the unit of work's one transaction, so a failure of + the outer handler still rolled all of it back; what this tier got wrong was the redundant claim and stamp for the + first nested Command and the skipped work of every later one. +- **Cosmos standalone inbox** (`src/Chatter.MessageBrokers.Reliability.Cosmos/src/Chatter.MessageBrokers.Reliability.Cosmos/Reliability/CosmosBrokeredMessageInbox.cs:132-157`): + the create of the nested marker returned 409, the conflicting marker was the outer delivery's pending one, and the + inbox took it over, ran the nested handler and then `CompleteClaim` (`:177-182`) marked it `Completed` while the + outer handler was still running. A later nested Command in the same delivery then found a completed marker and was + skipped (`:139-143`). `WithCosmosInbox` registers no unit of work + (`src/Chatter.MessageBrokers.Reliability.Cosmos/src/Chatter.MessageBrokers.Reliability.Cosmos/CosmosInboxServiceCollectionExtensions.cs:107-117`), + so nothing rolled that completion back. If the outer handler then failed, the redelivery found a completed marker, + was treated as a confirmed duplicate and skipped. The outer delivery's work was lost. + +**The behaviour is opt-in.** `InboxBehavior<>` runs only where an application registers it: directly with +`WithBehavior(typeof(InboxBehavior<>))`, through `WithInboxBehavior()` in the Entity Framework module +(`Extensions.cs:43`), or through `WithCosmosInbox` in the Cosmos module (`CosmosInboxServiceCollectionExtensions.cs:117`). + +## Considered Options + +### Option 1 — key the Inbox on the message id and the message type (REJECTED) + +It enumerates cases instead of closing them. A nested Command of the same type as the one the delivery admitted would +still collide with the delivery's claim. A nested Command of a different type would write its own reservation under a +key that no redelivery ever reads, so the store fills with entries that deduplicate nothing. It also changes the key +every store writes, which is a contract change on all three tiers. + +### Option 2 — move the gate to the receive seam (REJECTED) + +Gating in `ScopedReceivedMessageDispatcher.DispatchAsync` +(`src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Receiving/ScopedReceivedMessageDispatcher.cs:18-26`), before +the message enters the Command Pipeline, would gate only the delivery by construction. It does not work for the +relational tier: `ClaimMessageIdAsync` refuses to claim when its `DbContext` has no active transaction +(`BrokeredMessageInbox.cs:189-197`), and that transaction is opened by `UnitOfWorkBehavior` inside the pipeline. The +gate has to stay inside the unit of work, so it has to stay in the pipeline. It would also stop honouring the opt-in: +the Inbox would run for every received message whether or not the application registered it, and every existing +`WithBehavior` registration would become meaningless. A receiver that overrides `DispatchReceivedMessageAsync`, as +`ChangeFeedReceiver` does, would bypass it altogether. + +### Option 3 — make `InMemory()` dispatch with a context that is not a broker context (REJECTED) + +It would stop nested Commands from reaching the Inbox, but it would also take them out of the delivery they run +inside. `UnitOfWorkBehavior` (`Reliability/UnitOfWorkBehavior.cs:19`) and `OutboxProcessingBehavior` +(`Reliability/Outbox/OutboxProcessingBehavior.cs:26`) find the delivery's `TransactionContext` in the context's +container, where the receiver put it (`BrokeredMessageReceiver.cs:1077`). The Cosmos document tier finds the delivery +through `GetInboundBrokeredMessage()` (`DocumentTierBatchLifecycleBehavior.cs:83`), which is the same +`is IMessageBrokerContext` test (`MessageHandlerContextExtensions.cs:161-169`). A nested Command given a fresh +non-broker context would find neither, so the reliability behaviours would treat it as work outside the delivery's +unit of work. `GetInboundBrokeredMessage()` would also return `null` inside the nested handler, which breaks +routing-slip reads and the send and publish options that `BrokeredMessageDispatcher` derives from the inbound message +(`BrokeredMessageDispatcher.cs:268`, `:271`). + +### Option 4 — a re-entrancy flag: skip the Inbox while a gated dispatch is running (REJECTED) + +It fixes the reported case, but it keys on timing rather than on which message was admitted, and two cases fall +outside it. When the delivered payload is an Event, no gated dispatch encloses the Commands its handlers dispatch, so +the first of them is gated, clears the flag when it returns, and every later sibling is gated again under the same +message id and, finding the first one's receipt, skipped on every tier. And a handler that re-dispatches the very instance it received would +pass the gate and run itself again. + +### Option 5 — gate only the Delivery Entry (ACCEPTED) + +Recorded under *Decision*. + +## Decision + +**The Inbox deduplicates a delivery, and gates only that delivery's Delivery Entry.** The Delivery Entry is the +message instance the delivery admitted into the Command Pipeline. An internal type, `InboxDeliveryEntry`, binds it on +the delivery's `ContextContainer`: the first message to reach `InboxBehavior` on a delivery's context becomes the +Delivery Entry, and only a message that is that same instance is sent through `ReceiveViaInbox`. Every other message +dispatched on that context, at any depth and by any dispatcher, goes straight to `next()`. + +**A recovery retry is gated again.** The receiver deserializes the payload once (`BrokeredMessageReceiver.cs:1081`) and +its recovery strategy re-dispatches that same instance on the same context (`:1089-1094`). The retry therefore +presents the Delivery Entry again, and the Inbox gates it again, exactly as it did before this change. + +**One edit fixes all three tiers.** The change is in `InboxBehavior`, above every store. No store's contract, +`IBrokeredMessageInbox.ReceiveViaInbox`, or any store implementation changes. + +**No public API changes.** `InboxDeliveryEntry` is internal. `InboxBehavior` keeps its type, constructor and +signature. + +**The ADR-0006 behaviour order is unchanged.** The relational canonical order +`[OutboxProcessingBehavior, UnitOfWorkBehavior, InboxBehavior]` still holds, and the Inbox still runs inside the unit +of work, alongside the handler. + +The rule and its rationale are stated once, in the `INVARIANT:` comments on `InboxBehavior.Handle` and on +`InboxDeliveryEntry`, which name the oracles that falsify them. This ADR cites them rather than restating them +(ADR-0027). + +## Closed-by-Construction Acceptance Test + +> What class of future finding does this make impossible, and why? + +**ELIMINATED CLASS: "a message the delivery did not admit is gated by the Inbox as if it were the delivery".** The gate +no longer keys on the context's type, which every nested dispatch inherits and which any future nested-dispatch API +would inherit too. It keys on the admitted message instance, which nested dispatch cannot produce: a Command created +inside a handler is a different instance, whatever its type and whatever context it carries. That holds at any depth, +for Commands dispatched from Command handlers and from Event handlers, and through any dispatcher. + +Pinned by these facts in `src/Chatter.MessageBrokers/tests/Reliability/Inbox/UsingInboxBehavior/WhenHandling.cs`: + +- `MustInvokeTheNestedHandlerWhenAGatedHandlersOwnDispatchReEntersTheBehavior`: a Command dispatched from inside the + gated handler reaches its own handler. +- `MustReceiveViaInboxOnlyTheMessageTheDeliveryAdmitted`: only the Delivery Entry is sent through `ReceiveViaInbox`. +- `MustInvokeTheHandlerOfASiblingDispatchedAfterTheAdmittedMessageCompleted`: a Command dispatched on the same context + after the Delivery Entry's handler returned still runs. +- `MustReceiveViaInboxAgainWhenTheAdmittedMessageIsRedispatchedAfterItsHandlerFailed`: a recovery retry of the + Delivery Entry is gated again. + +## Consequences + +- **Work that the in-memory tier silently dropped now runs.** A Command dispatched with `context.InMemory()` from + inside a received handler reaches its handler. +- **The Entity Framework tier no longer claims and stamps the delivery's row for a nested Command.** The row is + stamped only when the Delivery Entry's handler returns, so a second nested Command is no longer skipped. +- **The Cosmos standalone inbox no longer completes the delivery's marker early.** Only the Delivery Entry's own + `ReceiveViaInbox` call completes it, after the outer handler returns, so an outer failure leaves the marker pending + and the redelivery takes it over. A second nested Command is no longer skipped either. +- **The Inbox is not a loop guard.** A handler that dispatches a new Command of its own type now dispatches it for real + on every tier, so a handler that does so unconditionally recurses without end. The Entity Framework and Cosmos tiers + already did this; the in-memory tier used to stop it by accident, by dropping the nested Command. +- **`InboxDeliveryEntry` takes no lock.** The context container is unsynchronized by decision (ADR-0011), and a + delivery's nested dispatches are awaited one at a time on that delivery's context. +- **`ChangeFeedReceiver` fans one delivery out as many Events on one context.** `DispatchReceivedMessageAsync` + (`src/Chatter.SqlChangeFeed/src/Chatter.SqlChangeFeed/ChangeFeedReceiver.cs:35-79`) dispatches one Event per changed + row on the delivery's context (`:57`, `:63`, `:69`). The first Command any of those Events' handlers dispatches + takes the Delivery Entry and is gated; every later one runs. Before this change every later one was gated under the + same message id and, once the first one's receipt existed, skipped on every tier. + +### Recorded residuals + +These are decisions, not open work. + +- **R1: a broker-received Event is not itself deduplicated, so the first Command its handler dispatches takes the + Delivery Entry.** `InboxBehavior` is a Command behaviour and never sees the Event. When the delivered payload is an + Event, the first Command dispatched on the delivery's context is the first message to reach the Inbox, and it is + gated under the delivery's message id. This is inherited, not introduced: that Command was gated before this change + as well. It is bounded: later sibling Commands now run, and nothing is silently dropped. The obvious remedy, gating + at the receive seam, was rejected as Option 2. No issue is filed for it. + +Related work that is tracked: the Cosmos document tier has its own gate with the same root class, a nested participant +Command being treated as the delivery, and it is not changed here. That is #538. + +## References + +- Issue #534 — *In-memory Inbox silently skips a Command dispatched with context.InMemory() from inside a received + handler*. The defect this ADR resolves. +- Issue #538 — *Cosmos document tier silently discards a nested participant command's writes*. The same root class on + the document tier, tracked separately. +- ADR-0006 — *Two-tier reliability*. The relational canonical behaviour order this decision leaves unchanged, and the + inbox-plus-handler atomicity inside the unit of work that rules out Option 2. +- ADR-0008 — *Document-tier participation model*. The document tier's own gate, which this decision does not change. +- ADR-0009 — *Standalone Cosmos inbox: confirm, not infer, and fail loud*. The pending-marker take-over that let a + nested Command complete the delivery's marker early. +- ADR-0011 — *Context Container: unsynchronized, documented rather than synchronized*. Why `InboxDeliveryEntry` takes no + lock. +- ADR-0027 — *An `INVARIANT:` comment names the oracle that falsifies it, and states its rationale once*. Why this ADR + cites the `INVARIANT:` comments on `InboxBehavior.Handle` and `InboxDeliveryEntry` rather than restating them. +- `src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/Inbox/InboxBehavior.cs` — the gate + (`Handle`, `:22-31` at master `77318ff`). +- `src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/Inbox/InboxDeliveryEntry.cs` — the Delivery Entry + binding. +- `src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Context/MessageHandlerContextExtensions.cs` — `InMemory()` + (`:34-35`) and `GetInboundBrokeredMessage()` (`:161-169`). +- `src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Sending/InMemoryDispatcher.cs` — nested dispatch on the + handler's own context (`:14-22`). +- `src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Receiving/BrokeredMessageReceiver.cs` — the transaction + context placed on the delivery's container (`:1077`), the single deserialization (`:1081`) and the recovery + re-dispatch (`:1089-1094`). +- `src/Chatter.MessageBrokers/tests/Reliability/Inbox/UsingInboxBehavior/WhenHandling.cs` — the oracles. diff --git a/src/Chatter.MessageBrokers/CONTEXT.md b/src/Chatter.MessageBrokers/CONTEXT.md index b4001b2b..7b6ed24b 100644 --- a/src/Chatter.MessageBrokers/CONTEXT.md +++ b/src/Chatter.MessageBrokers/CONTEXT.md @@ -101,13 +101,15 @@ _Avoid_: reading it as the claim the processed stamp carries — ADR-0031 Option **Deduplication Window**: The period after a receipt completes during which a redelivery of the same message id is skipped. It is the in-memory Inbox's guarantee (`InMemoryBrokeredMessageInbox`), configured as `ReliabilityOptions.InMemoryInboxDeduplicationWindowInMinutes` and sized against worst-case redelivery. The same period is also the lease on an in-flight reservation, so the term states one rule rather than two: an id is remembered for the window, and a reservation is honoured for the window. A reservation the window has stopped honouring is abandoned — a handler that has not returned holds its id for the window and no longer — and from then on the Inbox treats it exactly as it treats a receipt whose window has elapsed. The window is MANDATORY and at least one minute; a smaller value is refused while the options are being built, because the window is the store's only reclamation rule and a window that could be switched off would leave entries nothing was able to remove. Abandonment does not distinguish a dead handler from a slow one: a merely slow but live handler is pre-empted on exactly the same terms, a second handler for that id then runs concurrently with it, and the first one's completion is discarded when it eventually arrives — so a handler that can outrun the window must be idempotent, and the window is sized above the slowest handler as well as against worst-case redelivery. The entry cap (`InMemoryInboxMaxEntries`) is a safety valve, not the guarantee: under memory pressure it evicts completed-but-unexpired receipts to bound memory, which truncates the window this term describes, and it reports that once in the log per store when it does — a redelivery inside the advertised window can then be handled again. The cap never takes a reservation the window still honours; because the window is always positive, every such reservation is released in bounded time, which is what leaves the cap able to bound memory without cancelling a handler still running inside the window it was promised. _Avoid_: sizing deduplication by entry count — the cap is not the guarantee, and calling it one is exactly the confusion this term exists to prevent. -**Inbox Deduplicator** (`IInboxDeduplicator`): the tier-neutral inbox dedup contract expressing once-only-handling intent (`HasBeenReceived`), implemented by both reliability tiers. Distinct from the relational-only `ReceiveViaInbox` wrap seam: this contract carries only the shared once-only intent, not the relational `Func` handler-wrap mechanics. On the relational tier the marker carries the reservation and the outcome in one row: `ReceiveViaInbox` flushes a marker carrying no `ReceivedByInboxAtUtc` — the CLAIM — into the ambient transaction before invoking the handler, then flushes the handled stamp once the handler returns, and both become visible outside that transaction only when `UnitOfWorkBehavior`'s single `SaveChangesAsync` commits them together with the handler's work. `MessageId` is the primary key, so an existing marker is re-claimed in place rather than re-inserted (`docs/adr/0026-the-relational-inbox-decides-expiry-at-receive-so-purge-timing-cannot-suppress-a-legitimate-message.md`, `docs/adr/0033-the-relational-inbox-claims-before-the-handler-and-stamps-handled-after-it-in-the-same-row.md`). The claim's flush takes the row's lock, so a second delivery arriving while the first handler is in flight WAITS on it rather than running its own handler, and reads the first delivery's outcome once that transaction settles. `HasBeenReceived` therefore reports `true` only for a committed marker carrying a stamp — and, where a Deduplication Window is configured, only one still inside it — never for a message id whose handler has not finished. Skip-on-reservation is a property of the in-memory inbox, not a tier-neutral one. For the in-memory inbox, `HasBeenReceived` reports `true` for an in-flight reservation and across the completed receipt's Deduplication Window: it reports `true` while the receipt is within that window and `false` once the window elapses. The in-flight reservation is itself leased by that same window, so `HasBeenReceived` reports `true` for a reservation the window still honours and `false` for one it has abandoned, and both the sweep and the cap reclaim an abandoned reservation on the same terms as an expired receipt. A concurrent delivery that meets a reservation the window still honours is still skipped immediately: it never waits for the reservation's owner and never takes the delivery over. +**Inbox Deduplicator** (`IInboxDeduplicator`): the tier-neutral inbox dedup contract expressing once-only-handling intent (`HasBeenReceived`), implemented by both reliability tiers. Distinct from the relational-only `ReceiveViaInbox` wrap seam: this contract carries only the shared once-only intent, not the relational `Func` handler-wrap mechanics. On the relational tier the marker carries the reservation and the outcome in one row: `ReceiveViaInbox` flushes a marker carrying no `ReceivedByInboxAtUtc` — the CLAIM — into the ambient transaction before invoking the handler, then flushes the handled stamp once the handler returns, and both become visible outside that transaction only when `UnitOfWorkBehavior`'s single `SaveChangesAsync` commits them together with the handler's work. `MessageId` is the primary key, so an existing marker is re-claimed in place rather than re-inserted (`docs/adr/0026-the-relational-inbox-decides-expiry-at-receive-so-purge-timing-cannot-suppress-a-legitimate-message.md`, `docs/adr/0033-the-relational-inbox-claims-before-the-handler-and-stamps-handled-after-it-in-the-same-row.md`). The claim's flush takes the row's lock, so a second delivery arriving while the first handler is in flight WAITS on it rather than running its own handler, and reads the first delivery's outcome once that transaction settles. `HasBeenReceived` therefore reports `true` only for a committed marker carrying a stamp — and, where a Deduplication Window is configured, only one still inside it — never for a message id whose handler has not finished. Skip-on-reservation is a property of the in-memory inbox, not a tier-neutral one. For the in-memory inbox, `HasBeenReceived` reports `true` for an in-flight reservation and across the completed receipt's Deduplication Window: it reports `true` while the receipt is within that window and `false` once the window elapses. The in-flight reservation is itself leased by that same window, so `HasBeenReceived` reports `true` for a reservation the window still honours and `false` for one it has abandoned, and both the sweep and the cap reclaim an abandoned reservation on the same terms as an expired receipt. A concurrent delivery that meets a reservation the window still honours is still skipped immediately: it never waits for the reservation's owner and never takes the delivery over. What either tier's `HasBeenReceived` gates is the Delivery Entry — the message the delivery admitted — not every command that merely carries that delivery's context (ADR-0041). _Avoid_: consumer, listener, middleware (honor existing `_Avoid_` aliases). Also avoid reading the relational tier as handler-first — it is claim-first, pinned by `MustFlushAClaimCarryingNoTimestampBeforeInvokingTheHandler` and `MustFlushTheStampWhenTheHandlerReturnsAndCommitIt` in `src/Chatter.MessageBrokers.Reliability.EntityFramework/tests/UsingBrokeredMessageInbox/WhenReceivingViaInbox.cs`, so the claim goes red if the order ever changes. **Reliability-Store Facet Resolution**: the secondary reliability facet (`IPollableOutboxStore` for the outbox pair; `IInboxDeduplicator` for the inbox pair) is not an independent DI registration. It is obtained by casting the single resolved primary (`IBrokeredMessageOutbox` / `IBrokeredMessageInbox`) at the consumption site — the same precedent `OutboxProcessor` uses to obtain `IUnitOfWork` from the outbox. A custom primary that does not implement the required secondary facet throws `InvalidCastException` at the poll site (loud, not silent). Because there is exactly one resolved instance per pair, split-store is impossible by construction; no descriptor inspection, lifetime reconciliation, or fail-fast registration is required. The default in-memory outbox realizes the `IUnitOfWork` facet non-transactionally: it runs the supplied operation as-is rather than opening a transaction, because there is nothing to enlist in-memory, and a failure the operation raises travels out unchanged. The two SHIPPED in-memory stores (`InMemoryBrokeredMessageOutbox`, `InMemoryBrokeredMessageInbox`) are registered process-lifetime rather than per-scope, because each one's whole state lives in the instance while every consumer — the outbox poll loop, the receive dispatcher — opens a fresh `IServiceScopeFactory` scope per operation, distinct from the Receiver Scope that bounds a whole receive loop; a per-scope instance would poll or deduplicate against an empty dictionary every time. An internal marker interface, `IProcessLifetimeStore`, names which shipped classes carry that requirement, and the `WhenSharingReliabilityStoresAcrossScopes` tests are what actually enforce it — they resolve a shipped default from two scopes opened the way production opens them and assert the same instance and the same state come back. A custom primary store keeps whatever lifetime the application registered it with; process-lifetime promotion applies only to the shipped default. _Avoid_: "Reliability-Pair Lifetime Contract" (the per-pair DI forwarding mechanism is eliminated); "the marker makes mis-registration unrepresentable" — it does not: the marker is a signpost, `AddIfNotRegistered(ServiceLifetime.Scoped)` remains callable for a type carrying it, and the cross-scope oracle (`WhenSharingReliabilityStoresAcrossScopes`) is what actually catches a regression. -**ReceiveViaInbox / InboxBehavior**: the `ReceiveViaInbox(() => next())` wrap seam is relational-only mechanics. The inbox **contract** and intent (once-only dedup, idempotency) are shared across tiers; the seam that implements them is provider-specific. The document tier enlists the inbox marker via the Document-Tier Batch-Lifecycle Behavior and the document-tier reliability surface — not through `ReceiveViaInbox`. +**ReceiveViaInbox / InboxBehavior**: the `ReceiveViaInbox(() => next())` wrap seam is relational-only mechanics. The inbox **contract** and intent (once-only dedup, idempotency) are shared across tiers; the seam that implements them is provider-specific. The document tier enlists the inbox marker via the Document-Tier Batch-Lifecycle Behavior and the document-tier reliability surface — not through `ReceiveViaInbox`. What the wrap seam gates is the Delivery Entry, not every command that merely carries the delivery's context (ADR-0041). + +**Delivery Entry**: the one message per delivery the Inbox gates — the message the delivery admitted into the Command Pipeline. Every message dispatched in-process inside that delivery (`context.InMemory()`) is inside the same delivery, not a new one, and passes the Inbox through. A recovery retry re-presents the admitted message, so it is gated again. Pinned by `WhenHandling.MustInvokeTheNestedHandlerWhenAGatedHandlersOwnDispatchReEntersTheBehavior`, `.MustReceiveViaInboxOnlyTheMessageTheDeliveryAdmitted`, `.MustInvokeTheHandlerOfASiblingDispatchedAfterTheAdmittedMessageCompleted` and `.MustReceiveViaInboxAgainWhenTheAdmittedMessageIsRedispatchedAfterItsHandlerFailed`. See ADR-0041. **Document-Tier Reliability Surface**: the provider-shaped reliability surface the document (NoSQL) tier owns. It carries the document-store primitives the relational-shaped `TransactionContext` does not and cannot carry: the resolved partition key, bound container, `TransactionalBatch` handle and batch lifecycle, inbox-marker enlistment, and ETag concurrency token. The document tier is the doc-tier sibling of `IPersistanceTransaction` for the atomic-write handle; the ETag concurrency token lives here, not on `TransactionContext`. The surface lives entirely in the Cosmos provider module — not in core. _Avoid_: treating `TransactionContext.Container` as the semantic home of document-tier primitives (it is an untyped dict that unifies storage, not semantics; no partition concept exists in core). diff --git a/src/Chatter.MessageBrokers/src/README.md b/src/Chatter.MessageBrokers/src/README.md index 00f9b07f..2f0820bb 100644 --- a/src/Chatter.MessageBrokers/src/README.md +++ b/src/Chatter.MessageBrokers/src/README.md @@ -261,7 +261,7 @@ await context.Publish(new OrderPlaced { OrderId = message.OrderId }, "order-even await context.InMemory().Dispatch(new ReserveStock { OrderId = message.OrderId }); ``` -`context.Send` and `context.Publish` copy the entire inbound Message Context onto each outbound message, and options you pass win over inherited entries; see [Inbound header trust](#inbound-header-trust). They do nothing when the context holds no brokered dispatcher, which happens only if `AddMessageBrokers` was not called. `context.InMemory().Dispatch(...)` dispatches in-process on the caller's own Message Context, so await each nested dispatch before starting the next. If you use the in-memory Inbox, read [Inbox](#inbox) before dispatching a Command this way. +`context.Send` and `context.Publish` copy the entire inbound Message Context onto each outbound message, and options you pass win over inherited entries; see [Inbound header trust](#inbound-header-trust). They do nothing when the context holds no brokered dispatcher, which happens only if `AddMessageBrokers` was not called. `context.InMemory().Dispatch(...)` dispatches in-process on the caller's own Message Context, so await each nested dispatch before starting the next. ### Send options @@ -466,7 +466,9 @@ The Inbox reserves the message id before your handler runs, and a concurrent del - its id was evicted at the entry cap; - the process restarted, or the redelivery reached another instance, because each process keeps its own in-memory Inbox. -`InboxBehavior<>` acts on a command dispatched with an `IMessageBrokerContext`, which a Brokered Message Receiver supplies; a command dispatched from outside a received handler passes through. A command your handler dispatches with `context.InMemory()` carries the inbound message id, so the in-memory Inbox skips its handler while your handler holds that id ([#534](https://github.com/brenpike/Chatter/issues/534)). A received message without a `MessageId` makes the in-memory Inbox throw `ArgumentException`, and the handler does not run. +`InboxBehavior<>` acts on a command dispatched with an `IMessageBrokerContext`, which a Brokered Message Receiver supplies; a command dispatched from outside a received handler passes through. It gates only the **Delivery Entry** — the message the delivery admitted into the Command Pipeline, the first message it sees on that delivery's context; a recovery retry re-presents that same message and is gated again. A Command your handler dispatches in-process with `context.InMemory()` belongs to the same delivery, so it passes through the Inbox and runs. A broker-received event is not itself deduplicated, since the Inbox is a Command behavior; the first Command its handler dispatches in-process is the one the Inbox gates, and later ones run. See [ADR-0041](https://github.com/brenpike/Chatter/blob/master/docs/adr/0041-the-inbox-gates-the-message-the-delivery-admitted-not-every-command-that-carries-its-context.md). A received message without a `MessageId` makes the in-memory Inbox throw `ArgumentException`, and the handler does not run. + +The Inbox is not a loop guard: a handler that dispatches back into itself in-process belongs to the same delivery each time, so the dispatch passes through and the handler will now loop on every inbox tier. The in-memory Inbox has two settings: From ff0684cff0f2b60badc8959f01101975ad811807 Mon Sep 17 00:00:00 2001 From: brenpike Date: Fri, 25 Sep 2026 15:32:38 -0600 Subject: [PATCH 03/11] docs(messagebrokers): correct the outbox ordering and inbox re-dispatch comments (#534) --- .../Reliability/InMemoryBrokeredMessageOutbox.cs | 2 +- .../Reliability/Inbox/InboxBehavior.cs | 8 ++++++-- 2 files changed, 7 insertions(+), 3 deletions(-) diff --git a/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/InMemoryBrokeredMessageOutbox.cs b/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/InMemoryBrokeredMessageOutbox.cs index d6d59425..edb14b10 100644 --- a/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/InMemoryBrokeredMessageOutbox.cs +++ b/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/InMemoryBrokeredMessageOutbox.cs @@ -283,7 +283,7 @@ private void RemoveExpiredFromInboxOutbox() // timestamp, so no expiry instant is computed and there is nothing left to overflow - // subtracting two DateTime values always fits a TimeSpan and comparing two doubles is total. // AddMinutes was the one operation here that could throw, and OutboxProcessor calls - // UpdateProcessedDate inside its unit of work BEFORE dispatching and catches every exception + // UpdateProcessedDate inside its unit of work AFTER dispatching and catches every exception // around the whole block, so a ttl too large to add stamped each message processed, logged one // line and left it never dispatched and never retried. A ttl no elapsed time can reach now // simply never expires anything, which is what such a ttl asks for. diff --git a/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/Inbox/InboxBehavior.cs b/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/Inbox/InboxBehavior.cs index e357444b..31c6d329 100644 --- a/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/Inbox/InboxBehavior.cs +++ b/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/Inbox/InboxBehavior.cs @@ -31,8 +31,12 @@ public Task Handle(TMessage message, IMessageHandlerContext messageHandlerContex // MustReceiveViaInboxOnlyTheMessageTheDeliveryAdmitted and // MustInvokeTheHandlerOfASiblingDispatchedAfterTheAdmittedMessageCompleted; binding on a first-call flag // instead of the instance reddens MustReceiveViaInboxAgainWhenTheAdmittedMessageIsRedispatchedAfterItsHandlerFailed. - // A handler that re-dispatches the received INSTANCE itself is still gated, and so skipped while its - // own receipt holds the id; no test pins that. + // A handler that re-dispatches the received INSTANCE itself is gated again - the Admits() check + // reports true for the same instance on every call, so it reaches ReceiveViaInbox again - and what + // happens next is the store's own duplicate rule: the in-memory store skips it, while the relational + // and standalone Cosmos stores re-claim or take over the still-open claim and run it (EF + // BrokeredMessageInbox, Chatter.MessageBrokers.Reliability.Cosmos CosmosBrokeredMessageInbox); no test + // pins that. if (messageHandlerContext is IMessageBrokerContext messageBrokerContext && messageBrokerContext.Container.GetOrNew().Admits(message)) { From 5536abec44bf8b3e988c3233594e01bd9339b718 Mon Sep 17 00:00:00 2001 From: brenpike Date: Fri, 25 Sep 2026 15:34:33 -0600 Subject: [PATCH 04/11] chore(messagebrokers): bump Chatter.MessageBrokers to 0.35.3 --- .../src/Chatter.MessageBrokers/CHANGELOG.md | 7 +++++++ .../Chatter.MessageBrokers/Chatter.MessageBrokers.csproj | 2 +- 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/CHANGELOG.md b/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/CHANGELOG.md index 3b009748..e93c54e6 100644 --- a/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/CHANGELOG.md +++ b/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/CHANGELOG.md @@ -12,6 +12,13 @@ This project follows [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) an ### Fixed +## [0.35.3] - 2026-09-25 + +### Fixed + +- **A Command a received handler dispatches in-process now runs.** `InboxBehavior<>` gated every Command dispatched with the delivery's context, so a Command a handler dispatched with `context.InMemory()` looked like the delivery itself. The in-memory Inbox silently skipped it; the relational Inbox re-claimed it and could skip a second one; the standalone Cosmos Inbox completed the delivery's marker before the outer handler finished, so a later failure of the outer handler could be lost on redelivery. The Inbox now gates only the message the delivery admitted (a recovery retry of that message is still gated). No code change is needed. The Inbox is not a loop guard: a handler that dispatches a new Command back into itself now loops on every Inbox. ADR-0041 (#534). +- **A code comment on `InMemoryBrokeredMessageOutbox` that described `OutboxProcessor` as stamping a dispatched message's processed date BEFORE dispatching is corrected to say AFTER dispatching, matching the code it describes.** This carries no code change and no user-facing effect (#534). + ## [0.35.2] - 2026-09-25 ### Fixed diff --git a/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Chatter.MessageBrokers.csproj b/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Chatter.MessageBrokers.csproj index bbf24423..3ca31dbb 100644 --- a/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Chatter.MessageBrokers.csproj +++ b/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Chatter.MessageBrokers.csproj @@ -7,7 +7,7 @@ TryEncodeUnicodeScalar) is char*-pointer based, so the overrides must be unsafe. --> true Chatter.MessageBrokers - 0.35.2 + 0.35.3 Brennan Pike Brennan Pike A message broker adapter library built on top of Chatter.CQRS. From 0c3a86327e4c2728bb33a12bc975b5b0daceb958 Mon Sep 17 00:00:00 2001 From: brenpike Date: Fri, 25 Sep 2026 16:03:14 -0600 Subject: [PATCH 05/11] docs(messagebrokers): correct the outbox stamp-ordering comment consequent (#534) --- .../Reliability/InMemoryBrokeredMessageOutbox.cs | 14 +++++++++++--- .../WhenManagingOutbox.cs | 7 ++++--- 2 files changed, 15 insertions(+), 6 deletions(-) diff --git a/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/InMemoryBrokeredMessageOutbox.cs b/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/InMemoryBrokeredMessageOutbox.cs index edb14b10..24b7feb9 100644 --- a/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/InMemoryBrokeredMessageOutbox.cs +++ b/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/InMemoryBrokeredMessageOutbox.cs @@ -284,9 +284,17 @@ private void RemoveExpiredFromInboxOutbox() // subtracting two DateTime values always fits a TimeSpan and comparing two doubles is total. // AddMinutes was the one operation here that could throw, and OutboxProcessor calls // UpdateProcessedDate inside its unit of work AFTER dispatching and catches every exception - // around the whole block, so a ttl too large to add stamped each message processed, logged one - // line and left it never dispatched and never retried. A ttl no elapsed time can reach now - // simply never expires anything, which is what such a ttl asks for. + // around the whole block, so a ttl too large to add stamped each message processed - the stamp + // lands before this scan runs - and then threw on a message the publish had ALREADY put on the + // broker. Nothing was lost and nothing was published twice: the in-memory unit of work has no + // rollback to discard that stamp, and no later poll hands back a stamped row. What it cost was + // the failure ladder that followed every drain - one logged line, a second from the re-claim of + // the published row throwing the same way, and a dispatch attempt spent on a row already + // processed. A ttl no elapsed time can reach now simply never expires anything, which is what + // such a ttl asks for. + // Oracle: MustRetainProcessedMessageWithoutFaultingWhenMinutesToLiveOutrunsTheDateRange - + // restoring the AddMinutes form reddens it. The pre-fix failure ladder described above is + // history rather than behaviour this file has now, and no test pins it. if ((DateTime.UtcNow - message.ProcessedFromOutboxAtUtc.Value).TotalMinutes < ttl) { continue; diff --git a/src/Chatter.MessageBrokers/tests/Reliability/Outbox/UsingInMemoryBrokeredMessageOutbox/WhenManagingOutbox.cs b/src/Chatter.MessageBrokers/tests/Reliability/Outbox/UsingInMemoryBrokeredMessageOutbox/WhenManagingOutbox.cs index 0c01e390..bd866f74 100644 --- a/src/Chatter.MessageBrokers/tests/Reliability/Outbox/UsingInMemoryBrokeredMessageOutbox/WhenManagingOutbox.cs +++ b/src/Chatter.MessageBrokers/tests/Reliability/Outbox/UsingInMemoryBrokeredMessageOutbox/WhenManagingOutbox.cs @@ -174,9 +174,10 @@ public async Task MustRetainProcessedMessageWithoutFaultingWhenMinutesToLiveOutr { // INVARIANT: the scan compares ELAPSED minutes against the ttl instead of adding the ttl to the // processed timestamp, so a ttl no elapsed time can ever reach costs nothing and expires nothing. - // Adding it threw an ArgumentOutOfRangeException out of UpdateProcessedDate, which OutboxProcessor - // calls inside its unit of work BEFORE dispatching and catches around the whole block, so every - // message was stamped processed, logged and then never dispatched and never retried. + // This test is the oracle that INVARIANT names: restoring the AddMinutes form reddens it. What the + // AddMinutes form actually cost - a throw out of UpdateProcessedDate raised AFTER the publish and + // after the stamp, so no message was lost and none was published twice - is recorded once next to + // the mechanism, on InMemoryBrokeredMessageOutbox.RemoveExpiredFromInboxOutbox (ADR-0027). _reliabilityOptions.MinutesToLiveInMemory = 1e300; await _sut.SendToOutbox(CreateOutbound("id-1"), new TransactionContext()); var message = (await _sut.GetUnprocessedMessagesFromOutbox()).Single(); From 66c560aeee8ae13550af2923e3e95992a55a8774 Mon Sep 17 00:00:00 2001 From: brenpike Date: Fri, 25 Sep 2026 17:01:41 -0600 Subject: [PATCH 06/11] fix(messagebrokers): scope the inbox Delivery Entry to one receive attempt (#534) --- .../Receiving/BrokeredMessageReceiver.cs | 21 +- .../Reliability/Inbox/InboxBehavior.cs | 13 +- .../Reliability/Inbox/InboxDeliveryEntry.cs | 12 +- ...henGatingTheInboxAcrossRecoveryAttempts.cs | 244 ++++++++++++++++++ 4 files changed, 279 insertions(+), 11 deletions(-) create mode 100644 src/Chatter.MessageBrokers/tests/Receiving/UsingBrokeredMessageReceiver/WhenGatingTheInboxAcrossRecoveryAttempts.cs diff --git a/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Receiving/BrokeredMessageReceiver.cs b/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Receiving/BrokeredMessageReceiver.cs index 10131495..32622cea 100644 --- a/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Receiving/BrokeredMessageReceiver.cs +++ b/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Receiving/BrokeredMessageReceiver.cs @@ -4,6 +4,7 @@ using Chatter.MessageBrokers.Diagnostics; using Chatter.MessageBrokers.Exceptions; using Chatter.MessageBrokers.Recovery; +using Chatter.MessageBrokers.Reliability.Inbox; using Microsoft.Extensions.Logging; using System; using System.Collections.Generic; @@ -1061,6 +1062,24 @@ public virtual async Task DispatchReceivedMessageAsync(TMessage payload, Message } } + /// + /// Begins one Recovery attempt at : counts the attempt and installs a fresh + /// Delivery Entry, so the first command this attempt dispatches is the one the inbox gates. + /// + private static void BeginReceiveAttempt(MessageBrokerContext messageContext) + { + CountReceiveAttempt(); + + // INVARIANT: the Delivery Entry is attempt-scoped. A retry re-runs the dispatch on the SAME context, and a + // handler (or an overriding receiver such as ChangeFeedReceiver) may construct a fresh command per attempt, + // which an entry bound on an earlier attempt would never admit, so that attempt would bypass the inbox. The + // entry is installed here, not inside CountReceiveAttempt, because that returns early when BrokerDiagnostics + // has no listeners. Pinned by WhenGatingTheInboxAcrossRecoveryAttempts: deleting this Include reddens + // MustGateTheFirstCommandOfEveryRecoveryAttemptWhenTheHandlerBuildsAFreshOne and + // MustGateTheFirstCommandOfEveryAttemptWhenAReceiverOverridesTheDispatch. + messageContext.Container.Include(new InboxDeliveryEntry()); + } + async Task ProcessMessageAsync(MessageBrokerContext messageContext, TransactionContext transactionContext, CancellationToken receiverTokenSource) { receiverTokenSource.ThrowIfCancellationRequested(); @@ -1088,7 +1107,7 @@ async Task ProcessMessageAsync(MessageBrokerContext messageContext, TransactionC using var localTransaction = _infrastructureReceiver.CreateLocalTransaction(transactionContext); await _recoveryStrategy.ExecuteAsync(async () => { - CountReceiveAttempt(); + BeginReceiveAttempt(messageContext); await DispatchReceivedMessageAsync(brokeredMessagePayload, messageContext, receiverTokenSource); return true; }, receiverTokenSource); diff --git a/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/Inbox/InboxBehavior.cs b/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/Inbox/InboxBehavior.cs index 31c6d329..151ae36d 100644 --- a/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/Inbox/InboxBehavior.cs +++ b/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/Inbox/InboxBehavior.cs @@ -23,10 +23,15 @@ public Task Handle(TMessage message, IMessageHandlerContext messageHandlerContex { _logger.LogDebug($"Entering {nameof(InboxBehavior)}"); - // INVARIANT: only the Delivery Entry - the message this delivery admitted first - is received via the - // inbox; every other command dispatched on the same broker context (nested in-process dispatch - // inherits it) goes straight to next(), because the entry's handler already holds the delivery's - // message id (ADR-0041). Pinned by WhenHandling: restoring the bare `is IMessageBrokerContext` gate + // INVARIANT: only the Delivery Entry - the message the current receive attempt admitted first - is received + // via the inbox; every other command dispatched on the same broker context in that attempt (nested + // in-process dispatch inherits it) goes straight to next(), because the entry's handler already holds the + // delivery's message id (ADR-0041). The entry is attempt-scoped: the receiver installs a fresh one at the + // start of every Recovery attempt (BrokeredMessageReceiver.BeginReceiveAttempt), so a retry that constructs + // a fresh command is gated again; GetOrNew still creates one lazily for a dispatch that never passed that + // seam. Pinned by WhenGatingTheInboxAcrossRecoveryAttempts: deleting the Include in BeginReceiveAttempt + // reddens MustGateTheFirstCommandOfEveryRecoveryAttemptWhenTheHandlerBuildsAFreshOne and + // MustGateTheFirstCommandOfEveryAttemptWhenAReceiverOverridesTheDispatch. Pinned by WhenHandling: restoring the bare `is IMessageBrokerContext` gate // reddens MustInvokeTheNestedHandlerWhenAGatedHandlersOwnDispatchReEntersTheBehavior, // MustReceiveViaInboxOnlyTheMessageTheDeliveryAdmitted and // MustInvokeTheHandlerOfASiblingDispatchedAfterTheAdmittedMessageCompleted; binding on a first-call flag diff --git a/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/Inbox/InboxDeliveryEntry.cs b/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/Inbox/InboxDeliveryEntry.cs index 7d05fb37..1dd1e59c 100644 --- a/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/Inbox/InboxDeliveryEntry.cs +++ b/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/Inbox/InboxDeliveryEntry.cs @@ -1,16 +1,16 @@ namespace Chatter.MessageBrokers.Reliability.Inbox { /// - /// The Delivery Entry: the message a delivery admitted into the command pipeline first. It lives in the - /// delivery's , so it is scoped to one - /// and therefore to one delivery. + /// The Delivery Entry: the message a receive attempt admitted into the command pipeline first. It lives in the + /// delivery's , and the receiver replaces it at the start of + /// every Recovery attempt, so it is scoped to one attempt of one delivery. /// internal sealed class InboxDeliveryEntry { // INVARIANT: no lock. The entry lives in the delivery's ContextContainer, which is unsynchronized and must - // never be used from two threads at once (ADR-0011), and a delivery's nested dispatches are awaited inside - // its handler, so the first Admits call has bound the entry before any other call on this context runs. - // No test pins it. + // never be used from two threads at once (ADR-0011): one worker owns the delivery's container, runs its + // Recovery attempts one after another, and awaits an attempt's nested dispatches inside its handler, so the + // first Admits call has bound the attempt's entry before any other call on this context runs. No test pins it. private object _admittedMessage; /// diff --git a/src/Chatter.MessageBrokers/tests/Receiving/UsingBrokeredMessageReceiver/WhenGatingTheInboxAcrossRecoveryAttempts.cs b/src/Chatter.MessageBrokers/tests/Receiving/UsingBrokeredMessageReceiver/WhenGatingTheInboxAcrossRecoveryAttempts.cs new file mode 100644 index 00000000..31952cab --- /dev/null +++ b/src/Chatter.MessageBrokers/tests/Receiving/UsingBrokeredMessageReceiver/WhenGatingTheInboxAcrossRecoveryAttempts.cs @@ -0,0 +1,244 @@ +#nullable disable + +using Chatter.CQRS.Commands; +using Chatter.MessageBrokers.Configuration; +using Chatter.MessageBrokers.Context; +using Chatter.MessageBrokers.Receiving; +using Chatter.MessageBrokers.Recovery; +using Chatter.MessageBrokers.Recovery.CircuitBreaker; +using Chatter.MessageBrokers.Recovery.Options; +using Chatter.MessageBrokers.Recovery.Retry; +using Chatter.MessageBrokers.Reliability.Inbox; +using Chatter.MessageBrokers.Tests.Receiving.Fakes; +using FluentAssertions; +using Microsoft.Extensions.Logging.Abstractions; +using Moq; +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using Xunit; + +namespace Chatter.MessageBrokers.Tests.Receiving.UsingBrokeredMessageReceiver +{ + public class WhenGatingTheInboxAcrossRecoveryAttempts : Testing.Core.Context + { + private class FakeMessage : CQRS.IMessage + { + public string Value { get; set; } + } + + private class FakeCommand : ICommand { } + + /// + /// A handler that constructs a fresh on every call and runs it through a real + /// on the delivery's context; its first call fails from inside the + /// inbox, so the receiver's recovery strategy retries the delivery on the same context. The behavior logs through + /// a because Castle cannot proxy ILogger<InboxBehavior<FakeCommand>> + /// for the private . + /// + private sealed class FreshCommandHandler + { + private readonly InboxBehavior _inboxBehavior; + private int _attemptCount; + + public FreshCommandHandler(IBrokeredMessageInbox inbox) + { + _inboxBehavior = new InboxBehavior(inbox, NullLogger>.Instance); + } + + public List ConstructedCommands { get; } = new List(); + + public Task HandleAttemptAsync(MessageBrokerContext messageContext) + { + _attemptCount++; + var command = new FakeCommand(); + ConstructedCommands.Add(command); + var failThisAttempt = _attemptCount == 1; + return _inboxBehavior.Handle(command, messageContext, () => RunCommandHandlerAsync(failThisAttempt)); + } + + private static Task RunCommandHandlerAsync(bool failThisAttempt) + => failThisAttempt + ? Task.FromException(new InvalidOperationException("transient failure on the first attempt")) + : Task.CompletedTask; + } + + /// + /// A receiver that overrides the dispatch to build its own command per attempt, the shape + /// Chatter.SqlChangeFeed's ChangeFeedReceiver takes. + /// + private sealed class FreshCommandReceiver : BrokeredMessageReceiver + { + private readonly FreshCommandHandler _handler; + + public FreshCommandReceiver(IMessagingInfrastructureProvider infrastructureProvider, IRecoveryStrategy recoveryStrategy, FreshCommandHandler handler) + : base(infrastructureProvider, + BuildBrokerOptions(), + NullLogger>.Instance, + new Mock().Object, + new Mock().Object, + recoveryStrategy, + new Mock().Object) + { + _handler = handler; + } + + public override Task DispatchReceivedMessageAsync(FakeMessage payload, MessageBrokerContext messageContext, CancellationToken receiverTokenSource) + => _handler.HandleAttemptAsync(messageContext); + } + + private static MessageBrokerOptions BuildBrokerOptions() + { + var opts = new MessageBrokerOptions(); + opts.TransactionMode = TransactionMode.None; + return opts; + } + + private static ReceiverOptions BuildReceiverOptions() + => new ReceiverOptions + { + InfrastructureType = InMemoryMessagingInfrastructureProvider.InfrastructureType, + MessageReceiverPath = "test-queue", + SendingPath = "test-queue", + ErrorQueuePath = "error-queue", + DeadLetterQueuePath = "deadletter-queue", + TransactionMode = TransactionMode.None, + MaxReceiveAttempts = 10, + }; + + private static MessageBrokerContext BuildContext() + { + var converter = new JsonBodyConverter(); + var body = converter.Convert(new FakeMessage { Value = "hello" }); + return new MessageBrokerContext( + messageId: Guid.NewGuid().ToString(), + body: body, + applicationProperties: new Dictionary(), + messageReceiverPath: "test-queue", + receiverCancellationToken: CancellationToken.None, + bodyConverter: converter); + } + + /// + /// Builds the real retry-with-circuit-breaker strategy. Attempt 1 fails and attempt 2 succeeds, so + /// MaxRetryAttempts=3 leaves ONE attempt of headroom (see WhenRecoveringThroughTheSeam's attempt-arithmetic + /// note); the breaker never trips because its predicate list is empty. + /// + private static IRecoveryStrategy BuildStrategy() + { + var cbOptions = new CircuitBreakerOptions + { + OpenToHalfOpenWaitTimeInSeconds = 0, + ConcurrentHalfOpenAttempts = 1, + NumberOfFailuresBeforeOpen = 2, + NumberOfHalfOpenSuccessesToClose = 1, + SecondsOpenBeforeCriticalFailureNotification = 0, + }; + var circuitBreaker = new CircuitBreaker( + new InMemoryCircuitBreakerStateStore(NullLogger.Instance), + cbOptions, + NullLogger.Instance, + new CircuitBreakerExceptionEvaluator(Array.Empty())); + var recoveryOptions = new RecoveryOptions { MaxRetryAttempts = 3 }; + var retryEvaluator = new RetryExceptionEvaluator( + new[] { new ConfigRetryExceptionPredicatesProvider( + new Predicate[] { e => e is InvalidOperationException }) }); + var retryStrategy = new RetryStrategy( + recoveryOptions, + NullLogger.Instance, + new NoDelayRetry(), + retryEvaluator); + return new RetryWithCircuitBreakerStrategy(recoveryOptions, circuitBreaker, retryStrategy); + } + + private static Mock BuildRecordingInbox(List receivedViaInbox) + { + var inbox = new Mock(); + inbox.Setup(i => i.ReceiveViaInbox(It.IsAny(), It.IsAny(), It.IsAny>())) + .Callback>((command, _, __) => receivedViaInbox.Add(command)) + .Returns>((_, __, messageReceiver) => messageReceiver()); + return inbox; + } + + /// + /// Awaits the drain, bounded by the watchdog so a receiver loop that never reaches ReceiveMessageAsync fails + /// promptly instead of hanging (mirrors WhenRecoveringThroughTheSeam). + /// + private static async Task AwaitDrainedAsync(InMemoryMessagingInfrastructureReceiver infraReceiver, CancellationToken watchdog) + { + var watchdogTcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + using (watchdog.Register(() => watchdogTcs.TrySetCanceled(watchdog))) + { + var completed = await Task.WhenAny(infraReceiver.Drained, watchdogTcs.Task); + await completed; + } + } + + private static async Task WaitForDispositionAsync(InMemoryMessagingInfrastructureReceiver infraReceiver, ReceiverCall expectedDisposition, CancellationToken watchdog) + { + while (!infraReceiver.CallLog.Contains(expectedDisposition)) + { + watchdog.ThrowIfCancellationRequested(); + await Task.Yield(); + } + } + + private static async Task ReceiveUntilAckedAsync(BrokeredMessageReceiver sut, InMemoryMessagingInfrastructureReceiver infraReceiver) + { + using var cts = new CancellationTokenSource(); + var loop = Task.Run(() => sut.StartReceiver(BuildReceiverOptions(), cts.Token)); + + using var watchdog = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + await AwaitDrainedAsync(infraReceiver, watchdog.Token); + await WaitForDispositionAsync(infraReceiver, ReceiverCall.Ack, watchdog.Token); + + cts.Cancel(); + await loop; + } + + [Fact] + public async Task MustGateTheFirstCommandOfEveryRecoveryAttemptWhenTheHandlerBuildsAFreshOne() + { + var infraReceiver = new InMemoryMessagingInfrastructureReceiver(expectedMessageCount: 1); + infraReceiver.Enqueue(BuildContext()); + var receivedViaInbox = new List(); + var handler = new FreshCommandHandler(BuildRecordingInbox(receivedViaInbox).Object); + var dispatcher = new Mock(); + dispatcher + .Setup(d => d.DispatchAsync(It.IsAny(), It.IsAny(), It.IsAny())) + .Returns((_, messageContext, __) => handler.HandleAttemptAsync(messageContext)); + var sut = new BrokeredMessageReceiver( + infrastructureProvider: new InMemoryMessagingInfrastructureProvider(infraReceiver), + messageBrokerOptions: BuildBrokerOptions(), + logger: NullLogger>.Instance, + recoveryAction: new Mock().Object, + criticalFailureNotifier: new Mock().Object, + recoveryStrategy: BuildStrategy(), + receivedMessageDispatcher: dispatcher.Object); + + await ReceiveUntilAckedAsync(sut, infraReceiver); + + handler.ConstructedCommands.Should().HaveCount(2, because: "attempt 1 fails and the recovery strategy retries once"); + receivedViaInbox.Should().Equal(handler.ConstructedCommands, + because: "each recovery attempt's first command is that attempt's Delivery Entry and must be received via the inbox"); + } + + [Fact] + public async Task MustGateTheFirstCommandOfEveryAttemptWhenAReceiverOverridesTheDispatch() + { + var infraReceiver = new InMemoryMessagingInfrastructureReceiver(expectedMessageCount: 1); + infraReceiver.Enqueue(BuildContext()); + var receivedViaInbox = new List(); + var handler = new FreshCommandHandler(BuildRecordingInbox(receivedViaInbox).Object); + var sut = new FreshCommandReceiver(new InMemoryMessagingInfrastructureProvider(infraReceiver), BuildStrategy(), handler); + + await ReceiveUntilAckedAsync(sut, infraReceiver); + + handler.ConstructedCommands.Should().HaveCount(2, because: "attempt 1 fails and the recovery strategy retries once"); + receivedViaInbox.Should().Equal(handler.ConstructedCommands, + because: "each recovery attempt's first command is that attempt's Delivery Entry and must be received via the inbox"); + } + } +} From b89ef7e39955149ee73c5224b458f636f5715b16 Mon Sep 17 00:00:00 2001 From: brenpike Date: Fri, 25 Sep 2026 17:05:23 -0600 Subject: [PATCH 07/11] docs(messagebrokers): describe the attempt-scoped Delivery Entry (#534) --- ...-every-command-that-carries-its-context.md | 114 +++++++++++++----- src/Chatter.MessageBrokers/CONTEXT.md | 2 +- .../src/Chatter.MessageBrokers/CHANGELOG.md | 2 +- src/Chatter.MessageBrokers/src/README.md | 2 +- 4 files changed, 87 insertions(+), 33 deletions(-) diff --git a/docs/adr/0041-the-inbox-gates-the-message-the-delivery-admitted-not-every-command-that-carries-its-context.md b/docs/adr/0041-the-inbox-gates-the-message-the-delivery-admitted-not-every-command-that-carries-its-context.md index a6c0dfdd..ebfd40f8 100644 --- a/docs/adr/0041-the-inbox-gates-the-message-the-delivery-admitted-not-every-command-that-carries-its-context.md +++ b/docs/adr/0041-the-inbox-gates-the-message-the-delivery-admitted-not-every-command-that-carries-its-context.md @@ -5,9 +5,9 @@ date: 2026-09-25 # The Inbox gates the message the delivery admitted, not every command that carries its context -The Inbox deduplicates a delivery. `InboxBehavior` now gates only the Delivery Entry: the message instance -the delivery admitted into the Command Pipeline. Every other message dispatched on that delivery's context goes straight -to `next()`. Before this change the behaviour gated any Command whose context was an `IMessageBrokerContext`, and a +The Inbox deduplicates a delivery. `InboxBehavior` now gates only the Delivery Entry: the first message +instance the delivery's current receive attempt admitted into the Command Pipeline. Every other message dispatched on +that delivery's context in that attempt goes straight to `next()`. Before this change the behaviour gated any Command whose context was an `IMessageBrokerContext`, and a Command dispatched in-process from inside a received handler carries that same context. This ADR records what that did on each reliability tier, the options considered, the one that was chosen, and what it leaves as it was. @@ -81,7 +81,7 @@ the Inbox would run for every received message whether or not the application re It would stop nested Commands from reaching the Inbox, but it would also take them out of the delivery they run inside. `UnitOfWorkBehavior` (`Reliability/UnitOfWorkBehavior.cs:19`) and `OutboxProcessingBehavior` (`Reliability/Outbox/OutboxProcessingBehavior.cs:26`) find the delivery's `TransactionContext` in the context's -container, where the receiver put it (`BrokeredMessageReceiver.cs:1077`). The Cosmos document tier finds the delivery +container, where the receiver put it (`BrokeredMessageReceiver.cs:1096`). The Cosmos document tier finds the delivery through `GetInboundBrokeredMessage()` (`DocumentTierBatchLifecycleBehavior.cs:83`), which is the same `is IMessageBrokerContext` test (`MessageHandlerContextExtensions.cs:161-169`). A nested Command given a fresh non-broker context would find neither, so the reliability behaviours would treat it as work outside the delivery's @@ -101,17 +101,40 @@ pass the gate and run itself again. Recorded under *Decision*. -## Decision +### Option 6 — stamp the attempt ordinal on the delivery's container (REJECTED) + +The receiver would stamp each Recovery attempt's ordinal on the delivery's container, and `InboxDeliveryEntry` would +rebind whenever the ordinal it was bound under differs from the stamped one. It needs the same cooperation from the +receive seam as Option 5, which installs a fresh entry at that seam instead, and it has more parts: a second container +value, the ordinal's bookkeeping and a comparison in the entry. It closes no case that Option 5 leaves open. + +### Option 7 — key the Delivery Entry on the deserialized payload instance (REJECTED) + +The receiver would record the payload instance it deserialized, and the Inbox would gate only that instance. A retry +re-presents that instance, so it would be gated again. But `InboxBehavior` is a Command behaviour, so when the +delivered payload is an Event the payload never reaches the Inbox, and no Command dispatched during the delivery would +be gated at all. The same holds for every `ChangeFeedReceiver` delivery: its override of `DispatchReceivedMessageAsync` +never dispatches the payload and dispatches Events built from it instead. Those deliveries would lose the +deduplication they had on master, where the first Command their handlers dispatched was gated under the delivery's +message id (see R1). -**The Inbox deduplicates a delivery, and gates only that delivery's Delivery Entry.** The Delivery Entry is the -message instance the delivery admitted into the Command Pipeline. An internal type, `InboxDeliveryEntry`, binds it on -the delivery's `ContextContainer`: the first message to reach `InboxBehavior` on a delivery's context becomes the -Delivery Entry, and only a message that is that same instance is sent through `ReceiveViaInbox`. Every other message -dispatched on that context, at any depth and by any dispatcher, goes straight to `next()`. +## Decision -**A recovery retry is gated again.** The receiver deserializes the payload once (`BrokeredMessageReceiver.cs:1081`) and -its recovery strategy re-dispatches that same instance on the same context (`:1089-1094`). The retry therefore -presents the Delivery Entry again, and the Inbox gates it again, exactly as it did before this change. +**The Inbox deduplicates a delivery, and gates only the Delivery Entry of each receive attempt.** The Delivery Entry +is the first message instance the delivery's current receive attempt admitted into the Command Pipeline. An internal +type, `InboxDeliveryEntry`, binds it on the delivery's `ContextContainer`, and the receiver installs a fresh +`InboxDeliveryEntry` there at the start of every Recovery attempt (`BrokeredMessageReceiver.BeginReceiveAttempt`). The +first message to reach `InboxBehavior` on the delivery's context in an attempt becomes that attempt's Delivery Entry, +and only a message that is that same instance is sent through `ReceiveViaInbox`. Every other message dispatched on that +context in the attempt, at any depth and by any dispatcher, goes straight to `next()`. + +**A recovery retry is gated again, whoever constructs the Command.** The receiver runs every Recovery attempt on the +same context (`BrokeredMessageReceiver.cs:1108-1113`), and each attempt first calls `BeginReceiveAttempt` +(`:1069-1081`), which installs a fresh Delivery Entry before the attempt dispatches anything. The first Command a retry +dispatches is therefore that attempt's Delivery Entry, and the Inbox gates it again. That holds whether the Command is +the payload the receiver deserialized once (`:1100`) or a fresh Command that a handler or an overriding receiver builds +on each attempt. An entry bound by an earlier attempt would never admit such a fresh Command, so the retry would bypass +the Inbox. **One edit fixes all three tiers.** The change is in `InboxBehavior`, above every store. No store's contract, `IBrokeredMessageInbox.ReceiveViaInbox`, or any store implementation changes. @@ -144,8 +167,23 @@ Pinned by these facts in `src/Chatter.MessageBrokers/tests/Reliability/Inbox/Usi - `MustReceiveViaInboxOnlyTheMessageTheDeliveryAdmitted`: only the Delivery Entry is sent through `ReceiveViaInbox`. - `MustInvokeTheHandlerOfASiblingDispatchedAfterTheAdmittedMessageCompleted`: a Command dispatched on the same context after the Delivery Entry's handler returned still runs. -- `MustReceiveViaInboxAgainWhenTheAdmittedMessageIsRedispatchedAfterItsHandlerFailed`: a recovery retry of the - Delivery Entry is gated again. +- `MustReceiveViaInboxAgainWhenTheAdmittedMessageIsRedispatchedAfterItsHandlerFailed`: the instance the entry bound, + dispatched again on the same context after its handler failed, is gated again. + +**ELIMINATED CLASS: "a receive attempt inherits gate state bound by an earlier attempt of the same delivery".** +`BeginReceiveAttempt` installs a fresh `InboxDeliveryEntry` on the delivery's container at the start of every Recovery +attempt, before the attempt dispatches anything, so no attempt can find an entry that an earlier attempt bound. The seam +is in `ProcessMessageAsync`, above the virtual `DispatchReceivedMessageAsync`, so a receiver that overrides that +method, as `ChangeFeedReceiver` does, is covered by construction. + +Pinned by these facts in +`src/Chatter.MessageBrokers/tests/Receiving/UsingBrokeredMessageReceiver/WhenGatingTheInboxAcrossRecoveryAttempts.cs`, +which drive the real receiver loop, a real retry strategy and a real `InboxBehavior`: + +- `MustGateTheFirstCommandOfEveryRecoveryAttemptWhenTheHandlerBuildsAFreshOne`: when the handler builds a fresh + Command on every attempt, each attempt's Command is sent through `ReceiveViaInbox`. +- `MustGateTheFirstCommandOfEveryAttemptWhenAReceiverOverridesTheDispatch`: the same holds when a receiver overrides + `DispatchReceivedMessageAsync` and builds its own Command on every attempt. ## Consequences @@ -159,24 +197,36 @@ Pinned by these facts in `src/Chatter.MessageBrokers/tests/Reliability/Inbox/Usi - **The Inbox is not a loop guard.** A handler that dispatches a new Command of its own type now dispatches it for real on every tier, so a handler that does so unconditionally recurses without end. The Entity Framework and Cosmos tiers already did this; the in-memory tier used to stop it by accident, by dropping the nested Command. -- **`InboxDeliveryEntry` takes no lock.** The context container is unsynchronized by decision (ADR-0011), and a - delivery's nested dispatches are awaited one at a time on that delivery's context. +- **`InboxDeliveryEntry` takes no lock.** The context container is unsynchronized by decision (ADR-0011). One worker + runs a delivery's Recovery attempts one after another, and an attempt's nested dispatches are awaited one at a time + on that delivery's context. - **`ChangeFeedReceiver` fans one delivery out as many Events on one context.** `DispatchReceivedMessageAsync` (`src/Chatter.SqlChangeFeed/src/Chatter.SqlChangeFeed/ChangeFeedReceiver.cs:35-79`) dispatches one Event per changed - row on the delivery's context (`:57`, `:63`, `:69`). The first Command any of those Events' handlers dispatches - takes the Delivery Entry and is gated; every later one runs. Before this change every later one was gated under the - same message id and, once the first one's receipt existed, skipped on every tier. + row on the delivery's context (`:57`, `:63`, `:69`). In each receive attempt, the first Command any of those Events' + handlers dispatches takes that attempt's Delivery Entry and is gated; every later one in the attempt runs. A retry + starts with a fresh entry, so its first Command is gated again. Before this change every later one was gated under + the same message id and, once the first one's receipt existed, skipped on every tier. ### Recorded residuals These are decisions, not open work. -- **R1: a broker-received Event is not itself deduplicated, so the first Command its handler dispatches takes the - Delivery Entry.** `InboxBehavior` is a Command behaviour and never sees the Event. When the delivered payload is an - Event, the first Command dispatched on the delivery's context is the first message to reach the Inbox, and it is - gated under the delivery's message id. This is inherited, not introduced: that Command was gated before this change - as well. It is bounded: later sibling Commands now run, and nothing is silently dropped. The obvious remedy, gating - at the receive seam, was rejected as Option 2. No issue is filed for it. +- **R1: a broker-received Event is not itself deduplicated, so the first Command its handlers dispatch in each receive + attempt takes that attempt's Delivery Entry.** `InboxBehavior` is a Command behaviour and never sees the Event. When + the delivered payload is an Event, the first Command dispatched on the delivery's context in a receive attempt is the + first message to reach the Inbox in that attempt, and it is gated under the delivery's message id. This is inherited, + not introduced: that Command was gated before this change as well. It is bounded: later sibling Commands in the + attempt now run, and nothing is silently dropped. The obvious remedy, gating at the receive seam, was rejected as + Option 2. No issue is filed for it. +- **R2: a direct call to `DispatchReceivedMessageAsync` bypasses the attempt seam.** `DispatchReceivedMessageAsync` is + public and virtual (`BrokeredMessageReceiver.cs:1035`), and `BeginReceiveAttempt` runs in `ProcessMessageAsync` + before it, not inside it. A caller that invokes it directly installs no fresh entry, so `InboxBehavior` reuses + whatever `InboxDeliveryEntry` the context already holds, or creates one through `GetOrNew` when the context holds + none. The root cause is that the seam that scopes the entry to an attempt belongs to the receive loop, and the public + dispatch method can be reached without passing through it. It was raised as finding 2cc65ed4 in the local review of + #534. It is bounded: it needs a caller that drives the dispatch by hand and reuses one delivery context across + calls, and even then the gate behaves as it did in the previous revision of this change, where the entry was scoped + to the delivery. It is never worse than that revision. No issue is filed for it. Related work that is tracked: the Cosmos document tier has its own gate with the same root class, a nested participant Command being treated as the delivery, and it is not changed here. That is #538. @@ -204,7 +254,11 @@ Command being treated as the delivery, and it is not changed here. That is #538. (`:34-35`) and `GetInboundBrokeredMessage()` (`:161-169`). - `src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Sending/InMemoryDispatcher.cs` — nested dispatch on the handler's own context (`:14-22`). -- `src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Receiving/BrokeredMessageReceiver.cs` — the transaction - context placed on the delivery's container (`:1077`), the single deserialization (`:1081`) and the recovery - re-dispatch (`:1089-1094`). -- `src/Chatter.MessageBrokers/tests/Reliability/Inbox/UsingInboxBehavior/WhenHandling.cs` — the oracles. +- `src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Receiving/BrokeredMessageReceiver.cs` — the public virtual + `DispatchReceivedMessageAsync` (`:1035`), `BeginReceiveAttempt` (`:1069-1081`), which installs each attempt's fresh + Delivery Entry (`:1080`), the transaction context placed on the delivery's container (`:1096`), the single + deserialization (`:1100`) and the Recovery attempt that calls `BeginReceiveAttempt` before the dispatch + (`:1108-1113`). +- `src/Chatter.MessageBrokers/tests/Reliability/Inbox/UsingInboxBehavior/WhenHandling.cs` — the gate's oracles. +- `src/Chatter.MessageBrokers/tests/Receiving/UsingBrokeredMessageReceiver/WhenGatingTheInboxAcrossRecoveryAttempts.cs` + — the attempt-scope oracles. diff --git a/src/Chatter.MessageBrokers/CONTEXT.md b/src/Chatter.MessageBrokers/CONTEXT.md index 7b6ed24b..c4010621 100644 --- a/src/Chatter.MessageBrokers/CONTEXT.md +++ b/src/Chatter.MessageBrokers/CONTEXT.md @@ -109,7 +109,7 @@ _Avoid_: "Reliability-Pair Lifetime Contract" (the per-pair DI forwarding mechan **ReceiveViaInbox / InboxBehavior**: the `ReceiveViaInbox(() => next())` wrap seam is relational-only mechanics. The inbox **contract** and intent (once-only dedup, idempotency) are shared across tiers; the seam that implements them is provider-specific. The document tier enlists the inbox marker via the Document-Tier Batch-Lifecycle Behavior and the document-tier reliability surface — not through `ReceiveViaInbox`. What the wrap seam gates is the Delivery Entry, not every command that merely carries the delivery's context (ADR-0041). -**Delivery Entry**: the one message per delivery the Inbox gates — the message the delivery admitted into the Command Pipeline. Every message dispatched in-process inside that delivery (`context.InMemory()`) is inside the same delivery, not a new one, and passes the Inbox through. A recovery retry re-presents the admitted message, so it is gated again. Pinned by `WhenHandling.MustInvokeTheNestedHandlerWhenAGatedHandlersOwnDispatchReEntersTheBehavior`, `.MustReceiveViaInboxOnlyTheMessageTheDeliveryAdmitted`, `.MustInvokeTheHandlerOfASiblingDispatchedAfterTheAdmittedMessageCompleted` and `.MustReceiveViaInboxAgainWhenTheAdmittedMessageIsRedispatchedAfterItsHandlerFailed`. See ADR-0041. +**Delivery Entry**: the one message per receive attempt the Inbox gates — the first message the Inbox sees on the delivery's context in the delivery's current receive attempt. Every message dispatched in-process inside that delivery (`context.InMemory()`) is inside the same delivery, not a new one, and passes the Inbox through. Each receive attempt has its own Delivery Entry, which the Brokered Message Receiver installs before the attempt dispatches anything, so a recovery retry is gated again even when the handler builds a fresh Command. Pinned by `WhenHandling.MustInvokeTheNestedHandlerWhenAGatedHandlersOwnDispatchReEntersTheBehavior`, `.MustReceiveViaInboxOnlyTheMessageTheDeliveryAdmitted`, `.MustInvokeTheHandlerOfASiblingDispatchedAfterTheAdmittedMessageCompleted` and `.MustReceiveViaInboxAgainWhenTheAdmittedMessageIsRedispatchedAfterItsHandlerFailed`, and by `WhenGatingTheInboxAcrossRecoveryAttempts.MustGateTheFirstCommandOfEveryRecoveryAttemptWhenTheHandlerBuildsAFreshOne` and `.MustGateTheFirstCommandOfEveryAttemptWhenAReceiverOverridesTheDispatch`. See ADR-0041. **Document-Tier Reliability Surface**: the provider-shaped reliability surface the document (NoSQL) tier owns. It carries the document-store primitives the relational-shaped `TransactionContext` does not and cannot carry: the resolved partition key, bound container, `TransactionalBatch` handle and batch lifecycle, inbox-marker enlistment, and ETag concurrency token. The document tier is the doc-tier sibling of `IPersistanceTransaction` for the atomic-write handle; the ETag concurrency token lives here, not on `TransactionContext`. The surface lives entirely in the Cosmos provider module — not in core. _Avoid_: treating `TransactionContext.Container` as the semantic home of document-tier primitives (it is an untyped dict that unifies storage, not semantics; no partition concept exists in core). diff --git a/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/CHANGELOG.md b/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/CHANGELOG.md index e93c54e6..94576bf5 100644 --- a/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/CHANGELOG.md +++ b/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/CHANGELOG.md @@ -16,7 +16,7 @@ This project follows [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) an ### Fixed -- **A Command a received handler dispatches in-process now runs.** `InboxBehavior<>` gated every Command dispatched with the delivery's context, so a Command a handler dispatched with `context.InMemory()` looked like the delivery itself. The in-memory Inbox silently skipped it; the relational Inbox re-claimed it and could skip a second one; the standalone Cosmos Inbox completed the delivery's marker before the outer handler finished, so a later failure of the outer handler could be lost on redelivery. The Inbox now gates only the message the delivery admitted (a recovery retry of that message is still gated). No code change is needed. The Inbox is not a loop guard: a handler that dispatches a new Command back into itself now loops on every Inbox. ADR-0041 (#534). +- **A Command a received handler dispatches in-process now runs.** `InboxBehavior<>` gated every Command dispatched with the delivery's context, so a Command a handler dispatched with `context.InMemory()` looked like the delivery itself. The in-memory Inbox silently skipped it; the relational Inbox re-claimed it and could skip a second one; the standalone Cosmos Inbox completed the delivery's marker before the outer handler finished, so a later failure of the outer handler could be lost on redelivery. The Inbox now gates only the first message each receive attempt admits (each retry attempt's first Command is gated again, even when the handler builds a fresh one). No code change is needed. The Inbox is not a loop guard: a handler that dispatches a new Command back into itself now loops on every Inbox. ADR-0041 (#534). - **A code comment on `InMemoryBrokeredMessageOutbox` that described `OutboxProcessor` as stamping a dispatched message's processed date BEFORE dispatching is corrected to say AFTER dispatching, matching the code it describes.** This carries no code change and no user-facing effect (#534). ## [0.35.2] - 2026-09-25 diff --git a/src/Chatter.MessageBrokers/src/README.md b/src/Chatter.MessageBrokers/src/README.md index 2f0820bb..af139787 100644 --- a/src/Chatter.MessageBrokers/src/README.md +++ b/src/Chatter.MessageBrokers/src/README.md @@ -466,7 +466,7 @@ The Inbox reserves the message id before your handler runs, and a concurrent del - its id was evicted at the entry cap; - the process restarted, or the redelivery reached another instance, because each process keeps its own in-memory Inbox. -`InboxBehavior<>` acts on a command dispatched with an `IMessageBrokerContext`, which a Brokered Message Receiver supplies; a command dispatched from outside a received handler passes through. It gates only the **Delivery Entry** — the message the delivery admitted into the Command Pipeline, the first message it sees on that delivery's context; a recovery retry re-presents that same message and is gated again. A Command your handler dispatches in-process with `context.InMemory()` belongs to the same delivery, so it passes through the Inbox and runs. A broker-received event is not itself deduplicated, since the Inbox is a Command behavior; the first Command its handler dispatches in-process is the one the Inbox gates, and later ones run. See [ADR-0041](https://github.com/brenpike/Chatter/blob/master/docs/adr/0041-the-inbox-gates-the-message-the-delivery-admitted-not-every-command-that-carries-its-context.md). A received message without a `MessageId` makes the in-memory Inbox throw `ArgumentException`, and the handler does not run. +`InboxBehavior<>` acts on a command dispatched with an `IMessageBrokerContext`, which a Brokered Message Receiver supplies; a command dispatched from outside a received handler passes through. It gates only the **Delivery Entry** — the first message it sees on that delivery's context in the current receive attempt. Each receive attempt has its own Delivery Entry, so a retry is gated again even when the handler builds a fresh Command. A Command your handler dispatches in-process with `context.InMemory()` belongs to the same delivery, so it passes through the Inbox and runs. A broker-received event is not itself deduplicated, since the Inbox is a Command behavior; the first Command its handler dispatches in-process in each receive attempt is the one the Inbox gates, and later ones in that attempt run. See [ADR-0041](https://github.com/brenpike/Chatter/blob/master/docs/adr/0041-the-inbox-gates-the-message-the-delivery-admitted-not-every-command-that-carries-its-context.md). A received message without a `MessageId` makes the in-memory Inbox throw `ArgumentException`, and the handler does not run. The Inbox is not a loop guard: a handler that dispatches back into itself in-process belongs to the same delivery each time, so the dispatch passes through and the handler will now loop on every inbox tier. From 6bbf247d3cb39890ab6a24b0f831c2749fb019cb Mon Sep 17 00:00:00 2001 From: brenpike Date: Fri, 25 Sep 2026 20:06:10 -0600 Subject: [PATCH 08/11] docs(messagebrokers): state the interim inbox guarantee and point to #539 (#534) --- ...-every-command-that-carries-its-context.md | 71 +++++++++++-------- src/Chatter.MessageBrokers/CONTEXT.md | 2 +- .../src/Chatter.MessageBrokers/CHANGELOG.md | 2 +- .../Receiving/BrokeredMessageReceiver.cs | 5 +- .../Reliability/Inbox/InboxBehavior.cs | 5 +- src/Chatter.MessageBrokers/src/README.md | 2 +- 6 files changed, 52 insertions(+), 35 deletions(-) diff --git a/docs/adr/0041-the-inbox-gates-the-message-the-delivery-admitted-not-every-command-that-carries-its-context.md b/docs/adr/0041-the-inbox-gates-the-message-the-delivery-admitted-not-every-command-that-carries-its-context.md index ebfd40f8..901fae5e 100644 --- a/docs/adr/0041-the-inbox-gates-the-message-the-delivery-admitted-not-every-command-that-carries-its-context.md +++ b/docs/adr/0041-the-inbox-gates-the-message-the-delivery-admitted-not-every-command-that-carries-its-context.md @@ -116,10 +116,14 @@ delivered payload is an Event the payload never reaches the Inbox, and no Comman be gated at all. The same holds for every `ChangeFeedReceiver` delivery: its override of `DispatchReceivedMessageAsync` never dispatches the payload and dispatches Events built from it instead. Those deliveries would lose the deduplication they had on master, where the first Command their handlers dispatched was gated under the delivery's -message id (see R1). +message id (see G2). ## Decision +**This design is interim.** It is superseded when the design of epic #539 lands. The real fix is an idempotent +receiver at the delivery boundary: the unit of work, the Inbox and the outbox wrapped around the whole delivery. Until +then, the guarantee this design gives is stated, with its limits, under *Scope limits*. + **The Inbox deduplicates a delivery, and gates only the Delivery Entry of each receive attempt.** The Delivery Entry is the first message instance the delivery's current receive attempt admitted into the Command Pipeline. An internal type, `InboxDeliveryEntry`, binds it on the delivery's `ContextContainer`, and the receiver installs a fresh @@ -202,41 +206,48 @@ which drive the real receiver loop, a real retry strategy and a real `InboxBehav on that delivery's context. - **`ChangeFeedReceiver` fans one delivery out as many Events on one context.** `DispatchReceivedMessageAsync` (`src/Chatter.SqlChangeFeed/src/Chatter.SqlChangeFeed/ChangeFeedReceiver.cs:35-79`) dispatches one Event per changed - row on the delivery's context (`:57`, `:63`, `:69`). In each receive attempt, the first Command any of those Events' - handlers dispatches takes that attempt's Delivery Entry and is gated; every later one in the attempt runs. A retry - starts with a fresh entry, so its first Command is gated again. Before this change every later one was gated under - the same message id and, once the first one's receipt existed, skipped on every tier. - -### Recorded residuals - -These are decisions, not open work. - -- **R1: a broker-received Event is not itself deduplicated, so the first Command its handlers dispatch in each receive - attempt takes that attempt's Delivery Entry.** `InboxBehavior` is a Command behaviour and never sees the Event. When - the delivered payload is an Event, the first Command dispatched on the delivery's context in a receive attempt is the - first message to reach the Inbox in that attempt, and it is gated under the delivery's message id. This is inherited, - not introduced: that Command was gated before this change as well. It is bounded: later sibling Commands in the - attempt now run, and nothing is silently dropped. The obvious remedy, gating at the receive seam, was rejected as - Option 2. No issue is filed for it. -- **R2: a direct call to `DispatchReceivedMessageAsync` bypasses the attempt seam.** `DispatchReceivedMessageAsync` is - public and virtual (`BrokeredMessageReceiver.cs:1035`), and `BeginReceiveAttempt` runs in `ProcessMessageAsync` - before it, not inside it. A caller that invokes it directly installs no fresh entry, so `InboxBehavior` reuses - whatever `InboxDeliveryEntry` the context already holds, or creates one through `GetOrNew` when the context holds - none. The root cause is that the seam that scopes the entry to an attempt belongs to the receive loop, and the public - dispatch method can be reached without passing through it. It was raised as finding 2cc65ed4 in the local review of - #534. It is bounded: it needs a caller that drives the dispatch by hand and reuses one delivery context across - calls, and even then the gate behaves as it did in the previous revision of this change, where the entry was scoped - to the delivery. It is never worse than that revision. No issue is filed for it. - -Related work that is tracked: the Cosmos document tier has its own gate with the same root class, a nested participant -Command being treated as the delivery, and it is not changed here. That is #538. + row on the delivery's context (`:57`, `:63`, `:69`). What the Inbox guarantees for those deliveries is G2, under + *Scope limits*. + +### Scope limits + +This design is interim (see *Decision*). It covers a delivery fully only when the delivered payload is a Command. The +limits below are closed by epic #539's idempotent receiver, not by further work on this design. + +- **G1: a delivered Command, and every Command it dispatches in-process, is fully covered.** The payload is the + attempt's Delivery Entry, and nested Commands join the same unit of work. Pinned by the facts listed under + *Closed-by-Construction Acceptance Test*. +- **G2: a delivered Event is not itself deduplicated.** The first Command its handler(s) dispatch in each receive + attempt is gated; later Commands in that attempt are at-least-once on retry or redelivery (strictly better than + master, which silently dropped them). This covers `ChangeFeedReceiver`, which fans one delivery into N Events on one + context. `InboxBehavior` is a Command behaviour and never sees the Event, so the first Command dispatched on the + delivery's context in a receive attempt is the first message to reach the Inbox in that attempt, and it is gated + under the delivery's message id. A retry starts with a fresh entry, so its first Command is gated again. This is + inherited, not introduced: that Command was gated before this change as well. It is bounded: later sibling Commands + in the attempt now run, and nothing is silently dropped. Gating at the receive seam alone was rejected as Option 2, + because the unit of work opens inside the Command Pipeline. Tracked by epic #539. +- **G3: a direct call to the public virtual `DispatchReceivedMessageAsync` bypasses `BeginReceiveAttempt`, so it + reuses whatever entry the context holds, outside the guarantee.** `DispatchReceivedMessageAsync` is public and + virtual (`BrokeredMessageReceiver.cs:1035`), and `BeginReceiveAttempt` runs in `ProcessMessageAsync` before it, not + inside it. A caller that invokes it directly installs no fresh entry, so `InboxBehavior` reuses whatever + `InboxDeliveryEntry` the context already holds, or creates one through `GetOrNew` when the context holds none. The + root cause is that the seam that scopes the entry to an attempt belongs to the receive loop, and the public dispatch + method can be reached without passing through it. It was raised as finding 2cc65ed4 in the local review of #534. It + is bounded: it needs a caller that drives the dispatch by hand and reuses one delivery context across calls, and + even then the gate behaves as it did in the previous revision of this change, where the entry was scoped to the + delivery. It is never worse than that revision. No test pins it. Tracked by epic #539. + +Related work: the Cosmos document tier has its own gate with the same root class, a nested participant Command being +treated as the delivery, and it is not changed here. That was #538, which is folded into epic #539. ## References - Issue #534 — *In-memory Inbox silently skips a Command dispatched with context.InMemory() from inside a received handler*. The defect this ADR resolves. - Issue #538 — *Cosmos document tier silently discards a nested participant command's writes*. The same root class on - the document tier, tracked separately. + the document tier. Folded into epic #539. +- Epic #539 — *Idempotent receiver: move the unit of work, inbox and outbox processing to the delivery boundary*. The + design that supersedes this one and closes G2 and G3. - ADR-0006 — *Two-tier reliability*. The relational canonical behaviour order this decision leaves unchanged, and the inbox-plus-handler atomicity inside the unit of work that rules out Option 2. - ADR-0008 — *Document-tier participation model*. The document tier's own gate, which this decision does not change. diff --git a/src/Chatter.MessageBrokers/CONTEXT.md b/src/Chatter.MessageBrokers/CONTEXT.md index c4010621..73b942fe 100644 --- a/src/Chatter.MessageBrokers/CONTEXT.md +++ b/src/Chatter.MessageBrokers/CONTEXT.md @@ -109,7 +109,7 @@ _Avoid_: "Reliability-Pair Lifetime Contract" (the per-pair DI forwarding mechan **ReceiveViaInbox / InboxBehavior**: the `ReceiveViaInbox(() => next())` wrap seam is relational-only mechanics. The inbox **contract** and intent (once-only dedup, idempotency) are shared across tiers; the seam that implements them is provider-specific. The document tier enlists the inbox marker via the Document-Tier Batch-Lifecycle Behavior and the document-tier reliability surface — not through `ReceiveViaInbox`. What the wrap seam gates is the Delivery Entry, not every command that merely carries the delivery's context (ADR-0041). -**Delivery Entry**: the one message per receive attempt the Inbox gates — the first message the Inbox sees on the delivery's context in the delivery's current receive attempt. Every message dispatched in-process inside that delivery (`context.InMemory()`) is inside the same delivery, not a new one, and passes the Inbox through. Each receive attempt has its own Delivery Entry, which the Brokered Message Receiver installs before the attempt dispatches anything, so a recovery retry is gated again even when the handler builds a fresh Command. Pinned by `WhenHandling.MustInvokeTheNestedHandlerWhenAGatedHandlersOwnDispatchReEntersTheBehavior`, `.MustReceiveViaInboxOnlyTheMessageTheDeliveryAdmitted`, `.MustInvokeTheHandlerOfASiblingDispatchedAfterTheAdmittedMessageCompleted` and `.MustReceiveViaInboxAgainWhenTheAdmittedMessageIsRedispatchedAfterItsHandlerFailed`, and by `WhenGatingTheInboxAcrossRecoveryAttempts.MustGateTheFirstCommandOfEveryRecoveryAttemptWhenTheHandlerBuildsAFreshOne` and `.MustGateTheFirstCommandOfEveryAttemptWhenAReceiverOverridesTheDispatch`. See ADR-0041. +**Delivery Entry**: the one message per receive attempt the Inbox gates — the first message the Inbox sees on the delivery's context in the delivery's current receive attempt. Every message dispatched in-process inside that delivery (`context.InMemory()`) is inside the same delivery, not a new one, and passes the Inbox through. Each receive attempt has its own Delivery Entry, which the Brokered Message Receiver installs before the attempt dispatches anything, so a recovery retry is gated again even when the handler builds a fresh Command. Pinned by `WhenHandling.MustInvokeTheNestedHandlerWhenAGatedHandlersOwnDispatchReEntersTheBehavior`, `.MustReceiveViaInboxOnlyTheMessageTheDeliveryAdmitted`, `.MustInvokeTheHandlerOfASiblingDispatchedAfterTheAdmittedMessageCompleted` and `.MustReceiveViaInboxAgainWhenTheAdmittedMessageIsRedispatchedAfterItsHandlerFailed`, and by `WhenGatingTheInboxAcrossRecoveryAttempts.MustGateTheFirstCommandOfEveryRecoveryAttemptWhenTheHandlerBuildsAFreshOne` and `.MustGateTheFirstCommandOfEveryAttemptWhenAReceiverOverridesTheDispatch`. See ADR-0041. What this gate covers: a delivered Command, and every Command it dispatches in-process, is fully covered. A delivered Event is not itself deduplicated; the first Command its handler(s) dispatch in each receive attempt is the Delivery Entry, and a later Command in that same attempt is at-least-once on retry or redelivery rather than gated. A direct call to the public `DispatchReceivedMessageAsync` is outside this guarantee, since it does not start a new receive attempt. Deduplicating the whole delivery rather than only its Delivery Entry is tracked as #539. **Document-Tier Reliability Surface**: the provider-shaped reliability surface the document (NoSQL) tier owns. It carries the document-store primitives the relational-shaped `TransactionContext` does not and cannot carry: the resolved partition key, bound container, `TransactionalBatch` handle and batch lifecycle, inbox-marker enlistment, and ETag concurrency token. The document tier is the doc-tier sibling of `IPersistanceTransaction` for the atomic-write handle; the ETag concurrency token lives here, not on `TransactionContext`. The surface lives entirely in the Cosmos provider module — not in core. _Avoid_: treating `TransactionContext.Container` as the semantic home of document-tier primitives (it is an untyped dict that unifies storage, not semantics; no partition concept exists in core). diff --git a/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/CHANGELOG.md b/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/CHANGELOG.md index 94576bf5..7354896f 100644 --- a/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/CHANGELOG.md +++ b/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/CHANGELOG.md @@ -16,7 +16,7 @@ This project follows [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) an ### Fixed -- **A Command a received handler dispatches in-process now runs.** `InboxBehavior<>` gated every Command dispatched with the delivery's context, so a Command a handler dispatched with `context.InMemory()` looked like the delivery itself. The in-memory Inbox silently skipped it; the relational Inbox re-claimed it and could skip a second one; the standalone Cosmos Inbox completed the delivery's marker before the outer handler finished, so a later failure of the outer handler could be lost on redelivery. The Inbox now gates only the first message each receive attempt admits (each retry attempt's first Command is gated again, even when the handler builds a fresh one). No code change is needed. The Inbox is not a loop guard: a handler that dispatches a new Command back into itself now loops on every Inbox. ADR-0041 (#534). +- **A Command a received handler dispatches in-process now runs.** `InboxBehavior<>` gated every Command dispatched with the delivery's context, so a Command a handler dispatched with `context.InMemory()` looked like the delivery itself. The in-memory Inbox silently skipped it; the relational Inbox re-claimed it and could skip a second one; the standalone Cosmos Inbox completed the delivery's marker before the outer handler finished, so a later failure of the outer handler could be lost on redelivery. The Inbox now gates only the first message each receive attempt admits (each retry attempt's first Command is gated again, even when the handler builds a fresh one). No code change is needed. The Inbox is not a loop guard: a handler that dispatches a new Command back into itself now loops on every Inbox. For a delivered Event whose handler(s) dispatch several Commands in one receive attempt, only the first is deduplicated; later ones are at-least-once on retry or redelivery. A direct call to `DispatchReceivedMessageAsync` is outside this guarantee. Deduplicating the whole delivery is tracked in #539. ADR-0041 (#534). - **A code comment on `InMemoryBrokeredMessageOutbox` that described `OutboxProcessor` as stamping a dispatched message's processed date BEFORE dispatching is corrected to say AFTER dispatching, matching the code it describes.** This carries no code change and no user-facing effect (#534). ## [0.35.2] - 2026-09-25 diff --git a/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Receiving/BrokeredMessageReceiver.cs b/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Receiving/BrokeredMessageReceiver.cs index 32622cea..415a1eb3 100644 --- a/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Receiving/BrokeredMessageReceiver.cs +++ b/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Receiving/BrokeredMessageReceiver.cs @@ -1076,7 +1076,10 @@ private static void BeginReceiveAttempt(MessageBrokerContext messageContext) // entry is installed here, not inside CountReceiveAttempt, because that returns early when BrokerDiagnostics // has no listeners. Pinned by WhenGatingTheInboxAcrossRecoveryAttempts: deleting this Include reddens // MustGateTheFirstCommandOfEveryRecoveryAttemptWhenTheHandlerBuildsAFreshOne and - // MustGateTheFirstCommandOfEveryAttemptWhenAReceiverOverridesTheDispatch. + // MustGateTheFirstCommandOfEveryAttemptWhenAReceiverOverridesTheDispatch. The seam is in + // ProcessMessageAsync, above the public virtual DispatchReceivedMessageAsync, so a caller that invokes that + // method directly installs no fresh entry and is outside the guarantee (ADR-0041 G3; see #539); no test + // pins that. messageContext.Container.Include(new InboxDeliveryEntry()); } diff --git a/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/Inbox/InboxBehavior.cs b/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/Inbox/InboxBehavior.cs index 151ae36d..9f76b2a7 100644 --- a/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/Inbox/InboxBehavior.cs +++ b/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/Inbox/InboxBehavior.cs @@ -29,7 +29,10 @@ public Task Handle(TMessage message, IMessageHandlerContext messageHandlerContex // delivery's message id (ADR-0041). The entry is attempt-scoped: the receiver installs a fresh one at the // start of every Recovery attempt (BrokeredMessageReceiver.BeginReceiveAttempt), so a retry that constructs // a fresh command is gated again; GetOrNew still creates one lazily for a dispatch that never passed that - // seam. Pinned by WhenGatingTheInboxAcrossRecoveryAttempts: deleting the Include in BeginReceiveAttempt + // seam. When the delivered payload is an Event the Inbox never sees it, so the attempt's first Command + // takes the entry and later sibling Commands in that attempt are at-least-once on retry or redelivery + // (ADR-0041 G2; see #539). No test pins that bound. + // Pinned by WhenGatingTheInboxAcrossRecoveryAttempts: deleting the Include in BeginReceiveAttempt // reddens MustGateTheFirstCommandOfEveryRecoveryAttemptWhenTheHandlerBuildsAFreshOne and // MustGateTheFirstCommandOfEveryAttemptWhenAReceiverOverridesTheDispatch. Pinned by WhenHandling: restoring the bare `is IMessageBrokerContext` gate // reddens MustInvokeTheNestedHandlerWhenAGatedHandlersOwnDispatchReEntersTheBehavior, diff --git a/src/Chatter.MessageBrokers/src/README.md b/src/Chatter.MessageBrokers/src/README.md index af139787..255fdbd3 100644 --- a/src/Chatter.MessageBrokers/src/README.md +++ b/src/Chatter.MessageBrokers/src/README.md @@ -466,7 +466,7 @@ The Inbox reserves the message id before your handler runs, and a concurrent del - its id was evicted at the entry cap; - the process restarted, or the redelivery reached another instance, because each process keeps its own in-memory Inbox. -`InboxBehavior<>` acts on a command dispatched with an `IMessageBrokerContext`, which a Brokered Message Receiver supplies; a command dispatched from outside a received handler passes through. It gates only the **Delivery Entry** — the first message it sees on that delivery's context in the current receive attempt. Each receive attempt has its own Delivery Entry, so a retry is gated again even when the handler builds a fresh Command. A Command your handler dispatches in-process with `context.InMemory()` belongs to the same delivery, so it passes through the Inbox and runs. A broker-received event is not itself deduplicated, since the Inbox is a Command behavior; the first Command its handler dispatches in-process in each receive attempt is the one the Inbox gates, and later ones in that attempt run. See [ADR-0041](https://github.com/brenpike/Chatter/blob/master/docs/adr/0041-the-inbox-gates-the-message-the-delivery-admitted-not-every-command-that-carries-its-context.md). A received message without a `MessageId` makes the in-memory Inbox throw `ArgumentException`, and the handler does not run. +`InboxBehavior<>` acts on a command dispatched with an `IMessageBrokerContext`, which a Brokered Message Receiver supplies; a command dispatched from outside a received handler passes through. It gates only the **Delivery Entry** — the first message it sees on that delivery's context in the current receive attempt. Each receive attempt has its own Delivery Entry, so a retry is gated again even when the handler builds a fresh Command. A Command your handler dispatches in-process with `context.InMemory()` belongs to the same delivery, so it passes through the Inbox and runs. A delivered Command, and every Command it dispatches in-process, is therefore fully covered. A broker-received event is not itself deduplicated, since the Inbox is a Command behavior; the first Command its handler(s) dispatch in-process in each receive attempt is the one the Inbox gates, and a later Command in that same attempt is at-least-once on retry or redelivery rather than gated. A direct call to the receiver's public `DispatchReceivedMessageAsync` is outside this guarantee, since it does not start a new receive attempt. See [ADR-0041](https://github.com/brenpike/Chatter/blob/master/docs/adr/0041-the-inbox-gates-the-message-the-delivery-admitted-not-every-command-that-carries-its-context.md); deduplicating the whole delivery rather than only its Delivery Entry is tracked in [#539](https://github.com/brenpike/Chatter/issues/539). A received message without a `MessageId` makes the in-memory Inbox throw `ArgumentException`, and the handler does not run. The Inbox is not a loop guard: a handler that dispatches back into itself in-process belongs to the same delivery each time, so the dispatch passes through and the handler will now loop on every inbox tier. From 481831564c1818dab2b0d4c0e00e036d2ac990fb Mon Sep 17 00:00:00 2001 From: brenpike Date: Fri, 25 Sep 2026 20:21:43 -0600 Subject: [PATCH 09/11] docs(messagebrokers): state the Delivery Entry's coverage and threading limits per tier (#534) --- ...-every-command-that-carries-its-context.md | 20 ++++++++++++++----- .../Reliability/Inbox/InboxDeliveryEntry.cs | 11 +++++++--- 2 files changed, 23 insertions(+), 8 deletions(-) diff --git a/docs/adr/0041-the-inbox-gates-the-message-the-delivery-admitted-not-every-command-that-carries-its-context.md b/docs/adr/0041-the-inbox-gates-the-message-the-delivery-admitted-not-every-command-that-carries-its-context.md index 901fae5e..8e1e1736 100644 --- a/docs/adr/0041-the-inbox-gates-the-message-the-delivery-admitted-not-every-command-that-carries-its-context.md +++ b/docs/adr/0041-the-inbox-gates-the-message-the-delivery-admitted-not-every-command-that-carries-its-context.md @@ -202,8 +202,12 @@ which drive the real receiver loop, a real retry strategy and a real `InboxBehav on every tier, so a handler that does so unconditionally recurses without end. The Entity Framework and Cosmos tiers already did this; the in-memory tier used to stop it by accident, by dropping the nested Command. - **`InboxDeliveryEntry` takes no lock.** The context container is unsynchronized by decision (ADR-0011). One worker - runs a delivery's Recovery attempts one after another, and an attempt's nested dispatches are awaited one at a time - on that delivery's context. + running a delivery's Recovery attempts one after another is a construction fact; an attempt's nested dispatches + being awaited one at a time on that delivery's context is not — ADR-0011 places that on the application as a + requirement it deliberately does not enforce, and records that a missing `await` is enough to break it. An + application that breaks it corrupts the container's own dictionary before this entry's field is ever read, so the + entry adds no exposure the container does not already carry, and reopening the decision belongs to #333 rather than + here. - **`ChangeFeedReceiver` fans one delivery out as many Events on one context.** `DispatchReceivedMessageAsync` (`src/Chatter.SqlChangeFeed/src/Chatter.SqlChangeFeed/ChangeFeedReceiver.cs:35-79`) dispatches one Event per changed row on the delivery's context (`:57`, `:63`, `:69`). What the Inbox guarantees for those deliveries is G2, under @@ -214,9 +218,15 @@ which drive the real receiver loop, a real retry strategy and a real `InboxBehav This design is interim (see *Decision*). It covers a delivery fully only when the delivered payload is a Command. The limits below are closed by epic #539's idempotent receiver, not by further work on this design. -- **G1: a delivered Command, and every Command it dispatches in-process, is fully covered.** The payload is the - attempt's Delivery Entry, and nested Commands join the same unit of work. Pinned by the facts listed under - *Closed-by-Construction Acceptance Test*. +- **G1: a delivered Command, and every Command it dispatches in-process, is fully covered by the delivery's own + inbox receipt.** The payload is the attempt's Delivery Entry, so no nested Command is ever gated on its own and the + delivery's receipt is what covers all of them. Coverage is what G1 claims; atomicity is a per-tier property and is + not. Only the relational tier runs those nested Commands inside the delivery's ambient transaction, so only there + does an outer failure roll their writes back (ADR-0006). The in-memory tier and the standalone Cosmos inbox register + no unit of work — `WithCosmosInbox` is cited for that at *Context* above — so an outer failure rolls nothing back + and a nested Command's effects re-run when the delivery is redelivered. That is the at-least-once posture ADR-0009 + already records for non-batched effects, and this design leaves it exactly as it was. Pinned by the facts listed + under *Closed-by-Construction Acceptance Test*. - **G2: a delivered Event is not itself deduplicated.** The first Command its handler(s) dispatch in each receive attempt is gated; later Commands in that attempt are at-least-once on retry or redelivery (strictly better than master, which silently dropped them). This covers `ChangeFeedReceiver`, which fans one delivery into N Events on one diff --git a/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/Inbox/InboxDeliveryEntry.cs b/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/Inbox/InboxDeliveryEntry.cs index 1dd1e59c..0c27b5f9 100644 --- a/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/Inbox/InboxDeliveryEntry.cs +++ b/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/Inbox/InboxDeliveryEntry.cs @@ -8,9 +8,14 @@ internal sealed class InboxDeliveryEntry { // INVARIANT: no lock. The entry lives in the delivery's ContextContainer, which is unsynchronized and must - // never be used from two threads at once (ADR-0011): one worker owns the delivery's container, runs its - // Recovery attempts one after another, and awaits an attempt's nested dispatches inside its handler, so the - // first Admits call has bound the attempt's entry before any other call on this context runs. No test pins it. + // never be used from two threads at once (ADR-0011). One worker owning the delivery's container and running + // its Recovery attempts one after another is a construction fact; a handler awaiting each nested dispatch + // before starting the next is NOT - ADR-0011 places that on the application as a requirement it deliberately + // does not enforce, and records that a missing await is enough to break it. An application that breaks it + // corrupts the container's own Dictionary before _admittedMessage is ever read, so this field adds no + // exposure the container does not already carry; ADR-0011 holds the rationale and #333 is where reopening it + // belongs. While that requirement holds, the first Admits call has bound the attempt's entry before any other + // call on this context runs. No test pins it. private object _admittedMessage; /// From 341a2271099403eb6fcfa3af7344e6215c62692a Mon Sep 17 00:00:00 2001 From: brenpike Date: Sat, 26 Sep 2026 10:05:55 -0600 Subject: [PATCH 10/11] docs(messagebrokers): record the outer-behaviour Delivery Entry limit as G4 (#534) --- ...-every-command-that-carries-its-context.md | 21 +++++++++++++++++-- 1 file changed, 19 insertions(+), 2 deletions(-) diff --git a/docs/adr/0041-the-inbox-gates-the-message-the-delivery-admitted-not-every-command-that-carries-its-context.md b/docs/adr/0041-the-inbox-gates-the-message-the-delivery-admitted-not-every-command-that-carries-its-context.md index 8e1e1736..66cd82cb 100644 --- a/docs/adr/0041-the-inbox-gates-the-message-the-delivery-admitted-not-every-command-that-carries-its-context.md +++ b/docs/adr/0041-the-inbox-gates-the-message-the-delivery-admitted-not-every-command-that-carries-its-context.md @@ -226,7 +226,9 @@ limits below are closed by epic #539's idempotent receiver, not by further work no unit of work — `WithCosmosInbox` is cited for that at *Context* above — so an outer failure rolls nothing back and a nested Command's effects re-run when the delivery is redelivered. That is the at-least-once posture ADR-0009 already records for non-batched effects, and this design leaves it exactly as it was. Pinned by the facts listed - under *Closed-by-Construction Acceptance Test*. + under *Closed-by-Construction Acceptance Test*. G4, below, qualifies this claim: when another behaviour dispatches a + Command on the delivery's context before `InboxBehavior` sees the delivered Command, coverage shifts to that other + Command instead. - **G2: a delivered Event is not itself deduplicated.** The first Command its handler(s) dispatch in each receive attempt is gated; later Commands in that attempt are at-least-once on retry or redelivery (strictly better than master, which silently dropped them). This covers `ChangeFeedReceiver`, which fans one delivery into N Events on one @@ -246,6 +248,21 @@ limits below are closed by epic #539's idempotent receiver, not by further work is bounded: it needs a caller that drives the dispatch by hand and reuses one delivery context across calls, and even then the gate behaves as it did in the previous revision of this change, where the entry was scoped to the delivery. It is never worse than that revision. No test pins it. Tracked by epic #539. +- **G4: a Command another behaviour dispatches before `InboxBehavior` sees the delivered Command takes the Delivery + Entry, and the delivered Command is then at-least-once.** A behaviour registered outside `InboxBehavior` in the + Command Pipeline that dispatches a Command on the delivery's context before it invokes `next()` sends that Command + through `InboxBehavior` first, and `Admits` binds the attempt's entry to it. The delivered Command then goes + straight to `next()` and never passes through `ReceiveViaInbox`. On redelivery the Inbox skips only the other + Command's receipt, and the delivered handler runs again. The root cause is that the entry binds on arrival order at + the Inbox, not on the message the receive seam admitted, and only the receive seam knows which message that is. It + is bounded: it needs an application behaviour ordered outside `InboxBehavior` that dispatches before `next()`. No + behaviour this package ships does so, and ADR-0006's relational canonical order puts `InboxBehavior` innermost, so + any application behaviour sits outside it. The cost is a duplicate run of the delivered handler, never lost work. + This is inherited, and strictly better than master: there, the nested Command took the delivery's claim and the + delivered Command was then read as a duplicate and skipped, so its work was lost. Seeding the entry with the + delivered payload at the receive seam was rejected as Option 7 above, because `ChangeFeedReceiver` never dispatches + its payload, so its deliveries would lose the deduplication they have today. Raised in the review of PR #540. No + test pins it. Tracked by epic #539. Related work: the Cosmos document tier has its own gate with the same root class, a nested participant Command being treated as the delivery, and it is not changed here. That was #538, which is folded into epic #539. @@ -257,7 +274,7 @@ treated as the delivery, and it is not changed here. That was #538, which is fol - Issue #538 — *Cosmos document tier silently discards a nested participant command's writes*. The same root class on the document tier. Folded into epic #539. - Epic #539 — *Idempotent receiver: move the unit of work, inbox and outbox processing to the delivery boundary*. The - design that supersedes this one and closes G2 and G3. + design that supersedes this one and closes G2, G3 and G4. - ADR-0006 — *Two-tier reliability*. The relational canonical behaviour order this decision leaves unchanged, and the inbox-plus-handler atomicity inside the unit of work that rules out Option 2. - ADR-0008 — *Document-tier participation model*. The document tier's own gate, which this decision does not change. From 7857a09e83bc64013b93fd8ebf2e0b9017cdf656 Mon Sep 17 00:00:00 2001 From: brenpike Date: Sat, 26 Sep 2026 10:07:33 -0600 Subject: [PATCH 11/11] docs(messagebrokers): state the G4 Delivery Entry limit in README, CONTEXT, changelog (#534) --- src/Chatter.MessageBrokers/CONTEXT.md | 2 +- .../src/Chatter.MessageBrokers/CHANGELOG.md | 2 +- .../Chatter.MessageBrokers/Reliability/Inbox/InboxBehavior.cs | 4 +++- src/Chatter.MessageBrokers/src/README.md | 2 +- 4 files changed, 6 insertions(+), 4 deletions(-) diff --git a/src/Chatter.MessageBrokers/CONTEXT.md b/src/Chatter.MessageBrokers/CONTEXT.md index 73b942fe..99c5fc9f 100644 --- a/src/Chatter.MessageBrokers/CONTEXT.md +++ b/src/Chatter.MessageBrokers/CONTEXT.md @@ -109,7 +109,7 @@ _Avoid_: "Reliability-Pair Lifetime Contract" (the per-pair DI forwarding mechan **ReceiveViaInbox / InboxBehavior**: the `ReceiveViaInbox(() => next())` wrap seam is relational-only mechanics. The inbox **contract** and intent (once-only dedup, idempotency) are shared across tiers; the seam that implements them is provider-specific. The document tier enlists the inbox marker via the Document-Tier Batch-Lifecycle Behavior and the document-tier reliability surface — not through `ReceiveViaInbox`. What the wrap seam gates is the Delivery Entry, not every command that merely carries the delivery's context (ADR-0041). -**Delivery Entry**: the one message per receive attempt the Inbox gates — the first message the Inbox sees on the delivery's context in the delivery's current receive attempt. Every message dispatched in-process inside that delivery (`context.InMemory()`) is inside the same delivery, not a new one, and passes the Inbox through. Each receive attempt has its own Delivery Entry, which the Brokered Message Receiver installs before the attempt dispatches anything, so a recovery retry is gated again even when the handler builds a fresh Command. Pinned by `WhenHandling.MustInvokeTheNestedHandlerWhenAGatedHandlersOwnDispatchReEntersTheBehavior`, `.MustReceiveViaInboxOnlyTheMessageTheDeliveryAdmitted`, `.MustInvokeTheHandlerOfASiblingDispatchedAfterTheAdmittedMessageCompleted` and `.MustReceiveViaInboxAgainWhenTheAdmittedMessageIsRedispatchedAfterItsHandlerFailed`, and by `WhenGatingTheInboxAcrossRecoveryAttempts.MustGateTheFirstCommandOfEveryRecoveryAttemptWhenTheHandlerBuildsAFreshOne` and `.MustGateTheFirstCommandOfEveryAttemptWhenAReceiverOverridesTheDispatch`. See ADR-0041. What this gate covers: a delivered Command, and every Command it dispatches in-process, is fully covered. A delivered Event is not itself deduplicated; the first Command its handler(s) dispatch in each receive attempt is the Delivery Entry, and a later Command in that same attempt is at-least-once on retry or redelivery rather than gated. A direct call to the public `DispatchReceivedMessageAsync` is outside this guarantee, since it does not start a new receive attempt. Deduplicating the whole delivery rather than only its Delivery Entry is tracked as #539. +**Delivery Entry**: the one message per receive attempt the Inbox gates — the first message the Inbox sees on the delivery's context in the delivery's current receive attempt. Every message dispatched in-process inside that delivery (`context.InMemory()`) is inside the same delivery, not a new one, and passes the Inbox through. Each receive attempt has its own Delivery Entry, which the Brokered Message Receiver installs before the attempt dispatches anything, so a recovery retry is gated again even when the handler builds a fresh Command. Pinned by `WhenHandling.MustInvokeTheNestedHandlerWhenAGatedHandlersOwnDispatchReEntersTheBehavior`, `.MustReceiveViaInboxOnlyTheMessageTheDeliveryAdmitted`, `.MustInvokeTheHandlerOfASiblingDispatchedAfterTheAdmittedMessageCompleted` and `.MustReceiveViaInboxAgainWhenTheAdmittedMessageIsRedispatchedAfterItsHandlerFailed`, and by `WhenGatingTheInboxAcrossRecoveryAttempts.MustGateTheFirstCommandOfEveryRecoveryAttemptWhenTheHandlerBuildsAFreshOne` and `.MustGateTheFirstCommandOfEveryAttemptWhenAReceiverOverridesTheDispatch`. See ADR-0041. What this gate covers: a delivered Command, and every Command it dispatches in-process, is fully covered. A delivered Event is not itself deduplicated; the first Command its handler(s) dispatch in each receive attempt is the Delivery Entry, and a later Command in that same attempt is at-least-once on retry or redelivery rather than gated. A direct call to the public `DispatchReceivedMessageAsync` is outside this guarantee, since it does not start a new receive attempt. A Command another behavior dispatches on the delivery's context before `InboxBehavior` sees the delivered Command becomes that attempt's Delivery Entry instead, so the delivered Command is at-least-once on retry or redelivery; no Chatter-shipped behavior dispatches before `next()` (ADR-0041 G4). Deduplicating the whole delivery rather than only its Delivery Entry is tracked as #539. **Document-Tier Reliability Surface**: the provider-shaped reliability surface the document (NoSQL) tier owns. It carries the document-store primitives the relational-shaped `TransactionContext` does not and cannot carry: the resolved partition key, bound container, `TransactionalBatch` handle and batch lifecycle, inbox-marker enlistment, and ETag concurrency token. The document tier is the doc-tier sibling of `IPersistanceTransaction` for the atomic-write handle; the ETag concurrency token lives here, not on `TransactionContext`. The surface lives entirely in the Cosmos provider module — not in core. _Avoid_: treating `TransactionContext.Container` as the semantic home of document-tier primitives (it is an untyped dict that unifies storage, not semantics; no partition concept exists in core). diff --git a/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/CHANGELOG.md b/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/CHANGELOG.md index 7354896f..f29b3bef 100644 --- a/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/CHANGELOG.md +++ b/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/CHANGELOG.md @@ -16,7 +16,7 @@ This project follows [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) an ### Fixed -- **A Command a received handler dispatches in-process now runs.** `InboxBehavior<>` gated every Command dispatched with the delivery's context, so a Command a handler dispatched with `context.InMemory()` looked like the delivery itself. The in-memory Inbox silently skipped it; the relational Inbox re-claimed it and could skip a second one; the standalone Cosmos Inbox completed the delivery's marker before the outer handler finished, so a later failure of the outer handler could be lost on redelivery. The Inbox now gates only the first message each receive attempt admits (each retry attempt's first Command is gated again, even when the handler builds a fresh one). No code change is needed. The Inbox is not a loop guard: a handler that dispatches a new Command back into itself now loops on every Inbox. For a delivered Event whose handler(s) dispatch several Commands in one receive attempt, only the first is deduplicated; later ones are at-least-once on retry or redelivery. A direct call to `DispatchReceivedMessageAsync` is outside this guarantee. Deduplicating the whole delivery is tracked in #539. ADR-0041 (#534). +- **A Command a received handler dispatches in-process now runs.** `InboxBehavior<>` gated every Command dispatched with the delivery's context, so a Command a handler dispatched with `context.InMemory()` looked like the delivery itself. The in-memory Inbox silently skipped it; the relational Inbox re-claimed it and could skip a second one; the standalone Cosmos Inbox completed the delivery's marker before the outer handler finished, so a later failure of the outer handler could be lost on redelivery. The Inbox now gates only the first message each receive attempt admits (each retry attempt's first Command is gated again, even when the handler builds a fresh one). No code change is needed. The Inbox is not a loop guard: a handler that dispatches a new Command back into itself now loops on every Inbox. For a delivered Event whose handler(s) dispatch several Commands in one receive attempt, only the first is deduplicated; later ones are at-least-once on retry or redelivery. A direct call to `DispatchReceivedMessageAsync` is outside this guarantee. A behavior registered outside `InboxBehavior<>` that dispatches a command before it calls the next behavior takes the Delivery Entry, so the delivered command runs again on a retry or redelivery; no behavior this package ships does so. Deduplicating the whole delivery is tracked in #539. ADR-0041 (#534). - **A code comment on `InMemoryBrokeredMessageOutbox` that described `OutboxProcessor` as stamping a dispatched message's processed date BEFORE dispatching is corrected to say AFTER dispatching, matching the code it describes.** This carries no code change and no user-facing effect (#534). ## [0.35.2] - 2026-09-25 diff --git a/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/Inbox/InboxBehavior.cs b/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/Inbox/InboxBehavior.cs index 9f76b2a7..9d68545d 100644 --- a/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/Inbox/InboxBehavior.cs +++ b/src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Reliability/Inbox/InboxBehavior.cs @@ -31,7 +31,9 @@ public Task Handle(TMessage message, IMessageHandlerContext messageHandlerContex // a fresh command is gated again; GetOrNew still creates one lazily for a dispatch that never passed that // seam. When the delivered payload is an Event the Inbox never sees it, so the attempt's first Command // takes the entry and later sibling Commands in that attempt are at-least-once on retry or redelivery - // (ADR-0041 G2; see #539). No test pins that bound. + // (ADR-0041 G2; see #539). No test pins that bound. A behavior outside this one that dispatches on the + // delivery's context before calling next() binds the entry first, so the delivered Command goes straight + // to next() and is at-least-once on retry or redelivery (ADR-0041 G4; see #539). No test pins that bound. // Pinned by WhenGatingTheInboxAcrossRecoveryAttempts: deleting the Include in BeginReceiveAttempt // reddens MustGateTheFirstCommandOfEveryRecoveryAttemptWhenTheHandlerBuildsAFreshOne and // MustGateTheFirstCommandOfEveryAttemptWhenAReceiverOverridesTheDispatch. Pinned by WhenHandling: restoring the bare `is IMessageBrokerContext` gate diff --git a/src/Chatter.MessageBrokers/src/README.md b/src/Chatter.MessageBrokers/src/README.md index 255fdbd3..82448c9d 100644 --- a/src/Chatter.MessageBrokers/src/README.md +++ b/src/Chatter.MessageBrokers/src/README.md @@ -466,7 +466,7 @@ The Inbox reserves the message id before your handler runs, and a concurrent del - its id was evicted at the entry cap; - the process restarted, or the redelivery reached another instance, because each process keeps its own in-memory Inbox. -`InboxBehavior<>` acts on a command dispatched with an `IMessageBrokerContext`, which a Brokered Message Receiver supplies; a command dispatched from outside a received handler passes through. It gates only the **Delivery Entry** — the first message it sees on that delivery's context in the current receive attempt. Each receive attempt has its own Delivery Entry, so a retry is gated again even when the handler builds a fresh Command. A Command your handler dispatches in-process with `context.InMemory()` belongs to the same delivery, so it passes through the Inbox and runs. A delivered Command, and every Command it dispatches in-process, is therefore fully covered. A broker-received event is not itself deduplicated, since the Inbox is a Command behavior; the first Command its handler(s) dispatch in-process in each receive attempt is the one the Inbox gates, and a later Command in that same attempt is at-least-once on retry or redelivery rather than gated. A direct call to the receiver's public `DispatchReceivedMessageAsync` is outside this guarantee, since it does not start a new receive attempt. See [ADR-0041](https://github.com/brenpike/Chatter/blob/master/docs/adr/0041-the-inbox-gates-the-message-the-delivery-admitted-not-every-command-that-carries-its-context.md); deduplicating the whole delivery rather than only its Delivery Entry is tracked in [#539](https://github.com/brenpike/Chatter/issues/539). A received message without a `MessageId` makes the in-memory Inbox throw `ArgumentException`, and the handler does not run. +`InboxBehavior<>` acts on a command dispatched with an `IMessageBrokerContext`, which a Brokered Message Receiver supplies; a command dispatched from outside a received handler passes through. It gates only the **Delivery Entry** — the first message it sees on that delivery's context in the current receive attempt. Each receive attempt has its own Delivery Entry, so a retry is gated again even when the handler builds a fresh Command. A Command your handler dispatches in-process with `context.InMemory()` belongs to the same delivery, so it passes through the Inbox and runs. A delivered Command, and every Command it dispatches in-process, is therefore fully covered. A broker-received event is not itself deduplicated, since the Inbox is a Command behavior; the first Command its handler(s) dispatch in-process in each receive attempt is the one the Inbox gates, and a later Command in that same attempt is at-least-once on retry or redelivery rather than gated. A direct call to the receiver's public `DispatchReceivedMessageAsync` is outside this guarantee, since it does not start a new receive attempt. A behavior you register outside `InboxBehavior<>` that dispatches a command before it calls the next behavior sends that command through the Inbox first; that command then becomes the Delivery Entry, and the delivered command runs again on a retry or redelivery. The behaviors this package ships dispatch nothing before the next behavior. See [ADR-0041](https://github.com/brenpike/Chatter/blob/master/docs/adr/0041-the-inbox-gates-the-message-the-delivery-admitted-not-every-command-that-carries-its-context.md); deduplicating the whole delivery rather than only its Delivery Entry is tracked in [#539](https://github.com/brenpike/Chatter/issues/539). A received message without a `MessageId` makes the in-memory Inbox throw `ArgumentException`, and the handler does not run. The Inbox is not a loop guard: a handler that dispatches back into itself in-process belongs to the same delivery each time, so the dispatch passes through and the handler will now loop on every inbox tier.