spiral/

commit.rs

//! Staged table and name changes, published onto a pinned branch head as one Catalog commit.
use std::fmt::{self, Debug, Formatter};
use std::sync::Arc;

use serde::{Deserialize, Serialize};
use uuid::Uuid;

use crate::api::catalog::{
    AdvanceResourceChange, AssertNamespaceChange, AssertResourceChange, BindNamespaceChange,
    CatalogError, CommitId, CreateResourceChange, NamespaceChange, OperationId, ProjectOperation,
    Publication, PublicationConflict, PublishOperationRequest, PublishOperationResponse,
    PublishRequest, ResourceChange, ResourceKind, RetireResourceChange, UnbindNamespaceChange,
};
use crate::api::connections::{
    Connection as ExternalConnection, CreateConnectionRequest, Credential,
};
#[cfg(feature = "tables")]
use crate::api::enrichments::{
    CreateEnrichmentRequest, CreateVersionRequest as CreateEnrichmentVersionRequest,
};
use crate::catalog_types::{
    IdempotencyKey, NamespaceParent, NamespaceSegment, ProjectCommitId, ProjectTransactionId,
    ResourceDescriptor, ResourceId, ResourceServiceId, VersionId,
};
use crate::connections::require_connection;
#[cfg(feature = "tables")]
use crate::enrichment::{
    ENRICHMENT_KIND, ENRICHMENT_VALUE_COLUMN, EnrichmentDefinition, enrichment_coverage,
};
use crate::namespace::{Namespace, resource_path};
#[cfg(feature = "tables")]
use crate::table::CatalogBase;
#[cfg(feature = "tables")]
use crate::vector_index::{VECTOR_INDEX_KIND, VECTOR_INDEX_SERVICE};
use crate::{BranchHead, CallError, Error, Project};
#[cfg(feature = "tables")]
use crate::{Expression, NamedExpression, ScanOptions, Table, VectorIndex};

#[derive(Clone)]
struct ResourceMutation {
    resource_id: ResourceId,
    expected: Option<VersionId>,
    replacement: Option<VersionId>,
    create: Option<ResourceDescriptor>,
}

impl ResourceMutation {
    fn wire(&self) -> ResourceChange {
        match (&self.create, self.expected, self.replacement) {
            (Some(descriptor), None, Some(version_id)) => {
                ResourceChange::Create(CreateResourceChange {
                    descriptor: descriptor.clone(),
                    version_id,
                })
            }
            (None, Some(expected), Some(replacement)) if expected != replacement => {
                ResourceChange::Advance(AdvanceResourceChange {
                    resource_id: self.resource_id,
                    expected,
                    replacement,
                })
            }
            (None, Some(expected), None) => ResourceChange::Retire(RetireResourceChange {
                resource_id: self.resource_id,
                expected,
            }),
            (None, expected, replacement) if expected == replacement => {
                ResourceChange::Assert(AssertResourceChange {
                    resource_id: self.resource_id,
                    expected,
                })
            }
            _ => unreachable!("SDK staging only creates valid resource mutations"),
        }
    }
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) struct BindingMutation {
    pub(crate) parent: NamespaceParent,
    pub(crate) name: NamespaceSegment,
    pub(crate) expected: Option<ResourceId>,
    pub(crate) replacement: Option<ResourceId>,
}

impl BindingMutation {
    fn wire(&self) -> NamespaceChange {
        match (self.expected, self.replacement) {
            (expected, Some(replacement)) if expected != Some(replacement) => {
                NamespaceChange::Bind(BindNamespaceChange {
                    parent: self.parent.clone(),
                    name: self.name.clone(),
                    expected,
                    replacement,
                })
            }
            (Some(expected), None) => NamespaceChange::Unbind(UnbindNamespaceChange {
                parent: self.parent.clone(),
                name: self.name.clone(),
                expected,
            }),
            (expected, _) => NamespaceChange::Assert(AssertNamespaceChange {
                parent: self.parent.clone(),
                name: self.name.clone(),
                expected,
            }),
        }
    }
}

/// Staged changes to tables and names, published onto a pinned branch head as one Catalog commit.
///
/// Staging only records which versions and names change: tables write their new versions
/// themselves, and staging does no I/O except for Connections. Publishing succeeds only while
/// the branch still points at the pinned head, so readers see every staged change or none.
pub struct Commit {
    head: BranchHead,
    operation_id: Uuid,
    message: Option<String>,
    resources: Vec<ResourceMutation>,
    namespace: Vec<BindingMutation>,
}

impl Commit {
    pub(crate) fn new(head: BranchHead) -> Self {
        Self {
            head,
            operation_id: Uuid::now_v7(),
            message: None,
            resources: Vec::new(),
            namespace: Vec::new(),
        }
    }

    /// The branch head this commit publishes onto.
    pub fn head(&self) -> &BranchHead {
        &self.head
    }

    /// Durable operation identity used for publication and recovery.
    pub fn operation_id(&self) -> Uuid {
        self.operation_id
    }

    /// Record `message` with the commit.
    pub fn message(mut self, message: impl Into<String>) -> Self {
        self.message = Some(message.into());
        self
    }

    fn require_unstaged(&self, id: ResourceId) -> Result<(), Error> {
        if self.resources.iter().any(|change| change.resource_id == id) {
            return Err(Error::invalid("a resource may change once per commit"));
        }
        Ok(())
    }

