Merge pull request #79 from L2-Technology/batch-channel-opens

batch channel opens

.gitignore

@@ -1,6 +1,6 @@
 /target
 /.sensei-tests
-./senseicore/.sensei-tests
+/senseicore/.sensei-tests
 .DS_Store
 *_pb2.py
 *_pb2_grpc.py

Cargo.toml

@@ -27,7 +27,7 @@ bitcoin-bech32 = "0.12"
 bech32 = "0.8"
 futures = "0.3"
 chrono = "0.4"
-rand = "0.4"
+rand = "0.8.4"
 axum = { version = "0.4.2", features = ["headers"] }
 http = "0.2"
 tower = { version = "0.4", features = ["full"] }
@@ -64,6 +64,7 @@ tonic-build = "0.6"
 
 [dev-dependencies]
 bitcoind = { version = "0.26", features = [ "22_0" ] }
+serial_test = "0.6.0"
 
 [[test]]
 name = "senseicore"

proto/sensei.proto

@@ -20,7 +20,7 @@ service Node {
     rpc StopNode (StopNodeRequest) returns (StopNodeResponse);
     rpc GetUnusedAddress (GetUnusedAddressRequest) returns (GetUnusedAddressResponse);
     rpc GetBalance (GetBalanceRequest) returns (GetBalanceResponse);
-    rpc OpenChannel (OpenChannelRequest) returns (OpenChannelResponse);
+    rpc OpenChannels (OpenChannelsRequest) returns (OpenChannelsResponse);
     rpc PayInvoice (PayInvoiceRequest) returns (PayInvoiceResponse);
     rpc DecodeInvoice (DecodeInvoiceRequest) returns (DecodeInvoiceResponse);
     rpc Keysend (KeysendRequest) returns (KeysendResponse);
@@ -189,13 +189,24 @@ message GetBalanceResponse {
     uint64 balance_satoshis = 1;
 }
 
-message OpenChannelRequest {
+message OpenChannelInfo {
     string node_connection_string = 1;
     uint64 amt_satoshis = 2;
     bool public = 3;
 }
-message OpenChannelResponse {
-    string temp_channel_id = 1;
+
+message OpenChannelResult {
+    bool error = 1;
+    optional string error_message = 2;
+    optional string temp_channel_id = 3;
+}
+
+message OpenChannelsRequest {
+    repeated OpenChannelInfo channels = 1;
+}
+message OpenChannelsResponse {
+    repeated OpenChannelInfo channels = 1;
+    repeated OpenChannelResult results = 2;
 }
 
 message PayInvoiceRequest {

senseicore/Cargo.toml

@@ -21,7 +21,7 @@ bitcoin-bech32 = "0.12"
 bech32 = "0.8"
 futures = "0.3"
 chrono = "0.4"
-rand = "0.4"
+rand = "0.8.4"
 axum = { version = "0.4.2", features = ["headers"] }
 http = "0.2"
 tower = { version = "0.4", features = ["full"] }
@@ -49,3 +49,4 @@ migration = { path = "../migration" }
 
 [dev-dependencies]
 bitcoind = { version = "0.26", features = [ "22_0" ] }
+serial_test = "0.6.0"

senseicore/src/chain/broadcaster.rs

@@ -1,21 +1,45 @@
-use std::sync::{Arc, Mutex};
+use std::{
+    collections::HashMap,
+    sync::{Arc, Mutex},
+};
 
 use crate::events::SenseiEvent;
 
 use super::database::WalletDatabase;
-use bitcoin::Transaction;
+use bitcoin::{Transaction, Txid};
 use lightning::chain::chaininterface::BroadcasterInterface;
 use tokio::sync::broadcast;
 
 pub struct SenseiBroadcaster {
+    pub debounce: Mutex<HashMap<Txid, usize>>,
     pub node_id: String,
     pub broadcaster: Arc<dyn BroadcasterInterface + Send + Sync>,
     pub wallet_database: Arc<Mutex<WalletDatabase>>,
     pub event_sender: broadcast::Sender<SenseiEvent>,
 }
 
-impl BroadcasterInterface for SenseiBroadcaster {
-    fn broadcast_transaction(&self, tx: &Transaction) {
+impl SenseiBroadcaster {
+    pub fn new(
+        node_id: String,
+        broadcaster: Arc<dyn BroadcasterInterface + Send + Sync>,
+        wallet_database: Arc<Mutex<WalletDatabase>>,
+        event_sender: broadcast::Sender<SenseiEvent>,
+    ) -> Self {
+        Self {
+            node_id,
+            broadcaster,
+            wallet_database,
+            event_sender,
+            debounce: Mutex::new(HashMap::new()),
+        }
+    }
+
+    pub fn set_debounce(&self, txid: Txid, count: usize) {
+        let mut debounce = self.debounce.lock().unwrap();
+        debounce.insert(txid, count);
+    }
+
+    pub fn broadcast(&self, tx: &Transaction) {
         self.broadcaster.broadcast_transaction(tx);
 
         // TODO: there's a bug here if the broadcast fails
@@ -31,3 +55,23 @@ impl BroadcasterInterface for SenseiBroadcaster {
             .unwrap_or_default();
     }
 }
+
+impl BroadcasterInterface for SenseiBroadcaster {
+    fn broadcast_transaction(&self, tx: &Transaction) {
+        let txid = tx.txid();
+
+        let mut debounce = self.debounce.lock().unwrap();
+
+        let can_broadcast = match debounce.get_mut(&txid) {
+            Some(count) => {
+                *count -= 1;
+                *count == 0
+            }
+            None => true,
+        };
+
+        if can_broadcast {
+            self.broadcast(tx);
+        }
+    }
+}