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/07 10:05:22 UTC
(opendal-yinyang) branch main updated: feat(core): persist YinYang through OpenDAL (#43)
This is an automated email from the ASF dual-hosted git repository.
Xuanwo pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/opendal-yinyang.git
The following commit(s) were added to refs/heads/main by this push:
new 1f00d5d feat(core): persist YinYang through OpenDAL (#43)
1f00d5d is described below
commit 1f00d5db416d9799ee21e796bd0b29989148b880
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Mon Sep 7 18:05:17 2026 +0800
feat(core): persist YinYang through OpenDAL (#43)
---
Cargo.lock | 761 +++++++++++++++++++++++++++++++++++++++
Cargo.toml | 4 +
crates/core/Cargo.toml | 7 +-
crates/core/src/error.rs | 9 +
crates/core/src/lib.rs | 5 +-
crates/core/src/persistence.rs | 783 +++++++++++++++++++++++++++++++++++++++++
crates/core/src/publication.rs | 64 ++--
crates/core/src/storage.rs | 72 ----
crates/core/tests/core.rs | 494 +++++++++++++++++++++-----
specs/yinyang-format.md | 103 ++++--
10 files changed, 2099 insertions(+), 203 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 666aee4..21bd158 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -52,18 +52,106 @@ dependencies = [
"windows-sys",
]
+[[package]]
+name = "anyhow"
+version = "1.0.104"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470"
+
+[[package]]
+name = "arrayvec"
+version = "0.7.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d3fb67a6e08acf24fdeccbac2cb6ac4305825bd1f117462e0e6f2f193345ad56"
+
+[[package]]
+name = "asyncband"
+version = "0.6.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "94a214ba60d6231afd0e805e3c27c45a1626d9debaa5a5061c45a1ea1b2f1ed0"
+dependencies = [
+ "hashbrown",
+ "slab",
+]
+
+[[package]]
+name = "base64"
+version = "0.23.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ac07cdecf99051d9a5238b80f35af32cdeba5b336e55d957b318b50137e18da5"
+
+[[package]]
+name = "bitflags"
+version = "1.3.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a"
+
+[[package]]
+name = "blake3"
+version = "1.8.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6d9e454fc11f76977dc803893aff6304ed33d6a26efae8696573bea74baa27ae"
+dependencies = [
+ "arrayvec",
+ "cc",
+ "cfg-if",
+ "constant_time_eq",
+ "cpufeatures",
+]
+
+[[package]]
+name = "block-buffer"
+version = "0.12.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d2f6c7dbe95a6ed67ad9f18e57daf93a2f034c524b99fd2b76d18fdfeb6660aa"
+dependencies = [
+ "hybrid-array",
+]
+
+[[package]]
+name = "borsh"
+version = "1.8.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "553c5d846a6ba5150c65e3b1b8ec073bcf1abc20f9b7220de384a4443ea4e20a"
+dependencies = [
+ "bytes",
+ "cfg_aliases",
+]
+
[[package]]
name = "bumpalo"
version = "3.20.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649"
+[[package]]
+name = "bytes"
+version = "1.12.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04"
+
+[[package]]
+name = "cc"
+version = "1.4.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0ad534f4357a5264cce5019c989cf66a4f0dc4e0d1b1d15f8aacec0ff7360273"
+dependencies = [
+ "find-msvc-tools",
+ "shlex",
+]
+
[[package]]
name = "cfg-if"
version = "1.0.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801"
+[[package]]
+name = "cfg_aliases"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f079e83a288787bcd14a6aea84cee5c87a67c5a3e660c30f557a3d24761b3527"
+
[[package]]
name = "clap"
version = "4.6.6"
@@ -110,12 +198,157 @@ version = "1.0.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570"
+[[package]]
+name = "const-oid"
+version = "0.10.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a6ef517f0926dd24a1582492c791b6a4818a4d94e789a334894aa15b0d12f55c"
+
+[[package]]
+name = "constant_time_eq"
+version = "0.4.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3d52eff69cd5e647efe296129160853a42795992097e8af39800e1060caeea9b"
+
+[[package]]
+name = "cpufeatures"
+version = "0.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5ca28b0ae3115b884660db4118d803791fd6756b6e88f39c0f3f7859060d7566"
+dependencies = [
+ "libc",
+]
+
+[[package]]
+name = "crypto-common"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ce6e4c961d6cd6c9a86db418387425e8bdeaf05b3c8bc1411e6dca4c252f1453"
+dependencies = [
+ "hybrid-array",
+]
+
+[[package]]
+name = "defmt"
+version = "1.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e2953bfe4f93bbd20cc71198842756f77d161884c99ebbabc41d80231ded88d1"
+dependencies = [
+ "bitflags",
+ "defmt-macros",
+]
+
+[[package]]
+name = "defmt-macros"
+version = "1.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bad9c72e7ca2137e0dc3813245a0d282fd6daad32fd800af018306a9169b5fe8"
+dependencies = [
+ "defmt-parser",
+ "proc-macro2",
+ "quote",
+ "syn 2.0.119",
+]
+
+[[package]]
+name = "defmt-parser"
+version = "1.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "10d60334b3b2e7c9d91ef8150abfb6fa4c1c39ebbcf4a81c2e346aad939fee3e"
+dependencies = [
+ "thiserror",
+]
+
+[[package]]
+name = "digest"
+version = "0.11.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f1dd6dbb5841937940781866fa1281a1ff7bd3bf827091440879f9994983d5c2"
+dependencies = [
+ "block-buffer",
+ "const-oid",
+ "crypto-common",
+]
+
+[[package]]
+name = "displaydoc"
+version = "0.2.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c6232dd377dcc64799954cbd3a9bb882e9cdc1308ccd87b1c098f1fb2eaf82a8"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 3.0.3",
+]
+
+[[package]]
+name = "find-msvc-tools"
+version = "0.1.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d45db016d36b838f563236e9193d0ee6ce38f3f68b6c94e914b4929c96bbb890"
+
+[[package]]
+name = "form_urlencoded"
+version = "1.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cb4cb245038516f5f85277875cdaa4f7d2c9a0fa0468de06ed190163b1581fcf"
+dependencies = [
+ "percent-encoding",
+]
+
+[[package]]
+name = "futures"
+version = "0.3.34"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9a31d2a3fbaaeb2af2368bbdd904aa8e812d3c04a1ee10d3171f52d556e5d0a3"
+dependencies = [
+ "futures-channel",
+ "futures-core",
+ "futures-io",
+ "futures-sink",
+ "futures-task",
+ "futures-util",
+]
+
+[[package]]
+name = "futures-channel"
+version = "0.3.34"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b1f9e3d69d39e4862ffed03ed071a76f9a13ba1d9109d355b0f0aa6b15e393c4"
+dependencies = [
+ "futures-core",
+ "futures-sink",
+]
+
[[package]]
name = "futures-core"
version = "0.3.34"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e"
+[[package]]
+name = "futures-io"
+version = "0.3.34"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "53c0fa8157de1303bfffdaa1cc2a673bfffb60102f76b0ef4441659124373fed"
+
+[[package]]
+name = "futures-macro"
+version = "0.3.34"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9fb9654ba8355388abeb8dcb4fc62f511300867002afc858860463bdd9fe0c44"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 3.0.3",
+]
+
+[[package]]
+name = "futures-sink"
+version = "0.3.34"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1944426bf7d03f1d14f708785e4b33efd750b36d48a157b836b3efc15ede8e1d"
+
[[package]]
name = "futures-task"
version = "0.3.34"
@@ -128,8 +361,13 @@ version = "0.3.34"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0d50a92467f8ba5dd6e3ee5d4bd04d73ab2e4e1c44474a0674821dfce14b79bc"
dependencies = [
+ "futures-channel",
"futures-core",
+ "futures-io",
+ "futures-macro",
+ "futures-sink",
"futures-task",
+ "memchr",
"pin-project-lite",
"slab",
]
@@ -145,18 +383,208 @@ dependencies = [
"r-efi",
]
+[[package]]
+name = "hashbrown"
+version = "0.17.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a"
+
[[package]]
name = "heck"
version = "0.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea"
+[[package]]
+name = "http"
+version = "1.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "918d3568bebf352712bc2ef3d46a8bcf1a75b373be6539de198e9105cbbf9ce0"
+dependencies = [
+ "bytes",
+ "itoa",
+]
+
+[[package]]
+name = "hybrid-array"
+version = "0.4.14"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "707114b52a152fa7bdb290cd7cd5912d9467273b6d74e21b8d81aca1f8533f6b"
+dependencies = [
+ "typenum",
+]
+
+[[package]]
+name = "icu_collections"
+version = "2.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fa68d21081c4a05d5a901a1c62add574c77048b6a1c67be3b50ce0b60d4ca513"
+dependencies = [
+ "displaydoc",
+ "potential_utf",
+ "utf8_iter",
+ "yoke",
+ "zerofrom",
+ "zerovec",
+]
+
+[[package]]
+name = "icu_locale_core"
+version = "2.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d56e28588da92eee5c3201a6eff33fabdd49b62269c8938d4ff050ce4d900deb"
+dependencies = [
+ "displaydoc",
+ "litemap",
+ "tinystr",
+ "writeable",
+ "zerovec",
+]
+
+[[package]]
+name = "icu_normalizer"
+version = "2.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "12f9cf5f235641ed274641dd81c3f28d870e276763d0797aeeab72317b1c646f"
+dependencies = [
+ "icu_collections",
+ "icu_normalizer_data",
+ "icu_properties",
+ "icu_provider",
+ "smallvec",
+ "zerovec",
+]
+
+[[package]]
+name = "icu_normalizer_data"
+version = "2.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1563da1ed3e0b3bf3d74c9b85917ac9c56464d2f57242270c09c9e752f8021a0"
+
+[[package]]
+name = "icu_properties"
+version = "2.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7e7ca276ad3145661a65914e6daf131ca5120cd3dcee8f8f3214b8875184a148"
+dependencies = [
+ "displaydoc",
+ "icu_collections",
+ "icu_locale_core",
+ "icu_properties_data",
+ "icu_provider",
+ "zerotrie",
+ "zerovec",
+]
+
+[[package]]
+name = "icu_properties_data"
+version = "2.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e590f038c1464a96894fd6d10127e90a8be4509f56ff7ecef851b15cee0b7caa"
+
+[[package]]
+name = "icu_provider"
+version = "2.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d27bbb9d3abbefac45d55f647c9de1d44aafcd1186eb91879afef17c396c3e73"
+dependencies = [
+ "displaydoc",
+ "icu_locale_core",
+ "writeable",
+ "yoke",
+ "zerofrom",
+ "zerotrie",
+ "zerovec",
+]
+
+[[package]]
+name = "idna"
+version = "1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3b0875f23caa03898994f6ddc501886a45c7d3d62d04d2d90788d47be1b1e4de"
+dependencies = [
+ "idna_adapter",
+ "smallvec",
+ "utf8_iter",
+]
+
+[[package]]
+name = "idna_adapter"
+version = "1.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cb68373c0d6620ef8105e855e7745e18b0d00d3bdb07fb532e434244cdb9a714"
+dependencies = [
+ "icu_normalizer",
+ "icu_properties",
+]
+
[[package]]
name = "is_terminal_polyfill"
version = "1.70.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695"
+[[package]]
+name = "itoa"
+version = "1.0.18"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682"
+
+[[package]]
+name = "jiff"
+version = "0.2.35"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "668b7183bd07af9a4885f5c35b0cc5c83c4607a913c16b7e17291832910d2dcc"
+dependencies = [
+ "defmt",
+ "jiff-core",
+ "jiff-static",
+ "jiff-tzdb-platform",
+ "js-sys",
+ "log",
+ "portable-atomic",
+ "portable-atomic-util",
+ "serde_core",
+ "wasm-bindgen",
+ "windows-link",
+]
+
+[[package]]
+name = "jiff-core"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7feca88439efe53da3754500c1851dedf3cb36c524dd5cf8225cc0794de95d09"
+dependencies = [
+ "defmt",
+]
+
+[[package]]
+name = "jiff-static"
+version = "0.2.35"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3a69dcb3a21cfb32ce1cd056169337ca284af0766dd766e7878819b251a49204"
+dependencies = [
+ "jiff-core",
+ "proc-macro2",
+ "quote",
+ "syn 2.0.119",
+]
+
+[[package]]
+name = "jiff-tzdb"
+version = "0.1.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "142bd39932ad231f10513df9ab62661fead8719872150b7ad02a2df79f4e141e"
+
+[[package]]
+name = "jiff-tzdb-platform"
+version = "0.1.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "875a5a69ac2bab1a891711cf5eccbec1ce0341ea805560dcd90b7a2e925132e8"
+dependencies = [
+ "jiff-tzdb",
+]
+
[[package]]
name = "js-sys"
version = "0.3.104"
@@ -174,6 +602,34 @@ version = "0.2.189"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2"
+[[package]]
+name = "litemap"
+version = "0.8.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "47d9d19d1d6efa0109d2f65ff4c85cddd50bd572e5a00127ab10987290bcefae"
+
+[[package]]
+name = "log"
+version = "0.4.34"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6"
+
+[[package]]
+name = "md-5"
+version = "0.11.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "69b6441f590336821bb897fb28fc622898ccceb1d6cea3fde5ea86b090c4de98"
+dependencies = [
+ "cfg-if",
+ "digest",
+]
+
+[[package]]
+name = "memchr"
+version = "2.8.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98"
+
[[package]]
name = "once_cell"
version = "1.21.4"
@@ -186,12 +642,76 @@ version = "1.70.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe"
+[[package]]
+name = "opendal"
+version = "0.58.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "33dbff14cc9bb085224256d6a81289d2f3202e85b06f408d42534b42162a4231"
+dependencies = [
+ "opendal-core",
+]
+
+[[package]]
+name = "opendal-core"
+version = "0.58.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "48dbcef97d3eb7591db2c18d5cae95c836bcce07359b98d98dd6f4e861eb77b7"
+dependencies = [
+ "anyhow",
+ "asyncband",
+ "base64",
+ "bytes",
+ "futures",
+ "http",
+ "jiff",
+ "log",
+ "md-5",
+ "percent-encoding",
+ "quick-xml",
+ "serde",
+ "serde_json",
+ "tokio",
+ "url",
+ "uuid",
+ "web-time",
+]
+
+[[package]]
+name = "percent-encoding"
+version = "2.3.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220"
+
[[package]]
name = "pin-project-lite"
version = "0.2.17"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd"
+[[package]]
+name = "portable-atomic"
+version = "1.15.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "05c8b63e8d9609db387f0324918f81d68fe27748f084ef092fb35954d0539a85"
+
+[[package]]
+name = "portable-atomic-util"
+version = "0.2.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c2a106d1259c23fac8e543272398ae0e3c0b8d33c88ed73d0cc71b0f1d902618"
+dependencies = [
+ "portable-atomic",
+]
+
+[[package]]
+name = "potential_utf"
+version = "0.1.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d83eb9bc6d8e5cf568e7a1101d60ee05e81ed50ea106026f3d18deeb046d7661"
+dependencies = [
+ "zerovec",
+]
+
[[package]]
name = "proc-macro2"
version = "1.0.107"
@@ -201,6 +721,16 @@ dependencies = [
"unicode-ident",
]
+[[package]]
+name = "quick-xml"
+version = "0.41.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e660451e55124f798a69a5af3f49ccfbefbd41910eefd25caf2393e1f3473ec1"
+dependencies = [
+ "memchr",
+ "serde",
+]
+
[[package]]
name = "quote"
version = "1.0.47"
@@ -222,12 +752,73 @@ version = "1.0.23"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cf54715a573b99ac80df0bc206da022bcd442c974952c7b9720069370852e21f"
+[[package]]
+name = "serde"
+version = "1.0.229"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba"
+dependencies = [
+ "serde_core",
+ "serde_derive",
+]
+
+[[package]]
+name = "serde_core"
+version = "1.0.229"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "67dca2c9c51e58a4791a4b1ed58308b39c64224d349a935ab5039aa360942a48"
+dependencies = [
+ "serde_derive",
+]
+
+[[package]]
+name = "serde_derive"
+version = "1.0.229"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 3.0.3",
+]
+
+[[package]]
+name = "serde_json"
+version = "1.0.151"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c841b55ecdae098c80dcae9cf767f6f8a0c2cdb3416bbef72181df4d0fe73f14"
+dependencies = [
+ "itoa",
+ "memchr",
+ "serde",
+ "serde_core",
+ "zmij",
+]
+
+[[package]]
+name = "shlex"
+version = "2.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba"
+
[[package]]
name = "slab"
version = "0.4.12"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5"
+[[package]]
+name = "smallvec"
+version = "1.16.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b9be42f50aa861c555654aa3a37f52f4b1074bacf4e48fe0ef7fa584e80f1f0f"
+
+[[package]]
+name = "stable_deref_trait"
+version = "1.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596"
+
[[package]]
name = "strsim"
version = "0.11.1"
@@ -256,6 +847,47 @@ dependencies = [
"unicode-ident",
]
+[[package]]
+name = "synstructure"
+version = "0.13.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "728a70f3dbaf5bab7f0c4b1ac8d7ae5ea60a4b5549c8a5914361c99147a709d2"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.119",
+]
+
+[[package]]
+name = "thiserror"
+version = "2.0.20"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ec86235f5fcc2a73650310756d2ac5b138a5780bbbdfae3eeccec992c435ba4f"
+dependencies = [
+ "thiserror-impl",
+]
+
+[[package]]
+name = "thiserror-impl"
+version = "2.0.20"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bc04cd3e1236dd4a98afca4569f2deb3f120e5422a4023be2cb683f8486292af"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 3.0.3",
+]
+
+[[package]]
+name = "tinystr"
+version = "0.8.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b1e27c91459209c2986af3dcf603a5a74a4368754ce37414f59acc971167f643"
+dependencies = [
+ "displaydoc",
+ "zerovec",
+]
+
[[package]]
name = "tinyvec"
version = "1.12.0"
@@ -277,6 +909,7 @@ version = "1.53.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed"
dependencies = [
+ "bytes",
"pin-project-lite",
"tokio-macros",
]
@@ -292,6 +925,12 @@ dependencies = [
"syn 3.0.3",
]
+[[package]]
+name = "typenum"
+version = "1.20.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20"
+
[[package]]
name = "unicode-casefold"
version = "0.2.0"
@@ -313,6 +952,24 @@ dependencies = [
"tinyvec",
]
+[[package]]
+name = "url"
+version = "2.5.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ff67a8a4397373c3ef660812acab3268222035010ab8680ec4215f38ba3d0eed"
+dependencies = [
+ "form_urlencoded",
+ "idna",
+ "percent-encoding",
+ "serde",
+]
+
+[[package]]
+name = "utf8_iter"
+version = "1.0.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be"
+
[[package]]
name = "utf8parse"
version = "0.2.2"
@@ -327,6 +984,7 @@ checksum = "b5772d71c9be8a8a6ac2117d949c5b224c1b72241bb611d9a3012edcf8af7812"
dependencies = [
"getrandom",
"js-sys",
+ "serde_core",
"wasm-bindgen",
]
@@ -375,6 +1033,16 @@ dependencies = [
"unicode-ident",
]
+[[package]]
+name = "web-time"
+version = "1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5a6580f308b1fad9207618087a65c04e7a10bc77e02c8e84e9b00dd4b12fa0bb"
+dependencies = [
+ "js-sys",
+ "wasm-bindgen",
+]
+
[[package]]
name = "which"
version = "8.0.5"
@@ -399,6 +1067,12 @@ dependencies = [
"windows-link",
]
+[[package]]
+name = "writeable"
+version = "0.6.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3ad82d2a33cdc9674dc7465672f271e096168fcdbe0f799d9e6db8c5892679dc"
+
[[package]]
name = "x"
version = "0.0.0"
@@ -415,8 +1089,95 @@ version = "0.0.24"
name = "yinyang-core"
version = "0.0.24"
dependencies = [
+ "blake3",
+ "borsh",
+ "futures-util",
+ "opendal",
"tokio",
"unicode-casefold",
"unicode-normalization",
"uuid",
]
+
+[[package]]
+name = "yoke"
+version = "0.8.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "709fe23a0424b6a435d82152b1bd3fdfb0833487d5fa90d05d42762a9891fef5"
+dependencies = [
+ "stable_deref_trait",
+ "yoke-derive",
+ "zerofrom",
+]
+
+[[package]]
+name = "yoke-derive"
+version = "0.8.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "de844c262c8848816172cef550288e7dc6c7b7814b4ee56b3e1553f275f1858e"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.119",
+ "synstructure",
+]
+
+[[package]]
+name = "zerofrom"
+version = "0.1.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0ec05a11813ea801ff6d75110ad09cd0824ddba17dfe17128ea0d5f68e6c5272"
+dependencies = [
+ "zerofrom-derive",
+]
+
+[[package]]
+name = "zerofrom-derive"
+version = "0.1.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "11532158c46691caf0f2593ea8358fed6bbf68a0315e80aae9bd41fbade684a1"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.119",
+ "synstructure",
+]
+
+[[package]]
+name = "zerotrie"
+version = "0.2.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4ea269c3bd32f0a32c321907a2ae912ba6f4649bb0fc764a15627e99a7095a3f"
+dependencies = [
+ "displaydoc",
+ "yoke",
+ "zerofrom",
+]
+
+[[package]]
+name = "zerovec"
+version = "0.11.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bb0464e17806c1d976d5cba29399c7f08e516e279e2ba493f63123b5fca67dd8"
+dependencies = [
+ "yoke",
+ "zerofrom",
+ "zerovec-derive",
+]
+
+[[package]]
+name = "zerovec-derive"
+version = "0.11.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "34df6fc39dbd26ddc9c10e6a2984476e13acce22e64e4487636ef494369225da"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 3.0.3",
+]
+
+[[package]]
+name = "zmij"
+version = "1.0.23"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b"
diff --git a/Cargo.toml b/Cargo.toml
index 29e8b6d..ce7227f 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -42,7 +42,11 @@ repository = "https://github.com/apache/opendal-yinyang"
rust-version = "1.91.0"
[workspace.dependencies]
+blake3 = "1"
+borsh = { version = "1.8.1", default-features = false, features = ["std"] }
clap = { version = "4.6.5", features = ["derive"] }
+futures-util = "0.3"
+opendal = { version = "0.58.2", default-features = false }
unicode-casefold = "0.2"
unicode-normalization = "0.1"
uuid = { version = "1", features = ["v4"] }
diff --git a/crates/core/Cargo.toml b/crates/core/Cargo.toml
index 889e202..9b54dac 100644
--- a/crates/core/Cargo.toml
+++ b/crates/core/Cargo.toml
@@ -16,7 +16,7 @@
# under the License.
[package]
-description = "YinYang Format semantics and publication state machine"
+description = "YinYang Format values, OpenDAL persistence, and publication state machine"
edition.workspace = true
license.workspace = true
name = "yinyang-core"
@@ -26,11 +26,16 @@ rust-version.workspace = true
version = "0.0.24"
[dependencies]
+blake3.workspace = true
+borsh.workspace = true
+futures-util.workspace = true
+opendal.workspace = true
unicode-casefold.workspace = true
unicode-normalization.workspace = true
uuid.workspace = true
[dev-dependencies]
+opendal = { workspace = true, features = ["services-memory"] }
tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
[lints]
diff --git a/crates/core/src/error.rs b/crates/core/src/error.rs
index c055d2b..fb76738 100644
--- a/crates/core/src/error.rs
+++ b/crates/core/src/error.rs
@@ -23,6 +23,7 @@ pub enum ErrorKind {
Invalid,
Corrupt,
NotFound,
+ Unsupported,
Storage,
}
@@ -62,6 +63,14 @@ impl Error {
pub(crate) fn corrupt(operation: &'static str, message: impl Into<String>) -> Self {
Self::new(ErrorKind::Corrupt, operation, message)
}
+
+ pub(crate) fn unsupported(operation: &'static str, message: impl Into<String>) -> Self {
+ Self::new(ErrorKind::Unsupported, operation, message)
+ }
+
+ pub(crate) fn from_storage(operation: &'static str, error: opendal::Error) -> Self {
+ Self::new(ErrorKind::Storage, operation, error.to_string())
+ }
}
impl fmt::Display for Error {
diff --git a/crates/core/src/lib.rs b/crates/core/src/lib.rs
index 55ab8a6..028753b 100644
--- a/crates/core/src/lib.rs
+++ b/crates/core/src/lib.rs
@@ -15,18 +15,17 @@
// specific language governing permissions and limitations
// under the License.
-//! YinYang Format semantics and publication state machine.
+//! YinYang Format values, OpenDAL persistence, and publication state machine.
mod error;
mod filesystem;
mod identity;
+mod persistence;
mod publication;
-mod storage;
mod version;
pub use error::{Error, ErrorKind, Result};
pub use filesystem::{BlobRef, ContentId, File, FilePart, Node, NodeBody, Path, Tree};
pub use identity::{CommitId, Generation, NodeId};
pub use publication::{CommitOutcome, Fs, Observation};
-pub use storage::{HeadObservation, Storage};
pub use version::FsVersion;
diff --git a/crates/core/src/persistence.rs b/crates/core/src/persistence.rs
new file mode 100644
index 0000000..42468ca
--- /dev/null
+++ b/crates/core/src/persistence.rs
@@ -0,0 +1,783 @@
+// 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 futures_util::TryStreamExt as _;
+use opendal::{ErrorKind as StorageErrorKind, Operator, Writer};
+
+use crate::{
+ BlobRef, CommitId, ContentId, Error, File, FilePart, FsVersion, Generation, Node, NodeBody,
+ NodeId, Path, Result, Tree,
+};
+
+const HEAD_PATH: &str = ".yinyang/head";
+const VERSION_PREFIX: &str = ".yinyang/versions/";
+const HEAD_MAGIC: &[u8; 8] = b"YYHEAD01";
+const VERSION_MAGIC: &[u8; 8] = b"YYVER001";
+const MAX_HEAD_BYTES: usize = 4 * 1024;
+const MAX_MATERIALIZED_VERSION_BYTES: usize = 64 * 1024 * 1024;
+const VERSION_WRITE_BUFFER_BYTES: usize = 256 * 1024;
+const CHECKSUM_BYTES: usize = 32;
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub(crate) struct HeadObservation {
+ version: BlobRef,
+ etag: String,
+}
+
+impl HeadObservation {
+ pub(crate) const fn version(&self) -> &BlobRef {
+ &self.version
+ }
+}
+
+pub(crate) fn validate_operator(operator: &Operator) -> Result<()> {
+ let capability = operator.info().capability();
+ if !capability.read
+ || !capability.write
+ || !capability.write_can_multi
+ || !capability.write_with_if_match
+ || !capability.write_with_if_not_exists
+ {
+ return Err(Error::unsupported(
+ "use YinYang filesystem",
+ "OpenDAL backend must support read, streaming write, create-if-absent, and ETag if-match",
+ ));
+ }
+ Ok(())
+}
+
+pub(crate) async fn write_version(operator: &Operator, version: &FsVersion) -> Result<BlobRef> {
+ let path = new_version_path();
+ let writer = operator
+ .writer_with(&path)
+ .if_not_exists(true)
+ .await
+ .map_err(|error| Error::from_storage("write YinYang version", error))?;
+ let mut encoder = VersionEncoder::new(writer);
+ if let Err(error) = encoder.encode(version).await {
+ encoder.abort().await;
+ return Err(error);
+ }
+ let content = match encoder.finish().await {
+ Ok(content) => content,
+ Err(error) => {
+ encoder.abort().await;
+ return Err(error);
+ }
+ };
+ Ok(BlobRef::new(path, content))
+}
+
+pub(crate) async fn read_version(operator: &Operator, reference: &BlobRef) -> Result<FsVersion> {
+ let path = version_reference_path(reference)
+ .ok_or_else(|| Error::corrupt("read YinYang version", "version reference is invalid"))?;
+ if reference.content().length() > MAX_MATERIALIZED_VERSION_BYTES as u64 {
+ return Err(Error::unsupported(
+ "read YinYang version",
+ "version exceeds the current materialized implementation limit",
+ ));
+ }
+ if reference.content().length() < VERSION_MAGIC.len() as u64 {
+ return Err(Error::corrupt(
+ "read YinYang version",
+ "version reference is invalid",
+ ));
+ }
+ let object = read_bounded(
+ operator,
+ path,
+ MAX_MATERIALIZED_VERSION_BYTES,
+ "read YinYang version",
+ )
+ .await?
+ .ok_or_else(|| Error::corrupt("read YinYang version", "referenced version is missing"))?;
+ if content_id(&object.bytes) != reference.content() {
+ return Err(Error::corrupt(
+ "read YinYang version",
+ "version does not match its reference",
+ ));
+ }
+ decode_version(&object.bytes)
+}
+
+pub(crate) async fn create_head(operator: &Operator, version: &BlobRef) -> Result<()> {
+ let bytes = encode_head(version)?;
+ match operator
+ .write_with(HEAD_PATH, bytes)
+ .if_not_exists(true)
+ .await
+ {
+ Ok(_) => Ok(()),
+ Err(error)
+ if matches!(
+ error.kind(),
+ StorageErrorKind::AlreadyExists | StorageErrorKind::ConditionNotMatch
+ ) =>
+ {
+ Ok(())
+ }
+ Err(error) => Err(Error::from_storage("create YinYang head", error)),
+ }
+}
+
+pub(crate) async fn observe_head(operator: &Operator) -> Result<Option<HeadObservation>> {
+ let Some(object) =
+ read_bounded(operator, HEAD_PATH, MAX_HEAD_BYTES, "observe YinYang head").await?
+ else {
+ return Ok(None);
+ };
+ let etag = object.etag.ok_or_else(|| {
+ Error::unsupported(
+ "observe YinYang head",
+ "OpenDAL backend did not return an ETag with the head read",
+ )
+ })?;
+ Ok(Some(HeadObservation {
+ version: decode_head(&object.bytes)?,
+ etag,
+ }))
+}
+
+pub(crate) async fn replace_head(
+ operator: &Operator,
+ observed: &HeadObservation,
+ next: &BlobRef,
+) -> Result<bool> {
+ let bytes = encode_head(next)?;
+ match operator
+ .write_with(HEAD_PATH, bytes)
+ .if_match(&observed.etag)
+ .await
+ {
+ Ok(_) => Ok(true),
+ Err(error)
+ if matches!(
+ error.kind(),
+ StorageErrorKind::ConditionNotMatch | StorageErrorKind::NotFound
+ ) =>
+ {
+ Ok(false)
+ }
+ Err(error) => Err(Error::from_storage("replace YinYang head", error)),
+ }
+}
+
+struct StoredObject {
+ bytes: Vec<u8>,
+ etag: Option<String>,
+}
+
+async fn read_bounded(
+ operator: &Operator,
+ path: &str,
+ maximum_bytes: usize,
+ operation: &'static str,
+) -> Result<Option<StoredObject>> {
+ let reader = match operator.reader(path).await {
+ Ok(reader) => reader,
+ Err(error) if error.kind() == StorageErrorKind::NotFound => return Ok(None),
+ Err(error) => return Err(Error::from_storage(operation, error)),
+ };
+ let mut stream = match reader.into_stream(..).await {
+ Ok(stream) => stream,
+ Err(error) if error.kind() == StorageErrorKind::NotFound => return Ok(None),
+ Err(error) => return Err(Error::from_storage(operation, error)),
+ };
+ let (capacity, etag) = match stream.metadata().await {
+ Ok(metadata) => {
+ let length = usize::try_from(metadata.content_length())
+ .ok()
+ .filter(|length| *length <= maximum_bytes)
+ .ok_or_else(|| Error::corrupt(operation, "object exceeds its size limit"))?;
+ (length, metadata.etag().map(str::to_owned))
+ }
+ Err(error) if error.kind() == StorageErrorKind::NotFound => return Ok(None),
+ Err(error) if error.kind() == StorageErrorKind::Unsupported => (0, None),
+ Err(error) => return Err(Error::from_storage(operation, error)),
+ };
+
+ let mut bytes = Vec::with_capacity(capacity);
+ while let Some(buffer) = stream
+ .try_next()
+ .await
+ .map_err(|error| Error::from_storage(operation, error))?
+ {
+ if buffer.len() > maximum_bytes.saturating_sub(bytes.len()) {
+ return Err(Error::corrupt(operation, "object exceeds its size limit"));
+ }
+ for chunk in buffer {
+ bytes.extend_from_slice(&chunk);
+ }
+ }
+ Ok(Some(StoredObject { bytes, etag }))
+}
+
+fn decode_version(bytes: &[u8]) -> Result<FsVersion> {
+ if bytes.len() > MAX_MATERIALIZED_VERSION_BYTES {
+ return Err(Error::unsupported(
+ "decode YinYang version",
+ "version exceeds the current materialized implementation limit",
+ ));
+ }
+ if !bytes.starts_with(VERSION_MAGIC) {
+ return Err(Error::corrupt(
+ "decode YinYang version",
+ "version envelope is invalid",
+ ));
+ }
+
+ let mut decoder = Decoder::new(&bytes[VERSION_MAGIC.len()..], "decode YinYang version");
+ let entry_count: u32 = decoder.value()?;
+ let mut entries = Vec::new();
+ for _ in 0..entry_count {
+ let path = Path::new(decoder.value::<String>()?).map_err(corrupt_version_value)?;
+ entries.push((path, decode_node(&mut decoder)?));
+ }
+ let commit_count: u32 = decoder.value()?;
+ let mut commits = Vec::new();
+ for _ in 0..commit_count {
+ commits.push(CommitId::from_bytes(decoder.value()?));
+ }
+ decoder.finish()?;
+
+ let tree = Tree::from_entries(entries).map_err(corrupt_version_value)?;
+ FsVersion::new(tree, commits).map_err(corrupt_version_value)
+}
+
+fn encode_head(reference: &BlobRef) -> Result<Vec<u8>> {
+ let mut encoder = Encoder::new(HEAD_MAGIC, MAX_HEAD_BYTES, "encode YinYang head");
+ encode_blob_ref(&mut encoder, reference)?;
+ let checksum = blake3::hash(encoder.as_bytes());
+ encoder.raw_bytes(checksum.as_bytes())?;
+ Ok(encoder.finish())
+}
+
+fn decode_head(bytes: &[u8]) -> Result<BlobRef> {
+ if bytes.len() < HEAD_MAGIC.len() + CHECKSUM_BYTES
+ || bytes.len() > MAX_HEAD_BYTES
+ || !bytes.starts_with(HEAD_MAGIC)
+ {
+ return Err(Error::corrupt(
+ "decode YinYang head",
+ "head envelope is invalid",
+ ));
+ }
+ let checksum_offset = bytes.len() - CHECKSUM_BYTES;
+ if blake3::hash(&bytes[..checksum_offset]).as_bytes() != &bytes[checksum_offset..] {
+ return Err(Error::corrupt(
+ "decode YinYang head",
+ "head checksum is invalid",
+ ));
+ }
+ let mut decoder = Decoder::new(
+ &bytes[HEAD_MAGIC.len()..checksum_offset],
+ "decode YinYang head",
+ );
+ let reference = decode_blob_ref(&mut decoder)?;
+ decoder.finish()?;
+ Ok(reference)
+}
+
+fn content_id(bytes: &[u8]) -> ContentId {
+ ContentId::new(blake3::hash(bytes).into(), bytes.len() as u64)
+}
+
+fn new_version_path() -> String {
+ format!("{VERSION_PREFIX}{}", uuid::Uuid::new_v4().simple())
+}
+
+fn version_reference_path(reference: &BlobRef) -> Option<&str> {
+ let path = std::str::from_utf8(reference.as_bytes()).ok()?;
+ let key = path.strip_prefix(VERSION_PREFIX)?;
+ (key.len() == 32
+ && key
+ .bytes()
+ .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)))
+ .then_some(path)
+}
+
+fn encode_node(encoder: &mut Encoder, node: &Node) -> Result<()> {
+ encoder.value(node.id().as_bytes())?;
+ encoder.value(&node.generation().value())?;
+ encoder.value(&node.executable())?;
+ match node.body() {
+ NodeBody::Dir { entries_generation } => {
+ encoder.value(&0_u8)?;
+ encoder.value(&entries_generation.value())
+ }
+ NodeBody::File(file) => {
+ encoder.value(&1_u8)?;
+ encode_file(encoder, file)
+ }
+ }
+}
+
+fn encode_file(encoder: &mut Encoder, file: &File) -> Result<()> {
+ encode_content_id(encoder, file.content())?;
+ encoder.sequence_len(file.parts().len())?;
+ for part in file.parts() {
+ encoder.value(&part.range().start)?;
+ encoder.value(&part.range().end)?;
+ encoder.value(&part.blob_offset())?;
+ encode_blob_ref(encoder, part.blob())?;
+ }
+ Ok(())
+}
+
+fn encode_blob_ref(encoder: &mut Encoder, reference: &BlobRef) -> Result<()> {
+ encoder.value(reference.as_bytes())?;
+ encode_content_id(encoder, reference.content())
+}
+
+fn encode_content_id(encoder: &mut Encoder, content: ContentId) -> Result<()> {
+ encoder.value(content.digest())?;
+ encoder.value(&content.length())
+}
+
+fn decode_node(decoder: &mut Decoder<'_>) -> Result<Node> {
+ let id = NodeId::from_bytes(decoder.value()?);
+ let generation = Generation::from_value(decoder.value()?);
+ let executable = decoder.value()?;
+ match decoder.value::<u8>()? {
+ 0 => Ok(Node::dir(
+ id,
+ generation,
+ executable,
+ Generation::from_value(decoder.value()?),
+ )),
+ 1 => Ok(Node::file(
+ id,
+ generation,
+ executable,
+ decode_file(decoder)?,
+ )),
+ _ => Err(decoder.corrupt("node kind is invalid")),
+ }
+}
+
+fn decode_file(decoder: &mut Decoder<'_>) -> Result<File> {
+ let content = decode_content_id(decoder)?;
+ let part_count: u32 = decoder.value()?;
+ let mut parts = Vec::new();
+ for _ in 0..part_count {
+ let start = decoder.value()?;
+ let end = decoder.value()?;
+ let blob_offset = decoder.value()?;
+ let blob = decode_blob_ref(decoder)?;
+ parts.push(FilePart::new(start..end, blob_offset, blob).map_err(corrupt_version_value)?);
+ }
+ File::new(content, parts).map_err(corrupt_version_value)
+}
+
+fn decode_blob_ref(decoder: &mut Decoder<'_>) -> Result<BlobRef> {
+ let reference = decoder.value::<Vec<u8>>()?;
+ Ok(BlobRef::new(reference, decode_content_id(decoder)?))
+}
+
+fn decode_content_id(decoder: &mut Decoder<'_>) -> Result<ContentId> {
+ Ok(ContentId::new(decoder.value()?, decoder.value()?))
+}
+
+fn corrupt_version_value(error: Error) -> Error {
+ Error::corrupt("decode YinYang version", error.message())
+}
+
+struct VersionEncoder {
+ writer: Writer,
+ encoder: Encoder,
+ hasher: blake3::Hasher,
+}
+
+impl VersionEncoder {
+ fn new(writer: Writer) -> Self {
+ Self {
+ writer,
+ encoder: Encoder::new_with_implementation_limit(
+ VERSION_MAGIC,
+ MAX_MATERIALIZED_VERSION_BYTES,
+ "encode YinYang version",
+ ),
+ hasher: blake3::Hasher::new(),
+ }
+ }
+
+ async fn encode(&mut self, version: &FsVersion) -> Result<()> {
+ self.encoder.sequence_len(version.tree().iter().count())?;
+ for (path, node) in version.tree().iter() {
+ self.encoder.value(path.as_str())?;
+ encode_node(&mut self.encoder, node)?;
+ self.flush_if_full().await?;
+ }
+ self.encoder.sequence_len(version.commits().len())?;
+ for commit in version.commits() {
+ self.encoder.value(commit.as_bytes())?;
+ self.flush_if_full().await?;
+ }
+ Ok(())
+ }
+
+ async fn flush_if_full(&mut self) -> Result<()> {
+ if self.encoder.buffered_len() >= VERSION_WRITE_BUFFER_BYTES {
+ self.flush().await?;
+ }
+ Ok(())
+ }
+
+ async fn flush(&mut self) -> Result<()> {
+ let bytes = self.encoder.take_bytes();
+ if bytes.is_empty() {
+ return Ok(());
+ }
+ self.hasher.update(&bytes);
+ self.writer
+ .write(bytes)
+ .await
+ .map_err(|error| Error::from_storage("write YinYang version", error))
+ }
+
+ async fn finish(&mut self) -> Result<ContentId> {
+ self.flush().await?;
+ self.writer
+ .close()
+ .await
+ .map_err(|error| Error::from_storage("write YinYang version", error))?;
+ Ok(ContentId::new(
+ self.hasher.finalize().into(),
+ self.encoder.encoded_len() as u64,
+ ))
+ }
+
+ async fn abort(&mut self) {
+ let _ = self.writer.abort().await;
+ }
+}
+
+struct Encoder {
+ bytes: Vec<u8>,
+ flushed_bytes: usize,
+ maximum_bytes: usize,
+ operation: &'static str,
+ implementation_limit: bool,
+ limit_exceeded: bool,
+}
+
+impl Encoder {
+ fn new(magic: &[u8], maximum_bytes: usize, operation: &'static str) -> Self {
+ Self {
+ bytes: magic.to_vec(),
+ flushed_bytes: 0,
+ maximum_bytes,
+ operation,
+ implementation_limit: false,
+ limit_exceeded: false,
+ }
+ }
+
+ fn new_with_implementation_limit(
+ magic: &[u8],
+ maximum_bytes: usize,
+ operation: &'static str,
+ ) -> Self {
+ Self {
+ bytes: magic.to_vec(),
+ flushed_bytes: 0,
+ maximum_bytes,
+ operation,
+ implementation_limit: true,
+ limit_exceeded: false,
+ }
+ }
+
+ fn sequence_len(&mut self, value: usize) -> Result<()> {
+ let value = u32::try_from(value)
+ .map_err(|_| Error::invalid(self.operation, "length cannot be encoded"))?;
+ self.value(&value)
+ }
+
+ fn value<T: borsh::BorshSerialize + ?Sized>(&mut self, value: &T) -> Result<()> {
+ value
+ .serialize(self)
+ .map_err(|error| self.encode_error(error))
+ }
+
+ fn raw_bytes(&mut self, value: &[u8]) -> Result<()> {
+ borsh::io::Write::write_all(self, value).map_err(|error| self.encode_error(error))
+ }
+
+ fn as_bytes(&self) -> &[u8] {
+ &self.bytes
+ }
+
+ fn buffered_len(&self) -> usize {
+ self.bytes.len()
+ }
+
+ fn encoded_len(&self) -> usize {
+ self.flushed_bytes + self.bytes.len()
+ }
+
+ fn take_bytes(&mut self) -> Vec<u8> {
+ self.flushed_bytes += self.bytes.len();
+ std::mem::replace(
+ &mut self.bytes,
+ Vec::with_capacity(VERSION_WRITE_BUFFER_BYTES),
+ )
+ }
+
+ fn encode_error(&self, error: borsh::io::Error) -> Error {
+ if self.implementation_limit && self.limit_exceeded {
+ Error::unsupported(
+ self.operation,
+ "version exceeds the current materialized implementation limit",
+ )
+ } else {
+ Error::invalid(self.operation, error.to_string())
+ }
+ }
+
+ fn finish(self) -> Vec<u8> {
+ debug_assert_eq!(self.flushed_bytes, 0);
+ self.bytes
+ }
+}
+
+impl borsh::io::Write for Encoder {
+ fn write(&mut self, value: &[u8]) -> borsh::io::Result<usize> {
+ if value.len() > self.maximum_bytes.saturating_sub(self.encoded_len()) {
+ self.limit_exceeded = true;
+ return Err(borsh::io::Error::new(
+ borsh::io::ErrorKind::InvalidData,
+ "encoded object exceeds its size limit",
+ ));
+ }
+ self.bytes.extend_from_slice(value);
+ Ok(value.len())
+ }
+
+ fn flush(&mut self) -> borsh::io::Result<()> {
+ Ok(())
+ }
+}
+
+struct Decoder<'a> {
+ bytes: &'a [u8],
+ operation: &'static str,
+}
+
+impl<'a> Decoder<'a> {
+ const fn new(bytes: &'a [u8], operation: &'static str) -> Self {
+ Self { bytes, operation }
+ }
+
+ fn value<T: borsh::BorshDeserialize>(&mut self) -> Result<T> {
+ T::deserialize(&mut self.bytes).map_err(|error| self.corrupt(error.to_string()))
+ }
+
+ fn finish(self) -> Result<()> {
+ if !self.bytes.is_empty() {
+ return Err(self.corrupt("encoded object contains trailing bytes"));
+ }
+ Ok(())
+ }
+
+ fn corrupt(&self, message: impl Into<String>) -> Error {
+ Error::corrupt(self.operation, message)
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use opendal::services::Memory;
+
+ async fn persisted_version_bytes(version: &FsVersion) -> Vec<u8> {
+ let operator = Operator::new(Memory::default()).unwrap();
+ let reference = write_version(&operator, version).await.unwrap();
+ let path = version_reference_path(&reference).unwrap();
+ let bytes = operator.read(path).await.unwrap().to_bytes().to_vec();
+ assert_eq!(content_id(&bytes), reference.content());
+ bytes
+ }
+
+ #[tokio::test]
+ async fn version_wire_contract_is_stable() {
+ let root = NodeId::from_bytes([1; 16]);
+ let version = FsVersion::new(Tree::genesis(root), Vec::new()).unwrap();
+
+ let actual = persisted_version_bytes(&version).await;
+ let mut expected = VERSION_MAGIC.to_vec();
+ expected.extend_from_slice(&1_u32.to_le_bytes());
+ expected.extend_from_slice(&0_u32.to_le_bytes());
+ expected.extend_from_slice(&[1; 16]);
+ expected.extend_from_slice(&1_u64.to_le_bytes());
+ expected.push(0);
+ expected.push(0);
+ expected.extend_from_slice(&1_u64.to_le_bytes());
+ expected.extend_from_slice(&0_u32.to_le_bytes());
+
+ assert_eq!(actual, expected);
+ assert_eq!(decode_version(&actual).unwrap(), version);
+ }
+
+ #[tokio::test]
+ async fn file_version_wire_contract_is_stable() {
+ let root = NodeId::from_bytes([1; 16]);
+ let mut tree = Tree::genesis(root);
+ tree.insert(
+ Path::root(),
+ Node::dir(root, Generation::FIRST, false, Generation::from_value(2)),
+ );
+ let blob = BlobRef::new(b"blob".to_vec(), ContentId::new([4; 32], 8));
+ let file = File::new(
+ ContentId::new([3; 32], 4),
+ vec![FilePart::new(0..4, 1, blob).unwrap()],
+ )
+ .unwrap();
+ tree.insert(
+ Path::new("file").unwrap(),
+ Node::file(
+ NodeId::from_bytes([2; 16]),
+ Generation::from_value(3),
+ true,
+ file,
+ ),
+ );
+ let version = FsVersion::new(tree, vec![CommitId::from_bytes([5; 16])]).unwrap();
+
+ let actual = persisted_version_bytes(&version).await;
+ let mut expected = VERSION_MAGIC.to_vec();
+ expected.extend_from_slice(&2_u32.to_le_bytes());
+ expected.extend_from_slice(&0_u32.to_le_bytes());
+ expected.extend_from_slice(&[1; 16]);
+ expected.extend_from_slice(&1_u64.to_le_bytes());
+ expected.push(0);
+ expected.push(0);
+ expected.extend_from_slice(&2_u64.to_le_bytes());
+ expected.extend_from_slice(&4_u32.to_le_bytes());
+ expected.extend_from_slice(b"file");
+ expected.extend_from_slice(&[2; 16]);
+ expected.extend_from_slice(&3_u64.to_le_bytes());
+ expected.push(1);
+ expected.push(1);
+ expected.extend_from_slice(&[3; 32]);
+ expected.extend_from_slice(&4_u64.to_le_bytes());
+ expected.extend_from_slice(&1_u32.to_le_bytes());
+ expected.extend_from_slice(&0_u64.to_le_bytes());
+ expected.extend_from_slice(&4_u64.to_le_bytes());
+ expected.extend_from_slice(&1_u64.to_le_bytes());
+ expected.extend_from_slice(&4_u32.to_le_bytes());
+ expected.extend_from_slice(b"blob");
+ expected.extend_from_slice(&[4; 32]);
+ expected.extend_from_slice(&8_u64.to_le_bytes());
+ expected.extend_from_slice(&1_u32.to_le_bytes());
+ expected.extend_from_slice(&[5; 16]);
+
+ assert_eq!(actual, expected);
+ assert_eq!(decode_version(&actual).unwrap(), version);
+ }
+
+ #[tokio::test]
+ async fn version_decoder_rejects_invalid_borsh() {
+ let root = NodeId::from_bytes([1; 16]);
+ let version = FsVersion::new(Tree::genesis(root), Vec::new()).unwrap();
+ let encoded = persisted_version_bytes(&version).await;
+
+ let mut invalid_boolean = encoded.clone();
+ invalid_boolean[40] = 2;
+ assert_eq!(
+ decode_version(&invalid_boolean).unwrap_err().kind(),
+ crate::ErrorKind::Corrupt
+ );
+
+ let mut invalid_node_kind = encoded.clone();
+ invalid_node_kind[41] = 2;
+ assert_eq!(
+ decode_version(&invalid_node_kind).unwrap_err().kind(),
+ crate::ErrorKind::Corrupt
+ );
+
+ assert_eq!(
+ decode_version(&encoded[..encoded.len() - 1])
+ .unwrap_err()
+ .kind(),
+ crate::ErrorKind::Corrupt
+ );
+
+ let mut trailing = encoded;
+ trailing.push(0);
+ assert_eq!(
+ decode_version(&trailing).unwrap_err().kind(),
+ crate::ErrorKind::Corrupt
+ );
+ }
+
+ #[test]
+ fn head_wire_contract_is_stable() {
+ let reference = BlobRef::new(b"v".to_vec(), ContentId::new([2; 32], 3));
+
+ let actual = encode_head(&reference).unwrap();
+ let mut expected = HEAD_MAGIC.to_vec();
+ expected.extend_from_slice(&1_u32.to_le_bytes());
+ expected.push(b'v');
+ expected.extend_from_slice(&[2; 32]);
+ expected.extend_from_slice(&3_u64.to_le_bytes());
+ let checksum = blake3::hash(&expected);
+ expected.extend_from_slice(checksum.as_bytes());
+
+ assert_eq!(actual, expected);
+ assert_eq!(decode_head(&actual).unwrap(), reference);
+ }
+
+ #[test]
+ fn oversized_materialized_version_is_unsupported() {
+ let mut encoder = Encoder::new_with_implementation_limit(
+ VERSION_MAGIC,
+ VERSION_MAGIC.len(),
+ "encode YinYang version",
+ );
+
+ let error = encoder.value(&0_u8).unwrap_err();
+
+ assert_eq!(error.kind(), crate::ErrorKind::Unsupported);
+ }
+
+ #[tokio::test]
+ async fn rejects_invalid_version_object_keys() {
+ let operator = Operator::new(Memory::default()).unwrap();
+ let reference = BlobRef::new(
+ b".yinyang/versions/not-a-version-id".to_vec(),
+ ContentId::new([0; 32], VERSION_MAGIC.len() as u64),
+ );
+
+ let error = read_version(&operator, &reference).await.unwrap_err();
+
+ assert_eq!(error.kind(), crate::ErrorKind::Corrupt);
+ }
+
+ #[tokio::test]
+ async fn oversized_version_reference_is_unsupported() {
+ let operator = Operator::new(Memory::default()).unwrap();
+ let reference = BlobRef::new(
+ new_version_path(),
+ ContentId::new([0; 32], MAX_MATERIALIZED_VERSION_BYTES as u64 + 1),
+ );
+
+ let error = read_version(&operator, &reference).await.unwrap_err();
+
+ assert_eq!(error.kind(), crate::ErrorKind::Unsupported);
+ }
+}
diff --git a/crates/core/src/publication.rs b/crates/core/src/publication.rs
index 8f4c1cf..39799b3 100644
--- a/crates/core/src/publication.rs
+++ b/crates/core/src/publication.rs
@@ -15,9 +15,11 @@
// specific language governing permissions and limitations
// under the License.
-use crate::{
- CommitId, Error, ErrorKind, FsVersion, HeadObservation, NodeId, Result, Storage, Tree,
+use crate::persistence::{
+ HeadObservation, create_head, observe_head, read_version, replace_head, validate_operator,
+ write_version,
};
+use crate::{CommitId, Error, ErrorKind, FsVersion, NodeId, Result, Tree};
/// Current immutable version together with the condition needed for publication.
#[derive(Clone, Debug, Eq, PartialEq)]
@@ -43,43 +45,61 @@ pub enum CommitOutcome {
Conflict { current: u64 },
}
-/// Open YinYang filesystem over one storage implementation.
+/// Open YinYang filesystem over one OpenDAL operator.
+///
+/// YinYang owns the `.yinyang/` prefix below the operator root. The backend
+/// must support read, write, create-if-absent, and ETag if-match writes.
#[derive(Debug)]
-pub struct Fs<S> {
+pub struct Fs {
root: NodeId,
- storage: S,
+ operator: opendal::Operator,
}
-impl<S: Storage> Fs<S> {
+impl Fs {
/// Create a filesystem, or reopen the existing filesystem.
- pub async fn create(storage: S) -> Result<Self> {
- if let Some(head) = storage.observe_head().await? {
- return Self::from_head(storage, head).await;
+ ///
+ /// Returns [`ErrorKind::Unsupported`] when the operator lacks a required
+ /// capability or does not return an ETag with the head read.
+ pub async fn create(operator: opendal::Operator) -> Result<Self> {
+ validate_operator(&operator)?;
+ if let Some(head) = observe_head(&operator).await? {
+ return Self::from_head(operator, head).await;
}
let root = NodeId::generate();
let genesis = FsVersion::new(Tree::genesis(root), Vec::new())?;
- let blob = storage.write_version(&genesis).await?;
- storage.create_head(&blob).await?;
- Self::open(storage).await
+ let blob = write_version(&operator, &genesis).await?;
+ if let Err(create_error) = create_head(&operator, &blob).await {
+ return match Self::open(operator).await {
+ Ok(filesystem) => Ok(filesystem),
+ Err(open_error) if open_error.kind() == ErrorKind::NotFound => Err(create_error),
+ Err(open_error) => Err(open_error),
+ };
+ }
+ Self::open(operator).await
}
/// Open and validate an existing filesystem.
- pub async fn open(storage: S) -> Result<Self> {
- let head = storage.observe_head().await?.ok_or_else(|| {
+ ///
+ /// Returns [`ErrorKind::NotFound`] when no head exists and
+ /// [`ErrorKind::Unsupported`] when the operator lacks a required
+ /// capability or does not return an ETag with the head read.
+ pub async fn open(operator: opendal::Operator) -> Result<Self> {
+ validate_operator(&operator)?;
+ let head = observe_head(&operator).await?.ok_or_else(|| {
Error::new(
ErrorKind::NotFound,
"open YinYang filesystem",
"head is missing",
)
})?;
- Self::from_head(storage, head).await
+ Self::from_head(operator, head).await
}
- async fn from_head(storage: S, head: HeadObservation) -> Result<Self> {
- let version = storage.read_version(head.version()).await?;
+ async fn from_head(operator: opendal::Operator, head: HeadObservation) -> Result<Self> {
+ let version = read_version(&operator, head.version()).await?;
let root = version.tree().root_id()?;
- Ok(Self { root, storage })
+ Ok(Self { root, operator })
}
pub const fn root(&self) -> NodeId {
@@ -88,7 +108,7 @@ impl<S: Storage> Fs<S> {
/// Observe the head and materialize its immutable version.
pub async fn observe(&self) -> Result<Observation> {
- let head = self.storage.observe_head().await?.ok_or_else(|| {
+ let head = observe_head(&self.operator).await?.ok_or_else(|| {
Error::new(
ErrorKind::NotFound,
"observe YinYang filesystem",
@@ -118,8 +138,8 @@ impl<S: Storage> Fs<S> {
.tree()
.validate_successor(version.tree(), self.root)?;
let number = version.number();
- let blob = self.storage.write_version(&version).await?;
- match self.storage.replace_head(&observed.head, &blob).await {
+ let blob = write_version(&self.operator, &version).await?;
+ match replace_head(&self.operator, &observed.head, &blob).await {
Ok(true) => Ok(CommitOutcome::Committed { version: number }),
Ok(false) => self.resolve_conflict(id).await,
Err(error) => match self.find_commit(id).await {
@@ -130,7 +150,7 @@ impl<S: Storage> Fs<S> {
}
async fn read_observation(&self, head: HeadObservation) -> Result<Observation> {
- let version = self.storage.read_version(head.version()).await?;
+ let version = read_version(&self.operator, head.version()).await?;
version.validate_root(self.root)?;
Ok(Observation { head, version })
}
diff --git a/crates/core/src/storage.rs b/crates/core/src/storage.rs
deleted file mode 100644
index df39653..0000000
--- a/crates/core/src/storage.rs
+++ /dev/null
@@ -1,72 +0,0 @@
-// 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::future::Future;
-
-use crate::{BlobRef, FsVersion, Result};
-
-/// One head value and the opaque condition required to replace that observation.
-#[derive(Clone, Debug, Eq, PartialEq)]
-pub struct HeadObservation {
- version: BlobRef,
- condition: Vec<u8>,
-}
-
-impl HeadObservation {
- pub fn new(version: BlobRef, condition: Vec<u8>) -> Self {
- Self { version, condition }
- }
-
- pub const fn version(&self) -> &BlobRef {
- &self.version
- }
-
- pub fn condition(&self) -> &[u8] {
- &self.condition
- }
-}
-
-/// Persistent capabilities required by the YinYang Format state machine.
-pub trait Storage: Send + Sync {
- /// Persist one immutable filesystem version and return its verifiable reference.
- fn write_version<'a>(
- &'a self,
- version: &'a FsVersion,
- ) -> impl Future<Output = Result<BlobRef>> + Send + 'a;
-
- /// Read and verify one immutable filesystem version.
- fn read_version<'a>(
- &'a self,
- reference: &'a BlobRef,
- ) -> impl Future<Output = Result<FsVersion>> + Send + 'a;
-
- /// Create the mutable head when it does not exist.
- fn create_head<'a>(
- &'a self,
- version: &'a BlobRef,
- ) -> impl Future<Output = Result<()>> + Send + 'a;
-
- /// Read the head and a condition bound to the same observed value.
- fn observe_head(&self) -> impl Future<Output = Result<Option<HeadObservation>>> + Send + '_;
-
- /// Replace the complete head when `observed` is still current.
- fn replace_head<'a>(
- &'a self,
- observed: &'a HeadObservation,
- next: &'a BlobRef,
- ) -> impl Future<Output = Result<bool>> + Send + 'a;
-}
diff --git a/crates/core/tests/core.rs b/crates/core/tests/core.rs
index bc678af..ae5926e 100644
--- a/crates/core/tests/core.rs
+++ b/crates/core/tests/core.rs
@@ -18,104 +18,318 @@
use std::collections::BTreeMap;
use std::sync::{Arc, Mutex};
+use opendal::raw::*;
+use opendal::services::Memory;
+use opendal::{Buffer, BytesRange, EntryMode, Metadata, OperationContext, Operator};
use yinyang_core::{
- BlobRef, CommitId, CommitOutcome, ContentId, Error, ErrorKind, File, FilePart, Fs, FsVersion,
- Generation, HeadObservation, Node, NodeBody, NodeId, Path, Result, Storage, Tree,
+ BlobRef, CommitId, CommitOutcome, ContentId, ErrorKind, File, FilePart, Fs, Generation, Node,
+ NodeBody, NodeId, Path, Tree,
};
#[derive(Clone, Debug, Default)]
-struct MemoryStorage {
- state: Arc<Mutex<MemoryState>>,
+struct TestBackend {
+ state: Arc<Mutex<TestState>>,
}
#[derive(Debug, Default)]
-struct MemoryState {
- head: Option<BlobRef>,
- head_revision: u64,
- versions: BTreeMap<BlobRef, FsVersion>,
- next_blob: u64,
- fail_after_replace: bool,
+struct TestState {
+ objects: BTreeMap<String, StoredObject>,
+ next_revision: u64,
+ fail_after_head_write: bool,
+ stat_calls: u64,
+ version_write_calls: u64,
}
-impl MemoryStorage {
- fn fail_next_replace_after_success(&self) {
- self.state.lock().unwrap().fail_after_replace = true;
+#[derive(Clone, Debug)]
+struct StoredObject {
+ bytes: Vec<u8>,
+ etag: String,
+}
+
+impl TestBackend {
+ fn operator(&self) -> Operator {
+ self.operator_with_streaming_write(true)
}
- fn remove_current_version(&self) {
- let mut state = self.state.lock().unwrap();
- let reference = state
- .head
- .as_ref()
- .expect("the test filesystem has a head")
- .clone();
- state.versions.remove(&reference);
+ fn operator_without_streaming_write(&self) -> Operator {
+ self.operator_with_streaming_write(false)
}
-}
-impl Storage for MemoryStorage {
- async fn write_version(&self, version: &FsVersion) -> Result<BlobRef> {
- let mut state = self.state.lock().unwrap();
- state.next_blob += 1;
- let identity = state.next_blob;
- let mut digest = [0; 32];
- digest[..8].copy_from_slice(&identity.to_le_bytes());
- let reference = BlobRef::new(identity.to_le_bytes(), ContentId::new(digest, 1));
- state.versions.insert(reference.clone(), version.clone());
- Ok(reference)
+ fn operator_with_streaming_write(&self, write_can_multi: bool) -> Operator {
+ let service: Servicer = Arc::new(TestService {
+ state: self.state.clone(),
+ write_can_multi,
+ });
+ Operator::from_parts(OperationContext::default(), service)
+ }
+
+ fn fail_next_head_write_after_success(&self) {
+ self.state.lock().unwrap().fail_after_head_write = true;
}
- async fn read_version(&self, reference: &BlobRef) -> Result<FsVersion> {
+ fn stat_calls(&self) -> u64 {
+ self.state.lock().unwrap().stat_calls
+ }
+
+ fn reset_version_write_calls(&self) {
+ self.state.lock().unwrap().version_write_calls = 0;
+ }
+
+ fn version_write_calls(&self) -> u64 {
+ self.state.lock().unwrap().version_write_calls
+ }
+
+ fn version_objects(&self) -> Vec<(String, Vec<u8>)> {
self.state
.lock()
.unwrap()
- .versions
- .get(reference)
- .cloned()
- .ok_or_else(|| {
- Error::new(
- ErrorKind::Corrupt,
- "read in-memory YinYang version",
- "referenced version is missing",
- )
- })
+ .objects
+ .iter()
+ .filter(|(path, _)| path.starts_with(".yinyang/versions/"))
+ .map(|(path, object)| (path.clone(), object.bytes.clone()))
+ .collect()
}
- async fn create_head(&self, version: &BlobRef) -> Result<()> {
+ fn remove_current_version(&self) {
+ self.state
+ .lock()
+ .unwrap()
+ .objects
+ .retain(|path, _| !path.starts_with(".yinyang/versions/"));
+ }
+
+ fn corrupt_current_version(&self) {
let mut state = self.state.lock().unwrap();
- if state.head.is_some() {
- return Ok(());
+ let object = state
+ .objects
+ .iter_mut()
+ .find_map(|(path, object)| path.starts_with(".yinyang/versions/").then_some(object))
+ .expect("the test filesystem has a version object");
+ object.bytes[0] ^= 1;
+ }
+
+ fn corrupt_head(&self) {
+ let mut state = self.state.lock().unwrap();
+ let object = state
+ .objects
+ .get_mut(".yinyang/head")
+ .expect("the test filesystem has a head");
+ let checksum = object
+ .bytes
+ .last_mut()
+ .expect("the head contains a checksum");
+ *checksum ^= 1;
+ }
+}
+
+#[derive(Debug)]
+struct TestService {
+ state: Arc<Mutex<TestState>>,
+ write_can_multi: bool,
+}
+
+impl Service for TestService {
+ type Reader = TestReader;
+ type Writer = TestWriter;
+ type Lister = ();
+ type Deleter = ();
+ type Copier = ();
+
+ fn info(&self) -> ServiceInfo {
+ ServiceInfo::with_scheme("yinyang-test")
+ }
+
+ fn capability(&self) -> opendal::Capability {
+ opendal::Capability {
+ stat: true,
+ read: true,
+ write: true,
+ write_can_multi: self.write_can_multi,
+ write_can_empty: true,
+ write_with_if_match: true,
+ write_with_if_not_exists: true,
+ shared: true,
+ ..Default::default()
}
- state.head = Some(version.clone());
- state.head_revision = 1;
- Ok(())
}
- async fn observe_head(&self) -> Result<Option<HeadObservation>> {
+ async fn create_dir(
+ &self,
+ _: &OperationContext,
+ _: &str,
+ _: OpCreateDir,
+ ) -> opendal::Result<RpCreateDir> {
+ Err(unsupported())
+ }
+
+ async fn stat(&self, _: &OperationContext, path: &str, _: OpStat) -> opendal::Result<RpStat> {
+ let mut state = self.state.lock().unwrap();
+ state.stat_calls += 1;
+ let object = state.objects.get(path).ok_or_else(not_found)?;
+ Ok(RpStat::new(metadata(object)))
+ }
+
+ fn read(&self, _: &OperationContext, path: &str, _: OpRead) -> opendal::Result<Self::Reader> {
let state = self.state.lock().unwrap();
- Ok(state
- .head
- .clone()
- .map(|head| HeadObservation::new(head, state.head_revision.to_le_bytes().to_vec())))
+ let object = state.objects.get(path).cloned().ok_or_else(not_found)?;
+ Ok(TestReader { object })
+ }
+
+ fn write(
+ &self,
+ _: &OperationContext,
+ path: &str,
+ args: OpWrite,
+ ) -> opendal::Result<Self::Writer> {
+ Ok(TestWriter {
+ state: self.state.clone(),
+ path: path.to_owned(),
+ if_match: args.if_match().map(str::to_owned),
+ if_not_exists: args.if_not_exists(),
+ bytes: Vec::new(),
+ })
+ }
+
+ fn delete(&self, _: &OperationContext) -> opendal::Result<Self::Deleter> {
+ Err(unsupported())
+ }
+
+ fn list(&self, _: &OperationContext, _: &str, _: OpList) -> opendal::Result<Self::Lister> {
+ Err(unsupported())
}
- async fn replace_head(&self, observed: &HeadObservation, next: &BlobRef) -> Result<bool> {
+ fn copy(
+ &self,
+ _: &OperationContext,
+ _: &str,
+ _: &str,
+ _: OpCopy,
+ _: OpCopier,
+ ) -> opendal::Result<Self::Copier> {
+ Err(unsupported())
+ }
+
+ async fn rename(
+ &self,
+ _: &OperationContext,
+ _: &str,
+ _: &str,
+ _: OpRename,
+ ) -> opendal::Result<RpRename> {
+ Err(unsupported())
+ }
+
+ async fn presign(
+ &self,
+ _: &OperationContext,
+ _: &str,
+ _: OpPresign,
+ ) -> opendal::Result<RpPresign> {
+ Err(unsupported())
+ }
+}
+
+#[derive(Debug)]
+struct TestReader {
+ object: StoredObject,
+}
+
+impl oio::Read for TestReader {
+ async fn open(
+ &self,
+ range: BytesRange,
+ ) -> opendal::Result<(RpRead, Box<dyn oio::ReadStreamDyn>)> {
+ let (response, buffer) = self.read(range).await?;
+ Ok((response, Box::new(buffer)))
+ }
+
+ async fn read(&self, range: BytesRange) -> opendal::Result<(RpRead, Buffer)> {
+ let range = range.to_content_range(self.object.bytes.len())?;
+ Ok((
+ RpRead::new(metadata(&self.object)),
+ Buffer::from(self.object.bytes[range].to_vec()),
+ ))
+ }
+}
+
+#[derive(Debug)]
+struct TestWriter {
+ state: Arc<Mutex<TestState>>,
+ path: String,
+ if_match: Option<String>,
+ if_not_exists: bool,
+ bytes: Vec<u8>,
+}
+
+impl oio::Write for TestWriter {
+ async fn write(&mut self, buffer: Buffer) -> opendal::Result<()> {
+ if self.path.starts_with(".yinyang/versions/") {
+ self.state.lock().unwrap().version_write_calls += 1;
+ }
+ for chunk in buffer {
+ self.bytes.extend_from_slice(&chunk);
+ }
+ Ok(())
+ }
+
+ async fn close(&mut self) -> opendal::Result<Metadata> {
let mut state = self.state.lock().unwrap();
- if state.head.is_none() || observed.condition() != state.head_revision.to_le_bytes() {
- return Ok(false);
+ let current = state.objects.get(&self.path);
+ if self.if_not_exists && current.is_some() {
+ return Err(opendal::Error::new(
+ opendal::ErrorKind::ConditionNotMatch,
+ "object already exists",
+ ));
}
- state.head = Some(next.clone());
- state.head_revision += 1;
- if state.fail_after_replace {
- state.fail_after_replace = false;
- return Err(Error::new(
- ErrorKind::Storage,
- "replace in-memory YinYang head",
+ if self
+ .if_match
+ .as_ref()
+ .is_some_and(|etag| current.is_none_or(|object| object.etag != *etag))
+ {
+ return Err(opendal::Error::new(
+ opendal::ErrorKind::ConditionNotMatch,
+ "ETag does not match",
+ ));
+ }
+
+ state.next_revision += 1;
+ let object = StoredObject {
+ bytes: self.bytes.clone(),
+ etag: format!("\"{}\"", state.next_revision),
+ };
+ let metadata = metadata(&object);
+ state.objects.insert(self.path.clone(), object);
+ if self.path == ".yinyang/head" && state.fail_after_head_write {
+ state.fail_after_head_write = false;
+ return Err(opendal::Error::new(
+ opendal::ErrorKind::Unexpected,
"publication response was lost",
));
}
- Ok(true)
+ Ok(metadata)
}
+
+ async fn abort(&mut self) -> opendal::Result<()> {
+ self.bytes.clear();
+ Ok(())
+ }
+}
+
+fn metadata(object: &StoredObject) -> Metadata {
+ Metadata::new(EntryMode::FILE)
+ .with_content_length(object.bytes.len() as u64)
+ .with_etag(object.etag.clone())
+}
+
+fn not_found() -> opendal::Error {
+ opendal::Error::new(opendal::ErrorKind::NotFound, "object is missing")
+}
+
+fn unsupported() -> opendal::Error {
+ opendal::Error::new(
+ opendal::ErrorKind::Unsupported,
+ "operation is not supported by the test backend",
+ )
}
fn add_directory(tree: &Tree, name: &str, id: NodeId) -> Tree {
@@ -150,13 +364,21 @@ fn advance_root_membership(tree: &mut Tree) {
#[tokio::test]
async fn creates_and_reopens_one_filesystem() {
- let storage = MemoryStorage::default();
- let (left, right) = tokio::join!(Fs::create(storage.clone()), Fs::create(storage.clone()));
+ let backend = TestBackend::default();
+ let (left, right) = tokio::join!(
+ Fs::create(backend.operator()),
+ Fs::create(backend.operator())
+ );
let left = left.unwrap();
let right = right.unwrap();
assert_eq!(left.root(), right.root());
- let observed = Fs::open(storage).await.unwrap().observe().await.unwrap();
+ let observed = Fs::open(backend.operator())
+ .await
+ .unwrap()
+ .observe()
+ .await
+ .unwrap();
assert_eq!(observed.version().number(), 0);
assert_eq!(
observed.tree().get(&Path::root()).unwrap().id(),
@@ -164,10 +386,77 @@ async fn creates_and_reopens_one_filesystem() {
);
}
+#[tokio::test]
+async fn streams_versions_to_opaque_object_keys() {
+ let backend = TestBackend::default();
+ let filesystem = Fs::create(backend.operator()).await.unwrap();
+ let observed = filesystem.observe().await.unwrap();
+ let mut successor = observed.tree().clone();
+ advance_root_membership(&mut successor);
+ for index in 0..10_000 {
+ successor.insert(
+ Path::new(format!("dir-{index:05}")).unwrap(),
+ Node::dir(
+ NodeId::generate(),
+ Generation::FIRST,
+ false,
+ Generation::FIRST,
+ ),
+ );
+ }
+ backend.reset_version_write_calls();
+
+ filesystem
+ .commit(&observed, CommitId::generate(), successor)
+ .await
+ .unwrap();
+
+ assert!(backend.version_write_calls() > 1);
+ for (path, bytes) in backend.version_objects() {
+ let key = path.strip_prefix(".yinyang/versions/").unwrap();
+ assert_eq!(key.len(), 32);
+ assert!(
+ key.bytes()
+ .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
+ );
+ assert_ne!(key, blake3::hash(&bytes).to_hex().as_str());
+ }
+}
+
+#[tokio::test]
+async fn resolves_a_lost_head_creation_response() {
+ let backend = TestBackend::default();
+ backend.fail_next_head_write_after_success();
+
+ let filesystem = Fs::create(backend.operator()).await.unwrap();
+
+ assert_eq!(filesystem.observe().await.unwrap().version().number(), 0);
+}
+
+#[tokio::test]
+async fn rejects_backends_without_conditional_head_replacement() {
+ let operator = Operator::new(Memory::default()).unwrap();
+
+ let error = Fs::create(operator).await.unwrap_err();
+
+ assert_eq!(error.kind(), ErrorKind::Unsupported);
+}
+
+#[tokio::test]
+async fn rejects_backends_without_streaming_write() {
+ let backend = TestBackend::default();
+
+ let error = Fs::create(backend.operator_without_streaming_write())
+ .await
+ .unwrap_err();
+
+ assert_eq!(error.kind(), ErrorKind::Unsupported);
+}
+
#[tokio::test]
async fn publishes_and_retries_one_commit() {
- let storage = MemoryStorage::default();
- let filesystem = Fs::create(storage.clone()).await.unwrap();
+ let backend = TestBackend::default();
+ let filesystem = Fs::create(backend.operator()).await.unwrap();
let observed = filesystem.observe().await.unwrap();
let commit = CommitId::generate();
let successor = add_directory(observed.tree(), "dir", NodeId::generate());
@@ -187,16 +476,33 @@ async fn publishes_and_retries_one_commit() {
CommitOutcome::Committed { version: 1 }
);
- let reopened = Fs::open(storage).await.unwrap();
+ let reopened = Fs::open(backend.operator()).await.unwrap();
let current = reopened.observe().await.unwrap();
assert!(current.tree().get(&Path::new("dir").unwrap()).is_some());
assert_eq!(current.version().commits().len(), 1);
assert_eq!(current.version().commits()[0], commit);
}
+#[tokio::test]
+async fn uses_the_etag_from_the_same_head_read() {
+ let backend = TestBackend::default();
+ let filesystem = Fs::create(backend.operator()).await.unwrap();
+ let observed = filesystem.observe().await.unwrap();
+ let successor = add_directory(observed.tree(), "dir", NodeId::generate());
+
+ assert_eq!(
+ filesystem
+ .commit(&observed, CommitId::generate(), successor)
+ .await
+ .unwrap(),
+ CommitOutcome::Committed { version: 1 }
+ );
+ assert_eq!(backend.stat_calls(), 0);
+}
+
#[tokio::test]
async fn reports_conflict_for_competing_observations() {
- let filesystem = Fs::create(MemoryStorage::default()).await.unwrap();
+ let filesystem = Fs::create(TestBackend::default().operator()).await.unwrap();
let first = filesystem.observe().await.unwrap();
let second = filesystem.observe().await.unwrap();
@@ -220,11 +526,11 @@ async fn reports_conflict_for_competing_observations() {
#[tokio::test]
async fn resolves_a_lost_publication_response() {
- let storage = MemoryStorage::default();
- let filesystem = Fs::create(storage.clone()).await.unwrap();
+ let backend = TestBackend::default();
+ let filesystem = Fs::create(backend.operator()).await.unwrap();
let observed = filesystem.observe().await.unwrap();
let successor = add_directory(observed.tree(), "durable", NodeId::generate());
- storage.fail_next_replace_after_success();
+ backend.fail_next_head_write_after_success();
assert_eq!(
filesystem
@@ -237,7 +543,7 @@ async fn resolves_a_lost_publication_response() {
#[tokio::test]
async fn requires_directory_generation_for_membership_changes() {
- let filesystem = Fs::create(MemoryStorage::default()).await.unwrap();
+ let filesystem = Fs::create(TestBackend::default().operator()).await.unwrap();
let observed = filesystem.observe().await.unwrap();
let mut invalid = observed.tree().clone();
invalid.insert(
@@ -259,7 +565,7 @@ async fn requires_directory_generation_for_membership_changes() {
#[tokio::test]
async fn rejects_duplicate_nodes_and_missing_parents() {
- let filesystem = Fs::create(MemoryStorage::default()).await.unwrap();
+ let filesystem = Fs::create(TestBackend::default().operator()).await.unwrap();
let observed = filesystem.observe().await.unwrap();
let node = NodeId::generate();
let mut duplicate = add_directory(observed.tree(), "first", node);
@@ -298,7 +604,7 @@ async fn rejects_duplicate_nodes_and_missing_parents() {
#[tokio::test]
async fn rejects_case_folding_collisions_within_a_directory() {
- let filesystem = Fs::create(MemoryStorage::default()).await.unwrap();
+ let filesystem = Fs::create(TestBackend::default().operator()).await.unwrap();
let observed = filesystem.observe().await.unwrap();
let mut successor = observed.tree().clone();
advance_root_membership(&mut successor);
@@ -330,7 +636,7 @@ async fn rejects_case_folding_collisions_within_a_directory() {
#[tokio::test]
async fn requires_node_generation_for_file_changes() {
- let filesystem = Fs::create(MemoryStorage::default()).await.unwrap();
+ let filesystem = Fs::create(TestBackend::default().operator()).await.unwrap();
let genesis = filesystem.observe().await.unwrap();
let node = NodeId::generate();
let mut first = genesis.tree().clone();
@@ -372,7 +678,7 @@ async fn requires_node_generation_for_file_changes() {
#[tokio::test]
async fn preserves_node_generation_across_rename() {
- let filesystem = Fs::create(MemoryStorage::default()).await.unwrap();
+ let filesystem = Fs::create(TestBackend::default().operator()).await.unwrap();
let genesis = filesystem.observe().await.unwrap();
let node_id = NodeId::generate();
filesystem
@@ -446,10 +752,32 @@ fn accepts_only_canonical_portable_paths() {
#[tokio::test]
async fn rejects_a_missing_referenced_version() {
- let storage = MemoryStorage::default();
- let filesystem = Fs::create(storage.clone()).await.unwrap();
- storage.remove_current_version();
+ let backend = TestBackend::default();
+ let filesystem = Fs::create(backend.operator()).await.unwrap();
+ backend.remove_current_version();
let error = filesystem.observe().await.unwrap_err();
assert_eq!(error.kind(), ErrorKind::Corrupt);
}
+
+#[tokio::test]
+async fn rejects_a_version_that_does_not_match_its_reference() {
+ let backend = TestBackend::default();
+ let filesystem = Fs::create(backend.operator()).await.unwrap();
+ backend.corrupt_current_version();
+
+ let error = filesystem.observe().await.unwrap_err();
+
+ assert_eq!(error.kind(), ErrorKind::Corrupt);
+}
+
+#[tokio::test]
+async fn rejects_a_corrupt_head() {
+ let backend = TestBackend::default();
+ Fs::create(backend.operator()).await.unwrap();
+ backend.corrupt_head();
+
+ let error = Fs::open(backend.operator()).await.unwrap_err();
+
+ assert_eq!(error.kind(), ErrorKind::Corrupt);
+}
diff --git a/specs/yinyang-format.md b/specs/yinyang-format.md
index c1ea81f..9ac4a92 100644
--- a/specs/yinyang-format.md
+++ b/specs/yinyang-format.md
@@ -2,11 +2,11 @@
## Scope
-The YinYang Format core defines the filesystem values and publication state
-machine independently of storage representation. It owns the materialized
+The YinYang Format core defines the filesystem values, their OpenDAL
+persistence, and the publication state machine. It owns the materialized
`Tree`, immutable `FsVersion` values, durable commit identities, and the mutable
-head. A storage implementation supplies immutable blobs and conditional head
-replacement. The core defines no wire representation.
+head. `Fs` operates directly on one `opendal::Operator`; there is no storage
+provider abstraction inside YinYang.
The current implementation stores complete materialized versions and retains
all commits.
@@ -26,9 +26,8 @@ non-overlapping, and exactly cover
the content identified by its `BlobRef`.
`BlobRef` is an opaque persistent reference plus the identity of its referenced
-bytes. The storage implementation constructs and verifies it. The core does
-not interpret object keys, byte-range syntax, framing, or integrity
-metadata inside the reference.
+bytes. YinYang persistence constructs and verifies it. Filesystem values do not
+interpret object keys or integrity metadata inside the reference.
Paths use normalized portable components. They are relative, NFC-normalized,
at most 4096 bytes, and contain components of at most 255 bytes. Empty,
@@ -59,21 +58,81 @@ caller-assigned `CommitId` is its published version number.
The head contains the current version's `BlobRef`. Every version reached through
one head history uses the same root `NodeId`.
-## Storage contract
-
-A `Storage` implementation provides five operations:
-
-1. write an immutable `FsVersion` and return its `BlobRef`;
-2. read and verify an `FsVersion` through its `BlobRef`;
-3. create the initial head;
-4. observe the head together with an opaque condition bound to that exact read;
-5. replace the head when the observed condition remains current.
-
-Head creation is idempotent and never replaces an existing value.
-Compare-exchange reports whether it replaced the observed head. An observed
-condition is never reconstructed through a later metadata request. The storage
-implementation rejects a missing, truncated, or unverifiable blob as corrupt
-data.
+## OpenDAL contract
+
+The supplied OpenDAL operator is rooted at one YinYang filesystem. It must
+support read, streaming write, create-if-absent, and ETag if-match writes.
+Creation and open fail with `Unsupported` when any required capability is
+absent. An ETag used to replace the head comes from the same read that returned
+the head bytes; it is never reconstructed with a later metadata request.
+
+YinYang owns these paths below the operator root:
+
+```text
+.yinyang/head
+.yinyang/versions/<object-id>
+```
+
+`<object-id>` is a generated UUID encoded as 32 lowercase hexadecimal
+characters and selected before encoding starts. A version writer streams the
+encoded object to that path while computing its BLAKE3 digest and length. The
+resulting version `BlobRef` contains the object path and a `ContentId` covering
+the complete encoded object. Version writes use create-if-absent. Every later
+read verifies the stored length, digest, and encoding against that reference.
+
+The head is created with create-if-absent and replaced with ETag if-match. A
+condition mismatch is a publication conflict. Missing, truncated,
+length-mismatched, or unverifiable referenced objects are corrupt data. Direct
+external writes under `.yinyang/` are outside the format contract.
+
+## Persistent encoding
+
+The payload uses [Borsh encoding](https://borsh.io/). Integers use little-endian
+fixed-width encoding. Collection, string, and byte-sequence lengths are `u32`;
+enum discriminants are `u8`; booleans use one byte, with 0 for false and 1 for
+true.
+Strings contain UTF-8 bytes. Fixed byte arrays contain their bytes without a
+length. Fields appear in the order shown below, and readers reject unknown
+discriminants, invalid booleans, truncated values, and trailing bytes.
+
+```text
+VersionBody {
+ entries: [Entry],
+ commits: [[u8; 16]],
+}
+
+Entry { path: string, node: Node }
+
+Node {
+ id: [u8; 16],
+ generation: u64,
+ executable: bool,
+ body: Dir { entries_generation: u64 } | File(File),
+}
+
+File { content: ContentId, parts: [FilePart] }
+
+FilePart {
+ start: u64,
+ end: u64,
+ blob_offset: u64,
+ blob: BlobRef,
+}
+
+BlobRef { reference: [u8], content: ContentId }
+ContentId { digest: [u8; 32], length: u64 }
+```
+
+Entries are encoded in ascending canonical `Path` order. File parts and commits
+retain their logical order. `Dir` has discriminant 0 and `File` has
+discriminant 1. A version object is `YYVER001 || borsh(VersionBody)`. The
+current materialized reader and writer impose a 64 MiB resource limit including
+the magic; this is an implementation limit, not part of the persistent format.
+An otherwise valid version above that limit is unsupported by this
+implementation, not corrupt data. The head is
+`YYHEAD01 || borsh(BlobRef) || BLAKE3(magic || payload)` and is limited to 4
+KiB. The head checksum covers its magic and payload, excluding the checksum
+itself.
## Lifecycle