diff --git a/apps/chain-indexer/README.md b/apps/chain-indexer/README.md index 1e2d450838..0f034b3acc 100644 --- a/apps/chain-indexer/README.md +++ b/apps/chain-indexer/README.md @@ -58,6 +58,16 @@ BACKFILL_TO_HEIGHT=200000 Blocks are fetched from RPC in parallel (`BACKFILL_CONCURRENCY`, default 10) and committed strictly in order in batches of `BACKFILL_BATCH_SIZE` blocks (default 200), each batch in one Postgres transaction together with the checkpoint advance. Progress is checkpointed per range under the `indexer_state` stream `backfill:{from}-{to}`, so killing and restarting the job resumes at the checkpoint without gaps or duplicates, and re-running a completed range exits 0 immediately. Changing the range creates a fresh checkpoint row. All inserts are natural-keyed and conflict-ignoring, so a backfill can run against the same database as live sync, and a duplicate backfill pod on the same range is harmless. +## Balance ledger and activity log + +Every committed block also derives a balance ledger and an address activity log, in the same transaction as the block, so they never drift from the chain data they come from. `balance_changes` is the append-only ledger: one row per coin movement with the running `balance_after` and a classified `reason` (`mint`, `burn`, `slash`, `fee`, `reward`, `commission`, `staking`, `gov`, `ibc`, `escrow`, `bme`, or a plain `transfer`; genesis seeds are `genesis`). `account_balances` holds the current per-account per-denom balance, upserted from the ledger. `account_txs` is the activity log linking each account to the transactions that touched it. Addresses are interned to ids on first sight (`accounts`), so both live sync and backfill produce identical ledger rows for the same height. + +The reason heuristic is deliberately MVP: coincident mint/burn/slash win first, then the module account on the holder's side of the movement (falling back to the counterparty's), then the denom. Per-deployment/lease attribution of escrow movements is left for later. + +## Reconciliation + +`npm run reconcile` proves the ledger matches the chain at the `sync` checkpoint height. It samples the highest-balance accounts, compares each against the node's bank balance at that height, and checks the ledger's per-denom totals against the chain's total supply; it exits non-zero on any mismatch or misconfiguration, so it can gate a deploy. Querying at the checkpoint rather than the moving tip keeps the comparison race-free, which requires an unpruned (archival) node — sandbox is archival. `RECONCILE_SAMPLE_SIZE` overrides the default sample of 100 accounts. + ## Raw block archive Set `ARCHIVE_BUCKET` to a GCS bucket name to keep a zstd-compressed copy of every raw `/block` and `/block_results` payload, so handler fixes and new modules can be replayed without re-fetching history from RPC. Leave it unset and both roles behave exactly as before (the boot log says `ARCHIVE_DISABLED`). Authentication uses Application Default Credentials; no key material is configured in the app. diff --git a/apps/chain-indexer/drizzle/0002_clever_bulldozer.sql b/apps/chain-indexer/drizzle/0002_clever_bulldozer.sql new file mode 100644 index 0000000000..1c03820219 --- /dev/null +++ b/apps/chain-indexer/drizzle/0002_clever_bulldozer.sql @@ -0,0 +1,14 @@ +CREATE TYPE "cosmos"."account_tx_role" AS ENUM('signer', 'sender', 'receiver');--> statement-breakpoint +ALTER TYPE "cosmos"."balance_change_reason" ADD VALUE 'staking';--> statement-breakpoint +CREATE TABLE "cosmos"."account_txs" ( + "account_id" integer NOT NULL, + "height" bigint NOT NULL, + "tx_index" integer NOT NULL, + "role" "cosmos"."account_tx_role" NOT NULL, + CONSTRAINT "account_txs_account_id_height_tx_index_role_pk" PRIMARY KEY("account_id","height","tx_index","role") +); +--> statement-breakpoint +ALTER TABLE "cosmos"."balance_changes" ADD COLUMN "tx_index" integer;--> statement-breakpoint +ALTER TABLE "cosmos"."balance_changes" ADD COLUMN "event_index" integer NOT NULL;--> statement-breakpoint +ALTER TABLE "cosmos"."account_txs" ADD CONSTRAINT "account_txs_account_id_accounts_id_fk" FOREIGN KEY ("account_id") REFERENCES "cosmos"."accounts"("id") ON DELETE no action ON UPDATE no action;--> statement-breakpoint +CREATE UNIQUE INDEX "balance_changes_height_event_index_idx" ON "cosmos"."balance_changes" USING btree ("height","event_index"); \ No newline at end of file diff --git a/apps/chain-indexer/drizzle/meta/0002_snapshot.json b/apps/chain-indexer/drizzle/meta/0002_snapshot.json new file mode 100644 index 0000000000..423c04e22e --- /dev/null +++ b/apps/chain-indexer/drizzle/meta/0002_snapshot.json @@ -0,0 +1,801 @@ +{ + "id": "b507e1c8-31e3-46c8-8dc2-c0c6927e972d", + "prevId": "05402ce9-8c3f-4514-8150-e7911053df30", + "version": "7", + "dialect": "postgresql", + "tables": { + "cosmos.account_balances": { + "name": "account_balances", + "schema": "cosmos", + "columns": { + "account_id": { + "name": "account_id", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "denom": { + "name": "denom", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "amount": { + "name": "amount", + "type": "numeric(38, 0)", + "primaryKey": false, + "notNull": true + } + }, + "indexes": {}, + "foreignKeys": { + "account_balances_account_id_accounts_id_fk": { + "name": "account_balances_account_id_accounts_id_fk", + "tableFrom": "account_balances", + "tableTo": "accounts", + "schemaTo": "cosmos", + "columnsFrom": [ + "account_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": { + "account_balances_account_id_denom_pk": { + "name": "account_balances_account_id_denom_pk", + "columns": [ + "account_id", + "denom" + ] + } + }, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "cosmos.account_txs": { + "name": "account_txs", + "schema": "cosmos", + "columns": { + "account_id": { + "name": "account_id", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "height": { + "name": "height", + "type": "bigint", + "primaryKey": false, + "notNull": true + }, + "tx_index": { + "name": "tx_index", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "role": { + "name": "role", + "type": "account_tx_role", + "typeSchema": "cosmos", + "primaryKey": false, + "notNull": true + } + }, + "indexes": {}, + "foreignKeys": { + "account_txs_account_id_accounts_id_fk": { + "name": "account_txs_account_id_accounts_id_fk", + "tableFrom": "account_txs", + "tableTo": "accounts", + "schemaTo": "cosmos", + "columnsFrom": [ + "account_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": { + "account_txs_account_id_height_tx_index_role_pk": { + "name": "account_txs_account_id_height_tx_index_role_pk", + "columns": [ + "account_id", + "height", + "tx_index", + "role" + ] + } + }, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "cosmos.accounts": { + "name": "accounts", + "schema": "cosmos", + "columns": { + "id": { + "name": "id", + "type": "serial", + "primaryKey": true, + "notNull": true + }, + "address": { + "name": "address", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "account_number": { + "name": "account_number", + "type": "bigint", + "primaryKey": false, + "notNull": false + }, + "account_type": { + "name": "account_type", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "is_module_account": { + "name": "is_module_account", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": false + } + }, + "indexes": { + "accounts_address_idx": { + "name": "accounts_address_idx", + "columns": [ + { + "expression": "address", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "cosmos.balance_changes": { + "name": "balance_changes", + "schema": "cosmos", + "columns": { + "id": { + "name": "id", + "type": "bigserial", + "primaryKey": true, + "notNull": true + }, + "account_id": { + "name": "account_id", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "denom": { + "name": "denom", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "delta": { + "name": "delta", + "type": "numeric(38, 0)", + "primaryKey": false, + "notNull": true + }, + "balance_after": { + "name": "balance_after", + "type": "numeric(38, 0)", + "primaryKey": false, + "notNull": true + }, + "reason": { + "name": "reason", + "type": "balance_change_reason", + "typeSchema": "cosmos", + "primaryKey": false, + "notNull": true + }, + "height": { + "name": "height", + "type": "bigint", + "primaryKey": false, + "notNull": true + }, + "tx_index": { + "name": "tx_index", + "type": "integer", + "primaryKey": false, + "notNull": false + }, + "event_index": { + "name": "event_index", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "counterparty_account_id": { + "name": "counterparty_account_id", + "type": "integer", + "primaryKey": false, + "notNull": false + } + }, + "indexes": { + "balance_changes_account_denom_height_idx": { + "name": "balance_changes_account_denom_height_idx", + "columns": [ + { + "expression": "account_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "denom", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "height", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, + "balance_changes_height_event_index_idx": { + "name": "balance_changes_height_event_index_idx", + "columns": [ + { + "expression": "height", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "event_index", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "balance_changes_account_id_accounts_id_fk": { + "name": "balance_changes_account_id_accounts_id_fk", + "tableFrom": "balance_changes", + "tableTo": "accounts", + "schemaTo": "cosmos", + "columnsFrom": [ + "account_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + }, + "balance_changes_counterparty_account_id_accounts_id_fk": { + "name": "balance_changes_counterparty_account_id_accounts_id_fk", + "tableFrom": "balance_changes", + "tableTo": "accounts", + "schemaTo": "cosmos", + "columnsFrom": [ + "counterparty_account_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "cosmos.blocks": { + "name": "blocks", + "schema": "cosmos", + "columns": { + "height": { + "name": "height", + "type": "bigint", + "primaryKey": true, + "notNull": true + }, + "datetime": { + "name": "datetime", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true + }, + "hash": { + "name": "hash", + "type": "bytea", + "primaryKey": false, + "notNull": true + }, + "parent_hash": { + "name": "parent_hash", + "type": "bytea", + "primaryKey": false, + "notNull": false + }, + "proposer_address": { + "name": "proposer_address", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "tx_count": { + "name": "tx_count", + "type": "integer", + "primaryKey": false, + "notNull": true + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "cosmos.delegations": { + "name": "delegations", + "schema": "cosmos", + "columns": { + "delegator_account_id": { + "name": "delegator_account_id", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "validator_operator_address": { + "name": "validator_operator_address", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "shares": { + "name": "shares", + "type": "numeric(38, 18)", + "primaryKey": false, + "notNull": true + } + }, + "indexes": {}, + "foreignKeys": { + "delegations_delegator_account_id_accounts_id_fk": { + "name": "delegations_delegator_account_id_accounts_id_fk", + "tableFrom": "delegations", + "tableTo": "accounts", + "schemaTo": "cosmos", + "columnsFrom": [ + "delegator_account_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": { + "delegations_delegator_account_id_validator_operator_address_pk": { + "name": "delegations_delegator_account_id_validator_operator_address_pk", + "columns": [ + "delegator_account_id", + "validator_operator_address" + ] + } + }, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.indexer_state": { + "name": "indexer_state", + "schema": "", + "columns": { + "stream": { + "name": "stream", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "last_height": { + "name": "last_height", + "type": "bigint", + "primaryKey": false, + "notNull": true + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "cosmos.message_types": { + "name": "message_types", + "schema": "cosmos", + "columns": { + "id": { + "name": "id", + "type": "serial", + "primaryKey": true, + "notNull": true + }, + "type": { + "name": "type", + "type": "text", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "message_types_type_idx": { + "name": "message_types_type_idx", + "columns": [ + { + "expression": "type", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "cosmos.messages": { + "name": "messages", + "schema": "cosmos", + "columns": { + "height": { + "name": "height", + "type": "bigint", + "primaryKey": false, + "notNull": true + }, + "tx_index": { + "name": "tx_index", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "index": { + "name": "index", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "type_id": { + "name": "type_id", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "body": { + "name": "body", + "type": "jsonb", + "primaryKey": false, + "notNull": false + } + }, + "indexes": { + "messages_type_id_idx": { + "name": "messages_type_id_idx", + "columns": [ + { + "expression": "type_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "messages_type_id_message_types_id_fk": { + "name": "messages_type_id_message_types_id_fk", + "tableFrom": "messages", + "tableTo": "message_types", + "schemaTo": "cosmos", + "columnsFrom": [ + "type_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": { + "messages_height_tx_index_index_pk": { + "name": "messages_height_tx_index_index_pk", + "columns": [ + "height", + "tx_index", + "index" + ] + } + }, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "cosmos.transactions": { + "name": "transactions", + "schema": "cosmos", + "columns": { + "height": { + "name": "height", + "type": "bigint", + "primaryKey": false, + "notNull": true + }, + "index": { + "name": "index", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "hash": { + "name": "hash", + "type": "bytea", + "primaryKey": false, + "notNull": true + }, + "code": { + "name": "code", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "gas_used": { + "name": "gas_used", + "type": "bigint", + "primaryKey": false, + "notNull": true + }, + "gas_wanted": { + "name": "gas_wanted", + "type": "bigint", + "primaryKey": false, + "notNull": true + }, + "fee": { + "name": "fee", + "type": "jsonb", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "transactions_hash_idx": { + "name": "transactions_hash_idx", + "columns": [ + { + "expression": "hash", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": { + "transactions_height_index_pk": { + "name": "transactions_height_index_pk", + "columns": [ + "height", + "index" + ] + } + }, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "cosmos.validators": { + "name": "validators", + "schema": "cosmos", + "columns": { + "operator_address": { + "name": "operator_address", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "account_address": { + "name": "account_address", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "hex_address": { + "name": "hex_address", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "moniker": { + "name": "moniker", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "identity": { + "name": "identity", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "website": { + "name": "website", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "details": { + "name": "details", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "security_contact": { + "name": "security_contact", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "commission_rate": { + "name": "commission_rate", + "type": "numeric(20, 18)", + "primaryKey": false, + "notNull": false + }, + "commission_max_rate": { + "name": "commission_max_rate", + "type": "numeric(20, 18)", + "primaryKey": false, + "notNull": false + }, + "commission_max_change_rate": { + "name": "commission_max_change_rate", + "type": "numeric(20, 18)", + "primaryKey": false, + "notNull": false + }, + "min_self_delegation": { + "name": "min_self_delegation", + "type": "numeric(38, 0)", + "primaryKey": false, + "notNull": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + } + }, + "enums": { + "cosmos.account_tx_role": { + "name": "account_tx_role", + "schema": "cosmos", + "values": [ + "signer", + "sender", + "receiver" + ] + }, + "cosmos.balance_change_reason": { + "name": "balance_change_reason", + "schema": "cosmos", + "values": [ + "genesis", + "transfer", + "fee", + "reward", + "commission", + "slash", + "gov", + "ibc", + "escrow", + "bme", + "mint", + "burn", + "staking" + ] + } + }, + "schemas": { + "cosmos": "cosmos" + }, + "sequences": {}, + "roles": {}, + "policies": {}, + "views": {}, + "_meta": { + "columns": {}, + "schemas": {}, + "tables": {} + } +} \ No newline at end of file diff --git a/apps/chain-indexer/drizzle/meta/_journal.json b/apps/chain-indexer/drizzle/meta/_journal.json index d80e14e60d..43ea741cff 100644 --- a/apps/chain-indexer/drizzle/meta/_journal.json +++ b/apps/chain-indexer/drizzle/meta/_journal.json @@ -15,6 +15,13 @@ "when": 1786548202033, "tag": "0001_long_mach_iv", "breakpoints": true + }, + { + "idx": 2, + "version": "7", + "when": 1786566671632, + "tag": "0002_clever_bulldozer", + "breakpoints": true } ] } \ No newline at end of file diff --git a/apps/chain-indexer/package.json b/apps/chain-indexer/package.json index 17328b775a..6066df516e 100644 --- a/apps/chain-indexer/package.json +++ b/apps/chain-indexer/package.json @@ -13,6 +13,7 @@ "lint": "eslint .", "migration:gen": "drizzle-kit generate", "prod": "node --enable-source-maps --require ./dist/instrumentation.js ./dist/server.js", + "reconcile": "npm run build && node --enable-source-maps ./dist/reconcile.js", "start": "tsup --watch", "test": "vitest run --project unit", "test:cov": "vitest run --project unit --coverage", @@ -24,6 +25,7 @@ "@akashnetwork/instrumentation": "*", "@akashnetwork/logging": "*", "@akashnetwork/net": "*", + "@cosmjs/amino": "~0.38.0", "@cosmjs/encoding": "~0.38.0", "@cosmjs/proto-signing": "~0.38.0", "@cosmjs/stargate": "~0.38.0", diff --git a/apps/chain-indexer/src/config/env.config.spec.ts b/apps/chain-indexer/src/config/env.config.spec.ts index f448a6e043..edc4274aa7 100644 --- a/apps/chain-indexer/src/config/env.config.spec.ts +++ b/apps/chain-indexer/src/config/env.config.spec.ts @@ -81,6 +81,26 @@ describe("envSchema", () => { expect(config.BACKFILL_TO_HEIGHT).toBeUndefined(); }); + it("treats an empty RECONCILE_SAMPLE_SIZE as absent", () => { + const config = setup({ RECONCILE_SAMPLE_SIZE: "" }); + + expect(config.RECONCILE_SAMPLE_SIZE).toBeUndefined(); + }); + + it("coerces a numeric RECONCILE_SAMPLE_SIZE string", () => { + const config = setup({ RECONCILE_SAMPLE_SIZE: "250" }); + + expect(config.RECONCILE_SAMPLE_SIZE).toBe(250); + }); + + it("rejects a non-positive RECONCILE_SAMPLE_SIZE", () => { + expect(() => setup({ RECONCILE_SAMPLE_SIZE: "0" })).toThrow(); + }); + + it("rejects a fractional RECONCILE_SAMPLE_SIZE", () => { + expect(() => setup({ RECONCILE_SAMPLE_SIZE: "10.5" })).toThrow(); + }); + function setup(overrides?: Record) { return envSchema.parse({ POSTGRES_DB_URI: "postgres://unit:unit@localhost:5432/unit", ...overrides }); } diff --git a/apps/chain-indexer/src/config/env.config.ts b/apps/chain-indexer/src/config/env.config.ts index 6f3c849257..c62d647a5d 100644 --- a/apps/chain-indexer/src/config/env.config.ts +++ b/apps/chain-indexer/src/config/env.config.ts @@ -39,6 +39,8 @@ const rawEnvSchema = z.object({ ARCHIVE_STORAGE_API_ENDPOINT: z.preprocess(emptyStringAsUndefined, z.string().url().optional()), /** Decoded message bodies above this serialized size are stored as null to keep pathological messages out of Postgres. */ MESSAGE_BODY_MAX_BYTES: z.number({ coerce: true }).int().positive().default(65_536), + /** How many of the highest-balance accounts `npm run reconcile` checks against the chain. Unset defers to the service default. */ + RECONCILE_SAMPLE_SIZE: z.preprocess(emptyStringAsUndefined, z.number({ coerce: true }).int().positive().optional()), DRIZZLE_MIGRATIONS_FOLDER: z.string().default("./drizzle"), LOG_LEVEL: z.enum(["fatal", "error", "warn", "info", "debug", "trace"]).optional().default("info"), STD_OUT_LOG_FORMAT: z.enum(["json", "pretty"]).optional().default("json"), diff --git a/apps/chain-indexer/src/db/schema.spec.ts b/apps/chain-indexer/src/db/schema.spec.ts index 80c7fe9dce..a804cf4e13 100644 --- a/apps/chain-indexer/src/db/schema.spec.ts +++ b/apps/chain-indexer/src/db/schema.spec.ts @@ -1,7 +1,7 @@ import { getTableConfig } from "drizzle-orm/pg-core"; import { describe, expect, it } from "vitest"; -import { AccountBalances, Accounts, BalanceChanges, Delegations, Validators } from "@src/db/schema"; +import { AccountBalances, Accounts, AccountTxs, BalanceChanges, Delegations, Validators } from "@src/db/schema"; describe("cosmos genesis schema", () => { it("interns accounts under a unique address index", () => { @@ -25,10 +25,26 @@ describe("cosmos genesis schema", () => { const config = getTableConfig(BalanceChanges); expect(config.foreignKeys).toHaveLength(2); - expect(config.indexes).toHaveLength(1); config.foreignKeys.forEach(foreignKey => expect(foreignKey.reference().foreignColumns[0].name).toBe("id")); }); + it("makes the ledger idempotent with a unique (height, event_index) index", () => { + const config = getTableConfig(BalanceChanges); + + const uniqueOnHeightEvent = config.indexes.find(index => index.config.name === "balance_changes_height_event_index_idx"); + expect(uniqueOnHeightEvent?.config.unique).toBe(true); + expect(uniqueOnHeightEvent?.config.columns.map(column => (column as { name: string }).name)).toEqual(["height", "event_index"]); + }); + + it("keys the address activity log by account, height, tx and role", () => { + const config = getTableConfig(AccountTxs); + + expect(config.name).toBe("account_txs"); + expect(config.primaryKeys[0].columns.map(column => column.name)).toEqual(["account_id", "height", "tx_index", "role"]); + expect(config.foreignKeys).toHaveLength(1); + expect(config.foreignKeys[0].reference().foreignColumns[0].name).toBe("id"); + }); + it("keys delegations by delegator and validator with a delegator foreign key", () => { const config = getTableConfig(Delegations); diff --git a/apps/chain-indexer/src/db/schema.ts b/apps/chain-indexer/src/db/schema.ts index 93e9d6e184..19523cbb7d 100644 --- a/apps/chain-indexer/src/db/schema.ts +++ b/apps/chain-indexer/src/db/schema.ts @@ -93,7 +93,8 @@ export const balanceChangeReason = cosmosSchema.enum("balance_change_reason", [ "escrow", "bme", "mint", - "burn" + "burn", + "staking" ]); /** Addresses interned once and referenced by integer id, mirroring the message_types lookup. */ @@ -135,9 +136,14 @@ export const BalanceChanges = cosmosSchema.table( balanceAfter: numeric("balance_after", { precision: 38, scale: 0 }).notNull(), reason: balanceChangeReason("reason").notNull(), height: bigint("height", { mode: "number" }).notNull(), + txIndex: integer("tx_index"), + eventIndex: integer("event_index").notNull(), counterpartyAccountId: integer("counterparty_account_id").references(() => Accounts.id) }, - t => [index("balance_changes_account_denom_height_idx").on(t.accountId, t.denom, t.height)] + t => [ + index("balance_changes_account_denom_height_idx").on(t.accountId, t.denom, t.height), + uniqueIndex("balance_changes_height_event_index_idx").on(t.height, t.eventIndex) + ] ); export const Validators = cosmosSchema.table("validators", { @@ -166,3 +172,23 @@ export const Delegations = cosmosSchema.table( }, t => [primaryKey({ columns: [t.delegatorAccountId, t.validatorOperatorAddress] })] ); + +/** How an address participated in a transaction: it signed it, or it was the sender/recipient of a coin movement. */ +export const accountTxRole = cosmosSchema.enum("account_tx_role", ["signer", "sender", "receiver"]); + +/** + * Address activity log: one row per (address, tx, role). The leading `(accountId, height)` of the primary + * key serves "list an address's activity newest-first"; the composite key makes re-committing a block idempotent. + */ +export const AccountTxs = cosmosSchema.table( + "account_txs", + { + accountId: integer("account_id") + .notNull() + .references(() => Accounts.id), + height: bigint("height", { mode: "number" }).notNull(), + txIndex: integer("tx_index").notNull(), + role: accountTxRole("role").notNull() + }, + t => [primaryKey({ columns: [t.accountId, t.height, t.txIndex, t.role] })] +); diff --git a/apps/chain-indexer/src/genesis/bank-seeder.service.spec.ts b/apps/chain-indexer/src/genesis/bank-seeder.service.spec.ts index 6774f23516..a4fc80d4a7 100644 --- a/apps/chain-indexer/src/genesis/bank-seeder.service.spec.ts +++ b/apps/chain-indexer/src/genesis/bank-seeder.service.spec.ts @@ -18,9 +18,9 @@ describe(BankSeeder.name, () => { { accountId: 3, denom: "uakt", amount: "20" } ]); expect(rowsFor(inserts, BalanceChanges)).toEqual([ - { accountId: 1, denom: "uakt", delta: "10", balanceAfter: "10", reason: "genesis", height: 1, counterpartyAccountId: null }, - { accountId: 2, denom: "uakt", delta: "5", balanceAfter: "5", reason: "genesis", height: 1, counterpartyAccountId: null }, - { accountId: 3, denom: "uakt", delta: "20", balanceAfter: "20", reason: "genesis", height: 1, counterpartyAccountId: null } + { accountId: 1, denom: "uakt", delta: "10", balanceAfter: "10", reason: "genesis", height: 0, txIndex: null, eventIndex: 0, counterpartyAccountId: null }, + { accountId: 2, denom: "uakt", delta: "5", balanceAfter: "5", reason: "genesis", height: 0, txIndex: null, eventIndex: 1, counterpartyAccountId: null }, + { accountId: 3, denom: "uakt", delta: "20", balanceAfter: "20", reason: "genesis", height: 0, txIndex: null, eventIndex: 2, counterpartyAccountId: null } ]); }); diff --git a/apps/chain-indexer/src/genesis/bank-seeder.service.ts b/apps/chain-indexer/src/genesis/bank-seeder.service.ts index 5b27d5f690..7f803d6c1e 100644 --- a/apps/chain-indexer/src/genesis/bank-seeder.service.ts +++ b/apps/chain-indexer/src/genesis/bank-seeder.service.ts @@ -13,10 +13,15 @@ export class BankSeeder implements GenesisModuleSeeder { * Seeding all `bank.balances` (module and vesting accounts included) makes the current-balance total * reconcile to `bank.supply` by construction. Idempotency comes from the import marker, so the ledger * insert intentionally has no conflict target. + * + * The ledger rows sit at `initialHeight - 1` (the pre-block opening balance) so they never collide with + * block `initialHeight`'s own coin events on the `(height, event_index)` unique key and give that block's + * batch a correct running-balance baseline. */ async seed(tx: ChainTransaction, genesis: ParsedGenesis, context: GenesisSeedContext): Promise { const balanceRows: (typeof AccountBalances.$inferInsert)[] = []; const changeRows: (typeof BalanceChanges.$inferInsert)[] = []; + let eventIndex = 0; for (const balance of genesis.balances) { const accountId = context.accountIdByAddress.get(balance.address); @@ -32,7 +37,9 @@ export class BankSeeder implements GenesisModuleSeeder { delta: coin.amount, balanceAfter: coin.amount, reason: "genesis", - height: context.initialHeight, + height: context.initialHeight - 1, + txIndex: null, + eventIndex: eventIndex++, counterpartyAccountId: null }); } diff --git a/apps/chain-indexer/src/genesis/genesis-address.ts b/apps/chain-indexer/src/genesis/genesis-address.ts index bca386e366..c758d71c6d 100644 --- a/apps/chain-indexer/src/genesis/genesis-address.ts +++ b/apps/chain-indexer/src/genesis/genesis-address.ts @@ -1,6 +1,9 @@ import { fromBase64, fromBech32, toBech32, toHex } from "@cosmjs/encoding"; import { createHash } from "node:crypto"; +/** Bech32 human-readable prefix for account addresses. Shared by mainnet, sandbox and testnet — all Akash chains. */ +export const AKASH_ADDRESS_PREFIX = "akash"; + const ED25519_PUBKEY_TYPE = "/cosmos.crypto.ed25519.PubKey"; /** diff --git a/apps/chain-indexer/src/pipeline/balance/account-interner.service.spec.ts b/apps/chain-indexer/src/pipeline/balance/account-interner.service.spec.ts new file mode 100644 index 0000000000..e770e6f213 --- /dev/null +++ b/apps/chain-indexer/src/pipeline/balance/account-interner.service.spec.ts @@ -0,0 +1,108 @@ +import { describe, expect, it } from "vitest"; + +import { INSERT_CHUNK_SIZE } from "@src/db/insert-chunk-size"; +import { Accounts } from "@src/db/schema"; +import { AccountInterner } from "@src/pipeline/balance/account-interner.service"; +import type { ChainDatabase } from "@src/providers/db.provider"; + +describe(AccountInterner.name, () => { + it("returns existing ids without inserting when every address is already interned", async () => { + const { interner, insertedRows } = setup({ + selectResults: [ + [ + { id: 1, address: "akash1a" }, + { id: 2, address: "akash1b" } + ] + ] + }); + + const ids = await interner.resolve(["akash1a", "akash1b"]); + + expect(ids).toEqual( + new Map([ + ["akash1a", 1], + ["akash1b", 2] + ]) + ); + expect(insertedRows).toEqual([]); + }); + + it("inserts only the missing addresses and merges the returned ids", async () => { + const { interner, insertedRows } = setup({ selectResults: [[{ id: 1, address: "akash1a" }]], insertReturning: [{ id: 2, address: "akash1b" }] }); + + const ids = await interner.resolve(["akash1a", "akash1b"]); + + expect(insertedRows).toEqual([{ table: Accounts, rows: [{ address: "akash1b" }] }]); + expect(ids).toEqual( + new Map([ + ["akash1a", 1], + ["akash1b", 2] + ]) + ); + }); + + it("re-selects addresses lost to a concurrent insert", async () => { + const { interner } = setup({ selectResults: [[], [{ id: 5, address: "akash1c" }]], insertReturning: [] }); + + const ids = await interner.resolve(["akash1c"]); + + expect(ids).toEqual(new Map([["akash1c", 5]])); + }); + + it("dedups repeated addresses so each is interned once", async () => { + const { interner, insertedRows } = setup({ selectResults: [[]], insertReturning: [{ id: 1, address: "akash1a" }] }); + + await interner.resolve(["akash1a", "akash1a"]); + + expect(insertedRows).toEqual([{ table: Accounts, rows: [{ address: "akash1a" }] }]); + }); + + it("chunks the existence lookup so a batch past the bind-parameter limit stays within it", async () => { + const addresses = Array.from({ length: INSERT_CHUNK_SIZE + 1 }, (_, index) => `akash1_${index}`); + const firstChunk = addresses.slice(0, INSERT_CHUNK_SIZE).map((address, index) => ({ id: index + 1, address })); + const secondChunk = addresses.slice(INSERT_CHUNK_SIZE).map((address, index) => ({ id: INSERT_CHUNK_SIZE + 1 + index, address })); + const { interner, insertedRows, selectCount } = setup({ selectResults: [firstChunk, secondChunk] }); + + const ids = await interner.resolve(addresses); + + expect(selectCount()).toBe(2); + expect(insertedRows).toEqual([]); + expect(ids.size).toBe(addresses.length); + }); + + it("does nothing for an empty address set", async () => { + const { interner, insertedRows, selectCount } = setup(); + + const ids = await interner.resolve([]); + + expect(ids.size).toBe(0); + expect(insertedRows).toEqual([]); + expect(selectCount()).toBe(0); + }); + + function setup(input?: { selectResults?: Array>; insertReturning?: Array<{ id: number; address: string }> }) { + const selectResults = [...(input?.selectResults ?? [[]])]; + const insertedRows: Array<{ table: unknown; rows: unknown }> = []; + let selects = 0; + + const dbFake = { + select: () => ({ + from: () => ({ + where: () => { + selects++; + return Promise.resolve(selectResults.shift() ?? []); + } + }) + }), + insert: (table: unknown) => ({ + values: (rows: unknown) => { + insertedRows.push({ table, rows }); + return { onConflictDoNothing: () => ({ returning: () => Promise.resolve(input?.insertReturning ?? []) }) }; + } + }) + }; + + const interner = new AccountInterner(dbFake as unknown as ChainDatabase); + return { interner, insertedRows, selectCount: () => selects }; + } +}); diff --git a/apps/chain-indexer/src/pipeline/balance/account-interner.service.ts b/apps/chain-indexer/src/pipeline/balance/account-interner.service.ts new file mode 100644 index 0000000000..c252d7c5c4 --- /dev/null +++ b/apps/chain-indexer/src/pipeline/balance/account-interner.service.ts @@ -0,0 +1,64 @@ +import { inArray } from "drizzle-orm"; +import chunk from "lodash/chunk"; +import { inject, singleton } from "tsyringe"; + +import { INSERT_CHUNK_SIZE } from "@src/db/insert-chunk-size"; +import { Accounts } from "@src/db/schema"; +import type { ChainDatabase } from "@src/providers/db.provider"; +import { CHAIN_DB } from "@src/providers/db.provider"; + +/** + * Resolves mid-chain addresses to account ids, creating any that don't exist yet. Unlike the genesis + * account-seeder (which owns an empty table and reads ids straight back from `returning()`), addresses + * appear unboundedly during sync, so this holds no permanent cache and resolves per batch: select existing, + * insert the rest with `onConflictDoNothing().returning()`, then re-select any lost to a concurrent writer. + * It runs on the base connection, not the commit transaction, so the interned rows are visible when the + * transaction inserts ledger rows that reference them. + */ +@singleton() +export class AccountInterner { + readonly #db: ChainDatabase; + + constructor(@inject(CHAIN_DB) db: ChainDatabase) { + this.#db = db; + } + + async resolve(addresses: Iterable): Promise> { + const unique = [...new Set(addresses)]; + const idByAddress = new Map(); + + if (unique.length === 0) { + return idByAddress; + } + + await this.#selectInto(idByAddress, unique); + + const missing = unique.filter(address => !idByAddress.has(address)); + if (missing.length === 0) { + return idByAddress; + } + + for (const addressChunk of chunk(missing, INSERT_CHUNK_SIZE)) { + const inserted = await this.#db + .insert(Accounts) + .values(addressChunk.map(address => ({ address }))) + .onConflictDoNothing() + .returning({ id: Accounts.id, address: Accounts.address }); + inserted.forEach(row => idByAddress.set(row.address, row.id)); + } + + const stillMissing = missing.filter(address => !idByAddress.has(address)); + if (stillMissing.length > 0) { + await this.#selectInto(idByAddress, stillMissing); + } + + return idByAddress; + } + + async #selectInto(idByAddress: Map, addresses: string[]): Promise { + for (const addressChunk of chunk(addresses, INSERT_CHUNK_SIZE)) { + const rows = await this.#db.select({ id: Accounts.id, address: Accounts.address }).from(Accounts).where(inArray(Accounts.address, addressChunk)); + rows.forEach(row => idByAddress.set(row.address, row.id)); + } + } +} diff --git a/apps/chain-indexer/src/pipeline/balance/account-tx-deriver.spec.ts b/apps/chain-indexer/src/pipeline/balance/account-tx-deriver.spec.ts new file mode 100644 index 0000000000..130e42cecf --- /dev/null +++ b/apps/chain-indexer/src/pipeline/balance/account-tx-deriver.spec.ts @@ -0,0 +1,92 @@ +import { describe, expect, it } from "vitest"; + +import { deriveAccountTxs } from "@src/pipeline/balance/account-tx-deriver"; +import type { DecodedBlock, DecodedEvent, DecodedTransaction } from "@src/pipeline/decoded-block"; + +describe("deriveAccountTxs", () => { + it("records each signer of a transaction with the signer role", () => { + const block = buildBlock([buildTx({ index: 0, signerAddresses: ["akash1signer1", "akash1signer2"] })]); + + expect(deriveAccountTxs(block)).toEqual([ + { address: "akash1signer1", height: 10, txIndex: 0, role: "signer" }, + { address: "akash1signer2", height: 10, txIndex: 0, role: "signer" } + ]); + }); + + it("records the sender and recipient of a transfer event with their roles", () => { + const block = buildBlock([buildTx({ index: 0, events: [event("transfer", { sender: "akash1a", recipient: "akash1b", amount: "1uakt" })] })]); + + expect(deriveAccountTxs(block)).toEqual([ + { address: "akash1a", height: 10, txIndex: 0, role: "sender" }, + { address: "akash1b", height: 10, txIndex: 0, role: "receiver" } + ]); + }); + + it("dedups a repeated (address, tx, role) so the primary key never conflicts within a block", () => { + const block = buildBlock([ + buildTx({ + index: 0, + signerAddresses: ["akash1a", "akash1a"], + events: [ + event("transfer", { sender: "akash1a", recipient: "akash1b", amount: "1uakt" }), + event("transfer", { sender: "akash1a", recipient: "akash1b", amount: "2uakt" }) + ] + }) + ]); + + expect(deriveAccountTxs(block)).toEqual([ + { address: "akash1a", height: 10, txIndex: 0, role: "signer" }, + { address: "akash1a", height: 10, txIndex: 0, role: "sender" }, + { address: "akash1b", height: 10, txIndex: 0, role: "receiver" } + ]); + }); + + it("keeps the same address distinct across roles and transactions", () => { + const block = buildBlock([ + buildTx({ index: 0, signerAddresses: ["akash1a"] }), + buildTx({ index: 1, events: [event("transfer", { sender: "akash1a", recipient: "akash1b", amount: "1uakt" })] }) + ]); + + expect(deriveAccountTxs(block)).toEqual([ + { address: "akash1a", height: 10, txIndex: 0, role: "signer" }, + { address: "akash1a", height: 10, txIndex: 1, role: "sender" }, + { address: "akash1b", height: 10, txIndex: 1, role: "receiver" } + ]); + }); + + it("ignores block-level transfer events, which have no transaction to attribute", () => { + const block: DecodedBlock = { ...buildBlock([]), blockEvents: [event("transfer", { sender: "akash1a", recipient: "akash1b", amount: "1uakt" })] }; + + expect(deriveAccountTxs(block)).toEqual([]); + }); + + function buildBlock(transactions: DecodedTransaction[]): DecodedBlock { + return { + height: 10, + datetime: new Date("2026-08-11T00:00:00Z"), + hash: Buffer.alloc(0), + parentHash: null, + proposerAddress: "PROPOSER", + transactions, + blockEvents: [] + }; + } + + function buildTx(input: { index: number; signerAddresses?: string[]; events?: DecodedEvent[] }): DecodedTransaction { + return { + index: input.index, + hash: Buffer.alloc(0), + code: 0, + gasUsed: 0, + gasWanted: 0, + fee: [], + messages: [], + events: input.events ?? [], + signerAddresses: input.signerAddresses ?? [] + }; + } + + function event(type: string, attributes: Record): DecodedEvent { + return { type, attributes }; + } +}); diff --git a/apps/chain-indexer/src/pipeline/balance/account-tx-deriver.ts b/apps/chain-indexer/src/pipeline/balance/account-tx-deriver.ts new file mode 100644 index 0000000000..0acd8fad3f --- /dev/null +++ b/apps/chain-indexer/src/pipeline/balance/account-tx-deriver.ts @@ -0,0 +1,49 @@ +import type { accountTxRole } from "@src/db/schema"; +import type { DecodedBlock } from "@src/pipeline/decoded-block"; + +export type AccountTxRole = (typeof accountTxRole.enumValues)[number]; + +/** One address's participation in a transaction, before its address is interned to an account id. */ +export interface DerivedAccountTx { + address: string; + height: number; + txIndex: number; + role: AccountTxRole; +} + +/** + * Builds the address activity log for a block: every transaction's signers plus the sender and recipient + * of each of its `transfer` events. Rows are deduped per `(address, txIndex, role)` so they never collide + * on the `account_txs` primary key. Block-level events have no owning transaction and are skipped. + */ +export function deriveAccountTxs(block: DecodedBlock): DerivedAccountTx[] { + const rows: DerivedAccountTx[] = []; + const seen = new Set(); + + const add = (address: string, txIndex: number, role: AccountTxRole) => { + if (!address) { + return; + } + const key = `${address}|${txIndex}|${role}`; + if (seen.has(key)) { + return; + } + seen.add(key); + rows.push({ address, height: block.height, txIndex, role }); + }; + + for (const tx of block.transactions) { + for (const signer of tx.signerAddresses) { + add(signer, tx.index, "signer"); + } + + for (const event of tx.events) { + if (event.type === "transfer") { + add(event.attributes.sender, tx.index, "sender"); + add(event.attributes.recipient, tx.index, "receiver"); + } + } + } + + return rows; +} diff --git a/apps/chain-indexer/src/pipeline/balance/balance-deriver.spec.ts b/apps/chain-indexer/src/pipeline/balance/balance-deriver.spec.ts new file mode 100644 index 0000000000..a89afa42d2 --- /dev/null +++ b/apps/chain-indexer/src/pipeline/balance/balance-deriver.spec.ts @@ -0,0 +1,230 @@ +import { describe, expect, it } from "vitest"; + +import { AKASH_ADDRESS_PREFIX } from "@src/genesis/genesis-address"; +import { deriveBalanceChanges } from "@src/pipeline/balance/balance-deriver"; +import { buildModuleAddressRegistry, deriveModuleAddress } from "@src/pipeline/balance/module-address-registry"; +import type { DecodedBlock, DecodedEvent, DecodedTransaction } from "@src/pipeline/decoded-block"; + +const registry = buildModuleAddressRegistry(AKASH_ADDRESS_PREFIX); +const feeCollector = deriveModuleAddress("fee_collector", AKASH_ADDRESS_PREFIX); + +describe("deriveBalanceChanges", () => { + it("emits a debit and a credit for a simple transfer with correlated counterparties", () => { + const block = buildBlock({ + transactions: [ + buildTx({ + index: 0, + events: [ + event("coin_spent", { spender: "akash1a", amount: "100uakt", msg_index: "0" }, 0), + event("coin_received", { receiver: "akash1b", amount: "100uakt", msg_index: "0" }, 0), + event("transfer", { sender: "akash1a", recipient: "akash1b", amount: "100uakt", msg_index: "0" }, 0) + ] + }) + ] + }); + + expect(deriveBalanceChanges(block, registry)).toEqual([ + { address: "akash1a", counterpartyAddress: "akash1b", denom: "uakt", delta: -100n, reason: "transfer", height: 10, txIndex: 0, eventIndex: 0 }, + { address: "akash1b", counterpartyAddress: "akash1a", denom: "uakt", delta: 100n, reason: "transfer", height: 10, txIndex: 0, eventIndex: 1 } + ]); + }); + + it("assigns a deterministic block-wide event index across txs then block events, expanding per denom in amount order", () => { + const block = buildBlock({ + transactions: [ + buildTx({ index: 0, events: [event("coin_spent", { spender: "akash1a", amount: "5uakt,3uatom" }, undefined)] }), + buildTx({ index: 1, events: [event("coin_received", { receiver: "akash1b", amount: "7uakt" }, undefined)] }) + ], + blockEvents: [event("coin_received", { receiver: "akash1c", amount: "9uakt" }, undefined)] + }); + + expect( + deriveBalanceChanges(block, registry).map(change => ({ + eventIndex: change.eventIndex, + address: change.address, + denom: change.denom, + delta: change.delta, + txIndex: change.txIndex + })) + ).toEqual([ + { eventIndex: 0, address: "akash1a", denom: "uakt", delta: -5n, txIndex: 0 }, + { eventIndex: 1, address: "akash1a", denom: "uatom", delta: -3n, txIndex: 0 }, + { eventIndex: 2, address: "akash1b", denom: "uakt", delta: 7n, txIndex: 1 }, + { eventIndex: 3, address: "akash1c", denom: "uakt", delta: 9n, txIndex: null } + ]); + }); + + it("classifies a fee payment to the fee collector", () => { + const block = buildBlock({ + transactions: [ + buildTx({ + index: 0, + events: [ + event("coin_spent", { spender: "akash1payer", amount: "500uakt" }, undefined), + event("coin_received", { receiver: feeCollector, amount: "500uakt" }, undefined), + event("transfer", { sender: "akash1payer", recipient: feeCollector, amount: "500uakt" }, undefined) + ] + }) + ] + }); + + const changes = deriveBalanceChanges(block, registry); + expect(changes.map(change => change.reason)).toEqual(["fee", "fee"]); + expect(changes[0]).toMatchObject({ address: "akash1payer", counterpartyAddress: feeCollector, reason: "fee" }); + }); + + it("classifies a block-level inflation mint from the coincident coinbase event", () => { + const mintModule = deriveModuleAddress("mint", AKASH_ADDRESS_PREFIX); + const block = buildBlock({ + transactions: [], + blockEvents: [ + event("coinbase", { minter: mintModule, amount: "1000uakt" }, undefined), + event("coin_received", { receiver: mintModule, amount: "1000uakt" }, undefined) + ] + }); + + const changes = deriveBalanceChanges(block, registry); + expect(changes).toEqual([ + { address: mintModule, counterpartyAddress: null, denom: "uakt", delta: 1000n, reason: "mint", height: 10, txIndex: null, eventIndex: 0 } + ]); + }); + + it("classifies a debit coincident with a burn event as a burn", () => { + const block = buildBlock({ + transactions: [ + buildTx({ + index: 0, + events: [ + event("coin_spent", { spender: "akash1burner", amount: "42uakt" }, undefined), + event("burn", { burner: "akash1burner", amount: "42uakt" }, undefined) + ] + }) + ] + }); + + expect(deriveBalanceChanges(block, registry)[0]).toMatchObject({ reason: "burn", delta: -42n }); + }); + + it("applies the slash reason only to the coincident burn leg, leaving the block's inflation mint a mint", () => { + const mintModule = deriveModuleAddress("mint", AKASH_ADDRESS_PREFIX); + const bondedPool = deriveModuleAddress("bonded_tokens_pool", AKASH_ADDRESS_PREFIX); + const block = buildBlock({ + transactions: [], + blockEvents: [ + event("coinbase", { minter: mintModule, amount: "1000uakt" }, undefined), + event("coin_received", { receiver: mintModule, amount: "1000uakt" }, undefined), + event("slash", { address: "akashvalcons1jailed", amount: "50uakt" }, undefined), + event("coin_spent", { spender: bondedPool, amount: "50uakt" }, undefined), + event("burn", { burner: bondedPool, amount: "50uakt" }, undefined) + ] + }); + + const byAddress = new Map(deriveBalanceChanges(block, registry).map(change => [change.address, change.reason])); + expect(byAddress.get(mintModule)).toBe("mint"); + expect(byAddress.get(bondedPool)).toBe("slash"); + }); + + it("leaves an unrelated module burn a burn when it shares a block scope with a validator slash", () => { + const bondedPool = deriveModuleAddress("bonded_tokens_pool", AKASH_ADDRESS_PREFIX); + const govModule = deriveModuleAddress("gov", AKASH_ADDRESS_PREFIX); + const block = buildBlock({ + transactions: [], + blockEvents: [ + event("slash", { address: "akashvalcons1jailed", amount: "50uakt" }, undefined), + event("coin_spent", { spender: bondedPool, amount: "50uakt" }, undefined), + event("burn", { burner: bondedPool, amount: "50uakt" }, undefined), + event("coin_spent", { spender: govModule, amount: "10uakt" }, undefined), + event("burn", { burner: govModule, amount: "10uakt" }, undefined) + ] + }); + + const byAddress = new Map(deriveBalanceChanges(block, registry).map(change => [change.address, change.reason])); + expect(byAddress.get(bondedPool)).toBe("slash"); + expect(byAddress.get(govModule)).toBe("burn"); + }); + + it("correlates each debit to the transfer of matching amount when one sender pays several recipients in a block", () => { + const escrow = deriveModuleAddress("escrow", AKASH_ADDRESS_PREFIX); + const block = buildBlock({ + transactions: [], + blockEvents: [ + event("transfer", { sender: escrow, recipient: "akash1providerA", amount: "100uakt" }, undefined), + event("transfer", { sender: escrow, recipient: "akash1providerB", amount: "50uakt" }, undefined), + event("coin_spent", { spender: escrow, amount: "100uakt" }, undefined), + event("coin_spent", { spender: escrow, amount: "50uakt" }, undefined), + event("coin_received", { receiver: "akash1providerA", amount: "100uakt" }, undefined), + event("coin_received", { receiver: "akash1providerB", amount: "50uakt" }, undefined) + ] + }); + + const debits = deriveBalanceChanges(block, registry).filter(change => change.address === escrow); + expect(debits.find(change => change.delta === -100n)?.counterpartyAddress).toBe("akash1providerA"); + expect(debits.find(change => change.delta === -50n)?.counterpartyAddress).toBe("akash1providerB"); + }); + + it("classifies a distribution reward withdrawal using the message type at the coin's msg index", () => { + const distribution = deriveModuleAddress("distribution", AKASH_ADDRESS_PREFIX); + const block = buildBlock({ + transactions: [ + buildTx({ + index: 0, + messages: [{ index: 0, typeUrl: "/cosmos.distribution.v1beta1.MsgWithdrawValidatorCommission", body: null }], + events: [ + event("coin_spent", { spender: distribution, amount: "8uakt", msg_index: "0" }, 0), + event("coin_received", { receiver: "akash1val", amount: "8uakt", msg_index: "0" }, 0), + event("transfer", { sender: distribution, recipient: "akash1val", amount: "8uakt", msg_index: "0" }, 0) + ] + }) + ] + }); + + const creditToValidator = deriveBalanceChanges(block, registry).find(change => change.address === "akash1val"); + expect(creditToValidator?.reason).toBe("commission"); + }); + + it("ignores transfer, coinbase and message events as delta sources", () => { + const block = buildBlock({ + transactions: [ + buildTx({ + index: 0, + events: [ + event("transfer", { sender: "akash1a", recipient: "akash1b", amount: "1uakt" }, undefined), + event("message", { action: "/cosmos.bank.v1beta1.MsgSend" }, undefined) + ] + }) + ] + }); + + expect(deriveBalanceChanges(block, registry)).toEqual([]); + }); + + function buildBlock(input: { transactions: DecodedTransaction[]; blockEvents?: DecodedEvent[] }): DecodedBlock { + return { + height: 10, + datetime: new Date("2026-08-11T00:00:00Z"), + hash: Buffer.alloc(0), + parentHash: null, + proposerAddress: "PROPOSER", + transactions: input.transactions, + blockEvents: input.blockEvents ?? [] + }; + } + + function buildTx(input: { index: number; events: DecodedEvent[]; messages?: DecodedTransaction["messages"] }): DecodedTransaction { + return { + index: input.index, + hash: Buffer.alloc(0), + code: 0, + gasUsed: 0, + gasWanted: 0, + fee: [], + messages: input.messages ?? [], + events: input.events, + signerAddresses: [] + }; + } + + function event(type: string, attributes: Record, msgIndex: number | undefined): DecodedEvent { + return msgIndex === undefined ? { type, attributes } : { type, attributes, msgIndex }; + } +}); diff --git a/apps/chain-indexer/src/pipeline/balance/balance-deriver.ts b/apps/chain-indexer/src/pipeline/balance/balance-deriver.ts new file mode 100644 index 0000000000..9ebca7c13b --- /dev/null +++ b/apps/chain-indexer/src/pipeline/balance/balance-deriver.ts @@ -0,0 +1,154 @@ +import { parseCoins } from "@src/pipeline/balance/coin-amount"; +import type { ModuleAddressRegistry, ModuleRole } from "@src/pipeline/balance/module-address-registry"; +import type { BalanceReason } from "@src/pipeline/balance/reason-classifier"; +import { classifyReason } from "@src/pipeline/balance/reason-classifier"; +import type { DecodedBlock, DecodedEvent, DecodedTransaction } from "@src/pipeline/decoded-block"; + +/** A single balance movement before its address is interned to an account id. Deltas come only from coin_spent/coin_received. */ +export interface DerivedBalanceChange { + address: string; + counterpartyAddress: string | null; + denom: string; + delta: bigint; + reason: BalanceReason; + height: number; + txIndex: number | null; + eventIndex: number; +} + +const BLOCK_SCOPE = "block"; + +/** A slash burns from the staking pools, so a slashing block's burn leg only counts as the slash when its holder is one of those pools. */ +const STAKING_POOL_ROLES: ReadonlySet = new Set(["bonded_tokens_pool", "not_bonded_tokens_pool"]); + +interface ParsedTransfer { + sender: string; + recipient: string; + coins: Map; +} + +/** + * Per-scope classification context: the transfers to correlate a counterparty against, plus which addresses + * minted/burned and whether a slash occurred. `slashed` is scope-wide because a `slash` event names the + * validator, not the staking pool whose coins actually move, so it is only trusted to reclassify the + * coincident burn leg (see `deriveBalanceChanges`), never every movement sharing the scope. + */ +interface ScopeContext { + transfers: ParsedTransfer[]; + minters: Set; + burners: Set; + slashed: boolean; +} + +interface EventSource { + events: DecodedEvent[]; + txIndex: number | null; + msgTypeByIndex: Map; +} + +function scopeKeyOf(event: DecodedEvent): number | string { + return event.msgIndex ?? BLOCK_SCOPE; +} + +function buildScopeContexts(events: DecodedEvent[]): Map { + const scopes = new Map(); + const scopeOf = (event: DecodedEvent) => { + const key = scopeKeyOf(event); + const existing = scopes.get(key); + if (existing) { + return existing; + } + const created: ScopeContext = { transfers: [], minters: new Set(), burners: new Set(), slashed: false }; + scopes.set(key, created); + return created; + }; + + for (const event of events) { + const scope = scopeOf(event); + if (event.type === "transfer") { + scope.transfers.push({ + sender: event.attributes.sender, + recipient: event.attributes.recipient, + coins: new Map(parseCoins(event.attributes.amount ?? "").map(coin => [coin.denom, coin.amount])) + }); + } else if (event.type === "coinbase" && event.attributes.minter) { + scope.minters.add(event.attributes.minter); + } else if (event.type === "burn" && event.attributes.burner) { + scope.burners.add(event.attributes.burner); + } else if (event.type === "slash") { + scope.slashed = true; + } + } + + return scopes; +} + +function correlateCounterparty(scope: ScopeContext, holder: string, denom: string, amount: bigint, direction: "spent" | "received"): string | null { + const matchesHolder = (transfer: ParsedTransfer) => (direction === "spent" ? transfer.sender === holder : transfer.recipient === holder); + const other = (transfer: ParsedTransfer) => (direction === "spent" ? transfer.recipient : transfer.sender); + + const byAmount = scope.transfers.find(transfer => matchesHolder(transfer) && transfer.coins.get(denom) === amount); + const byDenom = scope.transfers.find(transfer => matchesHolder(transfer) && transfer.coins.has(denom)); + const byHolder = scope.transfers.find(matchesHolder); + const match = byAmount ?? byDenom ?? byHolder; + + return match ? other(match) : null; +} + +/** + * Turns a block's coin events into ordered balance movements. Deltas come solely from `coin_spent` + * (holder −amount) and `coin_received` (holder +amount); `transfer`/`coinbase`/`burn`/`slash` only + * inform the counterparty and reason. The `event_index` is a deterministic block-wide sequence — each tx + * in ascending order, its coin events in array order expanded per denom, then block-level events — so a + * re-derivation of the same block reproduces the exact `(height, event_index)` idempotency keys. + */ +export function deriveBalanceChanges(block: DecodedBlock, registry: ModuleAddressRegistry): DerivedBalanceChange[] { + const sources: EventSource[] = [ + ...[...block.transactions].sort((a, b) => a.index - b.index).map(tx => ({ events: tx.events, txIndex: tx.index, msgTypeByIndex: msgTypeByIndexOf(tx) })), + { events: block.blockEvents, txIndex: null, msgTypeByIndex: new Map() } + ]; + + const changes: DerivedBalanceChange[] = []; + let eventIndex = 0; + + for (const source of sources) { + const scopes = buildScopeContexts(source.events); + + for (const event of source.events) { + const direction = event.type === "coin_spent" ? "spent" : event.type === "coin_received" ? "received" : null; + if (!direction) { + continue; + } + + const holder = direction === "spent" ? event.attributes.spender : event.attributes.receiver; + const scope = scopes.get(scopeKeyOf(event)) ?? { transfers: [], minters: new Set(), burners: new Set(), slashed: false }; + const holderRole = registry.roleOf(holder); + const msgTypeUrl = event.msgIndex === undefined ? null : source.msgTypeByIndex.get(event.msgIndex) ?? null; + const isMint = direction === "received" && scope.minters.has(holder); + const isBurn = direction === "spent" && scope.burners.has(holder); + const isSlash = scope.slashed && isBurn && holderRole !== undefined && STAKING_POOL_ROLES.has(holderRole); + + for (const coin of parseCoins(event.attributes.amount ?? "")) { + const counterpartyAddress = correlateCounterparty(scope, holder, coin.denom, coin.amount, direction); + const reason = classifyReason({ address: holder, counterpartyAddress, denom: coin.denom, isMint, isBurn, isSlash, msgTypeUrl }, registry); + + changes.push({ + address: holder, + counterpartyAddress, + denom: coin.denom, + delta: direction === "spent" ? -coin.amount : coin.amount, + reason, + height: block.height, + txIndex: source.txIndex, + eventIndex: eventIndex++ + }); + } + } + } + + return changes; +} + +function msgTypeByIndexOf(tx: DecodedTransaction): Map { + return new Map(tx.messages.map(message => [message.index, message.typeUrl])); +} diff --git a/apps/chain-indexer/src/pipeline/balance/balance-writer.service.spec.ts b/apps/chain-indexer/src/pipeline/balance/balance-writer.service.spec.ts new file mode 100644 index 0000000000..3148fefc4f --- /dev/null +++ b/apps/chain-indexer/src/pipeline/balance/balance-writer.service.spec.ts @@ -0,0 +1,171 @@ +import type { SQL } from "drizzle-orm"; +import { PgDialect } from "drizzle-orm/pg-core"; +import { describe, expect, it } from "vitest"; + +import { AccountBalances, BalanceChanges } from "@src/db/schema"; +import type { ResolvedBalanceChange } from "@src/pipeline/balance/balance-writer.service"; +import { BalanceWriter } from "@src/pipeline/balance/balance-writer.service"; +import type { ChainTransaction } from "@src/providers/db.provider"; + +describe(BalanceWriter.name, () => { + it("computes each ledger entry's running balance from the ledger baseline", async () => { + const { writer, tx, balanceChangeRows } = setup({ baseline: [{ accountId: 1, denom: "uakt", balanceAfter: "100" }] }); + + await writer.write(tx, [change({ accountId: 1, delta: 50n, eventIndex: 0 }), change({ accountId: 1, delta: -30n, eventIndex: 1 })]); + + expect(balanceChangeRows().map(row => row.balanceAfter)).toEqual(["150", "120"]); + }); + + it("seeds the running balance from zero for an account with no prior ledger history", async () => { + const { writer, tx, balanceChangeRows } = setup(); + + await writer.write(tx, [change({ accountId: 2, delta: 10n, eventIndex: 0 })]); + + expect(balanceChangeRows().map(row => row.balanceAfter)).toEqual(["10"]); + }); + + it("carries the running balance across non-adjacent heights within a batch", async () => { + const { writer, tx, balanceChangeRows } = setup(); + + await writer.write(tx, [change({ accountId: 1, delta: 5n, height: 10, eventIndex: 0 }), change({ accountId: 1, delta: -2n, height: 13, eventIndex: 0 })]); + + expect(balanceChangeRows().map(row => row.balanceAfter)).toEqual(["5", "3"]); + }); + + it("sorts intents by height then event index before accumulating", async () => { + const { writer, tx, balanceChangeRows } = setup(); + + await writer.write(tx, [change({ accountId: 1, delta: -2n, height: 13, eventIndex: 0 }), change({ accountId: 1, delta: 5n, height: 10, eventIndex: 0 })]); + + expect(balanceChangeRows().map(row => ({ height: row.height, balanceAfter: row.balanceAfter }))).toEqual([ + { height: 10, balanceAfter: "5" }, + { height: 13, balanceAfter: "3" } + ]); + }); + + it("advances the current-balance snapshot additively with the summed net delta of the inserted rows", async () => { + const { writer, tx, balanceUpserts, conflictSet } = setup(); + + await writer.write(tx, [change({ accountId: 1, delta: 50n, eventIndex: 0 }), change({ accountId: 1, delta: -30n, eventIndex: 1 })]); + + expect(balanceUpserts()).toEqual([{ accountId: 1, denom: "uakt", amount: "20" }]); + expect(new PgDialect().sqlToQuery(conflictSet()!.amount).sql).toBe('"cosmos"."account_balances"."amount" + EXCLUDED.amount'); + }); + + it("applies zero deltas when a re-commit inserts no new rows", async () => { + const { writer, tx, balanceUpserts } = setup({ insertReturning: [] }); + + await writer.write(tx, [change({ accountId: 1, delta: 50n, eventIndex: 0 })]); + + expect(balanceUpserts()).toEqual([]); + }); + + it("keeps running balances correct while advancing the snapshot only by newly-inserted rows when a batch straddles the sync frontier", async () => { + const { writer, tx, balanceChangeRows, balanceUpserts } = setup({ + baseline: [{ accountId: 1, denom: "uakt", balanceAfter: "100" }], + insertReturning: [{ accountId: 1, denom: "uakt", delta: "7" }] + }); + + await writer.write(tx, [change({ accountId: 1, delta: 5n, height: 10, eventIndex: 0 }), change({ accountId: 1, delta: 7n, height: 11, eventIndex: 0 })]); + + expect(balanceChangeRows().map(row => row.balanceAfter)).toEqual(["105", "112"]); + expect(balanceUpserts()).toEqual([{ accountId: 1, denom: "uakt", amount: "7" }]); + }); + + it("does nothing for an empty intent set", async () => { + const { writer, tx, calls } = setup(); + + await writer.write(tx, []); + + expect(calls()).toBe(0); + }); + + it("reads the baseline in chunks so a batch touching more accounts than the chunk size stays under the bind-parameter limit", async () => { + const chunkSize = 2000; + const accountIds = Array.from({ length: chunkSize + 1 }, (_, index) => index + 1); + const { writer, tx, balanceChangeRows, baselineSelects } = setup({ + baselineByChunk: [[{ accountId: 1, denom: "uakt", balanceAfter: "100" }], [{ accountId: chunkSize + 1, denom: "uakt", balanceAfter: "500" }]] + }); + + await writer.write( + tx, + accountIds.map((accountId, index) => change({ accountId, delta: 10n, eventIndex: index })) + ); + + expect(baselineSelects()).toBe(2); + const balanceAfterByAccount = new Map(balanceChangeRows().map(row => [row.accountId, row.balanceAfter])); + expect(balanceAfterByAccount.get(1)).toBe("110"); + expect(balanceAfterByAccount.get(chunkSize + 1)).toBe("510"); + }); + + function change(input: Partial): ResolvedBalanceChange { + return { + accountId: 1, + counterpartyAccountId: null, + denom: "uakt", + delta: 0n, + reason: "transfer", + height: 10, + txIndex: 0, + eventIndex: 0, + ...input + }; + } + + function setup(input?: { + baseline?: Array<{ accountId: number; denom: string; balanceAfter: string }>; + baselineByChunk?: Array>; + insertReturning?: Array<{ accountId: number; denom: string; delta: string }>; + }) { + const balanceChangeInserts: Record[] = []; + const balanceBalanceUpserts: Record[] = []; + const baselineByChunk = [...(input?.baselineByChunk ?? [])]; + let conflictSet: { amount: SQL } | undefined; + let calls = 0; + let baselineSelects = 0; + + const txFake = { + selectDistinctOn: () => ({ + from: () => ({ + where: () => ({ + orderBy: () => { + calls++; + baselineSelects++; + return Promise.resolve(baselineByChunk.length > 0 ? baselineByChunk.shift()! : input?.baseline ?? []); + } + }) + }) + }), + insert: (table: unknown) => ({ + values: (rows: Record[]) => { + calls++; + if (table === BalanceChanges) { + balanceChangeInserts.push(...rows); + } else if (table === AccountBalances) { + balanceBalanceUpserts.push(...rows); + } + return { + onConflictDoNothing: () => ({ + returning: () => Promise.resolve(input?.insertReturning ?? rows.map(row => ({ accountId: row.accountId, denom: row.denom, delta: row.delta }))) + }), + onConflictDoUpdate: (config: { set: { amount: SQL } }) => { + conflictSet = config.set; + return Promise.resolve(); + } + }; + } + }) + }; + + const writer = new BalanceWriter(); + return { + writer, + tx: txFake as unknown as ChainTransaction, + balanceChangeRows: () => balanceChangeInserts, + balanceUpserts: () => balanceBalanceUpserts, + conflictSet: () => conflictSet, + calls: () => calls, + baselineSelects: () => baselineSelects + }; + } +}); diff --git a/apps/chain-indexer/src/pipeline/balance/balance-writer.service.ts b/apps/chain-indexer/src/pipeline/balance/balance-writer.service.ts new file mode 100644 index 0000000000..162fea71e9 --- /dev/null +++ b/apps/chain-indexer/src/pipeline/balance/balance-writer.service.ts @@ -0,0 +1,152 @@ +import { and, desc, inArray, lt, sql } from "drizzle-orm"; +import chunk from "lodash/chunk"; +import { singleton } from "tsyringe"; + +import { INSERT_CHUNK_SIZE } from "@src/db/insert-chunk-size"; +import { AccountBalances, BalanceChanges } from "@src/db/schema"; +import type { BalanceReason } from "@src/pipeline/balance/reason-classifier"; +import type { ChainTransaction } from "@src/providers/db.provider"; + +/** A derived balance change whose addresses have been interned to account ids, ready to persist. */ +export interface ResolvedBalanceChange { + accountId: number; + counterpartyAccountId: number | null; + denom: string; + delta: bigint; + reason: BalanceReason; + height: number; + txIndex: number | null; + eventIndex: number; +} + +const keyOf = (accountId: number, denom: string) => `${accountId}:${denom}`; + +/** + * Appends balance changes to the ledger and folds them into the current-balance snapshot, idempotently. + * Runs inside the committer's transaction. The `(height, event_index)` unique index is the serialization + * point: `onConflictDoNothing().returning()` yields only rows this call actually inserted, so re-committing + * a block (rolling deploy, backfill overlapping the frontier) inserts zero rows and applies zero deltas. + * The current-balance snapshot is only ever advanced by the returned rows, so it stays a pure projection of + * the ledger and remains rebuildable from it. + */ +@singleton() +export class BalanceWriter { + async write(tx: ChainTransaction, intents: ResolvedBalanceChange[]): Promise { + if (intents.length === 0) { + return; + } + + const ordered = [...intents].sort((a, b) => a.height - b.height || a.eventIndex - b.eventIndex); + const firstHeight = ordered[0].height; + + const baseline = await this.#readLedgerBaseline(tx, ordered, firstHeight); + const changeRows = this.#accumulateRunningBalances(ordered, baseline); + + const inserted = await this.#insertChanges(tx, changeRows); + if (inserted.length === 0) { + return; + } + + await this.#applyNetDeltas(tx, inserted); + } + + /** + * The running balance seeds from the ledger — the `balance_after` of the last change strictly before this + * batch — not from `account_balances`, whose snapshot a concurrent frontier writer may already have advanced. + */ + async #readLedgerBaseline(tx: ChainTransaction, intents: ResolvedBalanceChange[], firstHeight: number): Promise> { + const accountIds = [...new Set(intents.map(intent => intent.accountId))]; + const denoms = [...new Set(intents.map(intent => intent.denom))]; + const touched = new Set(intents.map(intent => keyOf(intent.accountId, intent.denom))); + + const baseline = new Map(); + for (const accountIdChunk of chunk(accountIds, INSERT_CHUNK_SIZE)) { + const rows = await tx + .selectDistinctOn([BalanceChanges.accountId, BalanceChanges.denom], { + accountId: BalanceChanges.accountId, + denom: BalanceChanges.denom, + balanceAfter: BalanceChanges.balanceAfter + }) + .from(BalanceChanges) + .where(and(lt(BalanceChanges.height, firstHeight), inArray(BalanceChanges.accountId, accountIdChunk), inArray(BalanceChanges.denom, denoms))) + .orderBy(BalanceChanges.accountId, BalanceChanges.denom, desc(BalanceChanges.height), desc(BalanceChanges.eventIndex)); + + for (const row of rows) { + const key = keyOf(row.accountId, row.denom); + if (touched.has(key)) { + baseline.set(key, BigInt(row.balanceAfter)); + } + } + } + return baseline; + } + + #accumulateRunningBalances(intents: ResolvedBalanceChange[], baseline: Map): (typeof BalanceChanges.$inferInsert)[] { + const running = new Map(); + + return intents.map(intent => { + const key = keyOf(intent.accountId, intent.denom); + const previous = running.get(key) ?? baseline.get(key) ?? 0n; + const balanceAfter = previous + intent.delta; + running.set(key, balanceAfter); + + return { + accountId: intent.accountId, + denom: intent.denom, + delta: intent.delta.toString(), + balanceAfter: balanceAfter.toString(), + reason: intent.reason, + height: intent.height, + txIndex: intent.txIndex, + eventIndex: intent.eventIndex, + counterpartyAccountId: intent.counterpartyAccountId + }; + }); + } + + async #insertChanges( + tx: ChainTransaction, + changeRows: (typeof BalanceChanges.$inferInsert)[] + ): Promise<{ accountId: number; denom: string; delta: string }[]> { + const inserted: { accountId: number; denom: string; delta: string }[] = []; + + for (const rowChunk of chunk(changeRows, INSERT_CHUNK_SIZE)) { + const returned = await tx + .insert(BalanceChanges) + .values(rowChunk) + .onConflictDoNothing() + .returning({ accountId: BalanceChanges.accountId, denom: BalanceChanges.denom, delta: BalanceChanges.delta }); + inserted.push(...returned); + } + + return inserted; + } + + /** Advances the current balance only by the rows actually inserted, summed per account+denom, so overlapping writers apply each delta exactly once. */ + async #applyNetDeltas(tx: ChainTransaction, inserted: { accountId: number; denom: string; delta: string }[]): Promise { + const netByKey = new Map(); + for (const row of inserted) { + const key = keyOf(row.accountId, row.denom); + const existing = netByKey.get(key); + if (existing) { + existing.amount += BigInt(row.delta); + } else { + netByKey.set(key, { accountId: row.accountId, denom: row.denom, amount: BigInt(row.delta) }); + } + } + + const balanceRows = [...netByKey.values()] + .sort((a, b) => a.accountId - b.accountId || a.denom.localeCompare(b.denom)) + .map(entry => ({ accountId: entry.accountId, denom: entry.denom, amount: entry.amount.toString() })); + + for (const rowChunk of chunk(balanceRows, INSERT_CHUNK_SIZE)) { + await tx + .insert(AccountBalances) + .values(rowChunk) + .onConflictDoUpdate({ + target: [AccountBalances.accountId, AccountBalances.denom], + set: { amount: sql`${AccountBalances.amount} + EXCLUDED.amount` } + }); + } + } +} diff --git a/apps/chain-indexer/src/pipeline/balance/coin-amount.spec.ts b/apps/chain-indexer/src/pipeline/balance/coin-amount.spec.ts new file mode 100644 index 0000000000..8142790775 --- /dev/null +++ b/apps/chain-indexer/src/pipeline/balance/coin-amount.spec.ts @@ -0,0 +1,35 @@ +import { describe, expect, it } from "vitest"; + +import { parseCoins } from "@src/pipeline/balance/coin-amount"; + +describe("parseCoins", () => { + it("parses a single coin", () => { + expect(parseCoins("100uakt")).toEqual([{ denom: "uakt", amount: 100n }]); + }); + + it("parses multiple comma-separated coins preserving order", () => { + expect(parseCoins("100uakt,5uatom")).toEqual([ + { denom: "uakt", amount: 100n }, + { denom: "uatom", amount: 5n } + ]); + }); + + it("parses ibc and factory denoms that contain slashes", () => { + expect(parseCoins("7ibc/27394FB092D2ECCD56123C74F36E4C1F926001CEADA9CA97EA622B25F41E5EB2")).toEqual([ + { denom: "ibc/27394FB092D2ECCD56123C74F36E4C1F926001CEADA9CA97EA622B25F41E5EB2", amount: 7n } + ]); + }); + + it("returns an empty array for an empty or whitespace value", () => { + expect(parseCoins("")).toEqual([]); + expect(parseCoins(" ")).toEqual([]); + }); + + it("skips segments that are not a leading integer followed by a denom", () => { + expect(parseCoins("100uakt,,garbage")).toEqual([{ denom: "uakt", amount: 100n }]); + }); + + it("parses amounts far beyond Number.MAX_SAFE_INTEGER without precision loss", () => { + expect(parseCoins("340282366920938463463374607431768211455uakt")).toEqual([{ denom: "uakt", amount: 340282366920938463463374607431768211455n }]); + }); +}); diff --git a/apps/chain-indexer/src/pipeline/balance/coin-amount.ts b/apps/chain-indexer/src/pipeline/balance/coin-amount.ts new file mode 100644 index 0000000000..bd08334dfb --- /dev/null +++ b/apps/chain-indexer/src/pipeline/balance/coin-amount.ts @@ -0,0 +1,24 @@ +export interface CoinAmount { + denom: string; + amount: bigint; +} + +const COIN_PATTERN = /^(\d+)(.+)$/; + +/** + * Parses a Cosmos coin string such as `"100uakt,5uatom"` into typed amounts. Amounts are `bigint` so + * u-denom values above `Number.MAX_SAFE_INTEGER` keep full precision. Denoms may contain slashes + * (ibc/factory), so the split is on the comma and the amount is the leading integer run only. + */ +export function parseCoins(value: string): CoinAmount[] { + const coins: CoinAmount[] = []; + + for (const segment of value.split(",")) { + const match = segment.trim().match(COIN_PATTERN); + if (match) { + coins.push({ amount: BigInt(match[1]), denom: match[2] }); + } + } + + return coins; +} diff --git a/apps/chain-indexer/src/pipeline/balance/module-address-registry.spec.ts b/apps/chain-indexer/src/pipeline/balance/module-address-registry.spec.ts new file mode 100644 index 0000000000..f6fc484b7c --- /dev/null +++ b/apps/chain-indexer/src/pipeline/balance/module-address-registry.spec.ts @@ -0,0 +1,38 @@ +import { describe, expect, it } from "vitest"; + +import { AKASH_ADDRESS_PREFIX } from "@src/genesis/genesis-address"; +import { BME_VAULT_ADDRESS, buildModuleAddressRegistry, deriveModuleAddress } from "@src/pipeline/balance/module-address-registry"; + +describe("deriveModuleAddress", () => { + it("matches the well-known cosmos-hub fee collector address", () => { + expect(deriveModuleAddress("fee_collector", "cosmos")).toBe("cosmos17xpfvakm2amg962yls6f84z3kell8c5lserqta"); + }); + + it("matches the well-known cosmos-hub distribution address", () => { + expect(deriveModuleAddress("distribution", "cosmos")).toBe("cosmos1jv65s3grqf6v6jl3dp4t6c9t9rk99cd88lyufl"); + }); +}); + +describe("buildModuleAddressRegistry", () => { + it("maps the derived module addresses back to their role", () => { + const registry = buildModuleAddressRegistry(AKASH_ADDRESS_PREFIX); + + expect(registry.roleOf(deriveModuleAddress("fee_collector", AKASH_ADDRESS_PREFIX))).toBe("fee_collector"); + expect(registry.roleOf(deriveModuleAddress("bonded_tokens_pool", AKASH_ADDRESS_PREFIX))).toBe("bonded_tokens_pool"); + expect(registry.roleOf(deriveModuleAddress("not_bonded_tokens_pool", AKASH_ADDRESS_PREFIX))).toBe("not_bonded_tokens_pool"); + expect(registry.roleOf(deriveModuleAddress("transfer", AKASH_ADDRESS_PREFIX))).toBe("ibc_transfer"); + expect(registry.roleOf(deriveModuleAddress("escrow", AKASH_ADDRESS_PREFIX))).toBe("escrow"); + }); + + it("maps the BME vault address to the bme role", () => { + const registry = buildModuleAddressRegistry(AKASH_ADDRESS_PREFIX); + + expect(registry.roleOf(BME_VAULT_ADDRESS)).toBe("bme_vault"); + }); + + it("returns undefined for a non-module address", () => { + const registry = buildModuleAddressRegistry(AKASH_ADDRESS_PREFIX); + + expect(registry.roleOf("akash1regularuseraddressxxxxxxxxxxxxxxxxxxx")).toBeUndefined(); + }); +}); diff --git a/apps/chain-indexer/src/pipeline/balance/module-address-registry.ts b/apps/chain-indexer/src/pipeline/balance/module-address-registry.ts new file mode 100644 index 0000000000..8b252330ff --- /dev/null +++ b/apps/chain-indexer/src/pipeline/balance/module-address-registry.ts @@ -0,0 +1,58 @@ +import { toBech32 } from "@cosmjs/encoding"; +import { createHash } from "node:crypto"; + +import { AKASH_ADDRESS_PREFIX } from "@src/genesis/genesis-address"; + +/** A recognized system account whose involvement in a coin movement identifies the movement's reason. */ +export type ModuleRole = + | "fee_collector" + | "distribution" + | "mint" + | "gov" + | "bonded_tokens_pool" + | "not_bonded_tokens_pool" + | "ibc_transfer" + | "bme_vault" + | "escrow"; + +/** The Akash BME vault, funded by escrow settlements and MsgMintACT and drained by burns; not a `x/auth` module account. */ +export const BME_VAULT_ADDRESS = "akash1klpwzlvfnw7j8gtdd0cuu9vaw9ermsmd37sg55"; + +const MODULE_NAME_ROLES: Record = { + fee_collector: "fee_collector", + distribution: "distribution", + mint: "mint", + gov: "gov", + bonded_tokens_pool: "bonded_tokens_pool", + not_bonded_tokens_pool: "not_bonded_tokens_pool", + transfer: "ibc_transfer", + escrow: "escrow" +}; + +/** + * The bech32 address of a Cosmos SDK module account: the first 20 bytes of `sha256(moduleName)`, matching + * `authtypes.NewModuleAddress`. Derivation is deterministic, so the classifier can recognize a module + * account without needing it seeded from genesis. + */ +export function deriveModuleAddress(moduleName: string, prefix: string): string { + const digest = createHash("sha256").update(Buffer.from(moduleName)).digest(); + return toBech32(prefix, digest.subarray(0, 20)); +} + +export interface ModuleAddressRegistry { + roleOf(address: string): ModuleRole | undefined; +} + +/** Precomputes the address→role map for every known system account so reason classification is a single map lookup. */ +export function buildModuleAddressRegistry(prefix: string = AKASH_ADDRESS_PREFIX): ModuleAddressRegistry { + const roleByAddress = new Map(); + + for (const [moduleName, role] of Object.entries(MODULE_NAME_ROLES)) { + roleByAddress.set(deriveModuleAddress(moduleName, prefix), role); + } + roleByAddress.set(BME_VAULT_ADDRESS, "bme_vault"); + + return { + roleOf: address => roleByAddress.get(address) + }; +} diff --git a/apps/chain-indexer/src/pipeline/balance/reason-classifier.spec.ts b/apps/chain-indexer/src/pipeline/balance/reason-classifier.spec.ts new file mode 100644 index 0000000000..39dfc4da60 --- /dev/null +++ b/apps/chain-indexer/src/pipeline/balance/reason-classifier.spec.ts @@ -0,0 +1,101 @@ +import { describe, expect, it } from "vitest"; + +import { AKASH_ADDRESS_PREFIX } from "@src/genesis/genesis-address"; +import { BME_VAULT_ADDRESS, buildModuleAddressRegistry, deriveModuleAddress } from "@src/pipeline/balance/module-address-registry"; +import type { ReasonContext } from "@src/pipeline/balance/reason-classifier"; +import { classifyReason } from "@src/pipeline/balance/reason-classifier"; + +const registry = buildModuleAddressRegistry(AKASH_ADDRESS_PREFIX); +const moduleAddress = (name: string) => deriveModuleAddress(name, AKASH_ADDRESS_PREFIX); + +describe("classifyReason", () => { + it("classifies a payment to the fee collector as a fee", () => { + expect(classifyReason(context({ counterpartyAddress: moduleAddress("fee_collector") }), registry)).toBe("fee"); + }); + + it("classifies a distribution outflow as a reward", () => { + expect(classifyReason(context({ counterpartyAddress: moduleAddress("distribution") }), registry)).toBe("reward"); + }); + + it("classifies a distribution outflow withdrawn by a validator commission message as commission", () => { + const ctx = context({ counterpartyAddress: moduleAddress("distribution"), msgTypeUrl: "/cosmos.distribution.v1beta1.MsgWithdrawValidatorCommission" }); + expect(classifyReason(ctx, registry)).toBe("commission"); + }); + + it("classifies a flow with the bonded pool as staking", () => { + expect(classifyReason(context({ counterpartyAddress: moduleAddress("bonded_tokens_pool") }), registry)).toBe("staking"); + }); + + it("classifies a flow with the not-bonded pool as staking", () => { + expect(classifyReason(context({ address: moduleAddress("not_bonded_tokens_pool") }), registry)).toBe("staking"); + }); + + it("classifies a gov flow as gov", () => { + expect(classifyReason(context({ counterpartyAddress: moduleAddress("gov") }), registry)).toBe("gov"); + }); + + it("classifies an ibc transfer module flow as ibc", () => { + expect(classifyReason(context({ counterpartyAddress: moduleAddress("transfer") }), registry)).toBe("ibc"); + }); + + it("classifies a flow of an ibc denom as ibc", () => { + expect(classifyReason(context({ denom: "ibc/ABCDEF", counterpartyAddress: "akash1peer" }), registry)).toBe("ibc"); + }); + + it("classifies a deposit whose counterparty is the escrow module as escrow", () => { + expect(classifyReason(context({ counterpartyAddress: moduleAddress("escrow") }), registry)).toBe("escrow"); + }); + + it("classifies a settlement paid out by the escrow module as escrow", () => { + expect(classifyReason(context({ address: moduleAddress("escrow"), counterpartyAddress: "akash1provider" }), registry)).toBe("escrow"); + }); + + it("classifies a flow with the BME vault as bme", () => { + expect(classifyReason(context({ counterpartyAddress: BME_VAULT_ADDRESS }), registry)).toBe("bme"); + }); + + it("prefers slash over any module role", () => { + expect(classifyReason(context({ counterpartyAddress: moduleAddress("bonded_tokens_pool"), isSlash: true }), registry)).toBe("slash"); + }); + + it("classifies a coinbase-coincident credit as mint", () => { + expect(classifyReason(context({ isMint: true }), registry)).toBe("mint"); + }); + + it("classifies a flow whose counterparty is the mint module as mint", () => { + expect(classifyReason(context({ counterpartyAddress: moduleAddress("mint") }), registry)).toBe("mint"); + }); + + it("classifies a burn-coincident debit as burn", () => { + expect(classifyReason(context({ isBurn: true }), registry)).toBe("burn"); + }); + + it("defaults a plain account-to-account movement to transfer", () => { + expect(classifyReason(context({ counterpartyAddress: "akash1peer" }), registry)).toBe("transfer"); + }); + + it("tags each leg of a fee_collector-to-distribution movement by the holder's own role", () => { + const feeCollector = moduleAddress("fee_collector"); + const distribution = moduleAddress("distribution"); + + expect(classifyReason(context({ address: feeCollector, counterpartyAddress: distribution }), registry)).toBe("fee"); + expect(classifyReason(context({ address: distribution, counterpartyAddress: feeCollector }), registry)).toBe("reward"); + }); + + it("tags the mint module's outgoing forwarding leg as mint rather than the counterparty's fee", () => { + expect(classifyReason(context({ address: moduleAddress("mint"), counterpartyAddress: moduleAddress("fee_collector") }), registry)).toBe("mint"); + }); + + function context(overrides: Partial): ReasonContext { + return { + address: "akash1self", + counterpartyAddress: null, + denom: "uakt", + isMint: false, + isBurn: false, + isSlash: false, + msgTypeUrl: null, + ...overrides + }; + } +}); diff --git a/apps/chain-indexer/src/pipeline/balance/reason-classifier.ts b/apps/chain-indexer/src/pipeline/balance/reason-classifier.ts new file mode 100644 index 0000000000..1e1bd748db --- /dev/null +++ b/apps/chain-indexer/src/pipeline/balance/reason-classifier.ts @@ -0,0 +1,60 @@ +import type { balanceChangeReason } from "@src/db/schema"; +import type { ModuleAddressRegistry } from "@src/pipeline/balance/module-address-registry"; + +export type BalanceReason = (typeof balanceChangeReason.enumValues)[number]; + +/** What the classifier knows about one coin movement: who moved it, the correlated counterparty, and any coincident mint/burn/slash. */ +export interface ReasonContext { + address: string; + counterpartyAddress: string | null; + denom: string; + isMint: boolean; + isBurn: boolean; + isSlash: boolean; + msgTypeUrl: string | null; +} + +const WITHDRAW_VALIDATOR_COMMISSION = "/cosmos.distribution.v1beta1.MsgWithdrawValidatorCommission"; + +/** + * MVP reason heuristic. Coincident mint/burn/slash win first (they are unambiguous), then the holder's own + * module role, falling back to the counterparty's, then the denom. Preferring the holder's own role keeps each + * leg of a module-to-module movement (e.g. fee_collector to distribution every block) tagged by the module + * whose balance actually changed, rather than mirroring the counterparty. Anything unrecognized is a plain + * `transfer`. Escrow-module movements classify as `escrow`; per-deployment/lease attribution of that escrow is + * deliberately left for later. + */ +export function classifyReason(ctx: ReasonContext, registry: ModuleAddressRegistry): BalanceReason { + if (ctx.isSlash) { + return "slash"; + } + if (ctx.isMint) { + return "mint"; + } + if (ctx.isBurn) { + return "burn"; + } + + const role = registry.roleOf(ctx.address) ?? (ctx.counterpartyAddress ? registry.roleOf(ctx.counterpartyAddress) : undefined); + switch (role) { + case "mint": + return "mint"; + case "fee_collector": + return "fee"; + case "distribution": + return ctx.msgTypeUrl === WITHDRAW_VALIDATOR_COMMISSION ? "commission" : "reward"; + case "bonded_tokens_pool": + case "not_bonded_tokens_pool": + return "staking"; + case "gov": + return "gov"; + case "ibc_transfer": + return "ibc"; + case "escrow": + return "escrow"; + case "bme_vault": + return "bme"; + } + + return ctx.denom.startsWith("ibc/") ? "ibc" : "transfer"; +} diff --git a/apps/chain-indexer/src/pipeline/block-committer.service.spec.ts b/apps/chain-indexer/src/pipeline/block-committer.service.spec.ts index 3da89ecba0..6c542d2268 100644 --- a/apps/chain-indexer/src/pipeline/block-committer.service.spec.ts +++ b/apps/chain-indexer/src/pipeline/block-committer.service.spec.ts @@ -1,13 +1,17 @@ import type { SQL } from "drizzle-orm"; import { PgDialect } from "drizzle-orm/pg-core"; import { describe, expect, it } from "vitest"; +import { mock } from "vitest-mock-extended"; -import { Blocks, IndexerState, Messages, MessageTypes } from "@src/db/schema"; +import { AccountTxs, Blocks, IndexerState, Messages, MessageTypes } from "@src/db/schema"; +import type { AccountInterner } from "@src/pipeline/balance/account-interner.service"; +import type { BalanceWriter } from "@src/pipeline/balance/balance-writer.service"; import { BlockCommitterService } from "@src/pipeline/block-committer.service"; -import type { DecodedBlock } from "@src/pipeline/decoded-block"; +import type { DecodedBlock, DecodedEvent } from "@src/pipeline/decoded-block"; import type { ChainDatabase } from "@src/providers/db.provider"; const MSG_SEND = "/cosmos.bank.v1beta1.MsgSend"; +const BALANCE_WRITE = Symbol("balance_write"); describe(BlockCommitterService.name, () => { it("reuses ids of message types that already exist instead of inserting them", async () => { @@ -99,6 +103,43 @@ describe(BlockCommitterService.name, () => { }); }); + describe("balance ledger and activity log", () => { + it("writes balances then the activity log inside the transaction, after messages and before the checkpoint", async () => { + const { committer, insertedRows } = setup({ selectResults: [[{ id: 7, type: MSG_SEND }]] }); + + await committer.commit(buildBlock([MSG_SEND], 10, { signerAddresses: ["akash1signer"], events: [transfer("akash1signer", "akash1b", "1uakt")] })); + + const order = insertedRows.map(call => call.table); + expect(order.indexOf(Messages)).toBeLessThan(order.indexOf(BALANCE_WRITE)); + expect(order.indexOf(BALANCE_WRITE)).toBeLessThan(order.indexOf(AccountTxs)); + expect(order.indexOf(AccountTxs)).toBeLessThan(order.indexOf(IndexerState)); + }); + + it("passes balance intents with interned account ids to the balance writer", async () => { + const { committer, balanceWriter } = setup({ selectResults: [[{ id: 7, type: MSG_SEND }]] }); + + await committer.commit(buildBlock([MSG_SEND], 10, { events: [coinSpent("akash1a", "100uakt")] })); + + expect(balanceWriter.write.mock.calls[0][1]).toEqual([ + expect.objectContaining({ accountId: 1, counterpartyAccountId: null, denom: "uakt", delta: -100n, height: 10 }) + ]); + }); + + it("interns signers, spenders, receivers and counterparties in one pass", async () => { + const { committer, interner } = setup({ selectResults: [[{ id: 7, type: MSG_SEND }]] }); + + await committer.commit( + buildBlock([MSG_SEND], 10, { + signerAddresses: ["akash1signer"], + events: [coinSpent("akash1a", "1uakt"), coinReceived("akash1b", "1uakt"), transfer("akash1a", "akash1b", "1uakt")] + }) + ); + + const interned = new Set([...interner.resolve.mock.calls[0][0]]); + expect(interned).toEqual(new Set(["akash1a", "akash1b", "akash1signer"])); + }); + }); + function setup(input?: { selectResults?: Array>; insertReturning?: Array<{ id: number; type: string }> }) { const selectResults = [...(input?.selectResults ?? [[]])]; const insertedRows: Array<{ table: unknown; rows: unknown }> = []; @@ -124,11 +165,23 @@ describe(BlockCommitterService.name, () => { transaction: (callback: (tx: unknown) => Promise) => callback(dbFake) }; - const committer = new BlockCommitterService(dbFake as unknown as ChainDatabase); - return { committer, insertedRows, conflictUpdates }; + const interner = mock(); + interner.resolve.mockImplementation(async addresses => new Map([...addresses].map((address, index) => [address, index + 1]))); + + const balanceWriter = mock(); + balanceWriter.write.mockImplementation(async () => { + insertedRows.push({ table: BALANCE_WRITE, rows: [] }); + }); + + const committer = new BlockCommitterService(dbFake as unknown as ChainDatabase, interner, balanceWriter); + return { committer, insertedRows, conflictUpdates, interner, balanceWriter }; } - function buildBlock(typeUrls: string[], height = 10): DecodedBlock { + function buildBlock( + typeUrls: string[], + height = 10, + tx?: { events?: DecodedEvent[]; signerAddresses?: string[]; blockEvents?: DecodedEvent[] } + ): DecodedBlock { return { height, datetime: new Date("2026-08-11T00:00:00Z"), @@ -143,9 +196,24 @@ describe(BlockCommitterService.name, () => { gasUsed: 0, gasWanted: 0, fee: [], - messages: typeUrls.map((typeUrl, index) => ({ index, typeUrl, body: null })) + messages: typeUrls.map((typeUrl, index) => ({ index, typeUrl, body: null })), + events: tx?.events ?? [], + signerAddresses: tx?.signerAddresses ?? [] } - ] + ], + blockEvents: tx?.blockEvents ?? [] }; } + + function coinSpent(spender: string, amount: string): DecodedEvent { + return { type: "coin_spent", attributes: { spender, amount } }; + } + + function coinReceived(receiver: string, amount: string): DecodedEvent { + return { type: "coin_received", attributes: { receiver, amount } }; + } + + function transfer(sender: string, recipient: string, amount: string): DecodedEvent { + return { type: "transfer", attributes: { sender, recipient, amount } }; + } }); diff --git a/apps/chain-indexer/src/pipeline/block-committer.service.ts b/apps/chain-indexer/src/pipeline/block-committer.service.ts index 17935209f2..b1e9a97c51 100644 --- a/apps/chain-indexer/src/pipeline/block-committer.service.ts +++ b/apps/chain-indexer/src/pipeline/block-committer.service.ts @@ -1,9 +1,16 @@ import { inArray, sql } from "drizzle-orm"; -import chunk from "lodash/chunk"; import { inject, singleton } from "tsyringe"; -import { INSERT_CHUNK_SIZE } from "@src/db/insert-chunk-size"; -import { Blocks, IndexerState, Messages, MessageTypes, Transactions } from "@src/db/schema"; +import { insertChunked } from "@src/db/insert-chunked"; +import { AccountTxs, Blocks, IndexerState, Messages, MessageTypes, Transactions } from "@src/db/schema"; +import { AccountInterner } from "@src/pipeline/balance/account-interner.service"; +import type { DerivedAccountTx } from "@src/pipeline/balance/account-tx-deriver"; +import { deriveAccountTxs } from "@src/pipeline/balance/account-tx-deriver"; +import type { DerivedBalanceChange } from "@src/pipeline/balance/balance-deriver"; +import { deriveBalanceChanges } from "@src/pipeline/balance/balance-deriver"; +import type { ResolvedBalanceChange } from "@src/pipeline/balance/balance-writer.service"; +import { BalanceWriter } from "@src/pipeline/balance/balance-writer.service"; +import { buildModuleAddressRegistry } from "@src/pipeline/balance/module-address-registry"; import type { DecodedBlock } from "@src/pipeline/decoded-block"; import type { ChainDatabase } from "@src/providers/db.provider"; import { CHAIN_DB } from "@src/providers/db.provider"; @@ -13,10 +20,15 @@ export const SYNC_STREAM = "sync"; @singleton() export class BlockCommitterService { readonly #db: ChainDatabase; + readonly #interner: AccountInterner; + readonly #balanceWriter: BalanceWriter; + readonly #moduleRegistry = buildModuleAddressRegistry(); readonly #typeIds = new Map(); - constructor(@inject(CHAIN_DB) db: ChainDatabase) { + constructor(@inject(CHAIN_DB) db: ChainDatabase, @inject(AccountInterner) interner: AccountInterner, @inject(BalanceWriter) balanceWriter: BalanceWriter) { this.#db = db; + this.#interner = interner; + this.#balanceWriter = balanceWriter; } async commit(block: DecodedBlock): Promise { @@ -70,20 +82,21 @@ export class BlockCommitterService { ) ); + const balanceChanges = blocks.flatMap(block => deriveBalanceChanges(block, this.#moduleRegistry)); + const accountTxs = blocks.flatMap(block => deriveAccountTxs(block)); + const accountIds = await this.#internAccounts(balanceChanges, accountTxs); + const balanceIntents = this.#resolveBalanceChanges(balanceChanges, accountIds); + const accountTxRows = this.#resolveAccountTxs(accountTxs, accountIds); + const lastHeight = blocks[blocks.length - 1].height; await this.#db.transaction(async tx => { - for (const blockChunk of chunk(blockRows, INSERT_CHUNK_SIZE)) { - await tx.insert(Blocks).values(blockChunk).onConflictDoNothing(); - } - - for (const transactionChunk of chunk(transactionRows, INSERT_CHUNK_SIZE)) { - await tx.insert(Transactions).values(transactionChunk).onConflictDoNothing(); - } + await insertChunked(tx, Blocks, blockRows); + await insertChunked(tx, Transactions, transactionRows); + await insertChunked(tx, Messages, messageRows); - for (const messageChunk of chunk(messageRows, INSERT_CHUNK_SIZE)) { - await tx.insert(Messages).values(messageChunk).onConflictDoNothing(); - } + await this.#balanceWriter.write(tx, balanceIntents); + await insertChunked(tx, AccountTxs, accountTxRows); await tx .insert(IndexerState) @@ -108,6 +121,52 @@ export class BlockCommitterService { }); } + /** + * Interns every address the batch touches — spenders, receivers, correlated counterparties and tx signers — + * on the base connection before the commit transaction, so the ledger and activity rows can reference their + * account ids by foreign key. + */ + async #internAccounts(balanceChanges: DerivedBalanceChange[], accountTxs: DerivedAccountTx[]): Promise> { + const addresses = new Set(); + + for (const change of balanceChanges) { + addresses.add(change.address); + if (change.counterpartyAddress) { + addresses.add(change.counterpartyAddress); + } + } + for (const row of accountTxs) { + addresses.add(row.address); + } + + return this.#interner.resolve(addresses); + } + + #resolveBalanceChanges(changes: DerivedBalanceChange[], accountIds: Map): ResolvedBalanceChange[] { + return changes.map(change => ({ + accountId: this.#requireId(accountIds, change.address), + counterpartyAccountId: change.counterpartyAddress ? accountIds.get(change.counterpartyAddress) ?? null : null, + denom: change.denom, + delta: change.delta, + reason: change.reason, + height: change.height, + txIndex: change.txIndex, + eventIndex: change.eventIndex + })); + } + + #resolveAccountTxs(rows: DerivedAccountTx[], accountIds: Map): (typeof AccountTxs.$inferInsert)[] { + return rows.map(row => ({ accountId: this.#requireId(accountIds, row.address), height: row.height, txIndex: row.txIndex, role: row.role })); + } + + #requireId(accountIds: Map, address: string): number { + const accountId = accountIds.get(address); + if (accountId === undefined) { + throw new Error(`No interned account id for address ${address}`); + } + return accountId; + } + async #internMessageTypes(blocks: DecodedBlock[]): Promise> { const typeUrls = new Set(blocks.flatMap(block => block.transactions.flatMap(tx => tx.messages.map(message => message.typeUrl)))); const uncached = [...typeUrls].filter(typeUrl => !this.#typeIds.has(typeUrl)); diff --git a/apps/chain-indexer/src/pipeline/block-decoder.service.spec.ts b/apps/chain-indexer/src/pipeline/block-decoder.service.spec.ts index ede66dbc3b..e5bee1a834 100644 --- a/apps/chain-indexer/src/pipeline/block-decoder.service.spec.ts +++ b/apps/chain-indexer/src/pipeline/block-decoder.service.spec.ts @@ -7,7 +7,7 @@ import { describe, expect, it } from "vitest"; import { envSchema } from "@src/config/env.config"; import { BlockDecoderService } from "@src/pipeline/block-decoder.service"; -import type { RpcBlockResult, RpcBlockResultsResult } from "@src/rpc/rpc-types"; +import type { RpcBlockResult, RpcBlockResultsResult, RpcEvent, RpcTxResult } from "@src/rpc/rpc-types"; describe(BlockDecoderService.name, () => { it("decodes block metadata with hashes as buffers", () => { @@ -81,6 +81,91 @@ describe(BlockDecoderService.name, () => { expect(decoded.transactions[0].code).toBe(11); }); + it("captures relevant tx events with their msg_index and drops irrelevant ones", () => { + const { decoder } = setup(); + const rawTx = buildMsgSendTx(); + const txResult: RpcTxResult = { + code: 0, + gas_used: "0", + gas_wanted: "0", + events: [ + event("coin_spent", { spender: "akash1from", amount: "42uakt", msg_index: "0" }), + event("tx", { fee: "5000uakt" }), + event("message", { action: "/cosmos.bank.v1beta1.MsgSend" }), + event("coin_received", { receiver: "akash1to", amount: "42uakt", msg_index: "0" }) + ] + }; + + const [tx] = decoder.decode(buildBlock({ txs: [rawTx.toString("base64")] }), buildBlockResults([txResult])).transactions; + + expect(tx.events).toEqual([ + { type: "coin_spent", attributes: { spender: "akash1from", amount: "42uakt", msg_index: "0" }, msgIndex: 0 }, + { type: "coin_received", attributes: { receiver: "akash1to", amount: "42uakt", msg_index: "0" }, msgIndex: 0 } + ]); + }); + + it("normalizes base64-encoded event attributes", () => { + const { decoder } = setup(); + const rawTx = buildMsgSendTx(); + const txResult: RpcTxResult = { + code: 0, + gas_used: "0", + gas_wanted: "0", + events: [event("coin_spent", { "c3BlbmRlcg==": "YWthc2gxZnJvbQ==" })] + }; + + const [tx] = decoder.decode(buildBlock({ txs: [rawTx.toString("base64")] }), buildBlockResults([txResult])).transactions; + + expect(tx.events[0].attributes).toEqual({ spender: "akash1from" }); + }); + + it("keeps a failed transaction's fee coin events", () => { + const { decoder } = setup(); + const rawTx = buildMsgSendTx(); + const txResult: RpcTxResult = { + code: 11, + gas_used: "70000", + gas_wanted: "70000", + events: [ + event("coin_spent", { spender: "akash1payer", amount: "5000uakt" }), + event("coin_received", { receiver: "akash1feecollector", amount: "5000uakt" }) + ] + }; + + const [tx] = decoder.decode(buildBlock({ txs: [rawTx.toString("base64")] }), buildBlockResults([txResult])).transactions; + + expect(tx.code).toBe(11); + expect(tx.events.map(e => e.type)).toEqual(["coin_spent", "coin_received"]); + }); + + it("prefers finalize_block_events for block-level events", () => { + const { decoder } = setup(); + + const decoded = decoder.decode( + buildBlock({ txs: [] }), + buildBlockResults([], { + finalize_block_events: [event("coinbase", { minter: "akash1mint", amount: "10uakt" })], + begin_block_events: [event("transfer", { sender: "ignored", recipient: "ignored", amount: "1uakt" })] + }) + ); + + expect(decoded.blockEvents).toEqual([{ type: "coinbase", attributes: { minter: "akash1mint", amount: "10uakt" } }]); + }); + + it("falls back to begin and end block events when finalize is absent", () => { + const { decoder } = setup(); + + const decoded = decoder.decode( + buildBlock({ txs: [] }), + buildBlockResults([], { + begin_block_events: [event("coin_received", { receiver: "akash1begin", amount: "1uakt" })], + end_block_events: [event("coin_spent", { spender: "akash1end", amount: "2uakt" })] + }) + ); + + expect(decoded.blockEvents.map(e => e.type)).toEqual(["coin_received", "coin_spent"]); + }); + function setup(input?: { maxBodyBytes?: number }) { const config = envSchema.parse({ POSTGRES_DB_URI: "postgres://unit:unit@localhost:5432/unit", @@ -105,8 +190,15 @@ describe(BlockDecoderService.name, () => { }; } - function buildBlockResults(txsResults: { code: number; gas_used: string; gas_wanted: string }[]): RpcBlockResultsResult { - return { height: "1234", txs_results: txsResults }; + function buildBlockResults( + txsResults: RpcTxResult[], + blockEvents?: Partial> + ): RpcBlockResultsResult { + return { height: "1234", txs_results: txsResults, ...blockEvents }; + } + + function event(type: string, attributes: Record): RpcEvent { + return { type, attributes: Object.entries(attributes).map(([key, value]) => ({ key, value })) }; } function buildMsgSendTx(typeUrl = "/cosmos.bank.v1beta1.MsgSend"): Buffer { diff --git a/apps/chain-indexer/src/pipeline/block-decoder.service.ts b/apps/chain-indexer/src/pipeline/block-decoder.service.ts index 987f68b935..1c9bed1ff8 100644 --- a/apps/chain-indexer/src/pipeline/block-decoder.service.ts +++ b/apps/chain-indexer/src/pipeline/block-decoder.service.ts @@ -5,11 +5,18 @@ import { inject, singleton } from "tsyringe"; import type { EnvConfig } from "@src/config/env.config"; import { toCanonicalJson } from "@src/pipeline/canonical-json"; -import type { DecodedBlock, DecodedMessage, DecodedTransaction } from "@src/pipeline/decoded-block"; +import { decodeIfBase64 } from "@src/pipeline/decode-if-base64"; +import type { DecodedBlock, DecodedEvent, DecodedMessage, DecodedTransaction } from "@src/pipeline/decoded-block"; +import { deriveSignerAddresses } from "@src/pipeline/signer-addresses"; import { APP_CONFIG } from "@src/providers/app-config.provider"; import type { Registry } from "@src/providers/type-registry.provider"; import { TYPE_REGISTRY } from "@src/providers/type-registry.provider"; -import type { RpcBlockResult, RpcBlockResultsResult, RpcTxResult } from "@src/rpc/rpc-types"; +import type { RpcBlockResult, RpcBlockResultsResult, RpcEvent, RpcTxResult } from "@src/rpc/rpc-types"; + +/** The ledger derives balances and reasons only from these event types; capturing the rest would waste memory across backfill batches. */ +const RELEVANT_EVENT_TYPES = new Set(["coin_spent", "coin_received", "transfer", "coinbase", "burn", "slash"]); + +const MSG_INDEX_ATTRIBUTE = "msg_index"; @singleton() export class BlockDecoderService { @@ -35,7 +42,8 @@ export class BlockDecoderService { hash: Buffer.from(block.block_id.hash, "hex"), parentHash: block.block.header.last_block_id?.hash ? Buffer.from(block.block.header.last_block_id.hash, "hex") : null, proposerAddress: block.block.header.proposer_address, - transactions: rawTxs.map((rawTx, index) => this.#decodeTransaction(rawTx, txResults[index], index)) + transactions: rawTxs.map((rawTx, index) => this.#decodeTransaction(rawTx, txResults[index], index)), + blockEvents: this.#decodeBlockEvents(blockResults) }; } @@ -50,10 +58,46 @@ export class BlockDecoderService { gasUsed: parseInt(txResult.gas_used ?? "0"), gasWanted: parseInt(txResult.gas_wanted ?? "0"), fee: decodedTx.authInfo.fee?.amount.map(({ denom, amount }) => ({ denom, amount })) ?? [], - messages: decodedTx.body.messages.map((message, messageIndex) => this.#decodeMessage(message, messageIndex)) + messages: decodedTx.body.messages.map((message, messageIndex) => this.#decodeMessage(message, messageIndex)), + events: this.#decodeEvents(txResult.events), + signerAddresses: deriveSignerAddresses(decodedTx.authInfo.signerInfos) }; } + /** + * ABCI 2.0 (CometBFT 0.38+) merges begin/end block events into `finalize_block_events`; older nodes split + * them. Mirrors the legacy indexer's `finalize_block_events ?? [...begin, ...end]` normalization. + */ + #decodeBlockEvents(blockResults: RpcBlockResultsResult): DecodedEvent[] { + const rawEvents = blockResults.finalize_block_events ?? [...(blockResults.begin_block_events ?? []), ...(blockResults.end_block_events ?? [])]; + return this.#decodeEvents(rawEvents); + } + + #decodeEvents(rawEvents: RpcEvent[] | undefined): DecodedEvent[] { + if (!rawEvents) { + return []; + } + + return rawEvents.filter(event => RELEVANT_EVENT_TYPES.has(event.type)).map(event => this.#decodeEvent(event)); + } + + #decodeEvent(event: RpcEvent): DecodedEvent { + const attributes: Record = {}; + let msgIndex: number | undefined; + + for (const attribute of event.attributes) { + const key = decodeIfBase64(attribute.key); + const value = attribute.value ? decodeIfBase64(attribute.value) : ""; + attributes[key] = value; + + if (key === MSG_INDEX_ATTRIBUTE) { + msgIndex = parseInt(value); + } + } + + return msgIndex === undefined ? { type: event.type, attributes } : { type: event.type, attributes, msgIndex }; + } + #decodeMessage(message: { typeUrl: string; value: Uint8Array }, index: number): DecodedMessage { return { index, diff --git a/apps/chain-indexer/src/pipeline/decode-if-base64.spec.ts b/apps/chain-indexer/src/pipeline/decode-if-base64.spec.ts new file mode 100644 index 0000000000..f87d075eda --- /dev/null +++ b/apps/chain-indexer/src/pipeline/decode-if-base64.spec.ts @@ -0,0 +1,29 @@ +import { describe, expect, it } from "vitest"; + +import { decodeIfBase64 } from "@src/pipeline/decode-if-base64"; + +describe("decodeIfBase64", () => { + it("decodes a base64-encoded printable-ascii value", () => { + expect(decodeIfBase64("c3BlbmRlcg==")).toBe("spender"); + }); + + it("returns an already-plaintext value untouched", () => { + expect(decodeIfBase64("spender")).toBe("spender"); + }); + + it("returns a plaintext value that happens to be valid base64 untouched when it stays printable", () => { + expect(decodeIfBase64("akash1abcd")).toBe("akash1abcd"); + }); + + it("leaves an empty string untouched", () => { + expect(decodeIfBase64("")).toBe(""); + }); + + it("returns a value whose length is not a multiple of four untouched", () => { + expect(decodeIfBase64("abc")).toBe("abc"); + }); + + it("keeps a value that decodes to non-printable bytes as the original", () => { + expect(decodeIfBase64("////")).toBe("////"); + }); +}); diff --git a/apps/chain-indexer/src/pipeline/decode-if-base64.ts b/apps/chain-indexer/src/pipeline/decode-if-base64.ts new file mode 100644 index 0000000000..26dd4ea6a2 --- /dev/null +++ b/apps/chain-indexer/src/pipeline/decode-if-base64.ts @@ -0,0 +1,35 @@ +const BASE64_PATTERN = /^[A-Za-z0-9+/]+={0,2}$/; + +function isPrintableAscii(value: string): boolean { + for (let i = 0; i < value.length; i++) { + const code = value.charCodeAt(i); + if (code < 32 || code > 126) { + return false; + } + } + return true; +} + +/** + * ABCI event attributes arrive base64-encoded on some CometBFT versions and plaintext on others, so the + * decoder normalizes every key/value through this guard. A value is only decoded when it round-trips + * through base64 exactly and yields printable ASCII, which keeps genuine plaintext (even plaintext that + * happens to be valid base64) untouched. Ported from the legacy indexer to stay node-version agnostic. + */ +export function decodeIfBase64(value: string): string { + if (!value || value.length % 4 !== 0 || !BASE64_PATTERN.test(value)) { + return value; + } + + try { + const decoded = atob(value); + + if (btoa(decoded) !== value) { + return value; + } + + return isPrintableAscii(decoded) ? decoded : value; + } catch { + return value; + } +} diff --git a/apps/chain-indexer/src/pipeline/decoded-block.ts b/apps/chain-indexer/src/pipeline/decoded-block.ts index c1225790a0..51acfb247a 100644 --- a/apps/chain-indexer/src/pipeline/decoded-block.ts +++ b/apps/chain-indexer/src/pipeline/decoded-block.ts @@ -6,6 +6,13 @@ export interface DecodedMessage { body: unknown | null; } +/** An ABCI event with base64-normalized attributes flattened to a key→value map. `msgIndex` links the event to its message where present. */ +export interface DecodedEvent { + type: string; + attributes: Record; + msgIndex?: number; +} + export interface DecodedTransaction { index: number; hash: Buffer; @@ -14,6 +21,8 @@ export interface DecodedTransaction { gasWanted: number; fee: FeeCoin[]; messages: DecodedMessage[]; + events: DecodedEvent[]; + signerAddresses: string[]; } export interface DecodedBlock { @@ -23,4 +32,5 @@ export interface DecodedBlock { parentHash: Buffer | null; proposerAddress: string; transactions: DecodedTransaction[]; + blockEvents: DecodedEvent[]; } diff --git a/apps/chain-indexer/src/pipeline/signer-addresses.spec.ts b/apps/chain-indexer/src/pipeline/signer-addresses.spec.ts new file mode 100644 index 0000000000..45bb11eea8 --- /dev/null +++ b/apps/chain-indexer/src/pipeline/signer-addresses.spec.ts @@ -0,0 +1,38 @@ +import { createMultisigThresholdPubkey, encodeSecp256k1Pubkey, pubkeyToAddress } from "@cosmjs/amino"; +import { encodePubkey } from "@cosmjs/proto-signing"; +import { SignerInfo } from "cosmjs-types/cosmos/tx/v1beta1/tx"; +import { describe, expect, it } from "vitest"; + +import { AKASH_ADDRESS_PREFIX } from "@src/genesis/genesis-address"; +import { deriveSignerAddresses } from "@src/pipeline/signer-addresses"; + +function secp256k1Pubkey(firstByte: number) { + return encodeSecp256k1Pubkey(new Uint8Array([firstByte, ...new Array(32).fill(1)])); +} + +function signerInfoWith(pubkey: Parameters[0]) { + return SignerInfo.fromPartial({ publicKey: encodePubkey(pubkey) }); +} + +describe("deriveSignerAddresses", () => { + it("derives the account address of a single secp256k1 signer", () => { + const pubkey = secp256k1Pubkey(2); + + expect(deriveSignerAddresses([signerInfoWith(pubkey)])).toEqual([pubkeyToAddress(pubkey, AKASH_ADDRESS_PREFIX)]); + }); + + it("derives every member address of a multisig signer", () => { + const first = secp256k1Pubkey(2); + const second = secp256k1Pubkey(3); + const multisig = createMultisigThresholdPubkey([first, second], 1); + + expect(deriveSignerAddresses([signerInfoWith(multisig)])).toEqual([ + pubkeyToAddress(first, AKASH_ADDRESS_PREFIX), + pubkeyToAddress(second, AKASH_ADDRESS_PREFIX) + ]); + }); + + it("skips signer infos without a public key", () => { + expect(deriveSignerAddresses([SignerInfo.fromPartial({})])).toEqual([]); + }); +}); diff --git a/apps/chain-indexer/src/pipeline/signer-addresses.ts b/apps/chain-indexer/src/pipeline/signer-addresses.ts new file mode 100644 index 0000000000..b0de801bb6 --- /dev/null +++ b/apps/chain-indexer/src/pipeline/signer-addresses.ts @@ -0,0 +1,44 @@ +import type { Pubkey, SinglePubkey } from "@cosmjs/amino"; +import { isMultisigThresholdPubkey, isSinglePubkey, pubkeyToAddress } from "@cosmjs/amino"; +import { decodePubkey } from "@cosmjs/proto-signing"; +import type { SignerInfo } from "cosmjs-types/cosmos/tx/v1beta1/tx"; + +import { AKASH_ADDRESS_PREFIX } from "@src/genesis/genesis-address"; + +function flattenSinglePubkeys(pubkey: Pubkey): SinglePubkey[] { + if (isMultisigThresholdPubkey(pubkey)) { + return pubkey.value.pubkeys.flatMap(flattenSinglePubkeys); + } + + return isSinglePubkey(pubkey) ? [pubkey] : []; +} + +/** + * Bech32 account addresses of every signer of a transaction. A multisig signer expands to one address per + * member key, mirroring the legacy indexer. An undecodable pubkey (e.g. a legacy amino multisig) yields no + * address for that signer rather than failing the whole block. + */ +export function deriveSignerAddresses(signerInfos: readonly SignerInfo[]): string[] { + const addresses: string[] = []; + + for (const signerInfo of signerInfos) { + if (!signerInfo.publicKey) { + continue; + } + + try { + const pubkey = decodePubkey(signerInfo.publicKey); + if (!pubkey) { + continue; + } + + for (const single of flattenSinglePubkeys(pubkey)) { + addresses.push(pubkeyToAddress(single, AKASH_ADDRESS_PREFIX)); + } + } catch { + continue; + } + } + + return addresses; +} diff --git a/apps/chain-indexer/src/reconcile/bank-query.spec.ts b/apps/chain-indexer/src/reconcile/bank-query.spec.ts new file mode 100644 index 0000000000..0944d3ad25 --- /dev/null +++ b/apps/chain-indexer/src/reconcile/bank-query.spec.ts @@ -0,0 +1,44 @@ +import { toBase64 } from "@cosmjs/encoding"; +import { QueryAllBalancesResponse, QueryTotalSupplyResponse } from "cosmjs-types/cosmos/bank/v1beta1/query"; +import { describe, expect, it } from "vitest"; + +import { + ALL_BALANCES_PATH, + decodeAllBalances, + decodeTotalSupply, + encodeAllBalancesRequest, + encodeTotalSupplyRequest, + TOTAL_SUPPLY_PATH +} from "@src/reconcile/bank-query"; + +describe("bank-query", () => { + it("encodes an all-balances request as hex carrying the address", () => { + const hex = encodeAllBalancesRequest("akash1abc"); + + expect(hex).toMatch(/^[0-9a-f]+$/); + expect(Buffer.from(hex, "hex").toString("utf8")).toContain("akash1abc"); + expect(ALL_BALANCES_PATH).toBe("/cosmos.bank.v1beta1.Query/AllBalances"); + }); + + it("encodes a total-supply request as hex", () => { + expect(encodeTotalSupplyRequest()).toMatch(/^[0-9a-f]*$/); + expect(TOTAL_SUPPLY_PATH).toBe("/cosmos.bank.v1beta1.Query/TotalSupply"); + }); + + it("decodes an all-balances response into typed coins", () => { + const value = toBase64(QueryAllBalancesResponse.encode({ balances: [{ denom: "uakt", amount: "42" }], pagination: undefined }).finish()); + + expect(decodeAllBalances(value)).toEqual([{ denom: "uakt", amount: 42n }]); + }); + + it("decodes a total-supply response into typed coins", () => { + const value = toBase64(QueryTotalSupplyResponse.encode({ supply: [{ denom: "uakt", amount: "1000" }], pagination: undefined }).finish()); + + expect(decodeTotalSupply(value)).toEqual([{ denom: "uakt", amount: 1000n }]); + }); + + it("decodes an empty value as no coins", () => { + expect(decodeAllBalances(null)).toEqual([]); + expect(decodeTotalSupply(null)).toEqual([]); + }); +}); diff --git a/apps/chain-indexer/src/reconcile/bank-query.ts b/apps/chain-indexer/src/reconcile/bank-query.ts new file mode 100644 index 0000000000..0a3770f6ad --- /dev/null +++ b/apps/chain-indexer/src/reconcile/bank-query.ts @@ -0,0 +1,40 @@ +import { fromBase64, toHex } from "@cosmjs/encoding"; +import { QueryAllBalancesRequest, QueryAllBalancesResponse, QueryTotalSupplyRequest, QueryTotalSupplyResponse } from "cosmjs-types/cosmos/bank/v1beta1/query"; + +import type { CoinAmount } from "@src/pipeline/balance/coin-amount"; + +export const ALL_BALANCES_PATH = "/cosmos.bank.v1beta1.Query/AllBalances"; +export const TOTAL_SUPPLY_PATH = "/cosmos.bank.v1beta1.Query/TotalSupply"; + +/** Cosmos paginates bank queries; a single large page covers any account's handful of denoms and the chain's denom set. */ +const PAGE_LIMIT = 10_000n; + +export function encodeAllBalancesRequest(address: string): string { + return toHex(QueryAllBalancesRequest.encode(QueryAllBalancesRequest.fromPartial({ address, pagination: pageRequest() })).finish()); +} + +export function encodeTotalSupplyRequest(): string { + return toHex(QueryTotalSupplyRequest.encode(QueryTotalSupplyRequest.fromPartial({ pagination: pageRequest() })).finish()); +} + +export function decodeAllBalances(value: string | null): CoinAmount[] { + if (!value) { + return []; + } + return toCoinAmounts(QueryAllBalancesResponse.decode(fromBase64(value)).balances); +} + +export function decodeTotalSupply(value: string | null): CoinAmount[] { + if (!value) { + return []; + } + return toCoinAmounts(QueryTotalSupplyResponse.decode(fromBase64(value)).supply); +} + +function pageRequest() { + return { key: new Uint8Array(), offset: 0n, limit: PAGE_LIMIT, countTotal: false, reverse: false }; +} + +function toCoinAmounts(coins: { denom: string; amount: string }[]): CoinAmount[] { + return coins.map(coin => ({ denom: coin.denom, amount: BigInt(coin.amount) })); +} diff --git a/apps/chain-indexer/src/reconcile/coin-diff.spec.ts b/apps/chain-indexer/src/reconcile/coin-diff.spec.ts new file mode 100644 index 0000000000..951f7dfe27 --- /dev/null +++ b/apps/chain-indexer/src/reconcile/coin-diff.spec.ts @@ -0,0 +1,33 @@ +import { describe, expect, it } from "vitest"; + +import { diffCoins } from "@src/reconcile/coin-diff"; + +describe("diffCoins", () => { + it("returns no differences when both sides match", () => { + expect(diffCoins([{ denom: "uakt", amount: 100n }], [{ denom: "uakt", amount: 100n }])).toEqual([]); + }); + + it("reports a denom whose amounts differ", () => { + expect(diffCoins([{ denom: "uakt", amount: 100n }], [{ denom: "uakt", amount: 90n }])).toEqual([{ denom: "uakt", expected: 100n, actual: 90n }]); + }); + + it("reports a denom present only on the chain as a zero ledger balance", () => { + expect(diffCoins([{ denom: "uakt", amount: 5n }], [])).toEqual([{ denom: "uakt", expected: 5n, actual: 0n }]); + }); + + it("reports a denom present only in the ledger as a zero chain balance", () => { + expect(diffCoins([], [{ denom: "uakt", amount: 5n }])).toEqual([{ denom: "uakt", expected: 0n, actual: 5n }]); + }); + + it("orders differences by denom", () => { + expect( + diffCoins( + [ + { denom: "uosmo", amount: 1n }, + { denom: "uakt", amount: 1n } + ], + [] + ).map(diff => diff.denom) + ).toEqual(["uakt", "uosmo"]); + }); +}); diff --git a/apps/chain-indexer/src/reconcile/coin-diff.ts b/apps/chain-indexer/src/reconcile/coin-diff.ts new file mode 100644 index 0000000000..eb8697dbb1 --- /dev/null +++ b/apps/chain-indexer/src/reconcile/coin-diff.ts @@ -0,0 +1,30 @@ +import type { CoinAmount } from "@src/pipeline/balance/coin-amount"; + +/** A denom whose chain-queried (`expected`) and ledger-derived (`actual`) amounts disagree. */ +export interface CoinDiff { + denom: string; + expected: bigint; + actual: bigint; +} + +function toMap(coins: CoinAmount[]): Map { + return new Map(coins.map(coin => [coin.denom, coin.amount])); +} + +/** Compares chain-queried balances (`expected`) against ledger-derived balances (`actual`), returning only the denoms that differ. */ +export function diffCoins(expected: CoinAmount[], actual: CoinAmount[]): CoinDiff[] { + const expectedByDenom = toMap(expected); + const actualByDenom = toMap(actual); + const denoms = [...new Set([...expectedByDenom.keys(), ...actualByDenom.keys()])].sort(); + + const diffs: CoinDiff[] = []; + for (const denom of denoms) { + const expectedAmount = expectedByDenom.get(denom) ?? 0n; + const actualAmount = actualByDenom.get(denom) ?? 0n; + if (expectedAmount !== actualAmount) { + diffs.push({ denom, expected: expectedAmount, actual: actualAmount }); + } + } + + return diffs; +} diff --git a/apps/chain-indexer/src/reconcile/reconcile.service.spec.ts b/apps/chain-indexer/src/reconcile/reconcile.service.spec.ts new file mode 100644 index 0000000000..5b28f62d37 --- /dev/null +++ b/apps/chain-indexer/src/reconcile/reconcile.service.spec.ts @@ -0,0 +1,132 @@ +import { toBase64 } from "@cosmjs/encoding"; +import { QueryAllBalancesResponse, QueryTotalSupplyResponse } from "cosmjs-types/cosmos/bank/v1beta1/query"; +import { describe, expect, it, vi } from "vitest"; +import { mock } from "vitest-mock-extended"; + +import { IndexerState } from "@src/db/schema"; +import type { ChainDatabase } from "@src/providers/db.provider"; +import type { LoggerService } from "@src/providers/logging.provider"; +import { ALL_BALANCES_PATH } from "@src/reconcile/bank-query"; +import { ReconcileService } from "@src/reconcile/reconcile.service"; +import type { RpcClientPool } from "@src/rpc/rpc-client-pool.service"; + +const coin = (denom: string, amount: string) => ({ denom, amount }); + +describe(ReconcileService.name, () => { + it("returns true when every sampled account and the total supply match the chain", async () => { + const { service, abciQuery } = setup({ + checkpoint: 100, + balanceRows: [{ address: "akash1a", denom: "uakt", amount: "100" }], + chainBalances: { akash1a: [coin("uakt", "100")] }, + chainSupply: [coin("uakt", "100")] + }); + + await expect(service.reconcile()).resolves.toBe(true); + expect(abciQuery.mock.calls.every(call => call[2] === 100)).toBe(true); + }); + + it("returns false when a sampled account balance disagrees with the chain", async () => { + const { service } = setup({ + checkpoint: 100, + balanceRows: [{ address: "akash1a", denom: "uakt", amount: "100" }], + chainBalances: { akash1a: [coin("uakt", "90")] }, + chainSupply: [coin("uakt", "100")] + }); + + await expect(service.reconcile()).resolves.toBe(false); + }); + + it("returns false when the total supply disagrees with the ledger", async () => { + const { service } = setup({ + checkpoint: 100, + balanceRows: [{ address: "akash1a", denom: "uakt", amount: "100" }], + chainBalances: { akash1a: [coin("uakt", "100")] }, + chainSupply: [coin("uakt", "999")] + }); + + await expect(service.reconcile()).resolves.toBe(false); + }); + + it("returns false when there is no sync checkpoint to reconcile against", async () => { + const { service } = setup({ checkpoint: undefined, balanceRows: [], chainBalances: {}, chainSupply: [] }); + + await expect(service.reconcile()).resolves.toBe(false); + }); + + it("fails fast on a non-integer sample size instead of silently checking nothing", async () => { + const { service, abciQuery } = setup({ + checkpoint: 100, + balanceRows: [{ address: "akash1a", denom: "uakt", amount: "100" }], + chainBalances: { akash1a: [coin("uakt", "100")] }, + chainSupply: [coin("uakt", "100")] + }); + + await expect(service.reconcile({ sampleSize: NaN })).resolves.toBe(false); + expect(abciQuery).not.toHaveBeenCalled(); + }); + + it("reads the checkpoint height and ledger balances from one repeatable-read snapshot", async () => { + const { service, transaction, baseSelect, txSelect } = setup({ + checkpoint: 100, + balanceRows: [{ address: "akash1a", denom: "uakt", amount: "100" }], + chainBalances: { akash1a: [coin("uakt", "100")] }, + chainSupply: [coin("uakt", "100")] + }); + + await service.reconcile(); + + expect(transaction).toHaveBeenCalledWith(expect.any(Function), { isolationLevel: "repeatable read", accessMode: "read only" }); + expect(baseSelect).not.toHaveBeenCalled(); + expect(txSelect).toHaveBeenCalledTimes(2); + }); + + it("caps concurrent balance queries instead of issuing them one at a time", async () => { + const balanceRows = Array.from({ length: 25 }, (_, index) => ({ address: `akash1a${index}`, denom: "uakt", amount: "100" })); + const chainBalances = Object.fromEntries(balanceRows.map(row => [row.address, [coin("uakt", "100")]])); + const { service, maxInFlight } = setup({ checkpoint: 100, balanceRows, chainBalances, chainSupply: [coin("uakt", "2500")] }); + + await service.reconcile(); + + expect(maxInFlight()).toBeGreaterThan(1); + expect(maxInFlight()).toBeLessThan(balanceRows.length); + }); + + function setup(input: { + checkpoint: number | undefined; + balanceRows: { address: string; denom: string; amount: string }[]; + chainBalances: Record; + chainSupply: { denom: string; amount: string }[]; + }) { + const buildSelect = () => ({ + from: (table: unknown) => { + if (table === IndexerState) { + return { where: () => Promise.resolve(input.checkpoint === undefined ? [] : [{ lastHeight: input.checkpoint }]) }; + } + return { innerJoin: () => Promise.resolve(input.balanceRows) }; + } + }); + + const baseSelect = vi.fn(buildSelect); + const txSelect = vi.fn(buildSelect); + const transaction = vi.fn(async (callback: (tx: unknown) => unknown) => callback({ select: txSelect })); + const dbFake = { select: baseSelect, transaction }; + + let inFlight = 0; + let maxInFlight = 0; + const rpc = mock(); + rpc.abciQuery.mockImplementation(async (path, dataHex) => { + if (path === ALL_BALANCES_PATH) { + inFlight++; + maxInFlight = Math.max(maxInFlight, inFlight); + await Promise.resolve(); + inFlight--; + const address = Object.keys(input.chainBalances).find(candidate => Buffer.from(dataHex, "hex").toString("utf8").includes(candidate)); + return { value: toBase64(QueryAllBalancesResponse.encode({ balances: input.chainBalances[address ?? ""] ?? [], pagination: undefined }).finish()) }; + } + return { value: toBase64(QueryTotalSupplyResponse.encode({ supply: input.chainSupply, pagination: undefined }).finish()) }; + }); + + const service = new ReconcileService(dbFake as unknown as ChainDatabase, rpc, mock()); + return { service, abciQuery: rpc.abciQuery, transaction, baseSelect, txSelect, maxInFlight: () => maxInFlight }; + } +}); diff --git a/apps/chain-indexer/src/reconcile/reconcile.service.ts b/apps/chain-indexer/src/reconcile/reconcile.service.ts new file mode 100644 index 0000000000..ff93a332e9 --- /dev/null +++ b/apps/chain-indexer/src/reconcile/reconcile.service.ts @@ -0,0 +1,167 @@ +import { eq } from "drizzle-orm"; +import { inject, singleton } from "tsyringe"; + +import { AccountBalances, Accounts, IndexerState } from "@src/db/schema"; +import type { CoinAmount } from "@src/pipeline/balance/coin-amount"; +import { SYNC_STREAM } from "@src/pipeline/block-committer.service"; +import type { ChainDatabase, ChainTransaction } from "@src/providers/db.provider"; +import { CHAIN_DB } from "@src/providers/db.provider"; +import { LoggerService } from "@src/providers/logging.provider"; +import { + ALL_BALANCES_PATH, + decodeAllBalances, + decodeTotalSupply, + encodeAllBalancesRequest, + encodeTotalSupplyRequest, + TOTAL_SUPPLY_PATH +} from "@src/reconcile/bank-query"; +import type { CoinDiff } from "@src/reconcile/coin-diff"; +import { diffCoins } from "@src/reconcile/coin-diff"; +import { RpcClientPool } from "@src/rpc/rpc-client-pool.service"; + +const DEFAULT_SAMPLE_SIZE = 100; + +/** Bounds concurrent ABCI round-trips so the sampled accounts load-balance across the RPC pool without flooding any single node. */ +const RECONCILE_CONCURRENCY = 10; + +interface AccountBalance { + address: string; + coins: CoinAmount[]; +} + +/** + * Proves the ledger matches the chain. At the indexer's `sync` checkpoint height it compares each sampled + * account's current balance against the node's bank balance, and the ledger's total per denom against the + * chain's supply. Querying at the checkpoint (not the moving tip) keeps the comparison race-free. + */ +@singleton() +export class ReconcileService { + readonly #db: ChainDatabase; + readonly #rpc: RpcClientPool; + readonly #logger: LoggerService; + + constructor(@inject(CHAIN_DB) db: ChainDatabase, @inject(RpcClientPool) rpc: RpcClientPool, @inject(LoggerService) logger: LoggerService) { + this.#db = db; + this.#rpc = rpc; + this.#logger = logger; + this.#logger.setContext("RECONCILE"); + } + + async reconcile({ sampleSize = DEFAULT_SAMPLE_SIZE }: { sampleSize?: number } = {}): Promise { + if (!Number.isInteger(sampleSize) || sampleSize <= 0) { + this.#logger.error({ event: "RECONCILE_INVALID_SAMPLE_SIZE", sampleSize }); + return false; + } + + const snapshot = await this.#readSnapshot(); + if (snapshot === undefined) { + this.#logger.warn({ event: "RECONCILE_NO_CHECKPOINT" }); + return false; + } + + const { height, balances } = snapshot; + const sampled = this.#sample(balances, sampleSize); + this.#logger.info({ event: "RECONCILE_START", height, accounts: balances.length, sampled: sampled.length }); + + const accountMatches = await mapWithConcurrency(sampled, RECONCILE_CONCURRENCY, async account => { + const chain = decodeAllBalances((await this.#rpc.abciQuery(ALL_BALANCES_PATH, encodeAllBalancesRequest(account.address), height)).value); + const diffs = diffCoins(chain, account.coins); + if (diffs.length === 0) return true; + this.#logger.error({ event: "RECONCILE_ACCOUNT_MISMATCH", address: account.address, diffs: format(diffs) }); + return false; + }); + + let mismatches = accountMatches.filter(matched => !matched).length; + + const chainSupply = decodeTotalSupply((await this.#rpc.abciQuery(TOTAL_SUPPLY_PATH, encodeTotalSupplyRequest(), height)).value); + const supplyDiffs = diffCoins(chainSupply, totals(balances)); + if (supplyDiffs.length > 0) { + mismatches++; + this.#logger.error({ event: "RECONCILE_SUPPLY_MISMATCH", diffs: format(supplyDiffs) }); + } + + const ok = mismatches === 0; + this.#logger[ok ? "info" : "error"]({ event: ok ? "RECONCILE_OK" : "RECONCILE_FAILED", height, mismatches }); + return ok; + } + + /** + * Reads the checkpoint height and the ledger balances from one REPEATABLE READ snapshot. The committer advances + * both in a single transaction, so reading them as two independent SELECTs could straddle a commit — stale height + * against post-commit balances — and flag spurious mismatches on any account touched by that block. + */ + async #readSnapshot(): Promise<{ height: number; balances: AccountBalance[] } | undefined> { + return this.#db.transaction( + async tx => { + const height = await this.#readCheckpointHeight(tx); + if (height === undefined) return undefined; + return { height, balances: await this.#readLedgerBalances(tx) }; + }, + { isolationLevel: "repeatable read", accessMode: "read only" } + ); + } + + async #readCheckpointHeight(tx: ChainTransaction): Promise { + const [row] = await tx.select().from(IndexerState).where(eq(IndexerState.stream, SYNC_STREAM)); + return row?.lastHeight; + } + + async #readLedgerBalances(tx: ChainTransaction): Promise { + const rows = await tx + .select({ address: Accounts.address, denom: AccountBalances.denom, amount: AccountBalances.amount }) + .from(AccountBalances) + .innerJoin(Accounts, eq(AccountBalances.accountId, Accounts.id)); + + const byAddress = new Map(); + for (const row of rows) { + const coins = byAddress.get(row.address) ?? []; + coins.push({ denom: row.denom, amount: BigInt(row.amount) }); + byAddress.set(row.address, coins); + } + + return [...byAddress.entries()].map(([address, coins]) => ({ address, coins })); + } + + /** Samples the highest-balance accounts, which carry the most reconciliation signal, capping RPC round-trips at `sampleSize`. */ + #sample(balances: AccountBalance[], sampleSize: number): AccountBalance[] { + return balances + .map(account => ({ account, total: sumCoins(account.coins) })) + .sort((a, b) => (b.total < a.total ? -1 : 1)) + .slice(0, sampleSize) + .map(entry => entry.account); + } +} + +/** Runs `worker` over `items` with at most `limit` in flight at once, preserving input order in the returned results. */ +async function mapWithConcurrency(items: T[], limit: number, worker: (item: T) => Promise): Promise { + const results = new Array(items.length); + let next = 0; + + async function runWorker(): Promise { + while (next < items.length) { + const index = next++; + results[index] = await worker(items[index]); + } + } + + await Promise.all(Array.from({ length: Math.min(limit, items.length) }, runWorker)); + return results; +} + +function totals(balances: AccountBalance[]): CoinAmount[] { + const byDenom = new Map(); + for (const account of balances) { + for (const coin of account.coins) { + byDenom.set(coin.denom, (byDenom.get(coin.denom) ?? 0n) + coin.amount); + } + } + return [...byDenom.entries()].map(([denom, amount]) => ({ denom, amount })); +} + +function sumCoins(coins: CoinAmount[]): bigint { + return coins.reduce((sum, coin) => sum + coin.amount, 0n); +} + +function format(diffs: CoinDiff[]): { denom: string; expected: string; actual: string }[] { + return diffs.map(diff => ({ denom: diff.denom, expected: diff.expected.toString(), actual: diff.actual.toString() })); +} diff --git a/apps/chain-indexer/src/reconcile/reconcile.ts b/apps/chain-indexer/src/reconcile/reconcile.ts new file mode 100644 index 0000000000..3a2d9a98f3 --- /dev/null +++ b/apps/chain-indexer/src/reconcile/reconcile.ts @@ -0,0 +1,37 @@ +import "@src/providers"; + +import { createOtelLogger } from "@akashnetwork/logging/otel"; +import { container } from "tsyringe"; + +import { envSchema } from "@src/config/env.config"; +import { PgClientService } from "@src/db/pg-client.service"; +import { ReconcileService } from "@src/reconcile/reconcile.service"; + +/** + * One-shot reconciliation entrypoint (`npm run reconcile`): exits 0 when the ledger matches the chain at the + * sync checkpoint height, non-zero on any mismatch or misconfiguration, so it can gate a deploy. + */ +async function main(): Promise { + const logger = createOtelLogger({ context: "RECONCILE_CLI" }); + + const parsed = envSchema.safeParse(process.env); + if (!parsed.success) { + logger.error({ event: "CONFIG_INVALID", issues: parsed.error.issues.map(issue => ({ path: issue.path.join(".") || "(root)", message: issue.message })) }); + process.exitCode = 1; + return; + } + + const sampleSize = parsed.data.RECONCILE_SAMPLE_SIZE; + + try { + const ok = await container.resolve(ReconcileService).reconcile(sampleSize === undefined ? {} : { sampleSize }); + process.exitCode = ok ? 0 : 1; + } catch (error) { + logger.error({ event: "RECONCILE_FATAL", error }); + process.exitCode = 1; + } finally { + await container.resolve(PgClientService).dispose(); + } +} + +void main(); diff --git a/apps/chain-indexer/src/rpc/rpc-client-pool.service.spec.ts b/apps/chain-indexer/src/rpc/rpc-client-pool.service.spec.ts index 6561c5120d..ec68268ff2 100644 --- a/apps/chain-indexer/src/rpc/rpc-client-pool.service.spec.ts +++ b/apps/chain-indexer/src/rpc/rpc-client-pool.service.spec.ts @@ -98,6 +98,42 @@ describe(RpcClientPool.name, () => { expect(fetchMock.mock.calls[0][0]).toBe("http://node-a/genesis_chunked?chunk=2"); }); + it("runs an abci query with a quoted path, hex data and historical height", async () => { + const { pool, fetchMock } = setup(); + fetchMock.mockResolvedValue(jsonResponse({ result: { response: { code: 0, value: "AA==" } } })); + + const response = await pool.abciQuery("/cosmos.bank.v1beta1.Query/TotalSupply", "0a00", 100); + + expect(response.value).toBe("AA=="); + expect(fetchMock.mock.calls[0][0]).toBe( + "http://node-a/abci_query?path=%22%2Fcosmos.bank.v1beta1.Query%2FTotalSupply%22&data=0x0a00&height=100&prove=false" + ); + }); + + it("fails over to the next node when a node answers with a non-zero abci code", async () => { + const { pool, fetchMock } = setup(); + fetchMock.mockResolvedValueOnce(jsonResponse({ result: { response: { code: 26, log: "height not available", value: null } } })); + fetchMock.mockResolvedValueOnce(jsonResponse({ result: { response: { code: 0, value: "AA==" } } })); + + const response = await pool.abciQuery("/cosmos.bank.v1beta1.Query/AllBalances", "00", 999); + + expect(response.value).toBe("AA=="); + expect(fetchMock).toHaveBeenCalledTimes(2); + expect(fetchMock.mock.calls[0][0]).toContain("http://node-a/"); + expect(fetchMock.mock.calls[1][0]).toContain("http://node-b/"); + }); + + it("throws an aggregate error carrying the abci log when every node answers with a non-zero code", async () => { + const { pool, fetchMock } = setup(); + fetchMock.mockResolvedValue(jsonResponse({ result: { response: { code: 26, log: "height not available", value: null } } })); + + const error = await pool.abciQuery("/cosmos.bank.v1beta1.Query/AllBalances", "00", 999).catch(caught => caught); + + expect(error).toBeInstanceOf(AggregateError); + expect(error.errors[0].message).toContain("height not available"); + expect(fetchMock).toHaveBeenCalledTimes(2); + }); + function setup() { const fetchMock = vi.fn(); vi.stubGlobal("fetch", fetchMock); diff --git a/apps/chain-indexer/src/rpc/rpc-client-pool.service.ts b/apps/chain-indexer/src/rpc/rpc-client-pool.service.ts index 3cee43a144..ee387a164d 100644 --- a/apps/chain-indexer/src/rpc/rpc-client-pool.service.ts +++ b/apps/chain-indexer/src/rpc/rpc-client-pool.service.ts @@ -4,7 +4,7 @@ import { inject, singleton } from "tsyringe"; import type { EnvConfig } from "@src/config/env.config"; import { APP_CONFIG } from "@src/providers/app-config.provider"; import { LoggerService } from "@src/providers/logging.provider"; -import type { RpcBlockResult, RpcBlockResultsResult, RpcGenesisChunkResult, RpcStatusResult } from "@src/rpc/rpc-types"; +import type { RpcAbciQueryResult, RpcBlockResult, RpcBlockResultsResult, RpcGenesisChunkResult, RpcStatusResult } from "@src/rpc/rpc-types"; interface RpcNodeState { endpoint: string; @@ -63,13 +63,35 @@ export class RpcClientPool { return await this.#get(`/genesis_chunked?chunk=${chunk}`); } - async #get(path: string): Promise { + /** + * Runs an ABCI query against historical state at `height`. Reconciliation reads bank balances at the + * indexer's checkpoint height (not the moving tip), which requires an unpruned node — sandbox is archival. + */ + async abciQuery(path: string, dataHex: string, height: number): Promise { + const result = await this.#get( + `/abci_query?path=${encodeURIComponent(`"${path}"`)}&data=0x${dataHex}&height=${height}&prove=false`, + result => { + if (result.response.code) { + throw new Error(`abci_query ${path} failed at height ${height}: ${result.response.log ?? `code ${result.response.code}`}`); + } + } + ); + + return result.response; + } + + /** + * A `validate` failure is treated like a transport failure so the failover loop tries the next node: a pruned + * node answering HTTP 200 with a non-zero ABCI code (e.g. "height not available") must fail over to an archival one. + */ + async #get(path: string, validate?: (result: T) => void): Promise { const errors: unknown[] = []; for (const node of this.#candidates()) { node.inFlight++; try { const result = await this.#fetchFromNode(node.endpoint, path); + validate?.(result); node.unhealthyUntil = 0; return result; } catch (error) { diff --git a/apps/chain-indexer/src/rpc/rpc-types.ts b/apps/chain-indexer/src/rpc/rpc-types.ts index 60fbec7b23..f0a8bc051d 100644 --- a/apps/chain-indexer/src/rpc/rpc-types.ts +++ b/apps/chain-indexer/src/rpc/rpc-types.ts @@ -26,17 +26,37 @@ export interface RpcBlockResult { }; } +/** An ABCI event. Attribute keys/values may be base64-encoded depending on the CometBFT version, so callers normalize them. */ +export interface RpcEvent { + type: string; + attributes: { key: string; value: string | null }[]; +} + /** Fields marshaled with proto3 omitempty semantics may be absent when zero (e.g. code 0 on success). */ export interface RpcTxResult { code?: number; log?: string; gas_used?: string; gas_wanted?: string; + events?: RpcEvent[]; } export interface RpcBlockResultsResult { height: string; txs_results: RpcTxResult[] | null; + finalize_block_events?: RpcEvent[]; + begin_block_events?: RpcEvent[]; + end_block_events?: RpcEvent[]; +} + +/** CometBFT `/abci_query` response. `value` is base64-encoded protobuf (or null when the queried key is absent). */ +export interface RpcAbciQueryResult { + response: { + code?: number; + log?: string; + value: string | null; + height?: string; + }; } /** CometBFT `/genesis_chunked` response. `chunk`/`total` are marshaled as strings; `data` is base64-encoded genesis JSON. */ diff --git a/apps/chain-indexer/tsup.config.ts b/apps/chain-indexer/tsup.config.ts index fb175af820..9457387734 100644 --- a/apps/chain-indexer/tsup.config.ts +++ b/apps/chain-indexer/tsup.config.ts @@ -13,6 +13,7 @@ export default defineConfig(async overrideOptions => prependEffectsToEntries: ["reflect-metadata", "@akashnetwork/env-loader"], entry: { server: "./src/server.ts", + reconcile: "./src/reconcile/reconcile.ts", instrumentation: fileURLToPath(import.meta.resolve("@akashnetwork/instrumentation/register")) }, target: tsconfig.compilerOptions.target, diff --git a/package-lock.json b/package-lock.json index 7b01409ede..97a6c32042 100644 --- a/package-lock.json +++ b/package-lock.json @@ -541,6 +541,7 @@ "@akashnetwork/instrumentation": "*", "@akashnetwork/logging": "*", "@akashnetwork/net": "*", + "@cosmjs/amino": "~0.38.0", "@cosmjs/encoding": "~0.38.0", "@cosmjs/proto-signing": "~0.38.0", "@cosmjs/stargate": "~0.38.0", @@ -576,6 +577,34 @@ "vitest-mock-extended": "^4.0.0" } }, + "apps/chain-indexer/node_modules/@cosmjs/amino": { + "version": "0.38.1", + "resolved": "https://registry.npmjs.org/@cosmjs/amino/-/amino-0.38.1.tgz", + "integrity": "sha512-WaThDpq2JwUyKuazq08Xa+FHzQ3jh1HcYnGL4xsyfqFwOlAvnl0EDvSSz9WSwz1oopIxFE9Qtf3OUKOlxBZbYA==", + "license": "Apache-2.0", + "dependencies": { + "@cosmjs/crypto": "^0.38.1", + "@cosmjs/encoding": "^0.38.1", + "@cosmjs/math": "^0.38.1", + "@cosmjs/utils": "^0.38.1" + } + }, + "apps/chain-indexer/node_modules/@cosmjs/crypto": { + "version": "0.38.1", + "resolved": "https://registry.npmjs.org/@cosmjs/crypto/-/crypto-0.38.1.tgz", + "integrity": "sha512-r1KQCjKAdMga2aZ/nkgULRF4fisPZMF6ErucVsMmkASBgDl0k9/vD9K9fHUdGClMv0oOYEfwOT/UTBR7K2OuYA==", + "license": "Apache-2.0", + "dependencies": { + "@cosmjs/encoding": "^0.38.1", + "@cosmjs/math": "^0.38.1", + "@cosmjs/utils": "^0.38.1", + "@noble/ciphers": "^1.3.0", + "@noble/curves": "^1.9.2", + "@noble/hashes": "^1.8.0", + "@scure/bip39": "^1.6.0", + "hash-wasm": "^4.12.0" + } + }, "apps/chain-indexer/node_modules/@cosmjs/encoding": { "version": "0.38.1", "resolved": "https://registry.npmjs.org/@cosmjs/encoding/-/encoding-0.38.1.tgz", @@ -587,6 +616,45 @@ "readonly-date-esm": "^2.0.0" } }, + "apps/chain-indexer/node_modules/@cosmjs/math": { + "version": "0.38.1", + "resolved": "https://registry.npmjs.org/@cosmjs/math/-/math-0.38.1.tgz", + "integrity": "sha512-MBk7p6kPNULi0TusD8O3xoBskFIkRzOtpmnea3sXbTVnguX7epNPVDITXM4tlsg8kAQrEOEIA0g5zAJxzH3Ikw==", + "license": "Apache-2.0" + }, + "apps/chain-indexer/node_modules/@cosmjs/utils": { + "version": "0.38.1", + "resolved": "https://registry.npmjs.org/@cosmjs/utils/-/utils-0.38.1.tgz", + "integrity": "sha512-ccQ5in6IvsQ+o/SstUdQH1jCJ2+MkJPZK7A/EYwMAFcjV8vzOgJ97LVy6AT24nwdi0/iVa2nbAG+fitUEsgLcA==", + "license": "Apache-2.0" + }, + "apps/chain-indexer/node_modules/@noble/ciphers": { + "version": "1.3.0", + "resolved": "https://registry.npmjs.org/@noble/ciphers/-/ciphers-1.3.0.tgz", + "integrity": "sha512-2I0gnIVPtfnMw9ee9h1dJG7tp81+8Ob3OJb3Mv37rx5L40/b0i7djjCVvGOVqc9AEIQyvyu1i6ypKdFw8R8gQw==", + "license": "MIT", + "engines": { + "node": "^14.21.3 || >=16" + }, + "funding": { + "url": "https://paulmillr.com/funding/" + } + }, + "apps/chain-indexer/node_modules/@noble/curves": { + "version": "1.9.7", + "resolved": "https://registry.npmjs.org/@noble/curves/-/curves-1.9.7.tgz", + "integrity": "sha512-gbKGcRUYIjA3/zCCNaWDciTMFI0dCkvou3TL8Zmy5Nc7sJ47a0jtOeZoTaMxkuqRo9cRhjOdZJXegxYE5FN/xw==", + "license": "MIT", + "dependencies": { + "@noble/hashes": "1.8.0" + }, + "engines": { + "node": "^14.21.3 || >=16" + }, + "funding": { + "url": "https://paulmillr.com/funding/" + } + }, "apps/chain-indexer/node_modules/@scure/base": { "version": "2.2.0", "resolved": "https://registry.npmjs.org/@scure/base/-/base-2.2.0.tgz",