fix: back off invalid metadata jobs and stop re-mints from wiping token metadata

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
 

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

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<void> {
+    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}
     `;

src/token-processor/queue/job/job.ts

@@ -43,6 +43,7 @@ export abstract class Job {
   async work(): Promise<void> {
     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<boolean> {
     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}`);

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
+      );
+    });
+  });
 });

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;