Merge branch 'perf/local-opt': faster local install and updates, lighter polling and ingest
This commit is contained in:
@@ -425,6 +425,9 @@ function writeSession(f, conv, now) {
|
||||
}
|
||||
|
||||
let running = null;
|
||||
// Fresh sessions the last pass left for a later one to settle (the poll keeps
|
||||
// its base pace while there are any).
|
||||
let lastPending = 0;
|
||||
|
||||
/**
|
||||
* One pass over every conversation file: a changed one is re-read, an
|
||||
@@ -440,12 +443,16 @@ function ingestAntigravityUsage({
|
||||
running = (async () => {
|
||||
let read = 0;
|
||||
let written = 0;
|
||||
let pending = 0; // still-fresh sessions a later pass may have to settle
|
||||
for (const f of conversationFiles()) {
|
||||
try {
|
||||
const seen = getSeen.get(f.path);
|
||||
if (seen && seen.mtime_ms === f.mtime) {
|
||||
const a = activity(f.id, f.mtime);
|
||||
if (fresh(a, now)) continue;
|
||||
if (fresh(a, now)) {
|
||||
pending += 1;
|
||||
continue;
|
||||
}
|
||||
const s = stmts.getSession.get(f.id);
|
||||
if (s && s.tool === "antigravity" && s.status === "active") {
|
||||
const at = new Date(a.last).toISOString();
|
||||
@@ -486,6 +493,7 @@ function ingestAntigravityUsage({
|
||||
}
|
||||
await new Promise((r) => setImmediate(r));
|
||||
}
|
||||
lastPending = pending;
|
||||
return { files: read, sessions: written };
|
||||
})().finally(() => {
|
||||
running = null;
|
||||
@@ -504,16 +512,16 @@ function startAntigravityUsagePoll(broadcast, broadcastLane) {
|
||||
if (!Number.isFinite(POLL_MS) || POLL_MS <= 0) return;
|
||||
// Files an older build could not read get another try.
|
||||
db.prepare("DELETE FROM antigravity_conversations WHERE session_id = ''").run();
|
||||
const tick = () => {
|
||||
ingestAntigravityUsage({ broadcast, broadcastLane })
|
||||
.then((r) => {
|
||||
// Slowed to once a minute after a pass that read nothing, wrote nothing and
|
||||
// left no fresh session to settle; back to POLL_MS once one finds work.
|
||||
require("./backoff-poll").startBackoffPoll(
|
||||
() =>
|
||||
ingestAntigravityUsage({ broadcast, broadcastLane }).then((r) => {
|
||||
if (r.sessions > 0 && pricingSync.syncEnabled()) void pricingSync.syncPricing(db);
|
||||
})
|
||||
.catch(() => {});
|
||||
};
|
||||
tick();
|
||||
const timer = setInterval(tick, POLL_MS);
|
||||
if (timer.unref) timer.unref();
|
||||
return r.files + r.sessions + lastPending > 0;
|
||||
}),
|
||||
{ baseMs: POLL_MS }
|
||||
);
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
|
||||
@@ -0,0 +1,43 @@
|
||||
/**
|
||||
* @file A background poll that slows down while it finds nothing to do: the
|
||||
* next pass runs `baseMs` after a pass that did work (or failed), `idleMs`
|
||||
* after one whose `tick` resolved `false` (nothing changed, nothing pending).
|
||||
* One pass at a time — the next is scheduled only once the last settled — and
|
||||
* every timer is `unref`'d. Used by the Cursor and Antigravity usage polls.
|
||||
* @author Nguyễn Ngọc Trí Vĩ <vinnt@smartgift.vn>
|
||||
*/
|
||||
|
||||
/** Default slow interval: 60 s, never faster than the base one. */
|
||||
const idleDefault = (baseMs) => Math.max(baseMs, 60_000);
|
||||
|
||||
/**
|
||||
* Run `tick` now, then again and again. `tick` resolves `false` for an idle
|
||||
* pass; anything else (or a throw) keeps the base pace.
|
||||
* @returns {{stop: () => void}}
|
||||
*/
|
||||
function startBackoffPoll(tick, { baseMs, idleMs = idleDefault(baseMs) }) {
|
||||
let timer = null;
|
||||
let stopped = false;
|
||||
const schedule = (ms) => {
|
||||
if (stopped) return;
|
||||
timer = setTimeout(run, ms);
|
||||
if (timer.unref) timer.unref();
|
||||
};
|
||||
function run() {
|
||||
Promise.resolve()
|
||||
.then(tick)
|
||||
.then(
|
||||
(busy) => schedule(busy === false ? idleMs : baseMs),
|
||||
() => schedule(baseMs)
|
||||
);
|
||||
}
|
||||
run();
|
||||
return {
|
||||
stop() {
|
||||
stopped = true;
|
||||
clearTimeout(timer);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
module.exports = { startBackoffPoll, idleDefault };
|
||||
@@ -235,7 +235,9 @@ function threadNames(home = getCodexHome()) {
|
||||
const names = new Map();
|
||||
let text = "";
|
||||
try {
|
||||
text = require("./run-as").readFileSafe(path.join(home, "session_index.jsonl"), "utf8", { max: 16 << 20 });
|
||||
text = require("./run-as").readFileSafe(path.join(home, "session_index.jsonl"), "utf8", {
|
||||
max: 16 << 20,
|
||||
});
|
||||
} catch {
|
||||
return names;
|
||||
}
|
||||
@@ -251,6 +253,17 @@ function threadNames(home = getCodexHome()) {
|
||||
}
|
||||
|
||||
const getRollout = db.prepare("SELECT * FROM codex_rollouts WHERE path = ?");
|
||||
// What a pass needs to skip an unchanged journal, for every journal at once:
|
||||
// one query per pass instead of one (with its whole parse state) per file.
|
||||
const rolloutStamps = db.prepare(
|
||||
`SELECT path, size, mtime_ms, instr(state, '"more":true') > 0 AS more FROM codex_rollouts`
|
||||
);
|
||||
// Journals that may still be growing outside the newest day folders: written
|
||||
// within RECENT_ROW_MS, or not yet read to the end.
|
||||
const liveRolloutPaths = db.prepare(
|
||||
`SELECT path FROM codex_rollouts WHERE mtime_ms > ? OR instr(state, '"more":true') > 0`
|
||||
);
|
||||
const threadPath = db.prepare("SELECT path FROM codex_rollouts WHERE thread_id = ? LIMIT 1");
|
||||
const upsertRollout = db.prepare(
|
||||
`INSERT INTO codex_rollouts (path, session_id, thread_id, size, mtime_ms, offset, state, updated_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
|
||||
@@ -400,6 +413,59 @@ function writeSession(sessionId, names, now, origin = LOCAL) {
|
||||
|
||||
const running = new Map(); // home ("" = CODEX_HOME) → the pass in flight
|
||||
|
||||
// A RECENT pass (the poll's, between full ones) stats only the newest
|
||||
// RECENT_DAY_DIRS day folders plus the journals the table says may still grow
|
||||
// (RECENT_ROW_MS) — not every rollout Codex ever wrote. A journal in an older
|
||||
// folder that wakes up again (a resumed thread quiet for over RECENT_ROW_MS)
|
||||
// is picked up by the next full pass.
|
||||
const RECENT_DAY_DIRS = 2;
|
||||
const RECENT_ROW_MS = 60 * 60_000;
|
||||
|
||||
function numericDirs(dir) {
|
||||
try {
|
||||
return fs
|
||||
.readdirSync(dir, { withFileTypes: true })
|
||||
.filter((e) => e.isDirectory() && /^\d+$/.test(e.name))
|
||||
.map((e) => e.name)
|
||||
.sort((a, b) => Number(b) - Number(a));
|
||||
} catch {
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
function recentRolloutFiles(home, now) {
|
||||
const root = path.join(home || getCodexHome(), "sessions");
|
||||
const days = [];
|
||||
outer: for (const y of numericDirs(root))
|
||||
for (const m of numericDirs(path.join(root, y)))
|
||||
for (const d of numericDirs(path.join(root, y, m))) {
|
||||
days.push(path.join(root, y, m, d));
|
||||
if (days.length >= RECENT_DAY_DIRS) break outer;
|
||||
}
|
||||
const paths = new Set();
|
||||
for (const dir of days) {
|
||||
try {
|
||||
for (const e of fs.readdirSync(dir, { withFileTypes: true }))
|
||||
if (e.isFile() && /^rollout-.*\.jsonl$/.test(e.name)) paths.add(path.join(dir, e.name));
|
||||
} catch {
|
||||
/* vanished mid-scan */
|
||||
}
|
||||
}
|
||||
const prefix = root + path.sep;
|
||||
for (const { path: p } of liveRolloutPaths.all(now - RECENT_ROW_MS))
|
||||
if (p.startsWith(prefix)) paths.add(p);
|
||||
const out = [];
|
||||
for (const p of paths) {
|
||||
try {
|
||||
const stat = fs.statSync(p);
|
||||
if (stat.isFile()) out.push({ path: p, mtime: stat.mtimeMs, size: stat.size });
|
||||
} catch {
|
||||
/* gone */
|
||||
}
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
/**
|
||||
* One pass over every journal. Yields to the event loop between files; a
|
||||
* second call for the same home while one runs joins it. `home`, `source` and
|
||||
@@ -413,21 +479,42 @@ function ingestCodexUsage({
|
||||
home,
|
||||
source = LOCAL.source,
|
||||
activeMs = LOCAL.activeMs,
|
||||
recent = false,
|
||||
} = {}) {
|
||||
const key = home || "";
|
||||
if (running.has(key)) return running.get(key);
|
||||
const pass = (async () => {
|
||||
const files = allRolloutFiles(home);
|
||||
const files = recent ? recentRolloutFiles(home, now) : allRolloutFiles(home);
|
||||
const byThread = new Map();
|
||||
for (const f of files) {
|
||||
const m = FILE_NAME.exec(path.basename(f.path));
|
||||
if (m) byThread.set(m[1], f.path);
|
||||
}
|
||||
// A recent pass lists only part of the tree: a fork's parent may live in
|
||||
// an older folder, known to the table (or, failing that, to a full walk).
|
||||
let walkedAll = !recent;
|
||||
const parentPath = (id) => {
|
||||
if (byThread.has(id)) return byThread.get(id);
|
||||
if (walkedAll) return null;
|
||||
const known = threadPath.get(id)?.path;
|
||||
const root = path.join(home || getCodexHome(), "sessions") + path.sep;
|
||||
if (known && known.startsWith(root) && fs.existsSync(known)) return known;
|
||||
walkedAll = true;
|
||||
for (const f of allRolloutFiles(home)) {
|
||||
const m = FILE_NAME.exec(path.basename(f.path));
|
||||
if (m && !byThread.has(m[1])) byThread.set(m[1], f.path);
|
||||
}
|
||||
return byThread.get(id) || null;
|
||||
};
|
||||
const stamps = new Map();
|
||||
for (const r of rolloutStamps.all()) stamps.set(r.path, r);
|
||||
const touched = new Set();
|
||||
let read = 0;
|
||||
for (const f of files) {
|
||||
try {
|
||||
const mtime = Math.floor(f.mtime);
|
||||
const seen = stamps.get(f.path);
|
||||
if (seen && seen.size === f.size && seen.mtime_ms === mtime && !seen.more) continue;
|
||||
const row = getRollout.get(f.path);
|
||||
let offset = row ? row.offset : 0;
|
||||
let state = row ? JSON.parse(row.state) : {};
|
||||
@@ -439,7 +526,8 @@ function ingestCodexUsage({
|
||||
let skip = null;
|
||||
const parent = offset === 0 ? firstLineMeta(f.path).forked_from_id : state.forkedFrom;
|
||||
if (parent && !state.pastFork) {
|
||||
if (byThread.has(parent)) skip = await parentTotals(byThread.get(parent));
|
||||
const pp = parentPath(parent);
|
||||
if (pp) skip = await parentTotals(pp);
|
||||
else if (offset === 0)
|
||||
console.warn(`[codex-usage] fork parent ${parent} not found; ${f.path} may overcount`);
|
||||
}
|
||||
@@ -508,10 +596,15 @@ function startCodexUsagePoll(broadcast) {
|
||||
.memberHomes()
|
||||
.map((h) => path.join(h, ".codex")),
|
||||
];
|
||||
// Every FULL_EVERY-th pass (and the first) walks every journal; the others
|
||||
// only the recent ones (recentRolloutFiles).
|
||||
const FULL_EVERY = Math.max(1, Math.round((5 * 60_000) / POLL_MS));
|
||||
let passNo = 0;
|
||||
const ingestAll = async () => {
|
||||
const total = { sessions: 0 };
|
||||
const recent = passNo++ % FULL_EVERY !== 0;
|
||||
for (const home of homes()) {
|
||||
const r = await ingestCodexUsage({ broadcast, home });
|
||||
const r = await ingestCodexUsage({ broadcast, home, recent });
|
||||
total.sessions += r?.sessions || 0;
|
||||
}
|
||||
return total;
|
||||
|
||||
+20
-16
@@ -182,14 +182,7 @@ function writeSession(f, now) {
|
||||
const endedAt = active ? null : lastWrite;
|
||||
const name = firstPrompt(f.path) || existing?.name || `Cursor ${f.id.slice(0, 8)}`;
|
||||
if (existing) {
|
||||
updateSession.run(
|
||||
name,
|
||||
status,
|
||||
endedAt,
|
||||
lastWrite,
|
||||
cwd,
|
||||
f.id
|
||||
);
|
||||
updateSession.run(name, status, endedAt, lastWrite, cwd, f.id);
|
||||
return "session_updated";
|
||||
}
|
||||
insertSession.run(
|
||||
@@ -206,6 +199,9 @@ function writeSession(f, now) {
|
||||
}
|
||||
|
||||
let running = null;
|
||||
// Fresh sessions the last pass left for a later one to settle (the poll keeps
|
||||
// its base pace while there are any).
|
||||
let lastPending = 0;
|
||||
|
||||
/**
|
||||
* One pass over every transcript: a changed one is re-read, an unchanged one
|
||||
@@ -217,11 +213,15 @@ function ingestCursorUsage({ broadcast = () => {}, now = Date.now() } = {}) {
|
||||
running = (async () => {
|
||||
let read = 0;
|
||||
let written = 0;
|
||||
let pending = 0; // still-fresh sessions a later pass may have to settle
|
||||
for (const f of transcriptFiles()) {
|
||||
try {
|
||||
const seen = getSeen.get(f.path);
|
||||
if (seen && seen.mtime_ms === f.mtime) {
|
||||
if (fresh(f.mtime, now)) continue;
|
||||
if (fresh(f.mtime, now)) {
|
||||
pending += 1;
|
||||
continue;
|
||||
}
|
||||
const s = stmts.getSession.get(f.id);
|
||||
if (s && s.tool === "cursor" && s.status === "active") {
|
||||
settle.run(new Date(f.mtime).toISOString(), s.id);
|
||||
@@ -242,6 +242,7 @@ function ingestCursorUsage({ broadcast = () => {}, now = Date.now() } = {}) {
|
||||
}
|
||||
await new Promise((r) => setImmediate(r));
|
||||
}
|
||||
lastPending = pending;
|
||||
return { files: read, sessions: written };
|
||||
})().finally(() => {
|
||||
running = null;
|
||||
@@ -249,18 +250,21 @@ function ingestCursorUsage({ broadcast = () => {}, now = Date.now() } = {}) {
|
||||
return running;
|
||||
}
|
||||
|
||||
/** First pass at once (in the background), then every DASHBOARD_CURSOR_POLL_MS (default 15 s; 0 disables). */
|
||||
/**
|
||||
* First pass at once (in the background), then every DASHBOARD_CURSOR_POLL_MS
|
||||
* (default 15 s; 0 disables) — slowed to once a minute after a pass that read
|
||||
* nothing, wrote nothing and left no fresh session to settle, and back to the
|
||||
* base pace as soon as one finds work.
|
||||
*/
|
||||
function startCursorUsagePoll(broadcast) {
|
||||
const POLL_MS = process.env.DASHBOARD_CURSOR_POLL_MS
|
||||
? Number(process.env.DASHBOARD_CURSOR_POLL_MS)
|
||||
: 15_000;
|
||||
if (!Number.isFinite(POLL_MS) || POLL_MS <= 0) return;
|
||||
const tick = () => {
|
||||
ingestCursorUsage({ broadcast }).catch(() => {});
|
||||
};
|
||||
tick();
|
||||
const timer = setInterval(tick, POLL_MS);
|
||||
if (timer.unref) timer.unref();
|
||||
require("./backoff-poll").startBackoffPoll(
|
||||
() => ingestCursorUsage({ broadcast }).then((r) => r.files + r.sessions + lastPending > 0),
|
||||
{ baseMs: POLL_MS }
|
||||
);
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
|
||||
@@ -0,0 +1,337 @@
|
||||
/**
|
||||
* @file Database size upkeep (Settings → Data → Session cleanup): the shared
|
||||
* session purge, the OPT-IN automatic retention, the bounded reclaim of free
|
||||
* pages, and the manual Compact.
|
||||
*
|
||||
* - `purgeSessionIds` is THE deletion of whole sessions — events, agents,
|
||||
* token usage (daily and hourly) and the session row — used by both the
|
||||
* manual `POST /api/settings/cleanup` and the automatic retention, so the two
|
||||
* can never leave different leftovers.
|
||||
* - Retention is OFF unless `AGENTHUB_RETENTION_DAYS` (settings file, 1–3650)
|
||||
* is set; it never deletes data by default. When set, the hourly
|
||||
* `runScheduledMaintenance` purges finished (completed/error/abandoned)
|
||||
* sessions that STARTED more than that many days ago — the manual purge's
|
||||
* rule — in transactions of `RETENTION_BATCH` sessions, yielding to the event
|
||||
* loop between them, and at most `RETENTION_MAX_PER_RUN` per run, so a first
|
||||
* run on a long history never blocks hooks for long. Each batch deletes by
|
||||
* `idx_events_session`, never a scan of the events table.
|
||||
* - With `auto_vacuum = INCREMENTAL` (server/db.js sets it on a new database
|
||||
* and converts a small existing one at startup), the same run returns free
|
||||
* pages to the OS once the freelist passes `RECLAIM_MIN_PAGES`, in slices of
|
||||
* `RECLAIM_SLICE_PAGES`, at most `RECLAIM_MAX_PAGES` per run.
|
||||
* - `runCompact` (`POST /api/settings/compact`) is the one full VACUUM: in a
|
||||
* worker thread on its own connection, setting `auto_vacuum = INCREMENTAL`
|
||||
* first so a large existing database is converted, then a TRUNCATE
|
||||
* checkpoint so the file actually shrinks. It holds the write lock for its
|
||||
* whole run (tens of seconds on a GB-size file), so hooks wait or are retried
|
||||
* meanwhile — it only ever runs on a click. This file is also its worker's
|
||||
* entry point.
|
||||
* @author Nguyễn Ngọc Trí Vĩ <vinnt@smartgift.vn>
|
||||
*/
|
||||
|
||||
"use strict";
|
||||
|
||||
const { Worker, isMainThread, workerData, parentPort } = require("node:worker_threads");
|
||||
|
||||
const KEYS = { days: "AGENTHUB_RETENTION_DAYS" };
|
||||
const MAX_DAYS = 3650;
|
||||
const RETENTION_BATCH = 25;
|
||||
const RETENTION_MAX_PER_RUN = 5000;
|
||||
const RECLAIM_MIN_PAGES = 1024;
|
||||
const RECLAIM_SLICE_PAGES = 1024;
|
||||
const RECLAIM_MAX_PAGES = 65536;
|
||||
const PURGE_CHUNK = 500;
|
||||
const PURGEABLE = "status IN ('completed','error','abandoned')";
|
||||
|
||||
const coded = (message, code) => Object.assign(new Error(message), { code });
|
||||
const yieldLoop = () => new Promise((r) => setImmediate(r));
|
||||
|
||||
/** `{ days }` — 0 means off. Anything outside 1–MAX_DAYS reads as off. */
|
||||
function retentionConfig(env = process.env) {
|
||||
const raw = env[KEYS.days];
|
||||
const n = Number(raw);
|
||||
const days = raw !== undefined && raw !== "" && Number.isInteger(n) && n >= 1 && n <= MAX_DAYS;
|
||||
return { days: days ? n : 0 };
|
||||
}
|
||||
|
||||
/**
|
||||
* Delete these sessions and everything hung off them, in chunks (a single
|
||||
* IN (...) of every id would pass SQLite's bound-variable limit). Runs in the
|
||||
* CALLER's transaction when there is one. Returns the rows removed.
|
||||
*/
|
||||
function purgeSessionIds(db, ids) {
|
||||
const { forgetHourly } = require("./hourly-usage");
|
||||
const out = { purged_sessions: 0, purged_events: 0, purged_agents: 0 };
|
||||
for (let i = 0; i < ids.length; i += PURGE_CHUNK) {
|
||||
const chunk = ids.slice(i, i + PURGE_CHUNK);
|
||||
const placeholders = chunk.map(() => "?").join(",");
|
||||
// Cascading deletes handle agents/events, but token_usage FK might not cascade on all setups
|
||||
out.purged_events += db
|
||||
.prepare(`DELETE FROM events WHERE session_id IN (${placeholders})`)
|
||||
.run(...chunk).changes;
|
||||
out.purged_agents += db
|
||||
.prepare(`DELETE FROM agents WHERE session_id IN (${placeholders})`)
|
||||
.run(...chunk).changes;
|
||||
db.prepare(`DELETE FROM token_usage WHERE session_id IN (${placeholders})`).run(...chunk);
|
||||
db.prepare(`DELETE FROM token_usage_hourly WHERE session_id IN (${placeholders})`).run(
|
||||
...chunk
|
||||
);
|
||||
forgetHourly(chunk);
|
||||
out.purged_sessions += db
|
||||
.prepare(`DELETE FROM sessions WHERE id IN (${placeholders})`)
|
||||
.run(...chunk).changes;
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
/** What `purgeSessionIds` would delete, without deleting it (the dry run). */
|
||||
function countSessionIds(db, ids) {
|
||||
const out = { purged_sessions: ids.length, purged_events: 0, purged_agents: 0 };
|
||||
for (let i = 0; i < ids.length; i += PURGE_CHUNK) {
|
||||
const chunk = ids.slice(i, i + PURGE_CHUNK);
|
||||
const placeholders = chunk.map(() => "?").join(",");
|
||||
out.purged_events += db
|
||||
.prepare(`SELECT COUNT(*) AS n FROM events WHERE session_id IN (${placeholders})`)
|
||||
.get(...chunk).n;
|
||||
out.purged_agents += db
|
||||
.prepare(`SELECT COUNT(*) AS n FROM agents WHERE session_id IN (${placeholders})`)
|
||||
.get(...chunk).n;
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
/** Finished sessions that started before `cutoff` (ISO), at most `limit`. */
|
||||
function purgeableBefore(db, cutoff, limit = -1) {
|
||||
return db
|
||||
.prepare(`SELECT id, started_at FROM sessions WHERE ${PURGEABLE} AND started_at < ? LIMIT ?`)
|
||||
.all(cutoff, limit);
|
||||
}
|
||||
|
||||
/**
|
||||
* Purge finished sessions older than `days`, a batch per transaction with a
|
||||
* yield between batches. Resolves to the totals; never throws.
|
||||
*/
|
||||
async function applyRetention({
|
||||
db = require("../db").db,
|
||||
days,
|
||||
now = Date.now(),
|
||||
batch = RETENTION_BATCH,
|
||||
maxSessions = RETENTION_MAX_PER_RUN,
|
||||
} = {}) {
|
||||
const out = { purged_sessions: 0, purged_events: 0, purged_agents: 0 };
|
||||
if (!Number.isInteger(days) || days < 1) return out;
|
||||
const cutoff = new Date(now - days * 86400 * 1000).toISOString();
|
||||
try {
|
||||
while (out.purged_sessions < maxSessions) {
|
||||
const ids = purgeableBefore(
|
||||
db,
|
||||
cutoff,
|
||||
Math.min(batch, maxSessions - out.purged_sessions)
|
||||
).map((r) => r.id);
|
||||
if (ids.length === 0) break;
|
||||
const r = db.transaction(() => purgeSessionIds(db, ids))();
|
||||
out.purged_sessions += r.purged_sessions;
|
||||
out.purged_events += r.purged_events;
|
||||
out.purged_agents += r.purged_agents;
|
||||
if (r.purged_sessions === 0) break; // nothing removable: never spin
|
||||
await yieldLoop();
|
||||
}
|
||||
} catch (err) {
|
||||
out.error = err.message;
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
/** One `incremental_vacuum(n)`, stepped to the end on either driver. */
|
||||
function incrementalVacuum(db, pages) {
|
||||
const sql = `PRAGMA incremental_vacuum(${Math.max(1, Math.floor(pages))})`;
|
||||
try {
|
||||
db.prepare(sql).all(); // node:sqlite: one row per freed page
|
||||
} catch (err) {
|
||||
if (!/does not return data/i.test(err.message)) throw err;
|
||||
db.pragma(`incremental_vacuum(${Math.max(1, Math.floor(pages))})`); // better-sqlite3
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Hand free pages back to the OS, a slice at a time. A no-op unless the
|
||||
* database is in INCREMENTAL auto_vacuum and its freelist is over `minPages`.
|
||||
*/
|
||||
async function reclaimFreePages({
|
||||
db = require("../db").db,
|
||||
minPages = RECLAIM_MIN_PAGES,
|
||||
slice = RECLAIM_SLICE_PAGES,
|
||||
maxPages = RECLAIM_MAX_PAGES,
|
||||
} = {}) {
|
||||
const out = { reclaimed_pages: 0 };
|
||||
try {
|
||||
if (db.pragma("auto_vacuum", { simple: true }) !== 2) return out;
|
||||
let free = db.pragma("freelist_count", { simple: true });
|
||||
if (free < minPages) return out;
|
||||
while (free > 0 && out.reclaimed_pages < maxPages) {
|
||||
incrementalVacuum(db, Math.min(slice, free, maxPages - out.reclaimed_pages));
|
||||
const after = db.pragma("freelist_count", { simple: true });
|
||||
if (after >= free) break;
|
||||
out.reclaimed_pages += free - after;
|
||||
free = after;
|
||||
await yieldLoop();
|
||||
}
|
||||
} catch (err) {
|
||||
out.error = err.message;
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
let lastRun = null;
|
||||
let maintaining = null;
|
||||
|
||||
/**
|
||||
* The hourly upkeep: `maintainDatabase` (checkpoint + optimize), then the
|
||||
* opt-in retention, then the bounded free-page reclaim. A call while one runs
|
||||
* joins it.
|
||||
*/
|
||||
function runScheduledMaintenance(opts = {}) {
|
||||
if (maintaining) return maintaining;
|
||||
maintaining = maintain(opts).finally(() => {
|
||||
maintaining = null;
|
||||
});
|
||||
return maintaining;
|
||||
}
|
||||
|
||||
async function maintain({ db } = {}) {
|
||||
const dbm = require("../db");
|
||||
const conn = db || dbm.db;
|
||||
const started = new Date();
|
||||
const out = { at: started.toISOString() };
|
||||
try {
|
||||
if (!db) dbm.maintainDatabase();
|
||||
} catch {
|
||||
/* maintainDatabase reports its own errors in its result */
|
||||
}
|
||||
const { days } = retentionConfig();
|
||||
if (days) {
|
||||
out.retention = await applyRetention({ db: conn, days });
|
||||
if (out.retention.purged_sessions > 0) {
|
||||
console.log(
|
||||
`[retention] purged ${out.retention.purged_sessions} session(s) older than ${days} days`
|
||||
);
|
||||
}
|
||||
}
|
||||
out.reclaim = await reclaimFreePages({ db: conn });
|
||||
out.ms = Date.now() - started;
|
||||
lastRun = out;
|
||||
return out;
|
||||
}
|
||||
|
||||
/** Page counts for the Settings card. */
|
||||
function spaceInfo(db = require("../db").db) {
|
||||
const pageSize = db.pragma("page_size", { simple: true });
|
||||
const pageCount = db.pragma("page_count", { simple: true });
|
||||
const free = db.pragma("freelist_count", { simple: true });
|
||||
const mode = db.pragma("auto_vacuum", { simple: true });
|
||||
return {
|
||||
auto_vacuum: mode === 2 ? "incremental" : mode === 1 ? "full" : "none",
|
||||
page_size: pageSize,
|
||||
page_count: pageCount,
|
||||
freelist_pages: free,
|
||||
free_bytes: free * pageSize,
|
||||
};
|
||||
}
|
||||
|
||||
let compacting = null;
|
||||
let lastCompact = null;
|
||||
|
||||
function vacuumInWorker(dbPath) {
|
||||
return new Promise((resolve, reject) => {
|
||||
const w = new Worker(__filename, { workerData: { agenthubCompactWorker: true, dbPath } });
|
||||
w.once("message", (m) => (m.error ? reject(coded(m.error.message, m.error.code)) : resolve(m)));
|
||||
w.once("error", reject);
|
||||
w.once(
|
||||
"exit",
|
||||
(code) => code !== 0 && reject(coded(`compact worker exited (${code})`, "ECOMPACT"))
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
/** Start (or join) the manual Compact. The outcome is `lastCompactResult()`. */
|
||||
function runCompact() {
|
||||
if (compacting) return compacting;
|
||||
const fs = require("node:fs");
|
||||
const { DB_PATH } = require("../db");
|
||||
const size = () => {
|
||||
try {
|
||||
return fs.statSync(DB_PATH).size;
|
||||
} catch {
|
||||
return 0;
|
||||
}
|
||||
};
|
||||
compacting = (async () => {
|
||||
const started = Date.now();
|
||||
const before = size();
|
||||
try {
|
||||
await vacuumInWorker(DB_PATH);
|
||||
lastCompact = {
|
||||
at: new Date(started).toISOString(),
|
||||
ok: true,
|
||||
size_before: before,
|
||||
size_after: size(),
|
||||
ms: Date.now() - started,
|
||||
};
|
||||
return lastCompact;
|
||||
} catch (err) {
|
||||
lastCompact = { at: new Date(started).toISOString(), ok: false, error: err.message };
|
||||
throw err;
|
||||
}
|
||||
})().finally(() => {
|
||||
compacting = null;
|
||||
});
|
||||
return compacting;
|
||||
}
|
||||
|
||||
const isCompacting = () => compacting !== null;
|
||||
const lastCompactResult = () => lastCompact;
|
||||
const lastMaintenance = () => lastRun;
|
||||
|
||||
if (!isMainThread && workerData?.agenthubCompactWorker) {
|
||||
// Same driver order as server/db.js: better-sqlite3, else node:sqlite.
|
||||
let conn;
|
||||
try {
|
||||
try {
|
||||
const Database = require("better-sqlite3");
|
||||
conn = new Database(workerData.dbPath);
|
||||
} catch {
|
||||
const Database = require("../compat-sqlite");
|
||||
conn = new Database(workerData.dbPath);
|
||||
}
|
||||
conn.pragma("busy_timeout = 30000");
|
||||
conn.pragma("auto_vacuum = INCREMENTAL"); // stored by the VACUUM below
|
||||
conn.exec("VACUUM");
|
||||
try {
|
||||
conn.pragma("wal_checkpoint(TRUNCATE)");
|
||||
} catch {
|
||||
/* a reader holds the WAL: the periodic checkpoint finishes it */
|
||||
}
|
||||
parentPort.postMessage({ ok: true });
|
||||
} catch (err) {
|
||||
parentPort.postMessage({ error: { message: err.message, code: err.code || "ECOMPACT" } });
|
||||
} finally {
|
||||
conn?.close();
|
||||
}
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
KEYS,
|
||||
MAX_DAYS,
|
||||
retentionConfig,
|
||||
purgeSessionIds,
|
||||
countSessionIds,
|
||||
purgeableBefore,
|
||||
applyRetention,
|
||||
reclaimFreePages,
|
||||
runScheduledMaintenance,
|
||||
spaceInfo,
|
||||
runCompact,
|
||||
isCompacting,
|
||||
lastCompactResult,
|
||||
lastMaintenance,
|
||||
};
|
||||
+16
-8
@@ -2,21 +2,29 @@
|
||||
* @file Self-hosted ReDoc API reference. ReDoc renders the OpenAPI spec as a
|
||||
* clean, three-panel reference document — a read-optimized complement to
|
||||
* Swagger UI's interactive "try it out" console (both are served from the same
|
||||
* `/api/openapi.json` spec). The ReDoc bundle ships with the `redoc`
|
||||
* dependency and is served straight from `node_modules` rather than a CDN, so
|
||||
* the docs render fully offline / air-gapped, consistent with the project's
|
||||
* no-external-assets policy.
|
||||
* `/api/openapi.json` spec). The ReDoc bundle is served from the repo itself
|
||||
* rather than a CDN, so the docs render fully offline / air-gapped, consistent
|
||||
* with the project's no-external-assets policy.
|
||||
*
|
||||
* The bundle is VENDORED (`server/vendor/redoc/`): `redoc.standalone.js` from
|
||||
* the `redoc` npm package 2.5.4, byte-for-byte, with its MIT `LICENSE` and the
|
||||
* bundled libraries' `redoc.standalone.js.LICENSE.txt`. The `redoc` package
|
||||
* itself is not a dependency — the server only ever serves this one browser
|
||||
* file, and the package pulled ~25 MB of Node-side dependencies into every
|
||||
* runtime install. To update: `npm pack redoc@<version>`, copy
|
||||
* `package/bundles/redoc.standalone.js` (+ `.LICENSE.txt`) over these files and
|
||||
* bump the version named here.
|
||||
* @author Nguyễn Ngọc Trí Vĩ <vinnt@smartgift.vn>
|
||||
*/
|
||||
|
||||
const path = require("path");
|
||||
|
||||
/**
|
||||
* Absolute path to the prebuilt ReDoc standalone bundle inside the installed
|
||||
* `redoc` package. Resolved through Node's module resolution so it works
|
||||
* regardless of hoisting / install layout. Throws if `redoc` is not installed.
|
||||
* Absolute path to the vendored ReDoc standalone bundle.
|
||||
* @returns {string}
|
||||
*/
|
||||
function redocBundlePath() {
|
||||
return require.resolve("redoc/bundles/redoc.standalone.js");
|
||||
return path.join(__dirname, "..", "vendor", "redoc", "redoc.standalone.js");
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -115,6 +115,19 @@ async function git(
|
||||
});
|
||||
return { stdout: result.stdout, stderr: result.stderr };
|
||||
} catch (err) {
|
||||
// ENOENT from execFile is ambiguous: a missing binary OR a missing cwd.
|
||||
// With the folder present it is git itself that is absent — say so, with
|
||||
// the fix, rather than `git status failed` (like ETMUXMISSING for tmux).
|
||||
if (err.code === "ENOENT" && cmd === "git" && fs.existsSync(cwd)) {
|
||||
const error = new Error(
|
||||
"git is not installed (or not on the dashboard's PATH), and every git action needs it. " +
|
||||
"Install it (sudo apt install git / brew install git / https://git-scm.com/download/win), " +
|
||||
"then restart the dashboard."
|
||||
);
|
||||
error.code = "EGITMISSING";
|
||||
error.git = { args, code: "EGITMISSING", stderr: "", stdout: "" };
|
||||
throw error;
|
||||
}
|
||||
const error = new Error(`git ${args[0]} failed`);
|
||||
error.code = err.code;
|
||||
error.git = { args, code: err.code, stderr: err.stderr, stdout: err.stdout };
|
||||
|
||||
Reference in New Issue
Block a user