diff --git a/docs/contracts/external-payload-storage.md b/docs/contracts/external-payload-storage.md index e23a674df..f1071be7e 100644 --- a/docs/contracts/external-payload-storage.md +++ b/docs/contracts/external-payload-storage.md @@ -261,6 +261,25 @@ curl -fsS "$SERVER/api/system/external-payload-backup-hold?owner=$BACKUP_ID" \ -H "Authorization: Bearer $ADMIN_TOKEN" -H 'X-Durable-Workflow-Control-Plane-Version: 2' ``` +A restore that is still offline, with no Server process running, uses the +native image's two console commands instead. They read only the database +settings (`DB_CONNECTION` and its connection variables), open only an already +initialized native database, never create or change its schema, print the +same JSON report, and start no listener or workers: + +```sh +# The hold as stored, read-only: no lock, no row created, no write. +docker run --rm --env-file db.env "$SERVER_IMAGE" backup-hold inspect +# Release the restored hold by its recorded owner; repeating it is harmless. +docker run --rm --env-file db.env "$SERVER_IMAGE" \ + backup-hold release --owner="$BACKUP_ID" +``` + +On refusal they print the reason (`backup_hold_not_owned`, +`native_schema_not_initialized`, or the owner validation message) and exit +non-zero. Acquire, renew and owner-scoped status stay on the endpoints, while +the Server runs. + Each answers the command's JSON report. Where a command exits unsuccessfully, the endpoint answers 409: `backup_hold_owned_elsewhere` when another backup holds the hold, and `backup_hold_not_owned` when the owner's hold expired, was diff --git a/docs/self-hosted-backup-and-restore.md b/docs/self-hosted-backup-and-restore.md index 06bb4db99..37c236228 100644 --- a/docs/self-hosted-backup-and-restore.md +++ b/docs/self-hosted-backup-and-restore.md @@ -4,8 +4,11 @@ This procedure applies to the PHP 2.x Compose stack, `docker-compose.php.yml`. The published Compose file runs the native server from 3.0.0. On native 3.x Server, take and release the backup hold through its admin -endpoints instead of `php artisan external-payloads:backup-hold`. The owner -UUID, TTL limits and status checks are the same; see +endpoints instead of `php artisan external-payloads:backup-hold`. To release a +restored hold before any Server process starts, run the image's offline +`backup-hold release --owner=` command against the restored database +(`backup-hold inspect` reads the hold without changing anything). The +owner UUID, TTL limits and status checks are the same; see [coordinating online backups](contracts/external-payload-storage.md#coordinating-online-backups). This procedure applies to the published single-node Compose stack with MySQL, diff --git a/rust/src/main.rs b/rust/src/main.rs index 95a36c2c0..ebc9f44d5 100644 --- a/rust/src/main.rs +++ b/rust/src/main.rs @@ -245,6 +245,42 @@ fn main() -> Result<(), Box> { .block_on(run()) } +/// `backup-hold release --owner ` and `backup-hold inspect`: what a +/// restore runs before any Server process starts (Server #973). It opens only +/// an initialized native database from the DB settings, `inspect` read-only, +/// prints the v1 report, and on refusal prints the reason and exits non-zero. +async fn backup_hold_command() -> Result<(), Box> { + use durable_workflow_server::runtime::backup_hold; + let args: Vec = env::args().skip(2).collect(); + let backend = env::var("DB_CONNECTION").unwrap_or_default(); + let result = match backup_hold::command(&args) { + Err(error) => Err(error), + Ok(command) => match backend.as_str() { + "sqlite" => { + let path = env::var("DB_DATABASE").map_err(|_| "DB_DATABASE is required")?; + backup_hold::console_sqlite(&path, &command).await + } + "pgsql" => backup_hold::console_postgres(postgres_options()?, &command).await, + "mysql" | "mariadb" => backup_hold::console_mysql(mysql_options()?, &command).await, + _ => { + return Err( + "backup-hold requires DB_CONNECTION=sqlite, pgsql, mysql or mariadb".into(), + ); + } + }, + }; + match result { + Ok(report) => { + println!("{report}"); + Ok(()) + } + Err(error) => { + eprintln!("{}", backup_hold::console_failure(&error)); + Err("backup hold command refused".into()) + } + } +} + /// A release build carries the version of its `native-` tag, built in as /// `DW_SERVER_VERSION`, and starts without the opt-in. A development build /// (no release version) still requires `DW_RUST_EXPERIMENTAL=1`, so an @@ -257,6 +293,9 @@ async fn run() -> Result<(), Box> { if !release_build() && env::var("DW_RUST_EXPERIMENTAL").as_deref() != Ok("1") { return Err("Rust development build requires DW_RUST_EXPERIMENTAL=1; release builds do not".into()); } + if env::args().nth(1).as_deref() == Some("backup-hold") { + return backup_hold_command().await; + } if let Some(command) = env::args().nth(1) { if env::args().count() != 2 || !matches!( diff --git a/rust/src/runtime/backup_hold.rs b/rust/src/runtime/backup_hold.rs index 794ca2014..8bdd8cbbc 100644 --- a/rust/src/runtime/backup_hold.rs +++ b/rust/src/runtime/backup_hold.rs @@ -187,6 +187,75 @@ fn sql_time(at: DateTime) -> String { at.format("%Y-%m-%d %H:%M:%S").to_string() } +/// One offline `backup-hold` invocation (Server #973): only what a restore +/// needs before any Server process starts. Live backups acquire, renew and +/// check their hold through #953's endpoints. +#[derive(Debug, PartialEq, Eq)] +pub enum Command { + /// `release --owner `: #953's release, by the hold's recorded + /// owner, idempotent, a lapsed restored hold included. + Release { owner: String }, + /// `inspect`: the hold as stored, read without a lock or a write. + Inspect, +} + +impl Command { + /// Whether the command writes; `inspect` opens the database read-only. + fn writes(&self) -> bool { + matches!(self, Self::Release { .. }) + } +} + +/// Parse the arguments after `backup-hold`: `release --owner ` (or +/// `--owner=`), or `inspect` alone. +pub fn command(args: &[String]) -> Result { + let invalid = || super::refuse(StatusCode::BAD_REQUEST, "backup_hold_command_invalid"); + match args { + [action, rest @ ..] if action == "release" => { + let owner = match rest { + [option] => option.strip_prefix("--owner=").map(str::to_owned), + [flag, value] if flag == "--owner" => Some(value.clone()), + _ => None, + }; + owner.map(|owner| Command::Release { owner }).ok_or_else(invalid) + } + [action] if action == "inspect" => Ok(Command::Inspect), + [action, ..] if action == "inspect" => Err(invalid()), + _ => Err(super::refuse(StatusCode::BAD_REQUEST, "backup_hold_action_invalid")), + } +} + +/// `backup-hold` against a native SQLite database that already exists. +pub async fn console_sqlite(database: &str, command: &Command) -> Result { + let store = super::store::Store::new(super::sqlite::open_existing(database, command.writes()).await?); + store.run_backup_hold_command(command).await +} + +/// `backup-hold` against an initialized native PostgreSQL database. +pub async fn console_postgres(options: sqlx::postgres::PgConnectOptions, command: &Command) -> Result { + let store = super::store::Store::new(super::postgres::PostgresStorage::open_existing(options, command.writes()).await?); + store.run_backup_hold_command(command).await +} + +/// `backup-hold` against an initialized native MySQL or MariaDB database. +pub async fn console_mysql(options: sqlx::mysql::MySqlConnectOptions, command: &Command) -> Result { + let store = super::store::Store::new(super::mysql::MySqlStorage::open_existing(options).await?); + store.run_backup_hold_command(command).await +} + +/// The line the command prints on failure: the refusal reason, or the +/// validation messages of an invalid owner or TTL. +pub fn console_failure(error: &super::RuntimeError) -> String { + match error { + super::RuntimeError::Refused { reason, .. } => (*reason).to_owned(), + super::RuntimeError::Protocol { response, .. } => response["message"] + .as_str() + .unwrap_or("backup_hold_request_invalid") + .to_owned(), + _ => "storage_unavailable".to_owned(), + } +} + store_methods! { /// The hold row and the database's clock, without a lock: the preflight /// PHP's `assertReclamationAllowed` makes before a prune stages its @@ -246,6 +315,35 @@ store_methods! { Ok(next.report()) } + /// The offline command's report. Release is #953's own method; the pool + /// closes whatever the outcome. + async fn run_backup_hold_command(&self, command: &Command) -> Result { + let result = match command { + Command::Inspect => self.inspect_backup_hold().await, + Command::Release { owner } => match request(&json!({ "owner": owner }), Action::Release) { + Ok(request) => self.change_backup_hold(Action::Release, &request).await, + Err(error) => Err(error), + }, + }; + self.close().await; + result + } + + /// The hold as stored, from plain reads in one read-only snapshot: no + /// lock, no row creation, no write. A missing row is no hold. + async fn inspect_backup_hold(&self) -> Result { + let mut tx = self.begin_read().await?; + let hold = match Self::read_backup_hold(&mut tx).await? { + Some(hold) => hold, + None => { + let now = Self::query(DB::BACKUP_HOLD_CLOCK).fetch_one(&mut *tx).await?; + Hold { owner: None, acquired_at: None, expires_at: None, now: DB::instant(&now, "database_time")? } + } + }; + tx.rollback().await?; + Ok(hold.report()) + } + /// `status`: the hold as it stands, refused when a scoped owner no longer /// holds it. pub(crate) async fn backup_hold_status(&self, owner: Option<&str>) -> Result { diff --git a/rust/src/runtime/mod.rs b/rust/src/runtime/mod.rs index 669799cf1..39d1681e9 100644 --- a/rust/src/runtime/mod.rs +++ b/rust/src/runtime/mod.rs @@ -78,7 +78,7 @@ mod memo_limits; mod history_meter; mod standalone_activities; mod archive; -pub(crate) mod backup_hold; +pub mod backup_hold; mod command_payloads; mod completion_rules; mod deployments; diff --git a/rust/src/runtime/mysql.rs b/rust/src/runtime/mysql.rs index 34cd8a13f..25193b22e 100644 --- a/rust/src/runtime/mysql.rs +++ b/rust/src/runtime/mysql.rs @@ -811,6 +811,34 @@ impl MySqlStorage { let pool = runtime_pool_options().connect_with(connect_options).await?; Ok(Self { pool }) } + /// Open an initialized native database for an offline console command + /// (Server #973). There is no bootstrap path: the pool's own connection + /// must find the native marker and the exact schema, or the database is + /// refused as it is. The check only reads, in one read-only transaction. + pub(super) async fn open_existing(connect_options: MySqlConnectOptions) -> Result { + let pool = runtime_pool_options() + .max_connections(1) + .connect_with(options(connect_options)) + .await?; + let result = async { + let mut connection = pool.acquire().await?; + configure(&mut connection).await?; + let flavor = Flavor::detect(&mut connection).await?; + let mut tx = connection.begin_with("START TRANSACTION READ ONLY").await?; + if !scope(&mut tx).await?.iter().any(|table| table == "dw_server_schema") { + return Err(refuse(StatusCode::CONFLICT, "native_schema_not_initialized")); + } + verify(&mut tx, flavor).await?; + tx.rollback().await?; + Ok::<_, super::RuntimeError>(()) + } + .await; + if let Err(error) = result { + pool.close().await; + return Err(error); + } + Ok(pool) + } pub async fn check(connect_options: MySqlConnectOptions) -> Result<()> { let mut connection = MySqlConnection::connect_with(&options(connect_options)).await?; let result = async { diff --git a/rust/src/runtime/postgres.rs b/rust/src/runtime/postgres.rs index 3ff5ef5b8..55b27aa93 100644 --- a/rust/src/runtime/postgres.rs +++ b/rust/src/runtime/postgres.rs @@ -390,6 +390,43 @@ impl PostgresStorage { Ok(Self { pool }) } + /// Open an initialized native database for an offline console command + /// (Server #973). There is no bootstrap path: under the bootstrap + /// advisory lock, the pool's own connection must find the native marker + /// and the exact schema, or the database is refused as it is. A command + /// that only reads checks in a read-only transaction, without the lock. + pub(super) async fn open_existing(connect_options: PgConnectOptions, writable: bool) -> Result { + let pool = PgPoolOptions::new() + .max_connections(1) + .acquire_timeout(Duration::from_secs(5)) + .connect_with(options(connect_options)) + .await?; + let result = async { + let mut tx = if writable { + let mut tx = pool.begin().await?; + sqlx::query("SELECT pg_advisory_xact_lock(4924501473315676162)") + .execute(&mut *tx) + .await?; + tx + } else { + pool.begin_with("BEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY").await? + }; + verify_scope(&mut tx).await?; + if !marked(&mut tx).await? { + return Err(refuse(StatusCode::CONFLICT, "native_schema_not_initialized")); + } + verify(&mut tx).await?; + tx.commit().await?; + Ok::<_, super::RuntimeError>(()) + } + .await; + if let Err(error) = result { + pool.close().await; + return Err(error); + } + Ok(pool) + } + /// Upgrade an earlier native schema version to this one, in one /// transaction under the bootstrap advisory lock. Returns the version /// found; a current database is only verified. diff --git a/rust/src/runtime/sqlite.rs b/rust/src/runtime/sqlite.rs index 19387878d..3fa2ed044 100644 --- a/rust/src/runtime/sqlite.rs +++ b/rust/src/runtime/sqlite.rs @@ -26,6 +26,50 @@ pub(super) async fn open(database: &str) -> Result { unreachable!("the final startup attempt always returns") } +/// Open an initialized native database for an offline console command +/// (Server #973). There is no bootstrap path: the file is never created or +/// switched to WAL, and the pool's own connection must find the native marker +/// and the exact schema, or the database is refused as it is. A command that +/// only reads opens the file read-only. +pub(super) async fn open_existing(database: &str, writable: bool) -> Result { + if database.is_empty() || database.contains('?') || database.contains(':') { + return Err(refuse( + StatusCode::BAD_REQUEST, + "invalid_development_configuration", + )); + } + let path = Path::new(database); + if !path.exists() { + return Err(refuse(StatusCode::CONFLICT, "native_schema_not_initialized")); + } + let pool = SqlitePoolOptions::new() + .max_connections(1) + .acquire_timeout(Duration::from_secs(5)) + .connect_with( + SqliteConnectOptions::new() + .filename(path) + .create_if_missing(false) + .read_only(!writable) + .synchronous(SqliteSynchronous::Full) + .foreign_keys(true) + .busy_timeout(Duration::from_secs(5)), + ) + .await?; + let result = async { + let mut connection = pool.acquire().await?; + if !inspect(&mut connection).await? { + return Err(refuse(StatusCode::CONFLICT, "native_schema_not_initialized")); + } + Ok::<_, super::RuntimeError>(()) + } + .await; + if let Err(error) = result { + pool.close().await; + return Err(error); + } + Ok(pool) +} + async fn open_once(database: &str) -> Result { if database.is_empty() || database.contains('?') || database.contains(':') { return Err(refuse( diff --git a/rust/tests/execution/external_payloads/backup_hold.rs b/rust/tests/execution/external_payloads/backup_hold.rs index 691b9370e..351ec95f1 100644 --- a/rust/tests/execution/external_payloads/backup_hold.rs +++ b/rust/tests/execution/external_payloads/backup_hold.rs @@ -215,3 +215,151 @@ async fn a_restored_lapsed_hold_is_released_by_its_recorded_owner() { let (code, body) = change(&router, "release", json!({"owner": OTHER})).await; assert_eq!((code, &body["reason"]), (StatusCode::CONFLICT, &json!("backup_hold_not_owned")), "{body}"); } + +/// The offline `backup-hold` command (Server #973), run as its own process +/// with only the DB settings: (succeeded, printed report, first stderr +/// line: the refusal reason; main's own error line follows it). +async fn console(database: &TestDatabase, args: &[&str]) -> (bool, Value, String) { + let mut command = std::process::Command::new(env!("CARGO_BIN_EXE_durable-workflow-server")); + command.env_clear().env("DW_RUST_EXPERIMENTAL", "1").arg("backup-hold").args(args); + match database { + TestDatabase::Sqlite(dir) => command.env("DB_CONNECTION", "sqlite").env("DB_DATABASE", dir.path().join("runtime.sqlite")), + TestDatabase::Postgres { url, .. } => command.env("DB_CONNECTION", "pgsql").env("DB_URL", url), + TestDatabase::MySql { url, .. } => command.env("DB_CONNECTION", "mysql").env("DB_URL", url), + }; + // Waiting on the child blocks, so it runs on the blocking pool. + let output = tokio::task::spawn_blocking(move || command.output().unwrap()).await.unwrap(); + let stdout = String::from_utf8_lossy(&output.stdout); + (output.status.success(), serde_json::from_str(stdout.trim()).unwrap_or(Value::Null), first_line(&output.stderr)) +} + +fn first_line(stderr: &[u8]) -> String { + String::from_utf8_lossy(stderr).lines().next().unwrap_or_default().to_owned() +} + +/// The hold row as stored, as text, for comparing before and after a read. +async fn hold_rows(database: &TestDatabase) -> Vec<(String, String)> { + text_rows(database, + "SELECT COALESCE(owner,'')||'|'||COALESCE(acquired_at,''), COALESCE(expires_at,'') FROM runtime_external_payload_backup_hold", + "SELECT COALESCE(owner::text,'')||'|'||COALESCE(acquired_at::text,''), COALESCE(expires_at::text,'') FROM runtime_external_payload_backup_hold", + "SELECT CONCAT(COALESCE(owner,''),'|',COALESCE(CAST(acquired_at AS CHAR),'')), COALESCE(CAST(expires_at AS CHAR),'') FROM runtime_external_payload_backup_hold").await +} + +#[tokio::test] +async fn the_offline_command_releases_a_restored_hold_before_any_server_starts() { + let temp = tempfile::tempdir().unwrap(); + let (database, runtime, router) = setup(ExternalPayloadConfig::default(), &temp.path().join("objects")).await; + + // Native schemas start without the hold row. Inspect reports no hold and + // still leaves no row. + let (ok, body, stderr) = console(&database, &["inspect"]).await; + assert!(ok, "{stderr}"); + assert_eq!(body["schema"], "durable-workflow.external-payload-backup-hold.v1"); + assert_eq!([&body["active"], &body["owner"], &body["expires_at"]], [&json!(false), &Value::Null, &Value::Null]); + assert!(body["database_time"].is_string(), "{body}"); + assert_eq!(hold_rows(&database).await, Vec::<(String, String)>::new(), "inspect created the hold row"); + + // A restored snapshot carries the capturing backup's hold, long expired. + let (code, body) = change(&router, "acquire", json!({"owner": OWNER, "ttl": 3600})).await; + assert_eq!(code, StatusCode::OK, "{body}"); + database.execute("UPDATE runtime_external_payload_backup_hold SET acquired_at='2000-01-01 00:00:00',expires_at='2000-01-01 01:00:00'").await; + // The cell is offline: no Server process holds the database. + drop(router); + runtime.close().await; + + let restored = hold_rows(&database).await; + let (ok, body, stderr) = console(&database, &["inspect"]).await; + assert!(ok, "{stderr}"); + assert_eq!([&body["active"], &body["owner"], &body["acquired_at"], &body["expires_at"]], + [&json!(false), &json!(OWNER), &json!("2000-01-01 00:00:00"), &json!("2000-01-01 01:00:00")]); + assert_eq!(hold_rows(&database).await, restored, "inspect changed the hold row"); + + // Only the recorded owner releases it, in either option form, and a + // second release reports the same released hold. + assert_eq!(console(&database, &["release", "--owner", OTHER]).await.2, "backup_hold_not_owned"); + assert_eq!(hold_rows(&database).await, restored); + let owner_option = format!("--owner={OWNER}"); + for args in [vec!["release", "--owner", OWNER], vec!["release", owner_option.as_str()]] { + let (ok, body, stderr) = console(&database, &args).await; + assert!(ok, "{stderr}"); + assert_eq!([&body["active"], &body["owner"], &body["acquired_at"], &body["expires_at"]], + [&json!(false), &json!(OWNER), &json!("2000-01-01 00:00:00"), &Value::Null]); + } + let (ok, body, _) = console(&database, &["inspect"]).await; + assert!(ok); + assert_eq!([&body["owner"], &body["expires_at"]], [&json!(OWNER), &Value::Null]); + + // Only release and inspect exist offline; everything else is refused + // before the database is opened. + for (args, refusal) in [ + (vec!["release"], "backup_hold_command_invalid"), + (vec!["release", "--owner"], "backup_hold_command_invalid"), + (vec!["release", "--owner", OWNER, "--owner", OTHER], "backup_hold_command_invalid"), + (vec!["release", "--owner", OWNER, "--ttl", "60"], "backup_hold_command_invalid"), + (vec!["release", "--force"], "backup_hold_command_invalid"), + (vec!["release", "--owner", "not-a-uuid"], "The owner field must be a valid UUID."), + (vec!["inspect", "--owner", OWNER], "backup_hold_command_invalid"), + (vec!["acquire", "--owner", OTHER], "backup_hold_action_invalid"), + (vec!["renew", "--owner", OWNER], "backup_hold_action_invalid"), + (vec!["status"], "backup_hold_action_invalid"), + (vec![], "backup_hold_action_invalid"), + ] { + let (ok, body, stderr) = console(&database, &args).await; + assert!(!ok && body.is_null(), "{args:?} succeeded: {body}"); + assert_eq!(stderr, refusal, "{args:?}"); + } +} + +#[tokio::test] +async fn the_offline_command_never_initializes_a_database() { + // A database with no native schema is refused, and left without one. + let fresh = TestDatabase::new().await; + for args in [["inspect"].as_slice(), &["release", "--owner", OWNER]] { + let (ok, _, stderr) = console(&fresh, args).await; + assert!(!ok); + assert_eq!(stderr, "native_schema_not_initialized", "{args:?}"); + } + match &fresh { + TestDatabase::Sqlite(dir) => assert!(!dir.path().join("runtime.sqlite").exists(), "the command created a database"), + _ => assert_eq!(table_counts(&fresh).await, Vec::<(String, i64)>::new(), "the command created tables"), + } + // Counterfactual: the server's own open initializes that same database, + // and the command then serves it. + fresh.open().await.unwrap().close().await; + let (ok, body, stderr) = console(&fresh, &["inspect"]).await; + assert!(ok, "{stderr}"); + assert_eq!(body["owner"], Value::Null); + + // A database that changes under the command, losing its marker or part of + // its schema, is refused and never repaired: there is no bootstrap path. + database_refuses_without_repair(&fresh, "DROP TABLE dw_query_cache", "dw_query_cache").await; + database_refuses_without_repair(&fresh, "DROP TABLE dw_server_schema", "dw_server_schema").await; + + // An existing SQLite file that is not a native database is refused + // unchanged. + let temp = tempfile::tempdir().unwrap(); + let empty = temp.path().join("empty.sqlite"); + std::fs::write(&empty, b"").unwrap(); + for args in [["inspect"].as_slice(), &["release", "--owner", OWNER]] { + let mut command = std::process::Command::new(env!("CARGO_BIN_EXE_durable-workflow-server")); + command.env_clear().env("DW_RUST_EXPERIMENTAL", "1").env("DB_CONNECTION", "sqlite").env("DB_DATABASE", &empty) + .arg("backup-hold").args(args); + let output = tokio::task::spawn_blocking(move || command.output().unwrap()).await.unwrap(); + assert!(!output.status.success()); + assert_eq!(first_line(&output.stderr), "native_schema_not_initialized"); + assert_eq!(std::fs::metadata(&empty).unwrap().len(), 0, "the command wrote to a non-native database"); + } +} + +/// Change `database` with `sql`, then check the command refuses it and leaves +/// `table` missing. +async fn database_refuses_without_repair(database: &TestDatabase, sql: &'static str, table: &str) { + database.execute(sql).await; + for args in [["inspect"].as_slice(), &["release", "--owner", OWNER]] { + let (ok, body, stderr) = console(database, args).await; + assert!(!ok && body.is_null(), "{sql}: {args:?} succeeded: {body}"); + assert!(!stderr.is_empty(), "{sql}: no refusal reason"); + } + let tables: Vec = table_counts(database).await.into_iter().map(|(name, _)| name).collect(); + assert!(!tables.iter().any(|name| name == table), "{sql}: the command recreated {table}"); +}