From 21454e5476fed966254fc0305340f0bb95cdca6e Mon Sep 17 00:00:00 2001 From: J Robert Ray Date: Tue, 11 Nov 2025 13:34:51 -0800 Subject: [PATCH 1/4] Fix ambiguous digest error when blob files exist If the repo still contains legacy blob files, resolving a partial digest can discover both the blob object and payload digest, which is then treated as ambiguous. Even when not using a partial digest, resolving the env spec can fail as ambiguous. This change makes it prefer finding a payload instead of being ambiguous in this situation. It ignores the possibility that a non-blob object may exist with the same payload as a digest, but in the past this would have been a bigger problem of the payload's blob conflicting with the non-blob object that both want to have the same filename. Signed-off-by: J Robert Ray --- crates/spfs/src/graph/database.rs | 44 ++++++++++++++++++++++++------- 1 file changed, 35 insertions(+), 9 deletions(-) diff --git a/crates/spfs/src/graph/database.rs b/crates/spfs/src/graph/database.rs index d9b2228a6..03800875c 100644 --- a/crates/spfs/src/graph/database.rs +++ b/crates/spfs/src/graph/database.rs @@ -2,7 +2,7 @@ // SPDX-License-Identifier: Apache-2.0 // https://github.com/spkenv/spk -use std::collections::VecDeque; +use std::collections::{HashMap, VecDeque}; use std::pin::Pin; use std::task::Poll; @@ -160,7 +160,7 @@ pub enum DigestSearchCriteria { } /// The types of digests that can exist in a database. -#[derive(PartialEq, Eq)] +#[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum FoundDigest { Object(encoding::Digest), Payload(encoding::Digest), @@ -263,16 +263,42 @@ pub trait DatabaseView: Sync + Send { /// - UnknownReferenceError: if the digest cannot be resolved /// - AmbiguousReferenceError: if the digest could point to multiple items async fn resolve_full_digest(&self, partial: &encoding::PartialDigest) -> Result { - let options: Vec<_> = self - .find_digests(&crate::graph::DigestSearchCriteria::StartsWith( - partial.clone(), - )) - .try_collect() - .await?; + #[derive(Debug)] + struct UpgradeToPayload(FoundDigest); + + impl UpgradeToPayload { + /// Replace the FoundDigest if fd is a Payload. + /// + /// If we observe both a legacy blob object and a payload with the + /// same digest we will only remember seeing the payload. This + /// method assumes it is only called with a FoundDigest that has the + /// same digest as the present one. + fn upgrade(&mut self, fd: FoundDigest) { + if matches!(fd, FoundDigest::Payload(_)) { + self.0 = fd; + } + } + } + + let mut options = HashMap::<_, UpgradeToPayload>::new(); + let filter = crate::graph::DigestSearchCriteria::StartsWith(partial.clone()); + let mut stream = self.find_digests(&filter); + while let Some(fd) = stream.try_next().await? { + // Hash on the raw digest to avoid double-counting legacy blobs and + // their payloads. + options + .entry(*fd.digest()) + .and_modify(|utp| utp.upgrade(fd)) + .or_insert_with(|| UpgradeToPayload(fd)); + } match options.len() { 0 => Err(Error::UnknownReference(partial.to_string())), - 1 => Ok(options.into_iter().next().unwrap()), + 1 => Ok(options + .into_iter() + .next() + .map(|(_, UpgradeToPayload(fd))| fd) + .unwrap()), _ => Err(Error::AmbiguousReference(partial.to_string())), } } From a22ab2aa3bfa444a6c4e3fe35ec4735224a70d62 Mon Sep 17 00:00:00 2001 From: J Robert Ray Date: Tue, 11 Nov 2025 13:52:15 -0800 Subject: [PATCH 2/4] Provide an optional digest type when resolving partials As mentioned in the previous commit, a possibility exists for a non-blob object to have the same digest as a payload. When resolving a partial digest for this case, the most correct behavior would be to return an ambiguous digest error. What happens as of the previous commit is it will always resolve to the payload. This new argument offers a way to still resolve the object, if the caller knows in advance that the type of item expected is an object. In practice there was only one place where the expected item type is already known. Signed-off-by: J Robert Ray --- crates/spfs/src/check.rs | 6 ++- crates/spfs/src/graph/database.rs | 42 +++++++++++++++++++- crates/spfs/src/graph/mod.rs | 1 + crates/spfs/src/resolve.rs | 22 +++++----- crates/spfs/src/storage/fs/database.rs | 35 ++++++++++++---- crates/spfs/src/storage/pinned/repository.rs | 10 ++++- crates/spfs/src/storage/repository.rs | 2 +- crates/spfs/src/storage/tar/repository.rs | 10 ++++- crates/spfs/src/sync.rs | 7 +++- crates/spfs/src/tracking/env.rs | 4 +- 10 files changed, 111 insertions(+), 28 deletions(-) diff --git a/crates/spfs/src/check.rs b/crates/spfs/src/check.rs index 47f8214ec..ea8aa5b5f 100644 --- a/crates/spfs/src/check.rs +++ b/crates/spfs/src/check.rs @@ -230,7 +230,11 @@ where &self, partial: encoding::PartialDigest, ) -> Result { - match self.repo.resolve_full_digest(&partial).await? { + match self + .repo + .resolve_full_digest(&partial, graph::PartialDigestType::Unknown) + .await? + { FoundDigest::Object(digest) => self .check_object_digest(digest) .await diff --git a/crates/spfs/src/graph/database.rs b/crates/spfs/src/graph/database.rs index 03800875c..beaf7c3c1 100644 --- a/crates/spfs/src/graph/database.rs +++ b/crates/spfs/src/graph/database.rs @@ -195,6 +195,30 @@ impl FoundDigest { } } +/// The types of items a partial digest can reference. +#[derive(Clone, Copy)] +pub enum PartialDigestType { + Object, + Payload, + Unknown, +} + +impl PartialDigestType { + /// Return true if the `FoundDigest` is a different item type. + /// + /// Unknown always return false. + #[inline] + pub fn conflicts_with(&self, fd: &FoundDigest) -> bool { + match (self, fd) { + (PartialDigestType::Unknown, _) + | (PartialDigestType::Object, FoundDigest::Object(_)) + | (PartialDigestType::Payload, FoundDigest::Payload(_)) => false, + (PartialDigestType::Object, FoundDigest::Payload(_)) + | (PartialDigestType::Payload, FoundDigest::Object(_)) => true, + } + } +} + /// A read-only object database. #[async_trait::async_trait] pub trait DatabaseView: Sync + Send { @@ -256,13 +280,25 @@ pub trait DatabaseView: Sync + Send { /// Resolve the complete item digest from a shortened one. /// + /// The type of item expected can be specified with `partial_digest_type`- + /// If `PartialDigestType::Unknown` is specified and both an object and + /// payload are found with the same digest, this will resolve to the payload + /// instead of fail with `AmbiguousReferenceError`, for interoperability + /// with repos containing legacy blob object files. Otherwise the type is + /// used to disambiguate in the unlikely case a non-blob object and payload + /// have the same digest. + /// /// By default this is an O(n) operation defined by the number of items. /// Other implementations may provide better results. /// /// # Errors /// - UnknownReferenceError: if the digest cannot be resolved /// - AmbiguousReferenceError: if the digest could point to multiple items - async fn resolve_full_digest(&self, partial: &encoding::PartialDigest) -> Result { + async fn resolve_full_digest( + &self, + partial: &encoding::PartialDigest, + partial_digest_type: PartialDigestType, + ) -> Result { #[derive(Debug)] struct UpgradeToPayload(FoundDigest); @@ -284,6 +320,10 @@ pub trait DatabaseView: Sync + Send { let filter = crate::graph::DigestSearchCriteria::StartsWith(partial.clone()); let mut stream = self.find_digests(&filter); while let Some(fd) = stream.try_next().await? { + if partial_digest_type.conflicts_with(&fd) { + continue; + } + // Hash on the raw digest to avoid double-counting legacy blobs and // their payloads. options diff --git a/crates/spfs/src/graph/mod.rs b/crates/spfs/src/graph/mod.rs index b5637594b..103d7ea92 100644 --- a/crates/spfs/src/graph/mod.rs +++ b/crates/spfs/src/graph/mod.rs @@ -34,6 +34,7 @@ pub use database::{ DatabaseWalker, DigestSearchCriteria, FoundDigest, + PartialDigestType, }; pub use entry::Entry; pub use kind::{HasKind, Kind, ObjectKind}; diff --git a/crates/spfs/src/resolve.rs b/crates/spfs/src/resolve.rs index f79b0af0c..a2e8d8f96 100644 --- a/crates/spfs/src/resolve.rs +++ b/crates/spfs/src/resolve.rs @@ -149,16 +149,20 @@ pub async fn compute_environment_manifest( .filter_map(|i| match i { tracking::EnvSpecItem::Digest(d) => Some(std::future::ready(Ok(*d)).boxed()), tracking::EnvSpecItem::PartialDigest(p) => Some( - repo.resolve_full_digest(p) - .and_then(|found_digest| async move { - match found_digest { - graph::FoundDigest::Object(digest) => Ok(digest), - graph::FoundDigest::Payload(_digest) => { - Err("unexpected payload digest in environment spec".into()) - } + repo.resolve_full_digest( + p, + // by the error below, we only expect to find object digests + graph::PartialDigestType::Object, + ) + .and_then(|found_digest| async move { + match found_digest { + graph::FoundDigest::Object(digest) => Ok(digest), + graph::FoundDigest::Payload(_digest) => { + Err("unexpected payload digest in environment spec".into()) } - }) - .boxed(), + } + }) + .boxed(), ), tracking::EnvSpecItem::TagSpec(t) => { Some(repo.resolve_tag(t).map_ok(|t| t.target).boxed()) diff --git a/crates/spfs/src/storage/fs/database.rs b/crates/spfs/src/storage/fs/database.rs index 8f1e7d376..3cdbfb9e8 100644 --- a/crates/spfs/src/storage/fs/database.rs +++ b/crates/spfs/src/storage/fs/database.rs @@ -47,8 +47,15 @@ impl DatabaseView for super::MaybeOpenFsRepository { graph::DatabaseWalker::new(self, *root) } - async fn resolve_full_digest(&self, partial: &encoding::PartialDigest) -> Result { - self.opened().await?.resolve_full_digest(partial).await + async fn resolve_full_digest( + &self, + partial: &encoding::PartialDigest, + partial_digest_type: graph::PartialDigestType, + ) -> Result { + self.opened() + .await? + .resolve_full_digest(partial, partial_digest_type) + .await } } @@ -127,12 +134,24 @@ impl DatabaseView for super::OpenFsRepository { graph::DatabaseWalker::new(self, *root) } - async fn resolve_full_digest(&self, partial: &encoding::PartialDigest) -> Result { - match self.objects.resolve_full_digest(partial).await { - Ok(digest) => Ok(FoundDigest::Object(digest)), - Err(_) => { - let digest = self.payloads.resolve_full_digest(partial).await?; - Ok(FoundDigest::Payload(digest)) + async fn resolve_full_digest( + &self, + partial: &encoding::PartialDigest, + partial_digest_type: graph::PartialDigestType, + ) -> Result { + match ( + partial_digest_type, + self.objects + .resolve_full_digest(partial) + .map_ok(FoundDigest::Object), + self.payloads + .resolve_full_digest(partial) + .map_ok(FoundDigest::Payload), + ) { + (graph::PartialDigestType::Object, f, _) => f.await, + (graph::PartialDigestType::Payload, _, f) => f.await, + (graph::PartialDigestType::Unknown, object_f, payload_f) => { + object_f.or_else(|_| payload_f).await } } } diff --git a/crates/spfs/src/storage/pinned/repository.rs b/crates/spfs/src/storage/pinned/repository.rs index 9f2958d84..58cab676d 100644 --- a/crates/spfs/src/storage/pinned/repository.rs +++ b/crates/spfs/src/storage/pinned/repository.rs @@ -78,8 +78,14 @@ where self.inner.walk_objects(root) } - async fn resolve_full_digest(&self, partial: &encoding::PartialDigest) -> Result { - self.inner.resolve_full_digest(partial).await + async fn resolve_full_digest( + &self, + partial: &encoding::PartialDigest, + partial_digest_type: graph::PartialDigestType, + ) -> Result { + self.inner + .resolve_full_digest(partial, partial_digest_type) + .await } } diff --git a/crates/spfs/src/storage/repository.rs b/crates/spfs/src/storage/repository.rs index 469124d1f..54d35280e 100644 --- a/crates/spfs/src/storage/repository.rs +++ b/crates/spfs/src/storage/repository.rs @@ -72,7 +72,7 @@ pub trait Repository: // this information? A new type could be added to wrap FoundDigest with // a variant that doesn't have the type information. Note how resolving // a tag above does not determine the type, or require the item exists. - self.resolve_full_digest(&partial) + self.resolve_full_digest(&partial, graph::PartialDigestType::Unknown) .await .map(|found_digest| found_digest.into_digest()) } diff --git a/crates/spfs/src/storage/tar/repository.rs b/crates/spfs/src/storage/tar/repository.rs index 38e448e86..579ad588c 100644 --- a/crates/spfs/src/storage/tar/repository.rs +++ b/crates/spfs/src/storage/tar/repository.rs @@ -240,8 +240,14 @@ impl graph::DatabaseView for TarRepository { self.repo.walk_objects(root) } - async fn resolve_full_digest(&self, partial: &encoding::PartialDigest) -> Result { - self.repo.resolve_full_digest(partial).await + async fn resolve_full_digest( + &self, + partial: &encoding::PartialDigest, + partial_digest_type: graph::PartialDigestType, + ) -> Result { + self.repo + .resolve_full_digest(partial, partial_digest_type) + .await } } diff --git a/crates/spfs/src/sync.rs b/crates/spfs/src/sync.rs index 3ac389a9e..4515472bd 100644 --- a/crates/spfs/src/sync.rs +++ b/crates/spfs/src/sync.rs @@ -251,7 +251,10 @@ impl<'src, 'dst> Syncer<'src, 'dst> { &self, partial: encoding::PartialDigest, ) -> Result { - let res = self.src.resolve_full_digest(&partial).await; + let res = self + .src + .resolve_full_digest(&partial, graph::PartialDigestType::Unknown) + .await; let found_digest = match res { Err(err) if self.policy.check_existing_objects() => { // there is a chance that this digest points to an existing object in @@ -263,7 +266,7 @@ impl<'src, 'dst> Syncer<'src, 'dst> { // on environments without checking what is or is not in the destination // first self.dest - .resolve_full_digest(&partial) + .resolve_full_digest(&partial, graph::PartialDigestType::Unknown) .await .map_err(|_| err) } diff --git a/crates/spfs/src/tracking/env.rs b/crates/spfs/src/tracking/env.rs index 2866d3c21..dd6e74c3c 100644 --- a/crates/spfs/src/tracking/env.rs +++ b/crates/spfs/src/tracking/env.rs @@ -14,7 +14,7 @@ use serde::Deserialize; use super::tag::TagSpec; use crate::runtime::{LiveLayer, SpecApiVersion}; -use crate::{Error, Result, encoding}; +use crate::{Error, Result, encoding, graph}; #[cfg(test)] #[path = "./env_test.rs"] @@ -244,7 +244,7 @@ impl EnvSpecItem { match self { Self::TagSpec(spec) => repo.resolve_tag(spec).await.map(|t| t.target), Self::PartialDigest(part) => repo - .resolve_full_digest(part) + .resolve_full_digest(part, graph::PartialDigestType::Unknown) .await .map(|found_digest| found_digest.into_digest()), Self::Digest(digest) => Ok(*digest), From b8315621afa452084251f21aa91241f5d38bc578 Mon Sep 17 00:00:00 2001 From: J Robert Ray Date: Tue, 11 Nov 2025 14:25:00 -0800 Subject: [PATCH 3/4] Prevent config from changing during this test If a parallel test changes the config to use legacy encoding then this test will fail when trying to write a layer with blob annotation. Signed-off-by: J Robert Ray --- crates/spfs/src/check_test.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/crates/spfs/src/check_test.rs b/crates/spfs/src/check_test.rs index 129b193d5..58c0c7ed2 100644 --- a/crates/spfs/src/check_test.rs +++ b/crates/spfs/src/check_test.rs @@ -210,6 +210,8 @@ impl CheckReporter for &DebugReporter { /// The check should complete successfully and report a missing payload. #[rstest] #[tokio::test] +// This test just needs the config to not change while it is running. +#[serial_test::serial(config)] async fn check_missing_annotation_payload(#[future] tmprepo: TempRepo) { init_logging(); let tmprepo = tmprepo.await; From 9553f01ad2ba45295c1623424d0ae18569047eca Mon Sep 17 00:00:00 2001 From: J Robert Ray Date: Tue, 11 Nov 2025 15:30:07 -0800 Subject: [PATCH 4/4] Test partial digest resolution with blob objects It order for the test to be able to create a blob object [on disk], add a `write_object_unchecked` operation that bypasses the check that the object being written is not a blob. Signed-off-by: J Robert Ray --- crates/spfs/src/graph/database.rs | 22 ++++ crates/spfs/src/proto/defs/database.proto | 11 ++ crates/spfs/src/proto/result.rs | 4 + crates/spfs/src/server/database.rs | 13 +++ crates/spfs/src/storage/database_test.rs | 103 ++++++++++++++++++ .../spfs/src/storage/fallback/repository.rs | 11 ++ crates/spfs/src/storage/fs/database.rs | 16 +++ crates/spfs/src/storage/handle.rs | 20 ++++ crates/spfs/src/storage/pinned/repository.rs | 13 +++ crates/spfs/src/storage/proxy/repository.rs | 11 ++ crates/spfs/src/storage/rpc/database.rs | 16 +++ crates/spfs/src/storage/tar/repository.rs | 12 ++ 12 files changed, 252 insertions(+) diff --git a/crates/spfs/src/graph/database.rs b/crates/spfs/src/graph/database.rs index beaf7c3c1..35525b7bf 100644 --- a/crates/spfs/src/graph/database.rs +++ b/crates/spfs/src/graph/database.rs @@ -404,7 +404,21 @@ impl Database for &T { #[async_trait::async_trait] pub trait DatabaseExt: Send + Sync { /// Write an object to the database, for later retrieval. + /// + /// It is not permitted to write blob objects. async fn write_object(&self, obj: &FlatObject) -> Result<()>; + + /// Write an object to the database, for later retrieval. + /// + /// # Safety + /// + /// This function does not check the type of the object being written. It + /// is expected that the caller will not write a blob object except in + /// specific cases, such as in test code. + async unsafe fn write_object_unchecked( + &self, + obj: &FlatObject, + ) -> Result<()>; } #[async_trait::async_trait] @@ -412,4 +426,12 @@ impl DatabaseExt for &T { async fn write_object(&self, obj: &FlatObject) -> Result<()> { DatabaseExt::write_object(&**self, obj).await } + + async unsafe fn write_object_unchecked( + &self, + obj: &FlatObject, + ) -> Result<()> { + // Safety: transitive unsafe call + unsafe { DatabaseExt::write_object_unchecked(&**self, obj).await } + } } diff --git a/crates/spfs/src/proto/defs/database.proto b/crates/spfs/src/proto/defs/database.proto index 902062f7c..754581e11 100644 --- a/crates/spfs/src/proto/defs/database.proto +++ b/crates/spfs/src/proto/defs/database.proto @@ -84,6 +84,16 @@ message WriteObjectResponse{ } } +message WriteObjectUncheckedRequest{ + Object object = 1; +} +message WriteObjectUncheckedResponse{ + oneof result { + Error error = 1; + Ok ok = 2; + } +} + message RemoveObjectRequest{ Digest digest = 1; } @@ -112,6 +122,7 @@ service DatabaseService { rpc IterObjects(IterObjectsRequest) returns (stream IterObjectsResponse); rpc WalkObjects(WalkObjectsRequest) returns (stream WalkObjectsResponse); rpc WriteObject(WriteObjectRequest) returns (WriteObjectResponse); + rpc WriteObjectUnchecked(WriteObjectUncheckedRequest) returns (WriteObjectUncheckedResponse); rpc RemoveObject(RemoveObjectRequest) returns (RemoveObjectResponse); rpc RemoveObjectIfOlderThan(RemoveObjectIfOlderThanRequest) returns (RemoveObjectIfOlderThanResponse); } diff --git a/crates/spfs/src/proto/result.rs b/crates/spfs/src/proto/result.rs index c299e7ce0..46ca4d502 100644 --- a/crates/spfs/src/proto/result.rs +++ b/crates/spfs/src/proto/result.rs @@ -120,6 +120,10 @@ rpc_result!( g::walk_objects_response::WalkObjectsItem ); rpc_result!(g::WriteObjectResponse, g::write_object_response::Result); +rpc_result!( + g::WriteObjectUncheckedResponse, + g::write_object_unchecked_response::Result +); rpc_result!(g::RemoveObjectResponse, g::remove_object_response::Result); rpc_result!( g::RemoveObjectIfOlderThanResponse, diff --git a/crates/spfs/src/server/database.rs b/crates/spfs/src/server/database.rs index 72ac409fd..6f9fe2cb0 100644 --- a/crates/spfs/src/server/database.rs +++ b/crates/spfs/src/server/database.rs @@ -104,6 +104,19 @@ impl proto::database_service_server::DatabaseService for DatabaseService { Ok(Response::new(result)) } + async fn write_object_unchecked( + &self, + request: Request, + ) -> Result, Status> { + let request = request.into_inner(); + let object = proto::handle_error!(request.object.try_into()); + // Safety: in spirit this generated trait method would be unsafe, and + // this would be a transitive unsafe call. + unsafe { proto::handle_error!(self.repo.write_object_unchecked(&object).await) }; + let result = proto::WriteObjectUncheckedResponse::ok(proto::Ok {}); + Ok(Response::new(result)) + } + async fn remove_object( &self, request: Request, diff --git a/crates/spfs/src/storage/database_test.rs b/crates/spfs/src/storage/database_test.rs index 6af386fc9..030903d91 100644 --- a/crates/spfs/src/storage/database_test.rs +++ b/crates/spfs/src/storage/database_test.rs @@ -6,6 +6,7 @@ use std::borrow::Cow; use rstest::rstest; +use crate::encoding::PartialDigest; use crate::fixtures::*; use crate::graph; use crate::prelude::*; @@ -40,3 +41,105 @@ async fn test_object_existence( let actual = tmprepo.has_object(digest).await; assert!(!actual, "object should not exist after being removed"); } + +#[rstest] +#[case::fs(tmprepo("fs"))] +#[case::tar(tmprepo("tar"))] +#[cfg_attr(feature = "server", case::rpc(tmprepo("rpc")))] +#[tokio::test] +async fn resolve_partial_digest_with_blob_object_file_present( + #[case] + #[future] + tmprepo: TempRepo, +) { + let tmprepo = tmprepo.await; + + let test_data = b"test data\n"; + + let payload = tmprepo + .commit_payload(Box::pin(test_data.as_slice())) + .await + .expect("failed to commit payload data"); + + let partial_digest = PartialDigest::from(&payload.as_bytes()[..8]); + + // First test baseline behavior without blob object file present + + // PartialDigestType::Unknown + { + let resolved = tmprepo + .resolve_full_digest(&partial_digest, graph::PartialDigestType::Unknown) + .await + .expect("failed to resolve partial digest"); + + assert_eq!(*resolved.digest(), payload); + assert!(matches!(resolved, graph::FoundDigest::Payload(_))); + } + + // PartialDigestType::Payload + { + let resolved = tmprepo + .resolve_full_digest(&partial_digest, graph::PartialDigestType::Payload) + .await + .expect("failed to resolve partial digest"); + + assert_eq!(*resolved.digest(), payload); + assert!(matches!(resolved, graph::FoundDigest::Payload(_))); + } + + // PartialDigestType::Object + { + let _ = tmprepo + .resolve_full_digest(&partial_digest, graph::PartialDigestType::Object) + .await + .expect_err("no object with this digest should exist"); + } + + // Write the blob object file + { + let blob = graph::Blob::new(payload, test_data.len() as u64); + + // Safety: we are writing a blob object for the purposes of this test + unsafe { + tmprepo + .write_object_unchecked(&blob) + .await + .expect("failed to write blob object data"); + } + } + + // Then test behavior with blob object file present + + // PartialDigestType::Unknown + { + let resolved = tmprepo + .resolve_full_digest(&partial_digest, graph::PartialDigestType::Unknown) + .await + .expect("failed to resolve partial digest"); + + assert_eq!(*resolved.digest(), payload); + assert!(matches!(resolved, graph::FoundDigest::Payload(_))); + } + + // PartialDigestType::Payload + { + let resolved = tmprepo + .resolve_full_digest(&partial_digest, graph::PartialDigestType::Payload) + .await + .expect("failed to resolve partial digest"); + + assert_eq!(*resolved.digest(), payload); + assert!(matches!(resolved, graph::FoundDigest::Payload(_))); + } + + // PartialDigestType::Object + { + let resolved = tmprepo + .resolve_full_digest(&partial_digest, graph::PartialDigestType::Object) + .await + .expect("failed to resolve partial digest"); + + assert_eq!(*resolved.digest(), payload); + assert!(matches!(resolved, graph::FoundDigest::Object(_))); + } +} diff --git a/crates/spfs/src/storage/fallback/repository.rs b/crates/spfs/src/storage/fallback/repository.rs index c1bd8a217..835c65748 100644 --- a/crates/spfs/src/storage/fallback/repository.rs +++ b/crates/spfs/src/storage/fallback/repository.rs @@ -264,6 +264,17 @@ impl graph::DatabaseExt for FallbackProxy { self.primary.write_object(obj).await?; Ok(()) } + + async unsafe fn write_object_unchecked( + &self, + obj: &graph::FlatObject, + ) -> Result<()> { + // Safety: transitive unsafe call + unsafe { + self.primary.write_object_unchecked(obj).await?; + } + Ok(()) + } } #[async_trait::async_trait] diff --git a/crates/spfs/src/storage/fs/database.rs b/crates/spfs/src/storage/fs/database.rs index 3cdbfb9e8..575d163ae 100644 --- a/crates/spfs/src/storage/fs/database.rs +++ b/crates/spfs/src/storage/fs/database.rs @@ -82,6 +82,14 @@ impl graph::DatabaseExt for super::MaybeOpenFsRepository { async fn write_object(&self, obj: &graph::FlatObject) -> Result<()> { self.opened().await?.write_object(obj).await } + + async unsafe fn write_object_unchecked( + &self, + obj: &graph::FlatObject, + ) -> Result<()> { + // Safety: transitive unsafe call + unsafe { self.opened().await?.write_object_unchecked(obj).await } + } } #[async_trait::async_trait] @@ -239,6 +247,14 @@ impl graph::DatabaseExt for super::OpenFsRepository { return Err("writing blob objects is not permitted".into()); }; + // Safety: we checked that the object is not a blob above + unsafe { self.write_object_unchecked(obj).await } + } + + async unsafe fn write_object_unchecked( + &self, + obj: &graph::FlatObject, + ) -> Result<()> { let digest = obj.digest()?; let filepath = self.objects.build_digest_path(&digest); if filepath.exists() { diff --git a/crates/spfs/src/storage/handle.rs b/crates/spfs/src/storage/handle.rs index c0b3278e1..2be12198f 100644 --- a/crates/spfs/src/storage/handle.rs +++ b/crates/spfs/src/storage/handle.rs @@ -339,6 +339,16 @@ impl DatabaseExt for RepositoryHandle { async fn write_object(&self, obj: &graph::FlatObject) -> Result<()> { each_variant!(self, repo, { repo.write_object(obj).await }) } + + async unsafe fn write_object_unchecked( + &self, + obj: &graph::FlatObject, + ) -> Result<()> { + each_variant!(self, repo, { + // Safety: transitive unsafe call + unsafe { repo.write_object_unchecked(obj).await } + }) + } } impl Address for Arc { @@ -523,4 +533,14 @@ impl DatabaseExt for Arc { async fn write_object(&self, obj: &graph::FlatObject) -> Result<()> { each_variant!(&**self, repo, { repo.write_object(obj).await }) } + + async unsafe fn write_object_unchecked( + &self, + obj: &graph::FlatObject, + ) -> Result<()> { + each_variant!(&**self, repo, { + // Safety: transitive unsafe call + unsafe { repo.write_object_unchecked(obj).await } + }) + } } diff --git a/crates/spfs/src/storage/pinned/repository.rs b/crates/spfs/src/storage/pinned/repository.rs index 58cab676d..83b9cbe61 100644 --- a/crates/spfs/src/storage/pinned/repository.rs +++ b/crates/spfs/src/storage/pinned/repository.rs @@ -119,6 +119,19 @@ where // on pinned repositories self.inner.write_object(obj).await } + + async unsafe fn write_object_unchecked( + &self, + obj: &graph::FlatObject, + ) -> Result<()> { + // objects are stored by digest, not time, and so can still + // be safely written to a past repository view. In practice, + // this allows some recovery and sync operations to still function + // on pinned repositories + // + // Safety: transitive unsafe call + unsafe { self.inner.write_object_unchecked(obj).await } + } } #[async_trait::async_trait] diff --git a/crates/spfs/src/storage/proxy/repository.rs b/crates/spfs/src/storage/proxy/repository.rs index eb2d1a92c..74a70f726 100644 --- a/crates/spfs/src/storage/proxy/repository.rs +++ b/crates/spfs/src/storage/proxy/repository.rs @@ -200,6 +200,17 @@ impl graph::DatabaseExt for ProxyRepository { self.primary.write_object(obj).await?; Ok(()) } + + async unsafe fn write_object_unchecked( + &self, + obj: &graph::FlatObject, + ) -> Result<()> { + // Safety: transitive unsafe call + unsafe { + self.primary.write_object_unchecked(obj).await?; + } + Ok(()) + } } pub(crate) async fn payload_size(repo: R, digest: encoding::Digest) -> Result diff --git a/crates/spfs/src/storage/rpc/database.rs b/crates/spfs/src/storage/rpc/database.rs index 4aebc795b..cec328d04 100644 --- a/crates/spfs/src/storage/rpc/database.rs +++ b/crates/spfs/src/storage/rpc/database.rs @@ -114,4 +114,20 @@ impl graph::DatabaseExt for super::RpcRepository { .to_result()?; Ok(()) } + + async unsafe fn write_object_unchecked( + &self, + obj: &graph::FlatObject, + ) -> Result<()> { + let request = proto::WriteObjectUncheckedRequest { + object: Some(obj.into()), + }; + self.db_client + .clone() + .write_object_unchecked(request) + .await? + .into_inner() + .to_result()?; + Ok(()) + } } diff --git a/crates/spfs/src/storage/tar/repository.rs b/crates/spfs/src/storage/tar/repository.rs index 579ad588c..f6c28253b 100644 --- a/crates/spfs/src/storage/tar/repository.rs +++ b/crates/spfs/src/storage/tar/repository.rs @@ -282,6 +282,18 @@ impl graph::DatabaseExt for TarRepository { self.up_to_date.store(false, Ordering::Release); Ok(()) } + + async unsafe fn write_object_unchecked( + &self, + obj: &graph::FlatObject, + ) -> Result<()> { + // Safety: transitive unsafe call + unsafe { + self.repo.write_object_unchecked(obj).await?; + } + self.up_to_date.store(false, Ordering::Release); + Ok(()) + } } #[async_trait::async_trait]