feat(load): retry a load while another load holds the table

skills/hotdata/SKILL.md

@@ -118,12 +118,12 @@ hotdata databases detach <database> [--database <id>]                     # or t
 
 # Preferred: load by catalog alias (server declares the table/schema if missing).
 # Loads csv, json, or parquet — format read from the extension.
-hotdata databases load --catalog <alias> --table <table> [--schema public] (--file <path> | --url <url> | --upload-id <id> | --result-id <id>) [--mode replace|append|delete|update|upsert] [--append] [--format csv|json|parquet] [--key <col>]... [--workspace-id <workspace_id>]
+hotdata databases load --catalog <alias> --table <table> [--schema public] (--file <path> | --url <url> | --upload-id <id> | --result-id <id>) [--mode replace|append|delete|update|upsert] [--append] [--format csv|json|parquet] [--key <col>]... [--retry-timeout <secs>] [--workspace-id <workspace_id>]
 
 # Also available via tables subcommand
 hotdata databases tables list [--database <id>] [--schema <name>] [--workspace-id <workspace_id>] [--output table|json|yaml]
 hotdata databases tables add <table|schema.table> [--database <id>] [--schema public] [--key <col>]... [--key-determines <col>]... [--sorted-by <col>[=asc|desc]]... [--partition-by <col>[=identity|year|month|day|hour]]... [--output table|json|yaml]
