api: Update 'missing_content' to give upload endpoint(s)

lann · lann · commit b000eb6bac40 · 2023-07-24T14:46:43.000-04:00
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/crates/api/Cargo.toml b/crates/api/Cargo.toml
@@ -11,3 +11,4 @@ warg-crypto = { workspace = true }
 serde = { workspace = true }
 serde_with = { workspace = true }
 thiserror = { workspace = true }
+itertools = { workspace = true }
diff --git a/crates/api/src/v1/package.rs b/crates/api/src/v1/package.rs
@@ -21,6 +21,26 @@ pub enum ContentSource {
     },
 }
 
+/// Represents the supported kinds of content upload endpoints.
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(tag = "type", rename_all = "camelCase")]
+pub enum UploadEndpoint {
+    /// Content may be uploaded via HTTP POST to the given URL.
+    /// If the endpoint responds with "201 Created" and a Location header, that
+    /// header's value will be the content source.
+    HttpPost {
+        /// The URL to POST content to.
+        url: String,
+    },
+}
+
+/// Information about missing content.
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct MissingContent {
+    /// Upload endpoint(s) that may be used to provide missing content.
+    pub upload: Vec<UploadEndpoint>,
+}
+
 /// Represents a request to publish a record to a package log.
 #[derive(Serialize, Deserialize)]
 #[serde(rename = "camelCase")]
@@ -45,13 +65,13 @@ pub struct PackageRecord {
 }
 
 impl PackageRecord {
-    /// Gets the missing content digests of the record.
-    pub fn missing_content(&self) -> &[AnyHash] {
+    /// Gets the missing content of the record.
+    pub fn missing_content(&self) -> impl Iterator<Item = (&AnyHash, &MissingContent)> {
         match &self.state {
             PackageRecordState::Sourcing {
                 missing_content, ..
-            } => missing_content,
-            _ => &[],
+            } => itertools::Either::Left(missing_content.iter()),
+            _ => itertools::Either::Right(std::iter::empty()),
         }
     }
 }
