fix(inbound): wire the compiled-handler cache as the IScriptArtifactChangeBus ApiMethod consumer (plan R2-06 T1)
This commit is contained in:
@@ -94,8 +94,12 @@ try
|
|||||||
builder.Services.AddTransport();
|
builder.Services.AddTransport();
|
||||||
// Script-artifact invalidation bus: in-process, per-node pub/sub
|
// Script-artifact invalidation bus: in-process, per-node pub/sub
|
||||||
// for ScriptArtifactsChanged. The Transport bundle importer publishes
|
// for ScriptArtifactsChanged. The Transport bundle importer publishes
|
||||||
// post-commit; the Inbound API compiled-handler cache subscribes as the
|
// post-commit; the Inbound API's ScriptArtifactChangeSubscriber (a hosted
|
||||||
// consumer (wired in plan 06). Central-only — it lives beside the importer.
|
// service registered by AddInboundAPI below) consumes ApiMethod notifications
|
||||||
|
// and invalidates the compiled-handler cache (wired in plan R2-06 T1; the
|
||||||
|
// per-request revision check remains the correctness fallback).
|
||||||
|
// SharedScript/Template notifications currently have NO consumer — nothing at
|
||||||
|
// central caches compiled artifacts of those kinds. Central-only.
|
||||||
builder.Services.AddSingleton<
|
builder.Services.AddSingleton<
|
||||||
ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Scripting.IScriptArtifactChangeBus,
|
ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Scripting.IScriptArtifactChangeBus,
|
||||||
ZB.MOM.WW.ScadaBridge.Host.Services.InProcessScriptArtifactChangeBus>();
|
ZB.MOM.WW.ScadaBridge.Host.Services.InProcessScriptArtifactChangeBus>();
|
||||||
|
|||||||
@@ -0,0 +1,87 @@
|
|||||||
|
using Microsoft.Extensions.Hosting;
|
||||||
|
using Microsoft.Extensions.Logging;
|
||||||
|
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Scripting;
|
||||||
|
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Scripting;
|
||||||
|
|
||||||
|
namespace ZB.MOM.WW.ScadaBridge.InboundAPI;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// N1 (arch-review 06 round 2): the Inbound API consumer of the script-artifact
|
||||||
|
/// invalidation contract (docs/plans/2026-07-08-script-artifact-invalidation-contract.md,
|
||||||
|
/// handoff item (a)). Subscribes to <see cref="IScriptArtifactChangeBus"/> on host
|
||||||
|
/// start and, for every <c>ApiMethod</c> notification (e.g. a Transport bundle
|
||||||
|
/// import overwriting method scripts, published post-commit), drops the compiled
|
||||||
|
/// handler AND the known-bad record via
|
||||||
|
/// <see cref="InboundScriptExecutor.InvalidateMethod"/>, so the next request
|
||||||
|
/// recompiles from the fresh row without waiting for the per-request
|
||||||
|
/// revision-check self-heal.
|
||||||
|
/// <para>
|
||||||
|
/// Fast-path only: the revision check in <c>ExecuteAsync</c> remains the
|
||||||
|
/// correctness fallback — this bus is in-process/per-node and at-least-once
|
||||||
|
/// (contract §4), so a missed or duplicate notification must always be harmless.
|
||||||
|
/// The bus is optional (site roles and plain test hosts register none) — the
|
||||||
|
/// subscriber then no-ops. SharedScript/Template notifications are ignored: this
|
||||||
|
/// executor caches only ApiMethod handlers.
|
||||||
|
/// </para>
|
||||||
|
/// </summary>
|
||||||
|
public sealed class ScriptArtifactChangeSubscriber : IHostedService
|
||||||
|
{
|
||||||
|
private readonly InboundScriptExecutor _executor;
|
||||||
|
private readonly ILogger<ScriptArtifactChangeSubscriber> _logger;
|
||||||
|
private readonly IScriptArtifactChangeBus? _bus;
|
||||||
|
private IDisposable? _subscription;
|
||||||
|
|
||||||
|
/// <summary>Initializes the subscriber.</summary>
|
||||||
|
/// <param name="executor">The compiled-handler cache to invalidate.</param>
|
||||||
|
/// <param name="logger">Logger instance.</param>
|
||||||
|
/// <param name="bus">The change bus, or null when the host registers none (site roles, tests).</param>
|
||||||
|
public ScriptArtifactChangeSubscriber(
|
||||||
|
InboundScriptExecutor executor,
|
||||||
|
ILogger<ScriptArtifactChangeSubscriber> logger,
|
||||||
|
IScriptArtifactChangeBus? bus = null)
|
||||||
|
{
|
||||||
|
_executor = executor;
|
||||||
|
_logger = logger;
|
||||||
|
_bus = bus;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public Task StartAsync(CancellationToken cancellationToken)
|
||||||
|
{
|
||||||
|
if (_bus is null)
|
||||||
|
{
|
||||||
|
_logger.LogDebug(
|
||||||
|
"No IScriptArtifactChangeBus registered; inbound handler-cache freshness relies on the per-request revision check alone.");
|
||||||
|
return Task.CompletedTask;
|
||||||
|
}
|
||||||
|
|
||||||
|
_subscription = _bus.Subscribe(OnChanged);
|
||||||
|
return Task.CompletedTask;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public Task StopAsync(CancellationToken cancellationToken)
|
||||||
|
{
|
||||||
|
_subscription?.Dispose();
|
||||||
|
_subscription = null;
|
||||||
|
return Task.CompletedTask;
|
||||||
|
}
|
||||||
|
|
||||||
|
private void OnChanged(ScriptArtifactsChanged notification)
|
||||||
|
{
|
||||||
|
if (!string.Equals(
|
||||||
|
notification.ArtifactKind, ScriptArtifactKinds.ApiMethod, StringComparison.Ordinal))
|
||||||
|
{
|
||||||
|
return; // only ApiMethod handlers are cached by this executor
|
||||||
|
}
|
||||||
|
|
||||||
|
foreach (var name in notification.Names)
|
||||||
|
{
|
||||||
|
_executor.InvalidateMethod(name);
|
||||||
|
}
|
||||||
|
|
||||||
|
_logger.LogInformation(
|
||||||
|
"Invalidated {Count} inbound API method handler(s) after a {Source} script-artifact change.",
|
||||||
|
notification.Names.Count, notification.Source);
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -27,6 +27,12 @@ public static class ServiceCollectionExtensions
|
|||||||
// body size cap and active-node gating for POST /api/{methodName}.
|
// body size cap and active-node gating for POST /api/{methodName}.
|
||||||
services.AddSingleton<InboundApiEndpointFilter>();
|
services.AddSingleton<InboundApiEndpointFilter>();
|
||||||
|
|
||||||
|
// 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
|
||||||
|
// and the executor's per-request revision check alone keeps handlers fresh.
|
||||||
|
services.AddHostedService<ScriptArtifactChangeSubscriber>();
|
||||||
|
|
||||||
return services;
|
return services;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,133 @@
|
|||||||
|
using Microsoft.Extensions.Logging.Abstractions;
|
||||||
|
using NSubstitute;
|
||||||
|
using ZB.MOM.WW.ScadaBridge.Commons.Entities.InboundApi;
|
||||||
|
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Scripting;
|
||||||
|
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
|
||||||
|
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Scripting;
|
||||||
|
|
||||||
|
namespace ZB.MOM.WW.ScadaBridge.InboundAPI.Tests;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Arch-review R2 N1: the Inbound API must be a REAL subscriber of
|
||||||
|
/// IScriptArtifactChangeBus — a bundle-import publish must invalidate a cached
|
||||||
|
/// handler immediately, without waiting for the per-request revision-check
|
||||||
|
/// self-heal (which remains the correctness fallback per the invalidation
|
||||||
|
/// contract's normative semantics).
|
||||||
|
/// </summary>
|
||||||
|
public class ScriptArtifactChangeSubscriberTests
|
||||||
|
{
|
||||||
|
/// <summary>Minimal in-test bus: records subscriptions, delivers synchronously.</summary>
|
||||||
|
private sealed class RecordingBus : IScriptArtifactChangeBus
|
||||||
|
{
|
||||||
|
private readonly List<Action<ScriptArtifactsChanged>> _handlers = new();
|
||||||
|
public int SubscriberCount => _handlers.Count;
|
||||||
|
|
||||||
|
public void Publish(ScriptArtifactsChanged notification)
|
||||||
|
{
|
||||||
|
foreach (var handler in _handlers.ToArray()) handler(notification);
|
||||||
|
}
|
||||||
|
|
||||||
|
public IDisposable Subscribe(Action<ScriptArtifactsChanged> handler)
|
||||||
|
{
|
||||||
|
_handlers.Add(handler);
|
||||||
|
return new Token(() => _handlers.Remove(handler));
|
||||||
|
}
|
||||||
|
|
||||||
|
private sealed class Token(Action dispose) : IDisposable
|
||||||
|
{
|
||||||
|
public void Dispose() => dispose();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private readonly InboundScriptExecutor _executor = new(
|
||||||
|
NullLogger<InboundScriptExecutor>.Instance, Substitute.For<IServiceProvider>());
|
||||||
|
private readonly RecordingBus _bus = new();
|
||||||
|
private readonly RouteHelper _route = new(
|
||||||
|
Substitute.For<IInstanceLocator>(), Substitute.For<IInstanceRouter>());
|
||||||
|
|
||||||
|
private ScriptArtifactChangeSubscriber CreateSubscriber(IScriptArtifactChangeBus? bus) =>
|
||||||
|
new(_executor, NullLogger<ScriptArtifactChangeSubscriber>.Instance, bus);
|
||||||
|
|
||||||
|
private Task<InboundScriptResult> Run(ApiMethod m) => _executor.ExecuteAsync(
|
||||||
|
m, new Dictionary<string, object?>(), _route, TimeSpan.FromSeconds(10));
|
||||||
|
|
||||||
|
private static ScriptArtifactsChanged ApiMethodChanged(params string[] names) =>
|
||||||
|
new(ScriptArtifactKinds.ApiMethod, names, "BundleImport", DateTimeOffset.UtcNow);
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task ApiMethodPublish_InvalidatesCachedHandler_WithoutRevisionCheckSelfHeal()
|
||||||
|
{
|
||||||
|
var subscriber = CreateSubscriber(_bus);
|
||||||
|
await subscriber.StartAsync(CancellationToken.None);
|
||||||
|
|
||||||
|
// Pin a SCRIPT-LESS handler: the RegisterHandler seam is exempt from the
|
||||||
|
// per-request revision check, so the ONLY mechanism that can stop it
|
||||||
|
// serving is an explicit InvalidateMethod — exactly what the bus must trigger.
|
||||||
|
_executor.RegisterHandler("m", async _ => { await Task.CompletedTask; return 99; });
|
||||||
|
var row = new ApiMethod("m", "return 1;") { Id = 1, TimeoutSeconds = 10 };
|
||||||
|
|
||||||
|
var before = await Run(row);
|
||||||
|
Assert.Contains("99", before.ResultJson); // pinned handler serves; no self-heal fires
|
||||||
|
|
||||||
|
_bus.Publish(ApiMethodChanged("m"));
|
||||||
|
|
||||||
|
var after = await Run(row);
|
||||||
|
Assert.True(after.Success);
|
||||||
|
Assert.Contains("1", after.ResultJson); // recompiled from the fresh row — invalidation, not self-heal
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task ApiMethodPublish_PurgesKnownBadRecord()
|
||||||
|
{
|
||||||
|
var subscriber = CreateSubscriber(_bus);
|
||||||
|
await subscriber.StartAsync(CancellationToken.None);
|
||||||
|
|
||||||
|
var broken = new ApiMethod("bad", "%%% not C# %%%") { Id = 2, TimeoutSeconds = 10 };
|
||||||
|
Assert.False(_executor.CompileAndRegister(broken));
|
||||||
|
Assert.Equal(1, _executor.KnownBadMethodCount);
|
||||||
|
|
||||||
|
_bus.Publish(ApiMethodChanged("bad"));
|
||||||
|
|
||||||
|
Assert.Equal(0, _executor.KnownBadMethodCount);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task NonApiMethodKinds_AreIgnored()
|
||||||
|
{
|
||||||
|
var subscriber = CreateSubscriber(_bus);
|
||||||
|
await subscriber.StartAsync(CancellationToken.None);
|
||||||
|
|
||||||
|
_executor.RegisterHandler("m", async _ => { await Task.CompletedTask; return 99; });
|
||||||
|
var row = new ApiMethod("m", "return 1;") { Id = 3, TimeoutSeconds = 10 };
|
||||||
|
|
||||||
|
_bus.Publish(new ScriptArtifactsChanged(
|
||||||
|
ScriptArtifactKinds.SharedScript, new[] { "m" }, "BundleImport", DateTimeOffset.UtcNow));
|
||||||
|
_bus.Publish(new ScriptArtifactsChanged(
|
||||||
|
ScriptArtifactKinds.Template, new[] { "m" }, "BundleImport", DateTimeOffset.UtcNow));
|
||||||
|
|
||||||
|
var result = await Run(row);
|
||||||
|
Assert.Contains("99", result.ResultJson); // pinned handler untouched
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task StopAsync_Unsubscribes()
|
||||||
|
{
|
||||||
|
var subscriber = CreateSubscriber(_bus);
|
||||||
|
await subscriber.StartAsync(CancellationToken.None);
|
||||||
|
Assert.Equal(1, _bus.SubscriberCount);
|
||||||
|
|
||||||
|
await subscriber.StopAsync(CancellationToken.None);
|
||||||
|
|
||||||
|
Assert.Equal(0, _bus.SubscriberCount);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task NullBus_NoOps()
|
||||||
|
{
|
||||||
|
// Site roles / plain test hosts have no bus registered — the hosted
|
||||||
|
// service must start and stop cleanly (the revision check alone applies).
|
||||||
|
var subscriber = CreateSubscriber(bus: null);
|
||||||
|
await subscriber.StartAsync(CancellationToken.None);
|
||||||
|
await subscriber.StopAsync(CancellationToken.None);
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user