Joseph Doherty
166f07fa68
LocalDb 0.1.1 creates the parent directory of LocalDb:Path itself, so the SiteLocalDbDirectory shim this repo carried through Phase 2 is deleted along with its call site. The gap was found here but was never ScadaBridge's alone — every LocalDb consumer had it — so the fix moved to the library. SiteLocalDbDirectoryTests is RETAINED and retargeted rather than deleted with the shim. It was already written against the site registration path, not the mechanism, so it needed only its Ensure() call removed: what a site node requires is that resolving ILocalDb not fail on a fresh machine, regardless of who provides that. Verified it still earns its place — pinned back to LocalDb 0.1.0 it fails inside SqliteLocalDb..ctor -> SqliteConnection.Open(), so it genuinely depends on the library behaviour and not on a coincidence. Also corrects the coverage-split note in StoreAndForwardStorageTests, which asserted that directory creation is "NOT LocalDb's" — true when written, wrong as of 0.1.1. Build 0 warnings; Host 330, StoreAndForward 130, LocalDb integration 20 pass. Claude-Session: https://claude.ai/code/session_01BL2Vu1ESDQ9SCN4gVKkdts
800 lines
34 KiB
C#
800 lines
34 KiB
C#
using Microsoft.Data.Sqlite;
|
|
using Microsoft.Extensions.Logging.Abstractions;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
|
|
using ZB.MOM.WW.ScadaBridge.TestSupport;
|
|
|
|
namespace ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests;
|
|
|
|
/// <summary>
|
|
/// WP-9: Tests for SQLite persistence layer.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// Backed by a real temp-file LocalDb rather than the shared-cache in-memory database
|
|
/// this class used before: <see cref="StoreAndForwardStorage"/> now takes
|
|
/// <c>ILocalDb</c>, and LocalDb has no in-memory mode (<c>LocalDb:Path</c> is a
|
|
/// filesystem path). Isolation still comes from a fresh database per test — xUnit
|
|
/// constructs one instance per test — it is just a file now.
|
|
/// </remarks>
|
|
public class StoreAndForwardStorageTests : IAsyncLifetime, IDisposable
|
|
{
|
|
private readonly TestLocalDb _localDb;
|
|
private readonly StoreAndForwardStorage _storage;
|
|
|
|
public StoreAndForwardStorageTests()
|
|
{
|
|
_localDb = TestLocalDb.CreateTemp("StorageTests");
|
|
_storage = new StoreAndForwardStorage(_localDb.Db, NullLogger<StoreAndForwardStorage>.Instance);
|
|
}
|
|
|
|
public async Task InitializeAsync() => await _storage.InitializeAsync();
|
|
|
|
public Task DisposeAsync() => Task.CompletedTask;
|
|
|
|
public void Dispose()
|
|
{
|
|
// Dispose first — the master connection anchors the WAL, so the sidecars
|
|
// cannot be removed while it is open.
|
|
var path = _localDb.Path;
|
|
_localDb.Dispose();
|
|
TestLocalDb.DeleteFiles(path);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task EnqueueAsync_StoresMessage()
|
|
{
|
|
var message = CreateMessage("msg1", StoreAndForwardCategory.ExternalSystem);
|
|
await _storage.EnqueueAsync(message);
|
|
|
|
var retrieved = await _storage.GetMessageByIdAsync("msg1");
|
|
Assert.NotNull(retrieved);
|
|
Assert.Equal("msg1", retrieved!.Id);
|
|
Assert.Equal(StoreAndForwardCategory.ExternalSystem, retrieved.Category);
|
|
Assert.Equal("target1", retrieved.Target);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task EnqueueAsync_AllCategories()
|
|
{
|
|
await _storage.EnqueueAsync(CreateMessage("es1", StoreAndForwardCategory.ExternalSystem));
|
|
await _storage.EnqueueAsync(CreateMessage("n1", StoreAndForwardCategory.Notification));
|
|
await _storage.EnqueueAsync(CreateMessage("db1", StoreAndForwardCategory.CachedDbWrite));
|
|
|
|
var es = await _storage.GetMessageByIdAsync("es1");
|
|
var n = await _storage.GetMessageByIdAsync("n1");
|
|
var db = await _storage.GetMessageByIdAsync("db1");
|
|
|
|
Assert.Equal(StoreAndForwardCategory.ExternalSystem, es!.Category);
|
|
Assert.Equal(StoreAndForwardCategory.Notification, n!.Category);
|
|
Assert.Equal(StoreAndForwardCategory.CachedDbWrite, db!.Category);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task RemoveMessageAsync_RemovesSuccessfully()
|
|
{
|
|
await _storage.EnqueueAsync(CreateMessage("rm1", StoreAndForwardCategory.ExternalSystem));
|
|
await _storage.RemoveMessageAsync("rm1");
|
|
|
|
var retrieved = await _storage.GetMessageByIdAsync("rm1");
|
|
Assert.Null(retrieved);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task UpdateMessageAsync_UpdatesFields()
|
|
{
|
|
var message = CreateMessage("upd1", StoreAndForwardCategory.ExternalSystem);
|
|
await _storage.EnqueueAsync(message);
|
|
|
|
message.RetryCount = 5;
|
|
message.LastAttemptAt = DateTimeOffset.UtcNow;
|
|
message.Status = StoreAndForwardMessageStatus.Parked;
|
|
message.LastError = "Connection refused";
|
|
await _storage.UpdateMessageAsync(message);
|
|
|
|
var retrieved = await _storage.GetMessageByIdAsync("upd1");
|
|
Assert.Equal(5, retrieved!.RetryCount);
|
|
Assert.Equal(StoreAndForwardMessageStatus.Parked, retrieved.Status);
|
|
Assert.Equal("Connection refused", retrieved.LastError);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task GetMessagesForRetryAsync_ReturnsOnlyPendingMessages()
|
|
{
|
|
var pending = CreateMessage("pend1", StoreAndForwardCategory.ExternalSystem);
|
|
pending.Status = StoreAndForwardMessageStatus.Pending;
|
|
await _storage.EnqueueAsync(pending);
|
|
|
|
var parked = CreateMessage("park1", StoreAndForwardCategory.ExternalSystem);
|
|
parked.Status = StoreAndForwardMessageStatus.Parked;
|
|
await _storage.EnqueueAsync(parked);
|
|
await _storage.UpdateMessageAsync(parked);
|
|
|
|
var forRetry = await _storage.GetMessagesForRetryAsync();
|
|
Assert.All(forRetry, m => Assert.Equal(StoreAndForwardMessageStatus.Pending, m.Status));
|
|
}
|
|
|
|
[Fact]
|
|
public async Task GetMessagesForRetryAsync_NonZeroInterval_ExcludesNotYetDueIncludesDue()
|
|
{
|
|
// StoreAndForward-013: exercise the julianday elapsed-time comparison with a
|
|
// non-zero retry interval. A message attempted just now must NOT be due; one
|
|
// attempted long ago must be due.
|
|
var notDue = CreateMessage("notdue", StoreAndForwardCategory.ExternalSystem);
|
|
notDue.RetryIntervalMs = (long)TimeSpan.FromHours(1).TotalMilliseconds;
|
|
notDue.LastAttemptAt = DateTimeOffset.UtcNow;
|
|
await _storage.EnqueueAsync(notDue);
|
|
|
|
var due = CreateMessage("due", StoreAndForwardCategory.ExternalSystem);
|
|
due.RetryIntervalMs = (long)TimeSpan.FromMinutes(5).TotalMilliseconds;
|
|
due.LastAttemptAt = DateTimeOffset.UtcNow.AddHours(-2);
|
|
await _storage.EnqueueAsync(due);
|
|
|
|
var neverAttempted = CreateMessage("never", StoreAndForwardCategory.ExternalSystem);
|
|
neverAttempted.RetryIntervalMs = (long)TimeSpan.FromHours(1).TotalMilliseconds;
|
|
neverAttempted.LastAttemptAt = null;
|
|
await _storage.EnqueueAsync(neverAttempted);
|
|
|
|
var forRetry = await _storage.GetMessagesForRetryAsync();
|
|
var ids = forRetry.Select(m => m.Id).ToHashSet();
|
|
|
|
Assert.DoesNotContain("notdue", ids);
|
|
Assert.Contains("due", ids);
|
|
Assert.Contains("never", ids);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task GetParkedMessagesAsync_ReturnsParkedOnly()
|
|
{
|
|
var msg = CreateMessage("prk1", StoreAndForwardCategory.Notification);
|
|
msg.Status = StoreAndForwardMessageStatus.Parked;
|
|
await _storage.EnqueueAsync(msg);
|
|
await _storage.UpdateMessageAsync(msg);
|
|
|
|
var (messages, total) = await _storage.GetParkedMessagesAsync();
|
|
Assert.True(total > 0);
|
|
Assert.All(messages, m => Assert.Equal(StoreAndForwardMessageStatus.Parked, m.Status));
|
|
}
|
|
|
|
[Fact]
|
|
public async Task RetryParkedMessageAsync_MovesToPending()
|
|
{
|
|
var msg = CreateMessage("retry1", StoreAndForwardCategory.ExternalSystem);
|
|
msg.Status = StoreAndForwardMessageStatus.Parked;
|
|
msg.RetryCount = 10;
|
|
await _storage.EnqueueAsync(msg);
|
|
await _storage.UpdateMessageAsync(msg);
|
|
|
|
var success = await _storage.RetryParkedMessageAsync("retry1");
|
|
Assert.True(success);
|
|
|
|
var retrieved = await _storage.GetMessageByIdAsync("retry1");
|
|
Assert.Equal(StoreAndForwardMessageStatus.Pending, retrieved!.Status);
|
|
Assert.Equal(0, retrieved.RetryCount);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task RetryParkedMessageAsync_ClearsLastAttemptAt_SoMessageIsImmediatelyDue()
|
|
{
|
|
// StoreAndForward-010: a re-queued parked message must be unambiguously due
|
|
// for the next sweep regardless of its (stale) last_attempt_at. Use a large
|
|
// retry interval so a leftover timestamp would otherwise exclude the message.
|
|
var msg = CreateMessage("requeue1", StoreAndForwardCategory.ExternalSystem);
|
|
msg.RetryIntervalMs = (long)TimeSpan.FromHours(1).TotalMilliseconds;
|
|
msg.LastAttemptAt = DateTimeOffset.UtcNow; // recent attempt
|
|
msg.Status = StoreAndForwardMessageStatus.Parked;
|
|
await _storage.EnqueueAsync(msg);
|
|
await _storage.UpdateMessageAsync(msg);
|
|
|
|
var requeued = await _storage.RetryParkedMessageAsync("requeue1");
|
|
Assert.True(requeued);
|
|
|
|
var retrieved = await _storage.GetMessageByIdAsync("requeue1");
|
|
Assert.Null(retrieved!.LastAttemptAt);
|
|
|
|
// It must appear in the retry-due set even though the configured interval
|
|
// (1 hour) has not elapsed since the original attempt.
|
|
var due = await _storage.GetMessagesForRetryAsync();
|
|
Assert.Contains(due, m => m.Id == "requeue1");
|
|
}
|
|
|
|
[Fact]
|
|
public async Task DiscardParkedMessageAsync_RemovesMessage()
|
|
{
|
|
var msg = CreateMessage("disc1", StoreAndForwardCategory.ExternalSystem);
|
|
msg.Status = StoreAndForwardMessageStatus.Parked;
|
|
await _storage.EnqueueAsync(msg);
|
|
await _storage.UpdateMessageAsync(msg);
|
|
|
|
var success = await _storage.DiscardParkedMessageAsync("disc1");
|
|
Assert.True(success);
|
|
|
|
var retrieved = await _storage.GetMessageByIdAsync("disc1");
|
|
Assert.Null(retrieved);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task GetBufferDepthByCategoryAsync_ReturnsCorrectCounts()
|
|
{
|
|
await _storage.EnqueueAsync(CreateMessage("bd1", StoreAndForwardCategory.ExternalSystem));
|
|
await _storage.EnqueueAsync(CreateMessage("bd2", StoreAndForwardCategory.ExternalSystem));
|
|
await _storage.EnqueueAsync(CreateMessage("bd3", StoreAndForwardCategory.Notification));
|
|
|
|
var depth = await _storage.GetBufferDepthByCategoryAsync();
|
|
Assert.True(depth.GetValueOrDefault(StoreAndForwardCategory.ExternalSystem) >= 2);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task GetMessageCountByOriginInstanceAsync_ReturnsCount()
|
|
{
|
|
var msg1 = CreateMessage("oi1", StoreAndForwardCategory.ExternalSystem);
|
|
msg1.OriginInstanceName = "Pump1";
|
|
await _storage.EnqueueAsync(msg1);
|
|
|
|
var msg2 = CreateMessage("oi2", StoreAndForwardCategory.Notification);
|
|
msg2.OriginInstanceName = "Pump1";
|
|
await _storage.EnqueueAsync(msg2);
|
|
|
|
var count = await _storage.GetMessageCountByOriginInstanceAsync("Pump1");
|
|
Assert.Equal(2, count);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task GetParkedMessagesAsync_Pagination()
|
|
{
|
|
for (int i = 0; i < 5; i++)
|
|
{
|
|
var msg = CreateMessage($"page{i}", StoreAndForwardCategory.ExternalSystem);
|
|
msg.Status = StoreAndForwardMessageStatus.Parked;
|
|
await _storage.EnqueueAsync(msg);
|
|
await _storage.UpdateMessageAsync(msg);
|
|
}
|
|
|
|
var (page1, total) = await _storage.GetParkedMessagesAsync(pageNumber: 1, pageSize: 2);
|
|
Assert.Equal(2, page1.Count);
|
|
Assert.True(total >= 5);
|
|
|
|
var (page2, _) = await _storage.GetParkedMessagesAsync(pageNumber: 2, pageSize: 2);
|
|
Assert.Equal(2, page2.Count);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task GetParkedMessagesAsync_TransactionedReads_CountMatchesFullResultSet()
|
|
{
|
|
// StoreAndForward-006: the COUNT(*) and paged SELECT now run inside one
|
|
// transaction so they share a consistent snapshot. This functional check
|
|
// guards the fix — it verifies the transaction wiring did not break paging:
|
|
// the reported TotalCount and the rows assembled across all pages agree, and
|
|
// a page wide enough to hold every parked row contains exactly TotalCount rows.
|
|
for (int i = 0; i < 25; i++)
|
|
{
|
|
var m = CreateMessage($"txn-{i}", StoreAndForwardCategory.ExternalSystem);
|
|
m.Status = StoreAndForwardMessageStatus.Parked;
|
|
await _storage.EnqueueAsync(m);
|
|
await _storage.UpdateMessageAsync(m);
|
|
}
|
|
|
|
var (wholePage, wholeTotal) = await _storage.GetParkedMessagesAsync(pageNumber: 1, pageSize: 1000);
|
|
Assert.Equal(25, wholeTotal);
|
|
Assert.Equal(wholeTotal, wholePage.Count);
|
|
|
|
var collected = new List<string>();
|
|
int reportedTotal = -1;
|
|
for (int page = 1; ; page++)
|
|
{
|
|
var (rows, total) = await _storage.GetParkedMessagesAsync(pageNumber: page, pageSize: 7);
|
|
reportedTotal = total;
|
|
collected.AddRange(rows.Select(r => r.Id));
|
|
if (rows.Count < 7) break;
|
|
}
|
|
Assert.Equal(reportedTotal, collected.Count);
|
|
Assert.Equal(25, collected.Distinct().Count());
|
|
}
|
|
|
|
[Fact]
|
|
public async Task GetMessageCountByStatusAsync_ReturnsAccurateCount()
|
|
{
|
|
var msg = CreateMessage("cnt1", StoreAndForwardCategory.ExternalSystem);
|
|
await _storage.EnqueueAsync(msg);
|
|
|
|
var count = await _storage.GetMessageCountByStatusAsync(StoreAndForwardMessageStatus.Pending);
|
|
Assert.True(count >= 1);
|
|
}
|
|
|
|
// ── Audit Log #23 (ExecutionId Task 4): execution_id / source_script ──
|
|
|
|
[Fact]
|
|
public async Task EnqueueAsync_RoundTripsExecutionIdAndSourceScript()
|
|
{
|
|
// A cached call buffered on a transient failure carries the originating
|
|
// script execution's ExecutionId + SourceScript; both must survive a
|
|
// persist + read-back so the retry loop can stamp them on audit rows.
|
|
var executionId = Guid.NewGuid();
|
|
var message = CreateMessage("exec1", StoreAndForwardCategory.ExternalSystem);
|
|
message.ExecutionId = executionId;
|
|
message.SourceScript = "Plant.Pump42/OnTick";
|
|
|
|
await _storage.EnqueueAsync(message);
|
|
|
|
var retrieved = await _storage.GetMessageByIdAsync("exec1");
|
|
Assert.NotNull(retrieved);
|
|
Assert.Equal(executionId, retrieved!.ExecutionId);
|
|
Assert.Equal("Plant.Pump42/OnTick", retrieved.SourceScript);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task EnqueueAsync_NullExecutionIdAndSourceScript_RoundTripAsNull()
|
|
{
|
|
// Non-cached-call enqueues (notifications) supply neither field — they
|
|
// must round-trip as null rather than throwing or coercing.
|
|
var message = CreateMessage("noexec1", StoreAndForwardCategory.Notification);
|
|
Assert.Null(message.ExecutionId);
|
|
Assert.Null(message.SourceScript);
|
|
|
|
await _storage.EnqueueAsync(message);
|
|
|
|
var retrieved = await _storage.GetMessageByIdAsync("noexec1");
|
|
Assert.NotNull(retrieved);
|
|
Assert.Null(retrieved!.ExecutionId);
|
|
Assert.Null(retrieved.SourceScript);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task ExecutionIdAndSourceScript_SurviveRetrySweepRead()
|
|
{
|
|
// The retry sweep reads due rows via GetMessagesForRetryAsync; the new
|
|
// fields must be present on that read path too (it is the path that
|
|
// feeds the CachedCallAttemptContext).
|
|
var executionId = Guid.NewGuid();
|
|
var message = CreateMessage("sweep1", StoreAndForwardCategory.CachedDbWrite);
|
|
message.ExecutionId = executionId;
|
|
message.SourceScript = "Plant.Tank/OnAlarm";
|
|
message.LastAttemptAt = null; // due immediately
|
|
await _storage.EnqueueAsync(message);
|
|
|
|
var due = await _storage.GetMessagesForRetryAsync();
|
|
|
|
var row = Assert.Single(due, m => m.Id == "sweep1");
|
|
Assert.Equal(executionId, row.ExecutionId);
|
|
Assert.Equal("Plant.Tank/OnAlarm", row.SourceScript);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task LegacyRowWithoutNewColumns_ReadsBackAsNull()
|
|
{
|
|
// Back-compat: a row persisted by a build that pre-dates the
|
|
// execution_id / source_script columns must still deserialize, with
|
|
// ExecutionId / SourceScript reading back as null. Simulate the legacy
|
|
// schema by dropping the table and recreating it without the columns,
|
|
// inserting directly, then running InitializeAsync (which ALTER-adds
|
|
// the columns) and reading the row back.
|
|
await using (var setup = _localDb.Db.CreateConnection())
|
|
{
|
|
await using var drop = setup.CreateCommand();
|
|
drop.CommandText = @"
|
|
DROP TABLE IF EXISTS sf_messages;
|
|
CREATE TABLE sf_messages (
|
|
id TEXT PRIMARY KEY,
|
|
category INTEGER NOT NULL,
|
|
target TEXT NOT NULL,
|
|
payload_json TEXT NOT NULL,
|
|
retry_count INTEGER NOT NULL DEFAULT 0,
|
|
max_retries INTEGER NOT NULL DEFAULT 50,
|
|
retry_interval_ms INTEGER NOT NULL DEFAULT 30000,
|
|
created_at TEXT NOT NULL,
|
|
last_attempt_at TEXT,
|
|
status INTEGER NOT NULL DEFAULT 0,
|
|
last_error TEXT,
|
|
origin_instance TEXT
|
|
);
|
|
INSERT INTO sf_messages (id, category, target, payload_json, created_at, status)
|
|
VALUES ('legacy1', 0, 'ERP', '{}', '2026-01-01T00:00:00.0000000+00:00', 0);";
|
|
await drop.ExecuteNonQueryAsync();
|
|
}
|
|
|
|
// InitializeAsync must additively ALTER-in the new columns without
|
|
// disturbing the pre-existing legacy row.
|
|
await _storage.InitializeAsync();
|
|
|
|
var retrieved = await _storage.GetMessageByIdAsync("legacy1");
|
|
Assert.NotNull(retrieved);
|
|
Assert.Equal("legacy1", retrieved!.Id);
|
|
Assert.Null(retrieved.ExecutionId);
|
|
Assert.Null(retrieved.SourceScript);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task MalformedExecutionId_ReadsBackAsNull_DoesNotAbortRetrySweep()
|
|
{
|
|
// Defensive read path: a corrupt (non-null, non-GUID) execution_id must
|
|
// be treated as "no execution id" rather than throwing FormatException
|
|
// — a single bad row must not abort the whole GetMessagesForRetryAsync
|
|
// sweep, which reads many rows. Persist two due rows, then corrupt the
|
|
// execution_id of one directly in the DB.
|
|
var goodId = Guid.NewGuid();
|
|
var good = CreateMessage("good1", StoreAndForwardCategory.ExternalSystem);
|
|
good.ExecutionId = goodId;
|
|
good.LastAttemptAt = null; // due immediately
|
|
await _storage.EnqueueAsync(good);
|
|
|
|
var bad = CreateMessage("bad1", StoreAndForwardCategory.ExternalSystem);
|
|
bad.ExecutionId = Guid.NewGuid();
|
|
bad.LastAttemptAt = null; // due immediately
|
|
await _storage.EnqueueAsync(bad);
|
|
|
|
await using (var conn = _localDb.Db.CreateConnection())
|
|
{
|
|
await using var corrupt = conn.CreateCommand();
|
|
corrupt.CommandText =
|
|
"UPDATE sf_messages SET execution_id = 'not-a-guid' WHERE id = 'bad1';";
|
|
await corrupt.ExecuteNonQueryAsync();
|
|
}
|
|
|
|
// The sweep must not throw; the corrupt row reads back with a null
|
|
// ExecutionId, the well-formed row keeps its value.
|
|
var due = await _storage.GetMessagesForRetryAsync();
|
|
Assert.Null(Assert.Single(due, m => m.Id == "bad1").ExecutionId);
|
|
Assert.Equal(goodId, Assert.Single(due, m => m.Id == "good1").ExecutionId);
|
|
|
|
// The single-row read path is equally defensive.
|
|
var retrieved = await _storage.GetMessageByIdAsync("bad1");
|
|
Assert.NotNull(retrieved);
|
|
Assert.Null(retrieved!.ExecutionId);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task InitializeAsync_IsIdempotent_WhenColumnsAlreadyExist()
|
|
{
|
|
// The additive ALTER must not fail on a second InitializeAsync call
|
|
// (SQLite has no ADD COLUMN IF NOT EXISTS — the probe must skip it).
|
|
await _storage.InitializeAsync();
|
|
await _storage.InitializeAsync();
|
|
|
|
var message = CreateMessage("idem1", StoreAndForwardCategory.ExternalSystem);
|
|
message.ExecutionId = Guid.NewGuid();
|
|
await _storage.EnqueueAsync(message);
|
|
var retrieved = await _storage.GetMessageByIdAsync("idem1");
|
|
Assert.NotNull(retrieved);
|
|
Assert.Equal(message.ExecutionId, retrieved!.ExecutionId);
|
|
}
|
|
|
|
// ── Audit Log #23 (ParentExecutionId Task 6): parent_execution_id ──
|
|
|
|
[Fact]
|
|
public async Task EnqueueAsync_RoundTripsParentExecutionId()
|
|
{
|
|
// A cached call buffered from an inbound-API-routed script run carries
|
|
// the spawning execution's ParentExecutionId; it must survive a persist
|
|
// + read-back so the retry loop can stamp it on audit rows.
|
|
var parentExecutionId = Guid.NewGuid();
|
|
var message = CreateMessage("parent1", StoreAndForwardCategory.ExternalSystem);
|
|
message.ParentExecutionId = parentExecutionId;
|
|
|
|
await _storage.EnqueueAsync(message);
|
|
|
|
var retrieved = await _storage.GetMessageByIdAsync("parent1");
|
|
Assert.NotNull(retrieved);
|
|
Assert.Equal(parentExecutionId, retrieved!.ParentExecutionId);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task EnqueueAsync_NullParentExecutionId_RoundTripsAsNull()
|
|
{
|
|
// A non-routed run supplies no ParentExecutionId — it must round-trip
|
|
// as null rather than throwing or coercing.
|
|
var message = CreateMessage("noparent1", StoreAndForwardCategory.ExternalSystem);
|
|
Assert.Null(message.ParentExecutionId);
|
|
|
|
await _storage.EnqueueAsync(message);
|
|
|
|
var retrieved = await _storage.GetMessageByIdAsync("noparent1");
|
|
Assert.NotNull(retrieved);
|
|
Assert.Null(retrieved!.ParentExecutionId);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task ParentExecutionId_SurvivesRetrySweepRead()
|
|
{
|
|
// The retry sweep reads due rows via GetMessagesForRetryAsync; the new
|
|
// parent_execution_id field must be present on that read path too — it
|
|
// is the path that feeds the CachedCallAttemptContext.
|
|
var parentExecutionId = Guid.NewGuid();
|
|
var message = CreateMessage("psweep1", StoreAndForwardCategory.CachedDbWrite);
|
|
message.ParentExecutionId = parentExecutionId;
|
|
message.LastAttemptAt = null; // due immediately
|
|
await _storage.EnqueueAsync(message);
|
|
|
|
var due = await _storage.GetMessagesForRetryAsync();
|
|
|
|
var row = Assert.Single(due, m => m.Id == "psweep1");
|
|
Assert.Equal(parentExecutionId, row.ParentExecutionId);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task LegacyRowWithoutParentExecutionIdColumn_ReadsBackAsNull()
|
|
{
|
|
// Back-compat: a row persisted by a build that pre-dates the
|
|
// parent_execution_id column must still deserialize, with
|
|
// ParentExecutionId reading back as null. Simulate the pre-Task-6
|
|
// schema (which already has execution_id / source_script from the
|
|
// ExecutionId rollout) by recreating the table without
|
|
// parent_execution_id, inserting directly, then running InitializeAsync
|
|
// which ALTER-adds the column.
|
|
await using (var setup = _localDb.Db.CreateConnection())
|
|
{
|
|
await using var drop = setup.CreateCommand();
|
|
drop.CommandText = @"
|
|
DROP TABLE IF EXISTS sf_messages;
|
|
CREATE TABLE sf_messages (
|
|
id TEXT PRIMARY KEY,
|
|
category INTEGER NOT NULL,
|
|
target TEXT NOT NULL,
|
|
payload_json TEXT NOT NULL,
|
|
retry_count INTEGER NOT NULL DEFAULT 0,
|
|
max_retries INTEGER NOT NULL DEFAULT 50,
|
|
retry_interval_ms INTEGER NOT NULL DEFAULT 30000,
|
|
created_at TEXT NOT NULL,
|
|
last_attempt_at TEXT,
|
|
status INTEGER NOT NULL DEFAULT 0,
|
|
last_error TEXT,
|
|
origin_instance TEXT,
|
|
execution_id TEXT,
|
|
source_script TEXT
|
|
);
|
|
INSERT INTO sf_messages (id, category, target, payload_json, created_at, status)
|
|
VALUES ('plegacy1', 0, 'ERP', '{}', '2026-01-01T00:00:00.0000000+00:00', 0);";
|
|
await drop.ExecuteNonQueryAsync();
|
|
}
|
|
|
|
// InitializeAsync must additively ALTER-in parent_execution_id without
|
|
// disturbing the pre-existing legacy row.
|
|
await _storage.InitializeAsync();
|
|
|
|
var retrieved = await _storage.GetMessageByIdAsync("plegacy1");
|
|
Assert.NotNull(retrieved);
|
|
Assert.Equal("plegacy1", retrieved!.Id);
|
|
Assert.Null(retrieved.ParentExecutionId);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task MalformedParentExecutionId_ReadsBackAsNull_DoesNotAbortRetrySweep()
|
|
{
|
|
// Defensive read path: a corrupt (non-null, non-GUID) parent_execution_id
|
|
// must be treated as "no parent execution id" rather than throwing
|
|
// FormatException — a single bad row must not abort the whole
|
|
// GetMessagesForRetryAsync sweep.
|
|
var goodParent = Guid.NewGuid();
|
|
var good = CreateMessage("pgood1", StoreAndForwardCategory.ExternalSystem);
|
|
good.ParentExecutionId = goodParent;
|
|
good.LastAttemptAt = null; // due immediately
|
|
await _storage.EnqueueAsync(good);
|
|
|
|
var bad = CreateMessage("pbad1", StoreAndForwardCategory.ExternalSystem);
|
|
bad.ParentExecutionId = Guid.NewGuid();
|
|
bad.LastAttemptAt = null; // due immediately
|
|
await _storage.EnqueueAsync(bad);
|
|
|
|
await using (var conn = _localDb.Db.CreateConnection())
|
|
{
|
|
await using var corrupt = conn.CreateCommand();
|
|
corrupt.CommandText =
|
|
"UPDATE sf_messages SET parent_execution_id = 'not-a-guid' WHERE id = 'pbad1';";
|
|
await corrupt.ExecuteNonQueryAsync();
|
|
}
|
|
|
|
var due = await _storage.GetMessagesForRetryAsync();
|
|
Assert.Null(Assert.Single(due, m => m.Id == "pbad1").ParentExecutionId);
|
|
Assert.Equal(goodParent, Assert.Single(due, m => m.Id == "pgood1").ParentExecutionId);
|
|
|
|
var retrieved = await _storage.GetMessageByIdAsync("pbad1");
|
|
Assert.NotNull(retrieved);
|
|
Assert.Null(retrieved!.ParentExecutionId);
|
|
}
|
|
|
|
// ── Task 7 (arch review 02, Performance): bounded due-rows query ──
|
|
|
|
[Fact]
|
|
public async Task GetMessagesForRetry_HonorsLimit_OldestFirst()
|
|
{
|
|
var baseTime = DateTimeOffset.UtcNow.AddMinutes(-10);
|
|
for (var i = 0; i < 5; i++)
|
|
{
|
|
await _storage.EnqueueAsync(new StoreAndForwardMessage
|
|
{
|
|
Id = $"m{i}",
|
|
Category = StoreAndForwardCategory.ExternalSystem,
|
|
Target = "t",
|
|
PayloadJson = "{}",
|
|
RetryCount = 0,
|
|
MaxRetries = 50,
|
|
RetryIntervalMs = 0, // always due
|
|
CreatedAt = baseTime.AddSeconds(i),
|
|
Status = StoreAndForwardMessageStatus.Pending
|
|
});
|
|
}
|
|
|
|
var page = await _storage.GetMessagesForRetryAsync(limit: 3);
|
|
Assert.Equal(3, page.Count);
|
|
Assert.Equal(new[] { "m0", "m1", "m2" }, page.Select(m => m.Id));
|
|
}
|
|
|
|
private static StoreAndForwardMessage CreateMessage(string id, StoreAndForwardCategory category)
|
|
{
|
|
return new StoreAndForwardMessage
|
|
{
|
|
Id = id,
|
|
Category = category,
|
|
Target = "target1",
|
|
PayloadJson = """{"method":"Test","args":{}}""",
|
|
RetryCount = 0,
|
|
MaxRetries = 50,
|
|
RetryIntervalMs = 30000,
|
|
CreatedAt = DateTimeOffset.UtcNow,
|
|
Status = StoreAndForwardMessageStatus.Pending
|
|
};
|
|
}
|
|
|
|
// ── Invariants that moved owner ──
|
|
//
|
|
// Directory creation and WAL journal mode used to be this storage class's own job
|
|
// (EnsureDatabaseDirectoryExists / an explicit PRAGMA in InitializeAsync). Both moved
|
|
// out when the store stopped owning its file. They landed in DIFFERENT places, so the
|
|
// coverage split rather than moving wholesale:
|
|
//
|
|
// * WAL is genuinely LocalDb's — it sets the journal mode on the database it owns.
|
|
// The test below still asserts it, now against the LocalDb-backed store.
|
|
//
|
|
// * Directory creation was NOT LocalDb's when Phase 2 landed: the library did not
|
|
// create the parent directory yet opened the file eagerly, so a missing directory
|
|
// was a hard boot failure, and the Host re-established the guarantee with a
|
|
// SiteLocalDbDirectory shim. LocalDb 0.1.1 took the guarantee back into the
|
|
// library, so the shim is gone — but either way it is not this class's to assert,
|
|
// and SiteLocalDbDirectoryTests pins the outcome for the site host.
|
|
|
|
[Fact]
|
|
public async Task Initialize_LeavesTheDatabaseInWalJournalMode()
|
|
{
|
|
// WAL lets the retry-sweep lanes, script enqueues, and standby replication
|
|
// applies read/write concurrently without "database is locked".
|
|
using var localDb = TestLocalDb.CreateTemp("sf-wal-test");
|
|
var path = localDb.Path;
|
|
try
|
|
{
|
|
var storage = new StoreAndForwardStorage(
|
|
localDb.Db, NullLogger<StoreAndForwardStorage>.Instance);
|
|
await storage.InitializeAsync();
|
|
|
|
// journal_mode is persistent + file-scoped, so any connection observes it.
|
|
await using var conn = localDb.Db.CreateConnection();
|
|
await using var cmd = conn.CreateCommand();
|
|
cmd.CommandText = "PRAGMA journal_mode";
|
|
Assert.Equal("wal", (string)(await cmd.ExecuteScalarAsync())!);
|
|
}
|
|
finally
|
|
{
|
|
localDb.Dispose();
|
|
TestLocalDb.DeleteFiles(path);
|
|
}
|
|
}
|
|
|
|
// ── 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 = _localDb.Db.CreateConnection();
|
|
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 = _localDb.Db.CreateConnection();
|
|
await using var cmd = conn.CreateCommand();
|
|
cmd.CommandText = sql;
|
|
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
|
|
}
|
|
|
|
// ReplaceAll_SwapsTheEntireBuffer_Atomically was DELETED with ReplaceAllAsync in
|
|
// LocalDb Phase 2, and deliberately not replaced. It asserted a destructive
|
|
// delete-all-then-insert-all, which was the standby's anti-entropy apply: a resync
|
|
// overwrote the standby's copy with the active node's snapshot. sf_messages is now a
|
|
// replicated table, so a mass DELETE would be CAPTURED and shipped to the peer — the
|
|
// method is not merely unused, it is unsafe to keep. LocalDb's own snapshot resync
|
|
// merges per row under last-writer-wins and never deletes, which is what
|
|
// LocalDbConfigConvergenceTests.ANodeWithNewerLocalRows_KeepsThem_WhenAPeerSnapshotArrives
|
|
// now covers.
|
|
|
|
// ── Task 23: oldest-parked-age health signal ──
|
|
|
|
[Fact]
|
|
public async Task GetOldestParkedCreatedAt_ReturnsOldestParkedRow_OrNull()
|
|
{
|
|
Assert.Null(await _storage.GetOldestParkedCreatedAtAsync());
|
|
|
|
var old = NewMsg("old-parked", StoreAndForwardMessageStatus.Parked, DateTimeOffset.UtcNow.AddDays(-3));
|
|
var recent = NewMsg("new-parked", StoreAndForwardMessageStatus.Parked, DateTimeOffset.UtcNow.AddHours(-1));
|
|
var pending = NewMsg("pending", StoreAndForwardMessageStatus.Pending, DateTimeOffset.UtcNow.AddDays(-9)); // must not count
|
|
await _storage.EnqueueAsync(old);
|
|
await _storage.EnqueueAsync(recent);
|
|
await _storage.EnqueueAsync(pending);
|
|
|
|
var oldest = await _storage.GetOldestParkedCreatedAtAsync();
|
|
Assert.Equal(old.CreatedAt, oldest!.Value, TimeSpan.FromSeconds(1));
|
|
}
|
|
}
|