fix: give up stalled graphs at the abandon height on every role

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<Vec<EngineAggregate>> {
+        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::<Result<_, _>>()?)
+    }
+
     /// Create the row if needed and apply `update`; bumps `version`.
     pub async fn upsert_engine_aggregate(
         &mut self,

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<ApplyOutcome> {
-        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<ApplyOutcome> {
+        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,

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<Uuid, Uuid> = {
         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<BTreeMap<Uuid, Uuid>> {
+    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(

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<Uuid, Uuid> {
+        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()
     }

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;
+}

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());