diff --git a/changelog.d/branch-create-at-a-commit.added.md b/changelog.d/branch-create-at-a-commit.added.md new file mode 100644 index 000000000..2b68aea72 --- /dev/null +++ b/changelog.d/branch-create-at-a-commit.added.md @@ -0,0 +1,14 @@ +- `branch create` starts a branch at a past graph commit: `--from-commit` on + the CLI, `from commit ""` in the `branch create` statement (GQ language + version 2.3) and `from_commit` on `POST /branches`. The new branch holds that + commit's rows and history, so a fix or revert can be tried where a change was + made and merged back with the commit as its base. A commit `cleanup` no + longer retains (`410`), one written under an earlier schema or before the + storage upgrade, one whose branch was deleted, an unpublished commit and an + unattached snapshot (`409`) start no branch; the new `branch_start_refusal` error detail names the + reason. Every branch creation receipt now carries `from_commit_id`, the + commit the new branch starts at, and the CLI prints it: + `created branch from at `. See + [Start a branch at a commit][branch-at-commit]. + +[branch-at-commit]: ../docs/user/branching/index.md#start-a-branch-at-a-commit diff --git a/crates/omnigraph-api-types/src/lib.rs b/crates/omnigraph-api-types/src/lib.rs index d8c42d4e6..cfdfd9ae5 100644 --- a/crates/omnigraph-api-types/src/lib.rs +++ b/crates/omnigraph-api-types/src/lib.rs @@ -390,8 +390,13 @@ pub struct SnapshotOutput { #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] pub struct BranchCreateRequest { - /// Parent branch to fork from. Defaults to `main`. + /// Parent branch to fork from; the new branch starts at its head. + /// Mutually exclusive with `from_commit`. Defaults to `main`. pub from: Option, + /// A published graph commit id to start the new branch at, with that + /// commit's data and history. Mutually exclusive with `from`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub from_commit: Option, /// Name of the new branch. Must not already exist. pub name: String, } @@ -402,6 +407,10 @@ pub struct BranchCreateOutput { pub from: String, pub name: String, pub actor_id: Option, + /// The graph commit the new branch starts at: `from`'s head when the + /// branch was created, or the commit `from_commit` named. For a commit + /// start, `from` is the branch that wrote the commit. + pub from_commit_id: String, } #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] @@ -612,8 +621,14 @@ pub struct ReadPageOutput { #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ToSchema)] #[serde(tag = "kind", rename_all = "snake_case")] pub enum BranchOutcomeOutput { - /// `name` was forked off `from`. - Created { from: String, name: String }, + /// `name` was forked off `from` at the commit `from_commit_id`: the + /// head of `from`, or the commit a `from commit ""` start named, + /// which `from` wrote. + Created { + from: String, + name: String, + from_commit_id: String, + }, /// `name` was removed. Deleted { name: String }, /// `source` was merged into `target`; `merge` is the three-way result, @@ -2106,6 +2121,42 @@ pub struct ChangeDiffRefusalOutput { pub type_name: Option, } +/// Why a branch could not start at the graph commit `from_commit` named. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, ToSchema)] +#[serde(rename_all = "snake_case")] +pub enum BranchStartRefusalReason { + /// HTTP 410: cleanup reclaimed data the commit's snapshot names. + NotRetained, + /// HTTP 409: the commit was written under an earlier schema. + SchemaChanged, + /// HTTP 409: the commit was written before the storage upgrade. + EarlierStorageFormat, + /// HTTP 409: the branch that wrote the commit was deleted, and may have + /// been created again under the same name since. + BranchDeleted, + /// HTTP 409: the commit is unpublished; publish it first. + UnpublishedCommit, + /// HTTP 409: the commit is an unattached snapshot, never branch history. + UnattachedSnapshot, + /// A reason added by a newer server; the branch was not created. + #[serde(other)] + Unknown, +} + +/// A branch create refused at the graph commit `from_commit` named; nothing +/// was created. +#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] +pub struct BranchStartRefusalOutput { + pub reason: BranchStartRefusalReason, + pub graph_commit_id: String, + /// For `not_retained`, the graph type whose stored version cleanup reclaimed. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub type_name: Option, + /// For `branch_deleted`, the branch that wrote the commit. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub branch: Option, +} + /// A selected full-text index cannot safely serve the current analyzer (HTTP 409). /// This is not a retryable write conflict: an operator must rebuild the live /// branch's indexes. Historical snapshots stay unchanged; branch old content @@ -2282,6 +2333,10 @@ pub struct ErrorOutput { /// commit or an unprovable schema boundary). #[serde(default, skip_serializing_if = "Option::is_none")] pub change_diff_refusal: Option, + /// Set with HTTP 409, or 410 for `not_retained`, when `branch create` + /// cannot start a branch at the commit `from_commit` named. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub branch_start_refusal: Option, /// Set with HTTP 409 when a selected full-text index requires an explicit /// rebuild before search can succeed. Unlike a write-authority conflict, /// this condition is not cleared by retrying. This additive discriminator @@ -2318,6 +2373,7 @@ impl ErrorOutput { blob_precondition_failure: None, change_feed_gap: None, change_diff_refusal: None, + branch_start_refusal: None, full_text_index_rebuild_required: None, full_text_index_required: None, diagnostic: None, diff --git a/crates/omnigraph-catalog/src/legacy.rs b/crates/omnigraph-catalog/src/legacy.rs index 58481105d..e2c18801e 100644 --- a/crates/omnigraph-catalog/src/legacy.rs +++ b/crates/omnigraph-catalog/src/legacy.rs @@ -479,7 +479,7 @@ pub(crate) enum RetiredHead { /// `Some` when `dataset`'s version is read from the legacy history of a root /// that has one: it predates native commit records, or is a supported pending /// conversion. The value says whether it carries the pending key. -fn legacy_version(dataset: &Dataset) -> Option { +pub(crate) fn legacy_version(dataset: &Dataset) -> Option { let metadata = &dataset.schema().metadata; let fenced = metadata.contains_key(UPGRADE_PENDING_KEY); let legacy = match metadata diff --git a/crates/omnigraph-catalog/src/lib.rs b/crates/omnigraph-catalog/src/lib.rs index 577770217..1a776d15b 100644 --- a/crates/omnigraph-catalog/src/lib.rs +++ b/crates/omnigraph-catalog/src/lib.rs @@ -2385,15 +2385,65 @@ impl ManifestCoordinator { #[doc(hidden)] pub async fn create_branch(&mut self, name: &str) -> Result<()> { crate::branch_names::ensure_logical_branch_name(name)?; - let mut ds = self.dataset.clone(); + let version = self.version(); + Self::create_branch_from(self.dataset.clone(), name, version).await + } + + /// Create the logical branch `name` as a fresh native incarnation whose + /// first `__manifest` version is a clone of this branch's version + /// `version`, the version whose publish made `graph_commit_id` the head: + /// the new branch starts with that commit's table pins, lineage and + /// schema contract, exactly as a branch created while the commit was the + /// head. The version must hold `graph_commit_id` as its head in the + /// format this build writes; a version only the legacy history serves is + /// refused. Every refusal here precedes any effect. + #[doc(hidden)] + pub async fn create_branch_at( + &mut self, + name: &str, + version: u64, + graph_commit_id: &str, + ) -> Result<()> { + crate::branch_names::ensure_logical_branch_name(name)?; + let dataset = if version == self.version() { + self.dataset.clone() + } else { + self.dataset + .checkout_version(version) + .await + .map_err(|error| OmniError::storage(error).before_effect())? + }; + if crate::legacy::legacy_version(&dataset).is_some() { + return Err(OmniError::branch_start_refused( + graph_commit_id, + crate::error::BranchStartRefusal::EarlierStorageFormat, + ) + .before_effect()); + } + let head = state::read_manifest_rows_projected(&dataset) + .await + .map_err(OmniError::before_effect)? + .head; + if head.graph_commit_id != graph_commit_id { + return Err(OmniError::manifest_internal(format!( + "`__manifest` version {version} of {} holds head commit '{}', not '{graph_commit_id}' \ + its record names; no branch was created", + self.native_branch.as_deref().unwrap_or("main"), + head.graph_commit_id + )) + .before_effect()); + } + Self::create_branch_from(dataset, name, version).await + } + + /// Create `name` as a clone of `ds`, checked out at `version`. + async fn create_branch_from(mut ds: Dataset, name: &str, version: u64) -> Result<()> { crate::branch_control::ensure_manifest_branch_create_namespace(&ds, name) .await .map_err(OmniError::before_effect)?; let native = crate::branch_names::native_branch_name(name, &crate::branch_names::mint_incarnation()); - match crate::branch_control::create_branch_recoverably(&mut ds, &native, self.version()) - .await? - { + match crate::branch_control::create_branch_recoverably(&mut ds, &native, version).await? { crate::branch_control::BranchCreateOutcome::Created => Ok(()), crate::branch_control::BranchCreateOutcome::RefAlreadyExists => Err( OmniError::manifest_conflict(format!("branch '{}' already exists", name)), diff --git a/crates/omnigraph-cli/src/app_tests.rs b/crates/omnigraph-cli/src/app_tests.rs index 14499267e..ad4b7c6b4 100644 --- a/crates/omnigraph-cli/src/app_tests.rs +++ b/crates/omnigraph-cli/src/app_tests.rs @@ -1082,7 +1082,11 @@ async fn maintenance_delivery_failure_preserves_one_rebuild_and_its_actor() { .await .unwrap(); client - .branch_create_from("main", "search", &mut Vec::new()) + .branch_create( + omnigraph::db::ReadTarget::branch("main"), + "search", + &mut Vec::new(), + ) .await .unwrap(); let main_before = client diff --git a/crates/omnigraph-cli/src/client.rs b/crates/omnigraph-cli/src/client.rs index 683be0e9c..644f0d22c 100644 --- a/crates/omnigraph-cli/src/client.rs +++ b/crates/omnigraph-cli/src/client.rs @@ -26,7 +26,7 @@ use std::sync::Arc; use color_eyre::Result; use color_eyre::eyre::{bail, eyre}; -use omnigraph::db::{Omnigraph, ReadTarget}; +use omnigraph::db::{Omnigraph, ReadTarget, SnapshotId}; use omnigraph::settings::{SessionSettings, SettingId, SettingValue, Source}; use omnigraph::{BLOB_READ_RANGE_MAX_BYTES, BlobContent, Session}; use omnigraph_api_types::{ @@ -43,7 +43,7 @@ use omnigraph_api_types::{ schema_apply_output, show_read_output, snapshot_payload, }; use omnigraph_compiler::catalog::Catalog; -use omnigraph_compiler::query::ast::{BranchRead, BranchWrite, CommitWrite}; +use omnigraph_compiler::query::ast::{BranchCreateFrom, BranchRead, BranchWrite, CommitWrite}; use omnigraph_compiler::query::parser::parse_query; use reqwest::header::{CONTENT_RANGE, RANGE}; use reqwest::{Method, StatusCode}; @@ -1544,12 +1544,22 @@ impl GraphClient { let query_name = write.statement_name().to_string(); let (branch, commit, outcome) = match write { BranchWrite::Create { name, from } => { - let from = from.unwrap_or_else(|| "main".to_string()); - self.branch_create_from(&from, &name, stderr).await?; + let start = match from { + Some(BranchCreateFrom::Commit(commit)) => { + ReadTarget::snapshot(SnapshotId::new(commit)) + } + Some(BranchCreateFrom::Branch(branch)) => ReadTarget::branch(branch), + None => ReadTarget::branch("main"), + }; + let created = self.branch_create(start, &name, stderr).await?; ( name.clone(), None, - BranchOutcomeOutput::Created { from, name }, + BranchOutcomeOutput::Created { + from: created.from, + name, + from_commit_id: created.from_commit_id, + }, ) } BranchWrite::Delete { name } => { @@ -2116,9 +2126,10 @@ impl GraphClient { } } - pub(crate) async fn branch_create_from( + /// Create `name` at `start`: a branch's head, or a published graph commit. + pub(crate) async fn branch_create( &self, - from: &str, + start: ReadTarget, name: &str, stderr: &mut dyn Write, ) -> Result { @@ -2133,9 +2144,17 @@ impl GraphClient { http, Method::POST, remote_url(base_url, &["branches"], &[])?, - Some(serde_json::to_value(BranchCreateRequest { - from: Some(from.to_string()), - name: name.to_string(), + Some(serde_json::to_value(match start { + ReadTarget::Branch(branch) => BranchCreateRequest { + from: Some(branch), + from_commit: None, + name: name.to_string(), + }, + ReadTarget::Snapshot(commit) => BranchCreateRequest { + from: None, + from_commit: Some(commit.as_str().to_string()), + name: name.to_string(), + }, })?), token.as_deref(), ) @@ -2144,13 +2163,13 @@ impl GraphClient { GraphClient::Embedded { uri, actor } => { let db = Self::open_write(uri, stderr).await?; let actor = actor.as_deref(); - db.branch_create_from_as(ReadTarget::branch(from), name, actor) - .await?; + let created = db.branch_create_from_as(start, name, actor).await?; Ok(BranchCreateOutput { uri: uri.clone(), - from: from.to_string(), + from: created.from, name: name.to_string(), actor_id: actor.map(String::from), + from_commit_id: created.from_commit_id, }) } GraphClient::Catalogue => Err(catalogue_scope_refusal("branch create from")), diff --git a/crates/omnigraph-cli/src/command_outcome.rs b/crates/omnigraph-cli/src/command_outcome.rs index b2e5cca24..aa5d5e790 100644 --- a/crates/omnigraph-cli/src/command_outcome.rs +++ b/crates/omnigraph-cli/src/command_outcome.rs @@ -231,6 +231,7 @@ impl Failure { || output.external_blob_source.is_some() || !output.merge_conflicts.is_empty() || output.diagnostic.is_some() + || output.branch_start_refusal.is_some() { Action::Refresh } else { @@ -293,6 +294,7 @@ pub(crate) fn is_plain_refusal(output: &ErrorOutput) -> bool { && output.blob_precondition_failure.is_none() && output.change_feed_gap.is_none() && output.change_diff_refusal.is_none() + && output.branch_start_refusal.is_none() && output.full_text_index_rebuild_required.is_none() && output.full_text_index_required.is_none() && output.diagnostic.is_none() diff --git a/crates/omnigraph-cli/src/commands/history/args.rs b/crates/omnigraph-cli/src/commands/history/args.rs index ece788303..17ac6bc21 100644 --- a/crates/omnigraph-cli/src/commands/history/args.rs +++ b/crates/omnigraph-cli/src/commands/history/args.rs @@ -50,8 +50,14 @@ pub(crate) enum BranchCommand { /// Graph URI #[arg(long)] uri: Option, + /// Branch to fork; the new branch starts at its head. Defaults to + /// `main`. #[arg(long)] from: Option, + /// A published graph commit id to start the branch at, with that + /// commit's data and history, instead of a branch head. + #[arg(long, conflicts_with = "from")] + from_commit: Option, name: String, #[arg(long)] json: bool, diff --git a/crates/omnigraph-cli/src/commands/history/branch.rs b/crates/omnigraph-cli/src/commands/history/branch.rs index 7585bf6da..dcd027968 100644 --- a/crates/omnigraph-cli/src/commands/history/branch.rs +++ b/crates/omnigraph-cli/src/commands/history/branch.rs @@ -5,6 +5,7 @@ use crate::client::{self, GraphClient}; use crate::helpers::{confirm_destructive, resolve_branch, write_target}; use crate::output::{write_json, write_merge_base_human}; use color_eyre::eyre::Result; +use omnigraph::db::{ReadTarget, SnapshotId}; use std::io::Write; pub(crate) async fn run_branch( @@ -18,6 +19,7 @@ pub(crate) async fn run_branch( BranchCommand::Create { uri, from, + from_commit, name, json, } => { @@ -33,7 +35,10 @@ pub(crate) async fn run_branch( stderr, ) .await?; - let from = resolve_branch(from, None, "main"); + let start = match from_commit { + Some(commit) => ReadTarget::snapshot(SnapshotId::new(commit)), + None => ReadTarget::branch(resolve_branch(from, None, "main")), + }; write_target( scope.quiet, "branch create", @@ -41,14 +46,14 @@ pub(crate) async fn run_branch( client.is_remote(), stderr, )?; - let payload = client.branch_create_from(&from, &name, stderr).await?; + let payload = client.branch_create(start, &name, stderr).await?; if json { write_json(&payload, stdout)?; } else { writeln!( stdout, - "created branch {} from {}", - payload.name, payload.from + "created branch {} from {} at {}", + payload.name, payload.from, payload.from_commit_id )?; } } diff --git a/crates/omnigraph-cli/src/output.rs b/crates/omnigraph-cli/src/output.rs index 419ccd589..6bd7bcdc6 100644 --- a/crates/omnigraph-cli/src/output.rs +++ b/crates/omnigraph-cli/src/output.rs @@ -1104,8 +1104,15 @@ pub(crate) fn write_blob_stat_human(output: &BlobStatOutput, writer: &mut dyn Wr /// prints; a mutation's prints the affected counts. pub(crate) fn write_change_human(output: &ChangeOutput, writer: &mut dyn Write) -> Result<()> { match &output.outcome { - Some(BranchOutcomeOutput::Created { from, name }) => { - writeln!(writer, "created branch {name} from {from}")?; + Some(BranchOutcomeOutput::Created { + from, + name, + from_commit_id, + }) => { + writeln!( + writer, + "created branch {name} from {from} at {from_commit_id}" + )?; } Some(BranchOutcomeOutput::Deleted { name }) => writeln!(writer, "deleted branch {name}")?, Some(BranchOutcomeOutput::Merged { diff --git a/crates/omnigraph-cli/tests/integration/cli_queries.rs b/crates/omnigraph-cli/tests/integration/cli_queries.rs index 872062271..3c580e547 100644 --- a/crates/omnigraph-cli/tests/integration/cli_queries.rs +++ b/crates/omnigraph-cli/tests/integration/cli_queries.rs @@ -67,7 +67,16 @@ fn branch_statements_embedded_print_the_verb_lines() { let (_temp, graph) = loaded_graph(); let created = output_success(embedded("mutate", &graph).arg("-e").arg("branch create b0")); - assert_eq!(stdout_string(&created), "created branch b0 from main\n"); + let start = parse_stdout_json(&output_success( + embedded("commit", &graph).args(["list", "--branch", "b0", "--json"]), + ))["commits"][0]["graph_commit_id"] + .as_str() + .unwrap() + .to_string(); + assert_eq!( + stdout_string(&created), + format!("created branch b0 from main at {start}\n") + ); let listed = output_success( embedded("query", &graph) @@ -236,7 +245,16 @@ fn branch_statements_remote_round_trip_and_delete_needs_consent() { let (_cluster, server) = served_graph(); let created = output_success(served("mutate", &server).arg("-e").arg("branch create b0")); - assert_eq!(stdout_string(&created), "created branch b0 from main\n"); + let start = parse_stdout_json(&output_success( + served("commit", &server).args(["list", "--branch", "b0", "--json"]), + ))["commits"][0]["graph_commit_id"] + .as_str() + .unwrap() + .to_string(); + assert_eq!( + stdout_string(&created), + format!("created branch b0 from main at {start}\n") + ); let table = stdout_string(&output_success( served("query", &server) diff --git a/crates/omnigraph-cli/tests/integration/parity_matrix.rs b/crates/omnigraph-cli/tests/integration/parity_matrix.rs index 8f6fb863d..a7261e9e9 100644 --- a/crates/omnigraph-cli/tests/integration/parity_matrix.rs +++ b/crates/omnigraph-cli/tests/integration/parity_matrix.rs @@ -322,6 +322,43 @@ fn parity_commit_list_branch() { assert_parity("commit list --branch", &l, &r); } +#[test] +fn parity_branch_create_at_a_commit() { + // The twins share their history, so one commit id names the same commit + // on both arms: the embedded arm starts the branch in the engine, the + // remote arm through `from_commit` on `POST /branches`. + let p = parity(); + let (listed, _) = p.run(&["commit", "list", "--branch", "main", "--json"]); + let head = parse_stdout_json(&listed)["commits"][0]["graph_commit_id"] + .as_str() + .unwrap() + .to_string(); + let (l, r) = p.run(&[ + "branch", + "create", + "--from-commit", + &head, + "at-head", + "--json", + ]); + assert_parity("branch create at a commit", &l, &r); + for (arm, output) in [("local", &l), ("remote", &r)] { + let payload = parse_stdout_json(output); + assert_eq!(payload["from"], "main", "{arm}"); + assert_eq!(payload["from_commit_id"], head, "{arm}"); + } + let (l, r) = p.run(&[ + "branch", + "create", + "--from-commit", + "01ARZ3NDEKTSV4RRFFQ69G5FAV", + "nowhere", + "--json", + ]); + assert_parity("branch create at an unknown commit", &l, &r); + assert!(!l.status.success(), "{l:?}"); +} + #[test] fn parity_commit_list_pages() { // The remote arm sends `?limit=&page_token=` while the embedded arm calls diff --git a/crates/omnigraph-cli/tests/support/mod.rs b/crates/omnigraph-cli/tests/support/mod.rs index 3673d39fa..576ebaea8 100644 --- a/crates/omnigraph-cli/tests/support/mod.rs +++ b/crates/omnigraph-cli/tests/support/mod.rs @@ -1269,6 +1269,7 @@ pub fn scrub_volatile(value: &mut serde_json::Value) { const VOLATILE_KEYS: &[&str] = &[ // identity-bearing per-instance values "commit_id", + "from_commit_id", "id", "__id", "parent_id", diff --git a/crates/omnigraph-compiler/src/lib.rs b/crates/omnigraph-compiler/src/lib.rs index c68d608f4..3c525642e 100644 --- a/crates/omnigraph-compiler/src/lib.rs +++ b/crates/omnigraph-compiler/src/lib.rs @@ -43,8 +43,9 @@ pub use ir::ParamMap; /// `(2, 0)` introduced the shared expression model; /// `(2, 1)` adds typed edge selections (`docs/rfcs/2026-09-30-typed-edge-alternation.md`); /// `(2, 2)` adds the `as commit` clause and the `commit publish` / `commit drop` -/// statements (`docs/rfcs/2026-10-10-unpublished-graph-commits.md`). -pub const GQ_LANGUAGE_VERSION: (u16, u16) = (2, 2); +/// statements (`docs/rfcs/2026-10-10-unpublished-graph-commits.md`); +/// `(2, 3)` adds `branch create from commit ""`. +pub const GQ_LANGUAGE_VERSION: (u16, u16) = (2, 3); pub use ir::lower::{lower_mutation_query, lower_query}; pub use lint::{DiagnosticCode, Family, SafetyTier, Severity}; pub use query::ast::Literal; diff --git a/crates/omnigraph-compiler/src/query/ast.rs b/crates/omnigraph-compiler/src/query/ast.rs index b634a3ca2..3aa19f804 100644 --- a/crates/omnigraph-compiler/src/query/ast.rs +++ b/crates/omnigraph-compiler/src/query/ast.rs @@ -390,14 +390,14 @@ impl BranchRead { } } -/// A control write: `branch create [from ]`, +/// A control write: `branch create [from | from commit ""]`, /// `branch delete `, or `branch merge [into ] [as commit]`. /// `from` and `into` are `None` when unspelled; no default is filled here. #[derive(Debug, Clone, PartialEq, Eq)] pub enum BranchWrite { Create { name: String, - from: Option, + from: Option, }, Delete { name: String, @@ -410,6 +410,16 @@ pub enum BranchWrite { }, } +/// Where `branch create` starts the new branch. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum BranchCreateFrom { + /// `from `: the new branch starts at the parent's head. + Branch(String), + /// `from commit ""`: the new branch starts at that published graph + /// commit. + Commit(String), +} + impl BranchWrite { /// The write's two keywords, `branch create` through `branch merge`. pub fn statement_name(&self) -> &'static str { diff --git a/crates/omnigraph-compiler/src/query/parser.rs b/crates/omnigraph-compiler/src/query/parser.rs index 449469b3f..239c5f0e0 100644 --- a/crates/omnigraph-compiler/src/query/parser.rs +++ b/crates/omnigraph-compiler/src/query/parser.rs @@ -323,6 +323,18 @@ fn parse_branch_stmt( let dry_run = has(Rule::dry_run_clause); let between = has(Rule::kw_to); let page = parse_diff_page_clause(&form)?; + let from_commit = form + .clone() + .into_inner() + .find(|inner| inner.as_rule() == Rule::create_from_commit) + .map(|pair| { + let literal = pair + .into_inner() + .find(|inner| inner.as_rule() == Rule::string_lit) + .expect("grammar: create_from_commit names one string literal"); + decode_graph_commit_id(&literal) + }) + .transpose()?; let names = form .into_inner() .filter(|inner| inner.as_rule() == Rule::branch_name) @@ -330,9 +342,12 @@ fn parse_branch_stmt( .collect::, QueryDiagnostic>>()?; let mut names = names.into_iter(); match (rule, names.next(), names.next()) { - (Rule::branch_create, Some(name), from) => { - Ok(BranchStmt::Write(BranchWrite::Create { name, from })) - } + (Rule::branch_create, Some(name), from) => Ok(BranchStmt::Write(BranchWrite::Create { + name, + from: from_commit + .map(BranchCreateFrom::Commit) + .or(from.map(BranchCreateFrom::Branch)), + })), (Rule::branch_delete, Some(name), None) => { Ok(BranchStmt::Write(BranchWrite::Delete { name })) } diff --git a/crates/omnigraph-compiler/src/query/parser_tests.rs b/crates/omnigraph-compiler/src/query/parser_tests.rs index dfc86b245..b485379f1 100644 --- a/crates/omnigraph-compiler/src/query/parser_tests.rs +++ b/crates/omnigraph-compiler/src/query/parser_tests.rs @@ -1489,7 +1489,14 @@ fn parse_branch(input: &str) -> BranchStmt { fn create(name: &str, from: Option<&str>) -> BranchStmt { BranchStmt::Write(BranchWrite::Create { name: name.to_string(), - from: from.map(str::to_string), + from: from.map(|from| BranchCreateFrom::Branch(from.to_string())), + }) +} + +fn create_at(name: &str, commit: &str) -> BranchStmt { + BranchStmt::Write(BranchWrite::Create { + name: name.to_string(), + from: Some(BranchCreateFrom::Commit(commit.to_string())), }) } @@ -1951,6 +1958,32 @@ fn branch_names_outside_the_identifier_alphabet_are_quoted() { ); assert!(parse_query("branch create from main").is_err()); assert!(parse_query("branch create b0 from").is_err()); +} + +#[test] +fn branch_create_from_commit_names_a_graph_commit() { + let id = "hb1.01ARZ3NDEKTSV4RRFFQ69G5FAV.3.01ARZ3NDEKTSV4RRFFQ69G5FAW"; + assert_eq!( + parse_branch(&format!(r#"branch create fix from commit "{id}""#)), + create_at("fix", id) + ); + assert_eq!( + parse_branch(r#"branch create "fix/import" from commit "01ARZ3NDEKTSV4RRFFQ69G5FAV""#), + create_at("fix/import", "01ARZ3NDEKTSV4RRFFQ69G5FAV") + ); + // A bare `commit` with no quoted id after it is a branch called `commit`. + assert_eq!( + parse_branch("branch create b0 from commit"), + create("b0", Some("commit")) + ); + assert_eq!( + parse_branch(r#"branch create b0 from "commit""#), + create("b0", Some("commit")) + ); + // The id is quoted; an unquoted word after `from commit` is refused. + assert!(parse_query("branch create b0 from commit main").is_err()); + assert!(parse_query(r#"branch create b0 from commit """#).is_err()); + assert!(parse_query(r#"branch create b0 from commit " x""#).is_err()); assert!(parse_query("branch merge b0 into").is_err()); assert!(parse_query("branch list all").is_err()); } diff --git a/crates/omnigraph-compiler/src/query/query.pest b/crates/omnigraph-compiler/src/query/query.pest index 14340d7dd..67066f9a1 100644 --- a/crates/omnigraph-compiler/src/query/query.pest +++ b/crates/omnigraph-compiler/src/query/query.pest @@ -54,7 +54,10 @@ kw_mode = @{ "mode" ~ !(ASCII_ALPHANUMERIC | "_") } kw_delimiter = @{ "delimiter" ~ !(ASCII_ALPHANUMERIC | "_") } kw_header = @{ "header" ~ !(ASCII_ALPHANUMERIC | "_") } kw_columns = @{ "columns" ~ !(ASCII_ALPHANUMERIC | "_") } -branch_create = { kw_create ~ branch_name ~ (kw_from ~ branch_name)? } +branch_create = { kw_create ~ branch_name ~ (kw_from ~ (create_from_commit | branch_name))? } +// `from commit ""`: the new branch starts at that graph commit. A bare +// `commit` with no quoted id after it names a branch called `commit`. +create_from_commit = { kw_commit ~ string_lit } branch_delete = { kw_delete ~ branch_name } branch_merge = { kw_merge ~ branch_name ~ (kw_into ~ branch_name)? ~ (as_commit_clause | dry_run_clause)? } branch_list = { kw_list } diff --git a/crates/omnigraph-core/src/error.rs b/crates/omnigraph-core/src/error.rs index be7ab0dc8..6710d7087 100644 --- a/crates/omnigraph-core/src/error.rs +++ b/crates/omnigraph-core/src/error.rs @@ -174,6 +174,61 @@ fn format_key_conflict(type_key: &str) -> String { format!("{} already has this id", graph_type_subject(type_key)) } +/// Why a branch cannot start at a graph commit. Each reason is decided before +/// any effect; none is approximated by a nearby commit. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum BranchStartRefusal { + /// Cleanup reclaimed a table version the commit's snapshot pins. + NotRetained { + table_key: String, + published_dataset_version: u64, + }, + /// The commit was written under another accepted schema than the graph's. + SchemaChanged, + /// The commit's `__manifest` version predates the storage upgrade; this + /// build reads it only through the legacy history. + EarlierStorageFormat, + /// The branch incarnation that wrote the commit was deleted: its name + /// names no branch now, or a later branch created under the same name. + BranchDeleted { branch: String }, + /// The commit is an unpublished proposal, which belongs to no branch's + /// history until it is published. + UnpublishedCommit, + /// The commit is an unattached snapshot, which never becomes branch + /// history. + UnattachedSnapshot, +} + +fn format_branch_start_refusal(graph_commit_id: &str, reason: &BranchStartRefusal) -> String { + let why = match reason { + BranchStartRefusal::NotRetained { + table_key, + published_dataset_version, + } => format!( + "it is no longer retained: cleanup reclaimed version {published_dataset_version} of \ + {}; start the branch at a retained commit", + graph_type_subject(table_key) + ), + BranchStartRefusal::SchemaChanged => "it was written under an earlier schema; start the \ + branch at a commit written under the current schema" + .to_string(), + BranchStartRefusal::EarlierStorageFormat => "it was written before the storage upgrade; \ + start the branch at a commit written after it" + .to_string(), + BranchStartRefusal::BranchDeleted { branch } => format!( + "branch '{branch}', which wrote it, was deleted; start the branch at a commit of a \ + live branch" + ), + BranchStartRefusal::UnpublishedCommit => "it is an unpublished commit; publish it with \ + `commit publish` and start the branch at the published commit" + .to_string(), + BranchStartRefusal::UnattachedSnapshot => { + "it is an unattached snapshot, which never becomes branch history".to_string() + } + }; + format!("cannot start a branch at commit '{graph_commit_id}': {why}") +} + fn merge_conflict_kind_label(kind: MergeConflictKind) -> &'static str { match kind { MergeConflictKind::DivergentInsert => "divergent_insert", @@ -293,6 +348,13 @@ pub enum OmniError { "'{target}' names both a branch and a commit; rename the branch, or name its head by commit id" )] AmbiguousReadTarget { target: String }, + /// `branch create` named a graph commit a branch cannot start at. Refused + /// before any effect; the commit itself is unchanged. + #[error("{}", format_branch_start_refusal(graph_commit_id, reason))] + BranchStartRefused { + graph_commit_id: String, + reason: BranchStartRefusal, + }, /// A change page or feed can no longer be reconstructed contiguously: /// cleanup reclaimed a published dataset version one of its commits pins. Recovery is /// the exact baseline handshake, never a retried continuation. @@ -943,6 +1005,16 @@ impl OmniError { ) } + pub fn branch_start_refused( + graph_commit_id: impl Into, + reason: BranchStartRefusal, + ) -> Self { + Self::BranchStartRefused { + graph_commit_id: graph_commit_id.into(), + reason, + } + } + pub fn manifest_row_level_cas_contention(message: impl Into) -> Self { Self::Manifest( ManifestError::new(ManifestErrorKind::Conflict, message) diff --git a/crates/omnigraph-gqt-core/src/lib.rs b/crates/omnigraph-gqt-core/src/lib.rs index 9923ee6c7..25e1dbdfd 100644 --- a/crates/omnigraph-gqt-core/src/lib.rs +++ b/crates/omnigraph-gqt-core/src/lib.rs @@ -16,9 +16,9 @@ use omnigraph::instrumentation::{QueryIoProbes, capture_query_io_probes, with_qu use omnigraph::loader::LoadMode; use omnigraph::{MutationReceipt, Session}; use omnigraph_compiler::query::ast::{ - BranchDiffSides, BranchRead, BranchStmt, BranchWrite, Clause, CommitWrite, EmptyFile, Expr, - FileBody, Literal, LoadStmt, MutationWriteMode, Param, QueryDecl, SettingStmt, UnattachedBase, - show_statement_name, + BranchCreateFrom, BranchDiffSides, BranchRead, BranchStmt, BranchWrite, Clause, CommitWrite, + EmptyFile, Expr, FileBody, Literal, LoadStmt, MutationWriteMode, Param, QueryDecl, SettingStmt, + UnattachedBase, show_statement_name, }; use omnigraph_compiler::query::parser::parse_query; use omnigraph_compiler::query::typecheck::{ @@ -289,7 +289,7 @@ pub struct ControlStep { pub enum ControlWrite { Create { name: String, - from: Option, + from: Option, expect: WriteExpect, }, Delete { @@ -2238,11 +2238,29 @@ pub fn parse_case(stem: &str, text: &str) -> Result { source: source.clone(), } } - BranchStmt::Write(write) => Pending::Control { - ordinal, - prefix: file.settings, - write: ControlSource::Branch(write), - }, + BranchStmt::Write(write) => { + if let BranchWrite::Create { + from: Some(BranchCreateFrom::Commit(commit)), + .. + } = &write + && let Some(name) = bound_operand(commit) + { + if !bound_names.contains(name) { + return Err(format!( + "`${{{name}}}` is not bound; `--- bind commit ${name}` must precede this step" + )); + } + if source.matches("${").count() == 1 { + substitutable_lines + .extend(section.body.iter().map(|(i, _)| *i)); + } + } + Pending::Control { + ordinal, + prefix: file.settings, + write: ControlSource::Branch(write), + } + } }); continue; } @@ -3893,11 +3911,17 @@ async fn run_control_step( from, expect, } => { - let parent = from.as_deref().unwrap_or(MAIN_BRANCH); + let start = match from { + Some(BranchCreateFrom::Commit(commit)) => ReadTarget::snapshot(SnapshotId::new( + resolve_operand(commit, bindings).map_err(&fail)?, + )), + Some(BranchCreateFrom::Branch(parent)) => ReadTarget::branch(parent.as_str()), + None => ReadTarget::branch(MAIN_BRANCH), + }; let outcome = operation( host, step.ordinal, - db.branch_create_from_as(ReadTarget::branch(parent), branch, None), + db.branch_create_from_as(start, branch, None), ) .await .map_err(&fail)?; diff --git a/crates/omnigraph-gqt-core/src/served.rs b/crates/omnigraph-gqt-core/src/served.rs index c9c4ce3a9..1f81004fd 100644 --- a/crates/omnigraph-gqt-core/src/served.rs +++ b/crates/omnigraph-gqt-core/src/served.rs @@ -11,7 +11,7 @@ use omnigraph_api_types::{ BranchMergeOutcome, BranchOutcomeOutput, ChangeOutput, ChangeRequest, HTTP_API_CONTRACT, HTTP_API_CONTRACT_HEADER, IngestRequest, QueryRequest, ReadOutput, SchemaOutput, }; -use omnigraph_compiler::query::ast::show_statement_name; +use omnigraph_compiler::query::ast::{BranchCreateFrom, show_statement_name}; use serde::de::DeserializeOwned; use serde::{Deserialize, Serialize}; use serde_json::Value; @@ -714,13 +714,22 @@ fn control_statement(write: &ControlWrite, bindings: &Bindings) -> Result format!( "branch create {} from {}", branch_name(name), branch_name(parent) ), + ControlWrite::Create { + name, + from: Some(BranchCreateFrom::Commit(commit)), + .. + } => format!( + "branch create {} from commit {}", + branch_name(name), + string_literal(&resolve_operand(commit, bindings)?) + ), ControlWrite::Create { name, .. } => format!("branch create {}", branch_name(name)), ControlWrite::Delete { name, .. } => format!("branch delete {}", branch_name(name)), ControlWrite::Merge { diff --git a/crates/omnigraph-gqt/README.md b/crates/omnigraph-gqt/README.md index 876952747..15c227e75 100644 --- a/crates/omnigraph-gqt/README.md +++ b/crates/omnigraph-gqt/README.md @@ -120,8 +120,9 @@ expected to succeed, binds the receipt's commit id to `$name` `--- query snapshot: $name` header reads at that commit instead of a branch, and a commit statement whose whole quoted operand is the name, `commit publish "${name}"` or `commit drop "${name}"`, is dispatched with the -bound id; this is the one place `${` may appear in a statement body, and the -engine receives an ordinary literal. A mutation declaration +bound id, as is a `branch create from commit "${name}"` whose +commit is the whole quoted operand; these are the places `${` may appear in a statement +body, and the engine receives an ordinary literal. A mutation declaration `as unattached branch on "${name}"` (RFC "Unattached graph branches") is substituted the same way, and a `--- params` body may spell a bound name, `{"base": "${name}"}`, beside the loop variable; a `--- mutate` header without diff --git a/crates/omnigraph-gqt/cases/branch_create_at_a_commit.gqt b/crates/omnigraph-gqt/cases/branch_create_at_a_commit.gqt new file mode 100644 index 000000000..f8f2cfdc5 --- /dev/null +++ b/crates/omnigraph-gqt/cases/branch_create_at_a_commit.gqt @@ -0,0 +1,115 @@ +# issue: none +# notes: `branch create from commit ""` starts a branch at a past commit of main: the branch reads that +# notes: commit's rows, its merge base with main is the commit, and it merges back three ways while main keeps its +# notes: later writes. Refusals are in branch_create_at_a_commit_refusals.gqt. + +--- runner +timeout_ms: 10000 +environments: + - target: omnigraph-engine + storage: local-filesystem + - target: omnigraph-engine-dst + storage: in-memory-object-store + seeds: [0] + - target: omnigraph-server + storage: local-filesystem + +--- schema +node Person { + name: String @key + age: I32 +} + +--- seed +{"type":"Person","data":{"name":"alice","age":30}} +{"type":"Person","data":{"name":"bob","age":40}} + +--- mutate +query insert_carol() { + insert Person { name: "carol", age: 22 } +} + +--- expect affected: nodes=1 edges=0 + +--- bind commit $carol + +--- mutate +query age_alice() { + update Person set { age: 31 } where name = "alice" +} + +--- expect affected: nodes=1 edges=0 + +--- mutate +query remove_bob() { + delete Person where name = "bob" +} + +--- expect affected: nodes=1 edges=0 + +--- bind commit $no_bob + +--- mutate +branch create fix from commit "${carol}" + +--- expect ok + +--- query branch: fix +query people() { + match { $p: Person } + return { $p.name, $p.age } +} + +--- expect unordered +{"p.name": "alice", "p.age": 30} +{"p.name": "bob", "p.age": 40} +{"p.name": "carol", "p.age": 22} + +--- expect shape +p.name: String +p.age: I32 + +--- mutate branch: fix +query insert_dave() { + insert Person { name: "dave", age: 50 } +} + +--- expect affected: nodes=1 edges=0 + +--- bind commit $dave + +--- query +branch merge-base fix + +--- expect ordered +{"source": "fix", "target": "main", "source_commit_id": "${dave}", "target_commit_id": "${no_bob}", "merge_base_commit_id": "${carol}"} + +--- mutate +branch merge fix into main + +--- expect outcome: merged + +--- query +query people() { + match { $p: Person } + return { $p.name, $p.age } +} + +--- expect unordered +{"p.name": "alice", "p.age": 31} +{"p.name": "carol", "p.age": 22} +{"p.name": "dave", "p.age": 50} + +--- expect shape +p.name: String +p.age: I32 + +--- query +branch list + +--- expect unordered +{"name": "fix"} +{"name": "main"} + +--- expect shape +name: String diff --git a/crates/omnigraph-gqt/cases/branch_create_at_a_commit_refusals.gqt b/crates/omnigraph-gqt/cases/branch_create_at_a_commit_refusals.gqt new file mode 100644 index 000000000..20b45f714 --- /dev/null +++ b/crates/omnigraph-gqt/cases/branch_create_at_a_commit_refusals.gqt @@ -0,0 +1,108 @@ +# issue: none +# notes: `branch create … from commit ""` refuses, before any effect, a start that is not branch history: an +# notes: unpublished commit (publish it first), an unattached snapshot, which never becomes branch history, and a commit +# notes: of a deleted branch, with one answer whether its name names nothing or a branch created again since; an id no +# notes: commit has is not found. +# notes: The start path itself is in branch_create_at_a_commit.gqt. + +--- runner +timeout_ms: 10000 +environments: + - target: omnigraph-engine + storage: local-filesystem + - target: omnigraph-engine-dst + storage: in-memory-object-store + seeds: [0] + - target: omnigraph-server + storage: local-filesystem + +--- schema +node Person { + name: String @key + age: I32 +} + +--- seed +{"type":"Person","data":{"name":"alice","age":30}} +{"type":"Person","data":{"name":"bob","age":40}} + +--- mutate +query propose_erin() as commit { + insert Person { name: "erin", age: 5 } +} + +--- expect affected: nodes=1 edges=0 + +--- bind commit $proposal + +--- mutate +branch create from_proposal from commit "${proposal}" + +--- expect error: is an unpublished commit + +--- mutate +query park_frank() as unattached branch { + insert Person { name: "frank", age: 6 } +} + +--- expect affected: nodes=1 edges=0 + +--- bind commit $snapshot + +--- mutate +branch create from_snapshot from commit "${snapshot}" + +--- expect error: is an unattached snapshot, which never becomes branch history + +--- mutate +branch create gone from main + +--- expect ok + +--- mutate branch: gone +query add_gil() { + insert Person { name: "gil", age: 7 } +} + +--- expect affected: nodes=1 edges=0 + +--- bind commit $gil + +--- mutate +branch delete gone + +--- expect ok + +--- mutate +branch create from_deleted from commit "${gil}" + +--- expect error: branch 'gone', which wrote it, was deleted + +--- mutate +branch create gone from main + +--- expect ok + +--- mutate +branch create from_reused from commit "${gil}" + +--- expect error: branch 'gone', which wrote it, was deleted + +--- mutate +branch delete gone + +--- expect ok + +--- mutate +branch create nowhere from commit "01ARZ3NDEKTSV4RRFFQ69G5FAV" + +--- expect error: commit '01ARZ3NDEKTSV4RRFFQ69G5FAV' not found + +--- query +branch list + +--- expect unordered +{"name": "main"} + +--- expect shape +name: String diff --git a/crates/omnigraph-server/src/handlers.rs b/crates/omnigraph-server/src/handlers.rs index bab1ce931..067ddcffc 100644 --- a/crates/omnigraph-server/src/handlers.rs +++ b/crates/omnigraph-server/src/handlers.rs @@ -11,7 +11,7 @@ use crate::serving::GraphRequest; use crate::workload::{AdmissionGuard, IngressLease}; use futures::StreamExt; use omnigraph::Session; -use omnigraph::db::MergeResult; +use omnigraph::db::{BranchCreated, MergeResult, SnapshotId}; use omnigraph::settings::{SettingId, SettingValue}; use omnigraph_compiler::Literal; use omnigraph_compiler::query::ast::{ @@ -2921,10 +2921,14 @@ async fn branch_list_body( request_body = BranchCreateRequest, responses( (status = 200, description = "Branch created", body = BranchCreateOutput), - (status = 400, description = "Bad request", body = ErrorOutput), + (status = 400, description = "Bad request, or both `from` and `from_commit`", body = ErrorOutput), (status = 401, description = "Unauthorized", body = ErrorOutput), (status = 403, description = "Forbidden", body = ErrorOutput), - (status = 409, description = "Branch already exists", body = ErrorOutput), + // A commit the actor may not branch from is indistinguishable from + // one that does not exist. + (status = 404, description = "`from` names no branch, or `from_commit` names no commit the actor may branch from", body = ErrorOutput), + (status = 409, description = "Branch already exists, or the commit `from_commit` names cannot start a branch; see branch_start_refusal", body = ErrorOutput), + (status = 410, description = "The commit `from_commit` names is no longer retained; see branch_start_refusal", body = ErrorOutput), (status = 429, description = "Per-actor admission cap exceeded; honor `Retry-After` header", body = ErrorOutput), (status = 503, description = "An overlapping durable recovery intent must be resolved before retry", body = ErrorOutput), ), @@ -2932,10 +2936,12 @@ async fn branch_list_body( )] /// Create a new branch. /// -/// Forks `name` off of `from` (defaults to `main`). The new branch shares -/// backing dataset data with its parent until it is mutated. Returns 409 if `name` -/// already exists. The GQ statement `branch create` on `POST /mutate` runs -/// the same body. +/// Forks `name` off of `from` (defaults to `main`) at its head, or starts it +/// at the published graph commit `from_commit` with that commit's data and +/// history. The receipt names the branch forked and the commit the new +/// branch starts at. The new branch shares backing dataset data with its +/// parent until it is mutated. Returns 409 if `name` already exists. The GQ +/// statement `branch create` on `POST /mutate` runs the same body. pub(crate) async fn server_branch_create( State(state): State, Extension(handle): Extension, @@ -2943,46 +2949,78 @@ pub(crate) async fn server_branch_create( actor: Option>, Json(request): Json, ) -> std::result::Result, ApiError> { - let from = request.from.unwrap_or_else(|| "main".to_string()); - branch_create_body( + let start = match (request.from, request.from_commit) { + (Some(_), Some(_)) => { + return Err(ApiError::bad_request( + "request may specify from or from_commit, not both", + )); + } + (_, Some(commit)) => ReadTarget::snapshot(SnapshotId::new(commit)), + (from, None) => ReadTarget::branch(from.unwrap_or_else(|| "main".to_string())), + }; + let created = branch_create_body( &state, &handle, actor.as_ref().map(|Extension(actor)| actor), ingress, - &from, + start, &request.name, ) .await?; Ok(Json(BranchCreateOutput { uri: handle.uri.clone(), - from, + from: created.from, name: request.name, actor_id: actor.map(|Extension(actor)| actor.actor_id.as_ref().to_string()), + from_commit_id: created.from_commit_id, })) } /// Body shared by `POST /branches` and the `branch create` statement: the -/// `branch_create` check on (`from`, `name`), admission, the engine call. +/// `branch_create` check on (the start's branch, `name`), admission, the +/// engine call. A start at a commit is checked on the branch a control on +/// the commit addresses, and a denial answers the 404 of an unknown commit, +/// as the commit controls do, so the route discloses no commit the actor +/// may not branch from. async fn branch_create_body( state: &AppState, handle: &GraphRequest, actor: Option<&AuthenticatedActor>, ingress: IngressLease, - from: &str, + start: ReadTarget, name: &str, -) -> std::result::Result<(), ApiError> { +) -> std::result::Result { let actor_arc = actor .map(|actor| Arc::clone(&actor.actor_id)) .unwrap_or_else(|| Arc::::from("anonymous")); - authorize_request( + let source = match &start { + ReadTarget::Branch(branch) => branch.clone(), + ReadTarget::Snapshot(commit_id) => handle + .engine + .commit_target_branch(commit_id.as_str()) + .await + .map_err(|error| change_route_commit_lookup_error(error, commit_id.as_str()))?, + }; + let decision = authorize( actor, handle.policy.as_deref(), PolicyRequest { action: PolicyAction::BranchCreate, - branch: Some(from.to_string()), + branch: Some(source), target_branch: Some(name.to_string()), }, )?; + match (decision, &start) { + (Authz::Allowed, _) => {} + (Authz::Denied(message), ReadTarget::Branch(_)) => { + return Err(ApiError::forbidden(message)); + } + (Authz::Denied(_), ReadTarget::Snapshot(commit_id)) => { + return Err(ApiError::not_found(format!( + "commit '{commit_id}' not found" + ))); + } + } // Branch metadata only — small constant bytes estimate. The Lance // shallow-clone work is bounded by the parent's manifest size, not // the request body. @@ -2991,12 +3029,11 @@ async fn branch_create_body( .try_admit(&actor_arc, 256) .map_err(ApiError::from_workload_reject)?; let engine = Arc::clone(&handle.engine); - let from = from.to_owned(); let name = name.to_owned(); let actor_id = actor.map(|actor| actor.actor_id.to_string()); owned_write(state, admission, ingress, handle.clone(), async move { engine - .branch_create_from_as(ReadTarget::branch(&from), &name, actor_id.as_deref()) + .branch_create_from_as(start, &name, actor_id.as_deref()) .await .map_err(ApiError::from_omni) }) diff --git a/crates/omnigraph-server/src/handlers/dispatch.rs b/crates/omnigraph-server/src/handlers/dispatch.rs index 79488b90e..b9b6c552c 100644 --- a/crates/omnigraph-server/src/handlers/dispatch.rs +++ b/crates/omnigraph-server/src/handlers/dispatch.rs @@ -7,8 +7,8 @@ use super::*; use crate::api::branch_statement_refusals as refusals; use omnigraph::loader::CsvLoadRequest; use omnigraph_compiler::query::ast::{ - BranchDiffSides, BranchRead, BranchStmt, BranchWrite, CommitWrite, EXPLAIN_STATEMENT_NAME, - LoadStmt, SettingStmt, show_statement_name, + BranchCreateFrom, BranchDiffSides, BranchRead, BranchStmt, BranchWrite, CommitWrite, + EXPLAIN_STATEMENT_NAME, LoadStmt, SettingStmt, show_statement_name, }; use omnigraph_compiler::settings::{SettingId, SettingRow}; @@ -321,12 +321,22 @@ pub(super) async fn run_branch_statement( let actor_id = actor.map(|actor| actor.actor_id.as_ref().to_string()); let (branch, commit, outcome) = match write { BranchWrite::Create { name, from } => { - let from = from.unwrap_or_else(|| "main".to_string()); - branch_create_body(state, handle, actor, ingress, &from, &name).await?; + let start = match from { + Some(BranchCreateFrom::Commit(commit)) => { + ReadTarget::snapshot(SnapshotId::new(commit)) + } + Some(BranchCreateFrom::Branch(branch)) => ReadTarget::branch(branch), + None => ReadTarget::branch("main"), + }; + let created = branch_create_body(state, handle, actor, ingress, start, &name).await?; ( name.clone(), None, - api::BranchOutcomeOutput::Created { from, name }, + api::BranchOutcomeOutput::Created { + from: created.from, + name, + from_commit_id: created.from_commit_id, + }, ) } BranchWrite::Delete { name } => { diff --git a/crates/omnigraph-server/src/lib.rs b/crates/omnigraph-server/src/lib.rs index a8b2590d6..fbdd9a7b9 100644 --- a/crates/omnigraph-server/src/lib.rs +++ b/crates/omnigraph-server/src/lib.rs @@ -68,7 +68,7 @@ use axum::routing::{any, delete, get, post}; use axum::{Json, Router}; use color_eyre::eyre::{Result, WrapErr, bail, eyre}; use omnigraph::db::{Omnigraph, ReadTarget}; -use omnigraph::error::{ManifestConflictDetails, ManifestErrorKind, OmniError}; +use omnigraph::error::{BranchStartRefusal, ManifestConflictDetails, ManifestErrorKind, OmniError}; use omnigraph::storage::normalize_root_uri; use omnigraph_compiler::catalog::Catalog; use omnigraph_compiler::json_params_to_param_map; @@ -493,6 +493,7 @@ enum ApiErrorDetails { BlobPreconditionFailure(api::BlobPreconditionFailureOutput), ChangeFeedGap(api::ChangeFeedGapOutput), ChangeDiffRefusal(api::ChangeDiffRefusalOutput), + BranchStartRefusal(api::BranchStartRefusalOutput), FullTextIndexRebuildRequired(api::FullTextIndexRebuildRequiredOutput), FullTextIndexRequired(api::FullTextIndexRequiredOutput), Diagnostic(api::DiagnosticOutput), @@ -1396,6 +1397,23 @@ impl ApiError { } } + /// HTTP 409, or 410 Gone when the commit is no longer retained: `branch + /// create` cannot start a branch at the commit `from` named, and nothing + /// was created. The `branch_start_refusal` detail carries the reason. + fn branch_start_refusal(message: String, details: api::BranchStartRefusalOutput) -> Self { + let status = match details.reason { + api::BranchStartRefusalReason::NotRetained => StatusCode::GONE, + _ => StatusCode::CONFLICT, + }; + Self { + completion_uncertain: false, + status, + code: (status == StatusCode::CONFLICT).then_some(ErrorCode::Conflict), + message: message.into_boxed_str(), + details: Some(Box::new(ApiErrorDetails::BranchStartRefusal(details))), + } + } + /// A refused query: HTTP 400 with the one-line message, plus the /// compiler's structured diagnostic when the refusal carries one. fn from_compiler(err: &omnigraph_compiler::error::CompilerError) -> Self { @@ -1546,6 +1564,60 @@ impl ApiError { } error @ OmniError::ReadTargetNotFound { .. } => Self::not_found(error.to_string()), error @ OmniError::AmbiguousReadTarget { .. } => Self::bad_request(error.to_string()), + error @ OmniError::BranchStartRefused { .. } => { + let message = error.to_string(); + let OmniError::BranchStartRefused { + graph_commit_id, + reason, + } = error + else { + unreachable!() + }; + let (reason, type_name, branch) = match reason { + BranchStartRefusal::NotRetained { table_key, .. } => { + let Ok((_, type_name)) = api::entity_type_parts(&table_key) else { + return Self::internal( + "branch start refusal named invalid graph type metadata", + ); + }; + ( + api::BranchStartRefusalReason::NotRetained, + Some(type_name.to_string()), + None, + ) + } + BranchStartRefusal::SchemaChanged => { + (api::BranchStartRefusalReason::SchemaChanged, None, None) + } + BranchStartRefusal::EarlierStorageFormat => ( + api::BranchStartRefusalReason::EarlierStorageFormat, + None, + None, + ), + BranchStartRefusal::BranchDeleted { branch } => ( + api::BranchStartRefusalReason::BranchDeleted, + None, + Some(branch), + ), + BranchStartRefusal::UnpublishedCommit => { + (api::BranchStartRefusalReason::UnpublishedCommit, None, None) + } + BranchStartRefusal::UnattachedSnapshot => ( + api::BranchStartRefusalReason::UnattachedSnapshot, + None, + None, + ), + }; + Self::branch_start_refusal( + message, + api::BranchStartRefusalOutput { + reason, + graph_commit_id, + type_name, + branch, + }, + ) + } // Retention loss under a change continuation: 410 with the // structured resume hint. Recovery is the baseline handshake. OmniError::ChangeFeedGap { @@ -1762,6 +1834,9 @@ impl ApiError { ApiErrorDetails::ChangeDiffRefusal(value) => { output.change_diff_refusal = Some(value) } + ApiErrorDetails::BranchStartRefusal(value) => { + output.branch_start_refusal = Some(value) + } ApiErrorDetails::FullTextIndexRebuildRequired(value) => { output.full_text_index_rebuild_required = Some(value) } @@ -1938,6 +2013,78 @@ mod api_error_tests { assert_eq!(gap.first_unreadable_commit_id, "01JTESTGAP"); } + #[tokio::test] + async fn branch_start_refusal_is_typed_409_or_410() { + for (reason, status, code, wire, type_name) in [ + ( + BranchStartRefusal::NotRetained { + table_key: "node:Person".to_string(), + published_dataset_version: 7, + }, + StatusCode::GONE, + None, + api::BranchStartRefusalReason::NotRetained, + Some("Person".to_string()), + ), + ( + BranchStartRefusal::SchemaChanged, + StatusCode::CONFLICT, + Some(ErrorCode::Conflict), + api::BranchStartRefusalReason::SchemaChanged, + None, + ), + ( + BranchStartRefusal::EarlierStorageFormat, + StatusCode::CONFLICT, + Some(ErrorCode::Conflict), + api::BranchStartRefusalReason::EarlierStorageFormat, + None, + ), + ( + BranchStartRefusal::BranchDeleted { + branch: "gone".to_string(), + }, + StatusCode::CONFLICT, + Some(ErrorCode::Conflict), + api::BranchStartRefusalReason::BranchDeleted, + None, + ), + ( + BranchStartRefusal::UnpublishedCommit, + StatusCode::CONFLICT, + Some(ErrorCode::Conflict), + api::BranchStartRefusalReason::UnpublishedCommit, + None, + ), + ( + BranchStartRefusal::UnattachedSnapshot, + StatusCode::CONFLICT, + Some(ErrorCode::Conflict), + api::BranchStartRefusalReason::UnattachedSnapshot, + None, + ), + ] { + let error = OmniError::branch_start_refused("01JTESTROOT", reason).before_effect(); + let message = error.to_string(); + let response = ApiError::from_omni(error).into_response(); + assert_eq!(response.status(), status); + let body = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + let error: ErrorOutput = serde_json::from_slice(&body).unwrap(); + assert_eq!(error.code, code); + assert_eq!(error.error, message); + let refusal = error.branch_start_refusal.expect("structured refusal"); + assert_eq!(refusal.reason, wire); + assert_eq!(refusal.graph_commit_id, "01JTESTROOT"); + assert_eq!(refusal.type_name, type_name); + assert_eq!( + refusal.branch.as_deref(), + (wire == api::BranchStartRefusalReason::BranchDeleted).then_some("gone") + ); + } + } + #[tokio::test] async fn change_diff_refusal_is_typed_409() { for (err, reason, type_name) in [ diff --git a/crates/omnigraph-server/src/mcp.rs b/crates/omnigraph-server/src/mcp.rs index 11337fb2e..ec7ab56f5 100644 --- a/crates/omnigraph-server/src/mcp.rs +++ b/crates/omnigraph-server/src/mcp.rs @@ -564,6 +564,8 @@ struct BranchCreateArguments { name: String, #[serde(default)] from: Option, + #[serde(default)] + from_commit: Option, } #[derive(Deserialize)] #[serde(deny_unknown_fields)] @@ -1159,6 +1161,7 @@ impl GraphTools { Some(Extension(actor)), Json(crate::api::BranchCreateRequest { from: args.from, + from_commit: args.from_commit, name: args.name, }), ) @@ -1765,7 +1768,8 @@ fn branch_create_schema() -> Value { arguments( json!({ "name": text_schema("The new branch's name."), - "from": text_schema("Branch to fork. Defaults to `main`."), + "from": text_schema("Branch to fork; the new branch starts at its head. Defaults to `main`."), + "from_commit": text_schema("A published graph commit id to start the branch at instead of a branch head; excludes `from`."), }), &["name"], ) diff --git a/crates/omnigraph-server/src/redacted_cause.rs b/crates/omnigraph-server/src/redacted_cause.rs index 5955c51b3..17a2d92fc 100644 --- a/crates/omnigraph-server/src/redacted_cause.rs +++ b/crates/omnigraph-server/src/redacted_cause.rs @@ -41,6 +41,7 @@ impl RedactedCause { OmniError::BranchNotFound { .. } => "BranchNotFound", OmniError::ReadTargetNotFound { .. } => "ReadTargetNotFound", OmniError::AmbiguousReadTarget { .. } => "AmbiguousReadTarget", + OmniError::BranchStartRefused { .. } => "BranchStartRefused", OmniError::ChangeFeedGap { .. } => "ChangeFeedGap", OmniError::CommitHasNoParent { .. } => "CommitHasNoParent", OmniError::ChangeSchemaBoundary { .. } => "ChangeSchemaBoundary", diff --git a/crates/omnigraph-server/tests/integration/auth_policy.rs b/crates/omnigraph-server/tests/integration/auth_policy.rs index 416bd5fe3..c1dc01b4c 100644 --- a/crates/omnigraph-server/tests/integration/auth_policy.rs +++ b/crates/omnigraph-server/tests/integration/auth_policy.rs @@ -2450,6 +2450,7 @@ async fn branch_statements_take_their_routes_policy_decision() { assert_forbidden(status, body, "branch create by a reader"); let create = BranchCreateRequest { from: None, + from_commit: None, name: "feature".to_string(), }; let (status, body) = json_response( @@ -2473,6 +2474,43 @@ async fn branch_statements_take_their_routes_policy_decision() { assert_eq!(status, StatusCode::OK, "{body}"); assert_eq!(body["actor_id"], "act-ragnor"); assert_eq!(body["outcome"]["kind"], "created"); + let head = body["outcome"]["from_commit_id"] + .as_str() + .expect("the receipt names the commit the branch starts at") + .to_string(); + + // A commit start the actor may not branch from answers the 404 of a + // `from` that names nothing, on both doors: the id discloses nothing. + let at_head = format!("branch create at_head from commit \"{head}\""); + let (status, body) = + json_response(&app, statement_request("/mutate", "reader-token", &at_head)).await; + assert_eq!(status, StatusCode::NOT_FOUND, "{body}"); + let create_at_head = BranchCreateRequest { + from: None, + from_commit: Some(head.clone()), + name: "at_head".to_string(), + }; + let (status, body) = json_response( + &app, + Request::builder() + .header(HTTP_API_CONTRACT_HEADER, HTTP_API_CONTRACT) + .uri(g("/branches")) + .method(Method::POST) + .header("authorization", "Bearer reader-token") + .header("content-type", "application/json") + .body(Body::from(serde_json::to_vec(&create_at_head).unwrap())) + .unwrap(), + ) + .await; + assert_eq!(status, StatusCode::NOT_FOUND, "{body}"); + let (status, body) = + json_response(&app, statement_request("/mutate", "admin-token", &at_head)).await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!( + body["outcome"]["from"], "main", + "the branch that wrote the commit" + ); + assert_eq!(body["outcome"]["from_commit_id"], head); let (status, body) = json_response( &app, @@ -2494,7 +2532,10 @@ async fn branch_statements_take_their_routes_policy_decision() { ) .await; assert_eq!(status, StatusCode::OK, "{body}"); - assert_eq!(body["rows"], json!([{"name": "feature"}, {"name": "main"}])); + assert_eq!( + body["rows"], + json!([{"name": "at_head"}, {"name": "feature"}, {"name": "main"}]) + ); let (status, body) = json_response( &app, statement_request("/query", "admin-token", "branch list"), @@ -2758,6 +2799,7 @@ async fn authenticated_branch_merge_stamps_merge_actor_on_head_commit() { let create = BranchCreateRequest { from: Some("main".to_string()), + from_commit: None, name: "feature".to_string(), }; let (create_status, _) = json_response( diff --git a/crates/omnigraph-server/tests/integration/data_routes.rs b/crates/omnigraph-server/tests/integration/data_routes.rs index 4d0c27b5f..b37aeae83 100644 --- a/crates/omnigraph-server/tests/integration/data_routes.rs +++ b/crates/omnigraph-server/tests/integration/data_routes.rs @@ -2572,6 +2572,7 @@ async fn branch_merge_receipts_keep_own_publication_when_a_later_writer_finishes #[tokio::test(flavor = "multi_thread")] async fn branch_statements_dispatch_to_their_route_bodies() { let (_temp, app) = app_for_loaded_graph().await; + let loaded = main_head_commit_id(&app).await; let (status, body) = json_response( &app, @@ -2591,7 +2592,12 @@ async fn branch_statements_dispatch_to_their_route_bodies() { "affected_edges": 0, "actor_id": null, "commit": null, - "outcome": {"kind": "created", "from": "main", "name": "feature"} + "outcome": { + "kind": "created", + "from": "main", + "name": "feature", + "from_commit_id": loaded, + } }) ); @@ -2626,7 +2632,13 @@ async fn branch_statements_dispatch_to_their_route_bodies() { assert_eq!(status, StatusCode::OK, "{body}"); assert_eq!( body["outcome"], - json!({"kind": "created", "from": "feature", "name": "review/x"}) + json!({ + "kind": "created", + "from": "feature", + "name": "review/x", + "from_commit_id": loaded, + }), + "a branch with no commit of its own starts at the head it inherited" ); let (_, body) = json_response(&app, json_post("/query", &json!({"query": "branch list"}))).await; @@ -3012,6 +3024,7 @@ async fn branch_statement_refusals_name_the_door_and_the_envelope() { let create = BranchCreateRequest { from: Some("main".to_string()), + from_commit: None, name: "b0".to_string(), }; let post_create = || { @@ -3395,8 +3408,26 @@ async fn commit_routes_create_list_refuse_publish_and_drop_unattached_branches() body["error"], format!("commit '{id}' is unattached and cannot publish") ); + // An unattached snapshot never becomes branch history, so no branch + // starts at it. + let (status, body) = json_response( + &app, + json_post( + "/branches", + &json!({"from_commit": id, "name": "from-snapshot"}), + ), + ) + .await; + assert_eq!(status, StatusCode::CONFLICT, "{body}"); + assert_eq!( + body["branch_start_refusal"], + json!({"reason": "unattached_snapshot", "graph_commit_id": id}), + "{body}" + ); } assert_eq!(main_head_commit_id(&app).await, head_before); + let (_, branches) = get_json(&app, g("/branches")).await; + assert_eq!(branches["branches"], json!(["main"])); let response = app .clone() @@ -5392,6 +5423,7 @@ async fn remote_branch_list_create_merge_flow_works() { let create = BranchCreateRequest { from: Some("main".to_string()), + from_commit: None, name: "feature".to_string(), }; let (create_status, create_body) = json_response( @@ -5408,6 +5440,11 @@ async fn remote_branch_list_create_merge_flow_works() { assert_eq!(create_status, StatusCode::OK); assert_eq!(create_body["from"], "main"); assert_eq!(create_body["name"], "feature"); + let loaded = main_head_commit_id(&app).await; + assert_eq!( + create_body["from_commit_id"], loaded, + "the receipt names the head the branch starts at" + ); // Target namespace inventory reads the ancestor ref before it can clone // or reclaim anything. A failed read must leave server admission open. @@ -5539,6 +5576,77 @@ async fn remote_branch_list_create_merge_flow_works() { assert_eq!(read_status, StatusCode::OK); assert_eq!(read_body["row_count"], 1); assert_eq!(read_body["rows"][0]["p.name"], "Zoe"); + + // A branch started at the commit main held before the merge reads that + // commit's rows, not main's head. + let (status, body) = json_response( + &app, + json_post( + "/branches", + &json!({"from_commit": loaded, "name": "before-zoe"}), + ), + ) + .await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body["from"], "main", "the branch that wrote the commit"); + assert_eq!(body["from_commit_id"], loaded); + let (status, body) = json_response( + &app, + json_post( + "/query", + &json!({ + "query": fs::read_to_string(fixture("test.gq")).unwrap(), + "name": "get_person", + "params": {"name": "Zoe"}, + "branch": "before-zoe", + }), + ), + ) + .await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body["row_count"], 0); + + // A commit id no commit has is not found, and a start is a branch or a + // commit, never both. + let unknown = "01ARZ3NDEKTSV4RRFFQ69G5FAV"; + let (status, body) = json_response( + &app, + json_post( + "/branches", + &json!({"from_commit": unknown, "name": "nowhere"}), + ), + ) + .await; + assert_eq!(status, StatusCode::NOT_FOUND, "{body}"); + assert_eq!(body["error"], format!("commit '{unknown}' not found")); + let (status, body) = json_response( + &app, + json_post( + "/branches", + &json!({"from": "main", "from_commit": loaded, "name": "both"}), + ), + ) + .await; + assert_eq!(status, StatusCode::BAD_REQUEST, "{body}"); + assert_eq!( + body["error"], + "request may specify from or from_commit, not both" + ); + let (_, body) = json_response( + &app, + Request::builder() + .header(HTTP_API_CONTRACT_HEADER, HTTP_API_CONTRACT) + .uri(g("/branches")) + .method(Method::GET) + .body(Body::empty()) + .unwrap(), + ) + .await; + assert_eq!( + body["branches"], + json!(["before-zoe", "feature", "main"]), + "refused starts create nothing" + ); } #[tokio::test(flavor = "multi_thread")] @@ -5547,6 +5655,7 @@ async fn remote_branch_delete_flow_works() { let create = BranchCreateRequest { from: Some("main".to_string()), + from_commit: None, name: "feature".to_string(), }; let (create_status, _) = json_response( @@ -5595,6 +5704,7 @@ async fn branch_merge_delete_branch_retires_parent_with_live_child() { let create = BranchCreateRequest { from: Some("main".to_string()), + from_commit: None, name: "feature".to_string(), }; let (create_status, _) = json_response( @@ -5632,6 +5742,7 @@ async fn branch_merge_delete_branch_retires_parent_with_live_child() { let create_child = BranchCreateRequest { from: Some("feature".to_string()), + from_commit: None, name: "feature-child".to_string(), }; let (create_child_status, _) = json_response( @@ -5693,6 +5804,7 @@ async fn branch_merge_delete_branch_refusal_is_non_fatal() { let create = BranchCreateRequest { from: Some("main".to_string()), + from_commit: None, name: "feature".to_string(), }; let (create_status, _) = json_response( @@ -6981,6 +7093,7 @@ async fn mutate_graph_commit_precondition_issue_365() { let inherited_head = head_commit_id(&app).await; let create = BranchCreateRequest { from: Some("main".to_string()), + from_commit: None, name: "fresh-cas".to_string(), }; let (status, body) = json_response( diff --git a/crates/omnigraph-server/tests/integration/openapi.rs b/crates/omnigraph-server/tests/integration/openapi.rs index 77b994e04..99fa0c3a8 100644 --- a/crates/omnigraph-server/tests/integration/openapi.rs +++ b/crates/omnigraph-server/tests/integration/openapi.rs @@ -1059,6 +1059,8 @@ const EXPECTED_SCHEMAS: &[&str] = &[ "ChangeBaselineRecord", "ChangeDiffRefusalOutput", "ChangeDiffRefusalReason", + "BranchStartRefusalOutput", + "BranchStartRefusalReason", "BlobRangeOutput", "ExternalBlobSourceOutput", "ExportRequest", @@ -1259,7 +1261,7 @@ fn branch_outcome_output_schema_is_tagged_by_kind() { .collect(); fields.sort_unstable(); let expected: &[&str] = match kind { - "created" => &["from", "name"], + "created" => &["from", "from_commit_id", "name"], "deleted" => &["name"], "merged" => &["merge", "source", "target"], "published" => &[], @@ -1399,6 +1401,45 @@ fn error_output_schema_has_expected_fields() { .iter() .any(|field| field == "full_text_index_rebuild_required") ); + // A refused branch start is an optional detail; `POST /branches` declares + // its 409 and its 410 (a commit cleanup no longer retains) with it. + let branch_start = &props["branch_start_refusal"]; + assert!( + branch_start["oneOf"] + .as_array() + .unwrap() + .iter() + .any(|schema| { schema["$ref"] == "#/components/schemas/BranchStartRefusalOutput" }) + ); + assert_eq!( + doc["components"]["schemas"]["BranchStartRefusalReason"]["enum"], + serde_json::json!([ + "not_retained", + "schema_changed", + "earlier_storage_format", + "branch_deleted", + "unpublished_commit", + "unattached_snapshot", + "unknown" + ]) + ); + let create = &doc["paths"]["/graphs/{graph_id}/branches"]["post"]["responses"]; + for status in ["404", "409", "410"] { + assert_eq!( + create[status]["content"]["application/json"]["schema"]["$ref"], + "#/components/schemas/ErrorOutput", + "POST /branches {status}" + ); + } + for status in ["409", "410"] { + assert!( + create[status]["description"] + .as_str() + .unwrap() + .contains("branch_start_refusal"), + "POST /branches {status} must name the refusal detail" + ); + } let details = &doc["components"]["schemas"]["FullTextIndexRebuildRequiredOutput"]; let required: HashSet<&str> = details["required"] .as_array() diff --git a/crates/omnigraph-server/tests/support/mod.rs b/crates/omnigraph-server/tests/support/mod.rs index 11db1aca3..12e5c9df2 100644 --- a/crates/omnigraph-server/tests/support/mod.rs +++ b/crates/omnigraph-server/tests/support/mod.rs @@ -645,6 +645,7 @@ pub mod matrix { pub async fn create_branch(&self, from: &str, name: &str) { let body = serde_json::to_vec(&BranchCreateRequest { from: Some(from.to_string()), + from_commit: None, name: name.to_string(), }) .unwrap(); @@ -950,6 +951,7 @@ pub mod matrix { barrier.wait().await; let body = serde_json::to_vec(&BranchCreateRequest { from: Some(from), + from_commit: None, name, }) .unwrap(); diff --git a/crates/omnigraph/src/db/graph_coordinator.rs b/crates/omnigraph/src/db/graph_coordinator.rs index b5b8de0fc..c611ad79d 100644 --- a/crates/omnigraph/src/db/graph_coordinator.rs +++ b/crates/omnigraph/src/db/graph_coordinator.rs @@ -494,7 +494,9 @@ impl GraphCoordinator { self.manifest.native_branch() } - pub(crate) async fn branch_create(&mut self, name: &str) -> Result<()> { + /// Create `name` from the `__manifest` version this coordinator holds and + /// return the commit the new branch starts at: that version's head. + pub(crate) async fn branch_create(&mut self, name: &str) -> Result { let branch = normalize_branch_name(name)? .ok_or_else(|| OmniError::manifest("cannot create branch 'main'".to_string()))?; @@ -502,7 +504,27 @@ impl GraphCoordinator { // it in two physical phases (shallow clone, then BranchContents); the // manifest coordinator classifies/reclaims a clone-only zombie before // a bounded retry. No graph-lineage branch is created or rolled back. - self.manifest.create_branch(&branch).await + self.manifest.create_branch(&branch).await?; + Ok(self.manifest.head().graph_commit_id.clone()) + } + + /// Create `name` at the published commit `record` names, which this + /// coordinator's branch incarnation wrote: the new branch's first + /// `__manifest` version is a clone of the version that published it. + pub(crate) async fn branch_create_at( + &mut self, + name: &str, + record: &HistoryRecord, + ) -> Result<()> { + let branch = normalize_branch_name(name)? + .ok_or_else(|| OmniError::manifest("cannot create branch 'main'".to_string()))?; + self.manifest + .create_branch_at( + &branch, + record.commit.graph_manifest_version, + &record.commit.graph_commit_id, + ) + .await } /// Delete the branch represented by an operation-local post-gate capture. @@ -672,6 +694,24 @@ impl GraphCoordinator { } } + /// The record of the published commit `id`: held by this coordinator, + /// else in `__history`, else in a live branch. `None` for an id no + /// published commit has. + pub(crate) async fn published_record(&self, id: &str) -> Result> { + if let Some(record) = self.manifest.held_record(id) { + return Ok(Some(record)); + } + let session = self.manifest.control_session(); + match self + .history() + .read_record(self.root_uri(), &session, id) + .await? + { + Some(record) => Ok(Some(record)), + None => self.record_in_live_branches(id).await, + } + } + /// The unpublished commit `id` names, read after every published /// resolution missed: one tag read on a miss of a published id. async fn unpublished_commit(&self, id: &str) -> Result> { diff --git a/crates/omnigraph/src/db/mod.rs b/crates/omnigraph/src/db/mod.rs index 193db63ef..c000378c6 100644 --- a/crates/omnigraph/src/db/mod.rs +++ b/crates/omnigraph/src/db/mod.rs @@ -16,17 +16,18 @@ pub use commit_graph::{GraphCommit, GraphCommitKind}; pub use graph_coordinator::{ReadTarget, ResolvedTarget, SnapshotId}; pub use manifest::{DatasetEntry, DatasetUpdate}; pub use omnigraph::{ - COMMIT_LIST_DEFAULT_PAGE_COMMITS, COMMIT_LIST_MAX_PAGE_COMMITS, CleanupPolicyOptions, - CollectorCost, CollectorPathSnapshot, CollectorReport, CollectorRowSummary, CommitListPage, - DatasetCleanupStats, DatasetOptimizeStats, DatasetRepairStats, EXPORT_CHUNK_MAX_BYTES, - ExportCut, FullTextIndexRebuildResult, GraphCreateReconciliation, InitOptions, MergeOutcome, - MergeResult, Omnigraph, OpenMode, PendingIndex, PreparedGraphCreate, PreparedSchemaApply, - PreparedSchemaSettlement, ReadCheck, RebuiltFullTextIndex, RepairAction, RepairClassification, - RepairOptions, RepairStats, RetainedManifestVersions, SYSTEM_COLUMNS_PREFLIGHT, - SchemaApplyReconciliation, SchemaApplyResult, SchemaApplySettlement, SchemaContractDigest, - SchemaNonPublicationProof, SkipReason, StagingVerdict, SystemColumnUpgradeFinding, - SystemColumnUpgradeOptions, SystemColumnUpgradeOutcome, SystemColumnUpgradeReport, - TableCollectionPlan, UnattachedTagRecord, UnpublishedManifest, unattached_release_order, + BranchCreated, COMMIT_LIST_DEFAULT_PAGE_COMMITS, COMMIT_LIST_MAX_PAGE_COMMITS, + CleanupPolicyOptions, CollectorCost, CollectorPathSnapshot, CollectorReport, + CollectorRowSummary, CommitListPage, DatasetCleanupStats, DatasetOptimizeStats, + DatasetRepairStats, EXPORT_CHUNK_MAX_BYTES, ExportCut, FullTextIndexRebuildResult, + GraphCreateReconciliation, InitOptions, MergeOutcome, MergeResult, Omnigraph, OpenMode, + PendingIndex, PreparedGraphCreate, PreparedSchemaApply, PreparedSchemaSettlement, ReadCheck, + RebuiltFullTextIndex, RepairAction, RepairClassification, RepairOptions, RepairStats, + RetainedManifestVersions, SYSTEM_COLUMNS_PREFLIGHT, SchemaApplyReconciliation, + SchemaApplyResult, SchemaApplySettlement, SchemaContractDigest, SchemaNonPublicationProof, + SkipReason, StagingVerdict, SystemColumnUpgradeFinding, SystemColumnUpgradeOptions, + SystemColumnUpgradeOutcome, SystemColumnUpgradeReport, TableCollectionPlan, + UnattachedTagRecord, UnpublishedManifest, unattached_release_order, }; pub(crate) use omnigraph::{ LogicalBlobValue, RangedExternalBlobs, logical_row_image, row_blob_values, diff --git a/crates/omnigraph/src/db/omnigraph.rs b/crates/omnigraph/src/db/omnigraph.rs index 92fba3d30..641bc1e04 100644 --- a/crates/omnigraph/src/db/omnigraph.rs +++ b/crates/omnigraph/src/db/omnigraph.rs @@ -20,7 +20,7 @@ use omnigraph_compiler::{ use crate::db::graph_coordinator::{ CapturedLineage, GraphCoordinator, PublishedSnapshot, commit_not_found, }; -use crate::error::{ManifestErrorKind, OmniError, Result, dataset_subject}; +use crate::error::{BranchStartRefusal, ManifestErrorKind, OmniError, Result, dataset_subject}; use crate::runtime_cache::RuntimeCache; use crate::seams::{decide_seam, fail}; use crate::storage::{ @@ -124,6 +124,41 @@ impl AddressedCommit { } } +/// What a branch create started from: the branch it forked and the commit +/// the new branch's head is. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct BranchCreated { + /// The branch the new branch was forked off: the branch named, or for a + /// start at a commit, the branch that wrote the commit. + pub from: String, + /// The graph commit the new branch starts at: the branch's head when the + /// branch was created, or the named commit. + pub from_commit_id: String, +} + +/// Where `branch_create_from_impl` starts the new branch: a branch's head +/// when it is captured, or a published commit of the branch that wrote it. +enum CreateStart { + Head(String), + Commit(GraphCommit), +} + +/// Table versions a commit start opens at once, and objects it probes at +/// once, to prove the commit is retained. +const BRANCH_START_RETENTION_PROBES: usize = 8; + +/// A start at `commit_id` refused because cleanup reclaimed the version of +/// `entry` the commit pins. +fn branch_start_not_retained(commit_id: &str, entry: &crate::db::DatasetEntry) -> OmniError { + OmniError::branch_start_refused( + commit_id, + BranchStartRefusal::NotRetained { + table_key: entry.type_key.clone(), + published_dataset_version: entry.published_dataset_version, + }, + ) +} + /// A merge's disposition and the graph commit published by that invocation. #[derive(Debug, Clone, PartialEq, Eq)] pub struct MergeResult { @@ -3703,51 +3738,97 @@ impl Omnigraph { Ok(()) } - pub async fn branch_create_from(&self, from: impl Into, name: &str) -> Result<()> { + pub async fn branch_create_from( + &self, + from: impl Into, + name: &str, + ) -> Result { self.branch_create_from_as(from, name, None).await } - /// Create a branch from a specific source branch with an explicit - /// actor for engine-layer policy enforcement (MR-722 fan-out). + /// Create `name` from `from` with an explicit actor for engine-layer + /// policy enforcement (MR-722 fan-out): a branch, whose head the new + /// branch starts at, or a published graph commit + /// ([`ReadTarget::Snapshot`]), which the new branch starts at with that + /// commit's data and history. Returns the branch forked and the commit + /// the new branch starts at. /// /// Scope is `BranchTransition { source, target }` — matches the /// HTTP-layer convention at `server_branch_create` /// (branch=Some(from), target_branch=Some(name)), so engine and - /// HTTP fire the same Cedar decision. Pinned-snapshot sources - /// (which aren't a branch ref) materialize as the sentinel - /// `` for the policy check; Cedar rules using - /// `branch_scope: any` still match, rules pinning a specific - /// source branch correctly do not. + /// HTTP fire the same Cedar decision. A commit start's source is the + /// branch a control on the commit addresses + /// ([`Self::commit_target_branch`]); a denial answers the not-found an + /// unknown id answers, so the id discloses neither existence nor + /// publication state. An unpublished commit and an unattached snapshot + /// are refused: neither is any branch's history. pub async fn branch_create_from_as( &self, from: impl Into, name: &str, actor: Option<&str>, - ) -> Result<()> { - let target = from.into(); - let source_branch = match &target { - ReadTarget::Branch(b) => b.clone(), - _ => "".to_string(), + ) -> Result { + let start = match from.into() { + ReadTarget::Branch(branch) => { + self.enforce( + omnigraph_policy::PolicyAction::BranchCreate, + &omnigraph_policy::ResourceScope::BranchTransition { + source: branch.clone(), + target: name.to_string(), + }, + actor, + )?; + CreateStart::Head(branch) + } + ReadTarget::Snapshot(id) => { + let addressed = self.addressed_commit(id.as_str()).await?; + let scope = omnigraph_policy::ResourceScope::BranchTransition { + source: addressed.target_branch().unwrap_or("main").to_string(), + target: name.to_string(), + }; + match self.enforce(omnigraph_policy::PolicyAction::BranchCreate, &scope, actor) { + Err(OmniError::Policy(_)) if actor.is_some() => { + return Err(commit_not_found(&id)); + } + decision => decision?, + } + match addressed { + AddressedCommit::Published(commit) => CreateStart::Commit(commit), + AddressedCommit::Proposal(_) => { + return Err(OmniError::branch_start_refused( + id.as_str(), + BranchStartRefusal::UnpublishedCommit, + ) + .before_effect()); + } + AddressedCommit::Unattached(_) => { + return Err(OmniError::branch_start_refused( + id.as_str(), + BranchStartRefusal::UnattachedSnapshot, + ) + .before_effect()); + } + } + } }; - self.enforce( - omnigraph_policy::PolicyAction::BranchCreate, - &omnigraph_policy::ResourceScope::BranchTransition { - source: source_branch, - target: name.to_string(), - }, - actor, - )?; - self.branch_create_from_impl(target, name).await + // Keep the capture and the start proof out of every caller's frame. + Box::pin(self.branch_create_from_impl(start, name)).await } - async fn branch_create_from_impl(&self, from: impl Into, name: &str) -> Result<()> { - let target = from.into(); - let ReadTarget::Branch(branch_name) = target else { - return Err(OmniError::manifest( - "branch creation from pinned snapshots is not supported yet".to_string(), - )); + async fn branch_create_from_impl( + &self, + start: CreateStart, + name: &str, + ) -> Result { + // The source a commit start clones is the branch incarnation that + // wrote the commit, where the version that published it lives. + let branch = match &start { + CreateStart::Head(branch) => normalize_branch_name(branch)?, + CreateStart::Commit(commit) => match commit.graph_branch.as_deref() { + Some(branch) => normalize_branch_name(branch)?, + None => None, + }, }; - let branch = normalize_branch_name(&branch_name)?; let target_branch = normalize_branch_name(name)? .ok_or_else(|| OmniError::manifest("cannot create branch 'main'".to_string()))?; let _export_exclusion = self.reserve_export_destructive_control()?; @@ -3771,10 +3852,21 @@ impl Omnigraph { self.ensure_schema_state_valid() .await .map_err(OmniError::before_effect)?; - let mut source_coord = self - .capture_branch_control_source(branch.as_deref()) - .await - .map_err(OmniError::before_effect)?; + let (mut source_coord, start_record) = match &start { + CreateStart::Head(_) => ( + self.capture_branch_control_source(branch.as_deref()) + .await + .map_err(OmniError::before_effect)?, + None, + ), + CreateStart::Commit(commit) => { + let (source, record) = + Box::pin(self.capture_commit_start_source(branch.as_deref(), commit)) + .await + .map_err(OmniError::before_effect)?; + (source, Some(record)) + } + }; let (_control_catalog, _table_guards) = self .join_control_catalog_to_capture( control_catalog, @@ -3789,9 +3881,158 @@ impl Omnigraph { // A locally owned source coordinator cannot be swapped by a concurrent // `branch_create_from`; the ref write is durable whichever handle // issued it. - source_coord.branch_create(&target_branch).await?; + let created = match (start, start_record) { + (CreateStart::Head(from), _) => BranchCreated { + from_commit_id: source_coord.branch_create(&target_branch).await?, + from, + }, + (CreateStart::Commit(commit), Some(record)) => { + self.prove_commit_start(&source_coord, &record) + .await + .map_err(OmniError::before_effect)?; + source_coord + .branch_create_at(&target_branch, &record) + .await?; + BranchCreated { + from: commit.graph_branch.unwrap_or_else(|| "main".to_string()), + from_commit_id: commit.graph_commit_id, + } + } + (CreateStart::Commit(_), None) => { + unreachable!("a commit start captures the commit's record with its source") + } + }; self.invalidate_read_caches().await; - Ok(()) + Ok(created) + } + + /// The source a branch starting at the published `commit` clones, and + /// the commit's record read from it, captured under the control gates. + /// The branch incarnation that wrote the commit must still be the live + /// one under its name. A deleted name and a name created again since + /// are the same fact, so both answer `BranchDeleted`, before any effect + /// and on every door. + async fn capture_commit_start_source( + &self, + branch: Option<&str>, + commit: &GraphCommit, + ) -> Result<(GraphCoordinator, super::manifest::HistoryRecord)> { + let commit_id = commit.graph_commit_id.as_str(); + let deleted = || { + OmniError::branch_start_refused( + commit_id, + BranchStartRefusal::BranchDeleted { + branch: commit + .graph_branch + .clone() + .unwrap_or_else(|| "main".to_string()), + }, + ) + }; + let source = match self.capture_branch_control_source(branch).await { + Err(OmniError::BranchNotFound { .. }) => return Err(deleted()), + captured => captured?, + }; + match source.published_record(commit_id).await? { + Some(record) if record.commit.native_branch.as_deref() == source.native_branch() => { + Ok((source, record)) + } + Some(_) => Err(deleted()), + // Addressed as published, its record is held by no live + // incarnation of its branch now: that branch was deleted. + None if commit.graph_branch.is_some() => Err(deleted()), + None => Err(commit_not_found(&SnapshotId::new(commit_id))), + } + } + + /// Prove under the control gates that a branch can start at `record`, + /// which the captured `source` wrote: the commit is at or below the + /// captured head; its schema is the accepted one; and every table + /// version its snapshot pins still opens with every object it + /// references, so the new branch's first manifest names a complete + /// snapshot that its base keeps from cleanup from then on. + async fn prove_commit_start( + &self, + source: &GraphCoordinator, + record: &super::manifest::HistoryRecord, + ) -> Result<()> { + let commit_id = record.commit.graph_commit_id.as_str(); + if record.commit.graph_manifest_version > source.version() { + return Err(OmniError::manifest_internal(format!( + "commit '{commit_id}' names `__manifest` version {} above the captured head \ + version {} of the branch that wrote it", + record.commit.graph_manifest_version, + source.version() + ))); + } + let snapshot = Snapshot::wrap(record.snapshot(self.root_uri())?); + // Every record carries the contract its commit was written under: + // the storage upgrade gives each converted commit one too. A commit + // from before the upgrade is refused where its `__manifest` version + // is cloned, which only the legacy history reads. + let Some(contract) = snapshot.schema_contract() else { + return Err(OmniError::manifest_internal(format!( + "the record of commit '{commit_id}' carries no schema contract" + ))); + }; + if Some(contract) != source.snapshot().schema_contract() { + return Err(OmniError::branch_start_refused( + commit_id, + BranchStartRefusal::SchemaChanged, + )); + } + // A pin the captured head also holds is a root of every collection, + // and the collector never deletes a root's files. Any other pin may + // be a version an interrupted cleanup already emptied: its sweep + // deletes a version's files before its manifest, so the manifest + // opening proves nothing about them, and each object it references + // is probed. These requests run under the exclusive schema permit, + // which every write in the process waits on, so the pins open and + // their objects are probed with bounded concurrency, stopping at the + // first missing object. + use futures::StreamExt; + let head = source.snapshot(); + let entries: std::collections::BTreeMap<_, _> = snapshot + .datasets() + .filter(|entry| { + !head + .datasets() + .any(|held| held.same_physical_version(entry)) + }) + .map(|entry| (entry.type_key.clone(), entry.clone())) + .collect(); + let mut opens = futures::stream::iter(entries.values().cloned()) + .map(|entry| async move { + let dataset = match self.table_store.open_at_entry(&entry).await { + Ok(dataset) => dataset, + Err(OmniError::HistoricalVersionReclaimed { .. }) => { + return Err(branch_start_not_retained(commit_id, &entry)); + } + Err(error) => return Err(error), + }; + let mut objects = collector::CollectorPathSnapshot::default(); + collector::capture_dataset_paths(&mut objects, &dataset, &entry.type_key).await?; + Ok(objects) + }) + .buffer_unordered(BRANCH_START_RETENTION_PROBES); + let mut objects = collector::CollectorPathSnapshot::default(); + while let Some(opened) = opens.next().await { + objects.absorb(opened?); + } + match objects + .first_missing_location(BRANCH_START_RETENTION_PROBES) + .await? + { + None => Ok(()), + Some(type_key) => Err(branch_start_not_retained( + commit_id, + entries.get(&type_key).ok_or_else(|| { + OmniError::manifest_internal(format!( + "a branch start probe reported '{type_key}', which no probed pin names" + )) + })?, + )), + } } pub async fn branch_list(&self) -> Result> { diff --git a/crates/omnigraph/src/db/omnigraph/collector.rs b/crates/omnigraph/src/db/omnigraph/collector.rs index 2caac87e0..93d2643bd 100644 --- a/crates/omnigraph/src/db/omnigraph/collector.rs +++ b/crates/omnigraph/src/db/omnigraph/collector.rs @@ -433,6 +433,42 @@ impl CollectorPathSnapshot { Ok(missing.into_iter().collect()) } + /// The location of a captured object that no longer exists, or `None` + /// when every one does. Probes at most `concurrency` objects at once and + /// stops at the first missing one, so a complete inventory costs one + /// request per object and an incomplete one usually far fewer. + pub(crate) async fn first_missing_location( + &self, + concurrency: usize, + ) -> Result> { + use futures::StreamExt; + if let Some((location, _)) = self.missing_at_capture.first() { + return Ok(Some(location.clone())); + } + let mut probes = futures::stream::iter(self.objects.iter().cloned()) + .map(|object| async move { + let present = object + .store + .exists(&object.path) + .await + .map_err(OmniError::storage)?; + Ok::<_, OmniError>((!present).then_some(object.location)) + }) + .buffer_unordered(concurrency.max(1)); + while let Some(probed) = probes.next().await { + if let Some(location) = probed? { + return Ok(Some(location)); + } + } + Ok(None) + } + + /// Move `other`'s captured objects into this inventory. + pub(crate) fn absorb(&mut self, other: CollectorPathSnapshot) { + self.objects.extend(other.objects); + self.missing_at_capture.extend(other.missing_at_capture); + } + fn insert( &mut self, location: &str, @@ -523,7 +559,7 @@ async fn capture_index_paths( Ok(()) } -async fn capture_dataset_paths( +pub(crate) async fn capture_dataset_paths( snapshot: &mut CollectorPathSnapshot, dataset: &lance::Dataset, location: &str, diff --git a/crates/omnigraph/src/db/upgrade/tests.rs b/crates/omnigraph/src/db/upgrade/tests.rs index e5778e4d6..f04b95e3e 100644 --- a/crates/omnigraph/src/db/upgrade/tests.rs +++ b/crates/omnigraph/src/db/upgrade/tests.rs @@ -3074,6 +3074,64 @@ async fn commit_list_and_change_feed_cross_the_upgrade() { } } +/// A commit written before the storage upgrade starts no branch: its +/// `__manifest` version is in the earlier format, which only the legacy +/// history reads, so the start is refused before any effect. A commit +/// written after the upgrade starts one. +#[tokio::test] +async fn branch_start_refuses_a_commit_written_before_the_upgrade() { + use crate::db::{ReadTarget, SnapshotId}; + use crate::error::{BranchStartRefusal, CompletionEvidence}; + #[cfg(feature = "failpoints")] + let _scenario = crate::seams::FailScenario::setup(); + let source = engine_source().await; + assert_eq!( + execute(&source.root).await.outcome, + UpgradeOutcome::Completed + ); + let graph = Arc::new(Omnigraph::open(&source.root).await.unwrap()); + insert(&graph, "main", "d").await; + let branches = graph.branch_list().await.unwrap(); + let main_version = version_of(&source.root, None).await; + for commit in [id(1), id(2), id(3), id(4)] { + let refused = graph + .branch_create_from(ReadTarget::snapshot(SnapshotId::new(commit.clone())), "old") + .await + .expect_err("a pre-upgrade commit starts no branch"); + assert_eq!( + refused.completion_evidence(), + Some(CompletionEvidence::BeforeEffect), + "{refused}" + ); + let refused = refused.into_completion_evidence().0; + assert!( + matches!( + &refused, + OmniError::BranchStartRefused { + reason: BranchStartRefusal::EarlierStorageFormat, + graph_commit_id, + } if *graph_commit_id == commit + ), + "{commit}: {refused:?}" + ); + } + assert_eq!(graph.branch_list().await.unwrap(), branches); + assert_eq!(version_of(&source.root, None).await, main_version); + + let newest = graph.list_commits(None).await.unwrap()[0] + .graph_commit_id + .clone(); + let created = graph + .branch_create_from(ReadTarget::snapshot(SnapshotId::new(newest.clone())), "new") + .await + .unwrap(); + assert_eq!( + (created.from.as_str(), created.from_commit_id.as_str()), + ("main", newest.as_str()) + ); + assert_eq!(names(&graph, "new").await, ["a", "c", "d"]); +} + /// Publish a lineage-only commit on `branch` of `root` under the release /// budget `release_bytes` and return the commit the head then holds. async fn publish( diff --git a/crates/omnigraph/src/error.rs b/crates/omnigraph/src/error.rs index 279a765c9..80e8fca32 100644 --- a/crates/omnigraph/src/error.rs +++ b/crates/omnigraph/src/error.rs @@ -2,6 +2,7 @@ pub(crate) use omnigraph_core::error::*; pub use omnigraph_core::error::{ - CompletionEvidence, ManifestConflictDetails, ManifestError, ManifestErrorKind, MergeConflict, - MergeConflictKind, OmniError, Result, StorageFailure, StorageFailureKind, + BranchStartRefusal, CompletionEvidence, ManifestConflictDetails, ManifestError, + ManifestErrorKind, MergeConflict, MergeConflictKind, OmniError, Result, StorageFailure, + StorageFailureKind, }; diff --git a/crates/omnigraph/tests/failpoints.rs b/crates/omnigraph/tests/failpoints.rs index 871419b81..1128a0303 100644 --- a/crates/omnigraph/tests/failpoints.rs +++ b/crates/omnigraph/tests/failpoints.rs @@ -431,6 +431,150 @@ async fn native_branch_controls_reclassify_lost_acknowledgements() { before_commits, "native branch controls must not manufacture graph lineage" ); + + // A start at a past commit clones the version that published it through + // the same native create: a failure between the inventory and the create + // leaves nothing, and a lost acknowledgement classifies as success. + let loaded = db.list_commits(Some("main")).await.unwrap()[0] + .graph_commit_id + .clone(); + mutate_main( + &db, + MUTATION_QUERIES, + "insert_person", + &mixed_params(&[("$name", "Eve")], &[("$age", 22)]), + ) + .await + .unwrap(); + let before_version = version_main(&db).await.unwrap(); + let before_commits = db.list_commits(Some("main")).await.unwrap(); + let at_load = ReadTarget::snapshot(SnapshotId::new(loaded.clone())); + { + let _fp = catalog::BRANCH_CREATE_POST_INVENTORY_PRE_NATIVE.fire_always(); + db.branch_create_from(at_load.clone(), "at-load") + .await + .expect_err("a failure before the native create creates nothing"); + } + assert_eq!(db.branch_list().await.unwrap(), vec!["main".to_string()]); + { + let _fp = catalog::BRANCH_CREATE_POST_NATIVE.fire_always(); + let created = db + .branch_create_from(at_load, "at-load") + .await + .expect("matching BranchContents must classify a lost create acknowledgement"); + assert_eq!(created.from_commit_id, loaded); + } + assert_eq!( + helpers::count_rows_branch(&db, "at-load", "node:Person").await, + 4, + "the branch holds the commit's rows, not main's head" + ); + assert_eq!(version_main(&db).await.unwrap(), before_version); + assert_eq!(db.list_commits(Some("main")).await.unwrap(), before_commits); +} + +/// Cleanup deletes a swept table version's files before its manifest, so a +/// run stopped between the two leaves a manifest that still opens over +/// objects that are gone. A branch cannot start at a commit pinning such a +/// version: a start probes every object of each pin its source head does not +/// hold, and refuses the commit as not retained before any effect. The +/// cleanup retry removes that manifest, and the start stays refused, now +/// because the version no longer opens; a start at the retained head works +/// throughout. +#[tokio::test] +#[serial] +async fn branch_start_refuses_a_commit_an_interrupted_cleanup_emptied() { + let _scenario = FailScenario::setup(); + let dir = tempfile::tempdir().unwrap(); + let db = helpers::init_and_load(&dir).await; + for name in ["Eve", "Frank"] { + mutate_main( + &db, + MUTATION_QUERIES, + "insert_person", + &mixed_params(&[("$name", name)], &[("$age", 40)]), + ) + .await + .unwrap(); + } + let commits = db.list_commits(Some("main")).await.unwrap(); + let eve = commits[1].graph_commit_id.clone(); + let frank = commits[0].graph_commit_id.clone(); + let at_eve = ReadTarget::snapshot(SnapshotId::new(eve.clone())); + let eve_person = db + .snapshot_of(at_eve.clone()) + .await + .unwrap() + .dataset("node:Person") + .unwrap() + .version_metadata + .staged_version() + .expect("Eve's insert pins a detached Person version"); + let person_uri = node_table_uri(&db, "Person").await; + + { + let _fp = catalog::CLEANUP_SWEEP_PRE_MANIFEST_DELETE.fire_always(); + let _ = db.cleanup(keep_one()).await; + } + assert!( + detached_versions(&person_uri).await.contains(&eve_person), + "precondition: the interrupted sweep left Eve's Person manifest" + ); + let main_version = version_main(&db).await.unwrap(); + let assert_not_retained = |refused: OmniError| { + assert_eq!( + refused.completion_evidence(), + Some(CompletionEvidence::BeforeEffect), + "{refused}" + ); + assert!( + matches!( + refused.into_completion_evidence().0, + OmniError::BranchStartRefused { + reason: omnigraph::error::BranchStartRefusal::NotRetained { table_key, .. }, + .. + } if table_key == "node:Person" + ), + "the refusal names the emptied table" + ); + }; + assert_not_retained( + db.branch_create_from(at_eve.clone(), "at-eve") + .await + .expect_err("a manifest whose objects are gone proves nothing"), + ); + assert_eq!(db.branch_list().await.unwrap(), vec!["main".to_string()]); + assert_eq!(version_main(&db).await.unwrap(), main_version); + + let stats = db.cleanup(keep_one()).await.unwrap(); + let person = stats + .iter() + .find(|row| row.type_key == "node:Person") + .unwrap(); + assert!(person.error.is_none(), "the retry converges: {person:?}"); + assert!( + !detached_versions(&person_uri).await.contains(&eve_person), + "the retry removes the half-swept manifest" + ); + assert_not_retained( + db.branch_create_from(at_eve, "at-eve") + .await + .expect_err("a reclaimed version starts no branch"), + ); + assert_eq!(db.branch_list().await.unwrap(), vec!["main".to_string()]); + + let created = db + .branch_create_from( + ReadTarget::snapshot(SnapshotId::new(frank.clone())), + "at-frank", + ) + .await + .expect("the retained head starts a branch"); + assert_eq!(created.from_commit_id, frank); + assert_eq!( + helpers::count_rows_branch(&db, "at-frank", "node:Person").await, + helpers::count_rows_branch(&db, "main", "node:Person").await, + ); } /// Branch delete leaves the deleted branch's detached staging for `cleanup`: diff --git a/crates/omnigraph/tests/forbidden_apis.rs b/crates/omnigraph/tests/forbidden_apis.rs index f405c2fb4..7fd64a42e 100644 --- a/crates/omnigraph/tests/forbidden_apis.rs +++ b/crates/omnigraph/tests/forbidden_apis.rs @@ -466,6 +466,11 @@ const LOW_LEVEL_READ_ONLY_SURFACES: &[(&str, &str, &str)] = &[ "GraphCoordinator", "resolve_commit", ), + ( + "db/graph_coordinator.rs", + "GraphCoordinator", + "published_record", + ), ( "db/graph_coordinator.rs", "GraphCoordinator", @@ -638,6 +643,12 @@ const LOW_LEVEL_WRITE_SURFACES: &[(&str, &str, &str, WriteProtocol)] = &[ "branch_create", WriteProtocol::NativeRefControl, ), + ( + "db/graph_coordinator.rs", + "GraphCoordinator", + "branch_create_at", + WriteProtocol::NativeRefControl, + ), ( "db/graph_coordinator.rs", "GraphCoordinator", @@ -674,6 +685,12 @@ const LOW_LEVEL_WRITE_SURFACES: &[(&str, &str, &str, WriteProtocol)] = &[ "create_branch", WriteProtocol::NativeRefControl, ), + ( + "omnigraph-catalog/lib.rs", + "ManifestCoordinator", + "create_branch_at", + WriteProtocol::NativeRefControl, + ), ( "omnigraph-catalog/lib.rs", "ManifestCoordinator", @@ -2863,6 +2880,7 @@ fn graph_manifest_writer_methods_are_not_public_escape_hatches() { let coordinator_methods = [ "init_commit_with_session", "branch_create", + "branch_create_at", "branch_delete_captured", "commit_updates_with_actor", "commit_updates_with_actor_with_expected", @@ -2906,6 +2924,7 @@ fn graph_manifest_writer_methods_are_not_public_escape_hatches() { "init_commit", "commit_changes_with_lineage_and_precondition", "create_branch", + "create_branch_at", "delete_branch", "delete_branch_with_expected", ] { @@ -2941,6 +2960,7 @@ fn graph_manifest_writer_methods_are_not_public_escape_hatches() { "commit_changes_with_lineage", "commit_changes_with_lineage_and_precondition", "create_branch", + "create_branch_at", "delete_branch", "delete_branch_with_expected", ]; @@ -3024,6 +3044,7 @@ const SPLIT_CRATE_REEXPORTS: &[&str] = &[ "db/manifest.rs: omnigraph_catalog::INTERNAL_MANIFEST_SCHEMA_VERSION", "db/manifest.rs: omnigraph_catalog::MIN_SUPPORTED_INTERNAL_SCHEMA_VERSION", "db/manifest.rs: omnigraph_catalog::READ_REFRESH_POST_STATE_PRE_LINEAGE", + "error.rs: omnigraph_core::error::BranchStartRefusal", "error.rs: omnigraph_core::error::CompletionEvidence", "error.rs: omnigraph_core::error::ManifestConflictDetails", "error.rs: omnigraph_core::error::ManifestError", diff --git a/crates/omnigraph/tests/integration/branch_control_cost.rs b/crates/omnigraph/tests/integration/branch_control_cost.rs index 5aa1f1485..49a57f32b 100644 --- a/crates/omnigraph/tests/integration/branch_control_cost.rs +++ b/crates/omnigraph/tests/integration/branch_control_cost.rs @@ -109,3 +109,70 @@ fn branch_delete_manifest_reads_bounded_per_surviving_branch() { }) }); } + +/// Budget for a start at a past commit: it probes every object of each pin +/// its source head does not also hold, so its requests grow with the +/// fragments of those pins. On this fixture each earlier insert adds one +/// `Person` fragment, one data file, so the slope is one probe request per +/// added fragment; the manifest, transaction and index objects stay +/// constant. A probe that opens, lists or reads an object more than once per +/// fragment re-trips this gate. +const START_PER_FRAGMENT_BUDGET: u64 = 1; + +#[test] +fn branch_start_at_a_commit_probes_once_per_pinned_object() { + use crate::helpers::cost::measure_insert; + use omnigraph::db::{ReadTarget, SnapshotId}; + on_big_stack(|| { + cost_harness(async { + let dir = tempfile::tempdir().unwrap(); + let db = local_graph(&dir).await; + + let mut curve: Vec<(u64, IoCounts)> = Vec::new(); + let mut inserted = 0u64; + for n in [4u64, 12] { + while inserted < n { + measure_insert(&db, &format!("p_{inserted:02}")).await; + inserted += 1; + } + let start = db.list_commits(Some("main")).await.unwrap()[0] + .graph_commit_id + .clone(); + // One more insert, so the start's `Person` pin is not the + // head's and its objects are probed. + measure_insert(&db, &format!("p_{inserted:02}")).await; + inserted += 1; + let (res, io) = measure(db.branch_create_from( + ReadTarget::snapshot(SnapshotId::new(start)), + &format!("at_{n:02}"), + )) + .await; + res.unwrap(); + eprintln!( + "inserts={n}: START data_reads={} manifest_reads={} data_open_count={}", + io.data_reads, io.manifest_reads, io.data_open_count + ); + curve.push((n, io)); + } + + let added_fragments = curve[1].0 - curve[0].0; + let budget = added_fragments * START_PER_FRAGMENT_BUDGET + SLOPE_SLACK; + let read_delta = curve[1].1.data_reads.saturating_sub(curve[0].1.data_reads); + assert!( + read_delta >= added_fragments, + "the start must probe each added fragment's data file: {read_delta} reads \ + across {added_fragments} added fragments" + ); + assert!( + read_delta <= budget, + "a start at a past commit grew {read_delta} data reads across \ + {added_fragments} added fragments (budget {budget}): the retention probe \ + must send one request per pinned object" + ); + assert_eq!( + curve[0].1.data_open_count, curve[1].1.data_open_count, + "the probe opens each probed pin once, whatever its fragment count" + ); + }) + }); +} diff --git a/crates/omnigraph/tests/integration/branching.rs b/crates/omnigraph/tests/integration/branching.rs index 2653cc708..388854f56 100644 --- a/crates/omnigraph/tests/integration/branching.rs +++ b/crates/omnigraph/tests/integration/branching.rs @@ -1,5 +1,6 @@ use crate::helpers; +use std::collections::BTreeMap; use std::fmt::Write as _; use std::fs; @@ -9,8 +10,10 @@ use lance::Dataset; use lance_index::is_system_index; use omnigraph::db::commit_graph::CommitGraph; -use omnigraph::db::{MergeOutcome, Omnigraph, ReadTarget}; -use omnigraph::error::{CompletionEvidence, ManifestErrorKind, MergeConflictKind, OmniError}; +use omnigraph::db::{BranchCreated, MergeOutcome, Omnigraph, ReadTarget, Snapshot, SnapshotId}; +use omnigraph::error::{ + BranchStartRefusal, CompletionEvidence, ManifestErrorKind, MergeConflictKind, OmniError, +}; use omnigraph::instrumentation::{MergeWriteProbes, with_merge_write_probes}; use omnigraph::loader::LoadMode; use omnigraph::{ @@ -3232,6 +3235,354 @@ async fn branch_created_from_non_main_inherits_branch_state() { assert_eq!(main_qr.num_rows(), 0); } +/// Every person on `branch` with its age, read from the table the branch pins. +async fn people_on(db: &Omnigraph, branch: &str) -> BTreeMap> { + let mut people = BTreeMap::new(); + for batch in read_table_branch(db, branch, "node:Person").await { + let names = batch + .column_by_name("name") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + let ages = batch + .column_by_name("age") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + for row in 0..batch.num_rows() { + let age = (!ages.is_null(row)).then(|| ages.value(row)); + people.insert(names.value(row).to_string(), age); + } + } + people +} + +/// Each table a snapshot pins, with the exact version and metadata of the pin. +fn table_pins(snapshot: &Snapshot) -> BTreeMap { + snapshot + .datasets() + .map(|entry| { + ( + entry.type_key.clone(), + ( + entry.published_dataset_version, + format!("{:?}", entry.version_metadata), + ), + ) + }) + .collect() +} + +fn at_commit(id: &str) -> ReadTarget { + ReadTarget::snapshot(SnapshotId::new(id)) +} + +/// A branch started at a past commit of main holds exactly that commit's +/// tables and history: its first manifest is the version that published the +/// commit, main is untouched, and the branch writes on the commit and merges +/// back three ways from it. +#[tokio::test] +async fn branch_create_at_a_main_commit_starts_at_its_tables_and_history() { + let dir = tempfile::tempdir().unwrap(); + let db = init_and_load(&dir).await; + let eve = db + .mutate_with_receipt( + "main", + MUTATION_QUERIES, + "insert_person", + &mixed_params(&[("$name", "Eve")], &[("$age", 22)]), + ) + .await + .unwrap() + .commit + .expect("the insert publishes a commit"); + mutate_main( + &db, + MUTATION_QUERIES, + "set_age", + &mixed_params(&[("$name", "Alice")], &[("$age", 31)]), + ) + .await + .unwrap(); + mutate_main( + &db, + MUTATION_QUERIES, + "remove_person", + ¶ms(&[("$name", "Bob")]), + ) + .await + .unwrap(); + let main_version = version_main(&db).await.unwrap(); + let main_commits = db.list_commits(Some("main")).await.unwrap(); + let main_people = people_on(&db, "main").await; + + let created = db + .branch_create_from(at_commit(&eve.graph_commit_id), "fix") + .await + .unwrap(); + assert_eq!( + created, + BranchCreated { + from: "main".to_string(), + from_commit_id: eve.graph_commit_id.clone(), + }, + "the receipt names the branch that wrote the commit, and the commit" + ); + + let fix = snapshot_branch(&db, "fix").await.unwrap(); + assert_eq!( + fix.graph_manifest_version(), + eve.graph_manifest_version, + "the first manifest is a clone of the version that published the commit" + ); + assert_eq!( + table_pins(&fix), + table_pins( + &db.snapshot_of(at_commit(&eve.graph_commit_id)) + .await + .unwrap() + ), + "the new branch pins exactly the commit's table versions" + ); + let fix_commits = db.list_commits(Some("fix")).await.unwrap(); + assert_eq!(fix_commits[0].graph_commit_id, eve.graph_commit_id); + assert_eq!( + fix_commits.as_slice(), + &main_commits[2..], + "the branch's history is the commit's: the two later main commits are not in it" + ); + let people = people_on(&db, "fix").await; + assert_eq!(people.get("Alice"), Some(&Some(30))); + assert!(people.contains_key("Bob") && people.contains_key("Eve")); + + assert_eq!(version_main(&db).await.unwrap(), main_version); + assert_eq!(db.list_commits(Some("main")).await.unwrap(), main_commits); + assert_eq!(people_on(&db, "main").await, main_people); + + let zed = db + .mutate_with_receipt( + "fix", + MUTATION_QUERIES, + "insert_person", + &mixed_params(&[("$name", "Zed")], &[("$age", 50)]), + ) + .await + .unwrap() + .commit + .unwrap(); + assert_eq!( + zed.parent_commit_id.as_deref(), + Some(eve.graph_commit_id.as_str()), + "the branch's first write builds on the commit" + ); + let merged = db.branch_merge("fix", "main").await.unwrap(); + assert_eq!(merged.outcome, MergeOutcome::Merged); + let after = people_on(&db, "main").await; + assert_eq!( + after.get("Alice"), + Some(&Some(31)), + "main's later update stays" + ); + assert!(!after.contains_key("Bob"), "main's later delete stays"); + assert_eq!( + after.get("Zed"), + Some(&Some(50)), + "the branch's write lands" + ); +} + +/// A commit a named branch wrote starts a branch from that branch's own +/// `__manifest` history, and reopening the graph reads the same start. +#[tokio::test] +async fn branch_create_at_a_named_branch_commit_reads_that_branch_version() { + let dir = tempfile::tempdir().unwrap(); + let uri = dir.path().to_str().unwrap(); + let db = init_and_load(&dir).await; + db.branch_create("feature").await.unwrap(); + let frank = db + .mutate_with_receipt( + "feature", + MUTATION_QUERIES, + "insert_person", + &mixed_params(&[("$name", "Frank")], &[("$age", 41)]), + ) + .await + .unwrap() + .commit + .unwrap(); + mutate_branch( + &db, + "feature", + MUTATION_QUERIES, + "insert_person", + &mixed_params(&[("$name", "Grace")], &[("$age", 42)]), + ) + .await + .unwrap(); + + let created = db + .branch_create_from(at_commit(&frank.graph_commit_id), "retry") + .await + .unwrap(); + assert_eq!(created.from, "feature", "the branch that wrote the commit"); + assert_eq!(created.from_commit_id, frank.graph_commit_id); + + let reopened = helpers::session(Omnigraph::open(uri).await.unwrap()); + let people = people_on(&reopened, "retry").await; + assert!(people.contains_key("Frank")); + assert!( + !people.contains_key("Grace"), + "a later commit of the branch is not in it" + ); + assert_eq!( + reopened.list_commits(Some("retry")).await.unwrap()[0].graph_commit_id, + frank.graph_commit_id + ); + assert_eq!( + people_on(&reopened, "feature").await.len(), + people.len() + 1, + "the source branch keeps its head" + ); +} + +/// What is not branch history is refused before any effect, never +/// approximated: an unknown id, an unpublished commit, an unattached +/// snapshot, a commit of a deleted branch (one answer whether its name +/// names nothing or a branch created again since), and a commit written +/// under an earlier schema. No refusal creates a branch or moves main. +#[tokio::test] +async fn branch_create_at_a_commit_refuses_what_is_not_branch_history() { + let dir = tempfile::tempdir().unwrap(); + let db = init_and_load(&dir).await; + let loaded = db.list_commits(Some("main")).await.unwrap()[0].clone(); + + let proposal = db + .mutate_with_receipt( + "main", + r#" +query propose($name: String, $age: I32) as commit { + insert Person { name: $name, age: $age } +} +"#, + "propose", + &mixed_params(&[("$name", "Pat")], &[("$age", 7)]), + ) + .await + .unwrap() + .commit + .unwrap(); + let snapshot = db + .mutate_with_receipt( + "main", + r#" +query park($name: String, $age: I32) as unattached branch { + insert Person { name: $name, age: $age } +} +"#, + "park", + &mixed_params(&[("$name", "Una")], &[("$age", 9)]), + ) + .await + .unwrap() + .commit + .unwrap(); + db.branch_create("gone").await.unwrap(); + let on_gone = db + .mutate_with_receipt( + "gone", + MUTATION_QUERIES, + "insert_person", + &mixed_params(&[("$name", "Gil")], &[("$age", 3)]), + ) + .await + .unwrap() + .commit + .unwrap(); + db.branch_delete("gone").await.unwrap(); + + let refusal = |result: omnigraph::error::Result| { + let error = result.expect_err("a refused start creates nothing"); + assert_eq!( + error.completion_evidence(), + Some(CompletionEvidence::BeforeEffect), + "{error}" + ); + match error.into_completion_evidence().0 { + OmniError::BranchStartRefused { reason, .. } => reason, + other => panic!("expected a branch start refusal, got {other:?}"), + } + }; + let unwrapped = |result: omnigraph::error::Result| { + result + .expect_err("a refused start creates nothing") + .into_completion_evidence() + .0 + }; + assert_eq!( + refusal( + db.branch_create_from(at_commit(&proposal.graph_commit_id), "b1") + .await + ), + BranchStartRefusal::UnpublishedCommit + ); + assert_eq!( + refusal( + db.branch_create_from(at_commit(&snapshot.graph_commit_id), "b2") + .await + ), + BranchStartRefusal::UnattachedSnapshot + ); + let unknown = ulid::Ulid::new().to_string(); + assert!(matches!( + unwrapped(db.branch_create_from(at_commit(&unknown), "b3").await), + OmniError::Manifest(error) if error.kind == ManifestErrorKind::NotFound + )); + let deleted = BranchStartRefusal::BranchDeleted { + branch: "gone".to_string(), + }; + assert_eq!( + refusal( + db.branch_create_from(at_commit(&on_gone.graph_commit_id), "b4") + .await + ), + deleted, + "the branch that wrote the commit is gone" + ); + db.branch_create("gone").await.unwrap(); + assert_eq!( + refusal( + db.branch_create_from(at_commit(&on_gone.graph_commit_id), "b4") + .await + ), + deleted, + "a branch created again under the name did not write the commit" + ); + db.branch_delete("gone").await.unwrap(); + assert!(matches!( + unwrapped(db.branch_create_from(at_commit(&loaded.graph_commit_id), "main").await), + OmniError::Manifest(error) if error.message.contains("cannot create branch 'main'") + )); + + let schema = TEST_SCHEMA.replace( + " age: I32?\n}", + " age: I32?\n nickname: String?\n}", + ); + db.apply_schema(&schema).await.unwrap(); + let main_version = version_main(&db).await.unwrap(); + assert_eq!( + refusal( + db.branch_create_from(at_commit(&loaded.graph_commit_id), "b5") + .await + ), + BranchStartRefusal::SchemaChanged + ); + + assert_eq!(db.branch_list().await.unwrap(), vec!["main".to_string()]); + assert_eq!(version_main(&db).await.unwrap(), main_version); +} + #[tokio::test] async fn ensure_indices_on_child_branch_keeps_inherited_table_when_no_work_is_needed() { let dir = tempfile::tempdir().unwrap(); @@ -3919,7 +4270,7 @@ async fn branch_namespace_rejects_live_physical_path_prefix_collisions() { db.branch_create("feature").await.unwrap(); for result in [ db.branch_create("feature").await, - db.branch_create_from("main", "feature").await, + db.branch_create_from("main", "feature").await.map(drop), ] { let error = result.unwrap_err(); assert!( @@ -3929,7 +4280,9 @@ async fn branch_namespace_rejects_live_physical_path_prefix_collisions() { } for result in [ db.branch_create("feature/child").await, - db.branch_create_from("main", "feature/child").await, + db.branch_create_from("main", "feature/child") + .await + .map(drop), ] { let error = result.unwrap_err(); assert!( @@ -3976,7 +4329,7 @@ async fn branch_namespace_rejects_live_physical_path_prefix_collisions() { ); for result in [ db.branch_create("feature").await, - db.branch_create_from("main", "feature").await, + db.branch_create_from("main", "feature").await.map(drop), ] { let error = result.unwrap_err(); assert!( diff --git a/crates/omnigraph/tests/integration/maintenance.rs b/crates/omnigraph/tests/integration/maintenance.rs index d39e45ac0..15afbc9e8 100644 --- a/crates/omnigraph/tests/integration/maintenance.rs +++ b/crates/omnigraph/tests/integration/maintenance.rs @@ -17,8 +17,9 @@ use lance_core::datatypes::BlobHandling; use omnigraph::IndexCoverage; use omnigraph::db::{ CleanupPolicyOptions, MergeOutcome, Omnigraph, ReadTarget, RepairAction, RepairClassification, - RepairOptions, + RepairOptions, SnapshotId, }; +use omnigraph::error::{BranchStartRefusal, OmniError}; use omnigraph::loader::LoadMode; use omnigraph::{ExternalBlobBase, ExternalBlobExecutionScope, ExternalBlobPolicy}; @@ -2340,6 +2341,74 @@ async fn cleanup_preserves_main_version_pinned_by_live_lazy_branch() { ); } +/// A branch started at a past commit keeps that commit retained: its first +/// `__manifest` version is its base, which cleanup always retains with every +/// table version it pins. A commit whose table versions cleanup reclaimed +/// starts no branch: the refusal names the reclaimed table, and nothing is +/// created. +#[tokio::test] +async fn cleanup_retains_the_commit_a_branch_starts_at() { + let dir = tempfile::tempdir().unwrap(); + let db = init_and_load(&dir).await; + let loaded = db.list_commits(Some("main")).await.unwrap()[0] + .graph_commit_id + .clone(); + mutate_main( + &db, + MUTATION_QUERIES, + "insert_person", + &mixed_params(&[("$name", "Eve")], &[("$age", 40)]), + ) + .await + .unwrap(); + let eve = db.list_commits(Some("main")).await.unwrap()[0] + .graph_commit_id + .clone(); + add_person_fragments(&db).await; + let at = |id: &str| ReadTarget::snapshot(SnapshotId::new(id)); + + db.branch_create_from(at(&loaded), "at-load").await.unwrap(); + let people_at_load = count_rows_branch(&db, "at-load", "node:Person").await; + db.cleanup(CleanupPolicyOptions { + keep_versions: Some(1), + older_than: None, + }) + .await + .unwrap(); + assert_eq!( + count_rows_branch(&db, "at-load", "node:Person").await, + people_at_load, + "the branch's base keeps the commit's table versions" + ); + db.branch_create_from(at(&loaded), "again") + .await + .expect("a commit a live branch starts at stays retained"); + + let refused = db + .branch_create_from(at(&eve), "at-eve") + .await + .expect_err("cleanup reclaimed the Person version Eve's commit pins") + .into_completion_evidence() + .0; + assert!( + matches!( + &refused, + OmniError::BranchStartRefused { + graph_commit_id, + reason: BranchStartRefusal::NotRetained { table_key, .. }, + } if graph_commit_id == &eve && table_key == "node:Person" + ), + "{refused:?}" + ); + assert!( + !db.branch_list() + .await + .unwrap() + .iter() + .any(|branch| branch == "at-eve") + ); +} + #[tokio::test] async fn cleanup_uses_oldest_pin_across_multiple_live_lazy_branches() { let dir = tempfile::tempdir().unwrap(); diff --git a/crates/omnigraph/tests/policy_engine_chassis.rs b/crates/omnigraph/tests/policy_engine_chassis.rs index b17f826d9..b0b806cc4 100644 --- a/crates/omnigraph/tests/policy_engine_chassis.rs +++ b/crates/omnigraph/tests/policy_engine_chassis.rs @@ -23,7 +23,7 @@ use std::path::Path; use std::sync::Arc; use omnigraph::Session; -use omnigraph::db::{Omnigraph, ReadTarget}; +use omnigraph::db::{Omnigraph, ReadTarget, SnapshotId}; use omnigraph::error::{ManifestErrorKind, OmniError}; use omnigraph::loader::LoadMode; use omnigraph_policy::{PolicyChecker, PolicyEngine}; @@ -562,6 +562,24 @@ async fn branch_create_from_as_denies_when_policy_rejects_actor() { .branch_create_from_as(ReadTarget::branch("main"), "feature", Some("act-denied")) .await; assert_denied(result, "branch_create_from_as"); + + // A commit start is checked on the branch that wrote the commit, and a + // denial is the not-found of an unknown id: the id discloses nothing. + let head = db.list_commits(Some("main")).await.unwrap()[0] + .graph_commit_id + .clone(); + let result = db + .branch_create_from_as( + ReadTarget::snapshot(SnapshotId::new(head)), + "feature", + Some("act-denied"), + ) + .await; + assert!( + matches!(&result, Err(OmniError::Manifest(error)) if error.kind == ManifestErrorKind::NotFound), + "{result:?}" + ); + assert_eq!(db.branch_list().await.unwrap(), vec!["main".to_string()]); } #[tokio::test] @@ -572,6 +590,20 @@ async fn branch_create_from_as_allows_when_policy_permits_actor() { db.branch_create_from_as(ReadTarget::branch("main"), "feature", Some("act-allowed")) .await .expect("act-allowed should be able to BranchCreate from main"); + + let head = db.list_commits(Some("main")).await.unwrap()[0] + .graph_commit_id + .clone(); + let created = db + .branch_create_from_as( + ReadTarget::snapshot(SnapshotId::new(head.clone())), + "at-head", + Some("act-allowed"), + ) + .await + .expect("act-allowed should be able to BranchCreate at a commit of main"); + assert_eq!(created.from_commit_id, head); + assert_eq!(created.from, "main"); } #[tokio::test] @@ -744,7 +776,8 @@ async fn non_canonical_branch_names_cannot_reach_a_protected_branch() { ); let result = db .branch_create_from_as(ReadTarget::branch(main), "fresh", actor) - .await; + .await + .map(drop); check( format!("branch_create_from_as({main:?} -> fresh)"), result, diff --git a/docs/dev/writes.md b/docs/dev/writes.md index d354108b3..b1e03fbbf 100644 --- a/docs/dev/writes.md +++ b/docs/dev/writes.md @@ -76,6 +76,36 @@ native branch lifetime; a stale or missing view takes the existing refresh/open path. Captures share immutable lineage and the Lance session, copy current table state, and leave the handle's active branch unchanged. +A start at a graph commit (`from_commit`, `branch create … from commit ""`) captures the +branch whose incarnation wrote the commit and clones that branch's +`__manifest` at the version that published it: Lance `create_branch` takes any +retained version, so the new branch's first manifest holds the commit's table +pins, head, buffer and schema contract, exactly as a create while the commit +was the head. No second publication follows. Under the gates the engine +captures the branch by name and reads the commit's record from the capture. +The incarnation that wrote the commit must be the one the name resolves to: +a name that resolves to nothing and a name created again since are one fact, +refused as `BranchDeleted` in that one place, on every door. It then refuses, +before any effect, a version above the captured head, a schema contract +other than the accepted one, a version only the legacy history serves, a +version whose head row is another commit, and any pinned table version that +no longer opens (cleanup reclaimed it) or misses an object its manifest +references. The collector's sweep deletes a version's files before its +manifest, so a manifest that opens proves nothing after an interrupted run: +every object of every pin the captured head does not also hold (the head's +pins are roots, whose files the collector never deletes) is probed, as the +collector oracle probes retained roots. The probe runs under the exclusive +schema permit every write waits on, so it opens the pins and probes their +objects with bounded concurrency and stops at the first missing object; it +still sends one request per object of those pins. The probe stands in for a +property the collector does not yet give every reader: a manifest that is +present names only objects that are present. A sweep that deleted a version's +manifest before its files would give it, and the probe would then reduce to +opening each pin. Collection keeps the new branch's base version, and with +it the commit's pins, while the branch lives. +An unpublished commit and an unattached snapshot are detached `__manifest` +versions that no branch history holds, and are refused. + Create admission lists native ref names and reads bodies only for the source, the target's incarnations and logical ancestors or descendants. Physical path checks include retired ref names. The diff --git a/docs/user/branching/index.md b/docs/user/branching/index.md index e21cb67b9..66fa553e6 100644 --- a/docs/user/branching/index.md +++ b/docs/user/branching/index.md @@ -28,8 +28,10 @@ omnigraph branch list --store graph.omni omnigraph branch delete review/abandoned --store graph.omni ``` -Creating a branch defaults to `main` when `--from` is omitted. A load can create -a missing target branch by combining `--branch ` with `--from `. +Creating a branch defaults to `main` when `--from` is omitted. To start a +branch at a past commit instead of a branch head, use `--from-commit`; see +[start a branch at a commit](#start-a-branch-at-a-commit). A load can create a +missing target branch by combining `--branch ` with `--from `. Each branch operation is also a GQ statement, so a client that already sends `.gq` source needs no second transport. The three control writes go to @@ -38,6 +40,7 @@ with no request target, name, or parameters: ```text branch create "review/add-benchmark" from main +branch create "fix/import" from commit "" branch merge "review/add-benchmark" into main branch delete "review/abandoned" branch list @@ -149,6 +152,40 @@ costs. Served, the same listing is Paging applies to a branch's history, not to `--unpublished` or `--unattached`. +### Start a branch at a commit + +A branch can start at a published graph commit instead of a branch's head. It +holds that commit's rows and history, as if it had been created while the +commit was the head. Use it to try a fix or a revert where a change was made, +then merge the branch as usual; the commit is its merge base. + +```bash +omnigraph commit list graph.omni --branch main --json +omnigraph branch create fix/import --from-commit --store graph.omni +``` + +The statement is `branch create "fix/import" from commit ""`, and +`POST /branches` takes the id in `from_commit`, which excludes `from` (a +request with both answers `400`). An id no commit has answers `404`. The +receipt names the branch the new +branch was forked off in `from`, for a commit start the branch that wrote the +commit, and the commit it starts at in `from_commit_id`. Every receipt carries +`from_commit_id`: for a start from a branch it is that branch's head when the +branch was created. A branch starts only at a commit whose whole snapshot is +still there under the graph's current schema; otherwise nothing is created: + +| `from_commit` names | Answer | `branch_start_refusal.reason` | +|---|---|---| +| A commit whose table versions `cleanup` reclaimed, wholly or in a run that stopped part way | `410` | `not_retained`, with the type in `type_name` | +| A commit written under an earlier schema than the graph's | `409` | `schema_changed` | +| A commit written before the storage upgrade | `409` | `earlier_storage_format` | +| A commit of a branch that was deleted, even if a branch has since been created under its name | `409` | `branch_deleted`, with the branch in `branch` | +| An [unpublished commit](#unpublished-commits) | `409` | `unpublished_commit`; publish it, then start at the published commit | +| An [unattached snapshot](#unattached-branches) | `409` | `unattached_snapshot`; it never becomes branch history | + +While the new branch lives, `cleanup` keeps the commit's table versions, so a +later start at the same commit succeeds too. + ### Unpublished commits A mutation or a merge written `as commit` records its result as an unpublished diff --git a/docs/user/cli/reference.md b/docs/user/cli/reference.md index 67721422a..5e2977386 100644 --- a/docs/user/cli/reference.md +++ b/docs/user/cli/reference.md @@ -44,7 +44,7 @@ server resolves the actor from the bearer token. Drop it, or use `--store ` | `mutate` | Run an insert/update/delete query (`as commit` leaves it unpublished; `as unattached branch [on ""]` records a snapshot that never publishes), or a `branch create`, `branch delete`, `branch merge`, `commit publish`, `commit drop`, or [`load csv`](../mutations/index.md#csv-loading) statement | direct or served | | `load` | Load graph JSONL in `overwrite`, `append`, or `merge` mode | direct or served | | `blob get/stat/put/clear` | Read, inspect, replace or clear one Blob cell | direct or served | -| `branch create/list/delete/merge/diff/merge-base` | Manage graph branches; `merge --as-commit` leaves the merge unpublished, `merge --dry-run` reports the outcome or conflicts and writes nothing, `diff` reviews what a branch changed since its merge base, `merge-base` prints that base | direct or served | +| `branch create/list/delete/merge/diff/merge-base` | Manage graph branches; `create --from-commit` starts a branch at a commit, `merge --as-commit` leaves the merge unpublished, `merge --dry-run` reports the outcome or conflicts and writes nothing, `diff` reviews what a branch changed since its merge base, `merge-base` prints that base | direct or served | | `snapshot` | Show a branch snapshot: `internal_schema_version`, `graph_manifest_version`, and the datasets of one captured graph version | direct or served | | `commit list/show/changes/publish/drop` | Inspect history (`list --limit`/`--page-token` for one page of it, `list --unpublished` for a branch's unpublished commits, `list --unattached` for its live unattached snapshots; the two flags exclude each other and paging) or one commit's entity changes; publish an unpublished commit onto its recorded target branch (an unattached snapshot refuses with `unattached_cannot_publish`), or drop it (an unattached snapshot takes every descendant with it); `publish` and `drop` take `--as` on a direct graph, the reads refuse it | direct or served | | `diff` | Net entity changes between two points: branches or commit ids (`--from`, `--to`) | direct or served | diff --git a/docs/user/operations/maintenance.md b/docs/user/operations/maintenance.md index e3d38da1a..c511c238b 100644 --- a/docs/user/operations/maintenance.md +++ b/docs/user/operations/maintenance.md @@ -208,7 +208,10 @@ At least one retention option is required: A graph commit survives when either option retains it. The current HEAD of every live branch is always retained, and a named branch also keeps its oldest commit, the copy of the parent's commit it was created from, while -the branch lives. Cleanup also retains the selected merge bases between +the branch lives; for a branch started at a past commit, that is the commit +it names. A commit cleanup no longer retains starts no branch +([start a branch at a commit](../branching/index.md#start-a-branch-at-a-commit)). +Cleanup also retains the selected merge bases between current branch heads and accepted merge inputs. Native tags protect the exact snapshots they name. A tagged graph snapshot retains all its table versions; a tagged table snapshot retains that version, including a detached version or a diff --git a/docs/user/operations/policy.md b/docs/user/operations/policy.md index 306a0c320..e5b930be3 100644 --- a/docs/user/operations/policy.md +++ b/docs/user/operations/policy.md @@ -137,6 +137,11 @@ the target the base snapshot recorded, whatever branch the request names. snapshot need `read` on that recorded target; `commit show ` keeps its scope-free `read`; `commit drop` takes `commit_drop` on the recorded target as for an unpublished commit (see [Unattached graph branches](../../rfcs/2026-10-11-unattached-graph-branches.md)). +`branch_create` from a branch checks that branch as the source and the new +branch as the target. A start at a commit (`from_commit`) checks the branch +that wrote the commit as the source; a denial answers `404`, as an id no +commit has does, so the id discloses nothing (see +[start a branch at a commit](../branching/index.md#start-a-branch-at-a-commit)). ## Validate and test diff --git a/openapi.json b/openapi.json index 51fb9fb81..d6fa7f255 100644 --- a/openapi.json +++ b/openapi.json @@ -3073,7 +3073,7 @@ "branches" ], "summary": "Create a new branch.", - "description": "Forks `name` off of `from` (defaults to `main`). The new branch shares\nbacking dataset data with its parent until it is mutated. Returns 409 if `name`\nalready exists. The GQ statement `branch create` on `POST /mutate` runs\nthe same body.", + "description": "Forks `name` off of `from` (defaults to `main`) at its head, or starts it\nat the published graph commit `from_commit` with that commit's data and\nhistory. The receipt names the branch forked and the commit the new\nbranch starts at. The new branch shares backing dataset data with its\nparent until it is mutated. Returns 409 if `name` already exists. The GQ\nstatement `branch create` on `POST /mutate` runs the same body.", "operationId": "cluster_createBranch", "parameters": [ { @@ -3131,7 +3131,7 @@ } }, "400": { - "description": "Bad request; api_contract_mismatch when the required API contract header is missing, duplicated or unsupported", + "description": "Bad request, or both `from` and `from_commit`; api_contract_mismatch when the required API contract header is missing, duplicated or unsupported", "headers": { "omnigraph-http-api": { "schema": { @@ -3193,6 +3193,27 @@ } } }, + "404": { + "description": "`from` names no branch, or `from_commit` names no commit the actor may branch from", + "headers": { + "omnigraph-http-api": { + "schema": { + "type": "string", + "enum": [ + "0.13" + ] + }, + "description": "Supported OmniGraph HTTP API contract." + } + }, + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorOutput" + } + } + } + }, "408": { "description": "Request body deadline exceeded before operation admission", "headers": { @@ -3215,7 +3236,28 @@ } }, "409": { - "description": "Branch already exists", + "description": "Branch already exists, or the commit `from_commit` names cannot start a branch; see branch_start_refusal", + "headers": { + "omnigraph-http-api": { + "schema": { + "type": "string", + "enum": [ + "0.13" + ] + }, + "description": "Supported OmniGraph HTTP API contract." + } + }, + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorOutput" + } + } + } + }, + "410": { + "description": "The commit `from_commit` names is no longer retained; see branch_start_refusal", "headers": { "omnigraph-http-api": { "schema": { @@ -9832,7 +9874,8 @@ "required": [ "uri", "from", - "name" + "name", + "from_commit_id" ], "properties": { "actor_id": { @@ -9844,6 +9887,10 @@ "from": { "type": "string" }, + "from_commit_id": { + "type": "string", + "description": "The graph commit the new branch starts at: `from`'s head when the\nbranch was created, or the commit `from_commit` named. For a commit\nstart, `from` is the branch that wrote the commit." + }, "name": { "type": "string" }, @@ -9863,7 +9910,14 @@ "string", "null" ], - "description": "Parent branch to fork from. Defaults to `main`." + "description": "Parent branch to fork from; the new branch starts at its head.\nMutually exclusive with `from_commit`. Defaults to `main`." + }, + "from_commit": { + "type": [ + "string", + "null" + ], + "description": "A published graph commit id to start the new branch at, with that\ncommit's data and history. Mutually exclusive with `from`." }, "name": { "type": "string", @@ -10019,16 +10073,20 @@ "oneOf": [ { "type": "object", - "description": "`name` was forked off `from`.", + "description": "`name` was forked off `from` at the commit `from_commit_id`: the\nhead of `from`, or the commit a `from commit \"\"` start named,\nwhich `from` wrote.", "required": [ "from", "name", + "from_commit_id", "kind" ], "properties": { "from": { "type": "string" }, + "from_commit_id": { + "type": "string" + }, "kind": { "type": "string", "enum": [ @@ -10123,6 +10181,49 @@ ], "description": "The effect of a branch statement sent to `POST /mutate`, tagged by `kind`.\nA merge conflict has no kind: it is the 409 `POST /branches/merge` answers.\nThe statement grammar is `BranchStmt` in `omnigraph-compiler`." }, + "BranchStartRefusalOutput": { + "type": "object", + "description": "A branch create refused at the graph commit `from_commit` named; nothing\nwas created.", + "required": [ + "reason", + "graph_commit_id" + ], + "properties": { + "branch": { + "type": [ + "string", + "null" + ], + "description": "For `branch_deleted`, the branch that wrote the commit." + }, + "graph_commit_id": { + "type": "string" + }, + "reason": { + "$ref": "#/components/schemas/BranchStartRefusalReason" + }, + "type_name": { + "type": [ + "string", + "null" + ], + "description": "For `not_retained`, the graph type whose stored version cleanup reclaimed." + } + } + }, + "BranchStartRefusalReason": { + "type": "string", + "description": "Why a branch could not start at the graph commit `from_commit` named.", + "enum": [ + "not_retained", + "schema_changed", + "earlier_storage_format", + "branch_deleted", + "unpublished_commit", + "unattached_snapshot", + "unknown" + ] + }, "ChangeBaselineOutput": { "type": "object", "description": "Terminal payload of a baseline stream: the captured snapshot commit and\nthe cursor that resumes the feed immediately after it.", @@ -10931,6 +11032,17 @@ } ] }, + "branch_start_refusal": { + "oneOf": [ + { + "type": "null" + }, + { + "$ref": "#/components/schemas/BranchStartRefusalOutput", + "description": "Set with HTTP 409, or 410 for `not_retained`, when `branch create`\ncannot start a branch at the commit `from_commit` named." + } + ] + }, "change_diff_refusal": { "oneOf": [ { diff --git a/skills/omnigraph/references/changes.md b/skills/omnigraph/references/changes.md index f044d2991..450a8cf6b 100644 --- a/skills/omnigraph/references/changes.md +++ b/skills/omnigraph/references/changes.md @@ -24,7 +24,8 @@ delete`, or `branch merge` statement without `name`, `params`, or a request outcome carries `merge` (`already_up_to_date`, `fast_forward`, or `merged`). Affected entity counts are zero because these are branch controls. -Creation and deletion return `commit: null` despite changing branch state. A +Creation and deletion return `commit: null` despite changing branch state; a +created outcome's `from_commit_id` names the commit the new branch starts at. A `fast_forward` or `merged` result returns `commit`, the exact commit that merge published, even when another writer has since advanced the target; an `already_up_to_date` merge publishes nothing and returns `commit: null`. diff --git a/skills/omnigraph/references/commands.md b/skills/omnigraph/references/commands.md index 21acd68f2..2abfe85f8 100644 --- a/skills/omnigraph/references/commands.md +++ b/skills/omnigraph/references/commands.md @@ -128,7 +128,8 @@ omnigraph branch merge --into main --delete-branch --store $REPO omnigraph branch delete --store $REPO ``` -All support `--json`. `--from` and `--into` default to `main`. +All support `--json`. `--from` and `--into` default to `main`. `branch create +--from-commit ` starts the branch at a published commit instead. `--delete-branch` removes the source only after a successful merge publication. A served `branch delete --json` needs `--yes`. diff --git a/skills/omnigraph/references/queries.md b/skills/omnigraph/references/queries.md index ca8b1563c..3c4341b54 100644 --- a/skills/omnigraph/references/queries.md +++ b/skills/omnigraph/references/queries.md @@ -450,7 +450,8 @@ branch delete "review/abandoned" branch list ``` -`from` and `into` default to `main`. Quote a name that is not a bare +`from` and `into` default to `main`; `branch create "fix" from commit ""` +starts the branch at a published commit. Quote a name that is not a bare identifier (a lowercase letter or `_`, then letters, digits, or `_`), such as one containing `/`, `-`, or `.`. Run the control writes with `omnigraph mutate -e '…'` (HTTP `POST /mutate`) and `branch list` with