Merge pull request #2224 from hirosystems/develop

release to master

Author: rafaelcr

migrations/1740074359683_last-node-events-index.js

@@ -0,0 +1,10 @@
+/* eslint-disable camelcase */
+
+exports.shorthands = undefined;
+
+exports.up = pgm => {
+  pgm.createIndex('event_observer_requests', [
+    'event_path',
+    { name: 'receive_timestamp', sort: 'DESC' },
+  ]);
+};

src/api/routes/address.ts

@@ -121,13 +121,19 @@ export const AddressRoutes: FastifyPluginAsync<
           blockHeight
         );
         let mempoolBalance: bigint | undefined = undefined;
+        let mempoolInbound: bigint | undefined = undefined;
+        let mempoolOutbound: bigint | undefined = undefined;
         if (req.query.until_block === undefined) {
-          const delta = await fastify.db.getPrincipalMempoolStxBalanceDelta(sql, stxAddress);
-          mempoolBalance = stxBalanceResult.balance + delta;
+          const pending = await fastify.db.getPrincipalMempoolStxBalanceDelta(sql, stxAddress);
+          mempoolInbound = pending.inbound;
+          mempoolOutbound = pending.outbound;
+          mempoolBalance = stxBalanceResult.balance + pending.delta;
         }
         const result: AddressStxBalance = {
           balance: stxBalanceResult.balance.toString(),
           estimated_balance: mempoolBalance?.toString(),
+          pending_balance_inbound: mempoolInbound?.toString(),
+          pending_balance_outbound: mempoolOutbound?.toString(),
           total_sent: stxBalanceResult.totalSent.toString(),
           total_received: stxBalanceResult.totalReceived.toString(),
           total_fees_sent: stxBalanceResult.totalFeesSent.toString(),
@@ -211,15 +217,21 @@ export const AddressRoutes: FastifyPluginAsync<
         });
 
         let mempoolBalance: bigint | undefined = undefined;
+        let mempoolInbound: bigint | undefined = undefined;
+        let mempoolOutbound: bigint | undefined = undefined;
         if (req.query.until_block === undefined) {
-          const delta = await fastify.db.getPrincipalMempoolStxBalanceDelta(sql, stxAddress);
-          mempoolBalance = stxBalanceResult.balance + delta;
+          const pending = await fastify.db.getPrincipalMempoolStxBalanceDelta(sql, stxAddress);
+          mempoolInbound = pending.inbound;
+          mempoolOutbound = pending.outbound;
+          mempoolBalance = stxBalanceResult.balance + pending.delta;
         }
 
         const result: AddressBalance = {
           stx: {
             balance: stxBalanceResult.balance.toString(),
             estimated_balance: mempoolBalance?.toString(),
+            pending_balance_inbound: mempoolInbound?.toString(),
+            pending_balance_outbound: mempoolOutbound?.toString(),
             total_sent: stxBalanceResult.totalSent.toString(),
             total_received: stxBalanceResult.totalReceived.toString(),
             total_fees_sent: stxBalanceResult.totalFeesSent.toString(),

src/api/schemas/entities/balances.ts

@@ -26,6 +26,16 @@ export const StxBalanceSchema = Type.Object(
         description: 'Total STX balance considering pending mempool transactions',
       })
     ),
+    pending_balance_inbound: Type.Optional(
+      Type.String({
+        description: 'Inbound STX balance from pending mempool transactions',
+      })
+    ),
+    pending_balance_outbound: Type.Optional(
+      Type.String({
+        description: 'Outbound STX balance from pending mempool transactions',
+      })
+    ),
     total_sent: Type.String(),
     total_received: Type.String(),
     total_fees_sent: Type.String(),

src/datastore/pg-store.ts

