use crate::branch_head::{
create_child_branch_head_from_parent, read_branch_head, BranchHead, BranchHeadError,
ChildBranchHeadRequest, ChildBranchSourceExpectation,
};
use crate::durability;
use crate::file_io::{self as fs, File, OpenOptions, TryLockError};
use crate::immutable_object::{
BranchReclamationEntry, BranchReclamationInventory, ImmutableObjectError, ImmutableObjectStore,
ObjectReference, ReclamationReport,
};
use crate::ownership::{DatabaseDirectoryLease, DatabaseDirectoryLeaseError};
use hawdb_core::Uuid;
use hawdb_integrity::crc32c;
use std::collections::{BTreeMap, BTreeSet};
use std::fmt::{self, Display, Formatter};
use std::io::{self, Read, Write};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
const MAGIC: &[u8; 8] = b"HBCATV2\0";
const VERSION: u16 = 2;
const MAX_CATALOG_BYTES: usize = 16 * 1024 * 1024;
const MAX_BRANCHES: u32 = 100_000;
const MAX_NAME_BYTES: usize = 128;
const MAX_OWNER_BYTES: usize = 256;
const MAX_REQUEST_KEY_BYTES: usize = 256;
const DIGEST_BYTES: usize = 32;
const METADATA_LOCK_FILE: &str = "metadata.hawdb.lock";
static CANDIDATE_SEQUENCE: AtomicU64 = AtomicU64::new(0);
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct BranchId(Uuid);
impl BranchId {
pub fn new(value: Uuid) -> Result<Self, CatalogError> {
if value.is_nil() {
return Err(CatalogError::InvalidIdentity("branch UUID must not be nil"));
}
Ok(Self(value))
}
pub fn parse(value: &str) -> Result<Self, CatalogError> {
let uuid = value
.parse::<Uuid>()
.map_err(|_| CatalogError::InvalidIdentity("branch UUID is not valid"))?;
Self::new(uuid)
}
pub const fn as_uuid(self) -> Uuid {
self.0
}
}
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct BranchName(String);
impl BranchName {
pub fn main() -> Self {
Self("main".to_string())
}
pub fn generated_agent(id: BranchId) -> Self {
Self(format!("agent/{}", id.as_uuid()))
}
pub fn new(value: impl Into<String>) -> Result<Self, CatalogError> {
let value = value.into();
validate_name(&value)?;
Ok(Self(value))
}
pub fn as_str(&self) -> &str {
&self.0
}
fn from_encoded(value: String) -> Result<Self, CatalogError> {
validate_catalog_name(&value)?;
Ok(Self(value))
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum BranchSelector<'a> {
Id(BranchId),
Name(&'a BranchName),
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum BranchState {
Creating,
Ready,
Deleting,
Deleted,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CreateOutcome {
Pending,
Succeeded,
Aborted,
Abandoned,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BranchRecord {
pub id: BranchId,
pub name: BranchName,
pub parent_id: Option<BranchId>,
pub source_commit_epoch: u64,
pub base_root_digest: Option<[u8; DIGEST_BYTES]>,
pub metadata_revision: u64,
pub state: BranchState,
pub owner: Option<String>,
pub create_request_key: String,
pub request_fingerprint: [u8; DIGEST_BYTES],
pub create_outcome: CreateOutcome,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Catalog {
pub project_id: BranchId,
pub revision: u64,
pub branches: Vec<BranchRecord>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CreateRequest {
pub id: BranchId,
pub name: BranchName,
pub parent_id: BranchId,
pub source_commit_epoch: u64,
pub base_root_digest: [u8; DIGEST_BYTES],
pub owner: Option<String>,
pub request_key: String,
pub request_fingerprint: [u8; DIGEST_BYTES],
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum CatalogTransitionError {
Conflict(&'static str),
InvalidState(&'static str),
MissingBranch,
MissingParent,
StaleRevision { expected: u64, actual: u64 },
Overflow(&'static str),
Validation(CatalogError),
}
impl Display for CatalogTransitionError {
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
match self {
Self::Conflict(message) => write!(formatter, "branch catalog conflict: {message}"),
Self::InvalidState(message) => {
write!(formatter, "invalid branch catalog state: {message}")
}
Self::MissingBranch => formatter.write_str("branch catalog branch is missing"),
Self::MissingParent => formatter.write_str("branch catalog parent is missing"),
Self::StaleRevision { expected, actual } => write!(
formatter,
"branch catalog revision mismatch: expected {expected}, found {actual}"
),
Self::Overflow(field) => write!(formatter, "branch catalog {field} overflow"),
Self::Validation(error) => error.fmt(formatter),
}
}
}
impl std::error::Error for CatalogTransitionError {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
match self {
Self::Validation(error) => Some(error),
_ => None,
}
}
}
impl Catalog {
pub fn bootstrap(project_id: BranchId, main_id: BranchId) -> Result<Self, CatalogError> {
if project_id == main_id {
return Err(CatalogError::InvalidIdentity(
"project UUID and branch UUID must be distinct",
));
}
let catalog = Self {
project_id,
revision: 1,
branches: vec![BranchRecord {
id: main_id,
name: BranchName::main(),
parent_id: None,
source_commit_epoch: 0,
base_root_digest: None,
metadata_revision: 1,
state: BranchState::Ready,
owner: None,
create_request_key: "bootstrap".to_string(),
request_fingerprint: [0; DIGEST_BYTES],
create_outcome: CreateOutcome::Succeeded,
}],
};
catalog.validate()?;
Ok(catalog)
}
pub fn validate(&self) -> Result<(), CatalogError> {
if self.branches.len() > MAX_BRANCHES as usize {
return Err(CatalogError::Limit("branch count"));
}
let mut branch_ids = self
.branches
.iter()
.map(|branch| branch.id)
.collect::<Vec<_>>();
branch_ids.sort_unstable();
for pair in branch_ids.windows(2) {
if pair[0] == pair[1] {
return Err(CatalogError::Duplicate("branch UUID"));
}
}
let mut names = BTreeSet::new();
let mut main_count = 0;
for branch in &self.branches {
if branch.state != BranchState::Deleted && !names.insert(branch.name.clone()) {
return Err(CatalogError::Duplicate("branch name"));
}
if branch.id == self.project_id {
return Err(CatalogError::InvalidIdentity(
"project UUID and branch UUID must be distinct",
));
}
validate_catalog_name(branch.name.as_str())?;
if branch.name.as_str() == "main" {
main_count += 1;
if branch.parent_id.is_some() {
return Err(CatalogError::InvalidState(
"main branch cannot have a parent",
));
}
if matches!(branch.state, BranchState::Deleting | BranchState::Deleted) {
return Err(CatalogError::InvalidState("main branch cannot be deleted"));
}
}
if branch.parent_id == Some(branch.id) {
return Err(CatalogError::InvalidState(
"branch cannot be its own parent",
));
}
if let Some(parent_id) = branch.parent_id
&& branch_ids.binary_search(&parent_id).is_err()
{
return Err(CatalogError::InvalidIdentity(
"branch parent UUID is not present in the catalog",
));
}
if branch.name.as_str().starts_with("agent/")
&& branch.name.as_str() != format!("agent/{}", branch.id.as_uuid())
{
return Err(CatalogError::InvalidName);
}
validate_bounded_string(
&branch.create_request_key,
MAX_REQUEST_KEY_BYTES,
"create request key",
)?;
if let Some(owner) = &branch.owner {
validate_bounded_string(owner, MAX_OWNER_BYTES, "owner")?;
}
if branch.create_outcome == CreateOutcome::Abandoned
&& branch.state != BranchState::Deleted
{
return Err(CatalogError::InvalidState(
"abandoned create must be deleted",
));
}
if branch.state == BranchState::Deleted
&& branch.create_outcome == CreateOutcome::Pending
{
return Err(CatalogError::InvalidState(
"deleted branch cannot have a pending create outcome",
));
}
}
if main_count > 1 {
return Err(CatalogError::Duplicate("main branch"));
}
Ok(())
}
pub fn reserve_create(
&mut self,
request: CreateRequest,
) -> Result<BranchId, CatalogTransitionError> {
if let Some(existing) = self
.branches
.iter()
.find(|branch| branch.create_request_key == request.request_key)
{
if existing.request_fingerprint == request.request_fingerprint {
return Ok(existing.id);
}
return Err(CatalogTransitionError::Conflict(
"create request key has a different fingerprint",
));
}
let parent = self
.branches
.iter()
.find(|branch| branch.id == request.parent_id)
.ok_or(CatalogTransitionError::MissingParent)?;
if parent.state != BranchState::Ready {
return Err(CatalogTransitionError::InvalidState(
"create parent is not ready",
));
}
if request.source_commit_epoch < parent.source_commit_epoch {
return Err(CatalogTransitionError::Conflict(
"create source revision is stale",
));
}
if self.branches.iter().any(|branch| branch.id == request.id) {
return Err(CatalogTransitionError::Conflict(
"branch UUID is already used",
));
}
if self
.branches
.iter()
.any(|branch| branch.name == request.name && branch.state != BranchState::Deleted)
{
return Err(CatalogTransitionError::Conflict("branch name is reserved"));
}
let catalog_revision = self.next_revision()?;
let mut candidate = self.clone();
candidate.revision = catalog_revision;
candidate.branches.push(BranchRecord {
id: request.id,
name: request.name,
parent_id: Some(request.parent_id),
source_commit_epoch: request.source_commit_epoch,
base_root_digest: Some(request.base_root_digest),
metadata_revision: 1,
state: BranchState::Creating,
owner: request.owner,
create_request_key: request.request_key,
request_fingerprint: request.request_fingerprint,
create_outcome: CreateOutcome::Pending,
});
candidate
.validate()
.map_err(CatalogTransitionError::Validation)?;
*self = candidate;
Ok(request.id)
}
pub fn complete_create(
&mut self,
id: BranchId,
expected_metadata_revision: u64,
) -> Result<(), CatalogTransitionError> {
self.transition_create(id, expected_metadata_revision, CreateOutcome::Succeeded)
}
pub fn abort_create(
&mut self,
id: BranchId,
expected_metadata_revision: u64,
) -> Result<(), CatalogTransitionError> {
self.transition_create(id, expected_metadata_revision, CreateOutcome::Aborted)
}
pub(crate) fn abandon_create(
&mut self,
id: BranchId,
expected_metadata_revision: u64,
) -> Result<(), CatalogTransitionError> {
self.transition_create(id, expected_metadata_revision, CreateOutcome::Abandoned)
}
fn transition_create(
&mut self,
id: BranchId,
expected_metadata_revision: u64,
outcome: CreateOutcome,
) -> Result<(), CatalogTransitionError> {
let index = self
.branches
.iter()
.position(|branch| branch.id == id)
.ok_or(CatalogTransitionError::MissingBranch)?;
let branch = &self.branches[index];
if branch.metadata_revision != expected_metadata_revision {
return Err(CatalogTransitionError::StaleRevision {
expected: expected_metadata_revision,
actual: branch.metadata_revision,
});
}
if branch.state != BranchState::Creating || branch.create_outcome != CreateOutcome::Pending
{
return Err(CatalogTransitionError::InvalidState(
"create transition requires a pending reservation",
));
}
let mut candidate = self.clone();
let catalog_revision = candidate.next_revision()?;
let metadata_revision = candidate.next_metadata_revision(index)?;
candidate.revision = catalog_revision;
candidate.branches[index].metadata_revision = metadata_revision;
candidate.branches[index].create_outcome = outcome;
candidate.branches[index].state = match outcome {
CreateOutcome::Succeeded => BranchState::Ready,
CreateOutcome::Aborted | CreateOutcome::Abandoned => BranchState::Deleted,
CreateOutcome::Pending => unreachable!("pending is rejected above"),
};
candidate
.validate()
.map_err(CatalogTransitionError::Validation)?;
*self = candidate;
Ok(())
}
pub fn rename(
&mut self,
id: BranchId,
expected_metadata_revision: u64,
new_name: BranchName,
) -> Result<(), CatalogTransitionError> {
if new_name.as_str() == "main" || new_name.as_str().starts_with("agent/") {
return Err(CatalogTransitionError::Conflict("name is reserved"));
}
let index = self.index_at_revision(id, expected_metadata_revision)?;
if self.branches[index].state != BranchState::Ready {
return Err(CatalogTransitionError::InvalidState(
"only a ready branch can be renamed",
));
}
if self.branches.iter().any(|branch| {
branch.id != id && branch.name == new_name && branch.state != BranchState::Deleted
}) {
return Err(CatalogTransitionError::Conflict("branch name is reserved"));
}
let mut candidate = self.clone();
candidate.revision = candidate.next_revision()?;
candidate.branches[index].metadata_revision = candidate.next_metadata_revision(index)?;
candidate.branches[index].name = new_name;
candidate
.validate()
.map_err(CatalogTransitionError::Validation)?;
*self = candidate;
Ok(())
}
pub fn begin_delete(
&mut self,
id: BranchId,
expected_metadata_revision: u64,
) -> Result<(), CatalogTransitionError> {
let index = self.index_at_revision(id, expected_metadata_revision)?;
let branch = &self.branches[index];
if branch.name.as_str() == "main" {
return Err(CatalogTransitionError::InvalidState(
"main branch is protected",
));
}
if branch.state != BranchState::Ready {
return Err(CatalogTransitionError::InvalidState(
"branch is not deletable",
));
}
self.transition_state(index, BranchState::Deleting)
}
pub fn finish_delete(
&mut self,
id: BranchId,
expected_metadata_revision: u64,
) -> Result<(), CatalogTransitionError> {
let index = self.index_at_revision(id, expected_metadata_revision)?;
if self.branches[index].state != BranchState::Deleting {
return Err(CatalogTransitionError::InvalidState(
"delete finalization requires a deleting branch",
));
}
self.transition_state(index, BranchState::Deleted)
}
fn index_at_revision(
&self,
id: BranchId,
expected_metadata_revision: u64,
) -> Result<usize, CatalogTransitionError> {
let index = self
.branches
.iter()
.position(|branch| branch.id == id)
.ok_or(CatalogTransitionError::MissingBranch)?;
let actual = self.branches[index].metadata_revision;
if actual != expected_metadata_revision {
return Err(CatalogTransitionError::StaleRevision {
expected: expected_metadata_revision,
actual,
});
}
Ok(index)
}
fn transition_state(
&mut self,
index: usize,
state: BranchState,
) -> Result<(), CatalogTransitionError> {
let mut candidate = self.clone();
candidate.revision = candidate.next_revision()?;
candidate.branches[index].metadata_revision = candidate.next_metadata_revision(index)?;
candidate.branches[index].state = state;
candidate
.validate()
.map_err(CatalogTransitionError::Validation)?;
*self = candidate;
Ok(())
}
fn next_revision(&self) -> Result<u64, CatalogTransitionError> {
self.revision
.checked_add(1)
.ok_or(CatalogTransitionError::Overflow("catalog revision"))
}
fn next_metadata_revision(&self, index: usize) -> Result<u64, CatalogTransitionError> {
self.branches[index]
.metadata_revision
.checked_add(1)
.ok_or(CatalogTransitionError::Overflow("branch metadata revision"))
}
pub fn encode(&self) -> Result<Vec<u8>, CatalogError> {
self.validate()?;
let mut branches = self.branches.clone();
branches.sort_by_key(|branch| branch.id);
let mut bytes = Vec::with_capacity(128);
bytes.extend_from_slice(MAGIC);
put_u16(&mut bytes, VERSION);
bytes.extend_from_slice(self.project_id.as_uuid().as_bytes());
put_u64(&mut bytes, self.revision);
put_u32(&mut bytes, branches.len() as u32);
for branch in branches {
encode_branch(&mut bytes, &branch)?;
}
let checksum = crc32c(&bytes).get();
put_u32(&mut bytes, checksum);
if bytes.len() > MAX_CATALOG_BYTES {
return Err(CatalogError::Limit("catalog bytes"));
}
Ok(bytes)
}
pub fn decode(encoded: &[u8]) -> Result<Self, CatalogError> {
if encoded.len() > MAX_CATALOG_BYTES {
return Err(CatalogError::Limit("catalog bytes"));
}
if encoded.len() < MAGIC.len() + 2 + 16 + 8 + 4 + 4 {
return Err(CatalogError::Truncated);
}
let checksum_offset = encoded.len() - 4;
let expected = u32::from_le_bytes(
encoded[checksum_offset..]
.try_into()
.map_err(|_| CatalogError::Truncated)?,
);
let actual = crc32c(&encoded[..checksum_offset]).get();
if actual != expected {
return Err(CatalogError::Checksum);
}
let mut reader = Reader::new(&encoded[..checksum_offset]);
if reader.take(MAGIC.len())? != MAGIC {
return Err(CatalogError::Version);
}
if reader.u16()? != VERSION {
return Err(CatalogError::Version);
}
let project_id = BranchId::new(Uuid::from_bytes(reader.array()?))?;
let revision = reader.u64()?;
let count = reader.u32()?;
if count > MAX_BRANCHES {
return Err(CatalogError::Limit("branch count"));
}
let mut branches = Vec::with_capacity(count as usize);
for _ in 0..count {
branches.push(decode_branch(&mut reader)?);
}
if !reader.is_empty() {
return Err(CatalogError::TrailingBytes);
}
let catalog = Self {
project_id,
revision,
branches,
};
catalog.validate()?;
Ok(catalog)
}
}
pub fn read_catalog(path: &Path) -> io::Result<Catalog> {
let length = fs::metadata(path)?.len();
if length > MAX_CATALOG_BYTES as u64 {
return Err(invalid_data("branch catalog exceeds its byte limit"));
}
let file = fs::File::open(path)?;
let mut encoded = Vec::with_capacity(length as usize);
file.take((MAX_CATALOG_BYTES + 1) as u64)
.read_to_end(&mut encoded)?;
Catalog::decode(&encoded).map_err(|error| invalid_data(error.to_string()))
}
pub fn write_catalog(path: &Path, catalog: &Catalog) -> io::Result<()> {
let parent = path
.parent()
.filter(|parent| !parent.as_os_str().is_empty())
.ok_or_else(|| invalid_data("branch catalog destination has no parent"))?;
let _metadata_lock = CatalogMetadataLease::acquire_blocking(parent)?;
write_catalog_locked(path, catalog)
}
pub fn initialize_catalog_file(
path: &Path,
project_id: BranchId,
main_id: BranchId,
) -> Result<BranchRecord, CatalogFileTransitionError> {
let parent = path
.parent()
.filter(|parent| !parent.as_os_str().is_empty())
.ok_or_else(|| {
CatalogFileTransitionError::Io(invalid_data("branch catalog destination has no parent"))
})?;
fs::create_dir_all(parent).map_err(CatalogFileTransitionError::Io)?;
let _metadata_lock =
CatalogMetadataLease::acquire_blocking(parent).map_err(CatalogFileTransitionError::Io)?;
match read_catalog(path) {
Ok(catalog) => main_record(&catalog, project_id, main_id),
Err(error) if error.kind() == io::ErrorKind::NotFound => {
let catalog = Catalog::bootstrap(project_id, main_id)
.map_err(|error| CatalogFileTransitionError::Io(invalid_data(error.to_string())))?;
let main = catalog.branches[0].clone();
write_catalog_locked(path, &catalog).map_err(CatalogFileTransitionError::Io)?;
Ok(main)
}
Err(error) => Err(CatalogFileTransitionError::Io(error)),
}
}
pub fn bind_main_head_file(
path: &Path,
project_id: BranchId,
main_id: BranchId,
sealed_root_digest: [u8; DIGEST_BYTES],
source_commit_epoch: u64,
) -> Result<BranchRecord, CatalogFileTransitionError> {
mutate_catalog_file(path, |catalog| {
let index = main_record_index(catalog, project_id, main_id)?;
let branch = &catalog.branches[index];
if let Some(digest) = branch.base_root_digest {
if digest != sealed_root_digest || branch.source_commit_epoch != source_commit_epoch {
return Err(CatalogTransitionError::Conflict(
"main catalog binding does not match its sealed head",
));
}
return Ok(CatalogMutation {
value: branch.clone(),
changed: false,
});
}
if branch.state != BranchState::Ready {
return Err(CatalogTransitionError::InvalidState(
"main branch is not ready for head binding",
));
}
let next_catalog_revision = catalog.next_revision()?;
let next_metadata_revision = catalog.next_metadata_revision(index)?;
let branch = &mut catalog.branches[index];
branch.base_root_digest = Some(sealed_root_digest);
branch.source_commit_epoch = source_commit_epoch;
branch.metadata_revision = next_metadata_revision;
catalog.revision = next_catalog_revision;
catalog
.validate()
.map_err(CatalogTransitionError::Validation)?;
Ok(CatalogMutation {
value: catalog.branches[index].clone(),
changed: true,
})
})
}
pub(crate) fn write_catalog_locked(path: &Path, catalog: &Catalog) -> io::Result<()> {
let encoded = catalog
.encode()
.map_err(|error| invalid_data(error.to_string()))?;
let parent = path
.parent()
.filter(|parent| !parent.as_os_str().is_empty())
.ok_or_else(|| invalid_data("branch catalog destination has no parent"))?;
let sequence = CANDIDATE_SEQUENCE.fetch_add(1, Ordering::Relaxed);
let candidate = parent.join(format!(
".{}.candidate-{}-{}",
path.file_name()
.and_then(|name| name.to_str())
.unwrap_or("catalog"),
std::process::id(),
sequence
));
let result = (|| {
{
let mut file = OpenOptions::new()
.create_new(true)
.write(true)
.open(&candidate)?;
file.write_all(&encoded)?;
file.sync_all()?;
}
durability::durable_replace_file(&candidate, path)
})();
if result.is_err() {
let _ = fs::remove_file(&candidate);
}
result
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct CreateReservation {
pub id: BranchId,
pub metadata_revision: u64,
pub replayed: bool,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct DeleteRequest {
pub id: BranchId,
pub expected_metadata_revision: u64,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct DeleteReservation {
pub id: BranchId,
pub metadata_revision: u64,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum DeleteBeginOutcome {
Deleting(DeleteReservation),
Deleted(BranchRecord),
}
#[derive(Debug)]
pub enum CatalogFileTransitionError {
Io(io::Error),
Transition(CatalogTransitionError),
}
impl Display for CatalogFileTransitionError {
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
match self {
Self::Io(error) => Display::fmt(error, formatter),
Self::Transition(error) => Display::fmt(error, formatter),
}
}
}
impl std::error::Error for CatalogFileTransitionError {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
match self {
Self::Io(error) => Some(error),
Self::Transition(error) => Some(error),
}
}
}
struct CatalogMutation<T> {
value: T,
changed: bool,
}
fn mutate_catalog_file<T>(
path: &Path,
transition: impl FnOnce(&mut Catalog) -> Result<CatalogMutation<T>, CatalogTransitionError>,
) -> Result<T, CatalogFileTransitionError> {
mutate_catalog_file_with_io(path, |catalog| {
transition(catalog).map_err(CatalogFileTransitionError::Transition)
})
}
fn mutate_catalog_file_with_io<T>(
path: &Path,
transition: impl FnOnce(&mut Catalog) -> Result<CatalogMutation<T>, CatalogFileTransitionError>,
) -> Result<T, CatalogFileTransitionError> {
let parent = path
.parent()
.filter(|parent| !parent.as_os_str().is_empty())
.ok_or_else(|| {
CatalogFileTransitionError::Io(invalid_data("branch catalog destination has no parent"))
})?;
let _metadata_lock =
CatalogMetadataLease::acquire_blocking(parent).map_err(CatalogFileTransitionError::Io)?;
let mut catalog = read_catalog(path).map_err(CatalogFileTransitionError::Io)?;
let mutation = transition(&mut catalog)?;
if mutation.changed {
write_catalog_locked(path, &catalog).map_err(CatalogFileTransitionError::Io)?;
}
Ok(mutation.value)
}
fn main_record(
catalog: &Catalog,
project_id: BranchId,
main_id: BranchId,
) -> Result<BranchRecord, CatalogFileTransitionError> {
let index = main_record_index(catalog, project_id, main_id)
.map_err(CatalogFileTransitionError::Transition)?;
Ok(catalog.branches[index].clone())
}
fn main_record_index(
catalog: &Catalog,
project_id: BranchId,
main_id: BranchId,
) -> Result<usize, CatalogTransitionError> {
if catalog.project_id != project_id {
return Err(CatalogTransitionError::Conflict(
"catalog project identity does not match root initialization",
));
}
let index = catalog
.branches
.iter()
.position(|branch| branch.id == main_id)
.ok_or(CatalogTransitionError::MissingBranch)?;
if catalog.branches[index].name.as_str() != "main" {
return Err(CatalogTransitionError::Conflict(
"root branch identity does not name main",
));
}
Ok(index)
}
#[cfg(test)]
fn reserve_create_file(
path: &Path,
request: CreateRequest,
) -> Result<CreateReservation, CatalogFileTransitionError> {
reserve_create_file_with_source(path, request, |_| Ok(()))
}
fn reserve_create_file_with_source(
path: &Path,
request: CreateRequest,
validate_source: impl FnOnce(&Catalog) -> Result<(), CatalogFileTransitionError>,
) -> Result<CreateReservation, CatalogFileTransitionError> {
mutate_catalog_file_with_io(path, move |catalog| {
let replayed = catalog
.branches
.iter()
.any(|branch| branch.create_request_key == request.request_key);
if !replayed {
validate_source(catalog)?;
}
let revision = catalog.revision;
let id = catalog
.reserve_create(request)
.map_err(CatalogFileTransitionError::Transition)?;
let metadata_revision = catalog
.branches
.iter()
.find(|branch| branch.id == id)
.map(|branch| branch.metadata_revision)
.ok_or(CatalogFileTransitionError::Transition(
CatalogTransitionError::MissingBranch,
))?;
Ok(CatalogMutation {
value: CreateReservation {
id,
metadata_revision,
replayed,
},
changed: catalog.revision != revision,
})
})
}
pub fn complete_create_file(
path: &Path,
reservation: CreateReservation,
) -> Result<(), CatalogFileTransitionError> {
mutate_catalog_file(path, |catalog| {
catalog.complete_create(reservation.id, reservation.metadata_revision)?;
Ok(CatalogMutation {
value: (),
changed: true,
})
})
}
pub fn abort_create_file(
path: &Path,
reservation: CreateReservation,
) -> Result<(), CatalogFileTransitionError> {
mutate_catalog_file(path, |catalog| {
catalog.abort_create(reservation.id, reservation.metadata_revision)?;
Ok(CatalogMutation {
value: (),
changed: true,
})
})
}
pub fn begin_delete_file(
path: &Path,
request: DeleteRequest,
) -> Result<DeleteBeginOutcome, CatalogFileTransitionError> {
mutate_catalog_file(path, |catalog| {
let branch = catalog
.branches
.iter()
.find(|branch| branch.id == request.id)
.cloned()
.ok_or(CatalogTransitionError::MissingBranch)?;
match branch.state {
BranchState::Ready => {
catalog.begin_delete(request.id, request.expected_metadata_revision)?;
let metadata_revision = catalog
.branches
.iter()
.find(|branch| branch.id == request.id)
.map(|branch| branch.metadata_revision)
.ok_or(CatalogTransitionError::MissingBranch)?;
Ok(CatalogMutation {
value: DeleteBeginOutcome::Deleting(DeleteReservation {
id: request.id,
metadata_revision,
}),
changed: true,
})
}
BranchState::Deleting => Ok(CatalogMutation {
value: DeleteBeginOutcome::Deleting(DeleteReservation {
id: branch.id,
metadata_revision: branch.metadata_revision,
}),
changed: false,
}),
BranchState::Deleted => Ok(CatalogMutation {
value: DeleteBeginOutcome::Deleted(branch),
changed: false,
}),
BranchState::Creating => Err(CatalogTransitionError::InvalidState(
"branch creation is not complete",
)),
}
})
}
pub fn finish_delete_file(
path: &Path,
reservation: DeleteReservation,
) -> Result<BranchRecord, CatalogFileTransitionError> {
mutate_catalog_file(path, |catalog| {
let branch = catalog
.branches
.iter()
.find(|branch| branch.id == reservation.id)
.cloned()
.ok_or(CatalogTransitionError::MissingBranch)?;
match branch.state {
BranchState::Deleting => {
catalog.finish_delete(reservation.id, reservation.metadata_revision)?;
let branch = catalog
.branches
.iter()
.find(|branch| branch.id == reservation.id)
.cloned()
.ok_or(CatalogTransitionError::MissingBranch)?;
Ok(CatalogMutation {
value: branch,
changed: true,
})
}
BranchState::Deleted => Ok(CatalogMutation {
value: branch,
changed: false,
}),
_ => Err(CatalogTransitionError::InvalidState(
"delete finalization requires a deleting branch",
)),
}
})
}
#[derive(Debug)]
pub enum BranchCreateError {
Catalog(CatalogFileTransitionError),
Head(BranchHeadError),
Lease(DatabaseDirectoryLeaseError),
InconsistentRequest(&'static str),
}
impl Display for BranchCreateError {
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
match self {
Self::Catalog(error) => Display::fmt(error, formatter),
Self::Head(error) => Display::fmt(error, formatter),
Self::Lease(error) => Display::fmt(error, formatter),
Self::InconsistentRequest(message) => formatter.write_str(message),
}
}
}
impl std::error::Error for BranchCreateError {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
match self {
Self::Catalog(error) => Some(error),
Self::Head(error) => Some(error),
Self::Lease(error) => Some(error),
Self::InconsistentRequest(_) => None,
}
}
}
pub fn create_branch_from_parent(
catalog_path: &Path,
parent_head_path: &Path,
child_head_request: ChildBranchHeadRequest,
expected_parent: ChildBranchSourceExpectation,
max_active_wal_bytes: u64,
request: CreateRequest,
) -> Result<BranchCreateResult, BranchCreateError> {
if request.id.as_uuid().as_bytes() != &child_head_request.branch_id {
return Err(BranchCreateError::InconsistentRequest(
"catalog and child head branch IDs differ",
));
}
if request.base_root_digest != *child_head_request.sealed_root.sha256.as_bytes() {
return Err(BranchCreateError::InconsistentRequest(
"catalog and child head sealed-root digests differ",
));
}
if request.parent_id.as_uuid().as_bytes() != &expected_parent.branch_id {
return Err(BranchCreateError::InconsistentRequest(
"catalog parent and selected parent head differ",
));
}
if request.source_commit_epoch != expected_parent.logical_commit_epoch
|| request.source_commit_epoch != child_head_request.logical_commit_epoch
|| child_head_request.sealed_root != expected_parent.sealed_root
{
return Err(BranchCreateError::InconsistentRequest(
"catalog source revision and selected parent head differ",
));
}
let child_head_path = child_head_request.head_path.clone();
let child_directory = child_head_path
.parent()
.filter(|parent| !parent.as_os_str().is_empty())
.unwrap_or_else(|| Path::new("."));
fs::create_dir_all(child_directory).map_err(|source| {
BranchCreateError::Head(BranchHeadError::Io {
operation: "create child branch directory",
source,
})
})?;
let lease =
DatabaseDirectoryLease::acquire(child_directory).map_err(BranchCreateError::Lease)?;
let reservation = reserve_create_file_with_source(catalog_path, request, |catalog| {
let parent = crate::branch_head::read_branch_head(parent_head_path)
.map_err(|error| CatalogFileTransitionError::Io(invalid_data(error.to_string())))?;
if parent.project_id != *catalog.project_id.as_uuid().as_bytes()
|| parent.branch_id != expected_parent.branch_id
|| parent.physical_generation != expected_parent.physical_generation
|| parent.logical_commit_epoch != expected_parent.logical_commit_epoch
|| parent.sealed_root != expected_parent.sealed_root
{
return Err(CatalogFileTransitionError::Transition(
CatalogTransitionError::Conflict("create source revision is stale"),
));
}
Ok(())
})
.map_err(BranchCreateError::Catalog)?;
if reservation.replayed {
if reservation.id.as_uuid().as_bytes() != &child_head_request.branch_id {
return Err(BranchCreateError::InconsistentRequest(
"replayed branch identity does not match the requested child paths",
));
}
let catalog = read_catalog(catalog_path)
.map_err(|error| BranchCreateError::Catalog(CatalogFileTransitionError::Io(error)))?;
let branch = catalog
.branches
.iter()
.find(|branch| branch.id == reservation.id)
.ok_or(BranchCreateError::InconsistentRequest(
"replayed branch is missing",
))?;
match (branch.state, branch.create_outcome) {
(BranchState::Creating, CreateOutcome::Pending) => {
if recover_create_file(
catalog_path,
reservation.id,
&child_head_path,
&child_head_request.wal_path,
max_active_wal_bytes,
)? == CreateRecoveryOutcome::Aborted
{
return Err(BranchCreateError::InconsistentRequest(
"replayed branch creation was aborted",
));
}
}
(BranchState::Ready, CreateOutcome::Succeeded) => {}
_ => {
return Err(BranchCreateError::InconsistentRequest(
"replayed branch is not ready",
))
}
}
let head = crate::branch_head::read_branch_head(&child_head_path)
.map_err(BranchCreateError::Head)?;
if head.project_id != *catalog.project_id.as_uuid().as_bytes()
|| head.branch_id != *reservation.id.as_uuid().as_bytes()
{
return Err(BranchCreateError::InconsistentRequest(
"replayed branch head identity mismatch",
));
}
let wal = crate::branch_head::active_wal_identity_from_file(
&child_head_request.wal_path,
head.active_wal.generation,
head.active_wal.replay_start_lsn,
max_active_wal_bytes,
)
.map_err(BranchCreateError::Head)?;
if wal != head.active_wal {
return Err(BranchCreateError::Head(BranchHeadError::InvalidWalIdentity));
}
return Ok(BranchCreateResult { head, lease });
}
match create_child_branch_head_from_parent(
parent_head_path,
&child_head_path,
child_head_request,
expected_parent,
max_active_wal_bytes,
) {
Ok(head) => {
complete_create_file(catalog_path, reservation).map_err(BranchCreateError::Catalog)?;
Ok(BranchCreateResult { head, lease })
}
Err(error @ BranchHeadError::CandidatePublicationUncertain { .. }) => {
Err(BranchCreateError::Head(error))
}
Err(error) => {
abort_create_file(catalog_path, reservation).map_err(BranchCreateError::Catalog)?;
Err(BranchCreateError::Head(error))
}
}
}
#[derive(Debug)]
pub struct BranchCreateResult {
pub head: BranchHead,
pub lease: DatabaseDirectoryLease,
}
#[derive(Debug, Clone)]
pub struct BranchReclamationPath {
pub id: BranchId,
pub directory: PathBuf,
pub head_path: PathBuf,
}
#[derive(Debug)]
pub enum BranchReclamationError {
Catalog(io::Error),
MissingPath(BranchId),
Head(BranchHeadError),
Lease(io::Error),
Objects(ImmutableObjectError),
PendingCreate {
branch_id: BranchId,
catalog_revision: u64,
metadata_revision: u64,
source: Box<BranchReclamationError>,
},
}
impl Display for BranchReclamationError {
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
match self {
Self::Catalog(error) => {
write!(formatter, "read branch catalog for reclamation: {error}")
}
Self::MissingPath(id) => write!(
formatter,
"missing reclamation path for branch {}",
id.as_uuid()
),
Self::Head(error) => write!(formatter, "read branch head for reclamation: {error}"),
Self::Lease(error) => {
write!(formatter, "inspect branch lease for reclamation: {error}")
}
Self::Objects(error) => write!(formatter, "reclaim branch objects: {error}"),
Self::PendingCreate {
branch_id, source, ..
} => write!(
formatter,
"pending branch {} blocks reclamation: {source}",
branch_id.as_uuid()
),
}
}
}
impl std::error::Error for BranchReclamationError {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
match self {
Self::Catalog(error) | Self::Lease(error) => Some(error),
Self::Head(error) => Some(error),
Self::Objects(error) => Some(error),
Self::PendingCreate { source, .. } => Some(source.as_ref()),
Self::MissingPath(_) => None,
}
}
}
impl From<BranchReclamationError> for crate::error::HawDBError {
fn from(error: BranchReclamationError) -> Self {
match error {
BranchReclamationError::PendingCreate {
branch_id,
catalog_revision,
metadata_revision,
source,
} => {
let cause = Self::from(*source);
if matches!(cause, Self::FileDescriptors(_)) {
return cause;
}
Self::BranchReclamationBlocked {
branch_id: branch_id.as_uuid(),
catalog_revision,
metadata_revision,
cause: Box::new(cause),
}
}
error => Self::from_storage_error(error),
}
}
}
pub(crate) fn pending_reclamation_error(
catalog: &Catalog,
record: &BranchRecord,
source: BranchReclamationError,
) -> BranchReclamationError {
if record.state == BranchState::Creating && record.create_outcome == CreateOutcome::Pending {
BranchReclamationError::PendingCreate {
branch_id: record.id,
catalog_revision: catalog.revision,
metadata_revision: record.metadata_revision,
source: Box::new(source),
}
} else {
source
}
}
pub(crate) fn reclamation_object_error(
catalog: &Catalog,
branches: &[BranchReclamationEntry],
error: ImmutableObjectError,
) -> BranchReclamationError {
if let ImmutableObjectError::ReclamationRoot { root, .. } = &error
&& let Some(record) = catalog
.branches
.iter()
.zip(branches)
.find_map(|(record, entry)| {
(record.state == BranchState::Creating
&& record.create_outcome == CreateOutcome::Pending
&& (entry.sealed_root == Some(*root)
|| record.base_root_digest == Some(*root.sha256.as_bytes())))
.then_some(record)
})
{
return pending_reclamation_error(catalog, record, BranchReclamationError::Objects(error));
}
BranchReclamationError::Objects(error)
}
pub fn reclaim_catalog_branches(
catalog_path: &Path,
object_store: &mut ImmutableObjectStore,
objects: &[ObjectReference],
paths: &[BranchReclamationPath],
) -> Result<ReclamationReport, BranchReclamationError> {
let project_directory = catalog_path
.parent()
.filter(|parent| !parent.as_os_str().is_empty())
.ok_or_else(|| BranchReclamationError::Catalog(invalid_data("catalog has no parent")))?;
let _metadata_lease = CatalogMetadataLease::acquire_blocking(project_directory)
.map_err(BranchReclamationError::Catalog)?;
let catalog = read_catalog(catalog_path).map_err(BranchReclamationError::Catalog)?;
let proofs = crate::branch_abandonment::reclamation_evidence(
project_directory,
&catalog,
Default::default(),
)
.map_err(|error| BranchReclamationError::Catalog(io::Error::other(error)))?;
let branches = reclamation_entries(&catalog, paths, None)?;
let inventory = BranchReclamationInventory {
objects: objects.to_vec(),
branches,
};
object_store
.reclaim_branches_with_evidence(&inventory, &[], &proofs)
.map_err(|error| reclamation_object_error(&catalog, &inventory.branches, error))
}
pub(crate) fn reclamation_entries(
catalog: &Catalog,
paths: &[BranchReclamationPath],
exclusive_owner: Option<(&DatabaseDirectoryLease, &BranchHead)>,
) -> Result<Vec<BranchReclamationEntry>, BranchReclamationError> {
let indexed_paths: BTreeMap<_, _> = paths.iter().map(|path| (path.id, path)).collect();
if indexed_paths.len() != paths.len() {
return Err(BranchReclamationError::Catalog(invalid_data(
"duplicate reclamation path identity",
)));
}
let mut branches = Vec::with_capacity(catalog.branches.len());
for record in &catalog.branches {
let path = indexed_paths
.get(&record.id)
.ok_or(BranchReclamationError::MissingPath(record.id))?;
branches.push(
reclamation_entry(catalog, record, path, exclusive_owner)
.map_err(|source| pending_reclamation_error(catalog, record, source))?,
);
}
Ok(branches)
}
fn reclamation_entry(
catalog: &Catalog,
record: &BranchRecord,
path: &BranchReclamationPath,
exclusive_owner: Option<(&DatabaseDirectoryLease, &BranchHead)>,
) -> Result<BranchReclamationEntry, BranchReclamationError> {
let owned = if let Some((lease, head)) = exclusive_owner {
record.state == BranchState::Ready
&& head.branch_id == *record.id.as_uuid().as_bytes()
&& lease
.owns_directory(&path.directory)
.map_err(BranchReclamationError::Lease)?
} else {
false
};
let active_lease = if owned
|| (matches!(record.state, BranchState::Deleted)
&& !fs::try_exists(&path.directory).map_err(BranchReclamationError::Lease)?)
{
false
} else {
match DatabaseDirectoryLease::acquire(&path.directory) {
Ok(lease) => {
drop(lease);
false
}
Err(DatabaseDirectoryLeaseError::AlreadyOpen) => true,
Err(DatabaseDirectoryLeaseError::Canonicalize(error))
| Err(DatabaseDirectoryLeaseError::OpenLockFile(error))
| Err(DatabaseDirectoryLeaseError::Lock(error)) => {
return Err(BranchReclamationError::Lease(error));
}
}
};
let sealed_root = if active_lease || matches!(record.state, BranchState::Deleted) {
None
} else {
let head = read_branch_head(&path.head_path).map_err(BranchReclamationError::Head)?;
if head.project_id != *catalog.project_id.as_uuid().as_bytes()
|| head.branch_id != *record.id.as_uuid().as_bytes()
|| (owned && exclusive_owner.is_some_and(|(_, expected)| *expected != head))
{
return Err(BranchReclamationError::Catalog(invalid_data(
"reclamation head does not match its catalog or admitted owner",
)));
}
Some(head.sealed_root)
};
Ok(BranchReclamationEntry {
state: record.state,
sealed_root,
directory: path.directory.clone(),
active_lease,
})
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CreateRecoveryOutcome {
Completed,
Aborted,
}
pub fn recover_create_file(
catalog_path: &Path,
branch_id: BranchId,
child_head_path: &Path,
child_wal_path: &Path,
max_active_wal_bytes: u64,
) -> Result<CreateRecoveryOutcome, BranchCreateError> {
recover_create_file_inner(
catalog_path,
branch_id,
child_head_path,
Some(child_wal_path),
max_active_wal_bytes,
|_, _| Ok(()),
)
}
pub fn recover_create_from_head_file(
catalog_path: &Path,
branch_id: BranchId,
child_head_path: &Path,
max_active_wal_bytes: u64,
) -> Result<CreateRecoveryOutcome, BranchCreateError> {
recover_create_with_validation(
catalog_path,
branch_id,
child_head_path,
max_active_wal_bytes,
|_, _| Ok(()),
)
}
pub(crate) fn recover_create_with_validation(
catalog_path: &Path,
branch_id: BranchId,
child_head_path: &Path,
max_active_wal_bytes: u64,
validate_dependencies: impl FnMut(Option<&BranchHead>, &Path) -> Result<(), BranchCreateError>,
) -> Result<CreateRecoveryOutcome, BranchCreateError> {
let directory = child_head_path
.parent()
.ok_or(BranchCreateError::InconsistentRequest(
"child head has no branch directory",
))?;
{
let project = catalog_path
.parent()
.ok_or(BranchCreateError::InconsistentRequest(
"catalog has no directory",
))?;
let _metadata = CatalogMetadataLease::acquire_blocking(project)
.map_err(|error| BranchCreateError::Catalog(CatalogFileTransitionError::Io(error)))?;
let catalog = read_catalog(catalog_path)
.map_err(|error| BranchCreateError::Catalog(CatalogFileTransitionError::Io(error)))?;
let branch = catalog
.branches
.iter()
.find(|branch| branch.id == branch_id)
.ok_or(BranchCreateError::InconsistentRequest(
"pending branch is missing",
))?;
match branch.create_outcome {
CreateOutcome::Succeeded => return Ok(CreateRecoveryOutcome::Completed),
CreateOutcome::Aborted | CreateOutcome::Abandoned => {
return Ok(CreateRecoveryOutcome::Aborted)
}
CreateOutcome::Pending => {}
}
fs::create_dir_all(directory).map_err(|source| {
BranchCreateError::Head(BranchHeadError::Io {
operation: "create pending child lease directory",
source,
})
})?;
}
let _lease = DatabaseDirectoryLease::acquire(directory).map_err(BranchCreateError::Lease)?;
recover_create_file_inner(
catalog_path,
branch_id,
child_head_path,
None,
max_active_wal_bytes,
validate_dependencies,
)
}
fn recover_create_file_inner(
catalog_path: &Path,
branch_id: BranchId,
child_head_path: &Path,
child_wal_path: Option<&Path>,
max_active_wal_bytes: u64,
mut validate_dependencies: impl FnMut(Option<&BranchHead>, &Path) -> Result<(), BranchCreateError>,
) -> Result<CreateRecoveryOutcome, BranchCreateError> {
let catalog = read_catalog(catalog_path)
.map_err(|error| BranchCreateError::Catalog(CatalogFileTransitionError::Io(error)))?;
let branch = catalog
.branches
.iter()
.find(|branch| branch.id == branch_id)
.ok_or(BranchCreateError::Catalog(
CatalogFileTransitionError::Transition(CatalogTransitionError::MissingBranch),
))?;
match branch.create_outcome {
CreateOutcome::Succeeded => return Ok(CreateRecoveryOutcome::Completed),
CreateOutcome::Aborted | CreateOutcome::Abandoned => {
return Ok(CreateRecoveryOutcome::Aborted)
}
CreateOutcome::Pending => {}
}
if branch.state != BranchState::Creating {
return Err(BranchCreateError::Catalog(
CatalogFileTransitionError::Transition(CatalogTransitionError::InvalidState(
"recovery requires a pending child create",
)),
));
}
if crate::branch_abandonment::has_abandonment(catalog_path, branch_id)
.map_err(|error| BranchCreateError::Catalog(CatalogFileTransitionError::Io(error)))?
{
return Err(BranchCreateError::InconsistentRequest(
"pending create has an operator abandonment intent",
));
}
let reservation = CreateReservation {
id: branch_id,
metadata_revision: branch.metadata_revision,
replayed: false,
};
validate_dependencies(None, child_head_path)?;
let head = match crate::branch_head::read_branch_head(child_head_path) {
Ok(head) => head,
Err(BranchHeadError::Io { source, .. }) if source.kind() == io::ErrorKind::NotFound => {
abort_create_file(catalog_path, reservation).map_err(BranchCreateError::Catalog)?;
return Ok(CreateRecoveryOutcome::Aborted);
}
Err(error) => return Err(BranchCreateError::Head(error)),
};
if head.project_id != *catalog.project_id.as_uuid().as_bytes()
|| head.branch_id != *branch_id.as_uuid().as_bytes()
|| branch.base_root_digest != Some(*head.sealed_root.sha256.as_bytes())
|| branch.source_commit_epoch != head.logical_commit_epoch
{
return Err(BranchCreateError::InconsistentRequest(
"pending child metadata does not match its head",
));
}
let derived_wal_path = child_head_path.with_file_name(
crate::artifact_files::wal_generation_file(head.active_wal.generation),
);
let child_wal_path = child_wal_path.unwrap_or(&derived_wal_path);
match fs::metadata(child_wal_path) {
Ok(_) => {}
Err(source) if source.kind() == io::ErrorKind::NotFound => {
abort_create_file(catalog_path, reservation).map_err(BranchCreateError::Catalog)?;
return Ok(CreateRecoveryOutcome::Aborted);
}
Err(source) => {
return Err(BranchCreateError::Head(BranchHeadError::Io {
operation: "read pending child WAL metadata",
source,
}));
}
}
validate_dependencies(Some(&head), child_wal_path)?;
let wal = crate::branch_head::active_wal_identity_from_file(
child_wal_path,
head.active_wal.generation,
head.active_wal.replay_start_lsn,
max_active_wal_bytes,
)
.map_err(BranchCreateError::Head)?;
if wal != head.active_wal {
return Err(BranchCreateError::Head(BranchHeadError::InvalidWalIdentity));
}
for path in [child_wal_path, child_head_path] {
OpenOptions::new()
.read(true)
.write(true)
.open(path)
.and_then(|file| file.sync_all())
.and_then(|_| {
crate::durability::sync_directory_ancestors(
path.parent().unwrap_or_else(|| Path::new(".")),
)
})
.map_err(|source| {
BranchCreateError::Head(BranchHeadError::CandidatePublicationUncertain { source })
})?;
}
complete_create_file(catalog_path, reservation).map_err(BranchCreateError::Catalog)?;
Ok(CreateRecoveryOutcome::Completed)
}
#[derive(Debug)]
pub struct CatalogMetadataLease {
file: File,
}
impl CatalogMetadataLease {
pub fn acquire(project_directory: &Path) -> io::Result<Self> {
let file = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(project_directory.join(METADATA_LOCK_FILE))?;
match file.try_lock() {
Ok(()) => Ok(Self { file }),
Err(TryLockError::WouldBlock) => Err(io::Error::new(
io::ErrorKind::WouldBlock,
"branch catalog metadata lock is held",
)),
Err(TryLockError::Error(error)) => Err(error),
}
}
pub fn acquire_blocking(project_directory: &Path) -> io::Result<Self> {
let file = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(project_directory.join(METADATA_LOCK_FILE))?;
file.lock()?;
Ok(Self { file })
}
}
impl Drop for CatalogMetadataLease {
fn drop(&mut self) {
let _ = self.file.unlock();
}
}
fn invalid_data(message: impl Into<String>) -> io::Error {
io::Error::new(io::ErrorKind::InvalidData, message.into())
}
fn validate_name(value: &str) -> Result<(), CatalogError> {
validate_bounded_string(value, MAX_NAME_BYTES, "branch name")?;
if value == "main" || value.starts_with("agent/") {
return Err(CatalogError::ReservedName);
}
validate_catalog_name(value)
}
fn validate_catalog_name(value: &str) -> Result<(), CatalogError> {
validate_bounded_string(value, MAX_NAME_BYTES, "branch name")?;
if value.starts_with('/') || value.ends_with('/') || value.contains('\\') {
return Err(CatalogError::InvalidName);
}
for component in value.split('/') {
if component.is_empty() || component == "." || component == ".." {
return Err(CatalogError::InvalidName);
}
if !component
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-' | b'.'))
{
return Err(CatalogError::InvalidName);
}
if component.bytes().any(|byte| byte.is_ascii_whitespace()) {
return Err(CatalogError::InvalidName);
}
}
Ok(())
}
fn validate_bounded_string(
value: &str,
maximum: usize,
field: &'static str,
) -> Result<(), CatalogError> {
if value.is_empty() || value.len() > maximum || !value.is_ascii() {
return Err(CatalogError::Limit(field));
}
if value.bytes().any(|byte| byte.is_ascii_control()) {
return Err(CatalogError::InvalidName);
}
Ok(())
}
fn encode_branch(output: &mut Vec<u8>, branch: &BranchRecord) -> Result<(), CatalogError> {
output.extend_from_slice(branch.id.as_uuid().as_bytes());
put_string(output, branch.name.as_str(), MAX_NAME_BYTES, "branch name")?;
put_optional_uuid(output, branch.parent_id);
put_u64(output, branch.source_commit_epoch);
put_optional_bytes(output, branch.base_root_digest);
put_u64(output, branch.metadata_revision);
output.push(state_byte(branch.state));
put_optional_string(output, branch.owner.as_deref(), MAX_OWNER_BYTES, "owner")?;
put_string(
output,
&branch.create_request_key,
MAX_REQUEST_KEY_BYTES,
"create request key",
)?;
output.extend_from_slice(&branch.request_fingerprint);
output.push(outcome_byte(branch.create_outcome));
Ok(())
}
fn decode_branch(reader: &mut Reader<'_>) -> Result<BranchRecord, CatalogError> {
let id = BranchId::new(Uuid::from_bytes(reader.array()?))?;
let name = BranchName::from_encoded(reader.string(MAX_NAME_BYTES, "branch name")?)?;
let parent_id = reader.optional_uuid()?;
let source_commit_epoch = reader.u64()?;
let base_root_digest = reader.optional_array()?;
let metadata_revision = reader.u64()?;
let state = parse_state(reader.byte()?)?;
let owner = reader.optional_string(MAX_OWNER_BYTES, "owner")?;
let create_request_key = reader.string(MAX_REQUEST_KEY_BYTES, "create request key")?;
let request_fingerprint = reader.array()?;
let create_outcome = parse_outcome(reader.byte()?)?;
Ok(BranchRecord {
id,
name,
parent_id,
source_commit_epoch,
base_root_digest,
metadata_revision,
state,
owner,
create_request_key,
request_fingerprint,
create_outcome,
})
}
fn state_byte(value: BranchState) -> u8 {
match value {
BranchState::Creating => 0,
BranchState::Ready => 1,
BranchState::Deleting => 3,
BranchState::Deleted => 4,
}
}
fn parse_state(value: u8) -> Result<BranchState, CatalogError> {
match value {
0 => Ok(BranchState::Creating),
1 => Ok(BranchState::Ready),
3 => Ok(BranchState::Deleting),
4 => Ok(BranchState::Deleted),
_ => Err(CatalogError::InvalidEnum("branch state")),
}
}
fn outcome_byte(value: CreateOutcome) -> u8 {
match value {
CreateOutcome::Pending => 0,
CreateOutcome::Succeeded => 1,
CreateOutcome::Aborted => 2,
CreateOutcome::Abandoned => 3,
}
}
fn parse_outcome(value: u8) -> Result<CreateOutcome, CatalogError> {
match value {
0 => Ok(CreateOutcome::Pending),
1 => Ok(CreateOutcome::Succeeded),
2 => Ok(CreateOutcome::Aborted),
3 => Ok(CreateOutcome::Abandoned),
_ => Err(CatalogError::InvalidEnum("create outcome")),
}
}
fn put_u16(output: &mut Vec<u8>, value: u16) {
output.extend_from_slice(&value.to_le_bytes());
}
fn put_u32(output: &mut Vec<u8>, value: u32) {
output.extend_from_slice(&value.to_le_bytes());
}
fn put_u64(output: &mut Vec<u8>, value: u64) {
output.extend_from_slice(&value.to_le_bytes());
}
fn put_string(
output: &mut Vec<u8>,
value: &str,
maximum: usize,
field: &'static str,
) -> Result<(), CatalogError> {
validate_bounded_string(value, maximum, field)?;
let length = u16::try_from(value.len()).map_err(|_| CatalogError::Limit(field))?;
put_u16(output, length);
output.extend_from_slice(value.as_bytes());
Ok(())
}
fn put_optional_string(
output: &mut Vec<u8>,
value: Option<&str>,
maximum: usize,
field: &'static str,
) -> Result<(), CatalogError> {
match value {
Some(value) => {
output.push(1);
put_string(output, value, maximum, field)?;
}
None => output.push(0),
}
Ok(())
}
fn put_optional_uuid(output: &mut Vec<u8>, value: Option<BranchId>) {
match value {
Some(value) => {
output.push(1);
output.extend_from_slice(value.as_uuid().as_bytes());
}
None => output.push(0),
}
}
fn put_optional_bytes(output: &mut Vec<u8>, value: Option<[u8; DIGEST_BYTES]>) {
match value {
Some(value) => {
output.push(1);
output.extend_from_slice(&value);
}
None => output.push(0),
}
}
struct Reader<'a> {
bytes: &'a [u8],
offset: usize,
}
impl<'a> Reader<'a> {
const fn new(bytes: &'a [u8]) -> Self {
Self { bytes, offset: 0 }
}
fn take(&mut self, count: usize) -> Result<&'a [u8], CatalogError> {
let end = self
.offset
.checked_add(count)
.ok_or(CatalogError::Truncated)?;
let value = self
.bytes
.get(self.offset..end)
.ok_or(CatalogError::Truncated)?;
self.offset = end;
Ok(value)
}
fn byte(&mut self) -> Result<u8, CatalogError> {
Ok(*self.take(1)?.first().ok_or(CatalogError::Truncated)?)
}
fn u16(&mut self) -> Result<u16, CatalogError> {
Ok(u16::from_le_bytes(
self.take(2)?
.try_into()
.map_err(|_| CatalogError::Truncated)?,
))
}
fn u32(&mut self) -> Result<u32, CatalogError> {
Ok(u32::from_le_bytes(
self.take(4)?
.try_into()
.map_err(|_| CatalogError::Truncated)?,
))
}
fn u64(&mut self) -> Result<u64, CatalogError> {
Ok(u64::from_le_bytes(
self.take(8)?
.try_into()
.map_err(|_| CatalogError::Truncated)?,
))
}
fn array<const N: usize>(&mut self) -> Result<[u8; N], CatalogError> {
self.take(N)?
.try_into()
.map_err(|_| CatalogError::Truncated)
}
fn optional_uuid(&mut self) -> Result<Option<BranchId>, CatalogError> {
match self.byte()? {
0 => Ok(None),
1 => Ok(Some(BranchId::new(Uuid::from_bytes(self.array()?))?)),
_ => Err(CatalogError::InvalidEnum("optional UUID marker")),
}
}
fn optional_array(&mut self) -> Result<Option<[u8; DIGEST_BYTES]>, CatalogError> {
match self.byte()? {
0 => Ok(None),
1 => Ok(Some(self.array()?)),
_ => Err(CatalogError::InvalidEnum("optional digest marker")),
}
}
fn string(&mut self, maximum: usize, field: &'static str) -> Result<String, CatalogError> {
let length = self.u16()? as usize;
if length == 0 || length > maximum {
return Err(CatalogError::Limit(field));
}
let value = std::str::from_utf8(self.take(length)?)
.map_err(|_| CatalogError::InvalidUtf8(field))?;
validate_bounded_string(value, maximum, field)?;
Ok(value.to_owned())
}
fn optional_string(
&mut self,
maximum: usize,
field: &'static str,
) -> Result<Option<String>, CatalogError> {
match self.byte()? {
0 => Ok(None),
1 => Ok(Some(self.string(maximum, field)?)),
_ => Err(CatalogError::InvalidEnum("optional string marker")),
}
}
fn is_empty(&self) -> bool {
self.offset == self.bytes.len()
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum CatalogError {
Checksum,
Duplicate(&'static str),
InvalidEnum(&'static str),
InvalidIdentity(&'static str),
InvalidName,
InvalidState(&'static str),
InvalidUtf8(&'static str),
Limit(&'static str),
ReservedName,
TrailingBytes,
Truncated,
Version,
}
impl Display for CatalogError {
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
match self {
Self::Checksum => formatter.write_str("branch catalog checksum mismatch"),
Self::Duplicate(field) => write!(formatter, "duplicate branch catalog {field}"),
Self::InvalidEnum(field) => write!(formatter, "invalid branch catalog {field}"),
Self::InvalidIdentity(message) => formatter.write_str(message),
Self::InvalidName => formatter.write_str("invalid branch catalog name"),
Self::InvalidState(message) => formatter.write_str(message),
Self::InvalidUtf8(field) => write!(formatter, "branch catalog {field} is not UTF-8"),
Self::Limit(field) => write!(formatter, "branch catalog {field} exceeds its limit"),
Self::ReservedName => formatter.write_str("branch catalog name is reserved"),
Self::TrailingBytes => formatter.write_str("branch catalog has trailing bytes"),
Self::Truncated => formatter.write_str("branch catalog is truncated"),
Self::Version => formatter.write_str("unsupported branch catalog version"),
}
}
}
impl std::error::Error for CatalogError {}
#[cfg(test)]
mod tests {
use super::*;
use crate::immutable_object::ObjectKind;
use std::path::PathBuf;
use std::sync::atomic::AtomicUsize;
static DIRECTORY_SEQUENCE: AtomicUsize = AtomicUsize::new(0);
fn id(byte: u8) -> BranchId {
BranchId::new(Uuid::from_bytes([byte; 16])).unwrap()
}
fn record(byte: u8, name: &str) -> BranchRecord {
BranchRecord {
id: id(byte),
name: BranchName::new(name).unwrap(),
parent_id: None,
source_commit_epoch: 7,
base_root_digest: Some([byte; DIGEST_BYTES]),
metadata_revision: 3,
state: BranchState::Ready,
owner: Some("agent-1".to_string()),
create_request_key: format!("request-{byte}"),
request_fingerprint: [byte.wrapping_add(1); DIGEST_BYTES],
create_outcome: CreateOutcome::Succeeded,
}
}
fn catalog() -> Catalog {
Catalog {
project_id: id(99),
revision: 11,
branches: vec![record(2, "second"), record(1, "first")],
}
}
#[test]
fn pending_reclamation_keeps_nested_descriptor_rejections_typed() {
use crate::error::HawDBError;
use hawdb_core::error::FileDescriptorError;
let mut catalog = catalog();
let pending = &mut catalog.branches[0];
pending.state = BranchState::Creating;
pending.create_outcome = CreateOutcome::Pending;
let descriptor = FileDescriptorError::BudgetExceeded {
requested: 1,
available: 0,
limit: 32,
};
let source = BranchReclamationError::Objects(ImmutableObjectError::ReclamationRoot {
root: ObjectReference::for_bytes(ObjectKind::SealedRoot, 2, b"pending root"),
source: Box::new(ImmutableObjectError::Io {
operation: "open immutable recovery dependency",
source: io::Error::other(descriptor.clone()),
}),
});
let error = pending_reclamation_error(&catalog, &catalog.branches[0], source);
assert_eq!(
HawDBError::from(error),
HawDBError::FileDescriptors(descriptor)
);
}
fn create_request() -> CreateRequest {
CreateRequest {
id: id(3),
name: BranchName::new("third").unwrap(),
parent_id: id(1),
source_commit_epoch: 7,
base_root_digest: [3; DIGEST_BYTES],
owner: Some("worker-1".to_string()),
request_key: "create-third".to_string(),
request_fingerprint: [9; DIGEST_BYTES],
}
}
fn temporary_catalog_path() -> (PathBuf, PathBuf) {
let directory = std::env::temp_dir().join(format!(
"hawdb-branch-catalog-{}-{}",
std::process::id(),
DIRECTORY_SEQUENCE.fetch_add(1, Ordering::Relaxed)
));
fs::create_dir(&directory).unwrap();
let path = directory.join("catalog.hawdb");
(directory, path)
}
#[test]
fn names_enforce_the_catalog_path_independent_contract() {
for value in [
"", "main", "agent/x", "/x", "x/", "x//y", "x/../y", "x/y\\z", "x y",
] {
assert!(BranchName::new(value).is_err(), "accepted {value:?}");
}
for value in ["Foo", "foo", "agentx", "a/b.c_2"] {
assert!(BranchName::new(value).is_ok(), "rejected {value:?}");
}
assert!(BranchName::new("agent/generated").is_err());
}
#[test]
fn catalog_codec_accepts_reserved_engine_names_only_in_catalog_records() {
let mut main = record(1, "ordinary-main");
main.name = BranchName::from_encoded("main".to_string()).unwrap();
main.parent_id = None;
let mut generated = record(2, "generated");
generated.name =
BranchName::from_encoded(format!("agent/{}", generated.id.as_uuid())).unwrap();
generated.parent_id = Some(main.id);
let catalog = Catalog {
project_id: id(99),
revision: 1,
branches: vec![main, generated],
};
let encoded = catalog.encode().unwrap();
assert_eq!(Catalog::decode(&encoded).unwrap().branches.len(), 2);
}
#[test]
fn codec_is_deterministic_and_round_trips_unsorted_records() {
let catalog = catalog();
let mut reversed = catalog.clone();
reversed.branches.reverse();
let encoded = catalog.encode().unwrap();
assert_eq!(encoded, reversed.encode().unwrap());
let mut expected = catalog.clone();
expected.branches.sort_by_key(|branch| branch.id);
assert_eq!(Catalog::decode(&encoded).unwrap(), expected);
}
#[test]
fn file_publication_syncs_and_reopens_the_canonical_catalog() {
let (directory, path) = temporary_catalog_path();
let catalog = catalog();
write_catalog(&path, &catalog).unwrap();
assert_eq!(read_catalog(&path).unwrap().revision, catalog.revision);
assert!(path.is_file());
assert!(directory.join(METADATA_LOCK_FILE).is_file());
fs::remove_dir_all(directory).unwrap();
}
#[test]
fn parent_validation_accepts_unsorted_lineage_and_rejects_missing_or_duplicate_ids() {
let mut parent = record(2, "parent");
parent.parent_id = Some(id(1));
let mut child = record(3, "child");
child.parent_id = Some(id(2));
let mut catalog = Catalog {
project_id: id(99),
revision: 1,
branches: vec![child, record(1, "root"), parent],
};
let encoded = catalog.encode().unwrap();
for _ in 0..catalog.branches.len() {
catalog.branches.rotate_left(1);
assert_eq!(catalog.encode().unwrap(), encoded);
}
let mut dangling = catalog.clone();
dangling
.branches
.iter_mut()
.find(|branch| branch.id == id(3))
.unwrap()
.parent_id = Some(id(4));
assert!(matches!(
dangling.validate(),
Err(CatalogError::InvalidIdentity(
"branch parent UUID is not present in the catalog"
))
));
catalog.branches.push(catalog.branches[0].clone());
assert_eq!(
catalog.validate(),
Err(CatalogError::Duplicate("branch UUID"))
);
}
#[test]
fn opened_catalog_read_retains_one_complete_publication() {
let (directory, path) = temporary_catalog_path();
let mut catalog = catalog();
write_catalog(&path, &catalog).unwrap();
let old_encoded = catalog.encode().unwrap();
let mut opened = File::open(&path).unwrap();
let mut observed = vec![0; MAGIC.len()];
opened.read_exact(&mut observed).unwrap();
catalog
.rename(id(2), 3, BranchName::new("renamed").unwrap())
.unwrap();
write_catalog(&path, &catalog).unwrap();
opened.read_to_end(&mut observed).unwrap();
assert_eq!(observed, old_encoded);
assert_eq!(Catalog::decode(&observed).unwrap().revision, 11);
let next = read_catalog(&path).unwrap();
assert_eq!(next.revision, 12);
assert_eq!(next.encode().unwrap(), catalog.encode().unwrap());
drop(opened);
fs::remove_dir_all(directory).unwrap();
}
#[test]
fn file_create_reservation_and_completion_are_restart_visible() {
let (directory, path) = temporary_catalog_path();
write_catalog(&path, &catalog()).unwrap();
let reservation = reserve_create_file(&path, create_request()).unwrap();
let pending = read_catalog(&path).unwrap();
assert_eq!(pending.revision, 12);
assert_eq!(pending.branches.len(), 3);
assert_eq!(
pending
.branches
.iter()
.find(|branch| branch.id == reservation.id)
.unwrap()
.state,
BranchState::Creating
);
complete_create_file(&path, reservation).unwrap();
let ready = read_catalog(&path).unwrap();
assert_eq!(
ready
.branches
.iter()
.find(|branch| branch.id == reservation.id)
.unwrap()
.state,
BranchState::Ready
);
fs::remove_dir_all(directory).unwrap();
}
#[test]
fn delete_transition_is_published_resumable_and_idempotent() {
let (directory, path) = temporary_catalog_path();
write_catalog(&path, &catalog()).unwrap();
let request = DeleteRequest {
id: id(1),
expected_metadata_revision: 3,
};
let reservation = match begin_delete_file(&path, request).unwrap() {
DeleteBeginOutcome::Deleting(reservation) => reservation,
DeleteBeginOutcome::Deleted(_) => panic!("ready branch must begin deletion"),
};
assert_eq!(reservation.metadata_revision, 4);
let deleting = read_catalog(&path).unwrap();
let branch = deleting
.branches
.iter()
.find(|branch| branch.id == request.id)
.unwrap();
assert_eq!(branch.state, BranchState::Deleting);
assert_eq!(branch.metadata_revision, reservation.metadata_revision);
assert_eq!(
begin_delete_file(&path, request).unwrap(),
DeleteBeginOutcome::Deleting(reservation)
);
let deleted = finish_delete_file(&path, reservation).unwrap();
assert_eq!(deleted.state, BranchState::Deleted);
assert_eq!(deleted.metadata_revision, 5);
assert_eq!(finish_delete_file(&path, reservation).unwrap(), deleted);
assert_eq!(
begin_delete_file(&path, request).unwrap(),
DeleteBeginOutcome::Deleted(deleted)
);
fs::remove_dir_all(directory).unwrap();
}
#[test]
fn delete_revision_cas_applies_before_transition_and_replays_keep_uuid_identity() {
let (directory, path) = temporary_catalog_path();
let mut initial = catalog();
initial
.rename(id(1), 3, BranchName::new("reusable").unwrap())
.unwrap();
write_catalog(&path, &initial).unwrap();
let before = fs::read(&path).unwrap();
let request = DeleteRequest {
id: id(1),
expected_metadata_revision: 3,
};
assert!(matches!(
begin_delete_file(&path, request),
Err(CatalogFileTransitionError::Transition(
CatalogTransitionError::StaleRevision {
expected: 3,
actual: 4,
}
))
));
assert_eq!(fs::read(&path).unwrap(), before);
let reservation = match begin_delete_file(
&path,
DeleteRequest {
expected_metadata_revision: 4,
..request
},
)
.unwrap()
{
DeleteBeginOutcome::Deleting(reservation) => reservation,
DeleteBeginOutcome::Deleted(_) => panic!("ready branch must begin deletion"),
};
assert!(matches!(
finish_delete_file(
&path,
DeleteReservation {
metadata_revision: 4,
..reservation
}
),
Err(CatalogFileTransitionError::Transition(
CatalogTransitionError::StaleRevision {
expected: 4,
actual: 5,
}
))
));
let deleted = finish_delete_file(&path, reservation).unwrap();
let mut reused = read_catalog(&path).unwrap();
reused.branches.push(record(9, "reusable"));
write_catalog(&path, &reused).unwrap();
let before_replay = fs::read(&path).unwrap();
assert_eq!(
begin_delete_file(&path, request).unwrap(),
DeleteBeginOutcome::Deleted(deleted.clone())
);
assert_eq!(finish_delete_file(&path, reservation).unwrap(), deleted);
assert_eq!(fs::read(&path).unwrap(), before_replay);
assert_eq!(
read_catalog(&path)
.unwrap()
.branches
.iter()
.find(|branch| branch.id == id(9))
.unwrap()
.state,
BranchState::Ready
);
fs::remove_dir_all(directory).unwrap();
}
#[test]
fn concurrent_create_and_delete_keep_both_catalog_transitions() {
let (directory, path) = temporary_catalog_path();
write_catalog(&path, &catalog()).unwrap();
let start = std::sync::Arc::new(std::sync::Barrier::new(3));
let delete_path = path.clone();
let delete_start = std::sync::Arc::clone(&start);
let delete = std::thread::spawn(move || {
delete_start.wait();
begin_delete_file(
&delete_path,
DeleteRequest {
id: id(1),
expected_metadata_revision: 3,
},
)
});
let create_path = path.clone();
let create_start = std::sync::Arc::clone(&start);
let create = std::thread::spawn(move || {
create_start.wait();
reserve_create_file(
&create_path,
CreateRequest {
id: id(3),
name: BranchName::new("third").unwrap(),
parent_id: id(2),
source_commit_epoch: 7,
base_root_digest: [3; DIGEST_BYTES],
owner: None,
request_key: "concurrent-create".to_string(),
request_fingerprint: [3; DIGEST_BYTES],
},
)
});
start.wait();
let reservation = match delete.join().unwrap().unwrap() {
DeleteBeginOutcome::Deleting(reservation) => reservation,
DeleteBeginOutcome::Deleted(_) => panic!("branch must begin deletion once"),
};
let created = create.join().unwrap().unwrap();
let published = read_catalog(&path).unwrap();
assert_eq!(published.revision, 13);
assert_eq!(
published
.branches
.iter()
.find(|branch| branch.id == id(1))
.unwrap()
.state,
BranchState::Deleting
);
assert_eq!(
published
.branches
.iter()
.find(|branch| branch.id == created.id)
.unwrap()
.state,
BranchState::Creating
);
let deleted = finish_delete_file(&path, reservation).unwrap();
assert_eq!(deleted.state, BranchState::Deleted);
assert_eq!(read_catalog(&path).unwrap().revision, 14);
fs::remove_dir_all(directory).unwrap();
}
#[test]
fn concurrent_create_reservations_serialize_the_read_modify_publish_cycle() {
let (directory, path) = temporary_catalog_path();
write_catalog(&path, &catalog()).unwrap();
let first_path = path.clone();
let first =
std::thread::spawn(move || reserve_with_retry(&first_path, create_request_for(3)));
let second_path = path.clone();
let second =
std::thread::spawn(move || reserve_with_retry(&second_path, create_request_for(4)));
let first = first.join().unwrap().unwrap();
let second = second.join().unwrap().unwrap();
assert_ne!(first.id, second.id);
let published = read_catalog(&path).unwrap();
assert!(published
.branches
.iter()
.any(|branch| branch.id == first.id));
assert!(published
.branches
.iter()
.any(|branch| branch.id == second.id));
assert_eq!(published.revision, 13);
fs::remove_dir_all(directory).unwrap();
}
fn create_request_for(byte: u8) -> CreateRequest {
CreateRequest {
id: id(byte),
name: BranchName::new(format!("branch-{byte}")).unwrap(),
parent_id: id(1),
source_commit_epoch: 7,
base_root_digest: [byte; DIGEST_BYTES],
owner: Some("worker-1".to_string()),
request_key: format!("create-{byte}"),
request_fingerprint: [byte.wrapping_add(10); DIGEST_BYTES],
}
}
fn reserve_with_retry(
path: &Path,
request: CreateRequest,
) -> Result<CreateReservation, CatalogFileTransitionError> {
for _ in 0..100 {
match reserve_create_file(path, request.clone()) {
Err(CatalogFileTransitionError::Io(error))
if error.kind() == io::ErrorKind::WouldBlock =>
{
std::thread::sleep(std::time::Duration::from_millis(1));
}
result => return result,
}
}
panic!("catalog metadata lock did not become available");
}
#[test]
fn interrupted_create_without_child_files_is_aborted_on_recovery() {
let (directory, path) = temporary_catalog_path();
write_catalog(&path, &catalog()).unwrap();
let reservation = reserve_create_file(&path, create_request()).unwrap();
let outcome = recover_create_file(
&path,
reservation.id,
&directory.join("missing.head"),
&directory.join("missing.wal"),
1024,
)
.unwrap();
assert_eq!(outcome, CreateRecoveryOutcome::Aborted);
let recovered = read_catalog(&path).unwrap();
let branch = recovered
.branches
.iter()
.find(|branch| branch.id == reservation.id)
.unwrap();
assert_eq!(branch.state, BranchState::Deleted);
assert_eq!(branch.create_outcome, CreateOutcome::Aborted);
fs::remove_dir_all(directory).unwrap();
}
#[test]
fn recovery_completes_a_pending_create_when_files_are_already_durable() {
let (directory, catalog_path) = temporary_catalog_path();
write_catalog(&catalog_path, &catalog()).unwrap();
let parent_head_path = directory.join("parent.head");
let child_directory = directory.join("child");
let child_head_path = child_directory.join("child.head");
let child_wal_path = child_directory.join("child.wal");
fs::create_dir_all(&child_directory).unwrap();
let root = crate::immutable_object::ObjectReference::for_bytes(
crate::immutable_object::ObjectKind::SealedRoot,
1,
b"parent-root",
);
let parent = BranchHead {
project_id: *catalog().project_id.as_uuid().as_bytes(),
branch_id: *id(1).as_uuid().as_bytes(),
physical_generation: 4,
sealed_root: root,
logical_commit_epoch: 7,
active_wal: crate::branch_head::ActiveWalIdentity {
generation: 5,
replay_start_lsn: 20,
byte_length: 1,
sha256: hawdb_integrity::sha256(b"x"),
},
};
fs::write(&parent_head_path, parent.encode().unwrap()).unwrap();
let mut request = create_request();
request.base_root_digest = *root.sha256.as_bytes();
let reservation = reserve_create_file(&catalog_path, request.clone()).unwrap();
let child_request = ChildBranchHeadRequest {
project_id: parent.project_id,
branch_id: *request.id.as_uuid().as_bytes(),
sealed_root: root,
logical_commit_epoch: 7,
active_wal_generation: 1,
replay_start_lsn: 42,
head_path: child_head_path.clone(),
wal_path: child_wal_path.clone(),
};
create_child_branch_head_from_parent(
&parent_head_path,
&child_head_path,
child_request,
ChildBranchSourceExpectation {
branch_id: parent.branch_id,
physical_generation: 4,
logical_commit_epoch: 7,
sealed_root: root,
},
1024,
)
.unwrap();
assert_eq!(
recover_create_file(
&catalog_path,
reservation.id,
&child_head_path,
&child_wal_path,
1024,
)
.unwrap(),
CreateRecoveryOutcome::Completed
);
assert_eq!(
read_catalog(&catalog_path)
.unwrap()
.branches
.iter()
.find(|branch| branch.id == reservation.id)
.unwrap()
.state,
BranchState::Ready
);
fs::remove_dir_all(directory).unwrap();
}
#[cfg(unix)]
#[test]
fn pending_recovery_syncs_complete_but_unsynchronized_files_before_ready() {
use crate::file_descriptors::ProjectFileDescriptors;
use crate::power_loss::{
image::{CrashPlan, ImageLimits},
PowerLossModel,
};
for inject_failure in [false, true] {
let (directory, catalog_path) = temporary_catalog_path();
let project = ProjectFileDescriptors::acquire(&directory, 32).unwrap();
let model = PowerLossModel::attach(&project, ImageLimits::default()).unwrap();
write_catalog(&catalog_path, &catalog()).unwrap();
let root = ObjectReference::for_bytes(ObjectKind::SealedRoot, 1, b"root");
let mut request = create_request();
request.base_root_digest = *root.sha256.as_bytes();
let reservation = reserve_create_file(&catalog_path, request).unwrap();
let child_directory = directory.join("new/child");
fs::create_dir_all(&child_directory).unwrap();
let wal_path = child_directory.join("child.wal");
let head_path = child_directory.join("child.head");
let wal = crate::wal::frame::encode_binary_wal_header(1, 42);
fs::write(&wal_path, wal).unwrap();
let identity =
crate::branch_head::active_wal_identity_from_file(&wal_path, 1, 42, 1024).unwrap();
let head = BranchHead {
project_id: *catalog().project_id.as_uuid().as_bytes(),
branch_id: *reservation.id.as_uuid().as_bytes(),
physical_generation: 1,
sealed_root: root,
logical_commit_epoch: 7,
active_wal: identity,
};
fs::write(&head_path, head.encode().unwrap()).unwrap();
let before_head = fs::read(&head_path).unwrap();
let before_wal = fs::read(&wal_path).unwrap();
if inject_failure {
let failure = crate::durability::fail_sync_directory_for(project.root());
assert!(matches!(
recover_create_file(&catalog_path, reservation.id, &head_path, &wal_path, 1024),
Err(BranchCreateError::Head(
BranchHeadError::CandidatePublicationUncertain { .. }
))
));
assert_eq!(fs::read(&head_path).unwrap(), before_head);
assert_eq!(fs::read(&wal_path).unwrap(), before_wal);
assert_eq!(
read_catalog(&catalog_path)
.unwrap()
.branches
.iter()
.find(|branch| branch.id == reservation.id)
.unwrap()
.state,
BranchState::Creating
);
drop(failure);
}
assert_eq!(
recover_create_file(&catalog_path, reservation.id, &head_path, &wal_path, 1024)
.unwrap(),
CreateRecoveryOutcome::Completed
);
let image = model
.capture()
.unwrap()
.crash(&CrashPlan::default())
.unwrap();
assert_eq!(
image.bytes(Path::new("new/child/child.head")).unwrap(),
before_head
);
assert_eq!(
image.bytes(Path::new("new/child/child.wal")).unwrap(),
before_wal
);
let recovered_catalog =
Catalog::decode(image.bytes(Path::new("catalog.hawdb")).unwrap()).unwrap();
assert_eq!(
recovered_catalog
.branches
.iter()
.find(|branch| branch.id == reservation.id)
.unwrap()
.state,
BranchState::Ready
);
assert_eq!(project.metrics().open, 0);
assert!(project.metrics().high_water <= 32);
std::fs::remove_dir_all(directory).unwrap();
}
}
#[test]
fn recovery_keeps_creating_for_corrupt_child_head() {
let (directory, path) = temporary_catalog_path();
write_catalog(&path, &catalog()).unwrap();
let reservation = reserve_create_file(&path, create_request()).unwrap();
let head_path = directory.join("corrupt.head");
let wal_path = directory.join("corrupt.wal");
fs::write(&head_path, b"corrupt").unwrap();
let error =
recover_create_file(&path, reservation.id, &head_path, &wal_path, 1024).unwrap_err();
assert!(matches!(error, BranchCreateError::Head(_)));
assert_eq!(
read_catalog(&path)
.unwrap()
.branches
.iter()
.find(|branch| branch.id == reservation.id)
.unwrap()
.state,
BranchState::Creating
);
fs::remove_dir_all(directory).unwrap();
}
#[derive(Clone)]
struct CreateFixture {
directory: PathBuf,
catalog_path: PathBuf,
parent_head_path: PathBuf,
child_request: ChildBranchHeadRequest,
expected_parent: ChildBranchSourceExpectation,
request: CreateRequest,
}
impl CreateFixture {
fn new() -> Self {
let (directory, catalog_path) = temporary_catalog_path();
write_catalog(&catalog_path, &catalog()).unwrap();
let parent_head_path = directory.join("parent.head");
let child_directory = directory.join("child");
let child_head_path = child_directory.join("child.head");
let child_wal_path = child_directory.join("child.wal");
let root = crate::immutable_object::ObjectReference::for_bytes(
crate::immutable_object::ObjectKind::SealedRoot,
1,
b"parent-root",
);
let parent = BranchHead {
project_id: *catalog().project_id.as_uuid().as_bytes(),
branch_id: *id(1).as_uuid().as_bytes(),
physical_generation: 4,
sealed_root: root,
logical_commit_epoch: 7,
active_wal: crate::branch_head::ActiveWalIdentity {
generation: 5,
replay_start_lsn: 20,
byte_length: 1,
sha256: hawdb_integrity::sha256(b"x"),
},
};
fs::write(&parent_head_path, parent.encode().unwrap()).unwrap();
let mut request = create_request();
request.base_root_digest = *root.sha256.as_bytes();
Self {
directory,
catalog_path,
parent_head_path,
child_request: ChildBranchHeadRequest {
project_id: parent.project_id,
branch_id: *request.id.as_uuid().as_bytes(),
sealed_root: root,
logical_commit_epoch: 7,
active_wal_generation: 1,
replay_start_lsn: 42,
head_path: child_head_path.clone(),
wal_path: child_wal_path.clone(),
},
expected_parent: ChildBranchSourceExpectation {
branch_id: parent.branch_id,
physical_generation: 4,
logical_commit_epoch: 7,
sealed_root: root,
},
request,
}
}
fn child_directory(&self) -> &Path {
self.child_request.head_path.parent().unwrap()
}
fn create(&self) -> Result<BranchCreateResult, BranchCreateError> {
create_branch_from_parent(
&self.catalog_path,
&self.parent_head_path,
self.child_request.clone(),
self.expected_parent,
1024,
self.request.clone(),
)
}
}
#[test]
fn create_branch_from_parent_completes_catalog_after_child_files() {
let fixture = CreateFixture::new();
let child = fixture.create().unwrap();
assert_eq!(child.head.sealed_root, fixture.child_request.sealed_root);
assert_eq!(
read_catalog(&fixture.catalog_path)
.unwrap()
.branches
.last()
.unwrap()
.state,
BranchState::Ready
);
assert!(fixture.child_request.head_path.is_file());
assert!(fixture.child_request.wal_path.is_file());
assert!(matches!(
DatabaseDirectoryLease::acquire(fixture.child_directory()),
Err(DatabaseDirectoryLeaseError::AlreadyOpen)
));
let first_head = child.head;
drop(child);
let retried = fixture.create().unwrap();
assert_eq!(retried.head, first_head);
drop(retried);
let reopened =
crate::branch_head::read_branch_head(&fixture.child_request.head_path).unwrap();
assert_eq!(reopened.sealed_root, fixture.child_request.sealed_root);
assert_eq!(
read_catalog(&fixture.catalog_path)
.unwrap()
.branches
.iter()
.find(|branch| branch.id == fixture.request.id)
.unwrap()
.state,
BranchState::Ready
);
let reopened_lease = DatabaseDirectoryLease::acquire(fixture.child_directory()).unwrap();
drop(reopened_lease);
fs::remove_dir_all(fixture.directory).unwrap();
}
#[test]
fn create_branch_directory_and_lease_failures_leave_no_pending_receipt() {
let fixture = CreateFixture::new();
let before = fs::read(&fixture.catalog_path).unwrap();
fs::write(fixture.child_directory(), b"not a directory").unwrap();
assert!(matches!(fixture.create(), Err(BranchCreateError::Head(_))));
assert_eq!(fs::read(&fixture.catalog_path).unwrap(), before);
fs::remove_file(fixture.child_directory()).unwrap();
fs::create_dir(fixture.child_directory()).unwrap();
let lease = DatabaseDirectoryLease::acquire(fixture.child_directory()).unwrap();
assert!(matches!(
fixture.create(),
Err(BranchCreateError::Lease(
DatabaseDirectoryLeaseError::AlreadyOpen
))
));
assert_eq!(fs::read(&fixture.catalog_path).unwrap(), before);
drop(lease);
let child = fixture.create().unwrap();
assert_eq!(
child.head.branch_id,
*fixture.request.id.as_uuid().as_bytes()
);
drop(child);
fs::remove_dir_all(fixture.directory).unwrap();
}
#[test]
fn create_branch_owns_child_before_waiting_for_catalog_reservation() {
let fixture = CreateFixture::new();
let before = fs::read(&fixture.catalog_path).unwrap();
let metadata = CatalogMetadataLease::acquire(&fixture.directory).unwrap();
let creator_fixture = fixture.clone();
let creator = std::thread::spawn(move || creator_fixture.create());
let child_directory = fs::canonicalize(&fixture.directory).unwrap().join("child");
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
let owns_child = loop {
if crate::ownership::has_process_lease(&child_directory) {
break true;
}
if creator.is_finished() || std::time::Instant::now() >= deadline {
break false;
}
std::thread::sleep(std::time::Duration::from_millis(1));
};
let waiting_catalog = fs::read(&fixture.catalog_path).unwrap();
drop(metadata);
let child = creator.join().unwrap().unwrap();
assert!(
owns_child,
"creation must own the child before publishing a pending receipt"
);
assert_eq!(waiting_catalog, before);
drop(child);
fs::remove_dir_all(fixture.directory).unwrap();
}
#[test]
fn metadata_lock_is_stable_and_serializes_catalog_writers() {
let (directory, _path) = temporary_catalog_path();
let first = CatalogMetadataLease::acquire(&directory).unwrap();
assert_eq!(
CatalogMetadataLease::acquire(&directory)
.unwrap_err()
.kind(),
io::ErrorKind::WouldBlock
);
drop(first);
CatalogMetadataLease::acquire(&directory).unwrap();
assert!(directory.join(METADATA_LOCK_FILE).is_file());
fs::remove_dir_all(directory).unwrap();
}
#[test]
fn root_initialization_and_head_binding_share_one_catalog_transition() {
let (directory, path) = temporary_catalog_path();
let start = std::sync::Arc::new(std::sync::Barrier::new(2));
let first_path = path.clone();
let first_start = std::sync::Arc::clone(&start);
let first = std::thread::spawn(move || {
first_start.wait();
initialize_catalog_file(&first_path, id(90), id(91)).unwrap()
});
start.wait();
let second = initialize_catalog_file(&path, id(90), id(91)).unwrap();
let first = first.join().unwrap();
assert_eq!(first, second);
assert_eq!(read_catalog(&path).unwrap().branches, vec![first.clone()]);
let bound = bind_main_head_file(&path, id(90), id(91), [7; DIGEST_BYTES], 12).unwrap();
assert_eq!(bound.base_root_digest, Some([7; DIGEST_BYTES]));
assert_eq!(bound.source_commit_epoch, 12);
assert_eq!(bound.metadata_revision, first.metadata_revision + 1);
assert_eq!(read_catalog(&path).unwrap().revision, 2);
assert_eq!(
bind_main_head_file(&path, id(90), id(91), [7; DIGEST_BYTES], 12).unwrap(),
bound
);
assert!(matches!(
bind_main_head_file(&path, id(90), id(91), [8; DIGEST_BYTES], 12),
Err(CatalogFileTransitionError::Transition(
CatalogTransitionError::Conflict(_)
))
));
fs::remove_dir_all(directory).unwrap();
}
#[test]
fn file_publication_failure_keeps_previous_bytes_and_cleans_candidate() {
let (directory, path) = temporary_catalog_path();
let first = catalog();
write_catalog(&path, &first).unwrap();
let before = fs::read(&path).unwrap();
let _failure = durability::fail_durable_replace_for_destination("catalog.hawdb");
assert!(write_catalog(
&path,
&Catalog {
revision: 12,
..first
}
)
.is_err());
assert_eq!(fs::read(&path).unwrap(), before);
assert!(path.is_file());
assert!(directory.join(METADATA_LOCK_FILE).is_file());
fs::remove_dir_all(directory).unwrap();
}
#[test]
fn catalog_reclamation_binds_heads_leases_and_deleted_directories() {
let (directory, catalog_path) = temporary_catalog_path();
let object_root = directory.join("objects");
let mut objects = ImmutableObjectStore::open(&object_root).unwrap();
let checkpoint = ObjectReference::for_bytes(ObjectKind::Checkpoint, 1, b"checkpoint");
let wal = ObjectReference::for_bytes(ObjectKind::SealedWal, 1, b"wal");
let manifest = ObjectReference::for_bytes(ObjectKind::DurableManifest, 1, b"manifest");
objects.publish(checkpoint, b"checkpoint").unwrap();
objects.publish(wal, b"wal").unwrap();
objects.publish(manifest, b"manifest").unwrap();
let sealed_root = crate::sealed_root::SealedRoot {
checkpoint_epoch: 1,
commit_epoch: 1,
wal_replay_start_lsn: 0,
durable_manifest: manifest,
checkpoint_references: vec![checkpoint],
checkpoint_bindings: vec![crate::sealed_root::CheckpointArtifactBinding {
relative_path: "checkpoint.hawdb".to_string(),
reference: checkpoint,
}],
sealed_wals: vec![crate::sealed_root::SealedWalReference {
start_lsn: 0,
end_lsn: 3,
object: wal,
}],
};
let sealed_root_bytes = sealed_root.encode().unwrap();
let root_reference =
ObjectReference::for_bytes(ObjectKind::SealedRoot, 2, &sealed_root_bytes);
let orphan = ObjectReference::for_bytes(ObjectKind::CheckpointArtifact, 1, b"orphan");
objects.publish(root_reference, &sealed_root_bytes).unwrap();
objects.publish(orphan, b"orphan").unwrap();
let parent_directory = directory.join("parent");
let deleted_directory = directory.join("deleted");
fs::create_dir_all(&parent_directory).unwrap();
fs::create_dir_all(&deleted_directory).unwrap();
fs::write(deleted_directory.join("head"), b"old").unwrap();
let parent_head_path = parent_directory.join("branch.head");
fs::write(
&parent_head_path,
BranchHead {
project_id: *id(99).as_uuid().as_bytes(),
branch_id: *id(1).as_uuid().as_bytes(),
physical_generation: 1,
sealed_root: root_reference,
logical_commit_epoch: 1,
active_wal: crate::branch_head::ActiveWalIdentity {
generation: 1,
replay_start_lsn: 3,
byte_length: 3,
sha256: hawdb_integrity::sha256(b"wal"),
},
}
.encode()
.unwrap(),
)
.unwrap();
let mut deleted = record(2, "deleted");
deleted.parent_id = Some(id(1));
deleted.state = BranchState::Deleted;
deleted.create_outcome = CreateOutcome::Succeeded;
write_catalog(
&catalog_path,
&Catalog {
project_id: id(99),
revision: 1,
branches: vec![record(1, "parent"), deleted],
},
)
.unwrap();
let report = reclaim_catalog_branches(
&catalog_path,
&mut objects,
&[root_reference, manifest, checkpoint, wal, orphan],
&[
BranchReclamationPath {
id: id(1),
directory: parent_directory,
head_path: parent_head_path,
},
BranchReclamationPath {
id: id(2),
directory: deleted_directory.clone(),
head_path: deleted_directory.join("branch.head"),
},
],
)
.unwrap();
assert_eq!(report.retained_objects, 4);
assert_eq!(report.reclaimed_objects, 1);
assert!(!deleted_directory.exists());
fs::remove_dir_all(directory).unwrap();
}
#[test]
fn read_rejects_an_oversized_catalog_before_loading_bytes() {
let (directory, path) = temporary_catalog_path();
let file = fs::File::create(&path).unwrap();
file.set_len((MAX_CATALOG_BYTES + 1) as u64).unwrap();
let error = read_catalog(&path).unwrap_err();
assert_eq!(error.kind(), io::ErrorKind::InvalidData);
fs::remove_dir_all(directory).unwrap();
}
#[test]
fn codec_rejects_tampering_versions_duplicates_and_trailing_bytes() {
let catalog = Catalog {
project_id: id(99),
revision: 1,
branches: vec![record(1, "one")],
};
let encoded = catalog.encode().unwrap();
let mut tampered = encoded.clone();
tampered[10] ^= 1;
assert_eq!(Catalog::decode(&tampered), Err(CatalogError::Checksum));
let mut trailing = encoded.clone();
trailing.insert(trailing.len() - 4, 0);
let checksum = crc32c(&trailing[..trailing.len() - 4]).get();
let checksum_offset = trailing.len() - 4;
trailing[checksum_offset..].copy_from_slice(&checksum.to_le_bytes());
assert_eq!(Catalog::decode(&trailing), Err(CatalogError::TrailingBytes));
let duplicate = Catalog {
project_id: id(99),
revision: 1,
branches: vec![record(1, "one"), record(1, "two")],
};
assert_eq!(
duplicate.encode(),
Err(CatalogError::Duplicate("branch UUID"))
);
}
#[test]
fn codec_rejects_old_headers_versions_and_retired_state_tag() {
let encoded = catalog().encode().unwrap();
assert_eq!(&encoded[..8], b"HBCATV2\0");
for (offset, replacement) in [(0, b"HBCATV1\0".as_slice()), (8, &[1, 0])] {
let mut old = encoded.clone();
old[offset..offset + replacement.len()].copy_from_slice(replacement);
let checksum_offset = old.len() - 4;
let checksum = crc32c(&old[..checksum_offset]).get();
old[checksum_offset..].copy_from_slice(&checksum.to_le_bytes());
assert_eq!(Catalog::decode(&old), Err(CatalogError::Version));
}
let branch = record(1, "first");
let mut bytes = Vec::new();
bytes.extend_from_slice(MAGIC);
put_u16(&mut bytes, VERSION);
bytes.extend_from_slice(id(99).as_uuid().as_bytes());
put_u64(&mut bytes, 1);
put_u32(&mut bytes, 1);
let record_start = bytes.len();
encode_branch(&mut bytes, &branch).unwrap();
let mut reader = Reader::new(&bytes[record_start..]);
let _ = reader.array::<16>().unwrap();
let _ = reader.string(MAX_NAME_BYTES, "branch name").unwrap();
let _ = reader.optional_uuid().unwrap();
let _ = reader.u64().unwrap();
let _ = reader.optional_array().unwrap();
let _ = reader.u64().unwrap();
let state_offset = reader.offset;
bytes[record_start + state_offset] = 2;
let checksum = crc32c(&bytes).get();
put_u32(&mut bytes, checksum);
assert_eq!(
Catalog::decode(&bytes),
Err(CatalogError::InvalidEnum("branch state"))
);
}
#[test]
fn lifecycle_transitions_are_idempotent_and_revision_bound() {
let mut catalog = catalog();
let request = create_request();
let before = catalog.encode().unwrap();
assert_eq!(catalog.reserve_create(request.clone()).unwrap(), id(3));
assert_eq!(catalog.branches.len(), 3);
assert_eq!(catalog.branches[2].state, BranchState::Creating);
assert_eq!(catalog.reserve_create(request).unwrap(), id(3));
assert_eq!(catalog.encode().unwrap(), catalog.encode().unwrap());
assert_ne!(catalog.encode().unwrap(), before);
assert_eq!(
catalog.complete_create(id(3), 99),
Err(CatalogTransitionError::StaleRevision {
expected: 99,
actual: 1
})
);
catalog.complete_create(id(3), 1).unwrap();
assert_eq!(catalog.branches[2].state, BranchState::Ready);
catalog
.rename(id(3), 2, BranchName::new("renamed").unwrap())
.unwrap();
catalog.begin_delete(id(3), 3).unwrap();
catalog.finish_delete(id(3), 4).unwrap();
assert_eq!(catalog.branches[2].state, BranchState::Deleted);
let mut reused = create_request();
reused.id = id(4);
reused.name = BranchName::new("renamed").unwrap();
reused.request_key = "create-reused-name".to_string();
reused.request_fingerprint = [10; DIGEST_BYTES];
assert_eq!(catalog.reserve_create(reused).unwrap(), id(4));
}
#[test]
fn failed_reservation_and_cas_leave_catalog_bytes_unchanged() {
let mut catalog = catalog();
let mut request = create_request();
request.source_commit_epoch = 6;
let before = catalog.encode().unwrap();
assert_eq!(
catalog.reserve_create(request),
Err(CatalogTransitionError::Conflict(
"create source revision is stale"
))
);
assert_eq!(catalog.encode().unwrap(), before);
assert_eq!(
catalog.rename(id(1), 99, BranchName::new("renamed").unwrap()),
Err(CatalogTransitionError::StaleRevision {
expected: 99,
actual: 3
})
);
assert_eq!(catalog.encode().unwrap(), before);
}
#[test]
fn invalid_uuid_and_deleted_pending_state_fail_closed() {
assert!(BranchId::new(Uuid::nil()).is_err());
assert!(BranchId::parse("not-a-uuid").is_err());
let mut branch = record(1, "one");
branch.state = BranchState::Deleted;
branch.create_outcome = CreateOutcome::Pending;
let catalog = Catalog {
project_id: id(99),
revision: 1,
branches: vec![branch],
};
assert_eq!(
catalog.encode(),
Err(CatalogError::InvalidState(
"deleted branch cannot have a pending create outcome"
))
);
}
}