mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 12:37:55 +01:00
fix(ui): stop refetching the @task mention pool on every task event (#19931)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - The board UI keeps its caches fresh from live events in `LiveUpdatesProvider` > - Each task activity event invalidates the whole `["issues", companyId]` query prefix > - That prefix also holds the `@task` mention pool: the 100 most recently updated tasks, about 315 KB per response > - A chat turn emits many comment, status and run events, so an open task page downloads that list again for almost every event > - This pull request keeps the pool fresh from push events without a refetch per event: a closed server flag marks real membership or title changes, progress on pooled tasks reorders the cache in place, and other progress refetches at most once per 30 seconds > - The benefit is less bandwidth and CPU for users with an open task or chat, and smaller eval traces ## Linked Issues or Issue Description **What happened** While a task chat page is open, the browser repeats `GET /api/companies/:id/issues?limit=100&sortField=updated&sortDir=desc` (about 315 KB each). One 10-minute chat-eval attempt recorded about 194 of these requests. The trigger is `invalidateActivityQueries` in `ui/src/context/LiveUpdatesProvider.tsx`. For every `entityType: "issue"` activity event, it invalidates `queryKeys.issues.list(companyId)`. That prefix matches `queryKeys.issues.mentionPool(companyId)`, which `IssueDetail` (and `NewIssueDialog`) keep active for the `@task` picker. Invalidation refetches active queries and ignores `staleTime`. **Expected behavior** Comments, run progress and status changes on any task do not download the full mention pool again. The pool still updates when a task that the picker can show is created, deleted or renamed. **Steps to reproduce** 1. Run `pnpm dev` and create a company with about 140 tasks. 2. Open a task page, for example `/REF/issues/REF-1`. 3. Every 4 seconds for 15 rounds, post comments on the open task, on a task in the top 100, and on a task outside the top 100, and change the open task's priority (60 events). 4. Count requests to `issues?limit=100&sortField=updated`. Before this change there are 15 (about 3.3 MB). After this change there are 2 (about 480 KB). ## What Changed - New shared helper `issueActivityChangesTaskIdentity` (`packages/shared/src/issue-activity-identity.ts`). It is true for a created, deleted, hidden, unhidden or renamed task. - `server/src/realtime/live-events-ws.ts`: task `activity.logged` events now carry the closed boolean `taskIdentityChanged`. Authenticated sockets still get no `details`; the flag carries no task content. - `ui/src/context/LiveUpdatesProvider.tsx` refreshes the mention pool this way: - The task identity flag is true (or full details show a create, delete, hide, unhide or rename): the pool refetches at once. - Other events on a task already in the pool: the task moves to the front of the cached list. There is no request. The freshness clock does not change, and a pending invalidation is never overwritten. - Other events on a task outside the pool: one trailing refetch at most every 30 seconds, because the task can now be in the top 100. - The picker is closed: the pool is only marked stale. - `ui/src/lib/query-invalidation-batcher.ts`: new `excludingQueryKeys()` keyed exclusion predicate. The batcher coalesces it and uses it for prefix coverage. The broad `["issues", companyId]` invalidation uses it to skip the pool, so a burst is again one broad refresh per batch. - Tests: `ui/src/context/LiveUpdatesProvider.mention-pool.test.ts` (authenticated event shape, hide and unhide, the 30-second window, reorder safety, a burst through the batcher), new batcher tests, `packages/shared/src/issue-activity-identity.test.ts`, and socket assertions in `server/src/__tests__/live-events-toast.test.ts`. ## Verification - `cd ui && npx vitest run src/context/LiveUpdatesProvider src/lib/query-invalidation-batcher`: 231 tests pass. - `cd server && npx vitest run src/__tests__/live-events-toast.test.ts`: 11 tests pass. In authenticated mode the socket keeps `details` off and sends `taskIdentityChanged` (false for routine updates, true for a create, rename or hide). - `cd packages/shared && npx vitest run src/issue-activity-identity.test.ts`: 14 tests pass. - The new UI tests fail against the first commit of this PR: the authenticated burst, hide, unhide, the 30-second window and the batcher burst. - `tsc --noEmit` for ui, server and shared: no new errors. - Browser measurement: a local `pnpm dev` instance (local trusted), headless Chromium on a task page, 141 tasks, 60 events over about 63 seconds (the reproduction steps above): - Before (master UI): **15** mention-pool requests, about 3.3 MB. - After: **2** requests, about 480 KB (one per 30-second window). - Freshness after the fix: one rename, hide, unhide or create each caused **1** refetch. A priority change caused **0**. - I did not run a browser session in authenticated mode. The authenticated event shape is covered by the socket test and the UI unit tests. ## Risks - A task that moves into the top 100 can be missing from the `@task` picker for up to 30 seconds. - The cached order is approximate between refetches: a moved task goes to the front, but its `updatedAt` value is not changed. - Other queries under `["issues", companyId]` (descendants, created-from, summaries, onboarding first task) still refresh for each event batch. They are small, but they can get the same treatment later. ## Model Used - Anthropic Claude Opus 5.5 (`claude-opus-5-5`), with tool use and code execution in Claude Code. ## 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 - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
b4842df1e4
commit
2a2d3f6e23
9 files changed
+465
-10
No files matched your search
@@ -2894,3 +2894,7 @@ export { readProviderFailure, type ProviderFailure } from "./provider-failure.ge
|
||||
export type { BoardBootstrap, BoardBootstrapSession } from "./types/board-bootstrap.js";
|
||||
export { readBoardBootstrap } from "./board-bootstrap.js";
|
||||
export * from "./provider-rate-limits.js";
|
||||
export {
|
||||
TASK_IDENTITY_CHANGED_FIELD,
|
||||
issueActivityChangesTaskIdentity,
|
||||
} from "./issue-activity-identity.js";
|
||||
@@ -0,0 +1,28 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { issueActivityChangesTaskIdentity } from "./issue-activity-identity.js";
|
||||
|
||||
describe("issueActivityChangesTaskIdentity", () => {
|
||||
it.each([
|
||||
["issue.created", {}],
|
||||
["issue.child_created", null],
|
||||
["issue.deleted", undefined],
|
||||
["issue.updated", { title: "Renamed" }],
|
||||
["issue.updated", { hiddenAt: "2026-11-07T00:00:00.000Z" }],
|
||||
["issue.updated", { hiddenAt: null }],
|
||||
["issue.updated", { changes: { title: { from: "a", to: "b" } } }],
|
||||
["issue.updated", { changes: { hiddenAt: { from: null, to: "x" } } }],
|
||||
])("is true for %s %j", (action, details) => {
|
||||
expect(issueActivityChangesTaskIdentity(action, details)).toBe(true);
|
||||
});
|
||||
|
||||
it.each([
|
||||
["issue.updated", { status: "in_progress", source: "comment" }],
|
||||
["issue.updated", { changes: { status: { from: "todo", to: "done" } } }],
|
||||
["issue.updated", null],
|
||||
["issue.comment_added", { title: "not a rename" }],
|
||||
["issue.checked_out", {}],
|
||||
[undefined, {}],
|
||||
])("is false for %s %j", (action, details) => {
|
||||
expect(issueActivityChangesTaskIdentity(action, details)).toBe(false);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,32 @@
|
||||
/**
|
||||
* Whether a task activity changes which tasks a viewer can list or how they
|
||||
* are named: a created, deleted, hidden, unhidden or renamed task.
|
||||
*
|
||||
* Live events reach authenticated browsers without `details`, so the server
|
||||
* computes this closed boolean from the full activity and sends it as
|
||||
* `taskIdentityChanged`. It carries no task content. Clients that receive full
|
||||
* details (local trusted mode) can compute the same answer themselves.
|
||||
*/
|
||||
export const TASK_IDENTITY_CHANGED_FIELD = "taskIdentityChanged" as const;
|
||||
|
||||
const TASK_IDENTITY_ACTIONS = new Set(["issue.created", "issue.child_created", "issue.deleted"]);
|
||||
const TASK_IDENTITY_FIELDS = ["title", "hiddenAt"] as const;
|
||||
|
||||
function asRecord(value: unknown): Record<string, unknown> | null {
|
||||
return value && typeof value === "object" && !Array.isArray(value)
|
||||
? (value as Record<string, unknown>)
|
||||
: null;
|
||||
}
|
||||
|
||||
export function issueActivityChangesTaskIdentity(
|
||||
action: unknown,
|
||||
details: unknown,
|
||||
): boolean {
|
||||
if (typeof action !== "string") return false;
|
||||
if (TASK_IDENTITY_ACTIONS.has(action)) return true;
|
||||
if (action !== "issue.updated") return false;
|
||||
const record = asRecord(details);
|
||||
if (!record) return false;
|
||||
const changes = asRecord(record.changes);
|
||||
return TASK_IDENTITY_FIELDS.some((field) => field in record || (!!changes && field in changes));
|
||||
}
|
||||
@@ -128,6 +128,8 @@ describe("authenticated activity notifications", () => {
|
||||
if (actorType === "agent") expect(event.payload.actorName).toBe("Coder");
|
||||
if (actorType === "user") expect(event.payload.actorName).toBe("Alex Example");
|
||||
if (deploymentMode === "authenticated") expect(event.payload).not.toHaveProperty("details");
|
||||
// Routine updates are not task identity changes.
|
||||
expect(event.payload.taskIdentityChanged).toBe(false);
|
||||
}
|
||||
expect(messages[3].payload.issueIdentifier).toBe("PAP-1234");
|
||||
} finally {
|
||||
@@ -260,6 +262,16 @@ describe("authenticated activity notifications", () => {
|
||||
await second;
|
||||
expect(messages[1].payload).toMatchObject({ issueIdentifier: "PAP-1234", actorName: "Coder", actorType: "agent", actorId: "agent", actorAppearance: { paletteId: "bubblegum-sky" } });
|
||||
if (deploymentMode === "authenticated") expect(messages[1].payload).not.toHaveProperty("details");
|
||||
expect(messages[0].payload.taskIdentityChanged).toBe(true);
|
||||
const third = once(socket, "message");
|
||||
publishLiveEvent({ companyId: "company", type: "activity.logged", payload: {
|
||||
action: "issue.updated", entityType: "issue", entityId: "visible", actorType: "user", actorId: "viewer",
|
||||
details: { title: "Secret new title", hiddenAt: null },
|
||||
} });
|
||||
await third;
|
||||
// A rename or hide is flagged with a closed boolean; its content stays off the socket.
|
||||
expect(messages[2].payload.taskIdentityChanged).toBe(true);
|
||||
if (deploymentMode === "authenticated") expect(JSON.stringify(messages[2].payload)).not.toContain("Secret new title");
|
||||
} finally {
|
||||
socket.terminate();
|
||||
await new Promise<void>((resolve) => server.close(() => resolve()));
|
||||
|
||||
@@ -7,7 +7,7 @@ import type { Duplex } from "node:stream";
|
||||
import { and, eq, isNull } from "drizzle-orm";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import { agentApiKeys, companyMemberships, instanceUserRoles, heartbeatRuns, projects } from "@paperclipai/db";
|
||||
import type { DeploymentMode, LiveEvent } from "@paperclipai/shared";
|
||||
import { issueActivityChangesTaskIdentity, type DeploymentMode, type LiveEvent } from "@paperclipai/shared";
|
||||
import type { BetterAuthSessionResult } from "../auth/better-auth.js";
|
||||
import { logger } from "../middleware/logger.js";
|
||||
import { activityToastIdentity } from "./activity-toast-identity.js";
|
||||
@@ -289,6 +289,7 @@ export function setupLiveEventsWebSocketServer(
|
||||
if (context!.actor.source === "local_implicit") {
|
||||
if (event.type === "activity.logged" && event.payload.entityType === "issue") {
|
||||
return { ...event, payload: { ...event.payload,
|
||||
taskIdentityChanged: issueActivityChangesTaskIdentity(event.payload.action, event.payload.details),
|
||||
...await activityToastIdentity(db, event.companyId, event.payload) } };
|
||||
}
|
||||
return event;
|
||||
@@ -351,6 +352,11 @@ export function setupLiveEventsWebSocketServer(
|
||||
// A handoff can attach the first case to a task. Its already-authorized
|
||||
// task identity lets an empty cases panel refresh without exposing content.
|
||||
...(payload.entityType === "case" && String(payload.action).startsWith("case.handoff_") && issueId ? { issueId } : {}),
|
||||
// A closed boolean derived from details, so task pickers can tell a
|
||||
// created, deleted, hidden or renamed task from routine progress.
|
||||
...(payload.entityType === "issue"
|
||||
? { taskIdentityChanged: issueActivityChangesTaskIdentity(payload.action, payload.details) }
|
||||
: {}),
|
||||
// The task read check above authorizes attribution, but not arbitrary details.
|
||||
...(issueId ? await activityToastIdentity(db, event.companyId, payload) : {}),
|
||||
} };
|
||||
|
||||
@@ -0,0 +1,174 @@
|
||||
// @vitest-environment node
|
||||
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { QueryClient, QueryObserver } from "@tanstack/react-query";
|
||||
import type { Issue } from "@paperclipai/shared";
|
||||
import { __liveUpdatesTestUtils } from "./LiveUpdatesProvider";
|
||||
import { createCoalescingQueryClient, createInvalidationBatcher } from "../lib/query-invalidation-batcher";
|
||||
import { queryKeys } from "../lib/queryKeys";
|
||||
|
||||
const COMPANY = "company-1";
|
||||
const ACTOR = { userId: "user-1", agentId: null };
|
||||
const POOL_KEY = queryKeys.issues.mentionPool(COMPANY);
|
||||
const WINDOW_MS = __liveUpdatesTestUtils.MENTION_POOL_REFRESH_WINDOW_MS;
|
||||
|
||||
function task(id: string): Issue {
|
||||
return { id, identifier: id.toUpperCase(), title: `Task ${id}` } as Issue;
|
||||
}
|
||||
|
||||
function mountQuery(client: QueryClient, queryKey: readonly unknown[], data: unknown[] = []) {
|
||||
const queryFn = vi.fn(async () => data);
|
||||
const observer = new QueryObserver(client, { queryKey, queryFn, staleTime: 60_000 });
|
||||
const unsubscribe = observer.subscribe(() => {});
|
||||
return { queryFn, unsubscribe };
|
||||
}
|
||||
|
||||
async function settle(ms = 0) {
|
||||
await vi.advanceTimersByTimeAsync(ms);
|
||||
}
|
||||
|
||||
/** Authenticated sockets carry no `details`, only the closed identity flag. */
|
||||
function authenticatedEvent(entityId: string, action: string, taskIdentityChanged: boolean) {
|
||||
return { entityType: "issue", entityId, action, actorType: "agent", actorId: "agent-1", taskIdentityChanged };
|
||||
}
|
||||
|
||||
function emit(client: QueryClient, payload: Record<string, unknown>) {
|
||||
__liveUpdatesTestUtils.invalidateActivityQueries(client, COMPANY, { entityType: "issue", ...payload }, ACTOR);
|
||||
}
|
||||
|
||||
const POOL = [task("t1"), task("t2"), task("t3")];
|
||||
|
||||
describe("LiveUpdatesProvider @task mention pool", () => {
|
||||
let client: QueryClient;
|
||||
beforeEach(() => {
|
||||
vi.useFakeTimers();
|
||||
client = new QueryClient();
|
||||
});
|
||||
afterEach(() => {
|
||||
client.clear();
|
||||
vi.useRealTimers();
|
||||
});
|
||||
|
||||
it("does not refetch for a burst of authenticated progress events on pooled tasks, and reorders in place", async () => {
|
||||
const pool = mountQuery(client, POOL_KEY, POOL);
|
||||
const otherList = mountQuery(client, [...queryKeys.issues.list(COMPANY), "descendants", "t1"]);
|
||||
await settle();
|
||||
expect(pool.queryFn).toHaveBeenCalledTimes(1);
|
||||
|
||||
for (let i = 0; i < 25; i += 1) {
|
||||
emit(client, authenticatedEvent("t3", "issue.comment_added", false));
|
||||
emit(client, authenticatedEvent("t3", "issue.updated", false));
|
||||
emit(client, authenticatedEvent("t2", "issue.checked_out", false));
|
||||
await settle();
|
||||
}
|
||||
await settle(WINDOW_MS * 2);
|
||||
|
||||
expect(pool.queryFn).toHaveBeenCalledTimes(1);
|
||||
expect(client.getQueryData<Issue[]>(POOL_KEY)?.map((issue) => issue.id)).toEqual(["t2", "t3", "t1"]);
|
||||
// Other company-scoped issue queries keep their live refresh.
|
||||
expect(otherList.queryFn.mock.calls.length).toBeGreaterThan(1);
|
||||
pool.unsubscribe();
|
||||
otherList.unsubscribe();
|
||||
});
|
||||
|
||||
it.each([
|
||||
["an authenticated identity change", authenticatedEvent("t9", "issue.updated", true)],
|
||||
["a created task", { entityId: "t9", action: "issue.created", details: { identifier: "T9" } }],
|
||||
["a deleted task", { entityId: "t1", action: "issue.deleted", details: {} }],
|
||||
["a renamed task", { entityId: "t1", action: "issue.updated", details: { title: "Renamed" } }],
|
||||
["a hidden task", { entityId: "t1", action: "issue.updated", details: { hiddenAt: "2026-11-07T00:00:00.000Z" } }],
|
||||
["an unhidden task", { entityId: "t1", action: "issue.updated", details: { hiddenAt: null } }],
|
||||
])("refetches immediately for %s", async (_label, payload) => {
|
||||
const pool = mountQuery(client, POOL_KEY, POOL);
|
||||
await settle();
|
||||
emit(client, payload);
|
||||
await settle();
|
||||
expect(pool.queryFn).toHaveBeenCalledTimes(2);
|
||||
pool.unsubscribe();
|
||||
});
|
||||
|
||||
it("refetches at most once per window when events touch tasks outside the pool", async () => {
|
||||
const pool = mountQuery(client, POOL_KEY, POOL);
|
||||
await settle();
|
||||
|
||||
for (let i = 0; i < 20; i += 1) {
|
||||
emit(client, authenticatedEvent(`outside-${i}`, "issue.comment_added", false));
|
||||
await settle(100);
|
||||
}
|
||||
expect(pool.queryFn).toHaveBeenCalledTimes(1);
|
||||
|
||||
// The trailing refresh lands one window after the last fetch.
|
||||
await settle(WINDOW_MS);
|
||||
expect(pool.queryFn).toHaveBeenCalledTimes(2);
|
||||
|
||||
// Quiet afterwards: no further refetch without new events.
|
||||
await settle(WINDOW_MS * 2);
|
||||
expect(pool.queryFn).toHaveBeenCalledTimes(2);
|
||||
pool.unsubscribe();
|
||||
});
|
||||
|
||||
it("only marks an unmounted pool stale", async () => {
|
||||
const pool = mountQuery(client, POOL_KEY, POOL);
|
||||
await settle();
|
||||
pool.unsubscribe();
|
||||
emit(client, authenticatedEvent("outside", "issue.updated", false));
|
||||
await settle(WINDOW_MS * 2);
|
||||
expect(pool.queryFn).toHaveBeenCalledTimes(1);
|
||||
expect(client.getQueryState(POOL_KEY)?.isInvalidated).toBe(true);
|
||||
});
|
||||
|
||||
it("keeps a pending refresh when a later event reorders a closed picker's pool", async () => {
|
||||
const pool = mountQuery(client, POOL_KEY, POOL);
|
||||
await settle();
|
||||
pool.unsubscribe();
|
||||
const fetchedAt = client.getQueryState(POOL_KEY)?.dataUpdatedAt;
|
||||
// A rename while the picker is closed marks the pool stale without a fetch.
|
||||
emit(client, { entityId: "t1", action: "issue.updated", details: { title: "Renamed" } });
|
||||
emit(client, authenticatedEvent("t3", "issue.comment_added", false));
|
||||
await settle();
|
||||
const state = client.getQueryState(POOL_KEY);
|
||||
expect(state?.isInvalidated).toBe(true);
|
||||
expect(state?.dataUpdatedAt).toBe(fetchedAt);
|
||||
|
||||
// Reopening the picker within staleTime still fetches the renamed list.
|
||||
const reopened = mountQuery(client, POOL_KEY, POOL);
|
||||
await settle();
|
||||
expect(reopened.queryFn).toHaveBeenCalledTimes(1);
|
||||
reopened.unsubscribe();
|
||||
});
|
||||
|
||||
it("does not reset the freshness clock when it reorders in place", async () => {
|
||||
const pool = mountQuery(client, POOL_KEY, POOL);
|
||||
await settle();
|
||||
const fetchedAt = client.getQueryState(POOL_KEY)?.dataUpdatedAt;
|
||||
await settle(5_000);
|
||||
emit(client, authenticatedEvent("t3", "issue.comment_added", false));
|
||||
await settle();
|
||||
expect(client.getQueryData<Issue[]>(POOL_KEY)?.[0]?.id).toBe("t3");
|
||||
expect(client.getQueryState(POOL_KEY)?.dataUpdatedAt).toBe(fetchedAt);
|
||||
pool.unsubscribe();
|
||||
});
|
||||
|
||||
it("coalesces a burst through the live-event batcher into one broad refresh", async () => {
|
||||
const batcher = createInvalidationBatcher(client);
|
||||
const coalescing = createCoalescingQueryClient(client, batcher);
|
||||
const invalidate = vi.spyOn(client, "invalidateQueries");
|
||||
const pool = mountQuery(client, POOL_KEY, POOL);
|
||||
const otherList = mountQuery(client, [...queryKeys.issues.list(COMPANY), "descendants", "t1"]);
|
||||
await settle();
|
||||
|
||||
for (let i = 0; i < 30; i += 1) {
|
||||
__liveUpdatesTestUtils.invalidateActivityQueries(coalescing, COMPANY, authenticatedEvent("t1", "issue.comment_added", false), ACTOR);
|
||||
}
|
||||
await settle(1_000);
|
||||
|
||||
const broad = invalidate.mock.calls.filter(([filters]) =>
|
||||
JSON.stringify(filters?.queryKey) === JSON.stringify(queryKeys.issues.list(COMPANY)));
|
||||
expect(broad).toHaveLength(1);
|
||||
expect(otherList.queryFn).toHaveBeenCalledTimes(2);
|
||||
expect(pool.queryFn).toHaveBeenCalledTimes(1);
|
||||
batcher.dispose();
|
||||
pool.unsubscribe();
|
||||
otherList.unsubscribe();
|
||||
});
|
||||
});
|
||||
@@ -1,5 +1,5 @@
|
||||
import { LiveTranscriptInterests, DISCONNECTED_TRANSCRIPT_STREAM } from "../lib/live-transcript-interests";
|
||||
import { resolveAgentAppearance } from "@paperclipai/shared";
|
||||
import { resolveAgentAppearance, TASK_IDENTITY_CHANGED_FIELD, issueActivityChangesTaskIdentity } from "@paperclipai/shared";
|
||||
import { getPageVisibility, usePageVisibility } from "../lib/page-visibility";
|
||||
import { createLiveQueryRecovery } from "../lib/live-query-recovery";
|
||||
import {
|
||||
@@ -22,6 +22,7 @@ import {
|
||||
import {
|
||||
createCoalescingQueryClient,
|
||||
createInvalidationBatcher,
|
||||
excludingQueryKeys,
|
||||
} from "../lib/query-invalidation-batcher";
|
||||
import {
|
||||
patchRunStatusInList,
|
||||
@@ -970,6 +971,96 @@ const ISSUE_TOAST_ACTIONS = new Set([
|
||||
"issue.updated",
|
||||
"issue.comment_added",
|
||||
]);
|
||||
/**
|
||||
* The `@task` mention pool is the 100 most recently updated tasks (one large
|
||||
* response) and only backs a client-side picker. A chat turn emits many task
|
||||
* events, and refetching the pool for each one re-downloads that list again
|
||||
* and again. The pool is kept fresh this way instead:
|
||||
*
|
||||
* - A created, deleted, hidden, unhidden or renamed task changes what the
|
||||
* picker can offer, so the pool refetches at once. Authenticated sockets
|
||||
* send no `details`; the server sends the closed `taskIdentityChanged`
|
||||
* boolean instead, and full-detail events are classified the same way.
|
||||
* - Any other event on a task already in the pool moves it to the front in
|
||||
* place. No request is needed: its title and identifier have not changed.
|
||||
* - Any other event on a task outside the pool may lift it into the top 100.
|
||||
* The pool refetches at most once per `MENTION_POOL_REFRESH_WINDOW_MS`,
|
||||
* trailing, so the picker catches up within that window without a refetch
|
||||
* per event. Until then a task that just entered the top 100 may be missing.
|
||||
*/
|
||||
const MENTION_POOL_REFRESH_WINDOW_MS = 30_000;
|
||||
const mentionPoolRefreshes = new WeakMap<
|
||||
object,
|
||||
Map<string, { lastRequestedAt: number; timer: ReturnType<typeof setTimeout> | null }>
|
||||
>();
|
||||
|
||||
function activityChangesTaskIdentity(
|
||||
payload: Record<string, unknown>,
|
||||
action: string | null,
|
||||
details: Record<string, unknown> | null,
|
||||
): boolean {
|
||||
const flag = payload[TASK_IDENTITY_CHANGED_FIELD];
|
||||
if (typeof flag === "boolean") return flag;
|
||||
return issueActivityChangesTaskIdentity(action, details);
|
||||
}
|
||||
|
||||
function refreshMentionPoolForActivity(
|
||||
queryClient: QueryClient,
|
||||
companyId: string,
|
||||
entityId: string | null,
|
||||
identityChanged: boolean,
|
||||
) {
|
||||
const queryKey = queryKeys.issues.mentionPool(companyId);
|
||||
if (identityChanged) {
|
||||
queryClient.invalidateQueries({ queryKey });
|
||||
return;
|
||||
}
|
||||
const pool = queryClient.getQueryData<Issue[]>(queryKey);
|
||||
if (!pool) return;
|
||||
const query = queryClient.getQueryCache().find({ queryKey, exact: true });
|
||||
if (!query) return;
|
||||
// A pending invalidation (for example a rename while the picker was closed)
|
||||
// must survive: writing data would clear it and make the old list look fresh.
|
||||
if (query.state.isInvalidated) return;
|
||||
const index = entityId
|
||||
? pool.findIndex((issue) => issue.id === entityId || issue.identifier === entityId)
|
||||
: -1;
|
||||
if (query.getObserversCount() === 0) {
|
||||
// Nobody shows the picker. A task outside the pool may now belong in it, so
|
||||
// refetch on the next mount; a reorder alone can wait for that mount too.
|
||||
if (index < 0) queryClient.invalidateQueries({ queryKey, refetchType: "none" });
|
||||
return;
|
||||
}
|
||||
if (index >= 0) {
|
||||
if (index > 0) {
|
||||
// Keep the freshness clock: a reorder is not a fetch.
|
||||
queryClient.setQueryData<Issue[]>(
|
||||
queryKey,
|
||||
[pool[index], ...pool.slice(0, index), ...pool.slice(index + 1)],
|
||||
{ updatedAt: query.state.dataUpdatedAt },
|
||||
);
|
||||
}
|
||||
return;
|
||||
}
|
||||
let refresh = mentionPoolRefreshes.get(queryClient);
|
||||
if (!refresh) {
|
||||
refresh = new Map();
|
||||
mentionPoolRefreshes.set(queryClient, refresh);
|
||||
}
|
||||
const state = refresh.get(companyId) ?? { lastRequestedAt: 0, timer: null };
|
||||
refresh.set(companyId, state);
|
||||
if (state.timer !== null) return;
|
||||
// Count both the last completed fetch and the last one requested here, so a
|
||||
// burst that arrives before the refetch finishes cannot start another.
|
||||
const lastRefresh = Math.max(query.state.dataUpdatedAt, state.lastRequestedAt);
|
||||
const wait = Math.max(0, lastRefresh + MENTION_POOL_REFRESH_WINDOW_MS - Date.now());
|
||||
state.timer = setTimeout(() => {
|
||||
state.timer = null;
|
||||
state.lastRequestedAt = Date.now();
|
||||
queryClient.invalidateQueries({ queryKey });
|
||||
}, wait);
|
||||
}
|
||||
|
||||
const ISSUE_DOCUMENT_ACTIVITY_ACTIONS = new Set([
|
||||
"issue.document_created",
|
||||
"issue.document_updated",
|
||||
@@ -1500,7 +1591,14 @@ function invalidateActivityQueries(
|
||||
}
|
||||
queryClient.invalidateQueries({
|
||||
queryKey: queryKeys.issues.list(companyId),
|
||||
predicate: excludingQueryKeys(queryKeys.issues.mentionPool(companyId)),
|
||||
});
|
||||
refreshMentionPoolForActivity(
|
||||
queryClient,
|
||||
companyId,
|
||||
entityId,
|
||||
activityChangesTaskIdentity(payload, action, details),
|
||||
);
|
||||
queryClient.invalidateQueries({
|
||||
queryKey: queryKeys.issues.listMineByMe(companyId),
|
||||
});
|
||||
@@ -2097,6 +2195,7 @@ function closeSocketQuietly(
|
||||
}
|
||||
|
||||
export const __liveUpdatesTestUtils = {
|
||||
MENTION_POOL_REFRESH_WINDOW_MS,
|
||||
buildActivityToast,
|
||||
applyRunLifecycleToCompanyLiveRuns,
|
||||
buildAgentStatusToast,
|
||||
|
||||
@@ -3,6 +3,7 @@ import { QueryClient, QueryObserver } from "@tanstack/react-query";
|
||||
import {
|
||||
createCoalescingQueryClient,
|
||||
createInvalidationBatcher,
|
||||
excludingQueryKeys,
|
||||
} from "./query-invalidation-batcher";
|
||||
|
||||
function fakeClient() {
|
||||
@@ -264,3 +265,57 @@ describe("overlapping live invalidations", () => {
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
describe("excludingQueryKeys", () => {
|
||||
beforeEach(() => vi.useFakeTimers());
|
||||
afterEach(() => vi.useRealTimers());
|
||||
|
||||
it("coalesces a burst of keyed exclusions into one invalidation per window", () => {
|
||||
const { client } = fakeClient();
|
||||
const batcher = createInvalidationBatcher(client, 300);
|
||||
for (let i = 0; i < 25; i += 1) {
|
||||
batcher.schedule({ queryKey: ["issues", "c1"], predicate: excludingQueryKeys(["issues", "c1", "mention-pool"]) });
|
||||
}
|
||||
vi.advanceTimersByTime(300);
|
||||
expect(client.invalidateQueries).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("keeps different exclusions separate", () => {
|
||||
const { client } = fakeClient();
|
||||
const batcher = createInvalidationBatcher(client, 300);
|
||||
batcher.schedule({ queryKey: ["issues", "c1"], predicate: excludingQueryKeys(["issues", "c1", "a"]) });
|
||||
batcher.schedule({ queryKey: ["issues", "c1"], predicate: excludingQueryKeys(["issues", "c1", "b"]) });
|
||||
vi.advanceTimersByTime(300);
|
||||
expect(client.invalidateQueries).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("covers non-excluded children but never the excluded subtree", () => {
|
||||
const { client, calls } = fakeClient();
|
||||
const batcher = createInvalidationBatcher(client, 300);
|
||||
const broad = { queryKey: ["issues", "c1"], predicate: excludingQueryKeys(["issues", "c1", "mention-pool"]) };
|
||||
batcher.schedule(broad);
|
||||
batcher.schedule({ queryKey: ["issues", "c1", "search", "q"] });
|
||||
batcher.schedule({ queryKey: ["issues", "c1", "mention-pool"] });
|
||||
batcher.schedule({ queryKey: ["issues", "c1"] });
|
||||
vi.advanceTimersByTime(300);
|
||||
// The plain prefix is not covered by the exclusion filter: it also reaches the pool.
|
||||
expect(calls).toEqual([{ queryKey: ["issues", "c1"] }]);
|
||||
});
|
||||
|
||||
it("is covered by a plain shorter prefix and covers a sibling child", () => {
|
||||
const { client, calls } = fakeClient();
|
||||
const batcher = createInvalidationBatcher(client, 300);
|
||||
const broad = { queryKey: ["issues", "c1"], predicate: excludingQueryKeys(["issues", "c1", "mention-pool"]) };
|
||||
batcher.schedule(broad);
|
||||
batcher.schedule({ queryKey: ["issues", "c1", "search", "q"] });
|
||||
batcher.schedule({ queryKey: ["issues", "c1", "mention-pool"] });
|
||||
vi.advanceTimersByTime(300);
|
||||
expect(calls).toEqual([broad, { queryKey: ["issues", "c1", "mention-pool"] }]);
|
||||
});
|
||||
|
||||
it("skips excluded queries when matching", () => {
|
||||
const predicate = excludingQueryKeys(["issues", "c1", "mention-pool"]);
|
||||
expect(predicate({ queryKey: ["issues", "c1", "mention-pool"] })).toBe(false);
|
||||
expect(predicate({ queryKey: ["issues", "c1", "search"] })).toBe(true);
|
||||
});
|
||||
});
|
||||
@@ -98,6 +98,36 @@ export function createInvalidationBatcher(
|
||||
|
||||
let uniqueFilterCounter = 0;
|
||||
|
||||
const EXCLUDED_QUERY_KEYS = Symbol("excludedQueryKeys");
|
||||
type QueryKeyPrefix = readonly unknown[];
|
||||
type ExclusionPredicate = ((query: { queryKey: readonly unknown[] }) => boolean) & {
|
||||
[EXCLUDED_QUERY_KEYS]: readonly QueryKeyPrefix[];
|
||||
};
|
||||
|
||||
function startsWithKey(key: readonly unknown[], prefix: QueryKeyPrefix): boolean {
|
||||
if (prefix.length > key.length) return false;
|
||||
return prefix.every((part, index) => JSON.stringify(part) === JSON.stringify(key[index]));
|
||||
}
|
||||
|
||||
/**
|
||||
* A `predicate` that skips queries under the given key prefixes. Unlike an
|
||||
* arbitrary predicate, the batcher can read its exclusions, so repeated
|
||||
* invalidations with the same key and exclusions still coalesce into one, and
|
||||
* the filter still covers its non-excluded children in the same batch.
|
||||
*/
|
||||
export function excludingQueryKeys(...excluded: QueryKeyPrefix[]): ExclusionPredicate {
|
||||
const predicate = ((query: { queryKey: readonly unknown[] }) =>
|
||||
!excluded.some((prefix) => startsWithKey(query.queryKey, prefix))) as ExclusionPredicate;
|
||||
predicate[EXCLUDED_QUERY_KEYS] = excluded;
|
||||
return predicate;
|
||||
}
|
||||
|
||||
function readExclusions(filters: InvalidateQueryFilters): readonly QueryKeyPrefix[] | null {
|
||||
const predicate = filters.predicate as Partial<ExclusionPredicate> | undefined;
|
||||
if (typeof predicate !== "function") return [];
|
||||
return predicate[EXCLUDED_QUERY_KEYS] ?? null;
|
||||
}
|
||||
|
||||
function filterBehaviorKey(filters: InvalidateQueryFilters): string {
|
||||
return JSON.stringify([
|
||||
filters.type ?? "all",
|
||||
@@ -109,26 +139,39 @@ function filterBehaviorKey(filters: InvalidateQueryFilters): string {
|
||||
|
||||
/** A collection prefix already refreshes its children in the same batch. */
|
||||
function removeCoveredFilters(filters: InvalidateQueryFilters[]): InvalidateQueryFilters[] {
|
||||
const prefixesByBehavior = new Map<string, Set<string>>();
|
||||
// Each prefix keeps the exclusions of every filter that names it; a plain
|
||||
// prefix has none.
|
||||
const prefixesByBehavior = new Map<string, Map<string, Array<readonly QueryKeyPrefix[]>>>();
|
||||
const isScalar = (value: unknown) => value === null || typeof value === "string"
|
||||
|| typeof value === "boolean" || typeof value === "number" && Number.isFinite(value);
|
||||
for (const filter of filters) {
|
||||
if (filter.predicate || filter.exact || !filter.queryKey?.every(isScalar)) continue;
|
||||
const exclusions = readExclusions(filter);
|
||||
if (!exclusions || filter.exact || !filter.queryKey?.every(isScalar)) continue;
|
||||
const behavior = filterBehaviorKey(filter);
|
||||
const prefixes = prefixesByBehavior.get(behavior) ?? new Set<string>();
|
||||
prefixes.add(JSON.stringify(filter.queryKey));
|
||||
const prefixes = prefixesByBehavior.get(behavior) ?? new Map<string, Array<readonly QueryKeyPrefix[]>>();
|
||||
const keyJson = JSON.stringify(filter.queryKey);
|
||||
prefixes.set(keyJson, [...(prefixes.get(keyJson) ?? []), exclusions]);
|
||||
prefixesByBehavior.set(behavior, prefixes);
|
||||
}
|
||||
return filters.filter(filter => {
|
||||
if (filter.predicate || !filter.queryKey) return true;
|
||||
const ownExclusions = readExclusions(filter);
|
||||
if (!ownExclusions || !filter.queryKey) return true;
|
||||
const prefixes = prefixesByBehavior.get(filterBehaviorKey(filter));
|
||||
if (!prefixes) return true;
|
||||
const key = filter.queryKey;
|
||||
// An excluded subtree overlapping this filter's subtree is not refreshed by
|
||||
// the covering prefix, so that prefix does not cover this filter.
|
||||
const coveredBy = (exclusions: readonly QueryKeyPrefix[]) =>
|
||||
!exclusions.some((excluded) => startsWithKey(key, excluded) || (!filter.exact && startsWithKey(excluded, key)));
|
||||
// A non-exact filter covers itself, so only check its shorter prefixes.
|
||||
const last = filter.exact ? key.length : key.length - 1;
|
||||
// A filter with exclusions is also covered by a plain filter on its own key.
|
||||
const hasExclusions = ownExclusions.length > 0;
|
||||
const last = filter.exact || hasExclusions ? key.length : key.length - 1;
|
||||
for (let length = 0; length <= last; length += 1) {
|
||||
if (length > 0 && !isScalar(key[length - 1])) break;
|
||||
if (prefixes.has(JSON.stringify(key.slice(0, length)))) return false;
|
||||
const entries = prefixes.get(JSON.stringify(key.slice(0, length)));
|
||||
const sameKeyNonExact = length === key.length && !filter.exact;
|
||||
if (entries?.some((exclusions) => sameKeyNonExact ? exclusions.length === 0 : coveredBy(exclusions))) return false;
|
||||
}
|
||||
return true;
|
||||
});
|
||||
@@ -138,7 +181,8 @@ function serializeFilters(filters: InvalidateQueryFilters): string {
|
||||
// A predicate is an opaque function — two predicate-based filters can never be
|
||||
// proven equal, so never coalesce them (that would silently drop one). Give
|
||||
// each its own key.
|
||||
if (typeof filters.predicate === "function") {
|
||||
const exclusions = readExclusions(filters);
|
||||
if (exclusions === null) {
|
||||
return `__predicate__:${(uniqueFilterCounter += 1)}`;
|
||||
}
|
||||
// Query selection and refetch behavior both belong to the identity. In
|
||||
@@ -149,6 +193,7 @@ function serializeFilters(filters: InvalidateQueryFilters): string {
|
||||
filters.queryKey ?? null,
|
||||
filterBehaviorKey(filters),
|
||||
filters.exact ?? false,
|
||||
...(exclusions.length > 0 ? [{ excluding: exclusions }] : []),
|
||||
]);
|
||||
} catch {
|
||||
// Non-serializable filter (shouldn't happen for our query keys) — fall back
|
||||
|
||||
Reference in new issue
Block a user