From 3eef9d1077859a6c75109161e4c9779f10146750 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 8 Jul 2026 21:22:25 -0400 Subject: [PATCH] feat(store-and-forward): GetAllMessagesAsync/ReplaceAllAsync storage primitives for standby buffer resync --- .../StoreAndForwardStorage.cs | 103 +++++++++++++++--- .../StoreAndForwardStorageTests.cs | 42 +++++++ 2 files changed, 130 insertions(+), 15 deletions(-) diff --git a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs index cfbe34ea..26a7a7d4 100644 --- a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs +++ b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs @@ -185,16 +185,11 @@ public class StoreAndForwardStorage } /// - /// Enqueues a new message with Pending status. + /// INSERT statement for a full message row. Shared by + /// and ; bind with + /// so the column list and the parameters never drift apart. /// - /// The message to enqueue. - /// A task that represents the asynchronous operation. - public async Task EnqueueAsync(StoreAndForwardMessage message) - { - await using var connection = await OpenConnectionAsync(); - - await using var cmd = connection.CreateCommand(); - cmd.CommandText = @" + private const string InsertMessageSql = @" INSERT INTO sf_messages (id, category, target, payload_json, retry_count, max_retries, retry_interval_ms, created_at, last_attempt_at, last_attempt_at_ms, status, last_error, origin_instance, execution_id, source_script, parent_execution_id) @@ -202,6 +197,14 @@ public class StoreAndForwardStorage @retryIntervalMs, @createdAt, @lastAttempt, @lastAttemptMs, @status, @lastError, @origin, @executionId, @sourceScript, @parentExecutionId)"; + /// + /// Binds every column parameter for a full message row (used by + /// ). GUID ids are stored in canonical "D" string + /// form and the epoch-ms sibling of last_attempt_at is derived here so a single + /// place owns the mapping. + /// + private static void BindMessageParameters(SqliteCommand cmd, StoreAndForwardMessage message) + { cmd.Parameters.AddWithValue("@id", message.Id); cmd.Parameters.AddWithValue("@category", (int)message.Category); cmd.Parameters.AddWithValue("@target", message.Target); @@ -217,21 +220,91 @@ public class StoreAndForwardStorage 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); - // The execution id is stored as its - // canonical string form ("D") so it round-trips cleanly through the - // TEXT column; null when not a cached call / not threaded. + // GUID ids stored as canonical "D" strings; null when not threaded. cmd.Parameters.AddWithValue("@executionId", message.ExecutionId.HasValue ? message.ExecutionId.Value.ToString("D") : DBNull.Value); cmd.Parameters.AddWithValue("@sourceScript", (object?)message.SourceScript ?? DBNull.Value); - // The parent execution id is - // stored as its canonical string form ("D") so it round-trips cleanly - // through the TEXT column; null when not a routed cached call. cmd.Parameters.AddWithValue("@parentExecutionId", message.ParentExecutionId.HasValue ? message.ParentExecutionId.Value.ToString("D") : DBNull.Value); + } + + /// + /// Enqueues a new message with Pending status. + /// + /// The message to enqueue. + /// A task that represents the asynchronous operation. + public async Task EnqueueAsync(StoreAndForwardMessage message) + { + await using var connection = await OpenConnectionAsync(); + + await using var cmd = connection.CreateCommand(); + cmd.CommandText = InsertMessageSql; + BindMessageParameters(cmd, message); await cmd.ExecuteNonQueryAsync(); } + /// + /// Returns every buffered message regardless of status, oldest-first, up to + /// rows, plus a flag indicating the buffer holds more + /// than that. Anti-entropy groundwork: the active node snapshots its buffer for + /// a standby resync. One extra row is fetched (limit + 1) purely to + /// compute the truncation flag without a second COUNT round-trip. + /// + /// Maximum rows to return; must be positive. + /// The oldest-first page (at most rows) and whether more rows exist beyond it. + public async Task<(List Messages, bool Truncated)> GetAllMessagesAsync(int limit) + { + await using var connection = await OpenConnectionAsync(); + + await using var cmd = connection.CreateCommand(); + cmd.CommandText = @" + SELECT 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 + FROM sf_messages + ORDER BY created_at ASC + LIMIT @limitPlusOne"; + cmd.Parameters.AddWithValue("@limitPlusOne", limit + 1); + + var rows = await ReadMessagesAsync(cmd); + var truncated = rows.Count > limit; + return (rows.Take(limit).ToList(), truncated); + } + + /// + /// Atomically replaces the entire buffer with in a + /// single transaction (delete-all then insert-all). Standby-side anti-entropy + /// apply: a peer-join resync overwrites the standby's divergent copy with the + /// active node's authoritative snapshot. Never call on an active node — + /// it discards every in-flight row. + /// + /// The full buffer snapshot to install. + /// A task that represents the asynchronous operation. + public async Task ReplaceAllAsync(IReadOnlyList messages) + { + await using var connection = await OpenConnectionAsync(); + await using var transaction = (SqliteTransaction)await connection.BeginTransactionAsync(); + + await using (var deleteCmd = connection.CreateCommand()) + { + deleteCmd.Transaction = transaction; + deleteCmd.CommandText = "DELETE FROM sf_messages"; + await deleteCmd.ExecuteNonQueryAsync(); + } + + foreach (var message in messages) + { + await using var insertCmd = connection.CreateCommand(); + insertCmd.Transaction = transaction; + insertCmd.CommandText = InsertMessageSql; + BindMessageParameters(insertCmd, message); + await insertCmd.ExecuteNonQueryAsync(); + } + + await transaction.CommitAsync(); + } + /// /// 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 diff --git a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardStorageTests.cs b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardStorageTests.cs index 3ee14b0f..317d704c 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardStorageTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardStorageTests.cs @@ -748,4 +748,46 @@ public class StoreAndForwardStorageTests : IAsyncLifetime, IDisposable var result = await cmd.ExecuteScalarAsync(); return result is DBNull ? null : result; } + + // ── Task 20: buffer-resync storage primitives ── + + private static StoreAndForwardMessage NewMsg( + string id, + StoreAndForwardMessageStatus status = StoreAndForwardMessageStatus.Pending, + DateTimeOffset? createdAt = null) => new() + { + Id = id, + Category = StoreAndForwardCategory.ExternalSystem, + Target = "t", + PayloadJson = "{}", + RetryCount = 0, + MaxRetries = 50, + RetryIntervalMs = 30000, + CreatedAt = createdAt ?? DateTimeOffset.UtcNow, + Status = status, + }; + + [Fact] + public async Task GetAllMessages_ReturnsEveryStatus_OldestFirst_UpToLimit() + { + var @base = DateTimeOffset.UtcNow; + await _storage.EnqueueAsync(NewMsg("p1", StoreAndForwardMessageStatus.Pending, @base)); + await _storage.EnqueueAsync(NewMsg("k1", StoreAndForwardMessageStatus.Parked, @base.AddSeconds(1))); + await _storage.EnqueueAsync(NewMsg("p2", StoreAndForwardMessageStatus.Pending, @base.AddSeconds(2))); + + var (all, truncated) = await _storage.GetAllMessagesAsync(limit: 2); + Assert.Equal(new[] { "p1", "k1" }, all.Select(m => m.Id)); // every status, oldest-first + Assert.True(truncated); // a third row exists beyond the limit + } + + [Fact] + public async Task ReplaceAll_SwapsTheEntireBuffer_Atomically() + { + await _storage.EnqueueAsync(NewMsg("stale")); + await _storage.ReplaceAllAsync(new[] { NewMsg("fresh-1"), NewMsg("fresh-2") }); + + Assert.Null(await _storage.GetMessageByIdAsync("stale")); + Assert.NotNull(await _storage.GetMessageByIdAsync("fresh-1")); + Assert.NotNull(await _storage.GetMessageByIdAsync("fresh-2")); + } }