@@ -69,7 +89,7 @@ pub enum PackageRecordState {
     #[serde(rename_all = "camelCase")]
     Sourcing {
         /// The digests of the missing content.
-        missing_content: Vec<AnyHash>,
+        missing_content: HashMap<AnyHash, MissingContent>,
     },
     /// The package record is processing.
     #[serde(rename_all = "camelCase")]
diff --git a/crates/client/src/api.rs b/crates/client/src/api.rs
@@ -301,15 +301,9 @@ impl Client {
     /// Uploads package content to the registry.
     pub async fn upload_content(
         &self,
-        log_id: &LogId,
-        record_id: &RecordId,
-        digest: &AnyHash,
+        url: &str,
         content: impl Into<Body>,
     ) -> Result<String, ClientError> {
-        let url = self
-            .url
-            .join(&paths::package_record_content(log_id, record_id, digest))
-            .unwrap();
         tracing::debug!("uploading content to `{url}`");
 
         let response = self.client.post(url).body(content).send().await?;
diff --git a/crates/client/src/lib.rs b/crates/client/src/lib.rs
@@ -13,7 +13,10 @@ use storage::{
 use thiserror::Error;
 use warg_api::v1::{
     fetch::{FetchError, FetchLogsRequest, FetchLogsResponse},
-    package::{PackageError, PackageRecord, PackageRecordState, PublishRecordRequest},
+    package::{
+        MissingContent, PackageError, PackageRecord, PackageRecordState, PublishRecordRequest,
+        UploadEndpoint,
+    },
     proof::{ConsistencyRequest, InclusionRequest},
 };
 use warg_crypto::{
@@ -156,34 +159,33 @@ impl<R: RegistryStorage, C: ContentStorage> Client<R, C> {
                 })
             })?;
 
-        let missing = record.missing_content();
-        if !missing.is_empty() {
-            // Upload the missing content
-            // TODO: parallelize this
-            for digest in record.missing_content() {
-                self.api
-                    .upload_content(
-                        &log_id,
-                        &record.id,
-                        digest,
-                        Body::wrap_stream(self.content.load_content(digest).await?.ok_or_else(
-                            || ClientError::ContentNotFound {
-                                digest: digest.clone(),
-                            },
-                        )?),
-                    )
-                    .await
-                    .map_err(|e| match e {
-                        api::ClientError::Package(PackageError::Rejection(reason)) => {
-                            ClientError::PublishRejected {
-                                id: package.id.clone(),
-                                record_id: record.id.clone(),
-                                reason,
-                            }
+        // TODO: parallelize this
+        for (digest, MissingContent { upload }) in record.missing_content() {
+            // Upload the missing content, if the registry supports it
+            let Some(UploadEndpoint::HttpPost {url}) = upload.first() else {
+                continue;
+            };
+
+            self.api
+                .upload_content(
+                    url,
+                    Body::wrap_stream(self.content.load_content(digest).await?.ok_or_else(
+                        || ClientError::ContentNotFound {
+                            digest: digest.clone(),
+                        },
+                    )?),
+                )
+                .await
+                .map_err(|e| match e {
+                    api::ClientError::Package(PackageError::Rejection(reason)) => {
+                        ClientError::PublishRejected {
+                            id: package.id.clone(),
+                            record_id: record.id.clone(),
+                            reason,
                         }
-                        _ => e.into(),
-                    })?;
-            }
+                    }
+                    _ => e.into(),
+                })?;
         }
 
         Ok(record.id)
diff --git a/crates/server/openapi.yaml b/crates/server/openapi.yaml
@@ -700,12 +700,9 @@ components:
           enum: [sourcing]
           example: sourcing
         missingContent:
-          type: array
-          description: The array of content digests that are missing for the package record.
-          minItems: 1
-          maxItems: 128
-          items:
-            "$ref": "#/components/schemas/AnyHash"
+          "$ref": "#/components/schemas/MissingContentMap"
+          description: The missing content for the package record.
+          minProperties: 1
     ProcessingRecord:
       type: object
       description: A record that is being processed.
@@ -825,6 +822,44 @@ components:
           description: The algorithm-prefixed bytes of the signature (base64 encoded).
           pattern: ^[a-z0-9-]+:(?:[A-Za-z0-9+\/]{4})*(?:[A-Za-z0-9+\/]{4}|[A-Za-z0-9+\/]{3}=|[A-Za-z0-9+\/]{2}={2})$
           example: "ecdsa-p256:MEUCIQCzWZBW6ux9LecP66Y+hjmLZTP/hZVz7puzlPTXcRT2wwIgQZO7nxP0nugtw18MwHZ26ROFWcJmgCtKOguK031Y1D0="
+    MissingContentMap:
+      type: object
+      description: The map of content digest to missing content info.
+      patternProperties:
+        "^[a-z0-9-]+:[a-f0-9]+$":
+          "$ref": "#/components/schemas/MissingContent"
+      example:
+        ? "sha256:7d865e959b2466918c9863afca942d0fb89d7c9ac0c99bafc3749504ded9773"
+        : upload:
+            - type: httpPost
+              url: https://example.com/7d865e959b2466918c9863afca942d0fb89d7c9ac0c99bafc3749504ded9773
+    MissingContent:
+      description: Information about missing content.
+      properties:
+        upload:
+          description: Upload endpoint(s) for the missing content.
+          type: array 
+          items:
+            oneOf:
+              - "$ref": "#/components/schemas/HttpPostUpload"
+            discriminator:
+              propertyName: type
+              mapping:
+                httpPost: "#/components/schemas/HttpPostUpload"
+    HttpPostUpload:
+      type: object
+      description: A HTTP POST upload endpoint.
+      properties:
+        type:
+          type: string
+          description: The type of upload endpoint.
+          enum: [httpPost]
+          example: httpPost
+        url:
+          type: string
+          description: The URL of the upload endpoint.
+          example: https://example.com/contents.wasm
+          format: uri
     ContentSourceMap:
       type: object
       description: The map of content digest to sources.
diff --git a/crates/server/src/api/v1/package.rs b/crates/server/src/api/v1/package.rs
@@ -16,13 +16,14 @@ use axum::{
     Router,
 };
 use futures::StreamExt;
-use std::path::PathBuf;
 use std::sync::Arc;
+use std::{collections::HashMap, path::PathBuf};
 use tempfile::NamedTempFile;
 use tokio::io::AsyncWriteExt;
 use url::Url;
 use warg_api::v1::package::{
-    ContentSource, PackageError, PackageRecord, PackageRecordState, PublishRecordRequest,
+    ContentSource, MissingContent, PackageError, PackageRecord, PackageRecordState,
+    PublishRecordRequest, UploadEndpoint,
 };
 use warg_crypto::hash::{AnyHash, Sha256};
 use warg_protocol::{
@@ -91,6 +92,25 @@ impl Config {
             .unwrap()
             .to_string()
     }
+
+    fn build_missing_content<'a>(
+        &self,
+        missing_digests: impl IntoIterator<Item = &'a AnyHash>,
+    ) -> HashMap<AnyHash, MissingContent> {
+        missing_digests
+            .into_iter()
+            .map(|digest| {
+                (
+                    digest.clone(),
+                    MissingContent {
+                        upload: vec![UploadEndpoint::HttpPost {
+                            url: self.content_url(digest),
+                        }],
+                    },
+                )
+            })
+            .collect()
+    }
 }
 
 struct PackageApiError(PackageError);
@@ -231,13 +251,12 @@ async fn publish_record(
         ));
     }
 
+    let missing_content = config.build_missing_content(missing);
     Ok((
         StatusCode::ACCEPTED,
         Json(PackageRecord {
             id: record_id,
-            state: PackageRecordState::Sourcing {
-                missing_content: missing.into_iter().cloned().collect(),
-            },
+            state: PackageRecordState::Sourcing { missing_content },
         }),
     ))
 }
@@ -254,12 +273,13 @@ async fn get_record(
         .await?;
 
     match record.status {
-        RecordStatus::MissingContent(missing) => Ok(Json(PackageRecord {
-            id: record_id,
-            state: PackageRecordState::Sourcing {
-                missing_content: missing,
-            },
-        })),
+        RecordStatus::MissingContent(missing) => {
+            let missing_content = config.build_missing_content(&missing);
+            Ok(Json(PackageRecord {
+                id: record_id,
+                state: PackageRecordState::Sourcing { missing_content },
+            }))
+        }
         // Validated is considered still processing until included in a checkpoint
         RecordStatus::Pending | RecordStatus::Validated => Ok(Json(PackageRecord {
             id: record_id,
@@ -275,11 +295,11 @@ async fn get_record(
                 .as_ref()
                 .contents()
                 .into_iter()
-                .map(|d| {
+                .map(|digest| {
                     (
-                        d.clone(),
+                        digest.clone(),
                         vec![ContentSource::Http {
-                            url: config.content_url(d),
+                            url: config.content_url(digest),
                         }],
                     )
                 })
@@ -363,7 +383,7 @@ async fn upload_content(
     {
         config
             .core_service
-            .submit_package_record(log_id, record_id.clone())
+            .submit_package_record(log_id, record_id)
             .await;
     }
 
