diff --git a/.github/branch-protection.json b/.github/branch-protection.json index a5a6f605..9156db88 100644 --- a/.github/branch-protection.json +++ b/.github/branch-protection.json @@ -1,5 +1,5 @@ { - "_comment": "Branch protection policy for main. Applied via scripts/apply-branch-protection.sh. See docs/dev/branch-protection.md for rationale. CODEOWNERS was removed (2-person team where both maintainers own everything, so code-owner review added friction without value). Review is no longer code-owner-scoped and no approvals are required; the reporting PR checks below are the gate. Graph Vocabulary Guard currently reports a successful PR skip while its full audit runs post-merge, on tags, and by manual dispatch. The full workspace suite, the format fence, and the RustFS S3 suite run on pull requests as reporting contexts and again after merge; the Azurite suite runs post-merge; immutable workflow refs are checked on pull requests. A red main branch is stop-the-line until fixed or reverted.", + "_comment": "Branch protection policy for main. Applied via scripts/apply-branch-protection.sh. See docs/dev/branch-protection.md for rationale. CODEOWNERS was removed (2-person team where both maintainers own everything, so code-owner review added friction without value). Review is no longer code-owner-scoped and no approvals are required; the reporting PR checks below are the gate. Graph Vocabulary Guard currently reports a successful PR skip while its full audit runs post-merge, on tags, and by manual dispatch. The full workspace suite, the format fence, and the RustFS S3 suite run on pull requests as reporting contexts and again after merge; the Azurite suite runs post-merge; immutable workflow refs are checked on pull requests. A red main branch is stop-the-line until fixed or reverted. Storage Upgrade Compatibility is a required context on every change, including documentation-only pull requests.", "required_status_checks": { "strict": true, "contexts": [ @@ -11,7 +11,8 @@ "Format (rustfmt)", "Lint (clippy)", "GQ Logic Tests", - "Fix Regression Gate" + "Fix Regression Gate", + "Storage Upgrade Compatibility" ] }, "enforce_admins": false, diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d67d65b7..2994d3c1 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -161,6 +161,9 @@ jobs: - name: Enforce release vocabulary gates run: python3 scripts/check-release-vocabulary-gates.py + - name: Check required storage upgrade coverage + run: python3 scripts/check-storage-upgrade-ci.py --self-test + azure_contract_guards: name: Azure Contract Guards runs-on: ubuntu-latest @@ -602,6 +605,86 @@ jobs: || { echo "::error::exact v6 format fence did not pass"; exit 1; } + + storage_upgrade_compatibility: + name: Storage Upgrade Compatibility + runs-on: ubuntu-latest + timeout-minutes: 60 + permissions: + contents: read + env: + CARGO_TERM_COLOR: always + RUST_MIN_STACK: 16777216 + OMNIGRAPH_REQUIRE_STORAGE_UPGRADE_TESTS: '1' + steps: + - name: Checkout source + uses: actions/checkout@93cb6efe18208431cddfb8368fd83d5badbf9bfd # v5.0.1 + + - name: Install system dependencies + run: | + sudo apt-get update + sudo apt-get install -y protobuf-compiler libprotobuf-dev + + - name: Install pinned toolchain + run: rustup toolchain install + + - name: Cache Rust build data + uses: Swatinem/rust-cache@e18b497796c12c097a38f9edb9d0641fb99eee32 # v2 + with: + workspaces: | + . -> target + save-if: ${{ github.ref == 'refs/heads/main' }} + cache-on-failure: true + + - name: Check storage compatibility gate configuration + run: python3 scripts/check-storage-upgrade-ci.py --self-test + + - name: Install genuine v0.9 and v0.10 migration predecessors + run: | + set -euo pipefail + v09_dir="$RUNNER_TEMP/omnigraph-v09" + v6_dir="$RUNNER_TEMP/omnigraph-v010" + REPO_SLUG=ModernRelay/omnigraph VERSION=v0.9.0 INSTALL_DIR="$v09_dir" \ + bash scripts/install.sh + REPO_SLUG=ModernRelay/omnigraph VERSION=v0.10.0 INSTALL_DIR="$v6_dir" \ + bash scripts/install.sh + test -x "$v09_dir/omnigraph" + test -x "$v6_dir/omnigraph" + [[ "$("$v09_dir/omnigraph" --version)" == "omnigraph 0.9.0" ]] \ + || { echo "::error::storage upgrade requires the genuine v0.9.0 CLI"; exit 1; } + [[ "$("$v6_dir/omnigraph" --version)" == "omnigraph 0.10.0" ]] \ + || { echo "::error::storage upgrade requires the genuine v0.10.0 CLI"; exit 1; } + echo "OMNIGRAPH_V09_BIN=$v09_dir/omnigraph" >> "$GITHUB_ENV" + echo "OMNIGRAPH_V6_BIN=$v6_dir/omnigraph" >> "$GITHUB_ENV" + + - name: Run required storage upgrade crossversion tests + run: | + set -euo pipefail + test_log="$RUNNER_TEMP/storage-upgrade-crossversion.log" + cargo test --workspace --locked --test crossversion_upgrade --features "$FAILPOINT_FEATURES" storage_upgrade -- --test-threads=1 2>&1 | tee "$test_log" + python3 scripts/check-storage-upgrade-ci.py --check-log crossversion "$test_log" + + - name: Run required storage upgrade engine tests + run: | + set -euo pipefail + test_log="$RUNNER_TEMP/storage-upgrade-engine.log" + cargo test --locked -p omnigraph-engine --lib --features failpoints db::manifest::upgrade::tests -- --test-threads=1 2>&1 | tee "$test_log" + python3 scripts/check-storage-upgrade-ci.py --check-log engine "$test_log" + + - name: Run required storage upgrade lance tests + run: | + set -euo pipefail + test_log="$RUNNER_TEMP/storage-upgrade-lance.log" + cargo test --locked -p omnigraph-engine --test lance_version_columns --features failpoints -- --test-threads=1 2>&1 | tee "$test_log" + python3 scripts/check-storage-upgrade-ci.py --check-log lance "$test_log" + + - name: Run required storage upgrade protocol tests + run: | + set -euo pipefail + test_log="$RUNNER_TEMP/storage-upgrade-protocol.log" + cargo test --locked -p omnigraph-engine --test forbidden_apis --features failpoints -- --test-threads=1 2>&1 | tee "$test_log" + python3 scripts/check-storage-upgrade-ci.py --check-log protocol "$test_log" + v5_v7_format_fence: name: V5 ↔ V7 Format Fence needs: classify_changes diff --git a/crates/omnigraph-cli/src/cli.rs b/crates/omnigraph-cli/src/cli.rs index f8d2fd08..8c5aab96 100644 --- a/crates/omnigraph-cli/src/cli.rs +++ b/crates/omnigraph-cli/src/cli.rs @@ -17,7 +17,7 @@ COMMANDS BY CAPABILITY:\n \ any — run against a graph, served (--server / --profile) or embedded (--store / a \ URI): query, mutate, load, blob, branch, snapshot, export, commit, changes, schema show/apply.\n \ served — require a server: graphs (registry scope).\n \ -direct — direct storage access; reject --server (init, optimize, rebuild-full-text-indexes, \ +direct — direct storage access; reject --server (init, upgrade, optimize, rebuild-full-text-indexes, \ repair, cleanup, schema plan, lint).\n \ control — manage or inspect a cluster (cluster via --config; policy & queries via \ --cluster).\n \ @@ -295,6 +295,19 @@ pub(crate) enum Command { #[arg(long)] force: bool, }, + /// Upgrade graph storage offline using registered migration handlers + Upgrade { + /// Standalone graph storage URI; alternatively use --store + uri: Option, + /// Run read-only preflight without conversion or recovery writes + #[arg(long)] + check: bool, + /// Requested storage format (defaults to the binary's declared target) + #[arg(long, value_name = "N")] + to_format: Option, + #[arg(long)] + json: bool, + }, /// Compact small Lance fragments in every backing dataset of the graph Optimize { /// Graph URI diff --git a/crates/omnigraph-cli/src/main.rs b/crates/omnigraph-cli/src/main.rs index 2fb96a6f..3a07c4ee 100644 --- a/crates/omnigraph-cli/src/main.rs +++ b/crates/omnigraph-cli/src/main.rs @@ -55,6 +55,7 @@ mod managed_http_fixture; mod output; mod planes; mod scope; +mod upgrade; use cli::*; use helpers::*; use output::*; @@ -1357,6 +1358,23 @@ async fn main() -> Result<()> { } } } + Command::Upgrade { + uri, + check, + to_format, + json, + } => { + upgrade::run( + &cli.profile, + &cli.store, + uri, + check, + to_format, + json, + cli.quiet, + ) + .await?; + } Command::Optimize { uri, json } => { let uri = resolve_maintenance_uri( cli.profile.as_deref(), diff --git a/crates/omnigraph-cli/src/planes.rs b/crates/omnigraph-cli/src/planes.rs index 52814493..7eb1f31c 100644 --- a/crates/omnigraph-cli/src/planes.rs +++ b/crates/omnigraph-cli/src/planes.rs @@ -265,6 +265,7 @@ pub(crate) fn command_plane(cmd: &Command) -> Plane { Command::Queries { .. } => Plane::Control, Command::Policy { .. } => Plane::Control, Command::Init { .. } + | Command::Upgrade { .. } | Command::Optimize { .. } | Command::RebuildFullTextIndexes { .. } | Command::Repair { .. } @@ -320,6 +321,7 @@ pub(crate) fn command_label(cmd: &Command) -> &'static str { Command::Mutate { .. } => "mutate", Command::Alias { .. } => "alias", Command::Policy { .. } => "policy", + Command::Upgrade { .. } => "upgrade", Command::Optimize { .. } => "optimize", Command::RebuildFullTextIndexes { .. } => "rebuild-full-text-indexes", Command::Repair { .. } => "repair", diff --git a/crates/omnigraph-cli/src/upgrade.rs b/crates/omnigraph-cli/src/upgrade.rs new file mode 100644 index 00000000..05f247db --- /dev/null +++ b/crates/omnigraph-cli/src/upgrade.rs @@ -0,0 +1,197 @@ +use super::*; + +pub(crate) async fn run( + profile: &Option, + store: &Option, + uri: Option, + check: bool, + to_format: Option, + json: bool, + quiet: bool, +) -> Result<()> { + let target = scope::resolve_scope( + &operator::load_operator_config()?, + planes::Capability::Direct, + scope::ScopeFlags { + profile: profile.as_deref(), + store: store.as_deref(), + server: None, + cluster: None, + graph: None, + uri, + }, + )?; + if target.cluster.is_some() { + bail!( + "upgrade refuses cluster-managed graphs; a qualified cluster upgrade operation is required" + ); + } + let uri = resolve_local_uri(target.uri, "upgrade")?; + let uri = omnigraph::storage::normalize_root_uri(&uri)?; + let uri = if omnigraph::storage::storage_kind_for_uri(&uri)? + == omnigraph::storage::StorageKind::Local + { + std::fs::canonicalize(&uri)? + .to_str() + .ok_or_else(|| color_eyre::eyre::eyre!("upgrade path is not valid UTF-8"))? + .to_owned() + } else { + uri + }; + if let Some(root) = omnigraph_cluster::cluster_root_for_graph_uri(&uri) + .await + .map_err(|diagnostic| { + color_eyre::eyre::eyre!("{}: {}", diagnostic.path, diagnostic.message) + })? + { + bail!( + "upgrade refuses graph `{uri}` inside cluster `{root}`; a qualified cluster upgrade operation is required" + ); + } + if !check { + echo_write_target(quiet, "upgrade", &uri, false); + } + let report = + omnigraph::db::upgrade_storage(&uri, omnigraph::db::UpgradeOptions { check, to_format }) + .await?; + if json { + print_json(&report)?; + } else { + print_human(&report)?; + } + if !report.success() { + std::process::exit(1); + } + Ok(()) +} + +fn print_human(report: &omnigraph::db::UpgradeReport) -> Result<()> { + let mode = serde_json::to_value(report.mode)?; + let outcome = serde_json::to_value(report.outcome)?; + println!( + "upgrade {}: {} ({})", + report.location, + outcome.as_str().unwrap_or("unknown"), + mode.as_str().unwrap_or("unknown") + ); + println!( + "graph identity: {}", + report.graph_identity.as_deref().unwrap_or("unknown") + ); + println!( + "format: {} -> {}{}", + report + .observed_format + .map_or_else(|| "unknown".into(), |v| v.to_string()), + report.target_format, + if report.target_defaulted { + " (default target)" + } else { + "" + } + ); + println!("route: {}", report.route.join(" -> ")); + println!( + "completed handlers: {}", + report.completed_handlers.join(", ") + ); + println!( + "last durable completed boundary: {}", + report + .last_durable_completed_boundary + .as_deref() + .unwrap_or("unknown") + ); + println!( + "work: {} metadata rows, {} retained snapshots, {} payload bytes copied, {} payload bytes rewritten", + report.work.metadata_rows, + report.work.retained_snapshots, + report.work.payload_bytes_copied, + report.work.payload_bytes_rewritten + ); + println!( + "validation bytes: {}", + report + .work + .validation_bytes + .map_or_else(|| "unknown".into(), |v| v.to_string()) + ); + for exclusion in &report.work.external_blob_exclusions { + println!("external bytes excluded from preservation: {exclusion}"); + } + for property in &report.work.historical_blob_identity_limits { + println!( + "historical Blob delivery retains the pre-0.10 property-lifetime restriction: {property}" + ); + } + for finding in &report.findings { + println!("{}: {}", finding.code, finding.message); + } + if let Some(recovery) = &report.recovery { + println!("failed handler: {}", recovery.failed_handler); + println!("recovery executable: {}", recovery.executable_compatibility); + println!("recovery action: {}", recovery.action); + } + if matches!(report.mode, omnigraph::db::UpgradeMode::Check) { + println!( + "Check is advisory. Stop all writers and maintenance and retain a verified backup before execution." + ); + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn storage_upgrade_parses_explicit_route_and_check_options() { + let cli = Cli::try_parse_from([ + "omnigraph", + "upgrade", + "graph.omni", + "--check", + "--to-format", + "7", + "--json", + ]) + .unwrap(); + assert!( + matches!(&cli.command, Command::Upgrade { uri: Some(uri), check: true, to_format: Some(7), json: true } if uri == "graph.omni") + ); + assert_eq!( + planes::command_capability(&cli.command), + planes::Capability::Direct + ); + assert!(planes::guard_addressing(&cli).is_ok()); + let cli = Cli::try_parse_from(["omnigraph", "--store", "graph.omni", "upgrade"]).unwrap(); + assert!(matches!( + cli.command, + Command::Upgrade { + uri: None, + check: false, + to_format: None, + json: false + } + )); + } + + #[test] + fn storage_upgrade_rejects_served_and_cluster_addressing() { + for flag in ["--server", "--cluster", "--graph"] { + let cli = + Cli::try_parse_from(["omnigraph", flag, "prod", "upgrade", "graph.omni"]).unwrap(); + assert!(planes::guard_addressing(&cli).is_err(), "{flag}"); + } + assert!( + Cli::try_parse_from([ + "omnigraph", + "upgrade", + "graph.omni", + "--to-format", + "not-a-version" + ]) + .is_err() + ); + } +} diff --git a/crates/omnigraph-cli/tests/crossversion_upgrade.rs b/crates/omnigraph-cli/tests/crossversion_upgrade.rs index 0540f7d4..8661b97d 100644 --- a/crates/omnigraph-cli/tests/crossversion_upgrade.rs +++ b/crates/omnigraph-cli/tests/crossversion_upgrade.rs @@ -1177,3 +1177,600 @@ query revise($body: String) { update Doc set { body: $body } where slug = "dl-ba ); eprintln!("v0.9 refusal and export/import rebuild completed"); } + +fn migration_bin(variable: &str, version: &str) -> Option { + let Some(path) = std::env::var_os(variable).map(PathBuf::from) else { + assert!( + std::env::var_os("OMNIGRAPH_REQUIRE_STORAGE_UPGRADE_TESTS").is_none(), + "required storage migration predecessor {variable} is unset" + ); + eprintln!("skipping explicit storage upgrade: {variable} is unset"); + return None; + }; + assert!( + path.is_file(), + "{variable} is not a binary: {}", + path.display() + ); + let output = run_old(&path, &["version"]); + assert_ok("migration predecessor version", &output); + assert_eq!( + String::from_utf8_lossy(&output.stdout).lines().next(), + Some(version) + ); + Some(path) +} + +#[test] +fn genuine_v09_explicit_storage_upgrade_preserves_history() { + if let Some(old) = migration_bin("OMNIGRAPH_V09_BIN", "omnigraph 0.9.0") { + explicit_storage_upgrade_journey(&old, false); + } +} + +#[test] +fn genuine_v010_explicit_storage_upgrade_preserves_history() { + if let Some(old) = migration_bin("OMNIGRAPH_V6_BIN", "omnigraph 0.10.0") { + explicit_storage_upgrade_journey(&old, true); + } +} + +fn graph_files(root: &Path) -> std::collections::BTreeMap> { + let mut files = std::collections::BTreeMap::new(); + let mut pending = vec![root.to_path_buf()]; + while let Some(directory) = pending.pop() { + for entry in std::fs::read_dir(directory).unwrap() { + let entry = entry.unwrap(); + let path = entry.path(); + if entry.file_type().unwrap().is_dir() { + pending.push(path); + } else { + files.insert( + path.strip_prefix(root).unwrap().to_path_buf(), + std::fs::read(path).unwrap(), + ); + } + } + } + files +} + +fn explicit_storage_upgrade_journey(old: &Path, has_property_lifetime_metadata: bool) { + let temp = tempdir().unwrap(); + let graph = temp.path().join("standalone.omni"); + let uri = graph.to_str().unwrap(); + let schema = temp.path().join("migration.pg"); + let data = temp.path().join("migration.jsonl"); + let queries = temp.path().join("migration.gq"); + std::fs::write(&schema, format!( + "{}\nedge Cites: Doc -> Doc {{ note: String }}\nnode BinaryAsset {{ name: String @key payload: Blob }}\n", + std::fs::read_to_string(fixture("search.pg")).unwrap() + )).unwrap(); + std::fs::write(&data, format!("{}\n{}\n{}\n", + std::fs::read_to_string(fixture("search.jsonl")).unwrap().trim_end(), + r#"{"edge":"Cites","from":"ml-intro","to":"dl-basics","data":{"id":"citation-1","note":"preserved edge"}}"#, + r#"{"type":"BinaryAsset","data":{"name":"blob-sentinel","payload":"base64:AAECA/8="}}"#, + )).unwrap(); + std::fs::write( + &queries, + r#" +query docs() { + match { $d: Doc } + return { $d.slug, $d.title, $d.body, $d.embedding } + order { $d.slug } +} +query edges() { + match { $a: Doc $a $c:cites $b } + return { $a.slug, $b.slug, $c.note } +} +query retitle($title: String) { update Doc set { title: $title } where slug = "ml-intro" } +query remove() { delete Doc where slug = "rl-intro" } +query revise() { update Doc set { body: "written after storage upgrade" } where slug = "dl-basics" } +query terms() { + match { $d: Doc search($d.title, "organism") } + return { $d.slug } + order { $d.slug } +} +query vectors($q: Vector(4)) { + match { $d: Doc } + return { $d.slug } + order { nearest($d.embedding, $q) } + limit 1 +} +"#, + ) + .unwrap(); + let query_path = queries.to_str().unwrap(); + assert_ok( + "migration init", + &run_old(old, &["init", "--schema", schema.to_str().unwrap(), uri]), + ); + assert_ok( + "migration load", + &run_old( + old, + &[ + "load", + "--mode", + "overwrite", + "--data", + data.to_str().unwrap(), + uri, + ], + ), + ); + for (branch, title) in [ + ("main", r#"{"title":"organism main"}"#), + ("review", r#"{"title":"organism review"}"#), + ] { + if branch == "review" { + assert_ok( + "migration branch", + &run_old(old, &["branch", "create", "review", "--uri", uri]), + ); + } + assert_ok( + "migration update", + &run_old( + old, + &[ + "mutate", "retitle", "--query", query_path, "--store", uri, "--branch", branch, + "--params", title, + ], + ), + ); + } + assert_ok( + "migration deletion", + &run_old( + old, + &[ + "mutate", "remove", "--query", query_path, "--store", uri, "--branch", "review", + ], + ), + ); + + let old_query = |selector: &str, value: &str, name: &str| { + let output = run_old( + old, + &[ + "query", name, "--query", query_path, "--store", uri, selector, value, "--json", + ], + ); + assert_ok("source snapshot query", &output); + let mut rows = support::parse_stdout_json(&output)["rows"].clone(); + normalize_f32_and_nulls(&mut rows); + rows + }; + let histories: Vec<_> = ["main", "review"] + .into_iter() + .map(|branch| { + let output = run_old(old, &["commit", "list", uri, "--branch", branch, "--json"]); + assert_ok("migration source commits", &output); + let mut commits = support::parse_stdout_json(&output)["commits"].clone(); + for commit in commits.as_array_mut().unwrap() { + let fields = commit.as_object_mut().unwrap(); + for (old, current) in [ + ("manifest_branch", "graph_branch"), + ("manifest_version", "graph_manifest_version"), + ] { + if let Some(value) = fields.remove(old) { + assert!(fields.insert(current.to_string(), value).is_none()); + } + } + } + commits + }) + .collect(); + let mut historical_rows = std::collections::BTreeMap::new(); + for history in &histories { + for commit in history.as_array().unwrap() { + let id = commit["graph_commit_id"].as_str().unwrap(); + historical_rows.entry(id.to_owned()).or_insert_with(|| { + [ + old_query("--snapshot", id, "docs"), + old_query("--snapshot", id, "edges"), + ] + }); + } + } + let exports: Vec<_> = ["main", "review"] + .into_iter() + .map(|branch| { + let output = run_old(old, &["export", uri, "--branch", branch]); + assert_ok("migration source export", &output); + canonical_export_rows(&output.stdout) + }) + .collect(); + let before = graph_files(&graph); + output_failure(cli().args(["snapshot", uri])); + let check = support::parse_stdout_json(&output_success(cli().args([ + "upgrade", + uri, + "--check", + "--to-format", + "7", + "--json", + ]))); + assert_eq!(check["outcome"], "check_passed"); + assert_eq!( + graph_files(&graph), + before, + "--check and refused open must leave every source byte unchanged" + ); + let upgraded = support::parse_stdout_json(&output_success(cli().args([ + "upgrade", + uri, + "--to-format", + "7", + "--json", + ]))); + assert_eq!(upgraded["outcome"], "completed"); + let after = graph_files(&graph); + let schema_identity = Path::new("_schema.ir.json"); + assert!(before.contains_key(schema_identity)); + assert_eq!( + after.get(schema_identity), + before.get(schema_identity), + "storage upgrade must preserve accepted schema identities exactly" + ); + let payloads = |files: &std::collections::BTreeMap>| { + files + .iter() + .filter(|(path, _)| path.starts_with("nodes") || path.starts_with("edges")) + .map(|(path, bytes)| (path.clone(), bytes.clone())) + .collect::>() + }; + assert!(!payloads(&before).is_empty()); + assert_eq!( + payloads(&after), + payloads(&before), + "storage migration must preserve every table object without adding table objects" + ); + for check_mode in [false, true] { + let mut command = cli(); + command.args(["upgrade", uri, "--json"]); + if check_mode { + command.arg("--check"); + } + let report = support::parse_stdout_json(&output_success(&mut command)); + assert_eq!(report["outcome"], "already_current"); + assert_eq!(graph_files(&graph), after, "rerun must be effect-free"); + } + assert!( + !run_old(old, &["snapshot", uri]).status.success(), + "predecessor writer must refuse upgraded graph" + ); + + let current_query = |selector: &str, value: &str, name: &str| { + let params = if name == "vectors" { + r#"{"q":[0.1,0.2,0.3,0.4]}"# + } else { + "{}" + }; + let output = output_success(cli().args([ + "query", name, "--query", query_path, "--store", uri, selector, value, "--params", + params, "--json", + ])); + let mut rows = support::parse_stdout_json(&output)["rows"].clone(); + normalize_f32_and_nulls(&mut rows); + rows + }; + let check_history = |after_maintenance: bool| { + tokio::runtime::Runtime::new().unwrap().block_on(async { + let db = Omnigraph::open(uri).await.unwrap(); + for history in &histories { + for commit in history.as_array().unwrap() { + let branch = commit["graph_branch"].as_str().unwrap_or("main"); + db.sync_branch(branch).await.unwrap(); + let version = commit["graph_manifest_version"].as_u64().unwrap(); + let numeric = db + .snapshot_at_graph_manifest_version(version) + .await + .unwrap(); + let by_id = db + .snapshot_of(ReadTarget::snapshot(omnigraph::db::SnapshotId::new( + commit["graph_commit_id"].as_str().unwrap(), + ))) + .await + .unwrap(); + assert_eq!(numeric.graph_manifest_version(), version); + assert_eq!(numeric.datasets().count(), by_id.datasets().count()); + for entry in numeric.datasets() { + assert!( + entry.same_registration(by_id.dataset(&entry.type_key).unwrap()), + "numeric snapshot and commit selector disagree at {branch}/{version}" + ); + if entry.type_key == "node:BinaryAsset" && entry.entity_count != 0 { + let table_uri = format!("{uri}/{}", entry.dataset_path); + let table = lance::Dataset::open(&table_uri).await.unwrap(); + let table = if let Some(native) = &entry.native_dataset_branch { + table.checkout_branch(native).await.unwrap() + } else { + table + }; + let table = std::sync::Arc::new( + table + .checkout_version(entry.published_dataset_version) + .await + .unwrap(), + ); + assert_eq!(table.count_rows(None).await.unwrap(), 1); + assert_eq!( + table + .schema() + .field("payload") + .unwrap() + .metadata + .contains_key("omnigraph.stable_property_id"), + has_property_lifetime_metadata + ); + let blobs = table.take_blobs_by_indices(&[0], "payload").await.unwrap(); + let blob = blobs[0].as_ref().unwrap(); + assert_eq!( + blob.read().await.unwrap().as_ref(), + [0, 1, 2, 3, 255], + "retained Blob bytes at {branch}/{version}" + ); + } + } + } + } + }); + for (id, expected) in &historical_rows { + assert_eq!( + current_query("--snapshot", id, "docs"), + expected[0], + "retained docs at {id}" + ); + assert_eq!( + current_query("--snapshot", id, "edges"), + expected[1], + "retained edges at {id}" + ); + if !expected[0].as_array().unwrap().is_empty() { + let deliverable = tokio::runtime::Runtime::new().unwrap().block_on(async { + let db = Omnigraph::open(uri).await.unwrap(); + match db + .read_blob_at( + ReadTarget::snapshot(omnigraph::db::SnapshotId::new(id)), + BlobCell { + entity: EntityKind::Node, + type_name: "BinaryAsset".into(), + id: "blob-sentinel".into(), + property: "payload".into(), + }, + ) + .await + { + Ok(_) => true, + Err(error) => { + assert!( + after_maintenance && !has_property_lifetime_metadata, + "historical Blob {id}: {error:?}" + ); + assert!( + error + .to_string() + .contains("no persisted property-lifetime witness"), + "{error:?}" + ); + false + } + } + }); + if !deliverable { + let refused = output_failure(cli().args([ + "blob", + "get", + "node", + "BinaryAsset", + "blob-sentinel", + "payload", + "--store", + uri, + "--snapshot", + id, + ])); + assert!( + String::from_utf8_lossy(&refused.stderr).contains("invalid Blob selector") + ); + continue; + } + let blob = output_success(cli().args([ + "blob", + "get", + "node", + "BinaryAsset", + "blob-sentinel", + "payload", + "--store", + uri, + "--snapshot", + id, + ])); + assert_eq!(blob.stdout, [0, 1, 2, 3, 255], "retained blob at {id}"); + } + } + }; + for (index, branch) in ["main", "review"].into_iter().enumerate() { + let exported = output_success(cli().args(["export", uri, "--branch", branch])); + assert_eq!(canonical_export_rows(&exported.stdout), exports[index]); + let history = support::parse_stdout_json(&output_success( + cli().args(["commit", "list", uri, "--branch", branch, "--json"]), + )); + for commit in histories[index].as_array().unwrap() { + assert!( + history["commits"].as_array().unwrap().contains(commit), + "retained commit identity and ancestry changed: {commit}" + ); + } + let blob = output_success(cli().args([ + "blob", + "get", + "node", + "BinaryAsset", + "blob-sentinel", + "payload", + "--store", + uri, + "--branch", + branch, + ])); + assert_eq!(blob.stdout, [0, 1, 2, 3, 255]); + assert_eq!( + current_query("--branch", branch, "vectors"), + serde_json::json!([{"d.slug":"ml-intro"}]) + ); + } + check_history(false); + for branch in ["main", "review"] { + output_success(cli().args([ + "rebuild-full-text-indexes", + uri, + "--branch", + branch, + "--json", + ])); + assert_eq!( + current_query("--branch", branch, "terms"), + serde_json::json!([{"d.slug":"ml-intro"}]) + ); + } + output_success(cli().args([ + "mutate", "revise", "--query", query_path, "--store", uri, "--branch", "review", + ])); + output_success(cli().args([ + "branch", "merge", "review", "--into", "main", "--store", uri, "--json", + ])); + let merged = current_query("--branch", "main", "docs"); + assert!(merged.as_array().unwrap().iter().any( + |row| row["d.slug"] == "dl-basics" && row["d.body"] == "written after storage upgrade" + )); + assert!( + !merged + .as_array() + .unwrap() + .iter() + .any(|row| row["d.slug"] == "rl-intro") + ); + output_success(cli().args([ + "cleanup", + uri, + "--older-than", + "7d", + "--confirm", + "--yes", + "--json", + ])); + check_history(true); + assert_eq!(current_query("--branch", "main", "docs"), merged); + std::fs::remove_dir_all(&graph).unwrap(); + let restored = &graph; + for (path, bytes) in &before { + let destination = restored.join(path); + std::fs::create_dir_all(destination.parent().unwrap()).unwrap(); + std::fs::write(destination, bytes).unwrap(); + } + for (index, branch) in ["main", "review"].into_iter().enumerate() { + let output = run_old( + old, + &["export", restored.to_str().unwrap(), "--branch", branch], + ); + assert_ok("whole-root backup restore", &output); + assert_eq!(canonical_export_rows(&output.stdout), exports[index]); + } +} + +#[test] +fn storage_upgrade_refuses_cluster_path_aliases() { + let temp = tempdir().unwrap(); + let cluster = temp.path().join("cluster"); + let graph = cluster.join("graphs/kb.omni"); + let schema = temp.path().join("schema.pg"); + std::fs::write(&schema, "node A { name: String @key }").unwrap(); + output_success(cli().args(["init", "--schema"]).arg(&schema).arg(&graph)); + std::fs::create_dir_all(cluster.join("__cluster")).unwrap(); + std::fs::write(cluster.join("__cluster/state.json"), "{}").unwrap(); + let before = graph_files(&cluster); + let mut aliases = vec![ + graph.to_str().unwrap().to_owned(), + "graphs/kb.omni".to_owned(), + "./graphs/kb.omni".to_owned(), + "graphs/kb.omni/.".to_owned(), + url::Url::from_file_path(&graph).unwrap().to_string(), + ]; + #[cfg(unix)] + { + let alias = temp.path().join("alias.omni"); + std::os::unix::fs::symlink(&graph, &alias).unwrap(); + aliases.push(alias.to_str().unwrap().to_owned()); + } + for alias in aliases { + for check in [true, false] { + let mut command = cli(); + command + .current_dir(&cluster) + .args(["upgrade", &alias, "--json"]); + if check { + command.arg("--check"); + } + let output = output_failure(&mut command); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!(stderr.contains("inside cluster"), "{alias}: {stderr}"); + assert_eq!( + graph_files(&cluster), + before, + "{alias} must refuse before effects" + ); + } + } +} + +#[test] +fn genuine_v09_storage_upgrade_refuses_ambiguous_branch_names() { + let Some(old) = migration_bin("OMNIGRAPH_V09_BIN", "omnigraph 0.9.0") else { + return; + }; + for sibling in [false, true] { + let temp = tempdir().unwrap(); + let graph = temp.path().join("legacy.omni"); + let uri = graph.to_str().unwrap(); + let schema = temp.path().join("schema.pg"); + std::fs::write(&schema, "node A { name: String @key }").unwrap(); + assert_ok( + "legacy branch init", + &run_old(&old, &["init", "--schema", schema.to_str().unwrap(), uri]), + ); + if sibling { + assert_ok( + "legacy sibling", + &run_old(&old, &["branch", "create", "feature", "--uri", uri]), + ); + } + let name = "feature.01ARZ3NDEKTSV4RRFFQ69G5FAV"; + assert_ok( + "legacy suffixed branch", + &run_old(&old, &["branch", "create", name, "--uri", uri]), + ); + let before = graph_files(&graph); + for check in [true, false] { + let mut command = cli(); + command.args(["upgrade", uri, "--json"]); + if check { + command.arg("--check"); + } + let report = support::parse_stdout_json(&output_failure(&mut command)); + assert_eq!(report["outcome"], "check_failed"); + assert!( + report["findings"].to_string().contains("branch identity"), + "{report}" + ); + assert_eq!(graph_files(&graph), before); + assert_ok( + "legacy source still opens", + &run_old(&old, &["snapshot", uri, "--branch", name]), + ); + } + } +} diff --git a/crates/omnigraph/src/db/manifest.rs b/crates/omnigraph/src/db/manifest.rs index 5ca4b22b..e19f8782 100644 --- a/crates/omnigraph/src/db/manifest.rs +++ b/crates/omnigraph/src/db/manifest.rs @@ -32,6 +32,12 @@ mod publisher; mod recovery; #[path = "manifest/state.rs"] mod state; +#[path = "manifest/upgrade.rs"] +mod upgrade; +pub use upgrade::{ + UpgradeFinding, UpgradeMode, UpgradeOptions, UpgradeOutcome, UpgradeRecovery, UpgradeReport, + UpgradeWork, upgrade_storage, upgrade_storage_as, +}; pub(crate) use graph::{GenesisManifestAttempt, ManifestInitError}; use graph::{ diff --git a/crates/omnigraph/src/db/manifest/migrations.rs b/crates/omnigraph/src/db/manifest/migrations.rs index 5964a8e8..9ab15a03 100644 --- a/crates/omnigraph/src/db/manifest/migrations.rs +++ b/crates/omnigraph/src/db/manifest/migrations.rs @@ -13,27 +13,13 @@ //! - One guard `refuse_if_stamp_unsupported` rejects any graph this binary //! cannot serve — in either direction — with a clear, actionable error. //! -//! ## Single-version contract (strand + export/import) +//! ## Explicit conversion and normal-open contract //! -//! This binary reads exactly ONE internal-schema version (`MIN_SUPPORTED == -//! CURRENT`). There is no in-place migration: a graph stamped below CURRENT is -//! refused on open with a "rebuild via `omnigraph export` + `init`/`load`" -//! message, not silently upgraded. This is the deliberate pre-release contract — -//! storage-format changes are a cutover, not a rolling in-place migration (see -//! `docs/user/operations/upgrade.md` and the versioning policy in `docs/dev`). -//! Fresh graphs are stamped at CURRENT *inside* the init `Dataset::write` -//! Create commit (`current_stamp_entry` rides the write's schema metadata), so -//! the stamp is atomic with manifest birth: no crash can leave `__manifest` -//! durable but unstamped. -//! -//! ## If an in-place migration is ever needed -//! -//! The stamp + `refuse_if_stamp_unsupported` are the seam a future migration -//! would plug into: re-introduce a dispatcher that walks the stamp forward and -//! lower `MIN_SUPPORTED` below CURRENT for exactly the versions it can upgrade. -//! Until a concrete graph demands it, that machinery is unearned complexity and -//! is deliberately absent. A future converter is best shaped as a standalone -//! one-shot tool, not a framework baked into the open path. +//! Normal open accepts only CURRENT and refuses an active storage-upgrade intent. +//! The explicit offline upgrade entry point converts supported v6 graphs before +//! serving. Retained v6 snapshots use the legacy decoder after root admission; +//! normal open never runs conversion or lowers MIN_SUPPORTED. +//! Fresh graphs receive their stamp atomically in the manifest Create commit. //! //! ## Forward-version protection //! @@ -75,14 +61,8 @@ use crate::error::{OmniError, Result}; /// is kept for provenance and to document what each stamp value meant. pub(crate) const INTERNAL_MANIFEST_SCHEMA_VERSION: u32 = 7; -/// The oldest on-disk internal-schema stamp this binary will open. With no -/// in-place migration, this equals `INTERNAL_MANIFEST_SCHEMA_VERSION`: a graph -/// stamped below it is refused (`refuse_if_stamp_unsupported`) with a -/// rebuild-via-export/import message rather than silently upgraded. -/// -/// Lowering this below CURRENT only makes sense alongside a re-introduced -/// migration dispatcher that can actually walk those versions forward (see the -/// module doc). +/// The oldest main-manifest stamp accepted by normal open. +/// Explicit conversion and retained-snapshot decoding do not lower this gate. pub(crate) const MIN_SUPPORTED_INTERNAL_SCHEMA_VERSION: u32 = INTERNAL_MANIFEST_SCHEMA_VERSION; /// The omnigraph release or exact development build that wrote a given @@ -111,7 +91,7 @@ pub(crate) fn release_for_internal_schema_version(stamp: u32) -> &'static str { } } -const INTERNAL_SCHEMA_VERSION_KEY: &str = "omnigraph:internal_schema_version"; +pub(super) const INTERNAL_SCHEMA_VERSION_KEY: &str = "omnigraph:internal_schema_version"; /// The schema-metadata entry stamping a fresh manifest at CURRENT. Folded into /// the Arrow schema of init's `Dataset::write` so the stamp lands in the same @@ -158,6 +138,15 @@ pub(crate) fn read_stamp(dataset: &Dataset) -> Option { /// treated as v1 and refused through the ordinary sub-floor message naming /// the 0.3.1 export path. pub(crate) fn guard_stamp(dataset: &Dataset) -> Result { + if dataset + .schema() + .metadata + .contains_key(super::upgrade::UPGRADE_PENDING_KEY) + { + return Err(OmniError::manifest( + "storage upgrade recovery required: stop all writers and maintenance, then rerun the same `omnigraph upgrade --to-format 7` command with the upgrade-capable executable", + )); + } match dataset.schema().metadata.get(INTERNAL_SCHEMA_VERSION_KEY) { Some(value) => match value.parse::() { Ok(stamp) => { @@ -214,6 +203,11 @@ pub(crate) fn refuse_if_stamp_unsupported(stamp: u32) -> Result<()> { ))); } if stamp < MIN_SUPPORTED_INTERNAL_SCHEMA_VERSION { + let explicit_upgrade = if stamp == 6 { + " A registered in-place route is also available: stop all writers and maintenance, retain a verified backup, and run `omnigraph upgrade --check --to-format 7` before execution." + } else { + "" + }; return Err(OmniError::manifest(format!( "__manifest is stamped at internal schema v{stamp}, but this omnigraph reads only v{current}. \ This graph was created by omnigraph {release}. Rebuild it: with an omnigraph {release} binary run \ @@ -221,7 +215,7 @@ pub(crate) fn refuse_if_stamp_unsupported(stamp: u32) -> Result<()> { `omnigraph init --schema ` and \ `omnigraph load --mode overwrite --data graph.jsonl `. \ (Data, vectors, and blobs are preserved; commit history and branches are not.) \ - See docs/user/operations/upgrade.md.", + See docs/user/operations/upgrade.md.{explicit_upgrade}", current = INTERNAL_MANIFEST_SCHEMA_VERSION, release = release_for_internal_schema_version(stamp), ))); diff --git a/crates/omnigraph/src/db/manifest/state.rs b/crates/omnigraph/src/db/manifest/state.rs index fd4e0f8e..0749922f 100644 --- a/crates/omnigraph/src/db/manifest/state.rs +++ b/crates/omnigraph/src/db/manifest/state.rs @@ -162,6 +162,18 @@ pub(super) fn manifest_schema() -> SchemaRef { ])) } +pub(super) async fn read_manifest_state_with_registration_clocks( + dataset: &Dataset, +) -> Result { + if super::migrations::read_stamp(dataset) != Some(6) { + return Err(OmniError::manifest_internal( + "registration-clock conversion requires a v6 source".to_string(), + )); + } + let scan = read_manifest_scan_with_clocks(dataset, false, None, true).await?; + manifest_state_from_scan(dataset.version().version, scan) +} + pub(super) async fn read_manifest_state(dataset: &Dataset) -> Result { let version = dataset.version().version; // The table-state hot path never needs lineage, so don't pay its JSON decode. @@ -673,6 +685,30 @@ fn require_clock_at_or_below(clock: u64, dataset: &Dataset, table_key: &str) -> Ok(()) } +fn registration_clock( + dataset: &Dataset, + legacy: bool, + key_version: u64, + table_version: u64, + update_versions: Option<&UInt64Array>, + row: usize, + table_key: &str, +) -> Result { + if legacy && key_version != table_version { + return Err(OmniError::manifest_internal(format!( + "v6 manifest row for {table_key} has key version {key_version}, expected table version {table_version}" + ))); + } + let clock = match update_versions { + Some(versions) => required_u64(versions, row, "_row_last_updated_at_version")?, + None => key_version, + }; + if !legacy || update_versions.is_some() { + require_clock_at_or_below(clock, dataset, table_key)?; + } + Ok(clock) +} + async fn read_manifest_scan(dataset: &Dataset, collect_lineage: bool) -> Result { read_manifest_scan_fragments(dataset, collect_lineage, None).await } @@ -686,12 +722,22 @@ async fn read_manifest_scan_fragments( collect_lineage: bool, fragments: Option>, ) -> Result { + read_manifest_scan_with_clocks(dataset, collect_lineage, fragments, false).await +} + +async fn read_manifest_scan_with_clocks( + dataset: &Dataset, + collect_lineage: bool, + fragments: Option>, + use_row_update_versions: bool, +) -> Result { + let legacy = super::migrations::read_stamp(dataset) == Some(6); crate::instrumentation::record_manifest_scan(); // Project only the columns the assembly below reads (RFC-013 PR2 #1c). The // `object_id` is needed for the bounded graph-head authority decode on every // path; `base_objects` remains reserved/unused. Mirrors Lance's own // directory-catalog `__manifest` reads, which project only needed columns. - let projection: Vec<&str> = vec![ + let mut projection: Vec<&str> = vec![ "object_id", "object_type", "location", @@ -703,6 +749,9 @@ async fn read_manifest_scan_fragments( "table_branch", "row_count", ]; + if use_row_update_versions { + projection.push("_row_last_updated_at_version"); + } let is_delta_scan = fragments.is_some(); let mut scanner = dataset.scan(); scanner.project(&projection).map_err(OmniError::storage)?; @@ -736,6 +785,9 @@ async fn read_manifest_scan_fragments( let versions = u64_column(batch, "table_version")?; let branches = string_column(batch, "table_branch")?; let row_counts = u64_column(batch, "row_count")?; + let update_versions = use_row_update_versions + .then(|| u64_column(batch, "_row_last_updated_at_version")) + .transpose()?; // `object_id` is needed for the exact graph-head authority even on the // table-state path. We still skip every `graph_commit` decode there, so // the added work is bounded by the number of branch-head rows rather @@ -795,12 +847,20 @@ async fn read_manifest_scan_fragments( OBJECT_TYPE_TABLE_VERSION, )?; let table_version = required_u64(versions, row, "table_version")?; - let manifest_version = manifest_version_from_object_id( + let key_version = manifest_version_from_object_id( object_ids.value(row), identity, OBJECT_TYPE_TABLE_VERSION, )?; - require_clock_at_or_below(manifest_version, dataset, &table_key)?; + let manifest_version = registration_clock( + dataset, + legacy, + key_version, + table_version, + update_versions, + row, + &table_key, + )?; let row_count = required_u64(row_counts, row, "row_count")?; if metadata.is_null(row) { return Err(OmniError::manifest_internal(format!( @@ -832,12 +892,20 @@ async fn read_manifest_scan_fragments( OBJECT_TYPE_TABLE_TOMBSTONE, )?; let tombstone_version = required_u64(versions, row, "table_version")?; - let manifest_version = manifest_version_from_object_id( + let key_version = manifest_version_from_object_id( object_ids.value(row), identity, OBJECT_TYPE_TABLE_TOMBSTONE, )?; - require_clock_at_or_below(manifest_version, dataset, &table_key)?; + let manifest_version = registration_clock( + dataset, + legacy, + key_version, + tombstone_version, + update_versions, + row, + &table_key, + )?; tombstones.push(TableTombstoneEntry { identity, table_key, @@ -900,7 +968,7 @@ async fn read_manifest_scan_fragments( .map(|tombstone| (tombstone.identity, tombstone.manifest_version)), ) { - if !clocks.insert((identity, clock)) { + if !clocks.insert((identity, clock)) && (!legacy || use_row_update_versions) { return Err(OmniError::manifest_internal(format!( "manifest has two rows for identity {identity} at manifest version {clock}" ))); diff --git a/crates/omnigraph/src/db/manifest/tests.rs b/crates/omnigraph/src/db/manifest/tests.rs index d6e9b905..f053b63a 100644 --- a/crates/omnigraph/src/db/manifest/tests.rs +++ b/crates/omnigraph/src/db/manifest/tests.rs @@ -3558,3 +3558,185 @@ async fn projection_refresh_matches_clean_full_reopen() { .any(|row| row.graph_commit_id == unacknowledged.graph_commit_id) ); } + +async fn legacy_manifest_fixture( + uri: &str, + source: &DatasetEntry, + table_version: u64, + key_version: u64, + mode: lance::dataset::WriteMode, +) -> Dataset { + let mut entry = source.clone(); + entry.published_dataset_version = table_version; + entry.manifest_version = key_version; + let metadata = HashMap::from([( + entry.identity, + entry.version_metadata.to_json_string().unwrap(), + )]); + let batch = super::state::entries_to_batch(&[entry], &metadata, &[]).unwrap(); + let batch = if matches!(mode, lance::dataset::WriteMode::Append) { + batch.slice(1, 1) + } else { + batch + }; + let schema = Arc::new( + batch + .schema() + .as_ref() + .clone() + .with_metadata(HashMap::from([( + "omnigraph:internal_schema_version".to_string(), + "6".to_string(), + )])), + ); + let batch = RecordBatch::try_new(schema.clone(), batch.columns().to_vec()).unwrap(); + Dataset::write( + RecordBatchIterator::new(vec![Ok(batch)], schema), + uri, + Some(lance::dataset::WriteParams { + mode, + enable_stable_row_ids: true, + data_storage_version: Some(lance_file::version::LanceFileVersion::V2_2), + ..Default::default() + }), + ) + .await + .unwrap() +} + +#[tokio::test] +async fn legacy_manifest_decoder_preserves_data_version_order() { + let dir = tempfile::tempdir().unwrap(); + let mc = ManifestCoordinator::init( + dir.path().join("graph").to_str().unwrap(), + &build_test_catalog(), + ) + .await + .unwrap(); + let source = mc.known_state.entries[0].clone(); + let fixture = dir.path().join("legacy"); + let uri = fixture.to_str().unwrap(); + legacy_manifest_fixture(uri, &source, 40, 40, lance::dataset::WriteMode::Create).await; + let dataset = + legacy_manifest_fixture(uri, &source, 20, 20, lance::dataset::WriteMode::Append).await; + let legacy = super::state::read_manifest_state(&dataset).await.unwrap(); + assert_eq!(legacy.entries[0].published_dataset_version, 40); + let by_update = super::state::read_manifest_state_with_registration_clocks(&dataset) + .await + .unwrap(); + assert_eq!(by_update.entries[0].published_dataset_version, 20); + assert_eq!(by_update.entries[0].manifest_version, 2); + let historical = dataset.checkout_version(1).await.unwrap(); + assert_eq!( + super::state::read_manifest_state(&historical) + .await + .unwrap() + .entries[0] + .published_dataset_version, + 40 + ); + assert!(super::migrations::guard_stamp(&dataset).is_err()); +} + +#[tokio::test] +async fn legacy_manifest_decoder_refuses_key_pointer_mismatch() { + let dir = tempfile::tempdir().unwrap(); + let mc = ManifestCoordinator::init( + dir.path().join("graph").to_str().unwrap(), + &build_test_catalog(), + ) + .await + .unwrap(); + let source = mc.known_state.entries[0].clone(); + let dataset = legacy_manifest_fixture( + dir.path().join("legacy").to_str().unwrap(), + &source, + 40, + 41, + lance::dataset::WriteMode::Create, + ) + .await; + let error = super::state::read_manifest_state(&dataset) + .await + .unwrap_err(); + assert!(error.to_string().contains("expected table version 40")); + let error = super::state::read_manifest_state_with_registration_clocks(&dataset) + .await + .unwrap_err(); + assert!(error.to_string().contains("expected table version 40")); +} + +#[tokio::test] +async fn legacy_manifest_decoder_preserves_equal_version_tombstone() { + let dir = tempfile::tempdir().unwrap(); + let mc = ManifestCoordinator::init( + dir.path().join("graph").to_str().unwrap(), + &build_test_catalog(), + ) + .await + .unwrap(); + let mut source = mc.known_state.entries[0].clone(); + source.published_dataset_version = 40; + source.manifest_version = 40; + let fixture = dir.path().join("legacy"); + let uri = fixture.to_str().unwrap(); + let original = + legacy_manifest_fixture(uri, &source, 40, 40, lance::dataset::WriteMode::Create).await; + let metadata = HashMap::from([( + source.identity, + source.version_metadata.to_json_string().unwrap(), + )]); + let batch = super::state::entries_to_batch(std::slice::from_ref(&source), &metadata, &[]) + .unwrap() + .slice(1, 1); + let mut columns = batch.columns().to_vec(); + columns[0] = Arc::new(StringArray::from(vec![super::layout::tombstone_object_id( + source.identity, + 40, + )])); + columns[1] = Arc::new(StringArray::from(vec![OBJECT_TYPE_TABLE_TOMBSTONE])); + let schema = Arc::new( + batch + .schema() + .as_ref() + .clone() + .with_metadata(original.schema().metadata.clone()), + ); + let batch = RecordBatch::try_new(schema.clone(), columns).unwrap(); + let mut dataset = Dataset::write( + RecordBatchIterator::new(vec![Ok(batch)], schema), + uri, + Some(lance::dataset::WriteParams { + mode: lance::dataset::WriteMode::Append, + ..Default::default() + }), + ) + .await + .unwrap(); + assert!( + super::state::read_manifest_state(&dataset) + .await + .unwrap() + .entries + .is_empty() + ); + assert!( + super::state::read_manifest_state_with_registration_clocks(&dataset) + .await + .unwrap() + .entries + .is_empty() + ); + dataset + .update_schema_metadata([("omnigraph:internal_schema_version", "7")]) + .await + .unwrap(); + let error = super::state::read_manifest_state(&dataset) + .await + .unwrap_err(); + assert!( + error + .to_string() + .contains("above the scanned dataset version") + ); +} diff --git a/crates/omnigraph/src/db/manifest/upgrade.rs b/crates/omnigraph/src/db/manifest/upgrade.rs new file mode 100644 index 00000000..f5f6846c --- /dev/null +++ b/crates/omnigraph/src/db/manifest/upgrade.rs @@ -0,0 +1,1140 @@ +use std::collections::{BTreeSet, HashMap, HashSet}; +use std::sync::Arc; + +use arrow_array::{Array, RecordBatch, StringArray, UInt64Array}; +use arrow_schema::{Schema, SchemaRef}; +use datafusion::physical_plan::stream::RecordBatchStreamAdapter; +use futures::TryStreamExt; +use lance::Dataset; +use lance::dataset::refs::BranchIdentifier; +use lance::dataset::transaction::{Operation, Transaction, UpdateMap}; +use lance::dataset::{CommitBuilder, InsertBuilder, WriteMode, WriteParams}; +use lance_file::version::LanceFileVersion; +use serde::{Deserialize, Serialize}; + +use crate::error::{OmniError, Result}; +use crate::storage::{normalize_root_uri, storage_for_uri}; + +use super::layout::open_manifest_dataset_native_with_session; +use super::migrations::{INTERNAL_SCHEMA_VERSION_KEY, read_stamp}; +use super::state::{read_manifest_state, read_manifest_state_with_registration_clocks}; +use super::{ + OBJECT_TYPE_GRAPH_COMMIT, OBJECT_TYPE_GRAPH_HEAD, OBJECT_TYPE_TABLE, + OBJECT_TYPE_TABLE_TOMBSTONE, OBJECT_TYPE_TABLE_VERSION, +}; + +pub(super) const UPGRADE_PENDING_KEY: &str = "omnigraph:storage_upgrade_pending"; +const UPGRADE_RECEIPT_KEY: &str = "omnigraph:storage_upgrade_receipt"; +const HANDLER: &str = "registration-clocks-v6-to-v7"; +const DEFAULT_TARGET: u32 = 7; +const MAX_BRANCHES: usize = 1024; +const MAX_VERSIONS: usize = 100_000; +const MAX_APPENDED_UPGRADE_VERSIONS: u64 = 3; +const MAX_ROWS: usize = 1_000_000; +const MAX_METADATA_BYTES: usize = 64 * 1024 * 1024; +const MAX_INTENT_BYTES: usize = 1024 * 1024; + +/// Explicit storage conversion options. Execution requires exclusive operator control. +#[derive(Debug, Clone, Copy, Default)] +pub struct UpgradeOptions { + pub check: bool, + pub to_format: Option, +} + +#[derive(Debug, Clone, Copy, Serialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum UpgradeMode { + Check, + Execute, +} + +#[derive(Debug, Clone, Copy, Serialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum UpgradeOutcome { + CheckPassed, + AlreadyCurrent, + Completed, + CheckFailed, + Interrupted, + RecoveryRequired, +} + +#[derive(Debug, Serialize)] +pub struct UpgradeFinding { + pub code: String, + pub message: String, +} + +#[derive(Debug, Serialize)] +pub struct UpgradeRecovery { + pub failed_handler: String, + pub executable_compatibility: String, + pub action: String, +} + +#[derive(Debug, Default, Serialize)] +pub struct UpgradeWork { + pub metadata_rows: u64, + pub retained_snapshots: u64, + pub payload_bytes_copied: u64, + pub payload_bytes_rewritten: u64, + pub validation_bytes: Option, + pub external_blob_exclusions: BTreeSet, + pub historical_blob_identity_limits: BTreeSet, +} + +#[derive(Debug, Serialize)] +pub struct UpgradeReport { + pub mode: UpgradeMode, + pub outcome: UpgradeOutcome, + pub location: String, + pub graph_identity: Option, + pub observed_format: Option, + pub target_format: u32, + pub target_defaulted: bool, + pub route: Vec, + pub completed_handlers: Vec, + pub findings: Vec, + pub last_durable_completed_boundary: Option, + pub recovery: Option, + pub work: UpgradeWork, +} + +impl UpgradeReport { + pub fn success(&self) -> bool { + matches!( + self.outcome, + UpgradeOutcome::CheckPassed + | UpgradeOutcome::AlreadyCurrent + | UpgradeOutcome::Completed + ) + } + + fn finding(&mut self, code: &str, message: impl Into) { + self.findings.push(UpgradeFinding { + code: code.into(), + message: message.into(), + }); + } + + fn recover(&mut self, code: &str, message: impl Into) { + self.outcome = UpgradeOutcome::RecoveryRequired; + self.finding(code, message); + self.recovery = Some(UpgradeRecovery { + failed_handler: HANDLER.into(), + executable_compatibility: "this storage-upgrade-capable v7 executable; do not use the v6 executable".into(), + action: "stop all writers and maintenance, retain the backup, then rerun the same upgrade command with --to-format 7 without --check".into(), + }); + } +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +struct SourceBranch { + native: Option, + identity: BranchIdentifier, + version: u64, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +struct UpgradeIntent { + protocol: u32, + attempt: String, + source_format: u32, + target_format: u32, + graph_identity: String, + branches: Vec, +} + +#[derive(Debug, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +struct BranchReceipt { + protocol: u32, + attempt: String, + source: SourceBranch, +} + +/// Upgrade a standalone graph offline; neither this function nor `--check` starts recovery on open. +pub async fn upgrade_storage(uri: &str, options: UpgradeOptions) -> Result { + upgrade_storage_as(uri, options, None, None).await +} + +/// Apply SchemaApply policy to every affected branch before any storage effects. +pub async fn upgrade_storage_as( + uri: &str, + options: UpgradeOptions, + actor: Option<&str>, + policy: Option<&dyn omnigraph_policy::PolicyChecker>, +) -> Result { + let root = normalize_root_uri(uri)?; + let mut report = UpgradeReport { + mode: if options.check { + UpgradeMode::Check + } else { + UpgradeMode::Execute + }, + outcome: UpgradeOutcome::CheckFailed, + location: root.clone(), + graph_identity: None, + observed_format: None, + target_format: options.to_format.unwrap_or(DEFAULT_TARGET), + target_defaulted: options.to_format.is_none(), + route: Vec::new(), + completed_handlers: Vec::new(), + findings: Vec::new(), + last_durable_completed_boundary: None, + recovery: None, + work: UpgradeWork::default(), + }; + let result = run(&root, options, actor, policy, &mut report).await; + if let Err(error) = result { + if report.last_durable_completed_boundary.is_some() || report.recovery.is_some() { + report.recover("upgrade_interrupted", error.to_string()); + } else { + report.finding("preflight_failed", error.to_string()); + } + } + Ok(report) +} + +async fn open(root: &str, native: Option<&str>) -> Result { + open_manifest_dataset_native_with_session(root, native, &crate::lance_access::control_session()) + .await +} + +fn invalid(message: impl Into) -> OmniError { + OmniError::manifest(message) +} + +fn intent_from(dataset: &Dataset) -> Result> { + let Some(json) = dataset.schema().metadata.get(UPGRADE_PENDING_KEY) else { + return Ok(None); + }; + if json.len() > MAX_INTENT_BYTES { + return Err(invalid( + "storage upgrade intent exceeds the metadata budget", + )); + } + let intent: UpgradeIntent = serde_json::from_str(json) + .map_err(|e| invalid(format!("unrecognized upgrade ownership: {e}")))?; + let mut names = HashSet::new(); + if intent.protocol != 1 + || intent.source_format != 6 + || intent.target_format != 7 + || intent.attempt.parse::().is_err() + || intent.graph_identity.is_empty() + || intent.branches.is_empty() + || intent.branches.len() > MAX_BRANCHES + || intent + .branches + .last() + .is_none_or(|branch| branch.native.is_some()) + || intent + .branches + .iter() + .any(|branch| branch.version == 0 || !names.insert(branch.native.clone())) + { + return Err(invalid("unsupported or ambiguous storage upgrade intent")); + } + Ok(Some(intent)) +} + +async fn run( + root: &str, + options: UpgradeOptions, + actor: Option<&str>, + policy: Option<&dyn omnigraph_policy::PolicyChecker>, + report: &mut UpgradeReport, +) -> Result<()> { + let main = open(root, None).await?; + report.observed_format = read_stamp(&main); + let pending = match intent_from(&main) { + Ok(value) => value, + Err(error) => { + report.recover("unknown_upgrade_ownership", error.to_string()); + return Ok(()); + } + }; + if pending.is_some() { + report.recover( + "pending_upgrade", + "an owned storage conversion requires explicit recovery", + ); + } + if report.target_format != DEFAULT_TARGET { + report.finding("unsupported_target", "this binary declares target format 7 only; no complete route exists to the requested target"); + return Ok(()); + } + let storage = storage_for_uri(root)?; + let (_, schema_state) = + crate::db::schema_state::load_validated_schema_contract(root, Arc::clone(&storage)).await?; + report.graph_identity = Some(schema_state.schema_identity_domain.clone()); + let sidecars = super::list_sidecars(root, storage.as_ref()).await?; + if !sidecars.is_empty() { + report.outcome = UpgradeOutcome::RecoveryRequired; + report.finding( + "source_recovery_required", + "pre-existing graph recovery must be resolved before storage conversion", + ); + report.recovery = Some(UpgradeRecovery { + failed_handler: HANDLER.into(), executable_compatibility: "the source-compatible executable for this graph's recovery format".into(), + action: "before starting migration, stop writers, preserve the backup and resolve pending recovery with the source executable; then rerun upgrade --check".into(), + }); + return Ok(()); + } + if pending.is_none() && report.observed_format == Some(DEFAULT_TARGET) { + super::migrations::guard_stamp(&main)?; + read_manifest_state(&main).await?; + let branches = crate::branch_control::list_branch_contents(&main).await?; + if branches.len() >= MAX_BRANCHES { + return Err(invalid("storage upgrade branch limit exceeded")); + } + for native in branches.keys() { + if crate::db::is_internal_system_branch(native) { + return Err(invalid( + "resolve internal branch recovery before storage upgrade", + )); + } + let branch = main + .checkout_branch(native) + .await + .map_err(OmniError::storage)?; + super::migrations::guard_stamp(&branch)?; + read_manifest_state(&branch).await?; + } + report.outcome = UpgradeOutcome::AlreadyCurrent; + return Ok(()); + } + if pending.is_none() && report.observed_format != Some(6) { + report.finding("unsupported_source", "only validated v6 graphs have a conversion handler; preserve the source and use its executable for export/import"); + return Ok(()); + } + report.route.push(HANDLER.into()); + let intent = match pending { + Some(intent) => { + if intent.graph_identity != schema_state.schema_identity_domain { + return Err(invalid("upgrade schema identity changed")); + } + intent + } + None => inventory(&main, schema_state.schema_identity_domain.clone()).await?, + }; + for branch in &intent.branches { + if let Some(checker) = policy { + let actor = actor.ok_or_else(|| { + OmniError::Policy( + "storage upgrade requires an actor when policy is installed".into(), + ) + })?; + let name = branch + .native + .as_deref() + .map(crate::branch_names::logical_branch_name) + .unwrap_or("main"); + checker + .check( + omnigraph_policy::PolicyAction::SchemaApply, + &omnigraph_policy::ResourceScope::TargetBranch(name.into()), + actor, + ) + .map_err(|e| OmniError::Policy(e.to_string()))?; + } + } + verify_inventory(root, &intent, report.recovery.is_some()).await?; + preflight(root, &intent, &mut report.work).await?; + if options.check { + if report.recovery.is_none() { + report.outcome = UpgradeOutcome::CheckPassed; + } + return Ok(()); + } + let _root_exclusion = crate::db::reserve_export_root_exclusion(root)?; + verify_inventory(root, &intent, report.recovery.is_some()).await?; + if report.recovery.is_none() { + let main = open(root, None).await?; + let json = serde_json::to_string(&intent).map_err(|e| invalid(e.to_string()))?; + if json.len() > MAX_INTENT_BYTES { + return Err(invalid( + "storage upgrade intent exceeds the metadata budget", + )); + } + report.recover( + "fence_publication_attempted", + "inspect durable ownership before retrying an uncertain fence publication", + ); + publish_fence(main, json).await?; + } + report.last_durable_completed_boundary = Some("source_fenced".into()); + crate::failpoints::maybe_fail(crate::failpoints::names::UPGRADE_AFTER_FENCE)?; + for branch in &intent.branches { + let current = open(root, branch.native.as_deref()).await?; + if current + .branch_identifier() + .await + .map_err(OmniError::storage)? + != branch.identity + { + return Err(invalid("native branch lifetime changed before conversion")); + } + if branch_completed(¤t, branch, &intent)? { + continue; + } + verify_source_head(¤t, branch, branch.native.is_none())?; + let source = current + .checkout_version(branch.version) + .await + .map_err(OmniError::storage)?; + publish_conversion(current, source, branch, &intent).await?; + report.last_durable_completed_boundary = Some(format!( + "converted:{}", + branch.native.as_deref().unwrap_or("main") + )); + crate::failpoints::maybe_fail(crate::failpoints::names::UPGRADE_AFTER_BRANCH)?; + } + for branch in &intent.branches { + let current = open(root, branch.native.as_deref()).await?; + if !branch_completed(¤t, branch, &intent)? { + return Err(invalid("activated graph has an incomplete branch")); + } + let source = current + .checkout_version(branch.version) + .await + .map_err(OmniError::storage)?; + equivalent(&source, ¤t).await?; + } + verify_inventory(root, &intent, true).await?; + let main = open(root, None).await?; + if intent_from(&main)?.as_ref() != Some(&intent) { + return Err(invalid("activation ownership changed")); + } + crate::failpoints::maybe_fail(crate::failpoints::names::UPGRADE_BEFORE_ACTIVATION)?; + publish_activation(main).await?; + report.last_durable_completed_boundary = Some("activated".into()); + crate::failpoints::maybe_fail(crate::failpoints::names::UPGRADE_AFTER_ACTIVATION)?; + report.outcome = UpgradeOutcome::Completed; + report.completed_handlers.push(HANDLER.into()); + report.recovery = None; + report.findings.clear(); + Ok(()) +} + +async fn inventory(main: &Dataset, graph_identity: String) -> Result { + let branches = crate::branch_control::list_branch_contents(main).await?; + if branches.len() >= MAX_BRANCHES { + return Err(invalid("storage upgrade branch limit exceeded")); + } + let mut names: Vec<_> = branches.into_keys().collect(); + names.sort(); + let mut sources = Vec::with_capacity(names.len() + 1); + for native in names { + if crate::db::is_internal_system_branch(&native) { + return Err(invalid( + "resolve schema recovery and internal branches before upgrade", + )); + } + let ds = main + .checkout_branch(&native) + .await + .map_err(OmniError::storage)?; + sources.push(SourceBranch { + native: Some(native), + identity: ds.branch_identifier().await.map_err(OmniError::storage)?, + version: ds.version().version, + }); + } + sources.push(SourceBranch { + native: None, + identity: main.branch_identifier().await.map_err(OmniError::storage)?, + version: main.version().version, + }); + Ok(UpgradeIntent { + protocol: 1, + attempt: ulid::Ulid::new().to_string(), + source_format: 6, + target_format: 7, + graph_identity, + branches: sources, + }) +} + +fn receipt(branch: &SourceBranch, intent: &UpgradeIntent) -> BranchReceipt { + BranchReceipt { + protocol: 1, + attempt: intent.attempt.clone(), + source: branch.clone(), + } +} + +fn branch_completed( + dataset: &Dataset, + source: &SourceBranch, + intent: &UpgradeIntent, +) -> Result { + let Some(raw) = dataset.schema().metadata.get(UPGRADE_RECEIPT_KEY) else { + return Ok(false); + }; + if raw.len() > MAX_INTENT_BYTES { + return Err(invalid( + "storage upgrade receipt exceeds the metadata budget", + )); + } + let found: BranchReceipt = + serde_json::from_str(raw).map_err(|e| invalid(format!("invalid upgrade receipt: {e}")))?; + if found != receipt(source, intent) || read_stamp(dataset) != Some(7) { + return Err(invalid("foreign upgrade receipt")); + } + let expected = source + .version + .checked_add(if source.native.is_none() { 2 } else { 1 }) + .ok_or_else(|| invalid("upgrade version overflow"))?; + let activated = + source.native.is_none() && !dataset.schema().metadata.contains_key(UPGRADE_PENDING_KEY); + let expected = expected + .checked_add(u64::from(activated)) + .ok_or_else(|| invalid("upgrade version overflow"))?; + if dataset.version().version != expected { + return Err(invalid( + "upgraded branch moved while conversion was incomplete", + )); + } + Ok(true) +} + +fn verify_source_head(dataset: &Dataset, source: &SourceBranch, fenced: bool) -> Result<()> { + let expected = source + .version + .checked_add(u64::from(fenced)) + .ok_or_else(|| invalid("upgrade version overflow"))?; + if dataset.version().version != expected { + return Err(invalid("source branch changed; refusing foreign movement")); + } + if !fenced && read_stamp(dataset) != Some(6) { + return Err(invalid("source branch is not v6")); + } + Ok(()) +} + +async fn verify_inventory(root: &str, intent: &UpgradeIntent, fenced: bool) -> Result<()> { + let main = open(root, None).await?; + let observed: BTreeSet<_> = crate::branch_control::list_branch_contents(&main) + .await? + .into_keys() + .collect(); + let expected: BTreeSet<_> = intent + .branches + .iter() + .filter_map(|branch| branch.native.clone()) + .collect(); + if observed != expected { + return Err(invalid("native branch inventory changed during upgrade")); + } + if fenced && intent_from(&main)?.as_ref() != Some(intent) { + return Err(invalid("main upgrade ownership changed")); + } + for source in &intent.branches { + let dataset = open(root, source.native.as_deref()).await?; + if dataset + .branch_identifier() + .await + .map_err(OmniError::storage)? + != source.identity + { + return Err(invalid("native branch lifetime changed during upgrade")); + } + if !branch_completed(&dataset, source, intent)? { + verify_source_head(&dataset, source, fenced && source.native.is_none())?; + } + } + Ok(()) +} + +fn dependency_root_identity(uri: &str) -> Result { + let uri = normalize_root_uri(uri)?; + if crate::storage::storage_kind_for_uri(&uri)? == crate::storage::StorageKind::Local { + std::fs::canonicalize(&uri)? + .to_str() + .map(str::to_owned) + .ok_or_else(|| invalid("storage dependency path is not valid UTF-8")) + } else { + Ok(uri) + } +} + +fn confined(root: &str, dataset: &Dataset) -> Result<()> { + let root = dependency_root_identity(root)?; + for base in dataset.manifest().base_paths.values() { + let path = dependency_root_identity(&base.path)?; + if path != root && !path.starts_with(&format!("{root}/")) { + return Err(invalid( + "shared Lance dependencies outside the graph root require separately qualified retention and backup; upgrade refuses", + )); + } + } + Ok(()) +} + +async fn preflight(root: &str, intent: &UpgradeIntent, work: &mut UpgradeWork) -> Result<()> { + let mut refs = HashSet::new(); + let mut logical_names = HashSet::from(["main"]); + for native in intent + .branches + .iter() + .filter_map(|branch| branch.native.as_deref()) + { + let logical = crate::branch_names::logical_branch_name(native); + crate::branch_names::ensure_logical_branch_name(logical)?; + if !logical_names.insert(logical) { + return Err(invalid( + "ambiguous source branch identity: duplicate logical branch name", + )); + } + } + for branch in &intent.branches { + let current = open(root, branch.native.as_deref()).await?; + let source = current + .checkout_version(branch.version) + .await + .map_err(OmniError::storage)?; + if read_stamp(&source) != Some(6) { + return Err(invalid("source history has an unsupported format")); + } + if source + .schema() + .unenforced_primary_key() + .iter() + .map(|field| field.name.as_str()) + .collect::>() + != ["object_id"] + || !source.manifest().uses_stable_row_ids() + { + return Err(invalid("source manifest primary-key evidence is invalid")); + } + validate_metadata_budget(&source).await?; + let (old, lineage) = super::state::read_manifest_state_and_lineage(&source).await?; + validate_source_branch_identity(branch, &old, &lineage)?; + let mut translated = read_manifest_state_with_registration_clocks(&source).await?; + compare_states(old, &mut translated)?; + let schema: Schema = source.schema().into(); + let expected = super::state::manifest_schema(); + if schema.fields().len() != expected.fields().len() + || schema + .fields() + .iter() + .zip(expected.fields()) + .any(|(actual, expected)| { + actual.name() != expected.name() + || actual.data_type() != expected.data_type() + || actual.is_nullable() != expected.is_nullable() + }) + { + return Err(invalid( + "source manifest columns do not match the registered v6 format", + )); + } + let schema = Arc::new(schema); + let mut scan = source.scan(); + let mut columns: Vec<_> = schema + .fields() + .iter() + .map(|field| field.name().clone()) + .collect(); + columns.push("_row_last_updated_at_version".into()); + scan.project(&columns).map_err(OmniError::storage)?; + scan.batch_size(1024); + let mut stream = scan.try_into_stream().await.map_err(OmniError::storage)?; + let mut seen = HashSet::new(); + let mut count = 0usize; + while let Some(batch) = stream.try_next().await.map_err(OmniError::storage)? { + convert_batch( + batch.clone(), + Arc::clone(&schema), + branch.version, + &mut seen, + )?; + count = count + .checked_add(batch.num_rows()) + .ok_or_else(|| invalid("metadata row count overflow"))?; + if count > MAX_ROWS { + return Err(invalid("storage upgrade metadata-row limit exceeded")); + } + } + work.metadata_rows += u64::try_from(count).map_err(|e| invalid(e.to_string()))?; + let versions = retained_version_refs(&source, MAX_VERSIONS).await?; + for version in versions { + let snapshot = source + .checkout_version(version.version) + .await + .map_err(OmniError::storage)?; + let unstamped_bootstrap = matches!(snapshot.version().version, 1 | 2) + && !snapshot + .schema() + .metadata + .contains_key(INTERNAL_SCHEMA_VERSION_KEY); + if read_stamp(&snapshot) != Some(6) && !unstamped_bootstrap { + return Err(invalid(format!( + "retained history contains an unsupported format at {:?} version {}", + branch.native, + snapshot.version().version + ))); + } + confined(root, &snapshot)?; + validate_metadata_budget(&snapshot).await?; + let (state, lineage) = super::state::read_manifest_state_and_lineage(&snapshot).await?; + if unstamped_bootstrap { + let genesis = lineage.first(); + let valid_genesis = lineage.len() == 1 + && genesis.is_some_and(|genesis| { + genesis.graph_manifest_version == 1 + && genesis.graph_branch.is_none() + && genesis.parent_commit_id.is_none() + && genesis.merged_parent_commit_id.is_none() + && genesis.actor_id.is_none() + && state.graph_heads.len() == 1 + && state.graph_heads.get("main") == Some(&genesis.graph_commit_id) + }); + let valid_entries = state.entries.iter().all(|entry| { + entry.entity_count == 0 + && entry.published_dataset_version == 1 + && entry.manifest_version == 1 + && entry.native_dataset_branch.is_none() + }); + let expected_rows = state.entries.len() * 2 + 2; + if !valid_genesis + || !valid_entries + || snapshot + .count_rows(None) + .await + .map_err(OmniError::storage)? + != expected_rows + { + return Err(invalid( + "unstamped history does not match the released v0.9 empty bootstrap contract", + )); + } + } + work.retained_snapshots += 1; + for entry in state.entries { + let key = ( + entry.dataset_path.clone(), + entry.native_dataset_branch.clone(), + entry.published_dataset_version, + ); + if !refs.insert(key) { + continue; + } + if refs.len() > MAX_ROWS { + return Err(invalid("storage upgrade dependency limit exceeded")); + } + let uri = format!("{root}/{}", entry.dataset_path); + let table = crate::instrumentation::open_dataset( + &uri, + crate::instrumentation::VersionResolution::Latest, + Some(&crate::lance_access::control_session()), + crate::instrumentation::manifest_wrapper(), + ) + .await?; + let table = if let Some(native) = &entry.native_dataset_branch { + table + .checkout_branch(native) + .await + .map_err(OmniError::storage)? + } else { + table + }; + let table = table + .checkout_version(entry.published_dataset_version) + .await + .map_err(OmniError::storage)?; + confined(root, &table)?; + if table + .schema() + .unenforced_primary_key() + .iter() + .map(|field| field.name.as_str()) + .collect::>() + != ["id"] + || table + .schema() + .unenforced_primary_key() + .iter() + .any(|field| field.nullable) + || !table.manifest().uses_stable_row_ids() + { + return Err(invalid( + "source table lacks v6 id primary-key or stable-row identity evidence", + )); + } + table.validate().await.map_err(OmniError::storage)?; + for field in table.schema().fields.iter().filter(|field| field.is_blob()) { + if !field + .metadata + .contains_key(crate::db::STABLE_PROPERTY_ID_METADATA_KEY) + { + work.historical_blob_identity_limits + .insert(format!("{}:{}", entry.dataset_path, field.name)); + } + } + validate_blobs(&table, work).await?; + } + } + } + Ok(()) +} + +fn validate_source_branch_identity( + branch: &SourceBranch, + state: &super::state::ManifestState, + lineage: &[super::state::GraphLineageRow], +) -> Result<()> { + let Some(native) = branch.native.as_deref() else { + return Ok(()); + }; + let (logical, suffix) = crate::branch_names::split_native_branch_name(native); + if suffix.is_none() { + return Ok(()); + } + let fork_version = branch + .identity + .version_mapping + .last() + .map(|(version, _)| *version) + .ok_or_else(|| invalid("missing source branch identity fork witness"))?; + let head = state.graph_heads.get(logical); + let witnessed = lineage.iter().any(|commit| { + head == Some(&commit.graph_commit_id) + && commit.graph_branch.as_deref() == Some(logical) + && commit.graph_manifest_version > fork_version + && commit.graph_manifest_version <= branch.version + }); + if !witnessed { + return Err(invalid(format!( + "ambiguous source branch identity for '{native}': no post-fork logical-name witness; preserve the source and resolve branch naming with its executable before upgrade" + ))); + } + Ok(()) +} + +async fn retained_version_refs( + dataset: &Dataset, + limit: usize, +) -> Result> { + let inventory_limit = u64::try_from(limit) + .map_err(|error| invalid(error.to_string()))? + .saturating_add(MAX_APPENDED_UPGRADE_VERSIONS); + if dataset.count_versions().await.map_err(OmniError::storage)? > inventory_limit { + return Err(invalid("storage upgrade retained-version limit exceeded")); + } + let mut versions = dataset.version_refs().await.map_err(OmniError::storage)?; + if versions.len() as u64 > inventory_limit { + return Err(invalid("storage upgrade retained-version limit exceeded")); + } + versions.retain(|version| version.version <= dataset.version().version); + if versions.len() > limit { + return Err(invalid("storage upgrade retained-version limit exceeded")); + } + Ok(versions) +} + +fn compare_states( + mut old: super::state::ManifestState, + translated: &mut super::state::ManifestState, +) -> Result<()> { + old.entries.sort_by_key(|entry| entry.identity); + translated.entries.sort_by_key(|entry| entry.identity); + if old.graph_heads != translated.graph_heads || old.entries.len() != translated.entries.len() { + return Err(invalid( + "registration-order damage: conversion would change the logical graph", + )); + } + for (mut before, after) in old.entries.into_iter().zip(&translated.entries) { + before.manifest_version = after.manifest_version; + if !before.same_registration(after) { + return Err(invalid( + "registration-order damage: conversion would change logical rows; automatic repair is unsupported", + )); + } + } + Ok(()) +} + +async fn equivalent(source: &Dataset, target: &Dataset) -> Result<()> { + let old = read_manifest_state(source).await?; + let mut converted = read_manifest_state(target).await?; + compare_states(old, &mut converted) +} + +async fn publish_fence(dataset: Dataset, intent: String) -> Result { + let operation = Operation::UpdateConfig { + config_updates: None, + table_metadata_updates: None, + field_metadata_updates: HashMap::new(), + schema_metadata_updates: Some(UpdateMap { + update_entries: vec![ + (INTERNAL_SCHEMA_VERSION_KEY.to_string(), "7".to_string()).into(), + (UPGRADE_PENDING_KEY.to_string(), intent).into(), + ], + replace: false, + }), + }; + let transaction = Transaction::new(dataset.version().version, operation, None); + CommitBuilder::new(Arc::new(dataset)) + .with_max_retries(0) + .with_skip_auto_cleanup(true) + .execute(transaction) + .await + .map_err(OmniError::storage) +} + +async fn publish_activation(dataset: Dataset) -> Result { + let operation = Operation::UpdateConfig { + config_updates: None, + table_metadata_updates: None, + field_metadata_updates: HashMap::new(), + schema_metadata_updates: Some(UpdateMap { + update_entries: vec![(UPGRADE_PENDING_KEY.to_string(), None::).into()], + replace: false, + }), + }; + let transaction = Transaction::new(dataset.version().version, operation, None); + CommitBuilder::new(Arc::new(dataset)) + .with_max_retries(0) + .with_skip_auto_cleanup(true) + .execute(transaction) + .await + .map_err(OmniError::storage) +} + +async fn publish_conversion( + current: Dataset, + source: Dataset, + branch: &SourceBranch, + intent: &UpgradeIntent, +) -> Result<()> { + let mut metadata = source.schema().metadata.clone(); + metadata.insert(INTERNAL_SCHEMA_VERSION_KEY.into(), "7".into()); + metadata.remove(UPGRADE_PENDING_KEY); + if branch.native.is_none() { + metadata.insert( + UPGRADE_PENDING_KEY.into(), + serde_json::to_string(intent).map_err(|e| invalid(e.to_string()))?, + ); + } + metadata.insert( + UPGRADE_RECEIPT_KEY.into(), + serde_json::to_string(&receipt(branch, intent)).map_err(|e| invalid(e.to_string()))?, + ); + let schema: Schema = source.schema().into(); + let schema = Arc::new(schema.with_metadata(metadata)); + let mut scan = source.scan(); + let mut columns: Vec = schema + .fields() + .iter() + .map(|field| field.name().clone()) + .collect(); + columns.push("_row_last_updated_at_version".into()); + scan.project(&columns).map_err(OmniError::storage)?; + scan.batch_size(1024); + let batches = scan.try_into_stream().await.map_err(OmniError::storage)?; + let output_schema = Arc::clone(&schema); + let mut seen = HashSet::new(); + let version = source.version().version; + let converted = batches + .map_err(datafusion::error::DataFusionError::from) + .and_then(move |batch| { + let result = convert_batch(batch, Arc::clone(&output_schema), version, &mut seen) + .map_err(|error| datafusion::error::DataFusionError::External(Box::new(error))); + futures::future::ready(result) + }); + let stream: datafusion::physical_plan::SendableRecordBatchStream = + Box::pin(RecordBatchStreamAdapter::new(schema, converted)); + let params = WriteParams { + mode: WriteMode::Overwrite, + enable_stable_row_ids: true, + data_storage_version: Some(LanceFileVersion::V2_2), + skip_auto_cleanup: true, + max_rows_per_file: 64 * 1024, + max_rows_per_group: 1024, + ..Default::default() + }; + let destination = Arc::new(current); + let transaction = InsertBuilder::new(Arc::clone(&destination)) + .with_params(¶ms) + .execute_uncommitted_stream(stream) + .await + .map_err(OmniError::storage)?; + crate::failpoints::maybe_fail(crate::failpoints::names::UPGRADE_AFTER_STAGE)?; + let target = CommitBuilder::new(destination) + .with_max_retries(0) + .with_skip_auto_cleanup(true) + .execute(transaction) + .await + .map_err(OmniError::storage)?; + if !branch_completed(&target, branch, intent)? { + return Err(invalid( + "storage conversion publication did not carry its receipt", + )); + } + equivalent(&source, &target).await +} + +fn convert_batch( + batch: RecordBatch, + schema: SchemaRef, + source_version: u64, + seen: &mut HashSet, +) -> Result { + let column = |name: &str| { + batch + .column_by_name(name) + .ok_or_else(|| invalid(format!("missing manifest column {name}"))) + }; + let ids = column("object_id")? + .as_any() + .downcast_ref::() + .ok_or_else(|| invalid("object_id must be Utf8"))?; + let types = column("object_type")? + .as_any() + .downcast_ref::() + .ok_or_else(|| invalid("object_type must be Utf8"))?; + let clocks = column("_row_last_updated_at_version")? + .as_any() + .downcast_ref::() + .ok_or_else(|| invalid("row update provenance must be UInt64"))?; + let mut converted = Vec::with_capacity(batch.num_rows()); + for row in 0..batch.num_rows() { + if ids.is_null(row) || types.is_null(row) { + return Err(invalid("null manifest identity")); + } + if !matches!( + types.value(row), + OBJECT_TYPE_TABLE + | OBJECT_TYPE_TABLE_VERSION + | OBJECT_TYPE_TABLE_TOMBSTONE + | OBJECT_TYPE_GRAPH_COMMIT + | OBJECT_TYPE_GRAPH_HEAD + ) { + return Err(invalid( + "source manifest contains an unsupported object type", + )); + } + let id = if matches!( + types.value(row), + OBJECT_TYPE_TABLE_VERSION | OBJECT_TYPE_TABLE_TOMBSTONE + ) { + if clocks.is_null(row) || clocks.value(row) == 0 || clocks.value(row) > source_version { + return Err(invalid("missing or invalid source registration provenance")); + } + let (prefix, _) = ids + .value(row) + .rsplit_once(':') + .ok_or_else(|| invalid("invalid source registration key"))?; + format!("{prefix}:{:020}", clocks.value(row)) + } else { + ids.value(row).to_string() + }; + if !seen.insert(id.clone()) { + return Err(invalid("duplicate converted manifest identity")); + } + if seen.len() > MAX_ROWS { + return Err(invalid("storage upgrade metadata-row limit exceeded")); + } + converted.push(id); + } + let mut output = Vec::with_capacity(schema.fields().len()); + for field in schema.fields() { + output.push(if field.name() == "object_id" { + Arc::new(StringArray::from(converted.clone())) as _ + } else { + Arc::clone(column(field.name())?) + }); + } + RecordBatch::try_new(schema, output).map_err(|error| invalid(error.to_string())) +} + +#[cfg(test)] +#[path = "upgrade/tests.rs"] +mod tests; + +async fn validate_blobs(table: &Dataset, work: &mut UpgradeWork) -> Result<()> { + let columns: Vec<_> = table + .schema() + .fields + .iter() + .filter(|field| field.is_blob()) + .map(|field| field.name.clone()) + .collect(); + if columns.is_empty() { + return Ok(()); + } + let table = Arc::new(table.clone()); + let mut scan = table.scan(); + scan.project::<&str>(&[]).map_err(OmniError::storage)?; + scan.with_row_id(); + scan.batch_size(1024); + let mut stream = scan.try_into_stream().await.map_err(OmniError::storage)?; + while let Some(batch) = stream.try_next().await.map_err(OmniError::storage)? { + let ids = batch + .column_by_name("_rowid") + .and_then(|column| column.as_any().downcast_ref::()) + .ok_or_else(|| invalid("missing stable row IDs while validating Blob dependencies"))?; + if ids.null_count() != 0 { + return Err(invalid("null stable row ID in Blob dependency scan")); + } + for column in &columns { + for blob in table + .take_blobs(ids.values(), column) + .await + .map_err(OmniError::storage)? + .into_iter() + .flatten() + { + if let Some(uri) = blob.uri() { + if work.external_blob_exclusions.len() >= MAX_ROWS { + return Err(invalid("external Blob dependency limit exceeded")); + } + work.external_blob_exclusions.insert(uri.to_string()); + } else { + let mut offset: u64 = 0; + while offset < blob.size() { + let end = offset.saturating_add(1024 * 1024).min(blob.size()); + let bytes = blob + .read_range(offset..end) + .await + .map_err(OmniError::storage)?; + if bytes.len() as u64 != end - offset { + return Err(invalid("truncated managed Blob dependency")); + } + offset = end; + } + } + } + } + } + Ok(()) +} + +async fn validate_metadata_budget(dataset: &Dataset) -> Result<()> { + if dataset.count_rows(None).await.map_err(OmniError::storage)? > MAX_ROWS { + return Err(invalid("storage upgrade metadata-row limit exceeded")); + } + let mut scan = dataset.scan(); + scan.batch_size(1024); + let mut stream = scan.try_into_stream().await.map_err(OmniError::storage)?; + let mut bytes = 0usize; + let mut rows = 0usize; + while let Some(batch) = stream.try_next().await.map_err(OmniError::storage)? { + bytes = bytes + .checked_add(batch.get_array_memory_size()) + .ok_or_else(|| invalid("metadata byte count overflow"))?; + rows = rows + .checked_add(batch.num_rows()) + .ok_or_else(|| invalid("metadata row count overflow"))?; + if bytes > MAX_METADATA_BYTES || rows > MAX_ROWS { + return Err(invalid( + "storage upgrade metadata budget exceeded (1,000,000 rows or 64 MiB per retained manifest)", + )); + } + } + Ok(()) +} diff --git a/crates/omnigraph/src/db/manifest/upgrade/tests.rs b/crates/omnigraph/src/db/manifest/upgrade/tests.rs new file mode 100644 index 00000000..0b5d45df --- /dev/null +++ b/crates/omnigraph/src/db/manifest/upgrade/tests.rs @@ -0,0 +1,548 @@ +use std::collections::BTreeMap; +use std::path::{Path, PathBuf}; + +use super::*; +use crate::db::Omnigraph; + +async fn synthetic_v6_fixture(root: &str) { + let db = Omnigraph::init(root, "node Person { name: String }") + .await + .unwrap(); + db.mutate( + "main", + "query seed($name: String) { insert Person { name: $name } }", + "seed", + &HashMap::from([( + "name".to_string(), + omnigraph_compiler::query::ast::Literal::String("before upgrade".to_string()), + )]), + ) + .await + .unwrap(); + drop(db); + let mut dataset = open(root, None).await.unwrap(); + dataset + .update_schema_metadata([(INTERNAL_SCHEMA_VERSION_KEY, "6")]) + .await + .unwrap(); + let policy = lance::dataset::cleanup::CleanupPolicy { + before_version: Some(dataset.version().version), + before_timestamp: None, + delete_unverified: false, + error_if_tagged_old_versions: false, + clean_referenced_branches: false, + delete_rate_limit: None, + }; + lance::dataset::cleanup::cleanup_old_versions(&dataset, policy) + .await + .unwrap(); + dataset + .create_branch("feature", dataset.version().version, None) + .await + .unwrap(); +} + +fn stored_files(root: &Path) -> BTreeMap, std::time::SystemTime)> { + fn collect(path: &Path, files: &mut BTreeMap, std::time::SystemTime)>) { + for entry in std::fs::read_dir(path).unwrap() { + let entry = entry.unwrap(); + let path = entry.path(); + let metadata = entry.metadata().unwrap(); + if metadata.is_dir() { + collect(&path, files); + } else { + files.insert( + path.clone(), + (std::fs::read(path).unwrap(), metadata.modified().unwrap()), + ); + } + } + } + let mut files = BTreeMap::new(); + collect(root, &mut files); + files +} + +#[tokio::test] +async fn storage_upgrade_check_has_no_local_store_effects() { + #[cfg(feature = "failpoints")] + let _scenario = crate::failpoints::FailScenario::setup(); + let dir = tempfile::tempdir().unwrap(); + let root = dir.path().to_str().unwrap(); + synthetic_v6_fixture(root).await; + let before = stored_files(dir.path()); + let tracker = lance_io::utils::tracking_store::IOTracker::default(); + let probes = crate::instrumentation::QueryIoProbes { + manifest_wrapper: Some(Arc::new(tracker.clone())), + table_wrapper: Some(Arc::new(tracker.clone())), + ..Default::default() + }; + let report = crate::instrumentation::with_query_io_probes( + probes, + upgrade_storage( + root, + UpgradeOptions { + check: true, + to_format: Some(7), + }, + ), + ) + .await + .unwrap(); + let stats = tracker.stats(); + assert!(stats.read_iops > 0); + assert_eq!(stats.write_iops, 0, "{stats:?}"); + assert!( + !stats + .requests + .iter() + .any(|request| request.method == "delete") + ); + assert_eq!(report.outcome, UpgradeOutcome::CheckPassed, "{report:?}"); + assert_eq!(stored_files(dir.path()), before); + assert!(!report.route.is_empty()); + assert!(report.work.retained_snapshots >= 2); + assert!(Omnigraph::open(root).await.is_err()); + assert!(Omnigraph::open_read_only(root).await.is_err()); + let unsupported = upgrade_storage( + root, + UpgradeOptions { + check: false, + to_format: Some(8), + }, + ) + .await + .unwrap(); + assert_eq!(unsupported.outcome, UpgradeOutcome::CheckFailed); + assert_eq!(stored_files(dir.path()), before); +} + +#[cfg(feature = "failpoints")] +#[tokio::test] +async fn storage_upgrade_interruption_boundaries_retry_without_mixed_visibility() { + use crate::failpoints::{FailScenario, ScopedFailPoint, names}; + let _scenario = FailScenario::setup(); + for boundary in [ + names::UPGRADE_AFTER_FENCE, + names::UPGRADE_AFTER_STAGE, + names::UPGRADE_AFTER_BRANCH, + names::UPGRADE_BEFORE_ACTIVATION, + names::UPGRADE_AFTER_ACTIVATION, + ] { + let dir = tempfile::tempdir().unwrap(); + let root = dir.path().to_str().unwrap(); + synthetic_v6_fixture(root).await; + let report = { + let _fault = ScopedFailPoint::new(boundary, "return"); + upgrade_storage(root, UpgradeOptions::default()) + .await + .unwrap() + }; + assert_eq!( + report.outcome, + UpgradeOutcome::RecoveryRequired, + "{boundary}: {report:?}" + ); + let activated = boundary == names::UPGRADE_AFTER_ACTIVATION; + assert_eq!(Omnigraph::open(root).await.is_ok(), activated, "{boundary}"); + assert_eq!( + Omnigraph::open_read_only(root).await.is_ok(), + activated, + "{boundary}" + ); + let before_check = stored_files(dir.path()); + let check = upgrade_storage( + root, + UpgradeOptions { + check: true, + to_format: Some(7), + }, + ) + .await + .unwrap(); + assert_eq!( + check.outcome, + if activated { + UpgradeOutcome::AlreadyCurrent + } else { + UpgradeOutcome::RecoveryRequired + }, + "{boundary}: {check:?}" + ); + assert_eq!(stored_files(dir.path()), before_check, "{boundary}"); + let retried = upgrade_storage(root, UpgradeOptions::default()) + .await + .unwrap(); + assert_eq!( + retried.outcome, + if activated { + UpgradeOutcome::AlreadyCurrent + } else { + UpgradeOutcome::Completed + }, + "{boundary}: {retried:?}" + ); + assert!(Omnigraph::open(root).await.is_ok(), "{boundary}"); + assert!(Omnigraph::open_read_only(root).await.is_ok(), "{boundary}"); + for branch in [None, Some("feature")] { + let dataset = open(root, branch).await.unwrap(); + assert_eq!(read_stamp(&dataset), Some(7), "{boundary}"); + } + let repeated = upgrade_storage(root, UpgradeOptions::default()) + .await + .unwrap(); + assert_eq!( + repeated.outcome, + UpgradeOutcome::AlreadyCurrent, + "{boundary}" + ); + } +} + +#[cfg(feature = "failpoints")] +#[tokio::test] +async fn storage_upgrade_recovery_refuses_foreign_head_movement() { + use crate::failpoints::{FailScenario, ScopedFailPoint, names}; + let _scenario = FailScenario::setup(); + let dir = tempfile::tempdir().unwrap(); + let root = dir.path().to_str().unwrap(); + synthetic_v6_fixture(root).await; + { + let _fault = ScopedFailPoint::new(names::UPGRADE_AFTER_FENCE, "return"); + let interrupted = upgrade_storage(root, UpgradeOptions::default()) + .await + .unwrap(); + assert_eq!( + interrupted.outcome, + UpgradeOutcome::RecoveryRequired, + "{interrupted:?}" + ); + } + let mut foreign = open(root, Some("feature")).await.unwrap(); + foreign + .update_schema_metadata([("test:foreign", "movement")]) + .await + .unwrap(); + let before_retry = stored_files(dir.path()); + let refused = upgrade_storage(root, UpgradeOptions::default()) + .await + .unwrap(); + assert_eq!( + refused.outcome, + UpgradeOutcome::RecoveryRequired, + "{refused:?}" + ); + assert!( + refused + .findings + .iter() + .any(|finding| finding.message.contains("foreign movement")), + "{refused:?}" + ); + assert_eq!(stored_files(dir.path()), before_retry); + assert!(Omnigraph::open(root).await.is_err()); + assert!(Omnigraph::open_read_only(root).await.is_err()); +} + +#[tokio::test] +async fn storage_upgrade_tracks_metadata_writes_and_no_payload_effects() { + #[cfg(feature = "failpoints")] + let _scenario = crate::failpoints::FailScenario::setup(); + let dir = tempfile::tempdir().unwrap(); + let root = dir.path().to_str().unwrap(); + synthetic_v6_fixture(root).await; + let table_files = |files: BTreeMap, std::time::SystemTime)>| { + files + .into_iter() + .filter(|(path, _)| { + path.components() + .any(|part| part.as_os_str() == "nodes" || part.as_os_str() == "edges") + }) + .collect::>() + }; + let before_tables = table_files(stored_files(dir.path())); + assert!(!before_tables.is_empty()); + let tracker = lance_io::utils::tracking_store::IOTracker::default(); + let probes = crate::instrumentation::QueryIoProbes { + manifest_wrapper: Some(Arc::new(tracker.clone())), + table_wrapper: Some(Arc::new(tracker.clone())), + ..Default::default() + }; + let report = crate::instrumentation::with_query_io_probes( + probes, + upgrade_storage(root, UpgradeOptions::default()), + ) + .await + .unwrap(); + assert_eq!(report.outcome, UpgradeOutcome::Completed, "{report:?}"); + assert_eq!(table_files(stored_files(dir.path())), before_tables); + let stats = tracker.stats(); + assert!(stats.write_iops > 0 && stats.written_bytes > 0, "{stats:?}"); + assert!( + stats.write_iops < 100 && stats.written_bytes < 1_048_576, + "{stats:?}" + ); + assert!( + !stats + .requests + .iter() + .any(|request| request.method == "copy"), + "metadata-only conversion must not perform storage-side copies: {stats:?}" + ); + for request in stats.requests { + match request.method { + "put" | "put_opts" | "put_part" | "copy" | "rename" => { + assert!( + request + .path + .as_ref() + .split('/') + .any(|part| part == "__manifest"), + "unexpected non-manifest write: {request:?}" + ); + } + "delete" => panic!("upgrade deleted an existing object: {request:?}"), + _ => {} + } + } +} + +#[tokio::test] +async fn storage_upgrade_policy_denial_precedes_effects() { + #[cfg(feature = "failpoints")] + let _scenario = crate::failpoints::FailScenario::setup(); + struct DenySchemaApply; + impl omnigraph_policy::PolicyChecker for DenySchemaApply { + fn check( + &self, + action: omnigraph_policy::PolicyAction, + scope: &omnigraph_policy::ResourceScope, + actor: &str, + ) -> std::result::Result<(), omnigraph_policy::PolicyError> { + assert_eq!(action, omnigraph_policy::PolicyAction::SchemaApply); + assert!(matches!( + scope, + omnigraph_policy::ResourceScope::TargetBranch(_) + )); + assert_eq!(actor, "blocked-actor"); + Err(omnigraph_policy::PolicyError::Denied("test denial".into())) + } + } + let dir = tempfile::tempdir().unwrap(); + let root = dir.path().to_str().unwrap(); + synthetic_v6_fixture(root).await; + let before = stored_files(dir.path()); + let report = upgrade_storage_as( + root, + UpgradeOptions::default(), + Some("blocked-actor"), + Some(&DenySchemaApply), + ) + .await + .unwrap(); + assert_eq!(report.outcome, UpgradeOutcome::CheckFailed, "{report:?}"); + assert!( + report + .findings + .iter() + .any(|finding| finding.message.contains("test denial")) + ); + assert_eq!(stored_files(dir.path()), before); +} + +#[tokio::test] +async fn storage_upgrade_refuses_unknown_ownership_and_source() { + #[cfg(feature = "failpoints")] + let _scenario = crate::failpoints::FailScenario::setup(); + for source_format in ["5", "99"] { + let dir = tempfile::tempdir().unwrap(); + let root = dir.path().to_str().unwrap(); + synthetic_v6_fixture(root).await; + let mut dataset = open(root, None).await.unwrap(); + dataset + .update_schema_metadata([(INTERNAL_SCHEMA_VERSION_KEY, source_format)]) + .await + .unwrap(); + let before = stored_files(dir.path()); + let report = upgrade_storage(root, UpgradeOptions::default()) + .await + .unwrap(); + assert_eq!(report.outcome, UpgradeOutcome::CheckFailed, "{report:?}"); + assert!( + report + .findings + .iter() + .any(|finding| finding.code == "unsupported_source") + ); + assert_eq!(stored_files(dir.path()), before); + } + let dir = tempfile::tempdir().unwrap(); + let root = dir.path().to_str().unwrap(); + synthetic_v6_fixture(root).await; + let mut dataset = open(root, None).await.unwrap(); + dataset + .update_schema_metadata([(UPGRADE_PENDING_KEY, "{}")]) + .await + .unwrap(); + let before = stored_files(dir.path()); + let report = upgrade_storage(root, UpgradeOptions::default()) + .await + .unwrap(); + assert_eq!( + report.outcome, + UpgradeOutcome::RecoveryRequired, + "{report:?}" + ); + assert!( + report + .findings + .iter() + .any(|finding| finding.code == "unknown_upgrade_ownership") + ); + assert_eq!(stored_files(dir.path()), before); +} + +#[tokio::test] +async fn storage_upgrade_refuses_preexisting_recovery_without_healing() { + #[cfg(feature = "failpoints")] + let _scenario = crate::failpoints::FailScenario::setup(); + let dir = tempfile::tempdir().unwrap(); + let root = dir.path().to_str().unwrap(); + synthetic_v6_fixture(root).await; + let dataset = open(root, None).await.unwrap(); + let entry = read_manifest_state(&dataset) + .await + .unwrap() + .entries + .remove(0); + let pin = super::super::recovery::SidecarTablePin { + identity: entry.identity, + table_key: entry.type_key, + table_path: format!( + "{}/{}", + normalize_root_uri(root).unwrap(), + entry.dataset_path + ), + expected_version: entry.published_dataset_version, + post_commit_pin: entry.published_dataset_version + 1, + confirmed_version: None, + table_branch: entry.native_dataset_branch, + }; + let sidecar = super::super::recovery::new_optimize_sidecar_v9(vec![pin]).unwrap(); + let recovery = dir.path().join("__recovery"); + std::fs::create_dir_all(&recovery).unwrap(); + std::fs::write( + recovery.join(format!("{}.json", sidecar.operation_id)), + serde_json::to_vec(&sidecar).unwrap(), + ) + .unwrap(); + let before = stored_files(dir.path()); + for check in [true, false] { + let report = upgrade_storage( + root, + UpgradeOptions { + check, + to_format: Some(7), + }, + ) + .await + .unwrap(); + assert_eq!( + report.outcome, + UpgradeOutcome::RecoveryRequired, + "{report:?}" + ); + assert!( + report + .findings + .iter() + .any(|finding| finding.code == "source_recovery_required") + ); + assert!( + report + .recovery + .unwrap() + .executable_compatibility + .contains("source-compatible") + ); + assert_eq!(stored_files(dir.path()), before); + } +} + +#[tokio::test] +async fn storage_upgrade_current_main_refuses_legacy_branch_without_effects() { + #[cfg(feature = "failpoints")] + let _scenario = crate::failpoints::FailScenario::setup(); + let dir = tempfile::tempdir().unwrap(); + let root = dir.path().to_str().unwrap(); + synthetic_v6_fixture(root).await; + let mut main = open(root, None).await.unwrap(); + main.update_schema_metadata([(INTERNAL_SCHEMA_VERSION_KEY, "7")]) + .await + .unwrap(); + assert!(!main.schema().metadata.contains_key(UPGRADE_PENDING_KEY)); + assert_eq!( + read_stamp(&open(root, Some("feature")).await.unwrap()), + Some(6) + ); + let before = stored_files(dir.path()); + for check in [true, false] { + let report = upgrade_storage( + root, + UpgradeOptions { + check, + to_format: Some(7), + }, + ) + .await + .unwrap(); + assert!(!report.success(), "{report:?}"); + assert_eq!(report.outcome, UpgradeOutcome::CheckFailed, "{report:?}"); + assert!(!report.findings.is_empty()); + assert_eq!(stored_files(dir.path()), before); + } +} + +#[tokio::test] +async fn storage_upgrade_history_budget_precedes_manifest_reads() { + #[cfg(feature = "failpoints")] + let _scenario = crate::failpoints::FailScenario::setup(); + let dir = tempfile::tempdir().unwrap(); + let root = dir.path().to_str().unwrap(); + synthetic_v6_fixture(root).await; + let dataset = open(root, None).await.unwrap(); + let version = dataset.version().version; + assert!(version > 1); + let write_unreadable_version = |number| { + let path = dir + .path() + .join("__manifest/_versions") + .join(format!("{:020}.manifest", u64::MAX - number)); + assert!(!path.exists()); + let mut malformed = vec![0_u8; 64]; + malformed[..4].copy_from_slice(&1_u32.to_le_bytes()); + malformed[4] = 0xff; + std::fs::write(path, malformed).unwrap(); + }; + write_unreadable_version(1); + let error = retained_version_refs(&dataset, 1).await.unwrap_err(); + assert!( + error.to_string().contains("retained-version limit"), + "{error}" + ); + let refs = retained_version_refs(&dataset, 2).await.unwrap(); + assert_eq!( + refs.iter().map(|entry| entry.version).collect::>(), + [1, version] + ); + assert!(dataset.checkout_version(1).await.is_err()); + for offset in 1..=MAX_APPENDED_UPGRADE_VERSIONS { + write_unreadable_version(version + offset); + } + let refs = retained_version_refs(&dataset, 2).await.unwrap(); + assert_eq!( + refs.len(), + 2, + "retry excludes the protocol's appended versions" + ); +} diff --git a/crates/omnigraph/src/db/mod.rs b/crates/omnigraph/src/db/mod.rs index 985ac409..686ed5e0 100644 --- a/crates/omnigraph/src/db/mod.rs +++ b/crates/omnigraph/src/db/mod.rs @@ -9,6 +9,10 @@ pub(crate) mod write_queue; pub use commit_graph::GraphCommit; pub use graph_coordinator::{ReadTarget, ResolvedTarget, SnapshotId}; pub use manifest::{DatasetEntry, DatasetUpdate, Snapshot, SnapshotDataset, SnapshotScanner}; +pub use manifest::{ + UpgradeFinding, UpgradeMode, UpgradeOptions, UpgradeOutcome, UpgradeRecovery, UpgradeReport, + UpgradeWork, upgrade_storage, upgrade_storage_as, +}; pub(crate) use omnigraph::ensure_public_branch_ref; pub use omnigraph::{ CleanupPolicyOptions, DatasetCleanupStats, DatasetOptimizeStats, DatasetRepairStats, diff --git a/crates/omnigraph/src/failpoints.rs b/crates/omnigraph/src/failpoints.rs index a3dab87f..c38b6e50 100644 --- a/crates/omnigraph/src/failpoints.rs +++ b/crates/omnigraph/src/failpoints.rs @@ -132,6 +132,11 @@ pub(crate) fn maybe_fail_retryable_contention(name: &str) -> Result<()> { /// reference these constants instead of bare string literals, so a typo is a /// compile error rather than a silently-never-firing failpoint. pub mod names { + pub const UPGRADE_AFTER_FENCE: &str = "upgrade.after_fence"; + pub const UPGRADE_AFTER_STAGE: &str = "upgrade.after_stage"; + pub const UPGRADE_AFTER_BRANCH: &str = "upgrade.after_branch"; + pub const UPGRADE_BEFORE_ACTIVATION: &str = "upgrade.before_activation"; + pub const UPGRADE_AFTER_ACTIVATION: &str = "upgrade.after_activation"; /// After Lance returns success from its two-phase native create, before /// OmniGraph acknowledges it. Recovery must classify the matching /// BranchContents as a completed create (lost acknowledgement). diff --git a/crates/omnigraph/tests/forbidden_apis.rs b/crates/omnigraph/tests/forbidden_apis.rs index aaad71d5..f496570f 100644 --- a/crates/omnigraph/tests/forbidden_apis.rs +++ b/crates/omnigraph/tests/forbidden_apis.rs @@ -143,6 +143,8 @@ const ALLOW_LIST_FILES: &[&str] = &[ "db/manifest/recovery.rs", // Recovery executor; exactly inventoried below. "db/manifest/tests.rs", // Out-of-line tests for the trusted gateways. "instrumentation.rs", // The instrumented dataset opener. + "db/manifest/upgrade.rs", + "db/manifest/upgrade/tests.rs", ]; /// Out-of-line test modules are parsed as standalone files, so their enclosing @@ -156,6 +158,7 @@ const PROTOCOL_SCAN_EXCLUDED_FILES: &[&str] = &[ // source walk cannot see that attribute. "db/manifest/namespace.rs", "db/manifest/tests.rs", + "db/manifest/upgrade/tests.rs", ]; const SENTINEL: &str = "// forbidden-api-allow:"; @@ -698,6 +701,9 @@ macro_rules! durable_calls { // manifest implementations are included; only standalone test-only sources // whose parent cfg is invisible to this file walker are excluded. durable_calls! { + ("db/manifest/upgrade.rs", "CommitBuilder::new(", 3, WriteProtocol::Exact("offline storage upgrade with main-owned intent")), + ("db/manifest/upgrade.rs", "InsertBuilder::new(", 1, WriteProtocol::Exact("manifest-only conversion under durable upgrade ownership")), + ("db/manifest/upgrade.rs", ".execute_uncommitted_stream(", 1, WriteProtocol::Exact("manifest-only conversion under durable upgrade ownership")), ("table_store/fts_compat.rs", ".put(", 1, WriteProtocol::Composed("staged index artifact")), // The `__manifest` Create write is the manifest's entire birth: entries, // genesis lineage, and the internal-schema stamp all ride the one commit, diff --git a/crates/omnigraph/tests/lance_version_columns.rs b/crates/omnigraph/tests/lance_version_columns.rs index fbe0cb48..bb12baa2 100644 --- a/crates/omnigraph/tests/lance_version_columns.rs +++ b/crates/omnigraph/tests/lance_version_columns.rs @@ -281,3 +281,51 @@ async fn lance_merge_insert_update_preserves_created_at_version() { "bob updated_at must bump to the commit version on a merge_insert UPDATE" ); } + +#[tokio::test] +async fn metadata_upgrade_preserves_historical_stamps_and_row_provenance() { + let dir = tempfile::tempdir().unwrap(); + let uri = dir.path().join("test.lance"); + let uri = uri.to_str().unwrap(); + let mut main = create_test_dataset(uri).await; + let stamp_key = "omnigraph:internal_schema_version"; + main.update_schema_metadata([(stamp_key, "6")]) + .await + .unwrap(); + let source_version = main.version().version; + let source_rows = scan_with_versions(&main).await; + let mut feature = main + .create_branch("feature", source_version, None) + .await + .unwrap(); + let feature_source_version = feature.version().version; + assert_eq!(scan_with_versions(&feature).await, source_rows); + + main.update_schema_metadata([(stamp_key, "7")]) + .await + .unwrap(); + let current_main = Dataset::open(uri).await.unwrap(); + let current_feature = current_main.checkout_branch("feature").await.unwrap(); + assert_eq!(current_main.schema().metadata[stamp_key], "7"); + assert_eq!(current_feature.schema().metadata[stamp_key], "6"); + assert_eq!(current_feature.version().version, feature_source_version); + assert_eq!(scan_with_versions(¤t_main).await, source_rows); + + feature + .update_schema_metadata([(stamp_key, "7")]) + .await + .unwrap(); + let current_feature = current_main.checkout_branch("feature").await.unwrap(); + assert_eq!(current_feature.schema().metadata[stamp_key], "7"); + assert_eq!(scan_with_versions(¤t_feature).await, source_rows); + + let historical_main = current_main.checkout_version(source_version).await.unwrap(); + let historical_feature = current_feature + .checkout_version(feature_source_version) + .await + .unwrap(); + for historical in [historical_main, historical_feature] { + assert_eq!(historical.schema().metadata[stamp_key], "6"); + assert_eq!(scan_with_versions(&historical).await, source_rows); + } +} diff --git a/docs/dev/branch-protection.md b/docs/dev/branch-protection.md index 5b29bc40..e28e37b6 100644 --- a/docs/dev/branch-protection.md +++ b/docs/dev/branch-protection.md @@ -17,6 +17,7 @@ protection. - `Lint (clippy)` - `GQ Logic Tests` - `Fix Regression Gate` +- `Storage Upgrade Compatibility` Checks are strict, so a PR must be current with `main`. `Graph Vocabulary Guard` remains as an always-reporting context but reports a successful skip on @@ -25,7 +26,11 @@ audit runs after merge, on tags, and by manual dispatch. User documentation is intentionally outside this exact-occurrence audit and is validated by the documentation structure check. Documentation-only PRs still receive every required context; work-heavy steps -may report as skipped. +may report as skipped. `Storage Upgrade Compatibility` is an explicit exception: +both genuine predecessor migration journeys execute on every change, including +documentation-only pull requests. Its fixture availability and required-context +contract are checked by `scripts/check-storage-upgrade-ci.py`. The repository +policy change must still be applied by an administrator to affect GitHub. `Test Workspace` runs on pull requests as a reporting context and is deliberately not required: with strict checks every merge invalidates every diff --git a/docs/dev/versioning.md b/docs/dev/versioning.md index 06f17456..16615943 100644 --- a/docs/dev/versioning.md +++ b/docs/dev/versioning.md @@ -10,13 +10,13 @@ version axes. Never derive one axis from another. |---|---|---| | Release | Published workspace artifacts move in lockstep. | Workspace manifests, lockfile, generated metadata, release automation. | | CLI ↔ server wire | Prefer additive changes; documented breaking release boundaries require coordinated upgrades. No global version handshake. | Shared DTOs, OpenAPI drift tests, and release-specific migration guidance. | -| Graph storage | Strict single version; rebuild across an incompatible change. | Main-manifest stamp with `MIN_SUPPORTED == CURRENT`. | +| Graph storage | Current-format serving; explicit registered upgrades, otherwise rebuild. | Main-manifest stamp with `MIN_SUPPORTED == CURRENT`. | | Recovery sidecar | Independently versioned persisted protocol. | Sidecar grammar/version refusal before classification. | | Lance dependency and file format | One deliberately pinned Lance family and explicit stable file version. | Lockfile, write parameters, and Lance surface guards. | ## Current storage contract -The current binary reads and writes exactly **internal manifest schema v7**. +Normal graph open and new writes require **internal manifest schema v7**. `INTERNAL_MANIFEST_SCHEMA_VERSION` and `MIN_SUPPORTED_INTERNAL_SCHEMA_VERSION` are both 7. @@ -38,9 +38,12 @@ are both 7. "upgrade omnigraph" advice cannot be satisfied for it, and such a graph is rebuilt from an export taken with the build that wrote it. -A lower stamp is refused with export/rebuild guidance. A higher stamp is -refused before recovery or table decoding. There is no in-place migration -dispatcher. +Normal open refuses lower and higher stamps before recovery or table decoding. +`omnigraph upgrade` explicitly converts supported standalone v6 graphs to v7. +It preserves original retained snapshots and decodes their v6 registrations +explicitly after main-root admission. A pending upgrade marker refuses normal +opens until every branch validates and main activation completes. +See [RFC 0064](../rfcs/0064-explicit-storage-upgrades.md) for the offline protocol. ## Recovery version @@ -61,17 +64,41 @@ or a behavior that changes persisted graph meaning does. Current compatibility fences and the required upstream reading set are in [lance.md](lance.md). -## Why rebuild instead of migrate +## Registered conversion and rebuild fallback -An in-place migration permanently adds legacy readers, crash windows, and -version-pair tests. The present strand model keeps one readable physical shape: -export the old logical graph with the old binary, initialize a fresh current -graph, and load the export. Rows, vectors, Blob values, and schema meaning are -preserved; physical histories and stable identities intentionally restart. +The registered v6-to-v7 handler appends manifest metadata and retains table +files, branch ancestry, commit IDs and historical locators. Operators must stop +all writers and maintenance and preserve a restorable backup before execution. +Cluster-managed conversion is refused until its admission protocol is qualified. -The operator procedure is documented in +Other source formats still require export with the source executable, fresh +initialization and load. Rebuild preserves logical values but intentionally +restarts physical history and identities. See [the upgrade guide](../user/operations/upgrade.md). +## Storage upgrade support matrix + +The `storage_upgrade_compatibility` CI job (Storage Upgrade Compatibility) +requires genuine predecessor migration and admission regressions, engine +conversion/recovery tests, Lance version qualification and protocol guards on +every change. Missing binaries, missing test cases, empty runs and skipped +required cases fail the job. The binaries are version-checked before fixture +creation. Branch naming without a post-fork logical-name witness is refused; +see the [admission limits](../user/operations/upgrade.md). + +| Source executable / format | Normal open | Explicit route | Required case in `crossversion_upgrade.rs` | +|---|---|---|---| +| 0.9.0 / v6 | Refused | v6 → v7 | `genuine_v09_explicit_storage_upgrade_preserves_history` | +| 0.10.0 / v6 | Refused | v6 → v7 | `genuine_v010_explicit_storage_upgrade_preserves_history` | +| Current / v7 | Accepted | Already-current no-op | Both migration journeys, after conversion | +| Older or unknown / not v6 or v7 | Refused | No route; source-compatible export/rebuild | Existing format fences and engine refusal tests | + +These journeys cover local standalone roots. Object-store backend qualification +and deployment branch-protection configuration require their own environment +evidence; a local pass is not evidence for those gates. Cluster-managed entry +points remain refused. Changing a declared route requires updating its fixture, +refusal expectations and this matrix together, with storage-maintainer review. + ## Wire compatibility Prefer additive wire changes so compatible CLI and server releases can roll @@ -109,10 +136,11 @@ GitHub Releases, and the TypeScript SDK ships through npm. Do not document ### Graph storage 1. Write an RFC for the irreversible format decision. -2. Bump the manifest stamp and keep `MIN_SUPPORTED == CURRENT` unless a real - converter is implemented. +2. Bump the manifest stamp and keep normal-open `MIN_SUPPORTED == CURRENT`. + Register explicit conversion separately from serving admission. 3. Refuse old/future formats before decoding. -4. Add genuine old-binary/new-binary refusal and rebuild evidence. +4. Add genuine predecessor evidence for every declared direct or migration route, + plus refusal and rebuild fallback evidence. 5. Update the upgrade guide and release notes. ### Recovery diff --git a/docs/releases/v0.11.0.md b/docs/releases/v0.11.0.md index d9bfa65d..8a869e20 100644 --- a/docs/releases/v0.11.0.md +++ b/docs/releases/v0.11.0.md @@ -4,8 +4,8 @@ Notes accumulate here until the release is cut. ## Highlights -- **Graph storage moves to internal manifest schema v7; v6 graphs are rebuilt - by export and load.** Every `__manifest` registration row is now keyed by the +- **Graph storage moves to internal manifest schema v7, with explicit offline + upgrades for qualified v6 graphs.** Every `__manifest` registration row is now keyed by the manifest version that wrote it, and a branch's view of a table is the registration with the greatest manifest version rather than the greatest per-native-ref Lance version. A branch merge that adopts a source table whose @@ -13,8 +13,12 @@ Notes accumulate here until the release is cut. switch and reads the adopted rows; before, such a merge reported `fast_forward` and kept stale rows (lower) or failed with `table version N already exists ... with different state` (equal). That error - no longer exists. The rebuild is per branch and does not carry branch - topology, commit history or historical snapshots across; see + no longer exists. `omnigraph upgrade --check --to-format 7` checks a + standalone graph before in-place conversion. Qualified migration retains + branches, commit history and snapshots without copying table payloads; + ambiguous branch identities and cluster-managed roots refuse. Writers and + maintenance must be stopped and a verified backup retained. Export/load + remains the fallback for unsupported inputs and does not retain history; see [upgrade](../user/operations/upgrade.md) and [RFC 0062](../rfcs/0062-manifest-version-clock.md), the sole reason for this release's storage-format change. diff --git a/docs/rfcs/0064-explicit-storage-upgrades.md b/docs/rfcs/0064-explicit-storage-upgrades.md new file mode 100644 index 00000000..ffacd3e2 --- /dev/null +++ b/docs/rfcs/0064-explicit-storage-upgrades.md @@ -0,0 +1,418 @@ +--- +rfc: "0064" +title: "Explicit storage upgrades" +track: maintainer +status: draft +implementation: in-progress +authors: + - Azim Afroozeh +created: 2026-09-09 +updated: 2026-09-09 +discussion: null +supersedes: [] +superseded_by: [] +blocked_on: + - "Qualify object-store backends and deferred orphan reclamation before expanding local standalone support" + - "Storage-maintainer review of the implemented protocol and genuine v0.9/v0.10 compatibility evidence before acceptance" +--- + +# RFC 0064: Explicit storage upgrades + +> Number provisional: recheck the registry and open reservations before publication. +> A term in ***bold italics*** is defined at that spot. + +## Summary + +Extend `omnigraph upgrade` to select developer-written +***migration handlers***, code for named conversions with declared inputs, +outputs, prerequisites, checks, effects, validation, and recovery. +Start with internal v6 to v7 registration metadata ([RFC 0062](0062-manifest-version-clock.md)); +reuse table data only where its meaning and references remain valid. +Later handlers can cover [RFC 0040](0040-system-column-namespace.md)'s system +columns and settled fork ownership. Arbitrary conversions are unsupported; +the default is fast, in-place migration that appends metadata and reuses table +data. Publication and recovery still require proof; reclamation runs later. + +## Motivation + +Today's [versioning policy](../dev/versioning.md) refuses incompatible graphs +and requires export/import, preserving exported entities but restarting commit +history, snapshots, and shared branch ancestry. RFC 0062 identifies a metadata +converter using `_row_last_updated_at_version`; RFC 0040 specifies a recoverable +explicit upgrade. Extend that entry point to cover both without conflating their protocols. + +Full backward compatibility by v1.0 is a product goal, not a release-number +guarantee. For storage, it means newer binaries directly read and write +supported older graphs. This RFC establishes no wire, query-language, or SDK +compatibility; explicit upgrade remains useful for future required conversions. + +## User and operational behavior + +Proposed migration behavior and options for `omnigraph upgrade`: + +```text +omnigraph upgrade +omnigraph upgrade --check --to-format 7 --json +omnigraph upgrade --to-format 7 --json +``` + +The binary selects registered handlers from the stored format/capabilities to +one declared default target, independent of its serving range and release +number. `--to-format` overrides the target, never names a handler; v7 can stop +before system-column conversion. The default does not adapt to the graph. +Missing steps, cycles, ambiguous routes, and unimplemented handlers block +execution before effects. Numeric adjacency proves no route: graph format, +schema IR, and recovery format are separate compatibility axes. + +| Stage | Contract | +|---|---| +| `--check` | Select the route and run read-only handler preflight: formats, branches, retained history, references, dependencies, unsupported state, and pending recovery. No conversion, recovery writes, indexes, or implicit initialization. Report work categories and copying/rewriting estimates; distinguish known, unknown, and deferred checks. Unknown work is never zero. | +| Execution | Stop servers, embedded writers, maintenance, and cluster applies; prevent restart. Local locks or unchanged heads do not prove exclusivity. Repeat preflight under exclusive operator control. Validate checks requiring intermediate output before the affected handler's effects. | +| Preparation | Preflight identifies the source-compatible executable and command for pre-existing recovery; resolve it before starting a new handler. Unknown recovery formats refuse. Retain the old executable and verified backup covering the root and required dependencies. | +| Retry | After interruption or lost output, rerun the same command and target under exclusive control. Inspect authoritative state before effects; resume owned recovery or refuse with the required executable and exact action. Unknown/ambiguous ownership refuses. Handler-owned recovery may follow an early format fence: the source executable is not automatically safe. | + +An online check is advisory: it proves neither converted-output correctness +nor future availability and never authorizes serving with the target binary. +`--check` may report recovery actions but cannot perform them. + +| Outcome | Meaning | Exit | +|---|---|---| +| `check_passed` | Read-only checks pass; conversion remains required | 0 | +| `already_current` | No handler needed; no owned upgrade recovery remains | 0 | +| `completed` | Execution validated every requested handler and the resulting graph | 0 | +| `check_failed` | Blocking check or execution-preflight finding | Nonzero | +| `interrupted` | Execution stopped before route completion | Nonzero | +| `recovery_required` | Durable state requires an identified recovery action | Nonzero | + +`--check` never returns `completed`; partial chains never succeed. JSON and +human results carry mode (`check`/`execute`), outcome, graph identity/location, +observed format, target and whether defaulted, ordered route, completed +handlers, blocking findings with stable codes, and last durable completed +boundary. When recovery is needed, include failed handler, executable +compatibility, and exact operator action. Unestablished values are unknown. + +Cluster-managed graphs use the same engine handlers through a qualified +cluster operation; direct execution cannot bypass applied configuration, +policy, or runtime ownership. Refuse that route until implemented and tested. + +## Design + +### Terms and shared machinery + +- ***Storage format***: persisted structure and meaning identified by the + graph's internal-schema stamp, independent of binary release. +- ***Retained snapshot***: historical state readable under source retention. +- ***Native ref***: a Lance branch reference for a graph or table; logical + names do not prove native lifetime identity. + +Use compiled concrete handler functions, sharing admission, planning, reporting, +and results. Developers supply rules and tests; the binary selects rather than +invents conversions. A v6 graph targeting v7 selects its registered v6 to v7 +handler; longer routes require every declared prerequisite. Compatible releases +need no new handler. No plugins, third-party scripts, schema-diff language, or +generic migration ledger/scheduler. Share recovery only when effect identities +and publication rules match. Durable graph state and owned recovery establish +completion; progress reports and early-fence target stamps cannot establish it alone. + +Each handler defaults to in-place, append-only metadata work. Give it a restricted +storage interface for source reads and owned metadata staging, without methods +to overwrite/delete existing objects or copy/rewrite table payloads. Handlers +return a conversion plan; the shared runner validates it and publishes through +the existing graph-content publication path. Raw storage handles must not bypass +these restrictions. Protocol-owned publication and fencing effects use the +sealed publication/recovery APIs, not unrestricted handler writes. + +This is a default, not a claim that every format change fits it. Encoding, +physical-type, or encryption changes may require data movement. Such a handler +must declare an explicit exception, its effects and scope, storage-maintainer +approval, separate qualification tests, and work budgets. `--check` reports the +exception and its expected cost; there is no silent fallback to full rewriting. +Measure metadata scans and validation reads separately from payload copying or +rewriting. The v6 to v7 prototype must prove the default contract. + +### v6 to v7 + +RFC 0062 changes `table_version`/`table_tombstone` key suffixes from table data +versions to the owning branch's `__manifest` publication versions, preserving +data-pointer meaning. The candidate conversion is: + +1. Admit genuine released v6 with valid schema, recovery, identity, and version + evidence. Stamp-only admission cannot distinguish reused experimental formats. +2. Inventory all live branches, native identities, retained snapshots, and table + references needed for reads, merge ancestry, and cleanup. Never drop non-main + branches to satisfy another handler. +3. Decode selected snapshots explicitly as v6. Capture each registration and + tombstone's original `_row_last_updated_at_version` before rewriting; retries + must reuse this source evidence, not the rewritten row's update version. +4. Preserve source clocks only with proved continuity; otherwise qualify an + order-preserving translation into each destination branch's `__manifest` + publication clock. All retained snapshots must accept translated registrations + and tombstones, including inherited precedence; later ordinary writes must + order after converted state. Locator translation alone is insufficient. +5. Preserve stable table/incarnation identities and `(table_version, table_branch)` + pointer meaning. Validate keys, ordering, tombstone precedence, lifetimes, + and exact data availability; duplicate/ambiguous identities refuse before + publication. Publish through the selected whole-graph protocol and validate + retained reads, reopen, and subsequent writes before success. + +This is not proof of historical conversion: overwritten registrations may exist +only in older retained snapshots; current rows cannot reconstruct missing history. + +Preserve node/edge IDs, properties, schema identity, vectors, stored Blob values +and external descriptors, logical branch names/ancestry, retained commit IDs, +historical logical rows, and logical change-feed contents. Compare canonical +snapshot data or explicitly equivalent query semantics, not query bugs or wire +representations. Release-level query changes need separate expectations and +cannot hide conversion damage. Lost history is not recreated; unavailable +full-text history retains existing index-compatibility restrictions without an +inline rebuild. The same rule applies to pre-0.10 Blob property-lifetime +restrictions: retained bytes and descriptors are preserved, but missing identity +evidence is not fabricated. Preflight reports affected fields and tests read +retained bytes independently of the delivery guard. Physical version changes require validated resolution of every +exposed locator and dependent reference; no implicit renumbering. + +For registration-order damage, distinguish preservation, explicit repair, and +refusal. This handler refuses diagnostically when reordering would change +logical rows; automatic repair is excluded. Test this independently of query +representation changes. + +### Physical protocol and dependencies + +The v6-to-v7 implementation uses the existing graph location and a sealed +manifest publication gateway. It requires exclusive operator control: stop all +servers, embedded writers, cluster reconciliation and maintenance. Process-local +root exclusion does not fence other processes or already-open old handles. + +1. Preflight reads exact source versions, branch lifetime identities and the + schema identity. It refuses ambiguous logical branch names before fencing: + a suffixed native ref needs its own post-fork logical-head witness, not an + inherited head. It counts version references before loading history. It compares the legacy data-version fold with the proposed + registration-clock fold and refuses changes in logical state. +2. A main-manifest UpdateConfig commit atomically sets format 7 and + `omnigraph:storage_upgrade_pending`. Its versioned intent contains an attempt + ULID, source/target formats, schema identity and every native branch's source + version and lifetime identity. Old executables reject the new stamp; new + normal opens reject the pending intent before recovery or decoding. +3. With main still fenced, each native branch and then main appends converted + manifest fragments from its pinned v6 source. Only registration/tombstone key + suffixes change. Publication includes `omnigraph:storage_upgrade_receipt`, + binding that branch to the attempt and source. Main keeps the pending intent. + Publication has no automatic retry or cleanup; a changed head refuses. +4. Validate all converted branches against their source state. Only then append + a main UpdateConfig commit removing the pending intent. This is activation; + success requires it. No new serving pointer, locator map or generic ledger + is introduced. Durable authority remains the main manifest. + +Released v0.9 creates two unstamped bootstrap snapshots before stamping v6. +Preflight accepts those only at versions 1 and 2 with the exact empty-table, +version-one-pointer, single-parentless-genesis contract. Their registration keys +are already clock 1; normal root admission still rejects unstamped graphs. + +Retained source snapshots are deliberately not rewritten. After normal root +admission, an explicit v6 decoder preserves their original data-version ordering; +v7 snapshots use registration clocks. Numeric locators, commit IDs and table +pointers remain unchanged. Current registration clocks come from the pinned +source row provenance and precede subsequent publication versions. + +Retry validates the exact branch inventory, lifetime identities, source versions +and receipts. Completed branch publications are reused; unpublished staging is +recreated from immutable source evidence. Unknown ownership or foreign movement +refuses. A failure after activation is an already-current no-op on retry. The +pending main intent owns manifest staging for the attempt; cleanup is excluded +throughout the operation and remains a later retention-controlled activity. + +This is an explicit amendment to the head-only proposal: legacy historical +decoding and main-owned intent/receipts are necessary protocol state. The +alternative of rewriting historical snapshots would require a locator map and +would break the existing numeric version contract without further machinery. + +Every staged artifact needs attributable ownership before effects. Cleanup must +distinguish abandoned/obsolete artifacts from active graph data, retained history, +shared dependencies, and recovery-owned state. Reclaim only after existing +retention and recovery rules allow it; names alone do not establish abandonment. +Reuse or extend the existing cleanup owner rather than add a migration scheduler. +Cleanup reclaims space; it must not finish logical conversion or make an already +successful migration correct. Its compatibility must be qualified before shipping. + +Inventory the ***dependency closure***, all objects required by retained states, +including shared files outside the root. Required bytes need enforced retention +and immutability or verified backup/restore. Stop or exclude every writer and +cleanup process that could invalidate them, including other graphs; graph-local +exclusion and root-only backup are insufficient. Unknown dependencies or +unprotected required bytes refuse. Test cleanup, source retirement, and rollback. + +Preserve external Blob descriptors exactly. Externally managed URI bytes are +excluded from preservation/rollback unless immutability, retention, or verified +backup is established; preflight and results enumerate exclusions. Identical +descriptors do not prove byte preservation. Required shared Lance files cannot +use this exclusion. + +### Later handlers + +RFC 0040 owns main-only admission, preflight, `SchemaApply` recovery, and ordered +stamp/schema effects. Its CLI can delegate here without changing its cluster +declaration or branch restriction; this RFC does not activate it. Its early +stamp fence precludes a universal stamp-last rule. + +Fork conversion requires an accepted ownership representation and durable proof +of existing ownership. Never infer it from plausible names, recreate missing +incarnations, or assign current ownership to historical borrowers. Ambiguity +refuses. Format number and ordering remain unassigned, not automatically v7 to v8. + +## Invariants + +The [architectural invariants](../dev/invariants.md) apply: one graph-content +publication door, coherent snapshots without mixed formats, stable identities, +and loud refusal of unsupported state or missing required history. Incompatible +effects require owned recovery or an isolated disposable destination, never CLI +progress authority. Use Lance APIs inside the sealed boundary: no custom +transaction manager, public writable dataset handle, or raw metadata rewriting. +Bound scan batches and retries; report progress. Upgrade may scale with retained +history, ordinary requests must not. Engine action/scope admission covers direct +and cluster callers; actor attribution grants no permission by itself. + +## Compatibility and reversibility + +Converter input support does not widen serving support. Historical legacy +decoding requires explicit acceptance, not silently lowering +`MIN_SUPPORTED_INTERNAL_SCHEMA_VERSION`. Earlier, future, and experimental +formats without handlers retain refusal and rebuild guidance. + +No reverse handler: restore the complete backup and old deployment, including +required dependencies and declared external-byte exclusions. Retained old objects +in the same root do not by themselves provide rollback after activation or fencing. Plan explicitly for post-cutover writes absent from that backup. +Later-handler failure never automatically undoes completed handlers. + +Acceptance amends rebuild-only policy for qualified routes while retaining the +fallback. Coordinate policy/RFC 0062 and RFC 0040 command wording when implemented; +current user docs must not advertise this unimplemented draft. + +## Alternatives + +| Approach | Tradeoff / disposition | +|---|---| +| Separate destination | Isolates source authority but adds relocation, storage and cutover cost. Retain as an explicitly qualified exception if in-place conversion cannot satisfy the contract; no assumed root-copy/rename support. Shared shallow-clone dependencies must survive cleanup and source retirement; disposable restart requires source immutability and output ownership. | +| Export/import only | Existing, no converter state; loses ancestry/history. Retain fallback. | +| Separate commands | Locally small; duplicate admission/reporting and obscure prerequisite order. Share entry point. | +| Automatic open conversion | Less interaction; expensive effects/recovery become startup behavior. Reject initially. | +| Generic ledger/scheduler | Arbitrary resume chains add authority; handler recovery suffices for current scope. Omit. | +| Current heads only | Small rewrite cannot preserve history without legacy decoding. | +| Direct compatibility now | Avoids conversion but commits the engine to older read/write semantics. Evaluate toward v1.0. | + +RFCs 0040 and 0062 are the nearest precedents; all-branch historical conversion +has a different visibility boundary from main-only schema apply, so recovery +formats need not match. PostgreSQL's [pg_upgrade](https://www.postgresql.org/docs/current/pgupgrade.html) +and Neo4j's [database migrate](https://neo4j.com/docs/operations-manual/current/database-administration/standard-databases/migrate-database/) +establish an operator pattern, not Lance safety. + +## Evidence and tests + +Existing evidence is limited: `db/manifest/migrations.rs` separates refusal from +the converter seam; `db/manifest/graph.rs` enables stable row IDs; +`tests/lance_version_columns.rs` exercises physical version columns. None proves +conversion. Read complete relevant [Lance pages](../dev/lance.md), then qualify +pinned Lance 11.0.0; current website prose alone is insufficient. + +Every handler needs rule tests, supported/refused inputs, genuine predecessor +fixtures, interruption/retry, and work assertions. Extend these owners, not a +parallel harness: + +| Owner | Required coverage | +|---|---| +| `crates/omnigraph-cli/tests/crossversion_upgrade.rs` | Genuine predecessor creates nodes, edges, properties, branches, updates, deletions, and retained snapshots. Capture expected state; check, upgrade, compare the full preservation contract above, then reopen/write/merge/cleanup and recheck history. Old/new refusal, rerun, rollback. Counts alone are insufficient. | +| `crates/omnigraph/tests/lance_version_columns.rs`, `lance_surface_guards.rs` | Update, inheritance, compaction, retained provenance, clock translation, and selected physical protocol. | +| GQT | Equal/lower table-version adoption, key replacement, tombstones, borrowed forks, delete/recreate, and equivalent-semantics historical rows; damaged-input refusal separate from query changes. | +| `tests/failpoints.rs`, `tests/recovery.rs` | Before/after every durable effect/completion boundary, lost acknowledgements, same-command retry, source versus handler recovery, old-binary refusal, no readable mixed state. Every interruption leaves usable source or defined recovery to complete target. | +| Branch/maintenance/change-feed owners | Every retained branch/commit addressable; identity and results survive later operations. External deletion/mutation either refuses or satisfies retention/restore. | +| Existing local/object-store journeys | Same contract on every advertised backend; no atomic-root-rename or process-local distributed-fencing assumption. | +| Shared runner tests | Selection/order, missing/cyclic/ambiguous routes, later-step failure, already-current state, check success/failure, unknown/unavailable input, JSON/exit reporting, zero writes from `--check`. | + +Independently instrument storage operations, not handler counters. Separate +metadata work from payload bytes read/copied/rewritten/reused. Metadata-only CI +asserts zero payload copying/rewriting, including storage-side copies, and tests +reused-file readability across branches/history. Data-moving handlers need +fixture-specific budgets and declared-scope coverage. For the default handler, +also fail forbidden existing-object overwrites/deletes at the restricted interface +and storage-operation boundary, allowing only separately specified runner-owned +publication/fencing effects. Set metadata-operation and metadata-byte budgets on +representative fixtures: append-only is not a performance bound. Measure downtime +separately; no universal latency or size promise. + +Crash tests must exercise staging, validation, activation, and later cleanup. +Verify old-state visibility or explicit recovery refusal before activation, +complete new state afterward, safe retries, retained history, and eventual +reclamation of eligible leftovers without deleting live/shared/recovery data. +Successful migration must not depend on having run cleanup. + +### Required compatibility CI + +Once this migration support ships, run a required job on every change, including those +without format bumps. A version-controlled support matrix specifies predecessor +binaries, fixture coverage, and expected routes; it is test configuration, not +graph state. Test the declared route, never whichever happens to pass: + +| Route | Required result | +|---|---| +| Direct compatibility | Candidate reads/writes/reopens predecessor graphs without migration; preservation checks pass. | +| Explicit upgrade | Incompatible serving refuses before conversion; complete registered route and all handler/post-upgrade tests pass. | + +Cover every persisted feature available in each predecessor; extend fixtures +for new persisted state. Format/target changes require complete routes from +every supported conversion input and tests in the same change. Keep refusal +tests for unsupported formats; no support deletion merely to pass CI. + +Storage-maintainer review covers persisted formats, handler contracts, fixtures, +expectations, support scope, and gate configuration. Required binaries, executed +case inventories, and non-skipped cases must all be present. Fixture coverage +limits detection of unexercised breaks; this RFC configures neither CI nor review +enforcement itself. + +| Evasion | Defense | +|---|---| +| Omit format bump | Real predecessor journeys on every change. | +| Empty handler | Exact state, later writes, recovery, and work assertions. | +| Missing binary, skipped/filtered cases | Fail missing executables, skipped cases, or zero matches against required inventory. | +| Weaken expectations/delete fixtures | Storage-maintainer review of coverage and expectation changes. | +| Bypass handler restrictions or call full copying metadata-only | Restricted capabilities plus independent operation accounting; reject undeclared payload work or existing-object mutations. | +| Hide unbounded metadata work or require cleanup to finish conversion | Fixture metadata budgets; crash and preservation checks before cleanup, followed by safe reclamation tests. | + +| Author route | Acceptance | +|---|---| +| Preserve compatibility | Direct journeys pass; no new handler. | +| Convert supported input | Format change, handler, and tests together; upgrade journeys pass. | +| Intentional query representation change | Reviewed equivalent-semantics expectations; retain storage assertions. | +| Explicit support-policy change | Maintainer-approved policy/matrix amendment plus refusal coverage; no automatic missing-handler exception. | + +## Rollout + +1. Resolve decisions below and record the protocol. Before design acceptance, + obtain prototype evidence for in-place staging/activation, history, clocks, + recovery, safe deferred cleanup, and genuine v6-binary compatibility. +2. Qualify the production handler against the full matrix on every advertised + backend; prototype acceptance does not replace shipping evidence. +3. Ship check and qualified v6 to v7 execution with upgrade/versioning guidance, + required compatibility CI, and genuine-binary evidence. Check may ship earlier + only if execution unavailability is explicit. +4. Integrate RFC 0040 and settled fork handlers under their own gates; refuse unsupported + chains before effects. + +Target v0.11 only if gates pass. The release maintainer chooses delay for this +work or the documented rebuild path; v1.0 goals waive no gates or imply a date. +This draft changes no runtime behavior. + +## Unresolved questions + +Proposed responsibilities, not named approvals. Close all before acceptance; +record actual decider and evidence in the decision log. + +| Decision | Responsible role | Closure evidence | +|---|---|---| +| How does in-place append-only activation use the existing publication path? | Storage maintainer | Prove restricted handler effects, all-branch activation, history/clocks, writer fencing, recovery, rollback and deferred cleanup on pinned Lance. Any exception needs explicit effects, cost and qualification. | +| Cluster entry point? | Cluster maintainer with policy maintainer concurrence | Invocation, applied configuration, cutover, runtime exclusion, action/scope admission and old-format loading; preserve RFC 0040 declaration; qualify direct/cluster refusals. | +| Preserve or translate historical locators? | Storage maintainer with change-feed maintainer concurrence | Enumerate snapshot selectors and cursor/token bindings; prove state resolution and clock ordering. Preserve continuations or specify explicit refusal/restart; no implicit renumbering. | + +Release maintainer verifies production qualification before shipping. + +## Decision log + +No maintainer decision recorded yet. diff --git a/docs/rfcs/README.md b/docs/rfcs/README.md index 3753ef78..fb0d1c65 100644 --- a/docs/rfcs/README.md +++ b/docs/rfcs/README.md @@ -38,7 +38,7 @@ issue and implementation PR are usually enough. - Do not create `pre-merge`, `final`, `v2`, `internal`, or review-ledger copies. Revise the canonical file; preserve meaningful changes in its decision log. -The next available number is **0064**; lower gaps are historical and must +The next available number is **0065**; lower gaps are historical and must not be reused (0047 and 0048 are allocated by PR #606; 0050 by the `rfc/0050-engine-crate-topology` branch; 0056 by PR #670; 0058 by PR #662 for retained merged ancestry; 0059 by PR #675; 0060 by PR #677). @@ -187,3 +187,4 @@ This table is the human index for the canonical RFC corpus. | [0061](0061-managed-cluster-lifecycle.md) | Managed cluster lifecycle and config preparation | maintainer | accepted | complete | | [0062](0062-manifest-version-clock.md) | Manifest version as the table registration clock | maintainer | draft | in-progress | | [0063](0063-self-contained-branch-lineage.md) | Self-contained branch lineage | maintainer | draft | in-progress | +| [0064](0064-explicit-storage-upgrades.md) | Explicit storage upgrades | maintainer | draft | in-progress | diff --git a/docs/user/cli/reference.md b/docs/user/cli/reference.md index 557d6c8a..46d1521e 100644 --- a/docs/user/cli/reference.md +++ b/docs/user/cli/reference.md @@ -54,6 +54,7 @@ server resolves the actor from the bearer token. Drop it, or use `--store ` | `schema apply` | Apply a schema to a standalone graph | direct | | `schema plan` | Preview a schema migration | direct | | `lint` | Validate `.gq` source | local schema or direct graph | +| `upgrade` | Check or execute a registered offline storage migration | direct standalone | | `optimize` | Compact data and reconcile declared indexes | direct | | `rebuild-full-text-indexes` | Replace full-text indexes on one branch | direct | | `repair` | Preview or publish classified storage drift | direct | @@ -123,6 +124,20 @@ when it changed unrelated data. A mismatch has no effect and exits with code 4; JSON output includes `precondition_failure` with `expected` and optional `actual` commit ids. Re-read and decide again instead of retrying blindly. +## Storage upgrade + +```bash +omnigraph upgrade ./graph.omni --check --to-format 7 --json +omnigraph upgrade ./graph.omni --to-format 7 --json +``` + +`--store` is an alternative to the positional storage URI. Target format defaults +to 7. `--check` performs read-only preflight; execution requires stopped writers, +stopped maintenance and a verified whole-root backup. A failed check, refusal or +required recovery exits 1. JSON reports the route, formats, findings, durable +boundary, recovery action and work categories. Server and cluster addressing +are refused. See [storage migration](../operations/upgrade.md#explicit-v6-to-v7-storage-migration). + ## Load modes `load --mode` is required: diff --git a/docs/user/operations/upgrade.md b/docs/user/operations/upgrade.md index c79c1150..db3d7a4f 100644 --- a/docs/user/operations/upgrade.md +++ b/docs/user/operations/upgrade.md @@ -1,19 +1,90 @@ # Upgrading OmniGraph -OmniGraph intentionally supports one storage format per binary. When a release -changes that format, the new binary refuses the old graph and tells you which -release line can export it. Upgrade by rebuilding at a new URI: - -1. export the schema and one branch with a compatible old binary; -2. initialize a new graph with the new binary; -3. load the export; -4. verify and cut over; -5. retire the old graph only after the cutover is proven. - -Ordinary patch/minor upgrades that keep the same storage format do not require -this export/import procedure. Derived indexes may still need rebuilding, as -below, and CLI/API compatibility is independent of graph storage. Check the -[release notes](../../releases/) before upgrading. +Normal open accepts the current storage format. Use explicit storage migration +for a registered route, or export/import with a source-compatible binary when +no route exists. Storage formats, release versions and full-text index formats +are separate; check the [release notes](../../releases/) before upgrading. + +## Explicit v6 to v7 storage migration + +`omnigraph upgrade` converts standalone v6 graphs created by the 0.9.x/0.10.x +release lines to v7 in the same location. It appends manifest metadata and reuses +table data. It preserves branch ancestry, IDs, property values, schema identity, +retained commit IDs and numeric snapshots. Historical v6 snapshots keep their +original metadata and use an explicit legacy decoder after root admission. + +1. Stop every server, embedded writer, maintenance process and cluster apply + that could touch the graph or its shared dependencies. A process-local lock + cannot stop an already-open old binary in another process. +2. Preserve and verify a restorable backup of the entire root, including branch + references and historical data. Keep the source-compatible executable. +3. Run preflight with the new binary: + + ```bash + omnigraph upgrade ./graph.omni --check --to-format 7 --json + ``` + +4. Inspect `outcome`, `findings`, `route` and `work`. A passing check is advisory; + execution repeats validation. Resolve source recovery with the compatible + source executable before retrying. Shared Lance files outside the root refuse. + `work.external_blob_exclusions` lists external URI bytes whose immutability + and backup are outside the migration guarantee; their descriptors are retained. + `work.historical_blob_identity_limits` lists pre-0.10 Blob fields without + stable property IDs. Their bytes are preserved, but existing historical + delivery restrictions remain after their current physical entry changes; + migration cannot invent missing property-lifetime evidence. See + [Blob identity](../../releases/v0.10.0.md#blob-identity-and-rollback). + Validation-read bytes may be unknown and are reported as JSON `null`. +5. Execute while the graph remains offline: + + ```bash + omnigraph upgrade ./graph.omni --to-format 7 --json + ``` + +6. Verify reads on every branch and retained snapshot, then start only the new + fleet. Keep the backup for rollback. Restore the complete pre-upgrade backup + with the old executable; old bytes remaining in the upgraded root do not + make downgrading safe. Post-upgrade writes are absent from that backup. + +`--to-format` defaults to the binary's declared target, currently 7. Unsupported +sources and targets refuse; there is no automatic data-moving fallback. Both +check and execution return zero only for success (`check_passed`, `completed` +or `already_current`). Repeated successful execution is a no-write no-op. + +After the early fence, ordinary opens refuse until every branch is converted +and validated. If interrupted, retain the backup and rerun the same command +with `--to-format 7`, without `--check`, using this upgrade-capable executable. +The report identifies the last durable boundary and required recovery action. +Unknown ownership or foreign branch movement requires investigation; never +delete the pending marker to force serving or point the source executable at it. + +Branch naming must also be unambiguous. A native name ending in a ULID-shaped +suffix could be either a v0.9 logical name or a newer branch incarnation. The +handler requires a logical-head commit written after that native branch's fork +to prove the interpretation. Otherwise it refuses before writing, including +unused suffixed branches without that evidence. Duplicate logical names and +incarnation-shaped inner path segments also refuse. Resolve the branch naming +with the source executable or use the export/rebuild fallback; do not rename +native Lance refs or edit their metadata manually. + +This initial handler bounds each retained manifest to 1,000,000 rows and 64 MiB +of decoded batch metadata, 1,024 native branches and 100,000 retained versions +per branch. Version references are counted before historical manifests are +loaded. Exceeding a bound refuses before conversion. Payload bytes copied +and rewritten are zero; managed Blob validation can still read substantial data. + +Server and cluster selectors, cluster profiles and recognized cluster-layout +roots refuse until a cluster upgrade protocol is qualified. Local paths, file +URIs and symlink aliases are resolved before the cluster ownership check. Direct path access +is an operator interface; it cannot prove that an arbitrary root is unmanaged. +Embedded callers must supply exclusive control and, where installed, the policy +checker to `upgrade_storage_as`, which checks `SchemaApply` for every branch. + +The genuine predecessor CI journeys cover local standalone roots. Other backend +qualification is separate; see the [support matrix](../../dev/versioning.md#storage-upgrade-support-matrix). +Storage migration does not rebuild full-text indexes. Use the procedure below +when old index analyzers are incompatible. Formats without a registered route +still use the export/import rebuild procedure later in this guide. ## v0.9 to v0.10 diff --git a/scripts/check-storage-upgrade-ci.py b/scripts/check-storage-upgrade-ci.py new file mode 100644 index 00000000..b73219e8 --- /dev/null +++ b/scripts/check-storage-upgrade-ci.py @@ -0,0 +1,230 @@ +#!/usr/bin/env python3 +"""Require every storage migration coverage scope, with no empty or skipped runs.""" + +import argparse +import json +import re +import sys +import unittest +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] +CONTEXT = "Storage Upgrade Compatibility" +FEATURES = "omnigraph-engine/failpoints,omnigraph-cluster/failpoints" +CASES = ( + "genuine_v09_explicit_storage_upgrade_preserves_history", + "genuine_v010_explicit_storage_upgrade_preserves_history", + "storage_upgrade_refuses_cluster_path_aliases", + "genuine_v09_storage_upgrade_refuses_ambiguous_branch_names", +) +ENGINE_CASES = ( + "storage_upgrade_check_has_no_local_store_effects", + "storage_upgrade_interruption_boundaries_retry_without_mixed_visibility", + "storage_upgrade_recovery_refuses_foreign_head_movement", + "storage_upgrade_tracks_metadata_writes_and_no_payload_effects", + "storage_upgrade_policy_denial_precedes_effects", + "storage_upgrade_refuses_unknown_ownership_and_source", + "storage_upgrade_refuses_preexisting_recovery_without_healing", + "storage_upgrade_current_main_refuses_legacy_branch_without_effects", + "storage_upgrade_history_budget_precedes_manifest_reads", +) +SCOPES = { + "crossversion": ( + 'cargo test --workspace --locked --test crossversion_upgrade --features "$FAILPOINT_FEATURES" storage_upgrade -- --test-threads=1', + "crates/omnigraph-cli/tests/crossversion_upgrade.rs", + "", + ), + "engine": ( + "cargo test --locked -p omnigraph-engine --lib --features failpoints db::manifest::upgrade::tests -- --test-threads=1", + "crates/omnigraph/src/db/manifest/upgrade/tests.rs", + "db::manifest::upgrade::tests::", + ), + "lance": ( + "cargo test --locked -p omnigraph-engine --test lance_version_columns --features failpoints -- --test-threads=1", + "crates/omnigraph/tests/lance_version_columns.rs", + "", + ), + "protocol": ( + "cargo test --locked -p omnigraph-engine --test forbidden_apis --features failpoints -- --test-threads=1", + "crates/omnigraph/tests/forbidden_apis.rs", + "", + ), +} + + +def scope_script(scope: str) -> str: + command = SCOPES[scope][0] + return ( + "set -euo pipefail\n" + f'test_log="$RUNNER_TEMP/storage-upgrade-{scope}.log"\n' + f'{command} 2>&1 | tee "$test_log"\n' + f'python3 scripts/check-storage-upgrade-ci.py --check-log {scope} "$test_log"' + ) + + +def validate(workflow: str, policy: dict) -> list[str]: + failures = [] + match = re.search( + r"^ storage_upgrade_compatibility:\n(.*?)(?=^ \w+:|\Z)", + workflow, + re.MULTILINE | re.DOTALL, + ) + if match is None: + return ["CI must define storage_upgrade_compatibility"] + job = match.group(1) + trigger = workflow.split("\njobs:\n", 1)[0] + if re.search(r"^\s+(?:paths|paths-ignore):", trigger, re.MULTILINE): + failures.append("CI storage compatibility must not filter changed paths") + for event in ("pull_request", "push"): + if not re.search(rf"^ {event}:", trigger, re.MULTILINE): + failures.append(f"CI must run storage compatibility on {event}") + if re.search(r"^ (?:if|needs|continue-on-error):", job, re.MULTILINE): + failures.append("storage compatibility must run unconditionally and fail closed") + if re.search(r"^ (?:- | )(?:if|continue-on-error):", job, re.MULTILINE): + failures.append("storage compatibility steps must not skip or ignore failures") + if not re.search(rf"^ FAILPOINT_FEATURES: {re.escape(FEATURES)}$", trigger, re.MULTILINE): + failures.append("storage compatibility requires the canonical workspace failpoint features") + for token in ( + f"name: {CONTEXT}", + "OMNIGRAPH_REQUIRE_STORAGE_UPGRADE_TESTS: '1'", + "VERSION=v0.9.0", + "VERSION=v0.10.0", + "OMNIGRAPH_V09_BIN=", + "OMNIGRAPH_V6_BIN=", + "bash scripts/install.sh", + "run: python3 scripts/check-storage-upgrade-ci.py --self-test", + ): + if token not in job: + failures.append(f"storage compatibility is missing {token!r}") + scripts = re.findall( + r"^ run: \|\n((?:^ .*\n|^\n)+)", job, re.MULTILINE + ) + scripts = {"\n".join(line[10:] for line in body.rstrip().splitlines()) for body in scripts} + for scope in SCOPES: + if scope_script(scope) not in scripts: + failures.append(f"storage compatibility requires the exact fail-closed {scope} command and log check") + contexts = policy.get("required_status_checks", {}).get("contexts", []) + if CONTEXT not in contexts: + failures.append(f"branch protection must require {CONTEXT}") + return failures + + +def expected_cases(scope: str, root: Path = ROOT) -> set[str]: + _, source, prefix = SCOPES[scope] + names = set(re.findall( + r"^#\[(?:tokio::)?test[^\n]*\]\n(?:#\[[^\n]*\]\n)*(?:async )?fn (\w+)\(", + (root / source).read_text(), re.MULTILINE, + )) + if scope == "crossversion": + names = {name for name in names if "storage_upgrade" in name} | set(CASES) + elif scope == "engine": + names |= set(ENGINE_CASES) + return {prefix + name for name in names} + + +def validate_log(log: str, expected: set[str]) -> list[str]: + failures = [] + log = re.sub(r"\x1b\[[0-9;]*m", "", log) + if not expected: + failures.append("required test inventory is empty") + if re.search(r"\b(?:skipping|skipped)\b", log, re.IGNORECASE): + failures.append("required storage upgrade coverage skipped") + passed = set(re.findall(r"^test ([\w:]+) \.\.\. ok$", log, re.MULTILINE)) + for name in sorted(expected - passed): + failures.append(f"required storage upgrade case {name} did not pass") + summaries = re.findall( + r"^test result: ok\. (\d+) passed; (\d+) failed; (\d+) ignored; (\d+) measured; \d+ filtered out;", log, re.MULTILINE + ) + if len(summaries) != 1: + failures.append("required test run must have exactly one successful summary") + elif summaries[0] != (str(len(passed)), "0", "0", "0") or not passed: + failures.append("required test run must execute positive coverage with no failures or ignored cases") + return failures + + +class GuardTests(unittest.TestCase): + def setUp(self): + self.workflow = (ROOT / ".github/workflows/ci.yml").read_text() + self.policy = json.loads((ROOT / ".github/branch-protection.json").read_text()) + + def test_current_configuration(self): + self.assertEqual(validate(self.workflow, self.policy), []) + + def test_missing_or_changed_execution_fails(self): + for scope, (command, _, _) in SCOPES.items(): + for replacement in ("true", f"if false; then {command}; fi", command + " || true"): + with self.subTest(scope=scope, replacement=replacement): + changed = self.workflow.replace(command, replacement) + self.assertNotEqual(changed, self.workflow) + self.assertTrue(validate(changed, self.policy)) + + def test_old_package_feature_selection_fails(self): + changed = self.workflow.replace( + "cargo test --workspace --locked --test crossversion_upgrade", + "cargo test --locked -p omnigraph-cli --test crossversion_upgrade", + ) + self.assertTrue(validate(changed, self.policy)) + + def test_conditional_job_and_steps_fail(self): + for line in (" if: false\n", " needs: classify_changes\n", " continue-on-error: true\n"): + changed = self.workflow.replace(" storage_upgrade_compatibility:\n", " storage_upgrade_compatibility:\n" + line) + self.assertTrue(validate(changed, self.policy)) + for line in (" if: false\n", " continue-on-error: true\n"): + changed = self.workflow.replace(" - name: Run required storage upgrade engine tests\n", " - name: Run required storage upgrade engine tests\n" + line) + self.assertNotEqual(changed, self.workflow) + self.assertTrue(validate(changed, self.policy)) + + def test_missing_log_check_and_failpoints_fail(self): + for scope in SCOPES: + changed = self.workflow.replace(f"--check-log {scope}", "--check-log invalid") + self.assertTrue(validate(changed, self.policy)) + self.assertTrue(validate(self.workflow.replace(FEATURES, ""), self.policy)) + + def test_missing_required_context_fails(self): + self.assertTrue(validate(self.workflow, {})) + + def test_log_requires_every_case_and_positive_unskipped_summary(self): + good = "test alpha ... ok\ntest beta ... ok\ntest result: ok. 2 passed; 0 failed; 0 ignored; 0 measured; 4 filtered out; finished in 1s\n" + expected = {"alpha", "beta"} + self.assertEqual(validate_log(good, expected), []) + for log in ( + "", good.replace("test beta ... ok\n", ""), + good.replace("test beta ... ok", "test beta ... ignored"), + good.replace("2 passed", "0 passed"), + good.replace("0 ignored", "1 ignored"), + good + "skipping explicit storage upgrade: missing predecessor\n", + good + good, + ): + with self.subTest(log=log): + self.assertTrue(validate_log(log, expected)) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--self-test", action="store_true") + parser.add_argument("--check-log", nargs=2, metavar=("SCOPE", "PATH")) + args = parser.parse_args() + if args.self_test: + result = unittest.TextTestRunner().run(unittest.defaultTestLoader.loadTestsFromTestCase(GuardTests)) + return 0 if result.wasSuccessful() else 1 + if args.check_log: + scope, path = args.check_log + if scope not in SCOPES: + parser.error(f"unknown test scope: {scope}") + failures = validate_log(Path(path).read_text(), expected_cases(scope)) + else: + failures = validate( + (ROOT / ".github/workflows/ci.yml").read_text(), + json.loads((ROOT / ".github/branch-protection.json").read_text()), + ) + if failures: + for failure in failures: + print(f"Storage upgrade CI: {failure}", file=sys.stderr) + return 1 + print("Storage upgrade CI OK (required predecessors, refusal, recovery, Lance and protocol coverage).") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())