feat(rest,persistence): resolve conditional interactions in Bundle entries

crates/persistence/src/backends/mongodb/search_impl.rs

@@ -477,7 +477,7 @@ impl ConditionalStorage for MongoBackend {
             1 => {
                 let current = matches.into_iter().next().expect("single match must exist");
                 self.delete(tenant, resource_type, current.id()).await?;
-                Ok(ConditionalDeleteResult::Deleted)
+                Ok(ConditionalDeleteResult::Deleted(current))
             }
             n => Ok(ConditionalDeleteResult::MultipleMatches(n)),
         }

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

@@ -17,7 +17,8 @@ use crate::core::{
     BundleEntry, BundleEntryResult, BundleMethod, BundleProvider, BundleResult, BundleType,
     HistoryEntry, HistoryMethod, HistoryPage, HistoryParams, InstanceHistoryProvider,
     PurgableStorage, ResourceStorage, SettingsStore, SystemHistoryProvider, TypeHistoryProvider,
-    VersionedStorage, bundle_if_match_gate, if_match_field_satisfied, normalize_etag,
+    VersionedStorage, bundle_if_match_gate, bundle_if_none_exist_gate, if_match_field_satisfied,
+    normalize_etag,
 };
 use crate::error::{
     BackendError, ConcurrencyError, QueryErrorExt, ResourceError, StorageError, StorageResult,
@@ -2695,29 +2696,11 @@ impl MongoBackend {
                         )
                         .await?;
 
-                    match matches.len() {
-                        0 => {}
-                        1 => {
-                            return Ok(BundleEntryResult::ok(
-                                matches.into_iter().next().expect("single match must exist"),
-                            ));
-                        }
-                        n => {
-                            return Ok(BundleEntryResult::error(
-                                412,
-                                serde_json::json!({
-                                    "resourceType": "OperationOutcome",
-                                    "issue": [{
-                                        "severity": "error",
-                                        "code": "multiple-matches",
-                                        "diagnostics": format!(
-                                            "Conditional create matched {} resources",
-                                            n
-                                        )
-                                    }]
-                                }),
-                            ));
-                        }
+                    // Shared with the SQLite and PostgreSQL executors so all
+                    // three answer 200-with-location / 412 identically, and so
+                    // the matched id reaches the fullUrl reference map.
+                    if let Some(gated) = bundle_if_none_exist_gate(matches) {
+                        return Ok(gated);
                     }
                 }
 

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

@@ -195,31 +195,22 @@ fn open_range_needs_empty_guard(query: &SearchQuery) -> bool {
     })
 }
 
