eng: reapply test leak fixes

Re-applies #190623 which I merged before everyone had a chance to review.

Alex and Ben, you were the two people whose code this touches and didn't
review it in the original PR.

Author: connor4312

src/vs/base/browser/ui/grid/gridview.ts

@@ -705,6 +705,7 @@ class BranchNode implements ISplitView<ILayoutContext>, IDisposable {
 		this.splitviewSashResetDisposable.dispose();
 		this.childrenSashResetDisposable.dispose();
 		this.childrenChangeDisposable.dispose();
+		this.onDidScrollDisposable.dispose();
 		this.splitview.dispose();
 	}
 }

src/vs/base/browser/ui/splitview/splitview.ts

@@ -565,11 +565,11 @@ export class SplitView<TLayoutContext = undefined> extends Disposable {
 		this.sashContainer = append(this.el, $('.sash-container'));
 		this.viewContainer = $('.split-view-container');
 
-		this.scrollable = new Scrollable({
+		this.scrollable = this._register(new Scrollable({
 			forceIntegerValues: true,
 			smoothScrollDuration: 125,
 			scheduleAtNextAnimationFrame
-		});
+		}));
 		this.scrollableElement = this._register(new SmoothScrollableElement(this.viewContainer, {
 			vertical: this.orientation === Orientation.VERTICAL ? (options.scrollbarVisibility ?? ScrollbarVisibility.Auto) : ScrollbarVisibility.Hidden,
 			horizontal: this.orientation === Orientation.HORIZONTAL ? (options.scrollbarVisibility ?? ScrollbarVisibility.Auto) : ScrollbarVisibility.Hidden

src/vs/base/browser/ui/toolbar/toolbar.ts

@@ -212,6 +212,7 @@ export class ToolBar extends Disposable {
 
 	override dispose(): void {
 		this.clear();
+		this.disposables.dispose();
 		super.dispose();
 	}
 }

src/vs/base/common/stream.ts

@@ -519,20 +519,22 @@ export function consumeStream<T, R = T>(stream: ReadableStreamEvents<T>, reducer
 	return new Promise((resolve, reject) => {
 		const chunks: T[] = [];
 
-		listenStream(stream, {
+		const l = listenStream(stream, {
 			onData: chunk => {
 				if (reducer) {
 					chunks.push(chunk);
 				}
 			},
 			onError: error => {
+				l.dispose();
 				if (reducer) {
 					reject(error);
 				} else {
 					resolve(undefined);
 				}
 			},
 			onEnd: () => {
+				l.dispose();
 				if (reducer) {
 					resolve(reducer(chunks));
 				} else {
@@ -570,15 +572,18 @@ export interface IStreamListener<T> {
 export function listenStream<T>(stream: ReadableStreamEvents<T>, listener: IStreamListener<T>): IDisposable {
 	let destroyed = false;
 
+	// error and end events are in the next microtask so that a stream that is
+	// closed synchronously (e.g from a memory `toStream`) can get its disposable
+	// and destroy the stream.
 	stream.on('error', error => {
 		if (!destroyed) {
-			listener.onError(error);
+			queueMicrotask(() => listener.onError(error));
 		}
 	});
 
 	stream.on('end', () => {
 		if (!destroyed) {
-			listener.onEnd();
+			queueMicrotask(() => listener.onEnd());
 		}
 	});
 
@@ -692,10 +697,16 @@ export function toReadable<T>(t: T): Readable<T> {
 export function transform<Original, Transformed>(stream: ReadableStreamEvents<Original>, transformer: ITransformer<Original, Transformed>, reducer: IReducer<Transformed>): ReadableStream<Transformed> {
 	const target = newWriteableStream<Transformed>(reducer);
 
-	listenStream(stream, {
+	const l = listenStream(stream, {
 		onData: data => target.write(transformer.data(data)),
-		onError: error => target.error(transformer.error ? transformer.error(error) : error),
-		onEnd: () => target.end()
+		onError: error => {
+			l.dispose();
+			target.error(transformer.error ? transformer.error(error) : error);
+		},
+		onEnd: () => {
+			l.dispose();
+			target.end();
+		}
 	});
 
 	return target;
@@ -740,7 +751,7 @@ export function prefixedStream<T>(prefix: T, stream: ReadableStream<T>, reducer:
 
 	const target = newWriteableStream<T>(reducer);
 
-	listenStream(stream, {
+	const l = listenStream(stream, {
 		onData: data => {
 
 			// Handle prefix only once
@@ -752,8 +763,12 @@ export function prefixedStream<T>(prefix: T, stream: ReadableStream<T>, reducer:
 
 			return target.write(data);
 		},
-		onError: error => target.error(error),
+		onError: error => {
+			l.dispose();
+			target.error(error);
+		},
 		onEnd: () => {
+			l.dispose();
 
 			// Handle prefix only once
 			if (!prefixHandled) {

src/vs/base/test/common/stream.test.ts

@@ -315,14 +315,14 @@ suite('Stream', () => {
 		assert.strictEqual(consumed, undefined);
 	});
 
-	test('listenStream', () => {
+	test('listenStream', async () => {
 		const stream = newWriteableStream<string>(strings => strings.join());
 
 		let error = false;
 		let end = false;
 		let data = '';
 
-		listenStream(stream, {
+		const l = listenStream(stream, {
 			onData: d => {
 				data = d;
 			},
@@ -345,10 +345,14 @@ suite('Stream', () => {
 		assert.strictEqual(end, false);
 
 		stream.error(new Error());
+		await new Promise<void>(r => queueMicrotask(r));
 		assert.strictEqual(error, true);
 
 		stream.end('Final Bit');
+		await new Promise<void>(r => queueMicrotask(r));
 		assert.strictEqual(end, true);
+
+		l.dispose();
 	});
 
 	test('listenStream - dispose', () => {

src/vs/editor/browser/widget/diffEditorWidget2/diffEditorEditors.ts

@@ -35,8 +35,8 @@ export class DiffEditorEditors extends Disposable {
 	) {
 		super();
 
-		this.original = this._createLeftHandSideEditor(_options.editorOptions.get(), codeEditorWidgetOptions.originalEditor || {});
-		this.modified = this._createRightHandSideEditor(_options.editorOptions.get(), codeEditorWidgetOptions.modifiedEditor || {});
+		this.original = this._register(this._createLeftHandSideEditor(_options.editorOptions.get(), codeEditorWidgetOptions.originalEditor || {}));
+		this.modified = this._register(this._createRightHandSideEditor(_options.editorOptions.get(), codeEditorWidgetOptions.modifiedEditor || {}));
 
 		this._register(autorunHandleChanges({
 			createEmptyChangeSummary: () => ({} as IDiffEditorConstructionOptions),

src/vs/editor/common/model/textModel.ts

@@ -67,19 +67,21 @@ export function createTextBufferFactoryFromStream(stream: ITextStream | VSBuffer
 
 		let done = false;
 
-		listenStream<string | VSBuffer>(stream, {
+		const l = listenStream<string | VSBuffer>(stream, {
 			onData: chunk => {
 				builder.acceptChunk((typeof chunk === 'string') ? chunk : chunk.toString());
 			},
 			onError: error => {
 				if (!done) {
 					done = true;
+					l.dispose();
 					reject(error);
 				}
 			},
 			onEnd: () => {
 				if (!done) {
 					done = true;
+					l.dispose();
 					resolve(builder.finish());
 				}
 			}

src/vs/editor/contrib/suggest/browser/suggestController.ts

@@ -143,7 +143,7 @@ export class SuggestController implements IEditorContribution {
 		// context key: update insert/replace mode
 		const ctxInsertMode = SuggestContext.InsertMode.bindTo(_contextKeyService);
 		ctxInsertMode.set(editor.getOption(EditorOption.suggest).insertMode);
-		this.model.onDidTrigger(() => ctxInsertMode.set(editor.getOption(EditorOption.suggest).insertMode));
+		this._toDispose.add(this.model.onDidTrigger(() => ctxInsertMode.set(editor.getOption(EditorOption.suggest).insertMode)));
 
 		this.widget = this._toDispose.add(new IdleValue(() => {
 

src/vs/platform/actions/browser/buttonbar.ts

@@ -55,7 +55,7 @@ export class MenuWorkbenchButtonBar extends ButtonBar {
 					'workbenchActionExecuted',
 					{ id: e.action.id, from: options.telemetrySource! }
 				);
-			}, this._store);
+			}, undefined, this._store);
 		}
 
 		const conifgProvider: IButtonConfigProvider = options?.buttonConfigProvider ?? (() => ({ showLabel: true }));

src/vs/platform/checksum/node/checksumService.ts

@@ -20,10 +20,16 @@ export class ChecksumService implements IChecksumService {
 		return new Promise<string>((resolve, reject) => {
 			const hash = createHash('md5');
 
-			listenStream(stream, {
+			const l = listenStream(stream, {
 				onData: data => hash.update(data.buffer),
-				onError: error => reject(error),
-				onEnd: () => resolve(hash.digest('base64').replace(/=+$/, ''))
+				onError: error => {
+					l.dispose();
+					reject(error);
+				},
+				onEnd: () => {
+					l.dispose();
+					resolve(hash.digest('base64').replace(/=+$/, ''));
+				}
 			});
 		});
 	}