Fix agent sidebar liveness churn (#9358)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - The board UI includes an agents sidebar so operators can see which agents are currently active. > - That sidebar depends on live-run polling, heartbeat events, and cross-tab cache sharing to stay current without overloading the API. > - The sidebar was visually churning because agents could leave the live section immediately after a run ended, while progress events and cross-tab broadcasts kept forcing hot query updates. > - This pull request stabilizes live sidebar membership and makes shared polling broadcasts monotonic/deduplicated. > - The benefit is a calmer operator sidebar that still reflects real live state without flashing between stale and fresh snapshots. ## Linked Issues or Issue Description No public GitHub issue exists for this operator-facing bug. Bug summary: - What happened: the agents sidebar could flash or reshuffle around active agents while live-run and heartbeat data was updating. - Expected behavior: active and recently-active agents should remain visually stable, and cross-tab cache sharing should not overwrite fresher data with older snapshots. - Reproduction context: run Paperclip with multiple tabs or rapid live-run/progress updates and watch the agents sidebar while agents enter/leave live execution. - Deployment mode: local/operator board UI. Related PR: - Supersedes #9357, which carried the same fixes on a branch/title/body that were not suitable for public contribution hygiene. ## What Changed - Restored the maintainer-only warning wording in the developer skill guide so the existing server skill-utils CI gate passes on current master. - Added a 120-second linger window for streamlined sidebar agent rows so an agent does not immediately disappear from the live section as soon as its last run ends. - Deferred the recent-agent fallback until there are no live or lingering agents, while keeping the live badge tied only to actually-live runs. - Stopped broad live-runs/heartbeats/agents-list invalidation on every run progress event, while preserving targeted agent-detail invalidation. - Added producer timestamps to cross-tab shared polling result messages so older-or-equal snapshots are dropped before `setQueryData`. - Added per-resource broadcast dedupe/rate limiting so tabs do not rebroadcast equivalent cached data in a loop. - Added focused coverage for sidebar linger behavior, staggered multi-agent linger expiry, live update invalidation scope, shared polling timestamp handling, and cross-tab broadcast dedupe. ## Verification Run locally on the rebased PR branch: - `pnpm --filter @paperclipai/ui exec vitest run src/components/SidebarAgents.test.tsx src/context/LiveUpdatesProvider.test.ts` — 44 tests passed. - `pnpm --filter @paperclipai/ui exec vitest run src/lib/cross-tab-poll.test.ts src/hooks/useSharedPolling.test.ts` — 10 tests passed. - `pnpm --filter @paperclipai/ui typecheck` — exit code 0. ## Risks - Low migration risk: the sidebar/polling changes are UI/client cache behavior only, with no database or API contract changes. - Sidebar visibility now intentionally lingers for 120 seconds after the last live run; stale rows could remain briefly visible, but their live badge is removed when they are no longer actually live. - Cross-tab broadcasts are now more conservative; a missed publish should be corrected by the next normal poll or accepted newer timestamp. > 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 GPT-5 Codex via the Paperclip Codex coding-agent runtime; exact API model identifier and context-window size are not exposed in this environment. The agent used terminal/tool execution for repository inspection, focused tests, branch preparation, and PR creation. ## 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
e84731af70
commit
be1fcb2b46
|
|
@ -6,7 +6,7 @@ description: >
|
|||
checkbox confirmations, ask_user_questions, or suggest_tasks.
|
||||
---
|
||||
|
||||
# Create a new issue-thread interaction UI (developer skill)
|
||||
# Create a new issue-thread interaction UI (Developer/maintainer skill)
|
||||
|
||||
This skill walks a Paperclip contributor through introducing a new issue-thread
|
||||
interaction kind from shared contract to issue-detail wiring, helpers, and
|
||||
|
|
@ -14,6 +14,8 @@ docs. It is intentionally a developer/maintainer skill: the audience is a
|
|||
human or coding agent making code changes inside `paperclipai/paperclip`, not
|
||||
the operational agents that run inside a deployed Paperclip company.
|
||||
|
||||
Do NOT install this on production Paperclip agents. This guide is for repository contributors changing Paperclip itself.
|
||||
|
||||
## When to use
|
||||
|
||||
- A new interaction kind is being introduced (compact picker, structured
|
||||
|
|
|
|||
|
|
@ -7,6 +7,7 @@ import { QueryClient, QueryClientProvider } from "@tanstack/react-query";
|
|||
import type { Agent, ResourceMemberships } from "@paperclipai/shared";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { SidebarAgents } from "./SidebarAgents";
|
||||
import { queryKeys } from "../lib/queryKeys";
|
||||
import { TooltipProvider } from "@/components/ui/tooltip";
|
||||
|
||||
const mockAgentsApi = vi.hoisted(() => ({
|
||||
|
|
@ -267,6 +268,7 @@ describe("SidebarAgents", () => {
|
|||
currentRoot.unmount();
|
||||
});
|
||||
}
|
||||
vi.useRealTimers();
|
||||
queryClient.clear();
|
||||
container.remove();
|
||||
document.body.innerHTML = "";
|
||||
|
|
@ -302,6 +304,23 @@ describe("SidebarAgents", () => {
|
|||
await flushReact();
|
||||
}
|
||||
|
||||
async function renderSidebarAgentsWithFakeTimers() {
|
||||
const currentRoot = createRoot(container);
|
||||
root = currentRoot;
|
||||
|
||||
await act(async () => {
|
||||
currentRoot.render(
|
||||
<QueryClientProvider client={queryClient}>
|
||||
<SidebarAgents streamlined />
|
||||
</QueryClientProvider>,
|
||||
);
|
||||
});
|
||||
await act(async () => {
|
||||
await Promise.resolve();
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
});
|
||||
}
|
||||
|
||||
async function renderRailSidebarAgents() {
|
||||
mockSidebarState.collapsed = true;
|
||||
const currentRoot = createRoot(container);
|
||||
|
|
@ -645,6 +664,110 @@ describe("SidebarAgents", () => {
|
|||
expect(seeAllAgentsLink(container)?.getAttribute("href")).toBe("/agents/all");
|
||||
});
|
||||
|
||||
it("keeps formerly live agents visible for the streamlined linger window", async () => {
|
||||
vi.useFakeTimers({ now: new Date("2026-01-01T00:00:00Z") });
|
||||
mockAgentsApi.list.mockResolvedValue([
|
||||
makeAgent({ id: "agent-a", name: "Alpha", urlKey: "alpha" }),
|
||||
makeAgent({ id: "agent-b", name: "Bravo", urlKey: "bravo" }),
|
||||
makeAgent({ id: "agent-c", name: "Charlie", urlKey: "charlie" }),
|
||||
makeAgent({ id: "agent-d", name: "Delta", urlKey: "delta" }),
|
||||
]);
|
||||
mockHeartbeatsApi.liveRunsForCompany.mockResolvedValue([
|
||||
{ id: "run-1", agentId: "agent-a", status: "running" },
|
||||
]);
|
||||
|
||||
await renderSidebarAgentsWithFakeTimers();
|
||||
|
||||
let labels = agentLinkLabels(container);
|
||||
expect(labels).toHaveLength(1);
|
||||
expect(labels[0]).toContain("Alpha");
|
||||
expect(labels[0]).toContain("1 live");
|
||||
|
||||
mockHeartbeatsApi.liveRunsForCompany.mockResolvedValue([]);
|
||||
await act(async () => {
|
||||
queryClient.setQueryData(queryKeys.liveRuns("company-1"), []);
|
||||
});
|
||||
await act(async () => {
|
||||
await Promise.resolve();
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
});
|
||||
|
||||
labels = agentLinkLabels(container);
|
||||
expect(labels).toEqual(["Alpha"]);
|
||||
expect(labels.join(" ")).not.toContain("live");
|
||||
|
||||
await act(async () => {
|
||||
await vi.advanceTimersByTimeAsync(120_000);
|
||||
});
|
||||
expect(agentLinkLabels(container)).toEqual(["Alpha"]);
|
||||
|
||||
await act(async () => {
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
});
|
||||
expect(agentLinkLabels(container)).toEqual(["Alpha", "Bravo", "Charlie"]);
|
||||
});
|
||||
|
||||
it("expires staggered lingering agents without unrelated sidebar updates", async () => {
|
||||
vi.useFakeTimers({ now: new Date("2026-01-01T00:00:00Z") });
|
||||
mockAgentsApi.list.mockResolvedValue([
|
||||
makeAgent({ id: "agent-a", name: "Alpha", urlKey: "alpha" }),
|
||||
makeAgent({ id: "agent-b", name: "Bravo", urlKey: "bravo" }),
|
||||
makeAgent({ id: "agent-c", name: "Charlie", urlKey: "charlie" }),
|
||||
makeAgent({ id: "agent-d", name: "Delta", urlKey: "delta" }),
|
||||
]);
|
||||
mockHeartbeatsApi.liveRunsForCompany.mockResolvedValue([
|
||||
{ id: "run-1", agentId: "agent-a", status: "running" },
|
||||
]);
|
||||
|
||||
await renderSidebarAgentsWithFakeTimers();
|
||||
expect(agentLinkLabels(container)[0]).toContain("Alpha");
|
||||
|
||||
mockHeartbeatsApi.liveRunsForCompany.mockResolvedValue([]);
|
||||
await act(async () => {
|
||||
queryClient.setQueryData(queryKeys.liveRuns("company-1"), []);
|
||||
});
|
||||
await act(async () => {
|
||||
await Promise.resolve();
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
});
|
||||
expect(agentLinkLabels(container)).toEqual(["Alpha"]);
|
||||
|
||||
mockHeartbeatsApi.liveRunsForCompany.mockResolvedValue([
|
||||
{ id: "run-2", agentId: "agent-b", status: "running" },
|
||||
]);
|
||||
await act(async () => {
|
||||
await vi.advanceTimersByTimeAsync(60_000);
|
||||
queryClient.setQueryData(queryKeys.liveRuns("company-1"), [
|
||||
{ id: "run-2", agentId: "agent-b", status: "running" },
|
||||
]);
|
||||
});
|
||||
await act(async () => {
|
||||
await Promise.resolve();
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
});
|
||||
expect(agentLinkLabels(container).join(" ")).toContain("Bravo");
|
||||
|
||||
mockHeartbeatsApi.liveRunsForCompany.mockResolvedValue([]);
|
||||
await act(async () => {
|
||||
queryClient.setQueryData(queryKeys.liveRuns("company-1"), []);
|
||||
});
|
||||
await act(async () => {
|
||||
await Promise.resolve();
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
});
|
||||
expect(agentLinkLabels(container)).toEqual(["Alpha", "Bravo"]);
|
||||
|
||||
await act(async () => {
|
||||
await vi.advanceTimersByTimeAsync(60_001);
|
||||
});
|
||||
expect(agentLinkLabels(container)).toEqual(["Bravo"]);
|
||||
|
||||
await act(async () => {
|
||||
await vi.advanceTimersByTimeAsync(60_000);
|
||||
});
|
||||
expect(agentLinkLabels(container)).toEqual(["Alpha", "Bravo", "Charlie"]);
|
||||
});
|
||||
|
||||
it("shows up to 3 recently-active agents plus a See all link when none are running", async () => {
|
||||
mockAgentsApi.list.mockResolvedValue(
|
||||
Array.from({ length: 7 }, (_, index) =>
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
import { useCallback, useEffect, useMemo, useState } from "react";
|
||||
import { useCallback, useEffect, useMemo, useRef, useState } from "react";
|
||||
import { Link, useLocation } from "@/lib/router";
|
||||
import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query";
|
||||
import {
|
||||
|
|
@ -63,6 +63,7 @@ import type { Agent } from "@paperclipai/shared";
|
|||
* recently-active agents plus a "See all agents" link (IA Phase 5).
|
||||
*/
|
||||
const RECENT_AGENT_LIMIT = 3;
|
||||
const LIVE_AGENT_LINGER_MS = 120_000;
|
||||
|
||||
const AGENT_SORT_CHOICES: SidebarSectionRadioChoice[] = [
|
||||
{ value: "top", label: "Top" },
|
||||
|
|
@ -298,6 +299,8 @@ function SidebarAgentItem({
|
|||
export function SidebarAgents({ streamlined = false }: { streamlined?: boolean } = {}) {
|
||||
const [open, setOpen] = useState(true);
|
||||
const [pendingAgentIds, setPendingAgentIds] = useState<Set<string>>(() => new Set());
|
||||
const [liveLingerVersion, setLiveLingerVersion] = useState(0);
|
||||
const lastSeenLiveAtRef = useRef<Map<string, number>>(new Map());
|
||||
const queryClient = useQueryClient();
|
||||
const { selectedCompanyId } = useCompany();
|
||||
const { openNewAgent } = useDialogActions();
|
||||
|
|
@ -354,6 +357,13 @@ export function SidebarAgents({ streamlined = false }: { streamlined?: boolean }
|
|||
}
|
||||
return counts;
|
||||
}, [liveRuns]);
|
||||
const liveAgentIds = useMemo(() => {
|
||||
const ids = new Set<string>();
|
||||
for (const [agentId, count] of liveCountByAgent) {
|
||||
if (count > 0) ids.add(agentId);
|
||||
}
|
||||
return ids;
|
||||
}, [liveCountByAgent]);
|
||||
|
||||
const visibleAgents = useMemo(() => {
|
||||
const filtered = (agents ?? []).filter(
|
||||
|
|
@ -384,15 +394,39 @@ export function SidebarAgents({ streamlined = false }: { streamlined?: boolean }
|
|||
() => sortAgents(orderedAgents, sortMode),
|
||||
[orderedAgents, sortMode],
|
||||
);
|
||||
const sortedAgentIdSet = useMemo(
|
||||
() => new Set(sortedAgents.map((agent: Agent) => agent.id)),
|
||||
[sortedAgents],
|
||||
);
|
||||
|
||||
useEffect(() => {
|
||||
const now = Date.now();
|
||||
for (const agentId of liveAgentIds) {
|
||||
lastSeenLiveAtRef.current.set(agentId, now);
|
||||
}
|
||||
for (const agentId of lastSeenLiveAtRef.current.keys()) {
|
||||
if (!sortedAgentIdSet.has(agentId)) {
|
||||
lastSeenLiveAtRef.current.delete(agentId);
|
||||
}
|
||||
}
|
||||
}, [liveAgentIds, sortedAgentIdSet]);
|
||||
|
||||
// IA Phase 5 (streamlined): if any agent has a live run, show only those
|
||||
// active agents. Otherwise fall back to up to RECENT_AGENT_LIMIT agents. Either
|
||||
// way a "See all agents" link is shown so the full list is always reachable.
|
||||
// active agents. Agents that just stopped running linger briefly so clustered
|
||||
// run boundaries do not make rows pop out and the section does not immediately
|
||||
// swap to the recent fallback during short all-idle gaps. Otherwise fall back
|
||||
// to up to RECENT_AGENT_LIMIT agents. Either way a "See all agents" link is
|
||||
// shown so the full list is always reachable.
|
||||
// Classic mode (PAP-89, flag OFF) restores the show-all behavior.
|
||||
const runningAgents = useMemo(
|
||||
() => sortedAgents.filter((agent: Agent) => (liveCountByAgent.get(agent.id) ?? 0) > 0),
|
||||
[sortedAgents, liveCountByAgent],
|
||||
);
|
||||
const runningAgents = useMemo(() => {
|
||||
const nowForLiveLinger = Date.now();
|
||||
const lastSeenLiveAtByAgent = lastSeenLiveAtRef.current;
|
||||
return sortedAgents.filter((agent: Agent) => {
|
||||
if ((liveCountByAgent.get(agent.id) ?? 0) > 0) return true;
|
||||
const lastSeenLiveAt = lastSeenLiveAtByAgent.get(agent.id);
|
||||
return lastSeenLiveAt !== undefined && nowForLiveLinger - lastSeenLiveAt <= LIVE_AGENT_LINGER_MS;
|
||||
});
|
||||
}, [liveCountByAgent, liveLingerVersion, sortedAgents]);
|
||||
const hasActiveAgents = runningAgents.length > 0;
|
||||
const displayedAgents = !streamlined
|
||||
? sortedAgents
|
||||
|
|
@ -437,6 +471,30 @@ export function SidebarAgents({ streamlined = false }: { streamlined?: boolean }
|
|||
};
|
||||
}, [sortModeStorageKey]);
|
||||
|
||||
useEffect(() => {
|
||||
if (!streamlined) return;
|
||||
|
||||
const now = Date.now();
|
||||
let nextExpiryAt: number | null = null;
|
||||
for (const agent of sortedAgents) {
|
||||
if ((liveCountByAgent.get(agent.id) ?? 0) > 0) continue;
|
||||
const lastSeenLiveAt = lastSeenLiveAtRef.current.get(agent.id);
|
||||
if (lastSeenLiveAt === undefined) continue;
|
||||
const expiresAt = lastSeenLiveAt + LIVE_AGENT_LINGER_MS;
|
||||
if (expiresAt < now) continue;
|
||||
nextExpiryAt = nextExpiryAt === null ? expiresAt : Math.min(nextExpiryAt, expiresAt);
|
||||
}
|
||||
if (nextExpiryAt === null) return;
|
||||
|
||||
const timeoutId = window.setTimeout(() => {
|
||||
setLiveLingerVersion((version) => version + 1);
|
||||
}, Math.max(0, nextExpiryAt - now + 1));
|
||||
|
||||
return () => {
|
||||
window.clearTimeout(timeoutId);
|
||||
};
|
||||
}, [streamlined, sortedAgents, liveCountByAgent, liveLingerVersion]);
|
||||
|
||||
const persistSortMode = useCallback(
|
||||
(value: string) => {
|
||||
const nextSortMode: AgentSidebarSortMode =
|
||||
|
|
|
|||
|
|
@ -100,7 +100,7 @@ describe("LiveUpdatesProvider issue invalidation", () => {
|
|||
});
|
||||
});
|
||||
|
||||
it("keeps heartbeat progress invalidation scoped to live run data", () => {
|
||||
it("keeps heartbeat progress invalidation scoped away from hot list queries", () => {
|
||||
const invalidations: unknown[] = [];
|
||||
const queryClient = {
|
||||
invalidateQueries: (input: unknown) => {
|
||||
|
|
@ -117,21 +117,21 @@ describe("LiveUpdatesProvider issue invalidation", () => {
|
|||
},
|
||||
);
|
||||
|
||||
expect(invalidations).toContainEqual({
|
||||
queryKey: queryKeys.liveRuns("company-1"),
|
||||
});
|
||||
expect(invalidations).toContainEqual({
|
||||
queryKey: queryKeys.heartbeats("company-1"),
|
||||
});
|
||||
expect(invalidations).toContainEqual({
|
||||
queryKey: queryKeys.agents.list("company-1"),
|
||||
});
|
||||
expect(invalidations).toContainEqual({
|
||||
queryKey: queryKeys.agents.detail("agent-1"),
|
||||
});
|
||||
expect(invalidations).toContainEqual({
|
||||
expect(invalidations).not.toContainEqual({
|
||||
queryKey: queryKeys.liveRuns("company-1"),
|
||||
});
|
||||
expect(invalidations).not.toContainEqual({
|
||||
queryKey: queryKeys.heartbeats("company-1"),
|
||||
});
|
||||
expect(invalidations).not.toContainEqual({
|
||||
queryKey: queryKeys.heartbeats("company-1", "agent-1"),
|
||||
});
|
||||
expect(invalidations).not.toContainEqual({
|
||||
queryKey: queryKeys.agents.list("company-1"),
|
||||
});
|
||||
expect(invalidations).not.toContainEqual({
|
||||
queryKey: queryKeys.dashboard("company-1"),
|
||||
});
|
||||
|
|
|
|||
|
|
@ -817,17 +817,12 @@ function invalidateHeartbeatQueries(
|
|||
|
||||
function invalidateHeartbeatProgressQueries(
|
||||
queryClient: ReturnType<typeof useQueryClient>,
|
||||
companyId: string,
|
||||
_companyId: string,
|
||||
payload: Record<string, unknown>,
|
||||
) {
|
||||
queryClient.invalidateQueries({ queryKey: queryKeys.liveRuns(companyId) });
|
||||
queryClient.invalidateQueries({ queryKey: queryKeys.heartbeats(companyId) });
|
||||
queryClient.invalidateQueries({ queryKey: queryKeys.agents.list(companyId) });
|
||||
|
||||
const agentId = readString(payload.agentId);
|
||||
if (agentId) {
|
||||
queryClient.invalidateQueries({ queryKey: queryKeys.agents.detail(agentId) });
|
||||
queryClient.invalidateQueries({ queryKey: queryKeys.heartbeats(companyId, agentId) });
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,43 @@
|
|||
import { QueryClient } from "@tanstack/react-query";
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { applySharedPollingResult } from "./useSharedPolling";
|
||||
|
||||
describe("applySharedPollingResult", () => {
|
||||
it("drops result messages that are older than local query state", () => {
|
||||
const queryClient = new QueryClient();
|
||||
const queryKey = ["live-runs", "company-1"];
|
||||
queryClient.setQueryData(queryKey, [{ id: "run-1", lastEventAt: "newer" }], { updatedAt: 2_000 });
|
||||
|
||||
const applied = applySharedPollingResult(queryClient, queryKey, {
|
||||
type: "result",
|
||||
key: "company:live-runs",
|
||||
from: "leader",
|
||||
at: 1_000,
|
||||
dataUpdatedAt: 1_000,
|
||||
data: [{ id: "run-1", lastEventAt: "older" }],
|
||||
});
|
||||
|
||||
expect(applied).toBe(false);
|
||||
expect(queryClient.getQueryData(queryKey)).toEqual([{ id: "run-1", lastEventAt: "newer" }]);
|
||||
expect(queryClient.getQueryState(queryKey)?.dataUpdatedAt).toBe(2_000);
|
||||
});
|
||||
|
||||
it("applies newer result messages with the producer dataUpdatedAt", () => {
|
||||
const queryClient = new QueryClient();
|
||||
const queryKey = ["live-runs", "company-1"];
|
||||
queryClient.setQueryData(queryKey, [{ id: "run-1", lastEventAt: "older" }], { updatedAt: 1_000 });
|
||||
|
||||
const applied = applySharedPollingResult(queryClient, queryKey, {
|
||||
type: "result",
|
||||
key: "company:live-runs",
|
||||
from: "leader",
|
||||
at: 4_000,
|
||||
dataUpdatedAt: 3_000,
|
||||
data: [{ id: "run-1", lastEventAt: "newer" }],
|
||||
});
|
||||
|
||||
expect(applied).toBe(true);
|
||||
expect(queryClient.getQueryData(queryKey)).toEqual([{ id: "run-1", lastEventAt: "newer" }]);
|
||||
expect(queryClient.getQueryState(queryKey)?.dataUpdatedAt).toBe(3_000);
|
||||
});
|
||||
});
|
||||
|
|
@ -1,7 +1,8 @@
|
|||
import { useCallback, useEffect, useMemo, useRef, useState } from "react";
|
||||
import { type QueryKey, useQueryClient } from "@tanstack/react-query";
|
||||
import { type QueryClient, type QueryKey, useQueryClient } from "@tanstack/react-query";
|
||||
import {
|
||||
SharedPollingCoordinator,
|
||||
type SharedMessage,
|
||||
type SharedPollingSnapshot,
|
||||
} from "../lib/cross-tab-poll";
|
||||
|
||||
|
|
@ -26,7 +27,7 @@ export interface SharedPollingQueryState<TData> {
|
|||
isLeader: boolean;
|
||||
enabled: boolean;
|
||||
refetchInterval: RefetchInterval;
|
||||
publish: (data: TData | undefined) => void;
|
||||
publish: (data: TData | undefined, dataUpdatedAt: number) => void;
|
||||
}
|
||||
|
||||
type RegistryEntry = {
|
||||
|
|
@ -61,6 +62,20 @@ function resourceKey(companyId: string, key: string, queryKeyHash: string): stri
|
|||
return `${companyId}:${key}:${queryKeyHash}`;
|
||||
}
|
||||
|
||||
export function applySharedPollingResult<TData>(
|
||||
queryClient: Pick<QueryClient, "getQueryState" | "setQueryData">,
|
||||
queryKey: QueryKey,
|
||||
message: SharedMessage,
|
||||
): boolean {
|
||||
if (message.type !== "result") return false;
|
||||
const incomingUpdatedAt = message.dataUpdatedAt ?? message.at;
|
||||
if (incomingUpdatedAt <= 0) return false;
|
||||
const localUpdatedAt = queryClient.getQueryState(queryKey)?.dataUpdatedAt ?? 0;
|
||||
if (localUpdatedAt >= incomingUpdatedAt) return false;
|
||||
queryClient.setQueryData(queryKey, message.data as TData, { updatedAt: incomingUpdatedAt });
|
||||
return true;
|
||||
}
|
||||
|
||||
export function useSharedPollingQuery<TData>({
|
||||
companyId,
|
||||
resourceKey: rawResourceKey,
|
||||
|
|
@ -89,8 +104,7 @@ export function useSharedPollingQuery<TData>({
|
|||
const coordinator = acquireCoordinator(activeCompanyId);
|
||||
const unsubscribeState = coordinator.subscribe(setSnapshot);
|
||||
const unsubscribeResource = coordinator.subscribeResource(fullResourceKey, (message) => {
|
||||
if (message.type !== "result") return;
|
||||
queryClient.setQueryData(queryKeyRef.current, message.data as TData);
|
||||
applySharedPollingResult(queryClient, queryKeyRef.current, message);
|
||||
});
|
||||
coordinator.request(fullResourceKey);
|
||||
|
||||
|
|
@ -105,10 +119,10 @@ export function useSharedPollingQuery<TData>({
|
|||
const queryEnabled = enabled && (!leaderOnly || snapshot.isLeader);
|
||||
const coordinatedInterval = leaderOnly && !snapshot.isLeader ? false : refetchInterval;
|
||||
|
||||
const publish = useCallback((data: TData | undefined) => {
|
||||
const publish = useCallback((data: TData | undefined, dataUpdatedAt: number) => {
|
||||
if (!activeCompanyId || !fullResourceKey || data === undefined) return;
|
||||
const entry = coordinators.get(activeCompanyId);
|
||||
entry?.coordinator.publish(fullResourceKey, data);
|
||||
entry?.coordinator.publish(fullResourceKey, data, dataUpdatedAt);
|
||||
}, [activeCompanyId, fullResourceKey]);
|
||||
|
||||
return useMemo(
|
||||
|
|
@ -129,6 +143,6 @@ export function usePublishSharedQueryData<TData>(
|
|||
): void {
|
||||
useEffect(() => {
|
||||
if (!shared.isLeader || dataUpdatedAt <= 0) return;
|
||||
shared.publish(data);
|
||||
shared.publish(data, dataUpdatedAt);
|
||||
}, [data, dataUpdatedAt, shared]);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,9 +1,11 @@
|
|||
import { describe, expect, it, vi } from "vitest";
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import {
|
||||
LeaderElection,
|
||||
SharedPollingCoordinator,
|
||||
createLeaseStore,
|
||||
createStorageRelayShared,
|
||||
type LeaseStore,
|
||||
type SharedChannel,
|
||||
type SharedMessage,
|
||||
} from "./cross-tab-poll";
|
||||
|
||||
|
|
@ -23,6 +25,48 @@ class MemoryLeaseStore implements LeaseStore {
|
|||
}
|
||||
}
|
||||
|
||||
class MemorySharedChannel implements SharedChannel {
|
||||
posts: SharedMessage[] = [];
|
||||
private handler: ((message: SharedMessage) => void) | null = null;
|
||||
|
||||
post(message: SharedMessage) {
|
||||
this.posts.push(message);
|
||||
}
|
||||
|
||||
subscribe(handler: (message: SharedMessage) => void) {
|
||||
this.handler = handler;
|
||||
return () => {
|
||||
this.handler = null;
|
||||
};
|
||||
}
|
||||
|
||||
emit(message: SharedMessage) {
|
||||
this.handler?.(message);
|
||||
}
|
||||
|
||||
close() {}
|
||||
}
|
||||
|
||||
function startLeaderCoordinator(channel: MemorySharedChannel) {
|
||||
const store = new MemoryLeaseStore();
|
||||
const election = new LeaderElection({ now: () => Date.now(), store, random: () => 0 }, {
|
||||
tabId: "leader",
|
||||
leaseTtlMs: 10_000,
|
||||
});
|
||||
const coordinator = new SharedPollingCoordinator("company-1", {
|
||||
tabId: "leader",
|
||||
channel,
|
||||
election,
|
||||
tickMs: 10_000,
|
||||
publishDebounceMs: 1_000,
|
||||
now: () => Date.now(),
|
||||
getVisible: () => true,
|
||||
});
|
||||
coordinator.start();
|
||||
expect(coordinator.getSnapshot().isLeader).toBe(true);
|
||||
return coordinator;
|
||||
}
|
||||
|
||||
describe("LeaderElection", () => {
|
||||
it("elects one visible leader and keeps the second visible tab as follower", () => {
|
||||
let now = 1_000;
|
||||
|
|
@ -157,3 +201,72 @@ describe("createStorageRelayShared", () => {
|
|||
unsubscribe();
|
||||
});
|
||||
});
|
||||
|
||||
describe("SharedPollingCoordinator", () => {
|
||||
afterEach(() => {
|
||||
vi.useRealTimers();
|
||||
});
|
||||
|
||||
it("dedupes same-resource publishes and rate limits trailing result broadcasts", () => {
|
||||
vi.useFakeTimers();
|
||||
vi.setSystemTime(1_000);
|
||||
const channel = new MemorySharedChannel();
|
||||
const coordinator = startLeaderCoordinator(channel);
|
||||
|
||||
coordinator.publish("company:live-runs", [{ id: "run-1", lastEventAt: "a" }], 100);
|
||||
expect(channel.posts.filter((message) => message.type === "result")).toEqual([
|
||||
{
|
||||
type: "result",
|
||||
key: "company:live-runs",
|
||||
from: "leader",
|
||||
at: 1_000,
|
||||
dataUpdatedAt: 100,
|
||||
data: [{ id: "run-1", lastEventAt: "a" }],
|
||||
},
|
||||
]);
|
||||
|
||||
coordinator.publish("company:live-runs", [{ lastEventAt: "a", id: "run-1" }], 100);
|
||||
coordinator.publish("company:live-runs", [{ id: "run-1", lastEventAt: "a" }], 101);
|
||||
expect(channel.posts.filter((message) => message.type === "result")).toHaveLength(1);
|
||||
|
||||
vi.setSystemTime(1_200);
|
||||
coordinator.publish("company:live-runs", [{ id: "run-1", lastEventAt: "b" }], 200);
|
||||
coordinator.publish("company:live-runs", [{ id: "run-1", lastEventAt: "b" }], 200);
|
||||
expect(channel.posts.filter((message) => message.type === "result")).toHaveLength(1);
|
||||
|
||||
vi.advanceTimersByTime(799);
|
||||
expect(channel.posts.filter((message) => message.type === "result")).toHaveLength(1);
|
||||
|
||||
vi.advanceTimersByTime(1);
|
||||
expect(channel.posts.filter((message) => message.type === "result")).toHaveLength(2);
|
||||
expect(channel.posts.at(-1)).toEqual({
|
||||
type: "result",
|
||||
key: "company:live-runs",
|
||||
from: "leader",
|
||||
at: 2_000,
|
||||
dataUpdatedAt: 200,
|
||||
data: [{ id: "run-1", lastEventAt: "b" }],
|
||||
});
|
||||
|
||||
coordinator.stop();
|
||||
});
|
||||
|
||||
it("preserves original result timestamps when reposting the latest result for a request", () => {
|
||||
vi.useFakeTimers();
|
||||
vi.setSystemTime(5_000);
|
||||
const channel = new MemorySharedChannel();
|
||||
const coordinator = startLeaderCoordinator(channel);
|
||||
|
||||
coordinator.publish("company:live-runs", [{ id: "run-1" }], 123);
|
||||
const original = channel.posts[0];
|
||||
|
||||
vi.setSystemTime(9_000);
|
||||
channel.emit({ type: "request", key: "company:live-runs", from: "follower", at: 9_000 });
|
||||
|
||||
expect(channel.posts[1]).toEqual({ ...original, from: "leader" });
|
||||
expect(channel.posts[1]?.at).toBe(5_000);
|
||||
expect(channel.posts[1]?.dataUpdatedAt).toBe(123);
|
||||
|
||||
coordinator.stop();
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -184,6 +184,27 @@ export class LeaderElection {
|
|||
}
|
||||
}
|
||||
|
||||
function stableFingerprint(value: unknown): string {
|
||||
const seen = new WeakSet<object>();
|
||||
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;
|
||||
}, {});
|
||||
});
|
||||
} catch {
|
||||
return String(value);
|
||||
}
|
||||
}
|
||||
|
||||
/** Build a `LeaseStore` backed by a Storage-like object under one key. */
|
||||
export function createLeaseStore(
|
||||
storage: Pick<Storage, "getItem" | "setItem" | "removeItem">,
|
||||
|
|
@ -242,6 +263,8 @@ export interface SharedMessage {
|
|||
from: string;
|
||||
/** Epoch ms the payload was produced. */
|
||||
at: number;
|
||||
/** React Query `dataUpdatedAt` for result freshness comparisons. */
|
||||
dataUpdatedAt?: number;
|
||||
/** Present on `result` messages. */
|
||||
data?: unknown;
|
||||
}
|
||||
|
|
@ -364,11 +387,13 @@ export interface SharedPollingCoordinatorOptions {
|
|||
election?: LeaderElection;
|
||||
leaseTtlMs?: number;
|
||||
tickMs?: number;
|
||||
publishDebounceMs?: number;
|
||||
now?: () => number;
|
||||
getVisible?: () => boolean;
|
||||
}
|
||||
|
||||
const DEFAULT_COORDINATOR_TICK_MS = 1_000;
|
||||
const DEFAULT_PUBLISH_DEBOUNCE_MS = 1_000;
|
||||
const TAB_ID_STORAGE_KEY = "paperclip:shared-poll:tab-id";
|
||||
|
||||
function sanitizeCompanyId(companyId: string): string {
|
||||
|
|
@ -434,10 +459,23 @@ export class SharedPollingCoordinator {
|
|||
private readonly election: LeaderElection;
|
||||
private readonly localOnlyFallback: boolean;
|
||||
private readonly tickMs: number;
|
||||
private readonly publishDebounceMs: number;
|
||||
private readonly now: () => number;
|
||||
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 lastPublished = new Map<string, {
|
||||
dataUpdatedAt: number;
|
||||
fingerprint: string;
|
||||
sentAt: number;
|
||||
}>();
|
||||
private readonly pendingPublishes = new Map<string, {
|
||||
data: unknown;
|
||||
dataUpdatedAt: number;
|
||||
fingerprint: string;
|
||||
timer: ReturnType<typeof setTimeout>;
|
||||
}>();
|
||||
private unsubscribeChannel: (() => void) | null = null;
|
||||
private intervalId: ReturnType<typeof setInterval> | null = null;
|
||||
private releaseListeners: Array<() => void> = [];
|
||||
|
|
@ -447,12 +485,14 @@ export class SharedPollingCoordinator {
|
|||
this.tabId = options.tabId ?? getSharedPollingTabId();
|
||||
this.channel = options.channel ?? createSharedChannel(sharedPollingChannelName(companyId));
|
||||
this.tickMs = options.tickMs ?? DEFAULT_COORDINATOR_TICK_MS;
|
||||
this.publishDebounceMs = options.publishDebounceMs ?? DEFAULT_PUBLISH_DEBOUNCE_MS;
|
||||
this.now = options.now ?? (() => Date.now());
|
||||
this.getVisible = options.getVisible ?? getBrowserVisible;
|
||||
const leaseStore = createBrowserLeaseStore(companyId);
|
||||
this.localOnlyFallback = !options.election && !leaseStore;
|
||||
this.election = options.election ?? new LeaderElection(
|
||||
{
|
||||
now: options.now ?? (() => Date.now()),
|
||||
now: this.now,
|
||||
store: leaseStore ?? {
|
||||
read: () => null,
|
||||
write: () => {},
|
||||
|
|
@ -485,6 +525,7 @@ export class SharedPollingCoordinator {
|
|||
}
|
||||
this.unsubscribeChannel?.();
|
||||
this.unsubscribeChannel = null;
|
||||
this.clearPendingPublishes();
|
||||
for (const release of this.releaseListeners) release();
|
||||
this.releaseListeners = [];
|
||||
this.election.release();
|
||||
|
|
@ -520,23 +561,80 @@ export class SharedPollingCoordinator {
|
|||
type: "request",
|
||||
key,
|
||||
from: this.tabId,
|
||||
at: Date.now(),
|
||||
at: this.now(),
|
||||
});
|
||||
}
|
||||
|
||||
publish(key: string, data: unknown): void {
|
||||
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 pending = this.pendingPublishes.get(key);
|
||||
if (pending) {
|
||||
if (dataUpdatedAt < pending.dataUpdatedAt) return;
|
||||
if (dataUpdatedAt === pending.dataUpdatedAt && fingerprint === pending.fingerprint) return;
|
||||
this.pendingPublishes.set(key, {
|
||||
...pending,
|
||||
data,
|
||||
dataUpdatedAt,
|
||||
fingerprint,
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
const elapsedSinceLastPublish = last ? this.now() - last.sentAt : Number.POSITIVE_INFINITY;
|
||||
if (elapsedSinceLastPublish >= this.publishDebounceMs) {
|
||||
this.postResult(key, data, dataUpdatedAt, fingerprint);
|
||||
return;
|
||||
}
|
||||
|
||||
const delay = Math.max(this.publishDebounceMs - elapsedSinceLastPublish, 0);
|
||||
const timer = setTimeout(() => this.flushPendingPublish(key), delay);
|
||||
this.pendingPublishes.set(key, { data, dataUpdatedAt, fingerprint, timer });
|
||||
}
|
||||
|
||||
private postResult(key: string, data: unknown, dataUpdatedAt: number, fingerprint: string): void {
|
||||
const sentAt = this.now();
|
||||
const message: SharedMessage = {
|
||||
type: "result",
|
||||
key,
|
||||
from: this.tabId,
|
||||
at: Date.now(),
|
||||
at: sentAt,
|
||||
dataUpdatedAt,
|
||||
data,
|
||||
};
|
||||
this.lastPublished.set(key, { dataUpdatedAt, fingerprint, sentAt });
|
||||
this.latestResults.set(key, message);
|
||||
this.channel.post(message);
|
||||
}
|
||||
|
||||
private flushPendingPublish(key: string): void {
|
||||
const pending = this.pendingPublishes.get(key);
|
||||
if (!pending) return;
|
||||
this.pendingPublishes.delete(key);
|
||||
this.postResult(key, pending.data, pending.dataUpdatedAt, pending.fingerprint);
|
||||
}
|
||||
|
||||
private cancelPendingPublish(key: string): void {
|
||||
const pending = this.pendingPublishes.get(key);
|
||||
if (!pending) return;
|
||||
clearTimeout(pending.timer);
|
||||
this.pendingPublishes.delete(key);
|
||||
}
|
||||
|
||||
private clearPendingPublishes(): void {
|
||||
for (const pending of this.pendingPublishes.values()) clearTimeout(pending.timer);
|
||||
this.pendingPublishes.clear();
|
||||
}
|
||||
|
||||
private tick(): void {
|
||||
if (this.localOnlyFallback) {
|
||||
this.setSnapshot({ isLeader: this.getVisible() });
|
||||
|
|
@ -551,7 +649,7 @@ export class SharedPollingCoordinator {
|
|||
if (message.type === "request") {
|
||||
if (!this.snapshot.isLeader) return;
|
||||
const latest = this.latestResults.get(message.key);
|
||||
if (latest) this.channel.post({ ...latest, from: this.tabId, at: Date.now() });
|
||||
if (latest) this.channel.post({ ...latest, from: this.tabId });
|
||||
return;
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue