From 283ce0647cd60a0357a3db342a78605e0edf398b Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 8 Jul 2026 20:42:39 -0400 Subject: [PATCH] fix(store-and-forward): standby applies Add/Park/Requeue as upserts so a lost Add self-heals --- .../ReplicationService.cs | 13 +++- .../StoreAndForwardStorage.cs | 61 +++++++++++++++++++ .../StoreAndForwardReplicationTests.cs | 57 +++++++++++++++++ 3 files changed, 128 insertions(+), 3 deletions(-) diff --git a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/ReplicationService.cs b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/ReplicationService.cs index abdb2393..60f2b162 100644 --- a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/ReplicationService.cs +++ b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/ReplicationService.cs @@ -101,6 +101,13 @@ public class ReplicationService /// /// Applies a replicated operation received from the active node. /// Used by the standby node to keep its SQLite in sync. + /// + /// Add/Park/Requeue are applied as upserts (), + /// not blind INSERT/UPDATE: the full message rides in every one of those operations, + /// so a Park/Requeue whose original Add was lost (fire-and-forget replication is + /// best-effort) self-heals by materialising the row, and a duplicate Add (e.g. + /// re-issued by an anti-entropy resync) applies newest-wins instead of throwing a + /// primary-key violation. Remove is a plain delete. /// /// The replication operation to apply. /// The standby node's store-and-forward storage to update. @@ -112,7 +119,7 @@ public class ReplicationService switch (operation.OperationType) { case ReplicationOperationType.Add when operation.Message != null: - await storage.EnqueueAsync(operation.Message); + await storage.UpsertMessageAsync(operation.Message); break; case ReplicationOperationType.Remove: @@ -121,13 +128,13 @@ public class ReplicationService case ReplicationOperationType.Park when operation.Message != null: operation.Message.Status = StoreAndForwardMessageStatus.Parked; - await storage.UpdateMessageAsync(operation.Message); + await storage.UpsertMessageAsync(operation.Message); break; case ReplicationOperationType.Requeue when operation.Message != null: operation.Message.Status = StoreAndForwardMessageStatus.Pending; operation.Message.RetryCount = 0; - await storage.UpdateMessageAsync(operation.Message); + await storage.UpsertMessageAsync(operation.Message); break; } } diff --git a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs index b84a0347..7c94ae58 100644 --- a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs +++ b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs @@ -176,6 +176,67 @@ public class StoreAndForwardStorage await cmd.ExecuteNonQueryAsync(); } + /// + /// Inserts a message, or updates every mutable column in place if a row with the + /// same id already exists (ON CONFLICT(id) DO UPDATE). Used by the standby + /// node when applying a replicated Add/Park/Requeue: a Park/Requeue whose original + /// Add was lost self-heals (the full message rides in the operation), and a + /// duplicate Add (e.g. after an anti-entropy resync) applies newest-wins instead of + /// violating the primary key. The original created_at is preserved — it + /// orders the retry sweep and must never be overwritten on update. + /// + /// The message to insert or update. + /// A task that represents the asynchronous operation. + public async Task UpsertMessageAsync(StoreAndForwardMessage message) + { + await using var connection = new SqliteConnection(_connectionString); + await connection.OpenAsync(); + + await using var cmd = connection.CreateCommand(); + cmd.CommandText = @" + INSERT INTO sf_messages (id, category, target, payload_json, retry_count, max_retries, + retry_interval_ms, created_at, last_attempt_at, status, last_error, + origin_instance, execution_id, source_script, parent_execution_id) + VALUES (@id, @category, @target, @payload, @retryCount, @maxRetries, + @retryIntervalMs, @createdAt, @lastAttempt, @status, @lastError, + @origin, @executionId, @sourceScript, @parentExecutionId) + ON CONFLICT(id) DO UPDATE SET + category = excluded.category, + target = excluded.target, + payload_json = excluded.payload_json, + retry_count = excluded.retry_count, + max_retries = excluded.max_retries, + retry_interval_ms = excluded.retry_interval_ms, + last_attempt_at = excluded.last_attempt_at, + status = excluded.status, + last_error = excluded.last_error, + origin_instance = excluded.origin_instance, + execution_id = excluded.execution_id, + source_script = excluded.source_script, + parent_execution_id = excluded.parent_execution_id"; + + cmd.Parameters.AddWithValue("@id", message.Id); + cmd.Parameters.AddWithValue("@category", (int)message.Category); + cmd.Parameters.AddWithValue("@target", message.Target); + cmd.Parameters.AddWithValue("@payload", message.PayloadJson); + cmd.Parameters.AddWithValue("@retryCount", message.RetryCount); + cmd.Parameters.AddWithValue("@maxRetries", message.MaxRetries); + cmd.Parameters.AddWithValue("@retryIntervalMs", message.RetryIntervalMs); + cmd.Parameters.AddWithValue("@createdAt", message.CreatedAt.ToString("O")); + cmd.Parameters.AddWithValue("@lastAttempt", message.LastAttemptAt.HasValue + ? message.LastAttemptAt.Value.ToString("O") : DBNull.Value); + cmd.Parameters.AddWithValue("@status", (int)message.Status); + cmd.Parameters.AddWithValue("@lastError", (object?)message.LastError ?? DBNull.Value); + cmd.Parameters.AddWithValue("@origin", (object?)message.OriginInstanceName ?? DBNull.Value); + cmd.Parameters.AddWithValue("@executionId", + message.ExecutionId.HasValue ? message.ExecutionId.Value.ToString("D") : DBNull.Value); + cmd.Parameters.AddWithValue("@sourceScript", (object?)message.SourceScript ?? DBNull.Value); + cmd.Parameters.AddWithValue("@parentExecutionId", + message.ParentExecutionId.HasValue ? message.ParentExecutionId.Value.ToString("D") : DBNull.Value); + + await cmd.ExecuteNonQueryAsync(); + } + /// /// Gets all messages that are due for retry (Pending status, last attempt older than retry interval). /// diff --git a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardReplicationTests.cs b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardReplicationTests.cs index e5c449da..64e584fb 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardReplicationTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardReplicationTests.cs @@ -203,4 +203,61 @@ public class StoreAndForwardReplicationTests : IAsyncLifetime, IDisposable Assert.Equal(StoreAndForwardMessageStatus.Pending, row!.Status); Assert.Equal(0, row.RetryCount); } + + private static ReplicationService NewReplicationService() => + new(new StoreAndForwardOptions { ReplicationEnabled = true }, + NullLogger.Instance); + + private static StoreAndForwardMessage NewMessage(string id) => new() + { + Id = id, + Category = StoreAndForwardCategory.ExternalSystem, + Target = "api", + PayloadJson = "{}", + RetryCount = 0, + MaxRetries = 1, + RetryIntervalMs = 0, + CreatedAt = DateTimeOffset.UtcNow, + Status = StoreAndForwardMessageStatus.Pending, + }; + + /// + /// A Park (or Requeue) whose original Add was lost — e.g. the Add's + /// fire-and-forget replication dropped — must still materialise the row on the + /// standby. The full message rides in the operation, so an upsert self-heals; + /// a blind UPDATE would affect 0 rows and the row would be gone forever. + /// + [Fact] + public async Task ApplyReplicatedPark_WhenAddWasLost_MaterializesTheParkedRow() + { + var service = NewReplicationService(); + var msg = NewMessage("lost-add-1"); // never Added on this (standby) storage + msg.Status = StoreAndForwardMessageStatus.Parked; + + await service.ApplyReplicatedOperationAsync( + new ReplicationOperation(ReplicationOperationType.Park, msg.Id, msg), _storage); + + var row = await _storage.GetMessageByIdAsync("lost-add-1"); + Assert.NotNull(row); // pre-fix: blind UPDATE affected 0 rows, row is gone forever + Assert.Equal(StoreAndForwardMessageStatus.Parked, row!.Status); + } + + /// + /// A duplicate Add (e.g. after Task 21's anti-entropy resync re-issues an Add + /// the standby already holds) must not violate the PK — the upsert applies + /// newest-wins instead of throwing. + /// + [Fact] + public async Task ApplyReplicatedAdd_Twice_IsIdempotent_NewestWins() + { + var service = NewReplicationService(); + var msg = NewMessage("dup-add"); + await service.ApplyReplicatedOperationAsync( + new ReplicationOperation(ReplicationOperationType.Add, msg.Id, msg), _storage); + msg.RetryCount = 3; + await service.ApplyReplicatedOperationAsync( // pre-fix: SqliteException PK violation + new ReplicationOperation(ReplicationOperationType.Add, msg.Id, msg), _storage); + + Assert.Equal(3, (await _storage.GetMessageByIdAsync("dup-add"))!.RetryCount); + } }