diff --git a/common/proto/tables_data.proto b/common/proto/tables_data.proto index 2be621c3..ed69872b 100644 --- a/common/proto/tables_data.proto +++ b/common/proto/tables_data.proto @@ -29,7 +29,7 @@ service TablesData { rpc PostAccountingTableData(PostAccountingTableDataRequest) returns (PostTableDataResponse); - // Insert multiple rows by applying PostTableData behavior to each row. + // Insert multiple rows atomically by applying PostTableData behavior to each row. // // Behavior: // - Accepts 1..10,000 rows in one gRPC request @@ -37,9 +37,27 @@ service TablesData { // - Each row is inserted through the same validation, script execution, // typed binding, database insert, and indexing path as PostTableData // - Stops at the first failing row and returns that row's gRPC error code - // with row index context; rows inserted before the failure remain inserted + // - Commits only when every row succeeds; a failure rolls the entire request back rpc PostTableDataBulk(PostTableDataBulkRequest) returns (PostTableDataBulkResponse); + // Starts a durable, profile-scoped import staging session. Staging never changes + // profile data; CommitTableDataImport applies every staged chunk atomically. + rpc BeginTableDataImport(BeginTableDataImportRequest) + returns (BeginTableDataImportResponse); + + // Adds one table chunk to an import session. Rows retain chunk and table order. + rpc StageTableDataImport(StageTableDataImportRequest) + returns (StageTableDataImportResponse); + + // Applies every staged row through the ordinary validated insert machinery in + // one PostgreSQL transaction. Any failure rolls back every imported side effect. + rpc CommitTableDataImport(CommitTableDataImportRequest) + returns (CommitTableDataImportResponse); + + // Discards a staged import. Already committed imports cannot be aborted. + rpc AbortTableDataImport(AbortTableDataImportRequest) + returns (AbortTableDataImportResponse); + // Update existing row data with strict type binding and script validation. // // Behavior: @@ -201,6 +219,42 @@ message PostTableDataBulkResponse { repeated PostTableDataResponse responses = 3; } +message BeginTableDataImportRequest { + string profile_name = 1; +} + +message BeginTableDataImportResponse { + string import_id = 1; +} + +message StageTableDataImportRequest { + string import_id = 1; + string table_name = 2; + repeated PostTableDataBulkRow rows = 3; +} + +message StageTableDataImportResponse { + int64 staged_rows = 1; + int64 total_staged_rows = 2; +} + +message CommitTableDataImportRequest { + string import_id = 1; +} + +message CommitTableDataImportResponse { + bool success = 1; + int64 inserted_rows = 2; +} + +message AbortTableDataImportRequest { + string import_id = 1; +} + +message AbortTableDataImportResponse { + bool success = 1; +} + // Update an existing row. message PutTableDataRequest { // Required. Profile (schema) name. diff --git a/common/src/proto/descriptor.bin b/common/src/proto/descriptor.bin index ddd598db..8035716c 100644 Binary files a/common/src/proto/descriptor.bin and b/common/src/proto/descriptor.bin differ diff --git a/common/src/proto/komp_ac.tables_data.rs b/common/src/proto/komp_ac.tables_data.rs index 2fd878ab..9be0e14b 100644 --- a/common/src/proto/komp_ac.tables_data.rs +++ b/common/src/proto/komp_ac.tables_data.rs @@ -113,6 +113,54 @@ pub struct PostTableDataBulkResponse { #[prost(message, repeated, tag = "3")] pub responses: ::prost::alloc::vec::Vec, } +#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +pub struct BeginTableDataImportRequest { + #[prost(string, tag = "1")] + pub profile_name: ::prost::alloc::string::String, +} +#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +pub struct BeginTableDataImportResponse { + #[prost(string, tag = "1")] + pub import_id: ::prost::alloc::string::String, +} +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct StageTableDataImportRequest { + #[prost(string, tag = "1")] + pub import_id: ::prost::alloc::string::String, + #[prost(string, tag = "2")] + pub table_name: ::prost::alloc::string::String, + #[prost(message, repeated, tag = "3")] + pub rows: ::prost::alloc::vec::Vec, +} +#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)] +pub struct StageTableDataImportResponse { + #[prost(int64, tag = "1")] + pub staged_rows: i64, + #[prost(int64, tag = "2")] + pub total_staged_rows: i64, +} +#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +pub struct CommitTableDataImportRequest { + #[prost(string, tag = "1")] + pub import_id: ::prost::alloc::string::String, +} +#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)] +pub struct CommitTableDataImportResponse { + #[prost(bool, tag = "1")] + pub success: bool, + #[prost(int64, tag = "2")] + pub inserted_rows: i64, +} +#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +pub struct AbortTableDataImportRequest { + #[prost(string, tag = "1")] + pub import_id: ::prost::alloc::string::String, +} +#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)] +pub struct AbortTableDataImportResponse { + #[prost(bool, tag = "1")] + pub success: bool, +} /// Update an existing row. #[derive(Clone, PartialEq, ::prost::Message)] pub struct PutTableDataRequest { @@ -583,7 +631,7 @@ pub mod tables_data_client { ); self.inner.unary(req, path, codec).await } - /// Insert multiple rows by applying PostTableData behavior to each row. + /// Insert multiple rows atomically by applying PostTableData behavior to each row. /// /// Behavior: /// @@ -592,7 +640,7 @@ pub mod tables_data_client { /// * Each row is inserted through the same validation, script execution, /// typed binding, database insert, and indexing path as PostTableData /// * Stops at the first failing row and returns that row's gRPC error code - /// with row index context; rows inserted before the failure remain inserted + /// * Commits only when every row succeeds; a failure rolls the entire request back pub async fn post_table_data_bulk( &mut self, request: impl tonic::IntoRequest, @@ -622,6 +670,128 @@ pub mod tables_data_client { ); self.inner.unary(req, path, codec).await } + /// Starts a durable, profile-scoped import staging session. Staging never changes + /// profile data; CommitTableDataImport applies every staged chunk atomically. + pub async fn begin_table_data_import( + &mut self, + request: impl tonic::IntoRequest, + ) -> std::result::Result< + tonic::Response, + tonic::Status, + > { + self.inner + .ready() + .await + .map_err(|e| { + tonic::Status::unknown( + format!("Service was not ready: {}", e.into()), + ) + })?; + let codec = tonic_prost::ProstCodec::default(); + let path = http::uri::PathAndQuery::from_static( + "/komp_ac.tables_data.TablesData/BeginTableDataImport", + ); + let mut req = request.into_request(); + req.extensions_mut() + .insert( + GrpcMethod::new( + "komp_ac.tables_data.TablesData", + "BeginTableDataImport", + ), + ); + self.inner.unary(req, path, codec).await + } + /// Adds one table chunk to an import session. Rows retain chunk and table order. + pub async fn stage_table_data_import( + &mut self, + request: impl tonic::IntoRequest, + ) -> std::result::Result< + tonic::Response, + tonic::Status, + > { + self.inner + .ready() + .await + .map_err(|e| { + tonic::Status::unknown( + format!("Service was not ready: {}", e.into()), + ) + })?; + let codec = tonic_prost::ProstCodec::default(); + let path = http::uri::PathAndQuery::from_static( + "/komp_ac.tables_data.TablesData/StageTableDataImport", + ); + let mut req = request.into_request(); + req.extensions_mut() + .insert( + GrpcMethod::new( + "komp_ac.tables_data.TablesData", + "StageTableDataImport", + ), + ); + self.inner.unary(req, path, codec).await + } + /// Applies every staged row through the ordinary validated insert machinery in + /// one PostgreSQL transaction. Any failure rolls back every imported side effect. + pub async fn commit_table_data_import( + &mut self, + request: impl tonic::IntoRequest, + ) -> std::result::Result< + tonic::Response, + tonic::Status, + > { + self.inner + .ready() + .await + .map_err(|e| { + tonic::Status::unknown( + format!("Service was not ready: {}", e.into()), + ) + })?; + let codec = tonic_prost::ProstCodec::default(); + let path = http::uri::PathAndQuery::from_static( + "/komp_ac.tables_data.TablesData/CommitTableDataImport", + ); + let mut req = request.into_request(); + req.extensions_mut() + .insert( + GrpcMethod::new( + "komp_ac.tables_data.TablesData", + "CommitTableDataImport", + ), + ); + self.inner.unary(req, path, codec).await + } + /// Discards a staged import. Already committed imports cannot be aborted. + pub async fn abort_table_data_import( + &mut self, + request: impl tonic::IntoRequest, + ) -> std::result::Result< + tonic::Response, + tonic::Status, + > { + self.inner + .ready() + .await + .map_err(|e| { + tonic::Status::unknown( + format!("Service was not ready: {}", e.into()), + ) + })?; + let codec = tonic_prost::ProstCodec::default(); + let path = http::uri::PathAndQuery::from_static( + "/komp_ac.tables_data.TablesData/AbortTableDataImport", + ); + let mut req = request.into_request(); + req.extensions_mut() + .insert( + GrpcMethod::new( + "komp_ac.tables_data.TablesData", + "AbortTableDataImport", + ), + ); + self.inner.unary(req, path, codec).await + } /// Update existing row data with strict type binding and script validation. /// /// Behavior: @@ -992,7 +1162,7 @@ pub mod tables_data_server { tonic::Response, tonic::Status, >; - /// Insert multiple rows by applying PostTableData behavior to each row. + /// Insert multiple rows atomically by applying PostTableData behavior to each row. /// /// Behavior: /// @@ -1001,7 +1171,7 @@ pub mod tables_data_server { /// * Each row is inserted through the same validation, script execution, /// typed binding, database insert, and indexing path as PostTableData /// * Stops at the first failing row and returns that row's gRPC error code - /// with row index context; rows inserted before the failure remain inserted + /// * Commits only when every row succeeds; a failure rolls the entire request back async fn post_table_data_bulk( &self, request: tonic::Request, @@ -1009,6 +1179,40 @@ pub mod tables_data_server { tonic::Response, tonic::Status, >; + /// Starts a durable, profile-scoped import staging session. Staging never changes + /// profile data; CommitTableDataImport applies every staged chunk atomically. + async fn begin_table_data_import( + &self, + request: tonic::Request, + ) -> std::result::Result< + tonic::Response, + tonic::Status, + >; + /// Adds one table chunk to an import session. Rows retain chunk and table order. + async fn stage_table_data_import( + &self, + request: tonic::Request, + ) -> std::result::Result< + tonic::Response, + tonic::Status, + >; + /// Applies every staged row through the ordinary validated insert machinery in + /// one PostgreSQL transaction. Any failure rolls back every imported side effect. + async fn commit_table_data_import( + &self, + request: tonic::Request, + ) -> std::result::Result< + tonic::Response, + tonic::Status, + >; + /// Discards a staged import. Already committed imports cannot be aborted. + async fn abort_table_data_import( + &self, + request: tonic::Request, + ) -> std::result::Result< + tonic::Response, + tonic::Status, + >; /// Update existing row data with strict type binding and script validation. /// /// Behavior: @@ -1353,6 +1557,190 @@ pub mod tables_data_server { }; Box::pin(fut) } + "/komp_ac.tables_data.TablesData/BeginTableDataImport" => { + #[allow(non_camel_case_types)] + struct BeginTableDataImportSvc(pub Arc); + impl< + T: TablesData, + > tonic::server::UnaryService + for BeginTableDataImportSvc { + type Response = super::BeginTableDataImportResponse; + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { + let inner = Arc::clone(&self.0); + let fut = async move { + ::begin_table_data_import(&inner, request) + .await + }; + Box::pin(fut) + } + } + let accept_compression_encodings = self.accept_compression_encodings; + let send_compression_encodings = self.send_compression_encodings; + let max_decoding_message_size = self.max_decoding_message_size; + let max_encoding_message_size = self.max_encoding_message_size; + let inner = self.inner.clone(); + let fut = async move { + let method = BeginTableDataImportSvc(inner); + let codec = tonic_prost::ProstCodec::default(); + let mut grpc = tonic::server::Grpc::new(codec) + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); + let res = grpc.unary(method, req).await; + Ok(res) + }; + Box::pin(fut) + } + "/komp_ac.tables_data.TablesData/StageTableDataImport" => { + #[allow(non_camel_case_types)] + struct StageTableDataImportSvc(pub Arc); + impl< + T: TablesData, + > tonic::server::UnaryService + for StageTableDataImportSvc { + type Response = super::StageTableDataImportResponse; + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { + let inner = Arc::clone(&self.0); + let fut = async move { + ::stage_table_data_import(&inner, request) + .await + }; + Box::pin(fut) + } + } + let accept_compression_encodings = self.accept_compression_encodings; + let send_compression_encodings = self.send_compression_encodings; + let max_decoding_message_size = self.max_decoding_message_size; + let max_encoding_message_size = self.max_encoding_message_size; + let inner = self.inner.clone(); + let fut = async move { + let method = StageTableDataImportSvc(inner); + let codec = tonic_prost::ProstCodec::default(); + let mut grpc = tonic::server::Grpc::new(codec) + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); + let res = grpc.unary(method, req).await; + Ok(res) + }; + Box::pin(fut) + } + "/komp_ac.tables_data.TablesData/CommitTableDataImport" => { + #[allow(non_camel_case_types)] + struct CommitTableDataImportSvc(pub Arc); + impl< + T: TablesData, + > tonic::server::UnaryService + for CommitTableDataImportSvc { + type Response = super::CommitTableDataImportResponse; + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { + let inner = Arc::clone(&self.0); + let fut = async move { + ::commit_table_data_import(&inner, request) + .await + }; + Box::pin(fut) + } + } + let accept_compression_encodings = self.accept_compression_encodings; + let send_compression_encodings = self.send_compression_encodings; + let max_decoding_message_size = self.max_decoding_message_size; + let max_encoding_message_size = self.max_encoding_message_size; + let inner = self.inner.clone(); + let fut = async move { + let method = CommitTableDataImportSvc(inner); + let codec = tonic_prost::ProstCodec::default(); + let mut grpc = tonic::server::Grpc::new(codec) + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); + let res = grpc.unary(method, req).await; + Ok(res) + }; + Box::pin(fut) + } + "/komp_ac.tables_data.TablesData/AbortTableDataImport" => { + #[allow(non_camel_case_types)] + struct AbortTableDataImportSvc(pub Arc); + impl< + T: TablesData, + > tonic::server::UnaryService + for AbortTableDataImportSvc { + type Response = super::AbortTableDataImportResponse; + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { + let inner = Arc::clone(&self.0); + let fut = async move { + ::abort_table_data_import(&inner, request) + .await + }; + Box::pin(fut) + } + } + let accept_compression_encodings = self.accept_compression_encodings; + let send_compression_encodings = self.send_compression_encodings; + let max_decoding_message_size = self.max_decoding_message_size; + let max_encoding_message_size = self.max_encoding_message_size; + let inner = self.inner.clone(); + let fut = async move { + let method = AbortTableDataImportSvc(inner); + let codec = tonic_prost::ProstCodec::default(); + let mut grpc = tonic::server::Grpc::new(codec) + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); + let res = grpc.unary(method, req).await; + Ok(res) + }; + Box::pin(fut) + } "/komp_ac.tables_data.TablesData/PutTableData" => { #[allow(non_camel_case_types)] struct PutTableDataSvc(pub Arc); diff --git a/web/locales/cs/main.ftl b/web/locales/cs/main.ftl index bfb6e40d..8d363df2 100644 --- a/web/locales/cs/main.ftl +++ b/web/locales/cs/main.ftl @@ -592,9 +592,6 @@ import-continue = Pokračovat import-carried = Import do tabulky { $table } v rozsahu { $scope }. import-error-title = CSV se nepodařilo importovat import-import-rows = Importovat -import-failure-message = Před zastavením importu se importovalo { $inserted } z { $source_rows } řádků. Selhal řádek CSV { $row }. - Backend: { $error } - Již importované řádky zůstávají v tabulce. Aby nevznikly duplicity, neimportujte znovu celý soubor: nejprve tyto řádky odstraňte nebo importujte pouze chybný a zbývající řádky. import-failure-no-rows-message = Nic se neimportovalo. Selhal řádek CSV { $row }. Backend: { $error } import-err-bad-date = Řádek CSV { $row }, sloupec { $column }: „{ $value }“ není platné datum ve formátu { $format }. @@ -611,7 +608,7 @@ import-success-message = Vloženo { $inserted -> # --- Import v průběhu ------------------------------------------------------- import-progress-heading = Importuje se do { $table } -import-progress-rows = Importováno { $inserted } z { $total } řádků +import-progress-rows = Připraveno { $inserted } z { $total } řádků import-progress-label = Průběh importu import-progress-elapsed = Uplynulo import-progress-rate = Rychlost diff --git a/web/locales/en/main.ftl b/web/locales/en/main.ftl index 06b5e8f5..3a6407cd 100644 --- a/web/locales/en/main.ftl +++ b/web/locales/en/main.ftl @@ -582,9 +582,6 @@ import-continue = Continue import-carried = Importing into { $table } in { $scope }. import-error-title = Could not import CSV import-import-rows = Import -import-failure-message = Imported { $inserted } of { $source_rows } rows before the import stopped. CSV row { $row } failed. - Backend: { $error } - Rows already imported remain in the table. To avoid duplicates, do not retry the whole file: remove those rows first, or import only the failed and remaining rows. import-failure-no-rows-message = Nothing was imported. CSV row { $row } failed. Backend: { $error } import-err-bad-date = CSV row { $row }, column { $column }: “{ $value }” is not a valid { $format } date. @@ -599,7 +596,7 @@ import-success-message = Inserted { $inserted -> # --- The import while it runs ---------------------------------------------- import-progress-heading = Importing into { $table } -import-progress-rows = { $inserted } of { $total } rows imported +import-progress-rows = { $inserted } of { $total } rows staged import-progress-label = Import progress import-progress-elapsed = Elapsed import-progress-rate = Speed diff --git a/web/locales/sk/main.ftl b/web/locales/sk/main.ftl index 09d69b82..bc81bafb 100644 --- a/web/locales/sk/main.ftl +++ b/web/locales/sk/main.ftl @@ -592,9 +592,6 @@ import-continue = Pokračovať import-carried = Import do tabuľky { $table } v rozsahu { $scope }. import-error-title = CSV sa nepodarilo importovať import-import-rows = Importovať -import-failure-message = Pred zastavením importu sa importovalo { $inserted } z { $source_rows } riadkov. Zlyhal riadok CSV { $row }. - Backend: { $error } - Už importované riadky zostávajú v tabuľke. Aby nevznikli duplicity, neimportujte znova celý súbor: najprv tieto riadky odstráňte alebo importujte iba chybný a zostávajúce riadky. import-failure-no-rows-message = Nič sa neimportovalo. Zlyhal riadok CSV { $row }. Backend: { $error } import-err-bad-date = Riadok CSV { $row }, stĺpec { $column }: „{ $value }“ nie je platný dátum vo formáte { $format }. @@ -609,7 +606,7 @@ import-success-message = Vložený { $inserted -> # --- Import počas behu ------------------------------------------------------ import-progress-heading = Importuje sa do { $table } -import-progress-rows = Importovaných { $inserted } z { $total } riadkov +import-progress-rows = Pripravených { $inserted } z { $total } riadkov import-progress-label = Priebeh importu import-progress-elapsed = Uplynulo import-progress-rate = Rýchlosť diff --git a/web/src/pages/import_export/import/logic.rs b/web/src/pages/import_export/import/logic.rs index 064b65ec..944bb603 100644 --- a/web/src/pages/import_export/import/logic.rs +++ b/web/src/pages/import_export/import/logic.rs @@ -13,7 +13,10 @@ use crate::{ AppState, definitions::{ table_structure::GetTableStructureRequest, - tables_data::{PostTableDataBulkRequest, PostTableDataBulkRow}, + tables_data::{ + AbortTableDataImportRequest, BeginTableDataImportRequest, CommitTableDataImportRequest, + PostTableDataBulkRow, StageTableDataImportRequest, + }, }, services::{authenticated_request, reject_cross_site}, {i18n::Locale, tr}, @@ -264,7 +267,10 @@ pub(crate) async fn import_csv( // practice; it is not worth a panic on the import path. None => ( StatusCode::INTERNAL_SERVER_ERROR, - Html(ui::render_error(locale, &tr!(locale, "import-progress-gone"))), + Html(ui::render_error( + locale, + &tr!(locale, "import-progress-gone"), + )), ) .into_response(), } @@ -350,37 +356,100 @@ struct Running { /// The insert, chunk by chunk, reporting after each one. async fn run_import(job: Running) { let jobs = job.state.imports.clone(); - let mut inserted = 0usize; + let begin = BeginTableDataImportRequest { + profile_name: job.profile_name.clone(), + }; + let Ok(begin) = authenticated_request(&job.headers, begin) else { + return jobs.finish(&job.id, 0, Outcome::SessionLost); + }; + let mut data = job.state.tables_data.clone(); + let import_id = match data.begin_table_data_import(begin).await { + Ok(response) => response.into_inner().import_id, + Err(error) => { + let (_, outcome) = import_failure(&error, 0, 0); + return jobs.finish(&job.id, 0, outcome); + } + }; + let mut staged = 0usize; - for (chunk_index, chunk) in job.rows.chunks(CHUNK_ROWS).enumerate() { - let request = PostTableDataBulkRequest { - profile_name: job.profile_name.clone(), + for chunk in job.rows.chunks(CHUNK_ROWS) { + let request = StageTableDataImportRequest { + import_id: import_id.clone(), table_name: job.table_name.clone(), rows: chunk.to_vec(), }; let Ok(request) = authenticated_request(&job.headers, request) else { - return jobs.finish(&job.id, inserted, Outcome::SessionLost); + abort_import(&job, &import_id).await; + return jobs.finish(&job.id, 0, Outcome::SessionLost); }; - let mut data = job.state.tables_data.clone(); - match data.post_table_data_bulk(request).await { + match data.stage_table_data_import(request).await { Ok(response) => { - inserted += response - .into_inner() - .responses - .iter() - .filter(|row| row.inserted_id > 0) - .count(); - jobs.advance(&job.id, inserted); + staged = usize::try_from(response.into_inner().total_staged_rows) + .unwrap_or(job.rows.len()); + jobs.advance(&job.id, staged); } Err(error) => { - let (inserted, outcome) = - import_failure(&error, inserted, chunk_index * CHUNK_ROWS); - return jobs.finish(&job.id, inserted, outcome); + abort_import(&job, &import_id).await; + let (_, outcome) = import_failure(&error, 0, 0); + return jobs.finish(&job.id, 0, outcome); } } } - jobs.finish(&job.id, inserted, Outcome::Succeeded); + let mut commit_attempt = 0; + let committed = loop { + let Ok(commit) = authenticated_request( + &job.headers, + CommitTableDataImportRequest { + import_id: import_id.clone(), + }, + ) else { + abort_import(&job, &import_id).await; + return jobs.finish(&job.id, 0, Outcome::SessionLost); + }; + match data.commit_table_data_import(commit).await { + Err(error) + if commit_attempt == 0 + && matches!( + error.code(), + tonic::Code::Cancelled + | tonic::Code::Unknown + | tonic::Code::DeadlineExceeded + | tonic::Code::Internal + | tonic::Code::Unavailable + ) => + { + // The first commit may have reached PostgreSQL even if its response was lost. + // Completed sessions are durable, so repeating this call cannot duplicate rows. + commit_attempt += 1; + } + result => break result, + } + }; + match committed { + Ok(response) => { + let inserted = usize::try_from(response.into_inner().inserted_rows).unwrap_or(staged); + jobs.finish(&job.id, inserted, Outcome::Succeeded); + } + Err(error) => { + abort_import(&job, &import_id).await; + let (_, outcome) = import_failure(&error, 0, 0); + jobs.finish(&job.id, 0, outcome); + } + } +} + +async fn abort_import(job: &Running, import_id: &str) { + let Ok(request) = authenticated_request( + &job.headers, + AbortTableDataImportRequest { + import_id: import_id.to_string(), + }, + ) else { + return; + }; + let mut data = job.state.tables_data.clone(); + let _ = data.abort_table_data_import(request).await; } /// POST /admin/import/prepared.csv — the same file the import would read, to @@ -487,8 +556,7 @@ async fn prepared( ) -> Result<(Destination, Source, Prepared), Response> { let locale = Locale::from_headers(headers); let destination = destination(state, headers, form).await?; - let source = read_source(locale, &form.csv_data) - .map_err(|message| reject(headers, message))?; + let source = read_source(locale, &form.csv_data).map_err(|message| reject(headers, message))?; // Every posted destination has to still exist and still be writable. A key // that resolves to nothing is refused rather than skipped: skipping it @@ -511,13 +579,8 @@ async fn prepared( let assignments = read_mapping(locale, &chosen, &source).map_err(|message| reject(headers, message))?; let mut prepared = prepare(&assignments, &source, &destination.names()); - normalize_dates( - locale, - &mut prepared, - &destination.types, - form.date_format, - ) - .map_err(|message| reject(headers, message))?; + normalize_dates(locale, &mut prepared, &destination.types, form.date_format) + .map_err(|message| reject(headers, message))?; Ok((destination, source, prepared)) } @@ -604,24 +667,21 @@ fn grpc_error(headers: &HeaderMap, error: &tonic::Status) -> Response { .into_response(Locale::from_headers(headers), ui::render_error) } -/// How a failed chunk ends the import: the final inserted count, and what the -/// page will say about it. -/// -/// The rows before the failure are in the table and stay there, so the count -/// the job keeps is the count the message is built from. +/// How a failed atomic commit ends the import. Staging progress is discarded: +/// a failed import commits no profile rows. fn import_failure( error: &tonic::Status, - inserted_before_chunk: usize, + _inserted_before_chunk: usize, chunk_start: usize, ) -> (usize, Outcome) { let failure = crate::ui::FormError::from_status_with_message(error, format!("{error:?}")); if matches!(failure, crate::ui::FormError::Unauthenticated) { - return (inserted_before_chunk, Outcome::SessionLost); + return (0, Outcome::SessionLost); } let status = failure.status_code(); match bulk_failure(error) { - Some((failed_row_index, inserted_in_chunk)) => ( - inserted_before_chunk + inserted_in_chunk, + Some((failed_row_index, _inserted_in_chunk)) => ( + 0, Outcome::RowFailed { status, // One header row precedes the data, and CSV rows are one-based. @@ -629,10 +689,9 @@ fn import_failure( backend: format!("{error:?}"), }, ), - // The backend refused the batch without saying which row did it, so - // there is no partial progress to report inside this chunk. + // The backend refused the import without identifying one row. None => ( - inserted_before_chunk, + 0, Outcome::Refused { status, message: failure.message().to_string(), diff --git a/web/src/pages/import_export/import/progress.rs b/web/src/pages/import_export/import/progress.rs index b6f88656..e2bba2a7 100644 --- a/web/src/pages/import_export/import/progress.rs +++ b/web/src/pages/import_export/import/progress.rs @@ -33,10 +33,9 @@ const KEEP_FINISHED: Duration = Duration::from_secs(300); /// in their language, in whichever alert their page expects. #[derive(Clone, Debug)] pub(crate) enum Outcome { - /// Every chunk was accepted. + /// Every staged row was committed atomically. Succeeded, - /// The backend stopped on one row. `inserted` on the snapshot says how much - /// of the file went in before it, and stays true: those rows remain. + /// The backend stopped on one row. No rows from the import were committed. RowFailed { status: StatusCode, /// One-based row of the file the user handed over, header included. @@ -222,8 +221,7 @@ mod tests { assert!(snapshot.outcome.is_none()); } - /// A failure keeps the count it reached: those rows are in the table, and - /// the message about what to do next is built from that number. + /// Finishing replaces staging progress with the number actually committed. #[test] fn a_finished_job_keeps_its_count_and_its_outcome() { let jobs = ImportJobs::default(); @@ -232,7 +230,7 @@ mod tests { jobs.advance(&id, 300); jobs.finish( &id, - 340, + 0, Outcome::RowFailed { status: StatusCode::UNPROCESSABLE_ENTITY, csv_row: 342, @@ -241,7 +239,7 @@ mod tests { ); let snapshot = jobs.snapshot(&id, "session").unwrap(); - assert_eq!(snapshot.inserted, 340); + assert_eq!(snapshot.inserted, 0); assert!(matches!( snapshot.outcome, Some(Outcome::RowFailed { csv_row: 342, .. }) diff --git a/web/src/pages/import_export/import/ui.rs b/web/src/pages/import_export/import/ui.rs index eb03a372..559c9f5c 100644 --- a/web/src/pages/import_export/import/ui.rs +++ b/web/src/pages/import_export/import/ui.rs @@ -90,9 +90,10 @@ pub(crate) fn render_progress(locale: Locale, id: &str, snapshot: &Snapshot) -> }, Stat { label: tr!(locale, "import-progress-rate"), - value: snapshot.rows_per_second().map_or_else(unknown, |rows| { - tr!(locale, "import-progress-rate-value", "rows" => rows as i64) - }), + value: snapshot.rows_per_second().map_or_else( + unknown, + |rows| tr!(locale, "import-progress-rate-value", "rows" => rows as i64), + ), }, Stat { label: tr!(locale, "import-progress-remaining"), @@ -134,24 +135,17 @@ pub(crate) fn render_error(locale: Locale, message: &str) -> String { pub(crate) fn render_import_failure( locale: Locale, - inserted: usize, - prepared_rows: usize, + _inserted: usize, + _prepared_rows: usize, csv_row: usize, backend_message: &str, ) -> String { - let message_key = if inserted == 0 { - "import-failure-no-rows-message" - } else { - "import-failure-message" - }; render(&Alert::error( locale, &tr!(locale, "import-error-title"), &tr!( locale, - message_key, - "inserted" => inserted as i64, - "source_rows" => prepared_rows as i64, + "import-failure-no-rows-message", "row" => csv_row as i64, "error" => backend_message.to_string(), ), @@ -215,10 +209,7 @@ mod tests { destination: strings(&["id:42", "id:57"]), source_position: strings(&["1", "2"]), }, - date_formats: DateFormat::options( - Locale::English, - DateFormat::YearMonthDayFourDigit, - ), + date_formats: DateFormat::options(Locale::English, DateFormat::YearMonthDayFourDigit), step, } } @@ -303,9 +294,18 @@ mod tests { assert!(html.contains("Original table columns"), "{html}"); assert!(html.contains(r#"data-source-position="1""#), "{html}"); assert!(html.contains("value-a"), "{html}"); - assert!(html.contains(r#"name="destination" value="id:42""#), "{html}"); - assert!(html.contains(r#""), "{html}"); + assert!( + html.contains(r#"name="destination" value="id:42""#), + "{html}" + ); + assert!( + html.contains(r#""), + "{html}" + ); } /// The two sides are two lists, not one zipped table: a source chip carries @@ -339,7 +339,10 @@ mod tests { fn a_destination_is_carried_as_its_identity() { let html = render_step(&page(mapping())); - assert!(html.contains(r#"name="destination" value="id:42""#), "{html}"); + assert!( + html.contains(r#"name="destination" value="id:42""#), + "{html}" + ); assert!(!html.contains(r#"name="destination" value="a""#), "{html}"); } @@ -368,11 +371,20 @@ mod tests { html.contains(r#"formaction="/admin/import/prepared.csv""#), "{html}" ); - assert!(html.contains(""a","b","c""), "{html}"); + assert!( + html.contains(""a","b","c""), + "{html}" + ); // The mapping travels with it, so the download and the import prepare // the identical file. - assert!(html.contains(r#"name="destination" value="id:42""#), "{html}"); - assert!(html.contains(r#"name="source_position" value="1""#), "{html}"); + assert!( + html.contains(r#"name="destination" value="id:42""#), + "{html}" + ); + assert!( + html.contains(r#"name="source_position" value="1""#), + "{html}" + ); } /// While it runs, the card says how far along it is and asks for itself @@ -391,10 +403,13 @@ mod tests { }, ); - assert!(html.contains(r#"hx-get="/admin/import/progress/7""#), "{html}"); + assert!( + html.contains(r#"hx-get="/admin/import/progress/7""#), + "{html}" + ); assert!(html.contains(r#"hx-trigger="every 1s""#), "{html}"); assert!(html.contains(r#"value="25""#), "{html}"); - assert!(html.contains("500 of 2000 rows imported"), "{html}"); + assert!(html.contains("500 of 2000 rows staged"), "{html}"); assert!(html.contains("40 s"), "{html}"); // 500 rows in 40 seconds, so the 1500 left are about two minutes away. assert!(html.contains("13 rows/s"), "{html}"); @@ -423,7 +438,7 @@ mod tests { } #[test] - fn an_import_failure_explains_partial_progress_and_the_backend_error() { + fn an_import_failure_explains_atomic_rollback_and_the_backend_error() { let html = render_import_failure( Locale::English, 599, @@ -432,10 +447,10 @@ mod tests { "Internal server error (reference: example-id)", ); - assert!(html.contains("Imported 599 of 1200 rows"), "{html}"); + assert!(html.contains("Nothing was imported"), "{html}"); assert!(html.contains("CSV row 601 failed"), "{html}"); assert!(html.contains("reference: example-id"), "{html}"); - assert!(html.contains("Rows already imported remain"), "{html}"); + assert!(!html.contains("already imported remain"), "{html}"); } #[test]