Skip to content

fix: backend statement timeouts answer 504, not 500 (#353) #439

New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

Open
mauripunzueta wants to merge 2 commits into
base: main
Choose a base branch
from
Open

fix: backend statement timeouts answer 504, not 500 (#353) #439

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8f8fccf
fix(persistence,rest): backend statement timeouts answer 504, not 500
mauripunzueta Jul 29, 2026
bd44ce7
refactor(persistence): classify driver errors with a method, not a cl…
mauripunzueta Jul 29, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 19 additions & 22 deletions crates/persistence/src/backends/mongodb/search_impl.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -74,11 +74,11 @@ async fn collect_documents(mut cursor: Cursor<Document>) -> StorageResult<Vec<Do
while cursor
.advance()
.await
.map_err(|e| internal_error(format!("Failed to advance MongoDB cursor: {}", e)))?
.or_query_error("Failed to advance MongoDB cursor")?
{
let doc = cursor.deserialize_current().map_err(|e| {
internal_error(format!("Failed to deserialize MongoDB document: {}", e))
})?;
let doc = cursor
.deserialize_current()
.or_query_error("Failed to deserialize MongoDB document")?;
docs.push(doc);
}
Ok(docs)
Expand Down Expand Up @@ -198,7 +198,7 @@ impl SearchProvider for MongoBackend {
let docs = collect_documents(
find_action
.await
.map_err(|e| internal_error(format!("Failed to execute MongoDB search: {}", e)))?,
.or_query_error("Failed to execute MongoDB search")?,
)
.await?;

Expand Down Expand Up @@ -319,7 +319,7 @@ impl SearchProvider for MongoBackend {
resources
.count_documents(filter)
.await
.map_err(|e| internal_error(format!("Failed to count MongoDB search results: {}", e)))
.or_query_error("Failed to count MongoDB search results")
}

fn search_param_registry(
Expand Down Expand Up @@ -584,7 +584,7 @@ impl MongoBackend {
let cursor = search_index
.aggregate(pipeline)
.await
.map_err(|e| internal_error(format!("Failed to aggregate contained search: {}", e)))?;
.or_query_error("Failed to aggregate contained search")?;
let docs = collect_documents(cursor).await?;

let mut out = Vec::new();
Expand Down Expand Up @@ -656,7 +656,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::<HashSet<_>>();
Expand Down Expand Up @@ -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::<HashSet<_>>();
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -1613,12 +1613,7 @@ impl RevincludeProvider for MongoBackend {
let matching_ids: Vec<String> = 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();
Expand All @@ -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 =
Expand Down
75 changes: 39 additions & 36 deletions crates/persistence/src/backends/mongodb/storage.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand All @@ -1557,7 +1560,7 @@ impl ResourceStorage for MongoBackend {
db.collection::<Document>(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)
}
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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")
}
}

Expand All @@ -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
Expand Down Expand Up @@ -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")
}
}

Expand All @@ -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
Expand Down Expand Up @@ -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")
}
}

Expand Down Expand Up @@ -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(),
Expand All @@ -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
Expand All @@ -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(())
}
Expand All @@ -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)
}
Expand Down Expand Up @@ -3861,7 +3864,7 @@ impl ReindexTarget for MongoBackend {
.collection::<Document>(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)
}
Expand Down
Loading
Loading