diff --git a/README.md b/README.md index 67f3f41..b99f541 100644 --- a/README.md +++ b/README.md @@ -97,6 +97,23 @@ dw workflow:history dw service-call:describe --json ``` +## Version branch usage + +Before retiring a version branch, inspect the open runs for its workflow type +and change ID: + +```bash +dw workflow:version-usage shipment carrier-api --min-version=2 --namespace=orders +dw workflow:version-usage shipment carrier-api --min-version=2 --state=unknown --json +``` + +This requires a Server advertising `workflow_version_usage`. Counts include all +legacy, current and unknown runs before filtering and paging. A missing marker +or unavailable history stays unknown. The command reads one live page; pass its +opaque `next_cursor` with the same query to continue. JSON and JSONL preserve the +whole page, including counts. Drain or fence admissions and restart the query +before making a branch-removal decision. + ## Automation Contract All read and mutating commands support machine-readable output. Use diff --git a/rust/parity/catalog-additions.json b/rust/parity/catalog-additions.json index 9f1ee77..c0ebd29 100644 --- a/rust/parity/catalog-additions.json +++ b/rust/parity/catalog-additions.json @@ -1709,6 +1709,206 @@ } }, "rust_only": true + }, + { + "name": "workflow:version-usage", + "description": "Show live legacy, current and unknown version usage across open runs", + "synopsis": "workflow:version-usage [-s|--server [SERVER]] [--namespace NAMESPACE] [--token [TOKEN]] [--tls-verify TLS-VERIFY] [--env ENV] [--output OUTPUT] [--min-version MIN-VERSION] [--state STATE] [--limit LIMIT] [--cursor CURSOR] [--json] [--] ", + "aliases": [], + "hidden": false, + "help": "Read one live, namespace-scoped page of open runs for a workflow type and\nchange ID. --min-version is required: decisions below that proposed minimum\nare legacy; decisions at or above it are current. Use 1 for a patched() branch.\nNo marker, pruned/incomplete history or invalid decisions remain unknown.\nUnknown is never zero usage or a current decision.\n\nCounts cover all states before --state and paging. --cursor is opaque and\nbound to the exact namespace/type/change/minimum/state query. The first page\nsets an upper run-ID bound. New admissions beyond that bound stay excluded;\nclosures and new decisions can still change counts/state between pages.\nRestart after draining or fencing admissions before retiring a branch.\n\nRequires literal capabilities.workflow_version_usage=true from the Server.\nWithout it, dw exits 8 before the version-usage request. Unsupported routes,\nstorage failures and invalid/inconsistent responses are errors, never empty\nsuccessful counts. This command never runs workflow code or changes markers.\n\n--limit is 1..200 (default 50). --state is legacy, current or unknown.\nJSON and JSONL retain one complete page, including counts and continuation;\nJSONL does not discard the envelope when no runs match the selected state.\n\nExamples:\n\n dw workflow:version-usage shipment carrier-api --min-version=2 --namespace=orders\n dw workflow:version-usage shipment carrier-api --min-version=2 --state=unknown --json\n dw workflow:version-usage shipment carrier-api --min-version=2 --limit=20 --cursor=OPAQUE --json", + "definition": { + "arguments": { + "workflow-type": { + "name": "workflow-type", + "is_required": true, + "is_array": false, + "description": "Exact workflow type (at most 255 UTF-8 bytes)", + "default": null + }, + "change-id": { + "name": "change-id", + "is_required": true, + "is_array": false, + "description": "Exact version change ID (at most 1024 UTF-8 bytes)", + "default": null + } + }, + "options": { + "server": { + "name": "--server", + "shortcut": "-s", + "accept_value": true, + "is_value_required": false, + "is_multiple": false, + "description": "Server URL", + "default": null + }, + "namespace": { + "name": "--namespace", + "shortcut": "", + "accept_value": true, + "is_value_required": true, + "is_multiple": false, + "description": "Target namespace", + "default": null + }, + "token": { + "name": "--token", + "shortcut": "", + "accept_value": true, + "is_value_required": false, + "is_multiple": false, + "description": "Auth token", + "default": null + }, + "tls-verify": { + "name": "--tls-verify", + "shortcut": "", + "accept_value": true, + "is_value_required": true, + "is_multiple": false, + "description": "Verify TLS certificates (true/false). Defaults to $DURABLE_WORKFLOW_TLS_VERIFY, then the selected profile, then true.", + "default": null + }, + "env": { + "name": "--env", + "shortcut": "", + "accept_value": true, + "is_value_required": true, + "is_multiple": false, + "description": "Named dw environment to use (overrides $DW_ENV and any env set via `dw env:use`). Hard-fails when the profile does not exist.", + "default": null + }, + "output": { + "name": "--output", + "shortcut": "", + "accept_value": true, + "is_value_required": true, + "is_multiple": false, + "description": "Output format: table, json, or one complete page per line (jsonl)", + "default": "table" + }, + "min-version": { + "name": "--min-version", + "shortcut": "", + "accept_value": true, + "is_value_required": true, + "is_multiple": false, + "description": "Required proposed minimum supported version (signed 32-bit integer)", + "default": null + }, + "state": { + "name": "--state", + "shortcut": "", + "accept_value": true, + "is_value_required": true, + "is_multiple": false, + "description": "Filter returned runs by legacy, current, or unknown; counts retain all states", + "default": null + }, + "limit": { + "name": "--limit", + "shortcut": "", + "accept_value": true, + "is_value_required": true, + "is_multiple": false, + "description": "Runs per page (1..200)", + "default": "50" + }, + "cursor": { + "name": "--cursor", + "shortcut": "", + "accept_value": true, + "is_value_required": true, + "is_multiple": false, + "description": "Opaque continuation bound to this exact query", + "default": null + }, + "json": { + "name": "--json", + "shortcut": "", + "accept_value": false, + "is_value_required": false, + "is_multiple": false, + "description": "Output the command response as JSON (alias for --output=json)", + "default": false + }, + "help": { + "name": "--help", + "shortcut": "-h", + "accept_value": false, + "is_value_required": false, + "is_multiple": false, + "description": "Display help for the given command. When no command is given display help for the list command", + "default": false + }, + "silent": { + "name": "--silent", + "shortcut": "", + "accept_value": false, + "is_value_required": false, + "is_multiple": false, + "description": "Do not output any message", + "default": false + }, + "quiet": { + "name": "--quiet", + "shortcut": "-q", + "accept_value": false, + "is_value_required": false, + "is_multiple": false, + "description": "Only errors are displayed. All other output is suppressed", + "default": false + }, + "verbose": { + "name": "--verbose", + "shortcut": "-v|-vv|-vvv", + "accept_value": false, + "is_value_required": false, + "is_multiple": false, + "description": "Increase the verbosity of messages: 1 for normal output, 2 for more verbose output and 3 for debug", + "default": false + }, + "version": { + "name": "--version", + "shortcut": "-V", + "accept_value": false, + "is_value_required": false, + "is_multiple": false, + "description": "Display this application version", + "default": false + }, + "ansi": { + "name": "--ansi", + "shortcut": "", + "accept_value": false, + "is_value_required": false, + "is_multiple": false, + "description": "Force (or disable --no-ansi) ANSI output", + "default": null + }, + "no-ansi": { + "name": "--no-ansi", + "shortcut": "", + "accept_value": false, + "is_value_required": false, + "is_multiple": false, + "description": "Negate the \"--ansi\" option", + "default": null + }, + "no-interaction": { + "name": "--no-interaction", + "shortcut": "-n", + "accept_value": false, + "is_value_required": false, + "is_multiple": false, + "description": "Do not ask any interactive question", + "default": false + } + } + }, + "rust_only": true } ] } diff --git a/rust/parity/test_parity.py b/rust/parity/test_parity.py index 11fd68b..b868db0 100644 --- a/rust/parity/test_parity.py +++ b/rust/parity/test_parity.py @@ -234,7 +234,7 @@ class RustOnlyCommandTest(unittest.TestCase): def test_the_catalog_marks_the_batch_activity_dispatch_and_update_outcome_commands_rust_only(self): self.assertEqual(["batch:cancel", "batch:resume", "batch:show", "batch:signal", "batch:update", "task-queue:activity-dispatch", "task-queue:pause-activities", - "task-queue:resume-activities", "workflow:update-outcome"], + "task-queue:resume-activities", "workflow:update-outcome", "workflow:version-usage"], parity.rust_only_commands()) def test_rust_only_help_is_skipped_but_2x_help_is_kept(self): diff --git a/rust/src/catalog.json b/rust/src/catalog.json index 66d4190..6feae6d 100644 --- a/rust/src/catalog.json +++ b/rust/src/catalog.json @@ -16078,6 +16078,206 @@ } }, "rust_only": true + }, + { + "name": "workflow:version-usage", + "description": "Show live legacy, current and unknown version usage across open runs", + "synopsis": "workflow:version-usage [-s|--server [SERVER]] [--namespace NAMESPACE] [--token [TOKEN]] [--tls-verify TLS-VERIFY] [--env ENV] [--output OUTPUT] [--min-version MIN-VERSION] [--state STATE] [--limit LIMIT] [--cursor CURSOR] [--json] [--] ", + "aliases": [], + "hidden": false, + "help": "Read one live, namespace-scoped page of open runs for a workflow type and\nchange ID. --min-version is required: decisions below that proposed minimum\nare legacy; decisions at or above it are current. Use 1 for a patched() branch.\nNo marker, pruned/incomplete history or invalid decisions remain unknown.\nUnknown is never zero usage or a current decision.\n\nCounts cover all states before --state and paging. --cursor is opaque and\nbound to the exact namespace/type/change/minimum/state query. The first page\nsets an upper run-ID bound. New admissions beyond that bound stay excluded;\nclosures and new decisions can still change counts/state between pages.\nRestart after draining or fencing admissions before retiring a branch.\n\nRequires literal capabilities.workflow_version_usage=true from the Server.\nWithout it, dw exits 8 before the version-usage request. Unsupported routes,\nstorage failures and invalid/inconsistent responses are errors, never empty\nsuccessful counts. This command never runs workflow code or changes markers.\n\n--limit is 1..200 (default 50). --state is legacy, current or unknown.\nJSON and JSONL retain one complete page, including counts and continuation;\nJSONL does not discard the envelope when no runs match the selected state.\n\nExamples:\n\n dw workflow:version-usage shipment carrier-api --min-version=2 --namespace=orders\n dw workflow:version-usage shipment carrier-api --min-version=2 --state=unknown --json\n dw workflow:version-usage shipment carrier-api --min-version=2 --limit=20 --cursor=OPAQUE --json", + "definition": { + "arguments": { + "workflow-type": { + "name": "workflow-type", + "is_required": true, + "is_array": false, + "description": "Exact workflow type (at most 255 UTF-8 bytes)", + "default": null + }, + "change-id": { + "name": "change-id", + "is_required": true, + "is_array": false, + "description": "Exact version change ID (at most 1024 UTF-8 bytes)", + "default": null + } + }, + "options": { + "server": { + "name": "--server", + "shortcut": "-s", + "accept_value": true, + "is_value_required": false, + "is_multiple": false, + "description": "Server URL", + "default": null + }, + "namespace": { + "name": "--namespace", + "shortcut": "", + "accept_value": true, + "is_value_required": true, + "is_multiple": false, + "description": "Target namespace", + "default": null + }, + "token": { + "name": "--token", + "shortcut": "", + "accept_value": true, + "is_value_required": false, + "is_multiple": false, + "description": "Auth token", + "default": null + }, + "tls-verify": { + "name": "--tls-verify", + "shortcut": "", + "accept_value": true, + "is_value_required": true, + "is_multiple": false, + "description": "Verify TLS certificates (true/false). Defaults to $DURABLE_WORKFLOW_TLS_VERIFY, then the selected profile, then true.", + "default": null + }, + "env": { + "name": "--env", + "shortcut": "", + "accept_value": true, + "is_value_required": true, + "is_multiple": false, + "description": "Named dw environment to use (overrides $DW_ENV and any env set via `dw env:use`). Hard-fails when the profile does not exist.", + "default": null + }, + "output": { + "name": "--output", + "shortcut": "", + "accept_value": true, + "is_value_required": true, + "is_multiple": false, + "description": "Output format: table, json, or one complete page per line (jsonl)", + "default": "table" + }, + "min-version": { + "name": "--min-version", + "shortcut": "", + "accept_value": true, + "is_value_required": true, + "is_multiple": false, + "description": "Required proposed minimum supported version (signed 32-bit integer)", + "default": null + }, + "state": { + "name": "--state", + "shortcut": "", + "accept_value": true, + "is_value_required": true, + "is_multiple": false, + "description": "Filter returned runs by legacy, current, or unknown; counts retain all states", + "default": null + }, + "limit": { + "name": "--limit", + "shortcut": "", + "accept_value": true, + "is_value_required": true, + "is_multiple": false, + "description": "Runs per page (1..200)", + "default": "50" + }, + "cursor": { + "name": "--cursor", + "shortcut": "", + "accept_value": true, + "is_value_required": true, + "is_multiple": false, + "description": "Opaque continuation bound to this exact query", + "default": null + }, + "json": { + "name": "--json", + "shortcut": "", + "accept_value": false, + "is_value_required": false, + "is_multiple": false, + "description": "Output the command response as JSON (alias for --output=json)", + "default": false + }, + "help": { + "name": "--help", + "shortcut": "-h", + "accept_value": false, + "is_value_required": false, + "is_multiple": false, + "description": "Display help for the given command. When no command is given display help for the list command", + "default": false + }, + "silent": { + "name": "--silent", + "shortcut": "", + "accept_value": false, + "is_value_required": false, + "is_multiple": false, + "description": "Do not output any message", + "default": false + }, + "quiet": { + "name": "--quiet", + "shortcut": "-q", + "accept_value": false, + "is_value_required": false, + "is_multiple": false, + "description": "Only errors are displayed. All other output is suppressed", + "default": false + }, + "verbose": { + "name": "--verbose", + "shortcut": "-v|-vv|-vvv", + "accept_value": false, + "is_value_required": false, + "is_multiple": false, + "description": "Increase the verbosity of messages: 1 for normal output, 2 for more verbose output and 3 for debug", + "default": false + }, + "version": { + "name": "--version", + "shortcut": "-V", + "accept_value": false, + "is_value_required": false, + "is_multiple": false, + "description": "Display this application version", + "default": false + }, + "ansi": { + "name": "--ansi", + "shortcut": "", + "accept_value": false, + "is_value_required": false, + "is_multiple": false, + "description": "Force (or disable --no-ansi) ANSI output", + "default": null + }, + "no-ansi": { + "name": "--no-ansi", + "shortcut": "", + "accept_value": false, + "is_value_required": false, + "is_multiple": false, + "description": "Negate the \"--ansi\" option", + "default": null + }, + "no-interaction": { + "name": "--no-interaction", + "shortcut": "-n", + "accept_value": false, + "is_value_required": false, + "is_multiple": false, + "description": "Do not ask any interactive question", + "default": false + } + } + }, + "rust_only": true } ] } diff --git a/rust/src/commands/mod.rs b/rust/src/commands/mod.rs index afa7229..1afe0ce 100644 --- a/rust/src/commands/mod.rs +++ b/rust/src/commands/mod.rs @@ -17,6 +17,7 @@ pub mod start_dev; pub mod system; pub mod task_queue; pub mod upgrade; +pub mod version_usage; pub mod worker; pub mod workflow; pub mod workflow_control; diff --git a/rust/src/commands/version_usage.rs b/rust/src/commands/version_usage.rs new file mode 100644 index 0000000..0a747ea --- /dev/null +++ b/rust/src/commands/version_usage.rs @@ -0,0 +1,421 @@ +//! One namespace-scoped, live page of open-run version decisions (shared #220). + +use crate::cli::Input; +use crate::client::Context; +use crate::error::CliError; +use crate::{compat, formatter, output}; +use durable_workflow::RawRequest; +use serde::Deserialize; +use serde_json::Value; + +const CAPABILITY: &str = "workflow_version_usage"; +const NEXT_STEP: &str = "Connect to a Server that advertises workflow_version_usage. Missing markers or unavailable history must remain unknown; inspect retained history before retiring a branch."; + +#[derive(Clone, Copy, Debug, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +enum State { + Legacy, + Current, + Unknown, +} + +impl State { + fn name(self) -> &'static str { + match self { + Self::Legacy => "legacy", + Self::Current => "current", + Self::Unknown => "unknown", + } + } +} + +struct Query<'a> { + workflow_type: &'a str, + change_id: &'a str, + minimum: i32, + state: Option, + limit: u16, + cursor: Option<&'a str>, +} + +impl<'a> Query<'a> { + fn parse(input: &'a Input, namespace: &str) -> Result { + if input.provided("namespace") + && input + .opt("namespace") + .flatten() + .is_none_or(|value| value.trim().is_empty()) + { + return Err(CliError::Invalid( + "--namespace requires a nonempty namespace.".into(), + )); + } + if namespace.trim().is_empty() || namespace.chars().any(char::is_control) { + return Err(CliError::Invalid( + "The selected namespace must be nonempty and contain no control characters.".into(), + )); + } + let workflow_type = required_text(input.arg("workflow-type"), "workflow-type", 255)?; + let change_id = required_text(input.arg("change-id"), "change-id", 1024)?; + let minimum = input + .opt("min-version") + .flatten() + .and_then(|value| value.parse::().ok()) + .ok_or_else(|| { + CliError::Invalid( + "--min-version is required and must be a signed 32-bit integer.".into(), + ) + })?; + let state = match input.opt("state").flatten() { + None => None, + Some("legacy") => Some(State::Legacy), + Some("current") => Some(State::Current), + Some("unknown") => Some(State::Unknown), + Some(_) => { + return Err(CliError::Invalid( + "--state must be legacy, current, or unknown.".into(), + )) + } + }; + let limit = input + .opt("limit") + .flatten() + .and_then(|value| value.parse::().ok()) + .filter(|limit| (1..=200).contains(limit)) + .ok_or_else(|| CliError::Invalid("--limit must be an integer from 1 to 200.".into()))?; + let cursor = input.opt("cursor").flatten(); + if cursor == Some("") { + return Err(CliError::Invalid( + "--cursor must be the nonempty opaque continuation from this query.".into(), + )); + } + Ok(Self { + workflow_type, + change_id, + minimum, + state, + limit, + cursor, + }) + } + + fn request(&self) -> RawRequest { + let mut request = RawRequest::get(["workflow-version-usage"]) + .query("workflow_type", self.workflow_type) + .query("change_id", self.change_id) + .query("min_version", self.minimum.to_string()) + .query("limit", self.limit.to_string()); + if let Some(state) = self.state { + request = request.query("state", state.name()); + } + if let Some(cursor) = self.cursor { + request = request.query("cursor", cursor); + } + request + } +} + +fn required_text<'a>( + value: Option<&'a str>, + name: &str, + maximum: usize, +) -> Result<&'a str, CliError> { + value + .filter(|value| !value.is_empty() && value.len() <= maximum) + .ok_or_else(|| { + CliError::Invalid(format!( + "{name} must be nonempty UTF-8 text of at most {maximum} bytes." + )) + }) +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct Counts { + total: u64, + legacy: u64, + current: u64, + unknown: u64, +} + +impl Counts { + fn for_state(&self, state: State) -> u64 { + match state { + State::Legacy => self.legacy, + State::Current => self.current, + State::Unknown => self.unknown, + } + } +} + +#[derive(Deserialize)] +#[serde(rename_all = "snake_case")] +enum UnknownReason { + MarkerAbsent, + HistoryUnavailable, + HistoryInvalid, +} + +impl UnknownReason { + fn name(&self) -> &'static str { + match self { + Self::MarkerAbsent => "marker_absent", + Self::HistoryUnavailable => "history_unavailable", + Self::HistoryInvalid => "history_invalid", + } + } +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct Run { + instance_id: String, + run_id: String, + status: String, + state: State, + recorded_version: Option, + unknown_reason: Option, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct Page { + schema_version: u8, + namespace: String, + workflow_type: String, + change_id: String, + min_version: i32, + state: Option, + consistency: String, + through_run_id: Option, + counts: Counts, + runs: Vec, + limit: u16, + returned_count: usize, + has_more: bool, + next_cursor: Option, +} + +fn invalid_response(detail: &str) -> CliError { + CliError::Failure(format!("Workflow version usage is unavailable: invalid Server response ({detail}). Counts have not been established.")) +} + +impl Page { + fn parse(body: &Value, query: &Query<'_>, namespace: &str) -> Result { + // Nullable fields are still required by v1. Serde alone treats missing + // Option fields as null, which would hide incomplete producer data. + for name in [ + "schema_version", + "namespace", + "workflow_type", + "change_id", + "min_version", + "state", + "consistency", + "through_run_id", + "counts", + "runs", + "limit", + "returned_count", + "has_more", + "next_cursor", + ] { + if body.get(name).is_none() { + return Err(invalid_response("missing required v1 fields")); + } + } + let page: Self = serde_json::from_value(body.clone()) + .map_err(|_| invalid_response("v1 field types or values"))?; + if page.schema_version != 1 + || page.namespace != namespace + || page.workflow_type != query.workflow_type + || page.change_id != query.change_id + || page.min_version != query.minimum + || page.state != query.state + || page.consistency != "live" + || page.limit != query.limit + { + return Err(invalid_response( + "query identity or live-page contract mismatch", + )); + } + let total = page + .counts + .legacy + .checked_add(page.counts.current) + .and_then(|n| n.checked_add(page.counts.unknown)); + if total != Some(page.counts.total) + || page.returned_count != page.runs.len() + || page.runs.len() > usize::from(query.limit) + { + return Err(invalid_response("counts or page length")); + } + let bound = page.through_run_id.as_deref(); + if bound == Some("") + || (bound.is_none() + && (page.counts.total != 0 || !page.runs.is_empty() || page.has_more)) + || page.has_more + != page + .next_cursor + .as_deref() + .is_some_and(|cursor| !cursor.is_empty()) + || (!page.has_more && page.next_cursor.is_some()) + || (page.has_more + && (page.runs.is_empty() || page.next_cursor.as_deref() == query.cursor)) + { + return Err(invalid_response("continuation or upper bound")); + } + page.validate_runs(body, query, bound)?; + Ok(page) + } + + fn validate_runs( + &self, + body: &Value, + query: &Query<'_>, + bound: Option<&str>, + ) -> Result<(), CliError> { + let raw_runs = body["runs"] + .as_array() + .ok_or_else(|| invalid_response("run list"))?; + let mut previous: Option<&str> = None; + let mut observed = [0u64; 3]; + for (run, raw) in self.runs.iter().zip(raw_runs) { + if raw.get("recorded_version").is_none() + || raw.get("unknown_reason").is_none() + || run.instance_id.is_empty() + || run.run_id.is_empty() + || !matches!(run.status.as_str(), "pending" | "running" | "waiting") + || previous.is_some_and(|id| id >= run.run_id.as_str()) + || bound.is_none_or(|id| run.run_id.as_str() > id) + || query.state.is_some_and(|state| run.state != state) + { + return Err(invalid_response("run identity, ordering or state filter")); + } + previous = Some(&run.run_id); + let valid = match (run.state, run.recorded_version, run.unknown_reason.as_ref()) { + (State::Legacy, Some(version), None) => version < query.minimum, + (State::Current, Some(version), None) => version >= query.minimum, + (State::Unknown, None, Some(_)) => true, + _ => false, + }; + if !valid { + return Err(invalid_response("version classification or unknown reason")); + } + observed[match run.state { + State::Legacy => 0, + State::Current => 1, + State::Unknown => 2, + }] += 1; + } + if [State::Legacy, State::Current, State::Unknown] + .into_iter() + .zip(observed) + .any(|(state, count)| count > self.counts.for_state(state)) + { + return Err(invalid_response("state counts exclude returned runs")); + } + Ok(()) + } +} + +pub fn run(context: &Context, input: &Input) -> Result { + let query = Query::parse(input, &context.connection.namespace)?; + let info = context.preflight(false)?; + compat::require_capability( + &info, + crate::build_info::version(), + CAPABILITY, + "Version usage is unavailable. No legacy/current/unknown counts have been established.", + NEXT_STEP, + ) + .map_err(CliError::Compatibility)?; + let body = context + .resource_request(query.request()) + .map_err(|error| match error { + CliError::Http { + status, + message, + body, + } if matches!(status, 404 | 405 | 501) => CliError::Http { + status, + message: format!( + "Workflow version usage is unavailable (HTTP {status}): {message}" + ), + body, + }, + other => other, + })?; + let page = Page::parse(&body, &query, &context.connection.namespace)?; + if context.mode.machine_readable() { + // One complete page per JSON/JSONL record retains counts even when the + // selected page is empty. Row-only JSONL would discard unknown counts. + output::print_json(&body); + } else { + render(&page); + } + Ok(0) +} + +fn render(page: &Page) { + let text = |value: &str| formatter::display_text(value); + formatter::writeln_out(&format!("Namespace: {}", text(&page.namespace))); + formatter::writeln_out(&format!( + "Workflow type: {} | Change ID: {} | Minimum version: {}", + text(&page.workflow_type), + text(&page.change_id), + page.min_version + )); + formatter::writeln_out(&format!( + "Open runs (all states before filter/page): total={} legacy={} current={} unknown={}", + page.counts.total, page.counts.legacy, page.counts.current, page.counts.unknown + )); + formatter::writeln_out( + "Live observation. Counts and state membership can change between pages.", + ); + if page.counts.unknown > 0 { + formatter::writeln_out("Unknown runs remain unresolved; missing markers and unavailable history are not current decisions."); + } + if page.runs.is_empty() { + formatter::writeln_out("No runs on this page."); + } else { + let rows = page + .runs + .iter() + .map(|run| { + vec![ + text(&run.instance_id), + text(&run.run_id), + text(&run.status), + run.state.name().into(), + run.recorded_version + .map(|v| v.to_string()) + .unwrap_or_else(|| "-".into()), + run.unknown_reason + .as_ref() + .map(|r| r.name().to_string()) + .unwrap_or_else(|| "-".into()), + ] + }) + .collect::>(); + formatter::writeln_out(&output::render_table( + &[ + "Workflow ID", + "Run ID", + "Status", + "State", + "Recorded version", + "Unknown reason", + ], + &rows, + )); + } + formatter::writeln_out(&format!( + "Returned: {} | Has more: {}", + page.returned_count, page.has_more + )); + if let Some(cursor) = &page.next_cursor { + formatter::writeln_out(&format!("Next cursor: {}", text(cursor))); + } + formatter::writeln_out("Keep unknowns visible and restart after draining or fencing admissions before retiring a branch."); +} diff --git a/rust/src/completion.json b/rust/src/completion.json index 1660c65..8d37192 100644 --- a/rust/src/completion.json +++ b/rust/src/completion.json @@ -109,6 +109,10 @@ "value": "workflow:migrate-v1", "description": "Project a Waterline v1 workflow into standalone read-only visibility" }, + { + "value": "workflow:version-usage", + "description": "Show live legacy, current and unknown version usage across open runs" + }, { "value": "service-call:describe", "description": "Inspect a durable Nexus service call and its original target" @@ -1357,6 +1361,20 @@ "jsonl" ] } + }, + "workflow:version-usage": { + "options": { + "state": [ + "legacy", + "current", + "unknown" + ], + "output": [ + "table", + "json", + "jsonl" + ] + } } } } diff --git a/rust/src/main.rs b/rust/src/main.rs index f720ad6..33717de 100644 --- a/rust/src/main.rs +++ b/rust/src/main.rs @@ -37,8 +37,8 @@ enum Handler { fn handler(name: &str) -> Option { use commands::{ activity_task, batch, completion, debug, doctor, env, namespace, runtime, schedule, schema, - search_attribute, server, start_dev, system, task_queue, upgrade, worker, workflow, - workflow_control, workflow_message, workflow_run, + search_attribute, server, start_dev, system, task_queue, upgrade, version_usage, worker, + workflow, workflow_control, workflow_message, workflow_run, }; Some(match name { "completion" => Handler::Local(completion::completion), @@ -53,6 +53,7 @@ fn handler(name: &str) -> Option { "server:health" => Handler::Server(server::health), "server:info" => Handler::Server(server::info), "workflow:list" => Handler::Server(workflow::list), + "workflow:version-usage" => Handler::Server(version_usage::run), "batch:cancel" => Handler::Server(batch::cancel), "batch:signal" => Handler::Server(batch::signal), "batch:update" => Handler::Server(batch::update), @@ -141,6 +142,7 @@ fn handler(name: &str) -> Option { /// Commands whose errors carry the resolved namespace. const NAMESPACE_SCOPED: &[&str] = &[ "workflow:list", + "workflow:version-usage", "workflow:list-runs", "workflow:describe", "workflow:show-run", @@ -225,6 +227,7 @@ const GROUPED_ALIASES: &[(&[&str], GroupVerbs)] = &[ &["workflow", "workflows"], &[ ("list", "workflow:list"), + ("version-usage", "workflow:version-usage"), ("migrate-v1", "workflow:migrate-v1"), ], ), @@ -728,8 +731,12 @@ fn run(raw: Vec) -> i32 { let result = match handler { Handler::Local(run) => run(&input), Handler::Server(run) => client::Context::new(&input).and_then(|context| { - // doctor and server:start-dev opt out of the session warning in 2.x. - if !matches!(command.as_str(), "doctor" | "server:start-dev") { + // Version usage validates before its own strict preflight. The two + // original commands opt out of the session warning in 2.x. + if !matches!( + command.as_str(), + "doctor" | "server:start-dev" | "workflow:version-usage" + ) { context.warn_if_incompatible( client::uses_worker_protocol(&command), NAMESPACE_SCOPED.contains(&command.as_str()) diff --git a/rust/tests/version_usage.rs b/rust/tests/version_usage.rs new file mode 100644 index 0000000..0c90c98 --- /dev/null +++ b/rust/tests/version_usage.rs @@ -0,0 +1,545 @@ +//! Namespace-scoped version usage through the actual CLI and SDK transport. +use serde_json::{json, Value}; +use std::fs; +use std::io::{Read, Write}; +use std::net::TcpListener; +use std::path::PathBuf; +use std::process::{Command, Output}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; +use std::thread::{self, JoinHandle}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +struct Scratch(PathBuf); +impl Scratch { + fn new() -> Self { + static NEXT: AtomicU64 = AtomicU64::new(0); + let now = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos(); + let path = std::env::temp_dir().join(format!( + "dw-version-usage-{}-{now}-{}", + std::process::id(), + NEXT.fetch_add(1, Ordering::Relaxed) + )); + let mut builder = fs::DirBuilder::new(); + #[cfg(unix)] + { + use std::os::unix::fs::DirBuilderExt; + builder.mode(0o700); + } + builder.create(&path).unwrap(); + Self(path) + } + fn run(&self, server: &Server, args: &[&str]) -> Output { + self.command(server, &["workflow:version-usage"], args) + } + fn command(&self, server: &Server, command: &[&str], args: &[&str]) -> Output { + Command::new(env!("CARGO_BIN_EXE_dw")) + .args(command) + .args([ + "--server", + &server.url, + "--token=fixture-token", + "--no-ansi", + ]) + .args(args) + .env_clear() + .env("DW_CONFIG_HOME", &self.0) + .env("HOME", &self.0) + .output() + .unwrap() + } +} +impl Drop for Scratch { + fn drop(&mut self) { + let _ = fs::remove_dir_all(&self.0); + } +} + +enum Reply { + Json(u16, Value), + Redirect(String), + Drop, +} +struct Server { + url: String, + requests: Arc>>, + stop: Arc, + thread: Option>, +} +impl Server { + fn new(info: Value, reply: Reply) -> Self { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let url = format!("http://{}/prefix", listener.local_addr().unwrap()); + listener.set_nonblocking(true).unwrap(); + let requests = Arc::new(Mutex::new(Vec::new())); + let stop = Arc::new(AtomicBool::new(false)); + let observed = requests.clone(); + let stopped = stop.clone(); + let thread = thread::spawn(move || { + while !stopped.load(Ordering::Relaxed) { + match listener.accept() { + Ok((mut stream, _)) => { + stream + .set_read_timeout(Some(Duration::from_secs(2))) + .unwrap(); + let mut bytes = Vec::new(); + let mut buffer = [0; 2048]; + while !bytes.windows(4).any(|part| part == b"\r\n\r\n") + && bytes.len() < 65536 + { + match stream.read(&mut buffer) { + Ok(0) | Err(_) => break, + Ok(size) => bytes.extend_from_slice(&buffer[..size]), + } + } + let request = String::from_utf8(bytes).unwrap(); + let cluster = request.starts_with("GET /prefix/api/cluster/info "); + let usage = request.starts_with("GET /prefix/api/workflow-version-usage?"); + observed.lock().unwrap().push(request); + let (status, body, extra) = if cluster { + (200, info.clone(), String::new()) + } else if usage { + match &reply { + Reply::Json(status, body) => (*status, body.clone(), String::new()), + Reply::Redirect(url) => { + (302, json!({}), format!("Location: {url}\r\n")) + } + Reply::Drop => continue, + } + } else { + ( + 404, + json!({"message":"Unknown fixture route"}), + String::new(), + ) + }; + let payload = serde_json::to_vec(&body).unwrap(); + let head = format!("HTTP/1.1 {status} Test\r\nX-Durable-Workflow-Control-Plane-Version: 2\r\nContent-Type: application/json\r\n{extra}Content-Length: {}\r\nConnection: close\r\n\r\n", payload.len()); + let _ = stream.write_all(head.as_bytes()); + let _ = stream.write_all(&payload); + } + Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { + thread::sleep(Duration::from_millis(2)) + } + Err(error) => panic!("fixture accept: {error}"), + } + } + }); + Self { + url, + requests, + stop, + thread: Some(thread), + } + } + fn requests(&self) -> Vec { + self.requests.lock().unwrap().clone() + } + fn usage(&self) -> Vec { + self.requests() + .into_iter() + .filter(|r| r.starts_with("GET /prefix/api/workflow-version-usage?")) + .collect() + } +} +impl Drop for Server { + fn drop(&mut self) { + self.stop.store(true, Ordering::Relaxed); + self.thread.take().unwrap().join().unwrap(); + } +} +fn info() -> Value { + let mut info: Value = serde_json::from_str(include_str!( + "../src/compat/fixtures/cluster-info-php-2.11.8.json" + )) + .unwrap(); + info["capabilities"]["workflow_version_usage"] = json!(true); + info +} +fn page() -> Value { + json!({"schema_version":1,"namespace":"orders","workflow_type":"orders.process","change_id":"billing-v2","min_version":1,"state":null,"consistency":"live","through_run_id":"run-z", + "counts":{"total":5,"legacy":1,"current":1,"unknown":3}, + "runs":[ + {"instance_id":"wf-1","run_id":"run-a","status":"running","state":"legacy","recorded_version":-1,"unknown_reason":null}, + {"instance_id":"wf-2","run_id":"run-b","status":"pending","state":"current","recorded_version":1,"unknown_reason":null}, + {"instance_id":"wf-3","run_id":"run-c","status":"waiting","state":"unknown","recorded_version":null,"unknown_reason":"marker_absent"}, + {"instance_id":"wf-4","run_id":"run-d","status":"waiting","state":"unknown","recorded_version":null,"unknown_reason":"history_unavailable"}, + {"instance_id":"wf-5","run_id":"run-e","status":"waiting","state":"unknown","recorded_version":null,"unknown_reason":"history_invalid"}], + "limit":50,"returned_count":5,"has_more":false,"next_cursor":null}) +} +fn args() -> Vec<&'static str> { + vec![ + "orders.process", + "billing-v2", + "--min-version=1", + "--namespace=orders", + ] +} +fn success(output: &Output) -> Value { + assert_eq!( + output.status.code(), + Some(0), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + serde_json::from_slice(&output.stdout).unwrap() +} +fn assert_failed(output: &Output) { + assert!( + !output.status.success(), + "unexpected success: {}", + String::from_utf8_lossy(&output.stdout) + ); + if !output.stdout.is_empty() { + let failure: Value = serde_json::from_slice(&output.stdout).expect("structured error"); + assert!(failure.get("error").is_some()); + assert!( + failure.get("counts").is_none(), + "failure emitted successful counts" + ); + } +} + +#[test] +fn preserves_mixed_decisions_and_all_counts_in_json_and_jsonl() { + let scratch = Scratch::new(); + for mode in ["--output=json", "--output=jsonl"] { + let body = page(); + let server = Server::new(info(), Reply::Json(200, body.clone())); + let mut query = args(); + query.push(mode); + let output = scratch.run(&server, &query); + assert_eq!(success(&output), body); + assert_eq!(String::from_utf8(output.stdout).unwrap().lines().count(), 1); + assert_eq!(server.usage().len(), 1); + assert_eq!(server.requests().len(), 2); + } +} + +#[test] +fn human_output_exposes_unknown_reasons_and_live_counts() { + let server = Server::new(info(), Reply::Json(200, page())); + let output = Scratch::new().run(&server, &args()); + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + let text = String::from_utf8(output.stdout).unwrap(); + for expected in [ + "total=5", + "legacy=1", + "current=1", + "unknown=3", + "marker_absent", + "history_unavailable", + "history_invalid", + "run-a", + "run-e", + "Live", + ] { + assert!(text.contains(expected), "missing {expected}: {text}"); + } + assert!(!text.contains("safe_to_remove")); +} + +#[test] +fn filtered_empty_page_retains_unfiltered_unknown_counts() { + let mut body = page(); + body["state"] = json!("legacy"); + body["runs"] = json!([]); + body["returned_count"] = json!(0); + let server = Server::new(info(), Reply::Json(200, body.clone())); + let mut query = args(); + query.extend(["--state=legacy", "--output=jsonl"]); + assert_eq!(success(&Scratch::new().run(&server, &query)), body); +} + +#[test] +fn encodes_exact_query_values_and_namespace_without_following_cursor() { + let workflow_type = "Commande/é?&"; + let change_id = "change/#?=& é"; + let mut body = page(); + body["workflow_type"] = json!(workflow_type); + body["change_id"] = json!(change_id); + body["state"] = json!("unknown"); + body["runs"] = json!([body["runs"][2].clone()]); + body["returned_count"] = json!(1); + body["limit"] = json!(1); + body["has_more"] = json!(true); + body["next_cursor"] = json!("next/opaque?&="); + let server = Server::new(info(), Reply::Json(200, body.clone())); + let query = [ + workflow_type, + change_id, + "--min-version=1", + "--namespace=orders", + "--state=unknown", + "--limit=1", + "--cursor=prior/opaque?&=", + "--output=json", + ]; + assert_eq!(success(&Scratch::new().run(&server, &query)), body); + let requests = server.usage(); + assert_eq!(requests.len(), 1); + let request = &requests[0]; + assert!(request + .to_ascii_lowercase() + .contains("x-namespace: orders\r\n")); + let target = request + .lines() + .next() + .unwrap() + .split_whitespace() + .nth(1) + .unwrap(); + let url = reqwest::Url::parse(&format!("http://fixture.test{target}")).unwrap(); + let values: std::collections::BTreeMap<_, _> = url.query_pairs().into_owned().collect(); + assert_eq!(values.get("workflow_type").unwrap(), workflow_type); + assert_eq!(values.get("change_id").unwrap(), change_id); + assert_eq!(values.get("cursor").unwrap(), "prior/opaque?&="); + assert_eq!(values.get("min_version").unwrap(), "1"); + assert_eq!(values.get("limit").unwrap(), "1"); + assert_eq!(values.get("state").unwrap(), "unknown"); + assert_eq!(values.len(), 6); +} + +#[test] +fn requires_literal_true_capability_before_usage_request() { + let scratch = Scratch::new(); + for capability in [ + Value::Null, + json!(false), + json!("true"), + json!(1), + json!({"supported":true}), + ] { + let mut cluster = info(); + cluster["capabilities"]["workflow_version_usage"] = capability; + let server = Server::new(cluster, Reply::Json(200, page())); + let mut query = args(); + query.push("--output=json"); + let output = scratch.run(&server, &query); + assert_eq!(output.status.code(), Some(8)); + assert_failed(&output); + assert!(format!( + "{}{}", + String::from_utf8_lossy(&output.stderr), + String::from_utf8_lossy(&output.stdout) + ) + .contains("workflow_version_usage")); + assert!(server.usage().is_empty()); + assert_eq!(server.requests().len(), 1); + } + let mut cluster = info(); + cluster["capabilities"] + .as_object_mut() + .unwrap() + .remove("workflow_version_usage"); + let server = Server::new(cluster, Reply::Json(200, page())); + assert_eq!(scratch.run(&server, &args()).status.code(), Some(8)); + assert!(server.usage().is_empty()); +} + +#[test] +fn rejects_invalid_arguments_before_any_http_request() { + let server = Server::new(info(), Reply::Json(200, page())); + let scratch = Scratch::new(); + let mut cases: Vec> = + vec![vec!["orders.process".into(), "billing-v2".into()], vec![]]; + for bad in ["2147483648", "-2147483649", "1.5", "1e2", "", " 1"] { + cases.push(vec![ + "orders.process".into(), + "billing-v2".into(), + format!("--min-version={bad}"), + ]); + } + for bad in [ + "--limit=0", + "--limit=201", + "--limit=1.5", + "--state=closed", + "--cursor=", + "--namespace=", + "--namespace= ", + "--namespace=bad\nnamespace", + ] { + let mut query: Vec<_> = args() + .into_iter() + .filter(|a| !a.starts_with("--namespace")) + .map(str::to_owned) + .collect(); + query.push(bad.into()); + cases.push(query); + } + for (workflow_type, change_id) in [ + ("".to_string(), "billing-v2".to_string()), + ("é".repeat(128), "billing-v2".into()), + ("orders.process".into(), "".into()), + ("orders.process".into(), "é".repeat(513)), + ] { + cases.push(vec![workflow_type, change_id, "--min-version=1".into()]); + } + for case in cases { + let query: Vec<_> = case.iter().map(String::as_str).collect(); + let output = scratch.run(&server, &query); + assert_eq!( + output.status.code(), + Some(if case.is_empty() { 1 } else { 2 }), + "{case:?}: {}", + String::from_utf8_lossy(&output.stderr) + ); + assert_failed(&output); + } + assert!(server.requests().is_empty()); +} + +#[test] +fn accepts_signed_integer_boundaries_and_selected_default_namespace() { + let scratch = Scratch::new(); + for minimum in [i32::MIN, i32::MAX] { + let mut body = page(); + body["namespace"] = json!("default"); + body["min_version"] = json!(minimum); + body["counts"] = json!({"total":0,"legacy":0,"current":0,"unknown":0}); + body["runs"] = json!([]); + body["returned_count"] = json!(0); + body["through_run_id"] = Value::Null; + let server = Server::new(info(), Reply::Json(200, body.clone())); + let version = format!("--min-version={minimum}"); + let output = scratch.run( + &server, + &["orders.process", "billing-v2", &version, "--output=json"], + ); + assert_eq!(success(&output), body); + assert!(server.usage()[0] + .to_ascii_lowercase() + .contains("x-namespace: default\r\n")); + } +} + +#[test] +fn rejects_inconsistent_scope_counts_run_states_and_continuations() { + let mut mutations = vec![ + ("/namespace", json!("other")), + ("/namespace", Value::Null), + ("/workflow_type", json!("other")), + ("/change_id", json!("other")), + ("/min_version", json!(2)), + ("/state", json!("legacy")), + ("/schema_version", json!(2)), + ("/consistency", json!("snapshot")), + ("/limit", json!(200)), + ("/counts/total", json!(0)), + ("/counts/unknown", json!(-1)), + ("/counts/current", json!(1.5)), + ("/returned_count", json!(0)), + ("/through_run_id", Value::Null), + ("/through_run_id", json!("run-0")), + ("/runs/0/status", json!("completed")), + ("/runs/0/instance_id", json!("")), + ("/runs/0/recorded_version", json!(1)), + ("/runs/1/recorded_version", json!(-1)), + ("/runs/0/unknown_reason", json!("marker_absent")), + ("/runs/2/recorded_version", json!(0)), + ("/runs/2/unknown_reason", Value::Null), + ("/runs/2/unknown_reason", json!("future_reason")), + ("/runs/1/run_id", json!("run-a")), + ("/has_more", json!(true)), + ("/next_cursor", json!("unexpected")), + ]; + let scratch = Scratch::new(); + for (pointer, value) in mutations.drain(..) { + let mut body = page(); + *body.pointer_mut(pointer).unwrap() = value; + let server = Server::new(info(), Reply::Json(200, body)); + let mut query = args(); + query.push("--output=json"); + let output = scratch.run(&server, &query); + assert_failed(&output); + assert!( + format!( + "{}{}", + String::from_utf8_lossy(&output.stderr), + String::from_utf8_lossy(&output.stdout) + ) + .contains("Counts have not been established"), + "{pointer}: {}", + String::from_utf8_lossy(&output.stderr) + ); + assert_eq!(server.usage().len(), 1); + } +} + +#[test] +fn required_nullable_fields_and_unknown_fields_cannot_hide_incomplete_data() { + let scratch = Scratch::new(); + for field in ["state", "through_run_id", "next_cursor"] { + let mut body = page(); + body.as_object_mut().unwrap().remove(field); + let server = Server::new(info(), Reply::Json(200, body)); + assert_failed(&scratch.run(&server, &args())); + } + for field in ["recorded_version", "unknown_reason"] { + let mut body = page(); + body["runs"][2].as_object_mut().unwrap().remove(field); + let server = Server::new(info(), Reply::Json(200, body)); + assert_failed(&scratch.run(&server, &args())); + } + let mut body = page(); + body["safe_to_remove"] = json!(true); + let server = Server::new(info(), Reply::Json(200, body)); + assert_failed(&scratch.run(&server, &args())); +} + +#[test] +fn errors_remain_failures_and_redirects_and_lost_replies_are_not_retried() { + let scratch = Scratch::new(); + for status in [401, 403, 404, 405, 422, 500, 503] { + let server = Server::new( + info(), + Reply::Json( + status, + json!({"reason":"validation_failed","message":"fixture failure"}), + ), + ); + assert_failed(&scratch.run(&server, &args())); + assert_eq!(server.usage().len(), 1, "HTTP {status}"); + assert_eq!(server.requests().len(), 2); + } + let destination = Server::new(info(), Reply::Json(200, page())); + let server = Server::new(info(), Reply::Redirect(destination.url.clone())); + assert_failed(&scratch.run(&server, &args())); + assert!(destination.requests().is_empty()); + assert_eq!(server.usage().len(), 1); + let server = Server::new(info(), Reply::Drop); + assert_failed(&scratch.run(&server, &args())); + assert_eq!(server.usage().len(), 1); +} + +#[test] +fn help_is_offline_and_grouped_alias_uses_the_same_command() { + let server = Server::new(info(), Reply::Json(200, page())); + let scratch = Scratch::new(); + let output = scratch.run(&server, &["--help"]); + assert!(output.status.success()); + let help = String::from_utf8(output.stdout).unwrap(); + for expected in [ + "--min-version", + "--state", + "--cursor", + "workflow_version_usage", + "unknown", + ] { + assert!(help.contains(expected), "{expected}"); + } + assert!(server.requests().is_empty()); + let mut query = args(); + query.push("--output=json"); + let output = scratch.command(&server, &["workflow", "version-usage"], &query); + assert_eq!(success(&output), page()); + assert_eq!(server.usage().len(), 1); +}