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

perf(store-and-forward): epoch-ms due-check with additive column, backfill, and (status, due) index

Browse Source
This commit is contained in:
Joseph Doherty
2026-07-08 21:14:26 -04:00
parent 992f343f11
commit 5fe1b9747b
2 changed files with 113 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs
+48 -7
View File
@@ -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);
tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardStorageTests.cs
+65
View File
@@ -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<object?> 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;
}
}