From a0fe8f00d210e6b00d8ddb252d091b3e9183ce2b Mon Sep 17 00:00:00 2001 From: David Snelling Date: Wed, 29 Oct 2025 19:31:00 -0700 Subject: [PATCH] fix: add atomic writes to ALL file operations to prevent concurrent write corruption MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CRITICAL FIX v4.10.3: The v4.10.2 release only fixed VFS initialization but missed the underlying file corruption bug. During concurrent bulk imports, files were being truncated because writes were not atomic. Changes: 1. saveNode() - Added atomic temp+rename pattern for HNSW JSON files (line 252) 2. writeObjectToPath() - Added atomic temp+rename for compressed/uncompressed metadata (line 654) 3. saveEdge() - Added atomic temp+rename for verb JSON files (line 461) This prevents the Workshop team's reported errors: - SyntaxError: Unexpected end of JSON input (HNSW files truncated) - Error: unexpected end of file (compressed metadata truncated) All file write operations now use atomic temp file + rename sequence: 1. Write to temp file with unique timestamp + random suffix 2. Atomic rename temp → final (OS-level atomicity guarantee) 3. Clean up temp file on error This matches the atomic pattern already added to saveHNSWData() in v4.10.1, but now applied consistently across ALL file write operations. --- src/storage/adapters/fileSystemStorage.ts | 105 +++++++++++++++++----- 1 file changed, 85 insertions(+), 20 deletions(-) diff --git a/src/storage/adapters/fileSystemStorage.ts b/src/storage/adapters/fileSystemStorage.ts index f7c25e50..e7c535fa 100644 --- a/src/storage/adapters/fileSystemStorage.ts +++ b/src/storage/adapters/fileSystemStorage.ts @@ -248,6 +248,7 @@ export class FileSystemStorage extends BaseStorage { /** * Save a node to storage + * CRITICAL FIX (v4.10.3): Added atomic write pattern to prevent file corruption during concurrent imports */ protected async saveNode(node: HNSWNode): Promise { await this.ensureInitialized() @@ -266,11 +267,25 @@ export class FileSystemStorage extends BaseStorage { } const filePath = this.getNodePath(node.id) - await this.ensureDirectoryExists(path.dirname(filePath)) - await fs.promises.writeFile( - filePath, - JSON.stringify(serializableNode, null, 2) - ) + const tempPath = `${filePath}.tmp.${Date.now()}.${Math.random().toString(36).substring(2)}` + + try { + // ATOMIC WRITE SEQUENCE (v4.10.3): + // 1. Write to temp file + await this.ensureDirectoryExists(path.dirname(tempPath)) + await fs.promises.writeFile(tempPath, JSON.stringify(serializableNode, null, 2)) + + // 2. Atomic rename temp → final (crash-safe, prevents truncation during concurrent writes) + await fs.promises.rename(tempPath, filePath) + } catch (error: any) { + // Clean up temp file on any error + try { + await fs.promises.unlink(tempPath) + } catch (cleanupError) { + // Ignore cleanup errors + } + throw error + } // Count tracking happens in baseStorage.saveNounMetadata_internal (v4.1.2) // This fixes the race condition where metadata didn't exist yet @@ -442,6 +457,7 @@ export class FileSystemStorage extends BaseStorage { /** * Save an edge to storage + * CRITICAL FIX (v4.10.3): Added atomic write pattern to prevent file corruption during concurrent imports */ protected async saveEdge(edge: Edge): Promise { await this.ensureInitialized() @@ -466,11 +482,25 @@ export class FileSystemStorage extends BaseStorage { } const filePath = this.getVerbPath(edge.id) - await this.ensureDirectoryExists(path.dirname(filePath)) - await fs.promises.writeFile( - filePath, - JSON.stringify(serializableEdge, null, 2) - ) + const tempPath = `${filePath}.tmp.${Date.now()}.${Math.random().toString(36).substring(2)}` + + try { + // ATOMIC WRITE SEQUENCE (v4.10.3): + // 1. Write to temp file + await this.ensureDirectoryExists(path.dirname(tempPath)) + await fs.promises.writeFile(tempPath, JSON.stringify(serializableEdge, null, 2)) + + // 2. Atomic rename temp → final (crash-safe, prevents truncation during concurrent writes) + await fs.promises.rename(tempPath, filePath) + } catch (error: any) { + // Clean up temp file on any error + try { + await fs.promises.unlink(tempPath) + } catch (cleanupError) { + // Ignore cleanup errors + } + throw error + } // Count tracking happens in baseStorage.saveVerbMetadata_internal (v4.1.2) // This fixes the race condition where metadata didn't exist yet @@ -634,6 +664,7 @@ export class FileSystemStorage extends BaseStorage { * Primitive operation: Write object to path * All metadata operations use this internally via base class routing * v4.0.0: Supports gzip compression for 60-80% disk savings + * CRITICAL FIX (v4.10.3): Added atomic write pattern to prevent file corruption during concurrent imports */ protected async writeObjectToPath(pathStr: string, data: any): Promise { await this.ensureInitialized() @@ -642,16 +673,33 @@ export class FileSystemStorage extends BaseStorage { await this.ensureDirectoryExists(path.dirname(fullPath)) if (this.compressionEnabled) { - // Write compressed data with .gz extension + // Write compressed data with .gz extension using atomic pattern const compressedPath = `${fullPath}.gz` - const jsonString = JSON.stringify(data, null, 2) - const compressed = await new Promise((resolve, reject) => { - zlib.gzip(Buffer.from(jsonString, 'utf-8'), { level: this.compressionLevel }, (err: any, result: Buffer) => { - if (err) reject(err) - else resolve(result) + const tempPath = `${compressedPath}.tmp.${Date.now()}.${Math.random().toString(36).substring(2)}` + + try { + // ATOMIC WRITE SEQUENCE (v4.10.3): + // 1. Compress and write to temp file + const jsonString = JSON.stringify(data, null, 2) + const compressed = await new Promise((resolve, reject) => { + zlib.gzip(Buffer.from(jsonString, 'utf-8'), { level: this.compressionLevel }, (err: any, result: Buffer) => { + if (err) reject(err) + else resolve(result) + }) }) - }) - await fs.promises.writeFile(compressedPath, compressed) + await fs.promises.writeFile(tempPath, compressed) + + // 2. Atomic rename temp → final (crash-safe, prevents truncation during concurrent writes) + await fs.promises.rename(tempPath, compressedPath) + } catch (error: any) { + // Clean up temp file on any error + try { + await fs.promises.unlink(tempPath) + } catch (cleanupError) { + // Ignore cleanup errors + } + throw error + } // Clean up uncompressed file if it exists (migration from uncompressed) try { @@ -663,8 +711,25 @@ export class FileSystemStorage extends BaseStorage { } } } else { - // Write uncompressed data - await fs.promises.writeFile(fullPath, JSON.stringify(data, null, 2)) + // Write uncompressed data using atomic pattern + const tempPath = `${fullPath}.tmp.${Date.now()}.${Math.random().toString(36).substring(2)}` + + try { + // ATOMIC WRITE SEQUENCE (v4.10.3): + // 1. Write to temp file + await fs.promises.writeFile(tempPath, JSON.stringify(data, null, 2)) + + // 2. Atomic rename temp → final (crash-safe, prevents truncation during concurrent writes) + await fs.promises.rename(tempPath, fullPath) + } catch (error: any) { + // Clean up temp file on any error + try { + await fs.promises.unlink(tempPath) + } catch (cleanupError) { + // Ignore cleanup errors + } + throw error + } } }