mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-03 01:27:09 +01:00
Merge qualified sleep review fixes and prerequisites
Co-Authored-By: Paperclip <noreply@paperclip.ing> * commit '2d273d7ebb66179fa5c7c51dcfed6fee7b1f4e16': fix(retention): wake blocker retention after no-run workspace export Bound blocker workspace history reads by candidate workspace Validate eligibility before first attention retention materialization fix(retention): exclude finalized blockers before graph bounds test: hold lifecycle verification before retry identity assertions Exercise saved legacy input through its eligibility producer fix: cancel merge confirmation acceptance under lock Bound repeated external-object eligibility reconstruction
This commit is contained in:
DottaandPaperclip committed
commit
2ef4fa60c3
19 files changed
+862
-140
No files matched your search
@@ -6,6 +6,8 @@ import { subscribeDatabaseWork, authUsers, companyMemberships, agents, approvals
|
||||
decisionRetention, decisions, costEvents, heartbeatRunEvents, heartbeatRuns, issueApprovals, issueRecoveryActions,
|
||||
issueRelations, issueThreadInteractions, issues, type Db } from "@paperclipai/db";
|
||||
import { attentionRetentionService, registerAttentionRetentionWork } from "../services/attention-retention.js";
|
||||
import { workspaceOperationService } from "../services/workspace-operations.js";
|
||||
import * as workspaceOperationLogs from "../services/workspace-operation-log-store.js";
|
||||
import { budgetService } from "../services/budgets.js";
|
||||
import { writeExecutionControl } from "../services/execution-control-notifications.js";
|
||||
import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js";
|
||||
@@ -35,15 +37,22 @@ const delay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000;
|
||||
const errors: unknown[] = [];
|
||||
const statements: string[] = [];
|
||||
const parameterCounts: number[] = [];
|
||||
let captureBlockerPlan = false;
|
||||
let blockerPlanQuery: { query: string; parameters: unknown[] } | undefined;
|
||||
beforeAll(async () => {
|
||||
if (externalUrl) await applyPendingMigrations(externalUrl);
|
||||
else temporary = await startEmbeddedPostgresTestDatabase("attention-retention-");
|
||||
db = createDb(externalUrl ?? temporary!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
|
||||
db.$client.options.onnotice = () => {};
|
||||
db.$client.options.debug = (_connection, query, parameters) => { statements.push(query); parameterCounts.push(parameters.length); };
|
||||
db.$client.options.debug = (_connection, query, parameters) => {
|
||||
statements.push(query); parameterCounts.push(parameters.length);
|
||||
if (captureBlockerPlan && !blockerPlanQuery && query.includes("latest_blocker_operation") && query.includes("issue_relations"))
|
||||
blockerPlanQuery = { query, parameters: [...parameters] };
|
||||
};
|
||||
}, 30_000);
|
||||
beforeEach(async () => {
|
||||
companyId = randomUUID(); agentId = randomUUID(); draining = false; errors.length = 0;
|
||||
captureBlockerPlan = false; blockerPlanQuery = undefined;
|
||||
await db.insert(companies).values({ id: companyId, name: "Retention", issuePrefix: `A${companyId.slice(0, 6)}` });
|
||||
await db.insert(authUsers).values({ id: "board", name: "Board", email: "board@retention.test", createdAt: new Date(), updatedAt: new Date() }).onConflictDoNothing();
|
||||
await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: "board", membershipRole: "owner", status: "active" });
|
||||
@@ -428,6 +437,169 @@ const delay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000;
|
||||
expect(await db.select().from(decisionArchiveNotificationOutbox)).toHaveLength(0);
|
||||
});
|
||||
|
||||
it.each(["running", "run", "confirmation", "completed"] as const)("filters completed blocker history before bounded %s attention and archive checks", async kind => {
|
||||
const root = await insertIssue("blocked"), workspaceId = randomUUID();
|
||||
await db.execute(sql`insert into execution_workspaces (id, company_id, name, mode, strategy_type)
|
||||
values (${workspaceId}::uuid, ${companyId}::uuid, 'Finalized shared history', 'shared', 'project_primary')`);
|
||||
await db.execute(sql`insert into workspace_operations (company_id, execution_workspace_id, phase, status, started_at)
|
||||
values (${companyId}::uuid, ${workspaceId}::uuid, 'workspace_finalize', 'succeeded', ${old().toISOString()}::timestamptz)`);
|
||||
await db.execute(sql`with history as (
|
||||
insert into issues (id, company_id, title, status, updated_at, execution_workspace_id)
|
||||
select ('00000000-0000-4000-8000-' || lpad(to_hex(n), 12, '0'))::uuid,
|
||||
${companyId}::uuid, 'completed-blocker-' || n, 'done', ${old().toISOString()}::timestamptz, ${workspaceId}::uuid
|
||||
from generate_series(1, 2100) n returning id)
|
||||
insert into issue_relations (company_id, issue_id, related_issue_id, type)
|
||||
select ${companyId}::uuid, id, ${root.id}::uuid, 'blocks' from history`);
|
||||
if (kind !== "completed") {
|
||||
const live = await insertIssue(kind === "running" ? "in_progress" : "todo", { id: "ffffffff-ffff-4fff-8fff-ffffffffffff", assigneeAgentId: agentId });
|
||||
await db.insert(issueRelations).values({ companyId, issueId: live.id, relatedIssueId: root.id, type: "blocks" });
|
||||
if (kind === "run") {
|
||||
const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, status: "running" }).returning();
|
||||
await db.update(issues).set({ executionRunId: run.id }).where(eq(issues.id, live.id));
|
||||
} else if (kind === "confirmation") await db.insert(issueThreadInteractions).values({ companyId, issueId: live.id,
|
||||
kind: "request_confirmation", status: "pending", payload: { prompt: "Still waiting" } });
|
||||
}
|
||||
statements.length = 0; parameterCounts.length = 0;
|
||||
captureBlockerPlan = kind === "completed";
|
||||
const batch = await issueService(db).listBlockerAttentionBatch(companyId, [root]);
|
||||
captureBlockerPlan = false;
|
||||
if (kind === "completed") {
|
||||
expect(blockerPlanQuery).toBeTruthy();
|
||||
const explain = await db.$client.unsafe("EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) " + blockerPlanQuery!.query,
|
||||
blockerPlanQuery!.parameters as never[]);
|
||||
type Plan = { "Relation Name"?: string; "Actual Rows": number; "Actual Loops": number; "Rows Removed by Filter"?: number; Plans?: Plan[] };
|
||||
const rootPlan = (explain[0]!["QUERY PLAN"] as Array<{ Plan: Plan }>)[0]!.Plan;
|
||||
const visit = (node: Plan): void => {
|
||||
if (node["Relation Name"] === "issues" && node["Actual Loops"] > 1) {
|
||||
expect(node["Actual Rows"]).toBeLessThanOrEqual(1);
|
||||
expect(node["Rows Removed by Filter"] ?? 0).toBe(0);
|
||||
}
|
||||
node.Plans?.forEach(visit);
|
||||
};
|
||||
visit(rootPlan);
|
||||
process.stdout.write(JSON.stringify({ completedBlockerHistoryPlan: explain }) + "\n");
|
||||
}
|
||||
expect(batch.nodeCount).toBeLessThanOrEqual(BLOCKER_ATTENTION_MAX_NODES);
|
||||
expect(batch.edgeCount).toBeLessThanOrEqual(BLOCKER_ATTENTION_MAX_NODES * 2);
|
||||
const service = attentionRetentionService(db), page = await service.load({ companyId, sourceKind: "blocker_attention" });
|
||||
const entry = page.rows!.find(row => row.rootId === root.id && row.variant === "terminal");
|
||||
if (kind !== "completed") {
|
||||
// Exercise the authoritative fresh check even if discovery correctly omits the live root.
|
||||
await service.check(entry ?? { id: `blocker_attention:${root.id}:terminal`, companyId, sourceKind: "blocker_attention",
|
||||
sourceId: root.id, rootId: root.id, issueId: root.id, variant: "terminal", activityAt: old().toISOString(), deadlineAt: Date.now(), dueAt: Date.now() });
|
||||
expect(await state("blocker_attention", root.id)).toBeNull();
|
||||
expect(await db.select().from(decisionArchiveNotificationOutbox)).toHaveLength(0);
|
||||
} else {
|
||||
expect(entry).toBeTruthy(); await service.check(entry!);
|
||||
expect((await state("blocker_attention", root.id))?.archivedAt).toBeInstanceOf(Date);
|
||||
expect(await db.select().from(decisionArchiveNotificationOutbox)).toHaveLength(1);
|
||||
}
|
||||
expect(batch.fetchedRows).toBeLessThanOrEqual(4);
|
||||
expect(batch.deferredRootIds.size).toBe(0);
|
||||
expect(Boolean(batch.attention.get(root.id)?.blockingTreeLive)).toBe(kind !== "completed");
|
||||
expect(parameterCounts.length).toBeLessThanOrEqual(100);
|
||||
});
|
||||
|
||||
it("does not archive an unknown truncated blocker tree during its fresh check", async () => {
|
||||
const root = await insertIssue("blocked");
|
||||
await db.execute(sql`with blockers as (
|
||||
insert into issues (company_id, title, status, updated_at)
|
||||
select ${companyId}::uuid, 'unresolved-' || n, 'backlog', ${old().toISOString()}::timestamptz
|
||||
from generate_series(1, 2100) n returning id)
|
||||
insert into issue_relations (company_id, issue_id, related_issue_id, type)
|
||||
select ${companyId}::uuid, id, ${root.id}::uuid, 'blocks' from blockers`);
|
||||
const service = attentionRetentionService(db), page = await service.load({ companyId, sourceKind: "blocker_attention" });
|
||||
const entry = page.rows!.find(row => row.rootId === root.id && row.variant === "terminal")!;
|
||||
expect(entry).toBeTruthy(); expect(await service.check(entry)).toBeNull();
|
||||
expect(await db.select().from(decisionRetention)).toHaveLength(0);
|
||||
expect(await db.select().from(decisionArchiveNotificationOutbox)).toHaveLength(0);
|
||||
await start(); await quiet();
|
||||
});
|
||||
|
||||
function simulateRecorderLogStorage() {
|
||||
// The recorder lifecycle and SQL are real; only its log-storage boundary is simulated.
|
||||
vi.spyOn(workspaceOperationLogs, "getWorkspaceOperationLogStore").mockReturnValue({
|
||||
begin: async () => ({ store: "local_file", logRef: "retention-finalize-proof" }),
|
||||
append: async () => {}, finalize: async () => ({ bytes: 0, compressed: false }), read: async () => ({ content: "" }),
|
||||
});
|
||||
}
|
||||
|
||||
it.each(["issue", "workspace"] as const)("publishes retention intent from ordinary no-run %s finalization", async scope => {
|
||||
simulateRecorderLogStorage();
|
||||
const source = await insertIssue(), workspaceId = randomUUID();
|
||||
await db.execute(sql`insert into execution_workspaces (id, company_id, name, mode, strategy_type)
|
||||
values (${workspaceId}::uuid, ${companyId}::uuid, 'Finalize producer', 'shared', 'project_primary')`);
|
||||
await start(); await quiet();
|
||||
const committedHints: string[] = [];
|
||||
const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.attentionRetention, (event, hints) => {
|
||||
if (event === "settled") committedHints.push(...hints ?? []);
|
||||
});
|
||||
try {
|
||||
const result = await workspaceOperationService(db).createRecorder({ companyId,
|
||||
executionWorkspaceId: scope === "workspace" ? workspaceId : undefined,
|
||||
issueId: scope === "issue" ? source.id : undefined,
|
||||
}).recordOperation({ phase: "workspace_finalize", run: async () => ({ status: "succeeded" }) });
|
||||
expect(result.status).toBe("succeeded");
|
||||
expect(committedHints.length).toBeGreaterThan(0);
|
||||
expect(committedHints.map(hint => JSON.parse(hint).companyId).every(id => id === companyId)).toBe(true);
|
||||
await quiet();
|
||||
if (scope === "workspace") {
|
||||
committedHints.length = 0;
|
||||
await workspaceOperationService(db).createRecorder({ companyId })
|
||||
.recordOperation({ phase: "workspace_finalize", run: async () => ({ status: "succeeded" }) });
|
||||
expect(committedHints).toEqual([]); await quiet();
|
||||
}
|
||||
} finally { unsubscribe(); }
|
||||
});
|
||||
|
||||
describe("no-run finalization with an incomplete blocker graph", () => {
|
||||
let rootIds: string[], workspaceId: string, subjectId: string;
|
||||
let loadedScopes: Array<Parameters<ReturnType<typeof attentionRetentionService>["load"]>[0]>;
|
||||
beforeEach(async () => {
|
||||
const began = performance.now();
|
||||
simulateRecorderLogStorage();
|
||||
const roots = await Promise.all([insertIssue("blocked"), insertIssue("blocked")]);
|
||||
rootIds = roots.map(root => root.id); workspaceId = randomUUID();
|
||||
await db.execute(sql`insert into execution_workspaces (id, company_id, name, mode, strategy_type)
|
||||
values (${workspaceId}::uuid, ${companyId}::uuid, 'Pending shared export', 'shared', 'project_primary')`);
|
||||
await workspaceOperationService(db).createRecorder({ companyId, executionWorkspaceId: workspaceId })
|
||||
.recordOperation({ phase: "workspace_provision", run: async () => ({ status: "succeeded" }) });
|
||||
await db.execute(sql`with history as (
|
||||
insert into issues (company_id, title, status, updated_at, execution_workspace_id)
|
||||
select ${companyId}::uuid, 'waiting-for-export-' || n, 'done', ${old().toISOString()}::timestamptz, ${workspaceId}::uuid
|
||||
from generate_series(1, 2100) n returning id)
|
||||
insert into issue_relations (company_id, issue_id, related_issue_id, type)
|
||||
select ${companyId}::uuid, history.id, root.id, 'blocks' from history
|
||||
cross join (values (${rootIds[0]}::uuid), (${rootIds[1]}::uuid)) root(id)`);
|
||||
const [subject] = await db.select({ id: issues.id }).from(issues)
|
||||
.where(and(eq(issues.companyId, companyId), eq(issues.status, "done"))).limit(1);
|
||||
subjectId = subject!.id;
|
||||
const review = await insertIssue("in_review", { assigneeUserId: "board" });
|
||||
await decisionRetentionService(db).setKeep({ ...mutation("review", review.id), keep: true });
|
||||
const service = attentionRetentionService(db), load = service.load.bind(service); loadedScopes = [];
|
||||
vi.spyOn(service, "load").mockImplementation((scope, signal) => { loadedScopes.push(scope); return load(scope, signal); });
|
||||
process.stdout.write(JSON.stringify({ incompleteFinalizePhase: "seeded", setupMs: performance.now() - began }) + "\n");
|
||||
await start(service); await quiet();
|
||||
for (const rootId of rootIds) expect(await state("blocker_attention", rootId)).toBeNull();
|
||||
process.stdout.write(JSON.stringify({ incompleteFinalizeFixtureRows: 2100, incompleteRoots: 2, setupMs: performance.now() - began }) + "\n");
|
||||
}, 30_000);
|
||||
it("archives only after the ordinary workspace-only finalize commits, then settles quiet", async () => {
|
||||
const finish = (database: Db) => workspaceOperationService(database).createRecorder({ companyId, executionWorkspaceId: workspaceId, issueId: subjectId })
|
||||
.recordOperation({ phase: "workspace_finalize", run: async () => ({ status: "succeeded" }) });
|
||||
await expect(db.transaction(async tx => {
|
||||
await finish(tx as unknown as Db); throw new Error("rollback finalization");
|
||||
})).rejects.toThrow("rollback finalization");
|
||||
await settle(); for (const rootId of rootIds) expect(await state("blocker_attention", rootId)).toBeNull();
|
||||
loadedScopes.length = 0;
|
||||
expect((await finish(db)).status).toBe("succeeded");
|
||||
for (const rootId of rootIds) await archived("blocker_attention", rootId);
|
||||
await quiet();
|
||||
expect(loadedScopes.length).toBeGreaterThan(0);
|
||||
expect(loadedScopes.every(scope => scope?.companyId === companyId && scope.sourceKind === "blocker_attention")).toBe(true);
|
||||
expect(await db.select().from(decisionArchiveNotificationOutbox)).toHaveLength(2);
|
||||
});
|
||||
});
|
||||
|
||||
it("disarms finite deferred-root checks when large retained graphs have no archive obligation", async () => {
|
||||
await db.execute(sql`insert into issues (id, company_id, title, status, updated_at)
|
||||
select gen_random_uuid(), ${companyId}::uuid, 'held-root-' || n, 'blocked', ${old().toISOString()}::timestamptz
|
||||
|
||||
@@ -0,0 +1,103 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { sql } from "drizzle-orm";
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
|
||||
import { applyPendingMigrations, companies, createDb, issues } from "@paperclipai/db";
|
||||
import { issueService } from "../services/issues.js";
|
||||
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
|
||||
const external = resolveExternalTestDatabaseUrl("PAPERCLIP_BLOCKER_HISTORY_TEST_DATABASE_URL");
|
||||
const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport();
|
||||
type Plan = { "Node Type": string; "Relation Name"?: string; "Actual Rows": number; "Actual Loops": number;
|
||||
"Rows Removed by Filter"?: number; "Rows Removed by Index Recheck"?: number; Plans?: Plan[] };
|
||||
|
||||
describe.skipIf(!support.supported)("blocker workspace retained history cost", () => {
|
||||
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
let capture = false;
|
||||
const queries: Array<{ query: string; parameters: unknown[] }> = [];
|
||||
beforeAll(async () => {
|
||||
if (external) await applyPendingMigrations(external);
|
||||
else temporary = await startEmbeddedPostgresTestDatabase("blocker-history-");
|
||||
db = createDb(external ?? temporary!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
|
||||
db.$client.options.onnotice = () => {};
|
||||
db.$client.options.debug = (_connection, query, parameters) => {
|
||||
if (capture && query.includes("workspace_operations")) queries.push({ query, parameters: [...parameters] });
|
||||
};
|
||||
}, 30_000);
|
||||
afterEach(async () => { capture = false; queries.length = 0; await db.execute(sql`truncate companies cascade`); });
|
||||
afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); });
|
||||
|
||||
it.each(["finalized", "pending"] as const)("bounds operation-history work for 2100 blockers sharing 5000 %s operations", async kind => {
|
||||
const companyId = randomUUID(), workspaceId = randomUUID(), unrelatedWorkspaceId = randomUUID();
|
||||
await db.insert(companies).values({ id: companyId, name: "Shared history", issuePrefix: `SH${companyId.slice(0, 6)}` });
|
||||
const [root] = await db.insert(issues).values({ companyId, title: "History dependent", status: "blocked" }).returning();
|
||||
await db.execute(sql`insert into execution_workspaces (id, company_id, name, mode, strategy_type)
|
||||
values (${workspaceId}::uuid, ${companyId}::uuid, 'Shared operation history', 'shared', 'project_primary'),
|
||||
(${unrelatedWorkspaceId}::uuid, ${companyId}::uuid, 'Unrelated history', 'shared', 'project_primary')`);
|
||||
await db.execute(sql`insert into workspace_operations (company_id, execution_workspace_id, phase, status, started_at)
|
||||
select ${companyId}::uuid, ${unrelatedWorkspaceId}::uuid, 'worktree_prepare', 'succeeded',
|
||||
'2025-01-01T00:00:00Z'::timestamptz + n * interval '1 second' from generate_series(1, 50000) n`);
|
||||
await db.execute(sql`insert into workspace_operations (company_id, execution_workspace_id, phase, status, started_at)
|
||||
select ${companyId}::uuid, ${workspaceId}::uuid,
|
||||
case when n = 5000 and ${kind} = 'finalized' then 'workspace_finalize' else 'worktree_prepare' end,
|
||||
'succeeded', '2026-01-01T00:00:00Z'::timestamptz + n * interval '1 second' from generate_series(1, 5000) n`);
|
||||
await db.execute(sql`with history as (
|
||||
insert into issues (company_id, title, status, execution_workspace_id)
|
||||
select ${companyId}::uuid, 'done blocker ' || n, 'done', ${workspaceId}::uuid
|
||||
from generate_series(1, 2100) n returning id)
|
||||
insert into issue_relations (company_id, issue_id, related_issue_id, type)
|
||||
select ${companyId}::uuid, id, ${root.id}::uuid, 'blocks' from history`);
|
||||
await db.execute(sql`analyze workspace_operations`);
|
||||
await db.execute(sql`analyze issues`);
|
||||
await db.execute(sql`analyze issue_relations`);
|
||||
capture = true;
|
||||
const batch = await issueService(db).listBlockerAttentionBatch(companyId, [root]);
|
||||
capture = false;
|
||||
expect(queries.length).toBeGreaterThan(0);
|
||||
expect(Math.max(...queries.map(statement => statement.parameters.length))).toBeLessThanOrEqual(1100);
|
||||
let operationRowsExamined = 0;
|
||||
const measurements: unknown[] = [];
|
||||
for (const statement of queries) {
|
||||
const result = await db.$client.unsafe("EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) " + statement.query,
|
||||
statement.parameters as never[]);
|
||||
const explain = (result[0]!["QUERY PLAN"] as Array<{ Plan: Plan }>)[0]!;
|
||||
const visit = (node: Plan) => {
|
||||
if (node["Relation Name"] === "workspace_operations") operationRowsExamined += node["Actual Loops"]
|
||||
* (node["Actual Rows"] + (node["Rows Removed by Filter"] ?? 0) + (node["Rows Removed by Index Recheck"] ?? 0));
|
||||
node.Plans?.forEach(visit);
|
||||
};
|
||||
visit(explain.Plan);
|
||||
measurements.push({ ...statement, explain });
|
||||
}
|
||||
process.stdout.write(JSON.stringify({ sharedWorkspaceHistory: { kind, blockers: 2100, retainedOperations: 5000, unrelatedOperations: 50000,
|
||||
operationRowsExamined, queryCount: queries.length, measurements } }) + "\n");
|
||||
// The finite graph/pair pages may each read the 5,000 candidate rows once.
|
||||
// Neither 2,100 blockers nor 50,000 unrelated operations may multiply that work.
|
||||
expect(operationRowsExamined).toBeLessThanOrEqual(55_000);
|
||||
expect(batch.deferredRootIds.has(root.id)).toBe(kind === "pending");
|
||||
if (kind === "finalized") expect(batch.nodeCount).toBe(1);
|
||||
}, 120_000);
|
||||
it("preserves attributed precedence and the strict later-finalize boundary", async () => {
|
||||
const companyId = randomUUID(), workspaceId = randomUUID();
|
||||
await db.insert(companies).values({ id: companyId, name: "Finalize policy", issuePrefix: `FP${companyId.slice(0, 6)}` });
|
||||
await db.execute(sql`insert into execution_workspaces (id, company_id, name, mode, strategy_type)
|
||||
values (${workspaceId}::uuid, ${companyId}::uuid, 'Policy workspace', 'shared', 'project_primary')`);
|
||||
const roots = await db.insert(issues).values(["older", "equal", "fallback"].map(title => ({ companyId, title, status: "blocked" }))).returning();
|
||||
const blockers = await db.insert(issues).values(roots.map(root => ({ companyId, title: root.title + " blocker", status: "done", executionWorkspaceId: workspaceId }))).returning();
|
||||
for (let index = 0; index < roots.length; index += 1) {
|
||||
await db.execute(sql`insert into issue_relations (company_id, issue_id, related_issue_id, type)
|
||||
values (${companyId}::uuid, ${blockers[index]!.id}::uuid, ${roots[index]!.id}::uuid, 'blocks')`);
|
||||
}
|
||||
await db.execute(sql`insert into workspace_operations (company_id, execution_workspace_id, issue_id, phase, status, started_at) values
|
||||
(${companyId}::uuid, ${workspaceId}::uuid, ${blockers[0]!.id}::uuid, 'worktree_prepare', 'succeeded', '2026-01-01T00:00:01Z'),
|
||||
(${companyId}::uuid, ${workspaceId}::uuid, ${blockers[1]!.id}::uuid, 'worktree_prepare', 'succeeded', '2026-01-01T00:00:02Z'),
|
||||
(${companyId}::uuid, ${workspaceId}::uuid, null, 'workspace_finalize', 'succeeded', '2026-01-01T00:00:02Z'),
|
||||
(${companyId}::uuid, ${workspaceId}::uuid, null, 'worktree_prepare', 'succeeded', '2026-01-01T00:00:03Z')`);
|
||||
const service = issueService(db);
|
||||
expect((await service.listBlockerAttentionBatch(companyId, [roots[0]!])).nodeCount).toBe(1);
|
||||
expect((await service.listBlockerAttentionBatch(companyId, [roots[1]!])).nodeCount).toBe(2);
|
||||
expect((await service.listBlockerAttentionBatch(companyId, [roots[2]!])).nodeCount).toBe(2);
|
||||
});
|
||||
|
||||
});
|
||||
@@ -0,0 +1,134 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { and, eq, inArray, sql } from "drizzle-orm";
|
||||
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it } from "vitest";
|
||||
import { agents, authUsers, companies, companyMemberships, createDb, decisionRetention, issueRelations, issues } from "@paperclipai/db";
|
||||
import type { AttentionSourceKind } from "@paperclipai/shared";
|
||||
import { attentionService } from "../services/attention.js";
|
||||
import { decisionRetentionService } from "../services/decision-retention.js";
|
||||
import { issueService } from "../services/issues.js";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
|
||||
|
||||
const external = resolveExternalTestDatabaseUrl("PAPERCLIP_RETENTION_MATERIALIZATION_TEST_DATABASE_URL");
|
||||
const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport();
|
||||
(support.supported ? describe : describe.skip)("retention first-materialization security", () => {
|
||||
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
let companyId: string, agentId: string, userId: string;
|
||||
const users: string[] = [];
|
||||
beforeAll(async () => {
|
||||
if (!external) temporary = await startEmbeddedPostgresTestDatabase("retention-security-");
|
||||
db = createDb(external ?? temporary!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
|
||||
db.$client.options.onnotice = () => {};
|
||||
}, 30_000);
|
||||
beforeEach(async () => {
|
||||
companyId = randomUUID(); agentId = randomUUID(); userId = randomUUID(); users.push(userId);
|
||||
await db.insert(authUsers).values({ id: userId, name: "Security operator", email: `${userId}@retention.test`, createdAt: new Date(), updatedAt: new Date() });
|
||||
await db.insert(companies).values({ id: companyId, name: "Materialization", issuePrefix: `S${companyId.slice(0, 6)}`, defaultResponsibleUserId: userId });
|
||||
await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: userId, membershipRole: "owner", status: "active" });
|
||||
await db.insert(agents).values({ id: agentId, companyId, name: "Source agent", status: "idle", adapterType: "process" });
|
||||
});
|
||||
afterEach(async () => { await db.execute(sql`truncate companies cascade`); });
|
||||
afterAll(async () => { if (db) { await db.delete(authUsers).where(inArray(authUsers.id, users)); await db.$client.end(); } await temporary?.cleanup(); });
|
||||
const mutation = (sourceKind: AttentionSourceKind, sourceId: string) => ({ companyId, sourceKind, sourceId,
|
||||
authActor: { type: "board" as const, source: "local_implicit" as const, userId, companyIds: [companyId], isInstanceAdmin: true },
|
||||
actor: { actorType: "user" as const, actorId: userId, userId, agentId: null, runId: null, agentApiKeyId: null, responsibleUserId: userId } });
|
||||
async function issue(values: Partial<typeof issues.$inferInsert> = {}) {
|
||||
return (await db.insert(issues).values({ companyId, title: "Security source", status: "todo", createdByAgentId: agentId, ...values }).returning())[0]!;
|
||||
}
|
||||
async function visible(sourceKind: AttentionSourceKind, sourceId: string) {
|
||||
const feed = await attentionService(db).list(companyId, { all: true, allowUnscopedAll: true, userId, excludeSources: ["approval", "join_request"] });
|
||||
return feed.items.some(item => item.sourceKind === sourceKind && item.subject.id === sourceId);
|
||||
}
|
||||
|
||||
it.each(["review", "blocker_attention"] as const)("cannot pre-archive a future %s before an ordinary transition makes it visible", async kind => {
|
||||
const row = await issue({ assigneeUserId: userId });
|
||||
let rejected: unknown;
|
||||
try { await decisionRetentionService(db).archive(mutation(kind, row.id)); } catch (error) { rejected = error; }
|
||||
await issueService(db).update(row.id, kind === "review"
|
||||
? { status: "in_review", actorUserId: userId }
|
||||
: { status: "blocked", unblockDescriptor: { owner: "board", action: "Choose the next owner" }, actorUserId: userId });
|
||||
expect(await visible(kind, row.id)).toBe(true);
|
||||
expect(rejected).toMatchObject({ status: 404 });
|
||||
expect((await decisionRetentionService(db).getState(companyId, kind, row.id))?.archivedAt).toBeNull();
|
||||
});
|
||||
|
||||
it.each(["review", "blocker_attention", "productivity_review"] as const)("does not materialize Keep for an ineligible %s", async kind => {
|
||||
const row = await issue();
|
||||
await expect(decisionRetentionService(db).setKeep({ ...mutation(kind, row.id), keep: true })).rejects.toMatchObject({ status: 404 });
|
||||
expect(await decisionRetentionService(db).getState(companyId, kind, row.id)).toBeNull();
|
||||
});
|
||||
|
||||
it("allows Keep on an unseen eligible review and preserves existing manual actions after it leaves review", async () => {
|
||||
const row = await issue({ status: "in_review", assigneeUserId: userId });
|
||||
const service = decisionRetentionService(db), input = mutation("review", row.id);
|
||||
expect(await service.getState(companyId, "review", row.id)).toBeNull();
|
||||
expect((await service.setKeep({ ...input, keep: true })).keep).toBe(true);
|
||||
await issueService(db).update(row.id, { status: "todo", actorUserId: userId });
|
||||
expect((await service.archive(input)).archivedReason).toBe("manual");
|
||||
expect((await service.setKeep({ ...input, keep: false })).keep).toBe(false);
|
||||
expect((await service.revive(input)).archivedAt).toBeNull();
|
||||
});
|
||||
|
||||
it("allows unseen current board-owned blocker attention", async () => {
|
||||
const row = await issue({ status: "blocked", unblockDescriptor: { owner: "board", action: "Choose next owner" }, blockedTransitionAt: new Date() });
|
||||
expect((await decisionRetentionService(db).setKeep({ ...mutation("blocker_attention", row.id), keep: true })).keep).toBe(true);
|
||||
});
|
||||
|
||||
it("requires current agent error status for a new alert but preserves existing row actions", async () => {
|
||||
const service = decisionRetentionService(db), input = mutation("agent_error_alert", agentId);
|
||||
await expect(service.archive(input)).rejects.toMatchObject({ status: 404 });
|
||||
expect(await service.getState(companyId, "agent_error_alert", agentId)).toBeNull();
|
||||
await db.update(agents).set({ status: "error" }).where(eq(agents.id, agentId));
|
||||
expect((await service.setKeep({ ...input, keep: true })).keep).toBe(true);
|
||||
await db.update(agents).set({ status: "idle" }).where(eq(agents.id, agentId));
|
||||
expect((await service.archive(input)).archivedAt).toBeInstanceOf(Date);
|
||||
expect((await service.revive(input)).archivedAt).toBeNull();
|
||||
});
|
||||
|
||||
it("preserves historical productivity and terminal-leaf rows without inventing a new source", async () => {
|
||||
const row = await issue(); const service = decisionRetentionService(db);
|
||||
await service.syncItems(companyId, ["productivity_review", "blocker_attention"].map(sourceKind => ({ sourceKind: sourceKind as AttentionSourceKind, subject: { id: row.id }, activityAt: row.updatedAt.toISOString() })));
|
||||
for (const kind of ["productivity_review", "blocker_attention"] as const) {
|
||||
const input = mutation(kind, row.id);
|
||||
expect((await service.archive(input)).archivedAt).toBeInstanceOf(Date);
|
||||
expect((await service.revive(input)).archivedAt).toBeNull();
|
||||
expect((await service.setKeep({ ...input, keep: true })).keep).toBe(true);
|
||||
}
|
||||
});
|
||||
|
||||
it("requires ordinary feed materialization for an unseen TODO terminal leaf, then preserves its manual controls", async () => {
|
||||
const root = await issue({ status: "blocked" });
|
||||
const leaf = await issue({ status: "todo" });
|
||||
await db.insert(issueRelations).values({ companyId, issueId: leaf.id, relatedIssueId: root.id, type: "blocks" });
|
||||
const service = decisionRetentionService(db), input = mutation("blocker_attention", leaf.id);
|
||||
await expect(service.setKeep({ ...input, keep: true })).rejects.toMatchObject({ status: 404 });
|
||||
expect(await service.getState(companyId, "blocker_attention", leaf.id)).toBeNull();
|
||||
const attention = await issueService(db).listBlockerAttention(companyId, [root]);
|
||||
expect(attention.get(root.id)).toMatchObject({ terminalBlockerIssueId: leaf.id, state: "needs_attention", blockingTreeLive: false });
|
||||
expect(await visible("blocker_attention", leaf.id)).toBe(true);
|
||||
expect((await service.archive(input)).archivedAt).toBeInstanceOf(Date);
|
||||
expect((await service.revive(input)).archivedAt).toBeNull();
|
||||
expect((await service.setKeep({ ...input, keep: true })).keep).toBe(true);
|
||||
});
|
||||
|
||||
it.each(["review", "blocker_attention"] as const)("does not pre-materialize hidden %s", async kind => {
|
||||
const row = await issue({ hiddenAt: new Date(), status: kind === "review" ? "in_review" : "blocked", assigneeUserId: userId,
|
||||
unblockDescriptor: { owner: "board", action: "Choose next owner" }, blockedTransitionAt: new Date() });
|
||||
await expect(decisionRetentionService(db).archive(mutation(kind, row.id))).rejects.toMatchObject({ status: 404 });
|
||||
expect(await decisionRetentionService(db).getState(companyId, kind, row.id)).toBeNull();
|
||||
});
|
||||
|
||||
it.each(["review", "agent_error_alert"] as const)("keeps company and agent authorization isolated for %s", async kind => {
|
||||
const foreign = randomUUID();
|
||||
await db.insert(companies).values({ id: foreign, name: "Foreign", issuePrefix: "FOREIGN" });
|
||||
const sourceId = kind === "review" ? (await issue({ companyId: foreign, status: "in_review", assigneeUserId: userId })).id
|
||||
: (await db.insert(agents).values({ companyId: foreign, name: "Foreign error", status: "error" }).returning())[0]!.id;
|
||||
const service = decisionRetentionService(db);
|
||||
await expect(service.archive(mutation(kind, sourceId))).rejects.toMatchObject({ status: 404 });
|
||||
const input = { ...mutation(kind, sourceId), companyId: foreign, authActor: { type: "agent" as const, agentId, companyId },
|
||||
actor: { actorType: "agent" as const, actorId: agentId, userId: null, agentId, runId: null, agentApiKeyId: null, responsibleUserId: null } };
|
||||
await expect(service.archive(input)).rejects.toMatchObject({ status: 404 });
|
||||
expect(await db.select().from(decisionRetention).where(and(eq(decisionRetention.companyId, foreign), eq(decisionRetention.sourceId, sourceId)))).toHaveLength(0);
|
||||
});
|
||||
});
|
||||
@@ -368,6 +368,56 @@ const active: ExternalObjectResolveResult = { ok: true, snapshot: {
|
||||
expect(provider).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("coalesces repeated company status writes without rebuilding every future object", async () => {
|
||||
clock(); const svc = service(); const own = await fixture(), foreign = await fixture();
|
||||
const future = new Date(Date.now() + 86_400_000);
|
||||
for (let offset = 0; offset < 1100; offset += 250) {
|
||||
const ids = Array.from({ length: Math.min(250, 1100 - offset) }, () => randomUUID());
|
||||
await db.insert(externalObjects).values(ids.map(id => ({ id, companyId: own.companyId,
|
||||
providerKey: "url", objectType: "link", externalId: id, nextRefreshAt: future })));
|
||||
await db.insert(externalObjectMentions).values(ids.map(objectId => ({ companyId: own.companyId,
|
||||
sourceIssueId: own.issueId, sourceKind: "description" as const, objectId })));
|
||||
}
|
||||
await svc.syncIssue(foreign.issueId);
|
||||
await db.update(externalObjects).set({ nextRefreshAt: future });
|
||||
await start(svc);
|
||||
await settle(() => coordinator!.nextWakeAt() === future.getTime());
|
||||
const reads: Array<{ query: string; params: unknown[] }> = [];
|
||||
const debug = db.$client.options.debug;
|
||||
db.$client.options.debug = (_connection, query, params) => {
|
||||
if (/^select/i.test(query) && query.includes('from "external_objects"') && query.includes('inner join "companies"'))
|
||||
reads.push({ query, params: [...params] });
|
||||
};
|
||||
try {
|
||||
const companiesSvc = companyService(db);
|
||||
await companiesSvc.update(own.companyId, { status: "paused" });
|
||||
await settle(() => reads.length === 1 && coordinator!.nextWakeAt() === future.getTime()
|
||||
&& !databaseWorkPending(db, DELIVERY_QUEUES.externalObjectRefresh));
|
||||
for (const status of ["active", "paused", "active", "paused", "active"] as const) {
|
||||
await companiesSvc.update(own.companyId, { status });
|
||||
await settle(() => !databaseWorkPending(db, DELIVERY_QUEUES.externalObjectRefresh)
|
||||
&& (coordinator!.nextWakeAt() ?? 0) > Date.now());
|
||||
expect(reads).toHaveLength(1);
|
||||
}
|
||||
const rebuildAt = coordinator!.nextWakeAt()!;
|
||||
expect(rebuildAt - Date.now()).toBeGreaterThan(20_000);
|
||||
expect(rebuildAt - Date.now()).toBeLessThanOrEqual(30_000);
|
||||
await vi.advanceTimersByTimeAsync(rebuildAt - Date.now());
|
||||
await settle(() => coordinator!.nextWakeAt() === future.getTime() && reads.length === 6);
|
||||
for (const read of reads) {
|
||||
expect(read.query).toContain('"external_objects"."company_id" =');
|
||||
expect(read.params).toContain(own.companyId);
|
||||
expect(read.params).not.toContain(foreign.companyId);
|
||||
expect(read.params.at(-1)).toBe(250);
|
||||
}
|
||||
const count = reads.length;
|
||||
await vi.advanceTimersByTimeAsync(20 * 60_000);
|
||||
expect(reads).toHaveLength(count); expect(provider).not.toHaveBeenCalled();
|
||||
expect(coordinator!.nextWakeAt()).toBe(future.getTime());
|
||||
expect(errors).not.toHaveBeenCalled();
|
||||
} finally { db.$client.options.debug = debug; }
|
||||
});
|
||||
|
||||
it("scopes repeated issue deletions to cascaded source objects and cancels their last future mention", async () => {
|
||||
clock(); const svc = service();
|
||||
const own = await fixture(), foreign = await fixture();
|
||||
|
||||
@@ -22,7 +22,7 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
|
||||
let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
|
||||
let draining = false, standby = false;
|
||||
const errors: unknown[] = [];
|
||||
const provider = vi.fn(async (_companyId: string, _reference: { number: number }): Promise<"open" | "merged" | "unknown"> => "merged");
|
||||
const provider = vi.fn(async (_companyId: string, _reference: { number: number }, _signal?: AbortSignal): Promise<"open" | "merged" | "unknown"> => "merged");
|
||||
const wakeup = vi.fn(async () => ({ id: "simulated-durable-wake" }));
|
||||
beforeAll(async () => {
|
||||
if (!external) temporary = await startEmbeddedPostgresTestDatabase("merged-pr-work-");
|
||||
@@ -248,6 +248,84 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
|
||||
expect(errors).toEqual([]);
|
||||
});
|
||||
|
||||
it("preserves manual acceptance without cancellation and rolls back a failed caller mutation", async () => {
|
||||
const card = await create();
|
||||
const issue = { id: issueId, companyId, projectId: null, goalId: null };
|
||||
await expect(service().acceptInteraction(issue, card.id, {}, { userId: "board" }, {
|
||||
afterResolveInTransaction: async () => { throw new Error("manual acceptance rollback"); },
|
||||
})).rejects.toThrow("manual acceptance rollback");
|
||||
expect((await stored(card.id)).status).toBe("pending");
|
||||
const accepted = await service().acceptInteraction(issue, card.id, {}, { userId: "board" });
|
||||
expect(accepted.interaction.status).toBe("accepted");
|
||||
expect((await stored(card.id)).status).toBe("accepted");
|
||||
});
|
||||
|
||||
it.each(["issue lock", "acceptance write"] as const)("keeps a stopped %s acceptance pending and accepts once after restart", async phase => {
|
||||
vi.useRealTimers();
|
||||
const card = await create();
|
||||
let signal: AbortSignal | undefined;
|
||||
provider.mockImplementation(async (_companyId, _reference, currentSignal) => { signal = currentSignal; return "merged"; });
|
||||
let reserved: Awaited<ReturnType<typeof db.$client.reserve>> | undefined = await db.$client.reserve();
|
||||
const [{ pid }] = await reserved<{ pid: number }[]>`select pg_backend_pid() as pid`;
|
||||
if (phase === "issue lock") {
|
||||
await reserved.unsafe("begin");
|
||||
await reserved`select id from issues where id=${issueId} for update`;
|
||||
} else {
|
||||
await db.$client.unsafe(`create function maintenance_hold_merge_acceptance() returns trigger language plpgsql as $$begin
|
||||
if new.id = '${card.id}'::uuid and new.status = 'accepted' then perform pg_advisory_xact_lock(930582); end if; return new; end$$`);
|
||||
await db.$client.unsafe(`create trigger maintenance_hold_merge_acceptance before update on issue_thread_interactions for each row execute function maintenance_hold_merge_acceptance()`);
|
||||
await reserved`select pg_advisory_lock(930582)`;
|
||||
}
|
||||
let stopping: Promise<void> | undefined;
|
||||
try {
|
||||
coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
|
||||
const worker = registerMergeConfirmationWork(coordinator, service(), () => true);
|
||||
let blocked: { query: string } | undefined;
|
||||
for (let attempt = 0; attempt < 200; attempt++) {
|
||||
[blocked] = await db.$client<{ query: string }[]>`select query from pg_stat_activity
|
||||
where datname=current_database() and wait_event_type='Lock' and ${pid} = any(pg_blocking_pids(pid))`;
|
||||
if (blocked) break;
|
||||
await delay(10);
|
||||
}
|
||||
expect(blocked?.query).toContain(phase === "issue lock" ? '"issues"' : '"issue_thread_interactions"');
|
||||
let stopped = false;
|
||||
stopping = coordinator.stop().then(() => { stopped = true; });
|
||||
await delay(20);
|
||||
expect(signal?.aborted).toBe(true);
|
||||
expect(stopped).toBe(false);
|
||||
expect((await stored(card.id)).status).toBe("pending");
|
||||
const releasedAt = performance.now();
|
||||
if (phase === "issue lock") await reserved.unsafe("commit");
|
||||
else await reserved`select pg_advisory_unlock(930582)`;
|
||||
reserved.release(); reserved = undefined;
|
||||
await stopping; await worker.ready; coordinator = undefined;
|
||||
expect(performance.now() - releasedAt).toBeLessThan(1000);
|
||||
expect((await stored(card.id)).status).toBe("pending");
|
||||
expect(wakeup).not.toHaveBeenCalled();
|
||||
expect(await db.select().from(activityLog).where(eq(activityLog.action, "issue.thread_interaction_accepted"))).toHaveLength(0);
|
||||
expect(await db.select().from(chatActions)).toHaveLength(0);
|
||||
await start();
|
||||
expect((await stored(card.id)).status).toBe("accepted");
|
||||
expect(wakeup).toHaveBeenCalledTimes(1);
|
||||
expect(await db.select().from(activityLog).where(eq(activityLog.action, "issue.thread_interaction_accepted"))).toHaveLength(1);
|
||||
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
|
||||
await quiet();
|
||||
expect(wakeup).toHaveBeenCalledTimes(1); expect(errors).toEqual([]);
|
||||
} finally {
|
||||
if (reserved) {
|
||||
if (phase === "issue lock") await reserved.unsafe("rollback");
|
||||
else await reserved`select pg_advisory_unlock(930582)`;
|
||||
reserved.release();
|
||||
}
|
||||
await stopping;
|
||||
await coordinator?.stop(); coordinator = undefined;
|
||||
if (phase === "acceptance write") {
|
||||
await db.$client.unsafe("drop trigger if exists maintenance_hold_merge_acceptance on issue_thread_interactions");
|
||||
await db.$client.unsafe("drop function if exists maintenance_hold_merge_acceptance()");
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
it("awaits an in-flight observation at shutdown and reconstructs its unaccepted card on restart", async () => {
|
||||
let release!: () => void, observed!: () => void;
|
||||
const barrier = new Promise<void>(resolve => { release = resolve; });
|
||||
|
||||
@@ -37,3 +37,41 @@ it("preserves a subsecond replacement deadline when the owner does not require p
|
||||
expect(check).toHaveBeenCalledOnce();
|
||||
expect(onError).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
|
||||
it("coalesces broad reconstruction without delaying an exact object hint or saved deadline", async () => {
|
||||
vi.useFakeTimers();
|
||||
const owner = installDatabaseWorkSignals({ execute: vi.fn(async () => []) });
|
||||
coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => true, onError: vi.fn() });
|
||||
const now = Date.now();
|
||||
let dueAt = now + 60_000;
|
||||
const load = vi.fn(async () => [{ id: "object", companyId: "company", dueAt }]);
|
||||
const check = vi.fn(async () => null);
|
||||
const worker = registerScopedDeadlineWork(coordinator, {
|
||||
queue: DELIVERY_QUEUES.externalObjectRefresh,
|
||||
minimumPassIntervalMs: 0,
|
||||
reconstructionIntervalMs: (scope: string | undefined) => scope === "company" ? 30_000 : 0,
|
||||
decodeScope: hint => hint === "*" ? undefined : hint,
|
||||
matches: () => true,
|
||||
load,
|
||||
check,
|
||||
}, () => true);
|
||||
await worker.ready;
|
||||
worker.invalidate("company"); await vi.advanceTimersByTimeAsync(0);
|
||||
expect(load).toHaveBeenCalledTimes(2);
|
||||
worker.invalidate("company"); await vi.advanceTimersByTimeAsync(0);
|
||||
expect(load).toHaveBeenCalledTimes(2);
|
||||
expect(coordinator.nextWakeAt()).toBe(now + 30_000);
|
||||
dueAt = now + 250;
|
||||
worker.invalidate("object"); await vi.advanceTimersByTimeAsync(0);
|
||||
expect(load).toHaveBeenCalledTimes(3);
|
||||
expect(coordinator.nextWakeAt()).toBe(dueAt);
|
||||
await vi.advanceTimersByTimeAsync(250);
|
||||
expect(check).toHaveBeenCalledOnce();
|
||||
load.mockResolvedValue([]);
|
||||
await vi.advanceTimersByTimeAsync(29_750);
|
||||
expect(load).toHaveBeenCalledTimes(4);
|
||||
expect(coordinator.nextWakeAt()).toBeNull();
|
||||
await vi.advanceTimersByTimeAsync(120_000);
|
||||
expect(load).toHaveBeenCalledTimes(4); expect(check).toHaveBeenCalledOnce();
|
||||
});
|
||||
Reference in new issue
Block a user