Inbox gates only the delivery's own message (#534)

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<Task>` 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<Task>` 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<TService, TImplementation>(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 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).

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 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
 
 ### Fixed

src/Chatter.MessageBrokers/src/Chatter.MessageBrokers/Chatter.MessageBrokers.csproj

@@ -7,7 +7,7 @@
          TryEncodeUnicodeScalar) is char*-pointer based, so the overrides must be unsafe. -->
     <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
     <PackageId>Chatter.MessageBrokers</PackageId>
-    <Version>0.35.2</Version>
+    <Version>0.35.3</Version>
     <Authors>Brennan Pike</Authors>
     <Owners>Brennan Pike</Owners>
     <Description>A message broker adapter library built on top of Chatter.CQRS.</Description>

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,27 @@ public virtual async Task DispatchReceivedMessageAsync(TMessage payload, Message
             }
         }
 
+        /// <summary>
+        /// Begins one Recovery attempt at <paramref name="messageContext"/>: counts the attempt and installs a fresh
+        /// Delivery Entry, so the first command this attempt dispatches is the one the inbox gates.
+        /// </summary>
+        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. 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());
+        }
+
         async Task ProcessMessageAsync(MessageBrokerContext messageContext, TransactionContext transactionContext, CancellationToken receiverTokenSource)
         {
             receiverTokenSource.ThrowIfCancellationRequested();
@@ -1088,7 +1110,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);