diff --git a/apps/web/migrations-pg/0006_failed_jobs.sql b/apps/web/migrations-pg/0006_failed_jobs.sql new file mode 100644 index 0000000..d29a0eb --- /dev/null +++ b/apps/web/migrations-pg/0006_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 bf345a4..ad92141 100644 --- a/apps/web/scripts/cloudflare-adapter.mjs +++ b/apps/web/scripts/cloudflare-adapter.mjs @@ -67,10 +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 ack/retry decision per message id; anything it did not - // decide on is retried so a crash never silently drops work. + // 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, @@ -82,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', @@ -98,10 +104,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/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/failed-jobs.ts b/apps/web/src/lib/server/failed-jobs.ts new file mode 100644 index 0000000..195fd39 --- /dev/null +++ b/apps/web/src/lib/server/failed-jobs.ts @@ -0,0 +1,127 @@ +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(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 + }) + }); + 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( + JSON.stringify({ + message: 'failed job alert could not be delivered', + cause: String(cause) + }) + ); + }) + ) + ); 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/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/consumer.ts b/apps/web/src/lib/server/jobs/consumer.ts index d57487b..52b8c17 100644 --- a/apps/web/src/lib/server/jobs/consumer.ts +++ b/apps/web/src/lib/server/jobs/consumer.ts @@ -1,7 +1,12 @@ 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 { 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 @@ -17,11 +22,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 +69,60 @@ 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); } }; +// 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, + purge: (job) => files.purgeOne(job.fileId).pipe(Effect.as('done')), + siteCleanup: (job) => + sites.cleanupSession(job.sessionId).pipe(Effect.as('done')) + } 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 +133,75 @@ 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)); + } + 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' as const; + 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. `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' }), { + local: true + }), + 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)) ) ); 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/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/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 }) ) ) 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..86f5b5c 100644 --- a/apps/web/src/lib/server/services/files/internals.ts +++ b/apps/web/src/lib/server/services/files/internals.ts @@ -12,10 +12,12 @@ 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'; 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 +28,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 +76,18 @@ 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 }); + // 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 ) { @@ -144,6 +159,7 @@ export const createInternals = (deps: CoreDeps) => { ) ) ); + yield* sendIndexJob(current.id, version); return { file: { ...current, @@ -168,9 +184,12 @@ export const createInternals = (deps: CoreDeps) => { config, tags, org, + jobs, compensateStoredBlob, 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 bb4f697..afb37fd 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, 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 @@ -86,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( @@ -101,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 { @@ -122,6 +128,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 @@ -145,22 +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 }) ) ); + 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, @@ -203,27 +216,42 @@ 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 }>` - 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') - ), + scheduleAllPurgesNow: Effect.gen(function* () { + 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 }) + ) + ); + // 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(); // 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..fcf170d 100644 --- a/apps/web/src/lib/server/services/files/purge.ts +++ b/apps/web/src/lib/server/services/files/purge.ts @@ -1,133 +1,197 @@ 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} + 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()); + 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 4af2cd0..d8f5c63 100644 --- a/apps/web/src/lib/server/services/files/upload.ts +++ b/apps/web/src/lib/server/services/files/upload.ts @@ -21,7 +21,9 @@ export const uploadOps = ( reserveBytes, compensateStoredBlob, commitStoredVersion, - findDashboardFile + findDashboardFile, + sendIndexJob, + sendPurgeJob } = internals; return { upload: Effect.fn('Files.upload')(function* (input) { @@ -102,6 +104,8 @@ 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/indexing.ts b/apps/web/src/lib/server/services/indexing.ts index c33dd7f..7a48c05 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,89 @@ 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. 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().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); + // 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()} + 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} + FOR UPDATE SKIP LOCKED + ) 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 +389,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/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 38bfadb..ffb648d 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, @@ -219,35 +227,6 @@ export const createInternals = ({ sql, blobs, config, org }: CoreDeps) => { 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 ) { @@ -277,12 +256,12 @@ export const createInternals = ({ sql, blobs, config, org }: CoreDeps) => { compensateStagedBlob, drainDeletes, cleanupStaged, - sweepExpiredSessions, sweepPendingDeletes, 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..e4c8d41 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,27 +24,15 @@ export const sessionOps = ( compensateStagedBlob, cleanupStaged, drainDeletes, - sweepExpiredSessions, sql, blobs, config, - org + org, + jobs } = internals; 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) => @@ -54,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 @@ -81,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(); @@ -118,6 +110,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(() => { @@ -426,6 +423,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/lib/server/services/sites/types.ts b/apps/web/src/lib/server/services/sites/types.ts index 7a32d6a..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({ @@ -114,6 +115,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/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/files/+server.ts b/apps/web/src/routes/api/files/+server.ts index 7eb2831..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'; @@ -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,39 +182,21 @@ 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) => { - 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); @@ -228,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 98b3fe8..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,39 +101,7 @@ export const PATCH: RequestHandler = async (event) => { forcedPublic: false })) ); - return { - reindex: - mutation.action === 'reindex' || - mutation.action === 'rename' || - mutation.action === 'restore-version', - purge: mutation.action === 'purge', - response: Response.json(result) - }; + return 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( - event.platform.env, - Effect.gen(function* () { - const files = yield* Files; - yield* files.sweepPurges(1); - }), - event.locals.auth - ) - ); - } - return output.response; }; 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..91826e7 100644 --- a/apps/web/src/routes/api/internal/jobs/+server.ts +++ b/apps/web/src/routes/api/internal/jobs/+server.ts @@ -1,69 +1,28 @@ 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 { consumeBatch, runJob } from '$lib/server/jobs/consumer'; -import { readBoundedText } from '$lib/server/request-json'; +import { InvalidRequest } from '$lib/server/errors'; +import { readSignedBatch } from '$lib/server/jobs/batch-request'; +import { consumeBatch, runJobForOrg } from '$lib/server/jobs/consumer'; // 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. -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 }) => +// 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. +export const POST: RequestHandler = ({ request, platform }) => runEdge( 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(); - - 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 decisions = yield* consumeBatch(batch, runJob); + const env = platform?.env; + if (!env) { + return yield* new InvalidRequest({ + status: 400, + message: 'Cloudflare bindings unavailable' + }); + } + 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/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/apps/web/wrangler.jsonc b/apps/web/wrangler.jsonc index 34a4c8e..69ce047 100644 --- a/apps/web/wrangler.jsonc +++ b/apps/web/wrangler.jsonc @@ -57,6 +57,17 @@ "max_retries": 5, "max_concurrency": 10, "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": 5, + "max_batch_timeout": 30, + "max_retries": 3, + "max_concurrency": 1, + "dead_letter_queue": "adrive-jobs-parked" } ] }, @@ -154,6 +165,14 @@ "max_retries": 5, "max_concurrency": 10, "dead_letter_queue": "adrive-jobs-production-dlq" + }, + { + "queue": "adrive-jobs-production-dlq", + "max_batch_size": 5, + "max_batch_timeout": 30, + "max_retries": 3, + "max_concurrency": 1, + "dead_letter_queue": "adrive-jobs-production-parked" } ] }, diff --git a/docs/release.md b/docs/release.md index 8132c00..e40b0f6 100644 --- a/docs/release.md +++ b/docs/release.md @@ -62,11 +62,12 @@ 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 @@ -162,16 +163,31 @@ 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, 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. -- Local development uses the `adrive-jobs` / `adrive-jobs-dlq` names and +- 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. +- 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 diff --git a/packages/shared/src/index.ts b/packages/shared/src/index.ts index 970c12f..f626d0f 100644 --- a/packages/shared/src/index.ts +++ b/packages/shared/src/index.ts @@ -385,28 +385,52 @@ 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 }) ]); 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;