refactor(BufferedWriter): flush task owns the VirtualFile & abstraction for cleanup on drop (#11549)

Main change:

- `BufferedWriter` owns the `W`; no more `Arc<W>`
- We introduce auto-delete-on-drop wrappers for `VirtualFile`.
  - `TempVirtualFile` for write-only users
- `TempVirtualFileCoOwnedByEphemeralFileAndBufferedWriter` for
EphemeralFile which requires read access to the immutable prefix of the
file (see doc comments for details)
- Users of `BufferedWriter` hand it such a wrapped `VirtualFile`.
- The wrapped `VirtualFile` moves to the background flush task.
- On `BufferedWriter` shutdown, ownership moves back.
- Callers remove the wrapper (`disarm_into_inner()`) after doing final
touches, e.g., flushing index blocks and summary for delta/image layer
writers.

If the BufferedWriter isn't shut down properly via
`BufferedWriter::shutdown`, or if there is an error during final
touches, the wrapper type ensures that the file gets unlinked.

We store a GateGuard inside the wrapper to ensure that the Timeline is
still alive when unlinking on drop.

Rust doesn't have async drop yet, so, the unlinking happens using a
synchronous syscall.
NB we don't fsync the surrounding directory.
This is how it's been before this PR; I believe it is correct because
all of these files are temporary paths that get cleaned up on timeline
load.
Again, timeline load does not need to fsync because the next timeline
load will unlink again if the file reappears.

The auto-delete-on-drop can happen after a higher-level mechanism
retries.
Therefore, we switch all users to monotonically increasing, never-reused
temp file disambiguators.

The aspects pointed out in the last two paragraphs will receive further
cleanup in follow-up task
- https://github.com/neondatabase/neon/issues/11692

Drive-by changes:
- It turns out we can remove the two-pronged code in the layer file
download code.
No need to make this a separate PR because all of production already
uses `tokio-epoll-uring` with the buffered writer for many weeks.


Refs
- epic https://github.com/neondatabase/neon/issues/9868
- alternative to https://github.com/neondatabase/neon/pull/11544

pageserver/src/virtual_file/owned_buffers_io/write.rs

@@ -1,5 +1,4 @@
 mod flush;
-use std::sync::Arc;
 
 pub(crate) use flush::FlushControl;
 use flush::FlushHandle;
@@ -41,7 +40,6 @@ pub trait OwnedAsyncWriter {
 // TODO(yuchen): For large write, implementing buffer bypass for aligned parts of the write could be beneficial to throughput,
 // since we would avoid copying majority of the data into the internal buffer.
 pub struct BufferedWriter<B: Buffer, W> {
-    writer: Arc<W>,
     /// Clone of the buffer that was last submitted to the flush loop.
     /// `None` if no flush request has been submitted, Some forever after.
     pub(super) maybe_flushed: Option<FullSlice<B::IoBuf>>,
@@ -72,7 +70,7 @@ where
     ///
     /// The `buf_new` function provides a way to initialize the owned buffers used by this writer.
     pub fn new(
-        writer: Arc<W>,
+        writer: W,
         buf_new: impl Fn() -> B,
         gate_guard: utils::sync::gate::GateGuard,
         cancel: CancellationToken,
@@ -80,7 +78,6 @@ where
         flush_task_span: tracing::Span,
     ) -> Self {
         Self {
-            writer: writer.clone(),
             mutable: Some(buf_new()),
             maybe_flushed: None,
             flush_handle: FlushHandle::spawn_new(
@@ -95,10 +92,6 @@ where
         }
     }
 
-    pub fn as_inner(&self) -> &W {
-        &self.writer
-    }
-
     /// Returns the number of bytes submitted to the background flush task.
     pub fn bytes_submitted(&self) -> u64 {
         self.bytes_submitted
@@ -116,20 +109,16 @@ where
     }
 
     #[cfg_attr(target_os = "macos", allow(dead_code))]
-    pub async fn flush_and_into_inner(
-        mut self,
-        ctx: &RequestContext,
-    ) -> Result<(u64, Arc<W>), FlushTaskError> {
+    pub async fn shutdown(mut self, ctx: &RequestContext) -> Result<(u64, W), FlushTaskError> {
         self.flush(ctx).await?;
 
         let Self {
             mutable: buf,
             maybe_flushed: _,
-            writer,
             mut flush_handle,
             bytes_submitted: bytes_amount,
         } = self;
-        flush_handle.shutdown().await?;
+        let writer = flush_handle.shutdown().await?;
         assert!(buf.is_some());
         Ok((bytes_amount, writer))
     }
@@ -329,7 +318,7 @@ mod tests {
     async fn test_write_all_borrowed_always_goes_through_buffer() -> anyhow::Result<()> {
         let ctx = test_ctx();
         let ctx = &ctx;
-        let recorder = Arc::new(RecorderWriter::default());
+        let recorder = RecorderWriter::default();
         let gate = utils::sync::gate::Gate::default();
         let cancel = CancellationToken::new();
         let mut writer = BufferedWriter::<_, RecorderWriter>::new(
@@ -350,7 +339,7 @@ mod tests {
         writer.write_buffered_borrowed(b"j", ctx).await?;
         writer.write_buffered_borrowed(b"klmno", ctx).await?;
 
-        let (_, recorder) = writer.flush_and_into_inner(ctx).await?;
+        let (_, recorder) = writer.shutdown(ctx).await?;
         assert_eq!(
             recorder.get_writes(),
             {

pageserver/src/virtual_file/owned_buffers_io/write/flush.rs

@@ -1,5 +1,4 @@
 use std::ops::ControlFlow;
-use std::sync::Arc;
 
 use tokio_util::sync::CancellationToken;
 use tracing::{Instrument, info, info_span, warn};
@@ -21,7 +20,7 @@ pub struct FlushHandleInner<Buf, W> {
     /// and receives recyled buffer.
     channel: duplex::mpsc::Duplex<FlushRequest<Buf>, FullSlice<Buf>>,
     /// Join handle for the background flush task.
-    join_handle: tokio::task::JoinHandle<Result<Arc<W>, FlushTaskError>>,
+    join_handle: tokio::task::JoinHandle<Result<W, FlushTaskError>>,
 }
 
 struct FlushRequest<Buf> {
@@ -120,7 +119,7 @@ where
     /// The queue depth is 1, and the passed-in `buf` seeds the queue depth.
     /// I.e., the passed-in buf is immediately available to the handle as a recycled buffer.
     pub fn spawn_new<B>(
-        file: Arc<W>,
+        file: W,
         buf: B,
         gate_guard: utils::sync::gate::GateGuard,
         cancel: CancellationToken,
@@ -183,7 +182,7 @@ where
     }
 
     /// Cleans up the channel, join the flush task.
-    pub async fn shutdown(&mut self) -> Result<Arc<W>, FlushTaskError> {
+    pub async fn shutdown(&mut self) -> Result<W, FlushTaskError> {
         let handle = self
             .inner
             .take()
@@ -207,7 +206,7 @@ pub struct FlushBackgroundTask<Buf, W> {
     /// and send back recycled buffer.
     channel: duplex::mpsc::Duplex<FullSlice<Buf>, FlushRequest<Buf>>,
     /// A writter for persisting data to disk.
-    writer: Arc<W>,
+    writer: W,
     ctx: RequestContext,
     cancel: CancellationToken,
     /// Prevent timeline from shuting down until the flush background task finishes flushing all remaining buffers to disk.
@@ -228,7 +227,7 @@ where
     /// Creates a new background flush task.
     fn new(
         channel: duplex::mpsc::Duplex<FullSlice<Buf>, FlushRequest<Buf>>,
-        file: Arc<W>,
+        file: W,
         gate_guard: utils::sync::gate::GateGuard,
         cancel: CancellationToken,
         ctx: RequestContext,
@@ -243,7 +242,7 @@ where
     }
 
     /// Runs the background flush task.
-    async fn run(mut self) -> Result<Arc<W>, FlushTaskError> {
+    async fn run(mut self) -> Result<W, FlushTaskError> {
         //  Exit condition: channel is closed and there is no remaining buffer to be flushed
         while let Some(request) = self.channel.recv().await {
             #[cfg(test)]

pageserver/src/virtual_file/temporary.rs

@@ -0,0 +1,106 @@
+use tracing::error;
+use utils::sync::gate::GateGuard;
+
+use crate::context::RequestContext;
+
+use super::{
+    MaybeFatalIo, VirtualFile,
+    owned_buffers_io::{
+        io_buf_aligned::IoBufAligned, io_buf_ext::FullSlice, write::OwnedAsyncWriter,
+    },
+};
+
+/// A wrapper around [`super::VirtualFile`] that deletes the file on drop.
+/// For use as a [`OwnedAsyncWriter`] in [`super::owned_buffers_io::write::BufferedWriter`].
+#[derive(Debug)]
+pub struct TempVirtualFile {
+    inner: Option<Inner>,
+}
+
+#[derive(Debug)]
+struct Inner {
+    file: VirtualFile,
+    /// Gate guard is held on as long as we need to do operations in the path (delete on drop)
+    _gate_guard: GateGuard,
+}
+
+impl OwnedAsyncWriter for TempVirtualFile {
+    fn write_all_at<Buf: IoBufAligned + Send>(
+        &self,
+        buf: FullSlice<Buf>,
+        offset: u64,
+        ctx: &RequestContext,
+    ) -> impl std::future::Future<Output = (FullSlice<Buf>, std::io::Result<()>)> + Send {
+        VirtualFile::write_all_at(self, buf, offset, ctx)
+    }
+}
+
+impl Drop for TempVirtualFile {
+    fn drop(&mut self) {
+        let Some(Inner { file, _gate_guard }) = self.inner.take() else {
+            return;
+        };
+        let path = file.path();
+        if let Err(e) =
+            std::fs::remove_file(path).maybe_fatal_err("failed to remove the virtual file")
+        {
+            error!(err=%e, path=%path, "failed to remove");
+        }
+        drop(_gate_guard);
+    }
+}
+
+impl std::ops::Deref for TempVirtualFile {
+    type Target = VirtualFile;
+
+    fn deref(&self) -> &Self::Target {
+        &self
+            .inner
+            .as_ref()
+            .expect("only None after into_inner or drop")
+            .file
+    }
+}
+
+impl std::ops::DerefMut for TempVirtualFile {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self
+            .inner
+            .as_mut()
+            .expect("only None after into_inner or drop")
+            .file
+    }
+}
+
+impl TempVirtualFile {
+    /// The caller is responsible for ensuring that the path of `virtual_file` is not reused
+    /// until after this TempVirtualFile's `Drop` impl has completed.
+    /// Failure to do so will result in unlinking of the reused path by the original instance's Drop impl.
+    /// The best way to do so is by using a monotonic counter as a disambiguator.
+    /// TODO: centralize this disambiguator pattern inside this struct.
+    ///   => <https://github.com/neondatabase/neon/pull/11549#issuecomment-2824592831>
+    pub fn new(virtual_file: VirtualFile, gate_guard: GateGuard) -> Self {
+        Self {
+            inner: Some(Inner {
+                file: virtual_file,
+                _gate_guard: gate_guard,
+            }),
+        }
+    }
+
+    /// Dismantle this wrapper and return the underlying [`VirtualFile`].
+    /// This disables auto-unlinking functionality that is the essence of this wrapper.
+    ///
+    /// The gate guard is dropped as well; it is the callers responsibility to ensure filesystem
+    /// operations after calls to this functions are still gated by some other gate guard.
+    ///
+    /// TODO:
+    /// - centralize the common usage pattern of callers (sync_all(self), rename(self, dst), sync_all(dst.parent))
+    ///   => <https://github.com/neondatabase/neon/pull/11549#issuecomment-2824592831>
+    pub fn disarm_into_inner(mut self) -> VirtualFile {
+        self.inner
+            .take()
+            .expect("only None after into_inner or drop, and we are into_inner, and we consume")
+            .file
+    }
+}