feat(memory): add distributed-safe maintenance scheduler jobs

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> | 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<unknown>} 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> | unknown,
+ *   lockProvider: { withLock: (key: string, task: () => Promise<unknown>) => Promise<{ acquired: boolean, result: unknown }> },
+ *   maxRetries: number,
+ *   retryDelayMs: number,
+ *   emitMetric?: (metricName: string, payload: Record<string, unknown>) => 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> | unknown,
+ *     consolidateObservations: () => Promise<unknown> | unknown,
+ *     pruningCycle: () => Promise<unknown> | unknown,
+ *     deepMergeArchivalMaintenance: () => Promise<unknown> | unknown,
+ *   },
+ *   lockProvider?: { withLock: (key: string, task: () => Promise<unknown>) => Promise<{ acquired: boolean, result: unknown }> },
+ *   emitMetric?: (metricName: string, payload: Record<string, unknown>) => 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();
+    },
+  };
+}

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');
+});