From d0aa2374f5252133931d6ee5df134983531da30d Mon Sep 17 00:00:00 2001 From: Ethan Date: Tue, 10 Nov 2026 07:08:33 +0000 Subject: [PATCH 1/3] fix: read Goat events before the write transaction and write them under BEGIN IMMEDIATE --- crates/client/src/graphs/graph_query.rs | 2 +- .../scheduled_tasks/event_watch/bridge_out.rs | 157 ++++++----- .../scheduled_tasks/event_watch/gateway.rs | 258 +++++++++++++++--- node/src/scheduled_tasks/event_watch/mod.rs | 100 ++++--- 4 files changed, 376 insertions(+), 141 deletions(-) diff --git a/crates/client/src/graphs/graph_query.rs b/crates/client/src/graphs/graph_query.rs index 35f26d234..2c959e9be 100644 --- a/crates/client/src/graphs/graph_query.rs +++ b/crates/client/src/graphs/graph_query.rs @@ -455,7 +455,7 @@ pub struct UserInput { pub amount_sats: String, } -#[derive(Debug, Serialize, Deserialize)] +#[derive(Clone, Debug, Serialize, Deserialize)] pub struct BridgeInRequestEvent { pub id: String, #[serde(rename = "transactionHash")] diff --git a/node/src/scheduled_tasks/event_watch/bridge_out.rs b/node/src/scheduled_tasks/event_watch/bridge_out.rs index f572396af..e38ca5d34 100644 --- a/node/src/scheduled_tasks/event_watch/bridge_out.rs +++ b/node/src/scheduled_tasks/event_watch/bridge_out.rs @@ -2,7 +2,10 @@ use crate::env::get_network; use crate::scheduled_tasks::get_timestamp_from_contract_data; -use crate::utils::evm_swap_utils::{extract_claim_data_from_tx, extract_escrow_data_from_tx}; +use crate::utils::evm_swap_utils::IEscrowManager::EscrowData; +use crate::utils::evm_swap_utils::{ + ClaimData, extract_claim_data_from_tx, extract_escrow_data_from_tx, +}; use crate::utils::{get_bridge_out_global_stats, strip_hex_prefix_owned}; use alloy::primitives::{Address as EvmAddress, U256}; use alloy::sol_types::SolValue; @@ -14,63 +17,107 @@ use client::graphs::graph_query::{ }; use std::ops::AddAssign; use std::str::FromStr; -use std::sync::Arc; use store::localdb::{StorageProcessor, SwapEscrowUpdate}; use store::{SwapEscrow, SwapEscrowStatus, normalize_escrow_hash}; use tracing::{info, warn}; -#[allow(clippy::too_many_arguments)] -pub async fn fetch_and_handle_bridge_out_events<'a>( - goat_client: Arc, +/// The swap events of a block range. The init and claim transactions are +/// traced while fetching, so applying them only writes. +#[derive(Default)] +pub struct SwapRange { + entities: Vec, + inits: Vec<(SwapInitializeEvent, Option)>, + claims: Vec<(SwapClaimEvent, Option)>, + refunds: Vec, +} + +fn escrow_hash_bytes(escrow_hash: &str) -> anyhow::Result<[u8; 32]> { + hex::decode(strip_hex_prefix_owned(escrow_hash))? + .try_into() + .map_err(|v: Vec| anyhow::anyhow!("escrow_hash length is {}, expected 32", v.len())) +} + +/// Query the range and trace its init and claim transactions; no database +/// access. Anyone can create an escrow, so the number of traces is not ours to +/// bound, and none of them may run inside the write transaction. +pub async fn fetch_bridge_out_events( + goat_client: &GOATClient, client: &GraphQueryClient, - storage_processor: &mut StorageProcessor<'a>, contract_addr: &EvmAddress, graph_url: &str, event_entities: &[SwapEventEntity], from_height: i64, to_height: i64, - gateway_peg_btc_address: EvmAddress, -) -> anyhow::Result<()> { +) -> anyhow::Result { let query_res = client.get_events(graph_url, event_entities, from_height, to_height).await?; - for event_entity in event_entities { - let entity = event_entity.clone(); - if let Some(value_vec) = query_res[entity.to_string()].as_array() { - match entity { - SwapEventEntity::Initializes => { - handle_swap_init_events( - storage_processor, - goat_client.clone(), + let mut range = SwapRange { entities: event_entities.to_vec(), ..Default::default() }; + for entity in event_entities { + let Some(value_vec) = query_res[entity.to_string()].as_array() else { + continue; + }; + let values = serde_json::Value::Array(value_vec.clone()); + match entity { + SwapEventEntity::Initializes => { + for event in serde_json::from_value::>(values)? { + let escrow_data = extract_escrow_data_from_tx( + goat_client, + &event.transaction_hash, contract_addr, - &gateway_peg_btc_address, - serde_json::from_value::>( - serde_json::Value::Array(value_vec.clone()), - )?, + &escrow_hash_bytes(&event.escrow_hash)?, ) .await?; + range.inits.push((event, escrow_data)); } - SwapEventEntity::Claims => { - handle_swap_claim_events( - storage_processor, - goat_client.clone(), + } + SwapEventEntity::Claims => { + for event in serde_json::from_value::>(values)? { + let claim_data = extract_claim_data_from_tx( + goat_client, + &event.transaction_hash, contract_addr, - &gateway_peg_btc_address, - serde_json::from_value::>(serde_json::Value::Array( - value_vec.clone(), - ))?, + &escrow_hash_bytes(&event.escrow_hash)?, ) .await?; + range.claims.push((event, claim_data)); } - SwapEventEntity::Refunds => { - handle_swap_refund_events( - storage_processor, - &gateway_peg_btc_address, - serde_json::from_value::>(serde_json::Value::Array( - value_vec.clone(), - ))?, - ) - .await?; + } + SwapEventEntity::Refunds => { + range.refunds = serde_json::from_value(values)?; + } + } + } + Ok(range) +} + +/// Write a fetched range, entity by entity in the configured order: a claim +/// needs the escrow its init created. +pub async fn apply_bridge_out_events<'a>( + storage_processor: &mut StorageProcessor<'a>, + range: SwapRange, + gateway_peg_btc_address: EvmAddress, +) -> anyhow::Result<()> { + let SwapRange { entities, inits, claims, refunds } = range; + let (mut inits, mut claims, mut refunds) = (Some(inits), Some(claims), Some(refunds)); + for entity in entities { + match entity { + SwapEventEntity::Initializes => { + if let Some(inits) = inits.take() { + handle_swap_init_events(storage_processor, &gateway_peg_btc_address, inits) + .await?; + } + } + SwapEventEntity::Claims => { + if let Some(claims) = claims.take() { + handle_swap_claim_events(storage_processor, &gateway_peg_btc_address, claims) + .await?; + } + } + SwapEventEntity::Refunds => { + if let Some(refunds) = refunds.take() { + handle_swap_refund_events(storage_processor, &gateway_peg_btc_address, refunds) + .await?; } - }; + } } } Ok(()) @@ -78,22 +125,11 @@ pub async fn fetch_and_handle_bridge_out_events<'a>( async fn handle_swap_init_events<'a>( storage_processor: &mut StorageProcessor<'a>, - goat_client: Arc, - swap_contract_address: &EvmAddress, gateway_peg_btc_address: &EvmAddress, - init_events: Vec, + init_events: Vec<(SwapInitializeEvent, Option)>, ) -> anyhow::Result<()> { - for event in init_events { - let Some(escrow_data) = extract_escrow_data_from_tx( - &goat_client, - &event.transaction_hash, - swap_contract_address, - &hex::decode(strip_hex_prefix_owned(&event.escrow_hash))?.try_into().map_err( - |v: Vec| anyhow::anyhow!("escrow_hash length is {}, expected 32", v.len()), - )?, - ) - .await? - else { + for (event, escrow_data) in init_events { + let Some(escrow_data) = escrow_data else { warn!("failed to parse escrow_data for event:{event:?}"); continue; }; @@ -148,22 +184,11 @@ async fn handle_swap_init_events<'a>( async fn handle_swap_claim_events<'a>( storage_processor: &mut StorageProcessor<'a>, - goat_client: Arc, - swap_contract_address: &EvmAddress, gateway_peg_btc_address: &EvmAddress, - claim_events: Vec, + claim_events: Vec<(SwapClaimEvent, Option)>, ) -> anyhow::Result<()> { - for event in claim_events { - let Some(claim_data) = extract_claim_data_from_tx( - &goat_client, - &event.transaction_hash, - swap_contract_address, - &hex::decode(strip_hex_prefix_owned(&event.escrow_hash))?.try_into().map_err( - |v: Vec| anyhow::anyhow!("escrow_hash length is {}, expected 32", v.len()), - )?, - ) - .await? - else { + for (event, claim_data) in claim_events { + let Some(claim_data) = claim_data else { warn!("failed to parse claim_data for event:{event:?}"); continue; }; diff --git a/node/src/scheduled_tasks/event_watch/gateway.rs b/node/src/scheduled_tasks/event_watch/gateway.rs index 4f528a027..5e05cb663 100644 --- a/node/src/scheduled_tasks/event_watch/gateway.rs +++ b/node/src/scheduled_tasks/event_watch/gateway.rs @@ -23,29 +23,49 @@ use client::graphs::graph_query::{ UserGraphWithdrawEvent, WithdrawDisprovedEvent, WithdrawHappyEvent, WithdrawPathsEvent, WithdrawUnhappyEvent, }; +use std::collections::HashMap; use std::ops::AddAssign; use std::str::FromStr; -use std::sync::Arc; -use store::localdb::{GraphRuntimeUpdate, NodeQuery, StorageProcessor}; +use store::localdb::{GraphRuntimeUpdate, LocalDB, NodeQuery, StorageProcessor}; use store::{ GoatTxProcessingStatus, GoatTxRecord, GoatTxType, GraphStatus, GraphStatusSource, - GraphStatusTransitionOutcome, InstanceBridgeInStatus, + GraphStatusTransitionOutcome, Instance, InstanceBridgeInStatus, }; use tracing::{info, warn}; use uuid::Uuid; #[allow(clippy::too_many_arguments)] -pub async fn fetch_and_handle_gateway_events<'a>( - actor: Actor, - btc_client: Arc, - goat_client: Arc, +/// The gateway events of a block range, with the instances its BridgeIns need +/// already built (that reads Goat and Bitcoin), so applying it only writes. +#[derive(Default)] +pub struct GatewayRange { + init_withdraw_events: Vec, + cancel_withdraw_events: Vec, + proceed_withdraw_events: Vec, + withdraw_paths_events: Vec, + withdraw_disproved_events: Vec, + bridge_in_request_events: Vec, + committee_response_events: Vec, + bridge_in_events: Vec, + post_graph_data_events: Vec, + /// By instance id: the instance built from its request, or `None` when + /// building failed. + rebuilt_instances: HashMap>, +} + +/// Query and parse the range, and build the instances its BridgeIns need; +/// the database is only read, outside any transaction. +#[allow(clippy::too_many_arguments)] +pub async fn fetch_gateway_events( + btc_client: &BTCClient, + goat_client: &GOATClient, client: &GraphQueryClient, - storage_processor: &mut StorageProcessor<'a>, + local_db: &LocalDB, graph_url: &str, event_entities: &[GatewayEventEntity], from_height: i64, to_height: i64, -) -> anyhow::Result<()> { +) -> anyhow::Result { let query_res = client.get_events(graph_url, event_entities, from_height, to_height).await?; let mut init_withdraw_events: Vec = vec![]; @@ -141,24 +161,128 @@ pub async fn fetch_and_handle_gateway_events<'a>( } else { tracing::debug!("{summary}"); } - handle_user_withdraw_events(storage_processor, init_withdraw_events, cancel_withdraw_events) - .await?; - handle_proceed_withdraw_events(storage_processor, proceed_withdraw_events).await?; - handle_withdraw_paths_events(storage_processor, withdraw_paths_events).await?; - handle_withdraw_disproved_events(storage_processor, withdraw_disproved_events).await?; - handle_bridge_in_request_events(storage_processor, bridge_in_request_events).await?; - handle_committee_response_events(storage_processor, committee_response_events).await?; - handle_bridge_in_events( - storage_processor, - btc_client.clone(), - goat_client.clone(), + let rebuilt_instances = rebuild_missing_instances( + btc_client, + goat_client, + local_db, + &bridge_in_events, + &bridge_in_request_events, + ) + .await?; + Ok(GatewayRange { + init_withdraw_events, + cancel_withdraw_events, + proceed_withdraw_events, + withdraw_paths_events, + withdraw_disproved_events, + bridge_in_request_events, + committee_response_events, bridge_in_events, + post_graph_data_events, + rebuilt_instances, + }) +} + +/// Write a fetched range. +pub async fn apply_gateway_events<'a>( + actor: &Actor, + storage_processor: &mut StorageProcessor<'a>, + range: GatewayRange, +) -> anyhow::Result<()> { + handle_user_withdraw_events( + storage_processor, + range.init_withdraw_events, + range.cancel_withdraw_events, ) .await?; - handle_post_graph_data_events(&actor, storage_processor, post_graph_data_events).await?; + handle_proceed_withdraw_events(storage_processor, range.proceed_withdraw_events).await?; + handle_withdraw_paths_events(storage_processor, range.withdraw_paths_events).await?; + handle_withdraw_disproved_events(storage_processor, range.withdraw_disproved_events).await?; + handle_bridge_in_request_events(storage_processor, range.bridge_in_request_events).await?; + handle_committee_response_events(storage_processor, range.committee_response_events).await?; + handle_bridge_in_events(storage_processor, &range.rebuilt_instances, range.bridge_in_events) + .await?; + handle_post_graph_data_events(actor, storage_processor, range.post_graph_data_events).await?; Ok(()) } +/// The request of a BridgeIn whose instance this node never built: from the +/// same range, or from the record of an earlier one. +async fn missing_instance_request( + storage_processor: &mut StorageProcessor<'_>, + instance_id: Uuid, + range_requests: &[BridgeInRequestEvent], +) -> anyhow::Result> { + if storage_processor.find_instance(&instance_id).await?.is_some() { + return Ok(None); + } + if let Some(request) = range_requests.iter().find(|request| { + Uuid::from_str(&strip_hex_prefix_owned(&request.instance_id)).ok() == Some(instance_id) + }) { + return Ok(Some(request.clone())); + } + let record = storage_processor + .find_graph_goat_tx_record( + &instance_id, + &Uuid::nil(), + &GoatTxType::BridgeInRequest.to_string(), + ) + .await?; + Ok(record + .and_then(|record| record.extra) + .and_then(|extra| serde_json::from_str::(&extra).ok())) +} + +/// Build the instances `handle_bridge_in_events` will insert. The pool +/// connection is released before Goat and Bitcoin are read. +async fn rebuild_missing_instances( + btc_client: &BTCClient, + goat_client: &GOATClient, + local_db: &LocalDB, + bridge_in_events: &[BridgeInEvent], + range_requests: &[BridgeInRequestEvent], +) -> anyhow::Result>> { + let mut requests = Vec::new(); + if !bridge_in_events.is_empty() { + let mut storage_processor = local_db.acquire().await?; + for event in bridge_in_events { + let Ok(instance_id) = Uuid::from_str(&strip_hex_prefix_owned(&event.instance_id)) + else { + continue; + }; + if requests.iter().any(|(id, _)| *id == instance_id) { + continue; + } + if let Some(request) = + missing_instance_request(&mut storage_processor, instance_id, range_requests) + .await? + { + requests.push((instance_id, request)); + } + } + } + let mut rebuilt = HashMap::new(); + for (instance_id, request) in requests { + let instance = + match generate_instance_from_bridge_in_request_event(btc_client, goat_client, &request) + .await + { + Ok(instance) => Some(instance), + Err(error) => { + warn!( + event = "event_watch", + instance_id = %instance_id, + error = %error, + "could not build the instance of a BridgeIn; mint event skipped" + ); + None + } + }; + rebuilt.insert(instance_id, instance); + } + Ok(rebuilt) +} + async fn handle_user_withdraw_events<'a>( storage_processor: &mut StorageProcessor<'a>, init_withdraw_events: Vec, @@ -559,8 +683,7 @@ async fn handle_committee_response_events<'a>( async fn handle_bridge_in_events<'a>( storage_processor: &mut StorageProcessor<'a>, - btc_client: Arc, - goat_client: Arc, + rebuilt_instances: &HashMap>, bridge_in_events: Vec, ) -> anyhow::Result<()> { for event in bridge_in_events { @@ -583,19 +706,14 @@ async fn handle_bridge_in_events<'a>( // it in its final state. warn!("Instance {instance_id} is finished but not find in db. we will create it"); if let Some(extra) = tx_record.extra.as_ref() - && let Ok(bridge_event) = serde_json::from_str::(extra) + && serde_json::from_str::(extra).is_ok() { - match generate_instance_from_bridge_in_request_event( - btc_client.as_ref(), - goat_client.as_ref(), - &bridge_event, - ) - .await - { - Ok(mut instance) => { + match rebuilt_instances.get(&instance_id) { + Some(Some(instance)) => { info!( "Instance {instance_id} is created and set status to RelayerL2Minted" ); + let mut instance = instance.clone(); instance.status = InstanceBridgeInStatus::RelayerL2Minted.to_string(); instance.post_pegin_txhash = Some(post_pegin_txhash.clone()); if !storage_processor.insert_instance_if_absent(&instance).await? { @@ -605,11 +723,11 @@ async fn handle_bridge_in_events<'a>( } row_exists = true; } - Err(error) => warn!( - event = "event_watch", - instance_id = %instance_id, - error = %error, - "could not build the instance of a BridgeIn; mint event skipped" + // Building it failed while fetching; that was logged. + Some(None) => {} + // The request record appeared after the fetch: the range runs again. + None => anyhow::bail!( + "BridgeIn of instance {instance_id} needs an instance the fetch did not build; retrying the range" ), } } @@ -763,4 +881,70 @@ mod tests { ); } } + + fn bridge_in_request(instance_id: Uuid) -> BridgeInRequestEvent { + serde_json::from_value(serde_json::json!({ + "id": "1", + "transactionHash": "0xbb", + "blockNumber": "1", + "blockTimestamp": "1", + "instanceId": format!("0x{}", instance_id.simple()), + "depositorAddress": "0x00", + "peginAmountSats": "1", + "txnFees": ["0", "0", "0"], + "userXonlyPubkey": "", + "userChangeAddress": "", + "userRefundAddress": "", + })) + .unwrap() + } + + fn bridge_in(instance_id: Uuid) -> BridgeInEvent { + serde_json::from_value(serde_json::json!({ + "id": "1", + "transactionHash": "0xcc", + "blockNumber": "2", + "instanceId": format!("0x{}", instance_id.simple()), + "depositorAddress": "0x00", + "peginAmountSats": "1", + "feeAmountSats": "0", + })) + .unwrap() + } + + #[tokio::test] + async fn a_missing_instance_is_built_from_the_request_of_the_range_or_of_its_record() { + let (db, _dir) = db().await; + let (in_range, recorded, unknown) = (Uuid::new_v4(), Uuid::new_v4(), Uuid::new_v4()); + let mut storage = db.acquire().await.unwrap(); + handle_bridge_in_request_events(&mut storage, vec![bridge_in_request(recorded)]) + .await + .unwrap(); + let range_requests = vec![bridge_in_request(in_range)]; + for (instance_id, wanted) in [(in_range, true), (recorded, true), (unknown, false)] { + let request = + missing_instance_request(&mut storage, instance_id, &range_requests).await.unwrap(); + assert_eq!(request.is_some(), wanted, "{instance_id}"); + } + } + + #[tokio::test] + async fn a_bridge_in_whose_instance_the_fetch_did_not_build_retries_the_range() { + let (db, _dir) = db().await; + let instance_id = Uuid::new_v4(); + let mut tx = db.start_immediate_transaction().await.unwrap(); + handle_bridge_in_request_events(&mut tx, vec![bridge_in_request(instance_id)]) + .await + .unwrap(); + // The request record appeared after the fetch: refuse rather than + // skip the mint for good. + let error = handle_bridge_in_events(&mut tx, &HashMap::new(), vec![bridge_in(instance_id)]) + .await + .unwrap_err(); + assert!(error.to_string().contains("retrying the range"), "{error:#}"); + // Building failed while fetching: skipped, as before. + let failed = HashMap::from([(instance_id, None)]); + handle_bridge_in_events(&mut tx, &failed, vec![bridge_in(instance_id)]).await.unwrap(); + assert!(tx.find_instance(&instance_id).await.unwrap().is_none()); + } } diff --git a/node/src/scheduled_tasks/event_watch/mod.rs b/node/src/scheduled_tasks/event_watch/mod.rs index f447f12be..09d4e5d00 100644 --- a/node/src/scheduled_tasks/event_watch/mod.rs +++ b/node/src/scheduled_tasks/event_watch/mod.rs @@ -32,8 +32,8 @@ use tokio::time::sleep; use tokio_util::sync::CancellationToken; use tracing::{info, warn}; -use bridge_out::fetch_and_handle_bridge_out_events; -use gateway::fetch_and_handle_gateway_events; +use bridge_out::{apply_bridge_out_events, fetch_bridge_out_events}; +use gateway::{apply_gateway_events, fetch_gateway_events}; /// How long a history sync waits when the indexer tip is below its cursor. const HISTORY_SYNC_TIP_BEHIND_WAIT: Duration = Duration::from_secs(5); @@ -109,48 +109,70 @@ async fn refresh_event_watch_metrics( Ok(()) } -#[allow(clippy::too_many_arguments)] -pub async fn fetch_and_handle_block_range_events<'a>( - actor: Actor, - btc_client: Arc, - goat_client: Arc, +/// The events of one block range, fetched with every network read they need. +pub enum RangeEvents { + Gateway(gateway::GatewayRange), + Swap(bridge_out::SwapRange), +} + +/// Fetch a block range: the subgraph, Goat and Bitcoin reads, outside any +/// write transaction. +pub async fn fetch_block_range_events( + btc_client: &BTCClient, + goat_client: &GOATClient, client: &GraphQueryClient, - storage_processor: &mut StorageProcessor<'a>, + local_db: &LocalDB, watch_events_config: &WatchEventConfig, from_height: i64, to_height: i64, -) -> anyhow::Result<()> { - match watch_events_config { - WatchEventConfig::Gateway(config) => { - fetch_and_handle_gateway_events( - actor, +) -> anyhow::Result { + Ok(match watch_events_config { + WatchEventConfig::Gateway(config) => RangeEvents::Gateway( + fetch_gateway_events( btc_client, - goat_client.clone(), + goat_client, client, - storage_processor, + local_db, &config.the_graph_url, &config.event_entities, from_height, to_height, ) - .await?; - } - WatchEventConfig::Swap(config) => { - fetch_and_handle_bridge_out_events( - goat_client.clone(), + .await?, + ), + WatchEventConfig::Swap(config) => RangeEvents::Swap( + fetch_bridge_out_events( + goat_client, client, - storage_processor, &config.graph_config.address, &config.graph_config.the_graph_url, &config.graph_config.event_entities, from_height, to_height, - config.peg_btc_address, ) - .await?; + .await?, + ), + }) +} + +/// Write a fetched range; the caller holds the write transaction. +pub async fn apply_block_range_events<'a>( + actor: &Actor, + storage_processor: &mut StorageProcessor<'a>, + watch_events_config: &WatchEventConfig, + events: RangeEvents, +) -> anyhow::Result<()> { + match (watch_events_config, events) { + (_, RangeEvents::Gateway(range)) => { + apply_gateway_events(actor, storage_processor, range).await + } + (WatchEventConfig::Swap(config), RangeEvents::Swap(range)) => { + apply_bridge_out_events(storage_processor, range, config.peg_btc_address).await + } + (WatchEventConfig::Gateway(_), RangeEvents::Swap(_)) => { + anyhow::bail!("swap events fetched for a gateway watch") } } - Ok(()) } #[allow(clippy::too_many_arguments)] @@ -206,19 +228,20 @@ pub async fn fetch_history_events( } let to_height = current_finalized.min(watch_contract.from_height + watch_contract.gap); - let mut tx = local_db.start_transaction().await?; - - fetch_and_handle_block_range_events( - actor.clone(), - btc_client.clone(), - goat_client.clone(), + let events = fetch_block_range_events( + &btc_client, + &goat_client, query_client, - &mut tx, + local_db, &watch_events_config, watch_contract.from_height, to_height, ) .await?; + // Only writes from here: IMMEDIATE takes the write lock up front + // instead of failing with BUSY_SNAPSHOT after reading. + let mut tx = local_db.start_immediate_transaction().await?; + apply_block_range_events(&actor, &mut tx, &watch_events_config, events).await?; info!( "Contract {task_name} finish load history event from: {}, to: {to_height}", watch_contract.from_height @@ -363,18 +386,21 @@ pub async fn monitor_events_item( } let to_height = current_finalized.min(watch_contract.from_height + watch_contract.gap); - let mut tx = local_db.start_transaction().await?; - fetch_and_handle_block_range_events( - actor.clone(), - btc_client, - goat_client, + drop(storage_processor); + let events = fetch_block_range_events( + &btc_client, + &goat_client, &query_client, - &mut tx, + local_db, &watch_events_config, watch_contract.from_height, to_height, ) .await?; + // Only writes from here: IMMEDIATE takes the write lock up front instead + // of failing with BUSY_SNAPSHOT after reading. + let mut tx = local_db.start_immediate_transaction().await?; + apply_block_range_events(&actor, &mut tx, &watch_events_config, events).await?; info!("finish monitor event from: {}, to: {to_height}", watch_contract.from_height); watch_contract.from_height = to_height + 1; watch_contract.status = WatchContractStatus::Synced.to_string(); From 8f9ce1988a64facef9e939ab61d4d9874c6e1b76 Mon Sep 17 00:00:00 2001 From: Ethan Date: Tue, 10 Nov 2026 07:08:33 +0000 Subject: [PATCH 2/3] fix: record inbound messages on a writer task instead of the network loop --- node/src/middleware/publisher.rs | 32 +++- node/src/p2p_admission/mod.rs | 5 + node/src/p2p_msg_handler.rs | 290 +++++++++++++++++++++++++++---- 3 files changed, 295 insertions(+), 32 deletions(-) diff --git a/node/src/middleware/publisher.rs b/node/src/middleware/publisher.rs index faa60b6bd..5de17b9b7 100644 --- a/node/src/middleware/publisher.rs +++ b/node/src/middleware/publisher.rs @@ -6,7 +6,7 @@ use futures::future::BoxFuture; use libp2p::request_response::{OutboundRequestId, ResponseChannel}; use libp2p::{ PeerId, Swarm, - gossipsub::{IdentTopic, MessageId}, + gossipsub::{self, IdentTopic, MessageId}, }; use std::collections::HashMap; use tokio::sync::{mpsc, oneshot}; @@ -65,6 +65,13 @@ pub enum PublishCommand { channel: ResponseChannel, response: GetGraphResponse, }, + /// The forwarding verdict on a gossip message whose inbox write ran off the + /// network loop; only the swarm can report it. + GossipValidation { + id: MessageId, + propagation_source: PeerId, + acceptance: gossipsub::MessageAcceptance, + }, } impl PublishCommand { @@ -88,6 +95,14 @@ impl PublishCommand { // The requester may have given up meanwhile; nothing to do then. let _ = swarm.behaviour_mut().get_graph.send_response(channel, response); } + PublishCommand::GossipValidation { id, propagation_source, acceptance } => { + crate::p2p_msg_handler::report_gossip_validation( + swarm, + &id, + &propagation_source, + acceptance, + ); + } } } } @@ -119,6 +134,21 @@ impl NetworkPublisher { .map_err(|_| anyhow::anyhow!("network publisher stopped")) } + /// Hand the swarm the verdict on a gossip message recorded off the loop. + pub async fn report_validation( + &self, + id: MessageId, + propagation_source: PeerId, + acceptance: gossipsub::MessageAcceptance, + ) -> anyhow::Result<()> { + let send = + self.0.send(PublishCommand::GossipValidation { id, propagation_source, acceptance }); + tokio::time::timeout(PUBLISH_TIMEOUT, send) + .await + .map_err(|_| anyhow::anyhow!("network task did not take the gossip verdict in time"))? + .map_err(|_| anyhow::anyhow!("network publisher stopped")) + } + /// Ask `peer` for a graph definition over the direct protocol. pub async fn request_graph( &self, diff --git a/node/src/p2p_admission/mod.rs b/node/src/p2p_admission/mod.rs index 50a84c8c4..4daca75ff 100644 --- a/node/src/p2p_admission/mod.rs +++ b/node/src/p2p_admission/mod.rs @@ -83,6 +83,9 @@ pub enum DropReason { /// room, by rows or by bytes. Distinct from `ImmediateRateLimited`: that one /// is a sender exceeding its rate, this one is the node falling behind. ControlQueueFull, + /// The bounded queue in front of the inbox writer had no room, by rows or + /// by bytes: the database is behind. The sender's outbox sends again. + InboxQueueFull, SenderQuota, GlobalQuota, /// Not a message drop: a `NodeInfo` from a sender that is not registered was @@ -114,6 +117,7 @@ impl DropReason { Self::UnexpectedBinaryKind => "unexpected_binary_kind", Self::ImmediateRateLimited => "immediate_rate_limited", Self::ControlQueueFull => "control_queue_full", + Self::InboxQueueFull => "inbox_queue_full", Self::SenderQuota => "sender_quota", Self::GlobalQuota => "global_quota", Self::NodeTableFull => "node_table_full", @@ -137,6 +141,7 @@ impl DropReason { | Self::ForwardRate | Self::ImmediateRateLimited | Self::ControlQueueFull + | Self::InboxQueueFull | Self::SenderQuota | Self::GlobalQuota ) diff --git a/node/src/p2p_msg_handler.rs b/node/src/p2p_msg_handler.rs index c5cbcdc54..da4619e27 100644 --- a/node/src/p2p_msg_handler.rs +++ b/node/src/p2p_msg_handler.rs @@ -43,9 +43,44 @@ pub struct ImmediateMessage { pub struct WorkerControl { immediate: mpsc::Sender, bytes: Arc, + inbox: InboxQueue, joins: tokio::sync::Mutex>>, } +/// A durable message to record, with what its gossip verdict needs. +pub struct InboxRecord { + event: engine::NewEvent, + message_type: &'static str, + from_peer_id: PeerId, + propagation_source: PeerId, + id: MessageId, + content_size: usize, + _bytes: OwnedSemaphorePermit, +} + +/// Database writes of the network handler, run by the inbox writer so that a +/// long write lock never stalls the network loop. +pub enum InboxJob { + Record(Box), + TouchPeer(PeerId), +} + +/// The bounded queue in front of the inbox writer, by rows and by bytes. +pub struct InboxQueue { + jobs: mpsc::Sender, + bytes: Arc, +} + +const INBOX_QUEUE_ROWS: usize = 64; +const INBOX_QUEUE_BYTES: usize = 64 * 1024 * 1024; + +impl InboxQueue { + fn new() -> (Self, mpsc::Receiver) { + let (jobs, rx) = mpsc::channel(INBOX_QUEUE_ROWS); + (Self { jobs, bytes: Arc::new(Semaphore::new(INBOX_QUEUE_BYTES)) }, rx) + } +} + fn spawn_isolated_worker( name: &'static str, shutdown: CancellationToken, @@ -91,6 +126,42 @@ impl WorkerControl { let Ok(permit) = self.bytes.clone().try_acquire_many_owned(bytes) else { return false }; self.immediate.try_send(ImmediateMessage { source, message, _bytes: permit }).is_ok() } + + /// Queue a durable message for the inbox writer; `false` when the queue + /// has no room. + #[allow(clippy::too_many_arguments)] + fn record( + &self, + event: engine::NewEvent, + message_type: &'static str, + from_peer_id: PeerId, + propagation_source: PeerId, + id: MessageId, + content_size: usize, + ) -> bool { + let Ok(bytes) = u32::try_from(content_size.max(1)) else { return false }; + let Ok(permit) = self.inbox.bytes.clone().try_acquire_many_owned(bytes) else { + return false; + }; + let record = InboxRecord { + event, + message_type, + from_peer_id, + propagation_source, + id, + content_size, + _bytes: permit, + }; + self.inbox.jobs.try_send(InboxJob::Record(Box::new(record))).is_ok() + } + + /// Refresh a peer's liveness at most once per cooldown; skipped when the + /// writer is behind. + fn touch_peer(&self, peer: PeerId, now: Instant) { + if crate::p2p_admission::peer_timestamp_gate().allow(&peer.to_string(), now) { + let _ = self.inbox.jobs.try_send(InboxJob::TouchPeer(peer)); + } + } } impl BitvmNodeProcessor { @@ -121,11 +192,15 @@ impl BitvmNodeProcessor { control_processor.run_immediate_worker(control_publisher, rx).await; }, ); + let (inbox, inbox_jobs) = InboxQueue::new(); + let inbox_join = + tokio::spawn(processor.clone().run_inbox_writer(publisher.clone(), inbox_jobs)); let maintenance_join = tokio::spawn(processor.run_admission_maintenance(publisher)); WorkerControl { immediate: tx, bytes: Arc::new(Semaphore::new(16 * 1024 * 1024)), - joins: tokio::sync::Mutex::new(vec![control_join, maintenance_join]), + inbox, + joins: tokio::sync::Mutex::new(vec![control_join, inbox_join, maintenance_join]), } } } @@ -151,6 +226,61 @@ impl BitvmNodeProcessor { } } + /// Record durable messages and refresh peer liveness, then hand the swarm + /// each message's forwarding verdict: Accept once recorded, Ignore when + /// the write failed (the sender's outbox sends again). + async fn run_inbox_writer( + self, + publisher: NetworkPublisher, + mut jobs: mpsc::Receiver, + ) { + loop { + let job = tokio::select! { + biased; + _ = self.shutdown_token.cancelled() => return, + job = jobs.recv() => match job { + Some(job) => job, + None => return, + }, + }; + match job { + InboxJob::Record(record) => { + let result = record_engine_message( + &self.local_db, + record.event, + record.message_type, + record.from_peer_id, + record.content_size, + ) + .await; + let acceptance = match &result { + Ok(()) => gossipsub::MessageAcceptance::Accept, + Err(error) => { + tracing::warn!(event = "engine_inbox", outcome = "record_failed", message_type = record.message_type, from_peer_id = %record.from_peer_id, error = %format!("{error:#}"), "could not record a durable message"); + gossipsub::MessageAcceptance::Ignore + } + }; + if let Err(error) = publisher + .report_validation(record.id, record.propagation_source, acceptance) + .await + { + tracing::debug!(event = "p2p_message", outcome = "validation_report_failed", error = %error, "could not hand the gossip verdict to the swarm"); + } + if result.is_ok() { + self.metrics_state.record_p2p_receive(true); + refresh_peer_timestamp_throttled( + &self.local_db, + record.from_peer_id, + Instant::now(), + ) + .await; + } + } + InboxJob::TouchPeer(peer) => update_peer_timestamp(&self.local_db, peer).await, + } + } + } + /// Refresh registrations, report drops and clock skew, purge stale node /// rows and flush deferred NodeInfo answers. This task is the sole caller /// of admission maintenance. @@ -315,7 +445,7 @@ impl P2pMessageHandler for BitvmNodeProcessor { const IMMEDIATE_DISPATCH_TIMEOUT: Duration = Duration::from_secs(30); /// Report exactly one forwarding verdict for each received gossipsub message. -fn report_gossip_validation( +pub(crate) fn report_gossip_validation( swarm: &mut Swarm, id: &MessageId, propagation_source: &PeerId, @@ -393,48 +523,61 @@ pub async fn handle_inbound_p2p_message( InboundVerdict::Forward => { report_gossip_validation(swarm, &id, &propagation_source, MessageAcceptance::Accept); metrics_state.record_p2p_receive(true); - refresh_peer_timestamp_throttled(local_db, from_peer_id, now).await; + worker.touch_peer(from_peer_id, now); Ok(()) } // Record durable messages before allowing forwarding: graph-phase // messages for the Graph aggregate, PeginConfirm round messages for // the Instance aggregate. The engine deduplicates by content id and - // ACKs heavy messages once they are durably applied. + // ACKs heavy messages once they are durably applied. The write and the + // verdict run on the inbox writer, off the network loop. InboundVerdict::Enqueue { message: decoded, class } => { let author = from_peer_id.to_string(); let event = inbound_event(&actor, &author, &crate::env::get_peer_id(), &decoded, message); - let recorded = match event { - Ok(Some(event)) => { - record_engine_message( - local_db, - event.with_admission_class(class.to_string()), - &decoded, - from_peer_id, - message.len(), - ) - .await - } + let event = match event { + Ok(Some(event)) => event.with_admission_class(class.to_string()), // Kept by `handled_by` but routed by no aggregate: a table // mismatch, not a bad message. - Ok(None) => Err(anyhow::anyhow!( - "{} is handled by {actor} but no aggregate routes it", - decoded.content.event_type() - )), - Err(error) => Err(error), + Ok(None) => { + report_gossip_validation( + swarm, + &id, + &propagation_source, + MessageAcceptance::Ignore, + ); + return Err(anyhow::anyhow!( + "{} is handled by {actor} but no aggregate routes it", + decoded.content.event_type() + )); + } + Err(error) => { + report_gossip_validation( + swarm, + &id, + &propagation_source, + MessageAcceptance::Ignore, + ); + return Err(error); + } }; - if let Err(error) = recorded { + let queued = worker.record( + event, + decoded.content.event_type(), + from_peer_id, + propagation_source, + id.clone(), + message.len(), + ); + if !queued { report_gossip_validation( swarm, &id, &propagation_source, MessageAcceptance::Ignore, ); - return Err(error); + p2p_admission::record_drop(p2p_admission::DropReason::InboxQueueFull); } - report_gossip_validation(swarm, &id, &propagation_source, MessageAcceptance::Accept); - metrics_state.record_p2p_receive(true); - refresh_peer_timestamp_throttled(local_db, from_peer_id, now).await; Ok(()) } InboundVerdict::Immediate(decoded) => { @@ -450,7 +593,7 @@ pub async fn handle_inbound_p2p_message( } report_gossip_validation(swarm, &id, &propagation_source, MessageAcceptance::Accept); metrics_state.record_p2p_receive(true); - refresh_peer_timestamp_throttled(local_db, from_peer_id, now).await; + worker.touch_peer(from_peer_id, now); Ok(()) } } @@ -531,10 +674,13 @@ fn strike_direct_peer( /// Refresh peer liveness at most once per cooldown. async fn refresh_peer_timestamp_throttled(local_db: &LocalDB, from_peer_id: PeerId, now: Instant) { - let peer = from_peer_id.to_string(); - if !crate::p2p_admission::peer_timestamp_gate().allow(&peer, now) { - return; + if crate::p2p_admission::peer_timestamp_gate().allow(&from_peer_id.to_string(), now) { + update_peer_timestamp(local_db, from_peer_id).await; } +} + +async fn update_peer_timestamp(local_db: &LocalDB, from_peer_id: PeerId) { + let peer = from_peer_id.to_string(); if let Err(error) = update_node_timestamp(local_db, &peer).await { tracing::debug!( event = "p2p_message", @@ -552,7 +698,7 @@ async fn refresh_peer_timestamp_throttled(local_db: &LocalDB, from_peer_id: Peer async fn record_engine_message( local_db: &LocalDB, event: engine::NewEvent, - decoded: &GOATMessage, + message_type: &str, from_peer_id: PeerId, content_size: usize, ) -> Result<()> { @@ -565,7 +711,7 @@ async fn record_engine_message( tracing::info!( event = "engine_inbox", outcome = ?outcome, - message_type = decoded.content.event_type(), + message_type, from_peer_id = %from_peer_id, content_size, "received durable message" @@ -603,12 +749,94 @@ mod tests { tokio::time::timeout(std::time::Duration::from_secs(1), worker).await.unwrap().unwrap(); } + #[tokio::test] + async fn the_inbox_writer_records_off_the_network_loop_and_hands_back_the_verdict() { + use crate::middleware::publisher::PublishCommand; + use std::sync::{Arc, Mutex}; + let dir = tempfile::tempdir().unwrap(); + let db = store::localdb::LocalDB::new_with_busy_timeout( + &format!("sqlite:{}", dir.path().join("inbox.db").display()), + false, + store::DEFAULT_BUSY_TIMEOUT, + ) + .await; + db.migrate().await; + let (goat, _) = client::goat_chain::GOATClient::new_mock_client(); + let processor = super::BitvmNodeProcessor { + local_db: db.clone(), + goat_client: Arc::new(goat), + metrics_state: crate::metrics_service::MetricsState::new(Arc::new(Mutex::new( + prometheus_client::registry::Registry::default(), + ))), + shutdown_token: tokio_util::sync::CancellationToken::new(), + worker: Default::default(), + }; + let (publisher, mut commands) = crate::middleware::publisher::channel(); + let (inbox, jobs) = super::InboxQueue::new(); + let writer = tokio::spawn(processor.clone().run_inbox_writer(publisher, jobs)); + let control = super::WorkerControl { + immediate: tokio::sync::mpsc::channel(1).0, + bytes: std::sync::Arc::new(tokio::sync::Semaphore::new(1)), + inbox, + joins: Default::default(), + }; + + let event = crate::aggregates::graph::graph_posted_event(uuid::Uuid::new_v4()).unwrap(); + let event_id = event.event_id.clone(); + let (id, source) = (libp2p::gossipsub::MessageId::new(b"m"), libp2p::PeerId::random()); + // Queuing returns at once; the write and the verdict follow on the writer. + assert!(control.record(event, "GraphPosted", source, source, id.clone(), 10)); + match commands.recv().await.unwrap() { + PublishCommand::GossipValidation { id: reported, propagation_source, acceptance } => { + assert_eq!(reported, id); + assert_eq!(propagation_source, source); + assert!(matches!(acceptance, libp2p::gossipsub::MessageAcceptance::Accept)); + } + _ => panic!("expected a gossip verdict"), + } + let mut storage = db.acquire().await.unwrap(); + assert!(storage.find_engine_inbox_event(&event_id).await.unwrap().is_some()); + processor.shutdown_token.cancel(); + writer.await.unwrap(); + } + + #[test] + fn a_full_inbox_queue_refuses_instead_of_waiting() { + let (jobs, _rx) = tokio::sync::mpsc::channel(1); + let control = super::WorkerControl { + immediate: tokio::sync::mpsc::channel(1).0, + bytes: std::sync::Arc::new(tokio::sync::Semaphore::new(1)), + inbox: super::InboxQueue { + jobs, + bytes: std::sync::Arc::new(tokio::sync::Semaphore::new(8)), + }, + joins: Default::default(), + }; + let event = || crate::aggregates::graph::graph_posted_event(uuid::Uuid::new_v4()).unwrap(); + let (id, source) = (libp2p::gossipsub::MessageId::new(b"m"), libp2p::PeerId::random()); + assert!( + !control.record(event(), "GraphPosted", source, source, id.clone(), 9), + "over the byte budget" + ); + assert!(control.record(event(), "GraphPosted", source, source, id.clone(), 4)); + assert!( + !control.record(event(), "GraphPosted", source, source, id.clone(), 1), + "no row left" + ); + assert_eq!( + control.inbox.bytes.available_permits(), + 4, + "a refused record releases its bytes" + ); + } + #[tokio::test] async fn ephemeral_queue_is_bounded_by_rows_and_bytes() { let (tx, mut rx) = tokio::sync::mpsc::channel(1); let control = super::WorkerControl { immediate: tx, bytes: std::sync::Arc::new(tokio::sync::Semaphore::new(4)), + inbox: super::InboxQueue::new().0, joins: Default::default(), }; let source = libp2p::PeerId::random(); From 6313318ead97f1009accd7c24fce5e9ce78caa1b Mon Sep 17 00:00:00 2001 From: Ethan Date: Tue, 10 Nov 2026 07:08:33 +0000 Subject: [PATCH 3/3] fix: set engine row gauges of emptied states to zero --- crates/engine/src/metrics.rs | 48 +++++++++++++++++++++++++++++++++--- crates/engine/src/runtime.rs | 10 ++++---- 2 files changed, 49 insertions(+), 9 deletions(-) diff --git a/crates/engine/src/metrics.rs b/crates/engine/src/metrics.rs index 68e701778..f83287007 100644 --- a/crates/engine/src/metrics.rs +++ b/crates/engine/src/metrics.rs @@ -7,6 +7,8 @@ use prometheus_client::metrics::family::Family; use prometheus_client::metrics::gauge::Gauge; use prometheus_client::metrics::histogram::{Histogram, exponential_buckets}; use prometheus_client::registry::Registry; +use std::collections::HashSet; +use std::sync::{Arc, Mutex}; use std::time::Duration; #[derive(Clone, Debug, Hash, PartialEq, Eq, EncodeLabelSet)] @@ -38,6 +40,9 @@ pub struct EngineMetrics { outbox_publish_total: Family, job_runs_total: Family, rows: Family, + /// Every table/state the row gauge has reported, so a state that empties + /// is set back to zero. + known_rows: Arc>>, acks_dropped_total: Counter, poisoned_total: Counter, } @@ -58,6 +63,7 @@ impl EngineMetrics { outbox_publish_total: Family::default(), job_runs_total: Family::default(), rows: Family::default(), + known_rows: Arc::default(), acks_dropped_total: Counter::default(), poisoned_total: Counter::default(), } @@ -132,10 +138,21 @@ impl EngineMetrics { .inc(); } - pub fn set_rows(&self, table: &str, state: &str, count: i64) { - self.rows - .get_or_create(&TableStateLabels { table: table.to_string(), state: state.to_string() }) - .set(count); + /// Publish one refresh of the row counts. The count query only returns + /// states that have rows: a state it no longer returns goes to zero + /// instead of keeping its last value. + pub fn set_row_counts<'a>(&self, counts: impl IntoIterator) { + let mut known = self.known_rows.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + let mut seen = HashSet::new(); + for (table, state, count) in counts { + let labels = TableStateLabels { table: table.to_string(), state: state.to_string() }; + self.rows.get_or_create(&labels).set(count); + known.insert(labels.clone()); + seen.insert(labels); + } + for labels in known.iter().filter(|labels| !seen.contains(*labels)) { + self.rows.get_or_create(labels).set(0); + } } pub fn record_ack_dropped(&self) { @@ -146,3 +163,26 @@ impl EngineMetrics { self.poisoned_total.inc(); } } + +#[cfg(test)] +mod tests { + use super::*; + + fn rows(metrics: &EngineMetrics, table: &str, state: &str) -> i64 { + metrics + .rows + .get_or_create(&TableStateLabels { table: table.to_string(), state: state.to_string() }) + .get() + } + + #[test] + fn a_state_that_empties_goes_back_to_zero() { + let metrics = EngineMetrics::new(); + metrics.set_row_counts([("job", "pending", 3), ("job", "running", 1)]); + metrics.set_row_counts([("job", "pending", 2)]); + assert_eq!(rows(&metrics, "job", "pending"), 2); + assert_eq!(rows(&metrics, "job", "running"), 0); + metrics.set_row_counts([]); + assert_eq!(rows(&metrics, "job", "pending"), 0); + } +} diff --git a/crates/engine/src/runtime.rs b/crates/engine/src/runtime.rs index 9fd3642db..139e9ab4f 100644 --- a/crates/engine/src/runtime.rs +++ b/crates/engine/src/runtime.rs @@ -436,11 +436,11 @@ async fn run_recheck(inner: Arc, cancel: CancellationToken) { } } match storage.engine_state_counts().await { - Ok(counts) => { - for count in counts { - inner.metrics.set_rows(&count.table, &count.state, count.count); - } - } + Ok(counts) => inner.metrics.set_row_counts( + counts + .iter() + .map(|count| (count.table.as_str(), count.state.as_str(), count.count)), + ), Err(error) => { warn!(event = "engine_recheck", error = %error, "row count refresh failed") }