Merge pull request #1357 from hirosystems/6.1.0

6.1.0

Author: rafaelcr

.env

@@ -5,6 +5,10 @@ PG_PASSWORD=postgres
 PG_DATABASE=stacks_blockchain_api
 PG_SCHEMA=public
 PG_SSL=false
+# Idle connection timeout in seconds, defaults to 30
+# PG_IDLE_TIMEOUT=30
+# Max connection lifetime in seconds, defaults to 60
+# PG_MAX_LIFETIME=60
 
 # Can be any string, use to specify a use case specific to a deployment
 PG_APPLICATION_NAME=stacks-blockchain-api
@@ -27,12 +31,13 @@ PG_APPLICATION_NAME=stacks-blockchain-api
 # PG_PRIMARY_DATABASE=
 # PG_PRIMARY_SCHEMA=
 # PG_PRIMARY_SSL=
+# PG_PRIMARY_IDLE_TIMEOUT=
+# PG_PRIMARY_MAX_LIFETIME=
 # The connection URI below can be used in place of the PG variables above,
 # but if enabled it must be defined without others or omitted.
 # PG_PRIMARY_CONNECTION_URI=
 
 # Limit to how many concurrent connections can be created, defaults to 10
-# See https://node-postgres.com/api/pool
 # PG_CONNECTION_POOL_MAX=10
 
 # If specified, controls the Stacks Blockchain API mode. The possible values are:

.gitpod.Dockerfile

@@ -1,4 +1,4 @@
-FROM blockstack/stacks-blockchain:2.05.0.3.0-stretch as corenode
+FROM blockstack/stacks-blockchain:2.05.0.4.0 as corenode
 
 FROM gitpod/workspace-postgres
 

docker/stx-rosetta.Dockerfile

@@ -1,5 +1,5 @@
 ARG STACKS_API_VERSION=v0.71.2
-ARG STACKS_NODE_VERSION=2.05.0.3.0
+ARG STACKS_NODE_VERSION=2.05.0.4.0
 ARG STACKS_API_REPO=blockstack/stacks-blockchain-api
 ARG STACKS_NODE_REPO=blockstack/stacks-blockchain
 ARG PG_VERSION=12

src/datastore/connection.ts

