diff --git a/node/src/aggregates/graph.rs b/node/src/aggregates/graph.rs index 898be72f..2e6e7a89 100644 --- a/node/src/aggregates/graph.rs +++ b/node/src/aggregates/graph.rs @@ -19,6 +19,7 @@ use crate::engine_bridge::record_event_in; use crate::messages::{GOATMessage, GOATMessageContent, verify_init_graph_signature}; use crate::metrics_service::node_metrics_state; use crate::middleware::get_topic_name; +use crate::utils::stored_definition_committee_pre_signed; use anyhow::anyhow; use async_trait::async_trait; use bitvm_lib::actors::Actor; @@ -1169,7 +1170,17 @@ impl AggregateHandler for GraphAggregate { ))); } }; - if row.is_none() { + // A finalized lookup also runs when only the operator's copy is + // stored, whatever the status. + let wanted = match (&row, &pull.lookup) { + (None, _) => true, + (Some(_), GraphLookup::Id { finalized: true, .. }) => { + stored_definition_committee_pre_signed(ctx.storage, &graph_id).await? + != Some(true) + } + (Some(_), _) => false, + }; + if wanted { let params = bincode::serialize(&pull.lookup) .map_err(|error| anyhow::anyhow!("encode graph lookup: {error}"))?; self.enqueue_setup_job(ctx, JOB_FETCH_GRAPH, ¶ms); diff --git a/node/src/aggregates/graph_fetch.rs b/node/src/aggregates/graph_fetch.rs index b5937507..3e661699 100644 --- a/node/src/aggregates/graph_fetch.rs +++ b/node/src/aggregates/graph_fetch.rs @@ -7,8 +7,9 @@ use crate::middleware::get_graph::{GetGraphRequest, GetGraphResponse}; use crate::middleware::publisher::NetworkPublisher; use crate::utils::{ - load_validated_graph_definition, parse_graph_raw_data, store_finalized_graph_if_needed, - store_operator_presigned_graph, validate_graph_instance_parameters, + load_validated_graph_definition, parse_graph_raw_data, raw_definition_committee_pre_signed, + store_finalized_graph_if_needed, store_operator_presigned_graph, + validate_graph_instance_parameters, }; use anyhow::{Context, bail}; use bitvm_lib::operator::verify_graph_operator_pre_signatures; @@ -18,7 +19,6 @@ use client::goat_chain::GOATClient; use libp2p::PeerId; use serde::{Deserialize, Serialize}; use std::str::FromStr; -use store::GraphStatus; use store::localdb::LocalDB; use tracing::{info, warn}; use uuid::Uuid; @@ -73,13 +73,15 @@ pub async fn serve_graph_request( { return Ok(GetGraphResponse::not_found()); } - // Only the operator's copy is stored while the graph is OperatorPresigned. - if request.finalized && row.status == GraphStatus::OperatorPresigned.to_string() { - return Ok(GetGraphResponse::not_found()); - } let Some(raw) = storage.find_graph_raw_data(&row.graph_id).await? else { return Ok(GetGraphResponse::not_found()); }; + // A finalized copy is one with committee pre-signatures. The status says + // nothing about it: a Goat event moves an operator-only copy past + // OperatorPresigned too. + if request.finalized && !raw_definition_committee_pre_signed(&raw.raw_data)? { + return Ok(GetGraphResponse::not_found()); + } Ok(GetGraphResponse::found(row.instance_id, row.graph_id, raw.raw_data)) } @@ -346,3 +348,65 @@ mod admission_tests { assert!(db.acquire().await.unwrap().find_graph(&graph_id).await.unwrap().is_none()); } } + +#[cfg(test)] +mod serve_tests { + use super::*; + use store::{Graph, GraphRawData, GraphStatus, GraphStatusSource}; + + #[tokio::test] + async fn a_finalized_request_is_served_only_a_copy_with_committee_pre_signatures() { + let dir = tempfile::tempdir().unwrap(); + let db = LocalDB::new_with_busy_timeout( + &format!("sqlite:{}", dir.path().join("serve.db").display()), + false, + store::DEFAULT_BUSY_TIMEOUT, + ) + .await; + db.migrate().await; + for committee_pre_signed in [false, true] { + let (instance_id, graph_id) = (Uuid::new_v4(), Uuid::new_v4()); + let mut storage = db.acquire().await.unwrap(); + storage + .upsert_graph_definition(&Graph { + graph_id, + instance_id, + status: GraphStatus::OperatorPresigned.to_string(), + definition_hash: "definition".to_string(), + kickoff_index: i64::from(committee_pre_signed), + ..Default::default() + }) + .await + .unwrap(); + // Past OperatorPresigned either way: a Goat event moves the status + // of an operator-only copy too. + storage + .transition_graph_status( + instance_id, + graph_id, + GraphStatus::OperatorDataPushed, + GraphStatusSource::ChainReconcile, + None, + ) + .await + .unwrap(); + let raw = + serde_json::json!({ "committee_pre_signed": committee_pre_signed }).to_string(); + storage + .upsert_graph_raw_data( + instance_id, + GraphRawData { graph_id, raw_data: raw, created_at: 0, updated_at: 0 }, + "definition", + ) + .await + .unwrap(); + drop(storage); + + let finalized = GetGraphRequest::by_id(instance_id, graph_id, true); + let served = serve_graph_request(&db, &finalized).await.unwrap(); + assert_eq!(served.graph.is_some(), committee_pre_signed); + let any = GetGraphRequest::by_id(instance_id, graph_id, false); + assert!(serve_graph_request(&db, &any).await.unwrap().graph.is_some()); + } + } +} diff --git a/node/src/aggregates/graph_tests.rs b/node/src/aggregates/graph_tests.rs index 9090e8c5..3dd0831b 100644 --- a/node/src/aggregates/graph_tests.rs +++ b/node/src/aggregates/graph_tests.rs @@ -526,6 +526,37 @@ async fn a_definition_waits_for_its_predecessor_and_pulls_it() { t.stop().await; } +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_finalized_pull_runs_while_only_the_operator_copy_is_stored() { + let t = start(Actor::Verifier, false).await; + for (nonce, committee_pre_signed) in [(0, false), (1, true)] { + let instance_id = Uuid::new_v4(); + let graph_id = t.insert_graph(instance_id, nonce, GraphStatus::OperatorDataPushed).await; + let raw = serde_json::json!({ "committee_pre_signed": committee_pre_signed }).to_string(); + let mut storage = t.db.acquire().await.unwrap(); + storage + .upsert_graph_raw_data( + instance_id, + store::GraphRawData { graph_id, raw_data: raw, created_at: 0, updated_at: 0 }, + "definition", + ) + .await + .unwrap(); + drop(storage); + let lookup = GraphLookup::Id { instance_id, graph_id, finalized: true }; + let event = graph_pull_event(graph_id, lookup.clone()).unwrap(); + let event_id = t.submit(event).await; + t.wait_inbox(&event_id, EngineInboxState::Applied).await; + if committee_pre_signed { + settle().await; + assert!(t.jobs_of(graph_id).await.iter().all(|job| job.action != JOB_FETCH_GRAPH)); + } else { + t.wait_job(graph_id, JOB_FETCH_GRAPH, EngineJobState::Done).await; + } + } + t.stop().await; +} + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn a_value_from_an_unregistered_author_waits_without_a_pull() { let t = start_with(Actor::Committee, false, Arc::new(Unacquainted)).await; diff --git a/node/src/scheduled_tasks/event_watch/gateway.rs b/node/src/scheduled_tasks/event_watch/gateway.rs index 4f528a02..91783cc8 100644 --- a/node/src/scheduled_tasks/event_watch/gateway.rs +++ b/node/src/scheduled_tasks/event_watch/gateway.rs @@ -10,7 +10,7 @@ use crate::clock::current_time_secs; use crate::engine_bridge::record_event_in; use crate::utils::{ generate_instance_from_bridge_in_request_event, obsolete_instance_graphs_except, - reflect_goat_address, strip_hex_prefix_owned, + reflect_goat_address, stored_definition_committee_pre_signed, strip_hex_prefix_owned, }; use alloy::primitives::U256; use bitvm_lib::actors::Actor; @@ -665,10 +665,16 @@ async fn handle_post_graph_data_events<'a>( ) .await?; // Watching roles fetch a finalized definition when the posted graph is absent - // locally. - if matches!(outcome, GraphStatusTransitionOutcome::NotFound) - && matches!(actor, Actor::Verifier | Actor::Watchtower) - { + // locally. Any role holding only the operator's copy fetches it too: the + // status just moved past OperatorPresigned, but the copy still lacks the + // committee pre-signatures. + let pull = if matches!(outcome, GraphStatusTransitionOutcome::NotFound) { + matches!(actor, Actor::Verifier | Actor::Watchtower) + } else { + stored_definition_committee_pre_signed(storage_processor, &graph_id).await? + == Some(false) + }; + if pull { record_event_in( storage_processor, graph_pull_event( @@ -763,4 +769,56 @@ mod tests { ); } } + + /// A graph row with a stored definition; only its committee flag is read. + async fn stored(db: &LocalDB, slot: i64, committee_pre_signed: bool) -> (Uuid, Uuid) { + let (instance_id, graph_id) = (Uuid::new_v4(), Uuid::new_v4()); + let mut storage = db.acquire().await.unwrap(); + storage + .upsert_graph_definition(&store::Graph { + graph_id, + instance_id, + status: GraphStatus::OperatorPresigned.to_string(), + definition_hash: "definition".to_string(), + kickoff_index: slot, + ..Default::default() + }) + .await + .unwrap(); + let raw = serde_json::json!({ "committee_pre_signed": committee_pre_signed }).to_string(); + storage + .upsert_graph_raw_data( + instance_id, + store::GraphRawData { graph_id, raw_data: raw, created_at: 0, updated_at: 0 }, + "definition", + ) + .await + .unwrap(); + (instance_id, graph_id) + } + + #[tokio::test] + async fn a_posted_graph_held_only_as_the_operator_copy_is_pulled_finalized() { + let (db, _dir) = db().await; + let mut slot = 0; + for actor in [Actor::Verifier, Actor::Watchtower, Actor::Operator, Actor::Committee] { + for committee_pre_signed in [false, true] { + slot += 1; + let (instance_id, graph_id) = stored(&db, slot, committee_pre_signed).await; + let mut tx = db.start_transaction().await.unwrap(); + handle_post_graph_data_events(&actor, &mut tx, vec![posted(instance_id, graph_id)]) + .await + .unwrap(); + tx.commit().await.unwrap(); + let mut storage = db.acquire().await.unwrap(); + let row = storage.find_graph(&graph_id).await.unwrap().unwrap(); + assert_eq!(row.status, GraphStatus::OperatorDataPushed.to_string()); + let pull = storage + .find_engine_inbox_event(&format!("graph:{graph_id}:pull:finalized")) + .await + .unwrap(); + assert_eq!(pull.is_some(), !committee_pre_signed, "{actor}"); + } + } + } } diff --git a/node/src/utils/graph_store.rs b/node/src/utils/graph_store.rs index 9fbbaebf..00ee0b61 100644 --- a/node/src/utils/graph_store.rs +++ b/node/src/utils/graph_store.rs @@ -427,6 +427,28 @@ async fn confirm_finalized_graph_definition( } } +/// Whether a raw definition carries committee pre-signatures, read without +/// parsing the graph. The store paths verify the signatures before they write +/// a definition with the flag set, so the flag is what "finalized" means. +pub fn raw_definition_committee_pre_signed(raw_data: &str) -> Result { + #[derive(serde::Deserialize)] + struct Flags { + committee_pre_signed: bool, + } + Ok(serde_json::from_str::(raw_data)?.committee_pre_signed) +} + +/// Whether the stored definition is finalized; `None` when none is stored. +pub async fn stored_definition_committee_pre_signed( + storage_processor: &mut StorageProcessor<'_>, + graph_id: &Uuid, +) -> Result> { + let Some(raw) = storage_processor.find_graph_raw_data(graph_id).await? else { + return Ok(None); + }; + raw_definition_committee_pre_signed(&raw.raw_data).map(Some) +} + /// Parse raw graph data JSON string to SimplifiedBitvmGcGraph using spawn_blocking /// to handle large data and potential stack overflow issues pub async fn parse_graph_raw_data(