Bugfix: annotation blob value syncing (#1265)

Bugfix for annotation blob value syncing, adds sync test for annotation blobs

Signed-off-by: David Gilligan-Cook <dcook@imageworks.com>

Author: dcookspi

crates/spfs/src/sync.rs

@@ -398,7 +398,10 @@ impl<'src, 'dst> Syncer<'src, 'dst> {
         match annotation.value() {
             AnnotationValue::String(_) => Ok(SyncAnnotationResult::InternalValue),
             AnnotationValue::Blob(digest) => {
-                if !self.processed_digests.insert(*digest) {
+                if self.processed_digests.contains(&digest) {
+                    // do not insert here because annotation blobs
+                    // share a digest with payloads which must be
+                    // visited at least once to sync
                     return Ok(SyncAnnotationResult::Duplicate);
                 }
                 self.reporter.visit_annotation(&annotation);

crates/spfs/src/sync_test.rs

@@ -11,7 +11,7 @@ use super::Syncer;
 use crate::config::Config;
 use crate::fixtures::*;
 use crate::prelude::*;
-use crate::{Error, encoding, storage, tracking};
+use crate::{Error, encoding, graph, storage, tracking};
 
 #[rstest]
 #[tokio::test]
@@ -133,6 +133,94 @@ async fn test_sync_ref(
     assert!(repo_a.has_object(layer.digest().unwrap()).await);
 }
 
+#[rstest]
+#[case::fs(tmprepo("fs"), tmprepo("fs"))]
+#[case::tar(tmprepo("tar"), tmprepo("tar"))]
+#[cfg_attr(feature = "server", case::rpc(tmprepo("rpc"), tmprepo("rpc")))]
+#[tokio::test]
+// This test just needs the config to not change while it is running.
+#[serial_test::serial(config)]
+async fn test_sync_ref_including_annotation_blob(
+    #[case]
+    #[future]
+    repo_a: TempRepo,
+    #[case]
+    #[future]
+    repo_b: TempRepo,
+    tmpdir: tempfile::TempDir,
+) {
+    use std::borrow::Cow;
+
+    use crate::graph::AnnotationValue;
+
+    init_logging();
+    let repo_a = repo_a.await;
+    let repo_b = repo_b.await;
+
+    let src_dir = tmpdir.path().join("source");
+    ensure(src_dir.join("dir/file.txt"), "hello");
+    ensure(src_dir.join("dir2/otherfile.txt"), "hello2");
+    ensure(src_dir.join("dir//dir/dir/file.txt"), "hello, world");
+
+    let manifest = crate::Committer::new(&repo_a)
+        .commit_dir(src_dir.as_path())
+        .await
+        .unwrap();
+
+    // Set up an annotation in a blob for this test
+    let value = "this is a test annotation blob for syncing";
+    let annotation_digest = repo_a
+        .commit_blob(Box::pin(std::io::Cursor::new(
+            value.to_owned().into_bytes(),
+        )))
+        .await
+        .unwrap();
+    let annotation_value = AnnotationValue::Blob(Cow::Owned(annotation_digest));
+    let layer_with_annotation = graph::Layer::new_with_manifest_and_annotation(
+        manifest.to_graph_manifest().digest().unwrap(),
+        "test key",
+        annotation_value,
+    );
+    repo_a.write_object(&layer_with_annotation).await.unwrap();
+
+    let platform = repo_a
+        .create_platform(layer_with_annotation.digest().unwrap().into())
+        .await
+        .unwrap();
+    let tag = tracking::TagSpec::parse("testing").unwrap();
+    repo_a
+        .push_tag(&tag, &platform.digest().unwrap())
+        .await
+        .unwrap();
+
+    Syncer::new(&repo_a, &repo_b)
+        .sync_ref("testing")
+        .await
+        .expect("failed to sync ref");
+
+    assert!(repo_b.read_ref("testing").await.is_ok());
+    assert!(repo_b.has_object(platform.digest().unwrap()).await);
+    assert!(
+        repo_b
+            .has_object(layer_with_annotation.digest().unwrap())
+            .await
+    );
+    assert!(repo_b.has_object(annotation_digest).await);
+
+    Syncer::new(&repo_b, &repo_a)
+        .sync_ref("testing")
+        .await
+        .expect("failed to sync back");
+
+    assert!(repo_a.read_ref("testing").await.is_ok());
+    assert!(
+        repo_a
+            .has_object(layer_with_annotation.digest().unwrap())
+            .await
+    );
+    assert!(repo_a.has_object(annotation_digest).await);
+}
+
 #[rstest]
 #[case::fs(tmprepo("fs"), tmprepo("fs"))]
 #[case::tar(tmprepo("tar"), tmprepo("tar"))]