fix(sv-publisher): Restore tx_pointer for transactions

pedronauck · pedronauck · commit 41c4cd8c56f9 · 2025-04-28T19:28:31.000-04:00
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
@@ -209,7 +209,7 @@ make run-publisher \
     PORT=4000 \
     TELEMETRY_PORT=9001 \
     NATS_URL=localhost:4222 \
-    FROM_HEIGHT=0
+    FROM_BLOCK=0
 ```
 
 - Use `testnet-dev` when developing features against testnet
@@ -221,7 +221,7 @@ make run-publisher \
     - `PORT`: Service port (default: 4000)
     - `TELEMETRY_PORT`: Metrics port (default: 9001)
     - `NATS_URL`: NATS server URL
-    - `FROM_HEIGHT`: Starting block height
+    - `FROM_BLOCK`: Starting block height
 
 ### Consumer Service
 
diff --git a/Makefile b/Makefile
@@ -227,26 +227,26 @@ run-publisher: PORT="4000"
 run-publisher: TELEMETRY_PORT="9001"
 run-publisher: NATS_URL="localhost:4222"
 run-publisher: ARGS=""
-run-publisher: FROM_HEIGHT="0"
+run-publisher: FROM_BLOCK="0"
 run-publisher: check-network
 	@./scripts/run_publisher.sh \
 		--mode $(MODE) \
 		--network $(NETWORK) \
 		--telemetry-port $(TELEMETRY_PORT) \
-		--from-height $(FROM_HEIGHT) \
+		--from-block $(FROM_BLOCK) \
 		--extra-args $(ARGS)
 
 run-publisher-mainnet-dev:
-	$(MAKE) run-publisher NETWORK=mainnet MODE=dev FROM_HEIGHT=0
+	$(MAKE) run-publisher NETWORK=mainnet MODE=dev FROM_BLOCK=0
 
 run-publisher-mainnet-profiling:
-	$(MAKE) run-publisher NETWORK=mainnet MODE=profiling FROM_HEIGHT=0
+	$(MAKE) run-publisher NETWORK=mainnet MODE=profiling FROM_BLOCK=0
 
 run-publisher-testnet-dev:
-	$(MAKE) run-publisher NETWORK=testnet MODE=dev FROM_HEIGHT=0
+	$(MAKE) run-publisher NETWORK=testnet MODE=dev FROM_BLOCK=0
 
 run-publisher-testnet-profiling:
-	$(MAKE) run-publisher NETWORK=testnet MODE=profiling FROM_HEIGHT=0
+	$(MAKE) run-publisher NETWORK=testnet MODE=profiling FROM_BLOCK=0
 
 # ------------------------------------------------------------
 #  Consumer Run Commands
diff --git a/cluster/charts/fuel-streams/Chart.yaml b/cluster/charts/fuel-streams/Chart.yaml
@@ -2,7 +2,7 @@ apiVersion: v2
 appVersion: "1.0"
 description: A Helm chart for Kubernetes
 name: fuel-streams
-version: 0.11.13
+version: 0.11.14
 dependencies:
   - name: nats
     version: 1.3.2
diff --git a/cluster/charts/fuel-streams/values.yaml b/cluster/charts/fuel-streams/values.yaml
@@ -419,12 +419,6 @@ dune:
     pullPolicy: Always
     tag: latest
 
-  storage:
-    size: 10Mi # Small size since we only store block height state
-    storageClass: "gp3-generic"
-    accessMode: ReadWriteOnce
-    mountPath: /data/dune
-
   cronjob:
     schedule: "0 * * * *" # Run every hour
     successfulJobsHistoryLimit: 3
diff --git a/crates/domains/src/transactions/db_item.rs b/crates/domains/src/transactions/db_item.rs
@@ -9,6 +9,7 @@ use fuel_streams_types::{
     TransactionStatus,
     TransactionType,
     TxId,
+    TxPointer,
 };
 use serde::{Deserialize, Serialize};
 
@@ -115,7 +116,14 @@ impl TryFrom<&RecordPacket> for TransactionDbItem {
         let transaction = Transaction::decode_json(&packet.value)?;
         let tx_pointer = match transaction.tx_pointer {
             Some(tx_pointer) => Some(serde_json::to_vec(&tx_pointer)?),
-            None => None,
+            None => Some(serde_json::to_vec(&TxPointer {
+                block_height: packet.pointer.block_height.to_owned(),
+                tx_index: packet
+                    .pointer
+                    .tx_index
+                    .expect("tx_index should be defined")
+                    as u16,
+            })?),
         };
 
         let subject: Subjects = packet
diff --git a/crates/domains/src/transactions/repository.rs b/crates/domains/src/transactions/repository.rs
@@ -32,7 +32,6 @@ impl Repository for Transaction {
         let mut conn = executor.acquire().await?;
         let mut db_tx = conn.begin().await?;
         let created_at = BlockTimestamp::now();
-
         let record = sqlx::query_as::<_, TransactionDbItem>(
             r#"
             INSERT INTO transactions (
@@ -171,7 +170,6 @@ impl Repository for Transaction {
         .map_err(RepositoryError::Insert)?;
 
         let tx = Transaction::decode_json(&db_item.value)?;
-
         if let Some(storage_slots) = &tx.storage_slots {
             for slot in storage_slots {
                 let slot_item = TransactionStorageSlotDbItem {
diff --git a/scripts/run_publisher.sh b/scripts/run_publisher.sh
@@ -18,15 +18,15 @@ usage() {
     echo "  --telemetry-port : Specify the telemetry port number"
     echo "                  Default: 8080"
     echo "  --extra-args  : Optional additional arguments to append (in quotes)"
-    echo "  --from-height : Specify the starting block height"
+    echo "  --from-block : Specify the starting block height"
     echo "                  Default: 0"
     echo ""
     echo "Examples:"
     echo "  $0                                              # Runs with all defaults"
     echo "  $0 --network mainnet                            # Runs mainnet with default settings"
     echo "  $0 --port 9000                                  # Runs on port 9000"
     echo "  $0 --network mainnet --port 4001 --telemetry-port 8080 --mode dev     # Custom network, port, telemetry-port and mode"
-    echo "  $0 --from-height 1000                           # Start from block height 1000"
+    echo "  $0 --from-block 1000                           # Start from block height 1000"
     exit 1
 }
 
@@ -35,7 +35,7 @@ NETWORK=${NETWORK:-"testnet"}
 MODE=${MODE:-"profiling"}
 PORT=${PORT:-"4004"}
 TELEMETRY_PORT=${TELEMETRY_PORT:-"9002"}
-FROM_HEIGHT=${FROM_HEIGHT:-"0"}
+FROM_BLOCK=${FROM_BLOCK:-"0"}
 
 while [[ "$#" -gt 0 ]]; do
     case $1 in
@@ -59,8 +59,8 @@ while [[ "$#" -gt 0 ]]; do
             ARGS="$2"
             shift 2
             ;;
-        --from-height)
-            FROM_HEIGHT="$2"
+        --from-block)
+            FROM_BLOCK="$2"
             shift 2
             ;;
         --help)
@@ -89,7 +89,7 @@ echo "  → Network: $NETWORK"
 echo "  → Mode: $MODE"
 echo "  → Fuel Block Consumer Port: $PORT"
 echo "  → Telemetry Port: $TELEMETRY_PORT"
-echo "  → From Height: $FROM_HEIGHT"
+echo "  → From Height: $FROM_BLOCK"
 if [ -n "$ARGS" ]; then
     echo "→ Extra Arguments: $ARGS"
 fi
@@ -131,7 +131,7 @@ COMMON_ARGS=(
     # Application specific
     "--nats-url" "nats://localhost:4222"
     "--telemetry-port" "${TELEMETRY_PORT}"
-    "--from-height" "${FROM_HEIGHT}"
+    "--from-block" "${FROM_BLOCK}"
 )
 
 # Execute based on mode
diff --git a/services/publisher/src/cli.rs b/services/publisher/src/cli.rs
@@ -42,11 +42,11 @@ pub struct Cli {
     /// Start from block height
     #[arg(
         long,
-        value_name = "FROM_HEIGHT",
+        value_name = "FROM_BLOCK",
         default_value = "0",
         help = "Start from block height"
     )]
-    pub from_height: u64,
+    pub from_block: u64,
     /// Use metrics
     #[arg(
         long,
diff --git a/services/publisher/src/lib.rs b/services/publisher/src/lib.rs
@@ -3,4 +3,5 @@ pub mod error;
 pub mod gaps;
 pub mod metrics;
 pub mod publish;
+pub mod recover;
 pub mod state;
diff --git a/services/publisher/src/main.rs b/services/publisher/src/main.rs
@@ -15,6 +15,7 @@ use sv_publisher::{
     gaps::{find_next_block_to_save, BlockHeightGap},
     metrics::Metrics,
     publish::publish_block,
+    recover::recover_tx_pointers,
     state::ServerState,
 };
 use tokio::{sync::Semaphore, task::JoinSet};
@@ -52,8 +53,9 @@ async fn main() -> anyhow::Result<()> {
     tokio::select! {
         result = async {
             tokio::join!(
+                recover_tx_pointers(&db),
                 process_historical_blocks(
-                    cli.from_height.into(),
+                    cli.from_block.into(),
                     &message_broker,
                     &fuel_core,
                     &last_block_height,
@@ -72,6 +74,7 @@ async fn main() -> anyhow::Result<()> {
         } => {
             result.0?;
             result.1?;
+            result.2?;
         }
         _ = shutdown.wait_for_shutdown() => {
             tracing::info!("Shutdown signal received, waiting for processing to complete...");
@@ -124,7 +127,7 @@ async fn process_live_blocks(
 }
 
 fn process_historical_blocks(
-    from_height: BlockHeight,
+    from_block: BlockHeight,
     message_broker: &Arc<NatsMessageBroker>,
     fuel_core: &Arc<dyn FuelCoreLike>,
     last_block_height: &Arc<BlockHeight>,
@@ -145,7 +148,7 @@ fn process_historical_blocks(
         }
 
         let Some(processed_gaps) =
-            get_historical_block_range(from_height, &gaps, last_block_height)
+            get_historical_block_range(from_block, &gaps, last_block_height)
         else {
             return Ok(());
         };
@@ -242,7 +245,7 @@ async fn process_blocks_with_join_set(
 }
 
 fn get_historical_block_range(
-    from_height: BlockHeight,
+    from_block: BlockHeight,
     gaps: &[BlockHeightGap],
     last_block_height: BlockHeight,
 ) -> Option<Vec<BlockHeightGap>> {
@@ -252,7 +255,7 @@ fn get_historical_block_range(
 
     let mut processed_gaps = Vec::new();
     for gap in gaps {
-        let start = std::cmp::max(from_height, gap.start);
+        let start = std::cmp::max(from_block, gap.start);
         let end = std::cmp::min(gap.end, last_block_height);
 
         if start <= end {
diff --git a/services/publisher/src/recover.rs b/services/publisher/src/recover.rs
@@ -0,0 +1,109 @@
+use std::sync::Arc;
+
+use anyhow::Result;
+use fuel_streams_domains::infra::Db;
+use fuel_streams_types::TxPointer;
+use serde_json;
+use sqlx::PgPool;
+use tokio::{sync::Semaphore, task::JoinSet};
+
+const QUERY_BATCH_SIZE: i64 = 50;
+
+#[derive(Clone, Debug, sqlx::FromRow)]
+struct TransactionRecord {
+    id: i32,
+    block_height: i64,
+    tx_index: i32,
+}
+
+async fn fetch_transaction_chunk(
+    pool: &PgPool,
+    offset: i64,
+) -> Result<Vec<TransactionRecord>> {
+    println!("Fetching transaction chunk with offset {}", offset);
+    sqlx::query_as::<_, TransactionRecord>(
+        "SELECT id, block_height, tx_index
+         FROM transactions
+         WHERE tx_pointer IS NULL
+         ORDER BY block_height ASC
+         LIMIT $1
+         OFFSET $2",
+    )
+    .bind(QUERY_BATCH_SIZE)
+    .bind(offset)
+    .fetch_all(pool)
+    .await
+    .map_err(Into::into)
+}
+
+async fn update_transaction_chunk(
+    pool: &PgPool,
+    transactions: Vec<TransactionRecord>,
+) -> Result<usize> {
+    if transactions.is_empty() {
+        return Ok(0);
+    }
+
+    let mut tx = pool.begin().await?;
+    for record in transactions.clone() {
+        let tx_pointer = TxPointer {
+            block_height: record.block_height.try_into()?,
+            tx_index: record.tx_index as u16,
+        };
+        let tx_pointer = serde_json::to_vec(&tx_pointer)?;
+        sqlx::query(
+            "UPDATE transactions
+             SET tx_pointer = $1
+             WHERE id = $2",
+        )
+        .bind(&tx_pointer)
+        .bind(record.id)
+        .execute(&mut *tx)
+        .await?;
+    }
+
+    tx.commit().await?;
+    Ok(transactions.len())
+}
+
+pub async fn recover_tx_pointers(db: &Arc<Db>) -> Result<()> {
+    let pool = db.pool_ref();
+    let semaphore = Arc::new(Semaphore::new(30));
+    let mut join_set = JoinSet::new();
+    let mut offset = 0;
+    let mut total_updated = 0;
+
+    loop {
+        let transactions = fetch_transaction_chunk(&pool, offset).await?;
+        if transactions.is_empty() {
+            break;
+        }
+
+        let pool = pool.clone();
+        let permit = semaphore.clone().acquire_owned().await?;
+        let chunk_transactions = transactions;
+        join_set.spawn(async move {
+            let updated =
+                update_transaction_chunk(&pool, chunk_transactions).await?;
+            println!(
+                "Completed chunk at offset {}, updated {} transactions",
+                offset, updated
+            );
+            drop(permit);
+            Ok::<usize, anyhow::Error>(updated)
+        });
+
+        offset += QUERY_BATCH_SIZE;
+        tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
+    }
+
+    while let Some(result) = join_set.join_next().await {
+        total_updated += result??;
+    }
+
+    println!(
+        "Successfully completed transaction updates. Total transactions updated: {}",
+        total_updated
+    );
+    Ok(())
+}
