feat(cu): reduce gql queries

jfrain99 · jfrain99 · commit cbd1c85669d0 · 2025-11-13T15:42:46.000-05:00
diff --git a/servers/cu/src/domain/lib/loadMessages.js b/servers/cu/src/domain/lib/loadMessages.js
@@ -1,6 +1,6 @@
 import { Transform } from 'node:stream'
 
-import { Rejected, Resolved, fromPromise, of } from 'hyper-async'
+import { Resolved, fromPromise, of } from 'hyper-async'
 import { T, always, ascend, cond, equals, identity, ifElse, isNil, last, length, pipe, prop, reduce, uniqBy } from 'ramda'
 import ms from 'ms'
 
@@ -9,6 +9,27 @@ import { findBlocksSchema, getLatestBlockSchema, loadBlocksMetaSchema, loadMessa
 
 export const toSeconds = (millis) => Math.floor(millis / 1000)
 
+const makeRanges = (missing, blocks) => {
+  if (missing.length === 0) return []
+  const sorted = [...missing].sort((a, b) => a - b)
+
+  const ranges = []
+  let current = [sorted[0]]
+
+  for (let i = 1; i < sorted.length; i++) {
+    if (sorted[i] === sorted[i - 1] + 1) {
+      current.push(sorted[i])
+    } else {
+      const maxTimestamp = blocks.find((block) => block.height === current[current.length - 1] + 1)?.timestamp - 1
+      ranges.push({ min: current[0], max: current[current.length - 1], maxTimestamp })
+      current = [sorted[i]]
+    }
+  }
+  const maxTimestamp = blocks.find((block) => block.height === current[current.length - 1] + 1)?.timestamp - 1
+  ranges.push({ min: current[0], max: current[current.length - 1], maxTimestamp })
+  return ranges
+}
+
 export function findMissingBlocksIn (blocks, { min, maxTimestamp }) {
   if (!blocks.length) return { min, maxTimestamp }
 
@@ -37,16 +58,8 @@ export function findMissingBlocksIn (blocks, { min, maxTimestamp }) {
     return { min: maxBlock.height, maxTimestamp }
   }
 
-  /**
-   * TODO:
-   *
-   * The purpose of this function to find the holes in the incremental sequence of block meta.
-   * This impl returns one "large" hole to fetch from the gateway.
-   *
-   * An optimization would be to split the "large" hold into more reasonably sized "small" holes,
-   * and fetch those. aka. more resolution and less data unnecessarily loaded from the gateway
-   */
-  return { min: Math.min(...missing), maxTimestamp }
+  const missingRanges = makeRanges(missing, blocks)
+  return { missingRanges, maxTimestamp }
 }
 
 export function mergeBlocks (fromDb, fromGateway) {
@@ -347,42 +360,40 @@ function reconcileBlocksWith ({ logger, loadBlocksMeta, findBlocks, saveBlocks,
            * between the minimum block, and the maxTimestamp
            */
           .map((fromDb) => findMissingBlocksIn(fromDb, { min, maxTimestamp }))
-          .chain((missingRange) => {
-            if (!missingRange) return Resolved(fromDb)
-            const latestBlocksMatch = missingRange.min === fromDb[fromDb.length - 1].height
-            if (latestBlocksMatch) {
-              logger('Latest blocks match at height %d. Checking Arweave for latest block', missingRange.min)
-              return of()
-                .chain(getLatestBlock)
-                .chain((latestBlock) => {
-                  if (latestBlock === missingRange.min) {
-                    logger('Latest block matches missing range min height %d. Bypassing GQL call', missingRange.min)
-                    return Resolved(fromDb)
-                  }
-                  logger('Latest blocks do not match (arweave: %d, db: %d). Fetching missing blocks from gateway', latestBlock, missingRange.min)
-                  return Rejected(missingRange)
-                })
-            }
-            return Rejected(missingRange)
-          })
-          .bichain((missingRange) => {
-            if (!missingRange) return Resolved(fromDb)
-            logger('Loading missing blocks within range of %j', missingRange)
+          .chain((missing) => {
+            if (!missing) return Resolved(fromDb)
 
             /**
-             * Load any missing blocks within the determined range,
-             * from the gateway
+             * Handle two cases:
+             * 1. missing.missingRanges - array of ranges with gaps in the middle
+             * 2. missing.min and missing.maxTimestamp - single range at the end
              */
-            return loadBlocksMeta(missingRange)
-              /**
-               * Cache any fetched blocks for next time
-               *
-               * This will definitely result in individual 409s for existing blocks,
-               * but those shouldn't impact anything and are swallowed
-               */
-              .chain((fromGateway) => saveBlocks(fromGateway).map(() => fromGateway))
-              .map((fromGateway) => mergeBlocks(fromDb, fromGateway))
-          }, Resolved)
+            const ranges = missing.missingRanges || [{ min: missing.min, maxTimestamp: missing.maxTimestamp }]
+
+            /**
+             * Load blocks for each missing range sequentially,
+             * accumulating the results
+             */
+            return ranges.reduce((acc, missingRange) => {
+              return acc.chain((accumulated) => {
+                logger('Loading missing blocks within range of %j', missingRange)
+
+                /**
+                 * Load any missing blocks within the determined range,
+                 * from the gateway
+                 */
+                return loadBlocksMeta(missingRange)
+                  /**
+                   * Cache any fetched blocks for next time
+                   *
+                   * This will definitely result in individual 409s for existing blocks,
+                   * but those shouldn't impact anything and are swallowed
+                   */
+                  .chain((fromGateway) => saveBlocks(fromGateway).map(() => fromGateway))
+                  .map((fromGateway) => mergeBlocks(accumulated, fromGateway))
+              })
+            }, Resolved(fromDb))
+          })
       })
   }
 }
