Initial implementation of HTTP registry client

commit-id:d95e5f26

Cargo.toml

@@ -52,7 +52,7 @@ deno_task_shell = "0.13"
 derive_builder = "0.12"
 directories = "5"
 dunce = "1"
-fs4 = "0.7"
+fs4 = { version = "0.7", features = ["tokio"] }
 futures = { version = "0.3", default-features = false, features = ["std", "async-await"] }
 gix = "0.54"
 glob = "0.3"

scarb/src/core/registry/client/http.rs

@@ -0,0 +1,160 @@
+use crate::core::registry::client::RegistryClient;
+use crate::core::registry::index::{IndexConfig, IndexRecords};
+use crate::core::{Config, Package, PackageId, PackageName, SourceId};
+use crate::flock::{FileLockGuard, Filesystem};
+use anyhow::{Context, Result};
+use async_trait::async_trait;
+use fs4::tokio::AsyncFileExt;
+use futures::StreamExt;
+use reqwest::StatusCode;
+use std::path::PathBuf;
+use std::sync::Arc;
+use tokio::fs::OpenOptions;
+use tokio::io;
+use tokio::io::BufWriter;
+use tokio::sync::OnceCell;
+use tracing::{debug, trace};
+
+// TODO(mkaput): Honour ETag and Last-Modified headers.
+// TODO(mkaput): Progressbar.
+// TODO(mkaput): Request timeout.
+
+pub struct HttpRegistryClient<'c> {
+    source_id: SourceId,
+    config: &'c Config,
+    cached_index_config: OnceCell<IndexConfig>,
+    dl_fs: Filesystem<'c>,
+}
+
+impl<'c> HttpRegistryClient<'c> {
+    pub fn new(source_id: SourceId, config: &'c Config) -> Result<Self> {
+        let dl_fs = config
+            .dirs()
+            .registry_dir()
+            .into_child("dl")
+            .into_child(source_id.ident());
+
+        Ok(Self {
+            source_id,
+            config,
+            cached_index_config: Default::default(),
+            dl_fs,
+        })
+    }
+
+    async fn index_config(&self) -> Result<&IndexConfig> {
+        // TODO(mkaput): Cache config locally, honouring ETag and Last-Modified headers.
+
+        async fn load(source_id: SourceId, config: &Config) -> Result<IndexConfig> {
+            let index_config_url = source_id
+                .url
+                .join(IndexConfig::WELL_KNOWN_PATH)
+                .expect("Registry config URL should always be valid.");
+            debug!("fetching registry config: {index_config_url}");
+
+            let index_config = config
+                .http()?
+                .get(index_config_url)
+                .send()
+                .await?
+                .error_for_status()?
+                .json::<IndexConfig>()
+                .await?;
+
+            trace!(index_config = %serde_json::to_string(&index_config).unwrap());
+
+            Ok(index_config)
+        }
+
+        self.cached_index_config
+            .get_or_try_init(|| load(self.source_id, self.config))
+            .await
+            .context("failed to fetch registry config")
+    }
+}
+
+#[async_trait]
+impl<'c> RegistryClient for HttpRegistryClient<'c> {
+    fn is_offline(&self) -> bool {
+        false
+    }
+
+    async fn get_records(&self, package: PackageName) -> Result<Option<Arc<IndexRecords>>> {
+        let index_config = self.index_config().await?;
+        let records_url = index_config.index.expand(package.into())?;
+
+        let response = self
+            .config
+            .http()?
+            .get(records_url)
+            .send()
+            .await?
+            .error_for_status();
+
+        if let Err(err) = &response {
+            if let Some(status) = err.status() {
+                if status == StatusCode::NOT_FOUND {
+                    return Ok(None);
+                }
+            }
+        }
+
+        let records = response?
+            .json()
+            .await
+            .context("failed to deserialize index records")?;
+
+        Ok(Some(Arc::new(records)))
+    }
+
+    async fn is_downloaded(&self, _package: PackageId) -> bool {
+        // TODO(mkaput): Cache downloaded packages.
+        false
+    }
+
+    async fn download(&self, package: PackageId) -> Result<PathBuf> {
+        let dl_url = self.index_config().await?.dl.expand(package.into())?;
+
+        let response = self
+            .config
+            .http()?
+            .get(dl_url)
+            .send()
+            .await?
+            .error_for_status()?;
+
+        let output_path = self.dl_fs.path_existent()?.join(package.tarball_name());
+        let output_file = OpenOptions::new()
+            .read(true)
+            .write(true)
+            .truncate(true)
+            .create(true)
+            .open(&output_path)
+            .await
+            .with_context(|| format!("failed to open: {output_path}"))?;
+
+        output_file
+            .lock_exclusive()
+            .with_context(|| format!("failed to lock file: {output_path}"))?;
+
+        let mut stream = response.bytes_stream();
+        let mut writer = BufWriter::new(output_file);
+        while let Some(chunk) = stream.next().await {
+            let chunk = chunk.context("failed to read response chunk")?;
+            io::copy_buf(&mut &*chunk, &mut writer)
+                .await
+                .context("failed to save response chunk on disk")?;
+        }
+
+        Ok(output_path.into_std_path_buf())
+    }
+
+    async fn supports_publish(&self) -> Result<bool> {
+        // TODO(mkaput): Publishing to HTTP registries is not implemented yet.
+        Ok(false)
+    }
+
+    async fn publish(&self, _package: Package, _tarball: FileLockGuard) -> Result<()> {
+        todo!("Publishing to HTTP registries is not implemented yet.")
+    }
+}

