diff --git a/service/Dockerfile b/service/Dockerfile index 8ae43ef..762790d 100644 --- a/service/Dockerfile +++ b/service/Dockerfile @@ -16,10 +16,12 @@ RUN cd /temp/prod && bun install --frozen-lockfile --production FROM base AS builder COPY --from=install /temp/dev/node_modules node_modules COPY service/src ./src +COPY service/scripts ./scripts COPY shared /shared COPY service/tsconfig.json ./ RUN bun build ./src/file-server.ts --minify --outdir .build --target bun --external '@opentelemetry/*' RUN bun build ./src/api-server.ts --minify --outdir .build-api --target bun --external '@opentelemetry/*' +RUN bun build ./scripts/rehydrate-session-cache.ts --minify --outdir .build-migrations --target bun --external '@opentelemetry/*' RUN bun build ./src/worker-server.ts --minify --outdir .build-worker --target bun --external '@opentelemetry/*' RUN bun build ./src/egress-gateway.ts --minify --outdir .build-egress-gateway --target bun --external '@opentelemetry/*' @@ -37,6 +39,7 @@ ENV NODE_ENV=production WORKDIR /app COPY --from=install /temp/prod/node_modules ./node_modules COPY --from=builder /app/.build-api ./.build-api +COPY --from=builder /app/.build-migrations ./.build-migrations COPY --from=builder /app/src/*.py ./src/ CMD ["bun", "run", ".build-api/api-server.js"] diff --git a/service/Dockerfile.api b/service/Dockerfile.api index 0e0d1a9..419bdc9 100644 --- a/service/Dockerfile.api +++ b/service/Dockerfile.api @@ -18,9 +18,11 @@ RUN cd /temp/prod && bun install --frozen-lockfile --production FROM base AS builder COPY --from=install /temp/dev/node_modules node_modules COPY service/src ./src +COPY service/scripts ./scripts COPY shared /shared COPY service/tsconfig.json ./ RUN bun build ./src/api-server.ts --minify --outdir .build --target bun --external '@opentelemetry/*' +RUN bun build ./scripts/rehydrate-session-cache.ts --minify --outdir .build-migrations --target bun --external '@opentelemetry/*' # Production stage FROM oven/bun:1.3.14 AS production @@ -30,6 +32,7 @@ WORKDIR /app RUN apt-get update && apt-get install -y --no-install-recommends curl && rm -rf /var/lib/apt/lists/* COPY --from=install /temp/prod/node_modules ./node_modules COPY --from=builder /app/.build ./.build +COPY --from=builder /app/.build-migrations ./.build-migrations COPY --from=builder /app/src ./src # Copy matplotlib templates to /app where __dirname points at runtime COPY --from=builder /app/src/matplotlib.py /app/src/matplotlib-async.py ./ diff --git a/service/scripts/rehydrate-session-cache.ts b/service/scripts/rehydrate-session-cache.ts new file mode 100644 index 0000000..c21afa9 --- /dev/null +++ b/service/scripts/rehydrate-session-cache.ts @@ -0,0 +1,486 @@ +import IORedis from 'ioredis'; +import type * as tls from 'tls'; +import { isValidId } from '../src/utils'; +import { redisKeepAliveOptions } from '../src/redis-options'; + +/** + * One-time recovery for session ownership keys that expired before the + * associated files were removed. + * + * Build the input from a trusted source that provides the exact session id and + * expected session key pairs. Keep recovery manifests outside this repository + * because session keys can contain tenant and user identifiers. + * + * Pipe newline-delimited JSON to this script. Run it without `--apply` first: + * + * {"type":"source","environment":"example","region":"region-1","namespace":"codeapi","query_start_utc":"2026-01-01T00:00:00Z","query_end_utc":"2026-01-02T00:00:00Z"} + * {"session_id":"<21-character id>","expected_session_key":""} + * + * The apply path uses SET NX and never replaces an existing owner. + */ + +const DEFAULT_SESSION_CACHE_TTL_SECONDS = 86400; +const MAX_RECOVERY_CONTEXT_LENGTH = 128; +const MAX_RECOVERY_SESSION_KEY_LENGTH = 512; +const MAX_RECONNECT_ATTEMPTS = 5; +const RECONNECT_DELAY_MS = 2000; + +export interface RecoverySource { + type: 'source'; + environment: string; + region: string; + namespace: string; + query_start_utc: string; + query_end_utc: string; +} + +export interface RecoveryRecord { + session_id: string; + expected_session_key: string; +} + +export interface RecoveryStore { + get(key: string): Promise; + set( + key: string, + value: string, + expiryMode: 'EX', + ttlSeconds: number, + condition: 'NX', + ): Promise<'OK' | null>; +} + +export interface RecoverySummary { + input: number; + missing: number; + restored: number; + matching: number; + conflicts: number; +} + +export class RecoveryInterruptedError extends Error { + readonly summary: RecoverySummary; + + constructor(error: unknown, summary: RecoverySummary) { + super(error instanceof Error ? error.message : String(error)); + this.name = 'RecoveryInterruptedError'; + this.summary = { ...summary }; + } +} + +interface RecoveryScope { + environment: string; + region: string; + namespace: string; +} + +interface Options extends RecoveryScope { + apply: boolean; + inputPath?: string; + ttlSeconds: number; +} + +function usage(): string { + return `Usage: bun run /app/.build-migrations/rehydrate-session-cache.js [options] + +Restores missing Redis session ownership keys so retained files can be cleaned +up normally. Input is JSONL on stdin by default. The first record must be a +source header whose environment, region, and namespace match the configured +recovery scope. + +Options: + --apply Write missing keys. Without this flag, only inspect. + --input Read JSONL from a file instead of stdin. + --environment Expected source environment. Defaults to + SESSION_RECOVERY_ENVIRONMENT. + --region Expected source region. Defaults to + SESSION_RECOVERY_REGION. + --namespace Expected source namespace. Defaults to + SESSION_RECOVERY_NAMESPACE. + --ttl-seconds Redis TTL for restored keys. Defaults to + SESSION_CACHE_TTL or ${DEFAULT_SESSION_CACHE_TTL_SECONDS}. + --help Show this help. + +Keep recovery manifests outside the repository because expected_session_key +values can contain tenant and user identifiers. Add --apply only after +reviewing the dry-run summary. + +Exit codes: + 0 Completed without conflicts (dry-run missing keys are expected). + 1 Invalid input, configuration error, or interrupted Redis operation. + 2 Apply left a key missing or found an ownership conflict.`; +} + +function parsePositiveInteger(raw: string, name: string): number { + if (!/^\d+$/.test(raw)) { + throw new Error(`${name} must be a positive integer`); + } + const value = Number(raw); + if (!Number.isSafeInteger(value) || value <= 0) { + throw new Error(`${name} must be a positive integer`); + } + return value; +} + +function parseRecoveryContext(raw: string | undefined, name: string): string { + const value = raw?.trim(); + if (value == null || value === '') { + throw new Error(`${name} is required`); + } + if ( + value.length > MAX_RECOVERY_CONTEXT_LENGTH + || !/^[A-Za-z0-9][A-Za-z0-9_.:/-]*$/.test(value) + ) { + throw new Error( + `${name} must be ${MAX_RECOVERY_CONTEXT_LENGTH} or fewer safe characters`, + ); + } + return value; +} + +function optionValue(args: string[], index: number, name: string): string { + if (index + 1 >= args.length) { + throw new Error(`${name} requires a value`); + } + const value = args[index + 1]; + if (value === '' || value.startsWith('--')) { + throw new Error(`${name} requires a value`); + } + return value; +} + +export function parseOptions(args: string[], env: NodeJS.ProcessEnv = process.env): Options { + const configuredTtl = env.SESSION_CACHE_TTL?.trim(); + let ttlSeconds = configuredTtl != null && configuredTtl !== '' + ? parsePositiveInteger(configuredTtl, 'SESSION_CACHE_TTL') + : DEFAULT_SESSION_CACHE_TTL_SECONDS; + let environment = env.SESSION_RECOVERY_ENVIRONMENT; + let region = env.SESSION_RECOVERY_REGION; + let namespace = env.SESSION_RECOVERY_NAMESPACE; + let apply = false; + let inputPath: string | undefined; + + for (let index = 0; index < args.length; index += 1) { + const arg = args[index]; + switch (arg) { + case '--apply': + apply = true; + break; + case '--input': + inputPath = optionValue(args, index, '--input'); + index += 1; + break; + case '--environment': + environment = optionValue(args, index, '--environment'); + index += 1; + break; + case '--region': + region = optionValue(args, index, '--region'); + index += 1; + break; + case '--namespace': + namespace = optionValue(args, index, '--namespace'); + index += 1; + break; + case '--ttl-seconds': + ttlSeconds = parsePositiveInteger( + optionValue(args, index, '--ttl-seconds'), + '--ttl-seconds', + ); + index += 1; + break; + case '--help': + break; + default: + throw new Error(`Unknown option: ${arg}`); + } + } + + return { + apply, + inputPath, + environment: parseRecoveryContext(environment, 'Recovery environment'), + region: parseRecoveryContext(region, 'Recovery region'), + namespace: parseRecoveryContext(namespace, 'Recovery namespace'), + ttlSeconds, + }; +} + +function isUtcTimestamp(value: unknown): value is string { + return typeof value === 'string' + && /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z$/.test(value) + && Number.isFinite(Date.parse(value)); +} + +function parseRecoverySource(value: unknown, expected: RecoveryScope): RecoverySource { + if (typeof value !== 'object' || value === null || Array.isArray(value)) { + throw new Error('The first input record must be a recovery source header'); + } + const source = value as Record; + let environment: string; + let region: string; + let namespace: string; + try { + environment = parseRecoveryContext( + typeof source.environment === 'string' ? source.environment : undefined, + 'Source environment', + ); + region = parseRecoveryContext( + typeof source.region === 'string' ? source.region : undefined, + 'Source region', + ); + namespace = parseRecoveryContext( + typeof source.namespace === 'string' ? source.namespace : undefined, + 'Source namespace', + ); + } catch { + throw new Error('Recovery source contains invalid environment, region, or namespace'); + } + if ( + source.type !== 'source' + || environment !== expected.environment + || region !== expected.region + || namespace !== expected.namespace + || !isUtcTimestamp(source.query_start_utc) + || !isUtcTimestamp(source.query_end_utc) + ) { + throw new Error( + `Recovery source must match ${expected.environment}/${expected.region}/${expected.namespace}`, + ); + } + if (Date.parse(source.query_start_utc) >= Date.parse(source.query_end_utc)) { + throw new Error('Recovery source query_start_utc must be before query_end_utc'); + } + return { + type: 'source', + environment, + region, + namespace, + query_start_utc: source.query_start_utc, + query_end_utc: source.query_end_utc, + }; +} + +function isRecoveryRecord(value: unknown): value is RecoveryRecord { + if (typeof value !== 'object' || value === null || Array.isArray(value)) { + return false; + } + const record = value as Record; + return typeof record.session_id === 'string' + && isValidId(record.session_id) + && typeof record.expected_session_key === 'string' + && record.expected_session_key.length > 0 + && record.expected_session_key.length <= MAX_RECOVERY_SESSION_KEY_LENGTH + && !/\p{Cc}/u.test(record.expected_session_key); +} + +export function parseRecoveryManifest( + input: string, + expected: RecoveryScope, +): { source: RecoverySource; records: RecoveryRecord[] } { + const bySessionId = new Map(); + const lines = input.split(/\r?\n/); + let source: RecoverySource | undefined; + + for (let index = 0; index < lines.length; index += 1) { + const line = lines[index].trim(); + if (!line || line.startsWith('#')) { + continue; + } + + let value: unknown; + try { + value = JSON.parse(line); + } catch { + throw new Error(`Input line ${index + 1} is not valid JSON`); + } + if (!source) { + source = parseRecoverySource(value, expected); + continue; + } + if (!isRecoveryRecord(value)) { + throw new Error( + `Input line ${index + 1} must contain a valid session_id and expected_session_key`, + ); + } + + const previous = bySessionId.get(value.session_id); + if (previous && previous.expected_session_key !== value.expected_session_key) { + throw new Error(`Input contains conflicting owners for session ${value.session_id}`); + } + bySessionId.set(value.session_id, value); + } + + if (bySessionId.size === 0) { + throw new Error('Input contains no recovery records'); + } + if (!source) { + throw new Error('Input contains no recovery source header'); + } + return { source, records: [...bySessionId.values()] }; +} + +export async function recoverSessionCache( + records: RecoveryRecord[], + store: RecoveryStore, + options: Pick, +): Promise { + const summary: RecoverySummary = { + input: records.length, + missing: 0, + restored: 0, + matching: 0, + conflicts: 0, + }; + + for (const record of records) { + try { + const redisKey = `session:${record.session_id}`; + const current = await store.get(redisKey); + if (current === record.expected_session_key) { + summary.matching += 1; + continue; + } + if (current !== null) { + summary.conflicts += 1; + // eslint-disable-next-line no-console + console.error(`Conflict: ${record.session_id} already has a different owner`); + continue; + } + + if (!options.apply) { + summary.missing += 1; + continue; + } + + const result = await store.set( + redisKey, + record.expected_session_key, + 'EX', + options.ttlSeconds, + 'NX', + ); + if (result === 'OK') { + summary.restored += 1; + continue; + } + + const racedValue = await store.get(redisKey); + if (racedValue === record.expected_session_key) { + summary.matching += 1; + } else if (racedValue === null) { + summary.missing += 1; + // eslint-disable-next-line no-console + console.error(`Missing: ${record.session_id} disappeared during recovery`); + } else { + summary.conflicts += 1; + // eslint-disable-next-line no-console + console.error(`Conflict: ${record.session_id} changed during recovery`); + } + } catch (error) { + throw new RecoveryInterruptedError(error, summary); + } + } + + return summary; +} + +function redisRetryStrategy(times: number): number | null { + return times > MAX_RECONNECT_ATTEMPTS ? null : RECONNECT_DELAY_MS; +} + +function createRedisClient(): IORedis { + const options = { + host: process.env.REDIS_HOST ?? 'redis', + port: Number(process.env.REDIS_PORT) || 6379, + password: process.env.REDIS_PASSWORD, + maxRetriesPerRequest: 1, + retryStrategy: redisRetryStrategy, + enableReadyCheck: true, + connectTimeout: 10000, + disconnectTimeout: 2000, + ...redisKeepAliveOptions(), + tls: process.env.REDIS_TLS === 'true' + ? { rejectUnauthorized: false } as tls.ConnectionOptions + : undefined, + ...(process.env.REDIS_USE_ALTERNATIVE_DNS_LOOKUP === 'true' + ? { + dnsLookup: ( + address: string, + callback: (err: Error | null, addr: string) => void, + ): void => callback(null, address), + } + : {}), + }; + return new IORedis(options); +} + +async function readInput(inputPath?: string): Promise { + if (inputPath != null && inputPath !== '') { + return Bun.file(inputPath).text(); + } + if (process.stdin.isTTY === true) { + throw new Error('No manifest input: pipe JSONL on stdin or use --input '); + } + return Bun.stdin.text(); +} + +function recoveryOutput( + options: Options, + source: RecoverySource, + summary: RecoverySummary, +): Record { + return { + mode: options.apply ? 'apply' : 'dry-run', + environment: source.environment, + region: source.region, + namespace: source.namespace, + ttlSeconds: options.ttlSeconds, + ...summary, + }; +} + +export async function main(args: string[] = process.argv.slice(2)): Promise { + if (args.includes('--help')) { + // eslint-disable-next-line no-console + console.log(usage()); + return 0; + } + + let client: IORedis | undefined; + let options: Options | undefined; + let source: RecoverySource | undefined; + try { + options = parseOptions(args); + const manifest = parseRecoveryManifest(await readInput(options.inputPath), options); + source = manifest.source; + client = createRedisClient(); + const summary = await recoverSessionCache(manifest.records, client, options); + // eslint-disable-next-line no-console + console.log(JSON.stringify(recoveryOutput(options, source, summary))); + const incompleteApply = options.apply && summary.missing > 0; + return summary.conflicts === 0 && !incompleteApply ? 0 : 2; + } catch (error) { + if ( + error instanceof RecoveryInterruptedError + && options != null + && source != null + ) { + // eslint-disable-next-line no-console + console.log(JSON.stringify(recoveryOutput(options, source, error.summary))); + } + // eslint-disable-next-line no-console + console.error(error instanceof Error ? error.message : String(error)); + return 1; + } finally { + if (client != null) { + await client.quit().catch(() => client?.disconnect()); + } + } +} + +if (require.main === module) { + void main().then((exitCode) => { + process.exitCode = exitCode; + }); +} diff --git a/service/src/rehydrate-session-cache.test.ts b/service/src/rehydrate-session-cache.test.ts new file mode 100644 index 0000000..22e391f --- /dev/null +++ b/service/src/rehydrate-session-cache.test.ts @@ -0,0 +1,244 @@ +import { describe, expect, it } from 'bun:test'; +import { + parseOptions, + parseRecoveryManifest, + recoverSessionCache, + RecoveryInterruptedError, + type RecoverySource, + type RecoveryStore, +} from '../scripts/rehydrate-session-cache'; + +const SESSION_ID = 'ABCDEFGHIJKLMNOPQRSTU'; +const OTHER_SESSION_ID = 'ZYXWVUTSRQPONMLKJIHGF'; +const SESSION_KEY = 'example-tenant:user:example-user'; +const SCOPE = { + environment: 'example', + region: 'region-1', + namespace: 'codeapi', +}; +const SOURCE = { + type: 'source', + ...SCOPE, + query_start_utc: '2026-01-01T00:00:00Z', + query_end_utc: '2026-01-02T00:00:00Z', +} as const satisfies RecoverySource; + +class MemoryStore implements RecoveryStore { + readonly values = new Map(); + + async get(key: string): Promise { + return this.values.get(key) ?? null; + } + + async set( + key: string, + value: string, + _expiryMode: 'EX', + _ttlSeconds: number, + _condition: 'NX', + ): Promise<'OK' | null> { + if (this.values.has(key)) { + return null; + } + this.values.set(key, value); + return 'OK'; + } +} + +describe('rehydrate-session-cache', () => { + it('parses, validates, and deduplicates JSONL records', () => { + const input = [ + '# trusted recovery source', + JSON.stringify(SOURCE), + JSON.stringify({ session_id: SESSION_ID, expected_session_key: SESSION_KEY }), + JSON.stringify({ session_id: SESSION_ID, expected_session_key: SESSION_KEY }), + '', + ].join('\n'); + + expect(parseRecoveryManifest(input, SCOPE)).toEqual({ + source: SOURCE, + records: [{ session_id: SESSION_ID, expected_session_key: SESSION_KEY }], + }); + }); + + it('rejects a manifest outside the configured recovery scope', () => { + const input = [ + JSON.stringify(SOURCE), + JSON.stringify({ session_id: SESSION_ID, expected_session_key: SESSION_KEY }), + ].join('\n'); + + expect(() => parseRecoveryManifest(input, { ...SCOPE, region: 'region-2' })).toThrow( + 'Recovery source must match example/region-2/codeapi', + ); + }); + + it('rejects invalid recovery context values', () => { + expect(() => parseOptions([ + '--environment', 'example', + '--region', 'region with spaces', + '--namespace', 'codeapi', + ], {})).toThrow('Recovery region'); + }); + + it('rejects conflicting owners before connecting to Redis', () => { + const input = [ + JSON.stringify(SOURCE), + JSON.stringify({ session_id: SESSION_ID, expected_session_key: SESSION_KEY }), + JSON.stringify({ session_id: SESSION_ID, expected_session_key: 'tenant-id:user:other' }), + ].join('\n'); + + expect(() => parseRecoveryManifest(input, SCOPE)).toThrow('conflicting owners'); + }); + + it('accepts a composite session key longer than one resource id', () => { + const longSessionKey = `${'t'.repeat(128)}:skill:${'s'.repeat(128)}:v:1`; + const input = [ + JSON.stringify(SOURCE), + JSON.stringify({ session_id: SESSION_ID, expected_session_key: longSessionKey }), + ].join('\n'); + + expect(parseRecoveryManifest(input, SCOPE).records).toEqual([ + { session_id: SESSION_ID, expected_session_key: longSessionKey }, + ]); + }); + + it('accepts session keys containing identity punctuation and Unicode', () => { + const emittedSessionKey = 'tenant+東京@example.com/user:user+東京@example.com'; + const input = [ + JSON.stringify(SOURCE), + JSON.stringify({ session_id: SESSION_ID, expected_session_key: emittedSessionKey }), + ].join('\n'); + + expect(parseRecoveryManifest(input, SCOPE).records).toEqual([ + { session_id: SESSION_ID, expected_session_key: emittedSessionKey }, + ]); + }); + + it('rejects empty session keys and control characters', () => { + for (const expectedSessionKey of ['', 'tenant:user:user\nid']) { + const input = [ + JSON.stringify(SOURCE), + JSON.stringify({ session_id: SESSION_ID, expected_session_key: expectedSessionKey }), + ].join('\n'); + + expect(() => parseRecoveryManifest(input, SCOPE)).toThrow( + 'must contain a valid session_id and expected_session_key', + ); + } + }); + + it('accepts recovery scope from arguments or configuration', () => { + expect(parseOptions([ + '--environment', 'argument-env', + '--region', 'argument-region', + '--namespace', 'argument-namespace', + '--ttl-seconds', '7200', + ], {})).toMatchObject({ + environment: 'argument-env', + region: 'argument-region', + namespace: 'argument-namespace', + ttlSeconds: 7200, + }); + expect(parseOptions([], { + SESSION_RECOVERY_ENVIRONMENT: 'configured-env', + SESSION_RECOVERY_REGION: 'configured-region', + SESSION_RECOVERY_NAMESPACE: 'configured-namespace', + SESSION_CACHE_TTL: '86400', + })).toMatchObject({ + environment: 'configured-env', + region: 'configured-region', + namespace: 'configured-namespace', + ttlSeconds: 86400, + }); + }); + + it('requires the recovery scope', () => { + expect(() => parseOptions([], {})).toThrow('Recovery environment'); + }); + + it('does not write during a dry run', async () => { + const store = new MemoryStore(); + const summary = await recoverSessionCache( + [{ session_id: SESSION_ID, expected_session_key: SESSION_KEY }], + store, + { apply: false, ttlSeconds: 86400 }, + ); + + expect(summary).toEqual({ input: 1, missing: 1, restored: 0, matching: 0, conflicts: 0 }); + expect(store.values.size).toBe(0); + }); + + it('restores only absent keys and reports existing owners', async () => { + const store = new MemoryStore(); + store.values.set(`session:${OTHER_SESSION_ID}`, 'tenant-id:user:someone-else'); + + const summary = await recoverSessionCache( + [ + { session_id: SESSION_ID, expected_session_key: SESSION_KEY }, + { session_id: OTHER_SESSION_ID, expected_session_key: SESSION_KEY }, + ], + store, + { apply: true, ttlSeconds: 86400 }, + ); + + expect(summary).toEqual({ input: 2, missing: 0, restored: 1, matching: 0, conflicts: 1 }); + expect(store.values.get(`session:${SESSION_ID}`)).toBe(SESSION_KEY); + expect(store.values.get(`session:${OTHER_SESSION_ID}`)).toBe('tenant-id:user:someone-else'); + }); + + it('preserves partial counts when a Redis operation fails', async () => { + let reads = 0; + const store: RecoveryStore = { + async get(): Promise { + reads += 1; + if (reads === 1) { + return SESSION_KEY; + } + throw new Error('Redis unavailable'); + }, + async set(): Promise<'OK' | null> { + throw new Error('unexpected set'); + }, + }; + + try { + await recoverSessionCache( + [ + { session_id: SESSION_ID, expected_session_key: SESSION_KEY }, + { session_id: OTHER_SESSION_ID, expected_session_key: SESSION_KEY }, + ], + store, + { apply: true, ttlSeconds: 86400 }, + ); + throw new Error('expected recovery to fail'); + } catch (error) { + expect(error).toBeInstanceOf(RecoveryInterruptedError); + expect((error as RecoveryInterruptedError).summary).toEqual({ + input: 2, + missing: 0, + restored: 0, + matching: 1, + conflicts: 0, + }); + } + }); + + it('reports a key that disappears during an apply race as missing', async () => { + const store: RecoveryStore = { + async get(): Promise { + return null; + }, + async set(): Promise { + return null; + }, + }; + + const summary = await recoverSessionCache( + [{ session_id: SESSION_ID, expected_session_key: SESSION_KEY }], + store, + { apply: true, ttlSeconds: 86400 }, + ); + + expect(summary).toEqual({ input: 1, missing: 1, restored: 0, matching: 0, conflicts: 0 }); + }); +});