diff --git a/apps/server/src/vcs/GitVcsDriver.test.ts b/apps/server/src/vcs/GitVcsDriver.test.ts index 3c2b62259bff5..0fcea5b44d8ec 100644 --- a/apps/server/src/vcs/GitVcsDriver.test.ts +++ b/apps/server/src/vcs/GitVcsDriver.test.ts @@ -15,6 +15,7 @@ import { ChildProcessSpawner } from "effect/process"; import { assert, it } from "@effect/vitest"; import { CheckpointRef, GitCommandError, VcsProcessExitError } from "@t3tools/contracts"; +import * as HostProcess from "@t3tools/shared/HostProcess"; import * as ServerConfig from "../config.ts"; import * as CheckpointStore from "../checkpointing/CheckpointStore.ts"; import * as ProcessRunner from "../processRunner.ts"; @@ -532,10 +533,247 @@ it.effect("checkpoint recovery preserves interruption and removes the private in assert.isTrue(Exit.isFailure(exit) && Cause.hasInterruptsOnly(exit.cause)); assert.isDefined(privateIndex); assert.isFalse(yield* fs.exists(privateIndex!)); + assert.isFalse(yield* fs.exists((yield* Path.Path).dirname(privateIndex!))); assert.isFalse(yield* driver.checkpoints.hasCheckpointRef({ cwd, checkpointRef })); }).pipe(Effect.scoped, Effect.provide(layerGitContract)), ); +// These tests rely on POSIX FIFOs, shell clean filters, and permission bits. +const windowsHost = HostProcess.Platform.defaultValue() === "win32"; + +const makeLargeFileFixture = Effect.fn("makeLargeFileFixture")(function* ( + driver: Effect.Success>, + cwd: string, +) { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const fixture = yield* makeCheckpointFixture(driver, cwd); + // Above core.bigFileThreshold, `git add` streams the file into a temporary pack. + yield* fixture.git(["config", "core.bigFileThreshold", "1k"]); + yield* fs.writeFile( + path.join(cwd, "a-large.bin"), + Uint8Array.from({ length: 8192 }, (_, index) => (index * 7919) % 251), + ); + const objects = path.join(cwd, ".git", "objects"); + const listObjectStore = fs + .readDirectory(objects, { recursive: true }) + .pipe(Effect.map((entries) => entries.toSorted())); + return { ...fixture, objects, listObjectStore }; +}); + +it.effect.skipIf(windowsHost)( + "concurrent checkpoint captures stay readable with shared permissions and without hard links", + () => + 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-publish-" }); + const { git, objects, listObjectStore } = yield* makeLargeFileFixture(driver, cwd); + yield* git(["config", "core.sharedRepository", "group"]); + const existingDirectories = new Set(yield* fs.readDirectory(objects)); + const refs = ["first", "second"].map((name) => + CheckpointRef.make(`refs/t3/checkpoints/${name}`), + ); + + const noLinksDriver = yield* GitVcsDriver.makeVcsDriverShape().pipe( + Effect.provideService(FileSystem.FileSystem, { + ...fs, + link: (_source, target) => + Effect.fail( + PlatformError.systemError({ + _tag: "PermissionDenied", + module: "FileSystem", + method: "link", + pathOrDescriptor: target, + description: "EPERM", + }), + ), + }), + ); + yield* Effect.forEach( + refs, + (checkpointRef, index) => + (index === 0 ? driver : noLinksDriver).checkpoints.captureCheckpoint({ + cwd, + checkpointRef, + }), + { concurrency: "unbounded" }, + ); + + const expected = yield* fs.readFileString(path.join(cwd, "a-large.bin")); + for (const checkpointRef of refs) { + assert.strictEqual((yield* git(["show", `${checkpointRef}:a-large.bin`])).stdout, expected); + } + yield* git(["fsck", "--strict", "--no-dangling"]); + assert.isFalse((yield* listObjectStore).some((entry) => entry.includes("tmp_"))); + const createdDirectories = (yield* fs.readDirectory(objects)).filter( + (entry) => !existingDirectories.has(entry), + ); + assert.isNotEmpty(createdDirectories); + for (const entry of createdDirectories) { + // Other group members' Git processes must be able to add objects here. + const { mode } = yield* fs.stat(path.join(objects, entry)); + assert.strictEqual(mode & 0o2070, 0o2070, entry); + } + }).pipe(Effect.scoped, Effect.provide(layerGitCaptureContract)), +); + +it.effect( + "failed checkpoint publication preserves an identical pack adopted by another capture", + () => + 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-publish-race-" }); + const { git, checkpointRef, listObjectStore } = yield* makeLargeFileFixture(driver, cwd); + const installingIndex = yield* Deferred.make(); + const failInstall = yield* Deferred.make(); + const diskFull = (method: string, target: string) => + Effect.fail( + PlatformError.systemError({ + _tag: "Unknown", + module: "FileSystem", + method, + pathOrDescriptor: target, + description: "ENOSPC", + }), + ); + const failingDriver = yield* GitVcsDriver.makeVcsDriverShape().pipe( + Effect.provideService(FileSystem.FileSystem, { + ...fs, + link: (source, target) => + target.endsWith(".idx") ? diskFull("link", target) : fs.link(source, target), + rename: (source, target) => + target.endsWith(".idx") + ? Deferred.succeed(installingIndex, undefined).pipe( + Effect.andThen(Deferred.await(failInstall)), + Effect.andThen(diskFull("rename", target)), + ) + : fs.rename(source, target), + }), + ); + const failedCapture = yield* failingDriver.checkpoints + .captureCheckpoint({ cwd, checkpointRef }) + .pipe(Effect.exit, Effect.forkScoped); + yield* Effect.addFinalizer(() => Deferred.succeed(failInstall, undefined)); + yield* Deferred.await(installingIndex); + + const adoptedRef = CheckpointRef.make("refs/t3/checkpoints/adopted"); + yield* driver.checkpoints.captureCheckpoint({ cwd, checkpointRef: adoptedRef }); + const expected = yield* fs.readFileString(path.join(cwd, "a-large.bin")); + assert.strictEqual((yield* git(["show", `${adoptedRef}:a-large.bin`])).stdout, expected); + + yield* Deferred.succeed(failInstall, undefined); + assert.isTrue(Exit.isFailure(yield* Fiber.join(failedCapture))); + assert.isFalse(yield* driver.checkpoints.hasCheckpointRef({ cwd, checkpointRef })); + const adopted = yield* driver.execute({ + operation: "checkpoint-test", + cwd, + args: ["show", `${adoptedRef}:a-large.bin`], + allowNonZeroExit: true, + }); + assert.strictEqual(adopted.exitCode, 0, adopted.stderr); + assert.strictEqual(adopted.stdout, expected); + assert.isFalse((yield* listObjectStore).some((entry) => entry.includes("tmp_"))); + yield* git(["fsck", "--strict", "--no-dangling"]); + }).pipe(Effect.scoped, Effect.provide(layerGitCaptureContract)), +); + +it.effect.skipIf(windowsHost)( + "failed, interrupted, and timed-out captures clean up only their own temporary objects", + () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const processRunner = yield* ProcessRunner.ProcessRunner; + const driver = yield* GitVcsDriver.makeVcsDriverShape(); + const cwd = yield* fs.makeTempDirectoryScoped({ prefix: "t3-checkpoint-killed-" }); + const gates = yield* fs.makeTempDirectoryScoped({ prefix: "t3-checkpoint-gates-" }); + const { git, checkpointRef, objects, listObjectStore } = yield* makeLargeFileFixture( + driver, + cwd, + ); + const existingRef = CheckpointRef.make("refs/t3/checkpoints/existing"); + yield* driver.checkpoints.captureCheckpoint({ cwd, checkpointRef: existingRef }); + const existingRefs = (yield* git(["show-ref"])).stdout; + yield* fs.writeFileString( + path.join(cwd, "a-large.bin"), + "new packed contents\n".repeat(1024), + ); + yield* fs.writeFileString(path.join(cwd, "b-loose.txt"), "new loose object\n"); + // A later file blocks in its clean filter while Git holds the large file's open pack. + const started = path.join(gates, "started"); + const hold = path.join(gates, "hold"); + yield* processRunner.run({ command: "mkfifo", args: [started, hold], cwd: gates }); + const filter = path.join(gates, "filter.sh"); + yield* fs.writeFileString( + filter, + `#!/bin/sh\necho started > '${started}'\ncat '${hold}' > /dev/null\nexit 1\n`, + ); + yield* fs.chmod(filter, 0o755); + yield* fs.writeFileString( + path.join(cwd, ".git", "info", "attributes"), + "z-gated.txt filter=gate\n", + ); + yield* git(["config", "filter.gate.clean", `'${filter}'`]); + yield* git(["config", "filter.gate.required", "true"]); + yield* fs.writeFileString(path.join(cwd, "z-gated.txt"), "gated\n"); + // Stand-ins for a concurrent Git command's pack and another capture's quarantine. + yield* fs.writeFileString(path.join(objects, "pack", "tmp_pack_concurrent"), "concurrent"); + yield* fs.makeDirectory(path.join(objects, "tmp_objdir-t3-checkpoint-concurrent")); + const originalIndex = yield* fs.readFile(path.join(cwd, ".git", "index")); + const objectStore = yield* listObjectStore; + + for (const outcome of ["failure", "interrupt", "timeout"] as const) { + const capture = yield* driver.checkpoints + .captureCheckpoint({ cwd, checkpointRef }) + .pipe(Effect.forkScoped); + yield* fs.readFileString(started); + assert.isTrue( + (yield* listObjectStore).some((entry) => /tmp_pack_(?!concurrent)/.test(entry)), + ); + if (outcome === "failure") yield* fs.writeFileString(hold, "release\n"); + else if (outcome === "interrupt") yield* Fiber.interrupt(capture); + else yield* TestClock.adjust("30 seconds"); + const exit = yield* Fiber.await(capture); + assert.isTrue(Exit.isFailure(exit)); + if (Exit.isFailure(exit)) { + if (outcome === "interrupt") assert.isTrue(Cause.hasInterruptsOnly(exit.cause)); + else { + const error = Cause.findErrorOption(exit.cause); + assert.isTrue(error._tag === "Some"); + if (error._tag === "Some") + assert.strictEqual( + error.value._tag, + outcome === "timeout" ? "VcsProcessTimeoutError" : "VcsProcessExitError", + ); + } + } + assert.deepEqual(yield* listObjectStore, objectStore); + assert.deepEqual(yield* fs.readFile(path.join(cwd, ".git", "index")), originalIndex); + assert.strictEqual((yield* git(["show-ref"])).stdout, existingRefs); + assert.strictEqual( + yield* fs.readFileString(path.join(objects, "pack", "tmp_pack_concurrent")), + "concurrent", + ); + assert.isFalse(yield* driver.checkpoints.hasCheckpointRef({ cwd, checkpointRef })); + } + + yield* git(["config", "--unset", "filter.gate.clean"]); + yield* git(["config", "--unset", "filter.gate.required"]); + yield* driver.checkpoints.captureCheckpoint({ cwd, checkpointRef }); + const captured = yield* git(["show", `${checkpointRef}:a-large.bin`]); + assert.strictEqual(captured.stdout, yield* fs.readFileString(path.join(cwd, "a-large.bin"))); + assert.strictEqual( + (yield* git(["show", `${checkpointRef}:b-loose.txt`])).stdout, + "new loose object\n", + ); + yield* git(["fsck", "--strict", "--no-dangling"]); + }).pipe(Effect.scoped, Effect.provide(layerGitCaptureContract)), +); + it.effect("checkpoint capture does not rerun clean filters for unchanged indexed files", () => Effect.gen(function* () { const fileSystem = yield* FileSystem.FileSystem; @@ -1151,8 +1389,11 @@ it.effect("GitVcsDriver forwards execute env to the VCS process", () => { it.effect("GitVcsDriver flushes checkpoint objects and refs to disk before publishing them", () => { const observedArgs: ReadonlyArray[] = []; + let objectDirectory = ""; return Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + objectDirectory = yield* fileSystem.makeTempDirectoryScoped({ prefix: "t3-checkpoint-fsync-" }); const driver = yield* GitVcsDriver.makeVcsDriverShape(); yield* driver.checkpoints.captureCheckpoint({ @@ -1185,6 +1426,7 @@ it.effect("GitVcsDriver flushes checkpoint objects and refs to disk before publi "commit0000", ]); }).pipe( + Effect.scoped, Effect.provide( Layer.mergeAll( NodeServices.layer, @@ -1196,8 +1438,8 @@ it.effect("GitVcsDriver flushes checkpoint objects and refs to disk before publi ? "tree0000\n" : input.args.includes("commit-tree") ? "commit0000\n" - : input.args.includes("--git-common-dir") - ? ".git\n" + : input.args.includes("--git-path") && input.args.includes("objects") + ? `${objectDirectory}\n` : ""; return { exitCode: ChildProcessSpawner.ExitCode(0), diff --git a/apps/server/src/vcs/GitVcsDriver.ts b/apps/server/src/vcs/GitVcsDriver.ts index 5d382f16d2ec2..d83317e7202bd 100644 --- a/apps/server/src/vcs/GitVcsDriver.ts +++ b/apps/server/src/vcs/GitVcsDriver.ts @@ -32,6 +32,7 @@ import { type VcsStatusResult, type WorktreeSubmodules, } from "@t3tools/contracts"; +import * as HostProcess from "@t3tools/shared/HostProcess"; import { makeGitVcsDriverCore, PATCH_RENDER_PREFIX_ARGS, @@ -407,6 +408,21 @@ export class GitVcsDriver extends Context.Service< const WORKSPACE_FILES_MAX_OUTPUT_BYTES = 16 * 1024 * 1024; const CHECKPOINT_RECOVERY_MAX_CANDIDATES = 64; const CHECKPOINT_RECOVERY_TIMEOUT = "5 seconds"; +// A pack becomes visible through its index, so publish the index last. +const PACK_PUBLISH_ORDER = ["keep", "pack", "rev", "mtimes", "promisor", "idx"]; + +/** Publication order of a quarantined object file, or undefined for anything that is not one. */ +const quarantinedObjectRank = (entry: string) => { + if (/^[0-9a-f]{2}\/[0-9a-f]{38,62}$/.test(entry)) return PACK_PUBLISH_ORDER.length; + const extension = /^pack\/pack-[0-9a-f]{40,64}\.([a-z]+)$/.exec(entry)?.[1]; + const rank = extension === undefined ? -1 : PACK_PUBLISH_ORDER.indexOf(extension); + return rank === -1 ? undefined : rank; +}; + +// Git C-unquotes alternates that start with a quote, keeping path delimiters and backslashes literal. +const quoteAlternateObjectDirectory = (directory: string) => + `"${directory.replace(/[\\"]/g, "\\$&")}"`; + const GIT_CHECK_IGNORE_MAX_STDIN_BYTES = 256 * 1024; const CHECKPOINT_DIFF_MAX_OUTPUT_BYTES = 10_000_000; const WORKSPACE_GIT_HARDENED_CONFIG_ARGS = [ @@ -769,17 +785,66 @@ export const makeVcsDriverShape = Effect.fn("makeGitVcsDriverShape")(function* ( }), ); - const resolveGitCommonDir = (cwd: string) => + const resolveObjectDirectory = (cwd: string) => Effect.gen(function* () { const result = yield* execute({ - operation: "GitVcsDriver.checkpoints.resolveGitCommonDir", + operation: "GitVcsDriver.checkpoints.resolveObjectDirectory", cwd, - args: ["rev-parse", "--git-common-dir"], + args: ["rev-parse", "--git-path", "objects"], }); - const gitCommonDir = result.stdout.trim(); - return path.isAbsolute(gitCommonDir) ? gitCommonDir : path.resolve(cwd, gitCommonDir); + return path.resolve(cwd, result.stdout.trim()); }); + // Published files are shared immediately: another capture can adopt a pack before we + // install its index. Even on failure, only the private quarantine is ours to remove. + const publishQuarantinedObjects = Effect.fnUntraced(function* ( + quarantine: string, + objectDirectory: string, + ) { + const objectFiles = (yield* fileSystem.readDirectory(quarantine, { recursive: true })) + .flatMap((entry) => { + const rank = quarantinedObjectRank(entry.replaceAll("\\", "/")); + return rank === undefined ? [] : [{ entry, rank }]; + }) + .toSorted((left, right) => left.rank - right.rank); + const readyDirectories = new Set(); + for (const { entry } of objectFiles) { + const directory = path.dirname(entry); + if (!readyDirectories.has(directory)) { + const target = path.join(objectDirectory, directory); + // Git created the quarantine's directory with core.sharedRepository permissions. + yield* fileSystem.makeDirectory(target).pipe( + Effect.andThen(fileSystem.stat(path.join(quarantine, directory))), + Effect.flatMap((info) => fileSystem.chmod(target, info.mode & 0o7777)), + Effect.catchIf( + (error) => error.reason._tag === "AlreadyExists", + () => Effect.void, + ), + ); + readyDirectories.add(directory); + } + const source = path.join(quarantine, entry); + const target = path.join(objectDirectory, entry); + // Objects are content-addressed, so an existing file already holds these bytes. Without + // hard links, Git renames instead, unless the object already exists. + yield* fileSystem + .link(source, target) + .pipe( + Effect.catch((error) => + error.reason._tag === "AlreadyExists" + ? Effect.void + : fileSystem + .exists(target) + .pipe( + Effect.flatMap((exists) => + exists ? Effect.void : fileSystem.rename(source, target), + ), + ), + ), + ); + } + }); + // Git renames loose objects and refs into place without fsync by default, so // an unclean restart can leave 0-byte files under refs/t3/** that break every // later fetch and push. Checkpoint writes flush before they are published; @@ -807,12 +872,23 @@ export const makeVcsDriverShape = Effect.fn("makeGitVcsDriverShape")(function* ( "-c", "splitIndex.sharedIndexExpire=never", ]; - const gitCommonDir = yield* resolveGitCommonDir(input.cwd); - const indexId = yield* crypto.randomUUIDv4().pipe(Effect.orDie); - const tempIndexPath = path.join(gitCommonDir, `t3-checkpoint-index-${indexId}`); + const objectDirectory = yield* resolveObjectDirectory(input.cwd); + // New objects go to a private directory until the capture succeeds, so a failed or + // interrupted `git add` cannot leave its temporary packs in the repository. Git names + // its own quarantines tmp_objdir-*, which lets `git gc` reclaim one a crash left behind. + const quarantineId = yield* crypto.randomUUIDv4().pipe(Effect.orDie); + const quarantine = path.join(objectDirectory, `tmp_objdir-t3-checkpoint-${quarantineId}`); + const tempIndexPath = path.join(quarantine, "index"); const commitEnv: NodeJS.ProcessEnv = { ...process.env, GIT_INDEX_FILE: tempIndexPath, + GIT_OBJECT_DIRECTORY: quarantine, + GIT_ALTERNATE_OBJECT_DIRECTORIES: [ + quoteAlternateObjectDirectory(objectDirectory), + ...(process.env.GIT_ALTERNATE_OBJECT_DIRECTORIES + ? [process.env.GIT_ALTERNATE_OBJECT_DIRECTORIES] + : []), + ].join((yield* HostProcess.Platform) === "win32" ? ";" : ":"), GIT_AUTHOR_NAME: "T3 Code", GIT_AUTHOR_EMAIL: "t3code@users.noreply.github.com", GIT_COMMITTER_NAME: "T3 Code", @@ -825,7 +901,28 @@ export const makeVcsDriverShape = Effect.fn("makeGitVcsDriverShape")(function* ( { discard: true }, ); + const removeQuarantine = fileSystem + .remove(quarantine, { recursive: true, force: true }) + .pipe( + Effect.catch((error) => + Effect.logWarning("checkpoint temporary objects were not removed", { error }), + ), + ); + yield* Effect.gen(function* () { + yield* fileSystem.makeDirectory(quarantine).pipe( + Effect.mapError( + (cause) => + new VcsProcessExitError({ + operation, + command: "git checkpoint storage", + cwd: input.cwd, + exitCode: 0, + detail: "Could not create checkpoint object storage.", + cause, + }), + ), + ); const headExists = yield* hasHeadCommit(input.cwd); const sparseConfig = yield* execute({ operation, @@ -1056,12 +1153,30 @@ export const makeVcsDriverShape = Effect.fn("makeGitVcsDriverShape")(function* ( }); } - yield* execute({ - operation, - cwd: input.cwd, - args: [...durableWrite, "update-ref", input.checkpointRef, commitOid], - }); - }).pipe(Effect.ensuring(cleanupTempIndex)); + // Finish publishing objects and their ref together when cancellation arrives. + yield* Effect.uninterruptible( + Effect.gen(function* () { + yield* publishQuarantinedObjects(quarantine, objectDirectory).pipe( + Effect.mapError( + (cause) => + new VcsProcessExitError({ + operation, + command: "git checkpoint storage", + cwd: input.cwd, + exitCode: 0, + detail: "Could not publish checkpoint objects.", + cause, + }), + ), + ); + yield* execute({ + operation, + cwd: input.cwd, + args: [...durableWrite, "update-ref", input.checkpointRef, commitOid], + }); + }), + ); + }).pipe(Effect.ensuring(removeQuarantine)); }), hasCheckpointRef: (input) => diff --git a/packages/contracts/src/vcs.ts b/packages/contracts/src/vcs.ts index f92e4fc1d5c48..f5848d4b710d3 100644 --- a/packages/contracts/src/vcs.ts +++ b/packages/contracts/src/vcs.ts @@ -122,6 +122,7 @@ export class VcsProcessExitError extends Schema.TaggedError retryable: Schema.optional(Schema.Boolean), stderrLength: Schema.optional(NonNegativeInt), stderrTruncated: Schema.optional(Schema.Boolean), + cause: Schema.optional(Schema.Defect()), }, ) { override get message(): string {