From 286822d83e20ca652f4345c7e8d697db44804f90 Mon Sep 17 00:00:00 2001 From: Eduard Dumitru Date: Wed, 1 Jul 2026 11:40:46 +0200 Subject: [PATCH 1/8] .NET: add IpcContext ambient context (POCO callback-capable contracts) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Introduce `IpcContext` with a static `IpcContext? Current` backed by AsyncLocal, published for the duration of a server-side handler (and callback) invocation in `Server.MethodCall`. It exposes the peer (`Client`) + the call's `CancellationToken` and a `GetCallback()` that mirrors `Message.Client.GetCallback()` — so a service-contract implementation can reach callbacks WITHOUT a `Message` parameter, letting the contract-defining assembly stay free of a UiPath.Ipc reference. Additive and non-breaking: `Message` injection is unchanged; `Current` is null outside a call and composes across nested calls (a callback serviced mid-call). Tests (xUnit, self-contained POCO contract with no `Message` param): `Current` is null outside a call and after it completes, set while honoring a call, and a Message-free contract reaches its callback purely via `IpcContext.Current.GetCallback()`. Builds on net461/net6.0/net6.0-windows. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/UiPath.CoreIpc.Tests/IpcContextTests.cs | 110 ++++++++++++++++++++ src/UiPath.CoreIpc/IpcContext.cs | 87 ++++++++++++++++ src/UiPath.CoreIpc/Server/Server.cs | 28 +++-- 3 files changed, 214 insertions(+), 11 deletions(-) create mode 100644 src/UiPath.CoreIpc.Tests/IpcContextTests.cs create mode 100644 src/UiPath.CoreIpc/IpcContext.cs diff --git a/src/UiPath.CoreIpc.Tests/IpcContextTests.cs b/src/UiPath.CoreIpc.Tests/IpcContextTests.cs new file mode 100644 index 00000000..dcff9e5d --- /dev/null +++ b/src/UiPath.CoreIpc.Tests/IpcContextTests.cs @@ -0,0 +1,110 @@ +using Microsoft.Extensions.Logging; +using UiPath.Ipc.Transport.NamedPipe; + +namespace UiPath.Ipc.Tests; + +// A POCO service contract: it declares NO `Message` parameter, yet its +// implementation reaches the caller's callback purely via `IpcContext.Current`. +// This is the whole point of IpcContext — a contract assembly need not reference +// UiPath.Ipc to be callback-capable. +public interface IContextProbe +{ + Task ReachCallbackViaContext(); + Task ContextIsSet(); +} + +public interface IContextProbeCallback +{ + Task Pong(); +} + +public sealed class ContextProbe : IContextProbe +{ + public const string PongValue = "pong-from-callback"; + + // No Message parameter anywhere: the peer is reached through the ambient context. + public Task ReachCallbackViaContext() + => IpcContext.Current!.GetCallback().Pong(); + + public Task ContextIsSet() => Task.FromResult(IpcContext.Current is not null); +} + +public sealed class ContextProbeCallback : IContextProbeCallback +{ + public Task Pong() => Task.FromResult(ContextProbe.PongValue); +} + +public sealed class IpcContextTests +{ + [Fact] + public void Current_IsNull_OutsideAnyIpcCall() + => IpcContext.Current.ShouldBeNull(); + + [Fact] + public async Task Current_IsSet_WhileHonoringACall() + { + await using var pair = await Pair.Create(); + (await pair.Proxy.ContextIsSet()).ShouldBeTrue(); + } + + [Fact] + public async Task PocoContract_ReachesCallback_ViaIpcContext() + { + await using var pair = await Pair.Create(); + (await pair.Proxy.ReachCallbackViaContext()).ShouldBe(ContextProbe.PongValue); + } + + [Fact] + public async Task Current_IsNullAgain_AfterTheCallCompletes() + { + await using var pair = await Pair.Create(); + await pair.Proxy.ContextIsSet(); + // The ambient value must not leak into the test's own async flow. + IpcContext.Current.ShouldBeNull(); + } + + private sealed class Pair : IAsyncDisposable + { + private readonly IpcServer _server; + public IContextProbe Proxy { get; } + + private Pair(IpcServer server, IContextProbe proxy) + { + _server = server; + Proxy = proxy; + } + + public static async Task Create() + { + var pipeName = $"ipctest_ctx_{Guid.NewGuid():N}"; + + var server = new IpcServer + { + Transport = new NamedPipeServerTransport { PipeName = pipeName }, + Endpoints = new() { typeof(IContextProbe) }, + ServiceProvider = new ServiceCollection() + .AddLogging() + .AddSingleton() + .BuildServiceProvider(), + }; + + var client = new IpcClient + { + Transport = new NamedPipeClientTransport { PipeName = pipeName }, + Callbacks = new() { { typeof(IContextProbeCallback), new ContextProbeCallback() } }, + }; + var proxy = client.GetProxy(); + + server.Start(); + await Task.Yield(); + return new Pair(server, proxy); + } + + public async ValueTask DisposeAsync() + { + (Proxy as IpcProxy)?.Dispose(); + await ((Proxy as IpcProxy)?.CloseConnection() ?? default); + await _server.DisposeAsync(); + } + } +} diff --git a/src/UiPath.CoreIpc/IpcContext.cs b/src/UiPath.CoreIpc/IpcContext.cs new file mode 100644 index 00000000..7d9439b9 --- /dev/null +++ b/src/UiPath.CoreIpc/IpcContext.cs @@ -0,0 +1,87 @@ +using System; +using System.Threading; + +namespace UiPath.Ipc; + +/// +/// Ambient, per-operation context for the IPC call currently being honored on +/// the executing async flow. is non-null exactly while a +/// server executes an inbound request handler (or a callback handler) and is +/// otherwise — so IpcContext.Current is not null +/// is a precise "am I inside an IPC call?" signal. +/// +/// It lets a POCO service-contract implementation reach the peer — e.g. +/// IpcContext.Current!.GetCallback<TCallback>() — without taking a +/// parameter, so the contract-defining assembly need not +/// reference UiPath.Ipc. It coexists with (does not replace) . +/// +public sealed class IpcContext +{ + private static readonly AsyncLocal CurrentContext = new(); + + /// + /// The context of the IPC call in progress on the current async flow, or + /// if no IPC call is being honored here. + /// + public static IpcContext? Current => CurrentContext.Value; + + internal IpcContext(IClient? client, CancellationToken cancellationToken) + { + Client = client; + CancellationToken = cancellationToken; + } + + /// + /// The peer of the in-flight call (the same handle as ). + /// when the current endpoint has no reachable peer + /// (e.g. a callback being serviced on the client side). + /// + public IClient? Client { get; } + + /// The cancellation token of the in-flight call. + public CancellationToken CancellationToken { get; } + + /// + /// Returns a proxy to the peer's callback contract without needing a + /// parameter — equivalent to + /// Message.Client.GetCallback<TCallback>(). + /// + /// + /// There is no peer for the current context ( is ). + /// + public TCallback GetCallback() where TCallback : class + => (Client ?? throw new InvalidOperationException( + $"{nameof(IpcContext)}.{nameof(Current)} has no peer client; " + + $"{nameof(GetCallback)} is only available while honoring a server-side IPC call.")) + .GetCallback(); + + /// + /// Publishes as for the + /// lifetime of the returned scope, restoring the previous value on dispose + /// (so a nested call — a callback serviced mid-call — composes correctly). + /// + internal static IDisposable Push(IpcContext context) + { + var previous = CurrentContext.Value; + CurrentContext.Value = context; + return new Scope(previous); + } + + private sealed class Scope : IDisposable + { + private readonly IpcContext? _previous; + private bool _disposed; + + public Scope(IpcContext? previous) => _previous = previous; + + public void Dispose() + { + if (_disposed) + { + return; + } + _disposed = true; + CurrentContext.Value = _previous; + } + } +} diff --git a/src/UiPath.CoreIpc/Server/Server.cs b/src/UiPath.CoreIpc/Server/Server.cs index e957d43f..f11401cc 100644 --- a/src/UiPath.CoreIpc/Server/Server.cs +++ b/src/UiPath.CoreIpc/Server/Server.cs @@ -167,21 +167,27 @@ async ValueTask InvokeMethod() Task ScheduleMethodCall() => defaultScheduler ? MethodCall() : RunOnScheduler(); async Task MethodCall() { - await (route.BeforeCall?.Invoke( - new CallInfo(newConnection: false, method.MethodInfo, arguments), - cancellationToken) ?? Task.CompletedTask); + // Publish the ambient IpcContext for the duration of the call so a + // POCO contract implementation can reach the peer + // (IpcContext.Current.GetCallback()) without a Message parameter. + // Restored on scope exit; nested calls (a callback serviced + // mid-call) compose via the saved-previous in IpcContext.Push. + using (IpcContext.Push(new IpcContext(_client, cancellationToken))) + { + await (route.BeforeCall?.Invoke( + new CallInfo(newConnection: false, method.MethodInfo, arguments), + cancellationToken) ?? Task.CompletedTask); - Task invocationTask = null!; + Task invocationTask = method.Invoke(service, arguments, cancellationToken); + await invocationTask; - invocationTask = method.Invoke(service, arguments, cancellationToken); - await invocationTask; + if (!returnTaskType.IsGenericType) + { + return null; + } - if (!returnTaskType.IsGenericType) - { - return null; + return GetTaskResult(returnTaskType, invocationTask); } - - return GetTaskResult(returnTaskType, invocationTask); } Task RunOnScheduler() => Task.Factory.StartNew(MethodCall, cancellationToken, TaskCreationOptions.DenyChildAttach, scheduler).Unwrap(); From 042c806dc4bfa42bf0d0d4a190515752cd0f8e6b Mon Sep 17 00:00:00 2001 From: Eduard Dumitru Date: Wed, 1 Jul 2026 12:01:24 +0200 Subject: [PATCH 2/8] Python: add IpcContext ambient context (contextvars, POCO callback contracts) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Python counterpart of the .NET IpcContext. `IpcContext.Current` (a metaclass property over a task-local ContextVar) is non-None exactly while a handler is being honored, and exposes `get_callback(contract)` — so a POCO handler reaches the peer without a `Message` parameter, and the contract module need not import uipath_ipc. Published in `IpcConnection._invoke_callback` via `IpcContext._activate(self)`; no explicit reset is needed because the dispatch runs in its own asyncio task, whose contextvars copy is task-local (never leaks to the receive loop or sibling tasks, dropped when the task ends). Additive and non-breaking: `Message` injection is unchanged. Exported from the package root. Tests (real Python↔Python pair over TCP loopback): Current is None outside a call and doesn't leak into the caller's task, is set while honoring a call, and a Message-free POCO service reaches the client's callback purely via IpcContext.Current.get_callback. Full unit suite: 241 passed, 26 skipped. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../uipath-ipc/src/uipath_ipc/__init__.py | 2 + .../src/uipath_ipc/client/connection.py | 7 ++ .../uipath-ipc/src/uipath_ipc/context.py | 65 +++++++++++++++ .../python/uipath-ipc/tests/test_context.py | 82 +++++++++++++++++++ 4 files changed, 156 insertions(+) create mode 100644 src/Clients/python/uipath-ipc/src/uipath_ipc/context.py create mode 100644 src/Clients/python/uipath-ipc/tests/test_context.py diff --git a/src/Clients/python/uipath-ipc/src/uipath_ipc/__init__.py b/src/Clients/python/uipath-ipc/src/uipath_ipc/__init__.py index 69bd2765..718455e2 100644 --- a/src/Clients/python/uipath-ipc/src/uipath_ipc/__init__.py +++ b/src/Clients/python/uipath-ipc/src/uipath_ipc/__init__.py @@ -1,6 +1,7 @@ """uipath-ipc — Python client and server for UiPath.Ipc.""" from .client import IpcClient, IpcConnection +from .context import IpcContext from .errors import EndpointNotFoundError, MethodNotFoundError, RemoteException from .hooks import BeforeCallHandler, BeforeConnectHandler, CallInfo from .markers import ipc_cancellable @@ -27,6 +28,7 @@ "MethodNotFoundError", "IpcClient", "IpcConnection", + "IpcContext", "IpcServer", "Message", "NamedPipeClientTransport", diff --git a/src/Clients/python/uipath-ipc/src/uipath_ipc/client/connection.py b/src/Clients/python/uipath-ipc/src/uipath_ipc/client/connection.py index a873126e..4bb39492 100644 --- a/src/Clients/python/uipath-ipc/src/uipath_ipc/client/connection.py +++ b/src/Clients/python/uipath-ipc/src/uipath_ipc/client/connection.py @@ -45,6 +45,7 @@ from ..hooks import BeforeCallHandler, CallInfo from ..errors import EndpointNotFoundError, MethodNotFoundError, RemoteException +from ..context import IpcContext from ..markers import is_ipc_cancellable from ..message import Message from ..transport.base import ClientTransport @@ -731,6 +732,12 @@ async def _invoke_callback(self, req: Request) -> None: exception only logged — mirroring .NET's non-generic `Task`. """ deferred_one_way: tuple[Callable[..., object], list, dict] | None = None + # Publish the ambient IpcContext for this dispatch so a POCO handler can + # reach the peer via IpcContext.Current.get_callback(...) without a + # Message parameter. No reset needed: _invoke_callback runs in its own + # asyncio task (see _handle_incoming_request), whose contextvars copy is + # task-local, so the value never leaks and is dropped when the task ends. + IpcContext._activate(self) try: entry = self._callbacks.get(req.endpoint) if entry is None: diff --git a/src/Clients/python/uipath-ipc/src/uipath_ipc/context.py b/src/Clients/python/uipath-ipc/src/uipath_ipc/context.py new file mode 100644 index 00000000..3338ab59 --- /dev/null +++ b/src/Clients/python/uipath-ipc/src/uipath_ipc/context.py @@ -0,0 +1,65 @@ +"""Ambient, per-operation IPC context (`IpcContext.Current`). + +The Python counterpart of the .NET ``IpcContext`` / ``AsyncLocal`` feature: it +lets a POCO service-contract handler reach the caller's callback WITHOUT a +`Message` parameter, so a contract module need not import anything from +``uipath_ipc``. +""" + +from __future__ import annotations + +import contextvars +from typing import TYPE_CHECKING, TypeVar + +if TYPE_CHECKING: + from .client.connection import IpcConnection + +T = TypeVar("T") + +#: Task-local by construction: an asyncio Task copies the current context at +#: creation, so a value set inside one handler task never leaks to the receive +#: loop or sibling handler tasks, and is dropped when that task ends. +_current: "contextvars.ContextVar[IpcContext | None]" = contextvars.ContextVar( + "uipath_ipc_current_context", default=None +) + + +class _IpcContextMeta(type): + @property + def Current(cls) -> "IpcContext | None": # noqa: N802 - PascalCase mirrors .NET/TS + """The context of the IPC call being honored on the current asyncio + task, or ``None`` if no call is in progress here.""" + return _current.get() + + +class IpcContext(metaclass=_IpcContextMeta): + """Ambient, per-operation context for the IPC call currently being honored. + + ``IpcContext.Current`` is non-``None`` exactly while a server handler (or a + callback) runs on this task, and ``None`` otherwise — so + ``IpcContext.Current is not None`` is a precise "am I inside an IPC call?" + signal. A POCO handler can reach the peer via + ``IpcContext.Current.get_callback(SomeCallback)`` without declaring a + `Message` parameter (the counterpart of .NET's ``IpcContext.Current``). + Coexists with — does not replace — `Message` injection. + """ + + __slots__ = ("_connection",) + + def __init__(self, connection: "IpcConnection") -> None: + self._connection = connection + + def get_callback(self, contract: type[T]) -> T: + """Return a proxy to the peer's callback contract — the ambient + equivalent of an injected ``message.client.get_callback(contract)``.""" + return self._connection.get_callback(contract) + + @staticmethod + def _activate(connection: "IpcConnection") -> None: + """Publish an `IpcContext` for `connection` as `Current` on this task. + + No explicit reset: the caller (`IpcConnection._invoke_callback`) runs in + its own asyncio task, whose contextvars copy is task-local — the value + never leaks to other tasks and is discarded when the task completes. + """ + _current.set(IpcContext(connection)) diff --git a/src/Clients/python/uipath-ipc/tests/test_context.py b/src/Clients/python/uipath-ipc/tests/test_context.py new file mode 100644 index 00000000..f7903e16 --- /dev/null +++ b/src/Clients/python/uipath-ipc/tests/test_context.py @@ -0,0 +1,82 @@ +"""Tests for the ambient IpcContext (POCO callback-capable contracts).""" + +from __future__ import annotations + +import asyncio +from abc import ABC, abstractmethod + +from uipath_ipc import ( + IpcClient, + IpcContext, + IpcServer, + TcpClientTransport, + TcpServerTransport, +) + +PONG = "pong-from-callback" + + +# A POCO service contract: it declares NO `Message` parameter, yet its handler +# reaches the caller's callback purely through `IpcContext.Current` — the whole +# point of IpcContext (a contract module need not import anything from uipath_ipc). +class IContextProbe(ABC): + @abstractmethod + async def ReachCallbackViaContext(self) -> str: ... + + @abstractmethod + async def ContextIsSet(self) -> bool: ... + + +class IContextProbeCallback(ABC): + @abstractmethod + async def Pong(self) -> str: ... + + +class ContextProbe: # duck-typed; no Message anywhere in sight + async def ReachCallbackViaContext(self) -> str: + ctx = IpcContext.Current + assert ctx is not None + return await ctx.get_callback(IContextProbeCallback).Pong() + + async def ContextIsSet(self) -> bool: + return IpcContext.Current is not None + + +class ContextProbeCallback: + async def Pong(self) -> str: + return PONG + + +def _endpoint(server: IpcServer) -> tuple[str, int]: + assert server.handle is not None + return server.handle.sockets[0].getsockname()[:2] # type: ignore[attr-defined] + + +def test_current_is_none_outside_a_call() -> None: + assert IpcContext.Current is None + + +async def test_current_is_set_while_honoring_a_call() -> None: + server = IpcServer(TcpServerTransport("127.0.0.1", 0), {IContextProbe: ContextProbe()}) + async with server: + host, port = _endpoint(server) + async with IpcClient( + TcpClientTransport(host, port), + callbacks={IContextProbeCallback: ContextProbeCallback()}, + ) as client: + svc = client.get_proxy(IContextProbe) + assert await asyncio.wait_for(svc.ContextIsSet(), timeout=5) is True + # The ambient value must not have leaked into the test's own task. + assert IpcContext.Current is None + + +async def test_poco_contract_reaches_callback_via_ipccontext() -> None: + server = IpcServer(TcpServerTransport("127.0.0.1", 0), {IContextProbe: ContextProbe()}) + async with server: + host, port = _endpoint(server) + async with IpcClient( + TcpClientTransport(host, port), + callbacks={IContextProbeCallback: ContextProbeCallback()}, + ) as client: + svc = client.get_proxy(IContextProbe) + assert await asyncio.wait_for(svc.ReachCallbackViaContext(), timeout=5) == PONG From c549985c1a1ad11969df7170e92ffbd6916d330f Mon Sep 17 00:00:00 2001 From: Eduard Dumitru Date: Wed, 1 Jul 2026 12:15:43 +0200 Subject: [PATCH 3/8] TypeScript: accept AbortSignal wherever a CancellationToken is accepted MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A contract method may now pass a Web/Node AbortSignal in place of the TS CancellationToken clone. RpcRequestFactory routes each argument through AbortSignalAdapter.ensureCancellationToken (check-and-adapt in one gulp: passes a CancellationToken through, bridges an AbortSignal, else undefined) and substitutes the result in-place, so the wire form AND the ending-CancellationToken handling are byte-identical to passing a CancellationToken. Purely additive: CancellationToken / CancellationTokenSource are unchanged; no wire change. The bridge (toCancellationToken) drives a CancellationTokenSource from the signal's 'abort' (immediately if already aborted) and disposes it once it fires; the source is created with no cancelAfter delay, so it holds no timer (dispose is belt-and-suspenders) and the 'abort' listener is registered `once`. Tests (Jasmine): isAbortSignal recognition; live signal cancels + fires registrations on abort; already-aborted -> already-cancelled; ensureCancellationToken passes a token through, adapts a signal, returns undefined otherwise. tsc (src + test) clean; 7/7 specs pass. Note: TS IpcContext (POCO callback contracts) and callee-side AbortSignal are intentionally NOT included — the js client has no callee cancellation at all (RpcCallContext.Incomming carries no CancellationToken; TS doesn't send/observe cancel frames, a known parity gap) and no handler-side reach-back (no getCallback; Message carries no Client). Both are net-new features, deferred. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../bcl/cancellation/AbortSignalAdapter.ts | 65 +++++++++++++++++ .../js/src/std/bcl/cancellation/index.ts | 1 + .../src/std/core/Proxies/RpcRequestFactory.ts | 29 +++++--- .../test/std/bcl/AbortSignalAdapter.test.ts | 73 +++++++++++++++++++ 4 files changed, 157 insertions(+), 11 deletions(-) create mode 100644 src/Clients/js/src/std/bcl/cancellation/AbortSignalAdapter.ts create mode 100644 src/Clients/js/test/std/bcl/AbortSignalAdapter.test.ts diff --git a/src/Clients/js/src/std/bcl/cancellation/AbortSignalAdapter.ts b/src/Clients/js/src/std/bcl/cancellation/AbortSignalAdapter.ts new file mode 100644 index 00000000..b110a840 --- /dev/null +++ b/src/Clients/js/src/std/bcl/cancellation/AbortSignalAdapter.ts @@ -0,0 +1,65 @@ +import { CancellationToken } from './CancellationToken'; +import { CancellationTokenSource } from './CancellationTokenSource'; + +/** + * Bridges a Web/Node `AbortSignal` to a UiPath.Ipc `CancellationToken`, so a + * contract method may accept an `AbortSignal` anywhere a `CancellationToken` is + * accepted. Purely additive: `CancellationToken` / `CancellationTokenSource` + * are unchanged and remain fully supported. + */ +export class AbortSignalAdapter { + /** Whether `arg` is an `AbortSignal` (guarded for runtimes without it). */ + public static isAbortSignal(arg: unknown): arg is AbortSignal { + return typeof AbortSignal !== 'undefined' && arg instanceof AbortSignal; + } + + /** + * Check-and-adapt in one gulp: returns `candidate` unchanged if it is + * already a `CancellationToken`, the bridged token if it is an + * `AbortSignal`, or `undefined` if it is neither (i.e. not a cancellation + * argument). Lets callers treat both cancellation shapes uniformly. + */ + public static ensureCancellationToken( + candidate: unknown, + ): CancellationToken | undefined { + if (candidate instanceof CancellationToken) { + return candidate; + } + if (AbortSignalAdapter.isAbortSignal(candidate)) { + return AbortSignalAdapter.toCancellationToken(candidate); + } + return undefined; + } + + /** + * Returns a `CancellationToken` that is cancelled when `signal` aborts — + * immediately if it is already aborted. + * + * The backing `CancellationTokenSource` is created with no `cancelAfter` + * delay, so its only disposable resource (the delay timer) is never + * allocated; we still `dispose()` it right after it fires — for hygiene and + * to stay correct if the source ever gains resources. A never-aborting + * signal's source holds nothing and is collected together with the signal + * (the `abort` listener is registered `once`, so it self-removes). + */ + public static toCancellationToken(signal: AbortSignal): CancellationToken { + const cts = new CancellationTokenSource(); + if (signal.aborted) { + cts.cancel(); + cts.dispose(); + } else { + signal.addEventListener( + 'abort', + () => { + try { + cts.cancel(); + } finally { + cts.dispose(); + } + }, + { once: true }, + ); + } + return cts.token; + } +} diff --git a/src/Clients/js/src/std/bcl/cancellation/index.ts b/src/Clients/js/src/std/bcl/cancellation/index.ts index a1481b17..ef53f188 100644 --- a/src/Clients/js/src/std/bcl/cancellation/index.ts +++ b/src/Clients/js/src/std/bcl/cancellation/index.ts @@ -1,6 +1,7 @@ export * from './CancellationTokenSource'; export * from './CancellationTokenRegistration'; export * from './CancellationToken'; +export * from './AbortSignalAdapter'; export * from './EmptyCancellationToken'; export * from './RandomCancellationToken'; export * from './LinkedCancellationTokenSource'; diff --git a/src/Clients/js/src/std/core/Proxies/RpcRequestFactory.ts b/src/Clients/js/src/std/core/Proxies/RpcRequestFactory.ts index 08c6a9eb..d37cb6a9 100644 --- a/src/Clients/js/src/std/core/Proxies/RpcRequestFactory.ts +++ b/src/Clients/js/src/std/core/Proxies/RpcRequestFactory.ts @@ -1,4 +1,5 @@ import { + AbortSignalAdapter, CancellationToken, Primitive, PublicCtor, @@ -41,13 +42,24 @@ export class RpcRequestFactory { let message: Message | undefined; let ct: CancellationToken | undefined; + // Copy so an AbortSignal can be substituted in-place (below). + const args = [...params.args]; - for (const arg of params.args) { + for (let i = 0; i < args.length; i++) { + const arg = args[i]; if (arg instanceof Message) { message = arg; + continue; } - if (arg instanceof CancellationToken) { - ct = arg; + // Accept a CancellationToken OR (bridged) an AbortSignal wherever a + // cancellation argument is expected, and substitute it in-place so + // the wire form AND the ending-CancellationToken handling below are + // identical for both. For a real CancellationToken this is a no-op + // substitution. Additive — CancellationToken still works unchanged. + const token = AbortSignalAdapter.ensureCancellationToken(arg); + if (token) { + ct = token; + args[i] = token; } } @@ -61,17 +73,12 @@ export class RpcRequestFactory { ) ?? Timeout.infiniteTimeSpan; - let args = params.args; - if ( hasEndingCt && - (params.args.length === 0 || - !( - params.args[params.args.length - 1] instanceof - CancellationToken - )) + (args.length === 0 || + !(args[args.length - 1] instanceof CancellationToken)) ) { - args = [...args, CancellationToken.none]; + args.push(CancellationToken.none); } const rpcRequest = Converter.toRpcRequest( diff --git a/src/Clients/js/test/std/bcl/AbortSignalAdapter.test.ts b/src/Clients/js/test/std/bcl/AbortSignalAdapter.test.ts new file mode 100644 index 00000000..286bc506 --- /dev/null +++ b/src/Clients/js/test/std/bcl/AbortSignalAdapter.test.ts @@ -0,0 +1,73 @@ +import { AbortSignalAdapter, CancellationToken } from '../../../src/std'; + +import { expect } from 'chai'; + +describe(`${AbortSignalAdapter.name}`, () => { + describe('isAbortSignal', () => { + it('recognizes an AbortSignal and rejects everything else', () => { + const ac = new AbortController(); + expect(AbortSignalAdapter.isAbortSignal(ac.signal)).to.equal(true); + expect(AbortSignalAdapter.isAbortSignal(CancellationToken.none)).to.equal(false); + expect(AbortSignalAdapter.isAbortSignal(undefined)).to.equal(false); + expect(AbortSignalAdapter.isAbortSignal(null)).to.equal(false); + expect(AbortSignalAdapter.isAbortSignal({})).to.equal(false); + }); + }); + + describe('toCancellationToken', () => { + it('yields a token that is not yet cancelled for a live signal', () => { + const ac = new AbortController(); + const ct = AbortSignalAdapter.toCancellationToken(ac.signal); + expect(ct.isCancellationRequested).to.equal(false); + }); + + it('cancels the token (and fires registrations) when the signal aborts', () => { + const ac = new AbortController(); + const ct = AbortSignalAdapter.toCancellationToken(ac.signal); + + let fired = false; + ct.register(() => { + fired = true; + }); + + ac.abort(); + + expect(ct.isCancellationRequested).to.equal(true); + expect(fired).to.equal(true); + }); + + it('yields an already-cancelled token for an already-aborted signal', () => { + const ac = new AbortController(); + ac.abort(); + + const ct = AbortSignalAdapter.toCancellationToken(ac.signal); + + expect(ct.isCancellationRequested).to.equal(true); + }); + }); + + describe('ensureCancellationToken', () => { + it('passes a CancellationToken through unchanged', () => { + const ct = CancellationToken.none; + expect(AbortSignalAdapter.ensureCancellationToken(ct)).to.equal(ct); + }); + + it('adapts an AbortSignal to a CancellationToken', () => { + const ac = new AbortController(); + const token = AbortSignalAdapter.ensureCancellationToken(ac.signal); + + expect(token).to.be.instanceOf(CancellationToken); + expect(token!.isCancellationRequested).to.equal(false); + + ac.abort(); + + expect(token!.isCancellationRequested).to.equal(true); + }); + + it('returns undefined for a non-cancellation value', () => { + expect(AbortSignalAdapter.ensureCancellationToken({})).to.equal(undefined); + expect(AbortSignalAdapter.ensureCancellationToken(42)).to.equal(undefined); + expect(AbortSignalAdapter.ensureCancellationToken(undefined)).to.equal(undefined); + }); + }); +}); From c391dc76cc076b272bee477c52caf4283e197c2e Mon Sep 17 00:00:00 2001 From: Eduard Dumitru Date: Wed, 1 Jul 2026 14:38:11 +0200 Subject: [PATCH 4/8] TypeScript: support cancellation in hosted callbacks (callee side) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A callback the peer invokes on a TS client can now be cancelled. Previously an inbound CancellationRequest frame hit a `Method not implemented.` stub in RpcChannel and was swallowed-and-logged, so the running handler never learned of the cancel. - RpcChannel now tracks each in-flight incoming call in an IncomingCallTable (a per-call CancellationTokenSource keyed by request id) and implements processIncommingCancellationRequest to cancel the matching call; in-flight calls are also cancelled when the channel is disposed. RpcCallContext.Incomming carries the per-call token (defaulted to none, so existing sites are unaffected). - ChannelManager.invokeCallback injects that token into the handler when the callback contract declares a trailing cancellation parameter: a live CancellationToken, or a bridged AbortSignal (via AbortSignalAdapter.toAbortSignal) when the contract asks for one. This is metadata-driven and unambiguous — it fires only for a registered (decorated) contract, resolved by endpoint name via the new ContractStore.maybeGetByEndpoint. Callbacks registered by bare endpoint name carry no parameter-type metadata and the empty-string wire form of a cancellation slot is indistinguishable from a real empty string, so absent metadata the arguments are left untouched. Fully additive; no breaking changes. Adds unit tests for IncomingCallTable, the RpcChannel frame->cancel path, the invokeCallback injection (CancellationToken, AbortSignal, and the no-metadata no-op), and AbortSignalAdapter.toAbortSignal; updates LIMITATIONS.md. 381 std specs pass. Co-Authored-By: Claude Opus 4.8 (1M context) --- LIMITATIONS.md | 2 +- .../bcl/cancellation/AbortSignalAdapter.ts | 21 ++ .../js/src/std/core/Contract/ContractStore.ts | 9 + .../src/std/core/Contract/IContractStore.ts | 7 + .../std/core/Contract/OperationDescriptor.ts | 1 + .../core/Contract/OperationDescriptorImpl.ts | 10 + .../core/Protocol/Rpc/IncomingCallTable.ts | 52 ++++ .../std/core/Protocol/Rpc/RpcCallContext.ts | 6 + .../src/std/core/Protocol/Rpc/RpcChannel.ts | 29 ++- .../js/src/std/core/Protocol/Rpc/index.ts | 1 + .../js/src/std/core/Proxies/ChannelManager.ts | 42 +++- .../test/std/bcl/AbortSignalAdapter.test.ts | 39 ++- .../std/core/CallbackCancellation.test.ts | 224 ++++++++++++++++++ .../test/std/core/IncomingCallTable.test.ts | 84 +++++++ .../std/core/RpcChannelCancellation.test.ts | 149 ++++++++++++ 15 files changed, 665 insertions(+), 11 deletions(-) create mode 100644 src/Clients/js/src/std/core/Protocol/Rpc/IncomingCallTable.ts create mode 100644 src/Clients/js/test/std/core/CallbackCancellation.test.ts create mode 100644 src/Clients/js/test/std/core/IncomingCallTable.test.ts create mode 100644 src/Clients/js/test/std/core/RpcChannelCancellation.test.ts diff --git a/LIMITATIONS.md b/LIMITATIONS.md index 2a5b3108..5eab3d72 100644 --- a/LIMITATIONS.md +++ b/LIMITATIONS.md @@ -34,7 +34,7 @@ Contracts are a **shared agreement** between both peers, and every client *trust - **Propagation to the peer differs:** - **.NET** — a fired token sends a `CancellationRequest`. - **Python** — sends one **only for methods marked `@ipc_cancellable`** (otherwise the cancel is local-only); a hosted handler honors an inbound cancel only when its contract method is `@ipc_cancellable`. - - **TypeScript** — **does not send a cancel frame at all** (local-only; the remote keeps running) and throws on an inbound one. Known parity gap. + - **TypeScript** — as a **caller**, still **does not send a cancel frame** for its own outbound calls (local-only; the remote keeps running) — the remaining parity gap. As a **callee** (a hosted callback), it now **honors an inbound cancel**: the running handler's per-call token fires, and — when the callback contract declares a trailing `CancellationToken`/`AbortSignal` — the live token (or a bridged `AbortSignal`) is injected so the handler can observe it. A callback registered by bare endpoint name carries no parameter-type metadata, so its arguments are left untouched (the cancel still fires the token, but nothing is injected into the handler). - A successful response that arrives before the cancel is delivered — the result wins. ## Transports & features diff --git a/src/Clients/js/src/std/bcl/cancellation/AbortSignalAdapter.ts b/src/Clients/js/src/std/bcl/cancellation/AbortSignalAdapter.ts index b110a840..f8a65a5c 100644 --- a/src/Clients/js/src/std/bcl/cancellation/AbortSignalAdapter.ts +++ b/src/Clients/js/src/std/bcl/cancellation/AbortSignalAdapter.ts @@ -62,4 +62,25 @@ export class AbortSignalAdapter { } return cts.token; } + + /** + * The reverse of {@link toCancellationToken}: returns an `AbortSignal` that + * aborts when `token` is cancelled — immediately if it already is. Used on + * the callee side to hand a callback handler an `AbortSignal` in place of a + * `CancellationToken`. Throws if the runtime has no `AbortController`. + */ + public static toAbortSignal(token: CancellationToken): AbortSignal { + if (typeof AbortController === 'undefined') { + throw new Error( + 'Cannot adapt a CancellationToken to an AbortSignal: this runtime has no AbortController.', + ); + } + const controller = new AbortController(); + if (token.isCancellationRequested) { + controller.abort(); + } else { + token.register(() => controller.abort()); + } + return controller.signal; + } } diff --git a/src/Clients/js/src/std/core/Contract/ContractStore.ts b/src/Clients/js/src/std/core/Contract/ContractStore.ts index 56efdcaf..33a7105d 100644 --- a/src/Clients/js/src/std/core/Contract/ContractStore.ts +++ b/src/Clients/js/src/std/core/Contract/ContractStore.ts @@ -12,6 +12,15 @@ export class ContractStore implements IContractStore { return this._map.get($class); } + maybeGetByEndpoint(endpoint: string): ServiceDescriptor | undefined { + for (const descriptor of this._map.values()) { + if (descriptor.endpoint === endpoint) { + return descriptor; + } + } + return undefined; + } + private add($class: PublicCtor): ServiceDescriptor { const descriptor = new ServiceDescriptorImpl($class); this._map.set($class, descriptor); diff --git a/src/Clients/js/src/std/core/Contract/IContractStore.ts b/src/Clients/js/src/std/core/Contract/IContractStore.ts index 5d86f98f..a7040a00 100644 --- a/src/Clients/js/src/std/core/Contract/IContractStore.ts +++ b/src/Clients/js/src/std/core/Contract/IContractStore.ts @@ -6,4 +6,11 @@ export interface IContractStore { getOrCreate($class: PublicCtor): ServiceDescriptor; maybeGet($class: PublicCtor): ServiceDescriptor | undefined; + + /** + * Looks up a registered contract by its endpoint name (defaults to the + * contract class name). Used on the callee side, where an incoming call + * identifies its contract by endpoint name rather than by class. + */ + maybeGetByEndpoint(endpoint: string): ServiceDescriptor | undefined; } diff --git a/src/Clients/js/src/std/core/Contract/OperationDescriptor.ts b/src/Clients/js/src/std/core/Contract/OperationDescriptor.ts index 40db16da..7d560c8a 100644 --- a/src/Clients/js/src/std/core/Contract/OperationDescriptor.ts +++ b/src/Clients/js/src/std/core/Contract/OperationDescriptor.ts @@ -6,6 +6,7 @@ export type OperationDescriptor = { readonly methodName: string; readonly hasEndingCancellationToken: boolean; + readonly hasEndingAbortSignal: boolean; readonly returnType: PublicCtor; readonly parameterTypes: readonly PublicCtor[]; returnsPromiseOf?: PublicCtor | Primitive; diff --git a/src/Clients/js/src/std/core/Contract/OperationDescriptorImpl.ts b/src/Clients/js/src/std/core/Contract/OperationDescriptorImpl.ts index 13c2f146..bec0f8cb 100644 --- a/src/Clients/js/src/std/core/Contract/OperationDescriptorImpl.ts +++ b/src/Clients/js/src/std/core/Contract/OperationDescriptorImpl.ts @@ -29,6 +29,16 @@ export class OperationDescriptorImpl implements OperationDescriptor { parameterTypes[parameterTypes.length - 1] === (CancellationToken as any) ); } + public get hasEndingAbortSignal(): boolean { + const parameterTypes = this.parameterTypes; + + return ( + typeof AbortSignal !== 'undefined' && + parameterTypes && + parameterTypes.length > 0 && + parameterTypes[parameterTypes.length - 1] === (AbortSignal as any) + ); + } public get returnType(): PublicCtor { return Reflect.getMetadata( 'design:returntype', diff --git a/src/Clients/js/src/std/core/Protocol/Rpc/IncomingCallTable.ts b/src/Clients/js/src/std/core/Protocol/Rpc/IncomingCallTable.ts new file mode 100644 index 00000000..50eef65b --- /dev/null +++ b/src/Clients/js/src/std/core/Protocol/Rpc/IncomingCallTable.ts @@ -0,0 +1,52 @@ +import { CancellationToken, CancellationTokenSource } from '../../../bcl'; + +/** + * Tracks in-flight *incoming* calls (callbacks the peer invokes on us) by their + * request id, each backed by a {@link CancellationTokenSource}. It is the callee + * mirror of the outgoing-call table: it lets an inbound `CancellationRequest` + * frame cancel the matching running handler. + * + * @internal + */ +/* @internal */ +export class IncomingCallTable { + private readonly _map = new Map(); + + /** + * Starts tracking `requestId` and returns the token that fires when a + * cancellation for it arrives (or the call is torn down). + */ + public register(requestId: string): CancellationToken { + const cts = new CancellationTokenSource(); + this._map.set(requestId, cts); + return cts.token; + } + + /** Cancels the tracked call `requestId`, if still in flight. */ + public tryCancel(requestId: string): void { + const cts = this._map.get(requestId); + if (cts && !cts.isCancellationRequested) { + cts.cancel(); + } + } + + /** Stops tracking `requestId` (the call finished) and releases its source. */ + public complete(requestId: string): void { + const cts = this._map.get(requestId); + if (cts) { + this._map.delete(requestId); + cts.dispose(); + } + } + + /** Cancels and releases every tracked call — used when the channel dies. */ + public clear(): void { + for (const cts of this._map.values()) { + if (!cts.isCancellationRequested) { + cts.cancel(); + } + cts.dispose(); + } + this._map.clear(); + } +} diff --git a/src/Clients/js/src/std/core/Protocol/Rpc/RpcCallContext.ts b/src/Clients/js/src/std/core/Protocol/Rpc/RpcCallContext.ts index 75052bef..0f410881 100644 --- a/src/Clients/js/src/std/core/Protocol/Rpc/RpcCallContext.ts +++ b/src/Clients/js/src/std/core/Protocol/Rpc/RpcCallContext.ts @@ -11,6 +11,12 @@ export module RpcCallContext { constructor( public readonly request: RpcMessage.Request, public readonly respond: (response: RpcMessage.Response) => Promise, + /** + * Cancelled when the peer sends a `CancellationRequest` for this + * call (or the channel tears it down). Defaults to a token that + * never cancels, so pre-existing construction sites are unaffected. + */ + public readonly ct: CancellationToken = CancellationToken.none, ) {} } diff --git a/src/Clients/js/src/std/core/Protocol/Rpc/RpcChannel.ts b/src/Clients/js/src/std/core/Protocol/Rpc/RpcChannel.ts index a906501e..e106003f 100644 --- a/src/Clients/js/src/std/core/Protocol/Rpc/RpcChannel.ts +++ b/src/Clients/js/src/std/core/Protocol/Rpc/RpcChannel.ts @@ -11,7 +11,7 @@ import { import { ConnectHelper, Address } from '../..'; import { MessageStream, IMessageStream, Network } from '..'; -import { IRpcChannel, RpcCallContext, RpcMessage } from '.'; +import { IRpcChannel, RpcCallContext, RpcMessage, IncomingCallTable } from '.'; /* @internal */ export class RpcChannel implements IRpcChannel { @@ -88,6 +88,7 @@ export class RpcChannel implements IRpcChannel { public async disposeAsync(): Promise { if (!this._isDisposed) { this._isDisposed = true; + this._incomingCalls.clear(); try { const messageStream = await this._$messageStream; await messageStream.disposeAsync(); @@ -137,6 +138,7 @@ export class RpcChannel implements IRpcChannel { private readonly _$messageStream: Promise; private readonly _outgoingCalls = new RpcChannel.OutgoingCallTable(); + private readonly _incomingCalls = new IncomingCallTable(); private readonly _networkObserver = new (class implements Observer { constructor(private readonly _owner: RpcChannel) {} @@ -189,20 +191,31 @@ export class RpcChannel implements IRpcChannel { private processIncommingRequest(message: Network.Message): void { const request = RpcMessage.Request.fromNetwork(message); - const context = new RpcCallContext.Incomming(request, async (response) => { - response.RequestId = request.Id; - await ( - await this._$messageStream - ).writeMessageAsync(response.toNetwork(), CancellationToken.none); - }); + const ct = this._incomingCalls.register(request.Id); + const context = new RpcCallContext.Incomming( + request, + async (response) => { + try { + response.RequestId = request.Id; + await ( + await this._$messageStream + ).writeMessageAsync(response.toNetwork(), CancellationToken.none); + } finally { + this._incomingCalls.complete(request.Id); + } + }, + ct, + ); try { this._observer.next(context); } catch (err) { + this._incomingCalls.complete(request.Id); Trace.log(UnknownError.ensureError(err)); } } private processIncommingCancellationRequest(message: Network.Message): void { - throw new Error('Method not implemented.'); + const cancellationRequest = RpcMessage.CancellationRequest.fromNetwork(message); + this._incomingCalls.tryCancel(cancellationRequest.RequestId); } } diff --git a/src/Clients/js/src/std/core/Protocol/Rpc/index.ts b/src/Clients/js/src/std/core/Protocol/Rpc/index.ts index 715c8ab9..4ba2f441 100644 --- a/src/Clients/js/src/std/core/Protocol/Rpc/index.ts +++ b/src/Clients/js/src/std/core/Protocol/Rpc/index.ts @@ -1,6 +1,7 @@ export * from './IRpcChannel'; export * from './RpcChannel'; export * from './RpcCallContext'; +export * from './IncomingCallTable'; export * from './RpcMessageBase'; export * from './RpcMessage'; export * from './IncommingInitiatingRpcMessage'; diff --git a/src/Clients/js/src/std/core/Proxies/ChannelManager.ts b/src/Clients/js/src/std/core/Proxies/ChannelManager.ts index c0eda5dd..c4f0f965 100644 --- a/src/Clients/js/src/std/core/Proxies/ChannelManager.ts +++ b/src/Clients/js/src/std/core/Proxies/ChannelManager.ts @@ -1,4 +1,4 @@ -import { CancellationToken, IAsyncDisposable, PublicCtor, Timeout, TimeSpan } from '../..'; +import { AbortSignalAdapter, CancellationToken, IAsyncDisposable, PublicCtor, Timeout, TimeSpan } from '../..'; import { IServiceProvider, @@ -80,6 +80,7 @@ export class ChannelManager implements IAsyncDisposable { private async invokeCallback( request: RpcMessage.Request, + ct: CancellationToken, ): Promise { const callback = this._sp.callbackStore.get( request.Endpoint, @@ -102,6 +103,7 @@ export class ChannelManager implements IAsyncDisposable { } const args = request.Parameters.map(jsonArg => JSON.parse(jsonArg)); + this.injectCancellation(request, args, ct); let data: string | null = null; let error: IpcError | null = null; @@ -114,6 +116,43 @@ export class ChannelManager implements IAsyncDisposable { return new RpcMessage.Response(request.Id, data, error); } + + /** + * If the callback contract declares a trailing cancellation parameter, + * replaces the (blanked) trailing wire argument with the per-call token — + * or a bridged `AbortSignal` when the contract asks for one — so the handler + * observes cancellation exactly like a .NET service handler would. + * + * This is deliberately metadata-driven: it fires only for a registered + * (decorated) callback contract. Callbacks registered by bare endpoint name + * carry no parameter-type metadata, and the empty-string wire form of a + * cancellation slot is indistinguishable from a legitimate empty string, so + * guessing would risk clobbering a real argument. Absent metadata we leave + * the arguments untouched — fully backward compatible. + */ + private injectCancellation( + request: RpcMessage.Request, + args: unknown[], + ct: CancellationToken, + ): void { + if (args.length === 0) { + return; + } + + const operation = this._sp.contractStore + .maybeGetByEndpoint(request.Endpoint) + ?.operations.maybeGet(request.MethodName as never); + if (!operation) { + return; + } + + if (operation.hasEndingAbortSignal) { + args[args.length - 1] = AbortSignalAdapter.toAbortSignal(ct); + } else if (operation.hasEndingCancellationToken) { + args[args.length - 1] = ct; + } + } + private readonly _incommingCallObserver = new (class implements Observer { @@ -125,6 +164,7 @@ export class ChannelManager implements IAsyncDisposable { incommingContext.respond( await this._channelManager.invokeCallback( incommingContext.request, + incommingContext.ct, ), ); } diff --git a/src/Clients/js/test/std/bcl/AbortSignalAdapter.test.ts b/src/Clients/js/test/std/bcl/AbortSignalAdapter.test.ts index 286bc506..c50bcd92 100644 --- a/src/Clients/js/test/std/bcl/AbortSignalAdapter.test.ts +++ b/src/Clients/js/test/std/bcl/AbortSignalAdapter.test.ts @@ -1,4 +1,4 @@ -import { AbortSignalAdapter, CancellationToken } from '../../../src/std'; +import { AbortSignalAdapter, CancellationToken, CancellationTokenSource } from '../../../src/std'; import { expect } from 'chai'; @@ -70,4 +70,41 @@ describe(`${AbortSignalAdapter.name}`, () => { expect(AbortSignalAdapter.ensureCancellationToken(undefined)).to.equal(undefined); }); }); + + describe('toAbortSignal', () => { + it('yields a signal that is not yet aborted for a live token', () => { + const cts = new CancellationTokenSource(); + const signal = AbortSignalAdapter.toAbortSignal(cts.token); + expect(signal).to.be.instanceOf(AbortSignal); + expect(signal.aborted).to.equal(false); + }); + + it('aborts the signal when the token is cancelled', () => { + const cts = new CancellationTokenSource(); + const signal = AbortSignalAdapter.toAbortSignal(cts.token); + + cts.cancel(); + + expect(signal.aborted).to.equal(true); + }); + + it('yields an already-aborted signal for an already-cancelled token', () => { + const cts = new CancellationTokenSource(); + cts.cancel(); + + const signal = AbortSignalAdapter.toAbortSignal(cts.token); + + expect(signal.aborted).to.equal(true); + }); + + it('round-trips CancellationToken -> AbortSignal -> CancellationToken', () => { + const cts = new CancellationTokenSource(); + const signal = AbortSignalAdapter.toAbortSignal(cts.token); + const ct = AbortSignalAdapter.toCancellationToken(signal); + + expect(ct.isCancellationRequested).to.equal(false); + cts.cancel(); + expect(ct.isCancellationRequested).to.equal(true); + }); + }); }); diff --git a/src/Clients/js/test/std/core/CallbackCancellation.test.ts b/src/Clients/js/test/std/core/CallbackCancellation.test.ts new file mode 100644 index 00000000..a9756d0b --- /dev/null +++ b/src/Clients/js/test/std/core/CallbackCancellation.test.ts @@ -0,0 +1,224 @@ +import 'reflect-metadata'; + +import { + Address, + CallbackStoreImpl, + CancellationToken, + CancellationTokenSource, + ChannelManager, + ConnectHelper, + ContractStore, + RpcCallContext, + RpcMessage, + Socket, + TimeSpan, +} from '../../../src/std'; + +import { NodeAddressBuilder } from '../../../src/node/NodeAddressBuilder'; +import { MockServiceProvider } from '../../infrastructure'; + +import { expect } from 'chai'; + +// A callback contract shaped as the @ipc decorators would leave it: registered +// in the contract store, with parameter-type metadata stamped on the prototype. +class IProbeCallback { + Ping(_message: string, _ct: CancellationToken): Promise { + throw void 0; + } + Wave(_message: string, _signal: AbortSignal): Promise { + throw void 0; + } + Plain(_message: string): Promise { + throw void 0; + } +} + +function stampContract(store: ContractStore): void { + const descriptor = store.getOrCreate(IProbeCallback); + descriptor.operations.getOrCreate('Ping' as never); + descriptor.operations.getOrCreate('Wave' as never); + descriptor.operations.getOrCreate('Plain' as never); + + Reflect.defineMetadata( + 'design:paramtypes', + [String, CancellationToken], + IProbeCallback.prototype, + 'Ping', + ); + Reflect.defineMetadata( + 'design:paramtypes', + [String, AbortSignal], + IProbeCallback.prototype, + 'Wave', + ); + Reflect.defineMetadata('design:paramtypes', [String], IProbeCallback.prototype, 'Plain'); +} + +class MockAddress extends Address { + get key() { + return 'mock:callback-cancellation'; + } + async connect(_h: ConnectHelper, _t: TimeSpan, _ct: CancellationToken): Promise { + throw new Error('unused'); + } +} + +interface InvokeResult { + received: any[]; + response: RpcMessage.Response | undefined; + cts: CancellationTokenSource; +} + +async function invokeCallback(options: { + method: string; + params: string[]; + withContract?: boolean; +}): Promise { + const address = new MockAddress(); + + const contractStore = new ContractStore(); + if (options.withContract !== false) { + stampContract(contractStore); + } + + const received: any[] = []; + const handler = { + async Ping(message: string, ct: CancellationToken): Promise { + received.push({ message, ct }); + return 'pong'; + }, + async Wave(message: string, signal: AbortSignal): Promise { + received.push({ message, signal }); + return 'wave'; + }, + async Plain(message: string): Promise { + received.push({ message }); + return 'plain'; + }, + }; + + const callbackStore = new CallbackStoreImpl(); + callbackStore.set('IProbeCallback', address, handler); + + const sp = new MockServiceProvider({ + implementation: { contractStore, callbackStore }, + }); + + const cm = new ChannelManager(sp, address, undefined as any); + + const cts = new CancellationTokenSource(); + const request = new RpcMessage.Request(0, 'IProbeCallback', options.method, options.params); + request.Id = '1'; + + let response: RpcMessage.Response | undefined; + const context = new RpcCallContext.Incomming( + request, + async (r) => { + response = r; + }, + cts.token, + ); + + await (cm as any)._incommingCallObserver.next(context); + + return { received, response, cts }; +} + +const HELLO = JSON.stringify('hi'); +const BLANK = JSON.stringify(''); // how the peer serializes a cancellation slot + +describe('callback (callee) cancellation injection', () => { + it('injects the per-call CancellationToken into a trailing token parameter', async () => { + const { received, response, cts } = await invokeCallback({ + method: 'Ping', + params: [HELLO, BLANK], + }); + + expect(received).to.have.lengthOf(1); + expect(received[0].message).to.equal('hi'); + expect(received[0].ct).to.equal(cts.token); + expect(response?.Data).to.equal(JSON.stringify('pong')); + }); + + it('the injected token actually fires when the call is cancelled', async () => { + const { received, cts } = await invokeCallback({ method: 'Ping', params: [HELLO, BLANK] }); + + const ct: CancellationToken = received[0].ct; + expect(ct.isCancellationRequested).to.equal(false); + + cts.cancel(); + + expect(ct.isCancellationRequested).to.equal(true); + }); + + it('injects a bridged AbortSignal into a trailing AbortSignal parameter', async () => { + const { received, cts } = await invokeCallback({ method: 'Wave', params: [HELLO, BLANK] }); + + const signal: AbortSignal = received[0].signal; + expect(signal).to.be.instanceOf(AbortSignal); + expect(signal.aborted).to.equal(false); + + cts.cancel(); + + expect(signal.aborted).to.equal(true); + }); + + it('does not touch arguments when the contract has no trailing cancellation param', async () => { + const { received } = await invokeCallback({ method: 'Plain', params: [HELLO] }); + + expect(received[0].message).to.equal('hi'); + expect(Object.keys(received[0])).to.deep.equal(['message']); + }); + + it('leaves arguments untouched when no contract metadata is registered', async () => { + // The common bare-endpoint / interface callback: the blanked slot must + // be passed through verbatim rather than guessed at. + const { received } = await invokeCallback({ + method: 'Ping', + params: [HELLO, BLANK], + withContract: false, + }); + + expect(received[0].message).to.equal('hi'); + expect(received[0].ct).to.equal(''); + }); +}); + +describe('ContractStore.maybeGetByEndpoint', () => { + it('finds a registered contract by its endpoint name', () => { + const store = new ContractStore(); + stampContract(store); + + const descriptor = store.maybeGetByEndpoint('IProbeCallback'); + + expect(descriptor).to.not.equal(undefined); + expect(descriptor!.endpoint).to.equal('IProbeCallback'); + }); + + it('returns undefined for an unknown endpoint', () => { + const store = new ContractStore(); + stampContract(store); + + expect(store.maybeGetByEndpoint('Nope')).to.equal(undefined); + }); +}); + +describe('OperationDescriptor cancellation-kind flags', () => { + it('detects a trailing CancellationToken vs AbortSignal vs neither', () => { + const store = new ContractStore(); + stampContract(store); + const operations = store.maybeGetByEndpoint('IProbeCallback')!.operations; + + const ping = operations.maybeGet('Ping' as never)!; + expect(ping.hasEndingCancellationToken).to.equal(true); + expect(ping.hasEndingAbortSignal).to.equal(false); + + const wave = operations.maybeGet('Wave' as never)!; + expect(wave.hasEndingAbortSignal).to.equal(true); + expect(wave.hasEndingCancellationToken).to.equal(false); + + const plain = operations.maybeGet('Plain' as never)!; + expect(plain.hasEndingCancellationToken).to.equal(false); + expect(plain.hasEndingAbortSignal).to.equal(false); + }); +}); diff --git a/src/Clients/js/test/std/core/IncomingCallTable.test.ts b/src/Clients/js/test/std/core/IncomingCallTable.test.ts new file mode 100644 index 00000000..c929860b --- /dev/null +++ b/src/Clients/js/test/std/core/IncomingCallTable.test.ts @@ -0,0 +1,84 @@ +import { CancellationToken, IncomingCallTable } from '../../../src/std'; + +import { expect } from 'chai'; + +describe(`${IncomingCallTable.name}`, () => { + it('register yields a live (uncancelled) token', () => { + const table = new IncomingCallTable(); + + const ct = table.register('1'); + + expect(ct.isCancellationRequested).to.equal(false); + }); + + it('tryCancel cancels the token (and fires registrations) of the matching call', () => { + const table = new IncomingCallTable(); + const ct = table.register('1'); + + let fired = false; + ct.register(() => { + fired = true; + }); + + table.tryCancel('1'); + + expect(ct.isCancellationRequested).to.equal(true); + expect(fired).to.equal(true); + }); + + it('tryCancel is a harmless no-op for an unknown id', () => { + const table = new IncomingCallTable(); + const ct = table.register('1'); + + expect(() => table.tryCancel('does-not-exist')).to.not.throw(); + expect(ct.isCancellationRequested).to.equal(false); + }); + + it('tryCancel twice does not throw and stays cancelled', () => { + const table = new IncomingCallTable(); + const ct = table.register('1'); + + table.tryCancel('1'); + expect(() => table.tryCancel('1')).to.not.throw(); + + expect(ct.isCancellationRequested).to.equal(true); + }); + + it('complete stops tracking so a later cancel is a no-op', () => { + const table = new IncomingCallTable(); + const ct = table.register('1'); + + table.complete('1'); + table.tryCancel('1'); + + expect(ct.isCancellationRequested).to.equal(false); + }); + + it('complete for an unknown id does not throw', () => { + const table = new IncomingCallTable(); + + expect(() => table.complete('nope')).to.not.throw(); + }); + + it('clear cancels every tracked call', () => { + const table = new IncomingCallTable(); + const a = table.register('a'); + const b = table.register('b'); + + table.clear(); + + expect(a.isCancellationRequested).to.equal(true); + expect(b.isCancellationRequested).to.equal(true); + }); + + it('distinct ids are cancelled independently', () => { + const table = new IncomingCallTable(); + const a = table.register('a'); + const b = table.register('b'); + + table.tryCancel('a'); + + expect(a.isCancellationRequested).to.equal(true); + expect(b.isCancellationRequested).to.equal(false); + }); +}); diff --git a/src/Clients/js/test/std/core/RpcChannelCancellation.test.ts b/src/Clients/js/test/std/core/RpcChannelCancellation.test.ts new file mode 100644 index 00000000..142ba15a --- /dev/null +++ b/src/Clients/js/test/std/core/RpcChannelCancellation.test.ts @@ -0,0 +1,149 @@ +import { Observer, Subject } from 'rxjs'; + +import { + Address, + CancellationToken, + ConnectHelper, + IMessageStream, + Network, + RpcCallContext, + RpcChannel, + RpcMessage, + Socket, + TimeSpan, +} from '../../../src/std'; + +import { expect } from 'chai'; + +// --- Test doubles --- + +class MockSocket extends Socket { + private readonly _data = new Subject(); + get $data() { + return this._data; + } + async write(_buffer: Buffer, _ct: CancellationToken): Promise {} + dispose(): void {} +} + +class MockAddress extends Address { + get key() { + return 'mock:rpc-cancellation'; + } + async connect(_h: ConnectHelper, _t: TimeSpan, _ct: CancellationToken): Promise { + return new MockSocket(); + } +} + +// A message stream that lets the test push inbound Network.Messages straight +// into the channel's network observer, bypassing real socket framing. +class StubMessageStream implements IMessageStream { + public readonly writes: Network.Message[] = []; + constructor(public readonly observer: Observer) {} + async writeMessageAsync(message: Network.Message, _ct: CancellationToken): Promise { + this.writes.push(message); + } + async disposeAsync(): Promise {} +} + +class StubMessageStreamFactory implements IMessageStream.Factory { + public last: StubMessageStream | undefined; + create(_stream: any, observer: Observer): IMessageStream { + return (this.last = new StubMessageStream(observer)); + } +} + +function requestFrame(id: string, endpoint: string, method: string, params: string[]): Network.Message { + const request = new RpcMessage.Request(0, endpoint, method, params); + request.Id = id; + return request.toNetwork(); +} + +function cancelFrame(id: string): Network.Message { + return new RpcMessage.CancellationRequest(id).toNetwork(); +} + +async function setup() { + const incomming: RpcCallContext.Incomming[] = []; + const rpcObserver: Observer = { + next: (context) => { + incomming.push(context); + }, + error() {}, + complete() {}, + }; + + const factory = new StubMessageStreamFactory(); + const channel = RpcChannel.create( + new MockAddress(), + {} as any as ConnectHelper, + TimeSpan.fromSeconds(30), + CancellationToken.none, + rpcObserver, + factory, + ); + + // Wait for the (async) message-stream creation, which captures the observer. + await (channel as any)._$messageStream; + + return { channel, factory: factory!, incomming }; +} + +// --- Tests --- + +describe(`${RpcChannel.name} (callee cancellation)`, () => { + it('exposes a live token per incoming call and cancels it on a Cancel frame', async () => { + const { factory, incomming } = await setup(); + + factory.last!.observer.next(requestFrame('1', 'IProbe', 'Work', [])); + + expect(incomming).to.have.lengthOf(1); + const call = incomming[0]; + expect(call.ct.isCancellationRequested).to.equal(false); + + factory.last!.observer.next(cancelFrame('1')); + + expect(call.ct.isCancellationRequested).to.equal(true); + }); + + it('cancels only the call whose id matches', async () => { + const { factory, incomming } = await setup(); + + factory.last!.observer.next(requestFrame('1', 'IProbe', 'Work', [])); + factory.last!.observer.next(requestFrame('2', 'IProbe', 'Work', [])); + + factory.last!.observer.next(cancelFrame('2')); + + expect(incomming[0].ct.isCancellationRequested).to.equal(false); + expect(incomming[1].ct.isCancellationRequested).to.equal(true); + }); + + it('ignores a Cancel frame that arrives after the call completed', async () => { + const { factory, incomming } = await setup(); + + factory.last!.observer.next(requestFrame('3', 'IProbe', 'Work', [])); + const call = incomming[0]; + + await call.respond(new RpcMessage.Response('3', JSON.stringify(null), null)); + + expect(() => factory.last!.observer.next(cancelFrame('3'))).to.not.throw(); + expect(call.ct.isCancellationRequested).to.equal(false); + }); + + it('does not throw on a Cancel frame for an unknown request id', async () => { + const { factory } = await setup(); + + expect(() => factory.last!.observer.next(cancelFrame('unknown'))).to.not.throw(); + }); + + it('cancels in-flight calls when the channel is disposed', async () => { + const { channel, factory, incomming } = await setup(); + + factory.last!.observer.next(requestFrame('4', 'IProbe', 'Work', [])); + const call = incomming[0]; + + await channel.disposeAsync(); + + expect(call.ct.isCancellationRequested).to.equal(true); + }); +}); From bae3b00635d59c8f0898f0dc7a8e827f6576cc2a Mon Sep 17 00:00:00 2001 From: Eduard Dumitru Date: Wed, 1 Jul 2026 15:17:20 +0200 Subject: [PATCH 5/8] TypeScript: propagate cancellation to the peer (caller sends cancel frames) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes the other half of the cancellation parity gap. A fired CancellationToken used to only reject the local awaiting promise; the peer never learned of it and ran the operation to completion. RpcChannel.call now sends a CancellationRequest frame when the token fires, mirroring .NET's Connection.RemoteCall — so the remote actually stops. - registerOutgoingCancellation arms a token registration (before sending, like .NET) that fire-and-forgets a CancellationRequest for the request id; it is disposed when the call settles, so it never fires for a completed call nor retains the token. Guarded by canBeCanceled, so CancellationToken.none is a no-op and untimed/untokened calls are unchanged. Two issues found by an adversarial review of the above and fixed here: - A token ALREADY cancelled at call time synchronously fires the registration, which (with no send lock) enqueued the cancel frame before the request frame — the peer dropped the orphan cancel and ran the request uncancelled. Fixed by suppressing the request entirely when ct.isCancellationRequested, matching .NET (a call abandoned before it is sent never reaches the wire). For a not-yet- cancelled token, register() only stores the callback, so any later cancel is necessarily enqueued after the request. - disposeAsync never settled pending outgoing calls, so a call parked at `await promise` under the default infinite timeout hung forever after the channel died, and the (channel-capturing) cancellation registration leaked on the caller's token. Added OutgoingCallTable.completeAll (mirroring .NET's Connection.CompleteRequests) to fault pending calls with an ObjectDisposedError, which unblocks the await, runs call()'s finally, and releases the registration. A no-throw guard on the pending-call promise prevents an unhandled rejection when a call is cancelled/disposed while still connecting. Adds unit tests (RpcChannelSendCancellation): fired-token propagation with Request-before-Cancel ordering (incl. cancel-during-connect), already-cancelled suppression, non-cancelable no-op, no-cancel-after-completion, and dispose-settles-and-releases. Updates LIMITATIONS.md. 387 std specs pass; the implementation was verified by two rounds of independent adversarial review. Co-Authored-By: Claude Opus 4.8 (1M context) --- LIMITATIONS.md | 2 +- .../std/core/Protocol/Rpc/RpcCallContext.ts | 5 + .../src/std/core/Protocol/Rpc/RpcChannel.ts | 69 ++++- .../core/RpcChannelSendCancellation.test.ts | 283 ++++++++++++++++++ 4 files changed, 356 insertions(+), 3 deletions(-) create mode 100644 src/Clients/js/test/std/core/RpcChannelSendCancellation.test.ts diff --git a/LIMITATIONS.md b/LIMITATIONS.md index 5eab3d72..9b4eba65 100644 --- a/LIMITATIONS.md +++ b/LIMITATIONS.md @@ -34,7 +34,7 @@ Contracts are a **shared agreement** between both peers, and every client *trust - **Propagation to the peer differs:** - **.NET** — a fired token sends a `CancellationRequest`. - **Python** — sends one **only for methods marked `@ipc_cancellable`** (otherwise the cancel is local-only); a hosted handler honors an inbound cancel only when its contract method is `@ipc_cancellable`. - - **TypeScript** — as a **caller**, still **does not send a cancel frame** for its own outbound calls (local-only; the remote keeps running) — the remaining parity gap. As a **callee** (a hosted callback), it now **honors an inbound cancel**: the running handler's per-call token fires, and — when the callback contract declares a trailing `CancellationToken`/`AbortSignal` — the live token (or a bridged `AbortSignal`) is injected so the handler can observe it. A callback registered by bare endpoint name carries no parameter-type metadata, so its arguments are left untouched (the cancel still fires the token, but nothing is injected into the handler). + - **TypeScript** — as a **caller**, a fired token now sends a `CancellationRequest` to the peer (like .NET), so the remote actually stops; a token already cancelled *before* the call suppresses the request entirely (the remote never runs it), also mirroring .NET. As a **callee** (a hosted callback), it **honors an inbound cancel**: the running handler's per-call token fires, and — when the callback contract declares a trailing `CancellationToken`/`AbortSignal` — the live token (or a bridged `AbortSignal`) is injected so the handler can observe it. A callback registered by bare endpoint name carries no parameter-type metadata, so its arguments are left untouched (the cancel still fires the token, but nothing is injected into the handler). - A successful response that arrives before the cancel is delivered — the result wins. ## Transports & features diff --git a/src/Clients/js/src/std/core/Protocol/Rpc/RpcCallContext.ts b/src/Clients/js/src/std/core/Protocol/Rpc/RpcCallContext.ts index 0f410881..8eb14feb 100644 --- a/src/Clients/js/src/std/core/Protocol/Rpc/RpcCallContext.ts +++ b/src/Clients/js/src/std/core/Protocol/Rpc/RpcCallContext.ts @@ -35,5 +35,10 @@ export module RpcCallContext { public complete(response: RpcMessage.Response): void { const _ = this._pcs.trySetResult(response); } + + /** Faults the pending call — used when the channel is torn down. */ + public fail(error: Error): void { + const _ = this._pcs.trySetFaulted(error); + } } } diff --git a/src/Clients/js/src/std/core/Protocol/Rpc/RpcChannel.ts b/src/Clients/js/src/std/core/Protocol/Rpc/RpcChannel.ts index e106003f..7c61b36f 100644 --- a/src/Clients/js/src/std/core/Protocol/Rpc/RpcChannel.ts +++ b/src/Clients/js/src/std/core/Protocol/Rpc/RpcChannel.ts @@ -1,9 +1,11 @@ import { Observer } from 'rxjs'; import { CancellationToken, + CancellationTokenRegistration, TimeSpan, Trace, ArgumentOutOfRangeError, + ObjectDisposedError, SocketStream, UnknownError, PromisePal, @@ -40,6 +42,18 @@ export class RpcChannel implements IRpcChannel { this._map.get(response.RequestId)?.complete(response); } + // Faults every in-flight call (mirroring .NET's Connection.CompleteRequests) + // so a call parked at `await promise` unblocks when the channel dies — + // rather than hanging forever under an infinite request timeout — which in + // turn lets RpcChannel.call's finally dispose its cancellation registration. + // Snapshot first: failing a call rejects its promise, whose own finally + // deletes the map entry. + public completeAll(error: Error): void { + for (const context of [...this._map.values()]) { + context.fail(error); + } + } + private generateId(): string { return `${this._sequence++}`; } @@ -89,6 +103,9 @@ export class RpcChannel implements IRpcChannel { if (!this._isDisposed) { this._isDisposed = true; this._incomingCalls.clear(); + this._outgoingCalls.completeAll( + new ObjectDisposedError('RpcChannel', 'The RPC channel was disposed.'), + ); try { const messageStream = await this._$messageStream; await messageStream.disposeAsync(); @@ -102,8 +119,56 @@ export class RpcChannel implements IRpcChannel { ct: CancellationToken, ): Promise { const promise = this._outgoingCalls.register(request, timeout, ct); - await (await this._$messageStream).writeMessageAsync(request.toNetwork(), ct); - return await promise; + // `promise` is awaited below, but the send can park (e.g. on the initial + // connect) while the token fires or the channel is disposed — faulting it + // before it is awaited. Attach a no-throw handler now so that is never an + // unhandled rejection; the awaits below still observe the real outcome. + Trace.traceErrorNoThrow(promise); + // A token already cancelled at call time has synchronously rejected + // `promise` (via ct.bind in RpcCallContext.Outgoing). Mirror .NET and do + // NOT put the request on the wire: the peer must never run a call the + // caller has already abandoned. (This also avoids arming the outgoing + // cancellation, whose synchronous fire would otherwise race a cancel + // frame ahead of the request frame.) + if (ct.isCancellationRequested) { + return await promise; + } + // Arm before sending — mirroring .NET's Connection.RemoteCall — so a + // fired token propagates a CancellationRequest to the peer, not just a + // local rejection. Because the token is not yet cancelled here, register() + // only stores the callback (no synchronous fire), so any later cancel + // frame is necessarily enqueued after the request frame. Disposed once the + // call settles so it can't fire for a completed request or retain the token. + const cancellationRegistration = this.registerOutgoingCancellation(request.Id, ct); + try { + await (await this._$messageStream).writeMessageAsync(request.toNetwork(), ct); + return await promise; + } finally { + cancellationRegistration?.dispose(); + } + } + + private registerOutgoingCancellation( + requestId: string, + ct: CancellationToken, + ): CancellationTokenRegistration | undefined { + if (!ct.canBeCanceled) { + return undefined; + } + return ct.register(() => this.sendCancellationRequest(requestId)); + } + + private sendCancellationRequest(requestId: string): void { + const frame = new RpcMessage.CancellationRequest(requestId).toNetwork(); + // Fire-and-forget: the callback that triggers this is synchronous, and the + // peer treats a cancel for an unknown/finished request as a no-op. Errors + // (e.g. a torn-down connection) are traced, never thrown. + Trace.traceErrorNoThrow( + (async () => { + const messageStream = await this._$messageStream; + await messageStream.writeMessageAsync(frame, CancellationToken.none); + })(), + ); } public get isDisposed(): boolean { diff --git a/src/Clients/js/test/std/core/RpcChannelSendCancellation.test.ts b/src/Clients/js/test/std/core/RpcChannelSendCancellation.test.ts new file mode 100644 index 00000000..927258cd --- /dev/null +++ b/src/Clients/js/test/std/core/RpcChannelSendCancellation.test.ts @@ -0,0 +1,283 @@ +import { Observer, Subject } from 'rxjs'; + +import { + Address, + CancellationToken, + CancellationTokenSource, + ConnectHelper, + IMessageStream, + Network, + ObjectDisposedError, + OperationCanceledError, + RpcCallContext, + RpcChannel, + RpcMessage, + Socket, + Timeout, + TimeSpan, +} from '../../../src/std'; + +import { expect } from 'chai'; + +// --- Test doubles --- + +class MockSocket extends Socket { + private readonly _data = new Subject(); + get $data() { + return this._data; + } + async write(_buffer: Buffer, _ct: CancellationToken): Promise {} + dispose(): void {} +} + +class MockAddress extends Address { + get key() { + return 'mock:rpc-send-cancellation'; + } + async connect(_h: ConnectHelper, _t: TimeSpan, _ct: CancellationToken): Promise { + return new MockSocket(); + } +} + +// A connect that the test resolves manually, to exercise a token firing while the +// very first call is still connecting (the message stream not yet created). +class DeferredMockAddress extends Address { + public resolveConnect!: () => void; + private readonly _connected = new Promise((resolve) => { + this.resolveConnect = () => resolve(new MockSocket()); + }); + get key() { + return 'mock:rpc-send-cancellation-deferred'; + } + async connect(_h: ConnectHelper, _t: TimeSpan, _ct: CancellationToken): Promise { + return this._connected; + } +} + +// Records everything the channel writes and exposes the inbound observer so the +// test can deliver a Response frame to settle an outgoing call. +class StubMessageStream implements IMessageStream { + public readonly writes: Network.Message[] = []; + constructor(public readonly observer: Observer) {} + async writeMessageAsync(message: Network.Message, _ct: CancellationToken): Promise { + this.writes.push(message); + } + async disposeAsync(): Promise {} +} + +class StubMessageStreamFactory implements IMessageStream.Factory { + public last: StubMessageStream | undefined; + create(_stream: any, observer: Observer): IMessageStream { + return (this.last = new StubMessageStream(observer)); + } +} + +const tick = () => new Promise((resolve) => setTimeout(resolve, 5)); + +function typesOf(writes: Network.Message[]): string[] { + return writes.map((w) => Network.Message.Type[w.type]); +} + +function firstOfType( + writes: Network.Message[], + type: Network.Message.Type, +): Network.Message | undefined { + return writes.find((w) => w.type === type); +} + +async function setup() { + const rpcObserver: Observer = { + next() {}, + error() {}, + complete() {}, + }; + + const factory = new StubMessageStreamFactory(); + const channel = RpcChannel.create( + new MockAddress(), + {} as any as ConnectHelper, + TimeSpan.fromSeconds(30), + CancellationToken.none, + rpcObserver, + factory, + ); + + await (channel as any)._$messageStream; + + return { channel, factory: factory! }; +} + +function settle(promise: Promise) { + return promise.then( + (response) => ({ ok: true as const, response }), + (err: unknown) => ({ ok: false as const, err }), + ); +} + +function responseFrame(requestId: string): Network.Message { + return new RpcMessage.Response(requestId, JSON.stringify(null), null).toNetwork(); +} + +// --- Tests --- + +describe(`${RpcChannel.name} (caller sends cancel frames)`, () => { + it('propagates a CancellationRequest frame to the peer when the token fires', async () => { + const { channel, factory } = await setup(); + const cts = new CancellationTokenSource(); + + const settled = settle( + channel.call(new RpcMessage.Request(0, 'ISvc', 'Work', []), Timeout.infiniteTimeSpan, cts.token), + ); + + await tick(); + const requestWrite = firstOfType(factory.last!.writes, Network.Message.Type.Request); + expect(requestWrite, `writes were ${typesOf(factory.last!.writes)}`).to.not.equal(undefined); + const requestId = RpcMessage.Request.fromNetwork(requestWrite!).Id; + + // Nothing cancelled yet. + expect(firstOfType(factory.last!.writes, Network.Message.Type.Cancel)).to.equal(undefined); + + cts.cancel(); + await tick(); + + const cancelWrite = firstOfType(factory.last!.writes, Network.Message.Type.Cancel); + expect(cancelWrite, `writes were ${typesOf(factory.last!.writes)}`).to.not.equal(undefined); + expect(RpcMessage.CancellationRequest.fromNetwork(cancelWrite!).RequestId).to.equal(requestId); + + // The cancel frame must never precede its request on the wire. + expect(typesOf(factory.last!.writes)).to.deep.equal(['Request', 'Cancel']); + + // The caller's own promise still rejects with a cancellation. + const outcome = await settled; + expect(outcome.ok).to.equal(false); + expect((outcome as any).err).to.be.instanceOf(OperationCanceledError); + }); + + it('does not send a cancel frame after the call has already completed', async () => { + const { channel, factory } = await setup(); + const cts = new CancellationTokenSource(); + + const settled = settle( + channel.call(new RpcMessage.Request(0, 'ISvc', 'Work', []), Timeout.infiniteTimeSpan, cts.token), + ); + + await tick(); + const requestId = RpcMessage.Request.fromNetwork( + firstOfType(factory.last!.writes, Network.Message.Type.Request)!, + ).Id; + + // Complete the call, then cancel: the (disposed) registration must not fire. + factory.last!.observer.next(responseFrame(requestId)); + const outcome = await settled; + expect(outcome.ok).to.equal(true); + + cts.cancel(); + await tick(); + + expect(firstOfType(factory.last!.writes, Network.Message.Type.Cancel)).to.equal(undefined); + }); + + it('never registers cancellation for a non-cancelable token', async () => { + const { channel, factory } = await setup(); + + const settled = settle( + channel.call( + new RpcMessage.Request(0, 'ISvc', 'Work', []), + Timeout.infiniteTimeSpan, + CancellationToken.none, + ), + ); + + await tick(); + const requestId = RpcMessage.Request.fromNetwork( + firstOfType(factory.last!.writes, Network.Message.Type.Request)!, + ).Id; + expect(firstOfType(factory.last!.writes, Network.Message.Type.Cancel)).to.equal(undefined); + + // Settle it so nothing is left pending. + factory.last!.observer.next(responseFrame(requestId)); + expect((await settled).ok).to.equal(true); + }); + + it('suppresses the request entirely when the token is already cancelled at call time', async () => { + const { channel, factory } = await setup(); + const cts = new CancellationTokenSource(); + cts.cancel(); + + const settled = settle( + channel.call(new RpcMessage.Request(0, 'ISvc', 'Work', []), Timeout.infiniteTimeSpan, cts.token), + ); + + await tick(); + + // Mirrors .NET: a call abandoned before it is sent never reaches the wire — + // no request, and hence no (pointless) cancel for a request the peer never saw. + expect(factory.last!.writes, `writes were ${typesOf(factory.last!.writes)}`).to.have.lengthOf(0); + + const outcome = await settled; + expect(outcome.ok).to.equal(false); + expect((outcome as any).err).to.be.instanceOf(OperationCanceledError); + }); + + it('settles an in-flight call when the channel is disposed, and disposes its cancellation registration', async () => { + const { channel, factory } = await setup(); + const cts = new CancellationTokenSource(); + + const settled = settle( + channel.call(new RpcMessage.Request(0, 'ISvc', 'Work', []), Timeout.infiniteTimeSpan, cts.token), + ); + await tick(); + + await channel.disposeAsync(); + + // The call is faulted (not left hanging under the infinite timeout). + const outcome = await settled; + expect(outcome.ok).to.equal(false); + expect((outcome as any).err).to.be.instanceOf(ObjectDisposedError); + + // Settling ran call()'s finally, disposing the registration — so firing the + // token afterwards emits nothing (no retained channel-capturing closure). + const writesAfterDispose = factory.last!.writes.length; + cts.cancel(); + await tick(); + expect(factory.last!.writes.length).to.equal(writesAfterDispose); + }); + + it('keeps Request before Cancel even when the token fires while the first call is still connecting', async () => { + const address = new DeferredMockAddress(); + const factory = new StubMessageStreamFactory(); + const channel = RpcChannel.create( + address, + {} as any as ConnectHelper, + TimeSpan.fromSeconds(30), + CancellationToken.none, + { next() {}, error() {}, complete() {} } as Observer, + factory, + ); + + // Do NOT await the message stream: the connect is still pending. + const cts = new CancellationTokenSource(); + const settled = settle( + channel.call(new RpcMessage.Request(0, 'ISvc', 'Work', []), Timeout.infiniteTimeSpan, cts.token), + ); + + await tick(); + // Nothing on the wire yet — the stream does not exist until connect resolves. + expect(factory.last).to.equal(undefined); + + // Fire the token mid-connect: the cancel's write is queued behind the request's. + cts.cancel(); + await tick(); + expect(factory.last).to.equal(undefined); + + // Now let the connection complete; both queued writes flush in order. + address.resolveConnect(); + await tick(); + + expect(typesOf(factory.last!.writes)).to.deep.equal(['Request', 'Cancel']); + + const outcome = await settled; + expect(outcome.ok).to.equal(false); + expect((outcome as any).err).to.be.instanceOf(OperationCanceledError); + }); +}); From 2689c011239a94a0172669591d3fafac103c0a9f Mon Sep 17 00:00:00 2001 From: Eduard Dumitru Date: Wed, 1 Jul 2026 21:15:48 +0200 Subject: [PATCH 6/8] TypeScript: fix serialization of a live CancellationToken proxy argument MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Passing a real CancellationToken (or an AbortSignal) to a proxy method threw `TypeError: Converting circular structure to JSON` before anything was sent: the argument serializer (Converter) ran JSON.stringify over the raw token, and a live CancellationTokenSource token is a circular graph (token -> source -> token). CancellationToken.none serializes to `{}`, so this was never hit — no existing test passed a live token as a proxy argument — but it made caller-side cancellation unusable in practice (you could not even issue the call). RpcRequestFactory now keeps the live token for cancellation (local binding + the CancellationRequest frame) but writes an inert placeholder (CancellationToken.none -> `{}`) into the wire slot. The cancellation signal is out-of-band and the receiver ignores the slot's content for a CancellationToken parameter, so this is purely a serialization fix — no behavioural change to a call that never cancels. This is a pre-existing bug (present on master), independent of the recent callee/caller cancellation work; it was surfaced by wiring up a real end-to-end cancellation call. Adds unit tests for a live CancellationToken, an AbortSignal, and the no-argument case. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../src/std/core/Proxies/RpcRequestFactory.ts | 13 ++- .../RpcRequestFactoryCancellation.test.ts | 107 ++++++++++++++++++ 2 files changed, 115 insertions(+), 5 deletions(-) create mode 100644 src/Clients/js/test/std/core/RpcRequestFactoryCancellation.test.ts diff --git a/src/Clients/js/src/std/core/Proxies/RpcRequestFactory.ts b/src/Clients/js/src/std/core/Proxies/RpcRequestFactory.ts index d37cb6a9..a2818006 100644 --- a/src/Clients/js/src/std/core/Proxies/RpcRequestFactory.ts +++ b/src/Clients/js/src/std/core/Proxies/RpcRequestFactory.ts @@ -52,14 +52,17 @@ export class RpcRequestFactory { continue; } // Accept a CancellationToken OR (bridged) an AbortSignal wherever a - // cancellation argument is expected, and substitute it in-place so - // the wire form AND the ending-CancellationToken handling below are - // identical for both. For a real CancellationToken this is a no-op - // substitution. Additive — CancellationToken still works unchanged. + // cancellation argument is expected. Capture the live token for + // cancellation (local binding + the CancellationRequest frame), but + // put an INERT placeholder in the wire slot: the cancellation signal + // is out-of-band, and the receiver ignores the slot's content for a + // CancellationToken parameter. A live CancellationTokenSource token is + // a circular object graph (token -> source -> token) that JSON cannot + // serialize, so the raw token must never reach Converter. const token = AbortSignalAdapter.ensureCancellationToken(arg); if (token) { ct = token; - args[i] = token; + args[i] = CancellationToken.none; } } diff --git a/src/Clients/js/test/std/core/RpcRequestFactoryCancellation.test.ts b/src/Clients/js/test/std/core/RpcRequestFactoryCancellation.test.ts new file mode 100644 index 00000000..47ff2bcd --- /dev/null +++ b/src/Clients/js/test/std/core/RpcRequestFactoryCancellation.test.ts @@ -0,0 +1,107 @@ +import { + Address, + CancellationToken, + CancellationTokenSource, + ConfigStore, + ConnectHelper, + ContractStore, + RpcRequestFactory, + Socket, + TimeSpan, +} from '../../../src/std'; + +import { NodeAddressBuilder } from '../../../src/node/NodeAddressBuilder'; +import { MockServiceProvider } from '../../infrastructure'; + +import { expect } from 'chai'; + +class MockAddress extends Address { + get key() { + return 'mock:rpc-request-factory'; + } + async connect(_h: ConnectHelper, _t: TimeSpan, _ct: CancellationToken): Promise { + throw new Error('unused'); + } +} + +class Contract { + Work(_ct: CancellationToken): Promise { + throw void 0; + } +} + +function serviceProvider() { + const contractStore = new ContractStore(); + return new MockServiceProvider({ + implementation: { + contractStore, + configStore: new ConfigStore( + new MockServiceProvider({ + implementation: { contractStore: new ContractStore() }, + }), + ), + }, + }); +} + +describe('RpcRequestFactory (cancellation argument)', () => { + it('serializes a live CancellationToken to an inert placeholder (no circular JSON)', () => { + const sp = serviceProvider(); + const cts = new CancellationTokenSource(); + + // A live token is a circular object graph (token -> source -> token). + // Serializing it raw would throw "Converting circular structure to JSON". + expect(() => JSON.stringify(cts.token)).to.throw(); + + const [request, , ct] = RpcRequestFactory.create({ + sp, + service: Contract, + address: new MockAddress(), + methodName: 'Work', + args: [cts.token], + }); + + // The request must have been built without throwing, and every wire slot + // must be plain, parseable JSON — the token was replaced by a placeholder. + expect(() => request.Parameters.map((p) => JSON.parse(p))).to.not.throw(); + expect(JSON.parse(request.Parameters[0])).to.deep.equal({}); + + // ...yet the live token is still returned, so the caller can bind local + // cancellation and emit a CancellationRequest frame when it fires. + expect(ct).to.equal(cts.token); + }); + + it('adapts and serializes an AbortSignal argument the same way', () => { + const sp = serviceProvider(); + const controller = new AbortController(); + + const [request, , ct] = RpcRequestFactory.create({ + sp, + service: Contract, + address: new MockAddress(), + methodName: 'Work', + args: [controller.signal], + }); + + expect(() => request.Parameters.map((p) => JSON.parse(p))).to.not.throw(); + expect(ct).to.be.instanceOf(CancellationToken); + expect(ct.isCancellationRequested).to.equal(false); + + controller.abort(); + expect(ct.isCancellationRequested).to.equal(true); + }); + + it('uses CancellationToken.none (no wire arg detected) when none is passed', () => { + const sp = serviceProvider(); + + const [, , ct] = RpcRequestFactory.create({ + sp, + service: Contract, + address: new MockAddress(), + methodName: 'Work', + args: [], + }); + + expect(ct).to.equal(CancellationToken.none); + }); +}); From e8d351f5669117d9528548c7c61855c58ac9f514 Mon Sep 17 00:00:00 2001 From: Eduard Dumitru Date: Wed, 1 Jul 2026 21:16:00 +0200 Subject: [PATCH 7/8] Test: no-mock TS<->.NET end-to-end caller-cancellation coverage MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds the real-connection test that the mock-based unit tests could not provide: the TypeScript client (over a live NamedPipe/WebSocket connection to the real .NET NodeInterop server) cancels an in-flight call, and we verify the .NET handler actually observes it. - .NET server (Contracts + ServiceImpls): IAlgebra gains WaitForCancellation(ct), which parks on its injected CancellationToken until cancelled, and CancellationCount(), which reports how many cancellations it has observed. The single-parameter WaitForCancellation aligns the ct at position 0, so a TS `WaitForCancellation(cts.token)` call maps cleanly (the server injects the per-request token; the wire slot is ignored). - TS contract (test IAlgebra) mirrors the two methods. - end-to-end.test.ts: calls WaitForCancellation(token), cancels, asserts the caller's promise rejects locally with OperationCanceledError AND — the point — polls CancellationCount until the server-side count increments, which only happens if the client transmits a CancellationRequest frame. Runs for both the WebSocket and NamedPipe transports (delta-asserted against the shared server). This test only passes with the caller-side cancel-frame sending and the CancellationToken-argument serialization fix; verified green against the real .NET server over a named pipe. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../UiPath.CoreIpc.NodeInterop/Contracts.cs | 5 ++ .../ServiceImpls.cs | 19 +++++++ .../js/test/node/Contracts/IAlgebra.ts | 10 +++- src/Clients/js/test/node/end-to-end.test.ts | 55 ++++++++++++++++++- 4 files changed, 87 insertions(+), 2 deletions(-) diff --git a/src/Clients/js/dotnet/UiPath.CoreIpc.NodeInterop/Contracts.cs b/src/Clients/js/dotnet/UiPath.CoreIpc.NodeInterop/Contracts.cs index d86c0d8e..4550b643 100644 --- a/src/Clients/js/dotnet/UiPath.CoreIpc.NodeInterop/Contracts.cs +++ b/src/Clients/js/dotnet/UiPath.CoreIpc.NodeInterop/Contracts.cs @@ -21,6 +21,11 @@ public interface IAlgebra Task Timeout(); Task Echo(int x); Task TestMessage(Message message); + // Blocks until its (injected) CancellationToken fires, then throws — used + // to prove a client actually transmits a cancellation to the server. + Task WaitForCancellation(CancellationToken ct = default); + // How many times WaitForCancellation has observed a cancellation so far. + Task CancellationCount(); } public interface ICalculus diff --git a/src/Clients/js/dotnet/UiPath.CoreIpc.NodeInterop/ServiceImpls.cs b/src/Clients/js/dotnet/UiPath.CoreIpc.NodeInterop/ServiceImpls.cs index e6de1f24..b68d6ef6 100644 --- a/src/Clients/js/dotnet/UiPath.CoreIpc.NodeInterop/ServiceImpls.cs +++ b/src/Clients/js/dotnet/UiPath.CoreIpc.NodeInterop/ServiceImpls.cs @@ -45,6 +45,25 @@ public async Task Sleep(int milliseconds, Message message = default!, Canc public Task Timeout() => Task.FromException(new TimeoutException()); public Task Echo(int x) => Task.FromResult(x); + + private int _cancellationObservationCount; + + public async Task WaitForCancellation(CancellationToken ct = default) + { + try + { + await Task.Delay(System.Threading.Timeout.InfiniteTimeSpan, ct); + return false; // unreachable: only completes by cancellation + } + catch (OperationCanceledException) + { + Interlocked.Increment(ref _cancellationObservationCount); + throw; + } + } + + public Task CancellationCount() => + Task.FromResult(Volatile.Read(ref _cancellationObservationCount)); } public class Calculus : ICalculus diff --git a/src/Clients/js/test/node/Contracts/IAlgebra.ts b/src/Clients/js/test/node/Contracts/IAlgebra.ts index 85baa76a..171f6491 100644 --- a/src/Clients/js/test/node/Contracts/IAlgebra.ts +++ b/src/Clients/js/test/node/Contracts/IAlgebra.ts @@ -1,4 +1,4 @@ -import { Message } from '../../../src/std'; +import { CancellationToken, Message } from '../../../src/std'; export class IAlgebra { public MultiplySimple(x: number, y: number): Promise { @@ -12,4 +12,12 @@ export class IAlgebra { public TestMessage(message: Message): Promise { throw void 0; } + + public WaitForCancellation(ct: CancellationToken): Promise { + throw void 0; + } + + public CancellationCount(): Promise { + throw void 0; + } } diff --git a/src/Clients/js/test/node/end-to-end.test.ts b/src/Clients/js/test/node/end-to-end.test.ts index 7443f802..01974de5 100644 --- a/src/Clients/js/test/node/end-to-end.test.ts +++ b/src/Clients/js/test/node/end-to-end.test.ts @@ -1,9 +1,29 @@ import { expect } from 'chai'; -import { Message, PromisePal } from '../../src/std'; +import { + CancellationTokenSource, + Message, + OperationCanceledError, + PromisePal, +} from '../../src/std'; import { ipc } from '../../src/node'; import { IAlgebra, IArithmetic } from './Contracts'; import { AddressHelper as TestContext } from './Fixtures'; +async function waitFor( + predicate: () => Promise, + timeoutMs: number, + intervalMs = 100, +): Promise { + const deadline = Date.now() + timeoutMs; + do { + if (await predicate()) { + return true; + } + await PromisePal.delay(intervalMs); + } while (Date.now() < deadline); + return false; +} + describe('node:end-to-end', () => { for (const context of [TestContext.WebSocket, TestContext.NamedPipe]) { @@ -84,6 +104,39 @@ describe('node:end-to-end', () => { expect(receivedMessage?.Payload).to.equal(originalMessage.Payload); }); + it('propagates caller-side cancellation to the .NET callee', async () => { + // The .NET Algebra server is a shared singleton across the + // WebSocket/NamedPipe runs, so assert a delta rather than an + // absolute count. + const before = await algebraProxy.CancellationCount(); + + const cts = new CancellationTokenSource(); + const local = algebraProxy + .WaitForCancellation(cts.token) + .then(() => 'resolved' as const, (err: unknown) => err); + + // Let the request reach the server and park on its token. + await PromisePal.delay(100); + + cts.cancel(); + + // The caller's own promise rejects locally with a cancellation... + const outcome = await local; + expect(outcome).to.be.instanceOf(OperationCanceledError); + + // ...and the .NET handler's injected CancellationToken fires — which + // can only happen if the TS client actually transmitted a + // CancellationRequest frame to the server. + const serverObserved = await waitFor( + async () => (await algebraProxy.CancellationCount()) === before + 1, + 10_000, + ); + expect( + serverObserved, + 'the .NET server never observed the cancellation', + ).to.equal(true); + }, 30_000); + }); } }); From 45976a59c2c3deed154c0a036ddc618b784ccbfd Mon Sep 17 00:00:00 2001 From: Eduard Dumitru Date: Thu, 2 Jul 2026 10:34:20 +0200 Subject: [PATCH 8/8] Test: no-mock TS<->.NET cancellation interchangeability, both directions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extends the real-connection e2e suite so every new TypeScript cancellation feature is exercised against the live .NET NodeInterop server, and shows that a TypeScript contract may use a CancellationToken or an AbortSignal interchangeably against the same C# CancellationToken counterpart — as caller AND as callee. Caller (TS -> .NET service): - WaitForCancellation now accepts CancellationToken | AbortSignal. A new case passes an AbortSignal (bridged to a CancellationToken on the wire) alongside the existing CancellationToken case; both make the .NET handler observe the cancel. Callee (.NET service -> TS-hosted callback): - New .NET ICancellationCallback { Wait(ct) } plus IAlgebra.CancelCallback, which invokes the callback on the client and then cancels it. - Two new cases register a TS handler whose Wait parameter is a CancellationToken in one and an AbortSignal in the other (contract metadata stamped, mirroring emitDecoratorMetadata); each asserts the handler's injected token/signal fires when the .NET server cancels, and that the server sees the callback complete. Verified green for all four (caller/callee x CancellationToken/AbortSignal) against the real .NET server over a named pipe. Unit suite: 390 specs, 0 failures. Note: callee-side injection is metadata-driven, so it requires a callback contract whose parameter types are available at runtime (emitDecoratorMetadata or an equivalent stamp); a bare-endpoint callback receives no injected cancellation. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../UiPath.CoreIpc.NodeInterop/Contracts.cs | 12 ++ .../ServiceImpls.cs | 18 +++ .../js/test/node/Contracts/IAlgebra.ts | 10 +- src/Clients/js/test/node/end-to-end.test.ts | 120 ++++++++++++++++++ 4 files changed, 159 insertions(+), 1 deletion(-) diff --git a/src/Clients/js/dotnet/UiPath.CoreIpc.NodeInterop/Contracts.cs b/src/Clients/js/dotnet/UiPath.CoreIpc.NodeInterop/Contracts.cs index 4550b643..3431f820 100644 --- a/src/Clients/js/dotnet/UiPath.CoreIpc.NodeInterop/Contracts.cs +++ b/src/Clients/js/dotnet/UiPath.CoreIpc.NodeInterop/Contracts.cs @@ -12,6 +12,14 @@ public interface IArithmetic Task SendMessage(Message message); } + // A callback the server invokes on the client; the client handler is expected + // to observe the cancellation the server triggers (the TypeScript side may + // receive it as a CancellationToken or, interchangeably, an AbortSignal). + public interface ICancellationCallback + { + Task Wait(CancellationToken ct = default); + } + public interface IAlgebra { Task Ping(); @@ -26,6 +34,10 @@ public interface IAlgebra Task WaitForCancellation(CancellationToken ct = default); // How many times WaitForCancellation has observed a cancellation so far. Task CancellationCount(); + // Invokes ICancellationCallback.Wait on the client, then cancels it — used + // to prove a client-hosted callback observes a server-initiated cancel. + // Returns what the callback returned (true once it observed the cancel). + Task CancelCallback(Message message = default!); } public interface ICalculus diff --git a/src/Clients/js/dotnet/UiPath.CoreIpc.NodeInterop/ServiceImpls.cs b/src/Clients/js/dotnet/UiPath.CoreIpc.NodeInterop/ServiceImpls.cs index b68d6ef6..64f130be 100644 --- a/src/Clients/js/dotnet/UiPath.CoreIpc.NodeInterop/ServiceImpls.cs +++ b/src/Clients/js/dotnet/UiPath.CoreIpc.NodeInterop/ServiceImpls.cs @@ -64,6 +64,24 @@ public async Task WaitForCancellation(CancellationToken ct = default) public Task CancellationCount() => Task.FromResult(Volatile.Read(ref _cancellationObservationCount)); + + public async Task CancelCallback(Message message = default!) + { + var callback = message.Client.GetCallback(); + using var cts = new CancellationTokenSource(); + var callbackTask = callback.Wait(cts.Token); + cts.CancelAfter(TimeSpan.FromMilliseconds(200)); + try + { + // The client handler resolves true once it observes the cancel. + return await callbackTask; + } + catch (OperationCanceledException) + { + // Or the handler propagated the cancellation — still observed. + return true; + } + } } public class Calculus : ICalculus diff --git a/src/Clients/js/test/node/Contracts/IAlgebra.ts b/src/Clients/js/test/node/Contracts/IAlgebra.ts index 171f6491..12a03578 100644 --- a/src/Clients/js/test/node/Contracts/IAlgebra.ts +++ b/src/Clients/js/test/node/Contracts/IAlgebra.ts @@ -13,11 +13,19 @@ export class IAlgebra { throw void 0; } - public WaitForCancellation(ct: CancellationToken): Promise { + // Accepts a CancellationToken OR an AbortSignal interchangeably; the .NET + // counterpart is a plain CancellationToken. + public WaitForCancellation(cancellation: CancellationToken | AbortSignal): Promise { throw void 0; } public CancellationCount(): Promise { throw void 0; } + + // Asks the .NET server to invoke ICancellationCallback.Wait on us and then + // cancel it. Resolves true once the server sees the callback observe the cancel. + public CancelCallback(): Promise { + throw void 0; + } } diff --git a/src/Clients/js/test/node/end-to-end.test.ts b/src/Clients/js/test/node/end-to-end.test.ts index 01974de5..c4927a39 100644 --- a/src/Clients/js/test/node/end-to-end.test.ts +++ b/src/Clients/js/test/node/end-to-end.test.ts @@ -1,5 +1,7 @@ +import 'reflect-metadata'; import { expect } from 'chai'; import { + CancellationToken, CancellationTokenSource, Message, OperationCanceledError, @@ -24,6 +26,28 @@ async function waitFor( return false; } +// A callback contract the .NET server invokes on this client. We register it in +// the contract store and stamp its parameter metadata (mirroring what a contract +// compiled with emitDecoratorMetadata would emit) so the callee can inject the +// per-call cancellation into the handler — as a CancellationToken or, for the +// same C# CancellationToken counterpart, an AbortSignal. +class ICancellationCallback { + Wait(_cancellation: CancellationToken | AbortSignal): Promise { + throw void 0; + } +} + +function stampCallbackParameter(parameterType: unknown): void { + const contractStore = (ipc as any).contractStore; + contractStore.getOrCreate(ICancellationCallback).operations.getOrCreate('Wait'); + Reflect.defineMetadata( + 'design:paramtypes', + [parameterType], + ICancellationCallback.prototype, + 'Wait', + ); +} + describe('node:end-to-end', () => { for (const context of [TestContext.WebSocket, TestContext.NamedPipe]) { @@ -137,6 +161,102 @@ describe('node:end-to-end', () => { ).to.equal(true); }, 30_000); + it('propagates a caller-side AbortSignal to the .NET callee (interchangeable with CancellationToken)', async () => { + const before = await algebraProxy.CancellationCount(); + + const controller = new AbortController(); + // Same contract method as the CancellationToken case above — here we + // hand it an AbortSignal instead, which the client bridges to a + // CancellationToken on the wire. + const local = algebraProxy + .WaitForCancellation(controller.signal) + .then(() => 'resolved' as const, (err: unknown) => err); + + await PromisePal.delay(100); + controller.abort(); + + const outcome = await local; + expect(outcome).to.be.instanceOf(OperationCanceledError); + + const serverObserved = await waitFor( + async () => (await algebraProxy.CancellationCount()) === before + 1, + 10_000, + ); + expect( + serverObserved, + 'the .NET server never observed the AbortSignal cancellation', + ).to.equal(true); + }, 30_000); + + it('delivers a .NET-initiated callback cancellation to a TS handler as a CancellationToken', async () => { + stampCallbackParameter(CancellationToken); + + let handlerObserved!: () => void; + const observed = new Promise((resolve) => { + handlerObserved = resolve; + }); + + const callback = { + async Wait(ct: CancellationToken): Promise { + return new Promise((resolve) => { + ct.register(() => { + handlerObserved(); + resolve(true); + }); + }); + }, + }; + ipc.callback + .forAddress(context.address) + .forService(ICancellationCallback) + .is(callback); + + const result = await algebraProxy.CancelCallback(); + + // The handler's injected CancellationToken actually fired... + await observed; + // ...and the server saw the callback complete as cancelled. + expect(result).to.equal(true); + }, 30_000); + + it('delivers a .NET-initiated callback cancellation to a TS handler as an AbortSignal (interchangeable with CancellationToken)', async () => { + stampCallbackParameter(AbortSignal); + + let handlerObserved!: () => void; + const observed = new Promise((resolve) => { + handlerObserved = resolve; + }); + + const callback = { + async Wait(signal: AbortSignal): Promise { + return new Promise((resolve) => { + if (signal.aborted) { + handlerObserved(); + resolve(true); + return; + } + signal.addEventListener( + 'abort', + () => { + handlerObserved(); + resolve(true); + }, + { once: true }, + ); + }); + }, + }; + ipc.callback + .forAddress(context.address) + .forService(ICancellationCallback) + .is(callback as any); + + const result = await algebraProxy.CancelCallback(); + + await observed; + expect(result).to.equal(true); + }, 30_000); + }); } });