-#[async_trait]
-impl SearchProvider for PostgresBackend {
-    async fn search(
+impl PostgresBackend {
+    /// The body of [`SearchProvider::search`], run on a caller-supplied client.
+    ///
+    /// `SearchProvider::search` takes a fresh pooled client, which cannot see
+    /// rows an open transaction has written. A bundle entry that must resolve
+    /// `ifNoneExist` against what earlier entries in the same transaction wrote
+    /// runs this on the transaction's own client instead (#511). `total` is
+    /// computed by the caller so the count query's client is not held across
+    /// this one's await points.
+    pub(crate) async fn search_with_client(
         &self,
+        client: &deadpool_postgres::Client,
         tenant: &TenantContext,
         query: &SearchQuery,
+        total: Option<u64>,
     ) -> StorageResult<SearchResult> {
-        reject_contained_missing(query)?;
-
-        // `_contained` search uses a dedicated path (different index columns and
-        // heterogeneous result types); standard search handles `_contained=false`.
-        if query.contained != crate::types::ContainedMode::Off {
-            return self.search_contained(tenant, query).await;
-        }
-
-        // Populate Bundle.total only when the client asked for it
-        // (`_total=accurate|estimate`). Computed up-front so the count query's
-        // client is not held across the main query's await points.
-        let total = if query.wants_total() {
-            Some(self.search_count(tenant, query).await?)
-        } else {
-            None
-        };
-
-        let client = self.get_client().await?;
         let tenant_id = tenant.tenant_id().as_str();
         let resource_type = &query.resource_type;
 
@@ -406,7 +397,7 @@ impl SearchProvider for PostgresBackend {
             .map(|p| p.as_ref() as &(dyn tokio_postgres::types::ToSql + Sync))
             .collect();
         let rows = if statement_is_reusable(query) {
-            query_dyn_cached(&client, &sql, &param_refs).await
+            query_dyn_cached(client, &sql, &param_refs).await
         } else {
             client.query(&sql, &param_refs).await
         }
@@ -490,6 +481,35 @@ impl SearchProvider for PostgresBackend {
             scores: Default::default(),
         })
     }
+}
+
+#[async_trait]
+impl SearchProvider for PostgresBackend {
+    async fn search(
+        &self,
+        tenant: &TenantContext,
+        query: &SearchQuery,
+    ) -> StorageResult<SearchResult> {
+        reject_contained_missing(query)?;
+
+        // `_contained` search uses a dedicated path (different index columns and
+        // heterogeneous result types); standard search handles `_contained=false`.
+        if query.contained != crate::types::ContainedMode::Off {
+            return self.search_contained(tenant, query).await;
+        }
+
+        // Populate Bundle.total only when the client asked for it
+        // (`_total=accurate|estimate`). Computed up-front so the count query's
+        // client is not held across the main query's await points.
+        let total = if query.wants_total() {
+            Some(self.search_count(tenant, query).await?)
+        } else {
+            None
+        };
+
+        let client = self.get_client().await?;
+        self.search_with_client(&client, tenant, query, total).await
+    }
 
     async fn search_count(
         &self,

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

@@ -2753,7 +2753,7 @@ impl ConditionalStorage for PostgresBackend {
                 // Exactly one match - delete it
                 let existing = matches.into_iter().next().unwrap();
                 self.delete(tenant, resource_type, existing.id()).await?;
-                Ok(ConditionalDeleteResult::Deleted)
+                Ok(ConditionalDeleteResult::Deleted(existing))
             }
             n => {
                 // Multiple matches - error condition
@@ -2815,29 +2815,66 @@ impl PostgresBackend {
         resource_type: &str,
         search_params_str: &str,
     ) -> StorageResult<Vec<StoredResource>> {
+        let Some(query) = self.conditional_query(tenant, resource_type, search_params_str)? else {
+            return Ok(Vec::new());
+        };
+
+        // Use the SearchProvider implementation
+        let result = <Self as SearchProvider>::search(self, tenant, &query).await?;
+
+        Ok(result.resources.items)
+    }
+
+    /// Resolves conditional criteria on the transaction's own 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 puts `ifNoneExist` on every entry therefore
+    /// forfeits create batching, which is the correct trade.
+    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>> {
+        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)
+    }
+
+    /// Builds the search a conditional interaction's criteria describe, or
+    /// `None` when the criteria are empty — matching everything would be the
+    /// literal reading, but no conditional interaction means that.
+    fn conditional_query(
+        &self,
+        tenant: &TenantContext,
+        resource_type: &str,
+        search_params_str: &str,
+    ) -> StorageResult<Option<SearchQuery>> {
         // Parse search parameters into (name, value) pairs
         let parsed_params = parse_simple_search_params(search_params_str);
 
         if parsed_params.is_empty() {
-            // No search params means match all - but for conditional ops this is unusual
-            return Ok(Vec::new());
+            return Ok(None);
         }
 
         // Build SearchParameter objects by looking up types from the registry
         let search_params = self.build_search_parameters(tenant, resource_type, &parsed_params)?;
 
-        // Build a SearchQuery
-        let query = SearchQuery {
+        Ok(Some(SearchQuery {
             resource_type: resource_type.to_string(),
             parameters: search_params,
             count: Some(1000),
             ..Default::default()
-        };
-
-        // Use the SearchProvider implementation
-        let result = <Self as SearchProvider>::search(self, tenant, &query).await?;
-
-        Ok(result.resources.items)
+        }))
     }
 
     /// Builds SearchParameter objects from parsed (name, value) pairs.
@@ -3103,7 +3140,7 @@ impl BundleProvider for PostgresBackend {
             }
 
             let creates_before = tx.creates_seen();
-            let result = self.process_bundle_entry_tx(&mut tx, entry).await;
+            let result = self.process_bundle_entry_tx(tenant, &mut tx, entry).await;
             for _ in creates_before..tx.creates_seen() {
                 create_entry_index.push(idx);
             }
@@ -3202,6 +3239,7 @@ impl PostgresBackend {
     /// Process a single bundle entry within a transaction.
     async fn process_bundle_entry_tx(
         &self,
+        tenant: &TenantContext,
         tx: &mut super::transaction::PostgresTransaction,
         entry: &BundleEntry,
     ) -> StorageResult<BundleEntryResult> {
@@ -3240,6 +3278,27 @@ impl PostgresBackend {
                         )
                     })?;
 
+                if let Some(criteria) = entry.if_none_exist.as_deref() {
+                    // 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 this arm would
+                    // create the duplicate `ifNoneExist` exists to prevent.
+                    // 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",
+                        ));
+                    }
+                    let matches = self
+                        .find_matching_resources_in_tx(tenant, tx, &resource_type, criteria)
+                        .await?;
+                    if let Some(gated) = crate::core::bundle_if_none_exist_gate(matches) {
+                        return Ok(gated);
+                    }
+                }
+
                 let created = tx.create(&resource_type, resource).await?;
                 Ok(BundleEntryResult::created(created))
             }

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

