From 3df2ab1f42c71f7c131687d757c9506f565ef36e Mon Sep 17 00:00:00 2001 From: scanash00 Date: Thu, 13 Aug 2026 13:34:39 -0800 Subject: [PATCH] moderation/database fixes --- crates/tranquil-api/src/admin/status.rs | 81 +++++++++++++------ crates/tranquil-api/src/identity/did.rs | 14 +++- crates/tranquil-api/src/repo/blob.rs | 11 +++ crates/tranquil-api/src/repo/record/read.rs | 31 +++++-- crates/tranquil-db/src/postgres/user.rs | 55 ++++++++----- .../src/endpoints/par.rs | 18 ++++- .../src/endpoints/token/grants.rs | 10 +-- crates/tranquil-pds/tests/admin_moderation.rs | 64 +++++++++++++++ crates/tranquil-sync/src/subscribe_repos.rs | 28 +++++-- .../20260813_record_blobs_backfill_marker.sql | 10 +-- .../20260814_database_hot_path_indexes.sql | 10 +++ 11 files changed, 248 insertions(+), 84 deletions(-) create mode 100644 migrations/20260814_database_hot_path_indexes.sql diff --git a/crates/tranquil-api/src/admin/status.rs b/crates/tranquil-api/src/admin/status.rs index aad7291..86f3b45 100644 --- a/crates/tranquil-api/src/admin/status.rs +++ b/crates/tranquil-api/src/admin/status.rs @@ -162,6 +162,16 @@ pub async fn update_subject_status( Some("com.atproto.admin.defs#repoRef") => { let did_str = input.subject.get("did").and_then(Value::as_str); if let Some(did_str) = did_str { + if input.takedown.as_ref().is_some_and(|status| status.applied) + && input + .deactivated + .as_ref() + .is_some_and(|status| !status.applied) + { + return Err(ApiError::InvalidRequest( + "Cannot activate and takedown an account at the same time".into(), + )); + } let did: Did = match did_str.parse() { Ok(d) => d, Err(_) => return Err(ApiError::InvalidDid("Invalid DID format".into())), @@ -201,8 +211,8 @@ pub async fn update_subject_status( } let takedown_update = input.takedown.as_ref().map(|t| t.applied); let deactivated_update = input.deactivated.as_ref().map(|d| d.applied); - if (takedown_update.is_some() || deactivated_update.is_some()) - && let Err(e) = state + if takedown_update.is_some() || deactivated_update.is_some() { + state .repos .repo .update_repo_status( @@ -212,37 +222,58 @@ pub async fn update_subject_status( deactivated_update, ) .await - { - warn!("failed to sync status to repo backend: {e:?}"); - } - if let Some(takedown) = &input.takedown { - let status = match takedown.applied { - true => tranquil_db_traits::AccountStatus::Takendown, - false => tranquil_db_traits::AccountStatus::Active, - }; - if let Err(e) = - tranquil_pds::repo_ops::sequence_account_event(&state, &did, status).await - { - warn!("Failed to sequence account event for takedown: {}", e); - } + .map_err(|e| { + error!("Failed to sync account status to repo backend: {e:?}"); + ApiError::InternalError(Some("Failed to update account status".into())) + })?; } - if let Some(deactivated) = &input.deactivated { - let status = match deactivated.applied { - true => tranquil_db_traits::AccountStatus::Deactivated, - false => tranquil_db_traits::AccountStatus::Active, - }; - if let Err(e) = - tranquil_pds::repo_ops::sequence_account_event(&state, &did, status).await - { - warn!("Failed to sequence account event for deactivation: {}", e); - } + if input.takedown.as_ref().is_some_and(|status| status.applied) { + state + .repos + .session + .delete_sessions_by_did(&did) + .await + .map_err(|e| { + error!("Failed to revoke sessions for taken-down account: {e:?}"); + ApiError::InternalError(Some( + "Failed to revoke account sessions".into(), + )) + })?; + state + .repos + .oauth + .delete_sessions_by_did(&did) + .await + .map_err(|e| { + error!("Failed to revoke OAuth sessions for taken-down account: {e:?}"); + ApiError::InternalError(Some("Failed to revoke OAuth sessions".into())) + })?; } + tranquil_pds::auth::invalidate_auth_cache(state.cache.as_ref(), &did).await; if let Ok(Some(handle)) = state.repos.user.get_handle_by_did(&did).await { let _ = state .cache .delete(&tranquil_pds::cache_keys::handle_key(&handle)) .await; } + if takedown_update.is_some() || deactivated_update.is_some() { + let status = tranquil_pds::sync::get_account_with_status( + state.repos.repo.as_ref(), + &did, + ) + .await + .map_err(|e| { + error!("Failed to read updated account status: {e:?}"); + ApiError::InternalError(Some("Failed to read account status".into())) + })? + .ok_or(ApiError::SubjectNotFound)? + .status; + if let Err(e) = + tranquil_pds::repo_ops::sequence_account_event(&state, &did, status).await + { + warn!("Failed to sequence account status event: {}", e); + } + } return Ok(Json(json!({ "subject": input.subject, "takedown": input.takedown.as_ref().map(|t| json!({ diff --git a/crates/tranquil-api/src/identity/did.rs b/crates/tranquil-api/src/identity/did.rs index a0a9055..ccc0609 100644 --- a/crates/tranquil-api/src/identity/did.rs +++ b/crates/tranquil-api/src/identity/did.rs @@ -50,22 +50,30 @@ pub async fn resolve_handle( return ApiError::InvalidHandle(Some("Invalid handle format".into())).into_response(); } }; + let is_service_handle = tranquil_pds::handle::is_service_domain_handle( + handle.as_str(), + tranquil_config::get().server.hostname_without_port(), + ); let cache_key = tranquil_pds::cache_keys::handle_key(&handle); - if let Some(cached) = state.cache.get(&cache_key).await + if !is_service_handle + && let Some(cached) = state.cache.get(&cache_key).await && let Ok(did) = tranquil_pds::Did::new(cached) { return DidResponse::response(did).into_response(); } let user = state.repos.user.get_by_handle(&handle).await; match user { - Ok(Some(row)) => { + Ok(Some(row)) if row.takedown_ref.is_none() && row.deactivated_at.is_none() => { let _ = state .cache .set(&cache_key, &row.did, std::time::Duration::from_secs(300)) .await; DidResponse::response(row.did).into_response() } - Ok(None) => match tranquil_pds::handle::resolve_handle(&handle).await { + Ok(Some(_)) | Ok(None) if is_service_handle => { + ApiError::InvalidRequest("Unable to resolve handle".into()).into_response() + } + Ok(Some(_)) | Ok(None) => match tranquil_pds::handle::resolve_handle(&handle).await { Ok(did) => { let _ = state .cache diff --git a/crates/tranquil-api/src/repo/blob.rs b/crates/tranquil-api/src/repo/blob.rs index aa0f7b1..2cfba2c 100644 --- a/crates/tranquil-api/src/repo/blob.rs +++ b/crates/tranquil-api/src/repo/blob.rs @@ -69,6 +69,17 @@ pub async fn upload_blob( } }; + let account_status = state + .repos + .user + .get_status_by_did(&did) + .await + .log_db_err("fetching account status for blob upload")? + .ok_or(ApiError::AccountNotFound)?; + if account_status.takedown_ref.is_some() { + return Err(ApiError::AccountTakedown); + } + if state .repos .user diff --git a/crates/tranquil-api/src/repo/record/read.rs b/crates/tranquil-api/src/repo/record/read.rs index 036b13a..76dfb7c 100644 --- a/crates/tranquil-api/src/repo/record/read.rs +++ b/crates/tranquil-api/src/repo/record/read.rs @@ -16,6 +16,7 @@ use std::str::FromStr; use tracing::error; use tranquil_pds::api::error::ApiError; use tranquil_pds::state::AppState; +use tranquil_pds::sync::{RepoAccessLevel, assert_repo_availability}; use tranquil_pds::types::{AtIdentifier, Nsid, Rkey}; fn ipld_to_json(ipld: Ipld) -> Value { @@ -60,14 +61,23 @@ pub async fn get_record( _headers: HeaderMap, Query(input): Query, ) -> Response { - let user_id = match common::resolve_repo_user_id(state.repos.user.as_ref(), &input.repo).await { - Ok(id) => id, + let resolved = match common::resolve_repo(state.repos.user.as_ref(), &input.repo).await { + Ok(resolved) => resolved, Err(e) => return e.into_response(), }; + if let Err(e) = assert_repo_availability( + state.repos.repo.as_ref(), + &resolved.did, + RepoAccessLevel::Public, + ) + .await + { + return e.into_response(); + } let record_row = state .repos .repo - .get_record_cid(user_id, &input.collection, &input.rkey) + .get_record_cid(resolved.user_id, &input.collection, &input.rkey) .await; let record_cid_link = match record_row { Ok(Some(cid)) => cid, @@ -129,10 +139,19 @@ pub async fn list_records( State(state): State, Query(input): Query, ) -> Response { - let user_id = match common::resolve_repo_user_id(state.repos.user.as_ref(), &input.repo).await { - Ok(id) => id, + let resolved = match common::resolve_repo(state.repos.user.as_ref(), &input.repo).await { + Ok(resolved) => resolved, Err(e) => return e.into_response(), }; + if let Err(e) = assert_repo_availability( + state.repos.repo.as_ref(), + &resolved.did, + RepoAccessLevel::Public, + ) + .await + { + return e.into_response(); + } let limit = input.limit.unwrap_or(50).clamp(1, 100); let limit_i64 = i64::from(limit); let cursor_rkey = input @@ -143,7 +162,7 @@ pub async fn list_records( .repos .repo .list_records( - user_id, + resolved.user_id, &input.collection, cursor_rkey.as_ref(), limit_i64, diff --git a/crates/tranquil-db/src/postgres/user.rs b/crates/tranquil-db/src/postgres/user.rs index bedb465..adcbe5a 100644 --- a/crates/tranquil-db/src/postgres/user.rs +++ b/crates/tranquil-db/src/postgres/user.rs @@ -19,6 +19,8 @@ use tranquil_db_traits::{ UserVerificationInfo, UserWithKey, WebauthnChallengeType, }; +const FIRST_ADMIN_ADVISORY_LOCK_KEY: i64 = 0x7472_616e_7175_696c; + pub struct PostgresUserRepository { pool: PgPool, } @@ -77,6 +79,27 @@ async fn consume_invite_code( Ok(()) } +async fn determine_first_admin( + conn: &mut sqlx::PgConnection, +) -> Result { + let map_err = |e: sqlx::Error| tranquil_db_traits::CreateAccountError::Database(e.to_string()); + + sqlx::query!( + "SELECT pg_advisory_xact_lock($1)", + FIRST_ADMIN_ADVISORY_LOCK_KEY + ) + .execute(&mut *conn) + .await + .map_err(map_err)?; + + let count = sqlx::query_scalar!("SELECT COUNT(*) FROM users") + .fetch_one(&mut *conn) + .await + .map_err(map_err)?; + + Ok(count.unwrap_or(0) == 0) +} + #[async_trait] impl UserRepository for PostgresUserRepository { async fn get_by_did(&self, did: &Did) -> Result, DbError> { @@ -2146,17 +2169,17 @@ impl UserRepository for PostgresUserRepository { async fn admin_delete_account_complete(&self, user_id: Uuid, did: &Did) -> Result<(), DbError> { let mut tx = self.pool.begin().await.map_err(map_sqlx_error)?; - sqlx::query!("DELETE FROM session_tokens WHERE did = $1", did.as_str()) - .execute(&mut *tx) - .await - .map_err(map_sqlx_error)?; sqlx::query!( "DELETE FROM used_refresh_tokens WHERE session_id IN (SELECT id FROM session_tokens WHERE did = $1)", did.as_str() ) .execute(&mut *tx) .await - .ok(); + .map_err(map_sqlx_error)?; + sqlx::query!("DELETE FROM session_tokens WHERE did = $1", did.as_str()) + .execute(&mut *tx) + .await + .map_err(map_sqlx_error)?; sqlx::query!("DELETE FROM records WHERE repo_id = $1", user_id) .execute(&mut *tx) .await @@ -2179,14 +2202,14 @@ impl UserRepository for PostgresUserRepository { ) .execute(&mut *tx) .await - .ok(); + .map_err(map_sqlx_error)?; sqlx::query!( "DELETE FROM invite_codes WHERE created_by_user = $1", user_id ) .execute(&mut *tx) .await - .ok(); + .map_err(map_sqlx_error)?; sqlx::query!("DELETE FROM user_keys WHERE user_id = $1", user_id) .execute(&mut *tx) .await @@ -2527,11 +2550,7 @@ impl UserRepository for PostgresUserRepository { tranquil_db_traits::CreateAccountError::Database(e.to_string()) })?; - let is_first_user: bool = sqlx::query_scalar!("SELECT COUNT(*) as count FROM users") - .fetch_one(&mut *tx) - .await - .map(|c| c.unwrap_or(0) == 0) - .unwrap_or(false); + let is_first_user = determine_first_admin(&mut tx).await?; let user_insert: Result<(uuid::Uuid,), _> = sqlx::query_as( r#"INSERT INTO users ( @@ -2765,11 +2784,7 @@ impl UserRepository for PostgresUserRepository { tranquil_db_traits::CreateAccountError::Database(e.to_string()) })?; - let is_first_user: bool = sqlx::query_scalar!("SELECT COUNT(*) as count FROM users") - .fetch_one(&mut *tx) - .await - .map(|c| c.unwrap_or(0) == 0) - .unwrap_or(false); + let is_first_user = determine_first_admin(&mut tx).await?; let user_insert: Result<(uuid::Uuid,), _> = sqlx::query_as( r#"INSERT INTO users ( @@ -2917,11 +2932,7 @@ impl UserRepository for PostgresUserRepository { return Err(tranquil_db_traits::CreateAccountError::InvalidToken); } - let is_first_user: bool = sqlx::query_scalar!("SELECT COUNT(*) as count FROM users") - .fetch_one(&mut *tx) - .await - .map(|c| c.unwrap_or(0) == 0) - .unwrap_or(false); + let is_first_user = determine_first_admin(&mut tx).await?; let user_insert: Result<(uuid::Uuid,), _> = sqlx::query_as( r#"INSERT INTO users ( diff --git a/crates/tranquil-oauth-server/src/endpoints/par.rs b/crates/tranquil-oauth-server/src/endpoints/par.rs index c90e5d7..15430a2 100644 --- a/crates/tranquil-oauth-server/src/endpoints/par.rs +++ b/crates/tranquil-oauth-server/src/endpoints/par.rs @@ -2,6 +2,7 @@ use axum::body::Bytes; use axum::{Json, extract::State, http::HeaderMap}; use chrono::{Duration, Utc}; use serde::{Deserialize, Serialize}; +use std::sync::atomic::{AtomicI64, Ordering}; use tranquil_pds::oauth::{ AuthorizationRequestParameters, ClientAuth, ClientMetadataCache, CodeChallengeMethod, OAuthError, Prompt, RequestData, RequestId, ResponseMode, ResponseType, @@ -12,6 +13,9 @@ use tranquil_pds::state::AppState; use tranquil_types::{ClientId, JwkThumbprint}; const PAR_EXPIRY_SECONDS: i64 = 600; +const AUTH_REQUEST_CLEANUP_INTERVAL_SECONDS: i64 = 60; + +static LAST_AUTH_REQUEST_CLEANUP: AtomicI64 = AtomicI64::new(0); #[derive(Debug, Deserialize)] pub struct ParRequest { @@ -119,14 +123,20 @@ pub async fn pushed_authorization_request( .create_authorization_request(&request_id, &request_data) .await .map_err(tranquil_pds::oauth::db_err_to_oauth)?; - tokio::spawn({ + let now = Utc::now().timestamp(); + let previous = LAST_AUTH_REQUEST_CLEANUP.load(Ordering::Relaxed); + if now.saturating_sub(previous) >= AUTH_REQUEST_CLEANUP_INTERVAL_SECONDS + && LAST_AUTH_REQUEST_CLEANUP + .compare_exchange(previous, now, Ordering::AcqRel, Ordering::Relaxed) + .is_ok() + { let oauth_repo = state.repos.oauth.clone(); - async move { + tokio::spawn(async move { if let Err(e) = oauth_repo.delete_expired_authorization_requests().await { tracing::warn!("Failed to cleanup expired authorization requests: {:?}", e); } - } - }); + }); + } Ok(( axum::http::StatusCode::CREATED, Json(ParResponse { diff --git a/crates/tranquil-oauth-server/src/endpoints/token/grants.rs b/crates/tranquil-oauth-server/src/endpoints/token/grants.rs index b9698d5..543f71f 100644 --- a/crates/tranquil-oauth-server/src/endpoints/token/grants.rs +++ b/crates/tranquil-oauth-server/src/endpoints/token/grants.rs @@ -217,15 +217,7 @@ pub async fn handle_authorization_code_grant( client_id = %authorized.client_id, "Authorization code grant completed, token created" ); - tokio::spawn({ - let oauth_repo = state.repos.oauth.clone(); - let did_clone = did.clone(); - async move { - if let Err(e) = enforce_token_limit_for_user(oauth_repo.as_ref(), &did_clone).await { - tracing::warn!("Failed to enforce token limit for user: {:?}", e); - } - } - }); + enforce_token_limit_for_user(state.repos.oauth.as_ref(), &did).await?; let mut response_headers = HeaderMap::new(); let config = AuthConfig::get(); let verifier = DPoPVerifier::new(config.dpop_secret().as_bytes()); diff --git a/crates/tranquil-pds/tests/admin_moderation.rs b/crates/tranquil-pds/tests/admin_moderation.rs index 03f6b86..2adefa2 100644 --- a/crates/tranquil-pds/tests/admin_moderation.rs +++ b/crates/tranquil-pds/tests/admin_moderation.rs @@ -128,6 +128,24 @@ async fn test_update_subject_status_takedown_user_without_ref_updates_sync_statu let client = client(); let (admin_jwt, _) = create_admin_account_and_login(&client).await; let (_, target_did) = create_account_and_login(&client).await; + let target_did_parsed = target_did.parse().unwrap(); + let target_handle = get_test_repos() + .await + .user + .get_handle_by_did(&target_did_parsed) + .await + .unwrap() + .unwrap(); + let resolve_before = client + .get(format!( + "{}/xrpc/com.atproto.identity.resolveHandle", + base_url().await + )) + .query(&[("handle", target_handle.as_str())]) + .send() + .await + .unwrap(); + assert_eq!(resolve_before.status(), StatusCode::OK); let payload = json!({ "subject": { "$type": "com.atproto.admin.defs#repoRef", @@ -149,6 +167,52 @@ async fn test_update_subject_status_takedown_user_without_ref_updates_sync_statu .expect("Failed to send request"); assert_eq!(res.status(), StatusCode::OK); + let resolve_after = client + .get(format!( + "{}/xrpc/com.atproto.identity.resolveHandle", + base_url().await + )) + .query(&[("handle", target_handle.as_str())]) + .send() + .await + .unwrap(); + assert_eq!(resolve_after.status(), StatusCode::BAD_REQUEST); + let resolve_body: Value = resolve_after.json().await.unwrap(); + assert_eq!(resolve_body["error"], "InvalidRequest"); + + let get_record_res = client + .get(format!( + "{}/xrpc/com.atproto.repo.getRecord", + base_url().await + )) + .query(&[ + ("repo", target_did.as_str()), + ("collection", "app.bsky.feed.post"), + ("rkey", "test"), + ]) + .send() + .await + .unwrap(); + assert_eq!(get_record_res.status(), StatusCode::BAD_REQUEST); + let get_record_body: Value = get_record_res.json().await.unwrap(); + assert_eq!(get_record_body["error"], "RepoTakendown"); + + let list_records_res = client + .get(format!( + "{}/xrpc/com.atproto.repo.listRecords", + base_url().await + )) + .query(&[ + ("repo", target_did.as_str()), + ("collection", "app.bsky.feed.post"), + ]) + .send() + .await + .unwrap(); + assert_eq!(list_records_res.status(), StatusCode::BAD_REQUEST); + let list_records_body: Value = list_records_res.json().await.unwrap(); + assert_eq!(list_records_body["error"], "RepoTakendown"); + let describe_res = client .get(format!( "{}/xrpc/com.atproto.repo.describeRepo", diff --git a/crates/tranquil-sync/src/subscribe_repos.rs b/crates/tranquil-sync/src/subscribe_repos.rs index ac818b4..1d4da18 100644 --- a/crates/tranquil-sync/src/subscribe_repos.rs +++ b/crates/tranquil-sync/src/subscribe_repos.rs @@ -4,8 +4,11 @@ use axum::{ }; use futures::{sink::SinkExt, stream::StreamExt}; use serde::Deserialize; -use std::sync::atomic::{AtomicUsize, Ordering}; -use tokio::sync::broadcast::error::RecvError; +use std::sync::{ + LazyLock, + atomic::{AtomicUsize, Ordering}, +}; +use tokio::sync::{Semaphore, broadcast::error::RecvError}; use tracing::{error, info, warn}; use tranquil_db_traits::SequenceNumber; use tranquil_pds::state::AppState; @@ -17,8 +20,11 @@ use tranquil_pds::sync::util::{ }; const BACKFILL_BATCH_SIZE: i64 = 1000; +const BACKFILL_DB_CONCURRENCY_LIMIT: usize = 8; static SUBSCRIBER_COUNT: AtomicUsize = AtomicUsize::new(0); +static BACKFILL_DB_PERMITS: LazyLock = + LazyLock::new(|| Semaphore::new(BACKFILL_DB_CONCURRENCY_LIMIT)); #[derive(Deserialize)] pub struct SubscribeReposParams { @@ -60,12 +66,14 @@ async fn recover_lagged_events( return Ok(()); } loop { - let events = match state + let permit = BACKFILL_DB_PERMITS.acquire().await.map_err(|_| ())?; + let events_result = state .repos .repo .get_events_since_cursor(*last_seen, BACKFILL_BATCH_SIZE) - .await - { + .await; + drop(permit); + let events = match events_result { Ok(e) => e, Err(e) => { error!("Lag recovery DB query failed: {:?}", e); @@ -183,11 +191,13 @@ async fn handle_socket_inner( last_seen = current_cursor; loop { + let permit = BACKFILL_DB_PERMITS.acquire().await.map_err(|_| ())?; let events = state .repos .repo .get_events_since_cursor(current_cursor, BACKFILL_BATCH_SIZE) .await; + drop(permit); match events { Ok(events) => { if events.is_empty() { @@ -233,7 +243,13 @@ async fn handle_socket_inner( } } - let cutover_events = state.repos.repo.get_events_since_seq(last_seen, None).await; + let permit = BACKFILL_DB_PERMITS.acquire().await.map_err(|_| ())?; + let cutover_events = state + .repos + .repo + .get_events_since_seq(last_seen, Some(BACKFILL_BATCH_SIZE)) + .await; + drop(permit); if let Ok(events) = cutover_events && !events.is_empty() diff --git a/migrations/20260813_record_blobs_backfill_marker.sql b/migrations/20260813_record_blobs_backfill_marker.sql index 6c5f1e8..b667851 100644 --- a/migrations/20260813_record_blobs_backfill_marker.sql +++ b/migrations/20260813_record_blobs_backfill_marker.sql @@ -2,12 +2,4 @@ ALTER TABLE repos ADD COLUMN record_blobs_backfilled_at TIMESTAMPTZ; ALTER TABLE repos - ALTER COLUMN record_blobs_backfilled_at SET DEFAULT NOW(); - -UPDATE repos r -SET record_blobs_backfilled_at = NOW() -WHERE EXISTS ( - SELECT 1 - FROM record_blobs rb - WHERE rb.repo_id = r.user_id -); \ No newline at end of file + ALTER COLUMN record_blobs_backfilled_at SET DEFAULT NOW(); \ No newline at end of file diff --git a/migrations/20260814_database_hot_path_indexes.sql b/migrations/20260814_database_hot_path_indexes.sql new file mode 100644 index 0000000..eeb9a46 --- /dev/null +++ b/migrations/20260814_database_hot_path_indexes.sql @@ -0,0 +1,10 @@ +CREATE INDEX IF NOT EXISTS idx_record_blobs_repo_blob + ON record_blobs (repo_id, blob_cid); + +CREATE INDEX IF NOT EXISTS idx_comms_jobs_stale_processing + ON comms_queue (updated_at, scheduled_for, id) + WHERE status = 'processing'; + +CREATE INDEX IF NOT EXISTS idx_users_lower_email + ON users (LOWER(email)) + WHERE email IS NOT NULL; \ No newline at end of file -- 2.51.2