Dropped() spawns a resubscribe loop per failed message — concurrent handler failures multiply into a herd of duplicate subscriptions

[alexeyzimarev]

Summary

EventSubscription.Dropped() spawns a new Task.Run(Resubscribe) on every invocation, with no guard against a resubscribe already being in flight. On the async handling path, Dropped() fires per failed message (NackOnAsyncWorker in EventSubscriptionWithCheckpoint), and the physical message pump is not stopped when a handler-side drop occurs. When a batch of K queued messages all fail (for example a handler that deliberately throws to defer processing via redelivery), these three behaviors compound into a multiplicative herd of concurrent resubscribe loops and duplicate physical subscriptions.

Observed on 0.16.5-alpha.0.7; the relevant code on current dev is identical (EventSubscription.Dropped / EventSubscription.Resubscribe, EventSubscriptionWithCheckpoint.NackOnAsyncWorker).

Mechanism

1. A catch-up subscription (AllStreamSubscription, ThrowOnError = true, default async pipe) delivers a batch of K events into the AsyncHandlingFilter channel. The handler throws for each of them.
2. Each failure surfaces through Nack → throw → NackOnAsyncWorker catches and calls Dropped(DropReason.SubscriptionError, …). Dropped only checks IsRunning (which stays true), so all K failures fire the full drop path: K log lines, K invocations of _onDropped, and K Task.Run(Resubscribe) tasks. The pump keeps feeding the channel the whole time because nothing stops the underlying subscription on a handler-side drop.
3. After the 2s delay, each of the K tasks enters Resubscribe's while (IsRunning && IsDropped …) loop and calls Subscribe. Now there are K concurrent physical subscriptions, all reading from the same (unadvanced) checkpoint, each re-delivering the same K events. If the handler still fails, that is up to K² failures → K² new resubscribe tasks in the next cycle. The IsDropped flag is shared and clobbered concurrently by all loops, and a loop whose re-check races a fresh drop re-subscribes again with no delay, so growth is limited only by timing.

The herd collapses on its own once the handler stops failing (IsDropped stays false, loops exit), but while it lasts:

• hundreds of interleaved Will resubscribe / Resubscribing / Resubscribed / commit-handler stop/start lines per second;
• one live gRPC $all subscription plus one CheckpointCommitHandler churn per loop;
• a correctness hazard: once the handler recovers, multiple concurrent duplicate subscriptions can deliver the same event to the handler concurrently before any of them commits a checkpoint. For idempotent projections this is waste; for reactors (handlers with side effects) it is a real double-processing race.

Production observation

A handler that throws to defer a small batch of events (redelivery-based backpressure) produced this per-second log line count for one subscription — each ~2s resubscribe cycle multiplies the loop population until the defer condition clears:

17:25:51    7
17:25:53   43
17:25:55  201
17:25:57    4   <- handler stopped failing, herd collapsed

Sample of the interleaved storm (many resubscribe cycles running concurrently within the same milliseconds):

17:26:49.609 WRN [judge_fact_dedup] Will resubscribe after 00:00:02
17:26:49.609 INF [judge_fact_dedup] Resubscribed
17:26:49.609 INF [judge_fact_dedup] Resubscribed
17:26:49.609 INF [judge_fact_dedup] Resubscribed
17:26:49.611 WRN [judge_fact_dedup] Will resubscribe after 00:00:02
17:26:49.614 WRN [judge_fact_dedup] Will resubscribe after 00:00:02
17:26:49.615 WRN [judge_fact_dedup] Resubscribing
17:26:49.616 INF [judge_fact_dedup] Stopping commit handler worker
17:26:49.617 INF [judge_fact_dedup] Commit handler worker stopped
...

Repro sketch

1. Register an AllStreamSubscription with ThrowOnError = true and a single handler.
2. Append ~10 matching events.
3. Make the handler throw for every event for ~10 seconds, then succeed.
4. Watch the subscription log: instead of one orderly drop → 2s → resubscribe cycle, the number of Will resubscribe/Resubscribed lines grows multiplicatively per cycle, and after recovery the handler observes the same events delivered multiple times, some concurrently.

Suggested fix

• Make the drop → resubscribe transition single-flight: an interlocked "resubscribe pending" guard in Dropped() so only the first failure of a cycle spawns the task; subsequent Dropped() calls while one is pending only record the reason.
• Stop/dispose the still-live physical subscription (message pump) as part of handling a handler-side drop, before resubscribing, so a dropped-but-alive pump can't keep generating failures and the old and new subscriptions never overlap.
• In Resubscribe, clear IsDropped (or swap the guard) before awaiting Subscribe, or re-check under the same single-flight guard, so a drop racing the loop's re-check can't cause a zero-delay re-subscribe from two places at once.

Happy to send a PR to dev if the approach sounds right.

[linear]

EVE-243