diff --git a/server/src/nest/storage/drivers/mirror.driver.ts b/server/src/nest/storage/drivers/mirror.driver.ts new file mode 100644 index 000000000..b89cf91d1 --- /dev/null +++ b/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 { + 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 { + 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 { + 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 { + 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(fn: (replica: StorageDriver) => Promise, primaryErr: unknown): Promise { + 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); +} diff --git a/server/tests/unit/nest/storage/mirror.driver.test.ts b/server/tests/unit/nest/storage/mirror.driver.test.ts new file mode 100644 index 000000000..cef273023 --- /dev/null +++ b/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 { + 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 { + 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([]); + }); +});