Harden RedXaiHM transactions, mutations, locks, and recovery

src/database.js

@@ -60,81 +60,90 @@ export class RedXaiDatabase {
   }
 
   setById(id, value) {
-    const node = this.requireById(id);
-    node.value = valueFrom(value);
-    this.commitMutation();
-    return node;
+    return this.atomicMutation(() => {
+      const node = this.requireById(id);
+      node.value = valueFrom(value);
+      return node;
+    });
   }
 
   renameById(id, name) {
     if (!/^[A-Za-z_][A-Za-z0-9_.-]*$/.test(name)) throw new TypeError('Invalid RedXai variable name');
-    const node = this.requireById(id);
-    node.name = name;
-    this.commitMutation();
-    return node;
+    return this.atomicMutation(() => {
+      const node = this.requireById(id);
+      node.name = name;
+      return node;
+    });
   }
 
   add(databaseIndex, name, value, id = null) {
-    const database = this.document.databases[databaseIndex];
-    if (!database) throw new RangeError(`Database index ${databaseIndex} does not exist`);
-    const actualId = id ?? this.nextId();
-    const node = createAssignment(name, value, actualId);
-    database.entries.push(node);
-    this.commitMutation();
-    return node;
+    return this.atomicMutation(() => {
+      const database = this.document.databases[databaseIndex];
+      if (!database) throw new RangeError(`Database index ${databaseIndex} does not exist`);
+      const actualId = id ?? this.nextId();
+      const node = createAssignment(name, value, actualId);
+      database.entries.push(node);
+      return node;
+    });
   }
 
   addToArray(arrayId, name, value, id = null) {
-    const owner = this.requireById(arrayId);
-    if (owner.value.valueType !== 'array') throw new TypeError(`ID ${arrayId} is not an array`);
-    const actualId = id ?? this.nextId();
-    const node = createAssignment(name, value, actualId);
-    owner.value.items.push(node);
-    this.commitMutation();
-    return node;
+    return this.atomicMutation(() => {
+      const owner = this.requireById(arrayId);
+      if (owner.value.valueType !== 'array') throw new TypeError(`ID ${arrayId} is not an array`);
+      const actualId = id ?? this.nextId();
+      const node = createAssignment(name, value, actualId);
+      owner.value.items.push(node);
+      return node;
+    });
   }
 
   deleteById(id) {
-    const record = this.idIndex.get(id);
-    if (!record) return false;
-    record.container.splice(record.index, 1);
-    this.commitMutation();
-    return true;
+    if (!this.idIndex.has(id)) return false;
+    return this.atomicMutation(() => {
+      const record = this.idIndex.get(id);
+      if (!record) return false;
+      record.container.splice(record.index, 1);
+      return true;
+    });
   }
 
   moveById(id, targetArrayId, index = null) {
-    const record = this.idIndex.get(id);
-    if (!record) throw new RangeError(`RedXai ID ${id} was not found`);
-    if (id === targetArrayId) throw new TypeError('A value cannot be moved inside itself');
-    const target = this.requireById(targetArrayId);
-    if (target.value.valueType !== 'array') throw new TypeError(`Target ID ${targetArrayId} is not an array`);
-    if (containsId(record.node, targetArrayId)) throw new TypeError('A value cannot be moved into one of its descendants');
-
-    const [node] = record.container.splice(record.index, 1);
-    const insertion = index == null ? target.value.items.length : Math.max(0, Math.min(index, target.value.items.length));
-    target.value.items.splice(insertion, 0, node);
-    this.commitMutation();
-    return node;
+    return this.atomicMutation(() => {
+      const record = this.idIndex.get(id);
+      if (!record) throw new RangeError(`RedXai ID ${id} was not found`);
+      if (id === targetArrayId) throw new TypeError('A value cannot be moved inside itself');
+      const target = this.requireById(targetArrayId);
+      if (target.value.valueType !== 'array') throw new TypeError(`Target ID ${targetArrayId} is not an array`);
+      if (containsId(record.node, targetArrayId)) throw new TypeError('A value cannot be moved into one of its descendants');
+
+      const [node] = record.container.splice(record.index, 1);
+      const insertion = index == null ? target.value.items.length : Math.max(0, Math.min(index, target.value.items.length));
+      target.value.items.splice(insertion, 0, node);
+      return node;
+    });
   }
 
   copyById(id, options = {}) {
-    const record = this.idIndex.get(id);
-    if (!record) throw new RangeError(`RedXai ID ${id} was not found`);
-    const clone = structuredClone(record.node);
-    const idMap = new Map();
-    let next = this.nextId();
-    remapIds(clone, (oldId, isRoot) => {
-      const replacement = isRoot && options.newId != null ? options.newId : next++;
-      if (oldId != null) idMap.set(oldId, replacement);
-      return replacement;
-    }, true);
-    if (options.newName) clone.name = options.newName;
-
-    const target = options.targetArrayId == null ? record.container : this.requireArray(options.targetArrayId).value.items;
-    const insertion = options.index == null ? target.length : Math.max(0, Math.min(options.index, target.length));
-    target.splice(insertion, 0, clone);
-    this.commitMutation();
-    return { node: clone, idMap };
+    return this.atomicMutation(() => {
+      const record = this.idIndex.get(id);
+      if (!record) throw new RangeError(`RedXai ID ${id} was not found`);
+      const clone = structuredClone(record.node);
+      const idMap = new Map();
+      let next = this.nextId();
+      remapIds(clone, (oldId, isRoot) => {
+        const replacement = isRoot && options.newId != null ? options.newId : next++;
+        if (oldId != null) idMap.set(oldId, replacement);
+        return replacement;
+      }, true);
+      remapInternalReferences(clone, idMap);
+      if (options.newName) clone.name = options.newName;
+
+      const target = options.targetArrayId == null ? record.container : this.requireArray(options.targetArrayId).value.items;
+      const insertion = options.index == null ? target.length : Math.max(0, Math.min(options.index, target.length));
+      target.splice(insertion, 0, clone);
+      return { node: clone, idMap };
+    });
   }
 
   requireArray(id) {
@@ -166,6 +175,20 @@ export class RedXaiDatabase {
     return structuredClone(this.document);
   }
 
+  atomicMutation(callback) {
+    const before = structuredClone(this.document);
+    try {
+      const result = callback();
+      if (this.options.validate) assertValid(this.document);
+      this.rebuildIndex();
+      return result;
+    } catch (error) {
+      this.document = before;
+      this.rebuildIndex();
+      throw error;
+    }
+  }
+
   commitMutation() {
     if (this.options.validate) assertValid(this.document);
     this.rebuildIndex();
@@ -188,3 +211,30 @@ function remapIds(node, allocator, isRoot = false) {
     }
   }
 }
