Merge pull request #23 from BadlyDrawnBoy/codex/implement-write-strategy-for-diskcache

Add locked atomic writes for disk OCR cache

Author: mad-sol-dev

src/utils/diskCache.ts

@@ -11,6 +11,8 @@ import type { OcrDiskCache, OcrImageResult, OcrPageResult } from '../types/cache
 import { createLogger } from './logger.js';
 
 const logger = createLogger('DiskCache');
+const LOCK_RETRY_MS = 25;
+const LOCK_TIMEOUT_MS = 5_000;
 
 /**
  * Generate cache file path from PDF path
@@ -57,34 +59,102 @@ export const loadOcrCache = (pdfPath: string): OcrDiskCache | null => {
   }
 };
 
+const sleepSync = (ms: number): void => {
+  const array = new Int32Array(new SharedArrayBuffer(4));
+  Atomics.wait(array, 0, 0, ms);
+};
+
+const acquireCacheLock = (lockPath: string): number => {
+  const start = Date.now();
+
+  // eslint-disable-next-line no-constant-condition
+  while (true) {
+    try {
+      return fs.openSync(lockPath, 'wx');
+    } catch (error: unknown) {
+      const err = error as NodeJS.ErrnoException;
+
+      if (err.code === 'EEXIST') {
+        if (Date.now() - start > LOCK_TIMEOUT_MS) {
+          throw new Error(`Timed out waiting for cache lock at ${lockPath}`);
+        }
+
+        sleepSync(LOCK_RETRY_MS);
+        continue;
+      }
+
+      throw error;
+    }
+  }
+};
+
+const releaseCacheLock = (lockPath: string, fd: number): void => {
+  fs.closeSync(fd);
+  fs.rmSync(lockPath, { force: true });
+};
+
+const writeCacheFile = (cachePath: string, cache: OcrDiskCache): void => {
+  cache.updated_at = new Date().toISOString();
+
+  // Ensure directory exists
+  const dir = path.dirname(cachePath);
+  if (!fs.existsSync(dir)) {
+    fs.mkdirSync(dir, { recursive: true });
+  }
+
+  const tempPath = `${cachePath}.${process.pid}.${Date.now()}.tmp`;
+  fs.writeFileSync(tempPath, JSON.stringify(cache, null, 2), 'utf-8');
+  fs.renameSync(tempPath, cachePath);
+};
+
+const mergeCaches = (existing: OcrDiskCache | null, incoming: OcrDiskCache): OcrDiskCache => {
+  const now = new Date().toISOString();
+
+  if (existing && existing.fingerprint === incoming.fingerprint) {
+    return {
+      ...existing,
+      ...incoming,
+      created_at: existing.created_at,
+      updated_at: now,
+      pages: { ...existing.pages, ...incoming.pages },
+      images: { ...existing.images, ...incoming.images },
+    } satisfies OcrDiskCache;
+  }
+
+  return {
+    ...incoming,
+    created_at: incoming.created_at ?? existing?.created_at ?? now,
+    updated_at: now,
+    pages: incoming.pages ?? {},
+    images: incoming.images ?? {},
+  } satisfies OcrDiskCache;
+};
+
 /**
  * Save OCR cache to disk
  */
 export const saveOcrCache = (pdfPath: string, cache: OcrDiskCache): void => {
   const cachePath = getCacheFilePath(pdfPath);
+  const lockPath = `${cachePath}.lock`;
+  const lockFd = acquireCacheLock(lockPath);
 
   try {
-    // Update timestamp
-    cache.updated_at = new Date().toISOString();
-
-    // Ensure directory exists
-    const dir = path.dirname(cachePath);
-    if (!fs.existsSync(dir)) {
-      fs.mkdirSync(dir, { recursive: true });
-    }
+    const latest = loadOcrCache(pdfPath);
+    const merged = mergeCaches(latest, cache);
 
-    // Write with formatting for readability
-    fs.writeFileSync(cachePath, JSON.stringify(cache, null, 2), 'utf-8');
+    writeCacheFile(cachePath, merged);
 
     logger.debug('Saved OCR cache to disk', {
       cachePath,
-      pageCount: Object.keys(cache.pages).length,
-      imageCount: Object.keys(cache.images).length,
+      pageCount: Object.keys(merged.pages).length,
+      imageCount: Object.keys(merged.images).length,
     });
   } catch (error: unknown) {
     const message = error instanceof Error ? error.message : String(error);
     logger.error('Failed to save OCR cache', { cachePath, error: message });
     throw new Error(`Failed to save OCR cache: ${message}`);
+  } finally {
+    releaseCacheLock(lockPath, lockFd);
   }
 };
 

