Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 24 additions & 0 deletions src/client/ingest.rs
Original file line number Diff line number Diff line change
Expand Up @@ -285,6 +285,17 @@ impl IngestClient {
)
}

/// Cancel a running or pending ingest. The dltHub drain may still be
/// running (we can't interrupt it remotely); the server marks the status
/// as `cancelled` — a non-terminal state that keeps the /rerun in-flight
/// guard active. A 409 from /rerun means the drain hasn't settled yet.
pub fn cancel(&self, ingest_id: &str) -> Result<IngestCancelAck, IngestError> {
self.send(
self.authed(reqwest::Method::POST, &format!("/jobs/{ingest_id}/cancel")),
None,
)
}

// --- read endpoints --------------------------------------------------

/// The connector catalog. REST entries carry a ready-to-edit `template`
Expand Down Expand Up @@ -460,6 +471,19 @@ pub struct JobStatus {
pub updated_at: Option<String>,
}

/// `POST /jobs/{id}/cancel` body.
#[derive(Debug, Deserialize, Serialize)]
pub struct IngestCancelAck {
pub ingest_id: String,
pub status: String,
#[serde(default)]
pub detail: Option<String>,
#[serde(default)]
pub name: Option<String>,
#[serde(default)]
pub database_id: Option<String>,
}

/// `DELETE /sources/{id}` body.
#[derive(Debug, Deserialize, Serialize)]
pub struct DeleteSourceAck {
Expand Down
50 changes: 47 additions & 3 deletions src/commands/ingest.rs
Original file line number Diff line number Diff line change
Expand Up @@ -190,6 +190,17 @@ pub enum IngestCommands {
#[command(flatten)]
poll: PollArgs,
},

/// Cancel a running or pending import
///
/// Marks the import cancelled immediately. The dltHub drain may still be
/// running and will finish on its own — use `status <id> --wait` to see
/// when it settles. To retry after cancellation, use `trigger-import <id>`
/// once the status reaches `failed`.
Cancel {
/// Import or datasource id to cancel
id: String,
},
}

/// Wait flags shared by the non-blocking commands (`new-import`,
Expand Down Expand Up @@ -246,6 +257,7 @@ pub fn dispatch(workspace_id: &str, output: &str, command: IngestCommands) {
trigger_import(workspace_id, output, &id, &poll)
}
IngestCommands::Status { id, poll } => status(workspace_id, output, &id, &poll),
IngestCommands::Cancel { id } => cancel_import(workspace_id, output, &id),
}
}

Expand All @@ -261,6 +273,7 @@ fn normalize_status(raw: &str) -> (&'static str, Option<&str>) {
"running" => ("running", None),
"done" => ("done", None),
"failed" => ("failed", None),
"cancelled" => ("cancelled", None),
stage => ("running", Some(stage)),
}
}
Expand Down Expand Up @@ -986,7 +999,7 @@ fn trigger_import(workspace_id: &str, output: &str, id: &str, poll: &PollArgs) {
fn status_exit_code(status: &str) -> i32 {
match normalize_status(status).0 {
"done" => 0,
"failed" => 1,
"failed" | "cancelled" => 1,

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: the adjacent unit tests (status_exit_codes_close_over_stage_states, statuses_normalize_to_the_closed_set) cover every other status but not the new cancelled branch. Adding assert_eq!(status_exit_code("cancelled"), 1) and assert_eq!(normalize_status("cancelled"), ("cancelled", None)) would lock in this behavior. (not blocking)

_ => 2,
}
}
Expand Down Expand Up @@ -1030,6 +1043,36 @@ fn status(workspace_id: &str, output: &str, id: &str, poll: &PollArgs) {
std::process::exit(status_exit_code(&st.status));
}

// --- cancel ----------------------------------------------------------------

fn cancel_import(workspace_id: &str, output: &str, id: &str) {
let client = IngestClient::new(workspace_id);
let ack = with_spinner("cancelling…", || client.cancel(id));
let machine = serde_json::json!({
"ingest_id": ack.ingest_id,
"status": ack.status,
"detail": ack.detail,
"database_id": ack.database_id,
});
render(output, &machine, || {
use crossterm::style::Stylize;
let label = |l: &str| format!("{:<14}", l).dark_grey().to_string();
println!("{}{}", label("status:"), util::color_status(&ack.status));

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: util::color_status (src/util.rs) has no arm for cancelled, so it falls through to the yellow "in-flight" color here and in one-shot status. But cancelled is now a terminal state (exit 1) and poll_ingest prints its message in red. Consider adding cancelled to color_status so the status cell doesn't render a settled job in the in-flight color. (not blocking)

if let Some(db) = ack.database_id.as_deref() {
println!("{}{}", label("database:"), db);
}
println!(
"{}",
format!(
"Drain may still be running — track with: hotdata ingest status {} --wait",
ack.ingest_id
)
.dark_grey()
);
});
std::process::exit(1);
}

// --- list-datasources ------------------------------------------------------

fn list_datasources(workspace_id: &str, output: &str, all: bool) {
Expand Down Expand Up @@ -1280,14 +1323,15 @@ fn poll_ingest(
spinner.finish_and_clear();
return st;
}
"failed" => {
"failed" | "cancelled" => {
spinner.finish_and_clear();
// Machine formats get the projected status object (with the
// server's detail) on stdout, matching one-shot `status`.
render(output, &status_json(&st), || {
use crossterm::style::Stylize;
let label = st.status.as_str();
let detail = st.detail.as_deref().unwrap_or("unknown error");
eprintln!("{}", format!("failed: {detail}").red());
eprintln!("{}", format!("{label}: {detail}").red());
if detail.contains("Forbidden") {
eprintln!(
"{}",
Expand Down
Loading