1import * as Context from "effect/Context";
2import * as Duration from "effect/Duration";
3import * as Effect from "effect/Effect";
4import * as Layer from "effect/Layer";
5import * as Match from "effect/Match";
6import * as Schedule from "effect/Schedule";
7import * as Semaphore from "effect/Semaphore";
8import { ChildProcessSpawner } from "effect/process";
9
10import {
11 type VcsError,
12 type VcsProcessErrorContext,
13 VcsProcessExitError,
14 type VcsProcessExitFailureKind,
15 VcsProcessMissingExitCodeError,
16 VcsProcessOutputLimitError,
17 VcsProcessOutputReadError,
18 VcsProcessSpawnError,
19 VcsProcessStdinWriteError,
20 VcsProcessTimeoutError,
21} from "@t3tools/contracts";
22import * as HostProcess from "@t3tools/shared/HostProcess";
23import type { SourceControlProcessInput } from "@t3tools/source-control-core/server/SourceControlHost";
24
25import { gitCommandDuration, gitCommandsTotal, withMetrics } from "../observability/Metrics.ts";
26import * as ProcessRunner from "../processRunner.ts";
27
28export interface VcsProcessInput {
29 readonly operation: string;
30 readonly command: string;
31 readonly args: ReadonlyArray<string>;
32 readonly cwd: string;
33 readonly spawnCwd?: string;
34 readonly stdin?: string;
35 readonly onStdoutChunk?: (chunk: Uint8Array) => void;
36 readonly env?: NodeJS.ProcessEnv;
37 readonly allowNonZeroExit?: boolean;
38 readonly timeoutMs?: number;
39 readonly maxOutputBytes?: number;
40 readonly outputMode?: ProcessRunner.ProcessRunInput["outputMode"];
41 readonly appendTruncationMarker?: boolean;
42 readonly classifyFailure?: SourceControlProcessInput["classifyFailure"];
43 readonly notFoundDetail?: string;
44 readonly concurrency?: SourceControlProcessInput["concurrency"];
45}
46
47export interface VcsProcessOutput {
48 readonly exitCode: ChildProcessSpawner.ExitCode;
49 readonly stdout: string;
50 readonly stderr: string;
51 readonly stdoutTruncated: boolean;
52 readonly stderrTruncated: boolean;
53 readonly stdoutInvalidUtf8?: boolean;
54 readonly stderrInvalidUtf8?: boolean;
55}
56
57export class VcsProcess extends Context.Service<
58 VcsProcess,
59 {
60 readonly run: (input: VcsProcessInput) => Effect.Effect<VcsProcessOutput, VcsError>;
61 }
62>()("t3/vcs/VcsProcess") {}
63
64const DEFAULT_TIMEOUT_MS = 30_000;
65const DEFAULT_MAX_OUTPUT_BYTES = 1_000_000;
66const OUTPUT_TRUNCATED_MARKER = "\n\n[truncated]";
67const VCS_PROCESS_CONCURRENCY = 8;
68// Forge CLIs can take seconds per call; cap them together so git, checkpoints included, keeps slots.
69const NON_GIT_PROCESS_CONCURRENCY = VCS_PROCESS_CONCURRENCY / 2;
70
71/** Shown to users when a Git command fails because another process holds the index lock. */
72export const GIT_INDEX_LOCKED_DETAIL =
73 "Git's index is locked. Wait for other Git operations to finish, then try again.";
74
75export const CHECKPOINT_CAPTURE_OPERATION = "GitVcsDriver.checkpoints.captureCheckpoint";
76
77// Caller operations are diagnostic strings; only known labels may become metric attributes.
78const gitMetricOperations = new Set([
79 CHECKPOINT_CAPTURE_OPERATION,
80 "GitVcsDriver.checkpoints.deleteCheckpointRefs",
81 "GitVcsDriver.checkpoints.diffCheckpoints",
82 "GitVcsDriver.checkpoints.hasHeadCommit",
83 "GitVcsDriver.checkpoints.listAuthoredPaths",
84 "GitVcsDriver.checkpoints.resolveCheckpointCommit",
85 "GitVcsDriver.checkpoints.resolveGitCommonDir",
86 "GitVcsDriver.checkpoints.resolveHeadCommit",
87 "GitVcsDriver.checkpoints.restoreCheckpoint",
88 "GitVcsDriver.detectRepository",
89 "GitVcsDriver.filterIgnoredPaths",
90 "GitVcsDriver.initRepository",
91 "GitVcsDriver.isInsideWorkTree",
92 "GitVcsDriver.listRemotes",
93 "GitVcsDriver.listWorkspaceFiles",
94 "GitVcsDriver.resolveGitCommonDir",
95]);
96
97/**
98 * Append after inherited entries without changing argv or the user's config.
99 * Windows env keys ignore case: read the last spelling of the count so command overrides win,
100 * then update every spelling because Node may keep any one.
101 * Leave invalid counts untouched so Git reports them.
102 */
103export const gitConfigEnv = (
104 env: NodeJS.ProcessEnv,
105 key: string,
106 value: string,
107 platform: NodeJS.Platform,
108): NodeJS.ProcessEnv => {
109 const countKeys = Object.keys(env).filter((name) =>
110 platform === "win32" ? name.toUpperCase() === "GIT_CONFIG_COUNT" : name === "GIT_CONFIG_COUNT",
111 );
112 const countKey = countKeys.at(-1) ?? "GIT_CONFIG_COUNT";
113 const inherited = env[countKey];
114 if (
115 inherited !== undefined &&
116 inherited !== "" &&
117 // Match Git's strtoul count syntax, including whitespace, signs and non-negative -0.
118 /^[ \t\r\n\v\f]*(?:\+?\d+|-0+)/.exec(inherited)?.[0] !== inherited
119 ) {
120 return {};
121 }
122 const count = inherited === undefined || inherited === "" ? 0 : Number.parseInt(inherited, 10);
123 return {
124 ...Object.fromEntries([...countKeys, countKey].map((name) => [name, String(count + 1)])),
125 [`GIT_CONFIG_KEY_${count}`]: key,
126 [`GIT_CONFIG_VALUE_${count}`]: value,
127 };
128};
129
130/**
131 * A project can contain a fake bare repository whose config runs commands such as `core.fsmonitor`.
132 * Require `--git-dir` or `GIT_DIR` to open bare repositories; Git before 2.38 ignores this setting.
133 */
134export const explicitBareRepositoryEnv = (env: NodeJS.ProcessEnv, platform: NodeJS.Platform) =>
135 gitConfigEnv(env, "safe.bareRepository", "explicit", platform);
136
137const REPOSITORY_SELECTORS = new Set([
138 "GIT_DIR",
139 "GIT_WORK_TREE",
140 "GIT_COMMON_DIR",
141 "GIT_INDEX_FILE",
142 "GIT_OBJECT_DIRECTORY",
143 "GIT_ALTERNATE_OBJECT_DIRECTORIES",
144]);
145
146/**
147 * Unsets repository selectors the server inherited, such as a Git hook's `GIT_DIR`, so Git finds
148 * the repository from cwd. Spread it before a command's own env, which may still set them.
149 */
150export const unsetInheritedRepositoryEnv = (
151 inherited: NodeJS.ProcessEnv,
152 platform: NodeJS.Platform,
153): NodeJS.ProcessEnv =>
154 Object.fromEntries(
155 Object.keys(inherited)
156 .filter((name) => REPOSITORY_SELECTORS.has(platform === "win32" ? name.toUpperCase() : name))
157 .map((name) => [name, undefined]),
158 );
159
160const classifyNonZeroExit = (
161 stderr: string,
162 classifyFailure: VcsProcessInput["classifyFailure"],
163): VcsProcessExitFailureKind => {
164 const normalized = stderr.toLowerCase();
165 const commandKind = classifyFailure?.(stderr) ?? null;
166
167 if (
168 commandKind === "authentication" ||
169 normalized.includes("authentication failed") ||
170 normalized.includes("not logged in") ||
171 normalized.includes("no oauth token") ||
172 normalized.includes("unauthorized")
173 ) {
174 return "authentication";
175 }
176
177 if (
178 commandKind === "rate-limited" ||
179 normalized.includes("api rate limit") ||
180 normalized.includes("rate limit exceeded") ||
181 normalized.includes("secondary rate limit") ||
182 normalized.includes("too many requests") ||
183 normalized.includes("http 429")
184 ) {
185 return "rate-limited";
186 }
187
188 return commandKind ?? "command-failed";
189};
190
191// Classify before discarding stderr; keep paths and process output out of errors.
192const isTransientGitExit = (stderr: string) =>
193 /unable to create [^\n]*\.lock['"]?: file exists/i.test(stderr) ||
194 /(?:unable to stat|lstat\(|error: open\()[^\n]+: no such file or directory/i.test(stderr);
195
196// On macOS, Git's fsync uses F_FULLFSYNC; writeout-only still uses fsync(2).
197// Network mounts can reject F_FULLFSYNC. Ref writes omit errno, so retry their
198// bare write error too; explicit unrelated errors remain failures.
199// Linux writeout-only only starts writeback, so it must not use this fallback.
200const FULL_FSYNC_ARG = "core.fsyncMethod=fsync";
201const BATCH_FSYNC_ARG = "core.fsyncMethod=batch";
202const WRITEOUT_ONLY_FSYNC_ARG = "core.fsyncMethod=writeout-only";
203const canRetryFullFsync = (stderr: string) =>
204 /fsync error on [^\r\n]+: operation not supported\r?$/im.test(stderr) ||
205 /cannot update ref [^\r\n]+: couldn't write '[^\r\n]+\.lock'\r?$/im.test(stderr);
206
207export const make = Effect.gen(function* () {
208 const processRunner = yield* ProcessRunner.ProcessRunner;
209 const hostPlatform = yield* HostProcess.Platform;
210 const vcsProcesses = yield* Semaphore.make(VCS_PROCESS_CONCURRENCY);
211 const nonGitProcesses = yield* Semaphore.make(NON_GIT_PROCESS_CONCURRENCY);
212 const keyedProcesses = new Map<string, Semaphore.Semaphore>();
213 const keyedPermit = (concurrency: NonNullable<VcsProcessInput["concurrency"]>) => {
214 let semaphore = keyedProcesses.get(concurrency.key);
215 if (semaphore === undefined) {
216 semaphore = Semaphore.makeUnsafe(concurrency.limit);
217 keyedProcesses.set(concurrency.key, semaphore);
218 }
219 return semaphore.withPermits(1);
220 };
221
222 const spawn = Effect.fn("VcsProcess.spawn")(function* (
223 input: VcsProcessInput,
224 baseError: VcsProcessErrorContext,
225 ) {
226 let failedExit = false;
227 const execution = processRunner
228 .run({
229 command: input.command,
230 args: input.args,
231 cwd: input.cwd,
232 ...(input.spawnCwd !== undefined ? { spawnCwd: input.spawnCwd } : {}),
233 ...(input.stdin !== undefined ? { stdin: input.stdin } : {}),
234 ...(input.onStdoutChunk !== undefined ? { onStdoutChunk: input.onStdoutChunk } : {}),
235 env: {
236 ...(input.command === "git"
237 ? unsetInheritedRepositoryEnv(process.env, hostPlatform)
238 : {}),
239 // Failures are read by English patterns; without LANG, macOS Git follows the system language.
240 // LANGUAGE leaves LC_CTYPE, which hooks inherit, alone.
241 ...(input.command === "git" ? { LANGUAGE: "C" } : {}),
242 ...input.env,
243 ...explicitBareRepositoryEnv({ ...process.env, ...input.env }, hostPlatform),
244 },
245 timeout: input.timeoutMs ?? DEFAULT_TIMEOUT_MS,
246 maxOutputBytes: input.maxOutputBytes ?? DEFAULT_MAX_OUTPUT_BYTES,
247 outputMode: input.outputMode ?? "truncate",
248 truncatedMarker: input.appendTruncationMarker ? OUTPUT_TRUNCATED_MARKER : "",
249 timeoutBehavior: "error",
250 })
251 .pipe(
252 Effect.tap((result) => {
253 failedExit = result.code === null || (!input.allowNonZeroExit && result.code !== 0);
254 return Effect.void;
255 }),
256 Effect.mapError(
257 Match.valueTags({
258 ProcessSpawnError: (error) =>
259 VcsProcessSpawnError.fromProcessSpawnError(baseError, error),
260 ProcessOutputLimitError: (error) =>
261 new VcsProcessOutputLimitError({
262 ...baseError,
263 stream: error.stream,
264 maxBytes: error.maxBytes,
265 observedBytes: error.observedBytes,
266 }),
267 ProcessTimeoutError: (error) =>
268 VcsProcessTimeoutError.fromProcessTimeoutError(baseError, error),
269 ProcessStdinError: (error) =>
270 new VcsProcessStdinWriteError({
271 ...baseError,
272 stdinBytes: error.stdinBytes,
273 cause: error.cause,
274 }),
275 ProcessReadError: (error) =>
276 new VcsProcessOutputReadError({
277 ...baseError,
278 stream: error.stream,
279 cause: error.cause,
280 }),
281 }),
282 ),
283 );
284
285 return yield* input.command === "git"
286 ? execution.pipe(
287 withMetrics({
288 counter: gitCommandsTotal,
289 timer: gitCommandDuration,
290 attributes: {
291 operation: gitMetricOperations.has(input.operation) ? input.operation : "other",
292 },
293 outcomeAttributes: (outcome) => ({
294 outcome: outcome === "success" && failedExit ? "failure" : outcome,
295 }),
296 }),
297 )
298 : execution;
299 });
300
301 const runUnbounded = Effect.fn("VcsProcess.runUnbounded")(function* (input: VcsProcessInput) {
302 const baseError = {
303 operation: input.operation,
304 command: input.command,
305 cwd: input.cwd,
306 argumentCount: input.args.length,
307 };
308
309 let result = yield* spawn(input, baseError);
310
311 if (
312 hostPlatform === "darwin" &&
313 input.operation === CHECKPOINT_CAPTURE_OPERATION &&
314 !input.allowNonZeroExit &&
315 result.code !== null &&
316 result.code !== 0 &&
317 input.command === "git" &&
318 (input.args.includes(FULL_FSYNC_ARG) || input.args.includes(BATCH_FSYNC_ARG)) &&
319 canRetryFullFsync(result.stderr)
320 ) {
321 result = yield* spawn(
322 {
323 ...input,
324 args: input.args.map((arg) =>
325 arg === FULL_FSYNC_ARG || arg === BATCH_FSYNC_ARG ? WRITEOUT_ONLY_FSYNC_ARG : arg,
326 ),
327 },
328 baseError,
329 );
330 }
331
332 if (result.code === null) {
333 return yield* new VcsProcessMissingExitCodeError(baseError);
334 }
335
336 if (!input.allowNonZeroExit && result.code !== 0) {
337 const failureKind = classifyNonZeroExit(result.stderr, input.classifyFailure);
338 const error = VcsProcessExitError.fromProcessExit(
339 baseError,
340 {
341 exitCode: result.code,
342 stderr: result.stderr,
343 stderrTruncated: result.stderrTruncated,
344 },
345 failureKind,
346 {
347 retryable:
348 input.command === "git" &&
349 failureKind === "command-failed" &&
350 isTransientGitExit(result.stderr),
351 ...(input.notFoundDetail === undefined ? {} : { notFoundDetail: input.notFoundDetail }),
352 },
353 );
354 if (
355 input.command === "git" &&
356 /^fatal: unable to create ['"][^\r\n]*[/\\]index\.lock['"]: file exists\.?$/im.test(
357 result.stderr,
358 )
359 ) {
360 return yield* new VcsProcessExitError({
361 ...error,
362 detail: GIT_INDEX_LOCKED_DETAIL,
363 });
364 }
365 return yield* error;
366 }
367
368 return {
369 exitCode: result.code,
370 stdout: result.stdout,
371 stderr: result.stderr,
372 stdoutTruncated: result.stdoutTruncated,
373 stderrTruncated: result.stderrTruncated,
374 stdoutInvalidUtf8: result.stdoutInvalidUtf8 ?? false,
375 stderrInvalidUtf8: result.stderrInvalidUtf8 ?? false,
376 } satisfies VcsProcessOutput;
377 });
378
379 const run = Effect.fn("VcsProcess.run")(function* (input: VcsProcessInput) {
380 const bounded = vcsProcesses.withPermits(1)(runUnbounded(input));
381 if (
382 input.command === "git" &&
383 input.operation === CHECKPOINT_CAPTURE_OPERATION &&
384 input.onStdoutChunk === undefined
385 ) {
386 return yield* bounded.pipe(
387 Effect.tapError((error) =>
388 Effect.logDebug("checkpoint Git command failed", {
389 operation: input.operation,
390 errorTag: error._tag,
391 }),
392 ),
393 Effect.retry({
394 times: 2,
395 while: (error) => error._tag === "VcsProcessExitError" && error.retryable === true,
396 schedule: Schedule.spaced(Duration.millis(75)),
397 }),
398 );
399 }
400 const shared = input.command === "git" ? bounded : nonGitProcesses.withPermits(1)(bounded);
401 return yield* input.concurrency === undefined ? shared : keyedPermit(input.concurrency)(shared);
402 });
403
404 return VcsProcess.of({ run });
405});
406
407export const layer = Layer.effect(VcsProcess, make).pipe(Layer.provide(ProcessRunner.layer));