| 1 | import { execFileSync } from "node:child_process"; |
| 2 | import { randomBytes } from "node:crypto"; |
| 3 | import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; |
| 4 | import { tmpdir } from "node:os"; |
| 5 | import { join } from "node:path"; |
| 6 | import { createHash } from "node:crypto"; |
| 7 | import { deflateSync, gzipSync, inflateSync } from "node:zlib"; |
| 8 | import { setFlagsFromString } from "node:v8"; |
| 9 | import { runInNewContext } from "node:vm"; |
| 10 | import { afterAll, describe, expect, it, vi } from "vitest"; |
| 11 | import { ENTRY_OVERHEAD, HASH_COPY_MAX, REF_DELTA_OVERHEAD, SCRATCH_BYTES, StreamedPack, TooMuchMemory, deflate, deflateChunks, hashObject, parsePack, writePack, type ChunkFeed, type GitObject, type PackSource, type StreamedObject } from "../src/index.js"; |
| 12 | |
| 13 | const dir = mkdtempSync(join(tmpdir(), "pack-stream-")); |
| 14 | afterAll(() => rmSync(dir, { recursive: true, force: true })); |
| 15 | const git = (args: string[], input?: string) => execFileSync("git", args, { cwd: dir, input, maxBuffer: 1 << 28, env: { ...process.env, GIT_AUTHOR_NAME: "a", GIT_AUTHOR_EMAIL: "a@x", GIT_COMMITTER_NAME: "a", GIT_COMMITTER_EMAIL: "a@x" } }); |
| 16 | |
| 17 | // A history whose files change a little each commit, so the pack has delta chains, plus a large blob. |
| 18 | git(["init", "-q", "-b", "main"]); |
| 19 | let text = Array.from({ length: 400 }, (_, i) => `line ${i} of a file that keeps changing\n`).join(""); |
| 20 | // Random bytes, so the pack itself is several blocks long. |
| 21 | const large = new Uint8Array(randomBytes(5 * 1024 * 1024)); |
| 22 | for (let c = 0; c < 12; c++) { |
| 23 | text = text.replace(`line ${c * 3}`, `changed ${c}`) + `appended ${c}\n`; |
| 24 | writeFileSync(join(dir, "a.txt"), text); |
| 25 | writeFileSync(join(dir, "b.txt"), text.split("").reverse().join("")); |
| 26 | if (c === 5) writeFileSync(join(dir, "big.bin"), large); |
| 27 | git(["add", "-A"]); |
| 28 | git(["commit", "-q", "-m", `c${c}`]); |
| 29 | } |
| 30 | const pack = (args: string[], revs: string) => new Uint8Array(git(["pack-objects", "--stdout", "-q", ...args], revs)); |
| 31 | const full = pack(["--revs"], "main\n"); |
| 32 | const thin = pack(["--revs", "--thin"], "main\n^main~4\n"); |
| 33 | |
| 34 | /** Reads count how many bytes were fetched, to show the pack is read in ranges. */ |
| 35 | function source(data: Uint8Array, reads: { bytes: number; calls: number } = { bytes: 0, calls: 0 }): PackSource { |
| 36 | return { |
| 37 | size: data.length, |
| 38 | read: async (offset, length) => { |
| 39 | reads.calls++; |
| 40 | reads.bytes += length; |
| 41 | return data.slice(offset, offset + length); |
| 42 | }, |
| 43 | }; |
| 44 | } |
| 45 | |
| 46 | /** A pack of deltas, each against the base with this oid (from outside the pack): `deltas[i]` is its instructions. */ |
| 47 | function refDeltaPack(baseOid: string, deltas: Uint8Array[]): Uint8Array { |
| 48 | const parts: Buffer[] = [Buffer.from("PACK"), Buffer.from([0, 0, 0, 2]), Buffer.alloc(4)]; |
| 49 | parts[2]!.writeUInt32BE(deltas.length); |
| 50 | for (const d of deltas) { |
| 51 | const head: number[] = []; |
| 52 | let size = d.length; |
| 53 | let c = (7 << 4) | (size & 15); |
| 54 | size = Math.floor(size / 16); |
| 55 | while (size > 0) { |
| 56 | head.push(c | 0x80); |
| 57 | c = size & 0x7f; |
| 58 | size = Math.floor(size / 128); |
| 59 | } |
| 60 | head.push(c); |
| 61 | parts.push(Buffer.from(head), Buffer.from(baseOid, "hex"), deflateSync(d)); |
| 62 | } |
| 63 | const body = Buffer.concat(parts); |
| 64 | return new Uint8Array(Buffer.concat([body, createHash("sha1").update(body).digest()])); |
| 65 | } |
| 66 | |
| 67 | /** REF_DELTAs, each against the oid given, then one blob (or the blob first). */ |
| 68 | function refDeltaChain(deltas: { base: string; delta: Uint8Array }[], blob: Uint8Array, blobFirst = false): Uint8Array { |
| 69 | const parts: Buffer[] = [Buffer.from("PACK"), Buffer.from([0, 0, 0, 2]), Buffer.alloc(4)]; |
| 70 | parts[2]!.writeUInt32BE(deltas.length + 1); |
| 71 | const header = (code: number, size: number) => { |
| 72 | const head: number[] = []; |
| 73 | let c = (code << 4) | (size & 15); |
| 74 | size = Math.floor(size / 16); |
| 75 | while (size > 0) { |
| 76 | head.push(c | 0x80); |
| 77 | c = size & 0x7f; |
| 78 | size = Math.floor(size / 128); |
| 79 | } |
| 80 | head.push(c); |
| 81 | return Buffer.from(head); |
| 82 | }; |
| 83 | const blobPart = [header(3, blob.length), deflateSync(blob)]; |
| 84 | if (blobFirst) parts.push(...blobPart); |
| 85 | for (const d of deltas) parts.push(header(7, d.delta.length), Buffer.from(d.base, "hex"), deflateSync(d.delta)); |
| 86 | if (!blobFirst) parts.push(...blobPart); |
| 87 | const body = Buffer.concat(parts); |
| 88 | return new Uint8Array(Buffer.concat([body, createHash("sha1").update(body).digest()])); |
| 89 | } |
| 90 | |
| 91 | const varint = (n: number) => { |
| 92 | const out: number[] = []; |
| 93 | do { |
| 94 | let b = n & 0x7f; |
| 95 | n = Math.floor(n / 128); |
| 96 | if (n > 0) b |= 0x80; |
| 97 | out.push(b); |
| 98 | } while (n > 0); |
| 99 | return out; |
| 100 | }; |
| 101 | |
| 102 | /** A delta that copies one byte of the base at `at`, then inserts `tag`. */ |
| 103 | function tinyDelta(baseSize: number, at: number, tag: number): Uint8Array { |
| 104 | return new Uint8Array([...varint(baseSize), ...varint(2), 0x80 | 0x0f | 0x10, at & 0xff, (at >> 8) & 0xff, (at >> 16) & 0xff, (at >> 24) & 0xff, 1, 1, tag]); |
| 105 | } |
| 106 | |
| 107 | async function unpack(data: Uint8Array, opts: Parameters<typeof StreamedPack.open>[1] = {}) { |
| 108 | const seen = new Map<string, { type: string; content: Uint8Array }>(); |
| 109 | const visit = async (o: StreamedObject, content: Uint8Array | null, feed: ChunkFeed | null) => { |
| 110 | let bytes = content; |
| 111 | if (!bytes) { |
| 112 | const parts: Uint8Array[] = []; |
| 113 | await feed!((c) => parts.push(c.slice())); |
| 114 | bytes = new Uint8Array(parts.reduce((n, p) => n + p.length, 0)); |
| 115 | let at = 0; |
| 116 | for (const p of parts) bytes.set(p, (at += p.length) - p.length); |
| 117 | } |
| 118 | expect(bytes.length).toBe(o.size); |
| 119 | seen.set(o.oid, { type: o.type, content: bytes }); |
| 120 | }; |
| 121 | const p = await StreamedPack.open(source(data), { largeObjectBytes: 4 * 1024 * 1024, blockBytes: 64 * 1024, maxBlocks: 4, cacheBytes: 256 * 1024, ...opts }, visit); |
| 122 | return { p, seen }; |
| 123 | } |
| 124 | |
| 125 | describe("StreamedPack", () => { |
| 126 | it("finds the same objects, with the same contents, as parsePack", async () => { |
| 127 | const want = await parsePack(full); |
| 128 | const { p, seen } = await unpack(full); |
| 129 | expect(p.count).toBe(want.length); |
| 130 | expect(seen.size).toBe(new Set(want.map((o) => o.oid)).size); |
| 131 | for (const o of want) { |
| 132 | expect(seen.get(o.oid)?.type).toBe(o.type); |
| 133 | expect(Buffer.from(seen.get(o.oid)!.content).equals(Buffer.from(o.content))).toBe(true); |
| 134 | } |
| 135 | // Every object comes back again for storing: a delta's content rebuilt, any other as its zlib data's place in the pack. |
| 136 | const listed = [...p.objects()]; |
| 137 | expect(listed.map((o) => o.oid).sort()).toEqual(want.map((o) => o.oid).sort()); |
| 138 | const big = listed.find((o) => o.size === large.length)!; |
| 139 | expect(big.deflated).toBeDefined(); |
| 140 | const deltas = listed.filter((x) => !x.deflated); |
| 141 | expect(deltas.length).toBeGreaterThan(0); |
| 142 | for (const o of [...deltas.slice(0, 25), ...listed.filter((x) => x.deflated && x !== big).slice(0, 25)]) { |
| 143 | expect(Buffer.from(await p.content(o.entry)).equals(Buffer.from(seen.get(o.oid)!.content))).toBe(true); |
| 144 | } |
| 145 | for (const o of listed.filter((x) => x.deflated && x !== big).slice(0, 25)) { |
| 146 | const parts: Uint8Array[] = []; |
| 147 | for await (const s of p.slices(o.deflated!.at, o.deflated!.length)) parts.push(s.slice()); |
| 148 | expect(Buffer.from(inflateSync(Buffer.concat(parts))).equals(Buffer.from(seen.get(o.oid)!.content))).toBe(true); |
| 149 | } |
| 150 | let zlib = 0; |
| 151 | for await (const s of p.slices(big.deflated!.at, big.deflated!.length)) zlib += s.length; |
| 152 | expect(zlib).toBe(big.deflated!.length); |
| 153 | }); |
| 154 | |
| 155 | it("resolves a thin pack's bases from outside the pack", async () => { |
| 156 | const stored = new Map<string, GitObject>(); |
| 157 | for (const o of await parsePack(full)) stored.set(o.oid, { type: o.type, content: o.content }); |
| 158 | const external = async (oid: string) => stored.get(oid) ?? null; |
| 159 | const want = await parsePack(thin, external); |
| 160 | const { seen } = await unpack(thin, { external }); |
| 161 | expect(seen.size).toBe(want.length); |
| 162 | for (const o of want) expect(Buffer.from(seen.get(o.oid)!.content).equals(Buffer.from(o.content))).toBe(true); |
| 163 | await expect(unpack(thin)).rejects.toThrow(/unresolvable delta base/); |
| 164 | }); |
| 165 | |
| 166 | it("reads the pack in ranges and keeps the #125 budget", async () => { |
| 167 | const reads = { bytes: 0, calls: 0 }; |
| 168 | await StreamedPack.open(source(full, reads), { blockBytes: 64 * 1024, maxBlocks: 2 }, () => {}); |
| 169 | expect(reads.calls).toBeGreaterThan(1); |
| 170 | // The same limit parsePack applies, counted the same way (deltas are resolved in another order, |
| 171 | // so the one that goes over can differ). |
| 172 | const over = /delta result of \d+ bytes is larger than the [\d.]+ bytes allowed|pack unpacks to more than/; |
| 173 | await expect(parsePack(full, undefined, { maxBytes: 5.2 * 1024 * 1024, largeObjectBytes: 4 * 1024 * 1024 })).rejects.toThrow(over); |
| 174 | await expect(unpack(full, { maxBytes: 5.2 * 1024 * 1024 })).rejects.toThrow(over); |
| 175 | await expect(unpack(full, { maxWork: 1024 * 1024 })).rejects.toThrow(/split it into smaller pushes/); |
| 176 | await expect(unpack(full, { maxObjects: 3 })).rejects.toThrow(/split it into smaller pushes/); |
| 177 | }); |
| 178 | |
| 179 | it("refuses a corrupt pack", async () => { |
| 180 | const bad = full.slice(); |
| 181 | bad[bad.length - 1]! ^= 1; |
| 182 | await expect(unpack(bad)).rejects.toThrow(/checksum mismatch/); |
| 183 | const longer = new Uint8Array(full.length + 5); |
| 184 | longer.set(full.subarray(0, full.length - 20)); |
| 185 | longer.set(full.subarray(full.length - 20), full.length + 5 - 20); |
| 186 | await expect(unpack(longer)).rejects.toThrow(/trailing garbage|zlib|bad object|size mismatch|truncated/); |
| 187 | await expect(unpack(new TextEncoder().encode("NOPE".padEnd(40, "x")))).rejects.toThrow(/not a packfile/); |
| 188 | }); |
| 189 | |
| 190 | it("refuses an object compressed with gzip instead of zlib", async () => { |
| 191 | // git can't read gzip, and the object would be stored as sent. pako takes gzip unless told not |
| 192 | // to, and only stumbles on what follows the stream; so these end exactly where a read does. |
| 193 | const content = new TextEncoder().encode("hello\n".repeat(100)); |
| 194 | const gzipped = new Uint8Array(gzipSync(content)); |
| 195 | const gz = await writePack([{ type: "blob", size: content.length, deflated: gzipped }]); |
| 196 | const blockBytes = 14 + gzipped.length; |
| 197 | await expect(unpack(gz, { blockBytes })).rejects.toThrow(/zlib error/); |
| 198 | await expect(unpack(gz, { blockBytes, largeObjectBytes: 16 })).rejects.toThrow(/zlib error/); |
| 199 | // In memory, a large blob is inflated a MiB at a time. |
| 200 | let n = 1024 * 1024 - 200; |
| 201 | while (gzipSync(new Uint8Array(n), { level: 0 }).length < 1024 * 1024) n++; |
| 202 | const big = new Uint8Array(randomBytes(n)); |
| 203 | const bigGz = new Uint8Array(gzipSync(big, { level: 0 })); |
| 204 | expect(bigGz.length).toBe(1024 * 1024); |
| 205 | const bigPack = await writePack([{ type: "blob", size: big.length, deflated: bigGz }]); |
| 206 | await expect(parsePack(bigPack, undefined, { largeObjectBytes: 16 })).rejects.toThrow(/zlib error/); |
| 207 | await expect(parsePack(gz)).rejects.toThrow(/zlib error/); |
| 208 | // The same objects in zlib come in. |
| 209 | const zl = await writePack([{ type: "blob", size: content.length, deflated: new Uint8Array(deflateSync(content)) }]); |
| 210 | expect((await unpack(zl)).seen.size).toBe(1); |
| 211 | expect((await unpack(zl, { largeObjectBytes: 16 })).seen.size).toBe(1); |
| 212 | expect(await parsePack(await writePack([{ type: "blob", size: big.length, deflated: new Uint8Array(deflateSync(big)) }]), undefined, { largeObjectBytes: 16 })).toHaveLength(1); |
| 213 | }); |
| 214 | |
| 215 | it("reads a base from outside the pack once for all its deltas, and counts it", async () => { |
| 216 | const base = new Uint8Array(randomBytes(1024 * 1024)); |
| 217 | const oid = await hashObject("blob", base); |
| 218 | let reads = 0; |
| 219 | const external = async (o: string) => (o === oid ? (reads++, { type: "blob" as const, content: base }) : null); |
| 220 | const many = refDeltaPack(oid, Array.from({ length: 200 }, (_, i) => tinyDelta(base.length, i * 4099, i))); |
| 221 | // However small the cache, tiny deltas can't make the server read it again and again. |
| 222 | for (const cacheBytes of [8 * 1024 * 1024, 64 * 1024]) { |
| 223 | reads = 0; |
| 224 | const { p, seen } = await unpack(many, { external, cacheBytes }); |
| 225 | expect(seen.size).toBe(200); |
| 226 | expect(reads).toBe(1); |
| 227 | expect(p.workDone).toBeLessThan(base.length + 64 * 1024); |
| 228 | } |
| 229 | // It counts toward what the pack unpacks to, as in parsePack. |
| 230 | await expect(unpack(many, { external, maxBytes: base.length })).rejects.toThrow(/unpacks to more than/); |
| 231 | await expect(parsePack(many, external, { maxBytes: base.length })).rejects.toThrow(/unpacks to more than/); |
| 232 | }); |
| 233 | |
| 234 | it("resolves a long delta chain in one pass", { timeout: 120_000 }, async () => { |
| 235 | // Each delta's base is listed after it: resolving round by round took a round, and a pass over |
| 236 | // every delta left, per link. |
| 237 | const n = 20_000; |
| 238 | const blob = new TextEncoder().encode("x".repeat(64)); |
| 239 | const contents = [blob]; |
| 240 | for (let i = 1; i <= n; i++) contents.push(new Uint8Array([...blob, i & 0xff, i >> 8])); |
| 241 | const oids = await Promise.all(contents.map((c) => hashObject("blob", c))); |
| 242 | // Copy the 64 bytes the chain shares, then insert the two that number it. |
| 243 | const deltas = contents.slice(1).map((c, i) => ({ base: oids[i]!, delta: new Uint8Array([...varint(contents[i]!.length), ...varint(c.length), 0x90, 64, 2, c[64]!, c[65]!]) })); |
| 244 | const timed = async (pack: Uint8Array) => { |
| 245 | const seen = new Map<string, Uint8Array>(); |
| 246 | const t = performance.now(); |
| 247 | await StreamedPack.open(source(pack), {}, (o, content) => void seen.set(o.oid, content!)); |
| 248 | return { ms: performance.now() - t, seen }; |
| 249 | }; |
| 250 | // Listed base first, every delta is resolvable when it is read: the time one pass takes on this machine. |
| 251 | const onePass = await timed(refDeltaChain(deltas, blob, true)); |
| 252 | const { ms, seen } = await timed(refDeltaChain(deltas.reverse(), blob)); |
| 253 | // Compared on the same machine, so a slower CI host doesn't change the verdict. Round by round, this took most of a minute. |
| 254 | expect(ms).toBeLessThan(4 * onePass.ms + 500); |
| 255 | expect(seen.size).toBe(n + 1); |
| 256 | expect(Buffer.from(seen.get(oids[n]!)!).equals(Buffer.from(contents[n]!))).toBe(true); |
| 257 | }); |
| 258 | |
| 259 | it("refuses a pack whose index alone wouldn't fit in memory", async () => { |
| 260 | // The caches unpack() uses (4 blocks of 64 KiB and 256 KiB), then ENTRY_OVERHEAD for each object. |
| 261 | const caches = 4 * 64 * 1024 + 256 * 1024; |
| 262 | const count = new DataView(full.buffer, full.byteOffset).getUint32(8); |
| 263 | await expect(unpack(full, { maxHeldBytes: caches + (count - 1) * ENTRY_OVERHEAD })).rejects.toThrow(/pack needs more than 0 MiB of memory to unpack; split it into smaller pushes/); |
| 264 | // A REF_DELTA costs more (git pack-objects writes them unless asked for OFS_DELTAs): charged as it is read. |
| 265 | await expect(unpack(full, { maxHeldBytes: caches + count * ENTRY_OVERHEAD })).rejects.toThrow(/split it into smaller pushes/); |
| 266 | // Then the objects themselves while they are in use, and the zlib data kept to inflate again. |
| 267 | expect((await unpack(full, { maxHeldBytes: caches + count * (ENTRY_OVERHEAD + REF_DELTA_OVERHEAD) + 16 * 1024 * 1024 })).seen.size).toBeGreaterThan(0); |
| 268 | }); |
| 269 | |
| 270 | it("charges every object it holds, so one huge object is refused rather than run the isolate out of memory", async () => { |
| 271 | const base = new Uint8Array(randomBytes(64 * 1024)); |
| 272 | const oid = await hashObject("blob", base); |
| 273 | // One delta that copies the 64 KiB base 6,000 times over: a 375 MiB blob from a 50 KB pack. |
| 274 | const ops: number[] = []; |
| 275 | for (let k = 0; k < 6000; k++) ops.push(0x80 | 0x20, 0x00); // copy 0x0000 -> 64 KiB (size 0 means 0x10000) |
| 276 | const result = 6000 * 64 * 1024; |
| 277 | const bomb = refDeltaChain([{ base: oid, delta: new Uint8Array([...varint(base.length), ...varint(result), ...ops]) }], base); |
| 278 | // A tree of zeros: 120 MiB from about 120 KB. |
| 279 | const zeros = await writePack([{ type: "tree", size: 120 * 1024 * 1024, deflated: new Uint8Array(deflateSync(new Uint8Array(120 * 1024 * 1024))) }]); |
| 280 | for (const pack of [bomb, zeros]) { |
| 281 | const before = process.memoryUsage().arrayBuffers; |
| 282 | let peak = before; |
| 283 | const t = setInterval(() => (peak = Math.max(peak, process.memoryUsage().arrayBuffers)), 1); |
| 284 | await expect(StreamedPack.open(source(pack), {}, () => {})).rejects.toThrow(/pack needs more than 80 MiB of memory to unpack; split it into smaller pushes/); |
| 285 | clearInterval(t); |
| 286 | expect(peak - before).toBeLessThan(64 * 1024 * 1024); |
| 287 | await expect(parsePack(pack)).rejects.toThrow(/split it into smaller pushes|larger than/); |
| 288 | } |
| 289 | }); |
| 290 | |
| 291 | it("reads each entry's own range again, keeps small ones from the first pass, and counts its reads", async () => { |
| 292 | const reads = { bytes: 0, calls: 0 }; |
| 293 | const kept = await StreamedPack.open(source(full, reads), { blockBytes: 64 * 1024 }, () => {}); |
| 294 | const keptCalls = reads.calls; |
| 295 | // Without keeping anything, every delta is read again, twice counted (storing replays it), each its own range. |
| 296 | const bare = { bytes: 0, calls: 0 }; |
| 297 | const p = await StreamedPack.open(source(full, bare), { blockBytes: 64 * 1024, keepBytes: 0 }, () => {}); |
| 298 | expect(bare.calls).toBeGreaterThan(keptCalls); |
| 299 | expect(bare.bytes).toBeLessThan(2 * full.length); |
| 300 | expect(p.readsDone).toBeGreaterThan(kept.readsDone); |
| 301 | await expect(StreamedPack.open(source(full), { blockBytes: 64 * 1024, keepBytes: 0, maxReads: p.readsDone - 1 }, () => {})).rejects.toThrow(/reads .* of storage to unpack; split it into smaller pushes/); |
| 302 | // Replaying reads only what was counted. |
| 303 | const counted = p.readsDone; |
| 304 | await p.replayDeltas(async () => {}); |
| 305 | expect(p.readsDone).toBe(counted); |
| 306 | }); |
| 307 | |
| 308 | it("says what storing it adds: each delta's content, not its few bytes in the pack", async () => { |
| 309 | const base = new Uint8Array(randomBytes(64 * 1024)); |
| 310 | const oid = await hashObject("blob", base); |
| 311 | // Each copies the base and adds one byte: 200 different 64 KiB blobs from a pack of a few KB. |
| 312 | const deltas = Array.from({ length: 200 }, (_, k) => ({ base: oid, delta: new Uint8Array([...varint(base.length), ...varint(base.length + 1), 0x80 | 0x20, 0x00, 1, k]) })); |
| 313 | const pack = refDeltaChain(deltas, base); |
| 314 | const p = await StreamedPack.open(source(pack), {}, () => {}); |
| 315 | expect(pack.length).toBeLessThan(200 * 1024); |
| 316 | expect(p.storedBytes).toBeGreaterThanOrEqual(200 * (64 * 1024 + 1)); |
| 317 | }); |
| 318 | |
| 319 | it("rebuilds the deltas for storing with exactly the work that already passed", async () => { |
| 320 | const stored = new Map<string, GitObject>(); |
| 321 | for (const o of await parsePack(full)) stored.set(o.oid, { type: o.type, content: o.content }); |
| 322 | const external = async (oid: string) => stored.get(oid) ?? null; |
| 323 | for (const [data, opts] of [ |
| 324 | [full, {}], |
| 325 | [thin, { external }], |
| 326 | ] as const) { |
| 327 | const unlimited = await unpack(data, opts); |
| 328 | // At the tightest limit that lets the pack in, storing it can't go over: nothing is refused halfway through. |
| 329 | const { p, seen } = await unpack(data, { ...opts, maxWork: unlimited.p.workDone, yieldEvery: 4096 }); |
| 330 | const done = p.workDone; |
| 331 | const rebuilt: string[] = []; |
| 332 | await p.replayDeltas(async (entry, content) => { |
| 333 | expect(Buffer.from(content).equals(Buffer.from(seen.get(p.oidOf(entry))!.content))).toBe(true); |
| 334 | rebuilt.push(p.oidOf(entry)); |
| 335 | }); |
| 336 | expect(p.workDone).toBe(done); |
| 337 | // It lets other work in the isolate run as it goes, as open does. |
| 338 | const timers = vi.spyOn(globalThis, "setTimeout"); |
| 339 | await p.replayDeltas(async () => {}); |
| 340 | expect(timers.mock.calls.length).toBeGreaterThan(1); |
| 341 | timers.mockRestore(); |
| 342 | const deltas = [...p.objects()].filter((x) => !x.deflated); |
| 343 | expect(rebuilt.length).toBe(deltas.length); |
| 344 | expect(new Set(rebuilt)).toEqual(new Set(deltas.map((x) => x.oid))); |
| 345 | } |
| 346 | }); |
| 347 | |
| 348 | it("hashes and compresses a large object without a second copy of it", async () => { |
| 349 | const content = new Uint8Array(randomBytes(8 * 1024 * 1024)); |
| 350 | const want = createHash("sha1").update(`blob ${content.length}\0`).update(content).digest("hex"); |
| 351 | const digest = vi.spyOn(crypto.subtle, "digest"); |
| 352 | expect(await hashObject("blob", content)).toBe(want); |
| 353 | // WebCrypto only hashes one whole buffer, so the header and content would be joined first. |
| 354 | for (const [, data] of digest.mock.calls) expect((data as Uint8Array).byteLength).toBeLessThanOrEqual(HASH_COPY_MAX + 64); |
| 355 | digest.mockRestore(); |
| 356 | expect(await hashObject("blob", content.subarray(0, 1000))).toBe(createHash("sha1").update("blob 1000\0").update(content.subarray(0, 1000)).digest("hex")); |
| 357 | // The same zlib data as deflate, in chunks that are each far smaller than the object. |
| 358 | const chunks = [...deflateChunks(content)]; |
| 359 | expect(Math.max(...chunks.map((c) => c.length))).toBeLessThanOrEqual(64 * 1024); |
| 360 | expect(Buffer.concat(chunks).equals(Buffer.from(deflate(content)))).toBe(true); |
| 361 | // A 50 MiB tree from a 51 KB pack is hashed where it lies: what it holds stays under the cap with the tree in it. |
| 362 | const tree = await writePack([{ type: "tree", size: 50 * 1024 * 1024, deflated: new Uint8Array(deflateSync(new Uint8Array(50 * 1024 * 1024))) }]); |
| 363 | const p = await StreamedPack.open(source(tree), {}, () => {}); |
| 364 | expect(p.peakHeldBytes).toBeLessThan(80 * 1024 * 1024); |
| 365 | expect(p.peakHeldBytes).toBeGreaterThan(50 * 1024 * 1024 + SCRATCH_BYTES); |
| 366 | }); |
| 367 | |
| 368 | it("checks what a base from the repository takes before reading it", async () => { |
| 369 | // A 400 MiB blob of zeros is about 400 KB stored; a thin pack's tiny delta against it would inflate it whole. |
| 370 | const oid = "1".repeat(40); |
| 371 | const asked: number[] = []; |
| 372 | let read = false; |
| 373 | const external = async (o: string, maxBytes: number) => { |
| 374 | if (o !== oid) return null; |
| 375 | asked.push(maxBytes); |
| 376 | if (400 * 1024 * 1024 > maxBytes) return 400 * 1024 * 1024; |
| 377 | read = true; |
| 378 | return { type: "blob" as const, content: new Uint8Array(0) }; |
| 379 | }; |
| 380 | const thinPack = refDeltaPack(oid, [tinyDelta(400 * 1024 * 1024, 0, 1)]); |
| 381 | await expect(StreamedPack.open(source(thinPack), { external }, () => {})).rejects.toThrow(TooMuchMemory); |
| 382 | await expect(parsePack(thinPack, external)).rejects.toThrow(/pack needs more than 80 MiB of memory to unpack; split it into smaller pushes/); |
| 383 | expect(read).toBe(false); |
| 384 | expect(asked).toHaveLength(2); |
| 385 | for (const n of asked) expect(n).toBeLessThan(80 * 1024 * 1024); |
| 386 | }); |
| 387 | |
| 388 | it("rebuilds a long history of one large file holding two versions at a time", { timeout: 60_000 }, async () => { |
| 389 | // A 2.6 MiB lockfile over 60 commits, each version a delta against the one before, as git packs it. |
| 390 | let version = new Uint8Array(Buffer.from(Array.from({ length: 110_000 }, (_, i) => `"pkg-${i}": "1.0.${i % 97}",\n`).join("").slice(0, 2_600_000))); |
| 391 | const blob = version; |
| 392 | const deltas: { base: string; delta: Uint8Array }[] = []; |
| 393 | for (let k = 0; k < 60; k++) { |
| 394 | const n = version.length; |
| 395 | const half = n >> 1; |
| 396 | const copy = (off: number, len: number) => [0x80 | 0x0f | 0x70, off & 0xff, (off >> 8) & 0xff, (off >> 16) & 0xff, (off >>> 24) & 0xff, len & 0xff, (len >> 8) & 0xff, (len >> 16) & 0xff]; |
| 397 | const delta = new Uint8Array([...varint(n), ...varint(n + 1), ...copy(0, half), 1, k, ...copy(half, n - half)]); |
| 398 | const next = new Uint8Array(n + 1); |
| 399 | next.set(version.subarray(0, half)); |
| 400 | next[half] = k; |
| 401 | next.set(version.subarray(half), half + 1); |
| 402 | deltas.push({ base: await hashObject("blob", version), delta }); |
| 403 | version = next; |
| 404 | } |
| 405 | const chain = refDeltaChain(deltas.reverse(), blob); |
| 406 | const last = await hashObject("blob", version); |
| 407 | let seen = false; |
| 408 | const p = await StreamedPack.open(source(chain), {}, (o, content) => { |
| 409 | if (o.oid === last) seen = Buffer.from(content!).equals(Buffer.from(version)); |
| 410 | }); |
| 411 | expect(seen).toBe(true); |
| 412 | // Its caches and scratch, then the version being rebuilt from and the one being rebuilt. |
| 413 | expect(p.peakHeldBytes).toBeLessThan(24 * 1024 * 1024 + 3 * version.length + 1024 * 1024); |
| 414 | let stored = 0; |
| 415 | await p.replayDeltas(async () => void stored++); |
| 416 | expect(stored).toBe(60); |
| 417 | // Parsed whole it holds every version: refused as memory, which a push takes as its cue to unpack it in pieces. |
| 418 | await expect(parsePack(chain)).rejects.toThrow(TooMuchMemory); |
| 419 | }); |
| 420 | |
| 421 | it("frees a chain's first version once the next is rebuilt from it", { timeout: 60_000 }, async () => { |
| 422 | setFlagsFromString("--expose-gc"); |
| 423 | // V8 frees ArrayBuffer memory on a background thread after a collection, so a reading right after |
| 424 | // gc() can still count buffers it just freed: a tenth of readings on a busy host (81.9 MB here on |
| 425 | // CI, one 26 MiB version too many). Swept on this thread, gc() returns with them uncounted. |
| 426 | setFlagsFromString("--no-concurrent-array-buffer-sweeping"); |
| 427 | const gc = runInNewContext("gc") as () => void; |
| 428 | // A 26 MiB text file, then three versions of it, each a small delta against the one before. |
| 429 | const n = 26 * 1024 * 1024; |
| 430 | let version = new Uint8Array(Buffer.from("a line of a large text file\n".repeat(Math.ceil(n / 28)).slice(0, n))); |
| 431 | const blob = version; |
| 432 | const deltas: { base: string; delta: Uint8Array }[] = []; |
| 433 | for (let k = 0; k < 3; k++) { |
| 434 | const len = version.length; |
| 435 | // Copies of at most 8 MiB (a copy's size has three bytes), then the byte that's new. |
| 436 | const ops: number[] = []; |
| 437 | for (let off = 0; off < len; off += 8 << 20) { |
| 438 | const m = Math.min(8 << 20, len - off); |
| 439 | ops.push(0x80 | 0x0f | 0x70, off & 0xff, (off >> 8) & 0xff, (off >> 16) & 0xff, (off >>> 24) & 0xff, m & 0xff, (m >> 8) & 0xff, (m >> 16) & 0xff); |
| 440 | } |
| 441 | const delta = new Uint8Array([...varint(len), ...varint(len + 1), ...ops, 1, k]); |
| 442 | const next = new Uint8Array(len + 1); |
| 443 | next.set(version); |
| 444 | next[len] = k; |
| 445 | deltas.push({ base: await hashObject("blob", version), delta }); |
| 446 | version = next; |
| 447 | } |
| 448 | const chain = new Uint8Array(refDeltaChain(deltas.reverse(), blob)); |
| 449 | const last = await hashObject("blob", version); |
| 450 | version = new Uint8Array(0); |
| 451 | let atLast = 0; |
| 452 | await StreamedPack.open(source(chain), {}, (o) => { |
| 453 | if (o.oid !== last) return; |
| 454 | gc(); |
| 455 | atLast = process.memoryUsage().arrayBuffers; |
| 456 | }); |
| 457 | // Measured against what is left once the unpack is gone. |
| 458 | gc(); |
| 459 | const live = atLast - process.memoryUsage().arrayBuffers; |
| 460 | // At most the version it was rebuilt from and itself, and what the caches hold: not the first |
| 461 | // version as well. (Only an upper bound: when the collector frees what is no longer used varies.) |
| 462 | expect(live).toBeLessThan(2 * n + 8 * 1024 * 1024); |
| 463 | }); |
| 464 | |
| 465 | it("measures what a large rebuilt delta compresses to, for counting and storing it", async () => { |
| 466 | // 40 versions of a 600 KB file that compresses well: each one stored inline, so no write of its own is counted. |
| 467 | let version = new Uint8Array(Buffer.from(Array.from({ length: 30_000 }, (_, i) => `"pkg-${i}": "1.0.${i % 7}",\n`).join("").slice(0, 600_000))); |
| 468 | const blob = version; |
| 469 | const deltas: { base: string; delta: Uint8Array }[] = []; |
| 470 | const versions: Uint8Array[] = []; |
| 471 | for (let k = 0; k < 40; k++) { |
| 472 | const len = version.length; |
| 473 | const delta = new Uint8Array([...varint(len), ...varint(len + 1), 0x80 | 0x70, len & 0xff, (len >> 8) & 0xff, (len >> 16) & 0xff, 1, k]); |
| 474 | const next = new Uint8Array(len + 1); |
| 475 | next.set(version); |
| 476 | next[len] = k; |
| 477 | deltas.push({ base: await hashObject("blob", version), delta }); |
| 478 | versions.push(next); |
| 479 | version = next; |
| 480 | } |
| 481 | const chain = refDeltaChain(deltas.reverse(), blob); |
| 482 | const counted = await StreamedPack.open(source(chain), {}, () => {}); |
| 483 | const p = await StreamedPack.open(source(chain), { writesOver: 512 * 1024 }, () => {}); |
| 484 | expect(p.readsDone).toBe(counted.readsDone); |
| 485 | const byOid = new Map(await Promise.all(versions.map(async (v) => [await hashObject("blob", v), v] as const))); |
| 486 | let measured = 0; |
| 487 | for (const x of p.objects()) { |
| 488 | if (x.deflated) continue; |
| 489 | expect(p.zlibSizeOf(x.entry)).toBe(deflate(byOid.get(x.oid)!).length); |
| 490 | measured++; |
| 491 | } |
| 492 | expect(measured).toBe(40); |
| 493 | // A pack that unpacks to more than it may says so for the whole push, not for one object. |
| 494 | await expect(StreamedPack.open(source(chain), { maxBytes: 4 * 1024 * 1024 }, () => {})).rejects.toThrow("pack unpacks to more than 4 MiB; split it into smaller pushes"); |
| 495 | await expect(parsePack(chain, undefined, { maxBytes: 4 * 1024 * 1024 })).rejects.toThrow("pack unpacks to more than 4 MiB; split it into smaller pushes"); |
| 496 | }); |
| 497 | }); |