Joseph Doherty
424 lines
22 KiB
C#
424 lines
22 KiB
C#
using Akka.Actor;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Messages.InboundApi;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Types;
|
|
|
|
namespace ZB.MOM.WW.ScadaBridge.InboundAPI;
|
|
|
|
/// <summary>
|
|
/// C3/N3: a routed call failed because the target site never answered — the routed
|
|
/// Ask <b>expired</b>. Unreachability is typed at the Ask boundary from
|
|
/// <see cref="AskTimeoutException"/> (the communication layer's own signal: an
|
|
/// unreachable/contactless site has its enveloped message dropped so the Ask times
|
|
/// out caller-side) — it is <em>never</em> inferred from an error-message substring.
|
|
/// Derives from <see cref="InvalidOperationException"/> so existing callers that
|
|
/// catch the base type are unaffected, while <see cref="InboundScriptExecutor"/>
|
|
/// catches this more specific type to surface the spec's <c>SITE_UNREACHABLE</c>
|
|
/// error code instead of a generic <c>SCRIPT_ERROR</c>. A <c>Success=false</c>
|
|
/// <em>response</em> means the site answered (reachable by definition) and stays a
|
|
/// plain <see cref="InvalidOperationException"/>.
|
|
/// </summary>
|
|
public sealed class SiteUnreachableException : InvalidOperationException
|
|
{
|
|
/// <summary>Initializes a new <see cref="SiteUnreachableException"/> with the given message.</summary>
|
|
/// <param name="message">The routing-level failure message that indicated the site was unreachable.</param>
|
|
public SiteUnreachableException(string message) : base(message) { }
|
|
|
|
/// <summary>Initializes a new <see cref="SiteUnreachableException"/> wrapping the underlying Ask-timeout.</summary>
|
|
/// <param name="message">The routing-level failure message.</param>
|
|
/// <param name="innerException">The underlying <see cref="AskTimeoutException"/> from the routed Ask.</param>
|
|
public SiteUnreachableException(string message, Exception innerException) : base(message, innerException) { }
|
|
}
|
|
|
|
/// <summary>
|
|
/// Route.To() helper for cross-site calls from inbound API scripts.
|
|
/// Resolves instance to site, routes via <see cref="IInstanceRouter"/>, blocks until
|
|
/// response or timeout. Site unreachable returns error (no store-and-forward).
|
|
///
|
|
/// The helper carries the executing method's <see cref="CancellationToken"/>
|
|
/// (the method-level timeout). Routed calls inherit that deadline by default, so a
|
|
/// natural script — <c>Route.To("inst").Call("doWork", p)</c> — is timeout-bounded
|
|
/// without the script having to thread a token explicitly.
|
|
/// </summary>
|
|
public class RouteHelper
|
|
{
|
|
private readonly IInstanceLocator _instanceLocator;
|
|
private readonly IInstanceRouter _instanceRouter;
|
|
private readonly CancellationToken _deadlineToken;
|
|
private readonly CancellationToken _requestAbortedToken;
|
|
private readonly Guid? _parentExecutionId;
|
|
|
|
/// <summary>
|
|
/// Initializes a new <see cref="RouteHelper"/> with no deadline token and no parent execution id.
|
|
/// </summary>
|
|
/// <param name="instanceLocator">Service to resolve the site id for a given instance code.</param>
|
|
/// <param name="instanceRouter">Service to route cross-site calls to the resolved site.</param>
|
|
public RouteHelper(
|
|
IInstanceLocator instanceLocator,
|
|
IInstanceRouter instanceRouter)
|
|
: this(instanceLocator, instanceRouter, CancellationToken.None, CancellationToken.None, parentExecutionId: null)
|
|
{
|
|
}
|
|
|
|
private RouteHelper(
|
|
IInstanceLocator instanceLocator,
|
|
IInstanceRouter instanceRouter,
|
|
CancellationToken deadlineToken,
|
|
CancellationToken requestAbortedToken,
|
|
Guid? parentExecutionId)
|
|
{
|
|
_instanceLocator = instanceLocator;
|
|
_instanceRouter = instanceRouter;
|
|
_deadlineToken = deadlineToken;
|
|
_requestAbortedToken = requestAbortedToken;
|
|
_parentExecutionId = parentExecutionId;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Returns a <see cref="RouteHelper"/> whose routed calls inherit
|
|
/// <paramref name="deadlineToken"/> (the executing method's timeout) by default.
|
|
/// <see cref="InboundScriptExecutor"/> calls this when it builds the script
|
|
/// context so the method timeout actually covers routed calls, as the design doc
|
|
/// requires.
|
|
/// </summary>
|
|
/// <param name="deadlineToken">The executing method's timeout cancellation token to inherit for routed calls.</param>
|
|
/// <returns>A new <see cref="RouteHelper"/> inheriting the given deadline token.</returns>
|
|
public RouteHelper WithDeadline(CancellationToken deadlineToken) =>
|
|
new(_instanceLocator, _instanceRouter, deadlineToken, _requestAbortedToken, _parentExecutionId);
|
|
|
|
/// <summary>
|
|
/// Returns a <see cref="RouteHelper"/> carrying the raw request-abort
|
|
/// token (a client disconnect) <em>separately</em> from the method deadline. Most
|
|
/// routed calls remain bounded by the method deadline (which already incorporates the
|
|
/// abort), but <see cref="RouteTarget.WaitForAttribute"/> uses this token so its wait
|
|
/// is bounded by the WAIT timeout — not the generic method deadline (spec §6) — while
|
|
/// still being cancelled by a client disconnect.
|
|
/// </summary>
|
|
/// <param name="requestAbortedToken">The raw client-disconnect token, independent of the method timeout.</param>
|
|
/// <returns>A new <see cref="RouteHelper"/> carrying the given request-abort token.</returns>
|
|
public RouteHelper WithRequestAborted(CancellationToken requestAbortedToken) =>
|
|
new(_instanceLocator, _instanceRouter, _deadlineToken, requestAbortedToken, _parentExecutionId);
|
|
|
|
/// <summary>
|
|
/// Audit Log (ParentExecutionId): returns a <see cref="RouteHelper"/> whose
|
|
/// routed <see cref="RouteTarget.Call"/> requests carry
|
|
/// <paramref name="parentExecutionId"/> as <see cref="RouteToCallRequest.ParentExecutionId"/>.
|
|
/// For an inbound API request this is the inbound request's own per-request
|
|
/// execution id, so the routed site script records the inbound request as its
|
|
/// parent. <see cref="InboundScriptExecutor"/> calls this when it builds the
|
|
/// script context.
|
|
/// </summary>
|
|
/// <param name="parentExecutionId">The inbound request's execution id to stamp as the spawning parent on routed calls, or null for non-routed runs.</param>
|
|
/// <returns>A new <see cref="RouteHelper"/> carrying the given parent execution id.</returns>
|
|
public RouteHelper WithParentExecutionId(Guid? parentExecutionId) =>
|
|
new(_instanceLocator, _instanceRouter, _deadlineToken, _requestAbortedToken, parentExecutionId);
|
|
|
|
/// <summary>
|
|
/// Creates a route target for the specified instance.
|
|
/// </summary>
|
|
/// <param name="instanceCode">The unique code of the instance to route calls to.</param>
|
|
/// <returns>A <see cref="RouteTarget"/> bound to the specified instance.</returns>
|
|
public RouteTarget To(string instanceCode)
|
|
{
|
|
return new RouteTarget(
|
|
instanceCode, _instanceLocator, _instanceRouter, _deadlineToken, _requestAbortedToken, _parentExecutionId);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Represents a route target (an instance) for cross-site calls.
|
|
/// </summary>
|
|
public class RouteTarget
|
|
{
|
|
private readonly string _instanceCode;
|
|
private readonly IInstanceLocator _instanceLocator;
|
|
private readonly IInstanceRouter _instanceRouter;
|
|
private readonly CancellationToken _deadlineToken;
|
|
private readonly CancellationToken _requestAbortedToken;
|
|
private readonly Guid? _parentExecutionId;
|
|
|
|
/// <summary>
|
|
/// Initializes a new <see cref="RouteTarget"/> for the specified instance.
|
|
/// </summary>
|
|
/// <param name="instanceCode">The unique code of the target instance.</param>
|
|
/// <param name="instanceLocator">Service to resolve the site id for the instance.</param>
|
|
/// <param name="instanceRouter">Service to route cross-site calls.</param>
|
|
/// <param name="deadlineToken">Cancellation token representing the method-level deadline.</param>
|
|
/// <param name="requestAbortedToken">Raw client-disconnect token, independent of the method timeout.</param>
|
|
/// <param name="parentExecutionId">Optional parent execution id for audit correlation on routed calls.</param>
|
|
internal RouteTarget(
|
|
string instanceCode,
|
|
IInstanceLocator instanceLocator,
|
|
IInstanceRouter instanceRouter,
|
|
CancellationToken deadlineToken,
|
|
CancellationToken requestAbortedToken,
|
|
Guid? parentExecutionId)
|
|
{
|
|
_instanceCode = instanceCode;
|
|
_instanceLocator = instanceLocator;
|
|
_instanceRouter = instanceRouter;
|
|
_deadlineToken = deadlineToken;
|
|
_requestAbortedToken = requestAbortedToken;
|
|
_parentExecutionId = parentExecutionId;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Calls a script on the remote instance. Synchronous from API caller's
|
|
/// perspective. <paramref name="parameters"/> may be a dictionary or an
|
|
/// anonymous object (<c>new { name = "Bob" }</c>) — see <see cref="ScriptArgs"/>.
|
|
///
|
|
/// When <paramref name="cancellationToken"/> is not supplied the
|
|
/// routed call inherits the executing method's timeout, so the call is bounded by
|
|
/// the method-level deadline with no token argument.
|
|
/// </summary>
|
|
/// <param name="scriptName">Name of the script to call on the remote instance.</param>
|
|
/// <param name="parameters">Optional parameters passed to the script; may be a dictionary or anonymous object.</param>
|
|
/// <param name="cancellationToken">Optional cancellation token; defaults to the method deadline when not supplied.</param>
|
|
/// <returns>A task that resolves to the remote script's return value, or null.</returns>
|
|
public async Task<object?> Call(
|
|
string scriptName,
|
|
object? parameters = null,
|
|
CancellationToken cancellationToken = default)
|
|
{
|
|
var token = Effective(cancellationToken);
|
|
var siteId = await ResolveSiteAsync(token);
|
|
var correlationId = Guid.NewGuid().ToString();
|
|
|
|
// Audit Log (ParentExecutionId): stamp the spawning execution's id
|
|
// (the inbound API request's ExecutionId) so the routed site script
|
|
// records this call's parent. CorrelationId above is a separate concern
|
|
// — the per-operation lifecycle id, freshly minted per routed call.
|
|
var request = new RouteToCallRequest(
|
|
correlationId, _instanceCode, scriptName, ScriptArgs.Normalize(parameters),
|
|
DateTimeOffset.UtcNow, _parentExecutionId);
|
|
|
|
var response = await RouteOrUnreachable(_instanceRouter.RouteToCallAsync(siteId, request, token));
|
|
|
|
if (!response.Success)
|
|
{
|
|
throw RoutingFailure(
|
|
response.ErrorMessage ?? "Remote script call failed");
|
|
}
|
|
|
|
return response.ReturnValue;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Gets a single attribute value from the remote instance.
|
|
/// </summary>
|
|
/// <param name="attributeName">Name of the attribute to read.</param>
|
|
/// <param name="cancellationToken">Optional cancellation token; defaults to the method deadline.</param>
|
|
/// <returns>A task that resolves to the attribute value, or null if the key is absent.</returns>
|
|
public async Task<object?> GetAttribute(
|
|
string attributeName,
|
|
CancellationToken cancellationToken = default)
|
|
{
|
|
var result = await GetAttributes(new[] { attributeName }, cancellationToken);
|
|
return result.TryGetValue(attributeName, out var value) ? value : null;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Gets multiple attribute values from the remote instance (batch read).
|
|
/// </summary>
|
|
/// <param name="attributeNames">Names of the attributes to read.</param>
|
|
/// <param name="cancellationToken">Optional cancellation token; defaults to the method deadline.</param>
|
|
/// <returns>A task that resolves to a map of attribute names to their current values.</returns>
|
|
public async Task<IReadOnlyDictionary<string, object?>> GetAttributes(
|
|
IEnumerable<string> attributeNames,
|
|
CancellationToken cancellationToken = default)
|
|
{
|
|
var token = Effective(cancellationToken);
|
|
var siteId = await ResolveSiteAsync(token);
|
|
var correlationId = Guid.NewGuid().ToString();
|
|
|
|
// Audit Log (ParentExecutionId): mirrors the Call path — stamp the
|
|
// spawning inbound request's ExecutionId so future site-side audit
|
|
// emission for routed reads can record this read's parent. Symmetric
|
|
// with RouteToCallRequest so script authors get the same correlation
|
|
// across Call / GetAttributes / SetAttributes.
|
|
var request = new RouteToGetAttributesRequest(
|
|
correlationId, _instanceCode, attributeNames.ToList(), DateTimeOffset.UtcNow,
|
|
_parentExecutionId);
|
|
|
|
var response = await RouteOrUnreachable(_instanceRouter.RouteToGetAttributesAsync(siteId, request, token));
|
|
|
|
if (!response.Success)
|
|
{
|
|
throw RoutingFailure(
|
|
response.ErrorMessage ?? "Remote attribute read failed");
|
|
}
|
|
|
|
return response.Values;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Blocks until a remote instance attribute reaches <paramref name="targetValue"/>
|
|
/// or <paramref name="timeout"/> elapses (spec §6). Value-equality ONLY across the
|
|
/// wire: the target is canonically encoded via <see cref="AttributeValueCodec"/> and
|
|
/// the site evaluates equality — there is no predicate and no quality flag in the
|
|
/// comparison.
|
|
///
|
|
/// <para>
|
|
/// Unlike the other routed calls (which inherit the method deadline
|
|
/// by default), the wait is bounded by <paramref name="timeout"/> — the WAIT
|
|
/// timeout — NOT the generic method deadline. The SITE enforces <paramref name="timeout"/>
|
|
/// and returns <c>Matched=false</c> when it elapses; the local token is built from a
|
|
/// per-wait CTS (the wait timeout plus a small grace), an explicit caller token, and
|
|
/// the client-disconnect token, but deliberately EXCLUDES the method deadline. So a
|
|
/// wait longer than the method timeout runs to its full wait timeout, as spec §6
|
|
/// requires, while a client disconnect still cancels it.
|
|
/// </para>
|
|
/// </summary>
|
|
/// <param name="attributeName">Name of the attribute to wait on.</param>
|
|
/// <param name="targetValue">Target value the attribute must equal for the wait to match.</param>
|
|
/// <param name="timeout">Maximum time to wait for the attribute to reach the target value; this — not the method deadline — bounds the wait.</param>
|
|
/// <param name="cancellationToken">Optional explicit cancellation token (a tighter caller-supplied bound); the wait is otherwise bounded by <paramref name="timeout"/>.</param>
|
|
/// <returns>A task that resolves to <c>true</c> if the attribute reached the target value, <c>false</c> if the wait timed out.</returns>
|
|
public async Task<bool> WaitForAttribute(
|
|
string attributeName,
|
|
object? targetValue,
|
|
TimeSpan timeout,
|
|
CancellationToken cancellationToken = default)
|
|
{
|
|
// Do NOT impose a local wait-timeout backstop here. The site
|
|
// enforces the wait `timeout` and returns Matched=false when it elapses, and the
|
|
// cluster Ask in CommunicationService.RouteToWaitForAttributeAsync already bounds
|
|
// the round trip by `timeout + IntegrationTimeout` — the authoritative backstop
|
|
// for a missing site response. A local CTS of `timeout + small grace` (the prior
|
|
// approach) was TIGHTER than that round-trip budget, so a
|
|
// slow-but-valid timed-out response could be cancelled into an exception instead
|
|
// of the spec-mandated `false`. Link ONLY the client-disconnect token and an
|
|
// explicit caller token — NOT the method deadline — so a client abort still
|
|
// cancels the wait while the wait timeout itself governs the duration.
|
|
using var linked = CancellationTokenSource.CreateLinkedTokenSource(
|
|
_requestAbortedToken, cancellationToken);
|
|
var token = linked.Token;
|
|
var siteId = await ResolveSiteAsync(token);
|
|
|
|
// Audit Log (ParentExecutionId): mirrors the Call path — stamp the
|
|
// spawning inbound request's ExecutionId so future site-side audit
|
|
// emission for routed waits can record this wait's parent. CorrelationId
|
|
// is the per-operation lifecycle id, freshly minted per routed wait.
|
|
var request = new RouteToWaitForAttributeRequest(
|
|
Guid.NewGuid().ToString(), _instanceCode, attributeName,
|
|
AttributeValueCodec.Encode(targetValue), timeout, DateTimeOffset.UtcNow,
|
|
_parentExecutionId);
|
|
|
|
var response = await RouteOrUnreachable(_instanceRouter.RouteToWaitForAttributeAsync(siteId, request, token));
|
|
|
|
if (!response.Success)
|
|
{
|
|
throw RoutingFailure(
|
|
response.ErrorMessage ?? "Remote attribute wait failed");
|
|
}
|
|
|
|
return response.Matched;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Sets a single attribute value on the remote instance.
|
|
/// </summary>
|
|
/// <param name="attributeName">Name of the attribute to write.</param>
|
|
/// <param name="value">Value to set on the attribute.</param>
|
|
/// <param name="cancellationToken">Optional cancellation token; defaults to the method deadline.</param>
|
|
/// <returns>A task that represents the asynchronous operation.</returns>
|
|
public async Task SetAttribute(
|
|
string attributeName,
|
|
string value,
|
|
CancellationToken cancellationToken = default)
|
|
{
|
|
await SetAttributes(
|
|
new Dictionary<string, string> { { attributeName, value } },
|
|
cancellationToken);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Sets multiple attribute values on the remote instance (batch write).
|
|
/// </summary>
|
|
/// <param name="attributeValues">Map of attribute names to values to write.</param>
|
|
/// <param name="cancellationToken">Optional cancellation token; defaults to the method deadline.</param>
|
|
/// <returns>A task that represents the asynchronous operation.</returns>
|
|
public async Task SetAttributes(
|
|
IReadOnlyDictionary<string, string> attributeValues,
|
|
CancellationToken cancellationToken = default)
|
|
{
|
|
var token = Effective(cancellationToken);
|
|
var siteId = await ResolveSiteAsync(token);
|
|
var correlationId = Guid.NewGuid().ToString();
|
|
|
|
// Audit Log (ParentExecutionId): mirrors the Call path — stamp the
|
|
// spawning inbound request's ExecutionId so future site-side audit
|
|
// emission for routed writes can record this write's parent. Symmetric
|
|
// with RouteToCallRequest so script authors get the same correlation
|
|
// across Call / GetAttributes / SetAttributes.
|
|
var request = new RouteToSetAttributesRequest(
|
|
correlationId, _instanceCode, attributeValues, DateTimeOffset.UtcNow,
|
|
_parentExecutionId);
|
|
|
|
var response = await RouteOrUnreachable(_instanceRouter.RouteToSetAttributesAsync(siteId, request, token));
|
|
|
|
if (!response.Success)
|
|
{
|
|
throw RoutingFailure(
|
|
response.ErrorMessage ?? "Remote attribute write failed");
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// A routed call with no explicit token inherits the executing
|
|
/// method's deadline. An explicitly supplied token (for a tighter bound) wins.
|
|
/// </summary>
|
|
private CancellationToken Effective(CancellationToken explicitToken) =>
|
|
explicitToken.CanBeCanceled ? explicitToken : _deadlineToken;
|
|
|
|
private async Task<string> ResolveSiteAsync(CancellationToken cancellationToken)
|
|
{
|
|
var siteId = await _instanceLocator.GetSiteIdForInstanceAsync(_instanceCode, cancellationToken);
|
|
if (siteId == null)
|
|
{
|
|
// A null site id is a plain not-found / no-assigned-site — never
|
|
// unreachability (nothing was routed), so it stays a generic
|
|
// InvalidOperationException (SCRIPT_ERROR).
|
|
throw RoutingFailure(
|
|
$"Instance '{_instanceCode}' not found or has no assigned site");
|
|
}
|
|
|
|
return siteId;
|
|
}
|
|
|
|
/// <summary>
|
|
/// N3: awaits a routed Ask and TYPES unreachability. AskTimeoutException means
|
|
/// the site never answered (no ClusterClient contact, site down, partition) —
|
|
/// the communication layer's own typed signal (CentralCommunicationActor drops
|
|
/// envelopes for contactless sites so the Ask expires caller-side) — and
|
|
/// surfaces as SiteUnreachableException so the executor maps it to the spec's
|
|
/// SITE_UNREACHABLE code. A Success=false RESPONSE means the site answered and
|
|
/// is therefore reachable; it flows through RoutingFailure as a plain
|
|
/// InvalidOperationException (SCRIPT_ERROR).
|
|
/// </summary>
|
|
private static async Task<TResponse> RouteOrUnreachable<TResponse>(Task<TResponse> routedAsk)
|
|
{
|
|
try
|
|
{
|
|
return await routedAsk;
|
|
}
|
|
catch (AskTimeoutException ex)
|
|
{
|
|
throw new SiteUnreachableException(
|
|
"Site did not respond to the routed request within the routing timeout — unreachable or not connected.",
|
|
ex);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// C3/N3: a routing-level failure REPORTED BY the site or the locator. The
|
|
/// remote end answered (or the instance simply has no site), so this is never
|
|
/// unreachability — always a plain <see cref="InvalidOperationException"/>
|
|
/// (SCRIPT_ERROR at the endpoint). Unreachability is typed at the Ask boundary
|
|
/// by <c>RouteOrUnreachable</c> (AskTimeoutException → SiteUnreachableException),
|
|
/// replacing the old message-substring sniff that a harmless rewording in the
|
|
/// communication layer could silently defeat.
|
|
/// </summary>
|
|
private static InvalidOperationException RoutingFailure(string message) => new(message);
|
|
}
|