    /// The Catalog base of a table opened at this commit's head.
    #[cfg(feature = "tables")]
    fn head_base<'t>(&self, table: &'t Table) -> Result<&'t CatalogBase, Error> {
        table
            .base
            .as_ref()
            .filter(|base| {
                Arc::ptr_eq(&table.connection, &self.head.project.client.connection)
                    && base.project_id == self.head.project.id
                    && base.commit_id == self.head.commit_id
            })
            .ok_or_else(|| Error::invalid("table was not opened at this commit's branch head"))
    }

    /// Parse a root name that no staged change binds; a staged drop or rename frees it.
    #[cfg(feature = "tables")]
    fn unclaimed_root_name(&self, name: &str) -> Result<NamespaceSegment, Error> {
        let name = NamespaceSegment::new(name).map_err(Error::invalid)?;
        if self.namespace.iter().any(|staged| {
            staged.parent == NamespaceParent::project_root()
                && staged.name == name
                && staged.replacement.is_some()
        }) {
            return Err(Error::invalid(format!("name `{name}` is already staged")));
        }
        Ok(name)
    }

    /// Stage one binding change per slot. A removal and a creation of the same slot
    /// merge into one rebinding, because the Catalog compares each slot once.
    fn stage_binding(&mut self, change: BindingMutation) -> Result<(), Error> {
        let staged = self
            .namespace
            .iter_mut()
            .find(|staged| staged.parent == change.parent && staged.name == change.name);
        match staged {
            None => self.namespace.push(change),
            Some(staged) if staged.expected.is_none() && change.replacement.is_none() => {
                staged.expected = change.expected;
            }
            Some(staged) if staged.replacement.is_none() && change.expected.is_none() => {
                staged.replacement = change.replacement;
            }
            Some(staged) => {
                return Err(Error::invalid(format!(
                    "binding `{}` is already staged",
                    staged.name
                )));
            }
        }
        Ok(())
    }

    /// Stage `table` as its resource's version, in place of the version the head binds.
    ///
    /// `table` is a table opened at the head or a version written from one, through any number
    /// of writes; only the head's version and `table` enter the commit. Stage each table once,
    /// with its newest version. Staging the head's own version asserts that it is unchanged.
    #[cfg(feature = "tables")]
    pub fn update_table(&mut self, table: &Table) -> Result<(), Error> {
        let base = self.head_base(table)?;
        let resource_id = ResourceId::new(table.identity().resource_id).map_err(Error::invalid)?;
        self.require_unstaged(resource_id)?;
        self.resources.push(ResourceMutation {
            resource_id,
            expected: Some(VersionId::new(base.version_id).map_err(Error::invalid)?),
            replacement: Some(VersionId::new(table.identity().version_id).map_err(Error::invalid)?),
            create: None,
        });
        Ok(())
    }

    /// Stage a new table, written by [`Project::write_table`] or [`Project::table_writer`],
    /// and bind it at the project root as `name`.
    ///
    /// Names are one namespace segment. Dropping a root name and creating it again publishes
    /// one rebinding.
    #[cfg(feature = "tables")]
    pub fn create_table(&mut self, name: &str, table: &Table) -> Result<(), Error> {
        if table.base.is_some() {
            return Err(Error::invalid(
                "table is already in the Catalog; rename it to change its name",
            ));
        }
        if !Arc::ptr_eq(&table.connection, &self.head.project.client.connection) {
            return Err(Error::invalid("table was written through another client"));
        }
        let identity = table.identity();
        self.stage_root_creation(
            name,
            identity.resource_id,
            identity.version_id,
            "arbor",
            "table",
        )
    }

    /// Stage a new Vector Index, created by [`Project::create_vector_index`], and bind it at
    /// the project root as `name`.
    #[cfg(feature = "tables")]
    pub fn create_vector_index(&mut self, name: &str, index: &VectorIndex) -> Result<(), Error> {
        if !index.is_unbound() {
            return Err(Error::invalid(
                "The Vector Index already belongs to a snapshot",
            ));
        }
        if index.project.id() != self.head.project.id()
            || !Arc::ptr_eq(
                &index.project.client.connection,
                &self.head.project.client.connection,
            )
        {
            return Err(Error::invalid(
                "The Vector Index belongs to another project or client",
            ));
        }
        self.stage_root_creation(
            name,
            index.resource_id(),
            index.version_id(),
            VECTOR_INDEX_SERVICE,
            VECTOR_INDEX_KIND,
        )
    }

    /// Stage a new resource at `version_id` under the root name `name`.
    ///
    /// Nothing is staged unless every step succeeds.
    #[cfg(feature = "tables")]
    fn stage_root_creation(
        &mut self,
        name: &str,
        resource_id: Uuid,
        version_id: Uuid,
        service: &str,
        kind: &str,
    ) -> Result<(), Error> {
        let name = self.unclaimed_root_name(name)?;
        let resource_id = ResourceId::new(resource_id).map_err(Error::invalid)?;
        self.require_unstaged(resource_id)?;
        let creation = ResourceMutation {
            resource_id,
            expected: None,
            replacement: Some(VersionId::new(version_id).map_err(Error::invalid)?),
            create: Some(ResourceDescriptor {
                namespace_root: None,
                namespace_managed: false,
                resource_id,
                service: ResourceServiceId::new(service).map_err(Error::invalid)?,
                kind: ResourceKind(kind.to_owned()),
            }),
        };
        self.stage_binding(BindingMutation {
            parent: NamespaceParent::project_root(),
            name,
            expected: None,
            replacement: Some(resource_id),
        })?;
        self.resources.push(creation);
        Ok(())
    }

    /// Create an Enrichment of `source`, a table opened at this commit's head, and stage it as
    /// `name` under the table, such as `frames/is_blurry`.
    ///
    /// The Enrichment computes `expression` over each source row. Creation computes nothing:
    /// it checks the expression against the source schema and records it with its output type.
    /// Run [`Enrichment::refresh`](crate::Enrichment::refresh) on the published commit to
    /// compute the values. Unlike table changes, this calls the Tables API.
    #[cfg(feature = "tables")]
    pub async fn create_enrichment(
        &mut self,
        source: &Table,
        name: &str,
        expression: Expression,
    ) -> Result<(), Error> {
        let base = self.head_base(source)?;
        let source_id = ResourceId::new(source.identity().resource_id).map_err(Error::invalid)?;
        let parent = NamespaceParent::resource(source_id);
        let name = NamespaceSegment::new(name).map_err(Error::invalid)?;
        if self
            .namespace
            .iter()
            .any(|staged| staged.parent == parent && staged.name == name)
        {
            return Err(Error::invalid(format!("name `{name}` is already staged")));
        }
        if source
            .schema()
            .await?
            .field_with_name(name.as_str())
            .is_ok()
        {
            return Err(Error::invalid(format!(
                "`{name}` is a column of the source table"
            )));
        }
        let output = source
            .scan(ScanOptions {
                projection: Some(vec![NamedExpression::new(
                    ENRICHMENT_VALUE_COLUMN,
                    expression.clone(),
                )]),
                ..ScanOptions::default()
            })
            .await?
            .schema()
            .field(0)
            .clone();
        let definition = EnrichmentDefinition { expression, output }.to_wire()?;
        let arbor = self.head.project.client.tables();
        let root = arbor
            .create_enrichment(&CreateEnrichmentRequest {
                project_id: self.head.project.id.to_string(),
                source_table_id: source.identity().resource_id,
            })
            .await?
            .root;
        let version = arbor
            .create_enrichment_version(
                root.enrichment_id,
                &CreateEnrichmentVersionRequest {
                    base_version_id: None,
                    source_version_id: base.version_id,
                    source_project_commit_id: self.head.commit_id.as_uuid(),
                    definition,
                    coverage: enrichment_coverage(false)?,
                    native_state: None,
                },
            )
            .await?
            .version;
        let resource_id = ResourceId::new(root.enrichment_id).map_err(Error::invalid)?;
        self.stage_binding(BindingMutation {
            parent,
            name,
            expected: None,
            replacement: Some(resource_id),
        })?;
        self.resources.push(ResourceMutation {
            resource_id,
            expected: None,
            replacement: Some(VersionId::new(version.version_id).map_err(Error::invalid)?),
            create: Some(ResourceDescriptor {
                namespace_root: None,
                namespace_managed: false,
                resource_id,
                service: ResourceServiceId::new("arbor").map_err(Error::invalid)?,
                kind: ResourceKind(ENRICHMENT_KIND.into()),
            }),
        });
        Ok(())
    }

    /// Stage moving a table opened by path at the head to a root name.
    ///
    /// Only bindings change: the resource keeps its identity and version, and the commit can
    /// still update it. The old path is unbound like a drop's, so its root name can be
    /// created again.
    #[cfg(feature = "tables")]
    pub fn rename_table(&mut self, table: &Table, name: &str) -> Result<(), Error> {
        self.head_base(table)?;
        let name = self.unclaimed_root_name(name)?;
        let unbinding = table.unbinding()?;
        if unbinding.parent == NamespaceParent::project_root() && unbinding.name == name {
            return Err(Error::invalid("table already has this name"));
        }
        let binding = BindingMutation {
            parent: NamespaceParent::project_root(),
            name,
            expected: None,
            replacement: unbinding.expected,
        };
        // The new root name is unclaimed, so only unbinding the old path can fail.
        self.stage_binding(unbinding)?;
        self.stage_binding(binding)
    }

    /// Stage retiring a table opened by path at the head and unbinding its path.
    ///
    /// The path is unbound from the project root; its leading segments are an implicit
    /// prefix, not a parent resource. Dropping a root name and creating it again
    /// publishes one rebinding.
    #[cfg(feature = "tables")]
    pub fn drop_table(&mut self, table: &Table) -> Result<(), Error> {
        let base = self.head_base(table)?;
        let resource_id = ResourceId::new(table.identity().resource_id).map_err(Error::invalid)?;
        self.require_unstaged(resource_id)?;
        let retirement = ResourceMutation {
            resource_id,
            expected: Some(VersionId::new(base.version_id).map_err(Error::invalid)?),
            replacement: None,
            create: None,
        };
        self.stage_binding(table.unbinding()?)?;
        self.resources.push(retirement);
        Ok(())
    }

    /// Create a Connection and stage its name for publication on main.
    ///
    /// Only the service request carries the credential. The Catalog receives IDs.
    /// The Connection is not discoverable by name until this commit publishes.
    /// Unlike table changes, this calls the Connection service and reads the names at the head.
    pub async fn create_connection(
        &mut self,
        path: &str,
        uri: String,
        credential: Credential,
        description: Option<String>,
    ) -> Result<ExternalConnection, Error> {
        let path = resource_path(path)?;
        self.head.require_main_branch().await?;
        let mut namespace = Namespace::read(&self.head).await?;
        namespace.apply(
            self.resources.iter().map(ResourceMutation::wire),
            self.namespace.iter().map(BindingMutation::wire),
        )?;
        if namespace.resolve(&path)?.is_some() {
            return Err(Error::invalid(
                "a resource already has this Connection name",
            ));
        }
        let (parent, name) = namespace.parent(&path)?;
        let created = self
            .head
            .project
            .client
            .connections()
            .create_connection(CreateConnectionRequest {
                project_id: self.head.project.id.as_str().into(),
                uri,
                credential,
                description,
            })
            .await?;
        let id = ResourceId::new(created.connection.connection_id).map_err(Error::invalid)?;
        let version = VersionId::new(created.version_id).map_err(Error::invalid)?;
        self.stage_binding(BindingMutation {
            parent,
            name,
            expected: None,
            replacement: Some(id),
        })?;
        self.resources.push(ResourceMutation {
            resource_id: id,
            expected: None,
            replacement: Some(version),
            create: Some(ResourceDescriptor {
                resource_id: id,
                service: ResourceServiceId::new("external-connection").map_err(Error::invalid)?,
                kind: ResourceKind("connection".into()),
                namespace_managed: false,
                namespace_root: None,
            }),
        });
        Ok(created.connection)
    }

    /// Stage retiring a Connection and unbinding its name on main.
    ///
    /// The service deletes the secret after observing the published retirement.
    /// Abandoning the commit or a publication conflict leaves the secret intact.
    /// Unlike table changes, this reads the names at the head.
    pub async fn drop_connection(&mut self, path: &str) -> Result<(), Error> {
        let path = resource_path(path)?;
        self.head.require_main_branch().await?;
        let mut namespace = Namespace::read(&self.head).await?;
        namespace.apply(
            self.resources.iter().map(ResourceMutation::wire),
            self.namespace.iter().map(BindingMutation::wire),
        )?;
        let binding = namespace
            .resolve(&path)?
            .ok_or_else(|| Error::NotFound("Connection".into()))?;
        require_connection(&binding.resource)?;
        let id = binding.resource.descriptor.resource_id;
        self.require_unstaged(id)?;
        self.stage_binding(BindingMutation {
            parent: binding.parent,
            name: binding.name,
            expected: Some(id),
            replacement: None,
        })?;
        self.resources.push(ResourceMutation {
            resource_id: id,
            expected: Some(binding.resource.version_id),
            replacement: None,
            create: None,
        });
        Ok(())
    }

    /// Fix the publication request without I/O, so it can be saved before publishing.
    pub fn freeze(self) -> Result<FrozenCommit, Error> {
        // A slot renamed away and back to one resource, as through an alias, ends
        // unchanged; the Catalog rejects such entries and the pinned head covers them.
        let namespace = self
            .namespace
            .iter()
            .filter(|binding| binding.expected != binding.replacement)
            .map(BindingMutation::wire)
            .collect::<Vec<_>>();
        if self.resources.is_empty() && namespace.is_empty() {
            return Err(Error::invalid("commit has no staged changes"));
        }
        let publication = PublishRequest {
            branch_id: self.head.branch_id(),
            base_commit_id: self.head.commit_id,
            expected_head_commit_id: Some(self.head.commit_id),
            transaction_id: ProjectTransactionId::new(Uuid::now_v7()).map_err(Error::invalid)?,
            idempotency_key: IdempotencyKey::new(self.operation_id.to_string())
                .map_err(Error::invalid)?,
            message: self.message,
            resources: self.resources.iter().map(ResourceMutation::wire).collect(),
            namespace,
        };
        publication.validate().map_err(Error::invalid)?;
        let input = serde_json::to_value(&publication).map_err(Error::invalid)?;
        Ok(FrozenCommit {
            head: self.head,
            request: Arc::new(PublishOperationRequest {
                operation: ProjectOperation {
                    operation_id: OperationId(self.operation_id),
                    input,
                    result: serde_json::Value::Null,
                },
                publication,
            }),
        })
    }

    /// Publish onto the pinned head and return the branch's new head.
    ///
    /// On [`PublishError::Unknown`], recover or publish the returned [`FrozenCommit`].
    pub async fn publish(self) -> Result<BranchHead, PublishError> {
        self.freeze()?.publish().await
    }
}

