clawft_kernel/
artifact_store.rs

1//! Content-addressed artifact store using BLAKE3 hashes (K3-G1).
2//!
3//! Stores arbitrary binary artifacts (WASM modules, app manifests,
4//! config bundles) indexed by their BLAKE3 content hash. Deduplicates
5//! automatically: storing the same content twice returns the same hash
6//! and increments the reference count.
7//!
8//! This module requires the `ecc` feature (which pulls `blake3`).
9
10use std::path::PathBuf;
11use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
12
13use async_trait::async_trait;
14use chrono::{DateTime, Utc};
15use dashmap::DashMap;
16use serde::{Deserialize, Serialize};
17use tracing::{info, warn};
18
19#[cfg(feature = "exochain")]
20use std::sync::Arc;
21
22use crate::error::KernelError;
23use crate::health::HealthStatus;
24use crate::service::{ServiceType, SystemService};
25
26// ---------------------------------------------------------------------------
27// ArtifactType
28// ---------------------------------------------------------------------------
29
30/// The kind of content stored in the artifact.
31#[non_exhaustive]
32#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
33pub enum ArtifactType {
34    /// WebAssembly module (.wasm).
35    WasmModule,
36    /// Application manifest (app.json).
37    AppManifest,
38    /// Configuration bundle.
39    ConfigBundle,
40    /// Untyped binary blob.
41    Generic,
42}
43
44impl std::fmt::Display for ArtifactType {
45    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
46        match self {
47            Self::WasmModule => write!(f, "wasm-module"),
48            Self::AppManifest => write!(f, "app-manifest"),
49            Self::ConfigBundle => write!(f, "config-bundle"),
50            Self::Generic => write!(f, "generic"),
51        }
52    }
53}
54
55// ---------------------------------------------------------------------------
56// StoredArtifact (metadata)
57// ---------------------------------------------------------------------------
58
59/// Metadata about a stored artifact (the data lives in the backend).
60pub struct StoredArtifact {
61    /// BLAKE3 hex hash of the content.
62    pub hash: String,
63    /// Content size in bytes.
64    pub size: u64,
65    /// Content type classification.
66    pub content_type: ArtifactType,
67    /// When this artifact was first stored.
68    pub stored_at: DateTime<Utc>,
69    /// Number of active references (for GC).
70    pub reference_count: AtomicU32,
71}
72
73// ---------------------------------------------------------------------------
74// ArtifactBackend
75// ---------------------------------------------------------------------------
76
77/// Storage backend for artifact data.
78#[non_exhaustive]
79pub enum ArtifactBackend {
80    /// In-memory storage (for tests / embedded use).
81    Memory(DashMap<String, Vec<u8>>),
82    /// File-system storage with two-level directory sharding.
83    File {
84        /// Base directory for artifact files.
85        base_path: PathBuf,
86    },
87}
88
89// ---------------------------------------------------------------------------
90// ArtifactStore
91// ---------------------------------------------------------------------------
92
93/// Content-addressed artifact store using BLAKE3 hashes.
94///
95/// Thread-safe via `DashMap`. The store guarantees:
96/// - **Deduplication**: same content always produces the same hash.
97/// - **Integrity**: every load verifies the content hash.
98/// - **Reference counting**: tracks how many consumers reference each artifact.
99pub struct ArtifactStore {
100    /// Hash -> metadata index.
101    artifacts: DashMap<String, StoredArtifact>,
102    /// Data storage backend.
103    backend: ArtifactBackend,
104    /// Total stored bytes across all artifacts.
105    total_size: AtomicU64,
106    /// Optional chain manager for ExoChain event logging.
107    #[cfg(feature = "exochain")]
108    chain_manager: Option<Arc<crate::chain::ChainManager>>,
109}
110
111impl ArtifactStore {
112    /// Create a new in-memory artifact store.
113    pub fn new_memory() -> Self {
114        Self {
115            artifacts: DashMap::new(),
116            backend: ArtifactBackend::Memory(DashMap::new()),
117            total_size: AtomicU64::new(0),
118            #[cfg(feature = "exochain")]
119            chain_manager: None,
120        }
121    }
122
123    /// Create a new file-backed artifact store.
124    pub fn new_file(base_path: PathBuf) -> Self {
125        Self {
126            artifacts: DashMap::new(),
127            backend: ArtifactBackend::File { base_path },
128            total_size: AtomicU64::new(0),
129            #[cfg(feature = "exochain")]
130            chain_manager: None,
131        }
132    }
133
134    /// Open a file-backed store, indexing artifacts already on disk
135    /// (`<base>/<2-hex>/<64-hex>`). Entries whose name is not a BLAKE3 hex
136    /// or that sit in the wrong shard are skipped; content is still
137    /// verified on every `load`. Indexed entries get type `Generic`.
138    pub fn open_file(base_path: PathBuf) -> Result<Self, KernelError> {
139        let store = Self::new_file(base_path.clone());
140        let Ok(shards) = std::fs::read_dir(&base_path) else {
141            return Ok(store);
142        };
143        // An unreadable shard or entry is skipped with a warning: a store that
144        // cannot index one directory is still a store, and one bad entry must
145        // not take the caller (the placement plane) down. The skipped blobs are
146        // simply not indexed, so nothing treats them as present or ours.
147        for shard in shards {
148            let shard = match shard {
149                Ok(s) => s,
150                Err(e) => {
151                    warn!(error = %e, "artifact index: skipping an unreadable directory entry");
152                    continue;
153                }
154            };
155            let prefix = shard.file_name().to_string_lossy().into_owned();
156            if prefix.len() != 2 || !shard.file_type().is_ok_and(|t| t.is_dir()) {
157                continue;
158            }
159            let entries = match std::fs::read_dir(shard.path()) {
160                Ok(e) => e,
161                Err(e) => {
162                    warn!(shard = %prefix, error = %e, "artifact index: skipping an unreadable shard");
163                    continue;
164                }
165            };
166            for entry in entries {
167                let entry = match entry {
168                    Ok(e) => e,
169                    Err(e) => {
170                        warn!(shard = %prefix, error = %e, "artifact index: skipping an unreadable entry");
171                        continue;
172                    }
173                };
174                let name = entry.file_name().to_string_lossy().into_owned();
175                let hex = name.len() == 64
176                    && name.bytes().all(|b| matches!(b, b'0'..=b'9' | b'a'..=b'f'));
177                let meta = match entry.metadata() {
178                    Ok(m) => m,
179                    Err(e) => {
180                        warn!(entry = %name, error = %e, "artifact index: skipping an entry with no metadata");
181                        continue;
182                    }
183                };
184                if !hex || !name.starts_with(&prefix) || !meta.is_file() {
185                    continue;
186                }
187                store.total_size.fetch_add(meta.len(), Ordering::Relaxed);
188                store.artifacts.insert(
189                    name.clone(),
190                    StoredArtifact {
191                        hash: name,
192                        size: meta.len(),
193                        content_type: ArtifactType::Generic,
194                        stored_at: Utc::now(),
195                        reference_count: AtomicU32::new(1),
196                    },
197                );
198            }
199        }
200        Ok(store)
201    }
202
203    /// Set the chain manager for ExoChain event logging.
204    #[cfg(feature = "exochain")]
205    pub fn set_chain_manager(&mut self, cm: Arc<crate::chain::ChainManager>) {
206        self.chain_manager = Some(cm);
207    }
208
209    /// Store content and return the BLAKE3 hash.
210    ///
211    /// If the content is already stored, increments the reference count
212    /// and returns the existing hash (deduplication).
213    pub fn store(&self, content: &[u8], content_type: ArtifactType) -> Result<String, KernelError> {
214        let hash = blake3::hash(content).to_hex().to_string();
215
216        // Dedup: if hash exists, just bump reference count.
217        if let Some(existing) = self.artifacts.get(&hash) {
218            existing.reference_count.fetch_add(1, Ordering::Relaxed);
219            return Ok(hash);
220        }
221
222        // Write to backend.
223        match &self.backend {
224            ArtifactBackend::Memory(map) => {
225                map.insert(hash.clone(), content.to_vec());
226            }
227            ArtifactBackend::File { base_path } => {
228                let prefix = &hash[..2.min(hash.len())];
229                let dir = base_path.join(prefix);
230                std::fs::create_dir_all(&dir)
231                    .map_err(|e| KernelError::Service(format!("artifact dir create: {e}")))?;
232                std::fs::write(dir.join(&hash), content)
233                    .map_err(|e| KernelError::Service(format!("artifact write: {e}")))?;
234            }
235        }
236
237        // Record metadata.
238        let size = content.len() as u64;
239        #[cfg(feature = "exochain")]
240        let content_type_str = content_type.to_string();
241        self.artifacts.insert(
242            hash.clone(),
243            StoredArtifact {
244                hash: hash.clone(),
245                size,
246                content_type,
247                stored_at: Utc::now(),
248                reference_count: AtomicU32::new(1),
249            },
250        );
251        self.total_size.fetch_add(size, Ordering::Relaxed);
252
253        #[cfg(feature = "exochain")]
254        if let Some(ref cm) = self.chain_manager {
255            cm.append(
256                "artifact_store",
257                crate::chain::EVENT_KIND_ARTIFACT_STORE,
258                Some(serde_json::json!({
259                    "hash": hash,
260                    "size": size,
261                    "content_type": content_type_str,
262                })),
263            );
264        }
265
266        info!(hash = %hash, size, "artifact stored");
267        Ok(hash)
268    }
269
270    /// Load content by hash, verifying integrity on read.
271    pub fn load(&self, hash: &str) -> Result<Vec<u8>, KernelError> {
272        if !self.artifacts.contains_key(hash) {
273            return Err(KernelError::Service(format!("artifact not found: {hash}")));
274        }
275
276        let content = match &self.backend {
277            ArtifactBackend::Memory(map) => map.get(hash).map(|v| v.value().clone()),
278            ArtifactBackend::File { base_path } => {
279                let prefix = &hash[..2.min(hash.len())];
280                let path = base_path.join(prefix).join(hash);
281                std::fs::read(&path).ok()
282            }
283        };
284
285        let content = content
286            .ok_or_else(|| KernelError::Service(format!("artifact data missing: {hash}")))?;
287
288        // Verify integrity.
289        let actual_hash = blake3::hash(&content).to_hex().to_string();
290        if actual_hash != hash {
291            return Err(KernelError::Service(format!(
292                "artifact integrity error: expected {hash}, got {actual_hash}"
293            )));
294        }
295
296        Ok(content)
297    }
298
299    /// Check whether an artifact exists by hash.
300    pub fn contains(&self, hash: &str) -> bool {
301        self.artifacts.contains_key(hash)
302    }
303
304    /// Decrement the reference count for an artifact.
305    ///
306    /// Returns `true` if the reference count reached zero (eligible for GC).
307    pub fn release(&self, hash: &str) -> bool {
308        if let Some(entry) = self.artifacts.get(hash) {
309            let prev = entry.reference_count.fetch_sub(1, Ordering::Relaxed);
310            return prev <= 1;
311        }
312        false
313    }
314
315    /// Remove an artifact and its data from storage.
316    pub fn remove(&self, hash: &str) -> Result<(), KernelError> {
317        if let Some((_, meta)) = self.artifacts.remove(hash) {
318            self.total_size.fetch_sub(meta.size, Ordering::Relaxed);
319            match &self.backend {
320                ArtifactBackend::Memory(map) => {
321                    map.remove(hash);
322                }
323                ArtifactBackend::File { base_path } => {
324                    let prefix = &hash[..2.min(hash.len())];
325                    let path = base_path.join(prefix).join(hash);
326                    let _ = std::fs::remove_file(path);
327                }
328            }
329
330            #[cfg(feature = "exochain")]
331            if let Some(ref cm) = self.chain_manager {
332                cm.append(
333                    "artifact_store",
334                    crate::chain::EVENT_KIND_ARTIFACT_REMOVE,
335                    Some(serde_json::json!({
336                        "hash": hash,
337                    })),
338                );
339            }
340        }
341        Ok(())
342    }
343
344    /// Total number of stored artifacts.
345    pub fn count(&self) -> usize {
346        self.artifacts.len()
347    }
348
349    /// Total stored bytes.
350    pub fn total_bytes(&self) -> u64 {
351        self.total_size.load(Ordering::Relaxed)
352    }
353
354    /// Get metadata for an artifact (hash, size, type, stored_at, refcount).
355    pub fn metadata(&self, hash: &str) -> Option<(String, u64, ArtifactType, DateTime<Utc>, u32)> {
356        self.artifacts.get(hash).map(|a| {
357            (
358                a.hash.clone(),
359                a.size,
360                a.content_type.clone(),
361                a.stored_at,
362                a.reference_count.load(Ordering::Relaxed),
363            )
364        })
365    }
366}
367
368#[async_trait]
369impl SystemService for ArtifactStore {
370    fn name(&self) -> &str {
371        "artifact-store"
372    }
373
374    fn service_type(&self) -> ServiceType {
375        ServiceType::Core
376    }
377
378    async fn start(&self) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
379        info!("artifact store started");
380        Ok(())
381    }
382
383    async fn stop(&self) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
384        info!(
385            count = self.count(),
386            total_bytes = self.total_bytes(),
387            "artifact store stopped"
388        );
389        Ok(())
390    }
391
392    async fn health_check(&self) -> HealthStatus {
393        HealthStatus::Healthy
394    }
395}
396
397// ── Tests ─────────────────────────────────────────────────────────────────
398
399#[cfg(test)]
400mod tests {
401    use super::*;
402
403    #[cfg(unix)]
404    #[test]
405    fn open_file_skips_an_unreadable_shard_instead_of_failing() {
406        use std::os::unix::fs::PermissionsExt;
407        let dir = tempfile::tempdir().unwrap();
408        let writer = ArtifactStore::new_file(dir.path().to_path_buf());
409        let kept = writer.store(b"readable blob", ArtifactType::Generic).unwrap();
410        let hidden = writer.store(b"blob in a locked shard", ArtifactType::Generic).unwrap();
411        let locked = dir.path().join(&hidden[..2]);
412        std::fs::set_permissions(&locked, std::fs::Permissions::from_mode(0o000)).unwrap();
413        // Root ignores permissions: nothing to prove then.
414        let readable = std::fs::read_dir(&locked).is_ok();
415        let opened = ArtifactStore::open_file(dir.path().to_path_buf());
416        std::fs::set_permissions(&locked, std::fs::Permissions::from_mode(0o755)).unwrap();
417        let store = opened.expect("an unreadable shard is skipped, not fatal");
418        if !readable && kept[..2] != hidden[..2] {
419            assert!(store.contains(&kept), "the readable shard is still indexed");
420            assert!(!store.contains(&hidden), "the locked one is not");
421        }
422    }
423
424    #[test]
425    fn store_and_load_roundtrip() {
426        let store = ArtifactStore::new_memory();
427        let content = b"hello weftos artifact";
428        let hash = store.store(content, ArtifactType::Generic).unwrap();
429        let loaded = store.load(&hash).unwrap();
430        assert_eq!(loaded, content);
431    }
432
433    #[test]
434    fn hash_verification_on_load() {
435        let store = ArtifactStore::new_memory();
436        let content = b"original content";
437        let hash = store.store(content, ArtifactType::WasmModule).unwrap();
438
439        // Tamper with stored data.
440        if let ArtifactBackend::Memory(map) = &store.backend {
441            map.insert(hash.clone(), b"tampered content".to_vec());
442        }
443
444        let result = store.load(&hash);
445        assert!(result.is_err());
446        let err = result.unwrap_err().to_string();
447        assert!(err.contains("integrity error"), "got: {err}");
448    }
449
450    #[test]
451    fn duplicate_store_returns_same_hash() {
452        let store = ArtifactStore::new_memory();
453        let content = b"dedup test";
454        let h1 = store.store(content, ArtifactType::Generic).unwrap();
455        let h2 = store.store(content, ArtifactType::Generic).unwrap();
456        assert_eq!(h1, h2);
457        assert_eq!(store.count(), 1);
458    }
459
460    #[test]
461    fn duplicate_store_increments_refcount() {
462        let store = ArtifactStore::new_memory();
463        let content = b"refcount test";
464        let hash = store.store(content, ArtifactType::Generic).unwrap();
465        let _ = store.store(content, ArtifactType::Generic).unwrap();
466        let (_, _, _, _, refcount) = store.metadata(&hash).unwrap();
467        assert_eq!(refcount, 2);
468    }
469
470    #[test]
471    fn reference_counting_release() {
472        let store = ArtifactStore::new_memory();
473        let content = b"rc data";
474        let hash = store.store(content, ArtifactType::Generic).unwrap();
475        let _ = store.store(content, ArtifactType::Generic).unwrap();
476
477        assert!(!store.release(&hash)); // 2 -> 1, not zero yet
478        assert!(store.release(&hash)); // 1 -> 0, eligible for GC
479    }
480
481    #[test]
482    fn load_nonexistent_fails() {
483        let store = ArtifactStore::new_memory();
484        let result = store.load("0000000000000000000000000000000000000000000000000000000000000000");
485        assert!(result.is_err());
486    }
487
488    #[test]
489    fn total_bytes_tracked() {
490        let store = ArtifactStore::new_memory();
491        store.store(b"aaaa", ArtifactType::Generic).unwrap();
492        store.store(b"bbb", ArtifactType::Generic).unwrap();
493        assert_eq!(store.total_bytes(), 7);
494    }
495
496    #[test]
497    fn remove_artifact() {
498        let store = ArtifactStore::new_memory();
499        let hash = store.store(b"remove me", ArtifactType::Generic).unwrap();
500        assert!(store.contains(&hash));
501        store.remove(&hash).unwrap();
502        assert!(!store.contains(&hash));
503        assert_eq!(store.count(), 0);
504    }
505
506    #[test]
507    fn file_backend_roundtrip() {
508        let dir = std::env::temp_dir().join(format!("artifact_test_{}", uuid::Uuid::new_v4()));
509        std::fs::create_dir_all(&dir).unwrap();
510
511        let store = ArtifactStore::new_file(dir.clone());
512        let content = b"file backend content";
513        let hash = store.store(content, ArtifactType::AppManifest).unwrap();
514
515        // Verify two-level directory structure.
516        let prefix = &hash[..2];
517        let file_path = dir.join(prefix).join(&hash);
518        assert!(file_path.exists(), "expected file at {file_path:?}");
519
520        let loaded = store.load(&hash).unwrap();
521        assert_eq!(loaded, content);
522
523        // Cleanup.
524        let _ = std::fs::remove_dir_all(&dir);
525    }
526
527    #[test]
528    fn different_content_different_hash() {
529        let store = ArtifactStore::new_memory();
530        let h1 = store.store(b"alpha", ArtifactType::Generic).unwrap();
531        let h2 = store.store(b"beta", ArtifactType::Generic).unwrap();
532        assert_ne!(h1, h2);
533        assert_eq!(store.count(), 2);
534    }
535
536    #[tokio::test]
537    async fn system_service_impl() {
538        let store = ArtifactStore::new_memory();
539        assert_eq!(store.name(), "artifact-store");
540        assert_eq!(store.service_type(), ServiceType::Core);
541        store.start().await.unwrap();
542        let health = store.health_check().await;
543        assert_eq!(health, HealthStatus::Healthy);
544        store.stop().await.unwrap();
545    }
546}
clawft_kernel/artifact_store.rs

clawft_kernel/
artifact_store.rs
