Repository navigation
fix: bound durable claim to free capacity (stop stale-reclaim under horizontal scaling) - #1
Merged
Merged
Conversation
added 10 commits
June 20, 2026 17:25
Add two failing tests that demonstrate the competing-consumers defect: FetchUnprocessedAsync reads rows under FOR UPDATE SKIP LOCKED but does not claim them in the same transaction, so every concurrent poller fetches the same Pending row and the message is delivered N times instead of once. - MsgFlux.Postgres.Tests/ConcurrentDeliveryTests: store-level, N concurrent fetches of one Pending row all return it. - MsgFlux.Integration.Tests (new project): end-to-end, 8 fully-wired instances sharing one PostgreSQL database handle a single message 8 times. Both assert exactly-once delivery and will pass once fetch becomes an atomic claim-on-fetch.
FetchUnprocessedAsync now claims rows in the same statement that reads them (WITH ... FOR UPDATE SKIP LOCKED ... UPDATE SET state=Processing RETURNING), so concurrent pollers across instances never receive the same row. This restores the competing-consumers invariant: a message is handled exactly once by the group instead of once per instance. - PostgresMessageStore: claim-on-fetch via CTE + UPDATE ... RETURNING. - PollingStoreSource: drop the now-redundant deferred claim (OnProcessing / pending-claims batch); keep in-flight dedup and batched acks. - Tests: InMemoryMessageStore fake mirrors claim-on-fetch; remove tests that asserted the deferred-claim behaviour; adjust store assertions to observe the post-claim Processing state. Note: processed_at (stale-processing clock) now starts at fetch, so size StaleProcessingTimeout above the engine's per-batch drain time. Both regression tests added in the previous commit now pass.
Claim-on-fetch made the batched-claim primitive dead: nothing calls it since PollingStoreSource stopped deferring claims. Drop it from the contract and every implementation/fake. MarkAsProcessingAsync (single) stays as a contract primitive still exercised by the store tests; AcknowledgeBatchAsync keeps using the shared UpdateStateBatchAsync helper.
Every instance runs its own MessagePurgeService timer, so under horizontal scaling N instances issued the same DELETE concurrently. PurgeCompletedAsync now wraps the delete in a transaction guarded by pg_try_advisory_xact_lock: one instance purges per cycle, the rest skip without blocking (the lock is non-blocking and auto-releases on commit/rollback). The key is exposed as PurgeAdvisoryLockKey so operators can avoid collisions. Adds a deterministic test: while an external transaction holds the lock the purge returns 0; once released it deletes as expected.
Claim-on-fetch stamps processed_at for a whole batch at fetch time, but the engine dispatches those rows gradually (gated by MaxDegreeOfParallelism). Rows waiting in the dispatch queue pass StaleProcessingTimeout while still owned, so another instance re-claims and re-runs them. Three integration scenarios (worker claims the batch, stealers grab the stale tail), all with per-message handle time UNDER the stale timeout so the cause is the claim/dispatch decoupling, not a slow consumer: - Scenario1: large batch + MaxDOP=1 -> 8 msgs handled 16 times. - Scenario2: fast consumer (100ms << 400ms) -> 8 msgs handled 16 times (queue wait alone). - Scenario3: publish spike, 3 instances -> 16 msgs handled ~48 times. distinct == messageCount in all three (no loss), total > distinct (duplicates). Kept fast (~8s for all three, stale=400ms) by waiting actively until every message is covered, then a short grace window, instead of fixed multi-second sleeps. The trigger has margin: (messageCount/maxDop)*handle is several times the stale timeout, so the worker cannot drain before its rows go stale. These fail today and should pass once the reservation window is decoupled from batch-claim time.
FetchUnprocessedAsync claims rows on fetch (state -> Processing, processed_at = now). Reserving a whole PollingBatchSize up front stamped tail rows long before they reached a free slot, so they passed StaleProcessingTimeout while still queued and another instance re-claimed them, processing the consumer twice. Bound the claim to free capacity (MaxDOP - local in-flight) so processed_at stays close to actual dispatch time; the stale clause then only fires for genuine crashes. Pin MaxDOP in the rapid-poll dedup test so capacity stays positive with one in-flight message regardless of host core count.
At-least-once may deliver a message more than once (crash after handle before ack, or a slow handle reclaimed past StaleProcessingTimeout), so consumers must be idempotent. Note that keeping StaleProcessingTimeout above the longest handle avoids legitimate re-delivery of in-progress work.
With one slot occupied by an in-flight message, the source must claim no further rows: only the in-flight row is Processing, the rest stay Pending. Pins MaxDOP=1 so the proof holds regardless of host core count. Without capacity-gating the source would claim the whole batch on first fetch.
…apacity The durable claim is bounded by free capacity (MaxDOP - in-flight), which is <= MaxDOP and thus always below the old default of 500. min(PollingBatchSize, capacity) therefore always resolved to capacity, making PollingBatchSize inert in any sane config (it only bit when set below MaxDOP, which just throttles below the configured parallelism). MaxDegreeOfParallelism already expresses how much is processed at once, so the redundant knob is removed: fetch claims exactly the free capacity.
imihalcea
force-pushed
the
fix/atomic-claim-on-fetch
branch
from
June 20, 2026 20:53
88f0201 to
807f54c
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Problème
Le claim-on-fetch atomique (point 1, déjà sur cette branche) a éliminé le double-dispatch systématique mais réintroduit un double traitement conditionnel : stale-reclaim.
FetchUnprocessedAsyncclaim jusqu'à 500 lignes d'un coup, toutes stampéesprocessed_at = T0, alors que l'engine les dispatche progressivement (sémaphoreMaxDOP). Les lignes en fin de file dépassentStaleProcessingTimeoutavant même d'être dispatchées → une autre instance les voitProcessing+ périmées, les re-claim et les retraite (exactly-once-par-consumer violé).Fix — capacity-gated fetch (piste A, alignée River/Oban)
Borner le claim à la capacité libre dans
PollingStoreSource:processed_atcolle alors à l'instant de dispatch → la clause stale existante ne se déclenche plus que sur de vrais crashes. Aucun SQL modifié, pas de nouveau composant.Simplification — retrait de
PollingBatchSizeLe claim étant désormais borné par
MaxDOP − in-flight(toujours ≤ MaxDOP, donc ≪ l'ancien défaut 500),min(PollingBatchSize, capacity)valait toujourscapacity. L'option était inerte dans toute config saine (elle ne « mordait » qu'en dessous deMaxDOP, ce qui ne fait que brider le débit).MaxDegreeOfParallelismexprime déjà l'intention → option retirée (changement d'API public :PollingBatchSize+WithPollingBatchSize).Contrat documenté
Semantics.AtLeastOnce: le consumer doit être idempotent ; garderStaleProcessingTimeoutau-dessus du handle le plus long évite la re-livraison légitime de travail en cours.Tests
StaleReclaimTests(3 scénarios) : passent désormais.SingleSlot_Claims_Only_One_Row_While_It_Is_InFlight: prouve le capacity-gating sur un slot (MaxDOP=1), indépendamment du nombre de cœurs.Limitation connue (documentée en commentaire)
_inFlight.Countne compte que la source durable, alors que le sémaphore est partagé avecInMemoryMessageSource. En charge mixte AtMostOnce+AtLeastOnce lourde, léger sur-claim possible ; le fix exact (source observant le sémaphore partagé) est un redesign reporté.Note
Cette branche inclut aussi un commit de refactor sans rapport direct : déplacement des types de configuration vers le namespace
MsgFlux.Core.Configuration.Commits
fix: bound durable claim to free capacity to stop stale-reclaimdocs: document at-least-once idempotency contracttest: prove capacity-gated claim on a single slot (MaxDOP=1)refactor: move configuration types into MsgFlux.Core.Configuration namespacerefactor: remove PollingBatchSize; durable claim is bounded by free capacity