feat(storage): MirrorDriver composite backend

server/src/nest/storage/drivers/mirror.driver.ts

@@ -0,0 +1,183 @@
+import fs from 'node:fs';
+import path from 'node:path';
+import { randomUUID } from 'node:crypto';
+import { pipeline } from 'node:stream/promises';
+import type { Readable } from 'node:stream';
+import { assertValidKey, assertValidPrefix } from '../storage-keys';
+import {
+  isLocalTempFile,
+  StorageInvalidKeyError,
+  StorageNotFoundError,
+  type ByteRange,
+  type LocalTempFile,
+  type ObjectStat,
+  type PutOptions,
+  type StorageDriver,
+} from '../storage.types';
+
+export interface ReplicaFailure {
+  backend: string;
+  key: string;
+  op: 'put' | 'delete';
+  error: string;
+  at: number;
+}
+
+/**
+ * Composite backend: one primary (source of truth) plus best-effort replicas.
+ * Runs entirely over the driver interface — the second in-tree consumer of
+ * the contract suite. Only the `backups` category may be assigned a mirror
+ * (registry validation enforces it).
+ *
+ * Writes hit the primary first and must succeed; replica failures are
+ * reported through `onReplicaFailure` (surfaced as health status), never
+ * thrown. Reads fall back to replicas only when the primary ERRORS — a plain
+ * miss stays a miss, replicas are copies, not a search path.
+ */
+export class MirrorDriver implements StorageDriver {
+  readonly id: string;
+  private readonly primary: StorageDriver;
+  private readonly replicas: readonly StorageDriver[];
+  private readonly tempDir: () => string;
+  private readonly onReplicaFailure?: (failure: ReplicaFailure) => void;
+
+  constructor(opts: {
+    id: string;
+    primary: StorageDriver;
+    replicas: StorageDriver[];
+    /** Global scratch dir (data/tmp) — mirrors have no same-volume spool of their own. */
+    tempDir: () => string;
+    onReplicaFailure?: (failure: ReplicaFailure) => void;
+  }) {
+    this.id = opts.id;
+    this.primary = opts.primary;
+    this.replicas = [...opts.replicas];
+    this.tempDir = opts.tempDir;
+    this.onReplicaFailure = opts.onReplicaFailure;
+  }
+
+  getSpoolDir(): null {
+    return null; // callers fall back to storage.tempDir()
+  }
+
+  getLocalPath(key: string): string | null {
+    assertValidKey(key);
+    return this.primary.getLocalPath?.(key) ?? null;
+  }
+
+  async put(key: string, source: Readable | LocalTempFile, opts?: PutOptions): Promise<void> {
+    assertValidKey(key);
+    // A Readable can only be consumed once, so materialize the source as a
+    // local file (stream → spool under tempDir(); temp-file sources as-is),
+    // then feed each target its own stream: the bytes survive the primary
+    // write for the replicas, and every child put stays atomic.
+    let file: string;
+    if (isLocalTempFile(source)) {
+      file = source.tmpPath;
+    } else {
+      file = path.join(this.tempDir(), randomUUID());
+      try {
+        await pipeline(source, fs.createWriteStream(file));
+      } catch (err) {
+        await fs.promises.rm(file, { force: true });
+        throw err;
+      }
+    }
+    try {
+      await this.primary.put(key, fs.createReadStream(file), opts);
+      for (const replica of this.replicas) {
+        try {
+          await replica.put(key, fs.createReadStream(file), opts);
+        } catch (err) {
+          this.reportReplicaFailure(replica.id, key, 'put', err);
+        }
+      }
+    } finally {
+      // Ownership of LocalTempFile sources transferred to us — consume it.
+      await fs.promises.rm(file, { force: true });
+    }
+  }
+
+  async getStream(key: string, range?: ByteRange): Promise<{ stream: Readable; stat: ObjectStat }> {
+    assertValidKey(key);
+    try {
+      return await this.primary.getStream(key, range);
+    } catch (err) {
+      if (!shouldFallback(err)) throw err;
+      return this.fromReplicas((replica) => replica.getStream(key, range), err);
+    }
+  }
+
+  async stat(key: string): Promise<ObjectStat | null> {
+    assertValidKey(key);
+    try {
+      return await this.primary.stat(key);
+    } catch (err) {
+      if (!shouldFallback(err)) throw err;
+      return this.fromReplicas((replica) => replica.stat(key), err);
+    }
+  }
+
+  async delete(key: string): Promise<void> {
+    assertValidKey(key);
+    await this.primary.delete(key); // hard primary failures propagate
+    for (const replica of this.replicas) {
+      try {
+        await replica.delete(key);
+      } catch (err) {
+        this.reportReplicaFailure(replica.id, key, 'delete', err);
+      }
+    }
+  }
+
+  async *list(prefix: string): AsyncIterable<ObjectStat> {
+    assertValidPrefix(prefix);
+    let yieldedAny = false;
+    try {
+      for await (const stat of this.primary.list(prefix)) {
+        yieldedAny = true;
+        yield stat;
+      }
+      return;
+    } catch (err) {
+      // Once items have been handed out, switching sources mid-stream would
+      // duplicate or drop keys — only an up-front primary failure falls back.
+      if (yieldedAny || !shouldFallback(err)) throw err;
+      for (const replica of this.replicas) {
+        try {
+          for await (const stat of replica.list(prefix)) yield stat;
+          return;
+        } catch {
+          /* try next replica */
+        }
+      }
+      throw err;
+    }
+  }
+
+  private async fromReplicas<T>(fn: (replica: StorageDriver) => Promise<T>, primaryErr: unknown): Promise<T> {
+    for (const replica of this.replicas) {
+      try {
+        return await fn(replica);
+      } catch {
+        /* try next replica */
+      }
+    }
+    throw primaryErr;
+  }
+
+  private reportReplicaFailure(backend: string, key: string, op: 'put' | 'delete', err: unknown): void {
+    this.onReplicaFailure?.({
+      backend,
+      key,
+      op,
+      error: err instanceof Error ? err.message : String(err),
+      at: Date.now(),
+    });
+  }
+}
+
+/** Fall back only on real errors — never on a miss (or a bad key, which is deterministic). */
+function shouldFallback(err: unknown): boolean {
+  return !(err instanceof StorageNotFoundError || err instanceof StorageInvalidKeyError);
+}

