From b58b84a81ef9227dd82bc1113d00d67f6465ade3 Mon Sep 17 00:00:00 2001 From: FragginWagon Date: Wed, 9 Sep 2026 18:02:35 -0400 Subject: [PATCH] feat: bound pricing job retries --- .../api/src/infrastructure/database/schema.js | 2 + .../pricing-refresh-job-repository.js | 44 ++++++++++++++----- .../tests/pricing-refresh-worker.test.js | 14 ++++++ .../migrations/0019_pricing_job_retries.sql | 2 + database/migrations/meta/_journal.json | 7 +++ docs/IMPLEMENTATION-STATUS.md | 1 + docs/PRICING.md | 5 +++ 7 files changed, 65 insertions(+), 10 deletions(-) create mode 100644 database/migrations/0019_pricing_job_retries.sql diff --git a/apps/api/src/infrastructure/database/schema.js b/apps/api/src/infrastructure/database/schema.js index 282631a..7c67e44 100644 --- a/apps/api/src/infrastructure/database/schema.js +++ b/apps/api/src/infrastructure/database/schema.js @@ -321,6 +321,8 @@ export const pricingRefreshJobs = pgTable('pricing_refresh_jobs', { conditionId: uuid('condition_id').notNull().references(() => conditions.id), providerId: text('provider_id').notNull(), status: text('status').default('PENDING').notNull(), + attemptCount: integer('attempt_count').default(0).notNull(), + nextAttemptAt: timestamp('next_attempt_at', { withTimezone: true }), observations: jsonb('observations').default([]).notNull(), valuation: jsonb('valuation'), error: text('error'), diff --git a/apps/api/src/modules/pricing/repositories/pricing-refresh-job-repository.js b/apps/api/src/modules/pricing/repositories/pricing-refresh-job-repository.js index 4f33d2b..7fac333 100644 --- a/apps/api/src/modules/pricing/repositories/pricing-refresh-job-repository.js +++ b/apps/api/src/modules/pricing/repositories/pricing-refresh-job-repository.js @@ -1,8 +1,9 @@ -import { and, asc, desc, eq } from 'drizzle-orm'; +import { and, asc, desc, eq, isNull, lte, or } from 'drizzle-orm'; import { pricingRefreshJobs } from '../../../infrastructure/database/schema.js'; export function createPricingRefreshJobRepository(db) { + const maxAttempts = 3; return { async create(job) { const [created] = await db.insert(pricingRefreshJobs).values(job).returning(); @@ -12,13 +13,16 @@ export function createPricingRefreshJobRepository(db) { async claimNext() { return db.transaction(async (transaction) => { const [job] = await transaction.select().from(pricingRefreshJobs) - .where(eq(pricingRefreshJobs.status, 'PENDING')) + .where(and( + eq(pricingRefreshJobs.status, 'PENDING'), + or(isNull(pricingRefreshJobs.nextAttemptAt), lte(pricingRefreshJobs.nextAttemptAt, new Date())), + )) .orderBy(asc(pricingRefreshJobs.createdAt)) .limit(1) .for('update', { skipLocked: true }); if (!job) return null; const [claimed] = await transaction.update(pricingRefreshJobs) - .set({ status: 'PROCESSING', updatedAt: new Date(), error: null }) + .set({ status: 'PROCESSING', attemptCount: pricingRefreshJobs.attemptCount, updatedAt: new Date(), error: null, nextAttemptAt: null }) .where(and(eq(pricingRefreshJobs.id, job.id), eq(pricingRefreshJobs.status, 'PENDING'))) .returning(); return claimed ?? null; @@ -46,8 +50,8 @@ export function createPricingRefreshJobRepository(db) { async retryForSeller(id, sellerId) { const [job] = await db.update(pricingRefreshJobs) - .set({ status: 'PENDING', error: null, observations: [], valuation: null, updatedAt: new Date() }) - .where(and(eq(pricingRefreshJobs.id, id), eq(pricingRefreshJobs.sellerId, sellerId), eq(pricingRefreshJobs.status, 'FAILED'))) + .set({ status: 'PENDING', error: null, observations: [], valuation: null, nextAttemptAt: new Date(), updatedAt: new Date() }) + .where(and(eq(pricingRefreshJobs.id, id), eq(pricingRefreshJobs.sellerId, sellerId), eq(pricingRefreshJobs.status, 'FAILED'), lte(pricingRefreshJobs.attemptCount, maxAttempts - 1))) .returning(); return job ?? null; }, @@ -61,11 +65,31 @@ export function createPricingRefreshJobRepository(db) { }, async fail(id, message) { - const [failed] = await db.update(pricingRefreshJobs) - .set({ status: 'FAILED', error: message, updatedAt: new Date() }) - .where(and(eq(pricingRefreshJobs.id, id), eq(pricingRefreshJobs.status, 'PROCESSING'))) - .returning(); - return failed ?? null; + return db.transaction(async (transaction) => { + const [current] = await transaction.select().from(pricingRefreshJobs) + .where(and(eq(pricingRefreshJobs.id, id), eq(pricingRefreshJobs.status, 'PROCESSING'))); + if (!current) return null; + const attemptCount = current.attemptCount + 1; + const retryable = attemptCount < maxAttempts; + const [updated] = await transaction.update(pricingRefreshJobs) + .set({ + status: retryable ? 'PENDING' : 'FAILED', + error: message, + attemptCount, + updatedAt: new Date(), + nextAttemptAt: retryable ? new Date(Date.now() + (2 ** attemptCount) * 1000) : null, + }) + .where(and(eq(pricingRefreshJobs.id, id), eq(pricingRefreshJobs.status, 'PROCESSING'))) + .returning(); + return updated ?? null; + }); + }, + + async purgeCompletedBefore(sellerId, cutoff) { + const deleted = await db.delete(pricingRefreshJobs) + .where(and(eq(pricingRefreshJobs.sellerId, sellerId), eq(pricingRefreshJobs.status, 'COMPLETED'), lte(pricingRefreshJobs.updatedAt, cutoff))) + .returning({ id: pricingRefreshJobs.id }); + return deleted.length; }, }; } \ No newline at end of file diff --git a/apps/worker/tests/pricing-refresh-worker.test.js b/apps/worker/tests/pricing-refresh-worker.test.js index b13023d..7465b54 100644 --- a/apps/worker/tests/pricing-refresh-worker.test.js +++ b/apps/worker/tests/pricing-refresh-worker.test.js @@ -62,3 +62,17 @@ test('persists completed and failed outcomes through the job repository', async assert.deepEqual(calls, [['complete', 'job-3', [{ id: 'observation-1' }], { valueMinor: 2500 }]]); }); + +test('persists failed pricing refresh outcomes through the repository', async () => { + const calls = []; + await processNextPersistentPricingRefreshJob({ + jobRepository: { + async claimNext() { return { id: 'job-4' }; }, + async complete() { throw new Error('not expected'); }, + async fail(id, error) { calls.push([id, error]); }, + }, + async pricingRefreshJob() { return { status: 'FAILED', error: 'Provider unavailable.' }; }, + }); + + assert.deepEqual(calls, [['job-4', 'Provider unavailable.']]); +}); diff --git a/database/migrations/0019_pricing_job_retries.sql b/database/migrations/0019_pricing_job_retries.sql new file mode 100644 index 0000000..1ff2005 --- /dev/null +++ b/database/migrations/0019_pricing_job_retries.sql @@ -0,0 +1,2 @@ +ALTER TABLE "pricing_refresh_jobs" ADD COLUMN "attempt_count" integer DEFAULT 0 NOT NULL; +ALTER TABLE "pricing_refresh_jobs" ADD COLUMN "next_attempt_at" timestamp with time zone; \ No newline at end of file diff --git a/database/migrations/meta/_journal.json b/database/migrations/meta/_journal.json index d7fb717..f4aaabf 100644 --- a/database/migrations/meta/_journal.json +++ b/database/migrations/meta/_journal.json @@ -134,6 +134,13 @@ "when": 1790150400000, "tag": "0018_provider_mappings", "breakpoints": true + }, + { + "idx": 19, + "version": "7", + "when": 1790236800000, + "tag": "0019_pricing_job_retries", + "breakpoints": true } ] } \ No newline at end of file diff --git a/docs/IMPLEMENTATION-STATUS.md b/docs/IMPLEMENTATION-STATUS.md index fdd81aa..58d6a30 100644 --- a/docs/IMPLEMENTATION-STATUS.md +++ b/docs/IMPLEMENTATION-STATUS.md @@ -154,6 +154,7 @@ Work through one bounded vertical slice at a time. The linked documentation rema - Product-level provider mappings now persist seller-scoped `PRODUCT`/`SEARCH` relationships and support active listing and controlled disabling. - Pricing refresh enqueue now resolves its provider from an active Product mapping and rejects unmapped Products explicitly. - Local development and CI can enable a deterministic fixture provider for end-to-end refresh coverage; production provider adapters remain explicitly configured deployment boundaries. +- Pricing refresh jobs have a bounded three-attempt retry policy with exponential backoff and an explicit completed-job purge operation for retention scheduling. - `npm install` currently reports dependency audit findings. Review and remediate them before production deployment; do not use `npm audit fix --force` without reviewing its changes. ## Local Commands diff --git a/docs/PRICING.md b/docs/PRICING.md index 1b11006..e825511 100644 --- a/docs/PRICING.md +++ b/docs/PRICING.md @@ -117,6 +117,11 @@ refresh-job storage and row-locked worker claim/complete/fail lifecycle are implemented in PostgreSQL. External provider configuration remains deployment work. +Refresh jobs allow three attempts. Failed attempts requeue with short +exponential backoff; the third failure becomes terminal `FAILED`. Completed-job +retention is handled explicitly through the repository cleanup operation so a +deployment can schedule cleanup according to its retention policy. + Provider failure should not invalidate other valid observations. The failed provider is recorded for admin diagnostics. ## Overrides