using Akka.Actor;
using Akka.TestKit;
using Akka.TestKit.Xunit2;
using Microsoft.CodeAnalysis.CSharp.Scripting;
using Microsoft.CodeAnalysis.Scripting;
using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.ScriptExecution;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Flattening;
using ZB.MOM.WW.ScadaBridge.HealthMonitoring;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Scripts;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests.TestSupport;
namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests.Actors;
///
/// WP3.1 test group 6 — the per-script in-flight cap and its shed policy.
///
/// Before WP3.1 every trigger spawned another run unconditionally: a trigger firing
/// faster than its script completes produced unbounded fan-out onto a bounded thread pool.
/// The cap ( , default 4) sheds the
/// NEWEST run instead. Keeping the four already queued/running — which are closest to their
/// own deadlines and already charged against them — is the policy that never reorders runs
/// and needs no queue at all: the scheduler's FIFO already IS the queue.
///
/// A shed is always counted on the health collector, emits a site event rate-limited to
/// one per script per minute (so a hot trigger cannot flood site_events ), and — for an
/// Ask-based CallScript — replies with an explicit error rather than letting a nested
/// call or inbound-API route hang to its Ask timeout.
///
public class ScriptRunShedTests : TestKit, IDisposable
{
private readonly SharedScriptLibrary _sharedLibrary;
private readonly ScriptExecutionScheduler _scheduler = new(8);
public ScriptRunShedTests()
{
var compilationService = new ScriptCompilationService(
NullLogger.Instance);
_sharedLibrary = new SharedScriptLibrary(
compilationService, NullLogger.Instance);
ShedHooks.Gate = new SemaphoreSlim(0);
}
void IDisposable.Dispose()
{
ShedHooks.Gate.Release(64);
Shutdown();
_scheduler.Dispose();
}
private static Script BlockingScript() => CompileRaw(
"ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests.Actors.ShedHooks.Gate.Wait(); return null;");
private static Script CompileRaw(string code)
{
var options = ScriptOptions.Default
.WithReferences(typeof(object).Assembly, typeof(Enumerable).Assembly,
typeof(ShedHooks).Assembly)
.WithImports("System", "System.Collections.Generic", "System.Linq", "System.Threading.Tasks");
var script = CSharpScript.Create(code, options, typeof(ScriptGlobals));
script.Compile();
return script;
}
private static SiteRuntimeOptions Options() => new()
{
MaxConcurrentRunsPerScript = 4,
// Long enough that nothing times out inside the test window — the cap, not the
// deadline, must be what refuses the fifth run.
ScriptExecutionTimeoutSeconds = 120,
StuckScriptGraceMs = 120_000
};
[Fact]
public void FifthConcurrentRun_IsShed_Counted_EventedOnce_AndAnsweredWithAnError()
{
var siteLog = new FakeSiteEventLogger();
var health = new SiteHealthCollector();
var instance = CreateTestProbe().Ref;
var options = Options();
var actor = ActorOfAsTestActorRef(
Props.Create(() => new ScriptActor(
"Hot", "Inst1", instance, BlockingScript(),
new ResolvedScript { CanonicalName = "Hot", TriggerType = "Call" },
_sharedLibrary, options, NullLogger.Instance,
null, null, health, new SingleServiceProvider(siteLog), _scheduler, null)),
"shed-" + Guid.NewGuid().ToString("N"));
// Fill the cap: four runs, all blocked in their bodies.
for (var i = 0; i < 4; i++)
actor.Tell(new ScriptCallRequest("Hot", null, 0, $"corr-{i}"), ActorRefs.NoSender);
AwaitAssert(() =>
{
Assert.Equal(4, actor.UnderlyingActor.RunsInFlight);
Assert.Equal(4, _scheduler.BusyThreadCount);
}, TimeSpan.FromSeconds(15));
// Fifth: shed. The Ask caller is answered explicitly instead of hanging.
var caller = CreateTestProbe();
actor.Tell(new ScriptCallRequest("Hot", null, 0, "corr-shed-1"), caller.Ref);
var result = caller.ExpectMsg(TimeSpan.FromSeconds(10));
Assert.False(result.Success);
Assert.Equal("corr-shed-1", result.CorrelationId);
Assert.Contains("shed", result.ErrorMessage, StringComparison.OrdinalIgnoreCase);
Assert.Contains("in flight", result.ErrorMessage, StringComparison.OrdinalIgnoreCase);
// Still exactly four in flight — the shed run was never launched.
Assert.Equal(4, actor.UnderlyingActor.RunsInFlight);
// Sixth: counted again, but the Warning site event is rate-limited to one per minute.
actor.Tell(new ScriptCallRequest("Hot", null, 0, "corr-shed-2"), caller.Ref);
var second = caller.ExpectMsg(TimeSpan.FromSeconds(10));
Assert.False(second.Success);
AwaitAssert(() =>
{
var shedEvents = siteLog.OfType("script")
.Where(r => r.Severity == "Warning" && r.Message.Contains("shed"))
.ToArray();
Assert.Single(shedEvents);
Assert.Equal("ScriptActor:Hot", shedEvents[0].Source);
Assert.Equal("Inst1", shedEvents[0].InstanceId);
}, TimeSpan.FromSeconds(5));
// Both sheds were counted on the health report even though only one was evented.
Assert.Equal(2, health.CollectReport("site-1").ScriptRunShedCount);
// One completion frees a slot, and the next trigger launches again.
ShedHooks.Gate.Release();
AwaitAssert(() => Assert.Equal(3, actor.UnderlyingActor.RunsInFlight), TimeSpan.FromSeconds(15));
actor.Tell(new ScriptCallRequest("Hot", null, 0, "corr-after"), ActorRefs.NoSender);
AwaitAssert(() => Assert.Equal(4, actor.UnderlyingActor.RunsInFlight), TimeSpan.FromSeconds(15));
}
[Fact]
public void AlarmOnTriggerRuns_ShedAtTheSameCap_WithAnAlarmScopedSiteEvent()
{
var siteLog = new FakeSiteEventLogger();
var health = new SiteHealthCollector();
var instanceProbe = CreateTestProbe();
var options = Options();
var alarm = ActorOfAsTestActorRef(
Props.Create(() => new AlarmActor(
"Flapper", "Inst1", instanceProbe.Ref,
new ResolvedAlarm
{
CanonicalName = "Flapper",
TriggerType = "ValueMatch",
TriggerConfiguration = "{\"attributeName\":\"Status\",\"matchValue\":\"Critical\"}",
PriorityLevel = 100
},
BlockingScript(), _sharedLibrary, options, NullLogger.Instance,
null, null, health, new SingleServiceProvider(siteLog), null, _scheduler, null)),
"alarm-shed-" + Guid.NewGuid().ToString("N"));
// Each raise edge spawns one on-trigger run; clear between raises to re-arm the edge.
void Flap(int cycle)
{
alarm.Tell(new AttributeValueChanged(
"Inst1", "Status", "Status", "Critical", "Good", DateTimeOffset.UtcNow.AddSeconds(cycle)));
alarm.Tell(new AttributeValueChanged(
"Inst1", "Status", "Status", "Normal", "Good", DateTimeOffset.UtcNow.AddSeconds(cycle)));
}
for (var i = 0; i < 4; i++) Flap(i);
AwaitAssert(() => Assert.Equal(4, alarm.UnderlyingActor.RunsInFlight), TimeSpan.FromSeconds(15));
// Fifth raise is shed — there is no Ask caller on this path, so it surfaces purely as
// a counter plus the rate-limited Warning event.
Flap(4);
Flap(5);
// The shed COUNT assertion lives inside the awaited block, not after it.
// ShedAlarmRun increments the counter and only then emits the (rate-limited)
// site event, and that event fires on the FIRST shed only — so an awaited
// gate on the event observes Flap(4)'s shed and orders nothing whatsoever
// with respect to Flap(5)'s, which is a separate mailbox message with no
// observable of its own on this path (an alarm on-trigger run has no Ask
// caller to reply to — AlarmActor.ShedAlarmRun). Asserting the count bare
// after the gate assumed both sheds had been dequeued by the time the
// first one's event landed; under load the second can still be pending and
// the count reads 1. Reproduced deterministically by deferring Flap(5) by
// 2 s: the bare form failed with "Expected: 2 / Actual: 1", while the
// ScriptActor sibling above — which IS ordered, because ShedRun replies
// ScriptCallResult after incrementing — kept passing.
//
// The count is ACCUMULATED across polls rather than re-read, because
// SiteHealthCollector.CollectReport DRAINS the interval counters
// (Interlocked.Exchange(ref _scriptRunShedCount, 0)) — a poll loop that
// simply re-read it would consume the first shed and never see 2.
var shedCounted = 0;
AwaitAssert(() =>
{
shedCounted += health.CollectReport("site-1").ScriptRunShedCount;
var shedEvents = siteLog.OfType("script")
.Where(r => r.Severity == "Warning" && r.Message.Contains("shed"))
.ToArray();
// Both sheds counted, even though the event is rate-limited to one.
Assert.Equal(2, shedCounted);
Assert.Single(shedEvents);
Assert.Equal("AlarmActor:Flapper", shedEvents[0].Source);
// ...and neither shed run was ever launched: still exactly four in flight.
Assert.Equal(4, alarm.UnderlyingActor.RunsInFlight);
}, TimeSpan.FromSeconds(10));
}
}
/// Test hook used to hold script runs in flight while the cap is exercised.
public static class ShedHooks
{
/// Gate the blocking test scripts wait on; reset per test.
public static SemaphoreSlim Gate = new(0);
}