Introduce efficient slot pagination

dcSpark · Dec 27, 2023 · d9b9e31 · d9b9e31
1 parent 97c8d4a
commit d9b9e31
Show file tree

Hide file tree

Showing 16 changed files with 211 additions and 92 deletions.
diff --git a/indexer/tasks/src/multiera/multiera_asset_utxo.rs b/indexer/tasks/src/multiera/multiera_asset_utxo.rs
@@ -1,5 +1,5 @@
 use super::multiera_stake_credentials::MultieraStakeCredentialTask;
-use crate::config::AddressConfig::PayloadAndReadonlyConfig;
+use crate::config::AddressConfig::AddressConfig;
 use crate::config::EmptyConfig::EmptyConfig;
 use crate::config::ReadonlyConfig::ReadonlyConfig;
 use crate::dsl::task_macro::*;

diff --git a/webserver/server/app/controllers/DelegationForPoolController.ts b/webserver/server/app/controllers/DelegationForPoolController.ts
@@ -6,8 +6,11 @@ import { genErrorMessage, type ErrorShape, Errors } from '../../../shared/errors
 import type { EndpointTypes } from '../../../shared/routes';
 import { Routes } from '../../../shared/routes';
 import { delegationsForPool } from '../services/DelegationForPool';
-import type { DelegationForPoolResponse } from '../../../shared/models/DelegationForPool';
-import { POOL_DELEGATION_LIMIT } from '../../../shared/constants';
+import type {
+  DelegationForPoolResponse,
+  DelegationForPoolSingleResponse
+} from '../../../shared/models/DelegationForPool';
+import {POOL_DELEGATION_LIMIT, PROJECTED_NFT_LIMIT} from '../../../shared/constants';
 
 const route = Routes.delegationForPool;
 
@@ -35,22 +38,31 @@ export class DelegationForPoolController extends Controller {
       );
     }
 
-    const slotRangeSize = requestBody.range.maxSlot - requestBody.range.minSlot;
-    if (slotRangeSize > POOL_DELEGATION_LIMIT.SLOT_RANGE) {
+    const after = requestBody.after != undefined ? requestBody.after : 0;
+    const until = requestBody.untilSlot != undefined ? requestBody.untilSlot : Number.MAX_VALUE;
+    const limit = requestBody.limit != undefined ? requestBody.limit : PROJECTED_NFT_LIMIT.MAX_LIMIT;
+
+    if (limit > POOL_DELEGATION_LIMIT.MAX_LIMIT) {
       // eslint-disable-next-line @typescript-eslint/no-unsafe-return
       return errorResponse(
-        StatusCodes.BAD_REQUEST,
-        genErrorMessage(Errors.SlotRangeLimitExceeded, {
-          limit: POOL_DELEGATION_LIMIT.SLOT_RANGE,
-          found: slotRangeSize,
-        })
+          StatusCodes.BAD_REQUEST,
+          genErrorMessage(Errors.SlotRangeLimitExceeded, {
+            limit: POOL_DELEGATION_LIMIT.MAX_LIMIT,
+            found: limit,
+          })
       );
     }
 
-    const response = await tx<DelegationForPoolResponse>(pool, async dbTx => {
+    let params = {
+      afterSlot: after,
+      untilSlot: until,
+      limit: limit
+    };
+
+    const result = await tx<DelegationForPoolSingleResponse[]>(pool, async dbTx => {
       const data = await delegationsForPool({
         pools: requestBody.pools.map(poolId => Buffer.from(poolId, 'hex')),
-        range: requestBody.range,
+        params: params,
         dbTx,
       });
 
@@ -62,6 +74,14 @@ export class DelegationForPoolController extends Controller {
       }));
     });
 
-    return response;
-  }
+    let newAfter = undefined;
+
+    if (result.length >= params.limit) {
+      newAfter = result[result.length - 1].slot;
+    }
+
+    return {
+      result: result,
+      after: newAfter,
+    };  }
 }
