feat: Recover interrupted dlc protocol on restart

CHANGELOG.md

@@ -16,6 +16,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Feat: allow force-close a DLC channel
 - Feat: made sure that rollover works with dlc-channels
 - Fix: correctly remember reserved utxos and don't accidentally double spend
+- Feat: Allow recovering from a stuck protocol state by resending last outbound dlc message on connect
 
 ## [1.7.4] - 2023-12-20
 

Cargo.toml

@@ -22,11 +22,11 @@ resolver = "2"
 # `p2pderivatives/rust-dlc#feature/ln-dlc-channels`: 4e104b4. This patch ensures backwards
 # compatibility for 10101 through the `rust-lightning:0.0.116` upgrade. We will be able to drop it
 # once all users have been upgraded and traded once.
-dlc-manager = { git = "https://github.com/p2pderivatives/rust-dlc", rev = "5373146" }
-dlc-messages = { git = "https://github.com/p2pderivatives/rust-dlc", rev = "5373146" }
-dlc = { git = "https://github.com/p2pderivatives/rust-dlc", rev = "5373146" }
-p2pd-oracle-client = { git = "https://github.com/p2pderivatives/rust-dlc", rev = "5373146" }
-dlc-trie = { git = "https://github.com/p2pderivatives/rust-dlc", rev = "5373146" }
+dlc-manager = { git = "https://github.com/p2pderivatives/rust-dlc", rev = "9735996802f64c0776a6969fa8ef24fba43f8630" }
+dlc-messages = { git = "https://github.com/p2pderivatives/rust-dlc", rev = "9735996802f64c0776a6969fa8ef24fba43f8630" }
+dlc = { git = "https://github.com/p2pderivatives/rust-dlc", rev = "9735996802f64c0776a6969fa8ef24fba43f8630" }
+p2pd-oracle-client = { git = "https://github.com/p2pderivatives/rust-dlc", rev = "9735996802f64c0776a6969fa8ef24fba43f8630" }
+dlc-trie = { git = "https://github.com/p2pderivatives/rust-dlc", rev = "9735996802f64c0776a6969fa8ef24fba43f8630" }
 
 # We should usually track the `p2pderivatives/split-tx-experiment[-10101]` branch.
 lightning = { git = "https://github.com/p2pderivatives/rust-lightning/", rev = "121bc324" }

coordinator/migrations/2024-01-10-105231_add_dlc_messages/down.sql

@@ -0,0 +1,6 @@
+-- This file should undo anything in `up.sql`
+DROP TABLE "last_outbound_dlc_messages";
+DROP TABLE "dlc_messages";
+
+DROP TYPE "Message_Type_Type";
+

coordinator/migrations/2024-01-10-105231_add_dlc_messages/up.sql

