using Akka.Actor; 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.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 4 — the stuck-script watchdog now REPLACES the thread a wedged script is /// holding, not just names it. /// /// Before WP3.1 the watchdog was observability only: a script blocked in synchronous, /// uninterruptible I/O never observes the cooperative cancellation the timeout requests, so /// the pool simply lost that dedicated thread — permanently, and silently apart from one log /// line. Eight such scripts left the site with no script execution at all. The watchdog now /// detaches the worker (it exits when its body finally returns) and starts a replacement, and /// the count of live detached workers is surfaced on the site health report as /// DetachedScriptThreads. Replacement is capped at the pool size, because bounded /// starvation beats unbounded thread growth when scripts wedge en masse. /// public class StuckScriptWatchdogTests : TestKit, IDisposable { private readonly SharedScriptLibrary _sharedLibrary; public StuckScriptWatchdogTests() { var compilationService = new ScriptCompilationService( NullLogger.Instance); _sharedLibrary = new SharedScriptLibrary( compilationService, NullLogger.Instance); WatchdogHooks.Gate = new SemaphoreSlim(0); } void IDisposable.Dispose() { // Free any still-wedged worker before tearing down, so no test leaves a blocked thread. WatchdogHooks.Gate.Release(8); Shutdown(); } private static Script CompileRaw(string code) { var scriptOptions = ScriptOptions.Default .WithReferences(typeof(object).Assembly, typeof(Enumerable).Assembly, typeof(WatchdogHooks).Assembly) .WithImports("System", "System.Collections.Generic", "System.Linq", "System.Threading.Tasks"); var script = CSharpScript.Create(code, scriptOptions, typeof(ScriptGlobals)); script.Compile(); return script; } /// A body that blocks its worker thread outright — cooperative cancellation is never observed. private static Script WedgeScript() => CompileRaw( "ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests.Actors.WatchdogHooks.Gate.Wait(); return null;"); private IActorRef BuildScriptActor( string name, Script compiled, SiteRuntimeOptions options, ScriptExecutionScheduler scheduler, IServiceProvider? serviceProvider) { var instance = CreateTestProbe().Ref; var config = new ResolvedScript { CanonicalName = name, TriggerType = "Call" }; return ActorOf(Props.Create(() => new ScriptActor( name, "Inst1", instance, compiled, config, _sharedLibrary, options, NullLogger.Instance, null, null, null, serviceProvider, scheduler, null))); } private static SiteRuntimeOptions WatchdogOptions() => new() { ScriptExecutionTimeoutSeconds = 1, StuckScriptGraceMs = 200, MaxScriptCallDepth = 10 }; [Fact] public void WedgedScript_DetachesItsWorker_StartsAReplacement_AndDrainsWhenFreed() { using var scheduler = new ScriptExecutionScheduler(1); var siteLog = new FakeSiteEventLogger(); var options = WatchdogOptions(); var wedged = BuildScriptActor("Wedged", WedgeScript(), options, scheduler, new SingleServiceProvider(siteLog)); wedged.Tell(new ScriptCallRequest("Wedged", null, 0, "corr-wedge"), ActorRefs.NoSender); // Timeout (1 s) + grace (200 ms) later the watchdog fires: the worker is detached and // a replacement thread starts. AwaitAssert( () => Assert.Equal(1, scheduler.DetachedThreadCount), TimeSpan.FromSeconds(15)); AwaitAssert( () => Assert.Contains(siteLog.OfType("script"), r => r.Severity == "Error" && r.Message.Contains("still executing", StringComparison.OrdinalIgnoreCase) && r.Message.Contains("Wedged") && r.Message.Contains("DETACHED")), TimeSpan.FromSeconds(5)); // The replacement worker is live: a fresh script runs even though the pool's only // original thread is still blocked. Before WP3.1 this reply never came. var healthy = BuildScriptActor("Healthy", CompileRaw("return 7;"), options, scheduler, null); var caller = CreateTestProbe(); healthy.Tell(new ScriptCallRequest("Healthy", null, 0, "corr-healthy"), caller.Ref); var result = caller.ExpectMsg(TimeSpan.FromSeconds(15)); Assert.True(result.Success, result.ErrorMessage); Assert.Equal(7, result.ReturnValue); // Freeing the wedged body lets the detached worker exit rather than pull more work, // so the pool does not silently end up double-sized. WatchdogHooks.Gate.Release(); AwaitAssert( () => Assert.Equal(0, scheduler.DetachedThreadCount), TimeSpan.FromSeconds(15)); Assert.Equal(1, scheduler.MaximumConcurrencyLevel); } [Fact] public void AtTheDetachCap_NoFurtherReplacementIsStarted_AndAnErrorEventIsEmitted() { // A one-thread pool caps live detached workers at one, so the SECOND wedge cannot be // replaced — the deliberate "bounded starvation beats unbounded thread growth" trade. using var scheduler = new ScriptExecutionScheduler(1); var siteLog = new FakeSiteEventLogger(); var options = WatchdogOptions(); var first = BuildScriptActor("WedgeOne", WedgeScript(), options, scheduler, new SingleServiceProvider(siteLog)); first.Tell(new ScriptCallRequest("WedgeOne", null, 0, "corr-w1"), ActorRefs.NoSender); AwaitAssert(() => Assert.Equal(1, scheduler.DetachedThreadCount), TimeSpan.FromSeconds(15)); var second = BuildScriptActor("WedgeTwo", WedgeScript(), options, scheduler, new SingleServiceProvider(siteLog)); second.Tell(new ScriptCallRequest("WedgeTwo", null, 0, "corr-w2"), ActorRefs.NoSender); AwaitAssert( () => Assert.Contains(siteLog.OfType("script"), r => r.Severity == "Error" && r.Message.Contains("WedgeTwo") && r.Message.Contains("at cap", StringComparison.OrdinalIgnoreCase)), TimeSpan.FromSeconds(20)); // Still exactly one detached worker: the second wedge was NOT replaced. Assert.Equal(1, scheduler.DetachedThreadCount); WatchdogHooks.Gate.Release(2); } [Fact] public async Task DetachedThreadCount_IsSurfacedOnTheSiteHealthReport() { using var scheduler = new ScriptExecutionScheduler(1); var collector = new SiteHealthCollector(); var wedged = BuildScriptActor("GaugeWedge", WedgeScript(), WatchdogOptions(), scheduler, null); wedged.Tell(new ScriptCallRequest("GaugeWedge", null, 0, "corr-gauge"), ActorRefs.NoSender); AwaitAssert(() => Assert.Equal(1, scheduler.DetachedThreadCount), TimeSpan.FromSeconds(15)); using var reporter = new ScriptSchedulerStatsReporter( collector, WatchdogOptions(), NullLogger.Instance, pollInterval: TimeSpan.FromMilliseconds(50), scheduler: scheduler); await reporter.StartAsync(CancellationToken.None); try { AwaitAssert( () => Assert.Equal(1, collector.CollectReport("site-1").DetachedScriptThreads), TimeSpan.FromSeconds(10)); } finally { await reporter.StopAsync(CancellationToken.None); WatchdogHooks.Gate.Release(); } } } /// /// Test hook the stuck-script watchdog tests use to block a script-execution thread /// deterministically: the compiled body waits on , which the test releases /// once it has observed the detach. /// public static class WatchdogHooks { /// Gate a blocking test script waits on; reset per test. public static SemaphoreSlim Gate = new(0); }