diff --git a/.changeset/por-indexer-streams-btc-indexer.md b/.changeset/por-indexer-streams-btc-indexer.md new file mode 100644 index 00000000000..500f981fddf --- /dev/null +++ b/.changeset/por-indexer-streams-btc-indexer.md @@ -0,0 +1,5 @@ +--- +'@chainlink/por-indexer-adapter': minor +--- + +Add opt-in streams Bitcoin indexer support for Bitcoin mainnet balances in `por-indexer`. Default behaviour remains the existing `bitcoin-por-indexer` HTTP path via `BITCOIN_*_POR_INDEXER_URL`. Enable BCY/streams UTXO queries by setting `BITCOIN_MAINNET_USE_STREAMS_INDEXER=true` and `BITCOIN_MAINNET_RPC_URL`. Preserves `minConfirmations` on the streams path. diff --git a/.pnp.cjs b/.pnp.cjs index 3a3afab2375..f08a691edcc 100644 --- a/.pnp.cjs +++ b/.pnp.cjs @@ -7001,11 +7001,9 @@ const RAW_RUNTIME_STATE = "packageDependencies": [\ ["@chainlink/external-adapter-framework", "npm:2.16.1"],\ ["@chainlink/six-adapter", "workspace:packages/sources/six"],\ - ["@date-fns/tz", "npm:1.4.1"],\ ["@sinonjs/fake-timers", "npm:15.4.0"],\ ["@types/jest", "npm:29.5.14"],\ ["@types/node", "npm:22.14.1"],\ - ["date-fns", "npm:4.1.0"],\ ["nock", "npm:13.5.6"],\ ["tslib", "npm:2.4.1"],\ ["typescript", "patch:typescript@npm%3A5.8.3#optional!builtin::version=5.8.3&hash=5786d5"]\ diff --git a/packages/sources/por-indexer/README.md b/packages/sources/por-indexer/README.md index ee45356b101..02167ba76db 100644 --- a/packages/sources/por-indexer/README.md +++ b/packages/sources/por-indexer/README.md @@ -8,9 +8,11 @@ This document was generated automatically. Please see [README Generator](../../s ### Dependencies -The `por-indexer` external adapter is strongly dependent on a synced `bitcoin-por-indexer` service and will not be functional without it. +By default, Bitcoin balances are fetched from a NOP-run **`bitcoin-por-indexer`** HTTP service via `BITCOIN_*_POR_INDEXER_URL`. -Follow [this deployment documentation](https://chainlink.notion.site/Bitcoin-Proof-of-Reserves-v2-e670b124e429466bbb31988c6836a9da) to set up a `bitcoin-por-indexer` service. Note that sync time may take weeks in certain cases. +Optionally, a NOP can opt into the **streams Bitcoin indexer** (Electrs-compatible REST API, same as `dlc-cbtc-por`) for **Bitcoin mainnet** by setting `BITCOIN_MAINNET_USE_STREAMS_INDEXER=true` and `BITCOIN_MAINNET_RPC_URL`. Dogecoin continues to use `DOGECOIN_*_POR_INDEXER_URL`. + +See [custom.md](./docs/custom.md) for more details. ### MAX_PAYLOAD_SIZE_LIMIT configuration @@ -18,15 +20,17 @@ The `MAX_PAYLOAD_SIZE_LIMIT` environment variable is used for controlling the ma ## Environment Variables -| Required? | Name | Description | Type | Options | Default | -| :-------: | :------------------------------: | :---------------------------------------------------------------------------------------: | :----: | :-----: | :-----------------------------------------------------------------: | -| | BITCOIN_MAINNET_POR_INDEXER_URL | Indexer URL for Bitcoin mainnet | string | | `` | -| | BITCOIN_TESTNET_POR_INDEXER_URL | Indexer URL for Bitcoin testnet | string | | `` | -| | DOGECOIN_MAINNET_POR_INDEXER_URL | Indexer URL for Dogecoin mainnet | string | | `` | -| | DOGECOIN_TESTNET_POR_INDEXER_URL | Indexer URL for Dogecoin testnet | string | | `` | -| | ZEUS_ZBTC_API_URL | API url for zeus zBTC | string | | `https://hermes.zeusnetwork.xyz/api/v2/chainlink/proof-of-reserves` | -| | BATCH_SIZE | Maximum number of addresses to send in a single request to the balance indexer | number | | `5000` | -| | BACKGROUND_EXECUTE_MS | The amount of time the background execute should sleep before performing the next request | number | | `10000` | +| Required? | Name | Description | Type | Options | Default | +| :-------: | :---------------------------------: | :---------------------------------------------------------------------------------------: | :-----: | :-----: | :-----------------------------------------------------------------: | +| | BITCOIN_MAINNET_POR_INDEXER_URL | bitcoin-por-indexer HTTP service URL for Bitcoin mainnet (default) | string | | `` | +| | BITCOIN_TESTNET_POR_INDEXER_URL | bitcoin-por-indexer HTTP service URL for Bitcoin testnet (default) | string | | `` | +| | BITCOIN_MAINNET_RPC_URL | Streams Bitcoin indexer endpoint for Bitcoin mainnet UTXO queries | string | | `` | +| | BITCOIN_MAINNET_USE_STREAMS_INDEXER | Use streams Bitcoin indexer for mainnet when `true` (requires `BITCOIN_MAINNET_RPC_URL`) | boolean | | `false` | +| | DOGECOIN_MAINNET_POR_INDEXER_URL | Indexer URL for Dogecoin mainnet | string | | `` | +| | DOGECOIN_TESTNET_POR_INDEXER_URL | Indexer URL for Dogecoin testnet | string | | `` | +| | ZEUS_ZBTC_API_URL | API url for zeus zBTC | string | | `https://hermes.zeusnetwork.xyz/api/v2/chainlink/proof-of-reserves` | +| | BATCH_SIZE | Maximum number of addresses to send in a single request to the balance indexer | number | | `5000` | +| | BACKGROUND_EXECUTE_MS | The amount of time the background execute should sleep before performing the next request | number | | `10000` | --- diff --git a/packages/sources/por-indexer/docs/custom.md b/packages/sources/por-indexer/docs/custom.md index 799205dd33d..bd019961ef3 100644 --- a/packages/sources/por-indexer/docs/custom.md +++ b/packages/sources/por-indexer/docs/custom.md @@ -2,9 +2,16 @@ ### Dependencies -The `por-indexer` external adapter is strongly dependent on a synced `bitcoin-por-indexer` service and will not be functional without it. +**Default (unchanged):** Bitcoin balances use a NOP-run **`bitcoin-por-indexer`** HTTP service. Set `BITCOIN_MAINNET_POR_INDEXER_URL` / `BITCOIN_TESTNET_POR_INDEXER_URL`. -Follow [this deployment documentation](https://chainlink.notion.site/Bitcoin-Proof-of-Reserves-v2-e670b124e429466bbb31988c6836a9da) to set up a `bitcoin-por-indexer` service. Note that sync time may take weeks in certain cases. +**Opt-in streams indexer (mainnet only):** To use the shared BCY/streams Bitcoin indexer instead, set: + +- `BITCOIN_MAINNET_USE_STREAMS_INDEXER=true` +- `BITCOIN_MAINNET_RPC_URL` to the Electrs-compatible base URL (same as `BITCOIN_RPC_ENDPOINT` in `dlc-cbtc-por`) + +PoR jobspecs can stay on `indexer: por_indexer`; only the `por-indexer` EA deployment env vars change. + +Dogecoin continues to use `DOGECOIN_*_POR_INDEXER_URL`. ### MAX_PAYLOAD_SIZE_LIMIT configuration diff --git a/packages/sources/por-indexer/src/config/index.ts b/packages/sources/por-indexer/src/config/index.ts index c2604cfc325..04d29d1c13a 100644 --- a/packages/sources/por-indexer/src/config/index.ts +++ b/packages/sources/por-indexer/src/config/index.ts @@ -11,6 +11,17 @@ export const configDefinition = { type: 'string', default: '', }, + BITCOIN_MAINNET_RPC_URL: { + description: 'Streams Bitcoin indexer URL for mainnet UTXO queries (opt-in path)', + type: 'string', + default: '', + }, + BITCOIN_MAINNET_USE_STREAMS_INDEXER: { + description: + 'When true, use BITCOIN_MAINNET_RPC_URL for mainnet Bitcoin instead of BITCOIN_MAINNET_POR_INDEXER_URL', + type: 'boolean', + default: false, + }, DOGECOIN_MAINNET_POR_INDEXER_URL: { description: 'Indexer URL for Dogecoin mainnet', type: 'string', @@ -50,3 +61,15 @@ export const config = new AdapterConfig( }, }, ) + +export const balanceEnvVarForAddress = ( + network: string, + chainId: string, + useStreamsMainnet: boolean, +): string => { + if (network === 'bitcoin' && chainId === 'mainnet' && useStreamsMainnet) { + return 'BITCOIN_MAINNET_RPC_URL' + } + + return `${network}_${chainId}`.toUpperCase() + '_POR_INDEXER_URL' +} diff --git a/packages/sources/por-indexer/src/endpoint/balance.ts b/packages/sources/por-indexer/src/endpoint/balance.ts index 3c2cade84aa..49d73f68199 100644 --- a/packages/sources/por-indexer/src/endpoint/balance.ts +++ b/packages/sources/por-indexer/src/endpoint/balance.ts @@ -1,9 +1,9 @@ -import { InputParameters } from '@chainlink/external-adapter-framework/validation' import { PoRTotalBalanceEndpoint } from '@chainlink/external-adapter-framework/adapter/por' -import { config, configDefinition } from '../config' -import { balanceTransport } from '../transport/balance' import { AdapterRequest } from '@chainlink/external-adapter-framework/util' +import { InputParameters } from '@chainlink/external-adapter-framework/validation' import { AdapterInputError } from '@chainlink/external-adapter-framework/validation/error' +import { balanceEnvVarForAddress, config } from '../config' +import { balanceTransport } from '../transport/balance' export const inputParameters = new InputParameters( { @@ -82,25 +82,26 @@ export const endpoint = new PoRTotalBalanceEndpoint({ }) } - // Check if based on input parameters corresponding env vars are set - const foundEnv = new Map() - const envVarsToCheck = Object.keys(configDefinition) + const checkedNetworkIds = new Set() for (const address of addresses) { - const id = `${address.network}_${address.chainId}` - const env = `${id}_POR_INDEXER_URL`.toUpperCase() + const networkId = `${address.network}_${address.chainId}`.toUpperCase() + if (checkedNetworkIds.has(networkId)) { + continue + } + checkedNetworkIds.add(networkId) + + const env = balanceEnvVarForAddress( + address.network, + address.chainId, + settings.BITCOIN_MAINNET_USE_STREAMS_INDEXER, + ) if (!settings[env as keyof typeof settings]) { throw new AdapterInputError({ statusCode: 400, message: `'${env}' environment variable is required.`, }) } - - foundEnv.set(env, true) - // Stop the loop if all `network_chainId` env vars are found - if (foundEnv.size === envVarsToCheck.length) { - break - } } return diff --git a/packages/sources/por-indexer/src/lib/btc/por.ts b/packages/sources/por-indexer/src/lib/btc/por.ts new file mode 100644 index 00000000000..13ae24befcc --- /dev/null +++ b/packages/sources/por-indexer/src/lib/btc/por.ts @@ -0,0 +1,153 @@ +import { Requester } from '@chainlink/external-adapter-framework/util/requester' + +type UTXO = { + txid: string + vout: number + value: number + status: { confirmed: boolean; block_height?: number } +} + +type MempoolTransaction = { + txid: string + vin: Array<{ + txid: string + vout: number + prevout: { scriptpubkey_address: string; value: number } + }> +} + +type TxStatus = { + confirmed: boolean + block_height?: number +} + +const MAX_STREAMS_ADDRESS_BATCH_SIZE = 10 + +const joinUrl = (base: string, path: string): string => { + const url = new URL(base) + url.pathname = url.pathname.replace(/\/$/, '') + path + return url.toString() +} + +const getConfirmations = (status: TxStatus, blockHeight: number): number => { + if (!status.confirmed || !status.block_height) return 0 + return blockHeight - status.block_height + 1 +} + +const sumConfirmedUtxos = (utxos: UTXO[], blockHeight: number, minConfirmations: number): bigint => + utxos + .filter((utxo) => getConfirmations(utxo.status, blockHeight) >= minConfirmations) + .reduce((sum, utxo) => sum + BigInt(utxo.value), 0n) + +const fetchTxStatus = async ( + requester: Requester, + endpoint: string, + txid: string, +): Promise => { + const txStatusResponse = await requester.request( + joinUrl(endpoint, `/tx/${txid}/status`), + { + url: joinUrl(endpoint, `/tx/${txid}/status`), + }, + ) + return txStatusResponse.response.data +} + +const sumPendingSpendInputs = async ( + requester: Requester, + endpoint: string, + mempoolTxs: MempoolTransaction[], + address: string, + blockHeight: number, + minConfirmations: number, +): Promise => { + const matchingInputs = mempoolTxs.flatMap((tx) => + tx.vin.filter((input) => input.prevout.scriptpubkey_address === address), + ) + + if (matchingInputs.length === 0) { + return 0n + } + + if (minConfirmations <= 0) { + return matchingInputs.reduce((sum, input) => sum + BigInt(input.prevout.value), 0n) + } + + const txStatusById = new Map>() + const eligiblePendingSpends = await Promise.all( + matchingInputs.map(async (input) => { + let txStatusPromise = txStatusById.get(input.txid) + if (!txStatusPromise) { + txStatusPromise = fetchTxStatus(requester, endpoint, input.txid) + txStatusById.set(input.txid, txStatusPromise) + } + + const txStatus = await txStatusPromise + return getConfirmations(txStatus, blockHeight) >= minConfirmations + ? BigInt(input.prevout.value) + : 0n + }), + ) + + let total = 0n + for (const value of eligiblePendingSpends) { + total += value + } + return total +} + +export async function calculateReserves( + requester: Requester, + endpoint: string, + addresses: string[], + minConfirmations: number, + batchSize: number, +): Promise { + const blockHeightResponse = await requester.request( + joinUrl(endpoint, '/blocks/tip/height'), + { + url: joinUrl(endpoint, '/blocks/tip/height'), + }, + ) + const blockHeight = blockHeightResponse.response.data as number + + let totalReserves = 0n + const addressBatchSize = Math.max(1, Math.min(batchSize, MAX_STREAMS_ADDRESS_BATCH_SIZE)) + + for (let i = 0; i < addresses.length; i += addressBatchSize) { + const batch = addresses.slice(i, i + addressBatchSize) + const batchTotals = await Promise.all( + batch.map(async (address) => { + const utxoResponse = await requester.request( + joinUrl(endpoint, `/address/${address}/utxo`), + { url: joinUrl(endpoint, `/address/${address}/utxo`) }, + ) + const mempoolResponse = await requester.request( + joinUrl(endpoint, `/address/${address}/txs/mempool`), + { url: joinUrl(endpoint, `/address/${address}/txs/mempool`) }, + ) + + const confirmed = sumConfirmedUtxos( + utxoResponse.response.data, + blockHeight, + minConfirmations, + ) + const pending = await sumPendingSpendInputs( + requester, + endpoint, + mempoolResponse.response.data, + address, + blockHeight, + minConfirmations, + ) + return confirmed + pending + }), + ) + + for (const reserves of batchTotals) { + totalReserves += reserves + } + } + + return totalReserves +} diff --git a/packages/sources/por-indexer/src/transport/balance.ts b/packages/sources/por-indexer/src/transport/balance.ts index 66603f0a011..f72840af66f 100644 --- a/packages/sources/por-indexer/src/transport/balance.ts +++ b/packages/sources/por-indexer/src/transport/balance.ts @@ -11,6 +11,7 @@ import { Requester } from '@chainlink/external-adapter-framework/util/requester' import { AdapterError } from '@chainlink/external-adapter-framework/validation/error' import Decimal from 'decimal.js' import { BaseEndpointTypes, inputParameters } from '../endpoint/balance' +import { calculateReserves } from '../lib/btc/por' export type TotalBalanceTransportTypes = BaseEndpointTypes @@ -61,55 +62,82 @@ export class TotalBalanceTransport extends SubscriptionTransport> { const { minConfirmations, addresses } = params - const porServiceRequests = new Map() + const balanceRequests = new Map< + string, + { network: string; chainId: string; addresses: string[] } + >() - // Collect addresses into their respective PoR requests - // Mapping from PoR ID to list of addresses for (const { network, chainId, address } of addresses) { const id = `${network}_${chainId}`.toUpperCase() - if (!porServiceRequests.has(id)) { - porServiceRequests.set(id, []) + if (!balanceRequests.has(id)) { + balanceRequests.set(id, { network, chainId, addresses: [] }) } - const existingAddresses = porServiceRequests.get(id)! - existingAddresses.push(address) + balanceRequests.get(id)!.addresses.push(address) } - // Fire off requests to each PoR indexer - const requestResultPromises = [] const providerDataRequestedUnixMs = Date.now() + let totalReserves = new Decimal(0) - for (const [porId, addresses] of porServiceRequests.entries()) { - const indexerEndpointEnvName = `${porId}_POR_INDEXER_URL` as keyof typeof this.config + for (const [networkId, { network, chainId, addresses: networkAddresses }] of balanceRequests) { + if ( + network === 'bitcoin' && + chainId === 'mainnet' && + this.config.BITCOIN_MAINNET_USE_STREAMS_INDEXER + ) { + const rpcUrl = this.config.BITCOIN_MAINNET_RPC_URL + + if (!rpcUrl) { + throw new AdapterError({ + message: + "'BITCOIN_MAINNET_RPC_URL' environment variable is required when BITCOIN_MAINNET_USE_STREAMS_INDEXER is enabled.", + }) + } + + const networkTotal = await calculateReserves( + this.requester, + rpcUrl, + networkAddresses, + minConfirmations, + this.config.BATCH_SIZE, + ) + totalReserves = totalReserves.add(new Decimal(networkTotal.toString())) + continue + } + + const indexerEndpointEnvName = `${networkId}_POR_INDEXER_URL` as keyof typeof this.config const indexerUrl = this.config[indexerEndpointEnvName] as string - const addressBatches = splitArrayIntoChunks(addresses, this.config.BATCH_SIZE) - for (const addressBatch of addressBatches) { - const requestResult = await this._makeRequest(indexerUrl, addressBatch, minConfirmations) - requestResultPromises.push(requestResult) + if (!indexerUrl) { + throw new AdapterError({ + message: `'${indexerEndpointEnvName}' environment variable is required.`, + }) } - } - // Sum up the total reserves from each PoR indexer - const requestResults = await Promise.all(requestResultPromises) - const summedTotalReserves = requestResults - .map((requestResult) => { - const totalReserves = new Decimal(requestResult.response.data.data.totalReserves) - if (!totalReserves.isFinite() || totalReserves.isNaN()) { + const addressBatches = splitArrayIntoChunks(networkAddresses, this.config.BATCH_SIZE) + for (const addressBatch of addressBatches) { + const requestResult = await this._makePorIndexerRequest( + indexerUrl, + addressBatch, + minConfirmations, + ) + const batchTotal = new Decimal(requestResult.response.data.data.totalReserves) + if (!batchTotal.isFinite() || batchTotal.isNaN()) { throw new AdapterError({ - message: `Invalid totalReserves answer: ${totalReserves.toString()}`, + message: `Invalid totalReserves answer: ${batchTotal.toString()}`, }) } - return totalReserves - }) - .reduce((p, c) => p.add(c), new Decimal(0)) - .toString() + totalReserves = totalReserves.add(batchTotal) + } + } + + const result = totalReserves.toString() return { data: { - result: summedTotalReserves, + result, }, statusCode: 200, - result: summedTotalReserves, + result, timestamps: { providerDataRequestedUnixMs, providerDataReceivedUnixMs: Date.now(), @@ -118,7 +146,7 @@ export class TotalBalanceTransport extends SubscriptionTransport { + let spy: jest.SpyInstance + let testAdapter: TestAdapter + let oldEnv: NodeJS.ProcessEnv + + beforeAll(async () => { + oldEnv = JSON.parse(JSON.stringify(process.env)) + process.env['BITCOIN_MAINNET_USE_STREAMS_INDEXER'] = 'true' + process.env['BITCOIN_MAINNET_RPC_URL'] = 'http://localhost:8546' + process.env['BACKGROUND_EXECUTE_MS'] = '0' + process.env['BATCH_SIZE'] = '2' + const mockDate = new Date('2001-01-01T11:11:11.111Z') + spy = jest.spyOn(Date, 'now').mockReturnValue(mockDate.getTime()) + + const adapter = (await import('./../../src')).adapter + testAdapter = await TestAdapter.startWithMockedCache(adapter, { + testAdapter: {} as TestAdapter, + }) + }) + + afterAll(async () => { + setEnvVariables(oldEnv) + await testAdapter.api.close() + nock.cleanAll() + spy.mockRestore() + nock.restore() + }) + + it('should return success', async () => { + mockStreamsResponseSuccess() + const response = await testAdapter.request({ + addresses: [ + { + network: 'bitcoin', + chainId: 'mainnet', + address: '39e7mxbeNmRRnjfy1qkphv1TiMcztZ8VuE', + }, + { + network: 'bitcoin', + chainId: 'mainnet', + address: '35ULMyVnFoYaPaMxwHTRmaGdABpAThM4QR', + }, + ], + minConfirmations: 6, + }) + expect(response.statusCode).toBe(200) + expect(response.json().result).toBe('24242') + }) + + it('should batch addresses', async () => { + mockStreamsSecondBatch() + const response = await testAdapter.request({ + addresses: [ + { + network: 'bitcoin', + chainId: 'mainnet', + address: '39e7mxbeNmRRnjfy1qkphv1TiMcztZ8VuE', + }, + { + network: 'bitcoin', + chainId: 'mainnet', + address: '35ULMyVnFoYaPaMxwHTRmaGdABpAThM4QR', + }, + { + network: 'bitcoin', + chainId: 'mainnet', + address: '3KLdeu9maZAfccm3TeRWEmUMuw2e8SLo4v', + }, + ], + minConfirmations: 6, + }) + expect(response.statusCode).toBe(200) + expect(response.json().result).toBe('24254') + }) + + it('should exclude UTXOs below minConfirmations', async () => { + mockMinConfirmationsExclusion() + const response = await testAdapter.request({ + addresses: [ + { + network: 'bitcoin', + chainId: 'mainnet', + address: '39e7mxbeNmRRnjfy1qkphv1TiMcztZ8VuE', + }, + ], + minConfirmations: 6, + }) + expect(response.statusCode).toBe(200) + expect(response.json().result).toBe('10000') + }) +}) diff --git a/packages/sources/por-indexer/test/integration/adapter.test.ts b/packages/sources/por-indexer/test/integration/adapter.test.ts index e6277c96b89..43cdecdce2d 100644 --- a/packages/sources/por-indexer/test/integration/adapter.test.ts +++ b/packages/sources/por-indexer/test/integration/adapter.test.ts @@ -3,9 +3,19 @@ import { setEnvVariables, } from '@chainlink/external-adapter-framework/util/testing-utils' import * as nock from 'nock' -import { mockResponseSuccess, mockResponseZeusMinerFeeSuccess, mockSecondBatch } from './fixtures' +import { + mockPorIndexerResponseSuccess, + mockPorIndexerSecondBatch, + mockResponseZeusMinerFeeSuccess, +} from './fixtures' + +const snapshotBody = (response: { json: () => Record }) => { + const body = response.json() + delete body.meta + return body +} -describe('execute', () => { +describe('execute (default bitcoin-por-indexer path)', () => { let spy: jest.SpyInstance let testAdapter: TestAdapter let oldEnv: NodeJS.ProcessEnv @@ -14,7 +24,8 @@ describe('execute', () => { oldEnv = JSON.parse(JSON.stringify(process.env)) process.env['BITCOIN_MAINNET_POR_INDEXER_URL'] = process.env['BITCOIN_MAINNET_POR_INDEXER_URL'] ?? 'http://localhost:8545' - process.env['ZEUS_ZBTC_API_URL'] = 'http://localhost:8546' + process.env['BITCOIN_MAINNET_USE_STREAMS_INDEXER'] = 'false' + process.env['ZEUS_ZBTC_API_URL'] = 'http://localhost:8547' process.env['BACKGROUND_EXECUTE_MS'] = '0' process.env['BATCH_SIZE'] = '2' const mockDate = new Date('2001-01-01T11:11:11.111Z') @@ -36,7 +47,8 @@ describe('execute', () => { describe('balance endpoint', () => { it('should return success', async () => { - const data = { + mockPorIndexerResponseSuccess() + const response = await testAdapter.request({ addresses: [ { network: 'bitcoin', @@ -50,15 +62,15 @@ describe('execute', () => { }, ], minConfirmations: 6, - } - mockResponseSuccess() - const response = await testAdapter.request(data) + }) expect(response.statusCode).toBe(200) - expect(response.json()).toMatchSnapshot() + expect(snapshotBody(response)).toMatchSnapshot() }) it('should batch addresses', async () => { - const data = { + mockPorIndexerResponseSuccess() + mockPorIndexerSecondBatch() + const response = await testAdapter.request({ addresses: [ { network: 'bitcoin', @@ -77,15 +89,13 @@ describe('execute', () => { }, ], minConfirmations: 6, - } - mockSecondBatch() - const response = await testAdapter.request(data) + }) expect(response.statusCode).toBe(200) - expect(response.json()).toMatchSnapshot() + expect(snapshotBody(response)).toMatchSnapshot() }) it('should return failure for missing env', async () => { - const data = { + const response = await testAdapter.request({ addresses: [ { network: 'dogecoin', @@ -94,32 +104,29 @@ describe('execute', () => { }, ], minConfirmations: 6, - } - const response = await testAdapter.request(data) + }) expect(response.statusCode).toBe(400) - expect(response.json()).toMatchSnapshot() + expect(snapshotBody(response)).toMatchSnapshot() }) it('should return failure for empty addresses', async () => { - const data = { + const response = await testAdapter.request({ addresses: [], minConfirmations: 6, - } - const response = await testAdapter.request(data) + }) expect(response.statusCode).toBe(400) - expect(response.json()).toMatchSnapshot() + expect(snapshotBody(response)).toMatchSnapshot() }) }) describe('zeusminerfees endpoint', () => { it('should return success', async () => { - const data = { - endpoint: 'zeusMinerFee', - } mockResponseZeusMinerFeeSuccess() - const response = await testAdapter.request(data) + const response = await testAdapter.request({ + endpoint: 'zeusMinerFee', + }) expect(response.statusCode).toBe(200) - expect(response.json()).toMatchSnapshot() + expect(snapshotBody(response)).toMatchSnapshot() }) }) }) diff --git a/packages/sources/por-indexer/test/integration/fixtures.ts b/packages/sources/por-indexer/test/integration/fixtures.ts index 1fd336e3df0..2d1145ff7da 100644 --- a/packages/sources/por-indexer/test/integration/fixtures.ts +++ b/packages/sources/por-indexer/test/integration/fixtures.ts @@ -1,12 +1,22 @@ import nock from 'nock' -export const mockResponseSuccess = (): nock.Scope => - nock('http://localhost:8545', {}) +export const MOCK_POR_INDEXER_URL = 'http://localhost:8545' +export const MOCK_BITCOIN_RPC_URL = 'http://localhost:8546' +export const MOCK_BLOCK_HEIGHT = 1000 + +export const ADDRESSES = { + addr1: '39e7mxbeNmRRnjfy1qkphv1TiMcztZ8VuE', + addr2: '35ULMyVnFoYaPaMxwHTRmaGdABpAThM4QR', + addr3: '3KLdeu9maZAfccm3TeRWEmUMuw2e8SLo4v', +} as const + +export const mockPorIndexerResponseSuccess = (): nock.Scope => + nock(MOCK_POR_INDEXER_URL, {}) .persist() .post('/', { id: '1', data: { - addresses: ['39e7mxbeNmRRnjfy1qkphv1TiMcztZ8VuE', '35ULMyVnFoYaPaMxwHTRmaGdABpAThM4QR'], + addresses: [ADDRESSES.addr1, ADDRESSES.addr2], minConfirmations: 6, }, }) @@ -22,13 +32,13 @@ export const mockResponseSuccess = (): nock.Scope => ]) .persist() -export const mockSecondBatch = (): nock.Scope => - nock('http://localhost:8545', {}) +export const mockPorIndexerSecondBatch = (): nock.Scope => + nock(MOCK_POR_INDEXER_URL, {}) .persist() .post('/', { id: '1', data: { - addresses: ['3KLdeu9maZAfccm3TeRWEmUMuw2e8SLo4v'], + addresses: [ADDRESSES.addr3], minConfirmations: 6, }, }) @@ -44,8 +54,77 @@ export const mockSecondBatch = (): nock.Scope => ]) .persist() +const mockBlockHeight = (baseUrl: string): nock.Scope => + nock(baseUrl).persist().get('/blocks/tip/height').reply(200, String(MOCK_BLOCK_HEIGHT)) + +const mockAddressUtxos = ( + baseUrl: string, + address: string, + utxos: Array<{ + txid: string + vout: number + value: number + block_height: number + }>, +): nock.Scope => + nock(baseUrl) + .persist() + .get(`/address/${address}/utxo`) + .reply( + 200, + utxos.map((utxo) => ({ + txid: utxo.txid, + vout: utxo.vout, + value: utxo.value, + status: { + confirmed: true, + block_height: utxo.block_height, + }, + })), + ) + +const mockEmptyMempool = (baseUrl: string, address: string): nock.Scope => + nock(baseUrl).persist().get(`/address/${address}/txs/mempool`).reply(200, []) + +export const mockStreamsResponseSuccess = (): void => { + mockBlockHeight(MOCK_BITCOIN_RPC_URL) + mockAddressUtxos(MOCK_BITCOIN_RPC_URL, ADDRESSES.addr1, [ + { txid: 'tx1', vout: 0, value: 10000, block_height: 994 }, + ]) + mockAddressUtxos(MOCK_BITCOIN_RPC_URL, ADDRESSES.addr2, [ + { txid: 'tx2', vout: 0, value: 14242, block_height: 995 }, + ]) + mockEmptyMempool(MOCK_BITCOIN_RPC_URL, ADDRESSES.addr1) + mockEmptyMempool(MOCK_BITCOIN_RPC_URL, ADDRESSES.addr2) +} + +export const mockStreamsSecondBatch = (): void => { + mockBlockHeight(MOCK_BITCOIN_RPC_URL) + mockAddressUtxos(MOCK_BITCOIN_RPC_URL, ADDRESSES.addr1, [ + { txid: 'tx1', vout: 0, value: 10000, block_height: 994 }, + ]) + mockAddressUtxos(MOCK_BITCOIN_RPC_URL, ADDRESSES.addr2, [ + { txid: 'tx2', vout: 0, value: 14242, block_height: 995 }, + ]) + mockAddressUtxos(MOCK_BITCOIN_RPC_URL, ADDRESSES.addr3, [ + { txid: 'tx3', vout: 0, value: 12, block_height: 994 }, + ]) + mockEmptyMempool(MOCK_BITCOIN_RPC_URL, ADDRESSES.addr1) + mockEmptyMempool(MOCK_BITCOIN_RPC_URL, ADDRESSES.addr2) + mockEmptyMempool(MOCK_BITCOIN_RPC_URL, ADDRESSES.addr3) +} + +export const mockMinConfirmationsExclusion = (): void => { + mockBlockHeight(MOCK_BITCOIN_RPC_URL) + mockAddressUtxos(MOCK_BITCOIN_RPC_URL, ADDRESSES.addr1, [ + { txid: 'included', vout: 0, value: 10000, block_height: 995 }, + { txid: 'excluded', vout: 1, value: 50000, block_height: 996 }, + ]) + mockEmptyMempool(MOCK_BITCOIN_RPC_URL, ADDRESSES.addr1) +} + export const mockResponseZeusMinerFeeSuccess = () => - nock('http://localhost:8546') + nock('http://localhost:8547') .persist() .get('/') .reply( diff --git a/packages/sources/por-indexer/test/unit/btc-reserves.test.ts b/packages/sources/por-indexer/test/unit/btc-reserves.test.ts new file mode 100644 index 00000000000..d71d143c5e2 --- /dev/null +++ b/packages/sources/por-indexer/test/unit/btc-reserves.test.ts @@ -0,0 +1,181 @@ +import { Requester } from '@chainlink/external-adapter-framework/util/requester' +import { calculateReserves } from '../../src/lib/btc/por' + +const waitFor = async (predicate: () => boolean): Promise => { + for (let i = 0; i < 50; i += 1) { + if (predicate()) { + return + } + await new Promise((resolve) => setTimeout(resolve, 0)) + } + + throw new Error('Timed out waiting for condition') +} + +describe('calculateReserves', () => { + it('includes pending spends when the source output meets minConfirmations', async () => { + const requester = { + request: async (_cacheKey: string, config: { url?: string }) => { + const url = config.url ?? '' + + if (url.endsWith('/blocks/tip/height')) { + return { response: { data: 1000 } } + } + + if (url.endsWith('/address/vault/utxo')) { + return { response: { data: [] } } + } + + if (url.endsWith('/address/vault/txs/mempool')) { + return { + response: { + data: [ + { + txid: 'pending-spend', + vin: [ + { + txid: 'eligible-prevout', + vout: 0, + prevout: { + scriptpubkey_address: 'vault', + value: 10000, + }, + }, + ], + }, + ], + }, + } + } + + if (url.endsWith('/tx/eligible-prevout/status')) { + return { + response: { + data: { + confirmed: true, + block_height: 995, + }, + }, + } + } + + throw new Error(`Unexpected URL: ${url}`) + }, + } as unknown as Requester + + await expect( + calculateReserves(requester, 'http://localhost:8546', ['vault'], 6, 5000), + ).resolves.toBe(10000n) + }) + + it('excludes pending spends when the source output is below minConfirmations', async () => { + const requester = { + request: async (_cacheKey: string, config: { url?: string }) => { + const url = config.url ?? '' + + if (url.endsWith('/blocks/tip/height')) { + return { response: { data: 1000 } } + } + + if (url.endsWith('/address/vault/utxo')) { + return { response: { data: [] } } + } + + if (url.endsWith('/address/vault/txs/mempool')) { + return { + response: { + data: [ + { + txid: 'pending-spend', + vin: [ + { + txid: 'ineligible-prevout', + vout: 0, + prevout: { + scriptpubkey_address: 'vault', + value: 10000, + }, + }, + ], + }, + ], + }, + } + } + + if (url.endsWith('/tx/ineligible-prevout/status')) { + return { + response: { + data: { + confirmed: true, + block_height: 996, + }, + }, + } + } + + throw new Error(`Unexpected URL: ${url}`) + }, + } as unknown as Requester + + await expect( + calculateReserves(requester, 'http://localhost:8546', ['vault'], 6, 5000), + ).resolves.toBe(0n) + }) + + it('caps concurrent streams address lookups at 10', async () => { + let activeUtxoRequests = 0 + let maxConcurrentUtxoRequests = 0 + let pendingUtxoResolvers: Array<() => void> = [] + + const requester = { + request: async (_cacheKey: string, config: { url?: string }) => { + const url = config.url ?? '' + + if (url.endsWith('/blocks/tip/height')) { + return { response: { data: 1000 } } + } + + if (url.endsWith('/utxo')) { + activeUtxoRequests += 1 + maxConcurrentUtxoRequests = Math.max(maxConcurrentUtxoRequests, activeUtxoRequests) + + await new Promise((resolve) => { + pendingUtxoResolvers.push(() => { + activeUtxoRequests -= 1 + resolve() + }) + }) + + return { response: { data: [] } } + } + + if (url.endsWith('/txs/mempool')) { + return { response: { data: [] } } + } + + throw new Error(`Unexpected URL: ${url}`) + }, + } as unknown as Requester + + const reservesPromise = calculateReserves( + requester, + 'http://localhost:8546', + Array.from({ length: 11 }, (_, index) => `address-${index}`), + 0, + 5000, + ) + + await waitFor(() => pendingUtxoResolvers.length === 10) + expect(maxConcurrentUtxoRequests).toBe(10) + + const firstBatchResolvers = pendingUtxoResolvers + pendingUtxoResolvers = [] + firstBatchResolvers.forEach((resolve) => resolve()) + + await waitFor(() => pendingUtxoResolvers.length === 1) + pendingUtxoResolvers.forEach((resolve) => resolve()) + + await expect(reservesPromise).resolves.toBe(0n) + }) +})