aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorNikolay Govorov <me@govorov.online>2026-07-16 01:00:47 +0100
committerNikolay Govorov <me@govorov.online>2026-07-16 08:36:29 +0100
commit08ad0701819267355df10b3904e601523096554e (patch)
treea01edcdebc794a9789e1b6f65d93b2e52a191553
parent353e3b6fe64659b33a6ca376f8c1e9aa4ae752d3 (diff)
downloadtar
tar.gz
tar.bz2
tar.lz
tar.xz
tar.zst
zip
Async typed hule-oci
Diffstat
-rw-r--r--Cargo.lock1+1 −0
-rw-r--r--crates/hule-oci/Cargo.toml1+1 −0
-rw-r--r--crates/hule-oci/src/error.rs119+119 −0
-rw-r--r--crates/hule-oci/src/lib.rs715+371 −344
-rw-r--r--crates/hule-oci/src/storage.rs213+213 −0
-rw-r--r--crates/hule/src/main.rs45+34 −11
6 files changed, 739 insertions, 355 deletions
diff --git a/Cargo.lock b/Cargo.lock
index cff42e0..d2d38d6 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -690,6 +690,7 @@ dependencies = [
"oci-client",
"serde_json",
"sha2 0.10.9",
+ "tokio",
"zstd",
]
diff --git a/crates/hule-oci/Cargo.toml b/crates/hule-oci/Cargo.toml
index 9fa92a9..c395abd 100644
--- a/crates/hule-oci/Cargo.toml
+++ b/crates/hule-oci/Cargo.toml
@@ -15,6 +15,7 @@ repository.workspace = true
[dependencies]
hule-hmi.workspace = true
serde_json.workspace = true
+tokio = { workspace = true, features = ["fs", "io-util", "rt"] }
oci-client = "0.17"
zstd = "0.13"
sha2 = "0.10"
diff --git a/crates/hule-oci/src/error.rs b/crates/hule-oci/src/error.rs
new file mode 100644
--- /dev/null
+++ b/crates/hule-oci/src/error.rs
@@ -0,0 +1,119 @@
+// SPDX-FileCopyrightText: 2026 Nikolay Govorov
+// SPDX-License-Identifier: Apache-2.0
+
+use oci_client::errors::OciDistributionError;
+use std::fmt;
+use std::path::PathBuf;
+
+#[derive(Debug)]
+pub enum Error {
+ Io(std::io::Error),
+ Json(serde_json::Error),
+ InvalidImage(hule_hmi::ParseError),
+ InvalidReference {
+ reference: String,
+ source: oci_client::ParseError,
+ },
+ Registry(OciDistributionError),
+ Task(tokio::task::JoinError),
+ InvalidImagePath(PathBuf),
+ MissingConfig(PathBuf),
+ ImageNotFound(String),
+ MissingManifest,
+ InvalidLayer(&'static str),
+ InvalidAnnotation {
+ name: &'static str,
+ value: String,
+ },
+ EmptyImageFile(PathBuf),
+ InvalidStoragePath {
+ kind: &'static str,
+ value: String,
+ },
+}
+
+impl fmt::Display for Error {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ match self {
+ Self::Io(error) => error.fmt(f),
+ Self::Json(error) => error.fmt(f),
+ Self::InvalidImage(error) => error.fmt(f),
+ Self::InvalidReference { reference, source } => {
+ write!(f, "invalid reference '{reference}': {source}")
+ }
+ Self::Registry(error) => error.fmt(f),
+ Self::Task(error) => write!(f, "background task failed: {error}"),
+ Self::InvalidImagePath(path) => {
+ write!(f, "{} has no file name", path.display())
+ }
+ Self::MissingConfig(path) => {
+ write!(f, "{} does not contain config.json", path.display())
+ }
+ Self::ImageNotFound(reference) => {
+ write!(f, "no local image tagged '{reference}'")
+ }
+ Self::MissingManifest => f.write_str("registry returned no image manifest"),
+ Self::InvalidLayer(reason) => write!(f, "invalid chunk layer: {reason}"),
+ Self::InvalidAnnotation { name, value } => {
+ write!(f, "invalid chunk {name} annotation '{value}'")
+ }
+ Self::EmptyImageFile(path) => write!(f, "{} is empty", path.display()),
+ Self::InvalidStoragePath { kind, value } => {
+ write!(f, "invalid {kind} '{value}'")
+ }
+ }
+ }
+}
+
+impl std::error::Error for Error {
+ fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+ match self {
+ Self::Io(error) => Some(error),
+ Self::Json(error) => Some(error),
+ Self::InvalidImage(error) => Some(error),
+ Self::InvalidReference { source, .. } => Some(source),
+ Self::Registry(error) => Some(error),
+ Self::Task(error) => Some(error),
+ Self::InvalidImagePath(_)
+ | Self::MissingConfig(_)
+ | Self::ImageNotFound(_)
+ | Self::MissingManifest
+ | Self::InvalidLayer(_)
+ | Self::InvalidAnnotation { .. }
+ | Self::EmptyImageFile(_)
+ | Self::InvalidStoragePath { .. } => None,
+ }
+ }
+}
+
+impl From<std::io::Error> for Error {
+ fn from(error: std::io::Error) -> Self {
+ Self::Io(error)
+ }
+}
+
+impl From<serde_json::Error> for Error {
+ fn from(error: serde_json::Error) -> Self {
+ Self::Json(error)
+ }
+}
+
+impl From<hule_hmi::ParseError> for Error {
+ fn from(error: hule_hmi::ParseError) -> Self {
+ Self::InvalidImage(error)
+ }
+}
+
+impl From<OciDistributionError> for Error {
+ fn from(error: OciDistributionError) -> Self {
+ Self::Registry(error)
+ }
+}
+
+impl From<tokio::task::JoinError> for Error {
+ fn from(error: tokio::task::JoinError) -> Self {
+ Self::Task(error)
+ }
+}
+
+pub type Result<T> = std::result::Result<T, Error>;
diff --git a/crates/hule-oci/src/lib.rs b/crates/hule-oci/src/lib.rs
index e53615b..990bd87 100644
--- a/crates/hule-oci/src/lib.rs
+++ b/crates/hule-oci/src/lib.rs
@@ -3,6 +3,9 @@
//! OCI image storage and registry operations for Hule machine images.
+mod error;
+mod storage;
+
use hule_hmi::MachineImage;
use oci_client::annotations::{ORG_OPENCONTAINERS_IMAGE_REF_NAME, ORG_OPENCONTAINERS_IMAGE_TITLE};
use oci_client::client::{ClientConfig, ClientProtocol, Config, ImageLayer};
@@ -12,9 +15,14 @@ use oci_client::manifest::{
use oci_client::secrets::RegistryAuth;
use oci_client::{Client, Reference};
use std::collections::{BTreeMap, BTreeSet};
-use std::io::Read;
use std::path::{Path, PathBuf};
-pub type Result<T> = std::result::Result<T, String>;
+use tokio::fs::{self, File};
+use tokio::io::AsyncReadExt;
+use tokio::task;
+
+pub use error::{Error, Result};
+pub use storage::Storage;
+
type R<T> = Result<T>;
// ---- local OCI-layout store -------------------------------------------
@@ -25,82 +33,6 @@ const HULE_CONFIG_MEDIA_TYPE: &str = "application/vnd.hule.machine.config.v1+jso
const ANNOTATION_CHUNK_OFFSET: &str = "io.hule.chunk.offset";
const ANNOTATION_CHUNK_LENGTH: &str = "io.hule.chunk.length";
-fn store_root() -> R<PathBuf> {
- let home = std::env::var("HOME").map_err(|_| "HOME is not set".to_string())?;
- Ok(PathBuf::from(home).join(".hule").join("store"))
-}
-
-fn blobs_dir() -> R<PathBuf> {
- Ok(store_root()?.join("blobs").join("sha256"))
-}
-
-fn materialized_dir(manifest_digest: &str) -> R<PathBuf> {
- Ok(store_root()?
- .join("materialized")
- .join(strip_sha256(manifest_digest)))
-}
-
-fn index_path() -> R<PathBuf> {
- Ok(store_root()?.join("index.json"))
-}
-
-fn strip_sha256(digest: &str) -> &str {
- digest.strip_prefix("sha256:").unwrap_or(digest)
-}
-
-fn sha256_hex(data: &[u8]) -> String {
- use sha2::{Digest, Sha256};
- let mut hasher = Sha256::new();
- hasher.update(data);
- format!("sha256:{:x}", hasher.finalize())
-}
-
-fn ensure_store() -> R<()> {
- std::fs::create_dir_all(blobs_dir()?).map_err(|e| e.to_string())?;
- std::fs::create_dir_all(store_root()?.join("materialized")).map_err(|e| e.to_string())?;
- let layout = store_root()?.join("oci-layout");
- if !layout.exists() {
- std::fs::write(&layout, br#"{"imageLayoutVersion":"1.0.0"}"#).map_err(|e| e.to_string())?;
- }
- Ok(())
-}
-
-/// Writes raw bytes as a content-addressed blob, no-op if already present.
-fn write_blob(data: &[u8]) -> R<String> {
- let digest = sha256_hex(data);
- let path = blobs_dir()?.join(strip_sha256(&digest));
- if !path.exists() {
- let tmp = path.with_extension("tmp");
- std::fs::write(&tmp, data).map_err(|e| e.to_string())?;
- std::fs::rename(&tmp, &path).map_err(|e| e.to_string())?;
- }
- Ok(digest)
-}
-
-fn read_blob(digest: &str) -> R<Vec<u8>> {
- std::fs::read(blobs_dir()?.join(strip_sha256(digest))).map_err(|e| e.to_string())
-}
-
-fn read_index() -> R<OciImageIndex> {
- let path = index_path()?;
- if !path.exists() {
- return Ok(OciImageIndex {
- schema_version: 2,
- media_type: None,
- manifests: vec![],
- artifact_type: None,
- annotations: None,
- });
- }
- let data = std::fs::read(&path).map_err(|e| e.to_string())?;
- serde_json::from_slice(&data).map_err(|e| e.to_string())
-}
-
-fn write_index(index: &OciImageIndex) -> R<()> {
- let data = serde_json::to_vec_pretty(index).map_err(|e| e.to_string())?;
- std::fs::write(index_path()?, data).map_err(|e| e.to_string())
-}
-
fn index_lookup(index: &OciImageIndex, reference: &str) -> Option<String> {
index
.manifests
@@ -138,79 +70,71 @@ fn index_set(index: &mut OciImageIndex, reference: &str, digest: &str, size: u64
});
}
-fn hardlink_or_copy(src: &Path, dest: &Path) -> R<()> {
- if std::fs::hard_link(src, dest).is_err() {
- std::fs::copy(src, dest).map_err(|e| e.to_string())?;
- }
- Ok(())
+/// Manages Hule images backed by a [`Storage`].
+#[derive(Clone, Debug)]
+pub struct ImageManager {
+ storage: Storage,
}
-fn write_at(file: &std::fs::File, offset: u64, data: &[u8]) -> R<()> {
- use std::os::unix::fs::FileExt;
- let mut written = 0usize;
- while written < data.len() {
- let n = file
- .write_at(&data[written..], offset + written as u64)
- .map_err(|e| e.to_string())?;
- if n == 0 {
- return Err("write_at wrote 0 bytes".to_string());
- }
- written += n;
+impl ImageManager {
+ pub fn new(storage: Storage) -> Self {
+ Self { storage }
}
- Ok(())
-}
-/// Splits `path` into fixed-size chunks, compresses each independently
-/// (own zstd frame, no state shared between chunks -- see the design notes
-/// in the project plan on why this must not be done the other way around),
-/// and writes each as its own blob. Returns one descriptor per chunk, all
-/// sharing a title annotation plus a chunk offset/length in terms of the
-/// *uncompressed* file so pull can reassemble it.
-fn write_file_chunks(path: &Path) -> R<Vec<OciDescriptor>> {
- let title = path
- .file_name()
- .ok_or_else(|| format!("{} has no file name", path.display()))?
- .to_string_lossy()
- .to_string();
- let mut file = std::fs::File::open(path).map_err(|e| e.to_string())?;
- let mut descriptors = Vec::new();
- let mut offset: u64 = 0;
- loop {
- let mut buf = Vec::with_capacity(CHUNK_SIZE);
- (&mut file)
- .take(CHUNK_SIZE as u64)
- .read_to_end(&mut buf)
- .map_err(|e| e.to_string())?;
- if buf.is_empty() {
- break;
- }
- let length = buf.len() as u64;
- let compressed = zstd::stream::encode_all(&buf[..], 0).map_err(|e| e.to_string())?;
- let digest = write_blob(&compressed)?;
-
- let mut annotations = BTreeMap::new();
- annotations.insert(ORG_OPENCONTAINERS_IMAGE_TITLE.to_string(), title.clone());
- annotations.insert(ANNOTATION_CHUNK_OFFSET.to_string(), offset.to_string());
- annotations.insert(ANNOTATION_CHUNK_LENGTH.to_string(), length.to_string());
-
- descriptors.push(OciDescriptor {
- media_type: HULE_CHUNK_MEDIA_TYPE.to_string(),
- digest,
- size: compressed.len() as i64,
- urls: None,
- annotations: Some(annotations),
- artifact_type: None,
- });
+ pub fn storage(&self) -> &Storage {
+ &self.storage
+ }
- offset += length;
- if length < CHUNK_SIZE as u64 {
- break;
+ /// Splits `path` into fixed-size chunks, compresses each independently
+ /// and writes each as its own blob.
+ async fn write_file_chunks(&self, path: &Path) -> R<Vec<OciDescriptor>> {
+ let title = path
+ .file_name()
+ .ok_or_else(|| Error::InvalidImagePath(path.to_path_buf()))?
+ .to_string_lossy()
+ .to_string();
+ let mut file = File::open(path).await?;
+ let mut descriptors = Vec::new();
+ let mut offset: u64 = 0;
+ loop {
+ let mut buf = Vec::with_capacity(CHUNK_SIZE);
+ (&mut file)
+ .take(CHUNK_SIZE as u64)
+ .read_to_end(&mut buf)
+ .await?;
+ if buf.is_empty() {
+ break;
+ }
+ let length = buf.len() as u64;
+ let compressed =
+ task::spawn_blocking(move || zstd::stream::encode_all(&buf[..], 0)).await??;
+ let compressed_len = compressed.len();
+ let (digest, _) = self.storage.write_blob(compressed).await?;
+
+ let mut annotations = BTreeMap::new();
+ annotations.insert(ORG_OPENCONTAINERS_IMAGE_TITLE.to_string(), title.clone());
+ annotations.insert(ANNOTATION_CHUNK_OFFSET.to_string(), offset.to_string());
+ annotations.insert(ANNOTATION_CHUNK_LENGTH.to_string(), length.to_string());
+
+ descriptors.push(OciDescriptor {
+ media_type: HULE_CHUNK_MEDIA_TYPE.to_string(),
+ digest,
+ size: compressed_len as i64,
+ urls: None,
+ annotations: Some(annotations),
+ artifact_type: None,
+ });
+
+ offset += length;
+ if length < CHUNK_SIZE as u64 {
+ break;
+ }
}
+ if descriptors.is_empty() {
+ return Err(Error::EmptyImageFile(path.to_path_buf()));
+ }
+ Ok(descriptors)
}
- if descriptors.is_empty() {
- return Err(format!("{} is empty", path.display()));
- }
- Ok(descriptors)
}
fn make_client(reference: &Reference) -> Client {
@@ -228,232 +152,335 @@ fn make_client(reference: &Reference) -> Client {
}
fn parse_reference(s: &str) -> R<Reference> {
- s.parse()
- .map_err(|e| format!("invalid reference '{s}': {e}"))
+ s.parse().map_err(|source| Error::InvalidReference {
+ reference: s.to_string(),
+ source,
+ })
}
-// ---- import / push / pull / run ---------------------------------------
+// ---- image operations -------------------------------------------------
+
+impl ImageManager {
+ /// Loads a prepared HMI image directory and optionally tags it.
+ pub async fn load(&self, path: &Path, reference: Option<&str>) -> R<String> {
+ let config_path = path.join("config.json");
+ match fs::metadata(&config_path).await {
+ Ok(metadata) if metadata.is_file() => {}
+ Ok(_) => return Err(Error::MissingConfig(path.to_path_buf())),
+ Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
+ return Err(Error::MissingConfig(path.to_path_buf()));
+ }
+ Err(error) => return Err(error.into()),
+ }
+ let config_bytes = fs::read(&config_path).await?;
+ MachineImage::from_json(&config_bytes)?;
-/// Imports an image directory into the local OCI layout and optionally tags it.
-pub async fn import(path: &Path, reference: Option<&str>) -> R<String> {
- ensure_store()?;
- let config_path = path.join("config.json");
- if !config_path.is_file() {
- return Err(format!("{} does not contain config.json", path.display()));
- }
- let config_bytes = std::fs::read(&config_path).map_err(|e| e.to_string())?;
- MachineImage::from_json(&config_bytes).map_err(|e| e.to_string())?;
-
- let mut entries: Vec<_> = std::fs::read_dir(path)
- .map_err(|e| e.to_string())?
- .collect::<std::result::Result<Vec<_>, _>>()
- .map_err(|e| e.to_string())?;
- entries.sort_by_key(|e| e.file_name());
-
- let mut layers = Vec::new();
- let mut sources: Vec<(String, PathBuf)> = Vec::new();
- for entry in entries {
- if !entry.file_type().map_err(|e| e.to_string())?.is_file() {
- continue;
+ let mut dir = fs::read_dir(path).await?;
+ let mut entries = Vec::new();
+ while let Some(entry) = dir.next_entry().await? {
+ entries.push(entry);
}
- let name = entry.file_name().to_string_lossy().to_string();
- if name == "config.json" {
- continue;
+ entries.sort_by_key(|entry| entry.file_name());
+
+ let mut layers = Vec::new();
+ let mut sources: Vec<(String, PathBuf)> = Vec::new();
+ for entry in entries {
+ if !entry.file_type().await?.is_file() {
+ continue;
+ }
+ let name = entry.file_name().to_string_lossy().to_string();
+ if name == "config.json" {
+ continue;
+ }
+ let file_path = entry.path();
+ layers.extend(self.write_file_chunks(&file_path).await?);
+ sources.push((name, file_path));
}
- let file_path = entry.path();
- layers.extend(write_file_chunks(&file_path)?);
- sources.push((name, file_path));
- }
- let config_digest = write_blob(&config_bytes)?;
- let config = OciDescriptor {
- media_type: HULE_CONFIG_MEDIA_TYPE.to_string(),
- digest: config_digest,
- size: config_bytes.len() as i64,
- urls: None,
- annotations: None,
- artifact_type: None,
- };
+ let config_size = config_bytes.len();
+ let (config_digest, _) = self.storage.write_blob(config_bytes).await?;
+ let config = OciDescriptor {
+ media_type: HULE_CONFIG_MEDIA_TYPE.to_string(),
+ digest: config_digest,
+ size: config_size as i64,
+ urls: None,
+ annotations: None,
+ artifact_type: None,
+ };
- let manifest = OciImageManifest {
- schema_version: 2,
- media_type: Some(OCI_IMAGE_MEDIA_TYPE.to_string()),
- config,
- layers,
- subject: None,
- artifact_type: None,
- annotations: None,
- };
- let manifest_bytes = serde_json::to_vec(&manifest).map_err(|e| e.to_string())?;
- let manifest_digest = write_blob(&manifest_bytes)?;
-
- let materialized = materialized_dir(&manifest_digest)?;
- std::fs::create_dir_all(&materialized).map_err(|e| e.to_string())?;
- for (title, source) in &sources {
- let dest = materialized.join(title);
- if !dest.exists() {
- hardlink_or_copy(source, &dest)?;
+ let manifest = OciImageManifest {
+ schema_version: 2,
+ media_type: Some(OCI_IMAGE_MEDIA_TYPE.to_string()),
+ config,
+ layers,
+ subject: None,
+ artifact_type: None,
+ annotations: None,
+ };
+ let manifest_bytes = serde_json::to_vec(&manifest)?;
+ let manifest_size = manifest_bytes.len();
+ let (manifest_digest, _) = self.storage.write_blob(manifest_bytes).await?;
+
+ self.storage.ensure_hmi_image(&manifest_digest).await?;
+ for (title, source) in &sources {
+ self.storage
+ .cache_hmi_file(&manifest_digest, title, source)
+ .await?;
}
- }
-
- if let Some(reference) = reference {
- let mut index = read_index()?;
- index_set(
- &mut index,
- reference,
- &manifest_digest,
- manifest_bytes.len() as u64,
- );
- write_index(&index)?;
- }
- Ok(manifest_digest)
-}
-/// Pushes a locally tagged image to its registry reference.
-pub async fn push(reference_str: &str) -> R<()> {
- ensure_store()?;
- let index = read_index()?;
- let manifest_digest = index_lookup(&index, reference_str)
- .ok_or_else(|| format!("no local image tagged '{reference_str}'"))?;
- let manifest_bytes = read_blob(&manifest_digest)?;
- let manifest: OciImageManifest =
- serde_json::from_slice(&manifest_bytes).map_err(|e| e.to_string())?;
-
- let reference = parse_reference(reference_str)?;
- let client = make_client(&reference);
- let auth = RegistryAuth::Anonymous;
-
- let mut layers = Vec::new();
- for descriptor in &manifest.layers {
- let data = read_blob(&descriptor.digest)?;
- layers.push(ImageLayer::new(
- data,
- descriptor.media_type.clone(),
- descriptor.annotations.clone(),
- ));
+ if let Some(reference) = reference {
+ let mut index = self.storage.read_index().await?;
+ index_set(
+ &mut index,
+ reference,
+ &manifest_digest,
+ manifest_size as u64,
+ );
+ self.storage.write_index(&index).await?;
+ }
+ Ok(manifest_digest)
}
- let config_bytes = read_blob(&manifest.config.digest)?;
- let config = Config::new(
- config_bytes,
- manifest.config.media_type.clone(),
- manifest.config.annotations.clone(),
- );
- client
- .push(&reference, &layers, config, &auth, Some(manifest))
- .await
- .map_err(|e| e.to_string())?;
+ /// Pushes a locally tagged image to its registry reference.
+ pub async fn push(&self, reference_str: &str) -> R<()> {
+ let index = self.storage.read_index().await?;
+ let manifest_digest = index_lookup(&index, reference_str)
+ .ok_or_else(|| Error::ImageNotFound(reference_str.to_string()))?;
+ let manifest_bytes = self.storage.read_blob(&manifest_digest).await?;
+ let manifest: OciImageManifest = serde_json::from_slice(&manifest_bytes)?;
+
+ let reference = parse_reference(reference_str)?;
+ let client = make_client(&reference);
+ let auth = RegistryAuth::Anonymous;
+
+ let mut layers = Vec::new();
+ for descriptor in &manifest.layers {
+ let data = self.storage.read_blob(&descriptor.digest).await?;
+ layers.push(ImageLayer::new(
+ data,
+ descriptor.media_type.clone(),
+ descriptor.annotations.clone(),
+ ));
+ }
+ let config_bytes = self.storage.read_blob(&manifest.config.digest).await?;
+ let config = Config::new(
+ config_bytes,
+ manifest.config.media_type.clone(),
+ manifest.config.annotations.clone(),
+ );
- Ok(())
-}
+ client
+ .push(&reference, &layers, config, &auth, Some(manifest))
+ .await?;
-/// Pulls an image into the local OCI layout and materializes its files.
-pub async fn pull(reference_str: &str) -> R<String> {
- ensure_store()?;
- let reference = parse_reference(reference_str)?;
- let client = make_client(&reference);
- let auth = RegistryAuth::Anonymous;
-
- let image_data = client
- .pull(&reference, &auth, vec![HULE_CHUNK_MEDIA_TYPE])
- .await
- .map_err(|e| e.to_string())?;
- let manifest = image_data
- .manifest
- .ok_or("registry returned no image manifest")?;
-
- // `client.pull()` fetches layers via `buffer_unordered`, so `image_data.layers` is in
- // completion order, NOT `manifest.layers` order -- do not zip the two. Each `ImageLayer`
- // already carries its own annotations, so it's self-describing without cross-referencing.
- for layer in &image_data.layers {
- write_blob(&layer.data)?;
+ Ok(())
}
- write_blob(&image_data.config.data)?;
- let manifest_bytes = serde_json::to_vec(&manifest).map_err(|e| e.to_string())?;
- let manifest_digest = write_blob(&manifest_bytes)?;
+ /// Pulls an image into the OCI layout and materializes it in the HMI store.
+ pub async fn pull(&self, reference_str: &str) -> R<String> {
+ let reference = parse_reference(reference_str)?;
+ let client = make_client(&reference);
+ let auth = RegistryAuth::Anonymous;
+
+ let mut image_data = client
+ .pull(&reference, &auth, vec![HULE_CHUNK_MEDIA_TYPE])
+ .await?;
+ let manifest = image_data.manifest.ok_or(Error::MissingManifest)?;
+
+ // `client.pull()` fetches layers via `buffer_unordered`, so `image_data.layers` is
+ // in completion order, not manifest order. Each layer carries its own annotations.
+ for layer in &mut image_data.layers {
+ let data = std::mem::take(&mut layer.data);
+ let (_, data) = self.storage.write_blob(data).await?;
+ layer.data = data;
+ }
+ let config_data = std::mem::take(&mut image_data.config.data);
+ let (_, config_data) = self.storage.write_blob(config_data).await?;
+ image_data.config.data = config_data;
+
+ let manifest_bytes = serde_json::to_vec(&manifest)?;
+ let manifest_size = manifest_bytes.len();
+ let (manifest_digest, _) = self.storage.write_blob(manifest_bytes).await?;
+ self.storage.ensure_hmi_image(&manifest_digest).await?;
+
+ let mut by_title: BTreeMap<String, Vec<(u64, ImageLayer)>> = BTreeMap::new();
+ for layer in image_data.layers {
+ let annotations = layer
+ .annotations
+ .as_ref()
+ .ok_or(Error::InvalidLayer("missing annotations"))?;
+ let title = annotations
+ .get(ORG_OPENCONTAINERS_IMAGE_TITLE)
+ .ok_or(Error::InvalidLayer("missing title annotation"))?;
+ let offset_value = annotations
+ .get(ANNOTATION_CHUNK_OFFSET)
+ .ok_or(Error::InvalidLayer("missing offset annotation"))?;
+ let offset: u64 = offset_value.parse().map_err(|_| Error::InvalidAnnotation {
+ name: "offset",
+ value: offset_value.clone(),
+ })?;
+ by_title
+ .entry(title.clone())
+ .or_default()
+ .push((offset, layer));
+ }
- let materialized = materialized_dir(&manifest_digest)?;
- std::fs::create_dir_all(&materialized).map_err(|e| e.to_string())?;
+ for (title, chunks) in by_title {
+ let mut decompressed_chunks = Vec::with_capacity(chunks.len());
+ for (offset, layer) in chunks {
+ let data = task::spawn_blocking(move || zstd::stream::decode_all(&layer.data[..]))
+ .await??;
+ decompressed_chunks.push((offset, data));
+ }
+ self.storage
+ .write_hmi_file(&manifest_digest, &title, decompressed_chunks)
+ .await?;
+ }
- let mut by_title: BTreeMap<String, Vec<(u64, &[u8])>> = BTreeMap::new();
- for layer in &image_data.layers {
- let annotations = layer
- .annotations
- .as_ref()
- .ok_or("chunk layer missing annotations")?;
- let title = annotations
- .get(ORG_OPENCONTAINERS_IMAGE_TITLE)
- .ok_or("chunk layer missing title annotation")?;
- let offset: u64 = annotations
- .get(ANNOTATION_CHUNK_OFFSET)
- .ok_or("chunk layer missing offset annotation")?
- .parse()
- .map_err(|_| "invalid chunk offset annotation".to_string())?;
- by_title
- .entry(title.clone())
- .or_default()
- .push((offset, &layer.data[..]));
+ let mut index = self.storage.read_index().await?;
+ index_set(
+ &mut index,
+ reference_str,
+ &manifest_digest,
+ manifest_size as u64,
+ );
+ self.storage.write_index(&index).await?;
+
+ Ok(manifest_digest)
}
- for (title, mut chunks) in by_title {
- chunks.sort_by_key(|(offset, _)| *offset);
- let file = std::fs::File::create(materialized.join(&title)).map_err(|e| e.to_string())?;
- for (offset, compressed) in chunks {
- let decompressed = zstd::stream::decode_all(compressed).map_err(|e| e.to_string())?;
- write_at(&file, offset, &decompressed)?;
- }
+ /// Returns whether a reference is present in the local OCI layout.
+ pub async fn contains(&self, reference_str: &str) -> R<bool> {
+ Ok(index_lookup(&self.storage.read_index().await?, reference_str).is_some())
}
- let mut index = read_index()?;
- index_set(
- &mut index,
- reference_str,
- &manifest_digest,
- manifest_bytes.len() as u64,
- );
- write_index(&index)?;
+ /// Materializes a locally tagged image into `destination` and returns config.json.
+ pub async fn materialize(&self, reference_str: &str, destination: &Path) -> R<Vec<u8>> {
+ let manifest_digest = index_lookup(&self.storage.read_index().await?, reference_str)
+ .ok_or_else(|| Error::ImageNotFound(reference_str.to_string()))?;
- Ok(manifest_digest)
-}
+ let manifest_bytes = self.storage.read_blob(&manifest_digest).await?;
+ let manifest: OciImageManifest = serde_json::from_slice(&manifest_bytes)?;
+ let config_bytes = self.storage.read_blob(&manifest.config.digest).await?;
-/// Returns whether a reference is present in the local OCI layout.
-pub fn contains(reference_str: &str) -> R<bool> {
- ensure_store()?;
- Ok(index_lookup(&read_index()?, reference_str).is_some())
-}
+ fs::create_dir_all(destination).await?;
+ fs::write(destination.join("config.json"), &config_bytes).await?;
-/// Materializes a locally tagged image into `destination` and returns config.json.
-pub fn materialize(reference_str: &str, destination: &Path) -> R<Vec<u8>> {
- ensure_store()?;
- let manifest_digest = index_lookup(&read_index()?, reference_str)
- .ok_or_else(|| format!("no local image tagged '{reference_str}'"))?;
-
- let manifest_bytes = read_blob(&manifest_digest)?;
- let manifest: OciImageManifest =
- serde_json::from_slice(&manifest_bytes).map_err(|e| e.to_string())?;
- let config_bytes = read_blob(&manifest.config.digest)?;
-
- std::fs::create_dir_all(destination).map_err(|e| e.to_string())?;
- std::fs::write(destination.join("config.json"), &config_bytes).map_err(|e| e.to_string())?;
-
- let materialized = materialized_dir(&manifest_digest)?;
- let mut titles: BTreeSet<String> = BTreeSet::new();
- for descriptor in &manifest.layers {
- if let Some(title) = descriptor
- .annotations
- .as_ref()
- .and_then(|a| a.get(ORG_OPENCONTAINERS_IMAGE_TITLE))
- {
- titles.insert(title.clone());
+ let mut titles: BTreeSet<String> = BTreeSet::new();
+ for descriptor in &manifest.layers {
+ if let Some(title) = descriptor
+ .annotations
+ .as_ref()
+ .and_then(|annotations| annotations.get(ORG_OPENCONTAINERS_IMAGE_TITLE))
+ {
+ titles.insert(title.clone());
+ }
}
- }
- for title in titles {
- let dest = destination.join(&title);
- if !dest.exists() {
- hardlink_or_copy(&materialized.join(&title), &dest)?;
+ for title in titles {
+ self.storage
+ .export_hmi_file(&manifest_digest, &title, &destination.join(&title))
+ .await?;
}
+
+ Ok(config_bytes)
}
+}
- Ok(config_bytes)
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use std::sync::atomic::{AtomicU64, Ordering};
+
+ const TEST_CONFIG: &[u8] = br#"{
+ "schemaVersion": 1,
+ "kind": "MachineImage",
+ "system": {
+ "os": "linux",
+ "name": "test",
+ "version": "1",
+ "architecture": "amd64"
+ },
+ "machine": {
+ "cpu": { "minimum": 1, "default": 1 },
+ "ram": { "minimum": 268435456, "default": 268435456 },
+ "boot": [{ "protocol": "firmware-disk/bios", "disk": "root" }],
+ "access": [{
+ "type": "ssh",
+ "port": 22,
+ "user": "root",
+ "auth": "empty-password"
+ }],
+ "network": { "mode": "dhcp" }
+ },
+ "disks": [{
+ "id": "root",
+ "format": "qcow2",
+ "path": "root.hmi",
+ "digest": "sha256:372409142c91c51316a7ab2b055596e145de56d99aa0f5eb110068876cef0be4",
+ "virtSize": 14,
+ "diskSize": 14
+ }]
+ }"#;
+
+ fn temporary_directory() -> PathBuf {
+ static NEXT_ID: AtomicU64 = AtomicU64::new(0);
+ std::env::temp_dir().join(format!(
+ "hule-oci-test-{}-{}",
+ std::process::id(),
+ NEXT_ID.fetch_add(1, Ordering::Relaxed)
+ ))
+ }
+
+ #[test]
+ fn manager_uses_separate_oci_and_hmi_roots() {
+ tokio::runtime::Builder::new_current_thread()
+ .build()
+ .unwrap()
+ .block_on(async {
+ let root = temporary_directory();
+ let source = root.join("source");
+ let oci_root = root.join("oci");
+ let hmi_root = root.join("hmi");
+ let destination = root.join("destination");
+ fs::create_dir_all(&source).await.unwrap();
+ fs::write(source.join("config.json"), TEST_CONFIG)
+ .await
+ .unwrap();
+ fs::write(source.join("root.hmi"), b"test hmi image")
+ .await
+ .unwrap();
+
+ let storage = Storage::open(&oci_root, &hmi_root).await.unwrap();
+ let images = ImageManager::new(storage);
+ let reference = "example.test/hule:tokio";
+ let digest = images.load(&source, Some(reference)).await.unwrap();
+
+ assert!(images.contains(reference).await.unwrap());
+ assert!(fs::try_exists(oci_root.join("index.json")).await.unwrap());
+ assert!(
+ fs::try_exists(
+ hmi_root
+ .join(digest.strip_prefix("sha256:").unwrap_or(&digest))
+ .join("root.hmi")
+ )
+ .await
+ .unwrap()
+ );
+
+ let config = images.materialize(reference, &destination).await.unwrap();
+ assert!(!config.is_empty());
+ assert_eq!(
+ fs::read(destination.join("root.hmi")).await.unwrap(),
+ b"test hmi image"
+ );
+
+ assert!(matches!(images.contains("missing").await, Ok(false)));
+ assert!(matches!(
+ images.materialize("missing", &destination).await,
+ Err(Error::ImageNotFound(reference)) if reference == "missing"
+ ));
+
+ fs::remove_dir_all(root).await.unwrap();
+ });
+ }
}
diff --git a/crates/hule-oci/src/storage.rs b/crates/hule-oci/src/storage.rs
new file mode 100644
--- /dev/null
+++ b/crates/hule-oci/src/storage.rs
@@ -0,0 +1,213 @@
+// SPDX-FileCopyrightText: 2026 Nikolay Govorov
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::{Error, Result};
+use oci_client::manifest::OciImageIndex;
+use std::io::SeekFrom;
+use std::path::{Component, Path, PathBuf};
+use tokio::fs::{self, File};
+use tokio::io::{AsyncSeekExt, AsyncWriteExt};
+use tokio::task;
+
+/// Filesystem storage for OCI content and materialized HMI images.
+///
+/// The roots are independent so callers can place content-addressed OCI data
+/// and large materialized disks on different filesystems.
+#[derive(Clone, Debug)]
+pub struct Storage {
+ oci_root: PathBuf,
+ hmi_root: PathBuf,
+}
+
+impl Storage {
+ /// Opens the storage and creates its filesystem layout when necessary.
+ pub async fn open(oci_root: impl Into<PathBuf>, hmi_root: impl Into<PathBuf>) -> Result<Self> {
+ let storage = Self {
+ oci_root: oci_root.into(),
+ hmi_root: hmi_root.into(),
+ };
+
+ fs::create_dir_all(&storage.hmi_root).await?;
+ fs::create_dir_all(storage.blobs_dir()).await?;
+
+ let layout = storage.oci_root.join("oci-layout");
+ if !fs::try_exists(&layout).await? {
+ fs::write(&layout, br#"{"imageLayoutVersion":"1.0.0"}"#).await?;
+ }
+
+ Ok(storage)
+ }
+
+ pub fn oci_root(&self) -> &Path {
+ &self.oci_root
+ }
+
+ pub fn hmi_root(&self) -> &Path {
+ &self.hmi_root
+ }
+
+ fn blobs_dir(&self) -> PathBuf {
+ self.oci_root.join("blobs").join("sha256")
+ }
+
+ fn index_path(&self) -> PathBuf {
+ self.oci_root.join("index.json")
+ }
+
+ fn hmi_image_dir(&self, manifest_digest: &str) -> Result<PathBuf> {
+ let digest = safe_component(strip_sha256(manifest_digest), "manifest digest")?;
+ Ok(self.hmi_root.join(digest))
+ }
+
+ fn hmi_file_path(&self, manifest_digest: &str, title: &str) -> Result<PathBuf> {
+ let title = safe_component(title, "image file name")?;
+ Ok(self.hmi_image_dir(manifest_digest)?.join(title))
+ }
+
+ fn blob_path(&self, digest: &str) -> Result<PathBuf> {
+ let digest = safe_component(strip_sha256(digest), "blob digest")?;
+ Ok(self.blobs_dir().join(digest))
+ }
+
+ pub(crate) async fn read_blob(&self, digest: &str) -> Result<Vec<u8>> {
+ Ok(fs::read(self.blob_path(digest)?).await?)
+ }
+
+ /// Writes raw bytes as a content-addressed blob, no-op if already present.
+ pub(crate) async fn write_blob<T>(&self, data: T) -> Result<(String, T)>
+ where
+ T: AsRef<[u8]> + Send + Sync + 'static,
+ {
+ let (digest, data) =
+ task::spawn_blocking(move || (sha256_hex(data.as_ref()), data)).await?;
+ let path = self.blob_path(&digest)?;
+ if !fs::try_exists(&path).await? {
+ let tmp = path.with_extension("tmp");
+ fs::write(&tmp, data.as_ref()).await?;
+ fs::rename(&tmp, &path).await?;
+ }
+ Ok((digest, data))
+ }
+
+ pub(crate) async fn read_index(&self) -> Result<OciImageIndex> {
+ let path = self.index_path();
+ if !fs::try_exists(&path).await? {
+ return Ok(OciImageIndex {
+ schema_version: 2,
+ media_type: None,
+ manifests: vec![],
+ artifact_type: None,
+ annotations: None,
+ });
+ }
+ let data = fs::read(&path).await?;
+ Ok(serde_json::from_slice(&data)?)
+ }
+
+ pub(crate) async fn write_index(&self, index: &OciImageIndex) -> Result<()> {
+ let data = serde_json::to_vec_pretty(index)?;
+ fs::write(self.index_path(), data).await?;
+ Ok(())
+ }
+
+ pub(crate) async fn ensure_hmi_image(&self, manifest_digest: &str) -> Result<()> {
+ fs::create_dir_all(self.hmi_image_dir(manifest_digest)?).await?;
+ Ok(())
+ }
+
+ pub(crate) async fn cache_hmi_file(
+ &self,
+ manifest_digest: &str,
+ title: &str,
+ source: &Path,
+ ) -> Result<()> {
+ let destination = self.hmi_file_path(manifest_digest, title)?;
+ if !fs::try_exists(&destination).await? {
+ Self::hardlink_or_copy(source, &destination).await?;
+ }
+ Ok(())
+ }
+
+ pub(crate) async fn write_hmi_file(
+ &self,
+ manifest_digest: &str,
+ title: &str,
+ mut chunks: Vec<(u64, Vec<u8>)>,
+ ) -> Result<()> {
+ chunks.sort_by_key(|(offset, _)| *offset);
+ let mut file = File::create(self.hmi_file_path(manifest_digest, title)?).await?;
+ for (offset, data) in chunks {
+ file.seek(SeekFrom::Start(offset)).await?;
+ file.write_all(&data).await?;
+ }
+ file.flush().await?;
+ Ok(())
+ }
+
+ pub(crate) async fn export_hmi_file(
+ &self,
+ manifest_digest: &str,
+ title: &str,
+ destination: &Path,
+ ) -> Result<()> {
+ let source = self.hmi_file_path(manifest_digest, title)?;
+ if !fs::try_exists(destination).await? {
+ Self::hardlink_or_copy(&source, destination).await?;
+ }
+ Ok(())
+ }
+
+ async fn hardlink_or_copy(source: &Path, destination: &Path) -> Result<()> {
+ if fs::hard_link(source, destination).await.is_err() {
+ fs::copy(source, destination).await?;
+ }
+ Ok(())
+ }
+}
+
+fn safe_component<'a>(value: &'a str, kind: &'static str) -> Result<&'a str> {
+ let mut components = Path::new(value).components();
+ match (components.next(), components.next()) {
+ (Some(Component::Normal(_)), None) => Ok(value),
+ _ => Err(Error::InvalidStoragePath {
+ kind,
+ value: value.to_string(),
+ }),
+ }
+}
+
+fn strip_sha256(digest: &str) -> &str {
+ digest.strip_prefix("sha256:").unwrap_or(digest)
+}
+
+fn sha256_hex(data: &[u8]) -> String {
+ use sha2::{Digest, Sha256};
+ let mut hasher = Sha256::new();
+ hasher.update(data);
+ format!("sha256:{:x}", hasher.finalize())
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn paths_cannot_escape_storage_roots() {
+ let storage = Storage {
+ oci_root: "oci".into(),
+ hmi_root: "hmi".into(),
+ };
+ assert!(matches!(
+ storage.blob_path("sha256:../../outside"),
+ Err(Error::InvalidStoragePath { .. })
+ ));
+ assert!(matches!(
+ storage.hmi_image_dir("sha256:../outside"),
+ Err(Error::InvalidStoragePath { .. })
+ ));
+ assert!(matches!(
+ storage.hmi_file_path("sha256:digest", "../outside"),
+ Err(Error::InvalidStoragePath { .. })
+ ));
+ }
+}
diff --git a/crates/hule/src/main.rs b/crates/hule/src/main.rs
index 644545b..cf9d5d9 100644
--- a/crates/hule/src/main.rs
+++ b/crates/hule/src/main.rs
@@ -1,7 +1,8 @@
// SPDX-FileCopyrightText: 2026 Nikolay Govorov
// SPDX-License-Identifier: Apache-2.0
-use std::path::Path;
+use std::fmt::Display;
+use std::path::{Path, PathBuf};
use std::process::exit;
use clap::{Parser, Subcommand};
@@ -60,19 +61,29 @@ enum MachineCommands {
},
}
-fn die(msg: impl AsRef<str>) -> ! {
- eprintln!("hule: {}", msg.as_ref());
+fn die(msg: impl Display) -> ! {
+ eprintln!("hule: {msg}");
exit(1);
}
-async fn cmd_run(reference: &str, port: u16) -> R<()> {
- if !hule_oci::contains(reference)? {
+async fn cmd_run(images: &hule_oci::ImageManager, reference: &str, port: u16) -> R<()> {
+ if !images
+ .contains(reference)
+ .await
+ .map_err(|error| error.to_string())?
+ {
eprintln!("hule: {reference} not found locally, pulling...");
- hule_oci::pull(reference).await?;
+ images
+ .pull(reference)
+ .await
+ .map_err(|error| error.to_string())?;
}
let scratch = std::env::temp_dir().join(format!("hule-run-{}", std::process::id()));
- let config = hule_oci::materialize(reference, &scratch)?;
+ let config = images
+ .materialize(reference, &scratch)
+ .await
+ .map_err(|error| error.to_string())?;
let image = MachineImage::from_json(&config).map_err(|e| e.to_string())?;
let hv = QemuHypervisor;
@@ -125,20 +136,30 @@ async fn cmd_run(reference: &str, port: u16) -> R<()> {
async fn main() {
let cli = Cli::parse();
+ let home = std::env::var("HOME").unwrap_or_else(|_| die("HOME is not set"));
+ let root = PathBuf::from(home).join(".hule");
+
+ let storage = hule_oci::Storage::open(root.join("oci"), root.join("hmi"))
+ .await
+ .unwrap_or_else(|e| die(e));
+
+ let images = hule_oci::ImageManager::new(storage);
+
match &cli.command {
Commands::Image { command } => match &command {
ImageCommands::Pull { name } => {
- let digest = hule_oci::pull(name).await.unwrap_or_else(|e| die(e));
+ let digest = images.pull(name).await.unwrap_or_else(|e| die(e));
eprintln!("hule: pulled {} ({digest})", name);
}
ImageCommands::Push { name } => {
- hule_oci::push(name).await.unwrap_or_else(|e| die(e));
+ images.push(name).await.unwrap_or_else(|e| die(e));
eprintln!("hule: pushed {}", name);
}
ImageCommands::Load { reference, image } => {
- let digest = hule_oci::import(Path::new(image), reference.as_deref())
+ let digest = images
+ .load(Path::new(image), reference.as_deref())
.await
.unwrap_or_else(|e| die(e));
@@ -152,7 +173,9 @@ async fn main() {
},
Commands::Machine { command } => match &command {
MachineCommands::Run { image, port } => {
- cmd_run(image, *port).await.unwrap_or_else(|e| die(e));
+ cmd_run(&images, image, *port)
+ .await
+ .unwrap_or_else(|e| die(e));
}
},
}
generated by Gilti