scarb/src/core/registry/index/config.rs

@@ -43,6 +43,10 @@ pub struct IndexConfig {
     pub index: TemplateUrl,
 }
 
+impl IndexConfig {
+    pub const WELL_KNOWN_PATH: &'_ str = "config.json";
+}
+
 #[derive(Copy, Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)]
 #[serde(into = "u8", try_from = "u8")]
 pub struct IndexVersion;

scarb/tests/http_registry.rs

@@ -0,0 +1,101 @@
+use assert_fs::prelude::*;
+use assert_fs::TempDir;
+use indoc::indoc;
+use std::fs;
+use std::time::Duration;
+
+use scarb_test_support::command::Scarb;
+use scarb_test_support::project_builder::{Dep, DepBuilder, ProjectBuilder};
+use scarb_test_support::registry::http::HttpRegistry;
+
+#[test]
+fn usage() {
+    let mut registry = HttpRegistry::serve();
+    registry.publish(|t| {
+        ProjectBuilder::start()
+            .name("bar")
+            .version("1.0.0")
+            .lib_cairo(r#"fn f() -> felt252 { 0 }"#)
+            .build(t);
+    });
+
+    let t = TempDir::new().unwrap();
+    ProjectBuilder::start()
+        .name("foo")
+        .version("0.1.0")
+        .dep("bar", Dep.version("1").registry(&registry))
+        .lib_cairo(r#"fn f() -> felt252 { bar::f() }"#)
+        .build(&t);
+
+    // FIXME(mkaput): Why are verbose statuses not appearing here?
+    Scarb::quick_snapbox()
+        .arg("fetch")
+        .current_dir(&t)
+        .timeout(Duration::from_secs(10))
+        .assert()
+        .success()
+        .stdout_matches(indoc! {r#"
+        [..] Downloading bar v1.0.0 ([..])
+        "#});
+}
+
+#[test]
+fn not_found() {
+    let mut registry = HttpRegistry::serve();
+    registry.publish(|t| {
+        // Publish a package so that the directory hierarchy is created.
+        // Note, however, that we declare a dependency on baZ.
+        ProjectBuilder::start()
+            .name("bar")
+            .version("1.0.0")
+            .lib_cairo(r#"fn f() -> felt252 { 0 }"#)
+            .build(t);
+    });
+
+    let t = TempDir::new().unwrap();
+    ProjectBuilder::start()
+        .name("foo")
+        .version("0.1.0")
+        .dep("baz", Dep.version("1").registry(&registry))
+        .build(&t);
+
+    Scarb::quick_snapbox()
+        .arg("fetch")
+        .current_dir(&t)
+        .timeout(Duration::from_secs(10))
+        .assert()
+        .failure()
+        .stdout_matches(indoc! {r#"
+        error: package not found in registry: baz ^1 (registry+http://[..])
+        "#});
+}
+
+#[test]
+fn missing_config_json() {
+    let registry = HttpRegistry::serve();
+    fs::remove_file(registry.child("config.json")).unwrap();
+
+    let t = TempDir::new().unwrap();
+    ProjectBuilder::start()
+        .name("foo")
+        .version("0.1.0")
+        .dep("baz", Dep.version("1").registry(&registry))
+        .build(&t);
+
+    Scarb::quick_snapbox()
+        .arg("fetch")
+        .current_dir(&t)
+        .timeout(Duration::from_secs(10))
+        .assert()
+        .failure()
+        .stdout_matches(indoc! {r#"
+        error: failed to lookup for `baz ^1 (registry+http://[..])` in registry: registry+http://[..]
+
+        Caused by:
+            0: failed to fetch registry config
+            1: HTTP status client error (404 Not Found) for url (http://[..]/config.json)
+        "#});
+}
+
+// TODO(mkaput): Test errors properly when package is in index, but tarball is missing.
+// TODO(mkaput): Test interdependencies.

utils/scarb-test-support/src/registry/http.rs

@@ -0,0 +1,79 @@
+use assert_fs::fixture::ChildPath;
+use assert_fs::prelude::*;
+use std::fmt;
+use std::path::Path;
+
+use assert_fs::TempDir;
+use once_cell::sync::Lazy;
+use serde_json::json;
+use tokio::runtime;
+
+use crate::registry::local::LocalRegistry;
+use crate::simple_http_server::SimpleHttpServer;
+
+pub struct HttpRegistry {
+    local: LocalRegistry,
+    url: String,
+
+    // This needs to be stored here so that it's dropped properly.
+    server: SimpleHttpServer,
+}
+
+impl HttpRegistry {
+    pub fn serve() -> Self {
+        // Keep a global multi-threading runtime to contain all running servers in one shared
+        // thread pool, while maintaining synchronous nature of tests.
+        static RUNTIME: Lazy<runtime::Runtime> = Lazy::new(|| {
+            runtime::Builder::new_multi_thread()
+                .worker_threads(1)
+                .enable_all()
+                .build()
+                .unwrap()
+        });
+
+        let local = LocalRegistry::create();
+        let server = {
+            let _guard = RUNTIME.enter();
+            SimpleHttpServer::serve(local.t.path().to_owned())
+        };
+        let url = server.url();
+
+        let config = json!({
+            "version": 1,
+            "dl": format!("{url}{{package}}-{{version}}.tar.zst"),
+            "index": format!("{url}index/{{prefix}}/{{package}}.json")
+        });
+        local
+            .t
+            .child("config.json")
+            .write_str(&serde_json::to_string(&config).unwrap())
+            .unwrap();
+
+        Self { local, url, server }
+    }
+
+    pub fn publish(&mut self, f: impl FnOnce(&TempDir)) -> &mut Self {
+        self.local.publish(f);
+        self
+    }
+
+    /// Enable this when writing tests to see what requests are being made in the test.
+    pub fn debug_log_requests(&self) {
+        self.server.log_requests(true);
+    }
+}
+
+impl PathChild for HttpRegistry {
+    fn child<P>(&self, path: P) -> ChildPath
+    where
+        P: AsRef<Path>,
+    {
+        self.local.t.child(path)
+    }
+}
+
+impl fmt::Display for HttpRegistry {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        fmt::Display::fmt(&self.url, f)
+    }
+}

utils/scarb-test-support/src/registry/mod.rs

@@ -1 +1,2 @@
+pub mod http;
 pub mod local;