aljaz/sphinx-key
1
1
Fork 0
mirror of https://github.com/stakwork/sphinx-key.git synced 2025-12-18 15:54:31 +01:00
Code Issues Packages Projects Releases Wiki Activity

broker: fix LSS getting out of step when new signer connects

Browse Source
This commit is contained in:
irriden
2023-09-08 02:21:57 +00:00
parent 128a29f298
commit 1b00559424
1 changed files with 28 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
broker/src/mqtt.rs
View File

@@ -172,20 +172,43 @@ fn pub_and_wait(
std::thread::sleep(Duration::from_secs(1)); std::thread::sleep(Duration::from_secs(1));
None None
} else { } else {
let current = current.unwrap(); let current_cid = current.clone().unwrap();
// Try the current connection // Try the current connection
let mut rep = pub_timeout(&current, &msg.topic, &msg.message, &msg_rx, link_tx); let mut rep = pub_timeout(&current_cid, &msg.topic, &msg.message, &msg_rx, link_tx);
// We restart the loop in case a new signer connects while pinging for connections
// as this could mean that the LSS state advanced broker side only, and signer side
// it's one step behind. Here and also in the for loop.
// We do it as soon as we know.
let cs = conns_.lock().unwrap();
let new_current = cs.current.clone();
drop(cs);
if new_current != current {
log::warn!("Client list changed, starting over!");
counter = 0;
continue;
}
// If that failed, try looking for some other signer // If that failed, try looking for some other signer
if rep.is_none() { if rep.is_none() {
for cid in client_list.into_keys().filter(|k| k != &current) { for cid in client_list.into_keys().filter(|k| k != &current_cid) {
rep = pub_timeout(&cid, &msg.topic, &msg.message, &msg_rx, link_tx); rep = pub_timeout(&cid, &msg.topic, &msg.message, &msg_rx, link_tx);
let mut cs = conns_.lock().unwrap();
log::debug!("got the list lock!");
let new_current = cs.current.clone();
if new_current != current {
log::info!("Client list changed, starting over!");
drop(cs);
counter = 0;
rep = None;
break;
}
if rep.is_some() { if rep.is_some() {
let mut cs = conns_.lock().unwrap();
log::debug!("got the list lock!");
cs.set_current(cid.to_string()); cs.set_current(cid.to_string());
drop(cs); drop(cs);
break; break;
} }
drop(cs);
} }
} }
rep rep