fix(inbound): background active-node-only method warm-up service (review 2026-10-21 05/U3)

This commit is contained in:
Joseph Doherty
2026-10-22 08:41:55 -04:00
parent 4921ebb0c8
commit fa73ddb453
4 changed files with 163 additions and 2 deletions
+2 -2
View File
@@ -175,7 +175,7 @@ The `sbk_<keyId>_<secret>` design already supports zero-downtime rotation — a
## Script Compilation & Hot-Reload
API method scripts are compiled at central startup — all method definitions are loaded from the configuration database and compiled into in-memory delegates.
API method scripts are pre-compiled in the **background after startup** by `InboundMethodWarmupService` (a hosted service registered after `AkkaHostedService`, so it never delays the cluster join or `/health/*`): it waits for `ApplicationStarted`, then — only once this node is the active central node (`IActiveNodeGate`; a standby pays nothing until a failover promotes it) — loads all method definitions and compiles them into in-memory delegates. Lazy compile + the per-request revision check guarantee correctness regardless. Compiles are **single-flight per method name + script revision**: N concurrent first callers share one compile (and one non-collectible loaded assembly). `InboundScriptExecutor.CompileAndRegister(ApiMethod, out IReadOnlyList<string> errors)` compiles once and returns the diagnostics, so a save needs no separate validate pass. `ApiMethodCache` prunes expired entries on a miss once large (scopes are exact method names — pinned by test — so growth is bounded).
### Update Workflow
@@ -367,7 +367,7 @@ A standby now **relays** the request to the active peer instead (`InboundApiEndp
- **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 total relay budget and should exceed the longest method timeout plus network overhead. 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 — the options snapshot is read per relay.
- **Pure relay.** Method, path, query string, body and credentials travel unchanged; hop-by-hop/Host/forwarding-provenance headers are excluded and authenticated provenance is rebuilt, 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 serving node owns its `ApiInbound` row; the originating standby retains a failure row when no peer serves or the outcome is uncertain. Verified internal standby refusals are not duplicated.
- **Order of checks.** The body-size cap (HTTP 413) runs before audit reads and 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.
- **Order of checks.** The body-size cap (HTTP 413) runs before audit reads and the active-node check, so a standby never buffers or relays a body the active node would refuse. A relaying standby skips the audit middleware's request/response capture entirely; the body is read once, by the forwarder (bounded by `MaxRequestBodyBytes`), so it can be re-sent to a second peer. On the node that executes, request bodies up to the 1 MiB cap are buffered in memory (`EnableBuffering` with the cap as threshold), not spilled to a temp file. The relay proof (`X-ScadaBridge-Relay`) carries a random nonce, is signed fresh per peer attempt, and is **single-use** within its 300 s window: the receiver keeps a bounded `(ExecutionId, nonce)` set and treats a repeat (or a nonce-less proof from an older peer) as unverified — the request still runs, but a captured proof cannot re-stamp ExecutionId/IP.
- **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`, proven name-resolution failure, or connection refusal permits trying the next peer. Timeout or ambiguous response loss returns 502 `OUTCOME_UNKNOWN` without replay. 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: <relaying host>`. 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.
@@ -0,0 +1,89 @@
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
namespace ZB.MOM.WW.ScadaBridge.InboundAPI;
/// <summary>
/// Pre-compiles every inbound API method script in the background AFTER the host has started,
/// so the warm-up never delays the cluster join or <c>/health/*</c> (review 2026-10-21 05/U3).
/// It used to run synchronously in <c>Program.cs</c> before <c>app.RunAsync()</c>, i.e. before
/// any hosted service started.
/// <para>
/// A node only executes inbound methods while it is the active central node, so the warm-up
/// waits (polling <see cref="IActiveNodeGate"/>) until this node is active and then runs once;
/// a standby therefore pays nothing until a failover promotes it. The per-request lazy compile
/// plus revision check in <see cref="InboundScriptExecutor"/> remains the correctness
/// guarantee — this is purely an optimisation for the first requests.
/// </para>
/// </summary>
public sealed class InboundMethodWarmupService : BackgroundService
{
private static readonly TimeSpan ActivePollInterval = TimeSpan.FromSeconds(5);
private readonly IServiceScopeFactory _scopeFactory;
private readonly InboundScriptExecutor _executor;
private readonly IHostApplicationLifetime _lifetime;
private readonly IActiveNodeGate? _gate;
private readonly ILogger<InboundMethodWarmupService> _logger;
/// <summary>Initializes the warm-up service.</summary>
/// <param name="scopeFactory">Creates the scope used to read the method list.</param>
/// <param name="executor">The executor whose handler cache is warmed.</param>
/// <param name="lifetime">Host lifetime; the warm-up starts only after ApplicationStarted.</param>
/// <param name="logger">Logger for warm-up diagnostics.</param>
/// <param name="gate">Optional active-node gate; absent means always active.</param>
public InboundMethodWarmupService(
IServiceScopeFactory scopeFactory,
InboundScriptExecutor executor,
IHostApplicationLifetime lifetime,
ILogger<InboundMethodWarmupService> logger,
IActiveNodeGate? gate = null)
{
_scopeFactory = scopeFactory;
_executor = executor;
_lifetime = lifetime;
_logger = logger;
_gate = gate;
}
/// <inheritdoc />
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
try
{
var started = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
using (_lifetime.ApplicationStarted.Register(() => started.TrySetResult()))
using (stoppingToken.Register(() => started.TrySetCanceled(stoppingToken)))
{
await started.Task.ConfigureAwait(false);
}
while (_gate is { IsActiveNode: false })
await Task.Delay(ActivePollInterval, stoppingToken).ConfigureAwait(false);
await Task.Run(() => WarmAsync(stoppingToken), stoppingToken).ConfigureAwait(false);
}
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
{
// shutting down
}
catch (Exception ex)
{
// Best effort: lazy compile on first request still guarantees correctness.
_logger.LogWarning(ex, "Inbound API method warm-up failed; methods will compile on first request.");
}
}
private async Task WarmAsync(CancellationToken ct)
{
using var scope = _scopeFactory.CreateScope();
var repo = scope.ServiceProvider.GetRequiredService<IInboundApiRepository>();
var methods = await repo.GetAllApiMethodsAsync(ct).ConfigureAwait(false);
// CompileAndRegister is safe for concurrent callers and single-flights per revision.
Parallel.ForEach(methods, new ParallelOptions { CancellationToken = ct },
method => _executor.CompileAndRegister(method));
_logger.LogInformation("Inbound API warm-up compiled {Count} method(s)", methods.Count);
}
}
@@ -59,6 +59,9 @@ public class InboundScriptExecutor
private const int KnownBadMethodsCap = 1000;
private readonly ConcurrentDictionary<string, string> _knownBadMethods = new();
/// <summary>Diagnostic helper — whether a compiled handler is cached for the method.</summary>
internal bool HasHandler(string methodName) => _scriptHandlers.ContainsKey(methodName);
/// <summary>
/// Diagnostic helper — returns the current size of the
/// known-bad-methods cache so tests can assert the cap is honoured. Internal
@@ -0,0 +1,69 @@
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging.Abstractions;
using NSubstitute;
using ZB.MOM.WW.ScadaBridge.Commons.Entities.InboundApi;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
namespace ZB.MOM.WW.ScadaBridge.InboundAPI.Tests;
/// <summary>Tests for <see cref="InboundMethodWarmupService"/> (review 2026-10-21 05/U3).</summary>
public class InboundMethodWarmupServiceTests
{
private sealed class Lifetime : IHostApplicationLifetime
{
public CancellationTokenSource StartedCts { get; } = new();
public CancellationToken ApplicationStarted => StartedCts.Token;
public CancellationToken ApplicationStopping => CancellationToken.None;
public CancellationToken ApplicationStopped => CancellationToken.None;
public void StopApplication() { }
}
private sealed class Gate(bool active) : IActiveNodeGate
{
public bool IsActiveNode => active;
}
private static (InboundMethodWarmupService svc, InboundScriptExecutor exec, Lifetime life) Build(bool active)
{
var repo = Substitute.For<IInboundApiRepository>();
repo.GetAllApiMethodsAsync(Arg.Any<CancellationToken>())
.Returns(new List<ApiMethod> { new("warm", "return 1;") { Id = 1, TimeoutSeconds = 10 } });
var provider = new ServiceCollection().AddSingleton(repo).BuildServiceProvider();
var exec = new InboundScriptExecutor(NullLogger<InboundScriptExecutor>.Instance, Substitute.For<IServiceProvider>());
var life = new Lifetime();
var svc = new InboundMethodWarmupService(
provider.GetRequiredService<IServiceScopeFactory>(), exec, life,
NullLogger<InboundMethodWarmupService>.Instance, new Gate(active));
return (svc, exec, life);
}
/// <summary>Verifies the warm-up waits for ApplicationStarted and then compiles on an active node.</summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task ActiveNode_CompilesAfterApplicationStarted_NotBefore()
{
var (svc, exec, life) = Build(active: true);
await svc.StartAsync(CancellationToken.None);
await Task.Delay(200);
Assert.False(exec.HasHandler("warm"));
life.StartedCts.Cancel();
for (var i = 0; i < 100 && !exec.HasHandler("warm"); i++) await Task.Delay(50);
Assert.True(exec.HasHandler("warm"));
await svc.StopAsync(CancellationToken.None);
}
/// <summary>Verifies a standby node does not warm up.</summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task StandbyNode_DoesNotCompile()
{
var (svc, exec, life) = Build(active: false);
await svc.StartAsync(CancellationToken.None);
life.StartedCts.Cancel();
await Task.Delay(500);
Assert.False(exec.HasHandler("warm"));
await svc.StopAsync(CancellationToken.None);
}
}