/// A commit whose publication request is fixed. Publishing it again sends the same request,
/// and the Catalog publishes one operation at most once.
#[derive(Clone)]
pub struct FrozenCommit {
    head: BranchHead,
    request: Arc<PublishOperationRequest>,
}

impl Debug for FrozenCommit {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("FrozenCommit")
            .field("operation_id", &self.operation_id())
            .field("base_commit_id", &self.head.commit_id)
            .finish_non_exhaustive()
    }
}

impl FrozenCommit {
    pub(crate) fn resume(project: &Project, saved: SavedCommit) -> Result<Self, Error> {
        let SavedCommit(request) = saved;
        request.publication.validate().map_err(Error::invalid)?;
        if request.publication.expected_head_commit_id != Some(request.publication.base_commit_id)
            || request.operation.input
                != serde_json::to_value(&request.publication).map_err(Error::invalid)?
            || request.publication.idempotency_key.as_str()
                != request.operation.operation_id.to_string()
            || request.operation.result != serde_json::Value::Null
        {
            return Err(Error::invalid(
                "saved commit does not match its base or operation identity",
            ));
        }
        Ok(Self {
            head: BranchHead::new(
                project.clone(),
                request.publication.base_commit_id,
                request.publication.branch_id,
            ),
            request: Arc::new(request),
        })
    }

