From 46bd7bb1157b59ea8d4663f6d8cfc6d26f4f161a Mon Sep 17 00:00:00 2001 From: Joncallim <64296013+Joncallim@users.noreply.github.com> Date: Mon, 27 Jul 2026 15:05:27 +0800 Subject: [PATCH 1/7] fix: durably retry session cache invalidation --- web/__tests__/auth.test.ts | 20 +- web/__tests__/epic-172-s4-context.test.ts | 3 + web/__tests__/epic-172-s4-postgres.test.ts | 66 ++++++ .../session-cache-invalidation.test.ts | 102 ++++++++++ .../0027_epic_172_s4_packet_context.sql | 37 +++- web/db/schema.ts | 32 +++ web/lib/session-cache-invalidation.ts | 57 ++++++ web/lib/session.ts | 188 +++++++++++++++++- web/scripts/reconcile-session-credentials.ts | 1 + web/worker/runtime.ts | 6 + 10 files changed, 504 insertions(+), 8 deletions(-) create mode 100644 web/__tests__/session-cache-invalidation.test.ts create mode 100644 web/lib/session-cache-invalidation.ts diff --git a/web/__tests__/auth.test.ts b/web/__tests__/auth.test.ts index f4544953..a778f836 100644 --- a/web/__tests__/auth.test.ts +++ b/web/__tests__/auth.test.ts @@ -663,7 +663,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 +684,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() }) }) diff --git a/web/__tests__/epic-172-s4-context.test.ts b/web/__tests__/epic-172-s4-context.test.ts index 8c5c9f86..59b68a63 100644 --- a/web/__tests__/epic-172-s4-context.test.ts +++ b/web/__tests__/epic-172-s4-context.test.ts @@ -424,6 +424,8 @@ 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).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 +445,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..0c19f3c0 100644 --- a/web/__tests__/epic-172-s4-postgres.test.ts +++ b/web/__tests__/epic-172-s4-postgres.test.ts @@ -874,6 +874,72 @@ describe.skipIf(!enabled)('Epic 172 S4 PostgreSQL boundaries', () => { expect(proof).toEqual({ digestRows: 1, rawIdRows: 0, retainedRawIds: 0 }) }) + it('records a revoked session cache purge through pending, claimed, and completed durable states', async () => { + const sessionId = randomUUID() + const generation = randomUUID() + const claimToken = randomUUID() + await admin` + insert into sessions (id, user_id, credential_digest_v1, expires_at, credential_storage_version) + values ( + ${sessionId}::uuid, ${ids.user}::uuid, ${randomBytes(32)}::bytea, + clock_timestamp() + interval '7 days', 2 + ) + ` + await expect(admin` + update sessions + set cache_purge_pending_at = clock_timestamp(), cache_purge_attempt_count = 1 + where id = ${sessionId}::uuid + `).rejects.toMatchObject({ code: '23514' }) + + await admin` + update sessions + set revoked_at = clock_timestamp(), + cache_purge_pending_at = clock_timestamp(), + cache_purge_generation = ${generation}::uuid, + cache_purge_claim_token = ${claimToken}::uuid, + cache_purge_claim_expires_at = clock_timestamp() + interval '30 seconds', + cache_purge_next_attempt_at = clock_timestamp(), + cache_purge_attempt_count = 1 + where id = ${sessionId}::uuid + ` + const [claimed] = await admin<{ + attempts: number + claimToken: string + generation: string + pending: boolean + }[]>` + select cache_purge_attempt_count::integer as attempts, + cache_purge_claim_token::text as "claimToken", + cache_purge_generation::text as generation, + cache_purge_pending_at is not null as pending + from sessions where id = ${sessionId}::uuid + ` + expect(claimed).toEqual({ attempts: 1, claimToken, generation, pending: true }) + + await admin` + update sessions + set cache_purge_pending_at = null, + cache_purge_claim_token = null, + cache_purge_claim_expires_at = null, + cache_purge_next_attempt_at = null, + cache_purge_completed_at = clock_timestamp() + where id = ${sessionId}::uuid + and cache_purge_generation = ${generation}::uuid + and cache_purge_claim_token = ${claimToken}::uuid + ` + const [completed] = await admin<{ + completed: boolean + pending: boolean + tokenCleared: boolean + }[]>` + select cache_purge_completed_at is not null as completed, + cache_purge_pending_at is not null as pending, + cache_purge_claim_token is null as "tokenCleared" + from sessions where id = ${sessionId}::uuid + ` + expect(completed).toEqual({ completed: true, pending: false, tokenCleared: true }) + }) + 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..c914cb38 --- /dev/null +++ b/web/__tests__/session-cache-invalidation.test.ts @@ -0,0 +1,102 @@ +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(undefined) + const defer = vi.fn().mockResolvedValue(undefined) + + await expect(reconcileSessionCacheInvalidations({ + limit: 100, + deleteKeys, + store: { claimDue: vi.fn().mockResolvedValue([claimed]), complete, defer }, + })).resolves.toEqual({ claimed: 1, completed: 1, deferred: 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(undefined) + const defer = vi.fn().mockResolvedValue(undefined) + 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 }) + 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 }) + 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(undefined) + + await expect(reconcileSessionCacheInvalidations({ + limit: 1, + deleteKeys, + store: { claimDue: vi.fn().mockResolvedValue([malformed]), complete, defer }, + })).resolves.toEqual({ claimed: 1, completed: 0, deferred: 1 }) + + expect(deleteKeys).not.toHaveBeenCalled() + expect(complete).not.toHaveBeenCalled() + expect(defer).toHaveBeenCalledWith(malformed) + }) +}) 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..9fc2aa8d 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,14 @@ 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_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 +278,33 @@ 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_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_generation IS NOT NULL + AND cache_purge_completed_at IS NULL) + OR + (cache_purge_pending_at 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..24b046ba 100644 --- a/web/db/schema.ts +++ b/web/db/schema.ts @@ -100,6 +100,13 @@ 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), + 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 +114,31 @@ 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.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.cachePurgeGeneration} is not null + and ${t.cachePurgeCompletedAt} is null) + or + (${t.cachePurgePendingAt} 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..a5aced25 --- /dev/null +++ b/web/lib/session-cache-invalidation.ts @@ -0,0 +1,57 @@ +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 }> { + const targets = await input.store.claimDue(input.limit) + let completed = 0 + let deferred = 0 + for (const target of targets) { + const keys = sessionCachePurgeKeys(target) + if (keys.length === 0) { + await input.store.defer(target) + deferred += 1 + continue + } + try { + await input.deleteKeys(keys) + await input.store.complete(target) + completed += 1 + } catch { + await input.store.defer(target) + deferred += 1 + } + } + return { claimed: targets.length, completed, deferred } +} diff --git a/web/lib/session.ts b/web/lib/session.ts index dcef69ed..de136e57 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,141 @@ 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)) + await db.update(sessions).set({ + cachePurgeClaimExpiresAt: null, + cachePurgeClaimToken: null, + cachePurgeCompletedAt: sql`pg_catalog.clock_timestamp()`, + cachePurgeNextAttemptAt: null, + cachePurgePendingAt: null, + }).where(and(...conditions)) +} + +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)) + await db.update(sessions).set({ + cachePurgeClaimExpiresAt: null, + cachePurgeClaimToken: null, + cachePurgeNextAttemptAt: sql`pg_catalog.clock_timestamp() + ${sessionCachePurgeBackoffSeconds(target.attemptCount)} * interval '1 second'`, + }).where(and(...conditions)) +} + +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 }> { + 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.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.credential_digest_v1, 'hex') AS "credentialDigestHex", + CASE WHEN session.credential_storage_version = 1 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') { @@ -377,12 +520,21 @@ 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, + 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 +543,38 @@ export async function destroySession(sessionCredential: string): Promise { }) .where(or( eq(sessions.credentialDigestV1, digest), - eq(sessions.id, sessionCredential), + and( + eq(sessions.id, sessionCredential), + eq(sessions.credentialStorageVersion, 1), + ), )) + .returning({ + attemptCount: sessions.cachePurgeAttemptCount, + credentialDigest: sessions.credentialDigestV1, + claimToken: sessions.cachePurgeClaimToken, + generation: sessions.cachePurgeGeneration, + legacyCredential: sql`CASE WHEN ${sessions.credentialStorageVersion} = 1 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..20eb1f9d 100644 --- a/web/worker/runtime.ts +++ b/web/worker/runtime.ts @@ -184,6 +184,7 @@ async function startWorkerOnce( { tasks, workPackages }, { enqueueDueBlockedHandoffRetries }, { convergeRecognizedOperatorHolds }, + { reconcilePendingSessionCacheInvalidations }, { reconcilePendingS4CompletionHandoffs }, { and, eq }, ] = await Promise.all([ @@ -191,6 +192,7 @@ async function startWorkerOnce( import('../db/schema'), import('./blocked-handoff-retry'), import('../lib/mcps/filesystem-grant-reconciliation'), + import('../lib/session'), import('./work-package-handoff'), import('drizzle-orm'), ]) @@ -207,6 +209,7 @@ async function startWorkerOnce( drain: options.startup === true, workerId, }) + const reconciledSessionCaches = await reconcilePendingSessionCacheInvalidations(100) const enqueued = await enqueueDueBlockedHandoffRetries(stuck) const converged = await convergeRecognizedOperatorHolds() if (enqueued > 0) { @@ -218,6 +221,9 @@ async function startWorkerOnce( if (recoveredS4Handoffs > 0) { console.info('[worker] Recovered protected completion handoffs', { count: recoveredS4Handoffs, workerId }) } + if (reconciledSessionCaches.claimed > 0) { + console.info('[worker] Reconciled revoked session caches', { ...reconciledSessionCaches, workerId }) + } } catch (err) { console.warn('[worker] Blocked-handoff sweep failed', { err: errorMessage(err), workerId }) } finally { From a6d148f492182e2b6ecf19b3462cd9fab6cc23f4 Mon Sep 17 00:00:00 2001 From: Joncallim <64296013+Joncallim@users.noreply.github.com> Date: Mon, 27 Jul 2026 15:14:58 +0800 Subject: [PATCH 2/7] test: prove session-authoritative history route --- web/__tests__/epic-172-s4-postgres.test.ts | 183 ++++++++++++++++++++- 1 file changed, 182 insertions(+), 1 deletion(-) diff --git a/web/__tests__/epic-172-s4-postgres.test.ts b/web/__tests__/epic-172-s4-postgres.test.ts index 0c19f3c0..26baa7ba 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,156 @@ 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 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', 'not accessible', 'running')` + }) + 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: 'route-policy', schemaVersion: 1 }), + entryId: 'requirement:route-policy', entryKind: 'requirement', projectionEligible: false, requirementKey: 'route-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 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 () => { + const [row] = await admin<{ count: number }[]>` + select count(*)::integer as count from architect_plan_history_reads where task_id = ${routeTask}::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() + const response = await routeRead(credential, routeTask, '1') + expect(response.status).toBe(401) + await expect(response.json()).resolves.toEqual({ error: 'Unauthorized' }) + expect(await auditCount()).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()) + + const crossTaskCredential = await passwordSession() + const beforeCrossTask = await auditCount() + 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()).toBe(beforeCrossTask) + + const beforeWrongVersion = await auditCount() + 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()).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() From 7def101b0217abf935dfb9fa4ffb3ecce50b5ac9 Mon Sep 17 00:00:00 2001 From: Joncallim <64296013+Joncallim@users.noreply.github.com> Date: Mon, 27 Jul 2026 15:20:47 +0800 Subject: [PATCH 3/7] test: use mandatory history policy fixture --- web/__tests__/epic-172-s4-postgres.test.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/web/__tests__/epic-172-s4-postgres.test.ts b/web/__tests__/epic-172-s4-postgres.test.ts index 26baa7ba..ca7a3c09 100644 --- a/web/__tests__/epic-172-s4-postgres.test.ts +++ b/web/__tests__/epic-172-s4-postgres.test.ts @@ -905,8 +905,8 @@ describe.skipIf(!enabled)('Epic 172 S4 PostgreSQL boundaries', () => { entryId: 'plan_body:000000', entryKind: 'plan_body', projectionEligible: false, requirementKey: null, }, { agent: null, bindingFingerprint: null, - content: JSON.stringify({ requirementKey: 'route-policy', schemaVersion: 1 }), - entryId: 'requirement:route-policy', entryKind: 'requirement', projectionEligible: false, requirementKey: 'route-policy', + content: JSON.stringify({ requirementKey: 'plan-policy', schemaVersion: 1 }), + entryId: 'requirement:plan-policy', entryKind: 'requirement', projectionEligible: false, requirementKey: 'plan-policy', }], }) From b9b20b34fe5573f022bb1ffa93acca0c4d4cb135 Mon Sep 17 00:00:00 2001 From: Joncallim <64296013+Joncallim@users.noreply.github.com> Date: Mon, 27 Jul 2026 15:43:45 +0800 Subject: [PATCH 4/7] fix: fence durable session cache purges --- web/__tests__/auth.test.ts | 65 +++++++- web/__tests__/epic-172-s4-context.test.ts | 1 + web/__tests__/epic-172-s4-postgres.test.ts | 141 ++++++++++-------- .../session-cache-invalidation.test.ts | 47 ++++-- web/__tests__/worker-retry-contract.test.ts | 11 ++ .../0027_epic_172_s4_packet_context.sql | 4 + web/db/schema.ts | 4 + web/lib/session-cache-invalidation.ts | 21 +-- web/lib/session.ts | 86 +++++++++-- web/worker/runtime.ts | 43 +++++- 10 files changed, 321 insertions(+), 102 deletions(-) diff --git a/web/__tests__/auth.test.ts b/web/__tests__/auth.test.ts index a778f836..97d6ef53 100644 --- a/web/__tests__/auth.test.ts +++ b/web/__tests__/auth.test.ts @@ -559,7 +559,7 @@ describe('getSession — write-behind logic', () => { vi.useRealTimers() }) - it('does NOT trigger a DB write when lastSeenAt is less than 60 seconds old', async () => { + it('writes no sliding refresh when lastSeenAt is recent, but still fences the cache write', async () => { const now = Date.now() vi.setSystemTime(now) @@ -572,7 +572,7 @@ describe('getSession — write-behind logic', () => { const req = fakeRequest('00000000-0000-4000-8000-000000000000') await getSession(req) - expect(mockDbUpdate).not.toHaveBeenCalled() + expect(mockDbUpdate).toHaveBeenCalledOnce() }) it('triggers a fire-and-forget DB write when lastSeenAt is older than 60 seconds', async () => { @@ -588,12 +588,69 @@ 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() + // One authoritative refresh plus one post-cache-write authority fence. + expect(mockDbUpdate).toHaveBeenCalledTimes(2) // Redis was also refreshed expect(mockRedisSet).toHaveBeenCalledOnce() }) + it('requeues and deletes a v2 cache key when logout wins after the cache write', async () => { + const now = Date.now() + vi.setSystemTime(now) + const credential = '00000000-0000-4000-8000-000000000000' + 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), + }])) + mockDbUpdate + .mockReturnValueOnce(chain([{ + attemptCount: 1, + claimToken: '00000000-0000-4000-8000-000000000011', + credentialDigest: Buffer.alloc(32, 1), + generation: '00000000-0000-4000-8000-000000000012', + legacyCredential: null, + sessionId: '00000000-0000-4000-8000-000000000010', + }])) + .mockReturnValueOnce(chain([{ id: '00000000-0000-4000-8000-000000000010' }])) + + await expect(getSession(fakeRequest(credential))).resolves.toEqual({ + sessionId: '00000000-0000-4000-8000-000000000010', userId: 'user-1', + }) + + expect(mockRedisSet.mock.invocationCallOrder[0]).toBeLessThan(mockRedisDel.mock.invocationCallOrder[0]) + expect(mockRedisDel).toHaveBeenCalledWith(expect.stringMatching(/^session:v2:/)) + }) + + it('fences both dual-write keys when revocation races the post-commit cache write', async () => { + const now = Date.now() + vi.setSystemTime(now) + const credential = '00000000-0000-4000-8000-000000000000' + mockDbSelect + .mockReturnValueOnce(chain([{ state: 'expansion' }])) + .mockReturnValueOnce(chain([{ + sessionId: credential, userId: 'user-1', + lastSeenAt: new Date(now - 30_000), expiresAt: new Date(now + 60_000), + revokedAt: null, credentialStorageVersion: 1, databaseNow: new Date(now), + }])) + mockDbUpdate + .mockReturnValueOnce(chain([{ + attemptCount: 1, + claimToken: '00000000-0000-4000-8000-000000000011', + credentialDigest: Buffer.alloc(32, 1), + generation: '00000000-0000-4000-8000-000000000012', + legacyCredential: credential, + sessionId: credential, + }])) + .mockReturnValueOnce(chain([{ id: credential }])) + + await getSession(fakeRequest(credential)) + + expect(mockRedisSet).toHaveBeenCalledTimes(2) + expect(mockRedisSet.mock.invocationCallOrder[1]).toBeLessThan(mockRedisDel.mock.invocationCallOrder[0]) + expect(mockRedisDel).toHaveBeenCalledWith(expect.stringMatching(/^session:v2:/), `session:${credential}`) + }) + it('does not write a legacy cache when the sliding-refresh transaction fails to commit', async () => { const now = Date.now() vi.setSystemTime(now) diff --git a/web/__tests__/epic-172-s4-context.test.ts b/web/__tests__/epic-172-s4-context.test.ts index 59b68a63..492e39c0 100644 --- a/web/__tests__/epic-172-s4-context.test.ts +++ b/web/__tests__/epic-172-s4-context.test.ts @@ -426,6 +426,7 @@ describe('Epic 172 S4 PostgreSQL CI contract', () => { 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).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') diff --git a/web/__tests__/epic-172-s4-postgres.test.ts b/web/__tests__/epic-172-s4-postgres.test.ts index 0c19f3c0..1f02618f 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,11 @@ 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 { closeDb } from '@/db' + +const { mockRedisDel } = vi.hoisted(() => ({ mockRedisDel: vi.fn() })) + +vi.mock('@/lib/redis', () => ({ redis: { del: mockRedisDel } })) const adminUrl = process.env.FORGE_S4_POSTGRES_TEST_DATABASE_URL?.trim() const issuerUrl = process.env.FORGE_PACKET_ISSUER_DATABASE_URL?.trim() @@ -62,11 +67,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 +216,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 })]) }) @@ -874,70 +884,81 @@ describe.skipIf(!enabled)('Epic 172 S4 PostgreSQL boundaries', () => { expect(proof).toEqual({ digestRows: 1, rawIdRows: 0, retainedRawIds: 0 }) }) - it('records a revoked session cache purge through pending, claimed, and completed durable states', async () => { - const sessionId = randomUUID() - const generation = randomUUID() - const claimToken = randomUUID() - await admin` - insert into sessions (id, user_id, credential_digest_v1, expires_at, credential_storage_version) - values ( - ${sessionId}::uuid, ${ids.user}::uuid, ${randomBytes(32)}::bytea, - clock_timestamp() + interval '7 days', 2 - ) - ` - await expect(admin` - update sessions - set cache_purge_pending_at = clock_timestamp(), cache_purge_attempt_count = 1 - where id = ${sessionId}::uuid - `).rejects.toMatchObject({ code: '23514' }) + 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 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}) + ` + } - await admin` - update sessions - set revoked_at = clock_timestamp(), - cache_purge_pending_at = clock_timestamp(), - cache_purge_generation = ${generation}::uuid, - cache_purge_claim_token = ${claimToken}::uuid, - cache_purge_claim_expires_at = clock_timestamp() + interval '30 seconds', - cache_purge_next_attempt_at = clock_timestamp(), - cache_purge_attempt_count = 1 - where id = ${sessionId}::uuid - ` - const [claimed] = await admin<{ - attempts: number - claimToken: string - generation: string - pending: boolean - }[]>` - select cache_purge_attempt_count::integer as attempts, - cache_purge_claim_token::text as "claimToken", - cache_purge_generation::text as generation, - cache_purge_pending_at is not null as pending - from sessions where id = ${sessionId}::uuid - ` - expect(claimed).toEqual({ attempts: 1, claimToken, generation, pending: true }) + 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` - update sessions - set cache_purge_pending_at = null, - cache_purge_claim_token = null, - cache_purge_claim_expires_at = null, - cache_purge_next_attempt_at = null, - cache_purge_completed_at = clock_timestamp() - where id = ${sessionId}::uuid - and cache_purge_generation = ${generation}::uuid - and cache_purge_claim_token = ${claimToken}::uuid + 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) ` - const [completed] = await admin<{ - completed: boolean - pending: boolean - tokenCleared: boolean - }[]>` - select cache_purge_completed_at is not null as completed, - cache_purge_pending_at is not null as pending, - cache_purge_claim_token is null as "tokenCleared" - from sessions where id = ${sessionId}::uuid + 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 ` - expect(completed).toEqual({ completed: true, pending: false, tokenCleared: true }) + 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('allow-once-single-winner: atomically keeps one audit and one nonce claim', async () => { diff --git a/web/__tests__/session-cache-invalidation.test.ts b/web/__tests__/session-cache-invalidation.test.ts index c914cb38..c8a280b8 100644 --- a/web/__tests__/session-cache-invalidation.test.ts +++ b/web/__tests__/session-cache-invalidation.test.ts @@ -33,14 +33,14 @@ describe('session cache invalidation', () => { 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(undefined) - const defer = vi.fn().mockResolvedValue(undefined) + 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 }) + })).resolves.toEqual({ claimed: 1, completed: 1, deferred: 0, stale: 0 }) expect(deleteKeys).toHaveBeenCalledWith([`session:v2:${digest}`, 'session:legacy-id']) expect(complete).toHaveBeenCalledWith(claimed) @@ -53,8 +53,8 @@ describe('session cache invalidation', () => { const deleteKeys = vi.fn() .mockRejectedValueOnce(new Error('redis unavailable')) .mockResolvedValueOnce(0) - const complete = vi.fn().mockResolvedValue(undefined) - const defer = vi.fn().mockResolvedValue(undefined) + const complete = vi.fn().mockResolvedValue(true) + const defer = vi.fn().mockResolvedValue(true) const claimDue = vi.fn() .mockResolvedValueOnce([first]) .mockResolvedValueOnce([recovered]) @@ -63,7 +63,7 @@ describe('session cache invalidation', () => { limit: 1, deleteKeys, store: { claimDue, complete, defer }, - })).resolves.toEqual({ claimed: 1, completed: 0, deferred: 1 }) + })).resolves.toEqual({ claimed: 1, completed: 0, deferred: 1, stale: 0 }) expect(defer).toHaveBeenCalledWith(first) expect(complete).not.toHaveBeenCalled() @@ -71,7 +71,7 @@ describe('session cache invalidation', () => { limit: 1, deleteKeys, store: { claimDue, complete, defer }, - })).resolves.toEqual({ claimed: 1, completed: 1, deferred: 0 }) + })).resolves.toEqual({ claimed: 1, completed: 1, deferred: 0, stale: 0 }) expect(complete).toHaveBeenCalledWith(recovered) expect(deleteKeys).toHaveBeenCalledTimes(2) }) @@ -87,16 +87,45 @@ describe('session cache invalidation', () => { const malformed = target({ credentialDigestHex: 'not-a-digest' }) const deleteKeys = vi.fn() const complete = vi.fn() - const defer = vi.fn().mockResolvedValue(undefined) + 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 }) + })).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 9fc2aa8d..70095566 100644 --- a/web/db/migrations/0027_epic_172_s4_packet_context.sql +++ b/web/db/migrations/0027_epic_172_s4_packet_context.sql @@ -58,6 +58,7 @@ ALTER TABLE public.sessions 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 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, @@ -283,6 +284,7 @@ ALTER TABLE public.sessions 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 @@ -290,10 +292,12 @@ ALTER TABLE public.sessions AND cache_purge_completed_at IS NULL) OR (cache_purge_pending_at 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 diff --git a/web/db/schema.ts b/web/db/schema.ts index 24b046ba..5b6b8260 100644 --- a/web/db/schema.ts +++ b/web/db/schema.ts @@ -101,6 +101,7 @@ export const sessions = pgTable( 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), @@ -121,6 +122,7 @@ export const sessions = pgTable( ${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 @@ -128,10 +130,12 @@ export const sessions = pgTable( and ${t.cachePurgeCompletedAt} is null) or (${t.cachePurgePendingAt} 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 diff --git a/web/lib/session-cache-invalidation.ts b/web/lib/session-cache-invalidation.ts index a5aced25..3f954fc9 100644 --- a/web/lib/session-cache-invalidation.ts +++ b/web/lib/session-cache-invalidation.ts @@ -9,8 +9,8 @@ export type SessionCachePurgeTarget = { export type SessionCachePurgeStore = { claimDue: (limit: number) => Promise - complete: (target: SessionCachePurgeTarget) => Promise - defer: (target: SessionCachePurgeTarget) => Promise + complete: (target: SessionCachePurgeTarget) => Promise + defer: (target: SessionCachePurgeTarget) => Promise } export function sessionCachePurgeKeys(target: Pick): string[] { @@ -33,25 +33,26 @@ export async function reconcileSessionCacheInvalidations(input: { deleteKeys: (keys: readonly string[]) => Promise limit: number store: SessionCachePurgeStore -}): Promise<{ completed: number; deferred: number; claimed: number }> { +}): 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) { - await input.store.defer(target) - deferred += 1 + if (await input.store.defer(target)) deferred += 1 + else stale += 1 continue } try { await input.deleteKeys(keys) - await input.store.complete(target) - completed += 1 + if (await input.store.complete(target)) completed += 1 + else stale += 1 } catch { - await input.store.defer(target) - deferred += 1 + if (await input.store.defer(target)) deferred += 1 + else stale += 1 } } - return { claimed: targets.length, completed, deferred } + return { claimed: targets.length, completed, deferred, stale } } diff --git a/web/lib/session.ts b/web/lib/session.ts index de136e57..e0d30065 100644 --- a/web/lib/session.ts +++ b/web/lib/session.ts @@ -72,40 +72,95 @@ function cachePurgeTarget(row: { } } -async function completeSessionCachePurge(target: SessionCachePurgeTarget): Promise { +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)) - await db.update(sessions).set({ + 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)) + }).where(and(...conditions)).returning({ id: sessions.id }) + return Array.isArray(completed) && completed.length === 1 } -async function deferSessionCachePurge(target: SessionCachePurgeTarget): Promise { +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)) - await db.update(sessions).set({ + 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)) + }).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)) } +/** + * Runs after a cache write. If logout/expiry won the race after the original + * database authorization, this locked compare-and-set creates a fresh purge + * claim before deleting the just-written keys. + */ +async function fenceSessionCacheWrite(digest: Buffer, sessionId: string): Promise { + const generation = crypto.randomUUID() + const claimToken = crypto.randomUUID() + const fenced = await db.update(sessions).set({ + cachePurgeAttemptCount: 1, + cachePurgeClaimExpiresAt: sql`pg_catalog.clock_timestamp() + ${SESSION_CACHE_PURGE_CLAIM_SECONDS} * interval '1 second'`, + cachePurgeClaimToken: claimToken, + cachePurgeCompletedAt: null, + cachePurgeCredentialDigestV1: digest, + cachePurgeGeneration: generation, + cachePurgeNextAttemptAt: sql`pg_catalog.clock_timestamp()`, + cachePurgePendingAt: sql`pg_catalog.clock_timestamp()`, + }).where(and( + eq(sessions.id, sessionId), + eq(sessions.credentialDigestV1, digest), + or( + isNotNull(sessions.revokedAt), + isNull(sessions.expiresAt), + sql`${sessions.expiresAt} <= pg_catalog.clock_timestamp()`, + ), + )).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, + }) + const target = Array.isArray(fenced) ? fenced[0] : undefined + if (!target?.credentialDigest || !target.claimToken || !target.generation) return + const purge = cachePurgeTarget({ + attemptCount: target.attemptCount, + claimToken: target.claimToken, + credentialDigest: target.credentialDigest, + generation: target.generation, + legacyCredential: target.legacyCredential, + sessionId: target.sessionId, + }) + if (!purge) return + try { + await deleteSessionCacheTarget(purge) + await completeSessionCachePurge(purge) + } catch { + await deferSessionCachePurge(purge).catch(() => {}) + } +} + /** * Bounded worker maintenance for already-revoked sessions. It never reads Redis * to authorize a request and only derives cache keys from persisted digest/legacy @@ -113,7 +168,7 @@ async function deleteSessionCacheTarget(target: SessionCachePurgeTarget): Promis */ export async function reconcilePendingSessionCacheInvalidations( limit = SESSION_CACHE_PURGE_BATCH_LIMIT, -): Promise<{ claimed: number; completed: number; deferred: number }> { +): 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}.`) } @@ -134,7 +189,7 @@ export async function reconcilePendingSessionCacheInvalidations( SELECT session.id FROM public.sessions session WHERE session.cache_purge_pending_at IS NOT NULL - AND pg_catalog.octet_length(session.credential_digest_v1) = 32 + 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 @@ -154,8 +209,8 @@ export async function reconcilePendingSessionCacheInvalidations( 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.credential_digest_v1, 'hex') AS "credentialDigestHex", - CASE WHEN session.credential_storage_version = 1 THEN session.id::text ELSE NULL END AS "legacyCredential" + 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 ( @@ -452,6 +507,7 @@ export async function getSession( await writeLegacySessionCache(credential, authorized).catch(() => {}) } await cacheAuthorizedSession(digest, authorized).catch(() => {}) + await fenceSessionCacheWrite(digest, authorized.sessionId).catch(() => {}) return { sessionId: authorized.sessionId, userId: authorized.userId } } @@ -529,6 +585,7 @@ export async function destroySession(sessionCredential: string): Promise { .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, @@ -545,15 +602,18 @@ export async function destroySession(sessionCredential: string): Promise { eq(sessions.credentialDigestV1, digest), and( eq(sessions.id, sessionCredential), - eq(sessions.credentialStorageVersion, 1), + or( + eq(sessions.credentialStorageVersion, 0), + eq(sessions.credentialStorageVersion, 1), + ), ), )) .returning({ attemptCount: sessions.cachePurgeAttemptCount, - credentialDigest: sessions.credentialDigestV1, + credentialDigest: sessions.cachePurgeCredentialDigestV1, claimToken: sessions.cachePurgeClaimToken, generation: sessions.cachePurgeGeneration, - legacyCredential: sql`CASE WHEN ${sessions.credentialStorageVersion} = 1 THEN ${sessions.id}::text ELSE NULL END`, + legacyCredential: sql`CASE WHEN ${sessions.credentialStorageVersion} < 2 THEN ${sessions.id}::text ELSE NULL END`, sessionId: sessions.id, }) diff --git a/web/worker/runtime.ts b/web/worker/runtime.ts index 20eb1f9d..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 @@ -184,7 +191,6 @@ async function startWorkerOnce( { tasks, workPackages }, { enqueueDueBlockedHandoffRetries }, { convergeRecognizedOperatorHolds }, - { reconcilePendingSessionCacheInvalidations }, { reconcilePendingS4CompletionHandoffs }, { and, eq }, ] = await Promise.all([ @@ -192,7 +198,6 @@ async function startWorkerOnce( import('../db/schema'), import('./blocked-handoff-retry'), import('../lib/mcps/filesystem-grant-reconciliation'), - import('../lib/session'), import('./work-package-handoff'), import('drizzle-orm'), ]) @@ -209,7 +214,6 @@ async function startWorkerOnce( drain: options.startup === true, workerId, }) - const reconciledSessionCaches = await reconcilePendingSessionCacheInvalidations(100) const enqueued = await enqueueDueBlockedHandoffRetries(stuck) const converged = await convergeRecognizedOperatorHolds() if (enqueued > 0) { @@ -221,9 +225,6 @@ async function startWorkerOnce( if (recoveredS4Handoffs > 0) { console.info('[worker] Recovered protected completion handoffs', { count: recoveredS4Handoffs, workerId }) } - if (reconciledSessionCaches.claimed > 0) { - console.info('[worker] Reconciled revoked session caches', { ...reconciledSessionCaches, workerId }) - } } catch (err) { console.warn('[worker] Blocked-handoff sweep failed', { err: errorMessage(err), workerId }) } finally { @@ -231,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 = { @@ -247,6 +266,7 @@ async function startWorkerOnce( hostRepositoryWritesRequested: hostWriteMode.requested, maxAttempts, providerHealthIntervalSeconds, + sessionCachePurgeIntervalSeconds, source, stuckJobRecoveryMs, workPackageExecutionEnabled: executionMode.enabled, @@ -278,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), @@ -575,6 +602,10 @@ async function startWorkerOnce( clearInterval(blockedHandoffSweepTimer) blockedHandoffSweepTimer = null } + if (sessionCachePurgeTimer !== null) { + clearInterval(sessionCachePurgeTimer) + sessionCachePurgeTimer = null + } taskQueue.disconnect() approvalQueue.disconnect() answersQueue.disconnect() From 33372fc1fc728c3fe2de364bf20cba27666e305a Mon Sep 17 00:00:00 2001 From: Joncallim <64296013+Joncallim@users.noreply.github.com> Date: Mon, 27 Jul 2026 15:48:07 +0800 Subject: [PATCH 5/7] test: prove cross-user history audit isolation --- web/__tests__/epic-172-s4-postgres.test.ts | 61 ++++++++++++++++++---- 1 file changed, 52 insertions(+), 9 deletions(-) diff --git a/web/__tests__/epic-172-s4-postgres.test.ts b/web/__tests__/epic-172-s4-postgres.test.ts index f9b1d127..edf388f0 100644 --- a/web/__tests__/epic-172-s4-postgres.test.ts +++ b/web/__tests__/epic-172-s4-postgres.test.ts @@ -866,6 +866,7 @@ describe.skipIf(!enabled)('Epic 172 S4 PostgreSQL boundaries', () => { 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 @@ -892,7 +893,9 @@ describe.skipIf(!enabled)('Epic 172 S4 PostgreSQL boundaries', () => { 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', 'not accessible', 'running')` + '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, @@ -909,9 +912,25 @@ describe.skipIf(!enabled)('Epic 172 S4 PostgreSQL boundaries', () => { 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 }), @@ -927,9 +946,9 @@ describe.skipIf(!enabled)('Epic 172 S4 PostgreSQL boundaries', () => { }) as never, { params: Promise.resolve({ id: taskId, planVersion }) }, ) - const auditCount = async () => { + const auditCount = async (taskId: string) => { const [row] = await admin<{ count: number }[]>` - select count(*)::integer as count from architect_plan_history_reads where task_id = ${routeTask}::uuid + select count(*)::integer as count from architect_plan_history_reads where task_id = ${taskId}::uuid ` return row.count } @@ -964,11 +983,11 @@ describe.skipIf(!enabled)('Epic 172 S4 PostgreSQL boundaries', () => { expect(auditTextColumns).toEqual({ count: 0 }) const assertSessionDenied = async (credential: string) => { - const before = await auditCount() + 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()).toBe(before) + expect(await auditCount(routeTask)).toBe(before) } const revokedCredential = await passwordSession() await admin`update sessions set revoked_at = clock_timestamp() @@ -993,18 +1012,42 @@ describe.skipIf(!enabled)('Epic 172 S4 PostgreSQL boundaries', () => { 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 beforeCrossTask = await auditCount() + 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()).toBe(beforeCrossTask) + expect(await auditCount(routeTask)).toBe(beforeCrossTaskOwner) + expect(await auditCount(otherTask)).toBe(beforeCrossTaskOther) - const beforeWrongVersion = await auditCount() + 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()).toBe(beforeWrongVersion) + expect(await auditCount(routeTask)).toBe(beforeWrongVersion) expect(mockRedisSet).toHaveBeenCalled() }) From 30f7b067d7d389067a9507cda9b751d75648a697 Mon Sep 17 00:00:00 2001 From: Joncallim <64296013+Joncallim@users.noreply.github.com> Date: Mon, 27 Jul 2026 16:03:17 +0800 Subject: [PATCH 6/7] fix: serialize session cache writes with revocation --- web/__tests__/auth.test.ts | 149 ++++++++++++------ web/__tests__/epic-172-s4-context.test.ts | 2 + web/__tests__/epic-172-s4-postgres.test.ts | 83 +++++++++- .../0027_epic_172_s4_packet_context.sql | 1 + web/db/schema.ts | 1 + web/lib/session.ts | 118 +++++++------- 6 files changed, 242 insertions(+), 112 deletions(-) diff --git a/web/__tests__/auth.test.ts b/web/__tests__/auth.test.ts index 97d6ef53..1e9dc135 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([{ @@ -559,7 +573,7 @@ describe('getSession — write-behind logic', () => { vi.useRealTimers() }) - it('writes no sliding refresh when lastSeenAt is recent, but still fences the cache write', async () => { + it('writes cache entries inside a second locked transaction when lastSeenAt is recent', async () => { const now = Date.now() vi.setSystemTime(now) @@ -572,10 +586,12 @@ describe('getSession — write-behind logic', () => { const req = fakeRequest('00000000-0000-4000-8000-000000000000') await getSession(req) - expect(mockDbUpdate).toHaveBeenCalledOnce() + 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,67 +604,78 @@ describe('getSession — write-behind logic', () => { const req = fakeRequest('00000000-0000-4000-8000-000000000000') await getSession(req) - // One authoritative refresh plus one post-cache-write authority fence. - expect(mockDbUpdate).toHaveBeenCalledTimes(2) - // Redis was also refreshed + expect(mockDbUpdate).toHaveBeenCalledOnce() expect(mockRedisSet).toHaveBeenCalledOnce() }) - it('requeues and deletes a v2 cache key when logout wins after the cache write', async () => { + 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' - mockDbSelect.mockReturnValue(chain([{ - sessionId: '00000000-0000-4000-8000-000000000010', userId: 'user-1', + 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: 2, databaseNow: new Date(now), + 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, }])) - mockDbUpdate - .mockReturnValueOnce(chain([{ - attemptCount: 1, - claimToken: '00000000-0000-4000-8000-000000000011', - credentialDigest: Buffer.alloc(32, 1), - generation: '00000000-0000-4000-8000-000000000012', - legacyCredential: null, - sessionId: '00000000-0000-4000-8000-000000000010', - }])) - .mockReturnValueOnce(chain([{ id: '00000000-0000-4000-8000-000000000010' }])) - - await expect(getSession(fakeRequest(credential))).resolves.toEqual({ - sessionId: '00000000-0000-4000-8000-000000000010', userId: 'user-1', + 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 }) - expect(mockRedisSet.mock.invocationCallOrder[0]).toBeLessThan(mockRedisDel.mock.invocationCallOrder[0]) - expect(mockRedisDel).toHaveBeenCalledWith(expect.stringMatching(/^session:v2:/)) + 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('fences both dual-write keys when revocation races the post-commit cache write', async () => { + it('skips cache population when the row-lock transaction fails before a write', async () => { const now = Date.now() vi.setSystemTime(now) - const credential = '00000000-0000-4000-8000-000000000000' - mockDbSelect - .mockReturnValueOnce(chain([{ state: 'expansion' }])) - .mockReturnValueOnce(chain([{ - sessionId: credential, userId: 'user-1', - lastSeenAt: new Date(now - 30_000), expiresAt: new Date(now + 60_000), - revokedAt: null, credentialStorageVersion: 1, databaseNow: new Date(now), - }])) - mockDbUpdate - .mockReturnValueOnce(chain([{ - attemptCount: 1, - claimToken: '00000000-0000-4000-8000-000000000011', - credentialDigest: Buffer.alloc(32, 1), - generation: '00000000-0000-4000-8000-000000000012', - legacyCredential: credential, - sessionId: credential, - }])) - .mockReturnValueOnce(chain([{ id: credential }])) - - await getSession(fakeRequest(credential)) + 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()) + }) - expect(mockRedisSet).toHaveBeenCalledTimes(2) - expect(mockRedisSet.mock.invocationCallOrder[1]).toBeLessThan(mockRedisDel.mock.invocationCallOrder[0]) - expect(mockRedisDel).toHaveBeenCalledWith(expect.stringMatching(/^session:v2:/), `session:${credential}`) + 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 () => { @@ -666,6 +693,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') @@ -693,6 +730,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()) diff --git a/web/__tests__/epic-172-s4-context.test.ts b/web/__tests__/epic-172-s4-context.test.ts index 492e39c0..690da623 100644 --- a/web/__tests__/epic-172-s4-context.test.ts +++ b/web/__tests__/epic-172-s4-context.test.ts @@ -427,6 +427,8 @@ describe('Epic 172 S4 PostgreSQL CI contract', () => { 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') diff --git a/web/__tests__/epic-172-s4-postgres.test.ts b/web/__tests__/epic-172-s4-postgres.test.ts index 1f02618f..31491c9d 100644 --- a/web/__tests__/epic-172-s4-postgres.test.ts +++ b/web/__tests__/epic-172-s4-postgres.test.ts @@ -13,9 +13,9 @@ import { computeCredentialDigest } from '@/lib/session-credential-digest' import { appendArchitectClarificationAnswer, readArchitectPlanHistory } from '@/lib/mcps/history-reader' import { closeDb } from '@/db' -const { mockRedisDel } = vi.hoisted(() => ({ mockRedisDel: vi.fn() })) +const { mockRedisDel, mockRedisSet } = vi.hoisted(() => ({ mockRedisDel: vi.fn(), mockRedisSet: vi.fn() })) -vi.mock('@/lib/redis', () => ({ redis: { del: mockRedisDel } })) +vi.mock('@/lib/redis', () => ({ redis: { del: mockRedisDel, set: mockRedisSet } })) const adminUrl = process.env.FORGE_S4_POSTGRES_TEST_DATABASE_URL?.trim() const issuerUrl = process.env.FORGE_PACKET_ISSUER_DATABASE_URL?.trim() @@ -886,6 +886,22 @@ describe.skipIf(!enabled)('Epic 172 S4 PostgreSQL boundaries', () => { 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 @@ -961,6 +977,69 @@ describe.skipIf(!enabled)('Epic 172 S4 PostgreSQL boundaries', () => { 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/db/migrations/0027_epic_172_s4_packet_context.sql b/web/db/migrations/0027_epic_172_s4_packet_context.sql index 70095566..b6ce989c 100644 --- a/web/db/migrations/0027_epic_172_s4_packet_context.sql +++ b/web/db/migrations/0027_epic_172_s4_packet_context.sql @@ -292,6 +292,7 @@ ALTER TABLE public.sessions 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) diff --git a/web/db/schema.ts b/web/db/schema.ts index 5b6b8260..d5099356 100644 --- a/web/db/schema.ts +++ b/web/db/schema.ts @@ -130,6 +130,7 @@ export const sessions = pgTable( 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) diff --git a/web/lib/session.ts b/web/lib/session.ts index e0d30065..df0dcfc7 100644 --- a/web/lib/session.ts +++ b/web/lib/session.ts @@ -109,58 +109,6 @@ async function deleteSessionCacheTarget(target: SessionCachePurgeTarget): Promis await redis.del(...sessionCachePurgeKeys(target)) } -/** - * Runs after a cache write. If logout/expiry won the race after the original - * database authorization, this locked compare-and-set creates a fresh purge - * claim before deleting the just-written keys. - */ -async function fenceSessionCacheWrite(digest: Buffer, sessionId: string): Promise { - const generation = crypto.randomUUID() - const claimToken = crypto.randomUUID() - const fenced = await db.update(sessions).set({ - cachePurgeAttemptCount: 1, - cachePurgeClaimExpiresAt: sql`pg_catalog.clock_timestamp() + ${SESSION_CACHE_PURGE_CLAIM_SECONDS} * interval '1 second'`, - cachePurgeClaimToken: claimToken, - cachePurgeCompletedAt: null, - cachePurgeCredentialDigestV1: digest, - cachePurgeGeneration: generation, - cachePurgeNextAttemptAt: sql`pg_catalog.clock_timestamp()`, - cachePurgePendingAt: sql`pg_catalog.clock_timestamp()`, - }).where(and( - eq(sessions.id, sessionId), - eq(sessions.credentialDigestV1, digest), - or( - isNotNull(sessions.revokedAt), - isNull(sessions.expiresAt), - sql`${sessions.expiresAt} <= pg_catalog.clock_timestamp()`, - ), - )).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, - }) - const target = Array.isArray(fenced) ? fenced[0] : undefined - if (!target?.credentialDigest || !target.claimToken || !target.generation) return - const purge = cachePurgeTarget({ - attemptCount: target.attemptCount, - claimToken: target.claimToken, - credentialDigest: target.credentialDigest, - generation: target.generation, - legacyCredential: target.legacyCredential, - sessionId: target.sessionId, - }) - if (!purge) return - try { - await deleteSessionCacheTarget(purge) - await completeSessionCachePurge(purge) - } catch { - await deferSessionCachePurge(purge).catch(() => {}) - } -} - /** * Bounded worker maintenance for already-revoked sessions. It never reads Redis * to authorize a request and only derives cache keys from persisted digest/legacy @@ -481,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> { @@ -501,13 +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(() => {}) - await fenceSessionCacheWrite(digest, authorized.sessionId).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 } } From ea140824825d1a8ea500a50140baf71240aaeed9 Mon Sep 17 00:00:00 2001 From: Joncallim <64296013+Joncallim@users.noreply.github.com> Date: Mon, 27 Jul 2026 16:16:36 +0800 Subject: [PATCH 7/7] test: restore auth transaction mock boundary --- web/__tests__/auth.test.ts | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/web/__tests__/auth.test.ts b/web/__tests__/auth.test.ts index 1e9dc135..f434e847 100644 --- a/web/__tests__/auth.test.ts +++ b/web/__tests__/auth.test.ts @@ -571,6 +571,8 @@ describe('getSession — write-behind logic', () => { afterEach(() => { vi.useRealTimers() + mockDbTransaction.mockReset() + mockDbTransaction.mockImplementation(async (callback: (tx: unknown) => unknown) => callback(transactionClient())) }) it('writes cache entries inside a second locked transaction when lastSeenAt is recent', async () => { @@ -818,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', @@ -832,6 +834,7 @@ describe('login/finish — clone detection', () => { mockDbSelect .mockReturnValueOnce(chain([storedCredential])) .mockReturnValueOnce(chain([user])) + .mockReturnValueOnce(chain([{ state: 'expansion' }])) mockVerifyAuthenticationResponse.mockResolvedValue({ verified: true, @@ -850,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 () => {