fix(server): vacuum after migrations, concurrent reindex (#31424)
This commit is contained in:
@@ -241,12 +241,21 @@ export class DatabaseRepository {
|
||||
SET DATA TYPE vector(${sql.raw(String(dimSize))})`.execute(tx);
|
||||
await sql.raw(vectorIndexQuery({ vectorExtension, table, indexName, lists })).execute(tx);
|
||||
});
|
||||
try {
|
||||
await sql`VACUUM ANALYZE ${sql.raw(table)}`.execute(this.db);
|
||||
} catch (error: any) {
|
||||
this.logger.warn(`Failed to vacuum table '${table}'. The DB will temporarily use more disk space: ${error}`);
|
||||
}
|
||||
this.logger.log(`Reindexed ${indexName}`);
|
||||
void this.vacuum({ table }).catch((error) => this.logger.warn(`Failed to vacuum ${table}: ${error}`));
|
||||
}
|
||||
|
||||
async vacuum({ analyze = false, table }: { analyze?: boolean; table?: keyof DB } = {}): Promise<void> {
|
||||
try {
|
||||
await sql`VACUUM ${sql.raw(analyze ? 'ANALYZE' : '')} ${sql.raw(table ?? '')}`.execute(this.db);
|
||||
} catch (error) {
|
||||
this.logger.warn(`Failed to vacuum ${table || 'database'}: ${error}`);
|
||||
this.logger.warn('If using Docker, consider increasing shm_size for the database.');
|
||||
}
|
||||
}
|
||||
|
||||
reindex(table: keyof DB, { concurrently = false } = {}): Promise<unknown> {
|
||||
return sql`REINDEX TABLE ${sql.raw(concurrently ? 'CONCURRENTLY' : '')} ${sql.raw(table)}`.execute(this.db);
|
||||
}
|
||||
|
||||
private async getDatabaseName(): Promise<string> {
|
||||
@@ -367,14 +376,14 @@ export class DatabaseRepository {
|
||||
return count;
|
||||
}
|
||||
|
||||
async runMigrations(): Promise<void> {
|
||||
async runMigrations(): Promise<number> {
|
||||
this.logger.log('Running migrations');
|
||||
|
||||
const migrator = this.createMigrator();
|
||||
|
||||
const { error, results } = await migrator.migrateToLatest();
|
||||
const { error, results = [] } = await migrator.migrateToLatest();
|
||||
|
||||
for (const result of results ?? []) {
|
||||
for (const result of results) {
|
||||
if (result.status === 'Success') {
|
||||
this.logger.log(`Migration "${result.migrationName}" succeeded`);
|
||||
} else if (result.status === 'Error') {
|
||||
@@ -399,6 +408,7 @@ export class DatabaseRepository {
|
||||
}
|
||||
|
||||
this.logger.log('Finished running migrations');
|
||||
return results.length;
|
||||
}
|
||||
|
||||
async migrateFilePaths(sourceFolder: string, targetFolder: string): Promise<void> {
|
||||
|
||||
@@ -727,15 +727,6 @@ export class PersonRepository {
|
||||
await this.db.updateTable('asset_face').set({ deletedAt: new Date() }).where('asset_face.id', '=', id).execute();
|
||||
}
|
||||
|
||||
async vacuum({ reindexVectors }: { reindexVectors: boolean }): Promise<void> {
|
||||
await sql`VACUUM ANALYZE asset_face, face_search, person`.execute(this.db);
|
||||
await sql`REINDEX TABLE asset_face`.execute(this.db);
|
||||
await sql`REINDEX TABLE person`.execute(this.db);
|
||||
if (reindexVectors) {
|
||||
await sql`REINDEX TABLE face_search`.execute(this.db);
|
||||
}
|
||||
}
|
||||
|
||||
@GenerateSql({ params: [[], []] })
|
||||
async updateVisibility(visible: AssetFace[], hidden: AssetFace[]): Promise<void> {
|
||||
if (visible.length === 0 && hidden.length === 0) {
|
||||
|
||||
@@ -111,28 +111,36 @@ export class DatabaseService extends BaseService {
|
||||
}
|
||||
}
|
||||
|
||||
const preparation = [];
|
||||
const { database } = this.configRepository.getEnv();
|
||||
if (!database.skipMigrations) {
|
||||
await this.databaseRepository.runMigrations();
|
||||
|
||||
this.logger.log('Checking for schema drift');
|
||||
const drift = await this.databaseRepository.getSchemaDrift();
|
||||
if (drift.items.length === 0) {
|
||||
this.logger.log('No schema drift detected');
|
||||
} else {
|
||||
this.logger.warn(`${ErrorMessages.SchemaDrift} or run \`immich-admin schema-check\``);
|
||||
for (const warning of drift.asHuman()) {
|
||||
this.logger.warn(` - ${warning}`);
|
||||
}
|
||||
const migrationCount = await this.databaseRepository.runMigrations();
|
||||
preparation.push(this.checkSchemaDrift());
|
||||
if (migrationCount > 0) {
|
||||
preparation.push(this.databaseRepository.vacuum({ analyze: true }));
|
||||
}
|
||||
}
|
||||
await Promise.all([
|
||||
preparation.push(
|
||||
this.databaseRepository.prewarm(VectorIndex.Clip),
|
||||
this.databaseRepository.prewarm(VectorIndex.Face),
|
||||
]);
|
||||
);
|
||||
await Promise.all(preparation);
|
||||
});
|
||||
}
|
||||
|
||||
private async checkSchemaDrift() {
|
||||
this.logger.log('Checking for schema drift');
|
||||
const drift = await this.databaseRepository.getSchemaDrift();
|
||||
if (drift.items.length === 0) {
|
||||
this.logger.log('No schema drift detected');
|
||||
} else {
|
||||
this.logger.warn(`${ErrorMessages.SchemaDrift} or run \`immich-admin schema-check\``);
|
||||
for (const warning of drift.asHuman()) {
|
||||
this.logger.warn(` - ${warning}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private async createExtension(extension: DatabaseExtension) {
|
||||
try {
|
||||
await this.databaseRepository.createExtension(extension);
|
||||
|
||||
@@ -626,7 +626,7 @@ describe(PersonService.name, () => {
|
||||
await sut.handleQueueDetectFaces({ force: false });
|
||||
|
||||
expect(mocks.assetJob.streamForDetectFacesJob).toHaveBeenCalledWith(false);
|
||||
expect(mocks.person.vacuum).not.toHaveBeenCalled();
|
||||
expect(mocks.database.vacuum).not.toHaveBeenCalled();
|
||||
expect(mocks.job.queueAll).toHaveBeenCalledWith([
|
||||
{
|
||||
name: JobName.AssetDetectFaces,
|
||||
@@ -648,7 +648,9 @@ describe(PersonService.name, () => {
|
||||
expect(mocks.person.deleteFaces).toHaveBeenCalledWith({ sourceType: SourceType.MachineLearning });
|
||||
expect(mocks.person.delete).toHaveBeenCalledWith([person.personGroupId], undefined);
|
||||
expect(mocks.person.deleteEmptyGroups).toHaveBeenCalledWith();
|
||||
expect(mocks.person.vacuum).toHaveBeenCalledWith({ reindexVectors: true });
|
||||
expect(mocks.database.vacuum).toHaveBeenCalledWith({ analyze: true, table: 'asset_face' });
|
||||
expect(mocks.database.vacuum).toHaveBeenCalledWith({ analyze: true, table: 'person' });
|
||||
expect(mocks.database.vacuum).toHaveBeenCalledWith({ analyze: true, table: 'face_search' });
|
||||
expect(mocks.storage.unlink).toHaveBeenCalledWith(person.thumbnailPath);
|
||||
expect(mocks.assetJob.streamForDetectFacesJob).toHaveBeenCalledWith(true);
|
||||
expect(mocks.job.queueAll).toHaveBeenCalledWith([
|
||||
@@ -667,7 +669,7 @@ describe(PersonService.name, () => {
|
||||
|
||||
expect(mocks.person.deleteGroups).not.toHaveBeenCalled();
|
||||
expect(mocks.person.deleteFaces).not.toHaveBeenCalled();
|
||||
expect(mocks.person.vacuum).not.toHaveBeenCalled();
|
||||
expect(mocks.database.vacuum).not.toHaveBeenCalled();
|
||||
expect(mocks.storage.unlink).not.toHaveBeenCalled();
|
||||
expect(mocks.assetJob.streamForDetectFacesJob).toHaveBeenCalledWith(undefined);
|
||||
expect(mocks.job.queueAll).toHaveBeenCalledWith([
|
||||
@@ -703,7 +705,9 @@ describe(PersonService.name, () => {
|
||||
expect(mocks.person.delete).toHaveBeenCalledWith([person.personGroupId], undefined);
|
||||
expect(mocks.person.deleteEmptyGroups).toHaveBeenCalledWith();
|
||||
expect(mocks.storage.unlink).toHaveBeenCalledWith(person.thumbnailPath);
|
||||
expect(mocks.person.vacuum).toHaveBeenCalledWith({ reindexVectors: true });
|
||||
expect(mocks.database.vacuum).toHaveBeenCalledWith({ analyze: true, table: 'asset_face' });
|
||||
expect(mocks.database.vacuum).toHaveBeenCalledWith({ analyze: true, table: 'person' });
|
||||
expect(mocks.database.vacuum).toHaveBeenCalledWith({ analyze: true, table: 'face_search' });
|
||||
});
|
||||
});
|
||||
|
||||
@@ -768,7 +772,7 @@ describe(PersonService.name, () => {
|
||||
expect(mocks.systemMetadata.set).toHaveBeenCalledWith(SystemMetadataKey.FacialRecognitionState, {
|
||||
lastRun: expect.any(String),
|
||||
});
|
||||
expect(mocks.person.vacuum).not.toHaveBeenCalled();
|
||||
expect(mocks.database.vacuum).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('should queue all assets', async () => {
|
||||
@@ -797,7 +801,8 @@ describe(PersonService.name, () => {
|
||||
expect(mocks.systemMetadata.set).toHaveBeenCalledWith(SystemMetadataKey.FacialRecognitionState, {
|
||||
lastRun: expect.any(String),
|
||||
});
|
||||
expect(mocks.person.vacuum).toHaveBeenCalledWith({ reindexVectors: false });
|
||||
expect(mocks.database.vacuum).toHaveBeenCalledWith({ analyze: true, table: 'asset_face' });
|
||||
expect(mocks.database.vacuum).toHaveBeenCalledWith({ analyze: true, table: 'person' });
|
||||
});
|
||||
|
||||
it('should run nightly if new face has been added since last run', async () => {
|
||||
@@ -834,7 +839,7 @@ describe(PersonService.name, () => {
|
||||
expect(mocks.systemMetadata.set).toHaveBeenCalledWith(SystemMetadataKey.FacialRecognitionState, {
|
||||
lastRun: expect.any(String),
|
||||
});
|
||||
expect(mocks.person.vacuum).not.toHaveBeenCalled();
|
||||
expect(mocks.database.vacuum).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('should skip nightly if no new face has been added since last run', async () => {
|
||||
@@ -852,7 +857,7 @@ describe(PersonService.name, () => {
|
||||
expect(mocks.person.getAllFaces).not.toHaveBeenCalled();
|
||||
expect(mocks.job.queueAll).not.toHaveBeenCalled();
|
||||
expect(mocks.systemMetadata.set).not.toHaveBeenCalled();
|
||||
expect(mocks.person.vacuum).not.toHaveBeenCalled();
|
||||
expect(mocks.database.vacuum).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('should delete existing people if forced', async () => {
|
||||
@@ -886,7 +891,8 @@ describe(PersonService.name, () => {
|
||||
expect(mocks.person.delete).toHaveBeenCalledWith([person.personGroupId], undefined);
|
||||
expect(mocks.person.deleteEmptyGroups).toHaveBeenCalledWith();
|
||||
expect(mocks.storage.unlink).toHaveBeenCalledWith(person.thumbnailPath);
|
||||
expect(mocks.person.vacuum).toHaveBeenCalledWith({ reindexVectors: false });
|
||||
expect(mocks.database.vacuum).toHaveBeenCalledWith({ analyze: true, table: 'asset_face' });
|
||||
expect(mocks.database.vacuum).toHaveBeenCalledWith({ analyze: true, table: 'person' });
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
@@ -35,6 +35,7 @@ import {
|
||||
} from 'src/enum.js';
|
||||
import { BoundingBox } from 'src/repositories/machine-learning.repository.js';
|
||||
import { PersonId, UpdateFacesData } from 'src/repositories/person.repository.js';
|
||||
import { DB } from 'src/schema/index.js';
|
||||
import { AssetFaceTable } from 'src/schema/tables/asset-face.table.js';
|
||||
import { FaceSearchTable } from 'src/schema/tables/face-search.table.js';
|
||||
import { PersonTable } from 'src/schema/tables/person.table.js';
|
||||
@@ -296,7 +297,7 @@ export class PersonService extends BaseService {
|
||||
if (force) {
|
||||
await this.personRepository.deleteFaces({ sourceType: SourceType.MachineLearning });
|
||||
await this.handlePersonCleanup();
|
||||
await this.personRepository.vacuum({ reindexVectors: true });
|
||||
await this.vacuum('asset_face', 'person', 'face_search');
|
||||
}
|
||||
|
||||
for await (const assets of batched(this.assetJobRepository.streamForDetectFacesJob(force))) {
|
||||
@@ -443,7 +444,7 @@ export class PersonService extends BaseService {
|
||||
if (force) {
|
||||
await this.personRepository.unassignFaces({ clusterGroupId, sourceType: SourceType.MachineLearning });
|
||||
await this.handlePersonCleanup();
|
||||
await this.personRepository.vacuum({ reindexVectors: false });
|
||||
await this.vacuum('asset_face', 'person');
|
||||
} else if (waiting) {
|
||||
this.logger.debug(
|
||||
`Skipping facial recognition queueing because ${waiting} job${waiting > 1 ? 's are' : ' is'} already queued`,
|
||||
@@ -738,4 +739,14 @@ export class PersonService extends BaseService {
|
||||
|
||||
return dto.force ? this.personRepository.deleteAssetFace(id) : this.personRepository.softDeleteAssetFaces(id);
|
||||
}
|
||||
|
||||
private vacuum(...tables: (keyof DB)[]): Promise<unknown> {
|
||||
return Promise.all(
|
||||
tables.map((table) =>
|
||||
this.databaseRepository
|
||||
.vacuum({ analyze: true, table })
|
||||
.then(() => this.databaseRepository.reindex(table, { concurrently: true })),
|
||||
),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -312,6 +312,7 @@ export const getMocks = () => {
|
||||
databaseMock.getPostgresVersion = vitest.fn().mockResolvedValue('14.10 (Debian 14.10-1.pgdg120+1)');
|
||||
databaseMock.getPostgresVersionRange = vitest.fn().mockReturnValue('>=14.0.0');
|
||||
databaseMock.createExtension = vitest.fn().mockResolvedValue(void 0);
|
||||
databaseMock.vacuum.mockResolvedValue(void 0);
|
||||
|
||||
const mocks: ServiceMocks = {
|
||||
access: newAccessRepositoryMock(),
|
||||
|
||||
Reference in New Issue
Block a user