    /// Durable operation identity used for publication and recovery.
    pub fn operation_id(&self) -> Uuid {
        self.request.operation.operation_id.0
    }

    /// The branch head this commit publishes onto.
    pub fn head(&self) -> &BranchHead {
        &self.head
    }

    /// The publication request to save durably, to resume with [`Project::resume_commit`].
    pub fn save(&self) -> SavedCommit {
        SavedCommit(PublishOperationRequest::clone(&self.request))
    }

    /// Publish onto the pinned head and return the branch's new head.
    ///
    /// Publishing a commit that already published returns that publication's head.
    pub async fn publish(&self) -> Result<BranchHead, PublishError> {
        let project = &self.head.project;
        let response = project
            .client
            .connection
            .catalog
            .publish_operation(&project.id, &self.request)
            .await;
        match response {
            Ok(PublishOperationResponse::Published(receipt)) => {
                Ok(self.published(receipt.publication))
            }
            Ok(PublishOperationResponse::Conflict(conflict)) => {
                Err(PublishError::Conflict(Box::new(conflict)))
            }
            Ok(PublishOperationResponse::Unknown { .. }) => {
                Err(self.unknown("unrecognized publication response"))
            }
            Err(Error::Catalog(error)) => Err(match moved_head(&error) {
                Some(current) => PublishError::BranchMoved { current },
                None if publication_outcome_uncertain(&error) => self.unknown(error),
                None => PublishError::Failed(Error::Catalog(error)),
            }),
            Err(error) => Err(PublishError::Failed(error)),
        }
    }

    /// Look up a publication whose acknowledgement may have been lost.
    ///
    /// `None` means no publication was observed; publishing again is safe.
    pub async fn recover(&self) -> Result<Option<BranchHead>, Error> {
        let project = &self.head.project;
        let receipt = project
            .client
            .connection
            .catalog
            .read_operation(&project.id, OperationId(self.operation_id()))
            .await?;
        let Some(receipt) = receipt.receipt else {
            return Ok(None);
        };
        if receipt.operation != self.request.operation {
            return Err(Error::invalid(
                "operation receipt does not match this commit",
            ));
        }
        Ok(Some(self.published(receipt.publication)))
    }

    fn published(&self, publication: Publication) -> BranchHead {
        BranchHead::new(
            self.head.project.clone(),
            publication.commit_id,
            self.head.branch_id(),
        )
    }

    fn unknown(&self, message: impl fmt::Display) -> PublishError {
        PublishError::Unknown {
            commit: Box::new(self.clone()),
            message: message.to_string(),
        }
    }
}

/// A frozen commit saved durably, so that it can be recovered or published again after the
/// process restarts. Serialize it with any serde format.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(transparent)]
pub struct SavedCommit(PublishOperationRequest);

