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..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(); @@ -21,6 +29,15 @@ 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 — 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 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..f8a65a5c --- /dev/null +++ b/src/Clients/js/src/std/bcl/cancellation/AbortSignalAdapter.ts @@ -0,0 +1,86 @@ +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; + } + + /** + * 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/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..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..8eb14feb 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, ) {} } @@ -29,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 a906501e..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, @@ -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,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++}`; } @@ -88,6 +102,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 +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 { @@ -137,6 +203,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 +256,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/src/std/core/Proxies/RpcRequestFactory.ts b/src/Clients/js/src/std/core/Proxies/RpcRequestFactory.ts index 08c6a9eb..a2818006 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,27 @@ 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. 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] = CancellationToken.none; } } @@ -61,17 +76,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..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,9 +1,53 @@ +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. 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]) { @@ -84,6 +128,135 @@ 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); + + 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); + }); } }); 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..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 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();