diff --git a/webserver/server/app/controllers/ProjectedNftRangeController.ts b/webserver/server/app/controllers/ProjectedNftRangeController.ts
@@ -6,7 +6,11 @@ import type { ErrorShape } from '../../../shared/errors';
 import type { EndpointTypes } from '../../../shared/routes';
 import { Routes } from '../../../shared/routes';
 import { projectedNftRange, projectedNftRangeByAddress } from '../services/ProjectedNftRange';
-import type {ProjectedNftRangeResponse, ProjectedNftStatus} from '../../../shared/models/ProjectedNftRange';
+import type {
+    ProjectedNftRangeResponse,
+    ProjectedNftRangeSingleResponse,
+    ProjectedNftStatus
+} from '../../../shared/models/ProjectedNftRange';
 import {PROJECTED_NFT_LIMIT} from "../../../shared/constants";
 import {Errors, genErrorMessage} from "../../../shared/errors";
 
@@ -25,45 +29,42 @@ export class ProjectedNftRangeController extends Controller {
             ErrorShape
         >
     ): Promise<EndpointTypes[typeof route]['response']> {
-        const slotRangeSize = requestBody.range.maxSlot - requestBody.range.minSlot;
+        const after = requestBody.after != undefined ? requestBody.after : 0;
+        const until = requestBody.untilSlot != undefined ? requestBody.untilSlot : Number.MAX_VALUE;
+        const limit = requestBody.limit != undefined ? requestBody.limit : PROJECTED_NFT_LIMIT.MAX_LIMIT;
 
-        if (requestBody.address !== undefined) {
-            if (slotRangeSize > PROJECTED_NFT_LIMIT.SINGLE_USER_SLOT_RANGE) {
-                // eslint-disable-next-line @typescript-eslint/no-unsafe-return
-                return errorResponse(
-                    StatusCodes.BAD_REQUEST,
-                    genErrorMessage(Errors.SlotRangeLimitExceeded, {
-                        limit: PROJECTED_NFT_LIMIT.SINGLE_USER_SLOT_RANGE,
-                        found: slotRangeSize,
-                    })
-                );
-            }
+        if (limit > PROJECTED_NFT_LIMIT.MAX_LIMIT) {
+            // eslint-disable-next-line @typescript-eslint/no-unsafe-return
+            return errorResponse(
+                StatusCodes.BAD_REQUEST,
+                genErrorMessage(Errors.SlotRangeLimitExceeded, {
+                    limit: PROJECTED_NFT_LIMIT.MAX_LIMIT,
+                    found: limit,
+                })
+            );
+        }
 
-            return await this.handle_by_address_query(requestBody.address, requestBody);
-        } else {
-            if (slotRangeSize > PROJECTED_NFT_LIMIT.SLOT_RANGE) {
-                // eslint-disable-next-line @typescript-eslint/no-unsafe-return
-                return errorResponse(
-                    StatusCodes.BAD_REQUEST,
-                    genErrorMessage(Errors.SlotRangeLimitExceeded, {
-                        limit: PROJECTED_NFT_LIMIT.SLOT_RANGE,
-                        found: slotRangeSize,
-                    })
-                );
-            }
+        let params = {
+            afterSlot: after,
+            untilSlot: until,
+            limit: limit
+        };
 
-            return await this.handle_general_query(requestBody);
+        if (requestBody.address !== undefined) {
+            return await this.handle_by_address_query(requestBody.address, params);
+        } else {
+            return await this.handle_general_query(params);
         }
     }
 
     async handle_general_query(
-        requestBody: EndpointTypes[typeof route]['input'],
+        params: { afterSlot: number, untilSlot: number, limit: number },
     ): Promise<EndpointTypes[typeof route]['response']> {
-        const response = await tx<
-            ProjectedNftRangeResponse
+        const result = await tx<
+            ProjectedNftRangeSingleResponse[]
         >(pool, async dbTx => {
             const data = await projectedNftRange({
-                range: requestBody.range,
+                params: params,
                 dbTx
             });
 
@@ -83,19 +84,28 @@ export class ProjectedNftRangeController extends Controller {
             }));
         });
 
