openvibers/openvibe.ai Commits
Billing readings never sent expire with their run (the outbox grew 4,000 a day with no relay) (#139)
Changes
@@@ -73,7 +73,9 @@ async function start({ config, db: givenDb = null, clock = { now: () => Date.nowtry {const pruned = await runs.prune();const expired = await cache.prune();-- if (pruned || expired) log.log(`[ai] retention: ${pruned} old run(s), ${expired} expired cache entr${expired === 1 ? 'y' : 'ies'}`);++ // Billing readings never sent go with their runs (server/usage-samples.js expireUnsent).++ const lapsed = await require('./usage-samples').expireUnsent(db, { days: config.runs.retentionDays, now: clock.now() });++ if (pruned || expired || lapsed) log.log(`[ai] retention: ${pruned} old run(s), ${expired} expired cache entr${expired === 1 ? 'y' : 'ies'}, ${lapsed} unsent usage reading(s)`);} catch (err) { log.error(`[ai] retention failed: ${err.message}`); }}, 60 * 60 * 1000);housekeeping.unref?.();@@@ -133,6 +133,18 @@ async function readingsFor(db, usageSampleIds) {return out;});}++/**++ * Readings Billing never took expire with their run (runs.prune, AI_RUN_RETENTION_DAYS): a reading lives no longer than++ * the run it explains. Until Billing rates AI's metrics and the relay is configured, every reading waits here; on++ * 2026-10-23 that was 80,793 rows growing 4,000 a day, nearly all OpenVibe's own sites (service:live), worth about $3++ * (plan E2). The run's tokens and cost stay in runs and usage_daily. Sent rows are the relay's own prune; rows Billing++ * refused stay for the console. → rows removed++ */++async function expireUnsent(db, { days, now = Date.now() } = {}) {++ if (!(Number(days) > 0)) return 0;++ const r = await db.prepare(`DELETE FROM ${TABLE} WHERE sent_at IS NULL AND rejected_at IS NULL AND created_at < ?`).run(now - Number(days) * 86400000);++ return Number(r.changes) || 0;++}/** The run a reading key names: `ai:<run id>:<attempt>:<kind>` (the run id is the second segment). */const runIdOf = (key) => { const p = String(key || '').split(':'); return p[0] === 'ai' && p[1] ? p[1] : null; };/** Staff console: readings by delivery state and the latest rows carrying a send error (never a public page: keys name runs). */@@@ -151,4 +163,4 @@ async function stop({ drainMs = 3000 } = {}) {if (r) { let timer; await Promise.race([r.flush().catch(() => {}), new Promise((res) => { timer = setTimeout(res, drainMs); if (timer.unref) timer.unref(); })]); clearTimeout(timer); await r.stop(); }}function _reset() { if (pruneTimer) clearInterval(pruneTimer); pruneTimer = null; if (relay) relay.stop(); relay = null; on = false; queued = 0; lastError = null; }--module.exports = { IMAGE_KIND, init, record, sampleOf, sampleFromUsage, keyOf, runIdOf, readingsFor, deliveryReport, status, stop, _reset, TABLE, KINDS, UNIT };++module.exports = { IMAGE_KIND, init, record, sampleOf, sampleFromUsage, keyOf, runIdOf, readingsFor, deliveryReport, expireUnsent, status, stop, _reset, TABLE, KINDS, UNIT };@@@ -202,6 +202,21 @@ t.test('a reading Billing refuses is failed, with its error, in explain and on tassert.deepStrictEqual([f.run_id, f.state], [r.body.run.id, 'failed']);});++t.test('a reading Billing never took expires with its run; sent, refused and recent rows stay', async () => {++ const DAY = 86400000;++ const now = Date.UTC(2026, 9, 23);++ const put = (id, ageDays, extra = {}) => h.db.prepare(`INSERT INTO usage_sample_outbox (event_id, envelope, created_at, sent_at, rejected_at) VALUES (?, ?, ?, ?, ?)`)++ .run(id, JSON.stringify({ id }), now - ageDays * DAY, extra.sent ? now : null, extra.rejected ? now : null);++ await put('ai:run_old:0:in', 31);++ await put('ai:run_old_sent:0:in', 31, { sent: true });++ await put('ai:run_old_refused:0:in', 31, { rejected: true });++ await put('ai:run_new:0:in', 29);++ assert.strictEqual(await usageSamples.expireUnsent(h.db, { days: 30, now }), 1);++ const left = new Set((await h.db.prepare("SELECT event_id FROM usage_sample_outbox WHERE event_id LIKE 'ai:run_old%' OR event_id = 'ai:run_new:0:in'").all()).map((r) => r.event_id));++ assert.deepStrictEqual([...left].sort(), ['ai:run_new:0:in', 'ai:run_old_refused:0:in', 'ai:run_old_sent:0:in']);++ assert.strictEqual(await usageSamples.expireUnsent(h.db, { days: 0, now }), 0, 'no retention configured: nothing goes');++});++t.test('shutdown', async () => { usageSamples._reset(); await h.stop(); await seam.close(); });t.run();
@@ -73,7 +73,9 @@ async function start({ config, db: givenDb = null, clock = { now: () => Date.now
@@ -73,7 +73,9 @@ async function start({ config, db: givenDb = null, clock = { now: () => Date.now
try {
try {
const pruned = await runs.prune();
const pruned = await runs.prune();
const expired = await cache.prune();
const expired = await cache.prune();
- if (pruned || expired) log.log(`[ai] retention: ${pruned} old run(s), ${expired} expired cache entr${expired === 1 ? 'y' : 'ies'}`);
+ // Billing readings never sent go with their runs (server/usage-samples.js expireUnsent).
+ const lapsed = await require('./usage-samples').expireUnsent(db, { days: config.runs.retentionDays, now: clock.now() });
+ if (pruned || expired || lapsed) log.log(`[ai] retention: ${pruned} old run(s), ${expired} expired cache entr${expired === 1 ? 'y' : 'ies'}, ${lapsed} unsent usage reading(s)`);
} catch (err) { log.error(`[ai] retention failed: ${err.message}`); }
} catch (err) { log.error(`[ai] retention failed: ${err.message}`); }
}, 60 * 60 * 1000);
}, 60 * 60 * 1000);
housekeeping.unref?.();
housekeeping.unref?.();
@@ -133,6 +133,18 @@ async function readingsFor(db, usageSampleIds) {
@@ -133,6 +133,18 @@ async function readingsFor(db, usageSampleIds) {
return out;
return out;
});
});
}
}
+/**
+ * Readings Billing never took expire with their run (runs.prune, AI_RUN_RETENTION_DAYS): a reading lives no longer than
+ * the run it explains. Until Billing rates AI's metrics and the relay is configured, every reading waits here; on
+ * 2026-10-23 that was 80,793 rows growing 4,000 a day, nearly all OpenVibe's own sites (service:live), worth about $3
+ * (plan E2). The run's tokens and cost stay in runs and usage_daily. Sent rows are the relay's own prune; rows Billing
+ * refused stay for the console. → rows removed
+ */
+async function expireUnsent(db, { days, now = Date.now() } = {}) {
+ if (!(Number(days) > 0)) return 0;
+ const r = await db.prepare(`DELETE FROM ${TABLE} WHERE sent_at IS NULL AND rejected_at IS NULL AND created_at < ?`).run(now - Number(days) * 86400000);
+ return Number(r.changes) || 0;
+}
/** The run a reading key names: `ai:<run id>:<attempt>:<kind>` (the run id is the second segment). */
/** The run a reading key names: `ai:<run id>:<attempt>:<kind>` (the run id is the second segment). */
const runIdOf = (key) => { const p = String(key || '').split(':'); return p[0] === 'ai' && p[1] ? p[1] : null; };
const runIdOf = (key) => { const p = String(key || '').split(':'); return p[0] === 'ai' && p[1] ? p[1] : null; };
/** Staff console: readings by delivery state and the latest rows carrying a send error (never a public page: keys name runs). */
/** Staff console: readings by delivery state and the latest rows carrying a send error (never a public page: keys name runs). */
@@ -151,4 +163,4 @@ async function stop({ drainMs = 3000 } = {}) {
@@ -151,4 +163,4 @@ async function stop({ drainMs = 3000 } = {}) {
if (r) { let timer; await Promise.race([r.flush().catch(() => {}), new Promise((res) => { timer = setTimeout(res, drainMs); if (timer.unref) timer.unref(); })]); clearTimeout(timer); await r.stop(); }
if (r) { let timer; await Promise.race([r.flush().catch(() => {}), new Promise((res) => { timer = setTimeout(res, drainMs); if (timer.unref) timer.unref(); })]); clearTimeout(timer); await r.stop(); }
}
}
function _reset() { if (pruneTimer) clearInterval(pruneTimer); pruneTimer = null; if (relay) relay.stop(); relay = null; on = false; queued = 0; lastError = null; }
function _reset() { if (pruneTimer) clearInterval(pruneTimer); pruneTimer = null; if (relay) relay.stop(); relay = null; on = false; queued = 0; lastError = null; }
-module.exports = { IMAGE_KIND, init, record, sampleOf, sampleFromUsage, keyOf, runIdOf, readingsFor, deliveryReport, status, stop, _reset, TABLE, KINDS, UNIT };
+module.exports = { IMAGE_KIND, init, record, sampleOf, sampleFromUsage, keyOf, runIdOf, readingsFor, deliveryReport, expireUnsent, status, stop, _reset, TABLE, KINDS, UNIT };
@@ -202,6 +202,21 @@ t.test('a reading Billing refuses is failed, with its error, in explain and on t
@@ -202,6 +202,21 @@ t.test('a reading Billing refuses is failed, with its error, in explain and on t
assert.deepStrictEqual([f.run_id, f.state], [r.body.run.id, 'failed']);
assert.deepStrictEqual([f.run_id, f.state], [r.body.run.id, 'failed']);
});
});
+t.test('a reading Billing never took expires with its run; sent, refused and recent rows stay', async () => {
+ const DAY = 86400000;
+ const now = Date.UTC(2026, 9, 23);
+ const put = (id, ageDays, extra = {}) => h.db.prepare(`INSERT INTO usage_sample_outbox (event_id, envelope, created_at, sent_at, rejected_at) VALUES (?, ?, ?, ?, ?)`)
+ .run(id, JSON.stringify({ id }), now - ageDays * DAY, extra.sent ? now : null, extra.rejected ? now : null);
+ await put('ai:run_old:0:in', 31);
+ await put('ai:run_old_sent:0:in', 31, { sent: true });
+ await put('ai:run_old_refused:0:in', 31, { rejected: true });
+ await put('ai:run_new:0:in', 29);
+ assert.strictEqual(await usageSamples.expireUnsent(h.db, { days: 30, now }), 1);
+ const left = new Set((await h.db.prepare("SELECT event_id FROM usage_sample_outbox WHERE event_id LIKE 'ai:run_old%' OR event_id = 'ai:run_new:0:in'").all()).map((r) => r.event_id));
+ assert.deepStrictEqual([...left].sort(), ['ai:run_new:0:in', 'ai:run_old_refused:0:in', 'ai:run_old_sent:0:in']);
+ assert.strictEqual(await usageSamples.expireUnsent(h.db, { days: 0, now }), 0, 'no retention configured: nothing goes');
+});
+
t.test('shutdown', async () => { usageSamples._reset(); await h.stop(); await seam.close(); });
t.test('shutdown', async () => { usageSamples._reset(); await h.stop(); await seam.close(); });
t.run();
t.run();