diff --git a/CLAUDE.md b/CLAUDE.md index 86f77d67..13ba7a6a 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -117,6 +117,7 @@ spec for each is `docs/requirements/Component-.md`, and `README.md` carrie - `Notify.Send` is async and **enqueue-only** — it buffers the notification into the local SQLite S&F store and returns a `NotificationId` (GUID, idempotency key) status handle immediately; it never runs the forwarder's central Ask inline on the script thread (`deferToSweep: true` buffers due-immediately + kicks a background sweep), so its worst-case latency is the local insert whether central is up or down. It enqueues with `maxRetries: 0` (the "no limit" escape hatch), so notifications retry until central acks and are **never parked for retry exhaustion** — only a corrupt payload parks them (arch-review 02, Tasks 13/14). `Notify.Status(notificationId)` returns a status record (status, retry count, last error, key timestamps); answered site-locally as `Forwarding` while still in the site S&F buffer, otherwise round-trips to central. - SMS delivery adapter (T9/T10, 2026-06-19): `SmsNotificationDeliveryAdapter` — Twilio REST, no SDK, HTTP Basic auth (`AccountSid:AuthToken`), one `POST .../Messages.json` per recipient. `SmsConfiguration` entity (`AccountSid`, `AuthToken` encrypted via Data Protection, `FromNumber`/`MessagingServiceSid`, `ApiBaseUrl`, timeout/retry) stored centrally; managed Admin-only via CLI `notification sms list|update` + Central UI `/notifications/sms`. Central-only (never deployed to sites). Per-recipient rollup: all-accepted → Delivered; any-transient → retry/park; mix → delivered-to-good + note; all-permanent → Park. `SmsConfiguration` (Auth Token in SecretsBlock) travels in Transport bundles via `--sms-configs`. - Inbound API: `POST /api/{methodName}`, `X-API-Key` header, flat JSON, extended type system (Object, List). +- **Inbound API on a STANDBY central node RELAYS to the active peer (2026-08-28)** instead of answering 503 `STANDBY_NODE`. Method scripts/`Route.To()` still run only on the active node (`IActiveNodeGate`); the standby's `InboundApiEndpointFilter` hands the request to `IStandbyRequestForwarder` (`HttpStandbyRequestForwarder`, plain HTTP, verbatim method/path/query/body/headers, peer response replayed unchanged) when `ScadaBridge:InboundApi:StandbyForwardPeerEndpoints` lists the *other* node's HTTP base URL — a static appsettings list, because the Akka member address carries the remoting port, not the HTTP one. Empty list = old 503 behaviour. Peers are tried in order; a peer 503 stamped `X-ScadaBridge-Standby: 1`, a connection failure or a `StandbyForwardTimeout` (60 s, must exceed the longest method timeout) expiry each move to the next, all failing → 503 `ACTIVE_NODE_UNREACHABLE`. **One-hop loop guard:** relayed requests carry `X-ScadaBridge-Forwarded` and are never relayed again. The 413 body cap runs *before* the relay, and the standby's `AuditWriteMiddleware` skips its own row for a relayed call (`InboundApiEndpointFilter.RelayedToActiveItemKey`) so it is audited once, on the node that executed it. Both rigs (`docker/`, `docker-env2/`) configure it on both central nodes; QA/prod appsettings are user-held and need the key added by hand. Spec: `Component-InboundAPI.md` → *Standby forwarding*. ### Templates & Deployment - Pre-deployment validation includes semantic checks (call targets, argument types, trigger operand types). diff --git a/docker-env2/central-node-a/appsettings.Central.json b/docker-env2/central-node-a/appsettings.Central.json index 31d66d41..d07ff9a5 100644 --- a/docker-env2/central-node-a/appsettings.Central.json +++ b/docker-env2/central-node-a/appsettings.Central.json @@ -68,7 +68,9 @@ "OfflineTimeout": "00:01:00" }, "InboundApi": { - "DefaultMethodTimeout": "00:00:30" + "DefaultMethodTimeout": "00:00:30", + "_standbyForwarding": "2026-08-28 — a standby central node relays POST /api/{methodName} to the active peer instead of answering 503 STANDBY_NODE. List the OTHER node only.", + "StandbyForwardPeerEndpoints": [ "http://scadabridge-env2-central-b:5000" ] }, "Notification": { "SmtpServer": "scadabridge-smtp", diff --git a/docker-env2/central-node-b/appsettings.Central.json b/docker-env2/central-node-b/appsettings.Central.json index b8b588cd..4e56ef5f 100644 --- a/docker-env2/central-node-b/appsettings.Central.json +++ b/docker-env2/central-node-b/appsettings.Central.json @@ -68,7 +68,9 @@ "OfflineTimeout": "00:01:00" }, "InboundApi": { - "DefaultMethodTimeout": "00:00:30" + "DefaultMethodTimeout": "00:00:30", + "_standbyForwarding": "2026-08-28 — a standby central node relays POST /api/{methodName} to the active peer instead of answering 503 STANDBY_NODE. List the OTHER node only.", + "StandbyForwardPeerEndpoints": [ "http://scadabridge-env2-central-a:5000" ] }, "Notification": { "SmtpServer": "scadabridge-smtp", diff --git a/docker/central-node-a/appsettings.Central.json b/docker/central-node-a/appsettings.Central.json index 0184c93e..b96c4ee3 100644 --- a/docker/central-node-a/appsettings.Central.json +++ b/docker/central-node-a/appsettings.Central.json @@ -76,7 +76,9 @@ "OfflineTimeout": "00:01:00" }, "InboundApi": { - "DefaultMethodTimeout": "00:00:30" + "DefaultMethodTimeout": "00:00:30", + "_standbyForwarding": "2026-08-28 — a standby central node relays POST /api/{methodName} to the active peer instead of answering 503 STANDBY_NODE. List the OTHER node only.", + "StandbyForwardPeerEndpoints": [ "http://scadabridge-central-b:5000" ] }, "Notification": { "SmtpServer": "scadabridge-smtp", diff --git a/docker/central-node-b/appsettings.Central.json b/docker/central-node-b/appsettings.Central.json index 8ba50360..85da9d39 100644 --- a/docker/central-node-b/appsettings.Central.json +++ b/docker/central-node-b/appsettings.Central.json @@ -76,7 +76,9 @@ "OfflineTimeout": "00:01:00" }, "InboundApi": { - "DefaultMethodTimeout": "00:00:30" + "DefaultMethodTimeout": "00:00:30", + "_standbyForwarding": "2026-08-28 — a standby central node relays POST /api/{methodName} to the active peer instead of answering 503 STANDBY_NODE. List the OTHER node only.", + "StandbyForwardPeerEndpoints": [ "http://scadabridge-central-a:5000" ] }, "Notification": { "SmtpServer": "scadabridge-smtp", diff --git a/docs/requirements/Component-Host.md b/docs/requirements/Component-Host.md index ea7f14af..08280365 100644 --- a/docs/requirements/Component-Host.md +++ b/docs/requirements/Component-Host.md @@ -59,7 +59,7 @@ The Host must bind configuration sections from `appsettings.json` to strongly-ty | `ScadaBridge:SiteEventLog` | `SiteEventLogOptions` | Site Event Logging | RetentionDays, MaxStorageMb, PurgeScheduleCron | | `ScadaBridge:Communication` | `CommunicationOptions` | Communication | DeploymentTimeout, LifecycleTimeout, QueryTimeout, TransportHeartbeatInterval, TransportFailureThreshold | | `ScadaBridge:Security` | `SecurityOptions` | Security & Auth | LdapServer, LdapPort, LdapUseTls, JwtSigningKey, JwtExpiryMinutes, IdleTimeoutMinutes | -| `ScadaBridge:InboundApi` | `InboundApiOptions` | Inbound API | DefaultMethodTimeout | +| `ScadaBridge:InboundApi` | `InboundApiOptions` | Inbound API | DefaultMethodTimeout, MaxRequestBodyBytes, ApiMethodCacheTtl, StandbyForwardPeerEndpoints (HTTP base URLs of the other central node — a standby relays `POST /api/*` there instead of answering 503), StandbyForwardTimeout | | `ScadaBridge:Notification` | `NotificationOptions` | Notification Service | (SMTP config is stored in the central config DB, not in appsettings) | | `ScadaBridge:NotificationOutbox` | `NotificationOutboxOptions` | Notification Outbox | Dispatcher poll interval, stuck-age threshold, retention window (delivery retry settings reuse the central SMTP configuration) | | `ScadaBridge:SiteCallAudit` | `SiteCallAuditOptions` | Site Call Audit | Reconciliation pull interval, stuck-age threshold, retention window | diff --git a/docs/requirements/Component-InboundAPI.md b/docs/requirements/Component-InboundAPI.md index d08aa534..2c1650ef 100644 --- a/docs/requirements/Component-InboundAPI.md +++ b/docs/requirements/Component-InboundAPI.md @@ -6,7 +6,7 @@ The Inbound API exposes a web API on the central cluster that external systems c ## Location -Central cluster only (active node). Not available at site clusters. +Central cluster only. Method scripts execute on the **active** central node; a **standby** central node accepts the same endpoint and relays the request to the active peer (see [Standby forwarding](#standby-forwarding)). Not available at site clusters. ## Responsibilities @@ -16,7 +16,7 @@ Central cluster only (active node). Not available at site clusters. - Enforce per-method API key authorization (only approved keys can call a given method). - Execute the C# script implementation for the called method. - Return structured responses to the caller. -- Failover: API becomes available on the new active node after central failover. +- Failover: API becomes available on the new active node after central failover; during the handover a standby relays to whichever peer is active, so a caller addressing either central node directly is served without waiting for the load balancer's health probe to catch up. ## API Key Management @@ -122,7 +122,8 @@ The `sbk__` design already supports zero-downtime rotation — a | `INVALID_JSON` | 400 | Request body was not valid JSON. | | `VALIDATION_FAILED` | 400 | Parameter validation failed (missing required field, wrong type, undeclared field). | | `BODY_TOO_LARGE` | 413 | Request body exceeded `MaxRequestBodyBytes`. | -| `STANDBY_NODE` | 503 | Request hit a standby central node; the inbound API serves only the active node. | +| `STANDBY_NODE` | 503 | Request hit a standby central node that could not relay it: no `StandbyForwardPeerEndpoints` configured, or the request was already relayed once (loop guard — neither node is active, e.g. mid-failover). Carries the `X-ScadaBridge-Standby: 1` response header. | +| `ACTIVE_NODE_UNREACHABLE` | 503 | A standby tried every configured peer and none served the relay (unreachable, timed out after `StandbyForwardTimeout`, or also standby). | | `TIMEOUT` | 500 | The method script exceeded its execution timeout. | | `SITE_UNREACHABLE` | 500 | The target site did not respond to the routed request within the routing timeout (no contact / down / partitioned) — typed from the routed Ask expiring (`AskTimeoutException`), never inferred from an error-message substring. A `Success=false` response means the site answered and is classified `SCRIPT_ERROR`. | | `SCRIPT_COMPILE_FAILED` | 500 | The method's script failed to compile (returned consistently on every node until a compiling version is saved). | @@ -282,6 +283,19 @@ For everything else, scripts interact with the system through the curated `Route 3. The key's provisioned scopes include the requested method name (case-sensitive). - Failed authentication returns an appropriate HTTP error (401 Unauthorized for invalid/missing credentials; 403 Forbidden for a valid key not approved for the method). For enumeration-safety, both "method not found" and "key not in scope" return 403 with an identical body. +## Standby forwarding + +Both central nodes expose `POST /api/{methodName}`, but only the active node (the oldest Up member — `ActiveNodeEvaluator.SelfIsOldestUp`, the same verdict behind `/health/active`) executes method scripts and `Route.To()` calls; running them on a standby could race the active node or read stale singleton state. Before 2026-08-28 a standby simply answered 503 `STANDBY_NODE`, and "both nodes work" relied entirely on the load balancer probing `/health/active` — a caller addressing a node directly, or hitting the balancer inside its probe interval after a failover, got the 503. + +A standby now **relays** the request to the active peer instead (`InboundApiEndpointFilter` → `IStandbyRequestForwarder` / `HttpStandbyRequestForwarder`): + +- **Configuration.** `ScadaBridge:InboundApi:StandbyForwardPeerEndpoints` — the HTTP base URL(s) of the *other* central node(s), e.g. `["http://scadabridge-central-b:5000"]` on central-a. Empty (default) keeps the old 503 behaviour. `StandbyForwardTimeout` (default 60 s) is the per-peer attempt deadline and must exceed the longest method timeout in use. Validated at startup (absolute http/https URLs, positive timeout). Note the standby discovers the active peer from **appsettings**, not from cluster state — the Akka member address carries the remoting port, not the HTTP port — so this is a static, restart-to-change list, like `CentralGrpcEndpoints` on sites. +- **Pure relay.** Method, path, query string, body and headers travel verbatim (hop-by-hop headers and `Host` excepted), so the **active node** authenticates the caller's own `Authorization`/`X-API-Key` credential, applies its own body cap and method timeout, and writes the audit row. The peer's status, `Content-Type` and body are replayed unchanged. The standby's `AuditWriteMiddleware` skips its own `ApiInbound` row for a relayed request (`HttpContext.Items[RelayedToActiveItemKey]`) so a call is audited once, where it ran. +- **Order of checks.** The body-size cap (413 `BODY_TOO_LARGE`) runs *before* the active-node check, so a standby never buffers or relays a body the active node would refuse. The body is buffered once on the standby (bounded by `MaxRequestBodyBytes`) so it can be re-sent to a second peer. +- **Peer selection.** The standby knows only that it is *not* active. It tries the peers in order and accepts the first response that is not a standby refusal; a peer answering 503 with `X-ScadaBridge-Standby: 1`, a connection failure, or a `StandbyForwardTimeout` expiry each move to the next peer. A 503 *without* the marker is the active node's own answer and is replayed, not retried. All peers failing → 503 `ACTIVE_NODE_UNREACHABLE`. +- **One-hop loop guard.** Every relayed request carries `X-ScadaBridge-Forwarded: `. A standby never relays a request that already carries it — it answers 503 `STANDBY_NODE` with the marker header — so a pair in which neither node is active (a transient during failover) bounces at most one hop and never ping-pongs. The active node ignores the header and serves the request normally. +- **Still not a retry layer.** The relay makes one HTTP attempt per peer and does not retry a peer that answered; at-least-once callers keep their own retry loop and the idempotent-method authoring note above still applies. + ## Error Handling - Invalid API key → 401 Unauthorized. @@ -298,7 +312,7 @@ For everything else, scripts interact with the system through the curated `Route - **Configuration Database (via IAuditService)**: All API key and method definition changes are audit logged. - **Script Analysis (#25)**: `ForbiddenApiChecker` delegates to `ScriptTrustValidator.FindViolations` for the authoritative forbidden-API verdict during script compilation and validation. - **Audit Log (#23)**: Every inbound API request emits an `ApiInbound.Completed` row via `ICentralAuditWriter` from request middleware (non-blocking for the HTTP response). Request and response bodies are captured in full up to `AuditLog:InboundMaxBytes` (default 1 MiB) per the Audit Log Payload Capture Policy; redaction (headers + per-target body redactors) still applies before persistence. -- **Cluster Infrastructure**: API is hosted on the active central node and fails over with it. +- **Cluster Infrastructure**: method execution is hosted on the active central node and fails over with it; `IActiveNodeGate` (Host) tells the endpoint filter whether to serve locally or relay. ## Interactions diff --git a/src/ZB.MOM.WW.ScadaBridge.InboundAPI/HttpStandbyRequestForwarder.cs b/src/ZB.MOM.WW.ScadaBridge.InboundAPI/HttpStandbyRequestForwarder.cs new file mode 100644 index 00000000..02fc6a56 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.InboundAPI/HttpStandbyRequestForwarder.cs @@ -0,0 +1,238 @@ +using System.Net; +using System.Net.Http.Headers; +using Microsoft.AspNetCore.Http; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; + +namespace ZB.MOM.WW.ScadaBridge.InboundAPI; + +/// +/// Production : relays a standby-node inbound +/// API request to the active central node over plain HTTP using the named +/// client. +/// +/// Peer selection. The standby does not know which peer is active — it +/// only knows it is not. It therefore tries +/// in order and accepts the +/// first response that is not itself a standby refusal. A peer that is also a standby +/// (mid-failover, or a misconfigured list that includes this node) answers 503 with the +/// marker, which the forwarder treats like a +/// connection failure: try the next peer. Connection errors and per-attempt timeouts +/// () also move to the next peer. +/// When every peer fails the caller gets 503 ACTIVE_NODE_UNREACHABLE. +/// +/// Loop guard. Every relayed request carries +/// . never +/// forwards a request that already carries it, so a pair in which neither node is +/// active (a transient during failover) bounces at most one hop, never ping-pongs. +/// +/// Fidelity. Method, path, query string, body and headers are relayed +/// verbatim (hop-by-hop headers and Host excepted), so the active node +/// authenticates the caller's own Authorization/X-API-Key credential and +/// applies its own body cap and method timeout. The active node's status, content type +/// and body are replayed to the caller unchanged. The body is buffered once on the +/// standby (bounded by , which the +/// filter enforces before calling this class) so it can be re-sent to a second peer. +/// +public sealed class HttpStandbyRequestForwarder : IStandbyRequestForwarder +{ + /// Name of the client used for relays. + public const string HttpClientName = "ScadaBridge.InboundApi.StandbyForward"; + + /// + /// Request header stamped on every relayed request (value: the forwarding node's + /// host name when known, else 1). The filter refuses to forward a request + /// that already carries it. + /// + public const string ForwardedRequestHeader = "X-ScadaBridge-Forwarded"; + + /// + /// Response header the filter adds to its own 503 STANDBY_NODE reply so a + /// forwarding peer can distinguish "that node is also standby — try the next one" + /// from a genuine 503 the active node's method script produced. + /// + public const string StandbyResponseHeader = "X-ScadaBridge-Standby"; + + /// Machine-readable code when no configured peer served the relay. + public const string ActiveNodeUnreachableCode = "ACTIVE_NODE_UNREACHABLE"; + + private static readonly HashSet HopByHopRequestHeaders = new(StringComparer.OrdinalIgnoreCase) + { + "Host", "Content-Length", "Transfer-Encoding", "Connection", "Keep-Alive", + "Proxy-Connection", "Upgrade", "TE", "Trailer", "Expect", + }; + + private readonly IHttpClientFactory _httpClientFactory; + private readonly IOptionsMonitor _options; + private readonly ILogger _logger; + + /// Initializes a new . + /// Factory for the named relay client. + /// Live-reloadable inbound API options (peer list + timeout are read per relay). + /// Logger for relay diagnostics. + public HttpStandbyRequestForwarder( + IHttpClientFactory httpClientFactory, + IOptionsMonitor options, + ILogger logger) + { + _httpClientFactory = httpClientFactory; + _options = options; + _logger = logger; + } + + /// + public bool IsConfigured => _options.CurrentValue.StandbyForwardPeerEndpoints.Count > 0; + + /// + public async Task ForwardAsync(HttpContext context, CancellationToken cancellationToken) + { + var opts = _options.CurrentValue; + var peers = opts.StandbyForwardPeerEndpoints; + if (peers.Count == 0) + return Unreachable(); + + // Buffer the body once so it can be replayed to a second peer. The filter has + // already enforced MaxRequestBodyBytes (Content-Length up front, and the + // per-request Kestrel cap for chunked bodies), so this is bounded. + byte[] body; + using (var buffer = new MemoryStream()) + { + await context.Request.Body.CopyToAsync(buffer, cancellationToken).ConfigureAwait(false); + body = buffer.ToArray(); + } + + var client = _httpClientFactory.CreateClient(HttpClientName); + var pathAndQuery = context.Request.Path.Value + context.Request.QueryString.Value; + var forwardedBy = string.IsNullOrEmpty(context.Request.Host.Host) ? "1" : context.Request.Host.Host; + + foreach (var peer in peers) + { + cancellationToken.ThrowIfCancellationRequested(); + + using var attemptCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + attemptCts.CancelAfter(opts.StandbyForwardTimeout); + + try + { + using var request = BuildRequest(context, peer, pathAndQuery, body, forwardedBy); + var response = await client + .SendAsync(request, HttpCompletionOption.ResponseHeadersRead, attemptCts.Token) + .ConfigureAwait(false); + + if (response.StatusCode == HttpStatusCode.ServiceUnavailable + && response.Headers.Contains(StandbyResponseHeader)) + { + response.Dispose(); + _logger.LogInformation( + "Inbound API relay: peer {Peer} is also a standby — trying the next peer", peer); + continue; + } + + var responseBody = await response.Content + .ReadAsByteArrayAsync(attemptCts.Token) + .ConfigureAwait(false); + var contentType = response.Content.Headers.ContentType?.ToString(); + var status = (int)response.StatusCode; + response.Dispose(); + + _logger.LogDebug( + "Inbound API relay: served {Path} via active peer {Peer} → HTTP {Status}", + context.Request.Path, peer, status); + return new RelayedResult(status, contentType, responseBody); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + throw; // caller went away — nothing to answer + } + catch (OperationCanceledException) + { + _logger.LogWarning( + "Inbound API relay: peer {Peer} did not answer within {Timeout} — trying the next peer", + peer, opts.StandbyForwardTimeout); + } + catch (HttpRequestException ex) + { + _logger.LogWarning(ex, + "Inbound API relay: peer {Peer} unreachable — trying the next peer", peer); + } + } + + _logger.LogWarning( + "Inbound API relay: no configured peer served {Path} ({PeerCount} tried)", + context.Request.Path, peers.Count); + return Unreachable(); + } + + private static HttpRequestMessage BuildRequest( + HttpContext context, string peer, string pathAndQuery, byte[] body, string forwardedBy) + { + var request = new HttpRequestMessage( + new HttpMethod(context.Request.Method), + new Uri(new Uri(peer.TrimEnd('/') + "/"), pathAndQuery.TrimStart('/'))); + + var content = new ByteArrayContent(body); + foreach (var header in context.Request.Headers) + { + if (HopByHopRequestHeaders.Contains(header.Key) + || string.Equals(header.Key, ForwardedRequestHeader, StringComparison.OrdinalIgnoreCase)) + continue; + + var values = header.Value.ToArray(); + if (!request.Headers.TryAddWithoutValidation(header.Key, values)) + content.Headers.TryAddWithoutValidation(header.Key, values); + } + + if (content.Headers.ContentType == null + && MediaTypeHeaderValue.TryParse(context.Request.ContentType, out var mediaType)) + content.Headers.ContentType = mediaType; + + request.Content = content; + request.Headers.TryAddWithoutValidation(ForwardedRequestHeader, forwardedBy); + return request; + } + + private static IResult Unreachable() => Results.Json( + new + { + error = "Inbound API request could not be relayed to the active central node", + code = ActiveNodeUnreachableCode, + }, + statusCode: StatusCodes.Status503ServiceUnavailable); + + /// + /// Replays a peer's status, content type and body to the original caller. + /// Exposed so tests can assert on what the relay produced without a server. + /// + public sealed class RelayedResult : IResult, IStatusCodeHttpResult + { + /// Initializes a new . + /// The peer's HTTP status. + /// The peer's Content-Type, if any. + /// The peer's response body. + public RelayedResult(int statusCode, string? contentType, byte[] body) + { + StatusCode = statusCode; + ContentType = contentType; + Body = body; + } + + /// + public int? StatusCode { get; } + + /// The peer's Content-Type, replayed verbatim. + public string? ContentType { get; } + + /// The peer's response body, replayed verbatim. + public byte[] Body { get; } + + /// + public async Task ExecuteAsync(HttpContext httpContext) + { + httpContext.Response.StatusCode = StatusCode ?? StatusCodes.Status200OK; + if (ContentType != null) + httpContext.Response.ContentType = ContentType; + httpContext.Response.ContentLength = Body.Length; + await httpContext.Response.Body.WriteAsync(Body, httpContext.RequestAborted).ConfigureAwait(false); + } + } +} diff --git a/src/ZB.MOM.WW.ScadaBridge.InboundAPI/IStandbyRequestForwarder.cs b/src/ZB.MOM.WW.ScadaBridge.InboundAPI/IStandbyRequestForwarder.cs new file mode 100644 index 00000000..f3a16cc6 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.InboundAPI/IStandbyRequestForwarder.cs @@ -0,0 +1,33 @@ +using Microsoft.AspNetCore.Http; + +namespace ZB.MOM.WW.ScadaBridge.InboundAPI; + +/// +/// Standby-node forwarding seam for POST /api/{methodName}. When a request +/// lands on a standby central node, hands it +/// here instead of returning HTTP 503, and the implementation relays it verbatim to +/// the active central node over HTTP so callers that address a central node directly +/// (or that hit the load balancer inside its health-probe window after a failover) +/// still get a served response. Method scripts and Route.To() calls still run +/// ONLY on the active node — the standby is a pure relay. +/// +public interface IStandbyRequestForwarder +{ + /// + /// true when at least one peer endpoint is configured + /// (). When false + /// the filter falls back to the pre-forwarding behaviour: HTTP 503 STANDBY_NODE. + /// + bool IsConfigured { get; } + + /// + /// Relays the current request to the first peer that serves it and returns a + /// result that replays the peer's status, content type and body to the caller. + /// Never throws for peer failures: when no peer served the request the result is + /// HTTP 503 ACTIVE_NODE_UNREACHABLE. + /// + /// The inbound HTTP context on the standby node. + /// Cancelled when the caller aborts the request. + /// A result replaying the peer's response, or a 503 when no peer served it. + Task ForwardAsync(HttpContext context, CancellationToken cancellationToken); +} diff --git a/src/ZB.MOM.WW.ScadaBridge.InboundAPI/InboundApiEndpointFilter.cs b/src/ZB.MOM.WW.ScadaBridge.InboundAPI/InboundApiEndpointFilter.cs index 4cfc3edf..03fd8bed 100644 --- a/src/ZB.MOM.WW.ScadaBridge.InboundAPI/InboundApiEndpointFilter.cs +++ b/src/ZB.MOM.WW.ScadaBridge.InboundAPI/InboundApiEndpointFilter.cs @@ -12,17 +12,32 @@ namespace ZB.MOM.WW.ScadaBridge.InboundAPI; /// /// /// -/// Active-node gating. The inbound API is central-active-node-only; -/// a standby node returns HTTP 503 so it never executes method scripts. +/// Request body size cap. Oversized bodies are rejected with HTTP +/// 413 before being buffered into a JsonDocument — and before any relay, +/// so a standby never buffers or forwards a body the active node would refuse. /// /// -/// Request body size cap. Oversized bodies are rejected with HTTP -/// 413 before being buffered into a JsonDocument. +/// Active-node gating. Method scripts execute only on the active central node. +/// A standby RELAYS the request to the active peer through +/// when peers are configured +/// (), and otherwise +/// returns HTTP 503 STANDBY_NODE. A request that already carries +/// is never +/// relayed again (one-hop loop guard) — it gets the 503, stamped with +/// so the relaying +/// peer knows to try its next endpoint. /// /// /// public sealed class InboundApiEndpointFilter : IEndpointFilter { + /// + /// key set when this node relayed the request to the + /// active peer. AuditWriteMiddleware reads it to skip its own audit row (the + /// active node audits the call it executed). + /// + public const string RelayedToActiveItemKey = "ScadaBridge.InboundApi.RelayedToActive"; + private readonly ILogger _logger; private readonly IOptionsMonitor _options; @@ -37,32 +52,20 @@ public sealed class InboundApiEndpointFilter : IEndpointFilter _options = options; } - /// Applies active-node gating and request body size checks before delegating to the next filter or handler. + /// Applies the request body size check and active-node gating/relay before delegating to the next filter or handler. /// The endpoint filter invocation context containing the HTTP context and arguments. /// The next filter or endpoint handler in the pipeline. - /// A value task that resolves to the filter pipeline result, or an error result if gating fails. + /// A value task that resolves to the filter pipeline result, a relayed peer response, or an error result if gating fails. public async ValueTask InvokeAsync( EndpointFilterInvocationContext context, EndpointFilterDelegate next) { var httpContext = context.HttpContext; - // Refuse to serve the inbound API on a standby central node. - // The gate is optional — when no IActiveNodeGate is registered (non-clustered - // host / tests) the API is served, preserving prior behaviour. - var gate = httpContext.RequestServices.GetService(); - if (gate is { IsActiveNode: false }) - { - _logger.LogWarning( - "Inbound API request rejected — this node is a standby (not the active central node)"); - return Results.Json( - new { error = "Inbound API is only available on the active central node", code = "STANDBY_NODE" }, - statusCode: StatusCodes.Status503ServiceUnavailable); - } - - // Cap the request body size. Reject an over-limit body up + // Cap the request body size FIRST. Reject an over-limit body up // front via Content-Length; also lower the per-request max body size so a - // chunked/unknown-length stream is cut off by Kestrel as it is read. + // chunked/unknown-length stream is cut off by Kestrel as it is read. Runs + // before the active-node check so a standby never relays an oversized body. var maxBytes = _options.CurrentValue.MaxRequestBodyBytes; if (httpContext.Request.ContentLength is { } declaredLength && declaredLength > maxBytes) { @@ -80,6 +83,33 @@ public sealed class InboundApiEndpointFilter : IEndpointFilter sizeFeature.MaxRequestBodySize = maxBytes; } + // Active-node gating. The gate is optional — when no IActiveNodeGate is + // registered (non-clustered host / tests) the API is served locally. + var gate = httpContext.RequestServices.GetService(); + if (gate is { IsActiveNode: false }) + { + var alreadyForwarded = httpContext.Request.Headers + .ContainsKey(HttpStandbyRequestForwarder.ForwardedRequestHeader); + var forwarder = httpContext.RequestServices.GetService(); + + if (!alreadyForwarded && forwarder is { IsConfigured: true }) + { + httpContext.Items[RelayedToActiveItemKey] = true; + _logger.LogDebug("Inbound API request on a standby node — relaying to the active peer"); + return await forwarder.ForwardAsync(httpContext, httpContext.RequestAborted) + .ConfigureAwait(false); + } + + _logger.LogWarning( + alreadyForwarded + ? "Inbound API request rejected — relayed here by a peer, but this node is also a standby (loop guard)" + : "Inbound API request rejected — this node is a standby (not the active central node) and no relay peers are configured"); + httpContext.Response.Headers[HttpStandbyRequestForwarder.StandbyResponseHeader] = "1"; + return Results.Json( + new { error = "Inbound API is only available on the active central node", code = "STANDBY_NODE" }, + statusCode: StatusCodes.Status503ServiceUnavailable); + } + return await next(context); } } diff --git a/src/ZB.MOM.WW.ScadaBridge.InboundAPI/InboundApiOptions.cs b/src/ZB.MOM.WW.ScadaBridge.InboundAPI/InboundApiOptions.cs index cd172a61..d4b41e8c 100644 --- a/src/ZB.MOM.WW.ScadaBridge.InboundAPI/InboundApiOptions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.InboundAPI/InboundApiOptions.cs @@ -48,4 +48,25 @@ public class InboundApiOptions /// serve minutes-old method definitions. /// public TimeSpan ApiMethodCacheTtl { get; set; } = TimeSpan.FromSeconds(5); + + /// + /// Standby-node forwarding (2026-08-28): the HTTP base URLs of the OTHER central + /// node(s) — e.g. ["http://scadabridge-central-b:5000"] on central-a — that a + /// standby relays POST /api/{methodName} to instead of answering 503 + /// STANDBY_NODE. Tried in order; the first peer that is not itself a standby + /// serves the call. Empty (the default) keeps the pre-forwarding behaviour: a standby + /// refuses with 503. Method scripts still execute only on the active node — the + /// standby is a pure relay (see ). Each entry + /// must be an absolute http/https URL; never list this node's own address + /// (it would cost one wasted round-trip per call — the loop guard stops it bouncing). + /// + public List StandbyForwardPeerEndpoints { get; set; } = new(); + + /// + /// Per-peer deadline for one relay attempt (connect + the active node's full method + /// execution). Must comfortably exceed and the + /// longest per-method timeout in use, otherwise the standby gives up on a peer that + /// is still legitimately working and moves on to the next one. Default 60 s. + /// + public TimeSpan StandbyForwardTimeout { get; set; } = TimeSpan.FromSeconds(60); } diff --git a/src/ZB.MOM.WW.ScadaBridge.InboundAPI/InboundApiOptionsValidator.cs b/src/ZB.MOM.WW.ScadaBridge.InboundAPI/InboundApiOptionsValidator.cs index 574dcd37..f9b35c74 100644 --- a/src/ZB.MOM.WW.ScadaBridge.InboundAPI/InboundApiOptionsValidator.cs +++ b/src/ZB.MOM.WW.ScadaBridge.InboundAPI/InboundApiOptionsValidator.cs @@ -31,5 +31,18 @@ public sealed class InboundApiOptionsValidator : OptionsValidatorBase TimeSpan.Zero, $"ScadaBridge:InboundApi:ApiMethodCacheTtl must be a positive duration " + $"(was {options.ApiMethodCacheTtl}); it bounds how long a resolved ApiMethod row is cached."); + + builder.RequireThat(options.StandbyForwardTimeout > TimeSpan.Zero, + $"ScadaBridge:InboundApi:StandbyForwardTimeout must be a positive duration " + + $"(was {options.StandbyForwardTimeout}); it bounds each standby→active relay attempt."); + + foreach (var peer in options.StandbyForwardPeerEndpoints) + { + var ok = Uri.TryCreate(peer, UriKind.Absolute, out var uri) + && (uri.Scheme == Uri.UriSchemeHttp || uri.Scheme == Uri.UriSchemeHttps); + builder.RequireThat(ok, + $"ScadaBridge:InboundApi:StandbyForwardPeerEndpoints entry '{peer}' must be an absolute " + + $"http/https URL (e.g. http://central-b:5000); it is the peer a standby relays inbound API calls to."); + } } } diff --git a/src/ZB.MOM.WW.ScadaBridge.InboundAPI/Middleware/AuditWriteMiddleware.cs b/src/ZB.MOM.WW.ScadaBridge.InboundAPI/Middleware/AuditWriteMiddleware.cs index 8aefd49d..664abf50 100644 --- a/src/ZB.MOM.WW.ScadaBridge.InboundAPI/Middleware/AuditWriteMiddleware.cs +++ b/src/ZB.MOM.WW.ScadaBridge.InboundAPI/Middleware/AuditWriteMiddleware.cs @@ -219,6 +219,12 @@ public sealed class AuditWriteMiddleware { sw.Stop(); + // Standby forwarding (2026-08-28): a request this node RELAYED to the active + // peer is audited there, by the node that actually authenticated and executed + // it. Emitting a second row here would double-count every relayed call, so the + // standby skips its own row when the filter flagged the relay. + var relayed = ctx.Items.ContainsKey(InboundApiEndpointFilter.RelayedToActiveItemKey); + // Restore the original stream and resolve the captured audit copy. // The forwarding wrapper has already written every byte to the // original sink; this just pulls back the bounded UTF-8 string. @@ -232,7 +238,8 @@ public sealed class AuditWriteMiddleware var responseBody = skipBody ? null : capturedResponseBody; var responseTruncated = skipBody ? false : capturedResponseTruncated; - EmitInboundAudit( + if (!relayed) + EmitInboundAudit( ctx, opts, sw.ElapsedMilliseconds, diff --git a/src/ZB.MOM.WW.ScadaBridge.InboundAPI/ServiceCollectionExtensions.cs b/src/ZB.MOM.WW.ScadaBridge.InboundAPI/ServiceCollectionExtensions.cs index 57669485..07111e34 100644 --- a/src/ZB.MOM.WW.ScadaBridge.InboundAPI/ServiceCollectionExtensions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.InboundAPI/ServiceCollectionExtensions.cs @@ -35,6 +35,15 @@ public static class ServiceCollectionExtensions // body size cap and active-node gating for POST /api/{methodName}. services.AddSingleton(); + // Standby-node forwarding (2026-08-28): a standby relays POST /api/{methodName} + // to the active peer over HTTP instead of answering 503 STANDBY_NODE. The + // named client has no HttpClient.Timeout of its own — the forwarder's per-attempt + // CTS (InboundApiOptions.StandbyForwardTimeout) is the single deadline, so a + // long-running method on the active node is not cut off by a second, hidden cap. + services.AddHttpClient(HttpStandbyRequestForwarder.HttpClientName) + .ConfigureHttpClient(c => c.Timeout = Timeout.InfiniteTimeSpan); + services.AddSingleton(); + // N1: the ApiMethod consumer of the script-artifact invalidation bus. // The bus parameter is optional — hosts without a registered // IScriptArtifactChangeBus (site roles, plain test hosts) start it as a no-op diff --git a/tests/ZB.MOM.WW.ScadaBridge.InboundAPI.Tests/InboundApiOptionsValidatorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.InboundAPI.Tests/InboundApiOptionsValidatorTests.cs index a5c9f381..57e7da3b 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.InboundAPI.Tests/InboundApiOptionsValidatorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.InboundAPI.Tests/InboundApiOptionsValidatorTests.cs @@ -83,4 +83,38 @@ public class InboundApiOptionsValidatorTests Assert.True(result.Failed); Assert.Contains("ApiMethodCacheTtl", result.FailureMessage); } + + // --- Standby forwarding (2026-08-28) --- + + [Fact] + public void StandbyForwardPeerEndpoints_AbsoluteHttpUrls_AreValid() + { + var result = Validate(new InboundApiOptions + { + StandbyForwardPeerEndpoints = { "http://central-b:5000", "https://central-b.example/" }, + }); + Assert.True(result.Succeeded, result.FailureMessage); + } + + [Theory] + [InlineData("central-b:5000")] + [InlineData("/api")] + [InlineData("ftp://central-b")] + [InlineData("")] + public void StandbyForwardPeerEndpoints_NonHttpOrRelative_IsRejected(string peer) + { + var result = Validate(new InboundApiOptions { StandbyForwardPeerEndpoints = { peer } }); + + Assert.True(result.Failed); + Assert.Contains("StandbyForwardPeerEndpoints", result.FailureMessage); + } + + [Fact] + public void ZeroStandbyForwardTimeout_IsRejected() + { + var result = Validate(new InboundApiOptions { StandbyForwardTimeout = TimeSpan.Zero }); + + Assert.True(result.Failed); + Assert.Contains("StandbyForwardTimeout", result.FailureMessage); + } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.InboundAPI.Tests/StandbyRequestForwardingTests.cs b/tests/ZB.MOM.WW.ScadaBridge.InboundAPI.Tests/StandbyRequestForwardingTests.cs new file mode 100644 index 00000000..a2cc47e0 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.InboundAPI.Tests/StandbyRequestForwardingTests.cs @@ -0,0 +1,400 @@ +using System.Net; +using System.Text; +using Microsoft.AspNetCore.Http; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; + +namespace ZB.MOM.WW.ScadaBridge.InboundAPI.Tests; + +/// +/// Standby-node forwarding (2026-08-28): a standby central node must relay +/// POST /api/{methodName} to the active peer instead of refusing with 503, +/// while never executing the method itself and never relaying more than one hop. +/// Covers both the decision and the +/// relay mechanics. +/// +public class StandbyRequestForwardingTests +{ + // ------------------------------------------------------------------ + // Filter: when does a standby relay, and when does it refuse? + // ------------------------------------------------------------------ + + [Fact] + public async Task Standby_WithForwarderConfigured_RelaysAndDoesNotRunHandler() + { + var forwarder = new StubForwarder(configured: true, result: Results.Text("relayed")); + var (ctx, invocation) = BuildInvocation(activeNode: false, forwarder); + var next = NextSentinel(out var handlerRan); + + var result = await CreateFilter().InvokeAsync(invocation, next); + + Assert.False(handlerRan()); + Assert.Equal(1, forwarder.Calls); + Assert.Same(forwarder.Result, result); + Assert.True(ctx.Items.ContainsKey(InboundApiEndpointFilter.RelayedToActiveItemKey)); + } + + [Fact] + public async Task Standby_WithForwarderUnconfigured_Returns503StandbyNode_MarkedForPeers() + { + var forwarder = new StubForwarder(configured: false, result: Results.Ok()); + var (ctx, invocation) = BuildInvocation(activeNode: false, forwarder); + var next = NextSentinel(out var handlerRan); + + var result = await CreateFilter().InvokeAsync(invocation, next); + + Assert.False(handlerRan()); + Assert.Equal(0, forwarder.Calls); + AssertStatus(result, StatusCodes.Status503ServiceUnavailable); + Assert.Equal("1", ctx.Response.Headers[HttpStandbyRequestForwarder.StandbyResponseHeader]); + Assert.False(ctx.Items.ContainsKey(InboundApiEndpointFilter.RelayedToActiveItemKey)); + } + + [Fact] + public async Task Standby_AlreadyForwardedRequest_IsNotRelayedAgain_LoopGuard() + { + var forwarder = new StubForwarder(configured: true, result: Results.Ok()); + var (ctx, invocation) = BuildInvocation(activeNode: false, forwarder); + ctx.Request.Headers[HttpStandbyRequestForwarder.ForwardedRequestHeader] = "central-b"; + var next = NextSentinel(out var handlerRan); + + var result = await CreateFilter().InvokeAsync(invocation, next); + + Assert.False(handlerRan()); + Assert.Equal(0, forwarder.Calls); + AssertStatus(result, StatusCodes.Status503ServiceUnavailable); + Assert.Equal("1", ctx.Response.Headers[HttpStandbyRequestForwarder.StandbyResponseHeader]); + } + + [Fact] + public async Task Standby_OversizedBody_Returns413_BeforeRelaying() + { + var options = new InboundApiOptions(); + var forwarder = new StubForwarder(configured: true, result: Results.Ok()); + var (_, invocation) = BuildInvocation(activeNode: false, forwarder, + contentLength: options.MaxRequestBodyBytes + 1); + var next = NextSentinel(out var handlerRan); + + var result = await CreateFilter(options).InvokeAsync(invocation, next); + + Assert.False(handlerRan()); + Assert.Equal(0, forwarder.Calls); + AssertStatus(result, StatusCodes.Status413PayloadTooLarge); + } + + [Fact] + public async Task ActiveNode_ForwardedRequest_IsServedLocally() + { + // The active node receives the relay from its standby peer and serves it — + // the marker header only matters when the receiving node is NOT active. + var forwarder = new StubForwarder(configured: true, result: Results.Ok()); + var (ctx, invocation) = BuildInvocation(activeNode: true, forwarder); + ctx.Request.Headers[HttpStandbyRequestForwarder.ForwardedRequestHeader] = "central-b"; + var next = NextSentinel(out var handlerRan); + + await CreateFilter().InvokeAsync(invocation, next); + + Assert.True(handlerRan()); + Assert.Equal(0, forwarder.Calls); + } + + // ------------------------------------------------------------------ + // Forwarder: relay mechanics against a fake HTTP peer + // ------------------------------------------------------------------ + + [Fact] + public async Task Forwarder_RelaysMethodPathQueryHeadersAndBody_AndReplaysPeerResponse() + { + HttpRequestMessage? seen = null; + string? seenBody = null; + var handler = new FakeHandler(async req => + { + seen = req; + seenBody = await req.Content!.ReadAsStringAsync(); + return new HttpResponseMessage(HttpStatusCode.Created) + { + Content = new StringContent("{\"ok\":true}", Encoding.UTF8, "application/json"), + }; + }); + var forwarder = CreateForwarder(handler, "http://central-b:5000/"); + var ctx = BuildRequestContext("{\"x\":1}"); + ctx.Request.Headers["X-API-Key"] = "sbk_key_secret"; + ctx.Request.Headers["Host"] = "central-a:5000"; + + var result = await forwarder.ForwardAsync(ctx, CancellationToken.None); + + var relayed = Assert.IsType(result); + Assert.Equal(201, relayed.StatusCode); + Assert.StartsWith("application/json", relayed.ContentType); + Assert.Equal("{\"ok\":true}", Encoding.UTF8.GetString(relayed.Body)); + + Assert.NotNull(seen); + Assert.Equal(HttpMethod.Post, seen!.Method); + Assert.Equal("http://central-b:5000/api/GetRecipe?v=2", seen.RequestUri!.ToString()); + Assert.Equal("{\"x\":1}", seenBody); + Assert.Equal("sbk_key_secret", Assert.Single(seen.Headers.GetValues("X-API-Key"))); + Assert.Equal("central-a", Assert.Single(seen.Headers.GetValues(HttpStandbyRequestForwarder.ForwardedRequestHeader))); + Assert.Null(seen.Headers.Host); // hop-by-hop / origin headers are not relayed + Assert.Equal("application/json", seen.Content!.Headers.ContentType!.MediaType); + } + + [Fact] + public async Task Forwarder_ReplaysPeerErrorStatusUnchanged() + { + var handler = new FakeHandler(_ => Task.FromResult(new HttpResponseMessage(HttpStatusCode.Forbidden) + { + Content = new StringContent("{\"code\":\"NOT_APPROVED\"}", Encoding.UTF8, "application/json"), + })); + var forwarder = CreateForwarder(handler, "http://central-b:5000"); + + var result = await forwarder.ForwardAsync(BuildRequestContext("{}"), CancellationToken.None); + + var relayed = Assert.IsType(result); + Assert.Equal(403, relayed.StatusCode); + Assert.Contains("NOT_APPROVED", Encoding.UTF8.GetString(relayed.Body)); + } + + [Fact] + public async Task Forwarder_PeerUnreachable_TriesNextPeer_WithSameBody() + { + var bodies = new List(); + var handler = new FakeHandler(async req => + { + bodies.Add(await req.Content!.ReadAsStringAsync()); + if (req.RequestUri!.Host == "dead") + throw new HttpRequestException("connection refused"); + return new HttpResponseMessage(HttpStatusCode.OK) { Content = new StringContent("served") }; + }); + var forwarder = CreateForwarder(handler, "http://dead:5000", "http://alive:5000"); + + var result = await forwarder.ForwardAsync(BuildRequestContext("{\"n\":7}"), CancellationToken.None); + + var relayed = Assert.IsType(result); + Assert.Equal(200, relayed.StatusCode); + Assert.Equal(new[] { "{\"n\":7}", "{\"n\":7}" }, bodies); + } + + [Fact] + public async Task Forwarder_PeerIsAlsoStandby_TriesNextPeer() + { + var handler = new FakeHandler(req => + { + if (req.RequestUri!.Host == "standby") + { + var refusal = new HttpResponseMessage(HttpStatusCode.ServiceUnavailable); + refusal.Headers.TryAddWithoutValidation(HttpStandbyRequestForwarder.StandbyResponseHeader, "1"); + return Task.FromResult(refusal); + } + return Task.FromResult(new HttpResponseMessage(HttpStatusCode.OK) { Content = new StringContent("served") }); + }); + var forwarder = CreateForwarder(handler, "http://standby:5000", "http://active:5000"); + + var result = await forwarder.ForwardAsync(BuildRequestContext("{}"), CancellationToken.None); + + var relayed = Assert.IsType(result); + Assert.Equal(200, relayed.StatusCode); + Assert.Equal("served", Encoding.UTF8.GetString(relayed.Body)); + } + + [Fact] + public async Task Forwarder_Genuine503FromActivePeer_IsReplayed_NotRetried() + { + // A 503 WITHOUT the standby marker is the active node's own answer (e.g. a + // method script returning 503) and must be replayed, not treated as a miss. + var calls = 0; + var handler = new FakeHandler(_ => + { + calls++; + return Task.FromResult(new HttpResponseMessage(HttpStatusCode.ServiceUnavailable) + { + Content = new StringContent("busy"), + }); + }); + var forwarder = CreateForwarder(handler, "http://a:5000", "http://b:5000"); + + var result = await forwarder.ForwardAsync(BuildRequestContext("{}"), CancellationToken.None); + + var relayed = Assert.IsType(result); + Assert.Equal(503, relayed.StatusCode); + Assert.Equal("busy", Encoding.UTF8.GetString(relayed.Body)); + Assert.Equal(1, calls); + } + + [Fact] + public async Task Forwarder_AllPeersFail_Returns503ActiveNodeUnreachable() + { + var handler = new FakeHandler(_ => throw new HttpRequestException("down")); + var forwarder = CreateForwarder(handler, "http://a:5000", "http://b:5000"); + + var result = await forwarder.ForwardAsync(BuildRequestContext("{}"), CancellationToken.None); + + Assert.IsNotType(result); + AssertStatus(result, StatusCodes.Status503ServiceUnavailable); + } + + [Fact] + public async Task Forwarder_PeerTimesOut_TriesNextPeer() + { + var handler = new FakeHandler(async (req, ct) => + { + if (req.RequestUri!.Host == "slow") + { + await Task.Delay(Timeout.InfiniteTimeSpan, ct); // honours the per-attempt deadline + } + return new HttpResponseMessage(HttpStatusCode.OK) { Content = new StringContent("served") }; + }); + var forwarder = CreateForwarder(handler, TimeSpan.FromMilliseconds(100), "http://slow:5000", "http://fast:5000"); + + var result = await forwarder.ForwardAsync(BuildRequestContext("{}"), CancellationToken.None); + + var relayed = Assert.IsType(result); + Assert.Equal(200, relayed.StatusCode); + } + + [Fact] + public async Task Forwarder_CallerAborts_Propagates_DoesNotTryNextPeer() + { + var calls = 0; + using var callerCts = new CancellationTokenSource(); + var handler = new FakeHandler(async (_, ct) => + { + calls++; + callerCts.Cancel(); + await Task.Delay(Timeout.InfiniteTimeSpan, ct); + throw new InvalidOperationException("unreachable"); + }); + var forwarder = CreateForwarder(handler, "http://a:5000", "http://b:5000"); + + await Assert.ThrowsAnyAsync( + () => forwarder.ForwardAsync(BuildRequestContext("{}"), callerCts.Token)); + Assert.Equal(1, calls); + } + + [Fact] + public void Forwarder_IsConfigured_TracksPeerList() + { + Assert.False(CreateForwarder(new FakeHandler(_ => throw new InvalidOperationException())).IsConfigured); + Assert.True(CreateForwarder(new FakeHandler(_ => throw new InvalidOperationException()), "http://b:5000").IsConfigured); + } + + [Fact] + public async Task RelayedResult_WritesStatusContentTypeAndBody() + { + var ctx = new DefaultHttpContext(); + ctx.Response.Body = new MemoryStream(); + var relayed = new HttpStandbyRequestForwarder.RelayedResult(418, "text/plain", Encoding.UTF8.GetBytes("tea")); + + await relayed.ExecuteAsync(ctx); + + Assert.Equal(418, ctx.Response.StatusCode); + Assert.Equal("text/plain", ctx.Response.ContentType); + Assert.Equal(3, ctx.Response.ContentLength); + ctx.Response.Body.Position = 0; + Assert.Equal("tea", new StreamReader(ctx.Response.Body).ReadToEnd()); + } + + // ------------------------------------------------------------------ + // helpers + // ------------------------------------------------------------------ + + private static InboundApiEndpointFilter CreateFilter(InboundApiOptions? options = null) => + new(NullLogger.Instance, + new StaticOptionsMonitor(options ?? new InboundApiOptions())); + + private static (DefaultHttpContext ctx, DefaultEndpointFilterInvocationContext invocation) + BuildInvocation(bool activeNode, IStandbyRequestForwarder forwarder, long contentLength = 2) + { + var services = new ServiceCollection(); + services.AddSingleton(new StubActiveNodeGate(activeNode)); + services.AddSingleton(forwarder); + var ctx = new DefaultHttpContext { RequestServices = services.BuildServiceProvider() }; + ctx.Request.Method = "POST"; + ctx.Request.ContentLength = contentLength; + return (ctx, new DefaultEndpointFilterInvocationContext(ctx)); + } + + private static EndpointFilterDelegate NextSentinel(out Func wasCalled) + { + var called = false; + wasCalled = () => called; + return _ => + { + called = true; + return ValueTask.FromResult(Results.Ok("handler-ran")); + }; + } + + private static void AssertStatus(object? result, int expected) + { + var status = Assert.IsAssignableFrom(result); + Assert.Equal(expected, status.StatusCode); + } + + private static DefaultHttpContext BuildRequestContext(string body) + { + var ctx = new DefaultHttpContext(); + ctx.Request.Method = "POST"; + ctx.Request.Path = "/api/GetRecipe"; + ctx.Request.QueryString = new QueryString("?v=2"); + ctx.Request.ContentType = "application/json"; + ctx.Request.Body = new MemoryStream(Encoding.UTF8.GetBytes(body)); + return ctx; + } + + private static HttpStandbyRequestForwarder CreateForwarder(FakeHandler handler, params string[] peers) => + CreateForwarder(handler, TimeSpan.FromSeconds(5), peers); + + private static HttpStandbyRequestForwarder CreateForwarder(FakeHandler handler, TimeSpan timeout, params string[] peers) => + new(new FakeHttpClientFactory(handler), + new StaticOptionsMonitor(new InboundApiOptions + { + StandbyForwardPeerEndpoints = peers.ToList(), + StandbyForwardTimeout = timeout, + }), + NullLogger.Instance); + + private sealed class StubForwarder : IStandbyRequestForwarder + { + public StubForwarder(bool configured, IResult result) { IsConfigured = configured; Result = result; } + public bool IsConfigured { get; } + public IResult Result { get; } + public int Calls { get; private set; } + public Task ForwardAsync(HttpContext context, CancellationToken cancellationToken) + { + Calls++; + return Task.FromResult(Result); + } + } + + private sealed class StubActiveNodeGate : IActiveNodeGate + { + public StubActiveNodeGate(bool isActive) => IsActiveNode = isActive; + public bool IsActiveNode { get; } + } + + private sealed class FakeHandler : HttpMessageHandler + { + private readonly Func> _send; + public FakeHandler(Func> send) => _send = (r, _) => send(r); + public FakeHandler(Func> send) => _send = send; + protected override Task SendAsync(HttpRequestMessage request, CancellationToken ct) => + _send(request, ct); + } + + private sealed class FakeHttpClientFactory : IHttpClientFactory + { + private readonly HttpMessageHandler _handler; + public FakeHttpClientFactory(HttpMessageHandler handler) => _handler = handler; + public HttpClient CreateClient(string name) => + new(_handler, disposeHandler: false) { Timeout = Timeout.InfiniteTimeSpan }; + } + + private sealed class StaticOptionsMonitor : IOptionsMonitor + { + public StaticOptionsMonitor(T value) => CurrentValue = value; + public T CurrentValue { get; } + public T Get(string? name) => CurrentValue; + public IDisposable? OnChange(Action listener) => null; + } +}