diff --git a/apps/server/src/processRunner.ts b/apps/server/src/processRunner.ts index 25ed54e44fc6ffb96dcc397e31d3d30e9e1ceac1..0a9bb9b04a43e20422069e04a30d7850601092c1 100644 --- a/apps/server/src/processRunner.ts +++ b/apps/server/src/processRunner.ts @@ -25,6 +25,8 @@ export interface ProcessRunInput { readonly timeout?: Duration.Input | undefined; readonly env?: NodeJS.ProcessEnv | undefined; readonly stdin?: string | undefined; + /** Receives every stdout chunk, including bytes beyond the buffered output limit. */ + readonly onStdoutChunk?: ((chunk: Uint8Array) => void) | undefined; readonly maxOutputBytes?: number | undefined; readonly outputMode?: "error" | "truncate" | undefined; readonly truncatedMarker?: string | undefined; @@ -327,6 +329,7 @@ const runProcessCore = Effect.fn("processRunner.runProcessCore")(function* ( ); const stdin = input.stdin; + const onStdoutChunk = input.onStdoutChunk; const writeStdin = stdin === undefined ? Effect.void @@ -352,7 +355,9 @@ const runProcessCore = Effect.fn("processRunner.runProcessCore")(function* ( cwd: input.cwd, spawnCwd: input.spawnCwd, streamName: "stdout", - stream: child.stdout, + stream: onStdoutChunk + ? child.stdout.pipe(Stream.tap((chunk) => Effect.sync(() => onStdoutChunk(chunk)))) + : child.stdout, maxOutputBytes, outputMode, truncatedMarker, diff --git a/apps/server/src/vcs/GitVcsDriver.test.ts b/apps/server/src/vcs/GitVcsDriver.test.ts index 9373d8aabdb878e7725616ee056ed88094f1e383..5ccc2c8c6b5bad8ffa706c1d2b9bf1515df55bd2 100644 --- a/apps/server/src/vcs/GitVcsDriver.test.ts +++ b/apps/server/src/vcs/GitVcsDriver.test.ts @@ -119,6 +119,216 @@ it.effect("checkpoint capture does not rerun clean filters for unchanged indexed }).pipe(Effect.scoped, Effect.provide(GitContractLayer)), ); +for (const nested of [false, true]) { + for (const indexState of [ + "sparse", + "flags", + "manual-skip", + "missing", + "non-cone-missing", + ] as const) { + it.effect( + `sparse checkpoint preserves two captures (nested=${nested}, index=${indexState})`, + () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const driver = yield* GitVcsDriver.makeVcsDriverShape(); + const cwd = yield* fs.makeTempDirectoryScoped({ prefix: "t3-checkpoint-sparse-" }); + const { git } = yield* makeCheckpointFixture(driver, cwd); + const write = Effect.fn(function* (name: string, contents: string) { + yield* fs.makeDirectory(path.dirname(path.join(cwd, name)), { recursive: true }); + yield* fs.writeFileString(path.join(cwd, name), contents); + }); + for (const name of [ + "scope/in/edit", + "scope/in/delete", + "scope/out/deep/absent", + "scope/out/present", + "elsewhere/file", + ]) { + yield* write(name, "original\n"); + } + yield* git(["add", "."]); + yield* git(["commit", "-m", "sparse fixture"]); + yield* git([ + "sparse-checkout", + "set", + "--cone", + "--sparse-index", + "scope/in", + "elsewhere", + ]); + if (indexState === "non-cone-missing") + yield* git(["sparse-checkout", "set", "--no-cone", "/scope/in/", "/elsewhere/"]); + yield* write("scope/in/edit", "staged\n"); + yield* write("elsewhere/file", "staged outside\n"); + yield* git(["add", "."]); + if (indexState === "flags") + yield* git(["update-index", "--assume-unchanged", "scope/in/delete"]); + if (indexState === "manual-skip") + yield* git(["update-index", "--skip-worktree", "scope/in/delete"]); + yield* git(["config", "sparse.expectFilesOutsideOfPatterns", "true"]); + yield* write("scope/in/edit", "working\n"); + yield* write("scope/out/present", "modified skipped\n"); + yield* write("scope/out/new file", "new outside cone\n"); + yield* write("elsewhere/file", "working outside\n"); + yield* fs.remove(path.join(cwd, "scope/in/delete")); + const indexPath = path.join(cwd, ".git/index"); + if (indexState.endsWith("missing")) yield* fs.remove(indexPath); + const originalIndex = yield* fs + .readFile(indexPath) + .pipe(Effect.orElseSucceed(() => null)); + const captureCwd = nested ? path.join(cwd, "scope") : cwd; + for (const turn of [1, 2]) { + const ref = CheckpointRef.make(`refs/t3/checkpoints/sparse/${turn}`); + if (turn === 2) { + yield* write("scope/in/edit", "second\n"); + yield* fs.remove(path.join(cwd, "scope/out/new file")); + yield* write("scope/out/second", "second addition\n"); + } + const capture = driver.checkpoints.captureCheckpoint({ + cwd: captureCwd, + checkpointRef: ref, + }); + if (indexState === "non-cone-missing") { + assert.strictEqual((yield* capture.pipe(Effect.flip))._tag, "VcsProcessExitError"); + assert.isFalse( + yield* driver.checkpoints.hasCheckpointRef({ cwd: captureCwd, checkpointRef: ref }), + ); + assert.isFalse(yield* fs.exists(indexPath)); + break; + } + yield* capture; + for (const [name, content] of [ + ["scope/out/deep/absent", "original\n"], + ["scope/out/present", "modified skipped\n"], + ["scope/in/edit", turn === 1 ? "working\n" : "second\n"], + ["elsewhere/file", nested ? "original\n" : "working outside\n"], + [ + turn === 1 ? "scope/out/new file" : "scope/out/second", + turn === 1 ? "new outside cone\n" : "second addition\n", + ], + ]) { + assert.strictEqual((yield* git(["show", `${ref}:${name}`])).stdout, content); + } + const files = (yield* git(["ls-tree", "-rz", "--name-only", ref])).stdout.split("\0"); + assert.notInclude(files, "scope/in/delete"); + if (turn === 2) assert.notInclude(files, "scope/out/new file"); + assert.deepEqual( + yield* fs.readFile(indexPath).pipe(Effect.orElseSucceed(() => null)), + originalIndex, + ); + assert.isFalse(yield* fs.exists(path.join(cwd, "scope/out/deep/absent"))); + assert.strictEqual( + yield* fs.readFileString(path.join(cwd, "elsewhere/file")), + "working outside\n", + ); + } + }).pipe(Effect.scoped, Effect.provide(GitContractLayer)), + ); + } +} + +it.effect("checkpoint capture keeps the legacy path when Git lacks add --sparse", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const liveProcess = yield* VcsProcess.VcsProcess; + const driver = yield* GitVcsDriver.makeVcsDriverShape(); + const cwd = yield* fs.makeTempDirectoryScoped({ prefix: "t3-checkpoint-legacy-" }); + const { git, checkpointRef } = yield* makeCheckpointFixture(driver, cwd); + yield* git(["sparse-checkout", "set", "--cone", "included"]); + const originalIndex = yield* fs.readFile(path.join(cwd, ".git/index")); + const captureDriver = yield* GitVcsDriver.makeVcsDriverShape().pipe( + Effect.provideService(VcsProcess.VcsProcess, { + run: (input) => { + if (input.args.includes("-h")) + return Effect.succeed({ + exitCode: ChildProcessSpawner.ExitCode(129), + stdout: "usage: git add", + stderr: "", + stdoutTruncated: false, + stderrTruncated: false, + }); + return liveProcess.run( + input.args.includes("--sparse") + ? { + ...input, + args: input.args.map((arg) => + arg === "--sparse" ? "--unsupported-sparse" : arg, + ), + } + : input, + ); + }, + }), + ); + yield* captureDriver.checkpoints.captureCheckpoint({ cwd, checkpointRef }); + assert.strictEqual((yield* git(["show", `${checkpointRef}:file.txt`])).stdout, "unstaged\n"); + assert.deepEqual(yield* fs.readFile(path.join(cwd, ".git/index")), originalIndex); + }).pipe(Effect.scoped, Effect.provide(GitContractLayer)), +); + +for (const indexMode of ["normal", "flags", "sparse"] as const) { + it.effect( + `checkpoint index inspection handles entries beyond the output cap (index=${indexMode})`, + () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const liveProcess = yield* VcsProcess.VcsProcess; + const driver = yield* GitVcsDriver.makeVcsDriverShape(); + const cwd = yield* fs.makeTempDirectoryScoped({ prefix: "t3-checkpoint-inspection-" }); + const { git, checkpointRef } = yield* makeCheckpointFixture(driver, cwd); + yield* fs.writeFileString(path.join(cwd, ".gitattributes"), "stable filter=probe\n"); + yield* fs.writeFileString(path.join(cwd, "stable"), "unchanged\n"); + yield* fs.writeFileString(path.join(cwd, "z-skipped"), "original\n"); + yield* fs.makeDirectory(path.join(cwd, "excluded")); + yield* fs.writeFileString(path.join(cwd, "excluded/file"), "absent\n"); + yield* fs.writeFileString( + path.join(cwd, ".git/filter.cjs"), + 'require("node:fs").appendFileSync(".git/reads", "read\\n"); process.stdin.pipe(process.stdout);', + ); + yield* git(["config", "filter.probe.clean", "node .git/filter.cjs"]); + yield* fs.utimes(path.join(cwd, "stable"), 1_700_000_000, 1_700_000_000); + yield* git(["add", "."]); + yield* git(["commit", "-m", "inspection fixture"]); + if (indexMode === "flags") yield* git(["update-index", "--skip-worktree", "z-skipped"]); + if (indexMode === "sparse") + yield* git(["sparse-checkout", "set", "--cone", "--sparse-index", "included"]); + yield* fs.writeFileString(path.join(cwd, "z-skipped"), "modified\n"); + yield* fs.writeFileString(path.join(cwd, ".git/reads"), ""); + const originalIndex = yield* fs.readFile(path.join(cwd, ".git/index")); + const captureDriver = yield* GitVcsDriver.makeVcsDriverShape().pipe( + Effect.provideService(VcsProcess.VcsProcess, { + run: (input) => + liveProcess.run( + input.args.includes("ls-files") + ? { + ...input, + maxOutputBytes: 8, + onStdoutChunk: (chunk) => { + for (let i = 0; i < chunk.length; i++) + input.onStdoutChunk?.(chunk.subarray(i, i + 1)); + }, + } + : input, + ), + }), + ); + yield* captureDriver.checkpoints.captureCheckpoint({ cwd, checkpointRef }); + assert.strictEqual( + (yield* git(["show", `${checkpointRef}:z-skipped`])).stdout, + "modified\n", + ); + if (indexMode !== "flags") + assert.strictEqual(yield* fs.readFileString(path.join(cwd, ".git/reads")), ""); + assert.deepEqual(yield* fs.readFile(path.join(cwd, ".git/index")), originalIndex); + }).pipe(Effect.scoped, Effect.provide(GitContractLayer)), + ); +} + for (const timestamp of [1_700_000_000, 1_700_000_000.9999]) { it.effect( `checkpoint capture preserves same-size edits with racy index timestamps (${timestamp})`, diff --git a/apps/server/src/vcs/GitVcsDriver.ts b/apps/server/src/vcs/GitVcsDriver.ts index 84dd763150fa9cb9bfd08dd9fa2453b41ee09422..3c6fbe04f89227bf03e13e1a011ef3ad05b3385e 100644 --- a/apps/server/src/vcs/GitVcsDriver.ts +++ b/apps/server/src/vcs/GitVcsDriver.ts @@ -1,4 +1,5 @@ import * as NodeCrypto from "node:crypto"; +import * as NodeBuffer from "node:buffer"; import * as Context from "effect/Context"; import * as DateTime from "effect/DateTime"; @@ -752,6 +753,12 @@ export const makeVcsDriverShape = Effect.fn("makeGitVcsDriverShape")(function* ( const checkpoints: VcsDriver.VcsCheckpointOps = { captureCheckpoint: Effect.fn("GitVcsDriver.checkpoints.captureCheckpoint")(function* (input) { const operation = "GitVcsDriver.checkpoints.captureCheckpoint"; + const indexConfig = [ + "-c", + "core.fsmonitor=false", + "-c", + "sparse.expectFilesOutsideOfPatterns=false", + ]; const gitCommonDir = yield* resolveGitCommonDir(input.cwd); const tempIndexPath = path.join( gitCommonDir, @@ -772,6 +779,22 @@ export const makeVcsDriverShape = Effect.fn("makeGitVcsDriverShape")(function* ( yield* Effect.gen(function* () { const headExists = yield* hasHeadCommit(input.cwd); + const sparseConfig = yield* execute({ + operation, + cwd: input.cwd, + args: ["config", "--bool", "core.sparseCheckout"], + allowNonZeroExit: true, + }); + let sparseCheckout = sparseConfig.stdout.trim() === "true"; + if (sparseCheckout) { + const help = yield* execute({ + operation, + cwd: input.cwd, + args: ["add", "-h"], + allowNonZeroExit: true, + }); + sparseCheckout = /--(?:\[no-\])?sparse\b/.test(`${help.stdout}${help.stderr}`); + } if (headExists) { const reusedIndex = yield* Effect.gen(function* () { const indexPath = yield* execute({ @@ -789,26 +812,92 @@ export const makeVcsDriverShape = Effect.fn("makeGitVcsDriverShape")(function* ( yield* execute({ operation, cwd: input.cwd, - args: ["-c", "core.fsmonitor=false", "read-tree", "--reset", "HEAD"], + args: [...indexConfig, "read-tree", "--reset", "HEAD"], env: commitEnv, }); // read-tree can rewrite the index, so restore its racy timestamp afterward. yield* fileSystem.utimes(tempIndexPath, indexTime, indexTime); - const entries = yield* execute({ + let specialFlags = false; + let recordStart = true; + let skipped = false; + let skippedRecord: number[] = []; + const skippedPaths: string[] = []; + yield* vcsProcess.run({ operation, + command: "git", cwd: input.cwd, - args: ["ls-files", "-v"], + args: [...indexConfig, "ls-files", "--full-name", "--sparse", "-v", "-z"], env: commitEnv, - maxOutputBytes: WORKSPACE_FILES_MAX_OUTPUT_BYTES, + maxOutputBytes: 4_096, + outputMode: "truncate", + // Inspect every tag; retain only skipped file paths for checking sparse rules. + onStdoutChunk: (chunk) => { + for (const byte of chunk) { + if (recordStart) skipped = byte === 83; + if (skipped && sparseCheckout) { + if (byte !== 0) skippedRecord.push(byte); + else { + if (skippedRecord.at(-1) !== 47) { + const name = Buffer.from(skippedRecord).subarray(2); + if (!NodeBuffer.isUtf8(name)) specialFlags = true; + else skippedPaths.push(name.toString("utf8")); + } + skippedRecord = []; + } + } + if ( + recordStart && + ((byte >= 97 && byte <= 122) || (!sparseCheckout && byte === 83)) + ) { + specialFlags = true; + } + recordStart = byte === 0; + } + }, }); - // A fresh index must still capture assume-unchanged/skip-worktree files. - return !entries.stdoutTruncated && !/^[a-zS] /m.test(entries.stdout); + if (skippedPaths.length > 0 && !specialFlags) { + const selected = yield* execute({ + operation, + cwd: input.cwd, + args: [...indexConfig, "sparse-checkout", "check-rules", "-z"], + stdin: skippedPaths.join("\0") + "\0", + env: commitEnv, + maxOutputBytes: 1, + outputMode: "truncate", + }); + // Any selected skipped file has a manual flag, not a sparse exclusion. + specialFlags = selected.stdout.length > 0 || selected.stdoutTruncated; + } + // Sparse Git clears skip-worktree for present files. Manual flags still need a reset. + return !specialFlags; }).pipe(Effect.orElseSucceed(() => false)); if (!reusedIndex) { + if (sparseCheckout) { + const cone = yield* execute({ + operation, + cwd: input.cwd, + args: ["config", "--bool", "core.sparseCheckoutCone"], + allowNonZeroExit: true, + }); + // Rebuilding a non-cone index loses exclusions; do not publish false deletions. + if (cone.stdout.trim() !== "true") { + return yield* new VcsProcessExitError({ + operation, + command: "git read-tree", + cwd: input.cwd, + exitCode: 1, + detail: "Cannot rebuild a checkpoint index for non-cone sparse checkout.", + }); + } + } + yield* cleanupTempIndex; yield* execute({ operation, cwd: input.cwd, - args: ["read-tree", "HEAD"], + // A fresh sparse index represents excluded directories without marking them deleted. + args: sparseCheckout + ? [...indexConfig, "-c", "index.sparse=true", "read-tree", "--reset", "HEAD"] + : ["read-tree", "HEAD"], env: commitEnv, }); } @@ -817,14 +906,15 @@ export const makeVcsDriverShape = Effect.fn("makeGitVcsDriverShape")(function* ( yield* execute({ operation, cwd: input.cwd, - args: ["add", "-A", "--", "."], + // Preserve absent skipped entries, but capture present nonignored files outside the cone. + args: [...indexConfig, "add", ...(sparseCheckout ? ["--sparse"] : []), "-A", "--", "."], env: commitEnv, }); const writeTreeResult = yield* execute({ operation, cwd: input.cwd, - args: ["write-tree"], + args: [...indexConfig, "write-tree"], env: commitEnv, }); const treeOid = writeTreeResult.stdout.trim(); diff --git a/apps/server/src/vcs/VcsProcess.test.ts b/apps/server/src/vcs/VcsProcess.test.ts index ac0ee442804690a3268386902cfb7c205b91fae4..80dea03b245c367bab9f9e11d3d8d1a69c3f2d92 100644 --- a/apps/server/src/vcs/VcsProcess.test.ts +++ b/apps/server/src/vcs/VcsProcess.test.ts @@ -378,6 +378,24 @@ describe("VcsProcess.run", () => { }).pipe(provideLive), ); + it.effect("streams all stdout bytes beyond the buffered output cap", () => + Effect.gen(function* () { + const chunks: Uint8Array[] = []; + const result = yield* run({ + operation: "test.stream-output", + command: "node", + args: ["-e", "process.stdout.write('x'.repeat(131072) + '\\0S final\\0')"], + cwd: process.cwd(), + maxOutputBytes: 8, + onStdoutChunk: (chunk) => chunks.push(chunk), + }); + + expect(result.stdout).toBe("xxxxxxxx"); + expect(result.stdoutTruncated).toBe(true); + expect(Buffer.concat(chunks).toString()).toBe("x".repeat(131072) + "\0S final\0"); + }).pipe(provideLive), + ); + it.effect("fails with VcsProcessTimeoutError on timeout", () => Effect.gen(function* () { const errorFiber = yield* run({ diff --git a/apps/server/src/vcs/VcsProcess.ts b/apps/server/src/vcs/VcsProcess.ts index 74608e116eed48c2d82e364297d86e35a578c5dc..fcf1cf96025a17624cdd5a16a580fa738f1cf0ea 100644 --- a/apps/server/src/vcs/VcsProcess.ts +++ b/apps/server/src/vcs/VcsProcess.ts @@ -25,6 +25,7 @@ export interface VcsProcessInput { readonly cwd: string; readonly spawnCwd?: string; readonly stdin?: string; + readonly onStdoutChunk?: (chunk: Uint8Array) => void; readonly env?: NodeJS.ProcessEnv; readonly allowNonZeroExit?: boolean; readonly timeoutMs?: number; @@ -123,6 +124,7 @@ export const make = Effect.gen(function* () { cwd: input.cwd, ...(input.spawnCwd !== undefined ? { spawnCwd: input.spawnCwd } : {}), ...(input.stdin !== undefined ? { stdin: input.stdin } : {}), + ...(input.onStdoutChunk !== undefined ? { onStdoutChunk: input.onStdoutChunk } : {}), ...(input.env !== undefined ? { env: input.env } : {}), timeout: input.timeoutMs ?? DEFAULT_TIMEOUT_MS, maxOutputBytes: input.maxOutputBytes ?? DEFAULT_MAX_OUTPUT_BYTES,