fix(hubble): improve reorg handling - use indexer_id

unionlabs · Sep 12, 2024 · 057cf91 · 057cf91
1 parent fc275c2
commit 057cf91
Show file tree

Hide file tree

Showing 7 changed files with 23 additions and 23 deletions.
diff --git a/...4cd08545e1c0f021367ef7172294150a1e8d.json → ...efc27655e8c4f347b5ab79a33f6739b6f31d.json b/...4cd08545e1c0f021367ef7172294150a1e8d.json → ...efc27655e8c4f347b5ab79a33f6739b6f31d.json
diff --git a/...b6cecf54ca5713262f8ad1a793fa9676423f.json → ...d6627f1d79b682ba4c0f5f47732df3bf2b0c.json b/...b6cecf54ca5713262f8ad1a793fa9676423f.json → ...d6627f1d79b682ba4c0f5f47732df3bf2b0c.json
diff --git a/...54af415447cd064a047f701255d88fca2934.json → ...c29178aa7ba7c4bbd704cb9c4080240be4c4.json b/...54af415447cd064a047f701255d88fca2934.json → ...c29178aa7ba7c4bbd704cb9c4080240be4c4.json
diff --git a/hubble/hubble.nix b/hubble/hubble.nix
@@ -55,20 +55,20 @@
         indexers = mkOption {
           type = types.listOf (
             types.submodule {
-              options.indexer_id = mkOption { 
-                type = types.nullOr types.str; 
+              options.indexer_id = mkOption {
+                type = types.nullOr types.str;
                 description = "Id of the indexer which is used by the internal administration of Hubble. Should never change.";
                 example = "amazing-testnet";
                 default = null;
               };
-              options.internal_chain_id = mkOption { 
-                type = types.nullOr types.number; 
+              options.internal_chain_id = mkOption {
+                type = types.nullOr types.number;
                 description = "Hubble internal chain id, used to fetch the current height when migrating to fetchers.";
                 example = "4";
                 default = null;
               };
-              options.new_chain_override = mkOption { 
-                type = types.nullOr types.bool; 
+              options.new_chain_override = mkOption {
+                type = types.nullOr types.bool;
                 description = "Indicator that this is a new chain, so the current height must not be used when migrating to fetchers.";
                 example = "false";
                 default = null;

diff --git a/hubble/src/fetcher/eth/eth.rs b/hubble/src/fetcher/eth/eth.rs
@@ -259,10 +259,10 @@ impl BlockHandle for EthBlock {
         let block_to_insert = self.fetch_block().await?;
 
         if let Some(block_to_insert) = block_to_insert {
-            debug!("{}: exists", block_height);
+            debug!("{}: exists => upsert", block_height);
             insert_batch_logs(tx, vec![block_to_insert.into()], InsertMode::Upsert).await?;
         } else {
-            debug!("{}: empty", block_height);
+            debug!("{}: empty => delete", block_height);
             delete_eth_log(tx, self.eth_client.chain_id.db, block_height).await?;
         }
 

diff --git a/hubble/src/fetcher/fetcher.rs b/hubble/src/fetcher/fetcher.rs
@@ -440,7 +440,7 @@ impl<T: FetcherClient> Indexer<T> {
                 if block_range_to_finalize.end - block_range_to_finalize.start
                     < self.chunk_size as u64
                 {
-                    info!("not much to finalize => sleep a bit");
+                    info!("not much to finalize => retry later");
                     sleep(Duration::from_millis(1000)).await;
                 }
             } else {

diff --git a/hubble/src/postgres.rs b/hubble/src/postgres.rs
@@ -689,7 +689,7 @@ pub async fn get_next_block_to_refresh(
 
 pub async fn get_block_range_to_fix(
     tx: &mut sqlx::Transaction<'_, Postgres>,
-    label: String,
+    indexer_id: IndexerId,
 ) -> sqlx::Result<Option<BlockRange>> {
     let record = sqlx::query!(
         "
@@ -698,11 +698,11 @@ pub async fn get_block_range_to_fix(
         WHERE start_height = (
             SELECT min(start_height)
             FROM hubble.block_fix
-            WHERE label = $1
+            WHERE indexer_id = $1
         )
         GROUP BY start_height
         ",
-        label,
+        indexer_id,
     )
     .fetch_optional(tx.as_mut())
     .await?;
@@ -715,18 +715,18 @@ pub async fn get_block_range_to_fix(
 
 pub async fn update_block_range_to_fix(
     tx: &mut sqlx::Transaction<'_, Postgres>,
-    label: String,
+    indexer_id: IndexerId,
     range: BlockRange,
 ) -> sqlx::Result<()> {
     // update start of ranges
     sqlx::query!(
         "
         UPDATE hubble.block_fix
         SET start_height = $3
-        WHERE label = $1
+        WHERE indexer_id = $1
         AND   start_height = $2
         ",
-        label,
+        indexer_id,
         range.start as i64,
         range.end as i64,
     )
@@ -737,11 +737,11 @@ pub async fn update_block_range_to_fix(
     sqlx::query!(
         "
         DELETE FROM hubble.block_fix
-        WHERE label = $1
+        WHERE indexer_id = $1
         AND   start_height = $2
         AND   end_height <= $2
         ",
-        label,
+        indexer_id,
         range.end as i64,
     )
     .execute(tx.as_mut())