/// Publishing a commit failed, conflicted, or has an unknown outcome.
#[derive(Debug, thiserror::Error)]
pub enum PublishError {
    /// The branch moved past the pinned head. Nothing was published; pin the branch again and
    /// redo the changes from the tables at the new head.
    #[error("the branch moved to {current} before publication")]
    BranchMoved {
        /// The branch's head when publication was rejected.
        current: ProjectCommitId,
    },
    /// A staged change conflicts with the pinned head, such as creating a name it already
    /// binds. Nothing was published.
    #[error("a staged change conflicts with the branch head")]
    Conflict(Box<PublicationConflict>),
    /// The commit may have published. Recover or publish this same frozen commit; do not
    /// redo its changes.
    #[error("publication outcome unknown for {}: {message}", .commit.operation_id())]
    Unknown {
        commit: Box<FrozenCommit>,
        message: String,
    },
    /// The Catalog rejected the commit, or it was never sent. Nothing was published.
    #[error(transparent)]
    Failed(#[from] Error),
}

impl From<PublishError> for Error {
    fn from(error: PublishError) -> Self {
        match error {
            PublishError::BranchMoved { current } => Self::BranchMoved { current },
            PublishError::Conflict(conflict) => Self::Conflict(conflict),
            PublishError::Unknown { commit, message } => Self::OutcomeUnknown {
                operation_id: commit.operation_id(),
                message,
            },
            PublishError::Failed(error) => error,
        }
    }
}

/// The branch's head when the Catalog rejected a publication because the branch moved.
fn moved_head(error: &CallError<CatalogError>) -> Option<ProjectCommitId> {
    let CallError::Service(failure) = error else {
        return None;
    };
    let Some(CatalogError::StaleHead { current, .. }) = failure.declared() else {
        return None;
    };
    current.parse().ok().map(CommitId)
}

fn publication_outcome_uncertain(error: &CallError<CatalogError>) -> bool {
    let status = match error {
        CallError::Service(failure) => Some(failure.status()),
        CallError::UnknownRemote(failure) => Some(failure.status()),
        CallError::Transport(failure) => failure.observed_status(),
        CallError::Protocol(failure) => failure.observed_status(),
        CallError::Internal { .. } => None,
    };
    !status.is_some_and(|status| (400..500).contains(&status))
}

#[cfg(test)]
#[cfg(feature = "tables")]
mod tests {
    use catalog_api::types::*;
    use catalog_model::journal_model::ProjectJournalModel;
    use uuid::Uuid;

    use super::{Commit, publication_outcome_uncertain};
    use crate::api::catalog::{
        BranchId as WireBranchId, CatalogError, CommitId as WireCommitId, NamespaceChange,
        NamespaceParent as WireParent, PublishRequest as WirePublishRequest, ResourceChange,
    };
    use crate::scandal::types::TableWriteResult;
    use crate::table::CatalogBase;
    use crate::{BranchHead, CallError, Client, Table};

    fn legacy_parent(parent: &WireParent) -> NamespaceParent {
        match parent {
            WireParent::ProjectRoot(_) => NamespaceParent::ProjectRoot,
            WireParent::Resource(parent) => NamespaceParent::Resource {
                resource_id: ResourceId::new(parent.resource_id.0).unwrap(),
            },
            WireParent::Prefix(parent) => NamespaceParent::Prefix {
                resource_id: parent.resource_id.map(|id| ResourceId::new(id.0).unwrap()),
                path: NamespacePath::new(parent.path.0.iter().map(|s| segment(&s.0)).collect())
                    .unwrap(),
            },
            WireParent::Unknown { .. } => panic!("unexpected namespace parent"),
        }
    }

    fn legacy_publication(request: &WirePublishRequest) -> PublishProjectRequest {
        PublishProjectRequest {
            branch_id: ProjectBranchId::new(request.branch_id.0).unwrap(),
            base_commit_id: ProjectCommitId::new(request.base_commit_id.0).unwrap(),
            expected_head_commit_id: request
                .expected_head_commit_id
                .map(|id| ProjectCommitId::new(id.0).unwrap()),
            transaction_id: ProjectTransactionId::new(request.transaction_id.0).unwrap(),
            idempotency_key: IdempotencyKey::new(request.idempotency_key.0.clone()).unwrap(),
            message: request.message.clone(),
            resources: request
                .resources
                .iter()
                .map(|change| {
                    let (id, expected, replacement, create) = match change {
                        ResourceChange::Create(c) => (
                            c.descriptor.resource_id,
                            None,
                            Some(c.version_id),
                            Some(ResourceDescriptor {
                                resource_id: ResourceId::new(c.descriptor.resource_id.0).unwrap(),
                                service: ResourceServiceId::new(c.descriptor.service.0.clone())
                                    .unwrap(),
                                kind: ResourceKindName::new(c.descriptor.kind.0.clone()).unwrap(),
                                namespace_root: c
                                    .descriptor
                                    .namespace_root
                                    .map(|id| ResourceId::new(id.0).unwrap()),
                                namespace_managed: c.descriptor.namespace_managed,
                            }),
                        ),
                        ResourceChange::Advance(c) => {
                            (c.resource_id, Some(c.expected), Some(c.replacement), None)
                        }
                        ResourceChange::Retire(c) => (c.resource_id, Some(c.expected), None, None),
                        ResourceChange::Assert(c) => (c.resource_id, c.expected, c.expected, None),
                        ResourceChange::Unknown { .. } => panic!("unexpected resource change"),
                    };
                    ResourceCas {
                        resource_id: ResourceId::new(id.0).unwrap(),
                        expected: expected.map(|id| VersionId::new(id.0).unwrap()),
                        replacement: replacement.map(|id| VersionId::new(id.0).unwrap()),
                        create,
                    }
                })
                .collect(),
            namespace: request
                .namespace
                .iter()
                .map(|change| {
                    let (parent, name, expected, replacement) = match change {
                        NamespaceChange::Bind(c) => {
                            (&c.parent, &c.name, c.expected, Some(c.replacement))
                        }
                        NamespaceChange::Unbind(c) => (&c.parent, &c.name, Some(c.expected), None),
                        NamespaceChange::Assert(c) => (&c.parent, &c.name, c.expected, c.expected),
                        NamespaceChange::Unknown { .. } => panic!("unexpected namespace change"),
                    };
                    NamespaceCas {
                        parent: legacy_parent(parent),
                        name: segment(&name.0),
                        expected: expected.map(|id| ResourceId::new(id.0).unwrap()),
                        replacement: replacement.map(|id| ResourceId::new(id.0).unwrap()),
                    }
                })
                .collect(),
        }
    }

    #[test]
    fn publication_distinguishes_rejections_from_uncertain_outcomes() {
        for status in [400, 401, 403, 404, 409, 422] {
            let error = CallError::<CatalogError>::unknown_remote(status, None, vec![], false);
            assert!(!publication_outcome_uncertain(&error));
        }
        for error in [
            CallError::<CatalogError>::unknown_remote(500, None, vec![], false),
            CallError::protocol_response(200, "invalid acknowledgement"),
            CallError::transport("connection closed"),
        ] {
            assert!(publication_outcome_uncertain(&error));
        }
    }

    fn actor() -> ActorId {
        ActorId::new("user:test").unwrap()
    }

