feat(scripting): PE cache, quarantine-race fix, recovery, readiness for the out-of-process runner (lmxopcua#780 P1)
v2-ci / unit-tests (push) Failing after 1h12m10s

Second P1 slice of docs/plans/2026-11-09-out-of-process-script-runner-design.md; still opt-in behind
Scripting:Runner=OutOfProcess (InProcess unchanged).

- Compiled-PE cache (design §7, §14 #8): Scripting:Worker:PeCacheDir holds <dir>/pe/<id>.pe entries authenticated
  with HMAC-SHA256 under a per-node random key (<dir>/pe-cache.key, created by the host, 0600/0700 on Unix), handed to
  the worker in Hello. Entry id = SHA-256(compile fingerprint + kind + source); the fingerprint covers the Roslyn
  version, the Core.Scripting build (analyzer, wrapper, stack guard), the sandbox reference set and the runtime. Edited,
  truncated, other-key or renamed entries are rejected, deleted and recompiled; the worker never loads unverified bytes.
  Pruned by recency at startup (30 d / 50 000). ScriptEvaluator splits into CompileToImage + FromImage. Measured on
  40 scripts, restart to every first result: macOS 820 ms no cache / 840 ms empty / 85 ms warm;
  windev (Windows) 9.5 s / 9.4 s / 1.0 s.
- Quarantine race: WorkerConnection runs exit attribution synchronously BEFORE completing the evaluations/loads it held
  (and before Exited), and the shard passes it straight to the runner — a kill strike is in place before the killed
  key's slot is released.
- Supervision: a dead worker restarts by itself after its backoff; Unavailable outcomes carry ScriptRunOutcome.Recovery
  (-> VirtualTagEvalResult.RetryWhen) and the VT actor / Calculation driver re-evaluate once against current inputs when
  it completes. Isolation worker stops after Scripting:Worker:IsolationIdleTimeout (10 min). Keys compiling when a
  worker dies are suspects too (isolation finds a poison compile and quarantines it). Startup orphan sweep by launch
  marker: kills only a process whose marker's host is gone and whose pid + start time + image match.
- Crash-loop circuit (Scripting:Worker:CircuitOpenAfter, 2 min): runner Unavailable, one Error, evaluations answered
  at once while shards keep retrying at the cap.
- Observability (§10): otopcua.script.runner.state, worker.up/inflight/queue_depth/heap_bytes (shard),
  ipc.roundtrip + eval.duration (kind), compile{cache}, isolation.suspects, exits reason=idle. `script-runner` health
  check on the ready tier, at worst Degraded (§14 #12: degraded but still ready).

Left on #780: self-contained worker publish, installer (account, pipe ACL, firewall), docker-dev live drill,
scripted-alarm seam, the §4.5 per-seam batch submit, the §8 perf harness.

Tests: SupervisionTests, PeCacheStoreTests/PeCacheWorkerTests (real workers), ScriptRunnerHealthCheckTests, VT + Calc
recovery seam tests; 14 mutations each turned their test red.

Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
This commit is contained in:
Joseph Doherty
2026-11-10 00:20:24 -05:00
parent 69a6e0b4c8
commit f50b29abae
27 changed files with 2291 additions and 78 deletions
@@ -95,6 +95,13 @@ public sealed record VirtualTagEvalResult(bool Success, object? Value, string? R
/// </summary>
public uint? StatusCode { get; init; }
/// <summary>
/// For a runner failure (<see cref="Unavailable"/>): completes when the runner can run the script again — its
/// worker is back — so the seam re-evaluates it then against its latest inputs instead of waiting for the next
/// input change (lmxopcua#780 design §9). Null for every other result.
/// </summary>
public Task? RetryWhen { get; init; }
/// <summary>A failure that is the runner's, not the script's: published <c>BadResourceUnavailable</c>.</summary>
/// <param name="reason">Description of the failure cause.</param>
/// <returns>A failed evaluation result carrying <see cref="OpcUa.OpcUaStatus.BadResourceUnavailable"/>.</returns>
@@ -1,17 +1,45 @@
using System.Collections.Concurrent;
using System.Diagnostics;
using System.Diagnostics.Metrics;
using System.Security.Cryptography;
using System.Text;
using Serilog;
using Serilog.Core;
using Serilog.Events;
using ZB.MOM.WW.OtOpcUa.Commons.Engines;
using ZB.MOM.WW.OtOpcUa.Commons.Observability;
using ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol;
using ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol.V1;
namespace ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess;
/// <summary>The out-of-process runner's health (the <c>otopcua.script.runner.state</c> gauge: 0 / 1 / 2).</summary>
public enum ScriptRunnerState
{
/// <summary>Every shard that has been needed is up.</summary>
Ready = 0,
/// <summary>At least one shard (or the isolation worker) is down after a failure; the rest serve.</summary>
Degraded = 1,
/// <summary>The crash-loop circuit is open: every shard has been down for <c>CircuitOpenAfter</c> (design §5.3.6).</summary>
Unavailable = 2,
}
/// <summary>A point-in-time view of the runner (the readiness check and diagnostics).</summary>
/// <param name="State">The overall state.</param>
/// <param name="Shards">Every shard, the isolation worker last.</param>
/// <param name="SuspectCount">Keys currently routed to the isolation worker.</param>
/// <param name="QuarantinedCount">Keys currently quarantined (all seams).</param>
/// <param name="PeCacheEnabled">Whether the compiled-PE cache is in use.</param>
public sealed record ScriptRunnerStatus(
ScriptRunnerState State,
IReadOnlyList<ScriptWorkerShardStatus> Shards,
int SuspectCount,
int QuarantinedCount,
bool PeCacheEnabled);
/// <summary>
/// The pre-emptible <see cref="IScriptRunner"/> (lmxopcua#780, design §3–§9): compiled scripts run in a pool of
/// The pre-emptible <see cref="IScriptRunner"/> (lmxopcua#780, design §3–§10): compiled scripts run in a pool of
/// <c>ScriptWorker</c> processes — K shards placed by a stable hash of the script key, plus an isolation worker
/// for suspects — so a script that overruns its hard deadline is stopped by killing its worker, and an OOM, a
/// native crash or a fail-fast takes down a worker instead of the server.
@@ -21,16 +49,22 @@ namespace ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess;
/// <item><b>Two deadlines.</b> At the soft deadline <c>T</c> the caller gets today's timeout failure (and the
/// run token is cancelled inside the worker). At <c>T + G</c> (<see cref="ScriptWorkerOptions.EffectiveHardGrace"/>)
/// the worker is killed and the key takes a kill strike.</item>
/// <item><b>Attribution.</b> A deadline kill blames exactly one key; the other evaluations in flight on that
/// worker are victims and report <see cref="ScriptRunOutcome.WorkerRestartedReason"/> (no strike). A crash with
/// no deadline crossed makes every key in flight a suspect, and suspects run in the isolation worker one at a
/// time, where a crash names exactly one key. A suspect returns to its shard after
/// <item><b>Attribution</b> runs synchronously in the worker's exit path, before any evaluation it held is
/// completed — so a quarantine earned by a kill is in place before the key's slot is released and the next
/// trigger is admitted. A deadline kill blames exactly one key; the other evaluations in flight on that worker
/// are victims and report <see cref="ScriptRunOutcome.WorkerRestartedReason"/> (no strike). A crash with no
/// deadline crossed makes every key in flight — evaluating or compiling — a suspect, and suspects run (and
/// compile) in the isolation worker one at a time, where a crash names exactly one key; that is also how a
/// script whose compile keeps crashing its worker is found. A suspect returns to its shard after
/// <see cref="ScriptWorkerOptions.IsolationProbationRuns"/> clean runs.</item>
/// <item><b>Quarantine</b> lives here, in the host, so it survives worker restarts: kill strikes (within an
/// hour), consecutive soft timeouts, or one allocation-budget breach. Reasons keep the
/// <see cref="ScriptRunGuard.QuarantinedReasonPrefix"/>, and a source change or <see cref="Reset"/> lifts them.</item>
/// <item><b>Fail closed.</b> When no worker is ready within <see cref="ScriptWorkerOptions.UnavailableGrace"/>
/// the result is <see cref="ScriptRunStatus.Unavailable"/> — never an in-process run.</item>
/// the result is <see cref="ScriptRunStatus.Unavailable"/> — never an in-process run — carrying a
/// <see cref="ScriptRunOutcome.Recovery"/> task the seam re-evaluates on.</item>
/// <item><b>PE cache</b> (design §7): with <see cref="ScriptWorkerOptions.PeCacheDir"/> set, workers load
/// HMAC-verified compiled images instead of re-entering Roslyn after a restart.</item>
/// </list>
/// </remarks>
public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, IDisposable
@@ -44,8 +78,52 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
/// <summary>Failure reason when the worker could not take more work.</summary>
public const string WorkerBusyReason = "script worker busy";
/// <summary>PE cache entries unused for this long are pruned at startup.</summary>
internal static readonly TimeSpan PeCacheMaxAge = TimeSpan.FromDays(30);
/// <summary>At most this many PE cache entries are kept.</summary>
internal const int PeCacheMaxEntries = 50_000;
private static readonly TimeSpan KillStrikeWindow = TimeSpan.FromHours(1);
private static readonly ConcurrentDictionary<OutOfProcessScriptRunner, byte> LiveRunners = new();
private static readonly Counter<long> CompileCounter = OtOpcUaTelemetry.Meter.CreateCounter<long>(
"otopcua.script.compile", unit: "{compile}",
description: "Out-of-process script loads (tag: cache = hit | miss | rejected | uncached | negative).");
private static readonly Histogram<double> RoundTripHistogram = OtOpcUaTelemetry.Meter.CreateHistogram<double>(
"otopcua.script.ipc.roundtrip", unit: "us",
description: "Host-measured submit-to-result time of one out-of-process evaluation (tag: kind).");
private static readonly Histogram<double> EvalDurationHistogram = OtOpcUaTelemetry.Meter.CreateHistogram<double>(
"otopcua.script.eval.duration", unit: "us",
description: "Worker-measured run time of one out-of-process evaluation (tag: kind).");
static OutOfProcessScriptRunner()
{
// Gauges observe every live runner (one per node in production; tests make several).
OtOpcUaTelemetry.Meter.CreateObservableGauge("otopcua.script.runner.state",
static () => LiveRunners.Keys.Select(r => new Measurement<int>((int)r.State)),
unit: "{state}", description: "Out-of-process script runner state: 0 ready, 1 degraded, 2 unavailable (circuit open).");
OtOpcUaTelemetry.Meter.CreateObservableGauge("otopcua.script.worker.up",
static () => ShardMeasurements(s => s.Up ? 1 : 0),
unit: "{process}", description: "1 when the shard's ScriptWorker is live (tag: shard).");
OtOpcUaTelemetry.Meter.CreateObservableGauge("otopcua.script.worker.inflight",
static () => ShardMeasurements(s => s.InFlight),
unit: "{eval}", description: "Evaluations outstanding on the shard's ScriptWorker (tag: shard).");
OtOpcUaTelemetry.Meter.CreateObservableGauge("otopcua.script.worker.queue_depth",
static () => ShardMeasurements(s => s.Queued),
unit: "{eval}", description: "Evaluations queued inside the shard's ScriptWorker, from its heartbeat (tag: shard).");
OtOpcUaTelemetry.Meter.CreateObservableGauge("otopcua.script.worker.heap_bytes",
static () => LiveRunners.Keys.SelectMany(r => r.AllShards.Select(s => s.Status()))
.Select(s => new Measurement<long>(s.HeapBytes, new KeyValuePair<string, object?>("shard", s.Label))),
unit: "By", description: "The shard's ScriptWorker managed heap, from its heartbeat (tag: shard).");
OtOpcUaTelemetry.Meter.CreateObservableGauge("otopcua.script.isolation.suspects",
static () => LiveRunners.Keys.Select(r => new Measurement<int>(r.SuspectKeys.Count)),
unit: "{script}", description: "Scripts currently routed to the isolation worker.");
}
private readonly ScriptingOptions _options;
private readonly ScriptWorkerOptions _worker;
private readonly ILogger _log;
@@ -54,6 +132,15 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
private readonly WorkerShard _isolation;
private readonly ConcurrentDictionary<string, KeyState> _keys = new(StringComparer.Ordinal);
private readonly ConcurrentDictionary<string, string> _compileFailures = new(StringComparer.Ordinal);
private readonly Timer _maintenance;
private readonly string _markerDirectory;
private readonly string? _peCacheDirectory;
private readonly byte[]? _peCacheKey;
private long _peCacheHits;
private long _peCacheMisses;
private long _peCacheRejected;
private int _maintaining;
private int _circuitLogged;
private int _disposed;
/// <summary>Initializes the runner. No process starts until the first evaluation.</summary>
@@ -66,7 +153,7 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
}
internal OutOfProcessScriptRunner(ScriptingOptions options, ILogger? log, ILogger? scriptRoot,
TimeSpan heartbeatInterval, TimeSpan loadTimeout)
TimeSpan heartbeatInterval, TimeSpan loadTimeout, TimeSpan? maintenanceInterval = null, string? markerDirectory = null)
{
ArgumentNullException.ThrowIfNull(options);
_options = options.Sanitized(out _);
@@ -76,6 +163,32 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
LoadTimeout = loadTimeout;
WorkerPath = ResolveWorkerPath(_worker);
// The PE cache is best effort: a directory or key that cannot be used is a Warning and a cache-less runner,
// never a startup failure (the path RULE is enforced at startup by ScriptRunnerSelection.Validate).
if (!string.IsNullOrWhiteSpace(_worker.PeCacheDir))
{
try
{
_peCacheKey = ScriptPeCacheStore.LoadOrCreateKey(_worker.PeCacheDir);
_peCacheDirectory = _worker.PeCacheDir;
var pruned = new ScriptPeCacheStore(_peCacheDirectory, _peCacheKey).Prune(PeCacheMaxAge, PeCacheMaxEntries);
if (pruned > 0) _log.Information("Script PE cache {Directory}: pruned {Count} stale entr(ies)", _peCacheDirectory, pruned);
}
catch (Exception ex) when (ex is IOException or UnauthorizedAccessException)
{
_peCacheKey = null;
_peCacheDirectory = null;
_log.Warning(ex, "Script PE cache {Directory} is unusable; ScriptWorkers will recompile every script on each start",
_worker.PeCacheDir);
}
}
// Design §3.3: workers a dead predecessor of THIS host instance left behind, identified by their launch
// markers — never by process name.
_markerDirectory = markerDirectory ?? WorkerLaunchMarker.DirectoryFor(_worker.PeCacheDir);
var swept = WorkerLaunchMarker.Sweep(_markerDirectory, _log);
if (swept.Count > 0) _log.Warning("Killed {Count} orphaned ScriptWorker process(es) at startup: {Pids}", swept.Count, swept);
var evalThreads = Math.Max(2, Environment.ProcessorCount / _worker.Shards);
_shards = new WorkerShard[_worker.Shards];
for (var i = 0; i < _shards.Length; i++)
@@ -84,11 +197,27 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
_shards[i] = NewShard(label, isolation: false, evalThreads, heartbeatInterval);
}
_isolation = NewShard("iso", isolation: true, evalThreads: 1, heartbeatInterval);
var tick = maintenanceInterval ?? TimeSpan.FromSeconds(1);
_maintenance = new Timer(_ => _ = MaintainAsync(), null, tick, tick);
LiveRunners.TryAdd(this, 0);
}
/// <summary>The worker entry assembly this runner launches.</summary>
public string WorkerPath { get; }
/// <summary>The overall state (the readiness check, the state gauge).</summary>
public ScriptRunnerState State => ComputeState(_shards.Select(s => s.DownFor).ToArray(), _isolation.DownFor, _worker.CircuitOpenAfter);
/// <summary>A snapshot of the runner for diagnostics and health.</summary>
/// <returns>The status.</returns>
public ScriptRunnerStatus Status() => new(
State,
AllShards.Select(s => s.Status()).ToArray(),
SuspectKeys.Count,
_keys.Count(kv => kv.Value.QuarantineReason is not null),
_peCacheDirectory is not null);
/// <summary>How long a compile in the worker may take before the worker is killed.</summary>
internal TimeSpan LoadTimeout { get; }
@@ -98,10 +227,22 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
/// <summary>The isolation worker (tests).</summary>
internal WorkerShard Isolation => _isolation;
/// <summary>The marker directory this runner writes launch markers to and sweeps at startup.</summary>
internal string MarkerDirectory => _markerDirectory;
/// <summary>PE cache outcomes reported by this runner's workers (tests / diagnostics).</summary>
internal (long Hits, long Misses, long Rejected) PeCacheCounts =>
(Interlocked.Read(ref _peCacheHits), Interlocked.Read(ref _peCacheMisses), Interlocked.Read(ref _peCacheRejected));
/// <summary>Test hook: runs inside the continuation that releases a key's single-flight slot, just before it does.</summary>
internal Action<string>? SlotReleasingForTest { get; set; }
/// <summary>Scoped keys currently treated as suspects (tests / diagnostics).</summary>
internal IReadOnlyCollection<string> SuspectKeys =>
_keys.Where(kv => kv.Value.Suspect).Select(kv => kv.Key).ToArray();
private IEnumerable<WorkerShard> AllShards => _shards.Append(_isolation);
/// <summary>The shard a key is placed on when it is not a suspect: a stable FNV-1a hash, so placement survives restarts.</summary>
/// <param name="scopedKey">The scoped key.</param>
/// <param name="shardCount">K.</param>
@@ -117,6 +258,23 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
return (int)(hash % (uint)shardCount);
}
/// <summary>
/// The state rule: Unavailable when every shard has been down for at least <paramref name="circuitOpenAfter"/>,
/// Degraded when any shard or the isolation worker is down, Ready otherwise.
/// </summary>
/// <param name="shardsDownFor">Each regular shard's down time (null = up / never needed).</param>
/// <param name="isolationDownFor">The isolation worker's.</param>
/// <param name="circuitOpenAfter">The circuit threshold.</param>
/// <returns>The state.</returns>
internal static ScriptRunnerState ComputeState(IReadOnlyList<TimeSpan?> shardsDownFor, TimeSpan? isolationDownFor, TimeSpan circuitOpenAfter)
{
if (shardsDownFor.Count > 0 && shardsDownFor.All(d => d is { } down && down >= circuitOpenAfter))
return ScriptRunnerState.Unavailable;
return shardsDownFor.Any(d => d is not null) || isolationDownFor is not null
? ScriptRunnerState.Degraded
: ScriptRunnerState.Ready;
}
/// <summary>The default worker path, or the configured one.</summary>
/// <param name="worker">The worker options.</param>
/// <returns>The absolute worker path.</returns>
@@ -137,7 +295,11 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
var sha = SHA256.HashData(Encoding.UTF8.GetBytes(invocation.Source));
var failureKey = InProcessScriptRunner.ScopedKey(invocation.Kind, Convert.ToHexString(sha));
if (_compileFailures.TryGetValue(failureKey, out var knownFailure)) return ScriptRunOutcome.Failed(knownFailure);
if (_compileFailures.TryGetValue(failureKey, out var knownFailure))
{
CompileCounter.Add(1, new KeyValuePair<string, object?>("cache", "negative"));
return ScriptRunOutcome.Failed(knownFailure);
}
if (Interlocked.CompareExchange(ref state.InFlight, 1, 0) != 0)
{
@@ -147,19 +309,23 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
var handedOff = false;
SemaphoreSlim? gate = null;
var shard = state.Suspect ? _isolation : _shards[ShardIndex(key, _shards.Length)];
WorkerConnection? connection = null;
try
{
state.ScriptLogger = invocation.ScriptLogger;
var shard = state.Suspect ? _isolation : _shards[ShardIndex(key, _shards.Length)];
// Circuit open (design §5.3.6): answer at once instead of making every trigger wait out the grace. A zero
// grace still kicks a start that is due, so the shards keep retrying at the backoff cap.
var grace = State == ScriptRunnerState.Unavailable ? TimeSpan.Zero : _worker.UnavailableGrace;
if (shard.IsolationGate is { } isolationGate)
{
if (!await isolationGate.WaitAsync(_worker.UnavailableGrace, ct).ConfigureAwait(false))
return ScriptRunOutcome.Unavailable(ScriptRunOutcome.RunnerUnavailableReason);
if (!await isolationGate.WaitAsync(grace, ct).ConfigureAwait(false))
return Unavailable(ScriptRunOutcome.RunnerUnavailableReason, shard, null);
gate = isolationGate;
}
var connection = await shard.GetReadyAsync(_worker.UnavailableGrace, ct).ConfigureAwait(false);
if (connection is null) return ScriptRunOutcome.Unavailable(ScriptRunOutcome.RunnerUnavailableReason);
connection = await shard.GetReadyAsync(grace, ct).ConfigureAwait(false);
if (connection is null) return Unavailable(ScriptRunOutcome.RunnerUnavailableReason, shard, null);
var paths = invocation.Inputs.Keys.Order(StringComparer.Ordinal).ToArray();
WorkerConnection.LoadEntry? entry;
@@ -178,7 +344,8 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
connection.Kill(WorkerExitReason.KilledLoadTimeout, key);
return ScriptRunOutcome.Failed($"compile failure: compile did not finish within {LoadTimeout.TotalSeconds:F0} s");
}
if (entry?.Outcome is not { } outcome) return ScriptRunOutcome.Unavailable(ScriptRunOutcome.WorkerRestartedReason);
if (entry?.Outcome is not { } outcome) return Unavailable(ScriptRunOutcome.WorkerRestartedReason, RouteFor(key, state), connection);
NoteLoad(entry);
if (!outcome.Ok)
{
var reason = outcome.Error?.Kind switch
@@ -192,16 +359,19 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
return ScriptRunOutcome.Failed(reason);
}
var submitted = Stopwatch.GetTimestamp();
var pending = connection.Submit(key, BuildRequest(key, entry, invocation));
ArmHardDeadline(connection, pending, invocation.Timeout + _worker.EffectiveHardGrace(invocation.Timeout));
// From here the in-flight slot (and the isolation gate) belong to the evaluation itself: they are released
// when the worker answers or dies — not when this caller stops waiting at the soft deadline. That keeps a
// key single-flight between T and T + G, and the isolation worker strictly one-at-a-time.
// key single-flight between T and T + G, and the isolation worker strictly one-at-a-time. On a worker
// death, attribution (strike, quarantine, suspect) has already run when this continuation fires.
handedOff = true;
var heldGate = gate;
_ = pending.Completion.ContinueWith(_ =>
{
SlotReleasingForTest?.Invoke(key);
Interlocked.Exchange(ref state.InFlight, 0);
heldGate?.Release();
}, CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);
@@ -221,8 +391,11 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
return ScriptRunOutcome.Failed($"{ScriptEvalFailureReasons.TimedOutPrefix}{invocation.Timeout.TotalSeconds:F1}s");
}
if (result is null) return ScriptRunOutcome.Unavailable(ScriptRunOutcome.WorkerRestartedReason);
return Complete(key, state, result);
if (result is null) return Unavailable(ScriptRunOutcome.WorkerRestartedReason, RouteFor(key, state), connection);
var kindTag = new KeyValuePair<string, object?>("kind", KindTag(invocation.Kind));
RoundTripHistogram.Record(Stopwatch.GetElapsedTime(submitted).TotalMicroseconds, kindTag);
EvalDurationHistogram.Record(result.RunMicros, kindTag);
return Complete(key, state, result, shard, connection);
}
catch (OperationCanceledException) when (ct.IsCancellationRequested)
{
@@ -251,7 +424,7 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
{
if (key.StartsWith(prefix, StringComparison.Ordinal)) _compileFailures.TryRemove(key, out _);
}
foreach (var shard in _shards.Append(_isolation))
foreach (var shard in AllShards)
shard.Current?.Unload(dropped);
}
@@ -270,12 +443,47 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
public async ValueTask DisposeAsync()
{
if (Interlocked.Exchange(ref _disposed, 1) != 0) return;
await Task.WhenAll(_shards.Append(_isolation).Select(s => s.DisposeAsync().AsTask())).ConfigureAwait(false);
LiveRunners.TryRemove(this, out _);
await _maintenance.DisposeAsync().ConfigureAwait(false);
await Task.WhenAll(AllShards.Select(s => s.DisposeAsync().AsTask())).ConfigureAwait(false);
}
/// <summary>Stops every worker (blocking).</summary>
public void Dispose() => DisposeAsync().AsTask().GetAwaiter().GetResult();
/// <summary>
/// The periodic pass: stop the isolation worker once it has been idle for
/// <see cref="ScriptWorkerOptions.IsolationIdleTimeout"/>, and log the circuit opening / closing once each.
/// </summary>
/// <returns>A task for the pass.</returns>
internal async Task MaintainAsync()
{
if (Volatile.Read(ref _disposed) != 0 || Interlocked.Exchange(ref _maintaining, 1) != 0) return;
try
{
await _isolation.StopIfIdleAsync(_worker.IsolationIdleTimeout).ConfigureAwait(false);
var open = State == ScriptRunnerState.Unavailable;
if (open && Interlocked.Exchange(ref _circuitLogged, 1) == 0)
{
_log.Error("Script runner circuit OPEN: every ScriptWorker shard has been down for at least {After}; every compiled script on this node reports {Reason} until a worker starts (retrying at the backoff cap)",
_worker.CircuitOpenAfter, ScriptRunOutcome.RunnerUnavailableReason);
}
else if (!open && Interlocked.Exchange(ref _circuitLogged, 0) == 1)
{
_log.Information("Script runner circuit closed: a ScriptWorker shard is serving again");
}
}
catch (Exception ex)
{
_log.Warning(ex, "Script runner maintenance pass failed");
}
finally
{
Interlocked.Exchange(ref _maintaining, 0);
}
}
private WorkerShard NewShard(string label, bool isolation, int evalThreads, TimeSpan heartbeatInterval)
{
var spec = new WorkerLaunchSpec(
@@ -285,12 +493,45 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
EvalThreads: evalThreads,
HardGrace: _worker.HardGrace ?? TimeSpan.FromSeconds(Math.Max(1, _options.RunTimeout.TotalSeconds)),
HeartbeatInterval: heartbeatInterval,
HandshakeTimeout: TimeSpan.FromSeconds(15));
HandshakeTimeout: TimeSpan.FromSeconds(15),
PeCacheDirectory: _peCacheDirectory,
PeCacheKey: _peCacheKey,
MarkerDirectory: _markerDirectory);
return new WorkerShard(label, isolation,
ct => WorkerConnection.StartAsync(spec, _log, OnLogs, ct),
(onExiting, ct) => WorkerConnection.StartAsync(spec, _log, OnLogs, onExiting, ct),
OnWorkerExit, _log);
}
/// <summary>The shard a key's NEXT evaluation goes to (attribution may just have made it a suspect).</summary>
private WorkerShard RouteFor(string key, KeyState state)
=> state.Suspect ? _isolation : _shards[ShardIndex(key, _shards.Length)];
private static ScriptRunOutcome Unavailable(string reason, WorkerShard shard, WorkerConnection? failed)
=> ScriptRunOutcome.Unavailable(reason) with { Recovery = shard.WhenRecoveredAsync(failed) };
private void NoteLoad(WorkerConnection.LoadEntry entry)
{
// Counted once per compile, by the caller that waited for it (an already-loaded key is not a load).
if (Interlocked.Exchange(ref entry.Counted, 1) != 0 || entry.Outcome is not { Ok: true } outcome) return;
var tag = outcome.Cache switch
{
PeCache.Hit => "hit",
PeCache.Miss => "miss",
PeCache.Rejected => "rejected",
_ => "uncached",
};
CompileCounter.Add(1, new KeyValuePair<string, object?>("cache", tag));
switch (outcome.Cache)
{
case PeCache.Hit: Interlocked.Increment(ref _peCacheHits); break;
case PeCache.Miss: Interlocked.Increment(ref _peCacheMisses); break;
case PeCache.Rejected:
Interlocked.Increment(ref _peCacheRejected);
_log.Warning("Script {ScriptKey}: its PE cache entry failed verification and was deleted; recompiled", outcome.Key);
break;
}
}
private KeyState Admit(string key, string source)
{
while (true)
@@ -355,7 +596,7 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
}, CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);
}
private ScriptRunOutcome Complete(string key, KeyState state, EvalResult result)
private ScriptRunOutcome Complete(string key, KeyState state, EvalResult result, WorkerShard shard, WorkerConnection connection)
{
// The script returned control (a value or a throw): the soft-timeout streak resets, as in process.
Interlocked.Exchange(ref state.ConsecutiveTimeouts, 0);
@@ -396,7 +637,7 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
ScriptError.Types.Kind.Threw => ScriptRunOutcome.Failed($"script threw: {result.Error.Message}"),
ScriptError.Types.Kind.UnsupportedType => ScriptRunOutcome.Failed($"unsupported type: {result.Error.Message}"),
ScriptError.Types.Kind.Busy => ScriptRunOutcome.Failed(WorkerBusyReason),
ScriptError.Types.Kind.NotLoaded => ScriptRunOutcome.Unavailable(ScriptRunOutcome.WorkerRestartedReason),
ScriptError.Types.Kind.NotLoaded => Unavailable(ScriptRunOutcome.WorkerRestartedReason, shard, connection),
_ => ScriptRunOutcome.Failed($"script failed: {result.Error.Message}"),
};
default:
@@ -415,7 +656,14 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
return true;
}
private void OnWorkerExit(WorkerShard shard, WorkerExit exit)
/// <summary>
/// Attribution for one worker exit. Called synchronously from the worker's exit path BEFORE any evaluation or
/// compile it held is completed (see <see cref="WorkerConnection.StartAsync"/>), so its verdict is in place
/// before the affected keys' single-flight slots are released.
/// </summary>
/// <param name="shard">The shard whose worker ended.</param>
/// <param name="exit">How it ended.</param>
internal void OnWorkerExit(WorkerShard shard, WorkerExit exit)
{
switch (exit.Reason)
{
@@ -426,22 +674,25 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
KillStrike(culprit, "its compile did not finish and its worker was killed", "killed");
break;
case WorkerExitReason.Crash or WorkerExitReason.KilledHeartbeat or WorkerExitReason.ProtocolFault:
// Compiles count as well as evaluations: a source that crashes Roslyn (or the load) must be found too.
var implicated = exit.InFlightKeys.Concat(exit.LoadingKeys).Distinct(StringComparer.Ordinal).ToArray();
if (shard.IsIsolation)
{
// One evaluation at a time here: the crash is this key's (design §5.3.3).
foreach (var key in exit.InFlightKeys)
KillStrike(key, "it crashed the isolation worker", "isolated_crash");
// One evaluation (or compile) at a time here: the crash is this key's (design §5.3.3).
foreach (var key in implicated)
KillStrike(key, exit.InFlightKeys.Contains(key) ? "it crashed the isolation worker" : "its compile crashed the isolation worker",
"isolated_crash");
}
else if (exit.InFlightKeys.Count > 0)
else if (implicated.Length > 0)
{
foreach (var key in exit.InFlightKeys)
foreach (var key in implicated)
{
if (!_keys.TryGetValue(key, out var state)) continue;
state.Suspect = true;
Interlocked.Exchange(ref state.ProbationRuns, 0);
}
_log.Warning("ScriptWorker[{Shard}] ended ({Reason}) with no deadline crossed; {Count} script(s) in flight move to the isolation worker: {Suspects}",
shard.Label, exit.Reason, exit.InFlightKeys.Count, string.Join(", ", exit.InFlightKeys));
shard.Label, exit.Reason, implicated.Length, string.Join(", ", implicated));
}
break;
}
@@ -491,6 +742,18 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
_log.Warning("ScriptWorker dropped {Dropped} script log line(s): a script is logging faster than the pipe drains", batch.Dropped);
}
private static string KindTag(ScriptContextKind kind) => kind switch
{
ScriptContextKind.VirtualTag => "virtual_tag",
ScriptContextKind.Calculation => "calculation",
ScriptContextKind.AlarmPredicate => "alarm",
_ => "other",
};
private static IEnumerable<Measurement<int>> ShardMeasurements(Func<ScriptWorkerShardStatus, int> value)
=> LiveRunners.Keys.SelectMany(r => r.AllShards.Select(s => s.Status()))
.Select(s => new Measurement<int>(value(s), new KeyValuePair<string, object?>("shard", s.Label)));
private sealed class KeyState(string source)
{
public readonly string Source = source;
@@ -33,6 +33,9 @@ internal enum WorkerExitReason
/// <summary>The runner is shutting down.</summary>
Disposed,
/// <summary>The host stopped an idle worker (the isolation worker after <c>IsolationIdleTimeout</c>); not a failure.</summary>
Idle,
}
/// <summary>The end of one worker process.</summary>
@@ -40,7 +43,16 @@ internal enum WorkerExitReason
/// <param name="CulpritKey">The scoped key the host blamed, for a deadline or load kill.</param>
/// <param name="ExitCode">The OS exit code, when known.</param>
/// <param name="InFlightKeys">The scoped keys that had an evaluation outstanding when it ended.</param>
internal sealed record WorkerExit(WorkerExitReason Reason, string? CulpritKey, int? ExitCode, IReadOnlyList<string> InFlightKeys);
/// <param name="LoadingKeys">The scoped keys whose compile (load) was outstanding when it ended.</param>
internal sealed record WorkerExit(WorkerExitReason Reason, string? CulpritKey, int? ExitCode, IReadOnlyList<string> InFlightKeys,
IReadOnlyList<string> LoadingKeys)
{
/// <summary>An exit with no load in flight.</summary>
public WorkerExit(WorkerExitReason reason, string? culpritKey, int? exitCode, IReadOnlyList<string> inFlightKeys)
: this(reason, culpritKey, exitCode, inFlightKeys, [])
{
}
}
/// <summary>What the host needs to launch one worker.</summary>
/// <param name="Shard">The shard label (<c>0</c>, <c>1</c>, … or <c>iso</c>).</param>
@@ -50,6 +62,9 @@ internal sealed record WorkerExit(WorkerExitReason Reason, string? CulpritKey, i
/// <param name="HardGrace">The worker watchdog's grace past the soft deadline.</param>
/// <param name="HeartbeatInterval">How often the worker sends a heartbeat.</param>
/// <param name="HandshakeTimeout">How long a launch may take to reach Ready.</param>
/// <param name="PeCacheDirectory">The compiled-PE cache directory handed to the worker; null = no cache.</param>
/// <param name="PeCacheKey">The cache's HMAC key (with <paramref name="PeCacheDirectory"/>).</param>
/// <param name="MarkerDirectory">Where the launch marker for the orphan sweep is written; null = none.</param>
internal sealed record WorkerLaunchSpec(
string Shard,
string WorkerPath,
@@ -57,7 +72,10 @@ internal sealed record WorkerLaunchSpec(
int EvalThreads,
TimeSpan HardGrace,
TimeSpan HeartbeatInterval,
TimeSpan HandshakeTimeout);
TimeSpan HandshakeTimeout,
string? PeCacheDirectory = null,
byte[]? PeCacheKey = null,
string? MarkerDirectory = null);
/// <summary>
/// One live ScriptWorker process and its pipe (lmxopcua#780, design §3–§6): launch with a cleared, allow-listed
@@ -85,12 +103,17 @@ internal sealed class WorkerConnection : IAsyncDisposable
private readonly Action<LogBatch> _onLogs;
private readonly Channel<Envelope> _outbound = Channel.CreateUnbounded<Envelope>(new UnboundedChannelOptions { SingleReader = true });
private readonly ConcurrentDictionary<ulong, PendingEval> _pending = new();
private readonly ConcurrentDictionary<ulong, TaskCompletionSource<LoadResult>> _loads = new();
private readonly ConcurrentDictionary<ulong, (string Key, TaskCompletionSource<LoadResult> Completion)> _loads = new();
private readonly ConcurrentDictionary<string, LoadEntry> _loaded = new(StringComparer.Ordinal);
private readonly TaskCompletionSource<WorkerExit> _exited = new(TaskCreationOptions.RunContinuationsAsynchronously);
private readonly CancellationTokenSource _stopping = new();
private readonly TimeSpan _heartbeatInterval;
private readonly Action<WorkerExit> _onExiting;
private readonly string? _markerPath;
private TaskCompletionSource<DiagnosticsReply>? _diag;
private Heartbeat? _lastHeartbeat;
private long _lastUsedTimestamp;
private int _finished;
private (WorkerExitReason Reason, string? Culprit)? _killedFor;
private long _lastFrameTimestamp;
private long _seq;
@@ -100,8 +123,10 @@ internal sealed class WorkerConnection : IAsyncDisposable
private int _disposed;
private WorkerConnection(string shard, Process process, NamedPipeServerStream pipe, IDisposable? job, string workDir,
ILogger log, Action<LogBatch> onLogs, TimeSpan heartbeatInterval)
ILogger log, Action<LogBatch> onLogs, TimeSpan heartbeatInterval, Action<WorkerExit> onExiting, string? markerPath)
{
_onExiting = onExiting;
_markerPath = markerPath;
Shard = shard;
_process = process;
_pipe = pipe;
@@ -131,13 +156,29 @@ internal sealed class WorkerConnection : IAsyncDisposable
/// <summary>Evaluations currently outstanding on this worker.</summary>
public int InFlightCount => _pending.Count;
/// <summary>Compiles currently outstanding on this worker.</summary>
public int LoadingCount => _loads.Count;
/// <summary>The worker's last heartbeat (its in-flight / queued counts and heap), or null before the first.</summary>
public Heartbeat? LastHeartbeat => Volatile.Read(ref _lastHeartbeat);
/// <summary><see cref="Stopwatch.GetTimestamp"/> of the last load or evaluation submitted (or of Ready).</summary>
public long LastUsedTimestamp => Interlocked.Read(ref _lastUsedTimestamp);
/// <summary>Launches a worker and completes the handshake.</summary>
/// <param name="spec">What to launch.</param>
/// <param name="log">Host diagnostics (stderr lines, exits).</param>
/// <param name="onLogs">Receives the worker's forwarded <c>ctx.Logger</c> batches.</param>
/// <param name="onExiting">
/// Attribution, called synchronously when the worker has ended and BEFORE any outstanding evaluation or load is
/// completed or <see cref="Exited"/> is signalled. A kill strike or suspect mark recorded here is therefore in
/// place before any caller can see its evaluation end and trigger the script again (the quarantine race,
/// lmxopcua#780).
/// </param>
/// <param name="ct">Cancellation.</param>
/// <returns>A ready connection.</returns>
public static async Task<WorkerConnection> StartAsync(WorkerLaunchSpec spec, ILogger log, Action<LogBatch> onLogs, CancellationToken ct)
public static async Task<WorkerConnection> StartAsync(WorkerLaunchSpec spec, ILogger log, Action<LogBatch> onLogs,
Action<WorkerExit> onExiting, CancellationToken ct)
{
var pipeName = "otopcua-sw-" + Convert.ToHexString(RandomNumberGenerator.GetBytes(12)).ToLowerInvariant();
var nonce = Convert.ToHexString(RandomNumberGenerator.GetBytes(32));
@@ -149,11 +190,15 @@ internal sealed class WorkerConnection : IAsyncDisposable
PipeTransmissionMode.Byte, PipeOptions.Asynchronous | PipeOptions.CurrentUserOnly);
Process? process = null;
IDisposable? job = null;
string? markerPath = null;
try
{
var psi = BuildStartInfo(spec, pipeName, nonce, workDir);
process = new Process { StartInfo = psi, EnableRaisingEvents = true };
if (!process.Start()) throw new InvalidOperationException("the worker process did not start");
// Before anything else can fail: a host that dies from here on leaves a marker the next start's orphan
// sweep can match to exactly this process (design §3.3).
if (spec.MarkerDirectory is { } markerDir) markerPath = WorkerLaunchMarker.Write(markerDir, process, pipeName, log);
if (OperatingSystem.IsWindows())
{
// Native and JIT memory backstop above the managed heap limit.
@@ -190,9 +235,15 @@ internal sealed class WorkerConnection : IAsyncDisposable
HeartbeatIntervalMs = (uint)spec.HeartbeatInterval.TotalMilliseconds,
MaxLogLines = 1000,
};
if (spec.PeCacheDirectory is { } cacheDir && spec.PeCacheKey is { } cacheKey)
{
hello.PeCacheDir = cacheDir;
hello.PeCacheKey = Google.Protobuf.ByteString.CopyFrom(cacheKey);
}
await HandshakeAsync(pipe, hello, nonce, process.Id, handshakeCts.Token).ConfigureAwait(false);
var connection = new WorkerConnection(spec.Shard, process, pipe, job, workDir, log, onLogs, spec.HeartbeatInterval);
var connection = new WorkerConnection(spec.Shard, process, pipe, job, workDir, log, onLogs, spec.HeartbeatInterval,
onExiting, markerPath);
connection.Begin();
return connection;
}
@@ -203,6 +254,7 @@ internal sealed class WorkerConnection : IAsyncDisposable
process?.Dispose();
await pipe.DisposeAsync().ConfigureAwait(false);
TryDeleteDirectory(workDir);
WorkerLaunchMarker.TryDelete(markerPath);
throw;
}
}
@@ -308,6 +360,7 @@ internal sealed class WorkerConnection : IAsyncDisposable
{
ReadyTimestamp = Stopwatch.GetTimestamp();
Volatile.Write(ref _lastFrameTimestamp, ReadyTimestamp);
Interlocked.Exchange(ref _lastUsedTimestamp, ReadyTimestamp);
_ = Task.Run(ReadLoopAsync);
_ = Task.Run(WriteLoopAsync);
_ = Task.Run(HeartbeatMonitorAsync);
@@ -329,9 +382,10 @@ internal sealed class WorkerConnection : IAsyncDisposable
var entry = new LoadEntry(hashHex, inputPaths);
_loaded[key] = entry;
Interlocked.Exchange(ref _lastUsedTimestamp, Stopwatch.GetTimestamp());
var loadId = (ulong)Interlocked.Increment(ref _nextLoadId);
var tcs = new TaskCompletionSource<LoadResult>(TaskCreationOptions.RunContinuationsAsynchronously);
_loads[loadId] = tcs;
_loads[loadId] = (key, tcs);
var load = new LoadScripts { LoadId = loadId };
var script = new ScriptSource { Key = key, Kind = kind, Source = source, SourceSha256 = Google.Protobuf.ByteString.CopyFrom(sourceHash) };
script.InputPaths.AddRange(inputPaths);
@@ -372,6 +426,7 @@ internal sealed class WorkerConnection : IAsyncDisposable
public PendingEval Submit(string key, EvalRequest request)
{
request.EvalId = (ulong)Interlocked.Increment(ref _nextEvalId);
Interlocked.Exchange(ref _lastUsedTimestamp, Stopwatch.GetTimestamp());
var pending = new PendingEval(key, request.EvalId);
_pending[request.EvalId] = pending;
var batch = new EvalBatch();
@@ -423,13 +478,18 @@ internal sealed class WorkerConnection : IAsyncDisposable
/// <summary>Asks the worker to stop, waits briefly, then kills it.</summary>
/// <returns>A task that completes when the worker has ended.</returns>
public async ValueTask DisposeAsync()
public ValueTask DisposeAsync() => StopAsync(WorkerExitReason.Disposed);
/// <summary>As <see cref="DisposeAsync"/>, recording <paramref name="reason"/> as the exit reason.</summary>
/// <param name="reason">Why the host is stopping it (<see cref="WorkerExitReason.Disposed"/> or <see cref="WorkerExitReason.Idle"/>).</param>
/// <returns>A task that completes when the worker has ended.</returns>
public async ValueTask StopAsync(WorkerExitReason reason)
{
// Both the shard (on exit) and the runner (on shutdown) dispose a connection; only the first does the work.
if (Interlocked.Exchange(ref _disposed, 1) != 0) return;
if (!_exited.Task.IsCompleted)
{
lock (_pending) { _killedFor ??= (WorkerExitReason.Disposed, null); }
lock (_pending) { _killedFor ??= (reason, null); }
Send(new Envelope { Shutdown = new Shutdown() });
try
{
@@ -447,6 +507,7 @@ internal sealed class WorkerConnection : IAsyncDisposable
_job?.Dispose();
_process.Dispose();
TryDeleteDirectory(_workDir);
WorkerLaunchMarker.TryDelete(_markerPath);
}
/// <summary>Test hook: closes the host end of the pipe without touching the process, as a dying host would.</summary>
@@ -521,7 +582,7 @@ internal sealed class WorkerConnection : IAsyncDisposable
}
break;
case Envelope.BodyOneofCase.LoadResult:
if (_loads.TryRemove(frame.LoadResult.LoadId, out var load)) load.TrySetResult(frame.LoadResult);
if (_loads.TryRemove(frame.LoadResult.LoadId, out var load)) load.Completion.TrySetResult(frame.LoadResult);
else throw new ScriptWorkerProtocolException($"load result for unknown load id {frame.LoadResult.LoadId}");
break;
case Envelope.BodyOneofCase.Logs:
@@ -529,6 +590,7 @@ internal sealed class WorkerConnection : IAsyncDisposable
catch (Exception ex) { _log.Warning(ex, "ScriptWorker[{Shard}] log forwarding threw", Shard); }
break;
case Envelope.BodyOneofCase.Heartbeat:
Volatile.Write(ref _lastHeartbeat, frame.Heartbeat);
break;
case Envelope.BodyOneofCase.DiagReply:
Volatile.Read(ref _diag)?.TrySetResult(frame.DiagReply);
@@ -604,6 +666,11 @@ internal sealed class WorkerConnection : IAsyncDisposable
private void Finish()
{
// The reader (on EOF) and the process watcher both get here; only the first reports.
if (Interlocked.Exchange(ref _finished, 1) != 0) return;
// No new frame can be queued from here on, so every Submit / load either fails its Send (and completes itself)
// or was registered before this point and is completed below.
_outbound.Writer.TryComplete();
int? code = null;
try { if (_process.HasExited) code = _process.ExitCode; } catch (InvalidOperationException) { }
@@ -611,13 +678,21 @@ internal sealed class WorkerConnection : IAsyncDisposable
lock (_pending) { killed = _killedFor; }
var reason = killed?.Reason ?? (code == 0 ? WorkerExitReason.Clean : WorkerExitReason.Crash);
var inFlight = _pending.Values.Select(p => p.Key).Distinct(StringComparer.Ordinal).ToArray();
if (!_exited.TrySetResult(new WorkerExit(reason, killed?.Culprit, code, inFlight))) return;
var loading = _loads.Values.Select(l => l.Key).Distinct(StringComparer.Ordinal).ToArray();
var exit = new WorkerExit(reason, killed?.Culprit, code, inFlight, loading);
// Attribution FIRST, synchronously: a kill strike / quarantine / suspect mark must be in place before any
// evaluation below completes, because completing it releases the key's single-flight slot and the very next
// trigger is admitted against whatever the quarantine table says at that instant (lmxopcua#780 race fix).
try { _onExiting(exit); }
catch (Exception ex) { _log.Error(ex, "ScriptWorker[{Shard}] exit attribution threw", Shard); }
_exited.TrySetResult(exit);
// Every outstanding evaluation and load ends now; null tells the waiter "the worker went away".
foreach (var id in _pending.Keys.ToArray())
if (_pending.TryRemove(id, out var pending)) pending.Complete(null);
foreach (var id in _loads.Keys.ToArray())
if (_loads.TryRemove(id, out var load)) load.TrySetCanceled();
if (_loads.TryRemove(id, out var load)) load.Completion.TrySetCanceled();
Volatile.Read(ref _diag)?.TrySetCanceled();
_outbound.Writer.TryComplete();
}
@@ -638,6 +713,9 @@ internal sealed class WorkerConnection : IAsyncDisposable
{
private readonly TaskCompletionSource<LoadEntry?> _ready = new(TaskCreationOptions.RunContinuationsAsynchronously);
/// <summary>Set once the runner has counted this load in its metrics.</summary>
public int Counted;
public LoadEntry(string sourceHash, IReadOnlyList<string> inputPaths)
{
SourceHash = sourceHash;
@@ -0,0 +1,166 @@
using System.Diagnostics;
using System.Security.Cryptography;
using System.Text;
using System.Text.Json;
using Serilog;
namespace ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess;
/// <summary>
/// The launch marker and the startup orphan sweep (lmxopcua#780 design §3.3). Every worker this host launches gets a
/// marker file in this host instance's marker directory naming the worker's pid, its start time and process name,
/// and the launching host's pid and start time. A host that dies uncleanly leaves its markers behind; the next start
/// of the same instance kills exactly the processes they name — and nothing else.
/// </summary>
/// <remarks>
/// <para>
/// <b>Never "every process with this name".</b> mxaccessgw's <c>OrphanWorkerTerminator</c> killed a live
/// gateway's workers when a second gateway started. Here a marker is acted on only when (1) the host that wrote
/// it is gone — no process with its pid and start time — and (2) a process with the marker's pid, the same
/// start time and the same process name is still running. A pid that was reused by an unrelated process
/// fails (2) and is left alone; a marker written by a live host (a second instance sharing the directory, or a
/// sibling runner in this very process) fails (1) and is left alone.
/// </para>
/// <para>
/// On Windows the Job Object (<c>KILL_ON_JOB_CLOSE</c>) already takes the workers down with the host, and on
/// Linux the worker exits on pipe EOF or when its parent pid disappears. The sweep is the backstop for a worker
/// wedged so hard it noticed neither.
/// </para>
/// </remarks>
internal static class WorkerLaunchMarker
{
private const string Extension = ".worker";
/// <summary>Start times are compared with this tolerance (platforms report them at different precisions).</summary>
private static readonly TimeSpan StartTimeTolerance = TimeSpan.FromSeconds(2);
/// <summary>The marker directory of this host instance: under the PE cache when there is one, else under temp.</summary>
/// <param name="peCacheDir">The configured <c>Scripting:Worker:PeCacheDir</c>, or null.</param>
/// <returns>The directory (not created).</returns>
public static string DirectoryFor(string? peCacheDir)
{
if (!string.IsNullOrWhiteSpace(peCacheDir)) return Path.Combine(peCacheDir, "workers");
// One directory per installation (base directory), so two installs on a box do not see each other's markers.
var install = Convert.ToHexString(SHA256.HashData(Encoding.UTF8.GetBytes(AppContext.BaseDirectory)))[..16].ToLowerInvariant();
return Path.Combine(Path.GetTempPath(), "otopcua-scriptworker", "hosts", install);
}
/// <summary>Writes the marker for a just-started worker. Best effort: a failure is logged, never fatal.</summary>
/// <param name="directory">The marker directory.</param>
/// <param name="worker">The started worker.</param>
/// <param name="pipeName">Its pipe name (diagnostics).</param>
/// <param name="log">Diagnostics.</param>
/// <returns>The marker path, or null when none was written.</returns>
public static string? Write(string directory, Process worker, string pipeName, ILogger log)
{
try
{
Directory.CreateDirectory(directory);
using var host = Process.GetCurrentProcess();
var record = new MarkerRecord(
worker.Id, worker.StartTime.ToUniversalTime().Ticks, SafeName(worker),
host.Id, host.StartTime.ToUniversalTime().Ticks, pipeName);
var path = Path.Combine(directory, worker.Id.ToString(System.Globalization.CultureInfo.InvariantCulture) + Extension);
File.WriteAllText(path, JsonSerializer.Serialize(record));
return path;
}
catch (Exception ex) when (ex is IOException or UnauthorizedAccessException or InvalidOperationException
or System.ComponentModel.Win32Exception or NotSupportedException)
{
log.Warning(ex, "Could not write the ScriptWorker launch marker under {Directory}; the startup orphan sweep cannot see pid {Pid}",
directory, SafeId(worker));
return null;
}
}
/// <summary>Deletes a marker (worker ended cleanly).</summary>
/// <param name="path">The marker path, or null.</param>
public static void TryDelete(string? path)
{
if (path is null) return;
try { File.Delete(path); } catch (Exception ex) when (ex is IOException or UnauthorizedAccessException) { }
}
/// <summary>Kills the workers a dead host of this instance left behind; returns the pids killed.</summary>
/// <param name="directory">The marker directory.</param>
/// <param name="log">Diagnostics.</param>
/// <returns>The pids killed.</returns>
public static IReadOnlyList<int> Sweep(string directory, ILogger log)
{
var killed = new List<int>();
if (!Directory.Exists(directory)) return killed;
IEnumerable<string> files;
try { files = Directory.EnumerateFiles(directory, "*" + Extension).ToArray(); }
catch (Exception ex) when (ex is IOException or UnauthorizedAccessException) { return killed; }
foreach (var path in files)
{
MarkerRecord? record;
try { record = JsonSerializer.Deserialize<MarkerRecord>(File.ReadAllText(path)); }
catch (Exception ex) when (ex is IOException or UnauthorizedAccessException or JsonException)
{
record = null;
}
if (record is null)
{
TryDelete(path);
continue;
}
// (1) A live host owns its markers — another instance sharing the directory, or a sibling runner here.
if (IsSameProcess(record.HostPid, record.HostStartTicks, expectedName: null)) continue;
// (2) Only the exact process the dead host launched: same pid, same start time, same image name.
if (IsSameProcess(record.Pid, record.StartTicks, record.ProcessName))
{
try
{
using var orphan = Process.GetProcessById(record.Pid);
orphan.Kill(entireProcessTree: true);
killed.Add(record.Pid);
log.Warning("Killed orphaned ScriptWorker pid {Pid}, left behind by host pid {HostPid} (pipe {Pipe})",
record.Pid, record.HostPid, record.PipeName);
}
catch (Exception ex) when (ex is ArgumentException or InvalidOperationException or System.ComponentModel.Win32Exception)
{
log.Warning(ex, "Could not kill orphaned ScriptWorker pid {Pid}", record.Pid);
continue; // keep the marker: the next start retries
}
}
TryDelete(path);
}
return killed;
}
private static bool IsSameProcess(int pid, long startTicks, string? expectedName)
{
try
{
using var process = Process.GetProcessById(pid);
if (process.HasExited) return false;
var started = process.StartTime.ToUniversalTime().Ticks;
if (Math.Abs(started - startTicks) > StartTimeTolerance.Ticks) return false;
return expectedName is null || string.Equals(SafeName(process), expectedName, StringComparison.Ordinal);
}
catch (Exception ex) when (ex is ArgumentException or InvalidOperationException or System.ComponentModel.Win32Exception
or NotSupportedException)
{
return false;
}
}
private static string SafeName(Process process)
{
try { return process.ProcessName; }
catch (InvalidOperationException) { return string.Empty; }
}
private static int SafeId(Process process)
{
try { return process.Id; }
catch (InvalidOperationException) { return 0; }
}
/// <summary>The marker's content.</summary>
internal sealed record MarkerRecord(int Pid, long StartTicks, string ProcessName, int HostPid, long HostStartTicks, string PipeName);
}
@@ -6,13 +6,39 @@ using ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol.V1;
namespace ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess;
/// <summary>A point-in-time view of one shard (diagnostics, the readiness check, the runner-state gauge).</summary>
/// <param name="Label">The shard label (<c>0</c>…<c>K-1</c>, or <c>iso</c>).</param>
/// <param name="IsIsolation">Whether it is the isolation worker.</param>
/// <param name="Up">A worker is live and serving.</param>
/// <param name="DownFor">How long it has been down after a failure; null when up or never needed.</param>
/// <param name="Pid">The live worker's pid.</param>
/// <param name="ConsecutiveFailures">Start failures and crashes since the last healthy period.</param>
/// <param name="InFlight">Evaluations outstanding on the live worker.</param>
/// <param name="Queued">Evaluations queued in the worker (its last heartbeat).</param>
/// <param name="HeapBytes">The worker's managed heap (its last heartbeat).</param>
public sealed record ScriptWorkerShardStatus(
string Label,
bool IsIsolation,
bool Up,
TimeSpan? DownFor,
int? Pid,
int ConsecutiveFailures,
int InFlight,
int Queued,
long HeapBytes);
/// <summary>
/// Supervises one shard's worker process (design §3.3, §5.3.5): starts it lazily on first use, reports its end to
/// the runner (for attribution), and restarts it after an exponential backoff — 250 ms doubling to a 30 s cap with
/// ±20 % jitter, reset once a worker has stayed up for 60 s. It never runs a script in the host when the worker
/// cannot start (fail closed, §14 #4): <see cref="GetReadyAsync"/> returns null and the caller reports
/// <c>BadResourceUnavailable</c>.
/// the runner (for attribution, synchronously — see <see cref="WorkerConnection.StartAsync"/>), and restarts it after
/// an exponential backoff — 250 ms doubling to a 30 s cap with ±20 % jitter, reset once a worker has stayed up for
/// 60 s. It never runs a script in the host when the worker cannot start (fail closed, §14 #4):
/// <see cref="GetReadyAsync"/> returns null and the caller reports <c>BadResourceUnavailable</c>.
/// </summary>
/// <remarks>
/// A caller that got <c>Unavailable</c> waits on <see cref="WhenRecoveredAsync"/>; that both tells it when to
/// re-evaluate and makes the shard restart proactively (with backoff) instead of waiting for the next trigger, so
/// victims of a crash are re-evaluated on the shard's return (design §9).
/// </remarks>
internal sealed class WorkerShard : IAsyncDisposable
{
internal static readonly TimeSpan InitialBackoff = TimeSpan.FromMilliseconds(250);
@@ -24,20 +50,24 @@ internal sealed class WorkerShard : IAsyncDisposable
private static readonly Counter<long> ExitsCounter = OtOpcUaTelemetry.Meter.CreateCounter<long>(
"otopcua.script.worker.exits", unit: "{process}",
description: "ScriptWorker processes ended (tags: shard, reason = clean | killed_deadline | killed_heartbeat | killed_load | crash | protocol | disposed).");
description: "ScriptWorker processes ended (tags: shard, reason = clean | killed_deadline | killed_heartbeat | killed_load | crash | protocol | idle | disposed).");
private readonly Func<CancellationToken, Task<WorkerConnection>> _launch;
private readonly Func<Action<WorkerExit>, CancellationToken, Task<WorkerConnection>> _launch;
private readonly Action<WorkerShard, WorkerExit> _onExit;
private readonly ILogger _log;
private readonly object _gate = new();
private readonly Random _jitter = new();
private readonly CancellationTokenSource _disposing = new();
private WorkerConnection? _current;
private Task<WorkerConnection?>? _starting;
private TaskCompletionSource? _recovered;
private bool _recovering;
private long _backoffUntil;
private long _downSince;
private int _consecutiveFailures;
private bool _disposed;
public WorkerShard(string label, bool isolation, Func<CancellationToken, Task<WorkerConnection>> launch,
public WorkerShard(string label, bool isolation, Func<Action<WorkerExit>, CancellationToken, Task<WorkerConnection>> launch,
Action<WorkerShard, WorkerExit> onExit, ILogger log)
{
Label = label;
@@ -70,9 +100,40 @@ internal sealed class WorkerShard : IAsyncDisposable
get { lock (_gate) return _consecutiveFailures; }
}
/// <summary>How long the shard has been down after a failure; null when it is up or was never needed.</summary>
public TimeSpan? DownFor
{
get
{
lock (_gate)
{
if (_current is { IsAlive: true } || _downSince == 0) return null;
return Stopwatch.GetElapsedTime(_downSince);
}
}
}
/// <summary>A snapshot for diagnostics and health.</summary>
/// <returns>The status.</returns>
public ScriptWorkerShardStatus Status()
{
WorkerConnection? live;
int failures;
TimeSpan? downFor;
lock (_gate)
{
live = _current is { IsAlive: true } c ? c : null;
failures = _consecutiveFailures;
downFor = live is null && _downSince != 0 ? Stopwatch.GetElapsedTime(_downSince) : null;
}
var heartbeat = live?.LastHeartbeat;
return new ScriptWorkerShardStatus(Label, IsIsolation, live is not null, downFor, live?.Pid, failures,
live?.InFlightCount ?? 0, (int)(heartbeat?.Queued ?? 0), heartbeat?.HeapBytes ?? 0);
}
/// <summary>
/// The ready worker, starting one if needed. Waits at most <paramref name="grace"/> — through a start in progress
/// or a backoff — and returns null if no worker is ready by then.
/// or a backoff — and returns null if no worker is ready by then. A zero grace still kicks a start that is due.
/// </summary>
/// <param name="grace">How long to wait.</param>
/// <param name="ct">Cancellation.</param>
@@ -125,18 +186,123 @@ internal sealed class WorkerShard : IAsyncDisposable
}
}
/// <summary>
/// Completes when this shard has a live worker other than <paramref name="failed"/>. Starts a background
/// recovery (start with backoff, retried at the cap) if none is live, so the caller does not depend on a later
/// trigger to bring the shard back.
/// </summary>
/// <param name="failed">The worker the caller's evaluation died on, if any.</param>
/// <returns>A task that completes on recovery (or when the shard is disposed).</returns>
public Task WhenRecoveredAsync(WorkerConnection? failed)
{
lock (_gate)
{
if (_disposed) return Task.CompletedTask;
if (_current is { IsAlive: true } live)
{
// Still the same worker (it answered NOT_LOADED, or the caller lost a race with a restart that has
// already happened): a short pause, so a seam cannot spin on a key the worker keeps refusing.
return ReferenceEquals(live, failed) ? Task.Delay(InitialBackoff) : Task.CompletedTask;
}
_recovered ??= new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
var task = _recovered.Task;
EnsureRecoveringLocked();
return task;
}
}
private void EnsureRecoveringLocked()
{
if (_recovering || _disposed) return;
_recovering = true;
_ = Task.Run(RecoverAsync);
}
/// <summary>Stops the live worker if it has had nothing to do for <paramref name="idle"/> (the isolation worker).</summary>
/// <param name="idle">The idle threshold.</param>
/// <returns>True when a worker was stopped.</returns>
public async Task<bool> StopIfIdleAsync(TimeSpan idle)
{
WorkerConnection? victim;
// Holding the isolation gate means no evaluation can pick this worker up while it is being stopped.
var gate = IsolationGate;
try
{
if (gate is not null && !gate.Wait(0)) return false;
}
catch (ObjectDisposedException)
{
return false;
}
try
{
lock (_gate)
{
if (_disposed || _current is not { IsAlive: true } live) return false;
if (live.InFlightCount > 0 || live.LoadingCount > 0) return false;
if (Stopwatch.GetElapsedTime(live.LastUsedTimestamp) < idle) return false;
victim = live;
_current = null;
}
_log.Information("ScriptWorker[{Shard}] pid {Pid} idle for {Idle}; stopping it (it restarts on demand)", Label, victim.Pid, idle);
await victim.StopAsync(WorkerExitReason.Idle).ConfigureAwait(false);
return true;
}
finally
{
gate?.Release();
}
}
private async Task RecoverAsync()
{
try
{
while (true)
{
// A long grace: GetReadyAsync waits out the backoff itself and returns null only on a failed start.
var live = await GetReadyAsync(MaxBackoff * 2, _disposing.Token).ConfigureAwait(false);
lock (_gate)
{
if (_disposed || live is not null)
{
_recovering = false;
TakeRecoveredLocked()?.TrySetResult();
return;
}
}
}
}
catch (OperationCanceledException)
{
lock (_gate)
{
_recovering = false;
TakeRecoveredLocked()?.TrySetResult();
}
}
}
private TaskCompletionSource? TakeRecoveredLocked()
{
var tcs = _recovered;
_recovered = null;
return tcs;
}
private async Task<WorkerConnection?> StartCoreAsync()
{
await Task.Yield();
WorkerConnection? connection = null;
try
{
connection = await _launch(CancellationToken.None).ConfigureAwait(false);
connection = await _launch(Attribute, CancellationToken.None).ConfigureAwait(false);
StartsCounter.Add(1, new KeyValuePair<string, object?>("shard", Label));
_log.Information("ScriptWorker[{Shard}] started, pid {Pid}", Label, connection.Pid);
var started = connection;
_ = started.Exited.ContinueWith(t => OnExited(started, t.Result), CancellationToken.None,
TaskContinuationOptions.ExecuteSynchronously | TaskContinuationOptions.OnlyOnRanToCompletion, TaskScheduler.Default);
TaskCompletionSource? recovered;
lock (_gate)
{
_starting = null;
@@ -146,7 +312,10 @@ internal sealed class WorkerShard : IAsyncDisposable
return null;
}
_current = started;
_downSince = 0;
recovered = TakeRecoveredLocked();
}
recovered?.TrySetResult();
return started;
}
catch (Exception ex)
@@ -155,6 +324,7 @@ internal sealed class WorkerShard : IAsyncDisposable
lock (_gate)
{
_starting = null;
MarkDownLocked();
ScheduleBackoffLocked();
}
if (connection is not null) await connection.DisposeAsync().ConfigureAwait(false);
@@ -162,29 +332,44 @@ internal sealed class WorkerShard : IAsyncDisposable
}
}
/// <summary>Synchronous attribution, from <see cref="WorkerConnection"/>'s exit path, before any evaluation completes.</summary>
private void Attribute(WorkerExit exit)
{
try { _onExit(this, exit); }
catch (Exception ex) { _log.Error(ex, "ScriptWorker[{Shard}] exit handler threw", Label); }
}
private void OnExited(WorkerConnection connection, WorkerExit exit)
{
ExitsCounter.Add(1, new KeyValuePair<string, object?>("shard", Label),
new KeyValuePair<string, object?>("reason", MetricReason(exit.Reason)));
var expected = exit.Reason is WorkerExitReason.Disposed or WorkerExitReason.Idle;
lock (_gate)
{
if (ReferenceEquals(_current, connection)) _current = null;
if (!_disposed)
if (!_disposed && !expected)
{
if (Stopwatch.GetElapsedTime(connection.ReadyTimestamp) >= HealthyResetAfter) _consecutiveFailures = 0;
MarkDownLocked();
ScheduleBackoffLocked();
// A worker that was serving and died comes back by itself (after the backoff), so the scripts placed
// on it find it warm and its victims are re-evaluated on its return (design §9).
EnsureRecoveringLocked();
}
}
if (exit.Reason != WorkerExitReason.Disposed)
if (!expected)
{
_log.Warning("ScriptWorker[{Shard}] pid {Pid} ended: {Reason} (exit code {ExitCode}, culprit {Culprit}, {InFlight} evaluation(s) in flight)",
Label, connection.Pid, exit.Reason, exit.ExitCode, exit.CulpritKey ?? "none", exit.InFlightKeys.Count);
_log.Warning("ScriptWorker[{Shard}] pid {Pid} ended: {Reason} (exit code {ExitCode}, culprit {Culprit}, {InFlight} evaluation(s) and {Loading} compile(s) in flight)",
Label, connection.Pid, exit.Reason, exit.ExitCode, exit.CulpritKey ?? "none", exit.InFlightKeys.Count, exit.LoadingKeys.Count);
}
try { _onExit(this, exit); }
catch (Exception ex) { _log.Error(ex, "ScriptWorker[{Shard}] exit handler threw", Label); }
_ = connection.DisposeAsync().AsTask();
}
private void MarkDownLocked()
{
if (_downSince == 0) _downSince = Stopwatch.GetTimestamp();
}
private void ScheduleBackoffLocked()
{
_consecutiveFailures++;
@@ -212,6 +397,7 @@ internal sealed class WorkerShard : IAsyncDisposable
WorkerExitReason.KilledLoadTimeout => "killed_load",
WorkerExitReason.ProtocolFault => "protocol",
WorkerExitReason.Disposed => "disposed",
WorkerExitReason.Idle => "idle",
_ => "crash",
};
@@ -221,13 +407,18 @@ internal sealed class WorkerShard : IAsyncDisposable
{
WorkerConnection? current;
Task<WorkerConnection?>? starting;
TaskCompletionSource? recovered;
lock (_gate)
{
if (_disposed) return;
_disposed = true;
current = _current;
starting = _starting;
_current = null;
recovered = TakeRecoveredLocked();
}
recovered?.TrySetResult();
_disposing.Cancel();
if (starting is not null)
{
try { current ??= await starting.WaitAsync(TimeSpan.FromSeconds(15)).ConfigureAwait(false); } catch { /* shutting down */ }
@@ -37,6 +37,8 @@ message Hello {
uint32 hard_grace_ms = 5; // worker watchdog: fail fast past soft + grace (+ margin)
uint32 heartbeat_interval_ms = 6;
uint32 max_log_lines = 7; // outbound log buffer cap
string pe_cache_dir = 8; // the node's compiled-PE cache (design §7); empty = no cache
bytes pe_cache_key = 9; // its HMAC key; the worker loads only entries that verify under it
}
message Ready {
@@ -69,6 +71,14 @@ message LoadOutcome {
string key = 1;
bool ok = 2;
ScriptError error = 3; // COMPILE or SANDBOX when !ok
PeCache cache = 4; // how the PE cache served this load
}
enum PeCache {
PE_CACHE_NONE = 0; // no cache configured (or the load failed before it was consulted)
PE_CACHE_HIT = 1; // loaded from a verified entry; Roslyn not entered
PE_CACHE_MISS = 2; // compiled; the entry was written
PE_CACHE_REJECTED = 3; // an entry failed verification, was deleted, and the source was compiled
}
message LoadResult {
@@ -0,0 +1,292 @@
using System.Security.Cryptography;
using System.Text;
namespace ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol;
/// <summary>What a <see cref="ScriptPeCacheStore.TryRead"/> found.</summary>
public enum PeCacheLookup
{
/// <summary>No entry for the id.</summary>
Miss = 0,
/// <summary>An entry whose HMAC verified; its PE bytes are returned.</summary>
Hit = 1,
/// <summary>
/// An entry that exists but did not verify — truncated, edited, written under another key, or renamed from
/// another id. It has been deleted; the caller recompiles.
/// </summary>
Rejected = 2,
}
/// <summary>
/// The node-local compiled-PE cache (lmxopcua#780 design §7, owner decision §14 #8): content-addressed files under
/// <c>Scripting:Worker:PeCacheDir</c>, each authenticated with HMAC-SHA256 under a per-node random key, so a worker
/// restart loads its scripts' PE images instead of re-entering Roslyn — and never loads bytes it did not write.
/// </summary>
/// <remarks>
/// <list type="bullet">
/// <item><b>Layout.</b> <c>&lt;dir&gt;/pe-cache.key</c> (32 random bytes, created once by the host) and
/// <c>&lt;dir&gt;/pe/&lt;id&gt;.pe</c>, one file per entry: an 8-byte magic, the 32-byte tag, then the PE.</item>
/// <item><b>Identity.</b> <see cref="EntryId"/> is SHA-256 over the compile fingerprint (Roslyn version, the
/// sandbox and analyzer build, the reference set, the runtime), the context kind and the source. A compiler or
/// analyzer change therefore misses every old entry; it can never be served an image a newer sandbox rule
/// would have refused.</item>
/// <item><b>Authentication.</b> The tag covers the magic, the id and the PE, so an entry copied to another id
/// is rejected as well as an edited or truncated one. Loading unauthenticated PE bytes would be code execution
/// outside the analyzer; a rejected entry is deleted and the source recompiled.</item>
/// <item><b>Key.</b> Readable only by the host account: on Linux/macOS the directory is 0700 and the key file
/// 0600. On Windows the file inherits the directory's ACL, which must be the installer's (lmxopcua#780 installer
/// work). The worker receives the key in <c>Hello</c>, never on disk or in its environment.</item>
/// <item><b>Writes</b> go to a unique temp file and are renamed into place, so a concurrent reader (another
/// shard) sees the old entry, the new one, or none — never a torn file.</item>
/// </list>
/// </remarks>
public sealed class ScriptPeCacheStore
{
/// <summary>The key file's name under the cache directory.</summary>
public const string KeyFileName = "pe-cache.key";
/// <summary>The entries' subdirectory.</summary>
public const string EntriesDirectoryName = "pe";
/// <summary>The HMAC key length.</summary>
public const int KeyLength = 32;
/// <summary>An entry above this size is not read (a 4 MiB source compiles to far less).</summary>
public const int MaxEntryBytes = 32 * 1024 * 1024;
private const string EntryExtension = ".pe";
private static readonly byte[] Magic = "OTPEC\u0001\0\0"u8.ToArray();
private const int TagLength = 32;
private readonly byte[] _key;
/// <summary>Opens the store over <paramref name="directory"/> with <paramref name="key"/>.</summary>
/// <param name="directory">The cache directory (absolute).</param>
/// <param name="key">The node's HMAC key (<see cref="KeyLength"/> bytes).</param>
public ScriptPeCacheStore(string directory, byte[] key)
{
ArgumentException.ThrowIfNullOrWhiteSpace(directory);
ArgumentNullException.ThrowIfNull(key);
if (key.Length != KeyLength) throw new ArgumentException($"the PE cache key must be {KeyLength} bytes", nameof(key));
Directory = directory;
EntriesDirectory = Path.Combine(directory, EntriesDirectoryName);
_key = (byte[])key.Clone();
}
/// <summary>The cache directory.</summary>
public string Directory { get; }
/// <summary>The entries' directory.</summary>
public string EntriesDirectory { get; }
/// <summary>
/// The host half: returns the node's key, creating the directory and the key on first use. An existing key of
/// the wrong length is not replaced (it may be a sign of tampering); the call throws and the caller runs without
/// a cache.
/// </summary>
/// <param name="directory">The cache directory (absolute).</param>
/// <returns>The key.</returns>
/// <exception cref="IOException">The directory or key cannot be created or read, or the key is malformed.</exception>
public static byte[] LoadOrCreateKey(string directory)
{
ArgumentException.ThrowIfNullOrWhiteSpace(directory);
CreatePrivateDirectory(directory);
CreatePrivateDirectory(Path.Combine(directory, EntriesDirectoryName));
var keyPath = Path.Combine(directory, KeyFileName);
if (!File.Exists(keyPath))
{
// Written to a temp name and moved without overwrite: two hosts racing on first start agree on one key.
var temp = keyPath + "." + Convert.ToHexString(RandomNumberGenerator.GetBytes(8)) + ".tmp";
var options = new FileStreamOptions { Mode = FileMode.CreateNew, Access = FileAccess.Write, Share = FileShare.None };
if (!OperatingSystem.IsWindows()) options.UnixCreateMode = UnixFileMode.UserRead | UnixFileMode.UserWrite;
using (var stream = new FileStream(temp, options))
{
stream.Write(RandomNumberGenerator.GetBytes(KeyLength));
}
try
{
File.Move(temp, keyPath, overwrite: false);
}
catch (IOException) when (File.Exists(keyPath))
{
File.Delete(temp);
}
}
var key = File.ReadAllBytes(keyPath);
if (key.Length != KeyLength)
throw new IOException($"the PE cache key '{keyPath}' is {key.Length} bytes, not {KeyLength}; delete it to start a fresh cache");
return key;
}
/// <summary>The entry id for one compile: hex SHA-256 over the fingerprint, the kind and the source.</summary>
/// <param name="fingerprint">The compile fingerprint (<c>ScriptEvaluator.CacheFingerprint</c>).</param>
/// <param name="kind">The context kind (the wire enum value).</param>
/// <param name="source">The script source.</param>
/// <returns>A 64-character lowercase hex id.</returns>
public static string EntryId(ReadOnlySpan<byte> fingerprint, int kind, string source)
{
ArgumentNullException.ThrowIfNull(source);
using var hash = IncrementalHash.CreateHash(HashAlgorithmName.SHA256);
hash.AppendData(fingerprint);
hash.AppendData(BitConverter.GetBytes(kind));
hash.AppendData(Encoding.UTF8.GetBytes(source));
return Convert.ToHexString(hash.GetHashAndReset()).ToLowerInvariant();
}
/// <summary>The path of an entry.</summary>
/// <param name="entryId">The id.</param>
/// <returns>The file path.</returns>
public string EntryPath(string entryId)
{
if (!IsValidId(entryId)) throw new ArgumentException("not a PE cache entry id", nameof(entryId));
return Path.Combine(EntriesDirectory, entryId + EntryExtension);
}
/// <summary>Reads and authenticates an entry. A rejected entry is deleted.</summary>
/// <param name="entryId">The id.</param>
/// <param name="peImage">The PE bytes on a hit.</param>
/// <returns>Hit, miss, or rejected.</returns>
public PeCacheLookup TryRead(string entryId, out byte[]? peImage)
{
peImage = null;
var path = EntryPath(entryId);
byte[] bytes;
try
{
var info = new FileInfo(path);
if (!info.Exists) return PeCacheLookup.Miss;
if (info.Length > MaxEntryBytes)
{
TryDelete(path);
return PeCacheLookup.Rejected;
}
bytes = File.ReadAllBytes(path);
}
catch (FileNotFoundException)
{
return PeCacheLookup.Miss;
}
catch (DirectoryNotFoundException)
{
return PeCacheLookup.Miss;
}
if (bytes.Length <= Magic.Length + TagLength || !bytes.AsSpan(0, Magic.Length).SequenceEqual(Magic))
{
TryDelete(path);
return PeCacheLookup.Rejected;
}
var tag = bytes.AsSpan(Magic.Length, TagLength);
var pe = bytes.AsSpan(Magic.Length + TagLength);
if (!CryptographicOperations.FixedTimeEquals(tag, Tag(entryId, pe)))
{
TryDelete(path);
return PeCacheLookup.Rejected;
}
peImage = pe.ToArray();
// Recency for Prune: a hit counts as a use.
try { File.SetLastWriteTimeUtc(path, DateTime.UtcNow); } catch (IOException) { } catch (UnauthorizedAccessException) { }
return PeCacheLookup.Hit;
}
/// <summary>Writes an entry (atomically replacing any existing one).</summary>
/// <param name="entryId">The id.</param>
/// <param name="peImage">The PE bytes.</param>
/// <returns>False when the write failed (the cache is best effort: a failed write costs a recompile later).</returns>
public bool TryWrite(string entryId, ReadOnlySpan<byte> peImage)
{
var path = EntryPath(entryId);
var temp = Path.Combine(EntriesDirectory, $".{entryId}.{Convert.ToHexString(RandomNumberGenerator.GetBytes(6))}.tmp");
try
{
System.IO.Directory.CreateDirectory(EntriesDirectory);
var options = new FileStreamOptions { Mode = FileMode.CreateNew, Access = FileAccess.Write, Share = FileShare.None };
if (!OperatingSystem.IsWindows()) options.UnixCreateMode = UnixFileMode.UserRead | UnixFileMode.UserWrite;
using (var stream = new FileStream(temp, options))
{
stream.Write(Magic);
stream.Write(Tag(entryId, peImage));
stream.Write(peImage);
}
File.Move(temp, path, overwrite: true);
return true;
}
catch (Exception ex) when (ex is IOException or UnauthorizedAccessException)
{
TryDelete(temp);
return false;
}
}
/// <summary>
/// Deletes entries not used for <paramref name="maxAge"/>, then the least recently used beyond
/// <paramref name="maxEntries"/>, plus stray temp files. The design prunes "entries not in the current
/// generation"; this runner has no generation object, so recency stands in for it.
/// </summary>
/// <param name="maxAge">Entries older than this go.</param>
/// <param name="maxEntries">At most this many entries stay.</param>
/// <returns>How many files were deleted.</returns>
public int Prune(TimeSpan maxAge, int maxEntries)
{
if (!System.IO.Directory.Exists(EntriesDirectory)) return 0;
var deleted = 0;
var now = DateTime.UtcNow;
var entries = new List<FileInfo>();
foreach (var file in new DirectoryInfo(EntriesDirectory).EnumerateFiles())
{
var stale = now - file.LastWriteTimeUtc;
if (file.Name.EndsWith(".tmp", StringComparison.Ordinal))
{
if (stale > TimeSpan.FromHours(1) && TryDelete(file.FullName)) deleted++;
continue;
}
if (!file.Name.EndsWith(EntryExtension, StringComparison.Ordinal)) continue;
if (stale > maxAge)
{
if (TryDelete(file.FullName)) deleted++;
continue;
}
entries.Add(file);
}
if (entries.Count > maxEntries)
{
foreach (var file in entries.OrderBy(f => f.LastWriteTimeUtc).Take(entries.Count - maxEntries))
if (TryDelete(file.FullName)) deleted++;
}
return deleted;
}
private byte[] Tag(string entryId, ReadOnlySpan<byte> pe)
{
using var hmac = IncrementalHash.CreateHMAC(HashAlgorithmName.SHA256, _key);
hmac.AppendData(Magic);
hmac.AppendData(Encoding.ASCII.GetBytes(entryId));
hmac.AppendData(pe);
return hmac.GetHashAndReset();
}
private static bool IsValidId(string id)
=> id is { Length: 64 } && id.All(static c => c is >= '0' and <= '9' or >= 'a' and <= 'f');
private static void CreatePrivateDirectory(string path)
{
if (OperatingSystem.IsWindows()) System.IO.Directory.CreateDirectory(path);
else System.IO.Directory.CreateDirectory(path, UnixFileMode.UserRead | UnixFileMode.UserWrite | UnixFileMode.UserExecute);
}
private static bool TryDelete(string path)
{
try
{
File.Delete(path);
return true;
}
catch (Exception ex) when (ex is IOException or UnauthorizedAccessException)
{
return false;
}
}
}
@@ -97,6 +97,13 @@ public sealed record ScriptRunOutcome(ScriptRunStatus Status, object? Value, str
/// <summary>Reason when no worker became ready within <see cref="ScriptWorkerOptions.UnavailableGrace"/>.</summary>
public const string RunnerUnavailableReason = "script runner unavailable";
/// <summary>
/// For <see cref="ScriptRunStatus.Unavailable"/>: completes when the runner can run this script again (its worker
/// is back), so the seam re-evaluates it then with its latest inputs (design §9, "re-evaluated on the shard's
/// return"). Null when there is nothing to wait for.
/// </summary>
public Task? Recovery { get; init; }
/// <summary>A successful run.</summary>
/// <param name="value">The script's result.</param>
/// <returns>The outcome.</returns>
@@ -130,7 +137,7 @@ public static class ScriptRunOutcomeExtensions
{
ScriptRunStatus.Ok => Commons.Engines.VirtualTagEvalResult.Ok(outcome.Value),
ScriptRunStatus.Unavailable => Commons.Engines.VirtualTagEvalResult.Unavailable(
outcome.Reason ?? ScriptRunOutcome.RunnerUnavailableReason),
outcome.Reason ?? ScriptRunOutcome.RunnerUnavailableReason) with { RetryWhen = outcome.Recovery },
_ => Commons.Engines.VirtualTagEvalResult.Failure(outcome.Reason ?? "evaluator failure"),
};
}
@@ -70,7 +70,18 @@ public sealed class ScriptEvaluator<TContext, TResult> : IDisposable
/// <summary>Compiles user script source into an evaluator.</summary>
/// <param name="scriptSource">The user script source code to compile.</param>
/// <returns>A ready-to-run <see cref="ScriptEvaluator{TContext, TResult}"/> wrapping the compiled script.</returns>
public static ScriptEvaluator<TContext, TResult> Compile(string scriptSource)
public static ScriptEvaluator<TContext, TResult> Compile(string scriptSource) => FromImage(CompileToImage(scriptSource));
/// <summary>
/// The compile half of <see cref="Compile"/>: parse, sandbox compile, <see cref="ForbiddenTypeAnalyzer"/>, the
/// stack-guard rewrite and the emit — returning the PE bytes without loading them. The out-of-process worker
/// caches these bytes on disk (lmxopcua#780 design §7) so a restart does not re-enter Roslyn.
/// </summary>
/// <param name="scriptSource">The user script source code to compile.</param>
/// <returns>The emitted PE image.</returns>
/// <exception cref="CompilationErrorException">The source does not compile.</exception>
/// <exception cref="ScriptSandboxViolationException">The source uses a forbidden API.</exception>
public static byte[] CompileToImage(string scriptSource)
{
if (scriptSource is null) throw new ArgumentNullException(nameof(scriptSource));
@@ -142,8 +153,22 @@ public sealed class ScriptEvaluator<TContext, TResult> : IDisposable
throw new CompilationErrorException(emitErrors);
}
peStream.Position = 0;
var alc = new ScriptAssemblyLoadContext(assemblyName);
return peStream.ToArray();
}
/// <summary>
/// The load half of <see cref="Compile"/>: loads a PE image produced by <see cref="CompileToImage"/> into a fresh
/// collectible ALC and binds its <c>Run</c> method. The caller vouches for the image — only bytes this pipeline
/// emitted (the worker's HMAC-verified PE cache) may be passed, because loading arbitrary PE bytes would run
/// code the analyzer never saw.
/// </summary>
/// <param name="peImage">The image.</param>
/// <returns>A ready-to-run evaluator.</returns>
public static ScriptEvaluator<TContext, TResult> FromImage(byte[] peImage)
{
ArgumentNullException.ThrowIfNull(peImage);
using var peStream = new MemoryStream(peImage, writable: false);
var alc = new ScriptAssemblyLoadContext("ZB.MOM.WW.OtOpcUa.Core.Scripting.Compiled." + Guid.NewGuid().ToString("N"));
Assembly assembly;
try
{
@@ -182,6 +207,33 @@ public sealed class ScriptEvaluator<TContext, TResult> : IDisposable
return new ScriptEvaluator<TContext, TResult>(alc, func);
}
/// <summary>
/// A digest of everything that decides what <see cref="CompileToImage"/> emits for a given source, other than
/// the source itself (lmxopcua#780 design §7): the Roslyn version, this assembly's build identity (the
/// analyzer's deny-lists, the wrapper template, the stack-guard rewrite), the sandbox reference set, the
/// runtime version and <typeparamref name="TContext"/> / <typeparamref name="TResult"/>. A compiled-PE cache
/// keys on it, so an analyzer or compiler change invalidates every cached image and a cached PE can never skip
/// a newer sandbox rule.
/// </summary>
public static byte[] CacheFingerprint => FingerprintHolder.Value;
private static readonly Lazy<byte[]> FingerprintHolder = new(ComputeFingerprint, LazyThreadSafetyMode.ExecutionAndPublication);
private static byte[] ComputeFingerprint()
{
var sb = new StringBuilder("otopcua-script-pe/1\n");
static void AppendAssembly(StringBuilder b, System.Reflection.Assembly a)
=> b.Append(a.FullName).Append('|').Append(a.ManifestModule.ModuleVersionId.ToString("N")).Append('\n');
AppendAssembly(sb, typeof(CSharpCompilation).Assembly);
AppendAssembly(sb, typeof(Compilation).Assembly);
AppendAssembly(sb, typeof(ScriptEvaluator<,>).Assembly);
sb.Append(Environment.Version).Append('\n');
sb.Append(typeof(TContext).AssemblyQualifiedName).Append('\n');
sb.Append(typeof(TResult).AssemblyQualifiedName).Append('\n');
foreach (var identity in ScriptSandbox.ReferenceIdentities(typeof(TContext))) sb.Append(identity).Append('\n');
return System.Security.Cryptography.SHA256.HashData(Encoding.UTF8.GetBytes(sb.ToString()));
}
/// <summary>Runs the script against an already-constructed context.</summary>
/// <param name="context">The script context.</param>
/// <param name="ct">Cancellation token for the operation.</param>
@@ -74,6 +74,12 @@ public sealed class ScriptWorkerOptions
/// <summary>Default <see cref="IsolationProbationRuns"/>: 20.</summary>
public const int DefaultIsolationProbationRuns = 20;
/// <summary>Default <see cref="IsolationIdleTimeout"/>: 10 min (design §3.3).</summary>
public static readonly TimeSpan DefaultIsolationIdleTimeout = TimeSpan.FromMinutes(10);
/// <summary>Default <see cref="CircuitOpenAfter"/>: 2 min (design §5.3.6).</summary>
public static readonly TimeSpan DefaultCircuitOpenAfter = TimeSpan.FromMinutes(2);
/// <summary>Shard workers per node; a script is placed by a stable hash of its key. Clamped to 1–8.</summary>
public int Shards { get; set; } = DefaultShards;
@@ -102,9 +108,22 @@ public sealed class ScriptWorkerOptions
public int IsolationProbationRuns { get; set; } = DefaultIsolationProbationRuns;
/// <summary>
/// Directory of the node-local compiled-PE cache (design §7). Must be absolute and outside the content root
/// when set; startup fails otherwise (the family's fail-closed path rule). Reserved: the cache itself has
/// not landed yet (lmxopcua#780 P1 remainder), so a worker restart recompiles.
/// How long the isolation worker may sit with nothing to run before it is stopped (it is started again by the
/// next suspect). Default 10 min; 1 s – 24 h.
/// </summary>
public TimeSpan IsolationIdleTimeout { get; set; } = DefaultIsolationIdleTimeout;
/// <summary>
/// The crash-loop circuit (design §5.3.6): when every shard worker has been down for this long, the runner
/// reports <c>Unavailable</c>, logs one Error and answers evaluations at once (no <see cref="UnavailableGrace"/>
/// wait) while it keeps retrying at the backoff cap. Default 2 min; 1 s – 1 h.
/// </summary>
public TimeSpan CircuitOpenAfter { get; set; } = DefaultCircuitOpenAfter;
/// <summary>
/// Directory of the node-local compiled-PE cache (design §7): HMAC-authenticated PE images, so a worker restart
/// does not re-enter Roslyn. Must be absolute and outside the content root when set; startup fails otherwise
/// (the family's fail-closed path rule). Null (the default) runs without a cache: every worker start recompiles.
/// </summary>
public string? PeCacheDir { get; set; }
@@ -175,9 +194,27 @@ public sealed class ScriptWorkerOptions
probation = clamped;
}
var idle = IsolationIdleTimeout;
if (idle < TimeSpan.FromSeconds(1) || idle > TimeSpan.FromHours(24))
{
var clamped = idle < TimeSpan.FromSeconds(1) ? TimeSpan.FromSeconds(1) : TimeSpan.FromHours(24);
notes.Add($"{prefix}{nameof(IsolationIdleTimeout)} {idle} is outside 00:00:01–1.00:00:00; clamped to {clamped}.");
idle = clamped;
}
var circuit = CircuitOpenAfter;
if (circuit < TimeSpan.FromSeconds(1) || circuit > TimeSpan.FromHours(1))
{
var clamped = circuit < TimeSpan.FromSeconds(1) ? TimeSpan.FromSeconds(1) : TimeSpan.FromHours(1);
notes.Add($"{prefix}{nameof(CircuitOpenAfter)} {circuit} is outside 00:00:01–01:00:00; clamped to {clamped}.");
circuit = clamped;
}
adjustments = notes;
return new ScriptWorkerOptions
{
IsolationIdleTimeout = idle,
CircuitOpenAfter = circuit,
Shards = shards,
HardGrace = grace,
HeapLimitBytes = heap,
@@ -99,6 +99,23 @@ public static class ScriptSandbox
return new SandboxConfig(references, imports);
}
/// <summary>
/// A stable textual identity of the reference set <see cref="Build"/> compiles against for
/// <paramref name="contextType"/> — each pinned assembly's name and module version id, then each BCL path — without
/// opening any of them. Part of <see cref="ScriptEvaluator{TContext, TResult}.CacheFingerprint"/> (lmxopcua#780 §7).
/// </summary>
/// <param name="contextType">The concrete script context type.</param>
/// <returns>One line per reference, in a stable order.</returns>
public static IEnumerable<string> ReferenceIdentities(Type contextType)
{
ArgumentNullException.ThrowIfNull(contextType);
var pinned = new[] { typeof(DataValueSnapshot).Assembly, typeof(ScriptContext).Assembly, typeof(Serilog.ILogger).Assembly, contextType.Assembly }
.Distinct()
.Select(static a => $"{a.FullName}|{a.ManifestModule.ModuleVersionId:N}")
.Order(StringComparer.Ordinal);
return pinned.Concat(EnumerateBclAssemblyPaths().Order(StringComparer.Ordinal)).ToArray();
}
private static readonly System.Collections.Concurrent.ConcurrentDictionary<System.Reflection.Assembly, MetadataReference> PinnedReferences = new();
// The trusted-platform set is process-constant, so the BCL reference list is context-independent.
@@ -647,6 +647,13 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba
if (toPublish is not null || !string.Equals(st.LastFailureReason, reason, StringComparison.Ordinal))
failureToLog = reason;
st.LastFailureReason = reason;
if (result.RetryWhen is { } retryWhen && !st.RecoveryArmed)
{
st.RecoveryArmed = true;
var rawPath = def.RawPath;
_ = retryWhen.ContinueWith(_ => OnRunnerRecovered(rawPath), CancellationToken.None,
TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);
}
}
else
{
@@ -659,6 +666,24 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba
if (toPublish is { } p) Publish(def.RawPath, p.Value, p.StatusCode, pending.TimestampUtc);
}
/// <summary>
/// The out-of-process runner that could not run <paramref name="rawPath"/>'s script is serving again
/// (lmxopcua#780 design §9): queue one evaluation against the CURRENT inputs, so a calc tag with static inputs
/// does not stay BadResourceUnavailable until an input happens to change.
/// </summary>
/// <param name="rawPath">The calc tag.</param>
private void OnRunnerRecovered(string rawPath)
{
lock (_lock)
{
if (_stateByRawPath.TryGetValue(rawPath, out var st)) st.RecoveryArmed = false;
if (!_running || _disposed) return;
if (PrepareEvaluation(rawPath, DateTime.UtcNow) is not { } prepared) return;
_pending.Enqueue(prepared);
SchedulePump();
}
}
/// <summary>Record a successful evaluation. MUST be called under <see cref="_lock"/>; returns what to
/// publish, or null when the result is deduped.
/// <para><b>The dedup key is (value, status), not value alone</b> — generalising the
@@ -852,6 +877,9 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba
/// <summary>The reason of the last failed evaluation (null after a success / inputs-Bad hold) — the
/// gate that keeps the script-log entry to the transition and to a CHANGE of reason.</summary>
public string? LastFailureReason;
/// <summary>A runner-recovery re-evaluation is armed (lmxopcua#780): at most one per tag (latest-wins).</summary>
public bool RecoveryArmed;
}
private sealed record CalculationSubscriptionHandle(string DiagnosticId) : ISubscriptionHandle;
@@ -29,6 +29,10 @@ public sealed class ScriptRunnerSelection : IAsyncDisposable, IDisposable
/// <summary>The out-of-process runner, when one is configured.</summary>
public IScriptRunner? OutOfProcessRunner => _runner;
/// <summary>The out-of-process runner's status (the readiness check), or null when every seam runs in process.</summary>
/// <returns>The status, or null.</returns>
public ScriptRunnerStatus? Status() => _runner?.Status();
/// <summary>The runner <paramref name="seam"/> must use, or null for its in-process path.</summary>
/// <param name="seam">The seam.</param>
/// <returns>The runner, or null.</returns>
@@ -62,6 +62,16 @@ public static class HealthEndpoints
sp => new OpcUaEndpointHealthCheck(sp.GetRequiredService<IOpcUaEndpointState>()),
failureStatus: null,
tags: tags));
// lmxopcua#780 design §10 / owner decision §14 #12: the out-of-process script runner is a READY-tier
// component that is at worst Degraded (still 200): OPC UA keeps serving driver data while scripts are
// Bad, so a sick runner must never fail readiness (on a pair that would move ServiceLevel to the peer).
// Healthy when every seam runs in process. Not on the active tier.
checks.Add(new HealthCheckRegistration(
ScriptRunnerHealthCheck.Name,
sp => new ScriptRunnerHealthCheck(sp.GetService<ScriptRunnerSelection>()),
failureStatus: null,
tags: new[] { ZbHealthTags.Ready }));
}
// configdb probe is admin-only (per-cluster mesh Phase 4). A driver-only node gets its config
@@ -0,0 +1,68 @@
using Microsoft.Extensions.Diagnostics.HealthChecks;
using ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess;
using ZB.MOM.WW.OtOpcUa.Host.Configuration;
namespace ZB.MOM.WW.OtOpcUa.Host.Health;
/// <summary>
/// Pure decision function for the script-runner probe (lmxopcua#780 design §10, owner decision §14 #12), factored
/// out of <see cref="ScriptRunnerHealthCheck"/> so the matrix is table-testable without worker processes.
/// </summary>
public static class ScriptRunnerHealthDecision
{
/// <summary>
/// Maps the runner's status to a health status. <b>Never Unhealthy:</b> a sick script runner does not stop the
/// node serving driver data over OPC UA, so it must not fail readiness — on a redundant pair that would shift
/// ServiceLevel to the peer over a script problem. Degraded is still ready (200) but visible.
/// </summary>
/// <param name="status">The out-of-process runner's status, or null when every seam runs in process.</param>
/// <returns>The status, a description and the per-shard data.</returns>
public static (HealthStatus Status, string Description, IReadOnlyDictionary<string, object> Data) Evaluate(ScriptRunnerStatus? status)
{
if (status is null)
return (HealthStatus.Healthy, "Scripts run in process (Scripting:Runner=InProcess).", new Dictionary<string, object>());
var data = new Dictionary<string, object>(StringComparer.Ordinal)
{
["state"] = status.State.ToString(),
["suspects"] = status.SuspectCount,
["quarantined"] = status.QuarantinedCount,
["peCache"] = status.PeCacheEnabled,
};
foreach (var shard in status.Shards)
{
data[$"shard.{shard.Label}"] = shard.Up
? $"up pid {shard.Pid}, {shard.InFlight} in flight"
: shard.DownFor is { } downFor ? $"down {downFor.TotalSeconds:F0} s ({shard.ConsecutiveFailures} failure(s))" : "not started";
}
var down = status.Shards.Where(s => s.DownFor is not null).Select(s => s.Label).ToArray();
return status.State switch
{
ScriptRunnerState.Ready => (HealthStatus.Healthy, "Out-of-process script runner ready.", data),
ScriptRunnerState.Degraded => (HealthStatus.Degraded,
$"Out-of-process script runner degraded: ScriptWorker {string.Join(", ", down)} down; scripts placed there report BadResourceUnavailable until it restarts.",
data),
_ => (HealthStatus.Degraded,
"Out-of-process script runner unavailable (circuit open): every ScriptWorker shard is down; every compiled script reports BadResourceUnavailable. OPC UA keeps serving driver data.",
data),
};
}
}
/// <summary>The script-runner readiness component: Healthy, or Degraded-but-ready (see <see cref="ScriptRunnerHealthDecision"/>).</summary>
/// <param name="selection">The node's runner selection; null on nodes that never register one.</param>
public sealed class ScriptRunnerHealthCheck(ScriptRunnerSelection? selection) : IHealthCheck
{
/// <summary>The check's registration name.</summary>
public const string Name = "script-runner";
/// <inheritdoc />
public Task<HealthCheckResult> CheckHealthAsync(HealthCheckContext context, CancellationToken cancellationToken = default)
{
var (status, description, data) = ScriptRunnerHealthDecision.Evaluate(selection?.Status());
return Task.FromResult(status == HealthStatus.Healthy
? HealthCheckResult.Healthy(description, data)
: HealthCheckResult.Degraded(description, data: data));
}
}
@@ -51,6 +51,12 @@ public sealed class VirtualTagActor : ReceiveActor, IWithUnboundedStash
/// when the tag has a <c>TimerIntervalMs</c>.</summary>
private sealed class TimerTick { public static readonly TimerTick Instance = new(); private TimerTick() { } }
/// <summary>
/// The out-of-process script runner that could not run this tag's script is serving again (lmxopcua#780 design
/// §9): re-evaluate once against the latest inputs. Private to the actor.
/// </summary>
private sealed class RunnerRecovered { public static readonly RunnerRecovered Instance = new(); private RunnerRecovered() { } }
/// <summary>The asynchronous evaluation started by <see cref="Evaluate"/> finished. Private to the actor.</summary>
private sealed record EvaluationCompleted(
Task<VirtualTagEvalResult> Evaluation, uint WorstInputStatus, DateTime TimestampUtc);
@@ -108,6 +114,8 @@ public sealed class VirtualTagActor : ReceiveActor, IWithUnboundedStash
private readonly bool _changeTriggered;
private readonly int? _timerIntervalMs;
private ICancelable? _timer;
// One outstanding RunnerRecovered wait at most: a burst of Unavailable results re-evaluates once (latest-wins).
private bool _recoveryArmed;
// R02:P6 — a script failing on every trigger of a 1 Hz input published one cluster-wide DPS
// ScriptLogEntry per second per tag; 200 such tags shed everyone's entries out of the bounded
// telemetry channels. The per-failure entry is now rate-limited per tag (20/min + one summary).
@@ -212,6 +220,11 @@ public sealed class VirtualTagActor : ReceiveActor, IWithUnboundedStash
Receive<DependencyValueBatch>(OnDependencyBatch);
Receive<ReassertValue>(_ => OnReassertValue());
Receive<TimerTick>(_ => Evaluate(DateTime.UtcNow));
Receive<RunnerRecovered>(_ =>
{
_recoveryArmed = false;
Evaluate(DateTime.UtcNow);
});
// Only reachable for a completion that outlived its behaviour (a supervision restart mid-evaluation
// resets the behaviour stack): the incarnation it belonged to is gone, so it is dropped.
Receive<EvaluationCompleted>(_ => { });
@@ -368,6 +381,21 @@ public sealed class VirtualTagActor : ReceiveActor, IWithUnboundedStash
ReceiveAny(_ => Stash.Stash());
}
/// <summary>
/// A runner failure (BadResourceUnavailable) carries the task that completes when the runner is back: evaluate
/// again then, with whatever inputs are current, instead of holding the Bad until the next input change — which,
/// for a static input, would never come (lmxopcua#780 design §9, "re-evaluated on the shard's return").
/// </summary>
/// <param name="retryWhen">The recovery task, or null.</param>
private void ArmRecovery(Task? retryWhen)
{
if (retryWhen is null || _recoveryArmed) return;
_recoveryArmed = true;
var self = Self;
retryWhen.ContinueWith(_ => self.Tell(RunnerRecovered.Instance),
CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);
}
private void OnEvaluatorThrew(Exception ex, DateTime timestampUtc)
{
_log.Warning(ex, "VirtualTag {Id}: evaluator threw", _virtualTagId);
@@ -393,6 +421,7 @@ public sealed class VirtualTagActor : ReceiveActor, IWithUnboundedStash
{
OnEvaluationFailed(result.Reason ?? "evaluator failure", "Warning", timestampUtc,
result.StatusCode ?? OpcUaStatus.BadInternalError);
ArmRecovery(result.RetryWhen);
return;
}
@@ -44,6 +44,7 @@ internal sealed class WorkerHost
private int _queueCapacity = 4096;
private TimeSpan _hardGrace = TimeSpan.FromSeconds(1);
private TimeSpan _heartbeatInterval = TimeSpan.FromSeconds(1);
private ScriptPeCacheStore? _peCache;
private long _seq;
public WorkerHost(NamedPipeClientStream pipe, string nonce, int parentPid)
@@ -102,6 +103,11 @@ internal sealed class WorkerHost
if (hello.QueueCapacity > 0) _queueCapacity = (int)hello.QueueCapacity;
if (hello.HardGraceMs > 0) _hardGrace = TimeSpan.FromMilliseconds(hello.HardGraceMs);
if (hello.HeartbeatIntervalMs > 0) _heartbeatInterval = TimeSpan.FromMilliseconds(hello.HeartbeatIntervalMs);
if (!string.IsNullOrEmpty(hello.PeCacheDir) && hello.PeCacheKey.Length == ScriptPeCacheStore.KeyLength)
{
// Design §7: the host owns the key; it reaches us only here, never through the environment or a file.
_peCache = new ScriptPeCacheStore(hello.PeCacheDir, hello.PeCacheKey.ToByteArray());
}
}
private void StartBackground(int evalThreads)
@@ -181,7 +187,8 @@ internal sealed class WorkerHost
var outcome = new LoadOutcome { Key = script.Key, Ok = true };
try
{
var loaded = LoadedScript.Compile(script);
var loaded = LoadedScript.Load(script, _peCache, out var cache);
outcome.Cache = cache;
if (_scripts.TryRemove(script.Key, out var previous)) previous.Dispose();
_scripts[script.Key] = loaded;
}
@@ -499,15 +506,52 @@ internal sealed class WorkerHost
public string[] InputPaths { get; }
public static LoadedScript Compile(ScriptSource source) => source.Kind switch
/// <summary>
/// Loads a script: from a verified PE cache entry when there is one (Roslyn is not entered), otherwise by
/// compiling it and writing the entry (design §7). A compile error is never cached — the host's negative
/// cache covers that — and an entry that fails verification is deleted by the store and recompiled.
/// </summary>
public static LoadedScript Load(ScriptSource source, ScriptPeCacheStore? cache, out PeCache how) => source.Kind switch
{
ContextKind.VirtualTag or ContextKind.Calculation => new LoadedScript(
[.. source.InputPaths], ScriptEvaluator<VirtualTagContext, object?>.Compile(source.Source), null),
[.. source.InputPaths], Load<VirtualTagContext, object?>(source, cache, out how), null),
ContextKind.AlarmPredicate => new LoadedScript(
[.. source.InputPaths], null, ScriptEvaluator<AlarmPredicateContext, bool>.Compile(source.Source)),
[.. source.InputPaths], null, Load<AlarmPredicateContext, bool>(source, cache, out how)),
_ => throw new ScriptWorkerProtocolException($"unknown script kind {source.Kind}"),
};
private static ScriptEvaluator<TContext, TResult> Load<TContext, TResult>(ScriptSource source, ScriptPeCacheStore? cache,
out PeCache how) where TContext : ScriptContext
{
if (cache is null)
{
how = PeCache.None;
return ScriptEvaluator<TContext, TResult>.Compile(source.Source);
}
var id = ScriptPeCacheStore.EntryId(ScriptEvaluator<TContext, TResult>.CacheFingerprint, (int)source.Kind, source.Source);
var lookup = cache.TryRead(id, out var image);
if (lookup == PeCacheLookup.Hit)
{
try
{
how = PeCache.Hit;
return ScriptEvaluator<TContext, TResult>.FromImage(image!);
}
catch (Exception ex) when (ex is BadImageFormatException or InvalidOperationException or TypeLoadException or MissingMethodException)
{
// Authenticated but unloadable: only possible across a runtime change the fingerprint missed.
try { File.Delete(cache.EntryPath(id)); } catch (IOException) { }
lookup = PeCacheLookup.Rejected;
}
}
var compiled = ScriptEvaluator<TContext, TResult>.CompileToImage(source.Source);
cache.TryWrite(id, compiled);
how = lookup == PeCacheLookup.Rejected ? PeCache.Rejected : PeCache.Miss;
return ScriptEvaluator<TContext, TResult>.FromImage(compiled);
}
public object? Run(IReadOnlyDictionary<string, DataValueSnapshot> inputs, ILogger logger, Func<DateTime> clock,
CancellationToken ct, out bool isPredicate)
{