Merge pull request #533 from autonomys/perf-improvements

feat: speed up publish-nodes queue with node indexes + safe batch updates

Author: jfrank-summit

apps/backend/__tests__/unit/OnchainPublisher.spec.ts

@@ -42,11 +42,9 @@ describe('OnchainPublisher', () => {
       .spyOn(transactionManager, 'submit')
       .mockResolvedValue(nodes.map(() => MOCK_PUBLISH_RESULT))
 
-    jest.spyOn(nodesRepository, 'getNodeCount').mockResolvedValue({
-      totalCount: 0,
-      publishedCount: 0,
-      archivedCount: 0,
-    })
+    jest
+      .spyOn(nodesRepository, 'getNodesBlockchainDataBatch')
+      .mockResolvedValue([])
 
     await OnchainPublisher.publishNodes(nodes.map((e) => e.cid))
 
@@ -76,21 +74,17 @@ describe('OnchainPublisher', () => {
       .spyOn(transactionManager, 'submit')
       .mockResolvedValue(nodes.map(() => MOCK_PUBLISH_RESULT))
 
-    jest.spyOn(nodesRepository, 'getNodeCount').mockImplementation((cid) => {
-      if (publishedNodes.includes(cid.cid!)) {
-        return Promise.resolve({
-          totalCount: 1,
-          publishedCount: 1,
-          archivedCount: 0,
-        })
-      }
-
-      return Promise.resolve({
-        totalCount: 1,
-        publishedCount: 0,
-        archivedCount: 0,
-      })
-    })
+    jest
+      .spyOn(nodesRepository, 'getNodesBlockchainDataBatch')
+      .mockResolvedValue(
+        publishedNodes.map((cid) => ({
+          cid,
+          block_published_on: 1,
+          tx_published_on: '0x123',
+          piece_index: 0,
+          piece_offset: 0,
+        })),
+      )
     await OnchainPublisher.publishNodes(nodes.map((e) => e.cid))
 
     const transactions = nodes

apps/backend/migrations/20251118215346-add-nodes-cid-indexes.js

@@ -0,0 +1,59 @@
+'use strict'
+
+var dbm
+var type
+var seed
+var fs = require('fs')
+var path = require('path')
+var Promise
+
+/**
+ * We receive the dbmigrate dependency from dbmigrate initially.
+ * This enables us to not have to rely on NODE_PATH.
+ */
+exports.setup = function (options, seedLink) {
+  dbm = options.dbmigrate
+  type = dbm.dataType
+  seed = seedLink
+  Promise = options.Promise
+}
+
+exports.up = function (db) {
+  var filePath = path.join(
+    __dirname,
+    'sqls',
+    '20251118215346-add-nodes-cid-indexes-up.sql',
+  )
+  return new Promise(function (resolve, reject) {
+    fs.readFile(filePath, { encoding: 'utf-8' }, function (err, data) {
+      if (err) return reject(err)
+      console.log('received data: ' + data)
+
+      resolve(data)
+    })
+  }).then(function (data) {
+    return db.runSql(data)
+  })
+}
+
+exports.down = function (db) {
+  var filePath = path.join(
+    __dirname,
+    'sqls',
+    '20251118215346-add-nodes-cid-indexes-down.sql',
+  )
+  return new Promise(function (resolve, reject) {
+    fs.readFile(filePath, { encoding: 'utf-8' }, function (err, data) {
+      if (err) return reject(err)
+      console.log('received data: ' + data)
+
+      resolve(data)
+    })
+  }).then(function (data) {
+    return db.runSql(data)
+  })
+}
+
+exports._meta = {
+  version: 1,
+}

apps/backend/migrations/sqls/20251118215346-add-nodes-cid-indexes-down.sql

@@ -0,0 +1,4 @@
+-- Remove indexes added in up migration
+DROP INDEX IF EXISTS idx_nodes_cid;
+DROP INDEX IF EXISTS idx_nodes_root_cid_cid;
+

apps/backend/migrations/sqls/20251118215346-add-nodes-cid-indexes-up.sql

@@ -0,0 +1,10 @@
+-- Add indexes for nodes table to improve query performance
+-- These indexes optimize queries that filter by cid and (root_cid, cid)
+
+-- Index for cid lookups (used in getNodesByCids, getNodeCount, getNodeBlockchainData)
+CREATE INDEX IF NOT EXISTS idx_nodes_cid ON public.nodes USING btree (cid);
+
+-- Composite index for UPDATE queries that filter by root_cid AND cid
+-- Used in updateNodeBlockchainData
+CREATE INDEX IF NOT EXISTS idx_nodes_root_cid_cid ON public.nodes USING btree (root_cid, cid);
+

apps/backend/src/core/objects/nodes.ts

@@ -272,31 +272,40 @@ const ensureObjectPublished = async (cid: string): Promise<void> => {
 }
 
 const handleRepeatedNodes = async (nodes: Node[]): Promise<void> => {
+  if (nodes.length === 0) return
+
   logger.info(
     'Handling repeated nodes (cids=%s)',
     nodes.map((node) => node.cid).join(', '),
   )
 
-  const nodesWithBlockchainData = await Promise.all(
-    nodes.map((node) => nodesRepository.getNodeBlockchainData(node.cid)),
-  ).then((e) => e.filter((e) => e !== undefined).map((e) => e!))
+  // Optimize: Batch fetch blockchain data for all unique CIDs
+  const uniqueCids = Array.from(new Set(nodes.map((node) => node.cid)))
+  const nodesWithBlockchainData =
+    await nodesRepository.getNodesBlockchainDataBatch(uniqueCids)
 
   const nodeBlockchainDataMap = new Map(
     nodesWithBlockchainData.map((e) => [e.cid, e]),
   )
 
-  const updatedNodes = nodes.map((node) => {
-    const nodeBlockchainData = nodeBlockchainDataMap.get(node.cid)
-    if (!nodeBlockchainData) return node
+  // Prepare batch updates - only update nodes that have blockchain data
+  const updates = nodes
+    .map((node) => {
+      const nodeBlockchainData = nodeBlockchainDataMap.get(node.cid)
+      if (!nodeBlockchainData) return null
 
-    return { ...node, ...nodeBlockchainData }
-  })
+      return {
+        rootCid: node.root_cid,
+        cid: node.cid,
+        blockchainData: nodeBlockchainData,
+      }
+    })
+    .filter((u): u is NonNullable<typeof u> => u !== null)
 
-  await Promise.all(
-    updatedNodes.map((node) =>
-      nodesRepository.updateNodeBlockchainData(node.root_cid, node.cid, node),
-    ),
-  )
+  // Optimize with batch update
+  if (updates.length > 0) {
+    await nodesRepository.updateNodesBlockchainDataBatch(updates)
+  }
 }
 
 export const NodesUseCases = {

apps/backend/src/infrastructure/repositories/objects/nodes.ts

@@ -253,12 +253,27 @@ const getNodeBlockchainData = async (
 
   return db
     .query<NodeBlockchainData>({
-      text: 'SELECT cid, block_published_on, tx_published_on, piece_index, piece_offset FROM nodes WHERE cid = $1',
+      text: 'SELECT cid, block_published_on, tx_published_on, piece_index, piece_offset FROM nodes WHERE cid = $1 AND block_published_on IS NOT NULL',
       values: [cid],
     })
     .then((e) => e.rows.at(0))
 }
 
+const getNodesBlockchainDataBatch = async (
+  cids: string[],
+): Promise<NodeBlockchainData[]> => {
+  if (cids.length === 0) return []
+
+  const db = await getDatabase()
+
+  return db
+    .query<NodeBlockchainData>({
+      text: 'SELECT DISTINCT ON (cid) cid, block_published_on, tx_published_on, piece_index, piece_offset FROM nodes WHERE cid = ANY($1) AND block_published_on IS NOT NULL ORDER BY cid, block_published_on DESC',
+      values: [cids],
+    })
+    .then((e) => e.rows)
+}
+
 const updateNodeBlockchainData = async (
   rootCid: string,
   cid: string,
@@ -278,6 +293,54 @@ const updateNodeBlockchainData = async (
   })
 }
 
+const updateNodesBlockchainDataBatch = async (
+  updates: Array<{
+    rootCid: string
+    cid: string
+    blockchainData: NodeBlockchainData
+  }>,
+) => {
+  if (updates.length === 0) return
+
+  const db = await getDatabase()
+
+  // Use a single query with VALUES to batch update
+  // Group by (rootCid, cid) to avoid duplicates
+  const uniqueUpdates = Array.from(
+    new Map(updates.map((u) => [`${u.rootCid}:${u.cid}`, u])).values(),
+  )
+
+  const payload = uniqueUpdates.map((u) => ({
+    block_published_on: u.blockchainData.block_published_on,
+    tx_published_on: u.blockchainData.tx_published_on,
+    piece_index: u.blockchainData.piece_index,
+    piece_offset: u.blockchainData.piece_offset,
+    root_cid: u.rootCid,
+    cid: u.cid,
+  }))
+
+  return db.query({
+    text: `
+      UPDATE nodes AS n
+      SET
+        block_published_on = data.block_published_on,
+        tx_published_on = data.tx_published_on,
+        piece_index = data.piece_index,
+        piece_offset = data.piece_offset
+      FROM jsonb_to_recordset($1::jsonb) AS data(
+        block_published_on INTEGER,
+        tx_published_on TEXT,
+        piece_index INTEGER,
+        piece_offset INTEGER,
+        root_cid TEXT,
+        cid TEXT
+      )
+      WHERE n.root_cid = data.root_cid AND n.cid = data.cid
+    `,
+    values: [JSON.stringify(payload)],
+  })
+}
+
 const hasEncodedNode = async (cid: string) => {
   const db = await getDatabase()
   return db
@@ -305,6 +368,8 @@ export const nodesRepository = {
   removeNodeByRootCid,
   getNodesCountWithoutDataByRootCid,
   getNodeBlockchainData,
+  getNodesBlockchainDataBatch,
   updateNodeBlockchainData,
+  updateNodesBlockchainDataBatch,
   hasEncodedNode,
 }

apps/backend/src/infrastructure/services/upload/onchainPublisher/index.ts

@@ -2,7 +2,7 @@ import { createLogger } from '../../../drivers/logger.js'
 import { NodesUseCases } from '../../../../core/objects/nodes.js'
 import { createTransactionManager } from './transactionManager.js'
 import { compactAddLength } from '@polkadot/util'
-import { nodesRepository, Node } from '../../../repositories/objects/nodes.js'
+import { nodesRepository } from '../../../repositories/objects/nodes.js'
 
 const logger = createLogger('upload:onchainPublisher')
 
@@ -12,22 +12,19 @@ const publishNodes = async (cids: string[]) => {
   logger.info('Uploading %d nodes', cids.length)
 
   const nodes = await nodesRepository.getNodesByCids(cids)
-  const nodeCounts = await Promise.all(
-    nodes.map((e) =>
-      nodesRepository.getNodeCount({
-        cid: e.cid,
-      }),
-    ),
-  )
 
-  const filter = (node: Node, index: number) => {
-    return nodeCounts[index] && (nodeCounts[index].publishedCount ?? 0) === 0
-  }
+  // Optimize: Batch-check which CIDs are already published anywhere in the table
+  // We consider a node "repeated" if there exists ANY row with the same CID that has blockchain data
+  const uniqueCids = Array.from(new Set(nodes.map((n) => n.cid)))
+  const publishedRows =
+    await nodesRepository.getNodesBlockchainDataBatch(uniqueCids)
+  const publishedCidSet = new Set(publishedRows.map((r) => r.cid))
 
-  const repeatedNodes = nodes.filter((_, index) => !filter(_, index))
+  const repeatedNodes = nodes.filter((node) => publishedCidSet.has(node.cid))
   await NodesUseCases.handleRepeatedNodes(repeatedNodes)
 
-  const publishingNodes = nodes.filter(filter)
+  // Nodes that are not known as published anywhere should be sent for publishing
+  const publishingNodes = nodes.filter((node) => !publishedCidSet.has(node.cid))
 
   const transactions = publishingNodes.map((node) => {
     const buffer = Buffer.from(node.encoded_node, 'base64')