Refactors Graph commit streaming to improve perf

Reworks `Git.stream` to properly handle errors
Adds `Git.stream` support to Live Share (hacky)
Removes unused `Git.spawn` and `Git.logStreamTo` methods

Author: eamodio

src/env/browser/providers.ts

@@ -13,16 +13,6 @@ export function git(_options: GitCommandOptions, ..._args: any[]): Promise<GitRe
 	return Promise.resolve({ stdout: '', exitCode: 0 });
 }
 
-export function gitLogStreamTo(
-	_repoPath: string,
-	_sha: string,
-	_limit: number,
-	_options?: { configs?: readonly string[]; stdin?: string },
-	..._args: string[]
-): Promise<[data: string[], count: number]> {
-	return Promise.resolve([[''], 0]);
-}
-
 export function getSupportedGitProviders(container: Container): Promise<GitProvider[]> {
 	return Promise.resolve([new GitHubGitProvider(container)]);
 }

src/env/node/git/git.ts

@@ -1,4 +1,4 @@
-import type { ChildProcess, SpawnOptions } from 'child_process';
+import type { SpawnOptions } from 'child_process';
 import { spawn } from 'child_process';
 import { accessSync } from 'fs';
 import { join as joinPath } from 'path';
@@ -349,7 +349,7 @@ export class Git {
 		}
 	}
 
