From 6dd4c2e64577f2140d72b9d993bc31045e1d1734 Mon Sep 17 00:00:00 2001 From: wheval Date: Sun, 26 Apr 2026 13:37:05 +0100 Subject: [PATCH 1/2] fix(admin): add configurable TTL for indexer replay lock Add a background job lock with configurable TTL and conflict handling for indexer replay requests. Log lock acquisition and lock-expiration reclaim events to improve debugging of stuck jobs. Closes accesslayerorg/accesslayer-server#187 Made-with: Cursor --- .env.example | 1 + src/config.ts | 1 + src/modules/admin/admin.controllers.ts | 39 +++++++++ src/utils/background-job-lock.utils.ts | 76 +++++++++++++++++ .../test/background-job-lock.utils.test.ts | 84 +++++++++++++++++++ 5 files changed, 201 insertions(+) create mode 100644 src/utils/background-job-lock.utils.ts create mode 100644 src/utils/test/background-job-lock.utils.test.ts diff --git a/.env.example b/.env.example index 048949f..494c220 100644 --- a/.env.example +++ b/.env.example @@ -25,3 +25,4 @@ API_VERSION=1.0.0 ENABLE_API_VERSION_HEADER=true ENABLE_RESPONSE_TIMING=true ENABLE_REQUEST_LOGGING=true +BACKGROUND_JOB_LOCK_TTL_MS=300000 diff --git a/src/config.ts b/src/config.ts index 9876cd4..50480f9 100644 --- a/src/config.ts +++ b/src/config.ts @@ -50,6 +50,7 @@ export const envSchema = z.object({ ENABLE_API_VERSION_HEADER: z.coerce.boolean().default(true), ENABLE_REQUEST_LOGGING: z.coerce.boolean().default(true), INDEXER_JITTER_FACTOR: z.coerce.number().min(0).max(1).default(0.1), + BACKGROUND_JOB_LOCK_TTL_MS: z.coerce.number().int().positive().default(300000), }); export const envConfig = envSchema.parse(process.env); diff --git a/src/modules/admin/admin.controllers.ts b/src/modules/admin/admin.controllers.ts index c1e1086..f2e9d6d 100644 --- a/src/modules/admin/admin.controllers.ts +++ b/src/modules/admin/admin.controllers.ts @@ -5,6 +5,9 @@ import { emitAuditEvent } from '../../utils/audit.utils'; import { AdminRequest } from '../../middlewares/admin-guard.middleware'; import { Response } from 'express'; import { z } from 'zod'; +import { acquireJobLock } from '../../utils/background-job-lock.utils'; +import { logger } from '../../utils/logger.utils'; +import { ErrorCode } from '../../constants/error.constants'; const UpdateCreatorMetadataSchema = z.object({ isVerified: z.boolean().optional(), @@ -86,6 +89,8 @@ export const httpReplayIndexerEvents: AsyncController = async (req: AdminRequest try { const { startLedger } = req.body as { startLedger?: number }; const adminId = req.adminId; + const lockName = 'indexer-replay'; + const lockOwner = adminId || 'unknown'; if (typeof startLedger !== 'number' || startLedger < 1) { return sendValidationError(res, 'Invalid request body', [ @@ -93,13 +98,47 @@ export const httpReplayIndexerEvents: AsyncController = async (req: AdminRequest ]); } + const lock = acquireJobLock({ + name: lockName, + owner: lockOwner, + }); + + if (!lock.acquired) { + return res.status(409).json({ + success: false, + error: { + code: ErrorCode.CONFLICT, + message: 'Indexer replay job is already running', + details: [ + { + field: 'indexerReplayLock', + message: `Lock is held by ${lock.holder || 'another worker'} until ${lock.expiresAt || 'unknown time'}`, + }, + ], + }, + }); + } + const replayInitiated = { type: 'INDEXER_REPLAY_INITIATED', startLedger, initiatedBy: adminId, + lock: { + name: lockName, + expiresAt: lock.expiresAt, + }, timestamp: new Date().toISOString(), }; + logger.info( + { + lockName, + lockOwner, + lockExpiresAt: lock.expiresAt, + }, + 'Acquired background job lock for indexer replay' + ); + await emitAuditEvent({ actor: adminId || 'unknown', action: 'replay_indexer_events', diff --git a/src/utils/background-job-lock.utils.ts b/src/utils/background-job-lock.utils.ts new file mode 100644 index 0000000..c940016 --- /dev/null +++ b/src/utils/background-job-lock.utils.ts @@ -0,0 +1,76 @@ +import { envConfig } from '../config'; +import { logger } from './logger.utils'; + +interface LockEntry { + owner: string; + expiresAtMs: number; +} + +const locks = new Map(); + +export interface AcquireJobLockParams { + name: string; + owner: string; + ttlMs?: number; +} + +export interface AcquireJobLockResult { + acquired: boolean; + expiresAt?: string; + holder?: string; +} + +export function acquireJobLock({ + name, + owner, + ttlMs = envConfig.BACKGROUND_JOB_LOCK_TTL_MS, +}: AcquireJobLockParams): AcquireJobLockResult { + const nowMs = Date.now(); + const existing = locks.get(name); + + if (existing && existing.expiresAtMs <= nowMs) { + logger.warn( + { + lockName: name, + previousOwner: existing.owner, + expiredAt: new Date(existing.expiresAtMs).toISOString(), + now: new Date(nowMs).toISOString(), + }, + 'Background job lock expired; reclaiming lock' + ); + locks.delete(name); + } + + if (locks.has(name)) { + const current = locks.get(name)!; + return { + acquired: false, + expiresAt: new Date(current.expiresAtMs).toISOString(), + holder: current.owner, + }; + } + + const expiresAtMs = nowMs + ttlMs; + locks.set(name, { owner, expiresAtMs }); + + return { + acquired: true, + expiresAt: new Date(expiresAtMs).toISOString(), + }; +} + +export function releaseJobLock(name: string, owner?: string): boolean { + const current = locks.get(name); + if (!current) return false; + + if (owner && current.owner !== owner) { + return false; + } + + locks.delete(name); + return true; +} + +export function resetJobLocks(): void { + locks.clear(); +} diff --git a/src/utils/test/background-job-lock.utils.test.ts b/src/utils/test/background-job-lock.utils.test.ts new file mode 100644 index 0000000..0953b77 --- /dev/null +++ b/src/utils/test/background-job-lock.utils.test.ts @@ -0,0 +1,84 @@ +import { + acquireJobLock, + releaseJobLock, + resetJobLocks, +} from '../background-job-lock.utils'; +import { logger } from '../logger.utils'; + +jest.mock('../../config', () => ({ + envConfig: { + BACKGROUND_JOB_LOCK_TTL_MS: 1000, + }, +})); + +jest.mock('../logger.utils', () => ({ + logger: { + warn: jest.fn(), + }, +})); + +describe('background-job-lock.utils', () => { + beforeEach(() => { + jest.useFakeTimers(); + jest.setSystemTime(new Date('2026-01-01T00:00:00.000Z')); + resetJobLocks(); + jest.clearAllMocks(); + }); + + afterEach(() => { + jest.useRealTimers(); + resetJobLocks(); + }); + + it('acquires lock with default TTL from config', () => { + const result = acquireJobLock({ name: 'indexer', owner: 'worker-a' }); + + expect(result.acquired).toBe(true); + expect(result.expiresAt).toBe('2026-01-01T00:00:01.000Z'); + }); + + it('supports per-call TTL override', () => { + const result = acquireJobLock({ + name: 'indexer', + owner: 'worker-a', + ttlMs: 5000, + }); + + expect(result.acquired).toBe(true); + expect(result.expiresAt).toBe('2026-01-01T00:00:05.000Z'); + }); + + it('rejects acquisition while active lock exists', () => { + acquireJobLock({ name: 'indexer', owner: 'worker-a' }); + const result = acquireJobLock({ name: 'indexer', owner: 'worker-b' }); + + expect(result).toEqual({ + acquired: false, + holder: 'worker-a', + expiresAt: '2026-01-01T00:00:01.000Z', + }); + }); + + it('logs expiration and allows lock reclaim after TTL', () => { + acquireJobLock({ name: 'indexer', owner: 'worker-a' }); + jest.advanceTimersByTime(1001); + + const result = acquireJobLock({ name: 'indexer', owner: 'worker-b' }); + + expect(result.acquired).toBe(true); + expect(logger.warn).toHaveBeenCalledWith( + expect.objectContaining({ + lockName: 'indexer', + previousOwner: 'worker-a', + }), + 'Background job lock expired; reclaiming lock' + ); + }); + + it('releases lock only when owner matches', () => { + acquireJobLock({ name: 'indexer', owner: 'worker-a' }); + + expect(releaseJobLock('indexer', 'worker-b')).toBe(false); + expect(releaseJobLock('indexer', 'worker-a')).toBe(true); + }); +}); From 10781fd46faf576c1f6e08328ca5d605a1d0e2ed Mon Sep 17 00:00:00 2001 From: wheval Date: Mon, 27 Apr 2026 22:43:56 +0100 Subject: [PATCH 2/2] test(admin): mock job lock in replay controller tests Made-with: Cursor --- src/modules/admin/admin.controllers.test.ts | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/src/modules/admin/admin.controllers.test.ts b/src/modules/admin/admin.controllers.test.ts index 6aaf68c..f22af3a 100644 --- a/src/modules/admin/admin.controllers.test.ts +++ b/src/modules/admin/admin.controllers.test.ts @@ -3,6 +3,13 @@ import { emitAuditEvent } from '../../utils/audit.utils'; import { AdminRequest } from '../../middlewares/admin-guard.middleware'; import { Response } from 'express'; +jest.mock('../../utils/background-job-lock.utils', () => ({ + acquireJobLock: jest.fn(() => ({ + acquired: true, + expiresAt: '2026-01-01T00:00:00.000Z', + })), +})); + jest.mock('../../utils/prisma.utils', () => ({ prisma: { creatorProfile: {