test(site-call-audit, deployment): cover error, retry and state-transition branches (ScadaBridge#73)
Backfills the branches arch-review 08 flagged as thin, chosen for what they decide rather than for count. Every new test was mutation-checked: the guarded line was broken and the test went red. SiteCallAudit (SiteCallAuditFailurePathTests, 17 cases, in-memory - the 18 MSSQL-backed actor tests skip without a database, so the ingest and read fault paths had no runnable coverage): - a failed page write holds the reconciliation cursor; the next tick re-pulls the same window and advances only after the write lands - the pinned latch publishes Pinned=true once across many pinned ticks and Pinned=false once when the site starts honouring after_id - a site-enumeration fault skips one tick without wedging the single-flight guard - an endless backlog drains at most 50 pages per tick, unpinned, and the next tick resumes from the advanced cursor - purge single-flight: ticks during a blocked purge are dropped - ingest fault replies Accepted=false and the next upsert is accepted - detail / KPI / per-site / per-node / query faults map to Success=false with the root-cause message and an empty payload; the actor keeps serving - query page size clamped to [1, 200]; an empty page carries no cursor - Discard NotParked / OperationFailed (site message verbatim) with no audit row; Discard before transport registration is SiteUnreachable - a throwing audit sink leaves an Applied relay Applied; a missing mirror row or unknown channel still audits the operator, falling back to ApiOutbound DeploymentManager (DeploymentServiceFailurePathTests, 11 cases): - a site that answers Failed commits a Failed record with the site's message (no timeout marker), leaves instance state and the deployed snapshot untouched, and audits Deploy (not DeployFailed) - a prior non-timeout Failed record skips the query-before-redeploy - a failing Failed-status write detaches pending changes and retries once (none when nothing was detached); the caller still gets the original error - Disable / Enable / Delete answered with Success=false leave the instance unchanged, audit the attempt, and surface the site's message or the default Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
This commit is contained in:
Joseph Doherty
+291
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// Gitea #73: <see cref="DeploymentService"/> 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.
|
||||
/// </summary>
|
||||
public class DeploymentServiceFailurePathTests : TestKit
|
||||
{
|
||||
private readonly IDeploymentManagerRepository _repo = Substitute.For<IDeploymentManagerRepository>();
|
||||
private readonly IFlatteningPipeline _pipeline = Substitute.For<IFlatteningPipeline>();
|
||||
private readonly IAuditService _audit = Substitute.For<IAuditService>();
|
||||
|
||||
/// <summary>
|
||||
/// Stand-in central communication actor: answers each site-bound message with whatever
|
||||
/// <c>reply</c> returns for it (null drops it), and counts every message it saw.
|
||||
/// </summary>
|
||||
private sealed class ScriptedSiteActor : ReceiveActor
|
||||
{
|
||||
/// <summary>Initializes the actor.</summary>
|
||||
/// <param name="reply">Maps a site-bound message to the site's answer.</param>
|
||||
/// <param name="seen">Sink for every message received.</param>
|
||||
public ScriptedSiteActor(Func<object, object?> reply, List<object> seen)
|
||||
{
|
||||
Receive<SiteEnvelope>(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<CommunicationService>.Instance);
|
||||
if (commActor is not null)
|
||||
comms.SetCommunicationActor(commActor);
|
||||
|
||||
var siteRepo = Substitute.For<ISiteRepository>();
|
||||
siteRepo.GetSiteByIdAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
|
||||
.Returns(ci => new Site($"Site {ci.ArgAt<int>(0)}", $"site-{ci.ArgAt<int>(0)}") { Id = ci.ArgAt<int>(0) });
|
||||
|
||||
return new DeploymentService(
|
||||
_repo, siteRepo, Substitute.For<ITemplateEngineRepository>(), _pipeline, comms,
|
||||
new OperationLockManager(), _audit,
|
||||
new DiffService(),
|
||||
new RevisionHashService(),
|
||||
new DeploymentStatusNotifier(NullLogger<DeploymentStatusNotifier>.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<DeploymentService>.Instance);
|
||||
}
|
||||
|
||||
private (IActorRef Actor, List<object> Seen) SiteActor(Func<object, object?> reply)
|
||||
{
|
||||
var seen = new List<object>();
|
||||
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<CancellationToken>()).Returns(instance);
|
||||
_pipeline.FlattenAndValidateAsync(id, Arg.Any<CancellationToken>())
|
||||
.Returns(Result<FlatteningPipelineResult>.Success(new FlatteningPipelineResult(
|
||||
new FlattenedConfiguration { InstanceUniqueName = instance.UniqueName },
|
||||
revisionHash,
|
||||
ValidationResult.Success())));
|
||||
return instance;
|
||||
}
|
||||
|
||||
// ── Deploy: the site answers, but with Failed ──
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
/// <param name="initialState">The instance state before the deploy.</param>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[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<DeploymentRecord>(r => record = r), Arg.Any<CancellationToken>());
|
||||
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<Instance>(), Arg.Any<CancellationToken>());
|
||||
await _repo.DidNotReceive().AddDeployedSnapshotAsync(Arg.Any<DeployedConfigSnapshot>(), Arg.Any<CancellationToken>());
|
||||
await _repo.DidNotReceive().UpdateDeployedSnapshotAsync(Arg.Any<DeployedConfigSnapshot>(), Arg.Any<CancellationToken>());
|
||||
// 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<object>(), Arg.Any<CancellationToken>());
|
||||
await _audit.DidNotReceive().LogAsync(Arg.Any<string>(), "DeployFailed", Arg.Any<string>(),
|
||||
Arg.Any<string>(), Arg.Any<string>(), Arg.Any<object>(), Arg.Any<CancellationToken>());
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task Redeploy_PriorFailedForNonTimeoutReason_SkipsSiteQuery()
|
||||
{
|
||||
ArrangeDeployable(11, InstanceState.Enabled);
|
||||
_repo.GetCurrentDeploymentStatusAsync(11, Arg.Any<CancellationToken>())
|
||||
.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 ──
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
/// <param name="detached">What <c>DetachPendingChanges</c> reports.</param>
|
||||
/// <param name="expectedFailedWrites">Failed-status write attempts expected.</param>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[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<DeploymentRecord>(r => record = r), Arg.Any<CancellationToken>());
|
||||
_repo.DetachPendingChanges().Returns(detached);
|
||||
|
||||
var failedWrites = 0;
|
||||
_repo.SaveChangesAsync(Arg.Any<CancellationToken>()).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<object>(), Arg.Any<CancellationToken>());
|
||||
}
|
||||
|
||||
// ── Lifecycle commands: the site answers, but refuses ──
|
||||
|
||||
/// <summary>Lifecycle actions the theory below drives.</summary>
|
||||
public enum LifecycleAction
|
||||
{
|
||||
/// <summary>DisableInstanceAsync.</summary>
|
||||
Disable,
|
||||
|
||||
/// <summary>EnableInstanceAsync.</summary>
|
||||
Enable,
|
||||
|
||||
/// <summary>DeleteInstanceAsync.</summary>
|
||||
Delete,
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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).
|
||||
/// </summary>
|
||||
/// <param name="action">The lifecycle action.</param>
|
||||
/// <param name="initialState">The instance state before the command.</param>
|
||||
/// <param name="siteError">The site's error message, or null.</param>
|
||||
/// <param name="expectedError">The error the caller must see.</param>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[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<CancellationToken>()).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<Instance>(), Arg.Any<CancellationToken>());
|
||||
await _repo.DidNotReceive().DeleteInstanceAsync(Arg.Any<int>(), Arg.Any<CancellationToken>());
|
||||
await _repo.DidNotReceive().SaveChangesAsync(Arg.Any<CancellationToken>());
|
||||
await _audit.Received(1).LogAsync("admin", action.ToString(), "Instance", "20", "LcInst",
|
||||
Arg.Any<object>(), Arg.Any<CancellationToken>());
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// Gitea #73: in-memory coverage of <see cref="SiteCallAuditActor"/>'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.
|
||||
/// </summary>
|
||||
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),
|
||||
};
|
||||
|
||||
/// <summary>Repository whose every member can be scripted; unscripted members are inert.</summary>
|
||||
private sealed class ScriptableRepo : ISiteCallAuditRepository
|
||||
{
|
||||
private int _upsertManyCalls;
|
||||
|
||||
/// <summary>Gets the rows written by successful batched upserts.</summary>
|
||||
public List<SiteCall> Written { get; } = new();
|
||||
|
||||
/// <summary>Number of leading UpsertManyAsync calls that throw.</summary>
|
||||
public int FailFirstUpsertMany { get; set; }
|
||||
|
||||
/// <summary>Optional override for single-row upserts.</summary>
|
||||
public Func<SiteCall, Task>? OnUpsert { get; set; }
|
||||
|
||||
/// <summary>Optional override for GetAsync.</summary>
|
||||
public Func<TrackedOperationId, Task<SiteCall?>>? OnGet { get; set; }
|
||||
|
||||
/// <summary>Optional override for QueryAsync.</summary>
|
||||
public Func<SiteCallQueryFilter, SiteCallPaging, Task<IReadOnlyList<SiteCall>>>? OnQuery { get; set; }
|
||||
|
||||
/// <summary>Optional override for PurgeTerminalAsync.</summary>
|
||||
public Func<DateTime, Task<int>>? OnPurge { get; set; }
|
||||
|
||||
/// <summary>When set, every KPI computation throws it.</summary>
|
||||
public Exception? KpiFault { get; set; }
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task UpsertAsync(SiteCall siteCall, CancellationToken ct = default) =>
|
||||
OnUpsert?.Invoke(siteCall) ?? Task.CompletedTask;
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task UpsertManyAsync(IReadOnlyList<SiteCall> 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;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task<SiteCall?> GetAsync(TrackedOperationId id, CancellationToken ct = default) =>
|
||||
OnGet?.Invoke(id) ?? Task.FromResult<SiteCall?>(null);
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task<IReadOnlyList<SiteCall>> QueryAsync(
|
||||
SiteCallQueryFilter filter, SiteCallPaging paging, CancellationToken ct = default) =>
|
||||
OnQuery?.Invoke(filter, paging) ?? Task.FromResult<IReadOnlyList<SiteCall>>(Array.Empty<SiteCall>());
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task<int> PurgeTerminalAsync(DateTime olderThanUtc, CancellationToken ct = default) =>
|
||||
OnPurge?.Invoke(olderThanUtc) ?? Task.FromResult(0);
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task<SiteCallKpiSnapshot> ComputeKpisAsync(
|
||||
DateTime stuckCutoff, DateTime intervalSince, CancellationToken ct = default) =>
|
||||
KpiFault is null
|
||||
? Task.FromResult(new SiteCallKpiSnapshot(1, 2, 3, 4, null, 5))
|
||||
: Task.FromException<SiteCallKpiSnapshot>(KpiFault);
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task<IReadOnlyList<SiteCallSiteKpiSnapshot>> ComputePerSiteKpisAsync(
|
||||
DateTime stuckCutoff, DateTime intervalSince, CancellationToken ct = default) =>
|
||||
KpiFault is null
|
||||
? Task.FromResult<IReadOnlyList<SiteCallSiteKpiSnapshot>>(Array.Empty<SiteCallSiteKpiSnapshot>())
|
||||
: Task.FromException<IReadOnlyList<SiteCallSiteKpiSnapshot>>(KpiFault);
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task<IReadOnlyList<SiteCallNodeKpiSnapshot>> ComputePerNodeKpisAsync(
|
||||
DateTime stuckCutoff, DateTime intervalSince, CancellationToken ct = default) =>
|
||||
KpiFault is null
|
||||
? Task.FromResult<IReadOnlyList<SiteCallNodeKpiSnapshot>>(Array.Empty<SiteCallNodeKpiSnapshot>())
|
||||
: Task.FromException<IReadOnlyList<SiteCallNodeKpiSnapshot>>(KpiFault);
|
||||
}
|
||||
|
||||
/// <summary>Enumerator that throws on its first N calls, counting every call.</summary>
|
||||
private sealed class FlakyEnumerator(int failFirst, params SiteEntry[] sites) : ISiteEnumerator
|
||||
{
|
||||
private int _calls;
|
||||
|
||||
/// <summary>Gets the number of enumerations performed so far.</summary>
|
||||
public int Calls => Volatile.Read(ref _calls);
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task<IReadOnlyList<SiteEntry>> EnumerateAsync(CancellationToken ct = default) =>
|
||||
Interlocked.Increment(ref _calls) <= failFirst
|
||||
? throw new InvalidOperationException("simulated config DB outage")
|
||||
: Task.FromResult<IReadOnlyList<SiteEntry>>(sites);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
private sealed class SwitchablePullClient(IEnumerable<SiteCall> rows) : IPullSiteCallsClient
|
||||
{
|
||||
private readonly List<SiteCall> _rows = rows
|
||||
.OrderBy(r => r.UpdatedAtUtc)
|
||||
.ThenBy(r => r.TrackedOperationId.ToString(), StringComparer.Ordinal)
|
||||
.ToList();
|
||||
|
||||
/// <summary>Gets every pull recorded, with the enumerator tick it ran under.</summary>
|
||||
public List<(DateTime Since, string? AfterId)> Calls { get; } = new();
|
||||
|
||||
/// <summary>Gets or sets whether the site ignores <c>after_id</c> (pre-keyset).</summary>
|
||||
public volatile bool Legacy;
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task<PullSiteCallsResponse> 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));
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>A site with an endless backlog: every pull returns one newer row and MoreAvailable.</summary>
|
||||
private sealed class EndlessBacklogPullClient(FlakyEnumerator ticks) : IPullSiteCallsClient
|
||||
{
|
||||
private static readonly DateTime Start = new(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc);
|
||||
private int _n;
|
||||
|
||||
/// <summary>Gets every pull with the enumeration (tick) number it belongs to.</summary>
|
||||
public List<(int Tick, DateTime Since, DateTime Returned)> Calls { get; } = new();
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task<PullSiteCallsResponse> 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));
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>No-op pull client that only arms the reconciliation scheduler.</summary>
|
||||
private sealed class NoOpPullClient : IPullSiteCallsClient
|
||||
{
|
||||
/// <inheritdoc />
|
||||
public Task<PullSiteCallsResponse> PullAsync(
|
||||
string siteId, DateTime sinceUtc, string? afterId, int batchSize, CancellationToken ct) =>
|
||||
Task.FromResult(new PullSiteCallsResponse(Array.Empty<SiteCall>(), MoreAvailable: false));
|
||||
}
|
||||
|
||||
/// <summary>Central audit sink that throws on every write.</summary>
|
||||
private sealed class ThrowingAuditWriter : ICentralAuditWriter
|
||||
{
|
||||
private int _calls;
|
||||
|
||||
/// <summary>Gets the number of write attempts.</summary>
|
||||
public int Calls => Volatile.Read(ref _calls);
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task WriteAsync(ZB.MOM.WW.Audit.AuditEvent evt, CancellationToken ct = default)
|
||||
{
|
||||
Interlocked.Increment(ref _calls);
|
||||
throw new InvalidOperationException("audit store down");
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Central audit sink that records every row.</summary>
|
||||
private sealed class RecordingAuditWriter : ICentralAuditWriter
|
||||
{
|
||||
/// <summary>Gets the recorded rows.</summary>
|
||||
public List<AuditRowProjection.AuditRowValues> Events { get; } = new();
|
||||
|
||||
/// <inheritdoc />
|
||||
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<SiteCallAuditActor>.Instance)));
|
||||
}
|
||||
|
||||
private void Stop(IActorRef actor)
|
||||
{
|
||||
Watch(actor);
|
||||
Sys.Stop(actor);
|
||||
ExpectTerminated(actor, TimeSpan.FromSeconds(3));
|
||||
}
|
||||
|
||||
// ── Reconciliation: retry, latch and guard transitions ──
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
[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);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
[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<SiteCallReconciliationPinnedChanged>(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<SiteCallReconciliationPinnedChanged>(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);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
[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);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
[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 ──
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void Purge_InFlight_FurtherTicksDropped_UntilItCompletes()
|
||||
{
|
||||
var release = new TaskCompletionSource<int>(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 ──
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
[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<UpsertSiteCallReply>();
|
||||
Assert.Equal(first.TrackedOperationId, failed.TrackedOperationId);
|
||||
Assert.False(failed.Accepted);
|
||||
|
||||
actor.Tell(new UpsertSiteCallCommand(second));
|
||||
var ok = ExpectMsg<UpsertSiteCallReply>();
|
||||
Assert.Equal(second.TrackedOperationId, ok.TrackedOperationId);
|
||||
Assert.True(ok.Accepted);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
[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<SiteCall?>(fault),
|
||||
OnQuery = (_, _) => Task.FromException<IReadOnlyList<SiteCall>>(fault),
|
||||
};
|
||||
var actor = CreateActor(repo, FastTicks());
|
||||
|
||||
actor.Tell(new SiteCallDetailRequest("d", Guid.NewGuid()));
|
||||
var detail = ExpectMsg<SiteCallDetailResponse>();
|
||||
Assert.False(detail.Success);
|
||||
Assert.Equal("SQL timeout", detail.ErrorMessage);
|
||||
Assert.Null(detail.Detail);
|
||||
|
||||
actor.Tell(new SiteCallKpiRequest("k"));
|
||||
var kpi = ExpectMsg<SiteCallKpiResponse>();
|
||||
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<PerSiteSiteCallKpiResponse>();
|
||||
Assert.False(perSite.Success);
|
||||
Assert.Equal("SQL timeout", perSite.ErrorMessage);
|
||||
Assert.Empty(perSite.Sites);
|
||||
|
||||
actor.Tell(new PerNodeSiteCallKpiRequest("n"));
|
||||
var perNode = ExpectMsg<PerNodeSiteCallKpiResponse>();
|
||||
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<SiteCallQueryResponse>();
|
||||
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<SiteCallKpiResponse>().Success);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
[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<int>();
|
||||
var repo = new ScriptableRepo
|
||||
{
|
||||
OnQuery = (_, paging) =>
|
||||
{
|
||||
lock (seen) seen.Add(paging.PageSize);
|
||||
return Task.FromResult<IReadOnlyList<SiteCall>>(Array.Empty<SiteCall>());
|
||||
},
|
||||
};
|
||||
var actor = CreateActor(repo, FastTicks());
|
||||
|
||||
actor.Tell(new SiteCallQueryRequest("q", null, null, null, null, false, null, null, null, null, requested));
|
||||
var response = ExpectMsg<SiteCallQueryResponse>();
|
||||
|
||||
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 ──
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
/// <param name="siteError">The error the site's ack carries, or null for a clean no-op.</param>
|
||||
/// <param name="expected">The outcome the relay must report.</param>
|
||||
[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<SiteCall?>(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<SiteEnvelope>().Message;
|
||||
central.Reply(new ParkedOperationActionAck(relay.CorrelationId, Applied: false, siteError));
|
||||
|
||||
var response = ExpectMsg<DiscardSiteCallResponse>();
|
||||
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);
|
||||
}
|
||||
|
||||
/// <summary>A Discard before any central→site transport is registered is reported unreachable.</summary>
|
||||
[Fact]
|
||||
public void DiscardSiteCall_BeforeCentralCommunicationRegistered_ReportsSiteUnreachable()
|
||||
{
|
||||
var actor = CreateActor(new ScriptableRepo(), FastTicks());
|
||||
|
||||
actor.Tell(new DiscardSiteCallRequest("corr", Guid.NewGuid(), "site-north"));
|
||||
|
||||
var response = ExpectMsg<DiscardSiteCallResponse>();
|
||||
Assert.Equal(SiteCallRelayOutcome.SiteUnreachable, response.Outcome);
|
||||
Assert.False(response.SiteReachable);
|
||||
Assert.False(response.Success);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void RetrySiteCall_AuditSinkThrows_RelayStillReportsApplied()
|
||||
{
|
||||
var id = Guid.NewGuid();
|
||||
var writer = new ThrowingAuditWriter();
|
||||
var repo = new ScriptableRepo { OnGet = _ => Task.FromResult<SiteCall?>(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<SiteEnvelope>().Message;
|
||||
central.Reply(new ParkedOperationActionAck(relay.CorrelationId, Applied: true));
|
||||
|
||||
var response = ExpectMsg<RetrySiteCallResponse>();
|
||||
Assert.Equal(SiteCallRelayOutcome.Applied, response.Outcome);
|
||||
Assert.True(response.Success);
|
||||
Assert.Equal(1, writer.Calls); // the write WAS attempted
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
/// <param name="mirrorChannel">The mirror row's channel, or null for no mirror row.</param>
|
||||
[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<SiteCall?>(
|
||||
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<SiteEnvelope>().Message;
|
||||
central.Reply(new ParkedOperationActionAck(relay.CorrelationId, Applied: true));
|
||||
|
||||
Assert.True(ExpectMsg<RetrySiteCallResponse>().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);
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user