From 7514fa3ca66f4493fa0582cb8199ff6f865ccd5f Mon Sep 17 00:00:00 2001 From: Simon Wicky Date: Thu, 9 Nov 2023 15:16:42 +0100 Subject: [PATCH] async credential sending --- .../connection_handler/authenticated.rs | 4 +- .../websocket/connection_handler/coconut.rs | 102 ++++++++++++++++-- gateway/src/node/mod.rs | 1 + nym-api/nym-api-requests/src/coconut.rs | 2 +- 4 files changed, 97 insertions(+), 12 deletions(-) diff --git a/gateway/src/node/client_handling/websocket/connection_handler/authenticated.rs b/gateway/src/node/client_handling/websocket/connection_handler/authenticated.rs index 1341e3acca..0365619f2d 100644 --- a/gateway/src/node/client_handling/websocket/connection_handler/authenticated.rs +++ b/gateway/src/node/client_handling/websocket/connection_handler/authenticated.rs @@ -251,11 +251,9 @@ where .check_payment(&credential, &aggregated_verification_key) .await?; - //SW PUT that in a new threads, ensuring it eventually gets sent self.inner .ecash_verifier - .post_credential(current_api_clients, credential.clone()) - .await?; + .post_credential(current_api_clients, credential.clone())?; self.inner.storage.insert_credential(credential).await?; diff --git a/gateway/src/node/client_handling/websocket/connection_handler/coconut.rs b/gateway/src/node/client_handling/websocket/connection_handler/coconut.rs index e7cf4c68cd..05961b9625 100644 --- a/gateway/src/node/client_handling/websocket/connection_handler/coconut.rs +++ b/gateway/src/node/client_handling/websocket/connection_handler/coconut.rs @@ -3,7 +3,9 @@ use super::authenticated::RequestHandlingError; use chrono::Utc; -use log::*; +use futures::channel::mpsc::{self, UnboundedReceiver, UnboundedSender}; +use futures::StreamExt; +use nym_api_requests::coconut::VerifyCredentialBody; use nym_compact_ecash::scheme::EcashCredential; use nym_compact_ecash::setup::Parameters; use nym_compact_ecash::{PayInfo, VerificationKeyAuth}; @@ -11,8 +13,10 @@ use nym_validator_client::coconut::all_ecash_api_clients; use nym_validator_client::{ nyxd::contract_traits::DkgQueryClient, CoconutApiClient, DirectSigningHttpRpcNyxdClient, }; +use std::collections::VecDeque; use std::sync::Arc; use std::sync::Mutex; +use tokio::time::{interval, Duration}; const TIME_RANGE_SEC: i64 = 30; @@ -21,6 +25,7 @@ pub(crate) struct EcashVerifier { ecash_parameters: Parameters, pk_bytes: [u8; 32], //bytes represenation of a pub key representing the verifier pay_infos: Arc>>, + cred_sender: UnboundedSender, } impl EcashVerifier { @@ -28,12 +33,18 @@ impl EcashVerifier { nyxd_client: DirectSigningHttpRpcNyxdClient, ecash_parameters: Parameters, pk_bytes: [u8; 32], + shutdown: nym_task::TaskClient, ) -> Self { + let (cred_sender, cred_receiver) = mpsc::unbounded(); + let cs = CredentialSender::new(cred_receiver); + cs.start(shutdown); + EcashVerifier { nyxd_client, ecash_parameters, pk_bytes, pay_infos: Arc::new(Mutex::new(Vec::new())), + cred_sender, } } @@ -161,7 +172,7 @@ impl EcashVerifier { Ok(()) } - pub async fn post_credential( + pub fn post_credential( &self, api_clients: Vec, credential: EcashCredential, @@ -173,23 +184,98 @@ impl EcashVerifier { ); for client in api_clients { - let ret = client.api_client.verify_bandwidth_credential(&req).await; + self.cred_sender + .unbounded_send(CredentialToBeSent(req.clone(), client)) + .map_err(|_| RequestHandlingError::InternalError)? + } + Ok(()) + } +} - match ret { +#[derive(Clone)] +struct CredentialToBeSent(VerifyCredentialBody, CoconutApiClient); +struct CredentialSender { + pending: VecDeque, + cred_receiver: UnboundedReceiver, +} + +impl CredentialSender { + fn new(cred_receiver: UnboundedReceiver) -> Self { + CredentialSender { + pending: VecDeque::new(), + cred_receiver, + } + } + + async fn handle_credential(&mut self, credential: CredentialToBeSent) { + self.pending.push_back(credential); + self.try_empty_pending().await; + } + async fn try_empty_pending(&mut self) { + let mut new_pending = VecDeque::new(); + for credential in &self.pending { + match credential + .1 + .api_client + .verify_bandwidth_credential(&credential.0) + .await + { Ok(res) => { if !res.verification_result { - debug!( + log::debug!( "Validator {} didn't accept the credential.", - client.api_client.nym_api.current_url() + credential.1.api_client.nym_api.current_url() ); } } Err(e) => { - warn!("Validator {} could not be reached. There might be a problem with the coconut endpoint - {:?}", client.api_client.nym_api.current_url(), e); + log::warn!("Validator {} could not be reached. There might be a problem with the coconut endpoint - {:?}", credential + .1.api_client.nym_api.current_url(), e); + new_pending.push_back(credential.clone()); } } } + self.pending = new_pending; + } - Ok(()) + fn start(self, shutdown: nym_task::TaskClient) { + tokio::spawn(async move { self.run(shutdown).await }); + + //spawn a new thread that tries to send all pending Credentials. + //if it cannot send something, back of the queue and retry in 5min + //if everything has been sent, stop running + } + + async fn run(mut self, mut shutdown: nym_task::TaskClient) { + log::debug!("Starting Ecash CredentialSender"); + let mut interval = interval(Duration::from_secs(300)); + + while !shutdown.is_shutdown() { + tokio::select! { + biased; + _ = shutdown.recv() => { + log::trace!("client_handling::credentialSender : received shutdown"); + }, + Some(credential) = self.cred_receiver.next() => self.handle_credential(credential).await, + _ = interval.tick(), if !self.pending.is_empty() => self.try_empty_pending().await, + + } + } } } + +//let ret = client.api_client.verify_bandwidth_credential(&req).await; + +// match ret { +// Ok(res) => { +// if !res.verification_result { +// debug!( +// "Validator {} didn't accept the credential.", +// client.api_client.nym_api.current_url() +// ); +// } +// } +// Err(e) => { +// warn!("Validator {} could not be reached. There might be a problem with the coconut endpoint - {:?}", client.api_client.nym_api.current_url(), e); +// } +// } diff --git a/gateway/src/node/mod.rs b/gateway/src/node/mod.rs index 21153d625d..26f99c213c 100644 --- a/gateway/src/node/mod.rs +++ b/gateway/src/node/mod.rs @@ -470,6 +470,7 @@ impl Gateway { nyxd_client, ecash_params, self.identity_keypair.public_key().to_bytes(), + shutdown.subscribe().named("EcashVerifier"), ) }; diff --git a/nym-api/nym-api-requests/src/coconut.rs b/nym-api/nym-api-requests/src/coconut.rs index 9eb4583122..6f65faa191 100644 --- a/nym-api/nym-api-requests/src/coconut.rs +++ b/nym-api/nym-api-requests/src/coconut.rs @@ -11,7 +11,7 @@ use nym_compact_ecash::{ }; use serde::{Deserialize, Serialize}; -#[derive(Serialize, Deserialize, Getters, CopyGetters)] +#[derive(Serialize, Deserialize, Getters, CopyGetters, Clone)] pub struct VerifyCredentialBody { #[getset(get = "pub")] credential: EcashCredential,