From f7da01d324493ca507b8a2e3171824c81deb543e Mon Sep 17 00:00:00 2001 From: Brandon Shippy Date: Wed, 5 Aug 2026 14:18:41 -0700 Subject: [PATCH 1/6] initial framework down --- .../sift_cli/assets/docs/src/SUMMARY.md | 1 + .../sift_cli/assets/docs/src/data/jobs.md | 59 ++++ .../assets/docs/src/reference/commands.md | 12 + .../assets/skills/sift/references/cli.md | 7 + rust/crates/sift_cli/src/cli/mod.rs | 91 +++++++ rust/crates/sift_cli/src/cmd/import/mod.rs | 101 ++++--- rust/crates/sift_cli/src/cmd/jobs.rs | 255 ++++++++++++++++++ rust/crates/sift_cli/src/cmd/mod.rs | 1 + rust/crates/sift_cli/src/main.rs | 10 + rust/crates/sift_cli/src/util/job.rs | 46 +++- 10 files changed, 527 insertions(+), 56 deletions(-) create mode 100644 rust/crates/sift_cli/assets/docs/src/data/jobs.md create mode 100644 rust/crates/sift_cli/src/cmd/jobs.rs diff --git a/rust/crates/sift_cli/assets/docs/src/SUMMARY.md b/rust/crates/sift_cli/assets/docs/src/SUMMARY.md index db9e4a4d7b..a575fc1006 100644 --- a/rust/crates/sift_cli/assets/docs/src/SUMMARY.md +++ b/rust/crates/sift_cli/assets/docs/src/SUMMARY.md @@ -19,6 +19,7 @@ - [ULog](./data/import-ulog.md) - [Backups](./data/import-backups.md) - [Exporting Data](./data/exporting.md) +- [Jobs](./data/jobs.md) # Reference diff --git a/rust/crates/sift_cli/assets/docs/src/data/jobs.md b/rust/crates/sift_cli/assets/docs/src/data/jobs.md new file mode 100644 index 0000000000..fc61e853b8 --- /dev/null +++ b/rust/crates/sift_cli/assets/docs/src/data/jobs.md @@ -0,0 +1,59 @@ +# Jobs + +Server-side work in Sift — data imports, exports, and rule evaluations — runs +as a job. The `get`, `status`, and `wait` commands inspect and poll those jobs, +which is what makes the CLI usable in CI/CD. + +## Why not just pass `--wait`? + +`sift-cli import ... --wait` blocks until the job finishes. That is the right +default for interactive use, but it serializes work: firing five imports means +five sequential blocking commands. And it does not compose — you cannot start +several imports in parallel, do other work, then confirm they all landed at +the end. + +The verb-first commands split those two concerns: + +1. **Kick off work without waiting.** `sift-cli import ...` (no `--wait`) + uploads the file and prints the assigned job ID. It exits `0` even if the + server-side job later fails. +2. **Poll the jobs later.** `sift-cli wait job [ID ...]` blocks until + every named job reaches a terminal state and exits non-zero if any failed. + +## The CI/CD pattern + +```sh +# Fire imports in parallel; capture each job id. +JOB_A=$(sift-cli import csv a.csv --asset engine | awk '/Job ID/ {print $NF}') +JOB_B=$(sift-cli import csv b.csv --asset engine | awk '/Job ID/ {print $NF}') +JOB_C=$(sift-cli import csv c.csv --asset engine | awk '/Job ID/ {print $NF}') + +# Gate the pipeline on all three finishing successfully. +sift-cli wait job "$JOB_A" "$JOB_B" "$JOB_C" +``` + +The `import` command prints a `Job ID: ` line on the no-wait path +specifically so scripts can capture it. + +## Inspecting jobs + +- `sift-cli get jobs` lists the 50 most recent jobs, newest first. Add + `--job-type` (`data-import`, `data-export`, `rule-evaluation`) or `--status` + (`created`, `running`, `finished`, `failed`, `cancelled`, `cancel-requested`) + to narrow it. `--limit` overrides the page size. +- `sift-cli get job ` prints the full details on one job: type, status, + timestamps, and failure details when the job failed. +- `sift-cli status job ` is the scripting form. It prints one status word + to stdout and exits with a code that reflects the job state. + +### Exit codes for `status job` + +| Code | Meaning | +| ---- | ------------------------------------------ | +| `0` | Job finished successfully. | +| `1` | Job failed. | +| `2` | Job was cancelled or cancel is requested. | +| `3` | Job is still running or has not started. | + +`wait job` uses `0` if every job finished and `1` if any job failed or was +cancelled; per-job status is printed to stderr for the non-success cases. diff --git a/rust/crates/sift_cli/assets/docs/src/reference/commands.md b/rust/crates/sift_cli/assets/docs/src/reference/commands.md index b699588a8b..f009285087 100644 --- a/rust/crates/sift_cli/assets/docs/src/reference/commands.md +++ b/rust/crates/sift_cli/assets/docs/src/reference/commands.md @@ -51,6 +51,18 @@ Export data from Sift. See [Exporting Data](../data/exporting.md). Verify credentials and connectivity. See [Verifying Your Setup](../getting-started/verifying.md). +## `get`, `status`, `wait` + +Inspect and poll Sift resources. See [Jobs](../data/jobs.md) for the CI/CD +pattern (fire imports without `--wait`, then gate on a single `wait job`). + +| Command | Description | +| ------------------------------------------ | ---------------------------------------------------------------------------------- | +| `get jobs [--job-type ...] [--status ...]` | List recent jobs, newest first. | +| `get job ` | Show full details for one job. | +| `status job ` | Print a terse status line. Exit `0` finished, `1` failed, `2` cancelled, `3` running. | +| `wait job [JOB_ID ...]` | Block until every named job reaches a terminal state. | + ## `install` Install optional tooling. diff --git a/rust/crates/sift_cli/assets/skills/sift/references/cli.md b/rust/crates/sift_cli/assets/skills/sift/references/cli.md index 7517e66698..6cbdff8c3f 100644 --- a/rust/crates/sift_cli/assets/skills/sift/references/cli.md +++ b/rust/crates/sift_cli/assets/skills/sift/references/cli.md @@ -8,6 +8,8 @@ Key subcommands: - `import`: `csv`, `parquet flat-dataset`, `parquet cpr`, `tdms`, `hdf5`, `ulog`, `backups`. - `export`: `run`, `asset` (to CSV and other formats). +- `get`, `status`, `wait`: inspect and poll server-side jobs (`get jobs`, + `get job `, `status job `, `wait job [ID ...]`). - `mcp`: start the MCP server. - `ping`: verify credentials and connectivity. - `config`: manage profiles and credentials. @@ -72,6 +74,11 @@ per session. The rest apply to each subcommand invocation. it you cannot confirm the data actually landed. Relay the final stdout line to the user verbatim. `import backups` is the one exception: it accepts no `--wait`, `--preview`, or `--run`. + + Skip `--wait` only when firing multiple imports in parallel. In that + case capture each `Job ID: ` line from the upload output, then + run `sift-cli wait job [ID ...]` to block on the whole batch and + report failures. `wait job` exits non-zero if any job failed. 7. **Surface the Explore link from import output.** Each profile must set `app_uri`. `sift-cli import` prints a `View in Sift: ` tip when this value is usable. Surface the URL as plain text, in full. Do not wrap it in a diff --git a/rust/crates/sift_cli/src/cli/mod.rs b/rust/crates/sift_cli/src/cli/mod.rs index e4653c16d7..b1b6e8ea2b 100644 --- a/rust/crates/sift_cli/src/cli/mod.rs +++ b/rust/crates/sift_cli/src/cli/mod.rs @@ -72,6 +72,97 @@ pub enum Cmd { #[cfg(feature = "mcp")] #[command(hide = true)] Mcp(McpArgs), + + /// Retrieve Sift resources (e.g. jobs) + #[command(subcommand)] + Get(GetCmd), + + /// Print a terse status line for a resource; exit code reflects state + #[command(subcommand)] + Status(StatusCmd), + + /// Block until one or more resources reach a terminal state + #[command(subcommand)] + Wait(WaitCmd), +} + +#[derive(Subcommand)] +pub enum GetCmd { + /// List recent jobs, newest first + Jobs(GetJobsArgs), + + /// Show full details for a single job + Job(GetJobArgs), +} + +#[derive(clap::Args)] +pub struct GetJobsArgs { + /// Filter by job type + #[arg(long, value_enum)] + pub job_type: Option, + + /// Filter by job status + #[arg(long, value_enum)] + pub status: Option, + + /// Max jobs to return (clamped to the server's cap) + #[arg(long, default_value_t = 50)] + pub limit: u32, +} + +#[derive(clap::Args)] +pub struct GetJobArgs { + /// Job ID + pub job_id: String, +} + +#[derive(Subcommand)] +pub enum StatusCmd { + /// Print the current status of a job. Exit code: 0 finished, 1 failed, + /// 2 cancelled or cancel-requested, 3 still running + Job(StatusJobArgs), +} + +#[derive(clap::Args)] +pub struct StatusJobArgs { + /// Job ID + pub job_id: String, +} + +#[derive(Subcommand)] +pub enum WaitCmd { + /// Block until every named job reaches a terminal state. Exit 0 if all + /// finished; nonzero if any failed or was cancelled + Job(WaitJobArgs), +} + +#[derive(clap::Args)] +#[command( + override_usage = "sift-cli wait job [JOB_ID ...]", + after_help = "Example:\n \ + sift-cli wait job " +)] +pub struct WaitJobArgs { + /// One or more job IDs to wait on, separated by spaces + #[arg(required = true, value_name = "JOB_ID")] + pub job_ids: Vec, +} + +#[derive(clap::ValueEnum, Clone, Debug)] +pub enum JobTypeArg { + DataImport, + DataExport, + RuleEvaluation, +} + +#[derive(clap::ValueEnum, Clone, Debug)] +pub enum JobStatusArg { + Created, + Running, + Finished, + Failed, + Cancelled, + CancelRequested, } #[cfg(feature = "mcp")] diff --git a/rust/crates/sift_cli/src/cmd/import/mod.rs b/rust/crates/sift_cli/src/cmd/import/mod.rs index 5b01f7742d..7d84fb1f67 100644 --- a/rust/crates/sift_cli/src/cmd/import/mod.rs +++ b/rust/crates/sift_cli/src/cmd/import/mod.rs @@ -1,7 +1,6 @@ use anyhow::Result; use crossterm::style::Stylize; -use std::{process::ExitCode, time::Duration}; -use tokio::time::sleep; +use std::process::ExitCode; use sift_rs::{SiftChannel, common::r#type::v1::ChannelConfig, jobs::v1::JobStatus}; @@ -40,6 +39,7 @@ pub async fn finish_import( if !wait { Output::new() .line(format!("{} file for processing", "Uploaded".green())) + .line(format!("{}: {job_id}", "Job ID".green())) .tip(pending_import_tip( &target.location, target.explore_url.as_deref(), @@ -60,11 +60,26 @@ pub async fn wait_for_job_completion( let spinner = Spinner::new(); spinner.set_message(format!("{} file for processing", "Uploaded".green())); - let mut job_service = JobServiceWrapper::new(grpc_channel.clone()); + let mut job_service = JobServiceWrapper::new(grpc_channel); - let Some(mut job) = job_service.get_job(&job_id).await? else { - spinner.finish_and_clear(); + let outcome = job_service + .poll_until_terminal(&job_id, |job| match job.job_status() { + JobStatus::Running => { + spinner.set_message(format!("{} imported file", "Processing".green())); + } + JobStatus::CancelRequested => { + spinner.set_message(format!( + "{} was requested but the job may still finish", + "Cancellation".green() + )); + } + _ => (), + }) + .await?; + + spinner.finish_and_clear(); + let Some(job) = outcome else { Output::new() .line("The file was successfully uploaded but the job was unexpectedly not found") .tip("Please notify Sift about this bug") @@ -72,60 +87,38 @@ pub async fn wait_for_job_completion( return Ok(ExitCode::FAILURE); }; - loop { - sleep(Duration::from_secs(3)).await; - - let Some(updated_job) = job_service.get_job(&job.job_id).await? else { - spinner.finish_and_clear(); + match job.job_status() { + JobStatus::Finished => { + let mut tip_text = + format!("The data should be available on the {import_output_location}"); + tip_text.push_str(&explore_or_note(explore_url.as_deref())); + Output::new() + .line(format!("{} data import job", "Completed".green())) + .tip(tip_text) + .print(); + Ok(ExitCode::SUCCESS) + } + JobStatus::Cancelled => { + Output::new() + .line(format!("{} data import job", "Cancelled".green())) + .print(); + Ok(ExitCode::SUCCESS) + } + JobStatus::Failed => { Output::new() - .line("The file was successfully uploaded but the job was unexpectedly not found") + .line("Processing failed") + .tip("Please check the Sift jobs manage page for further details") + .eprint(); + Ok(ExitCode::FAILURE) + } + other => { + Output::new() + .line(format!("unexpected job status `{other:?}`")) .tip("Please notify Sift about this bug") .eprint(); - return Ok(ExitCode::FAILURE); - }; - job = updated_job; - - match job.job_status() { - JobStatus::Created => (), - JobStatus::Running => { - spinner.set_message(format!("{} imported file", "Processing".green())); - } - JobStatus::CancelRequested => { - spinner.set_message(format!( - "{} was requested but the job may still finish", - "Cancellation".green() - )); - } - JobStatus::Cancelled => { - spinner.finish_and_clear(); - Output::new() - .line(format!("{} data import job", "Cancelled".green())) - .print(); - break; - } - JobStatus::Failed => { - spinner.finish_and_clear(); - Output::new() - .line("Processing failed") - .tip("Please check the Sift jobs manage page for further details") - .eprint(); - return Ok(ExitCode::FAILURE); - } - JobStatus::Finished => { - spinner.finish_and_clear(); - let mut tip_text = - format!("The data should be available on the {import_output_location}"); - tip_text.push_str(&explore_or_note(explore_url.as_deref())); - Output::new() - .line(format!("{} data import job", "Completed".green())) - .tip(tip_text) - .print(); - break; - } - _ => (), + Ok(ExitCode::FAILURE) } } - Ok(ExitCode::SUCCESS) } pub struct TimePreview<'a> { diff --git a/rust/crates/sift_cli/src/cmd/jobs.rs b/rust/crates/sift_cli/src/cmd/jobs.rs new file mode 100644 index 0000000000..34238629e1 --- /dev/null +++ b/rust/crates/sift_cli/src/cmd/jobs.rs @@ -0,0 +1,255 @@ +use std::process::ExitCode; + +use anyhow::{Context as AnyhowContext, Result}; +use chrono::{DateTime, SecondsFormat, Utc}; +use crossterm::style::Stylize; +use pbjson_types::Timestamp; +use sift_rs::jobs::v1::{Job, JobStatus, JobType, ListJobsRequest}; + +use crate::cli::{ + GetJobArgs, GetJobsArgs, JobStatusArg, JobTypeArg, StatusJobArgs, WaitJobArgs, +}; +use crate::util::{api::create_grpc_channel, job::JobServiceWrapper, tty::Output}; + +use super::Context; + +const EXIT_JOB_FAILED: u8 = 1; +const EXIT_JOB_CANCELLED: u8 = 2; +const EXIT_JOB_RUNNING: u8 = 3; + +pub async fn get_jobs(ctx: Context, args: GetJobsArgs) -> Result { + let grpc_channel = create_grpc_channel(&ctx)?; + let mut job_service = JobServiceWrapper::new(grpc_channel); + + let filter = build_list_filter(args.job_type.as_ref(), args.status.as_ref()); + let jobs = job_service + .list_jobs(ListJobsRequest { + page_size: args.limit, + filter, + order_by: "created_date desc".to_string(), + ..Default::default() + }) + .await + .context("failed to list jobs")? + .into_inner() + .jobs; + + if jobs.is_empty() { + Output::new().line("No jobs matched").print(); + return Ok(ExitCode::SUCCESS); + } + + print_jobs_table(&jobs); + Ok(ExitCode::SUCCESS) +} + +pub async fn get_job(ctx: Context, args: GetJobArgs) -> Result { + let grpc_channel = create_grpc_channel(&ctx)?; + let mut job_service = JobServiceWrapper::new(grpc_channel); + + let Some(job) = job_service.get_job(&args.job_id).await? else { + Output::new() + .line(format!("Job `{}` not found", args.job_id)) + .eprint(); + return Ok(ExitCode::FAILURE); + }; + + print_job_details(&job); + Ok(ExitCode::SUCCESS) +} + +pub async fn status_job(ctx: Context, args: StatusJobArgs) -> Result { + let grpc_channel = create_grpc_channel(&ctx)?; + let mut job_service = JobServiceWrapper::new(grpc_channel); + + let Some(job) = job_service.get_job(&args.job_id).await? else { + Output::new() + .line(format!("Job `{}` not found", args.job_id)) + .eprint(); + return Ok(ExitCode::FAILURE); + }; + + let status = job.job_status(); + println!("{}", status_label(status)); + Ok(exit_code_for_status(status)) +} + +pub async fn wait_job(ctx: Context, args: WaitJobArgs) -> Result { + let grpc_channel = create_grpc_channel(&ctx)?; + + let mut handles = Vec::with_capacity(args.job_ids.len()); + for job_id in args.job_ids { + let channel = grpc_channel.clone(); + handles.push(tokio::spawn(async move { + let mut svc = JobServiceWrapper::new(channel); + let outcome = svc.poll_until_terminal(&job_id, |_| {}).await; + (job_id, outcome) + })); + } + + let mut any_failed = false; + for handle in handles { + let (job_id, outcome) = handle.await.context("wait task panicked")?; + match outcome { + Ok(Some(job)) => { + let status = job.job_status(); + let line = format!("{}: {}", job_id.cyan(), status_label(status)); + match status { + JobStatus::Finished => Output::new().line(line).print(), + JobStatus::Cancelled => Output::new().line(line).print(), + _ => { + any_failed = true; + Output::new().line(line).eprint(); + } + } + } + Ok(None) => { + any_failed = true; + Output::new() + .line(format!("Job `{job_id}` disappeared before it finished")) + .eprint(); + } + Err(err) => { + any_failed = true; + Output::new() + .line(format!("Job `{job_id}` failed to poll: {err:#}")) + .eprint(); + } + } + } + + if any_failed { + Ok(ExitCode::FAILURE) + } else { + Ok(ExitCode::SUCCESS) + } +} + +fn build_list_filter( + job_type: Option<&JobTypeArg>, + status: Option<&JobStatusArg>, +) -> String { + let mut parts: Vec = Vec::new(); + if let Some(t) = job_type { + parts.push(format!("job_type == \"{}\"", job_type_cel(t))); + } + if let Some(s) = status { + parts.push(format!("job_status == \"{}\"", job_status_cel(s))); + } + parts.join(" && ") +} + +fn job_type_cel(t: &JobTypeArg) -> &'static str { + match t { + JobTypeArg::DataImport => "JOB_TYPE_DATA_IMPORT", + JobTypeArg::DataExport => "JOB_TYPE_DATA_EXPORT", + JobTypeArg::RuleEvaluation => "JOB_TYPE_RULE_EVALUATION", + } +} + +fn job_status_cel(s: &JobStatusArg) -> &'static str { + match s { + JobStatusArg::Created => "JOB_STATUS_CREATED", + JobStatusArg::Running => "JOB_STATUS_RUNNING", + JobStatusArg::Finished => "JOB_STATUS_FINISHED", + JobStatusArg::Failed => "JOB_STATUS_FAILED", + JobStatusArg::Cancelled => "JOB_STATUS_CANCELLED", + JobStatusArg::CancelRequested => "JOB_STATUS_CANCEL_REQUESTED", + } +} + +fn status_label(status: JobStatus) -> String { + match status { + JobStatus::Created => "created".to_string(), + JobStatus::Running => "running".to_string(), + JobStatus::Finished => "finished".to_string(), + JobStatus::Failed => "failed".to_string(), + JobStatus::Cancelled => "cancelled".to_string(), + JobStatus::CancelRequested => "cancel-requested".to_string(), + other => format!("{other:?}"), + } +} + +fn type_label(job_type: JobType) -> &'static str { + match job_type { + JobType::DataImport => "data-import", + JobType::DataExport => "data-export", + JobType::RuleEvaluation => "rule-evaluation", + JobType::Unspecified => "unspecified", + } +} + +fn exit_code_for_status(status: JobStatus) -> ExitCode { + match status { + JobStatus::Finished => ExitCode::SUCCESS, + JobStatus::Failed => ExitCode::from(EXIT_JOB_FAILED), + JobStatus::Cancelled | JobStatus::CancelRequested => ExitCode::from(EXIT_JOB_CANCELLED), + JobStatus::Created | JobStatus::Running => ExitCode::from(EXIT_JOB_RUNNING), + _ => ExitCode::from(EXIT_JOB_RUNNING), + } +} + +fn print_jobs_table(jobs: &[Job]) { + let mut out = Output::new(); + out.line(format!( + "{:<38} {:<15} {:<17} {}", + "JOB ID", "TYPE", "STATUS", "CREATED", + )); + for job in jobs { + out.line(format!( + "{:<38} {:<15} {:<17} {}", + job.job_id, + type_label(job.job_type()), + status_label(job.job_status()), + format_timestamp(job.created_date.as_ref()), + )); + } + out.print(); +} + +fn format_timestamp(ts: Option<&Timestamp>) -> String { + let Some(ts) = ts else { + return "-".to_string(); + }; + DateTime::::from_timestamp(ts.seconds, ts.nanos as u32) + .map(|dt| dt.to_rfc3339_opts(SecondsFormat::Secs, true)) + .unwrap_or_else(|| "-".to_string()) +} + +fn print_job_details(job: &Job) { + let mut out = Output::new(); + out.line(format!("{}: {}", "Job ID".green(), job.job_id)); + out.line(format!("{}: {}", "Type".green(), type_label(job.job_type()))); + out.line(format!( + "{}: {}", + "Status".green(), + status_label(job.job_status()) + )); + if job.created_date.is_some() { + out.line(format!( + "{}: {}", + "Created".green(), + format_timestamp(job.created_date.as_ref()) + )); + } + if job.started_date.is_some() { + out.line(format!( + "{}: {}", + "Started".green(), + format_timestamp(job.started_date.as_ref()) + )); + } + if job.completed_date.is_some() { + out.line(format!( + "{}: {}", + "Completed".green(), + format_timestamp(job.completed_date.as_ref()) + )); + } + if job.job_status() == JobStatus::Failed { + if let Some(details) = job.job_status_details.as_ref() { + out.line(format!("{}: {:?}", "Failure details".red(), details)); + } + } + out.print(); +} diff --git a/rust/crates/sift_cli/src/cmd/mod.rs b/rust/crates/sift_cli/src/cmd/mod.rs index 79dcb375a4..e6e70a73d6 100644 --- a/rust/crates/sift_cli/src/cmd/mod.rs +++ b/rust/crates/sift_cli/src/cmd/mod.rs @@ -14,6 +14,7 @@ pub mod doc; pub mod export; pub mod import; pub mod install; +pub mod jobs; #[cfg(feature = "mcp")] pub mod mcp; pub mod ping; diff --git a/rust/crates/sift_cli/src/main.rs b/rust/crates/sift_cli/src/main.rs index 2b6503386d..01c6c005ba 100644 --- a/rust/crates/sift_cli/src/main.rs +++ b/rust/crates/sift_cli/src/main.rs @@ -159,6 +159,16 @@ fn run(clargs: cli::Args) -> Result { cli::ExportCmd::Asset(args) => run_future(cmd::export::asset(ctx, args)), }, Cmd::Ping => run_future(cmd::ping::run(ctx)), + Cmd::Get(cmd) => match cmd { + cli::GetCmd::Jobs(args) => run_future(cmd::jobs::get_jobs(ctx, args)), + cli::GetCmd::Job(args) => run_future(cmd::jobs::get_job(ctx, args)), + }, + Cmd::Status(cmd) => match cmd { + cli::StatusCmd::Job(args) => run_future(cmd::jobs::status_job(ctx, args)), + }, + Cmd::Wait(cmd) => match cmd { + cli::WaitCmd::Job(args) => run_future_mt(cmd::jobs::wait_job(ctx, args)), + }, _ => Ok(ExitCode::SUCCESS), } } diff --git a/rust/crates/sift_cli/src/util/job.rs b/rust/crates/sift_cli/src/util/job.rs index 0bb8ff5815..716fce8515 100644 --- a/rust/crates/sift_cli/src/util/job.rs +++ b/rust/crates/sift_cli/src/util/job.rs @@ -1,10 +1,23 @@ -use std::ops::{Deref, DerefMut}; +use std::{ + ops::{Deref, DerefMut}, + time::Duration, +}; use anyhow::{Context, Result}; use sift_rs::{ SiftChannel, - jobs::v1::{Job, ListJobsRequest, job_service_client::JobServiceClient}, + jobs::v1::{Job, JobStatus, ListJobsRequest, job_service_client::JobServiceClient}, }; +use tokio::time::sleep; + +const POLL_INTERVAL: Duration = Duration::from_secs(3); + +pub fn is_terminal_status(status: JobStatus) -> bool { + matches!( + status, + JobStatus::Finished | JobStatus::Failed | JobStatus::Cancelled + ) +} pub struct JobServiceWrapper(JobServiceClient); @@ -41,4 +54,33 @@ impl JobServiceWrapper { Ok(res.jobs.first().cloned()) } + + pub async fn poll_until_terminal( + &mut self, + job_id: &str, + mut on_status_change: F, + ) -> Result> + where + F: FnMut(&Job), + { + let Some(mut job) = self.get_job(job_id).await? else { + return Ok(None); + }; + on_status_change(&job); + let mut last_status = job.job_status(); + + while !is_terminal_status(last_status) { + sleep(POLL_INTERVAL).await; + let Some(updated) = self.get_job(job_id).await? else { + return Ok(None); + }; + if updated.job_status() != last_status { + last_status = updated.job_status(); + on_status_change(&updated); + } + job = updated; + } + + Ok(Some(job)) + } } From 7b0c96389bfdd562fa6a4d833adbb9b6b8332f9a Mon Sep 17 00:00:00 2001 From: Brandon Shippy Date: Wed, 5 Aug 2026 14:27:04 -0700 Subject: [PATCH 2/6] cargo fmt --- rust/crates/sift_cli/src/cmd/jobs.rs | 15 +++++++-------- 1 file changed, 7 insertions(+), 8 deletions(-) diff --git a/rust/crates/sift_cli/src/cmd/jobs.rs b/rust/crates/sift_cli/src/cmd/jobs.rs index 34238629e1..c9aad5116f 100644 --- a/rust/crates/sift_cli/src/cmd/jobs.rs +++ b/rust/crates/sift_cli/src/cmd/jobs.rs @@ -6,9 +6,7 @@ use crossterm::style::Stylize; use pbjson_types::Timestamp; use sift_rs::jobs::v1::{Job, JobStatus, JobType, ListJobsRequest}; -use crate::cli::{ - GetJobArgs, GetJobsArgs, JobStatusArg, JobTypeArg, StatusJobArgs, WaitJobArgs, -}; +use crate::cli::{GetJobArgs, GetJobsArgs, JobStatusArg, JobTypeArg, StatusJobArgs, WaitJobArgs}; use crate::util::{api::create_grpc_channel, job::JobServiceWrapper, tty::Output}; use super::Context; @@ -125,10 +123,7 @@ pub async fn wait_job(ctx: Context, args: WaitJobArgs) -> Result { } } -fn build_list_filter( - job_type: Option<&JobTypeArg>, - status: Option<&JobStatusArg>, -) -> String { +fn build_list_filter(job_type: Option<&JobTypeArg>, status: Option<&JobStatusArg>) -> String { let mut parts: Vec = Vec::new(); if let Some(t) = job_type { parts.push(format!("job_type == \"{}\"", job_type_cel(t))); @@ -219,7 +214,11 @@ fn format_timestamp(ts: Option<&Timestamp>) -> String { fn print_job_details(job: &Job) { let mut out = Output::new(); out.line(format!("{}: {}", "Job ID".green(), job.job_id)); - out.line(format!("{}: {}", "Type".green(), type_label(job.job_type()))); + out.line(format!( + "{}: {}", + "Type".green(), + type_label(job.job_type()) + )); out.line(format!( "{}: {}", "Status".green(), From 05534d06322a04e6a268213a2dfdbe6af53af142 Mon Sep 17 00:00:00 2001 From: Brandon Shippy Date: Wed, 5 Aug 2026 14:56:01 -0700 Subject: [PATCH 3/6] tests --- rust/crates/sift_cli/src/cmd/jobs.rs | 85 ++++++++++++++++++++++++++++ rust/crates/sift_cli/src/util/job.rs | 19 +++++++ 2 files changed, 104 insertions(+) diff --git a/rust/crates/sift_cli/src/cmd/jobs.rs b/rust/crates/sift_cli/src/cmd/jobs.rs index c9aad5116f..3dfec7988b 100644 --- a/rust/crates/sift_cli/src/cmd/jobs.rs +++ b/rust/crates/sift_cli/src/cmd/jobs.rs @@ -252,3 +252,88 @@ fn print_job_details(job: &Job) { } out.print(); } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn build_list_filter_produces_empty_string_when_nothing_selected() { + assert_eq!(build_list_filter(None, None), ""); + } + + #[test] + fn build_list_filter_produces_type_only() { + assert_eq!( + build_list_filter(Some(&JobTypeArg::DataImport), None), + "job_type == \"JOB_TYPE_DATA_IMPORT\"" + ); + } + + #[test] + fn build_list_filter_produces_status_only() { + assert_eq!( + build_list_filter(None, Some(&JobStatusArg::Failed)), + "job_status == \"JOB_STATUS_FAILED\"" + ); + } + + #[test] + fn build_list_filter_ands_type_and_status() { + assert_eq!( + build_list_filter( + Some(&JobTypeArg::RuleEvaluation), + Some(&JobStatusArg::Running) + ), + "job_type == \"JOB_TYPE_RULE_EVALUATION\" && job_status == \"JOB_STATUS_RUNNING\"" + ); + } + + #[test] + fn exit_code_for_status_matches_documented_semantics() { + let cases = [ + (JobStatus::Finished, ExitCode::SUCCESS), + (JobStatus::Failed, ExitCode::from(EXIT_JOB_FAILED)), + (JobStatus::Cancelled, ExitCode::from(EXIT_JOB_CANCELLED)), + (JobStatus::CancelRequested, ExitCode::from(EXIT_JOB_CANCELLED)), + (JobStatus::Created, ExitCode::from(EXIT_JOB_RUNNING)), + (JobStatus::Running, ExitCode::from(EXIT_JOB_RUNNING)), + ]; + for (status, expected) in cases { + assert_eq!( + format!("{:?}", exit_code_for_status(status)), + format!("{expected:?}"), + "status: {status:?}" + ); + } + } + + #[test] + fn status_label_covers_every_variant() { + assert_eq!(status_label(JobStatus::Created), "created"); + assert_eq!(status_label(JobStatus::Running), "running"); + assert_eq!(status_label(JobStatus::Finished), "finished"); + assert_eq!(status_label(JobStatus::Failed), "failed"); + assert_eq!(status_label(JobStatus::Cancelled), "cancelled"); + assert_eq!(status_label(JobStatus::CancelRequested), "cancel-requested"); + } + + #[test] + fn type_label_covers_every_variant() { + assert_eq!(type_label(JobType::DataImport), "data-import"); + assert_eq!(type_label(JobType::DataExport), "data-export"); + assert_eq!(type_label(JobType::RuleEvaluation), "rule-evaluation"); + assert_eq!(type_label(JobType::Unspecified), "unspecified"); + } + + #[test] + fn format_timestamp_handles_missing_and_present() { + assert_eq!(format_timestamp(None), "-"); + + let ts = Timestamp { + seconds: 1_767_225_600, + nanos: 0, + }; + assert_eq!(format_timestamp(Some(&ts)), "2026-01-01T00:00:00Z"); + } +} diff --git a/rust/crates/sift_cli/src/util/job.rs b/rust/crates/sift_cli/src/util/job.rs index 716fce8515..0b531b273a 100644 --- a/rust/crates/sift_cli/src/util/job.rs +++ b/rust/crates/sift_cli/src/util/job.rs @@ -84,3 +84,22 @@ impl JobServiceWrapper { Ok(Some(job)) } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn terminal_statuses_stop_polling() { + assert!(is_terminal_status(JobStatus::Finished)); + assert!(is_terminal_status(JobStatus::Failed)); + assert!(is_terminal_status(JobStatus::Cancelled)); + } + + #[test] + fn non_terminal_statuses_keep_polling() { + assert!(!is_terminal_status(JobStatus::Created)); + assert!(!is_terminal_status(JobStatus::Running)); + assert!(!is_terminal_status(JobStatus::CancelRequested)); + } +} From 2f2dabf46912bb6726bdfd2e9c74908d1e7bb7f1 Mon Sep 17 00:00:00 2001 From: Brandon Shippy Date: Wed, 5 Aug 2026 15:00:52 -0700 Subject: [PATCH 4/6] cargo fmt --- rust/crates/sift_cli/src/cmd/jobs.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/rust/crates/sift_cli/src/cmd/jobs.rs b/rust/crates/sift_cli/src/cmd/jobs.rs index 3dfec7988b..92a14a726a 100644 --- a/rust/crates/sift_cli/src/cmd/jobs.rs +++ b/rust/crates/sift_cli/src/cmd/jobs.rs @@ -295,7 +295,10 @@ mod tests { (JobStatus::Finished, ExitCode::SUCCESS), (JobStatus::Failed, ExitCode::from(EXIT_JOB_FAILED)), (JobStatus::Cancelled, ExitCode::from(EXIT_JOB_CANCELLED)), - (JobStatus::CancelRequested, ExitCode::from(EXIT_JOB_CANCELLED)), + ( + JobStatus::CancelRequested, + ExitCode::from(EXIT_JOB_CANCELLED), + ), (JobStatus::Created, ExitCode::from(EXIT_JOB_RUNNING)), (JobStatus::Running, ExitCode::from(EXIT_JOB_RUNNING)), ]; From 95bb79458e8729919f0586dfec702ad2a02dfb2f Mon Sep 17 00:00:00 2001 From: Brandon Shippy Date: Wed, 5 Aug 2026 15:23:17 -0700 Subject: [PATCH 5/6] move tests to their own file --- .../sift_cli/src/cmd/{jobs.rs => jobs/mod.rs} | 87 +------------------ rust/crates/sift_cli/src/cmd/jobs/tests.rs | 84 ++++++++++++++++++ .../sift_cli/src/util/{job.rs => job/mod.rs} | 18 +--- rust/crates/sift_cli/src/util/job/tests.rs | 15 ++++ 4 files changed, 101 insertions(+), 103 deletions(-) rename rust/crates/sift_cli/src/cmd/{jobs.rs => jobs/mod.rs} (74%) create mode 100644 rust/crates/sift_cli/src/cmd/jobs/tests.rs rename rust/crates/sift_cli/src/util/{job.rs => job/mod.rs} (81%) create mode 100644 rust/crates/sift_cli/src/util/job/tests.rs diff --git a/rust/crates/sift_cli/src/cmd/jobs.rs b/rust/crates/sift_cli/src/cmd/jobs/mod.rs similarity index 74% rename from rust/crates/sift_cli/src/cmd/jobs.rs rename to rust/crates/sift_cli/src/cmd/jobs/mod.rs index 92a14a726a..4e5bbcae80 100644 --- a/rust/crates/sift_cli/src/cmd/jobs.rs +++ b/rust/crates/sift_cli/src/cmd/jobs/mod.rs @@ -254,89 +254,4 @@ fn print_job_details(job: &Job) { } #[cfg(test)] -mod tests { - use super::*; - - #[test] - fn build_list_filter_produces_empty_string_when_nothing_selected() { - assert_eq!(build_list_filter(None, None), ""); - } - - #[test] - fn build_list_filter_produces_type_only() { - assert_eq!( - build_list_filter(Some(&JobTypeArg::DataImport), None), - "job_type == \"JOB_TYPE_DATA_IMPORT\"" - ); - } - - #[test] - fn build_list_filter_produces_status_only() { - assert_eq!( - build_list_filter(None, Some(&JobStatusArg::Failed)), - "job_status == \"JOB_STATUS_FAILED\"" - ); - } - - #[test] - fn build_list_filter_ands_type_and_status() { - assert_eq!( - build_list_filter( - Some(&JobTypeArg::RuleEvaluation), - Some(&JobStatusArg::Running) - ), - "job_type == \"JOB_TYPE_RULE_EVALUATION\" && job_status == \"JOB_STATUS_RUNNING\"" - ); - } - - #[test] - fn exit_code_for_status_matches_documented_semantics() { - let cases = [ - (JobStatus::Finished, ExitCode::SUCCESS), - (JobStatus::Failed, ExitCode::from(EXIT_JOB_FAILED)), - (JobStatus::Cancelled, ExitCode::from(EXIT_JOB_CANCELLED)), - ( - JobStatus::CancelRequested, - ExitCode::from(EXIT_JOB_CANCELLED), - ), - (JobStatus::Created, ExitCode::from(EXIT_JOB_RUNNING)), - (JobStatus::Running, ExitCode::from(EXIT_JOB_RUNNING)), - ]; - for (status, expected) in cases { - assert_eq!( - format!("{:?}", exit_code_for_status(status)), - format!("{expected:?}"), - "status: {status:?}" - ); - } - } - - #[test] - fn status_label_covers_every_variant() { - assert_eq!(status_label(JobStatus::Created), "created"); - assert_eq!(status_label(JobStatus::Running), "running"); - assert_eq!(status_label(JobStatus::Finished), "finished"); - assert_eq!(status_label(JobStatus::Failed), "failed"); - assert_eq!(status_label(JobStatus::Cancelled), "cancelled"); - assert_eq!(status_label(JobStatus::CancelRequested), "cancel-requested"); - } - - #[test] - fn type_label_covers_every_variant() { - assert_eq!(type_label(JobType::DataImport), "data-import"); - assert_eq!(type_label(JobType::DataExport), "data-export"); - assert_eq!(type_label(JobType::RuleEvaluation), "rule-evaluation"); - assert_eq!(type_label(JobType::Unspecified), "unspecified"); - } - - #[test] - fn format_timestamp_handles_missing_and_present() { - assert_eq!(format_timestamp(None), "-"); - - let ts = Timestamp { - seconds: 1_767_225_600, - nanos: 0, - }; - assert_eq!(format_timestamp(Some(&ts)), "2026-01-01T00:00:00Z"); - } -} +mod tests; diff --git a/rust/crates/sift_cli/src/cmd/jobs/tests.rs b/rust/crates/sift_cli/src/cmd/jobs/tests.rs new file mode 100644 index 0000000000..a49fdb8f33 --- /dev/null +++ b/rust/crates/sift_cli/src/cmd/jobs/tests.rs @@ -0,0 +1,84 @@ +use super::*; + +#[test] +fn build_list_filter_produces_empty_string_when_nothing_selected() { + assert_eq!(build_list_filter(None, None), ""); +} + +#[test] +fn build_list_filter_produces_type_only() { + assert_eq!( + build_list_filter(Some(&JobTypeArg::DataImport), None), + "job_type == \"JOB_TYPE_DATA_IMPORT\"" + ); +} + +#[test] +fn build_list_filter_produces_status_only() { + assert_eq!( + build_list_filter(None, Some(&JobStatusArg::Failed)), + "job_status == \"JOB_STATUS_FAILED\"" + ); +} + +#[test] +fn build_list_filter_ands_type_and_status() { + assert_eq!( + build_list_filter( + Some(&JobTypeArg::RuleEvaluation), + Some(&JobStatusArg::Running) + ), + "job_type == \"JOB_TYPE_RULE_EVALUATION\" && job_status == \"JOB_STATUS_RUNNING\"" + ); +} + +#[test] +fn exit_code_for_status_matches_documented_semantics() { + let cases = [ + (JobStatus::Finished, ExitCode::SUCCESS), + (JobStatus::Failed, ExitCode::from(EXIT_JOB_FAILED)), + (JobStatus::Cancelled, ExitCode::from(EXIT_JOB_CANCELLED)), + ( + JobStatus::CancelRequested, + ExitCode::from(EXIT_JOB_CANCELLED), + ), + (JobStatus::Created, ExitCode::from(EXIT_JOB_RUNNING)), + (JobStatus::Running, ExitCode::from(EXIT_JOB_RUNNING)), + ]; + for (status, expected) in cases { + assert_eq!( + format!("{:?}", exit_code_for_status(status)), + format!("{expected:?}"), + "status: {status:?}" + ); + } +} + +#[test] +fn status_label_covers_every_variant() { + assert_eq!(status_label(JobStatus::Created), "created"); + assert_eq!(status_label(JobStatus::Running), "running"); + assert_eq!(status_label(JobStatus::Finished), "finished"); + assert_eq!(status_label(JobStatus::Failed), "failed"); + assert_eq!(status_label(JobStatus::Cancelled), "cancelled"); + assert_eq!(status_label(JobStatus::CancelRequested), "cancel-requested"); +} + +#[test] +fn type_label_covers_every_variant() { + assert_eq!(type_label(JobType::DataImport), "data-import"); + assert_eq!(type_label(JobType::DataExport), "data-export"); + assert_eq!(type_label(JobType::RuleEvaluation), "rule-evaluation"); + assert_eq!(type_label(JobType::Unspecified), "unspecified"); +} + +#[test] +fn format_timestamp_handles_missing_and_present() { + assert_eq!(format_timestamp(None), "-"); + + let ts = Timestamp { + seconds: 1_767_225_600, + nanos: 0, + }; + assert_eq!(format_timestamp(Some(&ts)), "2026-01-01T00:00:00Z"); +} diff --git a/rust/crates/sift_cli/src/util/job.rs b/rust/crates/sift_cli/src/util/job/mod.rs similarity index 81% rename from rust/crates/sift_cli/src/util/job.rs rename to rust/crates/sift_cli/src/util/job/mod.rs index 0b531b273a..96fa29fef0 100644 --- a/rust/crates/sift_cli/src/util/job.rs +++ b/rust/crates/sift_cli/src/util/job/mod.rs @@ -86,20 +86,4 @@ impl JobServiceWrapper { } #[cfg(test)] -mod tests { - use super::*; - - #[test] - fn terminal_statuses_stop_polling() { - assert!(is_terminal_status(JobStatus::Finished)); - assert!(is_terminal_status(JobStatus::Failed)); - assert!(is_terminal_status(JobStatus::Cancelled)); - } - - #[test] - fn non_terminal_statuses_keep_polling() { - assert!(!is_terminal_status(JobStatus::Created)); - assert!(!is_terminal_status(JobStatus::Running)); - assert!(!is_terminal_status(JobStatus::CancelRequested)); - } -} +mod tests; diff --git a/rust/crates/sift_cli/src/util/job/tests.rs b/rust/crates/sift_cli/src/util/job/tests.rs new file mode 100644 index 0000000000..6cda4988c2 --- /dev/null +++ b/rust/crates/sift_cli/src/util/job/tests.rs @@ -0,0 +1,15 @@ +use super::*; + +#[test] +fn terminal_statuses_stop_polling() { + assert!(is_terminal_status(JobStatus::Finished)); + assert!(is_terminal_status(JobStatus::Failed)); + assert!(is_terminal_status(JobStatus::Cancelled)); +} + +#[test] +fn non_terminal_statuses_keep_polling() { + assert!(!is_terminal_status(JobStatus::Created)); + assert!(!is_terminal_status(JobStatus::Running)); + assert!(!is_terminal_status(JobStatus::CancelRequested)); +} From 1d2970b95585b3e88e38943827f2f09951fae0f5 Mon Sep 17 00:00:00 2001 From: Brandon Shippy Date: Wed, 5 Aug 2026 15:57:48 -0700 Subject: [PATCH 6/6] fixed contradiction --- rust/crates/sift_cli/src/cmd/jobs/mod.rs | 12 +++++------- 1 file changed, 5 insertions(+), 7 deletions(-) diff --git a/rust/crates/sift_cli/src/cmd/jobs/mod.rs b/rust/crates/sift_cli/src/cmd/jobs/mod.rs index 4e5bbcae80..3259c570a5 100644 --- a/rust/crates/sift_cli/src/cmd/jobs/mod.rs +++ b/rust/crates/sift_cli/src/cmd/jobs/mod.rs @@ -92,13 +92,11 @@ pub async fn wait_job(ctx: Context, args: WaitJobArgs) -> Result { Ok(Some(job)) => { let status = job.job_status(); let line = format!("{}: {}", job_id.cyan(), status_label(status)); - match status { - JobStatus::Finished => Output::new().line(line).print(), - JobStatus::Cancelled => Output::new().line(line).print(), - _ => { - any_failed = true; - Output::new().line(line).eprint(); - } + if status == JobStatus::Finished { + Output::new().line(line).print(); + } else { + any_failed = true; + Output::new().line(line).eprint(); } } Ok(None) => {