@@ -2480,8 +2480,8 @@ export class PgStore extends BasePgStore {
    * Returns the total STX balance delta affecting a principal from transactions currently in the
    * mempool.
    */
-  async getPrincipalMempoolStxBalanceDelta(sql: PgSqlClient, principal: string): Promise<bigint> {
-    const results = await sql<{ delta: string }[]>`
+  async getPrincipalMempoolStxBalanceDelta(sql: PgSqlClient, principal: string) {
+    const results = await sql<{ inbound: string; outbound: string; delta: string }[]>`
       WITH sent AS (
         SELECT SUM(COALESCE(token_transfer_amount, 0) + fee_rate) AS total
         FROM mempool_txs
@@ -2496,14 +2496,19 @@ export class PgStore extends BasePgStore {
         SELECT SUM(COALESCE(token_transfer_amount, 0)) AS total
         FROM mempool_txs
         WHERE pruned = false AND token_transfer_recipient_address = ${principal}
+      ),
+      values AS (
+        SELECT
+          COALESCE((SELECT total FROM received), 0) AS inbound,
+          COALESCE((SELECT total FROM sent), 0) + COALESCE((SELECT total FROM sponsored), 0) AS outbound
       )
-      SELECT
-        COALESCE((SELECT total FROM received), 0)
-        - COALESCE((SELECT total FROM sent), 0)
-        - COALESCE((SELECT total FROM sponsored), 0)
-        AS delta
+      SELECT inbound, outbound, (inbound - outbound) AS delta FROM values
     `;
-    return BigInt(results[0]?.delta ?? '0');
+    return {
+      inbound: BigInt(results[0]?.inbound ?? '0'),
+      outbound: BigInt(results[0]?.outbound ?? '0'),
+      delta: BigInt(results[0]?.delta ?? '0'),
+    };
   }
 
   async getUnlockedStxSupply(
@@ -4670,4 +4675,14 @@ export class PgStore extends BasePgStore {
       tx_total_size: parseInt(result[0].tx_total_size),
     };
   }
+
+  /// Returns timestamps for the last Stacks node events received by the API Event Server, grouped
+  /// by event type.
+  async getLastStacksNodeEventTimestamps() {
+    return await this.sql<{ event_path: string; receive_timestamp: Date }[]>`
+      SELECT DISTINCT ON (event_path) event_path, receive_timestamp
+      FROM event_observer_requests
+      ORDER BY event_path, receive_timestamp DESC
+    `;
+  }
 }

src/event-stream/event-server.ts

@@ -639,6 +639,7 @@ function createMessageProcessorQueue(db: PgWriteStore): EventMessageHandler {
   let metrics:
     | {
         eventTimer: prom.Histogram;
+        lastEventTimestamps: prom.Gauge;
         blocksInPreviousBurnBlock: prom.Gauge;
       }
     | undefined;
@@ -650,6 +651,18 @@ function createMessageProcessorQueue(db: PgWriteStore): EventMessageHandler {
         labelNames: ['event'],
         buckets: prom.exponentialBuckets(50, 3, 10), // 10 buckets, from 50 ms to 15 minutes
       }),
+      lastEventTimestamps: new prom.Gauge({
+        name: 'stacks_last_event_timestamps',
+        help: 'Last Stacks node events received timestamp',
+        labelNames: ['event'] as const,
+        async collect() {
+          const events = await db.getLastStacksNodeEventTimestamps();
+          this.reset();
+          for (const event of events) {
+            this.set({ event: event.event_path }, event.receive_timestamp.getTime());
+          }
+        },
+      }),
       blocksInPreviousBurnBlock: new prom.Gauge({
         name: 'stacks_blocks_in_previous_burn_block',
         help: 'Number of Stacks blocks produced in the previous burn block',

tests/api/address.test.ts

@@ -1551,6 +1551,8 @@ describe('address tests', () => {
       stx: {
         balance: '88679',
         estimated_balance: '88679',
+        pending_balance_inbound: '0',
+        pending_balance_outbound: '0',
         total_sent: '6385',
         total_received: '100000',
         total_fees_sent: '4936',
@@ -1601,6 +1603,8 @@ describe('address tests', () => {
       stx: {
         balance: '91',
         estimated_balance: '91',
+        pending_balance_inbound: '0',
+        pending_balance_outbound: '0',
         total_sent: '15',
         total_received: '1350',
         total_fees_sent: '1244',
@@ -1637,6 +1641,8 @@ describe('address tests', () => {
     const expectedStxResp1 = {
       balance: '91',
       estimated_balance: '91',
+      pending_balance_inbound: '0',
+      pending_balance_outbound: '0',
       total_sent: '15',
       total_received: '1350',
       total_fees_sent: '1244',
@@ -1668,6 +1674,8 @@ describe('address tests', () => {
     const expectedStxResp1Sponsored = {
       balance: '3766',
       estimated_balance: '3766',
+      pending_balance_inbound: '0',
+      pending_balance_outbound: '0',
       total_sent: '0',
       total_received: '5000',
       total_fees_sent: '1234',

tests/api/mempool.test.ts

@@ -2208,6 +2208,8 @@ describe('mempool tests', () => {
     const balance0 = await supertest(api.server).get(url);
     expect(balance0.body.balance).toEqual('2000');
     expect(balance0.body.estimated_balance).toEqual('2000');
+    expect(balance0.body.pending_balance_inbound).toEqual('0');
+    expect(balance0.body.pending_balance_outbound).toEqual('0');
 
     // STX transfer in mempool
     await db.updateMempoolTxs({
@@ -2223,6 +2225,8 @@ describe('mempool tests', () => {
     const balance1 = await supertest(api.server).get(url);
     expect(balance1.body.balance).toEqual('2000');
     expect(balance1.body.estimated_balance).toEqual('1850'); // Minus amount and fee
+    expect(balance1.body.pending_balance_inbound).toEqual('0');
+    expect(balance1.body.pending_balance_outbound).toEqual('150');
 
     // Contract call in mempool
     await db.updateMempoolTxs({
@@ -2242,6 +2246,8 @@ describe('mempool tests', () => {
     const balance1b = await supertest(api.server).get(url);
     expect(balance1b.body.balance).toEqual('2000');
     expect(balance1b.body.estimated_balance).toEqual('1800'); // Minus fee
+    expect(balance1b.body.pending_balance_inbound).toEqual('0');
+    expect(balance1b.body.pending_balance_outbound).toEqual('200');
 
     // Sponsored tx in mempool
     await db.updateMempoolTxs({
@@ -2258,6 +2264,8 @@ describe('mempool tests', () => {
     const balance2 = await supertest(api.server).get(url);
     expect(balance2.body.balance).toEqual('2000');
     expect(balance2.body.estimated_balance).toEqual('1750'); // Minus fee
+    expect(balance2.body.pending_balance_inbound).toEqual('0');
+    expect(balance2.body.pending_balance_outbound).toEqual('250');
 
     // STX received in mempool
     await db.updateMempoolTxs({
@@ -2273,6 +2281,8 @@ describe('mempool tests', () => {
     const balance3 = await supertest(api.server).get(url);
     expect(balance3.body.balance).toEqual('2000');
     expect(balance3.body.estimated_balance).toEqual('1850'); // Plus amount
+    expect(balance3.body.pending_balance_inbound).toEqual('100');
+    expect(balance3.body.pending_balance_outbound).toEqual('250');
 
     // Confirm all txs
     await db.update(
@@ -2317,5 +2327,7 @@ describe('mempool tests', () => {
     const balance4 = await supertest(api.server).get(url);
     expect(balance4.body.balance).toEqual('1850');
     expect(balance4.body.estimated_balance).toEqual('1850');
+    expect(balance4.body.pending_balance_inbound).toEqual('0');
+    expect(balance4.body.pending_balance_outbound).toEqual('0');
   });
 });

tests/api/tx.test.ts

@@ -1031,6 +1031,8 @@ describe('tx tests', () => {
     const expectedSponsoredRespBefore = {
       balance: '0',
       estimated_balance: '0',
+      pending_balance_inbound: '0',
+      pending_balance_outbound: '0',
       total_sent: '0',
       total_received: '0',
       total_fees_sent: '0',
@@ -1139,6 +1141,8 @@ describe('tx tests', () => {
     const expectedResp = {
       balance: '0',
       estimated_balance: '0',
+      pending_balance_inbound: '0',
+      pending_balance_outbound: '0',
       total_sent: '0',
       total_received: '0',
       total_fees_sent: '0',
@@ -1158,6 +1162,8 @@ describe('tx tests', () => {
       stx: {
         balance: '0',
         estimated_balance: '0',
+        pending_balance_inbound: '0',
+        pending_balance_outbound: '0',
         total_sent: '0',
         total_received: '0',
         total_fees_sent: '0',
@@ -1181,6 +1187,8 @@ describe('tx tests', () => {
     const expectedSponsoredRespAfter = {
       balance: '-300',
       estimated_balance: '-300',
+      pending_balance_inbound: '0',
+      pending_balance_outbound: '0',
       total_sent: '0',
       total_received: '0',
       total_fees_sent: '300',