aljaz/sphinx-key
1
1
Fork 0
mirror of https://github.com/stakwork/sphinx-key.git synced 2025-12-24 02:34:28 +01:00
Code Issues Packages Projects Releases Wiki Activity

make LSS optional if no muts are received

Browse Source
This commit is contained in:
Evan Feenstra
2023-06-02 15:40:11 -07:00
parent 2b1933534d
commit ea7e91d1c3
4 changed files with 34 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
broker/Cargo.lock generated
View File

@@ -1713,7 +1713,7 @@ dependencies = [
[[package]] [[package]]
name = "lss-connector" name = "lss-connector"
version = "0.1.0" version = "0.1.0"
source = "git+https://github.com/stakwork/sphinx-rs#da6850edff5dd6ba61a3c6bb63c8ef2bb967ddd3" source = "git+https://github.com/stakwork/sphinx-rs#d532a155b80efc76925cf1fecf035ccc71a63b89"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"lightning-storage-server", "lightning-storage-server",
@@ -3173,7 +3173,7 @@ dependencies = [
[[package]] [[package]]
name = "sphinx-auther" name = "sphinx-auther"
version = "0.1.12" version = "0.1.12"
source = "git+https://github.com/stakwork/sphinx-rs#da6850edff5dd6ba61a3c6bb63c8ef2bb967ddd3" source = "git+https://github.com/stakwork/sphinx-rs#d532a155b80efc76925cf1fecf035ccc71a63b89"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"base64 0.13.1", "base64 0.13.1",
@@ -3185,7 +3185,7 @@ dependencies = [
[[package]] [[package]]
name = "sphinx-glyph" name = "sphinx-glyph"
version = "0.1.2" version = "0.1.2"
source = "git+https://github.com/stakwork/sphinx-rs#da6850edff5dd6ba61a3c6bb63c8ef2bb967ddd3" source = "git+https://github.com/stakwork/sphinx-rs#d532a155b80efc76925cf1fecf035ccc71a63b89"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"hex", "hex",
@@ -3231,7 +3231,7 @@ dependencies = [
[[package]] [[package]]
name = "sphinx-signer" name = "sphinx-signer"
version = "0.1.0" version = "0.1.0"
source = "git+https://github.com/stakwork/sphinx-rs#da6850edff5dd6ba61a3c6bb63c8ef2bb967ddd3" source = "git+https://github.com/stakwork/sphinx-rs#d532a155b80efc76925cf1fecf035ccc71a63b89"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"bip39", "bip39",

4
broker/src/conn.rs
View File

@@ -105,8 +105,8 @@ impl ChannelRequest {
// mpsc reply // mpsc reply
#[derive(Debug)] #[derive(Debug)]
pub struct ChannelReply { pub struct ChannelReply {
// the return topic // the return topic end part (after last "/")
pub topic: String, pub topic_end: String,
pub reply: Vec<u8>, pub reply: Vec<u8>,
} }

31
broker/src/looper.rs
View File

@@ -160,17 +160,24 @@ impl<C: 'static + Client> SignerLoop<C> {
let md = parser::raw_request_from_bytes(message, self.chan.sequence, peer_id, dbid)?; let md = parser::raw_request_from_bytes(message, self.chan.sequence, peer_id, dbid)?;
// send to signer // send to signer
log::info!("SEND ON {}", topics::VLS); log::info!("SEND ON {}", topics::VLS);
let (_res_topic, res) = self.send_request_and_get_reply(topics::VLS, md)?; let (res_topic, res) = self.send_request_and_get_reply(topics::VLS, md)?;
// send reply to LSS to store muts log::info!("GOT ON {}", res_topic);
log::info!("GOT ON {}", _res_topic); let mut the_res = res.clone();
let lss_reply = self.send_lss_and_get_reply(res)?; if res_topic == topics::LSS_RES {
log::info!("LSS REPLY LEN {}", &lss_reply.len()); // send reply to LSS to store muts
// send to signer for HMAC validation, and get final reply let lss_reply = self.send_lss_and_get_reply(res)?;
log::info!("SEND ON {}", topics::LSS_MSG); log::info!("LSS REPLY LEN {}", &lss_reply.len());
let (_res_topic, res2) = self.send_request_and_get_reply(topics::LSS_MSG, lss_reply)?; // send to signer for HMAC validation, and get final reply
// create reply for CLN log::info!("SEND ON {}", topics::LSS_MSG);
log::info!("GOT ON {}, send to CLN", _res_topic); let (res_topic2, res2) = self.send_request_and_get_reply(topics::LSS_MSG, lss_reply)?;
let reply = parser::raw_response_from_bytes(res2, self.chan.sequence)?; log::info!("GOT ON {}, send to CLN", res_topic2);
if res_topic2 != topics::VLS_RETURN {
log::warn!("got a topic NOT on {}", topics::VLS_RETURN);
}
the_res = res2;
}
// create reply bytes for CLN
let reply = parser::raw_response_from_bytes(the_res, self.chan.sequence)?;
// add to the sequence // add to the sequence
self.chan.sequence = self.chan.sequence.wrapping_add(1); self.chan.sequence = self.chan.sequence.wrapping_add(1);
// catch the pubkey if its the first one connection // catch the pubkey if its the first one connection
@@ -213,7 +220,7 @@ impl<C: 'static + Client> SignerLoop<C> {
.map_err(|_| Error::Eof)?; .map_err(|_| Error::Eof)?;
let reply = reply_rx.blocking_recv().map_err(|_| Error::Eof)?; let reply = reply_rx.blocking_recv().map_err(|_| Error::Eof)?;
Ok((reply.topic, reply.reply)) Ok((reply.topic_end, reply.reply))
} }
fn send_lss_and_get_reply(&mut self, message: Vec<u8>) -> Result<Vec<u8>> { fn send_lss_and_get_reply(&mut self, message: Vec<u8>) -> Result<Vec<u8>> {

15
broker/src/mqtt.rs
View File

@@ -87,8 +87,8 @@ pub fn start_broker(
if let Err(e) = link_tx.publish(pub_topic, msg.message.clone()) { if let Err(e) = link_tx.publish(pub_topic, msg.message.clone()) {
log::error!("failed to pub to link_tx! {} {:?}", cid, e); log::error!("failed to pub to link_tx! {} {:?}", cid, e);
} }
if let Ok((cid, topic, reply)) = msg_rx.recv() { if let Ok((cid, topic_end, reply)) = msg_rx.recv() {
if let Err(_) = msg.reply_tx.send(ChannelReply { reply, topic }) { if let Err(_) = msg.reply_tx.send(ChannelReply { reply, topic_end }) {
log::warn!("could not send on reply_tx {}", cid); log::warn!("could not send on reply_tx {}", cid);
} }
} }
@@ -111,9 +111,10 @@ pub fn start_broker(
} }
// and receive from the correct client (or timeout to next) // and receive from the correct client (or timeout to next)
let dur = Duration::from_secs(9); let dur = Duration::from_secs(9);
if let Ok((cid, topic, reply)) = msg_rx.recv_timeout(dur) { if let Ok((cid, topic_end, reply)) = msg_rx.recv_timeout(dur) {
if &cid == client { if &cid == client {
if let Err(_) = msg.reply_tx.send(ChannelReply { reply, topic }) { if let Err(_) = msg.reply_tx.send(ChannelReply { reply, topic_end })
{
log::warn!("could not send on reply_tx"); log::warn!("could not send on reply_tx");
} }
break 'retry_loop; break 'retry_loop;
@@ -149,8 +150,10 @@ pub fn start_broker(
continue; continue;
} }
let cid = ts[0].to_string(); let cid = ts[0].to_string();
let topic = ts[1].to_string(); let topic_end = ts[1].to_string();
if let Err(e) = msg_tx.send((cid, topic, f.publish.payload.to_vec())) { if let Err(e) =
msg_tx.send((cid, topic_end, f.publish.payload.to_vec()))
{
log::error!("failed to pub to msg_tx! {:?}", e); log::error!("failed to pub to msg_tx! {:?}", e);
} }
} }