use std::sync::{Mutex, MutexGuard};
use crate::store::NodeStore;
use crate::tree::Hash;
use crate::tree::mutate::batch_mutate_owned_with_source;
use crate::wal::{Mutation, WalBuffer};
pub use super::commit_error::BranchCommitError;
use super::conflict::ConflictPolicy;
use super::handle::{BranchHandle, RefBinding, ShardId, ShardState};
use super::merge::{MergeReport, merge_with_report};
use super::operation_error::{CommitOperationError, preserve_node_publication_fence};
use super::policy::check_merge;
use super::refstore::{BranchRefError, BranchRefStore};
use super::registry::{BranchRegistry, advance_in};
use super::snapshot::Timestamp;
#[derive(Debug)]
pub enum CommitDurability<'a> {
Volatile,
Durable {
refs: &'a mut BranchRefStore,
},
}
#[derive(Debug)]
pub struct CommitRequest<'a> {
pub durability: CommitDurability<'a>,
pub extra_parents: &'a [Hash],
pub timestamp: Timestamp,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CommitOutcome {
pub heads: Vec<(ShardId, Hash)>,
pub seq: Option<u64>,
pub advanced: bool,
}
pub fn commit_branch<S: NodeStore + ?Sized>(
branch: &BranchHandle,
store: &mut S,
registry: &BranchRegistry,
request: CommitRequest<'_>,
) -> Result<CommitOutcome, BranchCommitError> {
commit_branch_with_source(branch, store, registry, request)
.map_err(CommitOperationError::into_public)
}
pub(crate) fn commit_branch_with_source<S: NodeStore + ?Sized>(
branch: &BranchHandle,
store: &mut S,
registry: &BranchRegistry,
request: CommitRequest<'_>,
) -> Result<CommitOutcome, CommitOperationError<S::Error>> {
let CommitRequest {
durability,
extra_parents,
timestamp,
} = request;
let Some(registry_guard) = branch.registry_guard() else {
return Err(BranchCommitError::Unregistered.into());
};
if matches!(durability, CommitDurability::Volatile) && !extra_parents.is_empty() {
return Err(BranchCommitError::VolatileExtraParents.into());
}
let mut states = branch.lock_states_ascending()?;
let mut binding = branch.binding.as_deref().map(lock_binding);
let durable = match durability {
CommitDurability::Volatile => None,
CommitDurability::Durable { refs } => match binding.as_deref() {
Some(bound) => Some((refs, bound.name.clone(), bound.created, bound.seq)),
None => return Err(BranchCommitError::UnnamedBranch.into()),
},
};
let mut counts = registry.lock_counts();
if extra_parents.is_empty() && states.iter().all(|(_, state)| state.buffer.is_empty()) {
if let Some((refs, name, created, seq)) = &durable {
refs.cas_check(name, *created, *seq)?;
}
return Ok(CommitOutcome {
heads: current_heads(&states),
seq: durable.map(|(_, _, _, seq)| seq),
advanced: false,
});
}
let mut materialised: Vec<Option<Hash>> = Vec::with_capacity(states.len());
for (_, state) in &states {
if state.buffer.is_empty() {
materialised.push(None);
continue;
}
let new_root = batch_mutate_owned_with_source(
store,
state.current_root,
buffered_batch(&state.buffer),
)
.map_err(CommitOperationError::Tree)?;
materialised.push(Some(new_root));
}
let advanced: Vec<(ShardId, Hash, Hash)> = states
.iter()
.zip(&materialised)
.filter_map(|((shard_id, state), new_root)| {
new_root.and_then(|new_root| {
(new_root != state.current_root).then_some((
*shard_id,
state.current_root,
new_root,
))
})
})
.collect();
let new_seq = if let Some((refs, name, created, expected_seq)) = durable {
preserve_node_publication_fence(store.sync_dirty_dirs())
.map_err(CommitOperationError::Fence)?;
let Some(record) = refs.get(&name) else {
return Err(BranchCommitError::Ref(BranchRefError::BranchRemoved(name)).into());
};
let parents: Vec<Hash> = record
.shards
.iter()
.map(|shard| shard.head)
.chain(extra_parents.iter().copied())
.collect();
let new_heads: Vec<(ShardId, Hash)> = advanced
.iter()
.map(|&(shard_id, _, new_root)| (shard_id, new_root))
.collect();
Some(refs.advance(&name, created, expected_seq, &new_heads, parents, timestamp)?)
} else {
None
};
for &(_, old_root, new_root) in &advanced {
advance_in(&mut counts, old_root, new_root);
registry_guard.replace(old_root, new_root);
}
for ((_, state), new_root) in states.iter_mut().zip(&materialised) {
if let Some(new_root) = new_root {
state.current_root = *new_root;
state.buffer = WalBuffer::new();
}
}
if let (Some(bound), Some(seq)) = (binding.as_deref_mut(), new_seq) {
bound.seq = seq;
}
let heads = current_heads(&states);
drop(counts);
drop(binding);
drop(states);
Ok(CommitOutcome {
heads,
seq: new_seq,
advanced: true,
})
}
pub fn merge<S: NodeStore + ?Sized>(
store: &mut S,
source: &BranchHandle,
destination: &BranchHandle,
refs: &BranchRefStore,
policy: &ConflictPolicy,
) -> Result<MergeReport, BranchCommitError> {
let source_binding = source
.binding_snapshot()
.ok_or(BranchCommitError::UnnamedBranch)?;
let destination_binding = destination
.binding_snapshot()
.ok_or(BranchCommitError::UnnamedBranch)?;
let source_record = refs.cas_check(
&source_binding.name,
source_binding.created,
source_binding.seq,
)?;
let source_name = source_record.name.clone();
let source_kind = source_record.kind;
let source_lineage = source_record
.resolved_namespace_lineage()
.map(str::to_owned);
let destination_record = refs.cas_check(
&destination_binding.name,
destination_binding.created,
destination_binding.seq,
)?;
let destination_name = destination_record.name.clone();
let destination_kind = destination_record.kind;
let destination_lineage = destination_record
.resolved_namespace_lineage()
.map(str::to_owned);
check_merge(
&source_name,
source_kind,
source_lineage.as_deref(),
&destination_name,
destination_kind,
destination_lineage.as_deref(),
)?;
if source.shard_count() != 1 || destination.shard_count() != 1 {
return Err(super::merge::MergeError::Unimplemented {
feature: "multi-shard branch merge",
}
.into());
}
merge_with_report(
store,
destination.current_root(),
source.current_root(),
source.fork_point(),
policy,
)
.map_err(BranchCommitError::from)
}
fn lock_binding(binding: &Mutex<RefBinding>) -> MutexGuard<'_, RefBinding> {
match binding.lock() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
}
}
fn buffered_batch(buffer: &WalBuffer) -> Vec<(Vec<u8>, Option<Vec<u8>>)> {
buffer
.iter()
.map(|mutation| match mutation {
Mutation::Put { key, value } => (key.clone(), Some(value.clone())),
Mutation::Delete { key } => (key.clone(), None),
})
.collect()
}
fn current_heads(states: &[(ShardId, MutexGuard<'_, ShardState>)]) -> Vec<(ShardId, Hash)> {
states
.iter()
.map(|(shard_id, state)| (*shard_id, state.current_root))
.collect()
}
#[cfg(test)]
#[path = "commit_tests.rs"]
mod tests;
#[cfg(test)]
#[path = "commit_error_tests.rs"]
mod error_tests;