Skip to content

Commit 6911cbc

Browse files
committed
feat(ingest): imports return immediately; add status for tracking
new-import and trigger-import no longer block the terminal. They enqueue, fire the drain that processes the job (the worker is on-demand -- the old --no-wait skipped this, leaving jobs pending forever), print the ingest id + database + a tracking hint, and return. --wait opts into the old blocking poll. new-connection keeps its blocking default deliberately: its output IS the feedback (credentials validated, schema discovered); --no-wait still opts out there. New `ingest status <id>`: - one-shot by default with script-friendly exit codes mirroring `query status`: 0 done, 1 failed, 2 still in flight; prints the worker's failure detail when there is one - --wait attaches and polls to a terminal state; the poll still re-kicks a job stuck in `pending` (control-store read lag), so attaching also rescues an import whose first drain missed the row The wait poll now tolerates transient errors: a single gateway blip (403/5xx, dropped connection) mid-poll retried up to 3 consecutive times instead of killing the wait (seen live: a one-off empty-body 403 aborted an otherwise-healthy wait). Verified live against prod: non-blocking enqueue returns in ~2s with the pending ack, status one-shot exits 0/2 correctly, status --wait attaches to done.
1 parent e8d0119 commit 6911cbc

1 file changed

Lines changed: 175 additions & 35 deletions

File tree

‎src/commands/ingest.rs‎

Lines changed: 175 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -5,8 +5,15 @@
55
//! (managed DBs materialized from a connection). Commands: `new-connection`,
66
//! `list-connections`, `connectors` (the catalog), `new-import` (--all or
77
//! SQL), `list-imports`, `trigger-import` (re-run: refresh an ingest's DB
8-
//! from source). The pre-rename verbs (`new`/`list`/`import`/`update`) remain
9-
//! as hidden aliases for one release.
8+
//! from source), `status` (one-shot or `--wait` attach). The pre-rename verbs
9+
//! (`new`/`list`/`import`/`update`) remain as hidden aliases for one release.
10+
//!
11+
//! **Imports don't block.** `new-import` and `trigger-import` enqueue, fire
12+
//! the drain, print the ingest id, and return; progress is tracked with
13+
//! `status <id> [--wait]` or `list-imports`. `--wait` opts into the old
14+
//! blocking poll. `new-connection` is the deliberate exception — it blocks
15+
//! by default because its output IS the feedback (credentials validated,
16+
//! schema discovered); `--no-wait` opts out.
1017
//!
1118
//! `new-connection` **adds a connection and discovers its schema — it loads no
1219
//! data** (the server's `validate_only` mode: check credentials, reflect the
@@ -127,8 +134,14 @@ pub enum IngestCommands {
127134
#[arg(long = "database-id")]
128135
database_id: Option<String>,
129136

130-
#[command(flatten)]
131-
wait: WaitArgs,
137+
/// Block until the import completes (default: return immediately;
138+
/// track with `hotdata ingest status <id> --wait`)
139+
#[arg(long)]
140+
wait: bool,
141+
142+
/// Seconds to wait for completion with --wait (default 300)
143+
#[arg(long = "wait-timeout", default_value = "300")]
144+
wait_timeout: u64,
132145
},
133146

134147
/// List your imports: the SQL behind each and the database it landed in
@@ -145,13 +158,40 @@ pub enum IngestCommands {
145158
/// `list-connections` (re-validates and refreshes its schema)
146159
id: String,
147160

148-
/// Seconds to wait for completion (default 300)
161+
/// Block until the re-run completes (default: return immediately;
162+
/// track with `hotdata ingest status <id> --wait`)
163+
#[arg(long)]
164+
wait: bool,
165+
166+
/// Seconds to wait for completion with --wait (default 300)
167+
#[arg(long = "wait-timeout", default_value = "300")]
168+
wait_timeout: u64,
169+
},
170+
171+
/// Show an ingest's status (an import or a connection onboard)
172+
///
173+
/// One-shot by default with script-friendly exit codes: 0 done,
174+
/// 1 failed, 2 still in flight. `--wait` attaches and polls to a
175+
/// terminal state instead.
176+
Status {
177+
/// Ingest id (from `new-import`, `list-imports`, or `list-connections`)
178+
id: String,
179+
180+
/// Poll until the ingest reaches done/failed instead of returning
181+
/// the current status
182+
#[arg(long)]
183+
wait: bool,
184+
185+
/// Seconds to wait for completion with --wait (default 300)
149186
#[arg(long = "wait-timeout", default_value = "300")]
150187
wait_timeout: u64,
151188
},
152189
}
153190

