mirror of
https://github.com/block/buzz.git
synced 2026-10-31 14:31:02 +00:00
perf(desktop): bound observer archive to an LRU channel window
The channel-scoped observer archive (`archiveEventsByChannel`) grew only by explicit paged loads from SQLite and was never evicted, so a long session that visits many channels retains every channel's full scroll-back for the lifetime of the app — a renderer memory leak behind the v0.5.9 lag. Bound it to MAX_RETAINED_ARCHIVE_CHANNELS distinct channels via an LRU policy that evicts whole channels (every agent entry for a channel goes together). Channels with a mounted panel are pinned (refcounted) and never evicted; an evicted channel re-hydrates from SQLite on revisit, so eviction is a cache miss, not data loss. A post-unmount fence stops an in-flight page from resurrecting an evicted channel after its pin is released. Co-authored-by: Will Pfleger <pfleger.will@gmail.com> Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
This commit is contained in:
148
desktop/src/features/agents/lib/observerArchiveEviction.test.mjs
Normal file
148
desktop/src/features/agents/lib/observerArchiveEviction.test.mjs
Normal file
@@ -0,0 +1,148 @@
|
||||
/**
|
||||
* Unit tests for selectArchiveEvictionKeys — the pure LRU-with-pins policy that
|
||||
* bounds the channel-scoped observer archive.
|
||||
*
|
||||
* The policy evicts at channel granularity (all agent entries for a channel go
|
||||
* together) while respecting pins (a pinned channel is never evicted). These
|
||||
* tests exercise the selection logic directly, with no React/Tauri runtime.
|
||||
*/
|
||||
|
||||
import assert from "node:assert/strict";
|
||||
import { describe, it } from "node:test";
|
||||
|
||||
import { selectArchiveEvictionKeys } from "./observerArchiveEviction.ts";
|
||||
|
||||
/** Build one archive entry meta. */
|
||||
function meta(agent, channelId, accessSeq) {
|
||||
return { key: `${agent}:${channelId}`, channelId, accessSeq };
|
||||
}
|
||||
|
||||
const NO_PINS = new Set();
|
||||
|
||||
describe("selectArchiveEvictionKeys", () => {
|
||||
it("test_empty_entries_returns_no_eviction", () => {
|
||||
assert.deepEqual(selectArchiveEvictionKeys([], NO_PINS, 8), []);
|
||||
});
|
||||
|
||||
it("test_channel_count_below_cap_returns_no_eviction", () => {
|
||||
const entries = [
|
||||
meta("agentA", "chan-1", 1),
|
||||
meta("agentA", "chan-2", 2),
|
||||
meta("agentA", "chan-3", 3),
|
||||
];
|
||||
assert.deepEqual(selectArchiveEvictionKeys(entries, NO_PINS, 8), []);
|
||||
});
|
||||
|
||||
it("test_channel_count_exactly_at_cap_returns_no_eviction", () => {
|
||||
const entries = Array.from({ length: 8 }, (_, i) =>
|
||||
meta("agentA", `chan-${i}`, i),
|
||||
);
|
||||
assert.deepEqual(selectArchiveEvictionKeys(entries, NO_PINS, 8), []);
|
||||
});
|
||||
|
||||
it("test_over_cap_evicts_least_recently_accessed_channels", () => {
|
||||
// 10 channels, cap 8 → evict the 2 lowest-recency (chan-0, chan-1).
|
||||
const entries = Array.from({ length: 10 }, (_, i) =>
|
||||
meta("agentA", `chan-${i}`, i),
|
||||
);
|
||||
const evicted = selectArchiveEvictionKeys(entries, NO_PINS, 8);
|
||||
assert.deepEqual(evicted.sort(), ["agentA:chan-0", "agentA:chan-1"].sort());
|
||||
});
|
||||
|
||||
it("test_channel_recency_is_max_access_across_its_agent_entries", () => {
|
||||
// chan-old has two agent entries; its most-recent access (seq 100) keeps it
|
||||
// alive even though its other entry (seq 0) is the global minimum.
|
||||
const entries = [
|
||||
meta("agentA", "chan-old", 0),
|
||||
meta("agentB", "chan-old", 100),
|
||||
meta("agentA", "chan-1", 1),
|
||||
meta("agentA", "chan-2", 2),
|
||||
];
|
||||
// cap 2 → keep the 2 most-recent channels (chan-old@100, chan-2@2),
|
||||
// evict chan-1@1.
|
||||
const evicted = selectArchiveEvictionKeys(entries, NO_PINS, 2);
|
||||
assert.deepEqual(evicted, ["agentA:chan-1"]);
|
||||
});
|
||||
|
||||
it("test_evicting_a_channel_removes_all_its_agent_entries", () => {
|
||||
// chan-evict has 3 agent entries; all three keys must be returned together.
|
||||
const entries = [
|
||||
meta("agentA", "chan-evict", 0),
|
||||
meta("agentB", "chan-evict", 0),
|
||||
meta("agentC", "chan-evict", 0),
|
||||
meta("agentA", "chan-keep", 9),
|
||||
];
|
||||
const evicted = selectArchiveEvictionKeys(entries, NO_PINS, 1);
|
||||
assert.deepEqual(
|
||||
evicted.sort(),
|
||||
["agentA:chan-evict", "agentB:chan-evict", "agentC:chan-evict"].sort(),
|
||||
);
|
||||
});
|
||||
|
||||
it("test_pinned_channel_is_never_evicted_even_when_least_recent", () => {
|
||||
// chan-pinned is the least-recently accessed but pinned → must survive;
|
||||
// chan-1 (next lowest) is evicted instead.
|
||||
const entries = [
|
||||
meta("agentA", "chan-pinned", 0),
|
||||
meta("agentA", "chan-1", 1),
|
||||
meta("agentA", "chan-2", 2),
|
||||
meta("agentA", "chan-3", 3),
|
||||
];
|
||||
const evicted = selectArchiveEvictionKeys(
|
||||
entries,
|
||||
new Set(["chan-pinned"]),
|
||||
3,
|
||||
);
|
||||
assert.deepEqual(evicted, ["agentA:chan-1"]);
|
||||
});
|
||||
|
||||
it("test_pins_exceeding_cap_evict_all_unpinned_but_keep_all_pinned", () => {
|
||||
// 3 pinned + 2 unpinned, cap 2. Pins alone exceed the cap, so the unpinned
|
||||
// budget is 0 → both unpinned channels evicted, all pinned retained.
|
||||
const entries = [
|
||||
meta("agentA", "pin-1", 10),
|
||||
meta("agentA", "pin-2", 11),
|
||||
meta("agentA", "pin-3", 12),
|
||||
meta("agentA", "free-1", 1),
|
||||
meta("agentA", "free-2", 2),
|
||||
];
|
||||
const evicted = selectArchiveEvictionKeys(
|
||||
entries,
|
||||
new Set(["pin-1", "pin-2", "pin-3"]),
|
||||
2,
|
||||
);
|
||||
assert.deepEqual(evicted.sort(), ["agentA:free-1", "agentA:free-2"].sort());
|
||||
});
|
||||
|
||||
it("test_pinned_channels_count_against_the_cap_for_unpinned_budget", () => {
|
||||
// 1 pinned + 3 unpinned, cap 2 → unpinned budget = 2 - 1 = 1. Keep the
|
||||
// single most-recent unpinned channel (free-3), evict free-1 and free-2.
|
||||
const entries = [
|
||||
meta("agentA", "pin-1", 5),
|
||||
meta("agentA", "free-1", 1),
|
||||
meta("agentA", "free-2", 2),
|
||||
meta("agentA", "free-3", 3),
|
||||
];
|
||||
const evicted = selectArchiveEvictionKeys(entries, new Set(["pin-1"]), 2);
|
||||
assert.deepEqual(evicted.sort(), ["agentA:free-1", "agentA:free-2"].sort());
|
||||
});
|
||||
|
||||
it("test_tie_broken_deterministically_by_channel_id", () => {
|
||||
// Two channels share recency 1 at the eviction boundary. The keep-sort
|
||||
// breaks ties by ascending channelId, so "chan-a" is retained and the
|
||||
// higher "chan-b" is evicted — deterministic regardless of input order.
|
||||
const entries = [
|
||||
meta("agentA", "chan-b", 1),
|
||||
meta("agentA", "chan-a", 1),
|
||||
meta("agentA", "chan-keep", 9),
|
||||
];
|
||||
const evicted = selectArchiveEvictionKeys(entries, NO_PINS, 2);
|
||||
assert.deepEqual(evicted, ["agentA:chan-b"]);
|
||||
});
|
||||
|
||||
it("test_zero_cap_evicts_all_unpinned_channels", () => {
|
||||
const entries = [meta("agentA", "chan-1", 1), meta("agentA", "chan-2", 2)];
|
||||
const evicted = selectArchiveEvictionKeys(entries, NO_PINS, 0);
|
||||
assert.deepEqual(evicted.sort(), ["agentA:chan-1", "agentA:chan-2"].sort());
|
||||
});
|
||||
});
|
||||
91
desktop/src/features/agents/lib/observerArchiveEviction.ts
Normal file
91
desktop/src/features/agents/lib/observerArchiveEviction.ts
Normal file
@@ -0,0 +1,91 @@
|
||||
/**
|
||||
* Pure LRU-with-pins eviction policy for the channel-scoped observer archive.
|
||||
*
|
||||
* The archive store (`observerRelayStore`) keys paged history by
|
||||
* `${normalizedAgentPubkey}:${channelId}`, so a single channel observed for
|
||||
* several agents occupies several map entries. Memory pressure, however, scales
|
||||
* with the number of distinct *channels* whose scroll-back is retained — the
|
||||
* user opens channels, not (agent, channel) pairs. This module therefore evicts
|
||||
* at channel granularity: when the count of distinct retained channels exceeds
|
||||
* `cap`, the least-recently-accessed unpinned channels are dropped whole (every
|
||||
* agent entry for that channel). Eviction is not data loss — the next visit
|
||||
* re-hydrates the channel from SQLite through the normal paging path.
|
||||
*
|
||||
* Pins are inviolable: a pinned channel is never selected for eviction, even
|
||||
* when pins alone exceed `cap`. A pin marks a channel whose panel is mounted and
|
||||
* whose eviction would blank a live view, so the cap is enforced only against
|
||||
* unpinned channels.
|
||||
*
|
||||
* Pure and React/Tauri-free so the selection logic is unit-testable in
|
||||
* isolation, mirroring the `archivePagingState` extraction.
|
||||
*/
|
||||
|
||||
export interface ArchiveEntryMeta {
|
||||
/** Composite map key `${normalizedAgentPubkey}:${channelId}`. */
|
||||
key: string;
|
||||
/** Channel the entry belongs to; the unit of eviction. */
|
||||
channelId: string;
|
||||
/** Monotonic access counter; higher = more recently read or written. */
|
||||
accessSeq: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* Select the composite map keys to evict so that no more than `cap` distinct
|
||||
* unpinned channels are retained. Returns the keys for every entry belonging to
|
||||
* an evicted channel; pinned channels never appear in the result.
|
||||
*
|
||||
* @param entries One meta per archive map entry (composite key granularity).
|
||||
* @param pinnedChannels Channel ids whose entries must be retained regardless of cap.
|
||||
* @param cap Maximum distinct unpinned channels to keep.
|
||||
*/
|
||||
export function selectArchiveEvictionKeys(
|
||||
entries: readonly ArchiveEntryMeta[],
|
||||
pinnedChannels: ReadonlySet<string>,
|
||||
cap: number,
|
||||
): string[] {
|
||||
// Recency of each channel = the most recent access across all its entries,
|
||||
// so touching any agent's view of a channel refreshes the whole channel.
|
||||
const channelRecency = new Map<string, number>();
|
||||
for (const entry of entries) {
|
||||
const prev = channelRecency.get(entry.channelId);
|
||||
if (prev === undefined || entry.accessSeq > prev) {
|
||||
channelRecency.set(entry.channelId, entry.accessSeq);
|
||||
}
|
||||
}
|
||||
|
||||
// Only unpinned channels are eviction candidates; pins claim retained slots.
|
||||
const unpinned: Array<{ channelId: string; recency: number }> = [];
|
||||
let pinnedPresent = 0;
|
||||
for (const [channelId, recency] of channelRecency) {
|
||||
if (pinnedChannels.has(channelId)) {
|
||||
pinnedPresent += 1;
|
||||
} else {
|
||||
unpinned.push({ channelId, recency });
|
||||
}
|
||||
}
|
||||
|
||||
const unpinnedBudget = Math.max(0, cap - pinnedPresent);
|
||||
if (unpinned.length <= unpinnedBudget) {
|
||||
return [];
|
||||
}
|
||||
|
||||
// Keep the most-recent `unpinnedBudget` channels; evict the rest. Sort by
|
||||
// recency descending, breaking ties by channelId so selection is deterministic.
|
||||
unpinned.sort(
|
||||
(a, b) =>
|
||||
b.recency - a.recency ||
|
||||
(a.channelId < b.channelId ? -1 : a.channelId > b.channelId ? 1 : 0),
|
||||
);
|
||||
const evictChannels = new Set(
|
||||
unpinned.slice(unpinnedBudget).map((channel) => channel.channelId),
|
||||
);
|
||||
|
||||
// Expand evicted channels back to every composite map key they cover.
|
||||
const keys: string[] = [];
|
||||
for (const entry of entries) {
|
||||
if (evictChannels.has(entry.channelId)) {
|
||||
keys.push(entry.key);
|
||||
}
|
||||
}
|
||||
return keys;
|
||||
}
|
||||
@@ -0,0 +1,260 @@
|
||||
/**
|
||||
* Store-level eviction tests for the channel-scoped observer archive.
|
||||
*
|
||||
* These exercise the REAL ingest path (ingestArchivedObserverEvents with an
|
||||
* injected decrypt fn) plus the public pin API, asserting the LRU bound and its
|
||||
* invariants at the store boundary — the layer above the pure policy unit:
|
||||
*
|
||||
* - the retained-channel cap is enforced across ingest pages;
|
||||
* - eviction is least-recently-loaded first;
|
||||
* - a pinned channel is never evicted (even when least recent);
|
||||
* - evicting a channel drops every (agent, channel) entry it owns;
|
||||
* - an evicted channel reads back empty (cache miss → re-hydrate on revisit);
|
||||
* - resetAgentObserverStore clears the LRU bookkeeping so counts don't leak.
|
||||
*
|
||||
* No React/Tauri runtime: security guards are satisfied via _testRegisterKnownAgents
|
||||
* and decrypt is injected. The cap (MAX_RETAINED_ARCHIVE_CHANNELS = 8) is a
|
||||
* private constant; tests drive enough distinct channels to cross it and assert
|
||||
* behaviour relative to that threshold rather than importing the value.
|
||||
*/
|
||||
|
||||
import assert from "node:assert/strict";
|
||||
import { beforeEach, describe, it } from "node:test";
|
||||
|
||||
import {
|
||||
ingestArchivedObserverEvents,
|
||||
resetAgentObserverStore,
|
||||
pinArchiveChannel,
|
||||
unpinArchiveChannel,
|
||||
_testRegisterKnownAgents,
|
||||
_testGetArchivedChannelEvents,
|
||||
} from "@/features/agents/observerRelayStore.ts";
|
||||
|
||||
const CAP = 8; // MAX_RETAINED_ARCHIVE_CHANNELS (private constant, mirrored here)
|
||||
const SUB_ID = "test-eviction-sub";
|
||||
|
||||
/** 64-hex agent pubkey from a short label. */
|
||||
function agentPubkey(label) {
|
||||
return label.padEnd(64, "0");
|
||||
}
|
||||
|
||||
const AGENT_A = agentPubkey("a");
|
||||
const AGENT_B = agentPubkey("b");
|
||||
|
||||
/** Raw kind-24200 telemetry row for (agent, channel); content carries the event. */
|
||||
function makeRawEvent(agent, channelId, seq) {
|
||||
return {
|
||||
id: `${seq}`.padStart(64, "e"),
|
||||
pubkey: agent,
|
||||
created_at: 1000 + seq,
|
||||
kind: 24200,
|
||||
tags: [
|
||||
["p", "d".repeat(64)],
|
||||
["agent", agent],
|
||||
["frame", "telemetry"],
|
||||
],
|
||||
content: JSON.stringify({
|
||||
seq,
|
||||
timestamp: new Date(1_000_000 + seq * 1000).toISOString(),
|
||||
kind: "acp_write",
|
||||
agentIndex: 0,
|
||||
channelId,
|
||||
sessionId: "sess-1",
|
||||
turnId: "turn-1",
|
||||
payload: {},
|
||||
}),
|
||||
sig: "s".repeat(128),
|
||||
};
|
||||
}
|
||||
|
||||
/** Decrypt fn: content is the JSON-encoded ObserverEvent. */
|
||||
const decrypt = (event) => Promise.resolve(JSON.parse(event.content));
|
||||
|
||||
/** Ingest one telemetry frame for (agent, channel), awaiting the store write. */
|
||||
async function ingestOne(agent, channelId, seq) {
|
||||
await ingestArchivedObserverEvents(
|
||||
[makeRawEvent(agent, channelId, seq)],
|
||||
decrypt,
|
||||
);
|
||||
}
|
||||
|
||||
describe("observer archive eviction — store level", () => {
|
||||
beforeEach(() => {
|
||||
resetAgentObserverStore();
|
||||
_testRegisterKnownAgents(SUB_ID, [AGENT_A, AGENT_B]);
|
||||
});
|
||||
|
||||
it("test_channels_up_to_cap_are_all_retained", async () => {
|
||||
for (let i = 0; i < CAP; i++) {
|
||||
await ingestOne(AGENT_A, `chan-${i}`, i);
|
||||
}
|
||||
for (let i = 0; i < CAP; i++) {
|
||||
assert.equal(
|
||||
_testGetArchivedChannelEvents(AGENT_A, `chan-${i}`).length,
|
||||
1,
|
||||
`chan-${i} must be retained at exactly cap`,
|
||||
);
|
||||
}
|
||||
});
|
||||
|
||||
it("test_exceeding_cap_evicts_least_recently_loaded_channel", async () => {
|
||||
// Load CAP+1 distinct channels in ascending order. chan-0 is the oldest
|
||||
// load and unpinned, so it is the one evicted when the cap is crossed.
|
||||
for (let i = 0; i <= CAP; i++) {
|
||||
await ingestOne(AGENT_A, `chan-${i}`, i);
|
||||
}
|
||||
assert.equal(
|
||||
_testGetArchivedChannelEvents(AGENT_A, "chan-0").length,
|
||||
0,
|
||||
"least-recently-loaded channel must be evicted",
|
||||
);
|
||||
// The newest CAP channels survive.
|
||||
for (let i = 1; i <= CAP; i++) {
|
||||
assert.equal(
|
||||
_testGetArchivedChannelEvents(AGENT_A, `chan-${i}`).length,
|
||||
1,
|
||||
`chan-${i} must survive`,
|
||||
);
|
||||
}
|
||||
});
|
||||
|
||||
it("test_pinned_channel_survives_even_when_least_recently_loaded", async () => {
|
||||
// chan-0 is loaded first (oldest) but pinned; loading CAP more channels
|
||||
// crosses the cap. The eviction must fall on the oldest UNPINNED channel
|
||||
// (chan-1), never the pinned chan-0.
|
||||
await ingestOne(AGENT_A, "chan-0", 0);
|
||||
pinArchiveChannel("chan-0");
|
||||
for (let i = 1; i <= CAP; i++) {
|
||||
await ingestOne(AGENT_A, `chan-${i}`, i);
|
||||
}
|
||||
assert.equal(
|
||||
_testGetArchivedChannelEvents(AGENT_A, "chan-0").length,
|
||||
1,
|
||||
"pinned channel must never be evicted",
|
||||
);
|
||||
assert.equal(
|
||||
_testGetArchivedChannelEvents(AGENT_A, "chan-1").length,
|
||||
0,
|
||||
"oldest unpinned channel must be evicted instead of the pinned one",
|
||||
);
|
||||
});
|
||||
|
||||
it("test_unpinning_makes_channel_evictable_again", async () => {
|
||||
// Pin protects chan-0 across a first over-cap wave; after unpinning, a
|
||||
// second wave that again exceeds the cap must be free to evict chan-0.
|
||||
await ingestOne(AGENT_A, "chan-0", 0);
|
||||
pinArchiveChannel("chan-0");
|
||||
for (let i = 1; i <= CAP; i++) {
|
||||
await ingestOne(AGENT_A, `chan-${i}`, i);
|
||||
}
|
||||
assert.equal(
|
||||
_testGetArchivedChannelEvents(AGENT_A, "chan-0").length,
|
||||
1,
|
||||
"pinned channel survives the first wave",
|
||||
);
|
||||
|
||||
unpinArchiveChannel("chan-0");
|
||||
// chan-0 is now the least-recently-loaded unpinned channel. Load fresh
|
||||
// channels to push the retained set past the cap again.
|
||||
for (let i = CAP + 1; i <= CAP * 2; i++) {
|
||||
await ingestOne(AGENT_A, `chan-${i}`, i);
|
||||
}
|
||||
assert.equal(
|
||||
_testGetArchivedChannelEvents(AGENT_A, "chan-0").length,
|
||||
0,
|
||||
"unpinned channel becomes evictable and is dropped",
|
||||
);
|
||||
});
|
||||
|
||||
it("test_refcounted_pin_holds_until_last_release", async () => {
|
||||
// Two mounted consumers pin chan-0; one unpins. The channel must remain
|
||||
// protected until the SECOND unpin — refcount, not a boolean flag.
|
||||
await ingestOne(AGENT_A, "chan-0", 0);
|
||||
pinArchiveChannel("chan-0");
|
||||
pinArchiveChannel("chan-0");
|
||||
unpinArchiveChannel("chan-0");
|
||||
|
||||
for (let i = 1; i <= CAP; i++) {
|
||||
await ingestOne(AGENT_A, `chan-${i}`, i);
|
||||
}
|
||||
assert.equal(
|
||||
_testGetArchivedChannelEvents(AGENT_A, "chan-0").length,
|
||||
1,
|
||||
"channel stays pinned while one consumer still holds a pin",
|
||||
);
|
||||
|
||||
unpinArchiveChannel("chan-0");
|
||||
for (let i = CAP + 1; i <= CAP * 2; i++) {
|
||||
await ingestOne(AGENT_A, `chan-${i}`, i);
|
||||
}
|
||||
assert.equal(
|
||||
_testGetArchivedChannelEvents(AGENT_A, "chan-0").length,
|
||||
0,
|
||||
"channel is evictable only after the last pin is released",
|
||||
);
|
||||
});
|
||||
|
||||
it("test_eviction_drops_all_agent_entries_for_the_channel", async () => {
|
||||
// chan-shared is populated for two agents (fan-out: one channel view
|
||||
// populates multiple (agent, channel) keys). Both keys must be evicted
|
||||
// together when the channel is dropped.
|
||||
await ingestOne(AGENT_A, "chan-shared", 0);
|
||||
await ingestOne(AGENT_B, "chan-shared", 1);
|
||||
// Push past the cap with unrelated channels; chan-shared is oldest.
|
||||
for (let i = 0; i < CAP; i++) {
|
||||
await ingestOne(AGENT_A, `chan-other-${i}`, 100 + i);
|
||||
}
|
||||
assert.equal(
|
||||
_testGetArchivedChannelEvents(AGENT_A, "chan-shared").length,
|
||||
0,
|
||||
"agent-A entry for the evicted channel must be gone",
|
||||
);
|
||||
assert.equal(
|
||||
_testGetArchivedChannelEvents(AGENT_B, "chan-shared").length,
|
||||
0,
|
||||
"agent-B entry for the same evicted channel must be gone too",
|
||||
);
|
||||
});
|
||||
|
||||
it("test_evicted_channel_reads_back_empty_for_rehydration", async () => {
|
||||
// The revisit contract: an evicted channel returns an empty window, which
|
||||
// the paging hook treats as a cache miss and re-hydrates from SQLite. Here
|
||||
// we assert the empty read; re-ingesting the same page restores it.
|
||||
for (let i = 0; i <= CAP; i++) {
|
||||
await ingestOne(AGENT_A, `chan-${i}`, i);
|
||||
}
|
||||
assert.equal(
|
||||
_testGetArchivedChannelEvents(AGENT_A, "chan-0").length,
|
||||
0,
|
||||
"evicted channel reads back empty",
|
||||
);
|
||||
// Re-hydrate: re-ingesting restores the window (proves eviction is not a
|
||||
// permanent tombstone — the key is fully reusable).
|
||||
await ingestOne(AGENT_A, "chan-0", 0);
|
||||
assert.equal(
|
||||
_testGetArchivedChannelEvents(AGENT_A, "chan-0").length,
|
||||
1,
|
||||
"re-ingest after eviction restores the channel window",
|
||||
);
|
||||
});
|
||||
|
||||
it("test_reset_clears_pin_state_so_counts_do_not_leak", async () => {
|
||||
// A pin outstanding at reset must not survive it — otherwise a phantom pin
|
||||
// would protect a channel id in the next session. After reset we re-pin
|
||||
// NOTHING and confirm the same channel id is freely evictable.
|
||||
await ingestOne(AGENT_A, "chan-0", 0);
|
||||
pinArchiveChannel("chan-0");
|
||||
resetAgentObserverStore();
|
||||
_testRegisterKnownAgents(SUB_ID, [AGENT_A, AGENT_B]);
|
||||
|
||||
// If reset leaked the pin, chan-0 would survive this over-cap wave.
|
||||
for (let i = 0; i <= CAP; i++) {
|
||||
await ingestOne(AGENT_A, `chan-${i}`, i);
|
||||
}
|
||||
assert.equal(
|
||||
_testGetArchivedChannelEvents(AGENT_A, "chan-0").length,
|
||||
0,
|
||||
"pin state must not survive reset — chan-0 is freely evictable again",
|
||||
);
|
||||
});
|
||||
});
|
||||
@@ -25,10 +25,19 @@ import {
|
||||
createEmptyTranscriptState,
|
||||
processTranscriptEvent,
|
||||
} from "./ui/agentSessionTranscript";
|
||||
import { selectArchiveEvictionKeys } from "./lib/observerArchiveEviction";
|
||||
|
||||
const MAX_OBSERVER_EVENTS = 3000;
|
||||
const MAX_PENDING_UNKNOWN_AGENT_FRAMES = 100;
|
||||
|
||||
// Maximum number of distinct channels whose archive scroll-back is retained in
|
||||
// RAM. The archive window grows only by explicit paged loads from SQLite, so
|
||||
// without a bound a session that visits many channels accumulates their full
|
||||
// history for the lifetime of the app. Eviction drops the least-recently-loaded
|
||||
// unpinned channels whole; a revisit re-hydrates from SQLite (cache miss, not
|
||||
// data loss). Channels with a mounted panel are pinned and never evicted.
|
||||
const MAX_RETAINED_ARCHIVE_CHANNELS = 8;
|
||||
|
||||
export type ObserverSnapshot = {
|
||||
connectionState: ConnectionState;
|
||||
errorMessage: string | null;
|
||||
@@ -58,6 +67,53 @@ const snapshotByAgent = new Map<string, ObserverSnapshot>();
|
||||
// TranscriptState once over the combined window.
|
||||
const archiveEventsByChannel = new Map<string, ObserverEvent[]>();
|
||||
|
||||
// LRU bookkeeping for `archiveEventsByChannel`. `archiveChannelId` maps a
|
||||
// composite key back to its channelId (the eviction unit) so the pure policy
|
||||
// can group agent entries by channel. `archiveAccessSeq` records the value of
|
||||
// `archiveAccessCounter` at each key's most recent read or write, giving a
|
||||
// total order for least-recently-used selection.
|
||||
const archiveChannelId = new Map<string, string>();
|
||||
const archiveAccessSeq = new Map<string, number>();
|
||||
let archiveAccessCounter = 0;
|
||||
|
||||
// Channels whose archive must never be evicted because a panel is mounted on
|
||||
// them. Refcounted so co-mounted panels (e.g. the channel screen and the
|
||||
// profile panel viewing the same channel) compose: the channel stays pinned
|
||||
// until the last panel unmounts. A channel with a positive count is pinned.
|
||||
const pinnedArchiveChannelCounts = new Map<string, number>();
|
||||
|
||||
/**
|
||||
* Pin a channel's archive against eviction while a panel is mounted on it.
|
||||
* Refcounted so multiple mounted consumers of the same channel compose; the
|
||||
* channel is protected until every consumer calls `unpinArchiveChannel`.
|
||||
* No-op for a null/empty channelId (panels open before a channel resolves).
|
||||
*/
|
||||
export function pinArchiveChannel(channelId: string | null | undefined): void {
|
||||
if (!channelId) return;
|
||||
pinnedArchiveChannelCounts.set(
|
||||
channelId,
|
||||
(pinnedArchiveChannelCounts.get(channelId) ?? 0) + 1,
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Release one pin acquired via `pinArchiveChannel`. When the refcount reaches
|
||||
* zero the channel becomes eligible for LRU eviction again. No-op for a
|
||||
* null/empty channelId or a channel with no outstanding pins.
|
||||
*/
|
||||
export function unpinArchiveChannel(
|
||||
channelId: string | null | undefined,
|
||||
): void {
|
||||
if (!channelId) return;
|
||||
const current = pinnedArchiveChannelCounts.get(channelId);
|
||||
if (current === undefined) return;
|
||||
if (current <= 1) {
|
||||
pinnedArchiveChannelCounts.delete(channelId);
|
||||
} else {
|
||||
pinnedArchiveChannelCounts.set(channelId, current - 1);
|
||||
}
|
||||
}
|
||||
|
||||
// Per-agent, per-channel latest-live-session-id.
|
||||
// Key: `${normalizePubkey(agentPubkey)}:${channelId}`.
|
||||
// Set when a live relay observer event with a sessionId arrives.
|
||||
@@ -298,9 +354,42 @@ function appendArchivedChannelEvent(
|
||||
// order for consumers that call buildTranscriptState over the window.
|
||||
const sorted = [...current, event].sort(compareObserverEvents);
|
||||
archiveEventsByChannel.set(key, sorted);
|
||||
// Record LRU bookkeeping: which channel this key belongs to (the eviction
|
||||
// unit) and its most-recent access, so the policy can pick the least-recently
|
||||
// loaded unpinned channels when the retained-channel cap is exceeded.
|
||||
archiveChannelId.set(key, channelId);
|
||||
archiveAccessSeq.set(key, ++archiveAccessCounter);
|
||||
return true;
|
||||
}
|
||||
|
||||
/**
|
||||
* Evict the least-recently-loaded unpinned channels from the archive window
|
||||
* when the number of distinct retained channels exceeds
|
||||
* `MAX_RETAINED_ARCHIVE_CHANNELS`. Pinned channels (a panel is mounted on them)
|
||||
* are never evicted. Eviction removes every (agent, channel) entry for the
|
||||
* dropped channels and their LRU bookkeeping; a later revisit re-hydrates the
|
||||
* channel from SQLite through the normal paging path. Returns true if any key
|
||||
* was evicted so the caller can decide whether a notify is warranted.
|
||||
*/
|
||||
function evictArchiveChannelsIfNeeded(): boolean {
|
||||
const entries = Array.from(archiveEventsByChannel.keys(), (key) => ({
|
||||
key,
|
||||
channelId: archiveChannelId.get(key) ?? "",
|
||||
accessSeq: archiveAccessSeq.get(key) ?? 0,
|
||||
}));
|
||||
const evictKeys = selectArchiveEvictionKeys(
|
||||
entries,
|
||||
new Set(pinnedArchiveChannelCounts.keys()),
|
||||
MAX_RETAINED_ARCHIVE_CHANNELS,
|
||||
);
|
||||
for (const key of evictKeys) {
|
||||
archiveEventsByChannel.delete(key);
|
||||
archiveChannelId.delete(key);
|
||||
archiveAccessSeq.delete(key);
|
||||
}
|
||||
return evictKeys.length > 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* Read the channel-scoped archive raw events for a given (agent, channel)
|
||||
* pair. Returns an empty array when no archive has been loaded yet.
|
||||
@@ -763,7 +852,14 @@ export async function ingestArchivedObserverEvents(
|
||||
// Batch-notify once for the whole page of archive events. appendAgentEvent
|
||||
// already notifies individually for live/no-channelId events above, so we
|
||||
// only need one extra notify here for the archive path.
|
||||
if (archiveChanged) {
|
||||
//
|
||||
// Enforce the retained-channel bound after the page lands: the channel just
|
||||
// ingested has the highest access seq, so it is safe from its own eviction;
|
||||
// only older unpinned channels are dropped. Fold the eviction result into the
|
||||
// notify decision so a page that only triggers eviction still refreshes any
|
||||
// panel reading an evicted channel.
|
||||
const evicted = evictArchiveChannelsIfNeeded();
|
||||
if (archiveChanged || evicted) {
|
||||
notifyListeners();
|
||||
}
|
||||
}
|
||||
@@ -809,6 +905,10 @@ export function resetAgentObserverStore() {
|
||||
transcriptByAgent.clear();
|
||||
snapshotByAgent.clear();
|
||||
archiveEventsByChannel.clear();
|
||||
archiveChannelId.clear();
|
||||
archiveAccessSeq.clear();
|
||||
archiveAccessCounter = 0;
|
||||
pinnedArchiveChannelCounts.clear();
|
||||
knownAgentPubkeys.clear();
|
||||
knownAgentsBySubscription.clear();
|
||||
pendingUnknownAgentFrames.length = 0;
|
||||
|
||||
@@ -934,3 +934,80 @@ describe("useLoadArchivedObserverEvents — mounted hook lifecycle regressions",
|
||||
await unmount();
|
||||
});
|
||||
});
|
||||
|
||||
describe("useLoadArchivedObserverEvents — post-unmount ingest fence", () => {
|
||||
beforeEach(() => {
|
||||
resetAgentObserverStore();
|
||||
clearIpcHandlers();
|
||||
_testRegisterKnownAgents(SUB_ID, [AGENT_PUBKEY]);
|
||||
});
|
||||
|
||||
/**
|
||||
* Post-unmount fence: a page whose Tauri read is still in flight when the
|
||||
* panel unmounts must NOT write to the archive store. By unmount time the
|
||||
* panel's pin has been released, so the channel may already be evicted;
|
||||
* writing the late page would resurrect a dead channel as most-recently-used
|
||||
* and silently defeat the archive bound.
|
||||
*
|
||||
* Sequence:
|
||||
* 1. Mount on chan-a. Backfill returns immediately (no rows). The archive
|
||||
* read is DEFERRED so it is still in flight at unmount.
|
||||
* 2. Unmount the hook while the read is deferred (disposedRef → true).
|
||||
* 3. Release the archive read. The resolved page reaches the fence check.
|
||||
* 4. Assert chan-a's archive window is EMPTY — the fence skipped the write.
|
||||
*
|
||||
* Without the disposedRef fence, the late page ingests and the channel window
|
||||
* is non-empty after unmount.
|
||||
*/
|
||||
it("test_page_resolving_after_unmount_does_not_write_to_store", async () => {
|
||||
let resolveArchive;
|
||||
const archiveDeferred = new Promise((resolve) => {
|
||||
resolveArchive = resolve;
|
||||
});
|
||||
|
||||
setIpcHandler("list_save_subscriptions", async () =>
|
||||
makeOwnerPSubResponse(),
|
||||
);
|
||||
setIpcHandler("read_unindexed_observer_rows", async () => []);
|
||||
setIpcHandler("index_observer_channel_id", async () => null);
|
||||
setIpcHandler("read_archived_observer_events_for_channel", async (args) => {
|
||||
if (args.channelId === "chan-a") {
|
||||
await archiveDeferred; // hold the read in flight across unmount
|
||||
return [JSON.stringify(makeArchivedRow(1, "chan-a"))];
|
||||
}
|
||||
return [];
|
||||
});
|
||||
setIpcHandler("decrypt_observer_event", async (args) => {
|
||||
try {
|
||||
const event = JSON.parse(args.eventJson);
|
||||
return JSON.parse(event.content);
|
||||
} catch {
|
||||
return { kind: "telemetry", channelId: null };
|
||||
}
|
||||
});
|
||||
|
||||
const qc = makeQueryClient();
|
||||
const { render, unmount } = mountHook("chan-a", qc);
|
||||
|
||||
// Step 1: mount; hydration starts and blocks on the deferred archive read.
|
||||
await render("chan-a");
|
||||
await act(async () => {
|
||||
await new Promise((r) => setTimeout(r, 20));
|
||||
});
|
||||
|
||||
// Step 2: unmount while the read is still in flight.
|
||||
await unmount();
|
||||
|
||||
// Step 3: release the archive read; let the resolved page reach the fence.
|
||||
resolveArchive();
|
||||
await settle(10);
|
||||
|
||||
// Step 4: the late page must NOT have been written to the store.
|
||||
const archived = _testGetArchivedChannelEvents(AGENT_PUBKEY, "chan-a");
|
||||
assert.equal(
|
||||
archived.length,
|
||||
0,
|
||||
`page resolving after unmount must not write to the store — found ${archived.length} (disposedRef fence missing would leave this 1)`,
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -6,7 +6,9 @@ import {
|
||||
getAgentTranscript,
|
||||
getArchivedChannelEvents,
|
||||
ingestArchivedObserverEvents,
|
||||
pinArchiveChannel,
|
||||
subscribeAgentObserverStore,
|
||||
unpinArchiveChannel,
|
||||
} from "@/features/agents/observerRelayStore";
|
||||
import {
|
||||
listSaveSubscriptions,
|
||||
@@ -81,6 +83,18 @@ export function useArchivedChannelEvents(
|
||||
agentPubkey: string | null | undefined,
|
||||
channelId: string | null | undefined,
|
||||
): ObserverEvent[] {
|
||||
// Pin this channel's archive against LRU eviction while the panel is mounted,
|
||||
// so the history the user is looking at is never dropped out from under the
|
||||
// view. Refcounted in the store, so co-mounted consumers of the same channel
|
||||
// (this read hook plus the loader below, plus a second panel) compose. The
|
||||
// pin is keyed on channel only — agent identity does not affect eviction,
|
||||
// which is a per-channel operation.
|
||||
React.useEffect(() => {
|
||||
if (!channelId) return;
|
||||
pinArchiveChannel(channelId);
|
||||
return () => unpinArchiveChannel(channelId);
|
||||
}, [channelId]);
|
||||
|
||||
const getSnapshot = React.useCallback(
|
||||
() => getArchivedChannelEvents(agentPubkey, channelId),
|
||||
[agentPubkey, channelId],
|
||||
@@ -133,6 +147,19 @@ export function useLoadArchivedObserverEvents(
|
||||
}
|
||||
const ps = pagingStateRef.current;
|
||||
|
||||
// Fence for in-flight fetches that resolve after unmount. Set true by the
|
||||
// cleanup effect below; checked before every store write so a page still
|
||||
// decrypting when the panel closes cannot repopulate a channel whose pin has
|
||||
// just been released — which would otherwise defeat the archive bound by
|
||||
// resurrecting a dead channel as most-recently-used.
|
||||
const disposedRef = React.useRef(false);
|
||||
React.useEffect(() => {
|
||||
disposedRef.current = false;
|
||||
return () => {
|
||||
disposedRef.current = true;
|
||||
};
|
||||
}, []);
|
||||
|
||||
// React state mirrors the fields callers observe so re-renders fire on change.
|
||||
const [hasSubscription, setHasSubscription] = React.useState<boolean | null>(
|
||||
ps.hasSubscription,
|
||||
@@ -335,6 +362,15 @@ export function useLoadArchivedObserverEvents(
|
||||
createdAt: oldestEvent.created_at,
|
||||
id: oldestEvent.id,
|
||||
};
|
||||
// Fence post-unmount ingest: if the panel unmounted while this page
|
||||
// was in flight, its pin has been released and the channel may already
|
||||
// be evicted. Writing now would resurrect a dead channel as
|
||||
// most-recently-used, silently defeating the archive bound. Skip the
|
||||
// store write; the cursor advance above is harmless (the ref is
|
||||
// discarded on unmount).
|
||||
if (disposedRef.current) {
|
||||
return;
|
||||
}
|
||||
await ingestArchivedObserverEvents(events);
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user