@@ -138,6 +138,8 @@ export function getPostgres({
     ssl: pgEnvValue('SSL'),
     schema: pgEnvValue('SCHEMA'),
     applicationName: pgEnvValue('APPLICATION_NAME'),
+    idleTimeout: parseInt(pgEnvValue('IDLE_TIMEOUT') ?? '30'),
+    maxLifetime: parseInt(pgEnvValue('MAX_LIFETIME') ?? '60'),
     poolMax: parseInt(process.env['PG_CONNECTION_POOL_MAX'] ?? '10'),
   };
   const defaultAppName = 'stacks-blockchain-api';
@@ -180,6 +182,8 @@ export function getPostgres({
       host: pgEnvVars.host,
       port: parsePort(pgEnvVars.port),
       ssl: parseArgBoolean(pgEnvVars.ssl),
+      idle_timeout: pgEnvVars.idleTimeout,
+      max_lifetime: pgEnvVars.maxLifetime,
       max: pgEnvVars.poolMax,
       types: PG_TYPE_MAPPINGS,
       connection: {

src/datastore/pg-write-store.ts

@@ -188,6 +188,8 @@ export class PgWriteStore extends PgStore {
   async update(data: DataStoreBlockUpdateData): Promise<void> {
     const tokenMetadataQueueEntries: DbTokenMetadataQueueEntry[] = [];
     let garbageCollectedMempoolTxs: string[] = [];
+    let batchedTxData: DataStoreTxEventData[] = [];
+
     await this.sql.begin(async sql => {
       const chainTip = await this.getChainTip(sql, false);
       await this.handleReorg(sql, data.block, chainTip.blockHeight);
@@ -257,7 +259,7 @@ export class PgWriteStore extends PgStore {
       data.block.execution_cost_write_count = totalCost.execution_cost_write_count;
       data.block.execution_cost_write_length = totalCost.execution_cost_write_length;
 
-      let batchedTxData: DataStoreTxEventData[] = data.txs;
+      batchedTxData = data.txs;
 
       // Find microblocks that weren't already inserted via the unconfirmed microblock event.
       // This happens when a stacks-node is syncing and receives confirmed microblocks with their anchor block at the same time.
@@ -358,8 +360,6 @@ export class PgWriteStore extends PgStore {
             await this.updateNames(sql, entry.tx, bnsName);
           }
         }
-        await this.refreshNftCustody(sql, batchedTxData);
-        await this.refreshMaterializedView(sql, 'chain_tip');
         const mempoolGarbageResults = await this.deleteGarbageCollectedMempoolTxs(sql);
         if (mempoolGarbageResults.deletedTxs.length > 0) {
           logger.verbose(
@@ -399,6 +399,10 @@ export class PgWriteStore extends PgStore {
       }
     });
 
+    await this.refreshNftCustody(batchedTxData);
+    await this.refreshMaterializedView('chain_tip');
+    await this.refreshMaterializedView('mempool_digest');
+
     // Skip sending `PgNotifier` updates altogether if we're in the genesis block since this block is the
     // event replay of the v1 blockchain.
     if ((data.block.block_height > 1 || !isProdEnv) && this.notifier) {
@@ -529,6 +533,9 @@ export class PgWriteStore extends PgStore {
   }
 
   async updateMicroblocksInternal(data: DataStoreMicroblockUpdateData): Promise<void> {
+    const txData: DataStoreTxEventData[] = [];
+    let dbMicroblocks: DbMicroblock[] = [];
+
     await this.sql.begin(async sql => {
       // Sanity check: ensure incoming microblocks have a `parent_index_block_hash` that matches the API's
       // current known canonical chain tip. We assume this holds true so incoming microblock data is always
@@ -550,7 +557,7 @@ export class PgWriteStore extends PgStore {
 
       // The block height is just one after the current chain tip height
       const blockHeight = chainTip.blockHeight + 1;
-      const dbMicroblocks = data.microblocks.map(mb => {
+      dbMicroblocks = data.microblocks.map(mb => {
         const dbMicroBlock: DbMicroblock = {
           canonical: true,
           microblock_canonical: true,
@@ -570,8 +577,6 @@ export class PgWriteStore extends PgStore {
         return dbMicroBlock;
       });
 
-      const txs: DataStoreTxEventData[] = [];
-
       for (const entry of data.txs) {
         // Note: the properties block_hash and burn_block_time are empty here because the anchor block with that data doesn't yet exist.
         const dbTx: DbTx = {
@@ -582,7 +587,7 @@ export class PgWriteStore extends PgStore {
 
         // Set all the `block_height` properties for the related tx objects, since it wasn't known
         // when creating the objects using only the stacks-node message payload.
-        txs.push({
+        txData.push({
           tx: dbTx,
           stxEvents: entry.stxEvents.map(e => ({ ...e, block_height: blockHeight })),
           contractLogEvents: entry.contractLogEvents.map(e => ({
@@ -598,7 +603,7 @@ export class PgWriteStore extends PgStore {
         });
       }
 
-      await this.insertMicroblockData(sql, dbMicroblocks, txs);
+      await this.insertMicroblockData(sql, dbMicroblocks, txData);
 
       // Find any microblocks that have been orphaned by this latest microblock chain tip.
       // This function also checks that each microblock parent hash points to an existing microblock in the db.
@@ -646,24 +651,25 @@ export class PgWriteStore extends PgStore {
         );
       }
 
-      await this.refreshNftCustody(sql, txs, true);
-      await this.refreshMaterializedView(sql, 'chain_tip');
-
       if (!this.isEventReplay) {
         const mempoolStats = await this.getMempoolStatsInternal({ sql });
         this.eventEmitter.emit('mempoolStatsUpdate', mempoolStats);
       }
+    });
 
-      if (this.notifier) {
-        for (const microblock of dbMicroblocks) {
-          await this.notifier.sendMicroblock({ microblockHash: microblock.microblock_hash });
-        }
-        for (const tx of txs) {
-          await this.notifier.sendTx({ txId: tx.tx.tx_id });
-        }
-        await this.emitAddressTxUpdates(txs);
+    await this.refreshNftCustody(txData, true);
+    await this.refreshMaterializedView('chain_tip');
+    await this.refreshMaterializedView('mempool_digest');
+
+    if (this.notifier) {
+      for (const microblock of dbMicroblocks) {
+        await this.notifier.sendMicroblock({ microblockHash: microblock.microblock_hash });
       }
-    });
+      for (const tx of txData) {
+        await this.notifier.sendTx({ txId: tx.tx.tx_id });
+      }
+      await this.emitAddressTxUpdates(txData);
+    }
   }
 
   async updateStxLockEvent(sql: PgSqlClient, tx: DbTx, event: DbStxLockEvent) {
@@ -1313,13 +1319,12 @@ export class PgWriteStore extends PgStore {
           updatedTxs.push(tx);
         }
       }
-      await this.refreshMaterializedView(sql, 'mempool_digest');
-
       if (!this.isEventReplay) {
         const mempoolStats = await this.getMempoolStatsInternal({ sql });
         this.eventEmitter.emit('mempoolStatsUpdate', mempoolStats);
       }
     });
+    await this.refreshMaterializedView('mempool_digest');
     for (const tx of updatedTxs) {
       await this.notifier?.sendTx({ txId: tx.tx_id });
     }
@@ -1335,8 +1340,8 @@ export class PgWriteStore extends PgStore {
         RETURNING ${sql(MEMPOOL_TX_COLUMNS)}
       `;
       updatedTxs = updateResults.map(r => parseMempoolTxQueryResult(r));
-      await this.refreshMaterializedView(sql, 'mempool_digest');
     });
+    await this.refreshMaterializedView('mempool_digest');
     for (const tx of updatedTxs) {
       await this.notifier?.sendTx({ txId: tx.tx_id });
     }
@@ -1964,7 +1969,6 @@ export class PgWriteStore extends PgStore {
       WHERE tx_id IN ${sql(txIds)}
       RETURNING tx_id
     `;
-    await this.refreshMaterializedView(sql, 'mempool_digest');
     const restoredTxs = updateResults.map(r => r.tx_id);
     return { restoredTxs: restoredTxs };
   }
@@ -1988,7 +1992,6 @@ export class PgWriteStore extends PgStore {
       WHERE tx_id IN ${sql(txIds)}
       RETURNING tx_id
     `;
-    await this.refreshMaterializedView(sql, 'mempool_digest');
     const removedTxs = updateResults.map(r => r.tx_id);
     return { removedTxs: removedTxs };
   }
@@ -2002,7 +2005,9 @@ export class PgWriteStore extends PgStore {
     // Get threshold block.
     const blockThreshold = process.env['STACKS_MEMPOOL_TX_GARBAGE_COLLECTION_THRESHOLD'] ?? 256;
     const cutoffResults = await sql<{ block_height: number }[]>`
-      SELECT (block_height - ${blockThreshold}) AS block_height FROM chain_tip
+      SELECT (MAX(block_height) - ${blockThreshold}) AS block_height
+      FROM blocks
+      WHERE canonical = TRUE
     `;
     if (cutoffResults.length != 1) {
       return { deletedTxs: [] };
@@ -2016,7 +2021,6 @@ export class PgWriteStore extends PgStore {
       WHERE pruned = FALSE AND receipt_block_height < ${cutoffBlockHeight}
       RETURNING tx_id
     `;
-    await this.refreshMaterializedView(sql, 'mempool_digest');
     const deletedTxs = deletedTxResults.map(r => r.tx_id);
     for (const txId of deletedTxs) {
       await this.notifier?.sendTx({ txId: txId });
@@ -2437,11 +2441,12 @@ export class PgWriteStore extends PgStore {
 
   /**
    * Refreshes a Postgres materialized view.
-   * @param sql - Pg Client
    * @param viewName - Materialized view name
+   * @param sql - Pg scoped client. Will use the default client if none specified
    * @param skipDuringEventReplay - If we should skip refreshing during event replay
    */
-  async refreshMaterializedView(sql: PgSqlClient, viewName: string, skipDuringEventReplay = true) {
+  async refreshMaterializedView(viewName: string, sql?: PgSqlClient, skipDuringEventReplay = true) {
+    sql = sql ?? this.sql;
     if (this.isEventReplay && skipDuringEventReplay) {
       return;
     }
@@ -2454,34 +2459,32 @@ export class PgWriteStore extends PgStore {
    * @param txs - Transaction event data
    * @param unanchored - If this refresh is requested from a block or microblock
    */
-  async refreshNftCustody(
-    sql: PgSqlClient,
-    txs: DataStoreTxEventData[],
-    unanchored: boolean = false
-  ) {
-    const newNftEventCount = txs
-      .map(tx => tx.nftEvents.length)
-      .reduce((prev, cur) => prev + cur, 0);
-    if (newNftEventCount > 0) {
-      // Always refresh unanchored view since even if we're in a new anchored block we should update the
-      // unanchored state to the current one.
-      await this.refreshMaterializedView(sql, 'nft_custody_unanchored');
-      if (!unanchored) {
-        await this.refreshMaterializedView(sql, 'nft_custody');
-      }
-    } else if (!unanchored) {
-      // Even if we didn't receive new NFT events in a new anchor block, we should check if we need to
-      // update the anchored view to reflect any changes made by previous microblocks.
-      const result = await sql<{ outdated: boolean }[]>`
-        WITH anchored_height AS (SELECT MAX(block_height) AS anchored FROM nft_custody),
-          unanchored_height AS (SELECT MAX(block_height) AS unanchored FROM nft_custody_unanchored)
-        SELECT unanchored > anchored AS outdated
-        FROM anchored_height CROSS JOIN unanchored_height
-      `;
-      if (result.length > 0 && result[0].outdated) {
-        await this.refreshMaterializedView(sql, 'nft_custody');
+  async refreshNftCustody(txs: DataStoreTxEventData[], unanchored: boolean = false) {
+    await this.sql.begin(async sql => {
+      const newNftEventCount = txs
+        .map(tx => tx.nftEvents.length)
+        .reduce((prev, cur) => prev + cur, 0);
+      if (newNftEventCount > 0) {
+        // Always refresh unanchored view since even if we're in a new anchored block we should update the
+        // unanchored state to the current one.
+        await this.refreshMaterializedView('nft_custody_unanchored', sql);
+        if (!unanchored) {
+          await this.refreshMaterializedView('nft_custody', sql);
+        }
+      } else if (!unanchored) {
+        // Even if we didn't receive new NFT events in a new anchor block, we should check if we need to
+        // update the anchored view to reflect any changes made by previous microblocks.
+        const result = await sql<{ outdated: boolean }[]>`
+          WITH anchored_height AS (SELECT MAX(block_height) AS anchored FROM nft_custody),
+            unanchored_height AS (SELECT MAX(block_height) AS unanchored FROM nft_custody_unanchored)
+          SELECT unanchored > anchored AS outdated
+          FROM anchored_height CROSS JOIN unanchored_height
+        `;
+        if (result.length > 0 && result[0].outdated) {
+          await this.refreshMaterializedView('nft_custody', sql);
+        }
       }
-    }
+    });
   }
 
   /**
@@ -2492,10 +2495,10 @@ export class PgWriteStore extends PgStore {
       return;
     }
     await this.sql.begin(async sql => {
-      await this.refreshMaterializedView(sql, 'nft_custody', false);
-      await this.refreshMaterializedView(sql, 'nft_custody_unanchored', false);
-      await this.refreshMaterializedView(sql, 'chain_tip', false);
-      await this.refreshMaterializedView(sql, 'mempool_digest', false);
+      await this.refreshMaterializedView('nft_custody', sql, false);
+      await this.refreshMaterializedView('nft_custody_unanchored', sql, false);
+      await this.refreshMaterializedView('chain_tip', sql, false);
+      await this.refreshMaterializedView('mempool_digest', sql, false);
     });
   }
 }