diff --git a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs index 54859796..8b3b008f 100644 --- a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs +++ b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs @@ -123,34 +123,6 @@ public class StoreAndForwardStorage 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 = OpenConnection(); - - 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); - } - /// /// 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 @@ -224,23 +196,41 @@ public class StoreAndForwardStorage await using var connection = OpenConnection(); await using var cmd = connection.CreateCommand(); - cmd.CommandText = @" - SELECT id, category, target, payload_json, retry_count, max_retries, + const string cols = @"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 - WHERE status = @pending + execution_id, source_script, parent_execution_id"; + const string due = @"status = @pending AND (last_attempt_at_ms IS NULL OR retry_interval_ms = 0 - OR (@nowMs - last_attempt_at_ms) >= retry_interval_ms) - ORDER BY created_at ASC"; + OR (@nowMs - last_attempt_at_ms) >= retry_interval_ms)"; - // Bound the sweep batch (oldest-first). 0 = unlimited (legacy). if (limit > 0) { - cmd.CommandText += "\n LIMIT @limit"; + // Page FAIRLY across (category, target) lanes: rank each due row within its + // lane (oldest-first) and take the page by rank, so one dead target holding + // hundreds of due rows contributes one row per "round" like every other + // lane instead of owning the whole page. Within a lane rank order == created_at + // order, so per-target FIFO is preserved for the service's GroupBy. + cmd.CommandText = $@" + SELECT {cols} + FROM ( + SELECT {cols}, + ROW_NUMBER() OVER (PARTITION BY category, target ORDER BY created_at ASC) AS lane_rank + FROM sf_messages + WHERE {due} + ) + ORDER BY lane_rank ASC, created_at ASC + LIMIT @limit"; cmd.Parameters.AddWithValue("@limit", limit); } + else + { + cmd.CommandText = $@" + SELECT {cols} + FROM sf_messages + WHERE {due} + ORDER BY created_at ASC"; + } cmd.Parameters.AddWithValue("@pending", (int)StoreAndForwardMessageStatus.Pending); cmd.Parameters.AddWithValue("@nowMs", DateTimeOffset.UtcNow.ToUnixTimeMilliseconds()); diff --git a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardStorageTests.cs b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardStorageTests.cs index a3b1a26e..d55c495e 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardStorageTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardStorageTests.cs @@ -822,19 +822,35 @@ public class StoreAndForwardStorageTests : IAsyncLifetime, IDisposable Status = status, }; - /// Verifies that the all-messages query returns every status, oldest first, up to the limit. + /// + /// Review 2026-10-21 02/C2: a dead target holding more due rows than the sweep page must not + /// starve a healthy target — the page is ranked per (category, target) lane, and each lane + /// stays oldest-first. + /// /// A task that represents the asynchronous operation. [Fact] - public async Task GetAllMessages_ReturnsEveryStatus_OldestFirst_UpToLimit() + public async Task GetMessagesForRetry_PagesPerLane_DeadTargetCannotStarveOthers() { - 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 @base = DateTimeOffset.UtcNow.AddHours(-1); + for (var i = 0; i < 600; i++) + { + var m = NewMsg($"a{i:D3}", StoreAndForwardMessageStatus.Pending, @base.AddSeconds(i)); + m.Target = "dead"; + m.RetryIntervalMs = 0; + await _storage.EnqueueAsync(m); + } + var healthy = NewMsg("b0", StoreAndForwardMessageStatus.Pending, @base.AddSeconds(1000)); + healthy.Target = "healthy"; + healthy.RetryIntervalMs = 0; + await _storage.EnqueueAsync(healthy); - 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 + var page = await _storage.GetMessagesForRetryAsync(500); + + Assert.Equal(500, page.Count); + Assert.Contains(page, m => m.Id == "b0"); + var dead = page.Where(m => m.Target == "dead").Select(m => m.Id).ToList(); + Assert.Equal(dead.OrderBy(x => x).ToList(), dead); // per-target FIFO preserved + Assert.Equal("a000", dead[0]); } // ReplaceAll_SwapsTheEntireBuffer_Atomically was DELETED with ReplaceAllAsync in