From 6a2fa85fec1102425fadd3b43c2e8ec3e48be687 Mon Sep 17 00:00:00 2001 From: Brandur Date: Wed, 4 Nov 2026 08:13:42 -0600 Subject: [PATCH] [rust] Align CLI with Go reference implementation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This came up in #506, but it turns out that Rust's CLI behaves different with respect to migration output compared to the reference implementation in Go. Here, tighten that up so that: * `migrate-list` shows all versions and names, marking the latest applied with `*`. * Progress output follows Go’s format; successful validation is silent. * SQL output requires `--show-sql`, including during dry runs. --- rust/CHANGELOG.md | 4 + rust/riverqueue-cli/README.md | 10 ++ rust/riverqueue-cli/src/main.rs | 2 +- rust/riverqueue-cli/src/migrate.rs | 155 ++++++++++++++++-- rust/riverqueue-cli/tests/migrate.rs | 236 +++++++++++++++++++++++++++ 5 files changed, 390 insertions(+), 17 deletions(-) create mode 100644 rust/riverqueue-cli/tests/migrate.rs diff --git a/rust/CHANGELOG.md b/rust/CHANGELOG.md index 58ea18e38..47175ed11 100644 --- a/rust/CHANGELOG.md +++ b/rust/CHANGELOG.md @@ -10,6 +10,10 @@ Changes to River for Go are recorded in the [repository changelog](../CHANGELOG. ## [Unreleased] +### Fixed + +- Align migration CLI output with Go: `migrate-list` shows every bundled version and name, marking the latest applied version with `*`; migration progress uses the Go format and reports when nothing remains to apply; successful validation is silent. SQL output now requires `--show-sql`, including with `--dry-run`. [PR #1580](https://github.com/riverqueue/river/pull/1580). + ## [0.5.0] - 2026-11-02 ### Changed diff --git a/rust/riverqueue-cli/README.md b/rust/riverqueue-cli/README.md index 1a27fb7ec..a1dfde5d9 100644 --- a/rust/riverqueue-cli/README.md +++ b/rust/riverqueue-cli/README.md @@ -23,6 +23,16 @@ options. `--target-version N`, `--max-steps N`, and `--dry-run` limit or preview a migration run. Applications can instead migrate from Rust with the [`riverqueue-migrate`](https://docs.rs/riverqueue-migrate) crate. +`migrate-list` prints every bundled version and its name, including pending +migrations, and marks the latest applied version with `*`, like the Go CLI. +Migration progress uses Go's `applied migration 001 [up] ... [duration]` format. +A dry run prints `migration 001 [up] [DRY RUN]`; add `--show-sql` to print the +SQL as well. `--show-sql` also works when applying migrations. Runs with nothing +to apply print `no migrations to apply`. + +Successful `validate` is silent. Outstanding migrations are reported on stderr +with exit status 2; connection and usage errors exit with status 1. + ## Benchmark `riverqueue bench` measures worker throughput and end-to-end latency. It diff --git a/rust/riverqueue-cli/src/main.rs b/rust/riverqueue-cli/src/main.rs index 2546726db..71f3cd76d 100644 --- a/rust/riverqueue-cli/src/main.rs +++ b/rust/riverqueue-cli/src/main.rs @@ -24,7 +24,7 @@ Usage: Commands: migrate-down Run down migrations - migrate-list List applied migration versions + migrate-list List migrations, marking the latest applied version with * migrate-up Run up migrations validate Check that all migrations are applied bench Benchmark job throughput against a disposable database diff --git a/rust/riverqueue-cli/src/migrate.rs b/rust/riverqueue-cli/src/migrate.rs index 07f11ee8c..eb3d1b6ba 100644 --- a/rust/riverqueue-cli/src/migrate.rs +++ b/rust/riverqueue-cli/src/migrate.rs @@ -1,6 +1,6 @@ //! Migration commands for River's canonical migration line. -use std::error::Error; +use std::{error::Error, time::Duration}; #[cfg(feature = "postgres")] use riverqueue::database::SchemaName; @@ -8,7 +8,7 @@ use riverqueue::database::SchemaName; use riverqueue_migrate::PostgresMigrator; #[cfg(feature = "sqlite")] use riverqueue_migrate::SqliteMigrator; -use riverqueue_migrate::{Direction, MigrateOpts, MigrateResult, ValidateResult}; +use riverqueue_migrate::{Direction, MigrateOpts, MigrateResult, Migration, ValidateResult}; #[cfg(feature = "postgres")] use sqlx::PgPool; #[cfg(feature = "sqlite")] @@ -21,6 +21,7 @@ struct Args { dry_run: bool, max_steps: Option, schema: Option, + show_sql: bool, target_version: Option, } @@ -32,6 +33,15 @@ enum CommandMigrator { } impl CommandMigrator { + fn all_versions(&self) -> &'static [Migration] { + match self { + #[cfg(feature = "postgres")] + Self::Postgres(_) => PostgresMigrator::all_versions(), + #[cfg(feature = "sqlite")] + Self::Sqlite(_) => SqliteMigrator::all_versions(), + } + } + async fn existing_versions(&self) -> Result, riverqueue_migrate::Error> { match self { #[cfg(feature = "postgres")] @@ -106,18 +116,28 @@ pub(crate) async fn run( match args.command.as_str() { "migrate-down" => { print_migrations( - migrator + &migrator .migrate(Direction::Down, migrate_opts(&args)) .await?, + &args, ); } "migrate-list" => { - for version in migrator.existing_versions().await? { - println!("{version:03}"); + let current = migrator.existing_versions().await?.last().copied(); + for migration in migrator.all_versions() { + let prefix = match current { + Some(version) if version == migration.version => "* ", + Some(_) => " ", + None => "", + }; + println!("{prefix}{:03} {}", migration.version, migration.name); } } "migrate-up" => { - print_migrations(migrator.migrate(Direction::Up, migrate_opts(&args)).await?); + print_migrations( + &migrator.migrate(Direction::Up, migrate_opts(&args)).await?, + &args, + ); } "validate" => { let result = migrator.validate(args.target_version).await?; @@ -125,13 +145,54 @@ pub(crate) async fn run( eprintln!("{result}"); std::process::exit(2); } - println!("River migrations valid"); } command => return Err(format!("unknown command {command:?}\n{}", usage()).into()), } Ok(()) } +fn format_duration(duration: Duration) -> String { + // Match the Go CLI's truncation and duration units, including minutes and hours. + let precision = if duration > Duration::from_secs(1) { + 10_000_000 + } else if duration < Duration::from_millis(1) { + 10 + } else { + 10_000 + }; + let nanos = duration.as_nanos() / precision * precision; + if nanos == 0 { + return "0s".to_owned(); + } + let mut prefix = String::new(); + let (value, scale, digits, unit) = if nanos < 1_000 { + (nanos, 1, 0, "ns") + } else if nanos < 1_000_000 { + (nanos, 1_000, 3, "µs") + } else if nanos < 1_000_000_000 { + (nanos, 1_000_000, 6, "ms") + } else { + let seconds = nanos / 1_000_000_000; + if seconds >= 3_600 { + prefix = format!("{}h{}m", seconds / 3_600, seconds / 60 % 60); + } else if seconds >= 60 { + prefix = format!("{}m", seconds / 60); + } + (nanos % 60_000_000_000, 1_000_000_000, 9, "s") + }; + let fraction = value % scale; + if fraction == 0 { + format!("{prefix}{}{unit}", value / scale) + } else { + let fraction = format!("{fraction:0digits$}"); + format!( + "{prefix}{}.{}{unit}", + value / scale, + fraction.trim_end_matches('0') + ) + } +} + fn migrate_opts(args: &Args) -> MigrateOpts { let mut opts = MigrateOpts::new().with_dry_run(args.dry_run); if let Some(max_steps) = args.max_steps { @@ -163,6 +224,7 @@ fn parse_args( args.max_steps = Some(raw.next().ok_or("--max-steps requires a value")?.parse()?); } "--schema" => args.schema = Some(raw.next().ok_or("--schema requires a value")?), + "--show-sql" => args.show_sql = true, "--target-version" => { args.target_version = Some( raw.next() @@ -180,18 +242,79 @@ fn parse_args( Ok(Some(args)) } -fn print_migrations(result: MigrateResult) { - for version in result.versions { - println!( - "{:03} {:?} {} ({:?})", - version.version, result.direction, version.name, version.duration - ); - if !version.sql.is_empty() && version.duration.is_zero() { - println!("{}", version.sql); +fn print_migrations(result: &MigrateResult, args: &Args) { + if result.versions.is_empty() { + println!("no migrations to apply"); + return; + } + let direction = match result.direction { + Direction::Down => "down", + Direction::Up => "up", + }; + let name_width = result + .versions + .iter() + .map(|version| version.name.len()) + .max() + .unwrap_or(0); + for version in &result.versions { + if args.dry_run { + println!("migration {:03} [{direction}] [DRY RUN]", version.version); + } else { + println!( + "applied migration {:03} [{direction}] {:name_width$} [{}]", + version.version, + version.name, + format_duration(version.duration) + ); + } + if args.show_sql { + println!("{}", "-".repeat(80)); + println!( + "-- River main migration {:03} [{direction}]", + version.version + ); + println!("{}\n", version.sql.trim()); } } + if args + .max_steps + .is_some_and(|steps| result.versions.len() < steps) + { + println!("no more migrations to apply"); + } } fn usage() -> &'static str { - "usage: riverqueue [--database-url URL] [--schema NAME] [--target-version N] [--max-steps N] [--dry-run]" + "usage: riverqueue [--database-url URL] [--schema NAME] [--target-version N] [--max-steps N] [--dry-run] [--show-sql]" +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn duration_format_matches_go() { + for (nanos, expected) in [ + (0, "0s"), + (9, "0s"), + (999, "990ns"), + (1_000, "1µs"), + (1_332, "1.33µs"), + (765_625, "765.62µs"), + (1_000_000, "1ms"), + (4_422_125, "4.42ms"), + (13_280_834, "13.28ms"), + (234_910_750, "234.91ms"), + (1_000_000_000, "1s"), + (3_937_042_000, "3.93s"), + (34_042_234_000, "34.04s"), + (60_000_000_000, "1m0s"), + (154_042_234_000, "2m34.04s"), + (3_600_000_000_000, "1h0m0s"), + (3_754_042_234_000, "1h2m34.04s"), + ] { + assert_eq!(format_duration(Duration::from_nanos(nanos)), expected); + } + } } diff --git a/rust/riverqueue-cli/tests/migrate.rs b/rust/riverqueue-cli/tests/migrate.rs new file mode 100644 index 000000000..32d00c33a --- /dev/null +++ b/rust/riverqueue-cli/tests/migrate.rs @@ -0,0 +1,236 @@ +#![cfg(any(feature = "sqlite", all(feature = "postgres", river_postgres_tests)))] + +use std::process::{Command, Output}; + +#[cfg(all(feature = "postgres", river_postgres_tests))] +use sqlx::{AssertSqlSafe, PgPool}; + +#[cfg(all(feature = "postgres", river_postgres_tests))] +#[tokio::test] +async fn postgres_migration_output() { + let database_url = std::env::var("RIVER_RUST_DATABASE_URL") + .expect("set RIVER_RUST_DATABASE_URL to run Postgres CLI tests"); + let pool = PgPool::connect(&database_url).await.unwrap(); + let schema = format!("RiverCli_{}_{}", std::process::id(), nonce()); + sqlx::raw_sql(AssertSqlSafe(format!("CREATE SCHEMA \"{schema}\""))) + .execute(&pool) + .await + .unwrap(); + let command = |arguments: &[&str]| run(&database_url, Some(&schema), arguments); + + assert_listing(&command, None); + let table: Option = sqlx::query_scalar("SELECT to_regclass($1)::text") + .bind(format!("\"{schema}\".river_migration")) + .fetch_one(&pool) + .await + .unwrap(); + assert!(table.is_none(), "listing must not create migration tables"); + exercise_migrations(&command); + + sqlx::raw_sql(AssertSqlSafe(format!("DROP SCHEMA \"{schema}\" CASCADE"))) + .execute(&pool) + .await + .unwrap(); + pool.close().await; +} + +#[cfg(feature = "sqlite")] +#[tokio::test] +async fn sqlite_migration_output() { + let path = + std::env::temp_dir().join(format!("river-cli-{}-{}.db", std::process::id(), nonce())); + let database_url = format!("sqlite://{}?mode=rwc", path.display()); + let pool = sqlx::SqlitePool::connect(&database_url).await.unwrap(); + let command = |arguments: &[&str]| run(&database_url, None, arguments); + + assert_listing(&command, None); + let count: i64 = + sqlx::query_scalar("SELECT count(*) FROM sqlite_master WHERE name = 'river_migration'") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(count, 0, "listing must not create migration tables"); + exercise_migrations(&command); + + pool.close().await; + std::fs::remove_file(path).unwrap(); +} + +fn assert_listing(command: &impl Fn(&[&str]) -> Output, current: Option) { + let names = [ + "create_river_migration", + "initial_schema", + "river_job_tags_non_null", + "pending_and_more", + "migration_unique_client", + "bulk_unique", + "notification_outbox_sqlite_jsonb_and_sql_cleanup", + "job_id_autoincrement", + ]; + let expected: Vec<_> = (1..) + .zip(names) + .map(|(version, name)| { + let prefix = if current == Some(version) { + "* " + } else if current.is_some() { + " " + } else { + "" + }; + format!("{prefix}{version:03} {name}") + }) + .collect(); + let stdout = success(command(&["migrate-list"])); + assert_eq!(stdout.lines().collect::>(), expected); + assert!(stdout.ends_with('\n')); +} + +fn assert_progress(output: Output, expected: &[&str]) { + let stdout = success(output); + let lines: Vec<_> = stdout.lines().collect(); + assert_eq!(lines.len(), expected.len(), "{stdout}"); + let duration_column = lines[0].rfind(" [").unwrap(); + for (line, expected) in lines.into_iter().zip(expected) { + assert_eq!(line.rfind(" ["), Some(duration_column), "{line}"); + let (prefix, duration) = line.rsplit_once(" [").expect("duration suffix"); + assert_eq!(prefix.trim_end(), *expected); + let duration = duration.strip_suffix(']').unwrap(); + assert!( + duration.chars().next().unwrap().is_ascii_digit(), + "{duration}" + ); + assert!(duration.ends_with('s'), "{duration}"); + } +} + +fn exercise_completion(command: &impl Fn(&[&str]) -> Output) { + let stdout = success(command(&["migrate-up", "--max-steps", "100"])); + assert!( + stdout.ends_with("no more migrations to apply\n"), + "{stdout}" + ); + assert_listing(command, Some(8)); + assert_eq!(success(command(&["validate"])), ""); + assert_eq!( + success(command(&["migrate-up"])), + "no migrations to apply\n" + ); + + let preview = success(command(&["migrate-down", "--dry-run"])); + assert_eq!(preview, "migration 008 [down] [DRY RUN]\n"); + assert_listing(command, Some(8)); + + let stdout = success(command(&[ + "migrate-down", + "--target-version", + "-1", + "--max-steps", + "100", + ])); + assert!(stdout.starts_with("applied migration 008 [down] job_id_autoincrement")); + assert!(stdout.ends_with("no more migrations to apply\n")); + assert_listing(command, None); + assert_eq!( + success(command(&["migrate-down"])), + "no migrations to apply\n" + ); +} + +fn exercise_migrations(command: &impl Fn(&[&str]) -> Output) { + let validation = command(&["validate"]); + assert_eq!(validation.status.code(), Some(2)); + assert_eq!(String::from_utf8(validation.stdout).unwrap(), ""); + assert_eq!( + String::from_utf8(validation.stderr).unwrap(), + "unapplied migrations: [1, 2, 3, 4, 5, 6, 7, 8]\n" + ); + + assert_eq!( + success(command(&["migrate-up", "--dry-run", "--max-steps", "2"])), + "migration 001 [up] [DRY RUN]\nmigration 002 [up] [DRY RUN]\n" + ); + let sql = success(command(&[ + "migrate-up", + "--dry-run", + "--show-sql", + "--max-steps", + "1", + ])); + assert!( + sql.starts_with(&format!( + "migration 001 [up] [DRY RUN]\n{}\n-- River main migration 001 [up]\n", + "-".repeat(80) + )), + "{sql}" + ); + assert!(sql.contains("CREATE TABLE"), "{sql}"); + assert!(sql.ends_with("\n\n")); + assert_listing(command, None); + + assert_progress( + command(&["migrate-up", "--max-steps", "2"]), + &[ + "applied migration 001 [up] create_river_migration", + "applied migration 002 [up] initial_schema", + ], + ); + assert_listing(command, Some(2)); + assert_eq!(success(command(&["validate", "--target-version", "2"])), ""); + assert_eq!(command(&["validate"]).status.code(), Some(2)); + assert_eq!( + success(command(&["migrate-up", "--target-version", "2"])), + "no migrations to apply\n" + ); + + let sql = success(command(&[ + "migrate-up", + "--show-sql", + "--target-version", + "3", + ])); + assert!(sql.starts_with("applied migration 003 [up] river_job_tags_non_null [")); + assert!(sql.contains("-- River main migration 003 [up]\n")); + assert!(sql.ends_with("\n\n")); + assert_listing(command, Some(3)); + + assert_progress( + command(&["migrate-down"]), + &["applied migration 003 [down] river_job_tags_non_null"], + ); + assert_listing(command, Some(2)); + exercise_completion(command); +} + +fn nonce() -> u128 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos() +} + +fn run(database_url: &str, schema: Option<&str>, arguments: &[&str]) -> Output { + let mut command = Command::new(env!("CARGO_BIN_EXE_riverqueue")); + command.env_remove("DATABASE_URL"); + command + .args(arguments) + .args(["--database-url", database_url]); + if let Some(schema) = schema { + command.args(["--schema", schema]); + } + command.output().unwrap() +} + +fn success(output: Output) -> String { + assert!( + output.status.success(), + "stdout: {}\nstderr: {}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + assert!( + output.stderr.is_empty(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + String::from_utf8(output.stdout).unwrap() +}