diff --git a/Cargo.lock b/Cargo.lock index b3eefef689..8c0ecd0b9a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -13443,12 +13443,13 @@ dependencies = [ "chrono", "dashmap", "deadpool-redis", + "eyre", "futures", "lz4_flex", + "postcard", "prometheus", "redis", "serde", - "serde_json", "thiserror 2.0.17", "tokio", "tracing", diff --git a/apps/labrinth/.sqlx/query-04c04958c71c4fab903c46c9185286e7460a6ff7b03cbc90939ac6c7cb526433.json b/apps/labrinth/.sqlx/query-214cc9257db904fd2cc68b8aa8d61d32cab210cfaea2205c95d9ca3bbd0ec1cd.json similarity index 68% rename from apps/labrinth/.sqlx/query-04c04958c71c4fab903c46c9185286e7460a6ff7b03cbc90939ac6c7cb526433.json rename to apps/labrinth/.sqlx/query-214cc9257db904fd2cc68b8aa8d61d32cab210cfaea2205c95d9ca3bbd0ec1cd.json index 6c62c2b6b6..f7c409af75 100644 --- a/apps/labrinth/.sqlx/query-04c04958c71c4fab903c46c9185286e7460a6ff7b03cbc90939ac6c7cb526433.json +++ b/apps/labrinth/.sqlx/query-214cc9257db904fd2cc68b8aa8d61d32cab210cfaea2205c95d9ca3bbd0ec1cd.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "\n SELECT id, enum_id, value, ordering, metadata, created FROM loader_field_enum_values\n WHERE enum_id = ANY($1)\n ORDER BY enum_id, ordering, created DESC\n ", + "query": "\n SELECT id, enum_id, value, ordering,\n metadata->>'type' AS \"ty?\",\n (metadata->>'major')::boolean AS \"major?\",\n created FROM loader_field_enum_values\n WHERE enum_id = ANY($1)\n ORDER BY enum_id, ordering, created DESC\n ", "describe": { "columns": [ { @@ -25,11 +25,16 @@ }, { "ordinal": 4, - "name": "metadata", - "type_info": "Jsonb" + "name": "ty?", + "type_info": "Text" }, { "ordinal": 5, + "name": "major?", + "type_info": "Bool" + }, + { + "ordinal": 6, "name": "created", "type_info": "Timestamptz" } @@ -44,9 +49,10 @@ false, false, true, - true, + null, + null, false ] }, - "hash": "04c04958c71c4fab903c46c9185286e7460a6ff7b03cbc90939ac6c7cb526433" + "hash": "214cc9257db904fd2cc68b8aa8d61d32cab210cfaea2205c95d9ca3bbd0ec1cd" } diff --git a/apps/labrinth/.sqlx/query-d9c4d536ce0bea290f445c3bccb56b4743f2f3a9ce4b170fb439e0e135ca9d51.json b/apps/labrinth/.sqlx/query-83b7543e426ae348e589b591f249358c2b9eac5323ae0795581358ed29e5ba65.json similarity index 58% rename from apps/labrinth/.sqlx/query-d9c4d536ce0bea290f445c3bccb56b4743f2f3a9ce4b170fb439e0e135ca9d51.json rename to apps/labrinth/.sqlx/query-83b7543e426ae348e589b591f249358c2b9eac5323ae0795581358ed29e5ba65.json index 7141f46a13..d0bc6abfc1 100644 --- a/apps/labrinth/.sqlx/query-d9c4d536ce0bea290f445c3bccb56b4743f2f3a9ce4b170fb439e0e135ca9d51.json +++ b/apps/labrinth/.sqlx/query-83b7543e426ae348e589b591f249358c2b9eac5323ae0795581358ed29e5ba65.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "\n SELECT DISTINCT id, enum_id, value, ordering, created, metadata\n FROM loader_field_enum_values lfev\n WHERE id = ANY($1)\n ORDER BY enum_id, ordering, created ASC\n ", + "query": "\n SELECT DISTINCT id, enum_id, value, ordering, created,\n metadata->>'type' AS \"ty?\",\n (metadata->>'major')::boolean AS \"major?\"\n FROM loader_field_enum_values lfev\n WHERE id = ANY($1)\n ORDER BY enum_id, ordering, created DESC\n ", "describe": { "columns": [ { @@ -30,8 +30,13 @@ }, { "ordinal": 5, - "name": "metadata", - "type_info": "Jsonb" + "name": "ty?", + "type_info": "Text" + }, + { + "ordinal": 6, + "name": "major?", + "type_info": "Bool" } ], "parameters": { @@ -45,8 +50,9 @@ false, true, false, - true + null, + null ] }, - "hash": "d9c4d536ce0bea290f445c3bccb56b4743f2f3a9ce4b170fb439e0e135ca9d51" + "hash": "83b7543e426ae348e589b591f249358c2b9eac5323ae0795581358ed29e5ba65" } diff --git a/apps/labrinth/.sqlx/query-fe34673ce6d7bcb616a5ab2e8900d7dfb4e0fa2ee640128d29d6e4beafe60f4c.json b/apps/labrinth/.sqlx/query-a8569122a309057326be0d49630119f81d17197673d831d210e351c149e41fc8.json similarity index 63% rename from apps/labrinth/.sqlx/query-fe34673ce6d7bcb616a5ab2e8900d7dfb4e0fa2ee640128d29d6e4beafe60f4c.json rename to apps/labrinth/.sqlx/query-a8569122a309057326be0d49630119f81d17197673d831d210e351c149e41fc8.json index 0af23b8554..2f1c614db8 100644 --- a/apps/labrinth/.sqlx/query-fe34673ce6d7bcb616a5ab2e8900d7dfb4e0fa2ee640128d29d6e4beafe60f4c.json +++ b/apps/labrinth/.sqlx/query-a8569122a309057326be0d49630119f81d17197673d831d210e351c149e41fc8.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "\n SELECT DISTINCT id, enum_id, value, ordering, created, metadata\n FROM loader_field_enum_values lfev\n ORDER BY enum_id, ordering, created DESC\n ", + "query": "\n SELECT DISTINCT id, enum_id, value, ordering, created,\n metadata->>'type' AS \"ty?\",\n (metadata->>'major')::boolean AS \"major?\"\n FROM loader_field_enum_values lfev\n ORDER BY enum_id, ordering, created DESC\n ", "describe": { "columns": [ { @@ -30,8 +30,13 @@ }, { "ordinal": 5, - "name": "metadata", - "type_info": "Jsonb" + "name": "ty?", + "type_info": "Text" + }, + { + "ordinal": 6, + "name": "major?", + "type_info": "Bool" } ], "parameters": { @@ -43,8 +48,9 @@ false, true, false, - true + null, + null ] }, - "hash": "fe34673ce6d7bcb616a5ab2e8900d7dfb4e0fa2ee640128d29d6e4beafe60f4c" + "hash": "a8569122a309057326be0d49630119f81d17197673d831d210e351c149e41fc8" } diff --git a/apps/labrinth/.sqlx/query-ab9b4b383ce8431214eb26abde734d36eff6902ae55b7ed4d89a10c4d69506a9.json b/apps/labrinth/.sqlx/query-ab9b4b383ce8431214eb26abde734d36eff6902ae55b7ed4d89a10c4d69506a9.json deleted file mode 100644 index 340549e517..0000000000 --- a/apps/labrinth/.sqlx/query-ab9b4b383ce8431214eb26abde734d36eff6902ae55b7ed4d89a10c4d69506a9.json +++ /dev/null @@ -1,50 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n SELECT l.id id, l.loader loader, l.icon icon, l.metadata metadata,\n ARRAY_AGG(DISTINCT pt.name) filter (where pt.name is not null) project_types,\n ARRAY_AGG(DISTINCT g.slug) filter (where g.slug is not null) games\n FROM loaders l\n LEFT OUTER JOIN loaders_project_types lpt ON joining_loader_id = l.id\n LEFT OUTER JOIN project_types pt ON lpt.joining_project_type_id = pt.id\n LEFT OUTER JOIN loaders_project_types_games lptg ON lptg.loader_id = lpt.joining_loader_id AND lptg.project_type_id = lpt.joining_project_type_id\n LEFT OUTER JOIN games g ON lptg.game_id = g.id\n GROUP BY l.id;\n ", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "id", - "type_info": "Int4" - }, - { - "ordinal": 1, - "name": "loader", - "type_info": "Varchar" - }, - { - "ordinal": 2, - "name": "icon", - "type_info": "Varchar" - }, - { - "ordinal": 3, - "name": "metadata", - "type_info": "Jsonb" - }, - { - "ordinal": 4, - "name": "project_types", - "type_info": "VarcharArray" - }, - { - "ordinal": 5, - "name": "games", - "type_info": "VarcharArray" - } - ], - "parameters": { - "Left": [] - }, - "nullable": [ - false, - false, - false, - false, - null, - null - ] - }, - "hash": "ab9b4b383ce8431214eb26abde734d36eff6902ae55b7ed4d89a10c4d69506a9" -} diff --git a/apps/labrinth/.sqlx/query-43d4eafdbcb449a56551d3d6edeba0d6e196fa6539e3f9df107c23a74ba962af.json b/apps/labrinth/.sqlx/query-ec9abad348739217eb887e2ac84901ba8dca63912e9d7054115052ad89ab4718.json similarity index 58% rename from apps/labrinth/.sqlx/query-43d4eafdbcb449a56551d3d6edeba0d6e196fa6539e3f9df107c23a74ba962af.json rename to apps/labrinth/.sqlx/query-ec9abad348739217eb887e2ac84901ba8dca63912e9d7054115052ad89ab4718.json index c66e860181..478c4dcd62 100644 --- a/apps/labrinth/.sqlx/query-43d4eafdbcb449a56551d3d6edeba0d6e196fa6539e3f9df107c23a74ba962af.json +++ b/apps/labrinth/.sqlx/query-ec9abad348739217eb887e2ac84901ba8dca63912e9d7054115052ad89ab4718.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "\n SELECT DISTINCT id, enum_id, value, ordering, created, metadata\n FROM loader_field_enum_values lfev\n WHERE id = ANY($1)\n ORDER BY enum_id, ordering, created DESC\n ", + "query": "\n SELECT DISTINCT id, enum_id, value, ordering, created,\n metadata->>'type' AS \"ty?\",\n (metadata->>'major')::boolean AS \"major?\"\n FROM loader_field_enum_values lfev\n WHERE id = ANY($1)\n ORDER BY enum_id, ordering, created ASC\n ", "describe": { "columns": [ { @@ -30,8 +30,13 @@ }, { "ordinal": 5, - "name": "metadata", - "type_info": "Jsonb" + "name": "ty?", + "type_info": "Text" + }, + { + "ordinal": 6, + "name": "major?", + "type_info": "Bool" } ], "parameters": { @@ -45,8 +50,9 @@ false, true, false, - true + null, + null ] }, - "hash": "43d4eafdbcb449a56551d3d6edeba0d6e196fa6539e3f9df107c23a74ba962af" + "hash": "ec9abad348739217eb887e2ac84901ba8dca63912e9d7054115052ad89ab4718" } diff --git a/apps/labrinth/.sqlx/query-f21bfe5bcc157d442180bf49c97654bd8650dcdd16cb9f74148bef6e932e2837.json b/apps/labrinth/.sqlx/query-f21bfe5bcc157d442180bf49c97654bd8650dcdd16cb9f74148bef6e932e2837.json new file mode 100644 index 0000000000..9fda28ebba --- /dev/null +++ b/apps/labrinth/.sqlx/query-f21bfe5bcc157d442180bf49c97654bd8650dcdd16cb9f74148bef6e932e2837.json @@ -0,0 +1,50 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT l.id id, l.loader loader, l.icon icon,\n (l.metadata->>'platform')::boolean AS platform,\n ARRAY_AGG(DISTINCT pt.name) filter (where pt.name is not null) project_types,\n ARRAY_AGG(DISTINCT g.slug) filter (where g.slug is not null) games\n FROM loaders l\n LEFT OUTER JOIN loaders_project_types lpt ON joining_loader_id = l.id\n LEFT OUTER JOIN project_types pt ON lpt.joining_project_type_id = pt.id\n LEFT OUTER JOIN loaders_project_types_games lptg ON lptg.loader_id = lpt.joining_loader_id AND lptg.project_type_id = lpt.joining_project_type_id\n LEFT OUTER JOIN games g ON lptg.game_id = g.id\n GROUP BY l.id;\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Int4" + }, + { + "ordinal": 1, + "name": "loader", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "icon", + "type_info": "Varchar" + }, + { + "ordinal": 3, + "name": "platform", + "type_info": "Bool" + }, + { + "ordinal": 4, + "name": "project_types", + "type_info": "VarcharArray" + }, + { + "ordinal": 5, + "name": "games", + "type_info": "VarcharArray" + } + ], + "parameters": { + "Left": [] + }, + "nullable": [ + false, + false, + false, + null, + null, + null + ] + }, + "hash": "f21bfe5bcc157d442180bf49c97654bd8650dcdd16cb9f74148bef6e932e2837" +} diff --git a/apps/labrinth/AGENTS.md b/apps/labrinth/AGENTS.md index 0c1e193f6f..3c11a2248b 100644 --- a/apps/labrinth/AGENTS.md +++ b/apps/labrinth/AGENTS.md @@ -12,8 +12,10 @@ - no trailing punctuation - wrap code items e.g. type names in backticks - Prefer `wrap_internal_err`, `wrap_request_err` when attaching context to an existing error (like Anyhow `context` or Eyre `wrap_err`) +- Prefer importing `eyre::Result` and using `Result` instead of `eyre::Result` +- Prefer `eyre::Ok(value)` instead of `Ok::<_, eyre::Report>(value)` when an explicit Eyre result type is needed - All operations should ideally have some context attached - - Database operations can have a message like `.wrap_internal_err("failed to fetch XYZ")` + - Database operations can have a message like `.wrap_internal_err("fetching XYZ")` - You can perform real-time queries against the databases in the Docker Compose - `docker exec labrinth-postgres psql -c "select 1"` - `docker exec labrinth-redis redis-cli flushall` diff --git a/apps/labrinth/src/auth/mod.rs b/apps/labrinth/src/auth/mod.rs index bf0a963f81..c58a032f13 100644 --- a/apps/labrinth/src/auth/mod.rs +++ b/apps/labrinth/src/auth/mod.rs @@ -59,12 +59,6 @@ pub enum AuthenticationError { Url, } -impl From for AuthenticationError { - fn from(error: xredis::Error) -> Self { - Self::Database(error.into()) - } -} - impl actix_web::ResponseError for AuthenticationError { fn status_code(&self) -> StatusCode { match self { diff --git a/apps/labrinth/src/database/models/analytics_event_item.rs b/apps/labrinth/src/database/models/analytics_event_item.rs index 268c0fc538..47493582aa 100644 --- a/apps/labrinth/src/database/models/analytics_event_item.rs +++ b/apps/labrinth/src/database/models/analytics_event_item.rs @@ -9,7 +9,7 @@ use crate::{ }; use serde::{Deserialize, Serialize}; -const ANALYTICS_EVENTS_NAMESPACE: &str = "analytics_events:v3"; +const ANALYTICS_EVENTS_NAMESPACE: &str = "analytics_events:v4"; const ANALYTICS_EVENTS_ALL_KEY: &str = "all"; #[derive(Debug, Clone, Serialize, Deserialize)] diff --git a/apps/labrinth/src/database/models/categories.rs b/apps/labrinth/src/database/models/categories.rs index edddeb7d21..c2b60ebeb8 100644 --- a/apps/labrinth/src/database/models/categories.rs +++ b/apps/labrinth/src/database/models/categories.rs @@ -7,7 +7,7 @@ use super::ids::*; use futures::TryStreamExt; use serde::{Deserialize, Serialize}; -const TAGS_NAMESPACE: &str = "tags:v3"; +const TAGS_NAMESPACE: &str = "tags:v4"; pub struct ProjectType { pub id: ProjectTypeId, diff --git a/apps/labrinth/src/database/models/collection_item.rs b/apps/labrinth/src/database/models/collection_item.rs index b2c53e1f21..6f89b75e71 100644 --- a/apps/labrinth/src/database/models/collection_item.rs +++ b/apps/labrinth/src/database/models/collection_item.rs @@ -8,7 +8,7 @@ use futures::TryStreamExt; use serde::{Deserialize, Serialize}; use xredis::RedisPool; -const COLLECTIONS_NAMESPACE: &str = "collections:v3"; +const COLLECTIONS_NAMESPACE: &str = "collections:v4"; #[derive(Clone)] pub struct CollectionBuilder { diff --git a/apps/labrinth/src/database/models/flow_item.rs b/apps/labrinth/src/database/models/flow_item.rs index 649da111be..7e4ad2bd4e 100644 --- a/apps/labrinth/src/database/models/flow_item.rs +++ b/apps/labrinth/src/database/models/flow_item.rs @@ -13,7 +13,7 @@ use url::Url; use webauthn_rs::prelude::{DiscoverableAuthentication, PasskeyRegistration}; use xredis::RedisPool; -const FLOWS_NAMESPACE: &str = "flows:v3"; +const FLOWS_NAMESPACE: &str = "flows:v4"; #[derive(Deserialize, Serialize)] pub enum DBFlow { diff --git a/apps/labrinth/src/database/models/image_item.rs b/apps/labrinth/src/database/models/image_item.rs index edb3d5add9..f3403ed7fc 100644 --- a/apps/labrinth/src/database/models/image_item.rs +++ b/apps/labrinth/src/database/models/image_item.rs @@ -6,7 +6,7 @@ use dashmap::DashMap; use serde::{Deserialize, Serialize}; use xredis::RedisPool; -const IMAGES_NAMESPACE: &str = "images:v3"; +const IMAGES_NAMESPACE: &str = "images:v4"; #[derive(Clone, Debug, Serialize, Deserialize)] pub struct DBImage { diff --git a/apps/labrinth/src/database/models/legacy_loader_fields.rs b/apps/labrinth/src/database/models/legacy_loader_fields.rs index a11a8a504f..f24291c25e 100644 --- a/apps/labrinth/src/database/models/legacy_loader_fields.rs +++ b/apps/labrinth/src/database/models/legacy_loader_fields.rs @@ -117,17 +117,8 @@ impl MinecraftGameVersion { id: loader_field_enum_value.id, version: loader_field_enum_value.value, created: loader_field_enum_value.created, - type_: loader_field_enum_value - .metadata - .get("type") - .and_then(|x| x.as_str()) - .map(|x| x.to_string()) - .unwrap_or_default(), - major: loader_field_enum_value - .metadata - .get("major") - .and_then(|x| x.as_bool()) - .unwrap_or_default(), + type_: loader_field_enum_value.ty.unwrap_or_default(), + major: loader_field_enum_value.major.unwrap_or_default(), } } } diff --git a/apps/labrinth/src/database/models/loader_fields.rs b/apps/labrinth/src/database/models/loader_fields.rs index a794812b25..bb599a1764 100644 --- a/apps/labrinth/src/database/models/loader_fields.rs +++ b/apps/labrinth/src/database/models/loader_fields.rs @@ -12,14 +12,14 @@ use itertools::Itertools; use serde::{Deserialize, Serialize}; use xredis::RedisPool; -const GAMES_LIST_NAMESPACE: &str = "games:v3"; -const LOADER_ID: &str = "loader_id:v3"; -const LOADERS_LIST_NAMESPACE: &str = "loaders:v3"; -const LOADER_FIELDS_NAMESPACE: &str = "loader_fields:v3"; -const LOADER_FIELDS_NAMESPACE_ALL: &str = "loader_fields_all:v3"; -const LOADER_FIELD_ENUMS_ID_NAMESPACE: &str = "loader_field_enums:v3"; +const GAMES_LIST_NAMESPACE: &str = "games:v4"; +const LOADER_ID: &str = "loader_id:v4"; +const LOADERS_LIST_NAMESPACE: &str = "loaders:v4"; +const LOADER_FIELDS_NAMESPACE: &str = "loader_fields:v4"; +const LOADER_FIELDS_NAMESPACE_ALL: &str = "loader_fields_all:v4"; +const LOADER_FIELD_ENUMS_ID_NAMESPACE: &str = "loader_field_enums:v4"; pub const LOADER_FIELD_ENUM_VALUES_NAMESPACE: &str = - "loader_field_enum_values:v3"; + "loader_field_enum_values:v4"; #[derive(Clone, Serialize, Deserialize, Debug)] pub struct Game { @@ -87,6 +87,11 @@ impl Game { } } +#[derive(Clone, Serialize, Deserialize)] +pub struct LoaderMetadata { + pub platform: Option, +} + #[derive(Serialize, Deserialize, Clone)] pub struct Loader { pub id: LoaderId, @@ -94,7 +99,7 @@ pub struct Loader { pub icon: String, pub supported_project_types: Vec, pub supported_games: Vec, // slugs - pub metadata: serde_json::Value, + pub metadata: LoaderMetadata, } impl Loader { @@ -154,7 +159,8 @@ impl Loader { let result = sqlx::query!( " - SELECT l.id id, l.loader loader, l.icon icon, l.metadata metadata, + SELECT l.id id, l.loader loader, l.icon icon, + (l.metadata->>'platform')::boolean AS platform, ARRAY_AGG(DISTINCT pt.name) filter (where pt.name is not null) project_types, ARRAY_AGG(DISTINCT g.slug) filter (where g.slug is not null) games FROM loaders l @@ -179,7 +185,9 @@ impl Loader { supported_games: x .games .unwrap_or_default(), - metadata: x.metadata + metadata: LoaderMetadata { + platform: x.platform, + }, }) .try_collect::>() .await?; @@ -277,8 +285,9 @@ pub struct LoaderFieldEnumValue { pub value: String, pub ordering: Option, pub created: DateTime, - #[serde(flatten)] - pub metadata: serde_json::Value, + #[serde(rename = "type")] + pub ty: Option, + pub major: Option, } impl std::hash::Hash for LoaderFieldEnumValue { @@ -357,7 +366,8 @@ pub struct QueryLoaderFieldEnumValue { pub value: String, pub ordering: Option, pub created: DateTime, - pub metadata: Option, + pub ty: Option, + pub major: Option, } impl LoaderField { @@ -612,42 +622,50 @@ impl LoaderFieldEnumValue { where E: crate::database::Executor<'a, Database = sqlx::Postgres>, { - let val = redis.get_cached_keys_raw( - LOADER_FIELD_ENUM_VALUES_NAMESPACE, - &loader_field_enum_ids.iter().map(|x| x.0).collect::>(), - |loader_field_enum_ids| async move { - let values = sqlx::query!( - " - SELECT id, enum_id, value, ordering, metadata, created FROM loader_field_enum_values + let val = redis + .get_cached_keys_raw( + LOADER_FIELD_ENUM_VALUES_NAMESPACE, + &loader_field_enum_ids + .iter() + .map(|x| x.0) + .collect::>(), + |loader_field_enum_ids| async move { + let values = sqlx::query!( + r#" + SELECT id, enum_id, value, ordering, + metadata->>'type' AS "ty?", + (metadata->>'major')::boolean AS "major?", + created FROM loader_field_enum_values WHERE enum_id = ANY($1) ORDER BY enum_id, ordering, created DESC - ", - &loader_field_enum_ids - ) + "#, + &loader_field_enum_ids + ) .fetch(exec) - .try_fold(DashMap::new(), |acc: DashMap>, c| { - let value = LoaderFieldEnumValue { - id: LoaderFieldEnumValueId(c.id), - enum_id: LoaderFieldEnumId(c.enum_id), - value: c.value, - ordering: c.ordering, - created: c.created, - metadata: c.metadata.unwrap_or_default(), - }; - - acc.entry(c.enum_id) - .or_default() - .push(value); + .try_fold( + DashMap::new(), + |acc: DashMap>, c| { + let value = LoaderFieldEnumValue { + id: LoaderFieldEnumValueId(c.id), + enum_id: LoaderFieldEnumId(c.enum_id), + value: c.value, + ordering: c.ordering, + created: c.created, + ty: c.ty, + major: c.major, + }; - async move { - Ok(acc) - } - }) + acc.entry(c.enum_id).or_default().push(value); + + async move { Ok(acc) } + }, + ) .await?; - Ok::<_, DatabaseError>(values) - }, - ).await?; + Ok::<_, DatabaseError>(values) + }, + ) + .await?; Ok(val .into_iter() @@ -669,15 +687,16 @@ impl LoaderFieldEnumValue { .await? .into_iter() .filter(|x| { - let mut bool = true; - for (key, value) in &filter { - if let Some(metadata_value) = x.metadata.get(key) { - bool &= metadata_value == value; - } else { - bool = false; + filter.iter().all(|(key, value)| match key.as_str() { + "type" => { + x.ty.as_deref() + .is_some_and(|type_| value.as_str() == Some(type_)) } - } - bool + "major" => x + .major + .is_some_and(|major| value.as_bool() == Some(major)), + _ => false, + }) }) .collect(); @@ -1170,10 +1189,8 @@ impl VersionFieldValue { value: lfev.value.clone(), ordering: lfev.ordering, created: lfev.created, - metadata: lfev - .metadata - .clone() - .unwrap_or_default(), + ty: lfev.ty.clone(), + major: lfev.major, } }), )) @@ -1249,10 +1266,8 @@ impl VersionFieldValue { value: lfev.value.clone(), ordering: lfev.ordering, created: lfev.created, - metadata: lfev - .metadata - .clone() - .unwrap_or_default(), + ty: lfev.ty.clone(), + major: lfev.major, }) }) .collect::>()?, diff --git a/apps/labrinth/src/database/models/mod.rs b/apps/labrinth/src/database/models/mod.rs index 09eddcaed7..badfad0bdb 100644 --- a/apps/labrinth/src/database/models/mod.rs +++ b/apps/labrinth/src/database/models/mod.rs @@ -77,8 +77,6 @@ pub enum DatabaseError { SerdeCacheError(#[from] serde_json::Error), #[error("error while encoding or decoding the cache: {0}")] PostcardCacheError(#[from] postcard::Error), - #[error(transparent)] - Redis(#[from] xredis::Error), #[error("Schema error: {0}")] SchemaError(String), } diff --git a/apps/labrinth/src/database/models/moderation_note_item.rs b/apps/labrinth/src/database/models/moderation_note_item.rs index 5f837b5da1..b0f14d175e 100644 --- a/apps/labrinth/src/database/models/moderation_note_item.rs +++ b/apps/labrinth/src/database/models/moderation_note_item.rs @@ -7,9 +7,9 @@ use xredis::RedisPool; use super::{DBOrganizationId, DBUserId, DatabaseError}; -const MODERATION_NOTES_USERS_NAMESPACE: &str = "moderation_notes_users:v3"; +const MODERATION_NOTES_USERS_NAMESPACE: &str = "moderation_notes_users:v4"; const MODERATION_NOTES_ORGANIZATIONS_NAMESPACE: &str = - "moderation_notes_organizations:v3"; + "moderation_notes_organizations:v4"; #[derive(Debug, Clone, Serialize, Deserialize)] pub struct DBModerationNote { diff --git a/apps/labrinth/src/database/models/notification_item.rs b/apps/labrinth/src/database/models/notification_item.rs index b97972c647..6421a6d0bf 100644 --- a/apps/labrinth/src/database/models/notification_item.rs +++ b/apps/labrinth/src/database/models/notification_item.rs @@ -10,7 +10,7 @@ use futures::TryStreamExt; use serde::{Deserialize, Serialize}; use xredis::RedisPool; -const USER_NOTIFICATIONS_NAMESPACE: &str = "user_notifications:v3"; +const USER_NOTIFICATIONS_NAMESPACE: &str = "user_notifications:v4"; pub struct NotificationBuilder { pub body: NotificationBody, diff --git a/apps/labrinth/src/database/models/notifications_template_item.rs b/apps/labrinth/src/database/models/notifications_template_item.rs index e5b7911561..dbe908e39e 100644 --- a/apps/labrinth/src/database/models/notifications_template_item.rs +++ b/apps/labrinth/src/database/models/notifications_template_item.rs @@ -1,14 +1,15 @@ use crate::database::models::DatabaseError; use crate::models::v3::notifications::{NotificationChannel, NotificationType}; use crate::routes::ApiError; +use crate::util::error::Context; use serde::{Deserialize, Serialize}; use xredis::RedisPool; -const TEMPLATES_NAMESPACE: &str = "notifications_templates:v3"; +const TEMPLATES_NAMESPACE: &str = "notifications_templates:v4"; const TEMPLATES_HTML_DATA_NAMESPACE: &str = - "notifications_templates_html_data:v3"; + "notifications_templates_html_data:v4"; const TEMPLATES_DYNAMIC_HTML_NAMESPACE: &str = - "notifications_templates_dynamic_html:v3"; + "notifications_templates_dynamic_html:v4"; const HTML_DATA_CACHE_EXPIRY: i64 = 60 * 15; // 15 minutes const TEMPLATES_CACHE_EXPIRY: i64 = 60 * 30; // 30 minutes @@ -123,12 +124,16 @@ where html: String, } - let mut redis_conn = redis.connect().await?; + let mut redis_conn = redis.connect().await.wrap_internal_err( + "connecting to redis for dynamic notification html", + )?; let redis_key = redis_conn .key() .metadata(TEMPLATES_DYNAMIC_HTML_NAMESPACE, key); - if let Some(body) = - redis_conn.get_deserialized::(&redis_key).await? + if let Some(body) = redis_conn + .get_deserialized::(&redis_key) + .await + .wrap_internal_err("fetching dynamic notification html from redis")? { return Ok(body.html); } @@ -136,14 +141,17 @@ where drop(redis_conn); let cached = HtmlBody { html: get().await? }; - let mut redis_conn = redis.connect().await?; + let mut redis_conn = redis.connect().await.wrap_internal_err( + "connecting to redis for dynamic notification html", + )?; let redis_key = redis_conn .key() .metadata(TEMPLATES_DYNAMIC_HTML_NAMESPACE, key); redis_conn .set_serialized(&redis_key, &cached, Some(HTML_DATA_CACHE_EXPIRY)) - .await?; + .await + .wrap_internal_err("writing dynamic notification html to redis")?; Ok(cached.html) } diff --git a/apps/labrinth/src/database/models/notifications_type_item.rs b/apps/labrinth/src/database/models/notifications_type_item.rs index 4663d90920..c2d036a3f7 100644 --- a/apps/labrinth/src/database/models/notifications_type_item.rs +++ b/apps/labrinth/src/database/models/notifications_type_item.rs @@ -3,7 +3,7 @@ use crate::models::v3::notifications::NotificationType; use serde::{Deserialize, Serialize}; use xredis::RedisPool; -const NOTIFICATION_TYPES_NAMESPACE: &str = "notification_types:v3"; +const NOTIFICATION_TYPES_NAMESPACE: &str = "notification_types:v4"; #[derive(Serialize, Deserialize)] pub struct NotificationTypeItem { diff --git a/apps/labrinth/src/database/models/organization_item.rs b/apps/labrinth/src/database/models/organization_item.rs index 2cb83b6fc3..845ff1235d 100644 --- a/apps/labrinth/src/database/models/organization_item.rs +++ b/apps/labrinth/src/database/models/organization_item.rs @@ -9,8 +9,8 @@ use xredis::RedisPool; use super::{DBTeamMember, ids::*}; use serde::{Deserialize, Serialize}; -const ORGANIZATIONS_NAMESPACE: &str = "organizations:v3"; -const ORGANIZATIONS_TITLES_NAMESPACE: &str = "organizations_titles:v3"; +const ORGANIZATIONS_NAMESPACE: &str = "organizations:v4"; +const ORGANIZATIONS_TITLES_NAMESPACE: &str = "organizations_titles:v4"; #[derive(Deserialize, Serialize, Clone, Debug)] /// An organization of users who together control one or more projects and organizations. diff --git a/apps/labrinth/src/database/models/pat_item.rs b/apps/labrinth/src/database/models/pat_item.rs index 50a05c84c7..deb6f4a0b2 100644 --- a/apps/labrinth/src/database/models/pat_item.rs +++ b/apps/labrinth/src/database/models/pat_item.rs @@ -11,9 +11,9 @@ use std::fmt::{Debug, Display}; use std::hash::Hash; use xredis::RedisPool; -const PATS_NAMESPACE: &str = "pats:v3"; -const PATS_TOKENS_NAMESPACE: &str = "pats_tokens:v3"; -const PATS_USERS_NAMESPACE: &str = "pats_users:v3"; +const PATS_NAMESPACE: &str = "pats:v4"; +const PATS_TOKENS_NAMESPACE: &str = "pats_tokens:v4"; +const PATS_USERS_NAMESPACE: &str = "pats_users:v4"; #[derive(Deserialize, Serialize, Clone, Debug)] pub struct DBPersonalAccessToken { diff --git a/apps/labrinth/src/database/models/product_item.rs b/apps/labrinth/src/database/models/product_item.rs index 266eafa773..a1f4cc291e 100644 --- a/apps/labrinth/src/database/models/product_item.rs +++ b/apps/labrinth/src/database/models/product_item.rs @@ -9,7 +9,7 @@ use std::convert::TryFrom; use std::convert::TryInto; use xredis::RedisPool; -const PRODUCTS_NAMESPACE: &str = "products:v3"; +const PRODUCTS_NAMESPACE: &str = "products:v4"; pub struct DBProduct { pub id: DBProductId, @@ -136,7 +136,7 @@ pub struct QueryProductWithPrices { pub id: DBProductId, pub metadata: ProductMetadata, pub unitary: bool, - #[serde(skip_serializing_if = "Option::is_none", default)] + #[serde(default)] pub name: Option, pub prices: Vec, } diff --git a/apps/labrinth/src/database/models/project_item.rs b/apps/labrinth/src/database/models/project_item.rs index e862cf76f3..3b85483843 100644 --- a/apps/labrinth/src/database/models/project_item.rs +++ b/apps/labrinth/src/database/models/project_item.rs @@ -23,9 +23,9 @@ use std::fmt::{Debug, Display}; use std::hash::Hash; use xredis::RedisPool; -pub const PROJECTS_NAMESPACE: &str = "projects:v3"; -pub const PROJECTS_SLUGS_NAMESPACE: &str = "projects_slugs:v3"; -const PROJECTS_DEPENDENCIES_NAMESPACE: &str = "projects_dependencies:v3"; +pub const PROJECTS_NAMESPACE: &str = "projects:v4"; +pub const PROJECTS_SLUGS_NAMESPACE: &str = "projects_slugs:v4"; +const PROJECTS_DEPENDENCIES_NAMESPACE: &str = "projects_dependencies:v4"; #[derive(Clone, Debug, Serialize, Deserialize)] pub struct LinkUrl { @@ -657,12 +657,14 @@ impl DBProject { .await?; let loader_field_enum_values: Vec = sqlx::query!( - " - SELECT DISTINCT id, enum_id, value, ordering, created, metadata + r#" + SELECT DISTINCT id, enum_id, value, ordering, created, + metadata->>'type' AS "ty?", + (metadata->>'major')::boolean AS "major?" FROM loader_field_enum_values lfev WHERE id = ANY($1) ORDER BY enum_id, ordering, created DESC - ", + "#, &loader_field_enum_value_ids .iter() .map(|x| x.0) @@ -675,7 +677,8 @@ impl DBProject { value: m.value, ordering: m.ordering, created: m.created, - metadata: m.metadata, + ty: m.ty, + major: m.major, }) .try_collect() .await?; @@ -946,7 +949,7 @@ impl DBProject { }, ) .await - .wrap_internal_err("failed to fetch cached projects")?; + .wrap_internal_err("fetching cached projects")?; Ok(val) } @@ -1053,6 +1056,5 @@ pub struct ProjectQueryResult { pub gallery_items: Vec, pub thread_id: DBThreadId, pub aggregate_version_fields: Vec, - #[serde(flatten)] pub components: exp::ProjectQuery, } diff --git a/apps/labrinth/src/database/models/session_item.rs b/apps/labrinth/src/database/models/session_item.rs index 9d58dd4f12..9ca8ae560d 100644 --- a/apps/labrinth/src/database/models/session_item.rs +++ b/apps/labrinth/src/database/models/session_item.rs @@ -10,9 +10,9 @@ use std::fmt::{Debug, Display}; use std::hash::Hash; use xredis::RedisPool; -const SESSIONS_NAMESPACE: &str = "sessions:v3"; -const SESSIONS_IDS_NAMESPACE: &str = "sessions_ids:v3"; -const SESSIONS_USERS_NAMESPACE: &str = "sessions_users:v3"; +const SESSIONS_NAMESPACE: &str = "sessions:v4"; +const SESSIONS_IDS_NAMESPACE: &str = "sessions_ids:v4"; +const SESSIONS_USERS_NAMESPACE: &str = "sessions_users:v4"; pub struct SessionBuilder { pub session: String, diff --git a/apps/labrinth/src/database/models/team_item.rs b/apps/labrinth/src/database/models/team_item.rs index 698c9d08a5..4246e60d7d 100644 --- a/apps/labrinth/src/database/models/team_item.rs +++ b/apps/labrinth/src/database/models/team_item.rs @@ -10,7 +10,7 @@ use rust_decimal::Decimal; use serde::{Deserialize, Serialize}; use xredis::RedisPool; -const TEAMS_NAMESPACE: &str = "teams:v3"; +const TEAMS_NAMESPACE: &str = "teams:v4"; pub struct TeamBuilder { pub members: Vec, diff --git a/apps/labrinth/src/database/models/user_item.rs b/apps/labrinth/src/database/models/user_item.rs index 416d1bad18..a5efc5e701 100644 --- a/apps/labrinth/src/database/models/user_item.rs +++ b/apps/labrinth/src/database/models/user_item.rs @@ -16,9 +16,9 @@ use std::fmt::{Debug, Display}; use std::hash::Hash; use xredis::RedisPool; -const USERS_NAMESPACE: &str = "users:v3"; -const USER_USERNAMES_NAMESPACE: &str = "users_usernames:v3"; -const USERS_PROJECTS_NAMESPACE: &str = "users_projects:v3"; +const USERS_NAMESPACE: &str = "users:v4"; +const USER_USERNAMES_NAMESPACE: &str = "users_usernames:v4"; +const USERS_PROJECTS_NAMESPACE: &str = "users_projects:v4"; #[derive(Deserialize, Serialize, Clone, Debug)] pub struct DBUser { diff --git a/apps/labrinth/src/database/models/version_item.rs b/apps/labrinth/src/database/models/version_item.rs index 096bed6614..f16d5212c9 100644 --- a/apps/labrinth/src/database/models/version_item.rs +++ b/apps/labrinth/src/database/models/version_item.rs @@ -21,8 +21,8 @@ use std::cmp::Ordering; use std::collections::HashMap; use tracing::error; -pub const VERSIONS_NAMESPACE: &str = "versions:v3"; -const VERSION_FILES_NAMESPACE: &str = "versions_files:v3"; +pub const VERSIONS_NAMESPACE: &str = "versions:v4"; +const VERSION_FILES_NAMESPACE: &str = "versions_files:v4"; pub async fn cleanup_unused_attribution_files_and_groups( transaction: &mut PgTransaction<'_>, @@ -704,12 +704,14 @@ impl DBVersion { .await?; let loader_field_enum_values: Vec = sqlx::query!( - " - SELECT DISTINCT id, enum_id, value, ordering, created, metadata + r#" + SELECT DISTINCT id, enum_id, value, ordering, created, + metadata->>'type' AS "ty?", + (metadata->>'major')::boolean AS "major?" FROM loader_field_enum_values lfev WHERE id = ANY($1) ORDER BY enum_id, ordering, created ASC - ", + "#, &loader_field_enum_value_ids .iter() .map(|x| x.0) @@ -722,7 +724,8 @@ impl DBVersion { value: m.value, ordering: m.ordering, created: m.created, - metadata: m.metadata, + ty: m.ty, + major: m.major, }) .try_collect() .await?; @@ -1090,7 +1093,6 @@ pub struct VersionQueryResult { pub project_types: Vec, pub games: Vec, pub dependencies: Vec, - #[serde(flatten)] pub components: exp::VersionQuery, } diff --git a/apps/labrinth/src/database/redis.rs b/apps/labrinth/src/database/redis.rs deleted file mode 100644 index 0bc5f37231..0000000000 --- a/apps/labrinth/src/database/redis.rs +++ /dev/null @@ -1,56 +0,0 @@ -use std::sync::Arc; - -use crate::env::ENV; - -struct RedisConfig { - inner: xredis::RedisConfig, - cache_settings: xredis::CacheSettings, -} - -impl RedisConfig { - fn from_env() -> Result { - let inner = xredis::RedisConfig::new( - ENV.REDIS_TOPOLOGY, - ENV.REDIS_CONNECTION_TYPE, - &ENV.REDIS_URL, - ENV.REDIS_WAIT_TIMEOUT_MS, - ( - ENV.REDIS_MAX_CONNECTIONS as usize, - ENV.REDIS_MIN_CONNECTIONS, - ), - ( - ENV.REDIS_CLUSTER_MAX_CONNECTIONS as usize, - ENV.REDIS_CLUSTER_MIN_CONNECTIONS, - ), - (ENV.REDIS_BLOCKING_MAX_CONNECTIONS as usize, 0), - ENV.REDIS_CACHE_LOCKING_STRATEGY, - ENV.REDIS_READ_REPLICA_STRATEGY, - )?; - let cache_settings = xredis::CacheSettings { - default_expiry: ENV.REDIS_DEFAULT_EXPIRY, - actual_expiry: ENV.REDIS_ACTUAL_EXPIRY, - version_default_expiry: ENV.REDIS_VERSION_DEFAULT_EXPIRY, - version_actual_expiry: ENV.REDIS_VERSION_ACTUAL_EXPIRY, - encoding_format: ENV.REDIS_ENCODING_FORMAT, - compression_algorithm: ENV.REDIS_COMPRESSION_ALGORITHM, - compression_level: ENV.REDIS_COMPRESSION_LEVEL, - compression_threshold_bytes: ENV.REDIS_COMPRESSION_THRESHOLD_BYTES, - compression_min_savings_ratio: ENV - .REDIS_COMPRESSION_MIN_SAVINGS_RATIO, - }; - - Ok(Self { - inner, - cache_settings, - }) - } -} - -pub async fn from_env( - meta_namespace: impl Into>, -) -> xredis::RedisPool { - let config = RedisConfig::from_env().expect("invalid Redis configuration"); - xredis::RedisPool::new(meta_namespace, config.inner, config.cache_settings) - .await - .expect("failed to initialize Redis connections") -} diff --git a/apps/labrinth/src/database/redis/mod.rs b/apps/labrinth/src/database/redis/mod.rs new file mode 100644 index 0000000000..575bf91867 --- /dev/null +++ b/apps/labrinth/src/database/redis/mod.rs @@ -0,0 +1,413 @@ +pub(crate) mod serde; + +use std::sync::Arc; + +use crate::env::ENV; +use eyre::{Result, WrapErr}; + +struct RedisConfig { + inner: xredis::RedisConfig, + cache_settings: xredis::CacheSettings, +} + +impl RedisConfig { + fn from_env() -> Result { + let inner = xredis::RedisConfig::new( + ENV.REDIS_TOPOLOGY, + ENV.REDIS_CONNECTION_TYPE, + &ENV.REDIS_URL, + ENV.REDIS_WAIT_TIMEOUT_MS, + ( + ENV.REDIS_MAX_CONNECTIONS as usize, + ENV.REDIS_MIN_CONNECTIONS, + ), + ( + ENV.REDIS_CLUSTER_MAX_CONNECTIONS as usize, + ENV.REDIS_CLUSTER_MIN_CONNECTIONS, + ), + (ENV.REDIS_BLOCKING_MAX_CONNECTIONS as usize, 0), + ENV.REDIS_CACHE_LOCKING_STRATEGY, + ENV.REDIS_READ_REPLICA_STRATEGY, + ) + .wrap_err("loading Redis configuration from environment")?; + let cache_settings = xredis::CacheSettings { + default_expiry: ENV.REDIS_DEFAULT_EXPIRY, + actual_expiry: ENV.REDIS_ACTUAL_EXPIRY, + version_default_expiry: ENV.REDIS_VERSION_DEFAULT_EXPIRY, + version_actual_expiry: ENV.REDIS_VERSION_ACTUAL_EXPIRY, + encoding_format: ENV.REDIS_ENCODING_FORMAT, + compression_algorithm: ENV.REDIS_COMPRESSION_ALGORITHM, + compression_level: ENV.REDIS_COMPRESSION_LEVEL, + compression_threshold_bytes: ENV.REDIS_COMPRESSION_THRESHOLD_BYTES, + compression_min_savings_ratio: ENV + .REDIS_COMPRESSION_MIN_SAVINGS_RATIO, + }; + + Ok(Self { + inner, + cache_settings, + }) + } +} + +pub async fn from_env( + meta_namespace: impl Into>, +) -> xredis::RedisPool { + let config = RedisConfig::from_env().expect("invalid Redis configuration"); + xredis::RedisPool::new(meta_namespace, config.inner, config.cache_settings) + .await + .expect("failed to initialize Redis connections") +} + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + + use ::serde::{Serialize, de::DeserializeOwned}; + use chrono::Utc; + use url::Url; + use uuid::Uuid; + use webauthn_rs::WebauthnBuilder; + + use crate::database::models::flow_item::DBFlow; + use crate::database::models::ids::{ + DBNotificationId, DBProjectId, DBTeamId, DBThreadId, DBUserId, + DBVersionId, LoaderFieldEnumId, LoaderFieldEnumValueId, LoaderFieldId, + LoaderId, + }; + use crate::database::models::loader_fields::{ + Loader, LoaderFieldEnumValue, LoaderMetadata, VersionField, + VersionFieldValue, + }; + use crate::database::models::notification_item::DBNotification; + use crate::database::models::project_item::{ + DBProject, ProjectQueryResult, + }; + use crate::database::models::version_item::{ + DBVersion, VersionQueryResult, + }; + use crate::models::billing::{Price, ProductMetadata}; + use crate::models::exp::{self, minecraft}; + use crate::models::notifications::NotificationBody; + use crate::models::projects::{ + MonetizationStatus, ProjectStatus, SideTypesMigrationReviewStatus, + VersionStatus, + }; + + fn postcard_round_trip(value: &T) -> T + where + T: Serialize + DeserializeOwned, + { + let serialized = + postcard::to_allocvec(value).expect("serializing with postcard"); + postcard::from_bytes(&serialized).expect("deserializing with postcard") + } + + fn loader_field_enum_value(ty: &str, major: bool) -> LoaderFieldEnumValue { + LoaderFieldEnumValue { + id: LoaderFieldEnumValueId(1), + enum_id: LoaderFieldEnumId(2), + value: "1.21.8".to_string(), + ordering: None, + created: Utc::now(), + ty: Some(ty.to_string()), + major: Some(major), + } + } + + fn db_project() -> DBProject { + let now = Utc::now(); + + DBProject { + id: DBProjectId(1), + team_id: DBTeamId(2), + organization_id: None, + name: "project".to_string(), + summary: "summary".to_string(), + description: "description".to_string(), + published: now, + updated: now, + approved: Some(now), + queued: None, + status: ProjectStatus::Approved, + requested_status: None, + downloads: 3, + follows: 4, + icon_url: None, + raw_icon_url: None, + license_url: None, + license: "MIT".to_string(), + slug: Some("project".to_string()), + moderation_message: None, + moderation_message_body: None, + webhook_sent: false, + color: None, + monetization_status: MonetizationStatus::Monetized, + side_types_migration_review_status: + SideTypesMigrationReviewStatus::Reviewed, + loaders: vec!["fabric".to_string()], + components: exp::ProjectSerial::default(), + } + } + + fn db_version() -> DBVersion { + DBVersion { + id: DBVersionId(1), + project_id: DBProjectId(2), + author_id: DBUserId(3), + name: "version".to_string(), + version_number: "1.0.0".to_string(), + changelog: "changelog".to_string(), + date_published: Utc::now(), + downloads: 4, + version_type: "release".to_string(), + featured: true, + status: VersionStatus::Listed, + requested_status: None, + ordering: None, + components: exp::VersionSerial::default(), + } + } + + #[test] + fn loader_metadata_round_trips_with_postcard() { + for platform in [None, Some(false), Some(true)] { + let metadata = LoaderMetadata { platform }; + let round_tripped = postcard_round_trip(&metadata); + + assert_eq!(round_tripped.platform, platform); + } + } + + #[test] + fn loader_round_trips_with_postcard() { + for platform in [None, Some(false), Some(true)] { + let loader = Loader { + id: LoaderId(1), + loader: "paper".to_string(), + icon: "icon".to_string(), + supported_project_types: vec!["plugin".to_string()], + supported_games: vec!["minecraft-java".to_string()], + metadata: LoaderMetadata { platform }, + }; + let round_tripped = postcard_round_trip(&loader); + + assert_eq!(round_tripped.id, loader.id); + assert_eq!(round_tripped.loader, loader.loader); + assert_eq!(round_tripped.icon, loader.icon); + assert_eq!( + round_tripped.supported_project_types, + loader.supported_project_types + ); + assert_eq!(round_tripped.supported_games, loader.supported_games); + assert_eq!(round_tripped.metadata.platform, platform); + } + } + + #[test] + fn loader_field_enum_values_round_trip_with_postcard() { + let metadata_values = [ + (None, None), + (Some("snapshot"), Some(false)), + (Some("alpha"), Some(false)), + (Some("beta"), Some(true)), + (Some("release"), Some(true)), + (Some("beta"), Some(false)), + (Some("release"), Some(false)), + ]; + + for (ty, major) in metadata_values { + let enum_value = LoaderFieldEnumValue { + id: LoaderFieldEnumValueId(1), + enum_id: LoaderFieldEnumId(2), + value: "1.21.8".to_string(), + ordering: None, + created: Utc::now(), + ty: ty.map(str::to_string), + major, + }; + + assert_eq!(postcard_round_trip(&enum_value), enum_value); + } + } + + #[test] + fn loader_field_enum_value_keeps_flattened_json_layout() { + let enum_value = loader_field_enum_value("release", true); + let json = serde_json::to_value(&enum_value) + .expect("serializing loader field enum value as JSON"); + + assert_eq!(json.get("type"), Some(&serde_json::json!("release"))); + assert_eq!(json.get("major"), Some(&serde_json::json!(true))); + assert!(json.get("metadata").is_none()); + assert_eq!( + serde_json::from_value::(json) + .expect("deserializing loader field enum value from JSON"), + enum_value + ); + } + + #[test] + fn non_enum_version_fields_round_trip_with_postcard() { + let values = [ + VersionFieldValue::Integer(5), + VersionFieldValue::Text("value".to_string()), + VersionFieldValue::Boolean(true), + VersionFieldValue::ArrayInteger(vec![1, 2]), + VersionFieldValue::ArrayText(vec!["one".to_string()]), + VersionFieldValue::ArrayBoolean(vec![true, false]), + ]; + + for value in values { + let field = VersionField { + version_id: DBVersionId(1), + field_id: LoaderFieldId(2), + field_name: "field".to_string(), + value, + }; + + assert_eq!(postcard_round_trip(&field), field); + } + } + + #[test] + fn flattened_cache_values_round_trip_with_postcard() { + let enum_value = loader_field_enum_value("release", true); + postcard_round_trip(&enum_value); + + let version = VersionQueryResult { + inner: db_version(), + files: Vec::new(), + version_fields: vec![VersionField { + version_id: DBVersionId(1), + field_id: LoaderFieldId(2), + field_name: "game_versions".to_string(), + value: VersionFieldValue::Enum( + LoaderFieldEnumId(2), + enum_value.clone(), + ), + }], + loaders: vec!["fabric".to_string()], + project_types: vec!["mod".to_string()], + games: vec!["minecraft-java".to_string()], + dependencies: Vec::new(), + components: exp::VersionQuery::default(), + }; + postcard_round_trip(&version); + + let project = ProjectQueryResult { + inner: db_project(), + categories: Vec::new(), + additional_categories: Vec::new(), + versions: vec![DBVersionId(1)], + project_types: vec!["mod".to_string()], + games: vec!["minecraft-java".to_string()], + urls: Vec::new(), + gallery_items: Vec::new(), + thread_id: DBThreadId(1), + aggregate_version_fields: Vec::new(), + components: exp::ProjectQuery::default(), + }; + postcard_round_trip(&project); + } + + #[test] + fn skipped_cache_fields_round_trip_with_postcard() { + postcard_round_trip(&exp::ProjectSerial::default()); + postcard_round_trip(&exp::ProjectQuery::default()); + postcard_round_trip(&db_project()); + } + + #[test] + fn internally_tagged_cache_values_round_trip_with_postcard() { + for metadata in [ + ProductMetadata::Midas, + ProductMetadata::Pyro { + cpu: 1, + ram: 2, + swap: 3, + storage: 4, + }, + ProductMetadata::Medal { + cpu: 1, + ram: 2, + swap: 3, + storage: 4, + region: "us-east".to_string(), + }, + ] { + postcard_round_trip(&metadata); + } + + for price in [ + Price::OneTime { price: 500 }, + Price::Recurring { + intervals: HashMap::new(), + }, + ] { + postcard_round_trip(&price); + } + + postcard_round_trip(&NotificationBody::TwoFactorEnabled); + postcard_round_trip(&DBNotification { + id: DBNotificationId(1), + user_id: DBUserId(2), + body: NotificationBody::TwoFactorEnabled, + read: false, + created: Utc::now(), + }); + postcard_round_trip(&minecraft::ServerContent::Vanilla { + supported_game_versions: vec!["1.21.8".to_string()], + recommended_game_version: Some("1.21.8".to_string()), + }); + postcard_round_trip(&minecraft::ServerContentQuery::Vanilla { + supported_game_versions: vec!["1.21.8".to_string()], + recommended_game_version: Some("1.21.8".to_string()), + }); + } + + #[test] + fn simple_flow_variants_round_trip_with_postcard() { + assert!(matches!( + postcard_round_trip(&DBFlow::MinecraftAuth), + DBFlow::MinecraftAuth + )); + + let flow = postcard_round_trip(&DBFlow::Login2FA { + user_id: DBUserId(1), + }); + assert!(matches!( + flow, + DBFlow::Login2FA { + user_id: DBUserId(1) + } + )); + } + + #[test] + fn passkey_flow_variants_round_trip_with_postcard() { + let origin = Url::parse("https://example.com").unwrap(); + let webauthn = WebauthnBuilder::new("example.com", &origin) + .unwrap() + .build() + .unwrap(); + let (_, registration) = webauthn + .start_passkey_registration( + Uuid::from_u128(1), + "user@example.com", + "user", + None, + ) + .unwrap(); + postcard_round_trip(&DBFlow::RegisterPasskey { + user_id: DBUserId(1), + state: registration, + }); + + let (_, authentication) = + webauthn.start_discoverable_authentication().unwrap(); + postcard_round_trip(&DBFlow::AuthenticatePasskey { + state: authentication, + }); + } +} diff --git a/apps/labrinth/src/database/redis/serde.rs b/apps/labrinth/src/database/redis/serde.rs new file mode 100644 index 0000000000..fcf68708f5 --- /dev/null +++ b/apps/labrinth/src/database/redis/serde.rs @@ -0,0 +1,70 @@ +macro_rules! impl_redis_serde { + ( + $type:ty, + human = $human:ident, + binary = $binary:ident, + schema $(,)? + ) => { + $crate::database::redis::serde::impl_redis_serde!( + $type, + human = $human, + binary = $binary, + ); + + impl ::utoipa::PartialSchema for $type { + fn schema() + -> ::utoipa::openapi::RefOr<::utoipa::openapi::schema::Schema> { + <$human as ::utoipa::PartialSchema>::schema() + } + } + + impl ::utoipa::ToSchema for $type { + fn schemas( + schemas: &mut ::std::vec::Vec<( + ::std::string::String, + ::utoipa::openapi::RefOr<::utoipa::openapi::schema::Schema>, + )>, + ) { + <$human as ::utoipa::ToSchema>::schemas(schemas); + } + } + }; + ( + $type:ty, + human = $human:ident, + binary = $binary:ident $(,)? + ) => { + impl ::serde::Serialize for $type { + fn serialize( + &self, + serializer: S, + ) -> ::std::result::Result + where + S: ::serde::Serializer, + { + if serializer.is_human_readable() { + $human::serialize(self, serializer) + } else { + $binary::serialize(self, serializer) + } + } + } + + impl<'de> ::serde::Deserialize<'de> for $type { + fn deserialize( + deserializer: D, + ) -> ::std::result::Result + where + D: ::serde::Deserializer<'de>, + { + if deserializer.is_human_readable() { + $human::deserialize(deserializer) + } else { + $binary::deserialize(deserializer) + } + } + } + }; +} + +pub(crate) use impl_redis_serde; diff --git a/apps/labrinth/src/env.rs b/apps/labrinth/src/env.rs index 20ce98f546..c5d63a7f6b 100644 --- a/apps/labrinth/src/env.rs +++ b/apps/labrinth/src/env.rs @@ -165,7 +165,7 @@ vars! { REDIS_BLOCKING_MAX_CONNECTIONS: u32 = 256u32; // The encoding format used for Redis cache values. - REDIS_ENCODING_FORMAT: xredis::EncodingFormat = xredis::EncodingFormat::Json; + REDIS_ENCODING_FORMAT: xredis::EncodingFormat = xredis::EncodingFormat::Postcard; // The level of LZ4 compression used for Redis cache values. A value of 0 disables compression (supports 1-12) REDIS_COMPRESSION_LEVEL: i32 = 0i32; // The compression algorithm used for Redis cache values. Currently only LZ4 is supported. diff --git a/apps/labrinth/src/models/exp/minecraft.rs b/apps/labrinth/src/models/exp/minecraft.rs index 3b8fb5508c..7882932419 100644 --- a/apps/labrinth/src/models/exp/minecraft.rs +++ b/apps/labrinth/src/models/exp/minecraft.rs @@ -324,8 +324,7 @@ impl ComponentEdit for JavaServerProjectEdit { } /// What game content a [`JavaServerProject`] is using. -#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)] -#[serde(tag = "kind", rename_all = "snake_case")] +#[derive(Debug, Clone)] pub enum ServerContent { /// Server runs modded content with a modpack found on the Modrinth platform. Modpack { @@ -345,9 +344,41 @@ pub enum ServerContent { }, } +const _: () = { + #[derive(Deserialize, Serialize, utoipa::ToSchema)] + #[serde(remote = "ServerContent", tag = "kind", rename_all = "snake_case")] + enum HumanReadableProxy { + Modpack { + version_id: VersionId, + }, + Vanilla { + supported_game_versions: Vec, + recommended_game_version: Option, + }, + } + + #[derive(Deserialize, Serialize)] + #[serde(remote = "ServerContent")] + enum BinaryProxy { + Modpack { + version_id: VersionId, + }, + Vanilla { + supported_game_versions: Vec, + recommended_game_version: Option, + }, + } + + crate::database::redis::serde::impl_redis_serde!( + ServerContent, + human = HumanReadableProxy, + binary = BinaryProxy, + schema, + ); +}; + /// What game content a [`JavaServerProject`] is using. -#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)] -#[serde(tag = "kind", rename_all = "snake_case")] +#[derive(Debug, Clone)] pub enum ServerContentQuery { /// Server runs modded content with a modpack found on the Modrinth platform. Modpack { @@ -366,6 +397,49 @@ pub enum ServerContentQuery { }, } +const _: () = { + #[derive(Deserialize, Serialize, utoipa::ToSchema)] + #[serde( + remote = "ServerContentQuery", + tag = "kind", + rename_all = "snake_case" + )] + enum HumanReadableProxy { + Modpack { + version_id: VersionId, + project_id: ProjectId, + project_name: String, + project_icon: String, + }, + Vanilla { + supported_game_versions: Vec, + recommended_game_version: Option, + }, + } + + #[derive(Deserialize, Serialize)] + #[serde(remote = "ServerContentQuery")] + enum BinaryProxy { + Modpack { + version_id: VersionId, + project_id: ProjectId, + project_name: String, + project_icon: String, + }, + Vanilla { + supported_game_versions: Vec, + recommended_game_version: Option, + }, + } + + crate::database::redis::serde::impl_redis_serde!( + ServerContentQuery, + human = HumanReadableProxy, + binary = BinaryProxy, + schema, + ); +}; + impl Default for ServerContent { fn default() -> Self { ServerContent::Vanilla { diff --git a/apps/labrinth/src/models/exp/project.rs b/apps/labrinth/src/models/exp/project.rs index 46694b2ec8..4dcc06fcab 100644 --- a/apps/labrinth/src/models/exp/project.rs +++ b/apps/labrinth/src/models/exp/project.rs @@ -54,7 +54,7 @@ macro_rules! define_project_components { pub struct ProjectSerial { $( #[validate(nested)] - #[serde(default, skip_serializing_if = "Option::is_none")] + #[serde(default)] pub $field_name: Option<$ty>, )* } @@ -114,7 +114,6 @@ macro_rules! define_project_components { #[derive(Debug, Clone, Default, Serialize, Deserialize, utoipa::ToSchema)] pub struct ProjectQuery { $( - #[serde(skip_serializing_if = "Option::is_none")] pub $field_name: Option>, )* } diff --git a/apps/labrinth/src/models/v3/billing.rs b/apps/labrinth/src/models/v3/billing.rs index c3ef13facb..c03d3d35df 100644 --- a/apps/labrinth/src/models/v3/billing.rs +++ b/apps/labrinth/src/models/v3/billing.rs @@ -14,8 +14,6 @@ pub struct Product { pub unitary: bool, } -#[derive(Serialize, Deserialize)] -#[serde(tag = "type", rename_all = "kebab-case")] pub enum ProductMetadata { Midas, Pyro { @@ -33,6 +31,56 @@ pub enum ProductMetadata { }, } +const _: () = { + #[derive(Deserialize, Serialize)] + #[serde( + remote = "ProductMetadata", + tag = "type", + rename_all = "kebab-case" + )] + enum HumanReadableProxy { + Midas, + Pyro { + cpu: u32, + ram: u32, + swap: u32, + storage: u32, + }, + Medal { + cpu: u32, + ram: u32, + swap: u32, + storage: u32, + region: String, + }, + } + + #[derive(Deserialize, Serialize)] + #[serde(remote = "ProductMetadata")] + enum BinaryProxy { + Midas, + Pyro { + cpu: u32, + ram: u32, + swap: u32, + storage: u32, + }, + Medal { + cpu: u32, + ram: u32, + swap: u32, + storage: u32, + region: String, + }, + } + + crate::database::redis::serde::impl_redis_serde!( + ProductMetadata, + human = HumanReadableProxy, + binary = BinaryProxy, + ); +}; + impl ProductMetadata { pub fn is_pyro(&self) -> bool { matches!(self, ProductMetadata::Pyro { .. }) @@ -55,8 +103,6 @@ pub struct ProductPrice { pub currency_code: String, } -#[derive(Serialize, Deserialize)] -#[serde(tag = "type", rename_all = "kebab-case")] pub enum Price { OneTime { price: i32, @@ -66,6 +112,36 @@ pub enum Price { }, } +const _: () = { + #[derive(Deserialize, Serialize)] + #[serde(remote = "Price", tag = "type", rename_all = "kebab-case")] + enum HumanReadableProxy { + OneTime { + price: i32, + }, + Recurring { + intervals: HashMap, + }, + } + + #[derive(Deserialize, Serialize)] + #[serde(remote = "Price")] + enum BinaryProxy { + OneTime { + price: i32, + }, + Recurring { + intervals: HashMap, + }, + } + + crate::database::redis::serde::impl_redis_serde!( + Price, + human = HumanReadableProxy, + binary = BinaryProxy, + ); +}; + impl Price { pub fn get_interval(&self, interval: PriceDuration) -> Option { match self { diff --git a/apps/labrinth/src/models/v3/notifications.rs b/apps/labrinth/src/models/v3/notifications.rs index 6a655958f9..7f5d056306 100644 --- a/apps/labrinth/src/models/v3/notifications.rs +++ b/apps/labrinth/src/models/v3/notifications.rs @@ -151,8 +151,7 @@ impl NotificationType { } } -#[derive(Serialize, Deserialize, Clone)] -#[serde(tag = "type", rename_all = "snake_case")] +#[derive(Clone)] pub enum NotificationBody { ProjectUpdate { project_id: ProjectId, @@ -284,6 +283,157 @@ pub enum NotificationBody { Unknown, } +const _: () = { + macro_rules! define_proxy { + ($name:ident, $($attributes:meta),+ $(,)?) => { + #[derive(Deserialize, Serialize)] + $(#[$attributes])+ + enum $name { + ProjectUpdate { + project_id: ProjectId, + version_id: VersionId, + }, + TeamInvite { + project_id: ProjectId, + team_id: TeamId, + invited_by: UserId, + role: String, + }, + OrganizationInvite { + organization_id: OrganizationId, + invited_by: UserId, + team_id: TeamId, + role: String, + }, + ServerInvite { + server_id: Uuid, + server_name: String, + invited_by: UserId, + role: String, + }, + SharedInstanceInvite { + shared_instance_id: String, + shared_instance_name: String, + shared_instance_icon: Option, + invited_by: UserId, + }, + StatusChange { + project_id: ProjectId, + old_status: ProjectStatus, + new_status: ProjectStatus, + }, + ModeratorMessage { + thread_id: ThreadId, + message_id: ThreadMessageId, + project_id: Option, + report_id: Option, + }, + PatCreated { + token_name: String, + }, + ModerationMessageReceived { + project_id: ProjectId, + }, + ReportStatusUpdated { + report_id: ReportId, + }, + ReportSubmitted { + report_id: ReportId, + }, + ProjectStatusApproved { + project_id: ProjectId, + }, + ProjectStatusNeutral { + project_id: ProjectId, + old_status: ProjectStatus, + new_status: ProjectStatus, + }, + ProjectTransferred { + project_id: ProjectId, + new_owner_user_id: Option, + new_owner_organization_id: Option, + }, + LegacyMarkdown { + notification_type: Option, + name: String, + text: String, + link: String, + actions: Vec, + }, + ResetPassword { + flow: String, + }, + VerifyEmail { + flow: String, + }, + AuthProviderAdded { + provider: String, + }, + AuthProviderRemoved { + provider: String, + }, + TwoFactorEnabled, + TwoFactorRemoved, + PasswordChanged, + PasswordRemoved, + EmailChanged { + new_email: String, + to_email: String, + }, + SubscriptionCredited { + subscription_id: UserSubscriptionId, + days: i32, + previous_due: DateTime, + next_due: DateTime, + header_message: Option, + }, + PaymentFailed { + amount: String, + service: String, + }, + TaxNotification { + subscription_id: UserSubscriptionId, + new_amount: i64, + new_tax_amount: i64, + old_amount: i64, + old_tax_amount: i64, + billing_interval: PriceDuration, + currency: String, + due: DateTime, + service: String, + }, + PayoutAvailable { + date_available: DateTime, + amount: u64, + }, + DiscordRoleCreatorClub, + Custom { + key: String, + title: String, + body_md: String, + }, + Unknown, + } + }; + } + + define_proxy!( + HumanReadableProxy, + serde( + remote = "NotificationBody", + tag = "type", + rename_all = "snake_case" + ), + ); + define_proxy!(BinaryProxy, serde(remote = "NotificationBody")); + + crate::database::redis::serde::impl_redis_serde!( + NotificationBody, + human = HumanReadableProxy, + binary = BinaryProxy, + ); +}; + impl NotificationBody { pub fn notification_type(&self) -> NotificationType { match &self { diff --git a/apps/labrinth/src/queue/analytics/cache.rs b/apps/labrinth/src/queue/analytics/cache.rs index 386cbdaad5..a82c69be29 100644 --- a/apps/labrinth/src/queue/analytics/cache.rs +++ b/apps/labrinth/src/queue/analytics/cache.rs @@ -12,7 +12,7 @@ use crate::{ routes::analytics::MINECRAFT_SERVER_PLAYS, util::error::Context, }; -pub const MINECRAFT_SERVER_ANALYTICS: &str = "minecraft_server_analytics:v3"; +pub const MINECRAFT_SERVER_ANALYTICS: &str = "minecraft_server_analytics:v4"; #[derive(Debug, Clone, Serialize, Deserialize)] pub struct MinecraftServerAnalytics { diff --git a/apps/labrinth/src/queue/analytics/mod.rs b/apps/labrinth/src/queue/analytics/mod.rs index 18766e48da..91c3f553d2 100644 --- a/apps/labrinth/src/queue/analytics/mod.rs +++ b/apps/labrinth/src/queue/analytics/mod.rs @@ -4,6 +4,7 @@ use crate::models::analytics::{ }; use crate::routes::ApiError; use crate::routes::analytics::MINECRAFT_SERVER_PLAYS; +use crate::util::error::Context; use dashmap::{DashMap, DashSet}; use std::collections::HashMap; use tracing::trace; @@ -11,9 +12,9 @@ use xredis::RedisPool; pub mod cache; -const DOWNLOADS_NAMESPACE: &str = "downloads:v3"; -const VIEWS_NAMESPACE: &str = "views:v3"; -const MINECRAFT_SERVER_PLAYS_NAMESPACE: &str = "minecraft_server_plays:v3"; +const DOWNLOADS_NAMESPACE: &str = "downloads:v4"; +const VIEWS_NAMESPACE: &str = "views:v4"; +const MINECRAFT_SERVER_PLAYS_NAMESPACE: &str = "minecraft_server_plays:v4"; const MINECRAFT_SERVER_PLAYS_EXPIRY: u64 = 86_400; // 24 hours const MINECRAFT_SERVER_PLAYS_LIMIT: u32 = 5; @@ -142,10 +143,15 @@ impl AnalyticsQueue { ) }) .collect::>(); - let mut redis_connection = redis.connect().await?; - - let results = - redis_connection.get_many_typed::(&redis_keys).await?; + let mut redis_connection = + redis.connect().await.wrap_internal_err( + "connecting to redis for server play counts", + )?; + + let results = redis_connection + .get_many_typed::(&redis_keys) + .await + .wrap_internal_err("fetching server play counts from redis")?; for (idx, count) in results.into_iter().enumerate() { let new_count = if let Some(count) = count { if count >= MINECRAFT_SERVER_PLAYS_LIMIT { @@ -164,7 +170,8 @@ impl AnalyticsQueue { new_count, Some(MINECRAFT_SERVER_PLAYS_EXPIRY as i64), ) - .await?; + .await + .wrap_internal_err("writing server play count to redis")?; } let mut plays = client @@ -198,10 +205,15 @@ impl AnalyticsQueue { ) }) .collect::>(); - let mut redis_connection = redis.connect().await?; - - let results = - redis_connection.get_many_typed::(&redis_keys).await?; + let mut redis_connection = redis + .connect() + .await + .wrap_internal_err("connecting to redis for view counts")?; + + let results = redis_connection + .get_many_typed::(&redis_keys) + .await + .wrap_internal_err("fetching view counts from redis")?; for (idx, count) in results.into_iter().enumerate() { let new_count = if let Some((views, monetized)) = raw_views.get_mut(idx) { @@ -226,7 +238,8 @@ impl AnalyticsQueue { let key = &redis_keys[idx]; redis_connection .set(key, new_count, Some(6 * 60 * 60)) - .await?; + .await + .wrap_internal_err("writing view count to redis")?; } let mut views = client.insert::("views").await?; @@ -267,10 +280,15 @@ impl AnalyticsQueue { ) }) .collect::>(); - let mut redis_connection = redis.connect().await?; - - let results = - redis_connection.get_many_typed::(&redis_keys).await?; + let mut redis_connection = redis + .connect() + .await + .wrap_internal_err("connecting to redis for download counts")?; + + let results = redis_connection + .get_many_typed::(&redis_keys) + .await + .wrap_internal_err("fetching download counts from redis")?; for (idx, count) in results.into_iter().enumerate() { let new_count = if let Some(count) = count { if count > 5 { @@ -286,7 +304,8 @@ impl AnalyticsQueue { let key = &redis_keys[idx]; redis_connection .set(key, new_count, Some(6 * 60 * 60)) - .await?; + .await + .wrap_internal_err("writing download count to redis")?; } let mut transaction = pool.begin().await?; diff --git a/apps/labrinth/src/queue/server_ping.rs b/apps/labrinth/src/queue/server_ping.rs index 75e8728260..070095c0b7 100644 --- a/apps/labrinth/src/queue/server_ping.rs +++ b/apps/labrinth/src/queue/server_ping.rs @@ -26,9 +26,9 @@ pub struct ServerPingQueue { pub incremental_search_queue: IncrementalSearchQueue, } -pub const REDIS_NAMESPACE: &str = "minecraft_java_server_ping:v3"; +pub const REDIS_NAMESPACE: &str = "minecraft_java_server_ping:v4"; pub const REDIS_FAILURE_NAMESPACE: &str = - "minecraft_java_server_ping_failures:v3"; + "minecraft_java_server_ping_failures:v4"; pub const CLICKHOUSE_TABLE: &str = "minecraft_java_server_pings"; impl ServerPingQueue { diff --git a/apps/labrinth/src/routes/internal/campaign.rs b/apps/labrinth/src/routes/internal/campaign.rs index b440558cc5..5da7ebc414 100644 --- a/apps/labrinth/src/routes/internal/campaign.rs +++ b/apps/labrinth/src/routes/internal/campaign.rs @@ -68,7 +68,7 @@ pub struct CampaignInfo { cached_at: DateTime, } -const CAMPAIGN_INFO_CACHE_NAMESPACE: &str = "campaign_info:v3"; +const CAMPAIGN_INFO_CACHE_NAMESPACE: &str = "campaign_info:v4"; const CAMPAIGN_INFO_CACHE_STALE_SECONDS: i64 = 15 * 60; const CAMPAIGN_INFO_CACHE_TTL_SECONDS: i64 = 24 * 60 * 60; diff --git a/apps/labrinth/src/routes/internal/flows.rs b/apps/labrinth/src/routes/internal/flows.rs index e9effeb083..fd70c60c55 100644 --- a/apps/labrinth/src/routes/internal/flows.rs +++ b/apps/labrinth/src/routes/internal/flows.rs @@ -2213,7 +2213,7 @@ async fn validate_2fa_code( ) .map_err(|_| AuthenticationError::InvalidCredentials)?; - const TOTP_NAMESPACE: &str = "used_totp:v3"; + const TOTP_NAMESPACE: &str = "used_totp:v4"; let mut conn = redis.connect().await?; let logical_key = format!("{}-{}", input, user_id.0); let key = redis diff --git a/apps/labrinth/src/routes/mod.rs b/apps/labrinth/src/routes/mod.rs index fc75849d22..e9e606deba 100644 --- a/apps/labrinth/src/routes/mod.rs +++ b/apps/labrinth/src/routes/mod.rs @@ -267,12 +267,6 @@ pub enum ApiError { }, } -impl From for ApiError { - fn from(error: xredis::Error) -> Self { - Self::Database(error.into()) - } -} - impl ApiError { pub fn delphi(err: impl Into) -> Self { Self::Delphi(err.into()) diff --git a/apps/labrinth/src/routes/v2/tags.rs b/apps/labrinth/src/routes/v2/tags.rs index 6f3ef45283..56a86ca46d 100644 --- a/apps/labrinth/src/routes/v2/tags.rs +++ b/apps/labrinth/src/routes/v2/tags.rs @@ -210,18 +210,9 @@ pub async fn game_version_list( .into_iter() .map(|f| GameVersionQueryData { version: f.value, - version_type: f - .metadata - .get("type") - .and_then(|m| m.as_str()) - .unwrap_or_default() - .to_string(), + version_type: f.ty.unwrap_or_default(), date: f.created, - major: f - .metadata - .get("major") - .and_then(|m| m.as_bool()) - .unwrap_or_default(), + major: f.major.unwrap_or_default(), }) .collect::>(); HttpResponse::Ok().json(fields) diff --git a/apps/labrinth/src/routes/v3/content/mod.rs b/apps/labrinth/src/routes/v3/content/mod.rs index 41807babd7..4d8cda3b91 100644 --- a/apps/labrinth/src/routes/v3/content/mod.rs +++ b/apps/labrinth/src/routes/v3/content/mod.rs @@ -23,8 +23,8 @@ use sha2::{Digest, Sha256}; use std::collections::BTreeMap; use xredis::RedisPool; -const CONTENT_RESOLVE_CACHE_NAMESPACE: &str = "content_resolve:v3"; -const CONTENT_RESOLVE_CACHE_HEAT_NAMESPACE: &str = "content_resolve_heat:v3"; +const CONTENT_RESOLVE_CACHE_NAMESPACE: &str = "content_resolve:v4"; +const CONTENT_RESOLVE_CACHE_HEAT_NAMESPACE: &str = "content_resolve_heat:v4"; const CONTENT_RESOLVE_CACHE_SCHEMA_VERSION: &str = "v3"; const CONTENT_RESOLVE_CACHE_HEAT_WINDOW_SECONDS: i64 = 60 * 60 * 24; diff --git a/apps/labrinth/src/routes/v3/tags.rs b/apps/labrinth/src/routes/v3/tags.rs index 6a499df5df..4c1589cbbc 100644 --- a/apps/labrinth/src/routes/v3/tags.rs +++ b/apps/labrinth/src/routes/v3/tags.rs @@ -6,6 +6,7 @@ use crate::database::models::categories::{ }; use crate::database::models::loader_fields::{ Game, Loader, LoaderField, LoaderFieldEnumValue, LoaderFieldType, + LoaderMetadata, }; use actix_web::{HttpResponse, get, web}; use xredis::RedisPool; @@ -103,7 +104,7 @@ pub struct LoaderData { pub supported_project_types: Vec, pub supported_games: Vec, pub supported_fields: Vec, // Available loader fields for this loader - pub metadata: Value, + pub metadata: LoaderMetadata, } #[utoipa::path(tag = "tags", responses((status = OK)))] diff --git a/apps/labrinth/src/search/indexing.rs b/apps/labrinth/src/search/indexing.rs index 5180bfea9c..0abadfd2bf 100644 --- a/apps/labrinth/src/search/indexing.rs +++ b/apps/labrinth/src/search/indexing.rs @@ -394,11 +394,13 @@ async fn build_search_documents( let loader_field_enum_values: Vec = sqlx::query!( - " - SELECT DISTINCT id, enum_id, value, ordering, created, metadata + r#" + SELECT DISTINCT id, enum_id, value, ordering, created, + metadata->>'type' AS "ty?", + (metadata->>'major')::boolean AS "major?" FROM loader_field_enum_values lfev ORDER BY enum_id, ordering, created DESC - " + "# ) .fetch(pool) .map_ok(|m| QueryLoaderFieldEnumValue { @@ -407,7 +409,8 @@ async fn build_search_documents( value: m.value, ordering: m.ordering, created: m.created, - metadata: m.metadata, + ty: m.ty, + major: m.major, }) .try_collect() .await?; diff --git a/apps/labrinth/src/sync/friends.rs b/apps/labrinth/src/sync/friends.rs index a7327f38eb..32169677e8 100644 --- a/apps/labrinth/src/sync/friends.rs +++ b/apps/labrinth/src/sync/friends.rs @@ -14,7 +14,7 @@ use redis::{RedisWrite, ToRedisArgs, ToSingleRedisArg}; use serde::{Deserialize, Serialize}; use tokio::sync::mpsc; -pub const FRIENDS_CHANNEL_NAME: &str = "friends:v3"; +pub const FRIENDS_CHANNEL_NAME: &str = "friends:v4"; #[derive(Serialize, Deserialize)] pub enum RedisFriendsMessage { diff --git a/apps/labrinth/src/sync/status.rs b/apps/labrinth/src/sync/status.rs index 33f6bc3881..98375dccb7 100644 --- a/apps/labrinth/src/sync/status.rs +++ b/apps/labrinth/src/sync/status.rs @@ -5,7 +5,7 @@ use redis::AsyncCommands; use xredis::RedisPool; const EXPIRY_TIME_SECONDS: i64 = 60; -const USER_STATUS_NAMESPACE: &str = "user_status:v3"; +const USER_STATUS_NAMESPACE: &str = "user_status:v4"; pub async fn get_user_status( user: UserId, diff --git a/apps/labrinth/src/util/gotenberg.rs b/apps/labrinth/src/util/gotenberg.rs index 64d971b801..0c1a3248b0 100644 --- a/apps/labrinth/src/util/gotenberg.rs +++ b/apps/labrinth/src/util/gotenberg.rs @@ -14,7 +14,7 @@ pub const MODRINTH_GENERATED_PDF_TYPE: HeaderName = HeaderName::from_static("modrinth-generated-pdf-type"); pub const MODRINTH_PAYMENT_ID: HeaderName = HeaderName::from_static("modrinth-payment-id"); -pub const PAYMENT_STATEMENTS_NAMESPACE: &str = "payment_statements:v3"; +pub const PAYMENT_STATEMENTS_NAMESPACE: &str = "payment_statements:v4"; const REDIS_TIMEOUT_MARGIN_MS: u64 = 250; pub(crate) fn payment_statement_key( diff --git a/apps/labrinth/src/util/ratelimit.rs b/apps/labrinth/src/util/ratelimit.rs index 9088b6ad50..3bac6cabe8 100644 --- a/apps/labrinth/src/util/ratelimit.rs +++ b/apps/labrinth/src/util/ratelimit.rs @@ -12,7 +12,7 @@ use std::str::FromStr; use std::sync::Arc; use xredis::RedisPool; -const RATE_LIMIT_NAMESPACE: &str = "rate_limit:v3"; +const RATE_LIMIT_NAMESPACE: &str = "rate_limit:v4"; const RATE_LIMIT_EXPIRY: i64 = 300; // 5 minutes const MINUTE_IN_NANOS: i64 = 60_000_000_000; diff --git a/apps/labrinth/tests/loader_fields.rs b/apps/labrinth/tests/loader_fields.rs index 53a86bfac5..750fd18e91 100644 --- a/apps/labrinth/tests/loader_fields.rs +++ b/apps/labrinth/tests/loader_fields.rs @@ -574,12 +574,7 @@ async fn minecraft_game_version_update() { // A couple specific checks- in the dummy data, all game versions are marked as major=false except 1.20.5 let name_to_major = game_versions .iter() - .map(|x| { - ( - x.value.clone(), - x.metadata.get("major").unwrap().as_bool().unwrap(), - ) - }) + .map(|x| (x.value.clone(), x.major.unwrap())) .collect::>(); for (name, major) in name_to_major { if name == "1.20.5" { @@ -612,12 +607,7 @@ async fn minecraft_game_version_update() { let name_to_major = game_versions .iter() - .map(|x| { - ( - x.value.clone(), - x.metadata.get("major").unwrap().as_bool().unwrap(), - ) - }) + .map(|x| (x.value.clone(), x.major.unwrap())) .collect::>(); // Confirm that the new version is there assert!(name_to_major.contains_key("1.20.6")); diff --git a/apps/labrinth/tests/redis.rs b/apps/labrinth/tests/redis.rs index c812868fcf..bbd38fea94 100644 --- a/apps/labrinth/tests/redis.rs +++ b/apps/labrinth/tests/redis.rs @@ -22,7 +22,7 @@ use serde_json::json; use tokio::sync::{Barrier, Notify}; use tokio::time::timeout; use uuid::Uuid; -use xredis::{KeyBuilder, RedisPool, RedisTopology}; +use xredis::{KeyBuilder, RedisPool, RedisTopology, RedisValue}; pub mod common; @@ -250,7 +250,7 @@ async fn cache_lock_coalesces_concurrent_misses_for_one_key() { tasks.push(tokio::spawn(async move { barrier.wait().await; pool.get_cached_keys_raw( - "single_flight:v3", + "single_flight:v4", &["shared".to_string()], move |keys| async move { fetch_count.fetch_add(1, Ordering::SeqCst); @@ -292,7 +292,7 @@ async fn cache_lock_coalesces_only_overlapping_keys() { tasks.push(tokio::spawn(async move { barrier.wait().await; pool.get_cached_keys_raw( - "overlapping_locks:v3", + "overlapping_locks:v4", &requested, move |keys| async move { tokio::time::sleep(Duration::from_millis(75)).await; @@ -333,7 +333,7 @@ async fn cache_lock_does_not_block_independent_keys() { let slow = tokio::spawn(async move { slow_pool .get_cached_keys_raw( - "independent_locks:v3", + "independent_locks:v4", &["slow".to_string()], move |keys| async move { slow_started.notify_one(); @@ -350,7 +350,7 @@ async fn cache_lock_does_not_block_independent_keys() { let fast = timeout( Duration::from_secs(1), pool.get_cached_keys_raw( - "independent_locks:v3", + "independent_locks:v4", &["fast".to_string()], |keys| async move { let values = DashMap::new(); @@ -379,7 +379,7 @@ async fn cache_lock_is_released_after_error_and_cancellation() { let failed = pool .get_cached_keys_raw( - "error_recovery:v3", + "error_recovery:v4", &["key".to_string()], |_| async { Err::, _>(DatabaseError::Internal( @@ -393,7 +393,7 @@ async fn cache_lock_is_released_after_error_and_cancellation() { let recovered = timeout( Duration::from_secs(1), pool.get_cached_keys_raw( - "error_recovery:v3", + "error_recovery:v4", &["key".to_string()], |keys| async move { let values = DashMap::new(); @@ -413,7 +413,7 @@ async fn cache_lock_is_released_after_error_and_cancellation() { let cancelled = tokio::spawn(async move { cancelled_pool .get_cached_keys_raw( - "cancellation_recovery:v3", + "cancellation_recovery:v4", &["key".to_string()], move |_| async move { cancelled_started.notify_one(); @@ -432,7 +432,7 @@ async fn cache_lock_is_released_after_error_and_cancellation() { let recovered = timeout( Duration::from_secs(1), pool.get_cached_keys_raw( - "cancellation_recovery:v3", + "cancellation_recovery:v4", &["key".to_string()], |keys| async move { let values = DashMap::new(); @@ -452,19 +452,19 @@ async fn cache_lock_is_released_after_error_and_cancellation() { #[actix_rt::test] async fn expired_cache_value_serves_waiter_while_writer_refreshes() { let pool = isolated_redis_pool("stale_while_revalidate").await; - let namespace = "stale_while_revalidate:v3"; + let namespace = "stale_while_revalidate:v4"; let logical_key = "key".to_string(); let mut connection = pool.connect().await.unwrap(); let redis_key = connection.key().entity(namespace, &logical_key); connection .set_serialized( &redis_key, - json!({ - "key": logical_key, - "alias": null, - "iat": 0, - "val": "stale", - }), + RedisValue::::new( + logical_key.clone(), + None, + 0, + "stale".to_string(), + ), None, ) .await @@ -539,8 +539,8 @@ async fn case_insensitive_slug_requests_share_one_cache_lock() { let requested = vec![requested]; barrier.wait().await; pool.get_cached_keys_raw_with_slug( - "slug_values:v3", - Some("slug_aliases:v3"), + "slug_values:v4", + Some("slug_aliases:v4"), false, &requested, move |_| async move { @@ -651,11 +651,11 @@ async fn many_get_routes_handle_cross_slot_cache_lifecycle() { redis.key().entity(VERSIONS_NAMESPACE, alpha_version_id), redis.key().entity(VERSIONS_NAMESPACE, beta_version_id), redis.key().entity( - "versions_files:v3", + "versions_files:v4", format!("sha1_{}", alpha.file_hash), ), redis.key().entity( - "versions_files:v3", + "versions_files:v4", format!("sha1_{}", beta.file_hash), ), ]; diff --git a/apps/labrinth/tests/tags.rs b/apps/labrinth/tests/tags.rs index 5ca1bb49ad..e2852d461a 100644 --- a/apps/labrinth/tests/tags.rs +++ b/apps/labrinth/tests/tags.rs @@ -47,12 +47,7 @@ async fn get_tags_v3() { let loader_metadata = loaders .into_iter() - .map(|x| { - ( - x.name, - x.metadata.get("platform").and_then(|x| x.as_bool()), - ) - }) + .map(|x| (x.name, x.metadata.platform)) .collect::>(); let loader_names = loader_metadata.keys().cloned().collect::>(); diff --git a/packages/xredis/Cargo.toml b/packages/xredis/Cargo.toml index 7f9fd2035a..cc87779098 100644 --- a/packages/xredis/Cargo.toml +++ b/packages/xredis/Cargo.toml @@ -10,8 +10,10 @@ ariadne = { workspace = true } chrono = { workspace = true } dashmap = { workspace = true } deadpool-redis = { workspace = true, features = ["cluster-async"] } +eyre = { workspace = true } futures = { workspace = true } lz4_flex = { workspace = true } +postcard = { workspace = true } prometheus = { workspace = true } redis = { workspace = true, features = [ "ahash", @@ -20,7 +22,6 @@ redis = { workspace = true, features = [ "tokio-comp" ] } serde = { workspace = true, features = ["derive"] } -serde_json = { workspace = true } thiserror = { workspace = true } tokio = { workspace = true, features = ["rt", "sync", "time"] } tracing = { workspace = true } diff --git a/packages/xredis/src/blocking.rs b/packages/xredis/src/blocking.rs index 7a7b7f4e0f..5aaf62d772 100644 --- a/packages/xredis/src/blocking.rs +++ b/packages/xredis/src/blocking.rs @@ -1,14 +1,14 @@ use std::time::Duration; +use eyre::{Result, WrapErr, bail}; use prometheus::Registry; +use super::RedisPool; use super::config::{RedisConfig, RedisTopology}; -use super::connection::RedisBackendBuildError; use super::metrics::{ LogicalPoolStatus, LogicalPoolStatusProvider, register_blocking_pool_metrics, }; -use super::{Error, RedisPool}; const POOL_RETAIN_INTERVAL: Duration = Duration::from_secs(30); const MAX_IDLE_CONNECTION_AGE: Duration = Duration::from_secs(5 * 60); @@ -27,9 +27,7 @@ enum RedisBlockingPoolInner { } impl RedisBlockingPool { - pub(super) async fn new( - config: &RedisConfig, - ) -> Result { + pub(super) async fn new(config: &RedisConfig) -> Result { let pool_size = config.blocking_pool_size(); let inner = match config.topology() { RedisTopology::Standalone => { @@ -39,14 +37,16 @@ impl RedisBlockingPool { let manager = deadpool_redis::Manager::new_with_config( config.seed_urls()[0].clone(), connection_config, - )?; + ) + .wrap_err("configuring standalone blocking Redis client")?; let pool = deadpool_redis::Pool::builder(manager) .max_size(pool_size.max()) .wait_timeout(Some(Duration::from_millis( config.wait_timeout_ms(), ))) .runtime(deadpool_redis::Runtime::Tokio1) - .build()?; + .build() + .wrap_err("building standalone blocking Redis pool")?; retain_standalone_pool(pool.clone()); RedisBlockingPoolInner::Standalone(pool) } @@ -54,14 +54,16 @@ impl RedisBlockingPool { let manager = deadpool_redis::cluster::Manager::new( config.seed_urls().to_vec(), false, - )?; + ) + .wrap_err("configuring clustered blocking Redis client")?; let pool = deadpool_redis::cluster::Pool::builder(manager) .max_size(pool_size.max()) .wait_timeout(Some(Duration::from_millis( config.wait_timeout_ms(), ))) .runtime(deadpool_redis::Runtime::Tokio1) - .build()?; + .build() + .wrap_err("building clustered blocking Redis pool")?; retain_cluster_pool(pool.clone()); RedisBlockingPoolInner::Cluster(pool) } @@ -70,10 +72,7 @@ impl RedisBlockingPool { Ok(Self { inner }) } - pub(super) fn register_metrics( - &self, - registry: &Registry, - ) -> Result<(), prometheus::Error> { + pub(super) fn register_metrics(&self, registry: &Registry) -> Result<()> { register_blocking_pool_metrics(registry, self.clone()) } @@ -81,22 +80,33 @@ impl RedisBlockingPool { &self, key: &str, timeout: Duration, - ) -> Result; 2]>, Error> { + ) -> Result; 2]>> { if timeout.is_zero() { - return Err(Error::InvalidBlockingTimeout); + bail!("redis blocking timeout must be greater than zero"); } let mut command = redis::cmd("BRPOP"); command.arg(key).arg(timeout.as_secs_f64()); - let response: Option<(Vec, Vec)> = match &self.inner { - RedisBlockingPoolInner::Standalone(pool) => { - command.query_async(&mut pool.get().await?).await? - } - RedisBlockingPoolInner::Cluster(pool) => { - command.query_async(&mut pool.get().await?).await? - } - }; + let response: Option<(Vec, Vec)> = + match &self.inner { + RedisBlockingPoolInner::Standalone(pool) => { + let mut connection = pool.get().await.wrap_err( + "fetching standalone blocking Redis connection", + )?; + command.query_async(&mut connection).await.wrap_err( + "reading from standalone Redis blocking queue", + )? + } + RedisBlockingPoolInner::Cluster(pool) => { + let mut connection = pool.get().await.wrap_err( + "fetching clustered blocking Redis connection", + )?; + command.query_async(&mut connection).await.wrap_err( + "reading from clustered Redis blocking queue", + )? + } + }; Ok(response.map(|(key, value)| [key, value])) } @@ -120,7 +130,7 @@ impl RedisPool { &self, key: &str, timeout: Duration, - ) -> Result; 2]>, Error> { + ) -> Result; 2]>> { self.blocking.brpop(key, timeout).await } } diff --git a/packages/xredis/src/cache.rs b/packages/xredis/src/cache.rs index 5e2aba678b..8a4a3fa55b 100644 --- a/packages/xredis/src/cache.rs +++ b/packages/xredis/src/cache.rs @@ -8,6 +8,7 @@ use std::str::FromStr; use ariadne::ids::base62_impl::{parse_base62, to_base62}; use chrono::{TimeZone, Utc}; use dashmap::DashMap; +use eyre::{Result, WrapErr, eyre}; use futures::stream::{FuturesUnordered, StreamExt}; use redis::aio::ConnectionLike; use serde::de::DeserializeOwned; @@ -16,8 +17,6 @@ use thiserror::Error; use tokio::time::{Instant, timeout_at}; use tracing::{Instrument, info_span}; -use crate::Error; - use super::commands; use super::connection::RoutableConnection; use super::key::KeyBuilder; @@ -33,9 +32,7 @@ const FILL_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(60); pub(super) trait ConnectionProvider { type Connection: ConnectionLike + RoutableConnection; - fn connect( - &self, - ) -> impl Future> + Send; + fn connect(&self) -> impl Future> + Send; } #[derive(Clone, Copy)] @@ -53,7 +50,7 @@ pub enum Codec { #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum EncodingFormat { - Json, + Postcard, } #[derive(Debug, Error)] @@ -92,7 +89,7 @@ impl FromStr for EncodingFormat { fn from_str(value: &str) -> Result { match value { - "json" => Ok(Self::Json), + "postcard" => Ok(Self::Postcard), _ => Err(InvalidEncodingFormat), } } @@ -112,12 +109,10 @@ pub struct CacheSettings { } impl CacheSettings { - pub fn encode_value( - &self, - value: &T, - ) -> Result, Error> { + pub fn encode_value(&self, value: &T) -> Result> { let mut value = match self.encoding_format { - EncodingFormat::Json => serde_json::to_vec(value)?, + EncodingFormat::Postcard => postcard::to_allocvec(value) + .wrap_err("serializing Redis cache value with postcard")?, }; if self.compression_level > 0 @@ -148,16 +143,26 @@ impl CacheSettings { where T: for<'a> Deserialize<'a>, { - let (codec, value) = value.split_first()?; - let value = match Codec::try_from(*codec).ok()? { + let Some((codec, value)) = value.split_first() else { + return None; + }; + let Ok(codec) = Codec::try_from(*codec) else { + return None; + }; + let value = match codec { Codec::Raw => Cow::Borrowed(value), - Codec::Lz4 => Cow::Owned( - lz4_flex::block::decompress_size_prepended(value).ok()?, - ), + Codec::Lz4 => { + let Ok(value) = + lz4_flex::block::decompress_size_prepended(value) + else { + return None; + }; + Cow::Owned(value) + } }; match self.encoding_format { - EncodingFormat::Json => serde_json::from_slice(&value).ok(), + EncodingFormat::Postcard => postcard::from_bytes(&value).ok(), } } @@ -202,12 +207,12 @@ impl CacheManager { namespace: &str, keys: &[K], closure: F, - ) -> Result, E> + ) -> Result> where P: ConnectionProvider, F: FnOnce(Vec) -> Fut, Fut: Future, E>>, - E: From, + E: std::error::Error + Send + Sync + 'static, T: Serialize + DeserializeOwned, K: Display + Hash @@ -220,7 +225,8 @@ impl CacheManager { { Ok(self .get_cached_keys_raw(provider, namespace, keys, closure) - .await? + .await + .wrap_err("fetching Redis cache values")? .into_values() .collect()) } @@ -232,12 +238,12 @@ impl CacheManager { namespace: &str, keys: &[K], closure: F, - ) -> Result, E> + ) -> Result> where P: ConnectionProvider, F: FnOnce(Vec) -> Fut, Fut: Future, E>>, - E: From, + E: std::error::Error + Send + Sync + 'static, T: Serialize + DeserializeOwned, K: Display + Hash @@ -255,11 +261,16 @@ impl CacheManager { false, keys, |ids| async move { - Ok(closure(ids) - .await? - .into_iter() - .map(|(key, value)| (key, (None::, value))) - .collect()) + let values = match closure(ids).await { + Ok(values) => values, + Err(error) => return Err(error), + }; + Ok::<_, E>( + values + .into_iter() + .map(|(key, value)| (key, (None::, value))) + .collect(), + ) }, ) .await @@ -274,12 +285,12 @@ impl CacheManager { case_sensitive: bool, keys: &[I], closure: F, - ) -> Result, E> + ) -> Result> where P: ConnectionProvider, F: FnOnce(Vec) -> Fut, Fut: Future, T)>, E>>, - E: From, + E: std::error::Error + Send + Sync + 'static, T: Serialize + DeserializeOwned, I: Display + Hash + Eq + PartialEq + Clone + Debug, K: Display @@ -300,7 +311,8 @@ impl CacheManager { keys, closure, ) - .await? + .await + .wrap_err("fetching Redis cache values by slug")? .into_values() .collect()) } @@ -314,12 +326,12 @@ impl CacheManager { case_sensitive: bool, keys: &[I], closure: F, - ) -> Result, E> + ) -> Result> where P: ConnectionProvider, F: FnOnce(Vec) -> Fut, Fut: Future, T)>, E>>, - E: From, + E: std::error::Error + Send + Sync + 'static, T: Serialize + DeserializeOwned, I: Display + Hash + Eq + PartialEq + Clone + Debug, K: Display @@ -360,7 +372,9 @@ impl CacheManager { }) .collect::>(); let mut connection = - provider.connect().await.map_err(E::from)?; + provider.connect().await.wrap_err( + "connecting to Redis for slug lookup", + )?; let values = match routing { CacheReadRouting::ReplicaOptional => { commands::get_many_strings( @@ -377,8 +391,8 @@ impl CacheManager { .await } } - .map_err(E::from)?; - Ok::<_, E>( + .wrap_err("fetching Redis cache slug values")?; + eyre::Ok( values .into_iter() .flatten() @@ -386,7 +400,8 @@ impl CacheManager { ) } .instrument(info_span!("get slug ids")) - .await? + .await + .wrap_err("resolving Redis cache slugs")? } else { Vec::new() }; @@ -403,8 +418,10 @@ impl CacheManager { .map(|key| self.key_builder.entity(namespace, key)) .collect::>(); - let mut connection = - provider.connect().await.map_err(E::from)?; + let mut connection = provider + .connect() + .await + .wrap_err("connecting to Redis for cache lookup")?; let mut cached_values = HashMap::new(); let values = match routing { CacheReadRouting::ReplicaOptional => { @@ -415,7 +432,7 @@ impl CacheManager { .await } } - .map_err(E::from)?; + .wrap_err("fetching Redis cache values")?; for value in values { if let Some(value) = value.and_then(|value| { self.settings @@ -425,7 +442,7 @@ impl CacheManager { } } - Ok::<_, E>((cached_values, ids)) + eyre::Ok((cached_values, ids)) } .instrument(info_span!("get_cached_values_closure")) }; @@ -437,7 +454,9 @@ impl CacheManager { let deadline = Instant::now() + WAIT_TIMEOUT; let (cached_values_raw, ids) = - get_cached_values(ids, CacheReadRouting::ReplicaOptional).await?; + get_cached_values(ids, CacheReadRouting::ReplicaOptional) + .await + .wrap_err("reading Redis cache")?; let mut cached_values = cached_values_raw .into_iter() .filter_map(|(key, value)| { @@ -508,7 +527,10 @@ impl CacheManager { let values = timeout_at(fill_deadline, closure(fetch_ids)) .await - .map_err(|_| lock_timeout_error(0, waiters.len()))??; + .map_err(|_| lock_timeout_error(0, waiters.len())) + .wrap_err("waiting to fill Redis cache")?; + let values = + values.wrap_err("fetching values to fill Redis cache")?; let mut return_values = HashMap::new(); let mut encoded_values = Vec::with_capacity(values.len()); @@ -520,13 +542,17 @@ impl CacheManager { val: value, alias: slug.clone(), }; - let encoded = - self.settings.encode_value(&value).map_err(E::from)?; + let encoded = self + .settings + .encode_value(&value) + .wrap_err("encoding Redis cache value")?; encoded_values.push((key, slug, value, encoded)); } - let mut connection = - provider.connect().await.map_err(E::from)?; + let mut connection = provider + .connect() + .await + .wrap_err("connecting to Redis to fill cache")?; for (key, slug, _, encoded) in &encoded_values { let redis_key = self.key_builder.entity(namespace, key.to_string()); @@ -537,7 +563,7 @@ impl CacheManager { default_expiry, ) .await - .map_err(E::from)?; + .wrap_err("writing Redis cache value")?; if let Some(slug) = slug && let Some(slug_namespace) = slug_namespace { @@ -554,7 +580,7 @@ impl CacheManager { default_expiry, ) .await - .map_err(E::from)?; + .wrap_err("writing Redis cache slug")?; } } @@ -563,7 +589,7 @@ impl CacheManager { return_values.insert(key, value); } - Result::<_, E>::Ok(return_values) + Result::<_>::Ok(return_values) } .await } else { @@ -604,13 +630,14 @@ impl CacheManager { Err(error) => Err(error), } } - Err(error) => Err(E::from(error)), + Err(error) => Err(error), } } } Err(error) => Err(error), }; - cached_values.extend(operation_result?); + cached_values + .extend(operation_result.wrap_err("populating Redis cache")?); Ok(cached_values .into_iter() @@ -679,7 +706,7 @@ fn push_identity(identities: &mut Vec, identity: String) { async fn wait_for_locks( waiters: Vec<(I, LockWaiter)>, deadline: Instant, -) -> Result, Error> { +) -> Result> { let total = waiters.len(); let mut released = Vec::with_capacity(total); let mut futures = FuturesUnordered::new(); @@ -695,26 +722,24 @@ async fn wait_for_locks( Ok(()) => { released.push(key); } - Err(error) - if is_lock_timeout(&error) || Instant::now() >= deadline => - { + Err(_) if Instant::now() >= deadline => { return Err(lock_timeout_error(released.len(), total)); } - Err(error) => return Err(error), + Err(error) => { + return Err(error).wrap_err("waiting for Redis cache lock"); + } } } Ok(released) } -fn is_lock_timeout(error: &Error) -> bool { - matches!(error, Error::LocalCacheTimeout { .. }) -} - -fn lock_timeout_error(locks_released: usize, locks_waiting: usize) -> Error { - Error::LocalCacheTimeout { - released: locks_released, - total: locks_waiting, - } +fn lock_timeout_error( + locks_released: usize, + locks_waiting: usize, +) -> eyre::Report { + eyre!( + "timeout waiting on local Redis cache lock ({locks_released}/{locks_waiting} released)" + ) } #[derive(Serialize, Deserialize)] @@ -726,6 +751,15 @@ pub struct RedisValue { } impl RedisValue { + pub fn new(key: K, alias: Option, iat: i64, val: T) -> Self { + Self { + key, + alias, + iat, + val, + } + } + pub fn value(&self) -> &T { &self.val } diff --git a/packages/xredis/src/cache/locking/local.rs b/packages/xredis/src/cache/locking/local.rs index fab5f6f822..001a64f72b 100644 --- a/packages/xredis/src/cache/locking/local.rs +++ b/packages/xredis/src/cache/locking/local.rs @@ -3,11 +3,10 @@ use std::sync::atomic::{AtomicBool, Ordering}; use dashmap::DashMap; use dashmap::mapref::entry::Entry; +use eyre::{Result, WrapErr}; use tokio::sync::Notify; use tokio::time::{Instant, timeout_at}; -use crate::Error; - #[derive(Clone)] pub(in crate::cache) struct LockCoordinator { locks: Arc>>, @@ -76,10 +75,7 @@ pub(in crate::cache) struct LockWaiter { } impl LockWaiter { - pub(in crate::cache) async fn wait( - self, - deadline: Instant, - ) -> Result<(), Error> { + pub(in crate::cache) async fn wait(self, deadline: Instant) -> Result<()> { loop { if self.state.released.load(Ordering::Acquire) { return Ok(()); @@ -94,7 +90,7 @@ impl LockWaiter { timeout_at(deadline, notified) .await - .map_err(|_| lock_timeout())?; + .wrap_err("waiting for local Redis cache lock")?; } } } @@ -112,10 +108,3 @@ impl LockState { } } } - -fn lock_timeout() -> Error { - Error::LocalCacheTimeout { - released: 0, - total: 1, - } -} diff --git a/packages/xredis/src/commands.rs b/packages/xredis/src/commands.rs index fdf0b058c7..744e158a89 100644 --- a/packages/xredis/src/commands.rs +++ b/packages/xredis/src/commands.rs @@ -1,10 +1,9 @@ use std::fmt::Debug; +use eyre::{Result, WrapErr}; use redis::aio::ConnectionLike; use redis::{FromRedisValue, ToRedisArgs}; -use crate::Error; - use super::cache::CacheSettings; use super::connection::RoutableConnection; use super::routing::primary_mget_routing; @@ -18,7 +17,7 @@ pub async fn set( key: &str, data: D, expiry: i64, -) -> Result<(), Error> +) -> Result<()> where C: ConnectionLike, D: ToRedisArgs + Send + Sync + Debug, @@ -29,7 +28,8 @@ where .arg("EX") .arg(expiry) .query_async::<()>(connection) - .await?; + .await + .wrap_err("writing to Redis")?; Ok(()) } @@ -40,7 +40,7 @@ pub async fn set_serialized( data: D, expiry: Option, settings: &CacheSettings, -) -> Result<(), Error> +) -> Result<()> where C: ConnectionLike, D: serde::Serialize, @@ -48,21 +48,24 @@ where set( connection, key, - settings.encode_value(&data)?, + settings + .encode_value(&data) + .wrap_err("serializing Redis value")?, expiry.unwrap_or(settings.default_expiry), ) .await } #[tracing::instrument(skip_all)] -pub async fn get( - connection: &mut C, - key: &str, -) -> Result, Error> +pub async fn get(connection: &mut C, key: &str) -> Result> where C: ConnectionLike, { - Ok(cmd("GET").arg(key).query_async(connection).await?) + cmd("GET") + .arg(key) + .query_async(connection) + .await + .wrap_err("fetching from Redis") } /// Issues ordinary `MGET` commands in bounded chunks. Cluster routing and @@ -72,7 +75,7 @@ where pub async fn get_many( connection: &mut C, keys: &[String], -) -> Result>>, Error> +) -> Result>>> where C: ConnectionLike, { @@ -83,7 +86,7 @@ where pub async fn get_many_strings( connection: &mut C, keys: &[String], -) -> Result>, Error> +) -> Result>> where C: ConnectionLike, { @@ -93,7 +96,7 @@ where pub(super) async fn get_many_primary( connection: &mut C, keys: &[String], -) -> Result>>, Error> +) -> Result>>> where C: RoutableConnection, { @@ -103,7 +106,7 @@ where pub(super) async fn get_many_strings_primary( connection: &mut C, keys: &[String], -) -> Result>, Error> +) -> Result>> where C: RoutableConnection, { @@ -113,7 +116,7 @@ where pub(super) async fn get_many_as( connection: &mut C, keys: &[String], -) -> Result>, Error> +) -> Result>> where C: ConnectionLike, T: FromRedisValue, @@ -123,7 +126,8 @@ where let part = cmd("MGET") .arg(chunk) .query_async::>>(connection) - .await?; + .await + .wrap_err("fetching multiple values from Redis")?; values.extend(part); } Ok(values) @@ -132,7 +136,7 @@ where async fn get_many_primary_as( connection: &mut C, keys: &[String], -) -> Result>, Error> +) -> Result>> where C: RoutableConnection, T: FromRedisValue, @@ -143,10 +147,14 @@ where command.arg(chunk); let value = connection .route_command(command, primary_mget_routing(chunk)) - .await?; - let part = - redis::from_redis_value::>>(value.extract_error()?) - .map_err(redis::RedisError::from)?; + .await + .wrap_err("fetching multiple values from primary Redis nodes")?; + let value = value + .extract_error() + .wrap_err("extracting Redis response")?; + let part = redis::from_redis_value::>>(value) + .map_err(redis::RedisError::from) + .wrap_err("decoding Redis response")?; values.extend(part); } Ok(values) @@ -157,13 +165,16 @@ pub async fn get_deserialized( connection: &mut C, key: &str, settings: &CacheSettings, -) -> Result, Error> +) -> Result> where C: ConnectionLike, R: for<'a> serde::Deserialize<'a>, { - let value: Option> = - cmd("GET").arg(key).query_async(connection).await?; + let value: Option> = cmd("GET") + .arg(key) + .query_async(connection) + .await + .wrap_err("fetching serialized value from Redis")?; Ok(value.and_then(|value| settings.decode_value(&value))) } @@ -172,47 +183,49 @@ pub async fn get_many_deserialized( connection: &mut C, keys: &[String], settings: &CacheSettings, -) -> Result>, Error> +) -> Result>> where C: ConnectionLike, R: for<'a> serde::Deserialize<'a>, { Ok(get_many(connection, keys) - .await? + .await + .wrap_err("fetching serialized values from Redis")? .into_iter() .map(|value| value.and_then(|value| settings.decode_value(&value))) .collect()) } #[tracing::instrument(skip_all)] -pub async fn delete(connection: &mut C, key: &str) -> Result<(), Error> +pub async fn delete(connection: &mut C, key: &str) -> Result<()> where C: ConnectionLike, { - cmd("DEL").arg(key).query_async::<()>(connection).await?; + cmd("DEL") + .arg(key) + .query_async::<()>(connection) + .await + .wrap_err("deleting from Redis")?; Ok(()) } #[tracing::instrument(skip_all)] -pub async fn delete_many( - connection: &mut C, - keys: &[String], -) -> Result<(), Error> +pub async fn delete_many(connection: &mut C, keys: &[String]) -> Result<()> where C: ConnectionLike, { if !keys.is_empty() { - cmd("DEL").arg(keys).query_async::<()>(connection).await?; + cmd("DEL") + .arg(keys) + .query_async::<()>(connection) + .await + .wrap_err("deleting multiple values from Redis")?; } Ok(()) } #[tracing::instrument(skip_all)] -pub async fn lpush( - connection: &mut C, - key: &str, - value: D, -) -> Result<(), Error> +pub async fn lpush(connection: &mut C, key: &str, value: D) -> Result<()> where C: ConnectionLike, D: ToRedisArgs + Send + Sync + Debug, @@ -221,17 +234,19 @@ where .arg(key) .arg(value) .query_async::<()>(connection) - .await?; + .await + .wrap_err("pushing to Redis list")?; Ok(()) } #[tracing::instrument(skip_all)] -pub async fn incr( - connection: &mut C, - key: &str, -) -> Result, Error> +pub async fn incr(connection: &mut C, key: &str) -> Result> where C: ConnectionLike, { - Ok(cmd("INCR").arg(key).query_async(connection).await?) + cmd("INCR") + .arg(key) + .query_async(connection) + .await + .wrap_err("incrementing Redis value") } diff --git a/packages/xredis/src/config.rs b/packages/xredis/src/config.rs index 6e466b107d..d63b3b7f89 100644 --- a/packages/xredis/src/config.rs +++ b/packages/xredis/src/config.rs @@ -1,5 +1,6 @@ use std::{fmt, str::FromStr}; +use eyre::{Result, WrapErr}; use thiserror::Error; #[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] @@ -91,13 +92,11 @@ pub(crate) struct RedisPoolSize { } impl RedisPoolSize { - fn new( - name: &'static str, - max: usize, - min: usize, - ) -> Result { + fn new(name: &'static str, max: usize, min: usize) -> Result { if max == 0 || min > max { - return Err(RedisConfigError::InvalidPoolSize { name, max, min }); + return Err( + RedisConfigError::InvalidPoolSize { name, max, min }.into() + ); } Ok(Self { max, min }) @@ -193,11 +192,12 @@ impl RedisConfig { blocking_pool_size: (usize, usize), cache_locking_strategy: CacheLockingStrategy, read_replica_strategy: ReadReplicaStrategy, - ) -> Result { + ) -> Result { if cache_locking_strategy == CacheLockingStrategy::Distributed { return Err(RedisConfigError::UnsupportedCacheLockingStrategy { strategy: cache_locking_strategy, - }); + } + .into()); } let seed_urls = raw_urls @@ -208,26 +208,32 @@ impl RedisConfig { .collect::>(); if seed_urls.is_empty() { - return Err(RedisConfigError::MissingUrl); + return Err(RedisConfigError::MissingUrl.into()); } let backend = match (mode, connection_type) { (RedisTopology::Standalone, RedisConnectionType::Pooled) => { if seed_urls.len() != 1 { - return Err(RedisConfigError::MultipleStandaloneUrls); + return Err(RedisConfigError::MultipleStandaloneUrls.into()); } - RedisBackendConfig::StandalonePooled(RedisPoolSize::new( - "standalone", - standalone_pool_size.0, - standalone_pool_size.1, - )?) + RedisBackendConfig::StandalonePooled( + RedisPoolSize::new( + "standalone", + standalone_pool_size.0, + standalone_pool_size.1, + ) + .wrap_err("validating standalone Redis pool size")?, + ) } (RedisTopology::Cluster, RedisConnectionType::Pooled) => { - RedisBackendConfig::ClusterPooled(RedisPoolSize::new( - "cluster", - cluster_pool_size.0, - cluster_pool_size.1, - )?) + RedisBackendConfig::ClusterPooled( + RedisPoolSize::new( + "cluster", + cluster_pool_size.0, + cluster_pool_size.1, + ) + .wrap_err("validating clustered Redis pool size")?, + ) } (RedisTopology::Cluster, RedisConnectionType::Multiplexed) => { RedisBackendConfig::ClusterMultiplexed @@ -236,7 +242,8 @@ impl RedisConfig { return Err(RedisConfigError::UnsupportedConnectionType { mode, connection_type, - }); + } + .into()); } }; @@ -249,7 +256,8 @@ impl RedisConfig { "blocking", blocking_pool_size.0, blocking_pool_size.1, - )?, + ) + .wrap_err("validating blocking Redis pool size")?, cache_locking_strategy, read_replica_strategy, }) diff --git a/packages/xredis/src/connection.rs b/packages/xredis/src/connection.rs index 0d66fc818d..3632290d19 100644 --- a/packages/xredis/src/connection.rs +++ b/packages/xredis/src/connection.rs @@ -1,5 +1,6 @@ use std::time::Duration; +use eyre::{Result, WrapErr}; use futures::future::try_join_all; use prometheus::Registry; use redis::aio::ConnectionLike; @@ -7,7 +8,6 @@ use redis::cluster_read_routing::{ RandomReplicaStrategy, RoundRobinReplicaStrategy, }; use redis::cluster_routing::RoutingInfo; -use thiserror::Error; use tracing::warn; use crate::ReadReplicaStrategy; @@ -29,16 +29,6 @@ pub(crate) enum RedisBackend { ClusterMultiplexed(redis::cluster_async::ClusterConnection), } -#[derive(Debug, Error)] -pub enum RedisBackendBuildError { - #[error("failed to configure Redis client: {0}")] - Redis(#[from] redis::RedisError), - #[error("failed to build Redis pool: {0}")] - PoolBuild(#[from] deadpool_redis::BuildError), - #[error("failed to establish initial Redis pool connections: {0}")] - Pool(#[from] deadpool_redis::PoolError), -} - pub(crate) struct RedisConnection { inner: RedisConnectionInner, } @@ -58,9 +48,7 @@ pub(crate) trait RoutableConnection: ConnectionLike { } impl RedisBackend { - pub(crate) async fn new( - config: &RedisConfig, - ) -> Result { + pub(crate) async fn new(config: &RedisConfig) -> Result { match config.backend() { RedisBackendConfig::StandalonePooled(pool_size) => { Self::standalone_pooled(config, pool_size).await @@ -77,21 +65,25 @@ impl RedisBackend { async fn standalone_pooled( config: &RedisConfig, pool_size: RedisPoolSize, - ) -> Result { + ) -> Result { let connection_config = redis::AsyncConnectionConfig::new() .set_connection_timeout(None) .set_response_timeout(None); let manager = deadpool_redis::Manager::new_with_config( config.seed_urls()[0].clone(), connection_config, - )?; + ) + .wrap_err("configuring standalone Redis client")?; let pool = deadpool_redis::Pool::builder(manager) .max_size(pool_size.max()) .wait_timeout(Some(Duration::from_millis(config.wait_timeout_ms()))) .runtime(deadpool_redis::Runtime::Tokio1) - .build()?; + .build() + .wrap_err("building standalone Redis pool")?; - warm_standalone_pool(&pool, pool_size.min()).await?; + warm_standalone_pool(&pool, pool_size.min()) + .await + .wrap_err("warming standalone Redis pool")?; retain_standalone_pool(pool.clone()); Ok(Self::StandalonePooled(pool)) @@ -100,16 +92,18 @@ impl RedisBackend { async fn cluster_pooled( config: &RedisConfig, pool_size: RedisPoolSize, - ) -> Result { + ) -> Result { let manager = deadpool_redis::cluster::Manager::new( config.seed_urls().to_vec(), false, - )?; + ) + .wrap_err("configuring clustered Redis client")?; let pool = deadpool_redis::cluster::Pool::builder(manager) .max_size(pool_size.max()) .wait_timeout(Some(Duration::from_millis(config.wait_timeout_ms()))) .runtime(deadpool_redis::Runtime::Tokio1) - .build()?; + .build() + .wrap_err("building clustered Redis pool")?; if config.read_replica_strategy() != ReadReplicaStrategy::Primary { warn!( @@ -117,15 +111,15 @@ impl RedisBackend { ); } - warm_cluster_pool(&pool, pool_size.min()).await?; + warm_cluster_pool(&pool, pool_size.min()) + .await + .wrap_err("warming clustered Redis pool")?; retain_cluster_pool(pool.clone()); Ok(Self::ClusterPooled(pool)) } - async fn cluster_multiplexed( - config: &RedisConfig, - ) -> Result { + async fn cluster_multiplexed(config: &RedisConfig) -> Result { let mut builder = redis::cluster::ClusterClientBuilder::new( config.seed_urls().iter().map(String::as_str), ); @@ -141,22 +135,31 @@ impl RedisBackend { } } - let client = builder.build()?; - let connection = client.get_async_connection().await?; + let client = builder + .build() + .wrap_err("building multiplexed Redis client")?; + let connection = client + .get_async_connection() + .await + .wrap_err("connecting multiplexed Redis client")?; Ok(Self::ClusterMultiplexed(connection)) } - pub(crate) async fn connect( - &self, - ) -> Result { + pub(crate) async fn connect(&self) -> Result { let inner = match self { Self::StandalonePooled(pool) => { - RedisConnectionInner::StandalonePooled(pool.get().await?) - } - Self::ClusterPooled(pool) => { - RedisConnectionInner::ClusterPooled(pool.get().await?) + RedisConnectionInner::StandalonePooled( + pool.get() + .await + .wrap_err("fetching standalone Redis connection")?, + ) } + Self::ClusterPooled(pool) => RedisConnectionInner::ClusterPooled( + pool.get() + .await + .wrap_err("fetching clustered Redis connection")?, + ), Self::ClusterMultiplexed(connection) => { RedisConnectionInner::ClusterMultiplexed(connection.clone()) } @@ -165,10 +168,7 @@ impl RedisBackend { Ok(RedisConnection { inner }) } - pub(crate) fn register_metrics( - &self, - registry: &Registry, - ) -> Result<(), prometheus::Error> { + pub(crate) fn register_metrics(&self, registry: &Registry) -> Result<()> { register_command_pool_metrics(registry, self.clone()) } } @@ -282,8 +282,10 @@ impl RoutableConnection for RedisConnection { async fn warm_standalone_pool( pool: &deadpool_redis::Pool, min: usize, -) -> Result<(), deadpool_redis::PoolError> { - let connections = try_join_all((0..min).map(|_| pool.get())).await?; +) -> Result<()> { + let connections = try_join_all((0..min).map(|_| pool.get())) + .await + .wrap_err("fetching initial standalone Redis connections")?; drop(connections); Ok(()) } @@ -291,8 +293,10 @@ async fn warm_standalone_pool( async fn warm_cluster_pool( pool: &deadpool_redis::cluster::Pool, min: usize, -) -> Result<(), deadpool_redis::PoolError> { - let connections = try_join_all((0..min).map(|_| pool.get())).await?; +) -> Result<()> { + let connections = try_join_all((0..min).map(|_| pool.get())) + .await + .wrap_err("fetching initial clustered Redis connections")?; drop(connections); Ok(()) } diff --git a/packages/xredis/src/lib.rs b/packages/xredis/src/lib.rs index e84fce15eb..19d4afce1e 100644 --- a/packages/xredis/src/lib.rs +++ b/packages/xredis/src/lib.rs @@ -6,6 +6,7 @@ use std::hash::Hash; use std::sync::Arc; use dashmap::DashMap; +use eyre::{Result, WrapErr}; use prometheus::Registry; use redis::aio::ConnectionLike; use redis::{FromRedisValue, ToRedisArgs}; @@ -35,25 +36,7 @@ pub use config::{ RedisConfigError, RedisConnectionType, RedisTopology, }; use connection::RedisBackend; -pub use connection::RedisBackendBuildError; pub use key::KeyBuilder; -use thiserror::Error as ThisError; - -#[derive(Debug, ThisError)] -pub enum Error { - #[error("error while interacting with Redis: {0}")] - Redis(#[from] redis::RedisError), - #[error("Redis pool error: {0}")] - Pool(#[from] deadpool_redis::PoolError), - #[error("error while serializing a Redis cache value: {0}")] - Serialization(#[from] serde_json::Error), - #[error("Redis blocking timeout must be greater than zero")] - InvalidBlockingTimeout, - #[error( - "timeout waiting on local cache lock ({released}/{total} released)" - )] - LocalCacheTimeout { released: usize, total: usize }, -} #[derive(Clone)] pub struct RedisPool { @@ -75,15 +58,19 @@ impl RedisPool { meta_namespace: impl Into>, config: RedisConfig, cache_settings: CacheSettings, - ) -> Result { + ) -> Result { tracing::info!( strategy = %config.cache_locking_strategy(), "configured Redis cache locking" ); - let backend = RedisBackend::new(&config).await?; + let backend = RedisBackend::new(&config) + .await + .wrap_err("creating Redis command backend")?; - let blocking = blocking::RedisBlockingPool::new(&config).await?; + let blocking = blocking::RedisBlockingPool::new(&config) + .await + .wrap_err("creating Redis blocking pool")?; let key_builder = KeyBuilder::new(meta_namespace, config.topology()); let cache = CacheManager::new(key_builder.clone(), cache_settings); @@ -102,9 +89,13 @@ impl RedisPool { } impl RedisPool { - pub async fn connect(&self) -> Result { + pub async fn connect(&self) -> Result { Ok(RedisConnection { - inner: self.backend.connect().await?, + inner: self + .backend + .connect() + .await + .wrap_err("connecting to Redis")?, key_builder: self.key_builder.clone(), settings: self.cache.settings().clone(), }) @@ -113,9 +104,13 @@ impl RedisPool { pub async fn register_and_set_metrics( &self, registry: &Registry, - ) -> Result<(), prometheus::Error> { - self.backend.register_metrics(registry)?; - self.blocking.register_metrics(registry) + ) -> Result<()> { + self.backend + .register_metrics(registry) + .wrap_err("registering Redis command pool metrics")?; + self.blocking + .register_metrics(registry) + .wrap_err("registering Redis blocking pool metrics") } pub async fn get_cached_keys( @@ -123,11 +118,11 @@ impl RedisPool { namespace: &str, keys: &[K], closure: F, - ) -> Result, E> + ) -> Result> where F: FnOnce(Vec) -> Fut, Fut: Future, E>>, - E: From, + E: std::error::Error + Send + Sync + 'static, T: Serialize + DeserializeOwned, K: Display + Hash @@ -148,11 +143,11 @@ impl RedisPool { namespace: &str, keys: &[K], closure: F, - ) -> Result, E> + ) -> Result> where F: FnOnce(Vec) -> Fut, Fut: Future, E>>, - E: From, + E: std::error::Error + Send + Sync + 'static, T: Serialize + DeserializeOwned, K: Display + Hash @@ -175,11 +170,11 @@ impl RedisPool { case_sensitive: bool, keys: &[I], closure: F, - ) -> Result, E> + ) -> Result> where F: FnOnce(Vec) -> Fut, Fut: Future, T)>, E>>, - E: From, + E: std::error::Error + Send + Sync + 'static, T: Serialize + DeserializeOwned, I: Display + Hash + Eq + PartialEq + Clone + Debug, K: Display @@ -210,11 +205,11 @@ impl RedisPool { case_sensitive: bool, keys: &[I], closure: F, - ) -> Result, E> + ) -> Result> where F: FnOnce(Vec) -> Fut, Fut: Future, T)>, E>>, - E: From, + E: std::error::Error + Send + Sync + 'static, T: Serialize + DeserializeOwned, I: Display + Hash + Eq + PartialEq + Clone + Debug, K: Display @@ -242,9 +237,7 @@ impl RedisPool { impl ConnectionProvider for RedisPool { type Connection = RedisConnection; - fn connect( - &self, - ) -> impl Future> + Send { + fn connect(&self) -> impl Future> + Send { RedisPool::connect(self) } } @@ -259,7 +252,7 @@ impl RedisConnection { key: &str, data: D, expiry: Option, - ) -> Result<(), Error> + ) -> Result<()> where D: ToRedisArgs + Send + Sync + Debug, { @@ -277,7 +270,7 @@ impl RedisConnection { key: &str, data: D, expiry: Option, - ) -> Result<(), Error> + ) -> Result<()> where D: Serialize, { @@ -291,31 +284,28 @@ impl RedisConnection { .await } - pub async fn get(&mut self, key: &str) -> Result, Error> { + pub async fn get(&mut self, key: &str) -> Result> { commands::get(&mut self.inner, key).await } pub async fn get_many( &mut self, keys: &[String], - ) -> Result>>, Error> { + ) -> Result>>> { commands::get_many(&mut self.inner, keys).await } pub async fn get_many_typed( &mut self, keys: &[String], - ) -> Result>, Error> + ) -> Result>> where R: FromRedisValue, { commands::get_many_as(&mut self.inner, keys).await } - pub async fn get_deserialized( - &mut self, - key: &str, - ) -> Result, Error> + pub async fn get_deserialized(&mut self, key: &str) -> Result> where R: for<'a> serde::Deserialize<'a>, { @@ -325,7 +315,7 @@ impl RedisConnection { pub async fn get_many_deserialized( &mut self, keys: &[String], - ) -> Result>, Error> + ) -> Result>> where R: for<'a> serde::Deserialize<'a>, { @@ -333,22 +323,22 @@ impl RedisConnection { .await } - pub async fn delete(&mut self, key: &str) -> Result<(), Error> { + pub async fn delete(&mut self, key: &str) -> Result<()> { commands::delete(&mut self.inner, key).await } - pub async fn delete_many(&mut self, keys: &[String]) -> Result<(), Error> { + pub async fn delete_many(&mut self, keys: &[String]) -> Result<()> { commands::delete_many(&mut self.inner, keys).await } - pub async fn lpush(&mut self, key: &str, value: D) -> Result<(), Error> + pub async fn lpush(&mut self, key: &str, value: D) -> Result<()> where D: ToRedisArgs + Send + Sync + Debug, { commands::lpush(&mut self.inner, key, value).await } - pub async fn incr(&mut self, key: &str) -> Result, Error> { + pub async fn incr(&mut self, key: &str) -> Result> { commands::incr(&mut self.inner, key).await } } diff --git a/packages/xredis/src/metrics.rs b/packages/xredis/src/metrics.rs index d50e4cce47..fafd5dccbe 100644 --- a/packages/xredis/src/metrics.rs +++ b/packages/xredis/src/metrics.rs @@ -1,5 +1,6 @@ use std::time::Duration; +use eyre::{Result, WrapErr}; use prometheus::{IntGauge, Registry}; const METRICS_UPDATE_INTERVAL: Duration = Duration::from_secs(5); @@ -72,7 +73,7 @@ impl RedisPoolMetrics { fn register( registry: &Registry, kind: RedisPoolMetricsKind, - ) -> Result { + ) -> Result { let prefix = kind.metric_prefix(); let description = kind.description(); let max_size = IntGauge::new( @@ -80,28 +81,40 @@ impl RedisPoolMetrics { format!( "Maximum logical connection count for the {description}; clustered logical connections may own multiple physical sockets" ), - )?; + ) + .wrap_err("creating Redis pool maximum size metric")?; let size = IntGauge::new( format!("{prefix}_size"), format!( "Current logical connection count for the {description}; clustered logical connections may own multiple physical sockets" ), - )?; + ) + .wrap_err("creating Redis pool size metric")?; let available = IntGauge::new( format!("{prefix}_available"), format!("Available logical connections in the {description}"), - )?; + ) + .wrap_err("creating Redis pool availability metric")?; let waiting = IntGauge::new( format!("{prefix}_waiting"), format!( "Number of futures waiting for a logical connection from the {description}" ), - )?; - - registry.register(Box::new(max_size.clone()))?; - registry.register(Box::new(size.clone()))?; - registry.register(Box::new(available.clone()))?; - registry.register(Box::new(waiting.clone()))?; + ) + .wrap_err("creating Redis pool waiters metric")?; + + registry + .register(Box::new(max_size.clone())) + .wrap_err("registering Redis pool maximum size metric")?; + registry + .register(Box::new(size.clone())) + .wrap_err("registering Redis pool size metric")?; + registry + .register(Box::new(available.clone())) + .wrap_err("registering Redis pool availability metric")?; + registry + .register(Box::new(waiting.clone())) + .wrap_err("registering Redis pool waiters metric")?; Ok(Self { max_size, @@ -122,7 +135,7 @@ impl RedisPoolMetrics { pub(super) fn register_command_pool_metrics

( registry: &Registry, provider: P, -) -> Result<(), prometheus::Error> +) -> Result<()> where P: LogicalPoolStatusProvider, { @@ -132,7 +145,7 @@ where pub(super) fn register_blocking_pool_metrics

( registry: &Registry, provider: P, -) -> Result<(), prometheus::Error> +) -> Result<()> where P: LogicalPoolStatusProvider, { @@ -143,11 +156,12 @@ fn register_pool_metrics

( registry: &Registry, kind: RedisPoolMetricsKind, provider: P, -) -> Result<(), prometheus::Error> +) -> Result<()> where P: LogicalPoolStatusProvider, { - let metrics = RedisPoolMetrics::register(registry, kind)?; + let metrics = RedisPoolMetrics::register(registry, kind) + .wrap_err("registering Redis pool metrics")?; metrics.set(provider.logical_pool_status()); tokio::spawn(async move { diff --git a/packages/xredis/src/pubsub.rs b/packages/xredis/src/pubsub.rs index b6080a0f32..c4986ce043 100644 --- a/packages/xredis/src/pubsub.rs +++ b/packages/xredis/src/pubsub.rs @@ -1,11 +1,12 @@ use std::time::Duration; +use eyre::{Result, WrapErr}; use futures::StreamExt; use redis::ToRedisArgs; use tokio::sync::mpsc; use tracing::{info, warn}; -use super::{Error, RedisPool}; +use super::RedisPool; const PUBSUB_BUFFER_SIZE: usize = 1024; const INITIAL_RECONNECT_BACKOFF: Duration = Duration::from_millis(250); @@ -35,21 +36,20 @@ impl RedisPool { receiver } - pub async fn publish( - &self, - channel: &str, - message: M, - ) -> Result<(), Error> + pub async fn publish(&self, channel: &str, message: M) -> Result<()> where M: ToRedisArgs + Send + Sync, { - let mut connection = self.connect().await?; + let mut connection = self + .connect() + .await + .wrap_err("connecting to Redis for publishing")?; let _: usize = redis::cmd("PUBLISH") .arg(channel) .arg(message) .query_async(&mut connection) .await - .map_err(Error::from)?; + .wrap_err("publishing to Redis channel")?; Ok(()) } } @@ -111,10 +111,17 @@ async fn forward_from_seed( seed_url: &str, channel: &'static str, sender: &mpsc::Sender>, -) -> redis::RedisResult { - let client = redis::Client::open(seed_url)?; - let mut pubsub = client.get_async_pubsub().await?; - pubsub.subscribe(channel).await?; +) -> Result { + let client = redis::Client::open(seed_url) + .wrap_err("configuring Redis Pub/Sub client")?; + let mut pubsub = client + .get_async_pubsub() + .await + .wrap_err("connecting to Redis Pub/Sub")?; + pubsub + .subscribe(channel) + .await + .wrap_err("subscribing to Redis channel")?; info!(channel, "Established Redis Pub/Sub subscription"); let mut stream = pubsub.into_on_message();