ironflow_artifacts/
lib.rs

1//! # ironflow-artifacts
2//!
3//! Blob storage for **ironflow** workflow artifacts: the bytes a step produces
4//! and a later step consumes.
5//!
6//! This crate owns the payloads only. Their metadata (name, MIME type, size,
7//! SHA-256, owning step) lives in `ironflow-store` and is the source of truth:
8//! a blob without a metadata row is never listed and never served.
9//!
10//! # Implementations
11//!
12//! | Store | Feature | Description |
13//! |-------|---------|-------------|
14//! | [`LocalBlobStore`](local::LocalBlobStore) | `artifact-local` (default) | Local filesystem, for development, CI and shared-volume deployments. |
15//! | [`S3BlobStore`](s3::S3BlobStore) | `storage-s3` | S3-compatible object store (AWS, MinIO, R2, GCS). |
16//!
17//! # Quick start
18//!
19//! ```no_run
20//! use ironflow_artifacts::prelude::*;
21//!
22//! # async fn example() -> Result<(), ArtifactError> {
23//! let store = LocalBlobStore::new("/var/lib/ironflow/artifacts");
24//!
25//! let key = storage_key(uuid::Uuid::now_v7(), uuid::Uuid::now_v7(), uuid::Uuid::now_v7());
26//! let digest = store.put(&key, stream_from_bytes(b"report".to_vec())).await?;
27//!
28//! println!("{} bytes, sha256 {}", digest.size_bytes, digest.sha256);
29//! # Ok(())
30//! # }
31//! ```
32
33pub mod blob_store;
34pub mod error;
35pub mod gc;
36pub mod name;
37
38#[cfg(feature = "artifact-local")]
39pub mod local;
40
41#[cfg(feature = "storage-s3")]
42pub mod s3;
43
44use std::path::Path;
45
46use bytes::Bytes;
47use futures_util::stream::{once, unfold};
48use tokio::fs::File;
49use tokio::io::AsyncReadExt;
50
51use crate::blob_store::ByteStream;
52use crate::error::ArtifactError;
53
54/// Size of the chunks yielded when reading a file.
55const READ_CHUNK_BYTES: usize = 64 * 1024;
56
57/// Wrap an in-memory buffer as a [`ByteStream`].
58///
59/// Convenience for callers that already hold the whole payload -- custom
60/// operations, tests, small generated files. Large payloads should be streamed
61/// from their source instead.
62///
63/// # Examples
64///
65/// ```
66/// use ironflow_artifacts::stream_from_bytes;
67///
68/// let stream = stream_from_bytes(b"hello".to_vec());
69/// # drop(stream);
70/// ```
71pub fn stream_from_bytes(bytes: impl Into<Bytes>) -> ByteStream {
72    let bytes = bytes.into();
73    Box::pin(once(async move { Ok(bytes) }))
74}
75
76/// Read an open file as a [`ByteStream`], in fixed-size chunks.
77///
78/// # Examples
79///
80/// ```no_run
81/// use ironflow_artifacts::stream_from_file;
82/// use tokio::fs::File;
83///
84/// # async fn example() -> Result<(), ironflow_artifacts::error::ArtifactError> {
85/// let file = File::open("report.html").await?;
86/// let stream = stream_from_file(file);
87/// # drop(stream);
88/// # Ok(())
89/// # }
90/// ```
91pub fn stream_from_file(file: File) -> ByteStream {
92    Box::pin(unfold(Some(file), |state| async move {
93        let mut file = state?;
94        let mut buf = vec![0u8; READ_CHUNK_BYTES];
95        match file.read(&mut buf).await {
96            Ok(0) => None,
97            Ok(read) => {
98                buf.truncate(read);
99                Some((Ok(Bytes::from(buf)), Some(file)))
100            }
101            Err(err) => Some((Err(ArtifactError::from(err)), None)),
102        }
103    }))
104}
105
106/// Open a file and read it as a [`ByteStream`].
107///
108/// # Errors
109///
110/// Returns [`ArtifactError::NotFound`] when the path does not exist, and
111/// [`ArtifactError::Io`] on any other filesystem failure.
112///
113/// # Examples
114///
115/// ```no_run
116/// use ironflow_artifacts::stream_from_path;
117///
118/// # async fn example() -> Result<(), ironflow_artifacts::error::ArtifactError> {
119/// let stream = stream_from_path("target/report.html").await?;
120/// # drop(stream);
121/// # Ok(())
122/// # }
123/// ```
124pub async fn stream_from_path(path: impl AsRef<Path>) -> Result<ByteStream, ArtifactError> {
125    let path = path.as_ref();
126    let file = File::open(path).await.map_err(|err| match err.kind() {
127        std::io::ErrorKind::NotFound => ArtifactError::NotFound(path.display().to_string()),
128        _ => ArtifactError::from(err),
129    })?;
130
131    Ok(stream_from_file(file))
132}
133
134/// Convenience re-exports for common usage.
135pub mod prelude {
136    pub use crate::blob_store::{BlobDigest, BlobFuture, BlobStore, ByteStream};
137    pub use crate::error::ArtifactError;
138    pub use crate::name::{
139        MAX_ARTIFACT_NAME_LEN, guess_content_type, storage_key, validate_artifact_name,
140    };
141    pub use crate::{stream_from_bytes, stream_from_file, stream_from_path};
142
143    #[cfg(feature = "artifact-local")]
144    pub use crate::local::{DEFAULT_MAX_ARTIFACT_BYTES, LocalBlobStore};
145
146    #[cfg(feature = "storage-s3")]
147    pub use crate::s3::S3BlobStore;
148}
149
150#[cfg(test)]
151mod tests {
152    use futures_util::TryStreamExt;
153
154    use super::*;
155
156    #[tokio::test]
157    async fn stream_from_bytes_yields_the_whole_buffer() {
158        let chunks: Vec<Bytes> = stream_from_bytes(b"hello".to_vec())
159            .try_collect()
160            .await
161            .expect("collect");
162
163        assert_eq!(chunks.concat(), b"hello");
164    }
165
166    #[tokio::test]
167    async fn stream_from_bytes_supports_an_empty_buffer() {
168        let chunks: Vec<Bytes> = stream_from_bytes(Vec::new())
169            .try_collect()
170            .await
171            .expect("collect");
172
173        assert!(chunks.concat().is_empty());
174    }
175
176    #[tokio::test]
177    async fn stream_from_path_reads_a_file_larger_than_one_chunk() {
178        let dir = tempfile::TempDir::new().expect("temp dir");
179        let path = dir.path().join("big.bin");
180        let payload: Vec<u8> = (0..200_000u32).map(|i| (i % 256) as u8).collect();
181        std::fs::write(&path, &payload).expect("write");
182
183        let chunks: Vec<Bytes> = stream_from_path(&path)
184            .await
185            .expect("open")
186            .try_collect()
187            .await
188            .expect("collect");
189
190        assert_eq!(chunks.concat(), payload);
191        assert!(chunks.len() > 1, "large file should stream in chunks");
192    }
193
194    #[tokio::test]
195    async fn stream_from_path_reports_a_missing_file_as_not_found() {
196        let result = stream_from_path("/nonexistent/ironflow/artifact").await;
197
198        assert!(matches!(result, Err(ArtifactError::NotFound(_))));
199    }
200}
ironflow_artifacts/lib.rs

ironflow_artifacts/
lib.rs
