Merge branch 'perf/local-dbupd' into perf/local-opt
This commit is contained in:
@@ -0,0 +1,175 @@
|
||||
/**
|
||||
* @file Tests server/lib/db-retention.js and its Settings routes — a new
|
||||
* database is INCREMENTAL auto_vacuum, retention is off by default and purges
|
||||
* nothing then, when on it removes only finished sessions older than the
|
||||
* cutoff (with their events, agents and token rows) in bounded batches, the
|
||||
* free-page reclaim shrinks the freelist, GET/PUT /api/settings/retention
|
||||
* validate and persist the setting, and POST /api/settings/compact needs
|
||||
* confirm and runs a VACUUM in a worker. The database and settings file are
|
||||
* temp paths.
|
||||
* @author Nguyễn Ngọc Trí Vĩ <vinnt@smartgift.vn>
|
||||
*/
|
||||
|
||||
const os = require("node:os");
|
||||
const fs = require("node:fs");
|
||||
const path = require("node:path");
|
||||
const { describe, it, before, after } = require("node:test");
|
||||
const assert = require("node:assert/strict");
|
||||
|
||||
const ROOT = fs.mkdtempSync(path.join(os.tmpdir(), "agenthub-db-retention-"));
|
||||
process.env.DASHBOARD_DB_PATH = path.join(ROOT, "dashboard.db");
|
||||
process.env.AGENTHUB_SETTINGS_FILE = path.join(ROOT, "settings.env");
|
||||
process.env.DASHBOARD_LIVENESS_PROBE = "0";
|
||||
delete process.env.AGENTHUB_RETENTION_DAYS;
|
||||
|
||||
const { createApp, startServer } = require("../index");
|
||||
const { db } = require("../db");
|
||||
const retention = require("../lib/db-retention");
|
||||
|
||||
let server;
|
||||
let BASE;
|
||||
before(async () => {
|
||||
server = await startServer(createApp(), 0);
|
||||
BASE = `http://127.0.0.1:${server.address().port}`;
|
||||
});
|
||||
after(() => {
|
||||
server?.close();
|
||||
delete process.env.AGENTHUB_RETENTION_DAYS;
|
||||
fs.rmSync(ROOT, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
async function call(method, urlPath, body) {
|
||||
const res = await fetch(BASE + urlPath, {
|
||||
method,
|
||||
headers: body ? { "content-type": "application/json" } : {},
|
||||
body: body ? JSON.stringify(body) : undefined,
|
||||
});
|
||||
return { status: res.status, body: await res.json() };
|
||||
}
|
||||
|
||||
const daysAgo = (n) => new Date(Date.now() - n * 86400 * 1000).toISOString();
|
||||
|
||||
function seed(id, status, startedAt, events = 3) {
|
||||
db.prepare(
|
||||
"INSERT INTO sessions (id, name, status, cwd, started_at, updated_at) VALUES (?, ?, ?, '/tmp', ?, ?)"
|
||||
).run(id, id, status, startedAt, startedAt);
|
||||
const ins = db.prepare(
|
||||
"INSERT INTO events (session_id, event_type, summary, data, created_at) VALUES (?, 'PostToolUse', 's', ?, ?)"
|
||||
);
|
||||
for (let i = 0; i < events; i++)
|
||||
ins.run(id, JSON.stringify({ pad: "x".repeat(2000) }), startedAt);
|
||||
}
|
||||
const exists = (id) => !!db.prepare("SELECT 1 FROM sessions WHERE id = ?").get(id);
|
||||
const eventCount = (id) =>
|
||||
db.prepare("SELECT COUNT(*) AS n FROM events WHERE session_id = ?").get(id).n;
|
||||
|
||||
describe("auto_vacuum", () => {
|
||||
it("a new database is INCREMENTAL", () => {
|
||||
assert.equal(db.pragma("auto_vacuum", { simple: true }), 2);
|
||||
assert.equal(retention.spaceInfo(db).auto_vacuum, "incremental");
|
||||
});
|
||||
});
|
||||
|
||||
describe("retentionConfig", () => {
|
||||
it("is off by default and for anything out of range", () => {
|
||||
assert.equal(retention.retentionConfig({}).days, 0);
|
||||
for (const v of ["", "0", "-3", "1.5", "abc", "99999"]) {
|
||||
assert.equal(retention.retentionConfig({ AGENTHUB_RETENTION_DAYS: v }).days, 0, v);
|
||||
}
|
||||
assert.equal(retention.retentionConfig({ AGENTHUB_RETENTION_DAYS: "30" }).days, 30);
|
||||
});
|
||||
});
|
||||
|
||||
describe("scheduled maintenance", () => {
|
||||
it("purges nothing while retention is off", async () => {
|
||||
seed("ret-off-old", "completed", daysAgo(400));
|
||||
const out = await retention.runScheduledMaintenance({ db });
|
||||
assert.equal(out.retention, undefined);
|
||||
assert.ok(exists("ret-off-old"));
|
||||
db.prepare("DELETE FROM sessions WHERE id = 'ret-off-old'").run();
|
||||
});
|
||||
|
||||
it("purges only finished sessions older than the cutoff, with their rows", async () => {
|
||||
seed("ret-old-done", "completed", daysAgo(100));
|
||||
seed("ret-old-err", "error", daysAgo(100));
|
||||
seed("ret-old-active", "active", daysAgo(100));
|
||||
seed("ret-new-done", "completed", daysAgo(5));
|
||||
db.prepare(
|
||||
"INSERT INTO token_usage (session_id, model, input_tokens) VALUES ('ret-old-done', 'm', 1)"
|
||||
).run();
|
||||
process.env.AGENTHUB_RETENTION_DAYS = "30";
|
||||
try {
|
||||
const out = await retention.runScheduledMaintenance({ db });
|
||||
assert.equal(out.retention.purged_sessions, 2);
|
||||
assert.equal(out.retention.purged_events, 6);
|
||||
} finally {
|
||||
delete process.env.AGENTHUB_RETENTION_DAYS;
|
||||
}
|
||||
assert.ok(!exists("ret-old-done"));
|
||||
assert.ok(!exists("ret-old-err"));
|
||||
assert.equal(eventCount("ret-old-done"), 0);
|
||||
assert.equal(
|
||||
db.prepare("SELECT COUNT(*) AS n FROM token_usage WHERE session_id = 'ret-old-done'").get().n,
|
||||
0
|
||||
);
|
||||
assert.ok(exists("ret-old-active"), "an active session is never purged");
|
||||
assert.ok(exists("ret-new-done"), "a recent session is kept");
|
||||
assert.equal(eventCount("ret-new-done"), 3);
|
||||
});
|
||||
|
||||
it("works in bounded batches and stops at the per-run cap", async () => {
|
||||
for (let i = 0; i < 7; i++) seed(`ret-batch-${i}`, "abandoned", daysAgo(50), 1);
|
||||
const first = await retention.applyRetention({ db, days: 10, batch: 2, maxSessions: 5 });
|
||||
assert.equal(first.purged_sessions, 5);
|
||||
const second = await retention.applyRetention({ db, days: 10, batch: 2, maxSessions: 5 });
|
||||
assert.equal(second.purged_sessions, 2);
|
||||
const third = await retention.applyRetention({ db, days: 10 });
|
||||
assert.equal(third.purged_sessions, 0);
|
||||
});
|
||||
|
||||
it("returns free pages to the OS", async () => {
|
||||
for (let i = 0; i < 40; i++) seed(`ret-big-${i}`, "completed", daysAgo(60), 20);
|
||||
await retention.applyRetention({ db, days: 30 });
|
||||
const before = db.pragma("freelist_count", { simple: true });
|
||||
assert.ok(before > 50, `expected free pages, got ${before}`);
|
||||
const out = await retention.reclaimFreePages({ db, minPages: 10, slice: 16 });
|
||||
assert.ok(out.reclaimed_pages > 0);
|
||||
assert.ok(db.pragma("freelist_count", { simple: true }) < before);
|
||||
});
|
||||
});
|
||||
|
||||
describe("routes", () => {
|
||||
it("GET reports retention off by default", async () => {
|
||||
const r = await call("GET", "/api/settings/retention");
|
||||
assert.equal(r.status, 200);
|
||||
assert.equal(r.body.days, 0);
|
||||
assert.equal(r.body.compacting, false);
|
||||
assert.equal(r.body.space.auto_vacuum, "incremental");
|
||||
});
|
||||
|
||||
it("PUT validates and persists, 0 turns it off", async () => {
|
||||
for (const bad of [-1, 1.5, "7", 99999, null]) {
|
||||
const r = await call("PUT", "/api/settings/retention", { days: bad });
|
||||
assert.equal(r.status, 400, String(bad));
|
||||
}
|
||||
let r = await call("PUT", "/api/settings/retention", { days: 90 });
|
||||
assert.equal(r.status, 200);
|
||||
assert.equal(r.body.days, 90);
|
||||
assert.match(fs.readFileSync(process.env.AGENTHUB_SETTINGS_FILE, "utf8"), /RETENTION_DAYS=90/);
|
||||
r = await call("PUT", "/api/settings/retention", { days: 0 });
|
||||
assert.equal(r.body.days, 0);
|
||||
});
|
||||
|
||||
it("POST /compact needs confirm and then vacuums", async () => {
|
||||
let r = await call("POST", "/api/settings/compact", {});
|
||||
assert.equal(r.status, 400);
|
||||
r = await call("POST", "/api/settings/compact", { confirm: true });
|
||||
assert.equal(r.status, 202);
|
||||
for (let i = 0; i < 200 && retention.isCompacting(); i++) {
|
||||
await new Promise((res) => setTimeout(res, 50));
|
||||
}
|
||||
const last = retention.lastCompactResult();
|
||||
assert.equal(last?.ok, true, last?.error);
|
||||
assert.equal(db.pragma("auto_vacuum", { simple: true }), 2);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,76 @@
|
||||
/**
|
||||
* @file Tests server/update-scheduler.js — the default interval is an hour
|
||||
* (the env override and its 60 s floor still apply), a scheduled tick with no
|
||||
* browser connected runs no check, and a browser connecting after that gets
|
||||
* one catch-up check (only when the last check is stale). The check, the
|
||||
* connection count and the connection event are injected; nothing touches git
|
||||
* or the network.
|
||||
* @author Nguyễn Ngọc Trí Vĩ <vinnt@smartgift.vn>
|
||||
*/
|
||||
|
||||
const { describe, it, afterEach } = require("node:test");
|
||||
const assert = require("node:assert/strict");
|
||||
|
||||
delete process.env.DASHBOARD_UPDATE_CHECK;
|
||||
const { startUpdateScheduler, intervalMs } = require("../update-scheduler");
|
||||
|
||||
const wait = (ms) => new Promise((r) => setTimeout(r, ms));
|
||||
|
||||
let handle = null;
|
||||
afterEach(() => {
|
||||
handle?.stop();
|
||||
handle = null;
|
||||
delete process.env.DASHBOARD_UPDATE_CHECK_INTERVAL_MS;
|
||||
});
|
||||
|
||||
function harness(clients) {
|
||||
const h = { checks: 0, clients, listeners: [] };
|
||||
handle = startUpdateScheduler({
|
||||
broadcast: () => {},
|
||||
getConnectionCount: () => h.clients,
|
||||
onConnection: (fn) => {
|
||||
h.listeners.push(fn);
|
||||
return () => {};
|
||||
},
|
||||
getStatus: async () => {
|
||||
h.checks++;
|
||||
return { update_available: false };
|
||||
},
|
||||
initialDelayMs: 10,
|
||||
});
|
||||
h.connect = () => h.listeners.forEach((fn) => fn());
|
||||
return h;
|
||||
}
|
||||
|
||||
describe("intervalMs", () => {
|
||||
it("defaults to one hour and keeps the override with its floor", () => {
|
||||
delete process.env.DASHBOARD_UPDATE_CHECK_INTERVAL_MS;
|
||||
assert.equal(intervalMs(), 60 * 60 * 1000);
|
||||
process.env.DASHBOARD_UPDATE_CHECK_INTERVAL_MS = "120000";
|
||||
assert.equal(intervalMs(), 120000);
|
||||
process.env.DASHBOARD_UPDATE_CHECK_INTERVAL_MS = "1000";
|
||||
assert.equal(intervalMs(), 60 * 60 * 1000);
|
||||
});
|
||||
});
|
||||
|
||||
describe("startUpdateScheduler", () => {
|
||||
it("checks on the first tick when a browser is connected", async () => {
|
||||
const h = harness(1);
|
||||
await wait(40);
|
||||
assert.equal(h.checks, 1);
|
||||
});
|
||||
|
||||
it("skips the tick with no browser, then checks once when one connects", async () => {
|
||||
const h = harness(0);
|
||||
await wait(40);
|
||||
assert.equal(h.checks, 0);
|
||||
h.clients = 1;
|
||||
h.connect();
|
||||
h.connect(); // a second tab in the same window: still one check
|
||||
await wait(3_300);
|
||||
assert.equal(h.checks, 1);
|
||||
h.connect(); // fresh check: no catch-up
|
||||
await wait(3_300);
|
||||
assert.equal(h.checks, 1);
|
||||
});
|
||||
});
|
||||
+35
-1
@@ -141,6 +141,14 @@ migrateLegacyDatabase(DB_PATH);
|
||||
|
||||
const db = new Database(DB_PATH);
|
||||
|
||||
// Freed pages go back to the OS only with auto_vacuum. On a NEW database this
|
||||
// takes effect at once — it must come before `journal_mode = WAL`, which
|
||||
// writes the header; an existing one is converted further down, by a VACUUM,
|
||||
// only while it is small (`convertAutoVacuum`), else by Settings → Data →
|
||||
// Compact. `maintainDatabase` then runs a bounded `incremental_vacuum`.
|
||||
if (db.pragma("auto_vacuum", { simple: true }) === 0) {
|
||||
db.pragma("auto_vacuum = INCREMENTAL");
|
||||
}
|
||||
db.pragma("journal_mode = WAL");
|
||||
db.pragma("foreign_keys = ON");
|
||||
db.pragma("busy_timeout = 5000");
|
||||
@@ -1866,6 +1874,28 @@ for (const col of ["display_name", "email", "avatar_at"]) {
|
||||
}
|
||||
}
|
||||
|
||||
// An existing database made before auto_vacuum: the setting is stored only by
|
||||
// a VACUUM, which rewrites the whole file under an exclusive lock. Done here,
|
||||
// at startup and under the migration lock, only for a file up to
|
||||
// AUTO_VACUUM_CONVERT_MAX_BYTES (well under a second); a bigger one is
|
||||
// converted by the manual Compact (server/lib/db-retention.js). Best effort: a
|
||||
// second process holding a read open makes it fail, and the next start retries.
|
||||
const AUTO_VACUUM_CONVERT_MAX_BYTES = 32 * 1024 * 1024;
|
||||
function convertAutoVacuum() {
|
||||
try {
|
||||
if (db.pragma("auto_vacuum", { simple: true }) !== 0) return false;
|
||||
const bytes =
|
||||
db.pragma("page_count", { simple: true }) * db.pragma("page_size", { simple: true });
|
||||
if (bytes > AUTO_VACUUM_CONVERT_MAX_BYTES) return false;
|
||||
db.pragma("auto_vacuum = INCREMENTAL");
|
||||
db.exec("VACUUM");
|
||||
return db.pragma("auto_vacuum", { simple: true }) === 2;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
convertAutoVacuum();
|
||||
|
||||
migrationLock.exec("COMMIT");
|
||||
migrationLock.close();
|
||||
|
||||
@@ -2535,7 +2565,9 @@ const stmts = {
|
||||
* on a reader — the other dashboard's), after which the next write truncates
|
||||
* the WAL to `journal_size_limit`; and `PRAGMA optimize`, which refreshes the
|
||||
* planner's statistics only where they went stale. Never VACUUM: it rewrites
|
||||
* the whole file under an exclusive lock.
|
||||
* the whole file under an exclusive lock. The opt-in retention and the
|
||||
* bounded `incremental_vacuum` run after this, from
|
||||
* `server/lib/db-retention.js`'s `runScheduledMaintenance`.
|
||||
*/
|
||||
function maintainDatabase() {
|
||||
const out = {};
|
||||
@@ -2561,4 +2593,6 @@ module.exports = {
|
||||
applyIntroPricing,
|
||||
topUpDefaultPricing,
|
||||
maintainDatabase,
|
||||
convertAutoVacuum,
|
||||
AUTO_VACUUM_CONVERT_MAX_BYTES,
|
||||
};
|
||||
|
||||
+8
-2
@@ -500,11 +500,17 @@ function startBackgroundServices() {
|
||||
.catch(() => {});
|
||||
|
||||
// Hourly DB upkeep: a non-blocking checkpoint (the WAL then shrinks to
|
||||
// journal_size_limit) and PRAGMA optimize. Off with DASHBOARD_DB_MAINTAIN_MS=0.
|
||||
// journal_size_limit), PRAGMA optimize, the opt-in retention
|
||||
// (AGENTHUB_RETENTION_DAYS, off by default) and a bounded incremental_vacuum
|
||||
// (server/lib/db-retention.js). Off with DASHBOARD_DB_MAINTAIN_MS=0.
|
||||
{
|
||||
const every = Number(process.env.DASHBOARD_DB_MAINTAIN_MS ?? 60 * 60 * 1000);
|
||||
if (every > 0) {
|
||||
const t = setInterval(() => require("./db").maintainDatabase(), every);
|
||||
const t = setInterval(() => {
|
||||
require("./lib/db-retention")
|
||||
.runScheduledMaintenance()
|
||||
.catch(() => {});
|
||||
}, every);
|
||||
if (t.unref) t.unref();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
};
|
||||
@@ -11,6 +11,9 @@
|
||||
* - GET /api/settings/backup (Settings)
|
||||
* - PUT /api/settings/backup (Settings)
|
||||
* - POST /api/settings/backup/run (Settings)
|
||||
* - GET /api/settings/retention (Settings)
|
||||
* - PUT /api/settings/retention (Settings)
|
||||
* - POST /api/settings/compact (Settings)
|
||||
* - GET /api/workflows/runs (Workflows)
|
||||
* - GET /api/workflows/runs/{runId} (Workflows)
|
||||
* - GET /api/remote-sources (Remote Sources)
|
||||
@@ -1225,6 +1228,117 @@ const paths = {
|
||||
},
|
||||
},
|
||||
|
||||
"/api/settings/retention": (() => {
|
||||
const state = {
|
||||
description: "Automatic retention setting, database space and the last Compact",
|
||||
content: {
|
||||
"application/json": {
|
||||
schema: {
|
||||
type: "object",
|
||||
required: ["days", "max_days", "env_owned", "compacting"],
|
||||
properties: {
|
||||
days: {
|
||||
type: "integer",
|
||||
minimum: 0,
|
||||
maximum: 3650,
|
||||
description:
|
||||
"Finished sessions older than this many days are purged by the hourly maintenance; 0 = off (the default)",
|
||||
},
|
||||
max_days: { type: "integer" },
|
||||
env_owned: { type: "object", properties: { days: { type: "boolean" } } },
|
||||
last_run: {
|
||||
type: "object",
|
||||
nullable: true,
|
||||
description:
|
||||
"This process's last hourly maintenance: {at, retention?: {purged_sessions, purged_events, purged_agents, error?}, reclaim: {reclaimed_pages, error?}, ms}",
|
||||
},
|
||||
space: {
|
||||
type: "object",
|
||||
nullable: true,
|
||||
properties: {
|
||||
auto_vacuum: { type: "string", enum: ["none", "full", "incremental"] },
|
||||
page_size: { type: "integer" },
|
||||
page_count: { type: "integer" },
|
||||
freelist_pages: { type: "integer" },
|
||||
free_bytes: { type: "integer" },
|
||||
},
|
||||
},
|
||||
db_size: { type: "integer", description: "Database file size in bytes" },
|
||||
compacting: { type: "boolean" },
|
||||
last_compact: {
|
||||
type: "object",
|
||||
nullable: true,
|
||||
description:
|
||||
"This process's last Compact: {at, ok, size_before?, size_after?, ms?, error?}",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
};
|
||||
return {
|
||||
get: {
|
||||
tags: ["Settings"],
|
||||
summary: "Automatic retention and database space",
|
||||
operationId: "getRetention",
|
||||
responses: { 200: state },
|
||||
},
|
||||
put: {
|
||||
tags: ["Settings"],
|
||||
summary: "Set the automatic retention",
|
||||
description:
|
||||
"Persisted to the settings file as AGENTHUB_RETENTION_DAYS. 0 turns it off. Same-origin only; in server mode an admin's.",
|
||||
operationId: "setRetention",
|
||||
requestBody: {
|
||||
required: true,
|
||||
content: {
|
||||
"application/json": {
|
||||
schema: {
|
||||
type: "object",
|
||||
required: ["days"],
|
||||
properties: { days: { type: "integer", minimum: 0, maximum: 3650 } },
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
responses: {
|
||||
200: { ...state, description: "The new state" },
|
||||
400: { description: "INVALID_INPUT: days is not a whole number from 0 to 3650" },
|
||||
403: { description: "EBADORIGIN: another site asked" },
|
||||
409: { description: "EENVOWNED: the environment set AGENTHUB_RETENTION_DAYS" },
|
||||
500: { description: "ESAVE: the settings file could not be written" },
|
||||
},
|
||||
},
|
||||
};
|
||||
})(),
|
||||
|
||||
"/api/settings/compact": {
|
||||
post: {
|
||||
tags: ["Settings"],
|
||||
summary: "Compact the database now (full VACUUM)",
|
||||
description:
|
||||
"Rewrites the database in a worker thread, switching it to incremental auto_vacuum, then truncates the WAL. Hook writes wait while it runs. Answers at once; poll GET /api/settings/retention while `compacting`. Same-origin only; in server mode an admin's.",
|
||||
operationId: "compactDatabase",
|
||||
requestBody: {
|
||||
required: true,
|
||||
content: {
|
||||
"application/json": {
|
||||
schema: {
|
||||
type: "object",
|
||||
required: ["confirm"],
|
||||
properties: { confirm: { type: "boolean", enum: [true] } },
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
responses: {
|
||||
202: { description: "Started (or already running); the retention state" },
|
||||
400: { description: "ECONFIRM: confirm: true is missing" },
|
||||
403: { description: "EBADORIGIN: another site asked" },
|
||||
},
|
||||
},
|
||||
},
|
||||
|
||||
"/api/settings/codex-hooks": (() => {
|
||||
const status = {
|
||||
description: "Where AgentHub's Codex hooks stand",
|
||||
|
||||
+66
-41
@@ -1,5 +1,5 @@
|
||||
/**
|
||||
* @file Express router for settings-related endpoints, providing system info, database statistics, hook status, and operations to clear data, re-import sessions, reinstall hooks, reset pricing, export data, and perform cleanup of stale sessions, plus the privacy controls (secret redaction, ignored folders), the push-ingest token and the daily database backup (server/lib/db-backup.js). This allows the frontend to manage and maintain the agent monitoring system effectively.
|
||||
* @file Express router for settings-related endpoints, providing system info, database statistics, hook status, and operations to clear data, re-import sessions, reinstall hooks, reset pricing, export data, and perform cleanup of stale sessions, plus the privacy controls (secret redaction, ignored folders), the push-ingest token, the daily database backup (server/lib/db-backup.js) and the opt-in retention + Compact (server/lib/db-retention.js). This allows the frontend to manage and maintain the agent monitoring system effectively.
|
||||
* @author Nguyễn Ngọc Trí Vĩ <vinnt@smartgift.vn>
|
||||
*/
|
||||
|
||||
@@ -28,6 +28,7 @@ const { getCodexHome, setCodexHome } = require("../lib/codex-home");
|
||||
const { envOwned, applySetting } = require("../lib/settings-env");
|
||||
const privacy = require("../lib/privacy");
|
||||
const dbBackup = require("../lib/db-backup");
|
||||
const retention = require("../lib/db-retention");
|
||||
const pushToken = require("../lib/push-token");
|
||||
const pricingSync = require("../lib/pricing-sync");
|
||||
const {
|
||||
@@ -678,10 +679,66 @@ function pushTokenWrite(fn) {
|
||||
router.post("/push-token", sameOriginGuard, pushTokenWrite(pushToken.rotatePushToken));
|
||||
router.delete("/push-token", sameOriginGuard, pushTokenWrite(pushToken.revokePushToken));
|
||||
|
||||
// GET/PUT /api/settings/retention, POST /api/settings/compact — database size
|
||||
// upkeep (server/lib/db-retention.js). Retention is OFF by default: `days` 0
|
||||
// turns it off, 1–3650 makes the hourly maintenance purge finished sessions
|
||||
// older than that, the manual purge's rule. It persists to the settings file;
|
||||
// one the environment set is `env_owned` and refused. Compact (a full VACUUM
|
||||
// in a worker) answers 202 and the page polls GET while `compacting`.
|
||||
function retentionState() {
|
||||
let space = null;
|
||||
try {
|
||||
space = retention.spaceInfo(db);
|
||||
} catch {
|
||||
/* pragma unavailable: the card shows no numbers */
|
||||
}
|
||||
return {
|
||||
...retention.retentionConfig(),
|
||||
max_days: retention.MAX_DAYS,
|
||||
env_owned: { days: envOwned.has(retention.KEYS.days) },
|
||||
last_run: retention.lastMaintenance(),
|
||||
space,
|
||||
db_size: getDbSize(),
|
||||
compacting: retention.isCompacting(),
|
||||
last_compact: retention.lastCompactResult(),
|
||||
};
|
||||
}
|
||||
router.get("/retention", (_req, res) => res.json(retentionState()));
|
||||
router.put("/retention", sameOriginGuard, (req, res) => {
|
||||
const days = req.body?.days;
|
||||
if (!Number.isInteger(days) || days < 0 || days > retention.MAX_DAYS) {
|
||||
return res.status(400).json({
|
||||
error: {
|
||||
code: "INVALID_INPUT",
|
||||
message: `days must be a whole number from 0 (off) to ${retention.MAX_DAYS}`,
|
||||
},
|
||||
});
|
||||
}
|
||||
if (envOwned.has(retention.KEYS.days)) {
|
||||
return res.status(409).json({
|
||||
error: { code: "EENVOWNED", message: `${retention.KEYS.days} is set by the environment` },
|
||||
});
|
||||
}
|
||||
try {
|
||||
applySetting(retention.KEYS.days, days === 0 ? "" : String(days));
|
||||
} catch (err) {
|
||||
return res.status(500).json({ error: { code: "ESAVE", message: err.message } });
|
||||
}
|
||||
res.json({ ok: true, ...retentionState() });
|
||||
});
|
||||
router.post("/compact", sameOriginGuard, (req, res) => {
|
||||
if (req.body?.confirm !== true) {
|
||||
return res.status(400).json({
|
||||
error: { code: "ECONFIRM", message: "compacting needs confirm: true" },
|
||||
});
|
||||
}
|
||||
retention.runCompact().catch(() => {}); // the outcome is `last_compact` in GET
|
||||
res.status(202).json({ ok: true, ...retentionState() });
|
||||
});
|
||||
|
||||
// POST /api/settings/cleanup — abandon stale sessions, purge old data.
|
||||
// `purge_days` is a whole number ≥ 1 (0.1 days would purge almost everything);
|
||||
// `dry_run: true` answers the same counts and changes nothing.
|
||||
const PURGE_CHUNK = 500;
|
||||
router.post("/cleanup", (req, res) => {
|
||||
const { abandon_hours, purge_days } = req.body;
|
||||
const dryRun = req.body.dry_run === true;
|
||||
@@ -739,11 +796,7 @@ router.post("/cleanup", (req, res) => {
|
||||
if (purge_days && typeof purge_days === "number" && purge_days > 0) {
|
||||
const cutoff = new Date(Date.now() - purge_days * 86400 * 1000).toISOString();
|
||||
// Only purge completed/error/abandoned sessions, never active
|
||||
const toDelete = db
|
||||
.prepare(
|
||||
"SELECT id FROM sessions WHERE status IN ('completed','error','abandoned') AND started_at < ?"
|
||||
)
|
||||
.all(cutoff);
|
||||
const toDelete = retention.purgeableBefore(db, cutoff);
|
||||
|
||||
if (dryRun) {
|
||||
// Exactly the ids the real run deletes: those already purgeable, plus
|
||||
@@ -752,43 +805,15 @@ router.post("/cleanup", (req, res) => {
|
||||
...toDelete.map((r) => r.id),
|
||||
...wouldAbandon.filter((r) => r.started_at < cutoff).map((r) => r.id),
|
||||
];
|
||||
for (let i = 0; i < ids.length; i += PURGE_CHUNK) {
|
||||
const chunk = ids.slice(i, i + PURGE_CHUNK);
|
||||
const placeholders = chunk.map(() => "?").join(",");
|
||||
result.purged_events += db
|
||||
.prepare(`SELECT COUNT(*) AS n FROM events WHERE session_id IN (${placeholders})`)
|
||||
.get(...chunk).n;
|
||||
result.purged_agents += db
|
||||
.prepare(`SELECT COUNT(*) AS n FROM agents WHERE session_id IN (${placeholders})`)
|
||||
.get(...chunk).n;
|
||||
}
|
||||
result.purged_sessions = ids.length;
|
||||
Object.assign(result, retention.countSessionIds(db, ids));
|
||||
return res.json({ ok: true, ...result });
|
||||
}
|
||||
|
||||
// One transaction, in chunks: a single IN (...) list of every id would
|
||||
// pass SQLite's bound-variable limit on a large history.
|
||||
db.transaction(() => {
|
||||
const ids = toDelete.map((r) => r.id);
|
||||
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
|
||||
result.purged_events += db
|
||||
.prepare(`DELETE FROM events WHERE session_id IN (${placeholders})`)
|
||||
.run(...chunk).changes;
|
||||
result.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);
|
||||
db.prepare(`DELETE FROM sessions WHERE id IN (${placeholders})`).run(...chunk);
|
||||
}
|
||||
result.purged_sessions = ids.length;
|
||||
})();
|
||||
// One transaction; the deletion itself is shared with the automatic
|
||||
// retention (server/lib/db-retention.js).
|
||||
const ids = toDelete.map((r) => r.id);
|
||||
const purged = db.transaction(() => retention.purgeSessionIds(db, ids))();
|
||||
Object.assign(result, purged, { purged_sessions: ids.length });
|
||||
}
|
||||
|
||||
res.json({ ok: true, ...result });
|
||||
|
||||
+78
-10
@@ -1,5 +1,11 @@
|
||||
/**
|
||||
* @file Periodic git upstream check and WebSocket broadcast when update availability changes.
|
||||
* @file Periodic release/upstream check and WebSocket broadcast when update
|
||||
* availability changes. Hourly by default (`DASHBOARD_UPDATE_CHECK_INTERVAL_MS`,
|
||||
* floor 60 s); a tick with no browser connected is skipped — a checkout install
|
||||
* runs `git fetch` on every check, and nobody would see the answer — and the
|
||||
* first browser to connect after a skipped or stale check gets one promptly.
|
||||
* The manual "Check now" (`POST /api/updates/check`, `agenthub update-check`)
|
||||
* is not routed through here and always runs.
|
||||
* @author Nguyễn Ngọc Trí Vĩ <vinnt@smartgift.vn>
|
||||
*/
|
||||
|
||||
@@ -13,11 +19,39 @@ function isUpdateCheckDisabled() {
|
||||
function intervalMs() {
|
||||
const n = Number.parseInt(process.env.DASHBOARD_UPDATE_CHECK_INTERVAL_MS || "", 10);
|
||||
if (Number.isFinite(n) && n >= 60_000) return n;
|
||||
return 5 * 60 * 1000;
|
||||
return 60 * 60 * 1000;
|
||||
}
|
||||
|
||||
function startUpdateScheduler({ broadcast }) {
|
||||
/** Delay before the catch-up check a reconnecting browser triggers. */
|
||||
const CONNECT_CHECK_DELAY_MS = 3_000;
|
||||
|
||||
/**
|
||||
* @param {object} opts
|
||||
* @param {Function} opts.broadcast
|
||||
* @param {() => number} [opts.getConnectionCount] browsers connected now
|
||||
* (default: server/websocket.js's); 0 skips a scheduled tick.
|
||||
* @param {(fn: Function) => Function} [opts.onConnection] subscribe to new
|
||||
* browser connections; returns an unsubscribe.
|
||||
* @param {Function} [opts.getStatus] the check itself (tests inject it).
|
||||
* @param {number} [opts.initialDelayMs] first tick after start (default 8 s).
|
||||
*/
|
||||
function startUpdateScheduler({
|
||||
broadcast,
|
||||
getConnectionCount,
|
||||
onConnection,
|
||||
getStatus = getUpdatesStatus,
|
||||
initialDelayMs = 8_000,
|
||||
}) {
|
||||
if (isUpdateCheckDisabled()) return { stop: () => {} };
|
||||
if (!getConnectionCount || !onConnection) {
|
||||
const ws = require("./websocket");
|
||||
getConnectionCount = getConnectionCount || ws.getConnectionCount;
|
||||
onConnection = onConnection || ws.onConnection;
|
||||
}
|
||||
const every = intervalMs();
|
||||
let lastCheckAt = 0;
|
||||
let inFlight = false;
|
||||
let connectTimer = null;
|
||||
|
||||
let lastFingerprint = "";
|
||||
let lastHadUpdate = false;
|
||||
@@ -25,10 +59,24 @@ function startUpdateScheduler({ broadcast }) {
|
||||
let initialTimer = null;
|
||||
let intervalTimer = null;
|
||||
|
||||
async function tick() {
|
||||
// A scheduled tick: skipped while no browser is connected.
|
||||
function scheduledTick() {
|
||||
if (stopped) return;
|
||||
let clients = 1;
|
||||
try {
|
||||
const status = await getUpdatesStatus();
|
||||
clients = getConnectionCount();
|
||||
} catch {
|
||||
/* unknown: check as before */
|
||||
}
|
||||
if (clients > 0) tick();
|
||||
}
|
||||
|
||||
async function tick() {
|
||||
if (stopped || inFlight) return;
|
||||
inFlight = true;
|
||||
lastCheckAt = Date.now();
|
||||
try {
|
||||
const status = await getStatus();
|
||||
// Fingerprint must change whenever anything user-visible changes,
|
||||
// including the *shape* of manual_command (e.g. user switches branch
|
||||
// or adds an upstream remote — same remote_sha / commits_behind, but
|
||||
@@ -72,15 +120,33 @@ function startUpdateScheduler({ broadcast }) {
|
||||
}
|
||||
} catch {
|
||||
// Non-fatal — never block the server on update checks
|
||||
} finally {
|
||||
inFlight = false;
|
||||
}
|
||||
}
|
||||
|
||||
initialTimer = setTimeout(() => {
|
||||
tick();
|
||||
}, 8_000);
|
||||
// A browser came back after the scheduled checks were skipped (or the last
|
||||
// one is older than the interval): check once, shortly, so the banner is
|
||||
// current without waiting up to an hour.
|
||||
let unsubscribe = null;
|
||||
try {
|
||||
unsubscribe = onConnection(() => {
|
||||
if (stopped || connectTimer || inFlight) return;
|
||||
if (Date.now() - lastCheckAt < every) return;
|
||||
connectTimer = setTimeout(() => {
|
||||
connectTimer = null;
|
||||
tick();
|
||||
}, CONNECT_CHECK_DELAY_MS);
|
||||
if (typeof connectTimer.unref === "function") connectTimer.unref();
|
||||
});
|
||||
} catch {
|
||||
/* no WebSocket server: scheduled ticks only */
|
||||
}
|
||||
|
||||
initialTimer = setTimeout(scheduledTick, initialDelayMs);
|
||||
if (typeof initialTimer.unref === "function") initialTimer.unref();
|
||||
|
||||
intervalTimer = setInterval(tick, intervalMs());
|
||||
intervalTimer = setInterval(scheduledTick, every);
|
||||
if (typeof intervalTimer.unref === "function") intervalTimer.unref();
|
||||
|
||||
return {
|
||||
@@ -88,8 +154,10 @@ function startUpdateScheduler({ broadcast }) {
|
||||
stopped = true;
|
||||
if (initialTimer) clearTimeout(initialTimer);
|
||||
if (intervalTimer) clearInterval(intervalTimer);
|
||||
if (connectTimer) clearTimeout(connectTimer);
|
||||
if (typeof unsubscribe === "function") unsubscribe();
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
module.exports = { startUpdateScheduler };
|
||||
module.exports = { startUpdateScheduler, intervalMs };
|
||||
|
||||
@@ -48,6 +48,13 @@ function initWebSocket(server) {
|
||||
});
|
||||
|
||||
wss.on("connection", (ws, req) => {
|
||||
for (const fn of connectionListeners) {
|
||||
try {
|
||||
fn(ws);
|
||||
} catch {
|
||||
/* a listener never breaks a connection */
|
||||
}
|
||||
}
|
||||
// Server mode: who this socket belongs to, for broadcast filtering, and
|
||||
// the cookie to re-check it by (a logout or a disabled user ends it).
|
||||
if (isServerMode()) {
|
||||
@@ -252,6 +259,16 @@ function broadcast(type, data) {
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Subscribe to new `/ws` browser connections (the update scheduler checks once
|
||||
* when a browser returns after its ticks were skipped). Returns an unsubscribe.
|
||||
*/
|
||||
const connectionListeners = new Set();
|
||||
function onConnection(fn) {
|
||||
connectionListeners.add(fn);
|
||||
return () => connectionListeners.delete(fn);
|
||||
}
|
||||
|
||||
function getConnectionCount() {
|
||||
if (!wss) return 0;
|
||||
let count = 0;
|
||||
@@ -306,6 +323,7 @@ module.exports = {
|
||||
initPtyWebSocket,
|
||||
broadcast,
|
||||
getConnectionCount,
|
||||
onConnection,
|
||||
closeWebSocket,
|
||||
sizeFromQuery,
|
||||
};
|
||||
|
||||
Reference in New Issue
Block a user