refactor(sync): rely on the advisory lock only and make syncAll private

Each startSync request takes the Postgres advisory lock on its own
connection (its own session), so it already refuses concurrent calls on
the same instance: the in-process syncInProgress flag and its
isSyncInProgress getter were redundant and are removed. syncAll becomes
private: syncs start only through startSync.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
MastersGame
2026-11-08 10:21:27 +01:00
parent 05911f4230
commit 96fca40847
2 changed files with 11 additions and 45 deletions

View File

@@ -62,30 +62,17 @@ describe('HiboutikService', () => {
it('refuses a second synchronization while one is running, and unlocks at the end', async () => { it('refuses a second synchronization while one is running, and unlocks at the end', async () => {
let finishBrands: () => void = () => undefined; let finishBrands: () => void = () => undefined;
brandService.syncBrands.mockReturnValue(new Promise<void>((resolve) => (finishBrands = resolve))); brandService.syncBrands.mockReturnValue(new Promise<void>((resolve) => (finishBrands = resolve)));
// The lock is the only guard: the second request does not get it, even from the same instance
syncLock.tryAcquire.mockResolvedValueOnce(release).mockResolvedValueOnce(null);
await expect(service.startSync()).resolves.toBe(true); await expect(service.startSync()).resolves.toBe(true);
expect(service.isSyncInProgress).toBe(true);
await expect(service.startSync()).resolves.toBe(false); await expect(service.startSync()).resolves.toBe(false);
expect(syncLock.tryAcquire).toHaveBeenCalledTimes(1); expect(brandService.syncBrands).toHaveBeenCalledTimes(1);
expect(release).not.toHaveBeenCalled(); expect(release).not.toHaveBeenCalled();
finishBrands(); finishBrands();
await flush(); await flush();
expect(release).toHaveBeenCalledTimes(1); expect(release).toHaveBeenCalledTimes(1);
expect(service.isSyncInProgress).toBe(false);
});
it('refuses concurrent calls on the same instance while the lock is being requested', async () => {
let grantLock: (value: typeof release) => void = () => undefined;
syncLock.tryAcquire.mockReturnValue(new Promise((resolve) => (grantLock = resolve)));
const first = service.startSync();
await expect(service.startSync()).resolves.toBe(false);
grantLock(release);
await expect(first).resolves.toBe(true);
await flush();
expect(syncLock.tryAcquire).toHaveBeenCalledTimes(1);
}); });
it('starts nothing when another instance holds the lock', async () => { it('starts nothing when another instance holds the lock', async () => {
@@ -93,7 +80,6 @@ describe('HiboutikService', () => {
await expect(service.startSync()).resolves.toBe(false); await expect(service.startSync()).resolves.toBe(false);
expect(service.isSyncInProgress).toBe(false);
expect(brandService.syncBrands).not.toHaveBeenCalled(); expect(brandService.syncBrands).not.toHaveBeenCalled();
expect(release).not.toHaveBeenCalled(); expect(release).not.toHaveBeenCalled();
}); });
@@ -102,19 +88,18 @@ describe('HiboutikService', () => {
syncLock.tryAcquire.mockRejectedValueOnce(new Error('database unreachable')); syncLock.tryAcquire.mockRejectedValueOnce(new Error('database unreachable'));
await expect(service.startSync()).rejects.toThrow('database unreachable'); await expect(service.startSync()).rejects.toThrow('database unreachable');
expect(service.isSyncInProgress).toBe(false); expect(brandService.syncBrands).not.toHaveBeenCalled();
await expect(service.startSync()).resolves.toBe(true); await expect(service.startSync()).resolves.toBe(true);
await flush(); await flush();
}); });
it('releases the lock even when the synchronization crashes', async () => { it('releases the lock even when the synchronization crashes', async () => {
jest.spyOn(service, 'syncAll').mockRejectedValue(new Error('crash')); jest.spyOn(service as never, 'syncAll').mockRejectedValue(new Error('crash'));
await expect(service.startSync()).resolves.toBe(true); await expect(service.startSync()).resolves.toBe(true);
await flush(); await flush();
expect(release).toHaveBeenCalledTimes(1); expect(release).toHaveBeenCalledTimes(1);
expect(service.isSyncInProgress).toBe(false);
}); });
}); });

View File

@@ -10,8 +10,6 @@ import { ReleaseLock, SyncLockDatabaseRepository } from '../../database/syncLock
@Injectable() @Injectable()
export class HiboutikService { export class HiboutikService {
private readonly logger = new Logger('HiboutikService'); private readonly logger = new Logger('HiboutikService');
/** Cheap in-process guard, checked before the database lock (which is the one that holds across instances). */
private syncInProgress = false;
constructor( constructor(
private readonly brandHiboutikService: BrandHiboutikService, private readonly brandHiboutikService: BrandHiboutikService,
@@ -22,32 +20,16 @@ export class HiboutikService {
private readonly syncLock: SyncLockDatabaseRepository, private readonly syncLock: SyncLockDatabaseRepository,
) {} ) {}
get isSyncInProgress(): boolean {
return this.syncInProgress;
}
/** /**
* Starts a full synchronization in the background, if no other one is running on this instance or any other * Starts a full synchronization in the background, if no other one is running on this instance or any other.
* (Postgres advisory lock, held until the synchronization ends). * The Postgres advisory lock is the only guard: each request takes it on its own connection (its own session),
* so it also refuses concurrent calls on this very instance. It is held until the synchronization ends.
* @returns false if a synchronization is already running (nothing is started). * @returns false if a synchronization is already running (nothing is started).
* @throws if the lock cannot be requested (e.g. database unreachable). * @throws if the lock cannot be requested (e.g. database unreachable).
*/ */
async startSync(): Promise<boolean> { async startSync(): Promise<boolean> {
if (this.syncInProgress) return false; const release = await this.syncLock.tryAcquire();
// Set before the first await so that concurrent calls on this instance are refused while the lock is requested if (!release) return false;
this.syncInProgress = true;
let release: ReleaseLock | null;
try {
release = await this.syncLock.tryAcquire();
} catch (error) {
this.syncInProgress = false;
throw error;
}
if (!release) {
this.syncInProgress = false;
return false;
}
void this.runLocked(release); void this.runLocked(release);
return true; return true;
@@ -61,12 +43,11 @@ export class HiboutikService {
this.logger.error('Global Hiboutik synchronization crashed', (error as Error)?.stack); this.logger.error('Global Hiboutik synchronization crashed', (error as Error)?.stack);
} finally { } finally {
await release(); await release();
this.syncInProgress = false;
} }
} }
/** Runs every sync step in order. A failing step is logged and does not stop the next ones. */ /** Runs every sync step in order. A failing step is logged and does not stop the next ones. */
async syncAll() { private async syncAll() {
this.logger.log('Starting global Hiboutik synchronization'); this.logger.log('Starting global Hiboutik synchronization');
// Order matters: products connect to brands, categories and suppliers by hiboutikId. // Order matters: products connect to brands, categories and suppliers by hiboutikId.
const steps: [string, () => Promise<void>][] = [ const steps: [string, () => Promise<void>][] = [