diff --git a/drizzle/0011_onchain_attempts.sql b/drizzle/0011_onchain_attempts.sql new file mode 100644 index 0000000..f346815 --- /dev/null +++ b/drizzle/0011_onchain_attempts.sql @@ -0,0 +1,2 @@ +ALTER TABLE resources ADD COLUMN onchain_attempts integer NOT NULL DEFAULT 0; +ALTER TABLE payments ADD COLUMN delivery_status text NOT NULL DEFAULT 'delivered'; \ No newline at end of file diff --git a/src/config.ts b/src/config.ts index 93708c7..b8029ab 100644 --- a/src/config.ts +++ b/src/config.ts @@ -244,6 +244,11 @@ const envSchema = z.object({ WEBHOOK_TIMEOUT_MS: z.coerce.number().default(10_000), WEBHOOK_MAX_ATTEMPTS: z.coerce.number().default(3), + // Retry worker (#85): max items processed per tick and per-item timeout so a + // slow RPC can't make a tick run into the next one. + RETRY_WORKER_BATCH_SIZE: z.coerce.number().int().min(1).default(25), + RETRY_WORKER_ITEM_TIMEOUT_MS: z.coerce.number().int().min(1).default(30_000), + // Optional Redis URL for a shared sliding-window rate-limit store (multi-instance). REDIS_URL: z.string().url().optional(), diff --git a/src/db/schema.ts b/src/db/schema.ts index 5ebb6ce..c291fbc 100644 --- a/src/db/schema.ts +++ b/src/db/schema.ts @@ -91,6 +91,9 @@ export const resources = pgTable( adminDelistedAt: timestamp("admin_delisted_at"), onchainStatus: onchainStatusEnum("onchain_status").notNull().default("none"), onchainTxHash: text("onchain_tx_hash"), + // Number of on-chain registration attempts made so far. Persisted so the + // retry budget survives worker restarts (#84). + onchainAttempts: integer("onchain_attempts").notNull().default(0), // Set when onchainStatus transitions to "pending". The retry worker uses // this (not createdAt) for its cutoff so a freshly pending old resource // isn't retried before a live registration can finish (#82). @@ -175,6 +178,9 @@ export const payments = pgTable( payerAddress: text("payer_address").notNull(), recipientAddress: text("recipient_address").notNull(), amount: text("amount").notNull(), // USDC amount + // Delivery state of the purchased content: "delivered" once the file was + // served, "undelivered" when the download failed after payment (#86). + deliveryStatus: text("delivery_status").notNull().default("delivered"), // Settlement state of the payment. Only settled payments count toward // earnings (#99). status: paymentStatusEnum("status").notNull().default("pending"), diff --git a/src/routes/resources.ts b/src/routes/resources.ts index 2f44207..177c405 100644 --- a/src/routes/resources.ts +++ b/src/routes/resources.ts @@ -307,29 +307,39 @@ router.get("/resources/:id", dynamicPaywall, async (req, res) => { } } - const [payment] = await db - .insert(payments) - .values({ - resourceId: resource.id, - payerAddress, - recipientAddress: resource.walletAddress, - amount: resource.price, - }) - .returning(); - - // Fire-and-forget: webhook delivery never delays or fails content access. - const purchase = { - paymentId: payment.id, - resourceId: payment.resourceId, - amount: payment.amount, - payerAddress: payment.payerAddress, - recipientAddress: payment.recipientAddress, - paidAt: payment.paidAt, + // Records a payment row with an explicit delivery status so an undelivered + // purchase can be reconciled later (#86). + const recordPayment = async (deliveryStatus: "delivered" | "undelivered") => { + const [payment] = await db + .insert(payments) + .values({ + resourceId: resource.id, + payerAddress, + recipientAddress: resource.walletAddress, + amount: resource.price, + deliveryStatus, + }) + .returning(); + return payment; + }; + + const emitPurchase = (payment: typeof payments.$inferSelect) => { + // Fire-and-forget: webhook delivery never delays or fails content access. + const purchase = { + paymentId: payment.id, + resourceId: payment.resourceId, + amount: payment.amount, + payerAddress: payment.payerAddress, + recipientAddress: payment.recipientAddress, + paidAt: payment.paidAt, + }; + void emitToPublisher(resource.publisherId, "resource.purchased", purchase); + void emitToPublisher(resource.publisherId, "payment.received", purchase); }; - void emitToPublisher(resource.publisherId, "resource.purchased", purchase); - void emitToPublisher(resource.publisherId, "payment.received", purchase); if (resource.resourceType === "link") { + const payment = await recordPayment("delivered"); + emitPurchase(payment); res.json({ url: resource.externalUrl, receipt: { @@ -348,12 +358,39 @@ router.get("/resources/:id", dynamicPaywall, async (req, res) => { throw new AppError("INTERNAL_ERROR", "Resource file not found"); } + // Fetch the content *before* recording the payment so a storage failure + // never leaves a silently undelivered purchase. + let fileContent: Awaited>; + try { + fileContent = await downloadFile(resource.storagePath); + } catch (err) { + const payment = await recordPayment("undelivered"); + getLogger().error( + { + event: "resource_delivery_failed", + resourceId: resource.id, + paymentId: payment.id, + payerAddress, + storagePath: resource.storagePath, + err, + }, + "file download failed after payment; payment marked undelivered for reconciliation", + ); + throw new AppError("UPSTREAM_ERROR", "Failed to deliver resource content", { + paymentId: payment.id, + deliveryStatus: "undelivered", + }); + } + + const payment = await recordPayment("delivered"); + emitPurchase(payment); + // Add receipt info in headers for file downloads res.setHeader("X-Payment-Id", payment.id); res.setHeader("X-Payment-Amount", `${payment.amount} USDC`); res.setHeader("X-Payment-Recipient", payment.recipientAddress); - const { buffer, mimeType } = await downloadFile(resource.storagePath); + const { buffer, mimeType } = fileContent; res.setHeader("Content-Type", mimeType); res.setHeader( "Content-Disposition", diff --git a/src/workers/retryPendingWorker.ts b/src/workers/retryPendingWorker.ts index bfb3118..bd1971c 100644 --- a/src/workers/retryPendingWorker.ts +++ b/src/workers/retryPendingWorker.ts @@ -35,6 +35,7 @@ export async function enqueueThumbnail(resourceId: string): Promise { } let intervalHandle: ReturnType | null = null; +let inFlight = false; export function startRetryPendingWorker(): void { getLogger().info( @@ -51,10 +52,18 @@ export function stopRetryPendingWorker(): void { clearInterval(intervalHandle); intervalHandle = null; } - retryCounts.clear(); } async function tick(): Promise { + if (inFlight) { + getLogger().debug( + { event: "retry_worker_tick_skipped" }, + "previous retry worker tick still running; skipping", + ); + return; + } + + inFlight = true; try { const cutoff = sql`NOW() - INTERVAL '5 minutes'`; @@ -62,8 +71,12 @@ async function tick(): Promise { .select() .from(resources) .where( - and(eq(resources.onchainStatus, "pending"), lt(resources.onchainPendingSince, cutoff)), - ); + and( + eq(resources.onchainStatus, "pending"), + lt(resources.onchainPendingSince, cutoff), + ), + ) + .limit(config.RETRY_WORKER_BATCH_SIZE); if (stuck.length === 0) return; @@ -73,16 +86,39 @@ async function tick(): Promise { ); for (const resource of stuck) { - await retryResource(resource); + await withTimeout( + retryResource(resource), + config.RETRY_WORKER_ITEM_TIMEOUT_MS, + resource.id, + ); } } catch (err) { getLogger().error({ event: "retry_worker_tick_error", err }, "retry worker tick failed"); + } finally { + inFlight = false; } await processCleanupJobs(); await processThumbnailJobs(); } +async function withTimeout(promise: Promise, ms: number, resourceId: string): Promise { + let timer: ReturnType | undefined; + try { + return await Promise.race([ + promise, + new Promise((_, reject) => { + timer = setTimeout( + () => reject(new Error(`retryResource timed out after ${ms}ms for ${resourceId}`)), + ms, + ); + }), + ]); + } finally { + if (timer) clearTimeout(timer); + } +} + async function processCleanupJobs(): Promise { const log = getLogger(); try { @@ -182,7 +218,30 @@ async function processThumbnailJobs(): Promise { async function retryResource(resource: typeof resources.$inferSelect): Promise { const log = getLogger().child({ resourceId: resource.id }); + let attempts = 0; + try { + const [incremented] = await db + .update(resources) + .set({ onchainAttempts: sql`${resources.onchainAttempts} + 1` }) + .where(eq(resources.id, resource.id)) + .returning({ onchainAttempts: resources.onchainAttempts }); + + attempts = incremented?.onchainAttempts ?? 0; + + if (attempts > MAX_RETRY_ATTEMPTS) { + log.warn( + { event: "retry_exhausted", attempts }, + "max retry attempts exhausted; marking as failed", + ); + await db + .update(resources) + .set({ onchainStatus: "failed" }) + .where(eq(resources.id, resource.id)); + retryCounts.delete(resource.id); + return; + } + if (resource.onchainTxHash) { const status = await pollTransactionStatus(resource.onchainTxHash); @@ -229,23 +288,6 @@ async function retryResource(resource: typeof resources.$inferSelect): Promise MAX_RETRY_ATTEMPTS) { - log.warn( - { event: "retry_exhausted", attempts: attempts - 1 }, - "max retry attempts exhausted; marking as failed", - ); - await db - .update(resources) - .set({ onchainStatus: "failed" }) - .where(eq(resources.id, resource.id)); - retryCounts.delete(resource.id); return; } @@ -316,7 +358,6 @@ async function retryResource(resource: typeof resources.$inferSelect): Promise MAX_RETRY_ATTEMPTS) { log.warn( - { event: "retry_exhausted_on_error", attempts: attempts - 1 }, + { event: "retry_exhausted_on_error", attempts }, "max retry attempts exhausted after error; marking as failed", ); await db .update(resources) .set({ onchainStatus: "failed" }) .where(eq(resources.id, resource.id)); - retryCounts.delete(resource.id); } } }