pf_registry/
registry.rs

1// SPDX-License-Identifier: MIT
2//! The shared `Registry` trait. Adapter authors implement this; the CLI
3//! and SDKs talk to it.
4
5use async_trait::async_trait;
6use pf_core::cas::BlobStore;
7use pf_core::digest::Digest256;
8use pf_core::manifest::Manifest;
9
10use crate::image_ref::ImageRef;
11
12/// Backend errors. Adapter-specific failures (HTTP, S3 SigV4, IPFS daemon
13/// down, etc.) flow through `Backend`. Trait-level invariants surface as
14/// the typed variants above it.
15#[derive(Debug, thiserror::Error)]
16pub enum RegistryError {
17    /// Pull / push attempted against a scheme this build doesn't support
18    /// (feature flag off).
19    #[error("unsupported scheme: {0}")]
20    UnsupportedScheme(String),
21    /// Reference parser caught a malformed URL.
22    #[error("image ref: {0}")]
23    BadRef(#[from] crate::image_ref::ImageRefError),
24    /// Wrapped `pf-core` failure (CAS, manifest decode, …).
25    #[error("core: {0}")]
26    Core(#[from] pf_core::Error),
27    /// Adapter-specific backend failure (HTTP, S3, IPFS, …).
28    #[error("backend: {0}")]
29    Backend(String),
30    /// Manifest signature did not verify.
31    #[error("signature verify failed: {0}")]
32    SignatureVerify(String),
33}
34
35/// Output of [`Registry::pull`]: the manifest plus every blob it references.
36#[derive(Debug)]
37pub struct LayerSet {
38    pub manifest: Manifest,
39    pub blobs: Vec<(Digest256, Vec<u8>)>,
40}
41
42/// A registry. Push uploads a manifest + every reachable blob; pull
43/// returns both. Implementations MUST be `Send + Sync`.
44#[async_trait]
45pub trait Registry: Send + Sync {
46    /// Push a `.pfimg` to the destination encoded by `target`.
47    async fn push(
48        &self,
49        target: &ImageRef,
50        manifest: &Manifest,
51        blobs: &dyn BlobStore,
52    ) -> Result<(), RegistryError>;
53
54    /// Pull a `.pfimg` from `source`. Returns the manifest + all blobs
55    /// it references (so the caller can flush them into a local
56    /// [`BlobStore`] of choice).
57    async fn pull(&self, source: &ImageRef) -> Result<LayerSet, RegistryError>;
58
59    /// Cheap existence check (HEAD-style; doesn't pull blobs).
60    async fn exists(&self, source: &ImageRef) -> Result<bool, RegistryError>;
61}
62
63/// Walk the manifest's layer descriptors and yield every TOP-LEVEL
64/// digest. Use [`transitive_blob_digests`] when you actually want to
65/// materialise the full set (file blobs inside the FsTree, page blobs
66/// inside the PageManifest, etc.).
67#[must_use]
68pub fn manifest_blob_digests(m: &Manifest) -> Vec<Digest256> {
69    vec![
70        m.model.base.clone(),
71        m.model.diff.clone(),
72        m.cache.manifest.clone(),
73        m.world.fs.clone(),
74        m.world.env.clone(),
75        m.world.procs.clone(),
76        m.effects.ledger.clone(),
77        m.trace.messages.clone(),
78    ]
79}
80
81/// Walk every blob the manifest transitively references — including
82/// file blobs inside the world-layer `FsTree` and K/V page blobs inside
83/// the cache-layer `PageManifest`.
84///
85/// Returns digests in deterministic insertion order with duplicates
86/// removed. The model / effects / trace layers are self-contained in
87/// their top-level digest (JSON envelope or JSONL with no nested blob
88/// refs), so they aren't expanded here.
89pub fn transitive_blob_digests(
90    m: &Manifest,
91    blobs: &dyn BlobStore,
92) -> Result<Vec<Digest256>, RegistryError> {
93    use std::collections::BTreeSet;
94    let mut seen: BTreeSet<String> = BTreeSet::new();
95    let mut order: Vec<Digest256> = Vec::new();
96
97    // Top-level descriptors first.
98    for d in manifest_blob_digests(m) {
99        if seen.insert(d.as_str().to_owned()) {
100            order.push(d);
101        }
102    }
103
104    // World FsTree → file / symlink blobs.
105    if let Ok(fs_bytes) = blobs.get(&m.world.fs) {
106        if let Ok(tree) = serde_json::from_slice::<serde_json::Value>(&fs_bytes) {
107            if tree.get("kind").and_then(|v| v.as_str()) == Some("fs.tree.v1") {
108                if let Some(entries) = tree.get("entries").and_then(|v| v.as_array()) {
109                    for e in entries {
110                        if let Some(blob) = e.get("blob").and_then(|v| v.as_str()) {
111                            if let Ok(d) = Digest256::parse(blob) {
112                                if seen.insert(d.as_str().to_owned()) {
113                                    order.push(d);
114                                }
115                            }
116                        }
117                    }
118                }
119            }
120        }
121    }
122
123    // Cache PageManifest → per-page K and V blobs.
124    if let Ok(pm_bytes) = blobs.get(&m.cache.manifest) {
125        if let Ok(pm) = serde_json::from_slice::<serde_json::Value>(&pm_bytes) {
126            if let Some(pages) = pm.get("pages").and_then(|v| v.as_array()) {
127                for p in pages {
128                    for key in ["k", "v"] {
129                        if let Some(s) = p.get(key).and_then(|v| v.as_str()) {
130                            if let Ok(d) = Digest256::parse(s) {
131                                if seen.insert(d.as_str().to_owned()) {
132                                    order.push(d);
133                                }
134                            }
135                        }
136                    }
137                }
138            }
139        }
140    }
141
142    Ok(order)
143}
pf_registry/registry.rs

pf_registry/
registry.rs
