Skip to content

Commit 0ea75be

Browse files
TooTallNateTooTallNate
committed
WIP Rusty
1 parent f007849 commit 0ea75be

File tree

3 files changed

+70
-201
lines changed

3 files changed

+70
-201
lines changed

‎src/bootstrap‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -11,4 +11,4 @@ else
1111
export DENO_DIR="$LAMBDA_TASK_ROOT/.vercel/cache/deno"
1212
fi
1313

14-
exec ${AWS_LAMBDA_EXEC_WRAPPER-} deno run $args ".vercel-deno-runtime.ts"
14+
exec deno run $args ".vercel-deno-runtime.ts"

‎src/deno-lambda.ts‎

Lines changed: 13 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
/**
22
* The default version of Deno that will be downloaded at build-time.
33
*/
4-
const DEFAULT_DENO_VERSION = 'v1.44.4';
4+
const DEFAULT_DENO_VERSION = 'v1.46.3';
55

66
import { spawn } from 'node:child_process';
77
import { fileURLToPath, pathToFileURL } from 'node:url';
@@ -47,7 +47,7 @@ const bootstrapData = readFileSync(bootstrapPath, 'utf8');
4747
const bootstrapMode = statSync(bootstrapPath).mode;
4848

4949
export interface DenoLambdaOptions
50-
extends Omit<LambdaOptionsWithFiles, 'runtime' | 'supportsWrapper'> {}
50+
extends Omit<LambdaOptionsWithFiles, 'runtime' | 'supportsWrapper'> { }
5151

5252
export interface DenoLambdaBuildOptions {
5353
entrypoint: string;
@@ -99,11 +99,11 @@ export class DenoLambda extends Lambda {
9999
// then also download Deno binary for the build host
100100
process.platform !== 'linux' || process.arch !== 'x64'
101101
? downloadDeno(
102-
denoDir,
103-
denoVersion,
104-
process.platform,
105-
process.arch
106-
)
102+
denoDir,
103+
denoVersion,
104+
process.platform,
105+
process.arch
106+
)
107107
: undefined,
108108
]);
109109

@@ -215,7 +215,12 @@ export class DenoLambda extends Lambda {
215215
return new DenoLambda({
216216
files: outputFiles,
217217
handler: entrypoint,
218-
environment: args.env,
218+
environment: {
219+
...args.env,
220+
//VERCEL_FORCE_STREAMING_RUNTIME: '1',
221+
SCRIPT_PATH: '/var/task/bootstrap',
222+
FOO: 'bar'
223+
},
219224
});
220225
}
221226
}

‎src/runtime/runtime.ts‎

Lines changed: 56 additions & 192 deletions
Original file line numberDiff line numberDiff line change
@@ -1,209 +1,73 @@
1-
import * as base64 from 'https://deno.land/[email protected]/encoding/base64.ts';
1+
const initStart = Date.now();
2+
23
import type {
34
Handler,
45
ConnInfo,
56
} from 'https://deno.land/[email protected]/http/server.ts';
67