154-
/// Shared wait/no-wait flags for the enqueue commands.
191+
/// Wait flags for `new-connection` — the one enqueue command that still
192+
/// blocks by default, because its output IS the feedback: did the
193+
/// credentials work, and what schema was discovered. The import commands
194+
/// return immediately and are tracked with `ingest status`.
155195
#[derive(clap::Args)]
156196
pub struct WaitArgs {
157197
/// Enqueue only; print the ingest id and return without waiting
@@ -203,11 +243,28 @@ pub fn dispatch(workspace_id: &str, output: &str, command: IngestCommands) {
203243
source,
204244
database_id,
205245
wait,
206-
} => new_import(workspace_id, output, sql, all, source, database_id, &wait),
246+
wait_timeout,
247+
} => new_import(
248+
workspace_id,
249+
output,
250+
sql,
251+
all,
252+
source,
253+
database_id,
254+
wait,
255+
wait_timeout,
256+
),
207257
IngestCommands::ListImports => list_imports(workspace_id, output),
208-
IngestCommands::TriggerImport { id, wait_timeout } => {
209-
trigger_import(workspace_id, output, &id, wait_timeout)
210-
}
258+
IngestCommands::TriggerImport {
259+
id,
260+
wait,
261+
wait_timeout,
262+
} => trigger_import(workspace_id, output, &id, wait, wait_timeout),
263+
IngestCommands::Status {
264+
id,
265+
wait,
266+
wait_timeout,
267+
} => status(workspace_id, output, &id, wait, wait_timeout),
211268
}
212269
}
213270

@@ -607,14 +664,16 @@ fn active_source_names(client: &IngestClient) -> std::collections::HashSet<Strin
607664

608665
// --- new-import: SQL front-door -------------------------------------------
609666

667+
#[allow(clippy::too_many_arguments)] // mirrors the clap surface one-to-one
610668
fn new_import(
611669
workspace_id: &str,
612670
output: &str,
613671
sql: Option<String>,
614672
all: bool,
615673
source: Option<String>,
616674
database_id: Option<String>,
617-
wait: &WaitArgs,
675+
wait: bool,
676+
wait_timeout: u64,
618677
) {
619678
// A 32-char hex --source is a connection id (pins resolution); anything
620679
// else is a connector name (the FROM target).
@@ -659,18 +718,16 @@ fn new_import(
659718
e.exit()
660719
});
661720
spinner.finish_and_clear();
662-
if wait.no_wait {
663-
render_ack(&ack, output);
721+
if wait {
722+
let st = poll_ingest(&client, &ack.ingest_id, wait_timeout, "importing", true);
723+
render_done(&st, output);
664724
return;
665725
}
666-
let st = poll_ingest(
667-
&client,
668-
&ack.ingest_id,
669-
wait.wait_timeout,
670-
"importing",
671-
true,
672-
);
673-
render_done(&st, output);
726+
// Non-blocking default: the worker is on-demand, so fire the drain that
727+
// actually processes the job — without it the import would sit pending —
728+
// then hand the terminal back.
729+
let _ = client.drain();
730+
render_ack(&ack, output);
674731
}
675732

676733
fn looks_like_ingest_id(s: &str) -> bool {
@@ -713,7 +770,7 @@ fn pinned_connector_name(client: &IngestClient, ingest_id: &str) -> Option<Strin
713770

714771
// --- trigger-import --------------------------------------------------------
715772

716-
fn trigger_import(workspace_id: &str, output: &str, id: &str, wait_timeout: u64) {
773+
fn trigger_import(workspace_id: &str, output: &str, id: &str, wait: bool, wait_timeout: u64) {
717774
let client = IngestClient::new(workspace_id);
718775
let spinner = util::spinner("requesting re-run…");
719776
// The worker resets the ingest to pending and fires the drain (409 if one
@@ -723,9 +780,69 @@ fn trigger_import(workspace_id: &str, output: &str, id: &str, wait_timeout: u64)
723780
e.exit()
724781
});
725782
spinner.finish_and_clear();
726-
// The rerun already drained; poll only.
727-
let st = poll_ingest(&client, &ack.ingest_id, wait_timeout, "re-running", false);
728-
render_done(&st, output);
783+
if wait {
784+
// The rerun already drained; poll only.
785+
let st = poll_ingest(&client, &ack.ingest_id, wait_timeout, "re-running", false);
786+
render_done(&st, output);
787+
return;
788+
}
789+
render_ack(&ack, output);
790+
}
791+
792+
// --- status ------------------------------------------------------------------
793+
794+
/// Exit code for a one-shot status check, mirroring `query status`:
795+
/// 0 done, 1 failed, 2 still in flight (pending/running/stage states).
796+
fn status_exit_code(status: &str) -> i32 {
797+
match status {
798+
"done" => 0,
799+
"failed" => 1,
800+
_ => 2,
801+
}
802+
}
803+
804+
fn status(workspace_id: &str, output: &str, id: &str, wait: bool, wait_timeout: u64) {
805+
let client = IngestClient::new(workspace_id);
806+
if wait {
807+
// Attach and poll to a terminal state. No initial drain kick — the
808+
// enqueue already fired one — but the poll loop still re-kicks a job
809+
// stuck in `pending` (control-store read lag), so attaching also
810+
// rescues an import whose first drain missed the row.
811+
let st = poll_ingest(&client, id, wait_timeout, "waiting", false);
812+
render_done(&st, output);
813+
return;
814+
}
815+
let st = client.job_status(id).unwrap_or_else(|e| e.exit());
816+
match output {
817+
"json" => println!("{}", serde_json::to_string_pretty(&st).unwrap()),
818+
"yaml" => print!("{}", serde_yaml::to_string(&st).unwrap()),
819+
_ => {
820+
use crossterm::style::Stylize;
821+
let label = |l: &str| format!("{:<14}", l).dark_grey().to_string();
822+
let colored = match st.status.as_str() {
823+
"done" => st.status.as_str().green(),
824+
"failed" => st.status.as_str().red(),
825+
other => other.yellow(),
826+
};
827+
println!("{}{}", label("status:"), colored);
828+
if let Some(d) = st.detail.as_deref().filter(|d| !d.trim().is_empty()) {
829+
println!("{}{}", label("detail:"), d);
830+
}
831+
if let Some(db) = st.database_id.as_deref() {
832+
println!("{}{}", label("database:"), db);
833+
}
834+
if let Some(t) = st.updated_at.as_deref() {
835+
println!("{}{}", label("updated:"), t);
836+
}
837+
if status_exit_code(&st.status) == 2 {
838+
println!(
839+
"{}",
840+
format!("Attach with: hotdata ingest status {id} --wait").dark_grey()
841+
);
842+
}
843+
}
844+
}
845+
std::process::exit(status_exit_code(&st.status));
729846
}
730847

