From a0d123148604528e3cc138170285787f8e8d38c6 Mon Sep 17 00:00:00 2001 From: Ben Davis <45952064+bmdavis419@users.noreply.github.com> Date: Wed, 9 Sep 2026 06:31:12 -0700 Subject: [PATCH 1/9] FEATURE: Index uploads through the jobs queue Every Job variant now carries orgId and the consumer builds a layer per job for that org, so the services see only that tenant's rows. Upload, version upload and restore, rename, reindex, and site commit send an index job after their transaction commits (JobQueue.trySend logs a failed send instead of failing the request; the cron sweep covers it); the routes and MCP tools no longer index inline under waitUntil. Indexing.runOne skips a delivery whose version is no longer current and reports whether the attempt indexed, was skipped, failed permanently (recorded on the row), or needs a redelivery. Consumer decisions are { ack: true } | { retry: true, delaySeconds } with delay doubling per delivery up to an hour, and the Worker facade passes it to message.retry. Indexing.runDue is now a reconciliation sweep: rows still pending or leased fifteen minutes past their due time are re-sent as jobs, never indexed inline. Route tests swap the JOBS binding for a collecting fake with ctx.drainJobs(), which runs the consumer in-process and honours re-sends. Co-Authored-By: Claude Fable 5.1 --- apps/web/scripts/cloudflare-adapter.mjs | 14 +- apps/web/src/lib/server/job-policy.ts | 27 +++ apps/web/src/lib/server/jobs/consumer.ts | 167 +++++++++++++----- apps/web/src/lib/server/mcp/run.ts | 21 +-- apps/web/src/lib/server/mcp/server.ts | 7 +- apps/web/src/lib/server/services/files.ts | 4 +- .../lib/server/services/files/internals.ts | 12 +- .../lib/server/services/files/mutations.ts | 3 +- .../src/lib/server/services/files/upload.ts | 4 +- apps/web/src/lib/server/services/indexing.ts | 127 ++++++++++--- apps/web/src/lib/server/services/jobs.ts | 45 +++-- apps/web/src/lib/server/services/sites.ts | 4 +- .../lib/server/services/sites/internals.ts | 13 +- .../src/lib/server/services/sites/sessions.ts | 9 +- apps/web/src/routes/api/files/+server.ts | 49 ++--- apps/web/src/routes/api/files/[id]/+server.ts | 16 -- .../routes/api/files/[id]/versions/+server.ts | 38 +--- .../src/routes/api/internal/jobs/+server.ts | 17 +- .../api/sites/sessions/[id]/commit/+server.ts | 40 ++--- packages/shared/src/index.ts | 11 +- 20 files changed, 388 insertions(+), 240 deletions(-) create mode 100644 apps/web/src/lib/server/job-policy.ts diff --git a/apps/web/scripts/cloudflare-adapter.mjs b/apps/web/scripts/cloudflare-adapter.mjs index bf345a4..3e9354f 100644 --- a/apps/web/scripts/cloudflare-adapter.mjs +++ b/apps/web/scripts/cloudflare-adapter.mjs @@ -68,8 +68,9 @@ export default { }, // Queue batches are forwarded to the SvelteKit bundle in-process (the // consumer lives under $lib, unreachable from this facade). The endpoint - // returns one ack/retry decision per message id; anything it did not - // decide on is retried so a crash never silently drops work. + // returns one decision per message id, \`{ ack: true }\` or + // \`{ retry: true, delaySeconds }\`; anything it did not decide on is + // retried so a crash never silently drops work. async queue(batch, env, ctx) { const timestamp = String(Date.now()); const body = JSON.stringify({ @@ -98,10 +99,13 @@ export default { throw new Error(\`Queue consumer failed with status \${response.status}\`); } const { decisions } = await response.json(); - const actions = new Map(decisions.map((decision) => [decision.id, decision.action])); + const byId = new Map(decisions.map((decision) => [decision.id, decision])); for (const message of batch.messages) { - if (actions.get(message.id) === 'ack') message.ack(); - else message.retry(); + const decision = byId.get(message.id); + if (decision?.ack === true) message.ack(); + else if (Number.isFinite(decision?.delaySeconds) && decision.delaySeconds > 0) { + message.retry({ delaySeconds: decision.delaySeconds }); + } else message.retry(); } } }; diff --git a/apps/web/src/lib/server/job-policy.ts b/apps/web/src/lib/server/job-policy.ts new file mode 100644 index 0000000..3294eb8 --- /dev/null +++ b/apps/web/src/lib/server/job-policy.ts @@ -0,0 +1,27 @@ +// Timing rules for queue-driven work. Pure so the consumer and the +// services that send jobs can be tested without a queue. + +// Cloudflare Queues cap a message delay at 12 hours; anything later is +// re-sent by the consumer with the remainder when it arrives. +export const MAX_JOB_DELAY_SECONDS = 12 * 60 * 60; + +// A job whose row is still pending or leased this long after it should +// have run is presumed lost and re-sent by the cron reconciliation. +export const STUCK_JOB_MS = 15 * 60 * 1_000; + +export const delaySecondsUntil = (at: string | number, now = Date.now()) => { + const target = typeof at === 'number' ? at : new Date(at).getTime(); + if (!Number.isFinite(target)) return 0; + return Math.max( + 0, + Math.min(Math.ceil((target - now) / 1_000), MAX_JOB_DELAY_SECONDS) + ); +}; + +// Transient failures back off by delivery count (message.attempts is 1 +// on the first delivery): a minute, then doubling up to an hour. +export const retryDelaySeconds = (attempts: number) => + Math.min(30 * 2 ** Math.max(0, attempts), 3_600); + +export const stuckBefore = (now = Date.now()) => + new Date(now - STUCK_JOB_MS).toISOString(); diff --git a/apps/web/src/lib/server/jobs/consumer.ts b/apps/web/src/lib/server/jobs/consumer.ts index d57487b..ab5e0d9 100644 --- a/apps/web/src/lib/server/jobs/consumer.ts +++ b/apps/web/src/lib/server/jobs/consumer.ts @@ -1,7 +1,10 @@ import { JobSchema, type Job } from '@adrive/shared'; import { Effect, Schema } from 'effect'; -import { runWorkerProgram, type AppServices } from '../edge'; -import type { StorageError } from '../errors'; +import type { AppServices } from '../edge'; +import { StorageError } from '../errors'; +import { retryDelaySeconds } from '../job-policy'; +import { requestLayer } from '../layer'; +import { Indexing } from '../services/indexing'; // The subset of a Cloudflare MessageBatch the consumer needs. A real // MessageBatch satisfies it, and so does the JSON the Worker facade posts @@ -17,11 +20,46 @@ export interface JobBatch { readonly messages: ReadonlyArray; } -export type JobAction = 'ack' | 'retry'; +// What the facade does with one message. `retry` hands it back to the +// queue after `delaySeconds`; the queue's max_retries then bounds how +// often a job can come back before it dead-letters. +export type JobAction = + | { readonly ack: true } + | { readonly retry: true; readonly delaySeconds: number }; -export interface JobDecision { - readonly id: string; - readonly action: JobAction; +export type JobDecision = { readonly id: string } & JobAction; + +export const ack: JobAction = { ack: true }; +export const retryAfter = (delaySeconds: number): JobAction => ({ + retry: true, + delaySeconds +}); + +// What a job handler reports back. Handlers own their persistence: a +// failure they recorded on the row is `done`; only a failure they could +// not record (the storage itself was unreachable) asks for a redelivery, +// and a job that must run again later (a purge whose time is not yet +// up) re-sends itself and reports `done`. +export type JobOutcome = 'done' | 'retry'; + +export type JobOf = Extract; + +// One handler per job kind. To add a kind: extend JobSchema in +// packages/shared, add its handler here and in liveJobHandlers, and +// cover the dispatch in consumer.test.ts. +export interface JobHandlers { + readonly index: ( + job: JobOf<'index'> + ) => Effect.Effect; + readonly scan: ( + job: JobOf<'scan'> + ) => Effect.Effect; + readonly purge: ( + job: JobOf<'purge'> + ) => Effect.Effect; + readonly siteCleanup: ( + job: JobOf<'site-cleanup'> + ) => Effect.Effect; } const log = (entry: Record) => @@ -29,53 +67,57 @@ const log = (entry: Record) => console.log(JSON.stringify(entry)); }); -// Dispatch by job kind. Every branch only logs for now; the queue is wired -// end to end before any behaviour moves onto it. -export const runJob = ( - job: Job -): Effect.Effect => { +export const dispatchJob = (handlers: JobHandlers) => (job: Job) => { switch (job.kind) { case 'index': - // TODO(D2): indexing.runOne(job); ack when the version is stale. - return log({ - message: 'job received', - kind: job.kind, - fileId: job.fileId, - version: job.version - }); + return handlers.index(job); case 'scan': - // TODO(D2): content scanning runs beside indexing. - return log({ - message: 'job received', - kind: job.kind, - fileId: job.fileId, - version: job.version - }); + return handlers.scan(job); case 'purge': - // TODO(D3): files.purgeOne(job.fileId) after the retention delay. - return log({ - message: 'job received', - kind: job.kind, - fileId: job.fileId - }); + return handlers.purge(job); case 'site-cleanup': - // TODO(D3): sites.cleanupSession(job.sessionId). - return log({ - message: 'job received', - kind: job.kind, - sessionId: job.sessionId - }); + return handlers.siteCleanup(job); } }; +const indexOutcome = (outcome: 'indexed' | 'skipped' | 'retry' | 'failed') => + outcome === 'retry' ? ('retry' as const) : ('done' as const); + +const received = (job: Job) => + log({ message: 'job received', ...job }).pipe(Effect.as('done' as const)); + +export const liveJobHandlers = Effect.gen(function* () { + const indexing = yield* Indexing; + return { + index: (job) => indexing.runOne(job).pipe(Effect.map(indexOutcome)), + // Content scanning arrives with the abuse stack; until then the job + // is acknowledged so a stray send never dead-letters. + scan: received, + // TODO(D3): files.purgeOne(job.fileId) after the retention delay. + purge: received, + // TODO(D3): sites.cleanupSession(job.sessionId). + siteCleanup: received + } satisfies JobHandlers; +}); + +export const runJob = ( + job: Job +): Effect.Effect => + Effect.flatMap(liveJobHandlers, (handlers) => dispatchJob(handlers)(job)); + const decodeJob = Schema.decodeUnknownEffect(JobSchema); -// Invalid bodies are acked: retrying can never make them decode. Failed -// jobs are retried; the queue's max_retries and dead-letter queue bound it. +// Invalid bodies are acked: retrying can never make them decode. A +// StorageError (Postgres, R2, or Workers AI unreachable) is transient and +// retried with backoff; the queue's max_retries and dead-letter queue +// bound it. const consumeMessage = ( queue: string, message: JobMessage, - run: (job: Job) => Effect.Effect + run: ( + job: Job, + attempts: number + ) => Effect.Effect ) => Effect.gen(function* () { const decoded = yield* Effect.result(decodeJob(message.body)); @@ -86,32 +128,61 @@ const consumeMessage = ( id: message.id, cause: String(decoded.failure) }); - return 'ack' as const; + return ack; } - const outcome = yield* Effect.result(run(decoded.success)); + const job = decoded.success; + const outcome = yield* Effect.result(run(job, message.attempts)); if (outcome._tag === 'Failure') { yield* log({ message: 'job failed', queue, id: message.id, - kind: decoded.success.kind, + kind: job.kind, + orgId: job.orgId, attempts: message.attempts, cause: String(outcome.failure.cause) }); - return 'retry' as const; + return retryAfter(retryDelaySeconds(message.attempts)); } - return 'ack' as const; + if (outcome.success === 'retry') { + yield* log({ + message: 'job asked to run again', + queue, + id: message.id, + kind: job.kind, + orgId: job.orgId, + attempts: message.attempts + }); + return retryAfter(retryDelaySeconds(message.attempts)); + } + return ack; }); export const consumeBatch = ( batch: JobBatch, - run: (job: Job) => Effect.Effect + run: ( + job: Job, + attempts: number + ) => Effect.Effect ) => Effect.forEach(batch.messages, (message) => consumeMessage(batch.queue, message, run).pipe( - Effect.map((action): JobDecision => ({ id: message.id, action })) + Effect.map((action): JobDecision => ({ id: message.id, ...action })) + ) + ); + +// Each job runs in its own layer bound to the job's org (the same layer +// runWorkerProgram builds), so the services only ever see that tenant's +// rows. A defect (a bug, not a StorageError) escapes the batch as a +// rejection; the facade then retries every message, which is the safe +// default for an unknown failure. +export const runJobForOrg = (env: Env) => (job: Job) => + runJob(job).pipe( + Effect.provide(requestLayer(env, { orgId: job.orgId, userId: 'system' })), + Effect.catchTag('SqlError', (cause) => + Effect.fail(new StorageError({ operation: 'connect for job', cause })) ) ); export const handleJobBatch = (env: Env, batch: JobBatch) => - runWorkerProgram(env, consumeBatch(batch, runJob)); + Effect.runPromise(consumeBatch(batch, runJobForOrg(env))); diff --git a/apps/web/src/lib/server/mcp/run.ts b/apps/web/src/lib/server/mcp/run.ts index cbe8054..564f169 100644 --- a/apps/web/src/lib/server/mcp/run.ts +++ b/apps/web/src/lib/server/mcp/run.ts @@ -1,9 +1,8 @@ import { Cause, Effect, Exit } from 'effect'; -import { isAppError, runWorkerProgram, type AppServices } from '../edge'; +import { isAppError, type AppServices } from '../edge'; import type { AppError } from '../errors'; import type { ProgramIdentity } from '../identity'; import { requestLayer } from '../layer'; -import { Indexing } from '../services/indexing'; export type McpRunSuccess = { readonly ok: true; readonly value: A }; export type McpRunFailure = { @@ -62,21 +61,3 @@ export const runMcp = async ( if (Exit.isSuccess(exit)) return { ok: true, value: exit.value }; return failureFromCause(exit.cause); }; - -export const scheduleIndex = ( - env: Env, - ctx: ExecutionContext, - identity: ProgramIdentity, - fileId: string -) => { - ctx.waitUntil( - runWorkerProgram( - env, - Effect.gen(function* () { - const indexing = yield* Indexing; - yield* indexing.process(fileId); - }), - identity - ) - ); -}; diff --git a/apps/web/src/lib/server/mcp/server.ts b/apps/web/src/lib/server/mcp/server.ts index c8dcb6e..053dcf0 100644 --- a/apps/web/src/lib/server/mcp/server.ts +++ b/apps/web/src/lib/server/mcp/server.ts @@ -27,7 +27,7 @@ import { decodeExclusiveContent, mcpPageLimit } from './payload'; -import { runMcp, scheduleIndex } from './run'; +import { runMcp } from './run'; export const READ_TOOL_NAMES = [ 'whoami', @@ -304,7 +304,7 @@ const registerReadTools = (server: McpServer, input: McpServerInput) => { }; const registerWriteTools = (server: McpServer, input: McpServerInput) => { - const { env, ctx, credential } = input; + const { env, credential } = input; server.registerTool( 'put_file', @@ -370,7 +370,6 @@ const registerWriteTools = (server: McpServer, input: McpServerInput) => { if (uploaded.value.kind === 'rate-limit') { return errorResult(uploaded.value.message, 429); } - scheduleIndex(env, ctx, credential, uploaded.value.file.id); const { kind: _kind, ...value } = uploaded.value; return jsonResult(value); } @@ -395,7 +394,6 @@ const registerWriteTools = (server: McpServer, input: McpServerInput) => { }) ); if (!result.ok) return errorResult(result.message, result.status); - scheduleIndex(env, ctx, credential, result.value.file.id); return jsonResult(result.value); } ); @@ -579,7 +577,6 @@ const registerWriteTools = (server: McpServer, input: McpServerInput) => { if (published.value.kind === 'rate-limit') { return errorResult(published.value.message, 429); } - scheduleIndex(env, ctx, credential, published.value.file.id); const { kind: _kind, ...value } = published.value; return jsonResult(value); } diff --git a/apps/web/src/lib/server/services/files.ts b/apps/web/src/lib/server/services/files.ts index bd0d75d..6871bdd 100644 --- a/apps/web/src/lib/server/services/files.ts +++ b/apps/web/src/lib/server/services/files.ts @@ -2,6 +2,7 @@ import { Context, Effect, Layer } from 'effect'; import { AppConfig } from '../config'; import { PgSql } from '../pg'; import { Blobs } from './blobs'; +import { JobQueue } from './jobs'; import { Tags } from './tags'; import { CurrentOrg } from './current-org'; import { createInternals } from './files/internals'; @@ -29,7 +30,8 @@ const makeFiles = Effect.gen(function* () { const config = yield* AppConfig; const tags = yield* Tags; const org = yield* CurrentOrg; - const internals = createInternals({ sql, blobs, config, tags, org }); + const jobs = yield* JobQueue; + const internals = createInternals({ sql, blobs, config, tags, org, jobs }); return Files.of({ ...uploadOps(internals), diff --git a/apps/web/src/lib/server/services/files/internals.ts b/apps/web/src/lib/server/services/files/internals.ts index a6cbbec..f9fa35f 100644 --- a/apps/web/src/lib/server/services/files/internals.ts +++ b/apps/web/src/lib/server/services/files/internals.ts @@ -16,6 +16,7 @@ import { refreshSearchDocument } from '../../search-index'; import { ensureStorageHeadroom, reserveWithinPlan } from '../../storage-quota'; import type { AppConfig } from '../../config'; import type { Blobs } from '../blobs'; +import type { JobQueue } from '../jobs'; import type { Tags } from '../tags'; import type { CurrentOrg } from '../current-org'; import type { MutationResult } from './types'; @@ -26,10 +27,11 @@ export interface CoreDeps { readonly config: AppConfig['Service']; readonly tags: Tags['Service']; readonly org: CurrentOrg['Service']; + readonly jobs: JobQueue['Service']; } export const createInternals = (deps: CoreDeps) => { - const { sql, blobs, config, tags, org } = deps; + const { sql, blobs, config, tags, org, jobs } = deps; const compensateStoredBlob = ( failure: OriginalError, fileId: string, @@ -73,6 +75,11 @@ export const createInternals = (deps: CoreDeps) => { const reserveBytes = (orgId: string, delta: number) => reserveWithinPlan(sql, orgId, delta); + // Sent after the transaction that made the work necessary has + // committed; a lost send is caught by the cron reconciliation. + const sendIndexJob = (fileId: string, version: number) => + jobs.trySend({ kind: 'index', orgId: org.id, fileId, version }); + const findDashboardFile = Effect.fn('Files.findDashboardFile')(function* ( id: string ) { @@ -144,6 +151,7 @@ export const createInternals = (deps: CoreDeps) => { ) ) ); + yield* sendIndexJob(current.id, version); return { file: { ...current, @@ -168,9 +176,11 @@ export const createInternals = (deps: CoreDeps) => { config, tags, org, + jobs, compensateStoredBlob, ensureHeadroom, reserveBytes, + sendIndexJob, findDashboardFile, commitStoredVersion }; diff --git a/apps/web/src/lib/server/services/files/mutations.ts b/apps/web/src/lib/server/services/files/mutations.ts index bb4f697..9cf71ce 100644 --- a/apps/web/src/lib/server/services/files/mutations.ts +++ b/apps/web/src/lib/server/services/files/mutations.ts @@ -26,7 +26,7 @@ export const mutationOps = ( | 'recordDownload' > => { const { sql, org } = internals; - const { findDashboardFile } = internals; + const { findDashboardFile, sendIndexJob } = internals; return { setVisibility: Effect.fn('Files.setVisibility')(function* (id, isPublic) { const current = yield* findDashboardFile(id); @@ -161,6 +161,7 @@ export const mutationOps = ( (cause) => new StorageError({ operation: 'rename file', cause }) ) ); + yield* sendIndexJob(id, current.version); return { file: { ...current, diff --git a/apps/web/src/lib/server/services/files/upload.ts b/apps/web/src/lib/server/services/files/upload.ts index 4af2cd0..abaa33f 100644 --- a/apps/web/src/lib/server/services/files/upload.ts +++ b/apps/web/src/lib/server/services/files/upload.ts @@ -21,7 +21,8 @@ export const uploadOps = ( reserveBytes, compensateStoredBlob, commitStoredVersion, - findDashboardFile + findDashboardFile, + sendIndexJob } = internals; return { upload: Effect.fn('Files.upload')(function* (input) { @@ -102,6 +103,7 @@ export const uploadOps = ( ) ); forgetTagListCache(org.id); + yield* sendIndexJob(id, 1); return { file: { diff --git a/apps/web/src/lib/server/services/indexing.ts b/apps/web/src/lib/server/services/indexing.ts index c33dd7f..9bab6ad 100644 --- a/apps/web/src/lib/server/services/indexing.ts +++ b/apps/web/src/lib/server/services/indexing.ts @@ -17,9 +17,11 @@ import { } from '../semantic-policy'; import { isSearchableText, searchTextLimit } from '../search-text'; import { createTtlCache } from '../isolate-cache'; +import { stuckBefore } from '../job-policy'; import { PgSql } from '../pg'; import { Blobs } from './blobs'; import { CurrentOrg } from './current-org'; +import { JobQueue } from './jobs'; import { Embedder, VectorIndex } from './semantic'; const INDEX_LEASE_MS = 5 * 60 * 1_000; @@ -58,9 +60,24 @@ export interface SemanticStatus { readonly costNotice: string; } +// What one indexing attempt did. `retry` is the only outcome that asks +// the queue to deliver the job again; a permanent failure is already +// recorded on the row and a skipped job has nothing left to do. +export type IndexOutcome = 'indexed' | 'skipped' | 'retry' | 'failed'; + export interface IndexingShape { + // Resets the row and sends an index job for its current version. readonly enqueue: (fileId: string) => Effect.Effect; + // One attempt for the current version, errors recorded on the row. readonly process: (fileId: string) => Effect.Effect; + // The queue consumer's entry point: skips when `version` is no longer + // current so a stale delivery never overwrites a newer index. + readonly runOne: (job: { + readonly fileId: string; + readonly version: number; + }) => Effect.Effect; + // Reconciliation: re-sends jobs for rows stuck past their lease or + // retry time, never indexes inline. Returns how many were re-sent. readonly runDue: (limit: number) => Effect.Effect; readonly status: Effect.Effect; } @@ -75,6 +92,7 @@ const makeIndexing = Effect.gen(function* () { const blobs = yield* Blobs; const embedder = yield* Embedder; const vectors = yield* VectorIndex; + const jobs = yield* JobQueue; const storage = (operation: string) => (cause: unknown) => new StorageError({ operation, cause }); @@ -117,11 +135,32 @@ const makeIndexing = Effect.gen(function* () { error }) ); + if (!stateChanged) return 'skipped' as const; + return disposition.state === 'failed' + ? ('failed' as const) + : ('retry' as const); }); - const perform = Effect.fn('Indexing.perform')(function* (fileId: string) { + const perform = Effect.fn('Indexing.perform')(function* ( + fileId: string, + expectedVersion: number | null + ) { const initial = yield* findJob(fileId); - if (!initial) return; + if (!initial) return 'skipped' as const; + if ( + expectedVersion !== null && + initial.current_version !== expectedVersion + ) { + console.log( + JSON.stringify({ + message: 'stale index job skipped', + fileId, + version: expectedVersion, + currentVersion: initial.current_version + }) + ); + return 'skipped' as const; + } const now = new Date(); const leaseUntil = new Date(now.getTime() + INDEX_LEASE_MS).toISOString(); const lease = { @@ -138,7 +177,7 @@ const makeIndexing = Effect.gen(function* () { leaseUntil, MAX_INDEX_ATTEMPTS ).pipe(Effect.mapError(storage('claim indexing job'))); - if (!claimed) return; + if (!claimed) return 'skipped' as const; const stale = (message: string) => Effect.sync(() => { @@ -152,7 +191,7 @@ const makeIndexing = Effect.gen(function* () { ); }); - yield* Effect.gen(function* () { + return yield* Effect.gen(function* () { const text = initial.index_cursor >= 1 && initial.text_content !== null ? initial.text_content @@ -165,7 +204,8 @@ const makeIndexing = Effect.gen(function* () { Effect.mapError(storage('store extracted search text')) ); if (!extracted) { - return yield* stale('stale semantic extraction ignored'); + yield* stale('stale semantic extraction ignored'); + return 'skipped' as const; } if (!embedder.enabled || !vectors.enabled) { @@ -174,8 +214,9 @@ const makeIndexing = Effect.gen(function* () { ); if (!finished) { yield* stale('stale keyword-only indexing completion ignored'); + return 'skipped' as const; } - return; + return 'indexed' as const; } const chunks = chunkSearchText(initial.display_name, text); @@ -203,7 +244,9 @@ const makeIndexing = Effect.gen(function* () { ).pipe(Effect.mapError(storage('commit semantic index state'))); if (!committed) { yield* stale('stale semantic indexing completion rolled back'); + return 'skipped' as const; } + return 'indexed' as const; }).pipe( Effect.catchCause((cause) => markFailure(lease, Cause.pretty(cause)).pipe( @@ -218,6 +261,9 @@ const makeIndexing = Effect.gen(function* () { cause: String(recordCause) }) ); + // The lease still holds; the row is picked up again once + // it lapses, so the delivery itself is not retried. + return 'skipped' as const; }) ) ) @@ -226,7 +272,7 @@ const makeIndexing = Effect.gen(function* () { }); const process = Effect.fn('Indexing.process')(function* (fileId: string) { - yield* perform(fileId).pipe( + yield* perform(fileId, null).pipe( Effect.catchCause((cause) => Effect.sync(() => { console.error( @@ -241,39 +287,66 @@ const makeIndexing = Effect.gen(function* () { ); }); + const runOne = Effect.fn('Indexing.runOne')(function* (job: { + readonly fileId: string; + readonly version: number; + }) { + return yield* perform(job.fileId, job.version); + }); + + const sendIndexJob = (fileId: string, version: number) => + jobs.trySend({ kind: 'index', orgId: org.id, fileId, version }); + + // Rows still pending or leased long after they were due lost their + // delivery somewhere (a crashed consumer, a queue outage during the + // send); they get a fresh job. Stamping index_next_run_at throttles + // the re-send to once per stuck window and keeps a lapsed `running` + // lease claimable. Disabled rows are re-sent once semantic search is + // switched on so they gain embeddings. const runDue = Effect.fn('Indexing.runDue')(function* (limit: number) { const bounded = Math.max(1, Math.min(limit, 10)); - const now = new Date().toISOString(); + const now = new Date(); + const cutoff = stuckBefore(now.getTime()); const includeDisabled = embedder.enabled && vectors.enabled; - const rows = yield* sql<{ id: string }>` - SELECT id - FROM files - WHERE org_id = ${org.id} - AND deleted_at IS NULL - AND (expires_at IS NULL OR expires_at > ${now}) - AND index_attempts < ${MAX_INDEX_ATTEMPTS} - AND ( - (index_state = 'pending' - AND (index_next_run_at IS NULL OR index_next_run_at <= ${now})) - OR (index_state = 'running' AND index_next_run_at <= ${now}) - OR (index_state = 'disabled' AND ${includeDisabled}::boolean) - ) - ORDER BY COALESCE(index_next_run_at, updated_at), id - LIMIT ${bounded}`.pipe(Effect.mapError(storage('list due indexing jobs'))); - for (const row of rows) yield* process(row.id); + const rows = yield* sql<{ id: string; current_version: number }>` + UPDATE files + SET index_next_run_at = ${now.toISOString()} + WHERE id IN ( + SELECT id + FROM files + WHERE org_id = ${org.id} + AND deleted_at IS NULL + AND (expires_at IS NULL OR expires_at > ${now.toISOString()}) + AND index_attempts < ${MAX_INDEX_ATTEMPTS} + AND ( + (index_state IN ('pending', 'running') + AND COALESCE(index_next_run_at, updated_at) <= ${cutoff}) + OR (index_state = 'disabled' AND ${includeDisabled}::boolean + AND COALESCE(index_next_run_at, updated_at) <= ${cutoff}) + ) + ORDER BY COALESCE(index_next_run_at, updated_at), id + LIMIT ${bounded} + ) AND org_id = ${org.id} + RETURNING id, current_version`.pipe( + Effect.mapError(storage('list stuck indexing jobs')) + ); + for (const row of rows) yield* sendIndexJob(row.id, row.current_version); return rows.length; }); const enqueue = Effect.fn('Indexing.enqueue')(function* (fileId: string) { - yield* sql` + const rows = yield* sql<{ current_version: number }>` UPDATE files SET index_state = 'pending', index_cursor = 0, index_attempts = 0, index_error = NULL, index_next_run_at = NULL, index_lease_token = NULL WHERE id = ${fileId} AND org_id = ${org.id} AND deleted_at IS NULL - AND (expires_at IS NULL OR expires_at > ${new Date().toISOString()})`.pipe( + AND (expires_at IS NULL OR expires_at > ${new Date().toISOString()}) + RETURNING current_version`.pipe( Effect.mapError(storage('enqueue semantic indexing')) ); + const row = rows[0]; + if (row) yield* sendIndexJob(fileId, row.current_version); }); const status = Effect.gen(function* () { @@ -293,7 +366,7 @@ const makeIndexing = Effect.gen(function* () { return result; }).pipe(Effect.withSpan('Indexing.status')); - return Indexing.of({ enqueue, process, runDue, status }); + return Indexing.of({ enqueue, process, runOne, runDue, status }); }); export const IndexingLive = Layer.effect(Indexing, makeIndexing); diff --git a/apps/web/src/lib/server/services/jobs.ts b/apps/web/src/lib/server/services/jobs.ts index 7d4bc8a..a9ea672 100644 --- a/apps/web/src/lib/server/services/jobs.ts +++ b/apps/web/src/lib/server/services/jobs.ts @@ -12,24 +12,49 @@ export interface JobQueueShape { job: Job, options?: JobSendOptions ) => Effect.Effect; + // For sends that follow a committed write: the row already says the + // work is owed, so a queue outage is logged and left to the cron + // reconciliation rather than failing the request. + readonly trySend: (job: Job, options?: JobSendOptions) => Effect.Effect; } export class JobQueue extends Context.Service()( 'app/JobQueue' ) {} +const withTrySend = (send: JobQueueShape['send']): JobQueueShape => ({ + send, + trySend: (job, options) => + send(job, options).pipe( + Effect.catchCause((cause) => + Effect.sync(() => { + console.error( + JSON.stringify({ + message: 'job could not be sent', + kind: job.kind, + orgId: job.orgId, + cause: String(cause) + }) + ); + }) + ) + ) +}); + const makeJobQueue = Effect.gen(function* () { const queue = yield* Jobs; - return JobQueue.of({ - send: Effect.fn('JobQueue.send')(function* (job, options) { - yield* Effect.tryPromise({ - try: () => queue.send(job, { contentType: 'json', ...options }), - catch: (cause) => - new StorageError({ operation: `enqueue ${job.kind} job`, cause }) - }); - }) - }); + return JobQueue.of( + withTrySend( + Effect.fn('JobQueue.send')(function* (job, options) { + yield* Effect.tryPromise({ + try: () => queue.send(job, { contentType: 'json', ...options }), + catch: (cause) => + new StorageError({ operation: `enqueue ${job.kind} job`, cause }) + }); + }) + ) + ); }); export const JobQueueLive = Layer.effect(JobQueue, makeJobQueue); @@ -38,5 +63,5 @@ export const JobQueueLive = Layer.effect(JobQueue, makeJobQueue); // go nowhere. export const JobQueueNull = Layer.succeed( JobQueue, - JobQueue.of({ send: () => Effect.void }) + JobQueue.of(withTrySend(() => Effect.void)) ); diff --git a/apps/web/src/lib/server/services/sites.ts b/apps/web/src/lib/server/services/sites.ts index a1ab7f6..babe064 100644 --- a/apps/web/src/lib/server/services/sites.ts +++ b/apps/web/src/lib/server/services/sites.ts @@ -3,6 +3,7 @@ import { AppConfig } from '../config'; import { PgSql } from '../pg'; import { Blobs } from './blobs'; import { CurrentOrg } from './current-org'; +import { JobQueue } from './jobs'; import { cleanupOps } from './sites/cleanup'; import { createInternals } from './sites/internals'; import { readOps } from './sites/read'; @@ -23,8 +24,9 @@ const makeSites = Effect.gen(function* () { const blobs = yield* Blobs; const config = yield* AppConfig; const org = yield* CurrentOrg; + const jobs = yield* JobQueue; - const internals = createInternals({ sql, blobs, config, org }); + const internals = createInternals({ sql, blobs, config, org, jobs }); return Sites.of({ ...sessionOps(internals), diff --git a/apps/web/src/lib/server/services/sites/internals.ts b/apps/web/src/lib/server/services/sites/internals.ts index 38bfadb..10d6813 100644 --- a/apps/web/src/lib/server/services/sites/internals.ts +++ b/apps/web/src/lib/server/services/sites/internals.ts @@ -6,6 +6,7 @@ import { NotFound, StorageError } from '../../errors'; import { siteCleanupDisposition } from '../../site-policy'; import { Blobs } from '../blobs'; import type { CurrentOrg } from '../current-org'; +import type { JobQueue } from '../jobs'; import { PendingDeleteRow, SiteSessionRow, @@ -18,9 +19,16 @@ interface CoreDeps { readonly blobs: Blobs['Service']; readonly config: AppConfig['Service']; readonly org: CurrentOrg['Service']; + readonly jobs: JobQueue['Service']; } -export const createInternals = ({ sql, blobs, config, org }: CoreDeps) => { +export const createInternals = ({ + sql, + blobs, + config, + org, + jobs +}: CoreDeps) => { const all = ( statement: Effect.Effect, unknown>, schema: Schema.Codec, @@ -282,7 +290,8 @@ export const createInternals = ({ sql, blobs, config, org }: CoreDeps) => { sql, blobs, config, - org + org, + jobs }; }; diff --git a/apps/web/src/lib/server/services/sites/sessions.ts b/apps/web/src/lib/server/services/sites/sessions.ts index 9ad50bb..2f3d47c 100644 --- a/apps/web/src/lib/server/services/sites/sessions.ts +++ b/apps/web/src/lib/server/services/sites/sessions.ts @@ -27,7 +27,8 @@ export const sessionOps = ( sql, blobs, config, - org + org, + jobs } = internals; return { @@ -426,6 +427,12 @@ export const sessionOps = ( ) ); + yield* jobs.trySend({ + kind: 'index', + orgId: org.id, + fileId: session.fileId, + version: session.version + }); const cleanupPending = yield* drainDeletes(session.fileId).pipe( Effect.catchCause((cause) => Effect.sync(() => { diff --git a/apps/web/src/routes/api/files/+server.ts b/apps/web/src/routes/api/files/+server.ts index 7eb2831..0e5dee0 100644 --- a/apps/web/src/routes/api/files/+server.ts +++ b/apps/web/src/routes/api/files/+server.ts @@ -116,10 +116,9 @@ export const GET: RequestHandler = (event) => { ); }; -export const PUT: RequestHandler = async (event) => { +export const PUT: RequestHandler = (event) => { const { request, url } = event; - const output = await runEdgeWithEvent( - event, + return runEdge( Effect.gen(function* () { const authGuard = yield* AuthGuard; const files = yield* Files; @@ -129,13 +128,10 @@ export const PUT: RequestHandler = async (event) => { credential.credentialId ); if (!rateLimit.allowed) { - return { - fileId: null, - response: authRateLimitResponse( - rateLimit, - 'Too many uploads. Try again later.' - ) - }; + return authRateLimitResponse( + rateLimit, + 'Too many uploads. Try again later.' + ); } const displayName = yield* Effect.try({ try: () => decodeName(request.headers.get('x-adrive-file-name')), @@ -186,33 +182,16 @@ export const PUT: RequestHandler = async (event) => { }) }) }); - return { - fileId: result.file.id, - response: Response.json( - { - file: result.file, - url: `${yield* currentContentOrigin}/f/${result.file.id}`, - forcedPublic: result.forcedPublic - }, - { status: 201 } - ) - }; + return Response.json( + { + file: result.file, + url: `${yield* currentContentOrigin}/f/${result.file.id}`, + forcedPublic: result.forcedPublic + }, + { status: 201 } + ); }) ); - const uploadedFileId = output.fileId; - if (uploadedFileId !== null && event.platform) { - event.platform.ctx.waitUntil( - runWorkerProgram( - event.platform.env, - Effect.gen(function* () { - const indexing = yield* Indexing; - yield* indexing.process(uploadedFileId); - }), - event.locals.auth - ) - ); - } - return output.response; }; export const DELETE: RequestHandler = async (event) => { diff --git a/apps/web/src/routes/api/files/[id]/+server.ts b/apps/web/src/routes/api/files/[id]/+server.ts index 98b3fe8..f8e4e2d 100644 --- a/apps/web/src/routes/api/files/[id]/+server.ts +++ b/apps/web/src/routes/api/files/[id]/+server.ts @@ -103,27 +103,11 @@ export const PATCH: RequestHandler = async (event) => { })) ); return { - reindex: - mutation.action === 'reindex' || - mutation.action === 'rename' || - mutation.action === 'restore-version', purge: mutation.action === 'purge', response: Response.json(result) }; }) ); - if (output.reindex && event.platform) { - event.platform.ctx.waitUntil( - runWorkerProgram( - event.platform.env, - Effect.gen(function* () { - const indexing = yield* Indexing; - yield* indexing.process(params.id); - }), - event.locals.auth - ) - ); - } if (output.purge && event.platform) { event.platform.ctx.waitUntil( runWorkerProgram( diff --git a/apps/web/src/routes/api/files/[id]/versions/+server.ts b/apps/web/src/routes/api/files/[id]/versions/+server.ts index d8188dd..27a3857 100644 --- a/apps/web/src/routes/api/files/[id]/versions/+server.ts +++ b/apps/web/src/routes/api/files/[id]/versions/+server.ts @@ -1,16 +1,14 @@ import type { RequestHandler } from './$types'; import { Effect } from 'effect'; import { authRateLimitResponse } from '$lib/server/auth-rate-limit-response'; -import { runEdgeWithEvent, runWorkerProgram } from '$lib/server/edge'; +import { runEdge } from '$lib/server/edge'; import { requireWrite } from '$lib/server/request-auth'; import { AuthGuard } from '$lib/server/services/auth-guard'; import { Files } from '$lib/server/services/files'; -import { Indexing } from '$lib/server/services/indexing'; -export const PUT: RequestHandler = async (event) => { +export const PUT: RequestHandler = (event) => { const { params, request } = event; - const output = await runEdgeWithEvent( - event, + return runEdge( Effect.gen(function* () { const authGuard = yield* AuthGuard; const files = yield* Files; @@ -20,13 +18,10 @@ export const PUT: RequestHandler = async (event) => { credential.credentialId ); if (!rateLimit.allowed) { - return { - fileId: null, - response: authRateLimitResponse( - rateLimit, - 'Too many uploads. Try again later.' - ) - }; + return authRateLimitResponse( + rateLimit, + 'Too many uploads. Try again later.' + ); } const result = yield* files.uploadVersion({ id: params.id, @@ -35,24 +30,7 @@ export const PUT: RequestHandler = async (event) => { contentLength: request.headers.get('content-length'), body: request.body }); - return { - fileId: params.id, - response: Response.json(result, { status: 201 }) - }; + return Response.json(result, { status: 201 }); }) ); - const uploadedFileId = output.fileId; - if (uploadedFileId !== null && event.platform) { - event.platform.ctx.waitUntil( - runWorkerProgram( - event.platform.env, - Effect.gen(function* () { - const indexing = yield* Indexing; - yield* indexing.process(uploadedFileId); - }), - event.locals.auth - ) - ); - } - return output.response; }; diff --git a/apps/web/src/routes/api/internal/jobs/+server.ts b/apps/web/src/routes/api/internal/jobs/+server.ts index 5b2bc64..2fdac5e 100644 --- a/apps/web/src/routes/api/internal/jobs/+server.ts +++ b/apps/web/src/routes/api/internal/jobs/+server.ts @@ -4,13 +4,15 @@ import { AppConfig } from '$lib/server/config'; import { verifyJobsRequest } from '$lib/server/cron-auth'; import { runEdge } from '$lib/server/edge'; import { InvalidRequest, Unauthorized } from '$lib/server/errors'; -import { consumeBatch, runJob } from '$lib/server/jobs/consumer'; +import { consumeBatch, runJobForOrg } from '$lib/server/jobs/consumer'; import { readBoundedText } from '$lib/server/request-json'; // The Worker facade (scripts/cloudflare-adapter.mjs) receives queue // batches and forwards them here in-process, because the consumer has to // run inside the SvelteKit bundle to reach $lib. The response tells the -// facade which messages to ack and which to retry. +// facade which messages to ack and which to retry (and after how long). +// The request itself carries no tenant; every job builds its own layer +// for the org named in its body. const JobBatchBody = Schema.Struct({ queue: Schema.String, messages: Schema.Array( @@ -27,10 +29,17 @@ const MAX_BATCH_BYTES = 1024 * 1024; const unauthorized = () => new Unauthorized({ message: 'Queue request is unauthorized' }); -export const POST: RequestHandler = ({ request }) => +export const POST: RequestHandler = ({ request, platform }) => runEdge( Effect.gen(function* () { const config = yield* AppConfig; + const env = platform?.env; + if (!env) { + return yield* new InvalidRequest({ + status: 400, + message: 'Cloudflare bindings unavailable' + }); + } const text = yield* readBoundedText(request, { maxBytes: MAX_BATCH_BYTES, invalidLengthMessage: 'Queue batch is too large', @@ -63,7 +72,7 @@ export const POST: RequestHandler = ({ request }) => }) ) ); - const decisions = yield* consumeBatch(batch, runJob); + const decisions = yield* consumeBatch(batch, runJobForOrg(env)); return Response.json({ decisions }); }) ); diff --git a/apps/web/src/routes/api/sites/sessions/[id]/commit/+server.ts b/apps/web/src/routes/api/sites/sessions/[id]/commit/+server.ts index bfca5dd..d7beaf0 100644 --- a/apps/web/src/routes/api/sites/sessions/[id]/commit/+server.ts +++ b/apps/web/src/routes/api/sites/sessions/[id]/commit/+server.ts @@ -1,42 +1,24 @@ import type { RequestHandler } from './$types'; import { Effect } from 'effect'; -import { runEdgeWithEvent, runWorkerProgram } from '$lib/server/edge'; +import { runEdge } from '$lib/server/edge'; import { requireWrite } from '$lib/server/request-auth'; import { currentContentOrigin } from '$lib/server/services/current-org'; -import { Indexing } from '$lib/server/services/indexing'; import { Sites } from '$lib/server/services/sites'; -export const POST: RequestHandler = async (event) => { - const { params, request, url } = event; - const output = await runEdgeWithEvent( - event, +export const POST: RequestHandler = (event) => { + const { params } = event; + return runEdge( Effect.gen(function* () { const sites = yield* Sites; yield* requireWrite(event); const result = yield* sites.commit(params.id); - return { - fileId: result.file.id, - response: Response.json( - { - ...result, - url: `${yield* currentContentOrigin}/s/${result.file.id}/` - }, - { status: 201 } - ) - }; + return Response.json( + { + ...result, + url: `${yield* currentContentOrigin}/s/${result.file.id}/` + }, + { status: 201 } + ); }) ); - if (event.platform) { - event.platform.ctx.waitUntil( - runWorkerProgram( - event.platform.env, - Effect.gen(function* () { - const indexing = yield* Indexing; - yield* indexing.process(output.fileId); - }), - event.locals.auth - ) - ); - } - return output.response; }; diff --git a/packages/shared/src/index.ts b/packages/shared/src/index.ts index 970c12f..b9d9e6c 100644 --- a/packages/shared/src/index.ts +++ b/packages/shared/src/index.ts @@ -385,26 +385,31 @@ export const ErrorResponseSchema = Schema.Struct({ export const API_KEY_PATTERN = /^adr_([A-Za-z0-9]{8})_([A-Za-z0-9_-]{24,})$/; -// Background work carried over the Cloudflare Queue. Tenancy adds orgId -// to every variant later; keep the union closed so the consumer's switch -// stays exhaustive. +// Background work carried over the Cloudflare Queue. Every job names the +// org it acts for; the consumer runs it with that org as CurrentOrg so the +// services see only that tenant's rows. Keep the union closed so the +// consumer's switch stays exhaustive. export const JobSchema = Schema.Union([ Schema.Struct({ kind: Schema.Literal('index'), + orgId: Schema.String, fileId: Schema.String, version: Schema.Int }), Schema.Struct({ kind: Schema.Literal('purge'), + orgId: Schema.String, fileId: Schema.String }), Schema.Struct({ kind: Schema.Literal('scan'), + orgId: Schema.String, fileId: Schema.String, version: Schema.Int }), Schema.Struct({ kind: Schema.Literal('site-cleanup'), + orgId: Schema.String, sessionId: Schema.String }) ]); From ecdf66a322556ca7bcf5e41b9bb73582fc58cbfc Mon Sep 17 00:00:00 2001 From: Ben Davis <45952064+bmdavis419@users.noreply.github.com> Date: Wed, 9 Sep 2026 06:43:56 -0700 Subject: [PATCH 2/9] FEATURE: Purge and clean up through delayed jobs Trashing a file, setting an expiry (on upload or later), purging from the trash, and emptying the trash send a purge job delayed until the file's deadline, capped at the queue's twelve hour maximum. The consumer calls the new Files.purgeOne, extracted from sweepPurges: it re-sends itself with the remainder when the deadline is still ahead or the row is backing off from a failed delete, otherwise claims the row, deletes the blobs, and completes the purge. A restored file is left alone by its stale job. Site session creation sends a site-cleanup job delayed by the session TTL; Sites.cleanupSession (extracted from sweepLifecycle) aborts the session if it is still open and expired, and createSession no longer sweeps expired sessions inline. Both cron sweeps are now reconciliation only: sweepPurges re-sends jobs for files due more than fifteen minutes ago, sweepLifecycle for sessions expired that long ago, still under runAcrossOrgs. The routes no longer run sweeps under waitUntil. The per-job layer is built with { local: true } so the jobs route's tenant-less layer cannot leak its anonymous org into a job. Co-Authored-By: Claude Fable 5.1 --- apps/web/src/lib/server/jobs/consumer.ts | 32 +- .../lib/server/services/files/internals.ts | 9 + .../lib/server/services/files/mutations.ts | 29 +- .../src/lib/server/services/files/purge.ts | 283 +++++++++++------- .../src/lib/server/services/files/types.ts | 5 + .../src/lib/server/services/files/upload.ts | 4 +- .../src/lib/server/services/sites/cleanup.ts | 64 +++- .../lib/server/services/sites/internals.ts | 30 -- .../src/lib/server/services/sites/sessions.ts | 19 +- .../src/lib/server/services/sites/types.ts | 6 + apps/web/src/routes/api/files/+server.ts | 28 +- apps/web/src/routes/api/files/[id]/+server.ts | 27 +- 12 files changed, 309 insertions(+), 227 deletions(-) diff --git a/apps/web/src/lib/server/jobs/consumer.ts b/apps/web/src/lib/server/jobs/consumer.ts index ab5e0d9..527d6d2 100644 --- a/apps/web/src/lib/server/jobs/consumer.ts +++ b/apps/web/src/lib/server/jobs/consumer.ts @@ -4,7 +4,9 @@ import type { AppServices } from '../edge'; import { StorageError } from '../errors'; import { retryDelaySeconds } from '../job-policy'; import { requestLayer } from '../layer'; -import { Indexing } from '../services/indexing'; +import { Files } from '../services/files'; +import { Indexing, type IndexOutcome } from '../services/indexing'; +import { Sites } from '../services/sites'; // The subset of a Cloudflare MessageBatch the consumer needs. A real // MessageBatch satisfies it, and so does the JSON the Worker facade posts @@ -80,23 +82,26 @@ export const dispatchJob = (handlers: JobHandlers) => (job: Job) => { } }; -const indexOutcome = (outcome: 'indexed' | 'skipped' | 'retry' | 'failed') => - outcome === 'retry' ? ('retry' as const) : ('done' as const); +// Only an attempt that could not run asks for a redelivery; a permanent +// failure is already on the row and a stale version has nothing to do. +export const indexOutcome = (outcome: IndexOutcome): JobOutcome => + outcome === 'retry' ? 'retry' : 'done'; const received = (job: Job) => log({ message: 'job received', ...job }).pipe(Effect.as('done' as const)); export const liveJobHandlers = Effect.gen(function* () { const indexing = yield* Indexing; + const files = yield* Files; + const sites = yield* Sites; return { index: (job) => indexing.runOne(job).pipe(Effect.map(indexOutcome)), // Content scanning arrives with the abuse stack; until then the job // is acknowledged so a stray send never dead-letters. scan: received, - // TODO(D3): files.purgeOne(job.fileId) after the retention delay. - purge: received, - // TODO(D3): sites.cleanupSession(job.sessionId). - siteCleanup: received + purge: (job) => files.purgeOne(job.fileId).pipe(Effect.as('done')), + siteCleanup: (job) => + sites.cleanupSession(job.sessionId).pipe(Effect.as('done')) } satisfies JobHandlers; }); @@ -173,12 +178,17 @@ export const consumeBatch = ( // Each job runs in its own layer bound to the job's org (the same layer // runWorkerProgram builds), so the services only ever see that tenant's -// rows. A defect (a bug, not a StorageError) escapes the batch as a -// rejection; the facade then retries every message, which is the safe -// default for an unknown failure. +// rows. `local` matters: the jobs route already runs under a tenant-less +// layer, and nested provides otherwise share memoized services, which +// would hand the job the anonymous org. A defect (a bug, not a +// StorageError) escapes the batch as a rejection; the facade then +// retries every message, which is the safe default for an unknown +// failure. export const runJobForOrg = (env: Env) => (job: Job) => runJob(job).pipe( - Effect.provide(requestLayer(env, { orgId: job.orgId, userId: 'system' })), + Effect.provide(requestLayer(env, { orgId: job.orgId, userId: 'system' }), { + local: true + }), Effect.catchTag('SqlError', (cause) => Effect.fail(new StorageError({ operation: 'connect for job', cause })) ) diff --git a/apps/web/src/lib/server/services/files/internals.ts b/apps/web/src/lib/server/services/files/internals.ts index f9fa35f..86f5b5c 100644 --- a/apps/web/src/lib/server/services/files/internals.ts +++ b/apps/web/src/lib/server/services/files/internals.ts @@ -12,6 +12,7 @@ import { toDashboardFile } from '../../file-rows'; import { visibilityForFile } from '../../file-policy'; +import { delaySecondsUntil } from '../../job-policy'; import { refreshSearchDocument } from '../../search-index'; import { ensureStorageHeadroom, reserveWithinPlan } from '../../storage-quota'; import type { AppConfig } from '../../config'; @@ -79,6 +80,13 @@ export const createInternals = (deps: CoreDeps) => { // committed; a lost send is caught by the cron reconciliation. const sendIndexJob = (fileId: string, version: number) => jobs.trySend({ kind: 'index', orgId: org.id, fileId, version }); + // Delayed until the row's deadline (capped by the queue; the job + // re-sends itself with the remainder when it arrives early). + const sendPurgeJob = (fileId: string, dueAt: string) => + jobs.trySend( + { kind: 'purge', orgId: org.id, fileId }, + { delaySeconds: delaySecondsUntil(dueAt) } + ); const findDashboardFile = Effect.fn('Files.findDashboardFile')(function* ( id: string @@ -181,6 +189,7 @@ export const createInternals = (deps: CoreDeps) => { ensureHeadroom, reserveBytes, sendIndexJob, + sendPurgeJob, findDashboardFile, commitStoredVersion }; diff --git a/apps/web/src/lib/server/services/files/mutations.ts b/apps/web/src/lib/server/services/files/mutations.ts index 9cf71ce..226a0a2 100644 --- a/apps/web/src/lib/server/services/files/mutations.ts +++ b/apps/web/src/lib/server/services/files/mutations.ts @@ -26,7 +26,7 @@ export const mutationOps = ( | 'recordDownload' > => { const { sql, org } = internals; - const { findDashboardFile, sendIndexJob } = internals; + const { findDashboardFile, sendIndexJob, sendPurgeJob } = internals; return { setVisibility: Effect.fn('Files.setVisibility')(function* (id, isPublic) { const current = yield* findDashboardFile(id); @@ -78,6 +78,7 @@ export const mutationOps = ( if (rows.length !== 1) { return yield* new NotFound({ id }); } + yield* sendPurgeJob(id, purgeAt); return { file: { ...current, deletedAt, updatedAt: deletedAt }, forcedPublic: false @@ -122,6 +123,7 @@ export const mutationOps = ( new StorageError({ operation: 'update file expiration', cause }) ) ); + if (expiresAt !== null) yield* sendPurgeJob(id, expiresAt); return { file: { ...current, expiresAt, updatedAt }, forcedPublic: false @@ -204,27 +206,28 @@ export const mutationOps = ( message: 'This file is already being purged' }); } + yield* sendPurgeJob(id, EPOCH); return { file: current, forcedPublic: false }; }), - // Suspended so the org is read when the effect runs, not when the + // A generator body reads the org when the effect runs, not when the // layer is built (content routes build the layer with no tenant). - scheduleAllPurgesNow: Effect.suspend( - () => sql<{ id: string }>` + scheduleAllPurgesNow: Effect.gen(function* () { + const rows = yield* sql<{ id: string }>` UPDATE files SET purge_at = ${EPOCH}, purge_state = 'none', purge_error = NULL, purge_next_run_at = NULL WHERE org_id = ${org.id} AND deleted_at IS NOT NULL AND purge_state <> 'pending' RETURNING id - ` - ).pipe( - Effect.map((rows) => rows.length), - Effect.mapError( - (cause) => - new StorageError({ operation: 'schedule empty trash', cause }) - ), - Effect.withSpan('Files.scheduleAllPurgesNow') - ), + `.pipe( + Effect.mapError( + (cause) => + new StorageError({ operation: 'schedule empty trash', cause }) + ) + ); + for (const row of rows) yield* sendPurgeJob(row.id, EPOCH); + return rows.length; + }).pipe(Effect.withSpan('Files.scheduleAllPurgesNow')), recordDownload: Effect.fn('Files.recordDownload')(function* (id) { const now = new Date().toISOString(); // Content routes have no tenant; the file id alone identifies it. diff --git a/apps/web/src/lib/server/services/files/purge.ts b/apps/web/src/lib/server/services/files/purge.ts index 385fac6..72b5e0e 100644 --- a/apps/web/src/lib/server/services/files/purge.ts +++ b/apps/web/src/lib/server/services/files/purge.ts @@ -1,133 +1,196 @@ import { dashboardThumbnailPrefix } from '../../../file-thumbnail'; import { StorageError } from '../../errors'; +import { stuckBefore } from '../../job-policy'; import { completePurge } from '../../purge-sql'; import { retryAt, safeIndexError } from '../../semantic-policy'; -import { Effect } from 'effect'; +import { Effect, Schema } from 'effect'; import { forgetTagListCache } from '../tags'; import type { FileInternals } from './internals'; import type { FilesShape } from './types'; +const PURGE_LEASE_MS = 5 * 60 * 1_000; + +const PurgeRow = Schema.Struct({ + deleted_at: Schema.NullOr(Schema.String), + purge_at: Schema.NullOr(Schema.String), + expires_at: Schema.NullOr(Schema.String), + purge_state: Schema.String, + purge_next_run_at: Schema.NullOr(Schema.String) +}); + +const decodePurgeRows = Schema.decodeUnknownOption(Schema.Array(PurgeRow)); + +// When the file is owed a purge: the trash deadline once it is in the +// trash, or its expiry, whichever comes first. Null means it is neither +// trashed nor expiring (restored, or the expiry was cleared). +export const purgeDueAt = (row: { + readonly deleted_at: string | null; + readonly purge_at: string | null; + readonly expires_at: string | null; +}) => { + const candidates = [ + row.deleted_at !== null ? row.purge_at : null, + row.expires_at + ].filter((value): value is string => value !== null); + if (candidates.length === 0) return null; + return candidates.reduce((earliest, value) => + value < earliest ? value : earliest + ); +}; + export const purgeOps = ( internals: FileInternals -): Pick => { - const { sql, blobs, org } = internals; - return { - sweepPurges: Effect.fn('Files.sweepPurges')(function* (limit) { - const bounded = Math.max(1, Math.min(limit, 10)); - const now = new Date().toISOString(); - const due = yield* sql<{ id: string }>` - SELECT id - FROM files - WHERE org_id = ${org.id} AND (( - ( +): Pick => { + const { sql, blobs, org, sendPurgeJob } = internals; + const storage = (operation: string) => (cause: unknown) => + new StorageError({ operation, cause }); + + const claim = (fileId: string, now: string) => + sql<{ id: string }>` + UPDATE files + SET purge_state = 'pending', + deleted_at = COALESCE(deleted_at, ${now}), + purge_at = COALESCE(purge_at, ${now}), + purge_attempts = purge_attempts + 1, + purge_error = NULL, + purge_next_run_at = ${new Date(Date.parse(now) + PURGE_LEASE_MS).toISOString()} + WHERE id = ${fileId} AND org_id = ${org.id} AND ( + ( + purge_state IN ('none', 'failed') + AND (purge_next_run_at IS NULL OR purge_next_run_at <= ${now}) + AND ( (expires_at IS NOT NULL AND expires_at <= ${now}) OR (deleted_at IS NOT NULL AND purge_at IS NOT NULL AND purge_at <= ${now}) ) - AND purge_state IN ('none', 'failed') - AND (purge_next_run_at IS NULL OR purge_next_run_at <= ${now}) - ) OR ( - purge_state = 'pending' - AND (purge_next_run_at IS NULL OR purge_next_run_at <= ${now}) - )) - ORDER BY COALESCE(purge_at, expires_at), id - LIMIT ${bounded} - `.pipe( - Effect.mapError( - (cause) => - new StorageError({ operation: 'list files due for purge', cause }) ) - ); + OR (purge_state = 'pending' AND purge_next_run_at <= ${now}) + ) + RETURNING id`.pipe( + Effect.map((rows) => rows.length === 1), + Effect.mapError(storage('claim file purge')) + ); - for (const row of due) { - const leaseUntil = new Date(Date.now() + 5 * 60 * 1_000).toISOString(); - const claimed = yield* sql<{ id: string }>` - UPDATE files - SET purge_state = 'pending', - deleted_at = COALESCE(deleted_at, ${now}), - purge_at = COALESCE(purge_at, ${now}), - purge_attempts = purge_attempts + 1, - purge_error = NULL, purge_next_run_at = ${leaseUntil} - WHERE id = ${row.id} AND org_id = ${org.id} AND ( - ( - purge_state IN ('none', 'failed') - AND ( - (expires_at IS NOT NULL AND expires_at <= ${now}) - OR (deleted_at IS NOT NULL AND purge_at IS NOT NULL AND purge_at <= ${now}) + const deleteBlobs = (fileId: string) => + Effect.all( + [ + sql<{ version: number; r2_key: string }>` + SELECT version, r2_key FROM file_versions WHERE file_id = ${fileId}`, + sql<{ r2_key: string }>` + SELECT r2_key FROM site_assets WHERE file_id = ${fileId}` + ], + { concurrency: 'unbounded' } + ).pipe( + Effect.mapError(storage('list file purge keys')), + Effect.flatMap(([versions, assets]) => + blobs + .deleteMany([ + ...versions + .filter((item) => !item.r2_key.startsWith('site-version/')) + .map((item) => item.r2_key), + ...assets.map((item) => item.r2_key) + ]) + .pipe( + Effect.andThen( + blobs.deletePrefixes( + versions.map((item) => + dashboardThumbnailPrefix(fileId, item.version) + ) ) ) - OR ( - purge_state = 'pending' - AND (purge_next_run_at IS NULL OR purge_next_run_at <= ${now}) - ) ) - RETURNING id - `.pipe( - Effect.mapError( - (cause) => - new StorageError({ operation: 'claim file purge', cause }) - ) - ); - if (claimed.length !== 1) continue; + ) + ); - const deletion = yield* Effect.all( - [ - sql<{ version: number; r2_key: string }>` - SELECT version, r2_key FROM file_versions WHERE file_id = ${row.id}`, - sql<{ r2_key: string }>` - SELECT r2_key FROM site_assets WHERE file_id = ${row.id}` - ], - { concurrency: 'unbounded' } - ).pipe( - Effect.map(([versions, assets]) => ({ - keys: [ - ...versions - .filter((item) => !item.r2_key.startsWith('site-version/')) - .map((item) => item.r2_key), - ...assets.map((item) => item.r2_key) - ], - thumbnailPrefixes: versions.map((item) => - dashboardThumbnailPrefix(row.id, item.version) - ) - })), - Effect.mapError( - (cause) => - new StorageError({ operation: 'list file purge keys', cause }) - ) - ); + // A failed blob delete keeps the row, backs off on the row's own + // schedule, and re-sends the job for that moment; the queue's retry + // budget is reserved for failures that could not even be recorded. + const recordFailure = (fileId: string, failure: StorageError) => + Effect.gen(function* () { + const attempts = yield* sql<{ purge_attempts: number }>` + SELECT purge_attempts FROM files WHERE id = ${fileId}`.pipe( + Effect.map((rows) => rows[0]?.purge_attempts ?? 1) + ); + const nextRunAt = retryAt(attempts); + yield* sql` + UPDATE files + SET purge_state = 'failed', + purge_error = ${safeIndexError(failure.cause)}, + purge_next_run_at = ${nextRunAt} + WHERE id = ${fileId} AND purge_state = 'pending'`; + return nextRunAt; + }).pipe(Effect.mapError(storage('record file purge failure'))); - const deleted = yield* blobs.deleteMany(deletion.keys).pipe( - Effect.andThen(blobs.deletePrefixes(deletion.thumbnailPrefixes)), - Effect.as(true), - Effect.catch((failure) => - Effect.gen(function* () { - const attempts = yield* sql<{ purge_attempts: number }>` - SELECT purge_attempts FROM files WHERE id = ${row.id}`.pipe( - Effect.map((rows) => rows[0]?.purge_attempts ?? 1) - ); - yield* sql` - UPDATE files - SET purge_state = 'failed', - purge_error = ${safeIndexError(failure.cause)}, - purge_next_run_at = ${retryAt(attempts)} - WHERE id = ${row.id} AND purge_state = 'pending'`; - }).pipe( - Effect.mapError( - (cause) => - new StorageError({ - operation: 'record file purge failure', - cause - }) - ), - Effect.as(false) - ) + const purgeOne = Effect.fn('Files.purgeOne')(function* (fileId: string) { + const rows = yield* sql` + SELECT deleted_at, purge_at, expires_at, purge_state, purge_next_run_at + FROM files + WHERE id = ${fileId} AND org_id = ${org.id} + LIMIT 1`.pipe(Effect.mapError(storage('find file to purge'))); + const decoded = decodePurgeRows(rows); + const row = decoded._tag === 'Some' ? decoded.value[0] : undefined; + // Already purged, or restored with no expiry: nothing is owed. + if (!row) return; + const dueAt = purgeDueAt(row); + if (dueAt === null) return; + + const now = new Date().toISOString(); + // Not yet due (the queue caps delays at twelve hours, so long + // retentions arrive early), or backing off from a failed delete, or + // leased by another attempt: come back when the row says so. + const resumeAt = [dueAt, row.purge_next_run_at] + .filter((value): value is string => value !== null && value > now) + .sort() + .at(-1); + if (resumeAt !== undefined) { + return yield* sendPurgeJob(fileId, resumeAt); + } + + const claimed = yield* claim(fileId, now); + if (!claimed) return; + + const deleted = yield* deleteBlobs(fileId).pipe( + Effect.as(true), + Effect.catch((failure) => + recordFailure(fileId, failure).pipe( + Effect.tap((nextRunAt) => sendPurgeJob(fileId, nextRunAt)), + Effect.as(false) + ) + ) + ); + if (!deleted) return; + + yield* completePurge(sql, org.id, fileId); + forgetTagListCache(org.id); + }); + + // Reconciliation: rows whose purge was due long ago and that no + // attempt is currently leasing lost their delivery; they get a fresh + // job. Stamping purge_next_run_at keeps one re-send per stuck window. + const sweepPurges = Effect.fn('Files.sweepPurges')(function* (limit: number) { + const bounded = Math.max(1, Math.min(limit, 10)); + const now = new Date(); + const cutoff = stuckBefore(now.getTime()); + const rows = yield* sql<{ id: string }>` + UPDATE files + SET purge_next_run_at = ${now.toISOString()} + WHERE id IN ( + SELECT id + FROM files + WHERE org_id = ${org.id} + AND purge_state IN ('none', 'failed', 'pending') + AND (purge_next_run_at IS NULL OR purge_next_run_at <= ${cutoff}) + AND ( + (expires_at IS NOT NULL AND expires_at <= ${cutoff}) + OR (deleted_at IS NOT NULL AND purge_at IS NOT NULL + AND purge_at <= ${cutoff}) ) - ); - if (!deleted) continue; + ORDER BY COALESCE(purge_at, expires_at), id + LIMIT ${bounded} + ) AND org_id = ${org.id} + RETURNING id`.pipe(Effect.mapError(storage('list stuck file purges'))); + for (const row of rows) yield* sendPurgeJob(row.id, now.toISOString()); + return rows.length; + }); - yield* completePurge(sql, org.id, row.id); - forgetTagListCache(org.id); - } - return due.length; - }) - } satisfies Pick; + return { purgeOne, sweepPurges }; }; diff --git a/apps/web/src/lib/server/services/files/types.ts b/apps/web/src/lib/server/services/files/types.ts index 0131d3d..e8c1928 100644 --- a/apps/web/src/lib/server/services/files/types.ts +++ b/apps/web/src/lib/server/services/files/types.ts @@ -132,6 +132,11 @@ export interface FilesShape { DashboardThumbnailStoreResult, InvalidRequest | NotFound | StorageError >; + // One purge attempt, run from the queue. Re-sends itself when the + // file's deadline (trash retention or expiry) is still ahead. + readonly purgeOne: (fileId: string) => Effect.Effect; + // Reconciliation: re-sends jobs for files due long ago that the queue + // never finished. Returns how many were re-sent. readonly sweepPurges: (limit: number) => Effect.Effect; readonly findContent: ( id: string, diff --git a/apps/web/src/lib/server/services/files/upload.ts b/apps/web/src/lib/server/services/files/upload.ts index abaa33f..d8f5c63 100644 --- a/apps/web/src/lib/server/services/files/upload.ts +++ b/apps/web/src/lib/server/services/files/upload.ts @@ -22,7 +22,8 @@ export const uploadOps = ( compensateStoredBlob, commitStoredVersion, findDashboardFile, - sendIndexJob + sendIndexJob, + sendPurgeJob } = internals; return { upload: Effect.fn('Files.upload')(function* (input) { @@ -104,6 +105,7 @@ export const uploadOps = ( ); forgetTagListCache(org.id); yield* sendIndexJob(id, 1); + if (input.expiresAt !== null) yield* sendPurgeJob(id, input.expiresAt); return { file: { diff --git a/apps/web/src/lib/server/services/sites/cleanup.ts b/apps/web/src/lib/server/services/sites/cleanup.ts index 81bdee5..acbb3ca 100644 --- a/apps/web/src/lib/server/services/sites/cleanup.ts +++ b/apps/web/src/lib/server/services/sites/cleanup.ts @@ -1,19 +1,73 @@ import { Effect } from 'effect'; +import { delaySecondsUntil, stuckBefore } from '../../job-policy'; import type { SiteInternals } from './internals'; -import type { SitesShape } from './types'; +import { SiteSessionRow, type SitesShape } from './types'; export const cleanupOps = ( internals: SiteInternals -): Pick => { - const { sweepExpiredSessions, sweepPendingDeletes } = internals; +): Pick => { + const { all, cleanupStaged, sweepPendingDeletes, sql, org, jobs } = internals; + + const sendCleanupJob = (sessionId: string, at: string) => + jobs.trySend( + { kind: 'site-cleanup', orgId: org.id, sessionId }, + { delaySeconds: delaySecondsUntil(at) } + ); return { + // Runs from the queue once the session's TTL is up. A session that + // committed or aborted in the meantime has nothing left to clean; + // one still open but not yet expired (its expiry was pushed out) + // comes back when it is. + cleanupSession: Effect.fn('Sites.cleanupSession')(function* ( + sessionId: string + ) { + const rows = yield* all( + sql` + SELECT id, file_id, display_name, version, status, created_at, + expires_at + FROM site_upload_sessions + WHERE id = ${sessionId} AND org_id = ${org.id} + LIMIT 1`, + SiteSessionRow, + 'find site upload session to clean' + ); + const session = rows[0]; + if (!session || session.status !== 'open') return; + const now = new Date().toISOString(); + if (session.expires_at > now) { + return yield* sendCleanupJob(session.id, session.expires_at); + } + yield* cleanupStaged( + { id: session.id, fileId: session.file_id, version: session.version }, + 'aborted' + ); + }), + // Reconciliation: sessions expired long ago that the queue never + // cleaned get a fresh job, and R2 deletes that failed earlier are + // retried. Returns how many of either were handled. sweepLifecycle: Effect.fn('Sites.sweepLifecycle')(function* ( limit: number ) { - const expired = yield* sweepExpiredSessions(limit); + const bounded = Math.max(1, Math.min(limit, 25)); + const cutoff = stuckBefore(); + const stuck = yield* all( + sql` + SELECT id, file_id, display_name, version, status, created_at, + expires_at + FROM site_upload_sessions + WHERE org_id = ${org.id} AND status = 'open' + AND expires_at <= ${cutoff} + ORDER BY expires_at + LIMIT ${bounded}`, + SiteSessionRow, + 'list stuck site upload sessions' + ); + for (const session of stuck) { + yield* sendCleanupJob(session.id, session.expires_at); + } const pending = yield* sweepPendingDeletes(limit); - return expired + pending; + return stuck.length + pending; }) }; }; diff --git a/apps/web/src/lib/server/services/sites/internals.ts b/apps/web/src/lib/server/services/sites/internals.ts index 10d6813..ffb648d 100644 --- a/apps/web/src/lib/server/services/sites/internals.ts +++ b/apps/web/src/lib/server/services/sites/internals.ts @@ -227,35 +227,6 @@ export const createInternals = ({ yield* drainDeletes(session.fileId); }); - const sweepExpiredSessions = Effect.fn('Sites.sweepExpiredSessions')( - function* (limit = 10) { - const now = new Date().toISOString(); - const bounded = Math.max(1, Math.min(limit, 25)); - const rows = yield* all( - sql` - SELECT id, file_id, display_name, version, status, created_at, - expires_at - FROM site_upload_sessions - WHERE org_id = ${org.id} AND status = 'open' AND expires_at <= ${now} - ORDER BY expires_at - LIMIT ${bounded}`, - SiteSessionRow, - 'list expired site upload sessions' - ); - for (const row of rows) { - yield* cleanupStaged( - { - id: row.id, - fileId: row.file_id, - version: row.version - }, - 'aborted' - ); - } - return rows.length; - } - ); - const sweepPendingDeletes = Effect.fn('Sites.sweepPendingDeletes')(function* ( limit: number ) { @@ -285,7 +256,6 @@ export const createInternals = ({ compensateStagedBlob, drainDeletes, cleanupStaged, - sweepExpiredSessions, sweepPendingDeletes, sql, blobs, diff --git a/apps/web/src/lib/server/services/sites/sessions.ts b/apps/web/src/lib/server/services/sites/sessions.ts index 2f3d47c..25957ec 100644 --- a/apps/web/src/lib/server/services/sites/sessions.ts +++ b/apps/web/src/lib/server/services/sites/sessions.ts @@ -1,6 +1,7 @@ import { normalizeSitePath } from '@adrive/shared'; import { Effect } from 'effect'; import { InvalidRequest, NotFound, StorageError } from '../../errors'; +import { delaySecondsUntil } from '../../job-policy'; import { refreshSearchDocument } from '../../search-index'; import { ensureStorageHeadroom, reserveWithinPlan } from '../../storage-quota'; import { @@ -23,7 +24,6 @@ export const sessionOps = ( compensateStagedBlob, cleanupStaged, drainDeletes, - sweepExpiredSessions, sql, blobs, config, @@ -33,18 +33,6 @@ export const sessionOps = ( return { createSession: Effect.fn('Sites.createSession')(function* (input) { - yield* sweepExpiredSessions().pipe( - Effect.catchCause((cause) => - Effect.sync(() => { - console.error( - JSON.stringify({ - message: 'expired site session sweep failed', - cause: String(cause) - }) - ); - }) - ) - ); const prepared = yield* Effect.try({ try: () => prepareSiteManifest(input, config.maxUploadBytes), catch: (cause) => @@ -119,6 +107,11 @@ export const sessionOps = ( }) ) ); + // Abandoned sessions are cleaned when the TTL is up. + yield* jobs.trySend( + { kind: 'site-cleanup', orgId: org.id, sessionId: id }, + { delaySeconds: delaySecondsUntil(expiresAt) } + ); yield* drainDeletes(fileId).pipe( Effect.catchCause((cause) => Effect.sync(() => { diff --git a/apps/web/src/lib/server/services/sites/types.ts b/apps/web/src/lib/server/services/sites/types.ts index 7a32d6a..34c7137 100644 --- a/apps/web/src/lib/server/services/sites/types.ts +++ b/apps/web/src/lib/server/services/sites/types.ts @@ -114,6 +114,12 @@ export interface SitesShape { readonly version?: number; } ) => Effect.Effect; + // Aborts an open session whose TTL is up, run from the queue. + readonly cleanupSession: ( + sessionId: string + ) => Effect.Effect; + // Reconciliation: re-sends cleanup for sessions expired long ago and + // retries failed asset deletes. readonly sweepLifecycle: ( limit: number ) => Effect.Effect; diff --git a/apps/web/src/routes/api/files/+server.ts b/apps/web/src/routes/api/files/+server.ts index 0e5dee0..edf5e5e 100644 --- a/apps/web/src/routes/api/files/+server.ts +++ b/apps/web/src/routes/api/files/+server.ts @@ -1,6 +1,6 @@ import type { RequestHandler } from './$types'; import { Effect } from 'effect'; -import { runEdge, runEdgeWithEvent, runWorkerProgram } from '$lib/server/edge'; +import { runEdge } from '$lib/server/edge'; import { requireAuth, requireWrite } from '$lib/server/request-auth'; import { AppConfig } from '$lib/server/config'; import { validateExpiration } from '$lib/server/auth-policy'; @@ -194,10 +194,9 @@ export const PUT: RequestHandler = (event) => { ); }; -export const DELETE: RequestHandler = async (event) => { - const { request, url } = event; - const output = await runEdgeWithEvent( - event, +export const DELETE: RequestHandler = (event) => { + const { url } = event; + return runEdge( Effect.gen(function* () { const files = yield* Files; yield* requireWrite(event); @@ -207,23 +206,8 @@ export const DELETE: RequestHandler = async (event) => { message: 'Only trash can be emptied' }); } - return { - count: yield* files.scheduleAllPurgesNow, - response: Response.json({ ok: true as const }) - }; + yield* files.scheduleAllPurgesNow; + return Response.json({ ok: true as const }); }) ); - if (output.count > 0 && event.platform) { - event.platform.ctx.waitUntil( - runWorkerProgram( - event.platform.env, - Effect.gen(function* () { - const files = yield* Files; - yield* files.sweepPurges(10); - }), - event.locals.auth - ) - ); - } - return output.response; }; diff --git a/apps/web/src/routes/api/files/[id]/+server.ts b/apps/web/src/routes/api/files/[id]/+server.ts index f8e4e2d..1eca08c 100644 --- a/apps/web/src/routes/api/files/[id]/+server.ts +++ b/apps/web/src/routes/api/files/[id]/+server.ts @@ -1,7 +1,7 @@ import { FileMutationSchema } from '@adrive/shared'; import type { RequestHandler } from './$types'; import { Effect } from 'effect'; -import { runEdge, runEdgeWithEvent, runWorkerProgram } from '$lib/server/edge'; +import { runEdge } from '$lib/server/edge'; import { requireAuth, requireWrite } from '$lib/server/request-auth'; import { AppConfig } from '$lib/server/config'; import { validateExpiration } from '$lib/server/auth-policy'; @@ -59,10 +59,9 @@ export const GET: RequestHandler = (event) => { ); }; -export const PATCH: RequestHandler = async (event) => { - const { params, request, url } = event; - const output = await runEdgeWithEvent( - event, +export const PATCH: RequestHandler = (event) => { + const { params, request } = event; + return runEdge( Effect.gen(function* () { const files = yield* Files; const indexing = yield* Indexing; @@ -102,23 +101,7 @@ export const PATCH: RequestHandler = async (event) => { forcedPublic: false })) ); - return { - purge: mutation.action === 'purge', - response: Response.json(result) - }; + return Response.json(result); }) ); - if (output.purge && event.platform) { - event.platform.ctx.waitUntil( - runWorkerProgram( - event.platform.env, - Effect.gen(function* () { - const files = yield* Files; - yield* files.sweepPurges(1); - }), - event.locals.auth - ) - ); - } - return output.response; }; From 4d02e21ed7dd84fbabac543204923da25bf3cb2f Mon Sep 17 00:00:00 2001 From: Ben Davis <45952064+bmdavis419@users.noreply.github.com> Date: Wed, 9 Sep 2026 06:47:07 -0700 Subject: [PATCH 3/9] FEATURE: Record dead-lettered jobs Migration 0005 adds failed_jobs (id, org_id, kind, payload, error, attempts, failed_at, resolved_at). Both wrangler blocks gain a second consumer for the dead-letter queue; the Worker facade dispatches on the queue name, sending -dlq batches to /api/internal/jobs/dead, which records each message keyed by its id (so a duplicate delivery cannot double-record), acks it, and POSTs a summary to ALERT_WEBHOOK_URL when that secret is set. Bodies that no longer decode as a job are kept as kind 'invalid' with no org. The signed batch parsing is shared by both internal routes. GET /api/admin/failed-jobs lists the caller's org's rows for owners; the admin UI follows in stack E. Co-Authored-By: Claude Fable 5.1 --- apps/web/migrations-pg/0005_failed_jobs.sql | 20 +++ apps/web/scripts/cloudflare-adapter.mjs | 13 +- apps/web/src/lib/server/failed-jobs.ts | 120 ++++++++++++++++++ apps/web/src/lib/server/jobs/batch-request.ts | 59 +++++++++ apps/web/src/lib/server/jobs/dead-letters.ts | 55 ++++++++ .../routes/api/admin/failed-jobs/+server.ts | 25 ++++ .../src/routes/api/internal/jobs/+server.ts | 58 +-------- .../routes/api/internal/jobs/dead/+server.ts | 24 ++++ apps/web/wrangler.jsonc | 15 +++ docs/release.md | 11 +- packages/shared/src/index.ts | 19 +++ 11 files changed, 357 insertions(+), 62 deletions(-) create mode 100644 apps/web/migrations-pg/0005_failed_jobs.sql create mode 100644 apps/web/src/lib/server/failed-jobs.ts create mode 100644 apps/web/src/lib/server/jobs/batch-request.ts create mode 100644 apps/web/src/lib/server/jobs/dead-letters.ts create mode 100644 apps/web/src/routes/api/admin/failed-jobs/+server.ts create mode 100644 apps/web/src/routes/api/internal/jobs/dead/+server.ts diff --git a/apps/web/migrations-pg/0005_failed_jobs.sql b/apps/web/migrations-pg/0005_failed_jobs.sql new file mode 100644 index 0000000..d29a0eb --- /dev/null +++ b/apps/web/migrations-pg/0005_failed_jobs.sql @@ -0,0 +1,20 @@ +-- migrate:up +-- Jobs the queue gave up on. The Worker consumes the dead-letter queue +-- into this table so an operator (and the admin UI) can see what failed +-- and why, then re-send once the cause is fixed. Platform table: rows +-- are looked up by org, never joined into tenant queries, so no RLS. +CREATE TABLE failed_jobs ( + id text PRIMARY KEY, + org_id text, + kind text NOT NULL, + payload jsonb NOT NULL, + error text NOT NULL, + attempts integer NOT NULL, + failed_at timestamptz NOT NULL DEFAULT now(), + resolved_at timestamptz +); + +CREATE INDEX failed_jobs_org_idx ON failed_jobs (org_id, failed_at DESC); + +-- migrate:down +DROP TABLE IF EXISTS failed_jobs; diff --git a/apps/web/scripts/cloudflare-adapter.mjs b/apps/web/scripts/cloudflare-adapter.mjs index 3e9354f..ad92141 100644 --- a/apps/web/scripts/cloudflare-adapter.mjs +++ b/apps/web/scripts/cloudflare-adapter.mjs @@ -67,11 +67,16 @@ export default { ); }, // Queue batches are forwarded to the SvelteKit bundle in-process (the - // consumer lives under $lib, unreachable from this facade). The endpoint - // returns one decision per message id, \`{ ack: true }\` or - // \`{ retry: true, delaySeconds }\`; anything it did not decide on is + // consumer lives under $lib, unreachable from this facade). The main + // queue goes to the jobs endpoint, its dead-letter queue (named with a + // -dlq suffix in both environments) to the dead endpoint that records + // the job. Either returns one decision per message id, \`{ ack: true }\` + // or \`{ retry: true, delaySeconds }\`; anything it did not decide on is // retried so a crash never silently drops work. async queue(batch, env, ctx) { + const path = batch.queue.endsWith('-dlq') + ? '/api/internal/jobs/dead' + : '/api/internal/jobs'; const timestamp = String(Date.now()); const body = JSON.stringify({ queue: batch.queue, @@ -83,7 +88,7 @@ export default { }); const signature = await jobsSignature(env.MAINTENANCE_SECRET, timestamp, body); const response = await sveltekit.fetch( - new Request(new URL('/api/internal/jobs', env.DASHBOARD_ORIGIN), { + new Request(new URL(path, env.DASHBOARD_ORIGIN), { method: 'POST', headers: { 'Content-Type': 'application/json', diff --git a/apps/web/src/lib/server/failed-jobs.ts b/apps/web/src/lib/server/failed-jobs.ts new file mode 100644 index 0000000..4bd38a7 --- /dev/null +++ b/apps/web/src/lib/server/failed-jobs.ts @@ -0,0 +1,120 @@ +import { JobSchema, type FailedJob } from '@adrive/shared'; +import type { PgClient } from '@effect/sql-pg'; +import { Effect, Schema } from 'effect'; +import { StorageError } from './errors'; + +export interface DeadLetter { + readonly id: string; + readonly body: unknown; + readonly attempts: number; + readonly error: string; +} + +const decodeJob = Schema.decodeUnknownOption(JobSchema); + +const FailedJobRow = Schema.Struct({ + id: Schema.String, + kind: Schema.String, + payload: Schema.Unknown, + error: Schema.String, + attempts: Schema.Int, + failed_at: Schema.String, + resolved_at: Schema.NullOr(Schema.String) +}); + +const decodeRows = Schema.decodeUnknownOption(Schema.Array(FailedJobRow)); + +// One row per dead-lettered message, keyed by the message id so the +// at-least-once DLQ delivery cannot double-record. A body that no longer +// decodes as a job is kept too (kind 'invalid', no org) so nothing the +// queue rejected disappears. +export const recordFailedJob = (sql: PgClient.PgClient, letter: DeadLetter) => + Effect.gen(function* () { + const job = decodeJob(letter.body); + const orgId = job._tag === 'Some' ? job.value.orgId : null; + const kind = job._tag === 'Some' ? job.value.kind : 'invalid'; + const rows = yield* sql<{ id: string }>` + INSERT INTO failed_jobs (id, org_id, kind, payload, error, attempts) + VALUES ( + ${letter.id}, ${orgId}, ${kind}, ${JSON.stringify(letter.body ?? null)}::jsonb, + ${letter.error}, ${letter.attempts} + ) + ON CONFLICT (id) DO NOTHING + RETURNING id`; + return { orgId, kind, recorded: rows.length === 1 }; + }).pipe( + Effect.mapError( + (cause) => new StorageError({ operation: 'record failed job', cause }) + ) + ); + +export const listFailedJobs = ( + sql: PgClient.PgClient, + orgId: string, + limit = 100 +) => + sql` + SELECT id, kind, payload, error, attempts, failed_at, resolved_at + FROM failed_jobs + WHERE org_id = ${orgId} + ORDER BY failed_at DESC, id + LIMIT ${Math.max(1, Math.min(limit, 500))}`.pipe( + Effect.map((rows): ReadonlyArray => { + const decoded = decodeRows(rows); + return decoded._tag === 'Some' + ? decoded.value.map((row) => ({ + id: row.id, + kind: row.kind, + payload: row.payload, + error: row.error, + attempts: row.attempts, + failedAt: row.failed_at, + resolvedAt: row.resolved_at + })) + : []; + }), + Effect.mapError( + (cause) => new StorageError({ operation: 'list failed jobs', cause }) + ) + ); + +// Optional operator alert. The URL is a secret rather than a var so it +// stays out of wrangler.jsonc; absent means no alert. Delivery is best +// effort: the rows are already committed and the batch is acked either +// way, so a webhook outage cannot dead-letter the dead letters. +export const alertWebhookUrl = (env: Env) => { + const value: unknown = Reflect.get(env, 'ALERT_WEBHOOK_URL'); + return typeof value === 'string' && value.length > 0 ? value : null; +}; + +export const postFailedJobAlert = ( + url: string, + summary: { + readonly queue: string; + readonly recorded: number; + readonly kinds: ReadonlyArray; + readonly orgIds: ReadonlyArray; + } +) => + Effect.tryPromise(() => + fetch(url, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ + text: `adrive: ${summary.recorded} job(s) dead-lettered on ${summary.queue}`, + ...summary + }) + }) + ).pipe( + Effect.asVoid, + Effect.catchCause((cause) => + Effect.sync(() => { + console.error( + JSON.stringify({ + message: 'failed job alert could not be delivered', + cause: String(cause) + }) + ); + }) + ) + ); diff --git a/apps/web/src/lib/server/jobs/batch-request.ts b/apps/web/src/lib/server/jobs/batch-request.ts new file mode 100644 index 0000000..567dce9 --- /dev/null +++ b/apps/web/src/lib/server/jobs/batch-request.ts @@ -0,0 +1,59 @@ +import { Effect, Schema } from 'effect'; +import { AppConfig } from '../config'; +import { verifyJobsRequest } from '../cron-auth'; +import { InvalidRequest, Unauthorized } from '../errors'; +import { readBoundedText } from '../request-json'; + +// The batch JSON the Worker facade posts for either queue: verified +// against the HMAC over the exact body, then decoded. +const JobBatchBody = Schema.Struct({ + queue: Schema.String, + messages: Schema.Array( + Schema.Struct({ + id: Schema.String, + attempts: Schema.Int, + body: Schema.Unknown + }) + ) +}); + +const MAX_BATCH_BYTES = 1024 * 1024; + +const invalid = (status: 400 | 413, message: string) => + new InvalidRequest({ status, message }); + +const unauthorized = () => + new Unauthorized({ message: 'Queue request is unauthorized' }); + +export const readSignedBatch = (request: Request) => + Effect.gen(function* () { + const config = yield* AppConfig; + const text = yield* readBoundedText(request, { + maxBytes: MAX_BATCH_BYTES, + invalidLengthMessage: 'Queue batch is too large', + invalidTextMessage: 'Queue batch is invalid' + }); + const authorized = yield* Effect.tryPromise({ + try: () => + verifyJobsRequest( + config.maintenanceSecret, + request.headers.get('x-adrive-jobs-time'), + text, + request.headers.get('x-adrive-jobs-signature') + ), + catch: unauthorized + }); + if (!authorized) return yield* unauthorized(); + + return yield* Effect.try({ + try: (): unknown => JSON.parse(text), + catch: () => invalid(400, 'Queue batch is invalid') + }).pipe( + Effect.flatMap(Schema.decodeUnknownEffect(JobBatchBody)), + Effect.mapError((cause) => + cause instanceof InvalidRequest + ? cause + : invalid(400, 'Queue batch is invalid') + ) + ); + }); diff --git a/apps/web/src/lib/server/jobs/dead-letters.ts b/apps/web/src/lib/server/jobs/dead-letters.ts new file mode 100644 index 0000000..45002dd --- /dev/null +++ b/apps/web/src/lib/server/jobs/dead-letters.ts @@ -0,0 +1,55 @@ +import { Effect } from 'effect'; +import { + alertWebhookUrl, + postFailedJobAlert, + recordFailedJob +} from '../failed-jobs'; +import { PgSql } from '../pg'; +import { ack, type JobBatch, type JobDecision } from './consumer'; + +// The dead-letter consumer. Every message is recorded and acked; the +// error text is what the platform tells us, which is only that the main +// consumer's retry budget ran out. Any earlier cause is on the row the +// job was working on (index_error, purge_error) and in the logs. +export const consumeDeadLetters = (env: Env, batch: JobBatch) => + Effect.gen(function* () { + const sql = yield* PgSql; + const kinds = new Set(); + const orgIds = new Set(); + let recorded = 0; + const decisions: Array = []; + for (const message of batch.messages) { + const result = yield* recordFailedJob(sql, { + id: message.id, + body: message.body, + attempts: message.attempts, + error: `Retries exhausted on ${batch.queue}` + }); + if (result.recorded) { + recorded += 1; + kinds.add(result.kind); + if (result.orgId !== null) orgIds.add(result.orgId); + } + decisions.push({ id: message.id, ...ack }); + } + console.error( + JSON.stringify({ + message: 'jobs dead-lettered', + queue: batch.queue, + received: batch.messages.length, + recorded, + kinds: [...kinds], + orgIds: [...orgIds] + }) + ); + const webhook = alertWebhookUrl(env); + if (webhook !== null && recorded > 0) { + yield* postFailedJobAlert(webhook, { + queue: batch.queue, + recorded, + kinds: [...kinds], + orgIds: [...orgIds] + }); + } + return decisions; + }); diff --git a/apps/web/src/routes/api/admin/failed-jobs/+server.ts b/apps/web/src/routes/api/admin/failed-jobs/+server.ts new file mode 100644 index 0000000..3b7eabf --- /dev/null +++ b/apps/web/src/routes/api/admin/failed-jobs/+server.ts @@ -0,0 +1,25 @@ +import type { RequestHandler } from './$types'; +import { Effect } from 'effect'; +import { runEdge } from '$lib/server/edge'; +import { InvalidRequest } from '$lib/server/errors'; +import { listFailedJobs } from '$lib/server/failed-jobs'; +import { PgSql } from '$lib/server/pg'; +import { requireAuth } from '$lib/server/request-auth'; + +// The org's dead-lettered jobs, for its owners. Read scope is enough; +// nothing here changes state. +export const GET: RequestHandler = (event) => + runEdge( + Effect.gen(function* () { + const sql = yield* PgSql; + const auth = yield* requireAuth(event); + if (auth.role !== 'owner') { + return yield* new InvalidRequest({ + status: 403, + message: 'Only an owner can view failed jobs' + }); + } + const jobs = yield* listFailedJobs(sql, auth.orgId); + return Response.json({ jobs }); + }) + ); diff --git a/apps/web/src/routes/api/internal/jobs/+server.ts b/apps/web/src/routes/api/internal/jobs/+server.ts index 2fdac5e..91826e7 100644 --- a/apps/web/src/routes/api/internal/jobs/+server.ts +++ b/apps/web/src/routes/api/internal/jobs/+server.ts @@ -1,11 +1,9 @@ import type { RequestHandler } from './$types'; -import { Effect, Schema } from 'effect'; -import { AppConfig } from '$lib/server/config'; -import { verifyJobsRequest } from '$lib/server/cron-auth'; +import { Effect } from 'effect'; import { runEdge } from '$lib/server/edge'; -import { InvalidRequest, Unauthorized } from '$lib/server/errors'; +import { InvalidRequest } from '$lib/server/errors'; +import { readSignedBatch } from '$lib/server/jobs/batch-request'; import { consumeBatch, runJobForOrg } from '$lib/server/jobs/consumer'; -import { readBoundedText } from '$lib/server/request-json'; // The Worker facade (scripts/cloudflare-adapter.mjs) receives queue // batches and forwards them here in-process, because the consumer has to @@ -13,26 +11,9 @@ import { readBoundedText } from '$lib/server/request-json'; // facade which messages to ack and which to retry (and after how long). // The request itself carries no tenant; every job builds its own layer // for the org named in its body. -const JobBatchBody = Schema.Struct({ - queue: Schema.String, - messages: Schema.Array( - Schema.Struct({ - id: Schema.String, - attempts: Schema.Int, - body: Schema.Unknown - }) - ) -}); - -const MAX_BATCH_BYTES = 1024 * 1024; - -const unauthorized = () => - new Unauthorized({ message: 'Queue request is unauthorized' }); - export const POST: RequestHandler = ({ request, platform }) => runEdge( Effect.gen(function* () { - const config = yield* AppConfig; const env = platform?.env; if (!env) { return yield* new InvalidRequest({ @@ -40,38 +21,7 @@ export const POST: RequestHandler = ({ request, platform }) => message: 'Cloudflare bindings unavailable' }); } - const text = yield* readBoundedText(request, { - maxBytes: MAX_BATCH_BYTES, - invalidLengthMessage: 'Queue batch is too large', - invalidTextMessage: 'Queue batch is invalid' - }); - const authorized = yield* Effect.tryPromise({ - try: () => - verifyJobsRequest( - config.maintenanceSecret, - request.headers.get('x-adrive-jobs-time'), - text, - request.headers.get('x-adrive-jobs-signature') - ), - catch: unauthorized - }); - if (!authorized) return yield* unauthorized(); - - const batch = yield* Effect.try({ - try: (): unknown => JSON.parse(text), - catch: () => - new InvalidRequest({ status: 400, message: 'Queue batch is invalid' }) - }).pipe( - Effect.flatMap(Schema.decodeUnknownEffect(JobBatchBody)), - Effect.mapError((cause) => - cause instanceof InvalidRequest - ? cause - : new InvalidRequest({ - status: 400, - message: 'Queue batch is invalid' - }) - ) - ); + const batch = yield* readSignedBatch(request); const decisions = yield* consumeBatch(batch, runJobForOrg(env)); return Response.json({ decisions }); }) diff --git a/apps/web/src/routes/api/internal/jobs/dead/+server.ts b/apps/web/src/routes/api/internal/jobs/dead/+server.ts new file mode 100644 index 0000000..7cc8ba6 --- /dev/null +++ b/apps/web/src/routes/api/internal/jobs/dead/+server.ts @@ -0,0 +1,24 @@ +import type { RequestHandler } from './$types'; +import { Effect } from 'effect'; +import { runEdge } from '$lib/server/edge'; +import { InvalidRequest } from '$lib/server/errors'; +import { readSignedBatch } from '$lib/server/jobs/batch-request'; +import { consumeDeadLetters } from '$lib/server/jobs/dead-letters'; + +// Dead-letter queue deliveries from the Worker facade: each message is +// written to failed_jobs and acked. +export const POST: RequestHandler = ({ request, platform }) => + runEdge( + Effect.gen(function* () { + const env = platform?.env; + if (!env) { + return yield* new InvalidRequest({ + status: 400, + message: 'Cloudflare bindings unavailable' + }); + } + const batch = yield* readSignedBatch(request); + const decisions = yield* consumeDeadLetters(env, batch); + return Response.json({ decisions }); + }) + ); diff --git a/apps/web/wrangler.jsonc b/apps/web/wrangler.jsonc index 34a4c8e..ba58606 100644 --- a/apps/web/wrangler.jsonc +++ b/apps/web/wrangler.jsonc @@ -57,6 +57,14 @@ "max_retries": 5, "max_concurrency": 10, "dead_letter_queue": "adrive-jobs-dlq" + }, + // Dead letters land in failed_jobs (see /api/internal/jobs/dead). + { + "queue": "adrive-jobs-dlq", + "max_batch_size": 10, + "max_batch_timeout": 30, + "max_retries": 3, + "max_concurrency": 1 } ] }, @@ -154,6 +162,13 @@ "max_retries": 5, "max_concurrency": 10, "dead_letter_queue": "adrive-jobs-production-dlq" + }, + { + "queue": "adrive-jobs-production-dlq", + "max_batch_size": 10, + "max_batch_timeout": 30, + "max_retries": 3, + "max_concurrency": 1 } ] }, diff --git a/docs/release.md b/docs/release.md index 8132c00..3114de0 100644 --- a/docs/release.md +++ b/docs/release.md @@ -167,10 +167,13 @@ wrangler queues create adrive-jobs-production-dlq - The consumer retries a failed message up to `max_retries` (5) times, then moves it to `adrive-jobs-production-dlq`. Messages whose body does not decode as a job are acked and logged, never retried. -- Inspect the dead-letter queue with - `wrangler queues consumer` tooling or the dashboard; nothing drains it - automatically. Re-send a message from the DLQ only after fixing the - cause, since the same job will otherwise fail again. +- The Worker also consumes the dead-letter queue: each message is + written to the `failed_jobs` table (org, kind, payload, error, + attempts) and acked. Owners see their org's rows at + `GET /api/admin/failed-jobs`. Set the optional `ALERT_WEBHOOK_URL` + secret to have a JSON summary POSTed whenever a batch dead-letters. + Re-send a job only after fixing the cause, since it will otherwise + fail again. - Local development uses the `adrive-jobs` / `adrive-jobs-dlq` names and needs no provisioning; `wrangler dev` simulates the queue. diff --git a/packages/shared/src/index.ts b/packages/shared/src/index.ts index b9d9e6c..f626d0f 100644 --- a/packages/shared/src/index.ts +++ b/packages/shared/src/index.ts @@ -415,3 +415,22 @@ export const JobSchema = Schema.Union([ ]); export type Job = typeof JobSchema.Type; + +// A job the queue gave up on, as listed for an org's owner. +export const FailedJobSchema = Schema.Struct({ + id: Schema.String, + kind: Schema.String, + payload: Schema.Unknown, + error: Schema.String, + attempts: Schema.Int, + failedAt: Schema.String, + resolvedAt: Schema.NullOr(Schema.String) +}); + +export type FailedJob = typeof FailedJobSchema.Type; + +export const FailedJobListResponseSchema = Schema.Struct({ + jobs: Schema.Array(FailedJobSchema) +}); + +export type FailedJobListResponse = typeof FailedJobListResponseSchema.Type; From cd84f2e4f68f29fc13cc7c877ebba292001d4f85 Mon Sep 17 00:00:00 2001 From: Ben Davis <45952064+bmdavis419@users.noreply.github.com> Date: Wed, 9 Sep 2026 06:50:18 -0700 Subject: [PATCH 4/9] FIX: Acknowledge jobs whose org no longer exists After the subdomain work the per-job layer resolves the org slug from Postgres. A missing org used to surface as a StorageError, which the consumer retried until the message dead-lettered. It is now a typed OrgMissing error that the consumer logs and acks. Co-Authored-By: Claude Fable 5.1 --- apps/web/src/lib/server/errors.ts | 6 ++++++ apps/web/src/lib/server/jobs/consumer.ts | 9 +++++++++ apps/web/src/lib/server/layer.ts | 7 ++++--- 3 files changed, 19 insertions(+), 3 deletions(-) diff --git a/apps/web/src/lib/server/errors.ts b/apps/web/src/lib/server/errors.ts index cdc0594..4b3c9b2 100644 --- a/apps/web/src/lib/server/errors.ts +++ b/apps/web/src/lib/server/errors.ts @@ -22,6 +22,12 @@ export class StorageError extends Data.TaggedError('StorageError')<{ readonly cause: unknown; }> {} +// A program was asked to run for an org that no longer exists. Background +// jobs treat this as done: there is nothing left to act on. +export class OrgMissing extends Data.TaggedError('OrgMissing')<{ + readonly orgId: string; +}> {} + export type AppError = InvalidRequest | MisdirectedRequest | Unauthorized | NotFound | StorageError; diff --git a/apps/web/src/lib/server/jobs/consumer.ts b/apps/web/src/lib/server/jobs/consumer.ts index 527d6d2..52b8c17 100644 --- a/apps/web/src/lib/server/jobs/consumer.ts +++ b/apps/web/src/lib/server/jobs/consumer.ts @@ -191,6 +191,15 @@ export const runJobForOrg = (env: Env) => (job: Job) => }), Effect.catchTag('SqlError', (cause) => Effect.fail(new StorageError({ operation: 'connect for job', cause })) + ), + // The org was deleted after the job was sent. Nothing to do, and + // retrying would only dead-letter it. + Effect.catchTag('OrgMissing', (missing) => + log({ + message: 'job skipped for a missing org', + kind: job.kind, + orgId: missing.orgId + }).pipe(Effect.as('done' as const)) ) ); diff --git a/apps/web/src/lib/server/layer.ts b/apps/web/src/lib/server/layer.ts index a0d7785..0b93e4c 100644 --- a/apps/web/src/lib/server/layer.ts +++ b/apps/web/src/lib/server/layer.ts @@ -1,6 +1,6 @@ import { Effect, Layer } from 'effect'; import { ConfigLive } from './config'; -import { StorageError } from './errors'; +import { OrgMissing, StorageError } from './errors'; import type { ProgramTenant } from './identity'; import { PgSql } from './pg'; import { AuthGuardLive } from './services/auth-guard'; @@ -53,10 +53,11 @@ const currentOrgLayer = ( SELECT slug FROM orgs WHERE id = ${orgId} LIMIT 1 `; const row = rows[0]; - if (!row) return yield* Effect.fail(`Org ${orgId} does not exist`); + if (!row) return yield* new OrgMissing({ orgId }); return { id: orgId, slug: row.slug }; }).pipe( - Effect.mapError( + Effect.catchTag( + 'SqlError', (cause) => new StorageError({ operation: 'resolve org slug', cause }) ) ) From 40219b2da240ea9b23d34fc16ce314c0e6a0b901 Mon Sep 17 00:00:00 2001 From: Ben Davis <45952064+bmdavis419@users.noreply.github.com> Date: Thu, 10 Sep 2026 23:24:03 -0700 Subject: [PATCH 5/9] FIX: Preserve worker leases and recover storage cleanup safely --- ...005_failed_jobs.sql => 0006_failed_jobs.sql} | 0 apps/web/src/lib/server/failed-jobs.ts | 17 ++++++++++++----- .../src/lib/server/services/files/mutations.ts | 9 +++++++-- apps/web/src/lib/server/services/files/purge.ts | 1 + apps/web/src/lib/server/services/indexing.ts | 10 ++++++---- .../src/lib/server/services/sites/sessions.ts | 11 +++++++---- apps/web/src/lib/server/services/sites/types.ts | 3 ++- 7 files changed, 35 insertions(+), 16 deletions(-) rename apps/web/migrations-pg/{0005_failed_jobs.sql => 0006_failed_jobs.sql} (100%) diff --git a/apps/web/migrations-pg/0005_failed_jobs.sql b/apps/web/migrations-pg/0006_failed_jobs.sql similarity index 100% rename from apps/web/migrations-pg/0005_failed_jobs.sql rename to apps/web/migrations-pg/0006_failed_jobs.sql diff --git a/apps/web/src/lib/server/failed-jobs.ts b/apps/web/src/lib/server/failed-jobs.ts index 4bd38a7..195fd39 100644 --- a/apps/web/src/lib/server/failed-jobs.ts +++ b/apps/web/src/lib/server/failed-jobs.ts @@ -96,17 +96,24 @@ export const postFailedJobAlert = ( readonly orgIds: ReadonlyArray; } ) => - Effect.tryPromise(() => - fetch(url, { + Effect.tryPromise(async () => { + const response = await fetch(url, { method: 'POST', + signal: AbortSignal.timeout(5_000), headers: { 'content-type': 'application/json' }, body: JSON.stringify({ text: `adrive: ${summary.recorded} job(s) dead-lettered on ${summary.queue}`, ...summary }) - }) - ).pipe( - Effect.asVoid, + }); + await response.body?.cancel(); + return response; + }).pipe( + Effect.flatMap((response) => + response.ok + ? Effect.void + : Effect.fail(`Alert webhook returned HTTP ${response.status}`) + ), Effect.catchCause((cause) => Effect.sync(() => { console.error( diff --git a/apps/web/src/lib/server/services/files/mutations.ts b/apps/web/src/lib/server/services/files/mutations.ts index 226a0a2..e6276cd 100644 --- a/apps/web/src/lib/server/services/files/mutations.ts +++ b/apps/web/src/lib/server/services/files/mutations.ts @@ -87,12 +87,17 @@ export const mutationOps = ( restore: Effect.fn('Files.restore')(function* (id) { const current = yield* findDashboardFile(id); const updatedAt = new Date().toISOString(); + // A failed purge may already have deleted originals before a later + // blob or thumbnail operation failed. Once deletion has started, + // keep the row in trash until cleanup finishes; resetting its state + // through another trash action must not make it restorable either. const rows = yield* sql<{ id: string }>` UPDATE files SET deleted_at = NULL, purge_at = NULL, purge_state = 'none', purge_error = NULL, purge_next_run_at = NULL, updated_at = ${updatedAt} - WHERE id = ${id} AND org_id = ${org.id} AND purge_state <> 'pending' + WHERE id = ${id} AND org_id = ${org.id} + AND purge_state = 'none' AND purge_attempts = 0 RETURNING id `.pipe( Effect.mapError( @@ -102,7 +107,7 @@ export const mutationOps = ( if (rows.length !== 1) { return yield* new InvalidRequest({ status: 409, - message: 'This file is already being purged' + message: 'This file has already started permanent deletion' }); } return { diff --git a/apps/web/src/lib/server/services/files/purge.ts b/apps/web/src/lib/server/services/files/purge.ts index 72b5e0e..fcf170d 100644 --- a/apps/web/src/lib/server/services/files/purge.ts +++ b/apps/web/src/lib/server/services/files/purge.ts @@ -186,6 +186,7 @@ export const purgeOps = ( ) ORDER BY COALESCE(purge_at, expires_at), id LIMIT ${bounded} + FOR UPDATE SKIP LOCKED ) AND org_id = ${org.id} RETURNING id`.pipe(Effect.mapError(storage('list stuck file purges'))); for (const row of rows) yield* sendPurgeJob(row.id, now.toISOString()); diff --git a/apps/web/src/lib/server/services/indexing.ts b/apps/web/src/lib/server/services/indexing.ts index 9bab6ad..e30ab2a 100644 --- a/apps/web/src/lib/server/services/indexing.ts +++ b/apps/web/src/lib/server/services/indexing.ts @@ -302,7 +302,8 @@ const makeIndexing = Effect.gen(function* () { // send); they get a fresh job. Stamping index_next_run_at throttles // the re-send to once per stuck window and keeps a lapsed `running` // lease claimable. Disabled rows are re-sent once semantic search is - // switched on so they gain embeddings. + // switched on so they gain embeddings. Lock candidates before updating + // them; a consumer claiming a row concurrently keeps its fresh lease. const runDue = Effect.fn('Indexing.runDue')(function* (limit: number) { const bounded = Math.max(1, Math.min(limit, 10)); const now = new Date(); @@ -324,9 +325,10 @@ const makeIndexing = Effect.gen(function* () { OR (index_state = 'disabled' AND ${includeDisabled}::boolean AND COALESCE(index_next_run_at, updated_at) <= ${cutoff}) ) - ORDER BY COALESCE(index_next_run_at, updated_at), id - LIMIT ${bounded} - ) AND org_id = ${org.id} + ORDER BY COALESCE(index_next_run_at, updated_at), id + LIMIT ${bounded} + FOR UPDATE SKIP LOCKED + ) AND org_id = ${org.id} RETURNING id, current_version`.pipe( Effect.mapError(storage('list stuck indexing jobs')) ); diff --git a/apps/web/src/lib/server/services/sites/sessions.ts b/apps/web/src/lib/server/services/sites/sessions.ts index 25957ec..e4c8d41 100644 --- a/apps/web/src/lib/server/services/sites/sessions.ts +++ b/apps/web/src/lib/server/services/sites/sessions.ts @@ -43,21 +43,19 @@ export const sessionOps = ( message: 'Site manifest is invalid' }) }); - // Declared manifest sizes gate the whole publish before any asset - // bytes are accepted; per-asset uploads re-verify actual lengths. const declaredBytes = prepared.assets.reduce( (total, asset) => total + asset.sizeBytes, 0 ); - yield* ensureStorageHeadroom(sql, org.id, declaredBytes); let fileId: string = crypto.randomUUID(); let version = 1; let displayName = prepared.displayName; + let previousBytes = 0; if (input.fileId !== undefined) { const rows = yield* all( sql` - SELECT id, display_name, current_version + SELECT id, display_name, current_version, size_bytes FROM files WHERE id = ${input.fileId} AND org_id = ${org.id} AND is_site = true AND deleted_at IS NULL @@ -70,7 +68,12 @@ export const sessionOps = ( fileId = current.id; version = current.current_version + 1; displayName = current.display_name; + previousBytes = current.size_bytes; } + // A republish replaces the previous assets, so preflight the same + // byte delta the commit will charge. Actual asset lengths and the + // authoritative reservation are still checked during publication. + yield* ensureStorageHeadroom(sql, org.id, declaredBytes - previousBytes); const id = crypto.randomUUID(); const createdAt = new Date().toISOString(); diff --git a/apps/web/src/lib/server/services/sites/types.ts b/apps/web/src/lib/server/services/sites/types.ts index 34c7137..dc11c4f 100644 --- a/apps/web/src/lib/server/services/sites/types.ts +++ b/apps/web/src/lib/server/services/sites/types.ts @@ -23,7 +23,8 @@ export const StagedAssetRow = Schema.Struct({ export const ExistingSiteRow = Schema.Struct({ id: Schema.String, display_name: Schema.String, - current_version: Schema.Int + current_version: Schema.Int, + size_bytes: Schema.Int }); export const SiteFileRow = Schema.Struct({ From a04b67fed3e66294ec6ea82dc5034efe9a158728 Mon Sep 17 00:00:00 2001 From: Ben Davis <45952064+bmdavis419@users.noreply.github.com> Date: Thu, 10 Sep 2026 23:41:25 -0700 Subject: [PATCH 6/9] FIX: Close indexing recovery and rename queue races --- .../lib/server/services/files/mutations.ts | 13 ++++++++---- apps/web/src/lib/server/services/indexing.ts | 21 +++++++++++++++++++ 2 files changed, 30 insertions(+), 4 deletions(-) diff --git a/apps/web/src/lib/server/services/files/mutations.ts b/apps/web/src/lib/server/services/files/mutations.ts index e6276cd..2d883c8 100644 --- a/apps/web/src/lib/server/services/files/mutations.ts +++ b/apps/web/src/lib/server/services/files/mutations.ts @@ -152,23 +152,28 @@ export const mutationOps = ( current.public ); const updatedAt = new Date().toISOString(); - yield* sql + const rows = yield* sql .withTransaction( - sql` + sql<{ current_version: number }>` UPDATE files SET display_name = ${displayName}, public = ${visibility.public}, updated_at = ${updatedAt}, index_state = 'pending', index_cursor = 0, index_attempts = 0, index_error = NULL, index_next_run_at = NULL, index_lease_token = NULL WHERE id = ${id} AND org_id = ${org.id} - `.pipe(Effect.andThen(refreshSearchDocument(sql, id, org.id))) + RETURNING current_version + `.pipe(Effect.tap(() => refreshSearchDocument(sql, id, org.id))) ) .pipe( Effect.mapError( (cause) => new StorageError({ operation: 'rename file', cause }) ) ); - yield* sendIndexJob(id, current.version); + const renamed = rows[0]; + if (!renamed) return yield* new NotFound({ id }); + // A version upload may have committed after the initial dashboard + // read. Index the version whose state this rename actually reset. + yield* sendIndexJob(id, renamed.current_version); return { file: { ...current, diff --git a/apps/web/src/lib/server/services/indexing.ts b/apps/web/src/lib/server/services/indexing.ts index e30ab2a..7a48c05 100644 --- a/apps/web/src/lib/server/services/indexing.ts +++ b/apps/web/src/lib/server/services/indexing.ts @@ -309,6 +309,27 @@ const makeIndexing = Effect.gen(function* () { const now = new Date(); const cutoff = stuckBefore(now.getTime()); const includeDisabled = embedder.enabled && vectors.enabled; + // A worker can disappear during its final attempt, before recording + // success or failure. It cannot claim again at the attempt limit, so + // close the expired lease instead of leaving it running forever. + yield* sql` + UPDATE files + SET index_state = 'failed', + index_error = 'Indexing lease expired after the final attempt', + index_next_run_at = NULL, index_lease_token = NULL + WHERE id IN ( + SELECT id FROM files + WHERE org_id = ${org.id} + AND deleted_at IS NULL + AND (expires_at IS NULL OR expires_at > ${now.toISOString()}) + AND index_state = 'running' + AND index_attempts >= ${MAX_INDEX_ATTEMPTS} + AND index_next_run_at <= ${cutoff} + ORDER BY index_next_run_at, id + LIMIT ${bounded} + FOR UPDATE SKIP LOCKED + ) AND org_id = ${org.id} + `.pipe(Effect.mapError(storage('close exhausted indexing leases'))); const rows = yield* sql<{ id: string; current_version: number }>` UPDATE files SET index_next_run_at = ${now.toISOString()} From 3960c0b3edf9f61efb56c6268ad31e8ea62fe85a Mon Sep 17 00:00:00 2001 From: Ben Davis <45952064+bmdavis419@users.noreply.github.com> Date: Thu, 10 Sep 2026 23:53:25 -0700 Subject: [PATCH 7/9] Bound empty-trash sends and retain failed dead letters --- .../lib/server/services/files/mutations.ts | 32 +++++-- apps/web/wrangler.jsonc | 6 +- docs/release.md | 92 +++++++++++-------- 3 files changed, 81 insertions(+), 49 deletions(-) diff --git a/apps/web/src/lib/server/services/files/mutations.ts b/apps/web/src/lib/server/services/files/mutations.ts index 2d883c8..afb37fd 100644 --- a/apps/web/src/lib/server/services/files/mutations.ts +++ b/apps/web/src/lib/server/services/files/mutations.ts @@ -222,21 +222,35 @@ export const mutationOps = ( // A generator body reads the org when the effect runs, not when the // layer is built (content routes build the layer with no tenant). scheduleAllPurgesNow: Effect.gen(function* () { - const rows = yield* sql<{ id: string }>` - UPDATE files - SET purge_at = ${EPOCH}, purge_state = 'none', purge_error = NULL, - purge_next_run_at = NULL - WHERE org_id = ${org.id} AND deleted_at IS NOT NULL - AND purge_state <> 'pending' - RETURNING id + const rows = yield* sql<{ count: number; ids: string[] }>` + WITH scheduled AS ( + UPDATE files + SET purge_at = ${EPOCH}, purge_state = 'none', purge_error = NULL, + purge_next_run_at = NULL + WHERE org_id = ${org.id} AND deleted_at IS NOT NULL + AND purge_state <> 'pending' + RETURNING id + ) + SELECT count(*)::int AS count, + ARRAY(SELECT id FROM scheduled LIMIT 20) AS ids + FROM scheduled `.pipe( Effect.mapError( (cause) => new StorageError({ operation: 'schedule empty trash', cause }) ) ); - for (const row of rows) yield* sendPurgeJob(row.id, EPOCH); - return rows.length; + // All work is durable in Postgres. Kick a small batch now; the + // bounded reconciliation sweep picks up the rest or a slow send. + yield* Effect.forEach( + rows[0]?.ids ?? [], + (id) => sendPurgeJob(id, EPOCH), + { + concurrency: 5, + discard: true + } + ).pipe(Effect.timeoutOption('5 seconds')); + return rows[0]?.count ?? 0; }).pipe(Effect.withSpan('Files.scheduleAllPurgesNow')), recordDownload: Effect.fn('Files.recordDownload')(function* (id) { const now = new Date().toISOString(); diff --git a/apps/web/wrangler.jsonc b/apps/web/wrangler.jsonc index ba58606..25d2fda 100644 --- a/apps/web/wrangler.jsonc +++ b/apps/web/wrangler.jsonc @@ -64,7 +64,8 @@ "max_batch_size": 10, "max_batch_timeout": 30, "max_retries": 3, - "max_concurrency": 1 + "max_concurrency": 1, + "dead_letter_queue": "adrive-jobs-parked" } ] }, @@ -168,7 +169,8 @@ "max_batch_size": 10, "max_batch_timeout": 30, "max_retries": 3, - "max_concurrency": 1 + "max_concurrency": 1, + "dead_letter_queue": "adrive-jobs-production-parked" } ] }, diff --git a/docs/release.md b/docs/release.md index 3114de0..7ad7cfd 100644 --- a/docs/release.md +++ b/docs/release.md @@ -62,58 +62,62 @@ the repository root. loudly if the `AI` binding is missing. Embeddings run within the Workers Paid plan's included neuron allocation at personal scale. -5. From `apps/web`: create the job queue and its dead-letter queue: +5. From `apps/web`: create the job, dead-letter, and parked queues: ``` wrangler queues create adrive-jobs-production wrangler queues create adrive-jobs-production-dlq ``` +wrangler queues create adrive-jobs-production-parked --message-retention-period-secs 1209600 + +```` + 6. In the Cloudflare dashboard, open **Images → Transformations**, select - the zone that owns `CONTENT_DOMAIN` (`davis7.space` for - `files.davis7.space`), and enable transformations. Dashboard thumbnails - require this zone-level setting. +the zone that owns `CONTENT_DOMAIN` (`davis7.space` for +`files.davis7.space`), and enable transformations. Dashboard thumbnails +require this zone-level setting. 7. From `apps/web`, set each secret with `wrangler secret put --env production`: - `MAINTENANCE_SECRET` (12+ characters), `WORKOS_API_KEY`, `WORKOS_CLIENT_ID`, - `WORKOS_COOKIE_PASSWORD` (32+ characters), `WORKOS_WEBHOOK_SECRET`. Point the - WorkOS redirect URI at `/auth/callback` and the webhook at - `/api/webhooks/workos` (events `user.deleted`, - `organization_membership.deleted`). +`MAINTENANCE_SECRET` (12+ characters), `WORKOS_API_KEY`, `WORKOS_CLIENT_ID`, +`WORKOS_COOKIE_PASSWORD` (32+ characters), `WORKOS_WEBHOOK_SECRET`. Point the +WorkOS redirect URI at `/auth/callback` and the webhook at +`/api/webhooks/workos` (events `user.deleted`, +`organization_membership.deleted`). 8. Activate the zones for the hosted target's dashboard and content domains - in Cloudflare. The `custom_domain` routes create DNS records for their - exact hostnames; they do not create the tenant wildcard. Create a proxied - `AAAA *. 100::` record and configure its matching Worker - route. With the checked-in example, those are `*.files.davis7.space` and - `*.files.davis7.space/*` in the `davis7.space` zone. - - Confirm the edge certificate covers `*.`. Universal SSL - covers the zone apex and first-level hosts, so its `*.davis7.space` - certificate does not cover `example.files.davis7.space`. Provision - explicit coverage for `*.files.davis7.space` (for example, through - Advanced Certificate Manager), or use a separate content-domain zone's - apex as `CONTENT_DOMAIN`. Update the domain and Worker routes together. - Verify DNS and HTTPS for an actual tenant hostname before cutover. - - Use temporary origins for the separate hosted target described above; - preserve existing production records until cutover. Resolve conflicting - records when assigning the intended custom domains. The checked-in - `adrive.davis7.space` custom domain serves the static landing page - (`apps/site`, an assets-only Worker with no build step). +in Cloudflare. The `custom_domain` routes create DNS records for their +exact hostnames; they do not create the tenant wildcard. Create a proxied +`AAAA *. 100::` record and configure its matching Worker +route. With the checked-in example, those are `*.files.davis7.space` and +`*.files.davis7.space/*` in the `davis7.space` zone. + +Confirm the edge certificate covers `*.`. Universal SSL +covers the zone apex and first-level hosts, so its `*.davis7.space` +certificate does not cover `example.files.davis7.space`. Provision +explicit coverage for `*.files.davis7.space` (for example, through +Advanced Certificate Manager), or use a separate content-domain zone's +apex as `CONTENT_DOMAIN`. Update the domain and Worker routes together. +Verify DNS and HTTPS for an actual tenant hostname before cutover. + +Use temporary origins for the separate hosted target described above; +preserve existing production records until cutover. Resolve conflicting +records when assigning the intended custom domains. The checked-in +`adrive.davis7.space` custom domain serves the static landing page +(`apps/site`, an assets-only Worker with no build step). 9. From the repo root: `bun release` 10. From the repo root: set up backups on your backup host - (`scripts/backup/install-backup-host.sh`) and complete the restore drill - in `docs/backup-restore.md`. + (`scripts/backup/install-backup-host.sh`) and complete the restore drill + in `docs/backup-restore.md`. Semantic search notes for the first deploy: - Files uploaded before the index existed (or while bindings were absent) - sit in `index_state = 'disabled'` and are backfilled by the maintenance - cron at 5 files per 5 minutes. A large pre-existing corpus takes hours; - the settings page's "indexed chunks" count shows progress. +sit in `index_state = 'disabled'` and are backfilled by the maintenance +cron at 5 files per 5 minutes. A large pre-existing corpus takes hours; +the settings page's "indexed chunks" count shows progress. - Embeddings live in Postgres beside the file rows, so they are restored - with the database. Files whose embeddings are missing after a partial - restore regenerate on reindex. +with the database. Files whose embeddings are missing after a partial +restore regenerate on reindex. ## Database roles @@ -131,9 +135,9 @@ tools or equivalent SQL: ```sql CREATE ROLE adrive_runtime LOGIN INHERIT - NOSUPERUSER NOCREATEDB NOCREATEROLE NOBYPASSRLS; +NOSUPERUSER NOCREATEDB NOCREATEROLE NOBYPASSRLS; GRANT adrive_app TO adrive_runtime; -``` +```` Set its password through the provider or `\password adrive_runtime` in `psql`. Connect using `ADRIVE_RUNTIME_DATABASE_URL` and verify: @@ -162,6 +166,7 @@ the Worker as its consumer; the queue itself is created once: ``` wrangler queues create adrive-jobs-production wrangler queues create adrive-jobs-production-dlq +wrangler queues create adrive-jobs-production-parked --message-retention-period-secs 1209600 ``` - The consumer retries a failed message up to `max_retries` (5) times, @@ -174,7 +179,18 @@ wrangler queues create adrive-jobs-production-dlq secret to have a JSON summary POSTed whenever a batch dead-letters. Re-send a job only after fixing the cause, since it will otherwise fail again. -- Local development uses the `adrive-jobs` / `adrive-jobs-dlq` names and +- If Postgres remains unavailable through the DLQ consumer's three retries, + Cloudflare moves the message to `adrive-jobs-production-parked`. This queue + deliberately has no automatic consumer, so an outage cannot exhaust another + retry chain. Its retention is 14 days, not indefinite: alert on nonzero backlog + and recover before expiry. Once Postgres is healthy, use the Queues HTTP pull + API to pull parked messages and republish their original bodies to + `adrive-jobs-production-dlq`; acknowledge parked messages only after successful + publication. The DLQ consumer records them in `failed_jobs` for review. +- Emptying trash records every requested deletion in Postgres but sends at most + 20 immediate queue messages within a five-second budget. Cron reconciliation + continues the rest in bounded batches. +- Local development uses the `adrive-jobs` / `adrive-jobs-dlq` / `adrive-jobs-parked` names and needs no provisioning; `wrangler dev` simulates the queue. ## Releasing From 5705cf2449692825650955234f0e03c20df77a06 Mon Sep 17 00:00:00 2001 From: Ben Davis <45952064+bmdavis419@users.noreply.github.com> Date: Fri, 11 Sep 2026 00:08:08 -0700 Subject: [PATCH 8/9] Fix queue provisioning documentation fences --- docs/release.md | 77 ++++++++++++++++++++++++------------------------- 1 file changed, 37 insertions(+), 40 deletions(-) diff --git a/docs/release.md b/docs/release.md index 7ad7cfd..e40b0f6 100644 --- a/docs/release.md +++ b/docs/release.md @@ -67,57 +67,54 @@ the repository root. ``` wrangler queues create adrive-jobs-production wrangler queues create adrive-jobs-production-dlq + wrangler queues create adrive-jobs-production-parked --message-retention-period-secs 1209600 ``` -wrangler queues create adrive-jobs-production-parked --message-retention-period-secs 1209600 - -```` - 6. In the Cloudflare dashboard, open **Images → Transformations**, select -the zone that owns `CONTENT_DOMAIN` (`davis7.space` for -`files.davis7.space`), and enable transformations. Dashboard thumbnails -require this zone-level setting. + the zone that owns `CONTENT_DOMAIN` (`davis7.space` for + `files.davis7.space`), and enable transformations. Dashboard thumbnails + require this zone-level setting. 7. From `apps/web`, set each secret with `wrangler secret put --env production`: -`MAINTENANCE_SECRET` (12+ characters), `WORKOS_API_KEY`, `WORKOS_CLIENT_ID`, -`WORKOS_COOKIE_PASSWORD` (32+ characters), `WORKOS_WEBHOOK_SECRET`. Point the -WorkOS redirect URI at `/auth/callback` and the webhook at -`/api/webhooks/workos` (events `user.deleted`, -`organization_membership.deleted`). + `MAINTENANCE_SECRET` (12+ characters), `WORKOS_API_KEY`, `WORKOS_CLIENT_ID`, + `WORKOS_COOKIE_PASSWORD` (32+ characters), `WORKOS_WEBHOOK_SECRET`. Point the + WorkOS redirect URI at `/auth/callback` and the webhook at + `/api/webhooks/workos` (events `user.deleted`, + `organization_membership.deleted`). 8. Activate the zones for the hosted target's dashboard and content domains -in Cloudflare. The `custom_domain` routes create DNS records for their -exact hostnames; they do not create the tenant wildcard. Create a proxied -`AAAA *. 100::` record and configure its matching Worker -route. With the checked-in example, those are `*.files.davis7.space` and -`*.files.davis7.space/*` in the `davis7.space` zone. - -Confirm the edge certificate covers `*.`. Universal SSL -covers the zone apex and first-level hosts, so its `*.davis7.space` -certificate does not cover `example.files.davis7.space`. Provision -explicit coverage for `*.files.davis7.space` (for example, through -Advanced Certificate Manager), or use a separate content-domain zone's -apex as `CONTENT_DOMAIN`. Update the domain and Worker routes together. -Verify DNS and HTTPS for an actual tenant hostname before cutover. - -Use temporary origins for the separate hosted target described above; -preserve existing production records until cutover. Resolve conflicting -records when assigning the intended custom domains. The checked-in -`adrive.davis7.space` custom domain serves the static landing page -(`apps/site`, an assets-only Worker with no build step). + in Cloudflare. The `custom_domain` routes create DNS records for their + exact hostnames; they do not create the tenant wildcard. Create a proxied + `AAAA *. 100::` record and configure its matching Worker + route. With the checked-in example, those are `*.files.davis7.space` and + `*.files.davis7.space/*` in the `davis7.space` zone. + + Confirm the edge certificate covers `*.`. Universal SSL + covers the zone apex and first-level hosts, so its `*.davis7.space` + certificate does not cover `example.files.davis7.space`. Provision + explicit coverage for `*.files.davis7.space` (for example, through + Advanced Certificate Manager), or use a separate content-domain zone's + apex as `CONTENT_DOMAIN`. Update the domain and Worker routes together. + Verify DNS and HTTPS for an actual tenant hostname before cutover. + + Use temporary origins for the separate hosted target described above; + preserve existing production records until cutover. Resolve conflicting + records when assigning the intended custom domains. The checked-in + `adrive.davis7.space` custom domain serves the static landing page + (`apps/site`, an assets-only Worker with no build step). 9. From the repo root: `bun release` 10. From the repo root: set up backups on your backup host - (`scripts/backup/install-backup-host.sh`) and complete the restore drill - in `docs/backup-restore.md`. + (`scripts/backup/install-backup-host.sh`) and complete the restore drill + in `docs/backup-restore.md`. Semantic search notes for the first deploy: - Files uploaded before the index existed (or while bindings were absent) -sit in `index_state = 'disabled'` and are backfilled by the maintenance -cron at 5 files per 5 minutes. A large pre-existing corpus takes hours; -the settings page's "indexed chunks" count shows progress. + sit in `index_state = 'disabled'` and are backfilled by the maintenance + cron at 5 files per 5 minutes. A large pre-existing corpus takes hours; + the settings page's "indexed chunks" count shows progress. - Embeddings live in Postgres beside the file rows, so they are restored -with the database. Files whose embeddings are missing after a partial -restore regenerate on reindex. + with the database. Files whose embeddings are missing after a partial + restore regenerate on reindex. ## Database roles @@ -135,9 +132,9 @@ tools or equivalent SQL: ```sql CREATE ROLE adrive_runtime LOGIN INHERIT -NOSUPERUSER NOCREATEDB NOCREATEROLE NOBYPASSRLS; + NOSUPERUSER NOCREATEDB NOCREATEROLE NOBYPASSRLS; GRANT adrive_app TO adrive_runtime; -```` +``` Set its password through the provider or `\password adrive_runtime` in `psql`. Connect using `ADRIVE_RUNTIME_DATABASE_URL` and verify: From b892e713494a5efdced947d8567e7362f3a39385 Mon Sep 17 00:00:00 2001 From: Ben Davis <45952064+bmdavis419@users.noreply.github.com> Date: Fri, 11 Sep 2026 01:42:54 -0700 Subject: [PATCH 9/9] Keep dead-letter batches within the signed request limit --- apps/web/wrangler.jsonc | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/apps/web/wrangler.jsonc b/apps/web/wrangler.jsonc index 25d2fda..69ce047 100644 --- a/apps/web/wrangler.jsonc +++ b/apps/web/wrangler.jsonc @@ -59,9 +59,11 @@ "dead_letter_queue": "adrive-jobs-dlq" }, // Dead letters land in failed_jobs (see /api/internal/jobs/dead). + // Five 128 KiB messages leave JSON-envelope headroom under the + // internal 1 MiB request cap. Keep production batches aligned. { "queue": "adrive-jobs-dlq", - "max_batch_size": 10, + "max_batch_size": 5, "max_batch_timeout": 30, "max_retries": 3, "max_concurrency": 1, @@ -166,7 +168,7 @@ }, { "queue": "adrive-jobs-production-dlq", - "max_batch_size": 10, + "max_batch_size": 5, "max_batch_timeout": 30, "max_retries": 3, "max_concurrency": 1,