-	async spawn(options: GitSpawnOptions, ...args: readonly (string | undefined)[]): Promise<ChildProcess> {
+	async *stream(options: GitSpawnOptions, ...args: readonly (string | undefined)[]): AsyncGenerator<string> {
 		if (!workspace.isTrusted) throw new WorkspaceUntrustedError();
 
 		const start = hrtime();
@@ -389,10 +389,21 @@ export class Git {
 			runArgs.unshift('-c', 'core.longpaths=true');
 		}
 
-		if (cancellation) {
+		let disposable: Disposable | undefined;
+		if (cancellation != null) {
 			const aborter = new AbortController();
+			const onAbort = () => aborter.abort();
+
+			const signal = spawnOpts.signal;
+			disposable = {
+				dispose: () => {
+					cancellation?.onCancellationRequested(onAbort);
+					signal?.removeEventListener('abort', onAbort);
+				},
+			};
+
+			spawnOpts.signal?.addEventListener('abort', onAbort);
 			spawnOpts.signal = aborter.signal;
-			cancellation.onCancellationRequested(() => aborter.abort());
 		}
 
 		const command = await this.path();
@@ -402,111 +413,73 @@ export class Git {
 		}
 
 		let exception: Error | undefined;
-		proc.once('error', ex => {
-			if (ex?.name === 'AbortError') {
-				exception = new CancellationError(new CancelledRunError(command, true));
-			} else {
-				exception = new GitError(ex);
-			}
-		});
-		proc.once('exit', () => this.logGitCommand(gitCommand, exception, getDurationMilliseconds(start), false));
-		return proc;
-	}
-
-	async *stream(options: GitSpawnOptions, ...args: readonly (string | undefined)[]): AsyncGenerator<string> {
-		if (!workspace.isTrusted) throw new WorkspaceUntrustedError();
-
-		const start = hrtime();
-
-		const { cancellation, configs, stdin, stdinEncoding, ...opts } = options;
-		const runArgs = args.filter(a => a != null);
 
-		const spawnOpts: SpawnOptions = {
-			// Unless provided, ignore stdin and leave default streams for stdout and stderr
-			stdio: [stdin ? 'pipe' : 'ignore', null, null],
-			...opts,
-			// Adds GCM environment variables to avoid any possible credential issues -- from https://github.com/Microsoft/vscode/issues/26573#issuecomment-338686581
-			// Shouldn't *really* be needed but better safe than sorry
-			env: {
-				...process.env,
-				...this._gitEnv,
-				...(options.env ?? emptyObj),
-				GCM_INTERACTIVE: 'NEVER',
-				GCM_PRESERVE_CREDS: 'TRUE',
-				LC_ALL: 'C',
-			},
-		};
-
-		const gitCommand = `(spawn) [${spawnOpts.cwd as string}] git ${runArgs.join(' ')}`;
-
-		// Fixes https://github.com/gitkraken/vscode-gitlens/issues/73 & https://github.com/gitkraken/vscode-gitlens/issues/161
-		// See https://stackoverflow.com/questions/4144417/how-to-handle-asian-characters-in-file-names-in-git-on-os-x
-		runArgs.unshift(
-			'-c',
-			'core.quotepath=false',
-			'-c',
-			'color.ui=false',
-			...(configs !== undefined ? configs : emptyArray),
-		);
-
-		if (process.platform === 'win32') {
-			runArgs.unshift('-c', 'core.longpaths=true');
-		}
+		const promise = new Promise<void>((resolve, reject) => {
+			const stderrChunks: string[] = [];
+			if (proc.stderr) {
+				proc.stderr?.setEncoding('utf8');
+				proc.stderr.on('data', chunk => stderrChunks.push(chunk));
+			}
 
-		const aborter = new AbortController();
-		spawnOpts.signal = aborter.signal;
-		cancellation?.onCancellationRequested(() => aborter.abort());
+			proc.once('error', ex => {
+				if (ex?.name === 'AbortError') return;
 
-		const command = await this.path();
-		const proc = spawn(command, runArgs, spawnOpts);
-		if (stdin) {
-			proc.stdin?.end(stdin, (stdinEncoding ?? 'utf8') as BufferEncoding);
-		}
+				exception = new GitError(ex);
+			});
+			proc.once('close', (code, signal) => {
+				// If the process exited normally or the caller didn't iterate over the complete stream, just resolve
+				if (code === 0 || code === 141 /* SIGPIPE */) {
+					resolve();
+					return;
+				}
 
-		let completed = false;
-		let exception: Error | undefined;
+				if (signal === 'SIGTERM') {
+					reject(
+						new CancellationError(
+							new CancelledRunError(proc.spawnargs.join(' '), true, code ?? undefined, signal),
+						),
+					);
+					return;
+				}
 
-		proc.once(
-			'error',
-			ex =>
-				(exception =
-					ex?.name === 'AbortError'
-						? new CancellationError(new CancelledRunError(command, true))
-						: new GitError(ex)),
-		);
-		proc.once('exit', (code, signal) => {
-			if (signal === 'SIGTERM') {
-				exception = new CancellationError(new CancelledRunError(command, true, code ?? undefined, signal));
-			}
-			this.logGitCommand(gitCommand, exception, getDurationMilliseconds(start), false);
+				const stderr = stderrChunks.join('').trim();
+				reject(
+					new GitError(
+						new RunError(
+							{
+								message: `Error (${code}): ${stderr || 'Unknown'}`,
+								cmd: proc.spawnargs.join(' '),
+								killed: proc.killed,
+								code: proc.exitCode,
+							},
+							'',
+							stderr,
+						),
+					),
+				);
+			});
 		});
 
 		try {
-			if (!proc.stdout) {
-				aborter.abort();
-				throw new Error('Spawned Git process has no stdout');
-			}
-			proc.stdout.setEncoding('utf8');
-
-			for await (const chunk of proc.stdout) {
-				if (exception != null) {
-					if (isCancellationError(exception)) {
-						// TODO: Should we throw here?
-						break;
-					} else {
-						throw exception;
+			try {
+				if (proc.stdout) {
+					proc.stdout.setEncoding('utf8');
+					for await (const chunk of proc.stdout) {
+						yield chunk;
 					}
 				}
-
-				yield chunk;
+			} finally {
+				// I have NO idea why this HAS to be in a finally block, but it does
+				await promise;
 			}
-
-			completed = true;
+		} catch (ex) {
+			exception = ex;
+			throw ex;
 		} finally {
-			// If we didn't complete the iteration, then abort the process
-			if (!completed) {
-				aborter.abort();
-			}
+			disposable?.dispose();
+			proc.removeAllListeners();
+
+			this.logGitCommand(gitCommand, exception, getDurationMilliseconds(start), false);
 		}
 	}
 
@@ -1101,93 +1074,6 @@ export class Git {
 		}
 	}
 
-	async logStreamTo(
-		repoPath: string,
-		sha: string,
-		limit: number,
-		options?: { cancellation?: CancellationToken; configs?: readonly string[]; stdin?: string },
-		...args: string[]
-	): Promise<[data: string[], count: number]> {
-		const params = ['log', ...args];
-		if (options?.stdin) {
-			params.push('--stdin');
-		}
-
-		const proc = await this.spawn(
-			{
-				cwd: repoPath,
-				cancellation: options?.cancellation,
-				configs: options?.configs ?? gitLogDefaultConfigs,
-				stdin: options?.stdin,
-			},
-			...params,
-			'--',
-		);
-
-		// \x1E = ASCII Record Separator character
-		// \x1D = ASCII Group Separator character
-		const shaMatch = `\x1E${sha}\x1D`;
-		// eslint-disable-next-line no-control-regex
-		const shaMatchRegex = /\x1E.+?\x1D/g;
-		let found = false;
-		let count = 0;
-
-		return new Promise<[data: string[], count: number]>((resolve, reject) => {
-			const errData: string[] = [];
-			const data: string[] = [];
-
-			function onErrData(s: string) {
-				errData.push(s);
-			}
-
-			function onError(e: Error) {
-				reject(e);
-			}
-
-			function onExit(exitCode: number) {
-				if (exitCode !== 0) {
-					reject(new Error(errData.join('')));
-				}
-
-				resolve([data, count]);
-			}
-
-			function onData(s: string) {
-				data.push(s);
-
-				const matches = s.match(shaMatchRegex);
-				count += matches?.length ?? 0;
-
-				if (!found && matches?.includes(shaMatch)) {
-					found = true;
-					// Buffer a bit past the sha we are looking for
-					if (count > limit) {
-						limit = count + 50;
-					}
-				}
-
-				if (!found || count <= limit) return;
-
-				proc.removeListener('exit', onExit);
-				proc.removeListener('error', onError);
-				proc.stdout!.removeListener('data', onData);
-				proc.stderr!.removeListener('data', onErrData);
-				proc.kill();
-
-				resolve([data, count]);
-			}
-
-			proc.on('error', onError);
-			proc.on('exit', onExit);
-
-			proc.stdout!.setEncoding('utf8');
-			proc.stdout!.on('data', onData);
-
-			proc.stderr!.setEncoding('utf8');
-			proc.stderr!.on('data', onErrData);
-		});
-	}
-
 	async ls_files(
 		repoPath: string,
 		fileName: string,