diff --git a/server/src/nest/storage/storage-events.service.ts b/server/src/nest/storage/storage-events.service.ts new file mode 100644 index 000000000..551cf9aab --- /dev/null +++ b/server/src/nest/storage/storage-events.service.ts @@ -0,0 +1,27 @@ +import { Injectable } from '@nestjs/common'; +import type { ReplicaFailure } from './drivers/mirror.driver'; + +/** + * In-process pub/sub for storage health events. Exists so consumers OUTSIDE + * StorageModule (the notifications notifier) can observe replica failures + * without StorageModule importing NotificationsModule — that would cycle + * (Notifications → Auth → Storage). + */ +@Injectable() +export class StorageEventsService { + private readonly replicaFailureSubscribers: Array<(failure: ReplicaFailure) => void> = []; + + onReplicaFailure(subscriber: (failure: ReplicaFailure) => void): void { + this.replicaFailureSubscribers.push(subscriber); + } + + emitReplicaFailure(failure: ReplicaFailure): void { + for (const subscriber of this.replicaFailureSubscribers) { + try { + subscriber(failure); + } catch { + // A broken subscriber must never break the write path. + } + } + } +} diff --git a/server/src/nest/storage/storage-jobs.service.ts b/server/src/nest/storage/storage-jobs.service.ts new file mode 100644 index 000000000..a59c27d1c --- /dev/null +++ b/server/src/nest/storage/storage-jobs.service.ts @@ -0,0 +1,111 @@ +import { Injectable, Logger } from '@nestjs/common'; +import { STORAGE_CATEGORIES, type StorageBackfillStatus } from '@trek/shared'; +import { MirrorDriver } from './drivers/mirror.driver'; +import { StorageRegistryService } from './storage-registry.service'; + +export class BackfillTargetError extends Error {} +export class BackfillBusyError extends Error {} + +interface ActiveJob { + status: StorageBackfillStatus; + cancelled: boolean; +} + +/** + * One-at-a-time backfill registry (the ImportJobsService idiom, minus WS). + * Finished statuses linger for ttlMs so a remounted panel can show the + * outcome; the client polls the admin state while anything is running. + */ +@Injectable() +export class StorageJobsService { + private readonly logger = new Logger(StorageJobsService.name); + private readonly jobs = new Map(); + private ttlMs = 10 * 60_000; + + constructor(private readonly registry: StorageRegistryService) {} + + /** Test-only factory for a short TTL, since a second constructor param would confuse Nest DI. */ + static withTtl(registry: StorageRegistryService, ttlMs: number): StorageJobsService { + const svc = new StorageJobsService(registry); + svc.ttlMs = ttlMs; + return svc; + } + + startBackfill(mirrorName: string): void { + if ([...this.jobs.values()].some((job) => job.status.status === 'running')) { + throw new BackfillBusyError('a sync is already running — one backfill at a time'); + } + // Scope rule: the categories whose EFFECTIVE backend is this mirror. + const snapshot = this.registry.snapshot(); + const categories = STORAGE_CATEGORIES.filter( + (category) => snapshot.categories[category]?.backend === mirrorName, + ); + if (categories.length === 0) { + throw new BackfillTargetError(`'${mirrorName}' is not a mirror routed by any category`); + } + const resolved = categories.map((category) => this.registry.resolve(category)); + const driver = resolved[0]!.driver; + if (!(driver instanceof MirrorDriver)) { + throw new BackfillTargetError(`'${mirrorName}' is not a mirror backend`); + } + const prefixes = resolved.map((r) => r.keyPrefix); + + const job: ActiveJob = { + cancelled: false, + status: { + backend: mirrorName, + status: 'running', + done: 0, + total: 0, + copied: 0, + skipped: 0, + failed: 0, + startedAt: Date.now(), + }, + }; + this.jobs.set(mirrorName, job); + + // Detached: the driver instances are resolved above, so a registry + // reload() mid-run keeps this job on them (the in-flight guarantee). + void driver + .backfill(prefixes, { + onProgress: (progress) => { + job.status = { ...job.status, ...progress }; + }, + isCancelled: () => job.cancelled, + }) + .then((result) => { + job.status = { + ...job.status, + ...result, + status: result.cancelled ? 'cancelled' : 'done', + finishedAt: Date.now(), + }; + }) + .catch((err: unknown) => { + job.status = { + ...job.status, + status: 'error', + error: err instanceof Error ? err.message : String(err), + finishedAt: Date.now(), + }; + this.logger.error(`backfill '${mirrorName}' aborted: ${job.status.error}`); + }) + .finally(() => { + setTimeout(() => { + if (this.jobs.get(mirrorName) === job && job.status.status !== 'running') this.jobs.delete(mirrorName); + }, this.ttlMs).unref?.(); + }); + } + + cancelBackfill(mirrorName: string): boolean { + const job = this.jobs.get(mirrorName); + if (!job || job.status.status !== 'running') return false; + job.cancelled = true; + return true; + } + + statuses(): StorageBackfillStatus[] { + return [...this.jobs.values()].map((job) => ({ ...job.status })); + } +} diff --git a/server/src/nest/storage/storage-registry.service.ts b/server/src/nest/storage/storage-registry.service.ts index d810b27ab..a322d523a 100644 --- a/server/src/nest/storage/storage-registry.service.ts +++ b/server/src/nest/storage/storage-registry.service.ts @@ -7,6 +7,7 @@ import { decrypt_api_key } from '../common/crypto/apiKeyCrypto'; import { LocalDriver } from './drivers/local.driver'; import { MirrorDriver, type ReplicaFailure } from './drivers/mirror.driver'; import { S3Driver } from './drivers/s3.driver'; +import { StorageEventsService } from './storage-events.service'; import { DEFAULT_BACKUPS_ROOT, DEFAULT_UPLOADS_ROOT, GLOBAL_TEMP_DIR, SEED_CONFIG_PATH } from './storage-paths'; import { assertNoMaskSentinels, @@ -121,6 +122,7 @@ export class StorageRegistryService implements OnModuleInit { constructor( private readonly db: DatabaseService, private readonly env: RuntimeEnvService, + private readonly events: StorageEventsService, ) {} onModuleInit(): void { @@ -173,6 +175,7 @@ export class StorageRegistryService implements OnModuleInit { if (this.failures.length > REPLICA_FAILURE_RING_SIZE) { this.failures = this.failures.slice(-REPLICA_FAILURE_RING_SIZE); } + this.events.emitReplicaFailure(failure); } replicaFailures(): readonly ReplicaFailure[] { diff --git a/server/src/nest/storage/storage.module.ts b/server/src/nest/storage/storage.module.ts index a87336058..43937b221 100644 --- a/server/src/nest/storage/storage.module.ts +++ b/server/src/nest/storage/storage.module.ts @@ -1,6 +1,8 @@ import { Module } from '@nestjs/common'; import { AppConfigModule } from '../app-config/app-config.module'; import { AuditModule } from '../audit/audit.module'; +import { StorageEventsService } from './storage-events.service'; +import { StorageJobsService } from './storage-jobs.service'; import { StorageRegistryService } from './storage-registry.service'; import { StorageService } from './storage.service'; import { StorageAdminService } from './storage-admin.service'; @@ -22,7 +24,7 @@ import { StorageAdminController } from './storage-admin.controller'; @Module({ imports: [AppConfigModule, AuditModule], controllers: [StorageAdminController], - providers: [StorageRegistryService, StorageService, StorageAdminService], - exports: [StorageService], + providers: [StorageRegistryService, StorageService, StorageAdminService, StorageEventsService, StorageJobsService], + exports: [StorageService, StorageEventsService], }) export class StorageModule {} diff --git a/server/tests/unit/nest/storage/storage-admin.service.test.ts b/server/tests/unit/nest/storage/storage-admin.service.test.ts index 275cc245f..2fa3f8a43 100644 --- a/server/tests/unit/nest/storage/storage-admin.service.test.ts +++ b/server/tests/unit/nest/storage/storage-admin.service.test.ts @@ -20,6 +20,7 @@ import { DatabaseService } from '../../../../src/nest/database/database.service' import type { RuntimeEnvService } from '../../../../src/nest/app-config/runtime-env.service'; import { encrypt_api_key } from '../../../../src/nest/common/crypto/apiKeyCrypto'; import { StorageAdminService } from '../../../../src/nest/storage/storage-admin.service'; +import { StorageEventsService } from '../../../../src/nest/storage/storage-events.service'; import { StorageRegistryService, BACKENDS_KEY, CATEGORIES_KEY } from '../../../../src/nest/storage/storage-registry.service'; import { StorageService } from '../../../../src/nest/storage/storage.service'; @@ -62,7 +63,7 @@ function makeService(opts: { encryptionKeySet?: boolean; uploadsRoot?: string } const env = { env: () => ({ paths: {}, security: { encryptionKeySet: opts.encryptionKeySet ?? true } }), } as unknown as RuntimeEnvService; - const registry = new StorageRegistryService(db, env); + const registry = new StorageRegistryService(db, env, new StorageEventsService()); registry.onModuleInit(); const service = new StorageAdminService(db, registry, new StorageService(registry), env); return { service, registry, uploadsRoot }; diff --git a/server/tests/unit/nest/storage/storage-jobs.service.test.ts b/server/tests/unit/nest/storage/storage-jobs.service.test.ts new file mode 100644 index 000000000..0c8822890 --- /dev/null +++ b/server/tests/unit/nest/storage/storage-jobs.service.test.ts @@ -0,0 +1,136 @@ +import { describe, it, expect, vi, beforeAll, beforeEach, afterEach } from 'vitest'; + +const { testDb, dbMock } = vi.hoisted(() => { + const Database = require('better-sqlite3'); + const db = new Database(':memory:'); + db.exec('PRAGMA journal_mode = WAL'); + db.exec('PRAGMA foreign_keys = ON'); + return { testDb: db, dbMock: { db, closeDb: () => {}, reinitialize: () => {} } }; +}); +vi.mock('../../../../src/db/database', () => dbMock); +vi.mock('../../../../src/config', () => ({ ENCRYPTION_KEY: 'storage-jobs-test-key' })); + +import fs from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +import { Readable } from 'node:stream'; +import { createTables } from '../../../../src/db/schema'; +import { runMigrations } from '../../../../src/db/migrations'; +import { DatabaseService } from '../../../../src/nest/database/database.service'; +import type { RuntimeEnvService } from '../../../../src/nest/app-config/runtime-env.service'; +import { StorageEventsService } from '../../../../src/nest/storage/storage-events.service'; +import { StorageRegistryService } from '../../../../src/nest/storage/storage-registry.service'; +import { StorageService } from '../../../../src/nest/storage/storage.service'; +import { + BackfillBusyError, + BackfillTargetError, + StorageJobsService, +} from '../../../../src/nest/storage/storage-jobs.service'; + +const db = new DatabaseService(testDb); + +beforeAll(() => { + createTables(testDb); + runMigrations(testDb); +}); + +const tmpDirs: string[] = []; +function makeTmpDir(): string { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'trek-jobs-')); + tmpDirs.push(dir); + return dir; +} +function setSetting(key: string, value: string): void { + testDb.prepare('INSERT OR REPLACE INTO app_settings (key, value) VALUES (?, ?)').run(key, value); +} +beforeEach(() => { + testDb.prepare("DELETE FROM app_settings WHERE key LIKE 'storage.%'").run(); +}); +afterEach(() => { + vi.restoreAllMocks(); + while (tmpDirs.length) fs.rmSync(tmpDirs.pop()!, { recursive: true, force: true }); +}); + +/** A registry whose backups category routes through mirror 'm' (nas replica). */ +function makeWorld() { + const uploadsRoot = makeTmpDir(); + const backupsRoot = makeTmpDir(); + const nasRoot = makeTmpDir(); + setSetting( + 'storage.backends', + JSON.stringify([ + { name: 'uploads-local', type: 'local', options: { root: uploadsRoot } }, + { name: 'backups-local', type: 'local', options: { root: backupsRoot } }, + { name: 'nas', type: 'local', options: { root: nasRoot } }, + { name: 'm', type: 'mirror', options: { primary: 'backups-local', replicas: ['nas'] } }, + ]), + ); + setSetting('storage.categories', JSON.stringify({ backups: 'm' })); + const env = { env: () => ({ paths: {}, security: { encryptionKeySet: true } }) } as unknown as RuntimeEnvService; + const registry = new StorageRegistryService(db, env, new StorageEventsService()); + registry.onModuleInit(); + const storage = new StorageService(registry); + const jobs = new StorageJobsService(registry); + return { registry, storage, jobs, backupsRoot, nasRoot }; +} + +async function waitFor(predicate: () => boolean, ms = 5000): Promise { + const start = Date.now(); + while (!predicate()) { + if (Date.now() - start > ms) throw new Error('timed out waiting for job'); + await new Promise((r) => setTimeout(r, 20)); + } +} + +describe('StorageJobsService', () => { + it('JOBS-001 backfills the categories routed through the mirror and lands on done', async () => { + const { storage, jobs, nasRoot } = makeWorld(); + await storage.put('backups', 'old-backup.zip', Readable.from('zipzip')); + fs.rmSync(path.join(nasRoot, 'old-backup.zip'), { force: true }); // simulate pre-mirror object + jobs.startBackfill('m'); + await waitFor(() => jobs.statuses().some((s) => s.backend === 'm' && s.status === 'done')); + const status = jobs.statuses().find((s) => s.backend === 'm')!; + expect(status).toMatchObject({ status: 'done', total: 1, done: 1, copied: 1, failed: 0 }); + expect(fs.existsSync(path.join(nasRoot, 'old-backup.zip'))).toBe(true); + }); + + it('JOBS-002 rejects an unrouted or non-mirror name with BackfillTargetError', () => { + const { jobs } = makeWorld(); + expect(() => jobs.startBackfill('nas')).toThrow(BackfillTargetError); + expect(() => jobs.startBackfill('ghost')).toThrow(BackfillTargetError); + }); + + it('JOBS-003 a second start while one runs throws BackfillBusyError (global, either backend)', async () => { + const { storage, jobs } = makeWorld(); + for (let i = 0; i < 20; i++) await storage.put('backups', `b${i}.zip`, Readable.from('x'.repeat(1000))); + jobs.startBackfill('m'); + expect(() => jobs.startBackfill('m')).toThrow(BackfillBusyError); + await waitFor(() => jobs.statuses().some((s) => s.status !== 'running')); + }); + + it('JOBS-004 cancel flips a running job to cancelled; cancelling a finished/unknown one returns false', async () => { + const { storage, jobs } = makeWorld(); + for (let i = 0; i < 50; i++) await storage.put('backups', `c${i}.zip`, Readable.from('y'.repeat(2000))); + jobs.startBackfill('m'); + expect(jobs.cancelBackfill('m')).toBe(true); + await waitFor(() => jobs.statuses().some((s) => s.backend === 'm' && s.status !== 'running')); + expect(jobs.statuses().find((s) => s.backend === 'm')!.status).toBe('cancelled'); + expect(jobs.cancelBackfill('m')).toBe(false); + expect(jobs.cancelBackfill('ghost')).toBe(false); + }); + + it('JOBS-005 finished statuses expire after the TTL', async () => { + vi.useFakeTimers(); + try { + const { storage, jobs } = makeWorld(); + await storage.put('backups', 'one.zip', Readable.from('z')); + jobs.startBackfill('m'); + // Drain the real async job under fake timers by flushing microtasks. + await vi.waitFor(() => expect(jobs.statuses()[0]?.status).toBe('done')); + vi.advanceTimersByTime(10 * 60_000 + 1000); + expect(jobs.statuses()).toEqual([]); + } finally { + vi.useRealTimers(); + } + }); +}); diff --git a/server/tests/unit/nest/storage/storage-registry.service.test.ts b/server/tests/unit/nest/storage/storage-registry.service.test.ts index 010a3803d..78bb7b645 100644 --- a/server/tests/unit/nest/storage/storage-registry.service.test.ts +++ b/server/tests/unit/nest/storage/storage-registry.service.test.ts @@ -25,6 +25,7 @@ import { runMigrations } from '../../../../src/db/migrations'; import { DatabaseService } from '../../../../src/nest/database/database.service'; import type { RuntimeEnvService } from '../../../../src/nest/app-config/runtime-env.service'; import { encrypt_api_key } from '../../../../src/nest/common/crypto/apiKeyCrypto'; +import { StorageEventsService } from '../../../../src/nest/storage/storage-events.service'; import { StorageRegistryService } from '../../../../src/nest/storage/storage-registry.service'; import { LocalDriver } from '../../../../src/nest/storage/drivers/local.driver'; import { MirrorDriver, type ReplicaFailure } from '../../../../src/nest/storage/drivers/mirror.driver'; @@ -99,7 +100,7 @@ function makeRegistry(opts: RegistryOpts = {}) { { placePhotoDir: opts.placePhotoDir }, { encryptionKeySet: opts.encryptionKeySet ?? true }, ); - const registry = new StorageRegistryService(db, stub.env); + const registry = new StorageRegistryService(db, stub.env, new StorageEventsService()); if (opts.boot !== false) registry.onModuleInit(); return { registry, uploadsRoot, setUploadsRoot: (root: string) => rewriteUploadsOverride(root) }; } @@ -479,7 +480,7 @@ describe('seed-once storage-config.json import', () => { /** A registry with NO storage.* rows (makeRegistry seeds an override row, so build raw). */ function makeUnseededRegistry(opts: { encryptionKeySet?: boolean } = {}) { const stub = makeEnvStub({}, { encryptionKeySet: opts.encryptionKeySet ?? true }); - return new StorageRegistryService(db, stub.env); + return new StorageRegistryService(db, stub.env, new StorageEventsService()); } function readRow(key: string): string | undefined {