fix(watcher.node): Normalize paths and improve Node watcher robustness: restart/rescan on errors (including ENOSPC), clear stale state, and remove legacy throttler

This commit is contained in:
2025-12-11 19:13:35 +00:00
parent 0bab7e0296
commit da77d8a608
4 changed files with 54 additions and 73 deletions

View File

@@ -18,16 +18,6 @@ const EV = {
ERROR: 'error',
} as const;
/** Throttle action types */
type ThrottleType = 'change' | 'unlink' | 'add';
/** Throttler state for an action */
interface Throttler {
timeout: NodeJS.Timeout;
count: number;
clear: () => number;
}
/** Configuration constants */
const CONFIG = {
MAX_RETRIES: 3,
@@ -127,9 +117,6 @@ export class NodeWatcher implements IWatcher {
// Event stream
public readonly events$ = new smartrx.rxjs.Subject<IWatchEvent>();
// Throttling - inspired by chokidar's _throttle pattern
private throttled: Map<ThrottleType, Map<string, Throttler>> = new Map();
// Atomic write handling - pending unlinks that may become changes
private pendingUnlinks: Map<string, { timeout: NodeJS.Timeout; event: IWatchEvent }> = new Map();
@@ -156,12 +143,7 @@ export class NodeWatcher implements IWatcher {
// Closer registry - inspired by chokidar for clean resource management
private closers: Map<string, Array<() => void>> = new Map();
constructor(private options: IWatcherOptions) {
// Initialize throttle maps
this.throttled.set('change', new Map());
this.throttled.set('unlink', new Map());
this.throttled.set('add', new Map());
}
constructor(private options: IWatcherOptions) {}
get isWatching(): boolean {
return this._isWatching;
@@ -228,14 +210,6 @@ export class NodeWatcher implements IWatcher {
}
this.pendingUnlinks.clear();
// Clear throttles
for (const actionMap of this.throttled.values()) {
for (const throttler of actionMap.values()) {
clearTimeout(throttler.timeout);
}
actionMap.clear();
}
// Now set flag
this._isWatching = false;
@@ -285,41 +259,6 @@ export class NodeWatcher implements IWatcher {
}
}
// ===========================================================================
// Throttling - Inspired by chokidar's elegant _throttle pattern
// ===========================================================================
/**
* Throttle an action for a path. Returns false if already throttled.
* Unlike simple debounce, this tracks how many events were suppressed.
*/
private throttle(actionType: ThrottleType, filePath: string, timeout: number): Throttler | false {
const actionMap = this.throttled.get(actionType);
if (!actionMap) return false;
const existing = actionMap.get(filePath);
if (existing) {
existing.count++;
return false;
}
const clear = (): number => {
const item = actionMap.get(filePath);
const count = item?.count ?? 0;
actionMap.delete(filePath);
return count;
};
const throttler: Throttler = {
timeout: setTimeout(clear, timeout),
count: 0,
clear,
};
actionMap.set(filePath, throttler);
return throttler;
}
// ===========================================================================
// Closer Registry - Clean resource management
// ===========================================================================
@@ -424,6 +363,9 @@ export class NodeWatcher implements IWatcher {
// ===========================================================================
private async watchPath(watchPath: string): Promise<void> {
// Normalize path to absolute - critical for consistent lookups
watchPath = path.resolve(watchPath);
try {
const stats = await this.statSafe(watchPath);
if (!stats?.isDirectory()) return;
@@ -494,7 +436,8 @@ export class NodeWatcher implements IWatcher {
return;
}
const fullPath = path.join(basePath, filename);
// Normalize to absolute path - critical for consistent lookups
const fullPath = path.resolve(path.join(basePath, filename));
// Handle temp files from atomic writes
if (this.isTemporaryFile(fullPath)) {
@@ -602,10 +545,7 @@ export class NodeWatcher implements IWatcher {
this.safeEmit({ type: EV.CHANGE, path: fullPath, stats });
}
} else {
// Apply throttle for change events
if (!this.throttle('change', fullPath, 50)) {
return; // Throttled
}
// Debounce already handles rapid events - no extra throttle needed
this.safeEmit({ type: EV.CHANGE, path: fullPath, stats });
}
}
@@ -667,6 +607,9 @@ export class NodeWatcher implements IWatcher {
// ===========================================================================
private async scanDirectory(dirPath: string, depth: number): Promise<void> {
// Normalize path to absolute - critical for consistent lookups
dirPath = path.resolve(dirPath);
if (depth > this.options.depth) return;
try {
@@ -724,6 +667,8 @@ export class NodeWatcher implements IWatcher {
console.error('[smartwatch] ENOSPC: inotify watch limit exceeded!');
console.error('[smartwatch] Fix: echo fs.inotify.max_user_watches=524288 | sudo tee -a /etc/sysctl.conf && sudo sysctl -p');
this.safeEmit({ type: EV.ERROR, path: basePath, error });
// Trigger restart - watcher may be broken after ENOSPC
this.restartWatcher(basePath, error);
} else {
console.error(`[smartwatch] Health check error for ${basePath}:`, error);
}
@@ -772,6 +717,22 @@ export class NodeWatcher implements IWatcher {
this.watchers.delete(basePath);
}
// Clear pending unlinks for this base path (prevent stale events)
for (const [unlinkedPath, pending] of this.pendingUnlinks) {
if (unlinkedPath.startsWith(basePath)) {
clearTimeout(pending.timeout);
this.pendingUnlinks.delete(unlinkedPath);
}
}
// Clear stale DirEntry data (will be repopulated by rescan)
for (const [dirPath, entry] of this.watched) {
if (dirPath === basePath || dirPath.startsWith(basePath + path.sep)) {
entry.dispose();
this.watched.delete(dirPath);
}
}
// Exponential backoff with abort support
const delay = this.restartDelays.get(basePath) || CONFIG.INITIAL_RESTART_DELAY;
console.log(`[smartwatch] Waiting ${delay}ms before restart...`);
@@ -803,6 +764,8 @@ export class NodeWatcher implements IWatcher {
try {
await this.watchPath(basePath);
// Rescan to catch files created during restart window
await this.scanDirectory(basePath, 0);
console.log(`[smartwatch] Successfully restarted watcher for ${basePath}`);
this.restartDelays.set(basePath, CONFIG.INITIAL_RESTART_DELAY);
this.restartAttempts.set(basePath, 0);