Merge pull request #113 from okp4/feat/faucet-harassing

😵 Limit faucet harassing

.github/workflows/build.yml

@@ -34,7 +34,7 @@ jobs:
       - name: Setup rust
         uses: actions-rs/toolchain@v1
         with:
-          toolchain: 1.64
+          toolchain: 1.65
           default: true
           override: true
 
@@ -53,7 +53,7 @@ jobs:
       - name: Setup rust
         uses: actions-rs/toolchain@v1
         with:
-          toolchain: 1.64
+          toolchain: 1.65
           default: true
           override: true
 

.github/workflows/lint.yml

@@ -69,7 +69,7 @@ jobs:
         if: steps.changed-rust-cargo.outputs.any_changed == 'true'
         uses: actions-rs/toolchain@v1
         with:
-          toolchain: 1.64
+          toolchain: 1.65
           default: true
           override: true
 
@@ -102,7 +102,7 @@ jobs:
         if: steps.changed-rust-files.outputs.any_changed == 'true'
         uses: actions-rs/toolchain@v1
         with:
-          toolchain: 1.64
+          toolchain: 1.65
           default: true
           override: true
 
@@ -124,7 +124,7 @@ jobs:
       - name: Setup rust
         uses: actions-rs/toolchain@v1
         with:
-          toolchain: 1.64
+          toolchain: 1.65
           default: true
           override: true
 
@@ -155,7 +155,7 @@ jobs:
       - name: Setup rust
         uses: actions-rs/toolchain@v1
         with:
-          toolchain: 1.64
+          toolchain: 1.65
           default: true
           override: true
 

.github/workflows/publish.yml

@@ -20,7 +20,7 @@ jobs:
       - name: Setup rust
         uses: actions-rs/toolchain@v1
         with:
-          toolchain: 1.64
+          toolchain: 1.65
           default: true
           override: true
 

.github/workflows/release.yml

@@ -32,7 +32,7 @@ jobs:
       - name: Setup rust
         uses: actions-rs/toolchain@v1
         with:
-          toolchain: 1.64
+          toolchain: 1.65
           default: true
           override: true
 

.github/workflows/test.yml

@@ -34,7 +34,7 @@ jobs:
       - name: Setup rust
         uses: actions-rs/toolchain@v1
         with:
-          toolchain: 1.64
+          toolchain: 1.65
           default: true
           override: true
 

Cargo.toml

@@ -6,7 +6,7 @@ license-file = "LICENSE"
 name = "discord_bot"
 readme = "README.md"
 repository = "https://github.com/okp4/discord-bot"
-rust-version = "1.64"
+rust-version = "1.65"
 version = "0.1.0"
 
 [dependencies]

Dockerfile

@@ -1,5 +1,5 @@
 #--- Build stage
-FROM clux/muslrust:1.64.0-stable as builder
+FROM clux/muslrust:1.65.0-stable as builder
 
 WORKDIR /app
 

src/cli/commands/start.rs

@@ -92,6 +92,8 @@ impl Runnable for StartCmd {
                 |handler| {
                     handler.memo = config.faucet.memo.to_string();
                     handler.batch_window = config.chain.batch_transaction_window;
+                    handler.max_msg = config.chain.max_msg;
+                    handler.queue_limit = config.chain.queue_limit;
                     handler
                 },
             )

src/cli/config.rs

@@ -102,6 +102,13 @@ pub struct ChainSection {
 
     /// Duration between two transaction batch.
     pub batch_transaction_window: Duration,
+
+    /// Configure the maximum number of message that are sent by transaction
+    /// batch.
+    pub max_msg: usize,
+
+    /// Maximum number of transaction waiting in the queue.
+    pub queue_limit: usize,
 }
 
 impl Default for ChainSection {
@@ -112,6 +119,8 @@ impl Default for ChainSection {
             denom: "know".to_string(),
             prefix: "okp4".to_string(),
             batch_transaction_window: Duration::from_secs(8),
+            max_msg: 7,
+            queue_limit: 1000,
         }
     }
 }

src/cosmos/faucet/handlers/request_funds.rs

@@ -1,12 +1,13 @@
 use crate::cosmos::faucet::messages::request_funds::{RequestFunds, RequestFundsResult};
 use crate::cosmos::faucet::Faucet;
