use super::{
LocalGitService,
observation::{ObservationInput, ObservationOwner, ObservationRecord},
observation_store::{Cursor, Plan, Spool},
};
use crate::{
files::{absent_structure, io_error, save_request_digest, target_path},
operations::ledger::SaveObservation,
};
use buntrail_core::contracts::*;
use sha2::{Digest, Sha256};
use std::{
io::Read,
sync::{
Arc,
atomic::{AtomicBool, Ordering},
},
time::Instant,
};
fn refused(message: &str) -> CoreError {
CoreError::new(ErrorCode::StaleRevision, message)
}
fn check_proof(cancel: &AtomicBool, started: Instant) -> Result<(), CoreError> {
if cancel.load(Ordering::SeqCst) || started.elapsed() >= super::process::DEADLINE {
Err(CoreError::new(
ErrorCode::Cancelled,
"proof verification cancelled or timed out",
))
} else {
Ok(())
}
}
fn request_digest(original: &GitMutationRequest) -> Result<[u8; 32], CoreError> {
super::observation::serialized_digest(original)
}
#[derive(serde::Serialize, serde::Deserialize)]
struct Evidence {
operation: OperationID,
file: FileRef,
before: Option<SaveObservation>,
after: SaveObservation,
captured_absence: bool,
}
fn read_evidence(reader: &mut impl Read) -> Result<Option<Evidence>, CoreError> {
let mut size = [0; 4];
if reader.read(&mut size[..1]).map_err(io_error)? == 0 {
return Ok(None);
}
reader.read_exact(&mut size[1..]).map_err(io_error)?;
let size = u32::from_be_bytes(size) as usize;
if size > 128 * 1024 {
return Err(refused(
"save evidence exceeds existing ledger record bound",
));
}
let mut bytes = vec![0; size];
reader.read_exact(&mut bytes).map_err(io_error)?;
serde_json::from_slice(&bytes).map(Some).map_err(io_error)
}
fn read_approval(
reader: &mut impl Read,
owner: &ObservationOwner,
) -> Result<Option<GitSaveApproval>, CoreError> {
let mut bytes = [0; 96];
if reader.read(&mut bytes[..1]).map_err(io_error)? == 0 {
return Ok(None);
}
reader.read_exact(&mut bytes[1..]).map_err(io_error)?;
Ok(Some(GitSaveApproval {
operation: OperationID(bytes[..16].try_into().unwrap()),
request_digest: bytes[16..48].try_into().unwrap(),
actor: WorkspaceLease {
client: ClientID(bytes[48..64].try_into().unwrap()),
workspace: WorkspaceID(bytes[64..80].try_into().unwrap()),
lease: WorkspaceLeaseID(bytes[80..96].try_into().unwrap()),
runtime: owner.registration().runtime.clone(),
},
}))
}
fn cursor(spool: &Arc<Spool>) -> Cursor {
Cursor {
spool: spool.clone(),
offset: 0,
}
}
fn approval_stream(plan: &Plan, additional: Option<&Arc<Spool>>) -> impl Read {
let tail: Box<dyn Read> = match additional {
Some(spool) => Box::new(cursor(spool)),
None => Box::new(std::io::empty()),
};
plan.approvals().chain(tail)
}
fn endpoint(record: &ObservationRecord, version: &FileVersion) -> bool {
record.regular.as_ref().is_some_and(|regular| {
regular.canonical.as_slice() == version.opaque_token.as_slice()
&& regular.content_digest == version.content_digest
})
}
fn valid_endpoint(endpoint: &SaveObservation, target: &TargetIdentity) -> bool {
endpoint.requested.target == *target
&& endpoint.canonical.target == *target
&& endpoint.requested.opaque_token.len() == 32
&& endpoint.canonical.opaque_token.len() == 32
&& endpoint.requested.file_identity.len() == 16
&& endpoint.requested.file_identity == endpoint.canonical.file_identity
&& endpoint.requested.content_digest == endpoint.canonical.content_digest
}
#[derive(Debug)]
pub enum WorktreePreparation {
Existing(Box<OperationReceipt>),
Prepared([u8; 16]),
}
impl LocalGitService {
pub fn save_approval_input(
&self,
owner: &ObservationOwner,
approvals: impl IntoIterator<Item = GitSaveApproval>,
) -> Result<ObservationInput, CoreError> {
let mut writer = self.observation_input(owner)?;
for approval in approvals {
if approval.actor.client != owner.registration().client
|| approval.actor.runtime != owner.registration().runtime
{
return Err(CoreError::new(
ErrorCode::TargetChanged,
"save approval client or runtime changed",
));
}
for bytes in [
approval.operation.0.as_slice(),
approval.request_digest.as_slice(),
approval.actor.client.0.as_slice(),
approval.actor.workspace.0.as_slice(),
approval.actor.lease.0.as_slice(),
] {
writer.append(bytes)?;
}
}
writer.seal()
}
pub fn observation_alias_digest(
&self,
owner: &ObservationOwner,
observation: [u8; 16],
) -> Result<[u8; 32], CoreError> {
let mut reader = self.observation_reader(owner, observation)?;
let mut hash = Sha256::new();
hash.update(b"buntrail.original-aliases.v1\0");
while let Some(record) = reader.next_record()? {
if let Some(identity) = record.file_reference {
hash.update(identity);
}
}
Ok(hash.finalize().into())
}
pub fn prepare_worktree_sync(
&self,
owner: &ObservationOwner,
original: &GitMutationRequest,
observation: [u8; 16],
alias_digest: [u8; 32],
approvals: ObservationInput,
cancel: &AtomicBool,
) -> Result<WorktreePreparation, CoreError> {
if let Some(receipt) = self.existing_receipt_sync(original)? {
return Ok(WorktreePreparation::Existing(Box::new(receipt)));
}
#[cfg(test)]
if let Some(checkpoint) = &self.after_proof_lookup {
checkpoint();
}
let mutex = super::mutations::repository_mutex(&original.worktree);
let _repository = mutex.lock().map_err(io_error)?;
if let Some(receipt) = self.existing_receipt_sync(original)? {
return Ok(WorktreePreparation::Existing(Box::new(receipt)));
}
if !matches!(
original.command,
GitMutation::Checkout { .. } | GitMutation::PullFastForward { .. }
) {
return Err(refused("only worktree changes use prepared execution"));
}
let store = self.observation_store()?;
let pin = self.pin_observation(owner, observation)?;
let (worktree, state) = pin.original_state()?;
if worktree != &original.worktree
|| state != &original.expected
|| self.observation_alias_digest(owner, observation)? != alias_digest
{
return Err(refused(
"request differs from its original owned observation or aliases",
));
}
let _work = store.verification()?;
let plan = store.adopt_plan(owner, pin, original, alias_digest, approvals)?;
let result = (|| {
let evidence = self.freeze_evidence(owner, &plan, None, cancel)?;
self.verify_worktree(&plan, &evidence, cancel)?;
store.plan(owner, plan.id)?;
Ok(WorktreePreparation::Prepared(plan.id))
})();
if result.is_err() {
store.cancel_plan(owner, plan.id);
}
result
}
pub fn execute_worktree_sync(
&self,
owner: &ObservationOwner,
original: GitMutationRequest,
plan_id: [u8; 16],
alias_digest: [u8; 32],
approvals: ObservationInput,
cancel: &AtomicBool,
) -> Result<OperationReceipt, CoreError> {
if let Some(receipt) = self.existing_receipt_sync(&original)? {
return Ok(receipt);
}
#[cfg(test)]
if let Some(checkpoint) = &self.after_proof_lookup {
checkpoint();
}
let digest = request_digest(&original)?;
let operation = original.operation;
let store = self.observation_store()?;
let mut input = Some(approvals);
let mut execution = None;
let mut verify = |accepted: bool| {
if !accepted {
let plan = store.plan(owner, plan_id)?;
if plan.request_digest != digest
|| plan.operation != operation
|| plan.alias_digest != alias_digest
{
return Err(refused(
"prepared original request or alias identity changed",
));
}
let work = store.verification()?;
let additional = store.adopt_execution_input(
owner,
input
.take()
.ok_or_else(|| refused("execution input already consumed"))?,
)?;
let evidence = self.freeze_evidence(owner, &plan, Some(&additional), cancel)?;
let expected = self.verify_worktree(&plan, &evidence, cancel)?;
store.plan(owner, plan_id)?;
execution = Some((plan, evidence, work));
Ok(expected)
} else {
let (plan, evidence, _) = execution
.as_ref()
.ok_or_else(|| refused("missing accepted proof snapshot"))?;
self.verify_worktree(plan, evidence, cancel)
}
};
let expected = original.expected.clone();
let authorize = || store.plan(owner, plan_id).map(|_| ());
let receipt = self.mutate_guarded(
original,
expected,
cancel,
Some(super::mutations::WorktreeProofGuard {
owner,
verify: &mut verify,
authorize: &authorize,
}),
);
store.cancel_plan(owner, plan_id);
receipt
}
pub fn cancel_worktree_plan(&self, owner: &ObservationOwner, plan: [u8; 16]) {
if let Ok(store) = self.observation_store() {
store.cancel_plan(owner, plan);
}
}
fn freeze_evidence(
&self,
owner: &ObservationOwner,
plan: &Plan,
additional: Option<&Arc<Spool>>,
cancel: &AtomicBool,
) -> Result<Arc<Spool>, CoreError> {
let started = Instant::now();
let mut writer = self.observation_input(owner)?;
let mut approvals = approval_stream(plan, additional);
let mut seen = 0u64;
while let Some(approval) = read_approval(&mut approvals, owner)? {
check_proof(cancel, started)?;
let mut prior = approval_stream(plan, additional);
for _ in 0..seen {
check_proof(cancel, started)?;
if read_approval(&mut prior, owner)?
.is_some_and(|old| old.operation == approval.operation)
{
return Err(refused("duplicate approved Save operation"));
}
}
seen += 1;
let record = {
let ledger = self.files.ledger.lock().map_err(io_error)?;
let _lock = ledger.lock()?;
let record = ledger
.read(approval.operation)?
.ok_or_else(|| refused("approved Save history is missing"))?;
if ledger.expired(&record) {
return Err(refused("approved Save history expired"));
}
record
};
let proof = record
.save_proof
.as_ref()
.ok_or_else(|| refused("Save has no authenticated history"))?;
let after = proof
.after
.as_ref()
.ok_or_else(|| refused("Save has no durable historical after"))?;
let ExpectedState::File { file, version } = &record.envelope.expected else {
return Err(refused("Save expected state is malformed"));
};
let OperationState::Confirmed(MutationResult::Saved(saved)) = &record.receipt.state
else {
return Err(refused("Save is not durably confirmed"));
};
let destination = record
.destination
.as_ref()
.ok_or_else(|| refused("Save has no original destination"))?;
let resolution_token: [u8; 32] =
Sha256::digest(serde_json::to_vec(destination).map_err(io_error)?).into();
if record.envelope.kind != MutationKind::Save
|| record.envelope.target != owner.target
|| file.target != owner.target
|| record.envelope.request_digest != approval.request_digest
|| proof.actor != approval.actor
|| proof.actor.client != owner.registration().client
|| proof.actor.runtime != owner.registration().runtime
|| saved.saved_revision != record.revision
|| saved.version != after.requested
|| !valid_endpoint(after, &owner.target)
|| after.requested.content_digest != record.requested_digest
|| after.resolved != destination.resolved
|| after.resolution_token != resolution_token
|| save_request_digest(file, version, record.revision, record.requested_digest)?
!= approval.request_digest
{
return Err(refused(
"approved Save identity or historical result differs",
));
}
match (&proof.before, version) {
(Some(before), Some(expected))
if before.requested == *expected
&& valid_endpoint(before, &owner.target)
&& before.resolved == after.resolved
&& before.resolution_token == after.resolution_token => {}
(None, None) => {}
_ => return Err(refused("Save before/after chain is malformed")),
}
let mut original = plan.original();
let mut covered = false;
let mut captured_absence = false;
while let Some(node) = ObservationRecord::read(&mut original)? {
check_proof(cancel, started)?;
if node.path != file.path
|| node
.membership
.as_ref()
.is_some_and(|membership| !membership.affected)
{
continue;
}
if node.membership.is_some() {
if node.file_reference.is_none()
|| (file.resolved_identity.is_some() && !node.matches_alias(file)?)
{
continue;
}
} else if file.resolved_identity.is_some() {
continue;
}
if node.resolved.as_ref() != Some(&after.resolved) {
continue;
}
if node.regular.is_none()
&& node.structural_token != absent_structure(&Some(destination.clone()))?
{
continue;
}
if node
.regular
.as_ref()
.is_some_and(|regular| regular.resolution != resolution_token)
{
continue;
}
covered = true;
captured_absence = node.regular.is_none();
break;
}
if !covered {
return Err(refused(
"Save requested alias was not covered by the original observation",
));
}
let evidence = Evidence {
operation: approval.operation,
file: file.clone(),
before: proof.before.clone(),
after: after.clone(),
captured_absence,
};
let bytes = serde_json::to_vec(&evidence).map_err(io_error)?;
if bytes.len() > 128 * 1024 {
return Err(refused("Save evidence exceeds its bounded durable record"));
}
writer.append(&(bytes.len() as u32).to_be_bytes())?;
writer.append(&bytes)?;
}
self.observation_store()?
.adopt_execution_input(owner, writer.seal()?)
}
fn compare_node(
&self,
original: &ObservationRecord,
current: &ObservationRecord,
current_absence: [u8; 32],
evidence: &Arc<Spool>,
cancel: &AtomicBool,
started: Instant,
) -> Result<(), CoreError> {
let unaffected = original
.membership
.as_ref()
.is_some_and(|membership| !membership.affected);
if original.path != current.path
|| original.resolved != current.resolved
|| original.membership != current.membership
|| original.file_reference != current.file_reference
{
return Err(refused("original path, alias or membership changed"));
}
if unaffected {
return if original.structural_token == current.structural_token {
Ok(())
} else {
Err(refused("unaffected alias structure changed"))
};
}
let mut reader = cursor(evidence);
let mut last: Option<Evidence> = None;
let mut matches = 0;
while let Some(save) = read_evidence(&mut reader)? {
check_proof(cancel, started)?;
if original.resolved.as_ref() != Some(&save.after.resolved) {
continue;
}
if let Some(previous) = &last {
if save
.before
.as_ref()
.is_none_or(|before| before.canonical != previous.after.canonical)
{
return Err(refused("approved canonical Save chain is not contiguous"));
}
} else if save
.before
.as_ref()
.is_some_and(|before| endpoint(original, &before.canonical))
|| (save.before.is_none()
&& original.regular.is_none()
&& original.structural_token == current_absence)
{
matches += 1;
}
if endpoint(original, &save.after.canonical) {
matches += 1;
}
last = Some(save);
}
let Some(last) = last else {
return if original == current {
Ok(())
} else {
Err(refused(
"unapproved node changed since original observation",
))
};
};
if matches != 1 || !endpoint(current, &last.after.canonical) {
return Err(refused(
"original or final node is not an exact Save-chain boundary",
));
}
if let Some(regular) = &original.regular {
if original.structural_token != current.structural_token
|| regular.resolution != current.regular.as_ref().unwrap().resolution
{
return Err(refused("saved node structure or resolution changed"));
}
} else if original.structural_token != current_absence {
return Err(refused("created node lost its original absence structure"));
}
if last.file.path == current.path
&& current.regular.as_ref().unwrap().requested.as_slice()
!= last.after.requested.opaque_token.as_slice()
{
return Err(refused(
"saved requested endpoint differs from historical after",
));
}
Ok(())
}
fn verify_worktree(
&self,
plan: &Plan,
evidence: &Arc<Spool>,
cancel: &AtomicBool,
) -> Result<GitExpectedState, CoreError> {
let worktree = &plan.pin.data.worktree;
let root = self.validate(worktree, cancel)?;
let started = Instant::now();
let mut reader = plan.original();
let mut next = ObservationRecord::read(&mut reader)?;
let (state, _) = self.state_records(&root, cancel, |name, node| {
use std::os::unix::ffi::OsStrExt;
let current = ObservationRecord::from_node(
target_path(&root.join(std::ffi::OsStr::from_bytes(name))),
node,
None,
)?;
if next.as_ref().is_some_and(|original| {
original.membership.is_none() && original.path == current.path
}) {
self.compare_node(
next.as_ref().unwrap(),
¤t,
absent_structure(&node.destination)?,
evidence,
cancel,
started,
)?;
next = ObservationRecord::read(&mut reader)?;
} else {
let mut approved = cursor(evidence);
let mut created = false;
let mut final_matches = false;
while let Some(save) = read_evidence(&mut approved)? {
check_proof(cancel, started)?;
if save.after.resolved == current.path {
created |= save.before.is_none() && save.captured_absence;
final_matches = endpoint(¤t, &save.after.canonical);
}
}
if !created || !final_matches {
return Err(refused("unexpected newly enumerated Git path"));
}
}
Ok(())
})?;
if state.head != plan.pin.data.expected.head
|| state.index_fingerprint != plan.pin.data.expected.index_fingerprint
{
return Err(refused("original HEAD or index changed"));
}
while let Some(original) = next {
if original.membership.is_none() {
return Err(refused("original Git path disappeared"));
}
let (mut current, destination) = self.alias_node_record(
worktree,
FileRef {
target: worktree.target.clone(),
path: original.path.clone(),
resolved_identity: None,
},
cancel,
started,
)?;
current.file_reference = original.file_reference;
self.compare_node(
&original,
¤t,
absent_structure(&destination)?,
evidence,
cancel,
started,
)?;
next = ObservationRecord::read(&mut reader)?;
}
if self.state(&root, cancel)? != state {
return Err(refused("worktree changed during proof validation"));
}
check_proof(cancel, started)?;
Ok(state)
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::{connections::local::LocalConnection, files::target_path};
use std::{path::Path, sync::Arc};
#[test]
fn proof_scan_cannot_run_past_the_existing_git_deadline() {
assert_eq!(
check_proof(
&AtomicBool::new(false),
Instant::now() - std::time::Duration::from_secs(120)
)
.unwrap_err()
.code,
ErrorCode::Cancelled
);
assert!(check_proof(&AtomicBool::new(false), Instant::now()).is_ok());
}
fn git(root: &Path, args: &[&str]) {
let output = std::process::Command::new("git")
.arg("-C")
.arg(root)
.args(args)
.env("GIT_CONFIG_NOSYSTEM", "1")
.env("GIT_CONFIG_GLOBAL", "/dev/null")
.output()
.unwrap();
assert!(
output.status.success(),
"{}",
String::from_utf8_lossy(&output.stderr)
);
}
struct Fixture {
dir: tempfile::TempDir,
local: Arc<LocalConnection>,
owner: ObservationOwner,
worktree: WorktreeIdentity,
file: FileRef,
}
impl Fixture {
fn new() -> Self {
let dir = tempfile::tempdir().unwrap();
let root = dir.path().join("repo");
std::fs::create_dir(&root).unwrap();
git(&root, &["init", "-b", "main"]);
git(&root, &["config", "user.name", "Fixture"]);
git(&root, &["config", "user.email", "fixture@example.invalid"]);
std::fs::write(root.join("file"), "original").unwrap();
git(&root, &["add", "file"]);
git(&root, &["commit", "-m", "initial"]);
git(&root, &["branch", "other"]);
let local = LocalConnection::open(dir.path().join("state")).unwrap();
let lease = WorkspaceLeaseID([1; 16]);
local.register(WorkspaceID([2; 16]), lease).unwrap();
let owner = local.observation_owner(lease).unwrap();
let file = FileRef {
target: local.identity.clone(),
path: target_path(&root.join("file")),
resolved_identity: None,
};
let worktree = local
.git
.discover_sync(
FileRef {
path: target_path(&root),
..file.clone()
},
&AtomicBool::new(false),
)
.unwrap()
.unwrap();
Self {
dir,
local: Arc::new(local),
owner,
worktree,
file,
}
}
fn save(&self, file: FileRef, id: u8, bytes: &[u8]) -> GitSaveApproval {
self.save_as(self.owner.registration().lease, file, id, bytes)
}
fn save_as(
&self,
lease: WorkspaceLeaseID,
file: FileRef,
id: u8,
bytes: &[u8],
) -> GitSaveApproval {
let expected_version = self
.local
.files
.read_sync(file.clone())
.ok()
.map(|snapshot| snapshot.version);
let save = self
.local
.owned_files(lease)
.unwrap()
.prepare(SaveRequest {
operation: OperationID([id; 16]),
file,
expected_version,
submitted_revision: BufferRevision(id as u64),
source: ByteSource::Memory(Arc::from(bytes)),
})
.unwrap();
let approval = save.approval().clone();
assert!(matches!(save.execute().state, OperationState::Confirmed(_)));
approval
}
fn capture(&self, aliases: Vec<FileRef>) -> (GitMutationRequest, [u8; 16], [u8; 32]) {
let retained = self
.local
.git
.observe_owned_sync(
&self.owner,
self.worktree.clone(),
aliases.into_iter().map(Ok),
&AtomicBool::new(false),
)
.unwrap();
let digest = self
.local
.git
.observation_alias_digest(&self.owner, retained.id)
.unwrap();
(
GitMutationRequest {
operation: OperationID([90; 16]),
worktree: self.worktree.clone(),
expected: retained.snapshot.state,
command: GitMutation::Checkout {
reference: b"other".to_vec(),
},
},
retained.id,
digest,
)
}
fn execute(
&self,
original: GitMutationRequest,
observation: [u8; 16],
digest: [u8; 32],
approvals: Vec<GitSaveApproval>,
) {
let input = self
.local
.git
.save_approval_input(&self.owner, approvals)
.unwrap();
let WorktreePreparation::Prepared(plan) = self
.local
.git
.prepare_worktree_sync(
&self.owner,
&original,
observation,
digest,
input,
&AtomicBool::new(false),
)
.expect("durable save chain must prepare")
else {
panic!("unexpected duplicate")
};
let additional = self
.local
.git
.save_approval_input(&self.owner, [])
.expect("adopted approvals must release only the staging slot");
let receipt = self
.local
.git
.execute_worktree_sync(
&self.owner,
original.clone(),
plan,
digest,
additional,
&AtomicBool::new(false),
)
.unwrap();
assert!(
matches!(receipt.state, OperationState::Confirmed(_)),
"{receipt:?}"
);
assert_eq!(
self.local.git.existing_receipt_sync(&original).unwrap(),
Some(receipt)
);
}
}
#[test]
fn original_pending_save_chain_can_prepare_and_execute() {
let f = Fixture::new();
let (original, observation, digest) = f.capture(vec![f.file.clone()]);
let approval = f.save(f.file.clone(), 20, b"saved");
f.execute(original, observation, digest, vec![approval]);
}
#[test]
fn already_reflected_save_is_not_applied_twice() {
let f = Fixture::new();
let first = f.save(f.file.clone(), 21, b"first");
let (original, observation, digest) = f.capture(vec![f.file.clone()]);
let second = f.save(f.file.clone(), 22, b"second");
f.execute(original, observation, digest, vec![first, second]);
}
#[test]
fn captured_absence_allows_only_the_actual_owned_creation() {
let f = Fixture::new();
let absent = FileRef {
path: target_path(&f.dir.path().join("repo/new")),
..f.file.clone()
};
let (original, observation, digest) = f.capture(vec![absent.clone()]);
let approval = f.save(absent, 23, b"created");
f.execute(original, observation, digest, vec![approval]);
}
#[test]
fn captured_creation_then_later_save_forms_one_contiguous_chain() {
let f = Fixture::new();
let absent = FileRef {
path: target_path(&f.dir.path().join("repo/new")),
..f.file.clone()
};
let (original, observation, digest) = f.capture(vec![absent.clone()]);
let first = f.save(absent.clone(), 24, b"created");
let second = f.save(absent, 25, b"updated");
f.execute(original, observation, digest, vec![first, second]);
}
#[test]
fn interleaved_cross_window_saves_through_outside_aliases_are_contiguous() {
let f = Fixture::new();
let root = f.dir.path().join("repo");
std::fs::write(root.join("second"), "second").unwrap();
git(&root, &["add", "second"]);
git(&root, &["commit", "-m", "second"]);
git(&root, &["branch", "-f", "other"]);
let second = FileRef {
path: target_path(&root.join("second")),
..f.file.clone()
};
let alias_path = f.dir.path().join("outside");
std::os::unix::fs::symlink(root.join("file"), &alias_path).unwrap();
let alias = FileRef {
path: target_path(&alias_path),
..f.file.clone()
};
let lease = WorkspaceLeaseID([3; 16]);
f.local.register(WorkspaceID([4; 16]), lease).unwrap();
let (original, observation, digest) =
f.capture(vec![f.file.clone(), alias.clone(), second.clone()]);
let first = f.save_as(lease, alias, 30, b"alias save");
let other = f.save(second, 31, b"second save");
let last = f.save(f.file.clone(), 32, b"last save");
f.local.release(lease);
f.execute(original, observation, digest, vec![first, other, last]);
}
#[test]
fn external_changes_never_become_approved_save_history() {
use std::os::unix::fs::PermissionsExt;
for case in [
"same-content",
"mode",
"head",
"index",
"new-path",
"missing-chain",
"duplicate",
"reverse",
"external-create",
] {
let f = Fixture::new();
let root = f.dir.path().join("repo");
let file = if case == "external-create" {
FileRef {
path: target_path(&root.join("new")),
..f.file.clone()
}
} else {
f.file.clone()
};
let (original, observation, digest) = f.capture(vec![file.clone()]);
if case == "external-create" {
std::fs::write(root.join("new"), "external").unwrap();
}
let first = f.save(file.clone(), 40, b"saved");
let mut approvals = vec![first.clone()];
match case {
"same-content" => {
std::fs::write(root.join("replacement"), b"saved").unwrap();
std::fs::rename(root.join("replacement"), root.join("file")).unwrap();
}
"mode" => std::fs::set_permissions(
root.join("file"),
std::fs::Permissions::from_mode(0o600),
)
.unwrap(),
"head" => git(&root, &["checkout", "other"]),
"index" => git(&root, &["add", "file"]),
"new-path" => std::fs::write(root.join("unapproved"), "external").unwrap(),
"missing-chain" => approvals = vec![f.save(file, 41, b"later")],
"duplicate" => approvals.push(first),
"reverse" => approvals.insert(0, f.save(file, 41, b"later")),
_ => {}
}
let input = f
.local
.git
.save_approval_input(&f.owner, approvals)
.unwrap();
assert!(
f.local
.git
.prepare_worktree_sync(
&f.owner,
&original,
observation,
digest,
input,
&AtomicBool::new(false)
)
.is_err(),
"{case}"
);
assert!(
f.local
.git
.existing_receipt_sync(&original)
.unwrap()
.is_none(),
"{case}"
);
}
}
#[test]
fn missing_expired_malformed_and_reconciled_records_cannot_prove_saves() {
for case in [
"missing", "expired", "legacy", "unknown", "no-after", "digest", "actor", "runtime",
"kind", "revision", "result", "before",
] {
let f = Fixture::new();
let (original, observation, digest) = f.capture(vec![f.file.clone()]);
let mut approval = f.save(f.file.clone(), 45, b"saved");
{
let ledger = f.local.files.ledger.lock().unwrap();
let _lock = ledger.lock().unwrap();
let mut record = ledger.read(approval.operation).unwrap().unwrap();
match case {
"missing" => {
std::fs::remove_file(
ledger.directory.join(format!("{}.json", "2d".repeat(16))),
)
.unwrap();
}
"expired" => record.created = 0,
"legacy" => record.save_proof = None,
"unknown" => {
record.receipt = crate::operations::ledger::unknown(
record.envelope.id,
record.envelope.target.clone(),
"fixture uncertain",
)
}
"no-after" => record.save_proof.as_mut().unwrap().after = None,
"digest" => approval.request_digest[0] ^= 1,
"actor" => approval.actor.lease = WorkspaceLeaseID([9; 16]),
"runtime" => {
record
.save_proof
.as_mut()
.unwrap()
.actor
.runtime
.supervisor_boot[0] ^= 1
}
"kind" => record.envelope.kind = MutationKind::Git,
"revision" => record.revision = BufferRevision(99),
"result" => {
let OperationState::Confirmed(MutationResult::Saved(saved)) =
&mut record.receipt.state
else {
panic!()
};
saved.version.opaque_token[0] ^= 1;
}
"before" => {
record
.save_proof
.as_mut()
.unwrap()
.before
.as_mut()
.unwrap()
.canonical
.opaque_token[0] ^= 1
}
_ => unreachable!(),
}
if case != "missing" {
ledger.write(&record).unwrap();
}
}
let input = f
.local
.git
.save_approval_input(&f.owner, [approval])
.unwrap();
assert!(
f.local
.git
.prepare_worktree_sync(
&f.owner,
&original,
observation,
digest,
input,
&AtomicBool::new(false)
)
.is_err(),
"{case}"
);
assert!(
f.local
.git
.existing_receipt_sync(&original)
.unwrap()
.is_none()
);
}
}
#[test]
fn exact_registration_survives_durable_admission_or_refuses_without_effect() {
for phase in ["before", "during", "after"] {
let f = Fixture::new();
let (original, observation, digest) = f.capture(vec![f.file.clone()]);
let input = f.local.git.save_approval_input(&f.owner, []).unwrap();
let WorktreePreparation::Prepared(plan) = f
.local
.git
.prepare_worktree_sync(
&f.owner,
&original,
observation,
digest,
input,
&AtomicBool::new(false),
)
.unwrap()
else {
panic!()
};
let mut service = LocalGitService::new(f.local.files.clone());
let local = f.local.clone();
let lease = f.owner.registration().lease;
let hook: Arc<dyn Fn() + Send + Sync> = Arc::new(move || local.release(lease));
match phase {
"before" => service.before_accept = Some(hook),
"during" => service.after_accept = Some(hook),
_ => service.after_authorization = Some(hook),
}
let input = service.save_approval_input(&f.owner, []).unwrap();
let result = service.execute_worktree_sync(
&f.owner,
original.clone(),
plan,
digest,
input,
&AtomicBool::new(false),
);
if phase == "before" {
assert!(result.is_err());
assert!(service.existing_receipt_sync(&original).unwrap().is_none());
} else {
let receipt = result.unwrap();
assert!(
if phase == "after" {
matches!(receipt.state, OperationState::Confirmed(_))
} else {
matches!(receipt.state, OperationState::Failed(_))
},
"{phase}: {receipt:?}"
);
assert_eq!(
service.existing_receipt_sync(&original).unwrap(),
Some(receipt)
);
}
}
}
#[test]
fn historical_canonical_identity_must_agree_with_the_same_saved_descriptor() {
let f = Fixture::new();
let (original, observation, digest) = f.capture(vec![f.file.clone()]);
let approval = f.save(f.file.clone(), 46, b"saved");
{
let ledger = f.local.files.ledger.lock().unwrap();
let _lock = ledger.lock().unwrap();
let mut record = ledger.read(approval.operation).unwrap().unwrap();
record
.save_proof
.as_mut()
.unwrap()
.after
.as_mut()
.unwrap()
.canonical
.file_identity[0] ^= 1;
ledger.write(&record).unwrap();
}
let input = f
.local
.git
.save_approval_input(&f.owner, [approval])
.unwrap();
assert!(
f.local
.git
.prepare_worktree_sync(
&f.owner,
&original,
observation,
digest,
input,
&AtomicBool::new(false)
)
.is_err(),
"malformed canonical endpoint was accepted"
);
}
#[test]
fn duplicate_receipt_precedes_invalid_plan_owner_cancellation_and_store() {
let f = Fixture::new();
let (original, _, _) = f.capture(vec![]);
let first_input = f.local.git.save_approval_input(&f.owner, []).unwrap();
let other_lease = WorkspaceLeaseID([6; 16]);
f.local.register(WorkspaceID([7; 16]), other_lease).unwrap();
let other = f.local.observation_owner(other_lease).unwrap();
let second_input = f.local.git.save_approval_input(&other, []).unwrap();
let receipt = f
.local
.git
.mutate_sync(original.clone(), &AtomicBool::new(false));
assert!(matches!(receipt.state, OperationState::Confirmed(_)));
f.local.release(f.owner.registration().lease);
let mut service = LocalGitService::new(f.local.files.clone());
service.observations = Err(io_error("fixture storage unavailable"));
let WorktreePreparation::Existing(found) = service
.prepare_worktree_sync(
&f.owner,
&original,
[0; 16],
[0; 32],
first_input,
&AtomicBool::new(true),
)
.unwrap()
else {
panic!()
};
assert_eq!(*found, receipt);
assert_eq!(
service
.execute_worktree_sync(
&f.owner,
original,
[0; 16],
[0; 32],
second_input,
&AtomicBool::new(true)
)
.unwrap(),
receipt
);
}
#[test]
fn execute_accepts_later_approvals_but_keeps_the_original_request() {
let f = Fixture::new();
let (original, observation, digest) = f.capture(vec![f.file.clone()]);
let first = f.save(f.file.clone(), 47, b"first");
let input = f.local.git.save_approval_input(&f.owner, [first]).unwrap();
let WorktreePreparation::Prepared(plan) = f
.local
.git
.prepare_worktree_sync(
&f.owner,
&original,
observation,
digest,
input,
&AtomicBool::new(false),
)
.unwrap()
else {
panic!()
};
let second = f.save(f.file.clone(), 48, b"second");
let input = f.local.git.save_approval_input(&f.owner, [second]).unwrap();
let receipt = f
.local
.git
.execute_worktree_sync(
&f.owner,
original.clone(),
plan,
digest,
input,
&AtomicBool::new(false),
)
.unwrap();
assert!(matches!(receipt.state, OperationState::Confirmed(_)));
let ledger = f.local.files.ledger.lock().unwrap();
let record = ledger.read(original.operation).unwrap().unwrap();
assert_eq!(
record.envelope.request_digest,
request_digest(&original).unwrap()
);
}
#[test]
fn prepared_sibling_worktree_can_save_and_prepare_independently() {
let f = Fixture::new();
let root = f.dir.path().join("repo");
let linked = f.dir.path().join("linked");
git(
&root,
&["worktree", "add", "-b", "linked", linked.to_str().unwrap()],
);
let sibling_file = FileRef {
path: target_path(&linked.join("file")),
..f.file.clone()
};
let lease = WorkspaceLeaseID([6; 16]);
f.local.register(WorkspaceID([7; 16]), lease).unwrap();
let owner = f.local.observation_owner(lease).unwrap();
let sibling_worktree = f
.local
.git
.discover_sync(
FileRef {
path: target_path(&linked),
..f.file.clone()
},
&AtomicBool::new(false),
)
.unwrap()
.unwrap();
let (original, observation, digest) = f.capture(vec![f.file.clone(), sibling_file.clone()]);
let input = f.local.git.save_approval_input(&f.owner, []).unwrap();
let WorktreePreparation::Prepared(plan) = f
.local
.git
.prepare_worktree_sync(
&f.owner,
&original,
observation,
digest,
input,
&AtomicBool::new(false),
)
.unwrap()
else {
panic!()
};
let sibling = f
.local
.git
.observe_owned_sync(
&owner,
sibling_worktree.clone(),
[Ok(sibling_file.clone())],
&AtomicBool::new(false),
)
.unwrap();
let sibling_digest = f
.local
.git
.observation_alias_digest(&owner, sibling.id)
.unwrap();
let sibling_original = GitMutationRequest {
operation: OperationID([91; 16]),
worktree: sibling_worktree,
expected: sibling.snapshot.state,
command: GitMutation::Checkout {
reference: b"linked".to_vec(),
},
};
let input = f.local.git.save_approval_input(&owner, []).unwrap();
let WorktreePreparation::Prepared(sibling_plan) = f
.local
.git
.prepare_worktree_sync(
&owner,
&sibling_original,
sibling.id,
sibling_digest,
input,
&AtomicBool::new(false),
)
.unwrap()
else {
panic!()
};
f.save_as(lease, sibling_file, 49, b"sibling remains editable");
let input = f.local.git.save_approval_input(&f.owner, []).unwrap();
assert!(matches!(
f.local
.git
.execute_worktree_sync(
&f.owner,
original,
plan,
digest,
input,
&AtomicBool::new(false)
)
.unwrap()
.state,
OperationState::Confirmed(_)
));
f.local.git.cancel_worktree_plan(&owner, sibling_plan);
}
#[test]
fn ignored_alias_is_rechecked_after_durable_acceptance() {
let f = Fixture::new();
let root = f.dir.path().join("repo");
std::fs::write(root.join(".git/info/exclude"), "ignored\n").unwrap();
std::fs::write(root.join("ignored"), "original").unwrap();
let file = FileRef {
path: target_path(&root.join("ignored")),
..f.file.clone()
};
let (original, observation, digest) = f.capture(vec![file]);
let input = f.local.git.save_approval_input(&f.owner, []).unwrap();
let WorktreePreparation::Prepared(plan) = f
.local
.git
.prepare_worktree_sync(
&f.owner,
&original,
observation,
digest,
input,
&AtomicBool::new(false),
)
.unwrap()
else {
panic!()
};
let mut service = LocalGitService::new(f.local.files.clone());
service.after_authorization = Some(Arc::new(move || {
std::fs::write(root.join("ignored"), "external").unwrap()
}));
let input = service.save_approval_input(&f.owner, []).unwrap();
let receipt = service
.execute_worktree_sync(
&f.owner,
original.clone(),
plan,
digest,
input,
&AtomicBool::new(false),
)
.unwrap();
assert!(matches!(
receipt.state,
OperationState::Failed(OperationFailure {
effect: FailureEffect::ProvenNoEffect,
..
})
));
assert_eq!(
service.existing_receipt_sync(&original).unwrap(),
Some(receipt)
);
}
#[test]
fn concurrent_original_acceptance_wins_over_stale_cancelled_proof_after_preflight() {
for preparing in [false, true] {
let f = Fixture::new();
let (original, _, _) = f.capture(vec![]);
let input = f.local.git.save_approval_input(&f.owner, []).unwrap();
let entered = Arc::new(std::sync::Barrier::new(2));
let resume = Arc::new(std::sync::Barrier::new(2));
let mut service = LocalGitService::new(f.local.files.clone());
let (at_lookup, continue_lookup) = (entered.clone(), resume.clone());
service.after_proof_lookup = Some(Arc::new(move || {
at_lookup.wait();
continue_lookup.wait();
}));
let owner = f.owner.clone();
let duplicate = original.clone();
let worker = std::thread::spawn(move || {
if preparing {
let WorktreePreparation::Existing(receipt) = service
.prepare_worktree_sync(
&owner,
&duplicate,
[0; 16],
[0; 32],
input,
&AtomicBool::new(true),
)
.unwrap()
else {
panic!()
};
*receipt
} else {
service
.execute_worktree_sync(
&owner,
duplicate,
[0; 16],
[0; 32],
input,
&AtomicBool::new(true),
)
.unwrap()
}
});
entered.wait();
let expected = f
.local
.git
.mutate_sync(original.clone(), &AtomicBool::new(false));
resume.wait();
assert!(matches!(expected.state, OperationState::Confirmed(_)));
assert_eq!(worker.join().unwrap(), expected);
assert_eq!(
f.local.git.existing_receipt_sync(&original).unwrap(),
Some(expected)
);
}
}
#[test]
fn legacy_lookup_never_waits_for_repository_execution_and_existing_receipt_wins() {
let f = Fixture::new();
let (original, _, _) = f.capture(vec![]);
let mutex = super::super::mutations::repository_mutex(&original.worktree);
let held = mutex.lock().unwrap();
assert_eq!(
f.local
.git
.refuse_unprepared_worktree_sync(&original)
.unwrap_err()
.code,
ErrorCode::Unavailable
);
drop(held);
let expected = f
.local
.git
.mutate_sync(original.clone(), &AtomicBool::new(false));
assert!(matches!(expected.state, OperationState::Confirmed(_)));
let _held = mutex.lock().unwrap();
assert_eq!(
f.local
.git
.refuse_unprepared_worktree_sync(&original)
.unwrap(),
expected
);
}
#[test]
fn legacy_refusal_rechecks_original_accepted_after_fast_preflight() {
let f = Fixture::new();
let (original, _, _) = f.capture(vec![]);
let local = f.local.clone();
let concurrent = original.clone();
let mut service = LocalGitService::new(f.local.files.clone());
service.after_proof_lookup = Some(Arc::new(move || {
assert!(matches!(
local
.git
.mutate_sync(concurrent.clone(), &AtomicBool::new(false))
.state,
OperationState::Confirmed(_)
));
}));
let receipt = service.refuse_unprepared_worktree_sync(&original).unwrap();
assert_eq!(
Some(receipt),
f.local.git.existing_receipt_sync(&original).unwrap()
);
}
#[test]
fn legacy_existing_accepted_lookup_does_not_wait_for_the_executing_repository() {
let f = Fixture::new();
let (original, _, _) = f.capture(vec![]);
let local = f.local.clone();
let duplicate = original.clone();
let mut service = LocalGitService::new(f.local.files.clone());
service.after_authorization = Some(Arc::new(move || {
let receipt = local
.git
.refuse_unprepared_worktree_sync(&duplicate)
.unwrap();
assert!(matches!(receipt.state, OperationState::Unknown(_)));
}));
let receipt = service.mutate_sync(original.clone(), &AtomicBool::new(false));
assert!(matches!(receipt.state, OperationState::Confirmed(_)));
assert_eq!(
Some(receipt),
f.local.git.existing_receipt_sync(&original).unwrap()
);
}
#[test]
fn hintless_owned_save_uses_captured_same_path_without_rewriting_alias_identity() {
for changed_hint in [false, true] {
let f = Fixture::new();
let outside = f.dir.path().join("original-alias");
std::os::unix::fs::symlink(f.dir.path().join("repo/file"), &outside).unwrap();
let captured = FileRef {
path: target_path(&outside),
resolved_identity: Some(vec![7]),
..f.file.clone()
};
let (original, observation, digest) = f.capture(vec![captured.clone()]);
let effective = FileRef {
resolved_identity: if changed_hint { Some(vec![8]) } else { None },
..captured.clone()
};
let approval = f.save(effective, 48, b"owned save");
let input = f
.local
.git
.save_approval_input(&f.owner, [approval])
.unwrap();
let result = f.local.git.prepare_worktree_sync(
&f.owner,
&original,
observation,
digest,
input,
&AtomicBool::new(false),
);
if changed_hint {
assert!(result.is_err());
} else {
let WorktreePreparation::Prepared(plan) = result.unwrap() else {
panic!("expected prepared")
};
assert_eq!(
f.local
.git
.observation_alias_digest(&f.owner, observation)
.unwrap(),
digest
);
let mut reader = f
.local
.git
.observation_reader(&f.owner, observation)
.unwrap();
let mut found = false;
while let Some(record) = reader.next_record().unwrap() {
if record.membership.is_some() {
assert!(record.matches_alias(&captured).unwrap());
found = true;
}
}
assert!(found);
drop(reader);
f.local.git.cancel_worktree_plan(&f.owner, plan);
}
}
}
#[test]
fn original_alias_set_and_link_identity_cannot_be_reconstructed_later() {
for case in [
"uncovered",
"same-target-link-replaced",
"hint-changed",
"membership-changed",
] {
let f = Fixture::new();
let root = f.dir.path().join("repo");
let outside = f.dir.path().join("alias");
let destination = if case == "membership-changed" {
let path = f.dir.path().join("unrelated");
std::fs::write(&path, "unrelated").unwrap();
path
} else {
root.join("file")
};
std::os::unix::fs::symlink(&destination, &outside).unwrap();
let mut alias = FileRef {
path: target_path(&outside),
..f.file.clone()
};
let aliases = if case == "uncovered" {
vec![f.file.clone()]
} else {
vec![f.file.clone(), alias.clone()]
};
let (original, observation, digest) = f.capture(aliases);
let approvals = if case == "membership-changed" {
vec![]
} else {
if case == "hint-changed" {
alias.resolved_identity = Some(vec![7]);
}
vec![f.save(alias, 50, b"saved through alias")]
};
if matches!(case, "same-target-link-replaced" | "membership-changed") {
let replacement = f.dir.path().join("new-link");
std::os::unix::fs::symlink(root.join("file"), &replacement).unwrap();
std::fs::rename(replacement, outside).unwrap();
}
let input = f
.local
.git
.save_approval_input(&f.owner, approvals)
.unwrap();
assert!(
f.local
.git
.prepare_worktree_sync(
&f.owner,
&original,
observation,
digest,
input,
&AtomicBool::new(false)
)
.is_err(),
"{case}"
);
}
}
#[test]
fn owned_creation_through_original_directory_alias_is_verified() {
let f = Fixture::new();
let root = f.dir.path().join("repo");
let alias = f.dir.path().join("alias-dir");
std::os::unix::fs::symlink(&root, &alias).unwrap();
let file = FileRef {
path: target_path(&alias.join("new")),
..f.file.clone()
};
let (original, observation, digest) = f.capture(vec![file.clone()]);
let approval = f.save(file, 51, b"created through original parent link");
f.execute(original, observation, digest, vec![approval]);
}
#[test]
fn nested_repository_and_submodule_aliases_preserve_original_membership() {
let f = Fixture::new();
let root = f.dir.path().join("repo");
let nested = root.join("nested");
std::fs::create_dir(&nested).unwrap();
git(&nested, &["init", "-b", "nested"]);
std::fs::write(nested.join("file"), "nested").unwrap();
std::fs::write(root.join(".git/info/exclude"), "nested/\n").unwrap();
let source = f.dir.path().join("source");
std::fs::create_dir(&source).unwrap();
git(&source, &["init", "-b", "main"]);
git(&source, &["config", "user.name", "Fixture"]);
git(
&source,
&["config", "user.email", "fixture@example.invalid"],
);
std::fs::write(source.join("file"), "submodule").unwrap();
git(&source, &["add", "file"]);
git(&source, &["commit", "-m", "initial"]);
git(
&root,
&[
"-c",
"protocol.file.allow=always",
"submodule",
"add",
source.to_str().unwrap(),
"sub",
],
);
git(&root, &["commit", "-am", "submodule"]);
git(&root, &["branch", "-f", "other"]);
let files = [nested.join("file"), root.join("sub/file")].map(|path| FileRef {
path: target_path(&path),
..f.file.clone()
});
let (original, observation, digest) = f.capture(files.to_vec());
let mut reader = f
.local
.git
.observation_reader(&f.owner, observation)
.unwrap();
let mut unaffected = 0;
while let Some(record) = reader.next_record().unwrap() {
if let Some(membership) = record.membership {
assert!(!membership.affected);
assert!(membership.requested_repository.is_some());
unaffected += 1;
}
}
assert_eq!(unaffected, 2);
drop(reader);
f.execute(original, observation, digest, vec![]);
}
#[test]
fn truncated_approval_input_cannot_publish_a_plan_or_leak_its_pin() {
let f = Fixture::new();
let (original, observation, digest) = f.capture(vec![]);
let before = f.local.git.observation_storage_stats().unwrap().bytes;
let mut input = f.local.git.observation_input(&f.owner).unwrap();
input.append(&[0; 95]).unwrap();
assert!(
f.local
.git
.prepare_worktree_sync(
&f.owner,
&original,
observation,
digest,
input.seal().unwrap(),
&AtomicBool::new(false)
)
.is_err()
);
let stats = f.local.git.observation_storage_stats().unwrap();
assert_eq!(stats.bytes, before);
assert_eq!(stats.pins, 0);
assert_eq!(stats.captures, 0);
}
#[test]
fn plan_cancel_or_observation_release_must_survive_durable_admission() {
for retirement in ["cancel", "release", "expire"] {
for phase in ["before", "during", "after"] {
let f = Fixture::new();
let (original, observation, digest) = f.capture(vec![]);
let input = f.local.git.save_approval_input(&f.owner, []).unwrap();
let WorktreePreparation::Prepared(plan) = f
.local
.git
.prepare_worktree_sync(
&f.owner,
&original,
observation,
digest,
input,
&AtomicBool::new(false),
)
.unwrap()
else {
panic!()
};
let mut service = LocalGitService::new(f.local.files.clone());
let local = f.local.clone();
let owner = f.owner.clone();
let hook: Arc<dyn Fn() + Send + Sync> = Arc::new(move || match retirement {
"release" => local.git.release_observation(&owner, observation),
"expire" => super::super::observation_store::testing::expire_owner(
local.git.observation_store().unwrap(),
&owner,
),
_ => local.git.cancel_worktree_plan(&owner, plan),
});
match phase {
"before" => service.before_accept = Some(hook),
"during" => service.after_accept = Some(hook),
_ => service.after_authorization = Some(hook),
}
let input = service.save_approval_input(&f.owner, []).unwrap();
let result = service.execute_worktree_sync(
&f.owner,
original.clone(),
plan,
digest,
input,
&AtomicBool::new(false),
);
if phase == "before" {
assert!(
result.is_err(),
"cancelled/released plan accepted before fsync"
);
assert!(service.existing_receipt_sync(&original).unwrap().is_none());
} else {
let receipt = result.unwrap();
assert!(
if phase == "after" {
matches!(receipt.state, OperationState::Confirmed(_))
} else {
matches!(
receipt.state,
OperationState::Failed(OperationFailure {
effect: FailureEffect::ProvenNoEffect,
..
})
)
},
"{phase}, {retirement}: {receipt:?}"
);
}
}
}
}
#[test]
fn unrecordable_revocation_remains_unknown_and_the_accepted_id_never_replays() {
let f = Fixture::new();
let (original, observation, digest) = f.capture(vec![]);
let input = f.local.git.save_approval_input(&f.owner, []).unwrap();
let WorktreePreparation::Prepared(plan) = f
.local
.git
.prepare_worktree_sync(
&f.owner,
&original,
observation,
digest,
input,
&AtomicBool::new(false),
)
.unwrap()
else {
panic!()
};
let mut service = LocalGitService::new(f.local.files.clone());
let directory = f.local.files.ledger.lock().unwrap().directory.clone();
let relocated = directory.with_file_name("temporarily-unavailable-operations");
let (original_directory, unavailable_directory) = (directory.clone(), relocated.clone());
let local = f.local.clone();
let lease = f.owner.registration().lease;
service.after_accept = Some(Arc::new(move || {
local.release(lease);
std::fs::rename(&original_directory, &unavailable_directory).unwrap();
}));
let input = service.save_approval_input(&f.owner, []).unwrap();
let receipt = service
.execute_worktree_sync(
&f.owner,
original.clone(),
plan,
digest,
input,
&AtomicBool::new(false),
)
.unwrap();
assert!(matches!(receipt.state, OperationState::Unknown(_)));
std::fs::rename(relocated, directory).unwrap();
assert!(matches!(
service
.mutate_sync(original.clone(), &AtomicBool::new(false))
.state,
OperationState::Unknown(_)
));
assert_eq!(
service
.status_sync(original.worktree, &AtomicBool::new(false))
.unwrap()
.state,
original.expected
);
}
#[test]
fn immutable_original_request_bytes_are_retained_and_charged_with_the_plan() {
let f = Fixture::new();
let (original, observation, digest) = f.capture(vec![]);
let baseline = f.local.git.observation_storage_stats().unwrap().bytes;
let input = f.local.git.save_approval_input(&f.owner, []).unwrap();
let WorktreePreparation::Prepared(plan) = f
.local
.git
.prepare_worktree_sync(
&f.owner,
&original,
observation,
digest,
input,
&AtomicBool::new(false),
)
.unwrap()
else {
panic!()
};
assert_eq!(
f.local.git.observation_storage_stats().unwrap().bytes,
baseline + serde_json::to_vec(&original).unwrap().len() as u64,
"full immutable request is missing from charged plan storage"
);
let store = f.local.git.observation_store().unwrap();
let retained = store.plan(&f.owner, plan).unwrap();
assert_eq!(retained.request_digest, request_digest(&original).unwrap());
let mut bytes = Vec::new();
cursor(&retained._original_request)
.read_to_end(&mut bytes)
.unwrap();
assert_eq!(bytes, serde_json::to_vec(&original).unwrap());
drop(retained);
f.local.git.cancel_worktree_plan(&f.owner, plan);
assert_eq!(
f.local.git.observation_storage_stats().unwrap().bytes,
baseline
);
}
#[test]
#[ignore = "explicit modest verifier geometry measurement"]
fn actual_verifier_chain_and_alias_measurement() {
fn rss() -> i64 {
let mut usage = std::mem::MaybeUninit::uninit();
assert_eq!(
unsafe { libc::getrusage(libc::RUSAGE_SELF, usage.as_mut_ptr()) },
0
);
unsafe { usage.assume_init() }.ru_maxrss
}
let f = Fixture::new();
let root = f.dir.path().join("repo");
let mut files = vec![];
for n in 0..64 {
let path = root.join(format!("path-{n:026}"));
std::fs::write(&path, b"original\n").unwrap();
files.push(FileRef {
path: target_path(&path),
..f.file.clone()
});
}
git(&root, &["add", "."]);
git(&root, &["commit", "-m", "geometry"]);
git(&root, &["branch", "-f", "other"]);
let mut destination = root.join("file");
for n in 0..8 {
let link = f.dir.path().join(format!("indirect-{n}"));
std::os::unix::fs::symlink(destination, &link).unwrap();
destination = link;
}
let mut aliases = vec![f.file.clone()];
aliases.extend(files.iter().take(16).cloned());
for n in 0..16 {
let path = f.dir.path().join(format!("outside-{n}"));
std::os::unix::fs::symlink(&destination, &path).unwrap();
aliases.push(FileRef {
path: target_path(&path),
..f.file.clone()
});
}
let rss_before = rss();
let capture_start = Instant::now();
let (original, observation, digest) = f.capture(aliases);
let capture_seconds = capture_start.elapsed().as_secs_f64();
let original_bytes = f.local.git.observation_storage_stats().unwrap().bytes;
let mut approvals = vec![f.save(f.file.clone(), 100, b"alias endpoint saved")];
for round in 0..2 {
for (n, file) in files.iter().take(16).enumerate() {
approvals.push(f.save(
file.clone(),
(101 + round * 16 + n) as u8,
if round == 0 { b"first" } else { b"final" },
));
}
}
let input = f
.local
.git
.save_approval_input(&f.owner, approvals)
.unwrap();
let approval_bytes = input.len();
let prepare_start = Instant::now();
let WorktreePreparation::Prepared(plan_id) = f
.local
.git
.prepare_worktree_sync(
&f.owner,
&original,
observation,
digest,
input,
&AtomicBool::new(false),
)
.unwrap()
else {
panic!()
};
let prepare_seconds = prepare_start.elapsed().as_secs_f64();
let retained_plan_bytes = f.local.git.observation_storage_stats().unwrap().bytes;
let store = f.local.git.observation_store().unwrap();
let plan = store.plan(&f.owner, plan_id).unwrap();
let evidence = f
.local
.git
.freeze_evidence(&f.owner, &plan, None, &AtomicBool::new(false))
.unwrap();
let sampled_bytes = store.stats().bytes;
let mut reader = cursor(&evidence);
let mut max_record_bytes = 0;
let mut records = 0;
while let Some(record) = read_evidence(&mut reader).unwrap() {
max_record_bytes = max_record_bytes.max(serde_json::to_vec(&record).unwrap().len());
records += 1;
}
drop(reader);
drop(evidence);
drop(plan);
let additional = f.local.git.save_approval_input(&f.owner, []).unwrap();
let execute_start = Instant::now();
let receipt = f
.local
.git
.execute_worktree_sync(
&f.owner,
original,
plan_id,
digest,
additional,
&AtomicBool::new(false),
)
.unwrap();
let execute_seconds = execute_start.elapsed().as_secs_f64();
assert!(matches!(receipt.state, OperationState::Confirmed(_)));
println!(
"{}",
serde_json::json!({"git_nodes":65,"supplemental_aliases":33,"outside_aliases":16,"indirect_links":8,"save_records":records,
"original_bytes":original_bytes,"approval_bytes":approval_bytes,"retained_plan_bytes":retained_plan_bytes,"sampled_with_frozen_evidence_bytes":sampled_bytes,
"max_encoded_evidence_record_bytes":max_record_bytes,"capture_seconds":capture_seconds,"prepare_seconds":prepare_seconds,"execute_seconds":execute_seconds,
"rss_before":rss_before,"rss_after":rss(),"hash_scratch_bytes_per_worker":65536,"ledger_record_parse_bound":131072})
);
f.local.git.release_observation(&f.owner, observation);
assert_eq!(store.stats().bytes, 0);
}
}