Joseph Doherty
20098c6108
Reconciliation cursor becomes composite (UpdatedAtUtc, TrackedOperationId); IPullSiteCallsClient/GrpcPullSiteCallsClient forward the Task-15 after_id keyset (additive param, null preserves the legacy inclusive >= contract). A burst sharing one exact UpdatedAtUtc now drains via the id tiebreak instead of pinning forever. A legacy site that ignores after_id is latched + published as SiteCallReconciliationPinnedChanged on the EventStream (transition-only), replacing the prior silent log line.
586 lines
27 KiB
C#
586 lines
27 KiB
C#
using Akka.Actor;
|
||
using Akka.TestKit.Xunit2;
|
||
using Microsoft.Extensions.Logging.Abstractions;
|
||
using ZB.MOM.WW.ScadaBridge.AuditLog.Central;
|
||
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Audit;
|
||
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
|
||
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Integration;
|
||
using ZB.MOM.WW.ScadaBridge.Commons.Types;
|
||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
|
||
|
||
namespace ZB.MOM.WW.ScadaBridge.SiteCallAudit.Tests;
|
||
|
||
/// <summary>
|
||
/// Reconciliation-tick tests for <see cref="SiteCallAuditActor"/> (#22, Piece A).
|
||
/// These exercise the periodic per-site self-heal pull entirely in-memory —
|
||
/// fake <see cref="IPullSiteCallsClient"/> + <see cref="ISiteEnumerator"/> + a
|
||
/// recording <see cref="ISiteCallAuditRepository"/> — so they run in
|
||
/// milliseconds and do NOT depend on a live MSSQL fixture (unlike the
|
||
/// MSSQL-backed <see cref="SiteCallAuditActorTests"/>). The actor is built via
|
||
/// the internal test ctor that injects all three collaborators; the
|
||
/// repo-only test ctor used by the MSSQL tests passes no client/enumerator, so
|
||
/// the reconciliation tick is gated off there (see
|
||
/// <see cref="TestCtor_RepositoryOnly_DoesNotStartReconciliationTick"/>).
|
||
/// </summary>
|
||
public class SiteCallAuditReconciliationTests : TestKit
|
||
{
|
||
private static SiteCall NewRow(
|
||
TrackedOperationId id,
|
||
string sourceSite,
|
||
string status = "Submitted",
|
||
DateTime? updatedAtUtc = null)
|
||
{
|
||
var now = updatedAtUtc ?? DateTime.UtcNow;
|
||
return new SiteCall
|
||
{
|
||
TrackedOperationId = id,
|
||
Channel = "ApiOutbound",
|
||
Target = "ERP.GetOrder",
|
||
SourceSite = sourceSite,
|
||
SourceNode = null,
|
||
Status = status,
|
||
RetryCount = 0,
|
||
LastError = null,
|
||
HttpStatus = null,
|
||
CreatedAtUtc = now,
|
||
UpdatedAtUtc = now,
|
||
TerminalAtUtc = null,
|
||
IngestedAtUtc = now,
|
||
};
|
||
}
|
||
|
||
private static SiteCallAuditOptions FastTickOptions(int batchSize = 500) => new()
|
||
{
|
||
// 100 ms tick keeps each test under a second; AwaitAssert covers
|
||
// scheduler jitter so the tick has up to a few seconds to fire.
|
||
ReconciliationInterval = TimeSpan.FromMinutes(5),
|
||
ReconciliationIntervalOverride = TimeSpan.FromMilliseconds(100),
|
||
ReconciliationBatchSize = batchSize,
|
||
};
|
||
|
||
/// <summary>In-memory enumerator returning a static list of sites.</summary>
|
||
private sealed class StaticEnumerator : ISiteEnumerator
|
||
{
|
||
private readonly IReadOnlyList<SiteEntry> _sites;
|
||
public StaticEnumerator(params SiteEntry[] sites) => _sites = sites;
|
||
public Task<IReadOnlyList<SiteEntry>> EnumerateAsync(CancellationToken ct = default) =>
|
||
Task.FromResult(_sites);
|
||
}
|
||
|
||
/// <summary>
|
||
/// Scripted pull client — returns the next queued response for the site on
|
||
/// each call (looping the last entry once exhausted) and records every
|
||
/// invocation so tests can assert call counts + the <c>since</c> cursor.
|
||
/// </summary>
|
||
private sealed class ScriptedPullClient : IPullSiteCallsClient
|
||
{
|
||
public List<(string SiteId, DateTime SinceUtc, string? AfterId, int BatchSize)> Calls { get; } = new();
|
||
private readonly Dictionary<string, Queue<PullSiteCallsResponse>> _scripted = new();
|
||
private readonly Dictionary<string, Exception> _throwOnSite = new();
|
||
|
||
public ScriptedPullClient Script(string siteId, params PullSiteCallsResponse[] responses)
|
||
{
|
||
_scripted[siteId] = new Queue<PullSiteCallsResponse>(responses);
|
||
return this;
|
||
}
|
||
|
||
public ScriptedPullClient ThrowFor(string siteId, Exception ex)
|
||
{
|
||
_throwOnSite[siteId] = ex;
|
||
return this;
|
||
}
|
||
|
||
public Task<PullSiteCallsResponse> 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 PullSiteCallsResponse(Array.Empty<SiteCall>(), MoreAvailable: false));
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// Keyset-honouring in-memory pull client (Task 16). Holds one static set of
|
||
/// rows and answers each pull as a real site would: returns the next
|
||
/// <c>batchSize</c> rows strictly greater than the composite
|
||
/// <c>(sinceUtc, afterId)</c> cursor under the deterministic
|
||
/// <c>(UpdatedAtUtc, TrackedOperationId ordinal)</c> ordering, with
|
||
/// <c>MoreAvailable=true</c> when unseen rows remain. This is the modern site
|
||
/// that DOES honour <c>after_id</c>, so a burst sharing one exact
|
||
/// <c>UpdatedAtUtc</c> drains fully via the id tiebreak — no single-timestamp
|
||
/// pin.
|
||
/// </summary>
|
||
private sealed class KeysetHonoringPullClient : IPullSiteCallsClient
|
||
{
|
||
private readonly IReadOnlyList<SiteCall> _rows;
|
||
public int CallCount { get; private set; }
|
||
|
||
public KeysetHonoringPullClient(IEnumerable<SiteCall> rows) =>
|
||
_rows = rows
|
||
.OrderBy(r => r.UpdatedAtUtc)
|
||
.ThenBy(r => r.TrackedOperationId.ToString(), StringComparer.Ordinal)
|
||
.ToList();
|
||
|
||
public Task<PullSiteCallsResponse> PullAsync(
|
||
string siteId, DateTime sinceUtc, string? afterId, int batchSize, CancellationToken ct)
|
||
{
|
||
CallCount++;
|
||
var greater = _rows
|
||
.Where(r =>
|
||
r.UpdatedAtUtc > sinceUtc ||
|
||
(r.UpdatedAtUtc == sinceUtc &&
|
||
string.CompareOrdinal(r.TrackedOperationId.ToString(), afterId) > 0))
|
||
.ToList();
|
||
var page = greater.Take(batchSize).ToList();
|
||
var more = greater.Count > page.Count;
|
||
return Task.FromResult(
|
||
new PullSiteCallsResponse(page, MoreAvailable: more));
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// Legacy pull client (Task 16) that IGNORES <c>afterId</c> and always
|
||
/// returns the same saturated page (<c>MoreAvailable=true</c>) — models a
|
||
/// pre-keyset site the central actor cannot drain past. The composite cursor
|
||
/// cannot advance, so the actor must latch and publish
|
||
/// <see cref="SiteCallReconciliationPinnedChanged"/> instead of spinning.
|
||
/// </summary>
|
||
private sealed class LegacyIgnoresAfterIdPullClient : IPullSiteCallsClient
|
||
{
|
||
private readonly IReadOnlyList<SiteCall> _rows;
|
||
public int CallCount { get; private set; }
|
||
|
||
public LegacyIgnoresAfterIdPullClient(IReadOnlyList<SiteCall> rows) => _rows = rows;
|
||
|
||
public Task<PullSiteCallsResponse> PullAsync(
|
||
string siteId, DateTime sinceUtc, string? afterId, int batchSize, CancellationToken ct)
|
||
{
|
||
CallCount++;
|
||
return Task.FromResult(new PullSiteCallsResponse(_rows, MoreAvailable: true));
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// Pull client that ALWAYS returns the same saturated response
|
||
/// (<c>MoreAvailable=true</c>) regardless of the <c>since</c> cursor —
|
||
/// simulates the SiteCallAudit-009 single-timestamp no-progress pin: a backlog
|
||
/// larger than the batch size all sharing one exact <c>UpdatedAtUtc</c>, so
|
||
/// the inclusive max-timestamp cursor never advances. Records every call so
|
||
/// the test can assert the within-tick drain is BOUNDED (the actor must not
|
||
/// spin the dispatcher forever on this pathological input).
|
||
/// </summary>
|
||
private sealed class SaturatedPinPullClient : IPullSiteCallsClient
|
||
{
|
||
private readonly IReadOnlyList<SiteCall> _rows;
|
||
public int CallCount { get; private set; }
|
||
|
||
public SaturatedPinPullClient(IReadOnlyList<SiteCall> rows) => _rows = rows;
|
||
|
||
public Task<PullSiteCallsResponse> PullAsync(
|
||
string siteId, DateTime sinceUtc, string? afterId, int batchSize, CancellationToken ct)
|
||
{
|
||
CallCount++;
|
||
return Task.FromResult(new PullSiteCallsResponse(_rows, MoreAvailable: true));
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// Recording repository that captures every <see cref="UpsertAsync"/> call
|
||
/// (keyed by id, last-write-wins on the captured row). The reconciliation
|
||
/// tick only ever calls <see cref="UpsertAsync"/>; the read/KPI members are
|
||
/// inert stubs.
|
||
/// </summary>
|
||
private sealed class RecordingRepo : ISiteCallAuditRepository
|
||
{
|
||
public Dictionary<TrackedOperationId, SiteCall> Upserted { get; } = new();
|
||
public int UpsertCallCount { get; private set; }
|
||
|
||
public Task UpsertAsync(SiteCall siteCall, CancellationToken ct = default)
|
||
{
|
||
UpsertCallCount++;
|
||
Upserted[siteCall.TrackedOperationId] = siteCall;
|
||
return Task.CompletedTask;
|
||
}
|
||
|
||
public Task<SiteCall?> GetAsync(TrackedOperationId id, CancellationToken ct = default) =>
|
||
Task.FromResult(Upserted.TryGetValue(id, out var row) ? row : null);
|
||
|
||
public Task<IReadOnlyList<SiteCall>> QueryAsync(
|
||
SiteCallQueryFilter filter, SiteCallPaging paging, CancellationToken ct = default) =>
|
||
Task.FromResult<IReadOnlyList<SiteCall>>(Array.Empty<SiteCall>());
|
||
|
||
public Task<int> PurgeTerminalAsync(DateTime olderThanUtc, CancellationToken ct = default) =>
|
||
Task.FromResult(0);
|
||
|
||
public Task<SiteCallKpiSnapshot> ComputeKpisAsync(
|
||
DateTime stuckCutoff, DateTime intervalSince, CancellationToken ct = default) =>
|
||
Task.FromResult(new SiteCallKpiSnapshot(0, 0, 0, 0, null, 0));
|
||
|
||
public Task<IReadOnlyList<SiteCallSiteKpiSnapshot>> ComputePerSiteKpisAsync(
|
||
DateTime stuckCutoff, DateTime intervalSince, CancellationToken ct = default) =>
|
||
Task.FromResult<IReadOnlyList<SiteCallSiteKpiSnapshot>>(Array.Empty<SiteCallSiteKpiSnapshot>());
|
||
|
||
public Task<IReadOnlyList<SiteCallNodeKpiSnapshot>> ComputePerNodeKpisAsync(
|
||
DateTime stuckCutoff, DateTime intervalSince, CancellationToken ct = default) =>
|
||
Task.FromResult<IReadOnlyList<SiteCallNodeKpiSnapshot>>(Array.Empty<SiteCallNodeKpiSnapshot>());
|
||
}
|
||
|
||
private IActorRef CreateActor(
|
||
ISiteEnumerator sites,
|
||
IPullSiteCallsClient client,
|
||
ISiteCallAuditRepository repo,
|
||
SiteCallAuditOptions options) =>
|
||
Sys.ActorOf(Props.Create(() => new SiteCallAuditActor(
|
||
repo,
|
||
sites,
|
||
client,
|
||
NullLogger<SiteCallAuditActor>.Instance,
|
||
options)));
|
||
|
||
// ---------------------------------------------------------------------
|
||
// 1. AbsentRow_PulledFromSite_IsUpserted
|
||
// ---------------------------------------------------------------------
|
||
|
||
[Fact]
|
||
public void ReconciliationTick_AbsentRow_IsUpsertedFromSitePull()
|
||
{
|
||
var siteId = "siteA";
|
||
var id = TrackedOperationId.New();
|
||
var row = NewRow(id, sourceSite: siteId, status: "Parked");
|
||
|
||
var sites = new StaticEnumerator(new SiteEntry(siteId, "http://siteA:8083"));
|
||
var client = new ScriptedPullClient().Script(siteId,
|
||
new PullSiteCallsResponse(new[] { row }, MoreAvailable: false));
|
||
var repo = new RecordingRepo();
|
||
|
||
CreateActor(sites, client, repo, FastTickOptions());
|
||
|
||
AwaitAssert(
|
||
() =>
|
||
{
|
||
Assert.True(repo.Upserted.ContainsKey(id),
|
||
"reconciliation tick should upsert the row present at the site but absent centrally");
|
||
Assert.Equal("Parked", repo.Upserted[id].Status);
|
||
Assert.Equal(siteId, repo.Upserted[id].SourceSite);
|
||
},
|
||
duration: TimeSpan.FromSeconds(3),
|
||
interval: TimeSpan.FromMilliseconds(50));
|
||
}
|
||
|
||
// ---------------------------------------------------------------------
|
||
// 2. Cursor_Advances_ToMaxUpdatedAtUtc_NoRePullOfOldRows
|
||
// ---------------------------------------------------------------------
|
||
|
||
[Fact]
|
||
public void ReconciliationTick_SecondTick_AdvancesCursorPastAlreadyPulledRows()
|
||
{
|
||
var siteId = "siteA";
|
||
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 r1 = NewRow(TrackedOperationId.New(), siteId, updatedAtUtc: t1);
|
||
var r2 = NewRow(TrackedOperationId.New(), siteId, updatedAtUtc: t2);
|
||
var r3 = NewRow(TrackedOperationId.New(), siteId, updatedAtUtc: t3);
|
||
|
||
var sites = new StaticEnumerator(new SiteEntry(siteId, "http://siteA:8083"));
|
||
// First pull returns three rows (max UpdatedAtUtc = t3); subsequent
|
||
// pulls return empty. The second pull's `since` must be t3, proving the
|
||
// cursor advanced and old rows are not re-pulled from the start.
|
||
var client = new ScriptedPullClient().Script(siteId,
|
||
new PullSiteCallsResponse(new[] { r1, r2, r3 }, MoreAvailable: false));
|
||
var repo = new RecordingRepo();
|
||
|
||
CreateActor(sites, client, repo, FastTickOptions());
|
||
|
||
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));
|
||
|
||
Assert.Equal(DateTime.MinValue, client.Calls[0].SinceUtc);
|
||
Assert.Equal(t3, client.Calls[1].SinceUtc);
|
||
// The batch size flows through from options.
|
||
Assert.Equal(500, client.Calls[0].BatchSize);
|
||
}
|
||
|
||
// ---------------------------------------------------------------------
|
||
// 3. OneSiteThrows_OtherSitesStillProcessed (failure isolation)
|
||
// ---------------------------------------------------------------------
|
||
|
||
[Fact]
|
||
public void ReconciliationTick_OneSiteThrows_OtherSitesStillReconciled()
|
||
{
|
||
var siteB = "siteB";
|
||
var bId = TrackedOperationId.New();
|
||
var bRow = NewRow(bId, sourceSite: siteB, status: "Delivered");
|
||
|
||
var sites = new StaticEnumerator(
|
||
new SiteEntry("siteA", "http://siteA:8083"),
|
||
new SiteEntry(siteB, "http://siteB:8083"));
|
||
var client = new ScriptedPullClient()
|
||
.ThrowFor("siteA", new InvalidOperationException("simulated transport failure"))
|
||
.Script(siteB, new PullSiteCallsResponse(new[] { bRow }, MoreAvailable: false));
|
||
var repo = new RecordingRepo();
|
||
|
||
CreateActor(sites, client, repo, FastTickOptions());
|
||
|
||
AwaitAssert(
|
||
() =>
|
||
{
|
||
// siteA was attempted (and threw) yet siteB's row still landed —
|
||
// one offline site must not sink the rest of the tick.
|
||
Assert.Contains(client.Calls, c => c.SiteId == "siteA");
|
||
Assert.True(repo.Upserted.ContainsKey(bId),
|
||
"siteB must be reconciled even though siteA threw");
|
||
},
|
||
duration: TimeSpan.FromSeconds(3),
|
||
interval: TimeSpan.FromMilliseconds(50));
|
||
}
|
||
|
||
// ---------------------------------------------------------------------
|
||
// 4. RepoOnly test ctor does NOT start the reconciliation tick
|
||
// ---------------------------------------------------------------------
|
||
|
||
[Fact]
|
||
public void TestCtor_RepositoryOnly_DoesNotStartReconciliationTick()
|
||
{
|
||
// The repo-only test ctor (used by the MSSQL-backed actor tests) injects
|
||
// no client/enumerator, so the tick must be gated OFF — otherwise those
|
||
// tests would fire phantom pulls. Build the actor via that ctor and
|
||
// confirm no pull ever happens. We can't observe a non-event directly,
|
||
// so we share a ScriptedPullClient with an isolated actor that DOES run
|
||
// the tick to bound the wait, then assert the repo-only actor's client
|
||
// (a separate instance) recorded nothing.
|
||
var repo = new RecordingRepo();
|
||
Sys.ActorOf(Props.Create(() => new SiteCallAuditActor(
|
||
repo,
|
||
NullLogger<SiteCallAuditActor>.Instance,
|
||
FastTickOptions())));
|
||
|
||
// Run a parallel actor with the full reconciliation ctor and a fast
|
||
// tick; once IT has pulled we know enough wall-clock elapsed that the
|
||
// repo-only actor would have ticked too, had it been wired.
|
||
var liveClient = new ScriptedPullClient();
|
||
var liveRepo = new RecordingRepo();
|
||
CreateActor(
|
||
new StaticEnumerator(new SiteEntry("siteX", "http://siteX:8083")),
|
||
liveClient,
|
||
liveRepo,
|
||
FastTickOptions());
|
||
|
||
AwaitAssert(
|
||
() => Assert.True(liveClient.Calls.Count >= 1),
|
||
duration: TimeSpan.FromSeconds(3),
|
||
interval: TimeSpan.FromMilliseconds(50));
|
||
|
||
// The repo-only actor never reconciles: it has no client to pull with,
|
||
// so it upserts nothing on its own.
|
||
Assert.Equal(0, repo.UpsertCallCount);
|
||
}
|
||
|
||
// ---------------------------------------------------------------------
|
||
// 5. SiteCallAudit-009: MoreAvailable drives a within-tick continuation
|
||
// drain — a multi-page backlog whose timestamps advance is fully drained
|
||
// in ONE tick rather than one page per tick.
|
||
// ---------------------------------------------------------------------
|
||
|
||
[Fact]
|
||
public void ReconciliationTick_MoreAvailable_DrainsMultiplePagesWithinOneTick()
|
||
{
|
||
var siteId = "siteA";
|
||
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 p1a = NewRow(TrackedOperationId.New(), siteId, updatedAtUtc: t1);
|
||
var p1b = NewRow(TrackedOperationId.New(), siteId, updatedAtUtc: t2);
|
||
var p2 = NewRow(TrackedOperationId.New(), siteId, updatedAtUtc: t3);
|
||
|
||
var sites = new StaticEnumerator(new SiteEntry(siteId, "http://siteA:8083"));
|
||
// Page 1 saturates (MoreAvailable: true) → the actor continues pulling
|
||
// within the SAME tick; page 2 is the final page (MoreAvailable: false).
|
||
// The continuation pull's `since` must be t2 (page-1 max), proving the
|
||
// cursor advanced page-to-page inside one tick rather than across ticks.
|
||
var client = new ScriptedPullClient().Script(siteId,
|
||
new PullSiteCallsResponse(new[] { p1a, p1b }, MoreAvailable: true),
|
||
new PullSiteCallsResponse(new[] { p2 }, MoreAvailable: false));
|
||
var repo = new RecordingRepo();
|
||
|
||
// Slow tick so the multi-page drain CANNOT be the natural tick cadence —
|
||
// it must be the within-tick continuation loop. Long enough that only the
|
||
// first tick fires in the assert window.
|
||
var options = new SiteCallAuditOptions
|
||
{
|
||
ReconciliationIntervalOverride = TimeSpan.FromSeconds(2),
|
||
ReconciliationBatchSize = 2,
|
||
};
|
||
|
||
CreateActor(sites, client, repo, options);
|
||
|
||
AwaitAssert(
|
||
() =>
|
||
{
|
||
// All three rows reconciled — including the page-2 row that only a
|
||
// within-tick continuation pull could have fetched.
|
||
Assert.True(repo.Upserted.ContainsKey(p1a.TrackedOperationId));
|
||
Assert.True(repo.Upserted.ContainsKey(p1b.TrackedOperationId));
|
||
Assert.True(repo.Upserted.ContainsKey(p2.TrackedOperationId),
|
||
"the page-2 row must be reconciled within the same tick via the MoreAvailable continuation drain");
|
||
},
|
||
duration: TimeSpan.FromSeconds(3),
|
||
interval: TimeSpan.FromMilliseconds(50));
|
||
|
||
// Exactly two pulls happened (page 1 + the continuation page 2) and the
|
||
// second pull's `since` cursor advanced to the page-1 max (t2).
|
||
Assert.True(client.Calls.Count >= 2, $"expected >= 2 pulls within the tick, got {client.Calls.Count}");
|
||
Assert.Equal(DateTime.MinValue, client.Calls[0].SinceUtc);
|
||
Assert.Equal(t2, client.Calls[1].SinceUtc);
|
||
}
|
||
|
||
// ---------------------------------------------------------------------
|
||
// 6. SiteCallAudit-009: single-timestamp saturation pin does NOT spin —
|
||
// a saturated batch whose max UpdatedAtUtc never advances past `since`
|
||
// breaks the within-tick drain after one page (no unbounded re-pull),
|
||
// and still upserts the rows it saw.
|
||
// ---------------------------------------------------------------------
|
||
|
||
[Fact]
|
||
public void ReconciliationTick_SingleTimestampSaturation_DoesNotSpin_MakesNoProgressGracefully()
|
||
{
|
||
var siteId = "siteA";
|
||
// A burst sharing ONE exact UpdatedAtUtc that saturates the batch — the
|
||
// inclusive max-timestamp cursor cannot advance, so an unbounded
|
||
// continuation loop would re-pull this identical window forever.
|
||
var ts = new DateTime(2026, 5, 20, 10, 0, 0, DateTimeKind.Utc);
|
||
var r1 = NewRow(TrackedOperationId.New(), siteId, updatedAtUtc: ts);
|
||
var r2 = NewRow(TrackedOperationId.New(), siteId, updatedAtUtc: ts);
|
||
|
||
var sites = new StaticEnumerator(new SiteEntry(siteId, "http://siteA:8083"));
|
||
var client = new SaturatedPinPullClient(new[] { r1, r2 });
|
||
var repo = new RecordingRepo();
|
||
|
||
// Long interval so AT MOST one tick fires in the assert window — lets us
|
||
// bound the WITHIN-tick pull count. A no-progress pin must break after a
|
||
// single page, NOT loop up to MaxReconciliationPagesPerTick (50).
|
||
var options = new SiteCallAuditOptions
|
||
{
|
||
ReconciliationIntervalOverride = TimeSpan.FromSeconds(2),
|
||
ReconciliationBatchSize = 2,
|
||
};
|
||
|
||
CreateActor(sites, client, repo, options);
|
||
|
||
AwaitAssert(
|
||
() => Assert.True(client.CallCount >= 1, "the first reconciliation tick should have pulled"),
|
||
duration: TimeSpan.FromSeconds(3),
|
||
interval: TimeSpan.FromMilliseconds(50));
|
||
|
||
// The rows it saw were still upserted (idempotent mirror refresh).
|
||
Assert.True(repo.Upserted.ContainsKey(r1.TrackedOperationId));
|
||
Assert.True(repo.Upserted.ContainsKey(r2.TrackedOperationId));
|
||
|
||
// Critical SiteCallAudit-009 invariant: the within-tick drain BROKE on the
|
||
// no-progress pin rather than looping to the 50-page ceiling. With a 2s
|
||
// tick interval, only the first tick has fired in the window, so the pull
|
||
// count reflects ONE tick's within-loop behaviour. A correct break yields
|
||
// 1 pull for that tick; we allow a small margin for a possible second tick
|
||
// edge, but it must be far below the 50-page within-tick ceiling.
|
||
Assert.True(client.CallCount < 10,
|
||
$"a single-timestamp saturation pin must break the within-tick drain, not spin to the "
|
||
+ $"page ceiling; got {client.CallCount} pulls (an unbounded within-tick loop would be 50+)");
|
||
}
|
||
|
||
// ---------------------------------------------------------------------
|
||
// 7. Task 16: a keyset-honouring site drains a single-timestamp burst
|
||
// LARGER than the batch size in ONE tick — the composite (UpdatedAtUtc,
|
||
// TrackedOperationId) cursor advances by id even when every row shares
|
||
// one exact timestamp, so the tail no longer pins forever.
|
||
// ---------------------------------------------------------------------
|
||
|
||
[Fact]
|
||
public void ReconciliationTick_KeysetCursor_DrainsSingleTimestampBurstBeyondBatchSize()
|
||
{
|
||
var siteId = "siteA";
|
||
// 2 × batchSize rows all sharing ONE exact UpdatedAtUtc: the pre-keyset
|
||
// inclusive-timestamp cursor could never advance past this window.
|
||
var ts = new DateTime(2026, 5, 20, 10, 0, 0, DateTimeKind.Utc);
|
||
var rows = Enumerable.Range(0, 4)
|
||
.Select(_ => NewRow(TrackedOperationId.New(), siteId, updatedAtUtc: ts))
|
||
.ToList();
|
||
|
||
var sites = new StaticEnumerator(new SiteEntry(siteId, "http://siteA:8083"));
|
||
var client = new KeysetHonoringPullClient(rows);
|
||
var repo = new RecordingRepo();
|
||
|
||
// batchSize = 2 → the 4-row burst needs the id tiebreak to page through.
|
||
// Slow tick so the full drain must be the within-tick continuation loop,
|
||
// not multiple ticks.
|
||
var options = new SiteCallAuditOptions
|
||
{
|
||
ReconciliationIntervalOverride = TimeSpan.FromSeconds(2),
|
||
ReconciliationBatchSize = 2,
|
||
};
|
||
|
||
CreateActor(sites, client, repo, options);
|
||
|
||
AwaitAssert(
|
||
() =>
|
||
{
|
||
foreach (var r in rows)
|
||
{
|
||
Assert.True(repo.Upserted.ContainsKey(r.TrackedOperationId),
|
||
"every row in the single-timestamp burst must reconcile via the "
|
||
+ "composite keyset cursor within one tick");
|
||
}
|
||
},
|
||
duration: TimeSpan.FromSeconds(3),
|
||
interval: TimeSpan.FromMilliseconds(50));
|
||
|
||
// Two pages (2 + 2) drained the burst; well under the 50-page ceiling.
|
||
Assert.True(client.CallCount is >= 2 and < 10,
|
||
$"expected the burst to drain in a couple of keyset pages, got {client.CallCount}");
|
||
}
|
||
|
||
// ---------------------------------------------------------------------
|
||
// 8. Task 16: a LEGACY site that ignores after_id can never be drained, so
|
||
// the actor latches and publishes SiteCallReconciliationPinnedChanged
|
||
// instead of silently logging (the dead-end is now health-observable).
|
||
// ---------------------------------------------------------------------
|
||
|
||
[Fact]
|
||
public void ReconciliationTick_LegacySiteIgnoresAfterId_PublishesPinnedChanged()
|
||
{
|
||
var siteId = "siteLegacy";
|
||
var ts = new DateTime(2026, 5, 20, 10, 0, 0, DateTimeKind.Utc);
|
||
var r1 = NewRow(TrackedOperationId.New(), siteId, updatedAtUtc: ts);
|
||
var r2 = NewRow(TrackedOperationId.New(), siteId, updatedAtUtc: ts);
|
||
|
||
var sites = new StaticEnumerator(new SiteEntry(siteId, "http://siteLegacy:8083"));
|
||
var client = new LegacyIgnoresAfterIdPullClient(new[] { r1, r2 });
|
||
var repo = new RecordingRepo();
|
||
|
||
// Subscribe BEFORE the actor starts ticking so the transition isn't missed.
|
||
Sys.EventStream.Subscribe(TestActor, typeof(SiteCallReconciliationPinnedChanged));
|
||
|
||
var options = new SiteCallAuditOptions
|
||
{
|
||
ReconciliationIntervalOverride = TimeSpan.FromMilliseconds(100),
|
||
ReconciliationBatchSize = 2,
|
||
};
|
||
|
||
CreateActor(sites, client, repo, options);
|
||
|
||
var evt = ExpectMsg<SiteCallReconciliationPinnedChanged>(TimeSpan.FromSeconds(5));
|
||
Assert.Equal(siteId, evt.SiteId);
|
||
Assert.True(evt.Pinned, "a legacy site that ignores after_id must publish Pinned=true");
|
||
}
|
||
}
|