feat(sitecallaudit): PullSiteCalls reconciliation plumbing (store read + RPC + site handler + central client)
Site Call Audit (#22): build the documented periodic reconciliation PULL self-heal path for the eventually-consistent central SiteCalls mirror, as a dedicated PullSiteCalls gRPC RPC kept separate from the audit pull. This is the pull PLUMBING only; the central reconciliation tick is a separate follow-up. - IOperationTrackingStore.ReadChangedSinceAsync(sinceUtc, batchSize): inclusive UpdatedAtUtc cursor, oldest-first, batch-capped; SQLite impl projects tracking rows onto SiteCallOperational (Kind->Channel, TargetSummary->Target, SourceSite left empty - the store has no site-id column). - sitestream.proto: rpc PullSiteCalls + PullSiteCallsRequest/Response, mirroring PullAuditEvents; regenerated checked-in SiteStreamGrpc/*.cs. - SiteCallDtoMapper.ToDto(SiteCallOperational): inverse of FromDto for the handler. - SiteStreamGrpcServer.PullSiteCalls handler + SetOperationTrackingStore seam; Host wires the seam alongside SetSiteAuditQueue (site roles only). - Central IPullSiteCallsClient + GrpcPullSiteCallsClient (home: AuditLog/Central to reuse ISiteEnumerator; SiteCallAudit does not reference AuditLog). Re-stamps SourceSite from the dialed siteId; no-throw on tolerable transport faults; SpecifyKind (not ToUniversalTime) cursor handling. Central-only DI registration. Tests: ReadChangedSinceAsync (4), PullSiteCalls handler (6), GrpcPullSiteCallsClient (8). Full solution build 0 warnings/0 errors (TreatWarningsAsErrors).
This commit is contained in:
Joseph Doherty
+132
@@ -439,6 +439,138 @@ public class OperationTrackingStoreTests
|
||||
Assert.NotNull(await store.GetStatusAsync(cId)); // kept (non-terminal)
|
||||
}
|
||||
|
||||
// ── Site Call Audit #22: ReadChangedSinceAsync (reconciliation pull) ───
|
||||
|
||||
[Fact]
|
||||
public async Task ReadChangedSinceAsync_ReturnsRowsAtOrAfterCursor_OldestFirst()
|
||||
{
|
||||
var (store, dataSource) = CreateStore(nameof(ReadChangedSinceAsync_ReturnsRowsAtOrAfterCursor_OldestFirst));
|
||||
await using var _store = store;
|
||||
|
||||
// Three rows with distinct UpdatedAtUtc, written out of chronological
|
||||
// order to prove the read sorts by UpdatedAtUtc ascending.
|
||||
var older = TrackedOperationId.New();
|
||||
var middle = TrackedOperationId.New();
|
||||
var newer = TrackedOperationId.New();
|
||||
await store.RecordEnqueueAsync(older, nameof(AuditKind.ApiCallCached), "ERP.A", null, null, "node-a");
|
||||
await store.RecordEnqueueAsync(middle, nameof(AuditKind.DbWriteCached), "DB.B", null, null, "node-b");
|
||||
await store.RecordEnqueueAsync(newer, nameof(AuditKind.ApiCallCached), "ERP.C", null, null, null);
|
||||
|
||||
// Backdate UpdatedAtUtc so the ordering is deterministic and a cursor
|
||||
// can be placed cleanly between rows. (Enqueue stamps DateTime.UtcNow;
|
||||
// we cannot inject the clock, so set the timestamps directly.)
|
||||
var t0 = new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc);
|
||||
SetUpdatedAt(dataSource, older, t0);
|
||||
SetUpdatedAt(dataSource, middle, t0.AddMinutes(10));
|
||||
SetUpdatedAt(dataSource, newer, t0.AddMinutes(20));
|
||||
|
||||
// Cursor at the middle row's UpdatedAtUtc: inclusive lower bound, so
|
||||
// middle + newer come back, older is excluded.
|
||||
var result = await store.ReadChangedSinceAsync(t0.AddMinutes(10), batchSize: 100, CancellationToken.None);
|
||||
|
||||
Assert.Equal(2, result.Count);
|
||||
Assert.Equal(middle, result[0].TrackedOperationId);
|
||||
Assert.Equal(newer, result[1].TrackedOperationId);
|
||||
Assert.True(result[0].UpdatedAtUtc <= result[1].UpdatedAtUtc);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ReadChangedSinceAsync_FromMinValue_ReturnsAllRows()
|
||||
{
|
||||
var (store, _) = CreateStore(nameof(ReadChangedSinceAsync_FromMinValue_ReturnsAllRows));
|
||||
await using var _store = store;
|
||||
|
||||
await store.RecordEnqueueAsync(TrackedOperationId.New(), nameof(AuditKind.ApiCallCached), "A", null, null, null);
|
||||
await store.RecordEnqueueAsync(TrackedOperationId.New(), nameof(AuditKind.ApiCallCached), "B", null, null, null);
|
||||
|
||||
var result = await store.ReadChangedSinceAsync(DateTime.MinValue, batchSize: 100, CancellationToken.None);
|
||||
|
||||
Assert.Equal(2, result.Count);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ReadChangedSinceAsync_IsBatchCapped()
|
||||
{
|
||||
var (store, dataSource) = CreateStore(nameof(ReadChangedSinceAsync_IsBatchCapped));
|
||||
await using var _store = store;
|
||||
|
||||
var ids = new List<TrackedOperationId>();
|
||||
var t0 = new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc);
|
||||
for (var i = 0; i < 5; i++)
|
||||
{
|
||||
var id = TrackedOperationId.New();
|
||||
ids.Add(id);
|
||||
await store.RecordEnqueueAsync(id, nameof(AuditKind.ApiCallCached), $"T{i}", null, null, null);
|
||||
SetUpdatedAt(dataSource, id, t0.AddMinutes(i));
|
||||
}
|
||||
|
||||
var result = await store.ReadChangedSinceAsync(DateTime.MinValue, batchSize: 3, CancellationToken.None);
|
||||
|
||||
// Capped to 3 — and the cap takes the OLDEST 3 (asc order) so the
|
||||
// caller can advance the cursor monotonically across follow-up pulls.
|
||||
Assert.Equal(3, result.Count);
|
||||
Assert.Equal(ids[0], result[0].TrackedOperationId);
|
||||
Assert.Equal(ids[1], result[1].TrackedOperationId);
|
||||
Assert.Equal(ids[2], result[2].TrackedOperationId);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ReadChangedSinceAsync_MapsTrackingRowOntoSiteCallOperational()
|
||||
{
|
||||
var (store, _) = CreateStore(nameof(ReadChangedSinceAsync_MapsTrackingRowOntoSiteCallOperational));
|
||||
await using var _store = store;
|
||||
|
||||
var apiId = TrackedOperationId.New();
|
||||
var dbId = TrackedOperationId.New();
|
||||
await store.RecordEnqueueAsync(apiId, nameof(AuditKind.ApiCallCached), "ERP.GetOrder", "inst-1", "ScriptActor:OnTick", "node-a");
|
||||
await store.RecordEnqueueAsync(dbId, nameof(AuditKind.DbWriteCached), "Historian.Write", null, null, "node-b");
|
||||
await store.RecordAttemptAsync(apiId, nameof(AuditStatus.Attempted), 2, "HTTP 503", 503);
|
||||
await store.RecordTerminalAsync(dbId, nameof(AuditStatus.Parked), "max retries", null);
|
||||
|
||||
var result = await store.ReadChangedSinceAsync(DateTime.MinValue, batchSize: 100, CancellationToken.None);
|
||||
var api = result.Single(r => r.TrackedOperationId == apiId);
|
||||
var db = result.Single(r => r.TrackedOperationId == dbId);
|
||||
|
||||
// Kind → Channel projection.
|
||||
Assert.Equal("ApiOutbound", api.Channel);
|
||||
Assert.Equal("DbOutbound", db.Channel);
|
||||
|
||||
// TargetSummary → Target; SourceNode carried verbatim.
|
||||
Assert.Equal("ERP.GetOrder", api.Target);
|
||||
Assert.Equal("node-a", api.SourceNode);
|
||||
Assert.Equal("node-b", db.SourceNode);
|
||||
|
||||
// Status / RetryCount / LastError / HttpStatus carried through.
|
||||
Assert.Equal(nameof(AuditStatus.Attempted), api.Status);
|
||||
Assert.Equal(2, api.RetryCount);
|
||||
Assert.Equal("HTTP 503", api.LastError);
|
||||
Assert.Equal(503, api.HttpStatus);
|
||||
|
||||
// SourceSite is left empty by the store (the site id is not a tracking
|
||||
// column); the central client re-stamps it from the dialed siteId.
|
||||
Assert.Equal(string.Empty, api.SourceSite);
|
||||
|
||||
// Terminal row carries TerminalAtUtc (UTC kind); active row leaves it null.
|
||||
Assert.Null(api.TerminalAtUtc);
|
||||
Assert.NotNull(db.TerminalAtUtc);
|
||||
Assert.Equal(DateTimeKind.Utc, db.TerminalAtUtc!.Value.Kind);
|
||||
|
||||
// Timestamps round-trip as UTC.
|
||||
Assert.Equal(DateTimeKind.Utc, api.CreatedAtUtc.Kind);
|
||||
Assert.Equal(DateTimeKind.Utc, api.UpdatedAtUtc.Kind);
|
||||
}
|
||||
|
||||
/// <summary>Directly sets a row's UpdatedAtUtc so cursor/ordering tests are deterministic.</summary>
|
||||
private static void SetUpdatedAt(string dataSource, TrackedOperationId id, DateTime updatedAtUtc)
|
||||
{
|
||||
using var connection = OpenVerifierConnection(dataSource);
|
||||
using var cmd = connection.CreateCommand();
|
||||
cmd.CommandText = "UPDATE OperationTracking SET UpdatedAtUtc = $u WHERE TrackedOperationId = $id;";
|
||||
cmd.Parameters.AddWithValue("$u", updatedAtUtc.ToString("o", System.Globalization.CultureInfo.InvariantCulture));
|
||||
cmd.Parameters.AddWithValue("$id", id.ToString());
|
||||
cmd.ExecuteNonQuery();
|
||||
}
|
||||
|
||||
// ── SiteRuntime-024: read/write split + sync-safe Dispose ──────────────
|
||||
|
||||
[Fact]
|
||||
|
||||
Reference in New Issue
Block a user