diff --git a/crates/store/src/schema.rs b/crates/store/src/schema.rs index 2dbc4fea..73c4d82d 100644 --- a/crates/store/src/schema.rs +++ b/crates/store/src/schema.rs @@ -412,6 +412,14 @@ impl GraphStatus { ]; const SKIPPED: &[GraphStatus] = &[OperatorPresigned, CommitteePresigned, OperatorDataPushed, PreKickoff, Obsoleted]; + const OBSOLETED: &[GraphStatus] = &[ + OperatorPresigned, + CommitteePresigned, + OperatorDataPushed, + PreKickoff, + OperatorKickOff, + Challenge, + ]; match source { GraphStatusSource::Definition => match self { @@ -426,11 +434,9 @@ impl GraphStatus { }, GraphStatusSource::ChainReconcile => match self { CommitteePresigned => &[OperatorPresigned], - // Obsoleted is a provisional branch when GraphData was not - // yet visible. A later full chain scan may correct it, but - // ordinary Goat event replay may not. OperatorDataPushed => EARLY_OR_OBSOLETED, - PreKickoff | Obsoleted => PRE_KICKOFF, + Obsoleted => OBSOLETED, + PreKickoff => PRE_KICKOFF, OperatorKickOff => KICKOFF, Challenge => CHALLENGE, OperatorTake1 => TAKE1, diff --git a/node/src/action.rs b/node/src/action.rs index 1c56f2f2..eee7d432 100644 --- a/node/src/action.rs +++ b/node/src/action.rs @@ -2,7 +2,7 @@ #![allow(clippy::single_match)] #![allow(clippy::collapsible_else_if)] -use crate::env::get_local_node_info; +use crate::env::{get_local_node_info, get_p2p_inbox_batch_size, get_p2p_outbox_batch_size}; use crate::handle::{ HandlerContext, HeavyTaskContext, dispatch as handle_dispatch, heavy_task_from_content, is_heavy_task_message_type, run_heavy_task, @@ -46,7 +46,6 @@ pub struct GOATMessage { const GOAT_MESSAGE_BIN_PREFIX: &[u8] = b"GOATBIN1"; const TRANSIENT_PEGIN_RETRY_DELAY_SECS: usize = 30; -const P2P_INBOX_BATCH_SIZE: i64 = 8; const P2P_INBOX_LEASE_SECS: i64 = 5 * 60; const P2P_INBOX_LEASE_RENEW_INTERVAL_SECS: u64 = 60; const P2P_INBOX_ENQUEUE_ATTEMPTS: usize = 3; @@ -940,7 +939,7 @@ async fn handle_p2p_inbox_messages( .claim_p2p_inbox_messages( now, now + P2P_INBOX_LEASE_SECS, - P2P_INBOX_BATCH_SIZE, + get_p2p_inbox_batch_size(), &active_heavy_task_ids, ) .await?; @@ -1277,7 +1276,7 @@ async fn handle_p2p_outbox_messages( let now = current_time_secs(); let mut storage = local_db.start_immediate_transaction().await?; let messages = storage - .claim_p2p_outbox_messages(now, now + P2P_INBOX_LEASE_SECS, P2P_INBOX_BATCH_SIZE) + .claim_p2p_outbox_messages(now, now + P2P_INBOX_LEASE_SECS, get_p2p_outbox_batch_size()) .await?; storage.commit().await?; diff --git a/node/src/bin/send_pegout.rs b/node/src/bin/send_pegout.rs index 10de16d9..a3ae1298 100644 --- a/node/src/bin/send_pegout.rs +++ b/node/src/bin/send_pegout.rs @@ -51,6 +51,10 @@ enum Commands { /// Dry run (validate but do not call Gateway.initWithdraw) #[arg(long, default_value_t = false)] dry_run: bool, + + /// Skip graphs whose instances are locked by another withdrawal + #[arg(long, default_value_t = false)] + skip_locked: bool, }, /// Initiate pegouts repeatedly until target amount reached Batch { @@ -66,6 +70,10 @@ enum Commands { #[arg(long, default_value_t = false)] dry_run: bool, + /// Skip graphs whose instances are locked by another withdrawal + #[arg(long, default_value_t = false)] + skip_locked: bool, + /// Poll interval seconds for checking graph readiness between pegouts #[arg(long, default_value_t = 300)] poll_interval_secs: u64, @@ -81,6 +89,7 @@ struct PegoutApiRequest { #[serde(skip_serializing_if = "Option::is_none")] graph_id: Option, dry_run: bool, + skip_locked: bool, } #[derive(Debug, Deserialize)] @@ -114,9 +123,11 @@ async fn call_pegout( base_url: &str, graph_id: Option, dry_run: bool, + skip_locked: bool, ) -> Result { let url = format!("{}/v1/graphs/pegout", base_url.trim_end_matches('/')); - let body = PegoutApiRequest { graph_id: graph_id.map(|id| id.to_string()), dry_run }; + let body = + PegoutApiRequest { graph_id: graph_id.map(|id| id.to_string()), dry_run, skip_locked }; let keypair = get_bitvm_key().context("failed to load BITVM_SECRET")?; let (timestamp, signature) = sign_request_auth(&keypair); @@ -208,14 +219,15 @@ async fn main() -> Result<()> { let base_url = &args.rpc_url; match args.command { - Commands::Once { graph_id, dry_run } => { - let resp = call_pegout(&client, base_url, graph_id, dry_run).await?; + Commands::Once { graph_id, dry_run, skip_locked } => { + let resp = call_pegout(&client, base_url, graph_id, dry_run, skip_locked).await?; print_pegout_result(&resp); } Commands::Batch { max_total_amount_sats, max_count, dry_run, + skip_locked, poll_interval_secs, max_wait_secs, } => { @@ -236,7 +248,7 @@ async fn main() -> Result<()> { break; } - match call_pegout(&client, base_url, None, true).await { + match call_pegout(&client, base_url, None, true, skip_locked).await { Ok(preflight) => { let amount = u64::try_from(preflight.amount) .context("pegout API returned a negative amount")?; @@ -257,7 +269,9 @@ async fn main() -> Result<()> { let graph_id = Uuid::parse_str(&preflight.graph_id) .context("pegout API returned an invalid graph id")?; - let resp = call_pegout(&client, base_url, Some(graph_id), false).await?; + let resp = + call_pegout(&client, base_url, Some(graph_id), false, skip_locked) + .await?; print_pegout_result(&resp); total_sent = total_sent.saturating_add(amount); count = count.saturating_add(1); diff --git a/node/src/env.rs b/node/src/env.rs index fdf70c91..35f3d062 100644 --- a/node/src/env.rs +++ b/node/src/env.rs @@ -142,6 +142,10 @@ pub const ENV_INSTANCE_MAINTENANCE_BATCH_SIZE: &str = "INSTANCE_MAINTENANCE_BATC pub const DEFAULT_INSTANCE_MAINTENANCE_BATCH_SIZE: u32 = 50; pub const ENV_MAINTENANCE_RUN_TIMEOUT_SECS: &str = "MAINTENANCE_RUN_TIMEOUT_SECS"; pub const DEFAULT_MAINTENANCE_RUN_TIMEOUT_SECS: u64 = 60; +pub const ENV_P2P_INBOX_BATCH_SIZE: &str = "P2P_INBOX_BATCH_SIZE"; +pub const DEFAULT_P2P_INBOX_BATCH_SIZE: i64 = 16; +pub const ENV_P2P_OUTBOX_BATCH_SIZE: &str = "P2P_OUTBOX_BATCH_SIZE"; +pub const DEFAULT_P2P_OUTBOX_BATCH_SIZE: i64 = 16; pub const ENV_ENABLE_COMMITTEE_INSTANCE_KEY_DELETE: &str = "ENABLE_COMMITTEE_INSTANCE_KEY_DELETE"; pub const DEFAULT_ENABLE_COMMITTEE_INSTANCE_KEY_DELETE: bool = false; pub const ENV_COMMITTEE_INSTANCE_KEY_DELETE_TIMELOCK_BLOCKS: &str = @@ -652,6 +656,22 @@ pub fn get_maintenance_run_timeout_secs() -> u64 { .unwrap_or(DEFAULT_MAINTENANCE_RUN_TIMEOUT_SECS) } +pub fn get_p2p_inbox_batch_size() -> i64 { + std::env::var(ENV_P2P_INBOX_BATCH_SIZE) + .ok() + .and_then(|value| value.parse::().ok()) + .map(|size| size.clamp(1, 128)) + .unwrap_or(DEFAULT_P2P_INBOX_BATCH_SIZE) +} + +pub fn get_p2p_outbox_batch_size() -> i64 { + std::env::var(ENV_P2P_OUTBOX_BATCH_SIZE) + .ok() + .and_then(|value| value.parse::().ok()) + .map(|size| size.clamp(1, 128)) + .unwrap_or(DEFAULT_P2P_OUTBOX_BATCH_SIZE) +} + pub fn is_enable_committee_instance_key_delete() -> bool { // TODO: enable this feature when ready tracing::warn!( diff --git a/node/src/rpc_service/bitvm.rs b/node/src/rpc_service/bitvm.rs index 851b2d45..3be10219 100644 --- a/node/src/rpc_service/bitvm.rs +++ b/node/src/rpc_service/bitvm.rs @@ -111,6 +111,9 @@ pub struct PegoutRequest { pub graph_id: Option, #[serde(default)] pub dry_run: bool, + /// Skip graphs whose instances are temporarily locked by another withdrawal. + #[serde(default)] + pub skip_locked: bool, } #[derive(Debug, Deserialize, Serialize)] diff --git a/node/src/rpc_service/handler/bitvm_handler.rs b/node/src/rpc_service/handler/bitvm_handler.rs index 6979c504..1652b801 100644 --- a/node/src/rpc_service/handler/bitvm_handler.rs +++ b/node/src/rpc_service/handler/bitvm_handler.rs @@ -15,7 +15,7 @@ use crate::rpc_service::{AppState, current_time_secs}; use crate::utils::{ bridge_out_instance_id_from_escrow_hash, find_instances_by_escrow_hash, gen_instance_parameters_local, get_bridge_out_global_stats, load_validated_graph_definition, - send_challenge_tx, + obsolete_graph, send_challenge_tx, }; use alloy::primitives::{Address, U256}; use axum::Json; @@ -25,6 +25,7 @@ use bitvm_lib::types::BitvmGcGraph; use client::goat_chain::{PeginStatus, WithdrawStatus}; use goat::transactions::pre_signed::PreSignedTransaction; use http::{HeaderMap, StatusCode}; +use std::collections::HashSet; use std::default::Default; use std::str::FromStr; use std::sync::Arc; @@ -34,7 +35,7 @@ use store::{ GoatTxType, Graph, GraphStatus, Instance, InstanceBridgeInStatus, InstanceBridgeOutStatus, }; use tokio::time::{Duration, sleep}; -use tracing::warn; +use tracing::{info, warn}; use uuid::Uuid; fn bridge_out_retry_jitter_ms(attempt: u32) -> u64 { @@ -1786,8 +1787,11 @@ fn sats_to_token_amount(amount_sats: u64, token_decimals: u8) -> U256 { /// /// # Request Body /// -/// - `graph_id`: Optional UUID of the graph to pegout (auto-selects if omitted) +/// - `graph_id`: Optional preferred UUID of the graph to pegout. If it has +/// already been claimed by another graph, the next eligible graph is used. /// - `dry_run`: If true, validate but skip the actual initWithdraw call (default: false) +/// - `skip_locked`: If true, skip an instance temporarily locked by another +/// withdrawal and search for another withdrawable graph (default: false). /// /// # Returns /// @@ -1809,70 +1813,136 @@ pub async fn pegout( .api_error("PEGOUT_ERROR")?; let gateway_addr = get_goat_gateway_contract_from_env(); - let mut storage_process = app_state.local_db.acquire().await.api_error("PEGOUT_ERROR")?; + let mut preferred_graph_id = payload + .graph_id + .as_deref() + .map(|graph_id| InputValidator::validate_uuid(graph_id, "graph_id")) + .transpose()?; + let mut skipped_graph_ids = HashSet::new(); + + // A claimed instance can leave an old OperatorDataPushed graph in the + // local projection until the terminal withdraw event is observed. Reconcile + // it and select another candidate instead of failing pegout. + let (graph, pegin_data) = loop { + let graph = { + let mut storage_process = + app_state.local_db.acquire().await.api_error("PEGOUT_ERROR")?; + if let Some(graph_id) = preferred_graph_id { + let graph = storage_process + .find_graph(&graph_id) + .await + .api_error("PEGOUT_ERROR")? + .ok_or_else(|| { + ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(ErrorResponse { + error: "PEGOUT_ERROR".to_string(), + message: format!("graph {graph_id} not found"), + }), + ) + })?; + if graph.operator_pubkey != operator_pubkey { + return error_response( + "PEGOUT_ERROR".to_string(), + format!("graph {} not owned by this operator", graph.graph_id), + ); + } + if graph.status != GraphStatus::OperatorDataPushed.to_string() { + return error_response( + "PEGOUT_ERROR".to_string(), + format!( + "graph {} status {} is not OperatorDataPushed", + graph.graph_id, graph.status + ), + ); + } + if graph.init_withdraw_tx_hash.is_some() { + return error_response( + "PEGOUT_ERROR".to_string(), + format!("graph {} already initialized withdraw", graph.graph_id), + ); + } + graph + } else { + let graphs = storage_process + .get_operator_graphs( + GraphQuery::default() + .with_operator_pubkey(operator_pubkey.clone()) + .with_status(GraphStatus::OperatorDataPushed.to_string()) + .with_raw_condition("init_withdraw_tx_hash IS NULL".to_string()) + .with_order("kickoff_index ASC".to_string()), + ) + .await + .api_error("PEGOUT_ERROR")?; + graphs + .into_iter() + .find(|graph| !skipped_graph_ids.contains(&graph.graph_id)) + .ok_or_else(|| { + ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(ErrorResponse { + error: "PEGOUT_ERROR".to_string(), + message: format!( + "no eligible graph found for operator {operator_pubkey}" + ), + }), + ) + })? + } + }; - // Select graph - let graph = if let Some(ref graph_id_str) = payload.graph_id { - let graph_id_uuid = InputValidator::validate_uuid(graph_id_str, "graph_id")?; - let graph = storage_process - .find_graph(&graph_id_uuid) - .await - .api_error("PEGOUT_ERROR")? - .ok_or_else(|| { - ( - StatusCode::INTERNAL_SERVER_ERROR, - Json(ErrorResponse { - error: "PEGOUT_ERROR".to_string(), - message: format!("graph {graph_id_str} not found"), - }), - ) - })?; - if graph.operator_pubkey != operator_pubkey { - return error_response( - "PEGOUT_ERROR".to_string(), - format!("graph {} not owned by this operator", graph.graph_id), - ); - } - if graph.status != GraphStatus::OperatorDataPushed.to_string() { - return error_response( - "PEGOUT_ERROR".to_string(), - format!( - "graph {} status {} is not OperatorDataPushed", - graph.graph_id, graph.status - ), - ); - } - if graph.init_withdraw_tx_hash.is_some() { - return error_response( - "PEGOUT_ERROR".to_string(), - format!("graph {} already initialized withdraw", graph.graph_id), - ); - } - graph - } else { - // Auto-select: minimal kickoff-index graph with OperatorDataPushed, no init_withdraw - let graphs = storage_process - .get_operator_graphs( - GraphQuery::default() - .with_operator_pubkey(operator_pubkey.clone()) - .with_status(GraphStatus::OperatorDataPushed.to_string()) - .with_raw_condition("init_withdraw_tx_hash IS NULL".to_string()) - .with_order("kickoff_index ASC".to_string()) - .with_limit(1), - ) + let pegin_data = app_state + .goat_client + .gateway_get_pegin_data(&graph.instance_id) .await .api_error("PEGOUT_ERROR")?; - graphs.into_iter().next().ok_or_else(|| { - ( - StatusCode::INTERNAL_SERVER_ERROR, - Json(ErrorResponse { - error: "PEGOUT_ERROR".to_string(), - message: format!("no eligible graph found for operator {operator_pubkey}"), - }), - ) - })? + match pegin_data.status { + PeginStatus::Claimed => { + let withdraw_data = app_state + .goat_client + .gateway_get_withdraw_data(&graph.graph_id) + .await + .api_error("PEGOUT_ERROR")?; + if withdraw_data.status == WithdrawStatus::None { + let mut storage_process = + app_state.local_db.acquire().await.api_error("PEGOUT_ERROR")?; + let obsoleted = + obsolete_graph(&mut storage_process, graph.instance_id, graph.graph_id) + .await + .api_error("PEGOUT_ERROR")?; + warn!( + instance_id = %graph.instance_id, + skipped_graph_id = %graph.graph_id, + obsoleted, + "skipped obsolete graph while selecting pegout candidate" + ); + } + } + PeginStatus::Locked if !payload.skip_locked => { + return error_response( + "PEGOUT_IN_PROGRESS".to_string(), + format!( + "graph {} instance {} is locked by an in-progress withdrawal; retry later or set skip_locked=true", + graph.graph_id, graph.instance_id + ), + ); + } + PeginStatus::Locked => { + info!( + instance_id = %graph.instance_id, + skipped_graph_id = %graph.graph_id, + "skipped graph whose instance is locked by another withdrawal" + ); + } + _ => break (graph, pegin_data), + } + + skipped_graph_ids.insert(graph.graph_id); + preferred_graph_id = None; }; + let mut storage_process = app_state.local_db.acquire().await.api_error("PEGOUT_ERROR")?; + // Check previous graph readiness if graph.kickoff_index > 0 { let pre_graphs = storage_process @@ -1910,11 +1980,6 @@ pub async fn pegout( } // Check L2 pegin status - let pegin_data = app_state - .goat_client - .gateway_get_pegin_data(&graph.instance_id) - .await - .api_error("PEGOUT_ERROR")?; if pegin_data.status != PeginStatus::Withdrawable { return error_response( "PEGOUT_ERROR".to_string(), diff --git a/node/src/scheduled_tasks/event_watch_task.rs b/node/src/scheduled_tasks/event_watch_task.rs index e55389c5..e631c268 100644 --- a/node/src/scheduled_tasks/event_watch_task.rs +++ b/node/src/scheduled_tasks/event_watch_task.rs @@ -13,8 +13,8 @@ use crate::utils::evm_swap_utils::IEscrowManager::EscrowData; use crate::utils::evm_swap_utils::{extract_claim_data_from_tx, extract_escrow_data_from_tx}; use crate::utils::{ GenerateInstanceParams, bridge_out_instance_id_from_escrow_hash, find_instances_by_escrow_hash, - generate_instance, get_bridge_out_global_stats, outpoint_available, reflect_goat_address, - strip_hex_prefix_owned, + generate_instance, get_bridge_out_global_stats, obsolete_instance_graphs_except, + outpoint_available, reflect_goat_address, strip_hex_prefix_owned, }; use alloy::primitives::{Address as EvmAddress, U256}; use alloy::sol_types::{SolType, SolValue}; @@ -537,6 +537,16 @@ async fn handle_withdraw_paths_events<'a>( ) { continue; } + let obsoleted_graph_ids = + obsolete_instance_graphs_except(storage_processor, instance_id, Some(graph_id)).await?; + if !obsoleted_graph_ids.is_empty() { + info!( + instance_id = %instance_id, + completed_graph_id = %graph_id, + obsoleted_graph_ids = ?obsoleted_graph_ids, + "marked other instance graphs obsolete after completed withdrawal" + ); + } let is_new_event = storage_processor .find_graph_goat_tx_record(&instance_id, &graph_id, &tx_type) .await? diff --git a/node/src/utils.rs b/node/src/utils.rs index e6b637b7..511e86c7 100644 --- a/node/src/utils.rs +++ b/node/src/utils.rs @@ -1451,13 +1451,13 @@ async fn scan_graph_chain_state( if graph_data_on_goat.operator_pubkey != [0u8; 32] { current_status = GraphStatus::OperatorDataPushed; } - // check if Graph has been obsoleted on GoatChain + // A claimed pegin is terminal, so a graph which did not complete that + // withdrawal can never be used. A Locked pegin is intentionally excluded: + // the active withdrawal may still be cancelled and return to Withdrawable. if current_status == GraphStatus::OperatorDataPushed { let pegin_data = goat_client.gateway_get_pegin_data(&instance_id).await?; let withdraw_data = goat_client.gateway_get_withdraw_data(&graph_id).await?; - // NOTE: maybe obesolete graph when pegin is claimed rather than processing? - if pegin_data.status != PeginStatus::Withdrawable - && withdraw_data.status == WithdrawStatus::None + if pegin_data.status == PeginStatus::Claimed && withdraw_data.status == WithdrawStatus::None { current_status = GraphStatus::Obsoleted; } @@ -5641,6 +5641,67 @@ pub async fn update_graph_status( ) .await } + +/// Mark one graph obsolete and cancel its pending messages. +pub(crate) async fn obsolete_graph( + storage_processor: &mut StorageProcessor<'_>, + instance_id: Uuid, + graph_id: Uuid, +) -> Result { + let outcome = storage_processor + .transition_graph_status( + instance_id, + graph_id, + GraphStatus::Obsoleted, + GraphStatusSource::ChainReconcile, + None, + ) + .await?; + if !matches!( + outcome, + GraphStatusTransitionOutcome::Applied | GraphStatusTransitionOutcome::AlreadyCurrent + ) { + return Ok(false); + } + + // `None` means no message-type filter: cancel every durable pending message + // for this terminal graph so stale retries cannot consume queue capacity or + // trigger a later graph action. + storage_processor + .update_messages_state_by_business_id( + &graph_id, + None, + MessageState::Pending.to_string(), + MessageState::Cancelled.to_string(), + ) + .await?; + Ok(true) +} + +/// Mark every non-terminal graph for an already claimed instance as obsolete. +/// +/// `retained_graph_id` is the graph which completed the withdrawal. The caller +/// must only use this after a terminal Gateway withdraw event confirms that +/// this graph completed the withdrawal. +pub(crate) async fn obsolete_instance_graphs_except( + storage_processor: &mut StorageProcessor<'_>, + instance_id: Uuid, + retained_graph_id: Option, +) -> Result> { + let graphs = storage_processor.get_graphs_by_instance_id(&instance_id).await?; + let mut obsoleted_graph_ids = Vec::new(); + + for graph in graphs { + if Some(graph.graph_id) != retained_graph_id + && obsolete_graph(storage_processor, instance_id, graph.graph_id).await? + { + obsoleted_graph_ids.push(graph.graph_id); + } + } + + Ok(obsoleted_graph_ids) +} + pub async fn get_graph_ids_for_instance( local_db: &LocalDB, instance_id: Uuid,