[UIO] Open append handles through UniversalWriteFileOps (#10091)

A file handle that appends was only reachable through
`UniversalReadFs::open` with `writeable: true`, which ties the append
capability to the backend's read handle type. Give the write-side
filesystem trait its own opening path instead:

    type AppendFile: UniversalAppend;
    fn open_append(&self, path, options) -> UioResult<Self::AppendFile>;

`AppendFile` is deliberately not tied to `UniversalReadFs::File`: the two
capabilities live on independent traits, so a backend may serve reads
through one handle type and appends through another, and a filesystem
that opens no read handles at all still names an append handle. For the
same reason there is no `OpenExtra` parameter — the append handle may
come from a different backend, whose per-open knobs would not apply.
`OpenOptions::for_append` forces `writeable` on, since a read-only
append handle is a contradiction rather than an error worth propagating.

Drop the `UniversalWriteFileOps` impls on the two disk caches first.
Both were vestigial: `DiskCacheFs` got its forwarding-to-remote impl
mechanically in the read/write trait split (#9682) and no caller ever
used it, and `BlockCacheFs` lives in a module that is dead code. Neither
cache can open a writeable file, so neither can produce an append
handle; mutations go straight to the backing storage. An
`assert_not_impl_any!` locks this in next to the existing one on
`DiskCache`.

`BlobFs` consequently requires `AsyncAppend` rather than `AsyncWrite`:
only a backend with a native single-request append can hand out an
append handle. Object stores that can just put whole objects (GCS,
Azure) stay read-only through universal I/O; nothing used their write
side.

The conformance suite gains `run_open_append_conformance`, run by
`run_append_conformance` and public for filesystems whose own `File`
does not append. It covers `MmapFs`, `IoUringFs` and `BlobFs` over the
in-memory object store.

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

lib/common/common/src/universal_io/conformance.rs

@@ -153,4 +153,43 @@ where
         first.read_whole::<u8>().unwrap().as_ref(),
         b"aaabbbccc".as_slice(),
     );
+
+    run_open_append_conformance(fs, dir);
+}
+
+/// Exercise [`UniversalWriteFileOps::open_append`] against a backend: the
+/// filesystem hands out an append handle whose writes the read path observes,
+/// whatever file type either side is.
+///
+/// Called by [`run_append_conformance`]; run it directly for a filesystem
+/// whose own [`UniversalReadFs::File`] does not append.
+pub fn run_open_append_conformance<Fs>(fs: &Fs, dir: &Path)
+where
+    Fs: UniversalReadFs + UniversalWriteFileOps,
+    Fs::OpenExtra: Default,
+{
+    let path = dir.join("open_append.dat");
+    fs.create(&path, 0).unwrap();
+
+    // `writeable` is forced on: an append handle opened as read-only is a
+    // contradiction, not an error to propagate.
+    let mut file = fs.open_append(&path, open_options(false)).unwrap();
+
+    file.append(0, b"one".as_slice()).unwrap();
+    file.append_batch(3, [b"two".as_slice(), b"three".as_slice()])
+        .unwrap();
+    (file.flusher())().unwrap();
+
+    // Same compare-and-swap contract as any other append handle.
+    let err = file.append(0, b"x".as_slice()).unwrap_err();
+    assert!(matches!(err, UniversalIoError::AppendOffsetConflict { .. }));
+
+    // The read path sees the appended bytes.
+    let reader = fs
+        .open(&path, open_options(false), Fs::OpenExtra::default())
+        .unwrap();
+    assert_eq!(
+        reader.read_whole::<u8>().unwrap().as_ref(),
+        b"onetwothree".as_slice(),
+    );
 }

lib/common/common/src/universal_io/disk_cache/mod.rs

@@ -8,7 +8,7 @@ use crate::ext::aligned_vec::ACow;
 use crate::generic_consts::AccessPattern;
 use crate::universal_io::{
     ListedFile, OpenOptions, UioResult, UniversalIoError, UniversalRead, UniversalReadFileOps,
-    UniversalReadFs, UniversalWriteFileOps, UserData, local_file_ops,
+    UniversalReadFs, UserData, local_file_ops,
 };
 
 mod cached_slice;
@@ -100,27 +100,10 @@ impl UniversalReadFileOps for BlockCacheFs {
     }
 }
 
