From 5805f2c8bc351f25b0c4d5921b92b50488fb8212 Mon Sep 17 00:00:00 2001 From: angela-helios Date: Mon, 21 Sep 2026 16:47:32 -0400 Subject: [PATCH 1/2] fix(s3): keep same-id entries in a batch serial (concurrency correctness) The concurrent ingest path races two entries that target the same resource id: their conditional writes interleave to a non-deterministic final state, where the serial loop resolves them last-write-wins in file order. A batch with any id collision therefore falls back to the serial path; a bulk file usually carries distinct resources, so the common case still parallelizes. Entries with no client id are server-assigned a unique one and never collide. Test: two entries for the same id, second content wins. Flagged by the #945 perf measurement (5.7x at N=16, latency-bound, 7 PUTs/entry). --- .../src/backends/s3/bulk_submit.rs | 21 ++++++- crates/persistence/src/backends/s3/tests.rs | 57 +++++++++++++++++++ 2 files changed, 77 insertions(+), 1 deletion(-) diff --git a/crates/persistence/src/backends/s3/bulk_submit.rs b/crates/persistence/src/backends/s3/bulk_submit.rs index 460e5a083..9927f04e8 100644 --- a/crates/persistence/src/backends/s3/bulk_submit.rs +++ b/crates/persistence/src/backends/s3/bulk_submit.rs @@ -352,8 +352,27 @@ impl BulkSubmitProvider for S3Backend { // entry is captured and propagated after the successful receipts are // reported, matching the serial block's #1078 contract. A cap keeps the // serial early-stop semantics. + // + // Two entries in one batch that target the same resource id are + // order-dependent — last write wins — so processing them concurrently + // would race to a non-deterministic result. A batch with any such id + // collision therefore stays serial (a bulk file usually carries + // distinct resources, so the common case still parallelizes). Entries + // with no client id are server-assigned a unique one and never collide. + let has_id_collision = { + let mut seen = std::collections::HashSet::new(); + !entries + .iter() + .all(|entry| match entry.resource_id.as_deref() { + Some(id) => seen.insert((entry.resource_type.as_str(), id)), + None => true, + }) + }; let concurrency = Self::s3_ingest_concurrency(); - let walked: StorageResult<()> = if options.max_errors == 0 && concurrency > 1 { + let walked: StorageResult<()> = if options.max_errors == 0 + && concurrency > 1 + && !has_id_collision + { use futures::stream::{self, StreamExt}; let outcomes: Vec> = stream::iter(entries) .map(|entry| { diff --git a/crates/persistence/src/backends/s3/tests.rs b/crates/persistence/src/backends/s3/tests.rs index ffad89f91..a26789372 100644 --- a/crates/persistence/src/backends/s3/tests.rs +++ b/crates/persistence/src/backends/s3/tests.rs @@ -1013,6 +1013,63 @@ async fn bulk_submit_concurrent_ingest_preserves_order_and_writes_all() { } } +/// A batch with two entries for the same resource id is order-dependent +/// (last write wins), so it ingests serially even with concurrency enabled — +/// the concurrent path would race them to a non-deterministic result (#945). +#[tokio::test] +async fn bulk_submit_same_id_entries_stay_ordered() { + let mock = Arc::new(MockS3Client::with_buckets(&["test-bucket"])); + let backend = make_prefix_backend(mock); + let tenant = tenant("tenant-a"); + + let submission_id = SubmissionId::new("client-a", "sub-dup"); + backend + .create_submission(&tenant, &submission_id, None) + .await + .unwrap(); + let manifest = backend + .add_manifest(&tenant, &submission_id, None, None) + .await + .unwrap(); + + // Same id, different content: the second entry must be the one that lands. + let entries = vec![ + NdjsonEntry::new( + 1, + "Patient", + json!({"resourceType": "Patient", "id": "dup", "gender": "male"}), + ), + NdjsonEntry::new( + 2, + "Patient", + json!({"resourceType": "Patient", "id": "dup", "gender": "female"}), + ), + ]; + + let results = backend + .process_entries( + &tenant, + &submission_id, + &manifest.manifest_id, + entries, + &BulkProcessingOptions::new(), + ) + .await + .unwrap(); + assert_eq!(results.len(), 2); + + let stored = backend + .read(&tenant, "Patient", "dup") + .await + .unwrap() + .expect("the resource is stored"); + assert_eq!( + stored.content()["gender"], + "female", + "the last write in the batch must win" + ); +} + /// Two output files of one manifest, both starting at line 1, must each keep /// their own entry result and raw archive (issue #457). /// From 2c4aaca904c2721fd3ed15124c39ed0963f6b675 Mon Sep 17 00:00:00 2001 From: angela-helios Date: Mon, 21 Sep 2026 17:17:49 -0400 Subject: [PATCH 2/2] perf(s3): coalesce the raw NDJSON archive to one object per batch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The per-entry raw-line PUT was one of the 7 PUTs/resource the #1429 measurement found the ingest is bound by, and the raw archive has zero production readers — it is a write-only auditable copy of the input. Write it once per batch instead: persist_raw_batch archives the whole chunk's NDJSON in a single object, keyed by the batch's first line number (submit_raw_batch_key) so successive chunks of one file never collide, and still discriminated by file_url like the receipts (#457). Written upfront, before any entry processing, so a failure aborts before anything is stored. Drops the raw PUT from 1/entry to 1/batch — 7->6 PUTs per resource, and one of the follow-ups the #1429 phase breakdown pointed at (raw + change + receipt are the coalescible three; change and receipt have readers and come next). Tests: one raw object per batch holding every line, and the #457 two-file archive discrimination updated to the batch key. --- .../src/backends/s3/bulk_submit.rs | 61 ++++++++++------- .../persistence/src/backends/s3/keyspace.rs | 13 +++- crates/persistence/src/backends/s3/tests.rs | 66 ++++++++++++++++++- 3 files changed, 113 insertions(+), 27 deletions(-) diff --git a/crates/persistence/src/backends/s3/bulk_submit.rs b/crates/persistence/src/backends/s3/bulk_submit.rs index 9927f04e8..f76f1a39b 100644 --- a/crates/persistence/src/backends/s3/bulk_submit.rs +++ b/crates/persistence/src/backends/s3/bulk_submit.rs @@ -339,6 +339,14 @@ impl BulkSubmitProvider for S3Backend { let mut error_count = 0u32; let file_url = options.file_url.as_deref(); + // Archive the batch's raw NDJSON in one object upfront — the input is + // preserved in full before any processing, and the per-entry raw PUT is + // gone (#1429). A failure here aborts before any entry is written, so + // the #1078 "report what was written" contract is trivially satisfied + // (nothing was). + self.persist_raw_batch(&location, submission_id, manifest_id, file_url, &entries) + .await?; + // S3 writes each entry on its own, so an entry is durable as soon as // its write returns. The loop runs in a block so that however it ends — // exhausted, max errors reached, or a storage error part-way — the @@ -437,9 +445,8 @@ impl BulkSubmitProvider for S3Backend { continue; } - self.persist_raw_entry(&location, submission_id, manifest_id, file_url, &entry) - .await?; - + // The raw line was archived for the whole batch upfront + // (persist_raw_batch), so the loop no longer PUTs it per entry. let result = match self .process_single_entry(tenant, submission_id, manifest_id, &entry, options) .await @@ -852,8 +859,7 @@ impl S3Backend { entry: NdjsonEntry, options: &BulkProcessingOptions, ) -> StorageResult { - self.persist_raw_entry(location, submission_id, manifest_id, file_url, &entry) - .await?; + // The raw line was archived for the whole batch upfront (persist_raw_batch). let result = match self .process_single_entry(tenant, submission_id, manifest_id, &entry, options) .await @@ -1003,40 +1009,51 @@ impl S3Backend { } } - /// Archives the raw NDJSON payload for a single entry to S3. + /// Archives the raw NDJSON of one ingest batch to S3 in a single object — + /// every line of the batch, keyed by its first line number. /// - /// Stored under `raw///.ndjson` so that the original - /// data is preserved for auditing after ingestion. `file_url` is the - /// manifest output file the line came from, and is required for the same - /// reason it is on [`Self::persist_entry_result`]. - async fn persist_raw_entry( + /// Stored under `raw///batch-.ndjson` so the + /// original data is preserved for auditing after ingestion. This is written + /// once per batch rather than once per entry: the archive has no reader, so + /// coalescing it drops one PUT per resource with no read contract to + /// preserve (#1429). `file_url` is the manifest output file the lines came + /// from, and discriminates otherwise-colliding batches for the same reason + /// it is on [`Self::persist_entry_result`]. An empty batch writes nothing. + async fn persist_raw_batch( &self, location: &TenantLocation, submission_id: &SubmissionId, manifest_id: &str, file_url: Option<&str>, - entry: &NdjsonEntry, + entries: &[NdjsonEntry], ) -> StorageResult<()> { - let key = location.keyspace.submit_raw_line_key( + let Some(first) = entries.first() else { + return Ok(()); + }; + let key = location.keyspace.submit_raw_batch_key( &submission_id.submitter, &submission_id.submission_id, manifest_id, file_url, - entry.line_number, + first.line_number, ); - let mut line = serde_json::to_string(&entry.resource).map_err(|e| { - StorageError::BulkSubmit(BulkSubmitError::ParseError { - line: entry.line_number, - message: format!("failed to serialize raw NDJSON entry: {e}"), - }) - })?; - line.push('\n'); + let mut body = String::new(); + for entry in entries { + let line = serde_json::to_string(&entry.resource).map_err(|e| { + StorageError::BulkSubmit(BulkSubmitError::ParseError { + line: entry.line_number, + message: format!("failed to serialize raw NDJSON entry: {e}"), + }) + })?; + body.push_str(&line); + body.push('\n'); + } self.put_bytes_object( &location.bucket, &key, - line.as_bytes(), + body.as_bytes(), Some("application/fhir+ndjson"), ) .await?; diff --git a/crates/persistence/src/backends/s3/keyspace.rs b/crates/persistence/src/backends/s3/keyspace.rs index 6fb757883..c0dfe3a26 100644 --- a/crates/persistence/src/backends/s3/keyspace.rs +++ b/crates/persistence/src/backends/s3/keyspace.rs @@ -275,13 +275,20 @@ impl S3Keyspace { /// /// `file_url` names the manifest output file the line came from; see /// [`submit_file_segment`] for why it is part of the key. - pub fn submit_raw_line_key( + /// Key for the raw NDJSON archive of one ingest batch (chunk) of a file. + /// + /// One object holds every line of the batch, keyed by the batch's first + /// line number so successive chunks of the same file (each a separate + /// `process_entries` call) never collide (#1429). `file_url` names the + /// manifest output file the lines came from; see [`submit_file_segment`] + /// for why it is part of the key (two files' line-1 batches must differ). + pub fn submit_raw_batch_key( &self, submitter: &str, submission_id: &str, manifest_id: &str, file_url: Option<&str>, - line: u64, + first_line: u64, ) -> String { self.join(&[ "bulk", @@ -291,7 +298,7 @@ impl S3Keyspace { "raw", manifest_id, &submit_file_segment(file_url), - &format!("line-{}.ndjson", line), + &format!("batch-{}.ndjson", first_line), ]) } diff --git a/crates/persistence/src/backends/s3/tests.rs b/crates/persistence/src/backends/s3/tests.rs index a26789372..023ea111b 100644 --- a/crates/persistence/src/backends/s3/tests.rs +++ b/crates/persistence/src/backends/s3/tests.rs @@ -1013,6 +1013,67 @@ async fn bulk_submit_concurrent_ingest_preserves_order_and_writes_all() { } } +/// The raw NDJSON archive is one object per batch holding every line (#1429), +/// not one PUT per entry — the coalescing that drops a PUT per resource. +#[tokio::test] +async fn bulk_submit_raw_archive_is_one_object_per_batch() { + let mock = Arc::new(MockS3Client::with_buckets(&["test-bucket"])); + let backend = make_prefix_backend(mock.clone()); + let tenant = tenant("tenant-a"); + + let submission_id = SubmissionId::new("client-a", "sub-raw"); + backend + .create_submission(&tenant, &submission_id, None) + .await + .unwrap(); + let manifest = backend + .add_manifest(&tenant, &submission_id, None, None) + .await + .unwrap(); + + let entries: Vec = (1..=3) + .map(|i| { + NdjsonEntry::new( + i, + "Patient", + json!({"resourceType": "Patient", "id": format!("r{i}")}), + ) + }) + .collect(); + backend + .process_entries( + &tenant, + &submission_id, + &manifest.manifest_id, + entries, + &BulkProcessingOptions::new(), + ) + .await + .unwrap(); + + let raw_puts = mock + .recorded_puts() + .into_iter() + .filter(|put| put.key.contains("/raw/")) + .count(); + assert_eq!( + raw_puts, 1, + "the three-entry batch must archive its raw NDJSON in one object, not three" + ); + + // The resources themselves are still all stored. + for i in 1..=3 { + assert!( + backend + .read(&tenant, "Patient", &format!("r{i}")) + .await + .unwrap() + .is_some(), + "Patient/r{i} should be stored" + ); + } +} + /// A batch with two entries for the same resource id is order-dependent /// (last write wins), so it ingests serially even with concurrency enabled — /// the concurrent path would race them to a non-deterministic result (#945). @@ -1244,12 +1305,13 @@ async fn bulk_submit_entry_results_are_keyed_by_their_output_file() { ); // The raw NDJSON archive is discriminated too, so the auditable copy of the - // first file's payload is not replaced by the second's. + // first file's payload is not replaced by the second's. Coalesced to one + // batch object per file (#1429), keyed by the batch's first line. let raw_keys: Vec = mock .recorded_puts() .into_iter() .map(|put| put.key) - .filter(|key| key.contains("/raw/") && key.ends_with("line-1.ndjson")) + .filter(|key| key.contains("/raw/") && key.ends_with("batch-1.ndjson")) .collect(); assert_eq!(raw_keys.len(), 2, "one raw archive put per file"); assert_ne!(