-        return response;
+        let after = undefined;
+
+        if (result.length >= params.limit) {
+            after = result[result.length - 1].actionSlot;
+        }
+
+        return {
+            result: result,
+            after: after,
+        };
     }
 
     async handle_by_address_query(
         address: string,
-        requestBody: EndpointTypes[typeof route]['input'],
+        params: { afterSlot: number, untilSlot: number, limit: number },
     ): Promise<EndpointTypes[typeof route]['response']> {
-        const response = await tx<
-            ProjectedNftRangeResponse
+        const result = await tx<
+            ProjectedNftRangeSingleResponse[]
         >(pool, async dbTx => {
             const data = await projectedNftRangeByAddress({
                 address: address,
-                range: requestBody.range,
+                params: params,
                 dbTx
             });
 
@@ -115,6 +125,15 @@ export class ProjectedNftRangeController extends Controller {
             }));
         });
 
-        return response;
+        let after = undefined;
+
+        if (result.length >= params.limit) {
+            after = result[result.length - 1].actionSlot;
+        }
+
+        return {
+            result: result,
+            after: after,
+        };
     }
 }
diff --git a/webserver/server/app/models/asset/assetUtxos.queries.ts b/webserver/server/app/models/asset/assetUtxos.queries.ts
@@ -25,7 +25,7 @@ export interface IAssetUtxosQuery {
   result: IAssetUtxosResult;
 }
 
-const assetUtxosIR: any = {"usedParamSet":{"fingerprints":true,"min_slot":true,"max_slot":true},"params":[{"name":"fingerprints","required":true,"transform":{"type":"array_spread"},"locs":[{"a":677,"b":690}]},{"name":"min_slot","required":true,"transform":{"type":"scalar"},"locs":[{"a":712,"b":721}]},{"name":"max_slot","required":true,"transform":{"type":"scalar"},"locs":[{"a":744,"b":753}]}],"statement":"SELECT ENCODE(TXO.HASH,\n        'hex') OUTPUT_TX_HASH,\n    \"TransactionOutput\".OUTPUT_INDEX,\n\t\"NativeAsset\".CIP14_FINGERPRINT,\n\t\"AssetUtxo\".AMOUNT,\n\t\"Block\".SLOT,\n\tENCODE(\"Transaction\".HASH,\n        'hex') TX_HASH,\n\t\"Address\".PAYLOAD ADDRESS_RAW\nFROM \"AssetUtxo\"\nJOIN \"Transaction\" ON \"AssetUtxo\".TX_ID = \"Transaction\".ID\nJOIN \"TransactionOutput\" ON \"AssetUtxo\".UTXO_ID = \"TransactionOutput\".ID\nJOIN \"Transaction\" TXO ON \"TransactionOutput\".TX_ID = TXO.ID\nJOIN \"Address\" ON \"Address\".id = \"TransactionOutput\".address_id\nJOIN \"NativeAsset\" ON \"AssetUtxo\".ASSET_ID = \"NativeAsset\".ID\nJOIN \"Block\" ON \"Transaction\".BLOCK_ID = \"Block\".ID\nWHERE \n\t\"NativeAsset\".CIP14_FINGERPRINT IN :fingerprints! AND\n\t\"Block\".SLOT > :min_slot! AND\n\t\"Block\".SLOT <= :max_slot!\nORDER BY \"Transaction\".ID ASC"};
+const assetUtxosIR: any = {"usedParamSet":{"fingerprints":true,"min_slot":true,"max_slot":true},"params":[{"name":"fingerprints","required":true,"transform":{"type":"array_spread"},"locs":[{"a":677,"b":690}]},{"name":"min_slot","required":true,"transform":{"type":"scalar"},"locs":[{"a":712,"b":721}]},{"name":"max_slot","required":true,"transform":{"type":"scalar"},"locs":[{"a":744,"b":753}]}],"statement":"SELECT ENCODE(TXO.HASH,\n        'hex') OUTPUT_TX_HASH,\n    \"TransactionOutput\".OUTPUT_INDEX,\n\t\"NativeAsset\".CIP14_FINGERPRINT,\n\t\"AssetUtxo\".AMOUNT,\n\t\"Block\".SLOT,\n\tENCODE(\"Transaction\".HASH,\n        'hex') TX_HASH,\n\t\"Address\".PAYLOAD ADDRESS_RAW\nFROM \"AssetUtxo\"\nJOIN \"Transaction\" ON \"AssetUtxo\".TX_ID = \"Transaction\".ID\nJOIN \"TransactionOutput\" ON \"AssetUtxo\".UTXO_ID = \"TransactionOutput\".ID\nJOIN \"Transaction\" TXO ON \"TransactionOutput\".TX_ID = TXO.ID\nJOIN \"Address\" ON \"Address\".id = \"TransactionOutput\".address_id\nJOIN \"NativeAsset\" ON \"AssetUtxo\".ASSET_ID = \"NativeAsset\".ID\nJOIN \"Block\" ON \"Transaction\".BLOCK_ID = \"Block\".ID\nWHERE \n\t\"NativeAsset\".CIP14_FINGERPRINT IN :fingerprints! AND\n\t\"Block\".SLOT > :min_slot! AND\n\t\"Block\".SLOT <= :max_slot!\nORDER BY \"Transaction\".ID, \"AssetUtxo\".ID ASC"};
 
 /**
  * Query generated from SQL:
@@ -50,7 +50,7 @@ const assetUtxosIR: any = {"usedParamSet":{"fingerprints":true,"min_slot":true,"
  * 	"NativeAsset".CIP14_FINGERPRINT IN :fingerprints! AND
  * 	"Block".SLOT > :min_slot! AND
  * 	"Block".SLOT <= :max_slot!
- * ORDER BY "Transaction".ID ASC
+ * ORDER BY "Transaction".ID, "AssetUtxo".ID ASC
  * ```
  */
 export const assetUtxos = new PreparedQuery<IAssetUtxosParams,IAssetUtxosResult>(assetUtxosIR);

