From 86a7fca3755092d2eb251c02b880e2175a4c9a15 Mon Sep 17 00:00:00 2001 From: jubnl Date: Thu, 20 Aug 2026 22:35:45 +0200 Subject: [PATCH] =?UTF-8?q?feat(server):=20MirrorDriver.backfill=20?= =?UTF-8?q?=E2=80=94=20prefix-scoped,=20size-compared,=20cancellable=20rep?= =?UTF-8?q?lica=20catch-up?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../src/nest/storage/drivers/mirror.driver.ts | 92 ++++++++++++++ .../unit/nest/storage/mirror-backfill.test.ts | 112 ++++++++++++++++++ 2 files changed, 204 insertions(+) create mode 100644 server/tests/unit/nest/storage/mirror-backfill.test.ts diff --git a/server/src/nest/storage/drivers/mirror.driver.ts b/server/src/nest/storage/drivers/mirror.driver.ts index bcdb26996..351f9f2e2 100644 --- a/server/src/nest/storage/drivers/mirror.driver.ts +++ b/server/src/nest/storage/drivers/mirror.driver.ts @@ -23,6 +23,23 @@ export interface ReplicaFailure { at: number; } +export interface BackfillProgress { + done: number; + total: number; + copied: number; + skipped: number; + failed: number; +} + +export interface BackfillHooks { + onProgress(progress: BackfillProgress): void; + isCancelled(): boolean; +} + +export interface BackfillResult extends BackfillProgress { + cancelled: boolean; +} + /** * Composite backend: one primary (source of truth) plus best-effort replicas. * Runs entirely over the driver interface — the second in-tree consumer of @@ -157,6 +174,81 @@ export class MirrorDriver implements StorageDriver { } } + /** + * Copy existing primary objects under the given prefixes to every replica + * (backfill/stats spec). NOT part of the StorageDriver contract — mirror + * specific and additive. Two phases: enumerate (honest total), then copy. + * `done` counts keys examined; copied/skipped/failed count replica-level + * outcomes. Replica errors (stat or put) flow through the same failure + * funnel the write path uses and never abort the run; PRIMARY errors + * (list/getStream) propagate — a getStream failure is deliberately kept + * outside the per-replica try/catch so it can never be misclassified as a + * replica failure. Objects are compared by size only — the same criterion + * a re-put would settle anyway. + */ + async backfill(prefixes: readonly string[], hooks: BackfillHooks): Promise { + const progress: BackfillProgress = { done: 0, total: 0, copied: 0, skipped: 0, failed: 0 }; + for (const prefix of prefixes) { + for await (const _stat of this.primary.list(prefix)) progress.total += 1; + } + hooks.onProgress({ ...progress }); + + for (const prefix of prefixes) { + for await (const stat of this.primary.list(prefix)) { + if (hooks.isCancelled()) return { ...progress, cancelled: true }; + + // Phase A: decide, per replica, whether it needs this key. Replica + // stat failures are reported and counted here — they never touch the + // primary-only try/catch below. + const needy: StorageDriver[] = []; + for (const replica of this.replicas) { + try { + const existing = await replica.stat(stat.key); + if (existing && existing.size === stat.size) { + progress.skipped += 1; + } else { + needy.push(replica); + } + } catch (err) { + progress.failed += 1; + this.reportReplicaFailure(replica.id, stat.key, 'put', err); + } + } + + // Phase B: only if some replica needs the bytes, materialize the key + // once (the put() spool pattern). A primary getStream/pipeline + // failure here RETHROWS — it is a primary error, not a replica one. + if (needy.length > 0) { + const file = path.join(this.tempDir(), randomUUID()); + try { + const { stream } = await this.primary.getStream(stat.key); + await pipeline(stream, fs.createWriteStream(file)); + } catch (err) { + await fs.promises.rm(file, { force: true }); + throw err; + } + try { + for (const replica of needy) { + try { + await replica.put(stat.key, fs.createReadStream(file)); + progress.copied += 1; + } catch (err) { + progress.failed += 1; + this.reportReplicaFailure(replica.id, stat.key, 'put', err); + } + } + } finally { + await fs.promises.rm(file, { force: true }); + } + } + + progress.done += 1; + hooks.onProgress({ ...progress }); + } + } + return { ...progress, cancelled: false }; + } + private async fromReplicas(fn: (replica: StorageDriver) => Promise, primaryErr: unknown): Promise { for (const replica of this.replicas) { try { diff --git a/server/tests/unit/nest/storage/mirror-backfill.test.ts b/server/tests/unit/nest/storage/mirror-backfill.test.ts new file mode 100644 index 000000000..224d5db53 --- /dev/null +++ b/server/tests/unit/nest/storage/mirror-backfill.test.ts @@ -0,0 +1,112 @@ +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'; + +const tmpDirs: string[] = []; +function makeTmpDir(): string { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'trek-backfill-')); + tmpDirs.push(dir); + return dir; +} +afterEach(() => { + while (tmpDirs.length) fs.rmSync(tmpDirs.pop()!, { recursive: true, force: true }); +}); + +function makeLocal(id: string): { driver: LocalDriver; root: string } { + const root = makeTmpDir(); + const driver = new LocalDriver({ id, root }); + driver.init({ ensurePrefixes: [], cleanSpool: false }); + return { driver, root }; +} + +async function put(driver: LocalDriver, key: string, body: string): Promise { + await driver.put(key, Readable.from(body)); +} + +function makeMirror(primary: LocalDriver, replicas: LocalDriver[], onReplicaFailure?: (f: ReplicaFailure) => void): MirrorDriver { + return new MirrorDriver({ id: 'm', primary, replicas, tempDir: () => makeTmpDir(), onReplicaFailure }); +} + +const HOOKS = { onProgress: () => undefined, isCancelled: () => false }; + +describe('MirrorDriver.backfill', () => { + it('BKFL-001 copies missing objects to every replica, scoped to the given prefixes', async () => { + const { driver: primary } = makeLocal('p'); + const { driver: replica, root: replicaRoot } = makeLocal('r'); + await put(primary, 'backups/a.zip', 'aaa'); + await put(primary, 'covers/skip.jpg', 'not in scope'); + const result = await makeMirror(primary, [replica]).backfill(['backups/'], HOOKS); + expect(result).toMatchObject({ done: 1, total: 1, copied: 1, skipped: 0, failed: 0, cancelled: false }); + expect(fs.readFileSync(path.join(replicaRoot, 'backups/a.zip'), 'utf8')).toBe('aaa'); + expect(fs.existsSync(path.join(replicaRoot, 'covers/skip.jpg'))).toBe(false); // scope rule + }); + + it('BKFL-002 skips size-matched objects and re-copies size mismatches', async () => { + const { driver: primary } = makeLocal('p'); + const { driver: replica, root: replicaRoot } = makeLocal('r'); + await put(primary, 'backups/same.zip', 'equal'); + await put(replica, 'backups/same.zip', 'eq2al'); // same size → skipped + await put(primary, 'backups/diff.zip', 'longer-content'); + await put(replica, 'backups/diff.zip', 'short'); // size mismatch → re-copied + const result = await makeMirror(primary, [replica]).backfill(['backups/'], HOOKS); + expect(result).toMatchObject({ done: 2, copied: 1, skipped: 1, failed: 0 }); + expect(fs.readFileSync(path.join(replicaRoot, 'backups/diff.zip'), 'utf8')).toBe('longer-content'); + expect(fs.readFileSync(path.join(replicaRoot, 'backups/same.zip'), 'utf8')).toBe('eq2al'); // untouched + }); + + it('BKFL-003 a failing replica reports through the failure hook, counts failed, and the run continues', async () => { + const { driver: primary } = makeLocal('p'); + const { driver: good, root: goodRoot } = makeLocal('good'); + const bad = { + id: 'bad', + stat: vi.fn().mockResolvedValue(null), + put: vi.fn().mockRejectedValue(new Error('replica on fire')), + } as unknown as LocalDriver; + const failures: ReplicaFailure[] = []; + await put(primary, 'backups/a.zip', 'aaa'); + await put(primary, 'backups/b.zip', 'bbb'); + const result = await makeMirror(primary, [bad, good], (f) => failures.push(f)).backfill(['backups/'], HOOKS); + expect(result).toMatchObject({ done: 2, copied: 2, failed: 2, cancelled: false }); // good got both; bad failed both + expect(failures).toHaveLength(2); + expect(failures[0]).toMatchObject({ backend: 'bad', op: 'put' }); + expect(failures[0]!.error).toContain('replica on fire'); + expect(fs.existsSync(path.join(goodRoot, 'backups/b.zip'))).toBe(true); + }); + + it('BKFL-004 cancel stops after the in-flight key', async () => { + const { driver: primary } = makeLocal('p'); + const { driver: replica } = makeLocal('r'); + for (let i = 0; i < 5; i++) await put(primary, `backups/${i}.zip`, `content-${i}`); + let examined = 0; + const hooks = { + onProgress: (p: { done: number }) => { + examined = p.done; + }, + isCancelled: () => examined >= 2, + }; + const result = await makeMirror(primary, [replica]).backfill(['backups/'], hooks); + expect(result.cancelled).toBe(true); + expect(result.done).toBeLessThan(5); + }); + + it('BKFL-005 a primary error aborts (propagates) instead of being swallowed', async () => { + const { driver: replica } = makeLocal('r'); + const explodingPrimary = { + id: 'p', + list: () => ({ + [Symbol.asyncIterator]() { + return { + next: () => Promise.reject(new Error('primary listing failed')), + }; + }, + }), + } as unknown as LocalDriver; + await expect(makeMirror(explodingPrimary, [replica]).backfill(['backups/'], HOOKS)).rejects.toThrow( + 'primary listing failed', + ); + }); +});