From 90b85296dcb8a5a367cc526a57e8179134b15eaf Mon Sep 17 00:00:00 2001 From: angela-helios Date: Tue, 1 Sep 2026 22:14:54 -0400 Subject: [PATCH 1/3] fix(persistence,rest): _lastUpdated on Postgres; server meta in response bodies MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two findings from the #448 Postgres leg. _lastUpdated searches returned an opaque 500 on the Postgres backend at every precision and prefix: build_last_updated_condition bound the raw query string as a text parameter against the TIMESTAMPTZ last_updated column, which tokio-postgres refuses to serialize, and the path had none of the precision-range semantics the date parameters got in #463. It now binds real timestamps and mirrors build_date_condition: eq at day precision means [day, day+1), gt excludes the named period, and a full-precision instant falls back to scalar comparison. Response bodies carried no server-populated meta on any read path — versionId and lastUpdated lived only in the ETag and Last-Modified headers, so search entries, reads, vreads, history bundles, and batch/transaction responses returned resources with no server metadata, and the Resources workspace's UPDATED column rendered blank on every backend. StoredResource::content_with_meta now merges the row's version and timestamp into the returned body (client-supplied meta members like profile survive), and every handler and bundle builder that echoed stored content uses it. Verified live against postgres:16 with the #448 dataset: the full battery passes 22/22, and a Patient read returns meta.profile + versionId + lastUpdated together. Closes #871 Closes #873 --- .../backends/postgres/search/query_builder.rs | 124 +++++++++++++++++- crates/persistence/src/core/search.rs | 4 +- crates/persistence/src/core/transaction.rs | 4 +- .../persistence/src/types/stored_resource.rs | 29 ++++ crates/rest/src/handlers/create.rs | 20 +-- crates/rest/src/handlers/history.rs | 2 +- crates/rest/src/handlers/patch.rs | 2 +- crates/rest/src/handlers/read.rs | 2 +- .../rest/src/handlers/subscription_event.rs | 2 +- crates/rest/src/handlers/update.rs | 7 +- crates/rest/src/handlers/vread.rs | 13 +- crates/rest/tests/search_integration.rs | 90 +++++++++++++ 12 files changed, 268 insertions(+), 31 deletions(-) diff --git a/crates/persistence/src/backends/postgres/search/query_builder.rs b/crates/persistence/src/backends/postgres/search/query_builder.rs index a2b2df42b4..75847462d3 100644 --- a/crates/persistence/src/backends/postgres/search/query_builder.rs +++ b/crates/persistence/src/backends/postgres/search/query_builder.rs @@ -642,15 +642,76 @@ impl PostgresQueryBuilder { Some(combined) } + /// Builds a `last_updated` comparison against the `resources` column. + /// + /// Binds real timestamps — `last_updated` is `TIMESTAMPTZ`, and a text + /// parameter fails serialization outright (#871) — and applies the same + /// precision-range semantics as [`Self::build_date_condition`]: `eq` at + /// day precision means `[day, day+1)`, not a scalar equality that nothing + /// can ever hit. fn build_last_updated_condition(values: &[SearchValue], offset: usize) -> Option { let mut conditions = Vec::new(); - for (i, value) in values.iter().enumerate() { - let param_num = offset + i + 1; - let op = Self::prefix_to_operator(&value.prefix); - conditions.push(SqlFragment::with_params( - format!("last_updated {} ${}", op, param_num), - vec![SqlParam::text(&value.value)], - )); + let mut next = offset; + for value in values { + let (start, end) = date_precision_range(&value.value); + let degenerate = start == end; + let (sql, params): (String, Vec) = match value.prefix { + SearchPrefix::Eq if !degenerate => { + next += 1; + let a = next; + next += 1; + ( + format!("last_updated >= ${a} AND last_updated < ${next}"), + vec![SqlParam::Timestamp(start), SqlParam::Timestamp(end)], + ) + } + SearchPrefix::Ne if !degenerate => { + next += 1; + let a = next; + next += 1; + ( + format!("(last_updated < ${a} OR last_updated >= ${next})"), + vec![SqlParam::Timestamp(start), SqlParam::Timestamp(end)], + ) + } + SearchPrefix::Gt | SearchPrefix::Sa if !degenerate => { + next += 1; + ( + format!("last_updated >= ${next}"), + vec![SqlParam::Timestamp(end)], + ) + } + SearchPrefix::Lt | SearchPrefix::Eb if !degenerate => { + next += 1; + ( + format!("last_updated < ${next}"), + vec![SqlParam::Timestamp(start)], + ) + } + SearchPrefix::Ge if !degenerate => { + next += 1; + ( + format!("last_updated >= ${next}"), + vec![SqlParam::Timestamp(start)], + ) + } + SearchPrefix::Le if !degenerate => { + next += 1; + ( + format!("last_updated < ${next}"), + vec![SqlParam::Timestamp(end)], + ) + } + other => { + let op = Self::prefix_to_operator(&other); + next += 1; + ( + format!("last_updated {op} ${next}"), + vec![SqlParam::Timestamp(start)], + ) + } + }; + conditions.push(SqlFragment::with_params(sql, params)); } if conditions.is_empty() { return None; @@ -1839,6 +1900,55 @@ mod tests { assert_eq!(frag.params.len(), 1); } + #[test] + fn last_updated_binds_timestamps_with_precision_ranges() { + // #871: a text param bound against the TIMESTAMPTZ column fails + // serialization, and day-precision eq must mean [day, day+1). + let query = SearchQuery::new("Patient").with_parameter(special_param( + "_lastUpdated", + vec![SearchValue::eq("2026-09-01")], + )); + let frag = PostgresQueryBuilder::build_search_query(&query, 2) + .expect("_lastUpdated must produce a condition"); + + assert!( + frag.sql + .contains("last_updated >= $3 AND last_updated < $4"), + "{}", + frag.sql + ); + assert_eq!(frag.params.len(), 2); + assert!( + frag.params + .iter() + .all(|p| matches!(p, SqlParam::Timestamp(_))), + "must bind timestamps, not text: {:?}", + frag.params + ); + } + + #[test] + fn last_updated_gt_excludes_the_named_day() { + let query = SearchQuery::new("Patient").with_parameter(special_param( + "_lastUpdated", + vec![SearchValue { + prefix: SearchPrefix::Gt, + value: "2026-09-01".to_string(), + }], + )); + let frag = PostgresQueryBuilder::build_search_query(&query, 2) + .expect("_lastUpdated gt must produce a condition"); + + assert!(frag.sql.contains("last_updated >= $3"), "{}", frag.sql); + assert_eq!(frag.params.len(), 1); + match &frag.params[0] { + SqlParam::Timestamp(ts) => { + assert_eq!(ts.to_rfc3339(), "2026-09-02T00:00:00+00:00"); + } + other => panic!("must bind the period end as a timestamp: {other:?}"), + } + } + #[test] fn text_or_list_placeholders_are_gap_free() { // Two terms OR together, and a blank one must not consume a placeholder diff --git a/crates/persistence/src/core/search.rs b/crates/persistence/src/core/search.rs index 212d7d6b61..ab61ea11b1 100644 --- a/crates/persistence/src/core/search.rs +++ b/crates/persistence/src/core/search.rs @@ -138,7 +138,7 @@ impl SearchResult { // computed one for this resource (`Bundle.entry.search.score`). for resource in &self.resources.items { let full_url = format!("{}/{}", base_url, resource.url()); - let entry = BundleEntry::match_entry(full_url, resource.content().clone()) + let entry = BundleEntry::match_entry(full_url, resource.content_with_meta()) .with_score(self.scores.get(&resource.url()).copied()); bundle = bundle.with_entry(entry); } @@ -148,7 +148,7 @@ impl SearchResult { let full_url = format!("{}/{}", base_url, resource.url()); bundle = bundle.with_entry(BundleEntry::include_entry( full_url, - resource.content().clone(), + resource.content_with_meta(), )); } diff --git a/crates/persistence/src/core/transaction.rs b/crates/persistence/src/core/transaction.rs index 957bc236ee..806e8d696c 100644 --- a/crates/persistence/src/core/transaction.rs +++ b/crates/persistence/src/core/transaction.rs @@ -329,7 +329,7 @@ impl BundleEntryResult { location: Some(resource.versioned_url()), etag: Some(resource.etag().to_string()), last_modified: Some(resource.last_modified().to_rfc3339()), - resource: Some(resource.into_content()), + resource: Some(resource.content_with_meta()), outcome: None, } } @@ -341,7 +341,7 @@ impl BundleEntryResult { location: None, etag: Some(resource.etag().to_string()), last_modified: Some(resource.last_modified().to_rfc3339()), - resource: Some(resource.into_content()), + resource: Some(resource.content_with_meta()), outcome: None, } } diff --git a/crates/persistence/src/types/stored_resource.rs b/crates/persistence/src/types/stored_resource.rs index 885d61e5ab..4febb78e79 100644 --- a/crates/persistence/src/types/stored_resource.rs +++ b/crates/persistence/src/types/stored_resource.rs @@ -190,6 +190,35 @@ impl StoredResource { &self.content } + /// Returns the content with server-populated `meta.versionId` and + /// `meta.lastUpdated` merged in from this row's version and timestamp. + /// + /// Stored resources persist the content as submitted — the version and + /// timestamp live in their own columns — so bodies read back without this + /// carry no server metadata even though the ETag/Last-Modified headers do + /// (#873). Client-supplied `meta` members (`profile`, `tag`, `security`) + /// are preserved; `versionId`/`lastUpdated` are overwritten because the + /// server's row is authoritative for both. + pub fn content_with_meta(&self) -> Value { + let mut content = self.content.clone(); + if let Value::Object(obj) = &mut content { + let meta = obj + .entry("meta") + .or_insert_with(|| Value::Object(serde_json::Map::new())); + if let Value::Object(meta) = meta { + meta.insert("versionId".into(), Value::String(self.version_id.clone())); + meta.insert( + "lastUpdated".into(), + Value::String( + self.last_modified + .to_rfc3339_opts(chrono::SecondsFormat::Millis, true), + ), + ); + } + } + content + } + /// Returns a mutable reference to the resource content. pub fn content_mut(&mut self) -> &mut Value { &mut self.content diff --git a/crates/rest/src/handlers/create.rs b/crates/rest/src/handlers/create.rs index 0be6d78eed..518138f229 100644 --- a/crates/rest/src/handlers/create.rs +++ b/crates/rest/src/handlers/create.rs @@ -294,11 +294,10 @@ fn build_create_response( } _ => { // Default: return=representation - format_resource_response(status, header_map, stored.content(), format).map_err(|_| { - RestError::InternalError { + format_resource_response(status, header_map, &stored.content_with_meta(), format) + .map_err(|_| RestError::InternalError { message: "Failed to serialize response".to_string(), - } - }) + }) } } } @@ -314,9 +313,14 @@ fn build_existing_response( match prefer.return_preference() { Some("minimal") => Ok((StatusCode::OK, header_map).into_response()), - _ => format_resource_response(StatusCode::OK, header_map, stored.content(), format) - .map_err(|_| RestError::InternalError { - message: "Failed to serialize response".to_string(), - }), + _ => format_resource_response( + StatusCode::OK, + header_map, + &stored.content_with_meta(), + format, + ) + .map_err(|_| RestError::InternalError { + message: "Failed to serialize response".to_string(), + }), } } diff --git a/crates/rest/src/handlers/history.rs b/crates/rest/src/handlers/history.rs index 26bdd4f127..ff5e82d218 100644 --- a/crates/rest/src/handlers/history.rs +++ b/crates/rest/src/handlers/history.rs @@ -117,7 +117,7 @@ where // Deleted versions carry no resource body. if entry.method != HistoryMethod::Delete { - bundle_entry["resource"] = resource.content().clone(); + bundle_entry["resource"] = resource.content_with_meta(); } bundle_entry diff --git a/crates/rest/src/handlers/patch.rs b/crates/rest/src/handlers/patch.rs index 164813334f..7e12f19339 100644 --- a/crates/rest/src/handlers/patch.rs +++ b/crates/rest/src/handlers/patch.rs @@ -324,6 +324,6 @@ fn build_patch_response( }); Ok((StatusCode::OK, header_map, Json(outcome)).into_response()) } - _ => Ok((StatusCode::OK, header_map, Json(stored.content().clone())).into_response()), + _ => Ok((StatusCode::OK, header_map, Json(stored.content_with_meta())).into_response()), } } diff --git a/crates/rest/src/handlers/read.rs b/crates/rest/src/handlers/read.rs index 21c507f71f..171e0a2d76 100644 --- a/crates/rest/src/handlers/read.rs +++ b/crates/rest/src/handlers/read.rs @@ -142,7 +142,7 @@ where .get("_elements") .map(|v| v.split(',').map(|s| s.trim()).collect()); - let mut content = stored.content().clone(); + let mut content = stored.content_with_meta(); let mut subsetted = false; if let Some(mode) = summary_mode { diff --git a/crates/rest/src/handlers/subscription_event.rs b/crates/rest/src/handlers/subscription_event.rs index 8df93fa636..404fbb2f9a 100644 --- a/crates/rest/src/handlers/subscription_event.rs +++ b/crates/rest/src/handlers/subscription_event.rs @@ -30,7 +30,7 @@ pub fn emit_subscription_event( resource_id: stored.id().to_string(), version_id: stored.version_id().to_string(), event_type, - resource: Some(stored.content().clone()), + resource: Some(stored.content_with_meta()), previous_resource: None, timestamp: chrono::Utc::now(), }; diff --git a/crates/rest/src/handlers/update.rs b/crates/rest/src/handlers/update.rs index 79a5a161a9..adc79cafd1 100644 --- a/crates/rest/src/handlers/update.rs +++ b/crates/rest/src/handlers/update.rs @@ -421,10 +421,9 @@ fn build_update_response( } }) } - _ => format_resource_response(status, header_map, stored.content(), format).map_err(|_| { - RestError::InternalError { + _ => format_resource_response(status, header_map, &stored.content_with_meta(), format) + .map_err(|_| RestError::InternalError { message: "Failed to serialize response".to_string(), - } - }), + }), } } diff --git a/crates/rest/src/handlers/vread.rs b/crates/rest/src/handlers/vread.rs index 07c2d9e81f..53296a9ac9 100644 --- a/crates/rest/src/handlers/vread.rs +++ b/crates/rest/src/handlers/vread.rs @@ -106,10 +106,15 @@ where "Returning resource version" ); - format_resource_response(StatusCode::OK, headers, stored.content(), negotiated.format) - .map_err(|_| RestError::InternalError { - message: "Failed to serialize response".to_string(), - }) + format_resource_response( + StatusCode::OK, + headers, + &stored.content_with_meta(), + negotiated.format, + ) + .map_err(|_| RestError::InternalError { + message: "Failed to serialize response".to_string(), + }) } None => Err(RestError::VersionNotFound { resource_type, diff --git a/crates/rest/tests/search_integration.rs b/crates/rest/tests/search_integration.rs index 2e1ba9a9d5..76fb810407 100644 --- a/crates/rest/tests/search_integration.rs +++ b/crates/rest/tests/search_integration.rs @@ -3016,3 +3016,93 @@ mod date_precision { ); } } + +/// #873: response bodies must carry server-populated `meta.versionId` and +/// `meta.lastUpdated` on every read path. The row's version and timestamp were +/// header-only (ETag / Last-Modified), so search entries, reads, vreads, and +/// history bundles all returned resources with no server metadata — visibly, +/// the Resources workspace's UPDATED column rendered blank on every backend. +mod server_meta { + use super::*; + + #[tokio::test] + async fn test_bodies_carry_version_and_last_updated_on_every_read_path() { + let (server, _backend) = create_test_server().await; + let tenant_header = HeaderValue::from_static("test-tenant"); + + // Create: the response body already carries v1 meta, with + // client-supplied meta members preserved. + let created = server + .post("/Patient") + .add_header(X_TENANT_ID, tenant_header.clone()) + .json(&json!({ + "resourceType": "Patient", + "meta": {"profile": ["http://example.org/StructureDefinition/p"]}, + "name": [{"family": "Metadata"}] + })) + .await; + created.assert_status(StatusCode::CREATED); + let body: Value = created.json(); + assert_eq!(body["meta"]["versionId"], "1"); + assert!(body["meta"]["lastUpdated"].is_string()); + assert_eq!( + body["meta"]["profile"][0], "http://example.org/StructureDefinition/p", + "client-supplied meta members must survive the injection" + ); + let id = body["id"].as_str().expect("created id").to_string(); + + // Update to v2. + let mut updated_body = body.clone(); + updated_body["name"][0]["family"] = json!("Metadata2"); + server + .put(&format!("/Patient/{id}")) + .add_header(X_TENANT_ID, tenant_header.clone()) + .json(&updated_body) + .await + .assert_status_ok(); + + // Read: current version's meta. + let read: Value = server + .get(&format!("/Patient/{id}")) + .add_header(X_TENANT_ID, tenant_header.clone()) + .await + .json(); + assert_eq!(read["meta"]["versionId"], "2"); + assert!(read["meta"]["lastUpdated"].is_string()); + + // Vread: each version reports its own meta. + let v1: Value = server + .get(&format!("/Patient/{id}/_history/1")) + .add_header(X_TENANT_ID, tenant_header.clone()) + .await + .json(); + assert_eq!(v1["meta"]["versionId"], "1"); + + // History bundle entries carry their version's meta. + let history: Value = server + .get(&format!("/Patient/{id}/_history")) + .add_header(X_TENANT_ID, tenant_header.clone()) + .await + .json(); + let versions: Vec<&str> = history["entry"] + .as_array() + .expect("history entries") + .iter() + .map(|e| e["resource"]["meta"]["versionId"].as_str().unwrap_or("")) + .collect(); + assert!( + versions.contains(&"1") && versions.contains(&"2"), + "history entries must carry per-version meta, got {versions:?}" + ); + + // Search: match entries carry meta. + let search: Value = server + .get("/Patient?family=Metadata2") + .add_header(X_TENANT_ID, tenant_header) + .await + .json(); + let entry = &get_bundle_entries(&search)[0]["resource"]; + assert_eq!(entry["meta"]["versionId"], "2"); + assert!(entry["meta"]["lastUpdated"].is_string()); + } +} From 5c00800a8bb5bff411e89909395e3fc4540eb727 Mon Sep 17 00:00:00 2001 From: angela-helios Date: Wed, 2 Sep 2026 09:08:37 -0400 Subject: [PATCH 2/3] test(postgres): pin the new _lastUpdated precision-range semantics MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit test_prefix_operators asserted the scalar operators the pre-#871 code emitted — the exact semantics the fix replaces. Day-precision prefixes now assert their [day, day+1) range shapes and bound counts, and a full-precision instant pins the scalar fallback. --- crates/persistence/tests/postgres_tests.rs | 60 +++++++++++++++++----- 1 file changed, 46 insertions(+), 14 deletions(-) diff --git a/crates/persistence/tests/postgres_tests.rs b/crates/persistence/tests/postgres_tests.rs index f04e233e71..c4f77e7011 100644 --- a/crates/persistence/tests/postgres_tests.rs +++ b/crates/persistence/tests/postgres_tests.rs @@ -687,17 +687,27 @@ mod query_builder_tests { #[test] fn test_prefix_operators() { - // Test all prefix-to-operator mappings by using date search - let prefixes_and_ops = vec![ - (SearchPrefix::Eq, "="), - (SearchPrefix::Ne, "!="), - (SearchPrefix::Gt, ">"), - (SearchPrefix::Lt, "<"), - (SearchPrefix::Ge, ">="), - (SearchPrefix::Le, "<="), + // Day-precision prefixes compare against the value's [day, day+1) + // range (#871), matching the date-parameter semantics from #463: eq + // must mean "inside the named day", gt must exclude it entirely. + let cases = vec![ + ( + SearchPrefix::Eq, + "last_updated >= $1 AND last_updated < $2", + 2, + ), + ( + SearchPrefix::Ne, + "(last_updated < $1 OR last_updated >= $2)", + 2, + ), + (SearchPrefix::Gt, "last_updated >= $1", 1), + (SearchPrefix::Lt, "last_updated < $1", 1), + (SearchPrefix::Ge, "last_updated >= $1", 1), + (SearchPrefix::Le, "last_updated < $1", 1), ]; - for (prefix, expected_op) in prefixes_and_ops { + for (prefix, expected_sql, expected_params) in cases { let query = SearchQuery::new("Patient").with_parameter(SearchParameter { name: "_lastUpdated".to_string(), param_type: SearchParamType::Date, @@ -711,15 +721,37 @@ mod query_builder_tests { assert!(result.is_some(), "Failed for prefix {:?}", prefix); let fragment = result.unwrap(); assert!( - fragment - .sql - .contains(&format!("last_updated {} $", expected_op)), - "Expected operator '{}' for prefix {:?}, got SQL: {}", - expected_op, + fragment.sql.contains(expected_sql), + "Expected '{}' for prefix {:?}, got SQL: {}", + expected_sql, prefix, fragment.sql ); + assert_eq!( + fragment.params.len(), + expected_params, + "param count for prefix {:?}", + prefix + ); } + + // A full-precision instant is a degenerate range and falls back to + // scalar comparison. + let query = SearchQuery::new("Patient").with_parameter(SearchParameter { + name: "_lastUpdated".to_string(), + param_type: SearchParamType::Date, + modifier: None, + values: vec![SearchValue::new(SearchPrefix::Gt, "2024-01-01T10:00:00Z")], + chain: vec![], + components: vec![], + }); + let fragment = PostgresQueryBuilder::build_search_query(&query, 0) + .expect("full-precision instant must build"); + assert!( + fragment.sql.contains("last_updated > $1"), + "full-precision gt must stay scalar, got SQL: {}", + fragment.sql + ); } } From bdf160c6d598c825e55fac55478b870141301a05 Mon Sep 17 00:00:00 2001 From: Alan Cruz Date: Wed, 2 Sep 2026 15:49:23 -0400 Subject: [PATCH 3/3] test(persistence): make the bulk-submit directives test claim its own manifest MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `postgres_bulk_submit_import_directives_round_trip` called `claim_next_manifest` and asserted the claimed view carried the directives it had just set. The claim queue is cross-tenant and ordered by `added_at`, the test binary shares one container database, and the batch test that #880 added leaves its manifest as `processing` with no lease — which the claim query treats as an orphan to reclaim. Whenever that manifest was added first, the directives test claimed it instead of its own and failed with `left: []`; main's coverage job has been red on most runs since. The test now claims through `claim_specific_manifest`, the submit-side twin of the export tests' `claim_specific`: loop until the target manifest comes back, hold any foreign lease picked up along the way so it cannot be re-claimed, then release those back to the queue. The batch test's comment no longer presents its `process_entries` call as a defence against concurrent claims. No product code changes. Tests: the two submit tests pass five consecutive runs together; the full postgres_tests binary passes (156). --- crates/persistence/tests/postgres_tests.rs | 80 ++++++++++++++++++---- 1 file changed, 65 insertions(+), 15 deletions(-) diff --git a/crates/persistence/tests/postgres_tests.rs b/crates/persistence/tests/postgres_tests.rs index bc9f1fbcf3..99eb060ed2 100644 --- a/crates/persistence/tests/postgres_tests.rs +++ b/crates/persistence/tests/postgres_tests.rs @@ -5959,6 +5959,46 @@ mod postgres_integration { "expected a backend error from an unmigrated database, got {create:?}" ); } + /// Claims manifests in a loop until the lease for `target` comes back, + /// holding any other lease it picks up along the way (so the loop cannot + /// re-claim the same foreign manifest) and returning those to the queue + /// once the target is held. Robust to concurrent tests sharing the + /// testcontainers PostgreSQL instance — the submit-side twin of + /// `claim_specific` for exports. + async fn claim_specific_manifest( + backend: &PostgresBackend, + worker_id: &helios_persistence::core::WorkerId, + submission_id: &helios_persistence::core::SubmissionId, + target_manifest_id: &str, + lease_duration: std::time::Duration, + ) -> helios_persistence::core::ManifestLease { + use helios_persistence::core::SubmitClaimStrategy; + + let mut held = Vec::new(); + let mut found = None; + for _ in 0..100 { + match backend + .claim_next_manifest(worker_id, lease_duration) + .await + .unwrap() + { + Some(lease) + if lease.submission_id == *submission_id + && lease.manifest_id == target_manifest_id => + { + found = Some(lease); + break; + } + Some(other) => held.push(other), + None => tokio::time::sleep(std::time::Duration::from_millis(20)).await, + } + } + for other in held { + let _ = SubmitClaimStrategy::release(backend, other).await; + } + found.expect("never claimed the expected manifest") + } + /// The `import` / `metadata` kickoff directives must survive the PostgreSQL /// round-trip and reach the worker, and `merge` must actually merge. /// @@ -5970,8 +6010,7 @@ mod postgres_integration { async fn postgres_bulk_submit_import_directives_round_trip() { use helios_persistence::core::{ BulkProcessingOptions, BulkSubmitProvider, IMPORT_MODE_PARAMETER_URL, ImportMode, - ManifestFetchParams, NdjsonEntry, SubmissionId, SubmitClaimStrategy, - SubmitWorkerStorage, + ManifestFetchParams, NdjsonEntry, SubmissionId, SubmitWorkerStorage, }; let backend = create_backend().await; @@ -6003,15 +6042,25 @@ mod postgres_integration { .await .unwrap(); - // Claim the manifest the way the worker does, then read its view back. - let lease = backend - .claim_next_manifest( - &helios_persistence::core::WorkerId::new("pg-import-worker"), - std::time::Duration::from_secs(60), - ) - .await - .unwrap() - .expect("claimable manifest"); + // Claim *this* manifest the way the worker does, then read its view + // back. The claim queue is cross-tenant and ordered by `added_at`, and + // the test binary shares one container database, so a plain + // `claim_next_manifest` can hand back another test's manifest — and + // an unleased `processing` manifest (the synchronous `process_entries` + // path leaves one behind) is reclaimable, so "move it out of pending" + // elsewhere is no defence. Loop until ours comes back, returning + // anything else to the queue. + let lease = claim_specific_manifest( + &backend, + &helios_persistence::core::WorkerId::new(format!( + "pg-import-worker-{}", + uuid::Uuid::new_v4() + )), + &sub_id, + &manifest.manifest_id, + std::time::Duration::from_secs(60), + ) + .await; let view = backend.get_manifest_for_worker(&lease).await.unwrap(); assert_eq!(view.import_directives, directives); assert_eq!(view.metadata, metadata); @@ -6087,10 +6136,11 @@ mod postgres_integration { .add_manifest(&tenant, &sub_id, Some("https://provider/b.json"), None) .await .unwrap(); - // Move the manifest out of `pending` right away: the test binary - // shares one container database, and a concurrently running test that - // calls claim_next_manifest would otherwise claim this one (the claim - // queue is cross-tenant by design). + // Mark the manifest `processing` right away. Note this is not a + // defence against other tests' `claim_next_manifest` calls — an + // unleased `processing` manifest is reclaimable, so a claimant may + // still pick this one up transiently — which is why claiming tests use + // `claim_specific_manifest` and hand foreign manifests back. backend .process_entries( &tenant,