Merge branch 'worktree-agent-a6a0dffcb93fa5007' into arch-review-remediation
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -260,14 +260,24 @@ public class SiteEventLogger : ISiteEventLogger, IDisposable
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Commits a drained batch of pending events in one transaction. All-or-nothing per
|
||||
/// batch: <c>site_events</c> ids are freshly minted GUIDs per <see cref="LogEventAsync"/>
|
||||
/// call (no cross-call replay/de-dup concern the way <c>SqliteAuditWriter</c> 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. <c>site_events</c> 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.
|
||||
///
|
||||
/// <para>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 <c>details</c> payload, a constraint/trigger reject)
|
||||
/// faulted all ~<see cref="WriteBatchSize"/> of its innocent neighbours and inflated
|
||||
/// <see cref="FailedWriteCount"/> — 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.</para>
|
||||
///
|
||||
/// <para><c>site_events</c> ids are freshly minted GUIDs per <see cref="LogEventAsync"/>
|
||||
/// call (no cross-call replay/de-dup concern the way <c>SqliteAuditWriter</c> has), so a
|
||||
/// re-insert during the fallback cannot duplicate a row the rolled-back batch had already
|
||||
/// written. <c>site_events</c> 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.</para>
|
||||
/// </summary>
|
||||
private void FlushBatch(IReadOnlyList<PendingEvent> 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
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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 <see cref="FailedWriteCount"/>, while every other row in the batch still
|
||||
/// lands and completes successfully. Only reached from <see cref="FlushBatch"/> after the
|
||||
/// batched transaction has already rolled back.
|
||||
/// </summary>
|
||||
/// <param name="batch">The rolled-back batch to retry row by row.</param>
|
||||
private void FlushRowByRow(IReadOnlyList<PendingEvent> 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);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
/// <returns>The exception to fault the pending event with.</returns>
|
||||
private static ObjectDisposedException DisposedMidWrite() =>
|
||||
new(nameof(SiteEventLogger),
|
||||
"Event could not be recorded: the event logger was disposed before the write completed.");
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
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)
|
||||
""";
|
||||
|
||||
/// <summary>
|
||||
/// Stops accepting new events, drains the write queue, and disposes the SQLite connection.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <c>_disposed</c> is deliberately NOT set in the first lock block. Completing
|
||||
/// the channel writer is the whole shutdown signal: <see cref="LogEventAsync"/> observes it
|
||||
/// (<c>TryWrite</c> returns false) and the writer loop drains what is already buffered
|
||||
/// before exiting. Flipping <c>_disposed</c> up front would make <see cref="WithConnection"/>
|
||||
/// return false for every remaining <see cref="FlushBatch"/>, 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 <c>SqliteAuditWriter.DisposeAsync</c>.
|
||||
/// </remarks>
|
||||
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();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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.</para>
|
||||
///
|
||||
/// <para>The cap governs NEW work only (<paramref name="callDepth"/> 0 — trigger-driven
|
||||
/// runs and depth-0 Ask calls). A nested <c>CallScript</c> is exempt: the calling run is
|
||||
/// itself still counted in <see cref="_runsInFlight"/> while it awaits its callee (the
|
||||
/// slot is released only by <see cref="ScriptExecutionCompleted"/>, 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
|
||||
/// <see cref="SiteRuntimeOptions.MaxConcurrentRunsPerScript"/> with a misleading "shed",
|
||||
/// making <see cref="SiteRuntimeOptions.MaxScriptCallDepth"/> — the limit that actually
|
||||
/// owns this path, enforced in <c>ScriptRuntimeContext.CallScript</c> — unreachable.
|
||||
/// Nested depth is bounded by MaxScriptCallDepth instead, which is what the cap would
|
||||
/// otherwise be doing badly.</para>
|
||||
/// </summary>
|
||||
private void SpawnExecution(
|
||||
IReadOnlyDictionary<string, object?>? 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
|
||||
/// <see cref="SiteRuntimeOptions.MaxConcurrentRunsPerScript"/> 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 <c>CallScript</c> 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 <c>CallScript</c> from
|
||||
/// an unrelated script's run) gets an explicit error so the caller fails fast rather than
|
||||
/// hanging to its Ask timeout. Nested (<c>callDepth > 0</c>) launches never reach here —
|
||||
/// see <see cref="SpawnExecution"/>.
|
||||
/// </summary>
|
||||
private void ShedRun(IActorRef replyTo, string correlationId)
|
||||
{
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -16,10 +16,18 @@ namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Scripts;
|
||||
///
|
||||
/// <para>Lock-free: each script's counters are a small class of <c>long</c> fields mutated
|
||||
/// via <see cref="Interlocked"/>, reached through a <see cref="ConcurrentDictionary{TKey,TValue}"/>
|
||||
/// that <see cref="FlushAsync"/> atomically swaps out (<see cref="Interlocked.Exchange"/>) 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).</para>
|
||||
/// that <see cref="FlushAsync"/> atomically swaps out (<see cref="Interlocked.Exchange"/>), so
|
||||
/// the flush never blocks a recording call (or vice versa).</para>
|
||||
///
|
||||
/// <para>Accounting is BEST-EFFORT, not exact. A recorder that has already resolved its
|
||||
/// <see cref="Counters"/> 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 <see cref="Counters"/> 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.</para>
|
||||
/// </summary>
|
||||
public sealed class ScriptRunSummaryRecorder
|
||||
{
|
||||
@@ -99,8 +107,8 @@ public sealed class ScriptRunSummaryRecorder
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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).
|
||||
/// </summary>
|
||||
@@ -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)
|
||||
|
||||
@@ -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<Task>(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<Exception>(() => 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<SiteEventLogger>.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<Task>(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);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+8
-1
@@ -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));
|
||||
|
||||
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// The per-script in-flight cap (<see cref="SiteRuntimeOptions.MaxConcurrentRunsPerScript"/>)
|
||||
/// and the recursion limit (<see cref="SiteRuntimeOptions.MaxScriptCallDepth"/>) govern two
|
||||
/// different things, and the cap must not usurp the recursion limit's job.
|
||||
///
|
||||
/// <para>A run awaiting a nested <c>CallScript</c> still holds its in-flight slot — the slot is
|
||||
/// released only by <c>ScriptExecutionCompleted</c>, sent after the body returns — and a script
|
||||
/// that calls ITSELF routes the nested request back to the SAME <see cref="ScriptActor"/>. So
|
||||
/// counting nested launches against the cap made self-recursion fail at depth 4 (cap) instead
|
||||
/// of depth 10 (<c>MaxScriptCallDepth</c>), reported as a misleading "shed" with a spurious shed
|
||||
/// counter, and left the documented recursion-limit path — the site event emitted by
|
||||
/// <c>ScriptRuntimeContext.CallScript</c> — unreachable.</para>
|
||||
///
|
||||
/// <para>The cap now applies to <c>callDepth == 0</c> launches only: trigger-driven runs and
|
||||
/// depth-0 Ask calls, i.e. genuinely NEW work. Both halves are pinned below.</para>
|
||||
/// </summary>
|
||||
public class ScriptRecursionVsRunCapTests : TestKit, IDisposable
|
||||
{
|
||||
private readonly SharedScriptLibrary _sharedLibrary;
|
||||
private readonly ScriptExecutionScheduler _scheduler = new(8);
|
||||
|
||||
/// <summary>Initializes the shared script library and resets the per-test hooks.</summary>
|
||||
public ScriptRecursionVsRunCapTests()
|
||||
{
|
||||
var compilationService = new ScriptCompilationService(
|
||||
NullLogger<ScriptCompilationService>.Instance);
|
||||
_sharedLibrary = new SharedScriptLibrary(
|
||||
compilationService, NullLogger<SharedScriptLibrary>.Instance);
|
||||
RecursionHooks.Reset();
|
||||
}
|
||||
|
||||
void IDisposable.Dispose()
|
||||
{
|
||||
RecursionHooks.Gate.Release(64);
|
||||
Shutdown();
|
||||
_scheduler.Dispose();
|
||||
}
|
||||
|
||||
private static Script<object?> 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<object?>(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
|
||||
};
|
||||
|
||||
/// <summary>
|
||||
/// End-to-end self-recursion through the real actor path: the script calls itself, the
|
||||
/// nested <c>ScriptCallRequest</c> is routed back to the same actor by a stand-in instance
|
||||
/// actor, and the chain must run all the way to <c>MaxScriptCallDepth</c> and then be
|
||||
/// stopped by the recursion limit — with its site event — never by the concurrency cap.
|
||||
/// </summary>
|
||||
[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<ScriptActor>.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<ScriptCallResult>(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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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 (<c>callDepth > 0</c>) 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.
|
||||
/// </summary>
|
||||
[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<ScriptActor>(
|
||||
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<ScriptActor>.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<ScriptCallResult>(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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Stand-in Instance Actor that forwards every message to one script actor, so a script's
|
||||
/// nested <c>CallScript</c> Ask is routed back to itself (self-recursion) with the original
|
||||
/// Ask sender preserved.
|
||||
/// </summary>
|
||||
private sealed class SelfCallRouter : ReceiveActor
|
||||
{
|
||||
/// <summary>Sets the actor every subsequent message is forwarded to.</summary>
|
||||
/// <param name="Target">The script actor to forward to.</param>
|
||||
public sealed record SetTarget(IActorRef Target);
|
||||
|
||||
private IActorRef? _target;
|
||||
|
||||
/// <summary>Initializes the router with no target until <see cref="SetTarget"/> arrives.</summary>
|
||||
public SelfCallRouter()
|
||||
{
|
||||
Receive<SetTarget>(m => _target = m.Target);
|
||||
ReceiveAny(msg => _target?.Forward(msg));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Test hooks for the recursion-versus-cap tests.</summary>
|
||||
public static class RecursionHooks
|
||||
{
|
||||
private static int _runs;
|
||||
|
||||
/// <summary>Gate the blocking cap-filling script waits on; reset per test.</summary>
|
||||
public static SemaphoreSlim Gate = new(0);
|
||||
|
||||
/// <summary>Number of recursive script bodies that started, across the whole chain.</summary>
|
||||
public static int Runs => Volatile.Read(ref _runs);
|
||||
|
||||
/// <summary>Called at the top of each recursive script body.</summary>
|
||||
public static void Enter() => Interlocked.Increment(ref _runs);
|
||||
|
||||
/// <summary>Resets the hooks between tests.</summary>
|
||||
public static void Reset()
|
||||
{
|
||||
Interlocked.Exchange(ref _runs, 0);
|
||||
Gate = new SemaphoreSlim(0);
|
||||
}
|
||||
}
|
||||
+12
-3
@@ -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<ScriptCallResult>(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"),
|
||||
|
||||
+49
@@ -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()
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user