Fix SubQL runtime image smoke failures

Author: xlc

packages/chain-stats-subql/Dockerfile

@@ -19,6 +19,8 @@ RUN set -e \
     && rm -rf /app/dist \
     && ./node_modules/.bin/subql codegen \
     && ./node_modules/.bin/subql build --output dist \
+    && mkdir -p /app/dist/data \
+    && cp /app/src/data/*.json /app/dist/data/ \
     && for i in 1 2 3 4 5; do [ -d /app/dist ] && break; sleep 1; done \
     && if [ ! -d /app/dist ]; then \
         echo "SubQL build did not create /app/dist"; \

packages/chain-stats-subql/package.json

@@ -9,9 +9,9 @@
   "private": true,
   "scripts": {
     "start": "bun run build && ./node_modules/.bin/subql-node -f . --timeout=512 --batch-size=2 --port=3123",
-    "build": "subql codegen && subql build",
+    "build": "subql codegen && subql build && mkdir -p dist/data && cp src/data/*.json dist/data/",
     "prepack": "rm -rf dist && npm build",
-    "test": "jest",
+    "test": "bun test tests/*.test.ts",
     "codegen": "./node_modules/.bin/subql codegen"
   },
   "files": [

packages/chain-stats-subql/src/index.ts

@@ -1,2 +1,4 @@
+import './runtimePolyfills';
+
 //Exports all handler functions
 export * from './mappings/mappingHandlers'

packages/chain-stats-subql/src/mappings/handleBlock.ts

@@ -1,11 +1,8 @@
 import { SubstrateBlock } from "@subql/types/dist/interfaces";
 import { getBlock, getToken } from "../utils/records";
-import acalaNative from "../data/acala-1870000-native.json";
-import acalaNonNative from "../data/acala-1870000-non-native.json";
-import karuraNative from "../data/karura-2650000-native.json";
-import karuraNonNative from "../data/karura-2650000-non-native.json";
 import { updateAccountBalance } from "../utils/updateAccountBalance";
 import { insertBefore, startAt } from "./mappingHandlers";
+import { getInitialBalanceRecords } from "../utils/balanceData";
 
 
 export async function handleBlock(block: SubstrateBlock) {
@@ -17,24 +14,16 @@ export async function handleBlock(block: SubstrateBlock) {
   record.timestamp = block.timestamp;
 
   if (blockNumber <= insertBefore) {
-    const isAcala = api.registry.chainSS58 === 10
-    let accountData = []
-
-    if (isAcala) {
-      accountData = [...acalaNative as any, ...acalaNonNative as any]
-    } else {
-      accountData = [...karuraNative as any, ...karuraNonNative as any]
-    }
-
+    const network = api.registry.chainSS58 === 10 ? "acala" : "karura";
     const round = Number(blockNumber - startAt);
+    const startIndex = round * 1000;
+    const accountData = await getInitialBalanceRecords(network, startIndex, 1000);
 
-    logger.info(`total ${accountData.length}, start insert ${round * 1000}`);
-
-    for (let i = 0; i < 1000; i++) {
-      const data = accountData[i + round * 1000] as { account: string; token: string; free: string; reserved: string; frozen: string };
+    logger.info(`start insert ${startIndex}, batch ${accountData.length}`);
 
-      if (!data) return;
+    if (accountData.length === 0) return;
 
+    for (const data of accountData) {
       const token = await getToken(data.token);
 
       await token.save();
@@ -52,4 +41,4 @@ export async function handleBlock(block: SubstrateBlock) {
 
 
   await record.save();
-}
+}

packages/chain-stats-subql/src/runtimePolyfills.ts

@@ -0,0 +1,15 @@
+import { TextDecoder, TextEncoder } from "util";
+
+export const installRuntimePolyfills = () => {
+  const runtime = globalThis as any;
+
+  if (typeof runtime.TextEncoder === "undefined") {
+    runtime.TextEncoder = TextEncoder;
+  }
+
+  if (typeof runtime.TextDecoder === "undefined") {
+    runtime.TextDecoder = TextDecoder;
+  }
+};
+
+installRuntimePolyfills();

packages/chain-stats-subql/src/utils/balanceData.ts

@@ -0,0 +1,170 @@
+import { createReadStream, existsSync } from "fs";
+import { join } from "path";
+
+export type InitialBalanceRecord = {
+  account: string;
+  token: string;
+  free: string;
+  reserved: string;
+  frozen: string;
+};
+
+type Network = "acala" | "karura";
+
+const DATA_FILES: Record<Network, string[]> = {
+  acala: ["acala-1870000-native.json", "acala-1870000-non-native.json"],
+  karura: ["karura-2650000-native.json", "karura-2650000-non-native.json"],
+};
+
+const getWorkingDirectory = () => {
+  const cwd = (globalThis as { process?: { cwd?: unknown } }).process?.cwd;
+
+  return typeof cwd === "function" ? cwd() : "/app";
+};
+
+const getDataDirCandidates = () => {
+  const cwd = getWorkingDirectory();
+
+  return [
+    join(cwd, "dist", "data"),
+    join(cwd, "src", "data"),
+    join(cwd, "data"),
+    join(cwd, "packages", "chain-stats-subql", "dist", "data"),
+    join(cwd, "packages", "chain-stats-subql", "src", "data"),
+    "/app/dist/data",
+    "/app/src/data",
+  ];
+};
+
+type ReaderState = {
+  nextIndex: number;
+  iterator: AsyncIterator<InitialBalanceRecord>;
+};
+
+const readers = new Map<Network, ReaderState>();
+
+const getDataFilePath = (fileName: string) => {
+  for (const directory of getDataDirCandidates()) {
+    const filePath = join(directory, fileName);
+
+    if (existsSync(filePath)) {
+      return filePath;
+    }
+  }
+
+  throw new Error(`Unable to locate chain-stats balance data file ${fileName}`);
+};
+
+async function* readJsonArrayObjects(filePath: string): AsyncGenerator<InitialBalanceRecord> {
+  let buffer = "";
+  let depth = 0;
+  let inString = false;
+  let escaped = false;
+  let capturing = false;
+
+  for await (const chunk of createReadStream(filePath, { encoding: "utf8" })) {
+    for (const char of chunk) {
+      if (!capturing) {
+        if (char === "{") {
+          capturing = true;
+          depth = 1;
+          buffer = char;
+        }
+
+        continue;
+      }
+
+      buffer += char;
+
+      if (escaped) {
+        escaped = false;
+        continue;
+      }
+
+      if (char === "\\") {
+        escaped = true;
+        continue;
+      }
+
+      if (char === "\"") {
+        inString = !inString;
+        continue;
+      }
+
+      if (inString) {
+        continue;
+      }
+
+      if (char === "{") {
+        depth += 1;
+      } else if (char === "}") {
+        depth -= 1;
+
+        if (depth === 0) {
+          yield JSON.parse(buffer) as InitialBalanceRecord;
+          buffer = "";
+          capturing = false;
+        }
+      }
+    }
+  }
+}
+
+async function* readNetworkRecords(network: Network): AsyncGenerator<InitialBalanceRecord> {
+  for (const fileName of DATA_FILES[network]) {
+    yield* readJsonArrayObjects(getDataFilePath(fileName));
+  }
+}
+
+const getReader = (network: Network, startIndex: number) => {
+  const current = readers.get(network);
+
+  if (current && startIndex >= current.nextIndex) {
+    return current;
+  }
+
+  const next = {
+    nextIndex: 0,
+    iterator: readNetworkRecords(network)[Symbol.asyncIterator](),
+  };
+
+  readers.set(network, next);
+  return next;
+};
+
+export const getInitialBalanceRecords = async (
+  network: Network,
+  startIndex: number,
+  limit: number,
+) => {
+  const reader = getReader(network, startIndex);
+
+  while (reader.nextIndex < startIndex) {
+    const skipped = await reader.iterator.next();
+
+    if (skipped.done) {
+      return [];
+    }
+
+    reader.nextIndex += 1;
+  }
+
+  const records: InitialBalanceRecord[] = [];
+
+  while (records.length < limit) {
+    const next = await reader.iterator.next();
+
+    if (next.done) {
+      break;
+    }
+
+    records.push(next.value);
+    reader.nextIndex += 1;
+  }
+
+  return records;
+};
+
+export const resetInitialBalanceReadersForTests = () => {
+  readers.clear();
+};

packages/chain-stats-subql/src/utils/records.ts

@@ -1,7 +1,33 @@
 import { isSystemAccount, getNativeCurrency, getTokenDecimals, isTokenEqual, getSystemAccountName } from '@acala-network/subql-utils'
+import { stringToHex, u8aToHex } from '@polkadot/util'
+import { decodeAddress } from '@polkadot/util-crypto'
 import { Block, Token, Account, AccountBalance, DailyAccountBalance, HourAccountBalance, HourToken, DailyToken } from '../types/models'
 
 const nativeToken = getNativeCurrency(api as any);
+const SYSTEM_ACCOUNT_PREFIX = stringToHex('modl');
+
+const isSystemAccountByPublicKey = (address: string) => {
+    try {
+        return u8aToHex(decodeAddress(address, true)).startsWith(SYSTEM_ACCOUNT_PREFIX)
+    } catch {
+        return false
+    }
+}
+
+export const isSystemAccountSafe = (address: string) => {
+    try {
+        return isSystemAccount(address)
+    } catch {
+        const isSystem = isSystemAccountByPublicKey(address)
+        const logger = (globalThis as any).logger
+
+        if (typeof logger?.warn === 'function') {
+            logger.warn(`Unable to classify account ${address} with checksum validation; fallback system=${isSystem}`)
+        }
+
+        return isSystem
+    }
+}
 
 export async function getBlock(id: string) {
     let record = await Block.get(id)
@@ -79,7 +105,7 @@ export async function getAccount(id: string) {
     let record = await Account.get(id)
 
     if (!record) {
-        const isSystem = isSystemAccount(id)
+        const isSystem = isSystemAccountSafe(id)
 
         record = new Account(id, id)
 

packages/chain-stats-subql/tests/account.test.ts

@@ -0,0 +1,53 @@
+import { beforeEach, expect, test } from "bun:test";
+
+const records = new Map<string, Map<string, any>>();
+
+const getEntityRecords = (entity: string) => {
+  let entityRecords = records.get(entity);
+
+  if (!entityRecords) {
+    entityRecords = new Map<string, any>();
+    records.set(entity, entityRecords);
+  }
+
+  return entityRecords;
+};
+
+beforeEach(() => {
+  records.clear();
+
+  (globalThis as any).api = {
+    consts: {
+      currencies: {
+        getNativeCurrencyId: "KAR",
+      },
+    },
+  };
+
+  (globalThis as any).store = {
+    get: async (entity: string, id: string) => getEntityRecords(entity).get(id),
+    set: async (entity: string, id: string, value: any) => {
+      getEntityRecords(entity).set(id, { ...value });
+    },
+    remove: async (entity: string, id: string) => {
+      getEntityRecords(entity).delete(id);
+    },
+    getByField: async () => [],
+    getByFields: async () => [],
+  };
+
+  (globalThis as any).logger = {
+    warn: () => undefined,
+  };
+});
+
+test("does not throw when a Karura snapshot account has an invalid checksum", async () => {
+  const { getAccount, isSystemAccountSafe } = await import("../src/utils/records");
+
+  expect(isSystemAccountSafe("qHz7aFFfQCcVcVVkzeYeNGoaDMivV9jUXLMBqSMLzFYkfmC")).toBe(false);
+
+  const account = await getAccount("qHz7aFFfQCcVcVVkzeYeNGoaDMivV9jUXLMBqSMLzFYkfmC");
+
+  expect(account.address).toBe("qHz7aFFfQCcVcVVkzeYeNGoaDMivV9jUXLMBqSMLzFYkfmC");
+  expect(account.mark).toBe("user");
+});

packages/chain-stats-subql/tests/balanceData.test.ts

@@ -0,0 +1,25 @@
+import { expect, test } from "bun:test";
+import { getInitialBalanceRecords, resetInitialBalanceReadersForTests } from "../src/utils/balanceData";
+
+test("streams only the requested Acala initial-balance batch", async () => {
+  resetInitialBalanceReadersForTests();
+
+  const records = await getInitialBalanceRecords("acala", 0, 3);
+
+  expect(records).toHaveLength(3);
+  expect(records[0].account).toBeString();
+  expect(records[0].token).toBe("ACA");
+  expect(records[0].free).toBeString();
+});
+
+test("streams Karura batches without loading the full snapshot into the bundle", async () => {
+  resetInitialBalanceReadersForTests();
+
+  const firstBatch = await getInitialBalanceRecords("karura", 0, 2);
+  const secondBatch = await getInitialBalanceRecords("karura", 2, 2);
+
+  expect(firstBatch).toHaveLength(2);
+  expect(secondBatch).toHaveLength(2);
+  expect(secondBatch[0].account).not.toBe(firstBatch[0].account);
+  expect(secondBatch[0].token).toBe("KAR");
+});