    fn segment(value: &str) -> NamespaceSegment {
        NamespaceSegment::new(value).unwrap()
    }

    fn prefix(path: &[&str]) -> NamespaceParent {
        NamespaceParent::Prefix {
            resource_id: None,
            path: NamespacePath::new(path.iter().copied().map(segment).collect()).unwrap(),
        }
    }

    fn resource() -> ResourceId {
        ResourceId::new(Uuid::new_v4()).unwrap()
    }

    /// The Catalog's semantic oracle with one Arbor table per binding; a repeated
    /// resource is an alias.
    fn catalog(bindings: &[(NamespaceParent, &str, ResourceId)]) -> ProjectJournalModel {
        let mut model = ProjectJournalModel::new(actor(), 0);
        let main = model.branch(&ProjectBranchName::main()).unwrap().clone();
        let mut resources = Vec::<ResourceCas>::new();
        for (_, _, id) in bindings {
            if resources.iter().all(|created| created.resource_id != *id) {
                resources.push(ResourceCas {
                    resource_id: *id,
                    expected: None,
                    replacement: Some(VersionId::new(Uuid::now_v7()).unwrap()),
                    create: Some(ResourceDescriptor {
                        namespace_root: None,
                        namespace_managed: false,
                        resource_id: *id,
                        service: ResourceServiceId::new("arbor").unwrap(),
                        kind: ResourceKind::Table.name(),
                    }),
                });
            }
        }
        let seed = PublishProjectRequest {
            branch_id: main.branch_id,
            base_commit_id: main.head_commit_id,
            expected_head_commit_id: Some(main.head_commit_id),
            transaction_id: ProjectTransactionId::new(Uuid::now_v7()).unwrap(),
            idempotency_key: IdempotencyKey::new("seed").unwrap(),
            message: None,
            resources,
            namespace: bindings
                .iter()
                .map(|(parent, name, id)| NamespaceCas {
                    parent: parent.clone(),
                    name: segment(name),
                    expected: None,
                    replacement: Some(*id),
                })
                .collect(),
        };
        assert!(matches!(
            model.publish(actor(), 0, seed).unwrap(),
            PublishProjectResponse::Published { .. }
        ));
        model
    }

    fn commit(catalog: &ProjectJournalModel) -> Commit {
        let client = Client::builder()
            .api_url("https://api.example.test")
            .bearer_token("test-token")
            .build()
            .unwrap();
        let main = catalog.branch(&ProjectBranchName::main()).unwrap();
        BranchHead::new(
            client.project("project-demo").unwrap(),
            WireCommitId(main.head_commit_id.as_uuid()),
            WireBranchId(main.branch_id.as_uuid()),
        )
        .commit()
    }

    /// The table that `path` resolves to at the commit's head.
    fn table_at(
        commit: &Commit,
        catalog: &ProjectJournalModel,
        path: &str,
        id: ResourceId,
    ) -> Table {
        let head = commit.head();
        let version = catalog
            .state(ProjectCommitId::new(head.commit_id.as_uuid()).unwrap())
            .unwrap()
            .resource(&id)
            .unwrap()
            .version_id;
        let mut table = head
            .project
            .client
            .table_by_id(id.as_uuid(), version.as_uuid());
        table.base = Some(CatalogBase {
            project_id: head.project.id.clone(),
            commit_id: head.commit_id,
            branch_id: Some(head.branch_id()),
            version_id: version.as_uuid(),
            address: Some(path.parse().unwrap()),
        });
        table.address = Some(path.parse().unwrap());
        table
    }

    /// A new table's first version, written but not yet in the Catalog.
    fn written(commit: &Commit) -> Table {
        commit
            .head()
            .project
            .client
            .table_by_id(Uuid::new_v4(), Uuid::now_v7())
    }

    /// The next version written from `table`.
    fn next_version(table: &Table) -> Table {
        table
            .successor(TableWriteResult {
                resource_id: table.identity().resource_id,
                version_id: Uuid::now_v7(),
            })
            .unwrap()
    }

    fn retirement(table: &Table) -> ResourceCas {
        ResourceCas {
            resource_id: ResourceId::new(table.identity().resource_id).unwrap(),
            expected: Some(VersionId::new(table.identity().version_id).unwrap()),
            replacement: None,
            create: None,
        }
    }

    fn publication(commit: Commit) -> PublishProjectRequest {
        legacy_publication(&commit.freeze().unwrap().request.publication)
    }

    fn published_head(response: PublishProjectResponse) -> ProjectCommitId {
        match response {
            PublishProjectResponse::Published { commit_id, .. } => commit_id,
            conflict => panic!("unexpected conflict: {conflict:?}"),
        }
    }

    #[test]
    fn drops_retire_resources_and_unbind_root_and_prefixed_paths() {
        let (flat_id, nested_id, kept_id) = (resource(), resource(), resource());
        let mut catalog = catalog(&[
            (
                NamespaceParent::ProjectRoot,
                "datasets.warp_sim_bottles.episodes",
                flat_id,
            ),
            (prefix(&["datasets", "x"]), "episodes", nested_id),
            (NamespaceParent::ProjectRoot, "kept", kept_id),
        ]);
        let mut commit = commit(&catalog);
        let flat = table_at(
            &commit,
            &catalog,
            "datasets.warp_sim_bottles.episodes",
            flat_id,
        );
        let nested = table_at(&commit, &catalog, "datasets/x/episodes", nested_id);
        commit.drop_table(&flat).unwrap();
        commit.drop_table(&nested).unwrap();
        let publication = publication(commit);
        assert_eq!(
            publication.resources,
            [retirement(&flat), retirement(&nested)]
        );
        assert_eq!(
            publication.namespace,
            [
                NamespaceCas {
                    parent: NamespaceParent::ProjectRoot,
                    name: segment("datasets.warp_sim_bottles.episodes"),
                    expected: Some(flat_id),
                    replacement: None,
                },
                NamespaceCas {
                    parent: prefix(&["datasets", "x"]),
                    name: segment("episodes"),
                    expected: Some(nested_id),
                    replacement: None,
                },
            ]
        );
        let head = published_head(catalog.publish(actor(), 1, publication).unwrap());
        let state = catalog.state(head).unwrap();
        assert_eq!(state.namespace().values().collect::<Vec<_>>(), [&kept_id]);
        assert_eq!(state.resources().keys().collect::<Vec<_>>(), [&kept_id]);
    }

