feat: add strict ft/nft metadata processing mode for better error handling (#1165)

* refactor: create root dir for token-metadata, separate into files

* feat: start creating retryable errors

* feat: catch all errors in one place

* fix: current tests

* feat: add strict flag and max retry env var

* style: comment nits

* style: remove unused exports

* fix: existing unit tests

* test: strict mode

* test: block metadata retry

* test: metadata uri timeout and max size

* chore: move retry_count column to separate migration

* fix: make fetch timeout dynamic

* feat: retry metadata uri fetch 5 times before failing

Author: rafaelcr

.env

@@ -105,11 +105,21 @@ MAINNET_SEND_MANY_CONTRACT_ID=SP3FBR2AGK5H9QBDH3EEN6DF8EK8JY7RX8QJ5SVTE.send-man
 # STACKS_API_ENABLE_FT_METADATA=1
 # STACKS_API_ENABLE_NFT_METADATA=1
 
+# If token metadata processing is enabled, this variable determines how the API reacts to metadata processing failures.
+# When strict mode is enabled, any failures caused by recoverable errors will be retried indefinitely. Otherwise,
+# the API will give up after `STACKS_API_TOKEN_METADATA_MAX_RETRIES` is reached for that smart contract.
+# STACKS_API_TOKEN_METADATA_STRICT_MODE=1
+
+# Maximum number of times we'll try processing FT/NFT metadata for a specific smart contract if we've failed
+# because of a recoverable error.
+# Only used if `STACKS_API_TOKEN_METADATA_STRICT_MODE` is disabled.
+# STACKS_API_TOKEN_METADATA_MAX_RETRIES=5
+
 # Controls the token metadata error handling mode. The possible values are:
 # * `warning`: If required metadata is not found, the API will issue a warning and not display data for that token.
 # * `error`: If required metadata is not found, the API will throw an error.
 # If not specified or any other value is provided, the mode will be set to `warning`.
-# STACKS_API_TOKEN_METADATA_ERROR_MODE=
+# STACKS_API_TOKEN_METADATA_ERROR_MODE=warning
 
 # Configure a script to handle image URLs during token metadata processing.
 # This example script uses the `imgix.net` service to create CDN URLs.

src/api/routes/rosetta/account.ts

@@ -16,7 +16,7 @@ import { RosettaErrors, RosettaConstants, RosettaErrorsTypes } from '../../roset
 import { rosettaValidateRequest, ValidSchema, makeRosettaError } from '../../rosetta-validate';
 import { ChainID } from '@stacks/transactions';
 import { getValidatedFtMetadata } from '../../../rosetta-helpers';
-import { isFtMetadataEnabled } from '../../../event-stream/tokens-contract-handler';
+import { isFtMetadataEnabled } from '../../../token-metadata/helpers';
 
 export function createRosettaAccountRouter(db: DataStore, chainId: ChainID): express.Router {
   const router = express.Router();

src/api/routes/tokens/tokens.ts

@@ -14,10 +14,7 @@ import {
   NonFungibleTokensMetadataList,
 } from '@stacks/stacks-blockchain-api-types';
 import { parseLimitQuery, parsePagingQueryInput } from './../../pagination';
-import {
-  isFtMetadataEnabled,
-  isNftMetadataEnabled,
-} from '../../../event-stream/tokens-contract-handler';
+import { isFtMetadataEnabled, isNftMetadataEnabled } from '../../../token-metadata/helpers';
 import { bufferToHexPrefixString, has0xPrefix, isValidPrincipal } from '../../../helpers';
 import { booleanValueForParam, isUnanchoredRequest } from '../../../api/query-helpers';
 import { decodeClarityValueToRepr } from 'stacks-encoding-native-js';

src/core-rpc/client.ts

@@ -2,6 +2,7 @@
 import fetch, { RequestInit } from 'node-fetch';
 import { parsePort, stopwatch, logError, timeout } from '../helpers';
 import { CoreNodeFeeResponse } from '@stacks/stacks-blockchain-api-types';
+import { ClarityValue, cvToHex } from '@stacks/transactions';
 
 interface CoreRpcAccountInfo {
   /** Hex-prefixed uint128. */
@@ -49,6 +50,20 @@ export interface Neighbor {
   authenticated: boolean;
 }
 
+interface ReadOnlyContractCallSuccessResponse {
+  okay: true;
+  result: string;
+}
+
+interface ReadOnlyContractCallFailResponse {
+  okay: false;
+  cause: string;
+}
+
+export type ReadOnlyContractCallResponse =
+  | ReadOnlyContractCallSuccessResponse
+  | ReadOnlyContractCallFailResponse;
+
 interface CoreRpcNeighbors {
   sample: Neighbor[];
   inbound: Neighbor[];
@@ -208,6 +223,27 @@ export class StacksCoreRpcClient {
     };
   }
 
+  async sendReadOnlyContractCall(
+    contractAddress: string,
+    contractName: string,
+    functionName: string,
+    senderAddress: string,
+    functionArgs: ClarityValue[]
+  ): Promise<ReadOnlyContractCallResponse> {
+    const body = {
+      sender: senderAddress,
+      arguments: functionArgs.map(arg => cvToHex(arg)),
+    };
+    return await this.fetchJson<ReadOnlyContractCallResponse>(
+      `v2/contracts/call-read/${contractAddress}/${contractName}/${functionName}`,
+      {
+        method: 'POST',
+        headers: { 'Content-Type': 'application/json' },
+        body: JSON.stringify(body),
+      }
+    );
+  }
+
   async getNeighbors(): Promise<CoreRpcNeighbors> {
     const result = await this.fetchJson<CoreRpcNeighbors>(`v2/neighbors`, {
       method: 'GET',

src/datastore/common.ts

@@ -604,6 +604,7 @@ export interface DbTokenMetadataQueueEntry {
   contractAbi: ClarityAbi;
   blockHeight: number;
   processed: boolean;
+  retry_count: number;
 }
 
 export interface DbChainTip {
@@ -983,8 +984,8 @@ export interface DataStore extends DataStoreEventEmitter {
   getFtMetadata(contractId: string): Promise<FoundOrNot<DbFungibleTokenMetadata>>;
   getNftMetadata(contractId: string): Promise<FoundOrNot<DbNonFungibleTokenMetadata>>;
 
-  updateNFtMetadata(nftMetadata: DbNonFungibleTokenMetadata, dbQueueId: number): Promise<number>;
-  updateFtMetadata(ftMetadata: DbFungibleTokenMetadata, dbQueueId: number): Promise<number>;
+  updateNFtMetadata(nftMetadata: DbNonFungibleTokenMetadata): Promise<number>;
+  updateFtMetadata(ftMetadata: DbFungibleTokenMetadata): Promise<number>;
 
   getFtMetadataList(args: {
     limit: number;
@@ -1001,6 +1002,19 @@ export interface DataStore extends DataStoreEventEmitter {
    */
   getTokenMetadataQueueEntry(queueId: number): Promise<FoundOrNot<DbTokenMetadataQueueEntry>>;
 
+  /**
+   * Marks a token metadata queue entry as processed.
+   * @param queueId - queue entry id
+   */
+  updateProcessedTokenMetadataQueueEntry(queueId: number): Promise<void>;
+
+  /**
+   * Increases the retry count for a specific token metadata queue entry.
+   * @param queueId - queue entry id
+   * @returns new retry count
+   */
+  increaseTokenMetadataQueueEntryRetryCount(queueId: number): Promise<number>;
+
   getTokenMetadataQueue(
     limit: number,
     excludingEntries: number[]

src/datastore/postgres-store.ts

@@ -105,7 +105,7 @@ import {
   Block,
 } from '@stacks/stacks-blockchain-api-types';
 import { getTxTypeId } from '../api/controllers/db-controller';
-import { isProcessableTokenMetadata } from '../event-stream/tokens-contract-handler';
+import { isProcessableTokenMetadata } from '../token-metadata/helpers';
 import { ChainID, ClarityAbi, hexToCV, TupleCV } from '@stacks/transactions';
 import {
   PgAddressNotificationPayload,
@@ -714,6 +714,7 @@ interface DbTokenMetadataQueueEntryQuery {
   contract_abi: string;
   block_height: number;
   processed: boolean;
+  retry_count: number;
 }
 
 interface StxEventQueryResult {
@@ -1503,6 +1504,7 @@ export class PgDataStore
               contractAbi: contractAbi,
               blockHeight: entry.tx.block_height,
               processed: false,
+              retry_count: 0,
             };
             return queueEntry;
           })
@@ -5128,6 +5130,7 @@ export class PgDataStore
       contractAbi: JSON.parse(row.contract_abi),
       blockHeight: row.block_height,
       processed: row.processed,
+      retry_count: row.retry_count,
     };
     return { found: true, result: entry };
   }
@@ -5158,6 +5161,7 @@ export class PgDataStore
         contractAbi: JSON.parse(row.contract_abi),
         blockHeight: row.block_height,
         processed: row.processed,
+        retry_count: row.retry_count,
       };
       return entry;
     });
@@ -5239,7 +5243,7 @@ export class PgDataStore
     });
   }
 
-  async getSmartContract(contractId: string) {
+  async getSmartContract(contractId: string): Promise<FoundOrNot<DbSmartContract>> {
     return this.query(async client => {
       const result = await client.query<{
         tx_id: Buffer;
@@ -7616,7 +7620,7 @@ export class PgDataStore
     });
   }
 
-  async updateFtMetadata(ftMetadata: DbFungibleTokenMetadata, dbQueueId: number): Promise<number> {
+  async updateFtMetadata(ftMetadata: DbFungibleTokenMetadata): Promise<number> {
     const {
       token_uri,
       name,
@@ -7629,8 +7633,7 @@ export class PgDataStore
       tx_id,
       sender_address,
     } = ftMetadata;
-
-    const rowCount = await this.queryTx(async client => {
+    const rowCount = await this.query(async client => {
       const result = await client.query(
         `
         INSERT INTO ft_metadata(
@@ -7650,24 +7653,13 @@ export class PgDataStore
           sender_address,
         ]
       );
-      await client.query(
-        `
-        UPDATE token_metadata_queue
-        SET processed = true
-        WHERE queue_id = $1
-        `,
-        [dbQueueId]
-      );
       return result.rowCount;
     });
     await this.notifier?.sendTokens({ contractID: contract_id });
     return rowCount;
   }
 
-  async updateNFtMetadata(
-    nftMetadata: DbNonFungibleTokenMetadata,
-    dbQueueId: number
-  ): Promise<number> {
+  async updateNFtMetadata(nftMetadata: DbNonFungibleTokenMetadata): Promise<number> {
     const {
       token_uri,
       name,
@@ -7678,7 +7670,7 @@ export class PgDataStore
       tx_id,
       sender_address,
     } = nftMetadata;
-    const rowCount = await this.queryTx(async client => {
+    const rowCount = await this.query(async client => {
       const result = await client.query(
         `
         INSERT INTO nft_metadata(
@@ -7696,18 +7688,38 @@ export class PgDataStore
           sender_address,
         ]
       );
+      return result.rowCount;
+    });
+    await this.notifier?.sendTokens({ contractID: contract_id });
+    return rowCount;
+  }
+
+  async updateProcessedTokenMetadataQueueEntry(queueId: number): Promise<void> {
+    await this.query(async client => {
       await client.query(
         `
         UPDATE token_metadata_queue
         SET processed = true
         WHERE queue_id = $1
         `,
-        [dbQueueId]
+        [queueId]
       );
-      return result.rowCount;
     });
-    await this.notifier?.sendTokens({ contractID: contract_id });
-    return rowCount;
+  }
+
+  async increaseTokenMetadataQueueEntryRetryCount(queueId: number): Promise<number> {
+    return await this.query(async client => {
+      const result = await client.query<{ retry_count: number }>(
+        `
+        UPDATE token_metadata_queue
+        SET retry_count = retry_count + 1
+        WHERE queue_id = $1
+        RETURNING retry_count
+        `,
+        [queueId]
+      );
+      return result.rows[0].retry_count;
+    });
   }
 
   getFtMetadataList({