-use crate::cosmos::tx::messages::register_msg::RegisterMsg;
-use actix::Handler;
+use crate::cosmos::tx::error::Error::Mailbox;
+use crate::cosmos::tx::messages::register_msg::{RegisterMsg, RegisterMsgResult};
+use actix::{Handler, MailboxError, ResponseFuture};
 use cosmrs::bank::MsgSend;
 use tracing::info;
 
 impl Handler<RequestFunds> for Faucet {
-    type Result = RequestFundsResult;
+    type Result = ResponseFuture<RequestFundsResult>;
 
     fn handle(&mut self, msg: RequestFunds, _: &mut Self::Context) -> Self::Result {
         let msg_send = MsgSend {
@@ -15,9 +16,15 @@ impl Handler<RequestFunds> for Faucet {
             amount: vec![self.amount.clone()],
         };
 
-        self.tx_handler
-            .do_send(RegisterMsg::new(msg_send, Some(msg.requester)));
+        info!("✍️ Register request funds for {}", msg.address);
 
-        info!("✍️  Register request funds for {}", msg.address);
+        let tx_handler = self.tx_handler.clone();
+        Box::pin(async move {
+            let result: Result<RegisterMsgResult, MailboxError> = tx_handler
+                .send(RegisterMsg::new(msg_send, msg.requester))
+                .await;
+
+            result.map_err(|e| Mailbox(e.to_string())).and_then(|r| r)
+        })
     }
 }

src/cosmos/faucet/messages/request_funds.rs

@@ -1,10 +1,11 @@
 //! Holds Request funds messages
+use crate::cosmos::tx::error::Error;
 use actix::Message;
 use cosmrs::AccountId;
 use serenity::model::user::User;
 
 /// Represent the RequestFunds message result
-pub type RequestFundsResult = ();
+pub type RequestFundsResult = Result<(), Error>;
 
 /// Request funds from the faucet message
 #[derive(Message)]

src/cosmos/tx/error.rs

@@ -19,9 +19,21 @@ pub enum Error {
     #[error("An error occur at the grpc client: {0}")]
     Client(crate::cosmos::client::error::Error),
 
+    /// The transaction queue is full.
+    #[error("The transaction queue is full")]
+    QueueFull,
+
+    /// A transaction is already registered for this user
+    #[error("A transaction is already registered for this user")]
+    DuplicateUser,
+
     /// Failed communicate to actix.
     #[error("An error occurs when send message to actix: {0}.")]
     Mailbox(String),
+
+    /// Error occurs to lock mutex
+    #[error("Error occurs to lock mutex")]
+    Lock,
 }
 
 impl From<CosmosErrorReport> for Error {

src/cosmos/tx/handlers/register_msg.rs

@@ -1,24 +1,42 @@
 //! Register transaction handler
 
+use crate::cosmos::tx::error::Error::{DuplicateUser, Lock, QueueFull};
 use crate::cosmos::tx::messages::register_msg::{RegisterMsg, RegisterMsgResult};
 use crate::cosmos::tx::messages::response::TxResult;
 use crate::cosmos::tx::TxHandler;
 use actix::dev::ToEnvelope;
 use actix::{Actor, Handler};
 use cosmrs::tx::Msg;
+use tracing::info;
+use tracing::log::warn;
 
 impl<T, R> Handler<RegisterMsg<T>> for TxHandler<T, R>
 where
-    T: Msg + Unpin + 'static,
+    T: Msg + Unpin + 'static + PartialEq,
     R: Actor + Handler<TxResult>,
     R::Context: ToEnvelope<R, TxResult>,
 {
     type Result = RegisterMsgResult;
 
     fn handle(&mut self, msg: RegisterMsg<T>, _: &mut Self::Context) -> Self::Result {
-        if msg.subscriber.is_some() {
-            self.subscribers.push(msg.subscriber.unwrap())
+        let mut msgs = self.msgs_queue.lock().map_err(|_| Lock)?;
+
+        if msgs.len() >= self.queue_limit {
+            warn!(
+                "❌ Could not register funds for {}, the queue is full.",
+                msg.subscriber.name
+            );
+            Err(QueueFull)
+        } else if msgs.iter().any(|f| f.0.id == msg.subscriber.id) {
+            info!(
+                "👮‍ The user {} already register transaction, skip this one.",
+                msg.subscriber.name
+            );
+            Err(DuplicateUser)
+        } else {
+            info!("🤑 Register transaction for {}", msg.subscriber.name);
+            msgs.push_back((msg.subscriber, msg.msg));
+            Ok(())
         }
-        self.msgs.push(msg.msg);
     }
 }

src/cosmos/tx/handlers/trigger.rs

@@ -9,6 +9,8 @@ use crate::cosmos::tx::TxHandler;
 use actix::dev::ToEnvelope;
 use actix::{Actor, ActorFutureExt, Handler, MailboxError, ResponseActFuture, WrapFuture};
 use cosmrs::tx::{Body, Msg};
+use serenity::model::user::User;
+use std::cmp::min;
 use tracing::info;
 use tracing::log::error;
 
@@ -21,15 +23,21 @@ where
     type Result = ResponseActFuture<Self, TriggerTxResult>;
 
     fn handle(&mut self, msg: TriggerTx, _ctx: &mut Self::Context) -> Self::Result {
-        if self.msgs.is_empty() {
+        let Ok(mut msgs_queue) = self.msgs_queue.lock() else {
+            error!("❌ Failed lock msgs queue, request fund couldn't be registered.");
+            return Box::pin(async {}.into_actor(self));
+        };
+        let msgs_queue_len = msgs_queue.len();
+
+        if msgs_queue.is_empty() {
             info!("🥹 No message to submit");
             return Box::pin(async {}.into_actor(self));
         }
 
-        let msgs = self.msgs.clone();
-        let subscribers = self.subscribers.clone();
-        self.msgs.clear();
-        self.subscribers.clear();
+        let (subscribers, msgs): (Vec<User>, Vec<T>) = msgs_queue
+            .drain(..min(msgs_queue_len, self.max_msg))
+            .map(|it| (it.0, it.1))
+            .unzip();
 
         let grpc_client = self.grpc_client.clone();
         let sender_address = self.sender.address.to_string();

src/cosmos/tx/messages/register_msg.rs

@@ -1,11 +1,12 @@
 //! Register transaction message
 
+use crate::cosmos::tx::error::Error;
 use actix::Message;
 use cosmrs::tx::Msg;
 use serenity::model::user::User;
 
 /// Result of a RegisterMsg message.
-pub type RegisterMsgResult = ();
+pub type RegisterMsgResult = Result<(), Error>;
 
 /// Register transaction's message actor message.
 #[derive(Message)]
@@ -18,15 +19,15 @@ where
     pub msg: T,
 
     /// Transaction subscriber
-    pub subscriber: Option<User>,
+    pub subscriber: User,
 }
 
 impl<T> RegisterMsg<T>
 where
     T: Msg,
 {
     /// Create a new RegisterMsg.
-    pub fn new(msg: T, subscriber: Option<User>) -> Self
+    pub fn new(msg: T, subscriber: User) -> Self
     where
         T: Msg,
     {

src/cosmos/tx/mod.rs

@@ -14,6 +14,8 @@ use actix::{Actor, Addr, Handler};
 use cosmos_sdk_proto::cosmos::auth::v1beta1::BaseAccount;
 use cosmrs::tx::{Body, Fee, Msg, SignDoc, SignerInfo};
 use serenity::model::user::User;
+use std::collections::VecDeque;
+use std::sync::{Arc, Mutex};
 use std::time::Duration;
 use tonic::transport::Channel;
 
@@ -34,10 +36,12 @@ where
     pub fee: Fee,
     /// Duration between two transactions.
     pub batch_window: Duration,
+    /// Number of max message per transactions
+    pub max_msg: usize,
+    /// Number of message that can be in the queue
+    pub queue_limit: usize,
     /// Contains the batch of transaction message to sent as prost::Any.
-    msgs: Vec<T>,
-    /// Contains the list of all user that request transaction.
-    subscribers: Vec<User>,
+    msgs_queue: Arc<Mutex<VecDeque<(User, T)>>>,
     /// GRPC client to send transaction.
     grpc_client: Addr<Client<Channel>>,
     /// Hold address of actor that will receive message when a transaction has been broadcasted.
@@ -66,8 +70,9 @@ where
             memo: "".to_string(),
             fee,
             batch_window: Duration::new(8, 0),
-            msgs: vec![],
-            subscribers: vec![],
+            max_msg: 7,
+            queue_limit: 1000,
+            msgs_queue: Arc::new(Mutex::new(VecDeque::new())),
             grpc_client,
             response_handler: None,
         };

src/discord/discord_server/cmd/request.rs

@@ -1,9 +1,12 @@
 //! Holds request commands functions
 
-use crate::cosmos::faucet::messages::request_funds::RequestFunds;
+use crate::cosmos::faucet::messages::request_funds::{RequestFunds, RequestFundsResult};
+use crate::cosmos::tx::error::Error as RequestError;
+use crate::cosmos::tx::error::Error::Mailbox;
 use crate::discord::discord_server::cmd::CommandExecutable;
 use crate::discord::discord_server::error::Error;
 use crate::discord::discord_server::Actors;
+use actix::MailboxError;
 use serenity::async_trait;
 use serenity::client::Context;
 use serenity::model::application::interaction::application_command::ApplicationCommandInteraction;
@@ -29,11 +32,21 @@ impl CommandExecutable for RequestCmd {
         info!("💰 request fund slash command");
 
         let msg = if let Ok(address) = self.address.parse() {
-            actors.faucet.do_send(RequestFunds {
-                address,
-                requester: command.user.clone(),
-            });
-            "📥 Funds has been successfully requested.".to_string()
+            let request: Result<RequestFundsResult, MailboxError> = actors
+                .faucet
+                .send(RequestFunds {
+                    address,
+                    requester: command.user.clone(),
+                })
+                .await;
+            match request.map_err(|e| Mailbox(e.to_string())).and_then(|r| r) {
+                Ok(_) => "📥 Funds has been successfully requested.".to_string(),
+                Err(err) => match err {
+                    RequestError::QueueFull => "❌ Queue is full, please wait before re-submit funds request.".to_string(),
+                    RequestError::DuplicateUser => "❌ You have already requested funds, please wait before re-submiting your request.".to_string(),
+                    _ => "❌ An error occurs, please try again.".to_string(),
+                }
+            }
         } else {
             format!(
                 "❌ Your wallet address `{}` seems to be wrong.",