731848
// --- list-connections ------------------------------------------------------
@@ -890,14 +1007,29 @@ fn poll_ingest(
8901007
let _ = client.drain();
8911008
}
8921009
let mut last_drain = Instant::now();
1010+
let mut consecutive_errors: u32 = 0;
8931011

8941012
let spinner = util::spinner(&format!("{verb}…"));
8951013
let deadline = Instant::now() + Duration::from_secs(timeout_secs);
8961014
loop {
897-
let st = client.job_status(ingest_id).unwrap_or_else(|e| {
898-
spinner.finish_and_clear();
899-
e.exit()
900-
});
1015+
// A status poll is read-only and cheap to retry — one transient
1016+
// gateway error (a 403/5xx blip, a dropped connection) must not kill
1017+
// a wait that is otherwise progressing. Only persistent failure exits.
1018+
let st = match client.job_status(ingest_id) {
1019+
Ok(st) => {
1020+
consecutive_errors = 0;
1021+
st
1022+
}
1023+
Err(e) => {
1024+
consecutive_errors += 1;
1025+
if consecutive_errors >= 3 || Instant::now() > deadline {
1026+
spinner.finish_and_clear();
1027+
e.exit();
1028+
}
1029+
std::thread::sleep(POLL_INTERVAL);
1030+
continue;
1031+
}
1032+
};
9011033
match st.status.as_str() {
9021034
"done" => {
9031035
spinner.finish_and_clear();
@@ -943,11 +1075,8 @@ fn poll_ingest(
9431075
eprintln!("{}", "ingest timed out".red());
9441076
eprintln!(
9451077
"{}",
946-
format!(
947-
"Check progress with: hotdata ingest list-imports (or list-connections) — \
948-
or re-kick it: hotdata ingest trigger-import {ingest_id}"
949-
)
950-
.dark_grey()
1078+
format!("Keep tracking it with: hotdata ingest status {ingest_id} --wait")
1079+
.dark_grey()
9511080
);
9521081
std::process::exit(2);
9531082
}
@@ -979,7 +1108,7 @@ fn render_ack(ack: &IngestAck, output: &str) {
9791108
println!(
9801109
"{}",
9811110
format!(
982-
"Track it with: hotdata ingest trigger-import {}",
1111+
"Track it with: hotdata ingest status {} --wait (or: hotdata ingest list-imports)",
9831112
ack.ingest_id
9841113
)
9851114
.dark_grey()
@@ -1354,6 +1483,17 @@ mod tests {
13541483
assert!(err.contains("--source"), "got: {err}");
13551484
}
13561485

1486+
#[test]
1487+
fn status_exit_codes_mirror_query_status() {
1488+
assert_eq!(status_exit_code("done"), 0);
1489+
assert_eq!(status_exit_code("failed"), 1);
1490+
// Everything else is in flight — including worker stage states the
1491+
// CLI doesn't enumerate (extracting/normalizing/loading).
1492+
assert_eq!(status_exit_code("pending"), 2);
1493+
assert_eq!(status_exit_code("running"), 2);
1494+
assert_eq!(status_exit_code("loading"), 2);
1495+
}
1496+
13571497
#[test]
13581498
fn date_only_trims_iso_timestamps() {
13591499
assert_eq!(date_only(Some("2026-07-08T10:12:00+00:00")), "2026-07-08");

0 commit comments

Comments
 (0)