diff --git a/crates/sqlload-python/src/bindings/models.rs b/crates/sqlload-python/src/bindings/models.rs
index 4dc6685f..249036ca 100644
--- a/crates/sqlload-python/src/bindings/models.rs
+++ b/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),
             }
diff --git a/crates/sqlload-python/src/crossing/_helpers/parser.rs b/crates/sqlload-python/src/crossing/_helpers/parser.rs
index fa6cd15a..10b33e9a 100644
--- a/crates/sqlload-python/src/crossing/_helpers/parser.rs
+++ b/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),
diff --git a/crates/sqlload-python/src/crossing/errors.rs b/crates/sqlload-python/src/crossing/errors.rs
index 8c65e997..f25fe209 100644
--- a/crates/sqlload-python/src/crossing/errors.rs
+++ b/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,6 +8,8 @@ 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 {
@@ -14,6 +17,7 @@ impl std::fmt::Display for CrossError {
         match self {
             Self::Python(error) => write!(formatter, "{error}"),
             Self::Data(reason) => formatter.write_str(reason),
+            Self::Http(error) => write!(formatter, "{error}"),
         }
     }
 }
diff --git a/crates/sqlload-python/src/host/_helpers/python_errors.rs b/crates/sqlload-python/src/host/_helpers/python_errors.rs
index c2ed10fe..8ab09ffd 100644
--- a/crates/sqlload-python/src/host/_helpers/python_errors.rs
+++ b/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,
+    }
+}
diff --git a/crates/sqlload-python/src/host/models.rs b/crates/sqlload-python/src/host/models.rs
index cff06202..687a2c55 100644
--- a/crates/sqlload-python/src/host/models.rs
+++ b/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,
             }
diff --git a/crates/sqlload/src/adapters/postgres/_helpers/session.rs b/crates/sqlload/src/adapters/postgres/_helpers/session.rs
index 58634d89..93a7f743 100644
--- a/crates/sqlload/src/adapters/postgres/_helpers/session.rs
+++ b/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(),
             ));
         }
diff --git a/crates/sqlload/src/cdc/_helpers/stream.rs b/crates/sqlload/src/cdc/_helpers/stream.rs
index 2c73ed05..d8d42eeb 100644
--- a/crates/sqlload/src/cdc/_helpers/stream.rs
+++ b/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,
diff --git a/crates/sqlload/src/cdc/main/follow_cdc.rs b/crates/sqlload/src/cdc/main/follow_cdc.rs
index dc001835..6ca109e2 100644
--- a/crates/sqlload/src/cdc/main/follow_cdc.rs
+++ b/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),
diff --git a/crates/sqlload/src/cdc/main/load_cdc.rs b/crates/sqlload/src/cdc/main/load_cdc.rs
index dc668e04..24235298 100644
--- a/crates/sqlload/src/cdc/main/load_cdc.rs
+++ b/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
 }
diff --git a/crates/sqlload/src/command/_helpers/copy_command.rs b/crates/sqlload/src/command/_helpers/copy_command.rs
index e22014f9..04af1dae 100644
--- a/crates/sqlload/src/command/_helpers/copy_command.rs
+++ b/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,
diff --git a/crates/sqlload/src/command/_helpers/loader_commands.rs b/crates/sqlload/src/command/_helpers/loader_commands.rs
index 190cb051..73ae99ff 100644
--- a/crates/sqlload/src/command/_helpers/loader_commands.rs
+++ b/crates/sqlload/src/command/_helpers/loader_commands.rs
@@ -8,7 +8,8 @@ use crate::auth::main::allow_sign_in::allow_sign_in;
 use crate::command::_helpers::output::confirm_reload;
 use crate::command::_helpers::project_file::replace_file;
 use crate::command::_helpers::signals::wait_for_signal;
-use crate::command::constants::{FORCED_STOP_EXIT, RELOAD_DECLINED};
+use crate::command::constants::{FORCED_STOP_EXIT, RELOAD_DECLINED, RETRYABLE_EXIT};
+use crate::command::errors::CommandError;
 use crate::command::models::{Command, ContractAction, LoaderArgs, LoaderRequest, StateAction};
 use crate::contract::main::compare_contracts::compare_contracts;
 use crate::contract::main::render_contract::render_contract;
@@ -25,7 +26,7 @@ use crate::loader::main::render_plan::render_plan;
 use crate::loader::main::render_tally::render_tally;
 use crate::loader::main::state_command::state_command;
 use crate::loader::models::{
-    LoadRun, PlanOptions, ProjectPlan, RunReport, StateKind, StateRequest,
+    LoadRun, PlanOptions, ProjectPlan, RunReport, StateKind, StateRequest, TableOutcome,
 };
 use crate::project::main::parse_project::parse_project;
 use crate::project::main::parse_python_project::parse_python_project;
@@ -87,8 +88,19 @@ async fn load_project_config(request: &LoaderRequest<'_>) -> Result<ProjectConfi
     parse_python_project(request.text, &lookup, &found).map_err(|error| error.to_string())
 }
 
