Support for the referrers api (oci compliance tests fully passing) (#397

)

* change how manifests are handled:
  * before: they were special blobs
  * now: they're a different object in the DB, not written on disk
* compute digest on AsyncRead instead of blocking
* add referrers API endpoint (very basic, no pagination)

.github/workflows/pr-tests.yaml

@@ -99,7 +99,7 @@ jobs:
           OCI_NAMESPACE: oci-conformance/distribution-test
           OCI_TEST_PULL: 1
           OCI_TEST_PUSH: 1
-          OCI_TEST_CONTENT_DISCOVERY: 0
+          OCI_TEST_CONTENT_DISCOVERY: 1
           OCI_TEST_CONTENT_MANAGEMENT: 1
           OCI_HIDE_SKIPPED_WORKFLOWS: 0
           OCI_DEBUG: 0

Cargo.toml

@@ -65,7 +65,7 @@ async-trait = "0.1.74"
 walkdir = "2.0"
 rand = "0.8"
 humansize = "2.1"
-sqlx = { version = "0.8", features = ["runtime-tokio", "migrate", "sqlite"] }
+sqlx = { version = "0.8", features = ["runtime-tokio", "migrate", "sqlite", "json"] }
 oci-spec = "0.7.0"
 oci-client = "0.14.0"
 

migrations/01_initial.sql

@@ -1,40 +1,28 @@
 CREATE TABLE "blob" (
-    "digest" varchar NOT NULL PRIMARY KEY,
-    "size" integer NOT NULL,
-    "is_manifest" boolean NOT NULL,
-    "last_accessed" integer NOT NULL DEFAULT (unixepoch())
-);
-CREATE TABLE "blob_blob_association" (
-    "parent_digest" varchar NOT NULL,
-    "child_digest" varchar NOT NULL,
-    PRIMARY KEY ("parent_digest", "child_digest"),
-    FOREIGN KEY ("parent_digest")
-        REFERENCES "blob" ("digest")
-        ON DELETE CASCADE
-        ON UPDATE CASCADE,
-    FOREIGN KEY ("child_digest")
-        REFERENCES "blob" ("digest")
-);
+    "digest" TEXT NOT NULL PRIMARY KEY,
+    "size" INTEGER NOT NULL,
+    "last_accessed" INTEGER NOT NULL DEFAULT (unixepoch())
+) STRICT;
+CREATE TABLE "manifest" (
+    "digest" TEXT NOT NULL PRIMARY KEY,
+    "last_accessed" INTEGER NOT NULL DEFAULT (unixepoch()),
+    "json" BLOB NOT NULL,
+    "blob" BLOB NOT NULL
+) STRICT;
 CREATE TABLE "blob_upload" (
     "uuid" TEXT NOT NULL PRIMARY KEY,
-    "offset" integer NOT NULL,
-    "updated_at" timestamp_text NOT NULL DEFAULT CURRENT_TIMESTAMP,
-    "repo" varchar NOT NULL
-);
+    "offset" INTEGER NOT NULL,
+    "updated_at" TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "repo" TEXT NOT NULL
+) STRICT;
 CREATE TABLE "repo_blob_association" (
-    "repo_name" varchar NOT NULL,
-    "blob_digest" varchar NOT NULL,
-    PRIMARY KEY ("repo_name", "blob_digest"),
-    FOREIGN KEY ("blob_digest")
-        REFERENCES "blob" ("digest")
-        ON DELETE CASCADE
-);
+    "repo_name" TEXT NOT NULL,
+    "blob_digest" TEXT NOT NULL,
+    PRIMARY KEY ("repo_name", "blob_digest")
+) STRICT;
 CREATE TABLE "tag" (
-    "tag" varchar NOT NULL,
-    "repo" varchar NOT NULL,
-    "manifest_digest" varchar NOT NULL,
-    CONSTRAINT "IDX_repo_tag" PRIMARY KEY ("repo", "tag"),
-    FOREIGN KEY ("repo", "manifest_digest")
-        REFERENCES "repo_blob_association" ("repo_name", "blob_digest")
-        ON DELETE CASCADE
-);
+    "tag" TEXT NOT NULL,
+    "repo" TEXT NOT NULL,
+    "manifest_digest" TEXT NOT NULL,
+    CONSTRAINT "IDX_repo_tag" PRIMARY KEY ("repo", "tag")
+) STRICT;

src/registry/digest.rs

@@ -1,4 +1,3 @@
-use std::io::Read;
 use std::{fmt, io};
 
 use lazy_static::lazy_static;
@@ -7,6 +6,7 @@ use serde::{Deserialize, Serialize};
 use sha2::digest::OutputSizeUser;
 use sha2::{Digest as ShaDigest, Sha256};
 use thiserror::Error;
+use tokio::io::{AsyncRead, AsyncReadExt};
 
 // Buffer size for SHA2 hashing
 const BUFFER_SIZE: usize = 1024 * 1024;
@@ -72,17 +72,24 @@ impl Digest {
         &self.0[7..]
     }
 
