diff --git a/docs/reliability-repair-progress.md b/docs/reliability-repair-progress.md index dba6308..a8cacb4 100644 --- a/docs/reliability-repair-progress.md +++ b/docs/reliability-repair-progress.md @@ -16,18 +16,38 @@ Evidence: ## D03 containment: overlapping tenant runtime state -Status: containment implemented on `codex/tenant-runtime-containment`; immutable provider/client context refactor still required before D03 can be marked complete. +Status: implemented and proven by upstream pull-request CI on head `ee377775030d27117b2568bc10426da8304c7a24`. -The Cloudflare scheduled and authenticated `/tick` job-drain entry points now share one per-isolate exclusive run gate. This prevents two overlapping SaaS drains in the same Worker isolate from entering `withTenantRuntime` concurrently. Separate Worker isolates do not share process globals. +The Cloudflare scheduled and authenticated `/tick` job-drain entry points share one per-isolate exclusive run gate. This prevents two overlapping SaaS drains in the same Worker isolate from entering the mutable tenant runtime concurrently. Separate Worker isolates do not share process globals. -This is deliberately a containment layer, not the target architecture. The remaining D03 work is to remove tenant-specific mutation of shared `config` and shared token-persistence callbacks and pass immutable tenant/connection context to provider and generation clients. +The containment regression deliberately interleaves two executions, proves maximum concurrent execution is one, and proves a rejected execution releases the gate. -Acceptance evidence required before this containment is considered proven: +## D03 runtime isolation: process-global config and token callbacks -- deliberately interleaved async runs never overlap inside the gate; -- a failed run does not poison the next queued run; -- the complete CI gate passes on the exact branch SHA. +Status: implemented on `codex/immutable-tenant-runtime`; upstream CI evidence is still required before this layer can be marked proven. + +The Worker now installs async-scoped accessors on the existing config object only after Cloudflare bindings have been copied into `process.env`. Each scheduled/authenticated SaaS drain then runs inside its own `AsyncLocalStorage` context. + +Within that context: + +- tenant config writes made by the existing `withTenantRuntime` path are copy-on-write and remain inside the current async execution instead of mutating process-global values; +- OpenAI, Cloudinary, Instagram, Facebook and provider modules that already read the shared config object transparently resolve the scoped values without a flag-day call-signature rewrite; +- Threads, LinkedIn and X token-persistence setters use scope-local callback slots when a SaaS runtime scope exists; +- token rotation updates only the current scoped config snapshot while the Supabase persistence callback remains attached to that same async execution; +- outside a SaaS runtime scope, the existing local single-tenant behaviour is preserved. + +The current `processPendingSupabaseJobs()` implementation remains serial, so tenant runtime mutation is restored between jobs inside a drain. The earlier exclusive run gate remains defence-in-depth but is no longer the only boundary preventing overlapping Worker invocations from sharing config or token callbacks. + +Acceptance evidence required before this layer is considered proven: + +- two deliberately overlapping runtime scopes resolve different OpenAI/provider credentials; +- Threads, LinkedIn and X token rotations invoke only the persistence callback belonging to their own scope; +- rotated credentials remain visible inside the originating scope but do not change the base config or the other scope; +- a failed scoped execution cannot leak its config into the next execution; +- the complete `npm run ci` gate passes on the exact branch SHA. + +No provider is re-enabled and no deployment is performed by this refactor. ## Next bounded repair -After containment is green, replace shared tenant runtime mutation with immutable request-scoped/provider-scoped context. Do not re-enable Meta publication as part of that refactor. +After runtime isolation is green, make missing tenant platform settings fail closed (D22), then proceed to atomic database claims/fencing and publication-attempt identity. Meta publication remains disabled until the publication ledger and provider-specific restoration work are ready. diff --git a/package.json b/package.json index 32c794c..47ad77b 100644 --- a/package.json +++ b/package.json @@ -6,7 +6,7 @@ "scripts": { "build": "tsx scripts/build.ts", "typecheck": "tsc --noEmit --project tsconfig.json", - "test": "npm run build && node dist/test/security-hardening.test.js && node dist/test/source-ssrf.test.js && node dist/test/browser-collector-ingest.test.js && node dist/test/slot-scheduler.test.js && node dist/test/daily-inventory-planner.test.js && node dist/test/refresh-queue-finalization.test.js && node dist/test/publish-summary.test.js && node dist/test/threads-refresh.test.js && node dist/test/linkedin-refresh.test.js && node dist/test/instagram-image-timeout.test.js && node dist/test/cost-control.test.js && node dist/test/recovery-scheduler.test.js && node dist/test/social-connector.test.js && node dist/test/meta-publication-boundary.test.js && node dist/test/cloudflare-health.test.js && node dist/test/exclusive-run-gate.test.js", + "test": "npm run build && node dist/test/security-hardening.test.js && node dist/test/source-ssrf.test.js && node dist/test/browser-collector-ingest.test.js && node dist/test/slot-scheduler.test.js && node dist/test/daily-inventory-planner.test.js && node dist/test/refresh-queue-finalization.test.js && node dist/test/publish-summary.test.js && node dist/test/threads-refresh.test.js && node dist/test/linkedin-refresh.test.js && node dist/test/instagram-image-timeout.test.js && node dist/test/cost-control.test.js && node dist/test/recovery-scheduler.test.js && node dist/test/social-connector.test.js && node dist/test/meta-publication-boundary.test.js && node dist/test/cloudflare-health.test.js && node dist/test/exclusive-run-gate.test.js && node dist/test/runtime-scope.test.js", "smoke:dist": "node dist/src/cli.js status", "ci": "npm run typecheck && npm test && npm run smoke:dist", "dev": "tsx src/agent.ts", diff --git a/src/cloudflare-worker.ts b/src/cloudflare-worker.ts index 6437123..0db1540 100644 --- a/src/cloudflare-worker.ts +++ b/src/cloudflare-worker.ts @@ -1,4 +1,5 @@ import { createExclusiveRunGate } from './exclusive-run-gate'; +import { installScopedConfig, runWithRuntimeScope } from './runtime-scope'; interface WorkerVersionMetadata { id: string; @@ -120,18 +121,28 @@ function applyCloudflareEnv(env: Env): void { async function executeScheduledTick(env: Env): Promise { applyCloudflareEnv(env); - const [{ processPendingSupabaseJobs, runSupabaseAutomationScheduler }, logger] = await Promise.all([ - import('./supabase-worker'), - import('./logger'), - ]); - - const schedulerStats = await runSupabaseAutomationScheduler(); - const stats = await processPendingSupabaseJobs(); - logger.info( - `Cloudflare scheduled worker tick | scheduled_fetch:${schedulerStats.fetchJobsEnqueued} scheduled_fill:${schedulerStats.slotFillJobsEnqueued} scheduled_publish:${schedulerStats.publishJobsEnqueued} inventory_plans:${schedulerStats.inventoryPlansChecked} inventory_alerts:${schedulerStats.inventoryAlerts} stale_failed:${schedulerStats.staleJobsFailed} claimed:${stats.claimed} completed:${stats.completed} failed:${stats.failed}` - ); - - return Response.json({ ok: true, schedulerStats, stats }); + // Config must be constructed after Worker bindings are copied into process.env. + // Instrument the shared object once, then keep every tenant mutation inside this + // async execution scope rather than process-global state. + const { default: config } = await import('../config'); + installScopedConfig(config); + + return runWithRuntimeScope(async () => { + const [{ processPendingSupabaseJobs, runSupabaseAutomationScheduler }, logger] = await Promise.all([ + import('./supabase-worker'), + import('./logger'), + ]); + + const schedulerStats = await runSupabaseAutomationScheduler(); + const stats = await processPendingSupabaseJobs(); + logger.info( + `Cloudflare scheduled worker tick | scheduled_fetch:${schedulerStats.fetchJobsEnqueued} scheduled_fill:${schedulerStats.slotFillJobsEnqueued} scheduled_publish:${schedulerStats.publishJobsEnqueued} inventory_plans:${schedulerStats.inventoryPlansChecked} inventory_alerts:${schedulerStats.inventoryAlerts} stale_failed:${schedulerStats.staleJobsFailed} claimed:${stats.claimed} completed:${stats.completed} failed:${stats.failed}` + ); + + return Response.json({ ok: true, schedulerStats, stats }); + }, { + execution: 'cloudflare_saas_tick', + }); } function runScheduledTick(env: Env): Promise { diff --git a/src/linkedin.ts b/src/linkedin.ts index e440395..f9f1822 100644 --- a/src/linkedin.ts +++ b/src/linkedin.ts @@ -6,6 +6,7 @@ import { PlatformPublishError, safeBodySnippet, } from './platform-errors'; +import { getScopedHandler, setScopedHandler } from './runtime-scope'; interface LinkedInPublishSuccess { id?: string; @@ -47,6 +48,7 @@ type LinkedInOAuthTokenPersistence = ( ) => void | Promise; const REFRESH_WINDOW_MS = 7 * 24 * 60 * 60 * 1000; +const LINKEDIN_TOKEN_PERSISTENCE_HANDLER = 'linkedin_oauth2_token_persistence'; let persistOAuth2TokensHandler: LinkedInOAuthTokenPersistence = persistOAuth2TokensToLocalRuntime; export function hasRefreshConfig(): boolean { @@ -109,6 +111,12 @@ export async function refreshOAuth2AccessToken( } export function setOAuth2TokenPersistence(handler: LinkedInOAuthTokenPersistence): () => void { + const scopedRestore = setScopedHandler( + LINKEDIN_TOKEN_PERSISTENCE_HANDLER, + handler + ); + if (scopedRestore) return scopedRestore; + const previous = persistOAuth2TokensHandler; persistOAuth2TokensHandler = handler; return () => { @@ -118,7 +126,9 @@ export function setOAuth2TokenPersistence(handler: LinkedInOAuthTokenPersistence export async function persistOAuth2Tokens(tokens: LinkedInOAuthTokenSet): Promise { applyOAuth2TokensToConfig(tokens); - await persistOAuth2TokensHandler(tokens); + const handler = getScopedHandler(LINKEDIN_TOKEN_PERSISTENCE_HANDLER) + || persistOAuth2TokensHandler; + await handler(tokens); } export async function refreshAndPersistOAuth2AccessToken(): Promise { diff --git a/src/runtime-scope.ts b/src/runtime-scope.ts new file mode 100644 index 0000000..eb434b9 --- /dev/null +++ b/src/runtime-scope.ts @@ -0,0 +1,118 @@ +import { AsyncLocalStorage } from 'node:async_hooks'; + +type RuntimeValueMap = Record; +type RuntimeHandler = (...args: any[]) => unknown; + +interface RuntimeScopeStore { + configOverlay: Readonly; + handlers: Readonly>; + metadata: Readonly>; +} + +const runtimeScope = new AsyncLocalStorage(); +const installedConfigObjects = new WeakMap>(); + +function frozenCopy>(value: T): Readonly { + return Object.freeze({ ...value }); +} + +function hasOwn(value: object, key: PropertyKey): boolean { + return Object.prototype.hasOwnProperty.call(value, key); +} + +export function hasRuntimeScope(): boolean { + return Boolean(runtimeScope.getStore()); +} + +export function runWithRuntimeScope( + fn: () => T, + metadata: Record = {} +): T { + const store: RuntimeScopeStore = { + configOverlay: frozenCopy({}), + handlers: Object.freeze({}), + metadata: frozenCopy(metadata), + }; + return runtimeScope.run(store, fn); +} + +export function getScopedConfigValue(key: string): { found: boolean; value?: unknown } { + const store = runtimeScope.getStore(); + if (!store || !hasOwn(store.configOverlay, key)) { + return { found: false }; + } + return { found: true, value: store.configOverlay[key] }; +} + +export function setScopedConfigValue(key: string, value: unknown): boolean { + const store = runtimeScope.getStore(); + if (!store) return false; + store.configOverlay = frozenCopy({ + ...store.configOverlay, + [key]: value, + }); + return true; +} + +export function getScopedHandler(key: string): T | undefined { + const store = runtimeScope.getStore(); + return store?.handlers[key] as T | undefined; +} + +export function setScopedHandler( + key: string, + handler: T +): (() => void) | undefined { + const store = runtimeScope.getStore(); + if (!store) return undefined; + + const hadPrevious = hasOwn(store.handlers, key); + const previous = store.handlers[key]; + store.handlers = Object.freeze({ + ...store.handlers, + [key]: handler, + }); + + return () => { + const next = { ...store.handlers } as Record; + if (hadPrevious && previous) next[key] = previous; + else delete next[key]; + store.handlers = Object.freeze(next); + }; +} + +export function installScopedConfig(config: T): T { + if (installedConfigObjects.has(config)) return config; + + const baseValues = new Map(); + installedConfigObjects.set(config, baseValues); + + for (const key of Object.keys(config)) { + baseValues.set(key, (config as Record)[key]); + Object.defineProperty(config, key, { + configurable: true, + enumerable: true, + get() { + const scoped = getScopedConfigValue(key); + return scoped.found ? scoped.value : baseValues.get(key); + }, + set(value: unknown) { + if (!setScopedConfigValue(key, value)) { + baseValues.set(key, value); + } + }, + }); + } + + return config; +} + +export function runtimeScopeMetadata(): Readonly> | null { + return runtimeScope.getStore()?.metadata || null; +} + +export const __test__ = { + getScopedConfigValue, + getScopedHandler, + hasRuntimeScope, +}; diff --git a/src/threads.ts b/src/threads.ts index 953571f..d377662 100644 --- a/src/threads.ts +++ b/src/threads.ts @@ -5,6 +5,7 @@ import { safeBodySnippet, } from './platform-errors'; import { assertCanonicalMetaPublicationPath } from './meta-publication-boundary'; +import { getScopedHandler, setScopedHandler } from './runtime-scope'; interface ThreadsTokenResponse extends GraphErrorResponse { access_token?: string; @@ -44,6 +45,7 @@ interface GraphErrorResponse { id?: string; } +const THREADS_TOKEN_PERSISTENCE_HANDLER = 'threads_token_persistence'; let persistThreadsTokenHandler: ThreadsTokenPersistence = persistThreadsTokenToLocalRuntime; export async function refreshLongLivedAccessToken(): Promise { @@ -224,6 +226,12 @@ export async function prepareAccessTokenForPublish(): Promise void { + const scopedRestore = setScopedHandler( + THREADS_TOKEN_PERSISTENCE_HANDLER, + handler + ); + if (scopedRestore) return scopedRestore; + const previous = persistThreadsTokenHandler; persistThreadsTokenHandler = handler; return () => { @@ -233,7 +241,9 @@ export function setTokenPersistence(handler: ThreadsTokenPersistence): () => voi export async function persistLongLivedAccessToken(tokens: ThreadsTokenSet): Promise { config.THREADS_ACCESS_TOKEN = tokens.accessToken; - await persistThreadsTokenHandler(tokens); + const handler = getScopedHandler(THREADS_TOKEN_PERSISTENCE_HANDLER) + || persistThreadsTokenHandler; + await handler(tokens); } function persistThreadsTokenToLocalRuntime(tokens: ThreadsTokenSet): void { diff --git a/src/x.ts b/src/x.ts index 3c39bd3..b670c7e 100644 --- a/src/x.ts +++ b/src/x.ts @@ -2,6 +2,7 @@ import * as crypto from 'node:crypto'; import config from '../config'; import { requestJson } from './http-client'; import { PlatformPublishError, safeBodySnippet } from './platform-errors'; +import { getScopedHandler, setScopedHandler } from './runtime-scope'; interface XApiErrorDetail { message?: string; @@ -61,6 +62,7 @@ export type XAuthMode = 'oauth1-user' | 'oauth2-user' | 'unconfigured'; export type XSafeAuthMode = 'x_oauth1_user_context' | 'x_oauth2_user_context' | 'unconfigured'; export type XErrorKind = 'publish-access-tier' | 'project-required' | 'auth' | 'other'; +const X_TOKEN_PERSISTENCE_HANDLER = 'x_oauth2_token_persistence'; let persistOAuth2TokensHandler: XOAuth2TokenPersistence = persistOAuth2TokensToLocalRuntime; function encodeOAuthComponent(value: string): string { @@ -558,6 +560,12 @@ function persistOAuth2TokensToLocalRuntime(tokens: XOAuth2TokenSet): void { } export function setOAuth2TokenPersistence(handler: XOAuth2TokenPersistence): () => void { + const scopedRestore = setScopedHandler( + X_TOKEN_PERSISTENCE_HANDLER, + handler + ); + if (scopedRestore) return scopedRestore; + const previous = persistOAuth2TokensHandler; persistOAuth2TokensHandler = handler; return () => { @@ -567,5 +575,7 @@ export function setOAuth2TokenPersistence(handler: XOAuth2TokenPersistence): () export async function persistOAuth2Tokens(tokens: XOAuth2TokenSet): Promise { applyOAuth2TokensToConfig(tokens); - await persistOAuth2TokensHandler(tokens); + const handler = getScopedHandler(X_TOKEN_PERSISTENCE_HANDLER) + || persistOAuth2TokensHandler; + await handler(tokens); } diff --git a/test/runtime-scope.test.ts b/test/runtime-scope.test.ts new file mode 100644 index 0000000..6a36049 --- /dev/null +++ b/test/runtime-scope.test.ts @@ -0,0 +1,146 @@ +import assert from 'node:assert/strict'; + +import config from '../config'; +import * as linkedin from '../src/linkedin'; +import { installScopedConfig, runWithRuntimeScope } from '../src/runtime-scope'; +import * as threads from '../src/threads'; +import * as x from '../src/x'; + +async function test(name: string, fn: () => Promise): Promise { + try { + await fn(); + console.log(`ok - ${name}`); + } catch (error) { + console.error(`not ok - ${name}`); + throw error; + } +} + +function deferred(): { promise: Promise; resolve: () => void } { + let resolve!: () => void; + const promise = new Promise(resolvePromise => { + resolve = resolvePromise; + }); + return { promise, resolve }; +} + +async function main(): Promise { + installScopedConfig(config); + + await test('overlapping tenant scopes do not share config or token persistence callbacks', async () => { + const base = { + openai: config.OPENAI_API_KEY, + threads: config.THREADS_ACCESS_TOKEN, + linkedin: config.LINKEDIN_TOKEN, + linkedinRefresh: config.LINKEDIN_REFRESH_TOKEN, + x: config.X_OAUTH2_ACCESS_TOKEN, + xRefresh: config.X_OAUTH2_REFRESH_TOKEN, + }; + const bothReady = deferred(); + const release = deferred(); + let readyCount = 0; + const persisted: string[] = []; + + async function tenantRun(tenant: 'a' | 'b'): Promise { + await runWithRuntimeScope(async () => { + config.OPENAI_API_KEY = `openai-${tenant}`; + config.THREADS_ACCESS_TOKEN = `threads-${tenant}`; + config.LINKEDIN_TOKEN = `linkedin-${tenant}`; + config.LINKEDIN_REFRESH_TOKEN = `linkedin-refresh-${tenant}`; + config.X_OAUTH2_ACCESS_TOKEN = `x-${tenant}`; + config.X_OAUTH2_REFRESH_TOKEN = `x-refresh-${tenant}`; + + const restoreThreads = threads.setTokenPersistence(async tokens => { + persisted.push(`${tenant}:threads:${tokens.accessToken}`); + }); + const restoreLinkedIn = linkedin.setOAuth2TokenPersistence(async tokens => { + persisted.push(`${tenant}:linkedin:${tokens.accessToken}`); + }); + const restoreX = x.setOAuth2TokenPersistence(async tokens => { + persisted.push(`${tenant}:x:${tokens.accessToken}`); + }); + + readyCount++; + if (readyCount === 2) bothReady.resolve(); + await release.promise; + + assert.equal(config.OPENAI_API_KEY, `openai-${tenant}`); + assert.equal(config.THREADS_ACCESS_TOKEN, `threads-${tenant}`); + assert.equal(config.LINKEDIN_TOKEN, `linkedin-${tenant}`); + assert.equal(config.X_OAUTH2_ACCESS_TOKEN, `x-${tenant}`); + + await threads.persistLongLivedAccessToken({ + accessToken: `threads-rotated-${tenant}`, + source: 'refresh', + }); + await linkedin.persistOAuth2Tokens({ + accessToken: `linkedin-rotated-${tenant}`, + refreshToken: `linkedin-refresh-rotated-${tenant}`, + }); + await x.persistOAuth2Tokens({ + accessToken: `x-rotated-${tenant}`, + refreshToken: `x-refresh-rotated-${tenant}`, + }); + + await Promise.resolve(); + assert.equal(config.THREADS_ACCESS_TOKEN, `threads-rotated-${tenant}`); + assert.equal(config.LINKEDIN_TOKEN, `linkedin-rotated-${tenant}`); + assert.equal(config.LINKEDIN_REFRESH_TOKEN, `linkedin-refresh-rotated-${tenant}`); + assert.equal(config.X_OAUTH2_ACCESS_TOKEN, `x-rotated-${tenant}`); + assert.equal(config.X_OAUTH2_REFRESH_TOKEN, `x-refresh-rotated-${tenant}`); + + restoreThreads(); + restoreLinkedIn(); + restoreX(); + }, { tenant }); + } + + const tenantA = tenantRun('a'); + const tenantB = tenantRun('b'); + await bothReady.promise; + release.resolve(); + await Promise.all([tenantA, tenantB]); + + assert.deepEqual(new Set(persisted), new Set([ + 'a:threads:threads-rotated-a', + 'a:linkedin:linkedin-rotated-a', + 'a:x:x-rotated-a', + 'b:threads:threads-rotated-b', + 'b:linkedin:linkedin-rotated-b', + 'b:x:x-rotated-b', + ])); + + assert.equal(config.OPENAI_API_KEY, base.openai); + assert.equal(config.THREADS_ACCESS_TOKEN, base.threads); + assert.equal(config.LINKEDIN_TOKEN, base.linkedin); + assert.equal(config.LINKEDIN_REFRESH_TOKEN, base.linkedinRefresh); + assert.equal(config.X_OAUTH2_ACCESS_TOKEN, base.x); + assert.equal(config.X_OAUTH2_REFRESH_TOKEN, base.xRefresh); + }); + + await test('a failed scoped execution cannot leak mutated config into the next execution', async () => { + const baseOpenAIKey = config.OPENAI_API_KEY; + + await assert.rejects( + runWithRuntimeScope(async () => { + config.OPENAI_API_KEY = 'should-not-leak'; + await Promise.resolve(); + throw new Error('expected failure'); + }, { tenant: 'failing' }), + /expected failure/ + ); + + assert.equal(config.OPENAI_API_KEY, baseOpenAIKey); + await runWithRuntimeScope(async () => { + assert.equal(config.OPENAI_API_KEY, baseOpenAIKey); + config.OPENAI_API_KEY = 'next-tenant'; + assert.equal(config.OPENAI_API_KEY, 'next-tenant'); + }, { tenant: 'next' }); + assert.equal(config.OPENAI_API_KEY, baseOpenAIKey); + }); +} + +main().catch(error => { + console.error(error); + process.exitCode = 1; +});