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/21 11:46:18 UTC
(opendal-yinyang) branch xuanwo/metadata-service created (now 8a3cc0f)
This is an automated email from the ASF dual-hosted git repository.
Xuanwo pushed a change to branch xuanwo/metadata-service
in repository https://gitbox.apache.org/repos/asf/opendal-yinyang.git
at 8a3cc0f feat: add durable metadata service publication
This branch includes the following new commits:
new 8a3cc0f feat: add durable metadata service publication
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: add durable metadata service publication
Posted by xu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
Xuanwo pushed a commit to branch xuanwo/metadata-service
in repository https://gitbox.apache.org/repos/asf/opendal-yinyang.git
commit 8a3cc0faea855a31dcbd27ef1704719b2dfa0ebf
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Mon Sep 21 19:46:08 2026 +0800
feat: add durable metadata service publication
Co-authored-by: Chojan Shang <ps...@apache.org>
---
Cargo.lock | 122 +++++++++
Cargo.toml | 2 +-
crates/core/Cargo.toml | 4 +-
crates/core/src/authority.rs | 150 +++++++++++
crates/core/src/lib.rs | 3 +
crates/core/src/object.rs | 6 +
crates/core/src/service.rs | 496 +++++++++++++++++++++++++++++++++++
crates/core/src/service/transport.rs | 473 +++++++++++++++++++++++++++++++++
crates/core/src/transaction.rs | 231 ++++++++++++++++
crates/core/tests/service.rs | 422 +++++++++++++++++++++++++++++
specs/README.md | 3 +-
specs/metadata-service.md | 111 ++++++++
specs/transactional-filesystem.md | 8 +-
src/main.rs | 22 +-
tests/service.rs | 131 +++++++++
15 files changed, 2176 insertions(+), 8 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index fcfe274..5efd074 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -155,10 +155,24 @@ version = "1.8.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "553c5d846a6ba5150c65e3b1b8ec073bcf1abc20f9b7220de384a4443ea4e20a"
dependencies = [
+ "borsh-derive",
"bytes",
"cfg_aliases",
]
+[[package]]
+name = "borsh-derive"
+version = "1.8.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "12cdfe656708a01f89b451a7d36466e6fe6c414de0aa18fc54f864f6f9ca9f56"
+dependencies = [
+ "once_cell",
+ "proc-macro-crate",
+ "proc-macro2",
+ "quote",
+ "syn 3.0.3",
+]
+
[[package]]
name = "bumpalo"
version = "3.20.3"
@@ -456,6 +470,12 @@ version = "1.0.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813"
+[[package]]
+name = "equivalent"
+version = "1.0.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f"
+
[[package]]
name = "errno"
version = "0.3.14"
@@ -466,6 +486,18 @@ dependencies = [
"windows-sys 0.61.2",
]
+[[package]]
+name = "fallible-iterator"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2acce4a10f12dc2fb14a218589d4f1f62ef011b2d0cc4b3cb1bba8e94da14649"
+
+[[package]]
+name = "fallible-streaming-iterator"
+version = "0.1.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7360491ce676a36bf9bb3c56c1aa791658183a54d2744120f27285738d90465a"
+
[[package]]
name = "fastrand"
version = "2.5.0"
@@ -861,6 +893,16 @@ dependencies = [
"icu_properties",
]
+[[package]]
+name = "indexmap"
+version = "2.14.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cc4e190f5d26ca7051642629da2c52fc03bde85a03197c99408dcd291734c855"
+dependencies = [
+ "equivalent",
+ "hashbrown 0.17.1",
+]
+
[[package]]
name = "ipnet"
version = "2.12.2"
@@ -1010,6 +1052,17 @@ version = "0.2.189"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2"
+[[package]]
+name = "libsqlite3-sys"
+version = "0.38.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f1d20bef17f513b9b3004532233187769cd072d790971f4e4da0e346eb6401e8"
+dependencies = [
+ "cc",
+ "pkg-config",
+ "vcpkg",
+]
+
[[package]]
name = "linux-raw-sys"
version = "0.12.1"
@@ -1213,6 +1266,15 @@ dependencies = [
"zerovec",
]
+[[package]]
+name = "proc-macro-crate"
+version = "3.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e67ba7e9b2b56446f1d419b1d807906278ffa1a658a8a5d8a39dcb1f5a78614f"
+dependencies = [
+ "toml_edit",
+]
+
[[package]]
name = "proc-macro2"
version = "1.0.107"
@@ -1451,6 +1513,19 @@ dependencies = [
"windows-sys 0.52.0",
]
+[[package]]
+name = "rusqlite"
+version = "0.40.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "23f2a97da3e3873c73cb2a2e71b35c40ff95e0b1eefa8d72d8499a6928c3b5b3"
+dependencies = [
+ "bitflags 2.13.2",
+ "fallible-iterator",
+ "fallible-streaming-iterator",
+ "libsqlite3-sys",
+ "smallvec",
+]
+
[[package]]
name = "rust-ini"
version = "0.21.3"
@@ -1926,6 +2001,36 @@ dependencies = [
"tokio",
]
+[[package]]
+name = "toml_datetime"
+version = "1.1.1+spec-1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3165f65f62e28e0115a00b2ebdd37eb6f3b641855f9d636d3cd4103767159ad7"
+dependencies = [
+ "serde_core",
+]
+
+[[package]]
+name = "toml_edit"
+version = "0.25.15+spec-1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1340ea94a5856333492c9064b02c778b191dd2c853778d9609debdcdfea3a614"
+dependencies = [
+ "indexmap",
+ "toml_datetime",
+ "toml_parser",
+ "winnow",
+]
+
+[[package]]
+name = "toml_parser"
+version = "1.1.3+spec-1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1d38ac1cf9b95face32296c0a3ede1fdc270627c9d9c02a7274dd6d960dc4d56"
+dependencies = [
+ "winnow",
+]
+
[[package]]
name = "tower"
version = "0.5.3"
@@ -2065,6 +2170,12 @@ dependencies = [
"wasm-bindgen",
]
+[[package]]
+name = "vcpkg"
+version = "0.2.15"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "accd4ea62f7bb7a82fe23066fb0957d48ef677f6eeb8215f372f52e48bb32426"
+
[[package]]
name = "version_check"
version = "0.9.5"
@@ -2299,6 +2410,15 @@ version = "0.52.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec"
+[[package]]
+name = "winnow"
+version = "1.0.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "23b97319f7b8343df12cc98938e5c3eb436064524c8d2b4e30a1d3a36eecdf81"
+dependencies = [
+ "memchr",
+]
+
[[package]]
name = "writeable"
version = "0.6.4"
@@ -2334,6 +2454,8 @@ dependencies = [
"borsh",
"futures-util",
"opendal",
+ "rusqlite",
+ "tempfile",
"tokio",
"unicode-casefold",
"unicode-normalization",
diff --git a/Cargo.toml b/Cargo.toml
index d95b701..5f394af 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -56,7 +56,7 @@ rust-version = "1.91.0"
[workspace.dependencies]
blake3 = "1"
-borsh = { version = "1.8.1", default-features = false, features = ["std"] }
+borsh = { version = "1.8.1", default-features = false, features = ["derive", "std"] }
clap = { version = "4.6.5", features = ["derive"] }
futures-util = "0.3"
opendal = { version = "0.58.2", default-features = false }
diff --git a/crates/core/Cargo.toml b/crates/core/Cargo.toml
index 497f6e1..d4e0f13 100644
--- a/crates/core/Cargo.toml
+++ b/crates/core/Cargo.toml
@@ -30,7 +30,8 @@ blake3.workspace = true
borsh.workspace = true
futures-util.workspace = true
opendal.workspace = true
-tokio.workspace = true
+rusqlite = { version = "0.40.2", default-features = false, features = ["bundled"] }
+tokio = { workspace = true, features = ["net", "rt", "sync", "time"] }
unicode-casefold.workspace = true
unicode-normalization.workspace = true
uuid.workspace = true
@@ -41,6 +42,7 @@ opendal = { workspace = true, features = [
"services-memory",
"services-s3",
] }
+tempfile = "3"
tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
[lints]
diff --git a/crates/core/src/authority.rs b/crates/core/src/authority.rs
new file mode 100644
index 0000000..ec3550a
--- /dev/null
+++ b/crates/core/src/authority.rs
@@ -0,0 +1,150 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Shared publication interface; access runtimes do not select physical metadata.
+use crate::object::{ObjectFs, Outcome};
+use crate::{DataStore, NodeId, PreparedContent, Result, Revision, Snapshot, Transaction};
+use futures_util::future::BoxFuture;
+use tokio::io::AsyncRead;
+
+pub trait Authority: std::fmt::Debug + Send + Sync {
+ fn data(&self) -> &DataStore;
+ fn filesystem(&self) -> NodeId;
+ fn root(&self) -> NodeId;
+ fn observe_latest(&self) -> BoxFuture<'_, Result<Snapshot>>;
+ fn observe_revision(&self, revision: Revision) -> BoxFuture<'_, Result<Snapshot>>;
+ fn commit_batch<'a>(
+ &'a self,
+ requests: &'a [Transaction],
+ ) -> BoxFuture<'a, Result<Vec<Outcome>>>;
+ fn register<'a>(&'a self, content: &'a PreparedContent) -> BoxFuture<'a, Result<()>> {
+ Box::pin(async move {
+ self.data().accept(content)?;
+ Ok(())
+ })
+ }
+ fn prepare<'a>(
+ &'a self,
+ mut source: &'a mut (dyn AsyncRead + Unpin + Send),
+ ) -> BoxFuture<'a, Result<PreparedContent>> {
+ Box::pin(async move {
+ let content = self.data().prepare(&mut source).await?;
+ self.register(&content).await?;
+ Ok(content)
+ })
+ }
+ fn commit<'a>(&'a self, request: &'a Transaction) -> BoxFuture<'a, Result<Outcome>> {
+ Box::pin(async move {
+ Ok(self
+ .commit_batch(std::slice::from_ref(request))
+ .await?
+ .remove(0))
+ })
+ }
+ fn restore_transaction<'a>(&'a self, bytes: &'a [u8]) -> BoxFuture<'a, Result<Transaction>> {
+ Box::pin(async move {
+ let (revision, files) = Transaction::inspect(bytes)?;
+ let snapshot = self.observe_revision(revision).await?;
+ let mut prepared = Vec::new();
+ for file in files {
+ prepared.push(self.data().import(&file).await?);
+ }
+ Transaction::restore(bytes, &snapshot, &prepared).await
+ })
+ }
+}
+impl Authority for ObjectFs {
+ fn data(&self) -> &DataStore {
+ self.data()
+ }
+ fn filesystem(&self) -> NodeId {
+ self.filesystem()
+ }
+ fn root(&self) -> NodeId {
+ self.root()
+ }
+ fn observe_latest(&self) -> BoxFuture<'_, Result<Snapshot>> {
+ Box::pin(self.observe_latest())
+ }
+ fn observe_revision(&self, r: Revision) -> BoxFuture<'_, Result<Snapshot>> {
+ Box::pin(self.observe_revision(r))
+ }
+ fn commit_batch<'a>(
+ &'a self,
+ requests: &'a [Transaction],
+ ) -> BoxFuture<'a, Result<Vec<Outcome>>> {
+ Box::pin(self.commit_batch(requests))
+ }
+}
+impl Authority for crate::service::MetadataService {
+ fn data(&self) -> &DataStore {
+ self.data()
+ }
+ fn filesystem(&self) -> NodeId {
+ self.filesystem()
+ }
+ fn root(&self) -> NodeId {
+ self.root()
+ }
+ fn observe_latest(&self) -> BoxFuture<'_, Result<Snapshot>> {
+ Box::pin(self.observe_latest())
+ }
+ fn observe_revision(&self, r: Revision) -> BoxFuture<'_, Result<Snapshot>> {
+ Box::pin(self.observe_revision(r))
+ }
+ fn commit_batch<'a>(
+ &'a self,
+ requests: &'a [Transaction],
+ ) -> BoxFuture<'a, Result<Vec<Outcome>>> {
+ Box::pin(self.commit_batch(requests))
+ }
+ fn register<'a>(&'a self, content: &'a PreparedContent) -> BoxFuture<'a, Result<()>> {
+ Box::pin(self.remember(content))
+ }
+ fn restore_transaction<'a>(&'a self, bytes: &'a [u8]) -> BoxFuture<'a, Result<Transaction>> {
+ Box::pin(self.restore_transaction(bytes))
+ }
+}
+impl Authority for crate::service::ServiceClient {
+ fn data(&self) -> &DataStore {
+ self.data()
+ }
+ fn filesystem(&self) -> NodeId {
+ self.filesystem()
+ }
+ fn root(&self) -> NodeId {
+ self.root()
+ }
+ fn observe_latest(&self) -> BoxFuture<'_, Result<Snapshot>> {
+ Box::pin(self.observe_latest())
+ }
+ fn observe_revision(&self, r: Revision) -> BoxFuture<'_, Result<Snapshot>> {
+ Box::pin(self.observe_revision(r))
+ }
+ fn commit_batch<'a>(
+ &'a self,
+ requests: &'a [Transaction],
+ ) -> BoxFuture<'a, Result<Vec<Outcome>>> {
+ Box::pin(self.commit_batch(requests))
+ }
+ fn register<'a>(&'a self, content: &'a PreparedContent) -> BoxFuture<'a, Result<()>> {
+ Box::pin(self.register(content))
+ }
+ fn restore_transaction<'a>(&'a self, bytes: &'a [u8]) -> BoxFuture<'a, Result<Transaction>> {
+ Box::pin(self.restore_transaction(bytes))
+ }
+}
diff --git a/crates/core/src/lib.rs b/crates/core/src/lib.rs
index 1228ae1..b63638f 100644
--- a/crates/core/src/lib.rs
+++ b/crates/core/src/lib.rs
@@ -17,12 +17,15 @@
//! Indexed transactional filesystem and authenticated immutable content.
+mod authority;
pub mod data;
+pub use authority::Authority;
mod error;
mod identity;
mod index;
pub mod namespace;
pub mod object;
+pub mod service;
mod snapshot;
pub mod transaction;
diff --git a/crates/core/src/object.rs b/crates/core/src/object.rs
index f2b3998..abfe28b 100644
--- a/crates/core/src/object.rs
+++ b/crates/core/src/object.rs
@@ -497,6 +497,12 @@ async fn read_head(operator: &Operator) -> Result<Option<Head>> {
bytes.extend_from_slice(&chunk);
}
}
+ if bytes.starts_with(b"YYSERV01") {
+ return Err(Error::unsupported(
+ "open filesystem",
+ "metadata-service authority required",
+ ));
+ }
if bytes.starts_with(b"YYHEAD01") {
return Err(Error::unsupported(
"open filesystem",
diff --git a/crates/core/src/service.rs b/crates/core/src/service.rs
new file mode 100644
index 0000000..8126587
--- /dev/null
+++ b/crates/core/src/service.rs
@@ -0,0 +1,496 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Durable single-authority metadata, independent of object-store head CAS.
+//! SQLite files must reside on a local filesystem that honors sync and locking.
+
+use crate::namespace::{Node, NodeKind, decode, encode};
+use crate::snapshot::{
+ ChangeRecord, Cursor, Outcome, Receipt, RecordReader, Rows, Snapshot, Table,
+};
+use crate::{
+ ContentDescriptor, DataStore, Error, ErrorKind, NodeId, PreparedContent, Result, Revision,
+ Transaction,
+};
+use futures_util::future::BoxFuture;
+use opendal::Operator;
+use rusqlite::{Connection, OptionalExtension, params};
+use std::path::{Path, PathBuf};
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+use tokio::io::AsyncRead;
+
+mod transport;
+pub use transport::{ServiceClient, serve};
+
+const PROFILE: &str = "yinyang-sqlite-1";
+type Db = Arc<Mutex<Connection>>;
+
+/// One durable filesystem authority. Clones and independent processes coordinate
+/// through SQLite; no process-local cache determines the latest revision.
+#[derive(Clone, Debug)]
+pub struct MetadataService {
+ path: PathBuf,
+ data: DataStore,
+ filesystem: NodeId,
+ root: NodeId,
+}
+
+fn sql(error: rusqlite::Error) -> Error {
+ Error::new(ErrorKind::Storage, "metadata database", error.to_string())
+}
+async fn db_call<T: Send + 'static>(
+ db: &Db,
+ call: impl FnOnce(&mut Connection) -> Result<T> + Send + 'static,
+) -> Result<T> {
+ let db = db.clone();
+ tokio::task::spawn_blocking(move || {
+ let mut connection = db.lock().map_err(|_| {
+ Error::new(
+ ErrorKind::Storage,
+ "metadata database",
+ "connection lock poisoned",
+ )
+ })?;
+ call(&mut connection)
+ })
+ .await
+ .map_err(|e| {
+ Error::new(
+ ErrorKind::Storage,
+ "metadata database worker",
+ e.to_string(),
+ )
+ })?
+}
+async fn connect(path: &Path, create: bool) -> Result<Db> {
+ let path = path.to_owned();
+ tokio::task::spawn_blocking(move || {
+ let mut flags = rusqlite::OpenFlags::SQLITE_OPEN_READ_WRITE;
+ if create {
+ flags |= rusqlite::OpenFlags::SQLITE_OPEN_CREATE;
+ }
+ let conn = Connection::open_with_flags(path, flags).map_err(sql)?;
+ conn.busy_timeout(Duration::from_millis(250)).map_err(sql)?;
+ conn.execute_batch(
+ "PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL; PRAGMA foreign_keys=ON;",
+ )
+ .map_err(sql)?;
+ Ok(Arc::new(Mutex::new(conn)))
+ })
+ .await
+ .map_err(|e| Error::new(ErrorKind::Storage, "open metadata worker", e.to_string()))?
+}
+fn manifest(conn: &Connection) -> Result<(NodeId, NodeId, Revision)> {
+ let (profile, fs, root, revision): (String, Vec<u8>, Vec<u8>, Vec<u8>) = conn
+ .query_row(
+ "SELECT profile, filesystem, root, latest FROM authority WHERE singleton=1",
+ [],
+ |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?)),
+ )
+ .map_err(sql)?;
+ if profile != PROFILE {
+ return Err(Error::unsupported(
+ "open metadata authority",
+ "unknown schema profile",
+ ));
+ }
+ Ok((
+ NodeId::from_bytes(fixed(fs)?),
+ NodeId::from_bytes(fixed(root)?),
+ Revision::from_bytes(fixed(revision)?),
+ ))
+}
+fn fixed<const N: usize>(bytes: Vec<u8>) -> Result<[u8; N]> {
+ bytes.try_into().map_err(|_| {
+ Error::new(
+ ErrorKind::Corrupt,
+ "read metadata database",
+ "invalid identity length",
+ )
+ })
+}
+impl MetadataService {
+ /// Initialize a new local database or reopen its exact existing authority.
+ pub async fn create(path: impl AsRef<Path>, operator: Operator) -> Result<Self> {
+ let path = path.as_ref().to_owned();
+ let db = connect(&path, true).await?;
+ let (filesystem, root, _) = db_call(&db, |conn| {
+ let tx = conn.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate).map_err(sql)?;
+ tx.execute_batch("CREATE TABLE IF NOT EXISTS authority(singleton INTEGER PRIMARY KEY CHECK(singleton=1), profile TEXT NOT NULL, filesystem BLOB NOT NULL, root BLOB NOT NULL, latest BLOB NOT NULL);
+ CREATE TABLE IF NOT EXISTS revisions(token BLOB PRIMARY KEY, sequence INTEGER UNIQUE NOT NULL);
+ CREATE TABLE IF NOT EXISTS records(family INTEGER NOT NULL, key BLOB NOT NULL, start INTEGER NOT NULL, end INTEGER, value BLOB NOT NULL, PRIMARY KEY(family,key,start));
+ CREATE INDEX IF NOT EXISTS records_current ON records(family,key) WHERE end IS NULL;
+ CREATE TABLE IF NOT EXISTS prepared(descriptor BLOB PRIMARY KEY);").map_err(sql)?;
+ if tx.query_row("SELECT count(*) FROM authority", [], |r| r.get::<_,i64>(0)).map_err(sql)? == 0 {
+ let fs = NodeId::generate();
+ let root = NodeId::generate();
+ let revision = Revision::new(0);
+ tx.execute("INSERT INTO authority VALUES(1,?1,?2,?3,?4)",params![PROFILE,fs.as_bytes(),root.as_bytes(),revision.to_bytes().as_slice()]).map_err(sql)?;
+ tx.execute("INSERT INTO revisions VALUES(?1,0)", params![revision.to_bytes().as_slice()]).map_err(sql)?;
+ let node = Node {id:root, generation:1, executable:false, link:None, kind:NodeKind::Directory{membership:1}};
+ tx.execute("INSERT INTO records VALUES(0,?1,0,NULL,?2)", params![root.as_bytes(),node.encode()?]).map_err(sql)?;
+ }
+ let result = manifest(&tx)?;
+ tx.commit().map_err(sql)?;
+ Ok(result)
+ }).await?;
+ let fs = Self {
+ path,
+ data: DataStore::new(operator.clone(), filesystem)?,
+ filesystem,
+ root,
+ };
+ let marker = fs.marker()?;
+ match operator
+ .write_with(".yinyang/head", marker.clone())
+ .if_not_exists(true)
+ .await
+ {
+ Ok(_) => {}
+ Err(_) => {
+ fs.verify_binding(&operator).await?;
+ }
+ }
+ fs.verify_binding(&operator).await?;
+ Ok(fs)
+ }
+ pub async fn open(path: impl AsRef<Path>, operator: Operator) -> Result<Self> {
+ let path = path.as_ref().to_owned();
+ let db = connect(&path, false).await?;
+ let (filesystem, root, _) = db_call(&db, |c| manifest(c)).await?;
+ let fs = Self {
+ path,
+ data: DataStore::new(operator.clone(), filesystem)?,
+ filesystem,
+ root,
+ };
+ fs.verify_binding(&operator).await?;
+ Ok(fs)
+ }
+ fn marker(&self) -> Result<Vec<u8>> {
+ encode(&(
+ *b"YYSERV01",
+ *self.filesystem.as_bytes(),
+ *self.root.as_bytes(),
+ ))
+ }
+ async fn verify_binding(&self, operator: &Operator) -> Result<()> {
+ let marker = operator
+ .read(".yinyang/head")
+ .await
+ .map_err(|e| Error::from_storage("read authority binding", e))?;
+ if marker.to_vec() != self.marker()? {
+ return Err(Error::invalid(
+ "open metadata authority",
+ "storage belongs to another filesystem or publication mode",
+ ));
+ }
+ Ok(())
+ }
+ pub fn data(&self) -> &DataStore {
+ &self.data
+ }
+ pub fn filesystem(&self) -> NodeId {
+ self.filesystem
+ }
+ pub fn root(&self) -> NodeId {
+ self.root
+ }
+ fn snapshot(&self, db: Db, revision: Revision) -> Snapshot {
+ Snapshot {
+ data: self.data.clone(),
+ filesystem: self.filesystem,
+ root: self.root,
+ revision,
+ reader: Arc::new(SqlRecords {
+ db,
+ sequence: revision.sequence,
+ }),
+ }
+ }
+ pub async fn observe_latest(&self) -> Result<Snapshot> {
+ let db = connect(&self.path, false).await?;
+ let (fs, root, revision) = db_call(&db, |c| manifest(c)).await?;
+ if fs != self.filesystem || root != self.root {
+ return Err(Error::invalid("observe service", "authority changed"));
+ }
+ Ok(self.snapshot(db, revision))
+ }
+ pub async fn observe_revision(&self, revision: Revision) -> Result<Snapshot> {
+ if revision.sequence > i64::MAX as u64 {
+ return Err(Error::new(
+ ErrorKind::NotFound,
+ "observe service revision",
+ "revision not retained",
+ ));
+ }
+ let db = connect(&self.path, false).await?;
+ db_call(&db, move |c| {
+ let exists: bool = c
+ .query_row(
+ "SELECT EXISTS(SELECT 1 FROM revisions WHERE token=?1 AND sequence=?2)",
+ params![revision.to_bytes().as_slice(), revision.sequence as i64],
+ |r| r.get(0),
+ )
+ .map_err(sql)?;
+ if !exists {
+ return Err(Error::new(
+ ErrorKind::NotFound,
+ "observe service revision",
+ "revision not retained",
+ ));
+ }
+ Ok(())
+ })
+ .await?;
+ Ok(self.snapshot(db, revision))
+ }
+ /// In-process preparation already carries this service's trusted evidence.
+ pub async fn prepare(&self, source: &mut (impl AsyncRead + Unpin)) -> Result<PreparedContent> {
+ let content = self.data.prepare(source).await?;
+ self.remember(&content).await?;
+ Ok(content)
+ }
+ pub(crate) async fn remember(&self, content: &PreparedContent) -> Result<()> {
+ let descriptor = self.data.accept(content)?.to_bytes();
+ let db = connect(&self.path, false).await?;
+ db_call(&db, move |c| {
+ c.execute(
+ "INSERT OR IGNORE INTO prepared VALUES(?1)",
+ params![descriptor],
+ )
+ .map_err(sql)?;
+ Ok(())
+ })
+ .await
+ }
+ /// Remote descriptors are untrusted: verify once before issuing durable
+ /// readiness. Subsequent commit and restart do not download them again.
+ pub async fn register(&self, descriptor: &ContentDescriptor) -> Result<()> {
+ let content = self.data.import(descriptor).await?;
+ self.remember(&content).await
+ }
+ pub async fn restore_transaction(&self, bytes: &[u8]) -> Result<Transaction> {
+ let (revision, files) = Transaction::inspect(bytes)?;
+ let snapshot = self.observe_revision(revision).await?;
+ let db = connect(&self.path, false).await?;
+ let mut prepared = Vec::new();
+ for file in files {
+ let descriptor = file.to_bytes();
+ let present = db_call(&db, move |c| {
+ c.query_row(
+ "SELECT EXISTS(SELECT 1 FROM prepared WHERE descriptor=?1)",
+ params![descriptor],
+ |r| r.get::<_, bool>(0),
+ )
+ .map_err(sql)
+ })
+ .await?;
+ if !present {
+ return Err(Error::invalid(
+ "restore service plan",
+ "content has no durable preparation evidence",
+ ));
+ }
+ prepared.push(self.data.published(file)?);
+ }
+ Transaction::restore(bytes, &snapshot, &prepared).await
+ }
+ pub async fn commit(&self, request: &Transaction) -> Result<Outcome> {
+ Ok(self
+ .commit_batch(std::slice::from_ref(request))
+ .await?
+ .remove(0))
+ }
+ pub async fn commit_batch(&self, requests: &[Transaction]) -> Result<Vec<Outcome>> {
+ if requests.len() > 4096 {
+ return Err(Error::unsupported(
+ "service batch",
+ "more than 4096 requests",
+ ));
+ }
+ if requests.is_empty() {
+ return Ok(vec![]);
+ }
+ // Even an in-process caller must have registered durable readiness.
+ let mut restored = Vec::new();
+ for request in requests {
+ request.validate(self.filesystem)?;
+ restored.push(self.restore_transaction(&request.to_bytes()?).await?);
+ }
+ let result = self.publish(&restored).await;
+ match result {
+ Err(e) if e.kind() == ErrorKind::Storage => {
+ Ok(requests.iter().map(|_| Outcome::Retryable).collect())
+ }
+ other => other,
+ }
+ }
+ async fn publish(&self, requests: &[Transaction]) -> Result<Vec<Outcome>> {
+ let db = connect(&self.path, false).await?;
+ db_call(&db, |c| {
+ c.execute_batch("BEGIN IMMEDIATE").map_err(sql)?;
+ Ok(())
+ })
+ .await?;
+ let (_, _, old) = db_call(&db, |c| manifest(c)).await?;
+ let sequence = old
+ .sequence
+ .checked_add(1)
+ .filter(|n| *n <= i64::MAX as u64)
+ .ok_or_else(|| Error::unsupported("service commit", "revision exhausted"))?;
+ let revision = Revision::new(sequence);
+ // This view observes preceding accepted members inside this SQL transaction.
+ let candidate = self.snapshot(db.clone(), revision);
+ let mut outcomes = Vec::new();
+ let mut ordinal = 0;
+ for request in requests {
+ if let Some(receipt) = candidate.receipt(request.id()).await? {
+ if receipt.request_digest != request.digest() {
+ return Err(Error::invalid(
+ "service commit",
+ "identity reused for different request",
+ ));
+ }
+ outcomes.push(Outcome::Committed(receipt));
+ continue;
+ }
+ let Some(delta) = request.apply(&candidate).await? else {
+ outcomes.push(Outcome::Conflict);
+ continue;
+ };
+ let receipt = Receipt {
+ commit_id: request.id(),
+ request_digest: request.digest(),
+ cursor: Cursor { revision, ordinal },
+ };
+ let encoded_receipt = receipt.encode()?;
+ let change = ChangeRecord {
+ receipt: receipt.clone(),
+ changes: delta.changes,
+ }
+ .encode()?;
+ let id = *request.id().as_bytes();
+ let key = receipt.cursor.key();
+ db_call(&db, move |c| {
+ for (family, rows) in [
+ (0, delta.nodes),
+ (1, delta.entries),
+ (2, vec![(id.to_vec(), Some(encoded_receipt))]),
+ (3, vec![(key, Some(change))]),
+ ] {
+ for (key, value) in rows {
+ // Multiple batch members may touch a record in this revision.
+ c.execute(
+ "DELETE FROM records WHERE family=?1 AND key=?2 AND start=?3",
+ params![family, key, sequence as i64],
+ )
+ .map_err(sql)?;
+ c.execute(
+ "UPDATE records SET end=?3 WHERE family=?1 AND key=?2 AND end IS NULL",
+ params![family, key, sequence as i64],
+ )
+ .map_err(sql)?;
+ if let Some(value) = value {
+ c.execute(
+ "INSERT INTO records VALUES(?1,?2,?3,NULL,?4)",
+ params![family, key, sequence as i64, value],
+ )
+ .map_err(sql)?;
+ }
+ }
+ }
+ Ok(())
+ })
+ .await?;
+ outcomes.push(Outcome::Committed(receipt));
+ ordinal += 1;
+ }
+ if ordinal == 0 {
+ db_call(&db, |c| {
+ c.execute_batch("ROLLBACK").map_err(sql)?;
+ Ok(())
+ })
+ .await?;
+ return Ok(outcomes);
+ }
+ db_call(&db, move |c| {
+ c.execute(
+ "INSERT INTO revisions VALUES(?1,?2)",
+ params![revision.to_bytes().as_slice(), sequence as i64],
+ )
+ .map_err(sql)?;
+ c.execute(
+ "UPDATE authority SET latest=?1 WHERE singleton=1",
+ params![revision.to_bytes().as_slice()],
+ )
+ .map_err(sql)?;
+ Ok(())
+ })
+ .await?;
+ // A failed durability acknowledgement is not proof of rollback.
+ if db_call(&db, |c| {
+ c.execute_batch("COMMIT").map_err(sql)?;
+ Ok(())
+ })
+ .await
+ .is_err()
+ {
+ return Ok(requests.iter().map(|r| Outcome::Unknown(r.id())).collect());
+ }
+ Ok(outcomes)
+ }
+}
+
+#[derive(Debug)]
+struct SqlRecords {
+ db: Db,
+ sequence: u64,
+}
+impl RecordReader for SqlRecords {
+ fn get<'a>(&'a self, table: Table, key: &'a [u8]) -> BoxFuture<'a, Result<Option<Vec<u8>>>> {
+ Box::pin(async move {
+ let key = key.to_vec();
+ let seq = self.sequence;
+ db_call(&self.db,move |c|c.query_row("SELECT value FROM records WHERE family=?1 AND key=?2 AND start<=?3 AND (end IS NULL OR end>?3) ORDER BY start DESC LIMIT 1",
+ params![table as u8,key,seq as i64],|r|r.get(0)).optional().map_err(sql)).await
+ })
+ }
+ fn scan<'a>(
+ &'a self,
+ table: Table,
+ lower: &'a [u8],
+ upper: Option<&'a [u8]>,
+ after: Option<&'a [u8]>,
+ limit: usize,
+ ) -> BoxFuture<'a, Result<Rows>> {
+ Box::pin(async move {
+ let (lower, upper, after) = (
+ lower.to_vec(),
+ upper.map(<[u8]>::to_vec),
+ after.map(<[u8]>::to_vec),
+ );
+ let seq = self.sequence;
+ db_call(&self.db,move |c| {
+ let mut stmt=c.prepare("SELECT key,value FROM records WHERE family=?1 AND key>=?2 AND (?3 IS NULL OR key<?3) AND (?4 IS NULL OR key>?4) AND start<=?5 AND (end IS NULL OR end>?5) ORDER BY key LIMIT ?6").map_err(sql)?;
+ let rows=stmt.query_map(params![table as u8,lower,upper,after,seq as i64,limit as i64],|r|Ok((r.get(0)?,r.get(1)?))).map_err(sql)?;
+ rows.collect::<std::result::Result<Rows,_>>().map_err(sql)
+ }).await
+ })
+ }
+}
diff --git a/crates/core/src/service/transport.rs b/crates/core/src/service/transport.rs
new file mode 100644
index 0000000..deb8731
--- /dev/null
+++ b/crates/core/src/service/transport.rs
@@ -0,0 +1,473 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Versioned, bounded RPC for a trusted local service deployment.
+//! The initial transport is loopback-only; remote exposure requires an
+//! authenticated encrypted transport and is deliberately rejected here.
+
+use super::*;
+use borsh::{BorshDeserialize, BorshSerialize};
+use std::net::SocketAddr;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::{TcpListener, TcpStream};
+
+const MAX_FRAME: usize = 16 * 1024 * 1024;
+const TIMEOUT: Duration = Duration::from_secs(60);
+
+#[derive(BorshSerialize, BorshDeserialize)]
+struct Request {
+ version: u8,
+ token: String,
+ command: Command,
+}
+#[derive(BorshSerialize, BorshDeserialize)]
+enum Command {
+ Observe(Option<[u8; 24]>),
+ Get([u8; 24], u8, Vec<u8>),
+ Scan([u8; 24], u8, Vec<u8>, Option<Vec<u8>>, Option<Vec<u8>>, u32),
+ Register(Vec<u8>),
+ Commit(Vec<Vec<u8>>),
+}
+#[derive(BorshSerialize, BorshDeserialize)]
+enum Response {
+ Observation([u8; 16], [u8; 16], [u8; 24]),
+ Value(Option<Vec<u8>>),
+ Rows(Rows),
+ Registered,
+ Outcomes(Vec<(u8, Vec<u8>)>),
+ Error(u8, String),
+}
+fn io(e: std::io::Error) -> Error {
+ Error::from_io("metadata RPC", e)
+}
+fn protocol() -> Error {
+ Error::invalid("metadata RPC", "unexpected protocol response")
+}
+fn kind_tag(kind: ErrorKind) -> u8 {
+ match kind {
+ ErrorKind::Invalid => 0,
+ ErrorKind::Corrupt => 1,
+ ErrorKind::NotFound => 2,
+ ErrorKind::AlreadyExists => 3,
+ ErrorKind::Unsupported => 4,
+ ErrorKind::Storage => 5,
+ ErrorKind::Io => 6,
+ }
+}
+fn tag_kind(tag: u8) -> Result<ErrorKind> {
+ Ok(match tag {
+ 0 => ErrorKind::Invalid,
+ 1 => ErrorKind::Corrupt,
+ 2 => ErrorKind::NotFound,
+ 3 => ErrorKind::AlreadyExists,
+ 4 => ErrorKind::Unsupported,
+ 5 => ErrorKind::Storage,
+ 6 => ErrorKind::Io,
+ _ => return Err(protocol()),
+ })
+}
+async fn read_frame<T: BorshDeserialize>(stream: &mut TcpStream) -> Result<T> {
+ let length = stream.read_u32().await.map_err(io)? as usize;
+ if length > MAX_FRAME {
+ return Err(Error::unsupported("metadata RPC", "frame exceeds 16 MiB"));
+ }
+ let mut bytes = vec![0; length];
+ stream.read_exact(&mut bytes).await.map_err(io)?;
+ decode(&bytes)
+}
+async fn write_frame(stream: &mut TcpStream, value: &impl BorshSerialize) -> Result<()> {
+ let bytes = encode(value)?;
+ if bytes.len() > MAX_FRAME {
+ return Err(Error::unsupported("metadata RPC", "frame exceeds 16 MiB"));
+ }
+ stream.write_u32(bytes.len() as u32).await.map_err(io)?;
+ stream.write_all(&bytes).await.map_err(io)?;
+ stream.flush().await.map_err(io)
+}
+fn validate_endpoint(addr: SocketAddr, token: &str) -> Result<()> {
+ if !addr.ip().is_loopback() || token.len() < 32 || token.len() > 1024 {
+ return Err(Error::unsupported(
+ "metadata RPC",
+ "requires loopback and a 32..=1024 byte authentication token",
+ ));
+ }
+ Ok(())
+}
+fn table(tag: u8) -> Result<Table> {
+ Ok(match tag {
+ 0 => Table::Nodes,
+ 1 => Table::Entries,
+ 2 => Table::Receipts,
+ 3 => Table::Changes,
+ _ => return Err(protocol()),
+ })
+}
+
+/// Serve authenticated local clients. No object checkpoint can accept writes
+/// while this authority is unavailable. Stop the listener to stop accepting;
+/// in-flight requests may still complete, so callers resolve their commit IDs.
+pub async fn serve(listener: TcpListener, service: MetadataService, token: String) -> Result<()> {
+ validate_endpoint(listener.local_addr().map_err(io)?, &token)?;
+ let limit = Arc::new(tokio::sync::Semaphore::new(32));
+ loop {
+ let permit = limit
+ .clone()
+ .acquire_owned()
+ .await
+ .map_err(|_| protocol())?;
+ let (mut stream, _) = listener.accept().await.map_err(io)?;
+ let service = service.clone();
+ let token = token.clone();
+ tokio::spawn(async move {
+ let _permit = permit;
+ // A timeout drops the connection, never claims to roll back a commit.
+ let _ = tokio::time::timeout(TIMEOUT, async {
+ let response = match read_frame::<Request>(&mut stream).await {
+ Ok(request)
+ if request.version == 1
+ && blake3::hash(request.token.as_bytes())
+ == blake3::hash(token.as_bytes()) =>
+ {
+ dispatch(&service, request.command).await
+ }
+ _ => Err(Error::invalid(
+ "metadata RPC",
+ "invalid protocol or authentication",
+ )),
+ };
+ let response = response.unwrap_or_else(|e| {
+ Response::Error(kind_tag(e.kind()), e.message().to_owned())
+ });
+ write_frame(&mut stream, &response).await
+ })
+ .await;
+ });
+ }
+}
+async fn dispatch(service: &MetadataService, command: Command) -> Result<Response> {
+ Ok(match command {
+ Command::Observe(revision) => {
+ let snapshot = match revision {
+ Some(r) => service.observe_revision(Revision::from_bytes(r)).await?,
+ None => service.observe_latest().await?,
+ };
+ Response::Observation(
+ *snapshot.filesystem().as_bytes(),
+ *snapshot.root().as_bytes(),
+ snapshot.revision().to_bytes(),
+ )
+ }
+ Command::Get(revision, tag, key) => {
+ if key.len() > 1024 {
+ return Err(protocol());
+ }
+ let snapshot = service
+ .observe_revision(Revision::from_bytes(revision))
+ .await?;
+ Response::Value(snapshot.reader.get(table(tag)?, &key).await?)
+ }
+ Command::Scan(revision, tag, lower, upper, after, limit) => {
+ if limit == 0
+ || limit > 4097
+ || lower.len() > 1024
+ || upper.as_ref().is_some_and(|x| x.len() > 1024)
+ || after.as_ref().is_some_and(|x| x.len() > 1024)
+ {
+ return Err(protocol());
+ }
+ let snapshot = service
+ .observe_revision(Revision::from_bytes(revision))
+ .await?;
+ Response::Rows(
+ snapshot
+ .reader
+ .scan(
+ table(tag)?,
+ &lower,
+ upper.as_deref(),
+ after.as_deref(),
+ limit as usize,
+ )
+ .await?,
+ )
+ }
+ Command::Register(bytes) => {
+ service
+ .register(&ContentDescriptor::from_bytes(&bytes)?)
+ .await?;
+ Response::Registered
+ }
+ Command::Commit(plans) => {
+ if plans.len() > 4096 {
+ return Err(protocol());
+ }
+ let mut requests = Vec::new();
+ for plan in plans {
+ requests.push(service.restore_transaction(&plan).await?);
+ }
+ Response::Outcomes(
+ service
+ .commit_batch(&requests)
+ .await?
+ .into_iter()
+ .map(|outcome| {
+ Ok(match outcome {
+ Outcome::Committed(receipt) => (0, receipt.encode()?),
+ Outcome::Conflict => (1, vec![]),
+ Outcome::Retryable => (2, vec![]),
+ Outcome::Unknown(id) => (3, id.as_bytes().to_vec()),
+ })
+ })
+ .collect::<Result<_>>()?,
+ )
+ }
+ })
+}
+
+/// Authenticated local client with direct OpenDAL content transfer.
+#[derive(Clone)]
+pub struct ServiceClient {
+ addr: SocketAddr,
+ token: Arc<str>,
+ data: DataStore,
+ filesystem: NodeId,
+ root: NodeId,
+}
+impl std::fmt::Debug for ServiceClient {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("ServiceClient")
+ .field("addr", &self.addr)
+ .field("filesystem", &self.filesystem)
+ .finish_non_exhaustive()
+ }
+}
+async fn rpc(addr: SocketAddr, token: &str, command: Command) -> Result<Response> {
+ tokio::time::timeout(TIMEOUT, async {
+ let mut stream = TcpStream::connect(addr).await.map_err(io)?;
+ write_frame(
+ &mut stream,
+ &Request {
+ version: 1,
+ token: token.to_owned(),
+ command,
+ },
+ )
+ .await?;
+ match read_frame(&mut stream).await? {
+ Response::Error(tag, message) => {
+ Err(Error::new(tag_kind(tag)?, "metadata service", message))
+ }
+ response => Ok(response),
+ }
+ })
+ .await
+ .map_err(|_| Error::new(ErrorKind::Io, "metadata RPC", "response timed out"))?
+}
+impl ServiceClient {
+ pub async fn connect(addr: SocketAddr, token: String, operator: Operator) -> Result<Self> {
+ validate_endpoint(addr, &token)?;
+ let Response::Observation(fs, root, _) = rpc(addr, &token, Command::Observe(None)).await?
+ else {
+ return Err(protocol());
+ };
+ let marker = operator
+ .read(".yinyang/head")
+ .await
+ .map_err(|e| Error::from_storage("verify service data binding", e))?;
+ if marker.to_vec() != encode(&(*b"YYSERV01", fs, root))? {
+ return Err(Error::invalid(
+ "connect service",
+ "data binding disagrees with service",
+ ));
+ }
+ Ok(Self {
+ addr,
+ token: Arc::from(token),
+ data: DataStore::new(operator, NodeId::from_bytes(fs))?,
+ filesystem: NodeId::from_bytes(fs),
+ root: NodeId::from_bytes(root),
+ })
+ }
+ pub fn data(&self) -> &DataStore {
+ &self.data
+ }
+ pub fn filesystem(&self) -> NodeId {
+ self.filesystem
+ }
+ pub fn root(&self) -> NodeId {
+ self.root
+ }
+ async fn observe(&self, revision: Option<Revision>) -> Result<Snapshot> {
+ let Response::Observation(fs, root, token) = rpc(
+ self.addr,
+ &self.token,
+ Command::Observe(revision.map(Revision::to_bytes)),
+ )
+ .await?
+ else {
+ return Err(protocol());
+ };
+ if fs != *self.filesystem.as_bytes()
+ || root != *self.root.as_bytes()
+ || revision.is_some_and(|r| r.to_bytes() != token)
+ {
+ return Err(protocol());
+ }
+ let revision = Revision::from_bytes(token);
+ Ok(Snapshot {
+ data: self.data.clone(),
+ filesystem: self.filesystem,
+ root: self.root,
+ revision,
+ reader: Arc::new(RemoteRecords {
+ client: self.clone(),
+ revision,
+ }),
+ })
+ }
+ pub async fn observe_latest(&self) -> Result<Snapshot> {
+ self.observe(None).await
+ }
+ pub async fn observe_revision(&self, revision: Revision) -> Result<Snapshot> {
+ self.observe(Some(revision)).await
+ }
+ pub async fn prepare(&self, source: &mut (impl AsyncRead + Unpin)) -> Result<PreparedContent> {
+ let content = self.data.prepare(source).await?;
+ self.register(&content).await?;
+ Ok(content)
+ }
+ pub async fn register(&self, content: &PreparedContent) -> Result<()> {
+ let descriptor = self.data.accept(content)?;
+ match rpc(
+ self.addr,
+ &self.token,
+ Command::Register(descriptor.to_bytes()),
+ )
+ .await?
+ {
+ Response::Registered => Ok(()),
+ _ => Err(protocol()),
+ }
+ }
+ pub async fn restore_transaction(&self, bytes: &[u8]) -> Result<Transaction> {
+ let (revision, files) = Transaction::inspect(bytes)?;
+ let snapshot = self.observe_revision(revision).await?;
+ let mut prepared = Vec::new();
+ for file in files {
+ prepared.push(self.data.import(&file).await?);
+ }
+ Transaction::restore(bytes, &snapshot, &prepared).await
+ }
+ pub async fn commit(&self, request: &Transaction) -> Result<Outcome> {
+ Ok(self
+ .commit_batch(std::slice::from_ref(request))
+ .await?
+ .remove(0))
+ }
+ pub async fn commit_batch(&self, requests: &[Transaction]) -> Result<Vec<Outcome>> {
+ for request in requests {
+ request.validate(self.filesystem)?;
+ }
+ let plans = requests
+ .iter()
+ .map(Transaction::to_bytes)
+ .collect::<Result<Vec<_>>>()?;
+ let response = match rpc(self.addr, &self.token, Command::Commit(plans)).await {
+ Ok(response) => response,
+ Err(e) if matches!(e.kind(), ErrorKind::Io | ErrorKind::Storage) => {
+ return Ok(requests.iter().map(|r| Outcome::Unknown(r.id())).collect());
+ }
+ Err(e) => return Err(e),
+ };
+ let Response::Outcomes(outcomes) = response else {
+ return Err(protocol());
+ };
+ if outcomes.len() != requests.len() {
+ return Err(protocol());
+ }
+ outcomes
+ .into_iter()
+ .zip(requests)
+ .map(|((tag, bytes), request)| {
+ Ok(match tag {
+ 0 => {
+ let receipt = Receipt::decode(&bytes)?;
+ if receipt.commit_id != request.id()
+ || receipt.request_digest != request.digest()
+ {
+ return Err(protocol());
+ }
+ Outcome::Committed(receipt)
+ }
+ 1 => Outcome::Conflict,
+ 2 => Outcome::Retryable,
+ 3 if bytes == request.id().as_bytes() => Outcome::Unknown(request.id()),
+ _ => return Err(protocol()),
+ })
+ })
+ .collect()
+ }
+}
+#[derive(Debug)]
+struct RemoteRecords {
+ client: ServiceClient,
+ revision: Revision,
+}
+impl RecordReader for RemoteRecords {
+ fn get<'a>(&'a self, table: Table, key: &'a [u8]) -> BoxFuture<'a, Result<Option<Vec<u8>>>> {
+ Box::pin(async move {
+ match rpc(
+ self.client.addr,
+ &self.client.token,
+ Command::Get(self.revision.to_bytes(), table as u8, key.to_vec()),
+ )
+ .await?
+ {
+ Response::Value(v) => Ok(v),
+ _ => Err(protocol()),
+ }
+ })
+ }
+ fn scan<'a>(
+ &'a self,
+ table: Table,
+ lower: &'a [u8],
+ upper: Option<&'a [u8]>,
+ after: Option<&'a [u8]>,
+ limit: usize,
+ ) -> BoxFuture<'a, Result<Rows>> {
+ Box::pin(async move {
+ match rpc(
+ self.client.addr,
+ &self.client.token,
+ Command::Scan(
+ self.revision.to_bytes(),
+ table as u8,
+ lower.to_vec(),
+ upper.map(<[u8]>::to_vec),
+ after.map(<[u8]>::to_vec),
+ u32::try_from(limit).map_err(|_| protocol())?,
+ ),
+ )
+ .await?
+ {
+ Response::Rows(v) => Ok(v),
+ _ => Err(protocol()),
+ }
+ })
+ }
+}
diff --git a/crates/core/src/transaction.rs b/crates/core/src/transaction.rs
index d334866..59c8f23 100644
--- a/crates/core/src/transaction.rs
+++ b/crates/core/src/transaction.rs
@@ -102,6 +102,7 @@ impl Condition {
#[derive(Clone, Debug)]
pub struct Transaction {
filesystem: NodeId,
+ base: crate::Revision,
id: CommitId,
digest: [u8; 32],
conditions: Vec<Condition>,
@@ -155,6 +156,234 @@ impl Transaction {
}
}
+type RequestWire = (
+ [u8; 8],
+ [u8; 16],
+ [u8; 16],
+ [u8; 24],
+ [u8; 32],
+ Vec<(Vec<u8>, Option<Vec<u8>>)>,
+ Vec<Vec<u8>>,
+);
+
+impl Transaction {
+ /// Persist the exact plan, not permission to trust its content descriptors.
+ /// Restore through the selected authority before submitting after a restart.
+ pub fn to_bytes(&self) -> Result<Vec<u8>> {
+ let mutations = self
+ .mutations
+ .iter()
+ .map(|m| {
+ Ok(match m {
+ Mutation::Create {
+ id,
+ parent,
+ name,
+ file,
+ executable,
+ } => encode(&(
+ 0_u8,
+ *id.as_bytes(),
+ *parent.as_bytes(),
+ name,
+ file.as_ref().map(|f| f.descriptor().to_bytes()),
+ executable,
+ ))?,
+ Mutation::Content(id, file) => {
+ encode(&(1_u8, *id.as_bytes(), file.descriptor().to_bytes()))?
+ }
+ Mutation::Executable(id, value) => encode(&(2_u8, *id.as_bytes(), value))?,
+ Mutation::Rename(id, parent, name) => {
+ encode(&(3_u8, *id.as_bytes(), *parent.as_bytes(), name))?
+ }
+ Mutation::Remove(id) => encode(&(4_u8, *id.as_bytes()))?,
+ })
+ })
+ .collect::<Result<Vec<_>>>()?;
+ encode(&(
+ *b"YYPLAN01",
+ *self.filesystem.as_bytes(),
+ *self.id.as_bytes(),
+ self.base.to_bytes(),
+ self.digest,
+ self.conditions
+ .iter()
+ .map(|c| (&c.key, &c.expected))
+ .collect::<Vec<_>>(),
+ mutations,
+ ))
+ }
+
+ pub(crate) fn inspect(
+ bytes: &[u8],
+ ) -> Result<(crate::Revision, Vec<crate::ContentDescriptor>)> {
+ let (_, _, _, revision, _, _, mutations) = Self::wire(bytes)?;
+ let mut files = Vec::new();
+ for bytes in mutations {
+ match bytes.first() {
+ Some(0) => {
+ let (_, _, _, _, file, _): (
+ u8,
+ [u8; 16],
+ [u8; 16],
+ String,
+ Option<Vec<u8>>,
+ bool,
+ ) = crate::namespace::decode(&bytes)?;
+ if let Some(file) = file {
+ files.push(crate::ContentDescriptor::from_bytes(&file)?);
+ }
+ }
+ Some(1) => {
+ let (_, _, file): (u8, [u8; 16], Vec<u8>) = crate::namespace::decode(&bytes)?;
+ files.push(crate::ContentDescriptor::from_bytes(&file)?);
+ }
+ _ => {}
+ }
+ }
+ Ok((crate::Revision::from_bytes(revision), files))
+ }
+ fn wire(bytes: &[u8]) -> Result<RequestWire> {
+ if bytes.len() > 16 * 1024 * 1024 {
+ return Err(Error::unsupported(
+ "restore transaction",
+ "plan exceeds 16 MiB",
+ ));
+ }
+ let wire: RequestWire = crate::namespace::decode(bytes)?;
+ if wire.0 != *b"YYPLAN01" {
+ return Err(Error::invalid(
+ "restore transaction",
+ "unknown plan profile",
+ ));
+ }
+ Ok(wire)
+ }
+ pub(crate) async fn restore(
+ bytes: &[u8],
+ snapshot: &Snapshot,
+ prepared: &[PreparedContent],
+ ) -> Result<Self> {
+ let (_, fs, id, revision, digest, conditions, encoded) = Self::wire(bytes)?;
+ if fs != *snapshot.filesystem.as_bytes() || revision != snapshot.revision().to_bytes() {
+ return Err(Error::invalid(
+ "restore transaction",
+ "wrong authority or base revision",
+ ));
+ }
+ let mut map = BTreeMap::new();
+ for (key, expected) in conditions {
+ if key.is_empty()
+ || key[0] > ENTRY
+ || (key[0] != ENTRY && key.len() != 17)
+ || (key[0] == ENTRY && key.len() < 18)
+ || map.contains_key(&key)
+ {
+ return Err(Error::invalid(
+ "restore transaction",
+ "invalid or duplicate predicate",
+ ));
+ }
+ let condition = Condition {
+ key: key.clone(),
+ expected,
+ };
+ if !condition.matches(snapshot).await? {
+ return Err(Error::invalid(
+ "restore transaction",
+ "predicate disagrees with original observation",
+ ));
+ }
+ map.insert(key, condition);
+ }
+ let find = |bytes: Vec<u8>| -> Result<PreparedContent> {
+ prepared
+ .iter()
+ .find(|p| p.descriptor().to_bytes() == bytes)
+ .cloned()
+ .ok_or_else(|| {
+ Error::invalid("restore transaction", "content readiness not established")
+ })
+ };
+ let mut work = Working::new(snapshot.clone(), true);
+ let mut mutations = Vec::new();
+ for bytes in encoded {
+ let mutation = match bytes.first() {
+ Some(0) => {
+ let (_, node, parent, name, file, executable): (
+ u8,
+ [u8; 16],
+ [u8; 16],
+ String,
+ Option<Vec<u8>>,
+ bool,
+ ) = crate::namespace::decode(&bytes)?;
+ let seed = encode(&(fs, id, mutations.len() as u64))?;
+ let derived =
+ blake3::derive_key("Apache OpenDAL YinYang node identity profile 2", &seed);
+ if node != derived[..16] {
+ return Err(Error::invalid(
+ "restore transaction",
+ "invalid creation identity",
+ ));
+ }
+ Mutation::Create {
+ id: NodeId::from_bytes(node),
+ parent: NodeId::from_bytes(parent),
+ name,
+ file: file.map(&find).transpose()?,
+ executable,
+ }
+ }
+ Some(1) => {
+ let (_, node, file): (u8, [u8; 16], Vec<u8>) =
+ crate::namespace::decode(&bytes)?;
+ Mutation::Content(NodeId::from_bytes(node), find(file)?)
+ }
+ Some(2) => {
+ let (_, node, value): (u8, [u8; 16], bool) = crate::namespace::decode(&bytes)?;
+ Mutation::Executable(NodeId::from_bytes(node), value)
+ }
+ Some(3) => {
+ let (_, node, parent, name): (u8, [u8; 16], [u8; 16], String) =
+ crate::namespace::decode(&bytes)?;
+ Mutation::Rename(NodeId::from_bytes(node), NodeId::from_bytes(parent), name)
+ }
+ Some(4) => {
+ let (_, node): (u8, [u8; 16]) = crate::namespace::decode(&bytes)?;
+ Mutation::Remove(NodeId::from_bytes(node))
+ }
+ _ => return Err(Error::invalid("restore transaction", "unknown mutation")),
+ };
+ work.apply(&mutation).await?;
+ mutations.push(mutation);
+ }
+ // Lower-level clients may add dependencies, but cannot omit the guards
+ // required by the same planner used by trusted in-process callers.
+ for required in work.conditions.values() {
+ if map
+ .get(&required.key)
+ .is_none_or(|c| c.expected != required.expected)
+ {
+ return Err(Error::invalid(
+ "restore transaction",
+ "missing mandatory predicate",
+ ));
+ }
+ }
+ let request = Self {
+ filesystem: snapshot.filesystem,
+ base: snapshot.revision(),
+ id: CommitId::from_bytes(id),
+ digest,
+ conditions: map.into_values().collect(),
+ mutations,
+ };
+ request.validate(snapshot.filesystem)?;
+ Ok(request)
+ }
+}
+
/// Dependency-capturing operation planner; it never runs caller callbacks on retry.
pub struct Planner {
id: CommitId,
@@ -255,6 +484,7 @@ impl Planner {
pub fn finish(self) -> Result<Transaction> {
let mut request = Transaction {
filesystem: self.work.base.filesystem,
+ base: self.work.base.revision(),
id: self.id,
digest: [0; 32],
conditions: self.work.conditions.into_values().collect(),
@@ -623,6 +853,7 @@ mod tests {
fn canonical_noop_request_vector() {
let request = Transaction {
filesystem: NodeId::from_bytes([1; 16]),
+ base: crate::Revision::new(0),
id: CommitId::from_bytes([2; 16]),
digest: [0; 32],
conditions: vec![],
diff --git a/crates/core/tests/service.rs b/crates/core/tests/service.rs
new file mode 100644
index 0000000..c8e2bc8
--- /dev/null
+++ b/crates/core/tests/service.rs
@@ -0,0 +1,422 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+mod support;
+use std::sync::Arc;
+use support::TestBackend;
+use yinyang_core::object::{BackendProfile, ObjectFs};
+use yinyang_core::service::{MetadataService, ServiceClient, serve};
+use yinyang_core::{Authority, CommitId, CommitOutcome as Outcome, ErrorKind, Planner};
+
+fn committed(outcome: Outcome) -> yinyang_core::Receipt {
+ match outcome {
+ Outcome::Committed(r) => r,
+ other => panic!("expected committed: {other:?}"),
+ }
+}
+async fn suite(fs: &dyn Authority) {
+ let first = fs.observe_latest().await.unwrap();
+ let file = fs.prepare(&mut b"original".as_slice()).await.unwrap();
+ let mut create = Planner::new(&first, CommitId::generate());
+ let a = create
+ .create_file(fs.root(), "a", file.clone(), false)
+ .await
+ .unwrap();
+ let b = create
+ .create_file(fs.root(), "b", file.clone(), false)
+ .await
+ .unwrap();
+ let p = create.create_directory(fs.root(), "p").await.unwrap();
+ let q = create.create_directory(fs.root(), "q").await.unwrap();
+ let transaction = create.finish().unwrap();
+ let receipt = committed(fs.commit(&transaction).await.unwrap());
+ assert_eq!(committed(fs.commit(&transaction).await.unwrap()), receipt);
+ assert!(first.resolve("a").await.unwrap().is_none());
+ let base = fs.observe_latest().await.unwrap();
+ let serialized = transaction.to_bytes().unwrap();
+ let restored = fs.restore_transaction(&serialized).await.unwrap();
+ assert_eq!(committed(fs.commit(&restored).await.unwrap()), receipt);
+
+ // Independent same-parent name slots and distinct file states do not conflict.
+ let mut left = Planner::new(&base, CommitId::generate());
+ left.create_directory(p, "left").await.unwrap();
+ left.set_executable(a, true).await.unwrap();
+ let mut right = Planner::new(&base, CommitId::generate());
+ right.create_directory(p, "right").await.unwrap();
+ right.set_executable(b, true).await.unwrap();
+ let (left, right) = (left.finish().unwrap(), right.finish().unwrap());
+ let (l, r) = tokio::join!(fs.commit(&left), fs.commit(&right));
+ committed(l.unwrap());
+ committed(r.unwrap());
+ let current = fs.observe_latest().await.unwrap();
+ assert!(current.node(a).await.unwrap().unwrap().executable());
+ assert!(current.node(b).await.unwrap().unwrap().executable());
+ assert!(!base.node(a).await.unwrap().unwrap().executable());
+
+ // Exact same observed file has at most one successful replacement.
+ let newer = fs.prepare(&mut b"new bytes".as_slice()).await.unwrap();
+ let mut l = Planner::new(¤t, CommitId::generate());
+ let mut r = Planner::new(¤t, CommitId::generate());
+ l.set_content(a, newer.clone()).await.unwrap();
+ r.set_content(a, file.clone()).await.unwrap();
+ committed(fs.commit(&l.finish().unwrap()).await.unwrap());
+ assert_eq!(
+ fs.commit(&r.finish().unwrap()).await.unwrap(),
+ Outcome::Conflict
+ );
+
+ // Empty removal is protected against insertions; scans include phantoms.
+ let current = fs.observe_latest().await.unwrap();
+ let mut remove = Planner::new(¤t, CommitId::generate());
+ remove.remove(q).await.unwrap();
+ let mut scan = Planner::new(¤t, CommitId::generate());
+ scan.scan(q).await.unwrap();
+ let scan = scan.finish().unwrap();
+ let mut add = Planner::new(¤t, CommitId::generate());
+ add.create_directory(q, "child").await.unwrap();
+ committed(fs.commit(&add.finish().unwrap()).await.unwrap());
+ assert_eq!(
+ fs.commit(&remove.finish().unwrap()).await.unwrap(),
+ Outcome::Conflict
+ );
+ assert_eq!(fs.commit(&scan).await.unwrap(), Outcome::Conflict);
+
+ // Competing ancestry changes cannot introduce cycles.
+ let current = fs.observe_latest().await.unwrap();
+ let mut l = Planner::new(¤t, CommitId::generate());
+ let mut r = Planner::new(¤t, CommitId::generate());
+ l.rename(p, q, "p").await.unwrap();
+ r.rename(q, p, "q").await.unwrap();
+ committed(fs.commit(&l.finish().unwrap()).await.unwrap());
+ assert_eq!(
+ fs.commit(&r.finish().unwrap()).await.unwrap(),
+ Outcome::Conflict
+ );
+
+ // Batched results share a revision but retain transaction order.
+ let current = fs.observe_latest().await.unwrap();
+ let mut l = Planner::new(¤t, CommitId::generate());
+ let mut r = Planner::new(¤t, CommitId::generate());
+ l.set_content(a, file).await.unwrap();
+ r.set_content(b, newer).await.unwrap();
+ let batch = fs
+ .commit_batch(&[l.finish().unwrap(), r.finish().unwrap()])
+ .await
+ .unwrap();
+ let l = committed(batch[0].clone());
+ let r = committed(batch[1].clone());
+ assert_eq!(l.cursor.revision, r.cursor.revision);
+ assert!(l.cursor.ordinal < r.cursor.ordinal);
+ let pinned = fs.observe_revision(receipt.cursor.revision).await.unwrap();
+ let page = pinned.scan(fs.root(), None, 1).await.unwrap();
+ assert!(page.next.is_some());
+ let latest = fs.observe_latest().await.unwrap();
+ assert_eq!(
+ latest
+ .scan(fs.root(), page.next.as_ref(), 1)
+ .await
+ .unwrap_err()
+ .kind(),
+ ErrorKind::Invalid
+ );
+ assert_eq!(
+ pinned.receipt(transaction.id()).await.unwrap(),
+ Some(receipt)
+ );
+ assert!(latest.changes(None, 4096).await.unwrap().len() >= 6);
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn shared_contract_object_and_metadata_service() {
+ let object_backend = TestBackend::default();
+ let object = ObjectFs::create(object_backend.operator(), BackendProfile::Minio)
+ .await
+ .unwrap();
+ suite(&object).await;
+ let dir = tempfile::tempdir().unwrap();
+ let backend = TestBackend::default();
+ let service = MetadataService::create(dir.path().join("metadata.db"), backend.operator())
+ .await
+ .unwrap();
+ suite(&service).await;
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn authenticated_service_protocol_and_restart() {
+ let dir = tempfile::tempdir().unwrap();
+ let path = dir.path().join("metadata.db");
+ let backend = TestBackend::default();
+ let service = MetadataService::create(&path, backend.operator())
+ .await
+ .unwrap();
+ assert_eq!(
+ ObjectFs::open(backend.operator(), BackendProfile::Minio)
+ .await
+ .unwrap_err()
+ .kind(),
+ ErrorKind::Unsupported
+ );
+ let token = "test-only-service-token-32-bytes-long";
+ let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+ let addr = listener.local_addr().unwrap();
+ let server = tokio::spawn(serve(listener, service.clone(), token.into()));
+ assert_eq!(
+ ServiceClient::connect(
+ addr,
+ "wrong-token-but-at-least-32-bytes-long".into(),
+ backend.operator()
+ )
+ .await
+ .unwrap_err()
+ .kind(),
+ ErrorKind::Invalid
+ );
+ let client = ServiceClient::connect(addr, token.into(), backend.operator())
+ .await
+ .unwrap();
+ assert!(!format!("{client:?}").contains(token));
+ suite(&client).await;
+ let snapshot = client.observe_latest().await.unwrap();
+ let content = client.prepare(&mut b"persisted".as_slice()).await.unwrap();
+ let mut plan = Planner::new(&snapshot, CommitId::generate());
+ plan.create_file(client.root(), "persisted", content, false)
+ .await
+ .unwrap();
+ let plan = plan.finish().unwrap();
+ let receipt = committed(client.commit(&plan).await.unwrap());
+ server.abort();
+ let _ = server.await;
+ drop(service);
+ let reopened = MetadataService::open(&path, backend.operator())
+ .await
+ .unwrap();
+ let restored = reopened
+ .restore_transaction(&plan.to_bytes().unwrap())
+ .await
+ .unwrap();
+ // Even with object-head writes failing, service publication and lookup work.
+ backend.state.lock().unwrap().fail_after_head_write = true;
+ assert_eq!(
+ committed(reopened.commit(&restored).await.unwrap()),
+ receipt
+ );
+ assert!(
+ reopened
+ .observe_revision(snapshot.revision())
+ .await
+ .unwrap()
+ .resolve("persisted")
+ .await
+ .unwrap()
+ .is_none()
+ );
+ let mut plan = Planner::new(
+ &reopened.observe_latest().await.unwrap(),
+ CommitId::generate(),
+ );
+ plan.create_directory(reopened.root(), "after-restart")
+ .await
+ .unwrap();
+ committed(reopened.commit(&plan.finish().unwrap()).await.unwrap());
+ assert!(backend.state.lock().unwrap().fail_after_head_write);
+}
+
+#[tokio::test]
+async fn service_rejects_unregistered_content_and_missing_guards() {
+ let dir = tempfile::tempdir().unwrap();
+ let backend = TestBackend::default();
+ let service = Arc::new(
+ MetadataService::create(dir.path().join("metadata.db"), backend.operator())
+ .await
+ .unwrap(),
+ );
+ let snapshot = service.observe_latest().await.unwrap();
+ let content = service
+ .data()
+ .prepare(&mut b"unregistered".as_slice())
+ .await
+ .unwrap();
+ let mut plan = Planner::new(&snapshot, CommitId::generate());
+ plan.create_file(service.root(), "file", content.clone(), false)
+ .await
+ .unwrap();
+ let plan = plan.finish().unwrap();
+ assert_eq!(
+ service.commit(&plan).await.unwrap_err().kind(),
+ ErrorKind::Invalid
+ );
+ service.register(content.descriptor()).await.unwrap();
+ let reads = backend.state.lock().unwrap().data_reads;
+ let bytes = plan.to_bytes().unwrap();
+ type Wire = (
+ [u8; 8],
+ [u8; 16],
+ [u8; 16],
+ [u8; 24],
+ [u8; 32],
+ Vec<(Vec<u8>, Option<Vec<u8>>)>,
+ Vec<Vec<u8>>,
+ );
+ let mut wire: Wire = borsh::from_slice(&bytes).unwrap();
+ wire.5.clear();
+ let error = service
+ .restore_transaction(&borsh::to_vec(&wire).unwrap())
+ .await
+ .unwrap_err();
+ assert!(
+ error.message().contains("missing mandatory predicate"),
+ "{error}"
+ );
+ committed(service.commit(&plan).await.unwrap());
+ assert_eq!(backend.state.lock().unwrap().data_reads, reads);
+}
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn lost_and_delayed_rpc_responses_resolve_the_original_request() {
+ use std::sync::atomic::{AtomicU8, Ordering};
+ use tokio::io::{AsyncReadExt, AsyncWriteExt};
+ use tokio::net::{TcpListener, TcpStream};
+ let dir = tempfile::tempdir().unwrap();
+ let backend = TestBackend::default();
+ let service = MetadataService::create(dir.path().join("db"), backend.operator())
+ .await
+ .unwrap();
+ let token = "test-only-service-token-32-bytes-long";
+ let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+ let upstream = listener.local_addr().unwrap();
+ let server = tokio::spawn(serve(listener, service.clone(), token.into()));
+ let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+ let addr = listener.local_addr().unwrap();
+ let mode = Arc::new(AtomicU8::new(0));
+ let reached = Arc::new(tokio::sync::Notify::new());
+ let release = Arc::new(tokio::sync::Notify::new());
+ let proxy = {
+ let (mode, reached, release) = (mode.clone(), reached.clone(), release.clone());
+ tokio::spawn(async move {
+ loop {
+ let (mut downstream, _) = listener.accept().await.unwrap();
+ let (mode, reached, release) = (mode.clone(), reached.clone(), release.clone());
+ tokio::spawn(async move {
+ let mut upstream = TcpStream::connect(upstream).await.unwrap();
+ let length = downstream.read_u32().await.unwrap();
+ let mut request = vec![0; length as usize];
+ downstream.read_exact(&mut request).await.unwrap();
+ upstream.write_u32(length).await.unwrap();
+ upstream.write_all(&request).await.unwrap();
+ let length = upstream.read_u32().await.unwrap();
+ let mut response = vec![0; length as usize];
+ upstream.read_exact(&mut response).await.unwrap();
+ match mode.swap(0, Ordering::SeqCst) {
+ 1 => return, // Commit is durable, but acknowledgement is lost.
+ 2 => {
+ reached.notify_one();
+ release.notified().await;
+ }
+ _ => {}
+ }
+ let _ = downstream.write_u32(length).await;
+ let _ = downstream.write_all(&response).await;
+ });
+ }
+ })
+ };
+ let client = ServiceClient::connect(addr, token.into(), backend.operator())
+ .await
+ .unwrap();
+ let first = service.observe_latest().await.unwrap();
+ let mut plan = Planner::new(&first, CommitId::generate());
+ plan.create_directory(service.root(), "lost").await.unwrap();
+ let plan = plan.finish().unwrap();
+ mode.store(1, Ordering::SeqCst);
+ assert_eq!(
+ client.commit(&plan).await.unwrap(),
+ Outcome::Unknown(plan.id())
+ );
+ let receipt = service
+ .observe_latest()
+ .await
+ .unwrap()
+ .receipt(plan.id())
+ .await
+ .unwrap()
+ .unwrap();
+ assert_eq!(committed(client.commit(&plan).await.unwrap()), receipt);
+
+ let mut delayed = Planner::new(
+ &service.observe_latest().await.unwrap(),
+ CommitId::generate(),
+ );
+ delayed
+ .create_directory(service.root(), "delayed")
+ .await
+ .unwrap();
+ let delayed = delayed.finish().unwrap();
+ let id = delayed.id();
+ mode.store(2, Ordering::SeqCst);
+ let pending = tokio::spawn(async move { client.commit(&delayed).await });
+ reached.notified().await;
+ let intermediate = service.observe_latest().await.unwrap();
+ let old = intermediate.receipt(id).await.unwrap().unwrap();
+ let mut next = Planner::new(&intermediate, CommitId::generate());
+ next.create_directory(service.root(), "later")
+ .await
+ .unwrap();
+ committed(service.commit(&next.finish().unwrap()).await.unwrap());
+ release.notify_one();
+ assert_eq!(committed(pending.await.unwrap().unwrap()), old);
+ assert!(first.resolve("lost").await.unwrap().is_none());
+ proxy.abort();
+ server.abort();
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn independent_authorities_busy_retry_and_identity_misuse() {
+ let dir = tempfile::tempdir().unwrap();
+ let backend = TestBackend::default();
+ let path = dir.path().join("db");
+ let service = MetadataService::create(&path, backend.operator())
+ .await
+ .unwrap();
+ let second = MetadataService::open(&path, backend.operator())
+ .await
+ .unwrap();
+ let base = service.observe_latest().await.unwrap();
+ let mut plan = Planner::new(&base, CommitId::generate());
+ plan.create_directory(service.root(), "winner")
+ .await
+ .unwrap();
+ let plan = plan.finish().unwrap();
+ let lock = rusqlite::Connection::open(&path).unwrap();
+ lock.execute_batch("BEGIN IMMEDIATE").unwrap();
+ assert_eq!(second.commit(&plan).await.unwrap(), Outcome::Retryable);
+ assert!(base.receipt(plan.id()).await.unwrap().is_none());
+ lock.execute_batch("ROLLBACK").unwrap();
+ let receipt = committed(second.commit(&plan).await.unwrap());
+ assert_eq!(committed(service.commit(&plan).await.unwrap()), receipt);
+ let mut misuse = Planner::new(&base, plan.id());
+ misuse
+ .create_directory(service.root(), "different")
+ .await
+ .unwrap();
+ assert_eq!(
+ service
+ .commit(&misuse.finish().unwrap())
+ .await
+ .unwrap_err()
+ .kind(),
+ ErrorKind::Invalid
+ );
+}
diff --git a/specs/README.md b/specs/README.md
index a6e01e5..f281b15 100644
--- a/specs/README.md
+++ b/specs/README.md
@@ -7,7 +7,8 @@ This directory contains the maintained contracts for Apache OpenDAL™ YinYang.
| [YinYang object profile](yinyang-format.md) | Experimental indexed transactions and OpenDAL head publication. |
| [Directory transfer](directory-transfer.md) | One-shot local directory publication and restoration. |
| [Authenticated content profile](content-profile.md) | Experimental range verification and trusted preparation; independent of publication authority. |
-| [Transactional filesystem](transactional-filesystem.md) | Shared target contract. Object-storage binding implemented experimentally; metadata-service binding remains open. |
+| [Metadata service](metadata-service.md) | Experimental durable SQLite authority and authenticated loopback protocol. |
+| [Transactional filesystem](transactional-filesystem.md) | Shared target contract. Object-storage and metadata-service bindings implemented experimentally. |
A specification describes the currently supported behavior, APIs, wire formats,
invariants, compatibility rules, and implementation boundaries. Specifications
diff --git a/specs/metadata-service.md b/specs/metadata-service.md
new file mode 100644
index 0000000..26aaa92
--- /dev/null
+++ b/specs/metadata-service.md
@@ -0,0 +1,111 @@
+# Metadata service
+
+**Status: experimental single-host authority and authenticated loopback RPC.**
+This binding implements the shared [transactional filesystem contract](transactional-filesystem.md)
+using SQLite metadata and the existing OpenDAL content profile. It does not
+provide high availability, multi-host consensus, or a remotely exposed transport.
+
+## Authority and storage
+
+`MetadataService::create/open` binds one local SQLite database to one content
+prefix. The database uses WAL, `synchronous=FULL`, and `BEGIN IMMEDIATE` for
+publication. Its filesystem must honor SQLite locking and synchronization;
+network-mounted database files are unsupported. Keep the database and its WAL
+together; copying only the live main database is not a backup procedure.
+
+The schema profile is `yinyang-sqlite-1`. The authority row retains filesystem
+and root identities plus the latest revision. Indexed valid-time records retain
+nodes, entries, receipts, and ordered changes. Each record has an inclusive
+start and exclusive end sequence. Revision tokens are retained separately and
+validated before reads. History is application data, not an expiring database
+MVCC window. There is no history deletion or garbage collection.
+
+Writers serialize at the database boundary, but revalidate original logical
+predicates through the same transaction evaluator as object publication.
+Disjoint name slots and file states can both succeed from one observation;
+membership scans, file state, emptiness, and ancestor predicates detect real
+conflicts. Every accepted batch member's namespace delta, receipt, and changes
+commit atomically. Members share a revision and have ordered cursor ordinals.
+No publication rewrites a complete object snapshot.
+
+An immutable `.yinyang/head` marker contains Borsh
+`([u8;8] = YYSERV01, filesystem: [u8;16], root: [u8;16])`.
+Creation conditionally installs it; opening validates it. Object-mode readers
+reject this marker. Service commits neither replace it nor consult it as a
+latest checkpoint. A service outage cannot authorize object-mode writes.
+The database is the sole authority and is required for history and receipts.
+
+## Shared API and content readiness
+
+`Authority` supplies common observations, preparation, batch publication, and
+frozen-request restoration. `Snapshot`, `Planner`, `Transaction`, receipts,
+changes, and revision/scan tokens are shared with object mode.
+
+In-process preparation carries trusted `DataStore` evidence and records
+readiness durably. A client uploads through OpenDAL, then registers its content
+descriptor. The service fully verifies an untrusted descriptor before recording
+readiness. Publication and restart use that persisted evidence; they do not
+download the file again. Credentials and prefix isolation must prevent external
+writers from mutating immutable published content, as in object mode.
+
+`Transaction::to_bytes` freezes the original observation, identity, predicates,
+and mutations. The experimental Borsh envelope is
+`(YYPLAN01, filesystem[16], commit_id[16], base_revision[24], digest[32], conditions, mutations)`.
+Conditions are key/optional-value pairs. Mutation encodings reuse the logical
+canonical transaction tags and record encodings. Physical prepared descriptors
+are carried for file content; they do not enter the canonical logical digest.
+Restoration validates the original retained snapshot, derived node identities,
+all mandatory predicates, and digest. A caller cannot remove a condition to
+bypass concurrency checks. An untrusted serialized request is not preparation
+evidence. Object-mode restoration verifies referenced content through import.
+
+## Protocol and security boundary
+
+`yy serve --database PATH --listen 127.0.0.1:7447` opens or creates the authority.
+Storage uses the existing `YINYANG_S3_*` settings. Set
+`YINYANG_SERVICE_TOKEN` to a random secret with 32–1024 bytes; do not place it in
+command arguments. `ServiceClient::connect` checks the content-prefix binding.
+
+The initial transport accepts loopback addresses only, including IPv6 loopback.
+It is authenticated but not encrypted. Use it only among trusted local
+processes; it has one full-authority token, no per-user authorization, no TLS,
+and no supported public/network exposure. Tokens are omitted from client Debug.
+A future remote deployment requires a separately specified secure transport.
+
+Each TCP connection carries one request/response. Frames are a big-endian u32
+length followed by Borsh, bounded to 16 MiB. Request fields are protocol version
+1, token string, and a tagged command. Commands observe latest/specific
+revision, read/scan records at a revision, register content, or commit frozen
+requests. Responses carry observations, records, readiness, typed outcomes, or
+typed errors. Maximum batch size is 4096; scans request at most 4097 records.
+The server handles at most 32 active connections, each with a 60-second timeout.
+This wire profile is experimental and is not a stable interoperability promise.
+
+## Failure and recovery
+
+Database contention before commit returns `Retryable`; replay the original
+request, not a newly observed/replanned one. Failed commit acknowledgement and
+lost/timeout transport responses return `Unknown(commit_id)`. A timeout does
+not cancel or prove rollback of an in-flight durable transaction. Resolve via
+a retained receipt or retry the exact frozen request. Absent receipt is not
+proof that a delayed request cannot still commit.
+
+Duplicate identity/digest returns the original receipt. Identity reused with a
+different digest is invalid. Reopening the same database retains receipts,
+readiness, changes, and historical revisions. No automatic failover or
+checkpoint writer exists. Listener shutdown stops accepting new work; already
+accepted requests may finish.
+
+## Verification and operating limits
+
+Shared integration tests run the same predicate and snapshot suite against
+object publication, embedded service, and authenticated RPC. Fault injection
+covers dropped and delayed acknowledgements, independent database connections,
+content-readiness rejection, omitted predicates, and restart. The S3 daemon
+test kills and restarts the real process against durable MinIO content.
+
+SQLite indexes provide incremental record access, but historical query cost
+depends on retained versions and query ranges; this profile does not promise
+the object B+ tree's page bounds. SQLite's single writer is a throughput limit.
+There is no benchmark-based production capacity claim or stable storage
+migration support.
diff --git a/specs/transactional-filesystem.md b/specs/transactional-filesystem.md
index d5bf68b..4ff3ec6 100644
--- a/specs/transactional-filesystem.md
+++ b/specs/transactional-filesystem.md
@@ -1,9 +1,9 @@
# Transactional Filesystem
-**Status: shared target contract; object-storage binding implemented experimentally.**
-The [current object profile](yinyang-format.md) and [content profile](content-profile.md)
-describe implemented APIs and encodings. Metadata-service publication remains
-unimplemented. The persistent profile is not frozen for interoperable release;
+**Status: shared target contract; both publication bindings implemented experimentally.**
+The [object profile](yinyang-format.md), [metadata service](metadata-service.md),
+and [content profile](content-profile.md) describe implemented APIs and encodings.
+The persistent profiles are not frozen for interoperable release;
[Open decisions](#open-decisions) identifies the remaining evaluation work.
## Scope and cost model
diff --git a/src/main.rs b/src/main.rs
index aa2aaac..28bb574 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -31,6 +31,13 @@ struct Cli {
}
#[derive(Subcommand)]
enum Command {
+ /// Run the authenticated loopback metadata authority with a local SQLite database.
+ Serve {
+ #[arg(long)]
+ database: PathBuf,
+ #[arg(long, default_value = "127.0.0.1:7447")]
+ listen: std::net::SocketAddr,
+ },
/// Create a filesystem, or validate an existing one.
Create,
/// Publish a complete directory, removing remote-only paths.
@@ -75,6 +82,19 @@ async fn run(cli: Cli) -> Result<(), Box<dyn std::error::Error>> {
})
.collect::<Vec<_>>();
let operator = opendal::Operator::via_iter("s3", config)?;
+ if let Command::Serve { database, listen } = &cli.command {
+ let token = std::env::var("YINYANG_SERVICE_TOKEN").map_err(
+ |_| "set YINYANG_SERVICE_TOKEN to a random authentication token of at least 32 bytes",
+ )?;
+ if !listen.ip().is_loopback() || !(32..=1024).contains(&token.len()) {
+ return Err("metadata service requires loopback and a 32..=1024 byte token".into());
+ }
+ let service = yinyang::core::service::MetadataService::create(database, operator).await?;
+ let listener = tokio::net::TcpListener::bind(listen).await?;
+ println!("metadata service listening on {}", listener.local_addr()?);
+ yinyang::core::service::serve(listener, service, token).await?;
+ return Ok(());
+ }
if matches!(cli.command, Command::Create) {
let fs = Fs::create(operator, profile).await?;
println!(
@@ -86,7 +106,7 @@ async fn run(cli: Cli) -> Result<(), Box<dyn std::error::Error>> {
let fs = Fs::open(operator, profile).await?;
let observed = fs.observe_latest().await?;
match cli.command {
- Command::Create => unreachable!(),
+ Command::Create | Command::Serve { .. } => unreachable!(),
Command::Publish {
source,
replace,
diff --git a/tests/service.rs b/tests/service.rs
new file mode 100644
index 0000000..0e891b1
--- /dev/null
+++ b/tests/service.rs
@@ -0,0 +1,131 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::io::{BufRead, BufReader};
+use std::process::{Child, Command, Stdio};
+use yinyang::core::service::ServiceClient;
+use yinyang::core::{CommitId, CommitOutcome, Planner};
+
+struct Daemon(Child);
+impl Drop for Daemon {
+ fn drop(&mut self) {
+ let _ = self.0.kill();
+ let _ = self.0.wait();
+ }
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[ignore = "requires an isolated S3 bucket configured with YINYANG_S3_* variables"]
+async fn s3_service_process_restart_retains_original_request_and_history() {
+ opendal::install_default();
+ let temp = tempfile::tempdir().unwrap();
+ let root = format!("service-{}", uuid::Uuid::new_v4().simple());
+ let token = "test-only-service-token-at-least-32-bytes";
+ let config = std::env::vars()
+ .filter_map(|(k, v)| {
+ k.strip_prefix("YINYANG_S3_")
+ .map(|k| (k.to_ascii_lowercase(), v))
+ })
+ .filter(|(k, _)| k != "root")
+ .chain([("root".into(), root.clone())])
+ .collect::<Vec<_>>();
+ let operator = opendal::Operator::via_iter("s3", config).unwrap();
+ let start = || {
+ let mut child = Daemon(
+ Command::new(env!("CARGO_BIN_EXE_yy"))
+ .args(["serve", "--database"])
+ .arg(temp.path().join("metadata.db"))
+ .args(["--listen", "127.0.0.1:0"])
+ .env("YINYANG_S3_ROOT", &root)
+ .env("YINYANG_STORAGE_PROFILE", "minio")
+ .env("YINYANG_SERVICE_TOKEN", token)
+ .stdout(Stdio::piped())
+ .spawn()
+ .unwrap(),
+ );
+ let stdout = child.0.stdout.take().unwrap();
+ (child, stdout)
+ };
+ let connect = |stdout| async {
+ let addr = tokio::time::timeout(
+ std::time::Duration::from_secs(30),
+ tokio::task::spawn_blocking(move || {
+ let mut line = String::new();
+ BufReader::new(stdout).read_line(&mut line).unwrap();
+ line.trim()
+ .strip_prefix("metadata service listening on ")
+ .unwrap()
+ .parse()
+ .unwrap()
+ }),
+ )
+ .await
+ .unwrap()
+ .unwrap();
+ ServiceClient::connect(addr, token.into(), operator.clone())
+ .await
+ .unwrap()
+ };
+ let (daemon, stdout) = start();
+ let client = connect(stdout).await;
+ let first = client.observe_latest().await.unwrap();
+ let content = client
+ .prepare(&mut b"durable service bytes".as_slice())
+ .await
+ .unwrap();
+ let mut plan = Planner::new(&first, CommitId::generate());
+ plan.create_file(client.root(), "file", content, false)
+ .await
+ .unwrap();
+ let plan = plan.finish().unwrap();
+ let CommitOutcome::Committed(receipt) = client.commit(&plan).await.unwrap() else {
+ panic!("not committed")
+ };
+ let frozen = plan.to_bytes().unwrap();
+ drop(daemon); // Kill, do not rely on a graceful checkpoint or destructor.
+ let (daemon, stdout) = start();
+ let client = connect(stdout).await;
+ assert!(
+ client
+ .observe_revision(first.revision())
+ .await
+ .unwrap()
+ .resolve("file")
+ .await
+ .unwrap()
+ .is_none()
+ );
+ let restored = client.restore_transaction(&frozen).await.unwrap();
+ assert_eq!(
+ client.commit(&restored).await.unwrap(),
+ CommitOutcome::Committed(receipt)
+ );
+ let snapshot = client.observe_latest().await.unwrap();
+ let node = snapshot.resolve("file").await.unwrap().unwrap();
+ let mut bytes = Vec::new();
+ client
+ .data()
+ .read_range(
+ snapshot.content(node.id()).await.unwrap().descriptor(),
+ 0..21,
+ &mut bytes,
+ )
+ .await
+ .unwrap();
+ assert_eq!(bytes, b"durable service bytes");
+ drop(daemon);
+}