diff --git a/docs/requirements/Component-AuditLog.md b/docs/requirements/Component-AuditLog.md index b9364d22..7b85c8a2 100644 --- a/docs/requirements/Component-AuditLog.md +++ b/docs/requirements/Component-AuditLog.md @@ -196,7 +196,9 @@ known spawn points: `ExecutionId` as its `ParentExecutionId`, while the inbound `InboundRequest` row is top-level (`ParentExecutionId` NULL). - **Alarm-triggered on-trigger script** — when a write trips an alarm and its - on-trigger script runs (via `AlarmActor → AlarmExecutionActor`), the run + on-trigger script runs (via `AlarmActor → ScriptRunLauncher.LaunchAlarmScript`, + which builds the run's `ScriptRuntimeContext` directly on the script-execution + pool — the per-run `AlarmExecutionActor` was removed by WP3.1), the run records the **writing execution's** `ExecutionId` as its `ParentExecutionId`. The write's originating execution rides site-locally from `ScriptRuntimeContext.SetAttribute` (or the inbound API's diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteEventLogging/SiteEventLogger.cs b/src/ZB.MOM.WW.ScadaBridge.SiteEventLogging/SiteEventLogger.cs index 169477c0..952be959 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteEventLogging/SiteEventLogger.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteEventLogging/SiteEventLogger.cs @@ -260,14 +260,24 @@ public class SiteEventLogger : ISiteEventLogger, IDisposable } /// - /// Commits a drained batch of pending events in one transaction. All-or-nothing per - /// batch: site_events ids are freshly minted GUIDs per - /// call (no cross-call replay/de-dup concern the way SqliteAuditWriter has), so - /// unlike that reference there is no per-row duplicate-key swallow here — any failure - /// rolls back the whole batch and faults every event in it. site_events is a - /// CDC-replicated table (LocalDb Phase 1); its capture triggers fire per row on INSERT - /// same as before, so batching the commit changes only how many rows land per - /// transaction, not per-row trigger/replication behavior. + /// Commits a drained batch of pending events in one transaction, falling back to + /// row-by-row inserts if that transaction fails. + /// + /// The batch is the fast path: one transaction amortises SQLite's fsync/journal cost + /// across the whole drain. But a batch is all-or-nothing, so before the fallback existed a + /// SINGLE unwritable row (an oversized details payload, a constraint/trigger reject) + /// faulted all ~ of its innocent neighbours and inflated + /// — the health metric — by the same factor. The rollback + /// is therefore followed by a per-row retry: one poison row costs exactly one row, the + /// rest still land, and the failure count stays truthful. Row-by-row is the degraded + /// path only — it is entered solely after a batch has already failed. + /// + /// site_events ids are freshly minted GUIDs per + /// call (no cross-call replay/de-dup concern the way SqliteAuditWriter has), so a + /// re-insert during the fallback cannot duplicate a row the rolled-back batch had already + /// written. site_events is a CDC-replicated table (LocalDb Phase 1); its capture + /// triggers fire per row on INSERT either way, so neither shape changes + /// trigger/replication behavior. /// private void FlushBatch(IReadOnlyList batch) { @@ -283,10 +293,7 @@ public class SiteEventLogger : ISiteEventLogger, IDisposable { using var cmd = connection.CreateCommand(); cmd.Transaction = transaction; - cmd.CommandText = """ - INSERT INTO site_events (id, timestamp, event_type, severity, instance_id, source, message, details) - VALUES ($id, $timestamp, $event_type, $severity, $instance_id, $source, $message, $details) - """; + cmd.CommandText = InsertEventSql; var pId = cmd.Parameters.Add("$id", SqliteType.Text); var pTimestamp = cmd.Parameters.Add("$timestamp", SqliteType.Text); var pEventType = cmd.Parameters.Add("$event_type", SqliteType.Text); @@ -327,16 +334,21 @@ public class SiteEventLogger : ISiteEventLogger, IDisposable if (failure is not null) { - // A write failure must be observable. Count every event in the failed - // batch (Health Monitoring reads FailedWriteCount) and fault each - // caller's Task instead of silently discarding the exception. - Interlocked.Add(ref _failedWriteCount, batch.Count); - _logger.LogError(failure, "Failed to record {Count} event(s) in batch (sqlite {SqliteError})", - batch.Count, DescribeSqliteError(failure)); - foreach (var pending in batch) + if (batch.Count == 1) { - pending.Completion.TrySetException(failure); + // Nothing to isolate — retrying the single row would fail identically. + Interlocked.Increment(ref _failedWriteCount); + _logger.LogError(failure, "Failed to record 1 event (sqlite {SqliteError})", + DescribeSqliteError(failure)); + batch[0].Completion.TrySetException(failure); + return; } + + // Degraded path: isolate the failure instead of charging the whole batch for it. + _logger.LogWarning(failure, + "Batch insert of {Count} site event(s) failed (sqlite {SqliteError}); retrying row by row", + batch.Count, DescribeSqliteError(failure)); + FlushRowByRow(batch); return; } @@ -352,8 +364,7 @@ public class SiteEventLogger : ISiteEventLogger, IDisposable // WithConnection returns false only when the logger has been // disposed mid-drain; none of the batch was persisted. Fault every // Task instead of reporting false success for a dropped event. - var disposedEx = new ObjectDisposedException(nameof(SiteEventLogger), - "Event could not be recorded: the event logger was disposed before the write completed."); + var disposedEx = DisposedMidWrite(); foreach (var pending in batch) { pending.Completion.TrySetException(disposedEx); @@ -361,16 +372,111 @@ public class SiteEventLogger : ISiteEventLogger, IDisposable } } + /// + /// Re-inserts a failed batch one row at a time, each in its own implicit transaction, so a + /// single unwritable event costs exactly one event: its own Task faults and it alone is + /// counted in , while every other row in the batch still + /// lands and completes successfully. Only reached from after the + /// batched transaction has already rolled back. + /// + /// The rolled-back batch to retry row by row. + private void FlushRowByRow(IReadOnlyList batch) + { + var failed = 0; + + foreach (var pending in batch) + { + Exception? rowFailure = null; + bool written; + try + { + written = WithConnection(connection => + { + using var cmd = connection.CreateCommand(); + cmd.CommandText = InsertEventSql; + cmd.Parameters.AddWithValue("$id", pending.Id); + cmd.Parameters.AddWithValue("$timestamp", pending.Timestamp); + cmd.Parameters.AddWithValue("$event_type", pending.EventType); + cmd.Parameters.AddWithValue("$severity", pending.Severity); + cmd.Parameters.AddWithValue("$instance_id", (object?)pending.InstanceId ?? DBNull.Value); + cmd.Parameters.AddWithValue("$source", pending.Source); + cmd.Parameters.AddWithValue("$message", pending.Message); + cmd.Parameters.AddWithValue("$details", (object?)pending.Details ?? DBNull.Value); + cmd.ExecuteNonQuery(); + }); + } + catch (Exception ex) + { + rowFailure = ex; + written = false; + } + + if (written) + { + pending.Completion.TrySetResult(); + continue; + } + + failed++; + if (rowFailure is null) + { + // Disposed mid-fallback; the remaining rows will take this path too. + pending.Completion.TrySetException(DisposedMidWrite()); + continue; + } + + _logger.LogError(rowFailure, + "Failed to record event from {Source} on row-by-row retry (sqlite {SqliteError})", + pending.Source, DescribeSqliteError(rowFailure)); + pending.Completion.TrySetException(rowFailure); + } + + if (failed > 0) + { + // Health Monitoring reads FailedWriteCount: only the rows that could not be + // written are counted, so the metric measures poison rows, not batch size. + Interlocked.Add(ref _failedWriteCount, failed); + } + } + + /// + /// The exception a pending event's Task is faulted with when the logger was disposed + /// before its write could complete. A fresh instance per call so each faulted Task carries + /// its own stack. + /// + /// The exception to fault the pending event with. + private static ObjectDisposedException DisposedMidWrite() => + new(nameof(SiteEventLogger), + "Event could not be recorded: the event logger was disposed before the write completed."); + + /// + /// The single INSERT shared by the batched fast path and the row-by-row fallback, so the + /// two shapes can never drift in column list or ordering. + /// + private const string InsertEventSql = """ + INSERT INTO site_events (id, timestamp, event_type, severity, instance_id, source, message, details) + VALUES ($id, $timestamp, $event_type, $severity, $instance_id, $source, $message, $details) + """; + /// /// Stops accepting new events, drains the write queue, and disposes the SQLite connection. /// + /// + /// _disposed is deliberately NOT set in the first lock block. Completing + /// the channel writer is the whole shutdown signal: observes it + /// (TryWrite returns false) and the writer loop drains what is already buffered + /// before exiting. Flipping _disposed up front would make + /// return false for every remaining , so a graceful shutdown would + /// discard the entire queued backlog — the very events (shutdown diagnostics) most worth + /// keeping. It flips only after the loop has fully drained, where it guards the window + /// between drain and connection close. Mirrors SqliteAuditWriter.DisposeAsync. + /// public void Dispose() { - Task? writerLoop = null; + Task? writerLoop; lock (_writeLock) { if (_disposed) return; - _disposed = true; // Stop accepting new events and let the writer loop drain. _writeQueue.Writer.TryComplete(); writerLoop = _writerLoop; @@ -390,6 +496,8 @@ public class SiteEventLogger : ISiteEventLogger, IDisposable lock (_writeLock) { + if (_disposed) return; + _disposed = true; _connection.Dispose(); } } diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/ScriptActor.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/ScriptActor.cs index a8a6e71c..195d4176 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/ScriptActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/ScriptActor.cs @@ -526,6 +526,18 @@ public class ScriptActor : ReceiveActor, IWithTimers /// but the script body runs on the bounded set of dedicated threads, so blocking script /// I/O is contained there and cannot starve the shared .NET thread pool. No per-run child /// actor is created. + /// + /// The cap governs NEW work only ( 0 — trigger-driven + /// runs and depth-0 Ask calls). A nested CallScript is exempt: the calling run is + /// itself still counted in while it awaits its callee (the + /// slot is released only by , sent after the body + /// returns), and a script calling ITSELF routes back to this same actor — so counting the + /// nested launch against the cap would refuse legitimate self-recursion at depth + /// with a misleading "shed", + /// making — the limit that actually + /// owns this path, enforced in ScriptRuntimeContext.CallScript — unreachable. + /// Nested depth is bounded by MaxScriptCallDepth instead, which is what the cap would + /// otherwise be doing badly. /// private void SpawnExecution( IReadOnlyDictionary? parameters, @@ -534,7 +546,7 @@ public class ScriptActor : ReceiveActor, IWithTimers string correlationId, Guid? parentExecutionId = null) { - if (_runsInFlight >= _options.MaxConcurrentRunsPerScript) + if (callDepth == 0 && _runsInFlight >= _options.MaxConcurrentRunsPerScript) { ShedRun(replyTo, correlationId); return; @@ -604,8 +616,10 @@ public class ScriptActor : ReceiveActor, IWithTimers /// runs are already in flight. /// The four already queued/running are kept — they are closest to their own deadlines and /// already charged against them — so nothing is ever reordered. A trigger-driven run is - /// simply not launched; an Ask-based CallScript gets an explicit error so a nested - /// call or inbound-API route fails fast rather than hanging to its Ask timeout. + /// simply not launched; a depth-0 Ask (an inbound-API route, or a CallScript from + /// an unrelated script's run) gets an explicit error so the caller fails fast rather than + /// hanging to its Ask timeout. Nested (callDepth > 0) launches never reach here — + /// see . /// private void ShedRun(IActorRef replyTo, string correlationId) { diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Scripts/ScriptExecutionScheduler.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Scripts/ScriptExecutionScheduler.cs index eebbaa34..8caf4ca4 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Scripts/ScriptExecutionScheduler.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Scripts/ScriptExecutionScheduler.cs @@ -289,11 +289,18 @@ public sealed class ScriptExecutionScheduler : TaskScheduler, IDisposable if (Volatile.Read(ref worker.RunStamp) != observedRunStamp) return WorkerDetachOutcome.NotRunning; if (Volatile.Read(ref worker.Detached) != 0) return WorkerDetachOutcome.NotRunning; - // Bound: never hold more than 2x threads (N wedged + N live). - if (_detachedLive >= _configuredCount) return WorkerDetachOutcome.AtCap; + // Bound: never hold more than 2x threads (N wedged + N live). Volatile read — + // the decrement side runs outside _growLock (see below). + if (Volatile.Read(ref _detachedLive) >= _configuredCount) return WorkerDetachOutcome.AtCap; Volatile.Write(ref worker.Detached, 1); - _detachedLive++; + // Interlocked, NOT ++: the matching decrement in WorkerLoop is lock-free (the + // exiting worker never touches _growLock), so a plain read-modify-write here can + // lose that concurrent decrement. The drift is upward-only and permanent — the + // gauge over-reports and TryDetachWorker starts returning AtCap while real + // capacity is available. The surrounding sequence keeps the lock; only this one + // field is shared with a lock-free writer. + Interlocked.Increment(ref _detachedLive); _configuredCount--; // the detached worker no longer counts towards the pool … GrowTo(1); // … and GrowTo puts the count back by starting its replacement. return WorkerDetachOutcome.Detached; diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Scripts/ScriptRunSummaryRecorder.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Scripts/ScriptRunSummaryRecorder.cs index 92196677..c07354db 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Scripts/ScriptRunSummaryRecorder.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Scripts/ScriptRunSummaryRecorder.cs @@ -16,10 +16,18 @@ namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Scripts; /// /// Lock-free: each script's counters are a small class of long fields mutated /// via , reached through a -/// that atomically swaps out () so a -/// concurrent increment either lands in the snapshot being flushed or the fresh one that -/// replaces it — never lost, never double-counted, and the flush never blocks a recording -/// call (or vice versa). +/// that atomically swaps out (), so +/// the flush never blocks a recording call (or vice versa). +/// +/// Accounting is BEST-EFFORT, not exact. A recorder that has already resolved its +/// object from the pre-swap dictionary can increment it after the swap +/// has snapshotted it, so that increment lands in a window that has already been summarised +/// and is never reported. These are operational volume counters for one interval row, not a +/// ledger: an occasional count landing in neither window is acceptable, and nothing +/// downstream reconciles totals across intervals. What IS guaranteed is that a count is never +/// double-reported (each instance is summarised by exactly one flush) +/// and that runs spanning a flush boundary — started in window N, completed in N+1 — still +/// emit their completion, because the flush filter admits a completion-only entry. /// public sealed class ScriptRunSummaryRecorder { @@ -99,8 +107,8 @@ public sealed class ScriptRunSummaryRecorder } /// - /// Snapshots and resets the accumulated counters, then — if any script ran, failed, or - /// timed out during the interval — emits ONE aggregate "script" Info site event + /// Snapshots and resets the accumulated counters, then — if any script started, completed, + /// failed, or timed out during the interval — emits ONE aggregate "script" Info site event /// summarizing the window. An interval with zero activity emits no row (the standby, /// which never runs scripts, therefore never produces a summary row naturally). /// @@ -119,8 +127,13 @@ public sealed class ScriptRunSummaryRecorder ref _counters, new ConcurrentDictionary<(string InstanceName, string ScriptName), Counters>()); + // Completed > 0 is admitted on its own: a run that started in window N and finished in + // N+1 records its completion against a FRESH counters object whose Started is 0, so + // filtering on Started/Failed/TimedOut alone would silently drop that completion (and, + // with it, the run's duration) from every summary row that could ever carry it. var entries = snapshot - .Where(kvp => kvp.Value.Started > 0 || kvp.Value.Failed > 0 || kvp.Value.TimedOut > 0) + .Where(kvp => kvp.Value.Started > 0 || kvp.Value.Completed > 0 || + kvp.Value.Failed > 0 || kvp.Value.TimedOut > 0) .ToList(); if (entries.Count == 0) diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteEventLogging.Tests/SiteEventLoggerBatchingTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteEventLogging.Tests/SiteEventLoggerBatchingTests.cs index c274f871..150749c3 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.SiteEventLogging.Tests/SiteEventLoggerBatchingTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.SiteEventLogging.Tests/SiteEventLoggerBatchingTests.cs @@ -155,4 +155,135 @@ public class SiteEventLoggerBatchingTests : IDisposable Assert.True(_logger.FailedWriteCount >= burstCount, $"Expected FailedWriteCount to account for all {burstCount} events in the rolled-back batch, got {_logger.FailedWriteCount}."); } + + [Fact] + public async Task ProcessWriteQueue_PoisonRowMidBatch_OtherRowsLand_AndOnlyThePoisonRowIsCountedFailed() + { + // A batch is all-or-nothing, so a SINGLE unwritable row used to take its ~255 innocent + // neighbours down with it — every Task faulted and FailedWriteCount (a health metric) + // inflated by the batch size instead of by the number of bad rows. The rollback is now + // followed by a row-by-row retry: one poison row must cost exactly one row. + _logger.WithConnection(connection => + { + using var cmd = connection.CreateCommand(); + cmd.CommandText = """ + CREATE TRIGGER reject_poison BEFORE INSERT ON site_events + WHEN NEW.message = 'poison' + BEGIN SELECT RAISE(ABORT, 'poison row rejected'); END; + """; + cmd.ExecuteNonQuery(); + }); + + // Hold the writer off the connection so the whole burst is queued together and lands + // in one batch — the poison row is therefore mid-batch, not alone in its own. + var busyStarted = new ManualResetEventSlim(false); + var releaseBusy = new ManualResetEventSlim(false); + var busyThread = new Thread(() => + { + _logger.WithConnection(_ => + { + busyStarted.Set(); + releaseBusy.Wait(TimeSpan.FromSeconds(10)); + }); + }); + busyThread.Start(); + Assert.True(busyStarted.Wait(TimeSpan.FromSeconds(5)), "Busy thread did not start."); + + const int burstCount = 10; + const int poisonIndex = 5; + var tasks = new List(burstCount); + for (var i = 0; i < burstCount; i++) + { + tasks.Add(_logger.LogEventAsync( + "script", "Info", null, "Poison", i == poisonIndex ? "poison" : $"event {i}")); + } + + releaseBusy.Set(); + busyThread.Join(TimeSpan.FromSeconds(10)); + + // Every non-poison event still completes successfully… + for (var i = 0; i < burstCount; i++) + { + if (i == poisonIndex) continue; + await tasks[i].WaitAsync(TimeSpan.FromSeconds(15)); + } + + // …and only the poison event's own caller sees a failure. + await Assert.ThrowsAnyAsync(() => tasks[poisonIndex]) + .WaitAsync(TimeSpan.FromSeconds(15)); + + Assert.Equal(1, _logger.FailedWriteCount); + + var landed = _logger.WithConnection(connection => + { + using var cmd = connection.CreateCommand(); + cmd.CommandText = "SELECT COUNT(*) FROM site_events WHERE source = 'Poison'"; + return (long)cmd.ExecuteScalar()!; + }); + Assert.Equal(burstCount - 1, landed); + } + + [Fact] + public async Task Dispose_PersistsEventsStillQueuedAtShutdown() + { + // Graceful shutdown must DRAIN the queue, not discard it. _disposed is what makes + // WithConnection refuse a write, so flipping it before waiting for the writer loop + // would fault every remaining batch with ObjectDisposedException — losing exactly the + // shutdown-time diagnostics most worth keeping. It flips only after the drain. + var dbPath = Path.Combine(Path.GetTempPath(), $"test_dispose_{Guid.NewGuid()}.db"); + var localDb = TestLocalDb.Create(dbPath); + try + { + var logger = new SiteEventLogger( + Options.Create(new SiteEventLogOptions { DatabasePath = dbPath }), + NullLogger.Instance, + localDb.Db); + + // Block the writer so the whole burst is provably still queued when Dispose starts. + var busyStarted = new ManualResetEventSlim(false); + var releaseBusy = new ManualResetEventSlim(false); + var busyThread = new Thread(() => + { + logger.WithConnection(_ => + { + busyStarted.Set(); + releaseBusy.Wait(TimeSpan.FromSeconds(10)); + }); + }); + busyThread.Start(); + Assert.True(busyStarted.Wait(TimeSpan.FromSeconds(5)), "Busy thread did not start."); + + // More than WriteBatchSize, so the drain spans several batches and a mid-drain + // _disposed flip would be caught even if the first batch slipped through. + const int burstCount = 500; + var tasks = new List(burstCount); + for (var i = 0; i < burstCount; i++) + { + tasks.Add(logger.LogEventAsync("script", "Info", null, "Shutdown", $"event {i}")); + } + + // Dispose blocks on the write lock the busy thread holds; release it once Dispose + // is under way, so the drain and the shutdown genuinely overlap. + var dispose = Task.Run(logger.Dispose); + Thread.Sleep(250); + releaseBusy.Set(); + busyThread.Join(TimeSpan.FromSeconds(10)); + await dispose.WaitAsync(TimeSpan.FromSeconds(30)); + + // Every caller that got an accepted enqueue before shutdown sees a completed write… + await Task.WhenAll(tasks).WaitAsync(TimeSpan.FromSeconds(15)); + Assert.Equal(0, logger.FailedWriteCount); + + // …and the rows are really in the file, read back through a fresh connection. + using var connection = localDb.Db.CreateConnection(); + using var cmd = connection.CreateCommand(); + cmd.CommandText = "SELECT COUNT(*) FROM site_events WHERE source = 'Shutdown'"; + Assert.Equal(burstCount, (long)cmd.ExecuteScalar()!); + } + finally + { + localDb.Dispose(); + TestLocalDb.DeleteFiles(dbPath); + } + } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/ScriptDeadlineAtEnqueueTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/ScriptDeadlineAtEnqueueTests.cs index 8e708540..9ce81e89 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/ScriptDeadlineAtEnqueueTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/ScriptDeadlineAtEnqueueTests.cs @@ -85,7 +85,12 @@ public class ScriptDeadlineAtEnqueueTests : TestKit, IDisposable var lateOptions = new SiteRuntimeOptions { ScriptExecutionTimeoutSeconds = 1, - StuckScriptGraceMs = 1000 + StuckScriptGraceMs = 1000, + // WP3.2 defaults per-run Started/Completed events OFF, which would make the + // "no started event" assertion below vacuously true whether the body ran or not. + // Opt back in so that assertion actually discriminates: a body that reached the + // run loop WOULD emit "started" here, and its absence is therefore evidence. + PerRunScriptEvents = true }; var late = BuildScriptActor( "Late", @@ -112,6 +117,8 @@ public class ScriptDeadlineAtEnqueueTests : TestKit, IDisposable { var rows = siteLog.OfType("script"); // Timeout path only — no "started" Info event, because the body was skipped. + // Per-run events are ON for this script (see lateOptions), so this absence is a + // real signal rather than the global default. Assert.Contains(rows, r => r.Severity == "Error" && r.Message.Contains("timed out")); Assert.DoesNotContain(rows, r => r.Message.Contains("started", StringComparison.OrdinalIgnoreCase)); }, TimeSpan.FromSeconds(5)); diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/ScriptRecursionVsRunCapTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/ScriptRecursionVsRunCapTests.cs new file mode 100644 index 00000000..9f2a0f07 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/ScriptRecursionVsRunCapTests.cs @@ -0,0 +1,244 @@ +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; + +/// +/// The per-script in-flight cap () +/// and the recursion limit () govern two +/// different things, and the cap must not usurp the recursion limit's job. +/// +/// A run awaiting a nested CallScript still holds its in-flight slot — the slot is +/// released only by ScriptExecutionCompleted, sent after the body returns — and a script +/// that calls ITSELF routes the nested request back to the SAME . So +/// counting nested launches against the cap made self-recursion fail at depth 4 (cap) instead +/// of depth 10 (MaxScriptCallDepth), reported as a misleading "shed" with a spurious shed +/// counter, and left the documented recursion-limit path — the site event emitted by +/// ScriptRuntimeContext.CallScript — unreachable. +/// +/// The cap now applies to callDepth == 0 launches only: trigger-driven runs and +/// depth-0 Ask calls, i.e. genuinely NEW work. Both halves are pinned below. +/// +public class ScriptRecursionVsRunCapTests : TestKit, IDisposable +{ + private readonly SharedScriptLibrary _sharedLibrary; + private readonly ScriptExecutionScheduler _scheduler = new(8); + + /// Initializes the shared script library and resets the per-test hooks. + public ScriptRecursionVsRunCapTests() + { + var compilationService = new ScriptCompilationService( + NullLogger.Instance); + _sharedLibrary = new SharedScriptLibrary( + compilationService, NullLogger.Instance); + RecursionHooks.Reset(); + } + + void IDisposable.Dispose() + { + RecursionHooks.Gate.Release(64); + Shutdown(); + _scheduler.Dispose(); + } + + private static Script CompileRaw(string code) + { + var options = ScriptOptions.Default + .WithReferences(typeof(object).Assembly, typeof(Enumerable).Assembly, + typeof(RecursionHooks).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(int maxCallDepth) => new() + { + MaxConcurrentRunsPerScript = 4, + MaxScriptCallDepth = maxCallDepth, + // Long enough that nothing times out inside the test window — the depth limit, not a + // deadline and not the cap, must be what stops the recursion. + ScriptExecutionTimeoutSeconds = 120, + StuckScriptGraceMs = 120_000 + }; + + /// + /// End-to-end self-recursion through the real actor path: the script calls itself, the + /// nested ScriptCallRequest is routed back to the same actor by a stand-in instance + /// actor, and the chain must run all the way to MaxScriptCallDepth and then be + /// stopped by the recursion limit — with its site event — never by the concurrency cap. + /// + [Fact] + public void SelfRecursion_ReachesMaxScriptCallDepth_AndStopsAtTheRecursionLimit_NotAShed() + { + const int maxCallDepth = 6; // > MaxConcurrentRunsPerScript (4): the whole point + var siteLog = new FakeSiteEventLogger(); + var health = new SiteHealthCollector(); + var options = Options(maxCallDepth); + + // Stand-in Instance Actor: routes ScriptCallRequest straight back to the one script + // actor, which is exactly what an InstanceActor does for a same-instance CallScript — + // and makes the nested request land on the SAME actor whose cap is under test. + var router = ActorOf(Props.Create(() => new SelfCallRouter()), "router-" + Guid.NewGuid().ToString("N")); + + var actor = ActorOf( + Props.Create(() => new ScriptActor( + "Self", "Inst1", router, + CompileRaw( + "ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests.Actors.RecursionHooks.Enter();" + + "await Instance.CallScript(\"Self\");" + + "return null;"), + new ResolvedScript { CanonicalName = "Self", TriggerType = "Call" }, + _sharedLibrary, options, NullLogger.Instance, + null, null, health, new SingleServiceProvider(siteLog), _scheduler, null)), + "self-" + Guid.NewGuid().ToString("N")); + + router.Tell(new SelfCallRouter.SetTarget(actor)); + + var caller = CreateTestProbe(); + actor.Tell(new ScriptCallRequest("Self", null, 0, "corr-root"), caller.Ref); + + // The whole chain unwinds back to the root caller: the deepest call is refused by the + // recursion limit, that failure propagates up through each awaiting CallScript. + var result = caller.ExpectMsg(TimeSpan.FromSeconds(30)); + Assert.False(result.Success); + Assert.Equal("corr-root", result.CorrelationId); + // NOT a shed — the reply carries the depth diagnosis, which is the actionable one. + Assert.DoesNotContain("shed", result.ErrorMessage, StringComparison.OrdinalIgnoreCase); + + // Root run + one run per nesting level: recursion went the full documented distance + // instead of stopping at the cap (which would have given 4). + Assert.Equal(maxCallDepth + 1, RecursionHooks.Runs); + + AwaitAssert(() => + { + var scriptEvents = siteLog.OfType("script"); + + // The recursion-limit path is REACHABLE: its Error site event is emitted by + // ScriptRuntimeContext, sourced at the INSTANCE (the per-run failure events the + // unwind also produces are sourced at the script actor and merely quote it), and + // it fires exactly once — at the bottom of the chain. + var limitEvent = Assert.Single( + scriptEvents, r => r.Source == "InstanceScript:Inst1"); + Assert.Equal("Error", limitEvent.Severity); + Assert.StartsWith("Script call depth exceeded", limitEvent.Message); + Assert.Contains($"maximum of {maxCallDepth}", limitEvent.Message); + Assert.Contains($"rejected at depth {maxCallDepth + 1}", limitEvent.Message); + + // …and nothing was shed on the way there. + Assert.DoesNotContain(scriptEvents, r => r.Message.Contains("shed", StringComparison.OrdinalIgnoreCase)); + }, TimeSpan.FromSeconds(10)); + + Assert.Equal(0, health.CollectReport("site-1").ScriptRunShedCount); + } + + /// + /// The exemption is scoped to nesting: with the cap already full of depth-0 runs, another + /// depth-0 request is still shed, while a nested (callDepth > 0) request is + /// launched. This is the discriminating pin — a blanket "skip the cap" would fail the first + /// half, and the pre-fix behaviour fails the second. + /// + [Fact] + public void WithTheCapFull_DepthZeroIsStillShed_ButANestedCallIsLaunched() + { + var siteLog = new FakeSiteEventLogger(); + var health = new SiteHealthCollector(); + var instance = CreateTestProbe().Ref; + var options = Options(maxCallDepth: 10); + + var actor = ActorOfAsTestActorRef( + Props.Create(() => new ScriptActor( + "Hot", "Inst1", instance, + CompileRaw("ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests.Actors.RecursionHooks.Gate.Wait(); return null;"), + new ResolvedScript { CanonicalName = "Hot", TriggerType = "Call" }, + _sharedLibrary, options, NullLogger.Instance, + null, null, health, new SingleServiceProvider(siteLog), _scheduler, null)), + "cap-" + Guid.NewGuid().ToString("N")); + + // Fill the cap with four depth-0 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)); + + // Depth 0 — new work — is still refused. + var newWork = CreateTestProbe(); + actor.Tell(new ScriptCallRequest("Hot", null, 0, "corr-depth0"), newWork.Ref); + var shed = newWork.ExpectMsg(TimeSpan.FromSeconds(10)); + Assert.False(shed.Success); + Assert.Contains("shed", shed.ErrorMessage, StringComparison.OrdinalIgnoreCase); + Assert.Equal(4, actor.UnderlyingActor.RunsInFlight); + + // Depth 1 — a nested call, bounded by MaxScriptCallDepth instead — is launched. + var nested = CreateTestProbe(); + actor.Tell(new ScriptCallRequest("Hot", null, 1, "corr-depth1"), nested.Ref); + + AwaitAssert(() => + { + Assert.Equal(5, actor.UnderlyingActor.RunsInFlight); + Assert.Equal(5, _scheduler.BusyThreadCount); + }, TimeSpan.FromSeconds(15)); + + // It is running, not answered: no shed reply reached the nested caller. + nested.ExpectNoMsg(TimeSpan.FromMilliseconds(500)); + + // Exactly the one depth-0 refusal was counted. + Assert.Equal(1, health.CollectReport("site-1").ScriptRunShedCount); + } + + /// + /// Stand-in Instance Actor that forwards every message to one script actor, so a script's + /// nested CallScript Ask is routed back to itself (self-recursion) with the original + /// Ask sender preserved. + /// + private sealed class SelfCallRouter : ReceiveActor + { + /// Sets the actor every subsequent message is forwarded to. + /// The script actor to forward to. + public sealed record SetTarget(IActorRef Target); + + private IActorRef? _target; + + /// Initializes the router with no target until arrives. + public SelfCallRouter() + { + Receive(m => _target = m.Target); + ReceiveAny(msg => _target?.Forward(msg)); + } + } +} + +/// Test hooks for the recursion-versus-cap tests. +public static class RecursionHooks +{ + private static int _runs; + + /// Gate the blocking cap-filling script waits on; reset per test. + public static SemaphoreSlim Gate = new(0); + + /// Number of recursive script bodies that started, across the whole chain. + public static int Runs => Volatile.Read(ref _runs); + + /// Called at the top of each recursive script body. + public static void Enter() => Interlocked.Increment(ref _runs); + + /// Resets the hooks between tests. + public static void Reset() + { + Interlocked.Exchange(ref _runs, 0); + Gate = new SemaphoreSlim(0); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/ScriptRunLauncherParityTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/ScriptRunLauncherParityTests.cs index e452dc2c..cb31222c 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/ScriptRunLauncherParityTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/ScriptRunLauncherParityTests.cs @@ -1,3 +1,4 @@ +using System.Diagnostics; using Akka.Actor; using Akka.Event; using Akka.TestKit; @@ -253,13 +254,21 @@ public class ScriptRunLauncherParityTests : TestKit, IDisposable perScriptTimeoutSeconds: perScriptSeconds); var caller = CreateTestProbe(); + var started = Stopwatch.StartNew(); actor.Tell(new ScriptCallRequest("Runner", null, 0, "corr-timeout"), caller.Ref); - // If the effective timeout were the 300 s global (case 1) or ignored (cases 2/3) this - // would not answer inside the window. + // All three cases must resolve to the SAME effective 1 s deadline. "Answered inside + // 15 s" alone would not discriminate — a 15 s window is satisfied by anything from a + // 1 s cancel to a 14 s one — so the run must be shown to have been cancelled AT that + // deadline: the reported timeout value is 1 s, and the wall clock agrees. var result = caller.ExpectMsg(TimeSpan.FromSeconds(15)); + started.Stop(); Assert.False(result.Success); - Assert.Contains("timed out", result.ErrorMessage); + Assert.Contains("timed out after 1s", result.ErrorMessage); + + // The body loops until cancelled, so it cannot answer before its deadline; and a + // deadline resolved to either global (300 s / 30 s) could not answer this soon. + Assert.InRange(started.Elapsed, TimeSpan.FromMilliseconds(800), TimeSpan.FromSeconds(8)); AwaitAssert( () => Assert.Contains(siteLog.OfType("script"), diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Scripts/ScriptRunSummaryRecorderTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Scripts/ScriptRunSummaryRecorderTests.cs index 1019045e..6bf0606e 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Scripts/ScriptRunSummaryRecorderTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Scripts/ScriptRunSummaryRecorderTests.cs @@ -149,6 +149,55 @@ public class ScriptRunSummaryRecorderTests Assert.Contains($"{expectedTotal} completed", row.Message); } + [Fact] + public async Task RunSpanningAFlushBoundary_StillReportsItsCompletionInTheNextWindow() + { + // A long run starts in window N and finishes in window N+1. Its completion is recorded + // against a FRESH counters object whose Started is 0 — so a flush filter keyed on + // Started/Failed/TimedOut alone would discard it, and the completion (plus its + // duration) would appear in NO summary row at all: window N reports a run that started + // and never finished, window N+1 reports nothing. + var recorder = new ScriptRunSummaryRecorder(); + recorder.RecordStarted("Inst1", "Slow"); + + var firstLog = new FakeSiteEventLogger(); + Assert.True(await recorder.FlushAsync(firstLog)); + Assert.Equal("1 runs: 0 completed, 0 failed, 0 timed out across 1 scripts", + firstLog.OfType("script").Single().Message); + + // …the run finishes after that flush. + recorder.RecordCompleted("Inst1", "Slow", 42); + + var secondLog = new FakeSiteEventLogger(); + Assert.True(await recorder.FlushAsync(secondLog)); + + var row = secondLog.OfType("script").Single(); + Assert.Equal("0 runs: 1 completed, 0 failed, 0 timed out across 1 scripts", row.Message); + + // The duration is carried too, so the completion is not merely counted. + using var doc = JsonDocument.Parse(row.Details!); + var script = doc.RootElement.GetProperty("scripts")[0]; + Assert.Equal("Slow", script.GetProperty("scriptName").GetString()); + Assert.Equal(0, script.GetProperty("started").GetInt64()); + Assert.Equal(1, script.GetProperty("completed").GetInt64()); + Assert.Equal(42, script.GetProperty("maxDurationMs").GetInt64()); + } + + [Fact] + public async Task CompletionOnlyEntry_DoesNotResurrectTheIdleFlushSuppression() + { + // The Completed>0 admission must not weaken "an idle interval emits nothing": a window + // in which literally nothing was recorded still has zero entries to admit. + var recorder = new ScriptRunSummaryRecorder(); + recorder.RecordStarted("Inst1", "A"); + recorder.RecordCompleted("Inst1", "A", 1); + Assert.True(await recorder.FlushAsync(new FakeSiteEventLogger())); + + var idleLog = new FakeSiteEventLogger(); + Assert.False(await recorder.FlushAsync(idleLog)); + Assert.Empty(idleLog.Entries); + } + [Fact] public async Task DurationTracking_ReportsAverageAndMax() {