diff --git a/skills/hotdata/SKILL.md b/skills/hotdata/SKILL.md index 2b59281..26b3089 100644 --- a/skills/hotdata/SKILL.md +++ b/skills/hotdata/SKILL.md @@ -118,12 +118,12 @@ hotdata databases detach [--database ] # 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 --table [--schema public] (--file | --url | --upload-id | --result-id ) [--mode replace|append|delete|update|upsert] [--append] [--format csv|json|parquet] [--key ]... [--workspace-id ] +hotdata databases load --catalog --table
[--schema public] (--file | --url | --upload-id | --result-id ) [--mode replace|append|delete|update|upsert] [--append] [--format csv|json|parquet] [--key ]... [--retry-timeout ] [--workspace-id ] # Also available via tables subcommand hotdata databases tables list [--database ] [--schema ] [--workspace-id ] [--output table|json|yaml] hotdata databases tables add [--database ] [--schema public] [--key ]... [--key-determines ]... [--sorted-by [=asc|desc]]... [--partition-by [=identity|year|month|day|hour]]... [--output table|json|yaml] -hotdata databases tables load
[--database ] [--schema public] (--file | --url | --upload-id | --result-id ) [--mode replace|append|delete|update|upsert] [--append] [--format csv|json|parquet] [--key ]... [--workspace-id ] +hotdata databases tables load
[--database ] [--schema public] (--file | --url | --upload-id | --result-id ) [--mode replace|append|delete|update|upsert] [--append] [--format csv|json|parquet] [--key ]... [--retry-timeout ] [--workspace-id ] hotdata databases tables remove
[--database ] [--schema public] [--workspace-id ] ``` @@ -137,6 +137,7 @@ hotdata databases tables remove
[--database ] [--schema public] [--w - `show ` / `` — 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` (`..
`), `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 ` or `=desc` sets sort order; `--partition-by ` partitions on the value, `=month` (or `year`/`day`/`hour`) on a calendar part — one partition per calendar month needs **both** `=year` and `=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. diff --git a/src/client/sdk.rs b/src/client/sdk.rs index a519d0c..2e427a2 100644 --- a/src/client/sdk.rs +++ b/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)) }) } diff --git a/src/commands/databases.rs b/src/commands/databases.rs index aa638a8..bca232a 100644 --- a/src/commands/databases.rs +++ b/src/commands/databases.rs @@ -251,6 +251,13 @@ pub enum DatabasesCommands { /// table was declared with. #[arg(long = "key")] key: Vec, + + /// 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, + + /// 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 { + if status != reqwest::StatusCode::CONFLICT { + return None; + } + let code = serde_json::from_str::(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::().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(); diff --git a/src/main.rs b/src/main.rs index a99676a..0efa961 100644 --- a/src/main.rs +++ b/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,