feat(central-ui): FlowDebugService relays Retry/Discard for a parked outbound step

Claude-Session: https://claude.ai/code/session_01NcMG6xXdsJbmD2wYAgkK29
This commit is contained in:
Joseph Doherty
2026-09-06 11:04:03 -04:00
parent 9c31f70105
commit 2025ec346f
3 changed files with 363 additions and 3 deletions
@@ -3,10 +3,14 @@ using System.Text.Json;
using Akka.Actor;
using Microsoft.AspNetCore.Components.Authorization;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.ScadaBridge.CentralUI.Auth;
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Templates;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Audit;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Management;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Notification;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Debugging;
using ZB.MOM.WW.ScadaBridge.Communication;
using ZB.MOM.WW.ScadaBridge.ManagementService;
using ZB.MOM.WW.ScadaBridge.ScriptAnalysis;
using ZB.MOM.WW.ScadaBridge.Security;
@@ -22,6 +26,9 @@ namespace ZB.MOM.WW.ScadaBridge.CentralUI.Services;
/// so the server's Deployer gate runs against the real identity. The Live-run Ask waits up
/// to <see cref="RunAskTimeout"/> (the server bounds a run by the method timeout + grace and
/// a value-change by its wait window; both fit inside two minutes).
/// Retry / Discard of a parked outbound node (follow-ups plan 2026-09-06 D4) goes through
/// <see cref="CommunicationService"/> to the SAME central-local proxies the Site Calls and
/// Notification report pages use, so the page and the report never disagree on what a relay does.
/// </summary>
public sealed class FlowDebugService : IFlowDebugService
{
@@ -44,6 +51,7 @@ public sealed class FlowDebugService : IFlowDebugService
private readonly AuthenticationStateProvider _auth;
private readonly IInboundApiRepository _inboundApi;
private readonly ITemplateEngineRepository _templates;
private readonly CommunicationService _comms;
private readonly ILogger<FlowDebugService> _logger;
/// <summary>Initializes the service.</summary>
@@ -52,12 +60,14 @@ public sealed class FlowDebugService : IFlowDebugService
AuthenticationStateProvider auth,
IInboundApiRepository inboundApi,
ITemplateEngineRepository templates,
CommunicationService comms,
ILogger<FlowDebugService> logger)
{
_holder = holder ?? throw new ArgumentNullException(nameof(holder));
_auth = auth ?? throw new ArgumentNullException(nameof(auth));
_inboundApi = inboundApi ?? throw new ArgumentNullException(nameof(inboundApi));
_templates = templates ?? throw new ArgumentNullException(nameof(templates));
_comms = comms ?? throw new ArgumentNullException(nameof(comms));
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
}
@@ -177,6 +187,109 @@ public sealed class FlowDebugService : IFlowDebugService
public Task<FlowDebugOutcome> ReplayAsync(Guid executionId, CancellationToken cancellationToken = default)
=> SendAndBuildAsync(new FlowDebugReplayCommand(executionId), ReplayAskTimeout, cancellationToken);
/// <inheritdoc/>
public Task<FlowOutboundActionResult> RetryOutboundAsync(FlowStep step, CancellationToken cancellationToken = default)
=> RelayAsync(step, retry: true, cancellationToken);
/// <inheritdoc/>
public Task<FlowOutboundActionResult> DiscardOutboundAsync(FlowStep step, CancellationToken cancellationToken = default)
=> RelayAsync(step, retry: false, cancellationToken);
/// <summary>
/// The Retry / Discard relay behind <see cref="RetryOutboundAsync"/> and
/// <see cref="DiscardOutboundAsync"/>: guards first (nothing leaves the node for a step
/// that cannot be acted on), then the outbox or the site-call relay by step kind, with the
/// outcome mapped to the report pages' wording. A transport fault becomes a failed result,
/// never an exception on the page.
/// </summary>
private async Task<FlowOutboundActionResult> RelayAsync(FlowStep step, bool retry, CancellationToken ct)
{
ArgumentNullException.ThrowIfNull(step);
var verb = retry ? "Retry" : "Discard";
if (!FlowStepProjector.IsOutbound(step.Kind))
return new(false, "Only an API call, database call or notification node can be retried or discarded.");
if (step.Outcome != FlowStepOutcome.Parked)
return new(false, "This node is not parked — nothing to retry or discard.");
if (step.CorrelationId is not Guid c)
return new(false, "This call has no tracked operation id (a sync call is not store-and-forwarded), so there is nothing to retry or discard.");
var isNotify = step.Kind == FlowStepKind.Notify;
if (!isNotify && string.IsNullOrWhiteSpace(step.SiteId))
return new(false, "The owning site of this call is unknown.");
var shortId = c.ToString("N")[..12];
try
{
var requestedBy = await _auth.GetCurrentUsernameAsync();
var corr = Guid.NewGuid().ToString("N");
if (isNotify)
{
var notificationId = c.ToString("D");
if (retry)
{
var r = await _comms.RetryNotificationAsync(new RetryNotificationRequest(corr, notificationId, requestedBy), ct);
return new(r.Success, r.Success
? $"Notification {shortId} re-queued for delivery."
: r.ErrorMessage ?? "Retry failed.");
}
var d = await _comms.DiscardNotificationAsync(new DiscardNotificationRequest(corr, notificationId, requestedBy), ct);
return new(d.Success, d.Success
? $"Notification {shortId} discarded."
: d.ErrorMessage ?? "Discard failed.");
}
var site = step.SiteId!;
SiteCallRelayOutcome outcome;
bool siteReachable;
string? errorMessage;
if (retry)
{
var r = await _comms.RetrySiteCallAsync(new RetrySiteCallRequest(corr, c, site, requestedBy), ct);
(outcome, siteReachable, errorMessage) = (r.Outcome, r.SiteReachable, r.ErrorMessage);
}
else
{
var d = await _comms.DiscardSiteCallAsync(new DiscardSiteCallRequest(corr, c, site, requestedBy), ct);
(outcome, siteReachable, errorMessage) = (d.Outcome, d.SiteReachable, d.ErrorMessage);
}
return MapRelayOutcome(outcome, siteReachable, errorMessage, $"{verb} of {shortId} relayed to {site}.");
}
catch (OperationCanceledException)
{
throw;
}
catch (Exception ex)
{
_logger.LogWarning(ex, "Flow Debugger {Verb} relay failed for {Kind} {CorrelationId}", verb, step.Kind, c);
return new(false, $"{verb} failed: {ex.Message}");
}
}
/// <summary>
/// Maps a site-call relay outcome exactly as the Site Calls report page's
/// <c>ShowRelayOutcome</c> does: <see cref="SiteCallRelayOutcome.Applied"/> succeeds;
/// <see cref="SiteCallRelayOutcome.NotParked"/> is a definitive "nothing to do";
/// <see cref="SiteCallRelayOutcome.SiteUnreachable"/> — or an
/// <see cref="SiteCallRelayOutcome.OperationFailed"/> that also reports the site unreachable —
/// gets the unreachable wording; anything else the site's own message or a generic failure.
/// </summary>
private static FlowOutboundActionResult MapRelayOutcome(
SiteCallRelayOutcome outcome, bool siteReachable, string? errorMessage, string appliedMessage)
{
const string Unreachable = "Site unreachable — the relay did not reach the owning site. Try again once the site is back online.";
return outcome switch
{
SiteCallRelayOutcome.Applied => new(true, appliedMessage),
SiteCallRelayOutcome.NotParked => new(false, errorMessage ?? "The site reported nothing to do — the cached call is no longer parked."),
SiteCallRelayOutcome.SiteUnreachable => new(false, errorMessage ?? Unreachable),
SiteCallRelayOutcome.OperationFailed when !siteReachable => new(false, errorMessage ?? Unreachable),
_ => new(false, errorMessage ?? "The site could not apply the action."),
};
}
/// <summary>
/// Resolves <c>instanceCode.scriptName</c> to a script body: instance by unique name →
/// its template chain (nearest definition wins) → the named script's code.
@@ -23,6 +23,16 @@ public sealed record FlowDebugOutcome(FlowGraph? Graph, FlowDebugRunResult? Raw,
public static FlowDebugOutcome Fail(string error, string errorCode) => new(null, null, error, errorCode);
}
/// <summary>
/// Outcome of a Retry / Discard relayed from a Flow Debugger node (follow-ups plan 2026-09-06 D4).
/// <see cref="Message"/> is operator-facing and uses the same wording as the Site Calls /
/// Notification report pages, whether the relay was refused before it left the page or the
/// owning site / outbox answered.
/// </summary>
/// <param name="Success">True when the relay was applied (site <c>Applied</c>, or outbox success).</param>
/// <param name="Message">What happened, in the report pages' words.</param>
public sealed record FlowOutboundActionResult(bool Success, string Message);
/// <summary>
/// CentralUI facade over the Flow Debugger (design 2026-09-05 §5.5/§7.1): static shape
/// prediction from script syntax, and the three management commands (Live run of a method,
@@ -66,4 +76,24 @@ public interface IFlowDebugService
/// <param name="executionId">Any execution id in the chain.</param>
/// <param name="cancellationToken">Cancellation token.</param>
Task<FlowDebugOutcome> ReplayAsync(Guid executionId, CancellationToken cancellationToken = default);
/// <summary>
/// Relays a Retry of a parked outbound node to its owner — the Site Call Audit relay for an
/// <see cref="FlowStepKind.ApiCall"/> / <see cref="FlowStepKind.DbCall"/> (keyed by the step's
/// <see cref="FlowStep.CorrelationId"/> = <c>TrackedOperationId</c> and <see cref="FlowStep.SiteId"/>),
/// the Notification Outbox for a <see cref="FlowStepKind.Notify"/> (<c>CorrelationId</c> =
/// <c>NotificationId</c>). A step that is not outbound, not <see cref="FlowStepOutcome.Parked"/>,
/// has no correlation id, or (site kinds) no site id is refused with a message and never relayed.
/// </summary>
/// <param name="step">The outbound step to retry.</param>
/// <param name="cancellationToken">Cancellation token.</param>
Task<FlowOutboundActionResult> RetryOutboundAsync(FlowStep step, CancellationToken cancellationToken = default);
/// <summary>
/// Relays a Discard of a parked outbound node to its owner. Same routing and refusals as
/// <see cref="RetryOutboundAsync"/>.
/// </summary>
/// <param name="step">The outbound step to discard.</param>
/// <param name="cancellationToken">Cancellation token.</param>
Task<FlowOutboundActionResult> DiscardOutboundAsync(FlowStep step, CancellationToken cancellationToken = default);
}
@@ -3,6 +3,7 @@ using System.Text.Json;
using Akka.Actor;
using Microsoft.AspNetCore.Components.Authorization;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options;
using NSubstitute;
using ZB.MOM.WW.Audit;
using ZB.MOM.WW.ScadaBridge.CentralUI.Services;
@@ -10,10 +11,13 @@ using ZB.MOM.WW.ScadaBridge.Commons.Entities.InboundApi;
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Instances;
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Templates;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Audit;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Management;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Notification;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Debugging;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
using ZB.MOM.WW.ScadaBridge.Communication;
using ZB.MOM.WW.ScadaBridge.ManagementService;
using ZB.MOM.WW.ScadaBridge.ScriptAnalysis;
using ZB.MOM.WW.ScadaBridge.Security;
@@ -22,9 +26,17 @@ namespace ZB.MOM.WW.ScadaBridge.CentralUI.Tests.Services;
/// <summary>
/// Tests for <see cref="FlowDebugService"/> (Flow Debugger design 2026-09-05 §5.5/§7.1):
/// the management-command bridge (holder null, error mapping, success → graph) and the
/// prediction resolver wiring over substituted repositories.
/// the management-command bridge (holder null, error mapping, success → graph), the
/// prediction resolver wiring over substituted repositories, and the Retry/Discard relay
/// for a parked outbound node (follow-ups plan 2026-09-06 D4).
/// </summary>
/// <remarks>
/// <see cref="CommunicationService"/> is a concrete class with non-virtual methods, so the
/// relay tests wire its two central-local proxies (<c>SetSiteCallAudit</c> /
/// <c>SetNotificationOutbox</c>) to scripted <see cref="ReceiveActor"/>s on the test's
/// <see cref="ActorSystem"/> — the same seam <c>SiteCallsReportPageTests</c> and
/// <c>NotificationReportPageTests</c> use.
/// </remarks>
public sealed class FlowDebugServiceTests : IDisposable
{
private static readonly JsonSerializerOptions ServerJson = new()
@@ -38,6 +50,15 @@ public sealed class FlowDebugServiceTests : IDisposable
private readonly ITemplateEngineRepository _templates = Substitute.For<ITemplateEngineRepository>();
private readonly List<ManagementEnvelope> _received = new();
// Relay probes: everything the scripted Site Call Audit / Notification Outbox actors receive.
private readonly List<object> _siteCallRequests = new();
private readonly List<object> _outboxRequests = new();
// Scripted relay replies — overridable per test before the relay call.
private SiteCallRelayOutcome _siteCallOutcome = SiteCallRelayOutcome.Applied;
private bool _siteReachable = true;
private string? _siteCallError;
public void Dispose() => _system.Terminate().Wait(TimeSpan.FromSeconds(5));
private FlowDebugService CreateService(object? reply, string role = "Deployer")
@@ -52,7 +73,165 @@ public sealed class FlowDebugServiceTests : IDisposable
new Claim(JwtTokenService.RoleClaimType, role),
};
var principal = new ClaimsPrincipal(new ClaimsIdentity(claims, "TestAuth"));
return new FlowDebugService(holder, new StubAuth(principal), _inboundApi, _templates, NullLogger<FlowDebugService>.Instance);
var comms = new CommunicationService(
Options.Create(new CommunicationOptions()),
NullLogger<CommunicationService>.Instance);
comms.SetSiteCallAudit(_system.ActorOf(Props.Create(() => new ScriptedSiteCallAuditActor(this))));
comms.SetNotificationOutbox(_system.ActorOf(Props.Create(() => new ScriptedOutboxActor(this))));
return new FlowDebugService(holder, new StubAuth(principal), _inboundApi, _templates, comms, NullLogger<FlowDebugService>.Instance);
}
/// <summary>An External-lane step as <c>FlowStepProjector</c> would project it (positional per <see cref="FlowStep"/>).</summary>
private static FlowStep OutboundStep(FlowStepKind kind, FlowStepOutcome outcome, Guid? correlationId, string? siteId)
=> new(
Guid.NewGuid(), Guid.NewGuid(), null, kind, FlowLane.External,
"ERP.GetOrder", null, DateTimeOffset.UtcNow, null, outcome,
null, null, null, siteId, "node-a", "Sim01", "ERP.GetOrder", null,
Lifecycle: null, CorrelationId: correlationId);
[Fact]
public async Task RetryOutbound_ApiCallParked_RelaysRetrySiteCallWithCorrelationAndSite()
{
var c = Guid.NewGuid();
var svc = CreateService(reply: null);
var result = await svc.RetryOutboundAsync(OutboundStep(FlowStepKind.ApiCall, FlowStepOutcome.Parked, c, "site-a"));
Assert.True(result.Success, result.Message);
Assert.Contains("relayed", result.Message);
Assert.Contains("site-a", result.Message);
var request = Assert.IsType<RetrySiteCallRequest>(Assert.Single(_siteCallRequests));
Assert.Equal(c, request.TrackedOperationId);
Assert.Equal("site-a", request.SourceSite);
Assert.Equal("tester", request.RequestedBy);
Assert.Empty(_outboxRequests);
}
[Fact]
public async Task RetryOutbound_Notify_RelaysRetryNotification()
{
var c = Guid.NewGuid();
var svc = CreateService(reply: null);
var result = await svc.RetryOutboundAsync(OutboundStep(FlowStepKind.Notify, FlowStepOutcome.Parked, c, siteId: null));
Assert.True(result.Success, result.Message);
Assert.Contains("re-queued", result.Message);
var request = Assert.IsType<RetryNotificationRequest>(Assert.Single(_outboxRequests));
Assert.Equal(c.ToString("D"), request.NotificationId);
Assert.Equal("tester", request.RequestedBy);
Assert.Empty(_siteCallRequests);
}
[Fact]
public async Task DiscardOutbound_DbCall_RelaysDiscardSiteCall()
{
var c = Guid.NewGuid();
var svc = CreateService(reply: null);
var result = await svc.DiscardOutboundAsync(OutboundStep(FlowStepKind.DbCall, FlowStepOutcome.Parked, c, "site-b"));
Assert.True(result.Success, result.Message);
Assert.Contains("Discard of", result.Message);
Assert.Contains("site-b", result.Message);
var request = Assert.IsType<DiscardSiteCallRequest>(Assert.Single(_siteCallRequests));
Assert.Equal(c, request.TrackedOperationId);
Assert.Equal("site-b", request.SourceSite);
Assert.Equal("tester", request.RequestedBy);
}
[Fact]
public async Task RetryOutbound_NotParked_RefusesWithoutRelay()
{
var svc = CreateService(reply: null);
var result = await svc.RetryOutboundAsync(OutboundStep(FlowStepKind.ApiCall, FlowStepOutcome.Ok, Guid.NewGuid(), "site-a"));
Assert.False(result.Success);
Assert.Contains("not parked", result.Message);
Assert.Empty(_siteCallRequests);
Assert.Empty(_outboxRequests);
}
[Fact]
public async Task RetryOutbound_NoCorrelation_Refuses()
{
var svc = CreateService(reply: null);
var result = await svc.RetryOutboundAsync(OutboundStep(FlowStepKind.ApiCall, FlowStepOutcome.Parked, correlationId: null, "site-a"));
Assert.False(result.Success);
Assert.Contains("no tracked operation id", result.Message);
Assert.Empty(_siteCallRequests);
Assert.Empty(_outboxRequests);
}
[Fact]
public async Task RetryOutbound_SiteKindWithoutSiteId_Refuses()
{
var svc = CreateService(reply: null);
var result = await svc.RetryOutboundAsync(OutboundStep(FlowStepKind.DbCall, FlowStepOutcome.Parked, Guid.NewGuid(), siteId: " "));
Assert.False(result.Success);
Assert.Contains("owning site", result.Message);
Assert.Empty(_siteCallRequests);
Assert.Empty(_outboxRequests);
}
[Fact]
public async Task RetryOutbound_NonOutboundStep_Refuses()
{
var svc = CreateService(reply: null);
var result = await svc.RetryOutboundAsync(OutboundStep(FlowStepKind.ScriptRun, FlowStepOutcome.Parked, Guid.NewGuid(), "site-a"));
Assert.False(result.Success);
Assert.Contains("Only an API call", result.Message);
Assert.Empty(_siteCallRequests);
Assert.Empty(_outboxRequests);
}
[Fact]
public async Task RetryOutbound_SiteUnreachable_MapsToFailureMessage()
{
_siteCallOutcome = SiteCallRelayOutcome.SiteUnreachable;
_siteReachable = false;
var svc = CreateService(reply: null);
var result = await svc.RetryOutboundAsync(OutboundStep(FlowStepKind.ApiCall, FlowStepOutcome.Parked, Guid.NewGuid(), "site-a"));
Assert.False(result.Success);
Assert.Contains("unreachable", result.Message);
Assert.Single(_siteCallRequests);
}
[Fact]
public async Task RetryOutbound_NotParkedAtSite_MapsToFailureMessage()
{
_siteCallOutcome = SiteCallRelayOutcome.NotParked;
var svc = CreateService(reply: null);
var result = await svc.RetryOutboundAsync(OutboundStep(FlowStepKind.ApiCall, FlowStepOutcome.Parked, Guid.NewGuid(), "site-a"));
Assert.False(result.Success);
Assert.Contains("no longer parked", result.Message);
}
[Fact]
public async Task DiscardOutbound_OperationFailed_UsesTheSitesOwnErrorMessage()
{
_siteCallOutcome = SiteCallRelayOutcome.OperationFailed;
_siteCallError = "S&F buffer is locked by a sweep.";
var svc = CreateService(reply: null);
var result = await svc.DiscardOutboundAsync(OutboundStep(FlowStepKind.ApiCall, FlowStepOutcome.Parked, Guid.NewGuid(), "site-a"));
Assert.False(result.Success);
Assert.Equal("S&F buffer is locked by a sweep.", result.Message);
Assert.IsType<DiscardSiteCallRequest>(Assert.Single(_siteCallRequests));
}
private static string SuccessPayload(int events, bool completed = true, bool truncated = false)
@@ -384,4 +563,42 @@ public sealed class FlowDebugServiceTests : IDisposable
});
}
}
/// <summary>Stands in for the Site Call Audit singleton proxy: records the relay request, replies as scripted.</summary>
private sealed class ScriptedSiteCallAuditActor : ReceiveActor
{
public ScriptedSiteCallAuditActor(FlowDebugServiceTests test)
{
Receive<RetrySiteCallRequest>(r =>
{
lock (test._siteCallRequests) test._siteCallRequests.Add(r);
Sender.Tell(new RetrySiteCallResponse(r.CorrelationId, test._siteCallOutcome,
test._siteCallOutcome == SiteCallRelayOutcome.Applied, test._siteReachable, test._siteCallError));
});
Receive<DiscardSiteCallRequest>(r =>
{
lock (test._siteCallRequests) test._siteCallRequests.Add(r);
Sender.Tell(new DiscardSiteCallResponse(r.CorrelationId, test._siteCallOutcome,
test._siteCallOutcome == SiteCallRelayOutcome.Applied, test._siteReachable, test._siteCallError));
});
}
}
/// <summary>Stands in for the Notification Outbox singleton proxy: records the request, replies success.</summary>
private sealed class ScriptedOutboxActor : ReceiveActor
{
public ScriptedOutboxActor(FlowDebugServiceTests test)
{
Receive<RetryNotificationRequest>(r =>
{
lock (test._outboxRequests) test._outboxRequests.Add(r);
Sender.Tell(new RetryNotificationResponse(r.CorrelationId, true, null));
});
Receive<DiscardNotificationRequest>(r =>
{
lock (test._outboxRequests) test._outboxRequests.Add(r);
Sender.Tell(new DiscardNotificationResponse(r.CorrelationId, true, null));
});
}
}
}