Skip to content

fix(persistence): version-guarded restore_deleted, PUT race retry, and the two-handle cluster harness (Phase 0.1) #905

New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

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

fix(persistence): version-guarded restore_deleted, PUT race retry, and the two-handle cluster harness (Phase 0.1) #905

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
4927af5
fix(persistence): make restore_deleted a version-guarded CAS and retr…
aacruzgon Sep 2, 2026
864b218
chore: ignore local cluster planning docs under docs/
aacruzgon Sep 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions .gitignore
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
125 changes: 92 additions & 33 deletions crates/persistence/src/backends/mongodb/storage.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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, &current, 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, &current, 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),
}
}

Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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! {
Expand Down
165 changes: 124 additions & 41 deletions crates/persistence/src/backends/postgres/storage.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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, &current, 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, &current, 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),
}
}

Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -1135,33 +1184,67 @@ 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,
&now,
&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
Expand Down
19 changes: 19 additions & 0 deletions crates/persistence/src/core/storage.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
Loading