From 20098c6108717aa1327c9a67c61b88e6401258ab Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 9 Jul 2026 08:44:07 -0400 Subject: [PATCH] fix(site-call-audit): composite keyset cursor eliminates the single-timestamp reconciliation pin; pinned state now a published event 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. --- docs/requirements/Component-AuditLog.md | 8 + docs/requirements/Component-SiteCallAudit.md | 15 ++ .../Central/GrpcPullSiteCallsClient.cs | 5 + .../Central/IPullSiteCallsClient.cs | 10 ++ .../SiteCallAuditActor.cs | 163 ++++++++++++------ .../SiteCallReconciliationPinnedChanged.cs | 19 ++ .../Central/GrpcPullSiteCallsClientTests.cs | 14 +- .../SiteCallAuditPurgeTests.cs | 2 +- .../SiteCallAuditReconciliationTests.cs | 154 ++++++++++++++++- 9 files changed, 324 insertions(+), 66 deletions(-) create mode 100644 src/ZB.MOM.WW.ScadaBridge.SiteCallAudit/SiteCallReconciliationPinnedChanged.cs diff --git a/docs/requirements/Component-AuditLog.md b/docs/requirements/Component-AuditLog.md index 88267907..c1575305 100644 --- a/docs/requirements/Component-AuditLog.md +++ b/docs/requirements/Component-AuditLog.md @@ -278,6 +278,14 @@ non-draining (e.g., telemetry actor wedged), central issues a are flipped to `ForwardState = 'Reconciled'` site-side. Same self-healing pattern as Site Call Audit's reconciliation of `SiteCalls`. +> **Cursor keyset (tracked follow-up).** The `PullAuditEvents` cursor is still a +> single `sinceUtc` timestamp. Site Call Audit's pull now uses a composite +> `(UpdatedAtUtc, TrackedOperationId)` keyset to avoid a single-timestamp pin +> (see Component-SiteCallAudit.md → Reconciliation). The same keyset should be +> applied here, but it is lower urgency because the audit cursor already re-pulls +> idempotently on `EventId` — a saturated single-timestamp window re-inserts +> harmless no-ops rather than losing rows. + ### Central direct-write (central-originated events) Events originating at central never touch site SQLite. Inbound API writes one diff --git a/docs/requirements/Component-SiteCallAudit.md b/docs/requirements/Component-SiteCallAudit.md index 63008751..51af77ff 100644 --- a/docs/requirements/Component-SiteCallAudit.md +++ b/docs/requirements/Component-SiteCallAudit.md @@ -98,6 +98,21 @@ reconnect — pulls "all tracking rows changed since cursor X" from each site. Gaps left by lost telemetry self-heal. Central converges to the site; the site never depends on central. +The per-site cursor is a **composite `(UpdatedAtUtc, TrackedOperationId)` +keyset**, not a single timestamp. Each pull asks for rows strictly greater than +the cursor pair and advances it to the maximum row seen; a burst of more rows +than one batch all sharing one exact `UpdatedAtUtc` therefore drains via the +`TrackedOperationId` tiebreak instead of pinning the timestamp forever. The +`after_id` keyset field is additive on the pull contract — a first pull (or a +**legacy** site that predates it) sends no `after_id` and keeps the inclusive +`>=` timestamp behaviour. When such a legacy site keeps reporting +`MoreAvailable` yet the composite cursor cannot advance, the actor latches the +site as *pinned* and publishes `SiteCallReconciliationPinnedChanged(siteId, +Pinned)` on the EventStream (transition-only, mirroring +`SiteAuditTelemetryStalledChanged`) — the un-drainable tail is a +health-observable condition rather than a silent log line, and the latch clears +with `Pinned=false` once a later tick makes progress. + ## Retry / Discard Relay Parked cached calls live in the owning site's S&F buffer. Operator Retry/Discard diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcPullSiteCallsClient.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcPullSiteCallsClient.cs index 0a468d23..6d82ad24 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcPullSiteCallsClient.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcPullSiteCallsClient.cs @@ -80,6 +80,7 @@ public sealed class GrpcPullSiteCallsClient : IPullSiteCallsClient public async Task PullAsync( string siteId, DateTime sinceUtc, + string? afterId, int batchSize, CancellationToken ct) { @@ -100,6 +101,10 @@ public sealed class GrpcPullSiteCallsClient : IPullSiteCallsClient // EnsureUtc keeps Timestamp.FromDateTime happy (it requires UTC kind). SinceUtc = Timestamp.FromDateTime(EnsureUtc(sinceUtc)), BatchSize = batchSize, + // Composite-keyset tiebreak (Task 16). proto3 has no nullable string — + // an unset/empty AfterId is the site's signal to keep the legacy + // inclusive-timestamp contract (also what a first pull sends). + AfterId = afterId ?? string.Empty, }; ProtoPullResponse reply; diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/IPullSiteCallsClient.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/IPullSiteCallsClient.cs index 74dc9601..7a13d73e 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/IPullSiteCallsClient.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/IPullSiteCallsClient.cs @@ -46,12 +46,22 @@ public interface IPullSiteCallsClient /// /// The identifier of the site to pull cached-call operational rows from. /// Only rows with an UpdatedAtUtc at or after this cursor time are returned. + /// + /// The composite-keyset tiebreak cursor (Task 16). When non-null it is the + /// TrackedOperationId of the last row already consumed at + /// ; the site returns only rows strictly greater + /// than the composite (UpdatedAtUtc, TrackedOperationId) pair, so a + /// burst sharing one exact UpdatedAtUtc drains via the id tiebreak + /// instead of pinning the inclusive-timestamp cursor. Null on the first pull + /// (or against a legacy site) preserves the inclusive >= contract. + /// /// Maximum number of rows to return per call. /// Cancellation token. /// A task that resolves to the next reconciliation batch with a MoreAvailable flag. Task PullAsync( string siteId, DateTime sinceUtc, + string? afterId, int batchSize, CancellationToken ct); } diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteCallAudit/SiteCallAuditActor.cs b/src/ZB.MOM.WW.ScadaBridge.SiteCallAudit/SiteCallAuditActor.cs index 9d4a262c..1e64c856 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteCallAudit/SiteCallAuditActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteCallAudit/SiteCallAuditActor.cs @@ -131,17 +131,32 @@ public class SiteCallAuditActor : ReceiveActor private readonly bool _backgroundTimersEnabled; /// - /// Per-site reconciliation watermark — the highest - /// seen for that site on a previous - /// tick. The next tick asks for rows at or after this cursor; idempotent - /// monotonic swallows any - /// duplicate-with-same-timestamp rows. In-memory for the singleton's - /// lifetime — a failover / restart resets every cursor to - /// , which is conservative but correct - /// (the next tick re-pulls and idempotent upsert dedupes). Mirrors - /// SiteAuditReconciliationActor. + /// Per-site reconciliation watermark — the composite + /// (UpdatedAtUtc, TrackedOperationId) keyset of the highest row seen + /// for that site on a previous tick (Task 16). The next tick asks for rows + /// strictly greater than this pair, so a burst sharing one exact + /// drains via the id tiebreak rather than + /// pinning the timestamp forever; idempotent monotonic + /// still swallows any + /// boundary duplicate. In-memory for the singleton's lifetime — a + /// failover / restart resets every cursor to + /// (DateTime.MinValue, null), which is conservative but correct (the + /// next tick re-pulls from the start and idempotent upsert dedupes). Mirrors + /// SiteAuditReconciliationActor's cursor (now composite). /// - private readonly Dictionary _reconciliationCursors = new(); + private readonly Dictionary _reconciliationCursors = new(); + + /// + /// Per-site "pinned" latch (Task 16) — true once a site's composite + /// cursor stopped advancing while the site still reported + /// MoreAvailable=true (in practice only a legacy site that ignores the + /// after_id keyset field). Tracks the latch so only transitions + /// publish on the EventStream + /// (pinned on latch, unpinned when progress resumes), mirroring the sibling + /// SiteAuditReconciliationActor's stalled-state machine — no flood of + /// redundant notifications. + /// + private readonly Dictionary _reconciliationPinned = new(); private ICancelable? _reconciliationTimer; private ICancelable? _purgeTimer; @@ -581,52 +596,56 @@ public class SiteCallAuditActor : ReceiveActor /// not the source of truth, so a slow site simply lags rather than corrupts. /// /// - /// Inclusive cursor boundary. The cursor is advanced to the maximum - /// seen, and the pull asks for rows at or - /// after it (since is >=, not >). The row whose - /// timestamp equals the cursor is therefore re-pulled on the next tick and - /// deduplicated by the idempotent monotonic upsert — the same inclusive-boundary - /// contract as SiteAuditReconciliationActor's cursor. + /// Composite keyset cursor (Task 16). The cursor is the + /// (UpdatedAtUtc, TrackedOperationId) pair of the maximum row seen, and + /// the pull asks for rows strictly greater than it. The boundary row is NOT + /// re-pulled (unlike the prior inclusive-timestamp cursor), and — critically — + /// a burst of more than + /// rows sharing one exact now drains via the + /// id tiebreak instead of pinning the timestamp forever. The idempotent monotonic + /// upsert still dedupes any overlap. Matches the composite keyset the site's + /// ReadChangedSinceAsync honours (Task 15). /// /// /// Consumes - /// to guarantee forward progress. Whereas the prior implementation ignored - /// the flag entirely and relied solely on the tick cadence, this method now - /// continues pulling within the same tick while the site reports - /// MoreAvailable=true, bounded by - /// . This closes the - /// single-timestamp-saturation edge: if a backlog larger than - /// all shares one - /// exact , the inclusive max-timestamp cursor - /// cannot advance, so the previous code re-pulled the identical window forever - /// across ticks and never drained the tail. Here, a saturated batch whose - /// observed max timestamp did NOT advance past since is detected as a - /// no-progress pin: the loop stops and logs a Warning (the same observability - /// intent as the sibling's stalled signal, without its EventStream state - /// machine), so the pathological site surfaces rather than spinning silently. - /// This diverges from SiteAuditReconciliationActor, which reads - /// MoreAvailable to drive a SiteAuditTelemetryStalledChanged - /// stalled-detection state machine instead of a within-tick continuation drain. + /// to guarantee forward progress. The method continues pulling within the + /// same tick while the site reports MoreAvailable=true, bounded by + /// so a misbehaving site can never + /// spin the dispatcher. Each page advances the in-flight composite cursor. If a + /// page fails to advance the cursor yet the site still reports more available, + /// the site is NOT honouring the after_id keyset (a legacy pre-Task-15 + /// site): continuing would re-pull the identical window forever, so the loop + /// latches the site as pinned and publishes + /// on the EventStream — the + /// dead-end becomes a health-observable condition rather than a silent log line. + /// When a later tick makes progress the latch clears with Pinned=false. + /// This aligns with SiteAuditReconciliationActor's + /// SiteAuditTelemetryStalledChanged transition-only publication. /// /// private async Task ReconcileSiteAsync( SiteEntry site, IPullSiteCallsClient client, ISiteCallAuditRepository repository) { - var cursor = _reconciliationCursors.TryGetValue(site.SiteId, out var c) ? c : DateTime.MinValue; + var cursor = _reconciliationCursors.TryGetValue(site.SiteId, out var c) + ? c + : (Since: DateTime.MinValue, AfterId: (string?)null); // Drain within the tick while the site keeps reporting // MoreAvailable, bounded by MaxReconciliationPagesPerTick so a misbehaving // site can never spin the dispatcher. Each page advances the in-flight - // cursor; a saturated page that fails to advance the cursor is the - // single-timestamp no-progress pin — break and surface it. + // composite cursor; a page that fails to advance it while MoreAvailable is + // still true means the site ignores after_id (legacy) — latch + publish + // the pinned state and break. for (var page = 0; page < MaxReconciliationPagesPerTick; page++) { - var since = cursor; + var since = cursor.Since; + var afterId = cursor.AfterId; var response = await client - .PullAsync(site.SiteId, since, _options.ReconciliationBatchSize, CancellationToken.None) + .PullAsync(site.SiteId, since, afterId, _options.ReconciliationBatchSize, CancellationToken.None) .ConfigureAwait(false); var maxUpdated = since; + var maxAfterId = afterId; var nowUtc = DateTime.UtcNow; foreach (var row in response.SiteCalls) { @@ -637,40 +656,57 @@ public class SiteCallAuditActor : ReceiveActor var siteCall = row with { IngestedAtUtc = nowUtc }; await repository.UpsertAsync(siteCall).ConfigureAwait(false); - if (row.UpdatedAtUtc > maxUpdated) + // Advance the composite max: greater timestamp wins; on a tie the + // greater TrackedOperationId (ordinal, matching the site's SQLite + // BINARY text collation) wins. + var rowId = row.TrackedOperationId.ToString(); + if (row.UpdatedAtUtc > maxUpdated || + (row.UpdatedAtUtc == maxUpdated && string.CompareOrdinal(rowId, maxAfterId) > 0)) { maxUpdated = row.UpdatedAtUtc; + maxAfterId = rowId; } } + // Did the composite cursor move? (timestamp advanced OR — at the same + // timestamp — the id tiebreak advanced). CompareOrdinal handles nulls + // (CompareOrdinal(x, null) > 0 for any non-null x). + var advanced = maxUpdated > since || string.CompareOrdinal(maxAfterId, afterId) != 0; + // Persist the advanced cursor after every page so a fault on a later // page (caught per-site upstream) still keeps the rows already drained. - cursor = maxUpdated; + cursor = (maxUpdated, maxAfterId); _reconciliationCursors[site.SiteId] = cursor; + // Progress resumed → clear any prior pinned latch (publishes the + // Pinned=false transition exactly once). + if (advanced) + { + SetReconciliationPinned(site.SiteId, pinned: false); + } + if (!response.MoreAvailable) { // Backlog fully drained for this site this tick. return; } - if (maxUpdated <= since) + if (!advanced) { - // No-progress pin: the site saturated the batch yet the max - // observed UpdatedAtUtc did not advance past the inclusive cursor - // (a burst of > batch-size rows sharing one exact timestamp). - // Continuing would re-pull the identical window forever, so stop - // and surface it — the inclusive max-timestamp cursor cannot make - // progress on this input without a composite (timestamp,id) - // keyset, which the pull contract does not yet support. + // The composite cursor did not advance yet the site still reports + // MoreAvailable — the site ignored after_id (a legacy pre-keyset + // site), so a continuation would re-pull the identical window + // forever. Latch + publish the pinned state and stop; the next tick + // re-checks and unpins if the site catches up. + SetReconciliationPinned(site.SiteId, pinned: true); _logger.LogWarning( - "SiteCallAudit reconciliation for site {SiteId} cannot make progress: a saturated " - + "batch of more than {BatchSize} rows shares a single UpdatedAtUtc ({CursorUtc:o}), " - + "so the inclusive cursor is pinned. The backlog tail beyond the batch ceiling will " - + "not reconcile until those rows' timestamps differ.", + "SiteCallAudit reconciliation for site {SiteId} is pinned: the composite " + + "(UpdatedAtUtc, TrackedOperationId) cursor did not advance past ({CursorUtc:o}, {AfterId}) " + + "yet the site reports more rows available — the site does not honour the after_id keyset " + + "(legacy). The backlog tail will not reconcile until the site is upgraded.", site.SiteId, - _options.ReconciliationBatchSize, - since); + since, + afterId ?? "(none)"); return; } } @@ -687,6 +723,25 @@ public class SiteCallAuditActor : ReceiveActor MaxReconciliationPagesPerTick); } + /// + /// Latches the per-site reconciliation pinned state and publishes + /// on the actor system + /// EventStream ONLY on a transition — mirroring + /// SiteAuditReconciliationActor.UpdateStalledState so a downstream + /// health subscriber never sees a flood of same-state notifications. + /// + private void SetReconciliationPinned(string siteId, bool pinned) + { + var was = _reconciliationPinned.TryGetValue(siteId, out var prior) && prior; + if (was == pinned) + { + return; + } + + _reconciliationPinned[siteId] = pinned; + Context.System.EventStream.Publish(new SiteCallReconciliationPinnedChanged(siteId, pinned)); + } + // ── Piece B: daily terminal-row purge scheduler ── /// diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteCallAudit/SiteCallReconciliationPinnedChanged.cs b/src/ZB.MOM.WW.ScadaBridge.SiteCallAudit/SiteCallReconciliationPinnedChanged.cs new file mode 100644 index 00000000..0b962ef0 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.SiteCallAudit/SiteCallReconciliationPinnedChanged.cs @@ -0,0 +1,19 @@ +namespace ZB.MOM.WW.ScadaBridge.SiteCallAudit; + +/// +/// Published on the actor system EventStream when a site's Site Call Audit +/// reconciliation puller transitions into or out of the "pinned" state — the +/// composite (UpdatedAtUtc, TrackedOperationId) keyset cursor could not +/// advance yet the site still reports MoreAvailable=true, so the backlog +/// tail cannot be drained. In practice this only happens against a legacy site +/// that ignores the after_id keyset field (a modern site drains a +/// single-timestamp burst via the id tiebreak). Mirrors +/// SiteAuditTelemetryStalledChanged: only transitions are published +/// (Pinned=true on latch, Pinned=false when progress resumes), so +/// a downstream health subscriber never sees a flood of redundant notifications. +/// The dead-end that the previous implementation only logged is now a +/// health-observable condition. +/// +/// The site whose reconciliation cursor pinned (or unpinned). +/// True on entering the pinned state; false when the cursor makes progress again. +public sealed record SiteCallReconciliationPinnedChanged(string SiteId, bool Pinned); diff --git a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/GrpcPullSiteCallsClientTests.cs b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/GrpcPullSiteCallsClientTests.cs index 650b4a15..7d0e1da9 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/GrpcPullSiteCallsClientTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/GrpcPullSiteCallsClientTests.cs @@ -98,7 +98,7 @@ public class GrpcPullSiteCallsClientTests invoker, NullLogger.Instance); - var result = await sut.PullAsync("site-a", BaseTime, batchSize: 256, CancellationToken.None); + var result = await sut.PullAsync("site-a", BaseTime, afterId: null, batchSize: 256, CancellationToken.None); // Endpoint resolution + request shaping. Assert.Equal("http://site-a:8083", invoker.Endpoint); @@ -131,7 +131,7 @@ public class GrpcPullSiteCallsClientTests invoker, NullLogger.Instance); - var result = await sut.PullAsync("site-a", BaseTime, batchSize: 256, CancellationToken.None); + var result = await sut.PullAsync("site-a", BaseTime, afterId: null, batchSize: 256, CancellationToken.None); Assert.Empty(result.SiteCalls); Assert.False(result.MoreAvailable); @@ -150,7 +150,7 @@ public class GrpcPullSiteCallsClientTests invoker, NullLogger.Instance); - var result = await sut.PullAsync("site-a", BaseTime, batchSize: 256, CancellationToken.None); + var result = await sut.PullAsync("site-a", BaseTime, afterId: null, batchSize: 256, CancellationToken.None); Assert.Empty(result.SiteCalls); Assert.False(result.MoreAvailable); @@ -165,7 +165,7 @@ public class GrpcPullSiteCallsClientTests invoker, NullLogger.Instance); - var result = await sut.PullAsync("site-a", BaseTime, batchSize: 256, CancellationToken.None); + var result = await sut.PullAsync("site-a", BaseTime, afterId: null, batchSize: 256, CancellationToken.None); Assert.Empty(result.SiteCalls); Assert.False(result.MoreAvailable); @@ -180,7 +180,7 @@ public class GrpcPullSiteCallsClientTests invoker, NullLogger.Instance); - var result = await sut.PullAsync("site-a", BaseTime, batchSize: 256, CancellationToken.None); + var result = await sut.PullAsync("site-a", BaseTime, afterId: null, batchSize: 256, CancellationToken.None); Assert.Empty(result.SiteCalls); Assert.False(result.MoreAvailable); @@ -211,7 +211,7 @@ public class GrpcPullSiteCallsClientTests NullLogger.Instance); // Must NOT throw — the bad row is dropped, the good rows are returned. - var result = await sut.PullAsync("site-a", BaseTime, batchSize: 256, CancellationToken.None); + var result = await sut.PullAsync("site-a", BaseTime, afterId: null, batchSize: 256, CancellationToken.None); Assert.Equal(2, result.SiteCalls.Count); // Survivors are oldest-first and SourceSite re-stamped from the dialed siteId. @@ -239,7 +239,7 @@ public class GrpcPullSiteCallsClientTests invoker, NullLogger.Instance); - var result = await sut.PullAsync("site-a", minUnspecified, batchSize: 256, CancellationToken.None); + var result = await sut.PullAsync("site-a", minUnspecified, afterId: null, batchSize: 256, CancellationToken.None); Assert.Equal(1, invoker.CallCount); Assert.Equal("http://site-a:8083", invoker.Endpoint); diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteCallAudit.Tests/SiteCallAuditPurgeTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteCallAudit.Tests/SiteCallAuditPurgeTests.cs index 611da1a4..d6114fe0 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.SiteCallAudit.Tests/SiteCallAuditPurgeTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.SiteCallAudit.Tests/SiteCallAuditPurgeTests.cs @@ -42,7 +42,7 @@ public class SiteCallAuditPurgeTests : TestKit private sealed class NoOpPullClient : IPullSiteCallsClient { public Task PullAsync( - string siteId, DateTime sinceUtc, int batchSize, CancellationToken ct) => + string siteId, DateTime sinceUtc, string? afterId, int batchSize, CancellationToken ct) => Task.FromResult(new PullSiteCallsResponse(Array.Empty(), MoreAvailable: false)); } diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteCallAudit.Tests/SiteCallAuditReconciliationTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteCallAudit.Tests/SiteCallAuditReconciliationTests.cs index 9912456e..661e885b 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.SiteCallAudit.Tests/SiteCallAuditReconciliationTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.SiteCallAudit.Tests/SiteCallAuditReconciliationTests.cs @@ -74,7 +74,7 @@ public class SiteCallAuditReconciliationTests : TestKit /// private sealed class ScriptedPullClient : IPullSiteCallsClient { - public List<(string SiteId, DateTime SinceUtc, int BatchSize)> Calls { get; } = new(); + public List<(string SiteId, DateTime SinceUtc, string? AfterId, int BatchSize)> Calls { get; } = new(); private readonly Dictionary> _scripted = new(); private readonly Dictionary _throwOnSite = new(); @@ -91,9 +91,9 @@ public class SiteCallAuditReconciliationTests : TestKit } public Task PullAsync( - string siteId, DateTime sinceUtc, int batchSize, CancellationToken ct) + string siteId, DateTime sinceUtc, string? afterId, int batchSize, CancellationToken ct) { - Calls.Add((siteId, sinceUtc, batchSize)); + Calls.Add((siteId, sinceUtc, afterId, batchSize)); if (_throwOnSite.TryGetValue(siteId, out var ex)) { throw ex; @@ -107,6 +107,67 @@ public class SiteCallAuditReconciliationTests : TestKit } } + /// + /// 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 + /// batchSize rows strictly greater than the composite + /// (sinceUtc, afterId) cursor under the deterministic + /// (UpdatedAtUtc, TrackedOperationId ordinal) ordering, with + /// MoreAvailable=true when unseen rows remain. This is the modern site + /// that DOES honour after_id, so a burst sharing one exact + /// UpdatedAtUtc drains fully via the id tiebreak — no single-timestamp + /// pin. + /// + private sealed class KeysetHonoringPullClient : IPullSiteCallsClient + { + private readonly IReadOnlyList _rows; + public int CallCount { get; private set; } + + public KeysetHonoringPullClient(IEnumerable rows) => + _rows = rows + .OrderBy(r => r.UpdatedAtUtc) + .ThenBy(r => r.TrackedOperationId.ToString(), StringComparer.Ordinal) + .ToList(); + + public Task 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)); + } + } + + /// + /// Legacy pull client (Task 16) that IGNORES afterId and always + /// returns the same saturated page (MoreAvailable=true) — models a + /// pre-keyset site the central actor cannot drain past. The composite cursor + /// cannot advance, so the actor must latch and publish + /// instead of spinning. + /// + private sealed class LegacyIgnoresAfterIdPullClient : IPullSiteCallsClient + { + private readonly IReadOnlyList _rows; + public int CallCount { get; private set; } + + public LegacyIgnoresAfterIdPullClient(IReadOnlyList rows) => _rows = rows; + + public Task PullAsync( + string siteId, DateTime sinceUtc, string? afterId, int batchSize, CancellationToken ct) + { + CallCount++; + return Task.FromResult(new PullSiteCallsResponse(_rows, MoreAvailable: true)); + } + } + /// /// Pull client that ALWAYS returns the same saturated response /// (MoreAvailable=true) regardless of the since cursor — @@ -124,7 +185,7 @@ public class SiteCallAuditReconciliationTests : TestKit public SaturatedPinPullClient(IReadOnlyList rows) => _rows = rows; public Task PullAsync( - string siteId, DateTime sinceUtc, int batchSize, CancellationToken ct) + string siteId, DateTime sinceUtc, string? afterId, int batchSize, CancellationToken ct) { CallCount++; return Task.FromResult(new PullSiteCallsResponse(_rows, MoreAvailable: true)); @@ -436,4 +497,89 @@ public class SiteCallAuditReconciliationTests : TestKit $"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(TimeSpan.FromSeconds(5)); + Assert.Equal(siteId, evt.SiteId); + Assert.True(evt.Pinned, "a legacy site that ignores after_id must publish Pinned=true"); + } }