Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Update HandlerContext signal to trigger on other reasons than timeouts #531

Merged
merged 8 commits into from May 10, 2023
Merged

Update HandlerContext signal to trigger on other reasons than timeouts #531

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
8fc4a9d
Add an AbortSignal to HandlerContext
Mar 13, 2023
b95f70a
Honor HTTP/2 error codes
May 9, 2023
70ebc21
Expose a single AbortSignal in HandlerContext
May 9, 2023
9d3c443
Improve documentation of the new AbortSignal
May 9, 2023
3daa1e9
Wrap JSDoc at 80 columns
May 9, 2023
820d4ee
Remove an intermediate variable
May 9, 2023
2eb01ac
make format
timostamm May 10, 2023
cdd0e81
Add basic tests
timostamm May 10, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions packages/connect-node-test/src/helpers/test-routes.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,7 +69,7 @@ const testService: ServiceImpl<typeof TestService> = {
);
for (const param of request.responseParameters) {
await maybeDelayResponse(param);
context.deadline?.throwIfAborted();
context.signal.throwIfAborted();
yield {
payload: interop.makeServerPayload(request.responseType, param.size),
};
Expand All @@ -85,7 +85,7 @@ const testService: ServiceImpl<typeof TestService> = {
);
for (const param of request.responseParameters) {
await maybeDelayResponse(param);
context.deadline?.throwIfAborted();
context.signal.throwIfAborted();
yield {
payload: interop.makeServerPayload(request.responseType, param.size),
};
Expand Down Expand Up @@ -119,7 +119,7 @@ const testService: ServiceImpl<typeof TestService> = {
for await (const req of requests) {
for (const param of req.responseParameters) {
await maybeDelayResponse(param);
context.deadline?.throwIfAborted();
context.signal.throwIfAborted();
yield {
payload: interop.makeServerPayload(req.responseType, param.size),
};
Expand All @@ -141,7 +141,7 @@ const testService: ServiceImpl<typeof TestService> = {
for await (const req of buffer) {
for (const param of req.responseParameters) {
await maybeDelayResponse(param);
context.deadline?.throwIfAborted();
context.signal.throwIfAborted();
yield {
payload: interop.makeServerPayload(req.responseType, param.size),
};
Expand Down
21 changes: 20 additions & 1 deletion packages/connect-node/src/node-universal-handler.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,10 @@ import {
nodeHeaderToWebHeader,
webHeaderToNodeHeaders,
} from "./node-universal-header.js";
import { connectErrorFromNodeReason } from "./node-error.js";
import {
connectErrorFromH2ResetCode,
connectErrorFromNodeReason,
} from "./node-error.js";

/**
* NodeHandlerFn is compatible with http.RequestListener and its equivalent
Expand Down Expand Up @@ -89,12 +92,28 @@ export function universalRequestFromNodeRequest(
parsedJsonBody !== undefined
? parsedJsonBody
: asyncIterableFromNodeServerRequest(nodeRequest);
const abortController = new AbortController();
if ("stream" in nodeRequest) {
// HTTP/2 has error codes we want to honor
nodeRequest.on("close", () => {
const err = connectErrorFromH2ResetCode(nodeRequest.stream.rstCode);
if (err !== undefined) {
abortController.abort(err);
} else {
abortController.abort();
}
});
} else {
// HTTP/1.1 does not
nodeRequest.on("close", () => abortController.abort());
}
return {
httpVersion: nodeRequest.httpVersion,
method: nodeRequest.method ?? "",
url: new URL(pathname, `${protocol}://${authority}`),
header: nodeHeaderToWebHeader(nodeRequest.headers),
body,
signal: abortController.signal,
};
}

Expand Down
11 changes: 8 additions & 3 deletions packages/connect/src/implementation.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@ import type {
} from "@bufbuild/protobuf";
import { ConnectError } from "./connect-error.js";
import { Code } from "./code.js";
import { createLinkedAbortController } from "./protocol/linked-abort-controller.js";

// prettier-ignore
/**
Expand Down Expand Up @@ -71,8 +72,11 @@ export interface HandlerContext {
*/
readonly service: ServiceType;

// TODO
readonly deadline?: AbortSignal;
/**
* An AbortSignal that is aborted when the connection with the client is closed
* or when the deadline is reached.
*/
readonly signal: AbortSignal;

/**
* Incoming request headers.
Expand All @@ -99,14 +103,15 @@ export interface HandlerContext {
export function createHandlerContext(
spec: { service: ServiceType; method: MethodInfo },
deadline: AbortSignal | undefined, // TODO
signal: AbortSignal,
This conversation was marked as resolved.
Show resolved Hide resolved
requestHeader: HeadersInit,
responseHeader: HeadersInit,
responseTrailer: HeadersInit
): HandlerContext {
return {
method: spec.method,
service: spec.service,
deadline,
signal: createLinkedAbortController(deadline, signal).signal,
requestHeader: new Headers(requestHeader),
responseHeader: new Headers(responseHeader),
responseTrailer: new Headers(responseTrailer),
Expand Down
100 changes: 85 additions & 15 deletions packages/connect/src/protocol-connect/handler-factory.spec.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,15 +26,11 @@ import {
createUniversalHandlerClient,
encodeEnvelope,
pipeTo,
readAllBytes,
sinkAll,
transformSplitEnvelope,
} from "../protocol/index.js";
import { ConnectError } from "../connect-error.js";
import {
createAsyncIterableBytes,
readAllBytes,
} from "../protocol/async-iterable-helper.spec.js";
import { Code } from "../code.js";
import { Code, ConnectError } from "../index.js";
import { errorFromJsonBytes } from "./error-json.js";
import { endStreamFromJson } from "./end-stream.js";
import { createTransport } from "./transport.js";
Expand Down Expand Up @@ -147,6 +143,7 @@ describe("createHandlerFactory()", function () {
url: new URL("https://example.com"),
header: new Headers({ "Content-Type": "application/json" }),
body: 777,
signal: new AbortController().signal,
});
expect(res.status).toBe(400);
expect(res.body).toBeInstanceOf(Uint8Array);
Expand All @@ -171,6 +168,7 @@ describe("createHandlerFactory()", function () {
"Connect-Protocol-Version": "UNEXPECTED",
}),
body: 777,
signal: new AbortController().signal,
});
expect(res.status).toBe(400);
expect(res.body).toBeInstanceOf(Uint8Array);
Expand Down Expand Up @@ -201,14 +199,15 @@ describe("createHandlerFactory()", function () {
method: "POST",
url: new URL("https://example.com"),
header: new Headers({ "Content-Type": "application/connect+json" }),
body: createAsyncIterableBytes(new Uint8Array()),
body: createAsyncIterable([new Uint8Array()]),
signal: new AbortController().signal,
});
expect(res.status).toBe(200);
expect(res.body).not.toBeInstanceOf(Uint8Array);
expect(res.body).not.toBeUndefined();
if (res.body !== undefined && Symbol.asyncIterator in res.body) {
const end = endStreamFromJson(
(await readAllBytes(res.body)).slice(5)
(await readAllBytes(res.body, Number.MAX_SAFE_INTEGER)).slice(5)
);
expect(end.error?.message).toBe(
'[invalid_argument] missing required header: set Connect-Protocol-Version to "1"'
Expand All @@ -224,14 +223,15 @@ describe("createHandlerFactory()", function () {
"Content-Type": "application/connect+json",
"Connect-Protocol-Version": "UNEXPECTED",
}),
body: createAsyncIterableBytes(new Uint8Array()),
body: createAsyncIterable([new Uint8Array()]),
signal: new AbortController().signal,
});
expect(res.status).toBe(200);
expect(res.body).not.toBeInstanceOf(Uint8Array);
expect(res.body).not.toBeUndefined();
if (res.body !== undefined && Symbol.asyncIterator in res.body) {
const end = endStreamFromJson(
(await readAllBytes(res.body)).slice(5)
(await readAllBytes(res.body, Number.MAX_SAFE_INTEGER)).slice(5)
);
expect(end.error?.message).toBe(
'[invalid_argument] Connect-Protocol-Version must be "1": got "UNEXPECTED"'
Expand All @@ -242,15 +242,15 @@ describe("createHandlerFactory()", function () {
});

describe("deadlines", function () {
describe("unary", function () {
describe("with unary RPC", function () {
it("should raise an error with code DEADLINE_EXCEEDED if exceeded", async function () {
const timeoutMs = 1;
const { handler, service, method } = setupTestHandler(
testService.methods.foo,
{},
async (req, ctx) => {
await new Promise((r) => setTimeout(r, timeoutMs + 50));
ctx.deadline?.throwIfAborted();
ctx.signal.throwIfAborted();
return { value: req.value.toString(10) };
}
);
Expand All @@ -262,14 +262,15 @@ describe("createHandlerFactory()", function () {
),
header: requestHeader(method.kind, true, timeoutMs, undefined),
body: createAsyncIterable([new Uint8Array(0)]),
signal: new AbortController().signal,
});
expect(res.status).toBe(408);
expect(res.body).toBeDefined();
if (res.body !== undefined) {
const bodyBytes =
res.body instanceof Uint8Array
? res.body
: await readAllBytes(res.body);
: await readAllBytes(res.body, Number.MAX_SAFE_INTEGER);
const err = errorFromJsonBytes(
bodyBytes,
undefined,
Expand All @@ -282,7 +283,7 @@ describe("createHandlerFactory()", function () {
}
});
});
describe("streaming", function () {
describe("with streaming RPC", function () {
async function getLastEnvelope(res: UniversalServerResponse) {
expect(res.body).toBeDefined();
expect(res.body).not.toBeInstanceOf(Uint8Array);
Expand All @@ -306,7 +307,7 @@ describe("createHandlerFactory()", function () {
{},
async function* (req, ctx) {
await new Promise((r) => setTimeout(r, timeoutMs + 50));
ctx.deadline?.throwIfAborted();
ctx.signal.throwIfAborted();
yield { value: req.value.toString(10) };
}
);
Expand All @@ -318,6 +319,7 @@ describe("createHandlerFactory()", function () {
),
header: requestHeader(method.kind, true, timeoutMs, undefined),
body: createAsyncIterable([encodeEnvelope(0, new Uint8Array(0))]),
signal: new AbortController().signal,
});
expect(res.status).toBe(200);
expect(res.body).toBeDefined();
Expand All @@ -334,4 +336,72 @@ describe("createHandlerFactory()", function () {
});
});
});

describe("request abort signal", function () {
describe("with unary RPC", function () {
it("should trigger handler context signal", async function () {
let handlerContextSignal: AbortSignal | undefined;
const { handler, service, method } = setupTestHandler(
testService.methods.foo,
{},
async (req, ctx) => {
handlerContextSignal = ctx.signal;
for (;;) {
await new Promise((r) => setTimeout(r, 1));
ctx.signal.throwIfAborted();
}
}
);
const ac = new AbortController();
const resPromise = handler({
httpVersion: "2.0",
method: "POST",
url: new URL(
`https://example.com/${service.typeName}/${method.name}`
),
header: requestHeader(method.kind, true, undefined, undefined),
body: createAsyncIterable([new Uint8Array(0)]),
signal: ac.signal,
});
ac.abort("test-reason");
await resPromise;
expect(handlerContextSignal).toBeDefined();
expect(handlerContextSignal?.aborted).toBeTrue();
expect(handlerContextSignal?.reason).toBe("test-reason");
});
});
describe("with streaming RPC", function () {
it("should trigger handler context signal", async function () {
let handlerContextSignal: AbortSignal | undefined;
const { handler, service, method } = setupTestHandler(
testService.methods.bar,
{},
// eslint-disable-next-line require-yield
async function* (req, ctx) {
handlerContextSignal = ctx.signal;
for (;;) {
await new Promise((r) => setTimeout(r, 1));
ctx.signal.throwIfAborted();
}
}
);
const ac = new AbortController();
const resPromise = handler({
httpVersion: "2.0",
method: "POST",
url: new URL(
`https://example.com/${service.typeName}/${method.name}`
),
header: requestHeader(method.kind, true, undefined, undefined),
body: createAsyncIterable([encodeEnvelope(0, new Uint8Array(0))]),
signal: ac.signal,
});
ac.abort("test-reason");
await resPromise;
expect(handlerContextSignal).toBeDefined();
expect(handlerContextSignal?.aborted).toBeTrue();
expect(handlerContextSignal?.reason).toBe("test-reason");
});
});
});
});
2 changes: 2 additions & 0 deletions packages/connect/src/protocol-connect/handler-factory.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -157,6 +157,7 @@ function createUnaryHandler<I extends Message<I>, O extends Message<O>>(
const context = createHandlerContext(
spec,
deadline.signal,
req.signal,
req.header,
{
[headerContentType]: type.binary
Expand Down Expand Up @@ -292,6 +293,7 @@ function createStreamHandler<I extends Message<I>, O extends Message<O>>(
const context = createHandlerContext(
spec,
deadline.signal,
req.signal,
req.header,
{
[headerContentType]: type.binary
Expand Down
Loading