    #[test]
    fn dropping_and_creating_one_name_publishes_one_rebinding_in_either_order() {
        for drop_first in [true, false] {
            let old_id = resource();
            let mut catalog = catalog(&[(NamespaceParent::ProjectRoot, "episodes", old_id)]);
            let mut commit = commit(&catalog);
            let old = table_at(&commit, &catalog, "episodes", old_id);
            if drop_first {
                commit.drop_table(&old).unwrap();
            }
            let new = written(&commit);
            commit.create_table("episodes", &new).unwrap();
            if !drop_first {
                commit.drop_table(&old).unwrap();
            }
            assert!(commit.unclaimed_root_name("episodes").is_err());
            let new_id = ResourceId::new(new.identity().resource_id).unwrap();
            let publication = publication(commit);
            assert_eq!(
                publication.namespace,
                [NamespaceCas {
                    parent: NamespaceParent::ProjectRoot,
                    name: segment("episodes"),
                    expected: Some(old_id),
                    replacement: Some(new_id),
                }]
            );
            let head = published_head(catalog.publish(actor(), 1, publication).unwrap());
            let state = catalog.state(head).unwrap();
            assert_eq!(
                state.binding(&NamespaceParent::ProjectRoot, &segment("episodes")),
                Some(&new_id)
            );
            assert_eq!(state.resources().keys().collect::<Vec<_>>(), [&new_id]);
        }
    }

    #[test]
    fn drops_that_leave_the_table_bound_or_miss_its_binding_conflict() {
        let (aliased, anchor, anchored) = (resource(), resource(), resource());
        let mut catalog = catalog(&[
            (NamespaceParent::ProjectRoot, "episodes", aliased),
            (NamespaceParent::ProjectRoot, "episodes_alias", aliased),
            (NamespaceParent::ProjectRoot, "datasets", anchor),
            (
                NamespaceParent::Resource {
                    resource_id: anchor,
                },
                "frames",
                anchored,
            ),
        ]);
        let mut alias = commit(&catalog);
        alias
            .drop_table(&table_at(&alias, &catalog, "episodes", aliased))
            .unwrap();
        match catalog.publish(actor(), 1, publication(alias)).unwrap() {
            PublishProjectResponse::Conflict { constraints, .. } => assert!(matches!(
                constraints.as_slice(),
                [NamespaceConstraintConflict::BindingTargetsRetiredResource { .. }]
            )),
            published => panic!("an aliased drop must conflict: {published:?}"),
        }
        let mut beneath = commit(&catalog);
        beneath
            .drop_table(&table_at(&beneath, &catalog, "datasets/frames", anchored))
            .unwrap();
        match catalog.publish(actor(), 1, publication(beneath)).unwrap() {
            PublishProjectResponse::Conflict { namespace, .. } => assert_eq!(namespace.len(), 1),
            published => panic!("a resource-anchored drop must conflict: {published:?}"),
        }
    }

    #[test]
    fn a_commit_changes_each_table_once_from_its_head() {
        let id = resource();
        let catalog = catalog(&[(NamespaceParent::ProjectRoot, "episodes", id)]);
        let mut commit = commit(&catalog);
        let dropped = table_at(&commit, &catalog, "episodes", id);
        commit.drop_table(&dropped).unwrap();
        for error in [
            commit.drop_table(&dropped).unwrap_err(),
            commit.update_table(&next_version(&dropped)).unwrap_err(),
        ] {
            assert!(error.to_string().contains("once per commit"), "{error}");
        }

        let created = written(&commit);
        commit.create_table("created", &created).unwrap();
        let other = self::commit(&catalog);
        let foreign = table_at(&other, &catalog, "episodes", id);
        for table in [created, foreign] {
            let error = commit.drop_table(&table).unwrap_err();
            assert!(error.to_string().contains("branch head"), "{error}");
        }
    }

    #[test]
    fn updates_swap_the_head_version_for_the_newest_written_version() {
        let id = resource();
        let mut catalog = catalog(&[(NamespaceParent::ProjectRoot, "episodes", id)]);
        let mut commit = commit(&catalog);
        let opened = table_at(&commit, &catalog, "episodes", id);
        let newest = next_version(&next_version(&opened));
        commit.update_table(&newest).unwrap();
        let publication = publication(commit);
        assert_eq!(
            publication.resources,
            [ResourceCas {
                resource_id: id,
                expected: Some(VersionId::new(opened.identity().version_id).unwrap()),
                replacement: Some(VersionId::new(newest.identity().version_id).unwrap()),
                create: None,
            }]
        );
        let head = published_head(catalog.publish(actor(), 1, publication).unwrap());
        assert_eq!(
            catalog
                .state(head)
                .unwrap()
                .resource(&id)
                .unwrap()
                .version_id
                .as_uuid(),
            newest.identity().version_id
        );
    }

    fn binding(
        parent: NamespaceParent,
        name: &str,
        change: [Option<ResourceId>; 2],
    ) -> NamespaceCas {
        NamespaceCas {
            parent,
            name: segment(name),
            expected: change[0],
            replacement: change[1],
        }
    }

    #[test]
    fn renames_move_root_and_prefixed_bindings_and_keep_resources() {
        let (root, episodes, stream) = (resource(), resource(), resource());
        let mut catalog = catalog(&[
            (NamespaceParent::ProjectRoot, "recordings", root),
            (prefix(&["datasets", "x"]), "episodes", episodes),
            (prefix(&["datasets", "x", "messages"]), "front", stream),
        ]);
        let mut commit = commit(&catalog);
        for (path, id, name) in [
            ("recordings", root, "datasets.x.recordings"),
            ("datasets/x/episodes", episodes, "datasets.x.episodes"),
            (
                "datasets/x/messages/front",
                stream,
                "datasets.x.messages.front",
            ),
        ] {
            let table = table_at(&commit, &catalog, path, id);
            commit.rename_table(&table, name).unwrap();
        }
        let base = commit.head().commit_id;
        let publication = publication(commit);
        assert!(publication.resources.is_empty());
        assert_eq!(
            publication.namespace,
            [
                binding(
                    NamespaceParent::ProjectRoot,
                    "recordings",
                    [Some(root), None]
                ),
                binding(
                    NamespaceParent::ProjectRoot,
                    "datasets.x.recordings",
                    [None, Some(root)]
                ),
                binding(
                    prefix(&["datasets", "x"]),
                    "episodes",
                    [Some(episodes), None]
                ),
                binding(
                    NamespaceParent::ProjectRoot,
                    "datasets.x.episodes",
                    [None, Some(episodes)]
                ),
                binding(
                    prefix(&["datasets", "x", "messages"]),
                    "front",
                    [Some(stream), None]
                ),
                binding(
                    NamespaceParent::ProjectRoot,
                    "datasets.x.messages.front",
                    [None, Some(stream)]
                ),
            ]
        );
        let head = published_head(catalog.publish(actor(), 1, publication).unwrap());
        let state = catalog.state(head).unwrap();
        assert_eq!(
            state.resources(),
            catalog
                .state(ProjectCommitId::new(base.as_uuid()).unwrap())
                .unwrap()
                .resources()
        );
        assert_eq!(
            state
                .namespace()
                .iter()
                .map(|(key, id)| (&key.parent, key.name.as_str(), *id))
                .collect::<Vec<_>>(),
            [
                (
                    &NamespaceParent::ProjectRoot,
                    "datasets.x.episodes",
                    episodes
                ),
                (
                    &NamespaceParent::ProjectRoot,
                    "datasets.x.messages.front",
                    stream
                ),
                (&NamespaceParent::ProjectRoot, "datasets.x.recordings", root),
            ]
        );
    }

