diff --git a/docker-compose.yml b/docker-compose.yml index c7d6f393..d964731c 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -172,6 +172,7 @@ services: - LOGGER_MIN_SEVERITY=${LOGGER_MIN_SEVERITY} - METADATA_SERVER_URI=${METADATA_SERVER_URI} - POSTGRES_PORT=${POSTGRES_PORT} + - ASSET_BACKFILL_BATCH_SIZE=${ASSET_BACKFILL_BATCH_SIZE} restart: on-failure secrets: - postgres_db diff --git a/packages/api-cardano-db-hasura/src/HasuraBackgroundClient.ts b/packages/api-cardano-db-hasura/src/HasuraBackgroundClient.ts index 3d855ddf..4fad6a83 100644 --- a/packages/api-cardano-db-hasura/src/HasuraBackgroundClient.ts +++ b/packages/api-cardano-db-hasura/src/HasuraBackgroundClient.ts @@ -16,6 +16,9 @@ import { const epochInformationNotYetAvailable = 'Epoch information not yet available. This is expected during the initial chain-sync.' +const ASSET_BACKFILL_ADVISORY_LOCK_KEY = 4021991017 +const ASSET_BACKFILL_DEFAULT_BATCH_SIZE = 500000 + const withHexPrefix = (value: string) => `\\x${value !== undefined ? value : ''}` @@ -388,41 +391,74 @@ export class HasuraBackgroundClient { } } - public async backfillMissingAssets (dbConfig: DbConfig): Promise { + public async backfillMissingAssets (dbConfig: DbConfig, batchSize: number = ASSET_BACKFILL_DEFAULT_BATCH_SIZE): Promise { const client = new Client(dbConfig) await client.connect() try { - await client.query(` - UPDATE "Asset" a - SET fingerprint = ma.fingerprint - FROM multi_asset ma - WHERE a."assetId" = CAST(CONCAT(ma.policy, RIGHT(CONCAT(E'\\\\', ma.name), -3)) AS BYTEA) - AND a.fingerprint IS NULL - `) - const result = await client.query(` - INSERT INTO "Asset" ("assetId", "assetName", "policyId", "fingerprint", "firstAppearedInSlot") - SELECT - CAST(CONCAT(ma.policy, RIGHT(CONCAT(E'\\\\', ma.name), -3)) AS BYTEA), - ma.name, - ma.policy, - ma.fingerprint, - MIN(b.slot_no) - FROM multi_asset ma - JOIN ma_tx_mint mtm ON mtm.ident = ma.id - JOIN tx ON tx.id = mtm.tx_id - JOIN block b ON b.id = tx.block_id - LEFT JOIN "Asset" a - ON a."assetId" = CAST(CONCAT(ma.policy, RIGHT(CONCAT(E'\\\\', ma.name), -3)) AS BYTEA) - WHERE a."assetId" IS NULL - GROUP BY ma.id, ma.policy, ma.name, ma.fingerprint - ON CONFLICT ("assetId") DO NOTHING - RETURNING encode("assetId", 'hex') AS "assetId" - `) - this.logger.info( - { module: 'HasuraBackgroundClient', inserted: result.rowCount }, - 'Backfilled missing assets from multi_asset' + const lockResult = await client.query<{ locked: boolean }>( + 'SELECT pg_try_advisory_lock($1) AS locked', + [ASSET_BACKFILL_ADVISORY_LOCK_KEY] ) - return result.rows.map((row: { assetId: string }) => row.assetId) + if (!lockResult.rows[0].locked) { + this.logger.warn( + { module: 'HasuraBackgroundClient' }, + 'Asset backfill already in progress on another connection, skipping' + ) + return [] + } + try { + await client.query(` + UPDATE "Asset" a + SET fingerprint = ma.fingerprint + FROM multi_asset ma + WHERE a."assetId" = CAST(CONCAT(ma.policy, RIGHT(CONCAT(E'\\\\', ma.name), -3)) AS BYTEA) + AND a.fingerprint IS NULL + `) + const maxResult = await client.query<{ maxId: string }>( + 'SELECT COALESCE(MAX(id), 0) AS "maxId" FROM multi_asset' + ) + const maxId = Number(maxResult.rows[0].maxId) + const insertedAssetIds: string[] = [] + for (let cur = 0; cur < maxId; cur += batchSize) { + const result = await client.query(` + INSERT INTO "Asset" ("assetId", "assetName", "policyId", "fingerprint", "firstAppearedInSlot") + SELECT + CAST(CONCAT(ma.policy, RIGHT(CONCAT(E'\\\\', ma.name), -3)) AS BYTEA), + ma.name, + ma.policy, + ma.fingerprint, + MIN(b.slot_no) + FROM multi_asset ma + JOIN ma_tx_mint mtm ON mtm.ident = ma.id + JOIN tx ON tx.id = mtm.tx_id + JOIN block b ON b.id = tx.block_id + LEFT JOIN "Asset" a + ON a."assetId" = CAST(CONCAT(ma.policy, RIGHT(CONCAT(E'\\\\', ma.name), -3)) AS BYTEA) + WHERE a."assetId" IS NULL + AND ma.id > $1 + AND ma.id <= $2 + GROUP BY ma.id, ma.policy, ma.name, ma.fingerprint + ON CONFLICT ("assetId") DO NOTHING + RETURNING encode("assetId", 'hex') AS "assetId" + `, [cur, cur + batchSize]) + if (result.rowCount > 0) { + for (const row of result.rows as { assetId: string }[]) { + insertedAssetIds.push(row.assetId) + } + this.logger.info( + { module: 'HasuraBackgroundClient', upToId: cur + batchSize, inserted: result.rowCount, total: insertedAssetIds.length }, + 'Backfilled asset batch from multi_asset' + ) + } + } + this.logger.info( + { module: 'HasuraBackgroundClient', inserted: insertedAssetIds.length }, + 'Backfilled missing assets from multi_asset' + ) + return insertedAssetIds + } finally { + await client.query('SELECT pg_advisory_unlock($1)', [ASSET_BACKFILL_ADVISORY_LOCK_KEY]) + } } finally { await client.end() } diff --git a/packages/api-cardano-db-hasura/src/background.ts b/packages/api-cardano-db-hasura/src/background.ts index 8a250b0e..e558abb0 100644 --- a/packages/api-cardano-db-hasura/src/background.ts +++ b/packages/api-cardano-db-hasura/src/background.ts @@ -29,6 +29,7 @@ export interface BackgroundConfig { metadataUpdateInterval?: { assets: number; }; + assetBackfillBatchSize?: number; } async function getConfig (): Promise { @@ -94,6 +95,7 @@ function filterAndTypecastEnvs (env: any) { const { COMPOSE_PROFILES, ASSET_METADATA_UPDATE_INTERVAL, + ASSET_BACKFILL_BATCH_SIZE, HASURA_CLI_PATH, HASURA_CLI_EXT_PATH, HASURA_URI, @@ -120,6 +122,9 @@ function filterAndTypecastEnvs (env: any) { ? Number(ASSET_METADATA_UPDATE_INTERVAL) : undefined }, + assetBackfillBatchSize: ASSET_BACKFILL_BATCH_SIZE + ? Number(ASSET_BACKFILL_BATCH_SIZE) + : undefined, postgres: { db: POSTGRES_DB, dbFile: POSTGRES_DB_FILE, @@ -201,7 +206,7 @@ function startAssetPolling ( try { await hasuraBackgroundClient.initialize() const lastSeenId = await hasuraBackgroundClient.getMaxMultiAssetId(config.db) - const backfilledAssetIds = await hasuraBackgroundClient.backfillMissingAssets(config.db) + const backfilledAssetIds = await hasuraBackgroundClient.backfillMissingAssets(config.db, config.assetBackfillBatchSize) await worker.initQueue() startAssetPolling(hasuraBackgroundClient, worker, config.db, lastSeenId, logger) await metadataClient.initialize()