fix: double event race condition

BoltzExchange · Mar 7, 2025 · bb880f2 · bb880f2
1 parent 37394d2
commit bb880f2
Show file tree

Hide file tree

Showing 2 changed files with 51 additions and 26 deletions.
diff --git a/regtest b/regtest
diff --git a/src/grpc/service.rs b/src/grpc/service.rs
@@ -13,6 +13,7 @@ use crate::grpc::transformers::{transform_invoice_state, transform_route_hints};
 use crate::settler::Settler;
 use bitcoin::hashes::{sha256, Hash};
 use log::{debug, error, warn};
+use std::collections::HashMap;
 use std::pin::Pin;
 use tokio::sync::mpsc;
 use tonic::codegen::tokio_stream::wrappers::ReceiverStream;
@@ -217,6 +218,7 @@ where
         let params = request.into_inner();
         let (tx, rx) = mpsc::channel(16);
 
+        let mut initial_state = None;
         let mut state_rx = self.settler.state_rx();
 
         match self
@@ -225,17 +227,26 @@ where
         {
             Ok(res) => {
                 if let Some(res) = res {
-                    if let Ok(state) = InvoiceState::try_from(res.invoice.state.as_str()) {
-                        if let Err(err) = tx
-                            .send(Ok(TrackResponse {
-                                state: transform_invoice_state(state),
-                            }))
-                            .await
-                        {
-                            error!("Could not send invoice state update: {}", err);
+                    match InvoiceState::try_from(res.invoice.state.as_str()) {
+                        Ok(state) => {
+                            initial_state = Some(state);
+                            if let Err(err) = tx
+                                .send(Ok(TrackResponse {
+                                    state: transform_invoice_state(state),
+                                }))
+                                .await
+                            {
+                                error!("Could not send invoice state update: {}", err);
+                                return Err(Status::new(
+                                    Code::Internal,
+                                    format!("could not send initial invoice state: {}", err),
+                                ));
+                            }
+                        }
+                        Err(err) => {
                             return Err(Status::new(
                                 Code::Internal,
-                                format!("could not send initial invoice state: {}", err),
+                                format!("could transform invoice state: {}", err),
                             ));
                         }
                     }
@@ -257,6 +268,13 @@ where
                             continue;
                         }
 
+                        // Do not send the initial state twice
+                        if let Some(initial_state) = initial_state {
+                            if initial_state == update.state {
+                                continue;
+                            }
+                        }
+
                         if let Err(err) = tx
                             .send(Ok(TrackResponse {
                                 state: transform_invoice_state(update.state),
@@ -292,6 +310,7 @@ where
 
         let (tx, rx) = mpsc::channel(128);
 
+        let mut initial_states = HashMap::new();
         let invoice_helper = self.invoice_helper.clone();
         let mut state_rx = self.settler.state_rx();
 
@@ -320,26 +339,25 @@ where
                     }
                 };
 
-                let state = transform_invoice_state(
-                    match InvoiceState::try_from(invoice.invoice.state.as_str()) {
-                        Ok(state) => state,
-                        Err(err) => {
-                            let err = format!(
-                                "Could not parse state of invoice {}: {}",
-                                hex::encode(&hash),
-                                err
-                            );
-                            error!("{}", err);
-                            let _ = tx.send(Err(Status::new(Code::Internal, err))).await;
-                            return;
-                        }
-                    },
-                );
+                let state = match InvoiceState::try_from(invoice.invoice.state.as_str()) {
+                    Ok(state) => state,
+                    Err(err) => {
+                        let err = format!(
+                            "Could not parse state of invoice {}: {}",
+                            hex::encode(&hash),
+                            err
+                        );
+                        error!("{}", err);
+                        let _ = tx.send(Err(Status::new(Code::Internal, err))).await;
+                        return;
+                    }
+                };
+                initial_states.insert(hash, state);
 
                 if let Err(err) = tx
                     .send(Ok(TrackAllResponse {
-                        state,
                         bolt11: invoice.invoice.bolt11,
+                        state: transform_invoice_state(state),
                         payment_hash: invoice.invoice.payment_hash,
                     }))
                     .await
@@ -352,6 +370,13 @@ where
             loop {
                 match state_rx.recv().await {
                     Ok(update) => {
+                        // Do not send the initial state twice
+                        if let Some(initial_state) = initial_states.get(&update.payment_hash) {
+                            if initial_state == &update.state {
+                                continue;
+                            }
+                        }
+
                         if let Err(err) = tx
                             .send(Ok(TrackAllResponse {
                                 bolt11: update.bolt11,
+5 −1		README.md
+2 −4		build.py
+24 −0		data/backend-nginx/default.conf