From 1459f9407fa0a490cf9b0eb61651db5da5dfa7aa Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 8 Jul 2026 20:06:06 -0400 Subject: [PATCH] perf(store-and-forward): short-circuit a (category,target) lane after its first transient failure per sweep --- .../StoreAndForwardService.cs | 35 +++++++++++--- .../StoreAndForwardServiceTests.cs | 48 +++++++++++++++++++ 2 files changed, 76 insertions(+), 7 deletions(-) diff --git a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs index afeee811..d3ac792d 100644 --- a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs +++ b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs @@ -606,9 +606,19 @@ public class StoreAndForwardService _logger.LogDebug("Retry sweep: {Count} messages due for retry", messages.Count); + var failedTargets = new HashSet<(StoreAndForwardCategory, string)>(); foreach (var message in messages) { - await RetryMessageAsync(message); + // One transient failure per (category, target) per sweep: the target + // is down — burning a full timeout per remaining message serializes + // the sweep into hours under backlog (arch review 02, Performance #1). + // Skipped rows keep their RetryCount/LastAttemptAt untouched. + if (failedTargets.Contains((message.Category, message.Target))) + continue; + + var outcome = await RetryMessageAsync(message); + if (outcome == RetryOutcome.TransientFailure) + failedTargets.Add((message.Category, message.Target)); } } catch (Exception ex) @@ -621,12 +631,20 @@ public class StoreAndForwardService } } - private async Task RetryMessageAsync(StoreAndForwardMessage message) + /// + /// Outcome of a single message's retry attempt, used by the sweep to + /// short-circuit a (category, target) lane after its first transient + /// failure. = no attempt was made (no handler, or the + /// row's status changed under a concurrent apply); its RetryCount is untouched. + /// + internal enum RetryOutcome { Delivered, Parked, TransientFailure, Skipped } + + private async Task RetryMessageAsync(StoreAndForwardMessage message) { if (!_deliveryHandlers.TryGetValue(message.Category, out var handler)) { _logger.LogWarning("No delivery handler for category {Category}", message.Category); - return; + return RetryOutcome.Skipped; } // Measure per-attempt @@ -657,7 +675,7 @@ public class StoreAndForwardService httpStatus: null, occurredAtUtc: attemptStartUtc, durationMs: (int)attemptStopwatch.ElapsedMilliseconds); - return; + return RetryOutcome.Delivered; } // Permanent failure on retry — park immediately. @@ -671,7 +689,7 @@ public class StoreAndForwardService _logger.LogDebug( "Message {MessageId} changed status during delivery; sweep park skipped", message.Id); - return; + return RetryOutcome.Skipped; } Interlocked.Decrement(ref _bufferedCount); _replication?.ReplicatePark(message); @@ -687,6 +705,7 @@ public class StoreAndForwardService httpStatus: null, occurredAtUtc: attemptStartUtc, durationMs: (int)attemptStopwatch.ElapsedMilliseconds); + return RetryOutcome.Parked; } catch (Exception ex) { @@ -706,7 +725,7 @@ public class StoreAndForwardService _logger.LogDebug( "Message {MessageId} changed status during delivery; sweep park skipped", message.Id); - return; + return RetryOutcome.Skipped; } Interlocked.Decrement(ref _bufferedCount); _replication?.ReplicatePark(message); @@ -725,6 +744,7 @@ public class StoreAndForwardService httpStatus: null, occurredAtUtc: attemptStartUtc, durationMs: (int)attemptStopwatch.ElapsedMilliseconds); + return RetryOutcome.Parked; } else { @@ -734,7 +754,7 @@ public class StoreAndForwardService _logger.LogDebug( "Message {MessageId} changed status during delivery; sweep retry-count update skipped", message.Id); - return; + return RetryOutcome.Skipped; } RaiseActivity("Retried", message.Category, $"Retry {message.RetryCount}/{message.MaxRetries} for {message.Target}: {ex.Message}"); @@ -748,6 +768,7 @@ public class StoreAndForwardService httpStatus: null, occurredAtUtc: attemptStartUtc, durationMs: (int)attemptStopwatch.ElapsedMilliseconds); + return RetryOutcome.TransientFailure; } } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardServiceTests.cs b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardServiceTests.cs index 24454a27..7ffcaefe 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardServiceTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardServiceTests.cs @@ -638,4 +638,52 @@ public class StoreAndForwardServiceTests : IAsyncLifetime, IDisposable await _service.RetryPendingMessagesAsync(); // must not throw Assert.Equal(0, delivered); } + + // ── Task 8 (arch review 02, Performance): per-target short-circuit ── + // After the first transient failure to a (category, target) the remaining + // messages for that pair are skipped this sweep — N dead-target messages + // cost one timeout, not N. Skipped rows keep their RetryCount. + + [Fact] + public async Task RetrySweep_ShortCircuitsTarget_AfterFirstTransientFailure() + { + var attemptsByTarget = new Dictionary(); + _service.RegisterDeliveryHandler(StoreAndForwardCategory.ExternalSystem, msg => + { + attemptsByTarget[msg.Target] = attemptsByTarget.GetValueOrDefault(msg.Target) + 1; + if (msg.Target == "dead-target") throw new TimeoutException("down"); + return Task.FromResult(true); + }); + + for (var i = 0; i < 3; i++) + await _service.EnqueueAsync(StoreAndForwardCategory.ExternalSystem, "dead-target", "{}", + attemptImmediateDelivery: false, retryInterval: TimeSpan.Zero); + await _service.EnqueueAsync(StoreAndForwardCategory.ExternalSystem, "healthy-target", "{}", + attemptImmediateDelivery: false, retryInterval: TimeSpan.Zero); + + await _service.RetryPendingMessagesAsync(); + + Assert.Equal(1, attemptsByTarget["dead-target"]); // pre-fix: 3 + Assert.Equal(1, attemptsByTarget["healthy-target"]); // healthy lane unaffected + } + + [Fact] + public async Task RetrySweep_SkippedMessages_DoNotAccrueRetryCount() + { + _service.RegisterDeliveryHandler(StoreAndForwardCategory.ExternalSystem, + _ => throw new TimeoutException("down")); + var id1 = (await _service.EnqueueAsync(StoreAndForwardCategory.ExternalSystem, "dead", "{}", + maxRetries: 5, attemptImmediateDelivery: false, retryInterval: TimeSpan.Zero)).MessageId; + var id2 = (await _service.EnqueueAsync(StoreAndForwardCategory.ExternalSystem, "dead", "{}", + maxRetries: 5, attemptImmediateDelivery: false, retryInterval: TimeSpan.Zero)).MessageId; + + await _service.RetryPendingMessagesAsync(); + + var r1 = (await _service.GetMessageByIdAsync(id1))!.RetryCount; + var r2 = (await _service.GetMessageByIdAsync(id2))!.RetryCount; + // Exactly one was attempted (RetryCount 1); the other was skipped by the + // short-circuit, its RetryCount untouched (0). Order-independent so the + // assertion doesn't depend on created_at tie-breaking. + Assert.Equal(new[] { 0, 1 }, new[] { r1, r2 }.OrderBy(x => x).ToArray()); + } }