feat(load): exit 75 and report retryable when a rerun may fix every failure

crates/sqlload-python/src/bindings/models.rs

@@ -4,7 +4,7 @@ use std::time::Duration;
 use arrow_array::RecordBatch;
 use arrow_pyarrow::ToPyArrow;
 use arrow_schema::SchemaRef;
-use pyo3::exceptions::{PyRuntimeError, PyValueError};
+use pyo3::exceptions::{PyConnectionError, PyRuntimeError, PyValueError};
 use pyo3::pybacked::PyBackedBytes;
 use pyo3::types::{PyAnyMethods, PyDict, PyDictMethods, PyList, PyListMethods};
 use pyo3::{Bound, PyAny, PyRef, PyResult, Python, pyclass, pymethods};
@@ -145,6 +145,12 @@ impl BatchStream {
                 Polled::Failed(CrossError::Data(reason)) => {
                     return Err(PyValueError::new_err(reason));
                 }
+                Polled::Failed(CrossError::Http(error)) if error.retryable => {
+                    return Err(PyConnectionError::new_err(error.message));
+                }
+                Polled::Failed(CrossError::Http(error)) => {
+                    return Err(PyValueError::new_err(error.message));
+                }
                 Polled::Pending | Polled::State(_) => py.check_signals()?,
                 Polled::Finished => return Ok(None),
             }

crates/sqlload-python/src/crossing/_helpers/parser.rs

@@ -272,7 +272,7 @@ impl Output {
                 open && failed.is_none()
             },
         )
-        .map_err(CrossError::Data)?;
+        .map_err(CrossError::Http)?;
         match failed {
             Some(reason) => Err(CrossError::Data(reason)),
             None => Ok(open),

crates/sqlload-python/src/crossing/errors.rs

@@ -1,4 +1,5 @@
 use pyo3::PyErr;
+use sqlload::python_sources::errors::ReadError;
 
 /// Why a table generator's data could not be read.
 #[derive(Debug)]
@@ -7,13 +8,16 @@ pub enum CrossError {
     Python(PyErr),
     /// The data could not become Arrow (malformed JSON, a non-object row).
     Data(String),
+    /// An `sl.http` job failed; it says whether running it again may succeed.
+    Http(ReadError),
 }
 
 impl std::fmt::Display for CrossError {
     fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
             Self::Python(error) => write!(formatter, "{error}"),
             Self::Data(reason) => formatter.write_str(reason),
+            Self::Http(error) => write!(formatter, "{error}"),
         }
     }
 }

crates/sqlload-python/src/host/_helpers/python_errors.rs

@@ -1,5 +1,7 @@
+use pyo3::exceptions::{PyConnectionError, PyTimeoutError};
 use pyo3::types::{PyStringMethods, PyTracebackMethods, PyTypeMethods};
 use pyo3::{PyErr, Python};
+use sqlload::python_sources::errors::ReadError;
 
 /// A Python exception as one line: its type and message, then where in the user's code it was raised.
 pub(crate) fn error_text(error: &PyErr) -> String {
@@ -21,3 +23,14 @@ pub(crate) fn error_text(error: &PyErr) -> String {
         format!("{kind}: {}{place}", error.value(py))
     })
 }
+
+/// A table function's exception as a read error, retryable when it is a `ConnectionError` or `TimeoutError` (or a subclass), as a dropped connection or an `sl.http` job that outlasted its retries raises.
+pub(crate) fn read_error(error: &PyErr) -> ReadError {
+    let retryable = Python::attach(|py| {
+        error.is_instance_of::<PyConnectionError>(py) || error.is_instance_of::<PyTimeoutError>(py)
+    });
+    ReadError {
+        message: error_text(error),
+        retryable,
+    }
+}

crates/sqlload-python/src/host/models.rs

@@ -3,6 +3,7 @@ use std::time::Duration;
 
 use pyo3::types::PyAnyMethods;
 use pyo3::{PyResult, Python};
+use sqlload::python_sources::errors::ReadError;
 use sqlload::python_sources::models::{PythonSource, TableRead};
 use sqlload::python_sources::types::{BatchStream, PythonHost, Streamed};
 
