test(auditlog): end-to-end ParentExecutionId correlation + docs
This commit is contained in:
Joseph Doherty
@@ -0,0 +1,562 @@
|
||||
using System.Text;
|
||||
using System.Text.Json;
|
||||
using Akka.Actor;
|
||||
using Akka.TestKit.Xunit2;
|
||||
using Microsoft.AspNetCore.Builder;
|
||||
using Microsoft.AspNetCore.Hosting;
|
||||
using Microsoft.AspNetCore.Http;
|
||||
using Microsoft.AspNetCore.TestHost;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Microsoft.Extensions.Options;
|
||||
using NSubstitute;
|
||||
using ScadaLink.AuditLog.Central;
|
||||
using ScadaLink.AuditLog.Site;
|
||||
using ScadaLink.AuditLog.Site.Telemetry;
|
||||
using ScadaLink.AuditLog.Tests.Integration.Infrastructure;
|
||||
using ScadaLink.Commons.Entities.Audit;
|
||||
using ScadaLink.Commons.Interfaces.Repositories;
|
||||
using ScadaLink.Commons.Interfaces.Services;
|
||||
using ScadaLink.Commons.Messages.InboundApi;
|
||||
using ScadaLink.Commons.Messages.Notification;
|
||||
using ScadaLink.Commons.Types.Audit;
|
||||
using ScadaLink.Commons.Types.Enums;
|
||||
using ScadaLink.ConfigurationDatabase;
|
||||
using ScadaLink.ConfigurationDatabase.Repositories;
|
||||
using ScadaLink.ConfigurationDatabase.Tests.Migrations;
|
||||
using ScadaLink.InboundAPI;
|
||||
using ScadaLink.InboundAPI.Middleware;
|
||||
using ScadaLink.NotificationOutbox;
|
||||
using ScadaLink.NotificationOutbox.Delivery;
|
||||
using ScadaLink.NotificationOutbox.Messages;
|
||||
using ScadaLink.SiteRuntime.Scripts;
|
||||
using ScadaLink.StoreAndForward;
|
||||
|
||||
namespace ScadaLink.AuditLog.Tests.Integration;
|
||||
|
||||
/// <summary>
|
||||
/// Audit Log #23 — <b>ParentExecutionId cross-execution correlation</b> headline
|
||||
/// end-to-end suite. Verifies the inbound-API → routed-site-script bridge: an
|
||||
/// inbound HTTP request runs an inbound method script that calls
|
||||
/// <c>Route.Call</c> into a site instance; the routed site script does a sync
|
||||
/// <c>ExternalSystem.Call</c>, a cached call and a <c>Notify.Send</c>. Every
|
||||
/// audit row the routed run produces — site + central, sync + cached lifecycle
|
||||
/// + <c>NotifySend</c>/<c>NotifyDeliver</c> — must carry
|
||||
/// <see cref="AuditEvent.ParentExecutionId"/> equal to the inbound request's
|
||||
/// <see cref="AuditEvent.ExecutionId"/>, while the routed run has its own
|
||||
/// distinct <see cref="AuditEvent.ExecutionId"/> and the inbound
|
||||
/// <see cref="AuditKind.InboundRequest"/> row is top-level
|
||||
/// (<c>ParentExecutionId = NULL</c>).
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// This is the integration-level counterpart to <see cref="ExecutionIdCorrelationTests"/>:
|
||||
/// where that suite drives a single <see cref="ScriptRuntimeContext"/> run and
|
||||
/// asserts the shared per-run <c>ExecutionId</c>, this suite spans <b>two</b>
|
||||
/// executions on opposite sides of the inbound→routed bridge and asserts the
|
||||
/// cross-execution <c>ParentExecutionId</c> linkage plus
|
||||
/// <see cref="IAuditLogRepository.GetExecutionTreeAsync"/>.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// The bridge is exercised through the genuine production glue:
|
||||
/// <list type="bullet">
|
||||
/// <item><description>the real <see cref="AuditWriteMiddleware"/> in a
|
||||
/// Microsoft.AspNetCore.TestHost pipeline — mints the inbound request's
|
||||
/// per-request <c>ExecutionId</c> once, stashes it on
|
||||
/// <see cref="HttpContext.Items"/>, and emits the top-level
|
||||
/// <see cref="AuditKind.InboundRequest"/> row via the real
|
||||
/// <see cref="CentralAuditWriter"/>;</description></item>
|
||||
/// <item><description>the real <see cref="InboundScriptExecutor"/> +
|
||||
/// <see cref="RouteHelper"/> — the executor binds the stashed inbound
|
||||
/// <c>ExecutionId</c> via <see cref="RouteHelper.WithParentExecutionId"/>, so a
|
||||
/// <c>Route.To(...).Call(...)</c> inside the inbound script builds a
|
||||
/// <see cref="RouteToCallRequest"/> carrying
|
||||
/// <see cref="RouteToCallRequest.ParentExecutionId"/>.</description></item>
|
||||
/// </list>
|
||||
/// Only the cross-cluster routing transport is substituted: the test
|
||||
/// <see cref="BridgingInstanceRouter"/> stands in for
|
||||
/// <c>CommunicationServiceInstanceRouter</c> exactly as the production site
|
||||
/// (<c>DeploymentManagerActor</c> → <c>ScriptActor</c> → <c>ScriptExecutionActor</c>)
|
||||
/// would — it reads <see cref="RouteToCallRequest.ParentExecutionId"/> off the
|
||||
/// wire request and threads it into the routed <see cref="ScriptRuntimeContext"/>
|
||||
/// as <c>parentExecutionId</c>. A multi-node cluster is out of scope for an
|
||||
/// in-process test (mirroring <c>SiteAuditPushFlowTests</c>'s relay).
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// The central audit store is the real <see cref="AuditLogRepository"/> over the
|
||||
/// per-class <see cref="MsSqlMigrationFixture"/> MSSQL database; the routed run's
|
||||
/// site rows reach it through the real <see cref="SqliteAuditWriter"/> hot-path +
|
||||
/// <see cref="SiteAuditTelemetryActor"/> drain, the cached lifecycle rows through
|
||||
/// the production <see cref="CachedCallTelemetryForwarder"/>, and the
|
||||
/// <c>NotifyDeliver</c> rows through the real central
|
||||
/// <see cref="NotificationOutboxActor"/> dispatcher.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public class ParentExecutionIdCorrelationTests : TestKit, IClassFixture<MsSqlMigrationFixture>
|
||||
{
|
||||
private readonly MsSqlMigrationFixture _fixture;
|
||||
|
||||
public ParentExecutionIdCorrelationTests(MsSqlMigrationFixture fixture)
|
||||
{
|
||||
_fixture = fixture;
|
||||
}
|
||||
|
||||
private const string RoutedInstanceCode = "Plant.Pump42";
|
||||
private const string RoutedScriptName = "OnInboundRouted";
|
||||
private const string ExternalSystemName = "ERP";
|
||||
private const string ExternalMethodName = "GetOrder";
|
||||
private const string NotifyListName = "ops-team";
|
||||
|
||||
/// <summary>Per-run site id (Guid suffix) so concurrent tests sharing the MSSQL fixture stay isolated.</summary>
|
||||
private static string NewSiteId() =>
|
||||
"test-parentexec-" + Guid.NewGuid().ToString("N").Substring(0, 8);
|
||||
|
||||
private ScadaLinkDbContext CreateContext()
|
||||
{
|
||||
var options = new DbContextOptionsBuilder<ScadaLinkDbContext>()
|
||||
.UseSqlServer(_fixture.ConnectionString)
|
||||
.ConfigureWarnings(w => w.Ignore(
|
||||
Microsoft.EntityFrameworkCore.Diagnostics.RelationalEventId.PendingModelChangesWarning))
|
||||
.Options;
|
||||
return new ScadaLinkDbContext(options);
|
||||
}
|
||||
|
||||
[SkippableFact]
|
||||
public async Task InboundRoutedRun_AllRoutedRows_CarryInboundExecutionId_AsParentExecutionId()
|
||||
{
|
||||
Skip.IfNot(_fixture.Available, _fixture.SkipReason);
|
||||
|
||||
var siteId = NewSiteId();
|
||||
|
||||
// ── Central — repository + ingest actor + audit writer over the MSSQL fixture ──
|
||||
var centralServices = new ServiceCollection();
|
||||
centralServices.AddDbContext<ScadaLinkDbContext>(opts =>
|
||||
opts.UseSqlServer(_fixture.ConnectionString)
|
||||
.ConfigureWarnings(w => w.Ignore(
|
||||
Microsoft.EntityFrameworkCore.Diagnostics.RelationalEventId.PendingModelChangesWarning)));
|
||||
centralServices.AddScoped<IAuditLogRepository>(sp =>
|
||||
new AuditLogRepository(sp.GetRequiredService<ScadaLinkDbContext>()));
|
||||
centralServices.AddScoped<ISiteCallAuditRepository>(sp =>
|
||||
new SiteCallAuditRepository(sp.GetRequiredService<ScadaLinkDbContext>()));
|
||||
centralServices.AddScoped<INotificationOutboxRepository>(sp =>
|
||||
new NotificationOutboxRepository(sp.GetRequiredService<ScadaLinkDbContext>()));
|
||||
centralServices.AddScoped<INotificationRepository>(sp =>
|
||||
new NotificationRepository(sp.GetRequiredService<ScadaLinkDbContext>()));
|
||||
// The NotifyDeliver dispatch path runs through this same long-lived
|
||||
// provider — a stub adapter that always reports a successful delivery.
|
||||
centralServices.AddScoped<INotificationDeliveryAdapter>(_ => new AlwaysDeliversAdapter());
|
||||
await using var centralProvider = centralServices.BuildServiceProvider();
|
||||
|
||||
var ingestActor = Sys.ActorOf(Props.Create(() => new AuditLogIngestActor(
|
||||
(IServiceProvider)centralProvider,
|
||||
NullLogger<AuditLogIngestActor>.Instance)));
|
||||
var centralAuditWriter = new CentralAuditWriter(
|
||||
centralProvider, NullLogger<CentralAuditWriter>.Instance);
|
||||
|
||||
// ── Site — SQLite audit writer (hot-path) drained to central by the
|
||||
// real SiteAuditTelemetryActor through the stub gRPC client. The sync
|
||||
// ApiCall row and the NotifySend row flow through this chain. ──
|
||||
await using var sqliteWriter = new SqliteAuditWriter(
|
||||
Options.Create(new SqliteAuditWriterOptions
|
||||
{
|
||||
DatabasePath = "ignored",
|
||||
BatchSize = 64,
|
||||
ChannelCapacity = 1024,
|
||||
}),
|
||||
NullLogger<SqliteAuditWriter>.Instance,
|
||||
connectionStringOverride:
|
||||
$"Data Source=file:auditlog-parentexec-{Guid.NewGuid():N}?mode=memory&cache=shared");
|
||||
var ring = new RingBufferFallback();
|
||||
var siteAuditWriter = new FallbackAuditWriter(
|
||||
sqliteWriter, ring, new NoOpAuditWriteFailureCounter(),
|
||||
NullLogger<FallbackAuditWriter>.Instance);
|
||||
var stubClient = new DirectActorSiteStreamAuditClient(ingestActor);
|
||||
Sys.ActorOf(Props.Create(() => new SiteAuditTelemetryActor(
|
||||
(ISiteAuditQueue)sqliteWriter,
|
||||
stubClient,
|
||||
Options.Create(new SiteAuditTelemetryOptions
|
||||
{
|
||||
BatchSize = 256,
|
||||
BusyIntervalSeconds = 1,
|
||||
IdleIntervalSeconds = 1,
|
||||
}),
|
||||
NullLogger<SiteAuditTelemetryActor>.Instance)));
|
||||
|
||||
// Cached-call telemetry: production forwarder + dispatcher that also
|
||||
// pushes each combined packet through the stub client into the central
|
||||
// dual-write transaction (same wiring CombinedTelemetryHarness uses).
|
||||
var cachedForwarder = new CombinedTelemetryDispatcher(
|
||||
new CachedCallTelemetryForwarder(
|
||||
siteAuditWriter, trackingStore: null,
|
||||
NullLogger<CachedCallTelemetryForwarder>.Instance),
|
||||
stubClient);
|
||||
|
||||
// Site Store-and-Forward — Notify.Send buffers a NotificationSubmit here.
|
||||
using var safKeepAlive = new Microsoft.Data.Sqlite.SqliteConnection(
|
||||
$"Data Source=parentexec-saf-{Guid.NewGuid():N};Mode=Memory;Cache=Shared");
|
||||
safKeepAlive.Open();
|
||||
var safStorage = new StoreAndForwardStorage(
|
||||
safKeepAlive.ConnectionString, NullLogger<StoreAndForwardStorage>.Instance);
|
||||
await safStorage.InitializeAsync();
|
||||
var storeAndForward = new StoreAndForwardService(
|
||||
safStorage,
|
||||
new StoreAndForwardOptions
|
||||
{
|
||||
DefaultRetryInterval = TimeSpan.Zero,
|
||||
DefaultMaxRetries = 3,
|
||||
RetryTimerInterval = TimeSpan.FromMinutes(10),
|
||||
},
|
||||
NullLogger<StoreAndForwardService>.Instance);
|
||||
|
||||
// ── Outbound external-system client (routed run): sync Call succeeds,
|
||||
// CachedCall completes immediately (WasBuffered=false) so the script
|
||||
// helper emits the Submit + Attempted + CachedResolve lifecycle. ──
|
||||
var externalClient = Substitute.For<IExternalSystemClient>();
|
||||
externalClient
|
||||
.CallAsync(ExternalSystemName, ExternalMethodName,
|
||||
Arg.Any<IReadOnlyDictionary<string, object?>?>(), Arg.Any<CancellationToken>())
|
||||
.Returns(new ExternalCallResult(true, "{\"ok\":true}", null));
|
||||
externalClient
|
||||
.CachedCallAsync(ExternalSystemName, ExternalMethodName,
|
||||
Arg.Any<IReadOnlyDictionary<string, object?>?>(),
|
||||
Arg.Any<string?>(), Arg.Any<CancellationToken>(),
|
||||
Arg.Any<ScadaLink.Commons.Types.TrackedOperationId?>(),
|
||||
Arg.Any<Guid?>(), Arg.Any<string?>(), Arg.Any<Guid?>())
|
||||
.Returns(new ExternalCallResult(true, "{\"ok\":true}", null, WasBuffered: false));
|
||||
|
||||
// ── The routing transport stand-in: builds the routed ScriptRuntimeContext
|
||||
// carrying RouteToCallRequest.ParentExecutionId — exactly what the
|
||||
// production site handler (DeploymentManagerActor) does. ──
|
||||
var router = new BridgingInstanceRouter(
|
||||
siteId,
|
||||
externalClient,
|
||||
siteAuditWriter,
|
||||
cachedForwarder,
|
||||
storeAndForward);
|
||||
|
||||
// ── The inbound API method script: it calls Route.Call into the site
|
||||
// instance. The real InboundScriptExecutor binds the inbound request's
|
||||
// ExecutionId onto the RouteHelper, so the routed call carries it as
|
||||
// ParentExecutionId. ──
|
||||
var inboundMethod = new ScadaLink.Commons.Entities.InboundApi.ApiMethod(
|
||||
"submitOrder",
|
||||
$"return await Route.To(\"{RoutedInstanceCode}\").Call(\"{RoutedScriptName}\", new {{ order = 7 }});");
|
||||
var locator = Substitute.For<IInstanceLocator>();
|
||||
locator.GetSiteIdForInstanceAsync(RoutedInstanceCode, Arg.Any<CancellationToken>())
|
||||
.Returns(siteId);
|
||||
var scriptExecutor = new InboundScriptExecutor(
|
||||
NullLogger<InboundScriptExecutor>.Instance,
|
||||
new ServiceCollection().BuildServiceProvider());
|
||||
Assert.True(scriptExecutor.CompileAndRegister(inboundMethod));
|
||||
|
||||
// ── Act — issue the inbound HTTP request through a TestHost pipeline
|
||||
// fronted by the real AuditWriteMiddleware. The endpoint handler reads
|
||||
// the middleware-stashed inbound ExecutionId and runs the inbound
|
||||
// method script with it as parentExecutionId. ──
|
||||
using var host = await BuildInboundHostAsync(centralAuditWriter, async ctx =>
|
||||
{
|
||||
var inboundExecutionId = (Guid)ctx.Items[AuditWriteMiddleware.InboundExecutionIdItemKey]!;
|
||||
var route = new RouteHelper(locator, router);
|
||||
var result = await scriptExecutor.ExecuteAsync(
|
||||
inboundMethod,
|
||||
new Dictionary<string, object?>(),
|
||||
route,
|
||||
TimeSpan.FromSeconds(30),
|
||||
ctx.RequestAborted,
|
||||
parentExecutionId: inboundExecutionId);
|
||||
|
||||
ctx.Response.StatusCode = result.Success ? 200 : 500;
|
||||
await ctx.Response.WriteAsync(result.Success ? "ok" : "fail");
|
||||
});
|
||||
|
||||
var client = host.GetTestClient();
|
||||
var response = await client.PostAsync(
|
||||
"/api/submitOrder",
|
||||
new StringContent("{}", Encoding.UTF8, "application/json"));
|
||||
Assert.Equal(System.Net.HttpStatusCode.OK, response.StatusCode);
|
||||
|
||||
// The routed run produced a NotifySend that buffered a NotificationSubmit
|
||||
// into S&F. Drain that genuine site-produced submission to the central
|
||||
// NotificationOutboxActor so the NotifyDeliver dispatch rows materialise.
|
||||
await ForwardBufferedNotificationToCentralAsync(
|
||||
storeAndForward, router.NotificationId!, centralProvider, centralAuditWriter);
|
||||
|
||||
// ── Assert ──────────────────────────────────────────────────────────
|
||||
await AwaitAssertAsync(async () =>
|
||||
{
|
||||
await using var readContext = CreateContext();
|
||||
var repo = new AuditLogRepository(readContext);
|
||||
|
||||
// Every audit row this site produced (sync ApiCall + cached lifecycle
|
||||
// + NotifySend) plus the central NotifyDeliver rows.
|
||||
var siteRows = await repo.QueryAsync(
|
||||
new AuditLogQueryFilter(SourceSiteIds: new[] { siteId }),
|
||||
new AuditLogPaging(PageSize: 100));
|
||||
|
||||
// sync ApiCall (1) + cached Submit/Attempted/Resolve (3) + NotifySend (1)
|
||||
// + NotifyDeliver Attempted/Delivered (2) = 7 rows for the routed run.
|
||||
Assert.True(siteRows.Count == 7,
|
||||
"Expected 7 routed-run audit rows; saw: "
|
||||
+ string.Join(", ", siteRows.Select(r => $"{r.Channel}/{r.Kind}/{r.Status}")));
|
||||
Assert.Single(siteRows, r => r.Channel == AuditChannel.ApiOutbound && r.Kind == AuditKind.ApiCall);
|
||||
Assert.Single(siteRows, r => r.Kind == AuditKind.CachedSubmit);
|
||||
Assert.Single(siteRows, r => r.Kind == AuditKind.CachedResolve);
|
||||
Assert.Single(siteRows, r => r.Kind == AuditKind.NotifySend);
|
||||
Assert.Equal(2, siteRows.Count(r => r.Kind == AuditKind.NotifyDeliver));
|
||||
|
||||
// CORE PROMISE: every routed-run row carries the SAME non-null
|
||||
// ParentExecutionId — the inbound request's ExecutionId.
|
||||
var parentIds = siteRows.Select(r => r.ParentExecutionId).Distinct().ToList();
|
||||
Assert.Single(parentIds);
|
||||
Assert.NotNull(parentIds[0]);
|
||||
var inboundExecutionId = parentIds[0]!.Value;
|
||||
|
||||
// The routed run has its OWN distinct ExecutionId — not the parent's.
|
||||
var routedExecutionIds = siteRows
|
||||
.Select(r => r.ExecutionId)
|
||||
.Distinct()
|
||||
.ToList();
|
||||
Assert.Single(routedExecutionIds);
|
||||
Assert.NotNull(routedExecutionIds[0]);
|
||||
var routedExecutionId = routedExecutionIds[0]!.Value;
|
||||
Assert.NotEqual(inboundExecutionId, routedExecutionId);
|
||||
|
||||
// The inbound request's own InboundRequest row is TOP-LEVEL —
|
||||
// ExecutionId = the propagated id, ParentExecutionId = NULL.
|
||||
var inboundRows = await repo.QueryAsync(
|
||||
new AuditLogQueryFilter(ExecutionId: inboundExecutionId),
|
||||
new AuditLogPaging(PageSize: 10));
|
||||
var inboundRow = Assert.Single(inboundRows,
|
||||
r => r.Channel == AuditChannel.ApiInbound && r.Kind == AuditKind.InboundRequest);
|
||||
Assert.Equal(AuditStatus.Delivered, inboundRow.Status);
|
||||
Assert.Null(inboundRow.ParentExecutionId);
|
||||
|
||||
// The parentExecutionId filter pulls the routed run's complete
|
||||
// trust-boundary footprint (all 7 routed rows, none of the inbound).
|
||||
var byParent = await repo.QueryAsync(
|
||||
new AuditLogQueryFilter(ParentExecutionId: inboundExecutionId),
|
||||
new AuditLogPaging(PageSize: 100));
|
||||
Assert.Equal(7, byParent.Count);
|
||||
Assert.All(byParent, r => Assert.Equal(routedExecutionId, r.ExecutionId));
|
||||
|
||||
// GetExecutionTreeAsync returns BOTH executions in one chain —
|
||||
// inbound (root) and routed (child), regardless of entry point.
|
||||
var treeFromChild = await repo.GetExecutionTreeAsync(routedExecutionId);
|
||||
AssertChain(treeFromChild, inboundExecutionId, routedExecutionId);
|
||||
var treeFromRoot = await repo.GetExecutionTreeAsync(inboundExecutionId);
|
||||
AssertChain(treeFromRoot, inboundExecutionId, routedExecutionId);
|
||||
}, TimeSpan.FromSeconds(30));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Asserts the execution tree is the expected two-node inbound→routed chain:
|
||||
/// the inbound execution is the root (<c>ParentExecutionId = NULL</c>) and the
|
||||
/// routed execution's <c>ParentExecutionId</c> points back at it.
|
||||
/// </summary>
|
||||
private static void AssertChain(
|
||||
IReadOnlyList<ExecutionTreeNode> tree,
|
||||
Guid inboundExecutionId,
|
||||
Guid routedExecutionId)
|
||||
{
|
||||
Assert.Equal(2, tree.Count);
|
||||
var root = Assert.Single(tree, n => n.ExecutionId == inboundExecutionId);
|
||||
Assert.Null(root.ParentExecutionId);
|
||||
var child = Assert.Single(tree, n => n.ExecutionId == routedExecutionId);
|
||||
Assert.Equal(inboundExecutionId, child.ParentExecutionId);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Spins up a minimal in-memory ASP.NET host whose pipeline mirrors the
|
||||
/// production inbound-API arrangement: routing → the real
|
||||
/// <see cref="AuditWriteMiddleware"/> → the <c>POST /api/{methodName}</c>
|
||||
/// endpoint. The middleware mints + stashes the inbound request's
|
||||
/// <c>ExecutionId</c> and emits the top-level <see cref="AuditKind.InboundRequest"/>
|
||||
/// row via the supplied <see cref="ICentralAuditWriter"/>.
|
||||
/// </summary>
|
||||
private static async Task<IHost> BuildInboundHostAsync(
|
||||
ICentralAuditWriter centralAuditWriter,
|
||||
RequestDelegate endpointHandler)
|
||||
{
|
||||
var hostBuilder = new HostBuilder()
|
||||
.ConfigureWebHost(webBuilder =>
|
||||
{
|
||||
webBuilder
|
||||
.UseTestServer()
|
||||
.ConfigureServices(services =>
|
||||
{
|
||||
services.AddSingleton(centralAuditWriter);
|
||||
services.AddRouting();
|
||||
})
|
||||
.Configure(app =>
|
||||
{
|
||||
app.UseRouting();
|
||||
app.UseAuditWriteMiddleware();
|
||||
app.UseEndpoints(endpoints =>
|
||||
{
|
||||
endpoints.MapPost("/api/{methodName}", endpointHandler);
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
return await hostBuilder.StartAsync();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Reads the genuine site-produced <see cref="NotificationSubmit"/> the routed
|
||||
/// <c>Notify.Send</c> buffered into Store-and-Forward, then drives it through
|
||||
/// a real central <see cref="NotificationOutboxActor"/> so the
|
||||
/// <see cref="AuditKind.NotifyDeliver"/> dispatch rows materialise. The
|
||||
/// dispatcher echoes <c>OriginParentExecutionId</c> off the
|
||||
/// <c>NotificationSubmit</c> onto every <c>NotifyDeliver</c> row — the
|
||||
/// cross-execution linkage under test on the central side.
|
||||
/// </summary>
|
||||
private async Task ForwardBufferedNotificationToCentralAsync(
|
||||
StoreAndForwardService storeAndForward,
|
||||
string notificationId,
|
||||
IServiceProvider centralProvider,
|
||||
ICentralAuditWriter centralAuditWriter)
|
||||
{
|
||||
var buffered = await storeAndForward.GetMessageByIdAsync(notificationId);
|
||||
Assert.NotNull(buffered);
|
||||
var submit = JsonSerializer.Deserialize<NotificationSubmit>(buffered!.PayloadJson);
|
||||
Assert.NotNull(submit);
|
||||
// The routed Notify.Send stamped the inbound request's ExecutionId as the
|
||||
// submission's OriginParentExecutionId — proven separately on the
|
||||
// NotifyDeliver rows, but asserted here too as the central handoff input.
|
||||
Assert.NotNull(submit!.OriginParentExecutionId);
|
||||
|
||||
// The outbox actor runs over the long-lived central provider (which
|
||||
// carries the AlwaysDeliversAdapter) so the dispatch sweep — launched
|
||||
// asynchronously by the DispatchTick — still has a live IServiceProvider
|
||||
// to resolve its per-sweep scope from.
|
||||
var outboxActor = Sys.ActorOf(Props.Create(() => new NotificationOutboxActor(
|
||||
centralProvider,
|
||||
new NotificationOutboxOptions
|
||||
{
|
||||
// Long timers so PreStart's scheduled ticks never fire — the
|
||||
// test drives ingest + dispatch explicitly.
|
||||
DispatchInterval = TimeSpan.FromHours(1),
|
||||
PurgeInterval = TimeSpan.FromDays(1),
|
||||
},
|
||||
centralAuditWriter,
|
||||
NullLogger<NotificationOutboxActor>.Instance)));
|
||||
|
||||
// Ingest the genuine site submission, then run one dispatch sweep.
|
||||
var ack = await outboxActor.Ask<NotificationSubmitAck>(
|
||||
submit, TimeSpan.FromSeconds(15));
|
||||
Assert.True(ack.Accepted, ack.Error);
|
||||
outboxActor.Tell(InternalMessages.DispatchTick.Instance);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Stub <see cref="INotificationDeliveryAdapter"/> that always reports a
|
||||
/// successful delivery — a single dispatch sweep then yields one
|
||||
/// <see cref="AuditStatus.Attempted"/> + one <see cref="AuditStatus.Delivered"/>
|
||||
/// <see cref="AuditKind.NotifyDeliver"/> row.
|
||||
/// </summary>
|
||||
private sealed class AlwaysDeliversAdapter : INotificationDeliveryAdapter
|
||||
{
|
||||
public NotificationType Type => NotificationType.Email;
|
||||
|
||||
public Task<DeliveryOutcome> DeliverAsync(
|
||||
ScadaLink.Commons.Entities.Notifications.Notification notification,
|
||||
CancellationToken cancellationToken = default)
|
||||
=> Task.FromResult(DeliveryOutcome.Success("ops@example.com"));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// In-process stand-in for the cross-cluster routing transport
|
||||
/// (<c>CommunicationServiceInstanceRouter</c> →
|
||||
/// <c>CommunicationService</c> → site <c>DeploymentManagerActor</c>). On a
|
||||
/// routed <c>Call</c> it does exactly what the production site handler does:
|
||||
/// it reads <see cref="RouteToCallRequest.ParentExecutionId"/> off the wire
|
||||
/// request and threads it into a fresh routed <see cref="ScriptRuntimeContext"/>
|
||||
/// as <c>parentExecutionId</c>, then runs the routed script's three
|
||||
/// trust-boundary actions (sync <c>ExternalSystem.Call</c>, a cached call and
|
||||
/// a <c>Notify.Send</c>). The routed context still mints its OWN fresh
|
||||
/// <c>ExecutionId</c> — only the parent pointer is inherited.
|
||||
/// </summary>
|
||||
private sealed class BridgingInstanceRouter : IInstanceRouter
|
||||
{
|
||||
private readonly string _siteId;
|
||||
private readonly IExternalSystemClient _externalClient;
|
||||
private readonly IAuditWriter _auditWriter;
|
||||
private readonly ICachedCallTelemetryForwarder _cachedForwarder;
|
||||
private readonly StoreAndForwardService _storeAndForward;
|
||||
|
||||
/// <summary>
|
||||
/// The <c>NotificationId</c> the routed <c>Notify.Send</c> minted, captured
|
||||
/// so the test can drain the buffered <see cref="NotificationSubmit"/>.
|
||||
/// </summary>
|
||||
public string? NotificationId { get; private set; }
|
||||
|
||||
public BridgingInstanceRouter(
|
||||
string siteId,
|
||||
IExternalSystemClient externalClient,
|
||||
IAuditWriter auditWriter,
|
||||
ICachedCallTelemetryForwarder cachedForwarder,
|
||||
StoreAndForwardService storeAndForward)
|
||||
{
|
||||
_siteId = siteId;
|
||||
_externalClient = externalClient;
|
||||
_auditWriter = auditWriter;
|
||||
_cachedForwarder = cachedForwarder;
|
||||
_storeAndForward = storeAndForward;
|
||||
}
|
||||
|
||||
public async Task<RouteToCallResponse> RouteToCallAsync(
|
||||
string siteId, RouteToCallRequest request, CancellationToken cancellationToken)
|
||||
{
|
||||
var compilationService = new ScriptCompilationService(
|
||||
NullLogger<ScriptCompilationService>.Instance);
|
||||
var sharedScriptLibrary = new SharedScriptLibrary(
|
||||
compilationService, NullLogger<SharedScriptLibrary>.Instance);
|
||||
|
||||
// Mirror DeploymentManagerActor → ScriptActor → ScriptExecutionActor:
|
||||
// the routed script execution gets its OWN fresh ExecutionId, and the
|
||||
// inbound request's ExecutionId arrives as ParentExecutionId.
|
||||
var routedContext = new ScriptRuntimeContext(
|
||||
ActorRefs.Nobody,
|
||||
ActorRefs.Nobody,
|
||||
sharedScriptLibrary,
|
||||
currentCallDepth: 0,
|
||||
maxCallDepth: 10,
|
||||
askTimeout: TimeSpan.FromSeconds(5),
|
||||
instanceName: request.InstanceUniqueName,
|
||||
logger: NullLogger.Instance,
|
||||
externalSystemClient: _externalClient,
|
||||
databaseGateway: null,
|
||||
storeAndForward: _storeAndForward,
|
||||
siteCommunicationActor: null,
|
||||
siteId: _siteId,
|
||||
sourceScript: $"ScriptActor:{request.ScriptName}",
|
||||
auditWriter: _auditWriter,
|
||||
operationTrackingStore: null,
|
||||
cachedForwarder: _cachedForwarder,
|
||||
executionId: null,
|
||||
parentExecutionId: request.ParentExecutionId);
|
||||
|
||||
// The routed site script's body: a sync ExternalSystem.Call, a cached
|
||||
// call, and a Notify.Send — three distinct trust-boundary actions of
|
||||
// the one routed execution.
|
||||
await routedContext.ExternalSystem.Call(ExternalSystemName, ExternalMethodName);
|
||||
await routedContext.ExternalSystem.CachedCall(ExternalSystemName, ExternalMethodName);
|
||||
NotificationId = await routedContext.Notify
|
||||
.To(NotifyListName)
|
||||
.Send("Routed run alert", "inbound-routed script fired");
|
||||
|
||||
return new RouteToCallResponse(
|
||||
request.CorrelationId, true, "routed-ok", null, DateTimeOffset.UtcNow);
|
||||
}
|
||||
|
||||
public Task<RouteToGetAttributesResponse> RouteToGetAttributesAsync(
|
||||
string siteId, RouteToGetAttributesRequest request, CancellationToken cancellationToken)
|
||||
=> throw new NotSupportedException();
|
||||
|
||||
public Task<RouteToSetAttributesResponse> RouteToSetAttributesAsync(
|
||||
string siteId, RouteToSetAttributesRequest request, CancellationToken cancellationToken)
|
||||
=> throw new NotSupportedException();
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user