import page progress live
This commit is contained in:
Priec
@@ -108,6 +108,10 @@ pub(crate) struct AppState {
|
||||
validations: TableValidationServiceClient<Channel>,
|
||||
tables_data: TablesDataClient<Channel>,
|
||||
ecb: EcbServiceClient<Channel>,
|
||||
/// The imports this process is running, which the import page polls. Not a
|
||||
/// client: an import outlives the request that started it, so where it has
|
||||
/// got to has to live somewhere both the task and the next request can see.
|
||||
imports: pages::import_export::import::progress::ImportJobs,
|
||||
}
|
||||
|
||||
/// Starts the web UI as a detached task on the current Tokio runtime.
|
||||
@@ -141,6 +145,7 @@ pub async fn serve() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
|
||||
tables_data: TablesDataClient::new(channel.clone()),
|
||||
ecb: EcbServiceClient::new(channel.clone()),
|
||||
structures: TableStructureServiceClient::new(channel),
|
||||
imports: Default::default(),
|
||||
};
|
||||
|
||||
let listener = tokio::net::TcpListener::bind(listen_address).await?;
|
||||
@@ -206,6 +211,7 @@ mod tests {
|
||||
tables_data: TablesDataClient::new(channel.clone()),
|
||||
ecb: EcbServiceClient::new(channel.clone()),
|
||||
structures: TableStructureServiceClient::new(channel),
|
||||
imports: Default::default(),
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -1,8 +1,8 @@
|
||||
use std::collections::HashMap;
|
||||
|
||||
use axum::{
|
||||
extract::State,
|
||||
http::{HeaderMap, HeaderValue, header},
|
||||
extract::{Path, State},
|
||||
http::{HeaderMap, HeaderName, HeaderValue, StatusCode, header},
|
||||
response::{Html, IntoResponse, Redirect, Response},
|
||||
};
|
||||
// The review posts a destination and selected CSV position once per table row;
|
||||
@@ -29,6 +29,7 @@ use super::{
|
||||
prepare::{
|
||||
Prepared, Source, canonical_csv, normalize_dates, prepare, read_mapping, read_source,
|
||||
},
|
||||
progress::Outcome,
|
||||
state::{ImportForm, MappingRow, MappingStep, PreviewStep, SourceOption, Step},
|
||||
ui,
|
||||
};
|
||||
@@ -36,6 +37,10 @@ use super::{
|
||||
/// How many prepared rows the preview shows.
|
||||
const PREVIEW_ROWS: usize = 20;
|
||||
|
||||
/// How many rows one bulk insert carries. Also the granularity of the progress
|
||||
/// the page shows, since a chunk is what the running total counts.
|
||||
const CHUNK_ROWS: usize = 1_000;
|
||||
|
||||
/// The table an import writes into, as the server currently declares it.
|
||||
///
|
||||
/// Loaded again at every step rather than carried in the form. The destination
|
||||
@@ -230,46 +235,152 @@ pub(crate) async fn import_csv(
|
||||
Err(message) => return reject(&headers, message),
|
||||
};
|
||||
|
||||
let mut inserted = 0usize;
|
||||
for (chunk_index, chunk) in converted.chunks(1_000).enumerate() {
|
||||
let request = PostTableDataBulkRequest {
|
||||
profile_name: profile_name.clone(),
|
||||
table_name: destination.table_name.clone(),
|
||||
rows: chunk.to_vec(),
|
||||
};
|
||||
let mut data = state.tables_data.clone();
|
||||
let response = match data
|
||||
.post_table_data_bulk(match authenticated_request(&headers, request) {
|
||||
Ok(request) => request,
|
||||
Err(_) => return Redirect::to("/login").into_response(),
|
||||
})
|
||||
.await
|
||||
{
|
||||
Ok(response) => response.into_inner(),
|
||||
Err(error) => {
|
||||
return import_error(
|
||||
&headers,
|
||||
&error,
|
||||
inserted,
|
||||
chunk_index * 1_000,
|
||||
prepared.row_count(),
|
||||
);
|
||||
}
|
||||
};
|
||||
inserted += response
|
||||
.responses
|
||||
.iter()
|
||||
.filter(|row| row.inserted_id > 0)
|
||||
.count();
|
||||
// Everything above is the request's work — refusing it is still an answer
|
||||
// to this POST. The insert itself is not: it is as long as the file, so it
|
||||
// becomes a job, and the answer is the card that watches it.
|
||||
let Some(session) = crate::cookie_value(&headers, crate::ui::SESSION_COOKIE) else {
|
||||
return Redirect::to("/login").into_response();
|
||||
};
|
||||
let session = session.to_string();
|
||||
if state.imports.is_running(&session) {
|
||||
return reject(&headers, tr!(locale, "import-err-already-running"));
|
||||
}
|
||||
|
||||
Html(ui::render_success(
|
||||
locale,
|
||||
inserted,
|
||||
prepared.row_count(),
|
||||
&destination.table_name,
|
||||
))
|
||||
.into_response()
|
||||
let id = state
|
||||
.imports
|
||||
.start(&session, &destination.table_name, prepared.row_count());
|
||||
tokio::spawn(run_import(Running {
|
||||
state: state.clone(),
|
||||
headers,
|
||||
id: id.clone(),
|
||||
profile_name,
|
||||
table_name: destination.table_name.clone(),
|
||||
rows: converted,
|
||||
}));
|
||||
|
||||
match state.imports.snapshot(&id, &session) {
|
||||
Some(snapshot) => Html(ui::render_progress(locale, &id, &snapshot)).into_response(),
|
||||
// The job was registered a line ago, so this is unreachable in
|
||||
// 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"))),
|
||||
)
|
||||
.into_response(),
|
||||
}
|
||||
}
|
||||
|
||||
/// GET /admin/import/progress/{id} — how far the import has got, or how it
|
||||
/// ended.
|
||||
///
|
||||
/// The card polls this every second and swaps itself for the answer, so this
|
||||
/// endpoint renders both: another card while the import runs, and the same
|
||||
/// success or failure alert the POST used to answer with once it is over.
|
||||
pub(crate) async fn import_progress(
|
||||
State(state): State<AppState>,
|
||||
headers: HeaderMap,
|
||||
Path(id): Path<String>,
|
||||
) -> Response {
|
||||
let locale = Locale::from_headers(&headers);
|
||||
let Some(session) = crate::cookie_value(&headers, crate::ui::SESSION_COOKIE) else {
|
||||
return Redirect::to("/login").into_response();
|
||||
};
|
||||
// Someone else's job id, or one dropped after it finished. Either way there
|
||||
// is nothing to report, and the alert says so rather than the card polling
|
||||
// a job that will never answer.
|
||||
let Some(snapshot) = state.imports.snapshot(&id, session) else {
|
||||
return reject(&headers, tr!(locale, "import-progress-gone"));
|
||||
};
|
||||
|
||||
match &snapshot.outcome {
|
||||
None => Html(ui::render_progress(locale, &id, &snapshot)).into_response(),
|
||||
Some(Outcome::Succeeded) => Html(ui::render_success(
|
||||
locale,
|
||||
snapshot.inserted,
|
||||
snapshot.total_rows,
|
||||
&snapshot.table_name,
|
||||
))
|
||||
.into_response(),
|
||||
Some(Outcome::RowFailed {
|
||||
status,
|
||||
csv_row,
|
||||
backend,
|
||||
}) => (
|
||||
*status,
|
||||
Html(ui::render_import_failure(
|
||||
locale,
|
||||
snapshot.inserted,
|
||||
snapshot.total_rows,
|
||||
*csv_row,
|
||||
backend,
|
||||
)),
|
||||
)
|
||||
.into_response(),
|
||||
Some(Outcome::Refused { status, message }) => {
|
||||
(*status, Html(ui::render_error(locale, message))).into_response()
|
||||
}
|
||||
// A poll is an XHR, so a 303 here would be followed by the XHR and the
|
||||
// login page swapped into the card. Ask htmx to navigate the browser,
|
||||
// which is what redirecting to /login means everywhere else.
|
||||
Some(Outcome::SessionLost) => (
|
||||
[(
|
||||
HeaderName::from_static("hx-redirect"),
|
||||
HeaderValue::from_static("/login"),
|
||||
)],
|
||||
Html(String::new()),
|
||||
)
|
||||
.into_response(),
|
||||
}
|
||||
}
|
||||
|
||||
/// One spawned import: what the walk through the prepared rows needs, and
|
||||
/// nothing that belonged to the request that started it.
|
||||
struct Running {
|
||||
state: AppState,
|
||||
/// The session cookie the chunks are sent with. The import outlives the
|
||||
/// request, but not the session: a signed-out user's remaining chunks are
|
||||
/// refused by the backend, exactly as they would have been inline.
|
||||
headers: HeaderMap,
|
||||
id: String,
|
||||
profile_name: String,
|
||||
table_name: String,
|
||||
rows: Vec<PostTableDataBulkRow>,
|
||||
}
|
||||
|
||||
/// 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;
|
||||
|
||||
for (chunk_index, chunk) in job.rows.chunks(CHUNK_ROWS).enumerate() {
|
||||
let request = PostTableDataBulkRequest {
|
||||
profile_name: job.profile_name.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);
|
||||
};
|
||||
let mut data = job.state.tables_data.clone();
|
||||
match data.post_table_data_bulk(request).await {
|
||||
Ok(response) => {
|
||||
inserted += response
|
||||
.into_inner()
|
||||
.responses
|
||||
.iter()
|
||||
.filter(|row| row.inserted_id > 0)
|
||||
.count();
|
||||
jobs.advance(&job.id, inserted);
|
||||
}
|
||||
Err(error) => {
|
||||
let (inserted, outcome) =
|
||||
import_failure(&error, inserted, chunk_index * CHUNK_ROWS);
|
||||
return jobs.finish(&job.id, inserted, outcome);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
jobs.finish(&job.id, inserted, Outcome::Succeeded);
|
||||
}
|
||||
|
||||
/// POST /admin/import/prepared.csv — the same file the import would read, to
|
||||
@@ -493,32 +604,41 @@ fn grpc_error(headers: &HeaderMap, error: &tonic::Status) -> Response {
|
||||
.into_response(Locale::from_headers(headers), ui::render_error)
|
||||
}
|
||||
|
||||
fn import_error(
|
||||
headers: &HeaderMap,
|
||||
/// 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.
|
||||
fn import_failure(
|
||||
error: &tonic::Status,
|
||||
inserted_before_chunk: usize,
|
||||
chunk_start: usize,
|
||||
prepared_rows: usize,
|
||||
) -> Response {
|
||||
let Some((failed_row_index, inserted_in_chunk)) = bulk_failure(error) else {
|
||||
return grpc_error(headers, error);
|
||||
};
|
||||
let locale = Locale::from_headers(headers);
|
||||
let failure = crate::ui::FormError::from_status(error);
|
||||
) -> (usize, Outcome) {
|
||||
let failure = crate::ui::FormError::from_status_with_message(error, format!("{error:?}"));
|
||||
if matches!(failure, crate::ui::FormError::Unauthenticated) {
|
||||
return failure.into_response(locale, ui::render_error);
|
||||
return (inserted_before_chunk, Outcome::SessionLost);
|
||||
}
|
||||
let status = failure.status_code();
|
||||
let backend_error = format!("{error:?}");
|
||||
let message = ui::render_import_failure(
|
||||
locale,
|
||||
inserted_before_chunk + inserted_in_chunk,
|
||||
prepared_rows,
|
||||
// One header row precedes the data, and CSV rows are one-based.
|
||||
chunk_start + failed_row_index + 2,
|
||||
&backend_error,
|
||||
);
|
||||
(status, Html(message)).into_response()
|
||||
match bulk_failure(error) {
|
||||
Some((failed_row_index, inserted_in_chunk)) => (
|
||||
inserted_before_chunk + inserted_in_chunk,
|
||||
Outcome::RowFailed {
|
||||
status,
|
||||
// One header row precedes the data, and CSV rows are one-based.
|
||||
csv_row: chunk_start + failed_row_index + 2,
|
||||
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.
|
||||
None => (
|
||||
inserted_before_chunk,
|
||||
Outcome::Refused {
|
||||
status,
|
||||
message: failure.message().to_string(),
|
||||
},
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
fn bulk_failure(error: &tonic::Status) -> Option<(usize, usize)> {
|
||||
|
||||
@@ -2,6 +2,7 @@ mod destination;
|
||||
mod loader;
|
||||
mod logic;
|
||||
mod prepare;
|
||||
pub(crate) mod progress;
|
||||
mod state;
|
||||
mod ui;
|
||||
|
||||
@@ -21,6 +22,9 @@ pub(crate) fn router() -> Router<AppState> {
|
||||
.route("/admin/import/source", post(logic::source_step))
|
||||
.route("/admin/import/prepare", post(logic::prepare_step))
|
||||
.route("/admin/import/preview", post(logic::preview_step))
|
||||
// A running import, which answers the POST below straight away and
|
||||
// then reports itself here until it is done.
|
||||
.route("/admin/import/progress/{id}", get(logic::import_progress))
|
||||
// The two things a prepared import can be: rows in the table, or a file
|
||||
// to look at first. Both read the same preparation.
|
||||
.route("/admin/import", post(logic::import_csv))
|
||||
|
||||
298
web/src/pages/import_export/import/progress.rs
Normal file
298
web/src/pages/import_export/import/progress.rs
Normal file
@@ -0,0 +1,298 @@
|
||||
//! Where a running import has got to, while it is still running.
|
||||
//!
|
||||
//! An import of a large file is one long chunked walk through the prepared
|
||||
//! rows, and until now the browser saw none of it: the POST simply did not
|
||||
//! answer until the last chunk was in. The walk now runs as its own task and
|
||||
//! records what it has done here, so the page can ask — every second, through
|
||||
//! the ordinary poll-until-finished shape `admin/ecb` already uses — and show
|
||||
//! it.
|
||||
//!
|
||||
//! The registry is this process's memory and nothing more. A job is readable
|
||||
//! only by the session that started it, and finished jobs are dropped shortly
|
||||
//! after they are read: the durable record of an import is the rows it wrote.
|
||||
|
||||
use std::{
|
||||
collections::HashMap,
|
||||
sync::{
|
||||
Arc, Mutex,
|
||||
atomic::{AtomicU64, Ordering},
|
||||
},
|
||||
time::{Duration, Instant},
|
||||
};
|
||||
|
||||
use axum::http::StatusCode;
|
||||
|
||||
/// How long a finished import stays readable. Long enough for the poll that is
|
||||
/// already in flight to collect it, and for a reload to still find it.
|
||||
const KEEP_FINISHED: Duration = Duration::from_secs(300);
|
||||
|
||||
/// How an import ended.
|
||||
///
|
||||
/// Kept as what happened rather than as rendered markup: the import runs in a
|
||||
/// task with no request behind it, and the words belong to whoever asks —
|
||||
/// in their language, in whichever alert their page expects.
|
||||
#[derive(Clone, Debug)]
|
||||
pub(crate) enum Outcome {
|
||||
/// Every chunk was accepted.
|
||||
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.
|
||||
RowFailed {
|
||||
status: StatusCode,
|
||||
/// One-based row of the file the user handed over, header included.
|
||||
csv_row: usize,
|
||||
backend: String,
|
||||
},
|
||||
/// The backend refused the batch, or could not be reached at all.
|
||||
Refused { status: StatusCode, message: String },
|
||||
/// The session went away while the import ran, so the remaining chunks
|
||||
/// could not be sent.
|
||||
SessionLost,
|
||||
}
|
||||
|
||||
/// One import, from the first chunk to whatever ended it.
|
||||
struct Job {
|
||||
/// The session that started it, so a job id alone does not read someone
|
||||
/// else's import.
|
||||
session: String,
|
||||
table_name: String,
|
||||
total_rows: usize,
|
||||
inserted: usize,
|
||||
started: Instant,
|
||||
finished: Option<(Instant, Outcome)>,
|
||||
}
|
||||
|
||||
/// What one poll shows.
|
||||
pub(crate) struct Snapshot {
|
||||
pub table_name: String,
|
||||
pub inserted: usize,
|
||||
pub total_rows: usize,
|
||||
pub elapsed: Duration,
|
||||
/// `None` while the import is still walking through the file.
|
||||
pub outcome: Option<Outcome>,
|
||||
}
|
||||
|
||||
impl Snapshot {
|
||||
/// How far along, as whole percent. A file with no rows is finished.
|
||||
pub(crate) fn percent(&self) -> u64 {
|
||||
if self.total_rows == 0 {
|
||||
return 100;
|
||||
}
|
||||
(self.inserted as u64 * 100 / self.total_rows as u64).min(100)
|
||||
}
|
||||
|
||||
/// Rows per second so far, or `None` before there is anything to divide:
|
||||
/// a rate measured over no elapsed time or no rows is not a rate.
|
||||
pub(crate) fn rows_per_second(&self) -> Option<u64> {
|
||||
let seconds = self.elapsed.as_secs_f64();
|
||||
if self.inserted == 0 || seconds < 0.5 {
|
||||
return None;
|
||||
}
|
||||
Some((self.inserted as f64 / seconds).round() as u64)
|
||||
}
|
||||
|
||||
/// How long the rest should take at the rate measured so far.
|
||||
pub(crate) fn remaining(&self) -> Option<Duration> {
|
||||
let rate = self.rows_per_second()?;
|
||||
if rate == 0 {
|
||||
return None;
|
||||
}
|
||||
let left = self.total_rows.saturating_sub(self.inserted) as u64;
|
||||
Some(Duration::from_secs(left.div_ceil(rate)))
|
||||
}
|
||||
}
|
||||
|
||||
/// Every import this process is running or has just finished.
|
||||
#[derive(Clone, Default)]
|
||||
pub(crate) struct ImportJobs {
|
||||
jobs: Arc<Mutex<HashMap<String, Job>>>,
|
||||
}
|
||||
|
||||
impl ImportJobs {
|
||||
/// Registers an import about to start, and hands back the id the page polls.
|
||||
pub(crate) fn start(&self, session: &str, table_name: &str, total_rows: usize) -> String {
|
||||
let id = next_id();
|
||||
let mut jobs = self.lock();
|
||||
prune(&mut jobs);
|
||||
jobs.insert(
|
||||
id.clone(),
|
||||
Job {
|
||||
session: session.to_string(),
|
||||
table_name: table_name.to_string(),
|
||||
total_rows,
|
||||
inserted: 0,
|
||||
started: Instant::now(),
|
||||
finished: None,
|
||||
},
|
||||
);
|
||||
id
|
||||
}
|
||||
|
||||
/// Records the running total after a chunk was accepted.
|
||||
pub(crate) fn advance(&self, id: &str, inserted: usize) {
|
||||
if let Some(job) = self.lock().get_mut(id) {
|
||||
job.inserted = inserted;
|
||||
}
|
||||
}
|
||||
|
||||
/// Records how the import ended. `inserted` is the final count, which for a
|
||||
/// failure is what went in before the row that stopped it.
|
||||
pub(crate) fn finish(&self, id: &str, inserted: usize, outcome: Outcome) {
|
||||
if let Some(job) = self.lock().get_mut(id) {
|
||||
job.inserted = inserted;
|
||||
job.finished = Some((Instant::now(), outcome));
|
||||
}
|
||||
}
|
||||
|
||||
/// The job as `session` may see it: its own, or nothing.
|
||||
pub(crate) fn snapshot(&self, id: &str, session: &str) -> Option<Snapshot> {
|
||||
let mut jobs = self.lock();
|
||||
prune(&mut jobs);
|
||||
let job = jobs.get(id).filter(|job| job.session == session)?;
|
||||
Some(Snapshot {
|
||||
table_name: job.table_name.clone(),
|
||||
inserted: job.inserted,
|
||||
total_rows: job.total_rows,
|
||||
elapsed: job
|
||||
.finished
|
||||
.as_ref()
|
||||
.map_or_else(|| job.started.elapsed(), |(at, _)| *at - job.started),
|
||||
outcome: job.finished.as_ref().map(|(_, outcome)| outcome.clone()),
|
||||
})
|
||||
}
|
||||
|
||||
/// Whether this session already has an import walking through a file.
|
||||
///
|
||||
/// One at a time per session: the Import button used to be unusable until
|
||||
/// the whole import had answered, and now that it answers immediately a
|
||||
/// second click would send the same file again.
|
||||
pub(crate) fn is_running(&self, session: &str) -> bool {
|
||||
let mut jobs = self.lock();
|
||||
prune(&mut jobs);
|
||||
jobs.values()
|
||||
.any(|job| job.session == session && job.finished.is_none())
|
||||
}
|
||||
|
||||
/// A poisoned registry is one panicked handler, not a reason to refuse
|
||||
/// every later import: the map itself is plain data.
|
||||
fn lock(&self) -> std::sync::MutexGuard<'_, HashMap<String, Job>> {
|
||||
self.jobs.lock().unwrap_or_else(|error| error.into_inner())
|
||||
}
|
||||
}
|
||||
|
||||
fn prune(jobs: &mut HashMap<String, Job>) {
|
||||
jobs.retain(|_, job| match &job.finished {
|
||||
Some((at, _)) => at.elapsed() < KEEP_FINISHED,
|
||||
None => true,
|
||||
});
|
||||
}
|
||||
|
||||
/// A per-process counter, which is all the uniqueness an id needs: what keeps
|
||||
/// one session from reading another's import is the session check, not an
|
||||
/// unguessable id.
|
||||
fn next_id() -> String {
|
||||
static NEXT: AtomicU64 = AtomicU64::new(1);
|
||||
NEXT.fetch_add(1, Ordering::Relaxed).to_string()
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn a_job_is_readable_only_by_the_session_that_started_it() {
|
||||
let jobs = ImportJobs::default();
|
||||
let id = jobs.start("session-a", "customers", 10);
|
||||
|
||||
assert!(jobs.snapshot(&id, "session-a").is_some());
|
||||
assert!(jobs.snapshot(&id, "session-b").is_none());
|
||||
assert!(jobs.snapshot("does-not-exist", "session-a").is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn progress_is_what_the_last_chunk_reported() {
|
||||
let jobs = ImportJobs::default();
|
||||
let id = jobs.start("session", "customers", 400);
|
||||
|
||||
jobs.advance(&id, 300);
|
||||
let snapshot = jobs.snapshot(&id, "session").unwrap();
|
||||
|
||||
assert_eq!(snapshot.inserted, 300);
|
||||
assert_eq!(snapshot.percent(), 75);
|
||||
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.
|
||||
#[test]
|
||||
fn a_finished_job_keeps_its_count_and_its_outcome() {
|
||||
let jobs = ImportJobs::default();
|
||||
let id = jobs.start("session", "customers", 400);
|
||||
|
||||
jobs.advance(&id, 300);
|
||||
jobs.finish(
|
||||
&id,
|
||||
340,
|
||||
Outcome::RowFailed {
|
||||
status: StatusCode::UNPROCESSABLE_ENTITY,
|
||||
csv_row: 342,
|
||||
backend: "invalid value".to_string(),
|
||||
},
|
||||
);
|
||||
let snapshot = jobs.snapshot(&id, "session").unwrap();
|
||||
|
||||
assert_eq!(snapshot.inserted, 340);
|
||||
assert!(matches!(
|
||||
snapshot.outcome,
|
||||
Some(Outcome::RowFailed { csv_row: 342, .. })
|
||||
));
|
||||
assert!(!jobs.is_running("session"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_session_has_one_running_import_at_a_time() {
|
||||
let jobs = ImportJobs::default();
|
||||
let id = jobs.start("session", "customers", 10);
|
||||
|
||||
assert!(jobs.is_running("session"));
|
||||
assert!(!jobs.is_running("other"));
|
||||
|
||||
jobs.finish(&id, 10, Outcome::Succeeded);
|
||||
assert!(!jobs.is_running("session"));
|
||||
}
|
||||
|
||||
/// Nothing to insert is not a stalled bar at 0%.
|
||||
#[test]
|
||||
fn an_empty_file_is_a_finished_bar() {
|
||||
let snapshot = Snapshot {
|
||||
table_name: "customers".to_string(),
|
||||
inserted: 0,
|
||||
total_rows: 0,
|
||||
elapsed: Duration::from_secs(1),
|
||||
outcome: Some(Outcome::Succeeded),
|
||||
};
|
||||
|
||||
assert_eq!(snapshot.percent(), 100);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_rate_needs_both_rows_and_time_before_it_means_anything() {
|
||||
let measured = Snapshot {
|
||||
table_name: "customers".to_string(),
|
||||
inserted: 500,
|
||||
total_rows: 2_000,
|
||||
elapsed: Duration::from_secs(10),
|
||||
outcome: None,
|
||||
};
|
||||
assert_eq!(measured.rows_per_second(), Some(50));
|
||||
assert_eq!(measured.remaining(), Some(Duration::from_secs(30)));
|
||||
|
||||
let too_early = Snapshot {
|
||||
inserted: 0,
|
||||
elapsed: Duration::from_millis(100),
|
||||
..measured
|
||||
};
|
||||
assert_eq!(too_early.rows_per_second(), None);
|
||||
assert_eq!(too_early.remaining(), None);
|
||||
}
|
||||
}
|
||||
@@ -1,8 +1,11 @@
|
||||
use std::time::Duration;
|
||||
|
||||
use askama::Template;
|
||||
|
||||
use crate::ui::{Alert, Nav, render};
|
||||
use crate::{i18n::Locale, tr};
|
||||
|
||||
use super::progress::Snapshot;
|
||||
// `Step` is named by the step template's `{% match %}`, so it has to be in
|
||||
// scope here — the derive expands into this module.
|
||||
use super::state::{ImportPageState, Step};
|
||||
@@ -38,6 +41,88 @@ pub(crate) fn render_step(page: &ImportPageState) -> String {
|
||||
})
|
||||
}
|
||||
|
||||
/// The card a running import answers with, and answers again at every poll.
|
||||
///
|
||||
/// Every string it shows is built here rather than in the template: the card is
|
||||
/// a fragment with no `Nav` behind it, and the numbers it puts in front of the
|
||||
/// user — a rate, a time left — are arithmetic, which is easier to read and to
|
||||
/// test in Rust than in markup.
|
||||
#[derive(Template)]
|
||||
#[template(path = "pages/import_export/import/progress.html")]
|
||||
struct ImportProgress {
|
||||
/// Where the card asks for its own next version.
|
||||
poll_url: String,
|
||||
heading: String,
|
||||
rows: String,
|
||||
percent: u64,
|
||||
stats: [Stat; 3],
|
||||
hint: String,
|
||||
label: String,
|
||||
}
|
||||
|
||||
struct Stat {
|
||||
label: String,
|
||||
value: String,
|
||||
}
|
||||
|
||||
pub(crate) fn render_progress(locale: Locale, id: &str, snapshot: &Snapshot) -> String {
|
||||
// A measurement that does not exist yet says so, rather than showing a
|
||||
// zero that reads as a stalled import.
|
||||
let unknown = || tr!(locale, "import-progress-unknown-yet");
|
||||
render(&ImportProgress {
|
||||
poll_url: format!("/admin/import/progress/{id}"),
|
||||
heading: tr!(
|
||||
locale,
|
||||
"import-progress-heading",
|
||||
"table" => snapshot.table_name.clone(),
|
||||
),
|
||||
rows: tr!(
|
||||
locale,
|
||||
"import-progress-rows",
|
||||
"inserted" => snapshot.inserted as i64,
|
||||
"total" => snapshot.total_rows as i64,
|
||||
),
|
||||
percent: snapshot.percent(),
|
||||
stats: [
|
||||
Stat {
|
||||
label: tr!(locale, "import-progress-elapsed"),
|
||||
value: duration(locale, snapshot.elapsed),
|
||||
},
|
||||
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)
|
||||
}),
|
||||
},
|
||||
Stat {
|
||||
label: tr!(locale, "import-progress-remaining"),
|
||||
value: snapshot
|
||||
.remaining()
|
||||
.map_or_else(unknown, |left| duration(locale, left)),
|
||||
},
|
||||
],
|
||||
hint: tr!(locale, "import-progress-hint"),
|
||||
label: tr!(locale, "import-progress-label"),
|
||||
})
|
||||
}
|
||||
|
||||
/// Whole seconds up to a minute, then minutes and seconds. Nothing an import
|
||||
/// reports is worth more precision than that, and a bare "312 s" is worse to
|
||||
/// read than "5 m 12 s".
|
||||
fn duration(locale: Locale, value: Duration) -> String {
|
||||
let seconds = value.as_secs();
|
||||
if seconds < 60 {
|
||||
tr!(locale, "import-progress-seconds", "seconds" => seconds as i64)
|
||||
} else {
|
||||
tr!(
|
||||
locale,
|
||||
"import-progress-minutes",
|
||||
"minutes" => (seconds / 60) as i64,
|
||||
"seconds" => (seconds % 60) as i64,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
/// The #submission-status swaps.
|
||||
pub(crate) fn render_error(locale: Locale, message: &str) -> String {
|
||||
render(&Alert::error(
|
||||
@@ -290,6 +375,53 @@ mod tests {
|
||||
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
|
||||
/// again: the trigger is what keeps the numbers moving.
|
||||
#[test]
|
||||
fn a_running_import_reports_its_numbers_and_polls_for_the_next_ones() {
|
||||
let html = render_progress(
|
||||
Locale::English,
|
||||
"7",
|
||||
&Snapshot {
|
||||
table_name: "customers".to_string(),
|
||||
inserted: 500,
|
||||
total_rows: 2_000,
|
||||
elapsed: Duration::from_secs(40),
|
||||
outcome: None,
|
||||
},
|
||||
);
|
||||
|
||||
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("40 s"), "{html}");
|
||||
// 500 rows in 40 seconds, so the 1500 left are about two minutes away.
|
||||
assert!(html.contains("13 rows/s"), "{html}");
|
||||
assert!(html.contains("1 m 56 s"), "{html}");
|
||||
}
|
||||
|
||||
/// A rate needs rows and time behind it. Before there are either, the card
|
||||
/// shows a dash rather than a zero that reads as a stalled import.
|
||||
#[test]
|
||||
fn a_just_started_import_claims_no_rate_it_has_not_measured() {
|
||||
let html = render_progress(
|
||||
Locale::English,
|
||||
"7",
|
||||
&Snapshot {
|
||||
table_name: "customers".to_string(),
|
||||
inserted: 0,
|
||||
total_rows: 2_000,
|
||||
elapsed: Duration::from_millis(120),
|
||||
outcome: None,
|
||||
},
|
||||
);
|
||||
|
||||
assert!(html.contains(r#"value="0""#), "{html}");
|
||||
assert!(!html.contains("rows/s"), "{html}");
|
||||
assert_eq!(html.matches('\u{2014}').count(), 2, "{html}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_import_failure_explains_partial_progress_and_the_backend_error() {
|
||||
let html = render_import_failure(
|
||||
|
||||
Reference in New Issue
Block a user