CIP codec (#564): PatchU16 bounds guard; CipReply content equality; CipPath port/length/ASCII/odd/oversize validation with ArgumentException for malformed tag names and routes; CipPath.MessageRouter/ConnectionManager are now ReadOnlySpan<byte>; refused Forward Open surfaces ForwardOpenRefusedException with structured general/extended status; test CaptureFile uses BinaryPrimitives. EncapsulationClient (#565): frame write bounded by the exchange timeout, which faults (never cancels) the connection; finite positive timeouts enforced at every entry point; gate wait bounded by the timeout; deterministic reorder of the 300 ms test; new dispose-during-drain, gate-wait and RegisterSession zero-context tests. Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
@@ -21,7 +21,18 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Eip;
|
||||
/// (<see cref="AbStatus.BadReply"/>) closes the socket; <see cref="IsConnected"/> turns false and
|
||||
/// every later call throws <see cref="AbTransportException"/>(<see cref="AbStatus.BadConnection"/>).
|
||||
/// The owner reconnects with a fresh client.</para>
|
||||
/// <para><b>No half-written frames.</b> A frame write never observes a cancellation token.</para>
|
||||
/// <para><b>No half-written frames.</b> A frame write never observes a cancellation token. It is
|
||||
/// nevertheless bounded: the exchange timeout also covers the write, and if it elapses first the
|
||||
/// connection is <i>faulted</i> (socket closed, which fails the pending write) rather than the write
|
||||
/// being cancelled — so a peer that stops reading cannot hold the single-flight gate forever, and a
|
||||
/// torn frame can only ever sit on a socket that is already dead.</para>
|
||||
/// <para><b>Timeouts are finite.</b> Every entry point rejects a non-positive or infinite timeout
|
||||
/// (<see cref="ArgumentOutOfRangeException"/>): the abandoned-exchange drain is bounded by it, so an
|
||||
/// infinite one would let a cancelled caller's drain hold the gate forever. The gate wait itself is
|
||||
/// bounded by the same timeout (<see cref="AbStatus.Timeout"/>, nothing written, connection intact).</para>
|
||||
/// <para><b>RegisterSession.</b> It goes out with a zero sender context (ab_server refuses a non-zero
|
||||
/// one) and therefore relies on the unconnected "reply context == 0 is tolerated" rule in
|
||||
/// <c>TryMatch</c> to match its reply; both ends of that dependency are pinned by tests.</para>
|
||||
/// </remarks>
|
||||
public sealed class EncapsulationClient : IAsyncDisposable
|
||||
{
|
||||
@@ -55,6 +66,7 @@ public sealed class EncapsulationClient : IAsyncDisposable
|
||||
/// bounds the TCP connect and the RegisterSession exchange separately.</summary>
|
||||
public static async Task<EncapsulationClient> ConnectAsync(string host, int port, TimeSpan timeout, CancellationToken ct)
|
||||
{
|
||||
ValidateTimeout(timeout);
|
||||
var tcp = new TcpClient { NoDelay = true };
|
||||
try
|
||||
{
|
||||
@@ -96,6 +108,7 @@ public sealed class EncapsulationClient : IAsyncDisposable
|
||||
/// <summary>Send one unconnected (UCMM) CIP request and return the CIP reply bytes.</summary>
|
||||
public async Task<ReadOnlyMemory<byte>> SendRRDataAsync(ReadOnlyMemory<byte> cip, TimeSpan timeout, CancellationToken ct)
|
||||
{
|
||||
ValidateTimeout(timeout);
|
||||
var secs = (ushort)Math.Clamp((int)Math.Ceiling(timeout.TotalSeconds), 1, ushort.MaxValue);
|
||||
var (_, reply) = await ExchangeAsync(connected: false,
|
||||
(ctx, _) => EncapsulationCodec.SendRRData(SessionHandle, ctx, cip.Span, secs), timeout, ct).ConfigureAwait(false);
|
||||
@@ -107,12 +120,19 @@ public sealed class EncapsulationClient : IAsyncDisposable
|
||||
public async Task<ReadOnlyMemory<byte>> SendUnitDataAsync(uint otConnectionId, ReadOnlyMemory<byte> cip,
|
||||
TimeSpan timeout, CancellationToken ct)
|
||||
{
|
||||
ValidateTimeout(timeout);
|
||||
var (_, reply) = await ExchangeAsync(connected: true,
|
||||
(ctx, seq) => EncapsulationCodec.SendUnitData(SessionHandle, ctx, otConnectionId, seq, cip.Span), timeout, ct)
|
||||
.ConfigureAwait(false);
|
||||
return reply;
|
||||
}
|
||||
|
||||
private static void ValidateTimeout(TimeSpan timeout)
|
||||
{
|
||||
if (timeout <= TimeSpan.Zero || timeout == Timeout.InfiniteTimeSpan)
|
||||
throw new ArgumentOutOfRangeException(nameof(timeout), timeout, "Timeout must be finite and positive.");
|
||||
}
|
||||
|
||||
/// <summary>The connected sequence count after <paramref name="current"/>: increments, wrapping
|
||||
/// 65535 → 1 (0 is never put on the wire).</summary>
|
||||
internal static ushort NextSequence(ushort current) => current == ushort.MaxValue ? (ushort)1 : (ushort)(current + 1);
|
||||
@@ -123,7 +143,9 @@ public sealed class EncapsulationClient : IAsyncDisposable
|
||||
bool connected, Func<ulong, ushort, byte[]> build, TimeSpan timeout, CancellationToken ct)
|
||||
{
|
||||
ThrowIfUnusable();
|
||||
await _gate.WaitAsync(ct).ConfigureAwait(false);
|
||||
if (!await _gate.WaitAsync(timeout, ct).ConfigureAwait(false))
|
||||
throw new AbTransportException(AbStatus.Timeout,
|
||||
$"Timed out after {timeout.TotalMilliseconds:F0} ms waiting for the session's single-flight gate.");
|
||||
|
||||
Task<(EncapsulationHeader, ReadOnlyMemory<byte>)> exchange;
|
||||
try
|
||||
@@ -168,7 +190,13 @@ public sealed class EncapsulationClient : IAsyncDisposable
|
||||
using var timeoutCts = new CancellationTokenSource(timeout);
|
||||
try
|
||||
{
|
||||
await _stream.WriteAsync(request, CancellationToken.None).ConfigureAwait(false);
|
||||
// The write ignores cancellation (a half-written frame would desync the stream) but is
|
||||
// bounded: on expiry the catch below faults the connection, closing the socket, which
|
||||
// is what fails the still-pending write.
|
||||
var write = _stream.WriteAsync(request, CancellationToken.None).AsTask();
|
||||
_ = write.ContinueWith(static t => _ = t.Exception, CancellationToken.None,
|
||||
TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);
|
||||
await write.WaitAsync(timeoutCts.Token).ConfigureAwait(false);
|
||||
while (true)
|
||||
{
|
||||
var frame = await ReadFrameAsync(timeoutCts.Token).ConfigureAwait(false);
|
||||
@@ -212,6 +240,8 @@ public sealed class EncapsulationClient : IAsyncDisposable
|
||||
|
||||
if (!connected)
|
||||
{
|
||||
// "== 0" is load-bearing, not just ab_server tolerance: RegisterSession is sent with a ZERO
|
||||
// sender context (ab_server refuses non-zero), so its reply echoes 0 and matches ONLY here.
|
||||
if (h.SenderContext != ctx && h.SenderContext != 0) return false;
|
||||
var p = EncapsulationCodec.ParseReply(frame, expectConnected: false);
|
||||
reply = (p.Header, p.Cip);
|
||||
|
||||
Reference in New Issue
Block a user