use crate::foundation::{FileId, GenerationId, OperationId, VolumeId};
use crate::kernel::{
ExtentKind, FileKind, FileMetadata, FilePayload, LogicalName, MetadataField, NamespacePath,
};
use crate::model::{CheckoutMode, GenerationSelector};
use crate::path::PortablePath;
use crate::{
AsyncAuthorityStore, AsyncObjectStore, AuthoredMutation, Checkout, CheckoutCommitOutcome,
ConflictKey, ConflictKind, ConflictSide, ConflictValue, FsError, GenerationDiff, MergeConflict,
MergeDriverRegistry, MergePlan, MergePlanResolutionError, MergeResolution,
MergeResolutionCache, UnpublishedMergeCandidate, Volume, resolve_merge_plan,
};
use bytes::Bytes;
use serde::{Deserialize, Serialize};
use std::collections::{BTreeMap, BTreeSet};
use std::fmt;
use std::sync::Arc;
use thiserror::Error;
use unicode_normalization::UnicodeNormalization;
const WORKSPACE_ID_DOMAIN: &[u8] = b"acyclic-fs-workspace-id-v1\0";
const MAXIMUM_WORKSPACE_NAME_BYTES: usize = 255;
#[derive(Clone, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct WorkspaceName(Arc<str>);
impl WorkspaceName {
pub fn new(name: impl AsRef<str>) -> Result<Self, WorkspaceNameError> {
let canonical: String = name.as_ref().nfc().collect();
if canonical.is_empty() {
return Err(WorkspaceNameError::Empty);
}
if canonical == "." || canonical == ".." {
return Err(WorkspaceNameError::Reserved);
}
if canonical.len() > MAXIMUM_WORKSPACE_NAME_BYTES {
return Err(WorkspaceNameError::TooLong);
}
if canonical
.chars()
.any(|character| character.is_control() || matches!(character, '/' | '\\'))
{
return Err(WorkspaceNameError::InvalidCharacter);
}
Ok(Self(Arc::from(canonical)))
}
#[must_use]
pub fn as_str(&self) -> &str {
&self.0
}
}
impl fmt::Debug for WorkspaceName {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_tuple("WorkspaceName")
.field(&self.0)
.finish()
}
}
impl fmt::Display for WorkspaceName {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str(self.as_str())
}
}
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize, Deserialize)]
#[serde(transparent)]
pub struct WorkspaceId([u8; 16]);
impl WorkspaceId {
pub(crate) const fn from_volume_id(volume_id: VolumeId) -> Self {
Self(volume_id.into_bytes())
}
#[must_use]
pub const fn from_bytes(bytes: [u8; 16]) -> Self {
Self(bytes)
}
#[must_use]
pub(crate) fn derive(namespace: [u8; 16], name: &WorkspaceName) -> Self {
let mut hasher = blake3::Hasher::new();
hasher.update(WORKSPACE_ID_DOMAIN);
hasher.update(&namespace);
hasher.update(
&u64::try_from(name.as_str().len())
.unwrap_or(u64::MAX)
.to_le_bytes(),
);
hasher.update(name.as_str().as_bytes());
let mut bytes = [0_u8; 16];
bytes.copy_from_slice(&hasher.finalize().as_bytes()[..16]);
Self(bytes)
}
#[must_use]
pub const fn into_bytes(self) -> [u8; 16] {
self.0
}
pub(crate) const fn volume_id(self) -> VolumeId {
VolumeId::from_bytes(self.0)
}
}
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize, Deserialize)]
#[serde(transparent)]
pub struct IdempotencyKey([u8; 16]);
impl IdempotencyKey {
#[must_use]
pub fn new() -> Self {
Self(OperationId::new().into_bytes())
}
#[must_use]
pub const fn from_bytes(bytes: [u8; 16]) -> Self {
Self(bytes)
}
#[must_use]
pub const fn into_bytes(self) -> [u8; 16] {
self.0
}
pub(crate) const fn operation_id(self) -> OperationId {
OperationId::from_bytes(self.0)
}
}
impl Default for IdempotencyKey {
fn default() -> Self {
Self::new()
}
}
pub struct Workspace<A, O> {
pub(crate) name: WorkspaceName,
pub(crate) id: WorkspaceId,
pub(crate) volume: Volume<A, O>,
#[cfg(all(feature = "native-watch", not(target_arch = "wasm32")))]
pub(crate) source: Option<crate::Source<A, O>>,
}
impl<A, O> Clone for Workspace<A, O> {
fn clone(&self) -> Self {
Self {
name: self.name.clone(),
id: self.id,
volume: self.volume.clone(),
#[cfg(all(feature = "native-watch", not(target_arch = "wasm32")))]
source: self.source.clone(),
}
}
}
impl<A, O> Workspace<A, O> {
#[must_use]
pub const fn id(&self) -> WorkspaceId {
self.id
}
pub(crate) fn authority(&self) -> &A {
self.volume.fs.authority()
}
pub(crate) const fn authority_id(&self) -> crate::AuthorityId {
crate::kernel::volume_authority_id(self.id.volume_id())
}
#[must_use]
pub fn name(&self) -> &WorkspaceName {
&self.name
}
pub fn fork_workspace_id(
&self,
destination: impl AsRef<str>,
) -> Result<WorkspaceId, WorkspaceNameError> {
self.volume.fs.workspace_id(destination)
}
#[must_use]
pub const fn profile(&self) -> crate::model::FilesystemProfile {
self.volume.config.profile
}
#[must_use]
pub const fn limits(&self) -> crate::model::VolumeLimits {
self.volume.config.limits
}
}
impl<A: AsyncAuthorityStore, O: AsyncObjectStore> Workspace<A, O> {
pub async fn operation_generation(
&self,
idempotency_key: IdempotencyKey,
) -> Result<Option<Generation<A, O>>, WorkspaceError> {
self.volume
.fs
.workspace_operation_generation(&self.volume, idempotency_key.operation_id())
.await
.map(|generation| {
generation.map(|id| Generation {
workspace: self.clone(),
id,
})
})
}
pub async fn checkout(
&self,
selector: GenerationSelector,
mode: CheckoutMode,
) -> Result<Checkout<A, O>, WorkspaceError> {
self.engine_checkout(selector, mode).await
}
#[cfg(all(feature = "native-watch", not(target_arch = "wasm32")))]
#[must_use]
pub fn source(&self) -> Option<&crate::Source<A, O>> {
self.source.as_ref()
}
#[cfg(all(feature = "native-watch", not(target_arch = "wasm32")))]
pub async fn seal(&self) -> Result<Generation<A, O>, crate::SourceError> {
match &self.source {
Some(source) => source.seal().await,
None => self
.head()
.await
.map_err(|error| crate::SourceError::Engine(error.to_string())),
}
}
pub async fn head(&self) -> Result<Generation<A, O>, WorkspaceError> {
self.head_measured(
crate::WorkBudget::UNBOUNDED,
&crate::CancellationToken::new(),
)
.await
.map(|receipt| receipt.value)
}
pub(crate) async fn head_measured(
&self,
budget: crate::WorkBudget,
cancellation: &crate::CancellationToken,
) -> Result<crate::OperationReceipt<Generation<A, O>>, WorkspaceError> {
let receipt = self
.volume
.checkout(
GenerationSelector::Head,
CheckoutMode::read_only_pinned(),
budget,
cancellation,
)
.await
.map_err(|failure| WorkspaceError::from(failure.error))?;
Ok(crate::OperationReceipt {
value: Generation {
workspace: self.clone(),
id: receipt.value.generation_id(),
},
work: receipt.work,
})
}
pub async fn generation(
&self,
generation_id: GenerationId,
) -> Result<Generation<A, O>, WorkspaceError> {
self.generation_measured(
generation_id,
crate::WorkBudget::UNBOUNDED,
&crate::CancellationToken::new(),
)
.await
.map(|receipt| receipt.value)
}
pub(crate) async fn generation_measured(
&self,
generation_id: GenerationId,
budget: crate::WorkBudget,
cancellation: &crate::CancellationToken,
) -> Result<crate::OperationReceipt<Generation<A, O>>, WorkspaceError> {
let checkout = self
.engine_checkout_measured(
GenerationSelector::Exact(generation_id),
CheckoutMode::read_only_pinned(),
budget,
cancellation,
)
.await?;
Ok(crate::OperationReceipt {
value: Generation {
workspace: self.clone(),
id: checkout.value.generation_id(),
},
work: checkout.work,
})
}
pub async fn restore_generation(
&self,
generation: &Generation<A, O>,
if_current: GenerationId,
idempotency_key: IdempotencyKey,
) -> Result<WorkspaceRestore<A, O>, WorkspaceError> {
self.restore_generation_with_permit(
generation,
if_current,
idempotency_key,
crate::PublicationPermit::Unrestricted,
)
.await
}
pub async fn restore_generation_with_permit(
&self,
generation: &Generation<A, O>,
if_current: GenerationId,
idempotency_key: IdempotencyKey,
permit: crate::PublicationPermit,
) -> Result<WorkspaceRestore<A, O>, WorkspaceError> {
if generation.workspace.id != self.id {
return Err(WorkspaceError::ForeignGeneration);
}
let outcome = self
.volume
.fs
.restore_workspace_generation(
&self.volume,
generation.id,
if_current,
idempotency_key.operation_id(),
permit,
)
.await?;
let restored = |id| Generation {
workspace: self.clone(),
id,
};
Ok(match outcome {
WorkspaceRestoreOutcome::Restored(id) => WorkspaceRestore::Restored(restored(id)),
WorkspaceRestoreOutcome::AlreadyRestored(id) => {
WorkspaceRestore::AlreadyRestored(restored(id))
}
WorkspaceRestoreOutcome::Current(id) => WorkspaceRestore::Current(restored(id)),
WorkspaceRestoreOutcome::Stale(id) => WorkspaceRestore::Stale(restored(id)),
WorkspaceRestoreOutcome::Fenced => WorkspaceRestore::Fenced,
WorkspaceRestoreOutcome::IdempotencyConflict => WorkspaceRestore::IdempotencyConflict,
})
}
pub async fn restore_paths_from(
&self,
source: &Generation<A, O>,
paths: &[String],
if_current: GenerationId,
idempotency_key: IdempotencyKey,
) -> Result<TransactionCommit<A, O>, WorkspaceError> {
self.restore_paths_from_with_permit(
source,
paths,
if_current,
idempotency_key,
crate::PublicationPermit::Unrestricted,
)
.await
}
pub async fn restore_paths_from_with_permit(
&self,
source: &Generation<A, O>,
paths: &[String],
if_current: GenerationId,
idempotency_key: IdempotencyKey,
permit: crate::PublicationPermit,
) -> Result<TransactionCommit<A, O>, WorkspaceError> {
if !self.volume.fs.same_deployment(&source.workspace.volume.fs) {
return Err(WorkspaceError::ForeignGeneration);
}
let current = self.head().await?;
if current.id != if_current {
return Ok(TransactionCommit::Conflict { actual: current });
}
let mut source_checkout = source
.workspace
.engine_checkout(
GenerationSelector::Exact(source.id),
CheckoutMode::read_only_pinned(),
)
.await?;
let mut transaction = self.begin_transaction(idempotency_key).await?;
let config = self.volume.config;
let cancellation = crate::CancellationToken::new();
let parsed = paths
.iter()
.map(|path| {
let relative = path.trim_start_matches('/');
let absolute = format!("/{relative}");
customer_path(&absolute, config).map(|path| (relative, path))
})
.collect::<Result<Vec<_>, _>>()?;
let lookup_paths = parsed
.iter()
.map(|(_, path)| path.clone())
.collect::<Vec<_>>();
let source_records = source_checkout
.lookup_batch_no_follow(&lookup_paths, crate::WorkBudget::UNBOUNDED, &cancellation)
.await
.map_err(WorkspaceError::engine)?
.value
.entries
.into_iter()
.map(|entry| entry.record)
.collect::<Vec<_>>();
let missing_paths = parsed
.iter()
.zip(&source_records)
.filter(|(_, record)| record.is_none())
.map(|((_, path), _)| path.clone())
.collect::<Vec<_>>();
let current_records = if missing_paths.is_empty() {
Vec::new()
} else {
transaction
.checkout
.lookup_batch_no_follow(&missing_paths, crate::WorkBudget::UNBOUNDED, &cancellation)
.await
.map_err(WorkspaceError::engine)?
.value
.entries
.into_iter()
.map(|entry| entry.record)
.collect()
};
let mut current_records = current_records.into_iter();
let mut operations = Vec::with_capacity(paths.len());
for ((relative, path), source_record) in parsed.into_iter().zip(source_records) {
if let Some(record) = source_record {
if record.kind != FileKind::Directory
&& let Some((parent, _)) = relative.rsplit_once('/')
{
transaction.create_dir_all(&format!("/{parent}")).await?;
}
operations.push(crate::kernel::Mutation::Restore { path, record });
} else {
let current_record = current_records
.next()
.ok_or_else(|| WorkspaceError::engine("missing batch lookup result"))?;
if let Some(record) = current_record {
operations.push(crate::kernel::Mutation::Remove {
path,
expected_file_id: MetadataField::Value(record.file_id),
});
}
}
}
if !operations.is_empty() {
transaction
.checkout
.mutate(operations, crate::WorkBudget::UNBOUNDED, &cancellation)
.await
.map_err(WorkspaceError::engine)?;
}
transaction.commit_with_permit(permit).await
}
#[allow(
clippy::too_many_lines,
reason = "one atomic path application keeps conflict classification and publication together"
)]
pub async fn apply_paths_from(
&self,
base: Option<&Generation<A, O>>,
source: Option<&Generation<A, O>>,
paths: &[String],
if_current: GenerationId,
idempotency_key: IdempotencyKey,
) -> Result<WorkspacePathApply<A, O>, WorkspaceError> {
self.apply_paths_from_with_permit(
base,
source,
paths,
if_current,
idempotency_key,
crate::PublicationPermit::Unrestricted,
)
.await
}
#[allow(
clippy::too_many_lines,
reason = "one transaction keeps path planning, typed conflicts, and fenced publication atomic"
)]
pub async fn apply_paths_from_with_permit(
&self,
base: Option<&Generation<A, O>>,
source: Option<&Generation<A, O>>,
paths: &[String],
if_current: GenerationId,
idempotency_key: IdempotencyKey,
permit: crate::PublicationPermit,
) -> Result<WorkspacePathApply<A, O>, WorkspaceError> {
if base.is_some_and(|base| !self.volume.fs.same_deployment(&base.workspace.volume.fs))
|| source
.is_some_and(|source| !self.volume.fs.same_deployment(&source.workspace.volume.fs))
{
return Err(WorkspaceError::ForeignGeneration);
}
let current = self.head().await?;
if current.id != if_current {
return Ok(WorkspacePathApply::Stale(current));
}
let mut base_checkout = match base {
Some(base) => Some(
base.workspace
.engine_checkout(
GenerationSelector::Exact(base.id),
CheckoutMode::read_only_pinned(),
)
.await?,
),
None => None,
};
let mut source_checkout = match source {
Some(source) => Some(
source
.workspace
.engine_checkout(
GenerationSelector::Exact(source.id),
CheckoutMode::read_only_pinned(),
)
.await?,
),
None => None,
};
let mut transaction = self.begin_transaction(idempotency_key).await?;
let config = self.volume.config;
let cancellation = crate::CancellationToken::new();
let parsed = paths
.iter()
.map(|path| {
let relative = path.trim_start_matches('/');
let absolute = format!("/{relative}");
customer_path(&absolute, config).map(|path| (relative, path))
})
.collect::<Result<Vec<_>, _>>()?;
let lookup_paths = parsed
.iter()
.map(|(_, path)| path.clone())
.collect::<Vec<_>>();
let base_records = match &mut base_checkout {
Some(checkout) => checkout
.lookup_batch_no_follow(&lookup_paths, crate::WorkBudget::UNBOUNDED, &cancellation)
.await
.map_err(WorkspaceError::engine)?
.value
.entries
.into_iter()
.map(|entry| entry.record)
.collect(),
None => vec![None; parsed.len()],
};
let source_records = match &mut source_checkout {
Some(checkout) => checkout
.lookup_batch_no_follow(&lookup_paths, crate::WorkBudget::UNBOUNDED, &cancellation)
.await
.map_err(WorkspaceError::engine)?
.value
.entries
.into_iter()
.map(|entry| entry.record)
.collect(),
None => vec![None; parsed.len()],
};
let changed_paths = parsed
.iter()
.zip(base_records.iter().zip(&source_records))
.filter(|(_, (base, source))| base != source)
.map(|((_, path), _)| path.clone())
.collect::<Vec<_>>();
let current_records = if changed_paths.is_empty() {
Vec::new()
} else {
transaction
.checkout
.lookup_batch_no_follow(&changed_paths, crate::WorkBudget::UNBOUNDED, &cancellation)
.await
.map_err(WorkspaceError::engine)?
.value
.entries
.into_iter()
.map(|entry| entry.record)
.collect()
};
let mut current_records = current_records.into_iter();
let mut operations = Vec::with_capacity(paths.len());
let mut conflicts = Vec::new();
for (((relative, path), base_record), source_record) in
parsed.into_iter().zip(base_records).zip(source_records)
{
if base_record == source_record {
continue;
}
let current_record = current_records
.next()
.ok_or_else(|| WorkspaceError::engine("missing batch lookup result"))?;
if current_record == source_record {
continue;
}
if current_record != base_record {
conflicts.push(WorkspacePathConflict {
path: relative.to_owned(),
kind: path_conflict_kind(base_record, current_record, source_record),
});
continue;
}
match source_record {
Some(record) => {
if record.kind != FileKind::Directory
&& let Some((parent, _)) = relative.rsplit_once('/')
{
transaction.create_dir_all(&format!("/{parent}")).await?;
}
operations.push(crate::kernel::Mutation::Restore { path, record });
}
None => {
if let Some(record) = current_record {
operations.push(crate::kernel::Mutation::Remove {
path,
expected_file_id: MetadataField::Value(record.file_id),
});
}
}
}
}
if !conflicts.is_empty() {
return Ok(WorkspacePathApply::Conflicted(conflicts));
}
if operations.is_empty() {
return Ok(WorkspacePathApply::NoChanges(current));
}
transaction
.checkout
.mutate(operations, crate::WorkBudget::UNBOUNDED, &cancellation)
.await
.map_err(WorkspaceError::engine)?;
Ok(match transaction.commit_with_permit(permit).await? {
TransactionCommit::Committed(generation) => WorkspacePathApply::Applied(generation),
TransactionCommit::AlreadyCommitted(generation) => {
WorkspacePathApply::AlreadyApplied(generation)
}
TransactionCommit::Conflict { actual } => WorkspacePathApply::Stale(actual),
TransactionCommit::Fenced => WorkspacePathApply::Fenced,
TransactionCommit::IdempotencyConflict => WorkspacePathApply::IdempotencyConflict,
})
}
pub async fn sync(&self) -> Result<WorkspaceSync<A, O>, WorkspaceError> {
Ok(WorkspaceSync {
generation: self.head().await?,
})
}
pub async fn fork(
&self,
destination: impl AsRef<str>,
options: ForkOptions<A, O>,
) -> Result<Self, WorkspaceError> {
if options.generation.workspace.id != self.id {
return Err(WorkspaceError::ForeignGeneration);
}
self.fork_measured(
destination,
options,
crate::WorkBudget::UNBOUNDED,
&crate::CancellationToken::new(),
)
.await
.map(|receipt| receipt.value)
}
pub(crate) async fn fork_measured(
&self,
destination: impl AsRef<str>,
options: ForkOptions<A, O>,
budget: crate::WorkBudget,
cancellation: &crate::CancellationToken,
) -> Result<crate::OperationReceipt<Self>, WorkspaceError> {
if options.generation.workspace.id != self.id {
return Err(WorkspaceError::ForeignGeneration);
}
let destination = WorkspaceName::new(destination)?;
Box::pin(self.volume.fs.fork_workspace_measured(
destination,
&options.generation,
options.idempotency_key,
budget,
cancellation,
))
.await
}
pub async fn begin_transaction(
&self,
idempotency_key: IdempotencyKey,
) -> Result<Transaction<A, O>, WorkspaceError> {
self.begin_transaction_measured(
idempotency_key,
crate::WorkBudget::UNBOUNDED,
&crate::CancellationToken::new(),
)
.await
.map(|receipt| receipt.value)
}
pub(crate) async fn begin_transaction_measured(
&self,
idempotency_key: IdempotencyKey,
budget: crate::WorkBudget,
cancellation: &crate::CancellationToken,
) -> Result<crate::OperationReceipt<Transaction<A, O>>, WorkspaceError> {
let head = self.head_measured(budget, cancellation).await?;
let transaction = self
.open_transaction_at_measured(
&head.value,
idempotency_key,
false,
head.work
.remaining(budget)
.map_err(WorkspaceError::engine)?,
cancellation,
)
.await?;
Ok(crate::OperationReceipt {
value: transaction.value,
work: head
.work
.checked_add(transaction.work)
.map_err(WorkspaceError::engine)?,
})
}
pub async fn begin_transaction_at(
&self,
generation: &Generation<A, O>,
idempotency_key: IdempotencyKey,
) -> Result<Transaction<A, O>, WorkspaceError> {
if generation.workspace.id != self.id {
return Err(WorkspaceError::ForeignGeneration);
}
let requires_rebase = self.head().await?.id != generation.id;
self.open_transaction_at(generation, idempotency_key, requires_rebase)
.await
}
pub(crate) async fn begin_pinned_transaction_measured(
&self,
generation: &Generation<A, O>,
idempotency_key: IdempotencyKey,
budget: crate::WorkBudget,
cancellation: &crate::CancellationToken,
) -> Result<crate::OperationReceipt<Transaction<A, O>>, WorkspaceError> {
if generation.workspace.id != self.id {
return Err(WorkspaceError::ForeignGeneration);
}
self.open_transaction_at_measured(generation, idempotency_key, false, budget, cancellation)
.await
}
async fn open_transaction_at(
&self,
generation: &Generation<A, O>,
idempotency_key: IdempotencyKey,
requires_rebase: bool,
) -> Result<Transaction<A, O>, WorkspaceError> {
self.open_transaction_at_measured(
generation,
idempotency_key,
requires_rebase,
crate::WorkBudget::UNBOUNDED,
&crate::CancellationToken::new(),
)
.await
.map(|receipt| receipt.value)
}
async fn open_transaction_at_measured(
&self,
generation: &Generation<A, O>,
idempotency_key: IdempotencyKey,
requires_rebase: bool,
budget: crate::WorkBudget,
cancellation: &crate::CancellationToken,
) -> Result<crate::OperationReceipt<Transaction<A, O>>, WorkspaceError> {
let receipt = self
.engine_checkout_measured(
GenerationSelector::Exact(generation.id),
CheckoutMode::tracking_transaction(),
budget,
cancellation,
)
.await?;
let mut checkout = receipt.value;
checkout.bind_authored_operation(idempotency_key.operation_id());
Ok(crate::OperationReceipt {
value: Transaction {
checkout,
workspace: self.clone(),
idempotency_key,
requires_rebase,
},
work: receipt.work,
})
}
pub async fn checkpoint(
&self,
label: impl AsRef<str>,
) -> Result<Checkpoint<A, O>, WorkspaceError> {
let label = WorkspaceName::new(label)?;
let generation = self.head().await?;
self.volume
.fs
.retain_workspace_generation(
&self.volume,
generation.id,
crate::kernel::RetentionKind::Checkpoint,
label.as_str().to_owned(),
)
.await?;
Ok(Checkpoint { label, generation })
}
pub async fn pin(
&self,
identity: impl AsRef<str>,
) -> Result<GenerationPin<A, O>, WorkspaceError> {
self.head().await?.pin(identity).await
}
pub async fn delete(
&self,
idempotency_key: IdempotencyKey,
) -> Result<WorkspaceDelete, WorkspaceError> {
self.volume
.fs
.delete_workspace_volume(&self.volume, idempotency_key.operation_id())
.await
}
pub async fn diff(
&self,
from: &Generation<A, O>,
to: &Generation<A, O>,
maximum_changes: u32,
) -> Result<ChangeSet<A, O>, WorkspaceError> {
if from.workspace.id != self.id || to.workspace.id != self.id {
return Err(WorkspaceError::ForeignGeneration);
}
let receipt = self
.volume
.diff_generations(
from.id,
to.id,
maximum_changes,
crate::WorkBudget::UNBOUNDED,
&crate::CancellationToken::new(),
)
.await
.map_err(WorkspaceError::engine)?;
if receipt.value.truncated {
return Err(WorkspaceError::JoinLimit);
}
Ok(ChangeSet {
from: from.clone(),
to: to.clone(),
changes: receipt.value,
work: receipt.work,
})
}
#[must_use]
pub fn join_into(&self, target: &Workspace<A, O>) -> JoinBuilder<A, O> {
JoinBuilder {
source: self.clone(),
target: target.clone(),
history: JoinHistory::Merge,
maximum_generations: 4_096,
maximum_changes: self.volume.config().limits.maximum_paths_per_batch,
maximum_conflicts: 1_024,
}
}
pub async fn live_rebase(
&self,
idempotency_key: IdempotencyKey,
maximum_generations: u32,
maximum_changes: u32,
maximum_conflicts: u32,
) -> Result<WorkspaceRebase<A, O>, WorkspaceError> {
let outcome = self
.volume
.fs
.live_rebase_workspace(crate::facade::WorkspaceRebaseRequest {
target: &self.volume,
operation_id: idempotency_key.operation_id(),
maximum_generations,
maximum_changes,
maximum_conflicts,
})
.await?;
let generation = |id| Generation {
workspace: self.clone(),
id,
};
Ok(match outcome {
crate::facade::WorkspaceJoinOutcome::Applied(id) => {
WorkspaceRebase::Rebased(generation(id))
}
crate::facade::WorkspaceJoinOutcome::AlreadyApplied(id) => {
WorkspaceRebase::AlreadyRebased(generation(id))
}
crate::facade::WorkspaceJoinOutcome::NoChanges(id) => {
WorkspaceRebase::Current(generation(id))
}
crate::facade::WorkspaceJoinOutcome::Stale(id) => {
WorkspaceRebase::Stale(generation(id))
}
crate::facade::WorkspaceJoinOutcome::Conflicted(conflicts, truncated) => {
WorkspaceRebase::Conflicted {
conflicts,
truncated,
}
}
crate::facade::WorkspaceJoinOutcome::Fenced => WorkspaceRebase::Fenced,
crate::facade::WorkspaceJoinOutcome::IdempotencyConflict => {
WorkspaceRebase::IdempotencyConflict
}
})
}
pub async fn read(&self, path: &str, maximum_bytes: u64) -> Result<Bytes, WorkspaceError> {
read_generation(self, GenerationSelector::Head, path, maximum_bytes).await
}
pub async fn read_range(
&self,
path: &str,
offset: u64,
length: u64,
) -> Result<Bytes, WorkspaceError> {
read_generation_range(self, GenerationSelector::Head, path, offset, length).await
}
pub async fn stat(&self, path: &str) -> Result<WorkspaceStat, WorkspaceError> {
stat_generation(self, GenerationSelector::Head, path).await
}
pub(crate) async fn stat_optional_measured(
&self,
path: &str,
budget: crate::WorkBudget,
cancellation: &crate::CancellationToken,
) -> Result<crate::OperationReceipt<Option<WorkspaceStat>>, WorkspaceError> {
stat_generation_optional_measured(
self,
GenerationSelector::Head,
path,
budget,
cancellation,
)
.await
}
pub(crate) async fn record_by_id(
&self,
file_id: FileId,
) -> Result<crate::kernel::FileRecord, WorkspaceError> {
let mut checkout = self
.engine_checkout(GenerationSelector::Head, CheckoutMode::read_only_pinned())
.await?;
checkout
.read_file_record_by_id(
file_id,
crate::WorkBudget::UNBOUNDED,
&crate::CancellationToken::new(),
)
.await
.map(|receipt| receipt.value)
.map_err(WorkspaceError::engine)
}
pub(crate) fn detached_record(
&self,
record: crate::kernel::FileRecord,
) -> crate::DetachedFile<A, O> {
crate::DetachedFile::from_record(self.volume.clone(), record)
}
pub async fn list_directory(
&self,
path: &str,
after: Option<&LogicalName>,
maximum_entries: u32,
) -> Result<WorkspaceDirectoryPage, WorkspaceError> {
list_generation_directory(self, GenerationSelector::Head, path, after, maximum_entries)
.await
}
pub(crate) async fn list_directory_measured(
&self,
path: &str,
after: Option<&LogicalName>,
maximum_entries: u32,
budget: crate::WorkBudget,
cancellation: &crate::CancellationToken,
) -> Result<crate::OperationReceipt<WorkspaceDirectoryPage>, WorkspaceError> {
list_generation_directory_measured(
self,
GenerationSelector::Head,
path,
after,
maximum_entries,
budget,
cancellation,
)
.await
}
pub async fn read_symbolic_link(&self, path: &str) -> Result<Bytes, WorkspaceError> {
read_generation_symbolic_link(self, GenerationSelector::Head, path).await
}
pub async fn plan_extents(
&self,
path: &str,
offset: u64,
length: u64,
maximum_spans: u32,
) -> Result<WorkspaceExtentPlan, WorkspaceError> {
plan_generation_extents(
self,
GenerationSelector::Head,
path,
offset,
length,
maximum_spans,
)
.await
}
pub async fn write_text(
&self,
path: &str,
text: &str,
) -> Result<TransactionCommit<A, O>, WorkspaceError> {
self.write(path, Bytes::copy_from_slice(text.as_bytes()))
.await
}
pub async fn write(
&self,
path: &str,
bytes: Bytes,
) -> Result<TransactionCommit<A, O>, WorkspaceError> {
let mut transaction = self.begin_transaction(IdempotencyKey::new()).await?;
transaction.write(path, bytes).await?;
transaction.commit().await
}
pub async fn remove(&self, path: &str) -> Result<TransactionCommit<A, O>, WorkspaceError> {
let mut transaction = self.begin_transaction(IdempotencyKey::new()).await?;
transaction.remove(path).await?;
transaction.commit().await
}
pub async fn copy(
&self,
source: &str,
destination: &str,
) -> Result<TransactionCommit<A, O>, WorkspaceError> {
let mut transaction = self.begin_transaction(IdempotencyKey::new()).await?;
transaction.copy(source, destination).await?;
transaction.commit().await
}
async fn read_selected(
&self,
selector: GenerationSelector,
path: &str,
maximum_bytes: u64,
) -> Result<Bytes, WorkspaceError> {
let mut checkout = self
.engine_checkout(selector, CheckoutMode::read_only_pinned())
.await?;
let path = customer_path(path, checkout.volume_config())?;
let lookup = checkout
.lookup_no_follow(
&path,
crate::WorkBudget::UNBOUNDED,
&crate::CancellationToken::new(),
)
.await
.map_err(WorkspaceError::engine)?
.value
.record
.ok_or(WorkspaceError::NotFound)?;
if lookup.kind != FileKind::Regular {
return Err(WorkspaceError::NotRegularFile);
}
let logical_bytes = regular_file_bytes(lookup)?;
if logical_bytes > maximum_bytes {
return Err(WorkspaceError::ReadLimitExceeded);
}
checkout
.read_file_range(
&path,
crate::ByteRange {
offset: 0,
length: logical_bytes,
},
crate::WorkBudget::UNBOUNDED,
&crate::CancellationToken::new(),
)
.await
.map(|receipt| receipt.value.bytes)
.map_err(WorkspaceError::engine)
}
pub(crate) async fn engine_checkout(
&self,
selector: GenerationSelector,
mode: CheckoutMode,
) -> Result<Checkout<A, O>, WorkspaceError> {
self.engine_checkout_measured(
selector,
mode,
crate::WorkBudget::UNBOUNDED,
&crate::CancellationToken::new(),
)
.await
.map(|receipt| receipt.value)
}
pub(crate) async fn engine_checkout_measured(
&self,
selector: GenerationSelector,
mode: CheckoutMode,
budget: crate::WorkBudget,
cancellation: &crate::CancellationToken,
) -> Result<crate::OperationReceipt<Checkout<A, O>>, WorkspaceError> {
self.volume
.checkout(selector, mode, budget, cancellation)
.await
.map_err(|failure| WorkspaceError::from(failure.error))
}
}
async fn read_generation<A: AsyncAuthorityStore, O: AsyncObjectStore>(
workspace: &Workspace<A, O>,
selector: GenerationSelector,
path: &str,
maximum_bytes: u64,
) -> Result<Bytes, WorkspaceError> {
workspace.read_selected(selector, path, maximum_bytes).await
}
async fn read_generation_range<A: AsyncAuthorityStore, O: AsyncObjectStore>(
workspace: &Workspace<A, O>,
selector: GenerationSelector,
path: &str,
offset: u64,
length: u64,
) -> Result<Bytes, WorkspaceError> {
let mut checkout = workspace
.engine_checkout(selector, CheckoutMode::read_only_pinned())
.await?;
let path = customer_path(path, checkout.volume_config())?;
checkout
.read_file_range(
&path,
crate::ByteRange { offset, length },
crate::WorkBudget::UNBOUNDED,
&crate::CancellationToken::new(),
)
.await
.map(|receipt| receipt.value.bytes)
.map_err(|failure| WorkspaceError::from(failure.error))
}
async fn stat_generation<A: AsyncAuthorityStore, O: AsyncObjectStore>(
workspace: &Workspace<A, O>,
selector: GenerationSelector,
path: &str,
) -> Result<WorkspaceStat, WorkspaceError> {
stat_generation_measured(
workspace,
selector,
path,
crate::WorkBudget::UNBOUNDED,
&crate::CancellationToken::new(),
)
.await
.map(|receipt| receipt.value)
}
async fn stat_generation_measured<A: AsyncAuthorityStore, O: AsyncObjectStore>(
workspace: &Workspace<A, O>,
selector: GenerationSelector,
path: &str,
budget: crate::WorkBudget,
cancellation: &crate::CancellationToken,
) -> Result<crate::OperationReceipt<WorkspaceStat>, WorkspaceError> {
let receipt =
stat_generation_optional_measured(workspace, selector, path, budget, cancellation).await?;
Ok(crate::OperationReceipt {
value: receipt.value.ok_or(WorkspaceError::NotFound)?,
work: receipt.work,
})
}
async fn stat_generation_optional_measured<A: AsyncAuthorityStore, O: AsyncObjectStore>(
workspace: &Workspace<A, O>,
selector: GenerationSelector,
path: &str,
budget: crate::WorkBudget,
cancellation: &crate::CancellationToken,
) -> Result<crate::OperationReceipt<Option<WorkspaceStat>>, WorkspaceError> {
let checkout = workspace
.engine_checkout_measured(
selector,
CheckoutMode::read_only_pinned(),
budget,
cancellation,
)
.await?;
let mut work = checkout.work;
let mut checkout = checkout.value;
let path = customer_path(path, checkout.volume_config())?;
let lookup = checkout
.lookup_no_follow_with_metadata(
&path,
work.remaining(budget).map_err(WorkspaceError::from)?,
cancellation,
)
.await
.map_err(|failure| WorkspaceError::from(failure.error))?;
work = work
.checked_add(lookup.work)
.map_err(WorkspaceError::from)?;
let Some(lookup) = lookup.value else {
return Ok(crate::OperationReceipt { value: None, work });
};
let logical_bytes = match &lookup.record.payload {
FilePayload::InlineRegular(bytes) => {
Some(u64::try_from(bytes.as_bytes().len()).map_err(WorkspaceError::engine)?)
}
FilePayload::Regular { logical_bytes, .. }
| FilePayload::SymbolicLink {
target_bytes: logical_bytes,
..
}
| FilePayload::ReparsePoint {
payload_bytes: logical_bytes,
..
} => Some(*logical_bytes),
FilePayload::Directory { .. } | FilePayload::Device { .. } | FilePayload::Empty => None,
};
Ok(crate::OperationReceipt {
value: Some(WorkspaceStat {
file_id: lookup.record.file_id,
kind: lookup.record.kind,
link_count: lookup.record.link_count,
logical_bytes,
metadata: WorkspaceMetadata::from_engine(lookup.metadata),
}),
work,
})
}
async fn list_generation_directory<A: AsyncAuthorityStore, O: AsyncObjectStore>(
workspace: &Workspace<A, O>,
selector: GenerationSelector,
path: &str,
after: Option<&LogicalName>,
maximum_entries: u32,
) -> Result<WorkspaceDirectoryPage, WorkspaceError> {
list_generation_directory_measured(
workspace,
selector,
path,
after,
maximum_entries,
crate::WorkBudget::UNBOUNDED,
&crate::CancellationToken::new(),
)
.await
.map(|receipt| receipt.value)
}
#[allow(clippy::too_many_arguments)]
async fn list_generation_directory_measured<A: AsyncAuthorityStore, O: AsyncObjectStore>(
workspace: &Workspace<A, O>,
selector: GenerationSelector,
path: &str,
after: Option<&LogicalName>,
maximum_entries: u32,
budget: crate::WorkBudget,
cancellation: &crate::CancellationToken,
) -> Result<crate::OperationReceipt<WorkspaceDirectoryPage>, WorkspaceError> {
let checkout = workspace
.engine_checkout_measured(
selector,
CheckoutMode::read_only_pinned(),
budget,
cancellation,
)
.await?;
let prior = checkout.work;
let mut checkout = checkout.value;
let path = customer_path(path, checkout.volume_config())?;
let page = checkout
.list_directory(
&path,
after,
maximum_entries,
prior.remaining(budget).map_err(WorkspaceError::from)?,
cancellation,
)
.await
.map_err(|failure| WorkspaceError::from(failure.error))?;
let work = prior.checked_add(page.work).map_err(WorkspaceError::from)?;
Ok(crate::OperationReceipt {
value: WorkspaceDirectoryPage {
entries: page
.value
.entries
.into_iter()
.map(|entry| WorkspaceDirectoryEntry {
name: entry.name,
file_id: entry.file_id,
kind: entry.kind,
})
.collect(),
has_more: page.value.has_more,
},
work,
})
}
async fn read_generation_symbolic_link<A: AsyncAuthorityStore, O: AsyncObjectStore>(
workspace: &Workspace<A, O>,
selector: GenerationSelector,
path: &str,
) -> Result<Bytes, WorkspaceError> {
let mut checkout = workspace
.engine_checkout(selector, CheckoutMode::read_only_pinned())
.await?;
let path = customer_path(path, checkout.volume_config())?;
checkout
.read_symbolic_link(
&path,
crate::WorkBudget::UNBOUNDED,
&crate::CancellationToken::new(),
)
.await
.map(|receipt| receipt.value)
.map_err(WorkspaceError::engine)
}
async fn plan_generation_extents<A: AsyncAuthorityStore, O: AsyncObjectStore>(
workspace: &Workspace<A, O>,
selector: GenerationSelector,
path: &str,
offset: u64,
length: u64,
maximum_spans: u32,
) -> Result<WorkspaceExtentPlan, WorkspaceError> {
let mut checkout = workspace
.engine_checkout(selector, CheckoutMode::read_only_pinned())
.await?;
let path = customer_path(path, checkout.volume_config())?;
let plan = checkout
.plan_file_extents(
&path,
crate::ByteRange { offset, length },
maximum_spans,
crate::WorkBudget::UNBOUNDED,
&crate::CancellationToken::new(),
)
.await
.map_err(WorkspaceError::engine)?
.value;
let spans = match plan {
Some(plan) => plan
.spans
.into_iter()
.map(|span| WorkspaceExtentSpan {
offset: span.offset,
length: span.length,
source_end: span.source_end,
kind: match span.kind {
ExtentKind::Hole => WorkspaceExtentKind::Hole,
ExtentKind::AllocatedZero => WorkspaceExtentKind::AllocatedZero,
ExtentKind::Content { .. } => WorkspaceExtentKind::Content,
},
})
.collect(),
None if length == 0 => Vec::new(),
None => vec![WorkspaceExtentSpan {
offset,
length,
source_end: offset
.checked_add(length)
.ok_or_else(|| WorkspaceError::engine("file range overflows"))?,
kind: WorkspaceExtentKind::Content,
}],
};
Ok(WorkspaceExtentPlan { spans })
}
pub struct Transaction<A, O> {
workspace: Workspace<A, O>,
checkout: Checkout<A, O>,
idempotency_key: IdempotencyKey,
requires_rebase: bool,
}
impl<A: AsyncAuthorityStore, O: AsyncObjectStore> Transaction<A, O> {
#[cfg(feature = "native-mount")]
pub(crate) fn for_checkout_candidate(
workspace: &Workspace<A, O>,
checkout: Checkout<A, O>,
) -> Self {
Self {
workspace: workspace.clone(),
checkout,
idempotency_key: IdempotencyKey::new(),
requires_rebase: false,
}
}
#[cfg(feature = "native-mount")]
pub(crate) fn into_checkout_candidate(self) -> Checkout<A, O> {
self.checkout
}
pub(crate) fn workspace_id(&self) -> WorkspaceId {
self.workspace.id()
}
pub(crate) async fn restore_record_measured(
&mut self,
path: &str,
record: crate::kernel::FileRecord,
budget: crate::WorkBudget,
cancellation: &crate::CancellationToken,
) -> Result<crate::WorkCounters, WorkspaceError> {
let path = customer_path(path, self.checkout.volume_config())?;
self.apply_engine_measured(
vec![crate::kernel::Mutation::Restore { path, record }],
budget,
cancellation,
)
.await
}
pub(crate) async fn create_dir_all_measured(
&mut self,
path: &str,
budget: crate::WorkBudget,
cancellation: &crate::CancellationToken,
) -> Result<crate::WorkCounters, WorkspaceError> {
let portable = PortablePath::parse(path, self.checkout.volume_config().limits)
.map_err(WorkspaceError::path)?;
let mut current = String::new();
let mut work = crate::WorkCounters::default();
for component in portable.components() {
cancellation.check().map_err(WorkspaceError::from)?;
current.push('/');
current.push_str(component);
let path = customer_path(¤t, self.checkout.volume_config())?;
let lookup = self
.checkout
.lookup_no_follow(
&path,
work.remaining(budget).map_err(WorkspaceError::from)?,
cancellation,
)
.await
.map_err(|failure| WorkspaceError::from(failure.error))?;
work = work
.checked_add(lookup.work)
.map_err(WorkspaceError::from)?;
match lookup.value.record {
Some(value) if value.kind == FileKind::Directory => {}
Some(_) => return Err(WorkspaceError::NotDirectory),
None => {
let applied = self
.apply_measured(
vec![AuthoredMutation::CreateDirectory {
path,
metadata: FileMetadata::default(),
}],
work.remaining(budget).map_err(WorkspaceError::from)?,
cancellation,
)
.await?;
work = work.checked_add(applied).map_err(WorkspaceError::from)?;
}
}
}
Ok(work)
}
pub(crate) async fn apply_authored_measured(
&mut self,
operation: AuthoredMutation,
budget: crate::WorkBudget,
cancellation: &crate::CancellationToken,
) -> Result<crate::WorkCounters, WorkspaceError> {
self.apply_measured(vec![operation], budget, cancellation)
.await
}
pub(crate) async fn apply_authored_bulk_measured(
&mut self,
operations: Vec<AuthoredMutation>,
budget: crate::WorkBudget,
cancellation: &crate::CancellationToken,
) -> Result<crate::WorkCounters, WorkspaceError> {
self.checkout
.apply_authored_bulk_transaction(operations, budget, cancellation)
.await
.map(|receipt| receipt.work)
.map_err(|failure| WorkspaceError::from(failure.error))
}
pub(crate) async fn create_directory_measured(
&mut self,
path: &str,
budget: crate::WorkBudget,
cancellation: &crate::CancellationToken,
) -> Result<crate::WorkCounters, WorkspaceError> {
let path = customer_path(path, self.checkout.volume_config())?;
self.apply_authored_measured(
AuthoredMutation::CreateDirectory {
path,
metadata: FileMetadata::default(),
},
budget,
cancellation,
)
.await
}
pub(crate) async fn create_file_measured(
&mut self,
path: &str,
bytes: Bytes,
budget: crate::WorkBudget,
cancellation: &crate::CancellationToken,
) -> Result<crate::WorkCounters, WorkspaceError> {
let path = customer_path(path, self.checkout.volume_config())?;
self.apply_authored_measured(
AuthoredMutation::CreateFile {
path,
bytes,
metadata: FileMetadata::default(),
},
budget,
cancellation,
)
.await
}
pub(crate) async fn write_range_measured(
&mut self,
path: &str,
offset: u64,
bytes: Bytes,
budget: crate::WorkBudget,
cancellation: &crate::CancellationToken,
) -> Result<crate::WorkCounters, WorkspaceError> {
let path = customer_path(path, self.checkout.volume_config())?;
self.apply_authored_measured(
AuthoredMutation::Write {
path,
offset,
bytes,
},
budget,
cancellation,
)
.await
}
pub(crate) async fn create_symbolic_link_measured(
&mut self,
path: &str,
target: Bytes,
budget: crate::WorkBudget,
cancellation: &crate::CancellationToken,
) -> Result<crate::WorkCounters, WorkspaceError> {
let path = customer_path(path, self.checkout.volume_config())?;
self.apply_authored_measured(
AuthoredMutation::CreateSymbolicLink {
path,
target,
metadata: FileMetadata::default(),
},
budget,
cancellation,
)
.await
}
pub(crate) async fn create_special_measured(
&mut self,
path: &str,
kind: FileKind,
device: Option<(u32, u32)>,
budget: crate::WorkBudget,
cancellation: &crate::CancellationToken,
) -> Result<crate::WorkCounters, WorkspaceError> {
let path = customer_path(path, self.checkout.volume_config())?;
let operation = match (kind, device) {
(FileKind::Fifo | FileKind::Socket, None) => AuthoredMutation::CreateEmptySpecial {
path,
kind,
metadata: FileMetadata::default(),
},
(FileKind::CharacterDevice | FileKind::BlockDevice, Some((major, minor))) => {
AuthoredMutation::CreateDevice {
path,
kind,
major,
minor,
metadata: FileMetadata::default(),
}
}
_ => return Err(WorkspaceError::IncompatibleWorkspace),
};
self.apply_authored_measured(operation, budget, cancellation)
.await
}
pub(crate) async fn set_metadata_measured(
&mut self,
path: &str,
metadata: FileMetadata,
budget: crate::WorkBudget,
cancellation: &crate::CancellationToken,
) -> Result<crate::WorkCounters, WorkspaceError> {
let path = customer_path(path, self.checkout.volume_config())?;
self.apply_authored_measured(
AuthoredMutation::SetMetadata { path, metadata },
budget,
cancellation,
)
.await
}
pub(crate) async fn preserve_file_identity_measured(
&mut self,
path: &str,
file_id: FileId,
budget: crate::WorkBudget,
cancellation: &crate::CancellationToken,
) -> Result<crate::WorkCounters, WorkspaceError> {
let path = customer_path(path, self.checkout.volume_config())?;
self.apply_authored_measured(
AuthoredMutation::Reidentify { path, file_id },
budget,
cancellation,
)
.await
}
pub(crate) async fn hard_link_measured(
&mut self,
source: &str,
destination: &str,
budget: crate::WorkBudget,
cancellation: &crate::CancellationToken,
) -> Result<crate::WorkCounters, WorkspaceError> {
let config = self.checkout.volume_config();
self.apply_authored_measured(
AuthoredMutation::HardLink {
source: customer_path(source, config)?,
destination: customer_path(destination, config)?,
},
budget,
cancellation,
)
.await
}
async fn apply_engine_measured(
&mut self,
operations: Vec<crate::kernel::Mutation>,
budget: crate::WorkBudget,
cancellation: &crate::CancellationToken,
) -> Result<crate::WorkCounters, WorkspaceError> {
self.checkout
.mutate(operations, budget, cancellation)
.await
.map(|receipt| receipt.work)
.map_err(|failure| WorkspaceError::from(failure.error))
}
pub async fn preserve_file_identity(
&mut self,
path: &str,
file_id: FileId,
) -> Result<(), WorkspaceError> {
let path = customer_path(path, self.checkout.volume_config())?;
self.apply(vec![AuthoredMutation::Reidentify { path, file_id }])
.await
}
pub async fn create_file(
&mut self,
path: &str,
bytes: Bytes,
metadata: FileMetadata,
) -> Result<(), WorkspaceError> {
let path = customer_path(path, self.checkout.volume_config())?;
self.apply(vec![AuthoredMutation::CreateFile {
path,
bytes,
metadata,
}])
.await
}
pub async fn stage_content<R: crate::kernel::AsyncBlobSource>(
&self,
source: &mut R,
maximum_source_bytes: u64,
) -> Result<crate::StagedContent, WorkspaceError> {
self.stage_content_measured(
source,
maximum_source_bytes,
crate::WorkBudget::UNBOUNDED,
&crate::CancellationToken::new(),
)
.await
.map(|receipt| receipt.value)
}
pub(crate) async fn stage_content_measured<R: crate::kernel::AsyncBlobSource>(
&self,
source: &mut R,
maximum_source_bytes: u64,
budget: crate::WorkBudget,
cancellation: &crate::CancellationToken,
) -> Result<crate::OperationReceipt<crate::StagedContent>, WorkspaceError> {
self.checkout
.stage_content(source, maximum_source_bytes, budget, cancellation)
.await
.map(|receipt| crate::OperationReceipt {
value: receipt.value,
work: receipt.work,
})
.map_err(WorkspaceError::engine)
}
pub async fn create_dir_all(&mut self, path: &str) -> Result<(), WorkspaceError> {
let portable = PortablePath::parse(path, self.checkout.volume_config().limits)
.map_err(WorkspaceError::path)?;
let mut current = String::new();
for component in portable.components() {
current.push('/');
current.push_str(component);
let path = customer_path(¤t, self.checkout.volume_config())?;
let existing = self
.checkout
.lookup_no_follow(
&path,
crate::WorkBudget::UNBOUNDED,
&crate::CancellationToken::new(),
)
.await
.map_err(WorkspaceError::engine)?
.value
.record;
match existing {
Some(value) if value.kind == FileKind::Directory => {}
Some(_) => return Err(WorkspaceError::NotDirectory),
None => {
self.apply(vec![AuthoredMutation::CreateDirectory {
path,
metadata: FileMetadata::default(),
}])
.await?;
}
}
}
Ok(())
}
pub async fn create_directory(&mut self, path: &str) -> Result<(), WorkspaceError> {
let path = customer_path(path, self.checkout.volume_config())?;
self.apply(vec![AuthoredMutation::CreateDirectory {
path,
metadata: FileMetadata::default(),
}])
.await
}
pub async fn create_symbolic_link(
&mut self,
path: &str,
target: Bytes,
) -> Result<(), WorkspaceError> {
let path = customer_path(path, self.checkout.volume_config())?;
self.apply(vec![AuthoredMutation::CreateSymbolicLink {
path,
target,
metadata: FileMetadata::default(),
}])
.await
}
pub async fn create_special(
&mut self,
path: &str,
kind: FileKind,
device: Option<(u32, u32)>,
) -> Result<(), WorkspaceError> {
let path = customer_path(path, self.checkout.volume_config())?;
let mutation = match (kind, device) {
(FileKind::Fifo | FileKind::Socket, None) => AuthoredMutation::CreateEmptySpecial {
path,
kind,
metadata: FileMetadata::default(),
},
(FileKind::CharacterDevice | FileKind::BlockDevice, Some((major, minor))) => {
AuthoredMutation::CreateDevice {
path,
kind,
major,
minor,
metadata: FileMetadata::default(),
}
}
_ => return Err(WorkspaceError::IncompatibleWorkspace),
};
self.apply(vec![mutation]).await
}
pub async fn write_text(&mut self, path: &str, text: &str) -> Result<(), WorkspaceError> {
self.write(path, Bytes::copy_from_slice(text.as_bytes()))
.await
}
pub async fn write(&mut self, path: &str, bytes: Bytes) -> Result<(), WorkspaceError> {
let path = customer_path(path, self.checkout.volume_config())?;
let existing = self
.checkout
.lookup_no_follow(
&path,
crate::WorkBudget::UNBOUNDED,
&crate::CancellationToken::new(),
)
.await
.map_err(WorkspaceError::engine)?
.value
.record;
let operations = match existing {
None => vec![AuthoredMutation::CreateFile {
path,
bytes,
metadata: FileMetadata::default(),
}],
Some(value) if value.kind == FileKind::Regular => vec![
AuthoredMutation::Resize {
path: path.clone(),
logical_bytes: 0,
},
AuthoredMutation::Write {
path,
offset: 0,
bytes,
},
],
Some(_) => return Err(WorkspaceError::NotRegularFile),
};
self.apply(operations).await
}
pub async fn write_staged(
&mut self,
path: &str,
parts: &[crate::StagedContent],
) -> Result<(), WorkspaceError> {
if parts.is_empty() {
return Err(WorkspaceError::EmptyContentSet);
}
let path = customer_path(path, self.checkout.volume_config())?;
let existing = self
.checkout
.lookup_no_follow(
&path,
crate::WorkBudget::UNBOUNDED,
&crate::CancellationToken::new(),
)
.await
.map_err(WorkspaceError::engine)?
.value
.record;
let mut operations = match existing {
None => vec![AuthoredMutation::CreateFile {
path: path.clone(),
bytes: Bytes::new(),
metadata: FileMetadata::default(),
}],
Some(record) if record.kind == FileKind::Regular => vec![AuthoredMutation::Resize {
path: path.clone(),
logical_bytes: 0,
}],
Some(_) => return Err(WorkspaceError::NotRegularFile),
};
let mut offset = 0_u64;
for part in parts {
if part.logical_bytes() != 0 {
operations.push(AuthoredMutation::WriteFromContent {
path: path.clone(),
offset,
content: *part,
});
}
offset = offset
.checked_add(part.logical_bytes())
.ok_or(WorkspaceError::ContentLengthOverflow)?;
}
self.apply(operations).await
}
pub async fn remove(&mut self, path: &str) -> Result<(), WorkspaceError> {
let path = customer_path(path, self.checkout.volume_config())?;
let existing = self
.checkout
.lookup_no_follow(
&path,
crate::WorkBudget::UNBOUNDED,
&crate::CancellationToken::new(),
)
.await
.map_err(WorkspaceError::engine)?
.value
.record
.ok_or(WorkspaceError::NotFound)?;
self.apply(vec![AuthoredMutation::Remove {
path,
expected_file_id: Some(existing.file_id),
}])
.await
}
pub async fn remove_if(
&mut self,
path: &str,
expected_file_id: crate::FileId,
) -> Result<(), WorkspaceError> {
let path = customer_path(path, self.checkout.volume_config())?;
self.apply(vec![AuthoredMutation::Remove {
path,
expected_file_id: Some(expected_file_id),
}])
.await
}
pub async fn copy(&mut self, source: &str, destination: &str) -> Result<(), WorkspaceError> {
let config = self.checkout.volume_config();
let source = customer_path(source, config)?;
let destination = customer_path(destination, config)?;
let source_record = self
.checkout
.lookup_no_follow(
&source,
crate::WorkBudget::UNBOUNDED,
&crate::CancellationToken::new(),
)
.await
.map_err(WorkspaceError::engine)?
.value
.record
.ok_or(WorkspaceError::NotFound)?;
let logical_bytes = regular_file_bytes(source_record)?;
let destination_record = self
.checkout
.lookup_no_follow(
&destination,
crate::WorkBudget::UNBOUNDED,
&crate::CancellationToken::new(),
)
.await
.map_err(WorkspaceError::engine)?
.value
.record;
let mut operations = match destination_record {
None => vec![AuthoredMutation::CreateFile {
path: destination.clone(),
bytes: Bytes::new(),
metadata: FileMetadata::default(),
}],
Some(record) if record.kind == FileKind::Regular => vec![AuthoredMutation::Resize {
path: destination.clone(),
logical_bytes: 0,
}],
Some(_) => return Err(WorkspaceError::NotRegularFile),
};
if logical_bytes != 0 {
operations.push(AuthoredMutation::CloneRange(crate::FileCloneRequest {
source,
source_offset: 0,
destination,
destination_offset: 0,
length: logical_bytes,
}));
}
self.apply(operations).await
}
pub async fn rename(&mut self, source: &str, destination: &str) -> Result<(), WorkspaceError> {
self.rename_with_replace(source, destination, true).await
}
pub async fn rename_with_replace(
&mut self,
source: &str,
destination: &str,
replace: bool,
) -> Result<(), WorkspaceError> {
let config = self.checkout.volume_config();
self.apply(vec![AuthoredMutation::Rename {
source: customer_path(source, config)?,
destination: customer_path(destination, config)?,
replace,
}])
.await
}
pub async fn hard_link(
&mut self,
source: &str,
destination: &str,
) -> Result<(), WorkspaceError> {
let config = self.checkout.volume_config();
self.apply(vec![AuthoredMutation::HardLink {
source: customer_path(source, config)?,
destination: customer_path(destination, config)?,
}])
.await
}
pub async fn write_range(
&mut self,
path: &str,
offset: u64,
bytes: Bytes,
) -> Result<(), WorkspaceError> {
let path = customer_path(path, self.checkout.volume_config())?;
self.apply(vec![AuthoredMutation::Write {
path,
offset,
bytes,
}])
.await
}
pub async fn resize(&mut self, path: &str, logical_bytes: u64) -> Result<(), WorkspaceError> {
let path = customer_path(path, self.checkout.volume_config())?;
self.apply(vec![AuthoredMutation::Resize {
path,
logical_bytes,
}])
.await
}
pub async fn zero_range(
&mut self,
path: &str,
range: crate::ByteRange,
allocated: bool,
extend: bool,
) -> Result<(), WorkspaceError> {
let path = customer_path(path, self.checkout.volume_config())?;
self.apply(vec![AuthoredMutation::ZeroRange {
path,
range,
allocated,
extend,
}])
.await
}
pub async fn preallocate(
&mut self,
path: &str,
range: crate::ByteRange,
keep_size: bool,
) -> Result<(), WorkspaceError> {
let path = customer_path(path, self.checkout.volume_config())?;
self.apply(vec![AuthoredMutation::Preallocate {
path,
range,
keep_size,
}])
.await
}
pub async fn clone_range(
&mut self,
source: &str,
source_offset: u64,
destination: &str,
destination_offset: u64,
length: u64,
) -> Result<(), WorkspaceError> {
let config = self.checkout.volume_config();
self.apply(vec![AuthoredMutation::CloneRange(
crate::FileCloneRequest {
source: customer_path(source, config)?,
source_offset,
destination: customer_path(destination, config)?,
destination_offset,
length,
},
)])
.await
}
pub async fn set_metadata(
&mut self,
path: &str,
metadata: FileMetadata,
) -> Result<(), WorkspaceError> {
let path = customer_path(path, self.checkout.volume_config())?;
self.apply(vec![AuthoredMutation::SetMetadata { path, metadata }])
.await
}
pub async fn commit(&mut self) -> Result<TransactionCommit<A, O>, WorkspaceError> {
self.commit_with_permit(crate::PublicationPermit::Unrestricted)
.await
}
pub async fn commit_with_permit(
&mut self,
permit: crate::PublicationPermit,
) -> Result<TransactionCommit<A, O>, WorkspaceError> {
self.commit_with_permit_measured(
permit,
crate::WorkBudget::UNBOUNDED,
&crate::CancellationToken::new(),
)
.await
.map(|receipt| receipt.value)
}
pub(crate) async fn commit_with_permit_measured(
&mut self,
permit: crate::PublicationPermit,
budget: crate::WorkBudget,
cancellation: &crate::CancellationToken,
) -> Result<crate::OperationReceipt<TransactionCommit<A, O>>, WorkspaceError> {
cancellation.check().map_err(WorkspaceError::from)?;
if self.requires_rebase {
let receipt = Box::pin(self.checkout.retry_stale_commit(
self.idempotency_key.operation_id(),
permit,
budget,
cancellation,
))
.await
.map_err(|failure| WorkspaceError::from(failure.error))?;
let work = receipt.work;
let Some(outcome) = receipt.value else {
let actual = Box::pin(self.workspace.head_measured(
work.remaining(budget).map_err(WorkspaceError::from)?,
cancellation,
))
.await?;
return Ok(crate::OperationReceipt {
value: TransactionCommit::Conflict {
actual: actual.value,
},
work: work
.checked_add(actual.work)
.map_err(WorkspaceError::from)?,
});
};
return Box::pin(self.commit_outcome_measured(outcome, work, budget, cancellation))
.await;
}
let receipt = Box::pin(self.checkout.commit_with_permit(
self.idempotency_key.operation_id(),
permit,
budget,
cancellation,
))
.await
.map_err(|failure| WorkspaceError::from(failure.error))?;
Box::pin(self.commit_outcome_measured(receipt.value, receipt.work, budget, cancellation))
.await
}
async fn commit_outcome_measured(
&self,
outcome: CheckoutCommitOutcome,
work: crate::WorkCounters,
budget: crate::WorkBudget,
cancellation: &crate::CancellationToken,
) -> Result<crate::OperationReceipt<TransactionCommit<A, O>>, WorkspaceError> {
let value = match outcome {
CheckoutCommitOutcome::Committed { generation_id, .. } => {
TransactionCommit::Committed(Generation {
workspace: self.workspace.clone(),
id: generation_id,
})
}
CheckoutCommitOutcome::AlreadyCommitted { generation_id, .. } => {
TransactionCommit::AlreadyCommitted(Generation {
workspace: self.workspace.clone(),
id: generation_id,
})
}
CheckoutCommitOutcome::Conflict { .. } => {
let actual = Box::pin(self.workspace.head_measured(
work.remaining(budget).map_err(WorkspaceError::from)?,
cancellation,
))
.await?;
return Ok(crate::OperationReceipt {
value: TransactionCommit::Conflict {
actual: actual.value,
},
work: work
.checked_add(actual.work)
.map_err(WorkspaceError::from)?,
});
}
CheckoutCommitOutcome::Fenced { .. } => TransactionCommit::Fenced,
CheckoutCommitOutcome::IdempotencyConflict { .. } => {
TransactionCommit::IdempotencyConflict
}
};
Ok(crate::OperationReceipt { value, work })
}
pub async fn rebase(
&mut self,
maximum_conflicts: u32,
) -> Result<TransactionRebase<A, O>, WorkspaceError> {
let decision = self
.checkout
.rebase_head(
maximum_conflicts,
crate::WorkBudget::UNBOUNDED,
&crate::CancellationToken::new(),
)
.await
.map_err(WorkspaceError::engine)?
.value;
Ok(match decision {
crate::kernel::RebaseDecision::Safe { generation } => {
self.requires_rebase = false;
TransactionRebase::Rebased(Generation {
workspace: self.workspace.clone(),
id: generation,
})
}
crate::kernel::RebaseDecision::Conflicted {
conflicts,
truncated,
} => TransactionRebase::Conflicted {
conflicts: conflicts
.into_iter()
.map(TransactionConflict::from_engine)
.collect(),
truncated,
},
})
}
async fn apply(&mut self, operations: Vec<AuthoredMutation>) -> Result<(), WorkspaceError> {
self.apply_measured(
operations,
crate::WorkBudget::UNBOUNDED,
&crate::CancellationToken::new(),
)
.await
.map(|_| ())
}
async fn apply_measured(
&mut self,
operations: Vec<AuthoredMutation>,
budget: crate::WorkBudget,
cancellation: &crate::CancellationToken,
) -> Result<crate::WorkCounters, WorkspaceError> {
self.checkout
.apply_authored_transaction(operations, budget, cancellation)
.await
.map(|receipt| receipt.work)
.map_err(|failure| WorkspaceError::from(failure.error))
}
}
pub enum TransactionCommit<A, O> {
Committed(Generation<A, O>),
AlreadyCommitted(Generation<A, O>),
Conflict {
actual: Generation<A, O>,
},
Fenced,
IdempotencyConflict,
}
pub enum TransactionRebase<A, O> {
Rebased(Generation<A, O>),
Conflicted {
conflicts: Vec<TransactionConflict>,
truncated: bool,
},
}
pub enum WorkspaceRebase<A, O> {
Rebased(Generation<A, O>),
AlreadyRebased(Generation<A, O>),
Current(Generation<A, O>),
Stale(Generation<A, O>),
Conflicted {
conflicts: Vec<MergeConflict>,
truncated: bool,
},
Fenced,
IdempotencyConflict,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct TransactionConflict {
pub region: TransactionConflictRegion,
pub usage: TransactionDependencyUse,
pub expected: Option<crate::Digest>,
pub actual: Option<crate::Digest>,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum TransactionConflictRegion {
FileRecord(crate::FileId),
Metadata(crate::FileId),
FileLength(crate::FileId),
ContentRange {
file_id: crate::FileId,
offset: u64,
length: u64,
},
SparseSeek {
file_id: crate::FileId,
offset: u64,
target: TransactionSparseSeek,
},
DirectoryName {
directory_id: crate::FileId,
name: LogicalName,
},
DirectoryRange {
directory_id: crate::FileId,
after: Option<LogicalName>,
maximum_entries: u32,
},
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum TransactionDependencyUse {
Observation,
Mutation,
ObservationAndMutation,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum TransactionSparseSeek {
Data,
Hole,
}
impl TransactionConflict {
fn from_engine(value: crate::kernel::RebaseConflict) -> Self {
use crate::kernel::{DependencyRegion, DependencyState, DependencyUse, ExtentSeekTarget};
let state = |value| match value {
DependencyState::Absent => None,
DependencyState::Present(digest) => Some(digest),
};
Self {
region: match value.region {
DependencyRegion::FileRecord(id) => TransactionConflictRegion::FileRecord(id),
DependencyRegion::Metadata(id) => TransactionConflictRegion::Metadata(id),
DependencyRegion::FileLength(id) => TransactionConflictRegion::FileLength(id),
DependencyRegion::ContentRange {
file_id,
offset,
length,
} => TransactionConflictRegion::ContentRange {
file_id,
offset,
length,
},
DependencyRegion::SparseSeek {
file_id,
offset,
target,
} => TransactionConflictRegion::SparseSeek {
file_id,
offset,
target: match target {
ExtentSeekTarget::Data => TransactionSparseSeek::Data,
ExtentSeekTarget::Hole => TransactionSparseSeek::Hole,
},
},
DependencyRegion::DirectoryName { directory_id, name } => {
TransactionConflictRegion::DirectoryName { directory_id, name }
}
DependencyRegion::DirectoryRange {
directory_id,
after,
maximum_entries,
} => TransactionConflictRegion::DirectoryRange {
directory_id,
after,
maximum_entries,
},
},
usage: match value.usage {
DependencyUse::Observation => TransactionDependencyUse::Observation,
DependencyUse::Mutation => TransactionDependencyUse::Mutation,
DependencyUse::ObservationAndMutation => {
TransactionDependencyUse::ObservationAndMutation
}
},
expected: state(value.expected),
actual: state(value.actual),
}
}
}
pub struct Generation<A, O> {
pub(crate) workspace: Workspace<A, O>,
pub(crate) id: GenerationId,
}
pub struct WorkspaceSync<A, O> {
generation: Generation<A, O>,
}
impl<A, O> WorkspaceSync<A, O> {
#[must_use]
pub const fn generation(&self) -> &Generation<A, O> {
&self.generation
}
#[must_use]
pub fn into_generation(self) -> Generation<A, O> {
self.generation
}
}
pub struct Checkpoint<A, O> {
label: WorkspaceName,
generation: Generation<A, O>,
}
pub struct GenerationPin<A, O> {
identity: WorkspaceName,
generation: Generation<A, O>,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum WorkspaceDelete {
Deleted,
AlreadyDeleted,
Conflict,
IdempotencyConflict,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct WorkspaceStat {
pub file_id: crate::FileId,
pub kind: FileKind,
pub link_count: u64,
pub logical_bytes: Option<u64>,
pub metadata: WorkspaceMetadata,
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct WorkspaceMetadata {
pub posix_mode: Option<u32>,
pub posix_uid: Option<u32>,
pub posix_gid: Option<u32>,
pub posix_flags: Option<u64>,
pub windows_attributes: Option<u32>,
pub created_ns: Option<i64>,
pub modified_ns: Option<i64>,
pub accessed_ns: Option<i64>,
pub changed_ns: Option<i64>,
pub has_named_attributes: bool,
pub has_acl: bool,
pub has_security_descriptor: bool,
}
impl WorkspaceMetadata {
pub(crate) fn from_engine(metadata: FileMetadata) -> Self {
Self {
posix_mode: metadata_value(metadata.posix_mode),
posix_uid: metadata_value(metadata.posix_uid),
posix_gid: metadata_value(metadata.posix_gid),
posix_flags: metadata_value(metadata.posix_flags),
windows_attributes: metadata_value(metadata.windows_attributes),
created_ns: metadata_value(metadata.created_ns),
modified_ns: metadata_value(metadata.modified_ns),
accessed_ns: metadata_value(metadata.accessed_ns),
changed_ns: metadata_value(metadata.changed_ns),
has_named_attributes: matches!(metadata.named_attributes, MetadataField::Value(_)),
has_acl: matches!(metadata.acl, MetadataField::Value(_)),
has_security_descriptor: matches!(
metadata.security_descriptor,
MetadataField::Value(_)
),
}
}
}
const fn metadata_value<T: Copy>(field: MetadataField<T>) -> Option<T> {
match field {
MetadataField::Unavailable => None,
MetadataField::Value(value) => Some(value),
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct WorkspaceDirectoryEntry {
pub name: LogicalName,
pub file_id: crate::FileId,
pub kind: FileKind,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct WorkspaceDirectoryPage {
pub entries: Vec<WorkspaceDirectoryEntry>,
pub has_more: bool,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum WorkspaceExtentKind {
Hole,
AllocatedZero,
Content,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct WorkspaceExtentSpan {
pub offset: u64,
pub length: u64,
pub source_end: u64,
pub kind: WorkspaceExtentKind,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct WorkspaceExtentPlan {
pub spans: Vec<WorkspaceExtentSpan>,
}
pub struct ChangeSet<A, O> {
from: Generation<A, O>,
to: Generation<A, O>,
changes: GenerationDiff,
work: crate::WorkCounters,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ChangedPath {
pub path: NamespacePath,
pub before: Option<crate::kernel::FileRecord>,
pub after: Option<crate::kernel::FileRecord>,
}
impl<A, O> ChangeSet<A, O> {
#[must_use]
pub const fn from(&self) -> &Generation<A, O> {
&self.from
}
#[must_use]
pub const fn to(&self) -> &Generation<A, O> {
&self.to
}
#[must_use]
pub const fn changes(&self) -> &GenerationDiff {
&self.changes
}
#[must_use]
pub const fn work(&self) -> crate::WorkCounters {
self.work
}
}
fn merge_binding_endpoint(
paths: &mut BTreeMap<
NamespacePath,
(
Option<crate::kernel::FileRecord>,
Option<crate::kernel::FileRecord>,
),
>,
directories: &BTreeMap<FileId, Vec<(NamespacePath, crate::kernel::FileRecord)>>,
records: &BTreeMap<FileId, crate::kernel::FileRecord>,
change: &crate::DirectoryBindingChange,
limits: crate::model::VolumeLimits,
before: bool,
) -> Result<(), WorkspaceError> {
let Some(directories) = directories.get(&change.directory_id) else {
return Ok(());
};
let binding = if before {
change.before.as_ref()
} else {
change.after.as_ref()
};
let record = binding.and_then(|entry| records.get(&entry.file_id).copied());
for (directory, _) in directories {
let path = child_namespace_path(directory, change.name.clone(), limits)?;
let endpoints = paths.entry(path).or_default();
if before {
endpoints.0 = record;
} else {
endpoints.1 = record;
}
}
Ok(())
}
struct NamespaceRecordSearch {
records: BTreeMap<FileId, Vec<(NamespacePath, crate::kernel::FileRecord)>>,
complete: bool,
}
impl<A: AsyncAuthorityStore, O: AsyncObjectStore> ChangeSet<A, O> {
pub async fn changed_paths(
&self,
maximum_entries: u32,
) -> Result<Vec<ChangedPath>, WorkspaceError> {
self.changed_paths_bounded(
maximum_entries,
crate::WorkBudget::UNBOUNDED,
&crate::CancellationToken::new(),
)
.await
.map(|receipt| receipt.value)
}
#[allow(clippy::too_many_lines)]
pub async fn changed_paths_bounded(
&self,
maximum_entries: u32,
budget: crate::WorkBudget,
cancellation: &crate::CancellationToken,
) -> Result<crate::OperationReceipt<Vec<ChangedPath>>, WorkspaceError> {
if self.changes.truncated {
return Err(WorkspaceError::ChangedPathLimit);
}
let file_ids: BTreeSet<_> = self
.changes
.files
.iter()
.map(|change| change.file_id)
.chain(
self.changes
.bindings
.iter()
.map(|change| change.directory_id),
)
.chain(self.changes.bindings.iter().flat_map(|change| {
change
.before
.iter()
.chain(change.after.iter())
.map(|entry| entry.file_id)
}))
.collect();
self.work.verify(budget).map_err(WorkspaceError::engine)?;
let before_receipt = self
.from
.namespace_records_for_file_ids_bounded(
file_ids.iter().copied(),
maximum_entries,
self.work
.remaining(budget)
.map_err(WorkspaceError::engine)?,
cancellation,
)
.await?;
let mut work = self
.work
.checked_add(before_receipt.work)
.map_err(WorkspaceError::engine)?;
let after_receipt = self
.to
.namespace_records_for_file_ids_bounded(
file_ids.iter().copied(),
maximum_entries,
work.remaining(budget).map_err(WorkspaceError::engine)?,
cancellation,
)
.await?;
work = work
.checked_add(after_receipt.work)
.map_err(WorkspaceError::engine)?;
let before = before_receipt.value;
let after = after_receipt.value;
if !before.complete || !after.complete {
return Err(WorkspaceError::ChangedPathLimit);
}
let before = before.records;
let after = after.records;
let mut paths = BTreeMap::<NamespacePath, (Option<_>, Option<_>)>::new();
let before_records: BTreeMap<_, _> = before
.iter()
.filter_map(|(&file_id, records)| records.first().map(|(_, record)| (file_id, *record)))
.collect();
let after_records: BTreeMap<_, _> = after
.iter()
.filter_map(|(&file_id, records)| records.first().map(|(_, record)| (file_id, *record)))
.collect();
for records in before.values() {
for (path, record) in records {
paths.entry(path.clone()).or_default().0 = Some(*record);
}
}
for records in after.values() {
for (path, record) in records {
paths.entry(path.clone()).or_default().1 = Some(*record);
}
}
for change in &self.changes.bindings {
merge_binding_endpoint(
&mut paths,
&before,
&before_records,
change,
self.from.workspace.volume.config.limits,
true,
)?;
merge_binding_endpoint(
&mut paths,
&after,
&after_records,
change,
self.to.workspace.volume.config.limits,
false,
)?;
}
let paths = paths
.into_iter()
.filter(|(_, (before, after))| match (before, after) {
(Some(before), Some(after)) => {
before.kind != after.kind
|| before.metadata != after.metadata
|| (before.kind != FileKind::Directory && before.payload != after.payload)
}
(None, None) => false,
_ => true,
})
.map(|(path, (before, after))| ChangedPath {
path,
before,
after,
})
.collect();
Ok(crate::OperationReceipt { value: paths, work })
}
pub async fn compose(
&self,
next: &ChangeSet<A, O>,
maximum_changes: u32,
) -> Result<ChangeSet<A, O>, WorkspaceError> {
if self.to.id != next.from.id
|| self.to.workspace.id != next.from.workspace.id
|| !self
.to
.workspace
.volume
.fs
.same_deployment(&next.from.workspace.volume.fs)
{
return Err(WorkspaceError::ChangeSetContinuity);
}
self.from
.workspace
.diff(&self.from, &next.to, maximum_changes)
.await
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum JoinHistory {
Merge,
Rebase,
Squash,
CherryPick,
}
pub struct JoinBuilder<A, O> {
source: Workspace<A, O>,
target: Workspace<A, O>,
history: JoinHistory,
maximum_generations: u32,
maximum_changes: u32,
maximum_conflicts: u32,
}
impl<A, O> JoinBuilder<A, O> {
fn validate_bounds(&self) -> Result<(), WorkspaceError> {
if self.maximum_generations == 0 || self.maximum_changes == 0 || self.maximum_conflicts == 0
{
Err(WorkspaceError::JoinLimit)
} else {
Ok(())
}
}
#[must_use]
pub const fn history(mut self, history: JoinHistory) -> Self {
self.history = history;
self
}
#[must_use]
pub const fn bounds(
mut self,
maximum_generations: u32,
maximum_changes: u32,
maximum_conflicts: u32,
) -> Self {
self.maximum_generations = maximum_generations;
self.maximum_changes = maximum_changes;
self.maximum_conflicts = maximum_conflicts;
self
}
}
impl<A: AsyncAuthorityStore, O: AsyncObjectStore> JoinBuilder<A, O> {
pub async fn plan(self) -> Result<JoinPlan<A, O>, WorkspaceError> {
self.validate_bounds()?;
let source_head = self.source.head().await?;
let (target_head_id, target_authority_head) = self
.target
.volume
.fs
.workspace_head_state(&self.target.volume)
.await?;
let target_head = Generation {
workspace: self.target.clone(),
id: target_head_id,
};
self.plan_generations(source_head, target_head, target_authority_head)
.await
}
pub async fn plan_pinned(
self,
source_head: Generation<A, O>,
target_head: Generation<A, O>,
) -> Result<JoinPlan<A, O>, WorkspaceError> {
self.validate_bounds()?;
if source_head.workspace.id != self.source.id || target_head.workspace.id != self.target.id
{
return Err(WorkspaceError::ForeignGeneration);
}
let (current_target, target_authority_head) = self
.target
.volume
.fs
.workspace_head_state(&self.target.volume)
.await?;
if current_target != target_head.id {
return Err(WorkspaceError::StaleGeneration);
}
self.plan_generations(source_head, target_head, target_authority_head)
.await
}
pub(crate) async fn plan_generations(
self,
source_head: Generation<A, O>,
target_head: Generation<A, O>,
target_authority_head: crate::Head,
) -> Result<JoinPlan<A, O>, WorkspaceError> {
self.validate_bounds()?;
if source_head.workspace.id != self.source.id || target_head.workspace.id != self.target.id
{
return Err(WorkspaceError::ForeignGeneration);
}
let base = self
.source
.volume
.fs
.workspace_common_ancestor(&source_head, &target_head, self.maximum_generations)
.await?;
Ok(JoinPlan {
source: self.source,
target: self.target,
source_head,
target_head,
target_authority_head,
base,
history: self.history,
maximum_generations: self.maximum_generations,
maximum_changes: self.maximum_changes,
maximum_conflicts: self.maximum_conflicts,
})
}
}
pub struct JoinPlan<A, O> {
source: Workspace<A, O>,
target: Workspace<A, O>,
source_head: Generation<A, O>,
target_head: Generation<A, O>,
target_authority_head: crate::Head,
base: crate::facade::WorkspaceCommonAncestor,
history: JoinHistory,
maximum_generations: u32,
maximum_changes: u32,
maximum_conflicts: u32,
}
impl<A, O> JoinPlan<A, O> {
#[must_use]
pub const fn source_head(&self) -> GenerationId {
self.source_head.id
}
#[must_use]
pub const fn target_head(&self) -> GenerationId {
self.target_head.id
}
#[must_use]
pub const fn common_ancestor(&self) -> GenerationId {
self.base.id
}
#[must_use]
#[cfg(not(target_arch = "wasm32"))]
pub(crate) const fn common_ancestor_volume(&self) -> VolumeId {
self.base.volume_id
}
#[must_use]
pub const fn source(&self) -> &Workspace<A, O> {
&self.source
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct ApplyOptions {
pub if_target: GenerationId,
pub idempotency_key: IdempotencyKey,
}
pub enum JoinOutcome<A, O> {
Applied(Generation<A, O>),
AlreadyApplied(Generation<A, O>),
NoChanges(Generation<A, O>),
StaleTarget(Generation<A, O>),
Conflicted {
conflicts: Vec<MergeConflict>,
truncated: bool,
},
Fenced,
IdempotencyConflict,
}
pub(crate) enum WorkspaceRestoreOutcome {
Restored(GenerationId),
AlreadyRestored(GenerationId),
Current(GenerationId),
Stale(GenerationId),
Fenced,
IdempotencyConflict,
}
pub enum WorkspaceRestore<A, O> {
Restored(Generation<A, O>),
AlreadyRestored(Generation<A, O>),
Current(Generation<A, O>),
Stale(Generation<A, O>),
Fenced,
IdempotencyConflict,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct WorkspacePathConflict {
pub path: String,
pub kind: ConflictKind,
}
pub enum WorkspacePathApply<A, O> {
Applied(Generation<A, O>),
AlreadyApplied(Generation<A, O>),
NoChanges(Generation<A, O>),
Conflicted(Vec<WorkspacePathConflict>),
Stale(Generation<A, O>),
Fenced,
IdempotencyConflict,
}
#[derive(Debug, Error)]
pub enum DrivenJoinError {
#[error(transparent)]
Workspace(#[from] WorkspaceError),
#[error(transparent)]
Resolution(#[from] MergePlanResolutionError),
#[error("merge candidate does not match the immutable workspace join plan")]
StaleCandidate,
#[error("merge resolution for {0:?} is not representable by this join")]
UnsupportedResolution(ConflictKey),
}
impl<A: AsyncAuthorityStore, O: AsyncObjectStore> JoinPlan<A, O> {
#[cfg(not(target_arch = "wasm32"))]
pub(crate) async fn source_changes(
&self,
) -> Result<crate::FsReceipt<GenerationDiff>, WorkspaceError> {
self.source
.volume
.fs
.workspace_join_changes(self.base.id, &self.source_head, self.maximum_changes)
.await
}
pub async fn apply(&self, options: ApplyOptions) -> Result<JoinOutcome<A, O>, WorkspaceError> {
self.apply_with_permit(options, crate::PublicationPermit::Unrestricted)
.await
}
pub async fn apply_with_permit(
&self,
options: ApplyOptions,
permit: crate::PublicationPermit,
) -> Result<JoinOutcome<A, O>, WorkspaceError> {
self.apply_resolutions(options, BTreeMap::new(), permit)
.await
}
pub async fn describe_conflicts(
&self,
conflicts: &[MergeConflict],
truncated: bool,
) -> Result<MergePlan, WorkspaceError> {
self.typed_merge_plan(conflicts.to_vec(), truncated).await
}
pub async fn apply_with_drivers<C: MergeResolutionCache>(
&self,
options: ApplyOptions,
registry: &MergeDriverRegistry,
cache: &mut C,
replanning: bool,
) -> Result<JoinOutcome<A, O>, DrivenJoinError> {
self.apply_with_drivers_and_permit(
options,
registry,
cache,
replanning,
crate::PublicationPermit::Unrestricted,
)
.await
}
pub async fn apply_with_drivers_and_permit<C: MergeResolutionCache>(
&self,
options: ApplyOptions,
registry: &MergeDriverRegistry,
cache: &mut C,
replanning: bool,
permit: crate::PublicationPermit,
) -> Result<JoinOutcome<A, O>, DrivenJoinError> {
let initial = self.apply_with_permit(options, permit).await?;
let JoinOutcome::Conflicted {
conflicts,
truncated,
} = initial
else {
return Ok(initial);
};
let plan = self.typed_merge_plan(conflicts, truncated).await?;
let candidate = resolve_merge_plan(plan, registry, cache, replanning)?;
self.apply_candidate_with_permit(options, &candidate, permit)
.await
}
pub async fn apply_sides(
&self,
options: ApplyOptions,
selections: BTreeMap<MergeConflict, ConflictSide>,
) -> Result<JoinOutcome<A, O>, DrivenJoinError> {
let initial = self.apply(options).await?;
let JoinOutcome::Conflicted {
conflicts,
truncated,
} = initial
else {
return Ok(initial);
};
if truncated
|| conflicts.len() != selections.len()
|| conflicts
.iter()
.any(|conflict| !selections.contains_key(conflict))
{
return Err(DrivenJoinError::StaleCandidate);
}
let resolutions = selections
.into_iter()
.map(|(conflict, side)| {
let side = match side {
ConflictSide::Base => crate::kernel::MergeConflictSide::Base,
ConflictSide::Ours => crate::kernel::MergeConflictSide::Ours,
ConflictSide::Theirs => crate::kernel::MergeConflictSide::Theirs,
};
(
conflict,
crate::kernel::MergeConflictResolution::Select(side),
)
})
.collect();
self.apply_resolutions(options, resolutions, crate::PublicationPermit::Unrestricted)
.await
.map_err(Into::into)
}
pub async fn apply_candidate(
&self,
options: ApplyOptions,
candidate: &UnpublishedMergeCandidate,
) -> Result<JoinOutcome<A, O>, DrivenJoinError> {
self.apply_candidate_with_permit(options, candidate, crate::PublicationPermit::Unrestricted)
.await
}
pub async fn apply_candidate_with_permit(
&self,
options: ApplyOptions,
candidate: &UnpublishedMergeCandidate,
permit: crate::PublicationPermit,
) -> Result<JoinOutcome<A, O>, DrivenJoinError> {
let conflict_keys = candidate
.plan
.conflicts
.iter()
.map(|conflict| &conflict.key)
.collect::<BTreeSet<_>>();
if candidate.plan.base != self.base.id
|| candidate.plan.ours != self.target_head.id
|| candidate.plan.theirs != self.source_head.id
|| candidate.plan.truncated
|| candidate.plan.conflicts.len() != candidate.resolutions.len()
|| conflict_keys.len() != candidate.plan.conflicts.len()
|| !candidate.resolutions.keys().eq(conflict_keys.into_iter())
{
return Err(DrivenJoinError::StaleCandidate);
}
let mut resolutions = BTreeMap::new();
for conflict in &candidate.plan.conflicts {
let resolution = candidate
.resolutions
.get(&conflict.key)
.ok_or(DrivenJoinError::StaleCandidate)?;
if !resolution_matches_conflict(&conflict.key, resolution) {
return Err(DrivenJoinError::UnsupportedResolution(conflict.key.clone()));
}
let resolution = match resolution {
MergeResolution::Select(ConflictSide::Base) => {
crate::kernel::MergeConflictResolution::Select(
crate::kernel::MergeConflictSide::Base,
)
}
MergeResolution::Select(ConflictSide::Ours) => {
crate::kernel::MergeConflictResolution::Select(
crate::kernel::MergeConflictSide::Ours,
)
}
MergeResolution::Select(ConflictSide::Theirs) => {
crate::kernel::MergeConflictResolution::Select(
crate::kernel::MergeConflictSide::Theirs,
)
}
MergeResolution::Text(text) => crate::kernel::MergeConflictResolution::File(Some(
self.stage_regular_resolution(
&conflict.key,
Bytes::copy_from_slice(text.as_bytes()),
)
.await?,
)),
MergeResolution::Binary(bytes) => {
crate::kernel::MergeConflictResolution::File(Some(
self.stage_regular_resolution(&conflict.key, Bytes::copy_from_slice(bytes))
.await?,
))
}
MergeResolution::Metadata(bytes) => {
let file_id = conflict_file_id(&conflict.key).ok_or_else(|| {
DrivenJoinError::UnsupportedResolution(conflict.key.clone())
})?;
let mut record = self
.target
.volume
.fs
.workspace_conflict_record(
&self.target.volume,
self.target_head.id,
file_id,
)
.await?
.ok_or(WorkspaceError::InvalidMergeResolution)?;
record.metadata = self
.target
.volume
.fs
.stage_merge_metadata(&self.target.volume, bytes)
.await?;
crate::kernel::MergeConflictResolution::File(Some(record))
}
MergeResolution::Binding(file_id) => {
crate::kernel::MergeConflictResolution::Binding(*file_id)
}
MergeResolution::Unresolved => {
return Err(DrivenJoinError::UnsupportedResolution(conflict.key.clone()));
}
};
resolutions.insert(
kernel_conflict(
&conflict.key,
self.target.volume.config.limits.maximum_component_bytes,
)?,
resolution,
);
}
self.apply_resolutions(options, resolutions, permit)
.await
.map_err(Into::into)
}
async fn stage_regular_resolution(
&self,
key: &ConflictKey,
bytes: Bytes,
) -> Result<crate::kernel::FileRecord, DrivenJoinError> {
let file_id = conflict_file_id(key)
.ok_or_else(|| DrivenJoinError::UnsupportedResolution(key.clone()))?;
let record = self
.target
.volume
.fs
.workspace_conflict_record(&self.target.volume, self.target_head.id, file_id)
.await?
.ok_or(WorkspaceError::InvalidMergeResolution)?;
self.target
.volume
.fs
.stage_merge_regular_record(&self.target.volume, record, bytes)
.await
.map_err(Into::into)
}
async fn typed_merge_plan(
&self,
conflicts: Vec<MergeConflict>,
truncated: bool,
) -> Result<MergePlan, WorkspaceError> {
let mut typed = Vec::with_capacity(conflicts.len());
for conflict in conflicts {
typed.push(self.conflict_view(conflict).await?);
}
Ok(MergePlan {
base: self.base.id,
ours: self.target_head.id,
theirs: self.source_head.id,
conflicts: typed,
truncated,
})
}
async fn conflict_view(
&self,
conflict: MergeConflict,
) -> Result<crate::ConflictView, WorkspaceError> {
let MergeConflict::File(file_id) = conflict else {
let mut view = binding_conflict_view(self, conflict.clone());
if let MergeConflict::Binding { directory_id, name } = conflict
&& let Some(directory) = self.conflict_path(directory_id).await?
&& let Ok(name) = logical_name_text(&name)
{
view.path = Some(if directory == "/" {
format!("/{name}")
} else {
format!("{directory}/{name}")
});
}
return Ok(view);
};
let path = self.conflict_path(file_id).await?;
let fs = &self.target.volume.fs;
let base = fs
.workspace_conflict_record(&self.target.volume, self.base.id, file_id)
.await?;
let ours = fs
.workspace_conflict_record(&self.target.volume, self.target_head.id, file_id)
.await?;
let theirs = fs
.workspace_conflict_record(&self.target.volume, self.source_head.id, file_id)
.await?;
let fallback = |generation, record: Option<crate::kernel::FileRecord>| match record {
Some(_) => ConflictValue::Record {
generation,
file_id,
},
None => ConflictValue::Absent,
};
let Some((base_record, ours_record, theirs_record)) = base
.zip(ours)
.zip(theirs)
.map(|((base, ours), theirs)| (base, ours, theirs))
else {
return Ok(crate::ConflictView {
key: ConflictKey::File(file_id),
path,
kind: ConflictKind::Record,
base: fallback(self.base.id, base),
ours: fallback(self.target_head.id, ours),
theirs: fallback(self.source_head.id, theirs),
});
};
if base_record.link_count != ours_record.link_count
|| base_record.link_count != theirs_record.link_count
{
return Ok(record_conflict_view(
self,
file_id,
path,
ConflictKind::HardLink,
));
}
if base_record.metadata != ours_record.metadata
|| base_record.metadata != theirs_record.metadata
{
let values = [base_record, ours_record, theirs_record];
let mut metadata = Vec::with_capacity(3);
for record in values {
metadata.push(ConflictValue::Metadata(
fs.workspace_conflict_metadata(&self.target.volume, record)
.await?,
));
}
return Ok(crate::ConflictView {
key: ConflictKey::Metadata(file_id),
path,
kind: ConflictKind::Metadata,
base: metadata.remove(0),
ours: metadata.remove(0),
theirs: metadata.remove(0),
});
}
if let Some(view) = self
.regular_conflict_view(
file_id,
path.clone(),
[base_record, ours_record, theirs_record],
)
.await?
{
return Ok(view);
}
let kind = match base_record.kind {
FileKind::Directory => ConflictKind::Directory,
FileKind::SymbolicLink => ConflictKind::SymbolicLink,
_ => ConflictKind::Special,
};
Ok(record_conflict_view(self, file_id, path, kind))
}
async fn conflict_path(&self, file_id: FileId) -> Result<Option<String>, WorkspaceError> {
let maximum = self.maximum_changes.max(1);
let mut paths = self.target_head.paths_for_file_id(file_id, maximum).await?;
if paths.is_empty() {
paths = self.source_head.paths_for_file_id(file_id, maximum).await?;
}
Ok(paths.into_iter().next())
}
async fn regular_conflict_view(
&self,
file_id: FileId,
path: Option<String>,
records: [crate::kernel::FileRecord; 3],
) -> Result<Option<crate::ConflictView>, WorkspaceError> {
if records
.iter()
.any(|record| record.kind != FileKind::Regular)
{
return Ok(None);
}
let mut bodies = Vec::with_capacity(3);
for record in records {
let Some(bytes) = self
.target
.volume
.fs
.workspace_conflict_bytes(&self.target.volume, record)
.await?
else {
return Ok(Some(record_conflict_view(
self,
file_id,
path,
ConflictKind::Binary,
)));
};
bodies.push(bytes);
}
let text = bodies
.iter()
.all(|bytes| std::str::from_utf8(bytes).is_ok());
let mut values = bodies.into_iter().map(|bytes| {
if text {
ConflictValue::Text(String::from_utf8(bytes.to_vec()).unwrap_or_default())
} else {
ConflictValue::Binary(bytes.to_vec())
}
});
Ok(Some(crate::ConflictView {
key: ConflictKey::File(file_id),
path,
kind: if text {
ConflictKind::Text
} else {
ConflictKind::Binary
},
base: values
.next()
.ok_or(WorkspaceError::InvalidMergeResolution)?,
ours: values
.next()
.ok_or(WorkspaceError::InvalidMergeResolution)?,
theirs: values
.next()
.ok_or(WorkspaceError::InvalidMergeResolution)?,
}))
}
async fn apply_resolutions(
&self,
options: ApplyOptions,
resolutions: BTreeMap<MergeConflict, crate::kernel::MergeConflictResolution>,
permit: crate::PublicationPermit,
) -> Result<JoinOutcome<A, O>, WorkspaceError> {
if options.if_target != self.target_head.id {
return Ok(JoinOutcome::StaleTarget(self.target.head().await?));
}
let outcome = self
.target
.volume
.fs
.apply_workspace_join(crate::facade::WorkspaceJoinRequest {
target: &self.target.volume,
base: self.base,
source: &self.source_head,
expected_target: options.if_target,
expected_head: self.target_authority_head,
history: self.history,
operation_id: options.idempotency_key.operation_id(),
permit,
maximum_generations: self.maximum_generations,
maximum_changes: self.maximum_changes,
maximum_conflicts: self.maximum_conflicts,
resolutions,
})
.await?;
let generation = |id| Generation {
workspace: self.target.clone(),
id,
};
Ok(match outcome {
crate::facade::WorkspaceJoinOutcome::Applied(id) => {
JoinOutcome::Applied(generation(id))
}
crate::facade::WorkspaceJoinOutcome::AlreadyApplied(id) => {
JoinOutcome::AlreadyApplied(generation(id))
}
crate::facade::WorkspaceJoinOutcome::NoChanges(id) => {
JoinOutcome::NoChanges(generation(id))
}
crate::facade::WorkspaceJoinOutcome::Stale(id) => {
JoinOutcome::StaleTarget(generation(id))
}
crate::facade::WorkspaceJoinOutcome::Conflicted(conflicts, truncated) => {
JoinOutcome::Conflicted {
conflicts,
truncated,
}
}
crate::facade::WorkspaceJoinOutcome::Fenced => JoinOutcome::Fenced,
crate::facade::WorkspaceJoinOutcome::IdempotencyConflict => {
JoinOutcome::IdempotencyConflict
}
})
}
}
fn binding_conflict_view<A, O>(
plan: &JoinPlan<A, O>,
conflict: MergeConflict,
) -> crate::ConflictView {
match conflict {
MergeConflict::File(file_id) => crate::ConflictView {
key: ConflictKey::File(file_id),
path: None,
kind: ConflictKind::Record,
base: ConflictValue::RecordReference {
generation: plan.base.id,
file_id,
},
ours: ConflictValue::RecordReference {
generation: plan.target_head.id,
file_id,
},
theirs: ConflictValue::RecordReference {
generation: plan.source_head.id,
file_id,
},
},
MergeConflict::Binding { directory_id, name } => {
let name = canonical_name_key(&name);
let value = |generation| ConflictValue::BindingReference {
generation,
directory_id,
name: name.clone(),
};
crate::ConflictView {
key: ConflictKey::Binding {
directory_id,
name: name.clone(),
},
path: None,
kind: ConflictKind::Binding,
base: value(plan.base.id),
ours: value(plan.target_head.id),
theirs: value(plan.source_head.id),
}
}
}
}
fn record_conflict_view<A, O>(
plan: &JoinPlan<A, O>,
file_id: FileId,
path: Option<String>,
kind: ConflictKind,
) -> crate::ConflictView {
let value = |generation| ConflictValue::Record {
generation,
file_id,
};
crate::ConflictView {
key: ConflictKey::File(file_id),
path,
kind,
base: value(plan.base.id),
ours: value(plan.target_head.id),
theirs: value(plan.source_head.id),
}
}
fn canonical_name_key(name: &LogicalName) -> Vec<u8> {
let mut key = Vec::with_capacity(name.as_bytes().len().saturating_add(1));
key.push(match name.encoding() {
crate::kernel::NameEncoding::Utf8 => 1,
crate::kernel::NameEncoding::PosixBytes => 2,
crate::kernel::NameEncoding::WindowsUtf16Le => 3,
});
key.extend_from_slice(name.as_bytes());
key
}
fn logical_name_text(name: &LogicalName) -> Result<std::borrow::Cow<'_, str>, WorkspaceError> {
name.unicode_text().ok_or_else(|| {
WorkspaceError::path("non-Unicode path cannot be projected as a portable string")
})
}
fn namespace_path_text(path: &NamespacePath) -> Result<String, WorkspaceError> {
if path.is_root() {
return Ok("/".to_owned());
}
let components = path
.components()
.iter()
.map(logical_name_text)
.collect::<Result<Vec<_>, _>>()?;
Ok(format!("/{}", components.join("/")))
}
fn child_namespace_path(
directory: &NamespacePath,
name: LogicalName,
limits: crate::model::VolumeLimits,
) -> Result<NamespacePath, WorkspaceError> {
let mut components = directory.components().to_vec();
components.push(name);
NamespacePath::new(components, limits).map_err(WorkspaceError::path)
}
fn kernel_conflict(
key: &ConflictKey,
maximum_component_bytes: u32,
) -> Result<MergeConflict, DrivenJoinError> {
match key {
ConflictKey::File(file_id)
| ConflictKey::Metadata(file_id)
| ConflictKey::HardLinks(file_id)
| ConflictKey::Directory(file_id)
| ConflictKey::SpecialPayload(file_id)
| ConflictKey::ContentRange { file_id, .. }
| ConflictKey::Rename { file_id, .. } => Ok(MergeConflict::File(*file_id)),
ConflictKey::Binding { directory_id, name } => {
let (&tag, bytes) = name.split_first().ok_or(DrivenJoinError::StaleCandidate)?;
let encoding = match tag {
1 => crate::kernel::NameEncoding::Utf8,
2 => crate::kernel::NameEncoding::PosixBytes,
3 => crate::kernel::NameEncoding::WindowsUtf16Le,
_ => return Err(DrivenJoinError::StaleCandidate),
};
let name = LogicalName::new(encoding, bytes.to_vec(), maximum_component_bytes)
.map_err(|_| DrivenJoinError::StaleCandidate)?;
Ok(MergeConflict::Binding {
directory_id: *directory_id,
name,
})
}
}
}
fn conflict_file_id(key: &ConflictKey) -> Option<FileId> {
match key {
ConflictKey::File(file_id)
| ConflictKey::Metadata(file_id)
| ConflictKey::HardLinks(file_id)
| ConflictKey::Directory(file_id)
| ConflictKey::SpecialPayload(file_id)
| ConflictKey::ContentRange { file_id, .. }
| ConflictKey::Rename { file_id, .. } => Some(*file_id),
ConflictKey::Binding { .. } => None,
}
}
fn resolution_matches_conflict(key: &ConflictKey, resolution: &MergeResolution) -> bool {
match resolution {
MergeResolution::Select(_) | MergeResolution::Unresolved => true,
MergeResolution::Text(_) | MergeResolution::Binary(_) => {
matches!(key, ConflictKey::File(_) | ConflictKey::ContentRange { .. })
}
MergeResolution::Metadata(_) => matches!(key, ConflictKey::Metadata(_)),
MergeResolution::Binding(_) => matches!(key, ConflictKey::Binding { .. }),
}
}
#[cfg(test)]
mod merge_resolution_tests {
use super::*;
#[test]
fn synthesized_resolutions_match_only_their_exact_conflict_shape() {
let file_id = FileId::new();
let file = ConflictKey::File(file_id);
let metadata = ConflictKey::Metadata(file_id);
let binding = ConflictKey::Binding {
directory_id: file_id,
name: vec![1, b'x'],
};
for resolution in [
MergeResolution::Text("merged".into()),
MergeResolution::Binary(vec![1]),
] {
assert!(resolution_matches_conflict(&file, &resolution));
assert!(!resolution_matches_conflict(&metadata, &resolution));
assert!(!resolution_matches_conflict(&binding, &resolution));
}
let metadata_resolution = MergeResolution::Metadata(vec![1]);
assert!(resolution_matches_conflict(&metadata, &metadata_resolution));
assert!(!resolution_matches_conflict(&file, &metadata_resolution));
let binding_resolution = MergeResolution::Binding(Some(file_id));
assert!(resolution_matches_conflict(&binding, &binding_resolution));
assert!(!resolution_matches_conflict(&file, &binding_resolution));
assert!(resolution_matches_conflict(
&file,
&MergeResolution::Select(ConflictSide::Ours)
));
}
}
impl<A, O> GenerationPin<A, O> {
#[must_use]
pub fn identity(&self) -> &WorkspaceName {
&self.identity
}
#[must_use]
pub const fn generation(&self) -> &Generation<A, O> {
&self.generation
}
}
impl<A, O> Checkpoint<A, O> {
#[must_use]
pub fn label(&self) -> &WorkspaceName {
&self.label
}
#[must_use]
pub const fn generation(&self) -> &Generation<A, O> {
&self.generation
}
}
impl<A, O> Clone for Generation<A, O> {
fn clone(&self) -> Self {
Self {
workspace: self.workspace.clone(),
id: self.id,
}
}
}
impl<A, O> Generation<A, O> {
#[must_use]
pub const fn id(&self) -> GenerationId {
self.id
}
#[must_use]
pub const fn workspace_id(&self) -> WorkspaceId {
self.workspace.id
}
#[must_use]
pub fn workspace(&self) -> Workspace<A, O> {
self.workspace.clone()
}
}
impl<A: AsyncAuthorityStore, O: AsyncObjectStore> Generation<A, O> {
pub async fn reader(&self) -> Result<crate::PinnedReader<A, O>, WorkspaceError> {
self.workspace
.engine_checkout(
GenerationSelector::Exact(self.id),
CheckoutMode::read_only_pinned(),
)
.await?
.pinned_reader()
.map_err(WorkspaceError::engine)
}
pub async fn diff_to(
&self,
to: &Generation<A, O>,
maximum_changes: u32,
) -> Result<ChangeSet<A, O>, WorkspaceError> {
self.diff_to_bounded(
to,
maximum_changes,
crate::WorkBudget::UNBOUNDED,
&crate::CancellationToken::new(),
)
.await
}
pub async fn diff_to_bounded(
&self,
to: &Generation<A, O>,
maximum_changes: u32,
budget: crate::WorkBudget,
cancellation: &crate::CancellationToken,
) -> Result<ChangeSet<A, O>, WorkspaceError> {
let receipt = self
.workspace
.volume
.fs
.workspace_join_changes_bounded(self.id, to, maximum_changes, budget, cancellation)
.await?;
Ok(ChangeSet {
from: self.clone(),
to: to.clone(),
changes: receipt.value,
work: receipt.work,
})
}
#[cfg(all(feature = "native-mount", not(target_arch = "wasm32")))]
pub async fn materialize(
&self,
options: &crate::MaterializeOptions,
budget: crate::WorkBudget,
cancellation: &crate::CancellationToken,
) -> Result<crate::OperationReceipt<crate::MaterializationReceipt>, WorkspaceError> {
self.materialize_with_mode(
options,
budget,
cancellation,
crate::native_mount::MaterializeMode::DurableOutput,
)
.await
}
#[cfg(all(feature = "native-mount", not(target_arch = "wasm32")))]
pub(crate) async fn materialize_with_mode(
&self,
options: &crate::MaterializeOptions,
budget: crate::WorkBudget,
cancellation: &crate::CancellationToken,
mode: crate::native_mount::MaterializeMode,
) -> Result<crate::OperationReceipt<crate::MaterializationReceipt>, WorkspaceError> {
let checkout = self
.workspace
.engine_checkout_measured(
GenerationSelector::Exact(self.id),
CheckoutMode::read_only_pinned(),
budget,
cancellation,
)
.await?;
let mut value = checkout.value;
let receipt = crate::native_mount::materialize_checkout_with_mode(
&mut value,
options,
checkout
.work
.remaining(budget)
.map_err(WorkspaceError::from)?,
cancellation,
mode,
)
.await
.map_err(|failure| WorkspaceError::engine(failure.error))?;
merge_workspace_work(checkout.work, receipt, budget)
}
#[cfg(all(feature = "native-mount", not(target_arch = "wasm32")))]
pub async fn materialize_path(
&self,
path: &str,
options: &crate::MaterializeOptions,
budget: crate::WorkBudget,
cancellation: &crate::CancellationToken,
) -> Result<crate::OperationReceipt<crate::MaterializationReceipt>, WorkspaceError> {
self.materialize_path_with_mode(
path,
options,
budget,
cancellation,
crate::native_mount::MaterializeMode::DurableOutput,
)
.await
}
#[cfg(all(feature = "native-mount", not(target_arch = "wasm32")))]
pub(crate) async fn materialize_path_with_mode(
&self,
path: &str,
options: &crate::MaterializeOptions,
budget: crate::WorkBudget,
cancellation: &crate::CancellationToken,
mode: crate::native_mount::MaterializeMode,
) -> Result<crate::OperationReceipt<crate::MaterializationReceipt>, WorkspaceError> {
let checkout = self
.workspace
.engine_checkout_measured(
GenerationSelector::Exact(self.id),
CheckoutMode::read_only_pinned(),
budget,
cancellation,
)
.await?;
let mut value = checkout.value;
let path = customer_path(path, value.volume_config())?;
let receipt = crate::native_mount::materialize_checkout_paths_with_mode(
&mut value,
&[path],
options,
checkout
.work
.remaining(budget)
.map_err(WorkspaceError::from)?,
cancellation,
mode,
)
.await
.map_err(|failure| WorkspaceError::engine(failure.error))?;
merge_workspace_work(checkout.work, receipt, budget)
}
#[cfg(all(feature = "native-mount", not(target_arch = "wasm32")))]
pub async fn materialize_paths(
&self,
paths: &[String],
options: &crate::MaterializeOptions,
budget: crate::WorkBudget,
cancellation: &crate::CancellationToken,
) -> Result<crate::OperationReceipt<crate::MaterializationReceipt>, WorkspaceError> {
let checkout = self
.workspace
.engine_checkout_measured(
GenerationSelector::Exact(self.id),
CheckoutMode::read_only_pinned(),
budget,
cancellation,
)
.await?;
let mut value = checkout.value;
let paths = paths
.iter()
.map(|path| customer_path(path, value.volume_config()))
.collect::<Result<Vec<_>, _>>()?;
let receipt = crate::materialize_checkout_paths(
&mut value,
&paths,
options,
checkout
.work
.remaining(budget)
.map_err(WorkspaceError::from)?,
cancellation,
)
.await
.map_err(|failure| WorkspaceError::engine(failure.error))?;
merge_workspace_work(checkout.work, receipt, budget)
}
#[cfg(all(feature = "native-mount", not(target_arch = "wasm32")))]
pub async fn restore_host_path(
&self,
relative: &std::path::Path,
replacement: crate::HostPathReplacement,
options: &crate::MaterializeOptions,
budget: crate::WorkBudget,
cancellation: &crate::CancellationToken,
) -> Result<crate::OperationReceipt<crate::HostPathRestore>, WorkspaceError> {
let checkout = self
.workspace
.engine_checkout_measured(
GenerationSelector::Exact(self.id),
CheckoutMode::read_only_pinned(),
budget,
cancellation,
)
.await?;
let mut value = checkout.value;
let receipt = crate::restore_checkout_host_path(
&mut value,
relative,
replacement,
options,
checkout
.work
.remaining(budget)
.map_err(WorkspaceError::from)?,
cancellation,
)
.await
.map_err(|failure| WorkspaceError::engine(failure.error))?;
merge_workspace_work(checkout.work, receipt, budget)
}
#[cfg(all(feature = "native-mount", not(target_arch = "wasm32")))]
pub async fn restore_host_paths(
&self,
paths: &[std::path::PathBuf],
replacement: crate::HostPathReplacement,
options: &crate::MaterializeOptions,
cancellation: &crate::CancellationToken,
) -> Result<Vec<crate::HostPathRestore>, WorkspaceError> {
let mut checkout = self
.workspace
.engine_checkout_measured(
GenerationSelector::Exact(self.id),
CheckoutMode::read_only_pinned(),
crate::WorkBudget::UNBOUNDED,
cancellation,
)
.await?
.value;
let mut restored = Vec::new();
restored
.try_reserve_exact(paths.len())
.map_err(|_| WorkspaceError::engine("restore path result allocation failed"))?;
for path in paths {
restored.push(
crate::restore_checkout_host_path(
&mut checkout,
path,
replacement,
options,
crate::WorkBudget::UNBOUNDED,
cancellation,
)
.await
.map_err(|failure| WorkspaceError::engine(failure.error))?
.value,
);
}
Ok(restored)
}
pub async fn parents(&self) -> Result<Vec<GenerationId>, WorkspaceError> {
self.workspace
.volume
.fs
.generation_parents(&self.workspace.volume, self.id)
.await
}
pub async fn read(&self, path: &str, maximum_bytes: u64) -> Result<Bytes, WorkspaceError> {
read_generation(
&self.workspace,
GenerationSelector::Exact(self.id),
path,
maximum_bytes,
)
.await
}
pub async fn read_range(
&self,
path: &str,
offset: u64,
length: u64,
) -> Result<Bytes, WorkspaceError> {
read_generation_range(
&self.workspace,
GenerationSelector::Exact(self.id),
path,
offset,
length,
)
.await
}
pub async fn stat(&self, path: &str) -> Result<WorkspaceStat, WorkspaceError> {
stat_generation(&self.workspace, GenerationSelector::Exact(self.id), path).await
}
pub async fn lookup_paths(
&self,
paths: &[NamespacePath],
budget: crate::WorkBudget,
cancellation: &crate::CancellationToken,
) -> crate::FsResult<Vec<Option<crate::kernel::FileRecord>>> {
let checkout = self
.workspace
.volume
.checkout(
GenerationSelector::Exact(self.id),
CheckoutMode::read_only_pinned(),
budget,
cancellation,
)
.await?;
let checkout_work = checkout.work;
let remaining = checkout_work
.remaining(budget)
.map_err(|error| crate::OperationFailure::new(error.into(), checkout_work))?;
let mut checkout = checkout.value;
let lookup = checkout
.lookup_batch_no_follow(paths, remaining, cancellation)
.await
.map_err(|failure| failure.map_with_prior_work(checkout_work, |error| error))?;
let work = checkout_work
.checked_add(lookup.work)
.map_err(|error| crate::OperationFailure::new(error.into(), checkout_work))?;
Ok(crate::FsReceipt {
value: lookup
.value
.entries
.into_iter()
.map(|entry| entry.record)
.collect(),
work,
})
}
pub async fn list_directory(
&self,
path: &str,
after: Option<&LogicalName>,
maximum_entries: u32,
) -> Result<WorkspaceDirectoryPage, WorkspaceError> {
list_generation_directory(
&self.workspace,
GenerationSelector::Exact(self.id),
path,
after,
maximum_entries,
)
.await
}
pub async fn paths_for_file_id(
&self,
file_id: FileId,
maximum_entries: u32,
) -> Result<Vec<String>, WorkspaceError> {
Ok(self
.paths_for_file_ids([file_id], maximum_entries)
.await?
.remove(&file_id)
.unwrap_or_default())
}
pub(crate) async fn paths_for_file_id_measured(
&self,
file_id: FileId,
maximum_entries: u32,
budget: crate::WorkBudget,
cancellation: &crate::CancellationToken,
) -> Result<crate::OperationReceipt<Vec<String>>, WorkspaceError> {
let receipt = self
.namespace_records_for_file_ids_bounded(
[file_id],
maximum_entries,
budget,
cancellation,
)
.await?;
let paths = receipt
.value
.records
.get(&file_id)
.into_iter()
.flatten()
.map(|(path, _)| namespace_path_text(path))
.collect::<Result<Vec<_>, _>>()?;
Ok(crate::OperationReceipt {
value: paths,
work: receipt.work,
})
}
pub async fn paths_for_file_ids(
&self,
file_ids: impl IntoIterator<Item = FileId>,
maximum_entries: u32,
) -> Result<BTreeMap<FileId, Vec<String>>, WorkspaceError> {
let paths = self
.namespace_paths_for_file_ids(file_ids, maximum_entries)
.await?;
paths
.into_iter()
.map(|(file_id, paths)| {
paths
.into_iter()
.map(|path| namespace_path_text(&path))
.collect::<Result<Vec<_>, _>>()
.map(|paths| (file_id, paths))
})
.collect()
}
async fn namespace_paths_for_file_ids(
&self,
file_ids: impl IntoIterator<Item = FileId>,
maximum_entries: u32,
) -> Result<BTreeMap<FileId, Vec<NamespacePath>>, WorkspaceError> {
Ok(self
.namespace_records_for_file_ids(file_ids, maximum_entries)
.await?
.records
.into_iter()
.map(|(file_id, records)| {
(file_id, records.into_iter().map(|(path, _)| path).collect())
})
.collect())
}
async fn namespace_records_for_file_ids(
&self,
file_ids: impl IntoIterator<Item = FileId>,
maximum_entries: u32,
) -> Result<NamespaceRecordSearch, WorkspaceError> {
self.namespace_records_for_file_ids_bounded(
file_ids,
maximum_entries,
crate::WorkBudget::UNBOUNDED,
&crate::CancellationToken::new(),
)
.await
.map(|receipt| receipt.value)
}
#[allow(clippy::too_many_lines)]
async fn namespace_records_for_file_ids_bounded(
&self,
file_ids: impl IntoIterator<Item = FileId>,
maximum_entries: u32,
budget: crate::WorkBudget,
cancellation: &crate::CancellationToken,
) -> Result<crate::OperationReceipt<NamespaceRecordSearch>, WorkspaceError> {
let file_ids: BTreeSet<_> = file_ids.into_iter().collect();
if file_ids.is_empty() {
return Ok(crate::OperationReceipt {
value: NamespaceRecordSearch {
records: BTreeMap::new(),
complete: true,
},
work: crate::WorkCounters::default(),
});
}
if maximum_entries == 0 {
return Ok(crate::OperationReceipt {
value: NamespaceRecordSearch {
records: BTreeMap::new(),
complete: false,
},
work: crate::WorkCounters::default(),
});
}
let maximum = usize::try_from(maximum_entries).unwrap_or(usize::MAX);
let limits = self.workspace.volume.config.limits;
let cache_key = crate::path_index::cache_key(self.id, file_ids.iter().copied());
if let Some(bytes) = self.workspace.volume.fs.path_index().load(cache_key)
&& let Some(cached) = crate::path_index::GenerationPathQuery::decode(
&bytes,
self.id,
file_ids.iter().copied(),
)
{
let cached_paths = cached
.into_iter()
.flat_map(|(file_id, paths)| paths.into_iter().map(move |path| (file_id, path)))
.map(|(file_id, path)| path.into_namespace(limits).map(|path| (file_id, path)))
.collect::<Option<Vec<_>>>();
if let Some(cached_paths) = cached_paths {
if cached_paths.is_empty() {
return Ok(crate::OperationReceipt {
value: NamespaceRecordSearch {
records: BTreeMap::new(),
complete: true,
},
work: crate::WorkCounters::default(),
});
}
if cached_paths.len() <= maximum {
let paths = cached_paths
.iter()
.map(|(_, path)| path.clone())
.collect::<Vec<_>>();
let lookup = self
.lookup_paths(&paths, budget, cancellation)
.await
.map_err(WorkspaceError::engine)?;
let mut records = BTreeMap::<_, Vec<_>>::new();
let mut valid = true;
for ((expected, path), record) in cached_paths.into_iter().zip(lookup.value) {
let Some(record) = record else {
valid = false;
break;
};
if record.file_id != expected {
valid = false;
break;
}
records.entry(expected).or_default().push((path, record));
}
if valid {
return Ok(crate::OperationReceipt {
value: NamespaceRecordSearch {
records,
complete: true,
},
work: lookup.work,
});
}
}
}
}
let mut matches: BTreeMap<_, Vec<(NamespacePath, crate::kernel::FileRecord)>> =
BTreeMap::new();
let checkout = self
.workspace
.engine_checkout_measured(
GenerationSelector::Exact(self.id),
CheckoutMode::read_only_pinned(),
budget,
cancellation,
)
.await?;
let mut work = checkout.work;
let mut checkout = checkout.value;
let root = NamespacePath::new(Vec::new(), limits).map_err(WorkspaceError::path)?;
let root_lookup = checkout
.lookup_no_follow(
&root,
work.remaining(budget).map_err(WorkspaceError::engine)?,
cancellation,
)
.await
.map_err(WorkspaceError::engine)?;
work = work
.checked_add(root_lookup.work)
.map_err(WorkspaceError::engine)?;
if let Some(record) = root_lookup.value.record
&& file_ids.contains(&record.file_id)
{
matches
.entry(record.file_id)
.or_default()
.push((root.clone(), record));
}
let mut pending = vec![root];
let mut examined = 0_usize;
while let Some(directory) = pending.pop() {
let mut after = None;
loop {
let page_receipt = checkout
.list_directory_records(
&directory,
after.as_ref(),
1_024,
work.remaining(budget).map_err(WorkspaceError::engine)?,
cancellation,
)
.await
.map_err(WorkspaceError::engine)?;
work = work
.checked_add(page_receipt.work)
.map_err(WorkspaceError::engine)?;
let page = page_receipt.value;
for entry in &page.entries {
examined = examined.saturating_add(1);
if examined > maximum {
for records in matches.values_mut() {
records.sort_by(|left, right| left.0.cmp(&right.0));
}
return Ok(crate::OperationReceipt {
value: NamespaceRecordSearch {
records: matches,
complete: false,
},
work,
});
}
let mut components = directory.components().to_vec();
components.push(entry.name.clone());
let path =
NamespacePath::new(components, limits).map_err(WorkspaceError::path)?;
if file_ids.contains(&entry.record.file_id) {
matches
.entry(entry.record.file_id)
.or_default()
.push((path.clone(), entry.record));
}
if entry.record.kind == FileKind::Directory {
pending.push(path);
}
}
if !page.has_more {
break;
}
after = page.entries.last().map(|entry| entry.name.clone());
}
}
for records in matches.values_mut() {
records.sort_by(|left, right| left.0.cmp(&right.0));
}
let cached_paths = matches
.iter()
.map(|(&file_id, records)| {
(
file_id,
records
.iter()
.map(|(path, _)| crate::path_index::CachedPath::from_namespace(path))
.collect(),
)
})
.collect();
let cached = crate::path_index::GenerationPathQuery::new(
self.id,
file_ids.iter().copied(),
&cached_paths,
);
if let Some(bytes) = cached.encode() {
self.workspace
.volume
.fs
.path_index()
.store(cache_key, &bytes);
}
Ok(crate::OperationReceipt {
value: NamespaceRecordSearch {
records: matches,
complete: true,
},
work,
})
}
pub async fn read_symbolic_link(&self, path: &str) -> Result<Bytes, WorkspaceError> {
read_generation_symbolic_link(&self.workspace, GenerationSelector::Exact(self.id), path)
.await
}
pub async fn plan_extents(
&self,
path: &str,
offset: u64,
length: u64,
maximum_spans: u32,
) -> Result<WorkspaceExtentPlan, WorkspaceError> {
plan_generation_extents(
&self.workspace,
GenerationSelector::Exact(self.id),
path,
offset,
length,
maximum_spans,
)
.await
}
pub async fn pin(
&self,
identity: impl AsRef<str>,
) -> Result<GenerationPin<A, O>, WorkspaceError> {
let identity = WorkspaceName::new(identity)?;
self.workspace
.volume
.fs
.retain_workspace_generation(
&self.workspace.volume,
self.id,
crate::kernel::RetentionKind::Pin,
identity.as_str().to_owned(),
)
.await?;
Ok(GenerationPin {
identity,
generation: self.clone(),
})
}
}
#[derive(Clone)]
pub struct ForkOptions<A, O> {
pub(crate) generation: Generation<A, O>,
pub(crate) idempotency_key: IdempotencyKey,
}
impl<A, O> ForkOptions<A, O> {
#[must_use]
pub fn from_generation(generation: Generation<A, O>, idempotency_key: IdempotencyKey) -> Self {
Self {
generation,
idempotency_key,
}
}
}
#[derive(Clone, Copy, Debug, Eq, Error, PartialEq)]
pub enum WorkspaceNameError {
#[error("workspace name is empty")]
Empty,
#[error("workspace name is reserved")]
Reserved,
#[error("workspace name exceeds 255 UTF-8 bytes")]
TooLong,
#[error("workspace name contains a path separator or control character")]
InvalidCharacter,
}
#[derive(Debug, Error)]
pub enum WorkspaceError {
#[error(transparent)]
Name(#[from] WorkspaceNameError),
#[error("workspace engine failure: {0}")]
Engine(String),
#[error(transparent)]
Cancelled(#[from] crate::CancellationError),
#[error(transparent)]
Work(#[from] crate::WorkError),
#[error("fork generation belongs to another filesystem deployment")]
ForeignGeneration,
#[error("workspace generation is stale")]
StaleGeneration,
#[error("workspace file identity is stale")]
StaleIdentity,
#[error("invalid workspace path: {0}")]
Path(String),
#[error("workspace path does not exist")]
NotFound,
#[error("workspace path is not a regular file")]
NotRegularFile,
#[error("workspace path is not a directory")]
NotDirectory,
#[error("workspace file exceeds the caller's read bound")]
ReadLimitExceeded,
#[error("retention label is already bound to another generation")]
RetentionConflict,
#[error("workspaces are incompatible")]
IncompatibleWorkspace,
#[error("workspaces have no common ancestor")]
NoCommonAncestor,
#[error("workspace lineage exceeds the configured bound")]
LineageLimit,
#[error("workspace join exceeds its configured bound")]
JoinLimit,
#[error("merge resolution is incompatible with its conflict")]
InvalidMergeResolution,
#[error("change sets are not contiguous")]
ChangeSetContinuity,
#[error("changed paths exceed the caller's traversal bound")]
ChangedPathLimit,
#[error("staged content set is empty")]
EmptyContentSet,
#[error("workspace is not a fork")]
NotFork,
#[error("staged content length overflow")]
ContentLengthOverflow,
}
impl WorkspaceError {
pub(crate) fn engine(error: impl fmt::Display) -> Self {
Self::Engine(error.to_string())
}
pub(crate) fn path(error: impl fmt::Display) -> Self {
Self::Path(error.to_string())
}
}
impl From<FsError> for WorkspaceError {
fn from(error: FsError) -> Self {
match error {
FsError::Cancelled(error) => Self::Cancelled(error),
FsError::Work(error) => Self::Work(error),
FsError::NotFound => Self::NotFound,
FsError::NotDirectory => Self::NotDirectory,
FsError::FileRead(crate::kernel::FileRangeReadError::NotRegular) => {
Self::NotRegularFile
}
FsError::Mutation(crate::kernel::GenerationMutationError::FileIdentityConflict) => {
Self::StaleIdentity
}
error => Self::Engine(error.to_string()),
}
}
}
pub(crate) fn customer_path(
path: &str,
config: crate::model::VolumeConfig,
) -> Result<NamespacePath, WorkspaceError> {
let portable = PortablePath::parse(path, config.limits).map_err(WorkspaceError::path)?;
NamespacePath::from_portable_in_profile(&portable, config.profile, config.limits)
.map_err(WorkspaceError::path)
}
#[cfg(all(feature = "native-mount", not(target_arch = "wasm32")))]
fn merge_workspace_work<T>(
prior: crate::WorkCounters,
receipt: crate::OperationReceipt<T>,
budget: crate::WorkBudget,
) -> Result<crate::OperationReceipt<T>, WorkspaceError> {
let work = prior
.checked_add(receipt.work)
.map_err(WorkspaceError::from)?;
work.verify(budget).map_err(WorkspaceError::from)?;
Ok(crate::OperationReceipt {
value: receipt.value,
work,
})
}
fn path_conflict_kind(
base: Option<crate::kernel::FileRecord>,
current: Option<crate::kernel::FileRecord>,
source: Option<crate::kernel::FileRecord>,
) -> ConflictKind {
let (Some(base), Some(current), Some(source)) = (base, current, source) else {
return ConflictKind::Binding;
};
if base.kind != current.kind
|| base.kind != source.kind
|| base.file_id != current.file_id
|| base.file_id != source.file_id
{
return ConflictKind::Binding;
}
if base.link_count != current.link_count || base.link_count != source.link_count {
return ConflictKind::HardLink;
}
if base.metadata != current.metadata || base.metadata != source.metadata {
return ConflictKind::Metadata;
}
match base.kind {
FileKind::Regular => ConflictKind::Binary,
FileKind::Directory => ConflictKind::Directory,
FileKind::SymbolicLink => ConflictKind::SymbolicLink,
FileKind::Fifo
| FileKind::Socket
| FileKind::CharacterDevice
| FileKind::BlockDevice
| FileKind::ReparsePoint
| FileKind::MountBoundary => ConflictKind::Special,
}
}
fn regular_file_bytes(record: crate::kernel::FileRecord) -> Result<u64, WorkspaceError> {
match record.payload {
FilePayload::InlineRegular(bytes) => {
Ok(u64::try_from(bytes.as_bytes().len()).unwrap_or(u64::MAX))
}
FilePayload::Regular { logical_bytes, .. } => Ok(logical_bytes),
_ => Err(WorkspaceError::NotRegularFile),
}
}
#[cfg(test)]
#[path = "tests/workspace.rs"]
mod tests;