addievo
commented
1 year ago All instances of cancel or abort without a reason
RPCClient.ts
@ready(new rpcErrors.ErrorMissingCaller())
public async unaryCaller<I extends JSONValue, O extends JSONValue>(
method: string,
parameters: I,
ctx: Partial<ContextTimedInput> = {},
): Promise<O> {
const callerInterface = await this.duplexStreamCaller<I, O>(method, ctx);
const reader = callerInterface.readable.getReader();
const writer = callerInterface.writable.getWriter();
try {
await writer.write(parameters);
const output = await reader.read();
if (output.done) {
throw new rpcErrors.ErrorMissingCaller('Missing response', {
cause: ctx.signal?.reason,
});
}
await reader.cancel();
await writer.close();
return output.value;
} finally {
// Attempt clean up, ignore errors if already cleaned up
await reader.cancel().catch(() => {});
await writer.close().catch(() => {});
}
}Instance 2
// Hooking up agnostic stream side
let rpcStream: RPCStream<Uint8Array, Uint8Array>;
const streamFactoryProm = this.streamFactory({ signal, timer });
try {
rpcStream = await Promise.race([streamFactoryProm, abortRaceProm.p]);
} catch (e) {
cleanUp();
void streamFactoryProm.then((stream) =>
stream.cancel(ErrorRPCStreamEnded),
);
throw e;
}
void timer.then(
() => {
rpcStream.cancel(
new rpcErrors.ErrorRPCTimedOut('RPC has timed out', {
cause: ctx.signal?.reason,
}),
);
},
() => {}, // Ignore cancellation error
);
// Deciding if we want to allow refreshing
// We want to refresh timer if none was provided
const refreshingTimer: Timer | undefined =
ctx.timer == null ? timer : undefined;
// Composing stream transforms and middleware
const metadata = {
...(rpcStream.meta ?? {}),
command: method,
};
const outputMessageTransformStream =
rpcUtils.clientOutputTransformStream<O>(metadata, refreshingTimer);
const inputMessageTransformStream = rpcUtils.clientInputTransformStream<I>(
method,
refreshingTimer,
);
const middleware = this.middlewareFactory(
{ signal, timer },
rpcStream.cancel,
metadata,
);RPCServer.ts
@ready(new rpcErrors.ErrorRPCHandlerFailed())
public handleStream(rpcStream: RPCStream<Uint8Array, Uint8Array>) {
// This will take a buffer stream of json messages and set up service
// handling for it.
// Constructing the PromiseCancellable for tracking the active stream
const abortController = new AbortController();
// Setting up timeout timer logic
const timer = new Timer({
delay: this.handlerTimeoutTime,
handler: () => {
abortController.abort(new rpcErrors.ErrorRPCTimedOut());
if (this.onTimeoutCallback) {
this.onTimeoutCallback();
}
},
});
const prom = (async () => {
const id = await this.idGen();
const headTransformStream = rpcUtilsMiddleware.binaryToJsonMessageStream(
rpcUtils.parseJSONRPCRequest,
);
// Transparent transform used as a point to cancel the input stream from
const passthroughTransform = new TransformStream<
Uint8Array,
Uint8Array
>();
const inputStream = passthroughTransform.readable;
const inputStreamEndProm = rpcStream.readable
.pipeTo(passthroughTransform.writable)
// Ignore any errors here, we only care that it ended
.catch(() => {});
void inputStream
// Allow us to re-use the readable after reading the first message
.pipeTo(headTransformStream.writable, {
preventClose: true,
preventCancel: true,
})
// Ignore any errors here, we only care that it ended
.catch(() => {});
const cleanUp = async (reason: any) => {
await inputStream.cancel(reason);
await rpcStream.writable.abort(reason);
await inputStreamEndProm;
timer.cancel(cleanupReason);
await timer.catch(() => {});
};
// Read a single empty value to consume the first message
const reader = headTransformStream.readable.getReader();
// Allows timing out when waiting for the first message
let headerMessage:
| ReadableStreamDefaultReadResult<JSONRPCRequest>
| undefined
| void;
try {
headerMessage = await Promise.race([
reader.read(),
timer.then(
() => undefined,
() => {},
),
]);
} catch (e) {
const newErr = new rpcErrors.ErrorRPCHandlerFailed(
'Stream failed waiting for header',
{ cause: e },
);
await inputStreamEndProm;
timer.cancel(cleanupReason);
await timer.catch(() => {});
this.dispatchEvent(
new rpcEvents.RPCErrorEvent({
detail: new rpcErrors.ErrorRPCOutputStreamError(
'Stream failed waiting for header',
{
cause: newErr,
},
),
}),
);
return;
}
// Downgrade back to the raw stream
await reader.cancel();
// There are 2 conditions where we just end here
// 1. The timeout timer resolves before the first message
// 2. the stream ends before the first message
CMCDragonkai
amydevs
Specification
We cannot call
cancelorabortwithout reasons. This can causeundefinederrors, which is very difficult to debug.These are being used to ungracefully close streams, possibly in
RPCClientandRPCServer.This shouldn't be used anymore anyway because we can rely on timeouts instead, rather than ungracefully closing streams.
Additional Context
cancelandabortcalls