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,
}),
}
}
}
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(),
}
}
pub fn head(&self) -> &BranchHead {
&self.head
}
pub fn operation_id(&self) -> Uuid {
self.operation_id
}
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(())
}
#[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"))
}
#[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)
}
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(())
}
#[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(())
}
#[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",
)
}
#[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,
)
}
#[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(())
}
#[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(())
}
#[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,
};
self.stage_binding(unbinding)?;
self.stage_binding(binding)
}
#[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(())
}
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)
}
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(())
}
pub fn freeze(self) -> Result<FrozenCommit, Error> {
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,
}),
})
}
pub async fn publish(self) -> Result<BranchHead, PublishError> {
self.freeze()?.publish().await
}
}
#[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),
})
}
pub fn operation_id(&self) -> Uuid {
self.request.operation.operation_id.0
}
pub fn head(&self) -> &BranchHead {
&self.head
}
pub fn save(&self) -> SavedCommit {
SavedCommit(PublishOperationRequest::clone(&self.request))
}
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)),
}
}
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(),
}
}
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(transparent)]
pub struct SavedCommit(PublishOperationRequest);
#[derive(Debug, thiserror::Error)]
pub enum PublishError {
#[error("the branch moved to {current} before publication")]
BranchMoved {
current: ProjectCommitId,
},
#[error("a staged change conflicts with the branch head")]
Conflict(Box<PublicationConflict>),
#[error("publication outcome unknown for {}: {message}", .commit.operation_id())]
Unknown {
commit: Box<FrozenCommit>,
message: String,
},
#[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,
}
}
}
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()
}
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()
}
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
}
fn written(commit: &Commit) -> Table {
commit
.head()
.project
.client
.table_by_id(Uuid::new_v4(), Uuid::now_v7())
}
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();
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)
);
}
}
}