diff --git a/services/security-sync/README.md b/services/security-sync/README.md index 7800f552fc..0eaef52a02 100644 --- a/services/security-sync/README.md +++ b/services/security-sync/README.md @@ -9,10 +9,22 @@ Cloudflare Worker that syncs security alerts on a cron schedule, enqueuing one q - `POST /internal/dismiss-finding` - dismissal command ingress; `DISMISS_FINDING_COMMAND_ROUTING_ENABLED=false` pauses new Worker dismissal commands - Cron trigger (`0 */6 * * *`) — queries enabled owners from DB and enqueues sync messages -## Queue +## Queues -- Producer binding: `SYNC_QUEUE` -- Consumer queue: `security-sync-jobs` (`security-sync-jobs-dev` in dev) -- DLQ: `security-sync-jobs-dlq` +- Sync producer: `SYNC_QUEUE` → `security-sync-jobs` (`security-sync-jobs-dev` in dev) +- Sync DLQ: `security-sync-jobs-dlq` +- Dismiss producer: `DISMISS_QUEUE` → `security-dismiss-jobs` (`security-dismiss-jobs-dev` in dev) +- Dismiss DLQ: `security-dismiss-jobs-dlq` -The consumer calls `syncOwner` which fetches Dependabot alerts from GitHub, upserts findings into the database, keeps automatic-analysis queue eligibility synchronized, and prunes stale repos from the config for both scheduled and manual sync paths. +Create the dismiss queues before the first production deploy: + +```bash +pnpm --filter cloudflare-security-sync exec wrangler queues create security-dismiss-jobs +pnpm --filter cloudflare-security-sync exec wrangler queues create security-dismiss-jobs-dlq +pnpm --filter cloudflare-security-sync exec wrangler queues create security-dismiss-jobs-dev +pnpm --filter cloudflare-security-sync exec wrangler queues create security-dismiss-jobs-dlq-dev +``` + +The sync consumer calls `syncOwner`. A scheduled owner run stops at an 8-minute budget, persists progress on `agent_configs.runtime_state.sync_run`, and enqueues a continuation. Owner freshness and config pruning run only after every selected repository has a terminal outcome for that run. + +The dismiss consumer is isolated from scheduled sync occupancy. The sync consumer still accepts leftover dismiss messages during deploy drain. diff --git a/services/security-sync/src/dismiss.ts b/services/security-sync/src/dismiss.ts index ea44c85bcd..0a660a4f10 100644 --- a/services/security-sync/src/dismiss.ts +++ b/services/security-sync/src/dismiss.ts @@ -77,39 +77,75 @@ async function getDismissalAuditActor( return buildSecurityFindingAuditHumanActor(actor); } +async function timedDismissalStage( + stage: string, + context: { commandId: string; findingId: string; runId: string }, + work: () => Promise +): Promise { + const started = Date.now(); + try { + const result = await work(); + console.info('Security Agent dismissal stage completed', { + stage, + duration_ms: Date.now() - started, + command_id: context.commandId, + finding_id: context.findingId, + run_id: context.runId, + }); + return result; + } catch (error) { + console.error('Security Agent dismissal stage failed', { + stage, + duration_ms: Date.now() - started, + command_id: context.commandId, + finding_id: context.findingId, + run_id: context.runId, + error_type: error instanceof Error ? error.name : 'UnknownError', + }); + throw error; + } +} + export async function processSecurityFindingDismissal(params: { db: WorkerDb; gitTokenService: GitTokenService; message: SecurityDismissMessage; }): Promise { - const rows = await params.db - .select({ - id: security_findings.id, - source: security_findings.source, - source_id: security_findings.source_id, - repo_full_name: security_findings.repo_full_name, - title: security_findings.title, - severity: security_findings.severity, - status: security_findings.status, - package_name: security_findings.package_name, - package_ecosystem: security_findings.package_ecosystem, - manifest_path: security_findings.manifest_path, - patched_version: security_findings.patched_version, - ghsa_id: security_findings.ghsa_id, - cve_id: security_findings.cve_id, - cwe_ids: security_findings.cwe_ids, - cvss_score: security_findings.cvss_score, - dependabot_html_url: security_findings.dependabot_html_url, - first_detected_at: security_findings.first_detected_at, - fixed_at: security_findings.fixed_at, - sla_due_at: security_findings.sla_due_at, - session_id: security_findings.session_id, - owned_by_organization_id: security_findings.owned_by_organization_id, - owned_by_user_id: security_findings.owned_by_user_id, - }) - .from(security_findings) - .where(eq(security_findings.id, params.message.findingId)) - .limit(1); + const stageContext = { + commandId: params.message.commandId, + findingId: params.message.findingId, + runId: params.message.runId, + }; + const rows = await timedDismissalStage('load_finding', stageContext, () => + params.db + .select({ + id: security_findings.id, + source: security_findings.source, + source_id: security_findings.source_id, + repo_full_name: security_findings.repo_full_name, + title: security_findings.title, + severity: security_findings.severity, + status: security_findings.status, + package_name: security_findings.package_name, + package_ecosystem: security_findings.package_ecosystem, + manifest_path: security_findings.manifest_path, + patched_version: security_findings.patched_version, + ghsa_id: security_findings.ghsa_id, + cve_id: security_findings.cve_id, + cwe_ids: security_findings.cwe_ids, + cvss_score: security_findings.cvss_score, + dependabot_html_url: security_findings.dependabot_html_url, + first_detected_at: security_findings.first_detected_at, + fixed_at: security_findings.fixed_at, + sla_due_at: security_findings.sla_due_at, + session_id: security_findings.session_id, + owned_by_organization_id: security_findings.owned_by_organization_id, + owned_by_user_id: security_findings.owned_by_user_id, + }) + .from(security_findings) + .where(eq(security_findings.id, params.message.findingId)) + .limit(1) + ); const finding = rows[0]; if (!finding || !findingMatchesOwner(finding, params.message.owner)) { @@ -153,72 +189,78 @@ export async function processSecurityFindingDismissal(params: { }; } - const token = await params.gitTokenService.getToken(params.message.installationId); - const response = await fetch( - `https://api.github.com/repos/${target.repoOwner}/${target.repoName}/dependabot/alerts/${target.alertNumber}`, - - { - method: 'PATCH', - headers: { - Authorization: `Bearer ${token}`, - Accept: 'application/vnd.github+json', - 'Content-Type': 'application/json', - 'X-GitHub-Api-Version': '2022-11-28', - 'User-Agent': 'cloudflare-security-sync', - }, - body: JSON.stringify({ - state: 'dismissed', - dismissed_reason: params.message.reason, - dismissed_comment: params.message.comment, - }), - } - ); - - if (!response.ok) { - throw new Error( - `GitHub Dependabot dismissal failed with ${response.status} for finding ${finding.id}` + await timedDismissalStage('github_writeback', stageContext, async () => { + const token = await params.gitTokenService.getToken(params.message.installationId); + const response = await fetch( + `https://api.github.com/repos/${target.repoOwner}/${target.repoName}/dependabot/alerts/${target.alertNumber}`, + { + method: 'PATCH', + headers: { + Authorization: `Bearer ${token}`, + Accept: 'application/vnd.github+json', + 'Content-Type': 'application/json', + 'X-GitHub-Api-Version': '2022-11-28', + 'User-Agent': 'cloudflare-security-sync', + }, + body: JSON.stringify({ + state: 'dismissed', + dismissed_reason: params.message.reason, + dismissed_comment: params.message.comment, + }), + } ); - } + + if (!response.ok) { + throw new Error( + `GitHub Dependabot dismissal failed with ${response.status} for finding ${finding.id}` + ); + } + }); } - const actor = await getDismissalAuditActor(params.db, params.message.actor.id); + const actor = await timedDismissalStage('load_actor', stageContext, () => + getDismissalAuditActor(params.db, params.message.actor.id) + ); - await params.db.transaction(async tx => { - await tx - .update(security_findings) - .set({ - status: 'ignored', - ignored_reason: params.message.reason, - ignored_by: actor.email ?? actor.id, - updated_at: sql`now()`, - }) - .where(eq(security_findings.id, finding.id)); - - await insertSecurityFindingAuditEvent(tx, { - owner: toAuditOwner(params.message.owner), - finding: { ...finding, status: 'ignored' }, - actor, - action: SecurityAuditLogAction.FindingDismissed, - occurredAt: new Date(), - eventKey: dismissalEventKey({ - owner: params.message.owner, - findingId: finding.id, - commandId: params.message.commandId, - }), - sourceContext: SecurityFindingAuditSourceContext.SecuritySync, - beforeState: { status: finding.status }, - afterState: { status: 'ignored', reason_code: params.message.reason }, - metadata: { - source: finding.source, - run_id: params.message.runId, - command_id: params.message.commandId, - message_id: params.message.messageId, - trigger: 'worker_queue', - reason_code: params.message.reason, - source_writeback_outcome: finding.source === 'dependabot' ? 'dismissed' : 'not_applicable', - }, - }); - }); + await timedDismissalStage('persist_dismissal', stageContext, () => + params.db.transaction(async tx => { + await tx + .update(security_findings) + .set({ + status: 'ignored', + ignored_reason: params.message.reason, + ignored_by: actor.email ?? actor.id, + updated_at: sql`now()`, + }) + .where(eq(security_findings.id, finding.id)); + + await insertSecurityFindingAuditEvent(tx, { + owner: toAuditOwner(params.message.owner), + finding: { ...finding, status: 'ignored' }, + actor, + action: SecurityAuditLogAction.FindingDismissed, + occurredAt: new Date(), + eventKey: dismissalEventKey({ + owner: params.message.owner, + findingId: finding.id, + commandId: params.message.commandId, + }), + sourceContext: SecurityFindingAuditSourceContext.SecuritySync, + beforeState: { status: finding.status }, + afterState: { status: 'ignored', reason_code: params.message.reason }, + metadata: { + source: finding.source, + run_id: params.message.runId, + command_id: params.message.commandId, + message_id: params.message.messageId, + trigger: 'worker_queue', + reason_code: params.message.reason, + source_writeback_outcome: + finding.source === 'dependabot' ? 'dismissed' : 'not_applicable', + }, + }); + }) + ); return { dismissed: true, diff --git a/services/security-sync/src/index.test.ts b/services/security-sync/src/index.test.ts index c11f73e10b..3cbe0513dc 100644 --- a/services/security-sync/src/index.test.ts +++ b/services/security-sync/src/index.test.ts @@ -13,6 +13,7 @@ import worker, { collectScheduledSyncOwners, type SecuritySyncQueueMessage } fro import { processSecurityFindingDismissal } from './dismiss.js'; import { runSecurityNotificationSweep } from './notifications/sweep.js'; import { syncOwner } from './sync.js'; +import type * as SyncModule from './sync.js'; vi.mock('@kilocode/db', async importOriginal => { const { @@ -33,7 +34,10 @@ vi.mock('@kilocode/db/client', () => ({ getWorkerDb: vi.fn() })); vi.mock('@kilocode/db/operation-ledger', () => ({ settleOperation: vi.fn() })); vi.mock('./dismiss.js', () => ({ processSecurityFindingDismissal: vi.fn() })); vi.mock('./notifications/sweep.js', () => ({ runSecurityNotificationSweep: vi.fn() })); -vi.mock('./sync.js', () => ({ syncOwner: vi.fn() })); +vi.mock('./sync.js', async importOriginal => { + const actual = await importOriginal(); + return { ...actual, syncOwner: vi.fn() }; +}); beforeEach(() => { vi.clearAllMocks(); @@ -460,6 +464,70 @@ describe('manual sync dispatch', () => { expect(retry).not.toHaveBeenCalled(); }); + it('enqueues a continuation and keeps a manual command running when the owner budget is exhausted', async () => { + const queuedBatches: MessageSendRequest[][] = []; + vi.mocked(getWorkerDb).mockReturnValue(workerDbStub()); + vi.mocked(syncOwner).mockResolvedValue({ + synced: 1, + errors: 0, + staleRepos: [], + exhaustedBudget: true, + remainingRepoCount: 3, + } as never); + const ack = vi.fn(); + const retry = vi.fn(); + + await worker.queue( + { + messages: [ + { + attempts: 1, + body: { + schemaVersion: 1, + commandId: 'dddddddd-dddd-4ddd-8ddd-dddddddddddd', + runId: 'cccccccc-cccc-4ccc-8ccc-cccccccccccc', + messageId: 'run:org:manual', + trigger: 'manual', + owner: { organizationId: 'aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa' }, + ownerKey: 'org:aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa', + chunkIndex: 0, + chunkCount: 1, + dispatchedAt: '2026-05-18T08:30:00.000Z', + actor: { id: 'user-123' }, + }, + ack, + retry, + }, + ], + } as never, + { + HYPERDRIVE: { connectionString: 'postgres://worker' }, + GIT_TOKEN_SERVICE: {}, + SYNC_QUEUE: { + sendBatch: async (batch: MessageSendRequest[]) => { + queuedBatches.push(batch); + }, + }, + } as CloudflareEnv + ); + + expect(queuedBatches[0]?.[0]?.body).toMatchObject({ + commandId: 'dddddddd-dddd-4ddd-8ddd-dddddddddddd', + trigger: 'manual', + chunkIndex: 1, + chunkCount: 2, + messageId: + 'cccccccc-cccc-4ccc-8ccc-cccccccccccc:org:aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa:manual:1', + }); + expect(transitionSecurityAgentCommandWithCurrentState).toHaveBeenCalledTimes(1); + expect(transitionSecurityAgentCommandWithCurrentState).toHaveBeenCalledWith( + expect.anything(), + expect.objectContaining({ status: 'running' }) + ); + expect(ack).toHaveBeenCalledTimes(1); + expect(retry).not.toHaveBeenCalled(); + }); + it('accepts legacy OAuth user IDs in manual sync commands', async () => { const queuedBatches: MessageSendRequest[][] = []; const legacyUserId = 'oauth:google:1234567890'; @@ -599,12 +667,17 @@ describe('manual dismissal dispatch', () => { { INTERNAL_API_SECRET: { get: async () => 'worker-secret' }, HYPERDRIVE: { connectionString: 'postgres://worker' }, - SYNC_QUEUE: { - sendBatch: async batch => { + DISMISS_QUEUE: { + sendBatch: async (batch: MessageSendRequest[]) => { queuedBatches.push(batch); }, }, - } as CloudflareEnv + SYNC_QUEUE: { + sendBatch: async () => { + throw new Error('dismiss must not use SYNC_QUEUE'); + }, + }, + } as unknown as CloudflareEnv ); expect(response.status).toBe(202); @@ -643,7 +716,7 @@ describe('manual dismissal dispatch', () => { { INTERNAL_API_SECRET: { get: async () => 'worker-secret' }, HYPERDRIVE: { connectionString: 'postgres://worker' }, - SYNC_QUEUE: { + DISMISS_QUEUE: { sendBatch: async batch => { queuedBatches.push(batch); }, @@ -871,6 +944,42 @@ describe('manual dismissal dispatch', () => { expect(ack).not.toHaveBeenCalled(); expect(retry).toHaveBeenCalledTimes(1); }); + + it('acks an invalid message on the dismiss queue without running sync', async () => { + vi.mocked(getWorkerDb).mockReturnValue(workerDbStub()); + const ack = vi.fn(); + const retry = vi.fn(); + + await worker.queue( + { + queue: 'security-dismiss-jobs', + messages: [ + { + attempts: 1, + body: { + schemaVersion: 1, + runId: 'cccccccc-cccc-4ccc-8ccc-cccccccccccc', + messageId: 'scheduled-sync-message', + trigger: 'scheduled', + owner: { userId: 'user-123' }, + ownerKey: 'user:user-123', + chunkIndex: 0, + chunkCount: 1, + dispatchedAt: '2026-06-11T10:00:00.000Z', + }, + ack, + retry, + }, + ], + } as never, + { HYPERDRIVE: { connectionString: 'postgres://worker' } } as CloudflareEnv + ); + + expect(processSecurityFindingDismissal).not.toHaveBeenCalled(); + expect(syncOwner).not.toHaveBeenCalled(); + expect(ack).toHaveBeenCalledTimes(1); + expect(retry).not.toHaveBeenCalled(); + }); }); describe('same-key command dedupe (P1-A-08e)', () => { @@ -886,6 +995,23 @@ describe('same-key command dedupe (P1-A-08e)', () => { } as CloudflareEnv; } + function dismissEnv(queuedBatches: MessageSendRequest[][]) { + return { + INTERNAL_API_SECRET: { get: async () => 'worker-secret' }, + HYPERDRIVE: { connectionString: 'postgres://worker' }, + DISMISS_QUEUE: { + sendBatch: async (batch: MessageSendRequest[]) => { + queuedBatches.push(batch); + }, + }, + SYNC_QUEUE: { + sendBatch: async () => { + throw new Error('dismiss must not use SYNC_QUEUE'); + }, + }, + } as unknown as CloudflareEnv; + } + function manualSyncRequest(operationKey: string) { return new Request('https://security-sync.test/internal/manual-sync', { method: 'POST', @@ -994,11 +1120,11 @@ describe('same-key command dedupe (P1-A-08e)', () => { const first = await worker.fetch( dismissalRequest('dismiss-key-123'), - manualSyncEnv(queuedBatches) + dismissEnv(queuedBatches) ); const second = await worker.fetch( dismissalRequest('dismiss-key-123'), - manualSyncEnv(queuedBatches) + dismissEnv(queuedBatches) ); expect(first.status).toBe(202); diff --git a/services/security-sync/src/index.ts b/services/security-sync/src/index.ts index 52d3c46e6f..1b4abcd845 100644 --- a/services/security-sync/src/index.ts +++ b/services/security-sync/src/index.ts @@ -23,7 +23,7 @@ import { emitScheduledJobEvent, } from '@kilocode/worker-utils/scheduled-job-observability'; import { eq, and, isNotNull, or } from 'drizzle-orm'; -import { syncOwner } from './sync'; +import { SECURITY_SYNC_OWNER_BUDGET_MS, syncOwner } from './sync'; import { processSecurityFindingDismissal } from './dismiss'; import { runSecurityNotificationSweep } from './notifications/sweep'; @@ -243,7 +243,7 @@ function buildManualSyncQueueMessage( function buildDismissQueueMessage( command: z.infer, ids: KeyedEnqueueIds -): MessageSendRequest { +): MessageSendRequest { return { body: { ...command, @@ -265,15 +265,15 @@ function buildDismissQueueMessage( * A queue-send failure marks the command failed before rethrowing, so no * command is left waiting for a message that was never sent. */ -async function enqueueSecurityCommand( +async function enqueueSecurityCommand( db: WorkerDb, - queue: Queue, + queue: Queue, params: { operationKey?: string; create: CreateSecurityAgentCommandInput; runId: string; messageId: string; - buildMessage: (ids: KeyedEnqueueIds) => MessageSendRequest; + buildMessage: (ids: KeyedEnqueueIds) => MessageSendRequest; } ): Promise { const { command, created } = @@ -341,7 +341,7 @@ async function enqueueManualSyncCommand( async function enqueueDismissFindingCommand( db: WorkerDb, - queue: Queue, + queue: Queue, command: z.infer ): Promise { const runId = crypto.randomUUID(); @@ -652,8 +652,40 @@ async function processSecuritySyncMessage( env.SECURITY_NOTIFICATION_MATERIALIZATION_ENABLED, 'SECURITY_NOTIFICATION_MATERIALIZATION_ENABLED' ), + budgetMs: body.repoFullName ? undefined : SECURITY_SYNC_OWNER_BUDGET_MS, }); + if (result.exhaustedBudget) { + const nextChunkIndex = body.chunkIndex + 1; + await env.SYNC_QUEUE.sendBatch([ + { + body: { + ...body, + chunkIndex: nextChunkIndex, + chunkCount: nextChunkIndex + 1, + messageId: + body.trigger === 'manual' + ? `${body.runId}:${body.ownerKey}:manual:${nextChunkIndex}` + : `${body.runId}:${body.ownerKey}:${nextChunkIndex}`, + dispatchedAt: new Date().toISOString(), + }, + contentType: 'json', + }, + ]); + console.info('Security sync continuation enqueued', { + command_id: body.commandId, + command_type: body.commandId ? 'sync' : undefined, + owner_type: body.owner.organizationId ? 'org' : 'user', + runId: body.runId, + ownerKey: body.ownerKey, + chunkIndex: nextChunkIndex, + remainingRepoCount: result.remainingRepoCount, + attempts: message.attempts, + }); + message.ack(); + return; + } + const terminal = syncCommandTerminalState(result); if (body.commandId) { const terminalTransition = await transitionSecurityAgentCommandWithCurrentState(db, { @@ -772,7 +804,7 @@ export default { } const db = getWorkerDb(env.HYPERDRIVE.connectionString, { statement_timeout: 30_000 }); - const accepted = await enqueueDismissFindingCommand(db, env.SYNC_QUEUE, parsed.data); + const accepted = await enqueueDismissFindingCommand(db, env.DISMISS_QUEUE, parsed.data); return jsonResponse({ success: true, accepted: true, ...accepted }, 202); } @@ -885,11 +917,17 @@ export default { }, async queue(batch: MessageBatch, env: CloudflareEnv): Promise { + const dismissOnly = (batch.queue ?? '').startsWith('security-dismiss-jobs'); for (const message of batch.messages) { try { if (await processSecurityDismissMessage(message, env)) { continue; } + if (dismissOnly) { + console.error('Invalid security dismiss queue message'); + message.ack(); + continue; + } await processSecuritySyncMessage(message, env); } catch (error) { const correlation = commandCorrelation(message.body); diff --git a/services/security-sync/src/sync.test.ts b/services/security-sync/src/sync.test.ts index 6c447089b7..1165036931 100644 --- a/services/security-sync/src/sync.test.ts +++ b/services/security-sync/src/sync.test.ts @@ -15,6 +15,7 @@ afterEach(() => { type FakeDbOptions = { authInvalidAt?: string | null; repositories?: string[]; + runtimeState?: Record; }; function createFakeDb(options: FakeDbOptions = {}) { @@ -29,7 +30,14 @@ function createFakeDb(options: FakeDbOptions = {}) { limit: async () => { selectCount++; if (selectCount === 1) { - return [{ id: 'agent-config', config: {}, is_enabled: true }]; + return [ + { + id: 'agent-config', + config: {}, + is_enabled: true, + runtime_state: options.runtimeState ?? {}, + }, + ]; } if (selectCount === 2) { return [ @@ -68,6 +76,17 @@ function createFakeDb(options: FakeDbOptions = {}) { return { db, sets }; } +function runtimeStateSqlText(entry: Record): string { + const value = entry.runtime_state; + if (value == null || typeof value !== 'object') return ''; + const chunks = (value as { queryChunks?: Array<{ value?: unknown }> }).queryChunks; + if (!Array.isArray(chunks)) return ''; + return chunks + .flatMap(chunk => (Array.isArray(chunk.value) ? chunk.value : [])) + .filter(part => typeof part === 'string') + .join(''); +} + function createGitTokenService() { return { getToken: vi.fn(async () => 'github-token') }; } @@ -513,6 +532,126 @@ describe('Worker GitHub auth-invalid sync', () => { }); expect(auditRows[0]?.finding_snapshot).not.toHaveProperty('dependabot_html_url'); }); + + it('stops after the first repository when the owner budget is already exhausted', async () => { + const { db, sets } = createFakeDb({ repositories: ['acme/widgets', 'acme/api'] }); + const gitTokenService = createGitTokenService(); + const fetchStub = stubFetch(new Response(JSON.stringify([]), { status: 200 })); + + await expect( + syncOwner({ + db: db as never, + gitTokenService, + owner: { userId: 'user-1' }, + runId: 'run-budget-1', + budgetMs: 0, + }) + ).resolves.toMatchObject({ + exhaustedBudget: true, + remainingRepoCount: 1, + }); + + expect(fetchStub).toHaveBeenCalledTimes(1); + const progressWrite = sets.find(entry => entry.runtime_state != null)?.runtime_state; + expect(progressWrite).toBeDefined(); + expect(progressWrite).not.toHaveProperty('sync_run'); + expect(progressWrite).not.toHaveProperty('last_synced_at'); + }); + + it('counts a fresh-run GitHub failure toward the owner budget and does not mark it complete', async () => { + const { db } = createFakeDb({ repositories: ['acme/widgets', 'acme/api'] }); + const gitTokenService = createGitTokenService(); + const fetchStub = stubFetch(new Response('Service unavailable', { status: 500 })); + + await expect( + syncOwner({ + db: db as never, + gitTokenService, + owner: { userId: 'user-1' }, + runId: 'run-budget-fail', + budgetMs: 0, + }) + ).resolves.toMatchObject({ + exhaustedBudget: true, + remainingRepoCount: 2, + errors: 0, + }); + + expect(fetchStub).toHaveBeenCalledTimes(1); + }); + + it('does not keep an incomplete GitHub failure as an error after a successful retry', async () => { + const { db, sets } = createFakeDb({ + repositories: ['acme/widgets', 'acme/api'], + runtimeState: { + sync_run: { + runId: 'run-budget-retry', + completedRepos: [], + staleRepos: [], + authInvalidRepos: [], + synced: 0, + errors: 0, + skipped: 0, + authInvalid: 0, + reauthRequired: false, + }, + }, + }); + const gitTokenService = createGitTokenService(); + stubFetch(() => new Response(JSON.stringify([]), { status: 200 })); + + await expect( + syncOwner({ + db: db as never, + gitTokenService, + owner: { userId: 'user-1' }, + runId: 'run-budget-retry', + }) + ).resolves.toMatchObject({ + exhaustedBudget: false, + remainingRepoCount: 0, + errors: 0, + }); + + expect(sets.some(entry => runtimeStateSqlText(entry).includes('last_synced_at'))).toBe(true); + }); + + it('skips completed repositories and finalizes freshness on the last chunk', async () => { + const { db, sets } = createFakeDb({ + repositories: ['acme/widgets', 'acme/api'], + runtimeState: { + sync_run: { + runId: 'run-budget-1', + completedRepos: ['acme/widgets'], + staleRepos: [], + authInvalidRepos: [], + synced: 0, + errors: 0, + skipped: 0, + authInvalid: 0, + reauthRequired: false, + }, + }, + }); + const gitTokenService = createGitTokenService(); + const fetchStub = stubFetch(new Response(JSON.stringify([]), { status: 200 })); + + await expect( + syncOwner({ + db: db as never, + gitTokenService, + owner: { userId: 'user-1' }, + runId: 'run-budget-1', + }) + ).resolves.toMatchObject({ + exhaustedBudget: false, + remainingRepoCount: 0, + authInvalid: 0, + }); + + expect(fetchStub).toHaveBeenCalledTimes(1); + expect(sets).toContainEqual(expect.objectContaining({ runtime_state: expect.anything() })); + }); }); describe('Worker auto-analysis queue sync', () => { diff --git a/services/security-sync/src/sync.ts b/services/security-sync/src/sync.ts index e8c664ab68..ca68cfd56c 100644 --- a/services/security-sync/src/sync.ts +++ b/services/security-sync/src/sync.ts @@ -164,7 +164,23 @@ type SecurityReviewOwner = | { organizationId: string; userId?: never } | { userId: string; organizationId?: never }; -type SyncResult = { +export const SECURITY_SYNC_OWNER_BUDGET_MS = 8 * 60 * 1000; + +const SyncRunProgressSchema = z.object({ + runId: z.string().min(1), + completedRepos: z.array(z.string().min(1)), + staleRepos: z.array(z.string()), + authInvalidRepos: z.array(z.string()), + synced: z.number().int().nonnegative(), + errors: z.number().int().nonnegative(), + skipped: z.number().int().nonnegative(), + authInvalid: z.number().int().nonnegative(), + reauthRequired: z.boolean(), +}); + +type SyncRunProgress = z.infer; + +export type SyncResult = { synced: number; errors: number; /** Repos where Dependabot alerts are permanently disabled (safe to skip) */ @@ -177,6 +193,8 @@ type SyncResult = { staleRepos: string[]; /** Stable command-ledger result when an acknowledged sync resolves without normal success. */ commandResultCode?: string; + exhaustedBudget: boolean; + remainingRepoCount: number; }; type FetchAlertsResult = @@ -195,9 +213,87 @@ function createEmptySyncResult(): SyncResult { authInvalidRepos: [], reauthRequired: false, staleRepos: [], + exhaustedBudget: false, + remainingRepoCount: 0, }; } +function readSyncRunProgress( + runtimeState: Record, + runId: string +): SyncRunProgress | null { + const parsed = SyncRunProgressSchema.safeParse(runtimeState.sync_run); + if (!parsed.success || parsed.data.runId !== runId) return null; + return parsed.data; +} + +function applySyncRunProgress(result: SyncResult, progress: SyncRunProgress): void { + result.synced = progress.synced; + result.errors = progress.errors; + result.skipped = progress.skipped; + result.authInvalid = progress.authInvalid; + result.authInvalidRepos = [...progress.authInvalidRepos]; + result.reauthRequired = progress.reauthRequired; + result.staleRepos = [...progress.staleRepos]; +} + +function toSyncRunProgress( + runId: string, + result: SyncResult, + completedRepos: string[] +): SyncRunProgress { + return { + runId, + completedRepos, + staleRepos: result.staleRepos, + authInvalidRepos: result.authInvalidRepos, + synced: result.synced, + errors: result.errors, + skipped: result.skipped, + authInvalid: result.authInvalid, + reauthRequired: result.reauthRequired, + }; +} + +async function writeSyncRunProgress( + db: WorkerDb, + owner: SecurityReviewOwner, + progress: SyncRunProgress +): Promise { + await db + .update(agent_configs) + .set({ + runtime_state: sql`jsonb_set( + COALESCE(${agent_configs.runtime_state}, '{}'::jsonb), + '{sync_run}', + ${JSON.stringify(progress)}::jsonb, + true + )`, + }) + .where( + and( + eq(agent_configs.agent_type, 'security_scan'), + eq(agent_configs.platform, 'github'), + ownerFilter(owner) + ) + ); +} + +async function clearSyncRunProgress(db: WorkerDb, owner: SecurityReviewOwner): Promise { + await db + .update(agent_configs) + .set({ + runtime_state: sql`COALESCE(${agent_configs.runtime_state}, '{}'::jsonb) - 'sync_run'`, + }) + .where( + and( + eq(agent_configs.agent_type, 'security_scan'), + eq(agent_configs.platform, 'github'), + ownerFilter(owner) + ) + ); +} + function createAuthInvalidSyncResult(repositories: string[]): SyncResult { return { ...createEmptySyncResult(), @@ -272,6 +368,7 @@ type EnabledOwnerConfig = { /** Number of selected_repository_ids that are no longer accessible via the installation. * Non-zero means the app lost access to a configured repo — freshness must not advance. */ missingSelectedRepoCount: number; + runtimeState: Record; }; export async function getOwnerConfig( @@ -284,6 +381,7 @@ export async function getOwnerConfig( id: agent_configs.id, config: agent_configs.config, is_enabled: agent_configs.is_enabled, + runtime_state: agent_configs.runtime_state, }) .from(agent_configs) .where( @@ -401,6 +499,10 @@ export async function getOwnerConfig( autoAnalysisEnabledAt: ownerStates[0]?.autoAnalysisEnabledAt ?? null, authInvalidAt: integration.authInvalidAt, missingSelectedRepoCount, + runtimeState: + agentConfig.runtime_state && typeof agentConfig.runtime_state === 'object' + ? agentConfig.runtime_state + : {}, }; } @@ -1610,6 +1712,7 @@ export async function syncOwner(params: { actor?: { id: string; email?: string | null; name?: string | null }; repoFullName?: string; notificationMaterializationEnabled?: boolean; + budgetMs?: number; }): Promise { const { db: database, gitTokenService, owner, runId, actor, repoFullName } = params; const trigger = params.trigger ?? 'scheduled'; @@ -1646,9 +1749,23 @@ export async function syncOwner(params: { return createAuthInvalidSyncResult(repositories); } + const previousProgress = repoFullName ? null : readSyncRunProgress(config.runtimeState, runId); + const completedRepos = new Set(previousProgress?.completedRepos ?? []); + const remainingRepositories = repositories.filter(name => !completedRepos.has(name)); const totalResult = createEmptySyncResult(); + if (previousProgress) { + applySyncRunProgress(totalResult, previousProgress); + console.info('Resuming security sync owner run', { + runId, + completedRepoCount: completedRepos.size, + remainingRepoCount: remainingRepositories.length, + }); + } let firstError: Error | null = null; let successfulRepos = 0; + let processedThisPass = 0; + let incompleteFailures = 0; + let exhaustedBudget = false; const notificationPolicy = params.notificationMaterializationEnabled ? config.notificationPolicy : null; @@ -1656,7 +1773,16 @@ export async function syncOwner(params: { ? await resolveNotificationRecipientUserIds(database, owner) : []; - for (const repoFullName of repositories) { + for (const selectedRepoFullName of remainingRepositories) { + if ( + processedThisPass > 0 && + params.budgetMs !== undefined && + Date.now() - startTime >= params.budgetMs + ) { + exhaustedBudget = true; + break; + } + try { const repoResult = await syncRepo({ db: database, @@ -1665,7 +1791,7 @@ export async function syncOwner(params: { owner, runId, platformIntegrationId: config.platformIntegrationId, - repoFullName, + repoFullName: selectedRepoFullName, slaConfig: config.slaConfig, notificationPolicy, notificationRecipientUserIds, @@ -1679,27 +1805,48 @@ export async function syncOwner(params: { totalResult.reauthRequired = totalResult.reauthRequired || repoResult.reauthRequired; totalResult.staleRepos.push(...repoResult.staleRepos); successfulRepos++; + completedRepos.add(selectedRepoFullName); + processedThisPass++; if (repoResult.reauthRequired) { break; } } catch (error) { - totalResult.errors++; + incompleteFailures++; await recordSecurityAgentRepositorySyncFailure(database, { owner: toSecurityAgentCommandOwner(owner), - repoFullName, + repoFullName: selectedRepoFullName, failureCode: 'SYNC_FAILED', }); - console.error(`Failed to sync ${repoFullName}`, { + console.error(`Failed to sync ${selectedRepoFullName}`, { error: error instanceof Error ? error.message : String(error), }); if (!firstError && error instanceof Error) { firstError = error; } + processedThisPass++; } } - if (successfulRepos === 0 && firstError) { + const remainingRepoCount = repositories.filter(name => !completedRepos.has(name)).length; + if (exhaustedBudget && remainingRepoCount > 0 && !totalResult.reauthRequired) { + await writeSyncRunProgress( + database, + owner, + toSyncRunProgress(runId, totalResult, [...completedRepos]) + ); + console.info('Security sync owner budget exhausted; continuation required', { + runId, + completedRepoCount: completedRepos.size, + remainingRepoCount, + durationMs: Date.now() - startTime, + }); + return { ...totalResult, exhaustedBudget: true, remainingRepoCount }; + } + + totalResult.errors += incompleteFailures; + + if (successfulRepos === 0 && firstError && !previousProgress) { throw firstError; } @@ -1763,19 +1910,19 @@ export async function syncOwner(params: { // repo-level setting, and blocking here would leave the timestamp stuck. // Missing selected repos (installation lost access) also block — the repo // was configured but silently dropped from the accessible list. - if ( + const shouldAdvanceFreshness = !repoFullName && totalResult.errors === 0 && totalResult.authInvalid === 0 && totalResult.staleRepos.length === 0 && - config.missingSelectedRepoCount === 0 - ) { + config.missingSelectedRepoCount === 0; + if (shouldAdvanceFreshness) { try { await database .update(agent_configs) .set({ runtime_state: sql`jsonb_set( - COALESCE(${agent_configs.runtime_state}, '{}'::jsonb), + COALESCE(${agent_configs.runtime_state}, '{}'::jsonb) - 'sync_run', '{last_synced_at}', to_jsonb(now()) )`, @@ -1792,6 +1939,14 @@ export async function syncOwner(params: { error: error instanceof Error ? error.message : String(error), }); } + } else if (previousProgress) { + try { + await clearSyncRunProgress(database, owner); + } catch (error) { + console.error('Failed to clear security sync run progress', { + error: error instanceof Error ? error.message : String(error), + }); + } } const syncSummary = { @@ -1819,7 +1974,7 @@ export async function syncOwner(params: { console.info('Sync cycle summary', syncSummary); } - return totalResult; + return { ...totalResult, remainingRepoCount: 0 }; } async function syncRepo(params: { diff --git a/services/security-sync/worker-configuration.d.ts b/services/security-sync/worker-configuration.d.ts index c046c641c8..adb794fdf0 100644 --- a/services/security-sync/worker-configuration.d.ts +++ b/services/security-sync/worker-configuration.d.ts @@ -10,6 +10,7 @@ declare type Message = { }; declare type MessageBatch = { + queue?: string; messages: Array>; }; @@ -44,6 +45,7 @@ declare type ExecutionContext = { declare type CloudflareEnv = { INTERNAL_API_SECRET: SecretBinding; SYNC_QUEUE: Queue; + DISMISS_QUEUE: Queue; HYPERDRIVE: Hyperdrive; GIT_TOKEN_SERVICE: GitTokenService; ENVIRONMENT: string; diff --git a/services/security-sync/wrangler.jsonc b/services/security-sync/wrangler.jsonc index dbc0350818..995b58b5f7 100644 --- a/services/security-sync/wrangler.jsonc +++ b/services/security-sync/wrangler.jsonc @@ -46,6 +46,10 @@ "binding": "SYNC_QUEUE", "queue": "security-sync-jobs", }, + { + "binding": "DISMISS_QUEUE", + "queue": "security-dismiss-jobs", + }, ], "consumers": [ { @@ -55,6 +59,13 @@ "max_concurrency": 10, "dead_letter_queue": "security-sync-jobs-dlq", }, + { + "queue": "security-dismiss-jobs", + "max_batch_size": 1, + "max_retries": 3, + "max_concurrency": 20, + "dead_letter_queue": "security-dismiss-jobs-dlq", + }, ], }, "hyperdrive": [ @@ -98,6 +109,10 @@ "binding": "SYNC_QUEUE", "queue": "security-sync-jobs-dev", }, + { + "binding": "DISMISS_QUEUE", + "queue": "security-dismiss-jobs-dev", + }, ], "consumers": [ { @@ -107,6 +122,13 @@ "max_concurrency": 10, "dead_letter_queue": "security-sync-jobs-dlq-dev", }, + { + "queue": "security-dismiss-jobs-dev", + "max_batch_size": 1, + "max_retries": 3, + "max_concurrency": 20, + "dead_letter_queue": "security-dismiss-jobs-dlq-dev", + }, ], }, "hyperdrive": [