using Akka.Actor;
using Akka.TestKit.Xunit2;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options;
using ZB.MOM.WW.ScadaBridge.AuditLog.Central;
using ZB.MOM.WW.Audit;
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Entities;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Integration;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase;
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Repositories;
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests.Migrations;
namespace ZB.MOM.WW.ScadaBridge.AuditLog.Tests.Central;
///
/// Tests for . Most
/// tests substitute the with an in-memory
/// recording stub so the actor's tick / cursor / stalled state machinery can
/// be exercised in milliseconds without an MSSQL container. The duplicate /
/// idempotency assertion uses the real against
/// the so we verify InsertIfNotExistsAsync
/// actually swallows duplicate-key collisions (the race fix the
/// reconciliation puller depends on).
///
public class SiteAuditReconciliationActorTests : TestKit, IClassFixture
{
private readonly MsSqlMigrationFixture _fixture;
/// Initializes the test class with the shared MSSQL migration fixture.
/// The shared MSSQL migration fixture.
public SiteAuditReconciliationActorTests(MsSqlMigrationFixture fixture)
{
_fixture = fixture;
}
private static AuditEvent NewEvent(
string siteId,
DateTime? occurredAt = null,
Guid? id = null) => ScadaBridgeAuditEventFactory.Create(
eventId: id ?? Guid.NewGuid(),
occurredAtUtc: occurredAt ?? new DateTime(2026, 5, 20, 10, 0, 0, DateTimeKind.Utc),
channel: AuditChannel.ApiOutbound,
kind: AuditKind.ApiCall,
status: AuditStatus.Delivered,
sourceSiteId: siteId);
private static SiteAuditReconciliationOptions FastTickOptions(
int batchSize = 256,
int stalledAfter = 2) =>
new()
{
// 100 ms tick keeps each test under a second. AwaitAssert covers
// schedule jitter so a 100 ms tick has up to ~3 s to fire.
ReconciliationIntervalSeconds = 300,
ReconciliationIntervalOverride = TimeSpan.FromMilliseconds(100),
BatchSize = batchSize,
StalledAfterNonDrainingCycles = stalledAfter,
};
///
/// In-memory recording stub used for non-MSSQL tests. Captures every
/// call AND deduplicates on
/// so duplicate-handling assertions don't
/// need a real database for the simple cases.
///
private sealed class RecordingRepo : IAuditLogRepository
{
///
public Task> GetTraceEventsAsync(Guid executionId, CancellationToken ct = default, IReadOnlyList? permittedSiteIds = null)
=> throw new NotSupportedException("GetTraceEventsAsync is not exercised by this test double.");
/// Gets the audit events recorded by this stub.
public List Inserted { get; } = new();
private readonly HashSet _seen = new();
/// Gets the number of times InsertIfNotExistsAsync was called.
public int InsertCallCount { get; private set; }
///
public Task InsertIfNotExistsAsync(AuditEvent evt, CancellationToken ct = default)
{
InsertCallCount++;
if (_seen.Add(evt.EventId))
{
Inserted.Add(evt);
}
return Task.CompletedTask;
}
///
public Task> QueryAsync(
AuditLogQueryFilter filter, AuditLogPaging paging, CancellationToken ct = default) =>
Task.FromResult>(Inserted);
///
public Task SwitchOutPartitionAsync(DateTime monthBoundary, TimeSpan? commandTimeout = null, CancellationToken ct = default) =>
Task.FromResult(0L);
///
public Task PurgeChannelOlderThanAsync(
string channel, DateTime threshold, int batchSize, TimeSpan? commandTimeout = null, CancellationToken ct = default, int maxRows = 50_000) =>
Task.FromResult(0L);
///
public Task BackfillSourceNodeAsync(
string sentinel, DateTime before, int batchSize, CancellationToken ct = default) =>
Task.FromResult(0L);
///
public Task> GetPartitionBoundariesOlderThanAsync(
DateTime threshold, CancellationToken ct = default) =>
Task.FromResult>(Array.Empty());
///
public Task GetKpiSnapshotAsync(
TimeSpan window, DateTime? nowUtc = null, CancellationToken ct = default) =>
Task.FromResult(new ZB.MOM.WW.ScadaBridge.Commons.Types.AuditLogKpiSnapshot(0L, 0L, 0L, nowUtc ?? DateTime.UtcNow));
///
public Task> GetExecutionTreeAsync(
Guid executionId, CancellationToken ct = default, IReadOnlyList? permittedSiteIds = null) =>
Task.FromResult>(Array.Empty());
///
public Task> GetDistinctSourceNodesAsync(CancellationToken ct = default) =>
Task.FromResult>(Array.Empty());
}
///
/// Recording stub whose throws for ONE
/// poison (every attempt) but records every
/// other row — including the synthetic ReconciliationAbandoned row the actor
/// emits once it gives up on the poison row.
///
private sealed class PoisonInsertRepo : IAuditLogRepository
{
///
public Task> GetTraceEventsAsync(Guid executionId, CancellationToken ct = default, IReadOnlyList? permittedSiteIds = null)
=> throw new NotSupportedException("GetTraceEventsAsync is not exercised by this test double.");
private readonly Guid _poisonId;
private readonly Exception _fault;
/// Gets the audit events recorded by this stub.
public List Inserted { get; } = new();
private readonly HashSet _seen = new();
/// Initializes the repository so that inserting the given event id always throws the given fault.
/// The event id whose insert always fails.
/// The exception thrown on every attempt to insert the poison row.
public PoisonInsertRepo(Guid poisonId, Exception fault)
{
_poisonId = poisonId;
_fault = fault;
}
///
public Task InsertIfNotExistsAsync(AuditEvent evt, CancellationToken ct = default)
{
if (evt.EventId == _poisonId)
{
throw _fault;
}
if (_seen.Add(evt.EventId))
{
Inserted.Add(evt);
}
return Task.CompletedTask;
}
///
public Task> QueryAsync(
AuditLogQueryFilter filter, AuditLogPaging paging, CancellationToken ct = default) =>
Task.FromResult>(Inserted);
///
public Task SwitchOutPartitionAsync(DateTime monthBoundary, TimeSpan? commandTimeout = null, CancellationToken ct = default) =>
Task.FromResult(0L);
///
public Task PurgeChannelOlderThanAsync(
string channel, DateTime threshold, int batchSize, TimeSpan? commandTimeout = null, CancellationToken ct = default, int maxRows = 50_000) =>
Task.FromResult(0L);
///
public Task BackfillSourceNodeAsync(
string sentinel, DateTime before, int batchSize, CancellationToken ct = default) =>
Task.FromResult(0L);
///
public Task> GetPartitionBoundariesOlderThanAsync(
DateTime threshold, CancellationToken ct = default) =>
Task.FromResult>(Array.Empty());
///
public Task GetKpiSnapshotAsync(
TimeSpan window, DateTime? nowUtc = null, CancellationToken ct = default) =>
Task.FromResult(new ZB.MOM.WW.ScadaBridge.Commons.Types.AuditLogKpiSnapshot(0L, 0L, 0L, nowUtc ?? DateTime.UtcNow));
///
public Task> GetExecutionTreeAsync(
Guid executionId, CancellationToken ct = default, IReadOnlyList? permittedSiteIds = null) =>
Task.FromResult>(Array.Empty());
///
public Task> GetDistinctSourceNodesAsync(CancellationToken ct = default) =>
Task.FromResult>(Array.Empty());
}
///
/// Recording stub whose throws for EVERY
/// row except a synthetic row
/// (identified by its Action suffix, mirroring 's
/// assertion pattern). Used to model a site whose backlog contains one row that
/// NEVER inserts (the eventual permanent-abandon target) alongside a co-batch
/// row that also never inserts but changes EventId every pull — i.e. it never
/// individually reaches MaxPermanentInsertAttempts, so it keeps
/// hasUnresolvedFailure true and holds the cursor at since forever
/// (the defect trigger). Dedupes inserted rows on
/// exactly like the real InsertIfNotExistsAsync, so a test asserting on
/// 's count is exercising the same idempotency contract
/// production relies on.
///
private sealed class AlwaysFailUnlessSyntheticRepo : IAuditLogRepository
{
///
public Task> GetTraceEventsAsync(Guid executionId, CancellationToken ct = default, IReadOnlyList? permittedSiteIds = null)
=> throw new NotSupportedException("GetTraceEventsAsync is not exercised by this test double.");
/// Gets the audit events recorded by this stub.
public List Inserted { get; } = new();
private readonly HashSet _seen = new();
///
public Task InsertIfNotExistsAsync(AuditEvent evt, CancellationToken ct = default)
{
if (!evt.Action.EndsWith(".ReconciliationAbandoned", StringComparison.Ordinal))
{
throw new InvalidOperationException("simulated permanent insert failure");
}
if (_seen.Add(evt.EventId))
{
Inserted.Add(evt);
}
return Task.CompletedTask;
}
///
public Task> QueryAsync(
AuditLogQueryFilter filter, AuditLogPaging paging, CancellationToken ct = default) =>
Task.FromResult>(Inserted);
///
public Task SwitchOutPartitionAsync(DateTime monthBoundary, TimeSpan? commandTimeout = null, CancellationToken ct = default) =>
Task.FromResult(0L);
///
public Task PurgeChannelOlderThanAsync(
string channel, DateTime threshold, int batchSize, TimeSpan? commandTimeout = null, CancellationToken ct = default, int maxRows = 50_000) =>
Task.FromResult(0L);
///
public Task BackfillSourceNodeAsync(
string sentinel, DateTime before, int batchSize, CancellationToken ct = default) =>
Task.FromResult(0L);
///
public Task> GetPartitionBoundariesOlderThanAsync(
DateTime threshold, CancellationToken ct = default) =>
Task.FromResult>(Array.Empty());
///
public Task GetKpiSnapshotAsync(
TimeSpan window, DateTime? nowUtc = null, CancellationToken ct = default) =>
Task.FromResult(new ZB.MOM.WW.ScadaBridge.Commons.Types.AuditLogKpiSnapshot(0L, 0L, 0L, nowUtc ?? DateTime.UtcNow));
///
public Task> GetExecutionTreeAsync(
Guid executionId, CancellationToken ct = default, IReadOnlyList? permittedSiteIds = null) =>
Task.FromResult>(Array.Empty());
///
public Task> GetDistinctSourceNodesAsync(CancellationToken ct = default) =>
Task.FromResult>(Array.Empty());
}
///
/// In-memory enumerator returning a static list of sites.
///
private sealed class StaticEnumerator : ISiteEnumerator
{
private readonly IReadOnlyList _sites;
/// Initializes the enumerator with a fixed set of sites.
/// The sites returned on every enumeration.
public StaticEnumerator(params SiteEntry[] sites) => _sites = sites;
///
public Task> EnumerateAsync(CancellationToken ct = default) =>
Task.FromResult(_sites);
}
///
/// Scripted pull client — returns the next queued response for the site
/// on each call, looping the last entry if the queue is exhausted. Also
/// records every invocation so tests can assert call counts + arguments.
///
private sealed class ScriptedPullClient : IPullAuditEventsClient
{
/// Gets the recorded pull calls in the order they were made.
public List<(string SiteId, DateTime SinceUtc, string? AfterId, int BatchSize)> Calls { get; } = new();
private readonly Dictionary> _scripted = new();
private readonly Dictionary _throwOnSite = new();
/// Queues scripted responses for the given site.
/// The site whose pulls receive the scripted responses.
/// The responses returned in order on successive pulls.
/// This client, for chaining.
public ScriptedPullClient Script(string siteId, params PullAuditEventsResponse[] responses)
{
_scripted[siteId] = new Queue(responses);
return this;
}
/// Makes every pull for the given site throw the given exception.
/// The site whose pulls fail.
/// The exception to throw.
/// This client, for chaining.
public ScriptedPullClient ThrowFor(string siteId, Exception ex)
{
_throwOnSite[siteId] = ex;
return this;
}
///
public Task PullAsync(
string siteId, DateTime sinceUtc, string? afterId, int batchSize, CancellationToken ct)
{
Calls.Add((siteId, sinceUtc, afterId, batchSize));
if (_throwOnSite.TryGetValue(siteId, out var ex))
{
throw ex;
}
if (_scripted.TryGetValue(siteId, out var queue) && queue.Count > 0)
{
return Task.FromResult(queue.Dequeue());
}
return Task.FromResult(
new PullAuditEventsResponse(Array.Empty(), MoreAvailable: false));
}
}
private IServiceProvider BuildScopedProvider(IAuditLogRepository repo)
{
var services = new ServiceCollection();
// The actor opens a scope per tick and resolves IAuditLogRepository
// from that scope; registering as scoped mirrors how
// AddConfigurationDatabase wires the real repository.
services.AddScoped(_ => repo);
return services.BuildServiceProvider();
}
private IActorRef CreateActor(
ISiteEnumerator sites,
IPullAuditEventsClient client,
IAuditLogRepository repo,
SiteAuditReconciliationOptions options)
{
var sp = BuildScopedProvider(repo);
return Sys.ActorOf(Props.Create(() => new SiteAuditReconciliationActor(
sites,
client,
sp,
Options.Create(options),
NullLogger.Instance)));
}
///
/// Stops a reconciliation actor and blocks until it is actually dead.
///
///
///
/// Deferred-work register row 95, the same leak shape row 82 fixed in
/// PartitionPurgeTests / NotifyDispatcherAuditTrailTests. Every actor this
/// class creates ticks on a ~0 s repeat interval () and was
/// only ever stopped implicitly by 's Dispose →
/// Shutdown, which is bounded by a timeout and does NOT fail when it expires. A
/// survivor keeps pulling and INSERTING audit rows: in
/// Tick_Duplicates_NotDoubleInserted it holds a real
/// over the shared
/// database (and over a DbContext the test is about to dispose), so it writes into
/// the DB LATER test classes read from.
///
///
/// Stopping the actor inside the test that created it removes the dependency on
/// ActorSystem-shutdown timing entirely.
///
///
private void StopReconciliationActor(IActorRef actor)
{
Watch(actor);
Sys.Stop(actor);
// A stop is processed after the in-flight tick completes, so this also guarantees no
// pull or insert is still running against the fixture DB.
ExpectTerminated(actor, TimeSpan.FromSeconds(30));
Unwatch(actor);
}
///
/// Subscribes to the EventStream and collects every
/// publication into a list
/// the test can assert on. Uses a probe actor so the stream's
/// fire-and-forget delivery is observable from the test thread.
///
private (Akka.TestKit.TestProbe Probe, List Captured) SubscribeStalled()
{
var probe = CreateTestProbe();
Sys.EventStream.Subscribe(probe.Ref, typeof(SiteAuditTelemetryStalledChanged));
var captured = new List();
return (probe, captured);
}
// ---------------------------------------------------------------------
// 1. Timer_Fires_OnConfiguredInterval
// ---------------------------------------------------------------------
/// Verifies that the reconciliation timer fires on the configured interval.
[Fact]
public void Timer_Fires_OnConfiguredInterval()
{
var sites = new StaticEnumerator(new SiteEntry("siteA", "http://siteA:8083"));
var client = new ScriptedPullClient();
var repo = new RecordingRepo();
var opts = FastTickOptions();
var actor = CreateActor(sites, client, repo, opts);
// The first scheduled tick fires after `ReconciliationIntervalSeconds`,
// which is 0 for the test — Akka's scheduler still respects the
// ScheduleTellRepeatedlyCancelable contract that issues a Tell on the
// scheduler thread, so we await visible side effects (a PullAsync call)
// rather than racing on internal state.
AwaitAssert(
() => Assert.True(client.Calls.Count >= 1, $"expected >= 1 pull call, got {client.Calls.Count}"),
duration: TimeSpan.FromSeconds(3),
interval: TimeSpan.FromMilliseconds(50));
StopReconciliationActor(actor);
}
// ---------------------------------------------------------------------
// 2. Tick_PullsFromEachKnownSite
// ---------------------------------------------------------------------
/// Verifies that each tick pulls from every known site.
[Fact]
public void Tick_PullsFromEachKnownSite()
{
var sites = new StaticEnumerator(
new SiteEntry("siteA", "http://siteA:8083"),
new SiteEntry("siteB", "http://siteB:8083"));
var client = new ScriptedPullClient();
var repo = new RecordingRepo();
var actor = CreateActor(sites, client, repo, FastTickOptions());
AwaitAssert(() =>
{
Assert.Contains(client.Calls, c => c.SiteId == "siteA");
Assert.Contains(client.Calls, c => c.SiteId == "siteB");
},
duration: TimeSpan.FromSeconds(3),
interval: TimeSpan.FromMilliseconds(50));
StopReconciliationActor(actor);
}
// ---------------------------------------------------------------------
// 3. Tick_IngestEvents_ViaInsertIfNotExistsAsync
// ---------------------------------------------------------------------
/// Verifies that pulled events are ingested through the repository insert-if-not-exists call.
[Fact]
public void Tick_IngestEvents_ViaInsertIfNotExistsAsync()
{
var sites = new StaticEnumerator(new SiteEntry("siteA", "http://siteA:8083"));
var e1 = NewEvent("siteA");
var e2 = NewEvent("siteA");
var client = new ScriptedPullClient().Script("siteA",
new PullAuditEventsResponse(new[] { e1, e2 }, MoreAvailable: false));
var repo = new RecordingRepo();
var actor = CreateActor(sites, client, repo, FastTickOptions());
AwaitAssert(() => Assert.Equal(2, repo.InsertCallCount),
duration: TimeSpan.FromSeconds(3),
interval: TimeSpan.FromMilliseconds(50));
StopReconciliationActor(actor);
Assert.Contains(repo.Inserted, e => e.EventId == e1.EventId);
Assert.Contains(repo.Inserted, e => e.EventId == e2.EventId);
}
// ---------------------------------------------------------------------
// 4. Tick_Duplicates_NotDoubleInserted (real MSSQL idempotency)
// ---------------------------------------------------------------------
private ScadaBridgeDbContext CreateContext() =>
new(new DbContextOptionsBuilder()
.UseSqlServer(_fixture.ConnectionString).Options);
/// Verifies that duplicate events are not inserted twice.
/// A task that represents the asynchronous operation.
[SkippableFact]
public async Task Tick_Duplicates_NotDoubleInserted()
{
Skip.IfNot(_fixture.Available, _fixture.SkipReason);
var siteId = "bundle-b-" + Guid.NewGuid().ToString("N").Substring(0, 8);
var pre = NewEvent(siteId);
// Seed the row directly so the actor sees it already present when the
// pull returns it.
await using (var seedContext = CreateContext())
{
await new AuditLogRepository(seedContext).InsertIfNotExistsAsync(pre);
}
// Stack one new and the pre-existing row in the pull response. The
// second-pull script returns empty so the actor settles.
var fresh = NewEvent(siteId);
var sites = new StaticEnumerator(new SiteEntry(siteId, "http://x:8083"));
var client = new ScriptedPullClient().Script(siteId,
new PullAuditEventsResponse(new[] { pre, fresh }, MoreAvailable: false));
await using var context = CreateContext();
var repo = new AuditLogRepository(context);
var actor = CreateActor(sites, client, repo, FastTickOptions());
// Wait for the actor to ingest both rows.
await Task.Delay(TimeSpan.FromSeconds(1));
AwaitAssert(() => Assert.True(client.Calls.Count >= 1),
duration: TimeSpan.FromSeconds(3));
// Stop before reading: this actor holds a real repository over the SHARED fixture
// database (and over the DbContext this test is about to dispose), so it must not
// outlive the test (register row 95).
StopReconciliationActor(actor);
// Even though the pull returned 2 events, only 1 fresh row should
// exist in MSSQL alongside the pre-existing one — InsertIfNotExistsAsync
// is first-write-wins on EventId.
await using var read = CreateContext();
var rows = await read.Set()
.Where(e => e.SourceSiteId == siteId)
.ToListAsync();
Assert.Equal(2, rows.Count);
Assert.Contains(rows, r => r.EventId == pre.EventId);
Assert.Contains(rows, r => r.EventId == fresh.EventId);
}
// ---------------------------------------------------------------------
// 5. Cursor_Advances_ToMaxOccurredAtUtc
// ---------------------------------------------------------------------
/// Verifies that the cursor advances to the maximum occurred-at timestamp of the pulled events.
[Fact]
public void Cursor_Advances_ToMaxOccurredAtUtc()
{
var sites = new StaticEnumerator(new SiteEntry("siteA", "http://siteA:8083"));
var t1 = new DateTime(2026, 5, 20, 10, 0, 0, DateTimeKind.Utc);
var t2 = new DateTime(2026, 5, 20, 10, 1, 0, DateTimeKind.Utc);
var t3 = new DateTime(2026, 5, 20, 10, 2, 0, DateTimeKind.Utc);
var e1 = NewEvent("siteA", t1);
var e2 = NewEvent("siteA", t2);
var e3 = NewEvent("siteA", t3);
// First pull returns three events with t1, t2, t3. Subsequent pulls
// return empty — but the test asserts the SECOND pull's since argument
// is t3 (the max OccurredAtUtc from the first pull).
var client = new ScriptedPullClient().Script("siteA",
new PullAuditEventsResponse(new[] { e1, e2, e3 }, MoreAvailable: false));
var repo = new RecordingRepo();
var actor = CreateActor(sites, client, repo, FastTickOptions());
// Wait until we have at least two pulls — the second one must use t3
// as its `since` argument because that was the max OccurredAtUtc in
// the first response.
AwaitAssert(() => Assert.True(client.Calls.Count >= 2,
$"need at least 2 pulls to assert cursor advancement, got {client.Calls.Count}"),
duration: TimeSpan.FromSeconds(5),
interval: TimeSpan.FromMilliseconds(50));
// Stop before indexing into client.Calls: the list must not grow mid-assertion
// (register row 95).
StopReconciliationActor(actor);
Assert.Equal(DateTime.MinValue, client.Calls[0].SinceUtc);
Assert.Equal(t3, client.Calls[1].SinceUtc);
// The composite half travels too: the first pull has
// no cursor at all, the second carries the id of the row at t3 so the site can retire
// it exactly instead of leaving the boundary instant permanently servable.
Assert.Null(client.Calls[0].AfterId);
Assert.Equal(e3.EventId.ToString(), client.Calls[1].AfterId);
}
/// Verifies that the cursor advances on the event id tiebreak when every row shares one instant.
[Fact]
public void Cursor_AdvancesOnTheIdTiebreak_WhenEveryRowSharesOneInstant()
{
// The case a bare timestamp cursor could never drain: a burst all stamped at the same
// instant. The timestamp cannot move, so only the id half can — and it must, or the
// next pull re-serves the identical window forever.
var sites = new StaticEnumerator(new SiteEntry("siteA", "http://siteA:8083"));
var t = new DateTime(2026, 5, 20, 10, 0, 0, DateTimeKind.Utc);
// Deterministic ids so "the greatest ordinal" is a fact, not a coin flip.
var low = NewEvent("siteA", t, Guid.Parse("11111111-1111-1111-1111-111111111111"));
var high = NewEvent("siteA", t, Guid.Parse("99999999-9999-9999-9999-999999999999"));
var mid = NewEvent("siteA", t, Guid.Parse("55555555-5555-5555-5555-555555555555"));
var client = new ScriptedPullClient().Script("siteA",
new PullAuditEventsResponse(new[] { low, high, mid }, MoreAvailable: true));
var repo = new RecordingRepo();
var actor = CreateActor(sites, client, repo, FastTickOptions());
AwaitAssert(() => Assert.True(client.Calls.Count >= 2,
$"need at least 2 pulls, got {client.Calls.Count}"),
duration: TimeSpan.FromSeconds(5),
interval: TimeSpan.FromMilliseconds(50));
StopReconciliationActor(actor);
Assert.Equal(t, client.Calls[1].SinceUtc);
Assert.Equal(high.EventId.ToString(), client.Calls[1].AfterId);
}
/// Verifies that both cursor halves are held back while a row is still being retried.
[Fact]
public void BothCursorHalves_AreHeldBack_WhileARowIsStillBeingRetried()
{
// A held-back cursor must hold BOTH halves: advancing the id while pinning the
// timestamp would skip past the very rows being retried.
var sites = new StaticEnumerator(new SiteEntry("siteA", "http://siteA:8083"));
var t = new DateTime(2026, 5, 20, 10, 0, 0, DateTimeKind.Utc);
var evt = NewEvent("siteA", t);
var client = new ScriptedPullClient().Script("siteA",
new PullAuditEventsResponse(new[] { evt }, MoreAvailable: false));
var repo = new AlwaysThrowingRepo();
var actor = CreateActor(sites, client, repo, FastTickOptions());
AwaitAssert(() => Assert.True(client.Calls.Count >= 2,
$"need at least 2 pulls, got {client.Calls.Count}"),
duration: TimeSpan.FromSeconds(5),
interval: TimeSpan.FromMilliseconds(50));
StopReconciliationActor(actor);
Assert.Equal(DateTime.MinValue, client.Calls[1].SinceUtc);
Assert.Null(client.Calls[1].AfterId);
}
/// Repository whose every insert throws, so the retry hold-back path is taken.
private sealed class AlwaysThrowingRepo : IAuditLogRepository
{
///
public Task> GetTraceEventsAsync(Guid executionId, CancellationToken ct = default, IReadOnlyList? permittedSiteIds = null)
=> throw new NotSupportedException("GetTraceEventsAsync is not exercised by this test double.");
///
public Task InsertIfNotExistsAsync(AuditEvent evt, CancellationToken ct = default) =>
throw new InvalidOperationException("central insert failed");
///
public Task> QueryAsync(
AuditLogQueryFilter filter, AuditLogPaging paging, CancellationToken ct = default)
=> throw new NotSupportedException();
///
public Task SwitchOutPartitionAsync(
DateTime monthBoundary, TimeSpan? commandTimeout = null, CancellationToken ct = default)
=> throw new NotSupportedException();
///
public Task PurgeChannelOlderThanAsync(
string channel, DateTime threshold, int batchSize, TimeSpan? commandTimeout = null,
CancellationToken ct = default, int maxRows = 50_000)
=> throw new NotSupportedException();
///
public Task BackfillSourceNodeAsync(
string sentinel, DateTime before, int batchSize, CancellationToken ct = default)
=> throw new NotSupportedException();
///
public Task> GetPartitionBoundariesOlderThanAsync(
DateTime threshold, CancellationToken ct = default)
=> throw new NotSupportedException();
///
public Task GetKpiSnapshotAsync(
TimeSpan window, DateTime? nowUtc = null, CancellationToken ct = default)
=> throw new NotSupportedException();
///
public Task> GetExecutionTreeAsync(
Guid executionId, CancellationToken ct = default, IReadOnlyList? permittedSiteIds = null)
=> throw new NotSupportedException();
///
public Task> GetDistinctSourceNodesAsync(CancellationToken ct = default)
=> throw new NotSupportedException();
}
// ---------------------------------------------------------------------
// 6. Tick_OneSiteThrows_OtherSitesStillProcessed
// ---------------------------------------------------------------------
/// Verifies that a failure pulling from one site does not stop the other sites from being processed.
[Fact]
public void Tick_OneSiteThrows_OtherSitesStillProcessed()
{
var sites = new StaticEnumerator(
new SiteEntry("siteA", "http://siteA:8083"),
new SiteEntry("siteB", "http://siteB:8083"));
var bEvent = NewEvent("siteB");
var client = new ScriptedPullClient()
.ThrowFor("siteA", new InvalidOperationException("simulated transport failure"))
.Script("siteB",
new PullAuditEventsResponse(new[] { bEvent }, MoreAvailable: false));
var repo = new RecordingRepo();
var actor = CreateActor(sites, client, repo, FastTickOptions());
AwaitAssert(() =>
{
Assert.Contains(client.Calls, c => c.SiteId == "siteA");
Assert.Contains(repo.Inserted, e => e.EventId == bEvent.EventId);
},
duration: TimeSpan.FromSeconds(3),
interval: TimeSpan.FromMilliseconds(50));
StopReconciliationActor(actor);
}
// ---------------------------------------------------------------------
// 7. StalledDetection_TwoConsecutiveNonDrainingCycles_PublishesStalledTrue
// ---------------------------------------------------------------------
/// Verifies that two consecutive non-draining cycles publish a stalled state of true.
[Fact]
public void StalledDetection_TwoConsecutiveNonDrainingCycles_PublishesStalledTrue()
{
var sites = new StaticEnumerator(new SiteEntry("siteA", "http://siteA:8083"));
// Two scripted responses that each return events AND MoreAvailable=true
// — the second pull triggers the stalled transition.
var batch1 = Enumerable.Range(0, 3).Select(_ => NewEvent("siteA")).ToArray();
var batch2 = Enumerable.Range(0, 3).Select(_ => NewEvent("siteA")).ToArray();
var client = new ScriptedPullClient().Script("siteA",
new PullAuditEventsResponse(batch1, MoreAvailable: true),
new PullAuditEventsResponse(batch2, MoreAvailable: true));
var repo = new RecordingRepo();
var (probe, _) = SubscribeStalled();
var actor = CreateActor(sites, client, repo, FastTickOptions(stalledAfter: 2));
// Expect Stalled=true after the second non-draining tick. The probe
// waits with its own timeout (a few seconds gives the 0 s repeat
// interval ample slack).
var msg = probe.ExpectMsg(TimeSpan.FromSeconds(5));
StopReconciliationActor(actor);
Assert.Equal("siteA", msg.SiteId);
Assert.True(msg.Stalled);
}
// ---------------------------------------------------------------------
// 8. StalledDetection_DrainingCycle_PublishesStalledFalse
// ---------------------------------------------------------------------
/// Verifies that a draining cycle publishes a stalled state of false.
[Fact]
public void StalledDetection_DrainingCycle_PublishesStalledFalse()
{
var sites = new StaticEnumerator(new SiteEntry("siteA", "http://siteA:8083"));
// Two non-draining responses get the actor into Stalled=true, then a
// draining response (events but MoreAvailable=false) flips it back.
var batch1 = Enumerable.Range(0, 3).Select(_ => NewEvent("siteA")).ToArray();
var batch2 = Enumerable.Range(0, 3).Select(_ => NewEvent("siteA")).ToArray();
var batch3 = Enumerable.Range(0, 3).Select(_ => NewEvent("siteA")).ToArray();
var client = new ScriptedPullClient().Script("siteA",
new PullAuditEventsResponse(batch1, MoreAvailable: true),
new PullAuditEventsResponse(batch2, MoreAvailable: true),
new PullAuditEventsResponse(batch3, MoreAvailable: false));
var repo = new RecordingRepo();
var (probe, _) = SubscribeStalled();
var actor = CreateActor(sites, client, repo, FastTickOptions(stalledAfter: 2));
// First publication is the stalled=true transition; second is the
// back-to-draining flip. The actor publishes ONLY on transitions so we
// expect exactly these two messages in order.
var first = probe.ExpectMsg(TimeSpan.FromSeconds(5));
Assert.True(first.Stalled);
var second = probe.ExpectMsg(TimeSpan.FromSeconds(5));
StopReconciliationActor(actor);
Assert.False(second.Stalled);
Assert.Equal("siteA", second.SiteId);
}
// ---------------------------------------------------------------------
// 9. A permanently-abandoned row leaves a durable
// ReconciliationAbandoned audit record (not just a Critical log line),
// and the cursor advances past it.
// ---------------------------------------------------------------------
/// Verifies that a permanently abandoned row emits a ReconciliationAbandoned row and advances the cursor.
[Fact]
public void PermanentAbandonment_EmitsReconciliationAbandonedRow_AndAdvancesCursor()
{
var siteId = "siteAbandon";
var poisonTime = new DateTime(2026, 5, 20, 10, 5, 0, DateTimeKind.Utc);
var poison = NewEvent(siteId, poisonTime);
var sites = new StaticEnumerator(new SiteEntry(siteId, "http://x:8083"));
// Five pulls each returning the same poison row so its per-EventId attempt
// counter reaches MaxPermanentInsertAttempts (5); afterwards the scripted
// client returns empty and the actor settles.
var responses = Enumerable.Range(0, 5)
.Select(_ => new PullAuditEventsResponse(new[] { poison }, MoreAvailable: false))
.ToArray();
var client = new ScriptedPullClient().Script(siteId, responses);
var repo = new PoisonInsertRepo(
poison.EventId, new InvalidOperationException("poison row will never insert"));
var actor = CreateActor(sites, client, repo, FastTickOptions());
// After the 5th failed attempt the actor abandons the row and writes ONE
// synthetic ReconciliationAbandoned row (fresh EventId) whose details carry
// the abandoned id + the site.
AwaitAssert(
() =>
{
var synthetic = repo.Inserted.SingleOrDefault(
e => e.Action.EndsWith(".ReconciliationAbandoned", StringComparison.Ordinal));
Assert.NotNull(synthetic);
Assert.NotEqual(poison.EventId, synthetic!.EventId);
Assert.Contains(poison.EventId.ToString(), synthetic.DetailsJson);
Assert.Contains(siteId, synthetic.DetailsJson);
},
duration: TimeSpan.FromSeconds(8),
interval: TimeSpan.FromMilliseconds(100));
// The cursor advanced past MinValue to the abandoned row's timestamp — the
// poison row no longer pins the site's reconciliation forever. A later tick
// must pull with since = poisonTime (poll for it; the abandonment tick and
// the first advanced-cursor pull are separate ticks).
AwaitAssert(
() => Assert.Contains(client.Calls, c => c.SinceUtc == poisonTime),
duration: TimeSpan.FromSeconds(5),
interval: TimeSpan.FromMilliseconds(100));
StopReconciliationActor(actor);
}
// ---------------------------------------------------------------------
// 10. Deterministic synthetic ReconciliationAbandoned EventId
// ---------------------------------------------------------------------
///
/// Reproduces the rotating-poison defect end to end: a permanently-poisoned row sits
/// in a batch alongside a co-batch row that ALSO fails on every pull but
/// carries a fresh each time (modelling a
/// site that keeps producing new events with transient insert failures).
/// Per PullSiteAsync's hasUnresolvedFailure rule the rotating
/// row never individually reaches MaxPermanentInsertAttempts, so it
/// holds the reconciliation cursor at since forever — the poison row
/// keeps being re-pulled with a reset attempt counter (its
/// _failedInsertAttempts entry was cleared by its first abandonment)
/// and is permanently abandoned again roughly every
/// MaxPermanentInsertAttempts ticks. Ten ticks reproduces two full
/// abandon cycles for the SAME poisoned EventId. Pre-fix, each cycle mints a
/// fresh random synthetic EventId, so two DISTINCT rows land in the
/// repository. Post-fix, DeriveAbandonmentEventId derives the same id
/// both times and InsertIfNotExistsAsync's idempotency collapses the
/// second insert into a no-op, leaving exactly one row.
///
[Fact]
public void ReAbandonment_SameLostEventId_YieldsExactlyOneCentralRow()
{
var siteId = "siteReabandon";
var poisonTime = new DateTime(2026, 5, 20, 10, 5, 0, DateTimeKind.Utc);
var poison = NewEvent(siteId, poisonTime);
// 10 scripted responses: enough for exactly two 5-attempt abandon
// cycles of the poison row (the cursor never advances between them,
// so the SAME poison row + a freshly-guid'd rotating sibling are
// re-served every tick). Once exhausted, ScriptedPullClient returns
// empty responses, so no further cycles occur — the test's eventual
// row count is bounded and deterministic.
var responses = Enumerable.Range(0, 10)
.Select(i => new PullAuditEventsResponse(
new[] { poison, NewEvent(siteId, poisonTime.AddSeconds(i + 1)) },
MoreAvailable: false))
.ToArray();
var client = new ScriptedPullClient().Script(siteId, responses);
var sites = new StaticEnumerator(new SiteEntry(siteId, "http://x:8083"));
var repo = new AlwaysFailUnlessSyntheticRepo();
var actor = CreateActor(sites, client, repo, FastTickOptions());
// Wait for all 10 scripted pulls to have been consumed, i.e. both
// abandon cycles have run to completion.
AwaitAssert(
() => Assert.True(client.Calls.Count >= 10, $"expected >= 10 pull calls, got {client.Calls.Count}"),
duration: TimeSpan.FromSeconds(10),
interval: TimeSpan.FromMilliseconds(100));
// Give the actor's mailbox a chance to fully drain the 10th tick's
// processing (the call count above only proves PullAsync was invoked,
// not that the resulting insert(s) have landed yet) before reading
// the repository state.
AwaitAssert(
() => Assert.True(
repo.Inserted.Any(e => e.Action.EndsWith(".ReconciliationAbandoned", StringComparison.Ordinal)),
"expected at least one synthetic ReconciliationAbandoned row"),
duration: TimeSpan.FromSeconds(5),
interval: TimeSpan.FromMilliseconds(100));
StopReconciliationActor(actor);
var synthetic = repo.Inserted
.Where(e => e.Action.EndsWith(".ReconciliationAbandoned", StringComparison.Ordinal))
.ToList();
// The defect: pre-fix this is 2 (a fresh random EventId per abandon
// cycle defeats InsertIfNotExistsAsync's dedup). Post-fix it is 1.
Assert.Single(synthetic);
Assert.Contains(poison.EventId.ToString(), synthetic[0].DetailsJson);
}
///
/// Pure-function coverage of
/// directly: distinct lost rows must derive distinct synthetic ids (so two
/// different permanently-abandoned rows don't collide into one central row).
///
[Fact]
public void DeriveAbandonmentEventId_DifferentLostRows_YieldDistinctSyntheticIds()
{
var lostA = Guid.NewGuid();
var lostB = Guid.NewGuid();
var derivedA = SiteAuditReconciliationActor.DeriveAbandonmentEventId(lostA);
var derivedB = SiteAuditReconciliationActor.DeriveAbandonmentEventId(lostB);
Assert.NotEqual(derivedA, derivedB);
Assert.NotEqual(lostA, derivedA);
Assert.NotEqual(lostB, derivedB);
}
///
/// Pure-function coverage proving the derivation depends ONLY on the lost
/// row's VALUE — not on object identity, not
/// on any clock/random/process state. Two independently-parsed
/// values that are equal by value must derive the identical synthetic id, and
/// the exact value is pinned as a regression guard: since the function has no
/// external input besides , this literal must
/// reproduce identically in any process — this run, CI, or a reconciliation
/// singleton after a failover restart — which is the "stable across process
/// restarts" contract the fix depends on for idempotent re-abandonment.
///
[Fact]
public void DeriveAbandonmentEventId_IsPureAndStableAcrossIndependentInvocations()
{
var lostEventId = Guid.Parse("6b2f1e3a-8c9d-4a2b-9e3f-1a2b3c4d5e6f");
var roundTripped = Guid.Parse(lostEventId.ToString("D"));
var first = SiteAuditReconciliationActor.DeriveAbandonmentEventId(lostEventId);
var second = SiteAuditReconciliationActor.DeriveAbandonmentEventId(roundTripped);
Assert.Equal(first, second);
Assert.Equal(Guid.Parse("36baac53-f4c6-835e-9a77-d0b350a4e0d9"), first);
}
}