From c3a8576863690c03cbadf0920177f445d2368c46 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 8 Jul 2026 20:48:16 -0400 Subject: [PATCH] fix(notifications): park corrupt buffered notification payloads instead of silently discarding as delivered (supersedes StoreAndForward-018) --- .../requirements/Component-StoreAndForward.md | 2 ++ .../NotificationForwarder.cs | 25 +++++++++++++------ .../NotificationForwarderTests.cs | 25 ++++++++----------- 3 files changed, 30 insertions(+), 22 deletions(-) diff --git a/docs/requirements/Component-StoreAndForward.md b/docs/requirements/Component-StoreAndForward.md index 87f6cab6..4e14996f 100644 --- a/docs/requirements/Component-StoreAndForward.md +++ b/docs/requirements/Component-StoreAndForward.md @@ -51,6 +51,8 @@ flowchart TD For notifications, "delivery" means forwarding the message to the central cluster via Central–Site Communication; "success" is central's ack, on which the message is cleared. Notifications are retried at the fixed forward interval until central acks, but — like every other category — they are bounded by the engine's `DefaultMaxRetries` cap: a sustained central outage that exceeds `DefaultMaxRetries × forward-interval` will park the buffered notification, after which an operator can Retry/Discard it via the parked-message UI. Operationally, the cap is sized so the normal central-recovery window stays well inside it; "do not park" is the design's operational intent on the happy path, not an absolute invariant. Callers that genuinely require unbounded retry pass `maxRetries: 0` on `EnqueueAsync` (the documented "no limit" escape hatch — see `StoreAndForward-015`). +There are now exactly two ways a notification parks. First, a **corrupt buffered payload** (a row whose stored JSON no longer deserialises to a `NotificationSubmit`, or deserialises to null): the forwarder returns the delivery-handler contract's permanent-failure signal (`false`) so the engine parks the row, preserving the payload for operator forensics. This **supersedes `StoreAndForward-018`**, which discarded such rows by reporting them as delivered — a silent data loss with no parked row, no central `Notifications` row, and no audit trail. Retrying a corrupt payload is pointless (it can never deserialise), so parking, not retrying, is the correct home for it. Second, retry exhaustion under the `DefaultMaxRetries` cap as described above — and once a `Notify.Send` caller passes `maxRetries: 0` (its unbounded-retry contract), that second cause is off, leaving corrupt payload as the only parking cause for those notifications. + For the cached-call categories (`ExternalCall` and `DatabaseWrite`), the operation tracking table is the status record and the S&F buffer is purely the retry mechanism. A cached call that succeeds on its first immediate attempt is written directly as a terminal `Delivered` tracking row and never enters the S&F buffer. When immediate delivery fails transiently, the message is buffered and its tracking row moves to `Pending`/`Retrying`; the buffered message carries its `TrackedOperationId` so the tracking row and the retry record stay linked. When immediate delivery fails **permanently** (e.g. HTTP 4xx), the message is not buffered — the error is returned synchronously to the calling script as before — but the tracking row is written directly as a terminal `Failed` row capturing the error. On every tracking-table status transition the site emits `CachedCallTelemetry` to central. Every cached-call outcome maps to a tracking-table state: immediate success → `Delivered`; transient failure → `Pending`/`Retrying`, eventually `Delivered` or `Parked`; permanent failure → terminal `Failed`; operator discard of a parked row → terminal `Discarded`. diff --git a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/NotificationForwarder.cs b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/NotificationForwarder.cs index 71305c18..2b8500e9 100644 --- a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/NotificationForwarder.cs +++ b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/NotificationForwarder.cs @@ -20,6 +20,10 @@ namespace ZB.MOM.WW.ScadaBridge.StoreAndForward; /// ack not Accepted, or the Ask times out / fails → /// throws; the S&F engine treats any thrown /// exception as transient and retries the forward at the fixed interval. +/// the buffered payload is corrupt (undeserialisable / null) → +/// returns false (the handler contract's +/// permanent-failure signal); the engine parks the row so the payload is preserved +/// for operator forensics. Supersedes StoreAndForward-018's silent discard. /// /// /// The forward travels over the ClusterClient command/control transport: the handler @@ -51,8 +55,8 @@ public sealed class NotificationForwarder /// /// /// Optional logger. A corrupt buffered payload is logged at - /// Warning before being discarded so an operator has a forensic trail of the row - /// that vanished from the buffer. + /// Warning before the row is parked so an operator has a forensic trail pointing at + /// the preserved parked row. /// public NotificationForwarder( IActorRef siteCommunicationActor, @@ -69,21 +73,26 @@ public sealed class NotificationForwarder /// /// Store-and-Forward delivery handler entry point — matches the /// Func<StoreAndForwardMessage, Task<bool>> handler contract. - /// Returns true when central accepts the notification; throws on a - /// non-accepted ack or an Ask timeout/failure so the engine retries. + /// Returns true when central accepts the notification; returns false + /// when the buffered payload is corrupt (permanent failure → engine parks the row); + /// throws on a non-accepted ack or an Ask timeout/failure so the engine retries. /// /// The buffered store-and-forward message to deliver to central. - /// A task that resolves to true when central accepts (or the payload is corrupt and discarded); throws on a transient forward failure so the engine retries. + /// A task that resolves to true when central accepts; false when the payload is corrupt (the engine parks the row for forensics); throws on a transient forward failure so the engine retries. public async Task DeliverAsync(StoreAndForwardMessage message) { if (!TryBuildSubmit(message, out var submit)) { _logger.LogWarning( - "Discarding corrupt buffered notification {NotificationId} (payload is not deserialisable as NotificationSubmit). " + - "Payload preview: {PayloadPreview}", + "Parking corrupt buffered notification {NotificationId} (payload is not deserialisable as NotificationSubmit); " + + "the row is preserved for operator forensics. Payload preview: {PayloadPreview}", message.Id, PreviewPayload(message.PayloadJson)); - return true; + // false = permanent failure by the delivery-handler contract → the + // engine parks the row (payload preserved, operator can inspect or + // discard). Supersedes StoreAndForward-018's silent discard, which + // reported the notification as delivered while losing it entirely. + return false; } // The reply may legitimately be a non-accepted ack, so it is not requested as diff --git a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/NotificationForwarderTests.cs b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/NotificationForwarderTests.cs index 4c13806e..b6180556 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/NotificationForwarderTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/NotificationForwarderTests.cs @@ -199,16 +199,13 @@ public class NotificationForwarderTests : TestKit } [Fact] - public async Task Deliver_CorruptJsonPayload_ReturnsTrue_AndDoesNotForwardAnything() + public async Task Deliver_CorruptJsonPayload_ReturnsFalse_ParksForForensics_AndDoesNotForward() { - // Regression test for StoreAndForward-018. The design doc forbids parking - // notifications ("notifications do not park — they are retried at the fixed - // forward interval until central acks"; Component-StoreAndForward.md). The - // previous implementation returned false on a corrupt payload, which the S&F - // engine interprets as a permanent failure and parks the row — contradicting - // the invariant. The fix: discard a corrupt buffered notification by - // returning true (engine clears the buffer via its normal success path), - // with a Warning log line carrying the row id and a payload preview. + // Supersedes StoreAndForward-018 (discard-on-corrupt). Returning false is the + // handler contract's permanent-failure signal: the engine parks the row, which + // preserves the payload for operator forensics. Retrying a corrupt payload is + // pointless; deleting it silently (and reporting Delivered) loses data with no + // parked row, no central Notifications row, and no audit trail. var centralProbe = CreateTestProbe(); var forwarder = new NotificationForwarder( centralProbe.Ref, "site-7", ForwardTimeout); @@ -222,7 +219,7 @@ public class NotificationForwarderTests : TestKit OriginInstanceName = "Plant.Pump3", }; - Assert.True(await forwarder.DeliverAsync(corrupt)); + Assert.False(await forwarder.DeliverAsync(corrupt)); // The corrupt-payload path must NOT round-trip to central — no // NotificationSubmit / no Ask. ExpectNoMsg confirms nothing was forwarded. @@ -230,11 +227,11 @@ public class NotificationForwarderTests : TestKit } [Fact] - public async Task Deliver_NullDeserializedPayload_ReturnsTrue_AndDoesNotForwardAnything() + public async Task Deliver_NullDeserializedPayload_ReturnsFalse_ParksForForensics_AndDoesNotForward() { // The companion case to corrupt JSON: the payload is valid JSON but - // deserialises to null (e.g. "null"). Same treatment per StoreAndForward-018 - // — discard rather than park. + // deserialises to null (e.g. "null"). Same treatment — park (return false) + // rather than silently discard-as-delivered. Supersedes StoreAndForward-018. var centralProbe = CreateTestProbe(); var forwarder = new NotificationForwarder( centralProbe.Ref, "site-7", ForwardTimeout); @@ -248,7 +245,7 @@ public class NotificationForwarderTests : TestKit OriginInstanceName = "Plant.Pump3", }; - Assert.True(await forwarder.DeliverAsync(nullPayload)); + Assert.False(await forwarder.DeliverAsync(nullPayload)); centralProbe.ExpectNoMsg(TimeSpan.FromMilliseconds(200)); } }