Skip to content

Commit 367bfbd

Browse files
committed
Build platform tarballs from a single decompressed buffer (fix flaky 1102)
The streaming multipart build spiked past 128MB during R2 part uploads: while a part upload was in flight the consumer paused and the decompressor ran ahead. Decompress the upstream ONCE into one buffer, rewrite package.json in place (no second copy of the binary), then frame the buffer as gzip stored blocks into the multipart upload. With the buffer fully in memory, no decompressor runs during uploads, so nothing buffers ahead; peak footprint matches buildMetaLight.
1 parent 59d922b commit 367bfbd

5 files changed

Lines changed: 237 additions & 491 deletions

File tree

src/tarball/fetchUpstreamTarball.ts

Lines changed: 2 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -23,35 +23,10 @@ async function fetchUpstream(url: string, maxBytes: number) {
2323
return res.body
2424
}
2525

26-
/**
27-
* Stream an upstream pkg.pr.new tarball, enforcing a maximum size as bytes flow
28-
* so a malicious or oversized upstream cannot exhaust the Worker. The body is
29-
* never buffered whole, which is what keeps the large platform-binary tarballs
30-
* within the Worker memory budget.
31-
*/
32-
export async function fetchUpstreamTarballStream(
33-
url: string,
34-
maxBytes: number,
35-
): Promise<ReadableStream<Uint8Array>> {
36-
const body = await fetchUpstream(url, maxBytes)
37-
let total = 0
38-
return body.pipeThrough(
39-
new TransformStream<Uint8Array, Uint8Array>({
40-
transform(chunk, controller) {
41-
total += chunk.byteLength
42-
if (total > maxBytes) {
43-
throw new HttpError(413, 'Upstream tarball exceeds the maximum size')
44-
}
45-
controller.enqueue(chunk)
46-
},
47-
}),
48-
)
49-
}
50-
5126
/**
5227
* Download an upstream pkg.pr.new tarball into memory, enforcing a maximum size
53-
* while streaming. Used for the small preview packages whose integrity is
54-
* computed over the full bytes; large binaries use the streaming path instead.
28+
* while streaming so a malicious or oversized upstream cannot exhaust the
29+
* Worker.
5530
*/
5631
export async function fetchUpstreamTarball(
5732
url: string,

src/tarball/getPreviewBuild.ts

Lines changed: 72 additions & 55 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,10 @@ import { metaKey, tarballKey } from '../cache/r2Cache'
77
import { tarballCacheControl } from '../cache/headers'
88
import { rewritePackageJson } from './rewritePackageJson'
99
import { assertSafeTarballPath } from '../security/validateTarballPath'
10-
import { rewriteTarballEntryStream } from './rewriteTarballStream'
10+
import {
11+
emitStoredGzip,
12+
rewriteTarEntryInPlace,
13+
} from './rewriteTarballBuffer'
1114
import {
1215
buildPreviewTarball,
1316
encodePackageJson,
@@ -16,33 +19,82 @@ import {
1619
type PreviewBuild,
1720
type PreviewMeta,
1821
} from './buildPreviewTarball'
19-
import {
20-
fetchUpstreamTarball,
21-
fetchUpstreamTarballStream,
22-
} from './fetchUpstreamTarball'
22+
import { fetchUpstreamTarball } from './fetchUpstreamTarball'
2323

2424
// R2 multipart parts must be >=5 MiB and (except the last) equal-sized.
2525
const PART_SIZE = 10 * 1024 * 1024
2626

27+
function mergeChunks(chunks: Uint8Array[], total: number): Uint8Array {
28+
if (chunks.length === 1) return chunks[0]
29+
const out = new Uint8Array(total)
30+
let off = 0
31+
for (const c of chunks) {
32+
out.set(c, off)
33+
off += c.byteLength
34+
}
35+
return out
36+
}
37+
2738
/**
28-
* Build the rewritten tarball for a large non-preview package (a platform
29-
* binary) as a stream: swap only `package/package.json` and pass the multi-MB
30-
* native binary straight through, re-emitting gzip "stored" (uncompressed)
31-
* blocks. Nothing is materialized whole, so it stays within the Worker's 128MB
32-
* memory limit (buffering the ~tens-of-MB decompressed payload to re-tar/re-gzip
33-
* it returns Cloudflare 1102).
39+
* Decompress a gzipped tar into a single right-sized buffer. Presized from the
40+
* gzip ISIZE trailer (decompressed length; valid for the <4GB single-member
41+
* tarballs npm produces) so there is no doubling from a growing collector, and
42+
* read incrementally so the decompressor never holds the whole output on top of
43+
* the buffer. The compressed input goes out of scope on return.
3444
*/
35-
async function buildPlatformTarballStream(
45+
async function decompressToBuffer(gzipped: Uint8Array): Promise<Uint8Array> {
46+
const isize = new DataView(
47+
gzipped.buffer,
48+
gzipped.byteOffset + gzipped.length - 4,
49+
4,
50+
).getUint32(0, true)
51+
52+
const out = new Uint8Array(isize)
53+
let pos = 0
54+
const reader = new Response(gzipped)
55+
.body!.pipeThrough(new DecompressionStream('gzip'))
56+
.getReader()
57+
try {
58+
for (;;) {
59+
const { done, value } = await reader.read()
60+
if (done) break
61+
out.set(value, pos)
62+
pos += value.byteLength
63+
}
64+
} finally {
65+
reader.releaseLock()
66+
}
67+
return pos === isize ? out : out.subarray(0, pos)
68+
}
69+
70+
/**
71+
* Build a platform-binary tarball into R2, then leave it to be served straight
72+
* from R2 (a plain byte passthrough with a Content-Length that cannot be
73+
* truncated, unlike a Worker-generated transform response).
74+
*
75+
* Decompress the upstream ONCE into a single buffer and rewrite package.json in
76+
* place, so the ~tens-of-MB binary is never copied a second time. Then frame the
77+
* buffer as gzip "stored" blocks straight into a multipart upload: because the
78+
* buffer is fully in memory, no decompressor is running during the slow R2 part
79+
* uploads, so nothing buffers ahead. The previous streaming build spiked past
80+
* the 128MB limit (Cloudflare 1102) exactly there, the decompressor ran ahead
81+
* while a part upload was in flight. Peak footprint is ~the decompressed payload
82+
* plus one part, the same envelope as buildMetaLight, which the packument path
83+
* already runs on these binaries. Integrity is not pinned for these packages.
84+
*/
85+
async function buildPlatformTarballToR2(
3686
env: Env,
3787
name: string,
3888
version: string,
39-
): Promise<ReadableStream<Uint8Array>> {
89+
): Promise<void> {
4090
const url = toPkgPrNewUrl(env, name, version)
4191
if (!url) throw new HttpError(400, `Invalid preview version: ${version}`)
4292

43-
const upstream = await fetchUpstreamTarballStream(url, maxTarballBytes(env))
44-
return rewriteTarballEntryStream(
45-
upstream,
93+
const tar = await decompressToBuffer(
94+
await fetchUpstreamTarball(url, maxTarballBytes(env)),
95+
)
96+
rewriteTarEntryInPlace(
97+
tar,
4698
PACKAGE_JSON_NAMES,
4799
(data) => {
48100
let pkg: Record<string, any>
@@ -55,39 +107,7 @@ async function buildPlatformTarballStream(
55107
},
56108
assertSafeTarballPath,
57109
)
58-
}
59-
60-
function mergeChunks(chunks: Uint8Array[], total: number): Uint8Array {
61-
if (chunks.length === 1) return chunks[0]
62-
const out = new Uint8Array(total)
63-
let off = 0
64-
for (const c of chunks) {
65-
out.set(c, off)
66-
off += c.byteLength
67-
}
68-
return out
69-
}
70110

71-
/**
72-
* Build a platform-binary tarball into R2 via a multipart upload, then leave it
73-
* to be served straight from R2.
74-
*
75-
* This is the only shape that satisfies both constraints: the build STREAMS
76-
* (the stored-gzip consumer keeps pace with the decompressor, so the ~tens-of-MB
77-
* payload is never held whole, which a buffered re-tar/re-gzip would and OOM at
78-
* Cloudflare 1102), and the upload is chunked into bounded ~10MB parts (so it
79-
* never buffers the whole object the way a single put of an unsized stream
80-
* would). Serving the finished object from R2 is a plain byte passthrough with a
81-
* Content-Length, so it cannot be truncated the way a Worker-generated transform
82-
* response can. The work is awaited inside the handler so the Worker pumps it to
83-
* completion. Integrity is not pinned for these packages (see `buildMetaLight`).
84-
*/
85-
async function buildPlatformTarballToR2(
86-
env: Env,
87-
name: string,
88-
version: string,
89-
): Promise<void> {
90-
const stream = await buildPlatformTarballStream(env, name, version)
91111
const upload = await env.TARBALL_CACHE.createMultipartUpload(
92112
tarballKey(name, version),
93113
{
@@ -102,12 +122,9 @@ async function buildPlatformTarballToR2(
102122
const parts: R2UploadedPart[] = []
103123
let pending: Uint8Array[] = []
104124
let pendingLen = 0
105-
const reader = stream.getReader()
106-
for (;;) {
107-
const { done, value } = await reader.read()
108-
if (done) break
109-
pending.push(value)
110-
pendingLen += value.byteLength
125+
await emitStoredGzip(tar, async (chunk) => {
126+
pending.push(chunk)
127+
pendingLen += chunk.byteLength
111128
while (pendingLen >= PART_SIZE) {
112129
const merged = mergeChunks(pending, pendingLen)
113130
parts.push(
@@ -117,7 +134,7 @@ async function buildPlatformTarballToR2(
117134
pending = rest.byteLength ? [rest] : []
118135
pendingLen = rest.byteLength
119136
}
120-
}
137+
})
121138
// The final part may be smaller than PART_SIZE.
122139
if (pendingLen > 0 || parts.length === 0) {
123140
parts.push(
Lines changed: 163 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,163 @@
1+
/**
2+
* In-place tar rewriting and stored-gzip emission over a decompressed buffer.
3+
*
4+
* The platform-binary build decompresses the upstream ONCE into a single buffer
5+
* (within the Worker budget, the packument path already does this via
6+
* buildMetaLight), rewrites `package/package.json` in place, then re-emits the
7+
* buffer as gzip "stored" blocks. Because the buffer is fully in memory while it
8+
* is framed and uploaded, no decompressor is running during the slow R2 part
9+
* uploads, so nothing buffers ahead, which is what made the previous streaming
10+
* build spike past the 128MB limit (Cloudflare 1102) under upload backpressure.
11+
*/
12+
13+
import { HttpError } from '../httpError'
14+
15+
const BLOCK = 512
16+
const NAME_LEN = 100
17+
const SIZE_OFF = 124
18+
const SIZE_LEN = 12
19+
const CHKSUM_OFF = 148
20+
const CHKSUM_LEN = 8
21+
const MAGIC_OFF = 257
22+
const PREFIX_OFF = 345
23+
const PREFIX_LEN = 155
24+
25+
const textDecoder = new TextDecoder()
26+
27+
function isZeroBlock(tar: Uint8Array, off: number): boolean {
28+
for (let i = off; i < off + BLOCK; i++) if (tar[i] !== 0) return false
29+
return true
30+
}
31+
32+
function readCStr(tar: Uint8Array, off: number, len: number): string {
33+
let end = 0
34+
while (end < len && tar[off + end] !== 0) end++
35+
return textDecoder.decode(tar.subarray(off, off + end))
36+
}
37+
38+
function readName(tar: Uint8Array, off: number): string {
39+
const name = readCStr(tar, off, NAME_LEN)
40+
const isUstar =
41+
tar[off + MAGIC_OFF] === 0x75 &&
42+
tar[off + MAGIC_OFF + 1] === 0x73 &&
43+
tar[off + MAGIC_OFF + 2] === 0x74 &&
44+
tar[off + MAGIC_OFF + 3] === 0x61 &&
45+
tar[off + MAGIC_OFF + 4] === 0x72
46+
if (isUstar) {
47+
const prefix = readCStr(tar, off + PREFIX_OFF, PREFIX_LEN)
48+
if (prefix) return `${prefix}/${name}`
49+
}
50+
return name
51+
}
52+
53+
function readSize(tar: Uint8Array, off: number): number {
54+
if (tar[off + SIZE_OFF] & 0x80) {
55+
throw new HttpError(422, 'Unsupported base-256 size field in tarball')
56+
}
57+
let value = 0
58+
for (let i = off + SIZE_OFF; i < off + SIZE_OFF + SIZE_LEN; i++) {
59+
const c = tar[i]
60+
if (c === 0 || c === 0x20) continue
61+
value = value * 8 + (c - 0x30)
62+
}
63+
return value
64+
}
65+
66+
/** Rewrite the header in place for a new (smaller-or-equal) entry size. */
67+
function writeHeader(tar: Uint8Array, off: number, newSize: number): void {
68+
const octal = newSize.toString(8).padStart(SIZE_LEN - 1, '0')
69+
for (let i = 0; i < SIZE_LEN - 1; i++) {
70+
tar[off + SIZE_OFF + i] = octal.charCodeAt(i)
71+
}
72+
tar[off + SIZE_OFF + SIZE_LEN - 1] = 0
73+
74+
for (let i = 0; i < CHKSUM_LEN; i++) tar[off + CHKSUM_OFF + i] = 0x20
75+
let sum = 0
76+
for (let i = 0; i < BLOCK; i++) sum += tar[off + i]
77+
const chk = sum.toString(8).padStart(6, '0')
78+
for (let i = 0; i < 6; i++) tar[off + CHKSUM_OFF + i] = chk.charCodeAt(i)
79+
tar[off + CHKSUM_OFF + 6] = 0
80+
tar[off + CHKSUM_OFF + 7] = 0x20
81+
}
82+
83+
/**
84+
* Overwrite each `replaceNames` entry in `tar` with `replaceWith(originalData)`,
85+
* in place, preserving the tar layout (the replacement is zero-padded to the
86+
* entry's existing block allocation, so the large binary after it never moves).
87+
* Throws if a replacement does not fit (true only for an unexpectedly large
88+
* package.json) or if an entry name fails `validateName`.
89+
*/
90+
export function rewriteTarEntryInPlace(
91+
tar: Uint8Array,
92+
replaceNames: Set<string>,
93+
replaceWith: (data: Uint8Array) => Uint8Array,
94+
validateName?: (name: string) => void,
95+
): void {
96+
let off = 0
97+
while (off + BLOCK <= tar.length) {
98+
if (isZeroBlock(tar, off)) break // end-of-archive
99+
const name = readName(tar, off)
100+
validateName?.(name)
101+
const size = readSize(tar, off)
102+
const padded = Math.ceil(size / BLOCK) * BLOCK
103+
104+
if (replaceNames.has(name)) {
105+
const replacement = replaceWith(tar.subarray(off + BLOCK, off + BLOCK + size))
106+
if (replacement.length > padded) {
107+
throw new HttpError(
108+
500,
109+
'Rewritten package.json does not fit the tarball entry',
110+
)
111+
}
112+
tar.set(replacement, off + BLOCK)
113+
tar.fill(0, off + BLOCK + replacement.length, off + BLOCK + padded)
114+
writeHeader(tar, off, replacement.length)
115+
}
116+
off += BLOCK + padded
117+
}
118+
}
119+
120+
const CRC_TABLE = (() => {
121+
const table = new Uint32Array(256)
122+
for (let n = 0; n < 256; n++) {
123+
let c = n
124+
for (let k = 0; k < 8; k++) c = c & 1 ? 0xedb88320 ^ (c >>> 1) : c >>> 1
125+
table[n] = c >>> 0
126+
}
127+
return table
128+
})()
129+
130+
/**
131+
* Emit `tar` as a valid gzip using "stored" (uncompressed) deflate blocks,
132+
* calling `emit` for each output chunk. Re-emitting stored blocks is near-zero
133+
* CPU and emits views into the (in-memory) input, so the only copies are made by
134+
* the consumer (e.g. assembling R2 multipart parts). `emit` is awaited so the
135+
* consumer can apply backpressure.
136+
*/
137+
export async function emitStoredGzip(
138+
tar: Uint8Array,
139+
emit: (chunk: Uint8Array) => Promise<void>,
140+
): Promise<void> {
141+
// gzip header: magic, deflate method, no flags, no mtime, OS unknown.
142+
await emit(new Uint8Array([0x1f, 0x8b, 0x08, 0, 0, 0, 0, 0, 0, 0xff]))
143+
144+
let crc = 0xffffffff
145+
for (let off = 0; off < tar.length; off += 65535) {
146+
const n = Math.min(65535, tar.length - off)
147+
const seg = tar.subarray(off, off + n)
148+
for (let i = 0; i < n; i++) {
149+
crc = (crc >>> 8) ^ CRC_TABLE[(crc ^ seg[i]) & 0xff]
150+
}
151+
// Stored block header: BFINAL/BTYPE byte (0) + LEN + ~LEN, both LE.
152+
await emit(new Uint8Array([0, n & 0xff, (n >> 8) & 0xff, ~n & 0xff, (~n >> 8) & 0xff]))
153+
await emit(seg)
154+
}
155+
156+
// Final empty stored block, then CRC32 + ISIZE (both little-endian).
157+
await emit(new Uint8Array([1, 0, 0, 0xff, 0xff]))
158+
const trailer = new Uint8Array(8)
159+
const dv = new DataView(trailer.buffer)
160+
dv.setUint32(0, (crc ^ 0xffffffff) >>> 0, true)
161+
dv.setUint32(4, tar.length >>> 0, true)
162+
await emit(trailer)
163+
}

0 commit comments

Comments
 (0)