| | | 1 | | namespace AsyncResponse.Transports; |
| | | 2 | | |
| | | 3 | | // Shared source for every transport's outer BackgroundService.ExecuteAsync supervise-and-retry |
| | | 4 | | // loop: each csproj pulls this file in via <Compile Include="..\Shared\SubscriberSupervisor.cs" />, |
| | | 5 | | // so it compiles INTO each provider assembly. Per-transport setup that must run once before the |
| | | 6 | | // loop starts (topology warnings, resolving a queue/subscription name) stays in the caller, ahead |
| | | 7 | | // of the call into RunAsync below; only the retry-with-backoff loop shape is shared. |
| | | 8 | | |
| | | 9 | | /// <summary> |
| | | 10 | | /// Runs a subscriber's connect-and-consume operation with retry-with-backoff on failure. Any |
| | | 11 | | /// exception observed once host shutdown has been requested exits quietly — the cancellation |
| | | 12 | | /// itself, and equally the ObjectDisposed/connection-closed errors broker clients raise when the |
| | | 13 | | /// stop tears a connection down mid-consume; any other exception — including a cancellation NOT |
| | | 14 | | /// caused by host shutdown, e.g. a transport-internal timeout — increments the failure count, asks |
| | | 15 | | /// the caller-supplied delay policy how long to wait, reports the retry through the caller-supplied |
| | | 16 | | /// callback, and waits before trying again. The count is of CONSECUTIVE failures: a run that stayed |
| | | 17 | | /// up at least as long as the policy's saturated delay before failing starts the count over. |
| | | 18 | | /// </summary> |
| | | 19 | | internal static class SubscriberSupervisor |
| | | 20 | | { |
| | | 21 | | /// <summary> |
| | | 22 | | /// Runs <paramref name="run"/> until it completes or <paramref name="stoppingToken"/> requests |
| | | 23 | | /// shutdown. <paramref name="delayPolicy"/> receives the 1-based consecutive-failure count and |
| | | 24 | | /// returns how long to wait before the next attempt; <paramref name="logRetry"/> renders the |
| | | 25 | | /// per-transport log line for that wait. The policy must saturate: it is also asked for |
| | | 26 | | /// <see cref="int.MaxValue"/> failures, and that answer — its longest delay — is the healthy-run |
| | | 27 | | /// threshold past which a failed run no longer counts as consecutive with the one before it. |
| | | 28 | | /// <paramref name="timeProvider"/> clocks both the run and the wait (a test seam; the system |
| | | 29 | | /// clock when omitted). |
| | | 30 | | /// </summary> |
| | | 31 | | public static async Task RunAsync( |
| | | 32 | | Func<CancellationToken, Task> run, |
| | | 33 | | CancellationToken stoppingToken, |
| | | 34 | | Func<int, TimeSpan> delayPolicy, |
| | | 35 | | Action<Exception, TimeSpan> logRetry, |
| | | 36 | | TimeProvider? timeProvider = null) |
| | | 37 | | { |
| | 414 | 38 | | var clock = timeProvider ?? TimeProvider.System; |
| | 414 | 39 | | var failures = 0; |
| | 414 | 40 | | TimeSpan? healthyRun = null; |
| | 465 | 41 | | while (!stoppingToken.IsCancellationRequested) |
| | | 42 | | { |
| | 465 | 43 | | var startedAt = clock.GetTimestamp(); |
| | | 44 | | try |
| | | 45 | | { |
| | 465 | 46 | | await run(stoppingToken).ConfigureAwait(false); |
| | 412 | 47 | | return; |
| | | 48 | | } |
| | 53 | 49 | | catch (Exception) when (stoppingToken.IsCancellationRequested) |
| | | 50 | | { |
| | | 51 | | // Shutdown. Not only the OperationCanceledException: a client whose connection the |
| | | 52 | | // stop just closed throws its own ObjectDisposed/AlreadyClosed/connection error |
| | | 53 | | // instead, and with no arm for that shape it escaped ExecuteAsync — faulting the |
| | | 54 | | // BackgroundService (a critical "BackgroundService failed" log, and StopApplication |
| | | 55 | | // under the default behavior) on every clean redeploy. |
| | 0 | 56 | | return; |
| | | 57 | | } |
| | | 58 | | catch (Exception ex) |
| | | 59 | | { |
| | | 60 | | // Consecutive, not lifetime: a subscriber runs for weeks, and with a count that only |
| | | 61 | | // ever grew, a handful of unrelated blips spread over that time pinned EVERY later |
| | | 62 | | // reconnect at the policy's longest delay — a one-second broker failover then cost |
| | | 63 | | // the full maximum on every replica, for the rest of the process's life. A run that |
| | | 64 | | // outlived the longest delay the policy can impose was healthy, so this failure |
| | | 65 | | // starts a new streak. (A run that merely took that long to fail — a black-holed |
| | | 66 | | // connect — resets too, harmlessly: its own duration already paces the retries.) |
| | 53 | 67 | | if (failures > 0 && clock.GetElapsedTime(startedAt) >= (healthyRun ??= delayPolicy(int.MaxValue))) |
| | 0 | 68 | | failures = 0; |
| | | 69 | | |
| | 53 | 70 | | failures++; |
| | 53 | 71 | | var retryDelay = delayPolicy(failures); |
| | 53 | 72 | | logRetry(ex, retryDelay); |
| | 53 | 73 | | await Task.Delay(retryDelay, clock, stoppingToken).ConfigureAwait(false); |
| | | 74 | | } |
| | | 75 | | } |
| | 412 | 76 | | } |
| | | 77 | | } |