You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by xu...@apache.org on 2026/09/22 08:27:42 UTC
(opendal-yinyang) 01/01: feat: read immutable file versions without staging
This is an automated email from the ASF dual-hosted git repository.
Xuanwo pushed a commit to branch xuanwo/core-version-reads
in repository https://gitbox.apache.org/repos/asf/opendal-yinyang.git
commit 31fa0fd6c17521d512620b0a1ef5e5e931108dfc
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Tue Sep 22 16:27:32 2026 +0800
feat: read immutable file versions without staging
Co-authored-by: Chojan Shang <ps...@apache.org>
---
crates/core/src/lib.rs | 1 +
crates/core/src/snapshot.rs | 66 +++++++++++++++++++++++++++++-
specs/content-profile.md | 7 ++++
specs/file-runtime.md | 21 ++++++++++
src/runtime.rs | 7 +---
tests/runtime.rs | 97 +++++++++++++++++++++++++++++++++++++++++++--
6 files changed, 190 insertions(+), 9 deletions(-)
diff --git a/crates/core/src/lib.rs b/crates/core/src/lib.rs
index b63638f..a10826f 100644
--- a/crates/core/src/lib.rs
+++ b/crates/core/src/lib.rs
@@ -36,6 +36,7 @@ pub use namespace::{DirectoryEntry, Link, Node, NodeKind};
pub use object::{
BackendProfile, ObjectFs as Fs, Outcome as CommitOutcome, Receipt, Revision, Snapshot,
};
+pub use snapshot::{DirectoryPage, FileVersion, ScanToken};
pub use transaction::{Planner, Transaction};
#[cfg(test)]
diff --git a/crates/core/src/snapshot.rs b/crates/core/src/snapshot.rs
index 4ffc37e..fc8a864 100644
--- a/crates/core/src/snapshot.rs
+++ b/crates/core/src/snapshot.rs
@@ -15,13 +15,15 @@
// specific language governing permissions and limitations
// under the License.
-use crate::data::{DataStore, PreparedContent};
+use crate::data::{ContentDescriptor, DataStore, PreparedContent};
use crate::namespace::{
DirectoryEntry, Node, NodeKind, corrupt, decode, encode, entry_key, prefix_end,
};
use crate::{CommitId, Error, ErrorKind, NodeId, Result};
use futures_util::future::BoxFuture;
+use std::ops::Range;
use std::sync::Arc;
+use tokio::io::AsyncWrite;
/// Logical record families shared by publication authorities.
#[derive(Clone, Copy, Debug)]
@@ -182,6 +184,53 @@ pub struct DirectoryPage {
pub next: Option<ScanToken>,
}
+/// An immutable file observation. Opening reads metadata only; content is
+/// authenticated on demand without a local staging directory or write lease.
+#[derive(Clone, Debug)]
+pub struct FileVersion {
+ data: DataStore,
+ revision: Revision,
+ node: Node,
+}
+impl FileVersion {
+ /// The observation that selected this version, not the volume's latest revision.
+ pub const fn revision(&self) -> Revision {
+ self.revision
+ }
+ /// Metadata from the same observation as the content, including its NodeId.
+ pub fn node(&self) -> &Node {
+ &self.node
+ }
+ pub fn descriptor(&self) -> &ContentDescriptor {
+ match self.node.kind() {
+ NodeKind::File(content) => content,
+ NodeKind::Directory { .. } => unreachable!("FileVersion requires a file node"),
+ }
+ }
+ /// Return up to EOF. Memory is proportional to returned bytes; use
+ /// read_range with a streaming destination for large transfers.
+ pub async fn read(&self, offset: u64, length: usize) -> Result<Vec<u8>> {
+ let size = self.descriptor().content_id().length();
+ let start = offset.min(size);
+ let end = offset.saturating_add(length as u64).min(size);
+ let mut bytes = Vec::new();
+ self.read_range(start..end, &mut bytes).await?;
+ Ok(bytes)
+ }
+ /// Authenticate intersecting units before delivering their bytes. The range
+ /// must lie within the file. A later error can leave a verified prefix in
+ /// the destination, but never an unverified unit or bytes from another version.
+ pub async fn read_range(
+ &self,
+ range: Range<u64>,
+ destination: &mut (impl AsyncWrite + Unpin),
+ ) -> Result<()> {
+ self.data
+ .read_range(self.descriptor(), range, destination)
+ .await
+ }
+}
+
/// A coherent, lazily read observation pinned to one retained revision.
#[derive(Clone, Debug)]
pub struct Snapshot {
@@ -201,6 +250,21 @@ impl Snapshot {
pub const fn filesystem(&self) -> NodeId {
self.filesystem
}
+ /// Pin a file's identity, metadata and content without reading its payload.
+ /// Clone the result for independent concurrent reads of this exact version.
+ pub async fn open_file(&self, id: NodeId) -> Result<FileVersion> {
+ let node = self.node(id).await?.ok_or_else(|| {
+ Error::new(ErrorKind::NotFound, "open file version", "node is absent")
+ })?;
+ if node.is_directory() {
+ return Err(Error::invalid("open file version", "node is a directory"));
+ }
+ Ok(FileVersion {
+ data: self.data.clone(),
+ revision: self.revision,
+ node,
+ })
+ }
pub async fn node(&self, id: NodeId) -> Result<Option<Node>> {
let node = self
.reader
diff --git a/specs/content-profile.md b/specs/content-profile.md
index a1d1c97..fae54f0 100644
--- a/specs/content-profile.md
+++ b/specs/content-profile.md
@@ -69,6 +69,13 @@ Each unit's physical and logical hashes are checked before releasing its
requested bytes. A failure in a later unit may leave a verified prefix. Unread
units are not scrubbed as a side effect of a range read.
+`Snapshot::open_file(NodeId)` binds a `FileVersion` to published node metadata
+and its revision without fetching payload. Its range reads reuse this profile's
+verification and do not prepare content or materialize local staging. Content
+and metadata stay pinned across concurrent namespace and file changes; a fresh
+snapshot is required to observe a newer version. This is an immutable reader,
+not a writable file handle or an offline content cache.
+
## Preparation authority
`ContentDescriptor` is serializable and untrusted. `PreparedContent` is not
diff --git a/specs/file-runtime.md b/specs/file-runtime.md
index 61ce163..f8866cc 100644
--- a/specs/file-runtime.md
+++ b/specs/file-runtime.md
@@ -10,6 +10,27 @@ filesystem must honor file locks and SQLite synchronization.
## Handles and visibility
+`Snapshot::open_file(NodeId)` returns an immutable `FileVersion`, independent of
+Runtime and staging. It reads only metadata when opening. `read(offset, length)`
+returns up to EOF, including an empty result for zero length or offsets beyond
+EOF. `read_range(range, destination)` streams an exact in-bounds range; invalid
+ranges fail before payload I/O. Only intersecting verification units and proof
+paths are read. Each unit is verified before delivery; a streaming error may
+leave a verified prefix. A Vec-returning read exposes no partial result on error.
+
+The version carries its observed node metadata, content descriptor and revision;
+clones read that same content even after remote writes, rename or unlink. It has
+no write methods, publication, staging lease or implicit rebasing. It needs
+access to the retained remote content, not a full local copy. Retain the NodeId
+and revision and reopen through the same authority to recover this observation.
+
+The recoverable `FileHandle` API below deliberately retains full materialization
+for both read-only and writable handles, including its offline staged reads.
+Use FileVersion for on-demand reads and provider downloads; use FileHandle for
+durable writable staging. Opening FileVersion does not create a staging record.
+Both paths use the same authenticated range reader. No remote format changes
+or local staging migrations are required.
+
Identity queries and enumeration use the authority's `Snapshot::node`, `lookup`
and `scan` APIs. Keep one snapshot across directory pages: a continuation is
bound to its filesystem, directory and revision, and cannot be applied to a
diff --git a/src/runtime.rs b/src/runtime.rs
index af3ec05..8da0c73 100644
--- a/src/runtime.rs
+++ b/src/runtime.rs
@@ -241,7 +241,7 @@ impl Runtime {
node: NodeId,
writable: bool,
) -> Result<FileHandle> {
- let file = snapshot.content(node).await?;
+ let file = snapshot.open_file(node).await?;
if file.descriptor().content_id().length() > i64::MAX as u64 {
return Err(Error::Invalid("file exceeds staging length limit"));
}
@@ -263,10 +263,7 @@ impl Runtime {
while offset < r.length {
let end = (offset + CHUNK).min(r.length);
let mut bytes = Vec::new();
- self.authority()
- .data()
- .read_range(file.descriptor(), offset..end, &mut bytes)
- .await?;
+ file.read_range(offset..end, &mut bytes).await?;
self.inner
.stage
.initial_chunk(r.id, offset / CHUNK, bytes)
diff --git a/tests/runtime.rs b/tests/runtime.rs
index e869663..bf688cc 100644
--- a/tests/runtime.rs
+++ b/tests/runtime.rs
@@ -23,7 +23,7 @@ use yinyang::core::service::{MetadataService, ServiceClient, serve};
use yinyang::core::{Authority, BackendProfile, CommitId, CommitOutcome, Fs, Planner};
use yinyang::runtime::{Error, Runtime};
-async fn exercise(authority: Arc<dyn Authority>, path: &std::path::Path) {
+async fn exercise(authority: Arc<dyn Authority>, path: &std::path::Path, backend: &TestBackend) {
let runtime = Runtime::open(authority.clone(), path).await.unwrap();
runtime.create_file(authority.root(), "file").await.unwrap();
let mut first = runtime.open_file("file", true).await.unwrap();
@@ -87,6 +87,97 @@ async fn exercise(authority: Arc<dyn Authority>, path: &std::path::Path) {
runtime.acknowledge_errors(sequence).await.unwrap();
assert!(runtime.status().await.unwrap().errors.is_empty());
identity_access(&runtime).await;
+ version_reads(&runtime, backend).await;
+}
+
+async fn version_reads(runtime: &Runtime, backend: &TestBackend) {
+ let root = runtime.authority().root();
+ let bytes: Vec<u8> = (0..16 * 65536).map(|i| (i % 251) as u8).collect();
+ let snapshot = runtime.authority().observe_latest().await.unwrap();
+ let prepared = runtime
+ .authority()
+ .prepare(&mut bytes.as_slice())
+ .await
+ .unwrap();
+ let mut plan = Planner::new(&snapshot, CommitId::generate());
+ let id = plan
+ .create_file(root, "large", prepared, false)
+ .await
+ .unwrap();
+ runtime
+ .authority()
+ .commit(&plan.finish().unwrap())
+ .await
+ .unwrap();
+ let snapshot = runtime.authority().observe_latest().await.unwrap();
+ let before = backend.state.lock().unwrap().data_read_bytes;
+ let file = snapshot.open_file(id).await.unwrap();
+ assert_eq!(file.node().id(), id);
+ assert_eq!(file.revision(), snapshot.revision());
+ // Metadata shares pack storage with content; opening may read metadata,
+ // but must not read the 1 MiB payload or create a stage record.
+ assert!(backend.state.lock().unwrap().data_read_bytes - before < 65536);
+ assert!(runtime.recoverable().await.unwrap().is_empty());
+ let before = backend.state.lock().unwrap().data_read_bytes;
+ assert_eq!(file.read(31, 10).await.unwrap(), bytes[31..41]);
+ let read_bytes = backend.state.lock().unwrap().data_read_bytes - before;
+ assert!((65536..2 * 65536).contains(&read_bytes));
+ let before = backend.state.lock().unwrap().data_read_bytes;
+ assert!(file.read(u64::MAX, 100).await.unwrap().is_empty());
+ assert!(file.read(31, 0).await.unwrap().is_empty());
+ assert!(
+ file.read_range(0..bytes.len() as u64 + 1, &mut Vec::new())
+ .await
+ .is_err()
+ );
+ assert_eq!(backend.state.lock().unwrap().data_read_bytes, before);
+ let clone = file.clone();
+ let (a, b) = tokio::join!(
+ file.read(65530, 20),
+ clone.read(bytes.len() as u64 - 4, 100)
+ );
+ assert_eq!(a.unwrap(), bytes[65530..65550]);
+ assert_eq!(b.unwrap(), bytes[bytes.len() - 4..]);
+ let mut writer = runtime.open_node(id, true).await.unwrap();
+ writer.write(0, b"changed").await.unwrap();
+ writer.close().await.unwrap();
+ runtime.rename(id, root, "large-moved").await.unwrap();
+ let latest = runtime
+ .authority()
+ .observe_latest()
+ .await
+ .unwrap()
+ .open_file(id)
+ .await
+ .unwrap();
+ assert_eq!(latest.read(0, 7).await.unwrap(), b"changed");
+ runtime.unlink(id).await.unwrap();
+ assert_eq!(file.read(0, 10).await.unwrap(), bytes[..10]);
+ let retained = runtime
+ .authority()
+ .observe_revision(file.revision())
+ .await
+ .unwrap()
+ .open_file(id)
+ .await
+ .unwrap();
+ assert_eq!(retained.read(0, 10).await.unwrap(), bytes[..10]);
+ // Corrupt the exact packed bytes read by this version; verified data must
+ // not escape merely because a previous read of the same version succeeded.
+ {
+ let mut state = backend.state.lock().unwrap();
+ let object = state
+ .objects
+ .values_mut()
+ .find(|v| v.bytes.starts_with(&bytes[..65536]))
+ .unwrap();
+ object.bytes[0] ^= 1;
+ }
+ let mut destination = Vec::new();
+ let err = file.read_range(0..10, &mut destination).await.unwrap_err();
+ assert_eq!(err.kind(), yinyang::core::ErrorKind::Corrupt);
+ assert!(destination.is_empty());
+ assert!(file.read(0, 10).await.is_err());
}
async fn identity_access(runtime: &Runtime) {
@@ -149,7 +240,7 @@ async fn file_operations_share_object_and_service_contract() {
let object = Fs::create(backend.operator(), BackendProfile::Minio)
.await
.unwrap();
- exercise(Arc::new(object), &temp.path().join("object")).await;
+ exercise(Arc::new(object), &temp.path().join("object"), &backend).await;
let backend = TestBackend::default();
let service = MetadataService::create(temp.path().join("metadata.db"), backend.operator())
.await
@@ -161,7 +252,7 @@ async fn file_operations_share_object_and_service_contract() {
let client = ServiceClient::connect(addr, token.into(), backend.operator())
.await
.unwrap();
- exercise(Arc::new(client), &temp.path().join("service")).await;
+ exercise(Arc::new(client), &temp.path().join("service"), &backend).await;
server.abort();
}