From 3793baf6a12e36f309429146d388883902bd1db0 Mon Sep 17 00:00:00 2001 From: LuccaRebelloToledo Date: Tue, 8 Sep 2026 17:27:16 -0300 Subject: [PATCH] fix(aws-serverless): Keep the Lambda extension polling past 300s invocations MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `/event/next` acknowledges the previous event and waits for the next one, so the poll stays open for the whole of the following invocation. Node's `fetch` caps that at undici's 300s `headersTimeout`, and the rejection escapes a loop with no `try`/`catch` — the extension never asks for another event, and Lambda holds every later invocation on that execution environment until the function timeout. The poll now uses `http.request`, which has no default timeout. It carries no deadline either: the poll also spans the environment's frozen idle time, which is unbounded, so a socket deadline would fire on thaw and destroy a poll that was about to be answered. TCP keep-alive covers the case a deadline was there for. A failed poll is retried with capped backoff, bounded so a failure that stops recovering exits rather than logging every 5s forever. A refused poll — 4xx other than 408 and 429 — and a body that is not the event JSON are both failures rather than events, so neither resets the backoff or slips past the SHUTDOWN check. Exiting reports to the Extensions API first, so Lambda recycles the environment instead of leaving it registered and silent. Failures are reported through `console`: `debug` is only enabled from `Sentry.init`, which this process never calls. Fixes #24218 Co-authored-by: Claude Opus 5 --- .../lambda-extension/aws-lambda-extension.ts | 167 +++++++++- .../src/lambda-extension/index.ts | 36 ++- .../test/aws-lambda-extension.test.ts | 291 +++++++++++++++++- 3 files changed, 474 insertions(+), 20 deletions(-) diff --git a/packages/aws-serverless/src/lambda-extension/aws-lambda-extension.ts b/packages/aws-serverless/src/lambda-extension/aws-lambda-extension.ts index 8ecbfe2510ad..e94dab9ce184 100644 --- a/packages/aws-serverless/src/lambda-extension/aws-lambda-extension.ts +++ b/packages/aws-serverless/src/lambda-extension/aws-lambda-extension.ts @@ -10,6 +10,95 @@ import { } from '@sentry/core'; import { DEBUG_BUILD } from './debug-build'; +const POLL_RETRY_BASE_MS = 100; +const POLL_RETRY_MAX_MS = 5_000; +/** Bounded so a permanently unreachable API exits instead of logging every 5s forever. */ +const POLL_MAX_CONSECUTIVE_FAILURES = 20; + +/** The body lands in an error message that a failing poll writes to the console. */ +const ERROR_BODY_MAX_LENGTH = 200; + +/** + * Detects a peer that went away without a FIN/RST, which a request deadline cannot do here: the + * poll is open across the environment's frozen idle time, which is unbounded, and a socket + * deadline runs on real time and would fire on thaw after a long idle — destroying a poll that + * was about to be answered. Keep-alive probes only travel while the environment is running. + */ +const POLL_KEEPALIVE_MS = 30_000; + +/** 408 and 429 are the retryable ones; the rest of 4xx means the poll itself is refused. */ +const RETRYABLE_CLIENT_ERRORS = [408, 429]; + +interface ExtensionEvent { + eventType?: string; +} + +interface ExtensionsApiResponse { + statusCode: number; + body: string; +} + +export class ExtensionsApiError extends Error { + public constructor( + message: string, + public readonly statusCode: number, + ) { + super(message); + this.name = 'ExtensionsApiError'; + } +} + +/** + * Structural rather than `instanceof`: the check has to hold for an error that crossed a + * module boundary, and a transport failure carries `code`, never `statusCode`. + */ +function isClientError(err: unknown): boolean { + const statusCode = (err as { statusCode?: unknown } | null)?.statusCode; + return ( + typeof statusCode === 'number' && + statusCode >= 400 && + statusCode < 500 && + !RETRYABLE_CLIENT_ERRORS.includes(statusCode) + ); +} + +/** + * Exported only for testing purposes. + * + * `fetch` cannot be used for the long poll: Node's implementation applies undici's 300s + * `headersTimeout`, and lifting it would mean passing a dispatcher and depending on `undici` + * directly. `http.request` has no default timeout, and the Extensions API is plain HTTP on + * localhost. + */ +export function request(url: string, headers: Record): Promise { + return new Promise((resolve, reject) => { + const req = http.request(url, { headers }, res => { + const chunks: Buffer[] = []; + res.on('data', (chunk: Buffer) => chunks.push(chunk)); + res.on('end', () => resolve({ statusCode: res.statusCode ?? 0, body: Buffer.concat(chunks).toString() })); + res.on('error', err => { + req.destroy(); + reject(err); + }); + }); + + req.on('socket', socket => socket.setKeepAlive(true, POLL_KEEPALIVE_MS)); + + req.on('error', reject); + req.end(); + }); +} + +function truncate(body: string): string { + return body.length > ERROR_BODY_MAX_LENGTH ? `${body.slice(0, ERROR_BODY_MAX_LENGTH)}...` : body; +} + +function sleep(ms: number): Promise { + return new Promise(resolve => { + setTimeout(resolve, ms); + }); +} + /** * The Extension API Client. */ @@ -42,25 +131,85 @@ export class AwsLambdaExtension { } this._extensionId = res.headers.get('lambda-extension-identifier'); + + if (!this._extensionId) { + throw new Error('Extensions API accepted the registration without returning an extension identifier'); + } } /** - * Advances the extension to the next event. + * Advances the extension to the next event and returns it. */ - public async next(): Promise { + public async next(): Promise { if (!this._extensionId) { throw new Error('Extension ID is not set'); } - const res = await fetch(`${this._baseUrl}/event/next`, { - headers: { - 'Lambda-Extension-Identifier': this._extensionId, - 'Content-Type': 'application/json', - }, + // This request blocks until the next event arrives, so it stays open for the whole + // duration of the current invocation. Under `fetch` that is capped at 300s, so any + // invocation that runs longer than that loses the extension partway through. + const res = await request(`${this._baseUrl}/event/next`, { + 'Lambda-Extension-Identifier': this._extensionId, + 'Content-Type': 'application/json', }); - if (!res.ok) { - throw new Error(`Failed to advance to next event: ${await res.text()}`); + if (res.statusCode < 200 || res.statusCode > 299) { + throw new ExtensionsApiError(`Failed to advance to next event: ${truncate(res.body)}`, res.statusCode); + } + + try { + return JSON.parse(res.body) as ExtensionEvent; + } catch { + // Not an empty event: `run` reads `eventType` to decide when to stop, so a body it cannot + // read has to be a failed poll. Returning `{}` would look like an INVOKE — resetting the + // backoff and re-polling with no delay, which spins the loop on any endpoint answering + // 200 with something that is not JSON, and skips the SHUTDOWN exit. + throw new Error(`Failed to parse the event from the Extensions API: ${truncate(res.body)}`); + } + } + + /** + * Polls the Extensions API until the environment shuts down. + * + * A failed poll is retried rather than ending the loop. Lambda only completes an invocation + * once the runtime and every registered extension have asked for the next event, so an + * extension that stops polling does not fail loudly — it leaves every later invocation on + * that execution environment running until the function timeout kills it. + */ + public async run(): Promise { + let consecutiveFailures = 0; + + for (;;) { + try { + const event = await this.next(); + consecutiveFailures = 0; + + // The runtime API is torn down right after this, so polling again would only produce + // errors on the way out. + if (event.eventType === 'SHUTDOWN') { + return; + } + } catch (err) { + // A poll the API refuses outright is not going to start working; retrying only buries + // the reason under a console error every few seconds for the life of the environment. + if (isClientError(err)) { + throw err; + } + + consecutiveFailures++; + + // Same reasoning once a recoverable-looking failure stops recovering. + if (consecutiveFailures >= POLL_MAX_CONSECUTIVE_FAILURES) { + throw err; + } + + consoleSandbox(() => { + // eslint-disable-next-line no-console + console.error('Sentry Lambda extension: polling the Extensions API failed, retrying.', err); + }); + + await sleep(Math.min(POLL_RETRY_BASE_MS * 2 ** (consecutiveFailures - 1), POLL_RETRY_MAX_MS)); + } } } diff --git a/packages/aws-serverless/src/lambda-extension/index.ts b/packages/aws-serverless/src/lambda-extension/index.ts index f465dae9741d..712053912223 100644 --- a/packages/aws-serverless/src/lambda-extension/index.ts +++ b/packages/aws-serverless/src/lambda-extension/index.ts @@ -1,21 +1,37 @@ #!/usr/bin/env node -import { debug } from '@sentry/core'; +import { consoleSandbox } from '@sentry/core'; import { AwsLambdaExtension } from './aws-lambda-extension'; -import { DEBUG_BUILD } from './debug-build'; -async function main(): Promise { - const extension = new AwsLambdaExtension(); +const extension = new AwsLambdaExtension(); +async function main(): Promise { await extension.register(); extension.startSentryTunnel(); - // eslint-disable-next-line no-constant-condition - while (true) { - await extension.next(); - } + // Returns on SHUTDOWN. The process is left to idle rather than exiting, so envelopes the + // tunnel is still forwarding get their chance to land before Lambda reaps the environment. + await extension.run(); } -main().catch(err => { - DEBUG_BUILD && debug.error('Error in Lambda Extension', err); +main().catch(async err => { + // The debug logger is only enabled from `Sentry.init`, and this process never calls it, so + // nothing reported through the logger from here would ever be visible. + consoleSandbox(() => { + // eslint-disable-next-line no-console + console.error('Sentry Lambda extension: stopped, events will no longer be tunnelled.', err); + }); + + // Reporting lets Lambda recycle the environment; `error` rethrows, and a registration that + // never completed has no id to report with, so neither path should mask the exit. + await extension.error('exit', err as Error).catch(() => undefined); + + // Exiting here is not optional: the tunnel server holds a referenced handle, so the process + // would otherwise stay alive and registered while never asking for another event — and Lambda + // holds every later invocation on this execution environment open until the function timeout. + // + // Deferred by one turn of the loop because `process.exit` does not wait for stderr, which is a + // pipe under Lambda — exiting straight from the microtask above truncates the message written + // there to a single pipe buffer. + setImmediate(() => process.exit(1)); }); diff --git a/packages/aws-serverless/test/aws-lambda-extension.test.ts b/packages/aws-serverless/test/aws-lambda-extension.test.ts index 4c3143eea442..a711ade6a449 100644 --- a/packages/aws-serverless/test/aws-lambda-extension.test.ts +++ b/packages/aws-serverless/test/aws-lambda-extension.test.ts @@ -1,5 +1,13 @@ +import * as http from 'node:http'; +import * as net from 'node:net'; +import type { AddressInfo } from 'node:net'; import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest'; -import { getSentryDSNFromEnv } from '../src/lambda-extension/aws-lambda-extension'; +import { + AwsLambdaExtension, + ExtensionsApiError, + getSentryDSNFromEnv, + request, +} from '../src/lambda-extension/aws-lambda-extension'; describe('getSentryDSNFromEnv', () => { afterEach(() => { @@ -35,3 +43,284 @@ describe('getSentryDSNFromEnv', () => { expect(getSentryDSNFromEnv()).toEqual(undefined); }); }); + +/** + * Stands in for the Lambda Extensions API. `/register` always succeeds so tests can reach the + * poll; everything else is delegated so each test decides how `/event/next` behaves. + */ +async function startExtensionsApi( + onNext: (req: http.IncomingMessage, res: http.ServerResponse) => void, +): Promise<{ close: () => Promise }> { + const server = http.createServer((req, res) => { + if (req.url?.endsWith('/register')) { + res.writeHead(200, { 'lambda-extension-identifier': 'test-extension-id' }); + res.end('{}'); + return; + } + + onNext(req, res); + }); + + await new Promise(resolve => server.listen(0, '127.0.0.1', resolve)); + process.env.AWS_LAMBDA_RUNTIME_API = `127.0.0.1:${(server.address() as AddressInfo).port}`; + + return { + close: () => + new Promise(resolve => { + server.closeAllConnections(); + server.close(() => resolve()); + }), + }; +} + +describe('AwsLambdaExtension.next', () => { + let api: { close: () => Promise } | undefined; + + afterEach(async () => { + await api?.close(); + api = undefined; + delete process.env.AWS_LAMBDA_RUNTIME_API; + vi.restoreAllMocks(); + }); + + test("does not poll through fetch, which would cap the poll at undici's 300s headersTimeout", async () => { + api = await startExtensionsApi((_req, res) => { + res.writeHead(200, { 'Content-Type': 'application/json' }); + res.end(JSON.stringify({ eventType: 'INVOKE' })); + }); + const extension = new AwsLambdaExtension(); + await extension.register(); + + const fetchSpy = vi.spyOn(globalThis, 'fetch'); + await extension.next(); + + expect(fetchSpy).not.toHaveBeenCalled(); + }); + + test('resolves for an event that arrives long after the request was issued', async () => { + api = await startExtensionsApi((_req, res) => { + setTimeout(() => { + res.writeHead(200, { 'Content-Type': 'application/json' }); + res.end(JSON.stringify({ eventType: 'INVOKE' })); + }, 300); + }); + const extension = new AwsLambdaExtension(); + await extension.register(); + + await expect(extension.next()).resolves.toEqual({ eventType: 'INVOKE' }); + }); + + test('rejects with the response body and status when the Extensions API refuses the poll', async () => { + api = await startExtensionsApi((_req, res) => { + res.writeHead(403, { 'Content-Type': 'application/json' }); + res.end('extension not registered'); + }); + const extension = new AwsLambdaExtension(); + await extension.register(); + + const error = await extension.next().catch((err: Error) => err); + + expect(error.message).toBe('Failed to advance to next event: extension not registered'); + expect((error as { statusCode?: number }).statusCode).toBe(403); + }); + + test('rejects a 200 whose body is not the event JSON', async () => { + // Returning an empty event here would look like an INVOKE to `run`: backoff reset, no + // sleep, immediate re-poll — a tight silent loop, and the SHUTDOWN exit never taken. + api = await startExtensionsApi((_req, res) => { + res.writeHead(200, { 'Content-Type': 'application/json' }); + res.end('not json'); + }); + const extension = new AwsLambdaExtension(); + await extension.register(); + + await expect(extension.next()).rejects.toThrow('Failed to parse the event from the Extensions API'); + }); + + test('rejects registration that returns no extension identifier', async () => { + // Left unchecked this yields a null id, and every later poll throws synchronously. + const server = http.createServer((_req, res) => { + res.writeHead(200, { 'Content-Type': 'application/json' }); + res.end('{}'); + }); + await new Promise(resolve => server.listen(0, '127.0.0.1', resolve)); + process.env.AWS_LAMBDA_RUNTIME_API = `127.0.0.1:${(server.address() as AddressInfo).port}`; + api = { close: () => new Promise(resolve => server.close(() => resolve())) }; + + await expect(new AwsLambdaExtension().register()).rejects.toThrow('without returning an extension identifier'); + }); +}); + +type PolledEvent = { eventType?: string }; + +/** + * Drives `run` through a fixed sequence of polls. Anything past the script rejects with a + * client error, which `run` treats as fatal — so a regression that stops honouring SHUTDOWN + * fails in milliseconds instead of spinning the worker until it runs out of heap. + */ +function scriptPolls(extension: AwsLambdaExtension, script: Array) { + let poll = 0; + + return vi.spyOn(extension, 'next').mockImplementation(async () => { + const step = script[poll++]; + + if (step === undefined) { + throw new ExtensionsApiError(`unexpected poll #${poll}`, 400); + } + if (step instanceof Error) { + throw step; + } + return step; + }); +} + +describe('AwsLambdaExtension.run', () => { + let errorSpy: ReturnType; + + beforeEach(() => { + errorSpy = vi.spyOn(console, 'error').mockImplementation(() => {}); + }); + + afterEach(() => { + vi.restoreAllMocks(); + }); + + test('keeps polling after a failed poll', async () => { + // Lambda holds an invocation open until every registered extension asks for the next + // event, so a loop that exits on the first rejection leaves later invocations to be + // killed by the function timeout with no error reported anywhere. + const extension = new AwsLambdaExtension(); + const next = scriptPolls(extension, [ + new Error('socket hang up'), + { eventType: 'INVOKE' }, + { eventType: 'SHUTDOWN' }, + ]); + + await extension.run(); + + expect(next).toHaveBeenCalledTimes(3); + }); + + test('reports a failed poll on the console, which the debug logger cannot do here', async () => { + const extension = new AwsLambdaExtension(); + const pollFailure = new Error('socket hang up'); + scriptPolls(extension, [pollFailure, { eventType: 'SHUTDOWN' }]); + + await extension.run(); + + expect(errorSpy).toHaveBeenCalledWith( + 'Sentry Lambda extension: polling the Extensions API failed, retrying.', + pollFailure, + ); + }); + + test('retries a 408 and a 429 rather than treating them as unrecoverable', async () => { + // These are the two 4xx that do start working again; routing them into the fatal path + // would exit the process over a transient hiccup. + const extension = new AwsLambdaExtension(); + const next = scriptPolls(extension, [ + new ExtensionsApiError('request timeout', 408), + new ExtensionsApiError('too many requests', 429), + { eventType: 'SHUTDOWN' }, + ]); + + await extension.run(); + + expect(next).toHaveBeenCalledTimes(3); + }); + + test('gives up once a retryable failure stops recovering', async () => { + // Without a cap this writes one console error every 5s for as long as the environment is + // thawed, for a condition that is never going to clear. The backoff makes 20 attempts take + // over a minute of real time, hence the fake clock. + vi.useFakeTimers(); + try { + const extension = new AwsLambdaExtension(); + const next = vi.spyOn(extension, 'next').mockRejectedValue(new Error('ECONNREFUSED')); + + const running = expect(extension.run()).rejects.toThrow('ECONNREFUSED'); + await vi.advanceTimersByTimeAsync(120_000); + await running; + + expect(next).toHaveBeenCalledTimes(20); + } finally { + vi.useRealTimers(); + } + }); + + test('stops on SHUTDOWN instead of polling a runtime API that is being torn down', async () => { + // Polling after SHUTDOWN only produces failures on the way out, and the retry path would + // write one console error per attempt on every execution environment teardown. + const extension = new AwsLambdaExtension(); + const next = scriptPolls(extension, [{ eventType: 'SHUTDOWN' }]); + + await extension.run(); + + expect(next).toHaveBeenCalledTimes(1); + expect(errorSpy).not.toHaveBeenCalled(); + }); +}); + +describe('AwsLambdaExtension.run — unrecoverable poll', () => { + let api: { close: () => Promise } | undefined; + + beforeEach(() => { + vi.spyOn(console, 'error').mockImplementation(() => {}); + }); + + afterEach(async () => { + await api?.close(); + api = undefined; + delete process.env.AWS_LAMBDA_RUNTIME_API; + vi.restoreAllMocks(); + }); + + test('gives up on a client error rather than retrying it forever', async () => { + // Retrying a rejected registration never recovers; it just buries the reason under a + // console error every few seconds for the life of the execution environment. + api = await startExtensionsApi((_req, res) => { + res.writeHead(403, { 'Content-Type': 'application/json' }); + res.end('extension not registered'); + }); + const extension = new AwsLambdaExtension(); + await extension.register(); + + await expect(extension.run()).rejects.toThrow('extension not registered'); + }); +}); + +describe('request', () => { + afterEach(() => { + vi.restoreAllMocks(); + }); + + test('rejects when the peer goes away mid-poll', async () => { + // The poll has no deadline on purpose — it spans the environment's frozen idle time, which + // is unbounded — so a peer that disappears has to surface as a socket error instead. + const server = http.createServer(); + server.on('connection', socket => socket.destroy()); + await new Promise(resolve => server.listen(0, '127.0.0.1', resolve)); + const url = `http://127.0.0.1:${(server.address() as AddressInfo).port}/`; + + await expect(request(url, {})).rejects.toThrow(); + + await new Promise(resolve => server.close(() => resolve())); + }); + + test('enables TCP keep-alive so a dead peer is detected without a deadline', async () => { + const server = http.createServer((_req, res) => { + res.writeHead(200); + res.end('{}'); + }); + await new Promise(resolve => server.listen(0, '127.0.0.1', resolve)); + const url = `http://127.0.0.1:${(server.address() as AddressInfo).port}/`; + const setKeepAlive = vi.spyOn(net.Socket.prototype, 'setKeepAlive'); + + await request(url, {}); + + expect(setKeepAlive).toHaveBeenCalledWith(true, 30_000); + + server.closeAllConnections(); + await new Promise(resolve => server.close(() => resolve())); + }); +});