use std::collections::{BTreeMap, BTreeSet};
use std::fs;
use std::io::{Read, Write};
use std::path::{Path, PathBuf};
use std::sync::OnceLock;
use crate::legacy_dvc::{TreeEntry, directory_digest, tree_bytes, tree_manifest_bytes};
use crate::storage_format::{Checksum, Entry, Kind, Manifest, Verification, Version};
use md5::{Digest, Md5};
use serde_json::{Value, json};
use walkdir::WalkDir;
use crate::error::{Error, IoContext, Result};
use crate::git::GitRepo;
use crate::path::{reject_symlink_traversal, repo_path, resolved_under, to_slash};
use crate::process::CommandOutput;
use crate::storage_metadata;
#[derive(Debug, Clone)]
pub(crate) struct StorageEntry {
pub pointer: String,
pub object: String,
pub md5: Option<String>,
pub size: Option<u64>,
pub version_id: Option<String>,
pub etag: Option<String>,
pub verification: Option<Verification>,
pub hash_name: String,
}
#[derive(Debug, Clone)]
pub(crate) struct CasSource {
pub key: String,
pub version_id: Option<String>,
pub etag: String,
pub size: u64,
pub checksum: Checksum,
}
pub(crate) fn verification_for(
client: &crate::native_s3::S3Client,
entry: &StorageEntry,
version_id: &str,
sha256: &str,
method: &str,
) -> Result<Verification> {
let proof = Verification {
endpoint: client.endpoint_identity(),
bucket: client.bucket.clone(),
key: client.key_for(&entry.object),
version_id: version_id.into(),
checksum: Checksum {
algorithm: "sha256".into(),
digest: sha256.into(),
},
size: entry
.size
.ok_or_else(|| Error::message("verified storage entry has no size"))?,
method: method.into(),
};
proof.checksum.validate()?;
Ok(proof)
}
pub(crate) fn validate_verification_scope(
client: &crate::native_s3::S3Client,
entry: &StorageEntry,
) -> Result<()> {
if let Some(proof) = &entry.verification
&& (proof.endpoint != client.endpoint_identity()
|| proof.bucket != client.bucket
|| proof.key != client.key_for(&entry.object)
|| entry.version_id.as_deref() != Some(proof.version_id.as_str())
|| entry.size != Some(proof.size))
{
return Err(Error::message(format!(
"storage verification proof does not match its physical location: {}",
entry.object
)));
}
Ok(())
}
fn attest_legacy_entry(
repo: &GitRepo,
client: &crate::native_s3::S3Client,
entry: &StorageEntry,
) -> Result<Verification> {
use base64::Engine;
let version = entry
.version_id
.as_deref()
.ok_or_else(|| Error::message("legacy entry has no version"))?;
let request =
json!({"Bucket":client.bucket,"Key":client.key_for(&entry.object),"VersionId":version});
let info = client.head_with_checksums(&request)?.value;
let matching = |value: &Value| {
value["VersionId"].as_str() == Some(version)
&& value["DeleteMarker"] != true
&& value["ContentLength"].as_u64() == entry.size
&& entry.etag.as_deref().is_none_or(|wanted| {
value["ETag"]
.as_str()
.is_some_and(|got| got.trim_matches('"') == wanted.trim_matches('"'))
})
};
if !matching(&info) {
return Err(Error::message(
"legacy exact version metadata differs from control",
));
}
let mut local_hashes = HashInventory::new();
let local = repo.root.join(&entry.object);
reject_symlink_traversal(&repo.root, &entry.object, "legacy verification payload")?;
let local_hash = if local.is_file() {
let hashes = local_hashes.hashes(&local)?;
if fs::metadata(&local).at(&local)?.len() != entry.size.unwrap_or(u64::MAX)
|| hashes.digest(&entry.hash_name)? != entry.md5.as_deref().unwrap_or("")
{
return Err(Error::message(
"local legacy payload differs from its control checksum or size",
));
}
Some(hashes.sha256)
} else {
None
};
let provider = if info["ChecksumType"] == "FULL_OBJECT" {
info["ChecksumSHA256"]
.as_str()
.map(|raw| {
let bytes = base64::engine::general_purpose::STANDARD
.decode(raw)
.map_err(|_| Error::message("invalid provider SHA256 checksum"))?;
if bytes.len() != 32 {
return Err(Error::message("invalid provider SHA256 checksum"));
}
Ok(crate::hex::encode_lower(bytes))
})
.transpose()?
} else {
None
};
if let (Some(local), Some(provider)) = (&local_hash, &provider) {
if local != provider {
return Err(Error::message(
"legacy remote raw SHA256 differs from local payload",
));
}
if local_hashes.hashes(&repo.root.join(&entry.object))?.sha256 != *local {
return Err(Error::message(
"local legacy payload changed during verification",
));
}
return verification_for(client, entry, version, provider, "provider-checksum");
}
let mut request = request;
request["IfMatch"] = info["ETag"].clone();
let mut attempt = 0;
let (response, hashes) = loop {
let mut hasher = FileHashStream::new();
match client.get_stream(&request, |bytes| {
hasher.update(bytes);
Ok(())
}) {
Ok(response) => break (response, hasher.finish()),
Err(error) if error.is_retryable() && attempt < 2 => {
attempt += 1;
}
Err(error) => return Err(error.into()),
}
};
if !matching(&response.value)
|| response.value["ETag"] != info["ETag"]
|| hashes.digest(&entry.hash_name)? != entry.md5.as_deref().unwrap_or("")
|| local_hash
.as_ref()
.is_some_and(|local| local != &hashes.sha256)
|| provider
.as_ref()
.is_some_and(|provider| provider != &hashes.sha256)
{
return Err(Error::message(
"legacy exact-version content verification failed",
));
}
if let Some(before) = local_hash
&& local_hashes.hashes(&local)?.sha256 != before
{
return Err(Error::message(
"local legacy payload changed during verification",
));
}
verification_for(client, entry, version, &hashes.sha256, "verified-read")
}
pub(crate) fn upgrade_manifest_verification(
repo: &GitRepo,
client: &crate::native_s3::S3Client,
pointer: &str,
manifest: &mut Manifest,
) -> Result<()> {
if manifest.schema_version == crate::storage_format::SCHEMA_VERSION {
return Ok(());
}
let document = storage_metadata::logical_document(manifest);
let entries = entries_from_document(pointer, &document, &manifest.checksum.algorithm)?;
let bound = entries
.into_iter()
.filter(|entry| entry.version_id.is_some())
.collect::<Vec<_>>();
let proofs = crate::native_versions::bounded_map(&bound, |entry| {
Ok((
entry.object.clone(),
attest_legacy_entry(repo, client, entry)?,
))
})?
.into_iter()
.collect::<BTreeMap<_, _>>();
let object = output_object(pointer, &manifest.path)?;
if let Some(entries) = &mut manifest.entries {
for entry in entries {
if let Some(version) = &mut entry.version {
version.verification = proofs.get(&format!("{object}/{}", entry.path)).cloned();
}
}
} else if let Some(version) = &mut manifest.version {
version.verification = proofs.get(&object).cloned();
}
manifest.schema_version = crate::storage_format::SCHEMA_VERSION;
manifest.validate(pointer)
}
pub(crate) fn fetch_cas_to_cache(
repo: &GitRepo,
client: &crate::native_s3::S3Client,
source: &CasSource,
) -> Result<PathBuf> {
source.checksum.validate()?;
if source
.version_id
.as_deref()
.is_some_and(|version| version.trim().is_empty() || version == "null")
{
return Err(Error::message(
"legacy CAS source has an invalid pinned exact version ID",
));
}
if source.etag.trim_matches('"').is_empty() {
return Err(Error::message("legacy CAS source has no pinned ETag"));
}
let mut request = json!({"Bucket":client.bucket,"Key":source.key,"IfMatch":source.etag});
if let Some(version) = &source.version_id {
request["VersionId"] = version.clone().into();
}
let matches = |value: &Value| {
value["DeleteMarker"] != true
&& value["ContentLength"].as_u64() == Some(source.size)
&& value["ETag"]
.as_str()
.is_some_and(|etag| etag.trim_matches('"') == source.etag.trim_matches('"'))
&& source
.version_id
.as_deref()
.is_none_or(|version| value["VersionId"].as_str() == Some(version))
};
if !matches(&client.call_s3("head_object", &request, None)?.value) {
return Err(Error::message(
"legacy CAS source changed after its import inventory was recorded",
));
}
let cache =
cache_path_with_algorithm(repo, &source.checksum.digest, &source.checksum.algorithm)?;
let root = cache_root(repo)?;
fs::create_dir_all(&root).at(&root)?;
let temporary = tempfile::NamedTempFile::new_in(&root).at(&root)?;
let fetched = client.get_to_file(&request, temporary.path())?;
if !matches(&fetched.value)
|| fs::metadata(temporary.path()).at(temporary.path())?.len() != source.size
|| file_digest(temporary.path(), &source.checksum.algorithm)? != source.checksum.digest
{
return Err(Error::message(
"legacy CAS GET differs from its pinned ETag, exact version, physical size or checksum",
));
}
fs::create_dir_all(cache.parent().expect("cache parent")).at(&cache)?;
temporary.persist(&cache).map_err(|error| Error::Io {
path: cache.clone(),
source: error.error,
})?;
Ok(cache)
}
pub(crate) fn upload_verified(
repo: &GitRepo,
client: &crate::native_s3::S3Client,
entry: &StorageEntry,
source: &Path,
) -> Result<Version> {
let mut hashes = HashInventory::new();
let raw_sha256 = hashes.hashes(source)?.sha256;
let cache = CachePaths::new(repo)?;
let condition = import_destination_condition(repo, client, entry)?;
let (id, etag) = upload_version_in(
client,
repo,
entry,
source,
5 * (1 << 30),
64 * (1 << 20),
UploadPolicy {
namespace: "storage-import-uploads",
condition: condition.as_deref(),
raw_sha256: Some(&raw_sha256),
local: Some(cache.local()),
hashes: Some(&mut hashes),
},
)?;
let mut bound = entry.clone();
bound.version_id = Some(id.clone());
let proof = verification_for(client, entry, &id, &raw_sha256, "verified-upload")?;
bound.verification = Some(proof.clone());
cache.install_entry_with_inventory(&bound, source, &mut hashes)?;
if hashes.hashes(&cache.entry(&bound)?)?.sha256 != raw_sha256 {
return Err(Error::message(
"storage import source changed while installing its exact-version cache",
));
}
Ok(Version {
id,
etag: Some(etag),
verification: Some(proof),
})
}
pub(crate) fn is_historical_cas_read(repo: &GitRepo, pointers: &[String]) -> Result<bool> {
Ok(!pointers.is_empty() && historical_cas_pointers(repo, pointers)?.len() == pointers.len())
}
pub(crate) fn historical_cas_pointers(repo: &GitRepo, pointers: &[String]) -> Result<Vec<String>> {
if !pointers.iter().any(|pointer| pointer.ends_with(".dvc")) {
return Ok(Vec::new());
}
if !crate::legacy_dvc::is_content_addressed_checkout(repo)? {
return Ok(Vec::new());
}
let mut selected = Vec::new();
for pointer in pointers {
if !pointer.ends_with(".dvc") {
continue;
}
reject_symlink_traversal(&repo.root, pointer, "legacy CAS metadata")?;
let raw = fs::read_to_string(repo.root.join(pointer)).at(repo.root.join(pointer))?;
let yaml: serde_yaml::Value =
serde_yaml::from_str(&raw).map_err(|error| Error::message(error.to_string()))?;
let has_bindings = yaml["outs"].as_sequence().into_iter().flatten().any(|out| {
out.get("cloud").is_some()
|| out
.get("files")
.and_then(serde_yaml::Value::as_sequence)
.is_some_and(|files| files.iter().any(|file| file.get("cloud").is_some()))
});
if !has_bindings {
selected.push(pointer.clone());
}
}
Ok(selected)
}
pub(crate) fn resolve_cas_source(
client: &crate::native_s3::S3Client,
digest: &str,
algorithm: &str,
) -> Result<CasSource> {
let mut selected = None;
for object in crate::legacy_dvc::cas_key_candidates(digest, algorithm)? {
let key = client.key_for(&object);
let info = match client.call_s3(
"head_object",
&json!({"Bucket":client.bucket,"Key":key}),
None,
) {
Ok(response) => response.value,
Err(error) if matches!(error.code.as_str(), "NoSuchKey" | "NotFound" | "404") => {
continue;
}
Err(error) => return Err(error.into()),
};
let size = info["ContentLength"]
.as_u64()
.ok_or_else(|| Error::message("legacy CAS HEAD has no physical size"))?;
let etag = info["ETag"]
.as_str()
.filter(|etag| !etag.trim_matches('"').is_empty())
.ok_or_else(|| Error::message("legacy CAS HEAD has no ETag"))?
.to_owned();
let source = CasSource {
key,
version_id: info["VersionId"]
.as_str()
.filter(|version| !version.is_empty() && *version != "null")
.map(str::to_owned),
etag,
size,
checksum: Checksum {
algorithm: algorithm.into(),
digest: digest.trim_end_matches(".dir").into(),
},
};
if selected.as_ref().is_some_and(|previous: &CasSource| {
previous.size != source.size
|| previous.etag.trim_matches('"') != source.etag.trim_matches('"')
}) {
return Err(Error::message(
"legacy CAS object layouts contain conflicting source identities",
));
}
if selected.is_none() {
selected = Some(source);
}
}
selected
.ok_or_else(|| Error::message("legacy CAS object is absent from every supported layout"))
}
pub(crate) fn fetch_historical_cas(repo: &GitRepo, pointers: &[String]) -> Result<Option<Value>> {
if !is_historical_cas_read(repo, pointers)? {
return Ok(None);
}
let client = crate::native_s3::S3Client::historical_cas_from_repo(repo)?.ok_or_else(|| {
Error::message("historical metadata has no content-addressed source remote")
})?;
let mut checked = BTreeSet::new();
for pointer in pointers {
let raw = fs::read_to_string(repo.root.join(pointer)).at(repo.root.join(pointer))?;
let raw = storage_metadata::normalize_pointer_in_repo(repo, None, &raw, pointer)?;
let algorithm = storage_metadata::hash_algorithm(&raw, pointer)?;
let document = storage_metadata::parse_pointer_document(&raw, pointer)?;
let [out] = document.outs.as_slice() else {
return Err(Error::message(
"historical CAS pointer must define exactly one output",
));
};
let digest = out
.md5
.as_deref()
.ok_or_else(|| Error::message("historical CAS pointer has no checksum"))?;
let files = if digest.ends_with(".dir") {
let source = resolve_cas_source(&client, digest, &algorithm)?;
let mut directory_source = source.clone();
directory_source.checksum.algorithm = "md5".into();
let temporary_cache = fetch_cas_to_cache(repo, &client, &directory_source)?;
let bytes = fs::read(&temporary_cache).at(&temporary_cache)?;
let files = crate::legacy_dvc::parse_directory_manifest(&bytes, digest)?;
if out
.files
.as_ref()
.is_some_and(|inline| directory_digest(inline).ok().as_deref() != Some(digest))
{
return Err(Error::message(
"historical CAS inline directory inventory differs from its aggregate",
));
}
atomic_write(
&cache_path_with_algorithm(repo, digest, &algorithm)?,
&bytes,
)?;
files
.into_iter()
.map(|file| (file.relpath, file.md5.unwrap(), file.size))
.collect::<Vec<_>>()
} else {
vec![(String::new(), digest.to_owned(), out.size)]
};
let mut total = 0u64;
for (relative, digest, declared_size) in files {
let source = resolve_cas_source(&client, &digest, &algorithm)?;
if declared_size.is_some_and(|size| size != source.size) {
return Err(Error::message(
"historical CAS physical size differs from its metadata",
));
}
total = total
.checked_add(source.size)
.ok_or_else(|| Error::message("historical CAS size overflows"))?;
fetch_cas_to_cache(repo, &client, &source)?;
checked.insert(if relative.is_empty() {
output_object(pointer, &out.path)?
} else {
format!("{}/{}", output_object(pointer, &out.path)?, relative)
});
}
if out.size.is_some_and(|size| size != total) {
return Err(Error::message(
"historical CAS output physical size differs from its metadata",
));
}
}
Ok(Some(
json!({"mode":"legacy-content-addressed","checked_objects":checked}),
))
}
pub(crate) fn verify_import_destination(
repo: &GitRepo,
client: &crate::native_s3::S3Client,
entry: &StorageEntry,
) -> Result<bool> {
import_destination_condition(repo, client, entry).map(|condition| condition.is_some())
}
fn import_destination_condition(
repo: &GitRepo,
client: &crate::native_s3::S3Client,
entry: &StorageEntry,
) -> Result<Option<String>> {
let local = crate::local_state::directory_unmigrated(repo)?;
reject_symlink_traversal(
&local,
"storage-import-uploads",
"private storage import uploads",
)?;
let directory = local.join("storage-import-uploads");
let key = client.key_for(&entry.object);
let mut receipts = BTreeMap::new();
if directory.is_dir() {
for item in fs::read_dir(&directory).at(&directory)? {
let path = item.at(&directory)?.path();
if path.extension().is_none_or(|extension| extension != "json") {
continue;
}
if path.is_symlink() || !path.is_file() {
return Err(Error::message(
"private import upload receipt is not a regular file",
));
}
let journal: Value =
serde_json::from_slice(&fs::read(&path).at(&path)?).map_err(|error| {
Error::message(format!("invalid private import upload receipt: {error}"))
})?;
let context = &journal["context"];
if context["bucket"] != client.bucket || context["key"] != key {
continue;
}
let token = journal["token"]
.as_str()
.filter(|token| {
token.len() == 64 && token.bytes().all(|byte| byte.is_ascii_hexdigit())
})
.ok_or_else(|| {
Error::message("private import upload receipt has no valid ownership token")
})?;
let checksum = Checksum {
algorithm: context["hash_name"]
.as_str()
.ok_or_else(|| {
Error::message("private import upload receipt has no checksum algorithm")
})?
.into(),
digest: context["md5"]
.as_str()
.ok_or_else(|| Error::message("private import upload receipt has no checksum"))?
.into(),
};
checksum.validate()?;
let owned_entry = StorageEntry {
verification: None,
pointer: entry.pointer.clone(),
object: entry.object.clone(),
md5: Some(checksum.digest),
size: Some(context["size"].as_u64().ok_or_else(|| {
Error::message("private import upload receipt has no physical size")
})?),
version_id: None,
etag: None,
hash_name: checksum.algorithm,
};
let raw_sha256 = context["raw_sha256"]
.as_str()
.ok_or_else(|| Error::message("private import upload receipt has no raw SHA256"))?;
let (expected_path, expected_context) = upload_context(
repo,
client,
&owned_entry,
"storage-import-uploads",
Some(raw_sha256),
)?;
if path != expected_path
|| context != &expected_context
|| receipts
.insert(
token.to_owned(),
(
owned_entry,
raw_sha256.to_owned(),
if journal["verified_transfer"] == true
&& journal["transfer_endpoint"] == client.endpoint_identity()
{
journal["version_id"].as_str().map(str::to_owned)
} else {
None
},
),
)
.is_some()
{
return Err(Error::message(
"private import upload receipt identity does not match its destination",
));
}
}
}
let rows = client
.list_versions(&key)?
.into_iter()
.filter(|row| row["Key"] == key)
.collect::<Vec<_>>();
if rows.is_empty() {
return Ok(None);
}
let mut used = BTreeSet::new();
let mut latest = None;
for row in rows {
if row["delete_marker"] == true {
return Err(Error::message(
"storage import destination has an unowned delete marker",
));
}
let version = row["VersionId"]
.as_str()
.ok_or_else(|| Error::message("storage import destination has no exact version ID"))?;
let info = client
.call_s3(
"head_object",
&json!({"Bucket":client.bucket,"Key":key,"VersionId":version}),
None,
)?
.value;
let token = info["Metadata"][UPLOAD_TOKEN].as_str().ok_or_else(|| {
Error::message("storage import destination already has unowned object history")
})?;
let (owned_entry, raw_sha256, verified_version) = receipts.get(token).ok_or_else(|| {
Error::message("storage import destination has no matching private ownership receipt")
})?;
if !used.insert(token.to_owned()) {
return Err(Error::message(
"storage import ownership token matches conflicting destination versions",
));
}
verify_uploaded_version(
client,
repo,
owned_entry,
token,
version,
Some(raw_sha256),
verified_version.as_deref() == Some(version),
)?;
if row["IsLatest"] == true {
if latest.is_some() {
return Err(Error::message(
"storage import destination has conflicting latest versions",
));
}
latest = info["ETag"].as_str().map(str::to_owned);
}
}
latest.map(Some).ok_or_else(|| {
Error::message("storage import destination has no verified latest ownership binding")
})
}
#[derive(Debug, Clone)]
struct FileState {
relpath: String,
md5: String,
size: u64,
version_id: Option<String>,
verification: Option<Verification>,
}
#[derive(Debug, Clone)]
pub(crate) enum Operation {
#[cfg(test)]
Initialize,
Track {
paths: Vec<String>,
},
Record {
pointers: Vec<String>,
},
Upload {
pointers: Vec<String>,
},
Fetch {
pointers: Vec<String>,
},
Materialize {
pointers: Vec<String>,
},
Move {
source: String,
destination: String,
},
Untrack {
pointers: Vec<String>,
},
Status {
pointers: Vec<String>,
cloud: bool,
quiet: bool,
},
Changes {
outputs: Vec<String>,
},
}
pub(crate) fn execute(cwd: &Path, operation: &Operation) -> Result<CommandOutput> {
#[cfg(feature = "test-storage")]
if let Some(hook) = std::env::var_os("WORKSPACE_MGR_TEST_STORAGE_HOOK") {
let hook = hook
.to_str()
.ok_or_else(|| Error::message("storage test hook is not UTF-8"))?;
crate::process::run(hook, operation.test_arguments(), cwd)?;
}
let repo = GitRepo {
root: cwd.canonicalize().at(cwd)?,
};
match execute_inner(&repo, operation) {
Ok((code, stdout)) => Ok(CommandOutput {
code,
stdout,
stderr: String::new(),
}),
Err(error) => Ok(CommandOutput {
code: 1,
stdout: String::new(),
stderr: error.to_string(),
}),
}
}
#[cfg(feature = "test-storage")]
impl Operation {
fn test_arguments(&self) -> Vec<String> {
let (name, paths) = match self {
#[cfg(test)]
Self::Initialize => ("init", Vec::new()),
Self::Track { paths } => ("add", paths.clone()),
Self::Record { pointers } => ("commit", pointers.clone()),
Self::Upload { pointers } => ("push", pointers.clone()),
Self::Fetch { pointers } => ("fetch", pointers.clone()),
Self::Materialize { pointers } => ("checkout", pointers.clone()),
Self::Move {
source,
destination,
} => ("move", vec![source.clone(), destination.clone()]),
Self::Untrack { pointers } => ("remove", pointers.clone()),
Self::Status { pointers, .. } => ("status", pointers.clone()),
Self::Changes { outputs } => ("data", outputs.clone()),
};
std::iter::once(name.to_owned())
.chain(std::iter::once("--".to_owned()))
.chain(paths)
.collect()
}
}
fn execute_inner(repo: &GitRepo, operation: &Operation) -> Result<(i32, String)> {
match operation {
#[cfg(test)]
Operation::Initialize => initialize(repo)?,
Operation::Track { paths } => {
for path in paths {
add(repo, path)?;
}
}
Operation::Record { pointers } => {
let cache = CachePaths::new(repo)?;
let pointers = select_pointers(repo, pointers)?;
let cpus = std::thread::available_parallelism()
.map(usize::from)
.unwrap_or(1);
let workers = cpus.min(pointers.len()).max(1);
crate::native_versions::bounded_map_with_workers(
&pointers,
workers,
false,
|pointer| commit_with_cache(repo, &cache, pointer, cpus / workers),
)?;
}
Operation::Upload { pointers } => push(repo, &select_pointers(repo, pointers)?)?,
Operation::Fetch { pointers } => fetch(repo, &select_pointers(repo, pointers)?)?,
Operation::Materialize { pointers } => checkout(repo, &select_pointers(repo, pointers)?)?,
Operation::Move {
source,
destination,
} => move_output(repo, source, destination)?,
Operation::Untrack { pointers } => {
for pointer in select_pointers(repo, pointers)? {
remove(repo, &pointer)?;
}
}
Operation::Status {
pointers,
cloud,
quiet,
} => return status(repo, &select_pointers(repo, pointers)?, *cloud, *quiet),
Operation::Changes { outputs } => {
return Ok((
0,
serde_json::to_string(&data_status(repo, outputs)?)
.map_err(|error| Error::message(error.to_string()))?,
));
}
}
Ok((0, String::new()))
}
#[cfg(test)]
fn initialize(repo: &GitRepo) -> Result<()> {
let local = crate::local_state::directory_unmigrated(repo)?;
for path in [local.join("cache"), local.join("uploads")] {
fs::create_dir_all(&path).at(&path)?;
}
Ok(())
}
pub(crate) fn cache_root(repo: &GitRepo) -> Result<PathBuf> {
let local = crate::local_state::directory_unmigrated(repo)?;
reject_symlink_traversal(&local, "cache", "native storage cache")?;
let path = local.join("cache");
if path.exists() && !path.is_dir() {
return Err(Error::message("storage cache is not a directory"));
}
Ok(path)
}
pub(crate) struct CachePaths<'a> {
repo: &'a GitRepo,
root: PathBuf,
route: OnceLock<std::result::Result<CacheRouteSnapshot, String>>,
}
type CacheRoute = Option<(String, Option<String>)>;
struct CacheRouteSnapshot {
value: CacheRoute,
files: Vec<(PathBuf, Option<FileIdentity>)>,
}
fn optional_file_identity(path: &Path) -> Result<Option<FileIdentity>> {
match fs::symlink_metadata(path) {
Ok(metadata) => Ok(Some(FileIdentity::of(&metadata))),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None),
Err(source) => Err(Error::Io {
path: path.into(),
source,
}),
}
}
impl CacheRouteSnapshot {
fn capture(repo: &GitRepo) -> Result<Self> {
let files = [
repo.root.join(".workspace-mgr.toml"),
repo.root.join(".dvc/config"),
]
.into_iter()
.map(|path| optional_file_identity(&path).map(|identity| (path, identity)))
.collect::<Result<Vec<_>>>()?;
let snapshot = Self {
value: crate::native_s3::S3Client::cache_route(repo)?,
files,
};
snapshot.validate()?;
Ok(snapshot)
}
fn validate(&self) -> Result<()> {
for (path, identity) in &self.files {
if optional_file_identity(path)? != *identity {
return Err(Error::message(
"storage routing changed during cache verification",
));
}
}
Ok(())
}
}
impl<'a> CachePaths<'a> {
pub(crate) fn new(repo: &'a GitRepo) -> Result<Self> {
Ok(Self {
repo,
root: cache_root(repo)?,
route: OnceLock::new(),
})
}
pub(crate) fn root(&self) -> &Path {
&self.root
}
fn local(&self) -> &Path {
self.root.parent().expect("cache local state parent")
}
pub(crate) fn path(&self, digest: &str, hash_name: &str) -> Result<PathBuf> {
let (prefix, rest) = digest_parts(digest)?;
let relative = match hash_name {
"md5" => format!("objects/md5/{prefix}/{rest}"),
"md5-dos2unix" => format!("objects/md5-dos2unix/{prefix}/{rest}"),
_ => return Err(Error::message("unsupported storage hash algorithm")),
};
self.checked_path(&relative, "storage cache object")
}
fn checked_path(&self, relative: &str, field: &str) -> Result<PathBuf> {
let primary = self
.local()
.parent()
.and_then(Path::parent)
.expect("validated primary checkout cache location");
reject_symlink_traversal(
primary,
".workspace-mgr/local/cache",
"native storage cache",
)?;
reject_symlink_traversal(self.local(), "cache", "native storage cache")?;
if self.root.exists() && !self.root.is_dir() {
return Err(Error::message("storage cache is not a directory"));
}
reject_symlink_traversal(&self.root, relative, field)?;
Ok(self.root.join(relative))
}
pub(crate) fn existing(&self, digest: &str, hash_name: &str) -> Result<PathBuf> {
let preferred = self.path(digest, hash_name)?;
if preferred.is_file() {
return Ok(preferred);
}
Ok(
crate::legacy_dvc::existing_cache(&self.repo.root, digest, hash_name)
.unwrap_or(preferred),
)
}
pub(crate) fn entry(&self, entry: &StorageEntry) -> Result<PathBuf> {
if let Some(proof) = &entry.verification {
proof.checksum.validate()?;
let (first, rest) = proof.checksum.digest.split_at(2);
let path = self.checked_path(
&format!("objects/sha256/{first}/{rest}"),
"verified storage cache",
)?;
return Ok(path);
}
let digest = entry
.md5
.as_deref()
.ok_or_else(|| Error::message("storage cache entry has no checksum"))?;
let Some(version) = entry
.version_id
.as_deref()
.filter(|_| entry.hash_name == "md5-dos2unix")
else {
return self.existing(digest, &entry.hash_name);
};
Checksum {
algorithm: entry.hash_name.clone(),
digest: digest.into(),
}
.validate()?;
if version.trim().is_empty() || version == "null" {
return Err(Error::message(
"storage cache entry has no immutable exact version",
));
}
let route = self.route.get_or_init(|| {
CacheRouteSnapshot::capture(self.repo).map_err(|error| error.to_string())
});
let route = route
.as_ref()
.map_err(|error| Error::message(error.clone()))?;
route.validate()?;
let identity = crate::hex::encode_lower(sha2::Sha256::digest(
serde_json::to_vec(&json!([route.value, entry.object, version, digest]))
.map_err(|error| Error::message(error.to_string()))?,
));
self.checked_path(
&format!("versions/md5-dos2unix/{identity}"),
"exact-version storage cache",
)
}
#[cfg(test)]
pub(crate) fn install_entry(&self, entry: &StorageEntry, source: &Path) -> Result<()> {
self.install_entry_with_inventory(entry, source, &mut HashInventory::new())
}
pub(crate) fn install_entry_with_inventory(
&self,
entry: &StorageEntry,
source: &Path,
hashes: &mut HashInventory,
) -> Result<()> {
let digest = entry
.md5
.as_deref()
.ok_or_else(|| Error::message("storage cache entry has no checksum"))?;
let full = entry.verification.is_some()
|| entry.hash_name == "md5-dos2unix"
|| hashes
.files
.get(source)
.is_some_and(|(_, hashes)| !hashes.sha256.is_empty());
let source_hashes = hashes.inspect(source, full)?;
if source_hashes.digest(&entry.hash_name)? != digest
|| entry
.verification
.as_ref()
.is_some_and(|proof| source_hashes.sha256 != proof.checksum.digest)
|| entry.size.is_some_and(|size| {
fs::metadata(source).map(|metadata| metadata.len()).ok() != Some(size)
})
{
return Err(Error::message("downloaded content size or hash mismatch"));
}
let destination = if entry.verification.is_some()
|| (entry.hash_name == "md5-dos2unix" && entry.version_id.is_some())
{
self.entry(entry)?
} else {
self.path(digest, &entry.hash_name)?
};
atomic_copy(source, &destination)?;
let destination_hashes = hashes.inspect(&destination, full)?;
if destination_hashes.digest(&entry.hash_name)? != digest
|| entry
.verification
.as_ref()
.is_some_and(|proof| destination_hashes.sha256 != proof.checksum.digest)
|| (entry.hash_name == "md5-dos2unix"
&& hashes.hashes(source)?.sha256 != destination_hashes.sha256)
{
return Err(Error::message(
"downloaded content changed while installing its cache",
));
}
Ok(())
}
}
fn digest_parts(digest: &str) -> Result<(&str, &str)> {
let md5 = digest.strip_suffix(".dir").unwrap_or(digest);
if md5.len() != 32 || !md5.bytes().all(|b| matches!(b, b'0'..=b'9' | b'a'..=b'f')) {
return Err(Error::message(format!(
"invalid storage MD5 digest: {digest:?}"
)));
}
Ok(digest.split_at(2))
}
#[cfg(test)]
pub(crate) fn cache_path(repo: &GitRepo, digest: &str) -> Result<PathBuf> {
cache_path_with_algorithm(repo, digest, "md5")
}
pub(crate) fn cache_path_with_algorithm(
repo: &GitRepo,
digest: &str,
hash_name: &str,
) -> Result<PathBuf> {
CachePaths::new(repo)?.path(digest, hash_name)
}
#[cfg(test)]
pub(crate) fn existing_cache(repo: &GitRepo, digest: &str) -> Result<PathBuf> {
existing_cache_with_algorithm(repo, digest, "md5")
}
#[cfg(test)]
pub(crate) fn existing_cache_with_algorithm(
repo: &GitRepo,
digest: &str,
hash_name: &str,
) -> Result<PathBuf> {
CachePaths::new(repo)?.existing(digest, hash_name)
}
pub(crate) fn cache_path_for_entry(repo: &GitRepo, entry: &StorageEntry) -> Result<PathBuf> {
CachePaths::new(repo)?.entry(entry)
}
#[cfg(test)]
pub(crate) fn install_cache_for_entry(
repo: &GitRepo,
entry: &StorageEntry,
source: &Path,
) -> Result<()> {
CachePaths::new(repo)?.install_entry(entry, source)
}
pub(crate) fn exact_raw_bytes_match(
repo: &GitRepo,
entry: &StorageEntry,
local: &Path,
) -> Result<bool> {
if let Some(proof) = &entry.verification {
return Ok(local.is_file() && file_sha256(local)? == proof.checksum.digest);
}
if entry.hash_name != "md5-dos2unix" || entry.version_id.is_none() {
return Ok(true);
}
let cache = cache_path_for_entry(repo, entry)?;
Ok(cache.is_file() && local.is_file() && file_sha256(local)? == file_sha256(&cache)?)
}
pub(crate) fn normalized_exact_cache_missing(repo: &GitRepo, pointer: &str) -> Result<bool> {
if pointer_algorithm(repo, pointer)? != "md5-dos2unix" {
return Ok(false);
}
for entry in metadata_entries(repo, None, &[pointer.into()])? {
if entry.verification.is_none()
&& entry.version_id.is_some()
&& !cache_path_for_entry(repo, &entry)?.is_file()
{
return Ok(true);
}
}
Ok(false)
}
fn cache_for_recorded_file(
cache: &CachePaths<'_>,
object: &str,
file: &FileState,
algorithm: &str,
) -> Result<PathBuf> {
cache.entry(&StorageEntry {
pointer: String::new(),
object: if file.relpath.is_empty() {
object.into()
} else {
format!("{object}/{}", file.relpath)
},
md5: Some(file.md5.clone()),
size: Some(file.size),
version_id: file.version_id.clone(),
etag: None,
verification: file.verification.clone(),
hash_name: algorithm.into(),
})
}
fn normalized_exact_bytes_match(
repo: &GitRepo,
cache_paths: &CachePaths<'_>,
hashes: &mut HashInventory,
object: &str,
files: &[FileState],
algorithm: &str,
) -> Result<bool> {
for file in files.iter().filter(|file| file.verification.is_some()) {
let local = if file.relpath.is_empty() {
repo.root.join(object)
} else {
repo.root.join(object).join(&file.relpath)
};
if !local.is_file()
|| hashes.hashes(&local)?.sha256 != file.verification.as_ref().unwrap().checksum.digest
{
return Ok(false);
}
}
if algorithm != "md5-dos2unix" {
return Ok(true);
}
let cache_files = files
.iter()
.filter(|file| file.version_id.is_some() && file.verification.is_none())
.map(|file| cache_for_recorded_file(cache_paths, object, file, algorithm))
.collect::<Result<Vec<_>>>()?;
if cache_files.iter().any(|path| !path.is_file()) {
return Ok(false);
}
hashes.preload(&cache_files, algorithm)?;
for file in files
.iter()
.filter(|file| file.version_id.is_some() && file.verification.is_none())
{
let cache = cache_for_recorded_file(cache_paths, object, file, algorithm)?;
let local = if file.relpath.is_empty() {
repo.root.join(object)
} else {
repo.root.join(object).join(&file.relpath)
};
if !cache.is_file()
|| !local.is_file()
|| hashes.hashes(&local)?.sha256 != hashes.hashes(&cache)?.sha256
{
return Ok(false);
}
}
Ok(true)
}
pub(crate) fn file_digest(path: &Path, hash_name: &str) -> Result<String> {
let mut file = fs::File::open(path).at(path)?;
stream_digest(&mut file, path, hash_name)
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct FileHashes {
pub(crate) md5: String,
pub(crate) normalized_md5: String,
pub(crate) sha256: String,
}
impl FileHashes {
pub(crate) fn digest(&self, algorithm: &str) -> Result<&str> {
match algorithm {
"md5" => Ok(&self.md5),
"md5-dos2unix" => Ok(&self.normalized_md5),
_ => Err(Error::message("unsupported storage hash algorithm")),
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct FileIdentity {
length: u64,
#[cfg(unix)]
device: u64,
#[cfg(unix)]
inode: u64,
#[cfg(unix)]
modified: (i64, i64),
#[cfg(unix)]
changed: (i64, i64),
#[cfg(not(unix))]
modified: Option<std::time::SystemTime>,
#[cfg(not(unix))]
created: Option<std::time::SystemTime>,
}
impl FileIdentity {
fn of(metadata: &fs::Metadata) -> Self {
#[cfg(unix)]
use std::os::unix::fs::MetadataExt;
Self {
length: metadata.len(),
#[cfg(unix)]
device: metadata.dev(),
#[cfg(unix)]
inode: metadata.ino(),
#[cfg(unix)]
modified: (metadata.mtime(), metadata.mtime_nsec()),
#[cfg(unix)]
changed: (metadata.ctime(), metadata.ctime_nsec()),
#[cfg(not(unix))]
modified: metadata.modified().ok(),
#[cfg(not(unix))]
created: metadata.created().ok(),
}
}
}
#[derive(Clone)]
pub(crate) struct HashInventory {
files: BTreeMap<PathBuf, (FileIdentity, FileHashes)>,
workers: usize,
}
impl Default for HashInventory {
fn default() -> Self {
Self {
files: BTreeMap::new(),
workers: std::thread::available_parallelism()
.map(usize::from)
.unwrap_or(1),
}
}
}
impl HashInventory {
pub(crate) fn new() -> Self {
Self::default()
}
fn with_workers(workers: usize) -> Self {
Self {
files: BTreeMap::new(),
workers: workers.max(1),
}
}
pub(crate) fn hashes(&mut self, path: &Path) -> Result<FileHashes> {
self.inspect(path, true)
}
fn inspect(&mut self, path: &Path, full: bool) -> Result<FileHashes> {
let before = fs::metadata(path).at(path)?;
if !before.is_file() {
return Err(Error::message(format!(
"content verification requires a regular file: {}",
path.display()
)));
}
let identity = FileIdentity::of(&before);
#[cfg(unix)]
if let Some((previous, hashes)) = self.files.get(path)
&& *previous == identity
&& (!full || !hashes.sha256.is_empty())
{
return Ok(hashes.clone());
}
let mut input = fs::File::open(path).at(path)?;
if FileIdentity::of(&input.metadata().at(path)?) != identity {
return Err(Error::message("file changed before content verification"));
}
let hashes = if full {
stream_hashes(&mut input, path)?
} else {
FileHashes {
md5: stream_digest(&mut input, path, "md5")?,
normalized_md5: String::new(),
sha256: String::new(),
}
};
if FileIdentity::of(&input.metadata().at(path)?) != identity
|| FileIdentity::of(&fs::metadata(path).at(path)?) != identity
{
self.files.remove(path);
return Err(Error::message(format!(
"file changed during content verification: {}",
path.display()
)));
}
self.files
.insert(path.to_owned(), (identity, hashes.clone()));
Ok(hashes)
}
pub(crate) fn digest(&mut self, path: &Path, algorithm: &str) -> Result<String> {
self.inspect(path, algorithm != "md5")?
.digest(algorithm)
.map(str::to_owned)
}
fn preload(&mut self, paths: &[PathBuf], algorithm: &str) -> Result<()> {
let paths = paths
.iter()
.cloned()
.collect::<BTreeSet<_>>()
.into_iter()
.collect::<Vec<_>>();
let previous = &self.files;
let workers = self.workers;
let checked =
crate::native_versions::bounded_map_with_workers(&paths, workers, false, |path| {
let mut local = HashInventory::new();
if let Some(recorded) = previous.get(path) {
local.files.insert(path.clone(), recorded.clone());
}
local.inspect(path, algorithm != "md5")?;
let checked = local.files.remove(path).expect("verified file inventory");
Ok((path.clone(), checked))
})?;
self.files.extend(checked);
Ok(())
}
}
pub(crate) struct FileHashStream {
raw: Md5,
normalized: Md5,
sha256: sha2::Sha256,
pending: Vec<u8>,
}
impl FileHashStream {
pub(crate) fn new() -> Self {
Self {
raw: Md5::new(),
normalized: Md5::new(),
sha256: sha2::Sha256::new(),
pending: Vec::with_capacity(1024 * 1024),
}
}
pub(crate) fn update(&mut self, mut bytes: &[u8]) {
self.raw.update(bytes);
self.sha256.update(bytes);
while !bytes.is_empty() {
let count = (1024 * 1024 - self.pending.len()).min(bytes.len());
self.pending.extend_from_slice(&bytes[..count]);
bytes = &bytes[count..];
if self.pending.len() == 1024 * 1024 {
Self::normalize(&mut self.normalized, &self.pending);
self.pending.clear();
}
}
}
fn normalize(hasher: &mut Md5, chunk: &[u8]) {
let sample = &chunk[..chunk.len().min(512)];
let nontext = sample
.iter()
.filter(|&&byte| !matches!(byte, 32..=126 | b'\n' | b'\r' | b'\t' | b'\x0c' | b'\x08'))
.count();
let text = !sample.contains(&0) && nontext * 10 <= sample.len() * 3;
if !text {
hasher.update(chunk);
return;
}
let mut start = 0;
for index in 0..chunk.len().saturating_sub(1) {
if chunk[index] == b'\r' && chunk[index + 1] == b'\n' {
hasher.update(&chunk[start..index]);
start = index + 1;
}
}
hasher.update(&chunk[start..]);
}
pub(crate) fn finish(mut self) -> FileHashes {
if !self.pending.is_empty() {
Self::normalize(&mut self.normalized, &self.pending);
}
FileHashes {
md5: crate::hex::encode_lower(self.raw.finalize()),
normalized_md5: crate::hex::encode_lower(self.normalized.finalize()),
sha256: crate::hex::encode_lower(self.sha256.finalize()),
}
}
}
pub(crate) fn stream_hashes(input: &mut impl Read, path: &Path) -> Result<FileHashes> {
let mut hasher = FileHashStream::new();
let mut buffer = [0u8; 128 * 1024];
loop {
match input.read(&mut buffer) {
Ok(0) => break,
Ok(size) => hasher.update(&buffer[..size]),
Err(error) if error.kind() == std::io::ErrorKind::Interrupted => continue,
Err(source) => {
return Err(Error::Io {
path: path.into(),
source,
});
}
}
}
Ok(hasher.finish())
}
fn file_sha256(path: &Path) -> Result<String> {
let mut file = fs::File::open(path).at(path)?;
let mut hasher = sha2::Sha256::new();
let mut buffer = [0u8; 128 * 1024];
loop {
match file.read(&mut buffer) {
Ok(0) => break,
Ok(size) => hasher.update(&buffer[..size]),
Err(error) if error.kind() == std::io::ErrorKind::Interrupted => continue,
Err(source) => {
return Err(Error::Io {
path: path.to_owned(),
source,
});
}
}
}
Ok(crate::hex::encode_lower(hasher.finalize()))
}
pub(crate) fn stream_digest(input: &mut impl Read, path: &Path, hash_name: &str) -> Result<String> {
if !matches!(hash_name, "md5" | "md5-dos2unix") {
return Err(Error::message(format!(
"unsupported storage hash algorithm: {hash_name}"
)));
}
let mut hasher = Md5::new();
let legacy = hash_name == "md5-dos2unix";
let mut buffer = vec![0u8; if legacy { 1024 * 1024 } else { 128 * 1024 }];
loop {
let mut size = 0;
while size < buffer.len() {
match input.read(&mut buffer[size..]) {
Ok(0) => break,
Ok(count) => size += count,
Err(error) if error.kind() == std::io::ErrorKind::Interrupted => continue,
Err(source) => {
return Err(Error::Io {
path: path.to_owned(),
source,
});
}
}
}
if size == 0 {
break;
}
if !legacy {
hasher.update(&buffer[..size]);
continue;
}
let sample = &buffer[..size.min(512)];
let nontext = sample
.iter()
.filter(|&&byte| !matches!(byte, 32..=126 | b'\n' | b'\r' | b'\t' | b'\x0c' | b'\x08'))
.count();
let text = !sample.contains(&0) && nontext * 10 <= sample.len() * 3;
if !text {
hasher.update(&buffer[..size]);
continue;
}
let mut normalized = Vec::with_capacity(size);
for (index, &byte) in buffer[..size].iter().enumerate() {
if byte != b'\r' || index + 1 == size || buffer[index + 1] != b'\n' {
normalized.push(byte);
}
}
hasher.update(&normalized);
}
Ok(crate::hex::encode_lower(hasher.finalize()))
}
#[cfg(test)]
pub(crate) fn install_cache(repo: &GitRepo, digest: &str, bytes: &[u8]) -> Result<()> {
install_cache_with_algorithm(repo, digest, bytes, "md5")
}
#[cfg(test)]
fn install_cache_with_algorithm(
repo: &GitRepo,
digest: &str,
bytes: &[u8],
hash_name: &str,
) -> Result<()> {
let wanted = digest.strip_suffix(".dir").unwrap_or(digest);
if crate::hex::encode_lower(Md5::digest(bytes)) != wanted {
return Err(Error::message("downloaded content hash mismatch"));
}
atomic_write(&cache_path_with_algorithm(repo, digest, hash_name)?, bytes)
}
#[cfg(test)]
pub(crate) fn install_cache_file(repo: &GitRepo, digest: &str, path: &Path) -> Result<()> {
install_cache_file_with_algorithm(repo, digest, path, "md5")
}
#[cfg(test)]
pub(crate) fn install_cache_file_with_algorithm(
repo: &GitRepo,
digest: &str,
path: &Path,
hash_name: &str,
) -> Result<()> {
let wanted = digest.strip_suffix(".dir").unwrap_or(digest);
if file_digest(path, hash_name)? != wanted {
return Err(Error::message("downloaded content hash mismatch"));
}
atomic_copy(path, &cache_path_with_algorithm(repo, digest, hash_name)?)
}
fn read_manifest(repo: &GitRepo, pointer: &str) -> Result<Manifest> {
if !pointer.ends_with(crate::storage_format::SUFFIX) {
return Err(Error::message(
"legacy DVC metadata must be migrated before mutation; run `workspace-mgr manage`",
));
}
reject_symlink_traversal(&repo.root, pointer, "storage metadata")?;
let path = resolved_under(&repo.root, pointer);
if !fs::symlink_metadata(&path).at(&path)?.is_file() {
return Err(Error::message("storage metadata must be a regular file"));
}
Manifest::parse(&fs::read_to_string(&path).at(&path)?, pointer)
}
fn write_manifest(repo: &GitRepo, pointer: &str, manifest: &Manifest) -> Result<()> {
if !pointer.ends_with(crate::storage_format::SUFFIX) {
return Err(Error::message(
"native storage writes require a .wm-storage.json sidecar",
));
}
atomic_write(
&resolved_under(&repo.root, pointer),
manifest.serialize()?.as_bytes(),
)
}
fn output_object(pointer: &str, path: &str) -> Result<String> {
let parent = Path::new(pointer).parent().unwrap_or_else(|| Path::new(""));
repo_path(&to_slash(&parent.join(path)), "storage output")
}
fn pointer_algorithm(repo: &GitRepo, pointer: &str) -> Result<String> {
reject_symlink_traversal(&repo.root, pointer, "storage metadata")?;
let path = repo.root.join(pointer);
let raw = storage_metadata::normalize_pointer_in_repo(
repo,
None,
&fs::read_to_string(&path).at(&path)?,
pointer,
)?;
storage_metadata::hash_algorithm(&raw, pointer)
}
pub(crate) fn metadata_entries(
repo: &GitRepo,
revision: Option<&str>,
pointers: &[String],
) -> Result<Vec<StorageEntry>> {
let mut entries = Vec::new();
for pointer in pointers {
repo_path(pointer, "storage metadata")?;
}
let documents = storage_metadata::read_pointer_documents(repo, revision, pointers)?;
for pointer in pointers {
let (parsed, hash_name) = &documents[pointer];
entries.extend(entries_from_document(pointer, parsed, hash_name)?);
}
Ok(entries)
}
pub(crate) fn entries_from_document(
pointer: &str,
parsed: &storage_metadata::PointerDocument,
hash_name: &str,
) -> Result<Vec<StorageEntry>> {
repo_path(pointer, "storage metadata")?;
let mut entries = Vec::new();
if parsed.outs.is_empty() {
return Err(Error::message(format!(
"storage metadata has no outputs: {pointer}"
)));
}
for out in &parsed.outs {
if let Some(digest) = &out.md5 {
digest_parts(digest)?;
}
let object = output_object(pointer, &out.path)?;
if let Some(files) = &out.files {
if pointer.ends_with(".dvc")
&& out.md5.as_deref() != Some(directory_digest(files)?.as_str())
{
return Err(Error::message(format!(
"legacy directory manifest hash mismatch: {pointer}"
)));
}
for file in files {
if let Some(digest) = &file.md5 {
digest_parts(digest)?;
if digest.ends_with(".dir") {
return Err(Error::message(
"directory manifest cannot contain another directory digest",
));
}
}
entries.push(StorageEntry {
pointer: pointer.to_owned(),
object: repo_path(
&format!("{object}/{}", file.relpath),
"stored directory file",
)?,
md5: file.md5.clone(),
size: file.size,
version_id: file.version_id.clone(),
etag: file.etag.clone(),
verification: file.verification.clone(),
hash_name: hash_name.to_owned(),
});
}
} else if out
.md5
.as_deref()
.is_some_and(|digest| digest.ends_with(".dir"))
{
return Err(Error::message(format!(
"directory metadata is incomplete: {pointer}; restore its published file/version manifest"
)));
} else {
entries.push(StorageEntry {
pointer: pointer.to_owned(),
object,
md5: out.md5.clone(),
size: out.size,
version_id: out.version_id.clone(),
etag: out.etag.clone(),
verification: out.verification.clone(),
hash_name: hash_name.to_owned(),
});
}
}
Ok(entries)
}
pub(crate) fn install_directory_manifests(repo: &GitRepo, pointers: &[String]) -> Result<()> {
let documents = storage_metadata::read_pointer_documents(repo, None, pointers)?;
let cache = CachePaths::new(repo)?;
for pointer in pointers {
if pointer.ends_with(crate::storage_format::SUFFIX) {
read_manifest(repo, pointer)?;
continue;
}
let (document, algorithm) = &documents[pointer];
for out in &document.outs {
if let Some(files) = &out.files {
let bytes = tree_manifest_bytes(files)?;
let digest = format!("{}.dir", crate::hex::encode_lower(Md5::digest(&bytes)));
if out.md5.as_deref() != Some(&digest) {
return Err(Error::message("legacy directory manifest hash mismatch"));
}
atomic_write(&cache.path(&digest, algorithm)?, &bytes)?;
}
}
}
Ok(())
}
fn select_pointers(repo: &GitRepo, targets: &[String]) -> Result<Vec<String>> {
if targets.is_empty() {
return storage_metadata::discover(repo, &[]);
}
let mut pointers = BTreeSet::new();
let mut scopes = Vec::new();
for target in targets {
let target = repo_path(target, "storage target")?;
let pointer = if storage_metadata::is_pointer(&target) {
target.clone()
} else {
let native = storage_metadata::pointer_path(&target);
if repo.root.join(&native).is_file() {
native
} else {
format!("{target}.dvc")
}
};
if repo.root.join(&pointer).is_file() {
pointers.insert(pointer);
} else {
scopes.push(target);
}
}
if !scopes.is_empty() {
pointers.extend(storage_metadata::discover(repo, &scopes)?);
}
if pointers.is_empty() {
return Err(Error::message(
"storage targets did not select any metadata",
));
}
Ok(pointers.into_iter().collect())
}
pub(crate) fn payload_matches_manifest(
repo: &GitRepo,
object: &str,
manifest: &Manifest,
) -> Result<bool> {
let mut recorded = match manifest.kind {
crate::storage_format::Kind::File => vec![(
String::new(),
manifest.checksum.digest.clone(),
manifest.size,
)],
crate::storage_format::Kind::Directory => manifest
.entries
.as_ref()
.ok_or_else(|| Error::message("directory manifest has no entries"))?
.iter()
.map(|entry| {
(
entry.path.clone(),
entry.checksum.digest.clone(),
entry.size,
)
})
.collect(),
};
recorded.sort();
let current = current_files_with_algorithm(repo, object, &manifest.checksum.algorithm)?
.into_iter()
.map(|file| (file.relpath, file.md5, file.size))
.collect::<Vec<_>>();
Ok(!current.is_empty() && current == recorded)
}
#[cfg(test)]
fn current_files(repo: &GitRepo, object: &str) -> Result<Vec<FileState>> {
current_files_with_algorithm(repo, object, "md5")
}
fn current_files_with_algorithm(
repo: &GitRepo,
object: &str,
hash_name: &str,
) -> Result<Vec<FileState>> {
current_files_with_inventory(repo, object, hash_name, &mut HashInventory::new())
}
fn current_files_with_inventory(
repo: &GitRepo,
object: &str,
hash_name: &str,
hashes: &mut HashInventory,
) -> Result<Vec<FileState>> {
reject_symlink_traversal(&repo.root, object, "storage output")?;
let root = repo.root.join(object);
let mut discovered = Vec::new();
if !root.exists() {
return Ok(Vec::new());
}
for item in WalkDir::new(&root).follow_links(false).sort_by_file_name() {
let item =
item.map_err(|e| Error::message(format!("failed to inspect storage output: {e}")))?;
let metadata = fs::metadata(item.path()).at(item.path())?;
if (item.file_type().is_symlink() && !metadata.is_file())
|| (!metadata.is_dir() && !metadata.is_file())
{
return Err(Error::message(
"storage output contains a symlink or special file",
));
}
if metadata.is_dir() {
continue;
}
let relpath = if root.is_dir() {
let relative = item.path().strip_prefix(&root).expect("walk root");
if relative.to_str().is_none() {
return Err(Error::message(
"stored directory contains a filename that is not UTF-8",
));
}
to_slash(relative)
} else {
String::new()
};
if !relpath.is_empty() {
repo_path(&relpath, "stored directory file")?;
}
discovered.push((item.path().to_owned(), relpath));
}
hashes.preload(
&discovered
.iter()
.map(|(path, _)| path.clone())
.collect::<Vec<_>>(),
hash_name,
)?;
let mut files = Vec::with_capacity(discovered.len());
for (path, relpath) in discovered {
let md5 = hashes.digest(&path, hash_name)?;
let size = hashes.files[&path].0.length;
files.push(FileState {
relpath,
md5,
size,
version_id: None,
verification: None,
});
}
files.sort_by(|a, b| a.relpath.cmp(&b.relpath));
Ok(files)
}
#[cfg(test)]
fn recorded_files(
repo: &GitRepo,
pointer: &str,
out: &storage_metadata::PointerOutput,
algorithm: &str,
) -> Result<Vec<FileState>> {
recorded_files_with_cache(repo, &CachePaths::new(repo)?, pointer, out, algorithm)
}
fn recorded_files_with_cache(
repo: &GitRepo,
cache: &CachePaths<'_>,
pointer: &str,
out: &storage_metadata::PointerOutput,
algorithm: &str,
) -> Result<Vec<FileState>> {
if let Some(files) = &out.files {
if pointer.ends_with(".dvc")
&& out.md5.as_deref() != Some(directory_digest(files)?.as_str())
{
return Err(Error::message(format!(
"directory manifest hash mismatch: {pointer}"
)));
}
return files
.iter()
.map(|file| {
Ok(FileState {
relpath: repo_path(&file.relpath, "directory manifest entry")?,
md5: file
.md5
.clone()
.ok_or_else(|| Error::message("directory file has no digest"))?,
size: file.size.unwrap_or(0),
version_id: file.version_id.clone(),
verification: file.verification.clone(),
})
})
.collect();
}
let digest = out
.md5
.as_deref()
.ok_or_else(|| Error::message(format!("metadata has no content digest: {pointer}")))?;
if !digest.ends_with(".dir") {
return Ok(vec![FileState {
relpath: String::new(),
md5: digest.to_owned(),
size: out.size.unwrap_or(0),
version_id: out.version_id.clone(),
verification: out.verification.clone(),
}]);
}
let mut path = cache.existing(digest, algorithm)?;
if !path.is_file()
&& let Ok(remote) = remote_root(repo)
&& !remote.contains("://")
{
path = remote_cache_path(&remote, digest, repo, algorithm)?;
}
let bytes = fs::read(&path).at(&path)?;
if crate::hex::encode_lower(Md5::digest(&bytes)) != digest.trim_end_matches(".dir") {
return Err(Error::message("cached directory manifest hash mismatch"));
}
let files: Vec<TreeEntry> = serde_json::from_slice(&bytes)
.map_err(|e| Error::message(format!("invalid cached directory manifest: {e}")))?;
tree_bytes(&files)?;
files
.into_iter()
.map(|file| {
let path = cache.existing(&file.md5, algorithm)?;
Ok(FileState {
relpath: file.relpath,
md5: file.md5,
size: fs::metadata(path).map(|m| m.len()).unwrap_or(0),
version_id: None,
verification: None,
})
})
.collect()
}
fn add(repo: &GitRepo, target: &str) -> Result<()> {
let object = repo_path(target, "storage output")?;
storage_metadata::require_addressable(
&object,
"storage output",
"choose a path without backslashes",
)?;
let pointer = storage_metadata::pointer_path(&object);
reject_symlink_traversal(&repo.root, &pointer, "storage metadata")?;
if repo.root.join(format!("{object}.dvc")).exists() {
return Err(Error::message(
"legacy DVC metadata must be migrated before tracking this boundary; run `workspace-mgr manage`",
));
}
let parent = Path::new(&object).parent().unwrap_or_else(|| Path::new(""));
let filename = Path::new(&object)
.file_name()
.and_then(|name| name.to_str())
.ok_or_else(|| Error::message("output name is not UTF-8"))?;
let mut manifest = if repo.root.join(&pointer).exists() {
read_manifest(repo, &pointer)?
} else {
Manifest {
schema_version: crate::storage_format::SCHEMA_VERSION,
path: filename.to_owned(),
kind: Kind::File,
checksum: Checksum {
algorithm: "md5".into(),
digest: "00000000000000000000000000000000".into(),
},
size: 0,
version: None,
entries: None,
}
};
update_manifest(repo, &pointer, &mut manifest)?;
write_manifest(repo, &pointer, &manifest)?;
update_ignore(&repo.root.join(parent).join(".gitignore"), filename, true)
}
#[cfg(test)]
fn commit(repo: &GitRepo, pointer: &str) -> Result<()> {
commit_with_cache(
repo,
&CachePaths::new(repo)?,
pointer,
std::thread::available_parallelism()
.map(usize::from)
.unwrap_or(1),
)
}
fn commit_with_cache(
repo: &GitRepo,
cache: &CachePaths<'_>,
pointer: &str,
workers: usize,
) -> Result<()> {
let mut manifest = read_manifest(repo, pointer)?;
update_manifest_with_cache(repo, cache, pointer, &mut manifest, workers)?;
write_manifest(repo, pointer, &manifest)
}
fn update_manifest(repo: &GitRepo, pointer: &str, manifest: &mut Manifest) -> Result<()> {
update_manifest_with_cache(
repo,
&CachePaths::new(repo)?,
pointer,
manifest,
std::thread::available_parallelism()
.map(usize::from)
.unwrap_or(1),
)
}
fn update_manifest_with_cache(
repo: &GitRepo,
cache: &CachePaths<'_>,
pointer: &str,
manifest: &mut Manifest,
workers: usize,
) -> Result<()> {
let mut hashes = HashInventory::with_workers(workers);
let object = output_object(pointer, &manifest.path)?;
let path = repo.root.join(&object);
if !path.exists() {
return Err(Error::message(format!(
"storage output is missing: {object}"
)));
}
let previous = entries_from_document(
pointer,
&storage_metadata::logical_document(manifest),
&manifest.checksum.algorithm,
)?;
let proof_paths = previous
.iter()
.filter(|entry| entry.verification.is_some())
.map(|entry| repo.root.join(&entry.object))
.filter(|path| path.is_file())
.collect::<Vec<_>>();
hashes.preload(&proof_paths, "md5-dos2unix")?;
let changed_proofs = previous
.iter()
.filter_map(|entry| entry.verification.as_ref().map(|proof| (entry, proof)))
.map(|(entry, proof)| {
let local = repo.root.join(&entry.object);
Ok((
entry.object.clone(),
!local.is_file() || hashes.hashes(&local)?.sha256 != proof.checksum.digest,
))
})
.collect::<Result<BTreeMap<_, _>>>()?;
if let Some(entries) = &mut manifest.entries {
for entry in entries {
if changed_proofs.get(&format!("{object}/{}", entry.path)) == Some(&true) {
entry.version = None;
}
}
} else if changed_proofs.get(&object) == Some(&true) {
manifest.version = None;
}
if manifest.checksum.algorithm == "md5-dos2unix" {
let document = storage_metadata::read_pointer_document(repo, pointer)?;
let [out] = document.outs.as_slice() else {
return Err(Error::message(
"normalized storage metadata must define one output",
));
};
let recorded = recorded_files_with_cache(repo, cache, pointer, out, "md5-dos2unix")?;
let unchanged = recorded.iter().all(|file| file.version_id.is_some())
&& normalized_exact_bytes_match(
repo,
cache,
&mut hashes,
&object,
&recorded,
"md5-dos2unix",
)?;
if !unchanged {
manifest.checksum.algorithm = "md5".into();
manifest.clear_versions();
}
}
let files =
current_files_with_inventory(repo, &object, &manifest.checksum.algorithm, &mut hashes)?;
let size = files.iter().try_fold(0u64, |total, file| {
total
.checked_add(file.size)
.ok_or_else(|| Error::message("storage content sizes overflow"))
})?;
let (kind, digest, entries) = if path.is_dir() {
let previous = manifest
.entries
.as_deref()
.unwrap_or_default()
.iter()
.map(|entry| (entry.path.as_str(), entry))
.collect::<BTreeMap<_, _>>();
let entries = files
.iter()
.map(|file| Entry {
path: file.relpath.clone(),
checksum: Checksum {
algorithm: manifest.checksum.algorithm.clone(),
digest: file.md5.clone(),
},
size: file.size,
version: previous
.get(file.relpath.as_str())
.filter(|entry| entry.checksum.digest == file.md5 && entry.size == file.size)
.and_then(|entry| entry.version.clone()),
})
.collect::<Vec<_>>();
(
Kind::Directory,
crate::storage_format::directory_digest(&entries)?,
Some(entries),
)
} else {
(
Kind::File,
files
.first()
.ok_or_else(|| Error::message("storage output is not a file"))?
.md5
.clone(),
None,
)
};
if manifest.kind != kind || manifest.checksum.digest != digest || manifest.size != size {
manifest.version = None;
}
manifest.kind = kind;
manifest.checksum.digest = digest;
manifest.size = size;
manifest.entries = entries;
let bindings = entries_from_document(
pointer,
&storage_metadata::logical_document(manifest),
&manifest.checksum.algorithm,
)?
.into_iter()
.map(|entry| (entry.object.clone(), entry))
.collect::<BTreeMap<_, _>>();
let mut copies: BTreeMap<PathBuf, Vec<_>> = BTreeMap::new();
for file in files {
let source = if file.relpath.is_empty() {
path.clone()
} else {
path.join(&file.relpath)
};
if hashes.digest(&source, &manifest.checksum.algorithm)? != file.md5 {
return Err(Error::message("downloaded content hash mismatch"));
}
let object_path = if file.relpath.is_empty() {
object.clone()
} else {
format!("{object}/{}", file.relpath)
};
let bound = bindings
.get(&object_path)
.filter(|entry| entry.verification.is_some());
let expected_raw = bound
.and_then(|entry| entry.verification.as_ref())
.map(|proof| proof.checksum.digest.clone());
if let Some(expected) = &expected_raw
&& hashes.hashes(&source)?.sha256 != *expected
{
return Err(Error::message(
"storage payload changed after its raw content verification",
));
}
let destination = if let Some(bound) = bound {
cache.entry(bound)?
} else {
cache.path(&file.md5, &manifest.checksum.algorithm)?
};
let verified = hashes.files[&source].clone();
copies
.entry(destination)
.or_default()
.push((source, file.md5, verified, expected_raw));
}
let copies = copies.into_iter().collect::<Vec<_>>();
crate::native_versions::bounded_map_with_workers(
&copies,
workers,
false,
|(destination, sources)| {
for (source, wanted, verified, expected_raw) in sources {
let mut checked = HashInventory::with_workers(1);
checked.files.insert(source.clone(), verified.clone());
let full = expected_raw.is_some() || !verified.1.sha256.is_empty();
let current = checked.inspect(source, full)?;
if current.digest(&manifest.checksum.algorithm)? != *wanted
|| expected_raw
.as_ref()
.is_some_and(|expected| current.sha256 != *expected)
{
return Err(Error::message("downloaded content hash mismatch"));
}
atomic_copy(source, destination)?;
let copied = checked.inspect(destination, full)?;
let expected = expected_raw.as_deref().or_else(|| {
(!verified.1.sha256.is_empty()).then_some(verified.1.sha256.as_str())
});
if copied.digest(&manifest.checksum.algorithm)? != *wanted
|| expected.is_some_and(|expected| copied.sha256 != expected)
{
return Err(Error::message(
"content changed while installing the storage cache",
));
}
}
Ok(())
},
)?;
Ok(())
}
fn actual_directory_digest(pointer: &str, files: &[FileState], algorithm: &str) -> Result<String> {
if pointer.ends_with(".dvc") {
return Ok(format!(
"{}.dir",
crate::hex::encode_lower(Md5::digest(tree_bytes(
&files
.iter()
.map(|file| TreeEntry {
relpath: file.relpath.clone(),
md5: file.md5.clone()
})
.collect::<Vec<_>>()
)?))
));
}
let entries = files
.iter()
.map(|file| Entry {
path: file.relpath.clone(),
checksum: Checksum {
algorithm: algorithm.to_owned(),
digest: file.md5.clone(),
},
size: file.size,
version: None,
})
.collect::<Vec<_>>();
Ok(format!(
"{}.dir",
crate::storage_format::directory_digest(&entries)?
))
}
fn remote_root(repo: &GitRepo) -> Result<String> {
storage_metadata::internal_location(repo)?
.map(|location| location.0)
.ok_or_else(|| Error::message("storage remote is not configured"))
}
fn remote_cache_path(
remote: &str,
digest: &str,
repo: &GitRepo,
hash_name: &str,
) -> Result<PathBuf> {
let (first, rest) = digest_parts(digest)?;
let root = Path::new(remote);
let root = if root.is_absolute() {
root.to_owned()
} else {
repo.root.join(root)
};
let relative = match hash_name {
"md5" => format!("objects/md5/{first}/{rest}"),
"md5-dos2unix" => format!("objects/md5-dos2unix/{first}/{rest}"),
_ => return Err(Error::message("unsupported storage hash algorithm")),
};
reject_symlink_traversal(&root, &relative, "filesystem storage object")?;
Ok(root.join(relative))
}
fn algorithms_for_document(
repo: &GitRepo,
cache: &CachePaths<'_>,
pointer: &str,
document: &storage_metadata::PointerDocument,
algorithm: &str,
) -> Result<BTreeMap<String, String>> {
let mut hashes = BTreeMap::new();
for out in &document.outs {
if pointer.ends_with(".dvc")
&& let Some(digest) = &out.md5
{
hashes.insert(digest.clone(), algorithm.into());
}
for file in recorded_files_with_cache(repo, cache, pointer, out, algorithm)? {
hashes.insert(file.md5, algorithm.into());
}
}
Ok(hashes)
}
fn push(repo: &GitRepo, pointers: &[String]) -> Result<()> {
let remote = remote_root(repo)?;
if remote.starts_with("s3://") {
return push_versioned(repo, pointers);
}
if remote.contains("://") {
return Err(Error::message("unsupported storage remote scheme"));
}
let cache = CachePaths::new(repo)?;
let documents = storage_metadata::read_pointer_documents(repo, None, pointers)?;
for pointer in pointers {
let (document, algorithm) = &documents[pointer];
for (digest, algorithm) in
algorithms_for_document(repo, &cache, pointer, document, algorithm)?
{
let source = cache.existing(&digest, &algorithm)?;
if file_digest(&source, &algorithm)? != digest.trim_end_matches(".dir") {
return Err(Error::message("storage cache content hash mismatch"));
}
atomic_copy(
&source,
&remote_cache_path(&remote, &digest, repo, &algorithm)?,
)?;
}
}
Ok(())
}
fn push_versioned(repo: &GitRepo, pointers: &[String]) -> Result<()> {
let client = crate::native_s3::S3Client::from_repo(repo)?;
let versioning = client.call_s3(
"get_bucket_versioning",
&json!({"Bucket":client.bucket}),
None,
)?;
if versioning.value["Status"] != "Enabled" {
return Err(Error::message(
"version-aware storage requires enabled S3 bucket versioning",
));
}
let cache = CachePaths::new(repo)?;
let proof_pointers = pointers
.iter()
.filter_map(|pointer| {
read_manifest(repo, pointer)
.ok()
.filter(|manifest| manifest.schema_version == crate::storage_format::SCHEMA_VERSION)
.map(|_| pointer.clone())
})
.collect::<BTreeSet<_>>();
let entries = metadata_entries(repo, None, pointers)?;
let mut objects = BTreeSet::new();
for entry in &entries {
if !objects.insert(entry.object.clone()) {
return Err(Error::message(
"storage metadata repeats an upload object across overlapping boundaries",
));
}
}
let (bound, pending): (Vec<_>, Vec<_>) = entries
.into_iter()
.partition(|entry| entry.version_id.is_some());
crate::native_versions::verify_storage_entries(&client, repo, &bound)?;
let uploaded = crate::native_versions::bounded_map_with_workers(&pending, 4, true, |entry| {
let digest = entry
.md5
.as_deref()
.ok_or_else(|| Error::message("stored object has no content hash"))?;
let source = if proof_pointers.contains(&entry.pointer) {
let local = repo.root.join(&entry.object);
reject_symlink_traversal(&repo.root, &entry.object, "new storage payload")?;
if !local.is_file() {
return Err(Error::message(
"an unbound schema 2 payload must be materialized before its first upload",
));
}
local
} else {
cache.existing(digest, &entry.hash_name)?
};
let mut hashes = HashInventory::new();
let raw_sha256 = proof_pointers
.contains(&entry.pointer)
.then(|| hashes.hashes(&source).map(|hashes| hashes.sha256))
.transpose()?;
let (version, etag) = upload_version_in(
&client,
repo,
entry,
&source,
5 * (1 << 30),
64 * (1 << 20),
UploadPolicy {
namespace: "uploads",
condition: None,
raw_sha256: raw_sha256.as_deref(),
local: Some(cache.local()),
hashes: Some(&mut hashes),
},
)?;
let version = Version {
id: version.clone(),
etag: Some(etag),
verification: raw_sha256
.as_deref()
.map(|sha| verification_for(&client, entry, &version, sha, "verified-upload"))
.transpose()?,
};
if version.verification.is_some() {
let mut bound = entry.clone();
bound.version_id = Some(version.id.clone());
bound.verification = version.verification.clone();
cache.install_entry_with_inventory(&bound, &source, &mut hashes)?;
}
Ok((entry.clone(), version))
})?;
let mut by_pointer: BTreeMap<String, BTreeMap<String, (StorageEntry, Version)>> =
BTreeMap::new();
for (entry, version) in uploaded {
let pointer = by_pointer.entry(entry.pointer.clone()).or_default();
if pointer
.insert(entry.object.clone(), (entry, version))
.is_some()
{
return Err(Error::message("storage metadata repeats an upload object"));
}
}
for (pointer, mut uploaded) in by_pointer {
let mut manifest = read_manifest(repo, &pointer)?;
if (manifest.schema_version == crate::storage_format::SCHEMA_VERSION)
!= proof_pointers.contains(&pointer)
{
return Err(Error::message("storage schema changed during upload"));
}
let object = output_object(&pointer, &manifest.path)?;
if let Some(entries) = &mut manifest.entries {
for file in entries {
if let Some((entry, version)) = uploaded.remove(&format!("{object}/{}", file.path))
{
if !binding_is_unchanged(
&file.checksum,
file.size,
file.version.as_ref(),
&entry,
) {
return Err(Error::message(
"storage metadata changed during upload; retry after reconciling the pointer",
));
}
file.version = Some(version);
}
}
} else if let Some((entry, version)) = uploaded.remove(&object) {
if !binding_is_unchanged(
&manifest.checksum,
manifest.size,
manifest.version.as_ref(),
&entry,
) {
return Err(Error::message(
"storage metadata changed during upload; retry after reconciling the pointer",
));
}
manifest.version = Some(version);
}
if !uploaded.is_empty() {
return Err(Error::message(
"storage output changed during upload; its exact version remains recorded in the private upload journal",
));
}
write_manifest(repo, &pointer, &manifest)?;
}
Ok(())
}
fn binding_is_unchanged(
checksum: &Checksum,
size: u64,
version: Option<&Version>,
entry: &StorageEntry,
) -> bool {
Some(checksum.digest.as_str()) == entry.md5.as_deref()
&& checksum.algorithm == entry.hash_name
&& Some(size) == entry.size
&& version.map(|version| version.id.as_str()) == entry.version_id.as_deref()
&& version.and_then(|version| version.etag.as_deref()) == entry.etag.as_deref()
&& version.and_then(|version| version.verification.as_ref()) == entry.verification.as_ref()
}
const UPLOAD_TOKEN: &str = "workspace-mgr-upload";
#[cfg(test)]
fn upload_journal(
repo: &GitRepo,
client: &crate::native_s3::S3Client,
entry: &StorageEntry,
) -> Result<(PathBuf, Value)> {
upload_journal_in(repo, client, entry, "uploads", None)
}
fn upload_journal_in(
repo: &GitRepo,
client: &crate::native_s3::S3Client,
entry: &StorageEntry,
namespace: &str,
raw_sha256: Option<&str>,
) -> Result<(PathBuf, Value)> {
let local = crate::local_state::directory_unmigrated(repo)?;
upload_journal_at(&local, client, entry, namespace, raw_sha256)
}
fn upload_journal_at(
local: &Path,
client: &crate::native_s3::S3Client,
entry: &StorageEntry,
namespace: &str,
raw_sha256: Option<&str>,
) -> Result<(PathBuf, Value)> {
let (path, context) = upload_context_at(local, client, entry, namespace, raw_sha256)?;
if namespace == "uploads" && raw_sha256.is_some() {
let (old_path, old_context) = upload_context_at(local, client, entry, namespace, None)?;
if old_path.is_file() {
let mut old: Value = serde_json::from_slice(&fs::read(&old_path).at(&old_path)?)
.map_err(|error| {
Error::message(format!("invalid legacy upload journal: {error}"))
})?;
if old["context"] != old_context || !old["token"].is_string() {
return Err(Error::message(
"legacy upload journal differs from its object identity",
));
}
if path.is_file() {
let current: Value = serde_json::from_slice(&fs::read(&path).at(&path)?)
.map_err(|error| Error::message(error.to_string()))?;
if current["context"] != context || current["token"] != old["token"] {
return Err(Error::message(
"conflicting legacy and current owned upload journals",
));
}
} else {
old["context"] = context.clone();
old["verified_transfer"] = false.into();
save_upload(&path, &old)?;
}
fs::remove_file(&old_path).at(&old_path)?;
}
}
if path.is_file() {
let journal: Value = serde_json::from_slice(&fs::read(&path).at(&path)?)
.map_err(|e| Error::message(format!("invalid private storage upload journal: {e}")))?;
if journal["context"] != context || !journal["token"].is_string() {
return Err(Error::message(
"private storage upload journal does not match this object",
));
}
return Ok((path, journal));
}
let parent = path.parent().expect("journal parent");
fs::create_dir_all(parent).at(parent)?;
let nonce = tempfile::Builder::new()
.prefix("upload-")
.rand_bytes(32)
.tempfile_in(parent)
.at(parent)?;
use sha2::Sha256;
let token = crate::hex::encode_lower(Sha256::digest(
nonce
.path()
.file_name()
.expect("nonce filename")
.as_encoded_bytes(),
));
let journal = json!({"context":context,"token":token,"phase":"planned"});
save_upload(&path, &journal)?;
Ok((path, journal))
}
fn upload_context(
repo: &GitRepo,
client: &crate::native_s3::S3Client,
entry: &StorageEntry,
namespace: &str,
raw_sha256: Option<&str>,
) -> Result<(PathBuf, Value)> {
let local = crate::local_state::directory_unmigrated(repo)?;
upload_context_at(&local, client, entry, namespace, raw_sha256)
}
fn upload_context_at(
local: &Path,
client: &crate::native_s3::S3Client,
entry: &StorageEntry,
namespace: &str,
raw_sha256: Option<&str>,
) -> Result<(PathBuf, Value)> {
use sha2::Sha256;
let key = client.key_for(&entry.object);
let digest = entry
.md5
.as_deref()
.ok_or_else(|| Error::message("storage upload has no digest"))?;
let mut context = json!({"schema":1,"bucket":client.bucket,"key":key,"md5":digest,"hash_name":entry.hash_name,"size":entry.size});
if namespace == "storage-import-uploads" || raw_sha256.is_some() {
let raw_sha256 = raw_sha256
.filter(|digest| {
digest.len() == 64
&& digest
.bytes()
.all(|byte| matches!(byte, b'0'..=b'9' | b'a'..=b'f'))
})
.ok_or_else(|| {
Error::message("verified storage upload requires a raw SHA256 identity")
})?;
context["schema"] = 2.into();
context["raw_sha256"] = raw_sha256.into();
}
let identity = crate::hex::encode_lower(Sha256::digest(
serde_json::to_vec(&context).map_err(|e| Error::message(e.to_string()))?,
));
let relative = format!("{namespace}/{identity}.json");
reject_symlink_traversal(local, &relative, "private storage upload journal")?;
let path = local.join(relative);
Ok((path, context))
}
fn save_upload(path: &Path, journal: &Value) -> Result<()> {
atomic_write(
path,
&serde_json::to_vec_pretty(journal).map_err(|e| Error::message(e.to_string()))?,
)
}
fn verify_uploaded_version(
client: &crate::native_s3::S3Client,
_repo: &GitRepo,
entry: &StorageEntry,
token: &str,
version: &str,
raw_sha256: Option<&str>,
verified_transfer: bool,
) -> Result<String> {
let key = client.key_for(&entry.object);
let request = json!({"Bucket":client.bucket,"Key":key,"VersionId":version});
let info = if raw_sha256.is_some() {
client.head_with_checksums(&request)?
} else {
client.call_s3("head_object", &request, None)?
}
.value;
if version.is_empty()
|| version == "null"
|| info["DeleteMarker"] == true
|| info["VersionId"].as_str() != Some(version)
|| info["Metadata"][UPLOAD_TOKEN].as_str() != Some(token)
|| entry
.size
.is_some_and(|size| info["ContentLength"].as_u64() != Some(size))
{
return Err(Error::message(
"uploaded exact version does not match its private ownership journal",
));
}
let mut provider_verified = false;
if raw_sha256.is_some()
&& info["ChecksumType"] == "FULL_OBJECT"
&& let Some(checksum) = info["ChecksumSHA256"].as_str()
{
use base64::Engine;
let bytes = base64::engine::general_purpose::STANDARD
.decode(checksum)
.map_err(|_| Error::message("uploaded version returned a malformed provider SHA256"))?;
if bytes.len() != 32 || Some(crate::hex::encode_lower(bytes).as_str()) != raw_sha256 {
return Err(Error::message(
"uploaded version provider SHA256 conflicts with its expected raw bytes",
));
}
provider_verified = true;
}
if raw_sha256.is_some() && (verified_transfer || provider_verified) {
return info["ETag"]
.as_str()
.filter(|etag| !etag.trim_matches('"').is_empty())
.map(|etag| etag.trim_matches('"').to_owned())
.ok_or_else(|| Error::message("uploaded exact version has no ETag"));
}
let request =
json!({"Bucket":client.bucket,"Key":key,"VersionId":version,"IfMatch":info["ETag"]});
let mut attempt = 0;
let (fetched, hashes) = loop {
let mut hasher = FileHashStream::new();
match client.get_stream(&request, |bytes| {
hasher.update(bytes);
Ok(())
}) {
Ok(response) => break (response, hasher.finish()),
Err(error) if error.is_retryable() && attempt < 2 => {
attempt += 1;
}
Err(error) => return Err(error.into()),
}
};
if fetched.value["VersionId"].as_str() != Some(version)
|| fetched.value["DeleteMarker"] == true
|| fetched.value["ETag"] != info["ETag"]
|| fetched.value["ContentLength"] != info["ContentLength"]
|| entry
.size
.is_some_and(|size| fetched.value["ContentLength"].as_u64() != Some(size))
{
return Err(Error::message(
"uploaded exact-version GET response differs from its verified HEAD",
));
}
if hashes.digest(&entry.hash_name)? != entry.md5.as_deref().unwrap_or("") {
return Err(Error::message(
"uploaded exact version has a content hash mismatch; its private journal is retained for reconciliation",
));
}
if let Some(wanted) = raw_sha256
&& hashes.sha256 != wanted
{
return Err(Error::message(
"uploaded exact version has a raw content hash mismatch; its private journal is retained for reconciliation",
));
}
info["ETag"]
.as_str()
.map(|etag| etag.trim_matches('"').to_owned())
.ok_or_else(|| Error::message("uploaded exact version has no ETag"))
}
fn recover_uploaded_version(
client: &crate::native_s3::S3Client,
repo: &GitRepo,
entry: &StorageEntry,
journal: &Value,
) -> Result<Option<(String, String)>> {
let key = client.key_for(&entry.object);
let token = journal["token"]
.as_str()
.ok_or_else(|| Error::message("private upload journal has no ownership token"))?;
let mut found = Vec::new();
for row in client.list_versions(&key)? {
if row["Key"] != key || row["delete_marker"] == true {
continue;
}
let version = row["VersionId"]
.as_str()
.ok_or_else(|| Error::message("upload recovery listing has no exact ID"))?;
let info = client
.call_s3(
"head_object",
&json!({"Bucket":client.bucket,"Key":key,"VersionId":version}),
None,
)?
.value;
if info["Metadata"][UPLOAD_TOKEN].as_str() == Some(token) {
found.push(version.to_owned());
}
}
if found.len() > 1 {
return Err(Error::message(
"private upload ownership token matches multiple exact versions; refusing to duplicate or erase their history",
));
}
found
.into_iter()
.next()
.map(|version| {
verify_uploaded_version(
client,
repo,
entry,
token,
&version,
journal["context"]["raw_sha256"].as_str(),
false,
)
.map(|etag| (version, etag))
})
.transpose()
}
#[cfg(test)]
fn upload_version(
client: &crate::native_s3::S3Client,
repo: &GitRepo,
entry: &StorageEntry,
source: &Path,
single_limit: u64,
preferred_part_size: u64,
) -> Result<(String, String)> {
upload_version_in(
client,
repo,
entry,
source,
single_limit,
preferred_part_size,
UploadPolicy {
namespace: "uploads",
condition: None,
raw_sha256: None,
local: None,
hashes: None,
},
)
}
struct UploadPolicy<'a> {
namespace: &'a str,
condition: Option<&'a str>,
raw_sha256: Option<&'a str>,
local: Option<&'a Path>,
hashes: Option<&'a mut HashInventory>,
}
fn multipart_part_checksums(
source: &Path,
size: u64,
part_size: u64,
expected: &str,
) -> Result<Vec<String>> {
use sha2::Sha256;
let mut input = fs::File::open(source).at(source)?;
let mut whole = Sha256::new();
let mut parts = Vec::new();
let mut remaining = size;
let mut buffer = [0u8; 64 * 1024];
while remaining > 0 {
let length = remaining.min(part_size);
let mut part = Sha256::new();
let mut left = length;
while left > 0 {
let amount = left.min(buffer.len() as u64) as usize;
input.read_exact(&mut buffer[..amount]).at(source)?;
whole.update(&buffer[..amount]);
part.update(&buffer[..amount]);
left -= amount as u64;
}
parts.push(crate::hex::encode_lower(part.finalize()));
remaining -= length;
}
if input.read(&mut buffer[..1]).at(source)? != 0
|| crate::hex::encode_lower(whole.finalize()) != expected
{
return Err(Error::message(
"multipart source differs from its raw SHA256 identity",
));
}
Ok(parts)
}
fn upload_version_in(
client: &crate::native_s3::S3Client,
repo: &GitRepo,
entry: &StorageEntry,
source: &Path,
single_limit: u64,
preferred_part_size: u64,
policy: UploadPolicy<'_>,
) -> Result<(String, String)> {
let UploadPolicy {
namespace,
condition,
raw_sha256,
local,
hashes,
} = policy;
let (path, mut journal) = if let Some(local) = local {
upload_journal_at(local, client, entry, namespace, raw_sha256)?
} else {
upload_journal_in(repo, client, entry, namespace, raw_sha256)?
};
let key = client.key_for(&entry.object);
let token = journal["token"]
.as_str()
.ok_or_else(|| Error::message("private upload journal has no ownership token"))?
.to_owned();
if let Some(version) = journal["version_id"].as_str() {
return verify_uploaded_version(
client,
repo,
entry,
&token,
version,
raw_sha256,
journal["verified_transfer"] == true
&& journal["transfer_endpoint"] == client.endpoint_identity(),
)
.map(|etag| (version.to_owned(), etag));
}
if journal["phase"] != "planned" {
if let Some((version, etag)) = recover_uploaded_version(client, repo, entry, &journal)? {
journal["version_id"] = version.clone().into();
journal["etag"] = etag.clone().into();
journal["phase"] = "complete".into();
save_upload(&path, &journal)?;
return Ok((version, etag));
}
if journal["phase"] == "creating" {
return Err(Error::message(
"a multipart creation response was lost; reconcile the possibly outstanding upload before retrying this object",
));
}
if let Some(upload) = journal["upload_id"].as_str() {
match client.call_s3(
"abort_multipart_upload",
&json!({"Bucket":client.bucket,"Key":key,"UploadId":upload}),
None,
) {
Ok(_) => {}
Err(error) if error.code == "NoSuchUpload" => {}
Err(error) => return Err(error.into()),
}
journal
.as_object_mut()
.expect("journal object")
.remove("upload_id");
}
}
let size = fs::metadata(source).at(source)?.len();
if entry.size.is_some_and(|expected| size != expected) {
return Err(Error::message(
"cached upload size differs from storage metadata",
));
}
let mut private_hashes = HashInventory::new();
let hashes = hashes.unwrap_or(&mut private_hashes);
let source_hashes = hashes.hashes(source)?;
if source_hashes.digest(&entry.hash_name)? != entry.md5.as_deref().unwrap_or("")
|| raw_sha256.is_some_and(|expected| source_hashes.sha256 != expected)
{
return Err(Error::message(
"cached upload checksum differs from storage metadata",
));
}
let request = json!({"Bucket":client.bucket,"Key":key,"Metadata":{UPLOAD_TOKEN:token}});
let response = if size <= single_limit {
journal["phase"] = "uploading".into();
if let Some(sha256) = raw_sha256 {
journal["verified_transfer"] = true.into();
journal["transfer_endpoint"] = client.endpoint_identity().into();
journal["context"]["raw_sha256"] = sha256.into();
}
save_upload(&path, &journal)?;
let mut request = request;
if namespace == "storage-import-uploads" {
if let Some(etag) = condition {
request["IfMatch"] = etag.into();
} else {
request["IfNoneMatch"] = "*".into();
}
}
request["ExpectedMD5"] = source_hashes.md5.into();
if let Some(sha256) = raw_sha256 {
request["ExpectedSHA256"] = sha256.into();
}
request["ExpectedSize"] = size.into();
client.put_file(&request, source)
} else {
let part_size = preferred_part_size.max(size.div_ceil(10_000));
if part_size > 5 * (1 << 30) || size > 5 * (1 << 40) {
return Err(Error::message(
"stored object exceeds S3 multipart upload limits",
));
}
let part_sha256 = raw_sha256
.map(|expected| multipart_part_checksums(source, size, part_size, expected))
.transpose()?;
journal["phase"] = "creating".into();
save_upload(&path, &journal)?;
let created = client.call_s3("create_multipart_upload", &request, None)?;
let upload = created.value["UploadId"]
.as_str()
.filter(|id| !id.is_empty())
.ok_or_else(|| Error::message("multipart creation returned no upload ID"))?
.to_owned();
journal["upload_id"] = upload.clone().into();
journal["phase"] = "uploading".into();
save_upload(&path, &journal)?;
let result = (|| {
let requests = (0..size.div_ceil(part_size))
.map(|index| {
let offset = index * part_size;
(index + 1, offset, (size - offset).min(part_size))
})
.collect::<Vec<_>>();
let mut parts = crate::native_versions::bounded_map_with_workers(
&requests,
4,
true,
|&(part, offset, length)| {
let mut request = json!({"Bucket":client.bucket,"Key":key,"UploadId":upload,"PartNumber":part});
if let Some(parts) = &part_sha256 {
request["ExpectedSHA256"] = parts[(part - 1) as usize].clone().into();
}
let response = client.upload_part_file(&request, source, offset, length)?;
let etag = response.value["ETag"]
.as_str()
.ok_or_else(|| Error::message("multipart part returned no ETag"))?;
Ok(json!({"PartNumber":part,"ETag":etag}))
},
)?;
parts.sort_by_key(|part| part["PartNumber"].as_u64().unwrap_or_default());
if hashes.hashes(source)?.sha256 != source_hashes.sha256 {
return Err(Error::message(
"cached content changed during multipart upload",
));
}
journal["phase"] = "completing".into();
if raw_sha256.is_some() {
journal["verified_transfer"] = true.into();
journal["transfer_endpoint"] = client.endpoint_identity().into();
}
save_upload(&path, &journal)?;
let mut complete = json!({"Bucket":client.bucket,"Key":key,"UploadId":upload,"MultipartUpload":{"Parts":parts}});
if namespace == "storage-import-uploads" {
if let Some(etag) = condition {
complete["IfMatch"] = etag.into();
} else {
complete["IfNoneMatch"] = "*".into();
}
}
client
.call_s3("complete_multipart_upload", &complete, None)
.map_err(Error::from)
})();
match result {
Ok(response) => Ok(response),
Err(error) => {
if let Some((version, etag)) =
recover_uploaded_version(client, repo, entry, &journal)?
{
journal["version_id"] = version.clone().into();
journal["etag"] = etag.clone().into();
journal["phase"] = "complete".into();
save_upload(&path, &journal)?;
return Ok((version, etag));
}
match client.call_s3(
"abort_multipart_upload",
&json!({"Bucket":client.bucket,"Key":key,"UploadId":upload}),
None,
) {
Ok(_) => {}
Err(abort) if abort.code == "NoSuchUpload" => {}
Err(abort) => {
return Err(Error::message(format!(
"multipart upload failed ({error}); owned upload could not be aborted ({abort})"
)));
}
}
journal
.as_object_mut()
.expect("journal object")
.remove("upload_id");
journal["phase"] = "planned".into();
save_upload(&path, &journal)?;
return Err(error);
}
}
};
let response = match response {
Ok(response) => response,
Err(error) => {
if let Some((version, etag)) = recover_uploaded_version(client, repo, entry, &journal)?
{
journal["version_id"] = version.clone().into();
journal["etag"] = etag.clone().into();
journal["phase"] = "complete".into();
save_upload(&path, &journal)?;
return Ok((version, etag));
}
return Err(error.into());
}
};
let version = response.value["VersionId"].as_str().filter(|id|!id.is_empty() && *id != "null").ok_or_else(||Error::message("S3 upload returned no exact version ID; the private upload journal will recover it on retry"))?.to_owned();
journal["version_id"] = version.clone().into();
save_upload(&path, &journal)?;
let etag = verify_uploaded_version(
client,
repo,
entry,
&token,
&version,
raw_sha256,
journal["verified_transfer"] == true
&& journal["transfer_endpoint"] == client.endpoint_identity(),
)?;
journal["etag"] = etag.clone().into();
journal["phase"] = "complete".into();
save_upload(&path, &journal)?;
Ok((version, etag))
}
fn fetch(repo: &GitRepo, pointers: &[String]) -> Result<()> {
let remote = remote_root(repo)?;
if remote.starts_with("s3://") {
return Err(Error::message(
"version-aware S3 fetching must use the exact-version adapter",
));
}
let cache = CachePaths::new(repo)?;
let mut hashes = HashInventory::new();
let documents = storage_metadata::read_pointer_documents(repo, None, pointers)?;
for pointer in pointers {
let (document, algorithm) = &documents[pointer];
for out in &document.outs {
let algorithm = algorithm.as_str();
let digest = out
.md5
.as_deref()
.ok_or_else(|| Error::message("metadata has no content hash"))?;
if pointer.ends_with(".dvc") && digest.ends_with(".dir") {
install_cache_file_with_inventory(
&cache,
&mut hashes,
digest,
&remote_cache_path(&remote, digest, repo, algorithm)?,
algorithm,
)?;
}
for file in recorded_files_with_cache(repo, &cache, pointer, out, algorithm)? {
install_cache_file_with_inventory(
&cache,
&mut hashes,
&file.md5,
&remote_cache_path(&remote, &file.md5, repo, algorithm)?,
algorithm,
)?;
}
}
}
Ok(())
}
fn install_cache_file_with_inventory(
cache: &CachePaths<'_>,
hashes: &mut HashInventory,
digest: &str,
source: &Path,
algorithm: &str,
) -> Result<()> {
let wanted = digest.trim_end_matches(".dir");
if hashes.digest(source, algorithm)? != wanted {
return Err(Error::message("downloaded content hash mismatch"));
}
let destination = cache.path(digest, algorithm)?;
atomic_copy(source, &destination)?;
if hashes.digest(&destination, algorithm)? != wanted {
return Err(Error::message(
"downloaded content changed while installing its cache",
));
}
Ok(())
}
fn checkout(repo: &GitRepo, pointers: &[String]) -> Result<()> {
let mut staged = Vec::new();
let cache = CachePaths::new(repo)?;
let mut hashes = HashInventory::new();
let documents = storage_metadata::read_pointer_documents(repo, None, pointers)?;
for pointer in pointers {
let (document, algorithm) = &documents[pointer];
let mut proof_paths = Vec::new();
for entry in entries_from_document(pointer, document, algorithm)?
.into_iter()
.filter(|entry| entry.verification.is_some())
{
let local = repo.root.join(&entry.object);
if local.is_file() {
proof_paths.push(local);
}
let cached = cache.entry(&entry)?;
if cached.is_file() {
proof_paths.push(cached);
}
}
hashes.preload(&proof_paths, "md5-dos2unix")?;
for out in &document.outs {
let hash_name = algorithm.as_str();
let object = output_object(pointer, &out.path)?;
reject_symlink_traversal(&repo.root, &object, "storage checkout")?;
let root = repo.root.join(&object);
let current = if root.exists() {
current_files_with_inventory(repo, &object, hash_name, &mut hashes)?
} else {
Vec::new()
};
let directory = out.md5.as_deref().is_some_and(|m| m.ends_with(".dir"));
let files = recorded_files_with_cache(repo, &cache, pointer, out, hash_name)?;
let mut cached_size = 0u64;
for file in &files {
let source = cache_for_recorded_file(&cache, &object, file, hash_name)?;
let size = fs::metadata(&source).at(&source)?.len();
cached_size = cached_size
.checked_add(size)
.ok_or_else(|| Error::message("cached content size overflows"))?;
if (file.size != 0 && size != file.size)
|| hashes.digest(&source, hash_name)? != file.md5
|| file.verification.as_ref().is_some_and(|proof| {
!hashes
.hashes(&source)
.is_ok_and(|value| value.sha256 == proof.checksum.digest)
})
{
return Err(Error::message("cached content size or hash mismatch"));
}
}
if out.size.is_some_and(|size| cached_size != size) {
return Err(Error::message("cached output size differs from metadata"));
}
if root.exists() && root.is_dir() == directory {
let digest = if directory {
actual_directory_digest(pointer, ¤t, hash_name)?
} else {
current
.first()
.map(|file| file.md5.clone())
.unwrap_or_default()
};
if out.md5.as_deref() == Some(&digest)
&& out.size.is_none_or(|size| {
current.iter().map(|file| file.size).sum::<u64>() == size
})
&& normalized_exact_bytes_match(
repo,
&cache,
&mut hashes,
&object,
&files,
hash_name,
)?
{
continue;
}
}
if staged
.iter()
.any(|(previous, _, _): &(PathBuf, tempfile::TempDir, PathBuf)| {
root.starts_with(previous) || previous.starts_with(&root)
})
{
return Err(Error::message(
"storage checkout contains overlapping outputs",
));
}
let parent = root
.parent()
.ok_or_else(|| Error::message("storage output has no parent"))?;
fs::create_dir_all(parent).at(parent)?;
let temporary = tempfile::Builder::new()
.prefix(".workspace-mgr-checkout-")
.tempdir_in(parent)
.at(parent)?;
let replacement = temporary.path().join("output");
if directory {
fs::create_dir(&replacement).at(&replacement)?;
}
for file in files {
let source = cache_for_recorded_file(&cache, &object, &file, hash_name)?;
let existing = if file.relpath.is_empty() {
root.clone()
} else {
root.join(&file.relpath)
};
let destination = if file.relpath.is_empty() {
replacement.clone()
} else {
replacement.join(&file.relpath)
};
let relative =
to_slash(existing.strip_prefix(&repo.root).expect("validated output"));
let relative_parent = Path::new(&relative)
.parent()
.map(to_slash)
.unwrap_or_default();
reject_symlink_traversal(&repo.root, &relative_parent, "storage checkout")?;
let source_size = fs::metadata(&source).at(&source)?.len();
if (file.size != 0 && source_size != file.size)
|| hashes.digest(&source, hash_name)? != file.md5
|| file.verification.as_ref().is_some_and(|proof| {
!hashes
.hashes(&source)
.is_ok_and(|value| value.sha256 == proof.checksum.digest)
})
{
return Err(Error::message("cached content size or hash mismatch"));
}
atomic_copy(&source, &destination)?;
if fs::metadata(&destination).at(&destination)?.len() != source_size
|| hashes.digest(&destination, hash_name)? != file.md5
|| file.verification.as_ref().is_some_and(|proof| {
!hashes
.hashes(&destination)
.is_ok_and(|value| value.sha256 == proof.checksum.digest)
})
{
return Err(Error::message("cached content changed during checkout"));
}
if let Ok(metadata) = fs::metadata(&existing)
&& metadata.is_file()
{
fs::set_permissions(&destination, metadata.permissions()).at(&destination)?;
}
}
staged.push((root, temporary, replacement));
}
}
let mut installed: Vec<usize> = Vec::new();
for index in 0..staged.len() {
let (root, temporary, replacement) = &staged[index];
let backup = temporary.path().join("previous");
let result = (|| {
if root.exists() {
fs::rename(root, &backup).at(root)?;
}
if let Err(error) = fs::rename(replacement, root).at(root) {
if backup.exists() {
fs::rename(&backup, root).at(root)?;
}
return Err(error);
}
installed.push(index);
fs::File::open(root.parent().expect("validated parent"))
.at(root)?
.sync_all()
.at(root)
})();
if let Err(error) = result {
for previous in installed.into_iter().rev() {
let (root, temporary, _) = &staged[previous];
if root.is_dir() {
fs::remove_dir_all(root).at(root)?;
} else {
fs::remove_file(root).at(root)?;
}
let backup = temporary.path().join("previous");
if backup.exists() {
fs::rename(&backup, root).at(root)?;
}
}
return Err(error);
}
}
Ok(())
}
fn status(repo: &GitRepo, pointers: &[String], cloud: bool, quiet: bool) -> Result<(i32, String)> {
let mut result = serde_json::Map::new();
let cache = CachePaths::new(repo)?;
let documents = storage_metadata::read_pointer_documents(repo, None, pointers)?;
let remote = cloud.then(|| remote_root(repo)).transpose()?;
let cpus = std::thread::available_parallelism()
.map(usize::from)
.unwrap_or(1);
let workers = cpus.min(pointers.len()).max(1);
let checked =
crate::native_versions::bounded_map_with_workers(pointers, workers, false, |pointer| {
let mut hashes = HashInventory::with_workers(cpus / workers);
let mut changed = serde_json::Map::new();
let (document, algorithm) = &documents[pointer];
let proof_paths = entries_from_document(pointer, document, algorithm)?
.into_iter()
.filter(|entry| entry.verification.is_some())
.map(|entry| repo.root.join(entry.object))
.filter(|path| path.is_file())
.collect::<Vec<_>>();
hashes.preload(&proof_paths, "md5-dos2unix")?;
for out in &document.outs {
let algorithm = algorithm.as_str();
let object = output_object(pointer, &out.path)?;
let digest = out
.md5
.as_deref()
.ok_or_else(|| Error::message("metadata has no digest"))?;
let issue = if cloud {
let remote = remote.as_deref().expect("cloud remote");
if remote.starts_with("s3://") {
return Err(Error::message(
"version-aware remote status requires the exact-version adapter",
));
}
if algorithms_for_document(repo, &cache, pointer, document, algorithm)
.is_ok_and(|hashes| {
hashes.iter().all(|(hash, algorithm)| {
remote_cache_path(remote, hash, repo, algorithm).is_ok_and(|path| {
path.is_file()
&& file_digest(&path, algorithm)
.is_ok_and(|d| d == hash.trim_end_matches(".dir"))
})
})
})
{
None
} else {
Some("not in remote")
}
} else if !repo.root.join(&object).exists() {
Some("deleted")
} else if repo.root.join(&object).is_dir() != digest.ends_with(".dir") {
Some("modified")
} else {
let current =
current_files_with_inventory(repo, &object, algorithm, &mut hashes)?;
let actual = if digest.ends_with(".dir") {
actual_directory_digest(pointer, ¤t, algorithm)?
} else {
current
.first()
.map(|file| file.md5.clone())
.unwrap_or_default()
};
let recorded = if actual == digest {
recorded_files_with_cache(repo, &cache, pointer, out, algorithm)?
} else {
Vec::new()
};
let cached_paths = recorded
.iter()
.filter_map(|file| {
cache_for_recorded_file(&cache, &object, file, algorithm).ok()
})
.filter(|path| path.is_file())
.collect::<Vec<_>>();
let _ = hashes.preload(&cached_paths, algorithm);
if actual != digest
|| !normalized_exact_bytes_match(
repo,
&cache,
&mut hashes,
&object,
&recorded,
algorithm,
)?
{
Some("modified")
} else if !recorded.iter().all(|file| {
cache_for_recorded_file(&cache, &object, file, algorithm).is_ok_and(
|path| {
path.is_file()
&& hashes.digest(&path, algorithm).is_ok_and(|d| d == file.md5)
&& file.verification.as_ref().is_none_or(|proof| {
hashes.hashes(&path).is_ok_and(|value| {
value.sha256 == proof.checksum.digest
})
})
},
)
}) {
Some("not in cache")
} else {
None
}
};
if let Some(issue) = issue {
changed.insert(object, json!(issue));
}
}
Ok((pointer.clone(), changed))
})?;
for (pointer, changed) in checked {
if !changed.is_empty() {
result.insert(pointer, json!([{"changed outs":changed}]));
}
}
let code = if quiet && !result.is_empty() { 1 } else { 0 };
Ok((
code,
serde_json::to_string(&result).map_err(|e| Error::message(e.to_string()))?,
))
}
fn data_status(repo: &GitRepo, targets: &[String]) -> Result<Value> {
let pointers = select_pointers(repo, targets)?;
let documents = storage_metadata::read_pointer_documents(repo, None, &pointers)?;
let cache = CachePaths::new(repo)?;
let cpus = std::thread::available_parallelism()
.map(usize::from)
.unwrap_or(1);
let workers = cpus.min(pointers.len()).max(1);
let checked =
crate::native_versions::bounded_map_with_workers(&pointers, workers, false, |pointer| {
let mut hashes = HashInventory::with_workers(cpus / workers);
let mut added = BTreeSet::new();
let mut modified = BTreeSet::new();
let mut deleted = BTreeSet::new();
let mut not_in_cache = BTreeSet::new();
let mut unknown = BTreeSet::new();
let (document, algorithm) = &documents[pointer];
for out in &document.outs {
let algorithm = algorithm.as_str();
let object = output_object(pointer, &out.path)?;
let directory = out.md5.as_deref().is_some_and(|m| m.ends_with(".dir"));
let label = if directory {
format!("{object}/")
} else {
object.clone()
};
let files = match recorded_files_with_cache(repo, &cache, pointer, out, algorithm) {
Ok(files) => files,
Err(Error::Io { source, .. })
if source.kind() == std::io::ErrorKind::NotFound && directory =>
{
not_in_cache.insert(label.clone());
unknown.insert(label);
continue;
}
Err(error) => return Err(error),
};
let mut proof_paths = Vec::new();
for file in files.iter().filter(|file| file.verification.is_some()) {
let local = if file.relpath.is_empty() {
repo.root.join(&object)
} else {
repo.root.join(&object).join(&file.relpath)
};
if local.is_file() {
proof_paths.push(local);
}
}
hashes.preload(&proof_paths, "md5-dos2unix")?;
let current = current_files_with_inventory(repo, &object, algorithm, &mut hashes)?;
let old = files
.iter()
.map(|f| (f.relpath.clone(), (f.md5.clone(), f.size)))
.collect::<BTreeMap<_, _>>();
let new = current
.iter()
.map(|f| (f.relpath.clone(), (f.md5.clone(), f.size)))
.collect::<BTreeMap<_, _>>();
if old != new || !repo.root.join(&object).exists() {
if !repo.root.join(&object).exists() {
deleted.insert(label.clone());
} else if directory {
modified.insert(label.clone());
}
}
if pointer.ends_with(".dvc")
&& let Some(digest) = &out.md5
&& !cache.existing(digest, algorithm)?.is_file()
{
not_in_cache.insert(label.clone());
}
for file in &files {
let label = if file.relpath.is_empty() {
object.clone()
} else {
format!("{object}/{}", file.relpath)
};
let cached = if file.verification.is_some() {
cache_for_recorded_file(&cache, &object, file, algorithm)?
} else {
cache.existing(&file.md5, algorithm)?
};
let cached_matches = if let Some(proof) = &file.verification {
cached.is_file()
&& fs::metadata(&cached)
.is_ok_and(|metadata| metadata.len() == file.size)
&& hashes.hashes(&cached).is_ok_and(|actual| {
actual.sha256 == proof.checksum.digest
&& actual
.digest(algorithm)
.is_ok_and(|digest| digest == file.md5)
})
} else {
cached.is_file()
};
if !cached_matches {
not_in_cache.insert(label.clone());
}
match new.get(&file.relpath) {
None => {
deleted.insert(label);
}
Some(state) if state != &(file.md5.clone(), file.size) => {
modified.insert(label);
}
Some(_) => {
if let Some(proof) = &file.verification
&& hashes.hashes(&repo.root.join(&label))?.sha256
!= proof.checksum.digest
{
if directory {
modified.insert(format!("{object}/"));
}
modified.insert(label);
}
}
}
}
for file in ¤t {
if !old.contains_key(&file.relpath) {
added.insert(if file.relpath.is_empty() {
object.clone()
} else {
format!("{object}/{}", file.relpath)
});
}
}
}
Ok((added, modified, deleted, not_in_cache, unknown))
})?;
let mut added = BTreeSet::new();
let mut modified = BTreeSet::new();
let mut deleted = BTreeSet::new();
let mut not_in_cache = BTreeSet::new();
let mut unknown = BTreeSet::new();
for (a, m, d, n, u) in checked {
added.extend(a);
modified.extend(m);
deleted.extend(d);
not_in_cache.extend(n);
unknown.extend(u);
}
let mut changes = serde_json::Map::new();
for (key, set) in [
("added", added),
("modified", modified),
("deleted", deleted),
] {
if !set.is_empty() {
changes.insert(key.to_owned(), json!(set));
}
}
let mut value = serde_json::Map::new();
if !not_in_cache.is_empty() {
value.insert("not_in_cache".to_owned(), json!(not_in_cache));
}
if !changes.is_empty() {
value.insert("uncommitted".to_owned(), Value::Object(changes));
}
if !unknown.is_empty() {
value.insert("unknown".to_owned(), json!(unknown));
}
Ok(Value::Object(value))
}
fn move_output(repo: &GitRepo, source: &str, destination: &str) -> Result<()> {
let source = repo_path(source, "storage move source")?;
let destination = repo_path(destination, "storage move destination")?;
storage_metadata::require_addressable(
&destination,
"storage move destination",
"choose a path without backslashes",
)?;
let old_pointer = storage_metadata::pointer_path(&source);
let new_pointer = storage_metadata::pointer_path(&destination);
for path in [&source, &destination, &old_pointer, &new_pointer] {
reject_symlink_traversal(&repo.root, path, "storage move")?;
}
if repo.root.join(&destination).exists() || repo.root.join(&new_pointer).exists() {
return Err(Error::message("storage move destination already exists"));
}
let mut document = read_manifest(repo, &old_pointer)?;
document.clear_versions();
document.path = Path::new(&destination)
.file_name()
.and_then(|name| name.to_str())
.ok_or_else(|| Error::message("invalid output name"))?
.to_owned();
fs::create_dir_all(
repo.root
.join(&destination)
.parent()
.expect("destination parent"),
)
.at(repo.root.join(&destination))?;
let materialized = repo.root.join(&source).exists();
if materialized {
fs::rename(repo.root.join(&source), repo.root.join(&destination))
.at(repo.root.join(&source))?;
}
if let Err(error) = write_manifest(repo, &new_pointer, &document) {
if materialized {
let _ = fs::rename(repo.root.join(&destination), repo.root.join(&source));
}
return Err(error);
}
fs::remove_file(repo.root.join(&old_pointer)).at(repo.root.join(&old_pointer))?;
update_ignore(
&repo
.root
.join(&source)
.parent()
.expect("output parent")
.join(".gitignore"),
Path::new(&source)
.file_name()
.and_then(|s| s.to_str())
.expect("UTF-8"),
false,
)?;
update_ignore(
&repo
.root
.join(&destination)
.parent()
.expect("output parent")
.join(".gitignore"),
Path::new(&destination)
.file_name()
.and_then(|s| s.to_str())
.expect("UTF-8"),
true,
)
}
fn remove(repo: &GitRepo, pointer: &str) -> Result<()> {
let document = storage_metadata::read_pointer_document(repo, pointer)?;
for out in document.outs {
let object = output_object(pointer, &out.path)?;
update_ignore(
&repo
.root
.join(&object)
.parent()
.expect("output parent")
.join(".gitignore"),
Path::new(&object)
.file_name()
.and_then(|s| s.to_str())
.ok_or_else(|| Error::message("invalid output name"))?,
false,
)?;
}
fs::remove_file(repo.root.join(pointer)).at(repo.root.join(pointer))
}
fn ignore_rule(name: &str) -> String {
let mut escaped = String::from("/");
for ch in name.chars() {
if matches!(ch, '\\' | '*' | '?' | '[' | ']' | '!' | '#') {
escaped.push('\\');
}
escaped.push(ch);
}
escaped
}
fn update_ignore(path: &Path, name: &str, add: bool) -> Result<()> {
if fs::symlink_metadata(path).is_ok_and(|m| m.file_type().is_symlink()) {
return Err(Error::message("storage ignore file may not be a symlink"));
}
let raw = match fs::read_to_string(path) {
Ok(raw) => raw,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => String::new(),
Err(source) => {
return Err(Error::Io {
path: path.to_owned(),
source,
});
}
};
let rule = ignore_rule(name);
let mut lines = raw
.split_inclusive('\n')
.filter(|line| line.trim_end_matches(['\r', '\n']) != rule || add)
.map(ToOwned::to_owned)
.collect::<Vec<_>>();
if add && !raw.lines().any(|line| line == rule) {
if !raw.is_empty() && !raw.ends_with('\n') {
lines.push("\n".to_owned());
}
lines.push(format!("{rule}\n"));
}
let next = lines.concat();
if next != raw {
atomic_write(path, next.as_bytes())?;
}
Ok(())
}
fn atomic_write(path: &Path, bytes: &[u8]) -> Result<()> {
let parent = path
.parent()
.ok_or_else(|| Error::message("storage path has no parent"))?;
fs::create_dir_all(parent).at(parent)?;
let mut file = tempfile::NamedTempFile::new_in(parent).at(parent)?;
file.write_all(bytes).at(path)?;
if let Ok(metadata) = fs::metadata(path) {
file.as_file()
.set_permissions(metadata.permissions())
.at(path)?;
}
file.as_file().sync_all().at(path)?;
file.persist(path).map_err(|e| Error::Io {
path: path.to_owned(),
source: e.error,
})?;
fs::File::open(parent).at(parent)?.sync_all().at(parent)
}
fn atomic_copy(source: &Path, destination: &Path) -> Result<()> {
let parent = destination
.parent()
.ok_or_else(|| Error::message("storage object has no parent"))?;
fs::create_dir_all(parent).at(parent)?;
let mut file = tempfile::NamedTempFile::new_in(parent).at(parent)?;
let mut input = fs::File::open(source).at(source)?;
std::io::copy(&mut input, &mut file).at(destination)?;
let metadata = fs::metadata(destination)
.or_else(|_| fs::metadata(source))
.at(destination)?;
file.as_file()
.set_permissions(metadata.permissions())
.at(destination)?;
file.as_file().sync_all().at(destination)?;
file.persist(destination).map_err(|e| Error::Io {
path: destination.to_owned(),
source: e.error,
})?;
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
fn repository() -> (tempfile::TempDir, GitRepo) {
let temporary = tempfile::tempdir().unwrap();
let repo = GitRepo {
root: temporary.path().canonicalize().unwrap(),
};
repo.run(["init", "-q"]).unwrap();
initialize(&repo).unwrap();
(temporary, repo)
}
fn remote(repo: &GitRepo, path: &Path) {
fs::write(
repo.root.join(".workspace-mgr.toml"),
format!(
"[git]\nremote = \"origin\"\nbranch = \"main\"\n[s3]\nurl = {}\n",
serde_json::to_string(&path.to_string_lossy()).unwrap()
),
)
.unwrap();
}
#[test]
fn schema_two_data_status_checks_raw_content_and_verified_cache() {
for directory in [false, true] {
let (_temporary, repo) = repository();
let object = if directory { "data/a" } else { "data" };
let source = repo.root.join(object);
fs::create_dir_all(source.parent().unwrap()).unwrap();
fs::write(&source, b"alpha\r\nbeta\n").unwrap();
let mut hashes = HashInventory::new();
let actual = hashes.hashes(&source).unwrap();
let checksum = Checksum {
algorithm: "md5-dos2unix".into(),
digest: actual.normalized_md5,
};
let proof = Verification {
endpoint: "https://s3.example.invalid".into(),
bucket: "bucket".into(),
key: format!("prefix/{object}"),
version_id: "exact-version".into(),
checksum: Checksum {
algorithm: "sha256".into(),
digest: actual.sha256,
},
size: 12,
method: "verified-read".into(),
};
let version = Version {
id: "exact-version".into(),
etag: Some("exact-etag".into()),
verification: Some(proof),
};
let entries = directory.then(|| {
vec![Entry {
path: "a".into(),
checksum: checksum.clone(),
size: 12,
version: Some(version.clone()),
}]
});
let manifest = Manifest {
schema_version: crate::storage_format::SCHEMA_VERSION,
path: "data".into(),
kind: if directory {
Kind::Directory
} else {
Kind::File
},
checksum: Checksum {
algorithm: checksum.algorithm.clone(),
digest: match &entries {
Some(entries) => crate::storage_format::directory_digest(entries).unwrap(),
None => checksum.digest,
},
},
size: 12,
version: (!directory).then_some(version),
entries,
};
let pointer = "data.wm-storage.json";
write_manifest(&repo, pointer, &manifest).unwrap();
let entry = metadata_entries(&repo, None, &[pointer.into()])
.unwrap()
.remove(0);
let cache = CachePaths::new(&repo).unwrap();
cache
.install_entry_with_inventory(&entry, &source, &mut hashes)
.unwrap();
let verified_cache = cache.entry(&entry).unwrap();
assert_eq!(data_status(&repo, &["data".into()]).unwrap(), json!({}));
commit(&repo, pointer).unwrap();
assert_eq!(read_manifest(&repo, pointer).unwrap(), manifest);
assert_eq!(fs::read(&verified_cache).unwrap(), b"alpha\r\nbeta\n");
fs::write(&verified_cache, b"alpha\nbeta\r\n").unwrap();
let corrupt_cache = data_status(&repo, &["data".into()]).unwrap();
assert_eq!(corrupt_cache["not_in_cache"], json!([object]));
assert!(corrupt_cache["uncommitted"].is_null());
fs::write(&verified_cache, b"alpha\r\nbeta\n").unwrap();
fs::write(&source, b"alpha\nbeta\r\n").unwrap();
let local_change = data_status(&repo, &["data".into()]).unwrap();
assert_eq!(
local_change["uncommitted"]["modified"],
if directory {
json!(["data/", "data/a"])
} else {
json!(["data"])
}
);
assert!(local_change["not_in_cache"].is_null());
commit(&repo, pointer).unwrap();
let changed = read_manifest(&repo, pointer).unwrap();
assert!(changed.version.is_none());
assert!(
changed
.entries
.as_deref()
.unwrap_or_default()
.iter()
.all(|entry| entry.version.is_none())
);
assert_eq!(changed.checksum.algorithm, "md5");
assert_eq!(fs::read(&verified_cache).unwrap(), b"alpha\r\nbeta\n");
}
}
#[test]
fn optimistic_version_binding_rejects_changed_verification_evidence() {
let checksum = Checksum {
algorithm: "md5".into(),
digest: "0cc175b9c0f1b6a831c399e269772661".into(),
};
let proof = Verification {
endpoint: "https://s3.example.invalid".into(),
bucket: "bucket".into(),
key: "prefix/data".into(),
version_id: "exact-version".into(),
checksum: Checksum {
algorithm: "sha256".into(),
digest: "ca978112ca1bbdcafac231b39a23dc4da786eff8147c4e72b9807785afee48bb".into(),
},
size: 1,
method: "verified-read".into(),
};
let mut version = Version {
id: "exact-version".into(),
etag: Some("exact-etag".into()),
verification: Some(proof.clone()),
};
let captured = StorageEntry {
pointer: "data.wm-storage.json".into(),
object: "data".into(),
md5: Some(checksum.digest.clone()),
size: Some(1),
version_id: Some(version.id.clone()),
etag: version.etag.clone(),
verification: Some(proof),
hash_name: "md5".into(),
};
assert!(binding_is_unchanged(
&checksum,
1,
Some(&version),
&captured
));
version.verification.as_mut().unwrap().bucket = "other-bucket".into();
assert!(!binding_is_unchanged(
&checksum,
1,
Some(&version),
&captured
));
version.verification = None;
assert!(!binding_is_unchanged(
&checksum,
1,
Some(&version),
&captured
));
}
#[test]
fn dos2unix_hash_matches_fixed_dvc_oracles_across_short_reads_and_binary_chunks() {
let (_temporary, repo) = repository();
let path = repo.root.join("legacy.txt");
let boundary = [vec![b'a'; 1024 * 1024 - 1], b"\r\n".to_vec()].concat();
let over_threshold = [vec![0x80; 154], vec![b'a'; 358], b"\r\n".to_vec()].concat();
let below_threshold = [vec![0x80; 153], vec![b'a'; 359], b"\r\n".to_vec()].concat();
let later_binary = [vec![b'a'; 1024 * 1024], b"b\0\r\n".to_vec()].concat();
let later_text = [
b"a\0\r\n".to_vec(),
vec![0; 1024 * 1024 - 4],
b"x\r\n".to_vec(),
]
.concat();
let within_chunk = [vec![b'a'; 128 * 1024 - 1], b"\r\nb\rc\r\n\r".to_vec()].concat();
for (name, bytes, expected) in [
(
"NUL binary",
b"a\0\r\n".to_vec(),
"a200e344e12b35719025ffdb8b428ee8",
),
(
"1-MiB CRLF boundary",
boundary,
"2647acae3fed7a4538ecf474e9a98d1e",
),
(
"above 30-percent nontext",
over_threshold,
"2cf181203e54df156ffcb988c5897222",
),
(
"below 30-percent nontext",
below_threshold,
"035aaabce8d6cdfb644cf2dbf29cfb8b",
),
(
"text",
b"a\r\nb\r\n".to_vec(),
"dd8c6a395b5dd36c56d23275028f526c",
),
(
"binary later chunk",
later_binary,
"796ce23f04b52ede614636806c2a3b07",
),
(
"text later chunk",
later_text,
"842b31a117d59ef48c1ce05174fbbbc4",
),
(
"CRLF within logical chunk",
within_chunk,
"8cbb84bd2527ebbd2ca5ea12f049afaf",
),
] {
struct ShortReads(std::io::Cursor<Vec<u8>>);
impl Read for ShortReads {
fn read(&mut self, buffer: &mut [u8]) -> std::io::Result<usize> {
let maximum = buffer.len().min(257);
self.0.read(&mut buffer[..maximum])
}
}
fs::write(&path, &bytes).unwrap();
assert_eq!(
file_digest(&path, "md5-dos2unix").unwrap(),
expected,
"{name}"
);
assert_eq!(
stream_digest(
&mut ShortReads(std::io::Cursor::new(bytes.clone())),
&path,
"md5-dos2unix"
)
.unwrap(),
expected,
"short reads: {name}"
);
assert_eq!(
file_digest(&path, "md5").unwrap(),
crate::hex::encode_lower(Md5::digest(&bytes)),
"raw MD5: {name}"
);
let combined =
stream_hashes(&mut ShortReads(std::io::Cursor::new(bytes.clone())), &path).unwrap();
assert_eq!(
combined.normalized_md5, expected,
"combined short reads: {name}"
);
assert_eq!(combined.md5, crate::hex::encode_lower(Md5::digest(&bytes)));
assert_eq!(
combined.sha256,
crate::hex::encode_lower(sha2::Sha256::digest(&bytes))
);
assert_eq!(fs::read(&path).unwrap(), bytes, "raw bytes: {name}");
}
}
#[cfg(unix)]
#[test]
fn hash_inventory_rechecks_same_size_edits_even_when_mtime_is_restored() {
let directory = tempfile::tempdir().unwrap();
let path = directory.path().join("payload");
fs::write(&path, b"first").unwrap();
let modified = fs::metadata(&path).unwrap().modified().unwrap();
let mut hashes = HashInventory::new();
let original = hashes.hashes(&path).unwrap();
fs::write(&path, b"other").unwrap();
fs::File::open(&path)
.unwrap()
.set_times(fs::FileTimes::new().set_modified(modified))
.unwrap();
let changed = hashes.hashes(&path).unwrap();
assert_ne!(changed.md5, original.md5);
assert_eq!(changed.md5, "795f3202b17cb6bc3d4b771d8c6c9eaf");
let replacement = directory.path().join("replacement");
fs::write(&replacement, b"third").unwrap();
fs::File::open(&replacement)
.unwrap()
.set_times(fs::FileTimes::new().set_modified(modified))
.unwrap();
fs::rename(&replacement, &path).unwrap();
assert_ne!(hashes.hashes(&path).unwrap().md5, changed.md5);
}
#[test]
fn parallel_file_inventory_matches_serial_checks_and_rechecks_later_edits() {
let directory = tempfile::tempdir().unwrap();
let paths = (0..96)
.map(|index| {
let path = directory.path().join(index.to_string());
fs::write(&path, format!("payload {index}\r\n")).unwrap();
path
})
.collect::<Vec<_>>();
let mut parallel = HashInventory::with_workers(4);
let mut serial = HashInventory::with_workers(1);
parallel.preload(&paths, "md5-dos2unix").unwrap();
serial.preload(&paths, "md5-dos2unix").unwrap();
for path in &paths {
assert_eq!(parallel.hashes(path).unwrap(), serial.hashes(path).unwrap());
}
fs::write(&paths[95], "changed 95\r\n").unwrap();
parallel.preload(&paths, "md5-dos2unix").unwrap();
assert_eq!(
parallel.hashes(&paths[95]).unwrap().normalized_md5,
file_digest(&paths[95], "md5-dos2unix").unwrap()
);
assert_eq!(parallel.files.len(), paths.len());
}
#[cfg(unix)]
#[test]
fn command_cache_resolution_still_refuses_replaced_private_parent_links() {
let (_temporary, repo) = repository();
let cache = CachePaths::new(&repo).unwrap();
let local = cache.local().to_owned();
fs::rename(&local, local.with_extension("previous")).unwrap();
let foreign = tempfile::tempdir().unwrap();
std::os::unix::fs::symlink(foreign.path(), &local).unwrap();
assert!(
cache
.path("0cc175b9c0f1b6a831c399e269772661", "md5")
.is_err()
);
assert!(!foreign.path().join("cache").exists());
}
#[test]
fn command_exact_cache_resolution_rejects_mid_command_routing_changes() {
let (_temporary, repo) = repository();
remote(&repo, Path::new("first-remote"));
let cache = CachePaths::new(&repo).unwrap();
let entry = StorageEntry {
verification: None,
pointer: "source.wm-storage.json".into(),
object: "source".into(),
md5: Some("0cc175b9c0f1b6a831c399e269772661".into()),
size: Some(1),
version_id: Some("exact-version".into()),
etag: None,
hash_name: "md5-dos2unix".into(),
};
cache.entry(&entry).unwrap();
remote(&repo, Path::new("other-remote"));
assert!(
cache
.entry(&entry)
.unwrap_err()
.to_string()
.contains("routing changed")
);
}
#[test]
fn native_add_reports_granular_changes_and_preserves_remote_versions() {
let (_temporary, repo) = repository();
fs::create_dir(repo.root.join("data")).unwrap();
fs::write(repo.root.join("data/a"), "a").unwrap();
fs::write(repo.root.join("data/z"), "z").unwrap();
add(&repo, "data").unwrap();
let mut document = read_manifest(&repo, "data.wm-storage.json").unwrap();
document.schema_version = 1;
assert_eq!(document.checksum.algorithm, "md5");
for (entry, id) in document
.entries
.as_mut()
.unwrap()
.iter_mut()
.zip(["old-a", "old-z"])
{
entry.version = Some(Version {
verification: None,
id: id.into(),
etag: Some(format!("etag-{id}")),
});
}
write_manifest(&repo, "data.wm-storage.json", &document).unwrap();
assert_eq!(data_status(&repo, &["data".into()]).unwrap(), json!({}));
fs::write(repo.root.join("data/z"), "new z").unwrap();
fs::write(repo.root.join("data/new"), "new").unwrap();
let changes = data_status(&repo, &["data".into()]).unwrap();
assert_eq!(
changes["uncommitted"]["modified"],
json!(["data/", "data/z"])
);
assert_eq!(changes["uncommitted"]["added"], json!(["data/new"]));
commit(&repo, "data.wm-storage.json").unwrap();
let entries = metadata_entries(&repo, None, &["data.wm-storage.json".into()]).unwrap();
assert_eq!(
entries
.iter()
.find(|entry| entry.object == "data/a")
.unwrap()
.version_id
.as_deref(),
Some("old-a")
);
assert_eq!(
entries
.iter()
.find(|entry| entry.object == "data/z")
.unwrap()
.version_id,
None
);
}
#[test]
fn filesystem_fetch_recovers_directory_after_cache_removal() {
let (_temporary, repo) = repository();
let destination = tempfile::tempdir().unwrap();
remote(&repo, destination.path());
fs::create_dir(repo.root.join("data")).unwrap();
fs::write(repo.root.join("data/a"), b"old").unwrap();
add(&repo, "data").unwrap();
let pointers = vec!["data.wm-storage.json".into()];
push(&repo, &pointers).unwrap();
fs::remove_dir_all(cache_root(&repo).unwrap()).unwrap();
assert!(data_status(&repo, &["data".into()]).unwrap()["uncommitted"].is_null());
fs::remove_dir_all(repo.root.join("data")).unwrap();
fetch(&repo, &pointers).unwrap();
checkout(&repo, &pointers).unwrap();
assert_eq!(fs::read(repo.root.join("data/a")).unwrap(), b"old");
assert_eq!(status(&repo, &pointers, false, false).unwrap().1, "{}");
}
#[test]
fn authorized_checkout_replaces_directory_and_never_applies_corrupt_cache() {
let (_temporary, repo) = repository();
fs::create_dir(repo.root.join("data")).unwrap();
fs::write(repo.root.join("data/a"), "old a").unwrap();
fs::write(repo.root.join("data/retired"), "retired").unwrap();
add(&repo, "data").unwrap();
fs::remove_file(repo.root.join("data/retired")).unwrap();
fs::write(repo.root.join("data/a"), "new a").unwrap();
fs::write(repo.root.join("data/new"), "new file").unwrap();
commit(&repo, "data.wm-storage.json").unwrap();
fs::write(repo.root.join("data/a"), "old a").unwrap();
fs::write(repo.root.join("data/retired"), "retired").unwrap();
fs::remove_file(repo.root.join("data/new")).unwrap();
let pointers = vec!["data.wm-storage.json".into()];
checkout(&repo, &pointers).unwrap();
assert_eq!(
fs::read_to_string(repo.root.join("data/a")).unwrap(),
"new a"
);
assert!(!repo.root.join("data/retired").exists());
assert!(repo.root.join("data/new").exists());
let out = storage_metadata::read_pointer_document(&repo, "data.wm-storage.json")
.unwrap()
.outs
.remove(0);
let files = recorded_files(&repo, "data.wm-storage.json", &out, "md5").unwrap();
fs::write(existing_cache(&repo, &files[0].md5).unwrap(), "corrupt").unwrap();
let before = fs::read(repo.root.join("data/a")).unwrap();
assert!(checkout(&repo, &pointers).is_err());
assert_eq!(fs::read(repo.root.join("data/a")).unwrap(), before);
}
#[test]
fn legacy_dos2unix_pointer_is_read_without_rewriting_metadata() {
let (_temporary, repo) = repository();
fs::write(repo.root.join("text"), b"a\r\nb\r\n").unwrap();
let digest = "dd8c6a395b5dd36c56d23275028f526c";
let raw = format!("outs:\n- path: text\n md5: {digest}\n size: 6\n");
fs::write(repo.root.join("text.dvc"), &raw).unwrap();
install_cache_file_with_algorithm(&repo, digest, &repo.root.join("text"), "md5-dos2unix")
.unwrap();
let pointers = vec!["text.dvc".into()];
assert_eq!(status(&repo, &pointers, false, false).unwrap().1, "{}");
assert_eq!(data_status(&repo, &["text".into()]).unwrap(), json!({}));
fs::remove_file(repo.root.join("text")).unwrap();
checkout(&repo, &pointers).unwrap();
assert_eq!(fs::read(repo.root.join("text")).unwrap(), b"a\r\nb\r\n");
assert_eq!(fs::read_to_string(repo.root.join("text.dvc")).unwrap(), raw);
}
#[test]
fn legacy_checkout_refuses_normalized_cache_with_wrong_physical_size() {
let (_temporary, repo) = repository();
let digest = "dd8c6a395b5dd36c56d23275028f526c";
let raw = format!("outs:\n- path: text\n md5: {digest}\n size: 6\n");
fs::write(repo.root.join("text.dvc"), &raw).unwrap();
let normalized = repo.root.join("normalized");
fs::write(&normalized, b"a\nb\n").unwrap();
install_cache_file(&repo, digest, &normalized).unwrap();
let canonical = cache_path(&repo, digest).unwrap();
let error = checkout(&repo, &["text.dvc".into()]).unwrap_err();
assert!(
error.to_string().contains("size") || error.to_string().contains("No such file"),
"{error}"
);
assert!(!repo.root.join("text").exists());
assert_eq!(fs::read(canonical).unwrap(), b"a\nb\n");
assert_eq!(fs::read_to_string(repo.root.join("text.dvc")).unwrap(), raw);
}
#[test]
fn native_record_converts_an_unbound_legacy_directory_to_raw_hashes() {
let (_temporary, repo) = repository();
let storage = tempfile::tempdir().unwrap();
remote(&repo, storage.path());
fs::create_dir(repo.root.join("data")).unwrap();
fs::write(repo.root.join("data/a"), b"a\r\nb\r\n").unwrap();
let digest = "dd8c6a395b5dd36c56d23275028f526c";
let raw = format!(
"outs:\n- path: data\n md5: 178e38d9097fc874ace61e427874fc39.dir\n size: 6\n nfiles: 1\n files:\n - relpath: a\n md5: {digest}\n size: 6\n"
);
let manifest = crate::legacy_dvc::import_manifest(&raw, "data.dvc").unwrap();
write_manifest(&repo, "data.wm-storage.json", &manifest).unwrap();
let normalized = repo.root.join("normalized");
fs::write(&normalized, b"a\nb\n").unwrap();
install_cache_file(&repo, digest, &normalized).unwrap();
commit(&repo, "data.wm-storage.json").unwrap();
let canonical = cache_path(&repo, digest).unwrap();
let raw_digest = file_digest(&repo.root.join("data/a"), "md5").unwrap();
let raw_cache = cache_path(&repo, &raw_digest).unwrap();
assert_ne!(canonical, raw_cache);
assert_eq!(
read_manifest(&repo, "data.wm-storage.json")
.unwrap()
.checksum
.algorithm,
"md5"
);
assert_eq!(fs::read(&raw_cache).unwrap(), b"a\r\nb\r\n");
let pointers = vec!["data.wm-storage.json".into()];
push(&repo, &pointers).unwrap();
fs::remove_file(&raw_cache).unwrap();
fs::remove_dir_all(repo.root.join("data")).unwrap();
let before = fs::read(repo.root.join(&pointers[0])).unwrap();
fetch(&repo, &pointers).unwrap();
checkout(&repo, &pointers).unwrap();
assert_eq!(fs::read(repo.root.join("data/a")).unwrap(), b"a\r\nb\r\n");
assert_eq!(fs::read(canonical).unwrap(), b"a\nb\n");
assert_eq!(fs::read(repo.root.join(&pointers[0])).unwrap(), before);
}
#[test]
fn moving_an_unhydrated_pointer_preserves_metadata_and_ignore_rules() {
let (_temporary, repo) = repository();
fs::write(repo.root.join("data"), "data").unwrap();
add(&repo, "data").unwrap();
fs::remove_file(repo.root.join("data")).unwrap();
fs::write(repo.root.join(".gitignore"), "/data\n/keep-local\n").unwrap();
move_output(&repo, "data", "nested/moved").unwrap();
assert!(!repo.root.join("data.wm-storage.json").exists());
assert!(!repo.root.join("nested/moved").exists());
assert_eq!(
read_manifest(&repo, "nested/moved.wm-storage.json")
.unwrap()
.path,
"moved"
);
assert_eq!(
fs::read_to_string(repo.root.join(".gitignore")).unwrap(),
"/keep-local\n"
);
assert_eq!(
fs::read_to_string(repo.root.join("nested/.gitignore")).unwrap(),
"/moved\n"
);
remove(&repo, "nested/moved.wm-storage.json").unwrap();
assert_eq!(
fs::read_to_string(repo.root.join(".gitignore")).unwrap(),
"/keep-local\n"
);
}
#[cfg(unix)]
#[test]
fn file_symlinks_hash_target_bytes_but_directory_links_are_refused() {
let (_temporary, repo) = repository();
fs::create_dir(repo.root.join("data")).unwrap();
fs::write(repo.root.join("data/original"), "original").unwrap();
std::os::unix::fs::symlink("original", repo.root.join("data/link")).unwrap();
add(&repo, "data").unwrap();
let files = current_files(&repo, "data").unwrap();
assert_eq!(files.len(), 2);
assert_eq!(files[0].md5, files[1].md5);
checkout(&repo, &["data.wm-storage.json".into()]).unwrap();
assert!(
fs::symlink_metadata(repo.root.join("data/link"))
.unwrap()
.file_type()
.is_symlink()
);
fs::create_dir(repo.root.join("outside")).unwrap();
std::os::unix::fs::symlink("../outside", repo.root.join("data/dir-link")).unwrap();
assert!(current_files(&repo, "data").is_err());
let foreign = tempfile::tempdir().unwrap();
std::os::unix::fs::symlink(foreign.path(), repo.root.join("foreign")).unwrap();
assert!(add(&repo, "foreign/data").is_err());
assert!(!foreign.path().join("data.wm-storage.json").exists());
}
#[cfg(target_os = "linux")]
#[test]
fn non_utf8_directory_filenames_fail_before_metadata_or_payload_changes() {
use std::os::unix::ffi::OsStringExt;
let (_temporary, repo) = repository();
fs::create_dir(repo.root.join("data")).unwrap();
let name = std::ffi::OsString::from_vec(b"name-\xff".to_vec());
let source = repo.root.join("data").join(name);
fs::write(&source, b"local\0\r\nbytes").unwrap();
let error = add(&repo, "data").unwrap_err();
assert!(error.to_string().contains("not UTF-8"), "{error}");
assert_eq!(fs::read(source).unwrap(), b"local\0\r\nbytes");
assert!(!repo.root.join("data.wm-storage.json").exists());
assert!(!repo.root.join(".gitignore").exists());
assert_eq!(fs::read_dir(repo.root.join("data")).unwrap().count(), 1);
}
#[test]
fn transport_metadata_rejects_disabled_foreign_and_malformed_outputs() {
let (_temporary, repo) = repository();
let digest = "0cc175b9c0f1b6a831c399e269772661";
for extra in [
"cache: false",
"push: false",
"can_push: false",
"remote: another",
"hash: sha256",
"md5: FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF",
] {
let raw = if extra.starts_with("md5:") {
format!("outs:\n- path: data\n {extra}\n")
} else {
format!("outs:\n- path: data\n md5: {digest}\n {extra}\n")
};
fs::write(repo.root.join("data.dvc"), raw).unwrap();
assert!(
metadata_entries(&repo, None, &["data.dvc".into()]).is_err(),
"{extra}"
);
}
fs::write(repo.root.join("data.dvc"), "outs: []\n").unwrap();
assert!(metadata_entries(&repo, None, &["data.dvc".into()]).is_err());
assert!(
crate::legacy_dvc::hash_algorithm(
"outs:\n- path: data\n files:\n - relpath: a\n remote: foreign\n",
"data.dvc"
)
.is_err()
);
}
fn upload_fixture(repo: &GitRepo, bytes: &[u8]) -> (StorageEntry, PathBuf) {
let source = repo.root.join("source");
fs::write(&source, bytes).unwrap();
let digest = file_digest(&source, "md5").unwrap();
install_cache_file(repo, &digest, &source).unwrap();
(
StorageEntry {
verification: None,
pointer: "source.dvc".into(),
object: "task/source".into(),
md5: Some(digest),
size: Some(bytes.len() as u64),
version_id: None,
etag: None,
hash_name: "md5".into(),
},
source,
)
}
fn uploaded_head(token: &str, size: usize) -> crate::native_s3::tests::Reply {
crate::native_s3::tests::Reply {
status: 200,
headers: vec![
("x-amz-version-id", "owned-version".into()),
("etag", "\"owned-etag\"".into()),
("content-length", size.to_string()),
("x-amz-meta-workspace-mgr-upload", token.into()),
],
body: Vec::new(),
}
}
fn uploaded_get(bytes: &[u8], version: &str, etag: &str) -> crate::native_s3::tests::Reply {
crate::native_s3::tests::Reply {
status: 200,
headers: vec![("x-amz-version-id", version.into()), ("etag", etag.into())],
body: bytes.to_vec(),
}
}
fn upload_request_route(
request: &crate::native_s3::tests::WireRequest,
) -> (String, BTreeMap<String, String>) {
let target = url::Url::parse(&format!("http://fixture{}", request.target)).unwrap();
let query = target
.query_pairs()
.into_owned()
.collect::<BTreeMap<_, _>>();
assert_eq!(query.len(), target.query_pairs().count());
(target.path().into(), query)
}
fn assert_uploaded_read(request: &crate::native_s3::tests::WireRequest, object: &str) {
assert!(matches!(request.method.as_str(), "HEAD" | "GET"));
let (path, query) = upload_request_route(request);
assert_eq!(path, format!("/fixture-bucket/root/{object}"));
assert_eq!(
query,
BTreeMap::from([("versionId".into(), "owned-version".into())])
);
if request.method == "GET" {
assert_eq!(request.headers["if-match"], "\"owned-etag\"");
}
}
fn uploaded_read_fixture(
bytes: &'static [u8],
version: &'static str,
etag: &'static str,
) -> (
crate::native_s3::S3Client,
crate::native_s3::tests::RoutedFixture,
) {
crate::native_s3::tests::routed_fixture(move |request| {
assert_uploaded_read(request, "task/source");
if request.method == "HEAD" {
uploaded_head("owned-token", bytes.len())
} else {
uploaded_get(bytes, version, etag)
}
})
}
#[test]
fn multipart_upload_streams_complete_parts_and_verifies_its_exact_version() {
multipart_upload_with_proof(false);
}
#[test]
fn proven_multipart_upload_binds_ordered_signed_parts_without_downloading() {
multipart_upload_with_proof(true);
}
fn multipart_upload_with_proof(proven: bool) {
use crate::native_s3::tests::{Reply, routed_fixture};
let (_temporary, repo) = repository();
let bytes = b"abcdefghijkl";
let (entry, source) = upload_fixture(&repo, bytes);
let token = std::sync::Arc::new(std::sync::Mutex::new(String::new()));
let captured = token.clone();
let (client, worker) = routed_fixture(move |request| {
let (path, query) = upload_request_route(request);
assert_eq!(path, "/fixture-bucket/root/task/source");
if request.method == "POST" && query.contains_key("uploads") {
assert_eq!(query, BTreeMap::from([("uploads".into(), String::new())]));
assert!(captured.lock().unwrap().is_empty());
*captured.lock().unwrap() =
request.headers["x-amz-meta-workspace-mgr-upload"].clone();
return Reply::xml(
"<InitiateMultipartUploadResult><UploadId>owned-upload</UploadId></InitiateMultipartUploadResult>",
);
}
if request.method == "PUT" {
assert_eq!(
request.headers["x-amz-content-sha256"],
crate::hex::encode_lower(sha2::Sha256::digest(&request.body))
);
assert_eq!(query.len(), 2);
assert_eq!(query["uploadId"], "owned-upload");
assert!(matches!(query["partNumber"].as_str(), "1" | "2" | "3"));
return Reply {
status: 200,
headers: vec![("etag", "\"part-etag\"".into())],
body: Vec::new(),
};
}
if request.method == "POST" {
assert_eq!(
query,
BTreeMap::from([("uploadId".into(), "owned-upload".into())])
);
assert!(!captured.lock().unwrap().is_empty());
return Reply {status:200,headers:vec![("x-amz-version-id","owned-version".into())],body:b"<CompleteMultipartUploadResult><ETag>\"owned-etag\"</ETag></CompleteMultipartUploadResult>".to_vec()};
}
assert_uploaded_read(request, "task/source");
if request.method == "HEAD" {
return uploaded_head(&captured.lock().unwrap(), bytes.len());
}
uploaded_get(bytes, "owned-version", "\"owned-etag\"")
});
let raw = file_sha256(&source).unwrap();
assert_eq!(
upload_version_in(
&client,
&repo,
&entry,
&source,
4,
5,
UploadPolicy {
namespace: "uploads",
condition: None,
raw_sha256: proven.then_some(raw.as_str()),
local: None,
hashes: None,
}
)
.unwrap(),
("owned-version".into(), "owned-etag".into())
);
let requests = worker.finish_requests();
assert_eq!(
requests
.iter()
.filter(|request| request.method == "GET")
.count(),
usize::from(!proven)
);
assert_eq!(
requests
.iter()
.filter(|request| request.method == "POST")
.count(),
2
);
let mut parts = requests
.iter()
.filter(|request| request.method == "PUT")
.map(|request| {
(
upload_request_route(request).1["partNumber"]
.parse::<u64>()
.unwrap(),
request.body.clone(),
)
})
.collect::<Vec<_>>();
parts.sort_by_key(|(number, _)| *number);
assert_eq!(
parts
.into_iter()
.map(|(_, bytes)| bytes)
.collect::<Vec<_>>(),
vec![b"abcde".to_vec(), b"fghij".to_vec(), b"kl".to_vec()]
);
assert!(
requests
.iter()
.filter(|request| request.method == "PUT")
.all(|request| request.headers.contains_key("content-md5"))
);
let completion = String::from_utf8(
requests
.iter()
.find(|request| request.method == "POST" && !request.target.contains("uploads"))
.unwrap()
.body
.clone(),
)
.unwrap();
assert_eq!(completion.matches("<PartNumber>").count(), 3);
assert!(completion.contains("<PartNumber>3</PartNumber>"));
assert!(!token.lock().unwrap().is_empty());
}
#[test]
fn lost_put_response_recovers_owned_version_without_another_upload() {
lost_put_response_recovers_owned_version(false);
}
#[test]
fn lost_put_response_replays_interrupted_reads_without_another_upload() {
lost_put_response_recovers_owned_version(true);
}
fn lost_put_response_recovers_owned_version(interrupt: bool) {
use crate::native_s3::tests::{Reply, interrupt_response_once, routed_fixture};
let (_temporary, repo) = repository();
let bytes = b"recovered bytes";
let (entry, source) = upload_fixture(&repo, bytes);
let captured = std::sync::Arc::new(std::sync::Mutex::new(String::new()));
let token = captured.clone();
let (received, request_received) = std::sync::mpsc::channel();
let (client, worker) = routed_fixture(move |request| {
let (path, query) = upload_request_route(request);
if request.method == "PUT" {
assert_eq!(path, "/fixture-bucket/root/task/source");
assert!(query.is_empty());
assert!(token.lock().unwrap().is_empty());
*token.lock().unwrap() = request.headers["x-amz-meta-workspace-mgr-upload"].clone();
return Reply {status:500,headers:Vec::new(),body:b"<Error><Code>InternalError</Code><Message>response lost after commit</Message></Error>".to_vec()};
}
assert!(!token.lock().unwrap().is_empty());
if request.method == "GET" && query.contains_key("versions") {
assert_eq!(path, "/fixture-bucket");
assert_eq!(
query,
BTreeMap::from([
("versions".into(), String::new()),
("prefix".into(), "root/task/source".into()),
("max-keys".into(), "1000".into())
])
);
return Reply::xml(
"<ListVersionsResult><IsTruncated>false</IsTruncated><Version><Key>root/task/source</Key><VersionId>owned-version</VersionId><Size>15</Size><ETag>\"owned-etag\"</ETag></Version></ListVersionsResult>",
);
}
assert_uploaded_read(request, "task/source");
let _ = received.send(());
if request.method == "HEAD" {
return uploaded_head(&token.lock().unwrap(), bytes.len());
}
uploaded_get(bytes, "owned-version", "\"owned-etag\"")
});
let (client, injected) = if interrupt {
let (client, injected) = interrupt_response_once(
client,
"/fixture-bucket/root/task/source?versionId=owned-version".into(),
request_received,
);
(client, Some(injected))
} else {
(client, None)
};
assert_eq!(
upload_version(&client, &repo, &entry, &source, 100, 5)
.unwrap()
.0,
"owned-version"
);
let requests = worker.finish_requests();
assert_eq!(
requests
.iter()
.filter(|request| request.method == "PUT")
.count(),
1
);
let (_, journal) = upload_journal(&repo, &client, &entry).unwrap();
assert_eq!(journal["phase"], "complete");
assert_eq!(journal["version_id"], "owned-version");
if let Some(injected) = injected {
assert_eq!(injected.load(std::sync::atomic::Ordering::SeqCst), 1);
assert!(
requests
.iter()
.filter(|request| request.method == "HEAD")
.count()
>= 2
);
}
}
#[test]
fn multipart_failure_aborts_only_the_known_owned_upload() {
use crate::native_s3::tests::{Reply, routed_fixture};
let (_temporary, repo) = repository();
let (entry, source) = upload_fixture(&repo, b"multipart bytes");
let (client, worker) = routed_fixture(move |request| {
let (path, query) = upload_request_route(request);
if request.method == "POST" {
assert_eq!(path, "/fixture-bucket/root/task/source");
assert_eq!(query, BTreeMap::from([("uploads".into(), String::new())]));
return Reply::xml(
"<InitiateMultipartUploadResult><UploadId>owned-upload</UploadId></InitiateMultipartUploadResult>",
);
}
if request.method == "PUT" {
assert_eq!(path, "/fixture-bucket/root/task/source");
assert_eq!(query.len(), 2);
assert_eq!(query["uploadId"], "owned-upload");
assert!(matches!(query["partNumber"].as_str(), "1" | "2" | "3"));
return Reply {
status: 403,
headers: Vec::new(),
body: b"<Error><Code>AccessDenied</Code></Error>".to_vec(),
};
}
if request.method == "DELETE" {
assert_eq!(path, "/fixture-bucket/root/task/source");
assert_eq!(
query,
BTreeMap::from([("uploadId".into(), "owned-upload".into())])
);
return Reply {
status: 204,
headers: Vec::new(),
body: Vec::new(),
};
}
assert_eq!(request.method, "GET");
assert_eq!(path, "/fixture-bucket");
assert_eq!(
query,
BTreeMap::from([
("versions".into(), String::new()),
("prefix".into(), "root/task/source".into()),
("max-keys".into(), "1000".into())
])
);
Reply::xml("<ListVersionsResult><IsTruncated>false</IsTruncated></ListVersionsResult>")
});
assert!(upload_version(&client, &repo, &entry, &source, 4, 5).is_err());
let requests = worker.finish_requests();
for method in ["POST", "DELETE"] {
assert_eq!(
requests
.iter()
.filter(|request| request.method == method)
.count(),
1
);
}
let uploads = requests
.iter()
.filter(|request| request.method == "PUT")
.count();
assert!((1..=3).contains(&uploads));
let (_, journal) = upload_journal(&repo, &client, &entry).unwrap();
assert_eq!(journal["phase"], "planned");
assert!(journal["upload_id"].is_null());
}
#[test]
fn uploaded_get_cannot_substitute_another_exact_generation_or_etag() {
for (version, etag) in [
("foreign-version", "\"owned-etag\""),
("owned-version", "\"foreign-etag\""),
] {
let (_temporary, repo) = repository();
let bytes = b"same bytes";
let (entry, _source) = upload_fixture(&repo, bytes);
let (client, worker) = uploaded_read_fixture(bytes, version, etag);
assert!(
verify_uploaded_version(
&client,
&repo,
&entry,
"owned-token",
"owned-version",
None,
false
)
.is_err()
);
let requests = worker.finish_requests();
assert!(requests.iter().any(|request| request.method == "GET"));
}
}
#[test]
fn import_upload_receipts_distinguish_raw_variants_of_one_normalized_checksum() {
use crate::native_s3::tests::empty_fixture;
let (_temporary, repo) = repository();
let (mut entry, source) = upload_fixture(&repo, b"a\r\nb\n");
entry.hash_name = "md5-dos2unix".into();
entry.md5 = Some(file_digest(&source, &entry.hash_name).unwrap());
let a = file_sha256(&source).unwrap();
fs::write(&source, b"a\nb\r\n").unwrap();
assert_eq!(
file_digest(&source, &entry.hash_name).unwrap(),
entry.md5.as_deref().unwrap()
);
let b = file_sha256(&source).unwrap();
let (client, worker) = empty_fixture();
let (a_path, a_receipt) =
upload_journal_in(&repo, &client, &entry, "storage-import-uploads", Some(&a)).unwrap();
let (b_path, b_receipt) =
upload_journal_in(&repo, &client, &entry, "storage-import-uploads", Some(&b)).unwrap();
assert_ne!(a_path, b_path);
assert_ne!(a_receipt["token"], b_receipt["token"]);
assert_eq!(a_receipt["context"]["raw_sha256"], a);
assert_eq!(b_receipt["context"]["raw_sha256"], b);
assert_eq!(
serde_json::from_slice::<Value>(&fs::read(a_path).unwrap()).unwrap(),
a_receipt
);
assert!(upload_context(&repo, &client, &entry, "storage-import-uploads", None).is_err());
assert!(worker.finish_requests().is_empty());
}
#[test]
fn imported_exact_uploaded_get_rejects_a_different_raw_normalized_variant() {
let (_temporary, repo) = repository();
let (mut entry, source) = upload_fixture(&repo, b"a\r\nb\n");
entry.hash_name = "md5-dos2unix".into();
entry.md5 = Some(file_digest(&source, &entry.hash_name).unwrap());
let expected = file_sha256(&source).unwrap();
let bytes = b"a\nb\r\n";
let (client, worker) = uploaded_read_fixture(bytes, "owned-version", "\"owned-etag\"");
assert!(
verify_uploaded_version(
&client,
&repo,
&entry,
"owned-token",
"owned-version",
Some(&expected),
false
)
.is_err()
);
let requests = worker.finish_requests();
assert!(requests.iter().any(|request| request.method == "GET"));
}
#[test]
fn restoring_legacy_cache_preserves_hash_algorithm_and_exact_binding() {
let (_temporary, repo) = repository();
fs::write(repo.root.join("text"), b"a\r\nb\n").unwrap();
let digest = file_digest(&repo.root.join("text"), "md5-dos2unix").unwrap();
let raw = format!(
"outs:\n- path: text\n hash: md5-dos2unix\n md5: {digest}\n size: 5\n cloud:\n workspace-mgr:\n version_id: original-version\n etag: original-etag\n"
);
let manifest = crate::legacy_dvc::import_manifest(&raw, "text.dvc").unwrap();
write_manifest(&repo, "text.wm-storage.json", &manifest).unwrap();
let original = metadata_entries(&repo, None, &["text.wm-storage.json".into()])
.unwrap()
.remove(0);
assert!(
!storage_metadata::payload_matches_metadata(
&repo,
"text.wm-storage.json",
&manifest.serialize().unwrap()
)
.unwrap()
);
install_cache_for_entry(&repo, &original, &repo.root.join("text")).unwrap();
commit(&repo, "text.wm-storage.json").unwrap();
let entries = metadata_entries(&repo, None, &["text.wm-storage.json".into()]).unwrap();
assert_eq!(entries[0].hash_name, "md5-dos2unix");
assert_eq!(entries[0].md5.as_deref(), Some(digest.as_str()));
assert_eq!(entries[0].version_id.as_deref(), Some("original-version"));
assert_eq!(
fs::read(existing_cache_with_algorithm(&repo, &digest, "md5-dos2unix").unwrap())
.unwrap(),
b"a\r\nb\n"
);
fs::write(repo.root.join("text"), b"a\nb\r\n").unwrap();
assert_eq!(
file_digest(&repo.root.join("text"), "md5-dos2unix").unwrap(),
digest
);
assert_eq!(
status(&repo, &["text.wm-storage.json".into()], false, true)
.unwrap()
.0,
1
);
assert!(
!storage_metadata::payload_matches_metadata(
&repo,
"text.wm-storage.json",
&manifest.serialize().unwrap()
)
.unwrap()
);
commit(&repo, "text.wm-storage.json").unwrap();
let changed = read_manifest(&repo, "text.wm-storage.json").unwrap();
assert_eq!(changed.checksum.algorithm, "md5");
assert_eq!(
changed.checksum.digest,
file_digest(&repo.root.join("text"), "md5").unwrap()
);
assert!(
metadata_entries(&repo, None, &["text.wm-storage.json".into()]).unwrap()[0]
.version_id
.is_none()
);
}
#[test]
fn an_independent_cloud_binding_edit_during_upload_is_preserved() {
use crate::native_s3::tests::{Reply, configure_repo, routed_fixture};
let (_temporary, repo) = repository();
let bytes = b"same bytes";
fs::write(repo.root.join("source"), bytes).unwrap();
add(&repo, "source").unwrap();
let pointer = repo.root.join("source.wm-storage.json");
let mut legacy = read_manifest(&repo, "source.wm-storage.json").unwrap();
legacy.schema_version = 1;
write_manifest(&repo, "source.wm-storage.json", &legacy).unwrap();
let captured = std::sync::Arc::new(std::sync::Mutex::new((String::new(), Vec::new())));
let expected = captured.clone();
let (client, worker) = routed_fixture(move |request| {
let (path, query) = upload_request_route(request);
if request.method == "GET" && query.contains_key("versioning") {
assert_eq!(path, "/fixture-bucket");
assert_eq!(
query,
BTreeMap::from([("versioning".into(), String::new())])
);
return Reply::xml(
"<VersioningConfiguration><Status>Enabled</Status></VersioningConfiguration>",
);
}
if request.method == "PUT" {
assert_eq!(path, "/fixture-bucket/root/source");
assert!(query.is_empty());
assert!(expected.lock().unwrap().0.is_empty());
let mut raw = Manifest::parse(
&fs::read_to_string(&pointer).unwrap(),
"source.wm-storage.json",
)
.unwrap();
raw.version = Some(Version {
verification: None,
id: "independent-version".into(),
etag: Some("independent-etag".into()),
});
let rendered = raw.serialize().unwrap().into_bytes();
fs::write(&pointer, &rendered).unwrap();
*expected.lock().unwrap() = (
request.headers["x-amz-meta-workspace-mgr-upload"].clone(),
rendered,
);
return Reply {
status: 200,
headers: vec![
("x-amz-version-id", "owned-version".into()),
("etag", "\"owned-etag\"".into()),
],
body: Vec::new(),
};
}
assert_uploaded_read(request, "source");
if request.method == "HEAD" {
return uploaded_head(&expected.lock().unwrap().0, bytes.len());
}
uploaded_get(bytes, "owned-version", "\"owned-etag\"")
});
configure_repo(&client, &repo);
let error = push_versioned(&repo, &["source.wm-storage.json".into()]).unwrap_err();
assert!(error.to_string().contains("metadata changed during upload"));
let requests = worker.finish_requests();
assert_eq!(
requests
.iter()
.filter(|request| request.method == "PUT")
.count(),
1
);
assert_eq!(
fs::read(repo.root.join("source.wm-storage.json")).unwrap(),
captured.lock().unwrap().1
);
}
#[test]
fn changing_a_file_boundary_into_a_directory_is_dirty_even_if_one_file_matches() {
let (_temporary, repo) = repository();
fs::write(repo.root.join("source"), "same").unwrap();
add(&repo, "source").unwrap();
fs::remove_file(repo.root.join("source")).unwrap();
fs::create_dir(repo.root.join("source")).unwrap();
fs::write(repo.root.join("source/one"), "same").unwrap();
let value: Value = serde_json::from_str(
&status(&repo, &["source.wm-storage.json".into()], false, false)
.unwrap()
.1,
)
.unwrap();
assert_eq!(
value["source.wm-storage.json"][0]["changed outs"]["source"],
"modified"
);
}
#[test]
fn schema2_publication_records_exact_proof_and_sha_cache_without_payload_get() {
use crate::native_s3::tests::{Reply, configure_repo, routed_fixture};
let (_temporary, repo) = repository();
let bytes = b"raw bytes\r\n";
fs::write(repo.root.join("source"), bytes).unwrap();
add(&repo, "source").unwrap();
let token = std::sync::Arc::new(std::sync::Mutex::new(String::new()));
let captured = token.clone();
let (client, worker) = routed_fixture(move |request| {
let (_, query) = upload_request_route(request);
if query.contains_key("versioning") {
return Reply::xml(
"<VersioningConfiguration><Status>Enabled</Status></VersioningConfiguration>",
);
}
if request.method == "PUT" {
assert_eq!(request.body, bytes);
assert_eq!(
request.headers["x-amz-content-sha256"],
crate::hex::encode_lower(sha2::Sha256::digest(bytes))
);
*captured.lock().unwrap() =
request.headers["x-amz-meta-workspace-mgr-upload"].clone();
return Reply {
status: 200,
headers: vec![
("x-amz-version-id", "owned-version".into()),
("etag", "\"owned-etag\"".into()),
],
body: Vec::new(),
};
}
assert_eq!(request.method, "HEAD");
assert_uploaded_read(request, "source");
uploaded_head(&captured.lock().unwrap(), bytes.len())
});
configure_repo(&client, &repo);
push_versioned(&repo, &["source.wm-storage.json".into()]).unwrap();
let entry = metadata_entries(&repo, None, &["source.wm-storage.json".into()])
.unwrap()
.remove(0);
let proof = entry.verification.as_ref().unwrap();
assert_eq!(proof.version_id, "owned-version");
assert_eq!(
proof.checksum.digest,
file_sha256(&repo.root.join("source")).unwrap()
);
assert_eq!(proof.endpoint, client.endpoint_identity());
assert_eq!(
file_sha256(&cache_path_for_entry(&repo, &entry).unwrap()).unwrap(),
proof.checksum.digest
);
assert_eq!(data_status(&repo, &["source".into()]).unwrap(), json!({}));
push_versioned(&repo, &["source.wm-storage.json".into()]).unwrap();
let requests = worker.finish_requests();
assert_eq!(
requests
.iter()
.filter(|request| request.method == "PUT")
.count(),
1
);
assert!(
!requests
.iter()
.any(|request| request.method == "GET" && !request.target.contains("versioning"))
);
}
#[test]
fn unacknowledged_upload_recovery_requires_content_proof_beyond_its_token() {
use crate::native_s3::tests::{Reply, routed_fixture};
use base64::Engine;
for (remote, provider) in [
(b"a\r\nb\n".as_slice(), false),
(b"a\nb\r\n".as_slice(), false),
(b"a\r\nb\n".as_slice(), true),
] {
let (_temporary, repo) = repository();
let expected_bytes = b"a\r\nb\n";
let (mut entry, source) = upload_fixture(&repo, expected_bytes);
entry.hash_name = "md5-dos2unix".into();
entry.md5 = Some(file_digest(&source, &entry.hash_name).unwrap());
let raw = file_sha256(&source).unwrap();
let (client, worker) = routed_fixture(move |request| {
let (_, query) = upload_request_route(request);
if request.method == "GET" && query.contains_key("versions") {
return Reply::xml(
"<ListVersionsResult><IsTruncated>false</IsTruncated><Version><Key>root/task/source</Key><VersionId>owned-version</VersionId><Size>5</Size><ETag>\"owned-etag\"</ETag></Version></ListVersionsResult>",
);
}
assert_uploaded_read(request, "task/source");
if request.method == "HEAD" {
let mut reply = uploaded_head("owned-token", remote.len());
if provider
&& request
.headers
.get("x-amz-checksum-mode")
.map(String::as_str)
== Some("ENABLED")
{
reply
.headers
.push(("x-amz-checksum-type", "FULL_OBJECT".into()));
reply.headers.push((
"x-amz-checksum-sha256",
base64::engine::general_purpose::STANDARD
.encode(sha2::Sha256::digest(remote)),
));
}
return reply;
}
uploaded_get(remote, "owned-version", "\"owned-etag\"")
});
let journal = json!({
"token":"owned-token", "phase":"uploading",
"verified_transfer":true, "transfer_endpoint":client.endpoint_identity(),
"context":{"raw_sha256":raw},
});
let recovered = recover_uploaded_version(&client, &repo, &entry, &journal);
assert_eq!(recovered.is_ok(), remote == expected_bytes);
let requests = worker.finish_requests();
assert_eq!(
requests
.iter()
.filter(
|request| request.method == "GET" && request.target.contains("versionId=")
)
.count(),
usize::from(!provider),
);
assert!(!requests.iter().any(|request| request.method == "PUT"));
}
}
#[test]
fn legacy_upload_journal_is_recovered_before_creating_a_proven_upload() {
let (_temporary, repo) = repository();
let bytes = b"same bytes";
let (entry, source) = upload_fixture(&repo, bytes);
let (client, worker) = uploaded_read_fixture(bytes, "owned-version", "\"owned-etag\"");
let (old_path, mut old) = upload_journal(&repo, &client, &entry).unwrap();
old["token"] = "owned-token".into();
old["phase"] = "complete".into();
old["version_id"] = "owned-version".into();
save_upload(&old_path, &old).unwrap();
let raw = file_sha256(&source).unwrap();
let result = upload_version_in(
&client,
&repo,
&entry,
&source,
100,
5,
UploadPolicy {
namespace: "uploads",
condition: None,
raw_sha256: Some(&raw),
local: None,
hashes: None,
},
)
.unwrap();
assert_eq!(result.0, "owned-version");
assert!(!old_path.exists());
let (_, new) = upload_journal_in(&repo, &client, &entry, "uploads", Some(&raw)).unwrap();
assert_eq!(new["token"], "owned-token");
assert_ne!(new["verified_transfer"], true);
let requests = worker.finish_requests();
assert_eq!(
requests
.iter()
.filter(|request| request.method == "GET")
.count(),
1
);
assert!(!requests.iter().any(|request| request.method == "PUT"));
}
}