    #[test]
    fn renames_compose_with_writes_drops_and_creates_in_one_commit() {
        let (a, b, c) = (resource(), resource(), resource());
        let mut catalog = catalog(&[
            (NamespaceParent::ProjectRoot, "a", a),
            (NamespaceParent::ProjectRoot, "b", b),
            (NamespaceParent::ProjectRoot, "c", c),
        ]);
        let mut commit = commit(&catalog);
        let table_a = table_at(&commit, &catalog, "a", a);
        commit.rename_table(&table_a, "b").unwrap();
        // A renamed table can still be updated.
        let updated = next_version(&table_a);
        commit.update_table(&updated).unwrap();
        let table_b = table_at(&commit, &catalog, "b", b);
        commit.rename_table(&table_b, "c").unwrap();
        commit
            .drop_table(&table_at(&commit, &catalog, "c", c))
            .unwrap();
        let created = written(&commit);
        commit.create_table("a", &created).unwrap();
        let created = ResourceId::new(created.identity().resource_id).unwrap();
        let publication = publication(commit);
        assert_eq!(
            publication.namespace,
            [
                binding(NamespaceParent::ProjectRoot, "a", [Some(a), Some(created)]),
                binding(NamespaceParent::ProjectRoot, "b", [Some(b), Some(a)]),
                binding(NamespaceParent::ProjectRoot, "c", [Some(c), Some(b)]),
            ]
        );
        let head = published_head(catalog.publish(actor(), 1, publication).unwrap());
        let state = catalog.state(head).unwrap();
        for (name, id) in [("a", created), ("b", a), ("c", b)] {
            assert_eq!(
                state.binding(&NamespaceParent::ProjectRoot, &segment(name)),
                Some(&id)
            );
        }
        assert_eq!(
            state.resource(&a).unwrap().version_id.as_uuid(),
            updated.identity().version_id
        );
        assert!(state.resource(&c).is_none());
    }

    #[test]
    fn renames_reject_or_conflict_without_staging_partial_changes() {
        let (renamed, taken, dropped) = (resource(), resource(), resource());
        let mut catalog = catalog(&[
            (NamespaceParent::ProjectRoot, "renamed", renamed),
            (NamespaceParent::ProjectRoot, "taken", taken),
            (NamespaceParent::ProjectRoot, "dropped", dropped),
        ]);
        let mut commit = commit(&catalog);
        let renamed = table_at(&commit, &catalog, "renamed", renamed);
        let dropped = table_at(&commit, &catalog, "dropped", dropped);
        commit.rename_table(&renamed, "taken").unwrap();
        commit.drop_table(&dropped).unwrap();
        let staged = commit.namespace.clone();
        let created = written(&commit);
        commit.create_table("created", &created).unwrap();
        for (error, message) in [
            (
                commit.rename_table(&renamed, "elsewhere"),
                "binding `renamed` is already staged",
            ),
            (
                commit.rename_table(&dropped, "elsewhere"),
                "binding `dropped` is already staged",
            ),
            (
                commit.drop_table(&renamed),
                "binding `renamed` is already staged",
            ),
            (commit.rename_table(&created, "elsewhere"), "branch head"),
            (
                commit.rename_table(&dropped, "created"),
                "name `created` is already staged",
            ),
            (
                commit.rename_table(&dropped, "dropped"),
                "already has this name",
            ),
        ] {
            let error = error.unwrap_err().to_string();
            assert!(error.contains(message), "{error}");
        }
        assert_eq!(&commit.namespace[..staged.len()], staged.as_slice());
        assert_eq!(commit.namespace.len(), staged.len() + 1);
        let publication = publication(commit);
        match catalog.publish(actor(), 1, publication).unwrap() {
            PublishProjectResponse::Conflict { namespace, .. } => assert!(matches!(
                namespace.as_slice(),
                [NamespaceCasConflict {
                    expected: None,
                    current: Some(_),
                    ..
                }]
            )),
            published => panic!("renaming onto a bound name must conflict: {published:?}"),
        }
    }

    #[test]
    fn renaming_through_an_alias_publishes_only_changed_bindings() {
        let id = resource();
        let mut catalog = catalog(&[
            (NamespaceParent::ProjectRoot, "a", id),
            (NamespaceParent::ProjectRoot, "b", id),
        ]);
        let mut commit = commit(&catalog);
        let through_a = table_at(&commit, &catalog, "a", id);
        let through_b = table_at(&commit, &catalog, "b", id);
        commit.rename_table(&through_a, "b").unwrap();
        commit.rename_table(&through_b, "c").unwrap();
        let publication = publication(commit);
        assert_eq!(
            publication.namespace,
            [
                binding(NamespaceParent::ProjectRoot, "a", [Some(id), None]),
                binding(NamespaceParent::ProjectRoot, "c", [None, Some(id)]),
            ]
        );
        let head = published_head(catalog.publish(actor(), 1, publication).unwrap());
        let state = catalog.state(head).unwrap();
        for name in ["b", "c"] {
            assert_eq!(
                state.binding(&NamespaceParent::ProjectRoot, &segment(name)),
                Some(&id)
            );
        }
    }
}