diff --git a/web/__tests__/auth.test.ts b/web/__tests__/auth.test.ts index f4544953..f434e847 100644 --- a/web/__tests__/auth.test.ts +++ b/web/__tests__/auth.test.ts @@ -355,7 +355,7 @@ describe('getSession', () => { const req = fakeRequest('00000000-0000-4000-8000-000000000000') const result = await getSession(req) expect(result).toEqual({ sessionId: '00000000-0000-4000-8000-000000000010', userId: 'user-abc' }) - expect(mockDbTransaction).toHaveBeenCalledOnce() + expect(mockDbTransaction).toHaveBeenCalledTimes(2) expect(mockRedisSet).toHaveBeenCalledWith(expect.stringMatching(/^session:v2:/), expect.any(String), 'PXAT', expect.any(Number)) }) @@ -490,7 +490,17 @@ describe('getSession', () => { credentialDigestV1: null, credentialStorageVersion: 0, databaseNow: new Date(redisNowMs), - }])) + }])) + .mockReturnValueOnce(chain([{ + sessionId: credential, + userId: 'user-abc', + lastSeenAt: new Date(redisNowMs - 1_000), + expiresAt: new Date(expiresAtMs), + revokedAt: null, + credentialStorageVersion: 1, + databaseNow: new Date(redisNowMs), + }])) + .mockReturnValueOnce(chain([{ state: 'expansion' }])) mockDbUpdate.mockReturnValue(chain([{ id: credential }])) await expect(getSession(fakeRequest(credential))).resolves.toEqual({ @@ -548,6 +558,10 @@ describe('getSession', () => { describe('getSession — write-behind logic', () => { beforeEach(() => { vi.clearAllMocks() + mockDbSelect.mockReset() + mockDbUpdate.mockReset() + mockDbTransaction.mockReset() + mockDbTransaction.mockImplementation(async (callback: (tx: unknown) => unknown) => callback(transactionClient())) vi.useFakeTimers() mockRedisSet.mockResolvedValue('OK') mockDbUpdate.mockReturnValue(chain([{ @@ -557,9 +571,11 @@ describe('getSession — write-behind logic', () => { afterEach(() => { vi.useRealTimers() + mockDbTransaction.mockReset() + mockDbTransaction.mockImplementation(async (callback: (tx: unknown) => unknown) => callback(transactionClient())) }) - it('does NOT trigger a DB write when lastSeenAt is less than 60 seconds old', async () => { + it('writes cache entries inside a second locked transaction when lastSeenAt is recent', async () => { const now = Date.now() vi.setSystemTime(now) @@ -573,9 +589,11 @@ describe('getSession — write-behind logic', () => { await getSession(req) expect(mockDbUpdate).not.toHaveBeenCalled() + expect(mockDbTransaction).toHaveBeenCalledTimes(2) + expect(mockRedisSet).toHaveBeenCalledOnce() }) - it('triggers a fire-and-forget DB write when lastSeenAt is older than 60 seconds', async () => { + it('refreshes the authoritative row before the locked cache write when lastSeenAt is older than 60 seconds', async () => { const now = Date.now() vi.setSystemTime(now) @@ -588,12 +606,80 @@ describe('getSession — write-behind logic', () => { const req = fakeRequest('00000000-0000-4000-8000-000000000000') await getSession(req) - // DB update is kicked off fire-and-forget; the mock should have been invoked expect(mockDbUpdate).toHaveBeenCalledOnce() - // Redis was also refreshed expect(mockRedisSet).toHaveBeenCalledOnce() }) + it.each([ + ['v2', 2, null], + ['dual-write v1', 1, '00000000-0000-4000-8000-000000000000'], + ])('orders a completed %s cache write before a following revocation deletion even if its transaction fails', async ( + _label, + credentialStorageVersion, + expectedLegacyKey, + ) => { + const now = Date.now() + vi.setSystemTime(now) + const credential = '00000000-0000-4000-8000-000000000000' + const sessionId = credentialStorageVersion === 1 ? credential : '00000000-0000-4000-8000-000000000010' + const row = { + sessionId, userId: 'user-1', + lastSeenAt: new Date(now - 30_000), expiresAt: new Date(now + 60_000), + revokedAt: null, credentialStorageVersion, databaseNow: new Date(now), + } + mockDbSelect + .mockReturnValueOnce(chain([{ state: 'expansion' }])) + .mockReturnValueOnce(chain([row])) + .mockReturnValueOnce(chain([row])) + .mockReturnValueOnce(chain([{ state: 'expansion' }])) + mockDbUpdate.mockReturnValue(chain([{ + attemptCount: 1, + claimToken: '00000000-0000-4000-8000-000000000011', + credentialDigest: Buffer.alloc(32, 1), + generation: '00000000-0000-4000-8000-000000000012', + legacyCredential: expectedLegacyKey, + sessionId, + }])) + let transactions = 0 + mockDbTransaction.mockImplementation(async (callback: (tx: unknown) => unknown) => { + transactions += 1 + const result = await callback(transactionClient()) + if (transactions === 2) throw new Error('simulated process loss after Redis write') + return result + }) + + await expect(getSession(fakeRequest(credential))).resolves.toEqual({ sessionId, userId: 'user-1' }) + expect(mockRedisSet).toHaveBeenCalledTimes(credentialStorageVersion === 1 ? 2 : 1) + + await destroySession(credential) + expect(mockRedisSet.mock.invocationCallOrder.at(-1)!).toBeLessThan(mockRedisDel.mock.invocationCallOrder[0]) + expect(mockRedisDel).toHaveBeenCalledWith( + expect.stringMatching(/^session:v2:/), + ...(expectedLegacyKey ? [`session:${expectedLegacyKey}`] : []), + ) + }) + + it('skips cache population when the row-lock transaction fails before a write', async () => { + const now = Date.now() + vi.setSystemTime(now) + mockDbSelect.mockReturnValue(chain([{ + sessionId: '00000000-0000-4000-8000-000000000010', userId: 'user-1', + lastSeenAt: new Date(now - 30_000), expiresAt: new Date(now + 60_000), + revokedAt: null, credentialStorageVersion: 2, databaseNow: new Date(now), + }])) + let transactions = 0 + mockDbTransaction.mockImplementation(async (callback: (tx: unknown) => unknown) => { + transactions += 1 + if (transactions === 2) throw new Error('session row lock unavailable') + return callback(transactionClient()) + }) + + await expect(getSession(fakeRequest('00000000-0000-4000-8000-000000000000'))).resolves.toEqual({ + sessionId: '00000000-0000-4000-8000-000000000010', userId: 'user-1', + }) + expect(mockRedisSet).not.toHaveBeenCalled() + }) + it('does not write a legacy cache when the sliding-refresh transaction fails to commit', async () => { const now = Date.now() vi.setSystemTime(now) @@ -609,6 +695,16 @@ describe('getSession — write-behind logic', () => { credentialStorageVersion: 1, databaseNow: new Date(now), }])) + .mockReturnValueOnce(chain([{ + sessionId: '00000000-0000-4000-8000-000000000010', + userId: 'user-1', + lastSeenAt: new Date(now), + expiresAt: new Date(now + 604_800_000), + revokedAt: null, + credentialStorageVersion: 1, + databaseNow: new Date(now), + }])) + .mockReturnValueOnce(chain([{ state: 'expansion' }])) mockDbTransaction.mockImplementationOnce(async (callback: (tx: unknown) => unknown) => { await callback(transactionClient()) throw new Error('commit failed') @@ -636,6 +732,16 @@ describe('getSession — write-behind logic', () => { credentialStorageVersion: 1, databaseNow: new Date(now), }])) + .mockReturnValueOnce(chain([{ + sessionId: '00000000-0000-4000-8000-000000000010', + userId: 'user-1', + lastSeenAt: new Date(now), + expiresAt: new Date(now + 604_800_000), + revokedAt: null, + credentialStorageVersion: 1, + databaseNow: new Date(now), + }])) + .mockReturnValueOnce(chain([{ state: 'expansion' }])) let committed = false mockDbTransaction.mockImplementationOnce(async (callback: (tx: unknown) => unknown) => { const result = await callback(transactionClient()) @@ -663,7 +769,14 @@ describe('destroySession', () => { beforeEach(() => { vi.clearAllMocks() mockRedisDel.mockResolvedValue(1) - mockDbUpdate.mockReturnValue(chain(undefined)) + mockDbUpdate.mockReturnValue(chain([{ + attemptCount: 1, + claimToken: '00000000-0000-4000-8000-000000000011', + credentialDigest: Buffer.alloc(32, 1), + generation: '00000000-0000-4000-8000-000000000012', + legacyCredential: '00000000-0000-4000-8000-000000000000', + sessionId: '00000000-0000-4000-8000-000000000010', + }])) }) it('deletes both digest and legacy Redis keys after DB revocation', async () => { @@ -677,7 +790,16 @@ describe('destroySession', () => { it('sets revokedAt in the DB', async () => { await destroySession('00000000-0000-4000-8000-000000000000') - expect(mockDbUpdate).toHaveBeenCalledOnce() + expect(mockDbUpdate).toHaveBeenCalledTimes(2) + }) + + it('keeps the durable purge pending when Redis deletion fails', async () => { + mockRedisDel.mockRejectedValueOnce(new Error('redis unavailable')) + + await expect(destroySession('00000000-0000-4000-8000-000000000000')).resolves.toBeUndefined() + + expect(mockDbUpdate).toHaveBeenCalledTimes(2) + expect(mockRedisDel).toHaveBeenCalledOnce() }) }) @@ -698,7 +820,7 @@ describe('login/finish — clone detection', () => { mockDbUpdate.mockReturnValue(chain([])) }) - it('does NOT return 403 when newCounter=0 and storedCounter=0 (counter-0 passkeys are exempt)', async () => { + it('creates a session when newCounter=0 and storedCounter=0 (counter-0 passkeys are exempt)', async () => { const storedCredential = { id: 'cred-row-id', credentialId: 'abc123', @@ -712,6 +834,7 @@ describe('login/finish — clone detection', () => { mockDbSelect .mockReturnValueOnce(chain([storedCredential])) .mockReturnValueOnce(chain([user])) + .mockReturnValueOnce(chain([{ state: 'expansion' }])) mockVerifyAuthenticationResponse.mockResolvedValue({ verified: true, @@ -730,8 +853,9 @@ describe('login/finish — clone detection', () => { }) const res = await POST(req as never) - // Counter-0 passkeys are exempt — must NOT be 403 - expect(res.status).not.toBe(403) + expect(res.status).toBe(200) + await expect(res.json()).resolves.toEqual({ userId: 'user-1', displayName: 'Alice' }) + expect(res.headers.get('set-cookie')).toMatch(/^forge_session=[0-9a-f-]{36};/) }) it('returns 403 when newCounter < storedCounter and storedCounter > 0', async () => { diff --git a/web/__tests__/epic-172-s4-context.test.ts b/web/__tests__/epic-172-s4-context.test.ts index 8c5c9f86..690da623 100644 --- a/web/__tests__/epic-172-s4-context.test.ts +++ b/web/__tests__/epic-172-s4-context.test.ts @@ -424,6 +424,11 @@ describe('Epic 172 S4 PostgreSQL CI contract', () => { expect(s4Migration).toContain('session_credential_reconciliation') expect(s4Migration).toContain("state IN ('expansion','draining','strict')") expect(s4Migration).toContain('sessions_credential_cutover_guard_v1') + expect(s4Migration).toContain('sessions_cache_purge_state_chk') + expect(s4Migration).toContain('sessions_cache_purge_due_idx') + expect(s4Migration).toContain('cache_purge_credential_digest_v1 bytea') + expect(s4Migration).toContain('cache_purge_credential_digest_v1 IS NOT NULL') + expect(s4Migration).toContain('pg_catalog.octet_length(cache_purge_credential_digest_v1) = 32') expect(s4Migration).not.toContain("last_seen_at + interval '7 days'") expect(s4Migration).not.toContain('ALTER COLUMN credential_digest_v1 SET NOT NULL') expect(s4Migration).not.toContain('ALTER COLUMN expires_at SET NOT NULL') @@ -443,6 +448,7 @@ describe('Epic 172 S4 PostgreSQL CI contract', () => { expect(sessionReconciliation).toContain('Strict session cutover zero-scan failed') expect(sessionReconciliation).toContain('Strict session cutover Redis zero-scan failed') expect(sessionReconciliation).toContain('alter column credential_digest_v1 set not null') + expect(sessionReconciliation).toContain('validate constraint sessions_cache_purge_state_chk') expect(sessionReconciliation).toContain('alter column expires_at set not null') }) }) diff --git a/web/__tests__/epic-172-s4-postgres.test.ts b/web/__tests__/epic-172-s4-postgres.test.ts index 581da797..3bba4a46 100644 --- a/web/__tests__/epic-172-s4-postgres.test.ts +++ b/web/__tests__/epic-172-s4-postgres.test.ts @@ -1,6 +1,6 @@ import { createHash, randomBytes, randomUUID } from 'node:crypto' import postgres from 'postgres' -import { afterAll, beforeAll, describe, expect, it } from 'vitest' +import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest' import { bindArchitectReplanContext, executableReferenceForEntry, @@ -11,6 +11,32 @@ import { import { architectReplanReferenceForEntry } from '@/lib/mcps/architect-plan-entries' import { computeCredentialDigest } from '@/lib/session-credential-digest' import { appendArchitectClarificationAnswer, readArchitectPlanHistory } from '@/lib/mcps/history-reader' +import { hashPassword } from '@/lib/password' +import { closeDb } from '@/db' + +const { + mockRedisDel, + mockRedisEval, + mockRedisExpire, + mockRedisIncr, + mockRedisSet, +} = vi.hoisted(() => ({ + mockRedisDel: vi.fn(), + mockRedisEval: vi.fn(), + mockRedisExpire: vi.fn(), + mockRedisIncr: vi.fn(), + mockRedisSet: vi.fn(), +})) + +vi.mock('@/lib/redis', () => ({ + redis: { + del: mockRedisDel, + eval: mockRedisEval, + expire: mockRedisExpire, + incr: mockRedisIncr, + set: mockRedisSet, + }, +})) const adminUrl = process.env.FORGE_S4_POSTGRES_TEST_DATABASE_URL?.trim() const issuerUrl = process.env.FORGE_PACKET_ISSUER_DATABASE_URL?.trim() @@ -62,11 +88,13 @@ describe.skipIf(!enabled)('Epic 172 S4 PostgreSQL boundaries', () => { let admin: ReturnType let app: ReturnType let issuer: ReturnType + const previousDatabaseUrl = process.env.DATABASE_URL beforeAll(async () => { process.env.FORGE_ARCHITECT_PLAN_WRITER_DATABASE_URL = writerUrl! process.env.FORGE_ARCHITECT_PLAN_RESOLVER_DATABASE_URL = resolverUrl! process.env.FORGE_ARCHITECT_PLAN_HISTORY_READER_DATABASE_URL = historyReaderUrl! + process.env.DATABASE_URL = appUrl! admin = postgres(adminUrl!, { max: 1, onnotice: () => {} }) app = postgres(appUrl!, { max: 1, onnotice: () => {} }) issuer = postgres(issuerUrl!, { max: 2, onnotice: () => {} }) @@ -209,6 +237,9 @@ describe.skipIf(!enabled)('Epic 172 S4 PostgreSQL boundaries', () => { where singleton_id = 'epic-172' ` } + await closeDb() + if (previousDatabaseUrl === undefined) delete process.env.DATABASE_URL + else process.env.DATABASE_URL = previousDatabaseUrl await Promise.all([admin?.end({ timeout: 5 }), app?.end({ timeout: 5 }), issuer?.end({ timeout: 5 })]) }) @@ -827,6 +858,199 @@ describe.skipIf(!enabled)('Epic 172 S4 PostgreSQL boundaries', () => { await runStatefulHistoryProof() }) + it('serves protected Architect history through the real password session route with PostgreSQL as authority', async () => { + const ownerPassword = 'route-history-password' + const routeProject = randomUUID() + const routeTask = randomUUID() + const routeRun = randomUUID() + const otherUser = randomUUID() + const otherProject = randomUUID() + const otherTask = randomUUID() + const otherRun = randomUUID() + const digestKey = randomBytes(32) + + // Redis is unavailable and contains forged-looking state. The route must + // still use only the locked PostgreSQL session decision. + mockRedisDel.mockResolvedValue(0) + mockRedisEval.mockResolvedValue(['{"userId":"forged"}', Date.now() + 60_000, 0, 0]) + mockRedisExpire.mockResolvedValue(1) + mockRedisIncr.mockResolvedValue(1) + mockRedisSet.mockRejectedValue(new Error('Redis unavailable for route proof')) + + const [firstUser] = await admin<{ id: string }[]>`select id::text as id from users limit 1` + expect(firstUser?.id).toBeTruthy() + await admin`update users set password_hash = ${await hashPassword(ownerPassword)} where id = ${firstUser!.id}::uuid` + await admin.begin(async (tx) => { + await tx`insert into projects (id, name, submitted_by, grant_decision_revision, root_binding_revision) + values (${routeProject}::uuid, 'Route history project', ${firstUser!.id}::uuid, 1, 1)` + await tx`insert into tasks (id, project_id, submitted_by, title, prompt, status) + values (${routeTask}::uuid, ${routeProject}::uuid, ${firstUser!.id}::uuid, + 'Route history task', 'protected route fixture', 'running')` + await tx`insert into agent_runs (id, task_id, agent_type, model_id_used, status) + values (${routeRun}::uuid, ${routeTask}::uuid, 'architect', 'test', 'completed')` + await tx`insert into users (id, display_name) values (${otherUser}::uuid, 'Route history other user')` + await tx`insert into projects (id, name, submitted_by, grant_decision_revision, root_binding_revision) + values (${otherProject}::uuid, 'Route history other project', ${otherUser}::uuid, 1, 1)` + await tx`insert into tasks (id, project_id, submitted_by, title, prompt, status) + values (${otherTask}::uuid, ${otherProject}::uuid, ${otherUser}::uuid, + 'Route history other task', 'protected other-user fixture', 'running')` + await tx`insert into agent_runs (id, task_id, agent_type, model_id_used, status) + values (${otherRun}::uuid, ${otherTask}::uuid, 'architect', 'test', 'completed')` + }) + await recordArchitectPlanVersion({ + agentRunId: routeRun, + digestKey, + digestKeyId: 'route-history-key', + planVersion: '1', + taskId: routeTask, + entries: [{ + agent: null, bindingFingerprint: null, content: 'Route-visible protected plan.', + entryId: 'plan_body:000000', entryKind: 'plan_body', projectionEligible: false, requirementKey: null, + }, { + agent: null, bindingFingerprint: null, + content: JSON.stringify({ requirementKey: 'plan-policy', schemaVersion: 1 }), + entryId: 'requirement:plan-policy', entryKind: 'requirement', projectionEligible: false, requirementKey: 'plan-policy', + }], + }) + await recordArchitectPlanVersion({ + agentRunId: otherRun, + digestKey, + digestKeyId: 'route-history-key', + planVersion: '1', + taskId: otherTask, + entries: [{ + agent: null, bindingFingerprint: null, content: 'Other-user protected plan.', + entryId: 'plan_body:000000', entryKind: 'plan_body', projectionEligible: false, requirementKey: null, + }, { + agent: null, bindingFingerprint: null, + content: JSON.stringify({ requirementKey: 'plan-policy', schemaVersion: 1 }), + entryId: 'requirement:plan-policy', entryKind: 'requirement', projectionEligible: false, requirementKey: 'plan-policy', + }], + }) + + const { POST: passwordLogin } = await import('@/app/api/auth/login/password/route') + const { GET: protectedHistory } = await import('@/app/api/tasks/[id]/architect-plan-history/[planVersion]/route') + const { createSession } = await import('@/lib/session') + const passwordSession = async (): Promise => { + const response = await passwordLogin(new Request('http://localhost/api/auth/login/password', { + method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ password: ownerPassword }), + }) as never) + expect(response.status).toBe(200) + const match = response.headers.get('set-cookie')?.match(/(?:^|,\s*)forge_session=([^;]+)/) + expect(match?.[1]).toMatch(/^[0-9a-f-]{36}$/) + return match![1] + } + const routeRead = async (credential: string, taskId: string, planVersion: string) => protectedHistory( + new Request(`http://localhost/api/tasks/${taskId}/architect-plan-history/${planVersion}`, { + headers: { cookie: `forge_session=${credential}` }, + }) as never, + { params: Promise.resolve({ id: taskId, planVersion }) }, + ) + const auditCount = async (taskId: string) => { + const [row] = await admin<{ count: number }[]>` + select count(*)::integer as count from architect_plan_history_reads where task_id = ${taskId}::uuid + ` + return row.count + } + + const liveCredential = await passwordSession() + const successful = await routeRead(liveCredential, routeTask, '1') + expect(successful.status).toBe(200) + const successfulBody = await successful.json() as { taskId: string; planVersion: string; entries: Array<{ entryId: string; content: string }> } + expect(successfulBody).toEqual(expect.objectContaining({ taskId: routeTask, planVersion: '1' })) + expect(successfulBody.entries).toEqual(expect.arrayContaining([ + expect.objectContaining({ entryId: 'plan_body:000000', content: 'Route-visible protected plan.' }), + ])) + const responseText = JSON.stringify(successfulBody) + expect(responseText).not.toContain(liveCredential) + expect(responseText).not.toMatch(/credential|storage|session(?:_|-)?id/i) + const [successAudit] = await admin<{ count: number; digest: string; returned: number }[]>` + select count(*)::integer as count, max(entry_set_digest) as digest, + max(returned_entry_count)::integer as returned + from architect_plan_history_reads where task_id = ${routeTask}::uuid + ` + expect(successAudit).toEqual({ + count: 1, + digest: expect.stringMatching(/^sha256:[0-9a-f]{64}$/), + returned: successfulBody.entries.length, + }) + const [auditTextColumns] = await admin<{ count: number }[]>` + select count(*)::integer as count + from information_schema.columns + where table_schema = 'public' and table_name = 'architect_plan_history_reads' + and column_name in ('content', 'credential', 'credential_digest', 'storage_locator') + ` + expect(auditTextColumns).toEqual({ count: 0 }) + + const assertSessionDenied = async (credential: string) => { + const before = await auditCount(routeTask) + const response = await routeRead(credential, routeTask, '1') + expect(response.status).toBe(401) + await expect(response.json()).resolves.toEqual({ error: 'Unauthorized' }) + expect(await auditCount(routeTask)).toBe(before) + } + const revokedCredential = await passwordSession() + await admin`update sessions set revoked_at = clock_timestamp() + where credential_digest_v1 = ${computeCredentialDigest(revokedCredential).digest}::bytea` + await assertSessionDenied(revokedCredential) + + const expiryBoundaryCredential = await passwordSession() + await admin`update sessions set expires_at = clock_timestamp() + where credential_digest_v1 = ${computeCredentialDigest(expiryBoundaryCredential).digest}::bytea` + await assertSessionDenied(expiryBoundaryCredential) + + const rotatedOldCredential = await passwordSession() + const rotatedCurrentCredential = await passwordSession() + await admin`update sessions set revoked_at = clock_timestamp() + where credential_digest_v1 = ${computeCredentialDigest(rotatedOldCredential).digest}::bytea` + await assertSessionDenied(rotatedOldCredential) + const [rotatedCurrent] = await admin<{ live: boolean }[]>` + select revoked_at is null and expires_at > clock_timestamp() as live from sessions + where credential_digest_v1 = ${computeCredentialDigest(rotatedCurrentCredential).digest}::bytea + ` + expect(rotatedCurrent).toEqual({ live: true }) + + await assertSessionDenied(randomUUID()) + + // The password route is intentionally single-user today, so create the + // second user's canonical database-authoritative session through the same + // production session boundary that the route later validates. + const otherCredential = await createSession(otherUser, null, { ip: null, userAgent: null }) + const otherSuccessful = await routeRead(otherCredential, otherTask, '1') + expect(otherSuccessful.status).toBe(200) + const otherSuccessfulBody = await otherSuccessful.json() as { taskId: string; planVersion: string; entries: Array<{ entryId: string; content: string }> } + expect(otherSuccessfulBody).toEqual(expect.objectContaining({ taskId: otherTask, planVersion: '1' })) + expect(otherSuccessfulBody.entries).toEqual(expect.arrayContaining([ + expect.objectContaining({ entryId: 'plan_body:000000', content: 'Other-user protected plan.' }), + ])) + const [otherSuccessAudit] = await admin<{ count: number; digest: string; returned: number }[]>` + select count(*)::integer as count, max(entry_set_digest) as digest, + max(returned_entry_count)::integer as returned + from architect_plan_history_reads where task_id = ${otherTask}::uuid + ` + expect(otherSuccessAudit).toEqual({ + count: 1, + digest: expect.stringMatching(/^sha256:[0-9a-f]{64}$/), + returned: otherSuccessfulBody.entries.length, + }) + + const crossTaskCredential = await passwordSession() + const beforeCrossTaskOwner = await auditCount(routeTask) + const beforeCrossTaskOther = await auditCount(otherTask) + const crossTask = await routeRead(crossTaskCredential, otherTask, '1') + expect(crossTask.status).toBe(404) + await expect(crossTask.json()).resolves.toEqual({ error: 'Architect plan history not found.' }) + expect(await auditCount(routeTask)).toBe(beforeCrossTaskOwner) + expect(await auditCount(otherTask)).toBe(beforeCrossTaskOther) + + const beforeWrongVersion = await auditCount(routeTask) + const wrongVersion = await routeRead(crossTaskCredential, routeTask, '2') + expect(wrongVersion.status).toBe(404) + await expect(wrongVersion.json()).resolves.toEqual({ error: 'Architect plan history not found.' }) + expect(await auditCount(routeTask)).toBe(beforeWrongVersion) + expect(mockRedisSet).toHaveBeenCalled() + }) + it('resume-safely rekeys a crash-window legacy session and leaves no raw-id lookup target', async () => { const legacyCredential = randomUUID() const legacyUser = randomUUID() @@ -874,6 +1098,162 @@ describe.skipIf(!enabled)('Epic 172 S4 PostgreSQL boundaries', () => { expect(proof).toEqual({ digestRows: 1, rawIdRows: 0, retainedRawIds: 0 }) }) + it('uses production logout and claim paths for v0, v1, and v2 session cache purges', async () => { + const { destroySession, reconcilePendingSessionCacheInvalidations } = await import('@/lib/session') + for (const invalidDigest of [null, Buffer.alloc(31, 1)]) { + const invalidSession = randomUUID() + const liveDigest = computeCredentialDigest(randomUUID()).digest + await admin` + insert into sessions (id, user_id, credential_digest_v1, expires_at, credential_storage_version) + values (${invalidSession}::uuid, ${ids.user}::uuid, ${liveDigest}::bytea, + clock_timestamp() + interval '7 days', 2) + ` + await expect(admin` + update sessions + set cache_purge_pending_at = clock_timestamp(), + cache_purge_credential_digest_v1 = ${invalidDigest}::bytea, + cache_purge_generation = gen_random_uuid() + where id = ${invalidSession}::uuid + `).rejects.toMatchObject({ code: '23514' }) + } + for (const storageVersion of [0, 1, 2] as const) { + const credential = randomUUID() + const digest = computeCredentialDigest(credential).digest + const sessionId = storageVersion < 2 ? credential : randomUUID() + if (storageVersion === 0) { + await admin`insert into sessions (id, user_id, credential_storage_version) + values (${sessionId}::uuid, ${ids.user}::uuid, 0)` + } else { + await admin` + insert into sessions (id, user_id, credential_digest_v1, expires_at, credential_storage_version) + values (${sessionId}::uuid, ${ids.user}::uuid, ${digest}::bytea, + clock_timestamp() + interval '7 days', ${storageVersion}) + ` + } + + mockRedisDel.mockRejectedValueOnce(new Error('test redis outage')) + await destroySession(credential) + const [pending] = await admin<{ + digestMatches: boolean + legacy: boolean + pending: boolean + revoked: boolean + }[]>` + select cache_purge_credential_digest_v1 = ${digest}::bytea as "digestMatches", + credential_storage_version < 2 as legacy, + cache_purge_pending_at is not null as pending, + revoked_at is not null as revoked + from sessions where id = ${sessionId}::uuid + ` + expect(pending).toEqual({ digestMatches: true, legacy: storageVersion < 2, pending: true, revoked: true }) + + await admin`update sessions set cache_purge_next_attempt_at = clock_timestamp() + where id = ${sessionId}::uuid` + mockRedisDel.mockResolvedValueOnce(0) + await expect(reconcilePendingSessionCacheInvalidations(1)).resolves.toEqual({ + claimed: 1, completed: 1, deferred: 0, stale: 0, + }) + const [completed] = await admin<{ + completed: boolean + digestCleared: boolean + pending: boolean + }[]>` + select cache_purge_completed_at is not null as completed, + cache_purge_credential_digest_v1 is null as "digestCleared", + cache_purge_pending_at is not null as pending + from sessions where id = ${sessionId}::uuid + ` + expect(completed).toEqual({ completed: true, digestCleared: true, pending: false }) + } + + const expiredClaimCredential = randomUUID() + const expiredClaimDigest = computeCredentialDigest(expiredClaimCredential).digest + const expiredClaimSession = randomUUID() + await admin` + insert into sessions (id, user_id, credential_digest_v1, expires_at, credential_storage_version) + values (${expiredClaimSession}::uuid, ${ids.user}::uuid, ${expiredClaimDigest}::bytea, + clock_timestamp() + interval '7 days', 2) + ` + mockRedisDel.mockRejectedValueOnce(new Error('test redis outage')) + await destroySession(expiredClaimCredential) + await admin` + update sessions set cache_purge_claim_token = gen_random_uuid(), + cache_purge_claim_expires_at = clock_timestamp() - interval '1 second', + cache_purge_next_attempt_at = clock_timestamp() + where id = ${expiredClaimSession}::uuid + ` + mockRedisDel.mockResolvedValueOnce(0) + const [first, second] = await Promise.all([ + reconcilePendingSessionCacheInvalidations(1), + reconcilePendingSessionCacheInvalidations(1), + ]) + expect([first.claimed, second.claimed].sort()).toEqual([0, 1]) + expect(first.completed + second.completed).toBe(1) + }) + + it.each([ + ['v2', 2], + ['dual-write v1', 1], + ])('holds the %s session row through cache population so logout waits and then purges it', async ( + _label, + credentialStorageVersion, + ) => { + const { destroySession, getSession } = await import('@/lib/session') + const credential = randomUUID() + const digest = computeCredentialDigest(credential).digest + const sessionId = credentialStorageVersion === 1 ? credential : randomUUID() + mockRedisDel.mockClear() + mockRedisSet.mockClear() + await admin` + insert into sessions (id, user_id, credential_digest_v1, expires_at, credential_storage_version) + values (${sessionId}::uuid, ${ids.user}::uuid, ${digest}::bytea, + clock_timestamp() + interval '7 days', ${credentialStorageVersion}) + ` + + let releaseCacheWrite: (() => void) | undefined + const cacheWriteStarted = new Promise((resolve) => { + mockRedisSet.mockImplementationOnce(async () => { + resolve() + await new Promise((release) => { releaseCacheWrite = release }) + return 'OK' + }) + }) + mockRedisDel.mockResolvedValueOnce(0) + const read = getSession(new Request('http://localhost/', { + headers: { cookie: `forge_session=${credential}` }, + })) + await cacheWriteStarted + let revokeSettled = false + const revoke = destroySession(credential).then(() => { revokeSettled = true }) + const waitForBlockedLogout = async () => { + for (let attempt = 0; attempt < 40; attempt += 1) { + const [blocked] = await admin<{ count: number }[]>` + select count(*)::integer as count + from pg_catalog.pg_stat_activity + where datname = current_database() + and wait_event_type = 'Lock' + and query like 'update "sessions" set%' + ` + if (blocked?.count === 1) return + await new Promise((resolve) => setTimeout(resolve, 25)) + } + throw new Error('logout did not block on the cache-write session row lock') + } + try { + await waitForBlockedLogout() + expect(revokeSettled).toBe(false) + } finally { + releaseCacheWrite?.() + } + await expect(read).resolves.toEqual({ sessionId, userId: ids.user }) + await expect(revoke).resolves.toBeUndefined() + expect(mockRedisSet).toHaveBeenCalledTimes(credentialStorageVersion === 1 ? 2 : 1) + expect(mockRedisDel).toHaveBeenCalledWith( + expect.stringMatching(/^session:v2:/), + ...(credentialStorageVersion === 1 ? [`session:${credential}`] : []), + ) + }) + it('allow-once-single-winner: atomically keeps one audit and one nonce claim', async () => { const packageId = randomUUID() const decisionId = randomUUID() diff --git a/web/__tests__/session-cache-invalidation.test.ts b/web/__tests__/session-cache-invalidation.test.ts new file mode 100644 index 00000000..c8a280b8 --- /dev/null +++ b/web/__tests__/session-cache-invalidation.test.ts @@ -0,0 +1,131 @@ +import { describe, expect, it, vi } from 'vitest' +import { + reconcileSessionCacheInvalidations, + sessionCachePurgeBackoffSeconds, + sessionCachePurgeKeys, + type SessionCachePurgeTarget, +} from '@/lib/session-cache-invalidation' + +const digest = 'a'.repeat(64) + +function target(overrides: Partial = {}): SessionCachePurgeTarget { + return { + attemptCount: 1, + claimToken: 'claim-1', + credentialDigestHex: digest, + generation: 'generation-1', + legacyCredential: null, + sessionId: 'session-1', + ...overrides, + } +} + +describe('session cache invalidation', () => { + it('derives only the v2 key, plus an existing v1 legacy key when supplied', () => { + expect(sessionCachePurgeKeys(target())).toEqual([`session:v2:${digest}`]) + expect(sessionCachePurgeKeys(target({ legacyCredential: 'legacy-id' }))).toEqual([ + `session:v2:${digest}`, + 'session:legacy-id', + ]) + expect(sessionCachePurgeKeys(target({ credentialDigestHex: 'not-a-digest' }))).toEqual([]) + }) + + it('completes a claimed invalidation after idempotent Redis deletion', async () => { + const claimed = target({ legacyCredential: 'legacy-id' }) + const deleteKeys = vi.fn().mockResolvedValue(2) + const complete = vi.fn().mockResolvedValue(true) + const defer = vi.fn().mockResolvedValue(true) + + await expect(reconcileSessionCacheInvalidations({ + limit: 100, + deleteKeys, + store: { claimDue: vi.fn().mockResolvedValue([claimed]), complete, defer }, + })).resolves.toEqual({ claimed: 1, completed: 1, deferred: 0, stale: 0 }) + + expect(deleteKeys).toHaveBeenCalledWith([`session:v2:${digest}`, 'session:legacy-id']) + expect(complete).toHaveBeenCalledWith(claimed) + expect(defer).not.toHaveBeenCalled() + }) + + it('keeps failed Redis deletion retryable and safely completes a later claim', async () => { + const first = target({ claimToken: 'claim-first', attemptCount: 1 }) + const recovered = target({ claimToken: 'claim-recovered', attemptCount: 2 }) + const deleteKeys = vi.fn() + .mockRejectedValueOnce(new Error('redis unavailable')) + .mockResolvedValueOnce(0) + const complete = vi.fn().mockResolvedValue(true) + const defer = vi.fn().mockResolvedValue(true) + const claimDue = vi.fn() + .mockResolvedValueOnce([first]) + .mockResolvedValueOnce([recovered]) + + await expect(reconcileSessionCacheInvalidations({ + limit: 1, + deleteKeys, + store: { claimDue, complete, defer }, + })).resolves.toEqual({ claimed: 1, completed: 0, deferred: 1, stale: 0 }) + expect(defer).toHaveBeenCalledWith(first) + expect(complete).not.toHaveBeenCalled() + + await expect(reconcileSessionCacheInvalidations({ + limit: 1, + deleteKeys, + store: { claimDue, complete, defer }, + })).resolves.toEqual({ claimed: 1, completed: 1, deferred: 0, stale: 0 }) + expect(complete).toHaveBeenCalledWith(recovered) + expect(deleteKeys).toHaveBeenCalledTimes(2) + }) + + it('uses bounded exponential retry backoff', () => { + expect(sessionCachePurgeBackoffSeconds(1)).toBe(1) + expect(sessionCachePurgeBackoffSeconds(2)).toBe(2) + expect(sessionCachePurgeBackoffSeconds(9)).toBe(256) + expect(sessionCachePurgeBackoffSeconds(10)).toBe(300) + }) + + it('does not complete malformed targets or use them as cache authority', async () => { + const malformed = target({ credentialDigestHex: 'not-a-digest' }) + const deleteKeys = vi.fn() + const complete = vi.fn() + const defer = vi.fn().mockResolvedValue(true) + + await expect(reconcileSessionCacheInvalidations({ + limit: 1, + deleteKeys, + store: { claimDue: vi.fn().mockResolvedValue([malformed]), complete, defer }, + })).resolves.toEqual({ claimed: 1, completed: 0, deferred: 1, stale: 0 }) + + expect(deleteKeys).not.toHaveBeenCalled() + expect(complete).not.toHaveBeenCalled() + expect(defer).toHaveBeenCalledWith(malformed) + }) + + it('reports a reclaimed claim as stale instead of inventing completion', async () => { + const claimed = target({ claimToken: 'expired-claim' }) + const deleteKeys = vi.fn().mockResolvedValue(0) + const complete = vi.fn().mockResolvedValue(false) + const defer = vi.fn().mockResolvedValue(false) + + await expect(reconcileSessionCacheInvalidations({ + limit: 1, + deleteKeys, + store: { claimDue: vi.fn().mockResolvedValue([claimed]), complete, defer }, + })).resolves.toEqual({ claimed: 1, completed: 0, deferred: 0, stale: 1 }) + expect(defer).not.toHaveBeenCalled() + }) + + it('reports a late failed worker as stale when its defer claim was reclaimed', async () => { + const claimed = target({ claimToken: 'expired-claim' }) + const deleteKeys = vi.fn().mockRejectedValue(new Error('redis unavailable')) + const complete = vi.fn() + const defer = vi.fn().mockResolvedValue(false) + + await expect(reconcileSessionCacheInvalidations({ + limit: 1, + deleteKeys, + store: { claimDue: vi.fn().mockResolvedValue([claimed]), complete, defer }, + })).resolves.toEqual({ claimed: 1, completed: 0, deferred: 0, stale: 1 }) + expect(complete).not.toHaveBeenCalled() + expect(defer).toHaveBeenCalledWith(claimed) + }) +}) diff --git a/web/__tests__/worker-retry-contract.test.ts b/web/__tests__/worker-retry-contract.test.ts index 382d2a8e..846923bb 100644 --- a/web/__tests__/worker-retry-contract.test.ts +++ b/web/__tests__/worker-retry-contract.test.ts @@ -214,6 +214,17 @@ describe('answered-question retry contract', () => { ) }) + it('schedules session cache purges independently of blocked handoff recovery', () => { + const runtimeSource = fs.readFileSync(path.join(repoRoot, 'worker/runtime.ts'), 'utf8') + + expect(runtimeSource).toContain('FORGE_SESSION_CACHE_PURGE_INTERVAL_SECONDS') + expect(runtimeSource).toContain('const sweepSessionCachePurges') + expect(runtimeSource).toContain("import('../lib/session')") + expect(runtimeSource).toContain("void sweepSessionCachePurges({ startup: true })") + expect(runtimeSource).toContain('Session cache purge sweep failed') + expect(runtimeSource).toContain('clearInterval(sessionCachePurgeTimer)') + }) + it('restores answered rows and awaiting_answers status on retryable re-plan failure', async () => { const task = { id: 'task-answers', diff --git a/web/db/migrations/0027_epic_172_s4_packet_context.sql b/web/db/migrations/0027_epic_172_s4_packet_context.sql index d5e48ef0..b6ce989c 100644 --- a/web/db/migrations/0027_epic_172_s4_packet_context.sql +++ b/web/db/migrations/0027_epic_172_s4_packet_context.sql @@ -56,7 +56,15 @@ ALTER TABLE public.sessions ADD COLUMN IF NOT EXISTS expires_at timestamptz, ADD COLUMN IF NOT EXISTS credential_storage_version integer NOT NULL DEFAULT 0, ADD COLUMN IF NOT EXISTS legacy_redis_purge_pending_at timestamptz, - ADD COLUMN IF NOT EXISTS legacy_redis_invalidated_at timestamptz; + ADD COLUMN IF NOT EXISTS legacy_redis_invalidated_at timestamptz, + ADD COLUMN IF NOT EXISTS cache_purge_pending_at timestamptz, + ADD COLUMN IF NOT EXISTS cache_purge_credential_digest_v1 bytea, + ADD COLUMN IF NOT EXISTS cache_purge_generation uuid, + ADD COLUMN IF NOT EXISTS cache_purge_claim_token uuid, + ADD COLUMN IF NOT EXISTS cache_purge_claim_expires_at timestamptz, + ADD COLUMN IF NOT EXISTS cache_purge_attempt_count integer NOT NULL DEFAULT 0, + ADD COLUMN IF NOT EXISTS cache_purge_next_attempt_at timestamptz, + ADD COLUMN IF NOT EXISTS cache_purge_completed_at timestamptz; --> statement-breakpoint ALTER TABLE public.agent_runs ADD COLUMN provider_type_used text, @@ -271,7 +279,37 @@ ALTER TABLE public.sessions OR pg_catalog.octet_length(credential_digest_v1) = 32 ) NOT VALID, ADD CONSTRAINT sessions_credential_storage_version_chk - CHECK (credential_storage_version IN (0,1,2)) NOT VALID; + CHECK (credential_storage_version IN (0,1,2)) NOT VALID, + ADD CONSTRAINT sessions_cache_purge_state_chk CHECK ( + cache_purge_attempt_count >= 0 + AND ( + (cache_purge_pending_at IS NULL + AND cache_purge_credential_digest_v1 IS NULL + AND cache_purge_generation IS NULL + AND cache_purge_claim_token IS NULL + AND cache_purge_claim_expires_at IS NULL + AND cache_purge_next_attempt_at IS NULL + AND cache_purge_completed_at IS NULL) + OR + (cache_purge_pending_at IS NOT NULL + AND cache_purge_credential_digest_v1 IS NOT NULL + AND pg_catalog.octet_length(cache_purge_credential_digest_v1) = 32 + AND cache_purge_generation IS NOT NULL + AND cache_purge_completed_at IS NULL) + OR + (cache_purge_pending_at IS NULL + AND cache_purge_credential_digest_v1 IS NULL + AND cache_purge_generation IS NOT NULL + AND cache_purge_claim_token IS NULL + AND cache_purge_claim_expires_at IS NULL + AND cache_purge_next_attempt_at IS NULL + AND cache_purge_completed_at IS NOT NULL) + ) + ) NOT VALID; +--> statement-breakpoint +CREATE INDEX sessions_cache_purge_due_idx ON public.sessions + (cache_purge_next_attempt_at, cache_purge_pending_at, id) + WHERE cache_purge_pending_at IS NOT NULL; --> statement-breakpoint CREATE UNIQUE INDEX sessions_credential_digest_v1_idx ON public.sessions (credential_digest_v1) diff --git a/web/db/schema.ts b/web/db/schema.ts index 63566ef1..d5099356 100644 --- a/web/db/schema.ts +++ b/web/db/schema.ts @@ -100,6 +100,14 @@ export const sessions = pgTable( credentialStorageVersion: integer('credential_storage_version').notNull().default(0), legacyRedisPurgePendingAt: timestamp('legacy_redis_purge_pending_at', tsOpts), legacyRedisInvalidatedAt: timestamp('legacy_redis_invalidated_at', tsOpts), + cachePurgePendingAt: timestamp('cache_purge_pending_at', tsOpts), + cachePurgeCredentialDigestV1: bytea('cache_purge_credential_digest_v1'), + cachePurgeGeneration: uuid('cache_purge_generation'), + cachePurgeClaimToken: uuid('cache_purge_claim_token'), + cachePurgeClaimExpiresAt: timestamp('cache_purge_claim_expires_at', tsOpts), + cachePurgeAttemptCount: integer('cache_purge_attempt_count').notNull().default(0), + cachePurgeNextAttemptAt: timestamp('cache_purge_next_attempt_at', tsOpts), + cachePurgeCompletedAt: timestamp('cache_purge_completed_at', tsOpts), }, (t) => [ index('sessions_user_id_idx').on(t.userId), @@ -107,6 +115,35 @@ export const sessions = pgTable( uniqueIndex('sessions_credential_digest_v1_idx') .on(t.credentialDigestV1) .where(sql`${t.credentialDigestV1} is not null`), + index('sessions_cache_purge_due_idx') + .on(t.cachePurgeNextAttemptAt, t.cachePurgePendingAt, t.id) + .where(sql`${t.cachePurgePendingAt} is not null`), + check('sessions_cache_purge_state_chk', sql` + ${t.cachePurgeAttemptCount} >= 0 + and ( + (${t.cachePurgePendingAt} is null + and ${t.cachePurgeCredentialDigestV1} is null + and ${t.cachePurgeGeneration} is null + and ${t.cachePurgeClaimToken} is null + and ${t.cachePurgeClaimExpiresAt} is null + and ${t.cachePurgeNextAttemptAt} is null + and ${t.cachePurgeCompletedAt} is null) + or + (${t.cachePurgePendingAt} is not null + and ${t.cachePurgeCredentialDigestV1} is not null + and octet_length(${t.cachePurgeCredentialDigestV1}) = 32 + and ${t.cachePurgeGeneration} is not null + and ${t.cachePurgeCompletedAt} is null) + or + (${t.cachePurgePendingAt} is null + and ${t.cachePurgeCredentialDigestV1} is null + and ${t.cachePurgeGeneration} is not null + and ${t.cachePurgeClaimToken} is null + and ${t.cachePurgeClaimExpiresAt} is null + and ${t.cachePurgeNextAttemptAt} is null + and ${t.cachePurgeCompletedAt} is not null) + ) + `), ], ) diff --git a/web/lib/session-cache-invalidation.ts b/web/lib/session-cache-invalidation.ts new file mode 100644 index 00000000..3f954fc9 --- /dev/null +++ b/web/lib/session-cache-invalidation.ts @@ -0,0 +1,58 @@ +export type SessionCachePurgeTarget = { + attemptCount: number + claimToken: string | null + credentialDigestHex: string + generation: string + legacyCredential: string | null + sessionId: string +} + +export type SessionCachePurgeStore = { + claimDue: (limit: number) => Promise + complete: (target: SessionCachePurgeTarget) => Promise + defer: (target: SessionCachePurgeTarget) => Promise +} + +export function sessionCachePurgeKeys(target: Pick): string[] { + if (!/^[0-9a-f]{64}$/.test(target.credentialDigestHex)) return [] + const keys = [`session:v2:${target.credentialDigestHex}`] + if (target.legacyCredential) keys.push(`session:${target.legacyCredential}`) + return keys +} + +export function sessionCachePurgeBackoffSeconds(attemptCount: number): number { + return Math.min(2 ** Math.max(attemptCount - 1, 0), 300) +} + +/** + * Claims a bounded batch of already-revoked session cache entries. Redis + * deletion is idempotent; the store is responsible for crash-recoverable claim + * expiry and for accepting completion only from the matching claim generation. + */ +export async function reconcileSessionCacheInvalidations(input: { + deleteKeys: (keys: readonly string[]) => Promise + limit: number + store: SessionCachePurgeStore +}): Promise<{ completed: number; deferred: number; claimed: number; stale: number }> { + const targets = await input.store.claimDue(input.limit) + let completed = 0 + let deferred = 0 + let stale = 0 + for (const target of targets) { + const keys = sessionCachePurgeKeys(target) + if (keys.length === 0) { + if (await input.store.defer(target)) deferred += 1 + else stale += 1 + continue + } + try { + await input.deleteKeys(keys) + if (await input.store.complete(target)) completed += 1 + else stale += 1 + } catch { + if (await input.store.defer(target)) deferred += 1 + else stale += 1 + } + } + return { claimed: targets.length, completed, deferred, stale } +} diff --git a/web/lib/session.ts b/web/lib/session.ts index dcef69ed..df0dcfc7 100644 --- a/web/lib/session.ts +++ b/web/lib/session.ts @@ -1,12 +1,18 @@ import { db } from '@/db' import { sessionCredentialReconciliation, sessions } from '@/db/schema' -import { and, eq, isNull, or, sql } from 'drizzle-orm' +import { and, eq, isNotNull, isNull, or, sql } from 'drizzle-orm' import { redis } from '@/lib/redis' import { isIP } from 'node:net' import { computeCredentialDigest, isCanonicalSessionCredential, } from '@/lib/session-credential-digest' +import { + reconcileSessionCacheInvalidations, + sessionCachePurgeBackoffSeconds, + sessionCachePurgeKeys, + type SessionCachePurgeTarget, +} from '@/lib/session-cache-invalidation' export type SessionData = { userId: string @@ -31,6 +37,8 @@ type SessionMeta = { const SESSION_TTL_SECONDS = 60 * 60 * 24 * 7 const SESSION_TTL_MS = SESSION_TTL_SECONDS * 1000 const WRITE_BEHIND_INTERVAL_MS = 60 * 1000 +const SESSION_CACHE_PURGE_CLAIM_SECONDS = 30 +const SESSION_CACHE_PURGE_BATCH_LIMIT = 100 function redisKey(digest: Buffer): string { return `session:v2:${digest.toString('hex')}` @@ -40,6 +48,144 @@ function legacyRedisKey(credential: string): string { return `session:${credential}` } +type StoredSessionCachePurge = SessionCachePurgeTarget & { + credentialDigest: Buffer +} + +function cachePurgeTarget(row: { + attemptCount: number + claimToken: string | null + credentialDigest: Buffer + generation: string + legacyCredential: string | null + sessionId: string +}): StoredSessionCachePurge | null { + if (!Buffer.isBuffer(row.credentialDigest) || row.credentialDigest.length !== 32) return null + return { + attemptCount: row.attemptCount, + claimToken: row.claimToken, + credentialDigest: row.credentialDigest, + credentialDigestHex: row.credentialDigest.toString('hex'), + generation: row.generation, + legacyCredential: row.legacyCredential, + sessionId: row.sessionId, + } +} + +async function completeSessionCachePurge(target: SessionCachePurgeTarget): Promise { + const conditions = [ + eq(sessions.id, target.sessionId), + eq(sessions.cachePurgeGeneration, target.generation), + isNotNull(sessions.cachePurgePendingAt), + ] + if (target.claimToken) conditions.push(eq(sessions.cachePurgeClaimToken, target.claimToken)) + const completed = await db.update(sessions).set({ + cachePurgeCredentialDigestV1: null, + cachePurgeClaimExpiresAt: null, + cachePurgeClaimToken: null, + cachePurgeCompletedAt: sql`pg_catalog.clock_timestamp()`, + cachePurgeNextAttemptAt: null, + cachePurgePendingAt: null, + }).where(and(...conditions)).returning({ id: sessions.id }) + return Array.isArray(completed) && completed.length === 1 +} + +async function deferSessionCachePurge(target: SessionCachePurgeTarget): Promise { + const conditions = [ + eq(sessions.id, target.sessionId), + eq(sessions.cachePurgeGeneration, target.generation), + isNotNull(sessions.cachePurgePendingAt), + ] + if (target.claimToken) conditions.push(eq(sessions.cachePurgeClaimToken, target.claimToken)) + const deferred = await db.update(sessions).set({ + cachePurgeClaimExpiresAt: null, + cachePurgeClaimToken: null, + cachePurgeNextAttemptAt: sql`pg_catalog.clock_timestamp() + ${sessionCachePurgeBackoffSeconds(target.attemptCount)} * interval '1 second'`, + }).where(and(...conditions)).returning({ id: sessions.id }) + return Array.isArray(deferred) && deferred.length === 1 +} + +async function deleteSessionCacheTarget(target: SessionCachePurgeTarget): Promise { + await redis.del(...sessionCachePurgeKeys(target)) +} + +/** + * Bounded worker maintenance for already-revoked sessions. It never reads Redis + * to authorize a request and only derives cache keys from persisted digest/legacy + * identity fields returned by the locked database row. + */ +export async function reconcilePendingSessionCacheInvalidations( + limit = SESSION_CACHE_PURGE_BATCH_LIMIT, +): Promise<{ claimed: number; completed: number; deferred: number; stale: number }> { + if (!Number.isInteger(limit) || limit < 1 || limit > SESSION_CACHE_PURGE_BATCH_LIMIT) { + throw new Error(`Session cache purge limit must be an integer from 1 to ${SESSION_CACHE_PURGE_BATCH_LIMIT}.`) + } + return reconcileSessionCacheInvalidations({ + limit, + deleteKeys: async (keys) => redis.del(...keys), + store: { + claimDue: async (batchLimit) => db.transaction(async (tx) => { + const rows = await tx.execute(sql<{ + attemptCount: number + claimToken: string + credentialDigestHex: string + generation: string + legacyCredential: string | null + sessionId: string + }>` + WITH candidates AS ( + SELECT session.id + FROM public.sessions session + WHERE session.cache_purge_pending_at IS NOT NULL + AND pg_catalog.octet_length(session.cache_purge_credential_digest_v1) = 32 + AND (session.cache_purge_next_attempt_at IS NULL + OR session.cache_purge_next_attempt_at <= pg_catalog.clock_timestamp()) + AND (session.cache_purge_claim_expires_at IS NULL + OR session.cache_purge_claim_expires_at <= pg_catalog.clock_timestamp()) + ORDER BY session.cache_purge_pending_at, session.id + LIMIT ${batchLimit} + FOR UPDATE SKIP LOCKED + ) + UPDATE public.sessions session + SET cache_purge_claim_token = pg_catalog.gen_random_uuid(), + cache_purge_claim_expires_at = pg_catalog.clock_timestamp() + + ${SESSION_CACHE_PURGE_CLAIM_SECONDS} * interval '1 second', + cache_purge_attempt_count = session.cache_purge_attempt_count + 1 + FROM candidates + WHERE session.id = candidates.id + RETURNING session.id::text AS "sessionId", + session.cache_purge_generation::text AS generation, + session.cache_purge_claim_token::text AS "claimToken", + session.cache_purge_attempt_count AS "attemptCount", + pg_catalog.encode(session.cache_purge_credential_digest_v1, 'hex') AS "credentialDigestHex", + CASE WHEN session.credential_storage_version < 2 THEN session.id::text ELSE NULL END AS "legacyCredential" + `) + return rows.flatMap((row): SessionCachePurgeTarget[] => { + if ( + typeof row.sessionId !== 'string' + || typeof row.generation !== 'string' + || typeof row.claimToken !== 'string' + || typeof row.attemptCount !== 'number' + || typeof row.credentialDigestHex !== 'string' + || !/^[0-9a-f]{64}$/.test(row.credentialDigestHex) + || (row.legacyCredential !== null && typeof row.legacyCredential !== 'string') + ) return [] + return [{ + sessionId: row.sessionId, + generation: row.generation, + claimToken: row.claimToken, + attemptCount: row.attemptCount, + credentialDigestHex: row.credentialDigestHex, + legacyCredential: row.legacyCredential, + }] + }) + }), + complete: completeSessionCachePurge, + defer: deferSessionCachePurge, + }, + }) +} + function dualWriteSessions(): boolean { const mode = process.env.FORGE_SESSION_CREDENTIAL_MODE?.trim() || 'strict' if (mode !== 'strict' && mode !== 'dual') { @@ -283,6 +429,62 @@ async function cacheAuthorizedSession( ) } +/** + * Redis is not an authority, but its writes must be ordered with revocation. + * The exact live session row stays locked while these one or two cache writes + * run: a competing logout either committed first (so this skips), or waits and + * deletes the completed keys after this transaction releases its lock. + */ +async function cacheAuthorizedSessionWhileLocked( + credential: string, + digest: Buffer, + authorized: AuthorizedSession, +): Promise { + await db.transaction(async (tx) => { + const [row] = await tx + .select({ + credentialStorageVersion: sessions.credentialStorageVersion, + databaseNow: sql`pg_catalog.clock_timestamp()`, + expiresAt: sessions.expiresAt, + lastSeenAt: sessions.lastSeenAt, + revokedAt: sessions.revokedAt, + sessionId: sessions.id, + userId: sessions.userId, + }) + .from(sessions) + .where(and( + eq(sessions.id, authorized.sessionId), + eq(sessions.credentialDigestV1, digest), + )) + .limit(1) + .for('update') + if (!row || row.revokedAt || row.userId !== authorized.userId || !row.expiresAt) return + + const expiresAt = parseDatabaseTimestamp(row.expiresAt, 'locked expiry') + if (parseDatabaseTimestamp(row.databaseNow, 'locked clock') >= expiresAt) return + const lastSeenAt = parseDatabaseTimestamp(row.lastSeenAt, 'locked last-seen') + const [reconciliation] = await tx + .select({ state: sessionCredentialReconciliation.state }) + .from(sessionCredentialReconciliation) + .where(eq(sessionCredentialReconciliation.singleton, true)) + .limit(1) + .for('key share') + if (!reconciliation) throw new Error('Session credential reconciliation authority is unavailable') + + const locked: AuthorizedSession = { + sessionId: row.sessionId, + userId: row.userId, + lastSeenAt, + expiresAt, + refreshed: authorized.refreshed, + credentialStorageVersion: row.credentialStorageVersion, + writeLegacyCacheAfterCommit: row.credentialStorageVersion === 1 && reconciliation.state === 'expansion', + } + if (locked.writeLegacyCacheAfterCommit) await writeLegacySessionCache(credential, locked) + await cacheAuthorizedSession(digest, locked) + }) +} + export async function getSession( request: Request, ): Promise<{ sessionId: string; userId: string } | null> { @@ -303,12 +505,9 @@ export async function getSession( return null } - // Redis is a repairable cache only. Failure never turns a database-valid - // session into an authorization failure and never extends database expiry. - if (authorized.writeLegacyCacheAfterCommit) { - await writeLegacySessionCache(credential, authorized).catch(() => {}) - } - await cacheAuthorizedSession(digest, authorized).catch(() => {}) + // Redis is a repairable cache only. A failed lock or cache write never turns + // an already-authorized database session into a denial. + await cacheAuthorizedSessionWhileLocked(credential, digest, authorized).catch(() => {}) return { sessionId: authorized.sessionId, userId: authorized.userId } } @@ -377,12 +576,22 @@ export async function createSession( export async function destroySession(sessionCredential: string): Promise { if (!isCanonicalSessionCredential(sessionCredential)) return const digest = computeCredentialDigest(sessionCredential).digest + const generation = crypto.randomUUID() + const claimToken = crypto.randomUUID() // PostgreSQL revocation is authoritative and commits before cache deletion. - await db + const revoked = await db .update(sessions) .set({ revokedAt: sql`pg_catalog.clock_timestamp()`, + cachePurgeAttemptCount: 1, + cachePurgeCredentialDigestV1: digest, + cachePurgeClaimExpiresAt: sql`pg_catalog.clock_timestamp() + ${SESSION_CACHE_PURGE_CLAIM_SECONDS} * interval '1 second'`, + cachePurgeClaimToken: claimToken, + cachePurgeCompletedAt: null, + cachePurgeGeneration: generation, + cachePurgeNextAttemptAt: sql`pg_catalog.clock_timestamp()`, + cachePurgePendingAt: sql`pg_catalog.clock_timestamp()`, legacyRedisPurgePendingAt: sql`CASE WHEN ${sessions.credentialStorageVersion} < 2 THEN pg_catalog.clock_timestamp() @@ -391,10 +600,41 @@ export async function destroySession(sessionCredential: string): Promise { }) .where(or( eq(sessions.credentialDigestV1, digest), - eq(sessions.id, sessionCredential), + and( + eq(sessions.id, sessionCredential), + or( + eq(sessions.credentialStorageVersion, 0), + eq(sessions.credentialStorageVersion, 1), + ), + ), )) + .returning({ + attemptCount: sessions.cachePurgeAttemptCount, + credentialDigest: sessions.cachePurgeCredentialDigestV1, + claimToken: sessions.cachePurgeClaimToken, + generation: sessions.cachePurgeGeneration, + legacyCredential: sql`CASE WHEN ${sessions.credentialStorageVersion} < 2 THEN ${sessions.id}::text ELSE NULL END`, + sessionId: sessions.id, + }) - await redis.del(redisKey(digest), legacyRedisKey(sessionCredential)) + const target = Array.isArray(revoked) ? revoked[0] : undefined + if (!target?.generation || !target.credentialDigest || !target.claimToken) return + const stored = cachePurgeTarget({ + attemptCount: target.attemptCount, + claimToken: target.claimToken, + credentialDigest: target.credentialDigest, + generation: target.generation, + legacyCredential: target.legacyCredential, + sessionId: target.sessionId, + }) + if (!stored) return + try { + await deleteSessionCacheTarget(stored) + await completeSessionCachePurge(stored) + } catch { + // Revocation already committed. Preserve only bounded retry metadata. + await deferSessionCachePurge(stored).catch(() => {}) + } } export function sessionCookieOptions(): CookieOptions { diff --git a/web/scripts/reconcile-session-credentials.ts b/web/scripts/reconcile-session-credentials.ts index d703ac3d..38552a0e 100644 --- a/web/scripts/reconcile-session-credentials.ts +++ b/web/scripts/reconcile-session-credentials.ts @@ -316,6 +316,7 @@ async function main(): Promise { } await tx`alter table sessions validate constraint sessions_credential_digest_v1_length_chk` await tx`alter table sessions validate constraint sessions_credential_storage_version_chk` + await tx`alter table sessions validate constraint sessions_cache_purge_state_chk` await tx`alter table sessions alter column credential_digest_v1 set not null` await tx`alter table sessions alter column expires_at set not null` await tx` diff --git a/web/worker/runtime.ts b/web/worker/runtime.ts index b72a47ee..480391c0 100644 --- a/web/worker/runtime.ts +++ b/web/worker/runtime.ts @@ -8,6 +8,7 @@ const DEFAULT_MAX_ATTEMPTS = 3 const DEFAULT_STUCK_JOB_RECOVERY_SECONDS = 15 * 60 const DEFAULT_PROVIDER_HEALTH_INTERVAL_SECONDS = 5 * 60 const DEFAULT_BLOCKED_HANDOFF_SWEEP_INTERVAL_SECONDS = 5 * 60 +const DEFAULT_SESSION_CACHE_PURGE_INTERVAL_SECONDS = 60 type WorkerSource = 'standalone' | 'embedded' @@ -120,12 +121,18 @@ async function startWorkerOnce( 'FORGE_BLOCKED_HANDOFF_SWEEP_INTERVAL_SECONDS', DEFAULT_BLOCKED_HANDOFF_SWEEP_INTERVAL_SECONDS, ) + const sessionCachePurgeIntervalSeconds = getNonNegativeIntegerEnv( + 'FORGE_SESSION_CACHE_PURGE_INTERVAL_SECONDS', + DEFAULT_SESSION_CACHE_PURGE_INTERVAL_SECONDS, + ) const workerId = `${source}-${process.pid}-${Date.now().toString(36)}` let shuttingDown = false let providerHealthTimer: ReturnType | null = null let providerHealthRunning = false let blockedHandoffSweepTimer: ReturnType | null = null let blockedHandoffSweepRunning = false + let sessionCachePurgeTimer: ReturnType | null = null + let sessionCachePurgeRunning = false const taskExists = async (taskId: string): Promise => { const [row] = await db @@ -225,6 +232,24 @@ async function startWorkerOnce( } } + // This maintenance path is deliberately independent of handoff/S4 recovery: + // a failed handoff sweep must never strand a revoked session cache key. + const sweepSessionCachePurges = async (options: { startup?: boolean } = {}): Promise => { + if ((!options.startup && sessionCachePurgeIntervalSeconds === 0) || sessionCachePurgeRunning) return + sessionCachePurgeRunning = true + try { + const { reconcilePendingSessionCacheInvalidations } = await import('../lib/session') + const reconciled = await reconcilePendingSessionCacheInvalidations(100) + if (reconciled.claimed > 0) { + console.info('[worker] Reconciled revoked session caches', { ...reconciled, workerId }) + } + } catch (err) { + console.warn('[worker] Session cache purge sweep failed', { err: errorMessage(err), workerId }) + } finally { + sessionCachePurgeRunning = false + } + } + const run = async (): Promise => { const executionRequestFlag = defaultOnFeatureFlagState(process.env.FORGE_WORK_PACKAGE_EXECUTION) const executionMode = { @@ -241,6 +266,7 @@ async function startWorkerOnce( hostRepositoryWritesRequested: hostWriteMode.requested, maxAttempts, providerHealthIntervalSeconds, + sessionCachePurgeIntervalSeconds, source, stuckJobRecoveryMs, workPackageExecutionEnabled: executionMode.enabled, @@ -272,6 +298,13 @@ async function startWorkerOnce( blockedHandoffSweepIntervalSeconds * 1000, ) } + void sweepSessionCachePurges({ startup: true }) + if (sessionCachePurgeIntervalSeconds > 0) { + sessionCachePurgeTimer = setInterval( + () => void sweepSessionCachePurges(), + sessionCachePurgeIntervalSeconds * 1000, + ) + } const [recoveredApprovals, recoveredAnswers, recoveredTasks] = await Promise.all([ approvalQueue.recoverStuckJobs(stuckJobRecoveryMs), @@ -569,6 +602,10 @@ async function startWorkerOnce( clearInterval(blockedHandoffSweepTimer) blockedHandoffSweepTimer = null } + if (sessionCachePurgeTimer !== null) { + clearInterval(sessionCachePurgeTimer) + sessionCachePurgeTimer = null + } taskQueue.disconnect() approvalQueue.disconnect() answersQueue.disconnect()