-hotdata databases tables load <table> [--database <id>] [--schema public] (--file <path> | --url <url> | --upload-id <id> | --result-id <id>) [--mode replace|append|delete|update|upsert] [--append] [--format csv|json|parquet] [--key <col>]... [--workspace-id <workspace_id>]
+hotdata databases tables load <table> [--database <id>] [--schema public] (--file <path> | --url <url> | --upload-id <id> | --result-id <id>) [--mode replace|append|delete|update|upsert] [--append] [--format csv|json|parquet] [--key <col>]... [--retry-timeout <secs>] [--workspace-id <workspace_id>]
 hotdata databases tables remove <table> [--database <id>] [--schema public] [--workspace-id <workspace_id>]
 ```
 
@@ -137,6 +137,7 @@ hotdata databases tables remove <table> [--database <id>] [--schema public] [--w
 - `show <id>` / `<id>` — inspect one database (returns id, catalog, name, expires_at, attached databases; a fork also shows its `forked_from` record).
 - `remove` — removes the instant database; clears the active-database config if it matched.
 - `load` (top-level shorthand) — loads a file into `--catalog.--schema.--table`. Accepts `--file`, `--url`, `--upload-id`, or `--result-id` (load a saved query result by id — from `hotdata databases results` or a query's `[result-id: …]` footer — instead of a file; the result must belong to the target database). **Formats:** csv, json (`.json`/`.jsonl`/`.ndjson`), and parquet; the format comes from the file's extension, and `--format` overrides it (needed when the extension is absent or misleading). An unrecognised extension is not rejected — the server reads the bytes, and a file that plainly opens a json array is taken as json even without an extension. A json source in any shape (array, pretty-printed, one object per line) is reshaped locally to newline-delimited json before upload; an already-newline-delimited file is uploaded untouched. A table or schema that was never declared is declared by the server as part of the load, so no up-front `--table` is required.
+- **Concurrent loads** — loads into one table run one at a time. While another load holds the table the server answers 409 `RESOURCE_LOCKED` with a `Retry-After`; the CLI waits and retries for up to `--retry-timeout` seconds (default 60). Pass `--retry-timeout 0` to fail immediately instead. Loads into different tables never block each other.
 - **Load modes** (`--mode`, default `replace`) — `replace` supersedes the table's contents; `append` adds rows; `delete`, `update`, and `upsert` match existing rows **by key**. `--append` is the old shorthand for `--mode append` and still works, but the two cannot be combined. The keyed modes need a key: declare one with `databases tables add --key`, or name it per-load with `--key` (repeat for a composite key). `delete` uploads only the key columns; `update` replaces matching rows and ignores unmatched ones; `upsert` inserts the unmatched instead. Keyed modes are not available with `--result-id`.
 - `tables list` — lists tables with `TABLE` (`<catalog>.<schema>.<table>`), `SYNCED`, `LAST_SYNC`. Uses current database when `--database` is omitted.
 - `tables add` — declares a table **with its key and storage layout**, which a load cannot infer. `--key` (repeatable) is what enables the `delete`/`update`/`upsert` load modes on that table. `--sorted-by <col>` or `<col>=desc` sets sort order; `--partition-by <col>` partitions on the value, `<col>=month` (or `year`/`day`/`hour`) on a calendar part — one partition per calendar month needs **both** `<col>=year` and `<col>=month`, or every March shares a partition. Sort and partition are fixed once the table exists. `--key-determines` (repeatable) asserts a column's value is fixed by the key: it prunes keyed loads harder, and is **correctness-affecting** — declare it only where the invariant really holds, or a keyed load can leave a duplicate key behind. Re-adding an existing table is a conflict (409), and `tables remove` does not clear the declaration — the table leaves the listing but the name stays declared and still conflicts. So **a key cannot be retrofitted onto a table declared without one**: declare it with `--key` up front, or use a new table name.

src/client/sdk.rs

@@ -855,6 +855,19 @@ impl Api {
         path: &str,
         body: &serde_json::Value,
     ) -> Result<(reqwest::StatusCode, String), ApiError> {
+        self.post_raw_with_headers(path, body)
+            .map(|(status, _, body)| (status, body))
+    }
+
+    /// [`post_raw`](Self::post_raw) that also returns the response headers.
+    ///
+    /// For callers that act on a header as well as the status — the load
+    /// endpoint's `Retry-After` on a `RESOURCE_LOCKED` 409 is the one in use.
+    pub fn post_raw_with_headers(
+        &self,
+        path: &str,
+        body: &serde_json::Value,
+    ) -> Result<(reqwest::StatusCode, reqwest::header::HeaderMap, String), ApiError> {
         let cfg = self.client.configuration();
         let url = format!("{}/v1{path}", cfg.base_path);
         let database_id = self.database_id.clone();
@@ -867,11 +880,12 @@ impl Api {
                 .await
                 .map_err(|e| ApiError::Transport(format!("error connecting to API: {e}")))?;
             let status = resp.status();
+            let headers = resp.headers().clone();
             let body = resp
                 .text()
                 .await
                 .map_err(|e| ApiError::Transport(format!("error connecting to API: {e}")))?;
-            Ok((status, body))
+            Ok((status, headers, body))
         })
     }
 

src/commands/databases.rs

@@ -251,6 +251,13 @@ pub enum DatabasesCommands {
         /// table was declared with.
         #[arg(long = "key")]
         key: Vec<String>,
+
+        /// Seconds to keep retrying while another load holds the table
+        /// (default 60; 0 fails immediately). Concurrent loads into one table
+        /// are serialised server-side, and a busy table answers 409
+        /// `RESOURCE_LOCKED` with a `Retry-After` the CLI honours.
+        #[arg(long = "retry-timeout", default_value = "60")]
+        retry_timeout: u64,
     },
 
     /// Manage tables inside an instant database
@@ -480,6 +487,13 @@ pub enum DatabaseTablesCommands {
         /// table was declared with.
         #[arg(long = "key")]
         key: Vec<String>,
+
+        /// Seconds to keep retrying while another load holds the table
+        /// (default 60; 0 fails immediately). Concurrent loads into one table
+        /// are serialised server-side, and a busy table answers 409
+        /// `RESOURCE_LOCKED` with a `Retry-After` the CLI honours.
+        #[arg(long = "retry-timeout", default_value = "60")]
+        retry_timeout: u64,
     },
 
     /// Delete a table from an instant database
@@ -1221,6 +1235,70 @@ pub fn load_table_request_from_result(
     body
 }
 
+/// The wait a load 409 asks for, or `None` when the 409 is not the table lock.
+///
+/// Loads into one table are serialised server-side; a second writer gets a 409
+/// whose code is `RESOURCE_LOCKED` and whose `Retry-After` says how long to
+/// hold off. Matched on the code, never the message: a permanent `CONFLICT`
+/// (idempotency-key reuse, a catalog scoped elsewhere) is also a 409 and must
+/// not be retried. A missing or unparsable `Retry-After` falls back to
+/// [`LOCK_RETRY_FALLBACK`] — the lock is still transient without it.
+pub fn lock_retry_after(
+    status: reqwest::StatusCode,
+    headers: &reqwest::header::HeaderMap,
+    body: &str,
+) -> Option<std::time::Duration> {
+    if status != reqwest::StatusCode::CONFLICT {
+        return None;
+    }
+    let code = serde_json::from_str::<serde_json::Value>(body).ok()?;
+    if code["error"]["code"].as_str() != Some("RESOURCE_LOCKED") {
+        return None;
+    }
+    let secs = headers
+        .get(reqwest::header::RETRY_AFTER)
+        .and_then(|v| v.to_str().ok())
+        .and_then(|v| v.trim().parse::<u64>().ok())
+        .map(std::time::Duration::from_secs)
+        .unwrap_or(LOCK_RETRY_FALLBACK);
+    Some(secs)
+}
+
+/// Wait between load attempts when the lock 409 carries no usable
+/// `Retry-After`. Matches what the server advertises today.
+pub const LOCK_RETRY_FALLBACK: std::time::Duration = std::time::Duration::from_secs(5);
+
+/// `POST` a load, retrying while the table is held by another load.
+///
+/// Every attempt is the same request — a load is idempotent per upload, and
+/// the server either ran it or rejected it outright — so repeating it is safe.
+/// Retries stop at `retry_timeout_secs` after the first attempt; the final 409
+/// is then returned as-is so the caller renders the server's own message. A
+/// timeout of `0` sends exactly once. `on_wait` is told each pause before it
+/// happens, so a long wait is never silent.
+pub fn post_load_with_retry(
+    api: &Api,
+    path: &str,
+    body: &serde_json::Value,
+    retry_timeout_secs: u64,
+    mut on_wait: impl FnMut(std::time::Duration),
+) -> Result<(reqwest::StatusCode, String), ApiError> {
+    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(retry_timeout_secs);
+    loop {
+        let (status, headers, resp_body) = api.post_raw_with_headers(path, body)?;
+        let Some(wait) = lock_retry_after(status, &headers, &resp_body) else {
+            return Ok((status, resp_body));
+        };
+        // The deadline bounds the whole wait, so a pause that would overrun it
+        // is not taken: better to report "still busy" on time than late.
+        if std::time::Instant::now() + wait > deadline {
+            return Ok((status, resp_body));
+        }
+        on_wait(wait);
+        std::thread::sleep(wait);
+    }
+}
+
 /// The load-request `format` implied by `name`'s extension, or `None` when the
 /// extension names nothing we recognise.
 ///
@@ -2870,6 +2948,7 @@ pub fn tables_load(
     mode: Option<&str>,
     format: Option<&str>,
     key: &[String],
+    retry_timeout_secs: u64,
 ) {
     use crossterm::style::Stylize;
 
@@ -2964,10 +3043,17 @@ pub fn tables_load(
     let path = managed_table_load_path(&db.default_connection_id, schema, table);
 
     let spinner = crate::util::spinner("Loading table...");
-    let (status, resp_body) = api.post_raw(&path, &body).unwrap_or_else(|e| {
-        spinner.finish_and_clear();
-        e.exit()
-    });
+    let (status, resp_body) =
+        post_load_with_retry(&api, &path, &body, retry_timeout_secs, |wait| {
+            spinner.set_message(format!(
+                "Loading table... another load holds the table, retrying in {}s",
+                wait.as_secs()
+            ));
+        })
+        .unwrap_or_else(|e| {
+            spinner.finish_and_clear();
+            e.exit()
+        });
     spinner.finish_and_clear();
 
     // No undeclared-table recovery here: the load endpoint declares a missing
@@ -4422,6 +4508,93 @@ mod tests {
         fork.assert();
     }
 
+    #[test]
+    fn lock_retry_after_matches_resource_locked_only() {
+        use reqwest::header::{HeaderMap, HeaderValue, RETRY_AFTER};
+        let locked = r#"{"error":{"code":"RESOURCE_LOCKED","message":"another operation is already running for conn:c:public:t; retry shortly"}}"#;
+        let mut headers = HeaderMap::new();
+        headers.insert(RETRY_AFTER, HeaderValue::from_static("5"));
+        assert_eq!(
+            lock_retry_after(reqwest::StatusCode::CONFLICT, &headers, locked),
+            Some(std::time::Duration::from_secs(5))
+        );
+        // No header: still transient, fall back rather than give up.
+        assert_eq!(
+            lock_retry_after(reqwest::StatusCode::CONFLICT, &HeaderMap::new(), locked),
+            Some(LOCK_RETRY_FALLBACK)
+        );
+        // A permanent 409 is never retried, however it is worded.
+        let conflict = r#"{"error":{"code":"CONFLICT","message":"another operation is already running; retry shortly"}}"#;
+        assert_eq!(
+            lock_retry_after(reqwest::StatusCode::CONFLICT, &headers, conflict),
+            None
+        );
+        // Same code on a different status is not a lock.
+        assert_eq!(
+            lock_retry_after(reqwest::StatusCode::BAD_REQUEST, &headers, locked),
+            None
+        );
+    }
+
+    #[test]
+    fn load_retries_resource_locked_until_it_succeeds() {
+        let mut server = mockito::Server::new();
+        let locked = server
+            .mock("POST", "/v1/connections/c1/schemas/public/tables/t/loads")
+            .with_status(409)
+            .with_header("retry-after", "0")
+            .with_body(r#"{"error":{"code":"RESOURCE_LOCKED","message":"another operation is already running for conn:c1:public:t; retry shortly"}}"#)
+            .expect(2)
+            .create();
+        let ok = server
+            .mock("POST", "/v1/connections/c1/schemas/public/tables/t/loads")
+            .with_status(200)
+            .with_body(r#"{"connection_id":"c1","schema_name":"public","table_name":"t","row_count":1,"arrow_schema_json":"{}"}"#)
+            .create();
+
+        let api = Api::test_new(&server.url(), "k", Some("ws1"));
+        let body = load_table_request("upl_1", "append", None, &[]);
+        let mut waits = 0;
+        let (status, _) = post_load_with_retry(
+            &api,
+            "/connections/c1/schemas/public/tables/t/loads",
+            &body,
+            30,
+            |_| waits += 1,
+        )
+        .unwrap();
+        assert!(status.is_success());
+        assert_eq!(waits, 2, "caller is told about every pause");
+        locked.assert();
+        ok.assert();
+    }
+
+    #[test]
+    fn load_retry_timeout_zero_sends_once_and_returns_the_409() {
+        let mut server = mockito::Server::new();
+        let locked = server
+            .mock("POST", "/v1/connections/c1/schemas/public/tables/t/loads")
+            .with_status(409)
+            .with_header("retry-after", "5")
+            .with_body(r#"{"error":{"code":"RESOURCE_LOCKED","message":"busy"}}"#)
+            .expect(1)
+            .create();
+
+        let api = Api::test_new(&server.url(), "k", Some("ws1"));
+        let body = load_table_request("upl_1", "append", None, &[]);
+        let (status, resp) = post_load_with_retry(
+            &api,
+            "/connections/c1/schemas/public/tables/t/loads",
+            &body,
+            0,
+            |_| panic!("must not wait with a zero budget"),
+        )
+        .unwrap();
+        assert_eq!(status.as_u16(), 409);
+        assert_eq!(crate::util::api_error(resp), "busy");
+        locked.assert();
+    }
+
     #[test]
     fn tables_load_uses_default_connection_id() {
         let mut server = mockito::Server::new();

src/main.rs

@@ -320,6 +320,7 @@ fn main() {
                             mode,
                             format,
                             key,
+                            retry_timeout,
                         }) => databases::tables_load(
                             &workspace_id,
                             Some(catalog.as_str()),
@@ -333,6 +334,7 @@ fn main() {
                             mode.as_deref(),
                             format.as_deref(),
                             &key,
+                            retry_timeout,
                         ),
                         Some(DatabasesCommands::Tables { database, command }) => match command {
                             Some(DatabaseTablesCommands::List {
@@ -409,6 +411,7 @@ fn main() {
                                 mode,
                                 format,
                                 key,
+                                retry_timeout,
                             }) => databases::tables_load(
                                 &workspace_id,
                                 db_flag.as_deref().or(database.as_deref()),
@@ -422,6 +425,7 @@ fn main() {
                                 mode.as_deref(),
                                 format.as_deref(),
                                 &key,
+                                retry_timeout,
                             ),
                             Some(DatabaseTablesCommands::Delete {
                                 database: db_flag,