diff --git a/tests/ZB.MOM.WW.ScadaBridge.DeploymentManager.Tests/DeploymentServiceFailurePathTests.cs b/tests/ZB.MOM.WW.ScadaBridge.DeploymentManager.Tests/DeploymentServiceFailurePathTests.cs new file mode 100644 index 00000000..bd8f06aa --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.DeploymentManager.Tests/DeploymentServiceFailurePathTests.cs @@ -0,0 +1,291 @@ +using Akka.Actor; +using Akka.TestKit.Xunit2; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using NSubstitute; +using ZB.MOM.WW.ScadaBridge.Commons.Entities.Deployment; +using ZB.MOM.WW.ScadaBridge.Commons.Entities.Instances; +using ZB.MOM.WW.ScadaBridge.Commons.Entities.Sites; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Deployment; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Lifecycle; +using ZB.MOM.WW.ScadaBridge.Commons.Types; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Flattening; +using ZB.MOM.WW.ScadaBridge.Communication; +using ZB.MOM.WW.ScadaBridge.TemplateEngine.Flattening; + +namespace ZB.MOM.WW.ScadaBridge.DeploymentManager.Tests; + +/// +/// Gitea #73: branches the existing suites leave untested — +/// a site that answers but REFUSES (as opposed to timing out or throwing), the query-before- +/// redeploy trigger for a non-timeout failure, and the detach-and-retry-once recovery of the +/// Failed-status write. Each one decides an instance's or a deployment record's state. +/// +public class DeploymentServiceFailurePathTests : TestKit +{ + private readonly IDeploymentManagerRepository _repo = Substitute.For(); + private readonly IFlatteningPipeline _pipeline = Substitute.For(); + private readonly IAuditService _audit = Substitute.For(); + + /// + /// Stand-in central communication actor: answers each site-bound message with whatever + /// reply returns for it (null drops it), and counts every message it saw. + /// + private sealed class ScriptedSiteActor : ReceiveActor + { + /// Initializes the actor. + /// Maps a site-bound message to the site's answer. + /// Sink for every message received. + public ScriptedSiteActor(Func reply, List seen) + { + Receive(env => + { + lock (seen) seen.Add(env.Message); + var answer = reply(env.Message); + if (answer is not null) + Sender.Tell(answer); + }); + } + } + + private DeploymentService CreateService(IActorRef? commActor) + { + var comms = new CommunicationService( + Options.Create(new CommunicationOptions + { + QueryTimeout = TimeSpan.FromSeconds(5), + DeploymentTimeout = TimeSpan.FromSeconds(5), + }), + NullLogger.Instance); + if (commActor is not null) + comms.SetCommunicationActor(commActor); + + var siteRepo = Substitute.For(); + siteRepo.GetSiteByIdAsync(Arg.Any(), Arg.Any()) + .Returns(ci => new Site($"Site {ci.ArgAt(0)}", $"site-{ci.ArgAt(0)}") { Id = ci.ArgAt(0) }); + + return new DeploymentService( + _repo, siteRepo, Substitute.For(), _pipeline, comms, + new OperationLockManager(), _audit, + new DiffService(), + new RevisionHashService(), + new DeploymentStatusNotifier(NullLogger.Instance), + Options.Create(new DeploymentManagerOptions + { + OperationLockTimeout = TimeSpan.FromSeconds(5), + LifecycleCommandTimeout = TimeSpan.FromSeconds(5), + }), + Options.Create(new CommunicationOptions + { + CentralFetchBaseUrl = "https://central.test:9000", + PendingDeploymentTtl = TimeSpan.FromMinutes(5), + }), + NullLogger.Instance); + } + + private (IActorRef Actor, List Seen) SiteActor(Func reply) + { + var seen = new List(); + return (Sys.ActorOf(Props.Create(() => new ScriptedSiteActor(reply, seen))), seen); + } + + private Instance ArrangeDeployable(int id, InstanceState state, string revisionHash = "sha256:new") + { + var instance = new Instance($"Inst{id}") { Id = id, SiteId = 1, State = state }; + _repo.GetInstanceByIdAsync(id, Arg.Any()).Returns(instance); + _pipeline.FlattenAndValidateAsync(id, Arg.Any()) + .Returns(Result.Success(new FlatteningPipelineResult( + new FlattenedConfiguration { InstanceUniqueName = instance.UniqueName }, + revisionHash, + ValidationResult.Success()))); + return instance; + } + + // ── Deploy: the site answers, but with Failed ── + + /// + /// A site that receives the deploy but reports Failed (e.g. its own compile step rejected the + /// config) commits a Failed record carrying the SITE's message — not a timeout marker, so the + /// next redeploy does not waste a site query — and leaves the instance's state and deployed + /// snapshot exactly as they were: the site is not running the new config. + /// + /// The instance state before the deploy. + /// A task that represents the asynchronous operation. + [Theory] + [InlineData(InstanceState.NotDeployed)] + [InlineData(InstanceState.Disabled)] + public async Task Deploy_SiteReportsFailed_RecordFailedWithSiteError_InstanceAndSnapshotUntouched(InstanceState initialState) + { + var instance = ArrangeDeployable(10, initialState); + DeploymentRecord? record = null; + await _repo.AddDeploymentRecordAsync(Arg.Do(r => record = r), Arg.Any()); + var (actor, _) = SiteActor(msg => msg is RefreshDeploymentCommand d + ? new DeploymentStatusResponse(d.DeploymentId, d.InstanceUniqueName, + DeploymentStatus.Failed, "Script 'OnStart' failed to compile at site", DateTimeOffset.UtcNow) + : null); + + var result = await CreateService(actor).DeployInstanceAsync(10, "admin"); + + Assert.True(result.IsFailure); + Assert.Equal("Deployment failed: Script 'OnStart' failed to compile at site", result.Error); + Assert.NotNull(record); + Assert.Equal(DeploymentStatus.Failed, record!.Status); + Assert.Equal("Script 'OnStart' failed to compile at site", record.ErrorMessage); + Assert.NotNull(record.CompletedAt); + + Assert.Equal(initialState, instance.State); + await _repo.DidNotReceive().UpdateInstanceAsync(Arg.Any(), Arg.Any()); + await _repo.DidNotReceive().AddDeployedSnapshotAsync(Arg.Any(), Arg.Any()); + await _repo.DidNotReceive().UpdateDeployedSnapshotAsync(Arg.Any(), Arg.Any()); + // The outcome is audited as a Deploy with Failed status — not as a DeployFailed fault. + await _audit.Received(1).LogAsync("admin", "Deploy", "Instance", "10", "Inst10", + Arg.Any(), Arg.Any()); + await _audit.DidNotReceive().LogAsync(Arg.Any(), "DeployFailed", Arg.Any(), + Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()); + } + + /// + /// A prior deployment that failed for a NON-timeout reason (the site refused it) cannot have + /// been applied, so the redeploy goes straight to the site without a state query. Only an + /// InProgress or timeout-marked Failed record may hide an apply central never heard about. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task Redeploy_PriorFailedForNonTimeoutReason_SkipsSiteQuery() + { + ArrangeDeployable(11, InstanceState.Enabled); + _repo.GetCurrentDeploymentStatusAsync(11, Arg.Any()) + .Returns(new DeploymentRecord("dep-prior", "admin") + { + InstanceId = 11, + Status = DeploymentStatus.Failed, + ErrorMessage = "Script 'OnStart' failed to compile at site", + }); + var (actor, seen) = SiteActor(msg => msg switch + { + RefreshDeploymentCommand d => new DeploymentStatusResponse( + d.DeploymentId, d.InstanceUniqueName, DeploymentStatus.Success, null, DateTimeOffset.UtcNow), + DeploymentStateQueryRequest q => new DeploymentStateQueryResponse( + q.CorrelationId, q.InstanceUniqueName, true, "dep-x", "sha256:new", DateTimeOffset.UtcNow), + _ => null, + }); + + var result = await CreateService(actor).DeployInstanceAsync(11, "admin"); + + Assert.True(result.IsSuccess, result.IsFailure ? result.Error : null); + lock (seen) + { + Assert.DoesNotContain(seen, m => m is DeploymentStateQueryRequest); + Assert.Single(seen, m => m is RefreshDeploymentCommand); + } + } + + // ── Failed-status write: detach and retry once ── + + /// + /// When the cleanup write of the Failed status itself fails, the pending changes are detached + /// and the write is retried ONCE on the clean unit of work, so the record still reaches its + /// terminal state; with nothing to detach there is nothing to retry. Either way the caller + /// gets the ORIGINAL deployment error, never the cleanup error. + /// + /// What DetachPendingChanges reports. + /// Failed-status write attempts expected. + /// A task that represents the asynchronous operation. + [Theory] + [InlineData(2, 2)] + [InlineData(0, 1)] + public async Task Deploy_FailedStatusWriteFails_DetachesAndRetriesOnce(int detached, int expectedFailedWrites) + { + ArrangeDeployable(12, InstanceState.NotDeployed); + DeploymentRecord? record = null; + await _repo.AddDeploymentRecordAsync(Arg.Do(r => record = r), Arg.Any()); + _repo.DetachPendingChanges().Returns(detached); + + var failedWrites = 0; + _repo.SaveChangesAsync(Arg.Any()).Returns(_ => + { + if (record?.Status == DeploymentStatus.Failed && ++failedWrites == 1) + throw new InvalidOperationException("poisoned unit of work"); + return Task.FromResult(1); + }); + + // No communication actor: the send faults with a non-timeout error. + var result = await CreateService(commActor: null).DeployInstanceAsync(12, "admin"); + + Assert.True(result.IsFailure); + Assert.DoesNotContain("poisoned", result.Error); + Assert.Equal(DeploymentStatus.Failed, record!.Status); + Assert.StartsWith("Deployment error:", record.ErrorMessage); + Assert.Equal(expectedFailedWrites, failedWrites); + _repo.Received(1).DetachPendingChanges(); + // The failure audit row is written even when the status write could not be. + await _audit.Received(1).LogAsync("admin", "DeployFailed", "Instance", "12", "Inst12", + Arg.Any(), Arg.Any()); + } + + // ── Lifecycle commands: the site answers, but refuses ── + + /// Lifecycle actions the theory below drives. + public enum LifecycleAction + { + /// DisableInstanceAsync. + Disable, + + /// EnableInstanceAsync. + Enable, + + /// DeleteInstanceAsync. + Delete, + } + + /// + /// A lifecycle command the site answers with Success=false leaves the central instance + /// exactly as it was — state unchanged, nothing deleted — still audits the attempt, and + /// surfaces the site's message (or a default when the site gave none). + /// + /// The lifecycle action. + /// The instance state before the command. + /// The site's error message, or null. + /// The error the caller must see. + /// A task that represents the asynchronous operation. + [Theory] + [InlineData(LifecycleAction.Disable, InstanceState.Enabled, "Instance actor busy", "Instance actor busy")] + [InlineData(LifecycleAction.Disable, InstanceState.Enabled, null, "Disable failed.")] + [InlineData(LifecycleAction.Enable, InstanceState.Disabled, "No stored config", "No stored config")] + [InlineData(LifecycleAction.Enable, InstanceState.Disabled, null, "Enable failed.")] + [InlineData(LifecycleAction.Delete, InstanceState.Enabled, "Delete refused", "Delete refused")] + [InlineData(LifecycleAction.Delete, InstanceState.Disabled, null, "Delete failed. Site may be unreachable.")] + public async Task Lifecycle_SiteRefuses_InstanceUnchanged_AttemptAudited( + LifecycleAction action, InstanceState initialState, string? siteError, string expectedError) + { + var instance = new Instance("LcInst") { Id = 20, SiteId = 1, State = initialState }; + _repo.GetInstanceByIdAsync(20, Arg.Any()).Returns(instance); + var (actor, _) = SiteActor(msg => msg switch + { + DisableInstanceCommand c => new InstanceLifecycleResponse(c.CommandId, c.InstanceUniqueName, false, siteError, DateTimeOffset.UtcNow), + EnableInstanceCommand c => new InstanceLifecycleResponse(c.CommandId, c.InstanceUniqueName, false, siteError, DateTimeOffset.UtcNow), + DeleteInstanceCommand c => new InstanceLifecycleResponse(c.CommandId, c.InstanceUniqueName, false, siteError, DateTimeOffset.UtcNow), + _ => null, + }); + var service = CreateService(actor); + + var result = action switch + { + LifecycleAction.Disable => await service.DisableInstanceAsync(20, "admin"), + LifecycleAction.Enable => await service.EnableInstanceAsync(20, "admin"), + _ => await service.DeleteInstanceAsync(20, "admin"), + }; + + Assert.True(result.IsFailure); + Assert.Equal(expectedError, result.Error); + Assert.Equal(initialState, instance.State); + await _repo.DidNotReceive().UpdateInstanceAsync(Arg.Any(), Arg.Any()); + await _repo.DidNotReceive().DeleteInstanceAsync(Arg.Any(), Arg.Any()); + await _repo.DidNotReceive().SaveChangesAsync(Arg.Any()); + await _audit.Received(1).LogAsync("admin", action.ToString(), "Instance", "20", "LcInst", + Arg.Any(), Arg.Any()); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteCallAudit.Tests/SiteCallAuditFailurePathTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteCallAudit.Tests/SiteCallAuditFailurePathTests.cs new file mode 100644 index 00000000..993fee4a --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.SiteCallAudit.Tests/SiteCallAuditFailurePathTests.cs @@ -0,0 +1,662 @@ +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.Interfaces.Services; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Audit; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Integration; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.RemoteQuery; +using ZB.MOM.WW.ScadaBridge.Commons.Types; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; +using ZB.MOM.WW.ScadaBridge.Communication; + +namespace ZB.MOM.WW.ScadaBridge.SiteCallAudit.Tests; + +/// +/// Gitea #73: in-memory coverage of 's error, retry and +/// state-transition branches that the existing suites leave untested — or test only behind +/// the MSSQL fixture, which skips without a database. Covers: a failed page write holding the +/// reconciliation cursor so the next tick re-pulls the window; the pinned latch clearing (and +/// publishing only on transitions); an enumerator fault not wedging the single-flight guard; +/// the per-tick page ceiling; the purge single-flight guard; the read-side and ingest fault +/// mapping; and the Discard / audit-sink branches of the central→site relay. +/// +public class SiteCallAuditFailurePathTests : TestKit +{ + private static SiteCall Row(string site, DateTime updatedAtUtc, string channel = "ApiOutbound") => new() + { + TrackedOperationId = TrackedOperationId.New(), + Channel = channel, + Target = "ERP.GetOrder", + SourceSite = site, + SourceNode = "node-a", + Status = "Parked", + RetryCount = 1, + CreatedAtUtc = updatedAtUtc, + UpdatedAtUtc = updatedAtUtc, + IngestedAtUtc = updatedAtUtc, + }; + + private static SiteCallAuditOptions FastTicks(int batchSize = 500) => new() + { + ReconciliationIntervalOverride = TimeSpan.FromMilliseconds(100), + PurgeIntervalOverride = TimeSpan.FromMinutes(10), + ReconciliationBatchSize = batchSize, + RelayTimeout = TimeSpan.FromMilliseconds(500), + }; + + /// Repository whose every member can be scripted; unscripted members are inert. + private sealed class ScriptableRepo : ISiteCallAuditRepository + { + private int _upsertManyCalls; + + /// Gets the rows written by successful batched upserts. + public List Written { get; } = new(); + + /// Number of leading UpsertManyAsync calls that throw. + public int FailFirstUpsertMany { get; set; } + + /// Optional override for single-row upserts. + public Func? OnUpsert { get; set; } + + /// Optional override for GetAsync. + public Func>? OnGet { get; set; } + + /// Optional override for QueryAsync. + public Func>>? OnQuery { get; set; } + + /// Optional override for PurgeTerminalAsync. + public Func>? OnPurge { get; set; } + + /// When set, every KPI computation throws it. + public Exception? KpiFault { get; set; } + + /// + public Task UpsertAsync(SiteCall siteCall, CancellationToken ct = default) => + OnUpsert?.Invoke(siteCall) ?? Task.CompletedTask; + + /// + public Task UpsertManyAsync(IReadOnlyList siteCalls, CancellationToken ct = default) + { + if (Interlocked.Increment(ref _upsertManyCalls) <= FailFirstUpsertMany) + throw new InvalidOperationException("simulated page write failure"); + lock (Written) Written.AddRange(siteCalls); + return Task.CompletedTask; + } + + /// + public Task GetAsync(TrackedOperationId id, CancellationToken ct = default) => + OnGet?.Invoke(id) ?? Task.FromResult(null); + + /// + public Task> QueryAsync( + SiteCallQueryFilter filter, SiteCallPaging paging, CancellationToken ct = default) => + OnQuery?.Invoke(filter, paging) ?? Task.FromResult>(Array.Empty()); + + /// + public Task PurgeTerminalAsync(DateTime olderThanUtc, CancellationToken ct = default) => + OnPurge?.Invoke(olderThanUtc) ?? Task.FromResult(0); + + /// + public Task ComputeKpisAsync( + DateTime stuckCutoff, DateTime intervalSince, CancellationToken ct = default) => + KpiFault is null + ? Task.FromResult(new SiteCallKpiSnapshot(1, 2, 3, 4, null, 5)) + : Task.FromException(KpiFault); + + /// + public Task> ComputePerSiteKpisAsync( + DateTime stuckCutoff, DateTime intervalSince, CancellationToken ct = default) => + KpiFault is null + ? Task.FromResult>(Array.Empty()) + : Task.FromException>(KpiFault); + + /// + public Task> ComputePerNodeKpisAsync( + DateTime stuckCutoff, DateTime intervalSince, CancellationToken ct = default) => + KpiFault is null + ? Task.FromResult>(Array.Empty()) + : Task.FromException>(KpiFault); + } + + /// Enumerator that throws on its first N calls, counting every call. + private sealed class FlakyEnumerator(int failFirst, params SiteEntry[] sites) : ISiteEnumerator + { + private int _calls; + + /// Gets the number of enumerations performed so far. + public int Calls => Volatile.Read(ref _calls); + + /// + public Task> EnumerateAsync(CancellationToken ct = default) => + Interlocked.Increment(ref _calls) <= failFirst + ? throw new InvalidOperationException("simulated config DB outage") + : Task.FromResult>(sites); + } + + /// + /// Keyset-honouring site over a fixed row set that can be switched into "legacy" mode, where + /// it ignores the cursor and always reports a saturated page. + /// + private sealed class SwitchablePullClient(IEnumerable rows) : IPullSiteCallsClient + { + private readonly List _rows = rows + .OrderBy(r => r.UpdatedAtUtc) + .ThenBy(r => r.TrackedOperationId.ToString(), StringComparer.Ordinal) + .ToList(); + + /// Gets every pull recorded, with the enumerator tick it ran under. + public List<(DateTime Since, string? AfterId)> Calls { get; } = new(); + + /// Gets or sets whether the site ignores after_id (pre-keyset). + public volatile bool Legacy; + + /// + public Task PullAsync( + string siteId, DateTime sinceUtc, string? afterId, int batchSize, CancellationToken ct) + { + lock (Calls) Calls.Add((sinceUtc, afterId)); + if (Legacy) + return Task.FromResult(new PullSiteCallsResponse(_rows.Take(batchSize).ToList(), MoreAvailable: true)); + + 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(); + return Task.FromResult(new PullSiteCallsResponse(page, MoreAvailable: greater.Count > page.Count)); + } + } + + /// A site with an endless backlog: every pull returns one newer row and MoreAvailable. + private sealed class EndlessBacklogPullClient(FlakyEnumerator ticks) : IPullSiteCallsClient + { + private static readonly DateTime Start = new(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc); + private int _n; + + /// Gets every pull with the enumeration (tick) number it belongs to. + public List<(int Tick, DateTime Since, DateTime Returned)> Calls { get; } = new(); + + /// + public Task PullAsync( + string siteId, DateTime sinceUtc, string? afterId, int batchSize, CancellationToken ct) + { + var row = Row(siteId, Start.AddSeconds(Interlocked.Increment(ref _n))); + lock (Calls) Calls.Add((ticks.Calls, sinceUtc, row.UpdatedAtUtc)); + return Task.FromResult(new PullSiteCallsResponse(new[] { row }, MoreAvailable: true)); + } + } + + /// No-op pull client that only arms the reconciliation scheduler. + private sealed class NoOpPullClient : IPullSiteCallsClient + { + /// + public Task PullAsync( + string siteId, DateTime sinceUtc, string? afterId, int batchSize, CancellationToken ct) => + Task.FromResult(new PullSiteCallsResponse(Array.Empty(), MoreAvailable: false)); + } + + /// Central audit sink that throws on every write. + private sealed class ThrowingAuditWriter : ICentralAuditWriter + { + private int _calls; + + /// Gets the number of write attempts. + public int Calls => Volatile.Read(ref _calls); + + /// + public Task WriteAsync(ZB.MOM.WW.Audit.AuditEvent evt, CancellationToken ct = default) + { + Interlocked.Increment(ref _calls); + throw new InvalidOperationException("audit store down"); + } + } + + /// Central audit sink that records every row. + private sealed class RecordingAuditWriter : ICentralAuditWriter + { + /// Gets the recorded rows. + public List Events { get; } = new(); + + /// + public Task WriteAsync(ZB.MOM.WW.Audit.AuditEvent evt, CancellationToken ct = default) + { + lock (Events) Events.Add(evt.AsRow()); + return Task.CompletedTask; + } + } + + private IActorRef CreateActor( + ISiteCallAuditRepository repo, + SiteCallAuditOptions options, + ISiteEnumerator? sites = null, + IPullSiteCallsClient? client = null, + ICentralAuditWriter? auditWriter = null, + bool serialize = true) + { + var provider = TestSupport.StubRepositoryProvider.Build(repo, sites, client, auditWriter, serialize); + return Sys.ActorOf(Props.Create(() => new SiteCallAuditActor( + provider, options, NullLogger.Instance))); + } + + private void Stop(IActorRef actor) + { + Watch(actor); + Sys.Stop(actor); + ExpectTerminated(actor, TimeSpan.FromSeconds(3)); + } + + // ── Reconciliation: retry, latch and guard transitions ── + + /// + /// A page whose batched upsert fails must NOT advance the site's cursor: the next tick + /// re-pulls the same window from the old cursor, and once the write succeeds the cursor + /// moves past it. Advancing first would silently drop the failed page from the mirror. + /// + [Fact] + public void Reconciliation_PageWriteFails_CursorHeld_NextTickRePullsSameWindow() + { + var t1 = new DateTime(2026, 5, 20, 10, 0, 0, DateTimeKind.Utc); + var rows = new[] { Row("siteA", t1), Row("siteA", t1.AddMinutes(1)) }; + var client = new SwitchablePullClient(rows); + var repo = new ScriptableRepo { FailFirstUpsertMany = 1 }; + + var actor = CreateActor(repo, FastTicks(), + new FlakyEnumerator(0, new SiteEntry("siteA", "http://a")), client); + try + { + AwaitAssert(() => + { + lock (client.Calls) Assert.True(client.Calls.Count >= 3, $"pulls: {client.Calls.Count}"); + }, TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(50)); + + List<(DateTime Since, string? AfterId)> calls; + lock (client.Calls) calls = client.Calls.ToList(); + Assert.Equal(DateTime.MinValue, calls[0].Since); // failed write + Assert.Equal(DateTime.MinValue, calls[1].Since); // same window re-pulled + Assert.Equal(t1.AddMinutes(1), calls[2].Since); // advanced only after the write landed + lock (repo.Written) Assert.Equal(2, repo.Written.Count); + } + finally + { + Stop(actor); + } + } + + /// + /// The pinned latch is transition-only: a site that stays pinned across many ticks publishes + /// Pinned=true once, and when it starts honouring the keyset again it publishes Pinned=false + /// exactly once. + /// + [Fact] + public void Reconciliation_PinnedSiteResumesProgress_PublishesEachTransitionOnce() + { + var ts = new DateTime(2026, 5, 20, 10, 0, 0, DateTimeKind.Utc); + var rows = new[] { Row("siteP", ts), Row("siteP", ts), Row("siteP", ts) }; + var client = new SwitchablePullClient(rows) { Legacy = true }; + var probe = CreateTestProbe(); + Sys.EventStream.Subscribe(probe.Ref, typeof(SiteCallReconciliationPinnedChanged)); + + var actor = CreateActor(new ScriptableRepo(), FastTicks(batchSize: 2), + new FlakyEnumerator(0, new SiteEntry("siteP", "http://p")), client); + try + { + Assert.True(probe.ExpectMsg(TimeSpan.FromSeconds(5)).Pinned); + + // Several more pinned ticks: no repeat publication. + var pullsAtPin = client.Calls.Count; + AwaitCondition(() => client.Calls.Count >= pullsAtPin + 3, TimeSpan.FromSeconds(5)); + probe.ExpectNoMsg(TimeSpan.FromMilliseconds(200)); + + client.Legacy = false; // the site is upgraded and honours after_id + var unpinned = probe.ExpectMsg(TimeSpan.FromSeconds(5)); + Assert.Equal("siteP", unpinned.SiteId); + Assert.False(unpinned.Pinned); + probe.ExpectNoMsg(TimeSpan.FromMilliseconds(400)); + } + finally + { + Sys.EventStream.Unsubscribe(probe.Ref); + Stop(actor); + } + } + + /// + /// A site-enumeration fault skips that tick but lowers the single-flight guard, so the next + /// tick reconciles normally — one config-DB blip must not wedge the self-heal for good. + /// + [Fact] + public void Reconciliation_EnumeratorThrows_TickSkipped_NextTickReconciles() + { + var row = Row("siteA", new DateTime(2026, 5, 20, 10, 0, 0, DateTimeKind.Utc)); + var sites = new FlakyEnumerator(2, new SiteEntry("siteA", "http://a")); + var client = new SwitchablePullClient(new[] { row }); + var repo = new ScriptableRepo(); + + var actor = CreateActor(repo, FastTicks(), sites, client); + try + { + AwaitAssert(() => + { + lock (repo.Written) Assert.Contains(repo.Written, r => r.TrackedOperationId == row.TrackedOperationId); + }, TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(50)); + + Assert.True(sites.Calls >= 3); + // No pull happened during the two failed enumerations. + lock (client.Calls) Assert.Equal(DateTime.MinValue, client.Calls[0].Since); + } + finally + { + Stop(actor); + } + } + + /// + /// A site that always reports more rows is drained at most 50 pages per tick (bounded, not + /// pinned — the cursor advances each page), and the next tick resumes from the advanced + /// cursor rather than starting over. + /// + [Fact] + public void Reconciliation_EndlessBacklog_BoundedPerTick_NextTickResumesFromAdvancedCursor() + { + var sites = new FlakyEnumerator(0, new SiteEntry("siteB", "http://b")); + var client = new EndlessBacklogPullClient(sites); + var probe = CreateTestProbe(); + Sys.EventStream.Subscribe(probe.Ref, typeof(SiteCallReconciliationPinnedChanged)); + + var actor = CreateActor(new ScriptableRepo(), FastTicks(batchSize: 1), sites, client); + try + { + AwaitAssert(() => + { + lock (client.Calls) Assert.Contains(client.Calls, c => c.Tick >= 2); + }, TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(50)); + + List<(int Tick, DateTime Since, DateTime Returned)> calls; + lock (client.Calls) calls = client.Calls.ToList(); + var firstTick = calls.Where(c => c.Tick == 1).ToList(); + var secondTickFirstPull = calls.First(c => c.Tick == 2); + + Assert.Equal(50, firstTick.Count); + Assert.Equal(firstTick[^1].Returned, secondTickFirstPull.Since); + probe.ExpectNoMsg(TimeSpan.FromMilliseconds(200)); // progressing, so never pinned + } + finally + { + Sys.EventStream.Unsubscribe(probe.Ref); + Stop(actor); + } + } + + // ── Purge single-flight ── + + /// + /// A purge that outlives its interval is not stacked: further ticks are dropped while it is + /// in flight, and the next purge runs once it completes. + /// + [Fact] + public void Purge_InFlight_FurtherTicksDropped_UntilItCompletes() + { + var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var calls = 0; + var repo = new ScriptableRepo + { + OnPurge = _ => Interlocked.Increment(ref calls) == 1 ? release.Task : Task.FromResult(0), + }; + var options = FastTicks(); + options.ReconciliationIntervalOverride = TimeSpan.FromMinutes(10); + options.PurgeIntervalOverride = TimeSpan.FromMilliseconds(50); + + var actor = CreateActor(repo, options, new FlakyEnumerator(0), new NoOpPullClient(), serialize: false); + try + { + AwaitCondition(() => Volatile.Read(ref calls) == 1, TimeSpan.FromSeconds(3)); + // ~10 purge intervals elapse while the first purge is blocked. + ExpectNoMsg(TimeSpan.FromMilliseconds(500)); + Assert.Equal(1, Volatile.Read(ref calls)); + + release.SetResult(3); + AwaitCondition(() => Volatile.Read(ref calls) >= 2, TimeSpan.FromSeconds(3)); + } + finally + { + release.TrySetResult(0); + Stop(actor); + } + } + + // ── Ingest and read-side fault mapping ── + + /// + /// An ingest write fault replies Accepted=false (the telemetry sender keeps the row for a + /// later push) and the singleton stays alive to accept the next upsert. + /// + [Fact] + public void Upsert_RepositoryThrows_RepliesNotAccepted_ThenAcceptsNext() + { + var calls = 0; + var repo = new ScriptableRepo + { + OnUpsert = _ => Interlocked.Increment(ref calls) == 1 + ? throw new TimeoutException("SQL timeout") + : Task.CompletedTask, + }; + var actor = CreateActor(repo, FastTicks()); + var first = Row("siteA", DateTime.UtcNow); + var second = Row("siteA", DateTime.UtcNow); + + actor.Tell(new UpsertSiteCallCommand(first)); + var failed = ExpectMsg(); + Assert.Equal(first.TrackedOperationId, failed.TrackedOperationId); + Assert.False(failed.Accepted); + + actor.Tell(new UpsertSiteCallCommand(second)); + var ok = ExpectMsg(); + Assert.Equal(second.TrackedOperationId, ok.TrackedOperationId); + Assert.True(ok.Accepted); + } + + /// + /// Every read handler maps a repository fault onto Success=false with the root-cause message + /// and an empty payload — never a dead letter or a crashed singleton — and keeps serving. + /// + [Fact] + public void ReadHandlers_RepositoryThrows_ReplyFailureWithRootCause_AndStayAlive() + { + var fault = new InvalidOperationException("wrapper", new TimeoutException("SQL timeout")); + var repo = new ScriptableRepo + { + KpiFault = fault, + OnGet = _ => Task.FromException(fault), + OnQuery = (_, _) => Task.FromException>(fault), + }; + var actor = CreateActor(repo, FastTicks()); + + actor.Tell(new SiteCallDetailRequest("d", Guid.NewGuid())); + var detail = ExpectMsg(); + Assert.False(detail.Success); + Assert.Equal("SQL timeout", detail.ErrorMessage); + Assert.Null(detail.Detail); + + actor.Tell(new SiteCallKpiRequest("k")); + var kpi = ExpectMsg(); + Assert.False(kpi.Success); + Assert.Equal("SQL timeout", kpi.ErrorMessage); + Assert.Equal(0, kpi.BufferedCount + kpi.ParkedCount + kpi.StuckCount); + + actor.Tell(new PerSiteSiteCallKpiRequest("s")); + var perSite = ExpectMsg(); + Assert.False(perSite.Success); + Assert.Equal("SQL timeout", perSite.ErrorMessage); + Assert.Empty(perSite.Sites); + + actor.Tell(new PerNodeSiteCallKpiRequest("n")); + var perNode = ExpectMsg(); + Assert.False(perNode.Success); + Assert.Equal("SQL timeout", perNode.ErrorMessage); + Assert.Empty(perNode.Nodes); + + actor.Tell(new SiteCallQueryRequest("q", null, null, null, null, false, null, null, null, null, 50)); + var query = ExpectMsg(); + Assert.False(query.Success); + Assert.Equal("SQL timeout", query.ErrorMessage); + Assert.Empty(query.SiteCalls); + Assert.Null(query.NextAfterId); + + // Still alive: a healthy KPI read after the faults succeeds. + repo.KpiFault = null; + actor.Tell(new SiteCallKpiRequest("k2")); + Assert.True(ExpectMsg().Success); + } + + /// + /// The query page size is clamped to [1, 200] before it reaches SQL (an unbounded page from a + /// crafted request would materialise the whole table), and an empty page carries no cursor. + /// + [Theory] + [InlineData(0, 1)] + [InlineData(-5, 1)] + [InlineData(10_000, 200)] + [InlineData(37, 37)] + public void Query_PageSizeClamped_EmptyPageHasNoCursor(int requested, int expected) + { + var seen = new List(); + var repo = new ScriptableRepo + { + OnQuery = (_, paging) => + { + lock (seen) seen.Add(paging.PageSize); + return Task.FromResult>(Array.Empty()); + }, + }; + var actor = CreateActor(repo, FastTicks()); + + actor.Tell(new SiteCallQueryRequest("q", null, null, null, null, false, null, null, null, null, requested)); + var response = ExpectMsg(); + + Assert.True(response.Success); + lock (seen) Assert.Equal(new[] { expected }, seen); + Assert.Null(response.NextAfterCreatedAtUtc); + Assert.Null(response.NextAfterId); + } + + // ── Relay: Discard branches and the best-effort audit sink ── + + /// + /// Discard maps the site's ack the same way Retry does: a clean no-op is NotParked, an ack + /// with an error is OperationFailed carrying the site's message verbatim — the site was + /// reached either way — and neither emits an operator audit row. + /// + /// The error the site's ack carries, or null for a clean no-op. + /// The outcome the relay must report. + [Theory] + [InlineData(null, SiteCallRelayOutcome.NotParked)] + [InlineData("S&F buffer locked", SiteCallRelayOutcome.OperationFailed)] + public void DiscardSiteCall_SiteDidNotApply_MapsOutcome_NoAuditRow(string? siteError, SiteCallRelayOutcome expected) + { + var id = Guid.NewGuid(); + var writer = new RecordingAuditWriter(); + var repo = new ScriptableRepo { OnGet = _ => Task.FromResult(Row("site-north", DateTime.UtcNow)) }; + var central = CreateTestProbe(); + var actor = CreateActor(repo, FastTicks(), auditWriter: writer); + actor.Tell(new RegisterCentralCommunication(central.Ref)); + + actor.Tell(new DiscardSiteCallRequest("corr", id, "site-north", RequestedBy: "jdoe")); + var relay = (DiscardParkedOperation)central.ExpectMsg().Message; + central.Reply(new ParkedOperationActionAck(relay.CorrelationId, Applied: false, siteError)); + + var response = ExpectMsg(); + Assert.Equal(expected, response.Outcome); + Assert.False(response.Success); + Assert.True(response.SiteReachable); + if (siteError is not null) + Assert.Equal(siteError, response.ErrorMessage); + else + Assert.Contains("no longer parked", response.ErrorMessage); + + ExpectNoMsg(TimeSpan.FromMilliseconds(200)); + lock (writer.Events) Assert.Empty(writer.Events); + } + + /// A Discard before any central→site transport is registered is reported unreachable. + [Fact] + public void DiscardSiteCall_BeforeCentralCommunicationRegistered_ReportsSiteUnreachable() + { + var actor = CreateActor(new ScriptableRepo(), FastTicks()); + + actor.Tell(new DiscardSiteCallRequest("corr", Guid.NewGuid(), "site-north")); + + var response = ExpectMsg(); + Assert.Equal(SiteCallRelayOutcome.SiteUnreachable, response.Outcome); + Assert.False(response.SiteReachable); + Assert.False(response.Success); + } + + /// + /// The operator audit row is best-effort: a throwing audit sink must not turn an Applied + /// relay into a failure — the site already acted, and reporting otherwise would invite a + /// duplicate retry. + /// + [Fact] + public void RetrySiteCall_AuditSinkThrows_RelayStillReportsApplied() + { + var id = Guid.NewGuid(); + var writer = new ThrowingAuditWriter(); + var repo = new ScriptableRepo { OnGet = _ => Task.FromResult(Row("site-north", DateTime.UtcNow)) }; + var central = CreateTestProbe(); + var actor = CreateActor(repo, FastTicks(), auditWriter: writer); + actor.Tell(new RegisterCentralCommunication(central.Ref)); + + actor.Tell(new RetrySiteCallRequest("corr", id, "site-north", RequestedBy: "jdoe")); + var relay = (RetryParkedOperation)central.ExpectMsg().Message; + central.Reply(new ParkedOperationActionAck(relay.CorrelationId, Applied: true)); + + var response = ExpectMsg(); + Assert.Equal(SiteCallRelayOutcome.Applied, response.Outcome); + Assert.True(response.Success); + Assert.Equal(1, writer.Calls); // the write WAS attempted + } + + /// + /// When the mirror row is missing (telemetry not ingested yet) or carries an unknown channel, + /// the operator audit row is still written — falling back to ApiOutbound — so who acted is + /// never lost. + /// + /// The mirror row's channel, or null for no mirror row. + [Theory] + [InlineData(null)] + [InlineData("NotAChannel")] + public void RetrySiteCall_MirrorRowMissingOrUnknownChannel_AuditFallsBackToApiOutbound(string? mirrorChannel) + { + var id = Guid.NewGuid(); + var writer = new RecordingAuditWriter(); + var repo = new ScriptableRepo + { + OnGet = _ => Task.FromResult( + mirrorChannel is null ? null : Row("site-north", DateTime.UtcNow, mirrorChannel)), + }; + var central = CreateTestProbe(); + var actor = CreateActor(repo, FastTicks(), auditWriter: writer); + actor.Tell(new RegisterCentralCommunication(central.Ref)); + + actor.Tell(new RetrySiteCallRequest("corr", id, "site-north", RequestedBy: "jdoe")); + var relay = (RetryParkedOperation)central.ExpectMsg().Message; + central.Reply(new ParkedOperationActionAck(relay.CorrelationId, Applied: true)); + + Assert.True(ExpectMsg().Success); + AwaitAssert(() => + { + lock (writer.Events) + { + var evt = Assert.Single(writer.Events); + Assert.Equal(AuditChannel.ApiOutbound, evt.Channel); + Assert.Equal(AuditStatus.Submitted, evt.Status); + Assert.Equal("jdoe", evt.Actor); + Assert.Equal(id, evt.CorrelationId); + } + }); + } +}