Lambda Response Streaming: Backpressure When Clients Stall Mid-Transfer

// handler.ts — Node 20 Function URL streaming with backpressure
import { Readable } from "node:stream";
import { pipeline } from "node:stream/promises";

declare const awslambda: {
  streamifyResponse: (
    fn: (event: unknown, responseStream: NodeJS.WritableStream, context: unknown) => Promise<void>
  ) => unknown;
  HttpResponseStream: {
    from: (stream: NodeJS.WritableStream, meta: { statusCode: number; headers: Record<string, string> }) => NodeJS.WritableStream;
  };
};

export const handler = awslambda.streamifyResponse(async (event, responseStream, _ctx) => {
  const httpStream = awslambda.HttpResponseStream.from(responseStream, {
    statusCode: 200,
    headers: {
      "Content-Type": "application/octet-stream",
      "Cache-Control": "no-store",
    },
  });

  // ✅ Bound the writable buffer — default is often too generous for 128–512 MB functions
  (httpStream as any).writableHighWaterMark = 64 * 1024;

  const abort = new AbortController();
  httpStream.on("error", () => abort.abort());
  httpStream.on("close", () => abort.abort());

  const source = createProducer(event, abort.signal); // async generator → Readable
  try {
    await pipeline(Readable.from(source), httpStream);
  } catch (err) {
    // Client stall / reset — do not retry the whole payload in-process
    console.error(JSON.stringify({ msg: "stream_aborted", err: String(err) }));
  }
});

async function* createProducer(event: unknown, signal: AbortSignal) {
  for await (const chunk of fetchUpstreamChunks(event)) {
    if (signal.aborted) return; // ✅ stop producing
    // ❌ Never: chunks.push(chunk) into a growing array
    yield chunk;
  }
}

function wrapWithStallGuard(stream: NodeJS.WritableStream, opts: { minBps: number; windowMs: number }) {
  let written = 0;
  let windowStart = Date.now();
  const timer = setInterval(() => {
    const elapsed = Date.now() - windowStart;
    const bps = (written * 1000) / Math.max(elapsed, 1);
    if (bps < opts.minBps) {
      stream.destroy(new Error(`stall_below_${opts.minBps}_bps`));
    }
    written = 0;
    windowStart = Date.now();
  }, opts.windowMs);

  const origWrite = stream.write.bind(stream);
  (stream as any).write = (chunk: any, ...rest: any[]) => {
    written += Buffer.isBuffer(chunk) ? chunk.length : Buffer.byteLength(String(chunk));
    return origWrite(chunk, ...rest);
  };
  stream.on("close", () => clearInterval(timer));
  return stream;
}

# ✅ Probe backpressure locally (illustrative)
curl -N --limit-rate 1k "https://xxx.lambda-url.region.on.aws/export" -o /dev/null
# Watch CloudWatch: Duration rises, Max Memory used stays flat if backpressure works