@@ -16,6 +16,9 @@ import {
1616const epochInformationNotYetAvailable =
1717 'Epoch information not yet available. This is expected during the initial chain-sync.'
1818
19+ const ASSET_BACKFILL_ADVISORY_LOCK_KEY = 4021991017
20+ const ASSET_BACKFILL_DEFAULT_BATCH_SIZE = 500000
21+
1922const withHexPrefix = ( value : string ) =>
2023 `\\x${ value !== undefined ? value : '' }
2124
@@ -388,41 +391,74 @@ export class HasuraBackgroundClient {
388391 }
389392 }
390393
391- public async backfillMissingAssets ( dbConfig : DbConfig ) : Promise < string [ ] > {
394+ public async backfillMissingAssets ( dbConfig : DbConfig , batchSize : number = ASSET_BACKFILL_DEFAULT_BATCH_SIZE ) : Promise < string [ ] > {
392395 const client = new Client ( dbConfig )
393396 await client . connect ( )
394397 try {
395- await client . query ( `
396- UPDATE "Asset" a
397- SET fingerprint = ma.fingerprint
398- FROM multi_asset ma
399- WHERE a."assetId" = CAST(CONCAT(ma.policy, RIGHT(CONCAT(E'\\\\', ma.name), -3)) AS BYTEA)
400- AND a.fingerprint IS NULL
401- ` )
402- const result = await client . query ( `
403- INSERT INTO "Asset" ("assetId", "assetName", "policyId", "fingerprint", "firstAppearedInSlot")
404- SELECT
405- CAST(CONCAT(ma.policy, RIGHT(CONCAT(E'\\\\', ma.name), -3)) AS BYTEA),
406- ma.name,
407- ma.policy,
408- ma.fingerprint,
409- MIN(b.slot_no)
410- FROM multi_asset ma
411- JOIN ma_tx_mint mtm ON mtm.ident = ma.id
412- JOIN tx ON tx.id = mtm.tx_id
413- JOIN block b ON b.id = tx.block_id
414- LEFT JOIN "Asset" a
415- ON a."assetId" = CAST(CONCAT(ma.policy, RIGHT(CONCAT(E'\\\\', ma.name), -3)) AS BYTEA)
416- WHERE a."assetId" IS NULL
417- GROUP BY ma.id, ma.policy, ma.name, ma.fingerprint
418- ON CONFLICT ("assetId") DO NOTHING
419- RETURNING encode("assetId", 'hex') AS "assetId"
420- ` )
421- this . logger . info (
422- { module : 'HasuraBackgroundClient' , inserted : result . rowCount } ,
423- 'Backfilled missing assets from multi_asset'
398+ const lockResult = await client . query < { locked : boolean } > (
399+ 'SELECT pg_try_advisory_lock($1) AS locked' ,
400+ [ ASSET_BACKFILL_ADVISORY_LOCK_KEY ]
424401 )
425- return result . rows . map ( ( row : { assetId : string } ) => row . assetId )
402+ if ( ! lockResult . rows [ 0 ] . locked ) {
403+ this . logger . warn (
404+ { module : 'HasuraBackgroundClient' } ,
405+ 'Asset backfill already in progress on another connection, skipping'
406+ )
407+ return [ ]
408+ }
409+ try {
410+ await client . query ( `
411+ UPDATE "Asset" a
412+ SET fingerprint = ma.fingerprint
413+ FROM multi_asset ma
414+ WHERE a."assetId" = CAST(CONCAT(ma.policy, RIGHT(CONCAT(E'\\\\', ma.name), -3)) AS BYTEA)
415+ AND a.fingerprint IS NULL
416+ ` )
417+ const maxResult = await client . query < { maxId : string } > (
418+ 'SELECT COALESCE(MAX(id), 0) AS "maxId" FROM multi_asset'
419+ )
420+ const maxId = Number ( maxResult . rows [ 0 ] . maxId )
421+ const insertedAssetIds : string [ ] = [ ]
422+ for ( let cur = 0 ; cur < maxId ; cur += batchSize ) {
423+ const result = await client . query ( `
424+ INSERT INTO "Asset" ("assetId", "assetName", "policyId", "fingerprint", "firstAppearedInSlot")
425+ SELECT
426+ CAST(CONCAT(ma.policy, RIGHT(CONCAT(E'\\\\', ma.name), -3)) AS BYTEA),
427+ ma.name,
428+ ma.policy,
429+ ma.fingerprint,
430+ MIN(b.slot_no)
431+ FROM multi_asset ma
432+ JOIN ma_tx_mint mtm ON mtm.ident = ma.id
433+ JOIN tx ON tx.id = mtm.tx_id
434+ JOIN block b ON b.id = tx.block_id
435+ LEFT JOIN "Asset" a
436+ ON a."assetId" = CAST(CONCAT(ma.policy, RIGHT(CONCAT(E'\\\\', ma.name), -3)) AS BYTEA)
437+ WHERE a."assetId" IS NULL
438+ AND ma.id > $1
439+ AND ma.id <= $2
440+ GROUP BY ma.id, ma.policy, ma.name, ma.fingerprint
441+ ON CONFLICT ("assetId") DO NOTHING
442+ RETURNING encode("assetId", 'hex') AS "assetId"
443+ ` , [ cur , cur + batchSize ] )
444+ if ( result . rowCount > 0 ) {
445+ for ( const row of result . rows as { assetId : string } [ ] ) {
446+ insertedAssetIds . push ( row . assetId )
447+ }
448+ this . logger . info (
449+ { module : 'HasuraBackgroundClient' , upToId : cur + batchSize , inserted : result . rowCount , total : insertedAssetIds . length } ,
450+ 'Backfilled asset batch from multi_asset'
451+ )
452+ }
453+ }
454+ this . logger . info (
455+ { module : 'HasuraBackgroundClient' , inserted : insertedAssetIds . length } ,
456+ 'Backfilled missing assets from multi_asset'
457+ )
458+ return insertedAssetIds
459+ } finally {
460+ await client . query ( 'SELECT pg_advisory_unlock($1)' , [ ASSET_BACKFILL_ADVISORY_LOCK_KEY ] )
461+ }
426462 } finally {
427463 await client . end ( )
428464 }
0 commit comments