diff --git a/LIMITATIONS.md b/LIMITATIONS.md index 2a5b3108..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** — **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**, 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/dotnet/UiPath.CoreIpc.NodeInterop/Contracts.cs b/src/Clients/js/dotnet/UiPath.CoreIpc.NodeInterop/Contracts.cs index d86c0d8e..d002e338 100644 --- a/src/Clients/js/dotnet/UiPath.CoreIpc.NodeInterop/Contracts.cs +++ b/src/Clients/js/dotnet/UiPath.CoreIpc.NodeInterop/Contracts.cs @@ -12,6 +12,13 @@ public interface IArithmetic Task SendMessage(Message message); } + // A callback the server invokes then cancels; the client handler must observe it + // (as a CancellationToken or, interchangeably, an AbortSignal). + public interface ICancellationCallback + { + Task Wait(CancellationToken ct = default); + } + public interface IAlgebra { Task Ping(); @@ -21,6 +28,14 @@ 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(); + // Invokes ICancellationCallback.Wait on the client, then cancels it. Returns true + // once the client-hosted handler observed the server-initiated 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 e6de1f24..64f130be 100644 --- a/src/Clients/js/dotnet/UiPath.CoreIpc.NodeInterop/ServiceImpls.cs +++ b/src/Clients/js/dotnet/UiPath.CoreIpc.NodeInterop/ServiceImpls.cs @@ -45,6 +45,43 @@ 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 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/src/std/bcl/cancellation/AbortSignalAdapter.ts b/src/Clients/js/src/std/bcl/cancellation/AbortSignalAdapter.ts new file mode 100644 index 00000000..7fd7a9c6 --- /dev/null +++ b/src/Clients/js/src/std/bcl/cancellation/AbortSignalAdapter.ts @@ -0,0 +1,73 @@ +import { CancellationToken } from './CancellationToken'; +import { CancellationTokenSource } from './CancellationTokenSource'; + +/** + * Bridges a Web/Node `AbortSignal` to a `CancellationToken`, so a contract method may + * accept an `AbortSignal` anywhere a `CancellationToken` is accepted. + */ +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; + } + + /** + * `candidate` unchanged if it is a `CancellationToken`, the bridged token if it is + * an `AbortSignal`, or `undefined` if it is neither cancellation shape. + */ + public static ensureCancellationToken( + candidate: unknown, + ): CancellationToken | undefined { + if (candidate instanceof CancellationToken) { + return candidate; + } + if (AbortSignalAdapter.isAbortSignal(candidate)) { + return AbortSignalAdapter.toCancellationToken(candidate); + } + return undefined; + } + + /** + * A `CancellationToken` cancelled when `signal` aborts — immediately if already + * aborted. The `once` listener self-removes, so a never-aborting signal leaks nothing. + */ + 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; + } + + /** + * The reverse of {@link toCancellationToken}, for handing a callback handler an + * `AbortSignal`. 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/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/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..1aa90b72 100644 --- a/src/Clients/js/src/std/core/Contract/IContractStore.ts +++ b/src/Clients/js/src/std/core/Contract/IContractStore.ts @@ -6,4 +6,7 @@ export interface IContractStore { getOrCreate($class: PublicCtor): ServiceDescriptor; maybeGet($class: PublicCtor): ServiceDescriptor | undefined; + + /** For the callee side, where an incoming call identifies its contract by name. */ + 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..73337ab3 --- /dev/null +++ b/src/Clients/js/src/std/core/Protocol/Rpc/IncomingCallTable.ts @@ -0,0 +1,45 @@ +import { CancellationToken, CancellationTokenSource } from '../../../bcl'; + +/** + * The callee mirror of the outgoing-call table: tracks in-flight incoming calls by + * request id so an inbound `CancellationRequest` can cancel the running handler. + */ +/* @internal */ +export class IncomingCallTable { + private readonly _map = new Map(); + + /** Tracks `requestId`; the returned token fires when a cancellation for it arrives. */ + 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..83eb5eb3 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,8 @@ 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. */ + public readonly ct: CancellationToken = CancellationToken.none, ) {} } @@ -29,5 +31,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 a906501e..df996fb0 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, @@ -11,7 +13,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 { @@ -40,6 +42,14 @@ export class RpcChannel implements IRpcChannel { this._map.get(response.RequestId)?.complete(response); } + // Mirrors .NET's Connection.CompleteRequests, so a call parked at `await promise` + // unblocks when the channel dies. Snapshot first: failing a call deletes its entry. + public completeAll(error: Error): void { + for (const context of [...this._map.values()]) { + context.fail(error); + } + } + private generateId(): string { return `${this._sequence++}`; } @@ -88,6 +98,10 @@ export class RpcChannel implements IRpcChannel { public async disposeAsync(): Promise { 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(); @@ -101,8 +115,45 @@ 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; + // The send can park while `promise` is faulted, before it is awaited; a no-throw + // handler keeps that from being an unhandled rejection. + Trace.traceErrorNoThrow(promise); + // Already cancelled: `promise` has synchronously rejected. Mirror .NET and keep the + // request off the wire — the peer must never run a call the caller has abandoned. + if (ct.isCancellationRequested) { + return await promise; + } + // Arm before sending (as .NET's Connection.RemoteCall does) so a fired token reaches + // the peer. The token isn't cancelled yet, so any cancel frame follows the request. + 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 triggering callback is synchronous, and the peer treats a + // cancel for an unknown request as a no-op. Errors are traced, never thrown. + Trace.traceErrorNoThrow( + (async () => { + const messageStream = await this._$messageStream; + await messageStream.writeMessageAsync(frame, CancellationToken.none); + })(), + ); } public get isDisposed(): boolean { @@ -137,6 +188,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 +241,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..a700f89e 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,34 @@ export class ChannelManager implements IAsyncDisposable { return new RpcMessage.Response(request.Id, data, error); } + + /** + * Replaces the blanked trailing wire argument with the per-call token (or a bridged + * `AbortSignal`). Metadata-driven: guessing could clobber a real empty-string argument. + */ + 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 +155,7 @@ export class ChannelManager implements IAsyncDisposable { incommingContext.respond( await this._channelManager.invokeCallback( incommingContext.request, + incommingContext.ct, ), ); } diff --git a/src/Clients/js/src/std/core/Proxies/RpcRequestFactory.ts b/src/Clients/js/src/std/core/Proxies/RpcRequestFactory.ts index 08c6a9eb..1e8c70e7 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,21 @@ 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. Keep the live token for + // the out-of-band signal, but blank the wire slot: a live token can't be JSON'd. + const token = AbortSignalAdapter.ensureCancellationToken(arg); + if (token) { + ct = token; + args[i] = CancellationToken.none; } } @@ -61,17 +70,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/node/Contracts/IAlgebra.ts b/src/Clients/js/test/node/Contracts/IAlgebra.ts index 85baa76a..12a03578 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,20 @@ export class IAlgebra { public TestMessage(message: Message): Promise { throw void 0; } + + // 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 7443f802..96373a34 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,50 @@ +import 'reflect-metadata'; import { expect } from 'chai'; -import { Message, PromisePal } from '../../src/std'; +import { + CancellationToken, + 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; +} + +// A callback contract the .NET server invokes on this client. Its parameter metadata is +// stamped below (as emitDecoratorMetadata would) so the callee can inject the cancellation. +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]) { @@ -84,6 +125,132 @@ 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 singleton shared across the WebSocket and + // 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 token fires, which can only happen if the TS + // client actually transmitted a CancellationRequest frame. + 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); + + 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 above, handed an AbortSignal instead — which the + // client bridges to a CancellationToken. + 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); + }); } }); 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..c50bcd92 --- /dev/null +++ b/src/Clients/js/test/std/bcl/AbortSignalAdapter.test.ts @@ -0,0 +1,110 @@ +import { AbortSignalAdapter, CancellationToken, CancellationTokenSource } 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); + }); + }); + + 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); + }); +}); 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); + }); +}); 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); + }); +}); 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..662a1ef3 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,9 @@ 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 + # So a POCO handler can reach the peer without a Message parameter. No reset + # needed: this runs in its own asyncio task, which owns its contextvars copy. + 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..25a25e6a --- /dev/null +++ b/src/Clients/python/uipath-ipc/src/uipath_ipc/context.py @@ -0,0 +1,45 @@ +"""Ambient, per-operation IPC context (`IpcContext.Current`) — the Python +counterpart of .NET's `IpcContext`, letting a POCO handler skip `Message`.""" + +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 in one handler task never leaks to another. +_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 call being honored on the current asyncio task, or ``None``.""" + return _current.get() + + +class IpcContext(metaclass=_IpcContextMeta): + """Ambient context of the IPC call being honored, letting a POCO handler reach + the peer without a `Message` parameter. Coexists with `Message` injection.""" + + __slots__ = ("_connection",) + + def __init__(self, connection: "IpcConnection") -> None: + self._connection = connection + + def get_callback(self, contract: type[T]) -> T: + """The ambient equivalent of ``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 + reset needed — the caller's task owns its contextvars copy.""" + _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..80b19eca --- /dev/null +++ b/src/Clients/python/uipath-ipc/tests/test_context.py @@ -0,0 +1,80 @@ +"""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 contract: no `Message` parameter, yet callback-capable via `IpcContext.Current`. +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