@@ -13,7 +14,7 @@ use crate::crossing::errors::CrossError;
 use crate::crossing::main::default_parse_threads::default_parse_threads;
 use crate::crossing::main::table_batches::table_batches;
 use crate::crossing::models::{CrossingOptions, Polled, ReadContext, TableBatches};
-use crate::host::_helpers::python_errors::error_text;
+use crate::host::_helpers::python_errors::{error_text, read_error};
 use crate::host::constants::RUNTIME_MODULE;
 
 /// Runs Python sources in this interpreter through `sqlload._runtime`.
@@ -99,13 +100,14 @@ impl PythonHost for ExtensionHost {
 pub struct PythonBatches(TableBatches);
 
 impl BatchStream for PythonBatches {
-    fn next_batch(&mut self) -> Option<Result<Streamed, String>> {
+    fn next_batch(&mut self) -> Option<Result<Streamed, ReadError>> {
         loop {
             match self.0.poll(Duration::from_secs(3600)) {
                 Polled::Batch(batch) => return Some(Ok(Streamed::Batch(batch))),
                 Polled::State(state) => return Some(Ok(Streamed::State(state))),
-                Polled::Failed(CrossError::Python(error)) => return Some(Err(error_text(&error))),
-                Polled::Failed(CrossError::Data(reason)) => return Some(Err(reason)),
+                Polled::Failed(CrossError::Python(error)) => return Some(Err(read_error(&error))),
+                Polled::Failed(CrossError::Data(reason)) => return Some(Err(reason.into())),
+                Polled::Failed(CrossError::Http(error)) => return Some(Err(error)),
                 Polled::Pending => {}
                 Polled::Finished => return None,
             }

crates/sqlload/src/adapters/postgres/_helpers/session.rs

@@ -30,7 +30,7 @@ pub(crate) async fn session(
     match (closed, destination.reconnects) {
         (None, _) | (Some(true), true) => *held = Some(connect_session(destination).await?),
         (Some(true), false) => {
-            return Err(LoaderError::Invalid(
+            return Err(LoaderError::Transient(
                 "the table's Postgres session closed, and its staging table with it; nothing it had not committed was written".to_owned(),
             ));
         }

crates/sqlload/src/cdc/_helpers/stream.rs

@@ -1222,16 +1222,16 @@ pub(crate) async fn load_source(
     run: &LoadRun<'_>,
     name: &str,
     (tables, follow): (&[&TablePlan], Option<&CancellationToken>),
-) -> Result<(BTreeMap<String, u64>, Option<String>), String> {
+) -> Result<(BTreeMap<String, u64>, Option<String>), LoaderError> {
     let first = tables.first().map(|table| &table.spec);
     let (Some(source), Some(cdc), Some(destination)) = (
         run.project.sources.get(name),
         first.and_then(|spec| spec.cdc.as_ref()),
         first.and_then(|spec| run.destinations.get(&spec.target)),
     ) else {
-        return Err(format!(
+        return Err(LoaderError::Invalid(format!(
             "source {name} is not a connected postgres_cdc source"
-        ));
+        )));
     };
     let unit = CdcSource {
         run,
@@ -1242,13 +1242,14 @@ pub(crate) async fn load_source(
         destination,
         follow,
     };
-    unit.load().await.map_err(|error| error.to_string())
+    unit.load().await
 }
 
-/// Each table's outcome from its source's load.
+/// Each table's outcome from its source's load; `retryable` when its failure may pass on a rerun.
 pub(crate) fn source_outcomes(
     tables: &[&TablePlan],
     loaded: &Result<BTreeMap<String, u64>, String>,
+    retryable: bool,
     elapsed: Duration,
 ) -> Vec<TableOutcome> {
     let mut outcomes: Vec<TableOutcome> = Vec::with_capacity(tables.len());
@@ -1268,6 +1269,7 @@ pub(crate) fn source_outcomes(
                     ..TableLoaded::default()
                 })
                 .map_err(Clone::clone),
+            retryable: loaded.is_err() && retryable,
             elapsed,
         });
     }
@@ -1360,7 +1362,9 @@ pub(crate) async fn follow_round<'a>(
         );
         return (all, Err(error));
     }
-    let followed = load_source(run, name, (&tables, Some(stop))).await;
+    let followed = load_source(run, name, (&tables, Some(stop)))
+        .await
+        .map_err(|error| error.to_string());
     (tables, followed)
 }
 
@@ -1428,12 +1432,13 @@ pub(crate) fn failed_outcomes(
     error: String,
     elapsed: Duration,
 ) -> Vec<TableOutcome> {
-    let outcomes = source_outcomes(tables, &Err(error.clone()), elapsed);
+    let outcomes = source_outcomes(tables, &Err(error.clone()), false, elapsed);
     match outcomes.is_empty() {
         true => vec![TableOutcome {
             table: "load --follow".to_owned(),
             origin: None,
             loaded: Err(error),
+            retryable: false,
             elapsed,
         }],
         false => outcomes,

crates/sqlload/src/cdc/main/follow_cdc.rs

@@ -46,7 +46,7 @@ pub async fn follow_cdc(
         let reason = match next.1 {
             Ok(None) => {
                 tracing::info!(rows = rows.values().sum::<u64>(), "stopped following");
-                let outcomes = source_outcomes(&tables, &Ok(rows), started.elapsed());
+                let outcomes = source_outcomes(&tables, &Ok(rows), false, started.elapsed());
                 return with_changes(outcomes, applied);
             }
             Ok(Some(reason)) => Ok(reason),

crates/sqlload/src/cdc/main/load_cdc.rs

@@ -11,17 +11,28 @@ pub async fn load_cdc(run: &LoadRun<'_>, plan: &ProjectPlan) -> Vec<TableOutcome
         let loaded = match blocked.is_empty() {
             true => load_source(run, name, (&tables, None))
                 .await
-                .map(|(rows, _)| rows),
-            false => Err(format!(
-                "source {name} does not load while {} {} blocked: its tables share one slot, so loading the rest would skip the blocked tables' changes",
-                blocked.join(", "),
-                match blocked.len() {
-                    1 => "is",
-                    _ => "are",
-                }
+                .map(|(rows, _)| rows)
+                .map_err(|error| (error.is_retryable(), error.to_string())),
+            false => Err((
+                false,
+                format!(
+                    "source {name} does not load while {} {} blocked: its tables share one slot, so loading the rest would skip the blocked tables' changes",
+                    blocked.join(", "),
+                    match blocked.len() {
+                        1 => "is",
+                        _ => "are",
+                    }
+                ),
             )),
         };
-        outcomes.extend(source_outcomes(&tables, &loaded, started.elapsed()));
+        let retryable = matches!(loaded, Err((true, _)));
+        let loaded = loaded.map_err(|(_, error)| error);
+        outcomes.extend(source_outcomes(
+            &tables,
+            &loaded,
+            retryable,
+            started.elapsed(),
+        ));
     }
     outcomes
 }

crates/sqlload/src/command/_helpers/copy_command.rs

@@ -3,15 +3,18 @@
 use std::io::Write;
 
 use crate::command::_helpers::loader_commands::run_loader;
+use crate::command::errors::CommandError;
 use crate::command::models::{Command, LoaderArgs, LoaderRequest};
 use crate::copy::constants::PROJECT_HEADER;
 use crate::copy::main::copy_project::copy_project;
 use crate::copy::models::CopyRequest;
 
 /// Run `request`'s `copy`: its source into its destination with `request.text`'s connections (empty when there is no sqlload.toml), or print the copy as a project with `--print-toml`.
-pub(crate) async fn run_copy(request: LoaderRequest<'_>) -> Result<(Vec<String>, bool), String> {
+pub(crate) async fn run_copy(
+    request: LoaderRequest<'_>,
+) -> Result<(Vec<String>, u8), CommandError> {
     let Command::Copy(args) = request.command else {
-        return Err("copy runs only for sqlload copy".to_owned());
+        return Err("copy runs only for sqlload copy".to_owned().into());
     };
     let copy_request = CopyRequest {
         source: args.source,
@@ -26,7 +29,7 @@ pub(crate) async fn run_copy(request: LoaderRequest<'_>) -> Result<(Vec<String>,
         let _ = writeln!(std::io::stderr(), "warning: {warning}");
     }
     if args.print_toml {
-        return Ok((vec![format!("{PROJECT_HEADER}{}", copy.printed)], true));
+        return Ok((vec![format!("{PROJECT_HEADER}{}", copy.printed)], 0));
     }
     run_loader(LoaderRequest {
         text: &copy.text,