dohertj2/ScadaBridge
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(store-and-forward): page retry sweep per (category,target) lane; drop dead GetAllMessagesAsync (review 2026-10-21 02/C2, 02/U2)

Browse Source
This commit is contained in:
Joseph Doherty
2026-10-22 08:28:01 -04:00
parent e968566e9d
commit 76f7960a06
2 changed files with 52 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs
+27 -37
View File
@@ -123,34 +123,6 @@ public class StoreAndForwardStorage
await cmd.ExecuteNonQueryAsync();
}
/// <summary>
/// Returns every buffered message regardless of status, oldest-first, up to
/// <paramref name="limit"/> 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 (<c>limit + 1</c>) purely to
/// compute the truncation flag without a second COUNT round-trip.
/// </summary>
/// <param name="limit">Maximum rows to return; must be positive.</param>
/// <returns>The oldest-first page (at most <paramref name="limit"/> rows) and whether more rows exist beyond it.</returns>
public async Task<(List<StoreAndForwardMessage> 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);
}
/// <summary>
/// Inserts a message, or updates every mutable column in place if a row with the
/// same id already exists (<c>ON CONFLICT(id) DO UPDATE</c>). 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());
tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardStorageTests.cs
+25 -9
View File
@@ -822,19 +822,35 @@ public class StoreAndForwardStorageTests : IAsyncLifetime, IDisposable
Status = status,
};
/// <summary>Verifies that the all-messages query returns every status, oldest first, up to the limit.</summary>
/// <summary>
/// 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.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[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