authorgravatar for saphid@gmail.comAlex Southwell <saphid@gmail.com> 2026-09-19 04:42:28+10:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-18 11:42:28-07:00
log869347bc26051bba7c3c0274a7dfcae7770d3e3a
tree1d54942beb37ef79478bc9cab1f78b05c54239a5
parente84f23a97d587e0e58215f1fba2eab12717bec28
signature Signed by PGP key B5690EEEBB952194

fix(server): retry transient git failures during checkpoint capture (#11665)

Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> Co-authored-by: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com>

5 files changed, 348 insertions(+), 4 deletions(-)

apps/server/src/vcs/GitVcsDriver.test.ts+140-2
......@@ -1,5 +1,7 @@
11import * as NodeServices from "@effect/platform-node/NodeServices";
22import * as Cause from "effect/Cause";
3import * as Clock from "effect/Clock";
4import * as Duration from "effect/Duration";
35import * as Exit from "effect/Exit";
46import * as Deferred from "effect/Deferred";
57import * as Effect from "effect/Effect";
......@@ -14,7 +16,11 @@ import { assert, it } from "@effect/vitest";
1416
1517import { CheckpointRef, GitCommandError, VcsProcessExitError } from "@t3tools/contracts";
1618import * as ServerConfig from "../config.ts";
19import * as CheckpointStore from "../checkpointing/CheckpointStore.ts";
20import * as ProcessRunner from "../processRunner.ts";
1721import * as GitVcsDriver from "./GitVcsDriver.ts";
22import type * as VcsDriver from "./VcsDriver.ts";
23import * as VcsDriverRegistry from "./VcsDriverRegistry.ts";
1824import * as VcsProcess from "./VcsProcess.ts";
1925import { runVcsDriverContractSuite } from "./testing/VcsDriverContractHarness.ts";
2026
......@@ -26,6 +32,10 @@ const GitContractLayer = Layer.mergeAll(GitVcsDriver.vcsLayer, GitVcsDriver.laye
2632 Layer.provideMerge(VcsProcess.layer),
2733 Layer.provideMerge(NodeServices.layer),
2834);
35const GitCaptureContractLayer = Layer.merge(
36 GitContractLayer,
37 ProcessRunner.layer.pipe(Layer.provide(NodeServices.layer)),
38);
2939
3040const runGit = (cwd: string, args: ReadonlyArray<string>) =>
3141 Effect.gen(function* () {
......@@ -38,6 +48,22 @@ const runGit = (cwd: string, args: ReadonlyArray<string>) =>
3848 });
3949 });
4050
51const makeCaptureStore = Effect.fn("test.makeCaptureStore")(function* (
52 driver: VcsDriver.VcsDriver["Service"],
53 cwd: string,
54) {
55 const repository = yield* driver.detectRepository(cwd);
56 if (repository === null) return yield* Effect.die("Expected a test Git repository");
57 const handle = { kind: repository.kind, repository, driver };
58 return yield* CheckpointStore.make.pipe(
59 Effect.provideService(VcsDriverRegistry.VcsDriverRegistry, {
60 get: () => Effect.succeed(driver),
61 detect: () => Effect.succeed(handle),
62 resolve: () => Effect.succeed(handle),
63 }),
64 );
65});
66
4167type GitContractError = GitCommandError | PlatformError.PlatformError;
4268
4369runVcsDriverContractSuite<GitVcsDriver.GitVcsDriver, GitContractError>({
......@@ -260,6 +286,112 @@ it.effect("checkpoint recovery refuses excessive candidates before probing", ()
260286 }).pipe(Effect.scoped, Effect.provide(GitContractLayer)),
261287);
262288
289it.effect.each([
290 { phase: "add", nestedRecovery: false, expireRecovery: false },
291 { phase: "update-ref", nestedRecovery: false, expireRecovery: false },
292 { phase: "update-ref", nestedRecovery: true, expireRecovery: false },
293 { phase: "add", nestedRecovery: true, expireRecovery: false },
294 { phase: "add", nestedRecovery: true, expireRecovery: true },
295])(
296 "checkpoint handles a $phase lock with nestedRecovery=$nestedRecovery, expireRecovery=$expireRecovery",
297 ({ phase, nestedRecovery, expireRecovery }) =>
298 Effect.gen(function* () {
299 const fs = yield* FileSystem.FileSystem;
300 const path = yield* Path.Path;
301 const liveRunner = yield* ProcessRunner.ProcessRunner;
302 const driver = yield* GitVcsDriver.makeVcsDriverShape();
303 const cwd = yield* fs.makeTempDirectoryScoped({ prefix: "t3-checkpoint-ref-race-" });
304 const { git, checkpointRef } = yield* makeCheckpointFixture(driver, cwd);
305 if (nestedRecovery) yield* git(["init", "empty"]);
306 const originalIndex = yield* fs.readFile(path.join(cwd, ".git", "index"));
307 const refLockPath = path.join(cwd, ".git", `${checkpointRef}.lock`);
308 const failed = yield* Deferred.make<void>();
309 const retryReached = yield* Deferred.make<void>();
310 const allowRetry = yield* Deferred.make<void>();
311 const clock = yield* Clock.Clock;
312 const privateIndexes = new Set<string>();
313 let racedAttempts = 0;
314 let discoveries = 0;
315 let stageError: VcsProcessExitError | undefined;
316 const captureProcess = yield* VcsProcess.make.pipe(
317 Effect.provideService(ProcessRunner.ProcessRunner, {
318 run: (input) => {
319 if (input.args.includes("--others")) discoveries += 1;
320 if (input.env?.GIT_INDEX_FILE) privateIndexes.add(input.env.GIT_INDEX_FILE);
321 const initialStage =
322 phase === "add" &&
323 nestedRecovery &&
324 !input.args.some((arg) => arg.startsWith(":(exclude,literal)"));
325 if (!input.args.includes(phase) || initialStage || ++racedAttempts !== 1) {
326 return liveRunner.run(input);
327 }
328 const lockPath = phase === "add" ? `${input.env!.GIT_INDEX_FILE!}.lock` : refLockPath;
329 return Effect.gen(function* () {
330 yield* fs
331 .makeDirectory(path.dirname(lockPath), { recursive: true })
332 .pipe(Effect.orDie);
333 yield* fs.writeFileString(lockPath, "concurrent ref writer").pipe(Effect.orDie);
334 return yield* liveRunner.run(input).pipe(
335 Effect.ensuring(fs.remove(lockPath).pipe(Effect.orDie)),
336 Effect.tap(() => Deferred.succeed(failed, undefined)),
337 );
338 });
339 },
340 }),
341 );
342 const captureDriver = yield* GitVcsDriver.makeVcsDriverShape().pipe(
343 Effect.provideService(VcsProcess.VcsProcess, {
344 run: (input) =>
345 captureProcess.run(input).pipe(
346 Effect.tapError((error) => {
347 if (input.args.includes("add") && error._tag === "VcsProcessExitError")
348 stageError = error;
349 return Effect.void;
350 }),
351 ),
352 }),
353 );
354 const captureStore = yield* makeCaptureStore(captureDriver, cwd);
355 const fiber = yield* captureStore.captureCheckpoint({ cwd, checkpointRef }).pipe(
356 Effect.provideService(Clock.Clock, {
357 ...clock,
358 sleep: (duration) =>
359 Duration.toMillis(duration) === 75
360 ? Deferred.succeed(retryReached, undefined).pipe(
361 Effect.andThen(Deferred.await(allowRetry)),
362 )
363 : clock.sleep(duration),
364 }),
365 Effect.exit,
366 Effect.forkScoped,
367 );
368 yield* Deferred.await(failed);
369 yield* Deferred.await(retryReached);
370 if (expireRecovery) yield* TestClock.adjust("5 seconds");
371 else yield* Deferred.succeed(allowRetry, undefined);
372 const result = yield* Fiber.join(fiber);
373 if (expireRecovery) {
374 if (Exit.isSuccess(result))
375 return yield* Effect.die("Expected the recovery deadline to expire");
376 const error = Cause.findErrorOption(result.cause);
377 assert.isTrue(error._tag === "Some");
378 if (error._tag === "Some") assert.strictEqual(error.value, stageError);
379 } else assert.isTrue(Exit.isSuccess(result));
380 assert.strictEqual(racedAttempts, expireRecovery ? 1 : 2);
381 assert.strictEqual(discoveries, nestedRecovery ? 1 : 0);
382 assert.strictEqual(privateIndexes.size, 1);
383 assert.strictEqual(
384 yield* driver.checkpoints.hasCheckpointRef({ cwd, checkpointRef }),
385 !expireRecovery,
386 );
387 for (const index of privateIndexes) {
388 assert.isFalse(yield* fs.exists(index));
389 assert.isFalse(yield* fs.exists(`${index}.lock`));
390 }
391 assert.deepEqual(yield* fs.readFile(path.join(cwd, ".git", "index")), originalIndex);
392 }).pipe(Effect.scoped, Effect.provide(GitCaptureContractLayer)),
393);
394
263395for (const blockedPhase of ["discovery", "probe", "retry"] as const) {
264396 it.effect(`checkpoint recovery has one deadline including ${blockedPhase}`, () =>
265397 Effect.gen(function* () {
......@@ -276,11 +408,15 @@ for (const blockedPhase of ["discovery", "probe", "retry"] as const) {
276408 let stageError: VcsProcessExitError | undefined;
277409 let privateIndex: string | undefined;
278410 let interrupted = false;
411 let stagingAttempts = 0;
279412 const captureDriver = yield* GitVcsDriver.makeVcsDriverShape().pipe(
280413 Effect.provideService(VcsProcess.VcsProcess, {
281414 run: (input) => {
282415 const staging = input.args.includes("add") && input.args.includes("-A");
283 if (staging) privateIndex = input.env?.GIT_INDEX_FILE;
416 if (staging) {
417 privateIndex = input.env?.GIT_INDEX_FILE;
418 stagingAttempts += 1;
419 }
284420 const block =
285421 (blockedPhase === "discovery" && input.args.includes("--others")) ||
286422 (blockedPhase === "probe" && input.cwd !== cwd && input.args.includes("rev-parse")) ||
......@@ -322,7 +458,8 @@ for (const blockedPhase of ["discovery", "probe", "retry"] as const) {
322458 },
323459 }),
324460 );
325 const fiber = yield* captureDriver.checkpoints
461 const captureStore = yield* makeCaptureStore(captureDriver, cwd);
462 const fiber = yield* captureStore
326463 .captureCheckpoint({ cwd, checkpointRef })
327464 .pipe(Effect.flip, Effect.forkScoped);
328465 if (blockedPhase === "probe") {
......@@ -333,6 +470,7 @@ for (const blockedPhase of ["discovery", "probe", "retry"] as const) {
333470 yield* TestClock.adjust(blockedPhase === "probe" ? "2 seconds" : "5 seconds");
334471 const error = yield* Fiber.join(fiber);
335472 assert.strictEqual(error, stageError);
473 assert.strictEqual(stagingAttempts, blockedPhase === "retry" ? 2 : 1);
336474 assert.isTrue(interrupted);
337475 assert.isDefined(privateIndex);
338476 assert.isFalse(yield* fs.exists(privateIndex!));
apps/server/src/vcs/GitVcsDriver.ts+1-1
......@@ -766,7 +766,7 @@ export const makeVcsDriverShape = Effect.fn("makeGitVcsDriverShape")(function* (
766766
767767 const checkpoints: VcsDriver.VcsCheckpointOps = {
768768 captureCheckpoint: Effect.fn("GitVcsDriver.checkpoints.captureCheckpoint")(function* (input) {
769 const operation = "GitVcsDriver.checkpoints.captureCheckpoint";
769 const operation = VcsProcess.CHECKPOINT_CAPTURE_OPERATION;
770770 const indexConfig = [
771771 "-c",
772772 "core.fsmonitor=false",
apps/server/src/vcs/VcsProcess.test.ts+169
......@@ -4,10 +4,13 @@ import { HostProcessWorkingDirectory } from "@t3tools/shared/hostProcess";
44import * as Duration from "effect/Duration";
55import * as Deferred from "effect/Deferred";
66import * as Effect from "effect/Effect";
7import * as Exit from "effect/Exit";
8import * as Cause from "effect/Cause";
79import * as Fiber from "effect/Fiber";
810import * as Layer from "effect/Layer";
911import * as Queue from "effect/Queue";
1012import * as Ref from "effect/Ref";
13import * as Schema from "effect/Schema";
1114import { TestClock } from "effect/testing";
1215import { ChildProcessSpawner } from "effect/unstable/process";
1316
......@@ -19,6 +22,8 @@ import {
1922import * as ProcessRunner from "../processRunner.ts";
2023import * as VcsProcess from "./VcsProcess.ts";
2124
25const encodeExitError = Schema.encodeEffect(Schema.fromJsonString(VcsProcessExitError));
26
2227const run = (input: VcsProcess.VcsProcessInput) =>
2328 Effect.gen(function* () {
2429 const process = yield* VcsProcess.VcsProcess;
......@@ -50,6 +55,170 @@ const captureProcessResult = (
5055 );
5156
5257describe("VcsProcess.run", () => {
58 it.effect.each([
59 { stderr: "fatal: Unable to create '/private/repo/index.lock': File exists", retryable: true },
60 {
61 stderr: "fatal: Unable to create '/private/repo/index.lock': Permission denied",
62 retryable: false,
63 },
64 { stderr: 'error: open("/private/repo/file"): No such file or directory', retryable: true },
65 {
66 stderr: "fatal: unable to stat '/private/repo/file': No such file or directory",
67 retryable: true,
68 },
69 { stderr: 'error: open("/private/repo/file"): Permission denied', retryable: false },
70 {
71 stderr: "error: '/private/repo/child/' does not have a commit checked out",
72 retryable: false,
73 },
74 { stderr: "fatal: index file corrupt", retryable: false },
75 ])(
76 "classifies Git exit retryability without retaining stderr: $stderr",
77 ({ stderr, retryable }) =>
78 Effect.gen(function* () {
79 const error = yield* captureProcessResult(
80 Effect.succeed({
81 stdout: "",
82 stderr,
83 code: ChildProcessSpawner.ExitCode(128),
84 timedOut: false,
85 stdoutTruncated: false,
86 stderrTruncated: false,
87 stdoutInvalidUtf8: false,
88 stderrInvalidUtf8: false,
89 }),
90 );
91 assert.instanceOf(error, VcsProcessExitError);
92 expect(error.retryable === true).toBe(retryable);
93 const encoded = yield* encodeExitError(error);
94 expect(encoded).not.toContain("/private/repo");
95 }),
96 );
97
98 it.effect.each(["timeout", "spawn"] as const)(
99 "checkpoint commands do not retry %s failures",
100 (kind) =>
101 Effect.gen(function* () {
102 let attempts = 0;
103 const fields = { command: "git", argumentCount: 2, cwd: "/workspace" };
104 const failure =
105 kind === "timeout"
106 ? new ProcessRunner.ProcessTimeoutError({ ...fields, timeoutMs: 30_000 })
107 : new ProcessRunner.ProcessSpawnError({ ...fields, cause: new Error("spawn failed") });
108 const service = yield* VcsProcess.make.pipe(
109 Effect.provideService(ProcessRunner.ProcessRunner, {
110 run: () =>
111 Effect.sync(() => {
112 attempts += 1;
113 }).pipe(Effect.andThen(Effect.fail(failure))),
114 }),
115 );
116 const error = yield* service
117 .run({
118 ...baseInput,
119 operation: VcsProcess.CHECKPOINT_CAPTURE_OPERATION,
120 })
121 .pipe(Effect.flip);
122 expect(attempts).toBe(1);
123 expect(error._tag).toBe(
124 kind === "timeout" ? "VcsProcessTimeoutError" : "VcsProcessSpawnError",
125 );
126 }),
127 );
128
129 it.effect.each([
130 {
131 name: "recovers after two lock failures",
132 failures: 2,
133 attempts: 3,
134 transient: true,
135 capture: true,
136 streaming: false,
137 },
138 {
139 name: "bounds persistent lock failures",
140 failures: Infinity,
141 attempts: 3,
142 transient: true,
143 capture: true,
144 streaming: false,
145 },
146 {
147 name: "does not retry unknown exits",
148 failures: Infinity,
149 attempts: 1,
150 transient: false,
151 capture: true,
152 streaming: false,
153 },
154 {
155 name: "leaves other operations unchanged",
156 failures: Infinity,
157 attempts: 1,
158 transient: true,
159 capture: false,
160 streaming: false,
161 },
162 {
163 name: "does not replay stdout callbacks",
164 failures: Infinity,
165 attempts: 1,
166 transient: true,
167 capture: true,
168 streaming: true,
169 },
170 ])(
171 "checkpoint command retry $name",
172 ({ failures, attempts: expectedAttempts, transient, capture, streaming }) =>
173 Effect.gen(function* () {
174 let attempts = 0;
175 const service = yield* VcsProcess.make.pipe(
176 Effect.provideService(ProcessRunner.ProcessRunner, {
177 run: () =>
178 Effect.sync(() => {
179 attempts += 1;
180 return {
181 stdout: "",
182 stderr:
183 attempts > failures
184 ? ""
185 : transient
186 ? "fatal: Unable to create '/repo/index.lock': File exists"
187 : "fatal: index file corrupt",
188 code: ChildProcessSpawner.ExitCode(attempts > failures ? 0 : attempts),
189 timedOut: false,
190 stdoutTruncated: false,
191 stderrTruncated: false,
192 stdoutInvalidUtf8: false,
193 stderrInvalidUtf8: false,
194 };
195 }),
196 }),
197 );
198 const fiber = yield* service
199 .run({
200 ...baseInput,
201 operation: capture ? VcsProcess.CHECKPOINT_CAPTURE_OPERATION : baseInput.operation,
202 ...(streaming ? { onStdoutChunk: () => {} } : {}),
203 })
204 .pipe(Effect.exit, Effect.forkScoped);
205 yield* TestClock.adjust("150 millis");
206 const result = yield* Fiber.join(fiber);
207 expect(attempts).toBe(expectedAttempts);
208 if (failures === Infinity) {
209 if (Exit.isSuccess(result))
210 return yield* Effect.die("Expected a capture command failure");
211 const error = Cause.findErrorOption(result.cause);
212 if (error._tag === "None" || error.value._tag !== "VcsProcessExitError") {
213 return yield* Effect.die("Expected a Git exit error");
214 }
215 expect(error.value.exitCode).toBe(expectedAttempts);
216 } else {
217 expect(Exit.isSuccess(result)).toBe(true);
218 }
219 }),
220 );
221
53222 it.effect("bounds a synthetic burst of GitHub API processes", () =>
54223 Effect.gen(function* () {
55224 const gate = yield* Deferred.make<void>();
apps/server/src/vcs/VcsProcess.ts+34-1
......@@ -1,7 +1,9 @@
11import * as Context from "effect/Context";
2import * as Duration from "effect/Duration";
23import * as Effect from "effect/Effect";
34import * as Layer from "effect/Layer";
45import * as Match from "effect/Match";
6import * as Schedule from "effect/Schedule";
57import * as Semaphore from "effect/Semaphore";
68import { ChildProcessSpawner } from "effect/unstable/process";
79
......@@ -58,6 +60,8 @@ const OUTPUT_TRUNCATED_MARKER = "\n\n[truncated]";
5860const VCS_PROCESS_CONCURRENCY = 8;
5961const GITHUB_PROCESS_CONCURRENCY = 4;
6062
63export const CHECKPOINT_CAPTURE_OPERATION = "GitVcsDriver.checkpoints.captureCheckpoint";
64
6165const classifyNonZeroExit = (command: string, stderr: string): VcsProcessExitFailureKind => {
6266 const normalized = stderr.toLowerCase();
6367
......@@ -104,6 +108,11 @@ const classifyNonZeroExit = (command: string, stderr: string): VcsProcessExitFai
104108 return "command-failed";
105109};
106110
111// Classify before discarding stderr; keep paths and process output out of errors.
112const isTransientGitExit = (stderr: string) =>
113 /unable to create [^\n]*\.lock['"]?: file exists/i.test(stderr) ||
114 /(?:unable to stat|lstat\(|error: open\()[^\n]+: no such file or directory/i.test(stderr);
115
107116export const make = Effect.gen(function* () {
108117 const processRunner = yield* ProcessRunner.ProcessRunner;
109118 const vcsProcesses = yield* Semaphore.make(VCS_PROCESS_CONCURRENCY);
......@@ -167,6 +176,7 @@ export const make = Effect.gen(function* () {
167176 }
168177
169178 if (!input.allowNonZeroExit && result.code !== 0) {
179 const failureKind = classifyNonZeroExit(input.command, result.stderr);
170180 return yield* VcsProcessExitError.fromProcessExit(
171181 baseError,
172182 {
......@@ -174,7 +184,10 @@ export const make = Effect.gen(function* () {
174184 stderr: result.stderr,
175185 stderrTruncated: result.stderrTruncated,
176186 },
177 classifyNonZeroExit(input.command, result.stderr),
187 failureKind,
188 input.command === "git" &&
189 failureKind === "command-failed" &&
190 isTransientGitExit(result.stderr),
178191 );
179192 }
180193
......@@ -191,6 +204,26 @@ export const make = Effect.gen(function* () {
191204
192205 const run = Effect.fn("VcsProcess.run")(function* (input: VcsProcessInput) {
193206 const bounded = vcsProcesses.withPermits(1)(runUnbounded(input));
207 if (
208 input.command === "git" &&
209 input.operation === CHECKPOINT_CAPTURE_OPERATION &&
210 input.onStdoutChunk === undefined
211 ) {
212 // Retry the failed command, retaining the private index/tree and recovery's outer deadline.
213 return yield* bounded.pipe(
214 Effect.tapError((error) =>
215 Effect.logDebug("checkpoint Git command failed", {
216 operation: input.operation,
217 errorTag: error._tag,
218 }),
219 ),
220 Effect.retry({
221 times: 2,
222 while: (error) => error._tag === "VcsProcessExitError" && error.retryable === true,
223 schedule: Schedule.spaced(Duration.millis(75)),
224 }),
225 );
226 }
194227 return yield* input.command === "gh" ? githubProcesses.withPermits(1)(bounded) : bounded;
195228 });
196229
packages/contracts/src/vcs.ts+4
......@@ -119,6 +119,8 @@ export class VcsProcessExitError extends Schema.TaggedError<VcsProcessExitError>
119119 exitCode: Schema.Number,
120120 detail: Schema.String,
121121 failureKind: Schema.optional(VcsProcessExitFailureKind),
122 /** Process-boundary hint for a recognized transient failure; absence is not retryable. */
123 retryable: Schema.optional(Schema.Boolean),
122124 stderrLength: Schema.optional(NonNegativeInt),
123125 stderrTruncated: Schema.optional(Schema.Boolean),
124126 },
......@@ -131,6 +133,7 @@ export class VcsProcessExitError extends Schema.TaggedError<VcsProcessExitError>
131133 context: VcsProcessErrorContext,
132134 error: VcsProcessExitFailure,
133135 failureKind: VcsProcessExitFailureKind,
136 retryable?: boolean,
134137 ) {
135138 const detail =
136139 failureKind === "authentication"
......@@ -150,6 +153,7 @@ export class VcsProcessExitError extends Schema.TaggedError<VcsProcessExitError>
150153 exitCode: error.exitCode,
151154 detail,
152155 failureKind,
156 ...(retryable === true ? { retryable: true } : {}),
153157 stderrLength: error.stderr.length,
154158 stderrTruncated: error.stderrTruncated,
155159 });