diff --git a/src/ipc.test.ts b/src/ipc.test.ts new file mode 100644 index 0000000..6531f9a --- /dev/null +++ b/src/ipc.test.ts @@ -0,0 +1,73 @@ +import fs from 'fs'; +import os from 'os'; +import path from 'path'; + +import { afterEach, describe, expect, it } from 'vitest'; + +import { claimIpcFile, quarantineClaimedIpcFiles } from './ipc.js'; + +const tempDirs: string[] = []; + +function makeTempDir(): string { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ejclaw-ipc-')); + tempDirs.push(dir); + return dir; +} + +afterEach(() => { + for (const dir of tempDirs.splice(0, tempDirs.length)) { + fs.rmSync(dir, { recursive: true, force: true }); + } +}); + +describe('ipc file claiming', () => { + it('claims IPC files into a hidden processing directory before handling', () => { + const baseDir = makeTempDir(); + const messagesDir = path.join(baseDir, 'messages'); + fs.mkdirSync(messagesDir, { recursive: true }); + + const filePath = path.join(messagesDir, 'message.json'); + fs.writeFileSync(filePath, '{"type":"message"}'); + + const claimedPath = claimIpcFile(filePath); + + expect(claimedPath).toBe( + path.join(messagesDir, '.processing', 'message.json'), + ); + expect(fs.existsSync(filePath)).toBe(false); + expect(fs.existsSync(claimedPath!)).toBe(true); + }); + + it('returns null when the IPC file was already claimed or deleted', () => { + const baseDir = makeTempDir(); + const missingPath = path.join(baseDir, 'messages', 'missing.json'); + + expect(claimIpcFile(missingPath)).toBeNull(); + }); +}); + +describe('ipc claimed-file quarantine', () => { + it('moves stranded claimed files into the error directory without reprocessing', () => { + const baseDir = makeTempDir(); + const messagesDir = path.join(baseDir, 'messages'); + const processingDir = path.join(messagesDir, '.processing'); + const errorDir = path.join(baseDir, 'errors'); + + fs.mkdirSync(processingDir, { recursive: true }); + fs.mkdirSync(errorDir, { recursive: true }); + + const claimedPath = path.join(processingDir, 'message.json'); + fs.writeFileSync(claimedPath, '{"type":"message"}'); + + const movedPaths = quarantineClaimedIpcFiles( + messagesDir, + errorDir, + 'group-message-stale', + ); + + expect(movedPaths).toHaveLength(1); + expect(fs.existsSync(claimedPath)).toBe(false); + expect(fs.existsSync(movedPaths[0])).toBe(true); + expect(path.dirname(movedPaths[0])).toBe(errorDir); + }); +}); diff --git a/src/ipc.ts b/src/ipc.ts index 4f4ba41..3bb6954 100644 --- a/src/ipc.ts +++ b/src/ipc.ts @@ -35,6 +35,68 @@ export interface IpcDeps { } let ipcWatcherRunning = false; +const IPC_PROCESSING_DIRNAME = '.processing'; + +function buildIpcErrorPath( + errorDir: string, + prefix: string, + fileName: string, +): string { + return path.join(errorDir, `${prefix}-${Date.now()}-${fileName}`); +} + +export function claimIpcFile(filePath: string): string | null { + const processingDir = path.join( + path.dirname(filePath), + IPC_PROCESSING_DIRNAME, + ); + fs.mkdirSync(processingDir, { recursive: true }); + + const claimedPath = path.join(processingDir, path.basename(filePath)); + try { + fs.renameSync(filePath, claimedPath); + return claimedPath; + } catch (err) { + if ((err as NodeJS.ErrnoException).code === 'ENOENT') { + return null; + } + throw err; + } +} + +export function quarantineClaimedIpcFiles( + ipcDir: string, + errorDir: string, + prefix: string, +): string[] { + const processingDir = path.join(ipcDir, IPC_PROCESSING_DIRNAME); + if (!fs.existsSync(processingDir)) { + return []; + } + + const movedPaths: string[] = []; + for (const file of fs.readdirSync(processingDir).filter((f) => + f.endsWith('.json'), + )) { + const claimedPath = path.join(processingDir, file); + const errorPath = buildIpcErrorPath(errorDir, prefix, file); + fs.renameSync(claimedPath, errorPath); + movedPaths.push(errorPath); + } + + return movedPaths; +} + +function moveClaimedIpcFileToError( + claimedPath: string, + errorDir: string, + prefix: string, +): void { + fs.renameSync( + claimedPath, + buildIpcErrorPath(errorDir, prefix, path.basename(claimedPath)), + ); +} export function startIpcWatcher(deps: IpcDeps): void { if (ipcWatcherRunning) { @@ -72,6 +134,30 @@ export function startIpcWatcher(deps: IpcDeps): void { const isMain = folderIsMain.get(sourceGroup) === true; const messagesDir = path.join(ipcBaseDir, sourceGroup, 'messages'); const tasksDir = path.join(ipcBaseDir, sourceGroup, 'tasks'); + const errorDir = path.join(ipcBaseDir, 'errors'); + fs.mkdirSync(errorDir, { recursive: true }); + + for (const quarantinedPath of quarantineClaimedIpcFiles( + messagesDir, + errorDir, + `${sourceGroup}-message-stale`, + )) { + logger.warn( + { sourceGroup, quarantinedPath }, + 'Quarantined previously claimed IPC message after restart', + ); + } + + for (const quarantinedPath of quarantineClaimedIpcFiles( + tasksDir, + errorDir, + `${sourceGroup}-task-stale`, + )) { + logger.warn( + { sourceGroup, quarantinedPath }, + 'Quarantined previously claimed IPC task after restart', + ); + } // Process messages from this group's IPC directory try { @@ -81,8 +167,10 @@ export function startIpcWatcher(deps: IpcDeps): void { .filter((f) => f.endsWith('.json')); for (const file of messageFiles) { const filePath = path.join(messagesDir, file); + const claimedPath = claimIpcFile(filePath); + if (!claimedPath) continue; try { - const data = readJsonFile(filePath); + const data = readJsonFile(claimedPath); if (!data || typeof data !== 'object') throw new Error('Invalid JSON'); const msg = data as { @@ -122,17 +210,16 @@ export function startIpcWatcher(deps: IpcDeps): void { ); } } - fs.unlinkSync(filePath); + fs.unlinkSync(claimedPath); } catch (err) { logger.error( { file, sourceGroup, err }, 'Error processing IPC message', ); - const errorDir = path.join(ipcBaseDir, 'errors'); - fs.mkdirSync(errorDir, { recursive: true }); - fs.renameSync( - filePath, - path.join(errorDir, `${sourceGroup}-${file}`), + moveClaimedIpcFileToError( + claimedPath, + errorDir, + `${sourceGroup}-message-error`, ); } } @@ -152,8 +239,10 @@ export function startIpcWatcher(deps: IpcDeps): void { .filter((f) => f.endsWith('.json')); for (const file of taskFiles) { const filePath = path.join(tasksDir, file); + const claimedPath = claimIpcFile(filePath); + if (!claimedPath) continue; try { - const data = readJsonFile(filePath); + const data = readJsonFile(claimedPath); if (!data || typeof data !== 'object') throw new Error('Invalid JSON'); // Pass source group identity to processTaskIpc for authorization @@ -163,17 +252,16 @@ export function startIpcWatcher(deps: IpcDeps): void { isMain, deps, ); - fs.unlinkSync(filePath); + fs.unlinkSync(claimedPath); } catch (err) { logger.error( { file, sourceGroup, err }, 'Error processing IPC task', ); - const errorDir = path.join(ipcBaseDir, 'errors'); - fs.mkdirSync(errorDir, { recursive: true }); - fs.renameSync( - filePath, - path.join(errorDir, `${sourceGroup}-${file}`), + moveClaimedIpcFileToError( + claimedPath, + errorDir, + `${sourceGroup}-task-error`, ); } }