push.rocks/smartwatch
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(watcher.node): Add automatic restart, periodic health checks, and safe event emission to Node watcher; improve logging and stat handling

Browse Source
This commit is contained in:
Jürgen Kunz
2025-12-08 17:48:50 +00:00
parent 1c77a94b96
commit 2ce056f401
4 changed files with 204 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

180
ts/watchers/watcher.node.ts
View File

@@ -14,10 +14,114 @@ export class NodeWatcher implements IWatcher {
// Debounce: pending emits per file path
private pendingEmits: Map<string, NodeJS.Timeout> = new Map();
// Restart tracking
private restartDelays: Map<string, number> = new Map();
private restartAttempts: Map<string, number> = new Map();
private healthCheckInterval: NodeJS.Timeout | null = null;
// Configuration constants
private static readonly MAX_RETRIES = 3;
private static readonly INITIAL_RESTART_DELAY = 1000;
private static readonly MAX_RESTART_DELAY = 30000;
private static readonly HEALTH_CHECK_INTERVAL = 30000;
public readonly events$ = new smartrx.rxjs.Subject<IWatchEvent>();
constructor(private options: IWatcherOptions) {}
/**
* Safely emit an event, catching any subscriber errors
*/
private safeEmit(event: IWatchEvent): void {
try {
this.events$.next(event);
} catch (error) {
console.error('[smartwatch] Subscriber threw error (isolated):', error);
// Don't let subscriber errors kill the watcher
}
}
/**
* Restart a watcher after an error with exponential backoff
*/
private async restartWatcher(basePath: string, error: Error): Promise<void> {
const attempts = (this.restartAttempts.get(basePath) || 0) + 1;
this.restartAttempts.set(basePath, attempts);
console.log(`[smartwatch] Watcher error for ${basePath}: ${error.message}`);
console.log(`[smartwatch] Restart attempt ${attempts}/${NodeWatcher.MAX_RETRIES}`);
if (attempts > NodeWatcher.MAX_RETRIES) {
console.error(`[smartwatch] Max retries exceeded for ${basePath}, giving up`);
this.safeEmit({
type: 'error',
path: basePath,
error: new Error(`Max restart retries (${NodeWatcher.MAX_RETRIES}) exceeded`)
});
return;
}
// Close failed watcher
const oldWatcher = this.watchers.get(basePath);
if (oldWatcher) {
try {
oldWatcher.close();
} catch {
// Ignore close errors
}
this.watchers.delete(basePath);
}
// Exponential backoff
const delay = this.restartDelays.get(basePath) || NodeWatcher.INITIAL_RESTART_DELAY;
console.log(`[smartwatch] Waiting ${delay}ms before restart...`);
await new Promise((resolve) => setTimeout(resolve, delay));
this.restartDelays.set(basePath, Math.min(delay * 2, NodeWatcher.MAX_RESTART_DELAY));
try {
await this.watchPath(basePath, 0);
console.log(`[smartwatch] Successfully restarted watcher for ${basePath}`);
this.restartDelays.set(basePath, NodeWatcher.INITIAL_RESTART_DELAY);
this.restartAttempts.set(basePath, 0);
} catch (restartError) {
console.error(`[smartwatch] Restart failed for ${basePath}:`, restartError);
this.restartWatcher(basePath, restartError as Error); // Recursive retry
}
}
/**
* Start periodic health checks to detect silent failures
*/
private startHealthCheck(): void {
console.log('[smartwatch] Starting health check (every 30s)');
this.healthCheckInterval = setInterval(async () => {
console.log(`[smartwatch] Health check: ${this.watchers.size} watchers active`);
for (const [basePath] of this.watchers) {
const stats = await this.statSafe(basePath);
if (!stats && this._isWatching) {
console.error(`[smartwatch] Health check failed: ${basePath} no longer exists`);
this.safeEmit({
type: 'error',
path: basePath,
error: new Error('Watched path no longer exists')
});
this.restartWatcher(basePath, new Error('Watched path disappeared'));
}
}
}, NodeWatcher.HEALTH_CHECK_INTERVAL);
}
/**
* Stop health check interval
*/
private stopHealthCheck(): void {
if (this.healthCheckInterval) {
clearInterval(this.healthCheckInterval);
this.healthCheckInterval = null;
console.log('[smartwatch] Stopped health check');
}
}
/**
* Check if a file is a temporary file created by editors
*/
@@ -40,6 +144,8 @@ export class NodeWatcher implements IWatcher {
return;
}
console.log(`[smartwatch] Starting watcher for ${this.options.basePaths.length} base path(s)...`);
try {
// Start watching each base path
for (const basePath of this.options.basePaths) {
@@ -48,22 +154,31 @@ export class NodeWatcher implements IWatcher {
this._isWatching = true;
// Start health check monitoring
this.startHealthCheck();
// Perform initial scan to emit 'add' events for existing files
for (const basePath of this.options.basePaths) {
await this.scanDirectory(basePath, 0);
}
// Emit ready event
this.events$.next({ type: 'ready', path: '' });
this.safeEmit({ type: 'ready', path: '' });
console.log(`[smartwatch] Watcher started with ${this.watchers.size} active watcher(s)`);
} catch (error: any) {
this.events$.next({ type: 'error', path: '', error });
console.error('[smartwatch] Failed to start watcher:', error);
this.safeEmit({ type: 'error', path: '', error });
throw error;
}
}
async stop(): Promise<void> {
console.log('[smartwatch] Stopping watcher...');
this._isWatching = false;
// Stop health check monitoring
this.stopHealthCheck();
// Cancel all pending debounced emits
for (const timeout of this.pendingEmits.values()) {
clearTimeout(timeout);
@@ -71,11 +186,18 @@ export class NodeWatcher implements IWatcher {
this.pendingEmits.clear();
// Close all watchers
for (const [, watcher] of this.watchers) {
for (const [watchPath, watcher] of this.watchers) {
console.log(`[smartwatch] Closing watcher for: ${watchPath}`);
watcher.close();
}
this.watchers.clear();
this.watchedFiles.clear();
// Clear restart tracking state
this.restartDelays.clear();
this.restartAttempts.clear();
console.log('[smartwatch] Watcher stopped');
}
/**
@@ -105,13 +227,27 @@ export class NodeWatcher implements IWatcher {
);
watcher.on('error', (error) => {
this.events$.next({ type: 'error', path: watchPath, error });
console.error(`[smartwatch] FSWatcher error event on ${watchPath}:`, error);
this.safeEmit({ type: 'error', path: watchPath, error });
if (this._isWatching) {
this.restartWatcher(watchPath, error);
}
});
// Handle 'close' event - fs.watch can close without error
watcher.on('close', () => {
console.warn(`[smartwatch] FSWatcher closed unexpectedly for ${watchPath}`);
if (this._isWatching) {
this.restartWatcher(watchPath, new Error('Watcher closed unexpectedly'));
}
});
this.watchers.set(watchPath, watcher);
console.log(`[smartwatch] Started watching: ${watchPath}`);
}
} catch (error: any) {
this.events$.next({ type: 'error', path: watchPath, error });
console.error(`[smartwatch] Failed to watch path ${watchPath}:`, error);
this.safeEmit({ type: 'error', path: watchPath, error });
}
}
@@ -162,12 +298,12 @@ export class NodeWatcher implements IWatcher {
if (stats.isDirectory()) {
if (!this.watchedFiles.has(fullPath)) {
this.watchedFiles.add(fullPath);
this.events$.next({ type: 'addDir', path: fullPath, stats });
this.safeEmit({ type: 'addDir', path: fullPath, stats });
}
} else {
const wasWatched = this.watchedFiles.has(fullPath);
this.watchedFiles.add(fullPath);
this.events$.next({
this.safeEmit({
type: wasWatched ? 'change' : 'add',
path: fullPath,
stats
@@ -178,7 +314,7 @@ export class NodeWatcher implements IWatcher {
if (this.watchedFiles.has(fullPath)) {
const wasDir = this.isKnownDirectory(fullPath);
this.watchedFiles.delete(fullPath);
this.events$.next({
this.safeEmit({
type: wasDir ? 'unlinkDir' : 'unlink',
path: fullPath
});
@@ -191,18 +327,18 @@ export class NodeWatcher implements IWatcher {
if (!wasWatched) {
// This is actually an 'add' - file wasn't being watched before
this.watchedFiles.add(fullPath);
this.events$.next({ type: 'add', path: fullPath, stats });
this.safeEmit({ type: 'add', path: fullPath, stats });
} else {
this.events$.next({ type: 'change', path: fullPath, stats });
this.safeEmit({ type: 'change', path: fullPath, stats });
}
} else if (!stats && this.watchedFiles.has(fullPath)) {
// File was deleted
this.watchedFiles.delete(fullPath);
this.events$.next({ type: 'unlink', path: fullPath });
this.safeEmit({ type: 'unlink', path: fullPath });
}
}
} catch (error: any) {
this.events$.next({ type: 'error', path: fullPath, error });
this.safeEmit({ type: 'error', path: fullPath, error });
}
}
@@ -233,16 +369,16 @@ export class NodeWatcher implements IWatcher {
if (entry.isDirectory()) {
this.watchedFiles.add(fullPath);
this.events$.next({ type: 'addDir', path: fullPath, stats });
this.safeEmit({ type: 'addDir', path: fullPath, stats });
await this.scanDirectory(fullPath, depth + 1);
} else if (entry.isFile()) {
this.watchedFiles.add(fullPath);
this.events$.next({ type: 'add', path: fullPath, stats });
this.safeEmit({ type: 'add', path: fullPath, stats });
}
}
} catch (error: any) {
if (error.code !== 'ENOENT' && error.code !== 'EACCES') {
this.events$.next({ type: 'error', path: dirPath, error });
this.safeEmit({ type: 'error', path: dirPath, error });
}
}
}
@@ -252,12 +388,16 @@ export class NodeWatcher implements IWatcher {
*/
private async statSafe(filePath: string): Promise<fs.Stats | null> {
try {
if (this.options.followSymlinks) {
return await fs.promises.stat(filePath);
} else {
return await fs.promises.lstat(filePath);
return await (this.options.followSymlinks
? fs.promises.stat(filePath)
: fs.promises.lstat(filePath));
} catch (error: any) {
// Only silently return null for expected "file doesn't exist" errors
if (error.code === 'ENOENT' || error.code === 'ENOTDIR') {
return null;
}
} catch {
// Log other errors (permission, I/O) but still return null
console.warn(`[smartwatch] statSafe warning for ${filePath}: ${error.code} - ${error.message}`);
return null;
}
}