Payments - gateway timeout fix

core/payment/src/api/debit_notes.rs

@@ -16,6 +16,7 @@ use ya_net::RemoteEndpoint;
 use ya_persistence::executor::DbExecutor;
 use ya_persistence::types::Role;
 use ya_service_api_web::middleware::Identity;
+use ya_service_bus::timeout::IntoTimeoutFuture;
 use ya_service_bus::{typed as bus, RpcEndpoint};
 
 // Local uses
@@ -217,17 +218,19 @@ async fn send_debit_note(
             dao.mark_received(debit_note_id, node_id).await?;
             Ok(())
         }
+        .timeout(Some(timeout))
         .await
         {
-            Ok(_) => {
+            Ok(Ok(_)) => {
                 log::info!("DebitNote [{}] sent.", path.debit_note_id);
                 counter!("payment.debit_notes.provider.sent", 1);
                 response::ok(Null)
             }
-            Err(Error::Rpc(RpcMessageError::Send(SendError::BadRequest(e)))) => {
+            Ok(Err(Error::Rpc(RpcMessageError::Send(SendError::BadRequest(e))))) => {
                 response::bad_request(&e)
             }
-            Err(e) => response::server_error(&e),
+            Ok(Err(e)) => response::server_error(&e),
+            Err(_) => response::timeout(&"Timeout sending DebitNote to remote Node."),
         }
     })
     .await;
@@ -331,7 +334,7 @@ async fn accept_debit_note(
     }
 
     let timeout = query.timeout.unwrap_or(params::DEFAULT_ACK_TIMEOUT);
