Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
366 changes: 363 additions & 3 deletions crates/persistence/src/backends/mongodb/storage.rs

Large diffs are not rendered by default.

3 changes: 3 additions & 0 deletions crates/persistence/src/backends/postgres/backend.rs
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,9 @@ use crate::search::{
type StoredByTenant = Arc<RwLock<HashMap<String, Vec<SearchParameterDefinition>>>>;

/// PostgreSQL backend for FHIR resource storage.
///
/// Cheap to clone: a pool handle, shared registries and the config.
#[derive(Clone)]
pub struct PostgresBackend {
pool: Pool,
config: PostgresConfig,
Expand Down
111 changes: 92 additions & 19 deletions crates/persistence/src/backends/postgres/storage.rs
Original file line number Diff line number Diff line change
Expand Up @@ -2643,6 +2643,12 @@ impl PurgableStorage for PostgresBackend {

// Helper function to parse simple search parameters
// Supports basic formats like: identifier=X, _id=Y, name=Z
/// Why conditional criteria cannot be resolved inside a transaction when
/// search is offloaded to a secondary backend (#511, #859).
const OFFLOADED_CONDITIONAL_REFUSAL: &str = "conditional criteria cannot be resolved inside a \
transaction when search is offloaded to a secondary backend; submit the entry in a batch \
Bundle instead";

fn parse_simple_search_params(params: &str) -> Vec<(String, String)> {
params
.split('&')
Expand Down Expand Up @@ -2830,24 +2836,23 @@ impl PostgresBackend {
/// Buffered creates are flushed first, exactly as `read` does, so they are
/// visible too; a bundle that puts `ifNoneExist` on every entry therefore
/// forfeits create batching, which is the correct trade.
///
/// The string form `ifNoneExist` carries is parsed here and handed to the
/// typed [`ConditionalTransaction::find_matching`], so both criteria
/// forms run one search body (#859).
async fn find_matching_resources_in_tx(
&self,
tenant: &TenantContext,
tx: &mut crate::backends::postgres::transaction::PostgresTransaction,
resource_type: &str,
search_params_str: &str,
) -> StorageResult<Vec<StoredResource>> {
use crate::core::ConditionalTransaction;

let Some(query) = self.conditional_query(tenant, resource_type, search_params_str)? else {
return Ok(Vec::new());
};

tx.flush().await?;
let client = tx.client()?;
let result = self
.search_with_client(client, tenant, &query, None)
.await?;

Ok(result.resources.items)
tx.find_matching(resource_type, &query.parameters).await
}

/// Builds the search a conditional interaction's criteria describe, or
Expand Down Expand Up @@ -3097,6 +3102,14 @@ impl BundleProvider for PostgresBackend {
true
}

/// With search offloaded to a secondary backend the local index is empty
/// for every row, so an in-transaction search would always find nothing
/// and every conditional write would create the duplicate its criteria
/// exist to prevent.
fn supports_conditional_in_transaction(&self) -> bool {
!self.is_search_offloaded()
}

async fn process_transaction(
&self,
tenant: &TenantContext,
Expand All @@ -3117,6 +3130,25 @@ impl BundleProvider for PostgresBackend {
let mut results = Vec::with_capacity(entries.len());
let mut error_info: Option<(usize, String)> = None;

// URL-borne conditional entries (`PUT/DELETE [type]?[criteria]`)
// resolve against the transaction's starting view before any entry is
// written, and an overlap between resolved identities and the other
// entries fails the bundle (R4 §3.1.0.11.2; #859). Nothing is
// buffered yet, so no flush precedes these searches.
let targets = match crate::core::resolve_conditional_targets(
&mut tx,
&entries,
(!self.supports_conditional_in_transaction()).then_some(OFFLOADED_CONDITIONAL_REFUSAL),
)
.await
{
Ok(targets) => targets,
Err(e) => {
let _ = Box::new(tx).rollback().await;
return Err(e);
}
};

// `create` no longer sends its insert on the spot — the transaction
// batches consecutive creates and flushes them together, which is what
// takes a 1,632-entry import bundle from 3,264 statements to 26. A
Expand All @@ -3126,8 +3158,18 @@ impl BundleProvider for PostgresBackend {
// n-th `create` call back to the entry that made it.
let mut create_entry_index: Vec<usize> = Vec::with_capacity(entries.len());

// Build a map of fullUrl -> assigned reference for reference resolution
// Build a map of fullUrl -> assigned reference for reference resolution.
// A conditional entry that matched is known now, so `urn:uuid`
// references to it resolve regardless of entry order.
let mut reference_map: HashMap<String, String> = HashMap::new();
for target in targets.values() {
if let (Some(full_url), Some(identity)) = (
entries[target.entry_index].full_url.as_ref(),
target.identity(),
) {
reference_map.insert(full_url.clone(), identity);
}
}

// Whether any entry in this transaction writes a SearchParameter that
// affects this tenant's cached overlay (#787: transaction-bundle writes
Expand All @@ -3149,7 +3191,9 @@ impl BundleProvider for PostgresBackend {
}

let creates_before = tx.creates_seen();
let result = self.process_bundle_entry_tx(tenant, &mut tx, entry).await;
let result = self
.process_bundle_entry_tx(tenant, &mut tx, entry, targets.get(&idx))
.await;
for _ in creates_before..tx.creates_seen() {
create_entry_index.push(idx);
}
Expand Down Expand Up @@ -3187,17 +3231,22 @@ impl BundleProvider for PostgresBackend {
}) == Some("SearchParameter")
}
// Deleted: the emptied result carries no resource, so
// parse the type from the entry's URL instead.
204 => self
.parse_url(&entry.url)
.map(|(resource_type, _)| resource_type == "SearchParameter")
// take the type from the entry's URL instead.
204 => crate::core::conditional_resource_type(entry)
.map(|resource_type| resource_type == "SearchParameter")
.or_else(|| {
self.parse_url(&entry.url).ok().map(|(resource_type, _)| {
resource_type == "SearchParameter"
})
})
.unwrap_or(false),
_ => false,
};
}

// If this was a create (POST) and we have a fullUrl, record the mapping
if entry.method == BundleMethod::Post {
// A create (POST, or a conditional PUT that created) with a
// fullUrl records the assigned identity for later references.
if matches!(entry.method, BundleMethod::Post | BundleMethod::Put) {
if let Some(ref full_url) = entry.full_url {
if let Some(ref location) = entry_result.location {
let reference = location
Expand Down Expand Up @@ -3292,11 +3341,16 @@ fn attribute_entry_error(

impl PostgresBackend {
/// Process a single bundle entry within a transaction.
///
/// `target` is the pre-pass resolution of a URL-borne conditional entry
/// (#859): its `PUT` updates the match or creates, its `DELETE` deletes
/// the match or is a no-op `204`, without re-resolving the criteria.
async fn process_bundle_entry_tx(
&self,
tenant: &TenantContext,
tx: &mut super::transaction::PostgresTransaction,
entry: &BundleEntry,
target: Option<&crate::core::ConditionalTarget>,
) -> StorageResult<BundleEntryResult> {
use crate::core::transaction::Transaction;

Expand Down Expand Up @@ -3341,9 +3395,7 @@ impl PostgresBackend {
// Refuse the entry instead; the bundle rolls back (#511).
if self.is_search_offloaded() {
return Ok(crate::core::not_supported_entry(
"ifNoneExist cannot be resolved inside a transaction when search \
is offloaded to a secondary backend; submit the entry in a batch \
Bundle instead",
OFFLOADED_CONDITIONAL_REFUSAL,
));
}
let matches = self
Expand All @@ -3364,6 +3416,17 @@ impl PostgresBackend {
})
})?;

if let Some(target) = target {
return Ok(match &target.resolved {
Some(existing) => crate::core::conditional_update_entry(
tx.update(existing, resource).await?,
),
None => BundleEntryResult::created(
tx.create(&target.resource_type, resource).await?,
),
});
}

let (resource_type, id) = self.parse_url(&entry.url)?;

let existing = tx.read(&resource_type, &id).await?;
Expand Down Expand Up @@ -3396,6 +3459,16 @@ impl PostgresBackend {
}
}
BundleMethod::Delete => {
if let Some(target) = target {
return Ok(match &target.resolved {
Some(existing) => {
tx.delete(&target.resource_type, existing.id()).await?;
crate::core::conditional_delete_entry(existing)
}
None => BundleEntryResult::deleted(),
});
}

let (resource_type, id) = self.parse_url(&entry.url)?;

// Honor `ifMatch` on DELETE — previously ignored here, so a
Expand Down
57 changes: 49 additions & 8 deletions crates/persistence/src/backends/postgres/transaction.rs
Original file line number Diff line number Diff line change
Expand Up @@ -9,13 +9,15 @@ use deadpool_postgres::Client;
use helios_fhir::FhirVersion;
use serde_json::Value;

use crate::core::{Transaction, TransactionOptions, TransactionProvider};
use crate::core::{
ConditionalTransaction, Transaction, TransactionOptions, TransactionProvider, conditional_query,
};
use crate::error::{
BackendError, ConcurrencyError, ResourceError, StorageError, StorageResult, TransactionError,
};
use crate::search::SearchParameterExtractor;
use crate::tenant::{Operation, TenantContext};
use crate::types::StoredResource;
use crate::types::{SearchParameter, StoredResource};

use super::PostgresBackend;
use super::cached::{execute_cached, query_cached, query_opt_cached};
Expand Down Expand Up @@ -46,6 +48,10 @@ pub struct PostgresTransaction {
active: bool,
/// The tenant context for this transaction.
tenant: TenantContext,
/// The parent backend, for the transaction-scoped search
/// ([`ConditionalTransaction`]) to run the backend's own search body on
/// this transaction's client (#859).
backend: PostgresBackend,
/// Search parameter extractor for indexing resources.
search_extractor: Arc<SearchParameterExtractor>,
/// When true, search indexing is offloaded to a secondary backend.
Expand Down Expand Up @@ -126,15 +132,23 @@ impl std::fmt::Debug for PostgresTransaction {

impl PostgresTransaction {
/// Create a new transaction.
///
/// The extractor, the offload flag and the index layout are read off
/// `backend` rather than passed alongside it: the transaction holds the
/// backend for its conditional-URL search (#859), so taking them as
/// separate arguments would let a caller hand over settings that disagree
/// with the backend the transaction actually writes through.
async fn new(
client: Client,
tenant: TenantContext,
search_extractor: Arc<SearchParameterExtractor>,
search_offloaded: bool,
backend: PostgresBackend,
defer_search_indexing: bool,
fhir_version: FhirVersion,
index_layout: super::schema::IndexLayout,
) -> StorageResult<Self> {
let search_extractor = Arc::new(backend.tenant_extractor(tenant.tenant_id().as_str()));
let search_offloaded = backend.is_search_offloaded();
let index_layout = backend.index_layout();

// Start the transaction.
//
// `batch_execute` and not `execute`: `execute("BEGIN", &[])` takes the
Expand All @@ -156,6 +170,7 @@ impl PostgresTransaction {
client: Some(client),
active: true,
tenant,
backend,
search_extractor,
search_offloaded,
defer_search_indexing,
Expand Down Expand Up @@ -946,6 +961,34 @@ impl Drop for PostgresTransaction {
}
}

/// The transaction-scoped search surface (#859).
///
/// Runs the backend's search body on this transaction's client, so the match
/// set includes what earlier entries of the same bundle wrote (#511).
/// Buffered creates are flushed first, exactly as `read` does, so they are
/// visible too; a bundle that resolves criteria on every entry therefore
/// forfeits create batching, which is the correct trade.
#[async_trait]
impl ConditionalTransaction for PostgresTransaction {
async fn find_matching(
&mut self,
resource_type: &str,
criteria: &[SearchParameter],
) -> StorageResult<Vec<StoredResource>> {
if criteria.is_empty() {
return Ok(Vec::new());
}
let query = conditional_query(resource_type, criteria);
self.flush().await?;
let client = self.client()?;
let result = self
.backend
.search_with_client(client, &self.tenant, &query, None)
.await?;
Ok(result.resources.items)
}
}

#[async_trait]
impl TransactionProvider for PostgresBackend {
type Transaction = PostgresTransaction;
Expand All @@ -959,11 +1002,9 @@ impl TransactionProvider for PostgresBackend {
PostgresTransaction::new(
client,
tenant.clone(),
std::sync::Arc::new(self.tenant_extractor(tenant.tenant_id().as_str())),
self.is_search_offloaded(),
self.clone(),
options.defer_search_indexing,
options.fhir_version.unwrap_or(self.config().fhir_version),
self.index_layout(),
)
.await
}
Expand Down
5 changes: 5 additions & 0 deletions crates/persistence/src/backends/s3/bundle.rs
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,11 @@ impl BundleProvider for S3Backend {
false
}

/// No transaction, so nothing to resolve inside one.
fn supports_conditional_in_transaction(&self) -> bool {
false
}

async fn process_transaction(
&self,
_tenant: &TenantContext,
Expand Down
Loading
Loading