mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-08 01:57:31 +01:00
Merge commit '19ccf712e4b6f868b826952b17aea44a7b66bc86' into codex/sleep-09-aggregate
* commit '19ccf712e4b6f868b826952b17aea44a7b66bc86': fix(github): fence retry attempts and restored binding intent fix(server): fence status card refresh decisions and retry debt fix: park unchanged project budget execution waits test: remove duplicate external object database import
This commit is contained in:
commit
187ee70c3a
12 files changed
+310
-37
No files matched your search
@@ -1,4 +1,5 @@
|
||||
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
|
||||
import { DELIVERY_QUEUES, subscribeDeliveryWork } from "../services/delivery-work-notifications.js";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
@@ -264,6 +265,31 @@ describeEmbeddedPostgres("budgetService release gate enforcement", () => {
|
||||
});
|
||||
});
|
||||
|
||||
it.each([false, true])("only signals a real project budget pause with concurrent admission checks (already paused: %s)", async alreadyPaused => {
|
||||
const { companyId, agentId, projectId } = await createBudgetFixture();
|
||||
await db.insert(budgetPolicies).values({ companyId, scopeType: "project", scopeId: projectId,
|
||||
metric: "billed_cents", windowKind: "lifetime", amount: 100, hardStopEnabled: true, notifyEnabled: false });
|
||||
await insertCostEvent({ companyId, agentId, projectId, costCents: 125 });
|
||||
const pausedAt = new Date("2020-01-01T00:00:00.000Z");
|
||||
if (alreadyPaused) await db.update(projects).set({ pausedAt, pauseReason: "manual" }).where(eq(projects.id, projectId));
|
||||
const dispatch = vi.fn(), deferred = vi.fn();
|
||||
const unsubscribe = [subscribeDeliveryWork(db, DELIVERY_QUEUES.queuedRunDispatch, dispatch),
|
||||
subscribeDeliveryWork(db, DELIVERY_QUEUES.deferredExecutionReceipts, deferred)];
|
||||
try {
|
||||
const service = budgetService(db);
|
||||
const blocks = await Promise.all(Array.from({ length: 4 }, () => service.getInvocationBlock(companyId, agentId, { projectId })));
|
||||
expect(blocks.every(block => block?.scopeType === "project" && block.scopeId === projectId)).toBe(true);
|
||||
const [paused] = await db.select().from(projects).where(eq(projects.id, projectId));
|
||||
expect(paused).toMatchObject({ pauseReason: alreadyPaused ? "manual" : "budget", pausedAt: alreadyPaused ? pausedAt : expect.any(Date) });
|
||||
expect(dispatch).toHaveBeenCalledTimes(alreadyPaused ? 0 : 1);
|
||||
expect(deferred).toHaveBeenCalledTimes(alreadyPaused ? 0 : 1);
|
||||
await Promise.all(Array.from({ length: 4 }, () => service.getInvocationBlock(companyId, agentId, { projectId })));
|
||||
expect((await db.select().from(projects).where(eq(projects.id, projectId)))[0]).toEqual(paused);
|
||||
expect(dispatch).toHaveBeenCalledTimes(alreadyPaused ? 0 : 1);
|
||||
expect(deferred).toHaveBeenCalledTimes(alreadyPaused ? 0 : 1);
|
||||
} finally { unsubscribe.forEach(stop => stop()); }
|
||||
});
|
||||
|
||||
it("hard-stops project work until a valid budget raise resumes it and overview reconciles ledger spend", async () => {
|
||||
const { companyId, agentId, projectId } = await createBudgetFixture();
|
||||
const cancelWorkForScope = vi.fn().mockResolvedValue(undefined);
|
||||
|
||||
@@ -13,7 +13,7 @@ import { issueTreeControlService } from "../services/issue-tree-control.js";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { agents, agentWakeupRequests, authUsers, chatActions, chatEndpoints, toolApplications, toolConnections, companies, companyMemberships, costEvents, createDb, reconcileDatabaseWork, heartbeatRuns, issueComments, issues } from "@paperclipai/db";
|
||||
import { agents, agentWakeupRequests, authUsers, chatActions, chatEndpoints, toolApplications, toolConnections, companies, companyMemberships, costEvents, createDb, projects, reconcileDatabaseWork, heartbeatRuns, issueComments, issues } from "@paperclipai/db";
|
||||
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
|
||||
import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
|
||||
@@ -142,17 +142,22 @@ describe("deferred receipt durable dispatch", () => {
|
||||
expect(await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.contextSnapshot}->>'taskRetryRequestId' = ${receipt.requestId}`)).toHaveLength(1);
|
||||
await quiet();
|
||||
});
|
||||
async function blockedRetry(gate: "budget" | "paused agent", clocked = false) {
|
||||
async function blockedRetry(gate: "budget" | "project budget" | "paused agent", clocked = false) {
|
||||
const f = await fixture(), lifecycle = createAgentLifecycle(db), budgets = budgetService(db);
|
||||
// Host lifecycle completion is controlled; all state, admission and the
|
||||
// eventual process adapter execution use their production implementations.
|
||||
const lifecycleWorker = configureAgentLifecycle(db, { requiredPluginIds: async () => [],
|
||||
runHost: async () => "complete", runPlugin: async () => "complete" });
|
||||
lifecycleWorkers.add(lifecycleWorker);
|
||||
const policy = { scopeType: "agent" as const, scopeId: f.agent.id, amount: 100,
|
||||
const projectId = gate === "project budget" ? randomUUID() : null;
|
||||
if (projectId) {
|
||||
await db.insert(projects).values({ id: projectId, companyId: f.companyId, name: "Retry budget project" });
|
||||
await db.update(issues).set({ projectId }).where(eq(issues.id, f.issue.id));
|
||||
}
|
||||
const policy = { scopeType: projectId ? "project" as const : "agent" as const, scopeId: projectId ?? f.agent.id, amount: 100,
|
||||
metric: "billed_cents" as const, windowKind: "lifetime" as const, hardStopEnabled: true, notifyEnabled: false };
|
||||
if (gate === "budget") {
|
||||
await db.insert(costEvents).values({ companyId: f.companyId, agentId: f.agent.id,
|
||||
if (gate !== "paused agent") {
|
||||
await db.insert(costEvents).values({ companyId: f.companyId, agentId: f.agent.id, projectId,
|
||||
provider: "test", model: "receipt-proof", costCents: 200, occurredAt: new Date() });
|
||||
await budgets.upsertPolicy(f.companyId, policy, f.actorId);
|
||||
} else await lifecycle.pauseAgent(f.agent.id);
|
||||
@@ -166,18 +171,18 @@ describe("deferred receipt durable dispatch", () => {
|
||||
expect(request.accepted).toBe(true);
|
||||
const read = async () => (await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, request.requestId!)))[0];
|
||||
await vi.waitFor(async () => expect((await read()).payload?.executionWait).toMatchObject({
|
||||
reason: gate === "budget" ? "budget.blocked" : "agent.not_invokable",
|
||||
reason: gate !== "paused agent" ? "budget.blocked" : "agent.not_invokable",
|
||||
}), { timeout: 10_000 });
|
||||
const refusals: unknown[] = [];
|
||||
await heartbeat.resumeExecutionWaitComments({ companyId: f.companyId, requestId: request.requestId!, onError: error => refusals.push(error) });
|
||||
expect(refusals).toHaveLength(1);
|
||||
expect(refusals[0]).toBeInstanceOf(HttpError);
|
||||
expect(refusals[0]).toMatchObject({ status: 409,
|
||||
details: gate === "budget" ? { scopeType: "agent", scopeId: f.agent.id } : { status: "paused", reason: "paused" } });
|
||||
details: gate !== "paused agent" ? { scopeType: policy.scopeType, scopeId: policy.scopeId } : { status: "paused", reason: "paused" } });
|
||||
return { f, lifecycle, budgets, policy, lifecycleWorker, request, read };
|
||||
}
|
||||
|
||||
it.each(["budget", "paused agent"] as const)("parks real task Retry behind unchanged %s without recurring SQL", async gate => {
|
||||
it.each(["budget", "project budget", "paused agent"] as const)("parks real task Retry behind unchanged %s without recurring SQL", async gate => {
|
||||
const { request, read } = await blockedRetry(gate, true);
|
||||
// The clock was installed before worker registration and every timer.
|
||||
// No execution exists while we observe actual driver SQL across25min.
|
||||
@@ -190,11 +195,11 @@ describe("deferred receipt durable dispatch", () => {
|
||||
expect(await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.contextSnapshot}->>'taskRetryRequestId' = ${request.requestId}`)).toHaveLength(0);
|
||||
});
|
||||
|
||||
it.each(["budget", "paused agent"] as const)("releases real %s Retry into exactly one saved process execution", async gate => {
|
||||
it.each(["budget", "project budget", "paused agent"] as const)("releases real %s Retry into exactly one saved process execution", async gate => {
|
||||
const { f, lifecycle, budgets, policy, lifecycleWorker, request, read } = await blockedRetry(gate);
|
||||
// This independent warm-worker fixture stays on the real PostgreSQL clock
|
||||
// through gate release, controller authority and actual child-process exit.
|
||||
if (gate === "budget") await budgets.upsertPolicy(f.companyId, { ...policy, amount: 1000 }, f.actorId);
|
||||
if (gate !== "paused agent") await budgets.upsertPolicy(f.companyId, { ...policy, amount: 1000 }, f.actorId);
|
||||
else await lifecycle.resumeAgent(f.agent.id);
|
||||
await lifecycleWorker.process(f.agent.id);
|
||||
let runId: string | null = null;
|
||||
|
||||
@@ -1,4 +1,3 @@
|
||||
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
|
||||
@@ -6,9 +6,9 @@ import { createServer, type Server } from "node:http";
|
||||
import { once } from "node:events";
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import { eq, sql } from "drizzle-orm";
|
||||
import { applyPendingMigrations, companies, activityLog, companySecrets, connectionGrants, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, toolApplications, toolConnections } from "@paperclipai/db";
|
||||
import { applyPendingMigrations, companies, activityLog, companySecrets, connectionEventDeliveries, connectionGrants, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, toolApplications, toolConnections } from "@paperclipai/db";
|
||||
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
|
||||
import { createGitHubGrantContinuity, githubGrantContinuityDeadline } from "../services/tool-access/github-grant-continuity.js";
|
||||
import { createGitHubGrantContinuity, githubGrantContinuityDeadline, githubContinuityRevision } from "../services/tool-access/github-grant-continuity.js";
|
||||
import { createPaperclipCloudConnector } from "../services/paperclip-cloud-connector.js";
|
||||
import { githubConnectionEventService } from "../services/github-connection-events.js";
|
||||
import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
|
||||
@@ -36,6 +36,7 @@ describeDatabase("GitHub coordinator with PostgreSQL and controlled Cloud HTTP (
|
||||
const signing = generateKeyPairSync("ed25519"), sealing = generateKeyPairSync("x25519");
|
||||
let releaseRefresh: (() => void) | undefined, refreshArrived: (() => void) | undefined;
|
||||
let refreshGate: Promise<void> | undefined;
|
||||
let discoveryFailureGate: Promise<void> | undefined, discoveryArrived: (() => void) | undefined;
|
||||
let heldBindingCompany: string | undefined, bindingGate: Promise<void> | undefined;
|
||||
let bindingArrived: (() => void) | undefined, bindingClosed: (() => void) | undefined;
|
||||
const realFetch = globalThis.fetch;
|
||||
@@ -86,6 +87,10 @@ describeDatabase("GitHub coordinator with PostgreSQL and controlled Cloud HTTP (
|
||||
res.writeHead(failBindings ? 503 : 200, { "content-type": "application/json" }).end("{}"); return;
|
||||
}
|
||||
githubRequests++;
|
||||
if (req.url === "/user" && discoveryFailureGate) {
|
||||
discoveryArrived?.(); await discoveryFailureGate;
|
||||
res.writeHead(503, { "content-type": "application/json", "retry-after": "86400" }).end("{}"); return;
|
||||
}
|
||||
const body = req.url === "/user" ? { id: 42, login: "octocat" }
|
||||
: req.url!.startsWith("/user/installations?") ? { installations: [{ id: 101, repository_selection: "all", account: { login: "octocat" } }] }
|
||||
: { total_count: 1, repositories: [{ id: 999, full_name: "octocat/repo" }] };
|
||||
@@ -187,6 +192,108 @@ describeDatabase("GitHub coordinator with PostgreSQL and controlled Cloud HTTP (
|
||||
expect(await restarted.nextGitHubConnectionContinuityAt()).toBeNull();
|
||||
});
|
||||
|
||||
it.each(["credential", "installation"])("does not attach a withheld provider failure to newer %s work", async change => {
|
||||
const f = await fixture(["101"]);
|
||||
vi.spyOn(globalThis, "fetch").mockImplementation((url, init) => String(url).startsWith("https://api.github.com")
|
||||
? realFetch(`${origin}${new URL(String(url)).pathname}${new URL(String(url)).search}`, init) : realFetch(url, init));
|
||||
await db.update(connectionGrants).set({ providerTenant: { ...f.grant.providerTenant,
|
||||
oauth: { ...f.grant.providerTenant!.oauth, accessTokenExpiresAt: new Date(Date.now() + 30 * 60_000).toISOString() } } }).where(eq(connectionGrants.id, f.grant.id));
|
||||
let release!: () => void;
|
||||
const arrived = new Promise<void>(resolve => { discoveryArrived = resolve; });
|
||||
discoveryFailureGate = new Promise<void>(resolve => { release = resolve; });
|
||||
const pending = f.service.sweepGitHubConnectionContinuity(f.company.id);
|
||||
const finished = Promise.allSettled([pending]);
|
||||
try {
|
||||
await Promise.race([arrived, pending.then(() => { throw new Error("Expected held provider request"); })]);
|
||||
const revision = randomUUID();
|
||||
const expiry = new Date(Date.now() + 10 * 60_000).toISOString();
|
||||
await db.transaction(async tx => {
|
||||
await notifyGitHubCredentialChange(tx, f.connection);
|
||||
await tx.update(connectionGrants).set({ providerTenant: change === "credential"
|
||||
? sql`jsonb_set(${connectionGrants.providerTenant}, '{oauth,accessTokenExpiresAt}', ${JSON.stringify(expiry)}::jsonb)`
|
||||
: sql`jsonb_set(${connectionGrants.providerTenant}, '{github,bindingSync}', ${JSON.stringify({ id: revision, removeInstallationIds: ["202"] })}::jsonb)`,
|
||||
updatedAt: new Date() }).where(eq(connectionGrants.id, f.grant.id));
|
||||
});
|
||||
release();
|
||||
expect(await pending).toMatchObject({ due: 1, failed: 1 });
|
||||
const [saved] = await db.select().from(connectionGrants).where(eq(connectionGrants.id, f.grant.id));
|
||||
expect(saved!.providerTenant!.github!.continuityRetry).toBeUndefined();
|
||||
if (change === "credential") expect(saved!.providerTenant!.oauth!.accessTokenExpiresAt).toBe(expiry);
|
||||
else expect(saved!.providerTenant!.github!.bindingSync!.id).toBe(revision);
|
||||
expect(githubGrantContinuityDeadline(f.connection, saved!)).toBeLessThanOrEqual(Date.now());
|
||||
discoveryFailureGate = undefined;
|
||||
const restarted = createGitHubGrantContinuity(db, f.dependencies);
|
||||
expect(await restarted.sweepGitHubConnectionContinuity(f.company.id)).toMatchObject({ due: 1, refreshed: 1, failed: 0 });
|
||||
expect(bindings).toContainEqual(expect.objectContaining({ installationId: "101", active: true }));
|
||||
if (change === "installation") expect(bindings).toContainEqual(expect.objectContaining({ installationId: "202", active: false }));
|
||||
const [settled] = await db.select().from(connectionGrants).where(eq(connectionGrants.id, f.grant.id));
|
||||
expect(settled!.providerTenant!.github!.bindingSync).toBeUndefined();
|
||||
const next = await restarted.nextGitHubConnectionContinuityAt(f.company.id);
|
||||
expect(next).toBeLessThan(Date.now() + 24 * 60 * 60_000);
|
||||
} finally { release(); await finished; discoveryFailureGate = undefined; discoveryArrived = undefined; }
|
||||
});
|
||||
|
||||
it("retains provider backoff for the actual attempted revision after its own credential rotation", async () => {
|
||||
const f = await fixture(["101"]);
|
||||
await db.update(connectionGrants).set({ providerTenant: { ...f.grant.providerTenant,
|
||||
oauth: { ...f.grant.providerTenant!.oauth, accessTokenExpiresAt: new Date(Date.now() + 30 * 60_000).toISOString() } } }).where(eq(connectionGrants.id, f.grant.id));
|
||||
vi.spyOn(globalThis, "fetch").mockImplementation((url, init) => String(url).startsWith("https://api.github.com")
|
||||
? realFetch(`${origin}${new URL(String(url)).pathname}${new URL(String(url)).search}`, init) : realFetch(url, init));
|
||||
const owner = createGitHubGrantContinuity(db, { ...f.dependencies, refreshOAuthGrantCredentials: async () => {
|
||||
const [rotated] = await db.update(connectionGrants).set({ providerTenant: sql`jsonb_set(${connectionGrants.providerTenant}, '{oauth}', ${JSON.stringify({
|
||||
accessTokenExpiresAt: new Date(Date.now() + 10 * 60_000).toISOString(), refreshedAt: new Date().toISOString(),
|
||||
})}::jsonb)` }).where(eq(connectionGrants.id, f.grant.id)).returning();
|
||||
return rotated!;
|
||||
} });
|
||||
discoveryFailureGate = Promise.resolve();
|
||||
try {
|
||||
expect(await owner.sweepGitHubConnectionContinuity(f.company.id)).toMatchObject({ due: 1, failed: 1 });
|
||||
const [saved] = await db.select().from(connectionGrants).where(eq(connectionGrants.id, f.grant.id));
|
||||
expect(saved!.providerTenant!.github!.continuityRetry!.revision).toBe(githubContinuityRevision(saved!));
|
||||
expect(await owner.nextGitHubConnectionContinuityAt(f.company.id)).toBeGreaterThan(Date.now() + 23 * 60 * 60_000);
|
||||
const calls = githubRequests;
|
||||
expect(await createGitHubGrantContinuity(db, f.dependencies).sweepGitHubConnectionContinuity(f.company.id)).toMatchObject({ due: 0 });
|
||||
expect(githubRequests).toBe(calls);
|
||||
} finally { discoveryFailureGate = undefined; }
|
||||
});
|
||||
|
||||
it.each([["suspend", "unsuspend"], ["unsuspend", "suspend"]])("reconciles same-batch installation order %s then %s and duplicate replay", async (first, last) => {
|
||||
const f = await fixture(["101"]);
|
||||
await db.update(connectionGrants).set({ providerTenant: { ...f.grant.providerTenant,
|
||||
github: { ...f.grant.providerTenant!.github!, bindingSync: { id: randomUUID(), removeInstallationIds: ["202"] } } } }).where(eq(connectionGrants.id, f.grant.id));
|
||||
const events = [first, last].map(action => ({ id: randomUUID(), provider: "github" as const, event: "installation", action,
|
||||
installationId: "101", repositoryId: null, createdAt: new Date(Date.now() + 1000).toISOString(), bindingIds: [`${f.grant.id}_101`], payload: { installationId: "101" } }));
|
||||
const acknowledge = vi.fn(async () => events.length);
|
||||
const ingress = githubConnectionEventService(db, { connector: { ...f.connector, leaseEvents: async () => ({ leaseId: "installation-batch", events }), acknowledgeEvents: acknowledge } });
|
||||
expect(await ingress.pollOnce({ forDelivery: true })).toMatchObject({ processed: 2, failed: 0 });
|
||||
const [saved] = await db.select().from(connectionGrants).where(eq(connectionGrants.id, f.grant.id));
|
||||
const active = last === "unsuspend";
|
||||
expect(saved!.providerTenant!.github!.installationIds).toEqual(active ? ["101"] : []);
|
||||
expect(saved!.providerTenant!.github!.bindingSync!.removeInstallationIds.sort()).toEqual(active ? ["202"] : ["101", "202"]);
|
||||
expect(githubGrantContinuityDeadline(f.connection, saved!)).toBe(0);
|
||||
await f.service.syncGitHubGrantBindings(f.connection, saved!);
|
||||
expect(bindings.filter(binding => binding.installationId === "101")).toEqual([expect.objectContaining({ active })]);
|
||||
expect(bindings.filter(binding => binding.installationId === "202")).toEqual([expect.objectContaining({ active: false })]);
|
||||
const [settled] = await db.select().from(connectionGrants).where(eq(connectionGrants.id, f.grant.id));
|
||||
expect(settled!.providerTenant!.github!.bindingSync).toBeUndefined();
|
||||
// Batch duplicates must retain terminal receipts and emit no registration changes.
|
||||
expect(await ingress.pollOnce({ forDelivery: true })).toMatchObject({ processed: 0, duplicate: active ? 2 : 0, ignored: active ? 0 : 2, failed: 0 });
|
||||
await f.service.syncGitHubGrantBindings(f.connection, saved!);
|
||||
expect(bindings).toHaveLength(2);
|
||||
expect((await db.select().from(connectionGrants).where(eq(connectionGrants.id, f.grant.id)))[0]!.providerTenant).toEqual(settled!.providerTenant);
|
||||
expect((await db.select().from(connectionEventDeliveries).where(eq(connectionEventDeliveries.companyId, f.company.id))).map(receipt => receipt.status)).toEqual(["processed", "processed"]);
|
||||
expect(acknowledge).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("does not retire active installations from a legacy conflicting binding intent", async () => {
|
||||
const f = await fixture(["101"]);
|
||||
const [saved] = await db.update(connectionGrants).set({ providerTenant: { ...f.grant.providerTenant,
|
||||
github: { ...f.grant.providerTenant!.github!, bindingSync: { id: randomUUID(), removeInstallationIds: ["101", "202"] } } } }).where(eq(connectionGrants.id, f.grant.id)).returning();
|
||||
await f.service.syncGitHubGrantBindings(f.connection, saved!);
|
||||
expect(bindings.filter(binding => binding.installationId === "101")).toEqual([expect.objectContaining({ active: true })]);
|
||||
expect(bindings.filter(binding => binding.installationId === "202")).toEqual([expect.objectContaining({ active: false })]);
|
||||
});
|
||||
|
||||
it("preserves webhook removal debt across an actual signed, held Cloud token rotation and lease cleanup", async () => {
|
||||
const f = await fixture(["101"]);
|
||||
vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "01234567890123456789012345678901");
|
||||
@@ -327,6 +434,11 @@ describeDatabase("GitHub coordinator with PostgreSQL and controlled Cloud HTTP (
|
||||
expect(leases).toBeLessThanOrEqual(3); expect(load.mock.calls.length).toBeLessThanOrEqual(2);
|
||||
if (loadTimes.length === 2) expect(loadTimes[1]! - loadTimes[0]!).toBeGreaterThanOrEqual(5_000);
|
||||
if (leaseTimes.length === 3) expect(leaseTimes[2]! - leaseTimes[1]!).toBeGreaterThanOrEqual(5_000);
|
||||
// A hint arriving during the first SQL pass retains one spaced generation.
|
||||
// Start the empty-window assertion after that concrete work has settled.
|
||||
await vi.advanceTimersByTimeAsync(31_000); await settled(coordinator);
|
||||
expect(leases).toBeLessThanOrEqual(3); expect(load.mock.calls.length).toBeLessThanOrEqual(2);
|
||||
if (loadTimes.length === 2) expect(loadTimes[1]! - loadTimes[0]!).toBeGreaterThanOrEqual(5_000);
|
||||
const quiet = { leases, loads: load.mock.calls.length };
|
||||
await vi.advanceTimersByTimeAsync(60_000);
|
||||
expect({ leases, loads: load.mock.calls.length }).toEqual(quiet);
|
||||
|
||||
@@ -588,6 +588,88 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
|
||||
expect(wakeup).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it.each([false, true])("rechecks current policy after a committed edit during held search (changes=%s)", async changed => {
|
||||
const created = await card();
|
||||
if (changed) await issueService(db).create(companyId, { title: "Policy race blocker", status: "blocked", priority: "medium" });
|
||||
const entered = Promise.withResolvers<void>(), release = Promise.withResolvers<void>();
|
||||
const factory = companySearch.companySearchService;
|
||||
let hold = true;
|
||||
const searches = vi.fn();
|
||||
vi.spyOn(companySearch, "companySearchService").mockImplementation(owner => {
|
||||
const service = factory(owner), actual = service.search;
|
||||
service.search = async (...args) => {
|
||||
searches();
|
||||
const result = await actual(...args);
|
||||
if (hold) { hold = false; entered.resolve(); await release.promise; }
|
||||
return result;
|
||||
};
|
||||
return service;
|
||||
});
|
||||
const service = statusCardEvaluationService(db, { wakeup });
|
||||
const pass = service.run(new AbortController().signal);
|
||||
await entered.promise;
|
||||
const cards = statusCardService(db);
|
||||
let scheduled!: Date;
|
||||
try {
|
||||
scheduled = (await cards.update((await cards.getById(created.id))!, { refreshPolicy: policy("interval") }, actor)).nextEvalAt!;
|
||||
} finally { release.resolve(); }
|
||||
await pass;
|
||||
const current = (await cards.getById(created.id))!;
|
||||
expect(current.refreshPolicy.mode).toBe("interval");
|
||||
expect(current.nextEvalAt!.getTime()).toBe(scheduled.getTime());
|
||||
expect(current.nextEvalReason).toBe("refresh");
|
||||
expect(searches).toHaveBeenCalledOnce();
|
||||
expect(wakeup).not.toHaveBeenCalled();
|
||||
const intervalCard = (await cards.getById(created.id))!;
|
||||
await issueService(db).create(companyId, { title: "Next interval blocker", status: "blocked", priority: "medium" });
|
||||
vi.setSystemTime(intervalCard.nextEvalAt!);
|
||||
await service.run(new AbortController().signal);
|
||||
expect(wakeup).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it("keeps future failure retries fenced from ordinary company hints and permits explicit policy recovery", async () => {
|
||||
const created = await card();
|
||||
const factory = companySearch.companySearchService;
|
||||
let failing = true;
|
||||
const searches = vi.fn();
|
||||
vi.spyOn(companySearch, "companySearchService").mockImplementation(owner => {
|
||||
const service = factory(owner), actual = service.search;
|
||||
service.search = async (...args) => { searches(); if (failing) throw new Error("Controlled search unavailable"); return actual(...args); };
|
||||
return service;
|
||||
});
|
||||
const service = statusCardEvaluationService(db, { wakeup });
|
||||
await service.run(new AbortController().signal);
|
||||
const failed = (await statusCardService(db).getById(created.id))!;
|
||||
expect(failed.nextEvalReason).toBe("retry");
|
||||
const retryAt = failed.nextEvalAt!.getTime();
|
||||
expect(searches).toHaveBeenCalledOnce();
|
||||
for (let index = 0; index < 5; index++) {
|
||||
await issueService(db).create(companyId, { title: `Retry blocker ${index}`, status: "blocked", priority: "medium" });
|
||||
service.changed(new Set([`changes:${companyId}`]));
|
||||
await service.run(new AbortController().signal);
|
||||
}
|
||||
expect(searches).toHaveBeenCalledOnce();
|
||||
expect((await statusCardService(db).getById(created.id))!.nextEvalAt!.getTime()).toBe(retryAt);
|
||||
const restarted = statusCardEvaluationService(db, { wakeup });
|
||||
restarted.changed(new Set([`changes:${companyId}`]));
|
||||
await restarted.run(new AbortController().signal);
|
||||
expect(searches).toHaveBeenCalledOnce();
|
||||
expect(await restarted.next()).toBe(retryAt);
|
||||
vi.setSystemTime(retryAt);
|
||||
await restarted.run(new AbortController().signal);
|
||||
expect(searches).toHaveBeenCalledTimes(2);
|
||||
const cards = statusCardService(db);
|
||||
failing = false;
|
||||
await cards.update((await cards.getById(created.id))!, { refreshPolicy: { ...policy(), debounceSeconds: 1 } }, actor);
|
||||
await restarted.run(new AbortController().signal);
|
||||
expect(searches).toHaveBeenCalledTimes(3);
|
||||
const pending = (await cards.getById(created.id))!;
|
||||
expect(pending.nextEvalReason).toBe("policy");
|
||||
vi.setSystemTime(pending.nextEvalAt!);
|
||||
await restarted.run(new AbortController().signal);
|
||||
expect(wakeup).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it("retains the durable five-minute retry after evaluation failure across restart and drain", async () => {
|
||||
const created = await card();
|
||||
await db.update(statusCards).set({ queries: [{ scope: "invalid" } as never], nextEvalAt: new Date() }).where(eq(statusCards.id, created.id));
|
||||
|
||||
@@ -304,6 +304,11 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
|
||||
}
|
||||
|
||||
if (policy.scopeType === "project") {
|
||||
// Admission rechecks the same hard stop. Lock current eligibility before
|
||||
// publishing intent so a no-op pause cannot reactivate its own receipts.
|
||||
const [project] = await db.select({ pausedAt: projects.pausedAt }).from(projects)
|
||||
.where(and(eq(projects.id, policy.scopeId), eq(projects.companyId, policy.companyId))).for("update");
|
||||
if (!project || project.pausedAt !== null) return;
|
||||
await notifyRunDispatch(db, { companyId: policy.companyId });
|
||||
await notifyDeferredExecutionEligibility(db, { companyId: policy.companyId });
|
||||
await db
|
||||
@@ -313,7 +318,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
|
||||
pausedAt: now,
|
||||
updatedAt: now,
|
||||
})
|
||||
.where(and(eq(projects.id, policy.scopeId), isNull(projects.pausedAt)));
|
||||
.where(and(eq(projects.id, policy.scopeId), eq(projects.companyId, policy.companyId), isNull(projects.pausedAt)));
|
||||
return;
|
||||
}
|
||||
|
||||
|
||||
@@ -330,9 +330,9 @@ export function githubConnectionEventService(
|
||||
// Lifecycle webhooks carry IDs, not the user token’s complete repository view.
|
||||
// Discard the snapshot until Refresh access verifies it again.
|
||||
accessRevision: randomUUID(),
|
||||
...(unavailable ? { bindingSync: {
|
||||
...(unavailable || github.bindingSync || !github.installationIds.includes(binding.installationId) ? { bindingSync: {
|
||||
id: randomUUID(), removeInstallationIds: [...new Set([
|
||||
...(github.bindingSync?.removeInstallationIds ?? []), binding.installationId,
|
||||
...(github.bindingSync?.removeInstallationIds ?? []), ...(unavailable ? [binding.installationId] : []),
|
||||
])].filter(id => !installationIds.includes(id)),
|
||||
} } : {}),
|
||||
repositories: undefined,
|
||||
|
||||
@@ -643,14 +643,17 @@ export function statusCardService(
|
||||
? membershipAt : interval;
|
||||
}
|
||||
|
||||
async function requestRefresh(cardId: string, input: {
|
||||
async function refreshCurrentCard(cardId: string, input: {
|
||||
full?: boolean;
|
||||
trigger?: "manual" | "interval" | "reactive" | "restore";
|
||||
actor?: StatusCardActor;
|
||||
now?: Date;
|
||||
} = {}) {
|
||||
const card = await getById(cardId);
|
||||
if (!card) throw notFound("Status card not found");
|
||||
} = {}, superseded = false) {
|
||||
const [source] = await db.select({ card: statusCards, revision: sql<string>`${statusCards}.xmin::text` })
|
||||
.from(statusCards).where(eq(statusCards.id, cardId));
|
||||
if (!source) throw notFound("Status card not found");
|
||||
const { card, revision } = source;
|
||||
const currentRevision = sql`${statusCards}.xmin::text = ${revision}`;
|
||||
if (card.archivedAt) throw unprocessable("Archived status cards cannot be refreshed");
|
||||
if (card.queries.length === 0) throw conflict("Compile the status-card query before refreshing it");
|
||||
if (card.generatingIssueId) {
|
||||
@@ -663,6 +666,10 @@ export function statusCardService(
|
||||
}
|
||||
|
||||
const now = input.now ?? new Date();
|
||||
if (superseded && input.trigger !== undefined && input.trigger !== "manual" && input.trigger !== "restore"
|
||||
&& (card.refreshPolicy.mode === "manual" || (card.refreshPolicy.mode === "interval" && card.nextEvalAt && card.nextEvalAt > now))) {
|
||||
return { card, generatingIssue: null, alreadyGenerating: false, enqueued: false };
|
||||
}
|
||||
let membershipAt: Date | null = null;
|
||||
const snapshot = await executeQueries(card, {}, at => { if (at && (!membershipAt || at < membershipAt)) membershipAt = at; });
|
||||
const fingerprint = buildStatusCardFingerprint(snapshot);
|
||||
@@ -679,8 +686,9 @@ export function statusCardService(
|
||||
state: "active",
|
||||
nextEvalAt,
|
||||
nextEvalReason: nextEvalAt ? card.refreshPolicy.mode === "reactive" ? "membership" : "refresh" : null,
|
||||
}).where(eq(statusCards.id, card.id)).returning());
|
||||
return { card: next!, generatingIssue: null, alreadyGenerating: false, enqueued: false };
|
||||
}).where(and(eq(statusCards.id, card.id), currentRevision)).returning());
|
||||
if (!next) return undefined;
|
||||
return { card: next, generatingIssue: null, alreadyGenerating: false, enqueued: false };
|
||||
}
|
||||
|
||||
const hourAgo = new Date(now.getTime() - 60 * 60 * 1000);
|
||||
@@ -712,8 +720,9 @@ export function statusCardService(
|
||||
completedUpdateTimes: recent.filter(row => row.kind !== "compile" && row.finishedAt).map(row => row.startedAt),
|
||||
tokensToday: Number(daily[0]?.tokens ?? 0),
|
||||
}),
|
||||
}).where(eq(statusCards.id, card.id)).returning());
|
||||
return { card: next!, generatingIssue: null, alreadyGenerating: false, enqueued: false };
|
||||
}).where(and(eq(statusCards.id, card.id), currentRevision)).returning());
|
||||
if (!next) return undefined;
|
||||
return { card: next, generatingIssue: null, alreadyGenerating: false, enqueued: false };
|
||||
}
|
||||
|
||||
const history = await listUpdates(card.id);
|
||||
@@ -765,13 +774,14 @@ export function statusCardService(
|
||||
nextEvalAt,
|
||||
nextEvalReason: nextEvalAt ? card.refreshPolicy.mode === "reactive" ? "membership" : "refresh" : null,
|
||||
failureReason: null,
|
||||
}).where(and(eq(statusCards.id, card.id), isNull(statusCards.archivedAt), or(isNull(statusCards.generatingIssueId), priorGenerationPredicate))).returning());
|
||||
}).where(and(eq(statusCards.id, card.id), isNull(statusCards.archivedAt), currentRevision, or(isNull(statusCards.generatingIssueId), priorGenerationPredicate))).returning());
|
||||
if (!next) {
|
||||
const winner = await getById(card.id);
|
||||
if (!winner?.generatingIssueId) {
|
||||
if (!TERMINAL_ISSUE_STATUSES.has(generationIssue!.status)) {
|
||||
await issuesSvc().update(generationIssue!.id, { status: "cancelled" });
|
||||
}
|
||||
if (winner && !winner.archivedAt) return undefined;
|
||||
throw conflict("Status-card refresh claim was lost");
|
||||
}
|
||||
if (generationIssue!.id !== winner.generatingIssueId && !TERMINAL_ISSUE_STATUSES.has(generationIssue!.status)) {
|
||||
@@ -794,6 +804,14 @@ export function statusCardService(
|
||||
return { card: next, generatingIssue: generationIssue!, alreadyGenerating: deduplicated, enqueued: true, kind, changes };
|
||||
}
|
||||
|
||||
async function requestRefresh(cardId: string, input: Parameters<typeof refreshCurrentCard>[1] = {}) {
|
||||
for (let attempt = 0; attempt < 3; attempt += 1) {
|
||||
const result = await refreshCurrentCard(cardId, input, attempt > 0);
|
||||
if (result) return result;
|
||||
}
|
||||
throw conflict("Status-card refresh was superseded by concurrent edits");
|
||||
}
|
||||
|
||||
async function tickDueStatusCards(now = new Date(), reactiveCompanies?: ReadonlySet<string> | "changed", reconstruct = false) {
|
||||
const changedScope = reactiveCompanies === "changed" ? sql`true` : reactiveCompanies?.size
|
||||
? inArray(statusCards.companyId, [...reactiveCompanies]) : sql`false`;
|
||||
@@ -811,7 +829,9 @@ export function statusCardService(
|
||||
.innerJoin(companies, eq(companies.id, statusCards.companyId))
|
||||
.where(and(eq(companies.status, "active"), isNull(statusCards.archivedAt), isNull(statusCards.generatingIssueId),
|
||||
or(and(isNotNull(statusCards.nextEvalAt), lte(statusCards.nextEvalAt, now)),
|
||||
and(reactiveScope, ne(statusCards.state, "error"), ne(statusCards.state, "compiling"), sql`${statusCards.refreshPolicy}->>'mode' = 'reactive'`, sql`jsonb_array_length(${statusCards.queries}) > 0`))))
|
||||
and(reactiveScope,
|
||||
or(isNull(statusCards.nextEvalReason), ne(statusCards.nextEvalReason, "retry"), isNull(statusCards.nextEvalAt), lte(statusCards.nextEvalAt, now)),
|
||||
ne(statusCards.state, "error"), ne(statusCards.state, "compiling"), sql`${statusCards.refreshPolicy}->>'mode' = 'reactive'`, sql`jsonb_array_length(${statusCards.queries}) > 0`))))
|
||||
.then(rows => rows.map(row => row.card));
|
||||
const enqueued: Array<{ cardId: string; generatingIssue: typeof issues.$inferSelect }> = [];
|
||||
let evaluated = 0;
|
||||
|
||||
@@ -18,6 +18,8 @@ function asRecord(value: unknown): Record<string, unknown> {
|
||||
return {};
|
||||
}
|
||||
|
||||
type RecordGitHubAttempt = (grant: typeof connectionGrants.$inferSelect) => void;
|
||||
|
||||
type GitHubContinuityDependencies = {
|
||||
now: () => Date;
|
||||
currentCloudConnector: () => PaperclipCloudConnector | null;
|
||||
@@ -30,6 +32,7 @@ export function createGitHubGrantContinuity(db: Db, { now, currentCloudConnector
|
||||
connection: typeof toolConnections.$inferSelect,
|
||||
initialGrant: typeof connectionGrants.$inferSelect,
|
||||
actor?: ActorInfo,
|
||||
recordAttempt?: RecordGitHubAttempt,
|
||||
) {
|
||||
await actor?.managementContext?.reauthorize();
|
||||
return measureRunPhase("connectors.github.refresh", async () => {
|
||||
@@ -38,6 +41,7 @@ export function createGitHubGrantContinuity(db: Db, { now, currentCloudConnector
|
||||
connection,
|
||||
initialGrant,
|
||||
actor,
|
||||
recordAttempt,
|
||||
);
|
||||
} catch (error) {
|
||||
if (
|
||||
@@ -56,7 +60,7 @@ export function createGitHubGrantContinuity(db: Db, { now, currentCloudConnector
|
||||
),
|
||||
));
|
||||
if (!current || current.status !== "active") throw error;
|
||||
return refreshManagedGitHubGrantAccessOnce(connection, current, actor);
|
||||
return refreshManagedGitHubGrantAccessOnce(connection, current, actor, recordAttempt);
|
||||
}
|
||||
});
|
||||
}
|
||||
@@ -65,6 +69,7 @@ export function createGitHubGrantContinuity(db: Db, { now, currentCloudConnector
|
||||
connection: typeof toolConnections.$inferSelect,
|
||||
initialGrant: typeof connectionGrants.$inferSelect,
|
||||
actor?: ActorInfo,
|
||||
recordAttempt?: RecordGitHubAttempt,
|
||||
) {
|
||||
await actor?.managementContext?.reauthorize();
|
||||
let grant = await measureRunPhase("connectors.github.credentials", () => refreshOAuthGrantCredentials({
|
||||
@@ -73,6 +78,7 @@ export function createGitHubGrantContinuity(db: Db, { now, currentCloudConnector
|
||||
grantId: initialGrant.id,
|
||||
actor,
|
||||
}));
|
||||
recordAttempt?.(grant);
|
||||
const resolveAccessToken = async () => measureRunPhase("connectors.github.credentials", async () => {
|
||||
const accessRef = grant.credentialSecretRefs.find(
|
||||
(ref) => ref.configPath === "oauth.access_token",
|
||||
@@ -113,6 +119,7 @@ export function createGitHubGrantContinuity(db: Db, { now, currentCloudConnector
|
||||
forceRefresh: true,
|
||||
actor,
|
||||
}));
|
||||
recordAttempt?.(grant);
|
||||
accessToken = await resolveAccessToken();
|
||||
try {
|
||||
metadata = await measureRunPhase("connectors.github.repositories", () => loadGitHubGrantMetadata(
|
||||
@@ -204,7 +211,7 @@ export function createGitHubGrantContinuity(db: Db, { now, currentCloudConnector
|
||||
return { updated, previousGitHub };
|
||||
});
|
||||
|
||||
return measureRunPhase("connectors.github.webhooks", () => syncGitHubGrantBindings(connection, updated, actor, accessToken));
|
||||
return measureRunPhase("connectors.github.webhooks", () => syncGitHubGrantBindings(connection, updated, actor, accessToken, recordAttempt));
|
||||
}
|
||||
|
||||
async function syncGitHubGrantBindings(
|
||||
@@ -212,6 +219,7 @@ export function createGitHubGrantContinuity(db: Db, { now, currentCloudConnector
|
||||
initialGrant: typeof connectionGrants.$inferSelect,
|
||||
actor?: ActorInfo,
|
||||
accessToken?: string,
|
||||
recordAttempt?: RecordGitHubAttempt,
|
||||
) {
|
||||
await actor?.managementContext?.reauthorize();
|
||||
const cloudConnector = currentCloudConnector();
|
||||
@@ -243,9 +251,11 @@ export function createGitHubGrantContinuity(db: Db, { now, currentCloudConnector
|
||||
const subject = current.kind === "agent" && current.subjectAgentId ? `agent:${current.subjectAgentId}`
|
||||
: current.kind === "user" ? current.subjectUserId : null;
|
||||
if (!subject) throw unprocessable("GitHub webhook subject is unavailable");
|
||||
recordAttempt?.(grant);
|
||||
const changes = [
|
||||
...github.installationIds.map(installationId => ({ installationId, active: true as const })),
|
||||
...github.bindingSync.removeInstallationIds.map(installationId => ({ installationId, active: false as const })),
|
||||
...github.bindingSync.removeInstallationIds.filter(id => !github.installationIds.includes(id))
|
||||
.map(installationId => ({ installationId, active: false as const })),
|
||||
];
|
||||
for (let offset = 0; offset < changes.length; offset += 4) {
|
||||
const outcomes = await Promise.allSettled(changes.slice(offset, offset + 4).map(change => cloudConnector.setWebhookBinding({
|
||||
@@ -265,13 +275,14 @@ export function createGitHubGrantContinuity(db: Db, { now, currentCloudConnector
|
||||
});
|
||||
}
|
||||
|
||||
async function retainRetry(connection: typeof toolConnections.$inferSelect, grantId: string, error?: unknown, minimumMs = 30_000) {
|
||||
async function retainRetry(connection: typeof toolConnections.$inferSelect, grantId: string, attemptedRevision: string, error?: unknown, minimumMs = 30_000) {
|
||||
await db.transaction(async tx => {
|
||||
const [current] = await tx.select().from(connectionGrants).where(and(
|
||||
eq(connectionGrants.id, grantId), eq(connectionGrants.companyId, connection.companyId),
|
||||
eq(connectionGrants.connectionId, connection.id), eq(connectionGrants.status, "active"),
|
||||
)).for("update").limit(1);
|
||||
if (!current?.providerTenant?.github) return;
|
||||
// Backoff belongs to the attempted credential/installation revision.
|
||||
if (!current?.providerTenant?.github || githubContinuityRevision(current) !== attemptedRevision) return;
|
||||
const attempts = Math.min(16, (current.providerTenant.github.continuityRetry?.attempts ?? 0) + 1);
|
||||
const advised = error instanceof PaperclipCloudConnectorError ? error.retryAt
|
||||
: error instanceof HttpError ? asRecord(error.details).retryAt : undefined;
|
||||
@@ -280,7 +291,7 @@ export function createGitHubGrantContinuity(db: Db, { now, currentCloudConnector
|
||||
await notifyGitHubCredentialChange(tx, connection);
|
||||
await tx.update(connectionGrants).set({
|
||||
providerTenant: sql`jsonb_set(${connectionGrants.providerTenant}, '{github,continuityRetry}', ${JSON.stringify({
|
||||
at, attempts, revision: githubContinuityRevision(current),
|
||||
at, attempts, revision: attemptedRevision,
|
||||
})}::jsonb)`,
|
||||
updatedAt: now(),
|
||||
}).where(and(eq(connectionGrants.id, current.id), eq(connectionGrants.companyId, connection.companyId)));
|
||||
@@ -324,18 +335,20 @@ export function createGitHubGrantContinuity(db: Db, { now, currentCloudConnector
|
||||
let refreshed = 0;
|
||||
let failed = 0;
|
||||
for (const row of due) {
|
||||
let attemptedRevision = githubContinuityRevision(row.grant);
|
||||
const recordAttempt: RecordGitHubAttempt = grant => { attemptedRevision = githubContinuityRevision(grant); };
|
||||
try {
|
||||
const actor: ActorInfo = { actorType: "system", actorId: "github-connection-continuity" };
|
||||
const updated = row.grant.providerTenant?.github?.bindingSync
|
||||
? await syncGitHubGrantBindings(row.connection, row.grant, actor)
|
||||
: await refreshManagedGitHubGrantAccess(row.connection, row.grant, actor);
|
||||
? await syncGitHubGrantBindings(row.connection, row.grant, actor, undefined, recordAttempt)
|
||||
: await refreshManagedGitHubGrantAccess(row.connection, row.grant, actor, recordAttempt);
|
||||
// A successful provider response can still leave an unchanged/short-lived
|
||||
// credential due. Retain that obligation with bounded durable backoff.
|
||||
const next = githubGrantContinuityDeadline(row.connection, updated);
|
||||
if (next !== null && next <= now().getTime()) await retainRetry(row.connection, updated.id, undefined, 60 * 60_000);
|
||||
if (next !== null && next <= now().getTime()) await retainRetry(row.connection, updated.id, githubContinuityRevision(updated), undefined, 60 * 60_000);
|
||||
refreshed += 1;
|
||||
} catch (error) {
|
||||
await retainRetry(row.connection, row.grant.id, error);
|
||||
await retainRetry(row.connection, row.grant.id, attemptedRevision, error);
|
||||
failed += 1;
|
||||
}
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user