From 3183f2cb8869ce2201491efd050e97308840ac3a Mon Sep 17 00:00:00 2001 From: Yoni Davidson Date: Sun, 9 Aug 2026 10:55:36 +0300 Subject: [PATCH] perf: bound the daemon's warm mirror to recent history (#167) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The daemon mirrored the entire bus — every key AND every body — and refreshed it on every poll. On a snapshot-capable backend that is one round trip, but it read every blob in the store each time: pending mail, the roster, the whole 30-day archive, and every telemetry batch. The cost of a poll therefore grew with all history, forever, on a bus where nothing but the last few messages is hot. The mirror now holds what is worth holding. `agents/` is re-read every poll (records mutate); `inbox/` is always warm; `read/` and `events/` are warm only inside a window (AGENTCOMM_MIRROR_HISTORY_MS, default 7 days). Older history keeps its KEY, so list, log, purge and channel discovery see the whole store exactly as before, and a cold body loads on demand through the `get` passthrough. Message blobs are immutable, so a body already held is never re-read either: steady state is "the roster plus whatever is new". `Snapshottable.snapshot` grew a body filter and now reports the keys it saw alongside the bodies it read — listing names is cheap, reading blobs is not. --- src/backends/git.ts | 21 +++++++++----- src/cli.ts | 4 +++ src/daemon.ts | 64 +++++++++++++++++++++++++++++------------ src/types.ts | 15 ++++++++-- test/daemon.e2e.test.ts | 27 +++++++++++++++++ test/git.e2e.test.ts | 27 ++++++++++++++--- 6 files changed, 127 insertions(+), 31 deletions(-) diff --git a/src/backends/git.ts b/src/backends/git.ts index 410258a9..91e65bd5 100644 --- a/src/backends/git.ts +++ b/src/backends/git.ts @@ -244,19 +244,26 @@ export class GitBackend implements Backend, Batchable, Claimable, Snapshottable } /** - * One fetch, then every body read from local objects at that tip — the - * daemon's warm path. Key-by-key `get` would pay a fetch per key. + * One fetch, then bodies read from local objects at that tip — the daemon's + * warm path. Key-by-key `get` would pay a fetch per key. Listing names is + * cheap (`ls-tree`); reading blobs is not, so `opts.bodies` picks which + * ones are worth the batch (issue #167). */ - async snapshot(prefix: string): Promise> { + async snapshot( + prefix: string, + opts?: { bodies?: (key: string) => boolean }, + ): Promise<{ keys: string[]; bodies: Map }> { const out = new Map(); const tip = await this.tip(); - if (tip === null) return out; + if (tip === null) return { keys: [], bodies: out }; const full = this.k(prefix); - const names = (await this.git(['ls-tree', '-r', '--name-only', '-z', tip])) + const all = (await this.git(['ls-tree', '-r', '--name-only', '-z', tip])) .toString('utf8') .split('\0') .filter((p) => p.length > 0 && p.startsWith(full) && p.startsWith(this.keyPrefix)); - if (names.length === 0) return out; + const keys = all.map((p) => p.slice(this.keyPrefix.length)); + const names = opts?.bodies ? all.filter((p) => opts.bodies!(p.slice(this.keyPrefix.length))) : all; + if (names.length === 0) return { keys, bodies: out }; const batch = await this.git(['cat-file', '--batch'], { input: Buffer.from(names.map((p) => `${tip}:${p}`).join('\n') + '\n'), }); @@ -272,7 +279,7 @@ export class GitBackend implements Backend, Batchable, Claimable, Snapshottable out.set(p.slice(this.keyPrefix.length), Buffer.from(batch.subarray(off, off + size))); off += size + 1; // body + trailing newline } - return out; + return { keys, bodies: out }; } async list(prefix: string): Promise { diff --git a/src/cli.ts b/src/cli.ts index 796b222e..31cd6aef 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -158,6 +158,10 @@ Env: stderr) AGENTCOMM_DAEMON=1|0 Default all commands through / away from the daemon AGENTCOMM_POLL_MS Daemon remote-poll interval (default 10000) + AGENTCOMM_MIRROR_HISTORY_MS How much archive/telemetry history the daemon + keeps warm (default 7d). Older keys stay listable + and readable; their bodies just load on demand, + so a poll costs what is hot, not all history AGENTCOMM_BACKEND_PLUGINS comma/whitespace-separated module specifiers to import before resolving --backend, so a third-party package can register a new URI diff --git a/src/daemon.ts b/src/daemon.ts index 845c553a..5a6002a2 100644 --- a/src/daemon.ts +++ b/src/daemon.ts @@ -180,11 +180,38 @@ export async function runDaemon(uri: string): Promise { }); } - // warm mirror: full key set + bodies. On Snapshottable backends every poll - // is one round trip (bodies included); elsewhere the poll carries keys only - // and bodies fill lazily through the `get` passthrough — message blobs are - // immutable, and agent records (which mutate: heartbeats) are invalidated - // each poll so the next read refetches them. + // warm mirror: the full key set, plus the bodies worth holding. Message + // blobs are immutable, so a body already held is never re-read; agent + // records (which mutate: heartbeats) are refreshed every poll. + // + // What is NOT held warm is history (issue #167). A bus only grows, and + // re-reading every archived message and telemetry batch on every poll made + // the daemon's cost scale with all history instead of with what is hot — + // forever, on a store where nothing but the last few messages is live. + // Archives and events older than the window keep their KEY (so list, log, + // purge and channel discovery see the whole store, exactly as before) and + // their body fills on demand through the `get` passthrough. + const historyMs = Math.max(0, Number(process.env.AGENTCOMM_MIRROR_HISTORY_MS ?? 7 * 24 * 3600_000)); + const HOT_PREFIXES = ['agents/', 'inbox/']; + /** + * ms timestamp encoded in an archive/event key, or null when it has none. + * Both layouts put a zero-padded, monotonic ms prefix on the filename: + * read//-_.json and events/-_.json. + */ + const keyTime = (key: string): number | null => { + const m = /^0*(\d+)-/.exec(key.slice(key.lastIndexOf('/') + 1)); + return m ? Number(m[1]) : null; + }; + const withinWindow = (key: string): boolean => { + const ts = keyTime(key); + return ts === null || Date.now() - ts <= historyMs; + }; + /** Should the mirror hold this key's body at all? */ + const worthWarming = (key: string): boolean => HOT_PREFIXES.some((p) => key.startsWith(p)) || withinWindow(key); + /** Should THIS poll read it? Not if we already hold an immutable body. */ + const needsRead = (key: string): boolean => + key.startsWith('agents/') || (worthWarming(key) && !mirror.has(key)); + const mirror = new Map(); let keys = new Set(); @@ -237,23 +264,24 @@ export async function runDaemon(uri: string): Promise { async function doPoll(): Promise { const spooled = await snapshotSpool(); + // Bodies this poll actually read (fresh agent records + keys we do not + // already hold), merged onto the ones already warm. + const fetched = new Map(); + let next: Set; if (isSnapshottable(backend)) { - const snap = await backend.snapshot(''); - const present = new Set(snap.keys()); - replaySpool(spooled, snap, present); - for (const k of snap.keys()) if (!present.has(k)) snap.delete(k); - mirror.clear(); - for (const [k, v] of snap) mirror.set(k, v); - keys = present; - return; + const snap = await backend.snapshot('', { bodies: needsRead }); + next = new Set(snap.keys); + for (const [k, v] of snap.bodies) fetched.set(k, v); + } else { + next = new Set(await backend.list('')); } - const bodies = new Map(); - const next = new Set(await backend.list('')); - replaySpool(spooled, bodies, next); + replaySpool(spooled, fetched, next); for (const k of mirror.keys()) { - if (!next.has(k) || k.startsWith('agents/')) mirror.delete(k); + // drop what the store no longer has, what a fresher read supersedes, + // and history that has aged out of the warm window + if (!next.has(k) || k.startsWith('agents/') || !worthWarming(k)) mirror.delete(k); } - for (const [k, v] of bodies) if (next.has(k)) mirror.set(k, v); + for (const [k, v] of fetched) if (next.has(k)) mirror.set(k, v); keys = next; } // Coalesce: timer ticks, `refresh`, and post-claim polls that overlap a diff --git a/src/types.ts b/src/types.ts index 0c12b1e2..8726a7e7 100644 --- a/src/types.ts +++ b/src/types.ts @@ -73,8 +73,19 @@ export interface Claimable { * clock — the bus daemon uses this to warm its mirror in one round trip. */ export interface Snapshottable { - /** All keys under `prefix` with their bodies, read at a single consistent point. */ - snapshot(prefix: string): Promise>; + /** + * Every key under `prefix` at a single consistent point, with the bodies + * `opts.bodies` selects (all of them when it is absent). + * + * The split matters because a bus's history only grows: re-reading every + * archived body on every poll makes the daemon cost scale with all history + * rather than with what is hot (issue #167). Keys are cheap to list, so + * callers still see the whole store and fetch a cold body on demand. + */ + snapshot( + prefix: string, + opts?: { bodies?: (key: string) => boolean }, + ): Promise<{ keys: string[]; bodies: Map }>; } /** diff --git a/test/daemon.e2e.test.ts b/test/daemon.e2e.test.ts index bf09d62a..3f8cda68 100644 --- a/test/daemon.e2e.test.ts +++ b/test/daemon.e2e.test.ts @@ -313,6 +313,33 @@ describe('bus daemon: same semantics, immediate answers', () => { expect(archived).toHaveLength(3); }); + it('keeps only recent history warm, but still serves the old (issue #167)', async () => { + const dir = await mkTmp(); + // an archived message from long ago, and a fresh one — both on the store + const archive = path.join(dir, '.bus', 'read', 'alpha'); + await fs.mkdir(archive, { recursive: true }); + const oldTs = String(Date.now() - 30 * 24 * 3600_000).padStart(15, '0'); + await fs.writeFile( + path.join(archive, `${oldTs}-000000_ancient.json`), + JSON.stringify({ id: 'ancient', from: 'beta', to: 'alpha', body: 'from a month ago', ts: new Date(0).toISOString() }), + ); + + const WINDOW = { AGENTCOMM_MIRROR_HISTORY_MS: String(7 * 24 * 3600_000) }; + await run(['register', '--as', 'alpha', '--daemon'], dir, WINDOW); + await run(['send', 'alpha', 'today', '--as', 'beta', '--daemon', '--sync'], dir, WINDOW); + + // the aged-out archive is still a first-class key: log finds it and reads + // its body through the passthrough, exactly as if it were warm + const log = await run(['log', '--limit', '10', '--daemon', '--json'], dir, WINDOW); + const bodies = (JSON.parse(log.stdout) as { body: string }[]).map((m) => m.body); + expect(bodies).toContain('from a month ago'); + expect(bodies).toContain('today'); + + // and purge still sees it (list is unaffected by what the mirror holds) + const purge = await run(['purge', '--older-than', '14d', '--dry-run', '--daemon', '--json'], dir, WINDOW); + expect((JSON.parse(purge.stdout) as { count: number }).count).toBe(1); + }); + it('outbox survives a daemon crash: a fresh daemon delivers the leftovers', async () => { const dir = await mkTmp(); const FROZEN = { AGENTCOMM_FLUSH_MS: '600000' }; diff --git a/test/git.e2e.test.ts b/test/git.e2e.test.ts index cd724337..7ac14c7f 100644 --- a/test/git.e2e.test.ts +++ b/test/git.e2e.test.ts @@ -77,17 +77,36 @@ describe('GitBackend (local bare remotes — same code path as any host)', () => const { uri, cache } = await bareRemote(); const b = await open(uri, cache); - expect((await b.snapshot('')).size).toBe(0); // branch not born yet + expect((await b.snapshot('')).keys).toEqual([]); // branch not born yet await b.put('inbox/a/001.json', Buffer.from('one')); await b.put('inbox/b/001.json', Buffer.from('two')); await b.put('agents/a.json', Buffer.from('{"alias":"a"}')); const all = await b.snapshot(''); - expect([...all.keys()].sort()).toEqual(['agents/a.json', 'inbox/a/001.json', 'inbox/b/001.json']); - for (const [k, v] of all) expect(v.equals(await b.get(k))).toBe(true); // bodies match per-key reads + expect([...all.keys].sort()).toEqual(['agents/a.json', 'inbox/a/001.json', 'inbox/b/001.json']); + for (const [k, v] of all.bodies) expect(v.equals(await b.get(k))).toBe(true); // bodies match per-key reads - expect([...(await b.snapshot('inbox/a/')).keys()]).toEqual(['inbox/a/001.json']); + expect((await b.snapshot('inbox/a/')).keys).toEqual(['inbox/a/001.json']); + }, 60000); + + it('snapshot reads only the bodies asked for, but reports every key (issue #167)', async () => { + const { uri, cache } = await bareRemote(); + const b = await open(uri, cache); + await b.put('agents/a.json', Buffer.from('{"alias":"a"}')); + await b.put('read/a/000000000000001-000000_old.json', Buffer.from('ancient history')); + await b.put('inbox/a/000000000000002-000000_new.json', Buffer.from('live mail')); + + const warm = await b.snapshot('', { bodies: (key) => !key.startsWith('read/') }); + // the archive still EXISTS as far as list/log/purge are concerned... + expect(warm.keys).toContain('read/a/000000000000001-000000_old.json'); + // ...its body just was not read + expect([...warm.bodies.keys()].sort()).toEqual([ + 'agents/a.json', + 'inbox/a/000000000000002-000000_new.json', + ]); + // and it is still one `get` away + expect((await b.get('read/a/000000000000001-000000_old.json')).toString()).toBe('ancient history'); }, 60000); it('move is ATOMIC — one commit relocates the key', async () => {