test/utils/diskCache.test.ts

@@ -0,0 +1,122 @@
+import fs from 'node:fs';
+import os from 'node:os';
+import path from 'node:path';
+import { afterEach, describe, expect, it } from 'vitest';
+import type { OcrDiskCache } from '../../src/types/cache.js';
+import { loadOcrCache, saveOcrCache } from '../../src/utils/diskCache.js';
+
+const fingerprint = 'disk-cache-fingerprint';
+const provider = 'test-provider';
+
+const createBaseCache = (pdfPath: string): OcrDiskCache => ({
+  fingerprint,
+  pdf_path: pdfPath,
+  created_at: new Date().toISOString(),
+  updated_at: new Date().toISOString(),
+  ocr_provider: provider,
+  pages: {},
+  images: {},
+});
+
+const scheduleWrite = (fn: () => void): Promise<void> =>
+  new Promise((resolve) => {
+    setTimeout(() => {
+      fn();
+      resolve();
+    }, 0);
+  });
+
+const createdDirs: string[] = [];
+
+afterEach(() => {
+  for (const dir of createdDirs) {
+    fs.rmSync(dir, { recursive: true, force: true });
+  }
+
+  createdDirs.length = 0;
+});
+
+describe('disk cache writes', () => {
+  it('merges parallel page writes so no entries are lost', async () => {
+    const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), 'disk-cache-'));
+    createdDirs.push(tempDir);
+
+    const pdfPath = path.join(tempDir, 'test.pdf');
+    const baseCache = createBaseCache(pdfPath);
+
+    const cacheA: OcrDiskCache = {
+      ...baseCache,
+      pages: {
+        '1': {
+          text: 'page-one',
+          provider_hash: 'provider-hash',
+          cached_at: new Date().toISOString(),
+        },
+      },
+    };
+
+    const cacheB: OcrDiskCache = {
+      ...baseCache,
+      pages: {
+        '2': {
+          text: 'page-two',
+          provider_hash: 'provider-hash',
+          cached_at: new Date().toISOString(),
+        },
+      },
+    };
+
+    await Promise.all([
+      scheduleWrite(() => saveOcrCache(pdfPath, cacheA)),
+      scheduleWrite(() => saveOcrCache(pdfPath, cacheB)),
+    ]);
+
+    const cache = loadOcrCache(pdfPath);
+
+    expect(cache?.pages['1']?.text).toBe('page-one');
+    expect(cache?.pages['2']?.text).toBe('page-two');
+    expect(Object.keys(cache?.pages ?? {})).toHaveLength(2);
+  });
+
+  it('merges page and image entries from overlapping writes', async () => {
+    const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), 'disk-cache-'));
+    createdDirs.push(tempDir);
+
+    const pdfPath = path.join(tempDir, 'test.pdf');
+    const baseCache = createBaseCache(pdfPath);
+
+    const pageCache: OcrDiskCache = {
+      ...baseCache,
+      pages: {
+        '3': {
+          text: 'page-three',
+          provider_hash: 'provider-hash',
+          cached_at: new Date().toISOString(),
+        },
+      },
+    };
+
+    const imageCache: OcrDiskCache = {
+      ...baseCache,
+      images: {
+        '3/0': {
+          text: 'image-text',
+          provider_hash: 'provider-hash',
+          cached_at: new Date().toISOString(),
+        },
+      },
+    };
+
+    await Promise.all([
+      scheduleWrite(() => saveOcrCache(pdfPath, pageCache)),
+      scheduleWrite(() => saveOcrCache(pdfPath, imageCache)),
+    ]);
+
+    const cache = loadOcrCache(pdfPath);
+
+    expect(cache?.pages['3']?.text).toBe('page-three');
+    expect(cache?.images['3/0']?.text).toBe('image-text');
+    expect(Object.keys(cache?.pages ?? {})).toHaveLength(1);
+    expect(Object.keys(cache?.images ?? {})).toHaveLength(1);
+  });
+});