Skip to content

PG/RedisPool configuration/o11y improvements #5032

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
fetchfern wants to merge 9 commits into
base: main
Choose a base branch
from
Open

PG/RedisPool configuration/o11y improvements #5032

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
87f7708
Don't retain Redis connections while doing database queries
fetchfern Jan 3, 2026
4c3277b
Optional REDIS_WAIT_TIMEOUT_MS
fetchfern Jan 3, 2026
85c446b
Attach more data to CacheTimeout errrors
fetchfern Jan 3, 2026
4904a67
Fix locks_released
fetchfern Jan 3, 2026
1c6e941
Fmt
fetchfern Jan 3, 2026
f3dab34
Set default REDIS_WAIT_TIMEOUT_MS to 15s
fetchfern Jan 4, 2026
09d0eae
Fix lint
fetchfern Jan 4, 2026
5bd2a89
Close Redis connections idle for > 5 minutes
fetchfern Jan 4, 2026
19d18da
Exponential backoff on cache spin lock
fetchfern Jan 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
64 changes: 40 additions & 24 deletions apps/labrinth/src/database/models/categories.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -93,14 +93,16 @@ impl Category {
where
E: sqlx::Executor<'a, Database = sqlx::Postgres>,
{
let mut redis = redis.connect().await?;
{
let mut redis = redis.connect().await?;

let res: Option<Vec<Category>> = redis
.get_deserialized_from_json(TAGS_NAMESPACE, "category")
.await?;
let res: Option<Vec<Category>> = redis
.get_deserialized_from_json(TAGS_NAMESPACE, "category")
.await?;

if let Some(res) = res {
return Ok(res);
if let Some(res) = res {
return Ok(res);
}
}

let result = sqlx::query!(
Expand All @@ -122,6 +124,8 @@ impl Category {
.try_collect::<Vec<Category>>()
.await?;

let mut redis = redis.connect().await?;

redis
.set_serialized_to_json(TAGS_NAMESPACE, "category", &result, None)
.await?;
Expand Down Expand Up @@ -158,14 +162,16 @@ impl LinkPlatform {
where
E: sqlx::Executor<'a, Database = sqlx::Postgres>,
{
let mut redis = redis.connect().await?;
{
let mut redis = redis.connect().await?;

let res: Option<Vec<LinkPlatform>> = redis
.get_deserialized_from_json(TAGS_NAMESPACE, "link_platform")
.await?;
let res: Option<Vec<LinkPlatform>> = redis
.get_deserialized_from_json(TAGS_NAMESPACE, "link_platform")
.await?;

if let Some(res) = res {
return Ok(res);
if let Some(res) = res {
return Ok(res);
}
}

let result = sqlx::query!(
Expand All @@ -182,6 +188,8 @@ impl LinkPlatform {
.try_collect::<Vec<LinkPlatform>>()
.await?;

let mut redis = redis.connect().await?;

redis
.set_serialized_to_json(
TAGS_NAMESPACE,
Expand Down Expand Up @@ -223,14 +231,16 @@ impl ReportType {
where
E: sqlx::Executor<'a, Database = sqlx::Postgres>,
{
let mut redis = redis.connect().await?;
{
let mut redis = redis.connect().await?;

let res: Option<Vec<String>> = redis
.get_deserialized_from_json(TAGS_NAMESPACE, "report_type")
.await?;
let res: Option<Vec<String>> = redis
.get_deserialized_from_json(TAGS_NAMESPACE, "report_type")
.await?;

if let Some(res) = res {
return Ok(res);
if let Some(res) = res {
return Ok(res);
}
}

let result = sqlx::query!(
Expand All @@ -243,6 +253,8 @@ impl ReportType {
.try_collect::<Vec<String>>()
.await?;

let mut redis = redis.connect().await?;

redis
.set_serialized_to_json(
TAGS_NAMESPACE,
Expand Down Expand Up @@ -284,14 +296,16 @@ impl ProjectType {
where
E: sqlx::Executor<'a, Database = sqlx::Postgres>,
{
let mut redis = redis.connect().await?;
{
let mut redis = redis.connect().await?;

let res: Option<Vec<String>> = redis
.get_deserialized_from_json(TAGS_NAMESPACE, "project_type")
.await?;
let res: Option<Vec<String>> = redis
.get_deserialized_from_json(TAGS_NAMESPACE, "project_type")
.await?;

if let Some(res) = res {
return Ok(res);
if let Some(res) = res {
return Ok(res);
}
}

let result = sqlx::query!(
Expand All @@ -304,6 +318,8 @@ impl ProjectType {
.try_collect::<Vec<String>>()
.await?;

let mut redis = redis.connect().await?;

redis
.set_serialized_to_json(
TAGS_NAMESPACE,
Expand Down
85 changes: 52 additions & 33 deletions apps/labrinth/src/database/models/loader_fields.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,12 +50,14 @@ impl Game {
where
E: sqlx::Executor<'a, Database = sqlx::Postgres>,
{
let mut redis = redis.connect().await?;
let cached_games: Option<Vec<Game>> = redis
.get_deserialized_from_json(GAMES_LIST_NAMESPACE, "games")
.await?;
if let Some(cached_games) = cached_games {
return Ok(cached_games);
{
let mut redis = redis.connect().await?;
let cached_games: Option<Vec<Game>> = redis
.get_deserialized_from_json(GAMES_LIST_NAMESPACE, "games")
.await?;
if let Some(cached_games) = cached_games {
return Ok(cached_games);
}
}

let result = sqlx::query!(
Expand All @@ -74,6 +76,8 @@ impl Game {
.try_collect::<Vec<Game>>()
.await?;

let mut redis = redis.connect().await?;

redis
.set_serialized_to_json(
GAMES_LIST_NAMESPACE,
Expand Down Expand Up @@ -106,11 +110,13 @@ impl Loader {
where
E: sqlx::Executor<'a, Database = sqlx::Postgres>,
{
let mut redis = redis.connect().await?;
let cached_id: Option<i32> =
redis.get_deserialized_from_json(LOADER_ID, name).await?;
if let Some(cached_id) = cached_id {
return Ok(Some(LoaderId(cached_id)));
{
let mut redis = redis.connect().await?;
let cached_id: Option<i32> =
redis.get_deserialized_from_json(LOADER_ID, name).await?;
if let Some(cached_id) = cached_id {
return Ok(Some(LoaderId(cached_id)));
}
}

let result = sqlx::query!(
Expand All @@ -125,6 +131,7 @@ impl Loader {
.map(|r| LoaderId(r.id));

if let Some(result) = result {
let mut redis = redis.connect().await?;
redis
.set_serialized_to_json(LOADER_ID, name, &result.0, None)
.await?;
Expand All @@ -140,12 +147,14 @@ impl Loader {
where
E: sqlx::Executor<'a, Database = sqlx::Postgres>,
{
let mut redis = redis.connect().await?;
let cached_loaders: Option<Vec<Loader>> = redis
.get_deserialized_from_json(LOADERS_LIST_NAMESPACE, "all")
.await?;
if let Some(cached_loaders) = cached_loaders {
return Ok(cached_loaders);
{
let mut redis = redis.connect().await?;
let cached_loaders: Option<Vec<Loader>> = redis
.get_deserialized_from_json(LOADERS_LIST_NAMESPACE, "all")
.await?;
if let Some(cached_loaders) = cached_loaders {
return Ok(cached_loaders);
}
}

let result = sqlx::query!(
Expand Down Expand Up @@ -180,6 +189,8 @@ impl Loader {
.try_collect::<Vec<_>>()
.await?;

let mut redis = redis.connect().await?;

redis
.set_serialized_to_json(
LOADERS_LIST_NAMESPACE,
Expand Down Expand Up @@ -455,15 +466,17 @@ impl LoaderField {
where
E: sqlx::Executor<'a, Database = sqlx::Postgres>,
{
let mut redis = redis.connect().await?;
{
let mut redis = redis.connect().await?;

let cached_fields: Option<Vec<LoaderField>> = redis
.get(LOADER_FIELDS_NAMESPACE_ALL, "")
.await?
.and_then(|x| serde_json::from_str::<Vec<LoaderField>>(&x).ok());
let cached_fields: Option<Vec<LoaderField>> =
redis.get(LOADER_FIELDS_NAMESPACE_ALL, "").await?.and_then(
|x| serde_json::from_str::<Vec<LoaderField>>(&x).ok(),
);

if let Some(cached_fields) = cached_fields {
return Ok(cached_fields);
if let Some(cached_fields) = cached_fields {
return Ok(cached_fields);
}
}

let result = sqlx::query!(
Expand All @@ -489,6 +502,8 @@ impl LoaderField {
.flatten()
.collect();

let mut redis = redis.connect().await?;

redis
.set_serialized_to_json(
LOADER_FIELDS_NAMESPACE_ALL,
Expand All @@ -510,16 +525,18 @@ impl LoaderFieldEnum {
where
E: sqlx::Executor<'a, Database = sqlx::Postgres>,
{
let mut redis = redis.connect().await?;
{
let mut redis = redis.connect().await?;

let cached_enum = redis
.get_deserialized_from_json(
LOADER_FIELD_ENUMS_ID_NAMESPACE,
enum_name,
)
.await?;
if let Some(cached_enum) = cached_enum {
return Ok(cached_enum);
let cached_enum = redis
.get_deserialized_from_json(
LOADER_FIELD_ENUMS_ID_NAMESPACE,
enum_name,
)
.await?;
if let Some(cached_enum) = cached_enum {
return Ok(cached_enum);
}
}

let result = sqlx::query!(
Expand All @@ -540,6 +557,8 @@ impl LoaderFieldEnum {
hidable: l.hidable,
});

let mut redis = redis.connect().await?;

redis
.set_serialized_to_json(
LOADER_FIELD_ENUMS_ID_NAMESPACE,
Expand Down
11 changes: 9 additions & 2 deletions apps/labrinth/src/database/models/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,13 @@ pub enum DatabaseError {
SerdeCacheError(#[from] serde_json::Error),
#[error("Schema error: {0}")]
SchemaError(String),
#[error("Timeout when waiting for cache subscriber")]
CacheTimeout,
#[error(
"Timeout waiting on Redis cache lock ({locks_released}/{locks_waiting} released, spent {time_spent_pool_wait_ms}ms/{time_spent_total_ms}ms waiting on connections from pool)"
)]
CacheTimeout {
locks_released: usize,
locks_waiting: usize,
time_spent_pool_wait_ms: u64,
time_spent_total_ms: u64,
},
}
24 changes: 14 additions & 10 deletions apps/labrinth/src/database/models/notification_item.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -380,17 +380,19 @@ impl DBNotification {
where
E: sqlx::Executor<'a, Database = sqlx::Postgres> + Copy,
{
let mut redis = redis.connect().await?;

let cached_notifications: Option<Vec<DBNotification>> = redis
.get_deserialized_from_json(
USER_NOTIFICATIONS_NAMESPACE,
&user_id.0.to_string(),
)
.await?;
{
let mut redis = redis.connect().await?;

let cached_notifications: Option<Vec<DBNotification>> = redis
.get_deserialized_from_json(
USER_NOTIFICATIONS_NAMESPACE,
&user_id.0.to_string(),
)
.await?;

if let Some(notifications) = cached_notifications {
return Ok(notifications);
if let Some(notifications) = cached_notifications {
return Ok(notifications);
}
}

let db_notifications = sqlx::query!(
Expand Down Expand Up @@ -437,6 +439,8 @@ impl DBNotification {
.try_collect::<Vec<DBNotification>>()
.await?;

let mut redis = redis.connect().await?;

redis
.set_serialized_to_json(
USER_NOTIFICATIONS_NAMESPACE,
Expand Down
23 changes: 15 additions & 8 deletions apps/labrinth/src/database/models/notifications_template_item.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,14 +52,19 @@ impl NotificationTemplate {
exec: impl sqlx::Executor<'_, Database = sqlx::Postgres>,
redis: &RedisPool,
) -> Result<Vec<NotificationTemplate>, DatabaseError> {
let mut redis = redis.connect().await?;

let maybe_cached_templates = redis
.get_deserialized_from_json(TEMPLATES_NAMESPACE, channel.as_str())
.await?;

if let Some(cached) = maybe_cached_templates {
return Ok(cached);
{
let mut redis = redis.connect().await?;

let maybe_cached_templates = redis
.get_deserialized_from_json(
TEMPLATES_NAMESPACE,
channel.as_str(),
)
.await?;

if let Some(cached) = maybe_cached_templates {
return Ok(cached);
}
}

let results = sqlx::query_as!(
Expand All @@ -74,6 +79,8 @@ impl NotificationTemplate {

let templates = results.into_iter().map(Into::into).collect();

let mut redis = redis.connect().await?;

redis
.set_serialized_to_json(
TEMPLATES_NAMESPACE,
Expand Down
Loading