fix(chunks): stop the absolute-offset symlink from vanishing on rewrite

src/store/fs-chunk-data-store.test.ts

@@ -35,6 +35,92 @@ describe('FsChunkDataStore', () => {
     rmSync(tempDir, { recursive: true, force: true });
   });
 
+  describe('absolute offset symlink', () => {
+    const dataRoot = 'wRq6f05oRupfTW_M5dcYBtwK5P8rSNYu20vC6D_o-M4';
+    const relativeOffset = 0;
+    const absoluteOffset = 388149830525175;
+    const chunkData: ChunkData = {
+      chunk: Buffer.from('test chunk data'),
+      hash: crypto.createHash('sha256').update('test chunk data').digest(),
+    };
+
+    const symlinkPath = () =>
+      join(
+        tempDir,
+        'data',
+        'by-absolute-offset',
+        '388',
+        '149',
+        absoluteOffset.toString(),
+      );
+
+    it('should not unlink when the existing link already points at the target', async () => {
+      const fsp = (await import('node:fs')).promises;
+
+      // First write establishes the link.
+      await store.set(dataRoot, relativeOffset, chunkData, absoluteOffset);
+      const before = await fsp.readlink(symlinkPath());
+
+      // Unlinking before re-linking would leave a window where the index entry
+      // does not exist; a concurrent read would see ENOENT and refetch data
+      // that is already on disk. A rewrite of the same offset must therefore
+      // leave the link untouched.
+      let unlinked = false;
+      const realUnlink = fsp.unlink.bind(fsp);
+      (fsp as any).unlink = async (p: any) => {
+        if (String(p).includes('by-absolute-offset')) unlinked = true;
+        return realUnlink(p);
+      };
+      try {
+        await store.set(dataRoot, relativeOffset, chunkData, absoluteOffset);
+      } finally {
+        (fsp as any).unlink = realUnlink;
+      }
+
+      assert.equal(unlinked, false, 'must not unlink an already-correct link');
+      assert.equal(await fsp.readlink(symlinkPath()), before);
+    });
+
+    it('should replace the link when the target genuinely differs', async () => {
+      const fsp = (await import('node:fs')).promises;
+      const otherRoot = 'aBq6f05oRupfTW_M5dcYBtwK5P8rSNYu20vC6D_o-M4';
+
+      await store.set(dataRoot, relativeOffset, chunkData, absoluteOffset);
+      const first = await fsp.readlink(symlinkPath());
+
+      // Replacement must be atomic too: unlinking the live path would
+      // reintroduce the window this change closes, so the retarget has to go
+      // through rename() and must never unlink the index path itself.
+      let unlinkedIndexPath = false;
+      const realUnlink = fsp.unlink.bind(fsp);
+      (fsp as any).unlink = async (p: any) => {
+        if (String(p) === symlinkPath()) unlinkedIndexPath = true;
+        return realUnlink(p);
+      };
+      try {
+        // Same absolute offset, different data root: the index must follow it.
+        await store.set(otherRoot, relativeOffset, chunkData, absoluteOffset);
+      } finally {
+        (fsp as any).unlink = realUnlink;
+      }
+      const second = await fsp.readlink(symlinkPath());
+
+      assert.equal(
+        unlinkedIndexPath,
+        false,
+        'retarget must replace atomically, not unlink the live path',
+      );
+      assert.notEqual(second, first);
+      assert.ok(second.includes(otherRoot));
+
+      // No temporary links left behind.
+      const dir = await fsp.readdir(
+        join(tempDir, 'data', 'by-absolute-offset', '388', '149'),
+      );
+      assert.deepEqual(dir, [absoluteOffset.toString()]);
+    });
+  });
+
   describe('set', () => {
     it('should save chunk data to the correct path', async () => {
       const dataRoot = 'wRq6f05oRupfTW_M5dcYBtwK5P8rSNYu20vC6D_o-M4';

src/store/fs-chunk-data-store.ts

@@ -199,6 +199,16 @@ export class FsChunkDataStore implements ChunkDataStore {
     }
   }
 
+  /**
+   * Point the absolute-offset index at a chunk, creating or updating the link.
+   *
+   * The entry is never momentarily absent: the common case (an existing link
+   * already pointing at this target) is a no-op, and a genuine retarget is
+   * applied with an atomic rename. A concurrent reader therefore always
+   * resolves either the previous target or the new one. The index is
+   * best-effort -- failures are logged and swallowed so they cannot prevent
+   * the chunk itself from being cached.
+   */
   private async createAbsoluteOffsetSymlink(
     dataRoot: string,
     relativeOffset: number,
@@ -214,14 +224,42 @@ export class FsChunkDataStore implements ChunkDataStore {
         this.chunkDataRootPath(dataRoot, relativeOffset),
       );
 
-      // Remove existing symlink if present (allows updating)
+      // Link directly rather than unlinking first. Unlinking opens a window in
+      // which the index entry does not exist: a concurrent read of this offset
+      // sees ENOENT, treats it as a cache miss, and refetches data that is
+      // already on disk. Concurrent writers for the same offset resolve to the
+      // same target, so an EEXIST whose target already matches is a no-op, not
+      // an error -- which is what made this the hottest error in the log.
+      // A genuinely different target still gets replaced: that is the
+      // "allows updating" case the unlink was there for.
       try {
-        await fs.promises.unlink(symlinkPath);
-      } catch {
-        // Ignore if doesn't exist
+        await fs.promises.symlink(targetPath, symlinkPath);
+      } catch (error: any) {
+        if (error.code !== 'EEXIST') {
+          throw error;
+        }
+        const existing = await fs.promises
+          .readlink(symlinkPath)
+          .catch(() => undefined);
+        if (existing === targetPath) {
+          return;
+        }
+        // Replace atomically. Unlinking first would reintroduce the very
+        // window this change exists to close: rename() over an existing path
+        // is atomic within a filesystem, so a concurrent read always sees
+        // either the old link or the new one, never nothing. The temporary
+        // name is unique so concurrent replacements cannot collide on it.
+        const tmpPath = `${symlinkPath}.tmp-${crypto
+          .randomBytes(8)
+          .toString('hex')}`;
+        await fs.promises.symlink(targetPath, tmpPath);
+        try {
+          await fs.promises.rename(tmpPath, symlinkPath);
+        } catch (renameError: any) {
+          await fs.promises.unlink(tmpPath).catch(() => undefined);
+          throw renameError;
+        }
       }
-
-      await fs.promises.symlink(targetPath, symlinkPath);
     } catch (error: any) {
       this.log.error('Failed to create absolute offset symlink', {
         dataRoot,

src/store/fs-chunk-metadata-store.ts

@@ -4,6 +4,7 @@
  *
  * SPDX-License-Identifier: AGPL-3.0-or-later
  */
+import crypto from 'node:crypto';
 import fs from 'node:fs';
 import path from 'node:path';
 import winston from 'winston';
@@ -181,6 +182,16 @@ export class FsChunkMetadataStore implements ChunkMetadataStore {
     }
   }
 
+  /**
+   * Point the absolute-offset index at a chunk, creating or updating the link.
+   *
+   * The entry is never momentarily absent: the common case (an existing link
+   * already pointing at this target) is a no-op, and a genuine retarget is
+   * applied with an atomic rename. A concurrent reader therefore always
+   * resolves either the previous target or the new one. The index is
+   * best-effort -- failures are logged and swallowed so they cannot prevent
+   * the chunk itself from being cached.
+   */
   private async createAbsoluteOffsetSymlink(
     dataRoot: string,
     relativeOffset: number,
@@ -196,17 +207,42 @@ export class FsChunkMetadataStore implements ChunkMetadataStore {
         this.chunkMetadataPath(dataRoot, relativeOffset),
       );
 
-      // Remove existing symlink if present (allows updating).
-      // Note: Race condition possible between unlink and symlink if another
-      // process creates a symlink at the same path. We catch all errors below
-      // to ensure cache write succeeds - symlink index is best-effort.
+      // Link directly rather than unlinking first. Unlinking opens a window in
+      // which the index entry does not exist: a concurrent read of this offset
+      // sees ENOENT, treats it as a cache miss, and refetches data that is
+      // already on disk. Concurrent writers for the same offset resolve to the
+      // same target, so an EEXIST whose target already matches is a no-op, not
+      // an error -- which is what made this the hottest error in the log.
+      // A genuinely different target still gets replaced: that is the
+      // "allows updating" case the unlink was there for.
       try {
-        await fs.promises.unlink(symlinkPath);
-      } catch {
-        // Ignore if doesn't exist
+        await fs.promises.symlink(targetPath, symlinkPath);
+      } catch (error: any) {
+        if (error.code !== 'EEXIST') {
+          throw error;
+        }
+        const existing = await fs.promises
+          .readlink(symlinkPath)
+          .catch(() => undefined);
+        if (existing === targetPath) {
+          return;
+        }
+        // Replace atomically. Unlinking first would reintroduce the very
+        // window this change exists to close: rename() over an existing path
+        // is atomic within a filesystem, so a concurrent read always sees
+        // either the old link or the new one, never nothing. The temporary
+        // name is unique so concurrent replacements cannot collide on it.
+        const tmpPath = `${symlinkPath}.tmp-${crypto
+          .randomBytes(8)
+          .toString('hex')}`;
+        await fs.promises.symlink(targetPath, tmpPath);
+        try {
+          await fs.promises.rename(tmpPath, symlinkPath);
+        } catch (renameError: any) {
+          await fs.promises.unlink(tmpPath).catch(() => undefined);
+          throw renameError;
+        }
       }
-
-      await fs.promises.symlink(targetPath, symlinkPath);
     } catch (error: any) {
       this.log.error('Failed to create absolute offset symlink', {
         dataRoot,