From d0afc1034ba48a66c2f6790c407ab5cc0bb437b5 Mon Sep 17 00:00:00 2001 From: forkwright Date: Thu, 2 Jul 2026 12:19:17 -0500 Subject: [PATCH] fix(acquisition): low-severity audit batch findings (zetesis/ergasia/syntaxis/aitesis/syndesmos) Applies the verified low-severity findings across the acquisition crates: torznab/newznab parsing, extraction, dispatch, and request-workflow polish with added coverage. Closes #447 Closes #381 Closes #461 Closes #446 Closes #435 Closes #445 Gate-Passed: kanon 0.1.5 +stages:fmt,check,clippy,nextest,lint sha:c21076cbd9cbfbe6348cad404180ede27bc73199 --- Cargo.lock | 1 + crates/aitesis/Cargo.toml | 1 + crates/aitesis/src/lib.rs | 242 +++++++++++++++--- crates/aitesis/src/limits.rs | 24 +- crates/aitesis/src/repo.rs | 203 +++++++++++++-- crates/archon/src/serve.rs | 19 +- .../archon/tests/acquisition_integration.rs | 19 +- crates/ergasia/src/error.rs | 8 + crates/ergasia/src/extract/rar.rs | 50 ++++ crates/ergasia/src/extract/seven_zip.rs | 75 ++++++ crates/ergasia/src/session.rs | 109 +++++++- crates/paroche/src/routes/request.rs | 39 ++- crates/paroche/src/state.rs | 20 ++ crates/syndesmos/src/events.rs | 110 +++++++- crates/syndesmos/src/lastfm/auth.rs | 64 ++++- crates/syndesmos/src/plex/mod.rs | 14 + crates/syntaxis/src/error.rs | 5 +- crates/syntaxis/src/lib.rs | 81 +++++- crates/zetesis/src/client/newznab.rs | 20 +- crates/zetesis/src/client/torznab.rs | 52 +++- crates/zetesis/src/client/xml.rs | 100 +++++++- crates/zetesis/src/rate_limit.rs | 51 +++- crates/zetesis/src/search.rs | 162 ++++++++++-- 23 files changed, 1331 insertions(+), 138 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 855fff43..a1b107c0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -40,6 +40,7 @@ dependencies = [ "serde_json", "snafu", "sqlx", + "tempfile", "themelion", "tokio", "tracing", diff --git a/crates/aitesis/Cargo.toml b/crates/aitesis/Cargo.toml index cf03d05c..bdc4ef1f 100644 --- a/crates/aitesis/Cargo.toml +++ b/crates/aitesis/Cargo.toml @@ -20,6 +20,7 @@ jiff.workspace = true [dev-dependencies] rstest.workspace = true +tempfile = "3" serde_json.workspace = true tokio = { workspace = true, features = ["rt-multi-thread", "macros"] } diff --git a/crates/aitesis/src/lib.rs b/crates/aitesis/src/lib.rs index bb70c096..ca976a3e 100644 --- a/crates/aitesis/src/lib.rs +++ b/crates/aitesis/src/lib.rs @@ -12,9 +12,11 @@ pub mod repo; pub mod types; pub mod workflow; +use apotheke::error::TransactionSnafu; pub use approval::{IdentityValidator, MonitorService, UserRoleProvider}; pub use error::AitesisError; use horismos::AitesisConfig; +use snafu::ResultExt; use sqlx::SqlitePool; use themelion::{RequestId, UserId}; use tracing::instrument; @@ -58,7 +60,8 @@ pub trait RequestService: Send + Sync { /// Returns a single request by ID. async fn get_request(&self, request_id: RequestId) -> Result; - /// Lists requests, optionally filtered by user or status. + /// Lists requests, optionally filtered by user or status, windowed by + /// `limit`/`offset` (newest first). /// /// Authorization: admins may list any user's requests or all requests; /// members may only list their own (`user_id` must equal @@ -68,8 +71,20 @@ pub trait RequestService: Send + Sync { caller_id: UserId, user_id: Option, status: Option, + limit: u32, + offset: u32, ) -> Result, AitesisError>; + /// Counts requests matching the same filters as [`Self::list_requests`]. + /// + /// Same authorization rules as `list_requests`. + async fn count_requests( + &self, + caller_id: UserId, + user_id: Option, + status: Option, + ) -> Result; + /// Cancels a request. Users may cancel their own; admins may cancel any. async fn cancel_request( &self, @@ -131,15 +146,6 @@ where ) -> Result { let role = self.user_roles.role_of(user_id).await?; - limits::check_limits( - &self.read, - &user_id, - role, - self.config.max_pending_per_user, - self.config.max_requests_per_day, - ) - .await?; - let auto_approve = role == UserRole::Admin && self.config.auto_approve_admins; let now = jiff::Timestamp::now(); @@ -157,7 +163,28 @@ where created_at: now, }; - repo::insert_request(&self.write, &request).await?; + // WHY: the limit check and the insert run in ONE write transaction — + // checked against the pool they race, letting concurrent submissions + // exceed max_pending_per_user / max_requests_per_day. + let mut tx = self + .write + .begin() + .await + .context(TransactionSnafu) + .context(crate::error::DatabaseSnafu)?; + limits::check_limits( + &mut tx, + &user_id, + role, + self.config.max_pending_per_user, + self.config.max_requests_per_day, + ) + .await?; + repo::insert_request(&mut *tx, &request).await?; + tx.commit() + .await + .context(TransactionSnafu) + .context(crate::error::DatabaseSnafu)?; // WHY: the row is persisted as Submitted BEFORE the identity/monitor // handoff, so a handoff failure leaves a recoverable Submitted row @@ -226,21 +253,39 @@ where caller_id: UserId, user_id: Option, status: Option, + limit: u32, + offset: u32, ) -> Result, AitesisError> { let caller_role = self.user_roles.role_of(caller_id).await?; if caller_role != UserRole::Admin && user_id != Some(caller_id) { return InsufficientPermissionSnafu.fail(); } + let page = repo::Page::new(limit, offset); match (user_id, status) { (Some(uid), Some(st)) => { - let all = repo::list_by_user(&self.read, &uid).await?; - Ok(all.into_iter().filter(|r| r.status == st).collect()) + repo::list_by_user_and_status(&self.read, &uid, st, page).await } - (Some(uid), None) => repo::list_by_user(&self.read, &uid).await, - (None, Some(st)) => repo::list_by_status(&self.read, st).await, - (None, None) => repo::list_all(&self.read).await, + (Some(uid), None) => repo::list_by_user(&self.read, &uid, page).await, + (None, Some(st)) => repo::list_by_status(&self.read, st, page).await, + (None, None) => repo::list_all(&self.read, page).await, + } + } + + #[instrument(skip(self), fields(caller_id = %caller_id))] + async fn count_requests( + &self, + caller_id: UserId, + user_id: Option, + status: Option, + ) -> Result { + let caller_role = self.user_roles.role_of(caller_id).await?; + if caller_role != UserRole::Admin && user_id != Some(caller_id) { + return InsufficientPermissionSnafu.fail(); } + + let count = repo::count_requests(&self.read, user_id.as_ref(), status).await?; + Ok(u64::try_from(count).unwrap_or(0)) } #[instrument(skip(self), fields(request_id = %request_id, user_id = %user_id))] @@ -420,7 +465,9 @@ mod tests { .unwrap_err(); assert!(matches!(err, AitesisError::MediaIdentityInvalid { .. })); - let rows = crate::repo::list_all(&pool).await.unwrap(); + let rows = crate::repo::list_all(&pool, crate::repo::Page::new(100, 0)) + .await + .unwrap(); assert_eq!(rows.len(), 1); assert_eq!(rows[0].status, RequestStatus::Submitted); assert!(rows[0].want_id.is_none()); @@ -448,7 +495,9 @@ mod tests { .unwrap_err(); assert!(matches!(err, AitesisError::MediaIdentityInvalid { .. })); - let rows = crate::repo::list_all(&pool).await.unwrap(); + let rows = crate::repo::list_all(&pool, crate::repo::Page::new(100, 0)) + .await + .unwrap(); assert_eq!(rows.len(), 1); assert_eq!(rows[0].status, RequestStatus::Submitted); assert!(rows[0].want_id.is_none()); @@ -474,7 +523,9 @@ mod tests { .submit_request(admin_id, music_input()) .await .unwrap_err(); - let rows = crate::repo::list_all(&pool).await.unwrap(); + let rows = crate::repo::list_all(&pool, crate::repo::Page::new(100, 0)) + .await + .unwrap(); assert_eq!(rows[0].status, RequestStatus::Submitted); let working_svc = AitesisServiceImpl::new( @@ -577,6 +628,127 @@ mod tests { assert!(matches!(err, AitesisError::InsufficientPermission { .. })); } + fn raw_request(user_id: UserId) -> MediaRequest { + MediaRequest { + id: themelion::RequestId::new(), + user_id, + media_type: MediaType::Music, + title: "Test Album".to_string(), + external_id: None, + status: RequestStatus::Submitted, + decided_by: None, + decided_at: None, + deny_reason: None, + want_id: None, + created_at: jiff::Timestamp::now(), + } + } + + // ── Pagination tests ───────────────────────────────────────────────────── + + #[tokio::test] + async fn list_requests_windows_by_limit_and_offset() { + let (svc, pool) = make_service(UserRole::Admin).await; + let admin_id = UserId::new(); + let user_id = UserId::new(); + for i in 0..5 { + let mut req = raw_request(user_id); + req.title = format!("Album {i}"); + crate::repo::insert_request(&pool, &req).await.unwrap(); + } + + let first = svc.list_requests(admin_id, None, None, 2, 0).await.unwrap(); + assert_eq!(first.len(), 2); + + let tail = svc.list_requests(admin_id, None, None, 2, 4).await.unwrap(); + assert_eq!(tail.len(), 1); + + let total = svc.count_requests(admin_id, None, None).await.unwrap(); + assert_eq!(total, 5); + } + + #[tokio::test] + async fn count_requests_member_scoped_to_self() { + let (svc, pool) = make_service(UserRole::Member).await; + let member_id = UserId::new(); + let other_id = UserId::new(); + crate::repo::insert_request(&pool, &raw_request(member_id)) + .await + .unwrap(); + crate::repo::insert_request(&pool, &raw_request(other_id)) + .await + .unwrap(); + + let own = svc + .count_requests(member_id, Some(member_id), None) + .await + .unwrap(); + assert_eq!(own, 1); + + let err = svc + .count_requests(member_id, Some(other_id), None) + .await + .unwrap_err(); + assert!(matches!(err, AitesisError::InsufficientPermission { .. })); + } + + // ── Concurrency tests ──────────────────────────────────────────────────── + + #[tokio::test(flavor = "multi_thread")] + async fn concurrent_submissions_cannot_exceed_pending_limit() { + use std::sync::Arc; + + // WHY: file-backed DB — an in-memory pool gives each connection its + // own database, which would hide the cross-connection race under test. + let dir = tempfile::tempdir().unwrap(); + let url = format!( + "sqlite://{}?mode=rwc", + dir.path().join("aitesis.db").display() + ); + let pool = SqlitePool::connect(&url).await.unwrap(); + MIGRATOR.run(&pool).await.unwrap(); + + let config = AitesisConfig { + max_pending_per_user: 1, + max_requests_per_day: 100, + auto_approve_admins: false, + }; + let svc = Arc::new(AitesisServiceImpl::new( + pool.clone(), + pool.clone(), + config, + MockRoles { + role: UserRole::Member, + }, + AlwaysValidIdentity, + AlwaysCreateMonitor, + )); + let user_id = UserId::new(); + + let mut handles = Vec::new(); + for _ in 0..8 { + let svc = Arc::clone(&svc); + handles.push(tokio::spawn(async move { + svc.submit_request(user_id, music_input()).await + })); + } + let mut ok = 0usize; + for handle in handles { + if handle.await.unwrap().is_ok() { + ok += 1; + } + } + + assert_eq!( + ok, 1, + "exactly one concurrent submission may pass the limit" + ); + let pending = crate::repo::count_pending_by_user(&pool, &user_id) + .await + .unwrap(); + assert_eq!(pending, 1, "the pending limit must hold in the database"); + } + // ── Limit tests ─────────────────────────────────────────────────────────── #[tokio::test] @@ -741,7 +913,7 @@ mod tests { bob_svc.submit_request(bob, music_input()).await.unwrap(); let alice_requests = alice_svc - .list_requests(alice, Some(alice), None) + .list_requests(alice, Some(alice), None, 100, 0) .await .unwrap(); assert_eq!(alice_requests.len(), 2); @@ -785,13 +957,13 @@ mod tests { .unwrap(); let submitted = admin_svc - .list_requests(admin_id, None, Some(RequestStatus::Submitted)) + .list_requests(admin_id, None, Some(RequestStatus::Submitted), 100, 0) .await .unwrap(); assert_eq!(submitted.len(), 2); let monitoring = admin_svc - .list_requests(admin_id, None, Some(RequestStatus::Monitoring)) + .list_requests(admin_id, None, Some(RequestStatus::Monitoring), 100, 0) .await .unwrap(); assert!(monitoring.is_empty()); @@ -802,7 +974,10 @@ mod tests { let (svc, _pool) = make_service(UserRole::Member).await; let member_id = UserId::new(); - let err = svc.list_requests(member_id, None, None).await.unwrap_err(); + let err = svc + .list_requests(member_id, None, None, 100, 0) + .await + .unwrap_err(); assert!(matches!(err, AitesisError::InsufficientPermission { .. })); } @@ -814,13 +989,19 @@ mod tests { svc.submit_request(other_id, music_input()).await.unwrap(); let err = svc - .list_requests(member_id, Some(other_id), None) + .list_requests(member_id, Some(other_id), None, 100, 0) .await .unwrap_err(); assert!(matches!(err, AitesisError::InsufficientPermission { .. })); let err = svc - .list_requests(member_id, Some(other_id), Some(RequestStatus::Submitted)) + .list_requests( + member_id, + Some(other_id), + Some(RequestStatus::Submitted), + 100, + 0, + ) .await .unwrap_err(); assert!(matches!(err, AitesisError::InsufficientPermission { .. })); @@ -832,7 +1013,7 @@ mod tests { let member_id = UserId::new(); let err = svc - .list_requests(member_id, None, Some(RequestStatus::Submitted)) + .list_requests(member_id, None, Some(RequestStatus::Submitted), 100, 0) .await .unwrap_err(); assert!(matches!(err, AitesisError::InsufficientPermission { .. })); @@ -848,7 +1029,7 @@ mod tests { svc.submit_request(other_id, music_input()).await.unwrap(); let own = svc - .list_requests(member_id, Some(member_id), None) + .list_requests(member_id, Some(member_id), None, 100, 0) .await .unwrap(); assert_eq!(own.len(), 1); @@ -889,11 +1070,14 @@ mod tests { .unwrap(); member_svc.submit_request(bob, music_input()).await.unwrap(); - let all = admin_svc.list_requests(admin_id, None, None).await.unwrap(); + let all = admin_svc + .list_requests(admin_id, None, None, 100, 0) + .await + .unwrap(); assert_eq!(all.len(), 2); let bobs = admin_svc - .list_requests(admin_id, Some(bob), None) + .list_requests(admin_id, Some(bob), None, 100, 0) .await .unwrap(); assert_eq!(bobs.len(), 1); diff --git a/crates/aitesis/src/limits.rs b/crates/aitesis/src/limits.rs index e0deffd6..76ef50bf 100644 --- a/crates/aitesis/src/limits.rs +++ b/crates/aitesis/src/limits.rs @@ -2,7 +2,7 @@ //! //! Admin users are exempt from all limits. -use sqlx::SqlitePool; +use sqlx::SqliteConnection; use themelion::UserId; use crate::error::{AitesisError, RequestLimitExceededSnafu}; @@ -11,8 +11,12 @@ use crate::types::UserRole; /// Checks per-user request limits and returns an error if any limit is exceeded. /// /// Admin users are exempt. +/// +/// WHY: takes a connection (not a pool) so the caller can run the check inside +/// the same transaction as the subsequent insert — a pool-level check races +/// with concurrent submissions and lets users exceed the limits. pub(crate) async fn check_limits( - pool: &SqlitePool, + conn: &mut SqliteConnection, user_id: &UserId, role: UserRole, max_pending: u32, @@ -22,12 +26,12 @@ pub(crate) async fn check_limits( return Ok(()); } - let pending = crate::repo::count_pending_by_user(pool, user_id).await?; + let pending = crate::repo::count_pending_by_user(&mut *conn, user_id).await?; if pending >= i64::from(max_pending) { return RequestLimitExceededSnafu.fail(); } - let today = crate::repo::count_today_by_user(pool, user_id).await?; + let today = crate::repo::count_today_by_user(&mut *conn, user_id).await?; if today >= i64::from(max_per_day) { return RequestLimitExceededSnafu.fail(); } @@ -71,7 +75,8 @@ mod tests { async fn member_within_limits_passes() { let pool = setup().await; let user = UserId::new(); - let result = check_limits(&pool, &user, UserRole::Member, 25, 10).await; + let mut conn = pool.acquire().await.unwrap(); + let result = check_limits(&mut conn, &user, UserRole::Member, 25, 10).await; assert!(result.is_ok()); } @@ -85,7 +90,8 @@ mod tests { .await .unwrap(); } - let result = check_limits(&pool, &user, UserRole::Admin, 25, 10).await; + let mut conn = pool.acquire().await.unwrap(); + let result = check_limits(&mut conn, &user, UserRole::Admin, 25, 10).await; assert!(result.is_ok()); } @@ -98,7 +104,8 @@ mod tests { .await .unwrap(); } - let result = check_limits(&pool, &user, UserRole::Member, 3, 100).await; + let mut conn = pool.acquire().await.unwrap(); + let result = check_limits(&mut conn, &user, UserRole::Member, 3, 100).await; assert!(matches!( result, Err(AitesisError::RequestLimitExceeded { .. }) @@ -115,7 +122,8 @@ mod tests { .await .unwrap(); } - let result = check_limits(&pool, &user, UserRole::Member, 25, 2).await; + let mut conn = pool.acquire().await.unwrap(); + let result = check_limits(&mut conn, &user, UserRole::Member, 25, 2).await; assert!(matches!( result, Err(AitesisError::RequestLimitExceeded { .. }) diff --git a/crates/aitesis/src/repo.rs b/crates/aitesis/src/repo.rs index 922f50e9..d1e5d6fa 100644 --- a/crates/aitesis/src/repo.rs +++ b/crates/aitesis/src/repo.rs @@ -26,33 +26,57 @@ struct RequestRow { impl RequestRow { fn into_domain(self) -> Option { + // WHY: this is the decision site discarding a malformed row — the + // filter_map/and_then callers swallow the None, so an unlogged drop + // here makes a corrupt row silently vanish from every listing. + let row_id = self.id.clone(); + match self.try_into_domain() { + Ok(request) => Some(request), + Err(field) => { + tracing::warn!( + row_id = %format_args!("{row_id:02x?}"), + field, + "dropping requests row that failed to parse" + ); + None + } + } + } + + fn try_into_domain(self) -> Result { use uuid::Uuid; - let id = Uuid::from_slice(&self.id).ok()?; - let user_id_uuid = Uuid::from_slice(&self.user_id).ok()?; - let status = RequestStatus::parse(&self.status)?; - let media_type = media_type_from_str(&self.media_type)?; + let id = Uuid::from_slice(&self.id).map_err(|_| "id")?; + let user_id_uuid = Uuid::from_slice(&self.user_id).map_err(|_| "user_id")?; + let status = RequestStatus::parse(&self.status).ok_or("status")?; + let media_type = media_type_from_str(&self.media_type).ok_or("media_type")?; let decided_by = self .decided_by .as_deref() - .and_then(|b| Uuid::from_slice(b).ok()) + .map(|b| Uuid::from_slice(b).map_err(|_| "decided_by")) + .transpose()? .map(UserId::from_uuid); let decided_at = self .decided_at .as_deref() - .and_then(|s| s.parse::().ok()); + .map(|s| s.parse::().map_err(|_| "decided_at")) + .transpose()?; let want_id = self .want_id .as_deref() - .and_then(|b| Uuid::from_slice(b).ok()) + .map(|b| Uuid::from_slice(b).map_err(|_| "want_id")) + .transpose()? .map(WantId::from_uuid); - let created_at = self.created_at.parse::().ok()?; + let created_at = self + .created_at + .parse::() + .map_err(|_| "created_at")?; - Some(MediaRequest { + Ok(MediaRequest { id: RequestId::from_uuid(id), user_id: UserId::from_uuid(user_id_uuid), media_type, @@ -84,10 +108,16 @@ fn media_type_from_str(s: &str) -> Option { } /// Inserts a request row. -pub async fn insert_request( - pool: &SqlitePool, +/// +/// Generic over the executor so the limit-check + insert sequence can run +/// inside one transaction (see `submit_request`). +pub async fn insert_request<'e, E>( + executor: E, request: &MediaRequest, -) -> Result<(), crate::error::AitesisError> { +) -> Result<(), crate::error::AitesisError> +where + E: sqlx::Executor<'e, Database = sqlx::Sqlite>, +{ sqlx::query( "INSERT INTO requests (id, user_id, media_type, title, external_id, status, @@ -105,7 +135,7 @@ pub async fn insert_request( .bind(&request.deny_reason) .bind(request.want_id.as_ref().map(|id| id.as_bytes().to_vec())) .bind(request.created_at.to_string()) - .execute(pool) + .execute(executor) .await .context(DbQuerySnafu { table: "requests" }) .context(DatabaseSnafu)?; @@ -197,17 +227,54 @@ pub async fn delete_request( Ok(()) } +/// Upper bound a single page may request, regardless of caller input. +const MAX_PAGE_LIMIT: u32 = 1000; + +/// Pagination window for the list queries. +/// +/// WHY: every listing carries an explicit LIMIT/OFFSET — an unbounded +/// `SELECT *` over a large request table is an allocation hazard. +#[derive(Debug, Clone, Copy)] +pub struct Page { + limit: u32, + offset: u32, +} + +impl Page { + /// Builds a window, clamping `limit` into `1..=1000` so a hostile or + /// buggy caller cannot request an unbounded page. + #[must_use] + pub fn new(limit: u32, offset: u32) -> Self { + Self { + limit: limit.clamp(1, MAX_PAGE_LIMIT), + offset, + } + } + + fn limit_i64(self) -> i64 { + i64::from(self.limit) + } + + fn offset_i64(self) -> i64 { + i64::from(self.offset) + } +} + /// Lists requests submitted by a user, newest first. pub async fn list_by_user( pool: &SqlitePool, user_id: &UserId, + page: Page, ) -> Result, crate::error::AitesisError> { let rows = sqlx::query_as::<_, RequestRow>( "SELECT id, user_id, media_type, title, external_id, status, decided_by, decided_at, deny_reason, want_id, created_at - FROM requests WHERE user_id = ? ORDER BY created_at DESC", + FROM requests WHERE user_id = ? ORDER BY created_at DESC + LIMIT ? OFFSET ?", ) .bind(user_id.as_bytes().as_slice()) + .bind(page.limit_i64()) + .bind(page.offset_i64()) .fetch_all(pool) .await .context(DbQuerySnafu { table: "requests" }) @@ -223,13 +290,48 @@ pub async fn list_by_user( pub async fn list_by_status( pool: &SqlitePool, status: RequestStatus, + page: Page, ) -> Result, crate::error::AitesisError> { let rows = sqlx::query_as::<_, RequestRow>( "SELECT id, user_id, media_type, title, external_id, status, decided_by, decided_at, deny_reason, want_id, created_at - FROM requests WHERE status = ? ORDER BY created_at DESC", + FROM requests WHERE status = ? ORDER BY created_at DESC + LIMIT ? OFFSET ?", ) .bind(status.as_str()) + .bind(page.limit_i64()) + .bind(page.offset_i64()) + .fetch_all(pool) + .await + .context(DbQuerySnafu { table: "requests" }) + .context(DatabaseSnafu)?; + + Ok(rows + .into_iter() + .filter_map(RequestRow::into_domain) + .collect()) +} + +/// Lists a user's requests matching a status, newest first. +/// +/// WHY: filtering in SQL (not post-fetch) keeps LIMIT/OFFSET windows correct — +/// an in-memory status filter over a paginated fetch skips matching rows. +pub async fn list_by_user_and_status( + pool: &SqlitePool, + user_id: &UserId, + status: RequestStatus, + page: Page, +) -> Result, crate::error::AitesisError> { + let rows = sqlx::query_as::<_, RequestRow>( + "SELECT id, user_id, media_type, title, external_id, status, + decided_by, decided_at, deny_reason, want_id, created_at + FROM requests WHERE user_id = ? AND status = ? ORDER BY created_at DESC + LIMIT ? OFFSET ?", + ) + .bind(user_id.as_bytes().as_slice()) + .bind(status.as_str()) + .bind(page.limit_i64()) + .bind(page.offset_i64()) .fetch_all(pool) .await .context(DbQuerySnafu { table: "requests" }) @@ -242,12 +344,18 @@ pub async fn list_by_status( } /// Lists all requests, newest first. -pub async fn list_all(pool: &SqlitePool) -> Result, crate::error::AitesisError> { +pub async fn list_all( + pool: &SqlitePool, + page: Page, +) -> Result, crate::error::AitesisError> { let rows = sqlx::query_as::<_, RequestRow>( "SELECT id, user_id, media_type, title, external_id, status, decided_by, decided_at, deny_reason, want_id, created_at - FROM requests ORDER BY created_at DESC", + FROM requests ORDER BY created_at DESC + LIMIT ? OFFSET ?", ) + .bind(page.limit_i64()) + .bind(page.offset_i64()) .fetch_all(pool) .await .context(DbQuerySnafu { table: "requests" }) @@ -259,17 +367,51 @@ pub async fn list_all(pool: &SqlitePool) -> Result, crate::err .collect()) } -/// Count of requests in Submitted, Approved, or Monitoring states for a user. -pub async fn count_pending_by_user( +/// Counts requests matching the optional user/status filters. +pub async fn count_requests( pool: &SqlitePool, - user_id: &UserId, + user_id: Option<&UserId>, + status: Option, ) -> Result { + let query = match (user_id, status) { + (Some(uid), Some(st)) => sqlx::query_as::<_, (i64,)>( + "SELECT COUNT(*) FROM requests WHERE user_id = ? AND status = ?", + ) + .bind(uid.as_bytes().to_vec()) + .bind(st.as_str().to_string()), + (Some(uid), None) => { + sqlx::query_as::<_, (i64,)>("SELECT COUNT(*) FROM requests WHERE user_id = ?") + .bind(uid.as_bytes().to_vec()) + } + (None, Some(st)) => { + sqlx::query_as::<_, (i64,)>("SELECT COUNT(*) FROM requests WHERE status = ?") + .bind(st.as_str().to_string()) + } + (None, None) => sqlx::query_as::<_, (i64,)>("SELECT COUNT(*) FROM requests"), + }; + + let row = query + .fetch_one(pool) + .await + .context(DbQuerySnafu { table: "requests" }) + .context(DatabaseSnafu)?; + Ok(row.0) +} + +/// Count of requests in Submitted, Approved, or Monitoring states for a user. +pub async fn count_pending_by_user<'e, E>( + executor: E, + user_id: &UserId, +) -> Result +where + E: sqlx::Executor<'e, Database = sqlx::Sqlite>, +{ let row: (i64,) = sqlx::query_as( "SELECT COUNT(*) FROM requests WHERE user_id = ? AND status IN ('submitted', 'approved', 'monitoring')", ) .bind(user_id.as_bytes().as_slice()) - .fetch_one(pool) + .fetch_one(executor) .await .context(DbQuerySnafu { table: "requests" }) .context(DatabaseSnafu)?; @@ -277,17 +419,20 @@ pub async fn count_pending_by_user( } /// Count of requests created today (UTC) for a user. -pub async fn count_today_by_user( - pool: &SqlitePool, +pub async fn count_today_by_user<'e, E>( + executor: E, user_id: &UserId, -) -> Result { +) -> Result +where + E: sqlx::Executor<'e, Database = sqlx::Sqlite>, +{ let row: (i64,) = sqlx::query_as( "SELECT COUNT(*) FROM requests WHERE user_id = ? AND created_at >= strftime('%Y-%m-%dT00:00:00Z', 'now')", ) .bind(user_id.as_bytes().as_slice()) - .fetch_one(pool) + .fetch_one(executor) .await .context(DbQuerySnafu { table: "requests" }) .context(DatabaseSnafu)?; @@ -407,7 +552,9 @@ mod tests { .await .unwrap(); - let alice_requests = list_by_user(&pool, &alice).await.unwrap(); + let alice_requests = list_by_user(&pool, &alice, Page::new(100, 0)) + .await + .unwrap(); assert_eq!(alice_requests.len(), 2); assert!(alice_requests.iter().all(|r| r.user_id == alice)); } @@ -427,12 +574,12 @@ mod tests { .await .unwrap(); - let submitted = list_by_status(&pool, RequestStatus::Submitted) + let submitted = list_by_status(&pool, RequestStatus::Submitted, Page::new(100, 0)) .await .unwrap(); assert_eq!(submitted.len(), 2); - let approved = list_by_status(&pool, RequestStatus::Approved) + let approved = list_by_status(&pool, RequestStatus::Approved, Page::new(100, 0)) .await .unwrap(); assert_eq!(approved.len(), 1); diff --git a/crates/archon/src/serve.rs b/crates/archon/src/serve.rs index ca20cfdc..24a41937 100644 --- a/crates/archon/src/serve.rs +++ b/crates/archon/src/serve.rs @@ -351,11 +351,28 @@ impl DynRequestService for RequestAdapter { caller_id: themelion::UserId, user_id: Option, status: Option, + limit: u32, + offset: u32, ) -> RequestServiceFut<'_, Vec> { let service = Arc::clone(&self.0); Box::pin(async move { service - .list_requests(caller_id, user_id, status) + .list_requests(caller_id, user_id, status, limit, offset) + .await + .map_err(Into::into) + }) + } + + fn count_requests( + &self, + caller_id: themelion::UserId, + user_id: Option, + status: Option, + ) -> RequestServiceFut<'_, u64> { + let service = Arc::clone(&self.0); + Box::pin(async move { + service + .count_requests(caller_id, user_id, status) .await .map_err(Into::into) }) diff --git a/crates/archon/tests/acquisition_integration.rs b/crates/archon/tests/acquisition_integration.rs index 74e0a891..4df6380e 100644 --- a/crates/archon/tests/acquisition_integration.rs +++ b/crates/archon/tests/acquisition_integration.rs @@ -214,11 +214,28 @@ impl DynRequestService for MockRequestAdapter { caller_id: themelion::UserId, user_id: Option, status: Option, + limit: u32, + offset: u32, ) -> RequestServiceFut<'_, Vec> { let service = Arc::clone(&self.0); Box::pin(async move { service - .list_requests(caller_id, user_id, status) + .list_requests(caller_id, user_id, status, limit, offset) + .await + .map_err(Into::into) + }) + } + + fn count_requests( + &self, + caller_id: themelion::UserId, + user_id: Option, + status: Option, + ) -> RequestServiceFut<'_, u64> { + let service = Arc::clone(&self.0); + Box::pin(async move { + service + .count_requests(caller_id, user_id, status) .await .map_err(Into::into) }) diff --git a/crates/ergasia/src/error.rs b/crates/ergasia/src/error.rs index 65e99b78..33ea24dc 100644 --- a/crates/ergasia/src/error.rs +++ b/crates/ergasia/src/error.rs @@ -45,6 +45,14 @@ pub enum ErgasiaError { location: snafu::Location, }, + #[snafu(display("failed to delete torrent {download_id}"))] + DeleteAction { + download_id: DownloadId, + error: String, + #[snafu(implicit)] + location: snafu::Location, + }, + #[snafu(display("failed to open archive at {}", path.display()))] OpenArchive { path: PathBuf, diff --git a/crates/ergasia/src/extract/rar.rs b/crates/ergasia/src/extract/rar.rs index 033b2cbf..57c4ede0 100644 --- a/crates/ergasia/src/extract/rar.rs +++ b/crates/ergasia/src/extract/rar.rs @@ -231,4 +231,54 @@ mod tests { assert!(find_rar_first_volume(dir.path()).is_none()); } + + // NOTE: extract_rar's success path has no test — RAR compression is + // proprietary (unrar is extract-only, no OSS encoder exists), so a fixture + // archive cannot be generated at test time. Error paths are covered below. + + #[test] + fn extract_rar_missing_file_errors() { + let dir = tempfile::tempdir().unwrap(); + let output_dir = dir.path().join("output"); + fs::create_dir_all(&output_dir).unwrap(); + + let err = extract_rar(&dir.path().join("nonexistent.rar"), &output_dir).unwrap_err(); + assert!( + matches!(err, ErgasiaError::OpenArchive { .. }), + "expected OpenArchive for a missing file, got: {err}" + ); + } + + #[test] + fn extract_rar_corrupt_archive_errors() { + let dir = tempfile::tempdir().unwrap(); + let output_dir = dir.path().join("output"); + fs::create_dir_all(&output_dir).unwrap(); + let rar_path = dir.path().join("corrupt.rar"); + fs::write(&rar_path, b"Rar!\x1a\x07\x00 not actually a valid archive").unwrap(); + + let err = extract_rar(&rar_path, &output_dir).unwrap_err(); + assert!( + matches!( + err, + ErgasiaError::OpenArchive { .. } | ErgasiaError::ExtractFile { .. } + ), + "expected OpenArchive or ExtractFile for a corrupt archive, got: {err}" + ); + let leftovers: Vec<_> = fs::read_dir(&output_dir).unwrap().flatten().collect(); + assert!( + leftovers.is_empty(), + "corrupt archive must not produce output: {leftovers:?}" + ); + } + + #[test] + fn declared_uncompressed_size_missing_file_errors() { + let dir = tempfile::tempdir().unwrap(); + let err = declared_uncompressed_size(&dir.path().join("nonexistent.rar")).unwrap_err(); + assert!( + matches!(err, ErgasiaError::OpenArchive { .. }), + "expected OpenArchive, got: {err}" + ); + } } diff --git a/crates/ergasia/src/extract/seven_zip.rs b/crates/ergasia/src/extract/seven_zip.rs index dd0b0e35..38d6a2ff 100644 --- a/crates/ergasia/src/extract/seven_zip.rs +++ b/crates/ergasia/src/extract/seven_zip.rs @@ -28,3 +28,78 @@ pub(crate) fn declared_uncompressed_size(archive_path: &Path) -> Result PathBuf { + let staging = root.join("staging"); + fs::create_dir_all(&staging).unwrap(); + for (name, data) in contents { + fs::write(staging.join(name), data).unwrap(); + } + let archive_path = root.join("test.7z"); + sevenz_rust2::compress_to_path(&staging, &archive_path).unwrap(); + archive_path + } + + #[test] + fn extract_7z_success() { + let dir = tempfile::tempdir().unwrap(); + let archive_path = create_test_7z( + dir.path(), + &[("hello.txt", b"Hello, 7z!"), ("data.bin", &[0xAB; 64])], + ); + let output_dir = dir.path().join("output"); + fs::create_dir_all(&output_dir).unwrap(); + + extract_7z(&archive_path, &output_dir).unwrap(); + + assert_eq!( + fs::read_to_string(output_dir.join("hello.txt")).unwrap(), + "Hello, 7z!" + ); + assert_eq!(fs::read(output_dir.join("data.bin")).unwrap(), [0xAB; 64]); + } + + #[test] + fn extract_7z_corrupt_archive_errors() { + let dir = tempfile::tempdir().unwrap(); + let archive_path = dir.path().join("corrupt.7z"); + fs::write(&archive_path, b"7z\xBC\xAF\x27\x1C not actually valid").unwrap(); + let output_dir = dir.path().join("output"); + fs::create_dir_all(&output_dir).unwrap(); + + let err = extract_7z(&archive_path, &output_dir).unwrap_err(); + assert!( + matches!(err, ErgasiaError::ExtractFile { .. }), + "expected ExtractFile for a corrupt archive, got: {err}" + ); + } + + #[test] + fn extract_7z_missing_file_errors() { + let dir = tempfile::tempdir().unwrap(); + let output_dir = dir.path().join("output"); + fs::create_dir_all(&output_dir).unwrap(); + + let err = extract_7z(&dir.path().join("nonexistent.7z"), &output_dir).unwrap_err(); + assert!( + matches!(err, ErgasiaError::ExtractFile { .. }), + "expected ExtractFile for a missing archive, got: {err}" + ); + } + + #[test] + fn declared_size_sums_entries() { + let dir = tempfile::tempdir().unwrap(); + let archive_path = + create_test_7z(dir.path(), &[("a.bin", &[0u8; 100]), ("b.bin", &[0u8; 50])]); + + assert_eq!(declared_uncompressed_size(&archive_path).unwrap(), 150); + } +} diff --git a/crates/ergasia/src/session.rs b/crates/ergasia/src/session.rs index dbd20e01..d4fe4333 100644 --- a/crates/ergasia/src/session.rs +++ b/crates/ergasia/src/session.rs @@ -16,8 +16,8 @@ use tokio_util::sync::CancellationToken; use tracing::instrument; use crate::error::{ - AddTorrentSnafu, ErgasiaError, PauseActionSnafu, SessionInitSnafu, TorrentMapPersistenceSnafu, - TorrentNotFoundSnafu, + AddTorrentSnafu, DeleteActionSnafu, ErgasiaError, PauseActionSnafu, SessionInitSnafu, + TorrentMapPersistenceSnafu, TorrentNotFoundSnafu, }; use crate::seeding::SeedingPolicy; @@ -209,24 +209,30 @@ impl TorrentSession { } pub async fn delete_torrent(&self, download_id: DownloadId) -> Result<(), ErgasiaError> { - let torrent_id = self + // WHY: remove() is an atomic claim — of two concurrent deletes for the + // same id, exactly one proceeds into librqbit; the other sees the entry + // already gone and gets TorrentNotFound instead of a confusing + // wrapped librqbit failure. + let (_, torrent_id) = self .torrent_map - .get(&download_id) - .map(|v| *v) + .remove(&download_id) .ok_or_else(|| TorrentNotFoundSnafu { download_id }.build())?; - self.session + if let Err(e) = self + .session .delete(TorrentIdOrHash::Id(torrent_id), false) .await - .map_err(|e| { - PauseActionSnafu { - download_id, - error: e.to_string(), - } - .build() - })?; + { + // WHY: the torrent still exists in librqbit — restore the mapping + // so the caller can retry instead of orphaning it. + self.torrent_map.insert(download_id, torrent_id); + return Err(DeleteActionSnafu { + download_id, + error: e.to_string(), + } + .build()); + } - self.torrent_map.remove(&download_id); self.persist_torrent_map().await?; Ok(()) } @@ -464,6 +470,81 @@ mod tests { session.session.stop().await; } + #[tokio::test(flavor = "multi_thread")] + async fn unknown_download_id_errors_torrent_not_found() { + let _guard = SESSION_TEST_LOCK.lock().await; + let dir = tempfile::tempdir().unwrap(); + let config = test_config(dir.path(), 24401); + let session = TorrentSession::new(&config).await.unwrap(); + let unknown = DownloadId::new(); + + assert!(matches!( + session.get_torrent(unknown), + Err(ErgasiaError::TorrentNotFound { .. }) + )); + assert!(matches!( + session.get_stats(unknown), + Err(ErgasiaError::TorrentNotFound { .. }) + )); + assert!(matches!( + session.delete_torrent(unknown).await, + Err(ErgasiaError::TorrentNotFound { .. }) + )); + session.session.stop().await; + } + + #[tokio::test(flavor = "multi_thread")] + async fn concurrent_deletes_yield_one_ok_one_not_found() { + let _guard = SESSION_TEST_LOCK.lock().await; + let dir = tempfile::tempdir().unwrap(); + let config = test_config(dir.path(), 24501); + let download_id = DownloadId::new(); + + let session = TorrentSession::new(&config).await.unwrap(); + session + .add_torrent_from_bytes(download_id, minimal_torrent_bytes("race.bin")) + .await + .unwrap(); + + let (a, b) = tokio::join!( + session.delete_torrent(download_id), + session.delete_torrent(download_id) + ); + + let ok_count = [&a, &b].iter().filter(|r| r.is_ok()).count(); + assert_eq!(ok_count, 1, "exactly one delete must win: {a:?} / {b:?}"); + let loser = if a.is_err() { a } else { b }; + assert!( + matches!(loser, Err(ErgasiaError::TorrentNotFound { .. })), + "loser must see TorrentNotFound, got {loser:?}" + ); + session.session.stop().await; + } + + #[tokio::test(flavor = "multi_thread")] + async fn delete_failure_reports_delete_action_and_restores_mapping() { + let _guard = SESSION_TEST_LOCK.lock().await; + let dir = tempfile::tempdir().unwrap(); + let config = test_config(dir.path(), 24601); + let download_id = DownloadId::new(); + + let session = TorrentSession::new(&config).await.unwrap(); + // WHY: a mapping to a torrent id librqbit does not manage forces the + // session.delete failure path deterministically. + session.torrent_map.insert(download_id, 999_999); + + let err = session.delete_torrent(download_id).await.unwrap_err(); + assert!( + matches!(err, ErgasiaError::DeleteAction { .. }), + "expected DeleteAction (not PauseAction), got {err:?}" + ); + assert!( + session.torrent_map.contains_key(&download_id), + "mapping must be restored after a failed delete" + ); + session.session.stop().await; + } + #[tokio::test(flavor = "multi_thread")] async fn corrupt_torrent_map_is_quarantined() { let _guard = SESSION_TEST_LOCK.lock().await; diff --git a/crates/paroche/src/routes/request.rs b/crates/paroche/src/routes/request.rs index f2c7b1ac..2f22f91a 100644 --- a/crates/paroche/src/routes/request.rs +++ b/crates/paroche/src/routes/request.rs @@ -160,19 +160,23 @@ pub async fn list_requests( .as_deref() .map(parse_request_status) .transpose()?; + // WHY: the window and the total both come from SQL — fetching the full + // table to paginate in memory is an unbounded allocation on large + // libraries. + let limit = u32::try_from(per_page).unwrap_or(u32::MAX); + let offset = u32::try_from(offset).unwrap_or(u32::MAX); + let total = state + .requests + .count_requests(auth.user_id, user_id, status) + .await + .map_err(map_request_service_error)?; let requests = state .requests - .list_requests(auth.user_id, user_id, status) + .list_requests(auth.user_id, user_id, status, limit, offset) .await .map_err(map_request_service_error)?; - let total = requests.len() as u64; - let data: Vec = requests - .into_iter() - .skip(offset as usize) - .take(per_page as usize) - .map(Into::into) - .collect(); + let data: Vec = requests.into_iter().map(Into::into).collect(); Ok(ApiResponse::paginated(data, page, per_page, total)) } @@ -371,11 +375,28 @@ mod tests { caller_id: UserId, user_id: Option, status: Option, + limit: u32, + offset: u32, ) -> crate::state::RequestServiceFut<'_, Vec> { let service = Arc::clone(&self.0); Box::pin(async move { service - .list_requests(caller_id, user_id, status) + .list_requests(caller_id, user_id, status, limit, offset) + .await + .map_err(Into::into) + }) + } + + fn count_requests( + &self, + caller_id: UserId, + user_id: Option, + status: Option, + ) -> crate::state::RequestServiceFut<'_, u64> { + let service = Arc::clone(&self.0); + Box::pin(async move { + service + .count_requests(caller_id, user_id, status) .await .map_err(Into::into) }) diff --git a/crates/paroche/src/state.rs b/crates/paroche/src/state.rs index b56f42ba..71f506f5 100644 --- a/crates/paroche/src/state.rs +++ b/crates/paroche/src/state.rs @@ -150,8 +150,17 @@ pub trait DynRequestService: Send + Sync { caller_id: themelion::UserId, user_id: Option, status: Option, + limit: u32, + offset: u32, ) -> RequestServiceFut<'_, Vec>; + fn count_requests( + &self, + caller_id: themelion::UserId, + user_id: Option, + status: Option, + ) -> RequestServiceFut<'_, u64>; + fn cancel_request( &self, request_id: themelion::RequestId, @@ -317,10 +326,21 @@ impl DynRequestService for NullRequestService { _caller_id: themelion::UserId, _user_id: Option, _status: Option, + _limit: u32, + _offset: u32, ) -> RequestServiceFut<'_, Vec> { Box::pin(async { Err(RequestServiceError::NotAvailable) }) } + fn count_requests( + &self, + _caller_id: themelion::UserId, + _user_id: Option, + _status: Option, + ) -> RequestServiceFut<'_, u64> { + Box::pin(async { Err(RequestServiceError::NotAvailable) }) + } + fn cancel_request( &self, _request_id: themelion::RequestId, diff --git a/crates/syndesmos/src/events.rs b/crates/syndesmos/src/events.rs index c60e33f7..0433276d 100644 --- a/crates/syndesmos/src/events.rs +++ b/crates/syndesmos/src/events.rs @@ -31,7 +31,21 @@ pub async fn run_event_handler( } result = rx.recv() => { match result { - Ok(event) => handle_event(&service, event).await, + Ok(event) => { + // WHY: handle_event runs retry loops with backoff sleeps + // — raced against the token so shutdown is prompt even + // mid-retry, instead of waiting out the full backoff. + tokio::select! { + biased; + _ = ct.cancelled() => { + tracing::info!( + "syndesmos event handler cancelled mid-dispatch; shutting down" + ); + break; + } + () = handle_event(&service, event) => {} + } + } Err(RecvError::Lagged(n)) => { tracing::warn!(missed = n, "syndesmos event receiver lagged; events skipped"); } @@ -161,6 +175,100 @@ mod tests { assert_eq!(submitted[0].artist, "Autechre"); } + #[tokio::test] + async fn handler_continues_after_lag() { + use std::sync::Arc; + + use crate::plex::tests::MockPlexApi; + + let mock_plex = Arc::new(MockPlexApi::new()); + let sections_ref = mock_plex.sections_refreshed.clone(); + + // WHY: a small bus overflowed before the handler starts forces the + // first recv to hit RecvError::Lagged — the loop must survive it. + let (tx, rx) = create_event_bus(4); + let ct = CancellationToken::new(); + + for _ in 0..32 { + tx.send(HarmoniaEvent::SearchCompleted { + query_id: themelion::QueryId::new(), + result_count: 0, + }) + .unwrap(); + } + + let mut sections = std::collections::HashMap::new(); + sections.insert(themelion::MediaType::Music, 7u32); + let service = Arc::new( + ScrobbleClientBuilder::new(tx.clone(), crate::test_support::test_pool().await) + .with_mock_plex(mock_plex.clone(), sections) + .build(), + ); + + let ct_clone = ct.clone(); + let handler = tokio::spawn(async move { + run_event_handler(service, rx, ct_clone).await; + }); + + // The handler drains the lag, then must still process a fresh event. + tokio::time::sleep(Duration::from_millis(50)).await; + tx.send(HarmoniaEvent::PlexNotifyRequired { + media_id: MediaId::new(), + }) + .unwrap(); + + tokio::time::sleep(Duration::from_millis(50)).await; + ct.cancel(); + handler.await.unwrap(); + + assert_eq!( + *sections_ref.lock().unwrap(), + vec![7u32], + "the loop must keep dispatching after a Lagged error" + ); + } + + #[tokio::test] + async fn cancellation_interrupts_in_flight_event_dispatch() { + use std::sync::Arc; + + use crate::plex::tests::MockPlexApi; + + // WHY: 30s dwarfs the assertion window — if cancellation waited for + // handle_event to finish, the join below would time out. + let mock_plex = Arc::new(MockPlexApi::with_delay_ms(30_000)); + + let (tx, rx) = create_event_bus(32); + let ct = CancellationToken::new(); + + let mut sections = std::collections::HashMap::new(); + sections.insert(themelion::MediaType::Music, 1u32); + let service = Arc::new( + ScrobbleClientBuilder::new(tx.clone(), crate::test_support::test_pool().await) + .with_mock_plex(mock_plex.clone(), sections) + .build(), + ); + + let ct_clone = ct.clone(); + let handler = tokio::spawn(async move { + run_event_handler(service, rx, ct_clone).await; + }); + + tx.send(HarmoniaEvent::PlexNotifyRequired { + media_id: MediaId::new(), + }) + .unwrap(); + + // Let the handler enter the slow dispatch, then cancel. + tokio::time::sleep(Duration::from_millis(100)).await; + ct.cancel(); + + tokio::time::timeout(Duration::from_secs(2), handler) + .await + .expect("handler must exit promptly despite the in-flight dispatch") + .unwrap(); + } + #[tokio::test] async fn handler_exits_on_cancellation() { let (tx, rx) = create_event_bus(32); diff --git a/crates/syndesmos/src/lastfm/auth.rs b/crates/syndesmos/src/lastfm/auth.rs index 5e0ecf51..afddbc0c 100644 --- a/crates/syndesmos/src/lastfm/auth.rs +++ b/crates/syndesmos/src/lastfm/auth.rs @@ -56,6 +56,18 @@ pub async fn exchange_token( api_key: &str, shared_secret: &str, token: &str, +) -> Result { + exchange_token_at(http, LASTFM_API_URL, api_key, shared_secret, token).await +} + +// WHY: the api_url parameter exists so tests can point the exchange at a +// local mock server; production always passes LASTFM_API_URL. +async fn exchange_token_at( + http: &reqwest::Client, + api_url: &str, + api_key: &str, + shared_secret: &str, + token: &str, ) -> Result { let sig_params = [ ("api_key", api_key), @@ -65,7 +77,7 @@ pub async fn exchange_token( let api_sig = sign_params(&sig_params, shared_secret); let response = http - .post(LASTFM_API_URL) + .post(api_url) .form(&[ ("method", "auth.getSession"), ("api_key", api_key), @@ -78,7 +90,10 @@ pub async fn exchange_token( .context(LastfmApiCallSnafu)?; let body: serde_json::Value = response.json().await.context(LastfmApiCallSnafu)?; + parse_session_key(&body) +} +fn parse_session_key(body: &serde_json::Value) -> Result { body.get("session") .and_then(|s| s.get("key")) .and_then(|k| k.as_str()) @@ -136,6 +151,53 @@ mod tests { assert_eq!(md5_hex(&[0u8; 1024]).len(), 32); } + #[test] + fn parse_session_key_extracts_key_from_valid_response() { + let body = serde_json::json!({"session": {"key": "abc123", "name": "user"}}); + assert_eq!(parse_session_key(&body).unwrap(), "abc123"); + } + + #[test] + fn parse_session_key_errors_on_missing_session() { + let body = serde_json::json!({"error": 4, "message": "Invalid token"}); + let err = parse_session_key(&body).unwrap_err(); + assert!( + matches!(err, SyndesmodError::AuthenticationFailed { ref service, .. } if service == "lastfm"), + "expected AuthenticationFailed for lastfm, got {err:?}" + ); + } + + #[tokio::test] + async fn exchange_token_posts_signed_params_and_returns_key() { + let (url, server) = crate::test_support::spawn_one_shot_http( + 200, + "OK", + r#"{"session":{"key":"session-key-789"}}"#, + ) + .await; + + let key = exchange_token_at(&reqwest::Client::new(), &url, "key123", "sekrit", "tok456") + .await + .unwrap(); + assert_eq!(key, "session-key-789"); + + let request = server.await.unwrap(); + let expected_sig = sign_params( + &[ + ("api_key", "key123"), + ("method", "auth.getSession"), + ("token", "tok456"), + ], + "sekrit", + ); + assert!( + request.contains(&format!("api_sig={expected_sig}")), + "request must carry the signed api_sig, got: {request}" + ); + assert!(request.contains("method=auth.getSession")); + assert!(request.contains("format=json")); + } + #[test] fn sign_params_matches_known_answer_signature() { // WHY: independently computed offline via `md5sum` over the exact diff --git a/crates/syndesmos/src/plex/mod.rs b/crates/syndesmos/src/plex/mod.rs index 3e89a408..94cffaea 100644 --- a/crates/syndesmos/src/plex/mod.rs +++ b/crates/syndesmos/src/plex/mod.rs @@ -76,6 +76,7 @@ pub(crate) mod tests { pub(crate) struct MockPlexApi { pub(crate) sections_refreshed: Arc>>, pub(crate) fail_count: Arc, + pub(crate) delay_ms: Arc, } impl MockPlexApi { @@ -83,9 +84,18 @@ pub(crate) mod tests { Self { sections_refreshed: Arc::new(Mutex::new(Vec::new())), fail_count: Arc::new(std::sync::atomic::AtomicU32::new(0)), + delay_ms: Arc::new(std::sync::atomic::AtomicU64::new(0)), } } + /// Simulates a slow Plex endpoint — each call sleeps first. + pub(crate) fn with_delay_ms(delay_ms: u64) -> Self { + let mock = Self::new(); + mock.delay_ms + .store(delay_ms, std::sync::atomic::Ordering::SeqCst); + mock + } + #[expect( dead_code, reason = "available for future tests requiring pre-configured failures" @@ -109,7 +119,11 @@ pub(crate) mod tests { ) -> BoxFuture<'_, Result<(), SyndesmodError>> { let sections = self.sections_refreshed.clone(); let fail_count = self.fail_count.clone(); + let delay_ms = self.delay_ms.load(std::sync::atomic::Ordering::SeqCst); Box::pin(async move { + if delay_ms > 0 { + tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await; + } let remaining = fail_count.fetch_update( std::sync::atomic::Ordering::SeqCst, std::sync::atomic::Ordering::SeqCst, diff --git a/crates/syntaxis/src/error.rs b/crates/syntaxis/src/error.rs index 2e27ada3..23cd9450 100644 --- a/crates/syntaxis/src/error.rs +++ b/crates/syntaxis/src/error.rs @@ -63,8 +63,11 @@ pub enum SyntaxisError { location: snafu::Location, }, - #[snafu(display("failed to dispatch download to engine"))] + // NOTE: ErgasiaError is carried as a string — the engine trait is generic + // and its error is foreign to this enum's source-chain types. + #[snafu(display("failed to dispatch download to engine: {error}"))] DispatchFailed { + error: String, #[snafu(implicit)] location: snafu::Location, }, diff --git a/crates/syntaxis/src/lib.rs b/crates/syntaxis/src/lib.rs index 235d8ce7..48345703 100644 --- a/crates/syntaxis/src/lib.rs +++ b/crates/syntaxis/src/lib.rs @@ -72,6 +72,7 @@ struct ActiveEntry { want_id: themelion::ids::WantId, release_id: themelion::ids::ReleaseId, download_url: String, + info_hash: Option, retry_count: u32, } @@ -85,6 +86,7 @@ impl ActiveEntry { want_id: item.want_id, release_id: item.release_id, download_url: item.download_url.clone(), + info_hash: item.info_hash.clone(), retry_count: item.retry_count, } } @@ -234,6 +236,7 @@ impl DownloadQueue { })?; self.try_dispatch_next().await; return Err(SyntaxisError::DispatchFailed { + error: e.to_string(), location: snafu::location!(), }); } @@ -560,7 +563,7 @@ impl DownloadQueue { protocol: entry.protocol, priority: 2, tracker_id: entry.tracker_id, - info_hash: None, + info_hash: entry.info_hash, retry_count: retry_count + 1, }; @@ -830,8 +833,15 @@ impl QueueManager for Arc> { self.engine .cancel_download(download_id) .await - .map_err(|_| SyntaxisError::DispatchFailed { - location: snafu::location!(), + .map_err(|e| { + // WHY: this is the decision site — the engine failure is + // logged here and its message carried to the caller, matching + // the Database arm below instead of a bare opaque variant. + error!(error = %e, %download_id, "failed to cancel download on engine"); + SyntaxisError::DispatchFailed { + error: e.to_string(), + location: snafu::location!(), + } })?; repo::mark_failed(&self.pool, entry.queue_id, "cancelled by user") .await @@ -886,7 +896,7 @@ impl QueueManager for Arc> { protocol: e.protocol, priority: 4, tracker_id: e.tracker_id, - info_hash: None, + info_hash: e.info_hash.clone(), retry_count: e.retry_count, }) .collect(); @@ -1455,6 +1465,69 @@ mod tests { assert_eq!(retry_count, 1); } + #[tokio::test] + async fn retry_preserves_info_hash() { + let pool = test_pool().await; + let (engine, mut started_rx) = MockEngine::create(); + let svc = make_service(pool.clone(), Arc::clone(&engine), test_config(2, 3, 0)).await; + + let mut item = make_item(DownloadProtocol::Torrent, 2); + item.info_hash = Some("deadbeef1234567890".to_string()); + svc.enqueue(item).await.unwrap(); + + let (first_id, _) = tokio::time::timeout(RECV_TIMEOUT, started_rx.recv()) + .await + .unwrap() + .unwrap(); + svc.on_download_failed(first_id, "connection timeout".to_string()) + .await; + + settle_until(|| engine.start_calls() == 2).await; + let (second_id, _) = tokio::time::timeout(RECV_TIMEOUT, started_rx.recv()) + .await + .unwrap() + .unwrap(); + + let snapshot = svc.get_queue_state().await.unwrap(); + let active = snapshot + .active_downloads + .iter() + .find(|i| i.retry_count == 1) + .expect("the retried download must be active"); + assert_eq!( + active.info_hash.as_deref(), + Some("deadbeef1234567890"), + "info_hash must survive the retry round-trip" + ); + assert!(active_contains(&svc, second_id).await); + } + + #[tokio::test] + async fn cancel_engine_failure_carries_engine_error() { + let pool = test_pool().await; + let (engine, mut started_rx) = MockEngine::create(); + let svc = make_service(pool.clone(), Arc::clone(&engine), test_config(2, 3, 0)).await; + + let item = make_item(DownloadProtocol::Torrent, 2); + svc.enqueue(item).await.unwrap(); + let (download_id, _) = tokio::time::timeout(RECV_TIMEOUT, started_rx.recv()) + .await + .unwrap() + .unwrap(); + + engine.fail_cancels(); + let err = svc.cancel(download_id).await.unwrap_err(); + match err { + SyntaxisError::DispatchFailed { ref error, .. } => { + assert!( + error.contains("torrent not found"), + "the engine error message must be carried, got: {error}" + ); + } + other => panic!("expected DispatchFailed, got {other:?}"), + } + } + // ── #427: retry budget flows through dispatch and survives recovery ──── #[tokio::test] diff --git a/crates/zetesis/src/client/newznab.rs b/crates/zetesis/src/client/newznab.rs index 1a4d5236..44558cd9 100644 --- a/crates/zetesis/src/client/newznab.rs +++ b/crates/zetesis/src/client/newznab.rs @@ -5,7 +5,7 @@ use std::time::Duration; use bytes::Bytes; use snafu::ResultExt; use tokio_util::sync::CancellationToken; -use tracing::instrument; +use tracing::{instrument, warn}; use crate::cf_bypass::CloudflareProxy; use crate::client::xml::{get_attr_f64, get_attr_u32, parse_caps_xml, parse_feed_xml}; @@ -107,8 +107,18 @@ impl IndexerClient for NewznabClient { .channel .items .into_iter() - .map(|item| { - let download_url = item.link.unwrap_or_default(); + .filter_map(|item| { + // WHY: a result without is unusable downstream (DownloadId + // issuance assumes a fetchable URL) — skip it instead of emitting + // an empty download_url. + let Some(download_url) = item.link else { + warn!( + indexer_id = self.config.id, + title = %item.title, + "skipping newznab item with no " + ); + return None; + }; let category_id = get_attr_u32(&item.attrs, "category"); let download_volume_factor = get_attr_f64(&item.attrs, "downloadvolumefactor").unwrap_or(1.0); @@ -125,7 +135,7 @@ impl IndexerClient for NewznabClient { } } - SearchResult { + Some(SearchResult { title: item.title, guid: item.guid, download_url, @@ -140,7 +150,7 @@ impl IndexerClient for NewznabClient { download_volume_factor, upload_volume_factor, custom_attrs, - } + }) }) .collect(); diff --git a/crates/zetesis/src/client/torznab.rs b/crates/zetesis/src/client/torznab.rs index 7caa6b15..54369d45 100644 --- a/crates/zetesis/src/client/torznab.rs +++ b/crates/zetesis/src/client/torznab.rs @@ -5,7 +5,7 @@ use std::time::Duration; use bytes::Bytes; use snafu::ResultExt; use tokio_util::sync::CancellationToken; -use tracing::instrument; +use tracing::{instrument, warn}; use crate::cf_bypass::CloudflareProxy; use crate::client::xml::{get_attr, get_attr_f64, get_attr_u32, parse_caps_xml, parse_feed_xml}; @@ -107,8 +107,18 @@ impl IndexerClient for TorznabClient { .channel .items .into_iter() - .map(|item| { - let download_url = item.link.unwrap_or_default(); + .filter_map(|item| { + // WHY: a result without is unusable downstream (DownloadId + // issuance assumes a fetchable URL) — skip it instead of emitting + // an empty download_url. + let Some(download_url) = item.link else { + warn!( + indexer_id = self.config.id, + title = %item.title, + "skipping torznab item with no " + ); + return None; + }; let info_hash = get_attr(&item.attrs, "infohash").map(str::to_string); let seeders = get_attr_u32(&item.attrs, "seeders"); let leechers = get_attr_u32(&item.attrs, "leechers"); @@ -134,7 +144,7 @@ impl IndexerClient for TorznabClient { } } - SearchResult { + Some(SearchResult { title: item.title, guid: item.guid, download_url, @@ -149,7 +159,7 @@ impl IndexerClient for TorznabClient { download_volume_factor, upload_volume_factor, custom_attrs, - } + }) }) .collect(); @@ -259,6 +269,38 @@ mod tests { assert_eq!(results[0].protocol, ReleaseProtocol::Torrent); } + #[tokio::test] + async fn item_without_link_is_skipped() { + const MIXED_FEED: &str = r#" + + + Test Indexer + + No.Link.Release + nolink + 1024 + + + Has.Link.Release + haslink + 2048 + https://example.com/download/haslink + + +"#; + let (url, _server) = spawn_one_shot_http(200, "OK", &[], MIXED_FEED).await; + let results = client(url, None, 1 << 20) + .search(&query(), CancellationToken::new()) + .await + .unwrap(); + assert_eq!(results.len(), 1); + assert_eq!(results[0].title, "Has.Link.Release"); + assert_eq!( + results[0].download_url, + "https://example.com/download/haslink" + ); + } + #[tokio::test] async fn fetch_401_maps_to_auth_failed() { let (url, _server) = spawn_one_shot_http(401, "Unauthorized", &[], "").await; diff --git a/crates/zetesis/src/client/xml.rs b/crates/zetesis/src/client/xml.rs index cb279147..10747525 100644 --- a/crates/zetesis/src/client/xml.rs +++ b/crates/zetesis/src/client/xml.rs @@ -186,11 +186,50 @@ impl CapsRoot { } } -fn convert_category(c: CapsCategory) -> IndexerCategory { - IndexerCategory { - id: c.id.and_then(|v| v.parse().ok()).unwrap_or(0), - name: c.name.unwrap_or_default(), - subcategories: c.subcategories.into_iter().map(convert_category).collect(), +// WHY: iterative post-order traversal — category XML is third-party data, and +// recursive conversion would let a hostile deeply-nested caps document +// overflow the stack. +fn convert_category(root: CapsCategory) -> IndexerCategory { + struct Frame { + id: u32, + name: String, + pending: std::vec::IntoIter, + converted: Vec, + } + + fn open(c: CapsCategory) -> Frame { + Frame { + id: c.id.and_then(|v| v.parse().ok()).unwrap_or(0), + name: c.name.unwrap_or_default(), + pending: c.subcategories.into_iter(), + converted: Vec::new(), + } + } + + let mut stack = vec![open(root)]; + loop { + if let Some(child) = stack.last_mut().and_then(|f| f.pending.next()) { + stack.push(open(child)); + continue; + } + let Some(finished) = stack.pop() else { + // INVARIANT: unreachable — the root frame always exits via the + // `None => return` arm below; kept total for lint. + return IndexerCategory { + id: 0, + name: String::new(), + subcategories: Vec::new(), + }; + }; + let node = IndexerCategory { + id: finished.id, + name: finished.name, + subcategories: finished.converted, + }; + match stack.last_mut() { + Some(parent) => parent.converted.push(node), + None => return node, + } } } @@ -351,6 +390,57 @@ mod tests { assert!(caps.categories.is_empty()); } + #[test] + fn convert_category_survives_hostile_nesting_depth() { + const DEPTH: usize = 100_000; + let mut node = CapsCategory { + id: Some("1".to_string()), + name: Some("leaf".to_string()), + subcategories: Vec::new(), + }; + for i in 0..DEPTH { + node = CapsCategory { + id: Some(format!("{i}")), + name: Some(format!("level-{i}")), + subcategories: vec![node], + }; + } + + let converted = convert_category(node); + + // NOTE: the assertion walk (and teardown) is iterative too — a + // recursive walk or plain drop would re-introduce the overflow the + // conversion just avoided. + let mut count = 0usize; + let mut work = vec![converted]; + while let Some(mut n) = work.pop() { + count += 1; + work.append(&mut n.subcategories); + } + assert_eq!(count, DEPTH + 1); + } + + #[test] + fn convert_category_round_trips_shallow_tree() { + let node = CapsCategory { + id: Some("2000".to_string()), + name: Some("Movies".to_string()), + subcategories: vec![CapsCategory { + id: Some("2010".to_string()), + name: Some("Movies/Foreign".to_string()), + subcategories: Vec::new(), + }], + }; + + let converted = convert_category(node); + assert_eq!(converted.id, 2000); + assert_eq!(converted.name, "Movies"); + assert_eq!(converted.subcategories.len(), 1); + assert_eq!(converted.subcategories[0].id, 2010); + assert_eq!(converted.subcategories[0].name, "Movies/Foreign"); + assert!(converted.subcategories[0].subcategories.is_empty()); + } + #[test] fn attr_helpers() { let attrs = vec![ diff --git a/crates/zetesis/src/rate_limit.rs b/crates/zetesis/src/rate_limit.rs index 21f449f3..b4baac1e 100644 --- a/crates/zetesis/src/rate_limit.rs +++ b/crates/zetesis/src/rate_limit.rs @@ -4,6 +4,7 @@ use std::time::Duration; use dashmap::DashMap; use tokio::sync::Mutex; use tokio::time::Instant; +use tokio_util::sync::CancellationToken; pub struct RateLimiter { buckets: Arc>>, @@ -73,7 +74,12 @@ impl RateLimiter { } } - pub async fn acquire(&self, indexer_id: i64) { + /// Waits for a token, racing the back-off sleep against cancellation. + /// + /// Returns `true` when a token was acquired, `false` when the token was + /// cancelled first — callers must skip the guarded work on `false`. + #[must_use = "a false return means cancellation — the guarded work must be skipped"] + pub async fn acquire(&self, indexer_id: i64, ct: &CancellationToken) -> bool { loop { let wait = { let entry = self @@ -85,8 +91,15 @@ impl RateLimiter { }; match wait { - None => return, - Some(duration) => tokio::time::sleep(duration).await, + None => return true, + Some(duration) => { + // WHY: a cancelled search must not park fan-out tasks for the + // full back-off window — race the sleep against the token. + tokio::select! { + () = tokio::time::sleep(duration) => {} + () = ct.cancelled() => return false, + } + } } } } @@ -109,7 +122,7 @@ mod tests { async fn acquire_within_limit() { let limiter = RateLimiter::new(5, Duration::from_secs(10)); for _ in 0..5 { - limiter.acquire(1).await; + assert!(limiter.acquire(1, &CancellationToken::new()).await); } } @@ -118,9 +131,9 @@ mod tests { let limiter = RateLimiter::new(2, Duration::from_millis(200)); let start = Instant::now(); - limiter.acquire(1).await; - limiter.acquire(1).await; - limiter.acquire(1).await; + assert!(limiter.acquire(1, &CancellationToken::new()).await); + assert!(limiter.acquire(1, &CancellationToken::new()).await); + assert!(limiter.acquire(1, &CancellationToken::new()).await); let elapsed = start.elapsed(); assert!( @@ -132,8 +145,26 @@ mod tests { #[tokio::test] async fn separate_indexers_independent() { let limiter = RateLimiter::new(1, Duration::from_secs(10)); - limiter.acquire(1).await; - limiter.acquire(2).await; + assert!(limiter.acquire(1, &CancellationToken::new()).await); + assert!(limiter.acquire(2, &CancellationToken::new()).await); + } + + #[tokio::test] + async fn acquire_unblocks_on_cancellation_before_refill() { + let limiter = RateLimiter::new(1, Duration::from_secs(600)); + assert!(limiter.acquire(1, &CancellationToken::new()).await); + + let ct = CancellationToken::new(); + ct.cancel(); + let start = Instant::now(); + let acquired = limiter.acquire(1, &ct).await; + let elapsed = start.elapsed(); + + assert!(!acquired, "expected cancellation, not acquisition"); + assert!( + elapsed < Duration::from_secs(1), + "expected prompt unblock on cancel, got {elapsed:?}" + ); } #[tokio::test] @@ -142,7 +173,7 @@ mod tests { limiter.set_retry_after(1, Duration::from_millis(100)).await; let start = Instant::now(); - limiter.acquire(1).await; + assert!(limiter.acquire(1, &CancellationToken::new()).await); let elapsed = start.elapsed(); assert!( diff --git a/crates/zetesis/src/search.rs b/crates/zetesis/src/search.rs index 89353605..9f27f2f3 100644 --- a/crates/zetesis/src/search.rs +++ b/crates/zetesis/src/search.rs @@ -104,7 +104,16 @@ impl SearchIndexerService { let ct = ct.clone(); let q = query.clone(); async move { - rate_limiter.acquire(indexer.id).await; + if !rate_limiter.acquire(indexer.id, &ct).await { + // WHY: cancellation during rate-limit back-off — the + // search is abandoned, skip the fetch entirely. + info!( + indexer_id = indexer.id, + indexer_name = %indexer.name, + "search cancelled while awaiting rate limit" + ); + return Vec::new(); + } let client = make_client(&indexer, h, cf, timeout, max_body_bytes); match client.search_boxed(&q, ct).await { Ok(results) => results, @@ -309,8 +318,19 @@ fn filter_by_capability(indexers: &[IndexerRow], query: &SearchQuery) -> Vec(caps_json) else { - return false; + let caps = match serde_json::from_str::(caps_json) { + Ok(caps) => caps, + Err(e) => { + // WHY: this is the decision site excluding the indexer from a + // typed search — stale/corrupt caps must be visible, not a + // silent disappearance from results. + warn!( + indexer_id = indexer.id, + error = %e, + "invalid caps_json, excluding indexer from typed search" + ); + return false; + } }; crate::types::supports_function(&caps, function_type) @@ -325,16 +345,24 @@ fn deduplicate(results: Vec) -> Vec { let mut deduped: Vec = Vec::with_capacity(results.len()); for result in results { - if let Some(ref hash) = result.info_hash { - let hash_lower = hash.to_lowercase(); - if seen_hashes.contains_key(&hash_lower) { - continue; - } - seen_hashes.insert(hash_lower, deduped.len()); - } else if let Some(ref guid) = result.guid { - if seen_guids.contains_key(guid) { - continue; - } + // WHY: hash and guid are checked independently (not else-if) — a result + // carrying both must register both keys, or a later copy sharing only + // the guid slips past dedup. + let hash_lower = result.info_hash.as_ref().map(|h| h.to_lowercase()); + let is_dupe = hash_lower + .as_ref() + .is_some_and(|h| seen_hashes.contains_key(h)) + || result + .guid + .as_ref() + .is_some_and(|g| seen_guids.contains_key(g)); + if is_dupe { + continue; + } + if let Some(hash) = hash_lower { + seen_hashes.insert(hash, deduped.len()); + } + if let Some(ref guid) = result.guid { seen_guids.insert(guid.clone(), deduped.len()); } @@ -434,6 +462,32 @@ mod tests { assert_eq!(deduped[1].title, "NZB.B"); } + #[test] + fn dedup_registers_guid_of_hash_bearing_result() { + // WHY: a result carrying both keys must register its guid too — a later + // copy sharing only the guid must not slip past dedup. + let results = vec![ + make_result("Release.A", Some("abc123"), Some("guid-1"), 1), + make_result("Release.A.guid-dupe", None, Some("guid-1"), 2), + ]; + + let deduped = deduplicate(results); + assert_eq!(deduped.len(), 1); + assert_eq!(deduped[0].title, "Release.A"); + } + + #[test] + fn dedup_same_hash_different_guid_still_dedupes() { + let results = vec![ + make_result("Release.A", Some("abc123"), Some("guid-1"), 1), + make_result("Release.A.hash-dupe", Some("abc123"), Some("guid-2"), 2), + ]; + + let deduped = deduplicate(results); + assert_eq!(deduped.len(), 1); + assert_eq!(deduped[0].title, "Release.A"); + } + #[test] fn dedup_case_insensitive_hash() { let results = vec![ @@ -643,6 +697,67 @@ mod tests { assert_eq!(row.status, "failed"); } + #[tokio::test] + async fn handle_search_error_parse_response_marks_degraded() { + let (service, pool) = make_service().await; + let indexer = seed_indexer(&pool, "https://example.com/api").await; + + let error = SearchIndexerError::ParseResponse { + url: "https://example.com/api".to_string(), + error: "bad xml".to_string(), + location: snafu::Location::new(file!(), line!(), column!()), + }; + service.handle_search_error(&indexer, &error).await; + + let row = repo::get_indexer(&pool, indexer.id).await.unwrap().unwrap(); + assert_eq!(row.status, "degraded"); + } + + #[tokio::test] + async fn handle_search_error_http_request_active_marks_degraded() { + let (service, pool) = make_service().await; + let indexer = seed_indexer(&pool, "https://example.com/api").await; + assert_eq!(indexer.status, "active"); + + let error = SearchIndexerError::HttpRequest { + url: "https://example.com/api".to_string(), + source: reqwest::Client::new() + .get("http://127.0.0.1:9/") + .send() + .await + .unwrap_err(), + location: snafu::Location::new(file!(), line!(), column!()), + }; + service.handle_search_error(&indexer, &error).await; + + let row = repo::get_indexer(&pool, indexer.id).await.unwrap().unwrap(); + assert_eq!(row.status, "degraded"); + } + + #[tokio::test] + async fn handle_search_error_http_request_degraded_escalates_to_failed() { + let (service, pool) = make_service().await; + let mut indexer = seed_indexer(&pool, "https://example.com/api").await; + repo::update_indexer_status(&pool, indexer.id, "degraded") + .await + .unwrap(); + indexer.status = "degraded".to_string(); + + let error = SearchIndexerError::HttpRequest { + url: "https://example.com/api".to_string(), + source: reqwest::Client::new() + .get("http://127.0.0.1:9/") + .send() + .await + .unwrap_err(), + location: snafu::Location::new(file!(), line!(), column!()), + }; + service.handle_search_error(&indexer, &error).await; + + let row = repo::get_indexer(&pool, indexer.id).await.unwrap().unwrap(); + assert_eq!(row.status, "failed"); + } + // WHY: the clock is paused only around the limiter interaction — sqlx's // sqlite worker runs on a real thread, and a paused clock during pool // setup auto-advances straight into PoolTimedOut. @@ -657,7 +772,12 @@ mod tests { .await; let before = tokio::time::Instant::now(); - service.rate_limiter.acquire(indexer.id).await; + assert!( + service + .rate_limiter + .acquire(indexer.id, &CancellationToken::new()) + .await + ); let elapsed = before.elapsed(); tokio::time::resume(); assert!( @@ -681,7 +801,12 @@ mod tests { .await; let before = tokio::time::Instant::now(); - service.rate_limiter.acquire(indexer.id).await; + assert!( + service + .rate_limiter + .acquire(indexer.id, &CancellationToken::new()) + .await + ); let elapsed = before.elapsed(); tokio::time::resume(); assert!( @@ -701,7 +826,12 @@ mod tests { .await; let before = tokio::time::Instant::now(); - service.rate_limiter.acquire(indexer.id).await; + assert!( + service + .rate_limiter + .acquire(indexer.id, &CancellationToken::new()) + .await + ); let elapsed = before.elapsed(); tokio::time::resume(); assert!(