From 255713e0cd0505a471cc9f0d264a3891da31a47c Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 20:02:36 -0500 Subject: [PATCH] fix(server): bound and coalesce issue retention invalidations Bound reverse ancestor discovery before materializing adjacency, deduplicate cycles and emit one canonical affected-issue batch per write. Overflow delegates complete company reconstruction to existing paging instead of dropping eligibility or creating per-ancestor worker debt. Prove real high-fanin producer bursts, PostgreSQL execution limits, canonical cyclic batches and far ancestor archive deadlines. Co-Authored-By: Paperclip --- .../2026-10-30-attention-retention-sleep.md | 2 + .../attention-retention-work.test.ts | 107 +++++++++++++++++- server/src/services/attention-retention.ts | 12 +- .../services/retention-work-notifications.ts | 29 +++-- 4 files changed, 134 insertions(+), 16 deletions(-) diff --git a/doc/plans/2026-10-30-attention-retention-sleep.md b/doc/plans/2026-10-30-attention-retention-sleep.md index 2be011ccf9d..92241aefc4f 100644 --- a/doc/plans/2026-10-30-attention-retention-sleep.md +++ b/doc/plans/2026-10-30-attention-retention-sleep.md @@ -46,3 +46,5 @@ The focused suite supports `PAPERCLIP_ATTENTION_RETENTION_TEST_DATABASE_URL`, va Local embedded PostgreSQL could not start on the test host because system semaphores were exhausted. The accepted runs use isolated PostgreSQL 17 databases. A mixed consumer run also exposed fixture contamination from a lifecycle suite that retains native finalization rows; isolated reruns are required and preserve the failed run in the review evidence. Broad Linux CI remains the required full-suite gate. Review the source parity fixtures and producer hints together. The intended outcome is unchanged archive state and notifications with no recurring retention SQL when the sources remain unchanged. No merge or deployment is part of this change. + +Issue producer invalidation has its own finite discovery budget. Reverse parent/blocker traversal returns at most 101 distinct UUIDs in one recursive query, with each relation adjacency capped at 101 unique edges before materialization. At most 100 affected issues produce one sorted canonical batch hint; an overflow sentinel produces one company hint and the existing complete paged reconstruction. UNION deduplication terminates cycles. A PostgreSQL EXPLAIN ANALYZE on 2,100 reverse edges proves the outer limit stops at 101 CTE rows while the indexed relation branch returns only 100 rows. Repeated identical writes coalesce by canonical key. The shared scheduler is unchanged: a broader company hint does not cancel already pending narrower scopes; those remain finite batches of at most 100 IDs, processed at the existing 25-scope pass limit. This permits redundant bounded reconstruction instead of discarding invalidations or introducing cursor/subsumption policy across other owners. A 151-node cyclic chain proves overflow still registers the far ancestor's exact archive deadline and settles quietly afterward. diff --git a/server/src/__tests__/attention-retention-work.test.ts b/server/src/__tests__/attention-retention-work.test.ts index fb2dca6f6b5..f06355e1ae7 100644 --- a/server/src/__tests__/attention-retention-work.test.ts +++ b/server/src/__tests__/attention-retention-work.test.ts @@ -1,5 +1,6 @@ import { randomUUID } from "node:crypto"; -import { and, eq, sql } from "drizzle-orm"; +import { and, eq, sql, type SQL } from "drizzle-orm"; +import { PgDialect } from "drizzle-orm/pg-core"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { subscribeDatabaseWork, authUsers, companyMemberships, agents, approvals, budgetIncidents, budgetPolicies, companies, createDb, applyPendingMigrations, decisionArchiveNotificationOutbox, decisionRetention, decisions, costEvents, heartbeatRunEvents, heartbeatRuns, issueApprovals, issueRecoveryActions, @@ -425,6 +426,106 @@ const delay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000; expect(await db.select().from(decisionArchiveNotificationOutbox)).toHaveLength(0); }); + it("bounds high-fanin ordinary issue writes to one coalescible hint", async () => { + const service = attentionRetentionService(db), original = service.load, load = vi.spyOn(service, "load"); + await start(service); await settle(); load.mockClear(); + const leaf = await insertIssue("in_review", { updatedAt: new Date(), assigneeUserId: "board" }); + await db.execute(sql`with parents as ( + insert into issues (id, company_id, title, status) + select gen_random_uuid(), ${companyId}::uuid, 'fan-in-' || n, 'todo' + from generate_series(1, 2100) n returning id) + insert into issue_relations (company_id, issue_id, related_issue_id, type) + select ${companyId}::uuid, ${leaf.id}::uuid, id, 'blocks' from parents`); + const hints: string[] = []; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.attentionRetention, (event, values) => { + if (event === "settled") hints.push(...values ?? []); + }); + let release!: () => void, entered = false; + const gate = new Promise(resolve => { release = resolve; }); + load.mockImplementationOnce(async (...args) => { const page = await original(...args); entered = true; await gate; return page; }); + statements.length = 0; + try { + await issueService(db).update(leaf.id, { title: "High fan-in ordinary edit 0" }); + await vi.waitFor(() => expect(entered).toBe(true), { timeout: 15_000 }); + for (let n = 1; n < 8; n++) await issueService(db).update(leaf.id, { title: `High fan-in ordinary edit ${n}` }); + } finally { release(); unsubscribe(); } + expect(statements.filter(query => query.includes("with recursive affected"))).toHaveLength(8); + expect(hints).toHaveLength(8); + expect(new Set(hints)).toEqual(new Set([JSON.stringify({ companyId })])); + await settle(); + expect(load.mock.calls.length).toBeLessThanOrEqual(3); + expect(load.mock.calls.every(([scope]) => scope?.companyId === companyId && !scope.issueId && !scope.issueIds)).toBe(true); + let query!: Parameters[0], returned = 0; + await db.transaction(async tx => { + const execute = tx.execute.bind(tx); + vi.spyOn(tx, "execute").mockImplementation(async statement => { + const result = await execute(statement); + if (new PgDialect().sqlToQuery(statement as SQL).sql.includes("with recursive affected")) { + query = statement; returned = result.length; + } + return result; + }); + await notifyIssueAttentionRetentionWork(tx, { companyId, issueId: leaf.id }); + }); + expect(returned).toBe(101); + const plans = await db.execute(sql`explain (analyze, format json) ${query as SQL}`); + const plan = (plans[0]["QUERY PLAN"] as any)[0].Plan; + expect(plan["Node Type"]).toBe("Limit"); + expect(plan["Actual Rows"]).toBe(101); + const nodes: any[] = [], collect = (node: any) => { nodes.push(node); for (const child of node.Plans ?? []) collect(child); }; + collect(plan); + const recursive = nodes.filter(node => node["Node Type"] === "Recursive Union"); + const relations = nodes.filter(node => node["Relation Name"] === "issue_relations"); + expect(recursive).toHaveLength(1); + expect(recursive[0]["Actual Rows"]).toBe(101); + expect(relations.length).toBeGreaterThan(0); + expect(relations.every(node => node["Actual Rows"] * node["Actual Loops"] <= 101)).toBe(true); + process.stdout.write(JSON.stringify({ reverseFanInPlan: plan }) + "\n"); + }); + + it("retains far ancestor deadlines when a cyclic chain exceeds the precise hint budget", async () => { + const service = attentionRetentionService(db), load = vi.spyOn(service, "load"); + await start(service); await settle(); load.mockClear(); + await db.execute(sql`insert into issues (id, company_id, title, status, updated_at, assignee_user_id) + select gen_random_uuid(), ${companyId}::uuid, 'chain-' || n, + case when n = 150 then 'in_review' else 'todo' end, + case when n = 150 then ${new Date(Date.now() - delay + 15_000).toISOString()}::timestamptz else now() end, + case when n = 150 then 'board' else null end + from generate_series(0, 150) n`); + await db.execute(sql`update issues child set parent_id = parent.id from issues parent + where child.company_id = ${companyId}::uuid and parent.company_id = child.company_id + and parent.title = 'chain-' || ((substring(child.title from 7)::integer + 1) % 151)`); + const [leaf, far] = await Promise.all([db.select().from(issues).where(and(eq(issues.companyId, companyId), eq(issues.title, "chain-0"))).then(rows => rows[0]), + db.select().from(issues).where(and(eq(issues.companyId, companyId), eq(issues.title, "chain-150"))).then(rows => rows[0])]); + await issueService(db).update(leaf.id, { title: "Ordinary deep chain edit" }); + await settle(); + expect(load.mock.calls.some(([scope]) => scope?.companyId === companyId && !scope.issueId && !scope.issueIds)).toBe(true); + expect(await state("review", far.id)).toBeNull(); + expect(coordinator!.nextWakeAt()).not.toBeNull(); + await archived("review", far.id); + await quiet(); + }); + + it("canonicalizes a complete cyclic affected set into one bounded batch hint", async () => { + const a = await insertIssue(), b = await insertIssue(), c = await insertIssue(); + await db.update(issues).set({ parentId: b.id }).where(eq(issues.id, a.id)); + await db.update(issues).set({ parentId: c.id }).where(eq(issues.id, b.id)); + await db.update(issues).set({ parentId: a.id }).where(eq(issues.id, c.id)); + const hints: string[] = []; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.attentionRetention, (event, values) => { + if (event === "settled") hints.push(...values ?? []); + }); + try { + await db.transaction(tx => notifyIssueAttentionRetentionWork(tx, { companyId, issueId: a.id })); + await db.transaction(tx => notifyIssueAttentionRetentionWork(tx, { companyId, issueId: c.id })); + const wide = { companyId, issueIds: [b.id, a.id, c.id, a.id], ignored: {} as unknown }; + wide.ignored = wide; + await db.transaction(tx => notifyAttentionRetentionWork(tx, wide)); + } finally { unsubscribe(); } + expect(hints).toHaveLength(3); + expect(new Set(hints)).toEqual(new Set([JSON.stringify({ companyId, issueIds: [a.id, b.id, c.id].sort() })])); + }); + it("canonicalizes wide circular notification variables to the supported identity fields", async () => { const sourceId = randomUUID(), issueId = randomUUID(), hints: string[] = []; const wide = { companyId, sourceKind: "review" as const, sourceId, issueId, ignored: {} as unknown }; @@ -487,14 +588,14 @@ const delay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000; const terminal = await insertIssue("backlog", { parentId: oldParent.id, hiddenAt: old(), assigneeAgentId: agentId, responsibleUserId: "board" }); const service = attentionRetentionService(db), load = vi.spyOn(service, "load"); await start(service); await settle(); load.mockClear(); await issueService(db).update(terminal.id, { parentId: newParent.id }); await settle(); - const ids = load.mock.calls.map(([scope]) => scope?.issueId); + const ids = load.mock.calls.flatMap(([scope]) => scope?.issueIds ?? (scope?.issueId ? [scope.issueId] : [])); expect(ids).toEqual(expect.arrayContaining([oldParent.id, newParent.id, terminal.id])); load.mockClear(); await issueService(db).update(terminal.id, { hiddenAt: null }); await settle(); expect(await state("blocker_attention", terminal.id)).toBeNull(); const visible = await service.load({ companyId, issueId: newParent.id }); expect(visible.rows!.some(entry => entry.sourceId === terminal.id && entry.dueAt > Date.now())).toBe(true); - expect(load.mock.calls.some(([scope]) => scope?.issueId === newParent.id)).toBe(true); + expect(load.mock.calls.some(([scope]) => scope?.issueId === newParent.id || scope?.issueIds?.includes(newParent.id))).toBe(true); const next = await insertIssue("backlog", { assigneeAgentId: agentId }); await issueService(db).update(oldParent.id, { blockedByIssueIds: [next.id] }); await archived("blocker_attention", next.id); }); diff --git a/server/src/services/attention-retention.ts b/server/src/services/attention-retention.ts index ebe91416f52..a0bca4281f5 100644 --- a/server/src/services/attention-retention.ts +++ b/server/src/services/attention-retention.ts @@ -1,7 +1,6 @@ import { and, eq, getTableColumns, inArray, isNull, sql, type SQL } from "drizzle-orm"; import { agents, decisionRetention, decisions, heartbeatRuns, issueApprovals, issueRecoveryActions, issueThreadInteractions, issues, approvals, type Db } from "@paperclipai/db"; -import type { AttentionSourceKind } from "@paperclipai/shared"; import { evaluateAgentInvokability } from "./agent-invokability.js"; import { HUMAN_RECOVERY_OWNER_TYPES, ATTENTION_BUDGET_WARNING_PERCENT, ATTENTION_AGENT_ERROR_STATUS, collapsePendingConfirmationsToNewest, listFailedRunsWithNewerRun, readRunIssueId, requiresReviewAttention, terminalBlockerRequiresAttention } from "./attention-source-eligibility.js"; import { decisionRetentionService, DEFAULT_DECISION_ARCHIVE_DAYS } from "./decision-retention.js"; @@ -13,12 +12,13 @@ import { issueService } from "./issues.js"; import { parseIssueExecutionState } from "./issue-execution-policy.js"; import { isProspectiveBlockedTransition } from "./routable-blocked.js"; import { registerScopedDeadlineWork } from "./scoped-deadline-work.js"; +import type { AttentionRetentionScope } from "./retention-work-notifications.js"; const archiveDelay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000; const kinds = ["decision", "issue_thread_interaction", "recovery_action", "blocker_attention", "review", "failed_run", "budget_alert", "agent_error_alert"] as const; type Kind = typeof kinds[number]; -type Scope = { companyId?: string; sourceKind?: AttentionSourceKind; sourceId?: string; issueId?: string }; +type Scope = AttentionRetentionScope; type Candidate = { sourceKind: Kind; sourceId: string; companyId: string; activityAt: string; deadlineAt: number; issueId: string | null }; type Entry = Candidate & { id: string; dueAt: number; rootId: string; variant?: "owner" | "terminal" }; type Cursor = { sourceKind: string; sourceId: string }; @@ -66,8 +66,9 @@ const sources: Record = { export function attentionRetentionService(db: Db) { const cursors = new Map(); async function candidates(database: Db, scope: Scope = {}, cursor?: Cursor, sourceId?: string): Promise { + const issueIds = scope.issueIds ?? (scope.issueId ? [scope.issueId] : []); const selected = kinds.filter(kind => (!scope.sourceKind || scope.sourceKind === kind) - && (!scope.issueId || !["budget_alert", "agent_error_alert"].includes(kind))); + && (!issueIds.length || !["budget_alert", "agent_error_alert"].includes(kind))); if (!selected.length) return []; const rows = await database.execute(sql`with source_candidates(source_kind, source_id, company_id, activity_at, issue_id) as (${sql.join(selected.map(kind => sources[kind]), sql` union all `)}) select s.source_kind as "sourceKind", s.source_id as "sourceId", s.company_id as "companyId", @@ -75,7 +76,7 @@ export function attentionRetentionService(db: Db) { from source_candidates s inner join companies c on c.id = s.company_id left join decision_retention r on r.company_id = s.company_id and r.source_kind = s.source_kind and r.source_id = s.source_id where c.status = 'active' - and (${scope.issueId ?? null}::text is null or s.issue_id = ${scope.issueId ?? null}) + ${issueIds.length ? sql`and s.issue_id in (${sql.join(issueIds.map(id => sql`${id}`), sql`, `)})` : sql``} and (${scope.companyId ?? null}::uuid is null or s.company_id = ${scope.companyId ?? null}::uuid) and (${sourceId ?? scope.sourceId ?? null}::text is null or s.source_id = ${sourceId ?? scope.sourceId ?? null}) and (s.source_kind = 'blocker_attention' or ((r.id is null or r.keep = false) and r.archived_at is null)) @@ -312,7 +313,8 @@ export function registerAttentionRetentionWork(coordinator: ReturnType (!scope?.companyId || scope.companyId === entry.companyId) && (!scope?.sourceKind || scope.sourceKind === entry.sourceKind) && (!scope?.sourceId || scope.sourceId === entry.sourceId) - && (!scope?.issueId || scope.issueId === entry.issueId), + && (!scope?.issueId || scope.issueId === entry.issueId) + && (!scope?.issueIds?.length || entry.issueId !== null && scope.issueIds.includes(entry.issueId)), load: service.load, check: service.check, }, canRun); } diff --git a/server/src/services/retention-work-notifications.ts b/server/src/services/retention-work-notifications.ts index ba0379afee1..79cffd624a8 100644 --- a/server/src/services/retention-work-notifications.ts +++ b/server/src/services/retention-work-notifications.ts @@ -5,9 +5,16 @@ import type { Db } from "@paperclipai/db"; import type { AttentionSourceKind } from "@paperclipai/shared"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; -export async function notifyAttentionRetentionWork(transaction: object, scope: { companyId?: string; sourceKind?: AttentionSourceKind; issueId?: string; sourceId?: string } = {}) { +export const ATTENTION_RETENTION_MAX_ISSUE_IDS = 100; +export type AttentionRetentionScope = { companyId?: string; sourceKind?: AttentionSourceKind; + issueId?: string; issueIds?: string[]; sourceId?: string }; + +export async function notifyAttentionRetentionWork(transaction: object, scope: AttentionRetentionScope = {}) { + const issueIds = [...new Set([...(scope.issueIds ?? []), ...(scope.issueId ? [scope.issueId] : [])])].sort(); await notifyDeliveryWork(transaction, DELIVERY_QUEUES.attentionRetention, JSON.stringify({ - companyId: scope.companyId, sourceKind: scope.sourceKind, sourceId: scope.sourceId, issueId: scope.issueId, + companyId: scope.companyId, sourceKind: scope.sourceKind, sourceId: scope.sourceId, + issueId: issueIds.length === 1 ? issueIds[0] : undefined, + issueIds: issueIds.length > 1 && issueIds.length <= ATTENTION_RETENTION_MAX_ISSUE_IDS ? issueIds : undefined, })); } @@ -47,18 +54,24 @@ export async function notifyIssueAttentionRetentionWork(transaction: object, and id::text in (${sql.join([...new Set(references)].map(id => sql`${id}`), sql`, `)})`); for (const row of trusted) seeds.push(row.id as string); } + // UNION deduplicates visited UUIDs (including cycles). Each relation edge is + // unique by schema, so 101 neighbours guarantee overflow of a 100-ID scope. + // Bound that adjacency before materialization and stream at most 101 unique + // IDs to the caller; an inner UNION DISTINCT would consume the full fan-in. const affected = await database.execute(sql`with recursive affected(id) as ( select unnest(array[${sql.join([...new Set(seeds)].map(id => sql`${id}::uuid`), sql`, `)}]) union select parent.id from affected current cross join lateral ( - select related_issue_id as id from issue_relations where company_id = ${scope.companyId}::uuid - and issue_id = current.id and type = 'blocks' - union select parent_id from issues where company_id = ${scope.companyId}::uuid + (select related_issue_id as id from issue_relations where company_id = ${scope.companyId}::uuid + and issue_id = current.id and type = 'blocks' limit ${ATTENTION_RETENTION_MAX_ISSUE_IDS + 1}) + union all select parent_id from issues where company_id = ${scope.companyId}::uuid and id = current.id and parent_id is not null ) parent - ) select id::text as "issueId" from affected`); - for (const row of affected) await notifyAttentionRetentionWork(transaction, { - companyId: scope.companyId, issueId: row.issueId as string, + ) select id::text as "issueId" from affected limit ${ATTENTION_RETENTION_MAX_ISSUE_IDS + 1}`); + // The sentinel means the affected set is incomplete. Reconstruct the whole + // company through the existing paged authority instead of dropping ancestors. + await notifyAttentionRetentionWork(transaction, { companyId: scope.companyId, + issueIds: affected.length <= ATTENTION_RETENTION_MAX_ISSUE_IDS ? affected.map(row => row.issueId as string) : undefined, }); }