From 8fadf78f0e8716975469e9073cee8b5c160de453 Mon Sep 17 00:00:00 2001 From: Ethan Date: Tue, 10 Nov 2026 07:25:46 +0000 Subject: [PATCH] fix: bound public graph rebuilds and release RPC connections before external calls --- node/src/rpc_service/handler/bitvm_handler.rs | 130 +++++++++++------- node/src/utils/graph_store.rs | 122 +++++++++++++++- 2 files changed, 196 insertions(+), 56 deletions(-) diff --git a/node/src/rpc_service/handler/bitvm_handler.rs b/node/src/rpc_service/handler/bitvm_handler.rs index 599745d30..568f524ee 100644 --- a/node/src/rpc_service/handler/bitvm_handler.rs +++ b/node/src/rpc_service/handler/bitvm_handler.rs @@ -10,12 +10,12 @@ use crate::rpc_service::response::{ ApiErrorExt, ApiResult, ErrorResponse, error_response, ok_response, }; use crate::rpc_service::validation::InputValidator; -#[cfg(feature = "rpc-debug-endpoints")] -use crate::utils::send_challenge_tx; use crate::utils::{ - gen_instance_parameters_local, get_bridge_out_global_stats, load_validated_graph_definition, - obsolete_graph, + gen_instance_parameters_local, get_bridge_out_global_stats, obsolete_graph, + read_graph_definition, }; +#[cfg(feature = "rpc-debug-endpoints")] +use crate::utils::{load_validated_graph_definition, send_challenge_tx}; use alloy::primitives::U256; use axum::Json; use axum::extract::{Path, Query, State}; @@ -28,9 +28,31 @@ use std::collections::HashSet; use std::default::Default; use std::str::FromStr; use std::sync::Arc; -use store::localdb::{GraphQuery, InstanceQuery, StorageProcessor}; +use store::localdb::{GraphQuery, InstanceQuery, LocalDB, StorageProcessor}; use store::{Graph, GraphStatus, InstanceBridgeInStatus}; +use tokio::sync::Semaphore; use tracing::{info, warn}; +use uuid::Uuid; + +/// Public requests rebuilding a graph at a time; each rebuild and its +/// signature checks take a blocking thread. +static GRAPH_BUILDS: Semaphore = Semaphore::const_new(2); + +/// The validated graph of a public request. The connection is released +/// before the rebuild, which runs on the blocking pool. +async fn load_public_graph( + local_db: &LocalDB, + instance_id: Uuid, + graph_id: Uuid, +) -> anyhow::Result> { + let _permit = GRAPH_BUILDS.acquire().await?; + let stored = + read_graph_definition(&mut local_db.acquire().await?, instance_id, graph_id).await?; + let Some(stored) = stored else { + return Ok(None); + }; + Ok(Some(tokio::task::spawn_blocking(move || stored.build()).await??)) +} fn graph_tx_index_error( tx_name: GraphBtcTxName, @@ -177,6 +199,8 @@ pub async fn get_instances( let (instances, total) = storage_process.find_instances(query).await.api_error("GET_INSTANCE_ERROR")?; + // Released before the Goat and Bitcoin calls. + drop(storage_process); if instances.is_empty() { warn!("get_instances instance is empty: total {}", total); @@ -285,11 +309,16 @@ pub async fn get_instance( // Validate instance_id format let instance_id_uuid = InputValidator::validate_uuid(&instance_id, "instance_id")?; - let mut storage_process = app_state.local_db.acquire().await.api_error("GET_INSTANCE_ERROR")?; + let instance = app_state + .local_db + .acquire() + .await + .api_error("GET_INSTANCE_ERROR")? + .find_instance(&instance_id_uuid) + .await + .api_error("GET_INSTANCE_ERROR")?; - if let Some(instance) = - storage_process.find_instance(&instance_id_uuid).await.api_error("GET_INSTANCE_ERROR")? - { + if let Some(instance) = instance { let response_window_blocks = app_state .goat_client .gateway_get_response_window_blocks() @@ -470,11 +499,16 @@ pub async fn get_graph( // Validate graph_id format let graph_id_uuid = InputValidator::validate_uuid(&graph_id, "graph_id")?; - let mut storage_process = app_state.local_db.acquire().await.api_error("GET_GRAPH_ERROR")?; + let graph = app_state + .local_db + .acquire() + .await + .api_error("GET_GRAPH_ERROR")? + .find_graph(&graph_id_uuid) + .await + .api_error("GET_GRAPH_ERROR")?; - if let Some(graph) = - storage_process.find_graph(&graph_id_uuid).await.api_error("GET_GRAPH_ERROR")? - { + if let Some(graph) = graph { let graph_extended = GraphExtended::convert_from_graph(&app_state.btc_client, graph) .await .api_error("GET_GRAPH_ERROR")?; @@ -568,6 +602,8 @@ pub async fn get_graphs( let filter_params: GraphQuery = params.into(); let (graphs, total) = storage_process.find_graphs(filter_params).await.api_error("GET_GRAPHS_ERROR")?; + // Released before the Bitcoin calls. + drop(storage_process); resp.total = total; if graphs.is_empty() { @@ -813,23 +849,20 @@ pub async fn get_graph_tx( get_graph_btc_tx_process_data(&mut storage_process, tx_name, &graph) .await .api_error("GET_GRAPH_TX_ERROR")?; + drop(storage_process); - let simplified_bitvm_graph = - load_validated_graph_definition(&mut storage_process, graph.instance_id, graph_id_uuid) - .await - .api_error("GET_GRAPH_TXN_ERROR")? - .ok_or_else(|| { - ( - StatusCode::INTERNAL_SERVER_ERROR, - Json(ErrorResponse { - error: "GET_GRAPH_TXN_ERROR".to_string(), - message: format!("graph:{graph_id} raw data is not record in db"), - }), - ) - })?; - - let bitvm_graph: BitvmGcGraph = BitvmGcGraph::from_simplified(&simplified_bitvm_graph) - .api_error("GET_GRAPH_TX_ERROR")?; + let bitvm_graph = load_public_graph(&app_state.local_db, graph.instance_id, graph_id_uuid) + .await + .api_error("GET_GRAPH_TXN_ERROR")? + .ok_or_else(|| { + ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(ErrorResponse { + error: "GET_GRAPH_TXN_ERROR".to_string(), + message: format!("graph:{graph_id} raw data is not record in db"), + }), + ) + })?; let tx_index = params.index.unwrap_or(0); let raw_data = match tx_name { @@ -977,28 +1010,6 @@ pub async fn get_graph_txn( graph }; - let simplified_bitvm_graph = load_validated_graph_definition( - &mut storage_processor, - graph.instance_id, - graph.graph_id, - ) - .await - .api_error("GET_GRAPH_TXN_ERROR")? - .ok_or_else(|| { - ( - StatusCode::INTERNAL_SERVER_ERROR, - Json(ErrorResponse { - error: "GET_GRAPH_TXN_ERROR".to_string(), - message: format!( - "graph:{graph_id} with cursor:{} raw data is not record in db", - params.cursor - ), - }), - ) - })?; - let bitvm_graph: BitvmGcGraph = BitvmGcGraph::from_simplified(&simplified_bitvm_graph) - .api_error("GET_GRAPH_TXN_ERROR")?; - let (wt_progresses, wt_fail_reason) = get_graph_btc_tx_process_data( &mut storage_processor, GraphBtcTxName::WatchtowerChallengeInit, @@ -1014,6 +1025,23 @@ pub async fn get_graph_txn( ) .await .api_error("GET_GRAPH_TXN_ERROR")?; + drop(storage_processor); + + let bitvm_graph = load_public_graph(&app_state.local_db, graph.instance_id, graph.graph_id) + .await + .api_error("GET_GRAPH_TXN_ERROR")? + .ok_or_else(|| { + ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(ErrorResponse { + error: "GET_GRAPH_TXN_ERROR".to_string(), + message: format!( + "graph:{graph_id} with cursor:{} raw data is not record in db", + params.cursor + ), + }), + ) + })?; let mut resp = GraphTxnGetResponse { cur_prekickoff: BtcTxData::new(serialize_hex(bitvm_graph.cur_prekickoff.tx())), diff --git a/node/src/utils/graph_store.rs b/node/src/utils/graph_store.rs index 9fbbaebfb..abed7037a 100644 --- a/node/src/utils/graph_store.rs +++ b/node/src/utils/graph_store.rs @@ -508,6 +508,43 @@ pub(crate) async fn load_validated_graph_definition( instance_id: Uuid, graph_id: Uuid, ) -> Result> { + let Some(stored) = read_graph_definition(storage_processor, instance_id, graph_id).await? + else { + return Ok(None); + }; + let simplified_graph = parse_graph_raw_data(stored.raw_data, graph_id).await?; + check_graph_definition(instance_id, graph_id, &stored.definition_hash, &simplified_graph)?; + Ok(Some(simplified_graph)) +} + +/// A raw definition read with its row, not yet parsed or checked. +pub(crate) struct StoredGraphDefinition { + instance_id: Uuid, + graph_id: Uuid, + definition_hash: String, + raw_data: String, +} + +impl StoredGraphDefinition { + /// The checks of `load_validated_graph_definition` past the database + /// read, returning the rebuilt graph. CPU-bound. + pub(crate) fn build(self) -> Result { + let simplified_graph: SimplifiedBitvmGcGraph = serde_json::from_str(&self.raw_data)?; + check_graph_definition( + self.instance_id, + self.graph_id, + &self.definition_hash, + &simplified_graph, + ) + } +} + +/// The database half of `load_validated_graph_definition`. +pub(crate) async fn read_graph_definition( + storage_processor: &mut StorageProcessor<'_>, + instance_id: Uuid, + graph_id: Uuid, +) -> Result> { let Some(graph_row) = storage_processor.find_graph(&graph_id).await? else { return Ok(None); }; @@ -531,7 +568,22 @@ pub(crate) async fn load_validated_graph_definition( "graph {graph_id} has a runtime row but no raw definition; explicit repair is required" )) })?; - let simplified_graph = parse_graph_raw_data(graph_raw_data.raw_data, graph_id).await?; + Ok(Some(StoredGraphDefinition { + instance_id, + graph_id, + definition_hash: graph_row.definition_hash, + raw_data: graph_raw_data.raw_data, + })) +} + +/// Bind a parsed definition to its row and rebuild it with its pre-signatures +/// verified. +fn check_graph_definition( + instance_id: Uuid, + graph_id: Uuid, + stored_definition_hash: &str, + simplified_graph: &SimplifiedBitvmGcGraph, +) -> Result { if simplified_graph.parameters.graph_id != graph_id || simplified_graph.parameters.instance_parameters.instance_id != instance_id { @@ -540,13 +592,13 @@ pub(crate) async fn load_validated_graph_definition( ))); } let definition_hash = hex::encode(simplified_graph.parameters_hash()?); - if definition_hash != graph_row.definition_hash { + if definition_hash != stored_definition_hash { bail!(SpecialError::InvalidGraph(format!( "stored raw definition hash does not match graph row for {instance_id}:{graph_id}" ))); } - let full_graph = BitvmGcGraph::from_simplified(&simplified_graph).map_err(|error| { + let full_graph = BitvmGcGraph::from_simplified(simplified_graph).map_err(|error| { SpecialError::InvalidGraph(format!( "stored raw definition cannot rebuild graph for {instance_id}:{graph_id}: {error}" )) @@ -563,8 +615,7 @@ pub(crate) async fn load_validated_graph_definition( )) })?; } - - Ok(Some(simplified_graph)) + Ok(full_graph) } pub async fn get_graph( @@ -698,3 +749,64 @@ pub(crate) async fn pending_graph_belongs_to_operator( pending.instance_id == instance_id && pending.operator_pubkey == operator_pubkey.to_string() })) } + +#[cfg(test)] +mod definition_tests { + use super::*; + include!(concat!( + env!("CARGO_MANIFEST_DIR"), + "/../crates/bitvm-gc/tests/fixtures/admission.rs" + )); + + #[tokio::test] + async fn the_public_build_checks_what_the_validated_load_checks() { + let dir = tempfile::tempdir().unwrap(); + let db = + LocalDB::new(&format!("sqlite:{}", dir.path().join("defs.db").display()), false).await; + db.migrate().await; + let parameters = + admission_parameters(crate::env::get_network(), crate::utils::min_required_verifier()); + let (instance_id, graph_id) = + (parameters.instance_parameters.instance_id, parameters.graph_id); + let raw_data = serde_json::json!({"parameters":parameters,"operator_pre_signed":false,"committee_pre_signed":false,"operator_pre_sigs":null,"committee_pre_sigs":null}).to_string(); + let simplified: SimplifiedBitvmGcGraph = serde_json::from_str(&raw_data).unwrap(); + let definition_hash = hex::encode(simplified.parameters_hash().unwrap()); + let mut storage = db.acquire().await.unwrap(); + storage + .upsert_graph_definition(&Graph { + graph_id, + instance_id, + status: GraphStatus::OperatorPresigned.to_string(), + definition_hash: definition_hash.clone(), + operator_pubkey: parameters.operator_pubkey.to_string(), + ..Default::default() + }) + .await + .unwrap(); + storage + .upsert_graph_raw_data( + instance_id, + GraphRawData { graph_id, raw_data, created_at: 0, updated_at: 0 }, + &definition_hash, + ) + .await + .unwrap(); + + assert!( + read_graph_definition(&mut storage, instance_id, Uuid::new_v4()) + .await + .unwrap() + .is_none() + ); + assert!(read_graph_definition(&mut storage, Uuid::new_v4(), graph_id).await.is_err()); + // Without its operator pre-signatures the graph fails both ways alike. + let stored = read_graph_definition(&mut storage, instance_id, graph_id).await.unwrap(); + let built = stored.unwrap().build().err().unwrap(); + let loaded = load_validated_graph_definition(&mut storage, instance_id, graph_id) + .await + .err() + .unwrap(); + assert_eq!(format!("{built:#}"), format!("{loaded:#}")); + assert!(format!("{built:#}").contains("operator pre-signatures"), "{built:#}"); + } +}