From 3ddd05017a5ca1b6c418fb307643da3e97b8f843 Mon Sep 17 00:00:00 2001 From: thePR0M3TH3AN <53631862+PR0M3TH3AN@users.noreply.github.com> Date: Sat, 14 Feb 2026 15:05:59 -0500 Subject: [PATCH] feat(memory): add distributed-safe maintenance scheduler jobs --- src/services/memory/scheduler.js | 250 +++++++++++++++++++++++++++++++ test/memory-scheduler.test.mjs | 92 ++++++++++++ 2 files changed, 342 insertions(+) create mode 100644 test/memory-scheduler.test.mjs diff --git a/src/services/memory/scheduler.js b/src/services/memory/scheduler.js index a494565..562a835 100644 --- a/src/services/memory/scheduler.js +++ b/src/services/memory/scheduler.js @@ -1,3 +1,28 @@ +const MINUTE_MS = 60_000; +const HOUR_MS = 60 * MINUTE_MS; +const DAY_MS = 24 * HOUR_MS; +const WEEK_MS = 7 * DAY_MS; + +const JOBS = { + ingestRecentRuntimeEvents: { + lockKey: 'memory:job:ingest-recent-runtime-events', + minIntervalMs: MINUTE_MS, + maxIntervalMs: 5 * MINUTE_MS, + }, + consolidateObservations: { + lockKey: 'memory:job:consolidate-observations', + intervalMs: HOUR_MS, + }, + pruningCycle: { + lockKey: 'memory:job:pruning-cycle', + intervalMs: DAY_MS, + }, + deepMergeArchivalMaintenance: { + lockKey: 'memory:job:deep-merge-archival-maintenance', + intervalMs: WEEK_MS, + }, +}; + /** * @param {() => Promise | unknown} task * @param {{ intervalMs: number }} options @@ -14,3 +39,228 @@ export function startScheduler(task, options) { }, }; } + +function normalizeItemCount(result) { + if (typeof result === 'number' && Number.isFinite(result)) { + return Math.max(0, Math.floor(result)); + } + + if (result && typeof result === 'object') { + const count = result.itemCount ?? result.count ?? result.processed ?? result.rows; + if (typeof count === 'number' && Number.isFinite(count)) { + return Math.max(0, Math.floor(count)); + } + } + + return 0; +} + +/** + * @param {string} key + */ +function hashLockKey(key) { + let hash = 0; + for (let i = 0; i < key.length; i += 1) { + hash = ((hash * 31) + key.charCodeAt(i)) | 0; + } + return hash; +} + +/** + * @param {{ query: (sql: string, params?: unknown[]) => Promise<{ rows?: any[] }> }} db + * @param {{ namespace?: string }} [options] + */ +export function createDbAdvisoryLockProvider(db, options = {}) { + const namespace = options.namespace ?? 'torch-memory-jobs'; + + return { + /** + * @param {string} key + * @param {() => Promise} task + */ + async withLock(key, task) { + const lockId = `${namespace}:${key}`; + const keyA = hashLockKey(namespace); + const keyB = hashLockKey(lockId); + const acquiredResult = await db.query('SELECT pg_try_advisory_lock($1, $2) AS locked', [keyA, keyB]); + const acquired = Boolean(acquiredResult?.rows?.[0]?.locked); + if (!acquired) { + return { acquired: false, result: null }; + } + + try { + const result = await task(); + return { acquired: true, result }; + } finally { + await db.query('SELECT pg_advisory_unlock($1, $2)', [keyA, keyB]); + } + }, + }; +} + +function createInMemoryLockProvider() { + const locks = new Set(); + return { + async withLock(key, task) { + if (locks.has(key)) { + return { acquired: false, result: null }; + } + + locks.add(key); + try { + const result = await task(); + return { acquired: true, result }; + } finally { + locks.delete(key); + } + }, + }; +} + +/** + * @param {{ + * jobName: string, + * lockKey: string, + * handler: () => Promise | unknown, + * lockProvider: { withLock: (key: string, task: () => Promise) => Promise<{ acquired: boolean, result: unknown }> }, + * maxRetries: number, + * retryDelayMs: number, + * emitMetric?: (metricName: string, payload: Record) => void, + * }} options + */ +async function runScheduledJob(options) { + const startedAt = Date.now(); + let failures = 0; + let retries = 0; + + for (let attempt = 0; attempt <= options.maxRetries; attempt += 1) { + try { + const lockRun = await options.lockProvider.withLock(options.lockKey, async () => options.handler()); + + if (!lockRun.acquired) { + options.emitMetric?.('memory_scheduler_job', { + job: options.jobName, + status: 'skipped_lock_unavailable', + durationMs: Date.now() - startedAt, + itemCount: 0, + failures, + retries, + }); + return; + } + + options.emitMetric?.('memory_scheduler_job', { + job: options.jobName, + status: 'success', + durationMs: Date.now() - startedAt, + itemCount: normalizeItemCount(lockRun.result), + failures, + retries, + }); + return; + } catch (error) { + failures += 1; + if (attempt < options.maxRetries) { + retries += 1; + await new Promise((resolve) => { + setTimeout(resolve, options.retryDelayMs); + }); + continue; + } + + options.emitMetric?.('memory_scheduler_job', { + job: options.jobName, + status: 'failed', + durationMs: Date.now() - startedAt, + itemCount: 0, + failures, + retries, + error: error instanceof Error ? error.message : String(error), + }); + return; + } + } +} + +/** + * @param {{ + * handlers: { + * ingestRecentRuntimeEvents: () => Promise | unknown, + * consolidateObservations: () => Promise | unknown, + * pruningCycle: () => Promise | unknown, + * deepMergeArchivalMaintenance: () => Promise | unknown, + * }, + * lockProvider?: { withLock: (key: string, task: () => Promise) => Promise<{ acquired: boolean, result: unknown }> }, + * emitMetric?: (metricName: string, payload: Record) => void, + * maxRetries?: number, + * retryDelayMs?: number, + * runImmediately?: boolean, + * random?: () => number, + * }} options + */ +export function startMemoryMaintenanceScheduler(options) { + const lockProvider = options.lockProvider ?? createInMemoryLockProvider(); + const maxRetries = Number.isFinite(options.maxRetries) ? Math.max(0, Math.floor(options.maxRetries)) : 2; + const retryDelayMs = Number.isFinite(options.retryDelayMs) ? Math.max(0, Math.floor(options.retryDelayMs)) : 1_000; + const random = typeof options.random === 'function' ? options.random : Math.random; + const handles = []; + const timeouts = new Set(); + + const runJob = async (jobName, definition, handler) => runScheduledJob({ + jobName, + lockKey: definition.lockKey, + handler, + lockProvider, + maxRetries, + retryDelayMs, + emitMetric: options.emitMetric, + }); + + const startFixedIntervalJob = (jobName, definition, handler) => { + const run = () => { + void runJob(jobName, definition, handler); + }; + + if (options.runImmediately !== false) run(); + const intervalHandle = setInterval(run, definition.intervalMs); + handles.push(() => clearInterval(intervalHandle)); + }; + + const startRandomIntervalJob = (jobName, definition, handler) => { + const scheduleNext = () => { + const jitterRatio = Math.min(1, Math.max(0, random())); + const delayMs = definition.minIntervalMs + Math.floor((definition.maxIntervalMs - definition.minIntervalMs) * jitterRatio); + const timeout = setTimeout(async () => { + timeouts.delete(timeout); + await runJob(jobName, definition, handler); + scheduleNext(); + }, delayMs); + timeouts.add(timeout); + }; + + if (options.runImmediately !== false) { + void runJob(jobName, definition, handler).then(() => { + scheduleNext(); + }); + } else { + scheduleNext(); + } + }; + + startRandomIntervalJob('ingestRecentRuntimeEvents', JOBS.ingestRecentRuntimeEvents, options.handlers.ingestRecentRuntimeEvents); + startFixedIntervalJob('consolidateObservations', JOBS.consolidateObservations, options.handlers.consolidateObservations); + startFixedIntervalJob('pruningCycle', JOBS.pruningCycle, options.handlers.pruningCycle); + startFixedIntervalJob('deepMergeArchivalMaintenance', JOBS.deepMergeArchivalMaintenance, options.handlers.deepMergeArchivalMaintenance); + + return { + stop() { + for (const stopHandle of handles) { + stopHandle(); + } + for (const timeout of timeouts) { + clearTimeout(timeout); + } + timeouts.clear(); + }, + }; +} diff --git a/test/memory-scheduler.test.mjs b/test/memory-scheduler.test.mjs new file mode 100644 index 0000000..5770741 --- /dev/null +++ b/test/memory-scheduler.test.mjs @@ -0,0 +1,92 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; + +import { + createDbAdvisoryLockProvider, + startMemoryMaintenanceScheduler, +} from '../src/services/memory/scheduler.js'; + +const wait = (ms = 0) => new Promise((resolve) => setTimeout(resolve, ms)); + +test('createDbAdvisoryLockProvider acquires and releases postgres advisory locks', async () => { + const calls = []; + const db = { + async query(sql, params) { + calls.push({ sql, params }); + if (sql.includes('pg_try_advisory_lock')) { + return { rows: [{ locked: true }] }; + } + return { rows: [] }; + }, + }; + + const lockProvider = createDbAdvisoryLockProvider(db, { namespace: 'tests' }); + const result = await lockProvider.withLock('job', async () => 42); + + assert.equal(result.acquired, true); + assert.equal(result.result, 42); + assert.equal(calls.length, 2); + assert.match(calls[0].sql, /pg_try_advisory_lock/); + assert.match(calls[1].sql, /pg_advisory_unlock/); + assert.deepEqual(calls[0].params, calls[1].params); +}); + +test('startMemoryMaintenanceScheduler runs all jobs and emits metrics including retries', async () => { + const metrics = []; + let ingestAttempts = 0; + + const scheduler = startMemoryMaintenanceScheduler({ + runImmediately: true, + maxRetries: 2, + retryDelayMs: 1, + random: () => 0, + emitMetric(name, payload) { + metrics.push({ name, payload }); + }, + lockProvider: { + async withLock(key, task) { + if (key.includes('consolidate-observations')) { + return { acquired: false, result: null }; + } + return { acquired: true, result: await task() }; + }, + }, + handlers: { + async ingestRecentRuntimeEvents() { + ingestAttempts += 1; + if (ingestAttempts === 1) { + throw new Error('transient'); + } + return { itemCount: 3 }; + }, + async consolidateObservations() { + return { itemCount: 11 }; + }, + async pruningCycle() { + return { itemCount: 2 }; + }, + async deepMergeArchivalMaintenance() { + return { itemCount: 1 }; + }, + }, + }); + + await wait(25); + scheduler.stop(); + + const byJob = metrics.reduce((acc, entry) => { + const key = entry.payload.job; + acc[key] = acc[key] || []; + acc[key].push(entry); + return acc; + }, {}); + + assert.equal(byJob.ingestRecentRuntimeEvents.length, 1); + assert.equal(byJob.ingestRecentRuntimeEvents[0].payload.status, 'success'); + assert.equal(byJob.ingestRecentRuntimeEvents[0].payload.itemCount, 3); + assert.equal(byJob.ingestRecentRuntimeEvents[0].payload.retries, 1); + + assert.equal(byJob.consolidateObservations[0].payload.status, 'skipped_lock_unavailable'); + assert.equal(byJob.pruningCycle[0].payload.status, 'success'); + assert.equal(byJob.deepMergeArchivalMaintenance[0].payload.status, 'success'); +});