feat(parquet): route the manifest through the object-store seam (rfc0013 green)

crates/ourios-parquet/Cargo.toml

@@ -32,12 +32,14 @@ object_store = { version = "0.13" }
 # arrow/parquet tree; declared directly.
 bytes = { version = "1", default-features = false }
 # Drives the async `object_store` calls from the sync storage API
-# (`Store::*_blocking` bridge — `Writer`/`Reader`/compaction are sync,
-# `object_store` is async). Only `rt`: the bridge builds a `current_thread`
-# runtime per call on its own thread. The S3 backend adds `enable_all()` and the
-# `net`/`io`/`time` features in its slice. Already in the workspace tree via
-# DataFusion / object_store.
-tokio = { version = "1", default-features = false, features = ["rt"] }
+# (`Store::*_blocking` bridge — `Writer`/`Reader`/compaction/manifest are sync,
+# `object_store` is async). `rt-multi-thread` because the bridge shares one
+# process-wide runtime and drives it with `block_on` from many threads
+# concurrently (parallel queries resolving manifests); a multi-thread runtime
+# supports that, a current-thread one does not. The S3 backend adds
+# `enable_all()` and the `net`/`io`/`time` features in its slice. Already in the
+# workspace tree via DataFusion / object_store.
+tokio = { version = "1", default-features = false, features = ["rt", "rt-multi-thread"] }
 # UUIDv7 file naming per RFC 0005 §3.4 — timestamp-in-high-bits
 # means lexicographic sort = creation-order.
 uuid = { version = "1", default-features = false, features = ["v7", "std"] }

crates/ourios-parquet/src/manifest.rs

@@ -17,6 +17,14 @@ use std::path::Path;
 
 use serde::{Deserialize, Serialize};
 
+use crate::store::{Store, StoreError};
+
+/// Map a [`StoreError`] from the manifest seam onto [`ManifestError::Io`],
+/// keeping the backend cause in the error chain.
+fn store_io(err: StoreError) -> ManifestError {
+    ManifestError::Io(std::io::Error::other(err))
+}
+
 /// Canonical manifest filename inside a partition directory.
 pub const MANIFEST_FILENAME: &str = "manifest.json";
 
