use std::collections::{BTreeMap, BTreeSet};
use std::fs::OpenOptions;
use std::io::{self, Write};
use std::sync::atomic::AtomicBool;
use super::lock_wait::check_cancelled;
use super::store::{
Lock, check_expected, conflicts, io_error, read_optional, remove_loose, validate_target,
};
use super::{
Expected, FilesTransactionOptions, LockWait, RefName, ReferenceError, References, Reflog,
ReflogEntry, Target, packed, reflog,
};
use crate::ObjectId;
#[derive(Clone, Copy, Debug, Eq, PartialEq, Hash)]
#[non_exhaustive]
pub enum Dereference {
Stored,
Follow,
}
#[derive(Clone, Debug)]
#[non_exhaustive]
pub struct RefEdit {
pub name: RefName,
pub dereference: Dereference,
pub target: Option<Target>,
pub expected: Expected,
pub reflog: Reflog,
}
impl RefEdit {
pub fn new(name: RefName, target: Option<Target>, expected: Expected, reflog: Reflog) -> Self {
Self {
name,
target,
expected,
reflog,
dereference: crate::refs::Dereference::Stored,
}
}
pub fn create(name: RefName, id: ObjectId, reflog: Reflog) -> Self {
Self::new(name, Some(Target::Direct(id)), Expected::Absent, reflog)
}
pub fn update(name: RefName, old: ObjectId, new: ObjectId, reflog: Reflog) -> Self {
Self::new(
name,
Some(Target::Direct(new)),
Expected::Value(Target::Direct(old)),
reflog,
)
}
pub fn delete(name: RefName, old: ObjectId, reflog: Reflog) -> Self {
Self::new(name, None, Expected::Value(Target::Direct(old)), reflog)
}
#[must_use]
pub fn with_name(mut self, name: RefName) -> Self {
self.name = name;
self
}
#[must_use]
pub fn with_dereference(mut self, dereference: Dereference) -> Self {
self.dereference = dereference;
self
}
#[must_use]
pub fn with_target(mut self, target: Option<Target>) -> Self {
self.target = target;
self
}
#[must_use]
pub fn with_expected(mut self, expected: Expected) -> Self {
self.expected = expected;
self
}
#[must_use]
pub fn with_reflog(mut self, reflog: Reflog) -> Self {
self.reflog = reflog;
self
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[non_exhaustive]
pub enum RefOutcome {
Unchanged,
PackedRemoved,
Published,
}
#[derive(Clone, Debug, Eq, PartialEq)]
#[non_exhaustive]
pub enum LogOutcome {
NotAttempted,
Appended,
Deleted,
Failed {
bytes_written: usize,
},
}
#[derive(Clone, Debug, Eq, PartialEq)]
#[non_exhaustive]
pub struct RefEditOutcome {
pub name: RefName,
pub reference: RefOutcome,
pub logs: Vec<(RefName, LogOutcome)>,
}
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum TransactionError {
#[error("reference transaction preparation failed")]
Prepare {
operation: Option<usize>,
#[source]
source: ReferenceError,
},
#[error("reference transaction publication failed")]
Publish {
outcomes: Vec<RefEditOutcome>,
#[source]
source: ReferenceError,
},
}
impl References<'_> {
pub fn transaction(&self, edits: &[RefEdit]) -> Result<Vec<RefEditOutcome>, TransactionError> {
#[cfg(feature = "tracing")]
let span = tracing::debug_span!(
target: "girt",
"refs.transaction",
outcome = "incomplete",
failure_class = tracing::field::Empty,
effects = tracing::field::Empty,
edits = edits.len(),
);
let operation = || {
if edits.is_empty() {
return Ok(Vec::new());
}
for (index, edit) in edits.iter().enumerate() {
validate_edit(self.object_format, edit).map_err(|source| {
TransactionError::Prepare {
operation: Some(index),
source,
}
})?;
}
self.prepare_transaction(edits)?.publish()
};
#[cfg(feature = "tracing")]
let result = span.in_scope(operation);
#[cfg(not(feature = "tracing"))]
let result = { operation }();
#[cfg(feature = "tracing")]
crate::trace::finish(&span, &result, |error| {
crate::trace::transaction(error, &span)
});
result
}
pub fn transaction_files_with_options(
&self,
edits: &[RefEdit],
options: FilesTransactionOptions,
cancel: &AtomicBool,
) -> Result<Vec<RefEditOutcome>, TransactionError> {
self.transaction_files_with_log_identity(edits, options, cancel, LogIdentity::Resolved)
}
pub fn transaction_files_with_stored_log_ids(
&self,
edits: &[RefEdit],
options: FilesTransactionOptions,
cancel: &AtomicBool,
) -> Result<Vec<RefEditOutcome>, TransactionError> {
self.transaction_files_with_log_identity(edits, options, cancel, LogIdentity::Stored)
}
fn transaction_files_with_log_identity(
&self,
edits: &[RefEdit],
options: FilesTransactionOptions,
cancel: &AtomicBool,
identity: LogIdentity,
) -> Result<Vec<RefEditOutcome>, TransactionError> {
#[cfg(feature = "tracing")]
let span = tracing::debug_span!(
target: "girt", "refs.transaction", outcome = "incomplete",
failure_class = tracing::field::Empty, effects = tracing::field::Empty,
edits = edits.len(),
);
let operation = || {
if self.reference_backend != super::Backend::Files {
return Err(TransactionError::Prepare {
operation: None,
source: ReferenceError::Unsupported(
"files transaction options require files backend",
),
});
}
check_cancelled(cancel).map_err(|source| TransactionError::Prepare {
operation: None,
source,
})?;
if edits.is_empty() {
return Ok(Vec::new());
}
for (index, edit) in edits.iter().enumerate() {
if identity == LogIdentity::Stored
&& (edit.dereference == crate::refs::Dereference::Follow
|| matches!(edit.target, Some(Target::Symbolic(_))))
{
return Err(TransactionError::Prepare {
operation: Some(index),
source: ReferenceError::Unsupported(
"stored log identities require non-dereferencing direct edits or deletions",
),
});
}
validate_edit(self.object_format, edit).map_err(|source| {
TransactionError::Prepare {
operation: Some(index),
source,
}
})?;
}
let prepared =
self.prepare_files_transaction_with_options(edits, options, cancel, identity)?;
check_cancelled(cancel).map_err(|source| TransactionError::Prepare {
operation: None,
source,
})?;
prepared.publish()
};
#[cfg(feature = "tracing")]
let result = span.in_scope(operation);
#[cfg(not(feature = "tracing"))]
let result = operation();
#[cfg(feature = "tracing")]
crate::trace::finish(&span, &result, |error| {
crate::trace::transaction(error, &span)
});
result
}
pub(crate) fn prepare_transaction(
&self,
edits: &[RefEdit],
) -> Result<PreparedBackend, TransactionError> {
if self.reference_backend == super::Backend::Reftable {
super::reftable::backend::prepare(self, edits).map(PreparedBackend::Reftable)
} else {
self.prepare_files_transaction(edits)
.map(Box::new)
.map(PreparedBackend::Files)
}
}
pub(super) fn prepare_files_transaction(
&self,
edits: &[RefEdit],
) -> Result<Prepared, TransactionError> {
self.prepare_files_transaction_with_options(
edits,
FilesTransactionOptions::GIT_DEFAULT,
&AtomicBool::new(false),
LogIdentity::Resolved,
)
}
fn prepare_files_transaction_with_options(
&self,
edits: &[RefEdit],
options: FilesTransactionOptions,
cancel: &AtomicBool,
identity: LogIdentity,
) -> Result<Prepared, TransactionError> {
#[cfg(feature = "tracing")]
let span = tracing::debug_span!(
target: "girt",
"refs.prepare_transaction",
outcome = "incomplete",
failure_class = tracing::field::Empty,
effects = tracing::field::Empty,
);
let operation = || {
let batch_error = |source| TransactionError::Prepare {
operation: None,
source,
};
let packed_lock = Lock::acquire_wait(
self.common_dir.join("packed-refs"),
options.packed_refs_lock_wait,
cancel,
)
.map_err(batch_error)?;
let bytes = read_optional(&packed_lock.destination)
.map_err(batch_error)?
.unwrap_or_default();
let packed = packed::parse(self.object_format, &bytes, &packed_lock.destination)
.map_err(batch_error)?;
let mut plans = Vec::new();
let mut names = BTreeSet::new();
for (index, edit) in edits.iter().enumerate() {
let error = |source| TransactionError::Prepare {
operation: Some(index),
source,
};
let chain = self
.discover_chain(edit, &packed, identity)
.map_err(error)?;
let new_chain = if let Some(Target::Symbolic(target)) = &edit.target
&& edit.reflog.append_fields().is_some()
{
let dependency = RefEdit {
name: target.clone(),
dereference: crate::refs::Dereference::Follow,
target: None,
expected: Expected::Any,
reflog: Reflog::Preserve,
};
let new_chain = self
.discover_chain(&dependency, &packed, LogIdentity::Resolved)
.map_err(error)?;
if new_chain.iter().any(|(name, _)| name == &edit.name) {
return Err(error(ReferenceError::Cycle(edit.name.clone())));
}
new_chain
} else {
Vec::new()
};
let dependencies: BTreeSet<_> = chain
.iter()
.chain(&new_chain)
.map(|(name, _)| name)
.collect();
for name in dependencies {
if !names.insert(name.clone())
|| names
.iter()
.any(|other: &RefName| conflicts(name.as_bytes(), other.as_bytes()))
{
return Err(error(ReferenceError::Conflict(
self.path(name).map_err(error)?,
)));
}
}
plans.push((chain, new_chain));
}
let mut locks = BTreeMap::new();
for name in names {
self.check_packed_namespace(&name, &packed)
.map_err(batch_error)?;
locks.insert(
name.clone(),
Lock::acquire_wait(
self.path(&name).map_err(batch_error)?,
options.reference_lock_wait,
cancel,
)
.map_err(batch_error)?,
);
}
let mut operations = Vec::new();
let mut log_names = BTreeSet::new();
for (index, (edit, (chain, new_chain))) in edits.iter().zip(plans).enumerate() {
let error = |source| TransactionError::Prepare {
operation: Some(index),
source,
};
for (name, observed) in chain.iter().chain(&new_chain) {
check_expected(
self.read_locked(name, &packed).map_err(error)?,
match observed {
Some(target) => Expected::Value(target.clone()),
None => Expected::Absent,
},
)
.map_err(error)?;
}
let (name, actual) = if edit.dereference == crate::refs::Dereference::Follow {
chain.last().unwrap()
} else {
chain.first().unwrap()
};
check_expected(actual.clone(), edit.expected.clone()).map_err(error)?;
let mut logs = Vec::new();
let logged_chain = if edit.dereference == crate::refs::Dereference::Follow {
&chain[..]
} else {
&chain[..1]
};
if edit.reflog.append_fields().is_some() {
log_names.extend(logged_chain.iter().map(|(name, _)| name.clone()));
}
let skip_log = matches!(
edit.reflog,
Reflog::AppendIfChanged { .. } | Reflog::AppendExistingIfChanged { .. }
) && actual == &edit.target;
if let Some((committer, message)) = edit.reflog.append_fields()
&& !skip_log
{
let old = log_id(
self.object_format,
chain.last().unwrap().1.as_ref(),
identity,
)
.map_err(error)?;
let new_target = new_chain
.last()
.map_or(edit.target.as_ref(), |(_, value)| value.as_ref());
let new = log_id(self.object_format, new_target, identity).map_err(error)?;
let record = ReflogEntry {
old,
new,
committer: committer.clone(),
message: message.to_vec(),
};
let record = record.encode().map_err(error)?;
for (log_name, _) in logged_chain {
logs.push((log_name.clone(), record.clone()));
}
}
if edit.reflog == Reflog::Delete {
log_names.insert(name.clone());
logs.push((name.clone(), Vec::new()));
}
operations.push(Operation {
name: name.clone(),
target: edit.target.clone(),
delete_log: edit.reflog == Reflog::Delete,
packed_removed: edit.target.is_none() && packed.contains_key(name),
logs,
});
}
let mut log_locks = BTreeMap::new();
for name in log_names {
let path = self.reflog_path(&name).map_err(batch_error)?;
let lock =
Lock::acquire_wait(path, LockWait::Immediate, cancel).map_err(batch_error)?;
log_locks.insert(name, lock);
}
for (operation, edit) in operations.iter_mut().zip(edits) {
if matches!(edit.reflog, Reflog::AppendExistingIfChanged { .. }) {
let mut existing = Vec::new();
for (name, record) in std::mem::take(&mut operation.logs) {
if self.has_reflog(&name).map_err(batch_error)? {
existing.push((name, record));
}
}
operation.logs = existing;
}
if !operation.delete_log {
for (name, _) in &operation.logs {
reflog::check_append_tail(&log_locks[name].destination)
.map_err(batch_error)?;
}
}
}
let mut replacement = bytes;
for operation in &operations {
if operation.packed_removed {
replacement =
packed::without_ref(self.object_format, &replacement, &operation.name);
}
}
Ok(Prepared {
packed_lock,
replacement,
locks,
log_locks,
operations,
})
};
#[cfg(feature = "tracing")]
let result = span.in_scope(operation);
#[cfg(not(feature = "tracing"))]
let result = { operation }();
#[cfg(feature = "tracing")]
crate::trace::finish(&span, &result, |error| {
crate::trace::transaction(error, &span)
});
result
}
fn discover_chain(
&self,
edit: &RefEdit,
packed: &packed::Packed,
identity: LogIdentity,
) -> Result<Vec<(RefName, Option<Target>)>, ReferenceError> {
let resolve_old = identity == LogIdentity::Resolved
&& (edit.dereference == crate::refs::Dereference::Follow
|| edit.reflog.append_fields().is_some());
let mut chain = Vec::new();
let mut name = edit.name.clone();
loop {
if chain.iter().any(|(seen, _)| seen == &name) {
return Err(ReferenceError::Cycle(name));
}
let target = self.read_locked(&name, packed)?;
chain.push((name.clone(), target.clone()));
match target {
Some(Target::Symbolic(next)) if resolve_old => {
if chain.len() > 32 {
return Err(ReferenceError::Depth(32));
}
name = next;
}
_ => return Ok(chain),
}
}
}
}
pub(super) fn validate_edit(
format: crate::ObjectFormat,
edit: &RefEdit,
) -> Result<(), ReferenceError> {
super::store::validate_expected(format, &edit.expected)?;
if edit.reflog == Reflog::Delete && edit.target.is_some() {
return Err(ReferenceError::Unsupported(
"log deletion requires reference deletion",
));
}
if let Some(target) = &edit.target {
validate_target(format, target)?;
if edit.name.as_bytes() == b"HEAD"
&& matches!(target, Target::Symbolic(next) if next.as_bytes() == b"HEAD")
{
return Err(ReferenceError::InvalidHeadTarget);
}
if edit.dereference == crate::refs::Dereference::Follow
&& matches!(target, Target::Symbolic(_))
{
return Err(ReferenceError::Unsupported("resolved symbolic replacement"));
}
}
if let Some((committer, message)) = edit.reflog.append_fields() {
reflog::validate(committer, message)?;
}
Ok(())
}
#[derive(Clone, Copy, Eq, PartialEq)]
enum LogIdentity {
Resolved,
Stored,
}
fn log_id(
format: crate::ObjectFormat,
target: Option<&Target>,
identity: LogIdentity,
) -> Result<ObjectId, ReferenceError> {
match target {
Some(Target::Symbolic(_)) if identity == LogIdentity::Stored => Ok(ObjectId::null(format)),
None => Ok(ObjectId::null(format)),
Some(Target::Direct(id)) => Ok(*id),
Some(Target::Symbolic(_)) => Err(ReferenceError::Unsupported(
"stored symbolic edits with reflogs",
)),
}
}
struct Operation {
name: RefName,
target: Option<Target>,
packed_removed: bool,
delete_log: bool,
logs: Vec<(RefName, Vec<u8>)>,
}
pub(crate) struct Prepared {
packed_lock: Lock,
replacement: Vec<u8>,
locks: BTreeMap<RefName, Lock>,
log_locks: BTreeMap<RefName, Lock>,
operations: Vec<Operation>,
}
impl Prepared {
pub(crate) fn publish(self) -> Result<Vec<RefEditOutcome>, TransactionError> {
let _lock_operation = crate::locks::operation();
#[cfg(feature = "tracing")]
let span = tracing::debug_span!(
target: "girt",
"refs.publish",
outcome = "incomplete",
failure_class = tracing::field::Empty,
effects = tracing::field::Empty,
);
let operation = || {
let mut outcomes: Vec<_> = self
.operations
.iter()
.map(|op| RefEditOutcome {
name: op.name.clone(),
reference: RefOutcome::Unchanged,
logs: op
.logs
.iter()
.map(|(name, _)| (name.clone(), LogOutcome::NotAttempted))
.collect(),
})
.collect();
for lock in std::iter::once(&self.packed_lock)
.chain(self.locks.values())
.chain(self.log_locks.values())
{
lock.check_owned()
.map_err(|source| TransactionError::Publish {
outcomes: outcomes.clone(),
source,
})?;
}
if self.operations.iter().any(|op| op.packed_removed) {
self.packed_lock
.publish_retaining_lock(&self.replacement)
.map_err(|source| TransactionError::Publish {
outcomes: outcomes.clone(),
source,
})?;
for (operation, outcome) in self.operations.iter().zip(&mut outcomes) {
if operation.packed_removed {
outcome.reference = RefOutcome::PackedRemoved;
}
}
}
for (index, operation) in self.operations.iter().enumerate() {
let lock = &self.locks[&operation.name];
let result = match &operation.target {
Some(target) => {
let bytes = match target {
Target::Direct(id) => format!("{id}\n").into_bytes(),
Target::Symbolic(name) => {
let mut bytes = b"ref: ".to_vec();
bytes.extend_from_slice(name.as_bytes());
bytes.push(b'\n');
bytes
}
};
lock.publish_retaining_lock(&bytes)
}
None => lock
.check_owned()
.and_then(|()| remove_loose(&lock.destination, operation.packed_removed)),
};
if let Err(source) = result {
return Err(TransactionError::Publish { outcomes, source });
}
outcomes[index].reference = RefOutcome::Published;
for (log_index, (name, record)) in operation.logs.iter().enumerate() {
let path = &self.log_locks[name].destination;
let result = self.log_locks[name]
.check_owned()
.map_err(|error| (0, error))
.and_then(|()| {
if operation.delete_log {
remove_loose(path, false).map_err(|error| (0, error))
} else {
append(path, record)
}
});
match result {
Ok(()) => {
outcomes[index].logs[log_index].1 = if operation.delete_log {
LogOutcome::Deleted
} else {
LogOutcome::Appended
}
}
Err((bytes_written, source)) => {
outcomes[index].logs[log_index].1 =
LogOutcome::Failed { bytes_written };
return Err(TransactionError::Publish { outcomes, source });
}
}
}
}
Ok(outcomes)
};
#[cfg(feature = "tracing")]
let result = span.in_scope(operation);
#[cfg(not(feature = "tracing"))]
let result = { operation }();
#[cfg(feature = "tracing")]
crate::trace::finish(&span, &result, |error| {
crate::trace::transaction(error, &span)
});
result
}
}
fn append(path: &std::path::Path, record: &[u8]) -> Result<(), (usize, ReferenceError)> {
let mut file = OpenOptions::new()
.append(true)
.create(true)
.open(path)
.map_err(|source| (0, io_error(path, source)))?;
append_record(&mut file, record).map_err(|(count, source)| (count, io_error(path, source)))
}
fn append_record(writer: &mut impl Write, record: &[u8]) -> Result<(), (usize, io::Error)> {
let mut count = 0;
while count < record.len() {
match writer.write(&record[count..]) {
Ok(0) => return Err((count, io::Error::from(io::ErrorKind::WriteZero))),
Ok(written) => count += written,
Err(error) if error.kind() == io::ErrorKind::Interrupted => {}
Err(error) => return Err((count, error)),
}
}
Ok(())
}
#[cfg(test)]
#[path = "transaction_tests.rs"]
mod tests;
pub(crate) enum PreparedBackend {
Files(Box<Prepared>),
Reftable(super::reftable::backend::Prepared),
}
impl PreparedBackend {
pub(crate) fn publish(self) -> Result<Vec<RefEditOutcome>, TransactionError> {
let _lock_operation = crate::locks::operation();
match self {
Self::Files(prepared) => prepared.publish(),
Self::Reftable(prepared) => prepared.publish(),
}
}
}