fix(ui): bound shared polling cache (#9406)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - The board coordinates repeated API polling across tabs to reduce redundant requests > - The shared polling coordinator retained cached result and publication entries after the last subscriber left > - Dynamic polling keys could therefore grow those maps for the lifetime of the page > - This pull request evicts inactive keys while preserving useful short-lived handoff state and request deduplication > - The benefit is bounded client memory without regressing cross-tab polling behavior ## Linked Issues or Issue Description ### What happened? Shared polling cached result/publication entries indefinitely after a polling key no longer had subscribers. ### Expected behavior Inactive keys are eventually removed, while recently published values remain available long enough for normal subscriber handoff. ### Steps to reproduce 1. Create and unsubscribe many distinct shared polling keys in one page lifetime. 2. Inspect the coordinator's cached results and publication timestamps. 3. Observe that the old maps retain every historical key. ### Paperclip version or commit `origin/master` at `02e2dd271` ### Deployment mode Local dev; built from source; not adapter-specific; not database-related. ## What Changed - Track inactive polling keys and schedule bounded cache eviction. - Preserve cached data while a key is active or inside its retention window. - Cancel stale cleanup timers when polling resumes and clear coordinator caches during disposal. - Add focused fake-timer coverage for retention, resubscription, and disposal behavior. ## Verification - `vitest --project @paperclipai/ui src/lib/cross-tab-poll.test.ts` — 11 tests passed. ## Risks - Low-to-moderate risk: eviction timing affects client polling coordination. - Tests cover the retention boundary, resumed subscriptions, and coordinator cleanup to reduce regression risk. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - OpenAI Codex, GPT-5.3 Codex, reasoning with repository tool use and code execution; context-window size was not exposed by the runtime. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Dotta
GitHub
parent
5618ea91f6
commit
07e528256e
|
|
@ -67,6 +67,13 @@ function startLeaderCoordinator(channel: MemorySharedChannel) {
|
|||
return coordinator;
|
||||
}
|
||||
|
||||
function getCoordinatorCaches(coordinator: SharedPollingCoordinator) {
|
||||
return coordinator as unknown as {
|
||||
latestResults: Map<string, unknown>;
|
||||
lastPublished: Map<string, unknown>;
|
||||
};
|
||||
}
|
||||
|
||||
describe("LeaderElection", () => {
|
||||
it("elects one visible leader and keeps the second visible tab as follower", () => {
|
||||
let now = 1_000;
|
||||
|
|
@ -269,4 +276,118 @@ describe("SharedPollingCoordinator", () => {
|
|||
|
||||
coordinator.stop();
|
||||
});
|
||||
|
||||
it("bounds cached results with LRU eviction and removes idle entries on ticks", () => {
|
||||
vi.useFakeTimers();
|
||||
vi.setSystemTime(1_000);
|
||||
const channel = new MemorySharedChannel();
|
||||
const coordinator = startLeaderCoordinator(channel);
|
||||
|
||||
for (let index = 1; index <= 32; index += 1) {
|
||||
coordinator.publish(`company:resource-${index}`, { index }, index);
|
||||
}
|
||||
channel.emit({ type: "request", key: "company:resource-1", from: "follower", at: 1_000 });
|
||||
coordinator.publish("company:resource-1", { index: 1, refreshed: true }, 33);
|
||||
coordinator.publish("company:resource-33", { index: 33 }, 34);
|
||||
|
||||
const caches = getCoordinatorCaches(coordinator);
|
||||
expect(caches.latestResults.size).toBe(32);
|
||||
expect(caches.lastPublished.size).toBe(32);
|
||||
expect(caches.latestResults.has("company:resource-1")).toBe(true);
|
||||
expect(caches.lastPublished.has("company:resource-1")).toBe(true);
|
||||
expect(caches.latestResults.has("company:resource-2")).toBe(false);
|
||||
expect(caches.lastPublished.has("company:resource-2")).toBe(false);
|
||||
|
||||
vi.advanceTimersByTime(5 * 60_000 + 10_000);
|
||||
expect(caches.latestResults.size).toBe(0);
|
||||
expect(caches.lastPublished.size).toBe(0);
|
||||
|
||||
coordinator.stop();
|
||||
});
|
||||
|
||||
it("retains listenerless broadcasts through quick resource resubscriptions", () => {
|
||||
vi.useFakeTimers();
|
||||
vi.setSystemTime(1_000);
|
||||
const channel = new MemorySharedChannel();
|
||||
const coordinator = startLeaderCoordinator(channel);
|
||||
const message = {
|
||||
type: "result" as const,
|
||||
key: "company:live-runs",
|
||||
from: "follower",
|
||||
at: 1_000,
|
||||
dataUpdatedAt: 100,
|
||||
data: [{ id: "run-1" }],
|
||||
};
|
||||
|
||||
channel.emit(message);
|
||||
expect(getCoordinatorCaches(coordinator).latestResults.size).toBe(1);
|
||||
|
||||
const firstListener = vi.fn();
|
||||
const unsubscribe = coordinator.subscribeResource(message.key, firstListener);
|
||||
expect(firstListener).toHaveBeenCalledWith(message);
|
||||
expect(getCoordinatorCaches(coordinator).latestResults.size).toBe(1);
|
||||
|
||||
unsubscribe();
|
||||
expect(getCoordinatorCaches(coordinator).latestResults.size).toBe(1);
|
||||
|
||||
const remountedListener = vi.fn();
|
||||
const unsubscribeRemounted = coordinator.subscribeResource(message.key, remountedListener);
|
||||
expect(remountedListener).toHaveBeenCalledWith(message);
|
||||
|
||||
unsubscribeRemounted();
|
||||
vi.advanceTimersByTime(5 * 60_000 + 10_000);
|
||||
expect(getCoordinatorCaches(coordinator).latestResults.size).toBe(0);
|
||||
|
||||
coordinator.stop();
|
||||
});
|
||||
|
||||
it("keeps active publish dedupe entries when inactive keys exceed the cache limit", () => {
|
||||
vi.useFakeTimers();
|
||||
vi.setSystemTime(1_000);
|
||||
const channel = new MemorySharedChannel();
|
||||
const coordinator = startLeaderCoordinator(channel);
|
||||
const unsubscribe = coordinator.subscribeResource("company:resource-1", vi.fn());
|
||||
|
||||
for (let index = 1; index <= 33; index += 1) {
|
||||
coordinator.publish(`company:resource-${index}`, { index }, index);
|
||||
}
|
||||
coordinator.publish("company:resource-1", { index: 1 }, 34);
|
||||
|
||||
expect(channel.posts.filter((message) => message.type === "result")).toHaveLength(33);
|
||||
const caches = getCoordinatorCaches(coordinator);
|
||||
expect(caches.latestResults.size).toBe(33);
|
||||
expect(caches.lastPublished.size).toBe(33);
|
||||
expect(caches.lastPublished.has("company:resource-1")).toBe(true);
|
||||
|
||||
vi.advanceTimersByTime(5 * 60_000 + 10_000);
|
||||
expect(caches.latestResults.size).toBe(1);
|
||||
expect(caches.lastPublished.size).toBe(1);
|
||||
coordinator.publish("company:resource-1", { index: 1 }, 35);
|
||||
expect(channel.posts.filter((message) => message.type === "result")).toHaveLength(33);
|
||||
|
||||
unsubscribe();
|
||||
vi.advanceTimersByTime(5 * 60_000 - 1);
|
||||
expect(caches.latestResults.size).toBe(1);
|
||||
expect(caches.lastPublished.size).toBe(1);
|
||||
vi.advanceTimersByTime(10_001);
|
||||
expect(caches.latestResults.size).toBe(0);
|
||||
expect(caches.lastPublished.size).toBe(0);
|
||||
|
||||
coordinator.stop();
|
||||
});
|
||||
|
||||
it("skips fingerprint traversal for older-or-equal publish snapshots", () => {
|
||||
const channel = new MemorySharedChannel();
|
||||
const coordinator = startLeaderCoordinator(channel);
|
||||
const ownKeys = vi.fn(() => []);
|
||||
const staleData = new Proxy({}, { ownKeys });
|
||||
|
||||
coordinator.publish("company:live-runs", [{ id: "run-1" }], 100);
|
||||
coordinator.publish("company:live-runs", staleData, 100);
|
||||
|
||||
expect(ownKeys).not.toHaveBeenCalled();
|
||||
expect(channel.posts.filter((message) => message.type === "result")).toHaveLength(1);
|
||||
|
||||
coordinator.stop();
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -186,20 +186,86 @@ export class LeaderElection {
|
|||
|
||||
function stableFingerprint(value: unknown): string {
|
||||
const seen = new WeakSet<object>();
|
||||
let hashA = 0xdeadbeef;
|
||||
let hashB = 0x41c6ce57;
|
||||
|
||||
const write = (chunk: string) => {
|
||||
for (let index = 0; index < chunk.length; index += 1) {
|
||||
const code = chunk.charCodeAt(index);
|
||||
hashA = Math.imul(hashA ^ code, 2654435761);
|
||||
hashB = Math.imul(hashB ^ code, 1597334677);
|
||||
}
|
||||
};
|
||||
|
||||
const visit = (entry: unknown, inArray = false): void => {
|
||||
if (entry === null) {
|
||||
write("null");
|
||||
return;
|
||||
}
|
||||
switch (typeof entry) {
|
||||
case "string":
|
||||
write(JSON.stringify(entry));
|
||||
return;
|
||||
case "boolean":
|
||||
write(entry ? "true" : "false");
|
||||
return;
|
||||
case "number":
|
||||
write(Number.isFinite(entry) ? String(entry) : "null");
|
||||
return;
|
||||
case "undefined":
|
||||
write(inArray ? "null" : "undefined");
|
||||
return;
|
||||
case "bigint":
|
||||
write(`bigint:${entry}`);
|
||||
return;
|
||||
case "symbol":
|
||||
write(`symbol:${String(entry)}`);
|
||||
return;
|
||||
case "function":
|
||||
write(`function:${String(entry)}`);
|
||||
return;
|
||||
case "object":
|
||||
break;
|
||||
}
|
||||
|
||||
const object = entry as Record<string, unknown>;
|
||||
if (seen.has(object)) {
|
||||
write("[Circular]");
|
||||
return;
|
||||
}
|
||||
seen.add(object);
|
||||
const toJSON = object.toJSON;
|
||||
if (typeof toJSON === "function") {
|
||||
visit(toJSON.call(object), inArray);
|
||||
return;
|
||||
}
|
||||
if (Array.isArray(object)) {
|
||||
write("[");
|
||||
for (const item of object) {
|
||||
visit(item, true);
|
||||
write(",");
|
||||
}
|
||||
write("]");
|
||||
return;
|
||||
}
|
||||
|
||||
write("{");
|
||||
for (const key of Object.keys(object).sort()) {
|
||||
const item = object[key];
|
||||
if (item === undefined || typeof item === "function" || typeof item === "symbol") continue;
|
||||
write(JSON.stringify(key));
|
||||
write(":");
|
||||
visit(item);
|
||||
write(",");
|
||||
}
|
||||
write("}");
|
||||
};
|
||||
|
||||
try {
|
||||
return JSON.stringify(value, (_key, entry: unknown) => {
|
||||
if (!entry || typeof entry !== "object") return entry;
|
||||
if (seen.has(entry)) return "[Circular]";
|
||||
seen.add(entry);
|
||||
if (Array.isArray(entry)) return entry;
|
||||
const record = entry as Record<string, unknown>;
|
||||
return Object.keys(record)
|
||||
.sort()
|
||||
.reduce<Record<string, unknown>>((acc, key) => {
|
||||
acc[key] = record[key];
|
||||
return acc;
|
||||
}, {});
|
||||
});
|
||||
visit(value);
|
||||
hashA = Math.imul(hashA ^ (hashA >>> 16), 2246822507) ^ Math.imul(hashB ^ (hashB >>> 13), 3266489909);
|
||||
hashB = Math.imul(hashB ^ (hashB >>> 16), 2246822507) ^ Math.imul(hashA ^ (hashA >>> 13), 3266489909);
|
||||
return `${(hashB >>> 0).toString(36)}${(hashA >>> 0).toString(36)}`;
|
||||
} catch {
|
||||
return String(value);
|
||||
}
|
||||
|
|
@ -394,6 +460,8 @@ export interface SharedPollingCoordinatorOptions {
|
|||
|
||||
const DEFAULT_COORDINATOR_TICK_MS = 1_000;
|
||||
const DEFAULT_PUBLISH_DEBOUNCE_MS = 1_000;
|
||||
const MAX_COORDINATOR_CACHE_ENTRIES = 32;
|
||||
const COORDINATOR_CACHE_TTL_MS = 5 * 60_000;
|
||||
const TAB_ID_STORAGE_KEY = "paperclip:shared-poll:tab-id";
|
||||
|
||||
function sanitizeCompanyId(companyId: string): string {
|
||||
|
|
@ -464,11 +532,15 @@ export class SharedPollingCoordinator {
|
|||
private readonly getVisible: () => boolean;
|
||||
private readonly listeners = new Set<SharedPollingListener>();
|
||||
private readonly resourceListeners = new Map<string, Set<SharedPollingResourceListener>>();
|
||||
private readonly latestResults = new Map<string, SharedMessage>();
|
||||
private readonly latestResults = new Map<string, {
|
||||
message: SharedMessage;
|
||||
lastAccessedAt: number;
|
||||
}>();
|
||||
private readonly lastPublished = new Map<string, {
|
||||
dataUpdatedAt: number;
|
||||
fingerprint: string;
|
||||
sentAt: number;
|
||||
lastAccessedAt: number;
|
||||
}>();
|
||||
private readonly pendingPublishes = new Map<string, {
|
||||
data: unknown;
|
||||
|
|
@ -548,11 +620,14 @@ export class SharedPollingCoordinator {
|
|||
this.resourceListeners.set(key, listeners);
|
||||
}
|
||||
listeners.add(listener);
|
||||
const latest = this.latestResults.get(key);
|
||||
const latest = this.getLatestResult(key);
|
||||
if (latest) listener(latest);
|
||||
return () => {
|
||||
listeners?.delete(listener);
|
||||
if (listeners?.size === 0) this.resourceListeners.delete(key);
|
||||
if (listeners?.size === 0) {
|
||||
this.resourceListeners.delete(key);
|
||||
this.markInactive(key);
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
|
|
@ -568,16 +643,15 @@ export class SharedPollingCoordinator {
|
|||
publish(key: string, data: unknown, dataUpdatedAt = this.now()): void {
|
||||
if (!this.snapshot.isLeader) return;
|
||||
if (dataUpdatedAt <= 0) return;
|
||||
const fingerprint = stableFingerprint(data);
|
||||
const last = this.lastPublished.get(key);
|
||||
if (last) {
|
||||
if (dataUpdatedAt <= last.dataUpdatedAt) return;
|
||||
if (fingerprint === last.fingerprint) {
|
||||
this.cancelPendingPublish(key);
|
||||
return;
|
||||
}
|
||||
}
|
||||
const last = this.getLastPublished(key);
|
||||
if (last && dataUpdatedAt <= last.dataUpdatedAt) return;
|
||||
const pending = this.pendingPublishes.get(key);
|
||||
if (pending && dataUpdatedAt < pending.dataUpdatedAt) return;
|
||||
const fingerprint = stableFingerprint(data);
|
||||
if (last && fingerprint === last.fingerprint) {
|
||||
this.cancelPendingPublish(key);
|
||||
return;
|
||||
}
|
||||
if (pending) {
|
||||
if (dataUpdatedAt < pending.dataUpdatedAt) return;
|
||||
if (dataUpdatedAt === pending.dataUpdatedAt && fingerprint === pending.fingerprint) return;
|
||||
|
|
@ -611,8 +685,8 @@ export class SharedPollingCoordinator {
|
|||
dataUpdatedAt,
|
||||
data,
|
||||
};
|
||||
this.lastPublished.set(key, { dataUpdatedAt, fingerprint, sentAt });
|
||||
this.latestResults.set(key, message);
|
||||
this.setLastPublished(key, { dataUpdatedAt, fingerprint, sentAt, lastAccessedAt: sentAt });
|
||||
this.setLatestResult(key, message);
|
||||
this.channel.post(message);
|
||||
}
|
||||
|
||||
|
|
@ -636,6 +710,7 @@ export class SharedPollingCoordinator {
|
|||
}
|
||||
|
||||
private tick(): void {
|
||||
this.evictIdleEntries();
|
||||
if (this.localOnlyFallback) {
|
||||
this.setSnapshot({ isLeader: this.getVisible() });
|
||||
return;
|
||||
|
|
@ -648,17 +723,90 @@ export class SharedPollingCoordinator {
|
|||
if (message.from === this.tabId) return;
|
||||
if (message.type === "request") {
|
||||
if (!this.snapshot.isLeader) return;
|
||||
const latest = this.latestResults.get(message.key);
|
||||
const latest = this.getLatestResult(message.key);
|
||||
if (latest) this.channel.post({ ...latest, from: this.tabId });
|
||||
return;
|
||||
}
|
||||
|
||||
this.latestResults.set(message.key, message);
|
||||
this.setLatestResult(message.key, message);
|
||||
const listeners = this.resourceListeners.get(message.key);
|
||||
if (!listeners) return;
|
||||
if (!listeners || listeners.size === 0) return;
|
||||
for (const listener of listeners) listener(message);
|
||||
}
|
||||
|
||||
private getLatestResult(key: string): SharedMessage | undefined {
|
||||
const entry = this.latestResults.get(key);
|
||||
if (!entry) return undefined;
|
||||
entry.lastAccessedAt = this.now();
|
||||
this.latestResults.delete(key);
|
||||
this.latestResults.set(key, entry);
|
||||
return entry.message;
|
||||
}
|
||||
|
||||
private setLatestResult(key: string, message: SharedMessage): void {
|
||||
this.latestResults.delete(key);
|
||||
this.latestResults.set(key, { message, lastAccessedAt: this.now() });
|
||||
this.evictLeastRecentlyUsed(this.latestResults);
|
||||
}
|
||||
|
||||
private getLastPublished(key: string): {
|
||||
dataUpdatedAt: number;
|
||||
fingerprint: string;
|
||||
sentAt: number;
|
||||
lastAccessedAt: number;
|
||||
} | undefined {
|
||||
const entry = this.lastPublished.get(key);
|
||||
if (!entry) return undefined;
|
||||
entry.lastAccessedAt = this.now();
|
||||
this.lastPublished.delete(key);
|
||||
this.lastPublished.set(key, entry);
|
||||
return entry;
|
||||
}
|
||||
|
||||
private setLastPublished(key: string, entry: {
|
||||
dataUpdatedAt: number;
|
||||
fingerprint: string;
|
||||
sentAt: number;
|
||||
lastAccessedAt: number;
|
||||
}): void {
|
||||
this.lastPublished.delete(key);
|
||||
this.lastPublished.set(key, entry);
|
||||
this.evictLeastRecentlyUsed(this.lastPublished);
|
||||
}
|
||||
|
||||
private evictLeastRecentlyUsed<T>(entries: Map<string, T>): void {
|
||||
const inactiveKeys = Array.from(entries.keys()).filter(
|
||||
(key) => (this.resourceListeners.get(key)?.size ?? 0) === 0,
|
||||
);
|
||||
while (inactiveKeys.length > MAX_COORDINATOR_CACHE_ENTRIES) {
|
||||
const oldestInactiveKey = inactiveKeys.shift();
|
||||
if (oldestInactiveKey === undefined) return;
|
||||
entries.delete(oldestInactiveKey);
|
||||
}
|
||||
}
|
||||
|
||||
private evictIdleEntries(): void {
|
||||
const expiresBefore = this.now() - COORDINATOR_CACHE_TTL_MS;
|
||||
for (const [key, entry] of this.latestResults) {
|
||||
if ((this.resourceListeners.get(key)?.size ?? 0) > 0) continue;
|
||||
if (entry.lastAccessedAt < expiresBefore) this.latestResults.delete(key);
|
||||
}
|
||||
for (const [key, entry] of this.lastPublished) {
|
||||
if ((this.resourceListeners.get(key)?.size ?? 0) > 0) continue;
|
||||
if (entry.lastAccessedAt < expiresBefore) this.lastPublished.delete(key);
|
||||
}
|
||||
}
|
||||
|
||||
private markInactive(key: string): void {
|
||||
const inactiveAt = this.now();
|
||||
const latest = this.latestResults.get(key);
|
||||
if (latest) latest.lastAccessedAt = inactiveAt;
|
||||
const published = this.lastPublished.get(key);
|
||||
if (published) published.lastAccessedAt = inactiveAt;
|
||||
this.evictLeastRecentlyUsed(this.latestResults);
|
||||
this.evictLeastRecentlyUsed(this.lastPublished);
|
||||
}
|
||||
|
||||
private setSnapshot(snapshot: SharedPollingSnapshot): void {
|
||||
if (snapshot.isLeader === this.snapshot.isLeader) return;
|
||||
this.snapshot = snapshot;
|
||||
|
|
|
|||
Loading…
Reference in New Issue