-impl UniversalWriteFileOps for BlockCacheFs {
-    fn create(&self, path: &Path, expected_length: usize) -> UioResult<()> {
-        local_file_ops::local_create(path, expected_length)
-    }
-
-    fn create_dir(&self, path: &Path) -> UioResult<()> {
-        local_file_ops::local_create_dir(path)
-    }
-
-    fn remove(&self, path: &Path) -> UioResult<()> {
-        local_file_ops::local_remove(path)
-    }
-
-    fn remove_dir(&self, path: &Path) -> UioResult<()> {
-        local_file_ops::local_remove_dir(path)
-    }
-
-    fn atomic_save(&self, path: &Path, bytes: &[u8]) -> UioResult<()> {
-        local_file_ops::local_atomic_save(path, bytes)
-    }
-}
+// Deliberately no `UniversalWriteFileOps` impl: the block cache is strictly
+// read-only ([`CachedSlice`] neither writes nor appends, and `open` rejects
+// writeable opens). Mutations go straight to the underlying local
+// filesystem — `MmapFs`/`IoUringFs` over the very same paths.
 
 impl UniversalReadFs for BlockCacheFs {
     type File = CachedSlice;

lib/common/common/src/universal_io/io_uring/mod.rs

@@ -78,6 +78,8 @@ impl UniversalReadFileOps for IoUringFs {
 }
 
 impl UniversalWriteFileOps for IoUringFs {
+    type AppendFile = IoUringFile;
+
     fn create(&self, path: &Path, expected_length: usize) -> UioResult<()> {
         local_file_ops::local_create(path, expected_length)
     }
@@ -97,6 +99,13 @@ impl UniversalWriteFileOps for IoUringFs {
     fn atomic_save(&self, path: &Path, bytes: &[u8]) -> UioResult<()> {
         local_file_ops::local_atomic_save(path, bytes)
     }
+
+    /// The very handle [`UniversalReadFs::open`] hands out, opened with the
+    /// default extras: an `O_DIRECT` handle cannot append (its block-aligned
+    /// I/O requirements rule out appends of arbitrary sizes).
+    fn open_append(&self, path: impl AsRef<Path>, options: OpenOptions) -> UioResult<IoUringFile> {
+        self.open(path, options.for_append(), IoUringOpenExtra::default())
+    }
 }
 
 /// Per-open backend extras for [`IoUringFs::open`].

lib/common/common/src/universal_io/mmap/mod.rs

@@ -38,6 +38,8 @@ impl UniversalReadFileOps for MmapFs {
 }
 
 impl UniversalWriteFileOps for MmapFs {
+    type AppendFile = MmapFile;
+
     fn create(&self, path: &Path, expected_length: usize) -> UioResult<()> {
         local_file_ops::local_create(path, expected_length)
     }
@@ -57,6 +59,12 @@ impl UniversalWriteFileOps for MmapFs {
     fn atomic_save(&self, path: &Path, bytes: &[u8]) -> UioResult<()> {
         local_file_ops::local_atomic_save(path, bytes)
     }
+
+    /// The very handle [`UniversalReadFs::open`] hands out: an [`MmapFile`]
+    /// appends through a dedicated `O_APPEND` fd of its own.
+    fn open_append(&self, path: impl AsRef<Path>, options: OpenOptions) -> UioResult<MmapFile> {
+        MmapFile::open_inner(path, options.for_append())
+    }
 }
 
 impl UniversalReadFs for MmapFs {

lib/common/common/src/universal_io/simple_disk_cache/fs.rs

@@ -12,7 +12,7 @@ use crate::universal_io::simple_disk_cache::REMOTE_OPEN_OPTIONS;
 use crate::universal_io::simple_disk_cache::local_state::LocalState;
 use crate::universal_io::{
     ListedFile, OpenExtra, OpenOptions, OwnedPipeline, Populate, UioResult, UniversalIoError,
-    UniversalRead, UniversalReadFileOps, UniversalReadFs, UniversalWriteFileOps,
+    UniversalRead, UniversalReadFileOps, UniversalReadFs,
 };
 
 /// Construction context for [`DiskCacheFs`]: carries the
@@ -123,31 +123,11 @@ where
     }
 }
 
-impl<R> UniversalWriteFileOps for DiskCacheFs<R>
-where
-    R: UniversalRead,
-    R::Fs: UniversalWriteFileOps,
-{
-    fn create(&self, path: &Path, expected_length: usize) -> UioResult<()> {
-        self.remote_fs.create(path, expected_length)
-    }
-
-    fn create_dir(&self, path: &Path) -> UioResult<()> {
-        self.remote_fs.create_dir(path)
-    }
-
-    fn remove(&self, path: &Path) -> UioResult<()> {
-        self.remote_fs.remove(path)
-    }
-
-    fn remove_dir(&self, path: &Path) -> UioResult<()> {
-        self.remote_fs.remove_dir(path)
-    }
-
-    fn atomic_save(&self, path: &Path, bytes: &[u8]) -> UioResult<()> {
-        self.remote_fs.atomic_save(path, bytes)
-    }
-}
+// Deliberately no `UniversalWriteFileOps` impl: the disk cache is strictly
+// read-only, at the filesystem level as much as at the file level (see the
+// `assert_not_impl_any!` on `DiskCache`). Mutations — creating, removing and
+// appending to files — go straight to the backing storage, whose handle the
+// caller holds anyway to build this one.
 
 /// Make the mirror path unique per open, so concurrently-alive [`DiskCache`]
 /// instances for the same remote path never share (and truncate) each other's

lib/common/common/src/universal_io/simple_disk_cache/tests.rs

@@ -16,15 +16,18 @@ use crate::universal_io::cached_fs::FileInfo;
 use crate::universal_io::{
     CachedFs, CachedReadFs, MmapFile, OpenOptions, Populate, ReadPipeline, ReadRange,
     UniversalAppend, UniversalFlush, UniversalIoError, UniversalRead, UniversalReadFileOps,
-    UniversalReadFs, UniversalWrite,
+    UniversalReadFs, UniversalWrite, UniversalWriteFileOps,
 };
 
 // The disk cache is strictly read-only: mutating it must stay a
 // compile-time error, on top of writeable opens being rejected at runtime
-// (covered per backend variant below).
+// (covered per backend variant below). This holds at the filesystem level
+// too — creating, removing and appending to files goes to the backing
+// storage, never through the cache.
 static_assertions::assert_not_impl_any!(
     DiskCache<MmapFile>: UniversalAppend, UniversalFlush, UniversalWrite
 );
+static_assertions::assert_not_impl_any!(DiskCacheFs<MmapFile>: UniversalWriteFileOps);
 
 fn make_test_data(n_bytes: usize) -> Vec<u8> {
     (0..n_bytes).map(|i| (i % 251) as u8).collect()

lib/common/common/src/universal_io/traits/append.rs

@@ -39,8 +39,10 @@ use crate::universal_io::{ByteOffset, UioResult};
 ///   acknowledgement). Retrying at the *same* offset is safe — it conflicts
 ///   rather than duplicating; re-check the length before appending at a new
 ///   offset.
-/// - Requires a handle opened with `writeable: true`. Not supported on
-///   `prevent_caching` (`O_DIRECT`) handles.
+/// - Requires a handle opened with `writeable: true` — either through
+///   [`UniversalWriteFileOps::open_append`], or through
+///   [`UniversalReadFs::open`] on a backend whose read handle appends. Not
+///   supported on `prevent_caching` (`O_DIRECT`) handles.
 /// - Appending no bytes trivially succeeds, without touching the file or
 ///   validating `offset`.
 /// - Offsets are plain byte offsets; no `T`-alignment is guaranteed or
@@ -50,6 +52,7 @@ use crate::universal_io::{ByteOffset, UioResult};
 ///
 /// [`AppendOffsetConflict`]: crate::universal_io::UniversalIoError::AppendOffsetConflict
 /// [`MmapFile`]: crate::universal_io::MmapFile
+/// [`UniversalReadFs::open`]: super::UniversalReadFs::open
 /// [`UniversalWrite`]: super::UniversalWrite
 /// [`UniversalWrite::write`]: super::UniversalWrite::write
 /// [`len`]: UniversalRead::len

lib/common/common/src/universal_io/traits/file_ops.rs

@@ -2,6 +2,7 @@ use std::fmt::Debug;
 use std::path::Path;
 
 use crate::universal_io::cached_fs::FileInfo;
+use crate::universal_io::traits::append::UniversalAppend;
 use crate::universal_io::traits::open_extra::OpenExtra;
 use crate::universal_io::traits::read::UniversalRead;
 use crate::universal_io::{ListedFile, OpenOptions, UioResult};
@@ -51,11 +52,21 @@ pub trait UniversalReadFileOps: Clone + Debug + Send + Sync + Sized {
 
 /// Filesystem-level handle for mutating operations.
 ///
-/// Extends [`UniversalReadFileOps`] with create/remove/save operations.
-/// Read-only backends (e.g. `ReadOnlyFs`) implement only the read side,
-/// making the absence of write support a compile-time property instead of
-/// a runtime error.
+/// Extends [`UniversalReadFileOps`] with create/remove/save operations and
+/// with opening append handles ([`Self::open_append`]). Read-only backends
+/// (e.g. `ReadOnlyFs`, the disk caches) implement only the read side, making
+/// the absence of write support a compile-time property instead of a runtime
+/// error.
 pub trait UniversalWriteFileOps: UniversalReadFileOps {
+    /// File handle type produced by [`Self::open_append`].
+    ///
+    /// Deliberately not tied to [`UniversalReadFs::File`]: the two capabilities
+    /// live on independent traits, so a backend may serve reads through one
+    /// handle type and appends through another (and a filesystem that opens no
+    /// read handles at all still names an append handle here). Backends whose
+    /// read handle appends — every local one — simply name it twice.
+    type AppendFile: UniversalAppend;
+
     /// Create or truncate a file at the given path.
     ///
     /// Local backends use `expected_length` to pre-size the file. Backends
@@ -81,6 +92,20 @@ pub trait UniversalWriteFileOps: UniversalReadFileOps {
     /// backends may overwrite the full object.
     fn atomic_save(&self, path: &Path, bytes: &[u8]) -> UioResult<()>;
 
+    /// Open an existing file for appending, per the [`UniversalAppend`]
+    /// contract. The file must exist — [`Self::create`] it first.
+    ///
+    /// `options` are honored as by [`UniversalReadFs::open`], except that
+    /// `writeable` is forced on ([`OpenOptions::for_append`]). There is no
+    /// [`OpenExtra`] counterpart: the append handle may come from a different
+    /// backend than [`UniversalReadFs::File`], whose per-open knobs would then
+    /// not apply.
+    fn open_append(
+        &self,
+        path: impl AsRef<Path>,
+        options: OpenOptions,
+    ) -> UioResult<Self::AppendFile>;
+
     // When adding provided methods, don't forget to update impls in
     // `crate::universal_io::wrappers::*`.
 }

lib/common/common/src/universal_io/types.rs

@@ -119,6 +119,28 @@ pub struct OpenOptions {
 }
 
 impl OpenOptions {
+    /// The same options with `writeable` forced on, as
+    /// [`UniversalWriteFileOps::open_append`] opens them: an append handle
+    /// mutates the file by definition, so the flag carries no information
+    /// there.
+    ///
+    /// [`UniversalWriteFileOps::open_append`]: super::UniversalWriteFileOps::open_append
+    pub fn for_append(self) -> Self {
+        let Self {
+            writeable: _,
+            need_sequential,
+            populate,
+            advice,
+        } = self;
+
+        Self {
+            writeable: true,
+            need_sequential,
+            populate,
+            advice,
+        }
+    }
+
     /// Default values for [`OpenOptions`].
     #[cfg(any(test, feature = "testing"))]
     pub fn new_for_test() -> Self {

lib/common/io_bridge/src/fs.rs

@@ -6,7 +6,7 @@ use common::universal_io::{
     UniversalWriteFileOps,
 };
 
-use crate::{AsyncRead, AsyncWrite, BlobFile, BridgeRuntime};
+use crate::{AsyncAppend, AsyncRead, BlobFile, BridgeRuntime};
 
 /// Filesystem handle for an object-store backend: an [`AsyncRead`] handle plus
 /// the [`BridgeRuntime`] used to drive its async operations. Opens per-object
@@ -74,7 +74,14 @@ impl<A: AsyncRead + Clone> UniversalReadFileOps for BlobFs<A> {
     }
 }
 
-impl<A: AsyncWrite + Clone> UniversalWriteFileOps for BlobFs<A> {
+/// Requires [`AsyncAppend`], not just [`AsyncWrite`](crate::AsyncWrite): a write-capable
+/// filesystem must hand out append handles, and only backends with a native
+/// single-request append can produce one (`BlobFile` appends through
+/// [`AsyncAppend::append`]). Backends that can only put whole objects stay
+/// read-only through universal I/O.
+impl<A: AsyncAppend + Clone> UniversalWriteFileOps for BlobFs<A> {
+    type AppendFile = BlobFile<A>;
+
     fn create(&self, path: &Path, _expected_length: usize) -> UioResult<()> {
         // Object stores have no fixed-size preallocation; the expected
         // length is ignored, as the trait allows.
@@ -100,6 +107,16 @@ impl<A: AsyncWrite + Clone> UniversalWriteFileOps for BlobFs<A> {
         self.runtime
             .block_on(self.inner.save(path, Bytes::copy_from_slice(bytes)))
     }
+
+    /// The very handle [`UniversalReadFs::open`] hands out, always writeable.
+    /// Performs no IO — the object is not touched until the first append or
+    /// read. Blob handles have no open-time knobs, so `options` is unused.
+    fn open_append(&self, path: impl AsRef<Path>, _options: OpenOptions) -> UioResult<BlobFile<A>> {
+        Ok(
+            BlobFile::new(self.inner.clone(), self.runtime.clone(), path.as_ref())
+                .with_writeable(true),
+        )
+    }
 }
 
 impl<A: AsyncRead + Clone> UniversalReadFs for BlobFs<A> {