diff --git a/ui/src/lib/cross-tab-poll.test.ts b/ui/src/lib/cross-tab-poll.test.ts index 19d5c67c84..31e3d228f7 100644 --- a/ui/src/lib/cross-tab-poll.test.ts +++ b/ui/src/lib/cross-tab-poll.test.ts @@ -67,6 +67,13 @@ function startLeaderCoordinator(channel: MemorySharedChannel) { return coordinator; } +function getCoordinatorCaches(coordinator: SharedPollingCoordinator) { + return coordinator as unknown as { + latestResults: Map; + lastPublished: Map; + }; +} + 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(); + }); }); diff --git a/ui/src/lib/cross-tab-poll.ts b/ui/src/lib/cross-tab-poll.ts index 116438d46f..46af27f766 100644 --- a/ui/src/lib/cross-tab-poll.ts +++ b/ui/src/lib/cross-tab-poll.ts @@ -186,20 +186,86 @@ export class LeaderElection { function stableFingerprint(value: unknown): string { const seen = new WeakSet(); + 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; + 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; - return Object.keys(record) - .sort() - .reduce>((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(); private readonly resourceListeners = new Map>(); - private readonly latestResults = new Map(); + private readonly latestResults = new Map(); private readonly lastPublished = new Map(); private readonly pendingPublishes = new Map { 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(entries: Map): 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;