Skip to content
Open
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion packages/daemon/src/app.ts
Original file line number Diff line number Diff line change
Expand Up @@ -921,7 +921,7 @@ export async function createDaemonApp(options: CreateDaemonAppOptions): Promise<
SKILL_VALIDATE,
createSkillValidateHandler(skillsManager, db.appMcpServers)
);
jobProcessor.register(JOB_QUEUE_CLEANUP, createCleanupHandler(jobQueue));
jobProcessor.register(JOB_QUEUE_CLEANUP, createCleanupHandler(jobQueue, db.getDatabase()));
jobProcessor.register(
MEMORY_CONSOLIDATION,
createMemoryConsolidationHandler(db.agentMemory, jobQueue)
Expand Down
37 changes: 34 additions & 3 deletions packages/daemon/src/lib/job-handlers/cleanup.handler.ts
Original file line number Diff line number Diff line change
@@ -1,13 +1,44 @@
import type { Database as BunDatabase } from 'bun:sqlite';
import type { Job, JobQueueRepository } from '../../storage/repositories/job-queue-repository';
import { JOB_QUEUE_CLEANUP } from '../job-queue-constants';
import { Logger } from '../logger';
import {
loadRetentionConfig,
type RetentionConfig,
type RetentionStats,
runRetention,
} from './retention';

const DEFAULT_MAX_AGE_MS = 7 * 24 * 60 * 60 * 1000; // 7 days
const NEXT_RUN_DELAY_MS = 24 * 60 * 60 * 1000; // 24 hours

export function createCleanupHandler(jobQueue: JobQueueRepository) {
return async (_job: Job): Promise<{ deletedJobs: number; nextRunAt: number }> => {
const logger = new Logger('Cleanup');

/**
* Build the daily cleanup job handler.
*
* Two responsibilities, both self-perpetuating on a 24h cadence:
* 1. Reap terminal job_queue rows older than 7 days (pre-existing).
* 2. Run retention sweeps (events / audit / goal events) + reclaim freed pages
* via incremental_vacuum. Retention deletion is OFF by default; see
* `loadRetentionConfig`. `retentionConfig` is optional and mainly for tests —
* production reads env vars per run so operators can tune TTLs live.
*/
export function createCleanupHandler(
jobQueue: JobQueueRepository,
db: BunDatabase,
retentionConfig?: RetentionConfig
) {
return async (
_job: Job
): Promise<{ deletedJobs: number; retention: RetentionStats; nextRunAt: number }> => {
const deletedJobs = jobQueue.cleanup(Date.now() - DEFAULT_MAX_AGE_MS);

const retention = runRetention(db, retentionConfig ?? loadRetentionConfig());
logger.info(
`retention sweep: external=${retention.externalEvents} deliveries=${retention.deliveries} github=${retention.githubEvents} mcpAudit=${retention.mcpAudit} goalEvents=${retention.goalEvents} vacuumedPages=${retention.vacuumedPages}`
);

const nextRunAt = Date.now() + NEXT_RUN_DELAY_MS;

// Self-schedule: only enqueue next cleanup if none is already pending
Expand All @@ -16,6 +47,6 @@ export function createCleanupHandler(jobQueue: JobQueueRepository) {
jobQueue.enqueue({ queue: JOB_QUEUE_CLEANUP, payload: {}, runAt: nextRunAt });
}

return { deletedJobs, nextRunAt };
return { deletedJobs, retention, nextRunAt };
};
}
233 changes: 233 additions & 0 deletions packages/daemon/src/lib/job-handlers/retention.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,233 @@
/**
* Retention sweeps for append-heavy / event-log tables.
*
* The daemon DB grows monotonically: external/github events, the MCP audit log
* and space goal events are written continuously and never reaped. Over time the
* file outgrows the OS page cache and scans hit cold disk. This module prunes
* rows that have reached a terminal state and are older than a configurable TTL,
* then reclaims the freed pages via `incremental_vacuum`.
*
* Policy:
* - Deletion is OFF by default (`HYPERNEO_RETENTION_ENABLED=1` to activate) and
* each TTL is independently configurable. Default windows only matter once
* enabled — they are intentionally conservative.
* - Only terminal-state event rows are pruned; in-flight states (published /
* routed / received / pending) are always kept so an active pipeline never
* loses work.
* - `incremental_vacuum(500)` runs every cycle regardless of `enabled`: it is
* space maintenance, not deletion, and reclaims pages freed by any delete
* (including the pre-existing 7-day job_queue and worktree reapers). It is a
* no-op unless the DB is in incremental-vacuum mode (see migration 169 /
* DatabaseCore's fresh-DB pragma).
*/

import type { Database as BunDatabase } from 'bun:sqlite';

const DAY_MS = 24 * 60 * 60 * 1000;

export interface RetentionConfig {
/** Master switch for deletion sweeps. Vacuum still runs when disabled. */
enabled: boolean;
/** TTL (days) for terminal external + github events and their deliveries. */
eventsDays: number;
/** TTL (days) for mcp_audit_log rows. */
mcpAuditDays: number;
/** TTL (days) for space_goal_events rows. */
goalEventsDays: number;
/** Max pages reclaimed per incremental_vacuum. 0 disables vacuum. */
vacuumPages: number;
}

export interface RetentionStats {
externalEvents: number;
deliveries: number;
githubEvents: number;
mcpAudit: number;
goalEvents: number;
vacuumedPages: number;
}

function emptyStats(): RetentionStats {
return {
externalEvents: 0,
deliveries: 0,
githubEvents: 0,
mcpAudit: 0,
goalEvents: 0,
vacuumedPages: 0,
};
}

function readEnv(name: string): string | undefined {
return process.env[`HYPERNEO_RETENTION_${name}`];
}

function envBool(value: string | undefined): boolean {
return value === '1' || value === 'true';
}

function envInt(value: string | undefined, fallback: number): number {
if (value === undefined || value === '') return fallback;
const parsed = Number(value);
if (!Number.isFinite(parsed) || parsed < 0) return fallback;
return Math.floor(parsed);
}

/**
* Read retention configuration from `HYPERNEO_RETENTION_*` env vars.
*
* Read per-run (not cached at startup) so operators can adjust TTLs without a
* daemon restart. Defaults are conservative and only take effect once enabled.
*/
export function loadRetentionConfig(): RetentionConfig {
return {
enabled: envBool(readEnv('ENABLED')),
eventsDays: envInt(readEnv('EVENTS_DAYS'), 14),
mcpAuditDays: envInt(readEnv('MCP_AUDIT_DAYS'), 30),
goalEventsDays: envInt(readEnv('GOAL_EVENTS_DAYS'), 60),
vacuumPages: envInt(readEnv('VACUUM_PAGES'), 500),
};
}

// Terminal (resolved) states for the external-event pipeline. `published` /
// `routed` are in-flight (may still be delivered) and are never pruned.
const EXTERNAL_EVENT_TERMINAL_STATES = [
Comment thread
lsm marked this conversation as resolved.
Outdated
'delivered',
'delivery_failed',
'failed',
'ignored',
'ambiguous',
] as const;

// Deliveries: `pending` is in-flight; only resolved deliveries are pruned.
const DELIVERY_TERMINAL_STATES = ['delivered', 'failed'] as const;

// GitHub events: `received` / `routed` are in-flight; the rest are resolved.
const GITHUB_EVENT_TERMINAL_STATES = [
'processed',
'ignored',
'ambiguous',
'delivered',
'failed',
] as const;

function tableExists(db: BunDatabase, name: string): boolean {
return !!db.prepare(`SELECT name FROM sqlite_master WHERE type = 'table' AND name = ?`).get(name);
}

function placeholders(count: number): string {
return Array.from({ length: count }, () => '?').join(',');
}

/**
* Count the matching rows, then delete them, returning the pre-delete count.
*
* We count first because `DELETE ... .changes` includes rows removed by FK
* CASCADE (e.g. a pruned external event pulls its deliveries), which would
* inflate the per-table stat. A SELECT COUNT is accurate and cheap (the age
* columns are indexed). The `whereClause` may carry `?` placeholders bound by
Comment thread
lsm marked this conversation as resolved.
Outdated
* `params`, used identically for the count and the delete.
*/
function prune(
db: BunDatabase,
table: string,
whereClause: string,
params: (string | number)[]
): number {
const count = (
db.prepare(`SELECT COUNT(*) AS c FROM ${table} WHERE ${whereClause}`).get(...params) as {
c: number;
}
).c;
if (count > 0) {
db.prepare(`DELETE FROM ${table} WHERE ${whereClause}`).run(...params);
}
return count;
}

/** Current page count of the database file. */
function pageCount(db: BunDatabase): number {
const row = db.prepare('PRAGMA page_count').get() as { page_count?: number } | null;
return Number(row?.page_count ?? 0);
}

/**
* Reclaim up to `maxPages` free pages from the end of the file.
*
* Returns the number of pages freed (page_count delta). A no-op — returns 0 —
* when the DB is not in incremental-vacuum mode (auto_vacuum != INCREMENTAL),
* which covers pre-migration existing databases.
*/
export function incrementalVacuum(db: BunDatabase, maxPages: number): number {
if (maxPages <= 0) return 0;
const mode = db.prepare('PRAGMA auto_vacuum').get() as { auto_vacuum?: number } | null;
if (Number(mode?.auto_vacuum ?? 0) !== 2) return 0; // 2 == INCREMENTAL

const before = pageCount(db);
// Pragma arguments cannot be bound; maxPages is a validated non-negative int.
db.exec(`PRAGMA incremental_vacuum(${maxPages})`);
const after = pageCount(db);
const freed = before - after;
return freed > 0 ? freed : 0;
}

/**
* Run retention sweeps for the configured tables, then reclaim freed pages.
*
* Each table is guarded by `tableExists` so this is safe to run against a DB
* that hasn't yet had a given table created (e.g. minimal test schemas, or a
* fresh DB mid-bootstrap). When `config.enabled` is false, no rows are deleted
* but `incremental_vacuum` still runs.
*/
export function runRetention(db: BunDatabase, config: RetentionConfig): RetentionStats {
const stats = emptyStats();

if (config.enabled) {
const now = Date.now();
const eventsCutoff = now - config.eventsDays * DAY_MS;
const mcpCutoff = now - config.mcpAuditDays * DAY_MS;
const goalCutoff = now - config.goalEventsDays * DAY_MS;

// Prune resolved deliveries BEFORE their events: the independent sweep reaps
// old resolved deliveries under a kept (in-flight) event, and runs with an
// accurate count. The event delete below then cascades any remaining
// deliveries of pruned events (uncounted, by design).
if (tableExists(db, 'space_external_event_deliveries')) {
stats.deliveries = prune(
db,
'space_external_event_deliveries',
`state IN (${placeholders(DELIVERY_TERMINAL_STATES.length)}) AND updated_at < ?`,
[...DELIVERY_TERMINAL_STATES, eventsCutoff]
);
}

if (tableExists(db, 'space_external_events')) {
stats.externalEvents = prune(
db,
'space_external_events',
`state IN (${placeholders(EXTERNAL_EVENT_TERMINAL_STATES.length)}) AND updated_at < ?`,
[...EXTERNAL_EVENT_TERMINAL_STATES, eventsCutoff]
);
}

if (tableExists(db, 'space_github_events')) {
stats.githubEvents = prune(
db,
'space_github_events',
`state IN (${placeholders(GITHUB_EVENT_TERMINAL_STATES.length)}) AND updated_at < ?`,
[...GITHUB_EVENT_TERMINAL_STATES, eventsCutoff]
);
}

if (tableExists(db, 'mcp_audit_log')) {
stats.mcpAudit = prune(db, 'mcp_audit_log', 'timestamp < ?', [mcpCutoff]);
}

if (tableExists(db, 'space_goal_events')) {
stats.goalEvents = prune(db, 'space_goal_events', 'created_at < ?', [goalCutoff]);
}
}

stats.vacuumedPages = incrementalVacuum(db, config.vacuumPages);
return stats;
}
18 changes: 17 additions & 1 deletion packages/daemon/src/storage/database-core.ts
Original file line number Diff line number Diff line change
Expand Up @@ -44,9 +44,25 @@ export class DatabaseCore {
mkdirSync(dir, { recursive: true });
}

// Open database
// Open database. Detect a fresh file (didn't exist before opening) so we can
// set auto_vacuum = INCREMENTAL before any table is created — the pragma is
// only effective on an empty DB.
const dbFileExisted = existsSync(this.dbPath);
this.db = new BunDatabase(this.dbPath);

// Enable incremental auto-vacuum on FRESH databases only. auto_vacuum is a
// header flag that only takes effect when set before the first table is
// created; on an existing DB the flip is a silent no-op until a full VACUUM.
// Existing (auto_vacuum = NONE) databases are converted by the opt-in
// migration 169 (HYPERNEO_DB_VACUUM_MIGRATION) — VACUUM on a multi-GB DB is a
// long, disk-intensive operation to schedule deliberately. Once INCREMENTAL,
// the daily cleanup job runs incremental_vacuum(500) to reclaim pages freed
// by retention sweeps and normal deletes, so the file shrinks over time
// instead of growing monotonically.
if (!dbFileExisted) {
this.db.exec('PRAGMA auto_vacuum = INCREMENTAL');
}

// Enable WAL mode for better concurrency and crash recovery
// WAL mode provides:
// - Better performance for concurrent reads/writes
Expand Down
50 changes: 50 additions & 0 deletions packages/daemon/src/storage/schema/migrations.ts
Original file line number Diff line number Diff line change
Expand Up @@ -793,6 +793,20 @@ export function runMigrations(db: BunDatabase, createBackup: () => void): void {
// lookup. (The live-query task-scope nodeExecStmt already drives off
// idx_node_executions_run and is unaffected.)
run(migrationMarkerKey(168), () => runMigration168(db));

// Migration 169: Convert existing auto_vacuum = NONE databases to INCREMENTAL
// so the daily cleanup job's incremental_vacuum(500) can reclaim pages freed
// by retention sweeps and normal deletes. Fresh databases already get
// INCREMENTAL at creation (DatabaseCore), so this no-ops for them.
//
// GATED behind HYPERNEO_DB_VACUUM_MIGRATION: a full VACUUM rebuilds the entire
// file (temporarily doubling disk usage), so on a multi-GB production DB it is
// a long, disk-intensive operation that must be scheduled deliberately (e.g.
// during a maintenance window). When the flag is unset the migration is not
// registered and not marked, so setting the flag + restarting runs it once.
if (envFlag(process.env.HYPERNEO_DB_VACUUM_MIGRATION)) {
run(migrationMarkerKey(169), () => runMigration169(db));
}
}

function migrationMarkerKey(version: number): string {
Expand Down Expand Up @@ -11477,3 +11491,39 @@ export function runMigration168(db: BunDatabase): void {
ON node_executions(agent_session_id)
`);
}

/**
* Read the database's `auto_vacuum` mode from its header.
* 0 = NONE (SQLite default), 1 = FULL, 2 = INCREMENTAL.
*/
function readAutoVacuumMode(db: BunDatabase): number {
const row = db.prepare('PRAGMA auto_vacuum').get() as { auto_vacuum?: number } | undefined;
return Number(row?.auto_vacuum ?? 0);
}

/**
* Migration 169: Convert auto_vacuum = NONE to INCREMENTAL via a full VACUUM.
*
* Fresh databases are created with auto_vacuum = INCREMENTAL by DatabaseCore, so
* this is a no-op for them (mode is already 2). FULL (1) is left untouched — it
* is not a target mode and changing it is out of scope. For a NONE database,
* setting the pragma records the desired mode in the header and the subsequent
* VACUUM rebuilds the file in incremental mode. After conversion, freed pages
* land on the free-list and incremental_vacuum(500) (daily cleanup job) reclaims
* them so the file shrinks instead of growing monotonically.
*
* See registration site: gated behind HYPERNEO_DB_VACUUM_MIGRATION.
*/
export function runMigration169(db: BunDatabase): void {
const mode = readAutoVacuumMode(db);
if (mode === 2) return; // already INCREMENTAL
if (mode === 1) return; // FULL — leave untouched
// mode === 0 (NONE) → rebuild into incremental mode.
db.exec('PRAGMA auto_vacuum = INCREMENTAL');
db.exec('VACUUM');
}

/** Truthy check for an opt-in `HYPERNEO_*` env flag. */
function envFlag(value: string | undefined): boolean {
return value === '1' || value === 'true';
}
Loading