From 5fe1b9747b1c7c88e8063b86f418e9f6db789181 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 8 Jul 2026 21:14:26 -0400 Subject: [PATCH] perf(store-and-forward): epoch-ms due-check with additive column, backfill, and (status, due) index --- .../StoreAndForwardStorage.cs | 55 ++++++++++++++-- .../StoreAndForwardStorageTests.cs | 65 +++++++++++++++++++ 2 files changed, 113 insertions(+), 7 deletions(-) diff --git a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs index dc0bc06d..cfbe34ea 100644 --- a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs +++ b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs @@ -85,6 +85,34 @@ public class StoreAndForwardStorage // reads back ParentExecutionId = null (back-compat). await AddColumnIfMissingAsync(connection, "parent_execution_id", "TEXT"); + // Additively add the epoch-ms sibling of last_attempt_at (Task 18). The + // ISO-8601 text column stays authoritative for reads / back-compat; this + // INTEGER column drives the due predicate so the sweep no longer parses + // julianday() per row. + await AddColumnIfMissingAsync(connection, "last_attempt_at_ms", "INTEGER"); + + // One-time backfill for rows persisted before the ms column existed: derive + // epoch-ms from the text timestamp. The "... IS NULL" guard makes this run + // once per legacy DB and never match again — this is the only remaining + // julianday() use in the store. + await using (var backfill = connection.CreateCommand()) + { + backfill.CommandText = @" + UPDATE sf_messages + SET last_attempt_at_ms = CAST((julianday(last_attempt_at) - 2440587.5) * 86400000 AS INTEGER) + WHERE last_attempt_at IS NOT NULL AND last_attempt_at_ms IS NULL"; + await backfill.ExecuteNonQueryAsync(); + } + + // Covering index for the due query (status filter + ms ordering column). + // Created after the ALTER above so the column exists. + await using (var dueIndex = connection.CreateCommand()) + { + dueIndex.CommandText = + "CREATE INDEX IF NOT EXISTS idx_sf_messages_status_due ON sf_messages(status, last_attempt_at_ms)"; + await dueIndex.ExecuteNonQueryAsync(); + } + _logger.LogInformation("Store-and-forward SQLite storage initialized"); } @@ -168,10 +196,10 @@ public class StoreAndForwardStorage 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, + retry_interval_ms, created_at, last_attempt_at, last_attempt_at_ms, status, last_error, origin_instance, execution_id, source_script, parent_execution_id) VALUES (@id, @category, @target, @payload, @retryCount, @maxRetries, - @retryIntervalMs, @createdAt, @lastAttempt, @status, @lastError, + @retryIntervalMs, @createdAt, @lastAttempt, @lastAttemptMs, @status, @lastError, @origin, @executionId, @sourceScript, @parentExecutionId)"; cmd.Parameters.AddWithValue("@id", message.Id); @@ -184,6 +212,8 @@ public class StoreAndForwardStorage 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("@lastAttemptMs", + (object?)message.LastAttemptAt?.ToUnixTimeMilliseconds() ?? 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); @@ -220,10 +250,10 @@ public class StoreAndForwardStorage 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, + retry_interval_ms, created_at, last_attempt_at, last_attempt_at_ms, status, last_error, origin_instance, execution_id, source_script, parent_execution_id) VALUES (@id, @category, @target, @payload, @retryCount, @maxRetries, - @retryIntervalMs, @createdAt, @lastAttempt, @status, @lastError, + @retryIntervalMs, @createdAt, @lastAttempt, @lastAttemptMs, @status, @lastError, @origin, @executionId, @sourceScript, @parentExecutionId) ON CONFLICT(id) DO UPDATE SET category = excluded.category, @@ -233,6 +263,7 @@ public class StoreAndForwardStorage max_retries = excluded.max_retries, retry_interval_ms = excluded.retry_interval_ms, last_attempt_at = excluded.last_attempt_at, + last_attempt_at_ms = excluded.last_attempt_at_ms, status = excluded.status, last_error = excluded.last_error, origin_instance = excluded.origin_instance, @@ -250,6 +281,8 @@ public class StoreAndForwardStorage 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("@lastAttemptMs", + (object?)message.LastAttemptAt?.ToUnixTimeMilliseconds() ?? 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); @@ -277,9 +310,9 @@ public class StoreAndForwardStorage execution_id, source_script, parent_execution_id FROM sf_messages WHERE status = @pending - AND (last_attempt_at IS NULL + AND (last_attempt_at_ms IS NULL OR retry_interval_ms = 0 - OR (julianday('now') - julianday(last_attempt_at)) * 86400000 >= retry_interval_ms) + OR (@nowMs - last_attempt_at_ms) >= retry_interval_ms) ORDER BY created_at ASC"; // Bound the sweep batch (oldest-first). 0 = unlimited (legacy). @@ -290,6 +323,7 @@ public class StoreAndForwardStorage } cmd.Parameters.AddWithValue("@pending", (int)StoreAndForwardMessageStatus.Pending); + cmd.Parameters.AddWithValue("@nowMs", DateTimeOffset.UtcNow.ToUnixTimeMilliseconds()); return await ReadMessagesAsync(cmd); } @@ -308,6 +342,7 @@ public class StoreAndForwardStorage UPDATE sf_messages SET retry_count = @retryCount, last_attempt_at = @lastAttempt, + last_attempt_at_ms = @lastAttemptMs, status = @status, last_error = @lastError WHERE id = @id"; @@ -316,6 +351,8 @@ public class StoreAndForwardStorage cmd.Parameters.AddWithValue("@retryCount", message.RetryCount); cmd.Parameters.AddWithValue("@lastAttempt", message.LastAttemptAt.HasValue ? message.LastAttemptAt.Value.ToString("O") : DBNull.Value); + cmd.Parameters.AddWithValue("@lastAttemptMs", + (object?)message.LastAttemptAt?.ToUnixTimeMilliseconds() ?? DBNull.Value); cmd.Parameters.AddWithValue("@status", (int)message.Status); cmd.Parameters.AddWithValue("@lastError", (object?)message.LastError ?? DBNull.Value); @@ -342,6 +379,7 @@ public class StoreAndForwardStorage UPDATE sf_messages SET retry_count = @retryCount, last_attempt_at = @lastAttempt, + last_attempt_at_ms = @lastAttemptMs, status = @status, last_error = @lastError WHERE id = @id AND status = @expectedStatus"; @@ -350,6 +388,8 @@ public class StoreAndForwardStorage cmd.Parameters.AddWithValue("@retryCount", message.RetryCount); cmd.Parameters.AddWithValue("@lastAttempt", message.LastAttemptAt.HasValue ? message.LastAttemptAt.Value.ToString("O") : DBNull.Value); + cmd.Parameters.AddWithValue("@lastAttemptMs", + (object?)message.LastAttemptAt?.ToUnixTimeMilliseconds() ?? DBNull.Value); cmd.Parameters.AddWithValue("@status", (int)message.Status); cmd.Parameters.AddWithValue("@lastError", (object?)message.LastError ?? DBNull.Value); cmd.Parameters.AddWithValue("@expectedStatus", (int)expectedStatus); @@ -436,7 +476,8 @@ public class StoreAndForwardStorage await using var cmd = connection.CreateCommand(); cmd.CommandText = @" UPDATE sf_messages - SET status = @pending, retry_count = 0, last_error = NULL, last_attempt_at = NULL + SET status = @pending, retry_count = 0, last_error = NULL, + last_attempt_at = NULL, last_attempt_at_ms = NULL WHERE id = @id AND status = @parked"; cmd.Parameters.AddWithValue("@id", messageId); diff --git a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardStorageTests.cs b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardStorageTests.cs index acc9b82c..3ee14b0f 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardStorageTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardStorageTests.cs @@ -683,4 +683,69 @@ public class StoreAndForwardStorageTests : IAsyncLifetime, IDisposable Directory.Delete(directory, recursive: true); } } + + // ── Task 18: epoch-ms due-check ── + + [Fact] + public async Task GetMessagesForRetry_UsesEpochMs_RespectsInterval() + { + var msg = new StoreAndForwardMessage + { + Id = "due-check", + Category = StoreAndForwardCategory.ExternalSystem, + Target = "t", + PayloadJson = "{}", + RetryCount = 0, + MaxRetries = 50, + RetryIntervalMs = 60_000, + CreatedAt = DateTimeOffset.UtcNow, + Status = StoreAndForwardMessageStatus.Pending, + LastAttemptAt = DateTimeOffset.UtcNow.AddSeconds(-30), // not yet due + }; + await _storage.EnqueueAsync(msg); + Assert.Empty(await _storage.GetMessagesForRetryAsync()); + + msg.LastAttemptAt = DateTimeOffset.UtcNow.AddSeconds(-90); // due + await _storage.UpdateMessageAsync(msg); + Assert.Single(await _storage.GetMessagesForRetryAsync()); + } + + [Fact] + public async Task Initialize_BackfillsEpochMs_ForLegacyRows() + { + // Simulate a legacy row: text timestamp present, ms column NULL. + await ExecRawAsync( + "INSERT INTO sf_messages (id, category, target, payload_json, created_at, last_attempt_at, status) " + + "VALUES ('legacy', 0, 't', '{}', @c, @l, 0)", + ("@c", DateTimeOffset.UtcNow.ToString("O")), + ("@l", DateTimeOffset.UtcNow.AddHours(-1).ToString("O"))); + + await _storage.InitializeAsync(); // second init runs the one-time backfill + + var ms = await ScalarRawAsync("SELECT last_attempt_at_ms FROM sf_messages WHERE id='legacy'"); + Assert.NotNull(ms); + Assert.InRange(Convert.ToInt64(ms), + DateTimeOffset.UtcNow.AddHours(-1).AddMinutes(-1).ToUnixTimeMilliseconds(), + DateTimeOffset.UtcNow.ToUnixTimeMilliseconds()); + } + + private async Task ExecRawAsync(string sql, params (string, object)[] parameters) + { + await using var conn = new SqliteConnection($"Data Source={_dbName};Mode=Memory;Cache=Shared"); + await conn.OpenAsync(); + await using var cmd = conn.CreateCommand(); + cmd.CommandText = sql; + foreach (var (name, value) in parameters) cmd.Parameters.AddWithValue(name, value); + await cmd.ExecuteNonQueryAsync(); + } + + private async Task ScalarRawAsync(string sql) + { + await using var conn = new SqliteConnection($"Data Source={_dbName};Mode=Memory;Cache=Shared"); + await conn.OpenAsync(); + await using var cmd = conn.CreateCommand(); + cmd.CommandText = sql; + var result = await cmd.ExecuteScalarAsync(); + return result is DBNull ? null : result; + } }