From f06917c085b4a39af8659947c397ab01153dc607 Mon Sep 17 00:00:00 2001 From: Evgeny Shurakov Date: Wed, 30 Sep 2026 15:49:18 +0200 Subject: [PATCH] fix(security-sync): serialize owner syncs and guard run progress Concurrent scheduled and manual runs for one owner overwrote the single runtime_state.sync_run progress slot, so every continuation found a mismatched runId, restarted at repo 0, exhausted the budget, and re-enqueued forever. last_synced_at never advanced and the queue saturated, so manual syncs were delivered after the reconciler had already failed them. Add a single-flight owner lease (runtime_state.sync_lease) and fence every sync_run write, clear, and freshness update by runId and chunkIndex. A completed-run marker stops a finished run being resurrected, a no-progress cap stops a run that never advances, and the queue consumer now acks denials/refusals without enqueueing a successor. Add diagnostics: alerts_disabled status and a redacted body excerpt, per-repo timing and budget, the lease lifecycle, and a per-command stalled-command log in the auto-analysis dispatcher. --- .../src/dispatcher.test.ts | 92 +- .../security-auto-analysis/src/dispatcher.ts | 37 + services/security-auto-analysis/src/logger.ts | 10 + services/security-sync/package.json | 3 +- services/security-sync/src/index.test.ts | 338 ++++++- services/security-sync/src/index.ts | 116 ++- .../src/sync.integration.test.ts | 300 ++++++ services/security-sync/src/sync.test.ts | 902 +++++++++++++++++- services/security-sync/src/sync.ts | 769 +++++++++++++-- services/security-sync/vitest.config.ts | 1 + .../vitest.integration.config.ts | 10 + 11 files changed, 2477 insertions(+), 101 deletions(-) create mode 100644 services/security-sync/src/sync.integration.test.ts create mode 100644 services/security-sync/vitest.integration.config.ts diff --git a/services/security-auto-analysis/src/dispatcher.test.ts b/services/security-auto-analysis/src/dispatcher.test.ts index 5d34fda8254..576eee83574 100644 --- a/services/security-auto-analysis/src/dispatcher.test.ts +++ b/services/security-auto-analysis/src/dispatcher.test.ts @@ -96,11 +96,93 @@ describe('dispatchDueOwners telemetry', () => { requeued_pending_count: 1, failed_running_count: 2, }); - const serializedLogs = JSON.stringify(loggerMock.withTags.mock.calls); - expect(serializedLogs).not.toContain('sensitive-owner-id'); - expect(serializedLogs).not.toContain('sensitive-command-id'); - expect(serializedLogs).not.toContain('sensitive-attempt-id'); - expect(serializedLogs).not.toContain('postgres://'); + const serializedStageLogs = JSON.stringify( + loggerMock.withTags.mock.calls.filter( + ([tags]) => + (tags as { event_name?: string }).event_name === + 'security_auto_analysis.dispatcher_stage_succeeded' + ) + ); + expect(serializedStageLogs).not.toContain('sensitive-owner-id'); + expect(serializedStageLogs).not.toContain('sensitive-command-id'); + expect(serializedStageLogs).not.toContain('sensitive-attempt-id'); + expect(serializedStageLogs).not.toContain('postgres://'); + }); + + it('logs one command_stalled event per reconciled stale command', async () => { + vi.mocked(reconcileStaleSecurityAgentCommands).mockResolvedValueOnce({ + staleAccepted: [ + { + id: 'stale-accepted-id', + command_type: 'sync', + owned_by_organization_id: 'org-123', + accepted_at: new Date(Date.now() - 5_000).toISOString(), + result_code: 'COMMAND_STALLED', + repo_full_name: 'acme/widgets', + }, + ], + staleRunning: [ + { + id: 'stale-running-id', + command_type: 'sync', + owned_by_user_id: 'user-123', + accepted_at: new Date(Date.now() - 9_000).toISOString(), + started_at: new Date(Date.now() - 4_000).toISOString(), + last_error_redacted: 'must-not-be-logged', + }, + ], + } as never); + + await dispatchDueOwners(env(), 'dispatch-123'); + + const stallTags = loggerMock.withTags.mock.calls + .map(([tags]) => tags as Record) + .filter(tags => tags.event_name === 'security_auto_analysis.command_stalled'); + expect(stallTags).toHaveLength(2); + expect(stallTags[0]).toMatchObject({ + command_id: 'stale-accepted-id', + command_type: 'sync', + owner_type: 'org', + owner_id: 'org-123', + previous_status: 'accepted', + result_code: 'COMMAND_STALLED', + repo_full_name: 'acme/widgets', + }); + expect(Number(stallTags[0]?.age_ms)).toBeGreaterThanOrEqual(4_000); + expect(stallTags[1]).toMatchObject({ + command_id: 'stale-running-id', + owner_type: 'user', + owner_id: 'user-123', + previous_status: 'running', + result_code: null, + }); + expect(Number(stallTags[1]?.age_ms)).toBeGreaterThanOrEqual(3_000); + const serializedStallLogs = JSON.stringify(stallTags); + expect(serializedStallLogs).not.toContain('must-not-be-logged'); + expect(serializedStallLogs).not.toContain('postgres://'); + expect(stallTags[1]?.age_ms).not.toBeNull(); + }); + + it('logs command_stalled with a null age_ms when timestamps are absent', async () => { + vi.mocked(reconcileStaleSecurityAgentCommands).mockResolvedValueOnce({ + staleAccepted: [{ id: 'no-timestamp-id', command_type: 'sync' }], + staleRunning: [], + } as never); + + await dispatchDueOwners(env(), 'dispatch-123'); + + const stallTags = loggerMock.withTags.mock.calls + .map(([tags]) => tags as Record) + .filter(tags => tags.event_name === 'security_auto_analysis.command_stalled'); + expect(stallTags).toHaveLength(1); + expect(stallTags[0]).toMatchObject({ + command_id: 'no-timestamp-id', + age_ms: null, + started_at: null, + accepted_at: null, + result_code: null, + repo_full_name: null, + }); }); it.each([ diff --git a/services/security-auto-analysis/src/dispatcher.ts b/services/security-auto-analysis/src/dispatcher.ts index 8097a8c510d..b9ef28bfba5 100644 --- a/services/security-auto-analysis/src/dispatcher.ts +++ b/services/security-auto-analysis/src/dispatcher.ts @@ -2,6 +2,7 @@ import { randomUUID } from 'crypto'; import { deleteRetainedSecurityAgentCommands, reconcileStaleSecurityAgentCommands, + type SecurityAgentCommand, } from '@kilocode/db'; import { getWorkerDb } from '@kilocode/db/client'; import { discoverDueOwners, reconcileStaleAnalysisQueueRows } from './db/queries.js'; @@ -14,6 +15,40 @@ const DISPATCH_REMEDIATION_ATTEMPT_LIMIT = 100; const QUEUE_SEND_BATCH_LIMIT = 100; const DISPATCH_STAGE_SUCCEEDED_EVENT = 'security_auto_analysis.dispatcher_stage_succeeded'; const DISPATCH_FAILED_EVENT = 'security_auto_analysis.dispatcher_failed'; +const COMMAND_STALLED_EVENT = 'security_auto_analysis.command_stalled'; + +function logStalledCommands(reconciliation: { + staleAccepted: SecurityAgentCommand[]; + staleRunning: SecurityAgentCommand[]; +}): void { + for (const [previousStatus, commands] of [ + ['accepted', reconciliation.staleAccepted] as const, + ['running', reconciliation.staleRunning] as const, + ]) { + for (const command of commands) { + const timestamp = + previousStatus === 'running' + ? (command.started_at ?? command.accepted_at) + : command.accepted_at; + const parsed = timestamp ? Date.parse(timestamp) : Number.NaN; + logger + .withTags({ + event_name: COMMAND_STALLED_EVENT, + command_id: command.id, + command_type: command.command_type, + owner_type: command.owned_by_organization_id ? 'org' : 'user', + owner_id: command.owned_by_organization_id ?? command.owned_by_user_id ?? undefined, + previous_status: previousStatus, + result_code: command.result_code ?? null, + started_at: command.started_at ?? null, + accepted_at: command.accepted_at ?? null, + age_ms: Number.isFinite(parsed) ? Date.now() - parsed : null, + repo_full_name: command.repo_full_name ?? null, + }) + .info(COMMAND_STALLED_EVENT); + } + } +} async function runDispatcherStage(options: { stage: DispatcherStage; @@ -100,6 +135,8 @@ export async function dispatchDueOwners( }), }); + logStalledCommands(commandReconciliation); + const deletedCommandCount = await runDispatcherStage({ stage: 'retained_command_deletion', dispatchId, diff --git a/services/security-auto-analysis/src/logger.ts b/services/security-auto-analysis/src/logger.ts index f9eafcae8de..195f4891a3c 100644 --- a/services/security-auto-analysis/src/logger.ts +++ b/services/security-auto-analysis/src/logger.ts @@ -31,6 +31,16 @@ export type SecurityAutoAnalysisLogTags = { enqueued_owner_message_count?: number; discovered_remediation_attempt_count?: number; enqueued_remediation_message_count?: number; + command_id?: string; + command_type?: string; + owner_type?: 'org' | 'user'; + owner_id?: string; + previous_status?: string; + result_code?: string | null; + started_at?: string | null; + accepted_at?: string | null; + age_ms?: number | null; + repo_full_name?: string | null; }; export function sanitizedExceptionName(error: unknown): string { diff --git a/services/security-sync/package.json b/services/security-sync/package.json index 2c641dc491e..a4b45b3fc86 100644 --- a/services/security-sync/package.json +++ b/services/security-sync/package.json @@ -9,7 +9,8 @@ "typecheck": "tsgo --noEmit", "lint": "pnpm -w exec oxlint --config .oxlintrc.json services/security-sync/src", "types": "wrangler types --env-interface CloudflareEnv worker-configuration.d.ts", - "test": "vitest run" + "test": "vitest run", + "test:integration": "vitest run --config vitest.integration.config.ts" }, "dependencies": { "@kilocode/db": "workspace:*", diff --git a/services/security-sync/src/index.test.ts b/services/security-sync/src/index.test.ts index 3cbe0513dcf..94faf4a2b00 100644 --- a/services/security-sync/src/index.test.ts +++ b/services/security-sync/src/index.test.ts @@ -12,7 +12,7 @@ import { settleOperation } from '@kilocode/db/operation-ledger'; import worker, { collectScheduledSyncOwners, type SecuritySyncQueueMessage } from './index.js'; import { processSecurityFindingDismissal } from './dismiss.js'; import { runSecurityNotificationSweep } from './notifications/sweep.js'; -import { syncOwner } from './sync.js'; +import { syncOwner, releaseOwnerSyncLease } from './sync.js'; import type * as SyncModule from './sync.js'; vi.mock('@kilocode/db', async importOriginal => { @@ -36,7 +36,7 @@ vi.mock('./dismiss.js', () => ({ processSecurityFindingDismissal: vi.fn() })); vi.mock('./notifications/sweep.js', () => ({ runSecurityNotificationSweep: vi.fn() })); vi.mock('./sync.js', async importOriginal => { const actual = await importOriginal(); - return { ...actual, syncOwner: vi.fn() }; + return { ...actual, syncOwner: vi.fn(), releaseOwnerSyncLease: vi.fn() }; }); beforeEach(() => { @@ -55,6 +55,7 @@ beforeEach(() => { } as never); vi.mocked(runSecurityNotificationSweep).mockResolvedValue({} as never); vi.mocked(settleOperation).mockResolvedValue({ settled: true, row: {} } as never); + vi.mocked(releaseOwnerSyncLease).mockResolvedValue({ released: true }); }); /** Worker database stub; every repository call it would serve is mocked. */ @@ -644,6 +645,338 @@ describe('manual sync dispatch', () => { }); }); +describe('manual sync lease dispositions', () => { + function manualSyncBody() { + return { + schemaVersion: 1, + commandId: 'dddddddd-dddd-4ddd-8ddd-dddddddddddd', + runId: 'cccccccc-cccc-4ccc-8ccc-cccccccccccc', + messageId: 'manual-sync-message', + trigger: 'manual', + owner: { organizationId: 'aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa' }, + ownerKey: 'org:aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa', + chunkIndex: 1, + chunkCount: 2, + dispatchedAt: '2026-05-18T08:30:00.000Z', + actor: { id: 'user-123', email: 'owner@example.com' }, + }; + } + + function leaseEnv(sendBatch: ReturnType) { + return { + HYPERDRIVE: { connectionString: 'postgres://worker' }, + GIT_TOKEN_SERVICE: {}, + SYNC_QUEUE: { sendBatch }, + } as unknown as CloudflareEnv; + } + + it('acks a stale chunk delivery without failing, settling, or enqueueing', async () => { + const info = vi.spyOn(console, 'info').mockImplementation(() => undefined); + vi.mocked(getWorkerDb).mockReturnValue(workerDbStub()); + vi.mocked(syncOwner).mockResolvedValue({ + synced: 0, + errors: 0, + staleRepos: [], + staleChunk: true, + exhaustedBudget: false, + } as never); + const ack = vi.fn(); + const retry = vi.fn(); + const sendBatch = vi.fn(); + + await worker.queue( + { messages: [{ attempts: 1, body: manualSyncBody(), ack, retry }] } as never, + leaseEnv(sendBatch) + ); + + expect(syncOwner).toHaveBeenCalledWith(expect.objectContaining({ chunkIndex: 1 })); + expect(ack).toHaveBeenCalledTimes(1); + expect(retry).not.toHaveBeenCalled(); + expect(sendBatch).not.toHaveBeenCalled(); + expect(transitionSecurityAgentCommandWithCurrentState).toHaveBeenCalledTimes(1); + expect(settleOperation).not.toHaveBeenCalled(); + expect( + info.mock.calls.some(([message]) => message === 'Security sync completed for owner') + ).toBe(false); + info.mockRestore(); + }); + + it.each([ + { disposition: { staleChunk: true }, expectedLog: 'Security sync stale chunk acknowledged' }, + { + disposition: { claimDenied: true, holderRunId: 'other-run' }, + expectedLog: 'Security sync command terminated without continuation', + }, + ])( + 'acks a scheduled delivery with $expectedLog and no command transition', + async ({ disposition, expectedLog }) => { + const info = vi.spyOn(console, 'info').mockImplementation(() => undefined); + vi.mocked(getWorkerDb).mockReturnValue(workerDbStub()); + vi.mocked(syncOwner).mockResolvedValue({ + synced: 0, + errors: 0, + staleRepos: [], + exhaustedBudget: false, + ...disposition, + } as never); + const ack = vi.fn(); + const retry = vi.fn(); + const sendBatch = vi.fn(); + + await worker.queue( + { + 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-05-18T08:30:00.000Z', + }, + ack, + retry, + }, + ], + } as never, + leaseEnv(sendBatch) + ); + + expect(info.mock.calls.some(([message]) => message === expectedLog)).toBe(true); + expect( + info.mock.calls.some(([message]) => message === 'Security sync completed for owner') + ).toBe(false); + expect(transitionSecurityAgentCommandWithCurrentState).not.toHaveBeenCalled(); + expect(settleOperation).not.toHaveBeenCalled(); + expect(sendBatch).not.toHaveBeenCalled(); + expect(ack).toHaveBeenCalledTimes(1); + expect(retry).not.toHaveBeenCalled(); + info.mockRestore(); + } + ); + + it.each([ + { + disposition: { claimDenied: true, holderRunId: 'other-run' }, + resultCode: 'SYNC_ALREADY_RUNNING', + message: 'A sync is already in progress for this account. Wait for it to finish, then retry.', + }, + { + disposition: { checkpointRejected: true }, + resultCode: 'SYNC_SUPERSEDED', + message: 'This sync lost its place and cannot continue. Retry.', + }, + { + disposition: { noProgress: true }, + resultCode: 'SYNC_NO_PROGRESS', + message: 'Sync stopped because two consecutive passes made no progress. Retry.', + }, + ])( + 'fails the manual command with $resultCode without enqueueing', + async ({ disposition, resultCode, message }) => { + vi.mocked(getWorkerDb).mockReturnValue(workerDbStub()); + vi.mocked(syncOwner).mockResolvedValue({ + synced: 0, + errors: 0, + staleRepos: [], + exhaustedBudget: false, + ...disposition, + } as never); + const ack = vi.fn(); + const retry = vi.fn(); + const sendBatch = vi.fn(); + + await worker.queue( + { messages: [{ attempts: 1, body: manualSyncBody(), ack, retry }] } as never, + leaseEnv(sendBatch) + ); + + expect(transitionSecurityAgentCommandWithCurrentState).toHaveBeenNthCalledWith( + 2, + expect.anything(), + expect.objectContaining({ + status: 'failed', + resultCode, + lastErrorRedacted: message, + }) + ); + expect(sendBatch).not.toHaveBeenCalled(); + expect(ack).toHaveBeenCalledTimes(1); + expect(retry).not.toHaveBeenCalled(); + } + ); + + it('settles the ledger when a lease disposition fails a manual command', async () => { + vi.mocked(transitionSecurityAgentCommandWithCurrentState) + .mockResolvedValueOnce({ transitioned: true, command: {} } as never) + .mockResolvedValueOnce({ + transitioned: true, + command: { + id: 'dddddddd-dddd-4ddd-8ddd-dddddddddddd', + operation_key: 'retry-safe-key-123', + status: 'failed', + result_code: 'SYNC_ALREADY_RUNNING', + }, + } as never); + vi.mocked(getWorkerDb).mockReturnValue(ledgerDb([{ id: 'ledger-row-id' }])); + vi.mocked(syncOwner).mockResolvedValue({ + synced: 0, + errors: 0, + staleRepos: [], + claimDenied: true, + holderRunId: 'other-run', + exhaustedBudget: false, + } as never); + const ack = vi.fn(); + const retry = vi.fn(); + + await worker.queue( + { messages: [{ attempts: 1, body: manualSyncBody(), ack, retry }] } as never, + leaseEnv(vi.fn()) + ); + + expect(settleOperation).toHaveBeenCalledWith( + expect.anything(), + expect.objectContaining({ + rowId: 'ledger-row-id', + status: 'failed', + outcomeCode: 'SYNC_ALREADY_RUNNING', + }) + ); + expect(ack).toHaveBeenCalledTimes(1); + expect(retry).not.toHaveBeenCalled(); + }); + + it('releases the owner lease on the final delivery attempt before retrying', async () => { + const info = vi.spyOn(console, 'info').mockImplementation(() => undefined); + vi.mocked(getWorkerDb).mockReturnValue(workerDbStub()); + vi.mocked(syncOwner).mockRejectedValue(new Error('sync failed')); + const ack = vi.fn(); + const retry = vi.fn(); + + await worker.queue( + { messages: [{ attempts: 4, body: manualSyncBody(), ack, retry }] } as never, + leaseEnv(vi.fn()) + ); + + expect(releaseOwnerSyncLease).toHaveBeenCalledWith( + expect.anything(), + { organizationId: 'aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa' }, + 'cccccccc-cccc-4ccc-8ccc-cccccccccccc', + 1 + ); + expect(vi.mocked(releaseOwnerSyncLease).mock.invocationCallOrder[0]).toBeLessThan( + retry.mock.invocationCallOrder[0] ?? Infinity + ); + expect(retry).toHaveBeenCalledTimes(1); + expect( + info.mock.calls.some( + ([message, meta]) => + message === 'Security sync lease released' && + (meta as { reason?: string } | undefined)?.reason === 'retries_exhausted' + ) + ).toBe(true); + info.mockRestore(); + }); + + it('records exhaustion and still retries when the final-attempt lease release throws', async () => { + const errorLog = vi.spyOn(console, 'error').mockImplementation(() => undefined); + const workerDb = workerDbStub(); + vi.mocked(getWorkerDb).mockReturnValue(workerDb); + vi.mocked(syncOwner).mockRejectedValue(new Error('sync failed')); + vi.mocked(releaseOwnerSyncLease).mockRejectedValue(new Error('release unavailable')); + const ack = vi.fn(); + const retry = vi.fn(); + + await worker.queue( + { messages: [{ attempts: 4, body: manualSyncBody(), ack, retry }] } as never, + leaseEnv(vi.fn()) + ); + + expect(releaseOwnerSyncLease).toHaveBeenCalledWith( + expect.anything(), + { organizationId: 'aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa' }, + 'cccccccc-cccc-4ccc-8ccc-cccccccccccc', + 1 + ); + expect( + errorLog.mock.calls.some( + ([message, meta]) => + message === 'Security sync lease release rejected' && + (meta as { reason?: string; error_type?: string } | undefined)?.reason === + 'retries_exhausted' && + (meta as { error_type?: string } | undefined)?.error_type === 'Error' + ) + ).toBe(true); + expect(markSecurityAgentCommandRetriesExhausted).toHaveBeenCalledWith( + workerDb, + 'dddddddd-dddd-4ddd-8ddd-dddddddddddd' + ); + expect( + vi.mocked(markSecurityAgentCommandRetriesExhausted).mock.invocationCallOrder[0] + ).toBeLessThan(retry.mock.invocationCallOrder[0] ?? Infinity); + expect(retry).toHaveBeenCalledTimes(1); + expect(ack).not.toHaveBeenCalled(); + errorLog.mockRestore(); + }); + + it('does not release the lease below the final delivery attempt', async () => { + vi.mocked(getWorkerDb).mockReturnValue(workerDbStub()); + vi.mocked(syncOwner).mockRejectedValue(new Error('sync failed')); + const ack = vi.fn(); + const retry = vi.fn(); + + await worker.queue( + { messages: [{ attempts: 1, body: manualSyncBody(), ack, retry }] } as never, + leaseEnv(vi.fn()) + ); + + expect(releaseOwnerSyncLease).not.toHaveBeenCalled(); + expect(retry).toHaveBeenCalledTimes(1); + }); + + it('releases the owner lease before settling an already-terminal delivery', async () => { + const info = vi.spyOn(console, 'info').mockImplementation(() => undefined); + vi.mocked(transitionSecurityAgentCommandWithCurrentState).mockResolvedValueOnce({ + transitioned: false, + command: { status: 'succeeded', result_code: 'SYNC_COMPLETED' }, + } as never); + vi.mocked(getWorkerDb).mockReturnValue(ledgerDb([{ id: 'ledger-row-id' }])); + const ack = vi.fn(); + const retry = vi.fn(); + + await worker.queue( + { messages: [{ attempts: 2, body: manualSyncBody(), ack, retry }] } as never, + leaseEnv(vi.fn()) + ); + + expect(releaseOwnerSyncLease).toHaveBeenCalledWith( + expect.anything(), + { organizationId: 'aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa' }, + 'cccccccc-cccc-4ccc-8ccc-cccccccccccc', + 1 + ); + expect(vi.mocked(releaseOwnerSyncLease).mock.invocationCallOrder[0]).toBeLessThan( + ack.mock.invocationCallOrder[0] ?? Infinity + ); + expect( + info.mock.calls.some( + ([message, meta]) => + message === 'Security sync lease released' && + (meta as { reason?: string } | undefined)?.reason === 'command_terminal' + ) + ).toBe(true); + expect(ack).toHaveBeenCalledTimes(1); + expect(retry).not.toHaveBeenCalled(); + info.mockRestore(); + }); +}); + describe('manual dismissal dispatch', () => { it('accepts an authenticated dismissal command and enqueues actor-aware Worker processing', async () => { const queuedBatches: MessageSendRequest[][] = []; @@ -901,6 +1234,7 @@ describe('manual dismissal dispatch', () => { expect(vi.mocked(settleOperation).mock.invocationCallOrder[0]).toBeLessThan( retry.mock.invocationCallOrder[0] ?? Infinity ); + expect(releaseOwnerSyncLease).not.toHaveBeenCalled(); expect(ack).not.toHaveBeenCalled(); expect(retry).toHaveBeenCalledTimes(1); }); diff --git a/services/security-sync/src/index.ts b/services/security-sync/src/index.ts index ccab56b5c80..8bbffb16afd 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 { SECURITY_SYNC_OWNER_BUDGET_MS, syncOwner } from './sync'; +import { SECURITY_SYNC_OWNER_BUDGET_MS, releaseOwnerSyncLease, syncOwner } from './sync'; import { processSecurityFindingDismissal } from './dismiss'; import { runSecurityNotificationSweep } from './notifications/sweep'; @@ -185,6 +185,14 @@ function getEnvironment(env: CloudflareEnv): string | undefined { const QUEUE_SEND_BATCH_LIMIT = 100; const SECURITY_SYNC_COMMAND_MAX_ATTEMPTS = 4; +/** User-facing redacted messages for lease-driven sync failures; no repo names or tokens. */ +const SECURITY_SYNC_RESULT_MESSAGES: Record = { + SYNC_ALREADY_RUNNING: + 'A sync is already in progress for this account. Wait for it to finish, then retry.', + SYNC_SUPERSEDED: 'This sync lost its place and cannot continue. Retry.', + SYNC_NO_PROGRESS: 'Sync stopped because two consecutive passes made no progress. Retry.', +}; + function createOwnerKey(owner: SecuritySyncMessage['owner']): string { if (owner.organizationId) return `org:${owner.organizationId}`; if (owner.userId) return `user:${owner.userId}`; @@ -620,6 +628,20 @@ async function processSecuritySyncMessage( result_code: running.command?.result_code, attempts: message.attempts, }); + const released = await releaseOwnerSyncLease(db, owner, body.runId, body.chunkIndex); + if (released.released) { + console.info('Security sync lease released', { + runId: body.runId, + chunkIndex: body.chunkIndex, + reason: 'command_terminal', + }); + } else { + console.info('Security sync lease release rejected', { + runId: body.runId, + holderRunId: null, + reason: 'command_terminal', + }); + } if (!body.actor) throw new Error('Manual security sync command has no actor'); await settleSecurityLedgerForTerminalCommand(db, running.command, { actor: body.actor, @@ -639,6 +661,7 @@ async function processSecuritySyncMessage( trigger: body.trigger, actor: body.actor, repoFullName: body.repoFullName, + chunkIndex: body.chunkIndex, notificationMaterializationEnabled: isStrictTrueRolloutFlag( env.SECURITY_NOTIFICATION_MATERIALIZATION_ENABLED, 'SECURITY_NOTIFICATION_MATERIALIZATION_ENABLED' @@ -646,6 +669,59 @@ async function processSecuritySyncMessage( budgetMs: body.repoFullName ? undefined : SECURITY_SYNC_OWNER_BUDGET_MS, }); + if (result.staleChunk) { + console.info('Security sync stale chunk acknowledged', { + command_id: body.commandId, + command_type: body.commandId ? 'sync' : undefined, + owner_type: body.owner.organizationId ? 'org' : 'user', + runId: body.runId, + ownerKey: body.ownerKey, + chunkIndex: body.chunkIndex, + attempts: message.attempts, + }); + message.ack(); + return; + } + + const leaseFailure = result.claimDenied + ? 'SYNC_ALREADY_RUNNING' + : result.checkpointRejected + ? 'SYNC_SUPERSEDED' + : result.noProgress + ? 'SYNC_NO_PROGRESS' + : null; + if (leaseFailure) { + if (body.commandId) { + const terminalTransition = await transitionSecurityAgentCommandWithCurrentState(db, { + commandId: body.commandId, + fromStatuses: ['running'], + status: 'failed', + resultCode: leaseFailure, + lastErrorRedacted: SECURITY_SYNC_RESULT_MESSAGES[leaseFailure], + }); + requireSecurityAgentCommandTransitionOrTerminal(terminalTransition, 'terminal'); + if (!body.actor) throw new Error('Manual security sync command has no actor'); + await settleSecurityLedgerForTerminalCommand(db, terminalTransition.command, { + actor: body.actor, + intent: 'manual_sync', + dispatchedAt: body.dispatchedAt, + counts: { repo_count: result.synced, error_count: result.errors }, + }); + } + console.info('Security sync command terminated without continuation', { + command_id: body.commandId, + command_type: body.commandId ? 'sync' : undefined, + owner_type: body.owner.organizationId ? 'org' : 'user', + result_code: leaseFailure, + runId: body.runId, + ownerKey: body.ownerKey, + chunkIndex: body.chunkIndex, + attempts: message.attempts, + }); + message.ack(); + return; + } + if (result.exhaustedBudget) { const nextChunkIndex = body.chunkIndex + 1; await env.SYNC_QUEUE.sendBatch([ @@ -922,6 +998,44 @@ export default { await processSecuritySyncMessage(message, env); } catch (error) { const correlation = commandCorrelation(message.body); + const syncBody = SecuritySyncMessageSchema.safeParse(message.body); + if (message.attempts >= SECURITY_SYNC_COMMAND_MAX_ATTEMPTS && syncBody.success) { + const syncLeaseOwner = resolveOwner(syncBody.data.owner); + if (syncLeaseOwner) { + try { + const db = getWorkerDb(env.HYPERDRIVE.connectionString, { + statement_timeout: 30_000, + }); + const released = await releaseOwnerSyncLease( + db, + syncLeaseOwner, + syncBody.data.runId, + syncBody.data.chunkIndex + ); + if (released.released) { + console.info('Security sync lease released', { + runId: syncBody.data.runId, + chunkIndex: syncBody.data.chunkIndex, + reason: 'retries_exhausted', + }); + } else { + console.info('Security sync lease release rejected', { + runId: syncBody.data.runId, + holderRunId: null, + reason: 'retries_exhausted', + }); + } + } catch (releaseError) { + console.error('Security sync lease release rejected', { + runId: syncBody.data.runId, + chunkIndex: syncBody.data.chunkIndex, + holderRunId: null, + reason: 'retries_exhausted', + error_type: releaseError instanceof Error ? releaseError.name : 'UnknownError', + }); + } + } + } let exhaustionOutcome: SecurityAgentCommandTransitionOutcome | undefined; if (correlation.commandId && message.attempts >= SECURITY_SYNC_COMMAND_MAX_ATTEMPTS) { try { diff --git a/services/security-sync/src/sync.integration.test.ts b/services/security-sync/src/sync.integration.test.ts new file mode 100644 index 00000000000..c9cbfbb6dce --- /dev/null +++ b/services/security-sync/src/sync.integration.test.ts @@ -0,0 +1,300 @@ +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from 'vitest'; +import { randomUUID } from 'crypto'; +import { createDrizzleClient } from '@kilocode/db/client'; +import { agent_configs, kilocode_users } from '@kilocode/db/schema'; +import { eq } from 'drizzle-orm'; +import { + advanceOwnerSyncFreshness, + claimOwnerSyncLease, + clearSyncRunProgress, + releaseOwnerSyncLease, + syncOwner, + writeSyncRunProgress, + SECURITY_SYNC_LEASE_TTL_MS, +} from './sync.js'; + +const connectionString = + process.env.POSTGRES_URL ?? 'postgres://postgres:postgres@localhost:5432/postgres'; +const testUserId = `security-sync-lease-${randomUUID()}`; +const agentConfigId = randomUUID(); +const owner = { userId: testUserId }; + +let client: ReturnType; + +function progress(runId: string, completedRepos: string[], chunkIndex: number) { + return { + runId, + completedRepos, + staleRepos: [], + authInvalidRepos: [], + synced: 0, + errors: 0, + skipped: 0, + authInvalid: 0, + reauthRequired: false, + noProgressChunks: 0, + chunkIndex, + }; +} + +async function readRuntimeState(): Promise> { + const [row] = await client.db + .select({ runtime_state: agent_configs.runtime_state }) + .from(agent_configs) + .where(eq(agent_configs.id, agentConfigId)); + return (row?.runtime_state ?? {}) as Record; +} + +async function resetRuntimeState(state: Record = {}): Promise { + await client.db + .update(agent_configs) + .set({ runtime_state: state }) + .where(eq(agent_configs.id, agentConfigId)); +} + +describe('security sync owner lease in PostgreSQL', () => { + beforeAll(async () => { + client = createDrizzleClient({ connectionString, ssl: false }); + await client.db.insert(kilocode_users).values({ + id: testUserId, + google_user_email: `${testUserId}@example.com`, + google_user_name: 'Security Sync Lease DB Test', + google_user_image_url: 'https://example.com/avatar.png', + stripe_customer_id: `cus_${randomUUID()}`, + }); + await client.db.insert(agent_configs).values({ + id: agentConfigId, + owned_by_user_id: testUserId, + agent_type: 'security_scan', + platform: 'github', + config: {}, + created_by: 'test', + }); + }); + + afterEach(async () => { + vi.unstubAllGlobals(); + vi.restoreAllMocks(); + await resetRuntimeState({}); + }); + + afterAll(async () => { + await client.db.delete(agent_configs).where(eq(agent_configs.id, agentConfigId)); + await client.db.delete(kilocode_users).where(eq(kilocode_users.id, testUserId)); + await client.pool.end(); + }); + + it('denies a claim while another run holds the lease', async () => { + const before = Date.now(); + const claimed = await claimOwnerSyncLease(client.db as never, owner, 'holder-run', 0); + expect(claimed).not.toBeNull(); + expect(claimed?.runtimeState).toMatchObject({ + sync_lease: expect.objectContaining({ runId: 'holder-run', chunkIndex: 0 }), + }); + await expect( + claimOwnerSyncLease(client.db as never, owner, 'other-run', 0) + ).resolves.toBeNull(); + + const state = await readRuntimeState(); + const lease = state.sync_lease as { runId: string; chunkIndex: number; expiresAt: string }; + expect(lease).toMatchObject({ runId: 'holder-run', chunkIndex: 0 }); + expect(state.sync_run).toMatchObject({ runId: 'holder-run', completedRepos: [] }); + const expiresAt = new Date(lease.expiresAt).getTime(); + expect(expiresAt).toBeGreaterThan(before + SECURITY_SYNC_LEASE_TTL_MS - 60_000); + expect(expiresAt).toBeLessThan(Date.now() + SECURITY_SYNC_LEASE_TTL_MS + 60_000); + }); + + it('lets exactly one of two concurrent connections acquire the lease', async () => { + const clientB = createDrizzleClient({ connectionString, ssl: false }); + try { + const [a, b] = await Promise.all([ + claimOwnerSyncLease(client.db as never, owner, 'run-a', 0), + claimOwnerSyncLease(clientB.db as never, owner, 'run-b', 0), + ]); + expect([a, b].filter(result => result !== null)).toHaveLength(1); + + const winner = a ? 'run-a' : 'run-b'; + const state = await readRuntimeState(); + expect(state.sync_lease).toMatchObject({ runId: winner, chunkIndex: 0 }); + } finally { + await clientB.pool.end(); + } + }); + + it('takes over an expired lease with a fresh skeleton and no cursor copy', async () => { + await resetRuntimeState({ + sync_lease: { runId: 'old-run', chunkIndex: 0, expiresAt: '2000-01-01T00:00:00.000Z' }, + sync_run: progress('old-run', ['old-repo'], 0), + }); + + const claimed = await claimOwnerSyncLease(client.db as never, owner, 'new-run', 0); + expect(claimed).not.toBeNull(); + + const state = await readRuntimeState(); + expect(state.sync_run).toMatchObject({ runId: 'new-run', completedRepos: [] }); + expect(JSON.stringify(state.sync_run)).not.toContain('old-repo'); + expect(state.sync_run).not.toHaveProperty('chunkIndex'); + expect(state.sync_lease).toMatchObject({ runId: 'new-run', chunkIndex: 0 }); + + // The displaced run cannot reclaim the lease, even at a higher chunk index. + await expect(claimOwnerSyncLease(client.db as never, owner, 'old-run', 5)).resolves.toBeNull(); + }); + + it('rejects foreign write, clear, and freshness mutations', async () => { + await claimOwnerSyncLease(client.db as never, owner, 'run-f', 0); + + await expect( + writeSyncRunProgress(client.db as never, owner, progress('other-run', ['acme/a'], 0)) + ).resolves.toEqual({ written: false }); + await expect(clearSyncRunProgress(client.db as never, owner, 'other-run', 0)).resolves.toEqual({ + cleared: false, + }); + await expect( + advanceOwnerSyncFreshness(client.db as never, owner, 'other-run', 0) + ).resolves.toEqual({ advanced: false }); + + const state = await readRuntimeState(); + expect(state.sync_lease).toMatchObject({ runId: 'run-f', chunkIndex: 0 }); + expect(state.sync_run).toMatchObject({ runId: 'run-f', completedRepos: [] }); + expect(state.last_synced_at).toBeUndefined(); + expect(state.last_completed_run_id).toBeUndefined(); + }); + + it('rejects older same-run clear and freshness and leaves state unchanged', async () => { + await claimOwnerSyncLease(client.db as never, owner, 'run-older', 1); + await expect( + writeSyncRunProgress(client.db as never, owner, progress('run-older', ['acme/a'], 1)) + ).resolves.toEqual({ written: true }); + + await expect(clearSyncRunProgress(client.db as never, owner, 'run-older', 0)).resolves.toEqual({ + cleared: false, + }); + await expect( + advanceOwnerSyncFreshness(client.db as never, owner, 'run-older', 0) + ).resolves.toEqual({ advanced: false }); + + const state = await readRuntimeState(); + expect(state.sync_run).toMatchObject({ + runId: 'run-older', + completedRepos: ['acme/a'], + chunkIndex: 1, + }); + expect(state.sync_lease).toMatchObject({ runId: 'run-older', chunkIndex: 1 }); + expect(state.last_synced_at).toBeUndefined(); + expect(state.last_completed_run_id).toBeUndefined(); + }); + + it('fences chunk writes: older rejected, equal accepted, union preserved', async () => { + await claimOwnerSyncLease(client.db as never, owner, 'run-c', 1); + + await expect( + writeSyncRunProgress(client.db as never, owner, { + ...progress('run-c', ['acme/a'], 1), + synced: 3, + errors: 1, + }) + ).resolves.toEqual({ written: true }); + await expect( + writeSyncRunProgress(client.db as never, owner, { + ...progress('run-c', ['acme/b'], 1), + synced: 7, + errors: 2, + }) + ).resolves.toEqual({ written: true }); + + let state = await readRuntimeState(); + expect(state.sync_run).toMatchObject({ + completedRepos: ['acme/a', 'acme/b'], + synced: 7, + errors: 2, + chunkIndex: 1, + }); + + await expect( + writeSyncRunProgress(client.db as never, owner, progress('run-c', ['acme/c'], 0)) + ).resolves.toEqual({ written: false }); + + state = await readRuntimeState(); + expect(state.sync_run).toMatchObject({ + completedRepos: ['acme/a', 'acme/b'], + chunkIndex: 1, + }); + expect(state.sync_lease).toMatchObject({ runId: 'run-c', chunkIndex: 1 }); + }); + + it('guards release by run id and chunk fence', async () => { + await claimOwnerSyncLease(client.db as never, owner, 'run-r', 2); + + await expect(releaseOwnerSyncLease(client.db as never, owner, 'other-run', 2)).resolves.toEqual( + { released: false } + ); + await expect(releaseOwnerSyncLease(client.db as never, owner, 'run-r', 1)).resolves.toEqual({ + released: false, + }); + await expect(releaseOwnerSyncLease(client.db as never, owner, 'run-r', 2)).resolves.toEqual({ + released: true, + }); + expect((await readRuntimeState()).sync_lease).toBeUndefined(); + }); + + it('completion sets last_completed_run_id and removes the cursor', async () => { + await claimOwnerSyncLease(client.db as never, owner, 'run-done', 0); + + await expect( + advanceOwnerSyncFreshness(client.db as never, owner, 'run-done', 0) + ).resolves.toEqual({ advanced: true }); + + const state = await readRuntimeState(); + expect(state.last_completed_run_id).toBe('run-done'); + expect(state.sync_run).toBeUndefined(); + expect(typeof state.last_synced_at).toBe('string'); + + await expect(releaseOwnerSyncLease(client.db as never, owner, 'run-done', 0)).resolves.toEqual({ + released: true, + }); + + const getToken = vi.fn(async () => { + throw new Error('getToken must not be called for a completed redelivery'); + }); + const redelivered = await syncOwner({ + db: client.db as never, + gitTokenService: { getToken } as never, + owner, + runId: 'run-done', + chunkIndex: 0, + }); + + expect(redelivered).not.toHaveProperty('claimDenied'); + expect(redelivered).not.toHaveProperty('staleChunk'); + expect(redelivered).not.toHaveProperty('checkpointRejected'); + expect(redelivered.exhaustedBudget).toBe(false); + expect(getToken).not.toHaveBeenCalled(); + + const after = await readRuntimeState(); + expect(after.sync_run).toBeUndefined(); + expect(after.sync_lease).toBeUndefined(); + }); + + it('returns staleChunk for an older same-run delivery without minting a token', async () => { + await claimOwnerSyncLease(client.db as never, owner, 'run-stale', 2); + const getToken = vi.fn(async () => 'github-token'); + const fetchStub = vi.fn(); + vi.stubGlobal('fetch', fetchStub); + + const result = await syncOwner({ + db: client.db as never, + gitTokenService: { getToken } as never, + owner, + runId: 'run-stale', + chunkIndex: 1, + }); + + expect(result).toMatchObject({ staleChunk: true, exhaustedBudget: false }); + expect(result).not.toHaveProperty('claimDenied'); + expect(getToken).not.toHaveBeenCalled(); + expect(fetchStub).not.toHaveBeenCalled(); + + const state = await readRuntimeState(); + expect(state.sync_lease).toMatchObject({ runId: 'run-stale', chunkIndex: 2 }); + }); +}); diff --git a/services/security-sync/src/sync.test.ts b/services/security-sync/src/sync.test.ts index e5912473484..c2a804f0f63 100644 --- a/services/security-sync/src/sync.test.ts +++ b/services/security-sync/src/sync.test.ts @@ -16,16 +16,39 @@ type FakeDbOptions = { authInvalidAt?: string | null; repositories?: string[]; runtimeState?: Record; + /** Default true; false makes the claim `returning()` yield zero rows. */ + claimAcquired?: boolean; + /** Post-image of a successful claim; default `runtimeState`. */ + claimRuntimeState?: Record; + /** Default true; false makes the progress-write `returning()` yield zero rows. */ + writeAccepted?: boolean; + /** Default true; false makes clear/freshness/release `returning()` yield zero rows. */ + terminalAccepted?: boolean; + /** Runtime state returned by the deny/conflict read; default null (no row). */ + conflictRuntimeState?: Record | null; + /** Default false; true makes getOwnerConfig's first select return no row. */ + configMissing?: boolean; }; function createFakeDb(options: FakeDbOptions = {}) { const repositories = options.repositories ?? ['acme/widgets']; const sets: Array> = []; + const wheres: string[] = []; let selectCount = 0; + let updateCount = 0; + let conflictReadArmed = false; + const selection = { limit: async () => { + if (conflictReadArmed) { + conflictReadArmed = false; + return options.conflictRuntimeState + ? [{ runtime_state: options.conflictRuntimeState }] + : []; + } selectCount++; if (selectCount === 1) { + if (options.configMissing) return []; return [ { id: 'agent-config', @@ -54,18 +77,58 @@ function createFakeDb(options: FakeDbOptions = {}) { orderBy: () => selection, }; + const updateReturning = ( + values: Record, + isClaim: boolean + ): Array> => { + if (isClaim) { + if (options.claimAcquired === false) { + conflictReadArmed = true; + return []; + } + return [{ runtimeState: options.claimRuntimeState ?? options.runtimeState ?? {} }]; + } + if (isProgressWrite(values)) { + if (options.writeAccepted === false) { + conflictReadArmed = true; + return []; + } + return [{}]; + } + if (options.terminalAccepted === false) { + conflictReadArmed = true; + return []; + } + return [{}]; + }; + const db = { select: () => ({ from: () => ({ where: () => selection, }), }), - update: () => ({ - set: (values: Record) => { - sets.push(values); - return { where: async () => undefined }; - }, - }), + update: () => { + updateCount++; + const isClaim = updateCount === 1; + return { + set: (values: Record) => { + sets.push(values); + return { + where: (condition: unknown) => { + wheres.push(sqlToText(condition)); + return { + returning: async () => updateReturning(values, isClaim), + then: ( + resolve: (value: undefined) => unknown, + reject?: (error: unknown) => unknown + ) => Promise.resolve(undefined).then(resolve, reject), + }; + }, + }; + }, + }; + }, insert: () => ({ values: () => ({ onConflictDoUpdate: async () => undefined, @@ -75,18 +138,56 @@ function createFakeDb(options: FakeDbOptions = {}) { transaction: async (callback: (transaction: unknown) => Promise) => callback(db), }; - return { db, sets }; + return { db, sets, wheres, selectCount: () => selectCount }; +} + +function isProgressWrite(values: Record): boolean { + return ( + values.runtime_state != null && + typeof values.runtime_state === 'object' && + sqlToText(values.runtime_state).includes('jsonb_agg') + ); +} + +/** Flattens a Drizzle SQL object into text, including bound parameter values. */ +function sqlToText(node: unknown): string { + const parts: string[] = []; + collectSqlParts(node, parts); + return parts.join(''); +} + +function collectSqlParts(node: unknown, parts: string[]): void { + if (node == null) return; + if (typeof node === 'string' || typeof node === 'number' || typeof node === 'boolean') { + parts.push(String(node)); + return; + } + if (Array.isArray(node)) { + for (const item of node) collectSqlParts(item, parts); + return; + } + if (typeof node === 'object') { + const record = node as { queryChunks?: unknown; value?: unknown; name?: unknown }; + if (Array.isArray(record.queryChunks)) { + collectSqlParts(record.queryChunks, parts); + return; + } + if (Array.isArray(record.value)) { + collectSqlParts(record.value, parts); + return; + } + if ('value' in record && (record.value == null || typeof record.value !== 'object')) { + parts.push(String(record.value)); + return; + } + if (typeof record.name === 'string') { + parts.push(record.name); + } + } } 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(''); + return sqlToText(entry.runtime_state); } function createGitTokenService() { @@ -180,7 +281,7 @@ describe('Worker GitHub auth-invalid sync', () => { }); it('persists the first GitHub 401 and stops syncing remaining repos', async () => { - const { db, sets } = createFakeDb({ repositories: ['acme/widgets', 'acme/api'] }); + const { db, sets, wheres } = createFakeDb({ repositories: ['acme/widgets', 'acme/api'] }); const gitTokenService = createGitTokenService(); const fetchStub = stubFetch(new Response('Bad credentials', { status: 401 })); @@ -208,7 +309,14 @@ describe('Worker GitHub auth-invalid sync', () => { expect(sets).toContainEqual( expect.objectContaining({ auth_invalid_reason: 'github_dependabot_401' }) ); - expect(sets).not.toContainEqual(expect.objectContaining({ runtime_state: expect.anything() })); + expect(sets.some(entry => runtimeStateSqlText(entry).includes('last_synced_at'))).toBe(false); + const releaseIndex = sets.findIndex(entry => + runtimeStateSqlText(entry).includes("- 'sync_lease'") + ); + expect(releaseIndex).toBeGreaterThanOrEqual(0); + expect(wheres[releaseIndex]).toContain('sync_lease'); + expect(wheres[releaseIndex]).toContain("->>'runId'"); + expect(wheres[releaseIndex]).toContain('chunkIndex'); }); it('short-circuits a recent invalid marker before token minting or GitHub fetch', async () => { @@ -287,7 +395,7 @@ describe('Worker GitHub auth-invalid sync', () => { }); it('does not advance freshness after mixed success then GitHub 401', async () => { - const { db, sets } = createFakeDb({ repositories: ['acme/widgets', 'acme/api'] }); + const { db, sets, wheres } = createFakeDb({ repositories: ['acme/widgets', 'acme/api'] }); const gitTokenService = createGitTokenService(); const fetchStub = vi .fn() @@ -305,12 +413,20 @@ describe('Worker GitHub auth-invalid sync', () => { ).resolves.toMatchObject({ authInvalid: 1, reauthRequired: true }); expect(fetchStub).toHaveBeenCalledTimes(2); - expect(sets).not.toContainEqual(expect.objectContaining({ runtime_state: expect.anything() })); + expect(sets.some(entry => runtimeStateSqlText(entry).includes('last_synced_at'))).toBe(false); + const releaseIndex = sets.findIndex(entry => + runtimeStateSqlText(entry).includes("- 'sync_lease'") + ); + expect(releaseIndex).toBeGreaterThanOrEqual(0); + expect(wheres[releaseIndex]).toContain('sync_lease'); + expect(wheres[releaseIndex]).toContain("->>'runId'"); + expect(wheres[releaseIndex]).toContain('chunkIndex'); }); it('records disabled Dependabot alerts as a repository sync failure', async () => { const { db } = createFakeDb(); const gitTokenService = createGitTokenService(); + const info = vi.spyOn(console, 'info').mockImplementation(() => undefined); const upsertUpdates: Array> = []; const originalInsert = db.insert; db.insert = () => ({ @@ -320,7 +436,13 @@ describe('Worker GitHub auth-invalid sync', () => { }, }), }); - stubFetch(new Response('Dependabot alerts are disabled', { status: 422 })); + const longToken = `ghp_${'A'.repeat(250)}`; + stubFetch( + new Response( + `Dependabot alerts are disabled for this repository. Authorization: bearer ${longToken}`, + { status: 422 } + ) + ); await expect( syncOwner({ @@ -334,9 +456,54 @@ describe('Worker GitHub auth-invalid sync', () => { expect(upsertUpdates).toContainEqual( expect.objectContaining({ last_failure_code: 'DEPENDABOT_ALERTS_DISABLED' }) ); + const disabledLog = info.mock.calls.find( + ([message]) => + typeof message === 'string' && + message.includes('Dependabot alerts disabled for acme/widgets, skipping') + ); + expect(disabledLog?.[1]).toMatchObject({ httpStatus: 422 }); + const excerpt = (disabledLog?.[1] as { bodyExcerpt?: string } | undefined)?.bodyExcerpt ?? ''; + expect(excerpt).toContain('Dependabot alerts are disabled'); + expect(excerpt).not.toContain(longToken); + expect(excerpt).not.toContain('ghp_'); + expect(excerpt.length).toBeLessThanOrEqual(200); db.insert = originalInsert; }); + it('redacts a bare ghp_ token in a disabled-alerts body', async () => { + const { db } = createFakeDb(); + const gitTokenService = createGitTokenService(); + const info = vi.spyOn(console, 'info').mockImplementation(() => undefined); + const bareToken = `ghp_${'B'.repeat(250)}`; + stubFetch( + new Response(`Dependabot alerts are disabled for this repository. token ${bareToken}`, { + status: 422, + }) + ); + + await expect( + syncOwner({ + db: db as never, + gitTokenService, + owner: { userId: 'user-1' }, + runId: 'run-1', + }) + ).resolves.toMatchObject({ skipped: 1 }); + + const disabledLog = info.mock.calls.find( + ([message]) => + typeof message === 'string' && + message.includes('Dependabot alerts disabled for acme/widgets, skipping') + ); + const excerpt = (disabledLog?.[1] as { bodyExcerpt?: string } | undefined)?.bodyExcerpt ?? ''; + expect(excerpt).toContain('Dependabot alerts are disabled'); + expect(excerpt).toContain('[redacted]'); + expect(excerpt).not.toContain(bareToken); + expect(excerpt).not.toContain('ghp_'); + expect(excerpt.length).toBeLessThanOrEqual(200); + info.mockRestore(); + }); + it('throws non-401 GitHub errors', async () => { const { db } = createFakeDb(); const gitTokenService = createGitTokenService(); @@ -547,7 +714,7 @@ describe('Worker GitHub auth-invalid sync', () => { }); it('stops after the first repository when the owner budget is already exhausted', async () => { - const { db, sets } = createFakeDb({ repositories: ['acme/widgets', 'acme/api'] }); + const { db, sets, wheres } = createFakeDb({ repositories: ['acme/widgets', 'acme/api'] }); const gitTokenService = createGitTokenService(); const fetchStub = stubFetch(new Response(JSON.stringify([]), { status: 200 })); @@ -565,10 +732,14 @@ describe('Worker GitHub auth-invalid sync', () => { }); 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'); + const writeIndex = sets.findIndex(entry => runtimeStateSqlText(entry).includes('jsonb_agg')); + expect(writeIndex).toBeGreaterThanOrEqual(0); + expect(runtimeStateSqlText(sets[writeIndex] ?? {}).includes('sync_run')).toBe(true); + expect(runtimeStateSqlText(sets[writeIndex] ?? {}).includes('last_synced_at')).toBe(false); + expect(wheres[writeIndex]).toContain("->>'runId'"); + expect(wheres[writeIndex]).toContain('chunkIndex'); + expect(wheres[writeIndex]).toContain('<='); + expect(wheres[writeIndex]).toContain('run-budget-1'); }); it('counts a fresh-run GitHub failure toward the owner budget and does not mark it complete', async () => { @@ -594,7 +765,7 @@ describe('Worker GitHub auth-invalid sync', () => { }); it('does not keep an incomplete GitHub failure as an error after a successful retry', async () => { - const { db, sets } = createFakeDb({ + const { db, sets, wheres } = createFakeDb({ repositories: ['acme/widgets', 'acme/api'], runtimeState: { sync_run: { @@ -626,11 +797,17 @@ describe('Worker GitHub auth-invalid sync', () => { errors: 0, }); - expect(sets.some(entry => runtimeStateSqlText(entry).includes('last_synced_at'))).toBe(true); + const freshnessIndex = sets.findIndex(entry => + runtimeStateSqlText(entry).includes('last_completed_run_id') + ); + expect(freshnessIndex).toBeGreaterThanOrEqual(0); + expect(wheres[freshnessIndex]).toContain("->>'runId'"); + expect(wheres[freshnessIndex]).toContain('chunkIndex'); + expect(wheres[freshnessIndex]).toContain('run-budget-retry'); }); it('skips completed repositories and finalizes freshness on the last chunk', async () => { - const { db, sets } = createFakeDb({ + const { db, sets, wheres } = createFakeDb({ repositories: ['acme/widgets', 'acme/api'], runtimeState: { sync_run: { @@ -663,7 +840,674 @@ describe('Worker GitHub auth-invalid sync', () => { }); expect(fetchStub).toHaveBeenCalledTimes(1); - expect(sets).toContainEqual(expect.objectContaining({ runtime_state: expect.anything() })); + const freshnessIndex = sets.findIndex(entry => + runtimeStateSqlText(entry).includes('last_completed_run_id') + ); + expect(freshnessIndex).toBeGreaterThanOrEqual(0); + expect(wheres[freshnessIndex]).toContain("->>'runId'"); + expect(wheres[freshnessIndex]).toContain('chunkIndex'); + expect(wheres[freshnessIndex]).toContain('run-budget-1'); + }); +}); + +describe('owner sync lease', () => { + it('denies a claim held by another run without fetching or minting a token', async () => { + const { db, sets, selectCount } = createFakeDb({ + repositories: ['acme/widgets'], + claimAcquired: false, + conflictRuntimeState: { + sync_lease: { + runId: 'other-run', + chunkIndex: 0, + expiresAt: '2999-01-01T00:00:00.000Z', + }, + }, + }); + const gitTokenService = createGitTokenService(); + const fetchStub = stubFetch(new Response('unexpected')); + + const result = await syncOwner({ + db: db as never, + gitTokenService, + owner: { userId: 'user-1' }, + runId: 'run-1', + }); + + expect(result).toMatchObject({ + claimDenied: true, + exhaustedBudget: false, + holderRunId: 'other-run', + commandResultCode: 'SYNC_ALREADY_RUNNING', + }); + expect(result).not.toHaveProperty('staleChunk'); + expect(gitTokenService.getToken).not.toHaveBeenCalled(); + expect(fetchStub).not.toHaveBeenCalled(); + expect(selectCount()).toBe(0); + expect(sets.some(entry => runtimeStateSqlText(entry).includes('last_synced_at'))).toBe(false); + }); + + it('returns staleChunk when an older chunk of this run already holds the lease', async () => { + const { db } = createFakeDb({ + claimAcquired: false, + conflictRuntimeState: { + sync_lease: { + runId: 'run-1', + chunkIndex: 2, + expiresAt: '2999-01-01T00:00:00.000Z', + }, + }, + }); + const gitTokenService = createGitTokenService(); + const fetchStub = stubFetch(new Response('unexpected')); + const info = vi.spyOn(console, 'info').mockImplementation(() => undefined); + + const result = await syncOwner({ + db: db as never, + gitTokenService, + owner: { userId: 'user-1' }, + runId: 'run-1', + chunkIndex: 1, + }); + + expect(result).toMatchObject({ + staleChunk: true, + exhaustedBudget: false, + holderRunId: 'run-1', + }); + expect(result).not.toHaveProperty('claimDenied'); + expect(result).not.toHaveProperty('checkpointRejected'); + expect(gitTokenService.getToken).not.toHaveBeenCalled(); + expect(fetchStub).not.toHaveBeenCalled(); + expect( + info.mock.calls.some( + ([, meta]) => + typeof meta === 'object' && + meta !== null && + (meta as { reason?: string }).reason === 'stale_chunk' + ) + ).toBe(true); + info.mockRestore(); + }); + + it('treats a completed redelivery as a normal non-exhausted result', async () => { + const { db, selectCount } = createFakeDb({ + repositories: ['acme/widgets'], + claimAcquired: false, + conflictRuntimeState: { + last_completed_run_id: 'run-1', + sync_lease: { runId: 'run-1', chunkIndex: 5 }, + }, + }); + const gitTokenService = createGitTokenService(); + const fetchStub = stubFetch(new Response('unexpected')); + + const result = await syncOwner({ + db: db as never, + gitTokenService, + owner: { userId: 'user-1' }, + runId: 'run-1', + }); + + expect(result).not.toHaveProperty('claimDenied'); + expect(result).not.toHaveProperty('staleChunk'); + expect(result.exhaustedBudget).toBe(false); + expect(gitTokenService.getToken).not.toHaveBeenCalled(); + expect(fetchStub).not.toHaveBeenCalled(); + expect(selectCount()).toBe(0); + }); + + it('returns CONFIG_DISABLED without teardown when the claim finds no owner row', async () => { + const { db, sets } = createFakeDb({ + claimAcquired: false, + conflictRuntimeState: null, + configMissing: true, + }); + const gitTokenService = createGitTokenService(); + + const result = await syncOwner({ + db: db as never, + gitTokenService, + owner: { userId: 'user-1' }, + runId: 'run-1', + }); + + expect(result).toMatchObject({ commandResultCode: 'CONFIG_DISABLED' }); + expect(result).not.toHaveProperty('claimDenied'); + expect(sets.some(entry => runtimeStateSqlText(entry).includes("- 'sync_lease'"))).toBe(false); + expect(sets.some(entry => runtimeStateSqlText(entry).includes("- 'sync_run'"))).toBe(false); + }); + + it('clears then releases when a successful claim finds no enabled config', async () => { + const { db, sets } = createFakeDb({ + repositories: ['acme/widgets'], + configMissing: true, + }); + const gitTokenService = createGitTokenService(); + + const result = await syncOwner({ + db: db as never, + gitTokenService, + owner: { userId: 'user-1' }, + runId: 'run-config-disabled', + }); + + expect(result).toMatchObject({ commandResultCode: 'CONFIG_DISABLED' }); + const clearIndex = sets.findIndex(entry => runtimeStateSqlText(entry).includes("- 'sync_run'")); + const releaseIndex = sets.findIndex(entry => + runtimeStateSqlText(entry).includes("- 'sync_lease'") + ); + expect(clearIndex).toBeGreaterThanOrEqual(0); + expect(releaseIndex).toBeGreaterThan(clearIndex); + expect(sets.some(entry => runtimeStateSqlText(entry).includes('last_synced_at'))).toBe(false); + }); + + it('clears then releases on the auth-invalid short-circuit after a successful claim', async () => { + const { db, sets } = createFakeDb({ + authInvalidAt: new Date().toISOString(), + repositories: ['acme/widgets', 'acme/api'], + }); + const gitTokenService = createGitTokenService(); + const fetchStub = stubFetch(new Response('unexpected')); + + const result = await syncOwner({ + db: db as never, + gitTokenService, + owner: { userId: 'user-1' }, + runId: 'run-auth-invalid', + }); + + expect(result).toMatchObject({ authInvalid: 2, reauthRequired: true }); + expect(fetchStub).not.toHaveBeenCalled(); + const clearIndex = sets.findIndex(entry => runtimeStateSqlText(entry).includes("- 'sync_run'")); + const releaseIndex = sets.findIndex(entry => + runtimeStateSqlText(entry).includes("- 'sync_lease'") + ); + expect(clearIndex).toBeGreaterThanOrEqual(0); + expect(releaseIndex).toBeGreaterThan(clearIndex); + expect(sets.some(entry => runtimeStateSqlText(entry).includes('last_synced_at'))).toBe(false); + }); + + it('re-claims once and throws when a vanish-race deny leaves no visible holder', async () => { + const { db, selectCount } = createFakeDb({ + claimAcquired: false, + conflictRuntimeState: {}, + }); + const gitTokenService = createGitTokenService(); + const fetchStub = stubFetch(new Response('unexpected')); + + await expect( + syncOwner({ + db: db as never, + gitTokenService, + owner: { userId: 'user-1' }, + runId: 'run-1', + }) + ).rejects.toThrow('lost its lease holder and could not re-claim'); + + expect(gitTokenService.getToken).not.toHaveBeenCalled(); + expect(fetchStub).not.toHaveBeenCalled(); + expect(selectCount()).toBe(0); + }); + + it('reports staleChunk without releasing when the terminal clear is owned by a newer chunk', async () => { + const { db, sets } = createFakeDb({ + authInvalidAt: new Date().toISOString(), + terminalAccepted: false, + conflictRuntimeState: { sync_lease: { runId: 'run-auth-invalid', chunkIndex: 5 } }, + }); + const gitTokenService = createGitTokenService(); + + const result = await syncOwner({ + db: db as never, + gitTokenService, + owner: { userId: 'user-1' }, + runId: 'run-auth-invalid', + }); + + expect(result).toMatchObject({ staleChunk: true, exhaustedBudget: false }); + expect(result).not.toHaveProperty('checkpointRejected'); + expect(sets.some(entry => runtimeStateSqlText(entry).includes("- 'sync_lease'"))).toBe(false); + }); + + it('reports checkpointRejected and releases when the terminal clear is superseded', async () => { + const { db, sets } = createFakeDb({ + authInvalidAt: new Date().toISOString(), + terminalAccepted: false, + conflictRuntimeState: { sync_lease: { runId: 'other-run', chunkIndex: 0 } }, + }); + const gitTokenService = createGitTokenService(); + + const result = await syncOwner({ + db: db as never, + gitTokenService, + owner: { userId: 'user-1' }, + runId: 'run-auth-invalid', + }); + + expect(result).toMatchObject({ checkpointRejected: true, exhaustedBudget: false }); + expect(result).not.toHaveProperty('staleChunk'); + const releaseIndex = sets.findIndex(entry => + runtimeStateSqlText(entry).includes("- 'sync_lease'") + ); + expect(releaseIndex).toBeGreaterThanOrEqual(0); + }); + + it('returns the completed result and releases when the terminal clear finds this run completed', async () => { + const runId = 'run-completed-teardown'; + const { db, sets } = createFakeDb({ + authInvalidAt: new Date().toISOString(), + terminalAccepted: false, + conflictRuntimeState: { last_completed_run_id: runId }, + }); + const gitTokenService = createGitTokenService(); + + const result = await syncOwner({ + db: db as never, + gitTokenService, + owner: { userId: 'user-1' }, + runId, + }); + + // Exact empty result: dropping the completed-override would instead return the + // auth-invalid result for 'acme/widgets' (authInvalid 1, reauthRequired true). + expect(result).toEqual({ + synced: 0, + errors: 0, + skipped: 0, + authInvalid: 0, + authInvalidRepos: [], + reauthRequired: false, + staleRepos: [], + exhaustedBudget: false, + remainingRepoCount: 0, + }); + const releaseIndex = sets.findIndex(entry => + runtimeStateSqlText(entry).includes("- 'sync_lease'") + ); + expect(releaseIndex).toBeGreaterThanOrEqual(0); + }); + + it('clears and releases without a fetch when the acquired claim already completed this run', async () => { + const runId = 'run-completed-redelivery'; + const { db, sets } = createFakeDb({ + repositories: ['acme/widgets'], + claimRuntimeState: { + last_completed_run_id: runId, + sync_run: { + runId, + completedRepos: [], + staleRepos: [], + authInvalidRepos: [], + synced: 0, + errors: 0, + skipped: 0, + authInvalid: 0, + reauthRequired: false, + noProgressChunks: 0, + }, + }, + }); + const gitTokenService = createGitTokenService(); + const fetchStub = stubFetch(new Response('unexpected')); + + const result = await syncOwner({ + db: db as never, + gitTokenService, + owner: { userId: 'user-1' }, + runId, + }); + + expect(result).not.toHaveProperty('claimDenied'); + expect(result).not.toHaveProperty('staleChunk'); + expect(result.exhaustedBudget).toBe(false); + expect(gitTokenService.getToken).not.toHaveBeenCalled(); + expect(fetchStub).not.toHaveBeenCalled(); + const clearIndex = sets.findIndex(entry => runtimeStateSqlText(entry).includes("- 'sync_run'")); + const releaseIndex = sets.findIndex(entry => + runtimeStateSqlText(entry).includes("- 'sync_lease'") + ); + expect(clearIndex).toBeGreaterThanOrEqual(0); + expect(releaseIndex).toBeGreaterThan(clearIndex); + }); + + it('throws a first-repo failure on a fresh skeleton claim that ignores a snapshot cursor', async () => { + const runId = 'run-fresh-throw'; + const { db } = createFakeDb({ + repositories: ['acme/widgets', 'acme/api'], + claimRuntimeState: { + sync_run: { + runId, + completedRepos: [], + staleRepos: [], + authInvalidRepos: [], + synced: 0, + errors: 0, + skipped: 0, + authInvalid: 0, + reauthRequired: false, + noProgressChunks: 0, + }, + }, + runtimeState: { + sync_run: { + runId, + completedRepos: ['acme/widgets'], + staleRepos: [], + authInvalidRepos: [], + synced: 0, + errors: 0, + skipped: 0, + authInvalid: 0, + reauthRequired: false, + }, + }, + }); + 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, + }) + ).rejects.toThrow('GitHub API error 500 for acme/widgets'); + + // A snapshot cursor would skip acme/widgets and fetch once; a skeleton fetches both. + expect(fetchStub).toHaveBeenCalledTimes(2); + }); + + it('resumes only from the claim RETURNING and never from the returned skeleton', async () => { + const runId = 'run-skeleton'; + const { db } = createFakeDb({ + repositories: ['acme/widgets', 'acme/api'], + claimRuntimeState: { + sync_run: { + runId, + completedRepos: [], + staleRepos: [], + authInvalidRepos: [], + synced: 0, + errors: 0, + skipped: 0, + authInvalid: 0, + reauthRequired: false, + noProgressChunks: 0, + }, + }, + runtimeState: { + sync_run: { + runId, + 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, + }) + ).resolves.toMatchObject({ exhaustedBudget: false, remainingRepoCount: 0, errors: 0 }); + + // The same-runId cursor lives only on the getOwnerConfig snapshot; the claim + // RETURNING is a skeleton. Resuming from the snapshot would skip acme/widgets + // and fetch once. Reading only the RETURNING fetches both repos. + expect(fetchStub).toHaveBeenCalledTimes(2); + }); + + it('rejects a checkpoint write superseded by another run', async () => { + const { db, selectCount } = createFakeDb({ + repositories: ['acme/widgets', 'acme/api'], + writeAccepted: false, + conflictRuntimeState: { sync_lease: { runId: 'other-run', chunkIndex: 0 } }, + }); + const gitTokenService = createGitTokenService(); + stubFetch(new Response(JSON.stringify([]), { status: 200 })); + + const result = await syncOwner({ + db: db as never, + gitTokenService, + owner: { userId: 'user-1' }, + runId: 'run-budget-1', + budgetMs: 0, + }); + + expect(result).toMatchObject({ checkpointRejected: true, exhaustedBudget: false }); + expect(result).not.toHaveProperty('staleChunk'); + expect(selectCount()).toBe(3); + }); + + it('reports staleChunk when a newer chunk of this run rejects a checkpoint write', async () => { + const { db, sets } = createFakeDb({ + repositories: ['acme/widgets', 'acme/api'], + writeAccepted: false, + conflictRuntimeState: { sync_lease: { runId: 'run-budget-1', chunkIndex: 1 } }, + }); + const gitTokenService = createGitTokenService(); + stubFetch(new Response(JSON.stringify([]), { status: 200 })); + + const result = await syncOwner({ + db: db as never, + gitTokenService, + owner: { userId: 'user-1' }, + runId: 'run-budget-1', + budgetMs: 0, + }); + + expect(result).toMatchObject({ staleChunk: true, exhaustedBudget: false }); + expect(result).not.toHaveProperty('checkpointRejected'); + expect(sets.some(entry => runtimeStateSqlText(entry).includes("- 'sync_lease'"))).toBe(false); + }); + + it('releases the lease and returns the empty result when a checkpoint write finds this run completed', async () => { + const runId = 'run-budget-1'; + const { db, sets } = createFakeDb({ + repositories: ['acme/widgets', 'acme/api'], + writeAccepted: false, + conflictRuntimeState: { last_completed_run_id: runId }, + }); + const gitTokenService = createGitTokenService(); + stubFetch(new Response(JSON.stringify([]), { status: 200 })); + + const result = await syncOwner({ + db: db as never, + gitTokenService, + owner: { userId: 'user-1' }, + runId, + budgetMs: 0, + }); + + // The completed arm always releases and overrides the partial chunk result. + expect(result).toEqual({ + synced: 0, + errors: 0, + skipped: 0, + authInvalid: 0, + authInvalidRepos: [], + reauthRequired: false, + staleRepos: [], + exhaustedBudget: false, + remainingRepoCount: 0, + }); + const releaseIndex = sets.findIndex(entry => + runtimeStateSqlText(entry).includes("- 'sync_lease'") + ); + expect(releaseIndex).toBeGreaterThanOrEqual(0); + }); + + it('reports staleChunk without releasing when freshness is rejected by a newer chunk', async () => { + const { db, sets } = createFakeDb({ + repositories: ['acme/widgets'], + terminalAccepted: false, + conflictRuntimeState: { sync_lease: { runId: 'run-fresh', chunkIndex: 1 } }, + }); + const gitTokenService = createGitTokenService(); + stubFetch(new Response(JSON.stringify([]), { status: 200 })); + + const result = await syncOwner({ + db: db as never, + gitTokenService, + owner: { userId: 'user-1' }, + runId: 'run-fresh', + }); + + expect(result).toMatchObject({ staleChunk: true, remainingRepoCount: 0 }); + expect(sets.some(entry => runtimeStateSqlText(entry).includes("- 'sync_lease'"))).toBe(false); + }); + + it('stops with SYNC_NO_PROGRESS at the no-progress chunk limit', async () => { + const runId = 'run-noprogress'; + const { db, sets } = createFakeDb({ + repositories: ['acme/widgets', 'acme/api'], + runtimeState: { + sync_run: { + runId, + completedRepos: [], + staleRepos: [], + authInvalidRepos: [], + synced: 0, + errors: 0, + skipped: 0, + authInvalid: 0, + reauthRequired: false, + noProgressChunks: 1, + }, + }, + }); + const gitTokenService = createGitTokenService(); + const info = vi.spyOn(console, 'info').mockImplementation(() => undefined); + stubFetch(new Response('Service unavailable', { status: 500 })); + + const result = await syncOwner({ + db: db as never, + gitTokenService, + owner: { userId: 'user-1' }, + runId, + budgetMs: 0, + }); + + expect(result).toMatchObject({ + noProgress: true, + exhaustedBudget: false, + commandResultCode: 'SYNC_NO_PROGRESS', + }); + + const clearIndex = sets.findIndex(entry => runtimeStateSqlText(entry).includes("- 'sync_run'")); + const releaseIndex = sets.findIndex(entry => + runtimeStateSqlText(entry).includes("- 'sync_lease'") + ); + expect(clearIndex).toBeGreaterThanOrEqual(0); + expect(releaseIndex).toBeGreaterThan(clearIndex); + expect( + info.mock.calls.some( + ([message, meta]) => + message === 'Security sync lease released' && + (meta as { reason?: string } | undefined)?.reason === 'no_progress' + ) + ).toBe(true); + expect( + info.mock.calls.some( + ([message]) => message === 'Security sync owner budget exhausted; continuation required' + ) + ).toBe(false); + info.mockRestore(); + }); + + it('releases the lease and returns the empty result when the no-progress clear finds this run completed', async () => { + const runId = 'run-noprogress-completed'; + const { db, sets } = createFakeDb({ + repositories: ['acme/widgets', 'acme/api'], + runtimeState: { + sync_run: { + runId, + completedRepos: [], + staleRepos: [], + authInvalidRepos: [], + synced: 0, + errors: 0, + skipped: 0, + authInvalid: 0, + reauthRequired: false, + noProgressChunks: 1, + }, + }, + terminalAccepted: false, + conflictRuntimeState: { last_completed_run_id: runId }, + }); + const gitTokenService = createGitTokenService(); + stubFetch(new Response('Service unavailable', { status: 500 })); + + const result = await syncOwner({ + db: db as never, + gitTokenService, + owner: { userId: 'user-1' }, + runId, + budgetMs: 0, + }); + + // Without the completed release this arm would leave the lease held. + expect(result).toEqual({ + synced: 0, + errors: 0, + skipped: 0, + authInvalid: 0, + authInvalidRepos: [], + reauthRequired: false, + staleRepos: [], + exhaustedBudget: false, + remainingRepoCount: 0, + }); + const releaseIndex = sets.findIndex(entry => + runtimeStateSqlText(entry).includes("- 'sync_lease'") + ); + expect(releaseIndex).toBeGreaterThanOrEqual(0); + }); + + it('binds the claim fence and skeleton in the SQL text', async () => { + const { db, sets, wheres } = createFakeDb({ repositories: ['acme/widgets'] }); + const gitTokenService = createGitTokenService(); + stubFetch(new Response(JSON.stringify([]), { status: 200 })); + + await syncOwner({ + db: db as never, + gitTokenService, + owner: { userId: 'user-1' }, + runId: 'run-1', + chunkIndex: 0, + }); + + const claimWhere = wheres[0] ?? ''; + const claimSet = runtimeStateSqlText(sets[0] ?? {}); + expect(claimWhere).toContain('jsonb_exists'); + expect(claimWhere).toContain('expiresAt'); + expect(claimWhere).toContain('sync_lease'); + expect(claimWhere).toContain('chunkIndex'); + expect(claimWhere).toContain('<='); + expect(claimWhere).toContain('IS DISTINCT FROM'); + expect(claimSet).toContain("'chunkIndex'"); + expect(claimSet).not.toContain('"chunkIndex"'); + expect(claimSet).toContain('noProgressChunks'); + + const releaseIndex = sets.findIndex(entry => + runtimeStateSqlText(entry).includes("- 'sync_lease'") + ); + expect(releaseIndex).toBeGreaterThanOrEqual(0); + expect(wheres[releaseIndex]).toContain("->>'runId'"); + expect(wheres[releaseIndex]).toContain('<='); + expect(wheres[releaseIndex]).not.toContain(' OR '); }); }); diff --git a/services/security-sync/src/sync.ts b/services/security-sync/src/sync.ts index 424eb26aa35..64c7e3358bc 100644 --- a/services/security-sync/src/sync.ts +++ b/services/security-sync/src/sync.ts @@ -165,6 +165,8 @@ type SecurityReviewOwner = | { userId: string; organizationId?: never }; export const SECURITY_SYNC_OWNER_BUDGET_MS = 8 * 60 * 1000; +export const SECURITY_SYNC_LEASE_TTL_MS = 7 * 24 * 60 * 60 * 1000; +export const SECURITY_SYNC_NO_PROGRESS_CHUNK_LIMIT = 2; const SyncRunProgressSchema = z.object({ runId: z.string().min(1), @@ -176,10 +178,21 @@ const SyncRunProgressSchema = z.object({ skipped: z.number().int().nonnegative(), authInvalid: z.number().int().nonnegative(), reauthRequired: z.boolean(), + noProgressChunks: z.number().int().nonnegative().default(0), + chunkIndex: z.number().int().nonnegative().optional(), }); type SyncRunProgress = z.infer; +/** A writer's checkpoint: the schema fields plus the chunk fence this write installs. */ +type SyncRunProgressWrite = SyncRunProgress & { chunkIndex: number }; + +const SyncLeaseSchema = z.object({ + runId: z.string().optional(), + chunkIndex: z.number().int().optional(), + expiresAt: z.string().optional(), +}); + export type SyncResult = { synced: number; errors: number; @@ -195,15 +208,37 @@ export type SyncResult = { commandResultCode?: string; exhaustedBudget: boolean; remainingRepoCount: number; + /** A strictly older chunk of the holding run; ack without syncing or failing the command. */ + staleChunk?: boolean; + /** A different run holds the owner lease; fail the manual command with SYNC_ALREADY_RUNNING. */ + claimDenied?: boolean; + /** A lease-guarded write, clear, or freshness update matched zero rows. */ + checkpointRejected?: boolean; + /** Two consecutive budget chunks completed no repo; the run stopped. */ + noProgress?: boolean; + /** Run id currently holding the lease, when known. */ + holderRunId?: string | null; }; type FetchAlertsResult = | { status: 'success'; alerts: DependabotAlertRaw[] } | { status: 'repo_not_found' } - | { status: 'alerts_disabled' } + | { status: 'alerts_disabled'; httpStatus: number; bodyExcerpt: string } | { status: 'access_blocked' } | { status: 'auth_invalid' }; +const ALERTS_DISABLED_BODY_EXCERPT_LIMIT = 200; + +/** Collapse whitespace, redact token-shaped values, then bound the excerpt. */ +function redactAlertsDisabledBody(body: string): string { + return body + .replace(/\s+/g, ' ') + .replace(/bearer\s+\S+/gi, 'bearer [redacted]') + .replace(/gh[pousr]_[A-Za-z0-9]+/g, '[redacted]') + .replace(/github_pat_[A-Za-z0-9_]+/g, '[redacted]') + .slice(0, ALERTS_DISABLED_BODY_EXCERPT_LIMIT); +} + function createEmptySyncResult(): SyncResult { return { synced: 0, @@ -240,8 +275,10 @@ function applySyncRunProgress(result: SyncResult, progress: SyncRunProgress): vo function toSyncRunProgress( runId: string, result: SyncResult, - completedRepos: string[] -): SyncRunProgress { + completedRepos: string[], + chunkIndex: number, + noProgressChunks: number +): SyncRunProgressWrite { return { runId, completedRepos, @@ -252,21 +289,195 @@ function toSyncRunProgress( skipped: result.skipped, authInvalid: result.authInvalid, reauthRequired: result.reauthRequired, + noProgressChunks, + chunkIndex, }; } -async function writeSyncRunProgress( +function readLastCompletedRunId(runtimeState: Record): string | null { + const value = runtimeState.last_completed_run_id; + return typeof value === 'string' ? value : null; +} + +function readSyncRunRunId(runtimeState: Record): string | null { + const value = runtimeState.sync_run; + if ( + value && + typeof value === 'object' && + typeof (value as { runId?: unknown }).runId === 'string' + ) { + return (value as { runId: string }).runId; + } + return null; +} + +type SyncLeaseReadOutcome = + | { kind: 'completed' } + | { kind: 'stale_chunk'; holderRunId: string | null; holderChunkIndex: number | null } + | { kind: 'superseded'; holderRunId: string | null; holderChunkIndex: number | null }; + +/** + * Classifies a runtime_state read that did not acquire the lease. `completed` + * means this run already finished. `stale_chunk` means a newer chunk of this + * same run holds the lease, or no different run id is visible at all (the lease + * vanished in a millisecond race). `superseded` means a different run id is + * actually visible on the lease or the cursor. + */ +function classifySyncLeaseRead( + runtimeState: Record, + runId: string, + chunkIndex: number +): SyncLeaseReadOutcome { + if (readLastCompletedRunId(runtimeState) === runId) return { kind: 'completed' }; + const parsedLease = SyncLeaseSchema.safeParse(runtimeState.sync_lease); + const lease = parsedLease.success ? parsedLease.data : null; + const holderRunId = lease?.runId ?? null; + const holderChunkIndex = lease?.chunkIndex ?? null; + if (holderRunId === runId && (holderChunkIndex ?? -1) > chunkIndex) { + return { kind: 'stale_chunk', holderRunId, holderChunkIndex }; + } + const cursorRunId = readSyncRunRunId(runtimeState); + const foreignHolder = holderRunId !== null && holderRunId !== runId; + const foreignCursor = cursorRunId !== null && cursorRunId !== runId; + if (foreignHolder || foreignCursor) { + return { kind: 'superseded', holderRunId, holderChunkIndex }; + } + return { kind: 'stale_chunk', holderRunId, holderChunkIndex }; +} + +async function selectOwnerRuntimeState( + db: WorkerDb, + owner: SecurityReviewOwner +): Promise | null> { + const rows = await db + .select({ runtime_state: agent_configs.runtime_state }) + .from(agent_configs) + .where( + and( + eq(agent_configs.agent_type, 'security_scan'), + eq(agent_configs.platform, 'github'), + ownerFilter(owner) + ) + ) + .limit(1); + const value = rows[0]?.runtime_state; + return value ?? null; +} + +/** + * Claims the single owner-sync lease for `runId` at `chunkIndex`. Returns the + * post-image runtime_state, or null when another run holds the lease or a + * stricter same-run chunk is live. + */ +export async function claimOwnerSyncLease( db: WorkerDb, owner: SecurityReviewOwner, - progress: SyncRunProgress -): Promise { - await db + runId: string, + chunkIndex: number +): Promise<{ runtimeState: Record } | null> { + const ttlSeconds = Math.floor(SECURITY_SYNC_LEASE_TTL_MS / 1000); + const lease = sql`jsonb_build_object( + 'runId', ${runId}::text, + 'chunkIndex', to_jsonb(${chunkIndex}::int), + 'expiresAt', to_jsonb(now() + make_interval(secs => ${ttlSeconds})) + )`; + const skeleton = JSON.stringify({ + runId, + completedRepos: [], + staleRepos: [], + authInvalidRepos: [], + synced: 0, + errors: 0, + skipped: 0, + authInvalid: 0, + reauthRequired: false, + noProgressChunks: 0, + }); + const [row] = await db + .update(agent_configs) + .set({ + runtime_state: sql`CASE + WHEN ${agent_configs.runtime_state}->'sync_run'->>'runId' = ${runId} + THEN jsonb_set(COALESCE(${agent_configs.runtime_state}, '{}'::jsonb), '{sync_lease}', ${lease}, true) + ELSE jsonb_set( + jsonb_set(COALESCE(${agent_configs.runtime_state}, '{}'::jsonb), '{sync_lease}', ${lease}, true), + '{sync_run}', + ${skeleton}::jsonb, + true + ) + END`, + }) + .where( + and( + eq(agent_configs.agent_type, 'security_scan'), + eq(agent_configs.platform, 'github'), + ownerFilter(owner), + sql`( + NOT jsonb_exists(COALESCE(${agent_configs.runtime_state}, '{}'::jsonb), 'sync_lease') + OR ( + ${agent_configs.runtime_state}->'sync_lease'->>'runId' = ${runId} + AND COALESCE((${agent_configs.runtime_state}->'sync_lease'->>'chunkIndex')::int, -1) <= ${chunkIndex} + ) + OR ( + ${agent_configs.runtime_state}->'sync_lease'->>'runId' IS DISTINCT FROM ${runId} + AND (${agent_configs.runtime_state}->'sync_lease'->>'expiresAt')::timestamptz < now() + ) + )` + ) + ) + .returning({ runtimeState: agent_configs.runtime_state }); + + if (!row?.runtimeState) return null; + return { runtimeState: row.runtimeState }; +} + +export async function releaseOwnerSyncLease( + db: WorkerDb, + owner: SecurityReviewOwner, + runId: string, + chunkIndex: number +): Promise<{ released: boolean }> { + const rows = await db + .update(agent_configs) + .set({ + runtime_state: sql`COALESCE(${agent_configs.runtime_state}, '{}'::jsonb) - 'sync_lease'`, + }) + .where( + and( + eq(agent_configs.agent_type, 'security_scan'), + eq(agent_configs.platform, 'github'), + ownerFilter(owner), + sql`${agent_configs.runtime_state}->'sync_lease'->>'runId' = ${runId}`, + sql`COALESCE((${agent_configs.runtime_state}->'sync_lease'->>'chunkIndex')::int, -1) <= ${chunkIndex}` + ) + ) + .returning({ id: agent_configs.id }); + return { released: rows.length > 0 }; +} + +export async function writeSyncRunProgress( + db: WorkerDb, + owner: SecurityReviewOwner, + progress: SyncRunProgressWrite +): Promise<{ written: boolean }> { + const progressJson = JSON.stringify(progress); + const completedReposJson = JSON.stringify(progress.completedRepos); + const rows = await db .update(agent_configs) .set({ runtime_state: sql`jsonb_set( COALESCE(${agent_configs.runtime_state}, '{}'::jsonb), '{sync_run}', - ${JSON.stringify(progress)}::jsonb, + ${progressJson}::jsonb || jsonb_build_object( + 'completedRepos', + ( + SELECT COALESCE(jsonb_agg(DISTINCT elem), '[]'::jsonb) + FROM jsonb_array_elements_text( + COALESCE(${agent_configs.runtime_state}->'sync_run'->'completedRepos', '[]'::jsonb) + || ${completedReposJson}::jsonb + ) AS elem + ) + ), true )`, }) @@ -274,13 +485,24 @@ async function writeSyncRunProgress( and( eq(agent_configs.agent_type, 'security_scan'), eq(agent_configs.platform, 'github'), - ownerFilter(owner) + ownerFilter(owner), + sql`${agent_configs.runtime_state}->'sync_lease'->>'runId' = ${progress.runId}`, + sql`${agent_configs.runtime_state}->'sync_run'->>'runId' = ${progress.runId}`, + sql`COALESCE((${agent_configs.runtime_state}->'sync_lease'->>'chunkIndex')::int, -1) <= ${progress.chunkIndex}`, + sql`(${agent_configs.runtime_state}->'sync_run'->'chunkIndex' IS NULL OR (${agent_configs.runtime_state}->'sync_run'->>'chunkIndex')::int <= ${progress.chunkIndex})` ) - ); + ) + .returning({ id: agent_configs.id }); + return { written: rows.length > 0 }; } -async function clearSyncRunProgress(db: WorkerDb, owner: SecurityReviewOwner): Promise { - await db +export async function clearSyncRunProgress( + db: WorkerDb, + owner: SecurityReviewOwner, + runId: string, + chunkIndex: number +): Promise<{ cleared: boolean }> { + const rows = await db .update(agent_configs) .set({ runtime_state: sql`COALESCE(${agent_configs.runtime_state}, '{}'::jsonb) - 'sync_run'`, @@ -289,9 +511,166 @@ async function clearSyncRunProgress(db: WorkerDb, owner: SecurityReviewOwner): P and( eq(agent_configs.agent_type, 'security_scan'), eq(agent_configs.platform, 'github'), - ownerFilter(owner) + ownerFilter(owner), + sql`${agent_configs.runtime_state}->'sync_run'->>'runId' = ${runId}`, + sql`${agent_configs.runtime_state}->'sync_lease'->>'runId' = ${runId}`, + sql`COALESCE((${agent_configs.runtime_state}->'sync_lease'->>'chunkIndex')::int, -1) <= ${chunkIndex}`, + sql`(${agent_configs.runtime_state}->'sync_run'->'chunkIndex' IS NULL OR (${agent_configs.runtime_state}->'sync_run'->>'chunkIndex')::int <= ${chunkIndex})` ) - ); + ) + .returning({ id: agent_configs.id }); + return { cleared: rows.length > 0 }; +} + +export async function advanceOwnerSyncFreshness( + db: WorkerDb, + owner: SecurityReviewOwner, + runId: string, + chunkIndex: number +): Promise<{ advanced: boolean }> { + const rows = await db + .update(agent_configs) + .set({ + runtime_state: sql`jsonb_set( + COALESCE(${agent_configs.runtime_state}, '{}'::jsonb) - 'sync_run', + '{last_synced_at}', + to_jsonb(now()), + true + ) || jsonb_build_object('last_completed_run_id', ${runId}::text)`, + }) + .where( + and( + eq(agent_configs.agent_type, 'security_scan'), + eq(agent_configs.platform, 'github'), + ownerFilter(owner), + sql`${agent_configs.runtime_state}->'sync_run'->>'runId' = ${runId}`, + sql`${agent_configs.runtime_state}->'sync_lease'->>'runId' = ${runId}`, + sql`COALESCE((${agent_configs.runtime_state}->'sync_lease'->>'chunkIndex')::int, -1) <= ${chunkIndex}`, + sql`(${agent_configs.runtime_state}->'sync_run'->'chunkIndex' IS NULL OR (${agent_configs.runtime_state}->'sync_run'->>'chunkIndex')::int <= ${chunkIndex})` + ) + ) + .returning({ id: agent_configs.id }); + return { advanced: rows.length > 0 }; +} + +/** One read + classification for a lease-guarded mutation that matched zero rows. */ +async function classifyOwnerMutationMiss( + db: WorkerDb, + owner: SecurityReviewOwner, + runId: string, + chunkIndex: number +): Promise { + const runtimeState = await selectOwnerRuntimeState(db, owner); + return runtimeState ? classifySyncLeaseRead(runtimeState, runId, chunkIndex) : null; +} + +/** Lease release reasons emitted by owner-scoped terminal returns. */ +type OwnerLeaseReleaseReason = 'completed' | 'incomplete' | 'no_progress' | 'abandoned'; + +/** + * Releases the owner lease and logs the outcome. A zero-row release is logged + * and returned; a database error propagates so the bounded queue retry runs and + * the lease does not stay held for the 7-day TTL. + */ +async function releaseOwnerSyncLeaseLogged( + db: WorkerDb, + owner: SecurityReviewOwner, + runId: string, + chunkIndex: number, + reason: OwnerLeaseReleaseReason +): Promise { + const released = await releaseOwnerSyncLease(db, owner, runId, chunkIndex); + if (released.released) { + console.info('Security sync lease released', { runId, chunkIndex, reason }); + } else { + console.info('Security sync lease release rejected', { runId, holderRunId: null, reason }); + } +} + +/** + * The single decision for a lease-guarded mutation (write, clear, or freshness + * update) that matched zero rows: what to return and whether the owner lease + * must be released. A newer chunk of this same run never releases. A superseded + * or vanished lease releases only where the caller owns the foreign-release + * decision. A completed run always releases and returns the empty result. + */ +function reactToOwnerMutationMiss( + miss: SyncLeaseReadOutcome | null, + base: SyncResult, + options: { releaseOnSuperseded: boolean } +): { result: SyncResult; release: boolean } { + if (miss?.kind === 'stale_chunk') { + return { + result: { ...base, staleChunk: true, exhaustedBudget: false, holderRunId: miss.holderRunId }, + release: false, + }; + } + if (miss?.kind === 'completed') { + return { result: createEmptySyncResult(), release: true }; + } + return { + result: { ...base, checkpointRejected: true, exhaustedBudget: false }, + release: options.releaseOnSuperseded, + }; +} + +/** + * Fenced clear then release for an owner-scoped terminal return. The clear is + * best-effort so a failed clear does not mask the real sync result. A zero-row + * clear is classified against the live runtime state: a newer chunk of this run + * owns the command (`staleChunk`, no release), a different run supersedes it + * (`checkpointRejected`, release), or this run already completed (release and + * return the completed result). Returns an overriding result on those + * classifications, or null to return the caller's own result. Release errors + * propagate. + */ +async function teardownOwnerLease( + db: WorkerDb, + owner: SecurityReviewOwner, + runId: string, + chunkIndex: number, + reason: 'incomplete' | 'abandoned' +): Promise { + let cleared: boolean; + try { + cleared = (await clearSyncRunProgress(db, owner, runId, chunkIndex)).cleared; + } catch (error) { + console.error('Failed to clear Security sync run progress', { + runId, + chunkIndex, + reason, + error_type: error instanceof Error ? error.name : 'UnknownError', + }); + await releaseOwnerSyncLeaseLogged(db, owner, runId, chunkIndex, reason); + return null; + } + + if (cleared) { + await releaseOwnerSyncLeaseLogged(db, owner, runId, chunkIndex, reason); + return null; + } + + const miss = await classifyOwnerMutationMiss(db, owner, runId, chunkIndex); + if (miss?.kind === 'stale_chunk') { + console.info('Security sync checkpoint clear rejected', { + runId, + chunkIndex, + reason: 'stale_chunk', + }); + } else if (miss?.kind !== 'completed') { + console.info('Security sync checkpoint clear rejected', { + runId, + chunkIndex, + reason: 'superseded', + }); + } + const reaction = reactToOwnerMutationMiss(miss, createEmptySyncResult(), { + releaseOnSuperseded: true, + }); + if (reaction.release) { + await releaseOwnerSyncLeaseLogged(db, owner, runId, chunkIndex, reason); + } + return reaction.result; } function createAuthInvalidSyncResult(repositories: string[]): SyncResult { @@ -369,7 +748,6 @@ 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( @@ -382,7 +760,6 @@ 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( @@ -508,10 +885,6 @@ export async function getOwnerConfig( authInvalidAt: integration.authInvalidAt, githubAppType: integration.githubAppType ?? 'standard', missingSelectedRepoCount, - runtimeState: - agentConfig.runtime_state && typeof agentConfig.runtime_state === 'object' - ? agentConfig.runtime_state - : {}, }; } @@ -624,7 +997,11 @@ export async function fetchAllDependabotAlerts( body.includes('archived repositories') || body.includes('archived repository')) ) { - return { status: 'alerts_disabled' }; + return { + status: 'alerts_disabled', + httpStatus: response.status, + bodyExcerpt: redactAlertsDisabledBody(body), + }; } throw new Error(`GitHub API error ${response.status} for ${repoOwner}/${repoName}`); @@ -1722,14 +2099,142 @@ export async function syncOwner(params: { repoFullName?: string; notificationMaterializationEnabled?: boolean; budgetMs?: number; + chunkIndex?: number; }): Promise { const { db: database, gitTokenService, owner, runId, actor, repoFullName } = params; const trigger = params.trigger ?? 'scheduled'; + const chunkIndex = params.chunkIndex ?? 0; const startTime = Date.now(); + const ownerId = + 'organizationId' in owner ? (owner.organizationId ?? 'unknown') : (owner.userId ?? 'unknown'); + const ownerType = isOrgOwner(owner) ? 'org' : 'user'; + const ownerScoped = repoFullName === undefined; + + let claimRuntimeState: Record | null = null; + let previousProgress: SyncRunProgress | null = null; + let leaseHeld = false; + + if (ownerScoped) { + let claimed = await claimOwnerSyncLease(database, owner, runId, chunkIndex); + if (!claimed) { + const denyState = await selectOwnerRuntimeState(database, owner); + if (denyState) { + const outcome = classifySyncLeaseRead(denyState, runId, chunkIndex); + if (outcome.kind === 'completed') { + console.info('Security sync run already completed; delivery acknowledged', { + runId, + ownerId, + ownerType, + chunkIndex, + }); + return createEmptySyncResult(); + } + if (outcome.kind === 'stale_chunk' && outcome.holderRunId === null) { + // Only the vanish race reaches here: the holder released between the failed + // claim and this read, so no lease is visible and nothing else will + // terminalize or retry the command. A genuine same-run-stale classification + // always reports a non-null holder. Re-claim once and continue; if the retry + // misses too, throw so the bounded queue retry re-claims. + console.info('Security sync run claim denied', { + runId, + ownerId, + ownerType, + chunkIndex, + holderRunId: outcome.holderRunId, + holderChunkIndex: outcome.holderChunkIndex, + reason: 'stale_chunk', + }); + claimed = await claimOwnerSyncLease(database, owner, runId, chunkIndex); + if (!claimed) { + throw new Error( + `Security sync run ${runId} lost its lease holder and could not re-claim` + ); + } + } else if (outcome.kind === 'stale_chunk') { + console.info('Security sync run claim denied', { + runId, + ownerId, + ownerType, + chunkIndex, + holderRunId: outcome.holderRunId, + holderChunkIndex: outcome.holderChunkIndex, + reason: 'stale_chunk', + }); + return { + ...createEmptySyncResult(), + staleChunk: true, + exhaustedBudget: false, + holderRunId: outcome.holderRunId, + }; + } else { + console.info('Security sync run claim denied', { + runId, + ownerId, + ownerType, + chunkIndex, + holderRunId: outcome.holderRunId, + holderChunkIndex: outcome.holderChunkIndex, + reason: 'held_by_other', + }); + return { + ...createEmptySyncResult(), + claimDenied: true, + exhaustedBudget: false, + holderRunId: outcome.holderRunId, + commandResultCode: 'SYNC_ALREADY_RUNNING', + }; + } + } + // No config row exists; fall through so getOwnerConfig returns CONFIG_DISABLED. + } + + if (claimed) { + leaseHeld = true; + claimRuntimeState = claimed.runtimeState; + if (readLastCompletedRunId(claimRuntimeState) === runId) { + await clearSyncRunProgress(database, owner, runId, chunkIndex); + await releaseOwnerSyncLeaseLogged(database, owner, runId, chunkIndex, 'completed'); + console.info('Security sync run completed redelivery', { runId, ownerId, chunkIndex }); + return createEmptySyncResult(); + } + const parsedProgress = readSyncRunProgress(claimRuntimeState, runId); + const resumed = + parsedProgress !== null && + (parsedProgress.completedRepos.length > 0 || + parsedProgress.chunkIndex !== undefined || + parsedProgress.noProgressChunks > 0); + previousProgress = resumed ? parsedProgress : null; + console.info('Security sync run claim acquired', { + runId, + ownerId, + ownerType, + chunkIndex, + resumed, + }); + if (previousProgress) { + console.info('Resuming security sync owner run', { + runId, + storedRunId: previousProgress.runId, + chunkIndex, + ownerId, + }); + } else { + console.info('Security sync resume miss', { + runId, + chunkIndex, + reason: parsedProgress === null ? 'schema_invalid' : 'no_checkpoint', + }); + } + } + } const config = await getOwnerConfig(database, owner); if (!config) { console.info(`No enabled config for owner, skipping`, { runId, owner }); + if (leaseHeld) { + const teardown = await teardownOwnerLease(database, owner, runId, chunkIndex, 'abandoned'); + if (teardown) return teardown; + } return { ...createEmptySyncResult(), commandResultCode: 'CONFIG_DISABLED' }; } @@ -1755,26 +2260,25 @@ export async function syncOwner(params: { repositoryCount: repositories.length, authInvalidAt: config.authInvalidAt, }); + if (leaseHeld) { + const teardown = await teardownOwnerLease(database, owner, runId, chunkIndex, 'abandoned'); + if (teardown) return teardown; + } 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; + let nextRepo: string | null = null; const notificationPolicy = params.notificationMaterializationEnabled ? config.notificationPolicy : null; @@ -1789,9 +2293,23 @@ export async function syncOwner(params: { Date.now() - startTime >= params.budgetMs ) { exhaustedBudget = true; + nextRepo = selectedRepoFullName; break; } + const remainingBudgetMs = + params.budgetMs === undefined + ? null + : Math.max(0, params.budgetMs - (Date.now() - startTime)); + const repoStartedAt = Date.now(); + let outcome: 'synced' | 'skipped' | 'auth_invalid' | 'stale' | 'error' = 'synced'; + console.info('Security sync repo started', { + runId, + repoFullName: selectedRepoFullName, + remainingBudgetMs, + chunkIndex, + }); + try { const repoResult = await syncRepo({ db: database, @@ -1818,10 +2336,19 @@ export async function syncOwner(params: { completedRepos.add(selectedRepoFullName); processedThisPass++; + if (repoResult.authInvalid > 0 || repoResult.authInvalidRepos.length > 0) { + outcome = 'auth_invalid'; + } else if (repoResult.staleRepos.length > 0) { + outcome = 'stale'; + } else if (repoResult.skipped > 0) { + outcome = 'skipped'; + } + if (repoResult.reauthRequired) { break; } } catch (error) { + outcome = 'error'; incompleteFailures++; await recordSecurityAgentRepositorySyncFailure(database, { owner: toSecurityAgentCommandOwner(owner), @@ -1835,21 +2362,125 @@ export async function syncOwner(params: { firstError = error; } processedThisPass++; + } finally { + console.info('Security sync repo finished', { + runId, + repoFullName: selectedRepoFullName, + durationMs: Date.now() - repoStartedAt, + outcome, + remainingBudgetMs, + chunkIndex, + }); } } const remainingRepoCount = repositories.filter(name => !completedRepos.has(name)).length; if (exhaustedBudget && remainingRepoCount > 0 && !totalResult.reauthRequired) { - await writeSyncRunProgress( + const admission = previousProgress?.noProgressChunks ?? 0; + const noProgressChunks = successfulRepos > 0 ? 0 : admission + 1; + const stoppedForNoProgress = noProgressChunks >= SECURITY_SYNC_NO_PROGRESS_CHUNK_LIMIT; + + if (!stoppedForNoProgress) { + console.info('Security sync owner budget exhausted; continuation required', { + runId, + completedRepoCount: completedRepos.size, + remainingRepoCount, + durationMs: Date.now() - startTime, + budgetMs: params.budgetMs ?? null, + remainingBudgetMs: + params.budgetMs === undefined + ? null + : Math.max(0, params.budgetMs - (Date.now() - startTime)), + lastCompletedRepo: [...completedRepos].at(-1) ?? null, + nextRepo, + chunkIndex, + noProgressChunks, + stoppedForNoProgress, + }); + } + + if (stoppedForNoProgress) { + const cleared = await clearSyncRunProgress(database, owner, runId, chunkIndex); + if (cleared.cleared) { + await releaseOwnerSyncLeaseLogged(database, owner, runId, chunkIndex, 'no_progress'); + console.info('Security sync no progress; run stopped', { + runId, + chunkIndex, + noProgressChunks, + completedRepoCount: completedRepos.size, + stoppedForNoProgress, + }); + return { + ...totalResult, + noProgress: true, + exhaustedBudget: false, + commandResultCode: 'SYNC_NO_PROGRESS', + remainingRepoCount, + }; + } + const miss = await classifyOwnerMutationMiss(database, owner, runId, chunkIndex); + if (miss?.kind === 'stale_chunk') { + console.info('Security sync checkpoint clear rejected', { + runId, + chunkIndex, + reason: 'stale_chunk', + }); + } else if (miss?.kind !== 'completed') { + console.info('Security sync checkpoint clear rejected', { + runId, + chunkIndex, + reason: 'superseded', + }); + } + const reaction = reactToOwnerMutationMiss( + miss, + { ...totalResult, remainingRepoCount }, + { releaseOnSuperseded: false } + ); + if (reaction.release) { + await releaseOwnerSyncLeaseLogged(database, owner, runId, chunkIndex, 'completed'); + } + return reaction.result; + } + + const written = await writeSyncRunProgress( database, owner, - toSyncRunProgress(runId, totalResult, [...completedRepos]) + toSyncRunProgress(runId, totalResult, [...completedRepos], chunkIndex, noProgressChunks) ); - console.info('Security sync owner budget exhausted; continuation required', { + if (!written.written) { + const miss = await classifyOwnerMutationMiss(database, owner, runId, chunkIndex); + if (miss?.kind === 'stale_chunk') { + console.info('Security sync checkpoint write rejected', { + runId, + chunkIndex, + completedRepoCount: completedRepos.size, + reason: 'stale_chunk', + }); + } else if (miss?.kind !== 'completed') { + console.info('Security sync checkpoint write rejected', { + runId, + chunkIndex, + completedRepoCount: completedRepos.size, + reason: 'superseded', + }); + } + const reaction = reactToOwnerMutationMiss( + miss, + { ...totalResult, remainingRepoCount }, + { releaseOnSuperseded: false } + ); + if (reaction.release) { + await releaseOwnerSyncLeaseLogged(database, owner, runId, chunkIndex, 'completed'); + } + return reaction.result; + } + console.info('Security sync checkpoint write accepted', { runId, + chunkIndex, completedRepoCount: completedRepos.size, remainingRepoCount, - durationMs: Date.now() - startTime, + noProgressChunks, }); return { ...totalResult, exhaustedBudget: true, remainingRepoCount }; } @@ -1886,8 +2517,6 @@ export async function syncOwner(params: { } // Write audit log - const ownerId = - 'organizationId' in owner ? (owner.organizationId ?? 'unknown') : (owner.userId ?? 'unknown'); try { await writeAuditLog(database, { owner, @@ -1921,42 +2550,51 @@ export async function syncOwner(params: { // Missing selected repos (installation lost access) also block — the repo // was configured but silently dropped from the accessible list. const shouldAdvanceFreshness = - !repoFullName && + ownerScoped && totalResult.errors === 0 && totalResult.authInvalid === 0 && totalResult.staleRepos.length === 0 && config.missingSelectedRepoCount === 0; if (shouldAdvanceFreshness) { - try { - await database - .update(agent_configs) - .set({ - runtime_state: sql`jsonb_set( - COALESCE(${agent_configs.runtime_state}, '{}'::jsonb) - 'sync_run', - '{last_synced_at}', - to_jsonb(now()) - )`, - }) - .where( - and( - eq(agent_configs.agent_type, 'security_scan'), - eq(agent_configs.platform, 'github'), - ownerFilter(owner) - ) + const advanced = await advanceOwnerSyncFreshness(database, owner, runId, chunkIndex); + if (advanced.advanced) { + await releaseOwnerSyncLeaseLogged(database, owner, runId, chunkIndex, 'completed'); + } else { + const miss = await classifyOwnerMutationMiss(database, owner, runId, chunkIndex); + if (miss?.kind === 'stale_chunk') { + console.info('Security sync freshness update rejected', { + runId, + chunkIndex, + completedRepoCount: completedRepos.size, + reason: 'stale_chunk', + }); + } else if (miss?.kind !== 'completed') { + console.info('Security sync freshness update rejected', { + runId, + chunkIndex, + completedRepoCount: completedRepos.size, + reason: 'superseded', + }); + } + const reaction = reactToOwnerMutationMiss( + miss, + { ...totalResult, remainingRepoCount: 0 }, + { releaseOnSuperseded: true } + ); + if (reaction.release) { + await releaseOwnerSyncLeaseLogged( + database, + owner, + runId, + chunkIndex, + miss?.kind === 'completed' ? 'completed' : 'abandoned' ); - } catch (error) { - console.error('Failed to update last_synced_at in runtime_state', { - 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), - }); + } + return reaction.result; } + } else if (leaseHeld) { + const teardown = await teardownOwnerLease(database, owner, runId, chunkIndex, 'incomplete'); + if (teardown) return teardown; } const syncSummary = { @@ -2058,7 +2696,12 @@ async function syncRepo(params: { } if (fetchResult.status === 'alerts_disabled') { - console.info(`Dependabot alerts disabled for ${repoFullName}, skipping`); + console.info(`Dependabot alerts disabled for ${repoFullName}, skipping`, { + runId, + repoFullName, + httpStatus: fetchResult.httpStatus, + bodyExcerpt: fetchResult.bodyExcerpt, + }); result.skipped = 1; await recordSecurityAgentRepositorySyncFailure(database, { owner: commandOwner, diff --git a/services/security-sync/vitest.config.ts b/services/security-sync/vitest.config.ts index 7dd13254e75..b1d5b43bb98 100644 --- a/services/security-sync/vitest.config.ts +++ b/services/security-sync/vitest.config.ts @@ -5,5 +5,6 @@ export default defineConfig({ globals: true, environment: 'node', include: ['src/**/*.test.ts'], + exclude: ['src/**/*.integration.test.ts'], }, }); diff --git a/services/security-sync/vitest.integration.config.ts b/services/security-sync/vitest.integration.config.ts new file mode 100644 index 00000000000..5fec89faff5 --- /dev/null +++ b/services/security-sync/vitest.integration.config.ts @@ -0,0 +1,10 @@ +import { defineConfig } from 'vitest/config'; + +export default defineConfig({ + test: { + globals: true, + environment: 'node', + fileParallelism: false, + include: ['src/**/*.integration.test.ts'], + }, +});