From c657892a429ad0e191eb9d42e04f4b3c427041bc Mon Sep 17 00:00:00 2001 From: Ayobami Haastrup <47716486+AyobamiH@users.noreply.github.com> Date: Sat, 12 Sep 2026 15:56:49 -0700 Subject: [PATCH 1/4] Cancel source extraction deadlines and isolate scheduler failures --- docs/implementation-checkpoint-20260912.md | 113 ++++ docs/reliability-repair-progress.md | 4 + package.json | 4 +- src/ai.ts | 15 +- src/http-client.ts | 31 +- src/supabase-worker.ts | 721 +++++++++++---------- test/http-cancellation.test.ts | 118 ++++ test/publication-executor.test.ts | 2 +- test/publish-summary.test.ts | 2 +- test/scheduler-isolation.test.ts | 86 +++ 10 files changed, 746 insertions(+), 350 deletions(-) create mode 100644 docs/implementation-checkpoint-20260912.md create mode 100644 test/http-cancellation.test.ts create mode 100644 test/scheduler-isolation.test.ts diff --git a/docs/implementation-checkpoint-20260912.md b/docs/implementation-checkpoint-20260912.md new file mode 100644 index 0000000..4ecd98c --- /dev/null +++ b/docs/implementation-checkpoint-20260912.md @@ -0,0 +1,113 @@ +# Implementation checkpoint — 12 September 2026 + +## Recovered baseline + +The September 6 local branch is superseded. Continue from backend +`e9eb7fa8b2d10146480377e8c51854991c728194` and app/schema +`19a402ca28c61b53d17a6b12f14ca047ed5917e9`. Reapplying that old branch would +regress the manual deployment controls and duplicate already merged work. + +The persistent staging rehearsal and production schema cutover are complete +according to the September 9 rollout attestation in +[`reliability-repair-progress.md`](reliability-repair-progress.md). The guarded +[deployment run](https://github.com/OneClickPostFactory/social-agents/actions/runs/34354662402) +confirms Worker version `773bb17d-59ef-484d-ae74-2f7c004e3447`, from +`3a27a993742151a6558089c9ab0ade6b1a762ba4`, with generation and provider dispatch +disabled. This is an inert canary, not end-to-end product acceptance. No new +production deployment, schema change, paid API call or social post was made +during this continuation. + +## Repairs in this continuation + +The worker cancels angle extraction through the HTTP transport, response read +and retry wait. It awaits settlement before releasing the source claim. This +prevents abandoned local work; cancellation cannot guarantee that a provider +has not already incurred a charge. Durable generation accounting remains a +separate release requirement. + +Each scheduling stage, tenant and due queue row has an error boundary. A failed +fetch, tenant lookup, recovery query or log write cannot suppress independent +publish work. `SchedulerStats.errors` reports failures by stage. Existing +entitlement, canary, immutable hold and publication ledger gates still apply. +Scheduling remains serial; this change does not provide durable fairness, +enqueue uniqueness, or an overall tick deadline. + +The build uses Node's `--import tsx` entrypoint, avoiding the tsx CLI's unnecessary +IPC listener. Both new regression files are included in the required test command. +Two existing static wiring assertions now accept the scheduler's stage wrapper; +their prohibition on legacy publication reconciliation remains intact. + +The companion app repair checks billing database results, requires immutable +customer ownership, makes X OAuth state single use, and permits authenticated +credential clearing without paid access. Webhook acknowledgement and delivery +order are separate concerns; the former is repaired and the latter still needs +the durable inbox/projection release described below. + +## Reviewed completion plan + +Every phase must leave its current consumer, schema contract, negative tests, +rollout controls and recovery procedure executable before moving to the next. +Merge and deployment are distinct checkpoints. + +| Phase | Current state | Remaining implementation and acceptance gate | +| --- | --- | --- | +| Release identity and CI | Merged; inert deployment verified | Deploy only an approved exact SHA through the existing guarded workflow. | +| Tenant runtime and database claims | Scope isolation, source/angle fencing and publication ledger merged | Add durable job-enqueue uniqueness and terminal job ownership; demonstrate competing Workers cannot create duplicate logical jobs or finish a replacement owner's job. | +| Worker cancellation and isolation | Implemented in this repair | Full CI and exact-commit PR checks; retain all publication ambiguity tests. | +| Billing | Error acknowledgement, ownership and period compatibility repaired in companion app branch | Environment-owned configuration and environment/customer mapping; durable event inbox; serial canonical Stripe reconciliation; atomic projection/audit commit; checkout intent and lifetime trial eligibility. Test duplicate, reversed, concurrent and crash/replay deliveries without changing another tenant or environment. | +| Connection lifecycle | Single-use X callback and unpaid credential clearing implemented | Credential versions and compare-and-swap persistence for callback, refresh, reconnect and disconnect across app and Worker. A delayed refresh/callback must never restore a disconnected credential. | +| Generation | Claims and cancellation exist | Reserve a durable generation operation and budget before any paid request; classify ambiguous attempts; preserve outputs; enforce one spend for a logical operation across restarts. Never infer a durable budget from best-effort logs. | +| Scheduling | Tenant/stage failures isolated | Durable cursor, bounded tenant batches, per-tenant fairness, exact enqueue identity and queue-age monitoring. Test a healthy tenant behind more than one page of blocked rows. | +| Providers and ingestion | Existing retirement/quarantine boundaries enforced | Validate the installed Reddit connector end to end. Establish the hosted tenant contract for Threads/Instagram without reviving retired adapters. Verify current LinkedIn compatibility and tenant X identity. Facebook needs an explicit supported target; legacy Groups and frontend Page credentials are not interchangeable. | +| Product status and release acceptance | Health and ledger provide partial truth | UI consumes connection, generation and per-platform publication states, including unknown outcomes. Align marketing with enabled capabilities. Complete one authorised scheduled canary per supported provider, then a monitored soak and recovery drill. | + +## Original defect register reconciliation + +“Repaired here” means code and regression coverage on the repair branch. It does +not mean deployed. “Merged” refers to the recovered baseline; it does not prove +live account compatibility or every failure mode. + +| Finding | Current disposition | +| --- | --- | +| D01 release/deploy verification | Merged: manual exact-SHA gate and full CI. | +| D02 hosted Meta publisher gap | Open; retired adapters remain closed. | +| D03 shared tenant runtime | Runtime scope isolation merged; connection versioning remains. | +| D04 angle/source races | Atomic database claims and fenced finalisation merged. | +| D05 job creation races | Open for worker-owned enqueue paths. | +| D06 ambiguous database mutation retries | Merged: ordinary mutations single attempt; explicit idempotent RPC retry. | +| D07 provider success followed by bookkeeping failure | Publication ledger/executor merged; ambiguity cannot authorise resend. | +| D08 source URL used as publication identity | Removed from production reconciliation; exact queue/attempt identity merged. | +| D09 stale recovery ownership | Publication/source/angle protections merged; terminal agent-job fencing remains. | +| D10 uncancelled extraction deadline | Repaired here, including transport/body/backoff cancellation tests. | +| D11 generation spending from best-effort logs | Open: durable budget and generation operation ledger required. | +| D12 scheduler failure propagation | Repaired here at stage, tenant and row boundaries. | +| D13 first-page starvation | Open: durable fair scheduling and bounded cursors required. | +| D14 incomplete regression command | Full existing suite merged; new tests added to the same required command. | +| D15 readiness/typed UI truth | Health repaired; complete application state projection remains. | +| D16 ignored billing writes | Companion repair returns failures and requires affected profile rows; durable inbox remains. | +| D17 Stripe arrival-order projection and late deduplication | Open; trial reminders are now audit-only, but general ordering still needs transactional reconciliation. | +| D18 Stripe environment selection/partition | Incoming signed mode checked in companion repair; deployment selection and database partition remain open. | +| D19 first email customer fallback | Removed in companion repair; metadata conflicts fail closed for support review. | +| D20 duplicate checkout/trials | Open; customer-create idempotency alone is not checkout/trial protection. | +| D21 paid access required to clear credentials | App/API restriction removed in companion repair; disconnect/refresh fencing remains. | +| D22 implicit platform enablement | Explicit stored-true policy merged in app and Worker. | +| D23 concurrent X callback reuse | Atomic unexpired DELETE RETURNING implemented in companion repair. | +| D24 LinkedIn compatibility | Open; verify current Posts API contract before claiming availability. | +| D25 unsupported learning/marketing claims | Open: reconcile product promises with implemented evidence. | +| Additional: Stripe period fields | Companion repair reads current subscription item periods, with legacy-event compatibility. | + +## Evidence and limitations + +Local worker validation: `npm run ci` passes, including the new transport and +real scheduler failure tests. App validation and the exact review links are +recorded in the companion repository's September 12 checkpoint. + +Existing production data was not read again during the source repairs. The +148-row/11-hold inventory is the September 9 attestation, not a new snapshot. +The 16 real-Postgres scenarios are prior rehearsal evidence, not new tests run +in this continuation. The new tests intercept external transport and make no +paid or public calls. + +There is no defensible unconditional guarantee that the entire service will +work after these patches. Completion means measured success for the declared +platforms and failure/recovery contracts above, followed by monitored operation. diff --git a/docs/reliability-repair-progress.md b/docs/reliability-repair-progress.md index 64d2b85..a0c5897 100644 --- a/docs/reliability-repair-progress.md +++ b/docs/reliability-repair-progress.md @@ -1,5 +1,9 @@ # Reliability repair progress +Current continuation: [12 September implementation checkpoint](implementation-checkpoint-20260912.md). +It reconciles the superseded September 6 branch, current repairs and remaining +release gates. The September 9 deployment below remains an inert canary. + This file records implementation and rollout status. Exact-SHA CI receipts remain on the relevant pull requests and deployment runs. A passing component test is not live account evidence. ## Sequence 1: release identity and complete CI gate diff --git a/package.json b/package.json index 5c59d97..14edf18 100644 --- a/package.json +++ b/package.json @@ -4,9 +4,9 @@ "description": "Tenant-scoped source records -> OpenAI -> LinkedIn + Threads + X + Instagram autoposter with durable Cloudinary image persistence.", "main": "dist/src/agent.js", "scripts": { - "build": "tsx scripts/build.ts", + "build": "node --import 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/canary-policy.test.js && node dist/test/exclusive-run-gate.test.js && node dist/test/runtime-scope.test.js && node dist/test/tenant-platform-policy.test.js && node dist/test/supabase-client-retry.test.js && node dist/test/worker-claims.test.js && node dist/test/publication-ledger.test.js && node dist/test/publication-outcome.test.js && node dist/test/publication-executor.test.js && node dist/test/provider-single-dispatch.test.js && node dist/test/publication-receipts.test.js && node dist/test/legacy-revision-hold.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/canary-policy.test.js && node dist/test/exclusive-run-gate.test.js && node dist/test/runtime-scope.test.js && node dist/test/tenant-platform-policy.test.js && node dist/test/supabase-client-retry.test.js && node dist/test/worker-claims.test.js && node dist/test/publication-ledger.test.js && node dist/test/publication-outcome.test.js && node dist/test/publication-executor.test.js && node dist/test/provider-single-dispatch.test.js && node dist/test/publication-receipts.test.js && node dist/test/legacy-revision-hold.test.js && node dist/test/http-cancellation.test.js && node dist/test/scheduler-isolation.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/ai.ts b/src/ai.ts index fef9085..ca89386 100644 --- a/src/ai.ts +++ b/src/ai.ts @@ -543,8 +543,10 @@ async function chatComplete( userPrompt: string, maxTokens = 500, temperature = 0.8, - usageContext?: OpenAIUsageContext + usageContext?: OpenAIUsageContext, + signal?: AbortSignal ): Promise { + signal?.throwIfAborted(); const model = config.OPENAI_MODEL || 'gpt-4o'; const startedAt = Date.now(); const inputSizeEstimate = systemPrompt.length + userPrompt.length; @@ -566,6 +568,7 @@ async function chatComplete( }); try { + signal?.throwIfAborted(); const { data } = await requestJson('https://api.openai.com/v1/chat/completions', { method: 'POST', headers: { @@ -574,6 +577,7 @@ async function chatComplete( }, body, timeoutMs: config.HTTP_TIMEOUT_MS, + signal, }); if (data.error) { throw new Error('OpenAI: ' + (data.error.message || 'Unknown error')); @@ -763,9 +767,10 @@ function chatCompleteJson( userPrompt: string, maxTokens = 500, temperature = 0.6, - usageContext?: OpenAIUsageContext + usageContext?: OpenAIUsageContext, + signal?: AbortSignal ): Promise { - return chatComplete(systemPrompt, userPrompt, maxTokens, temperature, usageContext) + return chatComplete(systemPrompt, userPrompt, maxTokens, temperature, usageContext, signal) .then(raw => extractJson(raw)); } @@ -1672,6 +1677,7 @@ export async function extractSourceBank( contentStrategyProfile?: unknown; contentStrategyProfileVersion?: string | null; usageContext?: OpenAIUsageContext; + signal?: AbortSignal; } = {} ): Promise { const source = [post.title, post.selftext].filter(Boolean).join('\n\n').substring(0, 2400); @@ -1728,7 +1734,8 @@ ${source} ...options.usageContext, promptVersion: options.usageContext?.promptVersion || TEXT_PROMPT_VERSION, stage: OPENAI_TEXT_ANGLE_EXTRACTION_STAGE, - } + }, + options.signal ); return normalizeSourceExtraction(parsed, post); diff --git a/src/http-client.ts b/src/http-client.ts index 19b63e8..0bbd09a 100644 --- a/src/http-client.ts +++ b/src/http-client.ts @@ -7,6 +7,8 @@ export interface HttpJsonOptions { timeoutMs?: number; retryCount?: number; retryDelayMs?: number; + /** Cancels the complete operation, including response reads and retry delays. */ + signal?: AbortSignal; } export interface HttpJsonResponse { @@ -18,8 +20,19 @@ export interface HttpJsonResponse { const DEFAULT_TIMEOUT_MS = Number.parseInt(process.env.HTTP_TIMEOUT_MS || '15000', 10); -function sleep(ms: number): Promise { - return new Promise(resolve => setTimeout(resolve, ms)); +function sleep(ms: number, signal?: AbortSignal): Promise { + signal?.throwIfAborted(); + return new Promise((resolve, reject) => { + const onAbort = () => { + clearTimeout(timer); + reject(signal!.reason); + }; + const timer = setTimeout(() => { + signal?.removeEventListener('abort', onAbort); + resolve(); + }, ms); + signal?.addEventListener('abort', onAbort, { once: true }); + }); } function isRetryable(status: number): boolean { @@ -34,11 +47,15 @@ export async function requestJson(url: string, options: HttpJsonOptions = {}) timeoutMs = DEFAULT_TIMEOUT_MS, retryCount = 0, retryDelayMs = 250, + signal, } = options; for (let attempt = 0; attempt <= retryCount; attempt += 1) { + signal?.throwIfAborted(); const controller = new AbortController(); const timeout = setTimeout(() => controller.abort(), timeoutMs); + const onAbort = () => controller.abort(signal!.reason); + signal?.addEventListener('abort', onAbort, { once: true }); try { const response = await fetch(url, { @@ -49,6 +66,7 @@ export async function requestJson(url: string, options: HttpJsonOptions = {}) }); const text = await response.text(); + signal?.throwIfAborted(); let data: T; try { @@ -62,7 +80,7 @@ export async function requestJson(url: string, options: HttpJsonOptions = {}) } if (!response.ok && attempt < retryCount && isRetryable(response.status)) { - await sleep(retryDelayMs * (attempt + 1)); + await sleep(retryDelayMs * (attempt + 1), signal); continue; } @@ -73,18 +91,21 @@ export async function requestJson(url: string, options: HttpJsonOptions = {}) rawText: text, }; } catch (error) { + // Caller cancellation is terminal even when this request normally retries. + signal?.throwIfAborted(); if (attempt < retryCount) { - await sleep(retryDelayMs * (attempt + 1)); + await sleep(retryDelayMs * (attempt + 1), signal); continue; } const message = error instanceof Error ? error.message : String(error); - if (message === 'This operation was aborted') { + if (controller.signal.aborted) { upstreamFailure('Upstream request timed out', 'UPSTREAM_TIMEOUT'); } upstreamFailure(`Upstream request failed: ${message}`, 'UPSTREAM_REQUEST_FAILED'); } finally { clearTimeout(timeout); + signal?.removeEventListener('abort', onAbort); } } diff --git a/src/supabase-worker.ts b/src/supabase-worker.ts index fc7a910..6b57847 100644 --- a/src/supabase-worker.ts +++ b/src/supabase-worker.ts @@ -283,6 +283,7 @@ export interface SchedulerStats { inventoryPlansChecked: number; inventoryAlerts: number; skipped: Record; + errors: Record; } type AngleRecordStatus = 'unused' | 'in_progress' | 'drafted' | 'published' | 'rejected' | 'exhausted'; @@ -2350,26 +2351,26 @@ async function recordScheduledAutomationResult(job: AgentJobRow, status: string, } } -function withTimeout(promise: Promise, timeoutMs: number, message: string): Promise { - let timeout: ReturnType | undefined; - const timeoutPromise = new Promise((_, reject) => { - timeout = setTimeout(() => reject(new Error(message)), timeoutMs); - }); - return Promise.race([promise, timeoutPromise]).finally(() => { - if (timeout) clearTimeout(timeout); - }); -} - -function extractSourceBankWithJobTimeout( +async function extractSourceBankWithJobTimeout( post: RedditPost, usageContext?: ai.OpenAIUsageContext, contentStrategyOptions: ContentStrategyPromptOptions = {} ): Promise>> { - return withTimeout( - ai.extractSourceBank(post, { usageContext, ...contentStrategyOptions }), - Math.max(5_000, Math.min(ANGLE_EXTRACTION_TIMEOUT_MS, config.HTTP_TIMEOUT_MS - 1_000)), - 'OpenAI angle extraction timed out before the worker could finalize the job' + const controller = new AbortController(); + const timeout = setTimeout( + () => controller.abort(new Error('OpenAI angle extraction timed out before the worker could finalize the job')), + Math.max(5_000, Math.min(ANGLE_EXTRACTION_TIMEOUT_MS, config.HTTP_TIMEOUT_MS - 1_000)) ); + try { + // Await cancellation and usage bookkeeping before the source claim can be released. + const extraction = await ai.extractSourceBank(post, { + usageContext, ...contentStrategyOptions, signal: controller.signal, + }); + controller.signal.throwIfAborted(); + return extraction; + } finally { + clearTimeout(timeout); + } } async function completeJob(job: AgentJobRow, result: JsonMap): Promise { @@ -3984,6 +3985,23 @@ async function recordAutomationSkip( }); } +async function recordSchedulerFailure( + stats: SchedulerStats, stage: string, error: unknown, userId?: string +): Promise { + stats.errors[stage] = (stats.errors[stage] || 0) + 1; + // Do not put arbitrary provider/DB exception text into cross-tenant diagnostics. + const context = { stage, scope: userId ? 'tenant' : 'stage' }; + logger.error('automation_scheduler_failed', context); + if (userId) { + await writeWorkerLog(userId, 'error', 'automation_scheduler_failed', { + ...context, + message: 'This scheduling check failed. Other checks will continue.', + nextAction: 'Review this tenant’s worker logs and connection state before retrying.', + ...(error instanceof WorkerJobError ? { code: error.code } : {}), + }); + } +} + async function enqueueDueFetchJobs(stats: SchedulerStats, now: Date): Promise { if (!config.SUPABASE_WORKER_GENERATION_ENABLED) return; const allowedUserIds = rolloutAllowedUserIds(); @@ -4001,78 +4019,82 @@ async function enqueueDueFetchJobs(stats: SchedulerStats, now: Date): Promise('agent_jobs', { - user_id: settings.user_id, - kind: 'fetch_sources', - payload: { - source: SCHEDULED_SOURCE, - scheduler: SCHEDULER_NAME, - due_at: dueAt, - }, - }, true); - const job = inserted[0]; - const nextFetchAt = addMinutesIso(now, cadenceMinutes(settings)); - await supabaseUpdate('user_settings', { - last_scheduled_fetch_at: now.toISOString(), - last_scheduled_job_id: job?.id || null, - next_fetch_at: nextFetchAt, - last_automation_result: { - jobId: job?.id || null, + const dueAt = settings.next_fetch_at || now.toISOString(); + const inserted = await supabaseInsert('agent_jobs', { + user_id: settings.user_id, kind: 'fetch_sources', - status: 'pending', - origin: 'scheduled', - message: 'Scheduled fetch was queued by Cloudflare cron.', - nextAction: 'Wait for the worker to process this scheduled fetch.', + payload: { + source: SCHEDULED_SOURCE, + scheduler: SCHEDULER_NAME, + due_at: dueAt, + }, + }, true); + const job = inserted[0]; + const nextFetchAt = addMinutesIso(now, cadenceMinutes(settings)); + await supabaseUpdate('user_settings', { + last_scheduled_fetch_at: now.toISOString(), + last_scheduled_job_id: job?.id || null, + next_fetch_at: nextFetchAt, + last_automation_result: { + jobId: job?.id || null, + kind: 'fetch_sources', + status: 'pending', + origin: 'scheduled', + message: 'Scheduled fetch was queued by Cloudflare cron.', + nextAction: 'Wait for the worker to process this scheduled fetch.', + dueAt, + nextFetchAt, + }, + }, { + filters: [{ column: 'user_id', operator: 'eq', value: settings.user_id }], + }); + stats.fetchJobsEnqueued++; + await writeWorkerLog(settings.user_id, 'info', 'scheduled_fetch_enqueued', { + jobId: job?.id || null, dueAt, nextFetchAt, - }, - }, { - filters: [{ column: 'user_id', operator: 'eq', value: settings.user_id }], - }); - stats.fetchJobsEnqueued++; - await writeWorkerLog(settings.user_id, 'info', 'scheduled_fetch_enqueued', { - jobId: job?.id || null, - dueAt, - nextFetchAt, - scheduler: SCHEDULER_NAME, - }); + scheduler: SCHEDULER_NAME, + }); + } catch (error) { + await recordSchedulerFailure(stats, 'fetch', error, settings.user_id); + } } } @@ -4158,62 +4180,128 @@ async function enqueueDueSlotFillJobs(stats: SchedulerStats, now: Date): Promise }); for (const settings of settingsRows) { - stats.tenantsChecked++; + try { + stats.tenantsChecked++; - if (hasRecentOpenAIAutomationFailure(settings, now)) { - incrementSchedulerSkip(stats, 'fill_openai_generation_pause_active'); - continue; - } + if (hasRecentOpenAIAutomationFailure(settings, now)) { + incrementSchedulerSkip(stats, 'fill_openai_generation_pause_active'); + continue; + } - const entitlement = await loadEntitlement(settings.user_id); - if (!entitlement.canWrite) { - incrementSchedulerSkip(stats, `fill_access_${entitlement.reason}`); - await recordAutomationSkip( - settings.user_id, - 'refresh_queue', - entitlement.reason, - 'Scheduled slot fill is blocked by billing or access state.', - 'Restore billing access or dev/test access before automation can fill open slots.' - ); - continue; - } + const entitlement = await loadEntitlement(settings.user_id); + if (!entitlement.canWrite) { + incrementSchedulerSkip(stats, `fill_access_${entitlement.reason}`); + await recordAutomationSkip( + settings.user_id, + 'refresh_queue', + entitlement.reason, + 'Scheduled slot fill is blocked by billing or access state.', + 'Restore billing access or dev/test access before automation can fill open slots.' + ); + continue; + } - const tenant = await loadTenantContext(settings.user_id); - if (!tenant.activePlatforms.length) { - incrementSchedulerSkip(stats, 'fill_no_enabled_platforms'); - continue; - } + const tenant = await loadTenantContext(settings.user_id); + if (!tenant.activePlatforms.length) { + incrementSchedulerSkip(stats, 'fill_no_enabled_platforms'); + continue; + } - const timeZone = tenantAutomationTimeZone(tenant); - const targetLocalDate = tenantLocalDatePlusDays(now, timeZone, 1); - const activeRows = await loadActiveQueueRows(settings.user_id); - const occupiedSlots = buildPlatformSlotOccupancy(activeRows, timeZone); - - if (plannerActiveForTarget(targetLocalDate)) { - stats.inventoryPlansChecked++; - const plan = buildDailyInventoryPlan( - tenant.activePlatforms, - activeRows as DailyInventoryQueueRow[], - targetLocalDate - ); - if (plan.complete) { - incrementSchedulerSkip(stats, 'daily_inventory_complete'); + const timeZone = tenantAutomationTimeZone(tenant); + const targetLocalDate = tenantLocalDatePlusDays(now, timeZone, 1); + const activeRows = await loadActiveQueueRows(settings.user_id); + const occupiedSlots = buildPlatformSlotOccupancy(activeRows, timeZone); + + if (plannerActiveForTarget(targetLocalDate)) { + stats.inventoryPlansChecked++; + const plan = buildDailyInventoryPlan( + tenant.activePlatforms, + activeRows as DailyInventoryQueueRow[], + targetLocalDate + ); + if (plan.complete) { + incrementSchedulerSkip(stats, 'daily_inventory_complete'); + continue; + } + + if (await hasPendingOrRunningFetch(settings.user_id)) { + incrementSchedulerSkip(stats, 'daily_inventory_work_already_pending_or_running'); + continue; + } + + const missingPlatforms = tenant.activePlatforms.filter( + platform => plan.platforms[platform]?.missingSlotIndexes.length + ); + const hasAngles = await hasDraftableActiveAngle(settings.user_id, missingPlatforms); + const hasSources = hasAngles ? false : await hasProcessableSourceRecord(settings.user_id); + if (!hasAngles && !hasSources) { + incrementSchedulerSkip(stats, 'daily_inventory_insufficient'); + await recordDailyInventoryAlert(settings, plan, stats); + continue; + } + + const inserted = await supabaseInsert('agent_jobs', { + user_id: settings.user_id, + kind: 'refresh_queue', + payload: { + source: SCHEDULED_SOURCE, + scheduler: SCHEDULER_NAME, + mode: 'next_day_inventory', + fill_existing_angles_only: hasAngles, + target_local_date: targetLocalDate, + due_at: now.toISOString(), + }, + }, true); + const job = inserted[0]; + await updateAutomationResult(settings.user_id, { + jobId: job?.id || null, + kind: 'refresh_queue', + status: 'pending', + origin: 'scheduled', + mode: 'next_day_inventory', + targetLocalDate, + message: 'Bounded next-day inventory work was queued.', + nextAction: 'Wait for the worker to prepare approved drafts for the missing slots.', + dueAt: now.toISOString(), + missingSlotCount: plan.missingSlotCount, + }, { + last_scheduled_job_id: job?.id || null, + }); + stats.slotFillJobsEnqueued++; + await writeWorkerLog(settings.user_id, 'info', 'daily_inventory_fill_enqueued', { + jobId: job?.id || null, + targetLocalDate, + missingSlotCount: plan.missingSlotCount, + scheduler: SCHEDULER_NAME, + }); + continue; + } + + const capacity = platformSlotCapacitySnapshot(tenant.activePlatforms, occupiedSlots); + if (!capacity.hasOpenSlots) { + incrementSchedulerSkip(stats, 'refresh_queue_skipped_no_open_slots'); + await recordAutomationSkip( + settings.user_id, + 'refresh_queue', + 'refresh_queue_skipped_no_open_slots', + 'Queue slots are full. Automation skipped drafting until a slot opens.', + 'Wait for scheduled publishing to free slots, or add more future capacity.', + { + activeSlotsByPlatform: capacity.activeSlotsByPlatform, + openSlotsByPlatform: capacity.openSlotsByPlatform, + timezone: timeZone, + } + ); continue; } if (await hasPendingOrRunningFetch(settings.user_id)) { - incrementSchedulerSkip(stats, 'daily_inventory_work_already_pending_or_running'); + incrementSchedulerSkip(stats, 'fill_already_pending_or_running'); continue; } - const missingPlatforms = tenant.activePlatforms.filter( - platform => plan.platforms[platform]?.missingSlotIndexes.length - ); - const hasAngles = await hasDraftableActiveAngle(settings.user_id, missingPlatforms); - const hasSources = hasAngles ? false : await hasProcessableSourceRecord(settings.user_id); - if (!hasAngles && !hasSources) { - incrementSchedulerSkip(stats, 'daily_inventory_insufficient'); - await recordDailyInventoryAlert(settings, plan, stats); + if (!(await hasDraftableActiveAngle(settings.user_id, tenant.activePlatforms))) { + incrementSchedulerSkip(stats, 'fill_no_unused_angles'); continue; } @@ -4223,9 +4311,8 @@ async function enqueueDueSlotFillJobs(stats: SchedulerStats, now: Date): Promise payload: { source: SCHEDULED_SOURCE, scheduler: SCHEDULER_NAME, - mode: 'next_day_inventory', - fill_existing_angles_only: hasAngles, - target_local_date: targetLocalDate, + mode: 'fill_existing_angles', + fill_existing_angles_only: true, due_at: now.toISOString(), }, }, true); @@ -4235,83 +4322,22 @@ async function enqueueDueSlotFillJobs(stats: SchedulerStats, now: Date): Promise kind: 'refresh_queue', status: 'pending', origin: 'scheduled', - mode: 'next_day_inventory', - targetLocalDate, - message: 'Bounded next-day inventory work was queued.', - nextAction: 'Wait for the worker to prepare approved drafts for the missing slots.', + mode: 'fill_existing_angles', + message: 'Scheduled slot fill was queued from existing unused angles.', + nextAction: 'Wait for the worker to draft an unused angle into the next open slot.', dueAt: now.toISOString(), - missingSlotCount: plan.missingSlotCount, }, { last_scheduled_job_id: job?.id || null, }); stats.slotFillJobsEnqueued++; - await writeWorkerLog(settings.user_id, 'info', 'daily_inventory_fill_enqueued', { + await writeWorkerLog(settings.user_id, 'info', 'scheduled_slot_fill_enqueued', { jobId: job?.id || null, - targetLocalDate, - missingSlotCount: plan.missingSlotCount, + dueAt: now.toISOString(), scheduler: SCHEDULER_NAME, }); - continue; - } - - const capacity = platformSlotCapacitySnapshot(tenant.activePlatforms, occupiedSlots); - if (!capacity.hasOpenSlots) { - incrementSchedulerSkip(stats, 'refresh_queue_skipped_no_open_slots'); - await recordAutomationSkip( - settings.user_id, - 'refresh_queue', - 'refresh_queue_skipped_no_open_slots', - 'Queue slots are full. Automation skipped drafting until a slot opens.', - 'Wait for scheduled publishing to free slots, or add more future capacity.', - { - activeSlotsByPlatform: capacity.activeSlotsByPlatform, - openSlotsByPlatform: capacity.openSlotsByPlatform, - timezone: timeZone, - } - ); - continue; - } - - if (await hasPendingOrRunningFetch(settings.user_id)) { - incrementSchedulerSkip(stats, 'fill_already_pending_or_running'); - continue; - } - - if (!(await hasDraftableActiveAngle(settings.user_id, tenant.activePlatforms))) { - incrementSchedulerSkip(stats, 'fill_no_unused_angles'); - continue; + } catch (error) { + await recordSchedulerFailure(stats, 'slot_fill', error, settings.user_id); } - - const inserted = await supabaseInsert('agent_jobs', { - user_id: settings.user_id, - kind: 'refresh_queue', - payload: { - source: SCHEDULED_SOURCE, - scheduler: SCHEDULER_NAME, - mode: 'fill_existing_angles', - fill_existing_angles_only: true, - due_at: now.toISOString(), - }, - }, true); - const job = inserted[0]; - await updateAutomationResult(settings.user_id, { - jobId: job?.id || null, - kind: 'refresh_queue', - status: 'pending', - origin: 'scheduled', - mode: 'fill_existing_angles', - message: 'Scheduled slot fill was queued from existing unused angles.', - nextAction: 'Wait for the worker to draft an unused angle into the next open slot.', - dueAt: now.toISOString(), - }, { - last_scheduled_job_id: job?.id || null, - }); - stats.slotFillJobsEnqueued++; - await writeWorkerLog(settings.user_id, 'info', 'scheduled_slot_fill_enqueued', { - jobId: job?.id || null, - dueAt: now.toISOString(), - scheduler: SCHEDULER_NAME, - }); } } @@ -4366,102 +4392,106 @@ async function enqueueDuePublishJobs(stats: SchedulerStats, now: Date): Promise< }); for (const row of dueRows) { - if (row.legacy_revision_hold_id) { - incrementSchedulerSkip(stats, LEGACY_QUEUE_REVISION_HELD_CODE); - await recordAutomationSkip( - row.user_id, - 'publish_now', - LEGACY_QUEUE_REVISION_HELD_CODE, - LEGACY_QUEUE_REVISION_HELD_MESSAGE, - LEGACY_QUEUE_REVISION_HELD_NEXT_ACTION, - { queueItemId: row.id, platform: row.platform } - ); - continue; - } + try { + if (row.legacy_revision_hold_id) { + incrementSchedulerSkip(stats, LEGACY_QUEUE_REVISION_HELD_CODE); + await recordAutomationSkip( + row.user_id, + 'publish_now', + LEGACY_QUEUE_REVISION_HELD_CODE, + LEGACY_QUEUE_REVISION_HELD_MESSAGE, + LEGACY_QUEUE_REVISION_HELD_NEXT_ACTION, + { queueItemId: row.id, platform: row.platform } + ); + continue; + } - const settings = settingsByUser.get(row.user_id); - if (!settings || !settings.automation_publish_enabled) { - incrementSchedulerSkip(stats, 'publish_automation_disabled'); - continue; - } + const settings = settingsByUser.get(row.user_id); + if (!settings || !settings.automation_publish_enabled) { + incrementSchedulerSkip(stats, 'publish_automation_disabled'); + continue; + } - const priorFailedRecovery = await loadPriorFailedRecovery(row); - if (priorFailedRecovery) { - incrementSchedulerSkip(stats, 'publish_recovery_paused_after_failure'); - await supabaseUpdate('queue_items', { - status: 'skipped', - error_message: 'Recovery paused because an earlier recovery on this platform failed. Review the earlier attempt before rescheduling.', - }, { - filters: [ - { column: 'id', operator: 'eq', value: row.id }, - { column: 'user_id', operator: 'eq', value: row.user_id }, - { column: 'status', operator: 'in', value: ['pending', 'ready'] }, - ], - }); - await recordAutomationSkip( - row.user_id, - 'publish_now', - 'publish_recovery_paused_after_failure', - 'A scheduled recovery was paused because an earlier recovery on this platform failed.', - 'Review the earlier recovery attempt before rescheduling this row.', - { - queueItemId: row.id, - platform: row.platform, - priorQueueItemId: priorFailedRecovery.id, - } - ); - continue; - } + const priorFailedRecovery = await loadPriorFailedRecovery(row); + if (priorFailedRecovery) { + incrementSchedulerSkip(stats, 'publish_recovery_paused_after_failure'); + await supabaseUpdate('queue_items', { + status: 'skipped', + error_message: 'Recovery paused because an earlier recovery on this platform failed. Review the earlier attempt before rescheduling.', + }, { + filters: [ + { column: 'id', operator: 'eq', value: row.id }, + { column: 'user_id', operator: 'eq', value: row.user_id }, + { column: 'status', operator: 'in', value: ['pending', 'ready'] }, + ], + }); + await recordAutomationSkip( + row.user_id, + 'publish_now', + 'publish_recovery_paused_after_failure', + 'A scheduled recovery was paused because an earlier recovery on this platform failed.', + 'Review the earlier recovery attempt before rescheduling this row.', + { + queueItemId: row.id, + platform: row.platform, + priorQueueItemId: priorFailedRecovery.id, + } + ); + continue; + } - const entitlement = await loadEntitlement(row.user_id); - if (!entitlement.canWrite) { - incrementSchedulerSkip(stats, `publish_access_${entitlement.reason}`); - await recordAutomationSkip( - row.user_id, - 'publish_now', - entitlement.reason, - 'Scheduled publish is blocked by billing or access state.', - 'Restore billing access or dev/test access before scheduled publishing can continue.' - ); - continue; - } + const entitlement = await loadEntitlement(row.user_id); + if (!entitlement.canWrite) { + incrementSchedulerSkip(stats, `publish_access_${entitlement.reason}`); + await recordAutomationSkip( + row.user_id, + 'publish_now', + entitlement.reason, + 'Scheduled publish is blocked by billing or access state.', + 'Restore billing access or dev/test access before scheduled publishing can continue.' + ); + continue; + } - if (await hasPendingOrRunningPublish(row.user_id, row.id)) { - incrementSchedulerSkip(stats, 'publish_already_pending_or_running'); - continue; - } + if (await hasPendingOrRunningPublish(row.user_id, row.id)) { + incrementSchedulerSkip(stats, 'publish_already_pending_or_running'); + continue; + } - const inserted = await supabaseInsert('agent_jobs', { - user_id: row.user_id, - kind: 'publish_now', - payload: { - source: SCHEDULED_SOURCE, + const inserted = await supabaseInsert('agent_jobs', { + user_id: row.user_id, + kind: 'publish_now', + payload: { + source: SCHEDULED_SOURCE, + scheduler: SCHEDULER_NAME, + queue_item_id: row.id, + due_at: row.scheduled_for, + }, + }, true); + const job = inserted[0]; + await updateAutomationResult(row.user_id, { + jobId: job?.id || null, + kind: 'publish_now', + status: 'pending', + origin: 'scheduled', + message: `Scheduled ${row.platform} publish was queued.`, + nextAction: 'Wait for the worker to publish this due queue item.', + queueItemId: row.id, + dueAt: row.scheduled_for, + }, { + last_scheduled_job_id: job?.id || null, + }); + stats.publishJobsEnqueued++; + await writeWorkerLog(row.user_id, 'info', 'scheduled_publish_enqueued', { + jobId: job?.id || null, + queueItemId: row.id, + platform: row.platform, + dueAt: row.scheduled_for, scheduler: SCHEDULER_NAME, - queue_item_id: row.id, - due_at: row.scheduled_for, - }, - }, true); - const job = inserted[0]; - await updateAutomationResult(row.user_id, { - jobId: job?.id || null, - kind: 'publish_now', - status: 'pending', - origin: 'scheduled', - message: `Scheduled ${row.platform} publish was queued.`, - nextAction: 'Wait for the worker to publish this due queue item.', - queueItemId: row.id, - dueAt: row.scheduled_for, - }, { - last_scheduled_job_id: job?.id || null, - }); - stats.publishJobsEnqueued++; - await writeWorkerLog(row.user_id, 'info', 'scheduled_publish_enqueued', { - jobId: job?.id || null, - queueItemId: row.id, - platform: row.platform, - dueAt: row.scheduled_for, - scheduler: SCHEDULER_NAME, - }); + }); + } catch (error) { + await recordSchedulerFailure(stats, 'publish', error, row.user_id); + } } } @@ -4843,54 +4873,58 @@ async function cleanupStaleRunningJobs(stats: SchedulerStats, now: Date): Promis }); for (const job of jobs) { - const staleCutoff = addMinutesIso(now, -staleMinutesForJob(job)); - const startedAt = Date.parse(job.started_at || ''); - const staleCutoffMs = Date.parse(staleCutoff); - if ( - Number.isFinite(startedAt) - && Number.isFinite(staleCutoffMs) - && startedAt > staleCutoffMs - ) { - incrementSchedulerSkip(stats, 'stale_job_within_kind_runtime'); - continue; - } + try { + const staleCutoff = addMinutesIso(now, -staleMinutesForJob(job)); + const startedAt = Date.parse(job.started_at || ''); + const staleCutoffMs = Date.parse(staleCutoff); + if ( + Number.isFinite(startedAt) + && Number.isFinite(staleCutoffMs) + && startedAt > staleCutoffMs + ) { + incrementSchedulerSkip(stats, 'stale_job_within_kind_runtime'); + continue; + } - const logs = await staleJobLogs(job); - if (job.kind === 'refresh_queue' && hasRecentJobActivity(logs, cutoff)) { - incrementSchedulerSkip(stats, 'stale_refresh_recent_activity'); - continue; - } + const logs = await staleJobLogs(job); + if (job.kind === 'refresh_queue' && hasRecentJobActivity(logs, cutoff)) { + incrementSchedulerSkip(stats, 'stale_refresh_recent_activity'); + continue; + } - const releasedAngleLocks = await releaseStaleRefreshAngleLocks(job, logs); - const details = staleFailureFromLogs(logs); - const result = job.kind === 'publish_now' - ? await stalePublishJobResult(job, logs) - : staleJobResult(job, logs, details); - const summary = resultSummary(result); - const error = String(summary.failureCode || result.error || 'worker_job_timed_out'); - const status = terminalStatusForResult(result); - await supabaseUpdate('agent_jobs', { - status, - completed_at: now.toISOString(), - error, - result, - }, { - filters: [ - { column: 'id', operator: 'eq', value: job.id }, - { column: 'status', operator: 'eq', value: 'running' }, - ], - }); - await recordScheduledAutomationResult(job, status, result); - stats.staleJobsFailed++; - await writeWorkerLog(job.user_id, 'warn', 'stale_running_job_failed', { - jobId: job.id, - kind: job.kind, - reason: error, - status, - failedStage: typeof summary.failedStage === 'string' ? summary.failedStage : null, - releasedAngleLocks, - startedAt: job.started_at || null, - }); + const releasedAngleLocks = await releaseStaleRefreshAngleLocks(job, logs); + const details = staleFailureFromLogs(logs); + const result = job.kind === 'publish_now' + ? await stalePublishJobResult(job, logs) + : staleJobResult(job, logs, details); + const summary = resultSummary(result); + const error = String(summary.failureCode || result.error || 'worker_job_timed_out'); + const status = terminalStatusForResult(result); + await supabaseUpdate('agent_jobs', { + status, + completed_at: now.toISOString(), + error, + result, + }, { + filters: [ + { column: 'id', operator: 'eq', value: job.id }, + { column: 'status', operator: 'eq', value: 'running' }, + ], + }); + await recordScheduledAutomationResult(job, status, result); + stats.staleJobsFailed++; + await writeWorkerLog(job.user_id, 'warn', 'stale_running_job_failed', { + jobId: job.id, + kind: job.kind, + reason: error, + status, + failedStage: typeof summary.failedStage === 'string' ? summary.failedStage : null, + releasedAngleLocks, + startedAt: job.started_at || null, + }); + } catch (error) { + await recordSchedulerFailure(stats, 'job_recovery', error, job.user_id); + } } } @@ -4904,13 +4938,25 @@ export async function runSupabaseAutomationScheduler(): Promise inventoryPlansChecked: 0, inventoryAlerts: 0, skipped: {}, + errors: {}, }; const now = new Date(); - await recoverStalePublications(now.getTime(), rolloutAllowedUserIds()); - await cleanupStaleRunningJobs(stats, now); - await enqueueDueFetchJobs(stats, now); - await enqueueDueSlotFillJobs(stats, now); - await enqueueDuePublishJobs(stats, now); + const stages: Array<[string, () => Promise]> = [ + ['publication_recovery', () => recoverStalePublications(now.getTime(), rolloutAllowedUserIds())], + ['job_recovery', () => cleanupStaleRunningJobs(stats, now)], + ['fetch', () => enqueueDueFetchJobs(stats, now)], + ['slot_fill', () => enqueueDueSlotFillJobs(stats, now)], + ['publish', () => enqueueDuePublishJobs(stats, now)], + ]; + for (const [stage, run] of stages) { + try { + await run(); + } catch (error) { + // Each consumer still applies its own entitlement, claims and rollout gates. + // An unavailable input in one stage must not suppress independent stages. + await recordSchedulerFailure(stats, stage, error); + } + } return stats; } @@ -5016,6 +5062,7 @@ export function startSupabaseWorkerLoop(log = logger): { stop: () => void } | un } export const __test__ = { + extractSourceBankWithJobTimeout, assertQueueRevisionNotHeld, publishQueueRow, stalePublishJobResult, diff --git a/test/http-cancellation.test.ts b/test/http-cancellation.test.ts new file mode 100644 index 0000000..167781e --- /dev/null +++ b/test/http-cancellation.test.ts @@ -0,0 +1,118 @@ +import assert from 'node:assert/strict'; +import { test } from 'node:test'; +import config from '../config'; +import { requestJson } from '../src/http-client'; +import { extractSourceBank, setOpenAIUsageRecorder, type OpenAIUsageEvent } from '../src/ai'; +import { __test__ } from '../src/supabase-worker'; + +const post = { + id: 'post-1', title: 'Reliable queues', selftext: 'A real source body.', + url: 'https://reddit.example/post-1', score: 1, comments: 0, + subreddit: 'example', author: 'example', created: 1, +}; + +test('cancellation reaches transport, response reads and retry delays', async t => { + const originalFetch = globalThis.fetch; + try { + await t.test('already cancelled operations make no request', async () => { + let calls = 0; + globalThis.fetch = async () => { calls++; return new Response('{}'); }; + const reason = new Error('cancelled'); + await assert.rejects(requestJson('https://example.test', { + signal: AbortSignal.abort(reason), retryCount: 3, + }), error => error === reason); + assert.equal(calls, 0); + }); + + await t.test('caller cancellation aborts a pending request without retrying', async () => { + let calls = 0; + let transportAborted = false; + const controller = new AbortController(); + const reason = new Error('cancelled by job'); + globalThis.fetch = async (_url, options) => { + calls++; + return new Promise((_resolve, reject) => { + options!.signal!.addEventListener('abort', () => { + transportAborted = true; + reject(options!.signal!.reason); + }, { once: true }); + }); + }; + const result = requestJson('https://example.test', { signal: controller.signal, retryCount: 3 }); + controller.abort(reason); + await assert.rejects(result, error => error === reason); + assert.equal(transportAborted, true); + assert.equal(calls, 1); + }); + + await t.test('an abort during body reading cannot be returned as success', async () => { + const controller = new AbortController(); + globalThis.fetch = async () => ({ + ok: true, status: 200, headers: new Headers(), + text: async () => { controller.abort(new Error('body cancelled')); return '{}'; }, + }) as Response; + await assert.rejects(requestJson('https://example.test', { signal: controller.signal }), /body cancelled/); + }); + + await t.test('an abort interrupts backoff without another request', async () => { + let calls = 0; + const controller = new AbortController(); + globalThis.fetch = async () => { + calls++; + return new Response('{}', { status: 429 }); + }; + const result = requestJson('https://example.test', { + signal: controller.signal, retryCount: 3, retryDelayMs: 10_000, + }); + const timer = setTimeout(() => controller.abort(new Error('backoff cancelled')), 10); + try { await assert.rejects(result, /backoff cancelled/); } + finally { clearTimeout(timer); } + assert.equal(calls, 1); + }); + + await t.test('ordinary request timeout retains its typed error', async () => { + globalThis.fetch = async (_url, options) => new Promise((_resolve, reject) => { + options!.signal!.addEventListener('abort', () => reject(new Error('runtime-specific abort')), { once: true }); + }); + await assert.rejects(requestJson('https://example.test', { timeoutMs: 5 }), + (error: any) => error.code === 'UPSTREAM_TIMEOUT'); + }); + } finally { globalThis.fetch = originalFetch; } +}); + +test('extraction deadline cancels paid work before returning the source claim', async () => { + const originalFetch = globalThis.fetch; + const previousTimeout = config.HTTP_TIMEOUT_MS; + const events: OpenAIUsageEvent[] = []; + let transportSettled = false; + let calls = 0; + setOpenAIUsageRecorder(event => { events.push(event); }); + config.HTTP_TIMEOUT_MS = 6_000; // worker deadline is 5 seconds, before the HTTP timeout + globalThis.fetch = async (_url, options) => { + calls++; + return new Promise((_resolve, reject) => { + options!.signal!.addEventListener('abort', () => { + queueMicrotask(() => { + transportSettled = true; + reject(options!.signal!.reason); + }); + }, { once: true }); + }); + }; + try { + await assert.rejects(__test__.extractSourceBankWithJobTimeout(post), /OpenAI angle extraction timed out/); + assert.equal(transportSettled, true); + assert.equal(calls, 1); + assert.deepEqual(events.map(event => event.call_status), ['started', 'failed']); + assert.equal(events[1].stage, 'angle_extraction'); + + const eventCount = events.length; + await assert.rejects(extractSourceBank(post, { signal: AbortSignal.abort(new Error('job finished')) }), /job finished/); + assert.equal(calls, 1); + assert.equal(events.length, eventCount); + } finally { + globalThis.fetch = originalFetch; + config.HTTP_TIMEOUT_MS = previousTimeout; + setOpenAIUsageRecorder(undefined); + } +}); diff --git a/test/publication-executor.test.ts b/test/publication-executor.test.ts index 89dcb9a..ea1ff78 100644 --- a/test/publication-executor.test.ts +++ b/test/publication-executor.test.ts @@ -228,7 +228,7 @@ async function main() { assert.ok(!publish.includes("supabaseInsert")); assert.ok(!publish.includes("status: 'failed'")); assert.ok(!worker.includes('findPublishHistoryForQueueItem')); - assert.ok(worker.includes('await recoverStalePublications(now.getTime(), rolloutAllowedUserIds())')); + assert.ok(worker.includes('recoverStalePublications(now.getTime(), rolloutAllowedUserIds())')); }); } main().catch(error => { console.error(error); process.exitCode = 1; }); diff --git a/test/publish-summary.test.ts b/test/publish-summary.test.ts index d341ff2..46b1608 100644 --- a/test/publish-summary.test.ts +++ b/test/publish-summary.test.ts @@ -61,7 +61,7 @@ test('production stale publication recovery has no log/status reconciliation fal assert.ok(worker.includes( 'return reconcilePublication({ userId: job.user_id, queueItemId: row.id, platform: row.platform });' )); - assert.ok(worker.includes('await recoverStalePublications(now.getTime(), rolloutAllowedUserIds());')); + assert.ok(worker.includes('recoverStalePublications(now.getTime(), rolloutAllowedUserIds())')); }); test('unknown publication guidance never authorises a blind retry', () => { diff --git a/test/scheduler-isolation.test.ts b/test/scheduler-isolation.test.ts new file mode 100644 index 0000000..cad74dd --- /dev/null +++ b/test/scheduler-isolation.test.ts @@ -0,0 +1,86 @@ +import assert from 'node:assert/strict'; +import { test } from 'node:test'; +import config from '../config'; +import { runSupabaseAutomationScheduler } from '../src/supabase-worker'; + +test('the real scheduler isolates tenant, stage and logging failures', async t => { + const originalFetch = globalThis.fetch; + const originalConfig = { + SUPABASE_URL: config.SUPABASE_URL, + SUPABASE_SERVICE_ROLE_KEY: config.SUPABASE_SERVICE_ROLE_KEY, + CREDENTIAL_ENCRYPTION_KEY: config.CREDENTIAL_ENCRYPTION_KEY, + SUPABASE_WORKER_CANARY_REQUIRED: config.SUPABASE_WORKER_CANARY_REQUIRED, + SUPABASE_WORKER_CANARY_USER_IDS: config.SUPABASE_WORKER_CANARY_USER_IDS, + SUPABASE_WORKER_GENERATION_ENABLED: config.SUPABASE_WORKER_GENERATION_ENABLED, + SUPABASE_PROVIDER_DISPATCH_ENABLED: config.SUPABASE_PROVIDER_DISPATCH_ENABLED, + }; + Object.assign(config, { + SUPABASE_URL: 'https://example.supabase.co', SUPABASE_SERVICE_ROLE_KEY: 'test-key', + CREDENTIAL_ENCRYPTION_KEY: 'test-encryption', SUPABASE_WORKER_CANARY_REQUIRED: false, + SUPABASE_WORKER_CANARY_USER_IDS: '', SUPABASE_WORKER_GENERATION_ENABLED: true, + SUPABASE_PROVIDER_DISPATCH_ENABLED: true, + }); + try { + for (const failure of ['tenant', 'fetch_query', 'recovery_query', 'logging'] as const) { + await t.test(failure, async () => { + const enqueued: Array> = []; + const unexpected: string[] = []; + let slotFillVisited = false; + const settings = ['broken', 'healthy'].map(user_id => ({ + user_id, automation_enabled: true, automation_publish_enabled: true, + automation_fetch_enabled: true, next_fetch_at: null, + })); + const json = (data: unknown, status = 200) => new Response(JSON.stringify(data), { status }); + globalThis.fetch = async (input, init) => { + const url = new URL(String(input)); + const table = url.pathname.split('/').pop(); + const method = init?.method || 'GET'; + const user = url.searchParams.get('user_id'); + if (url.pathname.includes('/rpc/')) return json({ message: 'schema unavailable' }, 503); + if (table === 'worker_logs' && method === 'POST') return failure === 'logging' + ? json({ message: 'log unavailable' }, 503) : json([]); + if (table === 'user_settings' && method === 'PATCH') return json([]); + if (table === 'user_settings' && method === 'GET') { + if (url.searchParams.has('automation_fetch_enabled')) return failure === 'fetch_query' + ? json({ message: 'settings unavailable' }, 503) : json(settings); + if (url.searchParams.has('automation_publish_enabled')) { + slotFillVisited = true; + return json([]); + } + return json(settings); + } + if (table === 'profiles') return user === 'eq.broken' + ? json({ message: 'tenant data unavailable' }, 503) : json([{ subscription_status: 'active' }]); + if (table === 'internal_access_overrides') return json([]); + if (table === 'user_sources') return json([{ id: 'source' }]); + if (table === 'agent_jobs' && method === 'GET') return failure === 'recovery_query' && url.searchParams.has('started_at') + ? json({ message: 'recovery unavailable' }, 503) : json([]); + if (table === 'agent_jobs' && method === 'POST') { + const job = JSON.parse(String(init?.body)); + enqueued.push(job); + return json([{ ...job, id: `job-${enqueued.length}` }]); + } + if (table === 'queue_items' && method === 'GET') return json(['broken', 'healthy', 'healthy'].map((user_id, i) => ({ + id: `queue-${i}`, user_id, platform: 'x', status: 'ready', + scheduled_for: '2026-01-01T00:00:00Z', + }))); + unexpected.push(`${method} ${url.pathname}`); + return json({ message: 'unexpected request' }, 500); + }; + const stats = await runSupabaseAutomationScheduler(); + assert.equal(slotFillVisited, true); + assert.equal(stats.publishJobsEnqueued, 2); + assert.equal(stats.fetchJobsEnqueued, failure === 'fetch_query' ? 0 : 1); + assert.equal(stats.errors.fetch, 1); + assert.equal(stats.errors.publish, 1); + assert.equal(stats.errors.job_recovery || 0, failure === 'recovery_query' ? 1 : 0); + assert.equal(enqueued.every(job => job.user_id === 'healthy'), true); + assert.deepEqual(enqueued.filter(job => job.kind === 'publish_now').map(job => job.payload.queue_item_id), ['queue-1', 'queue-2']); + assert.deepEqual(unexpected, []); + }); + } + } finally { + globalThis.fetch = originalFetch; + Object.assign(config, originalConfig); + } +}); From 72fb0040fb2b256f522d2ba79bd40e7fc0de13a3 Mon Sep 17 00:00:00 2001 From: Ayobami Haastrup <47716486+AyobamiH@users.noreply.github.com> Date: Sat, 12 Sep 2026 22:56:18 -0700 Subject: [PATCH 2/4] Expose and stress concurrent publication database failures --- test/publication-database.integration.ts | 27 +++++++++++++++++++----- 1 file changed, 22 insertions(+), 5 deletions(-) diff --git a/test/publication-database.integration.ts b/test/publication-database.integration.ts index 7f6ce04..b1b9755 100644 --- a/test/publication-database.integration.ts +++ b/test/publication-database.integration.ts @@ -41,11 +41,16 @@ const texts = new Map(); const modes = new Map(); let lostRpc: { path: string; remaining: number } | undefined; let totalWrites = 0; +const databaseErrors: Array<{ rpc: string; code: string }> = []; const realFetch = globalThis.fetch; globalThis.fetch = (async (input, init) => { const url = new URL(String(input)); if (url.origin === api.origin) { const response = await realFetch(input, init); + if (!response.ok && url.pathname.includes('/rpc/')) { + const failure = await response.clone().json().catch(() => ({})) as { code?: unknown }; + databaseErrors.push({ rpc: url.pathname.split('/').pop() || 'unknown', code: String(failure.code || 'unknown') }); + } if (lostRpc && url.pathname.endsWith(lostRpc.path) && lostRpc.remaining > 0 && response.ok) { lostRpc.remaining--; await response.text(); throw new TypeError('injected response loss after database commit'); } @@ -123,11 +128,23 @@ async function main() { assert.throws(() => sql(`UPDATE public.queue_items SET draft_text = 'late edit' WHERE id = ${literal(f.row.id)}::uuid;`)); }); await test('concurrent real Postgres claimers have exactly one provider writer', async () => { - const f = seed(); - await Promise.all(Array.from({ length: 8 }, () => run(f))); - assert.equal(posts.get(f.user), 1); - assert.equal(Number(sql(`SELECT count(*) FROM public.publication_attempts WHERE user_id = ${literal(f.user)}::uuid;`)), 1); - assert.equal((await state(f)).attempt?.state, 'accepted'); + for (let round = 0; round < 20; round++) { + const f = seed(), errorStart = databaseErrors.length; + const results = await Promise.all(Array.from({ length: 8 }, () => run(f))); + const current = await state(f); + const observed = { + round, + writes: posts.get(f.user) || 0, + outcomes: results.map(r => ({ outcome: r.outcome, failureCode: r.failureCode })), + intentState: current.intent?.state, + attemptState: current.attempt?.state, + databaseErrors: databaseErrors.slice(errorStart), + }; + assert.equal(posts.get(f.user) || 0, 1, JSON.stringify(observed)); + assert.equal(Number(sql(`SELECT count(*) FROM public.publication_attempts WHERE user_id = ${literal(f.user)}::uuid;`)), 1); + assert.equal(current.attempt?.state, 'accepted', JSON.stringify(observed)); + assert.equal(databaseErrors.slice(errorStart).some(error => error.code === '40P01'), false, JSON.stringify(observed)); + } }); await test('two tenants interleave with their own encrypted credentials and source snapshots', async () => { const a = seed(), b = seed(); From e7b4cb6a8d005eda5602534d691b1ef93599703f Mon Sep 17 00:00:00 2001 From: Ayobami Haastrup <47716486+AyobamiH@users.noreply.github.com> Date: Sat, 12 Sep 2026 23:00:37 -0700 Subject: [PATCH 3/4] Reproduce publication lock inversion with controlled Postgres sessions --- test/publication-database.integration.ts | 12 ++ test/publication-lock-order.integration.ts | 144 +++++++++++++++++++++ 2 files changed, 156 insertions(+) create mode 100644 test/publication-lock-order.integration.ts diff --git a/test/publication-database.integration.ts b/test/publication-database.integration.ts index b1b9755..1c3c4c9 100644 --- a/test/publication-database.integration.ts +++ b/test/publication-database.integration.ts @@ -1,4 +1,5 @@ import assert from 'node:assert/strict'; +import { exercisePublicationLockOrder } from './publication-lock-order.integration'; import { execFileSync } from 'node:child_process'; import { randomUUID } from 'node:crypto'; import config from '../config'; @@ -113,6 +114,17 @@ async function test(name: string, fn: () => Promise) { } async function main() { + for (const operation of ['begin', 'release'] as const) { + await test(`controlled ${operation}/claim contention has no database deadlock`, async () => { + const f = seed(); + const intent = await claimPublicationIntent(f.user, f.row.id, randomUUID(), 120); + await exercisePublicationLockOrder({ + databaseName: names[0], sql, userId: f.user, queueId: f.row.id, + intentId: intent.id, claimToken: intent.claim_token!, claimVersion: intent.claim_version, operation, + }); + assert.equal(posts.get(f.user) || 0, 0); + }); + } await test('real scheduled Worker records exact immutable acceptance and handles duplicate delivery', async () => { const f = seed(); sql(`UPDATE public.queue_items SET draft_text = 'approved revision before claim' WHERE id = ${literal(f.row.id)}::uuid;`); diff --git a/test/publication-lock-order.integration.ts b/test/publication-lock-order.integration.ts new file mode 100644 index 0000000..7586ed8 --- /dev/null +++ b/test/publication-lock-order.integration.ts @@ -0,0 +1,144 @@ +import assert from 'node:assert/strict'; +import { spawn } from 'node:child_process'; +import { randomUUID } from 'node:crypto'; + +interface LockFixture { + databaseName: string; + sql: (statement: string) => string; + userId: string; + queueId: string; + intentId: string; + claimToken: string; + claimVersion: number; + operation: 'begin' | 'release'; +} +const literal = (value: string) => "'" + value.replace(/'/g, "''") + "'"; + +// Persistent psql sessions let the test hold real transaction locks. No database +// URL or provider credential is accepted: the caller has already proved this is +// the isolated local Supabase container. +function session(databaseName: string) { + assert.match(databaseName, /^supabase_db_[a-zA-Z0-9_-]+$/); + const child = spawn('docker', ['exec', '-i', databaseName, 'psql', '-X', '-qAt', + '-U', 'postgres', '-d', 'postgres', '-v', 'ON_ERROR_STOP=1', '-v', 'VERBOSITY=sqlstate']); + let stdout = '', stderr = '', exited = false; + let pending: { marker: string; resolve: (value: string) => void; reject: (error: Error) => void } | undefined; + const closed = new Promise(resolve => { + child.on('close', () => { + exited = true; + pending?.reject(new Error(stderr.trim() || 'Local SQL session closed')); + pending = undefined; + resolve(); + }); + }); + child.on('error', error => pending?.reject(error)); + child.stdin.on('error', error => pending?.reject(error)); + child.stderr.on('data', chunk => { stderr = (stderr + String(chunk)).slice(-2000); }); + child.stdout.on('data', chunk => { + stdout += String(chunk); + if (pending) { + const end = stdout.indexOf(pending.marker + '\n'); + if (end >= 0) { + const value = stdout.slice(0, end).trim(); + stdout = stdout.slice(end + pending.marker.length + 1); + const resolve = pending.resolve; + pending = undefined; + resolve(value); + } + } + }); + return { + query(statement: string): Promise { + assert.equal(exited, false, 'SQL session remains open'); + assert.equal(pending, undefined, 'one statement per SQL session at a time'); + const marker = 'barrier_' + randomUUID().replace(/-/g, ''); + return new Promise((resolve, reject) => { + pending = { marker, resolve, reject }; + child.stdin.write(statement + ';\nSELECT ' + literal(marker) + ';\n'); + }); + }, + async close() { + if (!exited) child.stdin.end('\n\\q\n'); + await closed; + }, + }; +} + +export async function exercisePublicationLockOrder(f: LockFixture): Promise { + const gate = session(f.databaseName), owner = session(f.databaseName), contender = session(f.databaseName); + const suffix = randomUUID().replace(/-/g, ''); + const gateKey = parseInt(suffix.slice(0, 7), 16); + const ownerName = 'publication_owner_' + suffix; + const contenderName = 'publication_contender_' + suffix; + const contenderToken = randomUUID(); + const table = f.operation === 'begin' ? 'publication_attempts' : 'publication_intents'; + async function waitForLock(name: string, advisory = false) { + const deadline = Date.now() + 7000; + while (Date.now() < deadline) { + const waiting = Number(f.sql("SELECT count(*) FROM pg_stat_activity WHERE application_name = " + + literal(name) + " AND wait_event_type = 'Lock'" + + (advisory ? " AND wait_event = 'advisory'" : ''))); + if (waiting === 1) return; + await new Promise(resolve => setTimeout(resolve, 20)); + } + throw new Error('Expected controlled database lock was not reached'); + } + try { + // This fixture-only trigger pauses the owner after its initial row lock, + // before the foreign-key check (begin) or queue projection (release). + f.sql("CREATE FUNCTION public.fixture_publication_lock_barrier() RETURNS trigger LANGUAGE plpgsql SET search_path = '' AS $$ BEGIN " + + "IF NEW.queue_item_id = " + literal(f.queueId) + "::uuid THEN PERFORM pg_catalog.pg_advisory_xact_lock(21475, " + gateKey + "); END IF; RETURN NEW; END; $$; " + + "CREATE TRIGGER fixture_publication_lock_barrier BEFORE " + + (f.operation === 'begin' ? 'INSERT' : 'UPDATE') + + " ON public." + table + " FOR EACH ROW EXECUTE FUNCTION public.fixture_publication_lock_barrier();"); + await gate.query('SELECT pg_advisory_lock(21475, ' + gateKey + ')'); + for (const [connection, name] of [[owner, ownerName], [contender, contenderName]] as const) { + await connection.query("SET application_name = " + literal(name) + + "; SET statement_timeout = '15s'; SET deadlock_timeout = '100ms'; SET ROLE service_role"); + } + const ownerSql = f.operation === 'begin' + ? "SELECT id FROM public.begin_publication_dispatch(" + [ + literal(f.userId), literal(f.intentId), literal(f.claimToken), String(f.claimVersion), + literal(randomUUID()), literal('fixture-account'), 'NULL', 'false', + ].join(',') + ')' + : "SELECT id FROM public.release_publication_claim(" + [ + literal(f.userId), literal(f.intentId), literal(f.claimToken), String(f.claimVersion), + literal('fixture_predispatch_release'), + ].join(',') + ')'; + const settle = (promise: Promise) => promise.then( + value => ({ ok: true, value, error: '' }), + error => ({ ok: false, value: '', error: String(error) }), + ); + const ownerResult = settle(owner.query(ownerSql)); + await waitForLock(ownerName, true); + const contenderResult = settle(contender.query("SELECT id FROM public.claim_publication_intent(" + + [literal(f.userId), literal(f.queueId), literal(contenderToken), '120'].join(',') + ')')); + await waitForLock(contenderName); + await gate.query('SELECT pg_advisory_unlock(21475, ' + gateKey + ')'); + const results = await Promise.all([ownerResult, contenderResult]); + assert.equal(results.some(result => result.error.includes('40P01')), false, + 'Controlled ' + f.operation + '/claim deadlock: ' + JSON.stringify(results)); + assert.equal(results[0].ok, true, JSON.stringify(results)); + if (f.operation === 'begin') { + assert.equal(results[1].ok, false, 'Duplicate claim must not steal dispatched intent'); + assert.match(results[1].error, /P0001/); + assert.equal(f.sql("SELECT state FROM public.publication_attempts WHERE queue_item_id = " + + literal(f.queueId) + "::uuid"), 'dispatching'); + } else { + assert.equal(results[1].ok, true, 'A safely released pre-dispatch claim may be acquired'); + assert.equal(f.sql("SELECT claim_token::text FROM public.publication_intents WHERE id = " + + literal(f.intentId) + "::uuid"), contenderToken); + assert.equal(Number(f.sql("SELECT claim_version FROM public.publication_intents WHERE id = " + + literal(f.intentId) + "::uuid")), f.claimVersion + 1); + assert.equal(Number(f.sql("SELECT count(*) FROM public.publication_attempts WHERE queue_item_id = " + + literal(f.queueId) + "::uuid")), 0); + } + } finally { + // Unlock even when a barrier assertion fails; statement timeouts bound all + // remaining local work. Closing a session rolls back an unfinished command. + await gate.close(); + await Promise.all([owner.close(), contender.close()]); + f.sql('DROP TRIGGER IF EXISTS fixture_publication_lock_barrier ON public.' + table + + '; DROP FUNCTION IF EXISTS public.fixture_publication_lock_barrier();'); + } +} From 7ee3403d3c2fea0c9ff9427070e33a7937724430 Mon Sep 17 00:00:00 2001 From: Ayobami Haastrup <47716486+AyobamiH@users.noreply.github.com> Date: Sat, 12 Sep 2026 23:06:44 -0700 Subject: [PATCH 4/4] Require the publication lock-order repair before provider dispatch --- docs/implementation-checkpoint-20260913.md | 49 ++++++++++++++++++++++ src/publication-ledger.ts | 1 + test/publication-ledger.test.ts | 16 +++++++ 3 files changed, 66 insertions(+) create mode 100644 docs/implementation-checkpoint-20260913.md diff --git a/docs/implementation-checkpoint-20260913.md b/docs/implementation-checkpoint-20260913.md new file mode 100644 index 0000000..b8fc362 --- /dev/null +++ b/docs/implementation-checkpoint-20260913.md @@ -0,0 +1,49 @@ +# Publication concurrency repair checkpoint — 13 September 2026 + +The full engineering blueprint remains the completion scope. This change closes +a newly reproduced database concurrency defect; it does not complete the +remaining billing, job ownership, connection lifecycle, generation, fairness, +provider or account-acceptance work in [the full register](implementation-checkpoint-20260912.md). + +## Reproduction and repair + +App PR #9 merged as c46dff045e79b537661c48b92b5f0b32c7be6b4b. Its post-merge +integration failed when eight concurrent publishers produced no provider write. +That run did not retain SQLSTATE diagnostics, so its exact historical cause +cannot be recovered from the assertion alone. + +A controlled test against the unchanged schema reproduced a begin/claim +deadlock with PostgreSQL SQLSTATE 40P01: +https://github.com/AyobamiH/oneclickpostfactory/actions/runs/34741743158 + +Claims lock queue then intent. Other publication mutations locked intent or +attempt first, then needed the queue row, including the attempt foreign-key +check. The companion forward migration takes the tenant-owned queue lock first +in every publication mutation. This serialises one publication without blocking +unrelated queue identities. + +The worker requires publication-queue-lock-order-v1 before dispatch. The v1 +contract migration identity remains compatible with the previous worker, while +the database advertises lock_order_migration=20260913061000 separately. + +## Paired review and verification + +- Canonical worker PR: https://github.com/OneClickPostFactory/social-agents/pull/11 +- Schema/app follow-up: https://github.com/AyobamiH/oneclickpostfactory/pull/10 +- The app workflow pins this worker candidate and checks out the canonical + OneClickPostFactory/social-agents repository. +- Controlled begin/claim and release/claim contention, twenty rounds of eight + Worker claimers, and the existing publication failure scenarios run against + real isolated Supabase/Postgres with provider transport intercepted. +- A missing lock-order capability must block the worker before provider dispatch. +- Passing receipts belong to the exact commits recorded in the paired PR checks. + Do not infer a pass from the existence of this checkpoint. + +Promote schema first, then the paired worker, through the existing guarded +release process. The old deployed worker remains compatible with the additive +schema. Do not re-enable generation or provider dispatch as part of source +integration. No live-provider acceptance or production migration is claimed. + +Next implementation slice: durable logical agent-job identity and fenced +terminal ownership (D05 and the remaining D09), with real database competing +claim/recovery tests before consumer promotion. diff --git a/src/publication-ledger.ts b/src/publication-ledger.ts index 16317ef..c702039 100644 --- a/src/publication-ledger.ts +++ b/src/publication-ledger.ts @@ -18,6 +18,7 @@ export const REQUIRED_PUBLICATION_CAPABILITIES = [ 'publication-queue-compatibility-fence-v1', 'publication-provenance-snapshot-v1', 'publication-legacy-queue-hold-v1', + 'publication-queue-lock-order-v1', ] as const; export interface PublicationSchemaContract { diff --git a/test/publication-ledger.test.ts b/test/publication-ledger.test.ts index 01a97d4..72e8cdd 100644 --- a/test/publication-ledger.test.ts +++ b/test/publication-ledger.test.ts @@ -77,6 +77,22 @@ async function main(): Promise { ); }); + await test('publication dispatch requires the queue lock-order migration capability', async () => { + globalThis.fetch = (async () => Response.json({ + contract: PUBLICATION_SCHEMA_CONTRACT, + migration: PUBLICATION_SCHEMA_MIGRATION, + capabilities: REQUIRED_PUBLICATION_CAPABILITIES.filter( + capability => capability !== 'publication-queue-lock-order-v1' + ), + })) as typeof fetch; + await assert.rejects( + () => assertPublicationLedgerContract(), + (error: unknown) => error instanceof PublicationLedgerContractError + && error.code === 'publication_ledger_schema_unavailable' + && error.message.includes('publication-queue-lock-order-v1') + ); + }); + await test('claim, release, and dispatch preserve exact fencing identities', async () => { const bodies: Array<{ url: string; body: Record }> = []; const claimToken = '61000000-0000-4000-8000-000000000001';