feat(inbound-api): standby central node relays POST /api/* to the active peer instead of 503

A caller addressing a standby central node directly (or hitting Traefik inside its
/health/active probe window after a failover) used to get 503 STANDBY_NODE. The
standby now relays the request verbatim to the active peer over HTTP
(IStandbyRequestForwarder / HttpStandbyRequestForwarder) and replays the response;
method scripts and Route.To() still execute only on the active node.

- ScadaBridge:InboundApi:StandbyForwardPeerEndpoints (static list of the OTHER
  central node's HTTP base URL; empty = old 503 behaviour) + StandbyForwardTimeout
  (60s), both startup-validated.
- Peers tried in order: peer 503 stamped X-ScadaBridge-Standby, connection failure
  or timeout → next peer; all fail → 503 ACTIVE_NODE_UNREACHABLE. A 503 without
  the marker is the active node's own answer and is replayed.
- One-hop loop guard via X-ScadaBridge-Forwarded; 413 body cap now runs BEFORE
  the active-node check so a standby never buffers/relays an oversized body.
- AuditWriteMiddleware skips its own ApiInbound row for a relayed request so a
  call is audited once, on the node that executed it.
- Rig + env2 central nodes configured; spec (Component-InboundAPI.md → Standby
  forwarding), Component-Host.md options row, CLAUDE.md updated.
- 20 new tests (filter decision + relay mechanics + validator).

Claude-Session: https://claude.ai/code/session_019ZxXT5sxZs5XrtdwYzJDPV
This commit is contained in:
Joseph Doherty
2026-08-28 13:34:20 -04:00
parent f07ba5cff5
commit a9b23556eb
16 changed files with 839 additions and 31 deletions
@@ -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;
/// <summary>
/// Production <see cref="IStandbyRequestForwarder"/>: relays a standby-node inbound
/// API request to the active central node over plain HTTP using the named
/// <see cref="HttpClientName"/> client.
///
/// <para><b>Peer selection.</b> The standby does not know which peer is active — it
/// only knows it is not. It therefore tries
/// <see cref="InboundApiOptions.StandbyForwardPeerEndpoints"/> 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
/// <see cref="StandbyResponseHeader"/> marker, which the forwarder treats like a
/// connection failure: try the next peer. Connection errors and per-attempt timeouts
/// (<see cref="InboundApiOptions.StandbyForwardTimeout"/>) also move to the next peer.
/// When every peer fails the caller gets 503 <c>ACTIVE_NODE_UNREACHABLE</c>.</para>
///
/// <para><b>Loop guard.</b> Every relayed request carries
/// <see cref="ForwardedRequestHeader"/>. <see cref="InboundApiEndpointFilter"/> 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.</para>
///
/// <para><b>Fidelity.</b> Method, path, query string, body and headers are relayed
/// verbatim (hop-by-hop headers and <c>Host</c> excepted), so the active node
/// authenticates the caller's own <c>Authorization</c>/<c>X-API-Key</c> 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 <see cref="InboundApiOptions.MaxRequestBodyBytes"/>, which the
/// filter enforces before calling this class) so it can be re-sent to a second peer.</para>
/// </summary>
public sealed class HttpStandbyRequestForwarder : IStandbyRequestForwarder
{
/// <summary>Name of the <see cref="IHttpClientFactory"/> client used for relays.</summary>
public const string HttpClientName = "ScadaBridge.InboundApi.StandbyForward";
/// <summary>
/// Request header stamped on every relayed request (value: the forwarding node's
/// host name when known, else <c>1</c>). The filter refuses to forward a request
/// that already carries it.
/// </summary>
public const string ForwardedRequestHeader = "X-ScadaBridge-Forwarded";
/// <summary>
/// Response header the filter adds to its own 503 <c>STANDBY_NODE</c> 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.
/// </summary>
public const string StandbyResponseHeader = "X-ScadaBridge-Standby";
/// <summary>Machine-readable <c>code</c> when no configured peer served the relay.</summary>
public const string ActiveNodeUnreachableCode = "ACTIVE_NODE_UNREACHABLE";
private static readonly HashSet<string> 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<InboundApiOptions> _options;
private readonly ILogger<HttpStandbyRequestForwarder> _logger;
/// <summary>Initializes a new <see cref="HttpStandbyRequestForwarder"/>.</summary>
/// <param name="httpClientFactory">Factory for the named relay client.</param>
/// <param name="options">Live-reloadable inbound API options (peer list + timeout are read per relay).</param>
/// <param name="logger">Logger for relay diagnostics.</param>
public HttpStandbyRequestForwarder(
IHttpClientFactory httpClientFactory,
IOptionsMonitor<InboundApiOptions> options,
ILogger<HttpStandbyRequestForwarder> logger)
{
_httpClientFactory = httpClientFactory;
_options = options;
_logger = logger;
}
/// <inheritdoc />
public bool IsConfigured => _options.CurrentValue.StandbyForwardPeerEndpoints.Count > 0;
/// <inheritdoc />
public async Task<IResult> 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);
/// <summary>
/// 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.
/// </summary>
public sealed class RelayedResult : IResult, IStatusCodeHttpResult
{
/// <summary>Initializes a new <see cref="RelayedResult"/>.</summary>
/// <param name="statusCode">The peer's HTTP status.</param>
/// <param name="contentType">The peer's <c>Content-Type</c>, if any.</param>
/// <param name="body">The peer's response body.</param>
public RelayedResult(int statusCode, string? contentType, byte[] body)
{
StatusCode = statusCode;
ContentType = contentType;
Body = body;
}
/// <inheritdoc />
public int? StatusCode { get; }
/// <summary>The peer's <c>Content-Type</c>, replayed verbatim.</summary>
public string? ContentType { get; }
/// <summary>The peer's response body, replayed verbatim.</summary>
public byte[] Body { get; }
/// <inheritdoc />
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);
}
}
}
@@ -0,0 +1,33 @@
using Microsoft.AspNetCore.Http;
namespace ZB.MOM.WW.ScadaBridge.InboundAPI;
/// <summary>
/// Standby-node forwarding seam for <c>POST /api/{methodName}</c>. When a request
/// lands on a standby central node, <see cref="InboundApiEndpointFilter"/> 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 <c>Route.To()</c> calls still run
/// ONLY on the active node — the standby is a pure relay.
/// </summary>
public interface IStandbyRequestForwarder
{
/// <summary>
/// <c>true</c> when at least one peer endpoint is configured
/// (<see cref="InboundApiOptions.StandbyForwardPeerEndpoints"/>). When <c>false</c>
/// the filter falls back to the pre-forwarding behaviour: HTTP 503 <c>STANDBY_NODE</c>.
/// </summary>
bool IsConfigured { get; }
/// <summary>
/// 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 <c>ACTIVE_NODE_UNREACHABLE</c>.
/// </summary>
/// <param name="context">The inbound HTTP context on the standby node.</param>
/// <param name="cancellationToken">Cancelled when the caller aborts the request.</param>
/// <returns>A result replaying the peer's response, or a 503 when no peer served it.</returns>
Task<IResult> ForwardAsync(HttpContext context, CancellationToken cancellationToken);
}
@@ -12,17 +12,32 @@ namespace ZB.MOM.WW.ScadaBridge.InboundAPI;
///
/// <list type="bullet">
/// <item><description>
/// 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 <c>JsonDocument</c> — and before any relay,
/// so a standby never buffers or forwards a body the active node would refuse.
/// </description></item>
/// <item><description>
/// Request body size cap. Oversized bodies are rejected with HTTP
/// 413 before being buffered into a <c>JsonDocument</c>.
/// Active-node gating. Method scripts execute only on the active central node.
/// A standby RELAYS the request to the active peer through
/// <see cref="IStandbyRequestForwarder"/> when peers are configured
/// (<see cref="InboundApiOptions.StandbyForwardPeerEndpoints"/>), and otherwise
/// returns HTTP 503 <c>STANDBY_NODE</c>. A request that already carries
/// <see cref="HttpStandbyRequestForwarder.ForwardedRequestHeader"/> is never
/// relayed again (one-hop loop guard) — it gets the 503, stamped with
/// <see cref="HttpStandbyRequestForwarder.StandbyResponseHeader"/> so the relaying
/// peer knows to try its next endpoint.
/// </description></item>
/// </list>
/// </summary>
public sealed class InboundApiEndpointFilter : IEndpointFilter
{
/// <summary>
/// <see cref="HttpContext.Items"/> key set when this node relayed the request to the
/// active peer. <c>AuditWriteMiddleware</c> reads it to skip its own audit row (the
/// active node audits the call it executed).
/// </summary>
public const string RelayedToActiveItemKey = "ScadaBridge.InboundApi.RelayedToActive";
private readonly ILogger<InboundApiEndpointFilter> _logger;
private readonly IOptionsMonitor<InboundApiOptions> _options;
@@ -37,32 +52,20 @@ public sealed class InboundApiEndpointFilter : IEndpointFilter
_options = options;
}
/// <summary>Applies active-node gating and request body size checks before delegating to the next filter or handler.</summary>
/// <summary>Applies the request body size check and active-node gating/relay before delegating to the next filter or handler.</summary>
/// <param name="context">The endpoint filter invocation context containing the HTTP context and arguments.</param>
/// <param name="next">The next filter or endpoint handler in the pipeline.</param>
/// <returns>A value task that resolves to the filter pipeline result, or an error result if gating fails.</returns>
/// <returns>A value task that resolves to the filter pipeline result, a relayed peer response, or an error result if gating fails.</returns>
public async ValueTask<object?> 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<IActiveNodeGate>();
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<IActiveNodeGate>();
if (gate is { IsActiveNode: false })
{
var alreadyForwarded = httpContext.Request.Headers
.ContainsKey(HttpStandbyRequestForwarder.ForwardedRequestHeader);
var forwarder = httpContext.RequestServices.GetService<IStandbyRequestForwarder>();
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);
}
}
@@ -48,4 +48,25 @@ public class InboundApiOptions
/// serve minutes-old method definitions.
/// </summary>
public TimeSpan ApiMethodCacheTtl { get; set; } = TimeSpan.FromSeconds(5);
/// <summary>
/// Standby-node forwarding (2026-08-28): the HTTP base URLs of the OTHER central
/// node(s) — e.g. <c>["http://scadabridge-central-b:5000"]</c> on central-a — that a
/// standby relays <c>POST /api/{methodName}</c> to instead of answering 503
/// <c>STANDBY_NODE</c>. 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 <see cref="HttpStandbyRequestForwarder"/>). Each entry
/// must be an absolute <c>http</c>/<c>https</c> URL; never list this node's own address
/// (it would cost one wasted round-trip per call — the loop guard stops it bouncing).
/// </summary>
public List<string> StandbyForwardPeerEndpoints { get; set; } = new();
/// <summary>
/// Per-peer deadline for one relay attempt (connect + the active node's full method
/// execution). Must comfortably exceed <see cref="DefaultMethodTimeout"/> 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.
/// </summary>
public TimeSpan StandbyForwardTimeout { get; set; } = TimeSpan.FromSeconds(60);
}
@@ -31,5 +31,18 @@ public sealed class InboundApiOptionsValidator : OptionsValidatorBase<InboundApi
builder.RequireThat(options.ApiMethodCacheTtl > 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.");
}
}
}
@@ -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,
@@ -35,6 +35,15 @@ public static class ServiceCollectionExtensions
// body size cap and active-node gating for POST /api/{methodName}.
services.AddSingleton<InboundApiEndpointFilter>();
// 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<IStandbyRequestForwarder, HttpStandbyRequestForwarder>();
// 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