From ce6b5877ccede50f2bf986156a6da39c8ca42fae Mon Sep 17 00:00:00 2001 From: Ethan Date: Tue, 10 Nov 2026 07:18:47 +0000 Subject: [PATCH] fix: give up stalled graphs at the abandon height on every role --- crates/store/src/engine/mod.rs | 17 ++++ node/src/aggregates/graph.rs | 64 ++++++++++++++- node/src/aggregates/graph_scan.rs | 77 ++++++++++++++----- node/src/aggregates/graph_tests.rs | 42 ++++++++++ .../graph_tests/setup_admission_tests.rs | 32 ++++++++ node/src/scheduled_tasks/mod.rs | 14 ++-- 6 files changed, 214 insertions(+), 32 deletions(-) diff --git a/crates/store/src/engine/mod.rs b/crates/store/src/engine/mod.rs index a8ab8cf92..eab98a975 100644 --- a/crates/store/src/engine/mod.rs +++ b/crates/store/src/engine/mod.rs @@ -828,6 +828,23 @@ impl<'a> StorageProcessor<'a> { Ok(query.fetch_all(self.conn()).await?) } + /// `aggregate_kind` aggregates whose mirrored status is `status`. + pub async fn list_engine_aggregates_by_status( + &mut self, + aggregate_kind: &str, + status: &str, + ) -> anyhow::Result> { + let rows = sqlx::query(&format!( + "SELECT {ENGINE_AGGREGATE_COLUMNS} FROM engine_aggregate \ + WHERE aggregate_kind = ? AND status = ?" + )) + .bind(aggregate_kind) + .bind(status) + .fetch_all(self.conn()) + .await?; + Ok(rows.iter().map(engine_aggregate_from_row).collect::>()?) + } + /// Create the row if needed and apply `update`; bumps `version`. pub async fn upsert_engine_aggregate( &mut self, diff --git a/node/src/aggregates/graph.rs b/node/src/aggregates/graph.rs index 898be72f5..e1f556f06 100644 --- a/node/src/aggregates/graph.rs +++ b/node/src/aggregates/graph.rs @@ -246,7 +246,7 @@ pub struct GraphState { /// The graph's Pre-Kickoff can never confirm (reported once). #[serde(default)] pub dead: bool, - /// The operator gave up its unfinished setup at the abandon height. + /// The setup was given up at the abandon height. #[serde(default)] pub abandoned: bool, /// Verifier: the authenticated InitGraph identity admitted to setup. @@ -263,7 +263,7 @@ pub struct SetupAdmission { } impl GraphState { - fn load(raw: Option<&str>) -> Self { + pub(super) fn load(raw: Option<&str>) -> Self { raw.and_then(|raw| serde_json::from_str(raw).ok()).unwrap_or_default() } @@ -944,7 +944,10 @@ impl GraphAggregate { row: Option<&store::Graph>, abandoned: &SetupAbandoned, ) -> anyhow::Result { - if self.actor != Actor::Operator || instance_failed(ctx, abandoned.instance_id).await? { + if self.actor != Actor::Operator { + return self.abandon_watched_setup(ctx, graph_id, state, row, abandoned).await; + } + if instance_failed(ctx, abandoned.instance_id).await? { return Ok(ApplyOutcome::Applied); } match row { @@ -992,6 +995,61 @@ impl GraphAggregate { Ok(ApplyOutcome::Applied) } + /// Other roles: a graph still without committee pre-signatures at the + /// abandon height turns Obsoleted (`finish` closes its deliveries and + /// jobs), and a verifier setup whose graph never arrived closes them here. + /// An Obsoleted graph is still probed for its Pre-Kickoff, which brings + /// it back. + async fn abandon_watched_setup( + &self, + ctx: &mut ApplyContext<'_>, + graph_id: Uuid, + state: &mut GraphState, + row: Option<&store::Graph>, + abandoned: &SetupAbandoned, + ) -> anyhow::Result { + match row { + Some(row) => { + if row.status != GraphStatus::OperatorPresigned.to_string() { + return Ok(ApplyOutcome::Applied); + } + ctx.storage + .transition_graph_status( + row.instance_id, + graph_id, + GraphStatus::Obsoleted, + GraphStatusSource::ChainReconcile, + None, + ) + .await?; + } + None => { + if state.abandoned || state.setup_admission.is_none() { + return Ok(ApplyOutcome::Applied); + } + ctx.effects.close_outbox( + EngineOutboxCloseTarget::Aggregate { + aggregate_kind: GRAPH_KIND.to_string(), + aggregate_id: graph_id.to_string(), + }, + "setup abandoned", + ); + ctx.effects.cancel_phase_jobs(JOB_PHASE); + } + } + state.abandoned = true; + warn!( + event = "graph_abandoned", + graph_id = %graph_id, + instance_id = %abandoned.instance_id, + built = row.is_some(), + height = abandoned.height, + abandon_height = abandoned.abandon_height, + "graph without committee pre-signatures at the abandon height; deliveries closed" + ); + Ok(ApplyOutcome::Applied) + } + /// Mirror graph status and close deliveries for completed or retired phases. async fn finish( &self, diff --git a/node/src/aggregates/graph_scan.rs b/node/src/aggregates/graph_scan.rs index f20135480..96792475d 100644 --- a/node/src/aggregates/graph_scan.rs +++ b/node/src/aggregates/graph_scan.rs @@ -2,11 +2,11 @@ //! //! Maintenance scans active graphs once per Bitcoin height and stored status; //! Obsoleted graphs only get a cheap probe until their Pre-Kickoff confirms. -//! Each result becomes a `chain_scan` event. Operator nodes also report setups +//! Each result becomes a `chain_scan` event. Every role also reports setups //! unfinished at their instance's abandon height. use super::common::ScanMarks; -use super::graph::{SetupAbandoned, setup_abandoned_event}; +use super::graph::{GRAPH_KIND, GraphState, SetupAbandoned, setup_abandoned_event}; use super::graph_pegout::{ChainScan, PegoutSignal, Unusable}; use super::instance::PeginDeadlines; use crate::clock::current_time_secs; @@ -16,6 +16,7 @@ use crate::utils::{ scan_graph_chain_state, }; use bitcoin::{OutPoint, Txid}; +use bitvm_lib::actors::Actor; use bitvm_lib::timelocks::{ connector_f_timelock_blocks, disprove_timelock_blocks, operator_ack_timelock_blocks, operator_commit_timelock_blocks, take1_timelock_blocks, take2_timelock_blocks, @@ -32,7 +33,7 @@ use serde::{Deserialize, Serialize}; use std::collections::{BTreeMap, HashMap}; use std::str::FromStr; use std::sync::{LazyLock, Mutex}; -use store::localdb::{GraphQuery, LocalDB}; +use store::localdb::{GraphQuery, LocalDB, StorageProcessor}; use store::{Graph, GraphBtcTxVoutMonitor, GraphStatus, InstanceBridgeInStatus, SerializableTxid}; use strum::{Display, EnumString}; use tracing::{debug, trace, warn}; @@ -250,30 +251,39 @@ async fn prekickoff_confirmed(btc_client: &BTCClient, row: &Graph) -> anyhow::Re Ok(btc_client.get_tx_status(&Txid::from(txid)).await?.confirmed) } -/// Report operator setups unfinished at the abandon height, including pending -/// sessions and built graphs without committee pre-signatures. Failed instances -/// are handled by their own cleanup. +/// Report setups unfinished at the abandon height. An operator reports its +/// pending sessions and built graphs without committee pre-signatures; failed +/// instances are handled by their own cleanup. Other roles report every graph +/// without committee pre-signatures and every admitted verifier setup whose +/// graph never arrived, failed instance or not: nothing else ends their +/// deliveries. pub async fn abandon_unfinished_setups( local_db: &LocalDB, btc_client: &BTCClient, engine: &EngineHandle, + actor: &Actor, ) -> anyhow::Result<()> { - let operator_pubkey = get_node_pubkey()?.to_string(); + let operator = *actor == Actor::Operator; let setups: BTreeMap = { let mut storage = local_db.acquire().await?; - let sessions = storage.list_pending_graph_inits_by_operator(&operator_pubkey).await?; - let graphs = storage - .get_operator_graphs( - GraphQuery::default() - .with_operator_pubkey(operator_pubkey) - .with_statuses(vec![GraphStatus::OperatorPresigned.to_string()]), - ) - .await?; - sessions - .into_iter() - .map(|session| (session.graph_id, session.instance_id)) - .chain(graphs.into_iter().map(|graph| (graph.graph_id, graph.instance_id))) - .collect() + if operator { + let operator_pubkey = get_node_pubkey()?.to_string(); + let sessions = storage.list_pending_graph_inits_by_operator(&operator_pubkey).await?; + let graphs = storage + .get_operator_graphs( + GraphQuery::default() + .with_operator_pubkey(operator_pubkey) + .with_statuses(vec![GraphStatus::OperatorPresigned.to_string()]), + ) + .await?; + sessions + .into_iter() + .map(|session| (session.graph_id, session.instance_id)) + .chain(graphs.into_iter().map(|graph| (graph.graph_id, graph.instance_id))) + .collect() + } else { + watched_setups(&mut storage).await? + } }; if setups.is_empty() { return Ok(()); @@ -284,7 +294,9 @@ pub async fn abandon_unfinished_setups( let Some(instance) = local_db.acquire().await?.find_instance(&instance_id).await? else { continue; }; - if InstanceBridgeInStatus::from_str(&instance.status).is_ok_and(|status| status.is_failed()) + if operator + && InstanceBridgeInStatus::from_str(&instance.status) + .is_ok_and(|status| status.is_failed()) { continue; } @@ -298,6 +310,29 @@ pub async fn abandon_unfinished_setups( Ok(()) } +/// `(graph, instance)` of what a non-operator still delivers for: graphs +/// without committee pre-signatures and admitted verifier setups whose graph +/// never arrived. +pub(super) async fn watched_setups( + storage: &mut StorageProcessor<'_>, +) -> anyhow::Result> { + let graphs = storage + .get_operator_graphs( + GraphQuery::default().with_statuses(vec![GraphStatus::OperatorPresigned.to_string()]), + ) + .await?; + let unbuilt = storage.list_engine_aggregates_by_status(GRAPH_KIND, "").await?; + Ok(graphs + .into_iter() + .map(|graph| (graph.graph_id, graph.instance_id)) + .chain(unbuilt.into_iter().filter_map(|aggregate| { + let state = GraphState::load(aggregate.state.as_deref()); + let admission = state.setup_admission.filter(|_| !state.abandoned)?; + Some((Uuid::parse_str(&aggregate.aggregate_id).ok()?, admission.instance_id)) + })) + .collect()) +} + /// The facts of one graph at `height`; `None` when its definition is not /// stored (nothing on chain can be decided without it). pub async fn gather_chain_scan( diff --git a/node/src/aggregates/graph_tests.rs b/node/src/aggregates/graph_tests.rs index 9090e8c5c..603ded384 100644 --- a/node/src/aggregates/graph_tests.rs +++ b/node/src/aggregates/graph_tests.rs @@ -1009,6 +1009,42 @@ async fn a_session_unfinished_at_the_abandon_height_is_released() { t.stop().await; } +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_watched_graph_unfinished_at_the_abandon_height_closes_its_deliveries() { + let t = start(Actor::Committee, false).await; + // The operator went offline mid-round: this member's nonces would be + // resent forever. Past the abandon height the graph turns Obsoleted, + // which closes them. + let instance_id = t.insert_instance(InstanceBridgeInStatus::Presigned).await; + let stalled = t.insert_graph(instance_id, 0, GraphStatus::OperatorPresigned).await; + t.script( + JOB_INGEST_NONCES, + GraphPlan { + messages: vec![planned(stalled, "NonceGeneration", PHASE_SIGNING, None)], + ..Default::default() + }, + ); + let nonce = t.submit(t.message("NonceGeneration", "peer-c", instance_id, stalled, None)).await; + t.wait_inbox(&nonce, EngineInboxState::Applied).await; + let sent = outbox_id(stalled, "NonceGeneration", None); + t.wait_outbox(&sent, EngineOutboxState::Open).await; + let finalized = t.insert_graph(instance_id, 1, GraphStatus::CommitteePresigned).await; + assert_eq!(t.watched().await, BTreeMap::from([(stalled, instance_id)])); + + // A finalized graph is left alone. + let abandoned = SetupAbandoned { instance_id, abandon_height: 166, height: 167 }; + for graph_id in [stalled, finalized] { + let event_id = t.submit(setup_abandoned_event(graph_id, &abandoned).unwrap()).await; + t.wait_inbox(&event_id, EngineInboxState::Applied).await; + } + t.wait_status(stalled, GraphStatus::Obsoleted).await; + t.wait_outbox(&sent, EngineOutboxState::Closed).await; + assert!(t.graph_state(stalled).await.abandoned); + assert_eq!(t.status(finalized).await, GraphStatus::CommitteePresigned); + assert!(t.watched().await.is_empty()); + t.stop().await; +} + // ------------------------------------------------------------------ peg-out const OTHER_OPERATOR: &str = "02other"; @@ -1051,6 +1087,12 @@ impl Harness { wait_for(|| async { (self.status(graph_id).await == status).then_some(()) }).await } + /// What the abandon scan of a non-operator would look at. + async fn watched(&self) -> BTreeMap { + let mut storage = self.db.acquire().await.unwrap(); + super::graph_scan::watched_setups(&mut storage).await.unwrap() + } + async fn graph_state(&self, graph_id: Uuid) -> GraphState { serde_json::from_str(self.aggregate(graph_id).await.state.as_deref().unwrap()).unwrap() } diff --git a/node/src/aggregates/graph_tests/setup_admission_tests.rs b/node/src/aggregates/graph_tests/setup_admission_tests.rs index 26ae833bd..e512e5904 100644 --- a/node/src/aggregates/graph_tests/setup_admission_tests.rs +++ b/node/src/aggregates/graph_tests/setup_admission_tests.rs @@ -155,3 +155,35 @@ async fn verifier_binds_the_admission_to_the_authenticated_init_graph() { assert_eq!(t.called(JOB_VERIFIER_SETUP).len(), 1); t.stop().await; } + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn verifier_closes_a_setup_whose_graph_never_arrives_at_the_abandon_height() { + let t = start(Actor::Verifier, false).await; + let (operator, peer) = (operator_key(7), PeerId::random()); + let instance_id = t.insert_instance(InstanceBridgeInStatus::UserBroadcastPeginPrepare).await; + let graph_id = Uuid::new_v4(); + + // The setup package goes out, then the operator goes offline before its + // graph: the package would be resent forever. + t.script( + JOB_VERIFIER_SETUP, + GraphPlan { + messages: vec![planned(graph_id, "GenCircuits", PHASE_SETUP, Some(&peer.to_string()))], + ..Default::default() + }, + ); + let event = t.submit(t.init_graph(&operator, peer, instance_id, graph_id, false).await).await; + t.wait_inbox(&event, EngineInboxState::Applied).await; + let sent = outbox_id(graph_id, "GenCircuits", None); + t.wait_outbox(&sent, EngineOutboxState::Open).await; + assert_eq!(t.watched().await, BTreeMap::from([(graph_id, instance_id)])); + + let abandoned = SetupAbandoned { instance_id, abandon_height: 166, height: 167 }; + let event_id = t.submit(setup_abandoned_event(graph_id, &abandoned).unwrap()).await; + t.wait_inbox(&event_id, EngineInboxState::Applied).await; + let closed = t.wait_outbox(&sent, EngineOutboxState::Closed).await; + assert_eq!(closed.close_reason.as_deref(), Some("setup abandoned")); + assert!(t.graph_state(graph_id).await.abandoned); + assert!(t.watched().await.is_empty(), "reported once"); + t.stop().await; +} diff --git a/node/src/scheduled_tasks/mod.rs b/node/src/scheduled_tasks/mod.rs index 1408dbb1b..e9edb6c1a 100644 --- a/node/src/scheduled_tasks/mod.rs +++ b/node/src/scheduled_tasks/mod.rs @@ -349,14 +349,12 @@ async fn run_bitcoin_scans( scan_instances_on_bitcoin(local_db, btc_client, engine, actor), ) .await; - if *actor == Actor::Operator { - failed |= !run_maintenance_subtask( - metrics_state, - "graph_setup_abandon_scan", - abandon_unfinished_setups(local_db, btc_client, engine), - ) - .await; - } + failed |= !run_maintenance_subtask( + metrics_state, + "graph_setup_abandon_scan", + abandon_unfinished_setups(local_db, btc_client, engine, actor), + ) + .await; let outcome = if failed { "failed" } else { "success" }; metrics_state.record_task_run("bitcoin_maintenance", outcome, started.elapsed());