@@ -103,7 +111,8 @@ fn is_partition_local_parquet(name: &str) -> bool {
 }
 
 impl Manifest {
-    /// Read `<partition_dir>/manifest.json`.
+    /// Read `<partition_dir>/manifest.json` through the object-storage
+    /// [`Store`] seam (RFC 0013).
     ///
     /// `Ok(None)` when the manifest is absent — the pre-compaction
     /// (and current) case, where the reader falls back to globbing
@@ -114,15 +123,24 @@ impl Manifest {
     /// [`ManifestError`] if the file exists but can't be read, or its
     /// bytes aren't valid manifest JSON.
     pub fn read(partition_dir: &Path) -> Result<Option<Self>, ManifestError> {
-        match std::fs::read(partition_dir.join(MANIFEST_FILENAME)) {
-            Ok(bytes) => {
+        // A missing partition directory means no manifest (matches the prior
+        // `std::fs` NotFound→None). Checked up front because `Store::local`
+        // canonicalises its root and would error on an absent directory.
+        if !partition_dir.try_exists().map_err(ManifestError::Io)? {
+            return Ok(None);
+        }
+        let store = Store::local(partition_dir).map_err(store_io)?;
+        match store
+            .get_blocking_opt(MANIFEST_FILENAME)
+            .map_err(store_io)?
+        {
+            Some(bytes) => {
                 let manifest: Self =
                     serde_json::from_slice(&bytes).map_err(ManifestError::Parse)?;
                 manifest.validate()?;
                 Ok(Some(manifest))
             }
-            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
-            Err(e) => Err(ManifestError::Io(e)),
+            None => Ok(None),
         }
     }
 
@@ -157,35 +175,35 @@ impl Manifest {
         serde_json::to_vec(self)
     }
 
-    /// Atomically (re)write the partition's manifest: serialize to a
-    /// sibling `manifest.json.tmp`, then `rename` it over
-    /// `manifest.json`. The rename is the **commit point** (RFC 0009
-    /// §3.4) — a *concurrent reader* observes either the old manifest
-    /// or the new one, never a partial write. The manifest is
-    /// validated before any bytes hit disk, so an invalid set is never
-    /// published.
+    /// Atomically (re)write the partition's manifest through the
+    /// object-storage [`Store`] seam (RFC 0013). The `Store` put is the
+    /// **commit point** (RFC 0009 §3.4): the local backend stages to a
+    /// private temp object and renames it into place, so a *concurrent
+    /// reader* observes either the old manifest or the new one, never a
+    /// partial write. The manifest is validated before any bytes are
+    /// written, so an invalid set is never published.
     ///
-    /// This is atomic, **not** crash-durable: without an `fsync` of the
-    /// file and its directory, a host crash mid-write can still leave a
-    /// truncated or missing `manifest.json` (the same caveat as the
-    /// `Writer`). That is safe by construction — a reader with no
-    /// manifest falls back to the `*.parquet` glob, and a compaction
-    /// that crashed before this commit left its inputs intact — so the
-    /// worst case is reverting to the prior generation, never data
-    /// loss. Durable fsync is a later refinement.
+    /// On the local backend this is last-writer-wins `Overwrite`, atomic but
+    /// **not** crash-durable (no `fsync`) — the same contract as before the
+    /// seam, and the same caveat as the `Writer`. That is safe by
+    /// construction: a reader with no manifest falls back to the `*.parquet`
+    /// glob, and a compaction that crashed before this commit left its inputs
+    /// intact, so the worst case is reverting to the prior generation, never
+    /// data loss. Compare-and-swap (generation CAS, `If-Match`) lands with the
+    /// S3 backend — `LocalFileSystem` does not support conditional update.
     ///
     /// # Errors
     ///
     /// [`ManifestError::InvalidFilename`] if any entry isn't a
-    /// partition-local `*.parquet` name; [`ManifestError::Io`] on a
-    /// write or rename failure.
+    /// partition-local `*.parquet` name; [`ManifestError::Io`] on a write
+    /// failure.
     pub fn write_atomic(&self, partition_dir: &Path) -> Result<(), ManifestError> {
         self.validate()?;
-        let bytes = serde_json::to_vec(self).map_err(ManifestError::Parse)?;
-        let tmp = partition_dir.join(format!("{MANIFEST_FILENAME}.tmp"));
-        let final_path = partition_dir.join(MANIFEST_FILENAME);
-        std::fs::write(&tmp, &bytes).map_err(ManifestError::Io)?;
-        std::fs::rename(&tmp, &final_path).map_err(ManifestError::Io)?;
+        let bytes = self.to_json().map_err(ManifestError::Parse)?;
+        let store = Store::local(partition_dir).map_err(store_io)?;
+        store
+            .put_blocking(MANIFEST_FILENAME, bytes)
+            .map_err(store_io)?;
         Ok(())
     }
 }

crates/ourios-parquet/src/store.rs

@@ -16,49 +16,74 @@
 //! depend on this crate, so the type is visible to every storage consumer.
 
 use std::future::Future;
-use std::sync::Arc;
+use std::sync::{Arc, OnceLock};
 
 use object_store::local::LocalFileSystem;
 use object_store::path::Path as ObjectPath;
-use object_store::{ObjectStore, ObjectStoreExt, PutPayload};
+use object_store::{ObjectStore, ObjectStoreExt, PutMode, PutOptions, PutPayload};
+use tokio::runtime::Runtime;
+
+/// The process-wide runtime that drives the async `object_store` calls behind
+/// the sync storage API. Built once, lazily, via `get_or_init` so there is no
+/// init-race that could drop a surplus runtime on a caller's thread (an
+/// earlier manual `get`/`set` did, panicking when the loser was inside a tokio
+/// runtime). The runtime lives for the process and is never dropped, so the
+/// "drop a runtime in async context" hazard can't arise.
+///
+/// Multi-threaded(1-worker) so concurrent `block_on` from many bridge threads
+/// (parallel queries / tests) is safe. No `enable_all()`: the local backend
+/// drives I/O via `spawn_blocking`, which needs only the bare runtime; the S3
+/// backend adds `enable_all()` plus the `net`/`io`/`time` tokio features in the
+/// slice that introduces it.
+fn bridge_runtime() -> Result<&'static Runtime, StoreError> {
+    static RT: OnceLock<std::io::Result<Runtime>> = OnceLock::new();
+    match RT.get_or_init(|| {
+        tokio::runtime::Builder::new_multi_thread()
+            .worker_threads(1)
+            .build()
+    }) {
+        Ok(rt) => Ok(rt),
+        // Build failure is cached (a permanent resource exhaustion); rebuild a
+        // fresh `io::Error` since it isn't `Clone`.
+        Err(e) => Err(StoreError::Runtime(std::io::Error::new(
+            e.kind(),
+            e.to_string(),
+        ))),
+    }
+}
 
 /// Drive `fut` to completion synchronously — the bridge from the **sync**
 /// storage API (`Writer`, `Reader`, `compaction`, the manifest) to async
 /// `object_store` (compaction must reach S3 per RFC0013.3, so a local-only
 /// `std::fs` shortcut won't do).
 ///
-/// Everything happens on a **fresh OS thread**: a single-threaded runtime is
-/// built, drives `fut`, and is dropped, all on a thread that never carries the
-/// caller's tokio context. That makes the bridge safe from any call site —
-/// including *inside* a runtime (a `#[tokio::test]` that opens the reader, or
-/// a future async consumer), where calling `block_on` (or dropping a runtime)
-/// on the caller's own thread would panic. [`std::thread::scope`] lets `fut`
-/// borrow the caller's `self`/`key` while still running off-thread.
+/// `block_on` runs on a **fresh OS thread** (the shared [`bridge_runtime`] is
+/// driven from there), not the caller's. A plain thread never carries the
+/// caller's tokio context, so this is safe from any call site — including
+/// *inside* a runtime (e.g. the querier resolving manifests on its async task,
+/// or a `#[tokio::test]`), where `block_on` on the caller's own thread would
+/// panic. [`std::thread::scope`] lets `fut` borrow the caller's `self`/`key`
+/// while still running off-thread. Reusing the shared runtime keeps the
+/// per-call cost to one thread spawn (no per-call runtime build), which matters
+/// on the query path (`resolve_live_files` reads one manifest per partition).
 ///
 /// `fut` already yields a [`StoreError`] result, returned directly; the extra
-/// error modes are building the bridge thread or its runtime
+/// error modes are building the bridge thread or runtime
 /// ([`StoreError::Runtime`]). A panic *inside* `fut` is not swallowed — it is
-/// re-raised on the caller's thread via [`std::panic::resume_unwind`]. No
-/// `enable_all()`: the local backend drives I/O via `spawn_blocking`, which
-/// needs only the bare runtime; the S3 backend adds `enable_all()` and the
-/// `net`/`io`/`time` tokio features in the slice that introduces it.
+/// re-raised on the caller's thread via [`std::panic::resume_unwind`].
 fn block_on_off_runtime<T>(
     fut: impl Future<Output = Result<T, StoreError>> + Send,
 ) -> Result<T, StoreError>
 where
     T: Send,
 {
+    let rt = bridge_runtime()?;
     std::thread::scope(|s| {
         // `Builder::spawn_scoped` (not `Scope::spawn`) so OS thread-creation
         // failure surfaces as `StoreError::Runtime` rather than panicking.
         let handle = std::thread::Builder::new()
             .name("ourios-store-bridge".into())
-            .spawn_scoped(s, || {
-                let rt = tokio::runtime::Builder::new_current_thread()
-                    .build()
-                    .map_err(StoreError::Runtime)?;
-                rt.block_on(fut)
-            })
+            .spawn_scoped(s, || rt.block_on(fut))
             .map_err(StoreError::Runtime)?;
         handle
             .join()
@@ -137,6 +162,15 @@ impl std::error::Error for StoreError {
     }
 }
 
+impl StoreError {
+    /// True if this is a "no such object" backend error — the caller may
+    /// treat the object as absent (see [`Store::get_blocking_opt`]).
+    #[must_use]
+    pub fn is_not_found(&self) -> bool {
+        matches!(self, Self::Backend(object_store::Error::NotFound { .. }))
+    }
+}
+
 impl Store {
     /// Local-filesystem backend rooted at `root` (dev / test / CI). Preserves
     /// today's on-disk layout — the RFC 0005 Hive keys become paths under
@@ -258,11 +292,54 @@ impl Store {
     pub fn put_blocking(&self, key: &str, bytes: Vec<u8>) -> Result<(), StoreError> {
         block_on_off_runtime(self.put(key, bytes))
     }
+
+    /// Write `bytes` to `key` only if no object exists there
+    /// (create-if-absent — `If-None-Match: *`). The local-testable half of
+    /// RFC 0013 conditional PUT; the compare-and-swap half (`If-Match`) needs
+    /// an S3 backend, since `LocalFileSystem` rejects `PutMode::Update`.
+    ///
+    /// # Errors
+    /// [`StoreError::Backend`] if an object already exists at `key`, or the
+    /// put otherwise fails.
+    pub async fn put_if_absent(&self, key: &str, bytes: Vec<u8>) -> Result<(), StoreError> {
+        self.inner
+            .put_opts(
+                &self.resolve(key),
+                PutPayload::from(bytes),
+                PutOptions::from(PutMode::Create),
+            )
+            .await
+            .map_err(StoreError::Backend)?;
+        Ok(())
+    }
+
+    /// Read the object at `key`, mapping a missing object to `None` rather
+    /// than an error — for sync call sites where absence is expected (e.g. a
+    /// partition with no manifest yet).
+    ///
+    /// # Errors
+    /// As [`Self::get_blocking`], except a not-found object yields `Ok(None)`.
+    pub fn get_blocking_opt(&self, key: &str) -> Result<Option<Vec<u8>>, StoreError> {
+        match self.get_blocking(key) {
+            Ok(bytes) => Ok(Some(bytes)),
+            Err(e) if e.is_not_found() => Ok(None),
+            Err(e) => Err(e),
+        }
+    }
+
+    /// Blocking [`Self::put_if_absent`] for the sync storage call sites.
+    ///
+    /// # Errors
+    /// As [`Self::put_if_absent`], plus [`StoreError::Runtime`] if the bridge
+    /// runtime can't be built.
+    pub fn put_if_absent_blocking(&self, key: &str, bytes: Vec<u8>) -> Result<(), StoreError> {
+        block_on_off_runtime(self.put_if_absent(key, bytes))
+    }
 }
 
 #[cfg(test)]
 mod tests {
-    use super::Store;
+    use super::{Store, StoreError};
 
     /// A byte object round-trips through the local backend, and a delete
     /// removes it. (Foundation for the RFC0013 consumer migration; the §5
@@ -312,4 +389,46 @@ mod tests {
             b"inside-runtime"
         );
     }
+
+    /// `get_blocking_opt` maps a missing object to `None` (the manifest's
+    /// "no manifest yet" case) and yields the bytes when present.
+    #[test]
+    fn get_blocking_opt_maps_missing_to_none() {
+        let dir = tempfile::TempDir::new().expect("temp dir");
+        let store = Store::local(dir.path()).expect("local store");
+        assert_eq!(
+            store.get_blocking_opt("manifest.json").expect("get_opt"),
+            None,
+            "absent object is None, not an error"
+        );
+        store
+            .put_blocking("manifest.json", b"{}".to_vec())
+            .expect("put");
+        assert_eq!(
+            store.get_blocking_opt("manifest.json").expect("get_opt"),
+            Some(b"{}".to_vec()),
+        );
+    }
+
+    /// `put_if_absent` (create-if-absent) writes when the key is free and
+    /// refuses to clobber an existing object — the local-testable half of
+    /// RFC 0013 conditional PUT.
+    #[test]
+    fn put_if_absent_refuses_to_clobber() {
+        let dir = tempfile::TempDir::new().expect("temp dir");
+        let store = Store::local(dir.path()).expect("local store");
+        let key = "manifest.json";
+        store
+            .put_if_absent_blocking(key, b"first".to_vec())
+            .expect("first create");
+        let err = store
+            .put_if_absent_blocking(key, b"second".to_vec())
+            .expect_err("create over an existing object must fail");
+        assert!(matches!(err, StoreError::Backend(_)), "got {err:?}");
+        assert_eq!(
+            store.get_blocking(key).expect("get"),
+            b"first",
+            "the original object is untouched"
+        );
+    }
 }