7-
interface VercelRequestPayload {
8-
method: string;
9-
path: string;
10-
headers: Record<string, string>;
11-
body: string;
12-
}
13-
14-
type VercelResponseHeaders = Record<string, string | string[]>;
15-
16-
interface VercelResponsePayload {
17-
statusCode: number;
18-
headers: VercelResponseHeaders;
19-
encoding: 'base64';
20-
body: string;
21-
}
22-
23-
const RUNTIME_PATH = '2018-06-01/runtime';
24-
25-
const { _HANDLER, ENTRYPOINT, AWS_LAMBDA_RUNTIME_API } = Deno.env.toObject();
26-
27-
Deno.env.delete('SHLVL');
28-
29-
function fromVercelRequest(payload: VercelRequestPayload): Request {
30-
const headers = new Headers(payload.headers);
31-
const base = `${headers.get('x-forwarded-proto')}://${headers.get(
32-
'x-forwarded-host'
33-
)}`;
34-
const url = new URL(payload.path, base);
35-
const body = payload.body ? base64.decode(payload.body) : undefined;
36-
return new Request(url.href, {
37-
method: payload.method,
38-
headers,
39-
body,
40-
});
41-
}
42-
43-
function headersToVercelHeaders(headers: Headers): VercelResponseHeaders {
44-
const h: VercelResponseHeaders = {};
45-
for (const [name, value] of headers) {
46-
const cur = h[name];
47-
if (typeof cur === 'string') {
48-
h[name] = [cur, value];
49-
} else if (Array.isArray(cur)) {
50-
cur.push(value);
51-
} else {
52-
h[name] = value;
53-
}
54-
}
55-
return h;
56-
}
57-
58-
async function toVercelResponse(res: Response): Promise<VercelResponsePayload> {
59-
let body = '';
60-
const bodyBuffer = await res.arrayBuffer();
61-
if (bodyBuffer.byteLength > 0) {
62-
body = base64.encode(bodyBuffer);
63-
}
64-
65-
return {
66-
statusCode: res.status,
67-
headers: headersToVercelHeaders(res.headers),
68-
encoding: 'base64',
69-
body,
70-
};
71-
}
72-
73-
async function processEvents(): Promise<void> {
74-
let handler: Handler | null = null;
75-
76-
while (true) {
77-
const { event, awsRequestId } = await nextInvocation();
78-
let result: VercelResponsePayload;
79-
try {
80-
if (!handler) {
81-
const mod = await import(`./${_HANDLER}`);
82-
handler = mod.default;
83-
if (typeof handler !== 'function') {
84-
throw new Error('Failed to load handler function');
85-
}
86-
}
87-
88-
const payload = JSON.parse(event.body) as VercelRequestPayload;
89-
const req = fromVercelRequest(payload);
8+
const { _HANDLER, ENTRYPOINT, VERCEL_IPC_FD } = Deno.env.toObject();
909

91-
const connInfo: ConnInfo = {
92-
// TODO: how to properly calculate these?
93-
// @ts-ignore - `rid` is not on the `ConnInfo` interface, but it's required by Oak
94-
rid: 0,
95-
localAddr: { hostname: '127.0.0.1', port: 0, transport: 'tcp' },
96-
remoteAddr: {
97-
hostname: '127.0.0.1',
98-
port: 0,
99-
transport: 'tcp',
100-
},
101-
};
102-
103-
// Run user code
104-
const res = await handler(req, connInfo);
105-
result = await toVercelResponse(res);
106-
} catch (e: unknown) {
107-
const err = e instanceof Error ? e : new Error(String(e));
108-
console.error(err);
109-
await invokeError(err, awsRequestId);
110-
continue;
111-
}
112-
await invokeResponse(result, awsRequestId);
113-
}
10+
function isNetAddr(v: any): v is Deno.NetAddr {
11+
return v && typeof v.port === 'number';
11412
}
11513

116-
async function nextInvocation() {
117-
const res = await request('invocation/next');
118-
119-
if (res.status !== 200) {
120-
throw new Error(
121-
`Unexpected "/invocation/next" response: ${JSON.stringify(res)}`
122-
);
123-
}
124-
125-
const traceId = res.headers.get('lambda-runtime-trace-id');
126-
if (typeof traceId === 'string') {
127-
Deno.env.set('_X_AMZN_TRACE_ID', traceId);
128-
} else {
129-
Deno.env.delete('_X_AMZN_TRACE_ID');
130-
}
14+
if (_HANDLER) {
13115

132-
const awsRequestId = res.headers.get('lambda-runtime-aws-request-id');
133-
if (typeof awsRequestId !== 'string') {
134-
throw new Error(
135-
'Did not receive "lambda-runtime-aws-request-id" header'
136-
);
16+
const mod = await import(`./${_HANDLER}`);
17+
const handler: Handler = mod.default;
18+
if (typeof handler !== 'function') {
19+
throw new Error('Failed to load handler function');
13720
}
13821

139-
const event = JSON.parse(res.body);
140-
return { event, awsRequestId };
141-
}
22+
// Spawn HTTP server on ephemeral port
23+
const listener = Deno.listen({ port: 3030 /* 0 */ });
14224

143-
async function invokeResponse(
144-
result: VercelResponsePayload,
145-
awsRequestId: string
146-
) {
147-
const res = await request(`invocation/${awsRequestId}/response`, {
148-
method: 'POST',
149-
headers: {
150-
'Content-Type': 'application/json',
151-
},
152-
body: JSON.stringify(result),
153-
});
154-
if (res.status !== 202) {
155-
throw new Error(
156-
`Unexpected "/invocation/response" response: ${JSON.stringify(res)}`
157-
);
25+
if (!isNetAddr(listener.addr)) {
26+
throw new Error('Server not listening on TCP port');
15827
}
159-
}
160-
161-
function invokeError(err: Error, awsRequestId: string) {
162-
return postError(`invocation/${awsRequestId}/error`, err);
163-
}
164-
165-
async function postError(path: string, err: Error): Promise<void> {
166-
const lambdaErr = toLambdaErr(err);
167-
const res = await request(path, {
168-
method: 'POST',
169-
headers: {
170-
'Content-Type': 'application/json',
171-
'Lambda-Runtime-Function-Error-Type': 'Unhandled',
172-
},
173-
body: JSON.stringify(lambdaErr),
174-
});
175-
if (res.status !== 202) {
176-
throw new Error(
177-
`Unexpected "${path}" response: ${JSON.stringify(res)}`
178-
);
28+
const { port } = listener.addr;
29+
console.log({ port });
30+
console.log({ VERCEL_IPC_FD });
31+
32+
const ipcSock = await Deno.connect({ path: `/dev/fd/${VERCEL_IPC_FD}`, transport: "unix" });
33+
ipcSock.write(new TextEncoder().encode(`${JSON.stringify({
34+
"type": "server-started",
35+
"payload": {
36+
"initDuration": Date.now() - initStart, // duration to init the process, connect to the unix domain socket & start the HTTP server in milliseconds
37+
"httpPort": port // the port of the HTTP server
38+
}
39+
})}\0`));
40+
41+
// Serve HTTP requests to handler function
42+
const conn = await listener.accept();
43+
const s = Deno.serveHttp(conn);
44+
for await (const req of s) {
45+
const connInfo: ConnInfo = {
46+
// @ts-ignore - `rid` is not on the `ConnInfo` interface, but it's required by Oak
47+
rid: conn.rid,
48+
localAddr: conn.localAddr,
49+
remoteAddr: conn.remoteAddr,
50+
};
51+
const requestId = req.headers.get('x-vercel-internal-request-id');
52+
const invocationId = req.headers.get('x-vercel-internal-invocation-id');
53+
Promise.resolve(handler(req.request, connInfo)).then((res: Response) => {
54+
req.respondWith(res);
55+
// TODO: figure out how to wait for HTTP request to complete
56+
setTimeout(() => {
57+
const endPayload = {
58+
"type": "end",
59+
"payload": {
60+
"context": {
61+
invocationId, // invocation-id from the http request
62+
requestId // request-id from the http request
63+
},
64+
"error": "" // optional
65+
}
66+
};
67+
ipcSock.write(new TextEncoder().encode(`${JSON.stringify(endPayload)}\0`));
68+
}, 1000);
69+
});
17970
}
180-
}
181-
182-
async function request(path: string, options?: RequestInit) {
183-
const url = `http://${AWS_LAMBDA_RUNTIME_API}/${RUNTIME_PATH}/${path}`;
184-
const res = await fetch(url, options);
185-
const body = await res.text();
186-
return {
187-
status: res.status,
188-
headers: res.headers,
189-
body,
190-
};
191-
}
192-
193-
function toLambdaErr({ name, message, stack }: Error) {
194-
return {
195-
errorType: name,
196-
errorMessage: message,
197-
stackTrace: (stack || '').split('\n').slice(1),
198-
};
199-
}
200-
201-
if (_HANDLER) {
202-
// Runtime - execute the runtime loop
203-
processEvents().catch((err) => {
204-
console.error(err);
205-
Deno.exit(1);
206-
});
20771
} else {
20872
// Build - import the entrypoint so that it gets cached
20973
await import(ENTRYPOINT);

0 commit comments

Comments
 (0)