remove rgb_send_lock: fail colored funding-prep instead of replaying the event (#67)

Author: dcorral

src/error.rs

@@ -313,9 +313,6 @@ pub enum APIError {
     #[error("No valid transport endpoint found")]
     NoValidTransportEndpoint,
 
-    #[error("Cannot perform this operation while an open channel operation is in progress")]
-    OpenChannelInProgress,
-
     #[error("Output below the dust limit")]
     OutputBelowDustLimit,
 
@@ -588,7 +585,6 @@ impl IntoResponse for APIError {
             | APIError::NoAvailableUtxos
             | APIError::NoRoute
             | APIError::NotInitialized
-            | APIError::OpenChannelInProgress
             | APIError::PaymentNotFound(_)
             | APIError::RecipientIDAlreadyUsed
             | APIError::SwapNotFound(_)

src/ldk.rs

@@ -1428,7 +1428,6 @@ fn normalize_funding_psbt_locktime(
 // FundingGenerationReady. Returns the value to propagate from the event handler: `Err(ReplayEvent)`
 // to retry the event (for transient network errors), or `Ok(())` after force-closing the channel
 // (for terminal errors).
-#[allow(dead_code)]
 fn handle_funding_prepare_err(
     e: RgbLibError,
     channel_manager: &ChannelManager,
@@ -1556,7 +1555,6 @@ async fn handle_ldk_events(
                         false,
                     );
                     unlocked_state.virtual_channel_draft_delete(&temporary_channel_id);
-                    *unlocked_state.rgb_send_lock.lock().unwrap() = false;
                 };
 
                 let mut virtual_funding_txo = virtual_channel_synthetic_outpoint(
@@ -1757,7 +1755,6 @@ async fn handle_ldk_events(
                             updated_at: get_current_timestamp(),
                         });
                         unlocked_state.virtual_channel_draft_delete(&temporary_channel_id);
-                        *unlocked_state.rgb_send_lock.lock().unwrap() = false;
                         tracing::info!(
                             "EVENT: registered trusted no-broadcast funding {} for virtual channel {}",
                             virtual_funding_txo,
@@ -1777,7 +1774,6 @@ async fn handle_ldk_events(
                             false,
                         );
                         unlocked_state.virtual_channel_draft_delete(&temporary_channel_id);
-                        *unlocked_state.rgb_send_lock.lock().unwrap() = false;
                     }
                 }
                 return Ok(());
@@ -1814,39 +1810,40 @@ async fn handle_ldk_events(
 
                 let unlocked_state_copy = unlocked_state.clone();
                 let res = tokio::task::spawn_blocking(
-                    move || -> Result<(String, Option<i32>), String> {
-                        let res = unlocked_state_copy
-                            .rgb_send_begin(
-                                recipient_map,
-                                true,
-                                FEE_RATE,
-                                MIN_CHANNEL_CONFIRMATIONS,
-                                None,
-                                false,
-                                // Final locktime: this colored tx funds an LN channel.
-                                Some(0),
-                            )
-                            .map_err(|e| e.to_string())?;
-                        let fascia_str = fs::read_to_string(&res.details.fascia_path)
-                            .map_err(|e| e.to_string())?;
-                        let fascia: Fascia =
-                            serde_json::from_str(&fascia_str).map_err(|e| e.to_string())?;
-                        unlocked_state_copy
-                            .rgb_consume_fascia(fascia, None)
-                            .map_err(|e| e.to_string())?;
-                        unlocked_state_copy
-                            .rgb_create_consignments(res.psbt.clone())
-                            .map_err(|e| e.to_string())?;
+                    move || -> Result<(String, Option<i32>), RgbLibError> {
+                        let res = unlocked_state_copy.rgb_send_begin(
+                            recipient_map,
+                            true,
+                            FEE_RATE,
+                            MIN_CHANNEL_CONFIRMATIONS,
+                            None,
+                            false,
+                            // Final locktime: this colored tx funds an LN channel.
+                            Some(0),
+                        )?;
+                        let fascia_str = fs::read_to_string(&res.details.fascia_path).unwrap();
+                        let fascia: Fascia = serde_json::from_str(&fascia_str).unwrap();
+                        unlocked_state_copy.rgb_consume_fascia(fascia, None)?;
+                        unlocked_state_copy.rgb_create_consignments(res.psbt.clone())?;
                         Ok((res.psbt, res.batch_transfer_idx))
                     },
                 )
                 .await
                 .unwrap();
                 let (unsigned_psbt, batch_transfer_idx) = match res {
                     Ok(result) => result,
+                    // A failed funding preparation (e.g. the asset allocation is
+                    // momentarily reserved by a concurrent open) must fail the
+                    // channel so the caller can retry, not retry the event
+                    // forever. handle_open_chan_fail (on ChannelClosed) then
+                    // releases any reserved allocation.
                     Err(e) => {
-                        tracing::error!("cannot prepare channel funding transfer: {e}");
-                        return Err(ReplayEvent());
+                        return handle_funding_prepare_err(
+                            e,
+                            &unlocked_state.channel_manager,
+                            &temporary_channel_id,
+                            &counterparty_node_id,
+                        );
                     }
                 };
                 // Record the batch transfer index on the channel's RGB info so a failed
@@ -1866,9 +1863,25 @@ async fn handle_ldk_events(
                 }
                 (unsigned_psbt, Some(asset_id))
             } else {
-                let raw_psbt = unlocked_state
-                    .rgb_send_btc_begin(addr.to_address(), channel_value_satoshis, FEE_RATE)
-                    .unwrap();
+                // Mirror the colored path: a failed funding preparation must fail
+                // the channel (so the caller can retry) rather than panic the event
+                // task. handle_funding_prepare_err force-closes on terminal errors
+                // and replays the event on transient network errors.
+                let raw_psbt = match unlocked_state.rgb_send_btc_begin(
+                    addr.to_address(),
+                    channel_value_satoshis,
+                    FEE_RATE,
+                ) {
+                    Ok(psbt) => psbt,
+                    Err(e) => {
+                        return handle_funding_prepare_err(
+                            e,
+                            &unlocked_state.channel_manager,
+                            &temporary_channel_id,
+                            &counterparty_node_id,
+                        );
+                    }
+                };
                 let current_best_height =
                     unlocked_state.channel_manager.current_best_block().height;
                 let unsigned_psbt =
@@ -1985,7 +1998,6 @@ async fn handle_ldk_events(
                 tracing::error!(
                     "ERROR: Channel went away before we could fund it. The peer disconnected or refused the channel.",
                 );
-                *unlocked_state.rgb_send_lock.lock().unwrap() = false;
             }
         }
         Event::FundingTxBroadcastSafe { .. } => {
@@ -2577,7 +2589,6 @@ async fn handle_ldk_events(
                 .virtual_channel_session_store()
                 .contains_key(&channel_id)
             {
-                *unlocked_state.rgb_send_lock.lock().unwrap() = false;
                 tracing::info!(
                     "EVENT: virtual channel {} is pending in trusted no-broadcast mode",
                     channel_id,
@@ -2611,8 +2622,6 @@ async fn handle_ldk_events(
                     })
                     .await;
 
-                    *unlocked_state.rgb_send_lock.lock().unwrap() = false;
-
                     let finalize_result = join_result.map_err(|join_err| {
                         tracing::error!("Channel opening finalization task failed: {join_err:?}");
                         ReplayEvent()
@@ -2696,8 +2705,6 @@ async fn handle_ldk_events(
                 reason
             );
 
-            *unlocked_state.rgb_send_lock.lock().unwrap() = false;
-
             // Release any funds locked for a funding tx that was never broadcast.
             handle_open_chan_fail(&channel_id, unlocked_state.clone()).await;
 
@@ -2729,7 +2736,6 @@ async fn handle_ldk_events(
                     &format!("virtual_channel_{}", channel_id),
                     false,
                 );
-                *unlocked_state.rgb_send_lock.lock().unwrap() = false;
 
                 tracing::warn!(
                     "EVENT: cleaned up failed virtual open draft {} after channel close {}",
@@ -4706,7 +4712,6 @@ pub(crate) async fn start_ldk(
         taker_swaps,
         router: Arc::clone(&router),
         output_sweeper: Arc::clone(&output_sweeper),
-        rgb_send_lock: Arc::new(Mutex::new(false)),
         channel_ids_map,
         proxy_endpoint: proxy_endpoint.to_string(),
         external_signer_mode,

src/routes.rs

@@ -3991,10 +3991,6 @@ pub(crate) async fn open_channel(
         let guard = state.check_unlocked().await?;
         let unlocked_state = guard.as_ref().unwrap();
 
-        if *unlocked_state.rgb_send_lock.lock().unwrap() {
-            return Err(APIError::OpenChannelInProgress);
-        }
-
         let is_virtual_open = match payload.virtual_open_mode.as_deref() {
             None => false,
             Some(VIRTUAL_OPEN_MODE_TRUSTED_NO_BROADCAST) => true,
@@ -4286,14 +4282,6 @@ pub(crate) async fn open_channel(
             (temporary_channel_id, None)
         };
 
-        // Only colored opens perform an RGB send during funding, so only they need
-        // the RGB send lock. Vanilla opens that stall (e.g. an unresponsive peer)
-        // must not hold it, or they would block all subsequent opens indefinitely.
-        if colored_info.is_some() {
-            *unlocked_state.rgb_send_lock.lock().unwrap() = true;
-            tracing::debug!("RGB send lock set to true");
-        }
-
         let temporary_channel_id = unlocked_state
             .channel_manager
             .create_channel(
@@ -4307,8 +4295,6 @@ pub(crate) async fn open_channel(
                 payload.push_asset_amount,
             )
             .map_err(|e| {
-                *unlocked_state.rgb_send_lock.lock().unwrap() = false;
-                tracing::debug!("RGB send lock set to false (open channel failure: {e:?})");
                 if let Some(temp_id_str) = rgb_metadata_temp_id_str.as_deref() {
                     let _ = unlocked_state
                         .kv_store

src/sdk/mod.rs

@@ -1647,10 +1647,6 @@ pub(crate) async fn send_rgb(
     let guard = check_unlocked(&state).await?;
     let unlocked_state = guard.as_ref().unwrap();
 
-    if *unlocked_state.rgb_send_lock.lock().unwrap() {
-        return Err(APIError::OpenChannelInProgress);
-    }
-
     let send_result = if unlocked_state.external_signer_mode {
         let unlocked_state_copy = unlocked_state.clone();
         let begin_result = tokio::task::spawn_blocking(move || {
@@ -2364,10 +2360,6 @@ pub(crate) async fn issue_asset_nia(
         ));
     }
 
-    if *unlocked_state.rgb_send_lock.lock().unwrap() {
-        return Err(APIError::OpenChannelInProgress);
-    }
-
     let asset = unlocked_state.rgb_issue_asset_nia(
         request.ticker,
         request.name,
@@ -2390,10 +2382,6 @@ pub(crate) async fn issue_asset_cfa(
         ));
     }
 
-    if *unlocked_state.rgb_send_lock.lock().unwrap() {
-        return Err(APIError::OpenChannelInProgress);
-    }
-
     let file_path = request.file_digest.map(|d| {
         unlocked_state
             .rgb_get_media_dir()
@@ -2425,10 +2413,6 @@ pub(crate) async fn issue_asset_ifa(
         ));
     }
 
-    if *unlocked_state.rgb_send_lock.lock().unwrap() {
-        return Err(APIError::OpenChannelInProgress);
-    }
-
     let asset = unlocked_state.rgb_issue_asset_ifa(
         request.ticker,
         request.name,
@@ -2453,10 +2437,6 @@ pub(crate) async fn issue_asset_uda(
         ));
     }
 
-    if *unlocked_state.rgb_send_lock.lock().unwrap() {
-        return Err(APIError::OpenChannelInProgress);
-    }
-
     let rgb_media_dir = unlocked_state.rgb_get_media_dir();
     let get_string_path = |d: String| {
         rgb_media_dir
@@ -2654,10 +2634,6 @@ pub(crate) async fn rgb_invoice(
     let guard = check_unlocked(&state).await?;
     let unlocked_state = guard.as_ref().unwrap();
 
-    if *unlocked_state.rgb_send_lock.lock().unwrap() {
-        return Err(APIError::OpenChannelInProgress);
-    }
-
     let assignment = match request.assignment_kind {
         Some(kind) => rgb_assignment_from_kind(kind, request.assignment_amount)?,
         None => RgbLibAssignment::Any,
@@ -2699,10 +2675,6 @@ pub(crate) async fn open_channel(
     let guard = check_unlocked(&state).await?;
     let unlocked_state = guard.as_ref().unwrap();
 
-    if *unlocked_state.rgb_send_lock.lock().unwrap() {
-        return Err(APIError::OpenChannelInProgress);
-    }
-
     let is_virtual_open = match request.virtual_open_mode.as_deref() {
         None => false,
         Some(SDK_VIRTUAL_OPEN_MODE_TRUSTED_NO_BROADCAST) => true,
@@ -2973,13 +2945,6 @@ pub(crate) async fn open_channel(
             (temporary_channel_id, None)
         };
 
-    // Only colored opens perform an RGB send during funding, so only they need the
-    // RGB send lock. Vanilla opens that stall must not hold it (see routes.rs).
-    if colored_info.is_some() {
-        *unlocked_state.rgb_send_lock.lock().unwrap() = true;
-        tracing::debug!("RGB send lock set to true");
-    }
-
     let temporary_channel_id = unlocked_state
         .channel_manager
         .create_channel(
@@ -2993,8 +2958,6 @@ pub(crate) async fn open_channel(
             request.push_asset_amount,
         )
         .map_err(|e| {
-            *unlocked_state.rgb_send_lock.lock().unwrap() = false;
-            tracing::debug!("RGB send lock set to false (open channel failure: {e:?})");
             if let Some(temp_id_str) = rgb_metadata_temp_id_str.as_deref() {
                 let _ = unlocked_state
                     .kv_store
@@ -4058,10 +4021,6 @@ pub(crate) async fn inflate(
         ));
     }
 
-    if *unlocked_state.rgb_send_lock.lock().unwrap() {
-        return Err(APIError::OpenChannelInProgress);
-    }
-
     let unlocked_state_copy = unlocked_state.clone();
     let inflate_result = tokio::task::spawn_blocking(move || {
         unlocked_state_copy.rgb_inflate(

src/test/mod.rs

@@ -1709,7 +1709,13 @@ async fn open_channel_raw(
             }
         }
         if (OffsetDateTime::now_utc() - t_0).as_seconds_f32() > 50.0 {
-            panic!("cannot find funding TX")
+            // The channel may have been force-closed before reaching
+            // ChannelPending (e.g. a colored funding preparation failed because
+            // the asset allocation was momentarily reserved by a concurrent
+            // open). Surface a retryable error so open_channel_with_retry tries
+            // again once the contending open has settled.
+            println!("cannot find funding TX for channel to {dest_peer_pubkey}");
+            return Err(reqwest::StatusCode::FORBIDDEN);
         }
     }
     let channel_id = channel_id.unwrap();

src/test/openchannel_fail.rs

@@ -580,34 +580,7 @@ async fn openchannel_fail() {
         .await
         .unwrap();
     assert!(res.status() == reqwest::StatusCode::OK);
-    // open a 2nd channel while the previous open is still in progess (fail)
-    let payload = OpenChannelRequest {
-        peer_pubkey_and_opt_addr: format!("{node2_pubkey}@127.0.0.1:{NODE2_PEER_PORT}"),
-        capacity_sat: 100_000,
-        push_msat: 3_500_000,
-        asset_amount: Some(100),
-        asset_id: Some(asset_id),
-        push_asset_amount: None,
-        public: true,
-        with_anchors: true,
-        fee_base_msat: None,
-        fee_proportional_millionths: None,
-        temporary_channel_id: None,
-        virtual_open_mode: None,
-    };
-    let res = reqwest::Client::new()
-        .post(format!("http://{node1_addr}/openchannel"))
-        .json(&payload)
-        .send()
-        .await
-        .unwrap();
-    check_response_is_nok(
-        res,
-        reqwest::StatusCode::FORBIDDEN,
-        "Cannot perform this operation while an open channel operation is in progress",
-        "OpenChannelInProgress",
-    )
-    .await;
+    let _ = asset_id;
 
     let t_0 = OffsetDateTime::now_utc();
     loop {