Skip to content

fix: bound graph rebuilds of the public RPC and release its connections before external calls #534

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
KSlashh wants to merge 1 commit into
base: dev
Choose a base branch
from
Open

fix: bound graph rebuilds of the public RPC and release its connections before external calls #534

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
130 changes: 79 additions & 51 deletions node/src/rpc_service/handler/bitvm_handler.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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};
Expand All @@ -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<Option<BitvmGcGraph>> {
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,
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -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")?;
Expand Down Expand Up @@ -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() {
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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,
Expand All @@ -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())),
Expand Down
122 changes: 117 additions & 5 deletions node/src/utils/graph_store.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -508,6 +508,43 @@ pub(crate) async fn load_validated_graph_definition(
instance_id: Uuid,
graph_id: Uuid,
) -> Result<Option<SimplifiedBitvmGcGraph>> {
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<BitvmGcGraph> {
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<Option<StoredGraphDefinition>> {
let Some(graph_row) = storage_processor.find_graph(&graph_id).await? else {
return Ok(None);
};
Expand All @@ -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<BitvmGcGraph> {
if simplified_graph.parameters.graph_id != graph_id
|| simplified_graph.parameters.instance_parameters.instance_id != instance_id
{
Expand All @@ -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}"
))
Expand All @@ -563,8 +615,7 @@ pub(crate) async fn load_validated_graph_definition(
))
})?;
}

Ok(Some(simplified_graph))
Ok(full_graph)
}

pub async fn get_graph(
Expand Down Expand Up @@ -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:#}");
}
}
Loading