diff --git a/crates/persistence/src/backends/mongodb/search_impl.rs b/crates/persistence/src/backends/mongodb/search_impl.rs index fe8e91557..afee20371 100644 --- a/crates/persistence/src/backends/mongodb/search_impl.rs +++ b/crates/persistence/src/backends/mongodb/search_impl.rs @@ -17,7 +17,7 @@ use crate::core::{ ConditionalUpdateResult, IncludeProvider, PatchFormat, ResourceStorage, RevincludeProvider, SearchProvider, SearchResult, }; -use crate::error::{BackendError, SearchError, StorageError, StorageResult}; +use crate::error::{BackendError, QueryErrorExt, SearchError, StorageError, StorageResult}; use crate::tenant::TenantContext; use crate::types::{ CompartmentMembership, CursorDirection, CursorValue, IncludeDirective, IncludeType, Page, @@ -74,11 +74,11 @@ async fn collect_documents(mut cursor: Cursor) -> StorageResult>(); @@ -738,7 +738,7 @@ impl MongoBackend { let ids = search_index .distinct("resource_id", filter) .await - .map_err(|e| internal_error(format!("Failed to query search_index: {}", e)))? + .or_query_error("Failed to query search_index")? .into_iter() .filter_map(|value| value.as_str().map(ToString::to_string)) .collect::>(); @@ -1500,7 +1500,7 @@ impl MongoBackend { "is_deleted": false, }) .await - .map_err(|e| internal_error(format!("Failed to fetch included resource: {}", e)))?; + .or_query_error("Failed to fetch included resource")?; match doc { Some(doc) => Ok(Some(self.document_to_stored_resource( @@ -1613,12 +1613,7 @@ impl RevincludeProvider for MongoBackend { let matching_ids: Vec = search_index .distinct("resource_id", index_filter) .await - .map_err(|e| { - internal_error(format!( - "Failed to query search_index for revinclude: {}", - e - )) - })? + .or_query_error("Failed to query search_index for revinclude")? .into_iter() .filter_map(|value| value.as_str().map(ToString::to_string)) .collect(); @@ -1635,11 +1630,13 @@ impl RevincludeProvider for MongoBackend { "id": { "$in": Bson::Array(id_bson) }, }; - let docs = - collect_documents(resources_collection.find(resource_filter).await.map_err( - |e| internal_error(format!("Failed to fetch revinclude resources: {}", e)), - )?) - .await?; + let docs = collect_documents( + resources_collection + .find(resource_filter) + .await + .or_query_error("Failed to fetch revinclude resources")?, + ) + .await?; for doc in docs { let stored = diff --git a/crates/persistence/src/backends/mongodb/storage.rs b/crates/persistence/src/backends/mongodb/storage.rs index 6bf591ddd..f159e42ac 100644 --- a/crates/persistence/src/backends/mongodb/storage.rs +++ b/crates/persistence/src/backends/mongodb/storage.rs @@ -20,7 +20,8 @@ use crate::core::{ bundle_if_match_gate, if_match_field_satisfied, normalize_etag, }; use crate::error::{ - BackendError, ConcurrencyError, ResourceError, StorageError, StorageResult, TransactionError, + BackendError, ConcurrencyError, QueryErrorExt, ResourceError, StorageError, StorageResult, + TransactionError, }; use crate::search::converters::IndexValue; use crate::search::extractor::ExtractedValue; @@ -1212,7 +1213,7 @@ impl ResourceStorage for MongoBackend { resources .count_documents(filter) .await - .map_err(|e| internal_error(format!("Failed to count resources: {}", e))) + .or_query_error("Failed to count resources") } async fn count_by_day( @@ -1252,17 +1253,17 @@ impl ResourceStorage for MongoBackend { let mut cursor = resources .aggregate(pipeline) .await - .map_err(|e| internal_error(format!("Failed to aggregate count_by_day: {}", e)))?; + .or_query_error("Failed to aggregate count_by_day")?; let mut out = Vec::new(); while cursor .advance() .await - .map_err(|e| internal_error(format!("count_by_day cursor advance: {}", e)))? + .or_query_error("count_by_day cursor advance")? { let doc = cursor .deserialize_current() - .map_err(|e| internal_error(format!("count_by_day cursor deserialize: {}", e)))?; + .or_query_error("count_by_day cursor deserialize")?; let day_str = doc.get_str("_id").unwrap_or_default(); // `$sum: 1` yields an int32 unless it overflows into int64. let n = doc @@ -1330,19 +1331,20 @@ impl ResourceStorage for MongoBackend { doc! { "$sort": { "_id": 1 } }, ]; - let mut cursor = history.aggregate(pipeline).await.map_err(|e| { - internal_error(format!("Failed to aggregate count_deltas_by_bucket: {}", e)) - })?; + let mut cursor = history + .aggregate(pipeline) + .await + .or_query_error("Failed to aggregate count_deltas_by_bucket")?; let mut out = Vec::new(); while cursor .advance() .await - .map_err(|e| internal_error(format!("count_deltas cursor advance: {}", e)))? + .or_query_error("count_deltas cursor advance")? { let doc = cursor .deserialize_current() - .map_err(|e| internal_error(format!("count_deltas cursor deserialize: {}", e)))?; + .or_query_error("count_deltas cursor deserialize")?; let bucket_ms_start = doc.get_i64("_id").unwrap_or_default(); // `$sum` yields an int32 for small totals and an int64 once it overflows, // so accept either width rather than assuming one. @@ -1387,19 +1389,20 @@ impl ResourceStorage for MongoBackend { }}, ]; - let mut cursor = history.aggregate(pipeline).await.map_err(|e| { - internal_error(format!("Failed to aggregate activity histogram: {}", e)) - })?; + let mut cursor = history + .aggregate(pipeline) + .await + .or_query_error("Failed to aggregate activity histogram")?; let mut out = Vec::new(); while cursor .advance() .await - .map_err(|e| internal_error(format!("activity cursor advance: {}", e)))? + .or_query_error("activity cursor advance")? { let doc = cursor .deserialize_current() - .map_err(|e| internal_error(format!("activity cursor deserialize: {}", e)))?; + .or_query_error("activity cursor deserialize")?; let id = match doc.get_document("_id") { Ok(id) => id, Err(_) => continue, @@ -1548,7 +1551,7 @@ impl ResourceStorage for MongoBackend { let removed = resources .count_documents(doc! { "tenant_id": id }) .await - .map_err(|e| internal_error(format!("purge count: {}", e)))?; + .or_query_error("purge count")?; for collection in [ MongoBackend::SEARCH_INDEX_COLLECTION, MongoBackend::RESOURCE_HISTORY_COLLECTION, @@ -1557,7 +1560,7 @@ impl ResourceStorage for MongoBackend { db.collection::(collection) .delete_many(doc! { "tenant_id": id }) .await - .map_err(|e| internal_error(format!("purge delete ({}): {}", collection, e)))?; + .or_query_error(&format!("purge delete ({collection})"))?; } Ok(removed) } @@ -1590,16 +1593,16 @@ async fn grouped_string_counts( let mut cursor = collection .aggregate(pipeline) .await - .map_err(|e| internal_error(format!("Failed to aggregate grouped counts: {}", e)))?; + .or_query_error("Failed to aggregate grouped counts")?; let mut out = Vec::new(); while cursor .advance() .await - .map_err(|e| internal_error(format!("grouped counts cursor advance: {}", e)))? + .or_query_error("grouped counts cursor advance")? { let doc = cursor .deserialize_current() - .map_err(|e| internal_error(format!("grouped counts cursor deserialize: {}", e)))?; + .or_query_error("grouped counts cursor deserialize")?; let key = doc.get_str("_id").unwrap_or_default().to_string(); if key.is_empty() { continue; @@ -2208,7 +2211,7 @@ impl VersionedStorage for MongoBackend { "id": id, }) .await - .map_err(|e| internal_error(format!("Failed to query version history: {}", e)))?; + .or_query_error("Failed to query version history")?; let docs = collect_documents(cursor).await?; let mut versions = docs @@ -2249,7 +2252,7 @@ impl InstanceHistoryProvider for MongoBackend { let cursor = history .find(filter) .await - .map_err(|e| internal_error(format!("Failed to query instance history: {}", e)))?; + .or_query_error("Failed to query instance history")?; let docs = collect_documents(cursor).await?; let mut rows = docs @@ -2311,7 +2314,7 @@ impl InstanceHistoryProvider for MongoBackend { "id": id, }) .await - .map_err(|e| internal_error(format!("Failed to count instance history: {}", e))) + .or_query_error("Failed to count instance history") } } @@ -2336,7 +2339,7 @@ impl TypeHistoryProvider for MongoBackend { let cursor = history .find(filter) .await - .map_err(|e| internal_error(format!("Failed to query type history: {}", e)))?; + .or_query_error("Failed to query type history")?; let docs = collect_documents(cursor).await?; let mut rows = docs @@ -2403,7 +2406,7 @@ impl TypeHistoryProvider for MongoBackend { "resource_type": resource_type, }) .await - .map_err(|e| internal_error(format!("Failed to count type history: {}", e))) + .or_query_error("Failed to count type history") } } @@ -2426,7 +2429,7 @@ impl SystemHistoryProvider for MongoBackend { let cursor = history .find(filter) .await - .map_err(|e| internal_error(format!("Failed to query system history: {}", e)))?; + .or_query_error("Failed to query system history")?; let docs = collect_documents(cursor).await?; let mut rows = docs @@ -2494,7 +2497,7 @@ impl SystemHistoryProvider for MongoBackend { "tenant_id": tenant_id, }) .await - .map_err(|e| internal_error(format!("Failed to count system history: {}", e))) + .or_query_error("Failed to count system history") } } @@ -3606,7 +3609,7 @@ impl PurgableStorage for MongoBackend { let in_history = history .count_documents(key.clone()) .await - .map_err(|e| internal_error(format!("Failed to check resource history: {e}")))?; + .or_query_error("Failed to check resource history")?; if in_resources == 0 && in_history == 0 { return Err(StorageError::Resource(ResourceError::NotFound { resource_type: resource_type.to_string(), @@ -3617,11 +3620,11 @@ impl PurgableStorage for MongoBackend { resources .delete_many(key.clone()) .await - .map_err(|e| internal_error(format!("Failed to purge resource: {e}")))?; + .or_query_error("Failed to purge resource")?; history .delete_many(key) .await - .map_err(|e| internal_error(format!("Failed to purge resource history: {e}")))?; + .or_query_error("Failed to purge resource history")?; // The search_index collection keys the resource as `resource_id`. search_index @@ -3631,7 +3634,7 @@ impl PurgableStorage for MongoBackend { "resource_id": id, }) .await - .map_err(|e| internal_error(format!("Failed to purge search index: {e}")))?; + .or_query_error("Failed to purge search index")?; Ok(()) } @@ -3652,20 +3655,20 @@ impl PurgableStorage for MongoBackend { let count = resources .count_documents(key.clone()) .await - .map_err(|e| internal_error(format!("Failed to count resources: {e}")))?; + .or_query_error("Failed to count resources")?; resources .delete_many(key.clone()) .await - .map_err(|e| internal_error(format!("Failed to purge resources: {e}")))?; + .or_query_error("Failed to purge resources")?; history .delete_many(key) .await - .map_err(|e| internal_error(format!("Failed to purge resource history: {e}")))?; + .or_query_error("Failed to purge resource history")?; search_index .delete_many(doc! { "tenant_id": tenant_id, "resource_type": resource_type }) .await - .map_err(|e| internal_error(format!("Failed to purge search index: {e}")))?; + .or_query_error("Failed to purge search index")?; Ok(count) } @@ -3861,7 +3864,7 @@ impl ReindexTarget for MongoBackend { .collection::(MongoBackend::SEARCH_INDEX_COLLECTION) .delete_many(doc! { "tenant_id": tenant.tenant_id().as_str() }) .await - .map_err(|e| internal_error(format!("Failed to clear search index: {e}")))?; + .or_query_error("Failed to clear search index")?; Ok(result.deleted_count) } diff --git a/crates/persistence/src/backends/postgres/search_impl.rs b/crates/persistence/src/backends/postgres/search_impl.rs index 2bed222f3..b100ae3eb 100644 --- a/crates/persistence/src/backends/postgres/search_impl.rs +++ b/crates/persistence/src/backends/postgres/search_impl.rs @@ -17,7 +17,7 @@ use crate::core::{ ChainedSearchProvider, IncludeProvider, MultiTypeSearchProvider, ResourceStorage, RevincludeProvider, SearchProvider, SearchResult, TextSearchProvider, }; -use crate::error::{BackendError, StorageError, StorageResult}; +use crate::error::{BackendError, QueryErrorExt, StorageError, StorageResult}; use crate::tenant::TenantContext; use crate::types::{ CursorDirection, CursorValue, IncludeDirective, Page, PageCursor, PageInfo, Pagination, @@ -200,7 +200,7 @@ impl SearchProvider for PostgresBackend { let rows = client .query(&sql, ¶m_refs) .await - .map_err(|e| internal_error(format!("Failed to execute search: {}", e)))?; + .or_query_error("Failed to execute search")?; // Parse rows, capturing the sort key for cursor construction. let mut parsed: Vec<(StoredResource, Option)> = Vec::new(); @@ -339,7 +339,7 @@ impl SearchProvider for PostgresBackend { let row = client .query_one(&sql, ¶m_refs) .await - .map_err(|e| internal_error(format!("Failed to count resources: {}", e)))?; + .or_query_error("Failed to count resources")?; let count: i64 = row.get(0); Ok(count as u64) @@ -402,7 +402,7 @@ impl MultiTypeSearchProvider for PostgresBackend { let rows = client .query(&sql, &[&tenant_id]) .await - .map_err(|e| internal_error(format!("Failed to execute multi-type search: {}", e)))?; + .or_query_error("Failed to execute multi-type search")?; let mut resources = Vec::new(); for row in &rows { @@ -567,9 +567,10 @@ impl RevincludeProvider for PostgresBackend { .map(|p| p.as_ref() as &(dyn tokio_postgres::types::ToSql + Sync)) .collect(); - let rows = client.query(&sql, ¶m_refs).await.map_err(|e| { - internal_error(format!("Failed to execute revinclude query: {}", e)) - })?; + let rows = client + .query(&sql, ¶m_refs) + .await + .or_query_error("Failed to execute revinclude query")?; for row in &rows { let id: String = row.get(0); @@ -662,7 +663,7 @@ impl ChainedSearchProvider for PostgresBackend { let rows = client .query(&sql, ¶m_refs) .await - .map_err(|e| internal_error(format!("Failed to execute chain query: {}", e)))?; + .or_query_error("Failed to execute chain query")?; Ok(rows.iter().map(|r| r.get(0)).collect()) } @@ -710,7 +711,7 @@ impl ChainedSearchProvider for PostgresBackend { let rows = client .query(&sql, ¶m_refs) .await - .map_err(|e| internal_error(format!("Failed to execute reverse chain query: {}", e)))?; + .or_query_error("Failed to execute reverse chain query")?; Ok(rows.iter().map(|r| r.get(0)).collect()) } @@ -746,7 +747,7 @@ impl TextSearchProvider for PostgresBackend { let rows = client .query(&sql, &[&tenant_id, &resource_type, &text]) .await - .map_err(|e| internal_error(format!("Failed to execute text search: {}", e)))?; + .or_query_error("Failed to execute text search")?; let mut resources = Vec::new(); for row in &rows { @@ -821,7 +822,7 @@ impl TextSearchProvider for PostgresBackend { let rows = client .query(&sql, &[&tenant_id, &resource_type, &content]) .await - .map_err(|e| internal_error(format!("Failed to execute content search: {}", e)))?; + .or_query_error("Failed to execute content search")?; let mut resources = Vec::new(); for row in &rows { @@ -948,9 +949,7 @@ impl PostgresBackend { let rows = client .query(&fragment.sql, ¶m_refs) .await - .map_err(|e| { - internal_error(format!("Failed to execute contained query: {e}")) - })?; + .or_query_error("Failed to execute contained query")?; rows.iter() .map(|row| { ( @@ -1168,7 +1167,7 @@ impl PostgresBackend { &[&tenant_id, &resource_type, &id], ) .await - .map_err(|e| internal_error(format!("Failed to fetch resource: {}", e)))?; + .or_query_error("Failed to fetch resource")?; if rows.is_empty() { return Ok(None); diff --git a/crates/persistence/src/backends/postgres/storage.rs b/crates/persistence/src/backends/postgres/storage.rs index fd6ef1cce..3960e9ede 100644 --- a/crates/persistence/src/backends/postgres/storage.rs +++ b/crates/persistence/src/backends/postgres/storage.rs @@ -18,7 +18,9 @@ use crate::core::{ if_match_field_satisfied, normalize_etag, }; use crate::error::TransactionError; -use crate::error::{BackendError, ConcurrencyError, ResourceError, StorageError, StorageResult}; +use crate::error::{ + BackendError, ConcurrencyError, QueryErrorExt, ResourceError, StorageError, StorageResult, +}; use crate::search::reindex::{ReindexSource, ReindexTarget, ResourcePage}; use crate::tenant::{Operation, TenantContext}; use crate::types::Pagination; @@ -491,7 +493,7 @@ impl ResourceStorage for PostgresBackend { &[&tenant_id, &rt], ) .await - .map_err(|e| internal_error(format!("Failed to count resources: {}", e)))?; + .or_query_error("Failed to count resources")?; row.get(0) } else { let row = client @@ -500,7 +502,7 @@ impl ResourceStorage for PostgresBackend { &[&tenant_id], ) .await - .map_err(|e| internal_error(format!("Failed to count resources: {}", e)))?; + .or_query_error("Failed to count resources")?; row.get(0) }; @@ -529,7 +531,7 @@ impl ResourceStorage for PostgresBackend { &[&tenant_id, &resource_type, &since], ) .await - .map_err(|e| internal_error(format!("Failed to count resources by day: {}", e)))?; + .or_query_error("Failed to count resources by day")?; let mut out = Vec::with_capacity(rows.len()); for row in rows { @@ -588,7 +590,7 @@ impl ResourceStorage for PostgresBackend { &[&tenant_id, &resource_type, &since_bound, &bucket_seconds], ) .await - .map_err(|e| internal_error(format!("Failed to count resource deltas: {}", e)))?; + .or_query_error("Failed to count resource deltas")?; let mut out = Vec::with_capacity(rows.len()); for row in rows { @@ -625,7 +627,7 @@ impl ResourceStorage for PostgresBackend { &[&tenant_id, &since], ) .await - .map_err(|e| internal_error(format!("Failed to compute activity histogram: {}", e)))?; + .or_query_error("Failed to compute activity histogram")?; let mut out = Vec::with_capacity(rows.len()); for row in rows { @@ -652,7 +654,7 @@ impl ResourceStorage for PostgresBackend { &[&tenant_id], ) .await - .map_err(|e| internal_error(format!("Failed to count all types: {}", e)))?; + .or_query_error("Failed to count all types")?; let mut out = Vec::with_capacity(rows.len()); for row in rows { let rt: String = row.get(0); @@ -702,7 +704,7 @@ impl ResourceStorage for PostgresBackend { let rows = client .query(&sql, ¶m_refs) .await - .map_err(|e| internal_error(format!("Failed to count by types: {}", e)))?; + .or_query_error("Failed to count by types")?; let mut out = Vec::with_capacity(rows.len()); for row in rows { let rt: String = row.get(0); @@ -722,7 +724,7 @@ impl ResourceStorage for PostgresBackend { &[], ) .await - .map_err(|e| internal_error(format!("Failed to count by tenant: {}", e)))?; + .or_query_error("Failed to count by tenant")?; let mut out = Vec::with_capacity(rows.len()); for row in rows { let tid: String = row.get(0); @@ -807,10 +809,7 @@ impl ResourceStorage for PostgresBackend { async fn purge_tenant_data(&self, id: &str) -> StorageResult { let mut client = self.get_client().await?; - let tx = client - .transaction() - .await - .map_err(|e| internal_error(format!("purge begin: {e}")))?; + let tx = client.transaction().await.or_query_error("purge begin")?; // Count current-version rows first (soft-deleted included) so we can // report what was removed. let removed: i64 = tx @@ -819,7 +818,7 @@ impl ResourceStorage for PostgresBackend { &[&id], ) .await - .map_err(|e| internal_error(format!("purge count: {e}")))? + .or_query_error("purge count")? .get(0); // search_index and resource_fts cascade from resources, but delete them // explicitly too, mirroring the purge/purge_all deletion order. @@ -831,11 +830,9 @@ impl ResourceStorage for PostgresBackend { ] { tx.execute(sql, &[&id]) .await - .map_err(|e| internal_error(format!("purge delete: {e}")))?; + .or_query_error("purge delete")?; } - tx.commit() - .await - .map_err(|e| internal_error(format!("purge commit: {e}")))?; + tx.commit().await.or_query_error("purge commit")?; Ok(removed.max(0) as u64) } } @@ -1449,7 +1446,7 @@ impl InstanceHistoryProvider for PostgresBackend { let rows = client .query(&sql, ¶m_refs) .await - .map_err(|e| internal_error(format!("Failed to query history: {}", e)))?; + .or_query_error("Failed to query history")?; let mut entries = Vec::new(); let mut last_version: Option = None; @@ -1531,7 +1528,7 @@ impl InstanceHistoryProvider for PostgresBackend { &[&tenant_id, &resource_type, &id], ) .await - .map_err(|e| internal_error(format!("Failed to count history: {}", e)))?; + .or_query_error("Failed to count history")?; let count: i64 = row.get(0); Ok(count as u64) @@ -1744,7 +1741,7 @@ impl TypeHistoryProvider for PostgresBackend { let rows = client .query(&sql, ¶m_refs) .await - .map_err(|e| internal_error(format!("Failed to query type history: {}", e)))?; + .or_query_error("Failed to query type history")?; let mut entries = Vec::new(); let mut last_entry: Option<(String, String)> = None; // (last_updated, id) @@ -1827,7 +1824,7 @@ impl TypeHistoryProvider for PostgresBackend { &[&tenant_id, &resource_type], ) .await - .map_err(|e| internal_error(format!("Failed to count type history: {}", e)))?; + .or_query_error("Failed to count type history")?; let count: i64 = row.get(0); Ok(count as u64) @@ -1915,7 +1912,7 @@ impl SystemHistoryProvider for PostgresBackend { let rows = client .query(&sql, ¶m_refs) .await - .map_err(|e| internal_error(format!("Failed to query system history: {}", e)))?; + .or_query_error("Failed to query system history")?; let mut entries = Vec::new(); let mut last_entry: Option<(String, String, String)> = None; @@ -1997,7 +1994,7 @@ impl SystemHistoryProvider for PostgresBackend { &[&tenant_id], ) .await - .map_err(|e| internal_error(format!("Failed to count system history: {}", e)))?; + .or_query_error("Failed to count system history")?; let count: i64 = row.get(0); Ok(count as u64) @@ -2155,7 +2152,7 @@ impl PurgableStorage for PostgresBackend { &[&tenant_id, &resource_type, &id], ) .await - .map_err(|e| internal_error(format!("Failed to check history: {}", e)))?; + .or_query_error("Failed to check history")?; if history_exists.is_none() { return Err(StorageError::Resource(ResourceError::NotFound { @@ -2172,7 +2169,7 @@ impl PurgableStorage for PostgresBackend { &[&tenant_id, &resource_type, &id], ) .await - .map_err(|e| internal_error(format!("Failed to purge search index: {}", e)))?; + .or_query_error("Failed to purge search index")?; // Delete from FTS table let _ = client @@ -2189,7 +2186,7 @@ impl PurgableStorage for PostgresBackend { &[&tenant_id, &resource_type, &id], ) .await - .map_err(|e| internal_error(format!("Failed to purge resource history: {}", e)))?; + .or_query_error("Failed to purge resource history")?; // Delete from resources table client @@ -2198,7 +2195,7 @@ impl PurgableStorage for PostgresBackend { &[&tenant_id, &resource_type, &id], ) .await - .map_err(|e| internal_error(format!("Failed to purge resource: {}", e)))?; + .or_query_error("Failed to purge resource")?; Ok(()) } @@ -2214,7 +2211,7 @@ impl PurgableStorage for PostgresBackend { &[&tenant_id, &resource_type], ) .await - .map_err(|e| internal_error(format!("Failed to count resources: {}", e)))?; + .or_query_error("Failed to count resources")?; let count: i64 = row.get(0); // Delete from search index first (due to FK constraint) @@ -2224,7 +2221,7 @@ impl PurgableStorage for PostgresBackend { &[&tenant_id, &resource_type], ) .await - .map_err(|e| internal_error(format!("Failed to purge search index: {}", e)))?; + .or_query_error("Failed to purge search index")?; // Delete from FTS table let _ = client @@ -2241,7 +2238,7 @@ impl PurgableStorage for PostgresBackend { &[&tenant_id, &resource_type], ) .await - .map_err(|e| internal_error(format!("Failed to purge resource history: {}", e)))?; + .or_query_error("Failed to purge resource history")?; // Delete from resources table client @@ -2250,7 +2247,7 @@ impl PurgableStorage for PostgresBackend { &[&tenant_id, &resource_type], ) .await - .map_err(|e| internal_error(format!("Failed to purge resources: {}", e)))?; + .or_query_error("Failed to purge resources")?; Ok(count as u64) } @@ -3299,7 +3296,7 @@ impl ReindexTarget for PostgresBackend { &[&tenant_id], ) .await - .map_err(|e| internal_error(format!("Failed to clear search index: {}", e)))?; + .or_query_error("Failed to clear search index")?; // Also clear FTS entries let _ = client diff --git a/crates/persistence/src/backends/sqlite/search_impl.rs b/crates/persistence/src/backends/sqlite/search_impl.rs index dfccdc388..8e3e646ff 100644 --- a/crates/persistence/src/backends/sqlite/search_impl.rs +++ b/crates/persistence/src/backends/sqlite/search_impl.rs @@ -18,7 +18,7 @@ use crate::core::{ ChainedSearchProvider, IncludeProvider, MultiTypeSearchProvider, ResourceStorage, RevincludeProvider, SearchProvider, SearchResult, }; -use crate::error::{BackendError, StorageError, StorageResult}; +use crate::error::{BackendError, QueryErrorExt, StorageError, StorageResult}; use crate::tenant::TenantContext; use crate::types::{ CursorDirection, CursorValue, IncludeDirective, Page, PageCursor, PageInfo, @@ -216,7 +216,7 @@ impl SearchProvider for SqliteBackend { let mut stmt = conn .prepare(&sql) - .map_err(|e| internal_error(format!("Failed to prepare search query: {}", e)))?; + .or_query_error("Failed to prepare search query")?; // Bind params: tenant, type, then (cursor) sort value + id, then filter params. let mut all_params: Vec> = vec![ @@ -256,9 +256,9 @@ impl SearchProvider for SqliteBackend { }; Ok((id, version_id, data, last_updated, fhir_version, sort_key)) }) - .map_err(|e| internal_error(format!("Failed to execute search: {}", e)))? + .or_query_error("Failed to execute search")? .collect::, _>>() - .map_err(|e| internal_error(format!("Failed to read row: {}", e)))?; + .or_query_error("Failed to read row")?; // Parse rows, carrying the sort key for cursor construction. let mut parsed: Vec<(StoredResource, Option)> = Vec::new(); @@ -389,7 +389,7 @@ impl SearchProvider for SqliteBackend { let count: i64 = conn .query_row(&sql, param_refs.as_slice(), |row| row.get(0)) - .map_err(|e| internal_error(format!("Failed to count resources: {}", e)))?; + .or_query_error("Failed to count resources")?; Ok(count as u64) } @@ -453,7 +453,7 @@ impl MultiTypeSearchProvider for SqliteBackend { let mut stmt = conn .prepare(&sql) - .map_err(|e| internal_error(format!("Failed to prepare multi-type search: {}", e)))?; + .or_query_error("Failed to prepare multi-type search")?; let rows = stmt .query_map(params![tenant_id], |row| { @@ -472,12 +472,12 @@ impl MultiTypeSearchProvider for SqliteBackend { fhir_version, )) }) - .map_err(|e| internal_error(format!("Failed to execute multi-type search: {}", e)))?; + .or_query_error("Failed to execute multi-type search")?; let mut resources = Vec::new(); for row in rows { let (resource_type, id, version_id, data, last_updated_str, fhir_version_str) = - row.map_err(|e| internal_error(format!("Failed to read row: {}", e)))?; + row.or_query_error("Failed to read row")?; let json_data: serde_json::Value = serde_json::from_slice(&data) .map_err(|e| internal_error(format!("Failed to deserialize resource: {}", e)))?; @@ -640,9 +640,9 @@ impl RevincludeProvider for SqliteBackend { .join(" OR ") ); - let mut stmt = conn.prepare(&sql).map_err(|e| { - internal_error(format!("Failed to prepare revinclude query: {}", e)) - })?; + let mut stmt = conn + .prepare(&sql) + .or_query_error("Failed to prepare revinclude query")?; // Build params: tenant_id, source_type, then all the patterns let mut param_values: Vec> = Vec::new(); @@ -664,13 +664,11 @@ impl RevincludeProvider for SqliteBackend { let fhir_version: String = row.get(4)?; Ok((id, version_id, data, last_updated, fhir_version)) }) - .map_err(|e| { - internal_error(format!("Failed to execute revinclude query: {}", e)) - })?; + .or_query_error("Failed to execute revinclude query")?; for row in rows { let (id, version_id, data, last_updated_str, fhir_version_str) = - row.map_err(|e| internal_error(format!("Failed to read row: {}", e)))?; + row.or_query_error("Failed to read row")?; // Skip if we've already included this resource let resource_key = format!("{}/{}", revinclude.source_type, id); @@ -766,7 +764,7 @@ impl ChainedSearchProvider for SqliteBackend { // Bind parameters: tenant_id, resource_type, then fragment params let mut stmt = conn .prepare(&sql) - .map_err(|e| internal_error(format!("Failed to prepare chain query: {}", e)))?; + .or_query_error("Failed to prepare chain query")?; // Build parameter vector for rusqlite let mut bound_params: Vec> = Vec::new(); @@ -786,11 +784,11 @@ impl ChainedSearchProvider for SqliteBackend { let rows = stmt .query_map(params_ref.as_slice(), |row| row.get::<_, String>(0)) - .map_err(|e| internal_error(format!("Failed to execute chain query: {}", e)))?; + .or_query_error("Failed to execute chain query")?; let mut ids = Vec::new(); for row in rows { - ids.push(row.map_err(|e| internal_error(format!("Failed to read row: {}", e)))?); + ids.push(row.or_query_error("Failed to read row")?); } Ok(ids) @@ -832,7 +830,7 @@ impl ChainedSearchProvider for SqliteBackend { let mut stmt = conn .prepare(&sql) - .map_err(|e| internal_error(format!("Failed to prepare reverse chain query: {}", e)))?; + .or_query_error("Failed to prepare reverse chain query")?; // Build parameter vector for rusqlite let mut bound_params: Vec> = Vec::new(); @@ -852,11 +850,11 @@ impl ChainedSearchProvider for SqliteBackend { let rows = stmt .query_map(params_ref.as_slice(), |row| row.get::<_, String>(0)) - .map_err(|e| internal_error(format!("Failed to execute reverse chain query: {}", e)))?; + .or_query_error("Failed to execute reverse chain query")?; let mut ids = Vec::new(); for row in rows { - ids.push(row.map_err(|e| internal_error(format!("Failed to read row: {}", e)))?); + ids.push(row.or_query_error("Failed to read row")?); } Ok(ids) @@ -925,9 +923,9 @@ impl SqliteBackend { let matches: Vec<(String, String, Option)> = match builder.build_contained(query) { Some(fragment) => { let conn = self.get_connection()?; - let mut stmt = conn.prepare(&fragment.sql).map_err(|e| { - internal_error(format!("Failed to prepare contained query: {e}")) - })?; + let mut stmt = conn + .prepare(&fragment.sql) + .or_query_error("Failed to prepare contained query")?; let mut all_params: Vec> = vec![ Box::new(tenant_id.to_string()), Box::new(contained_type.to_string()), @@ -949,9 +947,9 @@ impl SqliteBackend { row.get::<_, Option>(2)?, )) }) - .map_err(|e| internal_error(format!("Failed to execute contained query: {e}")))? + .or_query_error("Failed to execute contained query")? .collect::, _>>() - .map_err(|e| internal_error(format!("Failed to read contained row: {e}")))? + .or_query_error("Failed to read contained row")? } None => Vec::new(), }; @@ -1219,17 +1217,17 @@ impl SqliteBackend { let mut stmt = conn .prepare(&sql) - .map_err(|e| internal_error(format!("Failed to prepare find query: {}", e)))?; + .or_query_error("Failed to prepare find query")?; let rows = stmt .query_map(params![tenant_id, resource_type, param_name], |row| { row.get::<_, String>(0) }) - .map_err(|e| internal_error(format!("Failed to execute find query: {}", e)))?; + .or_query_error("Failed to execute find query")?; let mut ids = Vec::new(); for row in rows { - ids.push(row.map_err(|e| internal_error(format!("Failed to read row: {}", e)))?); + ids.push(row.or_query_error("Failed to read row")?); } Ok(ids) @@ -1248,7 +1246,7 @@ impl SqliteBackend { "SELECT id, version_id, data, last_updated, fhir_version FROM resources WHERE tenant_id = ?1 AND resource_type = ?2 AND is_deleted = 0", ) - .map_err(|e| internal_error(format!("Failed to prepare query: {}", e)))?; + .or_query_error("Failed to prepare query")?; let rows = stmt .query_map(params![tenant_id, resource_type], |row| { @@ -1259,12 +1257,12 @@ impl SqliteBackend { let fhir_version: String = row.get(4)?; Ok((id, version_id, data, last_updated, fhir_version)) }) - .map_err(|e| internal_error(format!("Failed to query resources: {}", e)))?; + .or_query_error("Failed to query resources")?; let mut resources = Vec::new(); for row in rows { let (id, version_id, data, last_updated_str, fhir_version_str) = - row.map_err(|e| internal_error(format!("Failed to read row: {}", e)))?; + row.or_query_error("Failed to read row")?; let json_data: serde_json::Value = serde_json::from_slice(&data) .map_err(|e| internal_error(format!("Failed to deserialize: {}", e)))?; diff --git a/crates/persistence/src/backends/sqlite/storage.rs b/crates/persistence/src/backends/sqlite/storage.rs index e0fbfb3ef..41ff00ef5 100644 --- a/crates/persistence/src/backends/sqlite/storage.rs +++ b/crates/persistence/src/backends/sqlite/storage.rs @@ -19,7 +19,9 @@ use crate::core::{ if_match_field_satisfied, normalize_etag, }; use crate::error::TransactionError; -use crate::error::{BackendError, ConcurrencyError, ResourceError, StorageError, StorageResult}; +use crate::error::{ + BackendError, ConcurrencyError, QueryErrorExt, ResourceError, StorageError, StorageResult, +}; use crate::search::extractor::ExtractedValue; use crate::search::reindex::{ReindexSource, ReindexTarget, ResourcePage}; use crate::tenant::{Operation, TenantContext}; @@ -497,7 +499,7 @@ impl ResourceStorage for SqliteBackend { |row| row.get(0), ) } - .map_err(|e| internal_error(format!("Failed to count resources: {}", e)))?; + .or_query_error("Failed to count resources")?; Ok(count as u64) } @@ -535,7 +537,7 @@ impl ResourceStorage for SqliteBackend { AND last_updated >= ?3 \ GROUP BY day ORDER BY day", ) - .map_err(|e| internal_error(format!("Failed to prepare count_by_day: {}", e)))?; + .or_query_error("Failed to prepare count_by_day")?; let rows = stmt .query_map(params![tenant_id, resource_type, since_bound], |row| { @@ -543,12 +545,11 @@ impl ResourceStorage for SqliteBackend { let n: i64 = row.get(1)?; Ok((day, n)) }) - .map_err(|e| internal_error(format!("Failed to query count_by_day: {}", e)))?; + .or_query_error("Failed to query count_by_day")?; let mut out = Vec::new(); for row in rows { - let (day_str, n) = - row.map_err(|e| internal_error(format!("Failed to read count_by_day row: {}", e)))?; + let (day_str, n) = row.or_query_error("Failed to read count_by_day row")?; if let Ok(day) = chrono::NaiveDate::parse_from_str(&day_str, "%Y-%m-%d") { out.push(crate::core::DailyResourceCount { day, @@ -597,9 +598,7 @@ impl ResourceStorage for SqliteBackend { WHERE tenant_id = ?1 AND resource_type = ?2 AND last_updated >= ?3 \ GROUP BY bucket HAVING delta != 0 ORDER BY bucket", ) - .map_err(|e| { - internal_error(format!("Failed to prepare count_deltas_by_bucket: {}", e)) - })?; + .or_query_error("Failed to prepare count_deltas_by_bucket")?; let rows = stmt .query_map( @@ -610,15 +609,12 @@ impl ResourceStorage for SqliteBackend { Ok((bucket, delta)) }, ) - .map_err(|e| { - internal_error(format!("Failed to query count_deltas_by_bucket: {}", e)) - })?; + .or_query_error("Failed to query count_deltas_by_bucket")?; let mut out = Vec::new(); for row in rows { - let (bucket, delta) = row.map_err(|e| { - internal_error(format!("Failed to read count_deltas_by_bucket row: {}", e)) - })?; + let (bucket, delta) = + row.or_query_error("Failed to read count_deltas_by_bucket row")?; if let Some(bucket_start) = chrono::DateTime::from_timestamp(bucket, 0) { out.push(crate::core::ResourceCountDelta { bucket_start, @@ -658,7 +654,7 @@ impl ResourceStorage for SqliteBackend { WHERE tenant_id = ?1 AND last_updated >= ?2 \ GROUP BY wd, hr", ) - .map_err(|e| internal_error(format!("Failed to prepare activity_histogram: {}", e)))?; + .or_query_error("Failed to prepare activity_histogram")?; let rows = stmt .query_map(params![tenant_id, since_bound], |row| { @@ -667,12 +663,11 @@ impl ResourceStorage for SqliteBackend { let n: i64 = row.get(2)?; Ok((wd, hr, n)) }) - .map_err(|e| internal_error(format!("Failed to query activity_histogram: {}", e)))?; + .or_query_error("Failed to query activity_histogram")?; let mut out = Vec::new(); for row in rows { - let (wd, hr, n) = - row.map_err(|e| internal_error(format!("Failed to read activity row: {}", e)))?; + let (wd, hr, n) = row.or_query_error("Failed to read activity row")?; out.push(crate::core::ActivityCell { weekday: wd.clamp(0, 6) as u8, hour: hr.clamp(0, 23) as u8, @@ -691,17 +686,17 @@ impl ResourceStorage for SqliteBackend { WHERE tenant_id = ?1 AND is_deleted = 0 \ GROUP BY resource_type", ) - .map_err(|e| internal_error(format!("Failed to prepare count_all_types: {}", e)))?; + .or_query_error("Failed to prepare count_all_types")?; let rows = stmt .query_map(params![tenant_id], |row| { let rt: String = row.get(0)?; let n: i64 = row.get(1)?; Ok((rt, n.max(0) as u64)) }) - .map_err(|e| internal_error(format!("Failed to query count_all_types: {}", e)))?; + .or_query_error("Failed to query count_all_types")?; let mut out = Vec::new(); for row in rows { - out.push(row.map_err(|e| internal_error(format!("count_all_types row: {}", e)))?); + out.push(row.or_query_error("count_all_types row")?); } Ok(out) } @@ -739,17 +734,17 @@ impl ResourceStorage for SqliteBackend { let mut stmt = conn .prepare(&sql) - .map_err(|e| internal_error(format!("Failed to prepare count_by_types: {}", e)))?; + .or_query_error("Failed to prepare count_by_types")?; let rows = stmt .query_map(rusqlite::params_from_iter(binds), |row| { let rt: String = row.get(0)?; let n: i64 = row.get(1)?; Ok((rt, n.max(0) as u64)) }) - .map_err(|e| internal_error(format!("Failed to query count_by_types: {}", e)))?; + .or_query_error("Failed to query count_by_types")?; let mut out = Vec::new(); for row in rows { - out.push(row.map_err(|e| internal_error(format!("count_by_types row: {}", e)))?); + out.push(row.or_query_error("count_by_types row")?); } Ok(out) } @@ -762,17 +757,17 @@ impl ResourceStorage for SqliteBackend { "SELECT tenant_id, COUNT(*) FROM resources \ WHERE is_deleted = 0 GROUP BY tenant_id", ) - .map_err(|e| internal_error(format!("Failed to prepare count_by_tenant: {}", e)))?; + .or_query_error("Failed to prepare count_by_tenant")?; let rows = stmt .query_map([], |row| { let tid: String = row.get(0)?; let n: i64 = row.get(1)?; Ok((tid, n.max(0) as u64)) }) - .map_err(|e| internal_error(format!("Failed to query count_by_tenant: {}", e)))?; + .or_query_error("Failed to query count_by_tenant")?; let mut out = Vec::new(); for row in rows { - out.push(row.map_err(|e| internal_error(format!("count_by_tenant row: {}", e)))?); + out.push(row.or_query_error("count_by_tenant row")?); } Ok(out) } @@ -867,9 +862,7 @@ impl ResourceStorage for SqliteBackend { async fn purge_tenant_data(&self, id: &str) -> StorageResult { let mut conn = self.get_connection()?; - let tx = conn - .transaction() - .map_err(|e| internal_error(format!("purge begin: {e}")))?; + let tx = conn.transaction().or_query_error("purge begin")?; // Count current-version rows first so we can report what was removed. let removed: i64 = tx .query_row( @@ -877,7 +870,7 @@ impl ResourceStorage for SqliteBackend { params![id], |row| row.get(0), ) - .map_err(|e| internal_error(format!("purge count: {e}")))?; + .or_query_error("purge count")?; // search_index has ON DELETE CASCADE from resources, but delete it // explicitly too in case foreign keys are not enforced on this handle. for sql in [ @@ -886,10 +879,9 @@ impl ResourceStorage for SqliteBackend { "DELETE FROM resources WHERE tenant_id = ?1", ] { tx.execute(sql, params![id]) - .map_err(|e| internal_error(format!("purge delete: {e}")))?; + .or_query_error("purge delete")?; } - tx.commit() - .map_err(|e| internal_error(format!("purge commit: {e}")))?; + tx.commit().or_query_error("purge commit")?; Ok(removed.max(0) as u64) } } @@ -1669,7 +1661,7 @@ impl InstanceHistoryProvider for SqliteBackend { let mut stmt = conn .prepare(&sql) - .map_err(|e| internal_error(format!("Failed to prepare history query: {}", e)))?; + .or_query_error("Failed to prepare history query")?; let rows = stmt .query_map(params![tenant_id, resource_type, id], |row| { @@ -1680,14 +1672,14 @@ impl InstanceHistoryProvider for SqliteBackend { let fhir_version: String = row.get(4)?; Ok((version_id, data, last_updated, is_deleted, fhir_version)) }) - .map_err(|e| internal_error(format!("Failed to query history: {}", e)))?; + .or_query_error("Failed to query history")?; let mut entries = Vec::new(); let mut last_version: Option = None; for row in rows { let (version_id, data, last_updated_str, is_deleted, fhir_version_str) = - row.map_err(|e| internal_error(format!("Failed to read history row: {}", e)))?; + row.or_query_error("Failed to read history row")?; // Stop if we've collected enough items (we fetched count+1 to detect more) if entries.len() >= params.pagination.count as usize { @@ -1775,7 +1767,7 @@ impl InstanceHistoryProvider for SqliteBackend { params![tenant_id, resource_type, id], |row| row.get(0), ) - .map_err(|e| internal_error(format!("Failed to count history: {}", e)))?; + .or_query_error("Failed to count history")?; Ok(count as u64) } @@ -1833,7 +1825,7 @@ impl InstanceHistoryProvider for SqliteBackend { WHERE tenant_id = ?1 AND resource_type = ?2 AND id = ?3 AND version_id != ?4", params![tenant_id, resource_type, id, current_version], ) - .map_err(|e| internal_error(format!("Failed to delete history: {}", e)))?; + .or_query_error("Failed to delete history")?; Ok(deleted as u64) } @@ -1978,7 +1970,7 @@ impl TypeHistoryProvider for SqliteBackend { let mut stmt = conn .prepare(&sql) - .map_err(|e| internal_error(format!("Failed to prepare type history query: {}", e)))?; + .or_query_error("Failed to prepare type history query")?; let rows = stmt .query_map(params![tenant_id, resource_type], |row| { @@ -1990,14 +1982,14 @@ impl TypeHistoryProvider for SqliteBackend { let fhir_version: String = row.get(5)?; Ok((id, version_id, data, last_updated, is_deleted, fhir_version)) }) - .map_err(|e| internal_error(format!("Failed to query type history: {}", e)))?; + .or_query_error("Failed to query type history")?; let mut entries = Vec::new(); let mut last_entry: Option<(String, String)> = None; // (last_updated, id) for row in rows { let (id, version_id, data, last_updated_str, is_deleted, fhir_version_str) = - row.map_err(|e| internal_error(format!("Failed to read type history row: {}", e)))?; + row.or_query_error("Failed to read type history row")?; // Stop if we've collected enough items (we fetched count+1 to detect more) if entries.len() >= params.pagination.count as usize { @@ -2160,9 +2152,9 @@ impl SystemHistoryProvider for SqliteBackend { sql.push_str(" ORDER BY last_updated DESC, resource_type DESC, id DESC, CAST(version_id AS INTEGER) DESC"); sql.push_str(&format!(" LIMIT {}", params.pagination.count + 1)); // +1 to detect if there are more - let mut stmt = conn.prepare(&sql).map_err(|e| { - internal_error(format!("Failed to prepare system history query: {}", e)) - })?; + let mut stmt = conn + .prepare(&sql) + .or_query_error("Failed to prepare system history query")?; let rows = stmt .query_map(params![tenant_id], |row| { @@ -2183,7 +2175,7 @@ impl SystemHistoryProvider for SqliteBackend { fhir_version, )) }) - .map_err(|e| internal_error(format!("Failed to query system history: {}", e)))?; + .or_query_error("Failed to query system history")?; let mut entries = Vec::new(); let mut last_entry: Option<(String, String, String)> = None; // (last_updated, resource_type, id) @@ -2197,8 +2189,7 @@ impl SystemHistoryProvider for SqliteBackend { last_updated_str, is_deleted, fhir_version_str, - ) = row - .map_err(|e| internal_error(format!("Failed to read system history row: {}", e)))?; + ) = row.or_query_error("Failed to read system history row")?; // Stop if we've collected enough items (we fetched count+1 to detect more) if entries.len() >= params.pagination.count as usize { @@ -2297,7 +2288,7 @@ impl SystemHistoryProvider for SqliteBackend { params![tenant_id], |row| row.get(0), ) - .map_err(|e| internal_error(format!("Failed to count system history: {}", e)))?; + .or_query_error("Failed to count system history")?; Ok(count as u64) } @@ -2346,21 +2337,21 @@ impl PurgableStorage for SqliteBackend { "DELETE FROM resources WHERE tenant_id = ?1 AND resource_type = ?2 AND id = ?3", params![tenant_id, resource_type, id], ) - .map_err(|e| internal_error(format!("Failed to purge resource: {}", e)))?; + .or_query_error("Failed to purge resource")?; // Delete from history table conn.execute( "DELETE FROM resource_history WHERE tenant_id = ?1 AND resource_type = ?2 AND id = ?3", params![tenant_id, resource_type, id], ) - .map_err(|e| internal_error(format!("Failed to purge resource history: {}", e)))?; + .or_query_error("Failed to purge resource history")?; // Delete from search index conn.execute( "DELETE FROM search_index WHERE tenant_id = ?1 AND resource_type = ?2 AND resource_id = ?3", params![tenant_id, resource_type, id], ) - .map_err(|e| internal_error(format!("Failed to purge search index: {}", e)))?; + .or_query_error("Failed to purge search index")?; Ok(()) } @@ -2383,21 +2374,21 @@ impl PurgableStorage for SqliteBackend { "DELETE FROM resources WHERE tenant_id = ?1 AND resource_type = ?2", params![tenant_id, resource_type], ) - .map_err(|e| internal_error(format!("Failed to purge resources: {}", e)))?; + .or_query_error("Failed to purge resources")?; // Delete from history table conn.execute( "DELETE FROM resource_history WHERE tenant_id = ?1 AND resource_type = ?2", params![tenant_id, resource_type], ) - .map_err(|e| internal_error(format!("Failed to purge resource history: {}", e)))?; + .or_query_error("Failed to purge resource history")?; // Delete from search index conn.execute( "DELETE FROM search_index WHERE tenant_id = ?1 AND resource_type = ?2", params![tenant_id, resource_type], ) - .map_err(|e| internal_error(format!("Failed to purge search index: {}", e)))?; + .or_query_error("Failed to purge search index")?; Ok(count as u64) } @@ -3665,7 +3656,7 @@ impl ReindexTarget for SqliteBackend { "DELETE FROM search_index WHERE tenant_id = ?1", params![tenant_id], ) - .map_err(|e| internal_error(format!("Failed to clear search index: {}", e)))?; + .or_query_error("Failed to clear search index")?; Ok(deleted as u64) } diff --git a/crates/persistence/src/error.rs b/crates/persistence/src/error.rs index 36a32ec5d..e059f5bb9 100644 --- a/crates/persistence/src/error.rs +++ b/crates/persistence/src/error.rs @@ -426,6 +426,32 @@ pub enum BackendError { backend_name: String, }, + /// The backend cancelled the operation because it exceeded a server-side + /// time limit. + /// + /// This is a *statement*-level deadline (PostgreSQL `statement_timeout` + /// → SQLSTATE `57014`, MongoDB `maxTimeMS` → `MaxTimeMSExpired`, an + /// explicit SQLite `interrupt`), not a connection-level failure: the + /// backend is healthy and reachable, and it deliberately stopped *this* + /// statement. That distinction is why it does not map onto + /// [`Self::Unavailable`] — the server is fine, the query was too + /// expensive — and why the REST layer answers `504` rather than `503` + /// (see `helios_rest::error`, issue #353). + /// + /// Lock-wait expiry is deliberately **not** this variant. SQLite's + /// `SQLITE_BUSY` after `busy_timeout` means "someone else held the write + /// lock", which a retry genuinely resolves, so it classifies as + /// [`Self::Unavailable`] (503 + `Retry-After`) instead. + #[error("{backend_name} operation timed out: {message}")] + Timeout { + /// Backend identifier (e.g., `postgres`). + backend_name: String, + /// Human-readable failure detail, including the driver context. Never + /// surfaced to HTTP clients — the REST layer logs it and replies with a + /// fixed, backend-agnostic message. + message: String, + }, + /// The requested capability is not supported by this backend. #[error("capability '{capability}' not supported by {backend_name}")] UnsupportedCapability { @@ -688,14 +714,59 @@ impl From for BackendError { } } +/// Classifies a `rusqlite` error into a [`BackendError`], preserving the +/// driver's `ErrorCode` rather than collapsing everything to `Internal`. +/// +/// `context` is prepended to the driver text so the caller's description of +/// *what* it was doing survives classification. +/// +/// - `SQLITE_INTERRUPT` — the statement was deliberately cancelled +/// (`Connection::interrupt`) → [`BackendError::Timeout`] (504). +/// - `SQLITE_BUSY` / `SQLITE_LOCKED` — the `busy_timeout` elapsed waiting for +/// the write lock. The database is healthy and merely contended, and a retry +/// usually succeeds, so this is [`BackendError::Unavailable`] (503 + +/// `Retry-After`) — *not* `Timeout`. Before #353 it was a 500, which told +/// clients a transient lock conflict was a server defect. +/// - everything else — unchanged: [`BackendError::Internal`], byte-for-byte the +/// message this helper's callers produced before. #[cfg(feature = "sqlite")] -impl From for StorageError { - fn from(err: rusqlite::Error) -> Self { - StorageError::Backend(BackendError::Internal { +pub fn classify_sqlite_error(context: &str, err: rusqlite::Error) -> BackendError { + use rusqlite::ErrorCode; + + // `sqlite_error_code()` is rusqlite's own accessor for the primary result + // code; it yields `None` for the non-`SqliteFailure` variants (e.g. + // `QueryReturnedNoRows`, a type-conversion failure), which correctly fall + // through to `Internal` below. + let code = err.sqlite_error_code(); + let message = if context.is_empty() { + err.to_string() + } else { + format!("{context}: {err}") + }; + + match code { + Some(ErrorCode::OperationInterrupted) => BackendError::Timeout { backend_name: "sqlite".to_string(), - message: err.to_string(), + message, + }, + Some(ErrorCode::DatabaseBusy) | Some(ErrorCode::DatabaseLocked) => { + BackendError::Unavailable { + backend_name: "sqlite".to_string(), + message, + } + } + _ => BackendError::Internal { + backend_name: "sqlite".to_string(), + message, source: Some(Box::new(err)), - }) + }, + } +} + +#[cfg(feature = "sqlite")] +impl From for StorageError { + fn from(err: rusqlite::Error) -> Self { + StorageError::Backend(classify_sqlite_error("", err)) } } @@ -708,25 +779,215 @@ impl From for StorageError { } } +/// Classifies a `tokio_postgres` error into a [`BackendError`] by SQLSTATE, +/// preserving the code rather than collapsing everything to `Internal`. +/// +/// `context` is prepended to the driver text so the caller's description of +/// *what* it was doing survives classification. +/// +/// SQLSTATE is the only stable signal here: PostgreSQL localizes error +/// *messages* through `lc_messages`, so matching on the text +/// ("canceling statement due to statement timeout") breaks on any server not +/// running an English locale. Classification must therefore happen while the +/// typed error is still in hand — once it has been through `format!` the code +/// is gone (issue #353). +/// +/// - `57014 query_canceled` — the statement exceeded `statement_timeout` (see +/// `HFS_PG_STATEMENT_TIMEOUT_MS`) or was cancelled by `pg_cancel_backend` +/// → [`BackendError::Timeout`] (504). +/// - `53300 too_many_connections`, `53400 configuration_limit_exceeded`, +/// `57P01 admin_shutdown`, `57P02 crash_shutdown`, `57P03 cannot_connect_now` +/// — the server is saturated or going away, and a retry may well land +/// → [`BackendError::Unavailable`] (503 + `Retry-After`). +/// - everything else — unchanged: [`BackendError::Internal`], byte-for-byte the +/// message this helper's callers produced before. +/// +/// Deliberately **not** reclassified here: `40001 serialization_failure` and +/// `40P01 deadlock_detected`. Both are retryable, but deciding what a FHIR +/// client should see for a write conflict is a separate question from +/// timeouts, and silently 503-ing a deadlock could mask a real lock-ordering +/// defect. +#[cfg(feature = "postgres")] +pub fn classify_postgres_error(context: &str, err: tokio_postgres::Error) -> BackendError { + use tokio_postgres::error::SqlState; + + // `SqlState` is compared with `==`, never matched as a pattern: it is a + // newtype over an enum with an `Other(Box)` variant, which makes it + // non-structural-match, so `SqlState::QUERY_CANCELED` in a pattern position + // does not compile. + let code = err.code().cloned(); + let message = if context.is_empty() { + err.to_string() + } else { + format!("{context}: {err}") + }; + + if code.as_ref() == Some(&SqlState::QUERY_CANCELED) { + return BackendError::Timeout { + backend_name: "postgres".to_string(), + message, + }; + } + + let unavailable = matches!( + code.as_ref(), + Some(c) if *c == SqlState::TOO_MANY_CONNECTIONS + || *c == SqlState::CONFIGURATION_LIMIT_EXCEEDED + || *c == SqlState::ADMIN_SHUTDOWN + || *c == SqlState::CRASH_SHUTDOWN + || *c == SqlState::CANNOT_CONNECT_NOW + ); + if unavailable { + return BackendError::Unavailable { + backend_name: "postgres".to_string(), + message, + }; + } + + BackendError::Internal { + backend_name: "postgres".to_string(), + message, + source: Some(Box::new(err)), + } +} + #[cfg(feature = "postgres")] impl From for StorageError { fn from(err: tokio_postgres::Error) -> Self { - StorageError::Backend(BackendError::Internal { - backend_name: "postgres".to_string(), - message: err.to_string(), - source: Some(Box::new(err)), - }) + StorageError::Backend(classify_postgres_error("", err)) + } +} + +/// MongoDB server error code for a query that exceeded `maxTimeMS`. +#[cfg(feature = "mongodb")] +const MONGO_MAX_TIME_MS_EXPIRED: i32 = 50; +/// MongoDB server error code for an operation that exceeded its time limit. +#[cfg(feature = "mongodb")] +const MONGO_EXCEEDED_TIME_LIMIT: i32 = 262; + +/// Classifies a MongoDB driver error into a [`BackendError`], preserving the +/// server error code rather than collapsing everything to `Internal`. +/// +/// `context` is prepended to the driver text so the caller's description of +/// *what* it was doing survives classification. +/// +/// - `MaxTimeMSExpired` (50) / `ExceededTimeLimit` (262) — the server stopped +/// the operation at its deadline → [`BackendError::Timeout`] (504). +/// - `Io` / `ConnectionPoolCleared` / `ServerSelection` — transport or +/// topology failure → [`BackendError::Unavailable`] (503 + `Retry-After`). +/// - everything else — unchanged: [`BackendError::Internal`], byte-for-byte the +/// message this helper's callers produced before. +/// +/// Note that HFS does not currently set `maxTimeMS` on its queries, so the +/// timeout arm fires only when the deadline comes from the server or a +/// connection-string option. Wiring an HFS-side query deadline is tracked +/// separately; the classification is in place either way. +#[cfg(feature = "mongodb")] +pub fn classify_mongodb_error(context: &str, err: mongodb::error::Error) -> BackendError { + use mongodb::error::ErrorKind; + + let message = if context.is_empty() { + err.to_string() + } else { + format!("{context}: {err}") + }; + + let timed_out = matches!( + err.kind.as_ref(), + ErrorKind::Command(cmd) + if cmd.code == MONGO_MAX_TIME_MS_EXPIRED || cmd.code == MONGO_EXCEEDED_TIME_LIMIT + ); + if timed_out { + return BackendError::Timeout { + backend_name: "mongodb".to_string(), + message, + }; + } + + let unreachable = matches!( + err.kind.as_ref(), + ErrorKind::Io(_) + | ErrorKind::ConnectionPoolCleared { .. } + | ErrorKind::ServerSelection { .. } + ); + if unreachable { + return BackendError::Unavailable { + backend_name: "mongodb".to_string(), + message, + }; + } + + BackendError::Internal { + backend_name: "mongodb".to_string(), + message, + source: Some(Box::new(err)), } } #[cfg(feature = "mongodb")] impl From for StorageError { fn from(err: mongodb::error::Error) -> Self { - StorageError::Backend(BackendError::Internal { - backend_name: "mongodb".to_string(), - message: err.to_string(), - source: Some(Box::new(err)), - }) + StorageError::Backend(classify_mongodb_error("", err)) + } +} + +/// Classifies a raw driver error into a [`StorageError`] in place, tagging it +/// with `context` — what the caller was doing when the driver failed. +/// +/// This is the form every backend call site uses. It replaces the pre-#353 +/// spelling, which stringified the driver error and so threw away the SQLSTATE +/// / `ErrorCode` that distinguishes a cancelled statement from a real defect: +/// +/// ```ignore +/// // before — classification impossible, everything is a 500 +/// .map_err(|e| internal_error(format!("Failed to prepare count_by_types: {e}")))? +/// // after +/// .or_query_error("Failed to prepare count_by_types")? +/// ``` +/// +/// A method rather than a `.map_err(|e| …)` closure, for two reasons beyond +/// brevity. First, the driver error type is named once, here, instead of being +/// re-bound at ~150 call sites. Second, the conversion sits on the call chain +/// rather than inside a closure body that only ever runs on failure, so +/// coverage tooling attributes it to the enclosing function instead of marking +/// every error-handling site in the backends as unexecuted — #353 rewrote all +/// of them at once, which made that reporting artifact impossible to miss. +/// +/// The classification itself is unchanged and still lives in +/// `classify_sqlite_error` / `classify_postgres_error` / `classify_mongodb_error`, +/// where it is unit tested: a server-side deadline becomes +/// [`BackendError::Timeout`] (504), an unreachable or saturated server becomes +/// [`BackendError::Unavailable`] (503 + `Retry-After`), and everything else +/// stays [`BackendError::Internal`] (500) with byte-identical text to the +/// `internal_error(format!(…))` these call sites used before. +/// +/// Because each impl is written for one concrete driver error type, a site +/// whose `Result` carries some other error (serde, chrono, a parse) fails to +/// compile rather than being silently mis-converted. +pub trait QueryErrorExt { + /// Converts a driver failure into a classified [`StorageError`], prefixing + /// the driver text with `context`. + fn or_query_error(self, context: &str) -> Result; +} + +#[cfg(feature = "sqlite")] +impl QueryErrorExt for Result { + fn or_query_error(self, context: &str) -> Result { + self.map_err(|err| StorageError::Backend(classify_sqlite_error(context, err))) + } +} + +#[cfg(feature = "postgres")] +impl QueryErrorExt for Result { + fn or_query_error(self, context: &str) -> Result { + self.map_err(|err| StorageError::Backend(classify_postgres_error(context, err))) + } +} + +#[cfg(feature = "mongodb")] +impl QueryErrorExt for Result { + fn or_query_error(self, context: &str) -> Result { + self.map_err(|err| StorageError::Backend(classify_mongodb_error(context, err))) } } @@ -810,6 +1071,223 @@ mod tests { assert!(err.to_string().contains("line 42")); } + // ── Driver-error classification (issue #353) ──────────────────────────── + // + // `rusqlite::Error` is constructible, so the SQLite classifier is unit + // testable. `tokio_postgres::Error` and `mongodb::error::Error` have no + // public constructors, so their classifiers are covered by the + // testcontainer integration tests instead (see `tests/postgres_tests.rs`). + + /// Builds a `rusqlite::Error` carrying the given primary result code. + #[cfg(feature = "sqlite")] + fn sqlite_failure(primary_code: std::os::raw::c_int) -> rusqlite::Error { + rusqlite::Error::SqliteFailure(rusqlite::ffi::Error::new(primary_code), None) + } + + /// An interrupted statement is a server-side deadline, not a server fault: + /// it must classify as `Timeout` so the REST layer answers 504. + #[cfg(feature = "sqlite")] + #[test] + fn test_classify_sqlite_interrupt_is_timeout() { + // SQLITE_INTERRUPT == 9 + let err = classify_sqlite_error("Failed to execute search", sqlite_failure(9)); + assert!( + matches!(err, BackendError::Timeout { ref backend_name, .. } if backend_name == "sqlite"), + "SQLITE_INTERRUPT must classify as Timeout, got {err:?}" + ); + // The caller's context survives classification. + assert!(err.to_string().contains("Failed to execute search")); + } + + /// A lock-wait expiry is contention, not an over-long query: a retry + /// genuinely succeeds, so it must be `Unavailable` (503 + Retry-After), + /// NOT `Timeout` (504, which advises against retrying). Before #353 this + /// was a 500. + #[cfg(feature = "sqlite")] + #[test] + fn test_classify_sqlite_busy_and_locked_are_unavailable() { + // SQLITE_BUSY == 5, SQLITE_LOCKED == 6 + for code in [5, 6] { + let err = classify_sqlite_error("Failed to insert resource", sqlite_failure(code)); + assert!( + matches!(err, BackendError::Unavailable { .. }), + "SQLite primary code {code} must classify as Unavailable, got {err:?}" + ); + } + } + + /// Everything else keeps today's behaviour exactly: `Internal`, with the + /// same `"{context}: {err}"` message the call sites built before. This is + /// the property that makes the call-site conversion safe — an unclassified + /// error is byte-identical to the pre-#353 result. + #[cfg(feature = "sqlite")] + #[test] + fn test_classify_sqlite_other_errors_are_unchanged_internal() { + // SQLITE_CONSTRAINT == 19 — a genuine defect, must stay a 500. + let raw = sqlite_failure(19); + let expected = format!("Failed to insert resource: {raw}"); + let err = classify_sqlite_error("Failed to insert resource", sqlite_failure(19)); + match err { + BackendError::Internal { + backend_name, + message, + .. + } => { + assert_eq!(backend_name, "sqlite"); + assert_eq!( + message, expected, + "unclassified errors must keep the pre-#353 message verbatim" + ); + } + other => panic!("SQLITE_CONSTRAINT must stay Internal, got {other:?}"), + } + + // A non-`SqliteFailure` variant has no code at all and must also pass + // through untouched. + let err = classify_sqlite_error("ctx", rusqlite::Error::QueryReturnedNoRows); + assert!(matches!(err, BackendError::Internal { .. })); + } + + /// The `From` impl (used by bare `?` call sites) classifies too, with no + /// context prefix — so the message is the driver text alone, as before. + #[cfg(feature = "sqlite")] + #[test] + fn test_sqlite_from_impl_classifies_without_context_prefix() { + let raw = sqlite_failure(9); + let expected = raw.to_string(); + let err: StorageError = sqlite_failure(9).into(); + match err { + StorageError::Backend(BackendError::Timeout { message, .. }) => { + assert_eq!(message, expected, "empty context must add no prefix"); + } + other => panic!("expected a classified Timeout, got {other:?}"), + } + } + + /// Builds a `mongodb::error::Error` carrying a server command error with + /// the given code. + /// + /// `CommandError` is `#[non_exhaustive]` and has a private field, so it + /// cannot be built with a struct literal from outside the driver — but it + /// derives `Deserialize`, which is exactly how the driver itself builds one + /// from a server reply. Going through serde therefore constructs the same + /// value the driver would, rather than a test-only approximation. + #[cfg(feature = "mongodb")] + fn mongo_command_error(code: i32, code_name: &str) -> mongodb::error::Error { + let command_error: mongodb::error::CommandError = + serde_json::from_value(serde_json::json!({ + "code": code, + "codeName": code_name, + "errmsg": "operation exceeded time limit", + "topologyVersion": null, + })) + .expect("CommandError deserializes from a server-shaped reply"); + mongodb::error::ErrorKind::Command(command_error).into() + } + + /// Both server-side deadline codes must classify as `Timeout` so the REST + /// layer answers 504 rather than 500. + #[cfg(feature = "mongodb")] + #[test] + fn test_classify_mongodb_deadline_codes_are_timeout() { + for (code, name) in [ + (MONGO_MAX_TIME_MS_EXPIRED, "MaxTimeMSExpired"), + (MONGO_EXCEEDED_TIME_LIMIT, "ExceededTimeLimit"), + ] { + let err = + classify_mongodb_error("Failed to execute search", mongo_command_error(code, name)); + assert!( + matches!(err, BackendError::Timeout { ref backend_name, .. } if backend_name == "mongodb"), + "MongoDB {name} ({code}) must classify as Timeout, got {err:?}" + ); + // The caller's context survives classification. + assert!(err.to_string().contains("Failed to execute search")); + } + } + + /// A transport failure is the server being unreachable, not a statement + /// running long: `Unavailable` (503 + `Retry-After`), not `Timeout` (504). + #[cfg(feature = "mongodb")] + #[test] + fn test_classify_mongodb_io_is_unavailable() { + let io = std::io::Error::new(std::io::ErrorKind::ConnectionReset, "connection reset"); + let err: mongodb::error::Error = + mongodb::error::ErrorKind::Io(std::sync::Arc::new(io)).into(); + let err = classify_mongodb_error("Failed to read resource", err); + assert!( + matches!(err, BackendError::Unavailable { ref backend_name, .. } if backend_name == "mongodb"), + "a MongoDB I/O failure must classify as Unavailable, got {err:?}" + ); + } + + /// Any other command failure keeps today's behaviour exactly: `Internal`, + /// with the same `"{context}: {err}"` text the call sites built before — + /// the property that makes converting ~150 call sites safe. + #[cfg(feature = "mongodb")] + #[test] + fn test_classify_mongodb_other_errors_are_unchanged_internal() { + // 11000 DuplicateKey — a genuine defect at these call sites, stays 500. + let expected = format!( + "Failed to insert resource: {}", + mongo_command_error(11000, "DuplicateKey") + ); + let err = classify_mongodb_error( + "Failed to insert resource", + mongo_command_error(11000, "DuplicateKey"), + ); + match err { + BackendError::Internal { + backend_name, + message, + .. + } => { + assert_eq!(backend_name, "mongodb"); + assert_eq!( + message, expected, + "unclassified errors must keep the pre-#353 message verbatim" + ); + } + other => panic!("DuplicateKey must stay Internal, got {other:?}"), + } + } + + /// The `From` impl (used by bare `?` call sites) classifies too, with no + /// context prefix — so the message is the driver text alone, as before. + #[cfg(feature = "mongodb")] + #[test] + fn test_mongodb_from_impl_classifies_without_context_prefix() { + let expected = + mongo_command_error(MONGO_MAX_TIME_MS_EXPIRED, "MaxTimeMSExpired").to_string(); + let err: StorageError = + mongo_command_error(MONGO_MAX_TIME_MS_EXPIRED, "MaxTimeMSExpired").into(); + match err { + StorageError::Backend(BackendError::Timeout { message, .. }) => { + assert_eq!(message, expected, "empty context must add no prefix"); + } + other => panic!("expected a classified Timeout, got {other:?}"), + } + } + + /// `QueryErrorExt` is the spelling every backend call site uses, so the + /// classification it performs is worth pinning independently of the free + /// functions: an `Ok` passes through untouched, and an `Err` arrives + /// classified and context-tagged. + #[cfg(feature = "sqlite")] + #[test] + fn test_query_error_ext_classifies_in_place() { + let ok: Result = Ok(7); + assert_eq!(ok.or_query_error("Failed to count resources").unwrap(), 7); + + // SQLITE_INTERRUPT == 9 + let err: Result = Err(sqlite_failure(9)); + match err.or_query_error("Failed to count resources") { + Err(StorageError::Backend(BackendError::Timeout { message, .. })) => { + assert!(message.starts_with("Failed to count resources: ")); + } + other => panic!("expected a context-tagged Timeout, got {other:?}"), + } + } + #[test] fn test_storage_error_from_bulk_errors() { let export_err = BulkExportError::JobNotFound { diff --git a/crates/persistence/tests/postgres_tests.rs b/crates/persistence/tests/postgres_tests.rs index e8844857a..fb61cbcac 100644 --- a/crates/persistence/tests/postgres_tests.rs +++ b/crates/persistence/tests/postgres_tests.rs @@ -706,7 +706,7 @@ mod postgres_integration { use helios_persistence::core::SettingsStore; use helios_persistence::core::history::{HistoryParams, InstanceHistoryProvider}; use helios_persistence::core::{Backend, BackendCapability, BackendKind, ResourceStorage}; - use helios_persistence::error::{ConcurrencyError, ResourceError, StorageError}; + use helios_persistence::error::{BackendError, ConcurrencyError, ResourceError, StorageError}; use helios_persistence::tenant::{TenantContext, TenantId, TenantPermissions}; use testcontainers::ImageExt; @@ -883,6 +883,87 @@ mod postgres_integration { } } + /// A statement cancelled by `statement_timeout` must classify as + /// [`BackendError::Timeout`] (→ HTTP 504), not `Internal` (→ 500). + /// + /// Regression for issue #353. `tokio_postgres::Error` has no public + /// constructor, so the SQLSTATE-classification path can only be exercised + /// against a live server — hence a testcontainer test rather than a unit + /// test. `SELECT pg_sleep()` is the cheapest statement guaranteed to + /// outlive the deadline. + /// + /// Note this asserts on the SQLSTATE (`57014`) reaching the classifier, not + /// on the driver's message text: PostgreSQL localizes error messages via + /// `lc_messages`, so matching the English string would make this test (and + /// the classifier it guards) locale-dependent. + #[tokio::test] + async fn statement_timeout_cancellation_classifies_as_backend_timeout() { + use helios_persistence::error::classify_postgres_error; + + let pg = shared_pg().await; + const TIMEOUT_MS: u64 = 250; + + let config = PostgresConfig { + host: pg.host.clone(), + port: pg.port, + dbname: "postgres".to_string(), + user: "postgres".to_string(), + password: Some("postgres".to_string()), + statement_timeout_ms: TIMEOUT_MS, + ..Default::default() + }; + let backend = PostgresBackend::new(config).await.expect("create backend"); + let client = backend.get_client().await.expect("get_client"); + + // Sleep well past the 250ms budget so the server cancels us. + let err = client + .query("SELECT pg_sleep(5)", &[]) + .await + .expect_err("pg_sleep(5) must be cancelled by a 250ms statement_timeout"); + + assert_eq!( + err.code().map(|c| c.code()), + Some("57014"), + "expected SQLSTATE 57014 query_canceled, got {err}" + ); + + let classified = classify_postgres_error("Failed to execute search", err); + match classified { + BackendError::Timeout { + ref backend_name, + ref message, + } => { + assert_eq!(backend_name, "postgres"); + assert!( + message.starts_with("Failed to execute search: "), + "caller context must survive classification, got {message:?}" + ); + } + other => panic!( + "statement_timeout cancellation must classify as BackendError::Timeout \ + (HTTP 504), got {other:?} — this is the #353 regression" + ), + } + + // Call sites that add no context of their own convert with a bare `?`, + // which goes through `impl From for StorageError` + // rather than the classifier directly. That path must classify + // identically, or the fix would hold only for the sites that happen to + // pass a context string. + let err = client + .query("SELECT pg_sleep(5)", &[]) + .await + .expect_err("pg_sleep(5) must be cancelled by a 250ms statement_timeout"); + let converted: StorageError = err.into(); + assert!( + matches!( + converted, + StorageError::Backend(BackendError::Timeout { .. }) + ), + "the `?` conversion must classify too, got {converted:?}" + ); + } + // ======================================================================== // CRUD Tests // ======================================================================== diff --git a/crates/rest/src/error.rs b/crates/rest/src/error.rs index 66d3804f6..4305a1da9 100644 --- a/crates/rest/src/error.rs +++ b/crates/rest/src/error.rs @@ -19,9 +19,22 @@ //! | UnsupportedResourceType | 400 | not-supported | //! | AccessDenied | 403 | forbidden | //! | BackendError::{Unavailable, ConnectionFailed, PoolExhausted} | 503 | transient | +//! | BackendError::Timeout | 504 | timeout | //! | BackendError::UnsupportedCapability | 501 | not-supported | //! | BackendError::{Migration, Internal, Query, Serialization} | 500 | exception | //! +//! `503` and `504` are both transient, but they say different things and are +//! read differently by the infrastructure in front of the server: +//! +//! - **503** — *this instance* cannot serve right now (backend down, pool +//! exhausted, lock contention). Carries `Retry-After`; a load balancer taking +//! the instance out of rotation is the correct response. +//! - **504** — the instance is fine; one *statement* exceeded a server-side +//! time limit and the backend cancelled it (PostgreSQL `statement_timeout`, +//! MongoDB `maxTimeMS`). No `Retry-After`: the query is usually +//! deterministically too slow, so a prompt retry only adds load. Ejecting the +//! instance would be wrong. See issue #353. +//! //! [`RestError::NotSupported`] (400 + `not-supported`) is reserved for //! spec-defined parameters/features that the server explicitly refuses; //! [`RestError::NotImplemented`] (501 + `not-supported`) signals work that @@ -176,6 +189,27 @@ pub enum RestError { message: String, }, + /// A backend cancelled the operation at a server-side time limit + /// (HTTP 504) — e.g. PostgreSQL `statement_timeout` (issue #353). + /// + /// Deliberately **not** 503. A 503 asserts that *this instance* cannot + /// serve requests, which load balancers and service meshes routinely read + /// as "eject this backend from rotation". A statement timeout says nothing + /// about instance health — one expensive query was stopped — so ejecting + /// the instance would be wrong and, under the load that provoked the + /// timeout, actively harmful. + /// + /// Also deliberately **not** accompanied by `Retry-After`. A cancelled + /// query is usually deterministically too slow for the configured budget, + /// so inviting a prompt retry just multiplies load on an already-strained + /// database. The 503 family keeps its `Retry-After`, where a retry + /// genuinely helps. + GatewayTimeout { + /// Error message. Internal only — sanitized before it reaches the + /// client (see [`RestError::client_response`]). + message: String, + }, + /// Not implemented (HTTP 501). NotImplemented { /// Description of what's not implemented. @@ -283,6 +317,9 @@ impl fmt::Display for RestError { RestError::ServiceUnavailable { message } => { write!(f, "Service unavailable: {}", message) } + RestError::GatewayTimeout { message } => { + write!(f, "Backend timeout: {}", message) + } RestError::NotImplemented { feature } => { write!(f, "Not implemented: {}", feature) } @@ -397,6 +434,30 @@ impl RestError { "transient", message.clone(), ), + RestError::GatewayTimeout { message } => { + // Keep the operator's diagnosis intact server-side — the + // SQLSTATE, the context string, and the driver text are all in + // `message`, and they are what tells an operator *which* + // statement blew the budget and whether to raise + // `HFS_PG_STATEMENT_TIMEOUT_MS` or fix the query. + tracing::warn!( + error.detail = %message, + "backend cancelled a statement at its server-side time limit" + ); + // The client gets none of that. Naming the backend product or + // an `HFS_*` environment variable here would fingerprint the + // deployment while being useless to the recipient — a client + // cannot set a server env var. Tell them the one thing they + // *can* act on: make the query cheaper. + ( + StatusCode::GATEWAY_TIMEOUT, + "timeout", + "The request exceeded the server's time limit for a single operation. \ + Narrow the request — add or tighten search parameters, reduce _count, \ + or request a smaller date range — and try again." + .to_string(), + ) + } RestError::NotImplemented { feature } => ( StatusCode::NOT_IMPLEMENTED, "not-supported", @@ -744,8 +805,15 @@ impl From for RestError { TransactionError::BundleError { index, message } => RestError::BadRequest { message: format!("Bundle entry {}: {}", index, message), }, - TransactionError::Timeout { .. } - | TransactionError::RolledBack { .. } + // A transaction that ran out of time is the same condition as a + // cancelled statement, one level up: the backend is healthy and + // stopped work that exceeded its budget. It answered 500 until + // #353, which is the identical mis-classification this issue fixes + // for `BackendError::Timeout`. + TransactionError::Timeout { .. } => RestError::GatewayTimeout { + message: err.to_string(), + }, + TransactionError::RolledBack { .. } | TransactionError::InvalidTransaction | TransactionError::NestedNotSupported | TransactionError::UnsupportedIsolationLevel { .. } => RestError::InternalError { @@ -778,6 +846,14 @@ impl From for RestError { message: format!("connection pool exhausted for {backend_name}"), }, + // A server-side statement deadline elapsed (e.g. PostgreSQL + // `statement_timeout` → SQLSTATE 57014). The backend is healthy and + // deliberately stopped one over-long statement, so this is neither + // a server defect (500) nor an instance-level outage (503) — see + // `RestError::GatewayTimeout` for why the distinction matters to + // load balancers. Issue #353. + BackendError::Timeout { message, .. } => RestError::GatewayTimeout { message }, + // Genuine server-side faults: retrying will not help, so keep them // 500. Their raw detail is sanitized and logged by // `RestError::client_response`, never leaked to the client. @@ -1311,6 +1387,13 @@ mod tests { }, StatusCode::INTERNAL_SERVER_ERROR, ), + ( + BackendError::Timeout { + backend_name: "postgres".to_string(), + message: "canceling statement due to statement timeout".to_string(), + }, + StatusCode::GATEWAY_TIMEOUT, + ), ]; for (err, expected) in cases { let (status, _, _) = RestError::from(err).client_response(); @@ -1318,6 +1401,95 @@ mod tests { } } + // ── BackendError::Timeout → 504 (issue #353) ─────────────────────────── + + /// A statement cancelled at a server-side deadline is a `504` with the FHIR + /// `timeout` issue code — not the `500`/`exception` it produced before. + #[test] + fn test_backend_timeout_maps_to_504_timeout() { + let err = BackendError::Timeout { + backend_name: "postgres".to_string(), + message: "Failed to execute search: db error: ERROR: canceling statement due to \ + statement timeout" + .to_string(), + }; + let rest_err = RestError::from(err); + let (status, code, _) = rest_err.client_response(); + assert_eq!(status, StatusCode::GATEWAY_TIMEOUT); + assert_eq!(code, "timeout"); + + // `Display` is what reaches the operator's logs and any `{err}` in a + // caller's own message, so it must keep the backend detail the client + // response strips. + let displayed = rest_err.to_string(); + assert!( + displayed.starts_with("Backend timeout: "), + "unexpected Display form: {displayed}" + ); + assert!(displayed.contains("canceling statement due to statement timeout")); + } + + /// The client-facing text must not carry driver internals, the backend + /// product name, or an `HFS_*` variable name — it should say only what the + /// caller can act on. The operator's copy goes to the log instead. + #[test] + fn test_backend_timeout_message_is_sanitized() { + let err = BackendError::Timeout { + backend_name: "postgres".to_string(), + message: "Failed to execute search: db error: ERROR: canceling statement due to \ + statement timeout; SELECT id FROM resources WHERE tenant_id = $1" + .to_string(), + }; + let (_, _, message) = RestError::from(err).client_response(); + for leak in [ + "canceling statement", + "db error", + "SELECT", + "resources", + "postgres", + "statement_timeout", + "HFS_PG_STATEMENT_TIMEOUT_MS", + ] { + assert!( + !message.contains(leak), + "504 diagnostic leaked {leak:?}: {message}" + ); + } + // It must still be actionable rather than merely opaque. + assert!(message.contains("time limit")); + assert!(message.contains("Narrow the request")); + } + + /// A transaction that exceeded its budget is the same condition one level + /// up, and answered 500 until #353. + #[test] + fn test_transaction_timeout_maps_to_504() { + let err = TransactionError::Timeout { timeout_ms: 30_000 }; + let (status, code, _) = RestError::from(err).client_response(); + assert_eq!(status, StatusCode::GATEWAY_TIMEOUT); + assert_eq!(code, "timeout"); + } + + /// The 504 deliberately carries no `Retry-After`: a cancelled query is + /// usually deterministically too slow, so inviting a prompt retry just + /// multiplies load on a strained database. `Retry-After` stays on the 503 + /// family, where a retry genuinely helps. + #[tokio::test] + async fn test_gateway_timeout_has_no_retry_after() { + let response = RestError::GatewayTimeout { + message: "statement cancelled".to_string(), + } + .into_response(); + assert_eq!(response.status(), StatusCode::GATEWAY_TIMEOUT); + assert!( + response + .headers() + .get(axum::http::header::RETRY_AFTER) + .is_none(), + "504 must not advise a retry deadline" + ); + } + #[tokio::test] async fn test_service_unavailable_response_carries_retry_after() { // The 503 must carry a Retry-After hint (delta-seconds) so clients and diff --git a/crates/rest/src/handlers/batch.rs b/crates/rest/src/handlers/batch.rs index 7f60b86d0..b040f79d0 100644 --- a/crates/rest/src/handlers/batch.rs +++ b/crates/rest/src/handlers/batch.rs @@ -1053,8 +1053,13 @@ fn transaction_error_response_parts(err: &TransactionError) -> (StatusCode, &'st "transient", "The transaction could not be completed and was rolled back.".to_string(), ), + // 504, not 500: the backend is healthy and deliberately stopped work + // that exceeded its time budget. Kept in step with + // `From for RestError`, so a transaction timeout + // reports the same status whether it surfaces through this bundle path + // or the single-resource one (issue #353). TransactionError::Timeout { timeout_ms } => ( - StatusCode::INTERNAL_SERVER_ERROR, + StatusCode::GATEWAY_TIMEOUT, "timeout", format!("Transaction timed out after {}ms", timeout_ms), ), @@ -1291,9 +1296,11 @@ mod tests { StatusCode::BAD_REQUEST, "processing", ), + // 504 since #353 — a backend that stopped over-budget work is not + // reporting a server defect. ( TransactionError::Timeout { timeout_ms: 1500 }, - StatusCode::INTERNAL_SERVER_ERROR, + StatusCode::GATEWAY_TIMEOUT, "timeout", ), (