store: Only close self-pipe when we're done

This took a crazy long time to debug but after lots of false
starts I think this is right. Basically what's going on is
we have async tasks that are talking over a `pipe()` inside
our own process.

We must not close the read side of the pipe until the
writer is done.

I believe this is dependent on tokio task scheduling order,
and it's way easier to reproduce when pinned to a single CPU.

Closes: #657
Signed-off-by: Colin Walters <[email protected]>

lib/src/container/store.rs

@@ -742,10 +742,15 @@ impl ImageImporter {
                         None
                     };
                     txn.commit(Some(cancellable))?;
-                    Ok::<_, anyhow::Error>(commit)
+                    // Pass back ownership, see below
+                    let blob = archive.into_inner().into_inner();
+                    Ok::<_, anyhow::Error>((blob, commit))
                 })
                 .map_err(|e| e.context(format!("Layer {}", layer.layer.digest())));
-            let commit = super::unencapsulate::join_fetch(import_task, driver).await?;
+            let (blob, commit) = super::unencapsulate::join_fetch(import_task, driver).await?;
+            // We can't close the read side until we've completed the rest of the processing
+            // to avoid breaking our own pipe-to-self. See https://github.com/ostreedev/ostree-rs-ext/issues/657
+            drop(blob);
             layer.commit = commit;
             if let Some(p) = self.layer_progress.as_ref() {
                 p.send(ImportProgress::OstreeChunkCompleted(layer.layer.clone()))
@@ -785,9 +790,14 @@ impl ImageImporter {
                     }
                     repo.mark_commit_partial(&commit, false)?;
                     txn.commit(Some(cancellable))?;
-                    Ok::<_, anyhow::Error>(commit)
+                    // Pass back ownership, see below
+                    let blob = archive.into_inner().into_inner();
+                    Ok::<_, anyhow::Error>((blob, commit))
                 });
-            let commit = super::unencapsulate::join_fetch(import_task, driver).await?;
+            let (blob, commit) = super::unencapsulate::join_fetch(import_task, driver).await?;
+            // We can't close the read side until we've completed the rest of the processing
+            // to avoid breaking our own pipe-to-self. See https://github.com/ostreedev/ostree-rs-ext/issues/657
+            drop(blob);
             import.ostree_commit_layer.commit = Some(commit);
             if let Some(p) = self.layer_progress.as_ref() {
                 p.send(ImportProgress::OstreeChunkCompleted(

lib/src/container/unencapsulate.rs

@@ -162,7 +162,7 @@ pub struct Import {
 /// to see if the worker function had an error *and* if the proxy
 /// had an error, but if the proxy's error ends in `broken pipe`
 /// then it means the real only error is from the worker.
-pub(crate) async fn join_fetch<T: std::fmt::Debug>(
+pub(crate) async fn join_fetch<T>(
     worker: impl Future<Output = Result<T>>,
     driver: impl Future<Output = Result<()>>,
 ) -> Result<T> {