ostree_ext/container/
layer_source.rs

1//! Abstraction over where the bytes of the image layers come from.
2
3use anyhow::Result;
4use containers_image_proxy::oci_spec::image as oci_image;
5use containers_image_proxy::{ConvertedLayerInfo, ImageProxy, OpenedImage, Transport};
6use futures_util::future::BoxFuture;
7use tokio::io::AsyncBufRead;
8use tokio::sync::{OnceCell, watch::Sender};
9
10use super::store::LayerProgress;
11use super::unencapsulate::fetch_layer;
12
13/// A layer opened for reading: its bytes, a driver future which must be polled
14/// alongside reading them, and the media type of those bytes. The media type is
15/// not necessarily the one in the layer descriptor, as some transports store
16/// layers uncompressed regardless of what the manifest says.
17pub type FetchedLayer<'a> = (
18    Box<dyn AsyncBufRead + Send + Unpin>,
19    BoxFuture<'a, Result<()>>,
20    oci_image::MediaType,
21);
22
23/// Abstraction for layer blob data.
24///
25/// Typically ProxyLayerSource which pulls from a registry, but for deltas
26/// we reconstruct data directly from the delta.
27pub trait LayerSource: std::fmt::Debug + Send + Sync {
28    /// Open one layer of `manifest` for reading.
29    fn fetch_layer<'a>(
30        &'a self,
31        manifest: &'a oci_image::ImageManifest,
32        layer: &'a oci_image::Descriptor,
33        progress: Option<&'a Sender<Option<LayerProgress>>>,
34    ) -> BoxFuture<'a, Result<FetchedLayer<'a>>>;
35
36    /// Release the source, at most once and after the last
37    /// [`Self::fetch_layer`].
38    ///
39    /// This is only reached when the import succeeds; an import that fails
40    /// part way through drops the source instead. Anything that must be
41    /// cleaned up either way belongs in a [`Drop`] impl.
42    fn finish(self: Box<Self>) -> BoxFuture<'static, Result<()>>;
43}
44
45/// Implementation of LayerSource via containers-image-proxy.
46#[derive(Debug)]
47pub(crate) struct ProxyLayerSource {
48    proxy: ImageProxy,
49    img: OpenedImage,
50    transport: Transport,
51    layer_info: OnceCell<Option<Vec<ConvertedLayerInfo>>>,
52}
53
54impl ProxyLayerSource {
55    pub(crate) fn new(proxy: ImageProxy, img: OpenedImage, transport: Transport) -> Self {
56        Self {
57            proxy,
58            img,
59            transport,
60            layer_info: OnceCell::new(),
61        }
62    }
63}
64
65impl LayerSource for ProxyLayerSource {
66    fn fetch_layer<'a>(
67        &'a self,
68        manifest: &'a oci_image::ImageManifest,
69        layer: &'a oci_image::Descriptor,
70        progress: Option<&'a Sender<Option<LayerProgress>>>,
71    ) -> BoxFuture<'a, Result<FetchedLayer<'a>>> {
72        Box::pin(async move {
73            let layer_info = self
74                .layer_info
75                .get_or_try_init(|| self.proxy.get_layer_info(&self.img))
76                .await?;
77            let (blob, driver, media_type) = fetch_layer(
78                &self.proxy,
79                &self.img,
80                manifest,
81                layer,
82                progress,
83                layer_info.as_ref(),
84                self.transport,
85            )
86            .await?;
87            Ok((blob, Box::pin(driver) as BoxFuture<'a, _>, media_type))
88        })
89    }
90
91    fn finish(self: Box<Self>) -> BoxFuture<'static, Result<()>> {
92        Box::pin(async move {
93            let Self { proxy, img, .. } = *self;
94            // TODO change the imageproxy API to ensure this happens automatically when
95            // the image reference is dropped
96            proxy.close_image(&img).await?;
97            proxy.finalize().await?;
98            Ok(())
99        })
100    }
101}
ostree_ext/container/layer_source.rs

ostree_ext/container/
layer_source.rs
