mirror of
https://github.com/mauriceboe/TREK.git
synced 2026-11-13 07:40:59 +00:00
fix(storage): keep the reclaim pass and a backup restore apart
A restore wipes and rehydrates the uploads under a swapped database, but it never took the storage job slot, so a reclaim pass could start or carry on beside it, and a delete already past the generation check could land on an object the restore had just put back. BackupService now marks a restore as running on StorageJobsService and waits for a pass in flight to let go; a pass refuses to start while a restore runs and stops at its next check when one begins. Backfills and migrations are not affected. The service test also reads its fake source sets through a helper instead of non-null assertions.
This commit is contained in:
@@ -1,4 +1,5 @@
|
||||
import { DATABASE_BACKUP, type DatabaseBackupStrategy } from '../database/database-backup.interface';
|
||||
import { StorageJobsService } from '../storage/storage-jobs.service';
|
||||
import { StorageService } from '../storage/storage.service';
|
||||
import * as svc from './backup.impl';
|
||||
import { Inject, Injectable } from '@nestjs/common';
|
||||
@@ -23,6 +24,7 @@ export class BackupService {
|
||||
constructor(
|
||||
private readonly storage: StorageService,
|
||||
@Inject(DATABASE_BACKUP) database: DatabaseBackupStrategy,
|
||||
private readonly jobs: StorageJobsService,
|
||||
) {
|
||||
this.deps = { storage, database };
|
||||
}
|
||||
@@ -34,10 +36,10 @@ export class BackupService {
|
||||
return svc.createBackup(this.deps, prefix);
|
||||
}
|
||||
restoreFromZip(zipPath: string) {
|
||||
return svc.restoreFromZip(this.deps, zipPath);
|
||||
return this.whileRestoring(() => svc.restoreFromZip(this.deps, zipPath));
|
||||
}
|
||||
restoreBackup(filename: string) {
|
||||
return svc.restoreBackup(this.deps, filename);
|
||||
return this.whileRestoring(() => svc.restoreBackup(this.deps, filename));
|
||||
}
|
||||
deleteBackup(filename: string) {
|
||||
return svc.deleteBackup(this.storage, filename);
|
||||
@@ -59,4 +61,19 @@ export class BackupService {
|
||||
get rateWindow() {
|
||||
return svc.BACKUP_RATE_WINDOW;
|
||||
}
|
||||
|
||||
/**
|
||||
* A restore wipes and rehydrates the uploads under a swapped database, so
|
||||
* the storage reclaim pass must not run beside it: the restore marks itself
|
||||
* running (a pass refuses to start and one in flight stops at its next
|
||||
* check) and waits for a pass in flight to let go before it touches anything.
|
||||
*/
|
||||
private async whileRestoring<T>(restore: () => Promise<T>): Promise<T> {
|
||||
await this.jobs.beginRestore();
|
||||
try {
|
||||
return await restore();
|
||||
} finally {
|
||||
this.jobs.endRestore();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -43,6 +43,11 @@ export class StorageJobsService {
|
||||
private ttlMs = 10 * 60_000;
|
||||
/** A pass that holds the one-job slot without a status of its own (the reclaim pass), by name. */
|
||||
private exclusive: string | null = null;
|
||||
/** Restores in progress: no exclusive pass starts while one runs, and one in flight stops. */
|
||||
private restores = 0;
|
||||
/** Settles once the exclusive pass holding the slot releases it; settled while none does. */
|
||||
private exclusiveReleased: Promise<void> = Promise.resolve();
|
||||
private onExclusiveReleased: (() => void) | null = null;
|
||||
|
||||
/** One storage job at a time, across backfills, category migrations and the reclaim pass. */
|
||||
private anyJobRunning(): boolean {
|
||||
@@ -60,17 +65,45 @@ export class StorageJobsService {
|
||||
|
||||
/**
|
||||
* Takes the one-job slot for a pass that runs in the caller (the reclaim
|
||||
* pass): false while a backfill, a migration or another such pass runs.
|
||||
* pass): false while a backfill, a migration, another such pass or a
|
||||
* restore runs.
|
||||
* The caller releases it in a finally.
|
||||
*/
|
||||
claimExclusive(name: string): boolean {
|
||||
if (this.anyJobRunning()) return false;
|
||||
if (this.restores > 0 || this.anyJobRunning()) return false;
|
||||
this.exclusive = name;
|
||||
this.exclusiveReleased = new Promise((resolve) => {
|
||||
this.onExclusiveReleased = resolve;
|
||||
});
|
||||
return true;
|
||||
}
|
||||
|
||||
releaseExclusive(name: string): void {
|
||||
if (this.exclusive === name) this.exclusive = null;
|
||||
if (this.exclusive !== name) return;
|
||||
this.exclusive = null;
|
||||
this.onExclusiveReleased?.();
|
||||
this.onExclusiveReleased = null;
|
||||
}
|
||||
|
||||
/** True while a backup restore swaps the database and the uploads. */
|
||||
get restoreRunning(): boolean {
|
||||
return this.restores > 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* Marks a restore as running and waits until an exclusive pass in flight
|
||||
* (the reclaim pass, which stops at its next check of `restoreRunning`) has
|
||||
* let go, so none of its deletes can land on the uploads the restore puts
|
||||
* back. Backfills and migrations are not waited for, as before. The caller
|
||||
* calls `endRestore` in a finally.
|
||||
*/
|
||||
async beginRestore(): Promise<void> {
|
||||
this.restores += 1;
|
||||
await this.exclusiveReleased;
|
||||
}
|
||||
|
||||
endRestore(): void {
|
||||
if (this.restores > 0) this.restores -= 1;
|
||||
}
|
||||
|
||||
constructor(private readonly registry: StorageRegistryService) {}
|
||||
|
||||
@@ -61,7 +61,10 @@ interface ReclaimRunOptions {
|
||||
now?: Date;
|
||||
}
|
||||
|
||||
/** The database was swapped under the pass (a restore): nothing it learnt before holds any more. */
|
||||
/**
|
||||
* A restore began, or the database was swapped under the pass: nothing it
|
||||
* learnt before holds any more. Reported as `database_reopened` either way.
|
||||
*/
|
||||
class ReclaimStopped extends Error {}
|
||||
|
||||
/**
|
||||
@@ -78,8 +81,9 @@ class ReclaimStopped extends Error {}
|
||||
* registered, whose objects share a location with another category's, whose
|
||||
* listing or lookups fail, or where more than a fifth of the objects look
|
||||
* unreferenced is skipped and its marks start over. A pass never runs beside
|
||||
* a backfill or a category migration (the one-job slot of StorageJobsService),
|
||||
* never on a demo instance, and stops when the database is reopened under it.
|
||||
* a backfill, a category migration or a restore (the one-job slot of
|
||||
* StorageJobsService, which a restore waits on), never on a demo instance,
|
||||
* and stops when a restore begins or the database is reopened under it.
|
||||
*/
|
||||
@Injectable()
|
||||
export class StorageReclaimService implements OnModuleInit {
|
||||
@@ -377,7 +381,9 @@ export class StorageReclaimService implements OnModuleInit {
|
||||
return true;
|
||||
}
|
||||
|
||||
/** A restore that began, or a reopen that happened, since the pass started ends it. */
|
||||
private assertSameDatabase(ctx: PassContext): void {
|
||||
if (this.jobs.restoreRunning) throw new ReclaimStopped('a restore began during the pass');
|
||||
if (this.generation !== ctx.generation) throw new ReclaimStopped('the database was reopened during the pass');
|
||||
}
|
||||
}
|
||||
|
||||
@@ -37,6 +37,8 @@ import { Module } from '@nestjs/common';
|
||||
* and follows database reopens (DatabaseLifecycleModule); the domains whose
|
||||
* rows name stored objects register their reference sources with the exported
|
||||
* StorageReclaimRegistryService, so storage never imports them.
|
||||
* StorageJobsService is exported for the backup restore, which marks itself
|
||||
* running there and waits for a reclaim pass in flight to let go.
|
||||
*/
|
||||
@Module({
|
||||
imports: [
|
||||
@@ -59,6 +61,6 @@ import { Module } from '@nestjs/common';
|
||||
StorageReclaimService,
|
||||
StorageReclaimJob,
|
||||
],
|
||||
exports: [StorageService, StorageEventsService, StorageReclaimRegistryService],
|
||||
exports: [StorageService, StorageEventsService, StorageJobsService, StorageReclaimRegistryService],
|
||||
})
|
||||
export class StorageModule {}
|
||||
|
||||
@@ -6,6 +6,8 @@ import { BackupController } from '../../../src/nest/backup/backup.controller';
|
||||
import * as backupSvc from '../../../src/nest/backup/backup.impl';
|
||||
import { BackupService as RealBackupService } from '../../../src/nest/backup/backup.service';
|
||||
import type { BackupService } from '../../../src/nest/backup/backup.service';
|
||||
import { StorageJobsService } from '../../../src/nest/storage/storage-jobs.service';
|
||||
import type { StorageRegistryService } from '../../../src/nest/storage/storage-registry.service';
|
||||
import { StorageNotFoundError } from '../../../src/nest/storage/storage.types';
|
||||
import type { User } from '../../../src/types';
|
||||
import { HttpException } from '@nestjs/common';
|
||||
@@ -374,7 +376,9 @@ describe('BackupService (wrapper)', () => {
|
||||
const database = {
|
||||
__sentinel: 'database',
|
||||
} as unknown as import('../../../src/nest/database/database-backup.interface').DatabaseBackupStrategy;
|
||||
const wrapper = new RealBackupService(storage, database);
|
||||
// The real one-job slot: the restore marks itself running there.
|
||||
const jobs = new StorageJobsService({} as StorageRegistryService);
|
||||
const wrapper = new RealBackupService(storage, database, jobs);
|
||||
|
||||
it('forwards every call straight to the legacy backup service', async () => {
|
||||
expect(await wrapper.listBackups()).toEqual([{ filename: 'svc.zip' }]);
|
||||
@@ -409,6 +413,33 @@ describe('BackupService (wrapper)', () => {
|
||||
it('exposes the legacy rate window', () => {
|
||||
expect(wrapper.rateWindow).toBe(backupSvc.BACKUP_RATE_WINDOW);
|
||||
});
|
||||
|
||||
it('BKS-RESTORE-001: a restore runs marked as such, so the reclaim pass can neither start nor go on beside it', async () => {
|
||||
let runningInside = false;
|
||||
vi.mocked(backupSvc.restoreFromZip).mockImplementationOnce(async () => {
|
||||
runningInside = jobs.restoreRunning;
|
||||
expect(jobs.claimExclusive('a storage reclaim')).toBe(false);
|
||||
return { success: true };
|
||||
});
|
||||
await expect(wrapper.restoreFromZip('/tmp/a.zip')).resolves.toEqual({ success: true });
|
||||
expect(runningInside).toBe(true);
|
||||
expect(jobs.restoreRunning).toBe(false);
|
||||
});
|
||||
|
||||
it('BKS-RESTORE-002: a restore waits for a reclaim pass in flight to let go, and a failed one still ends', async () => {
|
||||
expect(jobs.claimExclusive('a storage reclaim')).toBe(true);
|
||||
vi.mocked(backupSvc.restoreBackup).mockRejectedValueOnce(new Error('broken archive'));
|
||||
const restore = wrapper.restoreBackup('svc.zip');
|
||||
await Promise.resolve();
|
||||
expect(jobs.restoreRunning).toBe(true);
|
||||
expect(backupSvc.restoreBackup).not.toHaveBeenCalled();
|
||||
jobs.releaseExclusive('a storage reclaim');
|
||||
await expect(restore).rejects.toThrow('broken archive');
|
||||
expect(backupSvc.restoreBackup).toHaveBeenCalledWith({ storage, database }, 'svc.zip');
|
||||
expect(jobs.restoreRunning).toBe(false);
|
||||
expect(jobs.claimExclusive('a storage reclaim')).toBe(true);
|
||||
jobs.releaseExclusive('a storage reclaim');
|
||||
});
|
||||
});
|
||||
|
||||
describe('BackupModule', () => {
|
||||
|
||||
@@ -62,6 +62,16 @@ let named: Map<ServedCategory, Set<string>>;
|
||||
/** Called on every question a source is asked, before it answers. */
|
||||
let onAsk: ((category: ServedCategory, keys: readonly string[]) => void | Promise<void>) | null;
|
||||
|
||||
/** The keys the fake sources name in `category`, created empty on first use. */
|
||||
function namedIn(category: ServedCategory): Set<string> {
|
||||
let set = named.get(category);
|
||||
if (!set) {
|
||||
set = new Set<string>();
|
||||
named.set(category, set);
|
||||
}
|
||||
return set;
|
||||
}
|
||||
|
||||
function fakeSource(id: StorageReferenceSourceId): StorageReferenceSource {
|
||||
const categories: readonly ServedCategory[] =
|
||||
id === 'auth' ? ['avatars'] : id === 'files' ? ['files'] : STORAGE_RECLAIMABLE_CATEGORIES;
|
||||
@@ -208,12 +218,12 @@ describe('StorageReclaimService: mark, then sweep', () => {
|
||||
await seedCovers(['orphan.jpg']);
|
||||
await reclaim.setDeleteEnabled(true);
|
||||
await reclaim.run({ now: T0 });
|
||||
named.get('covers')!.add('orphan.jpg');
|
||||
namedIn('covers').add('orphan.jpg');
|
||||
const renamed = await reclaim.run({ now: new Date(T0.getTime() + DAY) });
|
||||
expect(renamed.categories.covers.unreferenced.objects).toBe(0);
|
||||
expect(await marksOf('covers')).toEqual([]);
|
||||
|
||||
named.get('covers')!.delete('orphan.jpg');
|
||||
namedIn('covers').delete('orphan.jpg');
|
||||
const again = await reclaim.run({ now: WEEK_LATER });
|
||||
expect(again.categories.covers.eligible.objects).toBe(0);
|
||||
expect(fs.existsSync(objectPath('covers', 'orphan.jpg'))).toBe(true);
|
||||
@@ -225,7 +235,8 @@ describe('StorageReclaimService: mark, then sweep', () => {
|
||||
await reclaim.setDeleteEnabled(true);
|
||||
await reclaim.run({ now: T0 });
|
||||
onAsk = (category, keys) => {
|
||||
if (category === 'covers' && keys.length === 1) named.get('covers')!.add(keys[0]!);
|
||||
const [only] = keys;
|
||||
if (category === 'covers' && keys.length === 1 && only !== undefined) namedIn('covers').add(only);
|
||||
};
|
||||
const report = await reclaim.run({ now: WEEK_LATER });
|
||||
expect(report.categories.covers.eligible.objects).toBe(1);
|
||||
@@ -462,4 +473,41 @@ describe('StorageReclaimService: the switch and the stored report', () => {
|
||||
await setAppSetting(orm, 'storage.reclaim_report', JSON.stringify({ startedAt: 'yesterday' }));
|
||||
expect((await reclaim.state()).lastRun).toBeNull();
|
||||
});
|
||||
|
||||
it('RCLS-022: a restore that begins mid-pass stops it before the delete, and waits for it to let go', async () => {
|
||||
await seedCovers(['orphan.jpg']);
|
||||
await reclaim.setDeleteEnabled(true);
|
||||
await reclaim.run({ now: T0 });
|
||||
const jobs = t.get(StorageJobsService);
|
||||
let restoreStarted: Promise<void> | null = null;
|
||||
let restoreWent = false;
|
||||
onAsk = (_category, keys) => {
|
||||
if (keys.length !== 1 || restoreStarted) return;
|
||||
restoreStarted = jobs.beginRestore().then(() => {
|
||||
restoreWent = true;
|
||||
});
|
||||
};
|
||||
try {
|
||||
const report = await reclaim.run({ now: WEEK_LATER });
|
||||
expect(report.stopped).toBe('database_reopened');
|
||||
expect(fs.existsSync(objectPath('covers', 'orphan.jpg'))).toBe(true);
|
||||
expect(restoreStarted).not.toBeNull();
|
||||
await restoreStarted;
|
||||
expect(restoreWent).toBe(true);
|
||||
} finally {
|
||||
jobs.endRestore();
|
||||
}
|
||||
});
|
||||
|
||||
it('RCLS-023: a pass refuses to start while a restore runs, and starts again once it ended', async () => {
|
||||
const jobs = t.get(StorageJobsService);
|
||||
await jobs.beginRestore();
|
||||
try {
|
||||
await expect(reclaim.run({ now: T0 })).rejects.toMatchObject({ code: 'reclaim_busy' });
|
||||
} finally {
|
||||
jobs.endRestore();
|
||||
}
|
||||
expect(jobs.restoreRunning).toBe(false);
|
||||
expect((await reclaim.run({ now: T0 })).stopped).toBeNull();
|
||||
});
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user