@@ -0,0 +1,32 @@
+CREATE TYPE "Message_Type_Type" AS ENUM (
+    'Offer',
+    'Accept',
+    'Sign',
+    'SettleOffer',
+    'SettleAccept',
+    'SettleConfirm',
+    'SettleFinalize',
+    'RenewOffer',
+    'RenewAccept',
+    'RenewConfirm',
+    'RenewFinalize',
+    'RenewRevoke',
+    'CollaborativeCloseOffer',
+    'Reject'
+);
+
+CREATE TABLE "dlc_messages" (
+    message_hash TEXT PRIMARY KEY NOT NULL,
+    inbound BOOLEAN NOT NULL,
+    peer_id TEXT NOT NULL,
+    message_type "Message_Type_Type" NOT NULL,
+    timestamp TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+CREATE TABLE "last_outbound_dlc_messages" (
+    peer_id TEXT PRIMARY KEY NOT NULL,
+    message_hash TEXT REFERENCES dlc_messages(message_hash) NOT NULL,
+    message TEXT NOT NULL,
+    timestamp TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+

coordinator/src/admin.rs

@@ -15,8 +15,8 @@ use bitcoin::secp256k1::PublicKey;
 use bitcoin::OutPoint;
 use commons::CollaborativeRevertCoordinatorExpertRequest;
 use commons::CollaborativeRevertCoordinatorRequest;
+use dlc_manager::channel::Channel;
 use dlc_manager::contract::Contract;
-use dlc_manager::subchannel::SubChannel;
 use lightning_invoice::Bolt11Invoice;
 use ln_dlc_node::node::NodeInfo;
 use serde::de;
@@ -246,10 +246,10 @@ pub struct DlcChannelDetails {
     pub user_registration_timestamp: Option<OffsetDateTime>,
 }
 
-impl From<(SubChannel, Option<Contract>, String, Option<OffsetDateTime>)> for DlcChannelDetails {
+impl From<(Channel, Option<Contract>, String, Option<OffsetDateTime>)> for DlcChannelDetails {
     fn from(
         (channel_details, contract, user_email, user_registration_timestamp): (
-            SubChannel,
+            Channel,
             Option<Contract>,
             String,
             Option<OffsetDateTime>,
@@ -273,36 +273,31 @@ pub async fn list_dlc_channels(
             AppError::InternalServerError(format!("Failed to acquire db lock: {e:#}"))
         })?;
 
-    let dlc_channels = state.node.inner.list_sub_channels().map_err(|e| {
+    let dlc_channels = state.node.inner.list_dlc_channels().map_err(|e| {
         AppError::InternalServerError(format!("Failed to list DLC channels: {e:#}"))
     })?;
 
     let dlc_channels = dlc_channels
         .into_iter()
-        .map(|subchannel| {
+        .map(|dlc_channel| {
             let (email, registration_timestamp) =
-                match db::user::by_id(&mut conn, subchannel.counter_party.to_string()) {
+                match db::user::by_id(&mut conn, dlc_channel.get_counter_party_id().to_string()) {
                     Ok(Some(user)) => (user.email, Some(user.timestamp)),
                     _ => ("unknown".to_string(), None),
                 };
 
-            let dlc_channel_id = subchannel.get_dlc_channel_id(0);
-
-            let contract = match dlc_channel_id {
-                Some(dlc_channel_id) => {
-                    match state
-                        .node
-                        .inner
-                        .get_contract_by_dlc_channel_id(&dlc_channel_id)
-                    {
-                        Ok(contract) => Some(contract),
-                        Err(_) => None,
-                    }
-                }
-                None => None,
+            let dlc_channel_id = dlc_channel.get_id();
+
+            let contract = match state
+                .node
+                .inner
+                .get_contract_by_dlc_channel_id(&dlc_channel_id)
+            {
+                Ok(contract) => Some(contract),
+                Err(_) => None,
             };
 
-            DlcChannelDetails::from((subchannel, contract, email, registration_timestamp))
+            DlcChannelDetails::from((dlc_channel, contract, email, registration_timestamp))
         })
         .collect::<Vec<_>>();
 

coordinator/src/bin/coordinator.rs

@@ -3,6 +3,8 @@ use anyhow::Result;
 use bitcoin::XOnlyPublicKey;
 use coordinator::backup::SledBackup;
 use coordinator::cli::Opts;
+use coordinator::dlc_handler;
+use coordinator::dlc_handler::DlcHandler;
 use coordinator::logger;
 use coordinator::message::spawn_delivering_messages_to_authenticated_users;
 use coordinator::message::NewUserMessage;
@@ -28,6 +30,7 @@ use diesel::r2d2;
 use diesel::r2d2::ConnectionManager;
 use diesel::PgConnection;
 use lightning::events::Event;
+use ln_dlc_node::node::event::NodeEventHandler;
 use ln_dlc_node::scorer;
 use ln_dlc_node::seed::Bip39Seed;
 use ln_dlc_node::CoordinatorEventHandler;
@@ -110,6 +113,7 @@ async fn main() -> Result<()> {
 
     let node_storage = Arc::new(NodeStorage::new(pool.clone()));
 
+    let node_event_handler = Arc::new(NodeEventHandler::new());
     let node = Arc::new(ln_dlc_node::node::Node::new(
         ln_dlc_node::config::coordinator_config(),
         scorer::persistent_scorer,
@@ -135,8 +139,13 @@ async fn main() -> Result<()> {
             .map(|o| o.into())
             .collect(),
         XOnlyPublicKey::from_str(&opts.oracle_pubkey).expect("valid public key"),
+        node_event_handler.clone(),
     )?);
 
+    let dlc_handler = DlcHandler::new(pool.clone(), node.clone());
+    let _handle =
+        dlc_handler::spawn_handling_dlc_messages(dlc_handler, node_event_handler.subscribe());
+
     let event_handler = CoordinatorEventHandler::new(node.clone(), Some(node_event_sender));
     let running = node.start(event_handler, false)?;
     let node = Node::new(node, running, pool.clone(), settings.to_node_settings());

coordinator/src/db/custom_types.rs

@@ -1,4 +1,5 @@
 use crate::db::channels::ChannelState;
+use crate::db::dlc_messages::MessageType;
 use crate::db::payments::HtlcStatus;
 use crate::db::payments::PaymentFlow;
 use crate::db::positions::ContractSymbol;
@@ -7,6 +8,7 @@ use crate::schema::sql_types::ChannelStateType;
 use crate::schema::sql_types::ContractSymbolType;
 use crate::schema::sql_types::DirectionType;
 use crate::schema::sql_types::HtlcStatusType;
+use crate::schema::sql_types::MessageTypeType;
 use crate::schema::sql_types::PaymentFlowType;
 use crate::schema::sql_types::PositionStateType;
 use diesel::deserialize;
@@ -159,3 +161,47 @@ impl FromSql<DirectionType, Pg> for Direction {
         }
     }
 }
+
+impl ToSql<MessageTypeType, Pg> for MessageType {
+    fn to_sql<'b>(&'b self, out: &mut Output<'b, '_, Pg>) -> serialize::Result {
+        match *self {
+            MessageType::Offer => out.write_all(b"Offer")?,
+            MessageType::Accept => out.write_all(b"Accept")?,
+            MessageType::Sign => out.write_all(b"Sign")?,
+            MessageType::SettleOffer => out.write_all(b"SettleOffer")?,
+            MessageType::SettleAccept => out.write_all(b"SettleAccept")?,
+            MessageType::SettleConfirm => out.write_all(b"SettleConfirm")?,
+            MessageType::SettleFinalize => out.write_all(b"SettleFinalize")?,
+            MessageType::RenewOffer => out.write_all(b"RenewOffer")?,
+            MessageType::RenewAccept => out.write_all(b"RenewAccept")?,
+            MessageType::RenewConfirm => out.write_all(b"RenewConfirm")?,
+            MessageType::RenewFinalize => out.write_all(b"RenewFinalize")?,
+            MessageType::RenewRevoke => out.write_all(b"RenewRevoke")?,
+            MessageType::CollaborativeCloseOffer => out.write_all(b"CollaborativeCloseOffer")?,
+            MessageType::Reject => out.write_all(b"Reject")?,
+        }
+        Ok(IsNull::No)
+    }
+}
+
+impl FromSql<MessageTypeType, Pg> for MessageType {
+    fn from_sql(bytes: PgValue<'_>) -> deserialize::Result<Self> {
+        match bytes.as_bytes() {
+            b"Offer" => Ok(MessageType::Offer),
+            b"Accept" => Ok(MessageType::Accept),
+            b"Sign" => Ok(MessageType::Sign),
+            b"SettleOffer" => Ok(MessageType::SettleOffer),
+            b"SettleAccept" => Ok(MessageType::SettleAccept),
+            b"SettleConfirm" => Ok(MessageType::SettleConfirm),
+            b"SettleFinalize" => Ok(MessageType::SettleFinalize),
+            b"RenewOffer" => Ok(MessageType::RenewOffer),
+            b"RenewAccept" => Ok(MessageType::RenewAccept),
+            b"RenewConfirm" => Ok(MessageType::RenewConfirm),
+            b"RenewFinalize" => Ok(MessageType::RenewFinalize),
+            b"RenewRevoke" => Ok(MessageType::RenewRevoke),
+            b"CollaborativeCloseOffer" => Ok(MessageType::CollaborativeCloseOffer),
+            b"Reject" => Ok(MessageType::Reject),
+            _ => Err("Unrecognized enum variant".into()),
+        }
+    }
+}