fix(bulk-submit): key entry results by their manifest output file

crates/persistence/src/backends/postgres/bulk_submit.rs

@@ -685,8 +685,14 @@ impl BulkSubmitProvider for PostgresBackend {
                 error_count += 1;
             }
 
-            self.store_entry_result(tenant, submission_id, manifest_id, &entry_result)
-                .await?;
+            self.store_entry_result(
+                tenant,
+                submission_id,
+                manifest_id,
+                options.file_url.as_deref().unwrap_or(""),
+                &entry_result,
+            )
+            .await?;
 
             results.push(entry_result);
         }
@@ -956,6 +962,7 @@ impl PostgresBackend {
         tenant: &TenantContext,
         submission_id: &SubmissionId,
         manifest_id: &str,
+        file_url: &str,
         result: &BulkEntryResult,
     ) -> StorageResult<()> {
         let client = self.get_client().await?;
@@ -965,14 +972,24 @@ impl PostgresBackend {
 
         client
             .execute(
+                // Upsert: the worker re-fetches a whole file after a transient
+                // failure, and the retry must overwrite its own earlier rows
+                // instead of colliding with them (#457).
                 "INSERT INTO bulk_entry_results
-                 (tenant_id, submitter, submission_id, manifest_id, line_number, resource_type, resource_id, created, outcome, operation_outcome)
-                 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)",
+                 (tenant_id, submitter, submission_id, manifest_id, file_url, line_number, resource_type, resource_id, created, outcome, operation_outcome)
+                 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
+                 ON CONFLICT (tenant_id, submitter, submission_id, manifest_id, file_url, line_number)
+                 DO UPDATE SET resource_type = EXCLUDED.resource_type,
+                               resource_id = EXCLUDED.resource_id,
+                               created = EXCLUDED.created,
+                               outcome = EXCLUDED.outcome,
+                               operation_outcome = EXCLUDED.operation_outcome",
                 &[
                     &tenant_id,
                     &submission_id.submitter.as_str(),
                     &submission_id.submission_id.as_str(),
                     &manifest_id,
+                    &file_url,
                     &(result.line_number as i32),
                     &result.resource_type.as_str(),
                     &result.resource_id,

crates/persistence/src/backends/postgres/schema.rs

@@ -536,13 +536,14 @@ async fn migrate_v5_to_v6(client: &deadpool_postgres::Client) -> StorageResult<(
                 submitter TEXT NOT NULL,
                 submission_id TEXT NOT NULL,
                 manifest_id TEXT NOT NULL,
+                file_url TEXT NOT NULL DEFAULT '',
                 line_number INTEGER NOT NULL,
                 resource_type TEXT NOT NULL,
                 resource_id TEXT,
                 created BOOLEAN,
                 outcome TEXT NOT NULL,
                 operation_outcome JSONB,
-                PRIMARY KEY (tenant_id, submitter, submission_id, manifest_id, line_number),
+                PRIMARY KEY (tenant_id, submitter, submission_id, manifest_id, file_url, line_number),
                 FOREIGN KEY (tenant_id, submitter, submission_id, manifest_id)
                     REFERENCES bulk_manifests(tenant_id, submitter, submission_id, manifest_id) ON DELETE CASCADE
             )",
@@ -551,6 +552,36 @@ async fn migrate_v5_to_v6(client: &deadpool_postgres::Client) -> StorageResult<(
         .await
         .map_err(|e| pg_error(format!("Failed to create bulk_entry_results table: {}", e)))?;
 
+    // #457 migration for pre-existing tables: line numbers restart per output
+    // file, so the file belongs in the key — without it every file after the
+    // first collided on its first entry. Pre-migration rows keep ''.
+    client
+        .execute(
+            "ALTER TABLE bulk_entry_results ADD COLUMN IF NOT EXISTS file_url TEXT NOT NULL DEFAULT ''",
+            &[],
+        )
+        .await
+        .map_err(|e| pg_error(format!("Failed to add bulk_entry_results.file_url: {}", e)))?;
+    client
+        .execute(
+            "DO $$
+             BEGIN
+               IF NOT EXISTS (
+                 SELECT 1 FROM information_schema.key_column_usage
+                 WHERE table_name = 'bulk_entry_results'
+                   AND constraint_name = 'bulk_entry_results_pkey'
+                   AND column_name = 'file_url'
+               ) THEN
+                 ALTER TABLE bulk_entry_results DROP CONSTRAINT bulk_entry_results_pkey;
+                 ALTER TABLE bulk_entry_results ADD PRIMARY KEY
+                   (tenant_id, submitter, submission_id, manifest_id, file_url, line_number);
+               END IF;
+             END $$",
+            &[],
+        )
+        .await
+        .map_err(|e| pg_error(format!("Failed to rekey bulk_entry_results: {}", e)))?;
+
     client
         .execute(
             "CREATE INDEX IF NOT EXISTS idx_bulk_entry_results_outcome

crates/persistence/src/backends/sqlite/bulk_submit.rs

@@ -671,9 +671,16 @@ impl BulkSubmitProvider for SqliteBackend {
                 error_count += 1;
             }
 
-            // Store the result
-            self.store_entry_result(tenant, submission_id, manifest_id, &entry_result)
-                .await?;
+            // Store the result, keyed by the file it came from (#457): line
+            // numbers restart per file, so the file is part of the identity.
+            self.store_entry_result(
+                tenant,
+                submission_id,
+                manifest_id,
+                options.file_url.as_deref().unwrap_or(""),
+                &entry_result,
+            )
+            .await?;
 
             results.push(entry_result);
         }
@@ -937,6 +944,7 @@ impl SqliteBackend {
         tenant: &TenantContext,
         submission_id: &SubmissionId,
         manifest_id: &str,
+        file_url: &str,
         result: &BulkEntryResult,
     ) -> StorageResult<()> {
         let conn = self.get_connection()?;
@@ -947,15 +955,19 @@ impl SqliteBackend {
             .as_ref()
             .and_then(|o| serde_json::to_vec(o).ok());
 
+        // OR REPLACE: the worker re-fetches a whole file after a transient
+        // failure, and the retry must overwrite its own earlier rows instead
+        // of colliding with them (#457).
         conn.execute(
-            "INSERT INTO bulk_entry_results
-             (tenant_id, submitter, submission_id, manifest_id, line_number, resource_type, resource_id, created, outcome, operation_outcome)
-             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
+            "INSERT OR REPLACE INTO bulk_entry_results
+             (tenant_id, submitter, submission_id, manifest_id, file_url, line_number, resource_type, resource_id, created, outcome, operation_outcome)
+             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)",
             params![
                 tenant_id,
                 &submission_id.submitter,
                 &submission_id.submission_id,
                 manifest_id,
+                file_url,
                 result.line_number as i64,
                 &result.resource_type,
                 &result.resource_id,
@@ -2054,6 +2066,49 @@ mod tests {
         assert!(results.iter().all(|r| r.created));
     }
 
+    /// #457: a manifest with several output files restarts line numbers in
+    /// each file, so the stored entry-result key must include the file — the
+    /// old key collided on every file after the first.
+    #[tokio::test]
+    async fn test_process_entries_from_multiple_files_do_not_collide() {
+        let backend = create_test_backend();
+        let tenant = create_test_tenant();
+
+        let sub_id = SubmissionId::generate("test-system");
+        backend
+            .create_submission(&tenant, &sub_id, None)
+            .await
+            .unwrap();
+        let manifest = backend
+            .add_manifest(&tenant, &sub_id, None, None)
+            .await
+            .unwrap();
+
+        for (file, family) in [
+            ("http://provider/Patient.ndjson", "FromPatients"),
+            ("http://provider/Practitioner.ndjson", "FromPractitioners"),
+        ] {
+            // Both files start at line 1 — the collision of the old key.
+            let entries = vec![NdjsonEntry::new(
+                1,
+                "Patient",
+                json!({"resourceType": "Patient", "name": [{"family": family}]}),
+            )];
+            let options = BulkProcessingOptions::new().with_file_url(file);
+            let results = backend
+                .process_entries(&tenant, &sub_id, &manifest.manifest_id, entries, &options)
+                .await
+                .unwrap_or_else(|e| panic!("file {file} must ingest: {e}"));
+            assert!(results.iter().all(|r| r.is_success()), "{file}");
+        }
+
+        let counts = backend
+            .get_entry_counts(&tenant, &sub_id, &manifest.manifest_id)
+            .await
+            .unwrap();
+        assert_eq!(counts.success, 2, "one stored entry result per file");
+    }
+
     #[tokio::test]
     async fn test_complete_submission() {
         let backend = create_test_backend();

crates/persistence/src/backends/sqlite/schema.rs

@@ -5,7 +5,7 @@ use rusqlite::Connection;
 use crate::error::StorageResult;
 
 /// Current schema version.
-pub const SCHEMA_VERSION: i32 = 14;
+pub const SCHEMA_VERSION: i32 = 15;
 
 /// Initialize the database schema.
 pub fn initialize_schema(conn: &Connection) -> StorageResult<()> {
@@ -295,6 +295,7 @@ fn migrate_schema(conn: &Connection, from_version: i32) -> StorageResult<()> {
             11 => migrate_v11_to_v12(conn)?,
             12 => migrate_v12_to_v13(conn)?,
             13 => migrate_v13_to_v14(conn)?,
+            14 => migrate_v14_to_v15(conn)?,
             _ => {
                 return Err(crate::error::StorageError::Backend(
                     crate::error::BackendError::Internal {
@@ -747,13 +748,14 @@ fn migrate_v5_to_v6(conn: &Connection) -> StorageResult<()> {
             submitter TEXT NOT NULL,
             submission_id TEXT NOT NULL,
             manifest_id TEXT NOT NULL,
+            file_url TEXT NOT NULL DEFAULT '',
             line_number INTEGER NOT NULL,
             resource_type TEXT NOT NULL,
             resource_id TEXT,
             created INTEGER,
             outcome TEXT NOT NULL,
             operation_outcome BLOB,
-            PRIMARY KEY (tenant_id, submitter, submission_id, manifest_id, line_number),
+            PRIMARY KEY (tenant_id, submitter, submission_id, manifest_id, file_url, line_number),
             FOREIGN KEY (tenant_id, submitter, submission_id, manifest_id)
                 REFERENCES bulk_manifests(tenant_id, submitter, submission_id, manifest_id) ON DELETE CASCADE
         )",
@@ -1258,6 +1260,41 @@ fn migrate_v13_to_v14(conn: &Connection) -> StorageResult<()> {
     ensure_tenants_table(conn)
 }
 
+/// v14 -> v15 migration: `bulk_entry_results` gains `file_url` in its primary
+/// key (#457). Line numbers restart in every manifest output file, so without
+/// the file in the key every file after the first collided on its first entry.
+/// SQLite cannot alter a primary key, so the table is rebuilt; pre-migration
+/// rows keep an empty file_url.
+fn migrate_v14_to_v15(conn: &Connection) -> StorageResult<()> {
+    conn.execute_batch(
+        "CREATE TABLE bulk_entry_results_v15 (
+            tenant_id TEXT NOT NULL,
+            submitter TEXT NOT NULL,
+            submission_id TEXT NOT NULL,
+            manifest_id TEXT NOT NULL,
+            file_url TEXT NOT NULL DEFAULT '',
+            line_number INTEGER NOT NULL,
+            resource_type TEXT NOT NULL,
+            resource_id TEXT,
+            created INTEGER,
+            outcome TEXT NOT NULL,
+            operation_outcome BLOB,
+            PRIMARY KEY (tenant_id, submitter, submission_id, manifest_id, file_url, line_number),
+            FOREIGN KEY (tenant_id, submitter, submission_id, manifest_id)
+                REFERENCES bulk_manifests(tenant_id, submitter, submission_id, manifest_id) ON DELETE CASCADE
+        );
+        INSERT INTO bulk_entry_results_v15
+            (tenant_id, submitter, submission_id, manifest_id, line_number, resource_type, resource_id, created, outcome, operation_outcome)
+        SELECT tenant_id, submitter, submission_id, manifest_id, line_number, resource_type, resource_id, created, outcome, operation_outcome
+        FROM bulk_entry_results;
+        DROP TABLE bulk_entry_results;
+        ALTER TABLE bulk_entry_results_v15 RENAME TO bulk_entry_results;
+        CREATE INDEX IF NOT EXISTS idx_bulk_entry_results_outcome
+            ON bulk_entry_results(tenant_id, submitter, submission_id, manifest_id, outcome);",
+    )
+    .map_err(|e| migration_err(format!("migrate bulk_entry_results to v15: {e}")))
+}
+
 fn migration_err(message: String) -> crate::error::StorageError {
     crate::error::StorageError::Backend(crate::error::BackendError::Internal {
         backend_name: "sqlite".to_string(),

crates/persistence/src/core/bulk_submit.rs

@@ -674,6 +674,11 @@ pub struct BulkProcessingOptions {
     /// How a submitted resource is applied over an existing one with the same id.
     #[serde(default)]
     pub import_mode: ImportMode,
+    /// The manifest output file the processed entries come from. Part of each
+    /// entry result's identity: line numbers restart per file, so without the
+    /// file every file after the first collides on the stored key (#457).
+    #[serde(default)]
+    pub file_url: Option<String>,
 }
 
 fn default_submit_batch_size() -> u32 {
@@ -703,9 +708,16 @@ impl BulkProcessingOptions {
             max_errors: 0,
             allow_updates: default_allow_updates(),
             import_mode: ImportMode::default(),
+            file_url: None,
         }
     }
 
+    /// Names the manifest output file the entries come from (#457).
+    pub fn with_file_url(mut self, file_url: impl Into<String>) -> Self {
+        self.file_url = Some(file_url.into());
+        self
+    }
+
     /// Sets the batch size.
     pub fn with_batch_size(mut self, batch_size: u32) -> Self {
         self.batch_size = batch_size;

crates/persistence/src/core/bulk_submit_worker.rs

@@ -468,6 +468,9 @@ where
                 }
             };
 
+            // Per-file options: the file url is part of every entry result's
+            // identity, since line numbers restart in each file (#457).
+            let file_opts = opts.clone().with_file_url(&file.url);
             match self
                 .jobs
                 .process_ndjson_stream(
@@ -476,7 +479,7 @@ where
                     &lease.manifest_id,
                     &resource_type,
                     stream,
-                    &opts,
+                    &file_opts,
                 )
                 .await
             {