fix(s3): re-land the same-id concurrency guard + raw coalescing dropped from #1411

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<StorageResult<BulkEntryResult>> = 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<BulkEntryResult> {
-        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/<manifest>/<file>/<line>.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/<manifest>/<file>/batch-<first_line>.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?;

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),
         ])
     }
 

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<NdjsonEntry> = (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<String> = 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!(