watcher - allow non-existent paths when correlating

src/vs/platform/files/common/files.ts

@@ -243,14 +243,6 @@ export interface IFileService {
 	 */
 	createWatcher(resource: URI, options: IWatchOptionsWithoutCorrelation): IFileSystemWatcher;
 
-	/**
-	 * Allows to start a watcher that reports file/folder change events on the provided resource.
-	 *
-	 * The watcher runs correlated and thus, file events will be reported on the returned
-	 * `IFileSystemWatcher` and not on the generic `IFileService.onDidFilesChange` event.
-	 */
-	watch(resource: URI, options: IWatchOptionsWithCorrelation): IFileSystemWatcher;
-
 	/**
 	 * Allows to start a watcher that reports file/folder change events on the provided resource.
 	 *

src/vs/platform/files/common/watcher.ts

@@ -71,7 +71,7 @@ export function isRecursiveWatchRequest(request: IWatchRequest): request is IRec
 
 export type IUniversalWatchRequest = IRecursiveWatchRequest | INonRecursiveWatchRequest;
 
-interface IWatcher {
+export interface IWatcher {
 
 	/**
 	 * A normalized file change event from the raw events

src/vs/platform/files/node/watcher/baseWatcher.ts

@@ -0,0 +1,173 @@
+/*---------------------------------------------------------------------------------------------
+ *  Copyright (c) Microsoft Corporation. All rights reserved.
+ *  Licensed under the MIT License. See License.txt in the project root for license information.
+ *--------------------------------------------------------------------------------------------*/
+
+import { watchFile, unwatchFile, Stats } from 'fs';
+import { Disposable, DisposableMap, IDisposable, toDisposable } from 'vs/base/common/lifecycle';
+import { ILogMessage, IUniversalWatchRequest, IWatcher } from 'vs/platform/files/common/watcher';
+import { Emitter, Event } from 'vs/base/common/event';
+import { FileChangeType, IFileChange } from 'vs/platform/files/common/files';
+import { URI } from 'vs/base/common/uri';
+
+export abstract class BaseWatcher extends Disposable implements IWatcher {
+
+	protected readonly _onDidChangeFile = this._register(new Emitter<IFileChange[]>());
+	readonly onDidChangeFile = this._onDidChangeFile.event;
+
+	protected readonly _onDidLogMessage = this._register(new Emitter<ILogMessage>());
+	readonly onDidLogMessage = this._onDidLogMessage.event;
+
+	private mapWatchMissingRequestPathToCorrelationId = this._register(new DisposableMap<number>());
+
+	private allWatchRequests = new Set<IUniversalWatchRequest>();
+	private suspendedWatchRequests = new Set<IUniversalWatchRequest>();
+
+	protected readonly missingRequestPathPollingInterval: number | undefined;
+
+	async watch(requests: IUniversalWatchRequest[]): Promise<void> {
+		this.allWatchRequests = new Set([...requests]);
+
+		const correlationIds = new Set<number>();
+		for (const request of requests) {
+
+			// Request with correlation: watch request path to support
+			// watching paths that do not exist yet or are potentially
+			// being deleted and recreated.
+			//
+			// We are not doing this for all watch requests yet to see
+			// how it goes, thus its limitd to correlated requests.
+
+			if (typeof request.correlationId === 'number') {
+				correlationIds.add(request.correlationId);
+
+				if (!this.mapWatchMissingRequestPathToCorrelationId.has(request.correlationId)) {
+					this.mapWatchMissingRequestPathToCorrelationId.set(request.correlationId, this.watchMissingRequestPath(request));
+				}
+			}
+		}
+
+		// Remove all watched correlated paths that are no longer
+		// needed because the request is no longer there
+		for (const [correlationId] of this.mapWatchMissingRequestPathToCorrelationId) {
+			if (!correlationIds.has(correlationId)) {
+				this.mapWatchMissingRequestPathToCorrelationId.deleteAndDispose(correlationId);
+			}
+		}
+
+		// Remove all suspended requests that are no longer needed
+		for (const request of this.suspendedWatchRequests) {
+			if (!this.allWatchRequests.has(request)) {
+				this.suspendedWatchRequests.delete(request);
+			}
+		}
+
+		return this.updateWatchers();
+	}
+
+	private updateWatchers(): Promise<void> {
+		return this.doWatch(Array.from(this.allWatchRequests).filter(request => !this.suspendedWatchRequests.has(request)));
+	}
+
+	private watchMissingRequestPath(request: IUniversalWatchRequest): IDisposable {
+		if (typeof request.correlationId !== 'number') {
+			return Disposable.None; // for now limit this to correlated watch requests only (reduces surface)
+		}
+
+		const that = this;
+		const resource = URI.file(request.path);
+
+		let disposed = false;
+		let pathNotFound = false;
+
+		const watchFileCallback: (curr: Stats, prev: Stats) => void = (curr, prev) => {
+			if (disposed) {
+				return; // return early if already disposed
+			}
+
+			const currentPathNotFound = this.isPathNotFound(curr);
+			const previousPathNotFound = this.isPathNotFound(prev);
+			const oldPathNotFound = pathNotFound;
+			pathNotFound = currentPathNotFound;
+
+			// Watch path created: resume watching request
+			if (
+				(previousPathNotFound && !currentPathNotFound) || 					// file was created
+				(oldPathNotFound && !currentPathNotFound && !previousPathNotFound) 	// file was created from a rename
+			) {
+				this.trace(`fs.watchFile() detected ${request.path} exists again, resuming watcher (correlationId: ${request.correlationId})`);
+
+				// Emit as event
+				const event: IFileChange = { resource, type: FileChangeType.ADDED, cId: request.correlationId };
+				that._onDidChangeFile.fire([event]);
+				this.traceEvent(event, request);
+
+				this.suspendedWatchRequests.delete(request);
+				this.updateWatchers();
+			}
+
+			// Watch path deleted or never existed: suspend watching request
+			else if (currentPathNotFound) {
+				this.trace(`fs.watchFile() detected ${request.path} not found, suspending watcher (correlationId: ${request.correlationId})`);
+
+				if (!previousPathNotFound) {
+					const event: IFileChange = { resource, type: FileChangeType.DELETED, cId: request.correlationId };
+					that._onDidChangeFile.fire([event]);
+					this.traceEvent(event, request);
+				}
+
+				this.suspendedWatchRequests.add(request);
+				this.updateWatchers();
+			}
+		};
+
+		this.trace(`starting fs.watchFile() on ${request.path} (correlationId: ${request.correlationId})`);
+		try {
+			watchFile(request.path, { persistent: false, interval: this.missingRequestPathPollingInterval }, watchFileCallback);
+		} catch (error) {
+			this.warn(`fs.watchFile() failed with error ${error} on path ${request.path} (correlationId: ${request.correlationId})`);
+
+			return Disposable.None;
+		}
+
+		return toDisposable(() => {
+			this.trace(`stopping fs.watchFile() on ${request.path} (correlationId: ${request.correlationId})`);
+
+			disposed = true;
+
+			this.suspendedWatchRequests.delete(request);
+
+			try {
+				unwatchFile(request.path, watchFileCallback);
+			} catch (error) {
+				this.warn(`fs.unwatchFile() failed with error ${error} on path ${request.path} (correlationId: ${request.correlationId})`);
+			}
+		});
+	}
+
+	private isPathNotFound(stats: Stats): boolean {
+		return stats.ctimeMs === 0 && stats.ino === 0;
+	}
+
+	async stop(): Promise<void> {
+		this.mapWatchMissingRequestPathToCorrelationId.clearAndDisposeAll();
+		this.suspendedWatchRequests.clear();
+	}
+
+	protected shouldRestartWatching(request: IUniversalWatchRequest): boolean {
+		return typeof request.correlationId !== 'number';
+	}
+
+	protected traceEvent(event: IFileChange, request: IUniversalWatchRequest): void {
+		const traceMsg = ` >> normalized ${event.type === FileChangeType.ADDED ? '[ADDED]' : event.type === FileChangeType.DELETED ? '[DELETED]' : '[CHANGED]'} ${event.resource.fsPath}`;
+		this.trace(typeof request.correlationId === 'number' ? `${traceMsg} (correlationId: ${request.correlationId})` : traceMsg);
+	}
+
+	protected abstract doWatch(requests: IUniversalWatchRequest[]): Promise<void>;
+
+	protected abstract trace(message: string): void;
+	protected abstract warn(message: string): void;
+
+	abstract onDidError: Event<string>;
+	abstract setVerboseLogging(enabled: boolean): Promise<void>;
+}

src/vs/platform/files/node/watcher/nodejs/nodejsClient.ts

@@ -21,6 +21,6 @@ export class NodeJSWatcherClient extends AbstractNonRecursiveWatcherClient {
 	}
 
 	protected override createWatcher(disposables: DisposableStore): INonRecursiveWatcher {
-		return disposables.add(new NodeJSWatcher());
+		return disposables.add(new NodeJSWatcher()) satisfies INonRecursiveWatcher;
 	}
 }