diff --git a/webserver/server/app/models/delegation/delegationsForPool.queries.ts b/webserver/server/app/models/delegation/delegationsForPool.queries.ts
@@ -3,6 +3,7 @@ import { PreparedQuery } from '@pgtyped/query';
 
 /** 'SqlStakeDelegationByPool' parameters type */
 export interface ISqlStakeDelegationByPoolParams {
+  limit: string;
   max_slot: number;
   min_slot: number;
   pools: readonly (Buffer)[];
@@ -22,7 +23,7 @@ export interface ISqlStakeDelegationByPoolQuery {
   result: ISqlStakeDelegationByPoolResult;
 }
 
-const sqlStakeDelegationByPoolIR: any = {"usedParamSet":{"pools":true,"min_slot":true,"max_slot":true},"params":[{"name":"pools","required":true,"transform":{"type":"array_spread"},"locs":[{"a":176,"b":182},{"a":590,"b":596},{"a":657,"b":663}]},{"name":"min_slot","required":true,"transform":{"type":"scalar"},"locs":[{"a":688,"b":697}]},{"name":"max_slot","required":true,"transform":{"type":"scalar"},"locs":[{"a":720,"b":729}]}],"statement":"SELECT \n\tencode(credential, 'hex') as credential,\n\tencode(\"Transaction\".hash, 'hex') as tx_id,\n\t\"Block\".slot,\n\tCASE WHEN \"StakeDelegationCredentialRelation\".pool_credential IN :pools! THEN encode(\"StakeDelegationCredentialRelation\".pool_credential, 'hex') ELSE NULL END AS pool\nFROM \"StakeDelegationCredentialRelation\"\nJOIN \"StakeCredential\" ON stake_credential = \"StakeCredential\".id\nJOIN \"Transaction\" ON \"Transaction\".id = \"StakeDelegationCredentialRelation\".tx_id\nJOIN \"Block\" ON \"Transaction\".block_id = \"Block\".id\nWHERE \n    (\n\t\t\"StakeDelegationCredentialRelation\".pool_credential IN :pools! OR\n\t \t\"StakeDelegationCredentialRelation\".previous_pool IN :pools!\n\t) AND\n\t\"Block\".slot > :min_slot! AND\n\t\"Block\".slot <= :max_slot!\nORDER BY (\"Block\".height, \"Transaction\".tx_index) ASC"};
+const sqlStakeDelegationByPoolIR: any = {"usedParamSet":{"pools":true,"min_slot":true,"max_slot":true,"limit":true},"params":[{"name":"pools","required":true,"transform":{"type":"array_spread"},"locs":[{"a":176,"b":182},{"a":590,"b":596},{"a":657,"b":663}]},{"name":"min_slot","required":true,"transform":{"type":"scalar"},"locs":[{"a":688,"b":697},{"a":1083,"b":1092}]},{"name":"max_slot","required":true,"transform":{"type":"scalar"},"locs":[{"a":720,"b":729},{"a":1126,"b":1135}]},{"name":"limit","required":true,"transform":{"type":"scalar"},"locs":[{"a":1151,"b":1157}]}],"statement":"SELECT \n\tencode(credential, 'hex') as credential,\n\tencode(\"Transaction\".hash, 'hex') as tx_id,\n\t\"Block\".slot,\n\tCASE WHEN \"StakeDelegationCredentialRelation\".pool_credential IN :pools! THEN encode(\"StakeDelegationCredentialRelation\".pool_credential, 'hex') ELSE NULL END AS pool\nFROM \"StakeDelegationCredentialRelation\"\nJOIN \"StakeCredential\" ON stake_credential = \"StakeCredential\".id\nJOIN \"Transaction\" ON \"Transaction\".id = \"StakeDelegationCredentialRelation\".tx_id\nJOIN \"Block\" ON \"Transaction\".block_id = \"Block\".id\nWHERE \n    (\n\t\t\"StakeDelegationCredentialRelation\".pool_credential IN :pools! OR\n\t \t\"StakeDelegationCredentialRelation\".previous_pool IN :pools!\n\t) AND\n\t\"Block\".slot > :min_slot! AND\n\t\"Block\".slot <= :max_slot!\n    AND \"Block\".height <= (\n        SELECT MAX(\"Heights\".height) FROM\n        (SELECT \"Block\".height as height FROM \"StakeDelegationCredentialRelation\"\n            JOIN \"Transaction\" ON \"Transaction\".id = \"StakeDelegationCredentialRelation\".tx_id\n            JOIN \"Block\" ON \"Transaction\".block_id = \"Block\".id\n        WHERE\n            \"Block\".slot > :min_slot!\n            AND \"Block\".slot <= :max_slot!\n        LIMIT :limit!) AS \"Heights\"\n    )\nORDER BY (\"Block\".height, \"Transaction\".tx_index) ASC"};
 
 /**
  * Query generated from SQL:
@@ -43,6 +44,16 @@ const sqlStakeDelegationByPoolIR: any = {"usedParamSet":{"pools":true,"min_slot"
  * 	) AND
  * 	"Block".slot > :min_slot! AND
  * 	"Block".slot <= :max_slot!
+ *     AND "Block".height <= (
+ *         SELECT MAX("Heights".height) FROM
+ *         (SELECT "Block".height as height FROM "StakeDelegationCredentialRelation"
+ *             JOIN "Transaction" ON "Transaction".id = "StakeDelegationCredentialRelation".tx_id
+ *             JOIN "Block" ON "Transaction".block_id = "Block".id
+ *         WHERE
+ *             "Block".slot > :min_slot!
+ *             AND "Block".slot <= :max_slot!
+ *         LIMIT :limit!) AS "Heights"
+ *     )
  * ORDER BY ("Block".height, "Transaction".tx_index) ASC
  * ```
  */

diff --git a/webserver/server/app/models/delegation/delegationsForPool.sql b/webserver/server/app/models/delegation/delegationsForPool.sql
@@ -18,4 +18,14 @@ WHERE
 	) AND
 	"Block".slot > :min_slot! AND
 	"Block".slot <= :max_slot!
+    AND "Block".height <= (
+        SELECT MAX("Heights".height) FROM
+        (SELECT "Block".height as height FROM "StakeDelegationCredentialRelation"
+            JOIN "Transaction" ON "Transaction".id = "StakeDelegationCredentialRelation".tx_id
+            JOIN "Block" ON "Transaction".block_id = "Block".id
+        WHERE
+            "Block".slot > :min_slot!
+            AND "Block".slot <= :max_slot!
+        LIMIT :limit!) AS "Heights"
+    )
 ORDER BY ("Block".height, "Transaction".tx_index) ASC;