Joseph Doherty
03ac57d5d6
- Host reader: the frame read and its handling are caught separately. Only the read's I/O exceptions mean the pipe ended; anything HandleFrame's inline completion work throws now kills that worker as ProtocolFault (outstanding runs -> worker restarted / BadResourceUnavailable) instead of escaping a dedicated thread (process crash) or being swallowed as a clean end (an InvalidOperationException used to be, leaving the worker unread until a Crash 2 s later). - Worker reader: the same catch-all around HandleFrame (exit ProtocolFault). - Worker: Cancel() runs on the pool, so a script's blocking ctx.CancellationToken callback cannot stop the worker reading its pipe. - Host write-stall check: ScriptWorkerFrameWriter.WriteInProgressFor; the heartbeat monitor kills a worker whose pipe write has been blocked longer than max(10 s, silence limit) as KilledUnresponsive, which fails the write and releases the producer (seam submit, deploy load, Cancel). Fixes the wrong 'heartbeat monitor ends it' comment on the pipe buffer quota. - StopAsync sends Shutdown off the caller so its 1 s budget holds on a full pipe. - WindowsBlockingPipeStream refuses the async surface (Stream's base versions serialise a write behind a waiting read). - Tests: PipeFaultTests (throwing completion callback -> ProtocolFault and a BadResourceUnavailable sibling, shard serves again; a worker that stops reading but heartbeats is killed within the bound and the blocked producer returns, via the Hello.test_stall_reader_on_diag seam); BlockingPipeStreamTests dispose with a write blocked on a full pipe, and with a read and a write both pending; Category=ScriptWorker on the protocol, handshake and pipe test classes. Mutation checks: no stall kill -> stall test red; catch-all without the fault -> callback test red (Crash). Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk
88 lines
3.5 KiB
C#
88 lines
3.5 KiB
C#
using System.IO.Pipes;
|
|
using System.Security.Cryptography;
|
|
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 §6 "Pipe hardening": only the launched process, echoing its nonce, is accepted.</summary>
|
|
[Trait("Category", "ScriptWorker")]
|
|
public sealed class HandshakeTests
|
|
{
|
|
private static CancellationToken Ct => TestContext.Current.CancellationToken;
|
|
|
|
private static async Task<Exception?> HandshakeWith(Func<Envelope, Envelope> reply, string expectedNonce = "right", int? expectedPid = 1234)
|
|
{
|
|
var name = "otopcua-sw-test-" + Convert.ToHexString(RandomNumberGenerator.GetBytes(6));
|
|
await using var server = new NamedPipeServerStream(name, PipeDirection.InOut, 1, PipeTransmissionMode.Byte,
|
|
PipeOptions.Asynchronous | PipeOptions.CurrentUserOnly);
|
|
await using var client = new NamedPipeClientStream(".", name, PipeDirection.InOut, PipeOptions.Asynchronous | PipeOptions.CurrentUserOnly);
|
|
var accept = server.WaitForConnectionAsync(Ct);
|
|
await client.ConnectAsync(TimeSpan.FromSeconds(5), Ct);
|
|
await accept;
|
|
|
|
var rogue = Task.Run(async () =>
|
|
{
|
|
var hello = await ScriptWorkerFrameCodec.ReadAsync(client, ScriptWorkerFrameCodec.DefaultMaxFrameBytes, Ct);
|
|
await ScriptWorkerFrameCodec.WriteAsync(client, reply(hello!), ScriptWorkerFrameCodec.DefaultMaxFrameBytes, Ct);
|
|
await client.FlushAsync(Ct);
|
|
}, Ct);
|
|
|
|
try
|
|
{
|
|
using var timeout = CancellationTokenSource.CreateLinkedTokenSource(Ct);
|
|
timeout.CancelAfter(TimeSpan.FromSeconds(5));
|
|
await WorkerConnection.HandshakeAsync(server, new Hello { Nonce = expectedNonce }, expectedNonce, expectedPid, timeout.Token);
|
|
await rogue;
|
|
return null;
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
return ex;
|
|
}
|
|
}
|
|
|
|
private static Envelope Ready(string nonce, int pid) => new()
|
|
{
|
|
ProtocolVersion = ScriptWorkerFrameCodec.ProtocolVersion,
|
|
Ready = new Ready { Nonce = nonce, Pid = pid },
|
|
};
|
|
|
|
[Fact]
|
|
public async Task The_launched_process_echoing_its_nonce_is_accepted()
|
|
=> (await HandshakeWith(_ => Ready("right", 1234))).ShouldBeNull("positive control");
|
|
|
|
[Fact]
|
|
public async Task A_wrong_nonce_is_refused()
|
|
=> (await HandshakeWith(_ => Ready("wrong", 1234))).ShouldBeOfType<ScriptWorkerProtocolException>()
|
|
.Message.ShouldContain("nonce");
|
|
|
|
[Fact]
|
|
public async Task A_client_that_is_not_the_launched_process_is_refused()
|
|
=> (await HandshakeWith(_ => Ready("right", 999))).ShouldBeOfType<ScriptWorkerProtocolException>()
|
|
.Message.ShouldContain("pid");
|
|
|
|
[Fact]
|
|
public async Task Anything_but_ready_first_is_refused()
|
|
=> (await HandshakeWith(_ => new Envelope
|
|
{
|
|
ProtocolVersion = ScriptWorkerFrameCodec.ProtocolVersion,
|
|
Heartbeat = new Heartbeat(),
|
|
})).ShouldBeOfType<ScriptWorkerProtocolException>();
|
|
|
|
[Fact]
|
|
public async Task The_hello_carries_the_nonce_to_echo()
|
|
{
|
|
string? seen = null;
|
|
(await HandshakeWith(hello =>
|
|
{
|
|
seen = hello.Hello.Nonce;
|
|
return Ready(hello.Hello.Nonce, 1234);
|
|
})).ShouldBeNull();
|
|
seen.ShouldBe("right");
|
|
}
|
|
}
|