src/vs/platform/files/node/watcher/nodejs/nodejsWatcher.ts

@@ -3,12 +3,11 @@
  *  Licensed under the MIT License. See License.txt in the project root for license information.
  *--------------------------------------------------------------------------------------------*/
 
-import { Event, Emitter } from 'vs/base/common/event';
+import { Event } from 'vs/base/common/event';
 import { patternsEquals } from 'vs/base/common/glob';
-import { Disposable } from 'vs/base/common/lifecycle';
+import { BaseWatcher } from 'vs/platform/files/node/watcher/baseWatcher';
 import { isLinux } from 'vs/base/common/platform';
-import { IFileChange } from 'vs/platform/files/common/files';
-import { ILogMessage, INonRecursiveWatchRequest, INonRecursiveWatcher } from 'vs/platform/files/common/watcher';
+import { INonRecursiveWatchRequest, INonRecursiveWatcher } from 'vs/platform/files/common/watcher';
 import { NodeJSFileWatcherLibrary } from 'vs/platform/files/node/watcher/nodejs/nodejsWatcherLib';
 
 export interface INodeJSWatcherInstance {
@@ -24,21 +23,15 @@ export interface INodeJSWatcherInstance {
 	readonly request: INonRecursiveWatchRequest;
 }
 
-export class NodeJSWatcher extends Disposable implements INonRecursiveWatcher {
-
-	private readonly _onDidChangeFile = this._register(new Emitter<IFileChange[]>());
-	readonly onDidChangeFile = this._onDidChangeFile.event;
-
-	private readonly _onDidLogMessage = this._register(new Emitter<ILogMessage>());
-	readonly onDidLogMessage = this._onDidLogMessage.event;
+export class NodeJSWatcher extends BaseWatcher implements INonRecursiveWatcher {
 
 	readonly onDidError = Event.None;
 
 	protected readonly watchers = new Map<string, INodeJSWatcherInstance>();
 
 	private verboseLogging = false;
 
-	async watch(requests: INonRecursiveWatchRequest[]): Promise<void> {
+	protected override async doWatch(requests: INonRecursiveWatchRequest[]): Promise<void> {
 
 		// Figure out duplicates to remove from the requests
 		const normalizedRequests = this.normalizeRequests(requests);
@@ -90,7 +83,9 @@ export class NodeJSWatcher extends Disposable implements INonRecursiveWatcher {
 		this.watchers.set(request.path, watcher);
 	}
 
-	async stop(): Promise<void> {
+	override async stop(): Promise<void> {
+		await super.stop();
+
 		for (const [path] of this.watchers) {
 			this.stopWatching(path);
 		}
@@ -134,13 +129,19 @@ export class NodeJSWatcher extends Disposable implements INonRecursiveWatcher {
 		}
 	}
 
-	private trace(message: string): void {
+	protected trace(message: string): void {
 		if (this.verboseLogging) {
 			this._onDidLogMessage.fire({ type: 'trace', message: this.toMessage(message) });
 		}
 	}
 
-	private toMessage(message: string, watcher?: INodeJSWatcherInstance): string {
-		return watcher ? `[File Watcher (node.js)] ${message} (path: ${watcher.request.path})` : `[File Watcher (node.js)] ${message}`;
+	protected warn(message: string): void {
+		if (this.verboseLogging) {
+			this._onDidLogMessage.fire({ type: 'warn', message: this.toMessage(message) });
+		}
+	}
+
+	private toMessage(message: string): string {
+		return `[File Watcher (node.js)] ${message}`;
 	}
 }

src/vs/platform/files/node/watcher/parcel/parcelWatcher.ts

@@ -13,15 +13,15 @@ import { toErrorMessage } from 'vs/base/common/errorMessage';
 import { Emitter } from 'vs/base/common/event';
 import { randomPath } from 'vs/base/common/extpath';
 import { GLOBSTAR, ParsedPattern, patternsEquals } from 'vs/base/common/glob';
-import { Disposable } from 'vs/base/common/lifecycle';
+import { BaseWatcher } from 'vs/platform/files/node/watcher/baseWatcher';
 import { TernarySearchTree } from 'vs/base/common/ternarySearchTree';
 import { normalizeNFC } from 'vs/base/common/normalization';
 import { dirname, normalize } from 'vs/base/common/path';
 import { isLinux, isMacintosh, isWindows } from 'vs/base/common/platform';
 import { realcaseSync, realpathSync } from 'vs/base/node/extpath';
 import { NodeJSFileWatcherLibrary } from 'vs/platform/files/node/watcher/nodejs/nodejsWatcherLib';
 import { FileChangeType, IFileChange } from 'vs/platform/files/common/files';
-import { ILogMessage, coalesceEvents, IRecursiveWatchRequest, IRecursiveWatcher, parseWatcherPatterns } from 'vs/platform/files/common/watcher';
+import { coalesceEvents, IRecursiveWatchRequest, IRecursiveWatcher, parseWatcherPatterns } from 'vs/platform/files/common/watcher';
 
 export interface IParcelWatcherInstance {
 
@@ -58,7 +58,7 @@ export interface IParcelWatcherInstance {
 	stop(): Promise<void>;
 }
 
-export class ParcelWatcher extends Disposable implements IRecursiveWatcher {
+export class ParcelWatcher extends BaseWatcher implements IRecursiveWatcher {
 
 	private static readonly MAP_PARCEL_WATCHER_ACTION_TO_FILE_CHANGE = new Map<parcelWatcher.EventType, number>(
 		[
@@ -70,12 +70,6 @@ export class ParcelWatcher extends Disposable implements IRecursiveWatcher {
 
 	private static readonly PARCEL_WATCHER_BACKEND = isWindows ? 'windows' : isLinux ? 'inotify' : 'fs-events';
 
-	private readonly _onDidChangeFile = this._register(new Emitter<IFileChange[]>());
-	readonly onDidChangeFile = this._onDidChangeFile.event;
-
-	private readonly _onDidLogMessage = this._register(new Emitter<ILogMessage>());
-	readonly onDidLogMessage = this._onDidLogMessage.event;
-
 	private readonly _onDidError = this._register(new Emitter<string>());
 	readonly onDidError = this._onDidError.event;
 
@@ -120,7 +114,7 @@ export class ParcelWatcher extends Disposable implements IRecursiveWatcher {
 		process.on('unhandledRejection', error => this.onUnexpectedError(error));
 	}
 
-	async watch(requests: IRecursiveWatchRequest[]): Promise<void> {
+	protected override async doWatch(requests: IRecursiveWatchRequest[]): Promise<void> {
 
 		// Figure out duplicates to remove from the requests
 		const normalizedRequests = this.normalizeRequests(requests);
@@ -370,8 +364,7 @@ export class ParcelWatcher extends Disposable implements IRecursiveWatcher {
 		// Logging
 		if (this.verboseLogging) {
 			for (const event of events) {
-				const traceMsg = ` >> normalized ${event.type === FileChangeType.ADDED ? '[ADDED]' : event.type === FileChangeType.DELETED ? '[DELETED]' : '[CHANGED]'} ${event.resource.fsPath}`;
-				this.trace(typeof watcher.request.correlationId === 'number' ? `${traceMsg} (correlationId: ${watcher.request.correlationId})` : traceMsg);
+				this.traceEvent(event, watcher.request);
 			}
 		}
 
@@ -383,7 +376,7 @@ export class ParcelWatcher extends Disposable implements IRecursiveWatcher {
 			this.warn(`started ignoring events due to too many file change events at once (incoming: ${events.length}, most recent change: ${events[0].resource.fsPath}). Use 'files.watcherExclude' setting to exclude folders with lots of changing files (e.g. compilation output).`);
 		} else {
 			if (this.throttledFileChangesEmitter.pending > 0) {
-				this.trace(`started throttling events due to large amount of file change events at once (pending: ${this.throttledFileChangesEmitter.pending}, most recent change: ${events[0].resource.fsPath}). Use 'files.watcherExclude' setting to exclude folders with lots of changing files (e.g. compilation output).`);
+				this.trace(`started throttling events due to large amount of file change events at once (pending: ${this.throttledFileChangesEmitter.pending}, most recent change: ${events[0].resource.fsPath}). Use 'files.watcherExclude' setting to exclude folders with lots of changing files (e.g. compilation output).`, watcher);
 			}
 		}
 	}
@@ -466,8 +459,14 @@ export class ParcelWatcher extends Disposable implements IRecursiveWatcher {
 	private onWatchedPathDeleted(watcher: IParcelWatcherInstance): void {
 		this.warn('Watcher shutdown because watched path got deleted', watcher);
 
+		if (!this.shouldRestartWatching(watcher.request)) {
+			return; // return if this deletion is handled outside
+		}
+
 		const parentPath = dirname(watcher.request.path);
 		if (existsSync(parentPath)) {
+			this.trace('Trying to watch on the parent path to restart the watcher...', watcher);
+
 			const nodeWatcher = new NodeJSFileWatcherLibrary({ path: parentPath, excludes: [], recursive: false, correlationId: watcher.request.correlationId }, changes => {
 				if (watcher.token.isCancellationRequested) {
 					return; // return early when disposed
@@ -522,7 +521,9 @@ export class ParcelWatcher extends Disposable implements IRecursiveWatcher {
 		}
 	}
 
-	async stop(): Promise<void> {
+	override async stop(): Promise<void> {
+		await super.stop();
+
 		for (const [path] of this.watchers) {
 			await this.stopWatching(path);
 		}
@@ -559,7 +560,7 @@ export class ParcelWatcher extends Disposable implements IRecursiveWatcher {
 	private async stopWatching(path: string): Promise<void> {
 		const watcher = this.watchers.get(path);
 		if (watcher) {
-			this.trace(`stopping file watcher on ${watcher.request.path}`);
+			this.trace(`stopping file watcher`, watcher);
 
 			this.watchers.delete(path);
 
@@ -664,13 +665,13 @@ export class ParcelWatcher extends Disposable implements IRecursiveWatcher {
 		this.verboseLogging = enabled;
 	}
 
-	private trace(message: string) {
+	protected trace(message: string, watcher?: IParcelWatcherInstance): void {
 		if (this.verboseLogging) {
-			this._onDidLogMessage.fire({ type: 'trace', message: this.toMessage(message) });
+			this._onDidLogMessage.fire({ type: 'trace', message: this.toMessage(message, watcher) });
 		}
 	}
 
-	private warn(message: string, watcher?: IParcelWatcherInstance) {
+	protected warn(message: string, watcher?: IParcelWatcherInstance) {
 		this._onDidLogMessage.fire({ type: 'warn', message: this.toMessage(message, watcher) });
 	}