-    pub fn digest_sha256<R: Read>(mut reader: R) -> io::Result<Digest> {
-        Self::digest::<Sha256, _>(&mut reader)
+    pub async fn digest_sha256<R: AsyncRead + Unpin>(mut reader: R) -> io::Result<Digest> {
+        Self::digest::<Sha256, _>(&mut reader).await
+    }
+
+    pub fn digest_sha256_slice(slice: &[u8]) -> Digest {
+        let hash = hex::encode(Sha256::digest(slice));
+        Self(format!("sha256:{hash}"))
     }
 
     #[allow(clippy::self_named_constructors)]
-    pub fn digest<D: ShaDigest + Default, R: Read>(reader: &mut R) -> io::Result<Digest> {
+    pub async fn digest<D: ShaDigest + Default, R: AsyncRead + Unpin>(
+        reader: &mut R,
+    ) -> io::Result<Digest> {
         let mut digest = D::default();
-        let mut buffer = [0u8; BUFFER_SIZE];
+        let mut buffer = vec![0u8; BUFFER_SIZE];
         let mut n = BUFFER_SIZE;
         while n == BUFFER_SIZE {
-            n = reader.read(&mut buffer)?;
+            n = reader.read(&mut buffer).await?;
             digest.update(&buffer[..n]);
         }
         let hash = hex::encode(digest.finalize());
@@ -100,17 +107,20 @@ impl Digest {
     pub fn as_str(&self) -> &str {
         &self.0
     }
+
+    pub fn into_string(self) -> String {
+        self.0
+    }
 }
 
 #[cfg(test)]
 mod test {
-    use std::io::BufReader;
 
     use crate::registry::digest::Digest;
 
     #[test]
     fn sha256_digest_test() {
-        let result = Digest::digest_sha256(BufReader::new("hello world".as_bytes())).unwrap();
+        let result = Digest::digest_sha256_slice("hello world".as_bytes());
         assert_eq!(
             &result.0,
             "sha256:b94d27b9934d3e08a52e52d7da7dabfac484efe37a5380ee9088f7ace2efcde9"
@@ -119,7 +129,7 @@ mod test {
 
     #[test]
     fn sha256_digest_empty_test() {
-        let result = Digest::digest_sha256(BufReader::new("".as_bytes())).unwrap();
+        let result = Digest::digest_sha256_slice("".as_bytes());
         assert_eq!(
             result.0,
             "sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855".to_string()
@@ -128,10 +138,8 @@ mod test {
 
     #[test]
     fn sha256_digest_brown_fox_test() {
-        let result = Digest::digest_sha256(BufReader::new(
-            "the quick brown fox jumps over the lazy dog".as_bytes(),
-        ))
-        .unwrap();
+        let result =
+            Digest::digest_sha256_slice("the quick brown fox jumps over the lazy dog".as_bytes());
         assert_eq!(
             result.0,
             "sha256:05c6e08f1d9fdafa03147fcb8f82f124c76d2f70e3d989dc8aadb5e7d7450bec".to_string()

src/registry/manifest.rs

@@ -1,8 +1,9 @@
 use std::borrow::Cow;
+use std::collections::HashMap;
 
 use anyhow::Result;
 use lazy_static::lazy_static;
-use oci_spec::image::{ImageIndex, ImageManifest, MediaType};
+use oci_spec::image::{Descriptor, ImageIndex, ImageManifest, MediaType};
 use regex::Regex;
 use serde::{Deserialize, Serialize};
 
@@ -101,6 +102,30 @@ pub mod manifest_media_type {
 }
 
 impl OCIManifest {
+    #[inline]
+    pub fn subject(&self) -> Option<Descriptor> {
+        match self {
+            OCIManifest::V2(m2) => m2.subject(),
+            OCIManifest::List(list) => list.subject(),
+        }
+        .clone()
+    }
+    #[inline]
+    pub fn artifact_type(&self) -> Option<MediaType> {
+        match self {
+            OCIManifest::V2(m2) => m2.artifact_type(),
+            OCIManifest::List(list) => list.artifact_type(),
+        }
+        .clone()
+    }
+    #[inline]
+    pub fn annotations(&self) -> &Option<HashMap<String, String>> {
+        match self {
+            OCIManifest::V2(m2) => m2.annotations(),
+            OCIManifest::List(list) => list.annotations(),
+        }
+    }
+
     /// Returns a Vector of the digests of all assets referenced in the Manifest
     /// With the exception of digests for "foreign blobs"
     pub fn get_local_asset_digests(&self) -> Vec<String> {
@@ -109,13 +134,7 @@ impl OCIManifest {
                 let mut digests: Vec<String> = m2
                     .layers()
                     .iter()
-                    .filter(|l| {
-                        l.media_type()
-                            != &MediaType::Other(
-                                "application/vnd.docker.image.rootfs.foreign.diff.tar.gzip"
-                                    .to_string(),
-                            )
-                    })
+                    .filter(|l| layer_is_distributable(l.media_type()))
                     .map(|x| x.digest().to_string())
                     .collect();
                 digests.push(m2.config().digest().to_string());
@@ -141,6 +160,17 @@ impl OCIManifest {
     }
 }
 
+pub fn layer_is_distributable(layer: &MediaType) -> bool {
+    let non_distributable = [
+        MediaType::ImageLayerNonDistributable,
+        MediaType::ImageLayerNonDistributableGzip,
+        MediaType::ImageLayerNonDistributableZstd,
+        MediaType::Other("application/vnd.docker.image.rootfs.foreign.diff.tar.gzip".to_string()),
+    ];
+
+    !non_distributable.contains(layer)
+}
+
 #[cfg(test)]
 mod test {
 

src/registry/mod.rs

@@ -59,7 +59,6 @@ impl From<StorageBackendError> for RegistryError {
             }
             StorageBackendError::InvalidContentRange => Self::InvalidContentRange,
             StorageBackendError::InvalidDigest => Self::InvalidDigest,
-            StorageBackendError::InvalidManifest(_msg) => Self::InvalidManifest,
             StorageBackendError::InvalidName(name) => Self::InvalidName(name),
             StorageBackendError::Io(e) => {
                 tracing::error!("Internal IO error: {e:?}");