diff --git a/README.md b/README.md index 4f771bb..fbc908e 100644 --- a/README.md +++ b/README.md @@ -188,6 +188,7 @@ All configuration is done via environment variables. Defaults are shown in paren | `JOB_QUEUE_MAX_RETRIES` | Max retry attempts for a failed job | `10` | | `JOB_QUEUE_TIMEOUT_MS` | Timeout per job (ms) | `60000` | | `JOB_QUEUE_RETRY_AFTER_MS` | Delay before retrying a failed job (ms) | `5000` | +| `JOB_QUEUE_INVALID_RETRY_AFTER_MS` | Delay before re-processing a job marked `invalid` (ms). Tokens are re-enqueued on every re-mint, so this keeps a contract with unparseable metadata from being re-fetched on each mint | `3600000` | ### Metadata fetching diff --git a/src/env.ts b/src/env.ts index 5388fb4..b7a9ea0 100644 --- a/src/env.ts +++ b/src/env.ts @@ -70,6 +70,13 @@ const schema = Type.Object({ JOB_QUEUE_TIMEOUT_MS: Type.Number({ default: 60_000 }), /** Minimum time we will wait to retry a job after it's been executed. */ JOB_QUEUE_RETRY_AFTER_MS: Type.Number({ default: 5_000 }), + /** + * Minimum time we will wait before processing a job again after it was marked as `invalid`. A + * token's job row is re-enqueued as `pending` every time that token is re-minted, so without this + * backoff a contract whose metadata never parses (but which mints constantly, e.g. an AMM pool + * minting one SFT per price bin) would be re-fetched and re-written on every single mint. + */ + JOB_QUEUE_INVALID_RETRY_AFTER_MS: Type.Number({ default: 3_600_000 }), /** * The max number of immediate attempts that will be made to retrieve metadata from external URIs diff --git a/src/pg/stacks-core-pg-store.ts b/src/pg/stacks-core-pg-store.ts index a4d0e8e..f5aeabb 100644 --- a/src/pg/stacks-core-pg-store.ts +++ b/src/pg/stacks-core-pg-store.ts @@ -497,13 +497,13 @@ export class StacksCorePgStore extends BasePgStoreModule { INSERT INTO tokens (smart_contract_id, type, token_number, block_height, index_block_hash, tx_id, tx_index) (SELECT * FROM filtered_values) ON CONFLICT ON CONSTRAINT tokens_smart_contract_id_token_number_unique DO - UPDATE SET - uri = EXCLUDED.uri, - name = EXCLUDED.name, - symbol = EXCLUDED.symbol, - decimals = EXCLUDED.decimals, - total_supply = EXCLUDED.total_supply, - updated_at = NOW() + -- Deliberately a no-op write, only here so RETURNING gives us the existing row's id. + -- This INSERT doesn't carry the metadata columns, so assigning them from EXCLUDED would + -- blank out metadata we already fetched. updated_at must not move either: a null + -- updated_at on a pending/queued job is how getTokenMetadataBundleInternal recognizes a + -- token that hasn't been processed yet, and a re-mint can land before the first job + -- runs. The metadata write and updateTokenSupply both set updated_at themselves. + UPDATE SET smart_contract_id = EXCLUDED.smart_contract_id RETURNING id ) INSERT INTO jobs (token_id) (SELECT id AS token_id FROM token_inserts) @@ -593,7 +593,18 @@ export class StacksCorePgStore extends BasePgStoreModule { id: number; status: DbJobStatus; invalidReason?: DbJobInvalidReason; + retryAfterMs?: number; }): Promise { + let retryFragment; + if (args.retryAfterMs !== undefined) { + const retryAfter = args.retryAfterMs.toString(); + retryFragment = this.sql`retry_count = 0, + retry_after = NOW() + INTERVAL '${this.sql(retryAfter)} ms',`; + } else if (args.status != DbJobStatus.pending) { + retryFragment = this.sql`retry_count = 0, retry_after = NULL,`; + } else { + retryFragment = this.sql``; + } await this.sql` UPDATE jobs SET status = ${args.status}, @@ -602,11 +613,7 @@ export class StacksCorePgStore extends BasePgStoreModule { ? args.invalidReason : this.sql`NULL` }, - ${ - args.status != DbJobStatus.pending - ? this.sql`retry_count = 0, retry_after = NULL,` - : this.sql`` - } + ${retryFragment} updated_at = NOW() WHERE id = ${args.id} `; diff --git a/src/token-processor/queue/job/job.ts b/src/token-processor/queue/job/job.ts index e8c80d2..3ea9159 100644 --- a/src/token-processor/queue/job/job.ts +++ b/src/token-processor/queue/job/job.ts @@ -43,6 +43,7 @@ export abstract class Job { async work(): Promise { let status: DbJobStatus | undefined; let invalidReason: DbJobInvalidReason | undefined; + let retryAfterMs: number | undefined; const sw = stopwatch(); // This block will catch any and all errors that are generated while processing the job. Each of @@ -86,13 +87,18 @@ export abstract class Job { logger.warn(error, `User error on Job ${this.description()}`); status = DbJobStatus.invalid; invalidReason = getUserErrorInvalidReason(error); + // Hold off on re-processing this job for a while. The metadata behind a token's URI can be + // fixed without the contract ever changing, so re-mints legitimately re-enqueue this job as + // `pending`, but a token that mints constantly would otherwise have us re-fetch a URI we + // already know is bad on every single mint. + retryAfterMs = ENV.JOB_QUEUE_INVALID_RETRY_AFTER_MS; } else { logger.error(error, `Job ${this.description()}`); status = DbJobStatus.failed; } } finally { if (status) { - if (await this.updateStatus(status, invalidReason)) { + if (await this.updateStatus(status, invalidReason, retryAfterMs)) { logger.info(`Job ${this.description()} ${status} in ${sw.getElapsed()}ms`); } } @@ -101,10 +107,11 @@ export abstract class Job { private async updateStatus( status: DbJobStatus, - invalidReason?: DbJobInvalidReason + invalidReason?: DbJobInvalidReason, + retryAfterMs?: number ): Promise { try { - await this.db.core.updateJobStatus({ id: this.job.id, status, invalidReason }); + await this.db.core.updateJobStatus({ id: this.job.id, status, invalidReason, retryAfterMs }); return true; } catch (error) { logger.error(`Job ${this.description()} could not update status to ${status}: ${error}`); diff --git a/tests/stacks-core/sft-events.test.ts b/tests/stacks-core/sft-events.test.ts index 368d61f..b124f70 100644 --- a/tests/stacks-core/sft-events.test.ts +++ b/tests/stacks-core/sft-events.test.ts @@ -11,6 +11,7 @@ import { setupEnv, } from '../helpers.js'; import { StacksCoreBlockProcessor } from '../../src/stacks-core/stacks-core-block-processor.js'; +import { TokenNotProcessedError } from '../../src/pg/errors.js'; import { afterEach, beforeEach, describe, test } from 'node:test'; describe('sft events', () => { @@ -77,4 +78,103 @@ describe('sft events', () => { assert.strictEqual(jobs.length, 1); assert.strictEqual(jobs[0].token_id, 1); }); + + describe('re-mints', () => { + const address = 'SP3K8BC0PPEVCV7NZ6QSRWPQ2JE9E5B6N3PA0KBR9'; + const contractId = `${address}.key-alex-autoalex-v1`; + + /** Builds a block that mints SFT #3 of `contractId`. */ + const mintBlock = (height: number, hash: string, parentHash: string, txId: string) => + new TestBlockBuilder({ + block_height: height, + index_block_hash: hash, + parent_index_block_hash: parentHash, + }) + .addTransaction( + new TestTransactionBuilder({ tx_id: txId, sender: address }) + .addContractEvent( + contractId, + cvToHex( + tupleCV({ + type: bufferCV(Buffer.from('sft_mint')), + recipient: bufferCV(Buffer.from(address)), + 'token-id': uintCV(3), + amount: uintCV(1000), + }) + ) + ) + .build() + ) + .build(); + + beforeEach(async () => { + await insertAndEnqueueTestContract(db, contractId, DbSipNumber.sip013); + await markAllJobsAsDone(db); + }); + + test('keeps already fetched metadata and re-enqueues the token', async () => { + await processor.processBlock(mintBlock(2, '0x000002', '0x000001', '0x01')); + + // Pretend the token's job already ran and wrote its metadata. + await db.sql` + UPDATE tokens + SET uri = 'https://example.com/3.json', name = 'Test Token', symbol = 'TEST', decimals = 6, + total_supply = '1000', updated_at = NOW() + WHERE id = 1 + `; + await markAllJobsAsDone(db); + + // The same token mints again in a later block. + await processor.processBlock(mintBlock(3, '0x000003', '0x000002', '0x02')); + + const token = await db.getToken({ id: 1 }); + assert.strictEqual(token?.uri, 'https://example.com/3.json'); + assert.strictEqual(token?.name, 'Test Token'); + assert.strictEqual(token?.symbol, 'TEST'); + assert.strictEqual(token?.decimals, 6); + assert.strictEqual(token?.total_supply, '1000'); + + // ...but it's queued for a refresh, since the metadata behind that URI may have changed. + const jobs = await db.getPendingJobBatch({ limit: 1 }); + assert.strictEqual(jobs.length, 1); + assert.strictEqual(jobs[0].token_id, 1); + }); + + test('do not clear the backoff on an invalid metadata job', async () => { + await processor.processBlock(mintBlock(2, '0x000002', '0x000001', '0x01')); + + // The token's job ran and hit a user error, so it's invalid with a backoff still pending. + await db.sql` + UPDATE jobs SET status = 'invalid', retry_after = NOW() + INTERVAL '1 hour' + WHERE token_id = 1 + `; + + await processor.processBlock(mintBlock(3, '0x000003', '0x000002', '0x02')); + + // The re-mint re-enqueues the job, but must leave `retry_after` intact so the queue keeps + // skipping it until the backoff elapses. + const jobs = await db.sql<{ status: string; retry_after: string | null }[]>` + SELECT status, retry_after FROM jobs WHERE token_id = 1 + `; + assert.strictEqual(jobs[0].status, 'pending'); + assert.notStrictEqual(jobs[0].retry_after, null); + const batch = await db.getPendingJobBatch({ limit: 1 }); + assert.strictEqual(batch.length, 0); + }); + + test('before first processing leave the token marked as unprocessed', async () => { + await processor.processBlock(mintBlock(2, '0x000002', '0x000001', '0x01')); + // Re-mint lands before the token's metadata job ever runs. + await processor.processBlock(mintBlock(3, '0x000003', '0x000002', '0x02')); + + // A null `updated_at` on a pending job is how an unprocessed token is recognized, so the + // re-mint must not touch it. + const token = await db.getToken({ id: 1 }); + assert.strictEqual(token?.updated_at, null); + await assert.rejects( + db.getTokenMetadataBundle({ contractPrincipal: contractId, tokenNumber: 3 }), + TokenNotProcessedError + ); + }); + }); }); diff --git a/tests/token-queue/job.test.ts b/tests/token-queue/job.test.ts index 951a92b..0cfbd76 100644 --- a/tests/token-queue/job.test.ts +++ b/tests/token-queue/job.test.ts @@ -87,6 +87,26 @@ describe('Job', () => { assert.strictEqual(dbJob1?.status, 'invalid'); }); + test('user error backs off a job that gets re-enqueued before retry_after', async () => { + ENV.JOB_QUEUE_INVALID_RETRY_AFTER_MS = 200; + const job = new TestUserErrorJob({ db, job: dbJob, network: 'mainnet' }); + + await assert.doesNotReject(job.work()); + const dbJob1 = await db.getJob({ id: dbJob.id }); + assert.strictEqual(dbJob1?.status, 'invalid'); + assert.notStrictEqual(dbJob1?.retry_after, undefined); + + // A token re-mint re-enqueues the job as `pending` without touching `retry_after`, so the queue + // should still skip it until the backoff elapses. + await db.sql`UPDATE jobs SET status = 'pending', updated_at = NOW() WHERE id = ${dbJob.id}`; + const jobs1 = await db.getPendingJobBatch({ limit: 1 }); + assert.strictEqual(jobs1.length, 0); + + await timeout(300); + const jobs2 = await db.getPendingJobBatch({ limit: 1 }); + assert.strictEqual(jobs2.length, 1); + }); + test('retry_count limit reached marks entry as failed', async () => { ENV.JOB_QUEUE_STRICT_MODE = false; ENV.JOB_QUEUE_MAX_RETRIES = 0;