Zero-Copy Node Streams: Pipe Large S3 Objects Without Buffering

// ❌ Full object in memory
import { S3Client, GetObjectCommand } from "@aws-sdk/client-s3";
const s3 = new S3Client({});
const out = await s3.send(new GetObjectCommand({ Bucket, Key }));
const bytes = await out.Body!.transformToByteArray(); // boom on large keys

// ✅ Stream with backpressure + incremental checksum
import { S3Client, GetObjectCommand, PutObjectCommand } from "@aws-sdk/client-s3";
import { Upload } from "@aws-sdk/lib-storage";
import { pipeline } from "node:stream/promises";
import { createHash } from "node:crypto";
import { createGunzip } from "node:zlib";
import { Readable, Transform } from "node:stream";

const s3 = new S3Client({});

function hashTap(algo: string) {
 const h = createHash(algo);
 return new Transform({
 transform(chunk, _e, cb) {
 h.update(chunk);
 cb(null, chunk);
 },
 flush(cb) {
 (this as any).digestHex = h.digest("hex");
 cb();
 },
 });
}

export async function mirrorS3(srcBucket: string, key: string, dstBucket: string) {
 const got = await s3.send(new GetObjectCommand({ Bucket: srcBucket, Key: key }));
 const body = got.Body as Readable;
 const tap = hashTap("sha256");

 const upload = new Upload({
 client: s3,
 params: {
 Bucket: dstBucket,
 Key: key,
 Body: body.pipe(tap), // still prefer pipeline() below for error propagation
 ContentType: got.ContentType,
 },
 queueSize: 16 * 1024 * 1024,
 });

 // Better: explicit pipeline into a PassThrough feeding Upload
 await upload.done();
 const digest = (tap as any).digestHex as string;
 if (got.ChecksumSHA256 && got.ChecksumSHA256 !== digest) {
 // Note: S3 checksum headers may be base64 — normalize before compare
 throw new Error("checksum_mismatch");
 }
 return digest;
}

import { pipeline } from "node:stream/promises";
import type { IncomingMessage, ServerResponse } from "node:http";

export async function downloadHandler(req: IncomingMessage, res: ServerResponse, key: string) {
 const got = await s3.send(new GetObjectCommand({ Bucket: process.env.BUCKET!, Key: key }));
 res.writeHead(200, {
 "Content-Type": got.ContentType ?? "application/octet-stream",
 "Content-Length": got.ContentLength ? String(got.ContentLength) : undefined,
 ETag: got.ETag,
 });
 await pipeline(got.Body as Readable, res);
}

import PQueue from "p-queue";
const queue = new PQueue({ concurrency: 4 });

export function enqueueMirror(job: () => Promise<void>) {
 return queue.add(job);
}

export async function readGunzipLimited(src: Readable, maxBytes: number) {
 let n = 0;
 const limiter = new Transform({
 transform(chunk, _e, cb) {
 n += chunk.length;
 if (n > maxBytes) cb(new Error("uncompressed_limit"));
 else cb(null, chunk);
 },
 });
 // pipeline(src, createGunzip(), limiter, dest)
 return { limiter };
}