-    let result = with_timeout(timeout, async move {
+    let result = async move {
         let issuer_id = debit_note.issuer_id;
         let accept_msg = AcceptDebitNote::new(debit_note_id.clone(), acceptance, issuer_id);
         let schedule_msg =
@@ -356,19 +359,23 @@ async fn accept_debit_note(
             log::trace!("Debit Note accepted successfully for [{}]", debit_note_id);
             Ok(())
         }
+        .timeout(Some(timeout))
         .await
         {
-            Ok(_) => {
+            Ok(Ok(_)) => {
                 log::info!("DebitNote [{}] accepted.", path.debit_note_id);
                 counter!("payment.debit_notes.requestor.accepted", 1);
                 response::ok(Null)
             }
-            Err(Error::Rpc(RpcMessageError::AcceptReject(AcceptRejectError::BadRequest(e)))) => {
+            Ok(Err(Error::Rpc(RpcMessageError::AcceptReject(AcceptRejectError::BadRequest(
+                e,
+            ))))) => {
                 return response::bad_request(&e);
             }
-            Err(e) => return response::server_error(&e),
+            Ok(Err(e)) => return response::server_error(&e),
+            Err(_) => response::timeout(&"Timeout sending Invoice to remote Node."),
         }
-    })
+    }
     .await;
 
     timing!(

core/payment/src/api/invoices.rs

@@ -17,6 +17,7 @@ use ya_net::RemoteEndpoint;
 use ya_persistence::executor::DbExecutor;
 use ya_persistence::types::Role;
 use ya_service_api_web::middleware::Identity;
+use ya_service_bus::timeout::IntoTimeoutFuture;
 use ya_service_bus::{typed as bus, RpcEndpoint};
 
 // Local uses
@@ -198,7 +199,7 @@ async fn send_invoice(
     }
     let timeout = query.timeout.unwrap_or(params::DEFAULT_ACK_TIMEOUT);
 
-    let result = with_timeout(timeout, async move {
+    let result = async move {
         match async move {
             log::debug!(
                 "Sending invoice [{}] to [{}].",
@@ -214,19 +215,21 @@ async fn send_invoice(
             dao.mark_received(invoice_id, node_id).await?;
             Ok(())
         }
+        .timeout(Some(timeout))
         .await
         {
-            Ok(_) => {
+            Ok(Ok(_)) => {
                 log::info!("Invoice [{}] sent.", path.invoice_id);
                 counter!("payment.invoices.provider.sent", 1);
                 response::ok(Null)
             }
-            Err(Error::Rpc(RpcMessageError::Send(SendError::BadRequest(e)))) => {
+            Ok(Err(Error::Rpc(RpcMessageError::Send(SendError::BadRequest(e))))) => {
                 response::bad_request(&e)
             }
-            Err(e) => response::server_error(&e),
+            Ok(Err(e)) => response::server_error(&e),
+            Err(_) => response::timeout(&"Timeout sending Invoice to remote Node."),
         }
-    })
+    }
     .await;
 
     timing!("payment.invoices.provider.sent.time", start, Instant::now());
@@ -265,7 +268,7 @@ async fn cancel_invoice(
     }
 
     let timeout = query.timeout.unwrap_or(params::DEFAULT_ACK_TIMEOUT);
-    let result = with_timeout(timeout, async move {
+    let result = async move {
         match async move {
             log::debug!(
                 "Canceling invoice [{}] sent to [{}].",
@@ -284,19 +287,21 @@ async fn cancel_invoice(
             dao.cancel(invoice_id, node_id).await?;
             Ok(())
         }
+        .timeout(Some(timeout))
         .await
         {
-            Ok(_) => {
+            Ok(Ok(_)) => {
                 counter!("payment.invoices.provider.cancelled", 1);
                 log::info!("Invoice [{}] cancelled.", path.invoice_id);
                 response::ok(Null)
             }
-            Err(Error::Rpc(RpcMessageError::Cancel(CancelError::Conflict))) => {
+            Ok(Err(Error::Rpc(RpcMessageError::Cancel(CancelError::Conflict)))) => {
                 response::conflict(&"Invoice already accepted by requestor")
             }
-            Err(e) => response::server_error(&e),
+            Ok(Err(e)) => response::server_error(&e),
+            Err(_) => response::timeout(&"Timeout canceling Invoice on remote Node."),
         }
-    })
+    }
     .await;
 
     timing!(
@@ -393,7 +398,7 @@ async fn accept_invoice(
     }
 
     let timeout = query.timeout.unwrap_or(params::DEFAULT_ACK_TIMEOUT);
-    let result = with_timeout(timeout, async move {
+    let result = async move {
         let issuer_id = invoice.issuer_id;
         let accept_msg = AcceptInvoice::new(invoice_id.clone(), acceptance, issuer_id);
         let schedule_msg = SchedulePayment::from_invoice(invoice, allocation_id, amount_to_pay);
@@ -413,19 +418,21 @@ async fn accept_invoice(
             log::trace!("Invoice accepted successfully for [{}]", invoice_id);
             Ok(())
         }
+        .timeout(Some(timeout))
         .await
         {
-            Ok(_) => {
+            Ok(Ok(_)) => {
                 counter!("payment.invoices.requestor.accepted", 1);
                 log::info!("Invoice [{}] accepted.", path.invoice_id);
                 response::ok(Null)
             }
-            Err(Error::Rpc(RpcMessageError::AcceptReject(AcceptRejectError::BadRequest(e)))) => {
-                return response::bad_request(&e)
-            }
-            Err(e) => return response::server_error(&e),
+            Ok(Err(Error::Rpc(RpcMessageError::AcceptReject(AcceptRejectError::BadRequest(
+                e,
+            ))))) => return response::bad_request(&e),
+            Ok(Err(e)) => return response::server_error(&e),
+            Err(_) => response::timeout(&"Timeout accepting Invoice on remote Node."),
         }
-    })
+    }
     .await;
 
     timing!(

core/payment/src/dao/invoice.rs

@@ -1,6 +1,6 @@
 use crate::dao::{agreement, invoice_event};
 use crate::error::{DbError, DbResult};
-use crate::models::invoice::{InvoiceXActivity, ReadObj, WriteObj};
+use crate::models::invoice::{equivalent, InvoiceXActivity, ReadObj, WriteObj};
 use crate::schema::pay_agreement::dsl as agreement_dsl;
 use crate::schema::pay_invoice::dsl;
 use crate::schema::pay_invoice_x_activity::dsl as activity_dsl;
@@ -77,6 +77,20 @@ impl<'c> InvoiceDao<'c> {
         let owner_id = invoice.owner_id.clone();
         let role = invoice.role.clone();
         do_with_transaction(self.pool, move |conn| {
+            if let Some(read_invoice) = query!()
+                .filter(dsl::id.eq(&invoice_id))
+                .filter(dsl::owner_id.eq(owner_id))
+                .first(conn)
+                .optional()?
+            {
+                return match equivalent(&read_invoice, &invoice) {
+                    true => Ok(()),
+                    false => Err(DbError::Integrity(format!(
+                        "Invoice with the same id and different content already exists."
+                    ))),
+                };
+            };
+
             agreement::set_amount_due(&invoice.agreement_id, &owner_id, &invoice.amount, conn)?;
 
             diesel::insert_into(dsl::pay_invoice)

core/payment/src/models/invoice.rs

@@ -105,3 +105,13 @@ pub struct InvoiceXActivity {
     pub activity_id: String,
     pub owner_id: NodeId,
 }
+
+pub fn equivalent(read_invoice: &ReadObj, write_invoice: &WriteObj) -> bool {
+    read_invoice.agreement_id == write_invoice.agreement_id
+        && read_invoice.amount == write_invoice.amount
+        && read_invoice.id == write_invoice.id
+        && read_invoice.owner_id == write_invoice.owner_id
+        && read_invoice.payment_due_date == write_invoice.payment_due_date
+        && read_invoice.role == write_invoice.role
+        && read_invoice.status == write_invoice.status
+}

core/payment/src/utils.rs

@@ -198,8 +198,10 @@ pub mod response {
         HttpResponse::Unauthorized().json(ErrorMessage { message: None })
     }
 
-    pub fn timeout() -> HttpResponse {
-        HttpResponse::GatewayTimeout().json(ErrorMessage { message: None })
+    pub fn timeout(e: &impl ToString) -> HttpResponse {
+        HttpResponse::GatewayTimeout().json(ErrorMessage {
+            message: Some(e.to_string()),
+        })
     }
 
     pub fn server_error(e: &impl ToString) -> HttpResponse {