feat(rest,persistence): resolve conditional URL entries inside transaction bundles

crates/persistence/src/backends/postgres/backend.rs

@@ -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,

crates/persistence/src/backends/postgres/storage.rs

@@ -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('&')
@@ -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
@@ -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,
@@ -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
@@ -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
@@ -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);
             }
@@ -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
@@ -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;
 
@@ -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
@@ -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?;
@@ -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

crates/persistence/src/backends/postgres/transaction.rs

@@ -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};
@@ -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.
@@ -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
@@ -156,6 +170,7 @@ impl PostgresTransaction {
             client: Some(client),
             active: true,
             tenant,
+            backend,
             search_extractor,
             search_offloaded,
             defer_search_indexing,
@@ -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;
@@ -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
     }

crates/persistence/src/backends/s3/bundle.rs

@@ -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,