Garbage-collect orphan inbox projection ledger rows
agent_message_projections accumulates one row per (message_id, session_id) pair and previously had no purge path — the table grew monotonically for the life of the process. Once a session is rotated away no agent references its id as the active one, and its ledger rows exist only to prevent replay against a shadow that no future run will touch. Add purgeOrphanProjections (delete rows whose session_id no longer appears in agents.session_id) and wire it into the housekeeping scheduler with a 24h safety margin so rotations in flight during the sweep cannot strip the ledger under an active run.
This commit is contained in:
1 parent
1d172c60f1
commit
f59acd2778
2 files changed
+34
No files matched your search
@@ -28,5 +28,23 @@ export function buildInboxProjectionMethods(getDb: () => DatabaseSync) {
|
||||
).run(messageId, sessionId, nowIso());
|
||||
return Number(result.changes ?? 0) > 0;
|
||||
},
|
||||
|
||||
/**
|
||||
* Purge ledger rows for sessions no agent currently references as
|
||||
* its active `session_id`. A rotated session is never read again —
|
||||
* its ledger entries only exist to prevent replays against that
|
||||
* specific shadow, which no future run will touch. The safety
|
||||
* margin (rows younger than the cutoff are preserved) covers
|
||||
* rotations that happen mid-sweep so a race cannot strip the
|
||||
* ledger under an active run.
|
||||
*/
|
||||
purgeOrphanProjections(cutoffIso: string): number {
|
||||
const result = getDb().prepare(
|
||||
`DELETE FROM agent_message_projections
|
||||
WHERE projected_at < ?
|
||||
AND session_id NOT IN (SELECT session_id FROM agents WHERE session_id IS NOT NULL)`,
|
||||
).run(cutoffIso);
|
||||
return Number(result.changes ?? 0);
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -88,6 +88,22 @@ export function createHousekeepingScheduler(
|
||||
} catch (error) {
|
||||
logger.error(`[housekeeping] Wakeup purge failed: ${error instanceof Error ? error.message : String(error)}`);
|
||||
}
|
||||
|
||||
try {
|
||||
// agent_message_projections accumulates one row per (inbox
|
||||
// message, session) forever. Once a session has rotated away
|
||||
// — no agent references it as its current session_id — its
|
||||
// projection rows only consume disk. Keep a 24h safety margin
|
||||
// so an active rotation window cannot strip the ledger out
|
||||
// from under a live run.
|
||||
const cutoff = new Date(Date.now() - 24 * 60 * 60 * 1000).toISOString();
|
||||
const purgedProjections = db.inboxProjections.purgeOrphanProjections(cutoff);
|
||||
if (purgedProjections > 0) {
|
||||
logger.info(`[housekeeping] Purged ${purgedProjections} orphan inbox projection ledger row(s)`);
|
||||
}
|
||||
} catch (error) {
|
||||
logger.error(`[housekeeping] Inbox projection purge failed: ${error instanceof Error ? error.message : String(error)}`);
|
||||
}
|
||||
} catch (error) {
|
||||
logger.error(`[housekeeping] Scheduler error: ${error instanceof Error ? error.message : String(error)}`);
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user