bulk import

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.

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<PostTableDataResponse>,
 }
+#[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<PostTableDataBulkRow>,
+}
+#[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<super::PostTableDataBulkRequest>,
@@ -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<super::BeginTableDataImportRequest>,
+        ) -> std::result::Result<
+            tonic::Response<super::BeginTableDataImportResponse>,
+            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<super::StageTableDataImportRequest>,
+        ) -> std::result::Result<
+            tonic::Response<super::StageTableDataImportResponse>,
+            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<super::CommitTableDataImportRequest>,
+        ) -> std::result::Result<
+            tonic::Response<super::CommitTableDataImportResponse>,
+            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<super::AbortTableDataImportRequest>,
+        ) -> std::result::Result<
+            tonic::Response<super::AbortTableDataImportResponse>,
+            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<super::PostTableDataResponse>,
             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<super::PostTableDataBulkRequest>,
@@ -1009,6 +1179,40 @@ pub mod tables_data_server {
             tonic::Response<super::PostTableDataBulkResponse>,
             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<super::BeginTableDataImportRequest>,
+        ) -> std::result::Result<
+            tonic::Response<super::BeginTableDataImportResponse>,
+            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<super::StageTableDataImportRequest>,
+        ) -> std::result::Result<
+            tonic::Response<super::StageTableDataImportResponse>,
+            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<super::CommitTableDataImportRequest>,
+        ) -> std::result::Result<
+            tonic::Response<super::CommitTableDataImportResponse>,
+            tonic::Status,
+        >;
+        /// Discards a staged import. Already committed imports cannot be aborted.
+        async fn abort_table_data_import(
+            &self,
+            request: tonic::Request<super::AbortTableDataImportRequest>,
+        ) -> std::result::Result<
+            tonic::Response<super::AbortTableDataImportResponse>,
+            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<T: TablesData>(pub Arc<T>);
+                    impl<
+                        T: TablesData,
+                    > tonic::server::UnaryService<super::BeginTableDataImportRequest>
+                    for BeginTableDataImportSvc<T> {
+                        type Response = super::BeginTableDataImportResponse;
+                        type Future = BoxFuture<
+                            tonic::Response<Self::Response>,
+                            tonic::Status,
+                        >;
+                        fn call(
+                            &mut self,
+                            request: tonic::Request<super::BeginTableDataImportRequest>,
+                        ) -> Self::Future {
+                            let inner = Arc::clone(&self.0);
+                            let fut = async move {
+                                <T as TablesData>::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<T: TablesData>(pub Arc<T>);
+                    impl<
+                        T: TablesData,
+                    > tonic::server::UnaryService<super::StageTableDataImportRequest>
+                    for StageTableDataImportSvc<T> {
+                        type Response = super::StageTableDataImportResponse;
+                        type Future = BoxFuture<
+                            tonic::Response<Self::Response>,
+                            tonic::Status,
+                        >;
+                        fn call(
+                            &mut self,
+                            request: tonic::Request<super::StageTableDataImportRequest>,
+                        ) -> Self::Future {
+                            let inner = Arc::clone(&self.0);
+                            let fut = async move {
+                                <T as TablesData>::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<T: TablesData>(pub Arc<T>);
+                    impl<
+                        T: TablesData,
+                    > tonic::server::UnaryService<super::CommitTableDataImportRequest>
+                    for CommitTableDataImportSvc<T> {
+                        type Response = super::CommitTableDataImportResponse;
+                        type Future = BoxFuture<
+                            tonic::Response<Self::Response>,
+                            tonic::Status,
+                        >;
+                        fn call(
+                            &mut self,
+                            request: tonic::Request<super::CommitTableDataImportRequest>,
+                        ) -> Self::Future {
+                            let inner = Arc::clone(&self.0);
+                            let fut = async move {
+                                <T as TablesData>::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<T: TablesData>(pub Arc<T>);
+                    impl<
+                        T: TablesData,
+                    > tonic::server::UnaryService<super::AbortTableDataImportRequest>
+                    for AbortTableDataImportSvc<T> {
+                        type Response = super::AbortTableDataImportResponse;
+                        type Future = BoxFuture<
+                            tonic::Response<Self::Response>,
+                            tonic::Status,
+                        >;
+                        fn call(
+                            &mut self,
+                            request: tonic::Request<super::AbortTableDataImportRequest>,
+                        ) -> Self::Future {
+                            let inner = Arc::clone(&self.0);
+                            let fut = async move {
+                                <T as TablesData>::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<T: TablesData>(pub Arc<T>);

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

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

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ť

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(),

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

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#"<option value="1" data-example="value-a" selected>"#), "{html}");
-        assert!(html.contains(">Not mapped — leave empty</option>"), "{html}");
+        assert!(
+            html.contains(r#"name="destination" value="id:42""#),
+            "{html}"
+        );
+        assert!(
+            html.contains(r#"<option value="1" data-example="value-a" selected>"#),
+            "{html}"
+        );
+        assert!(
+            html.contains(">Not mapped — leave empty</option>"),
+            "{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("&#34;a&#34;,&#34;b&#34;,&#34;c&#34;"), "{html}");
+        assert!(
+            html.contains("&#34;a&#34;,&#34;b&#34;,&#34;c&#34;"),
+            "{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]