Joseph Doherty
215 lines
14 KiB
C#
215 lines
14 KiB
C#
using System.Collections.Concurrent;
|
|
using System.Diagnostics;
|
|
using System.Net;
|
|
using System.Net.Sockets;
|
|
using System.Reflection;
|
|
using System.Text.Json;
|
|
using Opc.Ua;
|
|
using Opc.Ua.Configuration;
|
|
using Opc.Ua.Server;
|
|
using Shouldly;
|
|
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
|
|
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
|
using CoreHistoryReadResult = ZB.MOM.WW.OtOpcUa.Core.Abstractions.HistoryReadResult;
|
|
return args.Length == 3 && (args[2] == "--semantic-retention-control" || args[2] == "--semantic-terminal-dispose-control")
|
|
? await Canonical.SemanticRetentionAsync(args) : await Canonical.RunAsync(args);
|
|
internal static partial class Canonical
|
|
{
|
|
private const string Scalar = BaselineRig.RawDeviceFolder + "/SoakScalar";
|
|
private const string ArrayTag = BaselineRig.RawDeviceFolder + "/SoakArray";
|
|
private const string UnsScalar = "SoakEquipment/Scalar";
|
|
private static readonly DateTime Epoch = new(2026,10,1,0,0,0,DateTimeKind.Utc);
|
|
internal static readonly OperationLedger Ledger = new();
|
|
private static StreamWriter CurrentEvidence = null!;
|
|
internal static async Task<int> RunAsync(string[] args)
|
|
{
|
|
var root=Path.GetFullPath(args[0]); var cycles=int.Parse(args[1]);
|
|
var isolatedRestart = args.Length == 3 && args[2] == "--active-restart-control";
|
|
if (args.Length > 3 || (args.Length == 3 && !isolatedRestart)) throw new InvalidOperationException("Unknown short-control mode");
|
|
if(cycles<1 || cycles>4) throw new InvalidOperationException("Only a 1..4-cycle short diagnostic smoke is authorized");
|
|
if(Directory.Exists(root)) throw new InvalidOperationException("Use a fresh private artifact directory");
|
|
Directory.CreateDirectory(root);
|
|
if(!OperatingSystem.IsWindows()) File.SetUnixFileMode(root,UnixFileMode.UserRead|UnixFileMode.UserWrite|UnixFileMode.UserExecute);
|
|
await using var output=new StreamWriter(new FileStream(Path.Combine(root,"resources.jsonl"),FileMode.CreateNew,FileAccess.Write,FileShare.Read));
|
|
CurrentEvidence=output;
|
|
using var deadline=new CancellationTokenSource(TimeSpan.FromSeconds(120)); var ct=deadline.Token;
|
|
var sdk=typeof(NodeId).Assembly.GetCustomAttribute<AssemblyInformationalVersionAttribute>()?.InformationalVersion;
|
|
if(sdk is null || !sdk.StartsWith("1.5.378.182+",StringComparison.Ordinal)) throw new InvalidOperationException("The immutable .182 SDK is required");
|
|
var clientRoot=Path.Combine(root,"private-fixture","client-pki");
|
|
var telemetry=DefaultTelemetry.Create(_=>{});
|
|
var config=new ApplicationConfiguration(telemetry)
|
|
{
|
|
ApplicationName="SDK2LifecycleSoak",ApplicationUri=$"urn:SDK2LifecycleSoak:{Guid.NewGuid():N}",ApplicationType=ApplicationType.Client,
|
|
SecurityConfiguration=new SecurityConfiguration
|
|
{
|
|
ApplicationCertificate=new CertificateIdentifier{StoreType=CertificateStoreType.Directory,StorePath=Path.Combine(clientRoot,"own"),SubjectName="CN=SDK2LifecycleSoak"},
|
|
TrustedIssuerCertificates=new CertificateTrustList{StoreType=CertificateStoreType.Directory,StorePath=Path.Combine(clientRoot,"issuer")},
|
|
TrustedPeerCertificates=new CertificateTrustList{StoreType=CertificateStoreType.Directory,StorePath=Path.Combine(clientRoot,"trusted")},
|
|
RejectedCertificateStore=new CertificateTrustList{StoreType=CertificateStoreType.Directory,StorePath=Path.Combine(clientRoot,"rejected")},AutoAcceptUntrustedCertificates=true
|
|
},ClientConfiguration=new ClientConfiguration{DefaultSessionTimeout=15000},TransportQuotas=new TransportQuotas{OperationTimeout=5000}
|
|
};
|
|
var application=new ApplicationInstance(telemetry){ApplicationConfiguration=config};
|
|
await config.ValidateAsync(ApplicationType.Client,ct);await application.CheckApplicationInstanceCertificatesAsync(true,null,ct);
|
|
await using var rig=new BaselineRig(Path.Combine(root,"private-fixture"));
|
|
await rig.BootAsync(ct);Configure(rig);
|
|
var watch=Stopwatch.StartNew();var completed=0;var restarts=0;
|
|
await LogAsync(output,new {kind="start",sdk,productRebuilt=false,sourcePublication="83ed880e",requestedCycles=cycles,
|
|
processId=Environment.ProcessId,acceptance=false,scope="canonical operations short smoke; ownership differences remain explicit",
|
|
unsupportedObservables=new[]{"SDK2 CertificateManager disposed flag","SDK2 manager-owned cache-clear proof","ApplicationInstance.DisposeAsync"}});
|
|
try
|
|
{
|
|
if (isolatedRestart)
|
|
{
|
|
if (cycles != 1) throw new InvalidOperationException("The isolated restart control performs exactly one restart");
|
|
await RestartWithActiveResourcesAsync(rig,config,ct);Configure(rig);restarts++;
|
|
await LogAsync(output,new {kind="isolated-active-restart-completed",normalCycles=0,restarts,
|
|
acceptance=false,ownershipParity=false,ledger=Ledger.Snapshot()});
|
|
return 2;
|
|
}
|
|
while(completed<cycles)
|
|
{
|
|
await ExerciseAsync(rig,config,completed+1,output,ct);completed++;
|
|
await EventuallyAsync(()=>PublicCounts(rig)==(0,0,0),ct);
|
|
var counts=PublicCounts(rig);var fd=await BaselineMetrics.DescriptorsAsync(ct,clientRoot,rig.PkiRoot);
|
|
fd.Pki.ShouldBe(0);
|
|
foreach(var name in new[]{"LiveItemCount","LiveEventItemCount","MonitoredNodeCount"}) BaselineMetrics.Counter(rig.Nm,name).ShouldBe(0);
|
|
using var process=Process.GetCurrentProcess();process.Refresh();
|
|
await LogAsync(output,new {kind="quiescent",cycle=completed,seconds=watch.Elapsed.TotalSeconds,
|
|
sessions=counts.Sessions,subscriptions=counts.Subscriptions,items=counts.Items,
|
|
pkiDescriptors=fd.Pki,descriptors=fd.Total,managedBytes=GC.GetTotalMemory(true),rssBytes=process.WorkingSet64,
|
|
serverCachedStoreReferences=BaselineMetrics.CachedStoreReferences(rig.Validator),
|
|
clientCachedStoreReferences=BaselineMetrics.CachedStoreReferences(config.CertificateValidator),jit=System.Runtime.JitInfo.GetCompiledMethodCount(),timers=Timer.ActiveCount});
|
|
if(completed%4==0){await RestartWithActiveResourcesAsync(rig,config,ct);Configure(rig);restarts++;}
|
|
}
|
|
await LogAsync(output,new {kind="operations-completed",cycles=completed,restarts,seconds=watch.Elapsed.TotalSeconds,
|
|
acceptance=false,ownershipParity=false,reason="Original SDK has no equivalent manager-disposal observable; review required",ledger=Ledger.Snapshot()});
|
|
return 2;
|
|
}
|
|
catch(Exception error)
|
|
{
|
|
await LogAsync(output,new {kind="failed",cycles=completed,restarts,exceptionType=error.GetType().FullName,
|
|
uaStatus=error is ServiceResultException ua ? ua.StatusCode.ToString() : null,ledger=Ledger.Snapshot(),acceptance=false});
|
|
Console.Error.WriteLine(error);return 1;
|
|
}
|
|
}
|
|
private static int ScalarValue(Variant value)=>value.Value is int result?result:throw new InvalidOperationException("Expected scalar Int32 CLR shape");
|
|
private static void RequirePoint(byte[]? point)=>(point is {Length:>0}).ShouldBeTrue();
|
|
private static (int Browse,int History) ContinuationCounts(ISession session)=>BaselineMetrics.Points(session);
|
|
private static (int Sessions,int Subscriptions,int Items) PublicCounts(BaselineRig rig)
|
|
{var subscriptions=rig.Server.CurrentInstance.SubscriptionManager.GetSubscriptions();return(rig.Server.CurrentInstance.SessionManager.GetSessions().Count,subscriptions.Count,subscriptions.Sum(s=>s.MonitoredItemCount));}
|
|
private static async Task EventuallyAsync(Func<bool> condition,CancellationToken ct)
|
|
{var elapsed=Stopwatch.StartNew();while(!condition()&&elapsed.Elapsed<TimeSpan.FromSeconds(10))await Task.Delay(25,ct);condition().ShouldBeTrue();}
|
|
private static async Task LogAsync(StreamWriter output,object value){await output.WriteLineAsync(JsonSerializer.Serialize(value));await output.FlushAsync();}
|
|
private sealed class AcceptWrites:IOpcUaNodeWriteGateway
|
|
{public Task<NodeWriteOutcome> WriteAsync(string id,object? value,AddressSpaceRealm realm,CancellationToken ct)=>Task.FromResult(new NodeWriteOutcome(true,null));}
|
|
internal sealed class SqlLikeEventSource(List<HistoricalEvent> events) : IHistorianDataSource
|
|
{
|
|
private int _eventReads;
|
|
internal int EventReads => Volatile.Read(ref _eventReads);
|
|
private readonly List<HistoricalEvent> _all =
|
|
events.OrderBy(e => e.EventTimeUtc).ThenBy(e => e.EventId, StringComparer.Ordinal).ToList();
|
|
|
|
public async Task<HistoricalEventsResult> ReadEventsAsync(
|
|
string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, CancellationToken ct)
|
|
{
|
|
Interlocked.Increment(ref _eventReads);
|
|
await Task.Delay(1, ct).ConfigureAwait(false); // complete on a pool thread, like a real backend
|
|
var window = _all.Where(e => e.EventTimeUtc >= startUtc && e.EventTimeUtc < endUtc).ToList();
|
|
var cap = maxEvents <= 0 ? 1000 : maxEvents;
|
|
return new HistoricalEventsResult(window.Take(cap).ToList(), window.Count > cap ? new byte[] { 1 } : null);
|
|
}
|
|
|
|
public bool SupportsReverseEvents => true;
|
|
|
|
/// <summary>The #675 overload: reverse is <c>(lo, hi]</c> by (time, id) descending.</summary>
|
|
public async Task<HistoricalEventsResult> ReadEventsAsync(
|
|
string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, bool reverse, CancellationToken ct)
|
|
{
|
|
if (!reverse) return await ReadEventsAsync(sourceName, startUtc, endUtc, maxEvents, ct).ConfigureAwait(false);
|
|
await Task.Delay(1, ct).ConfigureAwait(false);
|
|
var (lo, hi) = startUtc <= endUtc ? (startUtc, endUtc) : (endUtc, startUtc);
|
|
var window = _all.Where(e => e.EventTimeUtc > lo && e.EventTimeUtc <= hi).Reverse().ToList();
|
|
var cap = maxEvents <= 0 ? 1000 : maxEvents;
|
|
return new HistoricalEventsResult(window.Take(cap).ToList(), window.Count > cap ? new byte[] { 1 } : null);
|
|
}
|
|
|
|
public Task<CoreHistoryReadResult> ReadRawAsync(string r, DateTime s, DateTime e, uint m, CancellationToken ct) =>
|
|
NullHistorianDataSource.Instance.ReadRawAsync(r, s, e, m, ct);
|
|
public Task<CoreHistoryReadResult> ReadProcessedAsync(
|
|
string r, DateTime s, DateTime e, TimeSpan i, HistoryAggregateType a, CancellationToken ct) =>
|
|
NullHistorianDataSource.Instance.ReadProcessedAsync(r, s, e, i, a, ct);
|
|
public Task<CoreHistoryReadResult> ReadAtTimeAsync(string r, IReadOnlyList<DateTime> t, CancellationToken ct) =>
|
|
NullHistorianDataSource.Instance.ReadAtTimeAsync(r, t, ct);
|
|
public HistorianHealthSnapshot GetHealthSnapshot() => NullHistorianDataSource.Instance.GetHealthSnapshot();
|
|
public void Dispose() { }
|
|
}
|
|
/// <summary>Task-owned TCP relay. Pause blocks forwarding in both directions without closing either socket.</summary>
|
|
private sealed class PauseProxy : IAsyncDisposable
|
|
{
|
|
private readonly TcpListener _listener = new(IPAddress.Loopback, 0);
|
|
private readonly CancellationTokenSource _stop = new();
|
|
private readonly ConcurrentBag<TcpClient> _sockets = [];
|
|
private readonly ConcurrentBag<Task> _pumps = [];
|
|
private readonly Task _accept;
|
|
private TaskCompletionSource? _pause;
|
|
internal int Port => ((IPEndPoint)_listener.LocalEndpoint).Port;
|
|
internal PauseProxy(int targetPort)
|
|
{
|
|
_listener.Start();
|
|
_accept = AcceptAsync(targetPort);
|
|
}
|
|
internal void Pause() => Volatile.Write(ref _pause, new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously));
|
|
internal void Resume() => Interlocked.Exchange(ref _pause, null)?.TrySetResult();
|
|
private async Task AcceptAsync(int port)
|
|
{
|
|
try
|
|
{
|
|
while (!_stop.IsCancellationRequested)
|
|
{
|
|
var incoming = await _listener.AcceptTcpClientAsync(_stop.Token);
|
|
_sockets.Add(incoming);
|
|
var outgoing = new TcpClient(); _sockets.Add(outgoing);
|
|
await outgoing.ConnectAsync(IPAddress.Loopback, port, _stop.Token);
|
|
_pumps.Add(PumpAsync(incoming.GetStream(), outgoing.GetStream()));
|
|
_pumps.Add(PumpAsync(outgoing.GetStream(), incoming.GetStream()));
|
|
}
|
|
}
|
|
catch (OperationCanceledException) when (_stop.IsCancellationRequested) { }
|
|
}
|
|
private async Task PumpAsync(NetworkStream source, NetworkStream destination)
|
|
{
|
|
var buffer = new byte[16 * 1024];
|
|
try
|
|
{
|
|
while (true)
|
|
{
|
|
var length = await source.ReadAsync(buffer, _stop.Token);
|
|
if (length == 0) return;
|
|
if (Volatile.Read(ref _pause) is { } paused) await paused.Task.WaitAsync(_stop.Token);
|
|
await destination.WriteAsync(buffer.AsMemory(0, length), _stop.Token);
|
|
}
|
|
}
|
|
catch (OperationCanceledException) when (_stop.IsCancellationRequested) { }
|
|
catch (IOException) when (_stop.IsCancellationRequested) { }
|
|
}
|
|
public async ValueTask DisposeAsync()
|
|
{
|
|
Resume(); _stop.Cancel(); _listener.Stop();
|
|
foreach (var socket in _sockets) socket.Dispose();
|
|
try { await _accept; await Task.WhenAll(_pumps); }
|
|
finally { _stop.Dispose(); }
|
|
}
|
|
}
|
|
}
|
|
|
|
internal sealed class OperationLedger
|
|
{
|
|
private readonly ConcurrentDictionary<string,long> _counts=new();
|
|
internal void Record(string name)=>_counts.AddOrUpdate(name,1,(_,count)=>count+1);
|
|
internal async Task<T> Call<T>(string name,Func<Task<T>> operation)
|
|
{Record(name+".attempted");try{var value=await operation();Record(name+".returned");return value;}catch{Record(name+".faulted");throw;}}
|
|
internal async Task Call(string name,Func<Task> operation)
|
|
{Record(name+".attempted");try{await operation();Record(name+".returned");}catch{Record(name+".faulted");throw;}}
|
|
internal IReadOnlyDictionary<string,long> Snapshot()=>new SortedDictionary<string,long>(_counts);
|
|
}
|