From 06a281ec7e0b47ec41c137ad0994befd20512c19 Mon Sep 17 00:00:00 2001 From: Zomatree Date: Mon, 15 Sep 2025 01:44:21 +0100 Subject: [PATCH] docs: document `revolt-coalesced` --- crates/core/coalesced/Cargo.toml | 2 +- crates/core/coalesced/src/config.rs | 1 + crates/core/coalesced/src/error.rs | 15 +++- crates/core/coalesced/src/lib.rs | 34 ++++++- crates/core/coalesced/src/service.rs | 88 ++++++++++++++----- crates/services/gifbox/Cargo.toml | 11 ++- .../services/gifbox/src/routes/categories.rs | 1 + crates/services/gifbox/src/routes/root.rs | 2 +- crates/services/gifbox/src/routes/search.rs | 5 ++ crates/services/gifbox/src/routes/trending.rs | 20 +++-- crates/services/gifbox/src/tenor/mod.rs | 48 ++++------ crates/services/gifbox/src/tenor/types.rs | 2 + 12 files changed, 154 insertions(+), 75 deletions(-) diff --git a/crates/core/coalesced/Cargo.toml b/crates/core/coalesced/Cargo.toml index 4485d65d..cb23a8ad 100644 --- a/crates/core/coalesced/Cargo.toml +++ b/crates/core/coalesced/Cargo.toml @@ -11,7 +11,7 @@ tokio = ["dep:tokio"] queue = ["dep:indexmap"] cache = ["dep:lru"] -default = ["tokio", "queue", "cache"] +default = ["tokio"] [dependencies] tokio = { version = "1.47.0", features = ["sync"], optional = true } diff --git a/crates/core/coalesced/src/config.rs b/crates/core/coalesced/src/config.rs index 00ef7990..23da1698 100644 --- a/crates/core/coalesced/src/config.rs +++ b/crates/core/coalesced/src/config.rs @@ -1,4 +1,5 @@ #[derive(Clone, PartialEq, Eq, Debug)] +/// Config values for [`CoalescionService`]. pub struct CoalescionServiceConfig { /// How many tasks are running at once pub max_concurrent: Option, diff --git a/crates/core/coalesced/src/error.rs b/crates/core/coalesced/src/error.rs index a5395597..6be89785 100644 --- a/crates/core/coalesced/src/error.rs +++ b/crates/core/coalesced/src/error.rs @@ -1,18 +1,25 @@ -use std::fmt::Display; +use std::fmt; -#[derive(Clone, PartialEq, Eq, Debug)] +#[derive(Clone, Copy, PartialEq, Eq, Debug, Hash)] +/// Coalescion service error. pub enum Error { + /// Failed to receive the actions return from the channel for unknown reason RecvError, + /// Reached the `max_concurrent` amount of actions running at once and could not queue the action MaxConcurrent, + /// Reached the `max_queue` amount of actions in the queue MaxQueue, + /// Failed to downcast the type to the current type being returned, this will be most likely an ID collision + DowncastError, } -impl Display for Error { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { +impl fmt::Display for Error { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { Error::RecvError => write!(f, "Unable to receive data from the channel"), Error::MaxConcurrent => write!(f, "Max number of tasks running at once"), Error::MaxQueue => write!(f, "Max number of tasks in queue"), + Error::DowncastError => write!(f, "Failed to downcast type, possible key collision with different types") } } } diff --git a/crates/core/coalesced/src/lib.rs b/crates/core/coalesced/src/lib.rs index 6fd18267..89839e23 100644 --- a/crates/core/coalesced/src/lib.rs +++ b/crates/core/coalesced/src/lib.rs @@ -1,7 +1,39 @@ +//! # Coalesced +//! +//! Coalescion service to group, caching and queue duplicate actions. +//! useful for deduplicating web requests, database lookups and other similar resource +//! intensive or rate-limited actions. +//! +//! ## Features +//! - `tokio`: Uses tokio for the async backend, this is currently the only backend. +//! - `queue`: Whether to support queueing requests to only allow X amount of actions running at once. +//! - `cache`: Whether to cache the actions results for future actions with the same id, uses an LRU cache internally. +//! +//! [`CoalescionService`] uses both [`Arc`] and [`RwLock`] internally and can be cheaply cloned to +//! use in your codebase. +//! +//! It is common practice to wrap the service and in your own which delegates the executions to ensure all ids are tracked in one location across your codebase. +//! +//! All values are stored using [`Any`] and must be [`'static`] + [`Send`] + [`Sync`], if there is an id mismatch +//! and a type is wrong the library will return an error, values returned from the service are also +//! wrapped in an [`Arc`] as they are shared to each duplicate action. +//! +//! ## Example: +//! ```rs +//! use revolt_coalesced::CoalescionService; +//! +//! let service = CoalescionService::new(); +//! +//! let user_id = "my_user_id"; +//! let user = service.execute(user_id, || async move { +//! database.fetch_user(user_id).await.unwrap() +//! }).await; +//! ``` + mod config; mod error; mod service; pub use config::CoalescionServiceConfig; pub use error::Error; -pub use service::CoalescionService; \ No newline at end of file +pub use service::CoalescionService; diff --git a/crates/core/coalesced/src/service.rs b/crates/core/coalesced/src/service.rs index 3d3b1c9b..aadadc8a 100644 --- a/crates/core/coalesced/src/service.rs +++ b/crates/core/coalesced/src/service.rs @@ -1,6 +1,9 @@ -use std::{collections::{HashMap}, fmt::Debug, hash::Hash, sync::Arc, future::Future}; +use std::{any::Any, collections::HashMap, fmt::Debug, future::Future, hash::Hash, sync::Arc}; -use tokio::{sync::{watch::{channel as watch_channel, Receiver}, RwLock, Mutex}}; +use tokio::sync::{ + watch::{channel as watch_channel, Receiver}, + RwLock, +}; #[cfg(feature = "cache")] use lru::LruCache; @@ -10,20 +13,23 @@ use indexmap::IndexMap; use crate::{CoalescionServiceConfig, Error}; -#[derive(Clone, Debug)] +#[derive(Debug, Clone)] #[allow(clippy::type_complexity)] -pub struct CoalescionService { +/// # Coalescion service +/// +/// See module description for example usage. +pub struct CoalescionService { config: Arc, - watchers: Arc, Error>>>>>>, + watchers: Arc, Error>>>>>>, #[cfg(feature = "queue")] - queue: Arc, Error>>>>>>, + queue: Arc, Error>>>>>>, #[cfg(feature = "cache")] - cache: Option>>>>, + cache: Option>>>>, } -impl CoalescionService { +impl CoalescionService { pub fn new() -> Self { - Self::default() + Default::default() } pub fn from_config(config: CoalescionServiceConfig) -> Self { @@ -38,22 +44,37 @@ impl CoalescionService>) -> Self { + pub fn from_cache( + config: CoalescionServiceConfig, + cache: LruCache>, + ) -> Self { Self { cache: Some(Arc::new(Mutex::new(cache))), ..Self::from_config(config) } } - async fn wait_for(&self, mut receiver: Receiver, Error>>>) -> Result, Error> { + async fn wait_for( + &self, + mut receiver: Receiver, Error>>>, + ) -> Result, Error> { receiver .wait_for(|v| v.is_some()) .await .map_err(|_| Error::RecvError) .and_then(|r| r.clone().unwrap()) + .and_then(|arc| Arc::downcast(arc).map_err(|_| Error::DowncastError)) } - async fn insert_and_execute Fut, Fut: Future>(&self, id: Id, func: F) -> Result, Error> { + async fn insert_and_execute< + Value: Send + Sync + 'static, + F: FnOnce() -> Fut, + Fut: Future, + >( + &self, + id: Id, + func: F, + ) -> Result, Error> { let (send, recv) = watch_channel(None); self.watchers.write().await.insert(id.clone(), recv); @@ -61,7 +82,7 @@ impl CoalescionService)); }); #[cfg(feature = "cache")] @@ -76,11 +97,21 @@ impl CoalescionService Fut, Fut: Future>(&self, id: Id, func: F) -> Result, Error> { + /// Coalesces an function, the actual function may not run if one with the same id is already running, + /// queued to be ran, or cached, the id should be globally unique for this specific action. + pub async fn execute< + Value: Send + Sync + 'static, + F: FnOnce() -> Fut, + Fut: Future, + >( + &self, + id: Id, + func: F, + ) -> Result, Error> { #[cfg(feature = "cache")] if let Some(cache) = self.cache.as_ref() { if let Some(value) = cache.lock().await.get(&id) { - return Ok(value.clone()) + return Arc::downcast::(value.clone()).map_err(|_| Error::DowncastError); } }; @@ -105,10 +136,14 @@ impl CoalescionService= length) { - return Err(Error::MaxQueue) + if self + .config + .max_queue + .is_some_and(|max_queue| max_queue >= length) + { + return Err(Error::MaxQueue); }; let (send, recv) = watch_channel(None); @@ -130,10 +165,14 @@ impl CoalescionService), + ); }); - return response + return response; } } } @@ -145,23 +184,24 @@ impl CoalescionService { - self.insert_and_execute(id, func).await - } + _ => self.insert_and_execute(id, func).await, } } } + /// Fetches the amount of currently running tasks pub async fn current_task_count(&self) -> usize { self.watchers.read().await.len() } + #[cfg(feature = "queue")] + /// Fetches the current length of the queue pub async fn current_queue_len(&self) -> usize { self.queue.read().await.len() } } -impl Default for CoalescionService { +impl Default for CoalescionService { fn default() -> Self { Self::from_config(CoalescionServiceConfig::default()) } diff --git a/crates/services/gifbox/Cargo.toml b/crates/services/gifbox/Cargo.toml index 94cd0732..f67a04e8 100644 --- a/crates/services/gifbox/Cargo.toml +++ b/crates/services/gifbox/Cargo.toml @@ -22,8 +22,13 @@ revolt-result = { version = "0.8.8", path = "../../core/result", features = [ "utoipa", "axum", ] } -revolt-coalesced = { version = "0.8.8", path = "../../core/coalesced" } -revolt-database = { version = "0.8.8", path = "../../core/database", features = ["axum-impl"] } +revolt-coalesced = { version = "0.8.8", path = "../../core/coalesced", features = [ + "queue", +] } +revolt-database = { version = "0.8.8", path = "../../core/database", features = [ + "axum-impl", +] } + # Axum / web server axum = { version = "0.7.5" } axum-extra = { version = "0.9", features = ["typed-header"] } @@ -37,4 +42,4 @@ tracing = "0.1" # Utils lru_time_cache = "0.11.11" -urlencoding = "2.1.3" \ No newline at end of file +urlencoding = "2.1.3" diff --git a/crates/services/gifbox/src/routes/categories.rs b/crates/services/gifbox/src/routes/categories.rs index ae85126d..a79f03c0 100644 --- a/crates/services/gifbox/src/routes/categories.rs +++ b/crates/services/gifbox/src/routes/categories.rs @@ -11,6 +11,7 @@ use crate::{tenor, types}; #[derive(Deserialize, IntoParams)] pub struct CategoriesQueryParams { + /// Users locale #[param(example = "en_US")] pub locale: String, } diff --git a/crates/services/gifbox/src/routes/root.rs b/crates/services/gifbox/src/routes/root.rs index 36acb877..4bf3aed5 100644 --- a/crates/services/gifbox/src/routes/root.rs +++ b/crates/services/gifbox/src/routes/root.rs @@ -19,4 +19,4 @@ pub async fn root() -> Json> { message: "Gifbox lives on!", version: CRATE_VERSION, }) -} \ No newline at end of file +} diff --git a/crates/services/gifbox/src/routes/search.rs b/crates/services/gifbox/src/routes/search.rs index ec0e6c6e..ac541dc2 100644 --- a/crates/services/gifbox/src/routes/search.rs +++ b/crates/services/gifbox/src/routes/search.rs @@ -11,12 +11,17 @@ use crate::{tenor, types}; #[derive(Deserialize, IntoParams)] pub struct SearchQueryParams { + /// Search query #[param(example = "Wave")] pub query: String, + /// Users locale #[param(example = "en_US")] pub locale: String, + /// Amount of results to respond with pub limit: Option, + /// Flag for if searching in a gif category pub is_category: Option, + /// Value of `next` for getting the next page of results with the current search query pub position: Option, } diff --git a/crates/services/gifbox/src/routes/trending.rs b/crates/services/gifbox/src/routes/trending.rs index f08ed242..b8105524 100644 --- a/crates/services/gifbox/src/routes/trending.rs +++ b/crates/services/gifbox/src/routes/trending.rs @@ -4,20 +4,20 @@ use axum::{ }; use revolt_database::User; use revolt_result::{create_error, Result}; -use serde::{Deserialize}; -use utoipa::{IntoParams}; +use serde::Deserialize; +use utoipa::IntoParams; -use crate::{ - tenor, - types, -}; +use crate::{tenor, types}; #[derive(Deserialize, IntoParams)] pub struct TrendingQueryParams { #[param(example = "en_US")] + /// Users locale pub locale: String, + /// Amount of results to respond with pub limit: Option, - pub position: Option + /// Value of `next` for getting the next page of results of featured gifs + pub position: Option, } /// Trending GIFs @@ -37,7 +37,11 @@ pub async fn trending( State(tenor): State, ) -> Result> { tenor - .featured(¶ms.locale, params.limit.unwrap_or(50), params.position.as_deref().unwrap_or_default()) + .featured( + ¶ms.locale, + params.limit.unwrap_or(50), + params.position.as_deref().unwrap_or_default(), + ) .await .map_err(|_| create_error!(InternalError)) .map(|results| results.as_ref().clone().into()) diff --git a/crates/services/gifbox/src/tenor/mod.rs b/crates/services/gifbox/src/tenor/mod.rs index da9b97aa..c90b74ce 100644 --- a/crates/services/gifbox/src/tenor/mod.rs +++ b/crates/services/gifbox/src/tenor/mod.rs @@ -1,3 +1,5 @@ +//! Interal Tenor API wrapper + use std::{sync::Arc, time::Duration}; use lru_time_cache::LruCache; @@ -20,18 +22,11 @@ pub enum TenorError { pub struct Tenor { pub key: Arc, pub client: Client, - + pub coalescion: CoalescionService, pub cache: Arc>>>, - pub cache_coalescion: - CoalescionService, TenorError>>, pub categories: Arc>>>, - pub categories_coalescion: - CoalescionService, TenorError>>, - pub featured: Arc>>>, - pub featured_coalescion: - CoalescionService, TenorError>>, } impl Tenor { @@ -39,39 +34,29 @@ impl Tenor { Self { key: Arc::from(key), client: Client::new(), + coalescion: CoalescionService::from_config(CoalescionServiceConfig { + max_concurrent: Some(100), + queue_requests: true, + max_queue: None, + }), // 1 hour, 1k requests cache: Arc::new(RwLock::new(LruCache::with_expiry_duration_and_capacity( Duration::from_secs(60 * 60), 1000, ))), - cache_coalescion: CoalescionService::from_config(CoalescionServiceConfig { - max_concurrent: Some(100), - queue_requests: true, - max_queue: None, - }), // 1 day, 1k requests categories: Arc::new(RwLock::new(LruCache::with_expiry_duration_and_capacity( Duration::from_secs(60 * 60 * 24), 1000, ))), - categories_coalescion: CoalescionService::from_config(CoalescionServiceConfig { - max_concurrent: Some(100), - queue_requests: true, - max_queue: None, - }), // 1 day, 1k requests featured: Arc::new(RwLock::new(LruCache::with_expiry_duration_and_capacity( Duration::from_secs(60 * 60 * 24), 1000, ))), - featured_coalescion: CoalescionService::from_config(CoalescionServiceConfig { - max_concurrent: Some(100), - queue_requests: true, - max_queue: None, - }), } } @@ -110,7 +95,7 @@ impl Tenor { } } - let res = self.cache_coalescion.execute(unique_key.clone(), || { + let res = self.coalescion.execute(unique_key.clone(), || { let mut path = format!( "/search?key={}&q={}&client_key=Gifbox&media_filter=webm,tinywebm&locale={}&contentfilter=high&limit={limit}", &self.key, @@ -127,7 +112,7 @@ impl Tenor { path.push_str("&component=categories"); } - self.request(path) + self.request::(path) }) .await .unwrap(); @@ -152,7 +137,7 @@ impl Tenor { } let res = self - .categories_coalescion + .coalescion .execute(unique_key.clone(), || { let path = format!( "/categories?key={}&client_key=Gifbox&locale={}&contentfilter=high", @@ -160,7 +145,7 @@ impl Tenor { url_encode(locale), ); - self.request(path) + self.request::(path) }) .await .unwrap(); @@ -189,7 +174,7 @@ impl Tenor { } } - let res = self.featured_coalescion.execute(unique_key.clone(), || { + let res = self.coalescion.execute(unique_key.clone(), || { let mut path = format!( "/featured?key={}&client_key=Gifbox&media_filter=webm,tinywebm&locale={}&contentfilter=high&limit={limit}", &self.key, @@ -201,16 +186,13 @@ impl Tenor { path.push_str(position); }; - self.request(path) + self.request::(path) }) .await .unwrap(); if let Ok(resp) = &*res { - self.featured - .write() - .await - .insert(unique_key, resp.clone()); + self.featured.write().await.insert(unique_key, resp.clone()); } (*res).clone() diff --git a/crates/services/gifbox/src/tenor/types.rs b/crates/services/gifbox/src/tenor/types.rs index 8165e4b4..15658252 100644 --- a/crates/services/gifbox/src/tenor/types.rs +++ b/crates/services/gifbox/src/tenor/types.rs @@ -1,3 +1,5 @@ +//! Tenor API models + use std::collections::HashMap; use serde::{Deserialize, Serialize};