diff --git a/.gitignore b/.gitignore index 33086a52a..384df4c7b 100644 --- a/.gitignore +++ b/.gitignore @@ -33,3 +33,8 @@ __pycache__/ # purpose: they are personal working material, not project documentation, and # they should never reach a PR. `tmp/` is the intended home for them. /tmp/ + +# Cluster-capable-state planning docs and draft-issue scratch files under +# docs/ are local working notes too (discussion #223); they never ship. +/docs/cluster-*.md +/docs/draft-issues-*.md diff --git a/crates/persistence/src/backends/mongodb/storage.rs b/crates/persistence/src/backends/mongodb/storage.rs index c56d1f7e5..c453d3dae 100644 --- a/crates/persistence/src/backends/mongodb/storage.rs +++ b/crates/persistence/src/backends/mongodb/storage.rs @@ -721,34 +721,60 @@ impl ResourceStorage for MongoBackend { resource: Value, fhir_version: FhirVersion, ) -> StorageResult<(StoredResource, bool)> { - // Check if exists - match self.read(tenant, resource_type, id).await { - // Update existing (preserves original FHIR version) - Ok(Some(current)) => { - let updated = self.update(tenant, ¤t, resource).await?; - Ok((updated, false)) - } - // Create new with specific ID - Ok(None) => { - let mut resource = resource; - if let Some(obj) = resource.as_object_mut() { - obj.insert("id".to_string(), Value::String(id.to_string())); + // Every branch is a check-then-act on the `read`, and every write is + // a compare-and-swap, so a race with another writer (on this or + // another instance) surfaces as a typed error that is never a + // legitimate answer from *this* method — it decides existence itself. + // Re-read and retry a bounded number of times; see the Postgres + // backend's `create_or_update` for the full reasoning. + const MAX_CAS_RETRIES: usize = 3; + let mut attempts = 0; + let mut resource = resource; + loop { + let body = if attempts < MAX_CAS_RETRIES { + resource.clone() + } else { + std::mem::take(&mut resource) + }; + + let result = match self.read(tenant, resource_type, id).await { + // Update existing (preserves original FHIR version) + Ok(Some(current)) => self + .update(tenant, ¤t, body) + .await + .map(|updated| (updated, false)), + // Create new with specific ID + Ok(None) => { + let mut body = body; + if let Some(obj) = body.as_object_mut() { + obj.insert("id".to_string(), Value::String(id.to_string())); + } + self.create(tenant, resource_type, body, fhir_version) + .await + .map(|created| (created, true)) } - let created = self - .create(tenant, resource_type, resource, fhir_version) - .await?; - Ok((created, true)) - } - // A deleted resource is brought back to life by a subsequent update - // (FHIR http.html#delete), continuing the existing version chain - // rather than being rejected with `Gone`. - Err(StorageError::Resource(ResourceError::Gone { .. })) => { - let restored = self - .restore_deleted(tenant, resource_type, id, resource) - .await?; - Ok((restored, true)) + // A deleted resource is brought back to life by a subsequent + // update (FHIR http.html#delete), continuing the existing + // version chain rather than being rejected with `Gone`. + Err(StorageError::Resource(ResourceError::Gone { .. })) => self + .restore_deleted(tenant, resource_type, id, body) + .await + .map(|restored| (restored, true)), + Err(e) => Err(e), + }; + + match result { + Err( + StorageError::Concurrency(ConcurrencyError::VersionConflict { .. }) + | StorageError::Resource( + ResourceError::NotFound { .. } | ResourceError::AlreadyExists { .. }, + ), + ) if attempts < MAX_CAS_RETRIES => { + attempts += 1; + continue; + } + other => return other, } - Err(e) => Err(e), } } @@ -1654,9 +1680,11 @@ impl MongoBackend { /// record keeps its version, the restore gets the next one) and keeps the /// FHIR version the resource was originally stored under. /// - /// Returns `NotFound` if no deleted document is present — the caller has - /// already established one exists, so that only happens under a concurrent - /// write. + /// The write is a compare-and-swap on the tombstone's version, so a lost + /// race reports `VersionConflict` (with the version actually stored) and + /// `create_or_update` retries from a fresh read. Returns `NotFound` only + /// if no document is present at all — the caller has already established + /// a tombstone exists, so either only happens under a concurrent write. async fn restore_deleted( &self, tenant: &TenantContext, @@ -1756,10 +1784,41 @@ impl MongoBackend { }; if update_result.matched_count == 0 { - return Err(StorageError::Resource(ResourceError::NotFound { - resource_type: resource_type.to_string(), - id: id.to_string(), - })); + // The tombstone the read observed is gone. A concurrent PUT + // restored it (a live document now) or restored and re-deleted it + // (a newer tombstone) — the same fact for the caller, so the + // re-read does not filter on `is_deleted`: any document is a + // conflict `create_or_update` retries from a fresh read; none at + // all is a hard delete. + let current_filter = doc! { + "tenant_id": tenant_id, + "resource_type": resource_type, + "id": id, + }; + let actual = if let Some(active_session) = session.as_mut() { + resources + .find_one(current_filter) + .session(active_session) + .await + } else { + resources.find_one(current_filter).await + } + .map_err(|e| internal_error(format!("Failed to get current version: {}", e)))?; + + return match actual { + Some(doc) => Err(StorageError::Concurrency( + ConcurrencyError::VersionConflict { + resource_type: resource_type.to_string(), + id: id.to_string(), + expected_version: deleted_version, + actual_version: doc.get_str("version_id").unwrap_or("").to_string(), + }, + )), + None => Err(StorageError::Resource(ResourceError::NotFound { + resource_type: resource_type.to_string(), + id: id.to_string(), + })), + }; } let history_doc = doc! { diff --git a/crates/persistence/src/backends/postgres/storage.rs b/crates/persistence/src/backends/postgres/storage.rs index fc09cd028..ee707ed9a 100644 --- a/crates/persistence/src/backends/postgres/storage.rs +++ b/crates/persistence/src/backends/postgres/storage.rs @@ -303,34 +303,68 @@ impl ResourceStorage for PostgresBackend { resource: Value, fhir_version: FhirVersion, ) -> StorageResult<(StoredResource, bool)> { - // Check if exists - match self.read(tenant, resource_type, id).await { - // Update existing (preserves original FHIR version) - Ok(Some(current)) => { - let updated = self.update(tenant, ¤t, resource).await?; - Ok((updated, false)) - } - // Create new with specific ID - Ok(None) => { - let mut resource = resource; - if let Some(obj) = resource.as_object_mut() { - obj.insert("id".to_string(), Value::String(id.to_string())); + // Unconditional PUT is last-writer-wins at the API surface, but every + // branch below is a check-then-act on the `read`: the row can be + // created, bumped, deleted or restored by another writer (on this or + // another instance) between the read and the write. Each write is a + // compare-and-swap, so a lost race surfaces as a typed error rather + // than corrupting anything — and because *this* method decides + // existence itself, none of those errors is a legitimate answer from + // it: `VersionConflict` from `update`/`restore_deleted`, + // `AlreadyExists` from `create`, and `NotFound` from `update` (the row + // was deleted underneath it) all mean "re-read and go again". A + // bounded number of retries absorbs the race; the last attempt's + // error propagates so a pathological caller cannot spin forever. + const MAX_CAS_RETRIES: usize = 3; + let mut attempts = 0; + let mut resource = resource; + loop { + // Each attempt consumes a body; keep a copy only while a retry is + // still possible so the final attempt moves rather than clones. + let body = if attempts < MAX_CAS_RETRIES { + resource.clone() + } else { + std::mem::take(&mut resource) + }; + + let result = match self.read(tenant, resource_type, id).await { + // Update existing (preserves original FHIR version) + Ok(Some(current)) => self + .update(tenant, ¤t, body) + .await + .map(|updated| (updated, false)), + // Create new with specific ID + Ok(None) => { + let mut body = body; + if let Some(obj) = body.as_object_mut() { + obj.insert("id".to_string(), Value::String(id.to_string())); + } + self.create(tenant, resource_type, body, fhir_version) + .await + .map(|created| (created, true)) } - let created = self - .create(tenant, resource_type, resource, fhir_version) - .await?; - Ok((created, true)) - } - // A deleted resource is brought back to life by a subsequent update - // (FHIR http.html#delete), continuing the existing version chain - // rather than being rejected with `Gone`. - Err(StorageError::Resource(ResourceError::Gone { .. })) => { - let restored = self - .restore_deleted(tenant, resource_type, id, resource) - .await?; - Ok((restored, true)) + // A deleted resource is brought back to life by a subsequent + // update (FHIR http.html#delete), continuing the existing + // version chain rather than being rejected with `Gone`. + Err(StorageError::Resource(ResourceError::Gone { .. })) => self + .restore_deleted(tenant, resource_type, id, body) + .await + .map(|restored| (restored, true)), + Err(e) => Err(e), + }; + + match result { + Err( + StorageError::Concurrency(ConcurrencyError::VersionConflict { .. }) + | StorageError::Resource( + ResourceError::NotFound { .. } | ResourceError::AlreadyExists { .. }, + ), + ) if attempts < MAX_CAS_RETRIES => { + attempts += 1; + continue; + } + other => return other, } - Err(e) => Err(e), } } @@ -1088,8 +1122,23 @@ impl PostgresBackend { /// record keeps its version, the restore gets the next one) and keeps the /// FHIR version the resource was originally stored under. /// - /// Returns `NotFound` if no deleted row is present — the caller has already - /// established one exists, so that only happens under a concurrent write. + /// The write is a version-guarded compare-and-swap against the tombstone + /// the caller's read observed, and it lands the resource row and its + /// history row in one statement (one implicit transaction) — the same + /// shape as `update`, and for the reason `delete` spells out: a history + /// row computed from a stale read can collide with one a concurrent + /// writer already inserted. Two PUTs racing onto one tombstone used to do + /// exactly that — both computed the same next version, the second history + /// insert hit the primary key, and the live row and history disagreed on + /// the body. Unlike `delete`, the restore keeps its read instead of + /// deriving the version in SQL: the read is what lets a lost race report + /// a truthful `expected_version` (the contract S3's restore already has), + /// and `create_or_update` turns that conflict into a fresh read-and-retry. + /// + /// Returns `NotFound` if no deleted row is present and `VersionConflict` + /// if the tombstone moved between the read and the write — the caller has + /// already established a tombstone exists, so either only happens under a + /// concurrent write. async fn restore_deleted( &self, tenant: &TenantContext, @@ -1135,13 +1184,23 @@ impl PostgresBackend { } let now = Utc::now(); - let is_deleted = false; - execute_cached( + // The tombstone's version is the CAS guard (`$7`): the UPDATE matches + // only the row the read observed, still deleted and still at that + // version, and the history row is fed from what it returned — so both + // land or neither does, and a stale read can never mint a duplicate + // version. + let restored = execute_cached( &client, - "UPDATE resources - SET version_id = $1, data = $2, last_updated = $3, is_deleted = FALSE, deleted_at = NULL - WHERE tenant_id = $4 AND resource_type = $5 AND id = $6", + "WITH restored AS ( + UPDATE resources + SET version_id = $1, data = $2, last_updated = $3, is_deleted = FALSE, deleted_at = NULL + WHERE tenant_id = $4 AND resource_type = $5 AND id = $6 + AND is_deleted = TRUE AND version_id = $7 + RETURNING tenant_id, resource_type, id, version_id, data, last_updated, is_deleted, fhir_version + ) + INSERT INTO resource_history (tenant_id, resource_type, id, version_id, data, last_updated, is_deleted, fhir_version) + SELECT tenant_id, resource_type, id, version_id, data, last_updated, is_deleted, fhir_version FROM restored", &[ &new_version_str, &resource, @@ -1149,19 +1208,43 @@ impl PostgresBackend { &tenant_id, &resource_type, &id, + &deleted_version, ], ) .await .map_err(|e| internal_error(format!("Failed to restore resource: {}", e)))?; - execute_cached( - &client, - "INSERT INTO resource_history (tenant_id, resource_type, id, version_id, data, last_updated, is_deleted, fhir_version) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8)", - &[&tenant_id, &resource_type, &id, &new_version_str, &resource, &now, &is_deleted, &fhir_version_str], - ) - .await - .map_err(|e| internal_error(format!("Failed to insert restore history: {}", e)))?; + if restored == 0 { + // Zero rows: the tombstone the read observed is gone. Whether a + // concurrent PUT restored it (a live row now) or restored and + // re-deleted it (a newer tombstone) is the same fact for the + // caller — so, unlike `update`'s re-select, this one does not + // filter on `is_deleted`. Any row is a conflict the caller can + // retry from a fresh read; no row at all is a hard delete. + let actual = client + .query_opt( + "SELECT version_id FROM resources + WHERE tenant_id = $1 AND resource_type = $2 AND id = $3", + &[&tenant_id, &resource_type, &id], + ) + .await + .map_err(|e| internal_error(format!("Failed to get current version: {}", e)))?; + + return match actual { + Some(row) => Err(StorageError::Concurrency( + ConcurrencyError::VersionConflict { + resource_type: resource_type.to_string(), + id: id.to_string(), + expected_version: deleted_version, + actual_version: row.get::<_, String>(0), + }, + )), + None => Err(StorageError::Resource(ResourceError::NotFound { + resource_type: resource_type.to_string(), + id: id.to_string(), + })), + }; + } // The delete dropped the search index entries; rebuild them for the // resource that is live again. As in `update`, `Replace` folds the diff --git a/crates/persistence/src/core/storage.rs b/crates/persistence/src/core/storage.rs index 9c68d381b..3ec9d5b07 100644 --- a/crates/persistence/src/core/storage.rs +++ b/crates/persistence/src/core/storage.rs @@ -410,6 +410,25 @@ pub trait ResourceStorage: Send + Sync { /// /// A tuple of (StoredResource, created: bool) where created indicates /// whether a new resource was created (true) or an existing one updated (false). + /// Restoring a soft-deleted resource counts as created. + /// + /// # Concurrency + /// + /// Unconditional PUT is last-writer-wins: two callers racing on one id — + /// on one instance or across instances sharing the store — must both + /// succeed, with distinct versions and a complete history. Implementations + /// decide existence with a read, write with a version-guarded + /// compare-and-swap, and retry a bounded number of times when the read + /// went stale (`VersionConflict`, `AlreadyExists`, or `NotFound` from the + /// write), so none of those errors reaches the caller for a race the + /// caller did not ask to observe. + /// + /// # Errors + /// + /// * `StorageError::Tenant` - If the tenant lacks create/update permission + /// * `StorageError::Validation` - If the resource content is invalid + /// * `StorageError::Concurrency(VersionConflict)` - Only after the bounded + /// retries are exhausted under sustained contention async fn create_or_update( &self, tenant: &TenantContext, diff --git a/crates/persistence/tests/common/cluster_harness.rs b/crates/persistence/tests/common/cluster_harness.rs new file mode 100644 index 000000000..8da8254d9 --- /dev/null +++ b/crates/persistence/tests/common/cluster_harness.rs @@ -0,0 +1,115 @@ +//! T2 cluster-harness helpers. +//! +//! A cluster bug is a multi-observer bug, and the faithful single-process +//! simulation of two `hfs` instances is two *freshly constructed* backend +//! handles that share nothing but the backing store (same connection +//! parameters) — never a cloned `Arc`, which would share an in-process heap +//! and prove nothing (the cloned-`Arc` anti-pattern). +//! +//! These helpers encode the mechanics every T2 cluster suite reuses: fresh +//! two-handle construction, barrier-synchronized racing, and the +//! definition-of-done assertion rows (visibility / isolation / exclusivity). +//! Calibrated against the already-cluster-safe bulk-export job store +//! (`postgres_integration_cluster_bulk_export_*` in `postgres_tests.rs`), so +//! the harness is proven before any product change relies on it; later +//! cluster-capable subsystems point the same helpers at their own stores. + +// This file is `#[path]`-included by several test binaries, and each uses +// only the helpers its subsystem needs; an unused helper in one binary is not +// dead code, so the per-binary dead-code lint is silenced here rather than +// per item. +#![allow(dead_code)] + +use std::future::Future; +use std::sync::Arc; + +/// Two independently constructed backend handles over one shared store. +/// +/// `a` and `b` play the roles of "instance A" and "instance B". They must be +/// separate constructions — the factory in [`two_handles`] is invoked twice — +/// so the only thing they can possibly share is the backing store itself. +pub struct ClusterHandles { + pub a: B, + pub b: B, +} + +/// Builds two fresh handles by invoking `factory` twice. +/// +/// The factory is the backend's normal constructor pointed at the shared +/// store (e.g. `create_backend()` against the shared Postgres container). +pub async fn two_handles(factory: F) -> ClusterHandles +where + F: Fn() -> Fut, + Fut: Future, +{ + ClusterHandles { + a: factory().await, + b: factory().await, + } +} + +/// Runs two futures on separate tasks released simultaneously by a barrier, +/// so both hit the shared store as close to concurrently as the runtime +/// allows. +/// +/// Each future must own everything it touches (`Send + 'static`); move the +/// handle into an `async move` block and re-observe afterwards through a +/// fresh handle. +pub async fn race2( + fa: impl Future + Send + 'static, + fb: impl Future + Send + 'static, +) -> (T, U) +where + T: Send + 'static, + U: Send + 'static, +{ + let barrier = Arc::new(tokio::sync::Barrier::new(2)); + let barrier_a = Arc::clone(&barrier); + let barrier_b = barrier; + let task_a = tokio::spawn(async move { + barrier_a.wait().await; + fa.await + }); + let task_b = tokio::spawn(async move { + barrier_b.wait().await; + fb.await + }); + ( + task_a.await.expect("racing task A panicked"), + task_b.await.expect("racing task B panicked"), + ) +} + +/// Exclusivity row: of two racing claimants, exactly one may win. +/// +/// Both winning is the cluster bug (double execution / double redeem); +/// neither winning means the race setup is wrong — also a failure, so the +/// test can't silently pass without exercising the claim. +pub fn assert_exactly_one(a: &Option, b: &Option, what: &str) { + match (a.is_some(), b.is_some()) { + (true, false) | (false, true) => {} + (true, true) => panic!("exclusivity violated: both handles won {what}"), + (false, false) => { + panic!("exclusivity check inconclusive: neither handle won {what}") + } + } +} + +/// Visibility row: state created via handle A must be observable via handle +/// B (same tenant). Returns the observed value for follow-on assertions. +pub fn assert_visible(got: Option, what: &str) -> T { + match got { + Some(value) => value, + None => panic!("visibility violated: {what} is not observable via the second handle"), + } +} + +/// Isolation row (mandatory on every suite): state created under one tenant +/// must not be observable under another — the observer sees nothing, exactly +/// as if the state did not exist. +pub fn assert_wrong_tenant_hidden(got: Option, what: &str) { + assert!( + got.is_none(), + "tenant isolation violated: {what} is observable under the wrong tenant" + ); +} diff --git a/crates/persistence/tests/mongodb_tests.rs b/crates/persistence/tests/mongodb_tests.rs index 572bcaf1f..8d317dfa5 100644 --- a/crates/persistence/tests/mongodb_tests.rs +++ b/crates/persistence/tests/mongodb_tests.rs @@ -39,6 +39,12 @@ use mongodb::Client; use mongodb::bson::{Document, doc}; use serde_json::json; +/// The T2 cluster harness (two independently constructed backends over one +/// shared store); path-included rather than placed in a `mod common`, which no +/// test target declares. +#[path = "common/cluster_harness.rs"] +mod cluster_harness; + const MONGODB_MAX_DATABASE_NAME_LEN: usize = 63; const MONGODB_TEST_DB_PREFIX: &str = "hfs_phase2_mongo_"; @@ -4515,3 +4521,206 @@ async fn mongodb_integration_export_until_is_inclusive() { "a resource exactly on the bound is included" ); } + +// ============================================================================ +// T2 cluster suite — F5 resource version-id race (restore path). +// ============================================================================ +// +// Two independently constructed `MongoBackend`s on ONE database play +// "instance A" and "instance B"; `create_backend` mints a fresh database per +// call, so the second handle is built from the first's config instead. Mongo's +// restore was already a version-guarded CAS; what these rows pin is that a +// lost race is *retried* by `create_or_update` rather than surfacing a 404 on +// a PUT, and that the loser reports `VersionConflict` with the stored version. + +use cluster_harness as harness; + +/// Two fresh handles, one database. `None` when Mongo is unavailable (the +/// suite's usual skip convention). +async fn two_mongo_handles(test_name: &str) -> Option<(MongoBackend, MongoBackend)> { + let connection_string = shared_mongo::connection_string().await?; + let a = build_backend(MongoBackendConfig { + connection_string: connection_string.clone(), + database_name: build_test_database_name(test_name), + max_connections: 1, + ..Default::default() + }) + .await?; + let b = build_backend(MongoBackendConfig { + connection_string, + database_name: a.config().database_name.clone(), + max_connections: 1, + ..Default::default() + }) + .await?; + Some((a, b)) +} + +/// Asserts a resource's history holds every version exactly once, +/// contiguous from 1 — no duplicate version_ids, no lost history rows. +async fn assert_mongo_history_versions( + backend: &MongoBackend, + tenant: &TenantContext, + resource_type: &str, + id: &str, + expected_len: u64, +) { + let history = backend + .history_instance( + tenant, + resource_type, + id, + &HistoryParams::new().include_deleted(true), + ) + .await + .unwrap(); + let mut versions: Vec = history + .items + .iter() + .map(|e| e.resource.version_id().parse().unwrap()) + .collect(); + versions.sort_unstable(); + assert_eq!( + versions, + (1..=expected_len).collect::>(), + "history must hold every version exactly once" + ); +} + +/// Creates a Patient and soft-deletes it, returning its id: version 1 is the +/// create, version 2 the tombstone. +async fn seed_mongo_tombstone(backend: &MongoBackend, tenant: &TenantContext) -> String { + let created = backend + .create( + tenant, + "Patient", + json!({"resourceType": "Patient", "name": [{"family": "Seed"}]}), + FhirVersion::default(), + ) + .await + .unwrap(); + backend + .delete(tenant, "Patient", created.id()) + .await + .unwrap(); + created.id().to_string() +} + +/// DoD rows (F5, restore path): two instances race unconditional PUTs onto +/// one soft-deleted id; both succeed (winner restores, loser converges through +/// the update arm), history is contiguous 1..=4 and the newest entry carries +/// the live body. Without the retry the loser's CAS miss surfaced as a 404. +#[tokio::test] +async fn mongodb_integration_cluster_restore_deleted_race_keeps_history_coherent() { + let Some((a, b)) = two_mongo_handles("cluster_restore_race").await else { + return; + }; + let (a, b) = (Arc::new(a), Arc::new(b)); + let tenant = create_tenant("cluster-f5-restore"); + + for round in 0..5 { + let id = seed_mongo_tombstone(&a, &tenant).await; + + let (ra, rb) = (Arc::clone(&a), Arc::clone(&b)); + let (tenant_a, tenant_b) = (tenant.clone(), tenant.clone()); + let (id_a, id_b) = (id.clone(), id.clone()); + let (res_a, res_b) = harness::race2( + async move { + ra.create_or_update( + &tenant_a, + "Patient", + &id_a, + json!({"resourceType": "Patient", "name": [{"family": "FromA"}]}), + FhirVersion::default(), + ) + .await + }, + async move { + rb.create_or_update( + &tenant_b, + "Patient", + &id_b, + json!({"resourceType": "Patient", "name": [{"family": "FromB"}]}), + FhirVersion::default(), + ) + .await + }, + ) + .await; + + let (stored_a, created_a) = + res_a.unwrap_or_else(|e| panic!("round {round}: PUT via A must succeed: {e:?}")); + let (stored_b, created_b) = + res_b.unwrap_or_else(|e| panic!("round {round}: PUT via B must succeed: {e:?}")); + + let restored = |flag: bool| flag.then_some(()); + harness::assert_exactly_one( + &restored(created_a), + &restored(created_b), + &format!("round {round}: the restore of one tombstone"), + ); + assert_ne!( + stored_a.version_id(), + stored_b.version_id(), + "round {round}: racing writers must never assign the same version" + ); + + let current = a.read(&tenant, "Patient", &id).await.unwrap().unwrap(); + assert_eq!(current.version_id(), "4", "round {round}"); + assert!(!current.is_deleted(), "round {round}"); + assert_mongo_history_versions(&b, &tenant, "Patient", &id, 4).await; + + let history = b + .history_instance( + &tenant, + "Patient", + &id, + &HistoryParams::new().include_deleted(true), + ) + .await + .unwrap(); + let newest = history + .items + .iter() + .max_by_key(|e| e.resource.version_id().parse::().unwrap()) + .unwrap(); + assert_eq!( + newest.resource.content(), + current.content(), + "round {round}: newest history entry must carry the live body" + ); + } +} + +/// DoD row (F5, restore path): isolation — a PUT onto the same id under +/// another tenant creates a fresh version-1 resource there and leaves the +/// first tenant's tombstone untouched. +#[tokio::test] +async fn mongodb_integration_cluster_restore_deleted_wrong_tenant_creates_not_restores() { + let Some((a, b)) = two_mongo_handles("cluster_restore_iso").await else { + return; + }; + let tenant = create_tenant("cluster-f5-restore-iso"); + let id = seed_mongo_tombstone(&a, &tenant).await; + + let other = create_tenant("cluster-f5-restore-other"); + let (stored, created) = b + .create_or_update( + &other, + "Patient", + &id, + json!({"resourceType": "Patient", "name": [{"family": "Other"}]}), + FhirVersion::default(), + ) + .await + .unwrap(); + assert!(created, "the other tenant must get a fresh resource"); + assert_eq!(stored.version_id(), "1"); + assert_eq!(stored.tenant_id().as_str(), other.tenant_id().as_str()); + + match a.read(&tenant, "Patient", &id).await { + Err(StorageError::Resource(ResourceError::Gone { .. })) => {} + got => panic!("first tenant's resource must still be deleted, got {got:?}"), + } + assert_mongo_history_versions(&a, &tenant, "Patient", &id, 2).await; +} diff --git a/crates/persistence/tests/postgres_tests.rs b/crates/persistence/tests/postgres_tests.rs index 40e92be1c..d78d5237c 100644 --- a/crates/persistence/tests/postgres_tests.rs +++ b/crates/persistence/tests/postgres_tests.rs @@ -63,6 +63,13 @@ mod meta_params_suite; #[path = "search/date_boundary_suite.rs"] mod date_boundary_suite; +/// The T2 cluster harness: two independently constructed backends over one +/// shared store, barrier-synchronized racing, and the definition-of-done +/// assertion rows. Declared at the top level for the same `#[path]` resolution +/// reason as `if_match_suite` above. +#[path = "common/cluster_harness.rs"] +mod cluster_harness; + // ============================================================================ // Backend Configuration Tests (no PostgreSQL instance required) // ============================================================================ @@ -6407,4 +6414,671 @@ mod postgres_integration { ) .await; } + + // ======================================================================== + // T2 cluster harness — calibration against the bulk-export job store. + // ======================================================================== + // + // Two freshly constructed backends over the one shared container play + // "instance A" and "instance B" (`create_backend()` builds a new pool each + // time; the only thing the two handles share is the database). The + // bulk-export job store is already cluster-safe — DB-leased and fenced — + // so these rows must pass on an unchanged tree: they prove the harness, + // not the product. Every later cluster-capable subsystem points the same + // helpers at its own store. + + use crate::cluster_harness as harness; + use helios_persistence::core::bulk_export::{ExportJobId, ExportProgress}; + use helios_persistence::error::BulkExportError; + use std::sync::Arc; + + /// Maps `get_export_status` to the harness's `Option` shape: a job that + /// doesn't exist *for this tenant* is `None`; any other error is a bug. + async fn export_status_across( + backend: &PostgresBackend, + tenant: &TenantContext, + job_id: &ExportJobId, + ) -> Option { + match backend.get_export_status(tenant, job_id).await { + Ok(progress) => Some(progress), + Err(StorageError::BulkExport(BulkExportError::JobNotFound { .. })) => None, + Err(other) => panic!("unexpected error reading export status: {other:?}"), + } + } + + /// Claims and finishes every eligible job so an exclusivity race starts + /// from an empty queue. Leftover `accepted` jobs from other suites + /// sharing the container would otherwise let both racers win on + /// different jobs. + async fn drain_export_queue(backend: &PostgresBackend) { + let reaper = WorkerId::new(format!("cluster-drain-{}", uuid::Uuid::new_v4())); + while let Some(lease) = backend + .claim_next(&reaper, StdDuration::from_secs(60)) + .await + .unwrap() + { + backend + .finish_export_job( + &lease.tenant, + &lease.job_id, + &lease.worker_id, + lease.fencing_token, + ) + .await + .expect("draining a leftover export job"); + } + } + + /// DoD rows: visibility + isolation. + #[tokio::test] + async fn postgres_integration_cluster_bulk_export_visible_and_isolated_across_handles() { + let _guard = BULK_EXPORT_TEST_LOCK.lock().await; + let handles = harness::two_handles(create_backend).await; + let tenant = create_tenant("cluster-vis"); + + let job_id = handles + .a + .start_export(&tenant, export_input(ExportRequest::system())) + .await + .unwrap(); + + // Visibility: created via instance A, observable via instance B. + let progress = harness::assert_visible( + export_status_across(&handles.b, &tenant, &job_id).await, + "export job created via instance A", + ); + assert_eq!(progress.status, ExportStatus::Accepted); + + // Isolation: another tenant on instance B sees nothing. + let other_tenant = create_tenant("cluster-vis-other"); + harness::assert_wrong_tenant_hidden( + export_status_across(&handles.b, &other_tenant, &job_id).await, + "export job", + ); + } + + /// DoD row: exclusivity — two instances race `claim_next` on one queued + /// job and exactly one wins (`FOR UPDATE SKIP LOCKED`). + #[tokio::test] + async fn postgres_integration_cluster_bulk_export_claim_exclusive_across_handles() { + let _guard = BULK_EXPORT_TEST_LOCK.lock().await; + let handles = harness::two_handles(create_backend).await; + let tenant = create_tenant("cluster-claim"); + + drain_export_queue(&handles.a).await; + + let job_id = handles + .a + .start_export(&tenant, export_input(ExportRequest::system())) + .await + .unwrap(); + + let worker_a = WorkerId::new(format!("cluster-claim-a-{}", uuid::Uuid::new_v4())); + let worker_b = WorkerId::new(format!("cluster-claim-b-{}", uuid::Uuid::new_v4())); + let (a, b) = (handles.a, handles.b); + let (lease_a, lease_b) = harness::race2( + async move { + a.claim_next(&worker_a, StdDuration::from_secs(60)) + .await + .unwrap() + }, + async move { + b.claim_next(&worker_b, StdDuration::from_secs(60)) + .await + .unwrap() + }, + ) + .await; + + harness::assert_exactly_one(&lease_a, &lease_b, "the claim on one queued job"); + let winner = lease_a.or(lease_b).unwrap(); + assert_eq!(winner.job_id, job_id); + assert_eq!( + winner.tenant.tenant_id().as_str(), + tenant.tenant_id().as_str() + ); + } + + /// DoD row: fencing — after a release/reclaim moves the lease to another + /// instance, the stale holder's heartbeat and guarded writes are refused. + /// Deterministic: the reclaim uses `release`, not lease expiry, so there + /// are no sleeps (coverage-safe). + #[tokio::test] + async fn postgres_integration_cluster_bulk_export_stale_handle_fenced_after_release() { + let _guard = BULK_EXPORT_TEST_LOCK.lock().await; + let handles = harness::two_handles(create_backend).await; + let tenant = create_tenant("cluster-fence"); + + let job_id = handles + .a + .start_export(&tenant, export_input(ExportRequest::system())) + .await + .unwrap(); + + // Instance A claims, then releases (graceful shutdown). + let worker_a = WorkerId::new(format!("cluster-fence-a-{}", uuid::Uuid::new_v4())); + let lease_a = + claim_specific(&handles.a, &worker_a, &job_id, StdDuration::from_secs(60)).await; + ExportClaimStrategy::release(&handles.a, lease_a.clone()) + .await + .unwrap(); + + // Instance B reclaims; the fencing token moves past A's. + let worker_b = WorkerId::new(format!("cluster-fence-b-{}", uuid::Uuid::new_v4())); + let lease_b = + claim_specific(&handles.b, &worker_b, &job_id, StdDuration::from_secs(60)).await; + assert!(lease_b.fencing_token > lease_a.fencing_token); + + // The stale handle is fenced out of heartbeat and guarded writes. + assert!(matches!( + handles.a.heartbeat(&lease_a).await, + Err(LeaseError::LeaseLost { .. }) + )); + assert!(matches!( + handles + .a + .mark_export_in_progress(&tenant, &job_id, &worker_a, lease_a.fencing_token) + .await, + Err(LeaseError::LeaseLost { .. }) + )); + + // The current holder is unaffected. + handles + .b + .finish_export_job(&tenant, &job_id, &worker_b, lease_b.fencing_token) + .await + .unwrap(); + } + + /// DoD row: durability — dropping the creating handle (simulated + /// redeploy) must not lose the job; a fresh handle still sees it. + #[tokio::test] + async fn postgres_integration_cluster_bulk_export_survives_handle_drop() { + let _guard = BULK_EXPORT_TEST_LOCK.lock().await; + let tenant = create_tenant("cluster-durable"); + + let first = create_backend().await; + let job_id = first + .start_export(&tenant, export_input(ExportRequest::system())) + .await + .unwrap(); + drop(first); + + let replacement = create_backend().await; + let progress = harness::assert_visible( + export_status_across(&replacement, &tenant, &job_id).await, + "export job after its creating handle was dropped", + ); + assert_eq!(progress.status, ExportStatus::Accepted); + } + + // ======================================================================== + // T2 cluster suite — F5 resource version-id race. + // ======================================================================== + // + // Every version bump is a version-guarded compare-and-swap that lands the + // resource row and its history row in one statement, so two instances + // racing unconditional writes can never both assign the same version_id + // or lose a history row. `create_or_update` retries transient CAS losses, + // keeping unconditional PUT last-writer-wins at the API surface — a PUT + // must never surface a 404 or 409 that only a concurrent PUT produced. + + /// Asserts a resource's history holds every version exactly once, + /// contiguous from 1 — no duplicate version_ids, no lost history rows. + async fn assert_history_versions( + backend: &PostgresBackend, + tenant: &TenantContext, + resource_type: &str, + id: &str, + expected_len: u64, + ) { + let history = backend + .history_instance( + tenant, + resource_type, + id, + &HistoryParams::new().include_deleted(true), + ) + .await + .unwrap(); + let mut versions: Vec = history + .items + .iter() + .map(|e| e.resource.version_id().parse().unwrap()) + .collect(); + versions.sort_unstable(); + assert_eq!( + versions, + (1..=expected_len).collect::>(), + "history must hold every version exactly once" + ); + } + + /// DoD rows (F5): exclusivity + isolation — two instances race + /// unconditional `update`s from the same version-1 snapshot. The CAS lets + /// exactly one through; the loser gets `VersionConflict` instead of + /// silently losing an update or 500ing on a duplicate history row, and a + /// fresh read-then-retry converges on version 3. + #[tokio::test] + async fn postgres_integration_cluster_resource_update_race_no_lost_version() { + let handles = harness::two_handles(create_backend).await; + let tenant = create_tenant("cluster-f5-upd"); + + let created = handles + .a + .create( + &tenant, + "Patient", + json!({"resourceType": "Patient", "name": [{"family": "V1"}]}), + FhirVersion::default(), + ) + .await + .unwrap(); + let id = created.id().to_string(); + + let (a, b) = (handles.a, handles.b); + let (tenant_a, tenant_b) = (tenant.clone(), tenant.clone()); + let (snapshot_a, snapshot_b) = (created.clone(), created); + let (res_a, res_b) = harness::race2( + async move { + a.update( + &tenant_a, + &snapshot_a, + json!({"resourceType": "Patient", "name": [{"family": "FromA"}]}), + ) + .await + }, + async move { + b.update( + &tenant_b, + &snapshot_b, + json!({"resourceType": "Patient", "name": [{"family": "FromB"}]}), + ) + .await + }, + ) + .await; + + let (ok_a, err_a) = match res_a { + Ok(r) => (Some(r), None), + Err(e) => (None, Some(e)), + }; + let (ok_b, err_b) = match res_b { + Ok(r) => (Some(r), None), + Err(e) => (None, Some(e)), + }; + harness::assert_exactly_one(&ok_a, &ok_b, "the version-guarded update"); + let winner = ok_a.or(ok_b).unwrap(); + assert_eq!(winner.version_id(), "2"); + let loser = err_a.or(err_b).unwrap(); + assert!( + matches!( + loser, + StorageError::Concurrency(ConcurrencyError::VersionConflict { .. }) + ), + "loser must see VersionConflict, got {loser:?}" + ); + + // A fresh read-then-retry from a third instance converges. + let verifier = create_backend().await; + let current = verifier + .read(&tenant, "Patient", &id) + .await + .unwrap() + .unwrap(); + assert_eq!(current.version_id(), "2"); + let retried = verifier + .update( + &tenant, + ¤t, + json!({"resourceType": "Patient", "name": [{"family": "Retry"}]}), + ) + .await + .unwrap(); + assert_eq!(retried.version_id(), "3"); + + assert_history_versions(&verifier, &tenant, "Patient", &id, 3).await; + + // Isolation: the resource does not exist for another tenant. + harness::assert_wrong_tenant_hidden( + verifier + .read(&create_tenant("cluster-f5-other"), "Patient", &id) + .await + .unwrap(), + "patient", + ); + } + + /// DoD row (F5): unconditional PUT stays last-writer-wins — two instances + /// race `create_or_update` on one existing resource; the bounded CAS + /// retry absorbs the losing race so both callers succeed, with distinct + /// version_ids and a complete history. + #[tokio::test] + async fn postgres_integration_cluster_resource_create_or_update_race_both_succeed() { + let handles = harness::two_handles(create_backend).await; + let tenant = create_tenant("cluster-f5-put"); + + let created = handles + .a + .create( + &tenant, + "Patient", + json!({"resourceType": "Patient", "name": [{"family": "V1"}]}), + FhirVersion::default(), + ) + .await + .unwrap(); + let id = created.id().to_string(); + + let (a, b) = (handles.a, handles.b); + let (tenant_a, tenant_b) = (tenant.clone(), tenant.clone()); + let (id_a, id_b) = (id.clone(), id.clone()); + let (res_a, res_b) = harness::race2( + async move { + a.create_or_update( + &tenant_a, + "Patient", + &id_a, + json!({"resourceType": "Patient", "name": [{"family": "FromA"}]}), + FhirVersion::default(), + ) + .await + }, + async move { + b.create_or_update( + &tenant_b, + "Patient", + &id_b, + json!({"resourceType": "Patient", "name": [{"family": "FromB"}]}), + FhirVersion::default(), + ) + .await + }, + ) + .await; + + let (stored_a, created_a) = res_a.expect("create_or_update must absorb CAS races"); + let (stored_b, created_b) = res_b.expect("create_or_update must absorb CAS races"); + assert!( + !created_a && !created_b, + "both racers must take the update arm" + ); + assert_ne!( + stored_a.version_id(), + stored_b.version_id(), + "racing writers must never assign the same version" + ); + + let verifier = create_backend().await; + let current = verifier + .read(&tenant, "Patient", &id) + .await + .unwrap() + .unwrap(); + assert_eq!(current.version_id(), "3"); + assert_history_versions(&verifier, &tenant, "Patient", &id, 3).await; + } + + /// DoD row (F5): an unconditional `update` racing an unconditional + /// `delete` can interleave either way, but the CAS keeps the invariants: + /// the delete always lands, no duplicate version_id, no lost history row, + /// and the resource ends deleted. + #[tokio::test] + async fn postgres_integration_cluster_resource_update_delete_race_keeps_history_coherent() { + let handles = harness::two_handles(create_backend).await; + let tenant = create_tenant("cluster-f5-del"); + + let created = handles + .a + .create( + &tenant, + "Patient", + json!({"resourceType": "Patient", "name": [{"family": "V1"}]}), + FhirVersion::default(), + ) + .await + .unwrap(); + let id = created.id().to_string(); + + let (a, b) = (handles.a, handles.b); + let (tenant_a, tenant_b) = (tenant.clone(), tenant.clone()); + let id_b = id.clone(); + let (res_update, res_delete) = harness::race2( + async move { + a.update( + &tenant_a, + &created, + json!({"resourceType": "Patient", "name": [{"family": "FromA"}]}), + ) + .await + }, + async move { b.delete(&tenant_b, "Patient", &id_b).await }, + ) + .await; + + // The delete derives the tombstone's version from the locked row + // inside its own statement, so it lands whichever way the race goes. + res_delete.expect("racing delete must succeed"); + + // The update either beat the delete (then the delete bumped again from + // the row it locked) or found the row gone. + let expected_versions = match res_update { + Ok(stored) => { + assert_eq!(stored.version_id(), "2"); + 3 + } + Err(err) => { + assert!( + matches!(err, StorageError::Resource(ResourceError::NotFound { .. })), + "update racing a delete must see NotFound, got {err:?}" + ); + 2 + } + }; + + let verifier = create_backend().await; + match verifier.read(&tenant, "Patient", &id).await { + Err(StorageError::Resource(ResourceError::Gone { .. })) => {} + other => panic!("resource must end deleted, got {other:?}"), + } + assert_history_versions(&verifier, &tenant, "Patient", &id, expected_versions).await; + } + + /// A fresh handle whose pool holds exactly one connection, so a warm-up + /// and the race that follows provably share one connection and one + /// prepared-statement cache — the race then measures the storage + /// protocol, not connection setup. + async fn create_race_backend() -> PostgresBackend { + let pg = shared_pg().await; + let data_dir = PathBuf::from(env!("CARGO_MANIFEST_DIR")) + .parent() + .and_then(|p| p.parent()) + .map(|p| p.join("data")) + .unwrap_or_else(|| PathBuf::from("data")); + let config = PostgresConfig { + host: pg.host.clone(), + port: pg.port, + dbname: "postgres".to_string(), + user: "postgres".to_string(), + password: Some("postgres".to_string()), + max_connections: 1, + data_dir: Some(data_dir), + ..Default::default() + }; + PostgresBackend::new(config) + .await + .expect("Failed to create single-connection PostgresBackend") + } + + /// Creates a Patient and soft-deletes it, returning its id: version 1 is + /// the create, version 2 the tombstone. + async fn seed_tombstone(backend: &PostgresBackend, tenant: &TenantContext) -> String { + let created = backend + .create( + tenant, + "Patient", + json!({"resourceType": "Patient", "name": [{"family": "Seed"}]}), + FhirVersion::default(), + ) + .await + .unwrap(); + backend + .delete(tenant, "Patient", created.id()) + .await + .unwrap(); + created.id().to_string() + } + + /// Runs one throwaway create → delete → restore cycle so every statement + /// on the restore path is prepared on the handle's single connection + /// before the timed race. + async fn warm_restore_path(backend: &PostgresBackend, tenant: &TenantContext) { + let id = seed_tombstone(backend, tenant).await; + backend + .create_or_update( + tenant, + "Patient", + &id, + json!({"resourceType": "Patient", "name": [{"family": "Warm"}]}), + FhirVersion::default(), + ) + .await + .unwrap(); + } + + /// DoD rows (F5, restore path): two instances race unconditional PUTs + /// onto one soft-deleted id. The restore is a version-guarded CAS that + /// lands the row and its history entry in one statement, and + /// `create_or_update` retries the loser from a fresh read, so both PUTs + /// succeed: the winner restores (v3, "created"), the loser converges + /// through the update arm (v4, "updated"), history is contiguous 1..=4, + /// and the newest history entry carries the live body. Without the CAS, + /// both racers compute v3 — the second history insert violates the + /// primary key (a 500) and the live row and history v3 carry different + /// bodies. Five rounds on fresh ids give the interleaving room to occur; + /// the post-conditions are the same for every interleaving. + #[tokio::test] + async fn postgres_integration_cluster_restore_deleted_race_keeps_history_coherent() { + let tenant = create_tenant("cluster-f5-restore"); + let a = Arc::new(create_race_backend().await); + let b = Arc::new(create_race_backend().await); + warm_restore_path(&a, &tenant).await; + warm_restore_path(&b, &tenant).await; + let verifier = create_backend().await; + + for round in 0..5 { + let id = seed_tombstone(&a, &tenant).await; + + let (ra, rb) = (Arc::clone(&a), Arc::clone(&b)); + let (tenant_a, tenant_b) = (tenant.clone(), tenant.clone()); + let (id_a, id_b) = (id.clone(), id.clone()); + let (res_a, res_b) = harness::race2( + async move { + ra.create_or_update( + &tenant_a, + "Patient", + &id_a, + json!({"resourceType": "Patient", "name": [{"family": "FromA"}]}), + FhirVersion::default(), + ) + .await + }, + async move { + rb.create_or_update( + &tenant_b, + "Patient", + &id_b, + json!({"resourceType": "Patient", "name": [{"family": "FromB"}]}), + FhirVersion::default(), + ) + .await + }, + ) + .await; + + let (stored_a, created_a) = + res_a.unwrap_or_else(|e| panic!("round {round}: PUT via A must succeed: {e:?}")); + let (stored_b, created_b) = + res_b.unwrap_or_else(|e| panic!("round {round}: PUT via B must succeed: {e:?}")); + + // Exactly one racer restores the tombstone; the other converges + // through the update arm onto the restored row. + let restored = |flag: bool| flag.then_some(()); + harness::assert_exactly_one( + &restored(created_a), + &restored(created_b), + &format!("round {round}: the restore of one tombstone"), + ); + assert_ne!( + stored_a.version_id(), + stored_b.version_id(), + "round {round}: racing writers must never assign the same version" + ); + + let current = verifier + .read(&tenant, "Patient", &id) + .await + .unwrap() + .unwrap(); + assert_eq!(current.version_id(), "4", "round {round}"); + assert!(!current.is_deleted(), "round {round}"); + assert_history_versions(&verifier, &tenant, "Patient", &id, 4).await; + + // The newest history entry must be the live body — a lost race + // that still "succeeded" leaves history v3 with one racer's body + // and the live row with the other's. + let history = verifier + .history_instance( + &tenant, + "Patient", + &id, + &HistoryParams::new().include_deleted(true), + ) + .await + .unwrap(); + let newest = history + .items + .iter() + .max_by_key(|e| e.resource.version_id().parse::().unwrap()) + .unwrap(); + assert_eq!( + newest.resource.content(), + current.content(), + "round {round}: newest history entry must carry the live body" + ); + } + } + + /// DoD row (F5, restore path): isolation — a PUT onto the same id under + /// another tenant creates a fresh version-1 resource in that tenant and + /// leaves the first tenant's tombstone exactly as it was. + #[tokio::test] + async fn postgres_integration_cluster_restore_deleted_wrong_tenant_creates_not_restores() { + let handles = harness::two_handles(create_backend).await; + let tenant = create_tenant("cluster-f5-restore-iso"); + let id = seed_tombstone(&handles.a, &tenant).await; + + let other = create_tenant("cluster-f5-restore-other"); + let (stored, created) = handles + .b + .create_or_update( + &other, + "Patient", + &id, + json!({"resourceType": "Patient", "name": [{"family": "Other"}]}), + FhirVersion::default(), + ) + .await + .unwrap(); + assert!(created, "the other tenant must get a fresh resource"); + assert_eq!(stored.version_id(), "1"); + assert_eq!(stored.tenant_id().as_str(), other.tenant_id().as_str()); + + // The first tenant's tombstone is untouched. + match handles.a.read(&tenant, "Patient", &id).await { + Err(StorageError::Resource(ResourceError::Gone { .. })) => {} + got => panic!("first tenant's resource must still be deleted, got {got:?}"), + } + assert_history_versions(&handles.a, &tenant, "Patient", &id, 2).await; + } }