diff --git a/crates/persistence/src/backends/postgres/bulk_submit.rs b/crates/persistence/src/backends/postgres/bulk_submit.rs index 10390b342..6ed831776 100644 --- a/crates/persistence/src/backends/postgres/bulk_submit.rs +++ b/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, diff --git a/crates/persistence/src/backends/postgres/schema.rs b/crates/persistence/src/backends/postgres/schema.rs index f6862977c..680186bcf 100644 --- a/crates/persistence/src/backends/postgres/schema.rs +++ b/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 diff --git a/crates/persistence/src/backends/sqlite/bulk_submit.rs b/crates/persistence/src/backends/sqlite/bulk_submit.rs index 6ccef110b..0f906004f 100644 --- a/crates/persistence/src/backends/sqlite/bulk_submit.rs +++ b/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(); diff --git a/crates/persistence/src/backends/sqlite/schema.rs b/crates/persistence/src/backends/sqlite/schema.rs index 86fcaba60..f63d45aef 100644 --- a/crates/persistence/src/backends/sqlite/schema.rs +++ b/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(), diff --git a/crates/persistence/src/core/bulk_submit.rs b/crates/persistence/src/core/bulk_submit.rs index 573c66425..a4d09e661 100644 --- a/crates/persistence/src/core/bulk_submit.rs +++ b/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, } 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) -> 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; diff --git a/crates/persistence/src/core/bulk_submit_worker.rs b/crates/persistence/src/core/bulk_submit_worker.rs index fcc6fd8bf..27fb7a190 100644 --- a/crates/persistence/src/core/bulk_submit_worker.rs +++ b/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 {