Merge pull request #84 from autonomys/add-step-size

Add step size in object mapping recovery

Author: clostao

common/rpc-apis/src/objectMappingIndexer.ts

@@ -21,6 +21,7 @@ export const ObjectMappingIndexerRPCApi = createApiDefinition({
     subscribe_recover_object_mappings: {
       params: z.object({
         pieceIndex: z.number(),
+        step: z.number().optional(),
       }),
       returns: z.object({
         subscriptionId: z.string(),

services/object-mapping-indexer/src/config.ts

@@ -7,6 +7,7 @@ export const config = {
   corsAllowOrigins: env('CORS_ALLOW_ORIGINS', ''),
   nodeRpcUrl: env('NODE_RPC_URL'),
   recoveryInterval: notNaN(Number(env('RECOVERY_INTERVAL', '1000'))),
+  maxRecoveryStep: notNaN(Number(env('MAX_RECOVERY_STEP', '1000'))),
   logLevel: env(
     'LOG_LEVEL',
     process.env.NODE_ENV === 'production' ? 'info' : 'debug',

services/object-mapping-indexer/src/repositories/objectMapping.ts

@@ -63,10 +63,22 @@ const getByHash = async (hash: string) => {
   return result.rows.at(0)
 }
 
+const getByPieceIndexRange = async (min: number, max: number) => {
+  const db = await getDatabase()
+
+  const result = await db.query<DBObjectMapping>(
+    'SELECT * FROM object_mappings WHERE "pieceIndex" >= $1 AND "pieceIndex" < $2',
+    [min, max],
+  )
+
+  return result.rows
+}
+
 export const objectMappingRepository = {
   saveObjectMappings,
   getByBlockNumber,
   getLatestBlockNumber,
   getByPieceIndex,
   getByHash,
+  getByPieceIndexRange,
 }

services/object-mapping-indexer/src/rpc/server.ts

@@ -23,13 +23,14 @@ const createObjectMappingsRPCServer = (app: Application) => {
         }
       },
       subscribe_recover_object_mappings: async (
-        { pieceIndex },
+        { pieceIndex, step },
         { connection },
       ) => {
         const subscriptionId =
           objectMappingRouter.subscribeRecoverObjectMappings(
             connection,
             pieceIndex,
+            step,
           )
 
         return { success: true, subscriptionId }

services/object-mapping-indexer/src/services/objectMappingRouter/index.ts

@@ -11,7 +11,7 @@ type RouterState = {
   objectMappingsSubscriptions: Map<string, Websocket.connection>
   recoverObjectMappingsSubscriptions: Map<
     string,
-    { connection: Websocket.connection; pieceIndex: number }
+    { connection: Websocket.connection; pieceIndex: number; step: number }
   >
   lastRealtimeBlockNumber: number
 }
@@ -71,6 +71,7 @@ const emitObjectMappings = (event: ObjectMappingListEntry) => {
 const subscribeRecoverObjectMappings = (
   connection: Websocket.connection,
   startingPieceIndex: number,
+  step: number = 1,
 ) => {
   logger.info(
     `IP (${connection.remoteAddress}) subscribing to recover object mappings`,
@@ -80,6 +81,7 @@ const subscribeRecoverObjectMappings = (
   state.recoverObjectMappingsSubscriptions.set(subscriptionId, {
     connection,
     pieceIndex,
+    step,
   })
 
   return subscriptionId
@@ -104,13 +106,16 @@ const emitRecoverObjectMappings = async () => {
   )
 
   const promises = recovering.map(
-    async ([subscriptionId, { connection, pieceIndex }]) => {
-      const result =
-        await objectMappingUseCase.getObjectByPieceIndex(pieceIndex)
+    async ([subscriptionId, { connection, pieceIndex, step }]) => {
+      const result = await objectMappingUseCase.getObjectByPieceIndexAndStep(
+        pieceIndex,
+        step,
+      )
 
       state.recoverObjectMappingsSubscriptions.set(subscriptionId, {
         connection,
-        pieceIndex: pieceIndex + 1,
+        pieceIndex: pieceIndex + step,
+        step,
       })
 
       if (pieceIndex >= state.lastRealtimeBlockNumber) {

services/object-mapping-indexer/src/useCases/objectMapping.ts

@@ -50,6 +50,18 @@ const getObjectByPieceIndex = async (
   return objectMappings.map((e) => [e.hash, e.pieceIndex, e.pieceOffset])
 }
 
+const getObjectByPieceIndexAndStep = async (
+  pieceIndex: number,
+  step: number,
+): Promise<ObjectMapping[]> => {
+  const objectMappings = await objectMappingRepository.getByPieceIndexRange(
+    pieceIndex,
+    pieceIndex + step,
+  )
+
+  return objectMappings.map((e) => [e.hash, e.pieceIndex, e.pieceOffset])
+}
+
 const getObjectByBlock = async (
   blockNumber: number,
 ): Promise<ObjectMapping[]> => {
@@ -64,4 +76,5 @@ export const objectMappingUseCase = {
   getObject,
   getObjectByPieceIndex,
   getObjectByBlock,
+  getObjectByPieceIndexAndStep,
 }