@@ -163,7 +163,7 @@ impl PostgresTransaction {
         })
     }
 
-    fn client(&self) -> StorageResult<&Client> {
+    pub(crate) fn client(&self) -> StorageResult<&Client> {
         self.client
             .as_ref()
             .ok_or_else(|| StorageError::Transaction(TransactionError::InvalidTransaction))

crates/persistence/src/backends/sqlite/search_impl.rs

@@ -83,31 +83,24 @@ fn bind_cursor_value(
     Ok(())
 }
 
-#[async_trait]
-impl SearchProvider for SqliteBackend {
-    async fn search(
+impl SqliteBackend {
+    /// The body of [`SearchProvider::search`], run on a caller-supplied
+    /// connection.
+    ///
+    /// `SearchProvider::search` takes a fresh pooled connection, which under
+    /// `BEGIN IMMEDIATE` sees pre-transaction state only. A bundle entry that
+    /// must resolve `ifNoneExist` against what earlier entries in the same
+    /// transaction wrote runs this on the transaction's own connection instead
+    /// (via `SqliteTransaction::with_connection`, #511). `total` is computed by
+    /// the caller because `search_count` is async and this is not.
+    #[allow(clippy::type_complexity)]
+    pub(crate) fn search_with_connection(
         &self,
+        conn: &rusqlite::Connection,
         tenant: &TenantContext,
         query: &SearchQuery,
+        total: Option<u64>,
     ) -> StorageResult<SearchResult> {
-        reject_contained_missing(query)?;
-
-        // `_contained` search uses a dedicated path (different index columns and
-        // heterogeneous result types); standard search handles `_contained=false`.
-        if query.contained != crate::types::ContainedMode::Off {
-            return self.search_contained(tenant, query).await;
-        }
-
-        // Populate Bundle.total only when the client asked for it
-        // (`_total=accurate|estimate`). Computed up-front, before acquiring the
-        // (non-Send) connection, so it is not held across this await.
-        let total = if query.wants_total() {
-            Some(self.search_count(tenant, query).await?)
-        } else {
-            None
-        };
-
-        let conn = self.get_connection()?;
         let tenant_id = tenant.tenant_id().as_str();
         let resource_type = &query.resource_type;
 
@@ -352,6 +345,35 @@ impl SearchProvider for SqliteBackend {
             scores: Default::default(),
         })
     }
+}
+
+#[async_trait]
+impl SearchProvider for SqliteBackend {
+    async fn search(
+        &self,
+        tenant: &TenantContext,
+        query: &SearchQuery,
+    ) -> StorageResult<SearchResult> {
+        reject_contained_missing(query)?;
+
+        // `_contained` search uses a dedicated path (different index columns and
+        // heterogeneous result types); standard search handles `_contained=false`.
+        if query.contained != crate::types::ContainedMode::Off {
+            return self.search_contained(tenant, query).await;
+        }
+
+        // Populate Bundle.total only when the client asked for it
+        // (`_total=accurate|estimate`). Computed up-front, before acquiring the
+        // (non-Send) connection, so it is not held across this await.
+        let total = if query.wants_total() {
+            Some(self.search_count(tenant, query).await?)
+        } else {
+            None
+        };
+
+        let conn = self.get_connection()?;
+        self.search_with_connection(&conn, tenant, query, total)
+    }
 
     async fn search_count(
         &self,