MrSphay/Modrinth-plus
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add SQLx operation tracing (#5223)

Browse Source 
* wip: vendor sqlx-tracing

* (compiles) standardize pg types used

* more standardization

* general log message improvements

* wip: improve sqlx-tracing architecture

* unify sqlx::Executor type

* wip: try fix sqlx tracing

* wip: sqlx-tracing compiles

* so close

* it compiles

* fix ci
This commit is contained in:
aecsocket
2026-01-28 13:38:57 +00:00
committed by GitHub
parent 7cb7e881fa
commit e57c15b3ce
146 changed files with 7320 additions and 801 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
apps/labrinth/src/queue/analytics.rs
View File

@@ -1,3 +1,4 @@
use crate::database::PgPool;
use crate::database::models::DatabaseError;
use crate::database::redis::RedisPool;
use crate::models::analytics::{
@@ -6,7 +7,6 @@ use crate::models::analytics::{
use crate::routes::ApiError;
use dashmap::{DashMap, DashSet};
use redis::cmd;
use sqlx::PgPool;
use std::collections::HashMap;
const DOWNLOADS_NAMESPACE: &str = "downloads";
@@ -255,7 +255,7 @@ impl AnalyticsQueue {
)
.bind(version_downloads.keys().copied().collect::<Vec<_>>())
.bind(version_downloads.values().copied().collect::<Vec<_>>())
.execute(&mut *transaction)
.execute(&mut transaction)
.await?;
sqlx::query(
@@ -268,7 +268,7 @@ impl AnalyticsQueue {
)
.bind(project_downloads.keys().copied().collect::<Vec<_>>())
.bind(project_downloads.values().copied().collect::<Vec<_>>())
.execute(&mut *transaction)
.execute(&mut transaction)
.await?;
transaction.commit().await?;

18
apps/labrinth/src/queue/billing.rs
View File

@@ -11,6 +11,7 @@ use crate::database::models::{
product_item, user_subscription_item, users_redeemals,
};
use crate::database::redis::RedisPool;
use crate::database::{PgPool, PgTransaction};
use crate::models::billing::{
ChargeStatus, ChargeType, PaymentPlatform, Price, PriceDuration,
ProductMetadata, SubscriptionMetadata, SubscriptionStatus,
@@ -28,7 +29,6 @@ use ariadne::ids::base62_impl::to_base62;
use chrono::Utc;
use futures::FutureExt;
use futures::stream::{FuturesUnordered, StreamExt};
use sqlx::PgPool;
use std::collections::HashSet;
use std::str::FromStr;
use std::time::Instant;
@@ -49,7 +49,7 @@ async fn update_tax_amounts(
loop {
let mut txn = pg.begin().await?;
let charges = DBCharge::get_updateable_lock(&mut *txn, 5).await?;
let charges = DBCharge::get_updateable_lock(&mut txn, 5).await?;
if charges.is_empty() {
info!("No more charges to process");
@@ -273,7 +273,7 @@ async fn update_anrok_transactions(
) -> Result<(), ApiError> {
async fn process_charge(
stripe_client: &stripe::Client,
txn: &mut sqlx::PgTransaction<'_>,
txn: &mut PgTransaction<'_>,
redis: &RedisPool,
anrok_client: &anrok::Client,
mut c: DBCharge,
@@ -341,7 +341,7 @@ async fn update_anrok_transactions(
.and_then(|x| x.billing_details.address);
let stripe_customer_id =
DBUser::get_id(c.user_id, &mut **txn, redis)
DBUser::get_id(c.user_id, &mut *txn, redis)
.await?
.ok_or_else(|| {
ApiError::from(DatabaseError::Database(
@@ -395,7 +395,7 @@ async fn update_anrok_transactions(
(address, tax_platform_id, customer_id)
};
let tax_id = DBProductsTaxIdentifier::get_price(c.price_id, &mut **txn)
let tax_id = DBProductsTaxIdentifier::get_price(c.price_id, &mut *txn)
.await?
.ok_or_else(|| DatabaseError::Database(sqlx::Error::RowNotFound))?;
@@ -466,7 +466,7 @@ async fn update_anrok_transactions(
let mut txn = pg.begin().await?;
let mut charges =
DBCharge::get_missing_tax_identifier_lock(&mut *txn, offset, 1)
DBCharge::get_missing_tax_identifier_lock(&mut txn, offset, 1)
.await?;
let Some(c) = charges.pop() else {
@@ -654,7 +654,7 @@ pub async fn try_process_user_redeemal(
// Update `users_redeemal`, mark subscription as redeemed.
user_redeemal.status = users_redeemals::Status::Processed;
user_redeemal.update(&mut *txn).await?;
user_redeemal.update(&mut txn).await?;
txn.commit().await?;
@@ -896,7 +896,7 @@ async fn unprovision_subscriptions(
badges.bits() as i64,
user.id as DBUserId,
)
.execute(&mut *transaction)
.execute(&mut transaction)
.await?;
true
@@ -942,7 +942,7 @@ async fn unprovision_subscriptions(
DBUsersSubscriptionsAffiliations::deactivate(
subscription.id,
&mut *transaction,
&mut transaction,
)
.await
.wrap_internal_err(

17
apps/labrinth/src/queue/email.rs
View File

@@ -4,6 +4,7 @@ use crate::database::models::notifications_deliveries_item::DBNotificationDelive
use crate::database::models::notifications_template_item::NotificationTemplate;
use crate::database::models::user_item::DBUser;
use crate::database::redis::RedisPool;
use crate::database::{PgPool, PgTransaction};
use crate::models::notifications::{NotificationBody, NotificationType};
use crate::models::v3::notifications::{
NotificationChannel, NotificationDeliveryStatus,
@@ -16,7 +17,6 @@ use lettre::transport::smtp::authentication::Credentials;
use lettre::transport::smtp::client::{Tls, TlsParameters};
use lettre::{AsyncSmtpTransport, AsyncTransport, Tokio1Executor};
use reqwest::Client;
use sqlx::PgPool;
use std::sync::Arc;
use thiserror::Error;
use tokio::sync::Mutex as TokioMutex;
@@ -204,12 +204,9 @@ impl EmailQueue {
futures.push(async move {
let mut txn = this.pg.begin().await?;
let maybe_user = DBUser::get_id(
notification.user_id,
&mut *txn,
&this.redis,
)
.await?;
let maybe_user =
DBUser::get_id(notification.user_id, &mut txn, &this.redis)
.await?;
let Some(mailbox) = maybe_user
.and_then(|user| user.email)
@@ -301,7 +298,7 @@ impl EmailQueue {
pub async fn send_one(
&self,
txn: &mut sqlx::PgTransaction<'_>,
txn: &mut PgTransaction<'_>,
notification: NotificationBody,
user_id: DBUserId,
address: Mailbox,
@@ -319,7 +316,7 @@ impl EmailQueue {
async fn send_one_with_transport(
&self,
txn: &mut sqlx::PgTransaction<'_>,
txn: &mut PgTransaction<'_>,
transport: Arc<AsyncSmtpTransport<Tokio1Executor>>,
notification: NotificationBody,
user_id: DBUserId,
@@ -330,7 +327,7 @@ impl EmailQueue {
let Some(template) = NotificationTemplate::list_channel(
NotificationChannel::Email,
&mut **txn,
&mut *txn,
&self.redis,
)
.await?

29
apps/labrinth/src/queue/email/templates.rs
View File

@@ -1,4 +1,5 @@
use super::MailError;
use crate::database::PgTransaction;
use crate::database::models::ids::*;
use crate::database::models::notifications_template_item::{
NotificationTemplate, get_or_set_cached_dynamic_html,
@@ -105,7 +106,7 @@ impl MailingIdentity {
#[allow(clippy::too_many_arguments)]
pub async fn build_email(
exec: &mut sqlx::PgTransaction<'_>,
exec: &mut PgTransaction<'_>,
redis: &RedisPool,
client: &reqwest::Client,
user_id: DBUserId,
@@ -147,7 +148,7 @@ pub async fn build_email(
reply_address,
} = from;
let db_user = DBUser::get_id(user_id, &mut **exec, redis)
let db_user = DBUser::get_id(user_id, &mut *exec, redis)
.await?
.ok_or(DatabaseError::Database(sqlx::Error::RowNotFound))?;
@@ -297,7 +298,7 @@ enum EmailTemplate {
}
async fn collect_template_variables(
exec: &mut sqlx::PgTransaction<'_>,
exec: &mut PgTransaction<'_>,
redis: &RedisPool,
user_id: DBUserId,
n: &NotificationBody,
@@ -339,7 +340,7 @@ async fn collect_template_variables(
"#,
report_id.0 as i64
)
.fetch_one(&mut **exec)
.fetch_one(&mut *exec)
.await?;
map.insert(REPORT_ID, to_base62(report_id.0));
@@ -361,7 +362,7 @@ async fn collect_template_variables(
"#,
report_id.0 as i64
)
.fetch_one(&mut **exec)
.fetch_one(&mut *exec)
.await?;
map.insert(REPORT_TITLE, result.title);
@@ -376,7 +377,7 @@ async fn collect_template_variables(
"#,
project_id.0 as i64
)
.fetch_one(&mut **exec)
.fetch_one(&mut *exec)
.await?;
map.insert(PROJECT_ID, to_base62(project_id.0));
@@ -414,7 +415,7 @@ async fn collect_template_variables(
} => {
let project = DBProject::get_id(
DBProjectId(project_id.0 as i64),
&mut **exec,
&mut *exec,
redis,
)
.await?
@@ -428,7 +429,7 @@ async fn collect_template_variables(
if let Some(new_owner_user_id) = new_owner_user_id {
let user = DBUser::get_id(
DBUserId(new_owner_user_id.0 as i64),
&mut **exec,
&mut *exec,
redis,
)
.await?
@@ -444,7 +445,7 @@ async fn collect_template_variables(
{
let org = DBOrganization::get_id(
DBOrganizationId(new_owner_organization_id.0 as i64),
&mut **exec,
&mut *exec,
redis,
)
.await?
@@ -484,7 +485,7 @@ async fn collect_template_variables(
project_id.0 as i64,
user_id.0 as i64
)
.fetch_one(&mut **exec)
.fetch_one(&mut *exec)
.await?;
map.insert(TEAMINVITE_INVITER_NAME, result.inviter_name);
@@ -516,7 +517,7 @@ async fn collect_template_variables(
organization_id.0 as i64,
user_id.0 as i64
)
.fetch_one(&mut **exec)
.fetch_one(&mut *exec)
.await?;
map.insert(ORGINVITE_INVITER_NAME, result.inviter_name);
@@ -544,7 +545,7 @@ async fn collect_template_variables(
project_id.0 as i64,
user_id.0 as i64,
)
.fetch_one(&mut **exec)
.fetch_one(&mut *exec)
.await?;
map.insert(STATUSCHANGE_PROJECT_NAME, result.project_name);
@@ -706,12 +707,12 @@ async fn collect_template_variables(
// Resolve product metadata via price_id join
if let Some(info) = crate::database::models::user_subscription_item::DBUserSubscription::get(
(*subscription_id).into(),
&mut **exec,
&mut *exec,
)
.await
.ok()
.flatten()
&& let Ok(Some(pinfo)) = crate::database::models::products_tax_identifier_item::product_info_by_product_price_id(info.price_id, &mut **exec).await {
&& let Ok(Some(pinfo)) = crate::database::models::products_tax_identifier_item::product_info_by_product_price_id(info.price_id, &mut *exec).await {
let label = match pinfo.product_metadata {
crate::models::billing::ProductMetadata::Pyro { .. } => "server".to_string(),
crate::models::billing::ProductMetadata::Medal { .. } => "server".to_string(),

2
apps/labrinth/src/queue/moderation.rs
View File

@@ -1,5 +1,6 @@
use crate::auth::checks::filter_visible_versions;
use crate::database;
use crate::database::PgPool;
use crate::database::models::notification_item::NotificationBuilder;
use crate::database::models::thread_item::ThreadMessageBuilder;
use crate::database::redis::RedisPool;
@@ -14,7 +15,6 @@ use hex::ToHex;
use itertools::Itertools;
use serde::{Deserialize, Serialize};
use sha1::Digest;
use sqlx::PgPool;
use std::collections::HashMap;
use std::fmt::Write;
use std::io::{Cursor, Read};

14
apps/labrinth/src/queue/payouts/affiliate.rs
View File

@@ -1,7 +1,7 @@
use crate::database::PgPool;
use chrono::{Datelike, Duration, TimeZone, Utc};
use eyre::{Context, Result, eyre};
use rust_decimal::{Decimal, dec};
use sqlx::PgPool;
use tracing::warn;
use crate::database::models::{DBAffiliateCodeId, DBUserId};
@@ -58,7 +58,7 @@ pub async fn process_affiliate_payouts(postgres: &PgPool) -> Result<()> {
)
"#
)
.fetch_all(&mut *txn)
.fetch_all(&mut txn)
.await
.wrap_err("failed to fetch charges awaiting affiliate payout")?;
@@ -147,7 +147,7 @@ pub async fn process_affiliate_payouts(postgres: &PgPool) -> Result<()> {
available,
affiliate_code_id as _,
)
.fetch_one(&mut *txn)
.fetch_one(&mut txn)
.await
.wrap_err_with(|| eyre!("failed to insert payout value for ({affiliate_user_id:?}, {affiliate_code_id:?})"))?
.id;
@@ -170,7 +170,7 @@ pub async fn process_affiliate_payouts(postgres: &PgPool) -> Result<()> {
&insert_usap_affiliate_codes[..],
&insert_usap_payout_values[..],
)
.execute(&mut *txn)
.execute(&mut txn)
.await
.wrap_err("failed to associate charges with affiliate payouts")?;
@@ -221,7 +221,7 @@ pub async fn remove_payouts_for_refunded_charges(
AND refund_charges.charge_type = 'refund'
"#
)
.fetch_all(&mut *txn)
.fetch_all(&mut txn)
.await
.wrap_err("failed to fetch refundable affiliate payouts")?;
@@ -248,7 +248,7 @@ pub async fn remove_payouts_for_refunded_charges(
",
&usap_ids[..]
)
.execute(&mut *txn)
.execute(&mut txn)
.await
.wrap_err("failed to delete affiliate payout associations")?;
@@ -260,7 +260,7 @@ pub async fn remove_payouts_for_refunded_charges(
",
&payout_value_ids[..]
)
.execute(&mut *txn)
.execute(&mut txn)
.await
.wrap_err("failed to delete payout values")?;

6
apps/labrinth/src/queue/payouts/flow/mod.rs
View File

@@ -4,7 +4,6 @@
use eyre::eyre;
use modrinth_util::decimal::Decimal2dp;
use rust_decimal::Decimal;
use sqlx::PgTransaction;
use thiserror::Error;
pub mod mural;
@@ -12,7 +11,10 @@ pub mod paypal;
pub mod tremendous;
use crate::{
database::models::{DBPayoutId, DBUser},
database::{
PgTransaction,
models::{DBPayoutId, DBUser},
},
models::payouts::{PayoutMethodRequest, Withdrawal},
queue::payouts::PayoutsQueue,
routes::ApiError,

20
apps/labrinth/src/queue/payouts/mod.rs
View File

@@ -1,6 +1,7 @@
use crate::database::models::notification_item::NotificationBuilder;
use crate::database::models::payouts_values_notifications;
use crate::database::redis::RedisPool;
use crate::database::{PgPool, PgTransaction};
use crate::models::payouts::{
PayoutDecimal, PayoutInterval, PayoutMethod, PayoutMethodType,
TremendousForexResponse,
@@ -26,7 +27,6 @@ use rust_decimal::prelude::ToPrimitive;
use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use sqlx::PgPool;
use sqlx::postgres::PgQueryResult;
use std::collections::HashMap;
use tokio::sync::RwLock;
@@ -1023,7 +1023,7 @@ pub async fn process_payout(
.map(|x| x.to_string())
.collect::<Vec<String>>(),
)
.fetch(&mut *transaction)
.fetch(&mut transaction)
.try_fold(DashMap::new(), |acc: DashMap<i64, HashMap<i64, Decimal>>, r| {
acc.entry(r.id)
.or_default()
@@ -1046,7 +1046,7 @@ pub async fn process_payout(
.map(|x| x.to_string())
.collect::<Vec<String>>(),
)
.fetch(&mut *transaction)
.fetch(&mut transaction)
.try_fold(
DashMap::new(),
|acc: DashMap<i64, HashMap<i64, Decimal>>, r| {
@@ -1193,7 +1193,7 @@ pub async fn process_payout(
&insert_starts[..],
&insert_availables[..]
)
.execute(&mut *transaction)
.execute(&mut transaction)
.await?;
transaction.commit().await?;
@@ -1208,7 +1208,7 @@ pub async fn insert_payouts(
insert_payouts: Vec<Decimal>,
insert_starts: Vec<DateTime<Utc>>,
insert_availables: Vec<DateTime<Utc>>,
transaction: &mut sqlx::Transaction<'_, sqlx::Postgres>,
transaction: &mut PgTransaction<'_>,
) -> sqlx::Result<PgQueryResult> {
sqlx::query!(
"
@@ -1221,7 +1221,7 @@ pub async fn insert_payouts(
&insert_starts[..],
&insert_availables[..],
)
.execute(&mut **transaction)
.execute(&mut *transaction)
.await
}
@@ -1234,11 +1234,11 @@ pub async fn index_payouts_notifications(
let mut transaction = pool.begin().await?;
payouts_values_notifications::synchronize_future_payout_values(
&mut *transaction,
&mut transaction,
200,
)
.await?;
let items = payouts_values_notifications::PayoutsValuesNotification::unnotified_users_with_available_payouts_with_limit(&mut *transaction, 200).await?;
let items = payouts_values_notifications::PayoutsValuesNotification::unnotified_users_with_available_payouts_with_limit(&mut transaction, 200).await?;
let payout_ref_ids = items.iter().map(|x| x.id).collect::<Vec<_>>();
let dates_available =
@@ -1254,7 +1254,7 @@ pub async fn index_payouts_notifications(
.await?;
payouts_values_notifications::PayoutsValuesNotification::set_notified_many(
&payout_ref_ids,
&mut *transaction,
&mut transaction,
)
.await?;
@@ -1320,7 +1320,7 @@ pub async fn insert_bank_balances_and_webhook(
&insert_pending[..],
&insert_recorded[..],
)
.fetch_one(&mut *transaction)
.fetch_one(&mut transaction)
.await?;
if inserted {

8
apps/labrinth/src/queue/payouts/mural.rs
View File

@@ -1,10 +1,10 @@
use crate::database::PgPool;
use chrono::Utc;
use eyre::{Result, eyre};
use futures::{StreamExt, TryFutureExt, stream::FuturesUnordered};
use modrinth_util::decimal::Decimal2dp;
use rust_decimal::{Decimal, prelude::ToPrimitive};
use serde::{Deserialize, Serialize};
use sqlx::PgPool;
use tracing::{info, trace, warn};
use crate::{
@@ -177,7 +177,7 @@ pub async fn sync_pending_payouts_from_mural(
.collect::<Vec<String>>(),
i64::from(limit),
)
.fetch_all(&mut *txn)
.fetch_all(&mut txn)
.await
.wrap_internal_err("failed to fetch incomplete Mural payouts")?;
@@ -235,7 +235,7 @@ pub async fn sync_pending_payouts_from_mural(
&payout_ids,
&payout_statuses,
)
.execute(&mut *txn)
.execute(&mut txn)
.await
.wrap_internal_err("failed to update payout statuses")?;
@@ -452,7 +452,7 @@ mod tests {
}
async fn setup_test_db_with_payouts(
db: &sqlx::PgPool,
db: &PgPool,
payouts: Vec<(i64, String, PayoutStatus)>,
) -> Result<(), eyre::Error> {
for (id, platform_id, status) in payouts {

16
apps/labrinth/src/queue/session.rs
View File

@@ -4,10 +4,10 @@ use crate::database::models::{
DBOAuthAccessTokenId, DBPatId, DBSessionId, DBUserId, DatabaseError,
};
use crate::database::redis::RedisPool;
use crate::database::{PgPool, PgTransaction};
use crate::routes::internal::session::SessionMetadata;
use chrono::Utc;
use itertools::Itertools;
use sqlx::PgPool;
use std::collections::{HashMap, HashSet};
use tokio::sync::Mutex;
@@ -55,14 +55,14 @@ impl AuthQueue {
let mut queue = self.session_queue.lock().await;
let len = queue.len();
std::mem::replace(&mut queue, HashMap::with_capacity(len))
std::mem::replace(&mut *queue, HashMap::with_capacity(len))
}
pub async fn take_hashset<T>(queue: &Mutex<HashSet<T>>) -> HashSet<T> {
let mut queue = queue.lock().await;
let len = queue.len();
std::mem::replace(&mut queue, HashSet::with_capacity(len))
std::mem::replace(&mut *queue, HashSet::with_capacity(len))
}
pub async fn index(
@@ -100,7 +100,7 @@ impl AuthQueue {
metadata.platform,
metadata.user_agent,
)
.execute(&mut *transaction)
.execute(&mut transaction)
.await?;
}
@@ -112,7 +112,7 @@ impl AuthQueue {
WHERE refresh_expires <= NOW()
"
)
.fetch(&mut *transaction)
.fetch(&mut transaction)
.map_ok(|x| (DBSessionId(x.id), x.session, DBUserId(x.user_id)))
.try_collect::<Vec<(DBSessionId, String, DBUserId)>>()
.await?;
@@ -143,7 +143,7 @@ impl AuthQueue {
&ids[..],
Utc::now(),
)
.execute(&mut *transaction)
.execute(&mut transaction)
.await?;
update_oauth_access_token_last_used(
@@ -162,7 +162,7 @@ impl AuthQueue {
async fn update_oauth_access_token_last_used(
oauth_access_token_queue: HashSet<DBOAuthAccessTokenId>,
transaction: &mut sqlx::Transaction<'_, sqlx::Postgres>,
transaction: &mut PgTransaction<'_>,
) -> Result<(), DatabaseError> {
let ids = oauth_access_token_queue.iter().map(|id| id.0).collect_vec();
sqlx::query!(
@@ -175,7 +175,7 @@ async fn update_oauth_access_token_last_used(
&ids[..],
Utc::now()
)
.execute(&mut **transaction)
.execute(&mut *transaction)
.await?;
Ok(())
}