+
+function remapInternalReferences(node, idMap) {
+  walkNodeValues(node, (value) => {
+    if (value.valueType !== 'reference' || value.target?.type !== 'id') return;
+    const replacement = idMap.get(value.target.value);
+    if (replacement != null) value.target.value = replacement;
+  });
+}
+
+function walkNodeValues(node, visitor) {
+  if (node.kind !== 'assignment') return;
+  visitor(node.value);
+  if (node.value.valueType !== 'array' && node.value.valueType !== 'collection') return;
+  node.value.items.forEach((item) => {
+    if (item.kind === 'assignment') walkNodeValues(item, visitor);
+    else if (item.kind === 'value') walkValue(item, visitor);
+  });
+}
+
+function walkValue(value, visitor) {
+  visitor(value);
+  if (value.valueType !== 'array' && value.valueType !== 'collection') return;
+  value.items.forEach((item) => {
+    if (item.kind === 'assignment') walkNodeValues(item, visitor);
+    else if (item.kind === 'value') walkValue(item, visitor);
+  });
+}

src/store.js

@@ -1,3 +1,4 @@
+import { randomUUID } from 'node:crypto';
 import { open, readFile, rename, copyFile, mkdir, stat, unlink } from 'node:fs/promises';
 import { dirname } from 'node:path';
 import { parse } from './parser.js';
