mirror of
https://github.com/PR0M3TH3AN/SeedPass.git
synced 2026-11-01 10:22:03 +00:00
315 lines
9.0 KiB
JavaScript
315 lines
9.0 KiB
JavaScript
import { getMemoryPruneMode, isMemoryEnabled, isMemoryIngestEnabled } from './feature-flags.js';
|
|
import { MINUTE_MS, HOUR_MS, DAY_MS, WEEK_MS } from './constants/time.js';
|
|
|
|
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
|
|
* @returns {{ stop: () => void }}
|
|
*/
|
|
export function startScheduler(task, options) {
|
|
const handle = setInterval(() => {
|
|
void task();
|
|
}, options.intervalMs);
|
|
|
|
return {
|
|
stop() {
|
|
clearInterval(handle);
|
|
},
|
|
};
|
|
}
|
|
|
|
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;
|
|
}
|
|
}
|
|
}
|
|
|
|
function shouldRunJob(jobName, options) {
|
|
const env = options.env ?? process.env;
|
|
if (!isMemoryEnabled(env)) return false;
|
|
|
|
if (jobName === 'ingestRecentRuntimeEvents') {
|
|
return isMemoryIngestEnabled(options.ingestAgentId, env);
|
|
}
|
|
|
|
if (jobName === 'pruningCycle') {
|
|
return getMemoryPruneMode(env) !== 'off';
|
|
}
|
|
|
|
return true;
|
|
}
|
|
|
|
/**
|
|
* @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 = () => {
|
|
if (!shouldRunJob(jobName, options)) {
|
|
options.emitMetric?.('memory_scheduler_job', {
|
|
job: jobName,
|
|
status: 'skipped_flag_disabled',
|
|
durationMs: 0,
|
|
itemCount: 0,
|
|
failures: 0,
|
|
retries: 0,
|
|
});
|
|
return;
|
|
}
|
|
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);
|
|
if (!shouldRunJob(jobName, options)) {
|
|
options.emitMetric?.('memory_scheduler_job', {
|
|
job: jobName,
|
|
status: 'skipped_flag_disabled',
|
|
durationMs: 0,
|
|
itemCount: 0,
|
|
failures: 0,
|
|
retries: 0,
|
|
});
|
|
scheduleNext();
|
|
return;
|
|
}
|
|
await runJob(jobName, definition, handler);
|
|
scheduleNext();
|
|
}, delayMs);
|
|
timeouts.add(timeout);
|
|
};
|
|
|
|
if (options.runImmediately !== false) {
|
|
if (!shouldRunJob(jobName, options)) {
|
|
options.emitMetric?.('memory_scheduler_job', {
|
|
job: jobName,
|
|
status: 'skipped_flag_disabled',
|
|
durationMs: 0,
|
|
itemCount: 0,
|
|
failures: 0,
|
|
retries: 0,
|
|
});
|
|
scheduleNext();
|
|
return;
|
|
}
|
|
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();
|
|
},
|
|
};
|
|
}
|