Joseph Doherty
8582ce91bc
v2-ci / unit-tests (push) Has been cancelled
#789 (partial, further tuning deferred): combining inline frame writer, buffered frame reader, one timer per evaluation for T and T+G, slot release on the reader before the caller resumes, a non-spinning per-connection result dispatcher, and blocking pipe I/O on Unix (both ends) so the socket never enters the socket engine; TerminateJobObject kill on Windows. macOS host CPU at 10k evals/s K=2: 228-240 % -> 79 % (InProcess 69 %, IPC share +10 vs budget 25); windev 153 % -> 123 % (+115, still a miss). Harness reports the "vs InProcess" share, warm-cache worker RSS, and times kill latency from the soft-timeout answer. #790: the worker sends LoadStarted; the compile timeout is armed then, so a load queued behind other compiles is never killed or struck. #788: one Error with stack per down period, rate-limited Warnings with attempt count, one Information on recovery. Design doc §8.2 notes + §12.2 record. Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
350 lines
16 KiB
C#
350 lines
16 KiB
C#
using System.Buffers.Binary;
|
|
using Google.Protobuf;
|
|
using Shouldly;
|
|
using Xunit;
|
|
using ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess;
|
|
using ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol;
|
|
using ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol.V1;
|
|
|
|
namespace ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests;
|
|
|
|
/// <summary>lmxopcua#780 design §13 — frame codec and the CLR ↔ Variant mapping.</summary>
|
|
public sealed class ProtocolTests
|
|
{
|
|
public static IEnumerable<object?[]> Values()
|
|
{
|
|
object?[] values =
|
|
[
|
|
null, true, false, (sbyte)-5, (byte)250, (short)-3, (ushort)7, -42, 42u, long.MinValue, ulong.MaxValue,
|
|
1.5f, double.NaN, 3.25, "text", string.Empty, new DateTime(2026, 11, 9, 1, 2, 3, DateTimeKind.Utc),
|
|
new[] { 1, 2, 3 }, new short[] { -1, 1 }, new[] { "a", "b" }, new[] { true, false }, Array.Empty<double>(),
|
|
new[] { new DateTime(2020, 1, 1, 0, 0, 0, DateTimeKind.Utc) }, (short)0, 0, 0L, 0.0,
|
|
];
|
|
return values.Select(v => new[] { v });
|
|
}
|
|
|
|
/// <summary>The exact boxed CLR type survives (design §4.4): <c>(short)ctx.GetTag(x).Value</c> must keep working.</summary>
|
|
[Theory]
|
|
[MemberData(nameof(Values))]
|
|
public void Every_type_tag_round_trips_with_its_exact_clr_type(object? value)
|
|
{
|
|
VariantCodec.TryEncode(value, out var variant, out var unsupported).ShouldBeTrue();
|
|
unsupported.ShouldBeNull();
|
|
|
|
// Through the wire, not just the object: a payload the serializer drops must still decode.
|
|
var wire = Variant.Parser.ParseFrom(variant.ToByteArray());
|
|
var decoded = VariantCodec.Decode(wire);
|
|
|
|
if (value is null)
|
|
{
|
|
decoded.ShouldBeNull();
|
|
return;
|
|
}
|
|
decoded.ShouldNotBeNull();
|
|
decoded.GetType().ShouldBe(value.GetType());
|
|
if (value is Array expected)
|
|
((Array)decoded).Cast<object?>().ShouldBe(expected.Cast<object?>());
|
|
else if (value is double d && double.IsNaN(d))
|
|
double.IsNaN((double)decoded).ShouldBeTrue();
|
|
else
|
|
decoded.ShouldBe(value);
|
|
}
|
|
|
|
[Fact]
|
|
public void Unsupported_types_are_refused_not_converted()
|
|
{
|
|
foreach (var value in new object[] { new List<int> { 1 }, 1.5m, Guid.NewGuid(), new { A = 1 }, new int[1, 1], new object[] { 1 } })
|
|
{
|
|
VariantCodec.TryEncode(value, out _, out var type).ShouldBeFalse(value.GetType().Name);
|
|
type.ShouldNotBeNullOrEmpty();
|
|
}
|
|
}
|
|
|
|
[Fact]
|
|
public void Local_datetime_crosses_as_utc()
|
|
{
|
|
var local = new DateTime(2026, 6, 1, 12, 0, 0, DateTimeKind.Local);
|
|
VariantCodec.TryEncode(local, out var variant, out _).ShouldBeTrue();
|
|
var decoded = (DateTime)VariantCodec.Decode(variant)!;
|
|
decoded.Kind.ShouldBe(DateTimeKind.Utc);
|
|
decoded.ShouldBe(local.ToUniversalTime());
|
|
}
|
|
|
|
[Fact]
|
|
public void A_payload_that_contradicts_its_tag_is_a_protocol_fault()
|
|
{
|
|
Should.Throw<ScriptWorkerProtocolException>(() => VariantCodec.Decode(new Variant { Type = Variant.Types.TypeTag.Int16, S = "x" }));
|
|
Should.Throw<ScriptWorkerProtocolException>(() => VariantCodec.Decode(new Variant { Type = (Variant.Types.TypeTag)99 }));
|
|
Should.Throw<OverflowException>(() => VariantCodec.Decode(new Variant { Type = Variant.Types.TypeTag.Int16, I = 1 << 20 }));
|
|
}
|
|
|
|
private static Envelope Sample() => new()
|
|
{
|
|
ProtocolVersion = ScriptWorkerFrameCodec.ProtocolVersion,
|
|
Seq = 7,
|
|
Heartbeat = new Heartbeat { Inflight = 3, HeapBytes = 1234 },
|
|
};
|
|
|
|
[Fact]
|
|
public async Task A_frame_round_trips()
|
|
{
|
|
using var stream = new MemoryStream();
|
|
await ScriptWorkerFrameCodec.WriteAsync(stream, Sample(), 1024, TestContext.Current.CancellationToken);
|
|
stream.Position = 0;
|
|
var read = await ScriptWorkerFrameCodec.ReadAsync(stream, 1024, TestContext.Current.CancellationToken);
|
|
read.ShouldBe(Sample());
|
|
(await ScriptWorkerFrameCodec.ReadAsync(stream, 1024, TestContext.Current.CancellationToken)).ShouldBeNull("a clean end of stream at a frame boundary");
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Oversize_frames_are_refused_on_write_and_on_read()
|
|
{
|
|
using var stream = new MemoryStream();
|
|
await Should.ThrowAsync<ScriptWorkerProtocolException>(async () =>
|
|
await ScriptWorkerFrameCodec.WriteAsync(stream, Sample(), 4, TestContext.Current.CancellationToken));
|
|
stream.Length.ShouldBe(0, "nothing is written for a refused frame");
|
|
|
|
await ScriptWorkerFrameCodec.WriteAsync(stream, Sample(), 1024, TestContext.Current.CancellationToken);
|
|
stream.Position = 0;
|
|
await Should.ThrowAsync<ScriptWorkerProtocolException>(async () =>
|
|
await ScriptWorkerFrameCodec.ReadAsync(stream, 4, TestContext.Current.CancellationToken));
|
|
}
|
|
|
|
public static TheoryData<string, byte[]> BadFrames()
|
|
{
|
|
var good = Sample().ToByteArray();
|
|
var header = new byte[4];
|
|
BinaryPrimitives.WriteInt32LittleEndian(header, good.Length);
|
|
var wrongVersion = new Envelope { ProtocolVersion = 99, Heartbeat = new Heartbeat() }.ToByteArray();
|
|
var noBody = new Envelope { ProtocolVersion = 1, Seq = 1 }.ToByteArray();
|
|
return new TheoryData<string, byte[]>
|
|
{
|
|
{ "zero length", new byte[] { 0, 0, 0, 0 } },
|
|
{ "negative length", new byte[] { 0xFF, 0xFF, 0xFF, 0xFF } },
|
|
{ "truncated header", new byte[] { 5, 0 } },
|
|
{ "truncated body", [.. header, .. good[..^1]] },
|
|
{ "garbage body", [5, 0, 0, 0, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF] },
|
|
{ "wrong protocol version", Frame(wrongVersion) },
|
|
{ "no body", Frame(noBody) },
|
|
};
|
|
}
|
|
|
|
private static byte[] Frame(byte[] body)
|
|
{
|
|
var header = new byte[4];
|
|
BinaryPrimitives.WriteInt32LittleEndian(header, body.Length);
|
|
return [.. header, .. body];
|
|
}
|
|
|
|
[Theory]
|
|
[MemberData(nameof(BadFrames))]
|
|
public async Task Malformed_frames_are_protocol_faults(string why, byte[] bytes)
|
|
{
|
|
using var stream = new MemoryStream(bytes);
|
|
await Should.ThrowAsync<ScriptWorkerProtocolException>(async () =>
|
|
await ScriptWorkerFrameCodec.ReadAsync(stream, 1024, TestContext.Current.CancellationToken), why);
|
|
}
|
|
|
|
/// <summary>Random bytes never escape as anything but a protocol fault (or a clean end of stream).</summary>
|
|
[Fact]
|
|
public async Task Fuzzed_input_only_ever_yields_a_protocol_fault()
|
|
{
|
|
var random = new Random(780);
|
|
for (var i = 0; i < 3000; i++)
|
|
{
|
|
var bytes = new byte[random.Next(0, 64)];
|
|
random.NextBytes(bytes);
|
|
if (bytes.Length >= 4 && random.Next(2) == 0)
|
|
BinaryPrimitives.WriteInt32LittleEndian(bytes, random.Next(1, bytes.Length));
|
|
using var stream = new MemoryStream(bytes);
|
|
try
|
|
{
|
|
while (await ScriptWorkerFrameCodec.ReadAsync(stream, 1024, TestContext.Current.CancellationToken) is { } envelope)
|
|
{
|
|
// A frame that parses must still decode its variants without an unexpected exception type.
|
|
if (envelope.BodyCase == Envelope.BodyOneofCase.Results)
|
|
foreach (var r in envelope.Results.Items)
|
|
try { VariantCodec.Decode(r.Value); } catch (Exception ex) when (ex is ScriptWorkerProtocolException or OverflowException) { }
|
|
}
|
|
}
|
|
catch (ScriptWorkerProtocolException)
|
|
{
|
|
}
|
|
}
|
|
}
|
|
|
|
// ---------------------------------------------------------------- buffered reader / combining writer (lmxopcua#789)
|
|
|
|
/// <summary>Hands out at most <c>chunk</c> bytes per read, so frames straddle reads at every possible offset.</summary>
|
|
private sealed class ChunkedStream(byte[] data, int chunk) : MemoryStream(data)
|
|
{
|
|
public override int Read(byte[] buffer, int offset, int count) => base.Read(buffer, offset, Math.Min(count, chunk));
|
|
|
|
public override ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default)
|
|
=> base.ReadAsync(buffer[..Math.Min(buffer.Length, chunk)], cancellationToken);
|
|
}
|
|
|
|
private static byte[] Frames(int count, int bigEvery = 0)
|
|
{
|
|
using var stream = new MemoryStream();
|
|
for (var i = 0; i < count; i++)
|
|
{
|
|
var env = new Envelope { ProtocolVersion = 1, Seq = (ulong)i + 1, Heartbeat = new Heartbeat { Inflight = (uint)i } };
|
|
if (bigEvery > 0 && i % bigEvery == 0)
|
|
env = new Envelope { ProtocolVersion = 1, Seq = (ulong)i + 1, Fault = new Fault { Message = new string('x', 100_000 + i) } };
|
|
ScriptWorkerFrameCodec.WriteAsync(stream, env, ScriptWorkerFrameCodec.DefaultMaxFrameBytes).AsTask().GetAwaiter().GetResult();
|
|
}
|
|
return stream.ToArray();
|
|
}
|
|
|
|
/// <summary>
|
|
/// The buffered reader yields exactly the frames the per-frame codec wrote, in order, however the bytes are
|
|
/// split across reads — including frames larger than its initial 64 KiB buffer — through both its async and its
|
|
/// blocking read, and ends cleanly at a frame boundary.
|
|
/// </summary>
|
|
[Theory]
|
|
[InlineData(1, false)]
|
|
[InlineData(3, true)]
|
|
[InlineData(7, false)]
|
|
[InlineData(65_536, true)]
|
|
[InlineData(1_000_000, false)]
|
|
public async Task The_buffered_reader_reassembles_frames_split_at_any_offset(int chunk, bool blocking)
|
|
{
|
|
var bytes = Frames(200, bigEvery: 37);
|
|
var reader = new ScriptWorkerFrameReader(new ChunkedStream(bytes, chunk));
|
|
for (var i = 0; i < 200; i++)
|
|
{
|
|
var env = (blocking ? reader.Read() : await reader.ReadAsync(TestContext.Current.CancellationToken)).ShouldNotBeNull($"frame {i}");
|
|
env.Seq.ShouldBe((ulong)i + 1);
|
|
if (i % 37 == 0) env.Fault.Message.Length.ShouldBe(100_000 + i);
|
|
else env.Heartbeat.Inflight.ShouldBe((uint)i);
|
|
}
|
|
(blocking ? reader.Read() : await reader.ReadAsync(TestContext.Current.CancellationToken)).ShouldBeNull("a clean end of stream");
|
|
}
|
|
|
|
/// <summary>The buffered reader rejects exactly what the codec rejects.</summary>
|
|
[Theory]
|
|
[MemberData(nameof(BadFrames))]
|
|
public async Task The_buffered_reader_rejects_malformed_frames_like_the_codec(string why, byte[] bytes)
|
|
{
|
|
await Should.ThrowAsync<ScriptWorkerProtocolException>(async () =>
|
|
await new ScriptWorkerFrameReader(new MemoryStream(bytes), 1024).ReadAsync(TestContext.Current.CancellationToken), why);
|
|
Should.Throw<ScriptWorkerProtocolException>(() => new ScriptWorkerFrameReader(new ChunkedStream(bytes, 1), 1024).Read(), why);
|
|
}
|
|
|
|
/// <summary>Random bytes never escape the buffered reader as anything but a protocol fault (or a clean end).</summary>
|
|
[Fact]
|
|
public void Fuzzed_input_only_ever_yields_a_protocol_fault_from_the_buffered_reader()
|
|
{
|
|
var random = new Random(789);
|
|
for (var i = 0; i < 3000; i++)
|
|
{
|
|
var bytes = new byte[random.Next(0, 64)];
|
|
random.NextBytes(bytes);
|
|
if (bytes.Length >= 4 && random.Next(2) == 0)
|
|
BinaryPrimitives.WriteInt32LittleEndian(bytes, random.Next(1, bytes.Length));
|
|
var reader = new ScriptWorkerFrameReader(new ChunkedStream(bytes, random.Next(1, 9)), 1024);
|
|
try { while (reader.Read() is not null) { } }
|
|
catch (ScriptWorkerProtocolException) { }
|
|
}
|
|
}
|
|
|
|
/// <summary>A stream whose first write waits on a gate, so frames queue up behind a write in progress.</summary>
|
|
private sealed class GatedStream : MemoryStream
|
|
{
|
|
public readonly TaskCompletionSource Gate = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
public int Writes;
|
|
|
|
public override async ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken = default)
|
|
{
|
|
if (Interlocked.Increment(ref Writes) == 1) await Gate.Task;
|
|
await base.WriteAsync(buffer, cancellationToken);
|
|
}
|
|
}
|
|
|
|
private static List<Envelope> ReadAll(byte[] bytes)
|
|
{
|
|
var reader = new ScriptWorkerFrameReader(new MemoryStream(bytes));
|
|
var frames = new List<Envelope>();
|
|
while (reader.Read() is { } env) frames.Add(env);
|
|
return frames;
|
|
}
|
|
|
|
private static Envelope Eval(ulong id) => new() { Eval = new EvalBatch { Items = { new EvalRequest { EvalId = id, Key = "k" } } } };
|
|
|
|
/// <summary>
|
|
/// The combining writer (lmxopcua#789): frames queued while a write is in progress go out in the next pass, with
|
|
/// consecutive evaluations merged into one batch, every other frame kept in its place (a Cancel is never
|
|
/// reordered ahead of the evaluation it cancels), and sequence numbers increasing.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task The_combining_writer_batches_consecutive_evaluations_and_keeps_frame_order()
|
|
{
|
|
var stream = new GatedStream();
|
|
Exception? fault = null;
|
|
var writer = new ScriptWorkerFrameWriter(stream, ex => fault = ex, firstSeq: 1);
|
|
|
|
writer.TryWrite(Eval(1)).ShouldBeTrue(); // starts a pass that blocks on the gate
|
|
writer.TryWrite(Eval(2)).ShouldBeTrue();
|
|
writer.TryWrite(Eval(3)).ShouldBeTrue();
|
|
writer.TryWrite(new Envelope { Cancel = new Cancel { EvalId = 2 } }).ShouldBeTrue();
|
|
writer.TryWrite(Eval(4)).ShouldBeTrue();
|
|
stream.Gate.SetResult();
|
|
await Scripts.Eventually(() =>
|
|
{
|
|
try { return ReadAll(stream.ToArray()).Count == 4; }
|
|
catch (ScriptWorkerProtocolException) { return false; } // a pass caught mid-copy
|
|
}, TimeSpan.FromSeconds(5), "both passes written");
|
|
|
|
fault.ShouldBeNull();
|
|
var frames = ReadAll(stream.ToArray());
|
|
frames.Select(f => f.BodyCase).ShouldBe([Envelope.BodyOneofCase.Eval, Envelope.BodyOneofCase.Eval,
|
|
Envelope.BodyOneofCase.Cancel, Envelope.BodyOneofCase.Eval]);
|
|
frames[0].Eval.Items.Select(e => e.EvalId).ShouldBe([1UL]);
|
|
frames[1].Eval.Items.Select(e => e.EvalId).ShouldBe([2UL, 3UL], "queued behind the write in progress: one batch");
|
|
frames[2].Cancel.EvalId.ShouldBe(2UL);
|
|
frames[3].Eval.Items.Select(e => e.EvalId).ShouldBe([4UL]);
|
|
frames.Select(f => f.Seq).ShouldBe([2UL, 3UL, 4UL, 5UL]);
|
|
frames.ShouldAllBe(f => f.ProtocolVersion == ScriptWorkerFrameCodec.ProtocolVersion);
|
|
stream.Writes.ShouldBe(2, "one write per pass");
|
|
}
|
|
|
|
/// <summary>A failed write completes the writer, reports once, and later frames are refused rather than lost silently.</summary>
|
|
[Fact]
|
|
public void A_failed_write_completes_the_writer_and_reports_once()
|
|
{
|
|
var faults = 0;
|
|
var stream = new MemoryStream();
|
|
stream.Dispose();
|
|
var writer = new ScriptWorkerFrameWriter(stream, _ => Interlocked.Increment(ref faults));
|
|
writer.TryWrite(Eval(1)).ShouldBeTrue();
|
|
faults.ShouldBe(1);
|
|
writer.IsCompleted.ShouldBeTrue();
|
|
writer.TryWrite(Eval(2)).ShouldBeFalse();
|
|
faults.ShouldBe(1);
|
|
}
|
|
|
|
[Fact]
|
|
public void Backoff_doubles_from_250ms_to_a_30s_cap_with_20_percent_jitter()
|
|
{
|
|
WorkerShard.NextBackoff(1, 0.5).ShouldBe(TimeSpan.FromMilliseconds(250));
|
|
WorkerShard.NextBackoff(2, 0.5).ShouldBe(TimeSpan.FromMilliseconds(500));
|
|
WorkerShard.NextBackoff(3, 0.5).ShouldBe(TimeSpan.FromSeconds(1));
|
|
WorkerShard.NextBackoff(20, 0.5).ShouldBe(TimeSpan.FromSeconds(30));
|
|
WorkerShard.NextBackoff(1, 0).ShouldBe(TimeSpan.FromMilliseconds(200));
|
|
WorkerShard.NextBackoff(1, 0.9999999).TotalMilliseconds.ShouldBe(300, tolerance: 0.01);
|
|
WorkerShard.NextBackoff(40, 1).ShouldBe(TimeSpan.FromSeconds(36));
|
|
}
|
|
|
|
[Fact]
|
|
public void Shard_placement_is_stable_and_in_range()
|
|
{
|
|
// FNV-1a, not string.GetHashCode (randomised per process): a script stays on its shard across restarts.
|
|
OutOfProcessScriptRunner.ShardIndex("1:vt-1", 2).ShouldBe(OutOfProcessScriptRunner.ShardIndex("1:vt-1", 2));
|
|
var spread = Enumerable.Range(0, 400).Select(i => OutOfProcessScriptRunner.ShardIndex($"1:vt-{i}", 4)).ToArray();
|
|
spread.ShouldAllBe(i => i >= 0 && i < 4);
|
|
spread.Distinct().Count().ShouldBe(4);
|
|
OutOfProcessScriptRunner.ShardIndex("abc", 1000).ShouldBe((int)(0x1A47E90Bu % 1000));
|
|
}
|
|
}
|