src: start using a bit of refcounting

Our `SplitStreamWriter` and `oci::ImageOp` structs contain simple
references to `Repository` which results in some awkward lifetime rules
on those structs.  We can simplify things substantially if we lean into
ref-counting a bit more.

I'm not yet ready to declare that Repository is always refcounted, but
for operations involving splitstreams (including oci downloads) it is
now required.

The ergonomics of this change surprised me.  The Deref trait on `Arc<>`
and the ability to define `self: &Arc<Self>` methods makes this all
quite nice to use.

Signed-off-by: Allison Karlitskaya <allison.karlitskaya@redhat.com>

Author: allisonkarlitskaya

src/bin/cfsctl.rs

@@ -1,4 +1,4 @@
-use std::path::PathBuf;
+use std::{path::PathBuf, sync::Arc};
 
 use anyhow::Result;
 use clap::{Parser, Subcommand};
@@ -141,7 +141,7 @@ async fn main() -> Result<()> {
         Command::Oci { cmd: oci_cmd } => match oci_cmd {
             OciCommand::ImportLayer { name, sha256 } => {
                 let object_id = oci::import_layer(
-                    &repo,
+                    &Arc::new(repo),
                     &parse_sha256(sha256)?,
                     name.as_deref(),
                     &mut std::io::stdin(),
@@ -159,11 +159,11 @@ async fn main() -> Result<()> {
                 println!("{}", image_id.to_hex());
             }
             OciCommand::Pull { ref image, name } => {
-                oci::pull(&repo, image, name.as_deref()).await?
+                oci::pull(&Arc::new(repo), image, name.as_deref()).await?
             }
             OciCommand::Seal { verity, ref name } => {
                 let (sha256, verity) = oci::seal(
-                    &repo,
+                    &Arc::new(repo),
                     name,
                     verity.map(Sha256HashValue::from_hex).transpose()?.as_ref(),
                 )?;

src/oci/mod.rs

@@ -3,7 +3,7 @@ use std::process::Command;
 pub mod image;
 pub mod tar;
 
-use std::{collections::HashMap, io::Read, iter::zip, path::Path};
+use std::{collections::HashMap, io::Read, iter::zip, path::Path, sync::Arc};
 
 use anyhow::{bail, ensure, Context, Result};
 use async_compression::tokio::bufread::{GzipDecoder, ZstdDecoder};
@@ -23,7 +23,7 @@ use crate::{
 };
 
 pub fn import_layer<ObjectID: FsVerityHashValue>(
-    repo: &Repository<ObjectID>,
+    repo: &Arc<Repository<ObjectID>>,
     sha256: &Sha256Digest,
     name: Option<&str>,
     tar_stream: &mut impl Read,
@@ -44,8 +44,8 @@ pub fn ls_layer<ObjectID: FsVerityHashValue>(
     Ok(())
 }
 
-struct ImageOp<'repo, ObjectID: FsVerityHashValue> {
-    repo: &'repo Repository<ObjectID>,
+struct ImageOp<ObjectID: FsVerityHashValue> {
+    repo: Arc<Repository<ObjectID>>,
     proxy: ImageProxy,
     img: OpenedImage,
     progress: MultiProgress,
@@ -67,8 +67,8 @@ fn sha256_from_digest(digest: &str) -> Result<Sha256Digest> {
 
 type ContentAndVerity<ObjectID> = (Sha256Digest, ObjectID);
 
-impl<'repo, ObjectID: FsVerityHashValue> ImageOp<'repo, ObjectID> {
-    async fn new(repo: &'repo Repository<ObjectID>, imgref: &str) -> Result<Self> {
+impl<ObjectID: FsVerityHashValue> ImageOp<ObjectID> {
+    async fn new(repo: &Arc<Repository<ObjectID>>, imgref: &str) -> Result<Self> {
         // See https://github.com/containers/skopeo/issues/2563
         let skopeo_cmd = if imgref.starts_with("containers-storage:") {
             let mut cmd = Command::new("podman");
@@ -87,7 +87,7 @@ impl<'repo, ObjectID: FsVerityHashValue> ImageOp<'repo, ObjectID> {
         let img = proxy.open_image(imgref).await.context("Opening image")?;
         let progress = MultiProgress::new();
         Ok(ImageOp {
-            repo,
+            repo: Arc::clone(repo),
             proxy,
             img,
             progress,
@@ -142,7 +142,7 @@ impl<'repo, ObjectID: FsVerityHashValue> ImageOp<'repo, ObjectID> {
     }
 
     pub async fn ensure_config(
-        &self,
+        self: &Arc<Self>,
         manifest_layers: &[Descriptor],
         descriptor: &Descriptor,
     ) -> Result<ContentAndVerity<ObjectID>> {
@@ -192,7 +192,7 @@ impl<'repo, ObjectID: FsVerityHashValue> ImageOp<'repo, ObjectID> {
         }
     }
 
-    pub async fn pull(&self) -> Result<ContentAndVerity<ObjectID>> {
+    pub async fn pull(self: &Arc<Self>) -> Result<ContentAndVerity<ObjectID>> {
         let (_manifest_digest, raw_manifest) = self
             .proxy
             .fetch_manifest_raw_oci(&self.img)
@@ -213,11 +213,11 @@ impl<'repo, ObjectID: FsVerityHashValue> ImageOp<'repo, ObjectID> {
 /// Pull the target image, and add the provided tag. If this is a mountable
 /// image (i.e. not an artifact), it is *not* unpacked by default.
 pub async fn pull(
-    repo: &Repository<impl FsVerityHashValue>,
+    repo: &Arc<Repository<impl FsVerityHashValue>>,
     imgref: &str,
     reference: Option<&str>,
 ) -> Result<()> {
-    let op = ImageOp::new(repo, imgref).await?;
+    let op = Arc::new(ImageOp::new(repo, imgref).await?);
     let (sha256, id) = op
         .pull()
         .await
@@ -280,7 +280,7 @@ pub fn open_config_shallow<ObjectID: FsVerityHashValue>(
 }
 
 pub fn write_config<ObjectID: FsVerityHashValue>(
-    repo: &Repository<ObjectID>,
+    repo: &Arc<Repository<ObjectID>>,
     config: &ImageConfiguration,
     refs: DigestMap<ObjectID>,
 ) -> Result<ContentAndVerity<ObjectID>> {
@@ -294,7 +294,7 @@ pub fn write_config<ObjectID: FsVerityHashValue>(
 }
 
 pub fn seal<ObjectID: FsVerityHashValue>(
-    repo: &Repository<ObjectID>,
+    repo: &Arc<Repository<ObjectID>>,
     name: &str,
     verity: Option<&ObjectID>,
 ) -> Result<ContentAndVerity<ObjectID>> {
@@ -421,7 +421,7 @@ mod test {
         let layer_id: [u8; 32] = context.finalize().into();
 
         let repo_dir = tempdir();
-        let repo = Repository::<Sha256HashValue>::open_path(CWD, &repo_dir).unwrap();
+        let repo = Arc::new(Repository::<Sha256HashValue>::open_path(CWD, &repo_dir).unwrap());
         let id = import_layer(&repo, &layer_id, Some("name"), &mut layer.as_slice()).unwrap();
 
         let mut dump = String::new();

src/oci/tar.rs

@@ -75,7 +75,7 @@ pub fn split(
 
 pub async fn split_async(
     mut tar_stream: impl AsyncRead + Unpin,
-    writer: &mut SplitStreamWriter<'_, impl FsVerityHashValue>,
+    writer: &mut SplitStreamWriter<impl FsVerityHashValue>,
 ) -> Result<()> {
     while let Some(header) = read_header_async(&mut tar_stream).await? {
         // the header always gets stored as inline data

src/repository.rs

@@ -5,6 +5,7 @@ use std::{
     io::{Read, Write},
     os::fd::{AsFd, OwnedFd},
     path::{Path, PathBuf},
+    sync::Arc,
 };
 
 use anyhow::{bail, ensure, Context, Result};
@@ -173,7 +174,7 @@ impl<ObjectID: FsVerityHashValue> Repository<ObjectID> {
     /// You should write the data to the returned object and then pass it to .store_stream() to
     /// store the result.
     pub fn create_stream(
-        &self,
+        self: &Arc<Self>,
         sha256: Option<Sha256Digest>,
         maps: Option<DigestMap<ObjectID>>,
     ) -> SplitStreamWriter<ObjectID> {
@@ -285,7 +286,7 @@ impl<ObjectID: FsVerityHashValue> Repository<ObjectID> {
     /// On success, the object ID of the new object is returned.  It is expected that this object
     /// ID will be used when referring to the stream from other linked streams.
     pub fn ensure_stream(
-        &self,
+        self: &Arc<Self>,
         sha256: &Sha256Digest,
         callback: impl FnOnce(&mut SplitStreamWriter<ObjectID>) -> Result<()>,
         reference: Option<&str>,

src/splitstream.rs

@@ -3,7 +3,10 @@
  * See doc/splitstream.md
  */
 
-use std::io::{BufReader, Read, Write};
+use std::{
+    io::{BufReader, Read, Write},
+    sync::Arc,
+};
 
 use anyhow::{bail, Result};
 use sha2::{Digest, Sha256};
@@ -60,14 +63,14 @@ impl<ObjectID: FsVerityHashValue> DigestMap<ObjectID> {
     }
 }
 
-pub struct SplitStreamWriter<'a, ObjectID: FsVerityHashValue> {
-    repo: &'a Repository<ObjectID>,
+pub struct SplitStreamWriter<ObjectID: FsVerityHashValue> {
+    repo: Arc<Repository<ObjectID>>,
     inline_content: Vec<u8>,
-    writer: Encoder<'a, Vec<u8>>,
+    writer: Encoder<'static, Vec<u8>>,
     pub sha256: Option<(Sha256, Sha256Digest)>,
 }
 
-impl<ObjectID: FsVerityHashValue> std::fmt::Debug for SplitStreamWriter<'_, ObjectID> {
+impl<ObjectID: FsVerityHashValue> std::fmt::Debug for SplitStreamWriter<ObjectID> {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         // writer doesn't impl Debug
         f.debug_struct("SplitStreamWriter")
@@ -78,9 +81,9 @@ impl<ObjectID: FsVerityHashValue> std::fmt::Debug for SplitStreamWriter<'_, Obje
     }
 }
 
-impl<'a, ObjectID: FsVerityHashValue> SplitStreamWriter<'a, ObjectID> {
+impl<ObjectID: FsVerityHashValue> SplitStreamWriter<ObjectID> {
     pub fn new(
-        repo: &'a Repository<ObjectID>,
+        repo: &Arc<Repository<ObjectID>>,
         refs: Option<DigestMap<ObjectID>>,
         sha256: Option<Sha256Digest>,
     ) -> Self {
@@ -98,7 +101,7 @@ impl<'a, ObjectID: FsVerityHashValue> SplitStreamWriter<'a, ObjectID> {
         }
 
         Self {
-            repo,
+            repo: Arc::clone(repo),
             inline_content: vec![],
             writer,
             sha256: sha256.map(|x| (Sha256::new(), x)),