-/// Run a loader command against the project in `request.text`: the lines to print, and whether it succeeded. Under `--json` every failure is a JSON error document.
-pub(crate) async fn run_loader(request: LoaderRequest<'_>) -> Result<(Vec<String>, bool), String> {
+/// The exit code of a command that succeeded (`ok`), or failed only in `retryable` ways, or failed otherwise.
+pub(crate) fn exit_code(ok: bool, retryable: bool) -> u8 {
+    match (ok, retryable) {
+        (true, _) => 0,
+        (false, true) => RETRYABLE_EXIT,
+        (false, false) => 1,
+    }
+}
+
+/// Run a loader command against the project in `request.text`: the lines to print, and its exit code. Under `--json` every failure is a JSON error document.
+pub(crate) async fn run_loader(
+    request: LoaderRequest<'_>,
+) -> Result<(Vec<String>, u8), CommandError> {
     let name = match &request.command {
         Command::Apply(_) => "apply",
         Command::Load(_) => "load",
@@ -98,8 +110,8 @@ pub(crate) async fn run_loader(request: LoaderRequest<'_>) -> Result<(Vec<String
         Command::Plan(args) | Command::Apply(args) if args.follow => {
             let error = "--follow and --max-duration apply to load only";
             return match args.json {
-                true => Ok((vec![json_error(name, error)], false)),
-                false => Err(error.to_owned()),
+                true => Ok((vec![json_error(name, error, false)], 1)),
+                false => Err(error.to_owned().into()),
             };
         }
         Command::Plan(args) | Command::Apply(args) | Command::Load(args) => args.json,
@@ -117,12 +129,15 @@ pub(crate) async fn run_loader(request: LoaderRequest<'_>) -> Result<(Vec<String
     allow_sign_in(!(json || contract_json));
     let project = match load_project_config(&request).await {
         Ok(project) => project,
-        Err(error) => return config_failure(&request.command, error),
+        Err(error) => return config_failure(&request.command, error).map_err(Into::into),
     };
     let outcome = match request.command {
         Command::Contract { action } => {
             let file = (request.config, request.text);
-            run_contract(&project, action, file, &request.variables).await
+            run_contract(&project, action, file, &request.variables)
+                .await
+                .map(|(lines, agrees)| (lines, exit_code(agrees, false)))
+                .map_err(Into::into)
         }
         command => run_command(&project, command, request.host.clone(), request.no_color).await,
     };
@@ -130,7 +145,10 @@ pub(crate) async fn run_loader(request: LoaderRequest<'_>) -> Result<(Vec<String
         let _ = tokio::task::spawn_blocking(move || host.finish()).await;
     }
     match (outcome, json) {
-        (Err(error), true) => Ok((vec![json_error(name, &error)], false)),
+        (Err(error), true) => Ok((
+            vec![json_error(name, &error.message, error.retryable)],
+            exit_code(false, error.retryable),
+        )),
         (outcome, _) => outcome,
     }
 }
@@ -143,8 +161,8 @@ fn problems(error: &str) -> Vec<String> {
     }
 }
 
-/// `error` in sqlbuild's error shape, `{"error": {"code", "message", "help"}}`: each problem's reason in `message` and its fix (the text after `Fix:`) in `help`, `null` when none has one.
-pub(crate) fn json_error(code: &str, error: &str) -> String {
+/// `error` in sqlbuild's error shape, `{"error": {"code", "message", "help", "retryable"}}`: each problem's reason in `message` and its fix (the text after `Fix:`) in `help`, `null` when none has one; `retryable` when running the command again unchanged may succeed.
+pub(crate) fn json_error(code: &str, error: &str, retryable: bool) -> String {
     let mut messages: Vec<String> = Vec::new();
     let mut helps: Vec<String> = Vec::new();
     for problem in problems(error) {
@@ -157,21 +175,24 @@ pub(crate) fn json_error(code: &str, error: &str) -> String {
         }
     }
     let help = (!helps.is_empty()).then(|| helps.join("\n"));
-    serde_json::json!({ "error": { "code": code, "message": messages.join("\n"), "help": help } })
+    serde_json::json!({ "error": { "code": code, "message": messages.join("\n"), "help": help, "retryable": retryable } })
         .to_string()
 }
 
 /// A project that does not parse: a failed plan's `Errors` for `plan`, `apply` and `load` (as JSON under `--json`), else the error.
-fn config_failure(command: &Command, error: String) -> Result<(Vec<String>, bool), String> {
+fn config_failure(command: &Command, error: String) -> Result<(Vec<String>, u8), String> {
     let (Command::Plan(args) | Command::Apply(args) | Command::Load(args)) = command else {
         return Err(error);
     };
     match args.json {
-        true => Ok((vec![json_error("config", &error)], false)),
-        false => Ok(render_plan(&ProjectPlan {
-            problems: problems(&error),
-            ..ProjectPlan::default()
-        })),
+        true => Ok((vec![json_error("config", &error, false)], 1)),
+        false => {
+            let (lines, ok) = render_plan(&ProjectPlan {
+                problems: problems(&error),
+                ..ProjectPlan::default()
+            });
+            Ok((lines, exit_code(ok, false)))
+        }
     }
 }
 
@@ -180,22 +201,23 @@ async fn run_command(
     command: Command,
     host: Option<Arc<dyn PythonHost>>,
     no_color: bool,
-) -> Result<(Vec<String>, bool), String> {
+) -> Result<(Vec<String>, u8), CommandError> {
     if let Command::Debug { no_connection } = command {
         let (lines, ok) = debug_project(project, !no_connection).await;
         return match ok {
-            true => Ok((lines, true)),
-            false => Err(lines.join("\n")),
+            true => Ok((lines, 0)),
+            false => Err(lines.join("\n").into()),
         };
     }
-    let destinations = connect_targets(project)
-        .await
-        .map_err(|error| error.to_string())?;
+    let destinations = connect_targets(project).await;
     if let Command::LoggedState { action } = command {
+        let destinations = destinations.map_err(|error| error.to_string())?;
         return run_state(project, &destinations, action)
             .await
-            .map(|lines| (lines, true));
+            .map(|lines| (lines, 0))
+            .map_err(Into::into);
     }
+    let destinations = destinations?;
     let (args, applies, loads) = match command {
         Command::Plan(args) => (args, false, false),
         Command::Apply(args) => (args, true, false),
@@ -203,7 +225,8 @@ async fn run_command(
         _ => {
             return Err(
                 "this command runs on exporter configurations; this file is a loader project"
-                    .to_owned(),
+                    .to_owned()
+                    .into(),
             );
         }
     };
@@ -214,22 +237,21 @@ async fn run_command(
     );
     let started = (chrono::Utc::now(), std::time::Instant::now());
     let options = options(args)?;
-    let plan = plan_project(project, &destinations, &options)
-        .await
-        .map_err(|error| error.to_string())?;
+    let plan = plan_project(project, &destinations, &options).await?;
     if json && !applies {
-        let (text, ok) = json_report(&plan, None);
-        return Ok((vec![text], ok));
+        let (text, ok, retryable) = json_report(&plan, None);
+        return Ok((vec![text], exit_code(ok, retryable)));
     }
     let (mut lines, planned) = match json {
         true => (Vec::new(), true),
         false => render_plan(&plan),
     };
+    let plan_retryable = plan.only_retryable_problems();
     if !applies {
-        return Ok((lines, planned));
+        return Ok((lines, exit_code(planned, plan_retryable)));
     }
     if let Some(problem) = follow.and_then(|_| follow_problem(&plan)) {
-        return Err(problem);
+        return Err(problem.into());
     }
     let flag = match (options.reload, options.resnapshot) {
         (true, true) => "--reload and --resnapshot",
@@ -251,17 +273,16 @@ async fn run_command(
         return Err(format!(
             "{flag} rebuilds {}, replacing what they hold, and --json cannot ask to confirm it; add --yes",
             rebuilt.join(", ")
-        ));
+        )
+        .into());
     }
     if !rebuilt.is_empty() {
         match confirm_reload(&lines, flag, &rebuilt, no_color)? {
             true => lines.clear(),
-            false => return Ok((vec![RELOAD_DECLINED.to_owned()], false)),
+            false => return Ok((vec![RELOAD_DECLINED.to_owned()], 1)),
         }
     }
-    let applied = apply_project(&plan, &destinations)
-        .await
-        .map_err(|error| error.to_string())?;
+    let applied = apply_project(&plan, &destinations).await?;
     let run = LoadRun {
         project,
         destinations: &destinations,
@@ -281,12 +302,16 @@ async fn run_command(
             started: started.0,
             elapsed: started.1.elapsed(),
         };
-        let (text, ok) = json_report(&plan, Some(&run));
-        return Ok((vec![text], ok));
+        let (text, ok, retryable) = json_report(&plan, Some(&run));
+        return Ok((vec![text], exit_code(ok, retryable)));
     }
     let (tallied, completed) = render_tally(&applied, &outcomes, started.1.elapsed());
     lines.extend(tallied);
-    Ok((lines, planned && completed))
+    let retryable = outcomes.iter().all(TableOutcome::passes_on_rerun);
+    Ok((
+        lines,
+        exit_code(planned && completed, plan_retryable && retryable),
+    ))
 }
 
 /// Why `load --follow` cannot follow `plan`: it follows exactly one `postgres_cdc` source, and nothing else, since it never finishes to load other tables after; when it selects none, the plan's problems say why.
diff --git a/crates/sqlload/src/command/constants.rs b/crates/sqlload/src/command/constants.rs
index b4ee69c6..c20fcfef 100644
--- a/crates/sqlload/src/command/constants.rs
+++ b/crates/sqlload/src/command/constants.rs
@@ -8,3 +8,6 @@ pub const RELOAD_DECLINED: &str = "✗ Reload cancelled: nothing was changed";
 
 /// The exit code of a `load --follow` stopped by a second Ctrl-C or SIGTERM, as a shell reports SIGINT.
 pub const FORCED_STOP_EXIT: i32 = 130;
+
+/// The exit code of a run that failed only in ways running it again unchanged may fix (`EX_TEMPFAIL`): a dropped or refused connection, throttling or server errors that outlasted their retries, or a table another run is loading.
+pub const RETRYABLE_EXIT: u8 = 75;
diff --git a/crates/sqlload/src/command/errors.rs b/crates/sqlload/src/command/errors.rs
new file mode 100644
index 00000000..80efc421
--- /dev/null
+++ b/crates/sqlload/src/command/errors.rs
@@ -0,0 +1,24 @@
+/// Why a loader command could not finish, and whether running it again unchanged may succeed.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct CommandError {
+    pub message: String,
+    pub retryable: bool,
+}
+
+impl From<String> for CommandError {
+    fn from(message: String) -> Self {
+        Self {
+            message,
+            retryable: false,
+        }
+    }
+}
+
+impl From<crate::loader::errors::LoaderError> for CommandError {
+    fn from(error: crate::loader::errors::LoaderError) -> Self {
+        Self {
+            retryable: error.is_retryable(),
+            message: error.to_string(),
+        }
+    }
+}
diff --git a/crates/sqlload/src/command/main/run_args_with_host.rs b/crates/sqlload/src/command/main/run_args_with_host.rs
index bf8c4978..0d2a9034 100644
--- a/crates/sqlload/src/command/main/run_args_with_host.rs
+++ b/crates/sqlload/src/command/main/run_args_with_host.rs
@@ -9,9 +9,10 @@ use clap::Parser;
 use crate::command::_helpers::copy_command::run_copy;
 use crate::command::_helpers::exporter_commands::run_exporter_command;
 use crate::command::_helpers::limits::runtime;
-use crate::command::_helpers::loader_commands::{json_error, run_loader};
+use crate::command::_helpers::loader_commands::{exit_code, json_error, run_loader};
 use crate::command::_helpers::logging::init_logging;
 use crate::command::_helpers::output::{config_text, print_lines};
+use crate::command::errors::CommandError;
 use crate::command::models::{Cli, Command, LoaderRequest};
 use crate::project::main::is_project::is_project;
 use crate::python_sources::types::PythonHost;
@@ -39,7 +40,7 @@ where
     let copies = matches!(&cli.command, Command::Copy(_));
     let text = match config_text(&cli.config, copies) {
         Ok(text) => text,
-        Err(message) => return failed(json, "config", &message),
+        Err(message) => return failed(json, "config", &message.into()),
     };
     let cli_threads = match &cli.command {
         Command::Plan(args) | Command::Apply(args) | Command::Load(args) => args.concurrency,
@@ -47,7 +48,7 @@ where
     };
     let runtime = match runtime(cli_threads, &text) {
         Ok(runtime) => runtime,
-        Err(message) => return failed(json, name, &message),
+        Err(message) => return failed(json, name, &message.into()),
     };
     let outcome = runtime.block_on(async {
         if copies || is_project(&text) {
@@ -66,24 +67,28 @@ where
         }
         run_exporter_command(&cli.config, cli.command)
             .await
-            .map(|lines| (lines, true))
+            .map(|lines| (lines, 0))
+            .map_err(CommandError::from)
     });
     match outcome {
-        Ok((lines, ok)) => {
+        Ok((lines, code)) => {
             print_lines(lines, cli.no_color);
-            u8::from(!ok)
+            code
         }
         Err(error) => failed(json, "config", &error),
     }
 }
 
-/// Report a failure and return exit code 1: as one JSON error on stdout under `--json`, so its output always parses, else as `error:` on stderr.
-fn failed(json: bool, code: &str, message: &str) -> u8 {
+/// Report a failure and return its exit code (75 when a rerun may succeed, else 1): as one JSON error on stdout under `--json`, so its output always parses, else as `error:` on stderr.
+fn failed(json: bool, code: &str, error: &CommandError) -> u8 {
     match json {
-        true => print_lines(vec![json_error(code, message)], true),
+        true => print_lines(
+            vec![json_error(code, &error.message, error.retryable)],
+            true,
+        ),
         false => {
-            let _ = writeln!(std::io::stderr(), "error: {message}");
+            let _ = writeln!(std::io::stderr(), "error: {}", error.message);
         }
     }
-    1
+    exit_code(false, error.retryable)
 }
diff --git a/crates/sqlload/src/command/mod.rs b/crates/sqlload/src/command/mod.rs
index f54f61ca..c1af6c04 100644
--- a/crates/sqlload/src/command/mod.rs
+++ b/crates/sqlload/src/command/mod.rs
@@ -2,6 +2,7 @@
 
 pub(crate) mod _helpers;
 pub mod constants;
+pub mod errors;
 pub mod main;
 pub mod models;
 
diff --git a/crates/sqlload/src/command/tests/helpers.rs b/crates/sqlload/src/command/tests/helpers.rs
index 39a64574..53c3ef48 100644
--- a/crates/sqlload/src/command/tests/helpers.rs
+++ b/crates/sqlload/src/command/tests/helpers.rs
@@ -2,7 +2,8 @@ use std::path::Path;
 
 use clap::Parser;
 
-use crate::command::_helpers::loader_commands::{options, run_loader};
+use crate::command::_helpers::loader_commands::{exit_code, options, run_loader};
+use crate::command::errors::CommandError;
 use crate::command::models::{Cli, Command, LoaderRequest};
 
 /// `sqlload debug --no-connection` on `project` without a Python host, as the standalone executable runs it; the error text, or `""`.
@@ -21,12 +22,16 @@ pub(crate) fn debug_without_host(project: &str) -> String {
         .enable_all()
         .build()
         .map_err(|error| error.to_string())
-        .and_then(|runtime| runtime.block_on(run_loader(request)))
+        .and_then(|runtime| {
+            runtime
+                .block_on(run_loader(request))
+                .map_err(|error| error.message)
+        })
         .err()
         .unwrap_or_default()
 }
 
-/// The command line `argv` (after `sqlload`) on `project` without a Python host: its output, or its error, then `exit 0` or `exit 1` on its own line.
+/// The command line `argv` (after `sqlload`) on `project` without a Python host: its output, or its error, then its exit code as `exit N` on its own line.
 pub(crate) fn run_without_host(project: &str, argv: &[&str]) -> String {
     let Ok(cli) = Cli::try_parse_from(["sqlload"].iter().chain(argv)) else {
         return "unparsed".to_owned();
@@ -42,11 +47,14 @@ pub(crate) fn run_without_host(project: &str, argv: &[&str]) -> String {
     tokio::runtime::Builder::new_current_thread()
         .enable_all()
         .build()
-        .map_err(|error| error.to_string())
+        .map_err(|error| CommandError::from(error.to_string()))
         .and_then(|runtime| runtime.block_on(run_loader(request)))
         .map_or_else(
-            |error| format!("{error}\nexit 1"),
-            |(lines, ok)| format!("{}\nexit {}", lines.join("\n"), u8::from(!ok)),
+            |error| {
+                let code = exit_code(false, error.retryable);
+                format!("{}\nexit {code}", error.message)
+            },
+            |(lines, code)| format!("{}\nexit {code}", lines.join("\n")),
         )
 }
 
diff --git a/crates/sqlload/src/command/tests/test_loader_commands.rs b/crates/sqlload/src/command/tests/test_loader_commands.rs
index bd7564c8..3efb4110 100644
--- a/crates/sqlload/src/command/tests/test_loader_commands.rs
+++ b/crates/sqlload/src/command/tests/test_loader_commands.rs
@@ -1,10 +1,12 @@
 use crate::command::tests::helpers::{debug_without_host, run_without_host};
 use crate::command::tests::test_types::{
-    FolderOverlapTestCase, LoaderCommandTestCase, PlanFailureTestCase,
+    ExitCodeTestCase, FolderOverlapTestCase, LoaderCommandTestCase, PlanFailureTestCase,
 };
 
 const TARGET: &str = "[connections.wh]\nkind = \"duckdb\"\ndatabase = \":memory:\"\n[targets.dev]\nconnection = \"wh\"\nloader_schema = \"RAW\"\n";
 const NO_LOAD_METHOD: &str = "[connections.wh]\nkind = \"snowflake\"\naccount = \"a\"\nuser = \"u\"\nauth = { method = \"pat\", token = \"t\" }\ndatabase = \"D\"\n[targets.dev]\nconnection = \"wh\"\nloader_schema = \"RAW\"\n";
+const REFUSING_TARGET: &str = "[connections.pg]\nkind = \"postgres\"\nhost = \"127.0.0.1\"\nport = 1\nuser = \"loader\"\npassword = \"p\"\ndatabase = \"app\"\nsslmode = \"disable\"\n[targets.dev]\nconnection = \"pg\"\nloader_schema = \"raw\"\n[sources.orders]\nkind = \"files\"\ndescription = \"Test table.\"\nconnection = \"lake\"\npath = \"orders/\"\nformat = \"csv\"\nwrite_strategy = \"append\"\n[connections.lake]\nkind = \"filesystem\"\nroot = \"{root}\"\n";
+const UNREACHABLE_SOURCE: &str = "[connections.wh]\nkind = \"duckdb\"\ndatabase = \":memory:\"\n[targets.dev]\nconnection = \"wh\"\nloader_schema = \"RAW\"\n[connections.app]\nkind = \"postgres\"\nhost = \"127.0.0.1\"\nport = 1\nuser = \"u\"\npassword = \"p\"\ndatabase = \"app\"\nsslmode = \"disable\"\n[sources.app]\nkind = \"postgres\"\nconnection = \"app\"\ndestination = \"dev\"\ninclude = [\"public.*\"]\n";
 
 #[test]
 fn given_the_standalone_executable_when_a_project_has_python_paths_then_it_says_to_install_the_python_package()
@@ -129,3 +131,75 @@ fn given_a_files_source_into_a_files_target_when_planning_then_an_overlapping_fo
     }
     Ok(())
 }
+
+#[test]
+fn given_a_failure_a_rerun_may_pass_when_loading_then_it_exits_75_and_says_so_else_1()
+-> Result<(), String> {
+    let root = tempfile::tempdir().map_err(|error| error.to_string())?;
+    std::fs::create_dir_all(root.path().join("orders")).map_err(|error| error.to_string())?;
+    std::fs::write(root.path().join("orders/a.csv"), "id\n1\n")
+        .map_err(|error| error.to_string())?;
+    let test_cases = [
+        ExitCodeTestCase {
+            description: "--json: a target that refuses the connection is retryable",
+            project: REFUSING_TARGET,
+            argv: &["load", "--json"],
+            expected_output: "\"retryable\":true",
+            expected_exit: "exit 75",
+        },
+        ExitCodeTestCase {
+            description: "text: the same failure exits 75 too",
+            project: REFUSING_TARGET,
+            argv: &["load"],
+            expected_output: "cannot connect to Postgres at",
+            expected_exit: "exit 75",
+        },
+        ExitCodeTestCase {
+            description: "--json: a source that refuses the connection is a retryable plan error",
+            project: UNREACHABLE_SOURCE,
+            argv: &["load", "--json"],
+            expected_output: "\"retryable\": true",
+            expected_exit: "exit 75",
+        },
+        ExitCodeTestCase {
+            description: "text: plan exits 75 for the same source",
+            project: UNREACHABLE_SOURCE,
+            argv: &["plan"],
+            expected_output: "cannot connect to Postgres at 127.0.0.1:1",
+            expected_exit: "exit 75",
+        },
+        ExitCodeTestCase {
+            description: "--json: an unreachable source beside a plan error that is not retryable exits 1",
+            project: "[connections.wh]\nkind = \"duckdb\"\ndatabase = \":memory:\"\n[targets.dev]\nconnection = \"wh\"\nloader_schema = \"RAW\"\n[connections.app]\nkind = \"postgres\"\nhost = \"127.0.0.1\"\nport = 1\nuser = \"u\"\npassword = \"p\"\ndatabase = \"app\"\nsslmode = \"disable\"\n[sources.app]\nkind = \"postgres\"\nconnection = \"app\"\ndestination = \"dev\"\ninclude = [\"public.*\"]\n[sources.orders]\nkind = \"files\"\ndescription = \"Test table.\"\nconnection = \"lake\"\npath = \"missing/\"\nformat = \"csv\"\nwrite_strategy = \"append\"\ndestination = \"dev\"\n[connections.lake]\nkind = \"filesystem\"\nroot = \"{root}\"\n",
+            argv: &["load", "--json"],
+            expected_output: "\"retryable\": false",
+            expected_exit: "exit 1",
+        },
+        ExitCodeTestCase {
+            description: "--json: a configuration that does not parse is not retryable",
+            project: NO_LOAD_METHOD,
+            argv: &["load", "--json"],
+            expected_output: "\"retryable\":false",
+            expected_exit: "exit 1",
+        },
+        ExitCodeTestCase {
+            description: "--json: a target that cannot open is not retryable",
+            project: "[connections.wh]\nkind = \"duckdb\"\ndatabase = \"/no/such/folder/wh.duckdb\"\n[targets.dev]\nconnection = \"wh\"\nloader_schema = \"RAW\"\n",
+            argv: &["load", "--json"],
+            expected_output: "\"retryable\":false",
+            expected_exit: "exit 1",
+        },
+    ];
+    for test_case in test_cases {
+        let project = test_case
+            .project
+            .replace("{root}", &root.path().display().to_string());
+        let output = run_without_host(&project, test_case.argv);
+        assert!(
+            output.contains(test_case.expected_output) && output.ends_with(test_case.expected_exit),
+            "{}: {output}",
+            test_case.description
+        );
+    }
+    Ok(())
+}
diff --git a/crates/sqlload/src/command/tests/test_types.rs b/crates/sqlload/src/command/tests/test_types.rs
index 79eead0b..8f27bd7c 100644
--- a/crates/sqlload/src/command/tests/test_types.rs
+++ b/crates/sqlload/src/command/tests/test_types.rs
@@ -46,3 +46,13 @@ pub(crate) struct FolderOverlapTestCase {
     /// The last line: `exit 0` or `exit 1`.
     pub(crate) expected_exit: &'static str,
 }
+
+pub(crate) struct ExitCodeTestCase {
+    pub(crate) description: &'static str,
+    pub(crate) project: &'static str,
+    pub(crate) argv: &'static [&'static str],
+    /// Text the output holds.
+    pub(crate) expected_output: &'static str,
+    /// The last line: `exit 1` or `exit 75`.
+    pub(crate) expected_exit: &'static str,
+}
diff --git a/crates/sqlload/src/http/errors.rs b/crates/sqlload/src/http/errors.rs
index c5d1be97..6d88cf02 100644
--- a/crates/sqlload/src/http/errors.rs
+++ b/crates/sqlload/src/http/errors.rs
@@ -47,7 +47,16 @@ impl ApiError {
 }
 
 impl From<reqwest::Error> for ApiError {
+    /// A request that could not be built, or an answer that is not the JSON expected, is a protocol error, which retrying would not change; anything else is transport, including a body cut off mid-read (which reqwest also calls a decode error).
     fn from(error: reqwest::Error) -> Self {
-        Self::Transport(error.without_url().to_string())
+        let not_json = error.is_decode()
+            && std::error::Error::source(&error)
+                .is_some_and(|source| source.is::<serde_json::Error>());
+        let protocol = error.is_builder() || not_json;
+        let text = error.without_url().to_string();
+        match protocol {
+            true => Self::Protocol(text),
+            false => Self::Transport(text),
+        }
     }
 }
diff --git a/crates/sqlload/src/http_source/_helpers/fetch.rs b/crates/sqlload/src/http_source/_helpers/fetch.rs
index 89629b48..c094127f 100644
--- a/crates/sqlload/src/http_source/_helpers/fetch.rs
+++ b/crates/sqlload/src/http_source/_helpers/fetch.rs
@@ -13,9 +13,10 @@ use crate::http_source::constants::{
     SKIPPED_REFUSAL_RENEWAL_AGE,
 };
 use crate::http_source::models::{Attempt, Calls, Credential, Fetched, StatusAction};
+use crate::python_sources::errors::ReadError;
 
-/// The body at `url`, or `None` when `on_status` skips its status; transport errors, 429 and 5xx are retried up to `calls.retries` times.
-pub(crate) async fn fetch(calls: &Calls, url: &Url) -> Result<Option<Fetched>, String> {
+/// The body at `url`, or `None` when `on_status` skips its status; transport errors, 429 and 5xx are retried up to `calls.retries` times, and are still worth retrying later once those run out.
+pub(crate) async fn fetch(calls: &Calls, url: &Url) -> Result<Option<Fetched>, ReadError> {
     let policy = RetryPolicy {
         first_delay: FIRST_RETRY_DELAY,
         max_delay: MAX_RETRY_DELAY,
@@ -43,7 +44,10 @@ pub(crate) async fn fetch(calls: &Calls, url: &Url) -> Result<Option<Fetched>, S
                 renewed = true;
             }
             Attempt::Retry { reason, .. } if attempt >= policy.max_attempts => {
-                return Err(format!("{reason} (after {} attempts)", policy.max_attempts));
+                return Err(ReadError::transient(format!(
+                    "{reason} (after {} attempts)",
+                    policy.max_attempts
+                )));
             }
             Attempt::Retry { after, .. } => {
                 let delay = after.unwrap_or_else(|| jittered(policy.delay(attempt)));
diff --git a/crates/sqlload/src/http_source/_helpers/job.rs b/crates/sqlload/src/http_source/_helpers/job.rs
index b35ba966..5d03c394 100644
--- a/crates/sqlload/src/http_source/_helpers/job.rs
+++ b/crates/sqlload/src/http_source/_helpers/job.rs
@@ -13,6 +13,7 @@ use crate::http_source::_helpers::oauth::token_url;
 use crate::http_source::_helpers::pages::paginate;
 use crate::http_source::_helpers::selector::{selected_rows, selector};
 use crate::http_source::_helpers::urls::{filled_path, placeholders};
+use crate::python_sources::errors::ReadError;
 use crate::row_batches::main::projection::projection;
 use crate::row_batches::main::projection_problem::projection_problem;
 use crate::row_batches::models::{Projection, ProjectionRules};
@@ -219,7 +220,7 @@ pub(crate) fn drive_job(
     job: HttpJob,
     connections: &Map<String, Value>,
     mut deliver: impl FnMut(HttpPage) -> bool,
-) -> Result<(), String> {
+) -> Result<(), ReadError> {
     let runtime = tokio::runtime::Builder::new_multi_thread()
         .worker_threads(JOB_THREADS)
         .enable_all()
@@ -237,7 +238,7 @@ pub(crate) fn drive_job(
     match runtime.block_on(task) {
         Ok(result) => result,
         Err(error) if error.is_cancelled() => Ok(()),
-        Err(error) => Err(format!("the sl.http job stopped: {error}")),
+        Err(error) => Err(format!("the sl.http job stopped: {error}").into()),
     }
 }
 
@@ -246,7 +247,7 @@ async fn run_job(
     job: HttpJob,
     connections: Map<String, Value>,
     pages: Sender<HttpPage>,
-) -> Result<(), String> {
+) -> Result<(), ReadError> {
     let http = job_client()?;
     match job {
         HttpJob::Paginate(request) => {
@@ -318,11 +319,11 @@ async fn run_fan_out(
     connections: &Map<String, Value>,
     http: &reqwest::Client,
     pages: Sender<HttpPage>,
-) -> Result<(), String> {
+) -> Result<(), ReadError> {
     let calls = request_calls(&fan_out.request, connections, http)?;
     let (key_sender, key_receiver) = channel::<Map<String, Value>>(fan_out.concurrency * 2);
     let keys = Arc::new(Mutex::new(key_receiver));
-    let mut tasks: JoinSet<Result<(), String>> = JoinSet::new();
+    let mut tasks: JoinSet<Result<(), ReadError>> = JoinSet::new();
     let feed = match fan_out.over.clone() {
         Over::Values { values } => KeyFeed::Values(values),
         Over::Paginate(over) => {
@@ -355,7 +356,7 @@ async fn feed_keys(
     feed: KeyFeed,
     wanted: Vec<String>,
     keys: Sender<Map<String, Value>>,
-) -> Result<(), String> {
+) -> Result<(), ReadError> {
     let single = |value: Value| -> Map<String, Value> {
         match value {
             Value::Object(key) => key,
@@ -393,7 +394,7 @@ async fn forward_keys(
     mut pages: Receiver<HttpPage>,
     (selected, field): (&Selector, &str),
     keys: &Sender<Map<String, Value>>,
-) -> Result<(), String> {
+) -> Result<(), ReadError> {
     while let Some(page) = pages.recv().await {
         let document: Value = serde_json::from_slice(&page.bytes)
             .map_err(|error| format!("the fan-out's over response is not JSON: {error}"))?;
diff --git a/crates/sqlload/src/http_source/_helpers/oauth.rs b/crates/sqlload/src/http_source/_helpers/oauth.rs
index f8379ea1..785dd268 100644
--- a/crates/sqlload/src/http_source/_helpers/oauth.rs
+++ b/crates/sqlload/src/http_source/_helpers/oauth.rs
@@ -18,6 +18,7 @@ use crate::http_source::constants::{
     MAX_TOKEN_RENEWAL_MARGIN,
 };
 use crate::http_source::models::{ClientAuth, Grant, Issued, OAuth2Spec, TokenSource, TokenState};
+use crate::python_sources::errors::ReadError;
 
 /// `spec`'s token URL, which must be `https`, or `http` on a loopback host, because the client's secret goes there.
 pub(crate) fn token_url(spec: &OAuth2Spec) -> Result<Url, String> {
@@ -129,7 +130,7 @@ impl TokenSource {
         http: &reqwest::Client,
         retries: u32,
         refused: Option<&str>,
-    ) -> Result<String, String> {
+    ) -> Result<String, ReadError> {
         let mut state = self.state.lock().await;
         let fresh = state
             .renew_at
@@ -163,7 +164,7 @@ impl TokenSource {
         http: &reqwest::Client,
         retries: u32,
         state: &TokenState,
-    ) -> Result<Issued, String> {
+    ) -> Result<Issued, ReadError> {
         let policy = RetryPolicy {
             first_delay: FIRST_RETRY_DELAY,
             max_delay: MAX_RETRY_DELAY,
@@ -194,10 +195,10 @@ impl TokenSource {
                 ),
             };
             if attempt >= policy.max_attempts {
-                return Err(format!(
+                return Err(ReadError::transient(format!(
                     "{failure} (after {} attempts)",
                     policy.max_attempts
-                ));
+                )));
             }
             tokio::time::sleep(jittered(policy.delay(attempt))).await;
             attempt += 1;
diff --git a/crates/sqlload/src/http_source/_helpers/pages.rs b/crates/sqlload/src/http_source/_helpers/pages.rs
index 553f3244..2ef1f26f 100644
--- a/crates/sqlload/src/http_source/_helpers/pages.rs
+++ b/crates/sqlload/src/http_source/_helpers/pages.rs
@@ -13,6 +13,7 @@ use crate::http_source::_helpers::job::request_projection;
 use crate::http_source::_helpers::selector::{pointer_text, selected_node, selector};
 use crate::http_source::_helpers::urls::{redacted, request_url};
 use crate::http_source::models::{Calls, Fetched, HttpPage, Paginator, Request, Selector};
+use crate::python_sources::errors::ReadError;
 
 /// Send every page of `request` at `path` (its placeholders filled) to `pages`, with where its rows are; a page whose selected node is missing or null has no rows and is not sent. Stops quietly once `pages` closes; a page read twice, or a response repeating the previous one, is an error rather than a loop.
 pub(crate) async fn paginate(
@@ -20,7 +21,7 @@ pub(crate) async fn paginate(
     request: &Request,
     path: &str,
     pages: &Sender<HttpPage>,
-) -> Result<(), String> {
+) -> Result<(), ReadError> {
     let rows = selector(request.select.as_deref())?;
     let projection = request_projection(request)?;
     let rows_path = Some(rows.rows.join(".")).filter(|path| !path.is_empty());
@@ -33,7 +34,8 @@ pub(crate) async fn paginate(
         if !seen.insert(url.clone()) {
             return Err(format!(
                 "{shown}: the paginator came back to a page it already read; check its pointer or parameter"
-            ));
+            )
+            .into());
         }
         let Some(fetched) = fetch(calls, &url).await? else {
             return Ok(());
@@ -42,7 +44,8 @@ pub(crate) async fn paginate(
         if digest.is_some() && previous == digest {
             return Err(format!(
                 "{shown}: the response repeats the previous page; the API may ignore the paginator's parameter"
-            ));
+            )
+            .into());
         }
         previous = digest;
         let document = parsed(request, &rows, &fetched.bytes)
diff --git a/crates/sqlload/src/http_source/main/run_http_job.rs b/crates/sqlload/src/http_source/main/run_http_job.rs
index 8f09c0c1..90c38022 100644
--- a/crates/sqlload/src/http_source/main/run_http_job.rs
+++ b/crates/sqlload/src/http_source/main/run_http_job.rs
@@ -4,20 +4,21 @@ use serde_json::{Map, Value};
 
 use crate::http_source::_helpers::job::{drive_job, job_problems};
 use crate::http_source::models::{HttpJob, HttpPage};
+use crate::python_sources::errors::ReadError;
 use crate::row_batches::models::ProjectionRules;
 
-/// Run the job `definition` (JSON from `sl.http`) with the table's `connections`, handing each page to `deliver` in arrival order; `deliver` returning `false` (the consumer has gone) cancels the requests in flight. The definition is checked whole before any request, including that projected rows meet the table's `rules`.
+/// Run the job `definition` (JSON from `sl.http`) with the table's `connections`, handing each page to `deliver` in arrival order; `deliver` returning `false` (the consumer has gone) cancels the requests in flight. The definition is checked whole before any request, including that projected rows meet the table's `rules`; the error says whether running the job again may succeed.
 pub fn run_http_job(
     definition: &str,
     connections: &Map<String, Value>,
     rules: ProjectionRules<'_>,
     deliver: impl FnMut(HttpPage) -> bool,
-) -> Result<(), String> {
+) -> Result<(), ReadError> {
     let job: HttpJob = serde_json::from_str(definition)
         .map_err(|error| format!("cannot read the sl.http job: {error}"))?;
     let problems = job_problems(&job, rules);
     match problems.is_empty() {
         true => drive_job(job, connections, deliver),
-        false => Err(problems.join("; ")),
+        false => Err(problems.join("; ").into()),
     }
 }
diff --git a/crates/sqlload/src/loader/_helpers/publish.rs b/crates/sqlload/src/loader/_helpers/publish.rs
index ed1828e8..90a81933 100644
--- a/crates/sqlload/src/loader/_helpers/publish.rs
+++ b/crates/sqlload/src/loader/_helpers/publish.rs
@@ -227,7 +227,7 @@ impl Publisher {
                     .transpose()?;
                 let lease = take_lease(&live, &load_id, lease_now())
                     .await?
-                    .map_err(LoaderError::Invalid)?;
+                    .map_err(LoaderError::Transient)?;
                 let replaced = match files_to_replace(&live, &table, output).await {
                     Ok(replaced) => replaced,
                     Err(error) => {
@@ -1020,26 +1020,26 @@ async fn feed(
                 let mut last: Option<String> = None;
                 let mut cut_from = table.window.as_ref().and_then(|window| window.from.clone());
                 read_source(reader, table, part, async |batch| {
-                    let values = batch
-                        .column_by_name(column)
-                        .cloned()
-                        .ok_or_else(|| format!("the read has no cursor column {column}"))?;
+                    let values = batch.column_by_name(column).cloned().ok_or_else(|| {
+                        LoaderError::Invalid(format!("the read has no cursor column {column}"))
+                    })?;
                     let at = match publisher.full() {
-                        true => cut_index(values.as_ref(), last.as_deref())?,
+                        true => cut_index(values.as_ref(), last.as_deref())
+                            .map_err(LoaderError::Invalid)?,
                         false => batch.num_rows(),
                     };
-                    last = last_value(values.as_ref())?.or(last.take());
+                    last = last_value(values.as_ref())
+                        .map_err(LoaderError::Invalid)?
+                        .or(last.take());
                     stage(&mut publisher, batch.slice(0, at)).await?;
                     if at < batch.num_rows() {
-                        let cut_at = cut_value(values.as_ref(), at, cursor_type)?;
+                        let cut_at = cut_value(values.as_ref(), at, cursor_type)
+                            .map_err(LoaderError::Invalid)?;
                         let range = CommitRange::Window {
                             from: cut_from.replace(cut_at.clone()),
                             to: Some(cut_at),
                         };
-                        publisher
-                            .cut(range)
-                            .await
-                            .map_err(|error| error.to_string())?;
+                        publisher.cut(range).await?;
                         stage(&mut publisher, batch.slice(at, batch.num_rows() - at)).await?;
                     }
                     Ok(())
@@ -1063,10 +1063,7 @@ async fn feed(
                         read_source(reader, table, part, async |batch| {
                             stage(&mut publisher, batch).await?;
                             match publisher.full() {
-                                true => publisher
-                                    .cut(CommitRange::Planned)
-                                    .await
-                                    .map_err(|error| error.to_string()),
+                                true => publisher.cut(CommitRange::Planned).await,
                                 false => Ok(()),
                             }
                         })
@@ -1091,11 +1088,8 @@ async fn feed(
     (publisher, fed)
 }
 
-async fn stage(publisher: &mut Publisher, batch: RecordBatch) -> Result<(), String> {
-    publisher
-        .stage(batch, None)
-        .await
-        .map_err(|error| error.to_string())
+async fn stage(publisher: &mut Publisher, batch: RecordBatch) -> Result<(), LoaderError> {
+    publisher.stage(batch, None).await
 }
 
 /// The source filter reading one cursor window, in the source's dialect; its bounds are literals, since `split_windows` cuts only number and time cursors, never a text one.
diff --git a/crates/sqlload/src/loader/_helpers/python.rs b/crates/sqlload/src/loader/_helpers/python.rs
index 5a1dbf3d..86ea384b 100644
--- a/crates/sqlload/src/loader/_helpers/python.rs
+++ b/crates/sqlload/src/loader/_helpers/python.rs
@@ -668,7 +668,7 @@ impl PythonFeed {
             stream = returned;
             match next {
                 None => return Ok(()),
-                Some(Err(reason)) => return Err(LoaderError::Invalid(reason)),
+                Some(Err(failure)) => return Err(LoaderError::Read(failure)),
                 Some(Ok(Streamed::State(state))) => {
                     let state = (state != NO_STATE).then_some(state);
                     let sized = self.table.commits == CommitRule::Size;
diff --git a/crates/sqlload/src/loader/_helpers/render.rs b/crates/sqlload/src/loader/_helpers/render.rs
index d796e467..4c8893e2 100644
--- a/crates/sqlload/src/loader/_helpers/render.rs
+++ b/crates/sqlload/src/loader/_helpers/render.rs
@@ -912,6 +912,7 @@ pub(crate) fn plan_view(plan: &ProjectPlan) -> PlanView {
     );
     PlanView {
         ok: errors.is_empty(),
+        retryable: !errors.is_empty() && plan.only_retryable_problems(),
         summary: summary.join(", "),
         selected: plan.tables.len(),
         objects: stages
@@ -1025,10 +1026,14 @@ pub(crate) fn run_result(plan: &ProjectPlan, run: &RunReport<'_>) -> RunResult {
     };
     let view = plan_view(plan);
     let finished = started + chrono::Duration::from_std(elapsed).unwrap_or_default();
+    let ok = view.ok && summary.failed == 0;
     RunResult {
         schema_version: RESULT_SCHEMA_VERSION,
         command,
-        ok: view.ok && summary.failed == 0,
+        ok,
+        retryable: !ok
+            && (view.ok || view.retryable)
+            && outcomes.iter().all(TableOutcome::passes_on_rerun),
         started_at: started.to_rfc3339_opts(SecondsFormat::Millis, true),
         finished_at: finished.to_rfc3339_opts(SecondsFormat::Millis, true),
         duration_seconds: seconds(elapsed),
@@ -1058,6 +1063,7 @@ fn table_result(outcome: &TableOutcome) -> TableResult {
         duration_seconds: seconds(outcome.elapsed),
         note,
         error,
+        retryable: outcome.retryable,
     }
 }
 
diff --git a/crates/sqlload/src/loader/_helpers/sources.rs b/crates/sqlload/src/loader/_helpers/sources.rs
index 5addeabb..b45cf884 100644
--- a/crates/sqlload/src/loader/_helpers/sources.rs
+++ b/crates/sqlload/src/loader/_helpers/sources.rs
@@ -42,7 +42,7 @@ use crate::loader::constants::{
 use crate::loader::errors::LoaderError;
 use crate::loader::models::{
     CommitRule, FilesSource, PlanOptions, ProjectPlan, ResolvedTables, SkippedTable, SourceEnd,
-    SourceReader, TablePlan, TableSpec,
+    SourceProblem, SourceReader, TablePlan, TableSpec,
 };
 use crate::postgres_source::constants::DEFAULT_BATCH_BYTES;
 use crate::postgres_source::main::connect_postgres::connect_postgres;
@@ -196,16 +196,17 @@ pub(crate) async fn source_tables(
     project: &ProjectConfig,
     source_name: &str,
     source: &SourceConfig,
-) -> Result<Result<(Vec<SourceTable>, Option<FilesSource>, Vec<String>), String>, LoaderError> {
+) -> Result<Result<(Vec<SourceTable>, Option<FilesSource>, Vec<String>), SourceProblem>, LoaderError>
+{
     match source.kind {
         SourceKind::Postgres | SourceKind::PostgresCdc => {
             let address = match postgres_address(project, &source.connection) {
                 Ok(address) => address,
-                Err(problem) => return Ok(Err(problem)),
+                Err(problem) => return Ok(Err(problem.into())),
             };
             let (client, encryption) = match connect_postgres(address.clone()).await {
                 Ok(connected) => connected,
-                Err(error) => return Ok(Err(LoaderError::from(error).to_string())),
+                Err(error) => return Ok(Err(LoaderError::from(error).into())),
             };
             let warnings: Vec<String> =
                 plain_text_warning((source_name, &source.connection), &address, encryption)
@@ -214,19 +215,19 @@ pub(crate) async fn source_tables(
             Ok(
                 match read_catalog(&client, &source.include, &source.exclude).await {
                     Ok(discovered) => Ok((discovered, None, warnings)),
-                    Err(error) => Err(LoaderError::from(error).to_string()),
+                    Err(error) => Err(LoaderError::from(error).into()),
                 },
             )
         }
         SourceKind::Files => {
             let warehouse = match source_target(project, source) {
                 Ok(target) => target.warehouse,
-                Err(problem) => return Ok(Err(problem)),
+                Err(problem) => return Ok(Err(problem.into())),
             };
             Ok(
                 match discover_files_table(project, source_name, source, warehouse).await {
                     Ok((table, files)) => Ok((vec![table], Some(files), Vec::new())),
-                    Err(error) => Err(error.to_string()),
+                    Err(error) => Err(error.into()),
                 },
             )
         }
@@ -235,7 +236,7 @@ pub(crate) async fn source_tables(
                 .and_then(|session| Ok(connect_snowflake_source(session, None)?))
             {
                 Ok(reader) => reader,
-                Err(error) => return Ok(Err(error.to_string())),
+                Err(error) => return Ok(Err(error.into())),
             };
             let queries: Vec<(String, String)> = source
                 .tables
@@ -247,7 +248,7 @@ pub(crate) async fn source_tables(
                     .await
                 {
                     Ok(discovered) => Ok((discovered, None, Vec::new())),
-                    Err(error) => Err(error.to_string()),
+                    Err(error) => Err(LoaderError::from(error).into()),
                 },
             )
         }
@@ -259,12 +260,13 @@ pub(crate) async fn source_tables(
                 .collect();
             Ok(discover_odbc(project, source, &queries)
                 .await
-                .map(|discovered| (discovered, None, Vec::new())))
+                .map(|discovered| (discovered, None, Vec::new()))
+                .map_err(SourceProblem::from))
         }
         SourceKind::BigQuery => {
             let reader = match bigquery_source_reader(project, source) {
                 Ok(reader) => reader,
-                Err(error) => return Ok(Err(error.to_string())),
+                Err(error) => return Ok(Err(error.into())),
             };
             let queries: Vec<(String, String)> = source
                 .tables
@@ -276,14 +278,14 @@ pub(crate) async fn source_tables(
                     .await
                 {
                     Ok(discovered) => Ok((discovered, None, Vec::new())),
-                    Err(error) => Err(error.to_string()),
+                    Err(error) => Err(LoaderError::from(error).into()),
                 },
             )
         }
         SourceKind::Python => Ok(Ok((python_tables(source), None, Vec::new()))),
         SourceKind::GoogleSheets => Ok(match sheets_tables(project, source).await {
             Ok((tables, warnings)) => Ok((tables, None, warnings)),
-            Err(error) => Err(error.to_string()),
+            Err(error) => Err(error.into()),
         }),
     }
 }
@@ -1148,8 +1150,32 @@ pub(crate) struct ReadPart {
     pub(crate) batch_bytes: Option<usize>,
 }
 
-/// Read `part` of `table`'s data from `reader`, passing each batch to `on_batch`; returns the row count.
+/// Read `part` of `table`'s rows from `reader`, handing each batch to `on_batch`; when `on_batch` fails, the read fails with the reader's text, and stays retryable when `on_batch`'s error was.
 pub(crate) async fn read_source(
+    reader: &SourceReader,
+    table: &TablePlan,
+    part: ReadPart,
+    mut on_batch: impl AsyncFnMut(RecordBatch) -> Result<(), LoaderError>,
+) -> Result<u64, LoaderError> {
+    let mut failed: Option<LoaderError> = None;
+    let read = read_each(reader, table, part, async |batch| {
+        on_batch(batch).await.map_err(|error| {
+            let text = error.to_string();
+            failed = Some(error);
+            text
+        })
+    })
+    .await;
+    match (read, failed) {
+        (Err(read), Some(failed)) if failed.is_retryable() => {
+            Err(LoaderError::Transient(read.to_string()))
+        }
+        (read, _) => read,
+    }
+}
+
+/// Read `part` of `table`'s data from `reader`, passing each batch to `on_batch`; returns the row count.
+async fn read_each(
     reader: &SourceReader,
     table: &TablePlan,
     part: ReadPart,
@@ -1359,19 +1385,19 @@ pub(crate) fn selected(
         && !options.exclude.iter().any(picks)
 }
 
-/// Source `source_name`'s tables resolved to their settings, with its files for a files source, or `Err` with its problem when it cannot list them; its problems count only when the selection in `options` takes something from it (see [`source_in_scope`]), and are dropped otherwise.
+/// Source `source_name`'s tables resolved to their settings, with its files for a files source, or `Err` with its problem when it cannot list them (`None` when out of scope); its problems count only when the selection in `options` takes something from it (see [`source_in_scope`]), and are dropped otherwise.
 pub(crate) async fn scoped_tables(
     (project, options): (&ProjectConfig, &PlanOptions),
     (source_name, source): (&str, &SourceConfig),
-) -> Result<Result<(ResolvedTables, Option<FilesSource>), Vec<String>>, LoaderError> {
+) -> Result<Result<(ResolvedTables, Option<FilesSource>), Option<SourceProblem>>, LoaderError> {
     let (discovered, files, warnings) = match source_tables(project, source_name, source).await? {
         Ok(found) => found,
         Err(problem) => {
             let in_scope = source_in_scope((project, options), (source_name, source), None);
-            return Ok(Err(in_scope
-                .then(|| format!("[sources.{source_name}]: {problem}"))
-                .into_iter()
-                .collect()));
+            return Ok(Err(in_scope.then(|| SourceProblem {
+                message: format!("[sources.{source_name}]: {}", problem.message),
+                retryable: problem.retryable,
+            })));
         }
     };
     let mut resolved = resolve_tables(project, source_name, &discovered);
diff --git a/crates/sqlload/src/loader/_helpers/table_queue.rs b/crates/sqlload/src/loader/_helpers/table_queue.rs
index a1f5d840..3f93db30 100644
--- a/crates/sqlload/src/loader/_helpers/table_queue.rs
+++ b/crates/sqlload/src/loader/_helpers/table_queue.rs
@@ -41,11 +41,13 @@ pub(crate) async fn load_task(shared: Arc<SharedRun>, table: TablePlan) -> Table
             table: name,
             origin: Some(table.origin()),
             loaded: Ok(loaded),
+            retryable: false,
             elapsed: started.elapsed(),
         },
         Err(error) => TableOutcome {
             table: table.outcome_name(),
             origin: Some(table.origin()),
+            retryable: error.is_retryable(),
             loaded: Err(error.to_string()),
             elapsed: started.elapsed(),
         },
@@ -118,6 +120,7 @@ impl<'a> TableQueue<'a> {
                 table: table.outcome_name(),
                 origin: Some(table.origin()),
                 loaded: Err(reason),
+                retryable: false,
                 elapsed: std::time::Duration::ZERO,
             }),
         ));
diff --git a/crates/sqlload/src/loader/constants.rs b/crates/sqlload/src/loader/constants.rs
index 1d9eefe7..2cd8562e 100644
--- a/crates/sqlload/src/loader/constants.rs
+++ b/crates/sqlload/src/loader/constants.rs
@@ -98,3 +98,25 @@ pub const RESULT_SCHEMA_VERSION: u32 = 1;
 /// The loss of an integer array pin over Arrow or Parquet lists.
 pub const CUT_ELEMENTS_NOTE: &str =
     "float elements of Arrow and Parquet lists are cut to whole numbers";
+
+/// The SQLSTATEs of Postgres errors worth retrying unchanged: a connection that failed or dropped (not 08P01, which PgBouncer also sends for a database it does not know), out of memory, too many connections, a server shutting down or starting, a serialization failure, a deadlock and a lock not available.
+pub const TRANSIENT_SQLSTATES: [&str; 11] = [
+    "08000", "08003", "08006", "53200", "53300", "57P01", "57P02", "57P03", "40001", "40P01",
+    "55P03",
+];
+
+/// HTTP statuses a rerun may get past whatever the service's code says: throttling, a bad gateway, unavailable, a gateway timeout.
+pub const TRANSIENT_STATUSES: [u16; 4] = [429, 502, 503, 504];
+
+/// Service codes that make an HTTP 500 worth retrying: ClickHouse's timeouts, socket and network errors, too many queries or parts, and memory limits; BigQuery's backend and internal errors. A 500 with any other code (ClickHouse answers a bad query with one) is not.
+pub const TRANSIENT_SERVER_CODES: [&str; 9] = [
+    "159",
+    "202",
+    "209",
+    "210",
+    "241",
+    "252",
+    "backendError",
+    "internalError",
+    "rateLimitExceeded",
+];
diff --git a/crates/sqlload/src/loader/errors.rs b/crates/sqlload/src/loader/errors.rs
index 5f26e096..b0b9f53b 100644
--- a/crates/sqlload/src/loader/errors.rs
+++ b/crates/sqlload/src/loader/errors.rs
@@ -1,10 +1,18 @@
+use std::error::Error;
+use std::io::ErrorKind;
+
+use object_store::client::HttpErrorKind;
+
 #[cfg(feature = "duckdb")]
 use crate::adapters::duckdb::errors::DuckDbError;
 use crate::adapters::files::errors::FilesDestinationError;
 use crate::file_store::errors::StoreError;
 use crate::files_source::errors::FilesError;
+use crate::http::constants::FIRST_SERVER_ERROR;
 use crate::http::errors::ApiError;
+use crate::loader::constants::{TRANSIENT_SERVER_CODES, TRANSIENT_SQLSTATES, TRANSIENT_STATUSES};
 use crate::postgres_source::errors::PostgresError;
+use crate::python_sources::errors::ReadError;
 use crate::snowflake_source::errors::SnowflakeSourceError;
 use crate::table_lease::errors::LeaseError;
 
@@ -35,7 +43,109 @@ pub enum LoaderError {
     Lease(#[from] LeaseError),
     #[error("{0}")]
     Invalid(String),
+    /// A Python table's read failed.
+    #[error(transparent)]
+    Read(#[from] ReadError),
+    /// A failure a rerun may pass: another run holds the table's lease, or the table's session dropped mid-load.
+    #[error("{0}")]
+    Transient(String),
     /// A table gained a column since the load was planned: nothing of the batch was written, and a new plan streams it.
     #[error("{0}")]
     Replan(String),
 }
+
+impl LoaderError {
+    /// True when running the load again unchanged may succeed: a dropped, refused or timed-out connection, throttling or a server error that outlasted its retries, a table another run is loading, or a column that appeared mid-load. Problems with the configuration, the data, credentials or permissions are not.
+    pub fn is_retryable(&self) -> bool {
+        match self {
+            Self::Postgres(
+                PostgresError::Connect { source, .. } | PostgresError::Query(source),
+            ) => Self::postgres_transient(source),
+            Self::Snowflake(error)
+            | Self::WarehouseSource(SnowflakeSourceError::Api(error))
+            | Self::BigQuerySource(crate::bigquery_source::errors::BigQuerySourceError::Api(
+                error,
+            )) => Self::api_transient(error),
+            Self::Store(store)
+            | Self::Files(FilesError::Store(store))
+            | Self::WarehouseSource(SnowflakeSourceError::Store(store))
+            | Self::FilesDestination(FilesDestinationError::Store(store))
+            | Self::Lease(LeaseError::Store(store)) => match store {
+                StoreError::Open { source, .. } | StoreError::Object { source, .. } => {
+                    Self::store_transient(source)
+                }
+            },
+            Self::Read(error) => error.retryable,
+            Self::Transient(_) | Self::Replan(_) => true,
+            other => Self::transient_cause(other),
+        }
+    }
+
+    /// Whether `error`, or anything it was caused by, is a dropped, refused or timed-out connection.
+    fn transient_cause(error: &(dyn Error + 'static)) -> bool {
+        let mut cause = Some(error);
+        while let Some(current) = cause {
+            if let Some(io) = current.downcast_ref::<std::io::Error>()
+                && matches!(
+                    io.kind(),
+                    ErrorKind::ConnectionRefused
+                        | ErrorKind::ConnectionReset
+                        | ErrorKind::ConnectionAborted
+                        | ErrorKind::NotConnected
+                        | ErrorKind::BrokenPipe
+                        | ErrorKind::TimedOut
+                        | ErrorKind::UnexpectedEof
+                        | ErrorKind::HostUnreachable
+                        | ErrorKind::NetworkUnreachable
+                        | ErrorKind::NetworkDown
+                )
+            {
+                return true;
+            }
+            if let Some(http) = current.downcast_ref::<reqwest::Error>()
+                && (http.is_timeout() || http.is_connect())
+            {
+                return true;
+            }
+            if let Some(http) = current.downcast_ref::<object_store::client::HttpError>()
+                && matches!(
+                    http.kind(),
+                    HttpErrorKind::Connect | HttpErrorKind::Timeout | HttpErrorKind::Interrupted
+                )
+            {
+                return true;
+            }
+            cause = current.source();
+        }
+        false
+    }
+
+    /// Whether a Postgres error may pass on a retry: the connection dropped or timed out, or the server answered with one of `TRANSIENT_SQLSTATES`; a server that refused the login or the SQL is not.
+    fn postgres_transient(error: &tokio_postgres::Error) -> bool {
+        match error.as_db_error() {
+            Some(answer) => TRANSIENT_SQLSTATES.contains(&answer.code().code()),
+            None => error.is_closed() || Self::transient_cause(error),
+        }
+    }
+
+    /// Whether an object store failure may pass on a retry: its own retries ran out on a dropped or timed-out connection.
+    fn store_transient(error: &object_store::Error) -> bool {
+        match error {
+            object_store::Error::Generic { source, .. } => Self::transient_cause(source.as_ref()),
+            _ => false,
+        }
+    }
+
+    /// Whether a warehouse or object store call may pass on a rerun: the connection failed, the service throttled or was unavailable, or it answered 500 with no code or a transient one.
+    fn api_transient(error: &ApiError) -> bool {
+        match error {
+            ApiError::Transport(_) => true,
+            ApiError::Status { status, code, .. } => {
+                TRANSIENT_STATUSES.contains(status)
+                    || (*status >= FIRST_SERVER_ERROR
+                        && (code.is_empty() || TRANSIENT_SERVER_CODES.contains(&code.as_str())))
+            }
+            ApiError::Protocol(_) | ApiError::Auth(_) => false,
+        }
+    }
+}
diff --git a/crates/sqlload/src/loader/main/json_report.rs b/crates/sqlload/src/loader/main/json_report.rs
index d8dcd224..91dbd068 100644
--- a/crates/sqlload/src/loader/main/json_report.rs
+++ b/crates/sqlload/src/loader/main/json_report.rs
@@ -5,16 +5,16 @@ use serde::Serialize;
 use crate::loader::_helpers::render::{plan_view, run_result};
 use crate::loader::models::{ProjectPlan, RunReport};
 
-/// The plan as one indented JSON document, or, with `run`, what `apply` or `load` did after it; and whether every selected table can load (and, after a run, loaded).
-pub fn json_report(plan: &ProjectPlan, run: Option<&RunReport<'_>>) -> (String, bool) {
+/// The plan as one indented JSON document, or, with `run`, what `apply` or `load` did after it; whether every selected table can load (and, after a run, loaded); and whether the run failed only in ways a rerun may fix.
+pub fn json_report(plan: &ProjectPlan, run: Option<&RunReport<'_>>) -> (String, bool, bool) {
     match run {
         None => {
             let view = plan_view(plan);
-            (pretty(&view), view.ok)
+            (pretty(&view), view.ok, view.retryable)
         }
         Some(run) => {
             let result = run_result(plan, run);
-            (pretty(&result), result.ok)
+            (pretty(&result), result.ok, result.retryable)
         }
     }
 }
diff --git a/crates/sqlload/src/loader/main/plan_project.rs b/crates/sqlload/src/loader/main/plan_project.rs
index 218a2453..0def8b7a 100644
--- a/crates/sqlload/src/loader/main/plan_project.rs
+++ b/crates/sqlload/src/loader/main/plan_project.rs
@@ -30,8 +30,13 @@ pub async fn plan_project(
         let (resolved, files) =
             match scoped_tables((project, options), (source_name, source)).await? {
                 Ok(found) => found,
-                Err(problems) => {
-                    plan.problems.extend(problems);
+                Err(problem) => {
+                    if let Some(problem) = problem {
+                        if problem.retryable {
+                            plan.retryable_problems.push(problem.message.clone());
+                        }
+                        plan.problems.push(problem.message);
+                    }
                     continue;
                 }
             };
diff --git a/crates/sqlload/src/loader/main/selected_specs.rs b/crates/sqlload/src/loader/main/selected_specs.rs
index 71cd7ec0..a79f8f26 100644
--- a/crates/sqlload/src/loader/main/selected_specs.rs
+++ b/crates/sqlload/src/loader/main/selected_specs.rs
@@ -19,7 +19,9 @@ pub async fn selected_specs(
             match scoped_tables((project, options), (source_name, source)).await? {
                 Ok(found) => found,
                 Err(problems) => {
-                    chosen.problems.extend(problems);
+                    chosen
+                        .problems
+                        .extend(problems.into_iter().map(|problem| problem.message));
                     continue;
                 }
             };
diff --git a/crates/sqlload/src/loader/models.rs b/crates/sqlload/src/loader/models.rs
index 98bb7c17..7f7c6fe0 100644
--- a/crates/sqlload/src/loader/models.rs
+++ b/crates/sqlload/src/loader/models.rs
@@ -356,9 +356,22 @@ pub struct ProjectPlan {
     pub problems: Vec<String>,
     /// Things to know that block nothing, such as a selector matching no table.
     pub warnings: Vec<String>,
+    /// The `problems` a rerun may get past: a source that could not be reached.
+    pub retryable_problems: Vec<String>,
 }
 
 impl ProjectPlan {
+    /// Whether everything that stops the plan is something a rerun may get past: no table is blocked, and every problem is a source that could not be reached.
+    pub fn only_retryable_problems(&self) -> bool {
+        self.tables
+            .iter()
+            .all(|table| table.schema.blocked.is_none())
+            && self
+                .problems
+                .iter()
+                .all(|problem| self.retryable_problems.contains(problem))
+    }
+
     /// The plan without its `postgres_cdc` tables, which load together per source rather than one by one.
     pub fn uncaptured(&self) -> Self {
         Self {
@@ -452,9 +465,21 @@ pub struct TableOutcome {
     /// The table's source, target and write strategy; `None` for a failure of no one table (a `load --follow` with none).
     pub origin: Option<OutcomeOrigin>,
     pub loaded: Result<TableLoaded, String>,
+    /// Whether a failed load may succeed when run again unchanged; `false` once loaded.
+    pub retryable: bool,
     pub elapsed: std::time::Duration,
 }
 
+impl TableOutcome {
+    /// Whether this table needs nothing a rerun cannot give it: it loaded, or its failure is retryable.
+    pub fn passes_on_rerun(&self) -> bool {
+        match &self.loaded {
+            Ok(_) => true,
+            Err(_) => self.retryable,
+        }
+    }
+}
+
 /// Where a loaded table comes from and goes, and how it is written.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct OutcomeOrigin {
@@ -722,6 +747,8 @@ pub struct PlanDetail {
 pub struct PlanView {
     /// Whether every selected table can load.
     pub ok: bool,
+    /// Whether the plan failed only in ways a rerun may get past, such as a source that could not be reached.
+    pub retryable: bool,
     /// The counts after `Plan ready`, nonzero clauses only.
     pub summary: String,
     pub selected: usize,
@@ -781,6 +808,8 @@ pub struct RunResult {
     pub command: &'static str,
     /// Whether every selected table planned, and every table loaded.
     pub ok: bool,
+    /// Whether the run failed only in ways that running it again unchanged may fix: every selected table planned, and each failed table's failure is retryable.
+    pub retryable: bool,
     pub started_at: String,
     pub finished_at: String,
     pub duration_seconds: f64,
@@ -823,4 +852,31 @@ pub struct TableResult {
     pub duration_seconds: f64,
     pub note: Option<String>,
     pub error: Option<String>,
+    /// Whether the failure may pass when run again unchanged; `false` once loaded.
+    pub retryable: bool,
+}
+
+/// Why a source's tables could not be listed, and whether a rerun may get past it.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct SourceProblem {
+    pub message: String,
+    pub retryable: bool,
+}
+
+impl From<String> for SourceProblem {
+    fn from(message: String) -> Self {
+        Self {
+            message,
+            retryable: false,
+        }
+    }
+}
+
+impl From<crate::loader::errors::LoaderError> for SourceProblem {
+    fn from(error: crate::loader::errors::LoaderError) -> Self {
+        Self {
+            retryable: error.is_retryable(),
+            message: error.to_string(),
+        }
+    }
 }
diff --git a/crates/sqlload/src/loader/tests.rs b/crates/sqlload/src/loader/tests.rs
index f75a04d3..e7a5c614 100644
--- a/crates/sqlload/src/loader/tests.rs
+++ b/crates/sqlload/src/loader/tests.rs
@@ -4,6 +4,8 @@ mod helpers;
 mod python_columns;
 #[path = "tests/test_render_tally.rs"]
 mod render_tally;
+#[path = "tests/test_retryable_errors.rs"]
+mod retryable_errors;
 #[path = "tests/test_source_filter.rs"]
 mod source_filter;
 #[path = "tests/test_types.rs"]
diff --git a/crates/sqlload/src/loader/tests/helpers.rs b/crates/sqlload/src/loader/tests/helpers.rs
index 646b3479..b4aadb46 100644
--- a/crates/sqlload/src/loader/tests/helpers.rs
+++ b/crates/sqlload/src/loader/tests/helpers.rs
@@ -6,11 +6,14 @@ use arrow_schema::{DataType, Field, Schema, TimeUnit};
 
 use crate::column_types::constants::ARROW_TYPE_METADATA_KEY;
 use crate::column_types::main::arrow_field::arrow_field;
+use crate::file_store::errors::StoreError;
+use crate::http::errors::ApiError;
 use crate::loader::_helpers::python::bound_array;
 use crate::loader::_helpers::python::{
     YieldedData, declared_problems, inferred_table, python_tables, source_spelling,
 };
 use crate::loader::_helpers::resolve::resolve_tables;
+use crate::loader::errors::LoaderError;
 use crate::loader::models::{PlanDetail, PlanItem, TableSpec};
 use crate::loader::tests::test_types::{
     BoundArrayTestCase, DeclaredProblemsTestCase, InferredTableTestCase, ItemSpec,
@@ -194,3 +197,53 @@ pub(crate) fn plan_items(specs: &[ItemSpec]) -> Vec<PlanItem> {
         })
         .collect()
 }
+
+/// A warehouse call that failed with `status` and service `code`.
+pub(crate) fn status(status: u16, code: &str) -> LoaderError {
+    LoaderError::Snowflake(ApiError::Status {
+        status,
+        code: code.to_owned(),
+        message: "m".to_owned(),
+    })
+}
+
+/// An object store failure.
+pub(crate) fn store(source: object_store::Error) -> LoaderError {
+    LoaderError::Store(StoreError::Object {
+        action: "get",
+        key: "k".to_owned(),
+        source,
+    })
+}
+
+/// An object store failure caused by a socket error of `kind`.
+pub(crate) fn io_store(kind: std::io::ErrorKind) -> LoaderError {
+    store(object_store::Error::Generic {
+        store: "S3",
+        source: Box::new(std::io::Error::new(kind, "socket")),
+    })
+}
+
+/// An object store failure whose client reported an HTTP error of `kind`.
+pub(crate) fn http_store(kind: object_store::client::HttpErrorKind) -> LoaderError {
+    store(object_store::Error::Generic {
+        store: "S3",
+        source: Box::new(object_store::client::HttpError::new(
+            kind,
+            std::io::Error::other("http"),
+        )),
+    })
+}
+
+/// A warehouse call whose request could not be built: a header value with a line break, such as a password read with its trailing newline.
+pub(crate) fn unbuildable_request() -> LoaderError {
+    crate::http::main::install_tls_provider::install_tls_provider();
+    let built = reqwest::Client::new()
+        .get("http://127.0.0.1:1/")
+        .header("X-Key", "secret\n")
+        .build();
+    built.map_or_else(
+        |error| LoaderError::Snowflake(ApiError::from(error)),
+        |_| LoaderError::Invalid("the request was built".to_owned()),
+    )
+}
diff --git a/crates/sqlload/src/loader/tests/test_render_tally.rs b/crates/sqlload/src/loader/tests/test_render_tally.rs
index 6f788385..d4fc01c7 100644
--- a/crates/sqlload/src/loader/tests/test_render_tally.rs
+++ b/crates/sqlload/src/loader/tests/test_render_tally.rs
@@ -57,6 +57,7 @@ fn given_a_load_when_rendering_its_tally_then_lines_align_and_the_last_counts_th
                         ..TableLoaded::default()
                     })
                     .map_err(ToString::to_string),
+                retryable: false,
                 elapsed: Duration::from_millis(*elapsed_ms),
             })
             .collect();
diff --git a/crates/sqlload/src/loader/tests/test_retryable_errors.rs b/crates/sqlload/src/loader/tests/test_retryable_errors.rs
new file mode 100644
index 00000000..b2ca0af4
--- /dev/null
+++ b/crates/sqlload/src/loader/tests/test_retryable_errors.rs
@@ -0,0 +1,138 @@
+use crate::http::errors::ApiError;
+use crate::loader::errors::LoaderError;
+use crate::loader::tests::helpers::{http_store, io_store, status, store, unbuildable_request};
+use crate::loader::tests::test_types::RetryableErrorTestCase;
+use crate::python_sources::errors::ReadError;
+
+#[test]
+fn given_a_failed_load_when_classifying_it_then_only_failures_a_rerun_may_pass_are_retryable() {
+    let test_cases = [
+        RetryableErrorTestCase {
+            description: "a service that is unavailable",
+            error: || status(503, ""),
+            expected_retryable: true,
+        },
+        RetryableErrorTestCase {
+            description: "throttling",
+            error: || status(429, "rateLimitExceeded"),
+            expected_retryable: true,
+        },
+        RetryableErrorTestCase {
+            description: "a 500 with no code",
+            error: || status(500, ""),
+            expected_retryable: true,
+        },
+        RetryableErrorTestCase {
+            description: "a ClickHouse timeout answered as 500",
+            error: || status(500, "159"),
+            expected_retryable: true,
+        },
+        RetryableErrorTestCase {
+            description: "a ClickHouse syntax error answered as 500 is not",
+            error: || status(500, "62"),
+            expected_retryable: false,
+        },
+        RetryableErrorTestCase {
+            description: "a failed statement reported as 400 is not",
+            error: || status(400, "invalidQuery"),
+            expected_retryable: false,
+        },
+        RetryableErrorTestCase {
+            description: "a refused credential is not",
+            error: || status(401, ""),
+            expected_retryable: false,
+        },
+        RetryableErrorTestCase {
+            description: "a missing permission is not",
+            error: || status(403, "accessDenied"),
+            expected_retryable: false,
+        },
+        RetryableErrorTestCase {
+            description: "a request that never reached the service",
+            error: || LoaderError::Snowflake(ApiError::Transport("connection reset".to_owned())),
+            expected_retryable: true,
+        },
+        RetryableErrorTestCase {
+            description: "an answer that could not be read is not",
+            error: || LoaderError::Snowflake(ApiError::Protocol("not JSON".to_owned())),
+            expected_retryable: false,
+        },
+        RetryableErrorTestCase {
+            description: "an object store connection reset after its own retries",
+            error: || io_store(std::io::ErrorKind::ConnectionReset),
+            expected_retryable: true,
+        },
+        RetryableErrorTestCase {
+            description: "an object store timeout",
+            error: || io_store(std::io::ErrorKind::TimedOut),
+            expected_retryable: true,
+        },
+        RetryableErrorTestCase {
+            description: "a missing object is not",
+            error: || {
+                store(object_store::Error::NotFound {
+                    path: "k".to_owned(),
+                    source: "gone".into(),
+                })
+            },
+            expected_retryable: false,
+        },
+        RetryableErrorTestCase {
+            description: "an object store refusing the request for another reason is not",
+            error: || io_store(std::io::ErrorKind::PermissionDenied),
+            expected_retryable: false,
+        },
+        RetryableErrorTestCase {
+            description: "an object store client that could not connect",
+            error: || http_store(object_store::client::HttpErrorKind::Connect),
+            expected_retryable: true,
+        },
+        RetryableErrorTestCase {
+            description: "an object store answer that could not be decoded is not",
+            error: || http_store(object_store::client::HttpErrorKind::Decode),
+            expected_retryable: false,
+        },
+        RetryableErrorTestCase {
+            description: "a request that cannot be built, such as a credential with a line break, is not",
+            error: unbuildable_request,
+            expected_retryable: false,
+        },
+        RetryableErrorTestCase {
+            description: "a table another run is loading",
+            error: || LoaderError::Transient("the table is being loaded by load x".to_owned()),
+            expected_retryable: true,
+        },
+        RetryableErrorTestCase {
+            description: "a column that appeared mid-load",
+            error: || LoaderError::Replan("new column".to_owned()),
+            expected_retryable: true,
+        },
+        RetryableErrorTestCase {
+            description: "an sl.http job that outlasted its retries",
+            error: || {
+                LoaderError::Read(ReadError::transient(
+                    "HTTP 503 (after 4 attempts)".to_owned(),
+                ))
+            },
+            expected_retryable: true,
+        },
+        RetryableErrorTestCase {
+            description: "a Python table that raised for any other reason is not",
+            error: || LoaderError::Read(ReadError::from("ValueError: bad row".to_owned())),
+            expected_retryable: false,
+        },
+        RetryableErrorTestCase {
+            description: "data or configuration problems are not",
+            error: || LoaderError::Invalid("column id: cannot cast".to_owned()),
+            expected_retryable: false,
+        },
+    ];
+    for test_case in test_cases {
+        assert_eq!(
+            (test_case.error)().is_retryable(),
+            test_case.expected_retryable,
+            "{}",
+            test_case.description
+        );
+    }
+}
diff --git a/crates/sqlload/src/loader/tests/test_types.rs b/crates/sqlload/src/loader/tests/test_types.rs
index 7e6a9454..46ae2c73 100644
--- a/crates/sqlload/src/loader/tests/test_types.rs
+++ b/crates/sqlload/src/loader/tests/test_types.rs
@@ -99,3 +99,9 @@ pub(crate) struct SourceFilterTestCase {
     /// The values bound to the filter's placeholders, in order.
     pub(crate) expected_parameters: &'static [&'static str],
 }
+
+pub(crate) struct RetryableErrorTestCase {
+    pub(crate) description: &'static str,
+    pub(crate) error: fn() -> crate::loader::errors::LoaderError,
+    pub(crate) expected_retryable: bool,
+}
diff --git a/crates/sqlload/src/loader/tests/test_view_lines.rs b/crates/sqlload/src/loader/tests/test_view_lines.rs
index 2ca26d6b..5e3893f1 100644
--- a/crates/sqlload/src/loader/tests/test_view_lines.rs
+++ b/crates/sqlload/src/loader/tests/test_view_lines.rs
@@ -103,6 +103,7 @@ fn given_a_plan_view_when_printing_it_then_it_follows_the_sqb_plan_layout() {
     for test_case in test_cases {
         let view = PlanView {
             ok: test_case.ok,
+            retryable: false,
             summary: test_case.summary.to_owned(),
             selected: test_case.selected,
             objects: plan_items(test_case.objects),
diff --git a/crates/sqlload/src/python_sources/errors.rs b/crates/sqlload/src/python_sources/errors.rs
new file mode 100644
index 00000000..0ddd1db1
--- /dev/null
+++ b/crates/sqlload/src/python_sources/errors.rs
@@ -0,0 +1,26 @@
+/// Why a Python table's read failed, and whether running it again unchanged may succeed (a connection or timeout error, or an `sl.http` request still throttled or failing on the server after its retries).
+#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
+#[error("{message}")]
+pub struct ReadError {
+    pub message: String,
+    pub retryable: bool,
+}
+
+impl ReadError {
+    /// A failure worth retrying unchanged.
+    pub fn transient(message: String) -> Self {
+        Self {
+            message,
+            retryable: true,
+        }
+    }
+}
+
+impl From<String> for ReadError {
+    fn from(message: String) -> Self {
+        Self {
+            message,
+            retryable: false,
+        }
+    }
+}
diff --git a/crates/sqlload/src/python_sources/mod.rs b/crates/sqlload/src/python_sources/mod.rs
index 696abac8..921db339 100644
--- a/crates/sqlload/src/python_sources/mod.rs
+++ b/crates/sqlload/src/python_sources/mod.rs
@@ -1,6 +1,7 @@
 //! Python sources: declarations found by importing `python_paths`, and the host that runs their table functions.
 
 pub(crate) mod _helpers;
+pub mod errors;
 pub mod main;
 pub mod models;
 pub mod types;
diff --git a/crates/sqlload/src/python_sources/types.rs b/crates/sqlload/src/python_sources/types.rs
index 3a68ed88..aaae1b86 100644
--- a/crates/sqlload/src/python_sources/types.rs
+++ b/crates/sqlload/src/python_sources/types.rs
@@ -2,6 +2,7 @@ use std::path::{Path, PathBuf};
 
 use arrow_array::RecordBatch;
 
+use crate::python_sources::errors::ReadError;
 use crate::python_sources::models::{PythonSource, TableRead};
 
 /// What a table function's read gives next.
@@ -15,7 +16,7 @@ pub enum Streamed {
 /// A table function's data, one Arrow batch at a time; `next_batch` blocks.
 pub trait BatchStream: Send {
     /// The next batch or state, an error, or `None` at the end.
-    fn next_batch(&mut self) -> Option<Result<Streamed, String>>;
+    fn next_batch(&mut self) -> Option<Result<Streamed, ReadError>>;
 }
 
 /// Runs Python sources: the `sqlload` Python package provides it; the standalone executable has none.
diff --git a/packages/sqlload/tests/test_cli.py b/packages/sqlload/tests/test_cli.py
index 3b07c296..9cb8b7a4 100644
--- a/packages/sqlload/tests/test_cli.py
+++ b/packages/sqlload/tests/test_cli.py
@@ -101,7 +101,7 @@ def test_given_no_project_when_copying_a_folder_then_its_rows_load_and_print_tom
 
 def test_given_a_literal_password_in_a_uri_when_the_copy_fails_then_no_output_shows_it(tmp_path: Path) -> None:
 	done = run("copy", "postgresql://app:hunter2-secret@127.0.0.1:1/shop#public.accounts", f"duckdb://{tmp_path / 'wh.duckdb'}#RAW.ACCOUNTS", "-c", str(tmp_path / "sqlload.toml"))
-	assert done.returncode == 1
+	assert done.returncode == 75, "a refused connection is worth retrying"
 	assert "app:***@127.0.0.1" in done.stderr, done.stderr
 	assert "cannot connect to Postgres at 127.0.0.1:1/shop" in done.stdout, done.stdout
 	assert "hunter2" not in done.stdout + done.stderr
diff --git a/packages/sqlload/tests/test_http_jobs.py b/packages/sqlload/tests/test_http_jobs.py
index d219291f..1ac92bd0 100644
--- a/packages/sqlload/tests/test_http_jobs.py
+++ b/packages/sqlload/tests/test_http_jobs.py
@@ -375,11 +375,26 @@ def test_given_a_failing_job_when_loading_then_the_error_names_the_request_and_h
 ) -> None:
 	project = make_project(tmp_path, api, tables)
 	done = sqlload(project, "load")
-	assert done.returncode == 1
+	assert done.returncode == (75 if "(after 4 attempts)" in expected_text else 1)
 	assert expected_text in done.stdout + done.stderr, done.stdout + done.stderr
 	assert TOKEN[:4] not in done.stdout + done.stderr
 
 
+@pytest.mark.parametrize(
+	("path", "expected_exit", "expected_retryable"),
+	[("/broken", 75, True), ("/missing", 1, False)],
+	ids=["server-errors-outlast-the-retries", "a-status-that-fails-for-good"],
+)
+def test_given_a_failing_job_when_loading_with_json_then_the_exit_code_and_report_say_whether_to_retry(
+	tmp_path: Path, api: ThreadingHTTPServer, path: str, expected_exit: int, expected_retryable: bool
+) -> None:
+	project = make_project(tmp_path, api, f"@shop.table(description='T.')\ndef t(ctx):\n    return api.paginate('{path}', paginator=sl.single())\n")
+	done = sqlload(project, "load", "--json")
+	assert done.returncode == expected_exit, done.stdout + done.stderr
+	result = json.loads(done.stdout)
+	assert (result["ok"], result["retryable"], result["tables"][0]["retryable"]) == (False, expected_retryable, expected_retryable)
+
+
 def test_given_a_bad_definition_when_loading_then_no_request_is_sent(tmp_path: Path, api: ThreadingHTTPServer) -> None:
 	project = make_project(tmp_path, api, "@shop.table(description='T.')\ndef t(ctx):\n    return api.paginate('/next', paginator=sl.single(), select='$.a[*]')\n")
 	assert sqlload(project, "load").returncode == 1
diff --git a/packages/sqlload/tests/test_python_sources.py b/packages/sqlload/tests/test_python_sources.py
index cd5ac51a..67a34982 100644
--- a/packages/sqlload/tests/test_python_sources.py
+++ b/packages/sqlload/tests/test_python_sources.py
@@ -216,7 +216,7 @@ def test_given_one_table_failing_when_loading_with_json_then_stdout_is_one_docum
 	done = sqlload(project, "load", "--json")
 	assert done.returncode == 1, done.stdout + done.stderr
 	result = json.loads(done.stdout)
-	assert (result["schema_version"], result["command"], result["ok"]) == (1, "load", False)
+	assert (result["schema_version"], result["command"], result["ok"], result["retryable"]) == (1, "load", False, False)
 	assert result["summary"] == {"created": 1, "altered": 0, "recreated": 0, "loaded": 1, "failed": 1, "rows": 2}
 	assert [(o["action"], o["name"]) for o in result["objects"]] == [("created", "RAW.shop__good")]
 	outcomes = {table["table"]: table for table in result["tables"]}
@@ -225,6 +225,7 @@ def test_given_one_table_failing_when_loading_with_json_then_stdout_is_one_docum
 		"RAW.shop__bad": ("failed", None, "shop", "dev", "replace"),
 	}
 	assert "RuntimeError: the API returned 503" in outcomes["RAW.shop__bad"]["error"]
+	assert {name: t["retryable"] for name, t in outcomes.items()} == {"RAW.shop__good": False, "RAW.shop__bad": False}
 	assert result["plan"]["selected"] == 2
 	reload = sqlload(project, "load", "--json", "--reload")
 	assert reload.returncode == 1, reload.stdout + reload.stderr
@@ -235,6 +236,81 @@ def test_given_one_table_failing_when_loading_with_json_then_stdout_is_one_docum
 	assert (json.loads(applied.stdout)["command"], json.loads(applied.stdout)["tables"]) == ("apply", [])
 
 
+#: Tables that raise what data/fail.json names: `pages` (logged, a checkpoint per page) at a page, `events` (a cursor) at a value, `broken` at all.
+RETRY_MODULE = """
+import json
+from pathlib import Path
+
+import sqlload as sl
+
+HERE = Path(__file__).resolve().parent.parent
+shop = sl.source("shop")
+
+
+def failure():
+    path = HERE / "data" / "fail.json"
+    return json.loads(path.read_text()) if path.exists() else {}
+
+
+@shop.table(description="Pages.", write_strategy="append", state="logged")
+def pages(ctx):
+    page = ctx.state or 0
+    while page < 4:
+        if failure().get("pages") == page:
+            raise ConnectionError("the API dropped the connection")
+        yield [{"n": page * 10 + row} for row in range(3)]
+        page += 1
+        yield sl.checkpoint(page)
+
+
+@shop.table(description="Events.", write_strategy="append", cursor_column="at")
+def events(ctx):
+    for at in range((ctx.cursor_value or 0) + 1, 7):
+        if failure().get("events") == at:
+            raise TimeoutError("the read timed out")
+        yield {"at": at}
+
+
+@shop.table(description="Broken.")
+def broken(ctx):
+    if failure().get("broken"):
+        raise ValueError("a row is malformed")
+    yield {"id": 1}
+"""
+
+
+def test_given_tables_failing_only_on_the_network_when_loading_then_it_exits_75_and_a_rerun_loads_every_row_once(tmp_path: Path) -> None:
+	project = make_project(tmp_path, DUCKDB_PROJECT, "", module=RETRY_MODULE)
+	give(project, "fail", {"pages": 2, "events": 4})
+	assert sqlload(project, "load", "shop__pages", "shop__events").returncode == 75
+	done = sqlload(project, "load", "--json", "shop__pages", "shop__events")
+	assert done.returncode == 75, done.stdout + done.stderr
+	result = json.loads(done.stdout)
+	assert (result["ok"], result["retryable"]) == (False, True)
+	assert {t["table"]: (t["status"], t["retryable"]) for t in result["tables"]} == {
+		"RAW.shop__pages": ("failed", True),
+		"RAW.shop__events": ("failed", True),
+	}
+	assert "ConnectionError: the API dropped the connection" in next(t["error"] for t in result["tables"] if t["table"] == "RAW.shop__pages")
+	give(project, "fail", {})
+	again = sqlload(project, "load", "--json", "shop__pages", "shop__events")
+	assert again.returncode == 0, again.stdout + again.stderr
+	assert (json.loads(again.stdout)["ok"], json.loads(again.stdout)["retryable"]) == (True, False)
+	assert query(project, 'SELECT n FROM "RAW".shop__pages ORDER BY n') == [(page * 10 + row,) for page in range(4) for row in range(3)]
+	assert query(project, 'SELECT "at" FROM "RAW".shop__events ORDER BY 1') == [(at,) for at in range(1, 7)]
+
+
+def test_given_a_retryable_and_a_lasting_failure_when_loading_then_it_exits_1_and_says_which_is_which(tmp_path: Path) -> None:
+	project = make_project(tmp_path, DUCKDB_PROJECT, "", module=RETRY_MODULE)
+	give(project, "fail", {"pages": 0, "broken": True})
+	assert sqlload(project, "load", "shop__pages", "shop__broken").returncode == 1
+	done = sqlload(project, "load", "--json", "shop__pages", "shop__broken")
+	assert done.returncode == 1, done.stdout + done.stderr
+	result = json.loads(done.stdout)
+	assert (result["ok"], result["retryable"]) == (False, False)
+	assert {t["table"]: t["retryable"] for t in result["tables"]} == {"RAW.shop__pages": True, "RAW.shop__broken": False}
+
+
 def test_given_a_plan_error_and_a_printing_source_when_loading_with_json_then_stdout_parses_and_the_run_is_not_ok(tmp_path: Path) -> None:
 	module = """import sqlload as sl
 
diff --git a/website/src/content/docs/docs/cli/index.mdx b/website/src/content/docs/docs/cli/index.mdx
index 80b8f4f7..cdc78ea6 100644
--- a/website/src/content/docs/docs/cli/index.mdx
+++ b/website/src/content/docs/docs/cli/index.mdx
@@ -53,8 +53,40 @@ Every command takes these:
 | 0 | Everything selected succeeded, or agreed (`contract diff`, `logged-state verify`). |
 | 1 | A table failed or was left out by a plan error, a contract difference or state disagreement was found, or the configuration is invalid. |
 | 2 | The command line couldn't be parsed. |
+| 75 | From 0.34.0 (not released yet). `plan`, `apply`, `load` or `copy` failed, but only in ways that running it again unchanged may fix. See [Retrying failed runs](#retrying-failed-runs). |
 | 130 | `load --follow` got a second Ctrl-C or SIGTERM and stopped at once. |
 
 `--json` doesn't change the exit code. A declined `--reload` exits 1. From 0.27.0,
 `apply --json` and `load --json` print their result as JSON; see
 [Results as JSON](/docs/cli/load/#results-as-json).
+
+## Retrying failed runs
+
+From 0.34.0 (not released yet), `plan`, `apply`, `load` and `copy` exit 75 instead of 1 when every
+failure may pass on a second try, so a scheduler can retry those runs and alert on the rest. A failure counts as
+retryable when it is one of these:
+
+- A connection that dropped, was refused or timed out, to a target, or to a source while planning
+  or loading.
+- Throttling (429) or a server error (5xx) that outlasted its retries, from a warehouse or an
+  [`sl.http`](/docs/sources/python/) job. An object store failure counts only when it's a
+  connection that dropped or timed out.
+- A Postgres connection failure, out of memory, too many connections, a server shutting down or
+  starting, a serialization failure, a deadlock, or a lock not available.
+- A table that another run is loading.
+- A Python table function that raised `ConnectionError` or `TimeoutError`, or a subclass of either.
+
+Anything else exits 1, including invalid configuration, bad credentials, a missing permission, a
+plan error and data that doesn't fit its column. If any one table fails in one of these ways, the
+whole run exits 1, even when other tables failed in retryable ways.
+
+A rerun after a failure loads no row twice and loses none:
+
+- Tables that load in batches keep the batches that committed and carry on from the destination's
+  cursor.
+- Logged tables resume from their last committed checkpoint.
+- `replace` swaps the new table in only at the end.
+
+See [Batches](/docs/concepts/cursors/#batches) and [Logged state](/docs/concepts/cursors/#logged-state).
+Under `--json`, each failed table, and the run as a whole, says whether it is retryable; see
+[Results as JSON](/docs/cli/load/#results-as-json).
diff --git a/website/src/content/docs/docs/cli/load.mdx b/website/src/content/docs/docs/cli/load.mdx
index 07bd24b3..798bb7c4 100644
--- a/website/src/content/docs/docs/cli/load.mdx
+++ b/website/src/content/docs/docs/cli/load.mdx
@@ -71,6 +71,7 @@ $ sqlload load --json
   "schema_version": 1,
   "command": "load",
   "ok": false,
+  "retryable": true,
   "started_at": "2026-11-08T16:33:56.655Z",
   "finished_at": "2026-11-08T16:33:56.731Z",
   "duration_seconds": 0.076,
@@ -96,7 +97,8 @@ $ sqlload load --json
       "rows": 1200,
       "duration_seconds": 0.032,
       "note": null,
-      "error": null
+      "error": null,
+      "retryable": false
     },
     {
       "table": "raw.shop__refunds",
@@ -107,7 +109,8 @@ $ sqlload load --json
       "rows": null,
       "duration_seconds": 0.009,
       "note": null,
-      "error": "RuntimeError: the API returned 503 (File \"/srv/project/sources/shop.py\", line 14, in refunds)"
+      "error": "GET http://api.example.com/v1/refunds?page=3: HTTP 503: unavailable (after 4 attempts)",
+      "retryable": true
     }
   ]
 }
@@ -118,15 +121,18 @@ $ sqlload load --json
 | `schema_version` | `1`. New fields can appear without a change; it changes only when a field changes meaning or goes away. |
 | `command` | `apply` or `load`. |
 | `ok` | Whether every selected table planned and loaded: `true` exactly when the exit code is 0. |
+| `retryable` | From 0.34.0 (not released yet). Whether running the command again unchanged may fix every failure: `true` exactly when the exit code is 75. It's `false` when `ok` is `true`, whenever a table failed for good, and whenever a plan error other than an unreachable source left one out. See [Retrying failed runs](/docs/cli/#retrying-failed-runs). |
 | `summary` | How many objects were `created`, `altered` (each table once, however many changes it had) and `recreated`; tables `loaded` and `failed`; `rows` loaded. |
-| `plan` | The same document `plan --json` prints. Tables a plan error left out are in its `errors`, and tables with nothing to load in `skipped`. |
+| `plan` | The same document `plan --json` prints, with its own `ok` and, from 0.34.0 (not released yet), `retryable`. Tables a plan error left out are in its `errors`, and tables with nothing to load in `skipped`. |
 | `objects` | One entry per change to a table or stage, by `apply` or during a load: its `action` (`created`, `altered` or `recreated`), its `name` and the `detail`. A table that gains two columns has two `altered` entries. |
-| `tables` | Each table loaded: its `status` (`loaded` or `failed`), `rows`, `duration_seconds`, a `note` (the cursor window it read, or that its table waits for first data), and the `error` when it failed. `source`, `target` and `write_strategy` are `null` only for a `load --follow` that failed before any table. `apply` loads nothing, so its list is empty. |
+| `tables` | Each table loaded: its `status` (`loaded` or `failed`), `rows`, `duration_seconds`, a `note` (the cursor window it read, or that its table waits for first data), and the `error` when it failed. From 0.34.0 (not released yet), `retryable` says whether that failure may pass on a rerun. `source`, `target` and `write_strategy` are `null` only for a `load --follow` that failed before any table. `apply` loads nothing, so its list is empty. |
 
 When the run can't start (the configuration doesn't parse, a connection fails), standard output
 holds one error document instead, the same as `plan --json`'s:
-`{"error": {"code": "config", "message": "…", "help": "…"}}`. `code` is `config` for the
-configuration, else the command's name.
+`{"error": {"code": "config", "message": "…", "help": "…", "retryable": false}}`. `code` is
+`config` for the configuration, else the command's name. From 0.34.0 (not released yet),
+`retryable` is `true`, and the exit code 75, when the run couldn't start for a reason a rerun may
+fix, such as a target that refused the connection or timed out.
 
 `--reload` and `--resnapshot` need `--yes` with `--json`, as there is no prompt to answer.