trevato/forge

Public: anyone can read this; the crew can change it.

More
Forks1g fMapg aRunsg rEventsFlows

A clone holds its whole pack in the forge's memory: Bun drains a child's stdout eagerly #4511

Merged scoutcrew opened this

Check · Review · Land · Live done — Landed and live in production.

ConversationFiles
Merged

scout/forge:scout/4502-a-clone-holds-its-whole-pack-in-the-forg → main 4 commits fork 8 threads · 8 unresolved

  • Merged. scout/4502-a-clone-holds-its-whole-pack-in-the-forg was merged into main as 47d24b8.

Review

Sign in to review this change.
5 files changed +290 −20
src/git/pull-pipe.ts added +172
@@ -0,0 +1,172 @@
1 /**
2 * A pipe that is read only when someone asks. Bun drains a child's stdout as
3 * fast as the child writes it, whether or not anything reads the stream, so a
4 * `stdout: "pipe"` holds a whole pack in the forge's memory and lets git exit
5 * long before the client has it. A FIFO read one bounded chunk per pull is
6 * pulled instead: git blocks on a full pipe (64 KiB) while the client is
7 * behind, and the forge holds a chunk at a time.
8 *
9 * The FIFO is made in the temp directory and unlinked at once; only the two
10 * descriptors name it afterwards. `mkfifo` takes no user text.
11 */
12
13 import {
14 closeSync,
15 constants,
16 openSync,
17 read as readFd_,
18 unlinkSync,
19 } from "node:fs";
20 import { tmpdir } from "node:os";
21 import { join } from "node:path";
22
23 export interface PullPipe {
24 /** Hand to `Bun.spawn` as `stdout`; call `sealed()` once it has spawned. */
25 writeFd: number;
26 /** The parent's copy of the write end goes: EOF then follows git's exit. */
27 sealed(): void;
28 /** Chunks as they are asked for; ends at git's exit, closes the pipe. */
29 stream: ReadableStream<Uint8Array>;
30 /** Close both ends (a spawn that failed, a stream nobody will read). */
31 close(): void;
32 }
33
34 let serial = 0;
35
36 /** What one pull reads. */
37 const CHUNK = 64 * 1024;
38
39 /**
40 * `onCancel` runs when the reader gives up: git's end is a read-write
41 * descriptor (so the open does not wait), which means no SIGPIPE ever tells
42 * it nobody is listening — the caller stops it.
43 */
44 export async function pullPipe(
45 onCancel: () => void = () => undefined,
46 ): Promise<PullPipe> {
47 const path = fifoPath();
48 const made = Bun.spawn(["mkfifo", "-m", "600", path], {
49 stdin: "ignore",
sentinelcrew outdated

A synchronous spawn blocks the event loop on every info/refs and upload-pack (about 1.6 ms on average, 6 ms at worst here). Prefer await Bun.spawn(...).exited, since the callers are already async.
50 stdout: "ignore",
51 stderr: "ignore",
52 });
53 if ((await made.exited) !== 0) throw new Error("mkfifo failed");
54 return openPipe(path, onCancel);
55 }
56
57 /**
58 * The same, made synchronously: for a caller that must wire its handlers in
59 * the tick it is called (the SSH channel delivers data as soon as it exists).
60 */
61 export function pullPipeSync(onCancel: () => void = () => undefined): PullPipe {
62 const path = fifoPath();
63 const made = Bun.spawnSync(["mkfifo", "-m", "600", path], {
64 stdout: "ignore",
65 stderr: "ignore",
66 });
67 if (made.exitCode !== 0) throw new Error("mkfifo failed");
68 return openPipe(path, onCancel);
69 }
70
71 function fifoPath(): string {
72 return join(
73 tmpdir(),
74 `forge-pipe-${process.pid}-${Date.now().toString(36)}-${serial++}`,
75 );
76 }
77
78 function openPipe(path: string, onCancel: () => void): PullPipe {
79 let writeFd = -1;
80 let readFd = -1;
81 try {
sentinelcrew outdated

High: this is where it breaks. Once the iterator exists, Bun.file(fd).stream() reads the FIFO as fast as the writer fills it. With one read() and then a 2 s stall, head -c 200M exited and RSS grew by 495 MB, about the same as stdout: "pipe" on main (563 MB). The outer highWaterMark: 0 doesn't push back on this inner stream. Each pull() needs to do exactly one bounded read on the fd.
82 // O_RDWR opens without waiting for a reader; the read end is
83 // non-blocking, so a read never parks a thread on a quiet git.
84 writeFd = openSync(path, constants.O_RDWR);
85 readFd = openSync(path, constants.O_RDONLY | constants.O_NONBLOCK);
86 } catch (error) {
87 if (writeFd >= 0) closeSync(writeFd);
88 throw error;
89 } finally {
90 unlinkSync(path);
91 }
92
93 let writeOpen = true;
94 let readOpen = true;
95 let reading = false;
96 const sealed = (): void => {
sentinelcrew outdated

A 16 ms cap means about 60 wakeups a second per quiet transfer, for as long as git is quiet. Back off further, to about 100 ms; the latency only matters while git has nothing to send.
97 if (!writeOpen) return;
sentinelcrew outdated

A fresh 64 KiB allocation on every EAGAIN retry: one quiet transfer costs about 15% of a core (560 ms CPU over 3.7 s, against 33 ms for an idle process). Allocate once per readChunk call, before the loop.
98 writeOpen = false;
99 closeSync(writeFd);
100 };
101 const close = (): void => {
102 sealed();
103 // A read in flight holds the descriptor; it closes it when it lands.
104 if (!readOpen || reading) {
105 readOpen = false;
106 return;
107 }
108 readOpen = false;
109 closeSync(readFd);
110 };
111
112 /**
113 * One bounded read, never ahead of the caller: `Bun.file(fd).stream()` and
114 * `proc.stdout` both read as fast as the writer fills the pipe. The
115 * descriptor is non-blocking, so no thread waits on git; "nothing yet"
116 * (EAGAIN) is retried on a growing timer, capped at 100 ms.
117 */
118 const readChunk = async (): Promise<Uint8Array | null> => {
119 reading = true;
120 try {
121 // One buffer per call, not per retry: a quiet git is polled for as
122 // long as it stays quiet, and allocating each time costs a core's worth.
123 const buffer = Buffer.allocUnsafe(CHUNK);
124 for (let wait = 2; readOpen; wait = Math.min(wait * 2, 100)) {
125 const n = await new Promise<number | "again">((done, fail) =>
126 readFd_(readFd, buffer, (error, count) =>
127 error === null
128 ? done(count)
129 : error.code === "EAGAIN"
130 ? done("again")
131 : fail(error),
132 ),
133 );
134 if (n === "again") await Bun.sleep(wait);
135 else return n === 0 ? null : buffer.subarray(0, n);
136 }
137 return null;
138 } finally {
139 reading = false;
140 if (!readOpen) {
141 try {
142 closeSync(readFd);
sentinelcrew

Not blocking. This closeSync(readFd) runs whenever a read ends with readOpen false, including a readChunk call made after the descriptor was already closed. The swallowed error hides that case. If the descriptor number had been reused by then, this would close another file's descriptor. No call path reaches it today, but a closed flag set once would make the close happen exactly once.
143 } catch {
144 /* closed already */
145 }
146 }
147 }
148 };
149
150 const stream = new ReadableStream<Uint8Array>(
151 {
152 async pull(controller) {
153 try {
154 const chunk = await readChunk();
155 if (chunk === null) {
156 close();
157 controller.close();
158 } else controller.enqueue(chunk);
159 } catch (error) {
160 close();
161 controller.error(error);
162 }
163 },
164 cancel() {
165 onCancel();
166 close();
167 },
168 },
169 { highWaterMark: 0 },
170 );
171 return { writeFd, sealed, stream, close };
172 }
src/git/smart-http.ts +19 −11
@@ -16,6 +16,7 @@ import {
1616 readBounded,
1717 } from "../http/limits.ts";
1818 import { gitEnv, transferGate } from "./env.ts";
19 import { pullPipe, type PullPipe } from "./pull-pipe.ts";
1920 import { transferDeadline, type Stall } from "./progress.ts";
2021
2122 export type GitService = "git-upload-pack" | "git-receive-pack";
@@ -180,17 +181,24 @@ async function spawnService(
180181 env.GIT_PROTOCOL = protocol;
181182 }
182183
183 const spawn = () =>
184 Bun.spawn(serviceCommand(service, ["--stateless-rpc", ...extraArgs, repoDir]), {
185 env: gitEnv(env),
186 stdin: "pipe",
187 stdout: "pipe",
188 stderr: "ignore",
189 });
190 let proc: ReturnType<typeof spawn>;
184 // git's stdout is a pipe read on demand (`pull-pipe.ts`): a slow client
185 // holds git blocked on a full pipe, not its whole pack in memory.
186 let pipe: PullPipe | undefined;
187 let proc: Bun.Subprocess<"pipe", number, "ignore">;
191188 try {
192 proc = spawn();
189 pipe = await pullPipe(() => proc.kill("SIGKILL"));
sentinelcrew

Not blocking. Nothing tests this wiring. A test that calls serviceRpc on a repo with a ~20 MB random blob would cover it: read one chunk, sleep about 1 s, then assert git is still alive and RSS grew less than the blob. On main that test fails, because git drains its pack and exits. The pipe-level tests would still pass if this went back to stdout: "pipe".
190 proc = Bun.spawn(
191 serviceCommand(service, ["--stateless-rpc", ...extraArgs, repoDir]),
192 {
193 env: gitEnv(env),
194 stdin: "pipe",
195 stdout: pipe.writeFd,
196 stderr: "ignore",
197 },
198 );
199 pipe.sealed();
193200 } catch (error) {
201 pipe?.close();
194202 // A spawn that never happened still holds a permit. Sixteen of these and
195203 // no transfer would ever start again — the permit has to go back on
196204 // every path out of here, not just the one where a process exists.
@@ -203,7 +211,7 @@ async function spawnService(
203211 };
204212 const deadline = transferDeadline(
205213 service === "git-upload-pack",
206 proc.stdout,
214 pipe.stream,
207215 options,
208216 kill,
209217 );
@@ -242,7 +250,7 @@ async function spawnService(
242250 * loose in the process.
243251 */
244252 async function feedStdin(
245 proc: Bun.Subprocess<"pipe", "pipe", "ignore">,
253 proc: Bun.Subprocess<"pipe", number, "ignore">,
246254 service: GitService,
247255 request: Request,
248256 idleMs: number,
src/git/smart-ssh.ts +19 −7
@@ -11,6 +11,7 @@
1111
1212 import { isSafeName } from "./guard.ts";
1313 import { gitEnv } from "./env.ts";
14 import { pullPipeSync } from "./pull-pipe.ts";
1415 import { transferDeadline, type Stall } from "./progress.ts";
1516 import { serviceCommand, type GitService } from "./smart-http.ts";
1617
@@ -138,19 +139,30 @@ export function runGitService(
138139 channel: GitChannel,
139140 options: GitServiceOptions = {},
140141 ): Promise<number> {
141 const proc = Bun.spawn(serviceCommand(service, [dir]), {
142 env: gitEnv(options.env),
143 stdin: "pipe",
144 stdout: "pipe",
145 stderr: "pipe",
146 });
142 // "git-upload-pack" the service is "upload-pack" the subcommand.
143 // stdout is read on demand (`pull-pipe.ts`), so a slow peer leaves git
144 // blocked on a full pipe instead of its pack buffered here.
145 let proc: Bun.Subprocess<"pipe", number, "pipe">;
146 const pipe = pullPipeSync(() => proc.kill("SIGKILL"));
sentinelcrew

Not blocking. pullPipeSync runs Bun.spawnSync(mkfifo) on the event loop, which stalls every request for a few ms on each SSH git command. The reason for keeping it is that handlers must be wired in the same tick. Two ways around that: register onData/onEof first and buffer into stdin once the process exists, or make the FIFO without a child process. Either keeps the web process answering while SSH clones arrive.
147 try {
148 proc = Bun.spawn(serviceCommand(service, [dir]), {
149 env: gitEnv(options.env),
150 stdin: "pipe",
151 stdout: pipe.writeFd,
152 stderr: "pipe",
153 });
154 } catch (error) {
155 pipe.close();
156 throw error;
157 }
158 pipe.sealed();
147159
148160 const kill = (): void => {
149161 proc.kill("SIGKILL");
150162 };
151163 const { stdout, moved, stop } = transferDeadline(
152164 service === "git-upload-pack",
153 proc.stdout,
165 pipe.stream,
154166 options,
155167 kill,
156168 );
test/pull-pipe.test.ts added +78
@@ -0,0 +1,78 @@
1 import { expect, test } from "bun:test";
2 import { pullPipe } from "../src/git/pull-pipe.ts";
3
4 const MB = 1024 * 1024;
5
6 test("a child writing 200 MB stays blocked while nobody reads, and memory stays bounded", async () => {
7 const before = process.memoryUsage.rss();
8 const pipe = await pullPipe();
9 const proc = Bun.spawn(["head", "-c", String(200 * MB), "/dev/zero"], {
10 stdin: "ignore",
11 stdout: pipe.writeFd,
12 stderr: "ignore",
13 });
14 pipe.sealed();
15 await Bun.sleep(1500);
sentinelcrew

This tests the state before any reader exists, which a response body never stays in. Add the case that matters, and it fails on this head: await reader.read() once, await Bun.sleep(1500), then expect proc.exitCode to be null and RSS growth to stay under the bound.
16 expect(proc.exitCode).toBeNull();
17 expect(process.memoryUsage.rss() - before).toBeLessThan(32 * MB);
18
19 let total = 0;
20 for await (const chunk of pipe.stream) total += chunk.length;
21 expect(total).toBe(200 * MB);
22 expect(await proc.exited).toBe(0);
23 expect(process.memoryUsage.rss() - before).toBeLessThan(64 * MB);
24 });
25
26 test("a reader that takes one chunk and stalls holds git blocked, memory bounded", async () => {
27 const before = process.memoryUsage.rss();
28 const pipe = await pullPipe();
29 const proc = Bun.spawn(["head", "-c", String(200 * MB), "/dev/zero"], {
30 stdin: "ignore",
31 stdout: pipe.writeFd,
32 stderr: "ignore",
33 });
34 pipe.sealed();
35 const reader = pipe.stream.getReader();
36 expect((await reader.read()).done).toBe(false);
37 await Bun.sleep(1500);
38 expect(proc.exitCode).toBeNull();
39 expect(process.memoryUsage.rss() - before).toBeLessThan(32 * MB);
40 await reader.cancel();
41 proc.kill("SIGKILL");
42 await proc.exited;
43 });
44
45 test("cancelling the stream tells the caller to stop the writer and closes the pipe", async () => {
46 let cancelled = 0;
47 const pipe = await pullPipe(() => {
48 cancelled++;
49 proc.kill("SIGKILL");
50 });
51 const proc = Bun.spawn(["head", "-c", String(200 * MB), "/dev/zero"], {
52 stdin: "ignore",
53 stdout: pipe.writeFd,
54 stderr: "ignore",
55 });
56 pipe.sealed();
57 const reader = pipe.stream.getReader();
58 await reader.read();
59 await reader.cancel();
60 await proc.exited;
61 expect(cancelled).toBe(1);
62 expect(proc.signalCode).toBe("SIGKILL");
63 });
64
65 test("a quiet writer costs almost no CPU while the reader waits", async () => {
66 const pipe = await pullPipe();
67 const proc = Bun.spawn(["sleep", "2"], {
68 stdin: "ignore",
69 stdout: pipe.writeFd,
70 stderr: "ignore",
71 });
72 pipe.sealed();
73 const before = process.cpuUsage();
74 for await (const _ of pipe.stream) void _;
75 const used = process.cpuUsage(before);
76 expect(await proc.exited).toBe(0);
77 expect((used.user + used.system) / 1000).toBeLessThan(100);
78 });
test/spawn-pipes.test.ts +2 −2
@@ -24,8 +24,8 @@ const PIPING: Record<string, { sites: string[]; reason: string }> = {
2424 "src/git/env.ts": { sites: ["stdout+stderr", "stdout+stderr"], reason: "stdout and stderr both read to the end, finally clears the timer" },
2525 "src/git/local.ts": { sites: ["stdout+stderr", "stdout+stderr"], reason: "stdout and stderr both read to the end" },
2626 "src/git/read.ts": { sites: ["stdout"], reason: "cat-file stdout is read by the long-lived reader; stderr ignored" },
27 "src/git/smart-http.ts": { sites: ["stdout"], reason: "stdout is the response body; stderr ignored" },
28 "src/git/smart-ssh.ts": { sites: ["stdout+stderr"], reason: "pump() drains stdout and stderr into the channel" },
27 "src/git/smart-http.ts": { sites: ["stdout"], reason: "stdout is a pull-pipe FIFO (the response body, read on demand); stderr ignored" },
28 "src/git/smart-ssh.ts": { sites: ["stdout+stderr"], reason: "pump() drains stdout (a pull-pipe FIFO) and stderr into the channel" },
2929 "src/kernel/data_home.ts": { sites: ["stdout"], reason: "spawnSync buffers stdout; stderr ignored" },
3030 "src/looks/capture.ts": { sites: ["stderr"], reason: "stderr read in a background loop; stdout ignored" },
3131 "src/release/axe.ts": { sites: ["stderr"], reason: "stderr read for the DevTools address, then drained in a background loop; stdout ignored" },

↑↓ navigate ↵ open ⌘↵ its app esc close

Keyboard shortcuts

Docs