diff --git a/crates/persistence/src/backends/s3/bulk_submit.rs b/crates/persistence/src/backends/s3/bulk_submit.rs index 460e5a083..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 @@ -352,8 +360,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| { @@ -418,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 @@ -833,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 @@ -984,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 ffad89f91..023ea111b 100644 --- a/crates/persistence/src/backends/s3/tests.rs +++ b/crates/persistence/src/backends/s3/tests.rs @@ -1013,6 +1013,124 @@ 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). +#[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). /// @@ -1187,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!(