@@ -46,46 +47,98 @@ export class RedXaiFileStore {
 
   async save(filePath, document) {
     assertExtension(filePath);
-    const warnings = assertValid(document);
-    const source = serialize(document, { indent: this.options.indent });
+    const prepared = this.prepareDocument(document);
+    await mkdir(dirname(filePath), { recursive: true });
+    const release = await this.acquireLock(filePath);
+    try {
+      return await this.writeLocked(filePath, prepared);
+    } finally {
+      await release();
+    }
+  }
+
+  async transaction(filePath, callback) {
+    assertExtension(filePath);
     await mkdir(dirname(filePath), { recursive: true });
     const release = await this.acquireLock(filePath);
-    const temporaryPath = `${filePath}.tmp-${process.pid}-${Date.now()}`;
     try {
-      if (this.options.backup && await exists(filePath)) await copyFile(filePath, `${filePath}.bak`);
+      // Loading while holding the same lock prevents the classic read-modify-write
+      // race where two writers both commit from an identical stale snapshot.
+      const loaded = await this.load(filePath);
+      const result = await loaded.database.transaction(callback);
+      const prepared = this.prepareDocument(loaded.database.document);
+      await this.writeLocked(filePath, prepared);
+      return result;
+    } finally {
+      await release();
+    }
+  }
+
+  prepareDocument(document) {
+    const warnings = assertValid(document);
+    const source = serialize(document, { indent: this.options.indent });
+    return { source, warnings };
+  }
+
+  async writeLocked(filePath, prepared) {
+    const temporaryPath = `${filePath}.tmp-${process.pid}-${randomUUID()}`;
+    const backupPath = `${filePath}.bak`;
+    const backupTemporaryPath = `${backupPath}.tmp-${process.pid}-${randomUUID()}`;
+    try {
+      if (this.options.backup && await exists(filePath)) {
+        await copyFile(filePath, backupTemporaryPath);
+        await syncFile(backupTemporaryPath);
+        await rename(backupTemporaryPath, backupPath);
+      }
+
       const handle = await open(temporaryPath, 'wx', 0o600);
       try {
-        await handle.writeFile(source, 'utf8');
+        await handle.writeFile(prepared.source, 'utf8');
         await handle.sync();
       } finally {
         await handle.close();
       }
       await rename(temporaryPath, filePath);
       await syncDirectory(dirname(filePath));
-      return { filePath, bytes: Buffer.byteLength(source), warnings };
+      return {
+        filePath,
+        bytes: Buffer.byteLength(prepared.source),
+        warnings: prepared.warnings,
+      };
     } catch (error) {
       await unlink(temporaryPath).catch(() => {});
+      await unlink(backupTemporaryPath).catch(() => {});
       throw new RedXaiStorageError(`Failed to save ${filePath}`, { cause: error });
-    } finally {
-      await release();
     }
   }
 
-  async transaction(filePath, callback) {
-    const loaded = await this.load(filePath);
-    const result = await loaded.database.transaction(callback);
-    await this.save(filePath, loaded.database.document);
-    return result;
-  }
-
   async acquireLock(filePath) {
     const lockPath = `${filePath}.lock`;
     const started = Date.now();
     while (true) {
       try {
         const handle = await open(lockPath, 'wx', 0o600);
-        await handle.writeFile(JSON.stringify({ pid: process.pid, createdAt: new Date().toISOString() }));
+        await handle.writeFile(JSON.stringify({
+          pid: process.pid,
+          createdAt: new Date().toISOString(),
+          token: randomUUID(),
+        }));
+        await handle.sync();
+
+        // Refresh the lock mtime while a valid writer is alive. A long transaction
+        // must not be mistaken for a crashed process merely because staleLockMs elapsed.
+        const heartbeatMs = Math.max(100, Math.floor(this.options.staleLockMs / 3));
+        const heartbeat = setInterval(() => {
+          const now = new Date();
+          handle.utimes(now, now).catch(() => {});
+        }, heartbeatMs);
+        heartbeat.unref?.();
+
+        let released = false;
         return async () => {
+          if (released) return;
+          released = true;
+          clearInterval(heartbeat);
           await handle.close().catch(() => {});
           await unlink(lockPath).catch(() => {});
         };
@@ -129,12 +182,21 @@ async function isStale(lockPath, staleMs) {
   }
 }
 
+async function syncFile(filePath) {
+  const handle = await open(filePath, 'r');
+  try {
+    await handle.sync();
+  } finally {
+    await handle.close();
+  }
+}
+
 async function syncDirectory(directory) {
   try {
     const handle = await open(directory, 'r');
     try { await handle.sync(); } finally { await handle.close(); }
   } catch {
-    // Some platforms do not support fsync on directories. The file itself is already synced.
+    // Some platforms do not support fsync on directories. The files themselves are synced.
   }
 }