Fix obsolete graph logic

crates/store/src/schema.rs

@@ -412,6 +412,14 @@ impl GraphStatus {
         ];
         const SKIPPED: &[GraphStatus] =
             &[OperatorPresigned, CommitteePresigned, OperatorDataPushed, PreKickoff, Obsoleted];
+        const OBSOLETED: &[GraphStatus] = &[
+            OperatorPresigned,
+            CommitteePresigned,
+            OperatorDataPushed,
+            PreKickoff,
+            OperatorKickOff,
+            Challenge,
+        ];
 
         match source {
             GraphStatusSource::Definition => match self {
@@ -426,11 +434,9 @@ impl GraphStatus {
             },
             GraphStatusSource::ChainReconcile => match self {
                 CommitteePresigned => &[OperatorPresigned],
-                // Obsoleted is a provisional branch when GraphData was not
-                // yet visible. A later full chain scan may correct it, but
-                // ordinary Goat event replay may not.
                 OperatorDataPushed => EARLY_OR_OBSOLETED,
-                PreKickoff | Obsoleted => PRE_KICKOFF,
+                Obsoleted => OBSOLETED,
+                PreKickoff => PRE_KICKOFF,
                 OperatorKickOff => KICKOFF,
                 Challenge => CHALLENGE,
                 OperatorTake1 => TAKE1,

node/src/action.rs

@@ -2,7 +2,7 @@
 #![allow(clippy::single_match)]
 #![allow(clippy::collapsible_else_if)]
 
-use crate::env::get_local_node_info;
+use crate::env::{get_local_node_info, get_p2p_inbox_batch_size, get_p2p_outbox_batch_size};
 use crate::handle::{
     HandlerContext, HeavyTaskContext, dispatch as handle_dispatch, heavy_task_from_content,
     is_heavy_task_message_type, run_heavy_task,
@@ -46,7 +46,6 @@ pub struct GOATMessage {
 
 const GOAT_MESSAGE_BIN_PREFIX: &[u8] = b"GOATBIN1";
 const TRANSIENT_PEGIN_RETRY_DELAY_SECS: usize = 30;
-const P2P_INBOX_BATCH_SIZE: i64 = 8;
 const P2P_INBOX_LEASE_SECS: i64 = 5 * 60;
 const P2P_INBOX_LEASE_RENEW_INTERVAL_SECS: u64 = 60;
 const P2P_INBOX_ENQUEUE_ATTEMPTS: usize = 3;
@@ -940,7 +939,7 @@ async fn handle_p2p_inbox_messages(
         .claim_p2p_inbox_messages(
             now,
             now + P2P_INBOX_LEASE_SECS,
-            P2P_INBOX_BATCH_SIZE,
+            get_p2p_inbox_batch_size(),
             &active_heavy_task_ids,
         )
         .await?;
@@ -1277,7 +1276,7 @@ async fn handle_p2p_outbox_messages(
     let now = current_time_secs();
     let mut storage = local_db.start_immediate_transaction().await?;
     let messages = storage
-        .claim_p2p_outbox_messages(now, now + P2P_INBOX_LEASE_SECS, P2P_INBOX_BATCH_SIZE)
+        .claim_p2p_outbox_messages(now, now + P2P_INBOX_LEASE_SECS, get_p2p_outbox_batch_size())
         .await?;
     storage.commit().await?;
 

node/src/bin/send_pegout.rs

@@ -51,6 +51,10 @@ enum Commands {
         /// Dry run (validate but do not call Gateway.initWithdraw)
         #[arg(long, default_value_t = false)]
         dry_run: bool,
+
+        /// Skip graphs whose instances are locked by another withdrawal
+        #[arg(long, default_value_t = false)]
+        skip_locked: bool,
     },
     /// Initiate pegouts repeatedly until target amount reached
     Batch {
@@ -66,6 +70,10 @@ enum Commands {
         #[arg(long, default_value_t = false)]
         dry_run: bool,
 
+        /// Skip graphs whose instances are locked by another withdrawal
+        #[arg(long, default_value_t = false)]
+        skip_locked: bool,
+
         /// Poll interval seconds for checking graph readiness between pegouts
         #[arg(long, default_value_t = 300)]
         poll_interval_secs: u64,
@@ -81,6 +89,7 @@ struct PegoutApiRequest {
     #[serde(skip_serializing_if = "Option::is_none")]
     graph_id: Option<String>,
     dry_run: bool,
+    skip_locked: bool,
 }
 
 #[derive(Debug, Deserialize)]
@@ -114,9 +123,11 @@ async fn call_pegout(
     base_url: &str,
     graph_id: Option<Uuid>,
     dry_run: bool,
+    skip_locked: bool,
 ) -> Result<PegoutApiResponse> {
     let url = format!("{}/v1/graphs/pegout", base_url.trim_end_matches('/'));
-    let body = PegoutApiRequest { graph_id: graph_id.map(|id| id.to_string()), dry_run };
+    let body =
+        PegoutApiRequest { graph_id: graph_id.map(|id| id.to_string()), dry_run, skip_locked };
 
     let keypair = get_bitvm_key().context("failed to load BITVM_SECRET")?;
     let (timestamp, signature) = sign_request_auth(&keypair);
@@ -208,14 +219,15 @@ async fn main() -> Result<()> {
     let base_url = &args.rpc_url;
 
     match args.command {
-        Commands::Once { graph_id, dry_run } => {
-            let resp = call_pegout(&client, base_url, graph_id, dry_run).await?;
+        Commands::Once { graph_id, dry_run, skip_locked } => {
+            let resp = call_pegout(&client, base_url, graph_id, dry_run, skip_locked).await?;
             print_pegout_result(&resp);
         }
         Commands::Batch {
             max_total_amount_sats,
             max_count,
             dry_run,
+            skip_locked,
             poll_interval_secs,
             max_wait_secs,
         } => {
@@ -236,7 +248,7 @@ async fn main() -> Result<()> {
                     break;
                 }
 
-                match call_pegout(&client, base_url, None, true).await {
+                match call_pegout(&client, base_url, None, true, skip_locked).await {
                     Ok(preflight) => {
                         let amount = u64::try_from(preflight.amount)
                             .context("pegout API returned a negative amount")?;
@@ -257,7 +269,9 @@ async fn main() -> Result<()> {
 
                         let graph_id = Uuid::parse_str(&preflight.graph_id)
                             .context("pegout API returned an invalid graph id")?;
-                        let resp = call_pegout(&client, base_url, Some(graph_id), false).await?;
+                        let resp =
+                            call_pegout(&client, base_url, Some(graph_id), false, skip_locked)
+                                .await?;
                         print_pegout_result(&resp);
                         total_sent = total_sent.saturating_add(amount);
                         count = count.saturating_add(1);

node/src/env.rs

@@ -142,6 +142,10 @@ pub const ENV_INSTANCE_MAINTENANCE_BATCH_SIZE: &str = "INSTANCE_MAINTENANCE_BATC
 pub const DEFAULT_INSTANCE_MAINTENANCE_BATCH_SIZE: u32 = 50;
 pub const ENV_MAINTENANCE_RUN_TIMEOUT_SECS: &str = "MAINTENANCE_RUN_TIMEOUT_SECS";
 pub const DEFAULT_MAINTENANCE_RUN_TIMEOUT_SECS: u64 = 60;
+pub const ENV_P2P_INBOX_BATCH_SIZE: &str = "P2P_INBOX_BATCH_SIZE";
+pub const DEFAULT_P2P_INBOX_BATCH_SIZE: i64 = 16;
+pub const ENV_P2P_OUTBOX_BATCH_SIZE: &str = "P2P_OUTBOX_BATCH_SIZE";
+pub const DEFAULT_P2P_OUTBOX_BATCH_SIZE: i64 = 16;
 pub const ENV_ENABLE_COMMITTEE_INSTANCE_KEY_DELETE: &str = "ENABLE_COMMITTEE_INSTANCE_KEY_DELETE";
 pub const DEFAULT_ENABLE_COMMITTEE_INSTANCE_KEY_DELETE: bool = false;
 pub const ENV_COMMITTEE_INSTANCE_KEY_DELETE_TIMELOCK_BLOCKS: &str =
@@ -652,6 +656,22 @@ pub fn get_maintenance_run_timeout_secs() -> u64 {
         .unwrap_or(DEFAULT_MAINTENANCE_RUN_TIMEOUT_SECS)
 }
 
+pub fn get_p2p_inbox_batch_size() -> i64 {
+    std::env::var(ENV_P2P_INBOX_BATCH_SIZE)
+        .ok()
+        .and_then(|value| value.parse::<i64>().ok())
+        .map(|size| size.clamp(1, 128))
+        .unwrap_or(DEFAULT_P2P_INBOX_BATCH_SIZE)
+}
+
+pub fn get_p2p_outbox_batch_size() -> i64 {
+    std::env::var(ENV_P2P_OUTBOX_BATCH_SIZE)
+        .ok()
+        .and_then(|value| value.parse::<i64>().ok())
+        .map(|size| size.clamp(1, 128))
+        .unwrap_or(DEFAULT_P2P_OUTBOX_BATCH_SIZE)
+}
+
 pub fn is_enable_committee_instance_key_delete() -> bool {
     // TODO: enable this feature when ready
     tracing::warn!(

node/src/rpc_service/bitvm.rs

@@ -111,6 +111,9 @@ pub struct PegoutRequest {
     pub graph_id: Option<String>,
     #[serde(default)]
     pub dry_run: bool,
+    /// Skip graphs whose instances are temporarily locked by another withdrawal.
+    #[serde(default)]
+    pub skip_locked: bool,
 }
 
 #[derive(Debug, Deserialize, Serialize)]