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:23:40 UTC
(opendal-yinyang) branch xuanwo/core-node-access created (now b60c7cf)
This is an automated email from the ASF dual-hosted git repository.
Xuanwo pushed a change to branch xuanwo/core-node-access
in repository https://gitbox.apache.org/repos/asf/opendal-yinyang.git
at b60c7cf feat: open managed files by node identity and revision
This branch includes the following new commits:
new b60c7cf feat: open managed files by node identity and revision
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(opendal-yinyang) 01/01: feat: open managed files by node identity and revision
Posted by xu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
Xuanwo pushed a commit to branch xuanwo/core-node-access
in repository https://gitbox.apache.org/repos/asf/opendal-yinyang.git
commit b60c7cf3b2a75c1206bddf877c743e769dc4a6c9
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Tue Sep 22 16:23:31 2026 +0800
feat: open managed files by node identity and revision
Co-authored-by: Chojan Shang <ps...@apache.org>
---
specs/file-runtime.md | 12 +++++++++++
src/runtime.rs | 43 +++++++++++++++++++++++++++++++++++-----
tests/runtime.rs | 55 +++++++++++++++++++++++++++++++++++++++++++++++++++
3 files changed, 105 insertions(+), 5 deletions(-)
diff --git a/specs/file-runtime.md b/specs/file-runtime.md
index b29170e..61ce163 100644
--- a/specs/file-runtime.md
+++ b/specs/file-runtime.md
@@ -10,6 +10,18 @@ filesystem must honor file locks and SQLite synchronization.
## Handles and visibility
+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
+fresh observation. NodeId is the identity; a path is only a convenience locator.
+
+`open_node(id, writable)` opens a node at latest, even after a rename.
+`open_node_at(revision, id, writable)` opens it in a retained revision validated
+by this runtime's authority. It can read a historical node after unlink; writes
+still conditionally publish against the original revision and cannot resurrect
+it. A revision from another filesystem is rejected. Reusing an old path for a
+new file never redirects an identity-based open to that new node.
+
`open_file(path, writable)` observes latest once, resolves a stable NodeId, and
materializes the pinned file into verified 64 KiB staging chunks. Memory used
for materialization and upload is bounded by chunks; local disk use includes
diff --git a/src/runtime.rs b/src/runtime.rs
index 6a73b9d..af3ec05 100644
--- a/src/runtime.rs
+++ b/src/runtime.rs
@@ -21,7 +21,9 @@ use std::collections::{BTreeMap, BTreeSet};
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use tokio::io::AsyncWriteExt;
-use yinyang_core::{Authority, CommitId, CommitOutcome, NodeId, Planner, Receipt, Revision};
+use yinyang_core::{
+ Authority, CommitId, CommitOutcome, NodeId, Planner, Receipt, Revision, Snapshot,
+};
mod staging;
use staging::{CHUNK, Record, Stage};
@@ -197,8 +199,8 @@ impl Runtime {
closed: false,
})
}
- /// Opens one pinned node version and materializes verified bytes in bounded
- /// chunks. Subsequent opens observe latest; existing handles are never rebased.
+ /// Resolves a path and opens its node in the same observation. Subsequent
+ /// opens observe latest; existing handles are never rebased.
pub async fn open_file(&self, path: &str, writable: bool) -> Result<FileHandle> {
if writable {
self.require_write()?;
@@ -208,13 +210,44 @@ impl Runtime {
.resolve(path)
.await?
.ok_or(Error::Invalid("file does not exist"))?;
- let file = snapshot.content(node.id()).await?;
+ self.open_snapshot_node(snapshot, node.id(), writable).await
+ }
+ /// Opens a stable node identity at latest, independently of its current name.
+ /// Use the authority's Snapshot for identity queries, lookup and enumeration.
+ pub async fn open_node(&self, node: NodeId, writable: bool) -> Result<FileHandle> {
+ if writable {
+ self.require_write()?;
+ }
+ let snapshot = self.authority().observe_latest().await?;
+ self.open_snapshot_node(snapshot, node, writable).await
+ }
+ /// Opens a node in a retained revision of this runtime's authority. A write
+ /// from an old revision still checks its original predicates at publication.
+ pub async fn open_node_at(
+ &self,
+ revision: Revision,
+ node: NodeId,
+ writable: bool,
+ ) -> Result<FileHandle> {
+ if writable {
+ self.require_write()?;
+ }
+ let snapshot = self.authority().observe_revision(revision).await?;
+ self.open_snapshot_node(snapshot, node, writable).await
+ }
+ async fn open_snapshot_node(
+ &self,
+ snapshot: Snapshot,
+ node: NodeId,
+ writable: bool,
+ ) -> Result<FileHandle> {
+ let file = snapshot.content(node).await?;
if file.descriptor().content_id().length() > i64::MAX as u64 {
return Err(Error::Invalid("file exceeds staging length limit"));
}
let mut r = Record {
id: *uuid::Uuid::new_v4().as_bytes(),
- node: *node.id().as_bytes(),
+ node: *node.as_bytes(),
base: snapshot.revision().to_bytes(),
length: file.descriptor().content_id().length(),
local: 0,
diff --git a/tests/runtime.rs b/tests/runtime.rs
index 41640ad..e869663 100644
--- a/tests/runtime.rs
+++ b/tests/runtime.rs
@@ -86,6 +86,61 @@ async fn exercise(authority: Arc<dyn Authority>, path: &std::path::Path) {
.sequence;
runtime.acknowledge_errors(sequence).await.unwrap();
assert!(runtime.status().await.unwrap().errors.is_empty());
+ identity_access(&runtime).await;
+}
+
+async fn identity_access(runtime: &Runtime) {
+ let root = runtime.authority().root();
+ runtime.create_directory(root, "dir").await.unwrap();
+ let snapshot = runtime.authority().observe_latest().await.unwrap();
+ let dir = snapshot.lookup(root, "dir").await.unwrap().unwrap().node_id;
+ runtime.create_file(dir, "a").await.unwrap();
+ runtime.create_file(dir, "b").await.unwrap();
+ let mut file = runtime.open_file("dir/a", true).await.unwrap();
+ file.write(0, b"original").await.unwrap();
+ file.close().await.unwrap();
+ let pinned = runtime.authority().observe_latest().await.unwrap();
+ let first = pinned.scan(dir, None, 1).await.unwrap();
+ let node = first.entries[0].node_id;
+ assert_eq!(first.entries[0].name, "a");
+ assert_eq!(pinned.node(node).await.unwrap().unwrap().id(), node);
+ runtime.rename(node, root, "moved").await.unwrap();
+ runtime.create_file(dir, "a").await.unwrap();
+ let latest = runtime.authority().observe_latest().await.unwrap();
+ assert!(latest.scan(dir, first.next.as_ref(), 1).await.is_err());
+ let second = pinned.scan(dir, first.next.as_ref(), 1).await.unwrap();
+ assert_eq!(second.entries[0].name, "b");
+ assert!(second.next.is_none());
+ let mut by_id = runtime.open_node(node, true).await.unwrap();
+ assert_eq!(by_id.read(0, 20).await.unwrap(), b"original");
+ let mut replacement = runtime.open_file("dir/a", false).await.unwrap();
+ assert_ne!(replacement.status().await.unwrap().node, node);
+ assert!(replacement.read(0, 20).await.unwrap().is_empty());
+ replacement.close().await.unwrap();
+ by_id.write(0, b"modified").await.unwrap();
+ by_id.close().await.unwrap();
+ let mut historical = runtime
+ .open_node_at(pinned.revision(), node, true)
+ .await
+ .unwrap();
+ assert_eq!(historical.read(0, 20).await.unwrap(), b"original");
+ historical.write(0, b"obsolete").await.unwrap();
+ assert!(matches!(historical.fsync().await, Err(Error::Conflict)));
+ historical.abort().await.unwrap();
+ runtime.unlink(node).await.unwrap();
+ assert!(runtime.open_node(node, false).await.is_err());
+ let mut retained = runtime
+ .open_node_at(pinned.revision(), node, false)
+ .await
+ .unwrap();
+ assert_eq!(retained.read(0, 20).await.unwrap(), b"original");
+ retained.close().await.unwrap();
+ assert!(runtime.open_node(root, false).await.is_err());
+ let foreign = Fs::create(TestBackend::default().operator(), BackendProfile::Minio)
+ .await
+ .unwrap();
+ let revision = foreign.observe_latest().await.unwrap().revision();
+ assert!(runtime.open_node_at(revision, node, false).await.is_err());
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn file_operations_share_object_and_service_contract() {