diff --git a/servers/cu/src/effects/ao-block.test.js b/servers/cu/src/effects/ao-block.test.js
@@ -136,6 +136,9 @@ describe('ao-block', () => {
     test('load the block data across multiple pages', async () => {
       const loadBlocksMeta = loadBlocksMetaSchema.implement(loadBlocksMetaWith({
         fetch,
+        gatewayCounter: {
+          inc: () => {}
+        },
         GRAPHQL_URLS,
         /**
          * Weird page size, so we know we are chopping off the excess
@@ -164,6 +167,9 @@ describe('ao-block', () => {
           errorCount++
           throw Error('Fetch error!')
         },
+        gatewayCounter: {
+          inc: () => {}
+        },
         GRAPHQL_URLS,
         pageSize: 17,
         logger
@@ -183,6 +189,9 @@ describe('ao-block', () => {
           assert.equal(url, GRAPHQL_URLS[errorsCount % GRAPHQL_URLS.length])
           throw Error('Fetch error!')
         },
+        gatewayCounter: {
+          inc: () => {}
+        },
         GRAPHQL_URLS,
         pageSize: 17,
         logger,
@@ -230,6 +239,9 @@ describe('ao-block', () => {
         fetch: () => {
           throw Error('Fetch error!')
         },
+        gatewayCounter: {
+          inc: () => {}
+        },
         GRAPHQL_URLS,
         pageSize: 17,
         logger,
@@ -334,6 +346,9 @@ describe('ao-block', () => {
             })
           }
         },
+        gatewayCounter: {
+          inc: () => {}
+        },
         GRAPHQL_URLS,
         pageSize: 5,
         logger,
diff --git a/servers/cu/src/effects/ao-process.test.js b/servers/cu/src/effects/ao-process.test.js
@@ -749,6 +749,9 @@ describe('ao-process', () => {
 
           return { data: { transactions: { edges } } }
         },
+        gatewayCounter: {
+          inc: () => {}
+        },
         queryCheckpointGateway: async () => assert.fail('should not call if default gateway is successful'),
         loadTransactionData: async (id) => {
           assert.ok(id.includes('tx-123-'))
@@ -1320,6 +1323,9 @@ describe('ao-process', () => {
           findRecordCheckpointBefore: async () => assert.fail('should not call if found in cache'),
           address: async () => assert.fail('should not call if found in file checkpoint'),
           queryGateway: async () => assert.fail('should not call if found in file checkpoint'),
+          gatewayCounter: {
+            inc: () => {}
+          },
           queryCheckpointGateway: async () => assert.fail('should not call if file checkpoint'),
           loadTransactionData: async (id) => {
             assert.equal(id, 'tx-123')
@@ -1361,6 +1367,9 @@ describe('ao-process', () => {
           address: async () => assert.fail('should not call if found in file checkpoint'),
           queryGateway: async () => assert.fail('should not call if found in file checkpoint'),
           queryCheckpointGateway: async () => assert.fail('should not call if file checkpoint'),
+          gatewayCounter: {
+            inc: () => {}
+          },
           loadTransactionData: async (id) => {
             assert.equal(id, 'tx-123')
             return new Response(Readable.toWeb(Readable.from(zipped)))
@@ -1417,6 +1426,9 @@ describe('ao-process', () => {
             }
           }),
           queryCheckpointGateway: async () => assert.fail('should not call if default gateway is successful'),
+          gatewayCounter: {
+            inc: () => {}
+          },
           loadTransactionData: async (id) => {
             assert.ok(id.includes('tx-123-'))
             return new Response(Readable.toWeb(Readable.from(zipped)))
@@ -1786,7 +1798,10 @@ describe('ao-process', () => {
   describe('saveCheckpointWith', () => {
     const logger = createTestLogger({ name: 'saveCheckpointWith' })
     const depsAll = {
-      logger
+      logger,
+      gatewayCounter: {
+        inc: () => {}
+      }
     }
 
     describe('should create an arweave checkpoint and file checkpoint', async () => {
diff --git a/servers/cu/src/effects/arweave.test.js b/servers/cu/src/effects/arweave.test.js
@@ -13,6 +13,9 @@ describe('arweave', () => {
     test('load transaction meta', async () => {
       const loadTransactionMeta = loadTransactionMetaSchema.implement(
         loadTransactionMetaWith({
+          gatewayCounter: {
+            inc: () => {}
+          },
           fetch: (url, options) => {
             const body = JSON.parse(options.body)
             assert.deepStrictEqual(body.variables, {
@@ -41,6 +44,9 @@ describe('arweave', () => {
       const loadTransactionMeta = loadTransactionMetaSchema.implement(
         loadTransactionMetaWith({
           GRAPHQL_URL,
+          gatewayCounter: {
+            inc: () => {}
+          },
           fetch: async (url, options) => {
             assert.equal(url, GRAPHQL_URL)
             const body = JSON.parse(options.body)