server/tests/unit/nest/storage/mirror.driver.test.ts

@@ -0,0 +1,180 @@
+import { describe, it, expect, vi, afterEach } from 'vitest';
+import fs from 'node:fs';
+import os from 'node:os';
+import path from 'node:path';
+import { Readable } from 'node:stream';
+import { LocalDriver } from '../../../../src/nest/storage/drivers/local.driver';
+import { MirrorDriver, type ReplicaFailure } from '../../../../src/nest/storage/drivers/mirror.driver';
+import {
+  StorageBackendError,
+  StorageNotFoundError,
+  type StorageDriver,
+} from '../../../../src/nest/storage/storage.types';
+import { describeStorageDriver, type DriverHarness } from './storage-driver.contract';
+
+function makeTmpDir(): string {
+  return fs.mkdtempSync(path.join(os.tmpdir(), 'trek-mirror-'));
+}
+
+function makeLocal(id: string): { driver: LocalDriver; root: string } {
+  const root = makeTmpDir();
+  const driver = new LocalDriver({ id, root });
+  driver.init({ cleanSpool: true });
+  return { driver, root };
+}
+
+interface MirrorFixture {
+  mirror: MirrorDriver;
+  primary: LocalDriver;
+  replica: LocalDriver;
+  spool: string;
+  failures: ReplicaFailure[];
+  dirs: string[];
+}
+
+function makeMirror(): MirrorFixture {
+  const p = makeLocal('primary-local');
+  const r = makeLocal('replica-local');
+  const spool = makeTmpDir();
+  const failures: ReplicaFailure[] = [];
+  const mirror = new MirrorDriver({
+    id: 'backup-mirror',
+    primary: p.driver,
+    replicas: [r.driver],
+    tempDir: () => spool,
+    onReplicaFailure: (f) => failures.push(f),
+  });
+  return { mirror, primary: p.driver, replica: r.driver, spool, failures, dirs: [p.root, r.root, spool] };
+}
+
+async function makeHarness(): Promise<DriverHarness> {
+  const fx = makeMirror();
+  const tmpFiles = makeTmpDir();
+  let counter = 0;
+  return {
+    driver: fx.mirror,
+    makeTempFile: async (contents) => {
+      const p = path.join(tmpFiles, `src-${counter++}.tmp`);
+      fs.writeFileSync(p, contents);
+      return p;
+    },
+    cleanup: async () => {
+      for (const dir of [...fx.dirs, tmpFiles]) fs.rmSync(dir, { recursive: true, force: true });
+    },
+  };
+}
+
+describeStorageDriver('MirrorDriver', makeHarness);
+
+async function readAll(driver: StorageDriver, key: string): Promise<string> {
+  const { stream } = await driver.getStream(key);
+  const chunks: Buffer[] = [];
+  for await (const chunk of stream) chunks.push(Buffer.from(chunk as Buffer));
+  return Buffer.concat(chunks).toString();
+}
+
+describe('MirrorDriver specifics', () => {
+  const cleanups: string[] = [];
+  const track = (fx: MirrorFixture): MirrorFixture => {
+    cleanups.push(...fx.dirs);
+    return fx;
+  };
+
+  afterEach(() => {
+    vi.restoreAllMocks();
+    while (cleanups.length) fs.rmSync(cleanups.pop()!, { recursive: true, force: true });
+  });
+
+  it('lands a stream put in the primary AND every replica', async () => {
+    const fx = track(makeMirror());
+    await fx.mirror.put('backup-1.zip', Readable.from('zip bytes'));
+    expect(await readAll(fx.primary, 'backup-1.zip')).toBe('zip bytes');
+    expect(await readAll(fx.replica, 'backup-1.zip')).toBe('zip bytes');
+  });
+
+  it('resolves put and reports when a replica write fails', async () => {
+    const fx = track(makeMirror());
+    vi.spyOn(fx.replica, 'put').mockRejectedValueOnce(new StorageBackendError('replica disk full'));
+
+    await expect(fx.mirror.put('backup-2.zip', Readable.from('payload'))).resolves.toBeUndefined();
+
+    expect(await readAll(fx.primary, 'backup-2.zip')).toBe('payload');
+    expect(fx.failures).toHaveLength(1);
+    expect(fx.failures[0]).toMatchObject({ backend: 'replica-local', key: 'backup-2.zip', op: 'put' });
+    expect(fx.failures[0].error).toContain('replica disk full');
+  });
+
+  it('rejects put and never touches replicas when the primary write fails', async () => {
+    const fx = track(makeMirror());
+    vi.spyOn(fx.primary, 'put').mockRejectedValueOnce(new StorageBackendError('primary gone'));
+    const replicaPut = vi.spyOn(fx.replica, 'put');
+
+    await expect(fx.mirror.put('backup-3.zip', Readable.from('x'))).rejects.toBeInstanceOf(StorageBackendError);
+    expect(replicaPut).not.toHaveBeenCalled();
+    expect(fx.failures).toHaveLength(0);
+  });
+
+  it('falls back to a replica when the primary read ERRORS', async () => {
+    const fx = track(makeMirror());
+    await fx.mirror.put('backup-4.zip', Readable.from('mirrored'));
+    vi.spyOn(fx.primary, 'getStream').mockRejectedValueOnce(new StorageBackendError('primary io error'));
+
+    expect(await readAll(fx.mirror, 'backup-4.zip')).toBe('mirrored');
+  });
+
+  it('does NOT fall back on a simple primary miss', async () => {
+    const fx = track(makeMirror());
+    // Plant the object only on the replica: a primary miss must stay a miss.
+    await fx.replica.put('only-on-replica.zip', Readable.from('ghost'));
+    const replicaGet = vi.spyOn(fx.replica, 'getStream');
+
+    await expect(fx.mirror.getStream('only-on-replica.zip')).rejects.toBeInstanceOf(StorageNotFoundError);
+    expect(replicaGet).not.toHaveBeenCalled();
+    expect(await fx.mirror.stat('only-on-replica.zip')).toBeNull();
+  });
+
+  it('falls back for stat and list on primary error', async () => {
+    const fx = track(makeMirror());
+    await fx.mirror.put('backup-5.zip', Readable.from('12345'));
+
+    vi.spyOn(fx.primary, 'stat').mockRejectedValueOnce(new StorageBackendError('io'));
+    expect((await fx.mirror.stat('backup-5.zip'))!.size).toBe(5);
+
+    vi.spyOn(fx.primary, 'list').mockImplementationOnce(() => {
+      throw new StorageBackendError('io');
+    });
+    const keys: string[] = [];
+    for await (const stat of fx.mirror.list('')) keys.push(stat.key);
+    expect(keys).toEqual(['backup-5.zip']);
+  });
+
+  it('fans delete out to all targets, tolerating a miss on either side', async () => {
+    const fx = track(makeMirror());
+    await fx.mirror.put('backup-6.zip', Readable.from('bye'));
+    await fx.replica.delete('backup-6.zip'); // already gone on the replica
+
+    await fx.mirror.delete('backup-6.zip');
+    expect(await fx.primary.stat('backup-6.zip')).toBeNull();
+    expect(await fx.replica.stat('backup-6.zip')).toBeNull();
+    expect(fx.failures).toHaveLength(0); // idempotent miss is not a failure
+  });
+
+  it('has no spool dir of its own and delegates getLocalPath to the primary', async () => {
+    const fx = track(makeMirror());
+    await fx.mirror.put('backup-7.zip', Readable.from('here'));
+
+    expect(fx.mirror.getSpoolDir()).toBeNull();
+    const local = fx.mirror.getLocalPath('backup-7.zip');
+    expect(local).toBe(fx.primary.getLocalPath('backup-7.zip'));
+  });
+
+  it('leaves no spool leftovers in tempDir on success or failure', async () => {
+    const fx = track(makeMirror());
+    await fx.mirror.put('backup-8.zip', Readable.from('ok'));
+    expect(fs.readdirSync(fx.spool)).toEqual([]);
+
+    vi.spyOn(fx.primary, 'put').mockRejectedValueOnce(new StorageBackendError('nope'));
+    await expect(fx.mirror.put('backup-9.zip', Readable.from('fail'))).rejects.toThrow();
+    expect(fs.readdirSync(fx.spool)).toEqual([]);
+  });
+});