diff --git a/scripts/validation/opcua-lifecycle-memory/SATURATION.md b/scripts/validation/opcua-lifecycle-memory/SATURATION.md new file mode 100644 index 00000000..d9b12a44 --- /dev/null +++ b/scripts/validation/opcua-lifecycle-memory/SATURATION.md @@ -0,0 +1,114 @@ +# Candidate saturation experiment — review before launch + +This is a prospective diagnostic protocol, not an acceptance run or a replacement +for the failed cold server soak. Production code, the cycle-3 resource baseline, +and the +64 MiB managed / +128 MiB RSS / +32 descriptor caps remain unchanged. +Only `OTOPCUA_SDK2_SOAK_MODE=saturation` selects this behavior; the existing +`diagnostic`, `profile`, and `full` modes keep their existing stopping rules. + +The fixed canonical phase performs exactly 384 lifecycle cycles, with the +existing active-resource restart after every fourth cycle: 96 completed restarts. +It then performs exactly one settling cycle, without another restart. The phase, +including observer handshakes, must finish within 1,800 seconds. Initial fixture +startup has a separate 120-second bound. There is no early completion, extension, +RSS-selected endpoint, budget reset, or two-hour/full-pass flag. Cleanup remains +bounded independently of the phase deadline. + +All canonical operations remain in use: encrypted opc.tcp sessions, namespace URI +binding, scalar/typed-array read/write, subscriptions, browse abandonment, history +continuation churn/release/cross-session/close, surgical changes/rebuilds, key +revocation, paused network reads/recovery, and same-PKI server restart. All +ownership, continuation, PKI-descriptor, store and node inventory checks remain +active. Cold budget rejection is latched at its original observation and rethrown, +even if diagnostic readiness later succeeds. A rejected readiness receipt does +not become a passing test or a new resource baseline. + +## Observer boundaries + +A fresh task-owned control directory and UUID bind every request/receipt to this +run and target PID. Files are published atomically; existing, malformed, foreign, +or incomplete receipts fail closed. The observer is a separate diagnostic process +using public `DiagnosticsClient` / EventPipe APIs. Product dependencies are +unchanged. + +One continuous trace covers both fixed windows, including their boundary gap: + +| Marker | Before/after operation | Completed restarts | +|---|---|---:| +| window1-start | Before cycle 321 | 80 | +| window1-end | After cycle 352 and its restart | 88 | +| window2-start | Before cycle 353 | 88 | +| window2-end | After cycle 384 and its restart | 96 | + +The target waits for the observer's active receipt and enabled marker provider +before its first window operation. The parser requires exactly these four +markers, the correct UUID and cycle/restart values, and strictly ordered +timestamps. A continuous veto also covers events outside the marker ranges +within the attached interval. Both windows and the entire attached interval +require zero `Opc.Ua.OpcUaExtensions` JIT starts and method-load events, zero new +`Opc.Ua.*` assembly-load events, and zero lost events. Other JIT counts and loaded +native method-byte sums are reported for review; they do not establish that all +runtime paths are warmed. Method-size sums do not equal resident memory. + +Runtime capture is Verbose `Microsoft-Windows-DotNETRuntime`, keywords `0x18` +(Loader + JIT), plus Informational `OtOpcUa-Sdk2-Saturation` markers, no rundown, +and a prospectively fixed **16 MiB** circular-buffer capacity. This is a new +experiment and does not reinterpret the earlier 256 MiB-capacity traces. + +After restart 96, the target requests stop and waits. The observer must stop its +session, drain the event stream, close it, parse the entire trace, and atomically +publish a PID/run-bound drained receipt. Its Stop/copy/dispose/parse stages have +10/10/5/10-second bounds. Target receipt waits are bounded at 45 seconds and remain +inside the 1,800-second phase deadline. Observer faults propagate immediately. +The target also verifies its marker provider is disabled before the settling +cycle. Missing markers, no events, event loss, or malformed stream cannot prove +readiness. A parser failure produces a failed observer receipt, not empty counts. + +The `saturation-result` separates readiness, cold budget rejection, fixed operation +completion and verified shutdown ownership. It always records `acceptance=false`. +It is emitted before rethrowing the original cold assertion. The launcher retains +nonzero test/container exits and records diagnostic readiness separately. + +## Prepare immutable Linux artifacts + +Commit the reviewed source first. Preparation refuses a dirty checkout and +publishes fresh, separate app/observer artifacts for `linux-arm64`, self-contained +.NET **10.0.14**, without trimming or single-file publishing. It never starts the +experiment. The bundle records source revision, source hashes, exact publish +command, fixed protocol, image digest and SHA-256 of every app/observer/tool file. + +```bash +python3 scripts/validation/opcua-lifecycle-memory/prepare_saturation.py \ + --output /tmp/opcua-saturation-reviewed \ + --libraries "$diagnosticToolLibraries" \ + --image sha256:b04611ee9e1b7d36797012ed921962c57189cb4dbb3dff12afb8e158110f293b +``` + +`diagnosticToolLibraries` is the installed dotnet-trace 9.0.661903 +`tools/net8.0/any` directory. No product package reference is added. Artifact +preparation and review are prerequisites; root and Astra must review the actual +code, control tests and receipt before launch. The coordinator owns the long +execution. + +```bash +python3 /tmp/opcua-saturation-reviewed/tools/launch_saturation.py \ + --bundle /tmp/opcua-saturation-reviewed --receipt-sha256 "$reviewedReceiptHash" \ + --output /tmp/opcua-saturation-run --verify-only +# After root/Astra review, root runs the same command without --verify-only. +``` + +The launcher rechecks every immutable file before and after the run. It creates a +fresh offline task-owned container, writable output/PKI directories, and read-only +app/observer/tool mounts. Descriptor controls run first in a separate process. +TRX guards require two descriptor tests and one canonical test actually executed, +with no skipped results. The launcher verifies the actual target and observer +mapped their own self-contained `libcoreclr.so`; the older SDK-image controller +runtime is not the target runtime. A 2,040-second outer timeout includes startup, +phase and bounded cleanup; it does not extend the 1,800-second phase. + +No long saturation run has been launched during implementation. Existing cold +failures remain untouched. No original-SDK full-workload adapter, warmed +acceptance baseline, additional two-hour server run, or production correction is +part of this change. The separately demonstrated Mac native certificate-related +inventory growth remains unresolved even if generated-factory compilation +saturates. diff --git a/scripts/validation/opcua-lifecycle-memory/SaturationObserver.cs b/scripts/validation/opcua-lifecycle-memory/SaturationObserver.cs new file mode 100644 index 00000000..52ee22a0 --- /dev/null +++ b/scripts/validation/opcua-lifecycle-memory/SaturationObserver.cs @@ -0,0 +1,127 @@ +// Diagnostic-only observer. The target remains the unchanged canonical lifecycle test process. +using System.Diagnostics.Tracing; +using System.Text.Json; +using Microsoft.Diagnostics.NETCore.Client; +using Microsoft.Diagnostics.Tracing; + +const string protocol = "sdk2-saturation-v1"; +const string markerProvider = "OtOpcUa-Sdk2-Saturation"; +var control = Path.GetFullPath(args[0]); +var runId = args[1]; +if (!Guid.TryParseExact(runId, "D", out _)) throw new InvalidDataException("Malformed run identity"); +using var lifetime = new CancellationTokenSource(TimeSpan.FromSeconds(1980)); +var ct = lifetime.Token; +try +{ + var request = await WaitJson("attach-request.json", ct); + using (request) + { + Identity(request.RootElement); + Require(request.RootElement.GetProperty("firstCycle").GetInt32() == 321 + && request.RootElement.GetProperty("completedRestarts").GetInt32() == 80, "Invalid attach boundary"); + var targetPid = request.RootElement.GetProperty("targetPid").GetInt32(); + Require(targetPid > 0 && targetPid != Environment.ProcessId, "Invalid target process"); + var providers = new[] + { + new EventPipeProvider("Microsoft-Windows-DotNETRuntime", EventLevel.Verbose, 0x18), + new EventPipeProvider(markerProvider, EventLevel.Informational) + }; + var client = new DiagnosticsClient(targetPid); + var session = await Task.Run(() => client.StartEventPipeSession(providers, requestRundown: false, circularBufferMB: 16), ct) + .WaitAsync(TimeSpan.FromSeconds(10), ct); + var trace = Path.Combine(control, "readiness.nettrace"); + await using (var output = new FileStream(trace, FileMode.CreateNew, FileAccess.Write, FileShare.Read)) + { + var copy = session.EventStream.CopyToAsync(output, ct); + await Publish("active.json", new { protocol, runId, targetPid, observerPid = Environment.ProcessId, active = true }); + using var stop = await WaitJson("detach-request.json", ct); + Identity(stop.RootElement); + Require(stop.RootElement.GetProperty("targetPid").GetInt32() == targetPid + && stop.RootElement.GetProperty("lastCycle").GetInt32() == 384 + && stop.RootElement.GetProperty("completedRestarts").GetInt32() == 96, "Invalid detach boundary"); + await Task.Run(session.Stop, ct).WaitAsync(TimeSpan.FromSeconds(10), ct); + await copy.WaitAsync(TimeSpan.FromSeconds(10), ct); + await output.FlushAsync(ct); + } + await Task.Run(session.Dispose, ct).WaitAsync(TimeSpan.FromSeconds(5), ct); + var summary = await Task.Run(() => Parse(trace, runId), ct).WaitAsync(TimeSpan.FromSeconds(10), ct); + // Published only after Stop, stream drain, disposal and complete parsing. Never just a file-exists ACK. + await Publish("drained.json", new + { + protocol, runId, targetPid, observerPid = Environment.ProcessId, active = false, drained = true, + summary.markersValid, summary.eventsLost, summary.sdkFactoryJit, summary.sdkFactoryLoads, + summary.sdkAssemblyLoads, summary.windows, summary.otherJit, summary.loadedNativeBytes, + configuration = new { runtimeProvider = "Microsoft-Windows-DotNETRuntime", level = "Verbose", keywords = "0x18", + markerProvider, circularBufferMiB = 16, requestRundown = false }, + acceptance = false + }); + } +} +catch (Exception error) +{ + await Publish("observer-failed.json", new { protocol, runId, exceptionType = error.GetType().FullName }); + // An IPC/Stop worker can still be blocked. The parent also bounds and reaps this separate process. + Environment.Exit(1); +} + +void Identity(JsonElement value) => Require(value.GetProperty("protocol").GetString() == protocol + && value.GetProperty("runId").GetString() == runId, "Mismatched observer identity"); +static void Require(bool condition, string message) { if (!condition) throw new InvalidDataException(message); } +async Task WaitJson(string name, CancellationToken token) +{ + var path = Path.Combine(control, name); + while (!File.Exists(path)) await Task.Delay(25, token); + return JsonDocument.Parse(await File.ReadAllTextAsync(path, token)); +} +async Task Publish(string name, object value) +{ + var temporary = Path.Combine(control, name + ".tmp"); + await using (var stream = new FileStream(temporary, FileMode.CreateNew, FileAccess.Write)) + await JsonSerializer.SerializeAsync(stream, value); + File.Move(temporary, Path.Combine(control, name)); +} +static TraceResult Parse(string path, string expectedRun) +{ + var markers = new List(); + var events = new List(); + using var source = new EventPipeEventSource(path); + source.Clr.MethodJittingStarted += data => events.Add(new(data.TimeStampRelativeMSec, + data.MethodNamespace == "Opc.Ua.OpcUaExtensions" ? "factory-jit" : "other-jit", 0)); + source.Clr.MethodLoadVerbose += data => events.Add(new(data.TimeStampRelativeMSec, + data.MethodNamespace == "Opc.Ua.OpcUaExtensions" ? "factory-load" : "other-load", data.MethodSize)); + source.Clr.LoaderAssemblyLoad += data => + { + if (data.FullyQualifiedAssemblyName.StartsWith("Opc.Ua.", StringComparison.Ordinal)) + events.Add(new(data.TimeStampRelativeMSec, "sdk-assembly", 0)); + }; + source.Dynamic.All += data => + { + if (data.ProviderName == "OtOpcUa-Sdk2-Saturation" && data.EventName == "Boundary") + markers.Add(new(data.TimeStampRelativeMSec, (string)data.PayloadByName("runId"), + (string)data.PayloadByName("name"), (int)data.PayloadByName("cycle"), (int)data.PayloadByName("restarts"))); + }; + source.Process(); + (string Name, int Cycle, int Restarts)[] expected = + [ ("window1-start", 321, 80), ("window1-end", 352, 88), + ("window2-start", 353, 88), ("window2-end", 384, 96) ]; + var valid = markers.Count == 4 && markers.Select((marker, index) => + marker.Run == expectedRun && (marker.Name, marker.Cycle, marker.Restarts) == expected[index]).All(x => x) + && markers.Zip(markers.Skip(1)).All(pair => pair.First.Ms < pair.Second.Ms); + var windows = Enumerable.Range(0, 2).Select(index => + { + var selected = valid ? events.Where(x => x.Ms >= markers[index * 2].Ms && x.Ms <= markers[index * 2 + 1].Ms).ToArray() : []; + return new Window($"window{index + 1}", 321 + index * 32, 352 + index * 32, 80 + index * 8, 88 + index * 8, + selected.Count(x => x.Kind == "factory-jit"), selected.Count(x => x.Kind == "factory-load"), + selected.Count(x => x.Kind == "sdk-assembly")); + }).ToArray(); + // Totals include the whole attached interval, including marker gaps, startup and stopping the collector. + return new TraceResult(valid, source.EventsLost, events.Count(x => x.Kind == "factory-jit"), + events.Count(x => x.Kind == "factory-load"), events.Count(x => x.Kind == "sdk-assembly"), windows, + events.Count(x => x.Kind == "other-jit"), events.Sum(x => x.Bytes)); +} +record Marker(double Ms, string Run, string Name, int Cycle, int Restarts); +record Observed(double Ms, string Kind, long Bytes); +record Window(string name, int firstCycle, int lastCycle, int startRestarts, int endRestarts, + int sdkFactoryJit, int sdkFactoryLoads, int sdkAssemblyLoads); +record TraceResult(bool markersValid, int eventsLost, int sdkFactoryJit, int sdkFactoryLoads, + int sdkAssemblyLoads, Window[] windows, int otherJit, long loadedNativeBytes); diff --git a/scripts/validation/opcua-lifecycle-memory/build_saturation_observer.py b/scripts/validation/opcua-lifecycle-memory/build_saturation_observer.py new file mode 100644 index 00000000..1b374ef3 --- /dev/null +++ b/scripts/validation/opcua-lifecycle-memory/build_saturation_observer.py @@ -0,0 +1,46 @@ +#!/usr/bin/env python3 +"""Build the opt-in saturation observer without changing product dependencies.""" +import argparse +import json +from pathlib import Path +import subprocess +import xml.etree.ElementTree as ET + + +def build(output, libraries, linux): + output = output.resolve() + output.mkdir(parents=True, exist_ok=False) + project = ET.Element("Project", Sdk="Microsoft.NET.Sdk") + properties = ET.SubElement(project, "PropertyGroup") + for key, value in {"OutputType": "Exe", "TargetFramework": "net10.0", "ImplicitUsings": "enable", + "TreatWarningsAsErrors": "true", "EnableDefaultCompileItems": "false"}.items(): + ET.SubElement(properties, key).text = value + items = ET.SubElement(project, "ItemGroup") + ET.SubElement(items, "FrameworkReference", Include="Microsoft.AspNetCore.App") + ET.SubElement(items, "Compile", Include=str(Path(__file__).resolve().parent / "SaturationObserver.cs")) + for name in ["Microsoft.Diagnostics.NETCore.Client", "Microsoft.Diagnostics.Tracing.TraceEvent", "Microsoft.Diagnostics.FastSerialization"]: + path = (libraries / (name + ".dll")).resolve(strict=True) + reference = ET.SubElement(items, "Reference", Include=name) + ET.SubElement(reference, "HintPath").text = str(path) + project_path = output / "observer.csproj" + ET.indent(project) + ET.ElementTree(project).write(project_path, encoding="unicode") + command = ["dotnet", "publish" if linux else "build", str(project_path), "-c", "Release", "-m:2", "--nologo"] + if linux: + command += ["-r", "linux-arm64", "--self-contained", "true", "-p:RuntimeFrameworkVersion=10.0.14", + "-p:PublishSingleFile=false", "-p:PublishTrimmed=false", "-p:UseAppHost=true", "-o", str(output / "published")] + with (output / "build.log").open("w") as log: + result = subprocess.run(command, stdout=log, stderr=subprocess.STDOUT) + (output / "build-command.json").write_text(json.dumps(command, indent=2) + "\n") + if result.returncode: + raise subprocess.CalledProcessError(result.returncode, command) + return output / ("published" if linux else "bin/Release/net10.0") + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--output", required=True, type=Path) + parser.add_argument("--libraries", required=True, type=Path) + parser.add_argument("--linux-self-contained", action="store_true") + args = parser.parse_args() + print(build(args.output, args.libraries, args.linux_self_contained)) diff --git a/scripts/validation/opcua-lifecycle-memory/container_saturation.sh b/scripts/validation/opcua-lifecycle-memory/container_saturation.sh new file mode 100644 index 00000000..c3aae42a --- /dev/null +++ b/scripts/validation/opcua-lifecycle-memory/container_saturation.sh @@ -0,0 +1,63 @@ +#!/usr/bin/env bash +# Diagnostic only. The published xUnit apphost and observer are self-contained .NET 10.0.14. +set -euo pipefail +mkdir -p /task/tmp /task/data /task/results /task/control +export TMPDIR=/task/tmp XDG_DATA_HOME=/task/data +export OTOPCUA_SDK2_SOAK=1 OTOPCUA_SDK2_SOAK_MODE=saturation OTOPCUA_SDK2_SOAK_SECONDS=1800 +export OTOPCUA_SDK2_SOAK_OUTPUT=/task/resources.jsonl +export OTOPCUA_SDK2_SATURATION_CONTROL=/task/control +export OPCUA_CERTIFICATE_LEAK_TRACKING=1 +export DOTNET_CLI_TELEMETRY_OPTOUT=1 DOTNET_NOLOGO=1 +[[ $(uname -s) == Linux && $(uname -m) == aarch64 && -d /proc/self/fd ]] +dotnet --info > /task/controller-runtime.txt +vstest_pid= observer_pid= +cleanup() { + for child in "$vstest_pid" "$observer_pid"; do + if [[ -n "$child" ]] && kill -0 "$child" 2>/dev/null; then kill "$child" 2>/dev/null || true; fi + done +} +trap cleanup EXIT +dotnet vstest /app/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests.dll \ + --TestCaseFilter:'FullyQualifiedName~Sdk2LifecycleSoakTests.Descriptor_probe_' \ + --ResultsDirectory:/task/results '--Logger:trx;LogFileName=descriptor-controls.trx' > /task/descriptor-controls.log 2>&1 +/observer/observer /task/control "$OTOPCUA_SDK2_SATURATION_RUN_ID" > /task/observer.log 2>&1 & +observer_pid=$! +dotnet vstest /app/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests.dll \ + --TestCaseFilter:'FullyQualifiedName=ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests.Sdk2LifecycleSoakTests.Repeated_lifecycles_return_resources_to_a_steady_baseline' \ + --ResultsDirectory:/task/results '--Logger:trx;LogFileName=saturation.trx' > /task/test.log 2>&1 & +vstest_pid=$! +observed=0 +for attempt in $(seq 1 1200); do + if [[ -s /task/resources.jsonl ]]; then + target_pid=$(head -n 1 /task/resources.jsonl | sed -n 's/.*"processId":\([0-9]*\),.*/\1/p') + [[ -n "$target_pid" ]] + target_exe=$(readlink "/proc/$target_pid/exe") + [[ "$target_exe" == /app/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests ]] + printf 'targetPid=%s\nactualExecutable=%s\n' "$target_pid" "$target_exe" > /task/target-runtime.txt + awk '$6 ~ /\/(libcoreclr.so|libhostfxr.so|libhostpolicy.so|System.Private.CoreLib.dll)$/ {print $6}' "/proc/$target_pid/maps" | sort -u >> /task/target-runtime.txt + grep -Fx /app/libcoreclr.so /task/target-runtime.txt > /dev/null + [[ $(readlink "/proc/$observer_pid/exe") == /observer/observer ]] + awk '$6 ~ /\/libcoreclr.so$/ {print $6}' "/proc/$observer_pid/maps" | sort -u > /task/observer-runtime.txt + grep -Fx /observer/libcoreclr.so /task/observer-runtime.txt > /dev/null + observed=1 + break + fi + kill -0 "$vstest_pid" 2>/dev/null || break + sleep 0.1 +done +[[ "$observed" == 1 ]] +set +e +wait "$vstest_pid" +test_status=$? +printf '%s\n' "$test_status" > /task/test-exit-status.txt +for attempt in $(seq 1 400); do + kill -0 "$observer_pid" 2>/dev/null || break + sleep 0.1 +done +if kill -0 "$observer_pid" 2>/dev/null; then kill "$observer_pid"; fi +wait "$observer_pid" +observer_status=$? +printf '%s\n' "$observer_status" > /task/observer-exit-status.txt +# Preserve a cold-cap failure even when diagnostic readiness succeeds. +if [[ "$test_status" != 0 ]]; then exit "$test_status"; fi +exit "$observer_status" diff --git a/scripts/validation/opcua-lifecycle-memory/launch_saturation.py b/scripts/validation/opcua-lifecycle-memory/launch_saturation.py new file mode 100644 index 00000000..aa09b37b --- /dev/null +++ b/scripts/validation/opcua-lifecycle-memory/launch_saturation.py @@ -0,0 +1,114 @@ +#!/usr/bin/env python3 +"""Launch only a previously reviewed immutable candidate saturation bundle. Never an acceptance run.""" +import argparse +import hashlib +import json +from pathlib import Path +import subprocess +import time +import uuid +import xml.etree.ElementTree as ET + + +def digest(path): + return hashlib.sha256(path.read_bytes()).hexdigest() + + +def require(condition, message): + if not condition: + raise ValueError(message) + + +def verify(root, expected): + receipt_path = root / "receipt.json" + if digest(receipt_path) != expected: + raise ValueError("Receipt hash differs from the reviewed artifact") + receipt = json.loads(receipt_path.read_text()) + for name, value in receipt["files"].items(): + if digest(root / name) != value: + raise ValueError("Immutable bundle file changed: " + name) + require(receipt["protocol"] == "sdk2-saturation-v1" and receipt["acceptance"] is False, "Invalid diagnostic protocol") + require((receipt["canonicalCycles"], receipt["completedRestarts"], receipt["settlingCycles"], receipt["maximumPhaseSeconds"]) == (384, 96, 1, 1800), "Changed prospective limits") + return receipt + + +def trx(path, minimum): + document = ET.parse(path) + counters = document.find(".//{*}Counters") + if counters is None: + raise ValueError("Missing TRX counters") + executed = int(counters.attrib.get("executed", "0")) + skipped = max(int(counters.attrib.get("notExecuted", "0")), + sum(result.get("outcome") == "NotExecuted" for result in document.findall("./{*}Results/{*}UnitTestResult"))) + if executed < minimum or skipped: + raise ValueError("Required diagnostic tests were skipped or not executed") + return {"executed": executed, "skipped": skipped} + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--bundle", type=Path, required=True) + parser.add_argument("--receipt-sha256", required=True, help="Hash reviewed by root and Astra before launch") + parser.add_argument("--output", type=Path, required=True) + parser.add_argument("--verify-only", action="store_true") + options = parser.parse_args() + root = options.bundle.resolve(strict=True) + receipt = verify(root, options.receipt_sha256) + if options.verify_only: + print("Immutable bundle verified; no experiment launched") + return 0 + output = options.output.resolve() + output.mkdir(parents=True, exist_ok=False) + run_id = str(uuid.uuid4()) + container = "opcua-saturation-" + uuid.uuid4().hex[:12] + command = ["docker", "run", "--rm", "--name", container, "--network", "none", + "-e", "OTOPCUA_SOAK_BUILD_SHA=" + receipt["sourceRevision"], + "-e", "OTOPCUA_SDK2_SATURATION_RUN_ID=" + run_id, + "-v", str(root / "app") + ":/app:ro", "-v", str(root / "observer") + ":/observer:ro", + "-v", str(root / "tools") + ":/tools:ro", "-v", str(output) + ":/task", + receipt["image"], "bash", "/tools/container_saturation.sh"] + (output / "launch.json").write_text(json.dumps({"command": command, "receiptSha256": options.receipt_sha256, + "sourceRevision": receipt["sourceRevision"], "acceptance": False}, indent=2) + "\n") + start = time.monotonic() + try: + with (output / "container.log").open("w") as log: + process = subprocess.run(command, stdout=log, stderr=subprocess.STDOUT, timeout=2040) + status = process.returncode + except subprocess.TimeoutExpired: + status = 124 + finally: + subprocess.run(["docker", "stop", "--time", "10", container], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=20) + (output / "container-exit-status.txt").write_text(str(status) + "\n") + results = {"acceptance": False, "containerExitStatus": status, "wallSeconds": time.monotonic() - start, + "sourceRevision": receipt["sourceRevision"], "receiptSha256": options.receipt_sha256} + try: + results["descriptorTests"] = trx(output / "results/descriptor-controls.trx", 2) + results["experimentTests"] = trx(output / "results/saturation.trx", 1) + rows = [json.loads(line) for line in (output / "resources.jsonl").read_text().splitlines()] + header = rows[0] + require(header["mode"] == "saturation" and header["requestedSeconds"] == 1800 and header["build"] == receipt["sourceRevision"], "Wrong executed source or mode") + require(not any(row["kind"] == "complete" for row in rows), "Saturation must never emit an ordinary completion result") + outcome = [row for row in rows if row["kind"] == "saturation-result"] + require(len(outcome) == 1, "Missing unique diagnostic result") + outcome = outcome[0] + require((outcome["canonicalCycles"], outcome["settlingCycles"], outcome["restarts"]) == (384, 1, 96), "Incomplete fixed work") + require(outcome["phaseSeconds"] <= 1800 and outcome["shutdownOwnershipVerified"], "Cutoff or cleanup proof failed") + require(outcome["acceptance"] is False, "This is not an acceptance run") + results["diagnosticOutcome"] = outcome + observer_status = int((output / "observer-exit-status.txt").read_text()) + test_status = int((output / "test-exit-status.txt").read_text()) + require(observer_status == 0, "Observer failed") + require((test_status != 0) == (outcome["coldBudgetRejected"] or not outcome["readiness"]), "Cold-cap exit was not preserved") + results["protocolCompleted"] = True + except Exception as error: + results["protocolCompleted"] = False + results["validationErrorType"] = type(error).__name__ + status = status or 1 + verify(root, options.receipt_sha256) + (output / "run-receipt.json").write_text(json.dumps(results, indent=2) + "\n") + print(json.dumps(results, indent=2)) + return status + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/validation/opcua-lifecycle-memory/prepare_saturation.py b/scripts/validation/opcua-lifecycle-memory/prepare_saturation.py new file mode 100644 index 00000000..b9187ea7 --- /dev/null +++ b/scripts/validation/opcua-lifecycle-memory/prepare_saturation.py @@ -0,0 +1,63 @@ +#!/usr/bin/env python3 +"""Prepare a fresh immutable candidate bundle. This command never launches the experiment.""" +import argparse +import hashlib +import json +from pathlib import Path +import shutil +import subprocess +from build_saturation_observer import build + + +def sha(path): + return hashlib.sha256(path.read_bytes()).hexdigest() + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--output", type=Path, required=True) + parser.add_argument("--libraries", type=Path, required=True) + parser.add_argument("--image", required=True) + options = parser.parse_args() + source = Path(__file__).resolve().parent + repo = source.parents[2] + revision = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=repo, text=True).strip() + if subprocess.check_output(["git", "status", "--porcelain"], cwd=repo, text=True).strip(): + raise RuntimeError("Commit and review the exact source before preparing immutable artifacts") + if not options.image.startswith("sha256:"): + raise ValueError("Use an immutable Linux .NET SDK image digest") + root = options.output.resolve() + root.mkdir(parents=True, exist_ok=False) + project = repo / "tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests.csproj" + command = ["dotnet", "publish", str(project), "-c", "Release", "-r", "linux-arm64", "--self-contained", "true", + "-p:RuntimeFrameworkVersion=10.0.14", "-p:PublishSingleFile=false", "-p:PublishTrimmed=false", "-p:UseAppHost=true", + "--artifacts-path", str(root / "build"), "-o", str(root / "app"), "-m:2", "--nologo"] + with (root / "publish.log").open("w") as log: + subprocess.run(command, cwd=repo, stdout=log, stderr=subprocess.STDOUT, check=True) + observer = build(root / "observer-build", options.libraries, True) + shutil.copytree(observer, root / "observer") + tools = root / "tools" + tools.mkdir() + for name in ["launch_saturation.py", "container_saturation.sh", "SATURATION.md"]: + shutil.copy2(source / name, tools / name) + manifest = {str(path.relative_to(root)): sha(path) for directory in ["app", "observer", "tools"] + for path in sorted((root / directory).rglob("*")) if path.is_file()} + runtime = json.loads((root / "app/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests.runtimeconfig.json").read_text()) + included = runtime["runtimeOptions"]["includedFrameworks"] + if not included or any(value["version"] != "10.0.14" for value in included): + raise RuntimeError("Published app is not the prospectively fixed self-contained runtime") + receipt = {"protocol": "sdk2-saturation-v1", "sourceRevision": revision, "image": options.image, + "rid": "linux-arm64", "runtimeVersion": "10.0.14", "selfContained": True, + "canonicalCycles": 384, "completedRestarts": 96, "settlingCycles": 1, "maximumPhaseSeconds": 1800, + "windows": [[321, 352], [353, 384]], "observerBufferMiB": 16, "runtimeKeywords": "0x18", "rundown": False, + "acceptance": False, "publishCommand": command, + "testSourceSha256": sha(project.parent / "Sdk2LifecycleSoakTests.cs"), + "observerSourceSha256": sha(source / "SaturationObserver.cs"), "files": manifest} + receipt_path = root / "receipt.json" + receipt_path.write_text(json.dumps(receipt, indent=2) + "\n") + print(json.dumps({"bundle": str(root), "receiptSha256": sha(receipt_path), "sourceRevision": revision, + "launched": False}, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/scripts/validation/opcua-lifecycle-memory/saturation_tool_controls.py b/scripts/validation/opcua-lifecycle-memory/saturation_tool_controls.py new file mode 100644 index 00000000..d5aba053 --- /dev/null +++ b/scripts/validation/opcua-lifecycle-memory/saturation_tool_controls.py @@ -0,0 +1,37 @@ +#!/usr/bin/env python3 +"""Small negative controls for artifact tampering and VSTest's misleading skipped counter.""" +import hashlib +import json +from pathlib import Path +import tempfile +from launch_saturation import trx, verify + +with tempfile.TemporaryDirectory(prefix="opcua-saturation-tools-") as temporary: + root = Path(temporary) + artifact = root / "immutable" + artifact.write_text("control") + receipt = {"protocol": "sdk2-saturation-v1", "acceptance": False, "canonicalCycles": 384, + "completedRestarts": 96, "settlingCycles": 1, "maximumPhaseSeconds": 1800, + "files": {"immutable": hashlib.sha256(artifact.read_bytes()).hexdigest()}} + path = root / "receipt.json" + path.write_text(json.dumps(receipt)) + digest = hashlib.sha256(path.read_bytes()).hexdigest() + verify(root, digest) + artifact.write_text("tampered") + try: + verify(root, digest) + except ValueError: + pass + else: + raise RuntimeError("Changed artifact was accepted") + xml = root / "skipped.trx" + xml.write_text('' + '' + '') + try: + trx(xml, 1) + except ValueError: + pass + else: + raise RuntimeError("Direct skipped result was hidden by the counter") +print("Artifact verification, tamper rejection and direct skipped-result controls passed (3/3)") diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/Sdk2LifecycleSoakTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/Sdk2LifecycleSoakTests.cs index e5fa50b7..ee6e531f 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/Sdk2LifecycleSoakTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/Sdk2LifecycleSoakTests.cs @@ -1,6 +1,7 @@ using System.Collections; using System.Collections.Concurrent; using System.Diagnostics; +using System.Diagnostics.Tracing; using System.Globalization; using System.Net; using System.Net.Sockets; @@ -39,14 +40,16 @@ public sealed class Sdk2LifecycleSoakTests private static readonly DateTime Epoch = new(2026, 10, 1, 0, 0, 0, DateTimeKind.Utc); public static bool Enabled => Environment.GetEnvironmentVariable("OTOPCUA_SDK2_SOAK") == "1"; - [Fact(SkipUnless = nameof(Enabled), Skip = "Opt in with OTOPCUA_SDK2_SOAK=1 and an explicit diagnostic/full mode.")] + [Fact(SkipUnless = nameof(Enabled), Skip = "Opt in with OTOPCUA_SDK2_SOAK=1 and an explicit soak mode.")] public async Task Repeated_lifecycles_return_resources_to_a_steady_baseline() { var mode = Environment.GetEnvironmentVariable("OTOPCUA_SDK2_SOAK_MODE"); - (mode is "diagnostic" or "profile" or "full").ShouldBeTrue("explicit diagnostic, profile or full mode is required"); + (mode is "diagnostic" or "profile" or "full" or "saturation").ShouldBeTrue("an explicit soak mode is required"); + var saturation = mode == "saturation" ? SaturationControl.FromEnvironment() : null; var seconds = int.Parse(Environment.GetEnvironmentVariable("OTOPCUA_SDK2_SOAK_SECONDS") - ?? (mode == "full" ? "7200" : "90"), CultureInfo.InvariantCulture); + ?? (mode == "full" ? "7200" : saturation is not null ? "1800" : "90"), CultureInfo.InvariantCulture); if (mode == "full") seconds.ShouldBeGreaterThanOrEqualTo(7200); + else if (saturation is not null) seconds.ShouldBe(1800, "the prospective saturation cutoff is fixed"); else if (mode == "profile") seconds.ShouldBeInRange(60, 1800); // Investigative only; never satisfies the full-duration flag. else seconds.ShouldBeInRange(60, 120); var output = Environment.GetEnvironmentVariable("OTOPCUA_SDK2_SOAK_OUTPUT") @@ -55,7 +58,7 @@ public sealed class Sdk2LifecycleSoakTests Directory.CreateDirectory(Path.GetDirectoryName(output)!); await using var evidence = new StreamWriter(new FileStream(output, FileMode.CreateNew, FileAccess.Write, FileShare.Read)); using var run = CancellationTokenSource.CreateLinkedTokenSource(TestContext.Current.CancellationToken); - run.CancelAfter(TimeSpan.FromSeconds(seconds + 120)); + run.CancelAfter(TimeSpan.FromSeconds(saturation is null ? seconds + 120 : 120)); var ct = run.Token; var watch = Stopwatch.StartNew(); var clientRoot = Path.Combine(Path.GetTempPath(), $"otua-soak-client-{Guid.NewGuid():N}"); @@ -87,6 +90,7 @@ public sealed class Sdk2LifecycleSoakTests Configure(rig); // The required observation period excludes initial certificate creation and server startup. watch.Restart(); + if (saturation is not null) run.CancelAfter(TimeSpan.FromSeconds(1800)); await LogAsync(evidence, new { kind = "start", utc = DateTimeOffset.UtcNow, processId = Environment.ProcessId, @@ -97,8 +101,19 @@ public sealed class Sdk2LifecycleSoakTests limits = new { retainedManagedGrowthBytes = 64L << 20, rssGrowthBytes = 128L << 20, fdGrowth = 32 }, coverage = "encrypted sessions; scalar/Int16[] Raw+UNS; subscription removal; browse abandonment; history release/cross-session/close; surgical changes/rebuild; revocation; paused TCP read/recovery; same-PKI restart", }); - while (watch.Elapsed < TimeSpan.FromSeconds(seconds)) + while (saturation is not null ? cycles < 385 : watch.Elapsed < TimeSpan.FromSeconds(seconds)) { + if (saturation is not null) + { + if (cycles == 320) await saturation.AttachAsync(ct); + if (cycles == 320) saturation.Boundary("window1-start", 321, restarts); + if (cycles == 352) saturation.Boundary("window2-start", 353, restarts); + if (cycles == 384) + { + await saturation.DetachAndDrainAsync(ct); + await LogAsync(evidence, new { kind = "saturation-settling-start", cycle = 385, restarts }); + } + } var cycle = Stopwatch.StartNew(); await ExerciseAsync(rig, config, ++cycles, evidence, ct); await EventuallyAsync(() => PublicCounts(rig) == (0, 0, 0), ct); @@ -123,7 +138,7 @@ public sealed class Sdk2LifecycleSoakTests sample.ManagedBytes.ShouldBeLessThanOrEqualTo(baseline.ManagedBytes + (64L << 20)); sample.RssBytes.ShouldBeLessThanOrEqualTo(baseline.RssBytes + (128L << 20)); } - catch (ShouldAssertException error) when (mode == "profile") + catch (ShouldAssertException error) when (mode == "profile" || saturation is not null) { // Investigative sampling may continue after rejection, but can never turn it into a pass. firstRejectedBudget = ExceptionDispatchInfo.Capture(error); @@ -151,22 +166,47 @@ public sealed class Sdk2LifecycleSoakTests restarts++; await LogAsync(evidence, new { kind = "restart", cycles, restarts, seconds = restart.Elapsed.TotalSeconds }); } + if (saturation is not null) + { + if (cycles == 352) saturation.Boundary("window1-end", cycles, restarts); + if (cycles == 384) saturation.Boundary("window2-end", cycles, restarts); + } } // Preserve the first assertion and its original stack even if later RSS samples decrease. // This happens before any completion record; finally still releases the owned rig and client. - firstRejectedBudget?.Throw(); - cycles.ShouldBeGreaterThanOrEqualTo(4, "diagnostic mode must exercise at least one actual restart"); - restarts.ShouldBeGreaterThan(0); - watch.Elapsed.TotalSeconds.ShouldBeGreaterThanOrEqualTo(seconds); + if (saturation is null) firstRejectedBudget?.Throw(); + var phaseSeconds = watch.Elapsed.TotalSeconds; + if (saturation is not null) + { + cycles.ShouldBe(385); restarts.ShouldBe(96); + phaseSeconds.ShouldBeLessThanOrEqualTo(1800, "there is no saturation cutoff extension"); + } + else + { + cycles.ShouldBeGreaterThanOrEqualTo(4, "diagnostic mode must exercise at least one actual restart"); + restarts.ShouldBeGreaterThan(0); + watch.Elapsed.TotalSeconds.ShouldBeGreaterThanOrEqualTo(seconds); + } + var cleanupToken = saturation is null ? ct : TestContext.Current.CancellationToken; var finalClientCores = CertificateCores(config.CertificateManager); var finalServerCores = CertificateCores(rig.Server.CertificateManager!); - await rig.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(20), ct); + await rig.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(20), cleanupToken); rig = null; - await application.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(20), ct); + await application.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(20), cleanupToken); application = null; finalClientCores.Sum(CachedStores).ShouldBe(0); finalServerCores.Sum(CachedStores).ShouldBe(0); - (await DescriptorCountsAsync(ct, clientRoot, serverRoot)).Pki.ShouldBe(0); + (await DescriptorCountsAsync(cleanupToken, clientRoot, serverRoot)).Pki.ShouldBe(0); + if (saturation is not null) + { + await LogAsync(evidence, new { kind = "saturation-result", utc = DateTimeOffset.UtcNow, + canonicalCycles = 384, settlingCycles = 1, restarts, phaseSeconds, + readiness = saturation.Readiness, coldBudgetRejected = firstRejectedBudget is not null, + shutdownOwnershipVerified = true, acceptance = false }); + firstRejectedBudget?.Throw(); + saturation.Readiness.ShouldBeTrue("diagnostic readiness requires both fixed observation windows"); + return; + } await LogAsync(evidence, new { kind = "complete", utc = DateTimeOffset.UtcNow, mode, cycles, restarts, seconds = watch.Elapsed.TotalSeconds, satisfiesTwoHourDuration = mode == "full" && watch.Elapsed.TotalSeconds >= 7200 }); } @@ -174,6 +214,7 @@ public sealed class Sdk2LifecycleSoakTests { await LogAsync(evidence, new { kind = "failed", cycles, restarts, seconds = watch.Elapsed.TotalSeconds, exceptionType = error.GetType().FullName }); + if (saturation is not null) firstRejectedBudget?.Throw(); throw; } finally @@ -631,6 +672,221 @@ public sealed class Sdk2LifecycleSoakTests await output.WriteLineAsync(JsonSerializer.Serialize(value)); await output.FlushAsync(); } + + // Test-only IPC. Receipts are fresh, run/PID-bound and published atomically by a separate observer. + private sealed class SaturationControl(string directory, string runId) + { + private const string Protocol = "sdk2-saturation-v1"; + private int _observerPid; + private int _nextMarker; + internal bool Readiness { get; private set; } + internal static SaturationControl FromEnvironment() + { + var directory = Environment.GetEnvironmentVariable("OTOPCUA_SDK2_SATURATION_CONTROL") + ?? throw new InvalidOperationException("A fresh saturation control directory is required."); + var runId = Environment.GetEnvironmentVariable("OTOPCUA_SDK2_SATURATION_RUN_ID") + ?? throw new InvalidOperationException("An explicit saturation run id is required."); + if (!Path.IsPathFullyQualified(directory) || !Guid.TryParseExact(runId, "D", out _)) + throw new InvalidOperationException("Saturation control identity is malformed."); + Directory.CreateDirectory(directory); + if (Directory.EnumerateFileSystemEntries(directory).Any()) + throw new InvalidOperationException("Saturation control directory must be empty."); + _ = SaturationMarkers.Log; + return new SaturationControl(directory, runId); + } + internal async Task AttachAsync(CancellationToken ct) + { + await PublishAsync("attach-request.json", new { protocol = Protocol, runId, + targetPid = Environment.ProcessId, firstCycle = 321, completedRestarts = 80 }, ct); + using var receipt = await WaitAsync("active.json", ct); + var root = receipt.RootElement; + ValidateIdentity(root); + root.GetProperty("active").GetBoolean().ShouldBeTrue(); + _observerPid = root.GetProperty("observerPid").GetInt32(); + if (_observerPid <= 0 || _observerPid == Environment.ProcessId) + throw new InvalidDataException("A distinct observer process is required."); + using var enabled = CancellationTokenSource.CreateLinkedTokenSource(ct); + enabled.CancelAfter(TimeSpan.FromSeconds(5)); + while (!SaturationMarkers.Log.IsEnabled()) await Task.Delay(10, enabled.Token); + } + internal void Boundary(string name, int cycle, int restarts) + { + (string Name, int Cycle, int Restarts)[] expected = + [ ("window1-start", 321, 80), ("window1-end", 352, 88), + ("window2-start", 353, 88), ("window2-end", 384, 96) ]; + if (_observerPid == 0 || _nextMarker >= expected.Length || expected[_nextMarker] != (name, cycle, restarts)) + throw new InvalidOperationException("Saturation boundary is out of order."); + if (!SaturationMarkers.Log.IsEnabled()) throw new InvalidOperationException("Saturation observer is not active."); + SaturationMarkers.Log.Boundary(runId, name, cycle, restarts); + _nextMarker++; + } + internal async Task DetachAndDrainAsync(CancellationToken ct) + { + if (_nextMarker != 4) throw new InvalidOperationException("Both fixed observation windows are required."); + await PublishAsync("detach-request.json", new { protocol = Protocol, runId, + targetPid = Environment.ProcessId, lastCycle = 384, completedRestarts = 96 }, ct); + using var receipt = await WaitAsync("drained.json", ct); + Readiness = ValidateDrained(receipt.RootElement, runId, Environment.ProcessId, _observerPid); + using var disabled = CancellationTokenSource.CreateLinkedTokenSource(ct); + disabled.CancelAfter(TimeSpan.FromSeconds(5)); + while (SaturationMarkers.Log.IsEnabled()) await Task.Delay(10, disabled.Token); + } + internal static bool ValidateDrained(JsonElement root, string expectedRun, int target, int observer) + { + if (root.GetProperty("protocol").GetString() != Protocol || root.GetProperty("runId").GetString() != expectedRun + || root.GetProperty("targetPid").GetInt32() != target || root.GetProperty("observerPid").GetInt32() != observer + || !root.GetProperty("drained").GetBoolean() || root.GetProperty("active").GetBoolean()) + throw new InvalidDataException("Saturation drained receipt identity or ownership is invalid."); + var windows = root.GetProperty("windows").EnumerateArray().ToArray(); + if (windows.Length != 2) throw new InvalidDataException("Exactly two fixed windows are required."); + var ready = root.GetProperty("markersValid").GetBoolean(); + foreach (var name in new[] { "eventsLost", "sdkFactoryJit", "sdkFactoryLoads", "sdkAssemblyLoads" }) + { + var value = root.GetProperty(name).GetInt64(); + if (value < 0) throw new InvalidDataException("Negative diagnostic count."); + ready &= value == 0; + } + for (var index = 0; index < 2; index++) + { + var window = windows[index]; + if (window.GetProperty("name").GetString() != $"window{index + 1}" + || window.GetProperty("firstCycle").GetInt32() != 321 + index * 32 + || window.GetProperty("lastCycle").GetInt32() != 352 + index * 32 + || window.GetProperty("startRestarts").GetInt32() != 80 + index * 8 + || window.GetProperty("endRestarts").GetInt32() != 88 + index * 8) + throw new InvalidDataException("Saturation window boundaries are invalid."); + foreach (var name in new[] { "sdkFactoryJit", "sdkFactoryLoads", "sdkAssemblyLoads" }) + { + var value = window.GetProperty(name).GetInt64(); + if (value < 0) throw new InvalidDataException("Negative window diagnostic count."); + ready &= value == 0; + } + } + return ready; + } + private void ValidateIdentity(JsonElement root) + { + if (root.GetProperty("protocol").GetString() != Protocol || root.GetProperty("runId").GetString() != runId + || root.GetProperty("targetPid").GetInt32() != Environment.ProcessId) + throw new InvalidDataException("Saturation observer receipt identity is invalid."); + } + private async Task PublishAsync(string name, object value, CancellationToken ct) + { + var temporary = Path.Combine(directory, name + ".tmp"); + await using (var stream = new FileStream(temporary, FileMode.CreateNew, FileAccess.Write)) + await JsonSerializer.SerializeAsync(stream, value, cancellationToken: ct); + File.Move(temporary, Path.Combine(directory, name)); + } + private async Task WaitAsync(string name, CancellationToken ct) + { + using var bounded = CancellationTokenSource.CreateLinkedTokenSource(ct); + bounded.CancelAfter(TimeSpan.FromSeconds(45)); + var path = Path.Combine(directory, name); + while (!File.Exists(path)) + { + if (File.Exists(Path.Combine(directory, "observer-failed.json"))) + throw new InvalidOperationException("The saturation observer failed before completing its receipt."); + await Task.Delay(25, bounded.Token); + } + return JsonDocument.Parse(await File.ReadAllTextAsync(path, bounded.Token)); + } + } + + [Fact] + public void Saturation_protocol_rejects_stale_identity_and_nonzero_readiness_counts() + { + var run = Guid.NewGuid().ToString("D"); + JsonDocument Receipt(string identity, int pid = 100, int lost = 0, int jit = 0, int loads = 0, int assemblies = 0) => + JsonDocument.Parse(JsonSerializer.Serialize(new { protocol = "sdk2-saturation-v1", runId = identity, + targetPid = pid, observerPid = 200, active = false, drained = true, markersValid = true, + eventsLost = lost, sdkFactoryJit = jit, sdkFactoryLoads = loads, sdkAssemblyLoads = assemblies, + windows = new[] { + new { name = "window1", firstCycle = 321, lastCycle = 352, startRestarts = 80, endRestarts = 88, + sdkFactoryJit = jit, sdkFactoryLoads = loads, sdkAssemblyLoads = assemblies }, + new { name = "window2", firstCycle = 353, lastCycle = 384, startRestarts = 88, endRestarts = 96, + sdkFactoryJit = 0, sdkFactoryLoads = 0, sdkAssemblyLoads = 0 } } })); + using var good = Receipt(run); + SaturationControl.ValidateDrained(good.RootElement, run, 100, 200).ShouldBeTrue(); + using var stale = Receipt(Guid.NewGuid().ToString("D")); + Should.Throw(() => SaturationControl.ValidateDrained(stale.RootElement, run, 100, 200)); + using var wrongPid = Receipt(run, pid: 101); + Should.Throw(() => SaturationControl.ValidateDrained(wrongPid.RootElement, run, 100, 200)); + foreach (var denied in new[] { Receipt(run, lost: 1), Receipt(run, jit: 1), Receipt(run, loads: 1), Receipt(run, assemblies: 1) }) + { + using (denied) SaturationControl.ValidateDrained(denied.RootElement, run, 100, 200).ShouldBeFalse(); + } + } + + [Fact] + public async Task Saturation_protocol_missing_observer_is_bounded() + { + var directory = Path.Combine(Path.GetTempPath(), $"otua-saturation-control-{Guid.NewGuid():N}"); + Directory.CreateDirectory(directory); + try + { + using var deadline = CancellationTokenSource.CreateLinkedTokenSource(TestContext.Current.CancellationToken); + deadline.CancelAfter(TimeSpan.FromMilliseconds(100)); + var control = new SaturationControl(directory, Guid.NewGuid().ToString("D")); + await Should.ThrowAsync(() => control.AttachAsync(deadline.Token)); + File.Exists(Path.Combine(directory, "active.json")).ShouldBeFalse(); + } + finally { Directory.Delete(directory, recursive: true); } + } + + public static bool SaturationObserverAvailable => File.Exists(Environment.GetEnvironmentVariable("OTOPCUA_SDK2_SATURATION_OBSERVER")); + + [Theory(SkipUnless = nameof(SaturationObserverAvailable), Skip = "Requires the separately built diagnostic observer.")] + [InlineData(false)] + [InlineData(true)] + public async Task Saturation_protocol_actual_observer_drains_and_rejects_extra_foreign_markers(bool foreignMarker) + { + var directory = Path.Combine(Path.GetTempPath(), $"otua-saturation-observer-{Guid.NewGuid():N}"); + Directory.CreateDirectory(directory); + var run = Guid.NewGuid().ToString("D"); + var control = new SaturationControl(directory, run); + var observerPath = Environment.GetEnvironmentVariable("OTOPCUA_SDK2_SATURATION_OBSERVER")!; + var frameworkDependent = observerPath.EndsWith(".dll", StringComparison.OrdinalIgnoreCase); + using var observer = new Process { StartInfo = new ProcessStartInfo(frameworkDependent ? "dotnet" : observerPath) + { RedirectStandardOutput = true, RedirectStandardError = true, UseShellExecute = false } }; + if (frameworkDependent) observer.StartInfo.ArgumentList.Add(observerPath); + observer.StartInfo.ArgumentList.Add(directory); + observer.StartInfo.ArgumentList.Add(run); + using var deadline = CancellationTokenSource.CreateLinkedTokenSource(TestContext.Current.CancellationToken); + deadline.CancelAfter(TimeSpan.FromSeconds(40)); + try + { + observer.Start().ShouldBeTrue(); + var stdout = observer.StandardOutput.ReadToEndAsync(deadline.Token); + var stderr = observer.StandardError.ReadToEndAsync(deadline.Token); + await control.AttachAsync(deadline.Token); + if (foreignMarker) SaturationMarkers.Log.Boundary(Guid.NewGuid().ToString("D"), "window1-start", 321, 80); + control.Boundary("window1-start", 321, 80); + await Task.Delay(20, deadline.Token); + control.Boundary("window1-end", 352, 88); + control.Boundary("window2-start", 353, 88); + await Task.Delay(20, deadline.Token); + control.Boundary("window2-end", 384, 96); + await control.DetachAndDrainAsync(deadline.Token); + control.Readiness.ShouldBe(!foreignMarker); + SaturationMarkers.Log.IsEnabled().ShouldBeFalse("Stop and stream drain precede return to the target"); + await observer.WaitForExitAsync(deadline.Token); + observer.ExitCode.ShouldBe(0, await stderr); + await stdout; + } + finally + { + if (!observer.HasExited) { observer.Kill(entireProcessTree: true); await observer.WaitForExitAsync(TestContext.Current.CancellationToken); } + Directory.Delete(directory, recursive: true); + } + } + + [EventSource(Name = "OtOpcUa-Sdk2-Saturation")] + private sealed class SaturationMarkers : EventSource + { + internal static readonly SaturationMarkers Log = new(); + [Event(1, Level = EventLevel.Informational)] + public void Boundary(string runId, string name, int cycle, int restarts) => WriteEvent(1, runId, name, cycle, restarts); + } private sealed class AcceptWrites : IOpcUaNodeWriteGateway { public Task WriteAsync(string id, object? value, AddressSpaceRealm realm, CancellationToken ct) =>