diff --git a/Cargo.lock b/Cargo.lock
index 9cf1a6d5..29312a4d 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -236,6 +236,43 @@ version = "1.5.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53"
+[[package]]
+name = "aws-config"
+version = "1.8.13"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c456581cb3c77fafcc8c67204a70680d40b61112d6da78c77bd31d945b65f1b5"
+dependencies = [
+ "aws-credential-types",
+ "aws-runtime",
+ "aws-sdk-sts",
+ "aws-smithy-async",
+ "aws-smithy-http",
+ "aws-smithy-json",
+ "aws-smithy-runtime",
+ "aws-smithy-runtime-api",
+ "aws-smithy-types",
+ "aws-types",
+ "bytes",
+ "fastrand",
+ "http 1.4.0",
+ "time",
+ "tokio",
+ "tracing",
+ "url",
+]
+
+[[package]]
+name = "aws-credential-types"
+version = "1.2.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3cd362783681b15d136480ad555a099e82ecd8e2d10a841e14dfd0078d67fee3"
+dependencies = [
+ "aws-smithy-async",
+ "aws-smithy-runtime-api",
+ "aws-smithy-types",
+ "zeroize",
+]
+
[[package]]
name = "aws-lc-rs"
version = "1.17.0"
@@ -259,6 +296,324 @@ dependencies = [
"fs_extra",
]
+[[package]]
+name = "aws-runtime"
+version = "1.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c635c2dc792cb4a11ce1a4f392a925340d1bdf499289b5ec1ec6810954eb43f5"
+dependencies = [
+ "aws-credential-types",
+ "aws-sigv4",
+ "aws-smithy-async",
+ "aws-smithy-eventstream",
+ "aws-smithy-http",
+ "aws-smithy-runtime",
+ "aws-smithy-runtime-api",
+ "aws-smithy-types",
+ "aws-types",
+ "bytes",
+ "fastrand",
+ "http 0.2.12",
+ "http 1.4.0",
+ "http-body 0.4.6",
+ "http-body 1.0.1",
+ "percent-encoding",
+ "pin-project-lite",
+ "tracing",
+ "uuid",
+]
+
+[[package]]
+name = "aws-sdk-s3"
+version = "1.122.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "94c2ca0cba97e8e279eb6c0b2d0aa10db5959000e602ab2b7c02de6b85d4c19b"
+dependencies = [
+ "aws-credential-types",
+ "aws-runtime",
+ "aws-sigv4",
+ "aws-smithy-async",
+ "aws-smithy-checksums",
+ "aws-smithy-eventstream",
+ "aws-smithy-http",
+ "aws-smithy-json",
+ "aws-smithy-observability",
+ "aws-smithy-runtime",
+ "aws-smithy-runtime-api",
+ "aws-smithy-types",
+ "aws-smithy-xml",
+ "aws-types",
+ "bytes",
+ "fastrand",
+ "hex",
+ "hmac 0.12.1",
+ "http 0.2.12",
+ "http 1.4.0",
+ "http-body 1.0.1",
+ "lru",
+ "percent-encoding",
+ "regex-lite",
+ "sha2 0.10.9",
+ "tracing",
+ "url",
+]
+
+[[package]]
+name = "aws-sdk-sts"
+version = "1.97.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e6443ccadc777095d5ed13e21f5c364878c9f5bad4e35187a6cdbd863b0afcad"
+dependencies = [
+ "aws-credential-types",
+ "aws-runtime",
+ "aws-smithy-async",
+ "aws-smithy-http",
+ "aws-smithy-json",
+ "aws-smithy-observability",
+ "aws-smithy-query",
+ "aws-smithy-runtime",
+ "aws-smithy-runtime-api",
+ "aws-smithy-types",
+ "aws-smithy-xml",
+ "aws-types",
+ "fastrand",
+ "http 0.2.12",
+ "http 1.4.0",
+ "regex-lite",
+ "tracing",
+]
+
+[[package]]
+name = "aws-sigv4"
+version = "1.3.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "efa49f3c607b92daae0c078d48a4571f599f966dce3caee5f1ea55c4d9073f99"
+dependencies = [
+ "aws-credential-types",
+ "aws-smithy-eventstream",
+ "aws-smithy-http",
+ "aws-smithy-runtime-api",
+ "aws-smithy-types",
+ "bytes",
+ "form_urlencoded",
+ "hex",
+ "hmac 0.12.1",
+ "http 0.2.12",
+ "http 1.4.0",
+ "percent-encoding",
+ "sha2 0.10.9",
+ "time",
+ "tracing",
+]
+
+[[package]]
+name = "aws-smithy-async"
+version = "1.2.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "52eec3db979d18cb807fc1070961cc51d87d069abe9ab57917769687368a8c6c"
+dependencies = [
+ "futures-util",
+ "pin-project-lite",
+ "tokio",
+]
+
+[[package]]
+name = "aws-smithy-checksums"
+version = "0.64.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ddcf418858f9f3edd228acb8759d77394fed7531cce78d02bdda499025368439"
+dependencies = [
+ "aws-smithy-http",
+ "aws-smithy-types",
+ "bytes",
+ "crc-fast",
+ "hex",
+ "http 1.4.0",
+ "http-body 1.0.1",
+ "http-body-util",
+ "md-5 0.10.6",
+ "pin-project-lite",
+ "sha1 0.10.6",
+ "sha2 0.10.9",
+ "tracing",
+]
+
+[[package]]
+name = "aws-smithy-eventstream"
+version = "0.60.18"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "35b9c7354a3b13c66f60fe4616d6d1969c9fd36b1b5333a5dfb3ee716b33c588"
+dependencies = [
+ "aws-smithy-types",
+ "bytes",
+ "crc32fast",
+]
+
+[[package]]
+name = "aws-smithy-http"
+version = "0.63.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "630e67f2a31094ffa51b210ae030855cb8f3b7ee1329bdd8d085aaf61e8b97fc"
+dependencies = [
+ "aws-smithy-eventstream",
+ "aws-smithy-runtime-api",
+ "aws-smithy-types",
+ "bytes",
+ "bytes-utils",
+ "futures-core",
+ "futures-util",
+ "http 1.4.0",
+ "http-body 1.0.1",
+ "http-body-util",
+ "percent-encoding",
+ "pin-project-lite",
+ "pin-utils",
+ "tracing",
+]
+
+[[package]]
+name = "aws-smithy-http-client"
+version = "1.1.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "12fb0abf49ff0cab20fd31ac1215ed7ce0ea92286ba09e2854b42ba5cabe7525"
+dependencies = [
+ "aws-smithy-async",
+ "aws-smithy-runtime-api",
+ "aws-smithy-types",
+ "h2",
+ "http 1.4.0",
+ "hyper",
+ "hyper-rustls",
+ "hyper-util",
+ "pin-project-lite",
+ "rustls",
+ "rustls-native-certs",
+ "rustls-pki-types",
+ "tokio",
+ "tokio-rustls",
+ "tower",
+ "tracing",
+]
+
+[[package]]
+name = "aws-smithy-json"
+version = "0.62.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3cb96aa208d62ee94104645f7b2ecaf77bf27edf161590b6224bfbac2832f979"
+dependencies = [
+ "aws-smithy-types",
+]
+
+[[package]]
+name = "aws-smithy-observability"
+version = "0.2.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c0a46543fbc94621080b3cf553eb4cbbdc41dd9780a30c4756400f0139440a1d"
+dependencies = [
+ "aws-smithy-runtime-api",
+]
+
+[[package]]
+name = "aws-smithy-query"
+version = "0.60.13"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0cebbddb6f3a5bd81553643e9c7daf3cc3dc5b0b5f398ac668630e8a84e6fff0"
+dependencies = [
+ "aws-smithy-types",
+ "urlencoding",
+]
+
+[[package]]
+name = "aws-smithy-runtime"
+version = "1.10.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f3df87c14f0127a0d77eb261c3bc45d5b4833e2a1f63583ebfb728e4852134ee"
+dependencies = [
+ "aws-smithy-async",
+ "aws-smithy-http",
+ "aws-smithy-http-client",
+ "aws-smithy-observability",
+ "aws-smithy-runtime-api",
+ "aws-smithy-types",
+ "bytes",
+ "fastrand",
+ "http 0.2.12",
+ "http 1.4.0",
+ "http-body 0.4.6",
+ "http-body 1.0.1",
+ "http-body-util",
+ "pin-project-lite",
+ "pin-utils",
+ "tokio",
+ "tracing",
+]
+
+[[package]]
+name = "aws-smithy-runtime-api"
+version = "1.11.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "49952c52f7eebb72ce2a754d3866cc0f87b97d2a46146b79f80f3a93fb2b3716"
+dependencies = [
+ "aws-smithy-async",
+ "aws-smithy-types",
+ "bytes",
+ "http 0.2.12",
+ "http 1.4.0",
+ "pin-project-lite",
+ "tokio",
+ "tracing",
+ "zeroize",
+]
+
+[[package]]
+name = "aws-smithy-types"
+version = "1.4.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3b3a26048eeab0ddeba4b4f9d51654c79af8c3b32357dc5f336cee85ab331c33"
+dependencies = [
+ "base64-simd",
+ "bytes",
+ "bytes-utils",
+ "futures-core",
+ "http 0.2.12",
+ "http 1.4.0",
+ "http-body 0.4.6",
+ "http-body 1.0.1",
+ "http-body-util",
+ "itoa",
+ "num-integer",
+ "pin-project-lite",
+ "pin-utils",
+ "ryu",
+ "serde",
+ "time",
+ "tokio",
+ "tokio-util",
+]
+
+[[package]]
+name = "aws-smithy-xml"
+version = "0.60.13"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "11b2f670422ff42bf7065031e72b45bc52a3508bd089f743ea90731ca2b6ea57"
+dependencies = [
+ "xmlparser",
+]
+
+[[package]]
+name = "aws-types"
+version = "1.3.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1d980627d2dd7bfc32a3c025685a033eeab8d365cc840c631ef59d1b8f428164"
+dependencies = [
+ "aws-credential-types",
+ "aws-smithy-async",
+ "aws-smithy-runtime-api",
+ "aws-smithy-types",
+ "rustc_version",
+ "tracing",
+]
+
[[package]]
name = "axum"
version = "0.8.9"
@@ -270,8 +625,8 @@ dependencies = [
"bytes",
"form_urlencoded",
"futures-util",
- "http",
- "http-body",
+ "http 1.4.0",
+ "http-body 1.0.1",
"http-body-util",
"hyper",
"hyper-util",
@@ -301,8 +656,8 @@ checksum = "08c78f31d7b1291f7ee735c1c6780ccde7785daae9a9206026862dab7d8792d1"
dependencies = [
"bytes",
"futures-core",
- "http",
- "http-body",
+ "http 1.4.0",
+ "http-body 1.0.1",
"http-body-util",
"mime",
"pin-project-lite",
@@ -333,8 +688,8 @@ dependencies = [
"bytes",
"either",
"fs-err",
- "http",
- "http-body",
+ "http 1.4.0",
+ "http-body 1.0.1",
"hyper",
"hyper-util",
"pin-project-lite",
@@ -357,6 +712,16 @@ version = "0.22.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6"
+[[package]]
+name = "base64-simd"
+version = "0.8.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "339abbe78e73178762e23bea9dfd08e697eb3f3301cd4be981c0f78ba5859195"
+dependencies = [
+ "outref",
+ "vsimd",
+]
+
[[package]]
name = "base64ct"
version = "1.8.3"
@@ -504,6 +869,16 @@ version = "1.11.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33"
+[[package]]
+name = "bytes-utils"
+version = "0.1.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7dafe3a8757b027e2be6e4e5601ed563c55989fcf1546e933c66c8eb3a058d35"
+dependencies = [
+ "bytes",
+ "either",
+]
+
[[package]]
name = "cc"
version = "1.2.62"
@@ -710,6 +1085,16 @@ dependencies = [
"libc",
]
+[[package]]
+name = "core-foundation"
+version = "0.10.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b2a6cd9ae233e7f62ba4e9353e81a88df7fc8a5987b8d445b4d90c879bd156f6"
+dependencies = [
+ "core-foundation-sys",
+ "libc",
+]
+
[[package]]
name = "core-foundation-sys"
version = "0.8.7"
@@ -736,9 +1121,9 @@ dependencies = [
[[package]]
name = "crc"
-version = "3.4.0"
+version = "3.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5eb8a2a1cd12ab0d987a5d5e825195d372001a4094a0376319d5a0ad71c1ba0d"
+checksum = "9710d3b3739c2e349eb44fe848ad0b7c8cb1e42bd87ee49371df2f7acaf3e675"
dependencies = [
"crc-catalog",
]
@@ -749,6 +1134,18 @@ version = "2.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "217698eaf96b4a3f0bc4f3662aaa55bdf913cd54d7204591faa790070c6d0853"
+[[package]]
+name = "crc-fast"
+version = "1.9.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2fd92aca2c6001b1bf5ba0ff84ee74ec8501b52bbef0cac80bf25a6c1d87a83d"
+dependencies = [
+ "crc",
+ "digest 0.10.7",
+ "rustversion",
+ "spin 0.10.1",
+]
+
[[package]]
name = "crc32fast"
version = "1.5.0"
@@ -1165,6 +1562,25 @@ dependencies = [
"zeroize",
]
+[[package]]
+name = "extenddb-backup-store"
+version = "0.1.11"
+dependencies = [
+ "aws-config",
+ "aws-sdk-s3",
+ "bytes",
+ "extenddb-backup-store",
+ "futures",
+ "serde",
+ "tempfile",
+ "thiserror",
+ "tokio",
+ "tokio-util",
+ "toml",
+ "tracing",
+ "uuid",
+]
+
[[package]]
name = "extenddb-cache"
version = "0.1.11"
@@ -1395,7 +1811,7 @@ checksum = "da0e4dd2a88388a1f4ccc7c9ce104604dab68d9f408dc34cd45823d5a9069095"
dependencies = [
"futures-core",
"futures-sink",
- "spin",
+ "spin 0.9.8",
]
[[package]]
@@ -1410,6 +1826,12 @@ version = "0.1.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2"
+[[package]]
+name = "foldhash"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "77ce24cb58228fbb8aa041425bb1050850ac19177686ea6e0f41a70416f56fdb"
+
[[package]]
name = "form_urlencoded"
version = "1.2.2"
@@ -1612,7 +2034,7 @@ dependencies = [
"fnv",
"futures-core",
"futures-sink",
- "http",
+ "http 1.4.0",
"indexmap",
"slab",
"tokio",
@@ -1638,7 +2060,18 @@ checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1"
dependencies = [
"allocator-api2",
"equivalent",
- "foldhash",
+ "foldhash 0.1.5",
+]
+
+[[package]]
+name = "hashbrown"
+version = "0.16.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100"
+dependencies = [
+ "allocator-api2",
+ "equivalent",
+ "foldhash 0.2.0",
]
[[package]]
@@ -1783,6 +2216,17 @@ dependencies = [
"windows-sys 0.61.2",
]
+[[package]]
+name = "http"
+version = "0.2.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "601cbb57e577e2f5ef5be8e7b83f0f63994f25aa94d673e54a92d5c516d101f1"
+dependencies = [
+ "bytes",
+ "fnv",
+ "itoa",
+]
+
[[package]]
name = "http"
version = "1.4.0"
@@ -1793,6 +2237,17 @@ dependencies = [
"itoa",
]
+[[package]]
+name = "http-body"
+version = "0.4.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7ceab25649e9960c0311ea418d17bee82c0dcec1bd053b5f9a66e265a693bed2"
+dependencies = [
+ "bytes",
+ "http 0.2.12",
+ "pin-project-lite",
+]
+
[[package]]
name = "http-body"
version = "1.0.1"
@@ -1800,7 +2255,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1efedce1fb8e6913f23e0c92de8e62cd5b772a67e7b3946df930a62566c93184"
dependencies = [
"bytes",
- "http",
+ "http 1.4.0",
]
[[package]]
@@ -1811,8 +2266,8 @@ checksum = "b021d93e26becf5dc7e1b75b1bed1fd93124b374ceb73f43d4d4eafec896a64a"
dependencies = [
"bytes",
"futures-core",
- "http",
- "http-body",
+ "http 1.4.0",
+ "http-body 1.0.1",
"pin-project-lite",
]
@@ -1848,14 +2303,31 @@ dependencies = [
"futures-channel",
"futures-core",
"h2",
- "http",
- "http-body",
+ "http 1.4.0",
+ "http-body 1.0.1",
"httparse",
"httpdate",
"itoa",
"pin-project-lite",
"smallvec",
"tokio",
+ "want",
+]
+
+[[package]]
+name = "hyper-rustls"
+version = "0.27.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "33ca68d021ef39cf6463ab54c1d0f5daf03377b70561305bb89a8f83aab66e0f"
+dependencies = [
+ "http 1.4.0",
+ "hyper",
+ "hyper-util",
+ "rustls",
+ "rustls-native-certs",
+ "tokio",
+ "tokio-rustls",
+ "tower-service",
]
[[package]]
@@ -1864,13 +2336,21 @@ version = "0.1.20"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "96547c2556ec9d12fb1578c4eaf448b04993e7fb79cbaad930a656880a6bdfa0"
dependencies = [
+ "base64 0.22.1",
"bytes",
- "http",
- "http-body",
+ "futures-channel",
+ "futures-util",
+ "http 1.4.0",
+ "http-body 1.0.1",
"hyper",
+ "ipnet",
+ "libc",
+ "percent-encoding",
"pin-project-lite",
+ "socket2",
"tokio",
"tower-service",
+ "tracing",
]
[[package]]
@@ -2131,7 +2611,7 @@ version = "1.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe"
dependencies = [
- "spin",
+ "spin 0.9.8",
]
[[package]]
@@ -2202,6 +2682,15 @@ version = "0.4.29"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897"
+[[package]]
+name = "lru"
+version = "0.16.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7f66e8d5d03f609abc3a39e6f08e4164ebf1447a732906d39eb9b99b7919ef39"
+dependencies = [
+ "hashbrown 0.16.1",
+]
+
[[package]]
name = "macro_magic"
version = "0.5.1"
@@ -2548,6 +3037,12 @@ version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381"
+[[package]]
+name = "openssl-probe"
+version = "0.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7c87def4c32ab89d880effc9e097653c8da5d6ef28e6b539d313baaacfbafcbe"
+
[[package]]
name = "ordered-multimap"
version = "0.7.3"
@@ -2558,6 +3053,12 @@ dependencies = [
"hashbrown 0.14.5",
]
+[[package]]
+name = "outref"
+version = "0.5.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1a80800c0488c3a21695ea981a54918fbb37abf04f4d0720c453632255e2ff0e"
+
[[package]]
name = "parking"
version = "2.2.1"
@@ -2685,6 +3186,12 @@ version = "0.2.17"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd"
+[[package]]
+name = "pin-utils"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184"
+
[[package]]
name = "pkcs1"
version = "0.7.5"
@@ -2979,6 +3486,12 @@ dependencies = [
"regex-syntax",
]
+[[package]]
+name = "regex-lite"
+version = "0.1.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cab834c73d247e67f4fae452806d17d3c7501756d98c8808d7c9c7aa7d18f973"
+
[[package]]
name = "regex-syntax"
version = "0.8.10"
@@ -3104,6 +3617,18 @@ dependencies = [
"zeroize",
]
+[[package]]
+name = "rustls-native-certs"
+version = "0.8.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "dab5152771c58876a2146916e53e35057e1a4dfa2b9df0f0305b07f611fdea4d"
+dependencies = [
+ "openssl-probe",
+ "rustls-pki-types",
+ "schannel",
+ "security-framework",
+]
+
[[package]]
name = "rustls-pki-types"
version = "1.14.1"
@@ -3158,12 +3683,44 @@ dependencies = [
"winapi-util",
]
+[[package]]
+name = "schannel"
+version = "0.1.29"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "91c1b7e4904c873ef0710c1f407dde2e6287de2bebc1bbbf7d430bb7cbffd939"
+dependencies = [
+ "windows-sys 0.61.2",
+]
+
[[package]]
name = "scopeguard"
version = "1.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49"
+[[package]]
+name = "security-framework"
+version = "3.7.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b7f4bc775c73d9a02cde8bf7b2ec4c9d12743edf609006c7facc23998404cd1d"
+dependencies = [
+ "bitflags",
+ "core-foundation 0.10.1",
+ "core-foundation-sys",
+ "libc",
+ "security-framework-sys",
+]
+
+[[package]]
+name = "security-framework-sys"
+version = "2.17.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6ce2691df843ecc5d231c0b14ece2acc3efb62c0a398c7e1d875f3983ce020e3"
+dependencies = [
+ "core-foundation-sys",
+ "libc",
+]
+
[[package]]
name = "semver"
version = "1.0.28"
@@ -3413,6 +3970,12 @@ dependencies = [
"lock_api",
]
+[[package]]
+name = "spin"
+version = "0.10.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "023a211cb3138dbc438680b32560ad89f699977624c9f8dbb95a47d5b4c07dd3"
+
[[package]]
name = "spki"
version = "0.7.3"
@@ -3700,7 +4263,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a13f3d0daba03132c0aa9767f98351b3488edc2c100cda2d2ec2b04f3d8d3c8b"
dependencies = [
"bitflags",
- "core-foundation",
+ "core-foundation 0.9.4",
"system-configuration-sys",
]
@@ -3970,8 +4533,8 @@ dependencies = [
"bitflags",
"bytes",
"futures-core",
- "http",
- "http-body",
+ "http 1.4.0",
+ "http-body 1.0.1",
"pin-project-lite",
"tokio",
"tokio-util",
@@ -4066,6 +4629,12 @@ dependencies = [
"tracing-serde",
]
+[[package]]
+name = "try-lock"
+version = "0.2.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b"
+
[[package]]
name = "typed-builder"
version = "0.22.0"
@@ -4225,6 +4794,12 @@ version = "0.9.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a"
+[[package]]
+name = "vsimd"
+version = "0.8.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5c3082ca00d5a5ef149bb8b555a72ae84c9c59f7250f013ac822ac2e49b19c64"
+
[[package]]
name = "wait-timeout"
version = "0.2.1"
@@ -4244,6 +4819,15 @@ dependencies = [
"winapi-util",
]
+[[package]]
+name = "want"
+version = "0.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bfa7760aed19e106de2c7c0b581b509f2f25d3dacaf737cb82ac61bc6d760b0e"
+dependencies = [
+ "try-lock",
+]
+
[[package]]
name = "wasi"
version = "0.11.1+wasi-snapshot-preview1"
@@ -4393,7 +4977,7 @@ version = "0.1.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22"
dependencies = [
- "windows-sys 0.61.2",
+ "windows-sys 0.48.0",
]
[[package]]
@@ -4715,6 +5299,12 @@ dependencies = [
"time",
]
+[[package]]
+name = "xmlparser"
+version = "0.13.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "66fee0b777b0f5ac1c69bb06d361268faafa61cd4682ae064a171c16c433e9e4"
+
[[package]]
name = "yaml-rust2"
version = "0.8.1"
diff --git a/Cargo.toml b/Cargo.toml
index 44ebd9fa..3fe61045 100755
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -5,6 +5,7 @@ resolver = "2"
members = [
"crates/core",
"crates/cache",
+ "crates/backup-store",
"crates/engine",
"crates/storage",
"crates/config",
@@ -27,6 +28,7 @@ license = "Apache-2.0"
# Internal crates
extenddb-core = { path = "crates/core" }
extenddb-cache = { path = "crates/cache" }
+extenddb-backup-store = { path = "crates/backup-store" }
extenddb-engine = { path = "crates/engine" }
extenddb-storage = { path = "crates/storage" }
extenddb-config = { path = "crates/config" }
@@ -72,6 +74,16 @@ mongodb = "3"
bson = "2.13"
dashmap = "6"
+# Backup store
+# Exact pins: the aws-sdk crates release in lockstep, and these are the newest
+# versions whose declared minimum supported rustc matches the workspace
+# rust-version (1.88); later releases require a newer compiler. Their
+# transitive aws-smithy-* crates are held in Cargo.lock to matching versions
+# for the same reason. Feature set mirrors tests/rust/Cargo.toml.
+aws-config = { version = "=1.8.13", default-features = false, features = ["behavior-version-latest", "rt-tokio", "default-https-client", "credentials-process"] }
+aws-sdk-s3 = { version = "=1.122.0", default-features = false, features = ["behavior-version-latest", "rt-tokio", "default-https-client"] }
+bytes = "1"
+
# Crypto & checksums
crc32fast = "1"
uuid = { version = "1", features = ["v4"] }
diff --git a/SOFTWARE-LICENSE-NOTICES-DEV.html b/SOFTWARE-LICENSE-NOTICES-DEV.html
index e18c5fc3..198643aa 100644
--- a/SOFTWARE-LICENSE-NOTICES-DEV.html
+++ b/SOFTWARE-LICENSE-NOTICES-DEV.html
@@ -1765,7 +1765,7 @@ Used by:
Apache License 2.0
Used by:
Apache License
Version 2.0 January 2004
diff --git a/SOFTWARE-LICENSE-NOTICES-MONGODB.html b/SOFTWARE-LICENSE-NOTICES-MONGODB.html
index 57d9567e..6ac65297 100644
--- a/SOFTWARE-LICENSE-NOTICES-MONGODB.html
+++ b/SOFTWARE-LICENSE-NOTICES-MONGODB.html
@@ -2191,7 +2191,7 @@ Used by:
Apache License 2.0
Used by:
Apache License
Version 2.0 January 2004
diff --git a/SOFTWARE-LICENSE-NOTICES.html b/SOFTWARE-LICENSE-NOTICES.html
index 2e6a38e4..84e55533 100644
--- a/SOFTWARE-LICENSE-NOTICES.html
+++ b/SOFTWARE-LICENSE-NOTICES.html
@@ -1766,7 +1766,7 @@ Used by:
Apache License 2.0
Used by:
Apache License
Version 2.0 January 2004
diff --git a/crates/backup-store/Cargo.toml b/crates/backup-store/Cargo.toml
new file mode 100644
index 00000000..bb07873e
--- /dev/null
+++ b/crates/backup-store/Cargo.toml
@@ -0,0 +1,31 @@
+# Copyright 2026 ExtendDB contributors
+# SPDX-License-Identifier: Apache-2.0
+[package]
+name = "extenddb-backup-store"
+version.workspace = true
+edition.workspace = true
+rust-version.workspace = true
+license.workspace = true
+description = "Backup object store abstraction with filesystem and S3 implementations."
+
+[dependencies]
+aws-config = { workspace = true }
+aws-sdk-s3 = { workspace = true }
+bytes = { workspace = true }
+futures = { workspace = true }
+serde = { workspace = true }
+thiserror = { workspace = true }
+tokio = { workspace = true }
+tokio-util = { workspace = true, features = ["io"] }
+tracing = { workspace = true }
+uuid = { workspace = true }
+
+[features]
+# Failure-injection seam for the crate's own test suite; not part of the
+# supported API surface and never enabled by production consumers.
+test-util = []
+
+[dev-dependencies]
+extenddb-backup-store = { path = ".", features = ["test-util"] }
+tempfile = "3"
+toml = { workspace = true }
diff --git a/crates/backup-store/README.md b/crates/backup-store/README.md
new file mode 100644
index 00000000..89f31ea9
--- /dev/null
+++ b/crates/backup-store/README.md
@@ -0,0 +1,55 @@
+# extenddb-backup-store
+
+Backup object store abstraction for ExtendDB: one `BackupStore` trait with a
+filesystem implementation and an S3 implementation built on the official
+`aws-sdk-s3` crate. The engine writes portable backups through this trait; see
+`docs/design/15-backup-restore.md` for the full design.
+
+## Behavior
+
+- Keys are `/`-separated component paths. Both stores reject empty
+ components, `.` and `..`, backslashes, control characters, and keys over
+ 1024 bytes, so the two stores accept the same key space.
+- The filesystem store canonicalizes its root at construction and resolves
+ every key one component at a time without following symlinks: a symlink
+ anywhere in a key's path, pointing inside or outside the root, is refused.
+ This defeats symlinks present when an operation starts; it does not defend
+ against another local process mutating the tree during an operation, so
+ the backup root must not be writable by less-trusted users (write access
+ inside the root already grants full read, write, and delete over every
+ backup). Writes go through a temporary sibling file renamed into place so
+ a crash never leaves a truncated object under its final name.
+- The S3 store takes credentials from the standard AWS SDK chain, uses
+ multipart upload for bodies larger than the configured part size (default
+ 8 MiB, minimum 5 MiB), aborts the multipart upload when a put fails,
+ paginates listings, and deletes prefixes in DeleteObjects batches of 1000.
+
+## Tests
+
+`cargo test -p extenddb-backup-store` runs the key validation and
+configuration unit tests, the filesystem conformance suite against a
+temporary directory, and the S3 conformance suite.
+
+The S3 suite needs a local S3-compatible endpoint and is skipped with a
+printed reason when `EXTENDDB_TEST_S3_ENDPOINT` is unset. To run it against a
+local MinIO:
+
+```sh
+docker run --rm -d -p 9000:9000 --name extenddb-backup-store-minio \
+ -e MINIO_ROOT_USER=minioadmin -e MINIO_ROOT_PASSWORD=minioadmin \
+ minio/minio server /data
+
+EXTENDDB_TEST_S3_ENDPOINT=http://127.0.0.1:9000 \
+AWS_ACCESS_KEY_ID=minioadmin \
+AWS_SECRET_ACCESS_KEY=minioadmin \
+cargo test -p extenddb-backup-store
+```
+
+A bare `minio server /data-dir` binary works the same way; only the endpoint
+and the two credential variables matter.
+
+`EXTENDDB_TEST_S3_BUCKET` (default `extenddb-backup-store-tests`) and
+`EXTENDDB_TEST_S3_REGION` (default `us-east-1`) can override the bucket and
+region. The suite creates the bucket when it does not exist and isolates each
+test under a unique key prefix. CI wiring for the MinIO service is tracked as
+its own task in the design doc.
diff --git a/crates/backup-store/src/config.rs b/crates/backup-store/src/config.rs
new file mode 100644
index 00000000..bbf37353
--- /dev/null
+++ b/crates/backup-store/src/config.rs
@@ -0,0 +1,271 @@
+// Copyright 2026 ExtendDB contributors
+// SPDX-License-Identifier: Apache-2.0
+
+//! Configuration shapes for the backup store.
+//!
+//! These model the store subset of the `[backup]` section:
+//!
+//! ```toml
+//! [backup]
+//! store = "filesystem" # "filesystem" | "s3"
+//!
+//! [backup.filesystem]
+//! path = "/var/lib/extenddb/backups" # required when store = "filesystem"
+//!
+//! [backup.s3]
+//! bucket = "my-extenddb-backups" # required when store = "s3"
+//! prefix = "extenddb/" # optional
+//! region = "us-east-1" # optional; SDK chain otherwise
+//! endpoint = "http://minio:9000" # optional; S3-compatible stores
+//! force_path_style = false
+//! upload_part_size_bytes = 8388608
+//! max_concurrent_uploads = 4
+//! ```
+//!
+//! Wiring into `crates/config` and the server startup path is deliberately
+//! not done here; this crate only defines the shapes and the
+//! [`crate::open`] constructor over them.
+
+use std::path::PathBuf;
+
+use serde::Deserialize;
+
+/// Which store implementation the `[backup]` section selects.
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum BackupStoreKind {
+ /// Local directory store. The default: local development works with no
+ /// configuration, and the backup manual states the failure-domain
+ /// consequence.
+ #[default]
+ Filesystem,
+ /// S3 bucket store.
+ S3,
+}
+
+/// The store subset of the `[backup]` configuration section.
+///
+/// Composition note for the wiring task: `deny_unknown_fields` does not
+/// survive `serde(flatten)`, so embedding this struct into a full `[backup]`
+/// section model via flatten would silently stop rejecting unknown fields.
+/// Model the full section as its own struct with these fields inline instead
+/// of composing by flattening.
+#[derive(Debug, Clone, Default, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct BackupStoreConfig {
+ /// Selected store kind; `filesystem` when absent.
+ #[serde(default)]
+ pub store: BackupStoreKind,
+ /// `[backup.filesystem]`, required when `store = "filesystem"`.
+ #[serde(default)]
+ pub filesystem: Option,
+ /// `[backup.s3]`, required when `store = "s3"`.
+ #[serde(default)]
+ pub s3: Option,
+}
+
+/// `[backup.filesystem]` settings.
+#[derive(Debug, Clone, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct FilesystemStoreConfig {
+ /// Root directory for backup objects. Must exist at startup.
+ pub path: PathBuf,
+}
+
+/// `[backup.s3]` settings. Credentials are absent by design: they come from
+/// the standard AWS SDK chain.
+#[derive(Debug, Clone, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct S3StoreConfig {
+ /// Destination bucket.
+ pub bucket: String,
+ /// Key prefix inside the bucket. Optional; empty means the bucket root.
+ #[serde(default)]
+ pub prefix: String,
+ /// Region override. When absent the SDK chain resolves the region.
+ #[serde(default)]
+ pub region: Option,
+ /// Endpoint override for S3-compatible stores.
+ #[serde(default)]
+ pub endpoint: Option,
+ /// Use path-style addressing; required by most S3-compatible stores.
+ #[serde(default)]
+ pub force_path_style: bool,
+ /// Multipart part size in bytes. Defaults to 8 MiB; the enforced minimum
+ /// is 5 MiB, the smallest part S3 accepts.
+ #[serde(default = "default_upload_part_size_bytes")]
+ pub upload_part_size_bytes: u64,
+ /// Maximum concurrently in-flight part uploads per put. Defaults to 4.
+ #[serde(default = "default_max_concurrent_uploads")]
+ pub max_concurrent_uploads: usize,
+}
+
+fn default_upload_part_size_bytes() -> u64 {
+ 8 * 1024 * 1024
+}
+
+fn default_max_concurrent_uploads() -> usize {
+ 4
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ /// `Arc` has no `Debug`, so `unwrap_err` cannot be used.
+ async fn open_err(config: &BackupStoreConfig) -> crate::StoreError {
+ match crate::open(config).await {
+ Err(e) => e,
+ Ok(_) => panic!("expected open to fail"),
+ }
+ }
+
+ #[test]
+ fn parses_the_documented_filesystem_shape() {
+ let config: BackupStoreConfig = toml::from_str(
+ r#"
+ store = "filesystem"
+
+ [filesystem]
+ path = "/var/lib/extenddb/backups"
+ "#,
+ )
+ .unwrap();
+ assert_eq!(config.store, BackupStoreKind::Filesystem);
+ assert_eq!(
+ config.filesystem.unwrap().path,
+ PathBuf::from("/var/lib/extenddb/backups")
+ );
+ assert!(config.s3.is_none());
+ }
+
+ #[test]
+ fn parses_the_documented_s3_shape_with_defaults() {
+ let config: BackupStoreConfig = toml::from_str(
+ r#"
+ store = "s3"
+
+ [s3]
+ bucket = "my-extenddb-backups"
+ prefix = "extenddb/"
+ region = "us-east-1"
+ endpoint = "http://minio:9000"
+ force_path_style = true
+ "#,
+ )
+ .unwrap();
+ assert_eq!(config.store, BackupStoreKind::S3);
+ let s3 = config.s3.unwrap();
+ assert_eq!(s3.bucket, "my-extenddb-backups");
+ assert_eq!(s3.prefix, "extenddb/");
+ assert_eq!(s3.region.as_deref(), Some("us-east-1"));
+ assert_eq!(s3.endpoint.as_deref(), Some("http://minio:9000"));
+ assert!(s3.force_path_style);
+ assert_eq!(s3.upload_part_size_bytes, 8 * 1024 * 1024);
+ assert_eq!(s3.max_concurrent_uploads, 4);
+ }
+
+ #[test]
+ fn store_kind_defaults_to_filesystem() {
+ let config: BackupStoreConfig = toml::from_str(
+ r#"
+ [filesystem]
+ path = "/backups"
+ "#,
+ )
+ .unwrap();
+ assert_eq!(config.store, BackupStoreKind::Filesystem);
+ }
+
+ #[test]
+ fn unknown_fields_are_rejected_at_every_level() {
+ assert!(toml::from_str::("stire = \"s3\"").is_err());
+ assert!(
+ toml::from_str::(
+ "store = \"filesystem\"\n[filesystem]\npath = \"/b\"\npth = \"/b\"\n"
+ )
+ .is_err()
+ );
+ assert!(
+ toml::from_str::(
+ "store = \"s3\"\n[s3]\nbucket = \"b\"\nbukcet = \"b\"\n"
+ )
+ .is_err()
+ );
+ }
+
+ #[test]
+ fn unknown_store_kind_is_rejected() {
+ assert!(toml::from_str::("store = \"gcs\"").is_err());
+ }
+
+ #[tokio::test]
+ async fn open_requires_the_matching_section() {
+ let missing_fs = BackupStoreConfig {
+ store: BackupStoreKind::Filesystem,
+ filesystem: None,
+ s3: None,
+ };
+ let err = open_err(&missing_fs).await;
+ assert!(err.to_string().contains("[backup.filesystem]"), "{err}");
+
+ let missing_s3 = BackupStoreConfig {
+ store: BackupStoreKind::S3,
+ filesystem: None,
+ s3: None,
+ };
+ let err = open_err(&missing_s3).await;
+ assert!(err.to_string().contains("[backup.s3]"), "{err}");
+ }
+
+ #[tokio::test]
+ async fn open_enforces_the_part_size_floor() {
+ let config = BackupStoreConfig {
+ store: BackupStoreKind::S3,
+ filesystem: None,
+ s3: Some(S3StoreConfig {
+ bucket: "b".to_owned(),
+ prefix: String::new(),
+ region: Some("us-east-1".to_owned()),
+ endpoint: None,
+ force_path_style: false,
+ upload_part_size_bytes: 4 * 1024 * 1024,
+ max_concurrent_uploads: 4,
+ }),
+ };
+ let err = open_err(&config).await;
+ assert!(err.to_string().contains("minimum is 5242880"), "{err}");
+ }
+
+ #[tokio::test]
+ async fn open_rejects_zero_concurrency() {
+ let config = BackupStoreConfig {
+ store: BackupStoreKind::S3,
+ filesystem: None,
+ s3: Some(S3StoreConfig {
+ bucket: "b".to_owned(),
+ prefix: String::new(),
+ region: Some("us-east-1".to_owned()),
+ endpoint: None,
+ force_path_style: false,
+ upload_part_size_bytes: 8 * 1024 * 1024,
+ max_concurrent_uploads: 0,
+ }),
+ };
+ let err = open_err(&config).await;
+ assert!(err.to_string().contains("max_concurrent_uploads"), "{err}");
+ }
+
+ #[tokio::test]
+ async fn open_rejects_a_missing_filesystem_root() {
+ let config = BackupStoreConfig {
+ store: BackupStoreKind::Filesystem,
+ filesystem: Some(FilesystemStoreConfig {
+ path: PathBuf::from("/nonexistent/extenddb-backup-root"),
+ }),
+ s3: None,
+ };
+ let err = open_err(&config).await;
+ assert!(err.to_string().contains("not usable"), "{err}");
+ }
+}
diff --git a/crates/backup-store/src/error.rs b/crates/backup-store/src/error.rs
new file mode 100644
index 00000000..239290b8
--- /dev/null
+++ b/crates/backup-store/src/error.rs
@@ -0,0 +1,60 @@
+// Copyright 2026 ExtendDB contributors
+// SPDX-License-Identifier: Apache-2.0
+
+//! Store error type.
+//!
+//! Display messages carry operation context, key names, and service error
+//! codes. They never carry credentials: S3 service errors are reduced to
+//! their error code and service message before formatting, and transport
+//! errors wrap the SDK error whose display contains no credential material.
+
+/// Error returned by every [`crate::BackupStore`] operation.
+#[derive(Debug, thiserror::Error)]
+pub enum StoreError {
+ /// The requested object does not exist.
+ #[error("object not found")]
+ NotFound,
+
+ /// The store refused the operation: filesystem permissions, a resolved
+ /// path escaping the configured root, or an S3 `AccessDenied`. The message
+ /// names what was refused.
+ #[error("permission denied: {0}")]
+ PermissionDenied(String),
+
+ /// The key or prefix failed validation. The message names the rule that
+ /// rejected it.
+ #[error("invalid key: {0}")]
+ InvalidKey(String),
+
+ /// A local I/O failure. `context` names the operation and path.
+ #[error("{context}: {source}")]
+ Io {
+ /// Operation and path the failure occurred on.
+ context: String,
+ /// Underlying I/O error.
+ #[source]
+ source: std::io::Error,
+ },
+
+ /// A network or protocol failure talking to a remote store.
+ #[error("transport error: {source}")]
+ Transport {
+ /// Underlying transport error.
+ #[source]
+ source: Box,
+ },
+
+ /// Any other failure. The message is self-contained.
+ #[error("{0}")]
+ Other(String),
+}
+
+impl StoreError {
+ /// Wrap an I/O error with the operation and path it occurred on.
+ pub(crate) fn io(context: impl Into, source: std::io::Error) -> Self {
+ Self::Io {
+ context: context.into(),
+ source,
+ }
+ }
+}
diff --git a/crates/backup-store/src/fs.rs b/crates/backup-store/src/fs.rs
new file mode 100644
index 00000000..dd29ca06
--- /dev/null
+++ b/crates/backup-store/src/fs.rs
@@ -0,0 +1,544 @@
+// Copyright 2026 ExtendDB contributors
+// SPDX-License-Identifier: Apache-2.0
+
+//! Filesystem-backed store.
+//!
+//! Keys map to paths under a root directory that is canonicalized at
+//! construction. Every operation resolves its key one component at a time
+//! with `symlink_metadata` and refuses any component that is a symlink,
+//! whether it points inside or outside the root, so a key never resolves
+//! through a link. These checks defeat symlinks present when an operation
+//! starts. They do not defend against another local process mutating the
+//! tree while an operation runs: between a component check and the next
+//! path step the kernel re-resolves the textual path, so a process with
+//! write access inside the root can swap a checked directory for a symlink
+//! during the operation. Closing that window requires resolving through
+//! directory handles (`openat2` with `RESOLVE_BENEATH`, or an `openat` walk
+//! with `O_NOFOLLOW` at every step), which the standard library does not expose;
+//! that is left as a follow-up. The backup root must therefore not be
+//! writable by less-trusted users: a process with write access inside the
+//! root already holds full read, write, and delete over every backup in it.
+//!
+//! Writes go to a temporary sibling file and rename into place, so a crash
+//! never leaves a truncated file under its final name.
+//!
+//! Two limits of this store narrow the shared key space. A key component
+//! longer than the filesystem's name limit (`NAME_MAX`, 255 bytes on Linux)
+//! fails with an I/O error even though key validation accepts it. An object
+//! whose file name matches the temporary-file pattern (`.put-*.tmp`) can be
+//! written and read but is hidden from listings.
+
+use std::path::{Path, PathBuf};
+
+use futures::future::BoxFuture;
+use futures::stream::BoxStream;
+use futures::{FutureExt, StreamExt, TryStreamExt};
+use tokio::io::AsyncWriteExt;
+
+use crate::key::{key_matches_prefix, normalize_prefix, validate_key, validate_prefix};
+use crate::{BackupStore, ByteStream, ObjectMeta, StoreError};
+
+/// Prefix and suffix marking in-flight temporary files, hidden from listings.
+const TEMP_PREFIX: &str = ".put-";
+const TEMP_SUFFIX: &str = ".tmp";
+
+/// Store writing objects under a local root directory.
+pub struct FilesystemStore {
+ /// Canonicalized root; every resolved target must stay under it.
+ root: PathBuf,
+}
+
+impl FilesystemStore {
+ /// Open a store rooted at `root`. The directory must already exist; it is
+ /// canonicalized here and the canonical form anchors every escape check.
+ ///
+ /// # Errors
+ ///
+ /// Returns an error when the root does not exist, cannot be resolved, or
+ /// is not a directory.
+ pub async fn open(root: impl AsRef) -> Result {
+ let root = root.as_ref();
+ let canonical = tokio::fs::canonicalize(root).await.map_err(|e| {
+ StoreError::io(format!("backup root {} is not usable", root.display()), e)
+ })?;
+ let meta = tokio::fs::metadata(&canonical)
+ .await
+ .map_err(|e| StoreError::io(format!("stat {}", canonical.display()), e))?;
+ if !meta.is_dir() {
+ return Err(StoreError::Other(format!(
+ "backup root {} is not a directory",
+ canonical.display()
+ )));
+ }
+ Ok(Self { root: canonical })
+ }
+
+ /// Resolve a validated key to its path under the root without following
+ /// symlinks: every component is checked with `symlink_metadata` and any
+ /// symlink, intermediate or final, is refused whether it points inside
+ /// or outside the root, so `get` and `head` agree with `put`, `list`,
+ /// and `delete_prefix` on what a key names. Key validation has already
+ /// refused `..`, empty components, and separators, so with no symlink in
+ /// the walk the path cannot leave the root. Missing paths surface as
+ /// `NotFound`. The returned metadata describes the final component.
+ async fn resolve_no_follow(
+ &self,
+ key: &str,
+ ) -> Result<(PathBuf, std::fs::Metadata), StoreError> {
+ let mut path = self.root.clone();
+ let mut meta = None;
+ for component in key.split('/') {
+ path.push(component);
+ let m = match tokio::fs::symlink_metadata(&path).await {
+ Ok(m) => m,
+ Err(e)
+ if e.kind() == std::io::ErrorKind::NotFound
+ || e.kind() == std::io::ErrorKind::NotADirectory =>
+ {
+ return Err(StoreError::NotFound);
+ }
+ Err(e) => return Err(StoreError::io(format!("stat {}", path.display()), e)),
+ };
+ if m.is_symlink() {
+ return Err(symlink_refused(&path));
+ }
+ meta = Some(m);
+ }
+ let meta = meta.expect("a validated key has at least one component");
+ Ok((path, meta))
+ }
+
+ /// Prepare the directory a key's object will be written into, without
+ /// ever creating a directory outside the root.
+ ///
+ /// `create_dir_all` follows symlinks in existing path components, so
+ /// checking the parent only after creating it would already have created
+ /// directories on the far side of a planted symlink. Instead: walk to the
+ /// deepest existing ancestor of the destination, refusing any component
+ /// that is a symlink (in-root aliases included), and verify the ancestor
+ /// still resolves under the root. The remaining components are then
+ /// created one at a time with `create_dir`, re-verifying after each step
+ /// that the component is a real directory and not a symlink.
+ async fn prepare_parent(&self, parent_components: &[&str]) -> Result {
+ let mut existing = self.root.clone();
+ let mut created_from = parent_components.len();
+ for (index, component) in parent_components.iter().enumerate() {
+ let candidate = existing.join(component);
+ match tokio::fs::symlink_metadata(&candidate).await {
+ Ok(meta) if meta.is_symlink() => {
+ return Err(symlink_refused(&candidate));
+ }
+ Ok(_) => existing = candidate,
+ // NotADirectory: an existing ancestor is a file, so the walk
+ // stops there and the checks below decide (an in-root file
+ // is an ancestor collision).
+ Err(e)
+ if e.kind() == std::io::ErrorKind::NotFound
+ || e.kind() == std::io::ErrorKind::NotADirectory =>
+ {
+ created_from = index;
+ break;
+ }
+ Err(e) => {
+ return Err(StoreError::io(format!("stat {}", candidate.display()), e));
+ }
+ }
+ }
+
+ // The deepest existing ancestor is checked before anything is
+ // created: no component was a symlink at the walk above, and as
+ // defense in depth its canonical form must still sit under the root.
+ let canonical = tokio::fs::canonicalize(&existing)
+ .await
+ .map_err(|e| StoreError::io(format!("resolve {}", existing.display()), e))?;
+ if !canonical.starts_with(&self.root) {
+ return Err(StoreError::PermissionDenied(format!(
+ "{} resolves outside the backup root",
+ existing.display()
+ )));
+ }
+ let meta = tokio::fs::metadata(&canonical)
+ .await
+ .map_err(|e| StoreError::io(format!("stat {}", canonical.display()), e))?;
+ if !meta.is_dir() {
+ return Err(StoreError::io(
+ format!("create under {}", canonical.display()),
+ std::io::Error::new(
+ std::io::ErrorKind::NotADirectory,
+ "an object already exists at an ancestor of the key",
+ ),
+ ));
+ }
+
+ let mut current = canonical;
+ for component in &parent_components[created_from..] {
+ let next = current.join(component);
+ match tokio::fs::create_dir(&next).await {
+ Ok(()) => {}
+ // A concurrent put may have created it; the check below
+ // verifies what is actually there.
+ Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {}
+ Err(e) => return Err(StoreError::io(format!("create {}", next.display()), e)),
+ }
+ let meta = tokio::fs::symlink_metadata(&next)
+ .await
+ .map_err(|e| StoreError::io(format!("stat {}", next.display()), e))?;
+ if meta.is_symlink() || !meta.is_dir() {
+ return Err(StoreError::PermissionDenied(format!(
+ "{} is not a directory inside the backup root",
+ next.display()
+ )));
+ }
+ current = next;
+ }
+ Ok(current)
+ }
+
+ async fn put_impl(&self, key: &str, mut body: ByteStream) -> Result<(), StoreError> {
+ validate_key(key)?;
+ let components: Vec<&str> = key.split('/').collect();
+ let (file_name, parent_components) = components
+ .split_last()
+ .expect("a validated key has at least one component");
+ let parent = self.prepare_parent(parent_components).await?;
+ let final_path = parent.join(file_name);
+
+ // A symlink at the destination, in-root or escaping, is refused: a
+ // put never resolves through a link.
+ match tokio::fs::symlink_metadata(&final_path).await {
+ Ok(meta) if meta.is_symlink() => {
+ return Err(symlink_refused(&final_path));
+ }
+ Ok(_) | Err(_) => {}
+ }
+
+ let temp_path = parent.join(format!(
+ "{TEMP_PREFIX}{}{TEMP_SUFFIX}",
+ uuid::Uuid::new_v4()
+ ));
+
+ // Durability contract: the temp file is synced before the rename and
+ // the parent directory is synced after it, so an Ok return means both
+ // the object bytes and its directory entry are on disk. Without the
+ // directory sync the rename itself could be lost on power failure
+ // even though the file data had been written.
+ let write_result = async {
+ let mut file = tokio::fs::File::create(&temp_path)
+ .await
+ .map_err(|e| StoreError::io(format!("create {}", temp_path.display()), e))?;
+ while let Some(chunk) = body.next().await {
+ let chunk = chunk?;
+ file.write_all(&chunk)
+ .await
+ .map_err(|e| StoreError::io(format!("write {}", temp_path.display()), e))?;
+ }
+ file.sync_all()
+ .await
+ .map_err(|e| StoreError::io(format!("sync {}", temp_path.display()), e))?;
+ drop(file);
+ tokio::fs::rename(&temp_path, &final_path)
+ .await
+ .map_err(|e| {
+ StoreError::io(
+ format!("rename {} to {}", temp_path.display(), final_path.display()),
+ e,
+ )
+ })?;
+ sync_dir(&parent).await
+ }
+ .await;
+
+ if write_result.is_err() {
+ // Best effort: never leave the temporary behind on failure.
+ let _ = tokio::fs::remove_file(&temp_path).await;
+ }
+ write_result
+ }
+
+ async fn get_impl(&self, key: &str) -> Result {
+ validate_key(key)?;
+ let (path, meta) = self.resolve_no_follow(key).await?;
+ if !meta.is_file() {
+ return Err(StoreError::NotFound);
+ }
+ let file = tokio::fs::File::open(&path)
+ .await
+ .map_err(|e| StoreError::io(format!("open {}", path.display()), e))?;
+ let display = path.display().to_string();
+ let stream = tokio_util::io::ReaderStream::new(file)
+ .map_err(move |e| StoreError::io(format!("read {display}"), e));
+ Ok(Box::pin(stream))
+ }
+
+ async fn head_impl(&self, key: &str) -> Result, StoreError> {
+ validate_key(key)?;
+ let (_, meta) = match self.resolve_no_follow(key).await {
+ Ok(resolved) => resolved,
+ Err(StoreError::NotFound) => return Ok(None),
+ Err(e) => return Err(e),
+ };
+ if !meta.is_file() {
+ return Ok(None);
+ }
+ Ok(Some(ObjectMeta {
+ key: key.to_owned(),
+ size: meta.len(),
+ last_modified: meta.modified().ok(),
+ }))
+ }
+
+ async fn list_impl(&self, prefix: &str) -> Result, StoreError> {
+ validate_prefix(prefix)?;
+ let normalized = normalize_prefix(prefix).to_owned();
+ let root = self.root.clone();
+ tokio::task::spawn_blocking(move || collect_objects(&root, &normalized))
+ .await
+ .map_err(|e| StoreError::Other(format!("listing task failed: {e}")))?
+ }
+
+ async fn delete_prefix_impl(&self, prefix: &str) -> Result {
+ validate_prefix(prefix)?;
+ let normalized = normalize_prefix(prefix).to_owned();
+ let root = self.root.clone();
+ tokio::task::spawn_blocking(move || delete_objects(&root, &normalized))
+ .await
+ .map_err(|e| StoreError::Other(format!("deletion task failed: {e}")))?
+ }
+
+ async fn validate_impl(&self) -> Result<(), StoreError> {
+ let meta = tokio::fs::metadata(&self.root).await.map_err(|e| {
+ StoreError::io(
+ format!("backup root {} is not usable", self.root.display()),
+ e,
+ )
+ })?;
+ if !meta.is_dir() {
+ return Err(StoreError::Other(format!(
+ "backup root {} is not a directory",
+ self.root.display()
+ )));
+ }
+ let probe = self.root.join(format!(
+ "{TEMP_PREFIX}validate-{}{TEMP_SUFFIX}",
+ uuid::Uuid::new_v4()
+ ));
+ tokio::fs::write(&probe, b"probe").await.map_err(|e| {
+ map_permission(
+ format!("backup root {} is not writable", self.root.display()),
+ e,
+ )
+ })?;
+ tokio::fs::remove_file(&probe)
+ .await
+ .map_err(|e| StoreError::io(format!("remove probe {}", probe.display()), e))?;
+ Ok(())
+ }
+}
+
+impl BackupStore for FilesystemStore {
+ fn put<'a>(&'a self, key: &'a str, body: ByteStream) -> BoxFuture<'a, Result<(), StoreError>> {
+ self.put_impl(key, body).boxed()
+ }
+
+ fn get<'a>(&'a self, key: &'a str) -> BoxFuture<'a, Result> {
+ self.get_impl(key).boxed()
+ }
+
+ fn head<'a>(&'a self, key: &'a str) -> BoxFuture<'a, Result, StoreError>> {
+ self.head_impl(key).boxed()
+ }
+
+ fn list<'a>(&'a self, prefix: &'a str) -> BoxStream<'a, Result> {
+ let items = async move {
+ match self.list_impl(prefix).await {
+ Ok(metas) => metas.into_iter().map(Ok).collect::>(),
+ Err(e) => vec![Err(e)],
+ }
+ };
+ Box::pin(
+ futures::stream::once(items)
+ .map(futures::stream::iter)
+ .flatten(),
+ )
+ }
+
+ fn delete_prefix<'a>(&'a self, prefix: &'a str) -> BoxFuture<'a, Result> {
+ self.delete_prefix_impl(prefix).boxed()
+ }
+
+ fn validate(&self) -> BoxFuture<'_, Result<(), StoreError>> {
+ self.validate_impl().boxed()
+ }
+}
+
+fn map_permission(context: String, e: std::io::Error) -> StoreError {
+ if e.kind() == std::io::ErrorKind::PermissionDenied {
+ StoreError::PermissionDenied(context)
+ } else {
+ StoreError::io(context, e)
+ }
+}
+
+/// Refusal for a symlink found in a key's path. The message names the path
+/// under the root that was refused, never the link's target.
+fn symlink_refused(path: &Path) -> StoreError {
+ StoreError::PermissionDenied(format!(
+ "{} is a symlink; keys never resolve through symlinks",
+ path.display()
+ ))
+}
+
+/// Fsync a directory so a rename inside it is durable. Windows has no
+/// directory handle to sync; there the rename's durability is left to the
+/// operating system.
+async fn sync_dir(dir: &std::path::Path) -> Result<(), StoreError> {
+ #[cfg(unix)]
+ {
+ let handle = tokio::fs::File::open(dir)
+ .await
+ .map_err(|e| StoreError::io(format!("open directory {}", dir.display()), e))?;
+ handle
+ .sync_all()
+ .await
+ .map_err(|e| StoreError::io(format!("sync directory {}", dir.display()), e))?;
+ }
+ #[cfg(not(unix))]
+ {
+ let _ = dir;
+ }
+ Ok(())
+}
+
+fn is_temp_name(name: &str) -> bool {
+ name.starts_with(TEMP_PREFIX) && name.ends_with(TEMP_SUFFIX)
+}
+
+/// Path for a normalized (possibly empty) prefix.
+fn prefix_path(root: &Path, prefix: &str) -> PathBuf {
+ if prefix.is_empty() {
+ return root.to_path_buf();
+ }
+ let mut path = root.to_path_buf();
+ for component in prefix.split('/') {
+ path.push(component);
+ }
+ path
+}
+
+/// Collect object metadata under a normalized prefix, sorted by key.
+fn collect_objects(root: &Path, prefix: &str) -> Result, StoreError> {
+ let base = prefix_path(root, prefix);
+ let mut out = Vec::new();
+ match std::fs::symlink_metadata(&base) {
+ Ok(meta) if meta.is_file() => {
+ // The prefix names an object directly (key == prefix).
+ out.push(ObjectMeta {
+ key: prefix.to_owned(),
+ size: meta.len(),
+ last_modified: meta.modified().ok(),
+ });
+ }
+ Ok(meta) if meta.is_dir() => {
+ walk_files(root, &base, &mut out)?;
+ }
+ // A symlink at the prefix, or nothing there at all: no objects.
+ Ok(_) | Err(_) => {}
+ }
+ // Defense in depth: the walk starts at the prefix directory, so every key
+ // matches by construction; keep the shared rule as the final filter.
+ out.retain(|meta| key_matches_prefix(&meta.key, prefix));
+ out.sort_by(|a, b| a.key.cmp(&b.key));
+ Ok(out)
+}
+
+fn walk_files(root: &Path, dir: &Path, out: &mut Vec) -> Result<(), StoreError> {
+ let entries = std::fs::read_dir(dir)
+ .map_err(|e| StoreError::io(format!("read directory {}", dir.display()), e))?;
+ for entry in entries {
+ let entry =
+ entry.map_err(|e| StoreError::io(format!("read directory {}", dir.display()), e))?;
+ let path = entry.path();
+ let file_type = entry
+ .file_type()
+ .map_err(|e| StoreError::io(format!("stat {}", path.display()), e))?;
+ if file_type.is_dir() {
+ walk_files(root, &path, out)?;
+ } else if file_type.is_file() {
+ let name = entry.file_name();
+ if is_temp_name(&name.to_string_lossy()) {
+ continue;
+ }
+ let meta = entry
+ .metadata()
+ .map_err(|e| StoreError::io(format!("stat {}", path.display()), e))?;
+ let key = path
+ .strip_prefix(root)
+ .expect("walk stays under the root")
+ .components()
+ .map(|c| c.as_os_str().to_string_lossy().into_owned())
+ .collect::>()
+ .join("/");
+ out.push(ObjectMeta {
+ key,
+ size: meta.len(),
+ last_modified: meta.modified().ok(),
+ });
+ }
+ // Symlinks are never objects this store wrote; skip them.
+ }
+ Ok(())
+}
+
+/// Delete everything under a normalized prefix; returns the number of files
+/// removed. Directories emptied by the deletion are removed too, including
+/// the prefix directory itself.
+fn delete_objects(root: &Path, prefix: &str) -> Result {
+ let base = prefix_path(root, prefix);
+ match std::fs::symlink_metadata(&base) {
+ Ok(meta) if meta.is_file() => {
+ std::fs::remove_file(&base)
+ .map_err(|e| StoreError::io(format!("remove {}", base.display()), e))?;
+ Ok(1)
+ }
+ Ok(meta) if meta.is_dir() => {
+ if prefix.is_empty() {
+ // Deleting everything must not remove the root itself.
+ let mut count = 0;
+ let entries = std::fs::read_dir(&base)
+ .map_err(|e| StoreError::io(format!("read directory {}", base.display()), e))?;
+ for entry in entries {
+ let entry = entry.map_err(|e| {
+ StoreError::io(format!("read directory {}", base.display()), e)
+ })?;
+ count += delete_entry(&entry.path())?;
+ }
+ Ok(count)
+ } else {
+ delete_entry(&base)
+ }
+ }
+ Ok(_) | Err(_) => Ok(0),
+ }
+}
+
+fn delete_entry(path: &Path) -> Result {
+ let meta = std::fs::symlink_metadata(path)
+ .map_err(|e| StoreError::io(format!("stat {}", path.display()), e))?;
+ if meta.is_dir() {
+ let mut count = 0;
+ let entries = std::fs::read_dir(path)
+ .map_err(|e| StoreError::io(format!("read directory {}", path.display()), e))?;
+ for entry in entries {
+ let entry = entry
+ .map_err(|e| StoreError::io(format!("read directory {}", path.display()), e))?;
+ count += delete_entry(&entry.path())?;
+ }
+ std::fs::remove_dir(path)
+ .map_err(|e| StoreError::io(format!("remove directory {}", path.display()), e))?;
+ Ok(count)
+ } else {
+ std::fs::remove_file(path)
+ .map_err(|e| StoreError::io(format!("remove {}", path.display()), e))?;
+ Ok(1)
+ }
+}
diff --git a/crates/backup-store/src/key.rs b/crates/backup-store/src/key.rs
new file mode 100644
index 00000000..a6726102
--- /dev/null
+++ b/crates/backup-store/src/key.rs
@@ -0,0 +1,181 @@
+// Copyright 2026 ExtendDB contributors
+// SPDX-License-Identifier: Apache-2.0
+
+//! Key and prefix validation shared by both store implementations.
+//!
+//! A key is a `/`-separated component path. The rules exist so a key can be
+//! mapped onto a filesystem path without ever escaping the store root, and so
+//! the two stores accept exactly the same key space:
+//!
+//! - every component is non-empty (no leading, trailing, or doubled `/`)
+//! - no component is `.` or `..`
+//! - no component contains a backslash (a path separator on other platforms)
+//! - no component contains a control character
+//! - the whole key is at most [`MAX_KEY_LEN`] bytes
+
+use crate::StoreError;
+
+/// Maximum total key length in bytes, matching the S3 key length limit.
+pub const MAX_KEY_LEN: usize = 1024;
+
+/// Validate an object key.
+///
+/// # Errors
+///
+/// Returns [`StoreError::InvalidKey`] naming the rule the key violated.
+pub fn validate_key(key: &str) -> Result<(), StoreError> {
+ if key.is_empty() {
+ return Err(StoreError::InvalidKey("key is empty".to_owned()));
+ }
+ if key.len() > MAX_KEY_LEN {
+ return Err(StoreError::InvalidKey(format!(
+ "key is {} bytes; the maximum is {MAX_KEY_LEN}",
+ key.len()
+ )));
+ }
+ for component in key.split('/') {
+ validate_component(component)?;
+ }
+ Ok(())
+}
+
+/// Validate a listing or deletion prefix.
+///
+/// A prefix follows the key rules with two relaxations: it may be empty
+/// (matching every object) and it may carry one trailing `/`.
+///
+/// # Errors
+///
+/// Returns [`StoreError::InvalidKey`] naming the rule the prefix violated.
+pub fn validate_prefix(prefix: &str) -> Result<(), StoreError> {
+ if prefix.is_empty() {
+ return Ok(());
+ }
+ let normalized = prefix.strip_suffix('/').unwrap_or(prefix);
+ if normalized.is_empty() {
+ return Err(StoreError::InvalidKey(
+ "prefix consists only of a separator".to_owned(),
+ ));
+ }
+ validate_key(normalized)
+}
+
+/// Strip at most one trailing `/` from a validated prefix, yielding the form
+/// the matching rule operates on.
+pub(crate) fn normalize_prefix(prefix: &str) -> &str {
+ prefix.strip_suffix('/').unwrap_or(prefix)
+}
+
+/// Whether `key` is matched by the normalized prefix `prefix`: the empty
+/// prefix matches everything, otherwise the key must equal the prefix or sit
+/// under `prefix/`. Component-aligned, so prefix `a/b` never matches `a/bc`.
+pub(crate) fn key_matches_prefix(key: &str, prefix: &str) -> bool {
+ if prefix.is_empty() {
+ return true;
+ }
+ match key.strip_prefix(prefix) {
+ Some("") => true,
+ Some(rest) => rest.starts_with('/'),
+ None => false,
+ }
+}
+
+fn validate_component(component: &str) -> Result<(), StoreError> {
+ if component.is_empty() {
+ return Err(StoreError::InvalidKey(
+ "key has an empty component (leading, trailing, or doubled '/')".to_owned(),
+ ));
+ }
+ if component == "." || component == ".." {
+ return Err(StoreError::InvalidKey(format!(
+ "key component '{component}' is not allowed"
+ )));
+ }
+ if component.contains('\\') {
+ return Err(StoreError::InvalidKey(
+ "key component contains a backslash".to_owned(),
+ ));
+ }
+ if component.chars().any(char::is_control) {
+ return Err(StoreError::InvalidKey(
+ "key component contains a control character".to_owned(),
+ ));
+ }
+ Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn accepts_well_formed_keys() {
+ for key in [
+ "a",
+ "a/b",
+ "123456789012/orders/backup-01J5/extenddb-manifest.json",
+ "data/000001.json.gz",
+ "with space/and.dots.json",
+ "unicode/ключ/键",
+ ] {
+ assert!(validate_key(key).is_ok(), "{key} should be valid");
+ }
+ }
+
+ #[test]
+ fn rejects_malformed_keys() {
+ let cases: &[&str] = &[
+ "", "/", "/a", "a/", "a//b", ".", "..", "a/./b", "a/../b", "../a", "a/..", "a\\b",
+ "a/b\\c", "a\x00b", "a/\x1fb", "a\x7f",
+ ];
+ for key in cases {
+ assert!(
+ matches!(validate_key(key), Err(StoreError::InvalidKey(_))),
+ "{key:?} should be rejected"
+ );
+ }
+ }
+
+ #[test]
+ fn rejects_overlong_keys() {
+ let key = "a/".repeat(512) + "b";
+ assert!(key.len() > MAX_KEY_LEN);
+ assert!(matches!(validate_key(&key), Err(StoreError::InvalidKey(_))));
+ let max = "a".repeat(MAX_KEY_LEN);
+ assert!(validate_key(&max).is_ok());
+ }
+
+ #[test]
+ fn prefix_rules() {
+ assert!(validate_prefix("").is_ok());
+ assert!(validate_prefix("a").is_ok());
+ assert!(validate_prefix("a/b/").is_ok());
+ assert!(matches!(
+ validate_prefix("/"),
+ Err(StoreError::InvalidKey(_))
+ ));
+ assert!(matches!(
+ validate_prefix("a//b"),
+ Err(StoreError::InvalidKey(_))
+ ));
+ assert!(matches!(
+ validate_prefix("a/../b"),
+ Err(StoreError::InvalidKey(_))
+ ));
+ assert!(matches!(
+ validate_prefix("a\x01"),
+ Err(StoreError::InvalidKey(_))
+ ));
+ }
+
+ #[test]
+ fn prefix_matching_is_component_aligned() {
+ assert!(key_matches_prefix("a/b/c", ""));
+ assert!(key_matches_prefix("a/b/c", "a"));
+ assert!(key_matches_prefix("a/b/c", "a/b"));
+ assert!(key_matches_prefix("a/b", "a/b"));
+ assert!(!key_matches_prefix("a/bc", "a/b"));
+ assert!(!key_matches_prefix("ab/c", "a"));
+ assert!(!key_matches_prefix("a", "a/b"));
+ }
+}
diff --git a/crates/backup-store/src/lib.rs b/crates/backup-store/src/lib.rs
new file mode 100644
index 00000000..f4e8716f
--- /dev/null
+++ b/crates/backup-store/src/lib.rs
@@ -0,0 +1,137 @@
+// Copyright 2026 ExtendDB contributors
+// SPDX-License-Identifier: Apache-2.0
+
+//! Backup object store abstraction.
+//!
+//! One trait, [`BackupStore`], and two implementations: [`FilesystemStore`]
+//! writes objects under a local root directory, [`S3Store`] writes them to an
+//! S3 bucket through the official AWS SDK. Both are exercised by the same
+//! conformance test module in `tests/conformance.rs`.
+//!
+//! Keys are `/`-separated component paths validated by [`key::validate_key`];
+//! the same rules apply to both stores so a backup written by one store can be
+//! read by the other. See `docs/design/15-backup-restore.md` for the full
+//! design.
+
+use std::sync::Arc;
+
+use bytes::Bytes;
+use futures::future::BoxFuture;
+use futures::stream::BoxStream;
+
+mod config;
+mod error;
+mod fs;
+pub mod key;
+mod s3;
+
+pub use config::{BackupStoreConfig, BackupStoreKind, FilesystemStoreConfig, S3StoreConfig};
+pub use error::StoreError;
+pub use fs::FilesystemStore;
+pub use s3::S3Store;
+
+/// Byte stream flowing into and out of a store.
+///
+/// `put` consumes one, `get` returns one. Errors from the underlying source
+/// propagate as [`StoreError`], which lets tests inject a mid-body failure to
+/// exercise abort paths.
+pub type ByteStream = BoxStream<'static, Result>;
+
+/// Build a [`ByteStream`] from an in-memory buffer.
+pub fn byte_stream_from(bytes: impl Into) -> ByteStream {
+ let bytes = bytes.into();
+ Box::pin(futures::stream::once(async move { Ok(bytes) }))
+}
+
+/// Metadata for one stored object.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ObjectMeta {
+ /// Object key, relative to the store root (and, for S3, the configured
+ /// bucket prefix).
+ pub key: String,
+ /// Object size in bytes.
+ pub size: u64,
+ /// Last modification time, when the store reports one.
+ pub last_modified: Option,
+}
+
+/// A destination for backup objects.
+///
+/// Implementations must be safe to share across tasks; the engine holds one
+/// behind an `Arc` for the lifetime of the process.
+pub trait BackupStore: Send + Sync {
+ /// Store `body` under `key`, replacing any existing object.
+ ///
+ /// A failed put must leave nothing observable under `key`: the filesystem
+ /// store writes to a temporary sibling and renames into place, the S3
+ /// store aborts an in-flight multipart upload.
+ ///
+ /// One key-space divergence between the stores: the filesystem store
+ /// cannot hold an object at `a` and another at `a/b` (a file and a
+ /// directory cannot share a path), while S3 can. The backup layout keys
+ /// objects only under per-backup prefixes, so it never produces that
+ /// shape.
+ fn put<'a>(&'a self, key: &'a str, body: ByteStream) -> BoxFuture<'a, Result<(), StoreError>>;
+
+ /// Stream the object stored under `key`.
+ ///
+ /// Returns [`StoreError::NotFound`] when no object exists.
+ fn get<'a>(&'a self, key: &'a str) -> BoxFuture<'a, Result>;
+
+ /// Metadata for the object under `key`, or `None` when no object exists.
+ fn head<'a>(&'a self, key: &'a str) -> BoxFuture<'a, Result, StoreError>>;
+
+ /// Stream metadata for every object whose key equals `prefix` or sits
+ /// under `prefix/`, in lexicographic key order. An empty prefix lists
+ /// every object. Prefix matching is component-aligned: prefix `a/b` does
+ /// not match key `a/bc`.
+ fn list<'a>(&'a self, prefix: &'a str) -> BoxStream<'a, Result>;
+
+ /// Delete every object matched by `prefix` (same matching rule as
+ /// [`BackupStore::list`]) and return how many objects were removed. The
+ /// filesystem store also removes directories the deletion emptied, and
+ /// its count includes any files under the prefix that listings hide
+ /// (in-flight temporaries, symlinks), so the count can exceed the number
+ /// of objects `list` reported. A prefix that matches nothing returns
+ /// `Ok(0)`.
+ fn delete_prefix<'a>(&'a self, prefix: &'a str) -> BoxFuture<'a, Result>;
+
+ /// Startup validation. The filesystem store checks the root exists and is
+ /// writable by writing and removing a probe file; the S3 store calls
+ /// `HeadBucket`. Called once at server startup so a misconfigured store
+ /// fails then, not at the first `CreateBackup`.
+ fn validate(&self) -> BoxFuture<'_, Result<(), StoreError>>;
+}
+
+/// Open the store described by `config`.
+///
+/// Validates the per-kind section is present and its values are usable, then
+/// constructs the store. S3 credentials come from the standard AWS SDK chain;
+/// they are never part of the configuration.
+///
+/// # Errors
+///
+/// Returns an error when the section matching the selected store kind is
+/// absent, a value fails validation (part size floor, zero concurrency,
+/// malformed prefix), or the filesystem root is unusable.
+pub async fn open(config: &BackupStoreConfig) -> Result, StoreError> {
+ match config.store {
+ BackupStoreKind::Filesystem => {
+ let fs_config = config.filesystem.as_ref().ok_or_else(|| {
+ StoreError::Other(
+ "[backup.filesystem] with a path is required when store = \"filesystem\""
+ .to_owned(),
+ )
+ })?;
+ Ok(Arc::new(FilesystemStore::open(&fs_config.path).await?))
+ }
+ BackupStoreKind::S3 => {
+ let s3_config = config.s3.as_ref().ok_or_else(|| {
+ StoreError::Other(
+ "[backup.s3] with a bucket is required when store = \"s3\"".to_owned(),
+ )
+ })?;
+ Ok(Arc::new(S3Store::open(s3_config).await?))
+ }
+ }
+}
diff --git a/crates/backup-store/src/s3.rs b/crates/backup-store/src/s3.rs
new file mode 100644
index 00000000..4e686da3
--- /dev/null
+++ b/crates/backup-store/src/s3.rs
@@ -0,0 +1,596 @@
+// Copyright 2026 ExtendDB contributors
+// SPDX-License-Identifier: Apache-2.0
+
+//! S3-backed store built on the official `aws-sdk-s3` crate.
+//!
+//! Credentials come from the standard AWS SDK chain (environment, shared
+//! config, IMDS, IRSA, container credentials); the store never holds them
+//! itself. Uploads larger than the configured part size use multipart, and a
+//! failed multipart upload is aborted so no orphaned parts accrue storage.
+//! Listing paginates; prefix deletion batches `DeleteObjects` requests.
+
+use aws_sdk_s3::error::{ProvideErrorMetadata, SdkError};
+use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart, Delete, ObjectIdentifier};
+use bytes::{Bytes, BytesMut};
+use futures::future::BoxFuture;
+use futures::stream::BoxStream;
+use futures::{FutureExt, StreamExt, TryStreamExt};
+
+use crate::config::S3StoreConfig;
+use crate::key::{key_matches_prefix, normalize_prefix, validate_key, validate_prefix};
+use crate::{BackupStore, ByteStream, ObjectMeta, StoreError};
+
+/// The smallest part size S3 accepts for any part except the last.
+pub const MIN_PART_SIZE_BYTES: u64 = 5 * 1024 * 1024;
+
+/// Largest number of keys one `DeleteObjects` request accepts.
+const DELETE_BATCH: usize = 1000;
+
+/// Store writing objects to an S3 bucket (or an S3-compatible endpoint).
+pub struct S3Store {
+ client: aws_sdk_s3::Client,
+ bucket: String,
+ /// Normalized bucket-side key prefix: empty, or ending in `/`.
+ prefix: String,
+ part_size: usize,
+ max_concurrent_uploads: usize,
+}
+
+impl S3Store {
+ /// Open a store against the configured bucket.
+ ///
+ /// Enforces the part size floor and resolves credentials, region, and
+ /// retry behavior through the SDK default chain. `endpoint` and
+ /// `force_path_style` support S3-compatible stores such as `MinIO`.
+ ///
+ /// # Errors
+ ///
+ /// Returns an error when the part size is below the 5 MiB floor, the
+ /// concurrency is zero, or the configured prefix fails key validation.
+ pub async fn open(config: &S3StoreConfig) -> Result {
+ Self::open_inner(config, None).await
+ }
+
+ /// Open with an additional client interceptor. This is the failure
+ /// injection seam the crate's own test suite uses to make a chosen
+ /// operation fail; it compiles only with the `test-util` feature and is
+ /// not part of the supported API surface.
+ #[cfg(feature = "test-util")]
+ #[doc(hidden)]
+ pub async fn open_with_interceptor(
+ config: &S3StoreConfig,
+ interceptor: impl aws_sdk_s3::config::Intercept + 'static,
+ ) -> Result {
+ Self::open_inner(
+ config,
+ Some(aws_sdk_s3::config::SharedInterceptor::new(interceptor)),
+ )
+ .await
+ }
+
+ async fn open_inner(
+ config: &S3StoreConfig,
+ interceptor: Option,
+ ) -> Result {
+ if config.upload_part_size_bytes < MIN_PART_SIZE_BYTES {
+ return Err(StoreError::Other(format!(
+ "upload_part_size_bytes is {}; the minimum is {MIN_PART_SIZE_BYTES} (5 MiB)",
+ config.upload_part_size_bytes
+ )));
+ }
+ if config.max_concurrent_uploads == 0 {
+ return Err(StoreError::Other(
+ "max_concurrent_uploads must be at least 1".to_owned(),
+ ));
+ }
+ let part_size = usize::try_from(config.upload_part_size_bytes).map_err(|_| {
+ StoreError::Other(format!(
+ "upload_part_size_bytes {} does not fit this platform",
+ config.upload_part_size_bytes
+ ))
+ })?;
+ let prefix = normalize_store_prefix(&config.prefix)?;
+
+ let mut loader = aws_config::defaults(aws_config::BehaviorVersion::latest());
+ if let Some(region) = &config.region {
+ loader = loader.region(aws_config::Region::new(region.clone()));
+ }
+ if let Some(endpoint) = &config.endpoint {
+ loader = loader.endpoint_url(endpoint.clone());
+ }
+ let shared = loader.load().await;
+ let mut builder = aws_sdk_s3::config::Builder::from(&shared);
+ if config.force_path_style {
+ builder = builder.force_path_style(true);
+ }
+ if let Some(interceptor) = interceptor {
+ builder.push_interceptor(interceptor);
+ }
+ let client = aws_sdk_s3::Client::from_conf(builder.build());
+
+ Ok(Self {
+ client,
+ bucket: config.bucket.clone(),
+ prefix,
+ part_size,
+ max_concurrent_uploads: config.max_concurrent_uploads,
+ })
+ }
+
+ /// Bucket-side key for a store key.
+ fn full_key(&self, key: &str) -> String {
+ format!("{}{key}", self.prefix)
+ }
+
+ async fn put_impl(&self, key: &str, body: ByteStream) -> Result<(), StoreError> {
+ validate_key(key)?;
+ let full_key = self.full_key(key);
+ let mut chunks = std::pin::pin!(chunk_stream(body, self.part_size));
+
+ let first = match chunks.try_next().await? {
+ Some(chunk) => chunk,
+ None => Bytes::new(),
+ };
+ let second = chunks.try_next().await?;
+
+ let Some(second) = second else {
+ // The whole body fits in one part: a single PutObject.
+ self.client
+ .put_object()
+ .bucket(&self.bucket)
+ .key(&full_key)
+ .body(first.into())
+ .send()
+ .await
+ .map_err(|e| map_sdk_error("PutObject", e))?;
+ return Ok(());
+ };
+
+ let upload_id = self
+ .client
+ .create_multipart_upload()
+ .bucket(&self.bucket)
+ .key(&full_key)
+ .send()
+ .await
+ .map_err(|e| map_sdk_error("CreateMultipartUpload", e))?
+ .upload_id()
+ .ok_or_else(|| {
+ StoreError::Other("CreateMultipartUpload returned no upload id".to_owned())
+ })?
+ .to_owned();
+
+ // Every failure after CreateMultipartUpload must reach the abort,
+ // including a failed CompleteMultipartUpload, so no orphaned parts
+ // accrue storage. A complete that failed after S3 committed it makes
+ // the abort answer NoSuchUpload, which the best-effort warn tolerates.
+ let outcome: Result<(), StoreError> = async {
+ let parts = self
+ .upload_parts(&full_key, &upload_id, first, second, chunks)
+ .await?;
+ self.client
+ .complete_multipart_upload()
+ .bucket(&self.bucket)
+ .key(&full_key)
+ .upload_id(&upload_id)
+ .multipart_upload(
+ CompletedMultipartUpload::builder()
+ .set_parts(Some(parts))
+ .build(),
+ )
+ .send()
+ .await
+ .map_err(|e| map_sdk_error("CompleteMultipartUpload", e))?;
+ Ok(())
+ }
+ .await;
+
+ if let Err(e) = outcome {
+ // Best effort: the original failure is what the caller needs.
+ if let Err(abort_err) = self
+ .client
+ .abort_multipart_upload()
+ .bucket(&self.bucket)
+ .key(&full_key)
+ .upload_id(&upload_id)
+ .send()
+ .await
+ {
+ tracing::warn!(
+ key = %full_key,
+ error = %map_sdk_error("AbortMultipartUpload", abort_err),
+ "failed to abort multipart upload after a put failure"
+ );
+ }
+ return Err(e);
+ }
+ Ok(())
+ }
+
+ /// Upload every part with bounded concurrency; returns the completed part
+ /// list sorted by part number.
+ async fn upload_parts(
+ &self,
+ full_key: &str,
+ upload_id: &str,
+ first: Bytes,
+ second: Bytes,
+ rest: impl futures::Stream- > + Send,
+ ) -> Result, StoreError> {
+ let all_chunks = futures::stream::iter([Ok(first), Ok(second)]).chain(rest);
+ let mut parts: Vec = all_chunks
+ .enumerate()
+ .map(|(index, chunk)| async move {
+ let chunk = chunk?;
+ let part_number = i32::try_from(index + 1).map_err(|_| {
+ StoreError::Other("multipart upload exceeds the part number range".to_owned())
+ })?;
+ let output = self
+ .client
+ .upload_part()
+ .bucket(&self.bucket)
+ .key(full_key)
+ .upload_id(upload_id)
+ .part_number(part_number)
+ .body(chunk.into())
+ .send()
+ .await
+ .map_err(|e| map_sdk_error("UploadPart", e))?;
+ let e_tag = output.e_tag().ok_or_else(|| {
+ StoreError::Other("UploadPart response carried no ETag".to_owned())
+ })?;
+ Ok::<_, StoreError>(
+ CompletedPart::builder()
+ .part_number(part_number)
+ .e_tag(e_tag)
+ .build(),
+ )
+ })
+ .buffered(self.max_concurrent_uploads)
+ .try_collect()
+ .await?;
+ parts.sort_by_key(CompletedPart::part_number);
+ Ok(parts)
+ }
+
+ async fn get_impl(&self, key: &str) -> Result {
+ validate_key(key)?;
+ let output = self
+ .client
+ .get_object()
+ .bucket(&self.bucket)
+ .key(self.full_key(key))
+ .send()
+ .await
+ .map_err(|e| map_sdk_error("GetObject", e))?;
+ Ok(Box::pin(futures::stream::unfold(
+ output.body,
+ |mut body| async move {
+ match body.next().await {
+ Some(Ok(bytes)) => Some((Ok(bytes), body)),
+ Some(Err(e)) => Some((
+ Err(StoreError::Transport {
+ source: Box::new(e),
+ }),
+ body,
+ )),
+ None => None,
+ }
+ },
+ )))
+ }
+
+ async fn head_impl(&self, key: &str) -> Result, StoreError> {
+ validate_key(key)?;
+ let output = match self
+ .client
+ .head_object()
+ .bucket(&self.bucket)
+ .key(self.full_key(key))
+ .send()
+ .await
+ {
+ Ok(output) => output,
+ Err(e) => {
+ return match map_sdk_error("HeadObject", e) {
+ StoreError::NotFound => Ok(None),
+ other => Err(other),
+ };
+ }
+ };
+ let size = output
+ .content_length()
+ .and_then(|len| u64::try_from(len).ok());
+ Ok(Some(ObjectMeta {
+ key: key.to_owned(),
+ size: size.unwrap_or(0),
+ last_modified: output.last_modified().map(smithy_time_to_system),
+ }))
+ }
+
+ /// One page of matching objects plus the continuation token, already
+ /// filtered by the component-aligned matching rule and mapped back to
+ /// store keys.
+ async fn list_page(
+ &self,
+ match_prefix: &str,
+ token: Option,
+ ) -> Result<(Vec, Option), StoreError> {
+ let request_prefix = self.full_key(match_prefix);
+ let output = self
+ .client
+ .list_objects_v2()
+ .bucket(&self.bucket)
+ .prefix(request_prefix)
+ .set_continuation_token(token)
+ .send()
+ .await
+ .map_err(|e| map_sdk_error("ListObjectsV2", e))?;
+ let metas = output
+ .contents()
+ .iter()
+ .filter_map(|object| {
+ let full = object.key()?;
+ let key = full.strip_prefix(&self.prefix)?;
+ if !key_matches_prefix(key, match_prefix) {
+ return None;
+ }
+ Some(ObjectMeta {
+ key: key.to_owned(),
+ size: object
+ .size()
+ .and_then(|s| u64::try_from(s).ok())
+ .unwrap_or(0),
+ last_modified: object.last_modified().map(smithy_time_to_system),
+ })
+ })
+ .collect();
+ Ok((metas, output.next_continuation_token().map(str::to_owned)))
+ }
+
+ async fn delete_prefix_impl(&self, prefix: &str) -> Result {
+ validate_prefix(prefix)?;
+ let match_prefix = normalize_prefix(prefix);
+ let mut deleted: u64 = 0;
+ let mut token: Option = None;
+ loop {
+ let (metas, next) = self.list_page(match_prefix, token).await?;
+ for batch in metas.chunks(DELETE_BATCH) {
+ if batch.is_empty() {
+ continue;
+ }
+ let objects = batch
+ .iter()
+ .map(|meta| {
+ ObjectIdentifier::builder()
+ .key(self.full_key(&meta.key))
+ .build()
+ .map_err(|e| StoreError::Other(format!("building delete request: {e}")))
+ })
+ .collect::, _>>()?;
+ let count = objects.len() as u64;
+ let delete = Delete::builder()
+ .set_objects(Some(objects))
+ .quiet(true)
+ .build()
+ .map_err(|e| StoreError::Other(format!("building delete request: {e}")))?;
+ let output = self
+ .client
+ .delete_objects()
+ .bucket(&self.bucket)
+ .delete(delete)
+ .send()
+ .await
+ .map_err(|e| map_sdk_error("DeleteObjects", e))?;
+ let errors = output.errors();
+ if let Some(first) = errors.first() {
+ return Err(StoreError::Other(format!(
+ "DeleteObjects failed for {} of {count} keys; first: {} on {}",
+ errors.len(),
+ first.code().unwrap_or("unknown"),
+ first.key().unwrap_or("unknown"),
+ )));
+ }
+ deleted += count;
+ }
+ match next {
+ Some(next) => token = Some(next),
+ None => break,
+ }
+ }
+ Ok(deleted)
+ }
+
+ async fn validate_impl(&self) -> Result<(), StoreError> {
+ self.client
+ .head_bucket()
+ .bucket(&self.bucket)
+ .send()
+ .await
+ .map_err(|e| {
+ StoreError::Other(format!(
+ "backup bucket {} failed validation: {}",
+ self.bucket,
+ map_sdk_error("HeadBucket", e)
+ ))
+ })?;
+ Ok(())
+ }
+}
+
+impl BackupStore for S3Store {
+ fn put<'a>(&'a self, key: &'a str, body: ByteStream) -> BoxFuture<'a, Result<(), StoreError>> {
+ self.put_impl(key, body).boxed()
+ }
+
+ fn get<'a>(&'a self, key: &'a str) -> BoxFuture<'a, Result> {
+ self.get_impl(key).boxed()
+ }
+
+ fn head<'a>(&'a self, key: &'a str) -> BoxFuture<'a, Result, StoreError>> {
+ self.head_impl(key).boxed()
+ }
+
+ fn list<'a>(&'a self, prefix: &'a str) -> BoxStream<'a, Result> {
+ enum State {
+ Start,
+ Next(String),
+ Done,
+ }
+ let pages = futures::stream::try_unfold(State::Start, move |state| async move {
+ let token = match state {
+ State::Start => {
+ validate_prefix(prefix)?;
+ None
+ }
+ State::Next(token) => Some(token),
+ State::Done => return Ok(None),
+ };
+ let (metas, next) = self.list_page(normalize_prefix(prefix), token).await?;
+ let next_state = match next {
+ Some(token) => State::Next(token),
+ None => State::Done,
+ };
+ Ok(Some((metas, next_state)))
+ });
+ Box::pin(
+ pages
+ .map_ok(|metas| futures::stream::iter(metas.into_iter().map(Ok)))
+ .try_flatten(),
+ )
+ }
+
+ fn delete_prefix<'a>(&'a self, prefix: &'a str) -> BoxFuture<'a, Result> {
+ self.delete_prefix_impl(prefix).boxed()
+ }
+
+ fn validate(&self) -> BoxFuture<'_, Result<(), StoreError>> {
+ self.validate_impl().boxed()
+ }
+}
+
+/// Normalize the configured bucket prefix: strip leading `/`, require a
+/// trailing `/` when non-empty, and hold it to the same component rules as
+/// keys.
+fn normalize_store_prefix(prefix: &str) -> Result {
+ let trimmed = prefix.trim_start_matches('/');
+ if trimmed.is_empty() {
+ return Ok(String::new());
+ }
+ let without_slash = trimmed.strip_suffix('/').unwrap_or(trimmed);
+ validate_key(without_slash)
+ .map_err(|e| StoreError::Other(format!("invalid [backup.s3] prefix {prefix:?}: {e}")))?;
+ Ok(format!("{without_slash}/"))
+}
+
+/// Split a byte stream into chunks of exactly `size` bytes, with a final
+/// short chunk carrying the remainder. An empty body yields no chunks.
+fn chunk_stream(
+ body: ByteStream,
+ size: usize,
+) -> impl futures::Stream
- > + Send {
+ futures::stream::try_unfold(
+ (body, BytesMut::new(), false),
+ move |(mut body, mut buffer, mut exhausted)| async move {
+ loop {
+ if buffer.len() >= size {
+ let chunk = buffer.split_to(size).freeze();
+ return Ok(Some((chunk, (body, buffer, exhausted))));
+ }
+ if exhausted {
+ if buffer.is_empty() {
+ return Ok(None);
+ }
+ let chunk = buffer.split().freeze();
+ return Ok(Some((chunk, (body, buffer, exhausted))));
+ }
+ match body.next().await {
+ Some(Ok(bytes)) => buffer.extend_from_slice(&bytes),
+ Some(Err(e)) => return Err(e),
+ None => exhausted = true,
+ }
+ }
+ },
+ )
+}
+
+fn smithy_time_to_system(dt: &aws_sdk_s3::primitives::DateTime) -> std::time::SystemTime {
+ std::time::SystemTime::try_from(*dt).unwrap_or(std::time::UNIX_EPOCH)
+}
+
+/// Reduce an SDK error to a [`StoreError`].
+///
+/// Service errors are classified by HTTP status and error code and formatted
+/// from the code and service message only, so no request or credential
+/// material reaches the Display output. Everything else (connection,
+/// timeout, response deserialization) becomes [`StoreError::Transport`].
+fn map_sdk_error(operation: &'static str, err: SdkError) -> StoreError
+where
+ E: ProvideErrorMetadata + std::error::Error + Send + Sync + 'static,
+{
+ match err {
+ SdkError::ServiceError(ctx) => {
+ let status = ctx.raw().status().as_u16();
+ let code = ctx.err().code().unwrap_or("unknown").to_owned();
+ match (status, code.as_str()) {
+ (404, _) | (_, "NoSuchKey" | "NotFound" | "NoSuchBucket") => StoreError::NotFound,
+ (403, _) | (_, "AccessDenied") => {
+ StoreError::PermissionDenied(format!("s3 {operation} was denied"))
+ }
+ _ => {
+ let message = ctx.err().message().unwrap_or("no message").to_owned();
+ StoreError::Other(format!("s3 {operation} failed: {code}: {message}"))
+ }
+ }
+ }
+ other => StoreError::Transport {
+ source: Box::new(other),
+ },
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn store_prefix_normalization() {
+ assert_eq!(normalize_store_prefix("").unwrap(), "");
+ assert_eq!(normalize_store_prefix("extenddb/").unwrap(), "extenddb/");
+ assert_eq!(normalize_store_prefix("extenddb").unwrap(), "extenddb/");
+ assert_eq!(normalize_store_prefix("/a/b").unwrap(), "a/b/");
+ assert!(normalize_store_prefix("a//b").is_err());
+ assert!(normalize_store_prefix("a/../b").is_err());
+ }
+
+ #[tokio::test]
+ async fn chunker_splits_exactly() {
+ let body = crate::byte_stream_from(vec![7u8; 10]);
+ let chunks: Vec = chunk_stream(body, 4).try_collect().await.unwrap();
+ assert_eq!(
+ chunks.iter().map(Bytes::len).collect::>(),
+ vec![4, 4, 2]
+ );
+ }
+
+ #[tokio::test]
+ async fn chunker_handles_exact_multiple_and_empty() {
+ let body = crate::byte_stream_from(vec![1u8; 8]);
+ let chunks: Vec = chunk_stream(body, 4).try_collect().await.unwrap();
+ assert_eq!(chunks.len(), 2);
+
+ let empty = crate::byte_stream_from(Vec::new());
+ let chunks: Vec = chunk_stream(empty, 4).try_collect().await.unwrap();
+ assert!(chunks.is_empty());
+ }
+
+ #[tokio::test]
+ async fn chunker_propagates_source_errors() {
+ let body: ByteStream = Box::pin(futures::stream::iter([
+ Ok(Bytes::from_static(b"abcd")),
+ Err(StoreError::Other("injected".to_owned())),
+ ]));
+ let result: Result, StoreError> = chunk_stream(body, 2).try_collect().await;
+ assert!(matches!(result, Err(StoreError::Other(msg)) if msg == "injected"));
+ }
+}
diff --git a/crates/backup-store/tests/conformance.rs b/crates/backup-store/tests/conformance.rs
new file mode 100644
index 00000000..7fc35690
--- /dev/null
+++ b/crates/backup-store/tests/conformance.rs
@@ -0,0 +1,929 @@
+// Copyright 2026 ExtendDB contributors
+// SPDX-License-Identifier: Apache-2.0
+
+//! Conformance suite run against both store implementations.
+//!
+//! The `suite` module holds store-agnostic checks; the `filesystem` and `s3`
+//! modules bind them to a tempdir-backed [`FilesystemStore`] and an
+//! [`S3Store`] against a local S3-compatible endpoint. S3 tests run only when
+//! `EXTENDDB_TEST_S3_ENDPOINT` is set and skip with a printed reason
+//! otherwise; see the crate README for how to run them.
+
+use bytes::Bytes;
+use extenddb_backup_store::{BackupStore, ByteStream, StoreError};
+use futures::TryStreamExt;
+
+/// Part size every conformance run uses: the smallest S3 accepts, so the
+/// multipart body stays cheap to generate.
+const PART_SIZE: u64 = 5 * 1024 * 1024;
+
+/// Deterministic pseudo-random bytes, so content equality proves the store
+/// reassembled parts in order.
+fn patterned_bytes(len: usize) -> Vec {
+ let mut state: u64 = 0x2545_f491_4f6c_dd1d;
+ let mut out = Vec::with_capacity(len);
+ while out.len() < len {
+ state ^= state << 13;
+ state ^= state >> 7;
+ state ^= state << 17;
+ out.extend_from_slice(&state.to_le_bytes());
+ }
+ out.truncate(len);
+ out
+}
+
+/// Stream a buffer in 1 MiB pieces so puts exercise real chunk reassembly
+/// rather than a single-frame fast path.
+fn chunked_body(data: impl AsRef<[u8]>) -> ByteStream {
+ let pieces: Vec> = data
+ .as_ref()
+ .chunks(1024 * 1024)
+ .map(|c| Ok(Bytes::copy_from_slice(c)))
+ .collect();
+ Box::pin(futures::stream::iter(pieces))
+}
+
+/// A body that yields `good` and then fails, for abort-path checks.
+fn failing_body(good: impl AsRef<[u8]>) -> ByteStream {
+ let pieces: Vec> = good
+ .as_ref()
+ .chunks(1024 * 1024)
+ .map(|c| Ok(Bytes::copy_from_slice(c)))
+ .chain(std::iter::once(Err(StoreError::Other(
+ "injected mid-body failure".to_owned(),
+ ))))
+ .collect();
+ Box::pin(futures::stream::iter(pieces))
+}
+
+async fn read_all(stream: ByteStream) -> Vec {
+ let chunks: Vec = stream.try_collect().await.expect("stream reads cleanly");
+ chunks.concat()
+}
+
+async fn list_keys(store: &dyn BackupStore, prefix: &str) -> Vec {
+ store
+ .list(prefix)
+ .map_ok(|meta| meta.key)
+ .try_collect()
+ .await
+ .expect("list succeeds")
+}
+
+mod suite {
+ use super::*;
+
+ pub async fn round_trip(store: &dyn BackupStore) {
+ let body = b"hello backup store".to_vec();
+ store
+ .put(
+ "account/table/backup/object.json",
+ chunked_body(body.clone()),
+ )
+ .await
+ .expect("put succeeds");
+
+ let read = read_all(
+ store
+ .get("account/table/backup/object.json")
+ .await
+ .expect("get succeeds"),
+ )
+ .await;
+ assert_eq!(read, body);
+
+ let meta = store
+ .head("account/table/backup/object.json")
+ .await
+ .expect("head succeeds")
+ .expect("object exists");
+ assert_eq!(meta.key, "account/table/backup/object.json");
+ assert_eq!(meta.size, body.len() as u64);
+ assert!(meta.last_modified.is_some(), "both stores report mtime");
+ }
+
+ pub async fn overwrite_replaces_content(store: &dyn BackupStore) {
+ store
+ .put("over/write.bin", chunked_body(b"first"))
+ .await
+ .unwrap();
+ store
+ .put("over/write.bin", chunked_body(b"second and longer"))
+ .await
+ .unwrap();
+ let read = read_all(store.get("over/write.bin").await.unwrap()).await;
+ assert_eq!(read, b"second and longer");
+ }
+
+ pub async fn empty_body_round_trips(store: &dyn BackupStore) {
+ store
+ .put("empty/object", chunked_body(Vec::new()))
+ .await
+ .unwrap();
+ let read = read_all(store.get("empty/object").await.unwrap()).await;
+ assert!(read.is_empty());
+ let meta = store.head("empty/object").await.unwrap().unwrap();
+ assert_eq!(meta.size, 0);
+ }
+
+ pub async fn list_with_and_without_prefix(store: &dyn BackupStore) {
+ for key in ["a/b/1", "a/b/2", "a/c/3", "ab/4", "d"] {
+ store.put(key, chunked_body(vec![1, 2, 3])).await.unwrap();
+ }
+
+ assert_eq!(
+ list_keys(store, "").await,
+ vec!["a/b/1", "a/b/2", "a/c/3", "ab/4", "d"],
+ "empty prefix lists everything in key order"
+ );
+ assert_eq!(list_keys(store, "a").await, vec!["a/b/1", "a/b/2", "a/c/3"]);
+ assert_eq!(list_keys(store, "a/b").await, vec!["a/b/1", "a/b/2"]);
+ assert_eq!(
+ list_keys(store, "a/b/").await,
+ vec!["a/b/1", "a/b/2"],
+ "a trailing slash means the same prefix"
+ );
+ assert_eq!(
+ list_keys(store, "d").await,
+ vec!["d"],
+ "a prefix equal to a key matches that key"
+ );
+ assert!(
+ list_keys(store, "a/b/1/x").await.is_empty(),
+ "prefix under a leaf matches nothing"
+ );
+ assert!(
+ list_keys(store, "nope").await.is_empty(),
+ "unmatched prefix lists nothing"
+ );
+ }
+
+ pub async fn prefix_matching_is_component_aligned(store: &dyn BackupStore) {
+ store.put("pre/fix", chunked_body(vec![1])).await.unwrap();
+ store
+ .put("pre/fixture", chunked_body(vec![2]))
+ .await
+ .unwrap();
+ assert_eq!(
+ list_keys(store, "pre/fix").await,
+ vec!["pre/fix"],
+ "string-prefix sibling must not match"
+ );
+ assert_eq!(
+ store.delete_prefix("pre/fix").await.unwrap(),
+ 1,
+ "delete_prefix uses the same matching rule"
+ );
+ assert_eq!(list_keys(store, "pre").await, vec!["pre/fixture"]);
+ }
+
+ pub async fn delete_prefix_removes_and_counts(store: &dyn BackupStore) {
+ for key in ["del/a/1", "del/a/2", "del/b/3", "keep/4"] {
+ store.put(key, chunked_body(vec![9])).await.unwrap();
+ }
+ assert_eq!(store.delete_prefix("del").await.unwrap(), 3);
+ assert!(list_keys(store, "del").await.is_empty());
+ assert_eq!(list_keys(store, "keep").await, vec!["keep/4"]);
+ assert_eq!(
+ store.delete_prefix("del").await.unwrap(),
+ 0,
+ "a prefix that matches nothing deletes nothing"
+ );
+ }
+
+ pub async fn missing_object_is_not_found(store: &dyn BackupStore) {
+ let err = store.get("no/such/object").await.err().expect("get fails");
+ assert!(matches!(err, StoreError::NotFound), "{err}");
+ let head = store.head("no/such/object").await.expect("head succeeds");
+ assert!(head.is_none());
+ }
+
+ pub async fn invalid_keys_are_rejected(store: &dyn BackupStore) {
+ let bad_keys: &[&str] = &[
+ "",
+ "/",
+ "/abs",
+ "trail/",
+ "dou//ble",
+ ".",
+ "..",
+ "dot/./inner",
+ "dot/../escape",
+ "../up",
+ "up/..",
+ "back\\slash",
+ "seg/back\\slash",
+ "ctl\x00char",
+ "ctl/\x1f",
+ ];
+ for key in bad_keys {
+ let err = store
+ .put(key, chunked_body(vec![1]))
+ .await
+ .err()
+ .unwrap_or_else(|| panic!("put must reject {key:?}"));
+ assert!(
+ matches!(err, StoreError::InvalidKey(_)),
+ "put {key:?}: {err}"
+ );
+
+ let err = store
+ .get(key)
+ .await
+ .err()
+ .unwrap_or_else(|| panic!("get must reject {key:?}"));
+ assert!(
+ matches!(err, StoreError::InvalidKey(_)),
+ "get {key:?}: {err}"
+ );
+
+ let err = store
+ .head(key)
+ .await
+ .err()
+ .unwrap_or_else(|| panic!("head must reject {key:?}"));
+ assert!(
+ matches!(err, StoreError::InvalidKey(_)),
+ "head {key:?}: {err}"
+ );
+ }
+
+ let overlong = "a/".repeat(600);
+ let err = store
+ .put(overlong.trim_end_matches('/'), chunked_body(vec![1]))
+ .await
+ .expect_err("overlong key rejected");
+ assert!(matches!(err, StoreError::InvalidKey(_)), "{err}");
+
+ for bad_prefix in ["/", "dou//ble", "dot/../escape", "ctl\x02"] {
+ let err = store
+ .list(bad_prefix)
+ .try_collect::>()
+ .await
+ .err()
+ .unwrap_or_else(|| panic!("list must reject {bad_prefix:?}"));
+ assert!(matches!(err, StoreError::InvalidKey(_)), "{err}");
+ let err = store
+ .delete_prefix(bad_prefix)
+ .await
+ .err()
+ .unwrap_or_else(|| panic!("delete_prefix must reject {bad_prefix:?}"));
+ assert!(matches!(err, StoreError::InvalidKey(_)), "{err}");
+ }
+ }
+
+ /// A body larger than the part size: multipart on S3, plain streaming on
+ /// the filesystem. Content equality proves part order.
+ pub async fn body_larger_than_part_size(store: &dyn BackupStore) {
+ let len = usize::try_from(PART_SIZE).unwrap() * 2 + 4096;
+ let body = patterned_bytes(len);
+ store
+ .put("multi/part.bin", chunked_body(body.clone()))
+ .await
+ .expect("large put succeeds");
+ let meta = store.head("multi/part.bin").await.unwrap().unwrap();
+ assert_eq!(meta.size, len as u64);
+ let read = read_all(store.get("multi/part.bin").await.unwrap()).await;
+ assert_eq!(read.len(), body.len());
+ assert_eq!(read, body, "reassembled content must match exactly");
+ }
+
+ /// A put whose body fails midway must surface the error and leave nothing
+ /// under the key.
+ pub async fn failed_put_leaves_nothing(store: &dyn BackupStore) {
+ let good = patterned_bytes(usize::try_from(PART_SIZE).unwrap() + 4096);
+ let err = store
+ .put("aborted/upload.bin", failing_body(good))
+ .await
+ .expect_err("mid-body failure surfaces");
+ assert!(
+ matches!(&err, StoreError::Other(msg) if msg.contains("injected")),
+ "the injected error must propagate, got: {err}"
+ );
+ assert!(
+ store.head("aborted/upload.bin").await.unwrap().is_none(),
+ "a failed put must leave nothing under the key"
+ );
+ assert!(list_keys(store, "aborted").await.is_empty());
+ }
+}
+
+mod filesystem {
+ use super::*;
+ use extenddb_backup_store::FilesystemStore;
+
+ async fn fs_store() -> (tempfile::TempDir, FilesystemStore) {
+ let dir = tempfile::tempdir().expect("tempdir");
+ let store = FilesystemStore::open(dir.path()).await.expect("open");
+ (dir, store)
+ }
+
+ #[tokio::test]
+ async fn round_trip() {
+ let (_dir, store) = fs_store().await;
+ suite::round_trip(&store).await;
+ }
+
+ #[tokio::test]
+ async fn overwrite_replaces_content() {
+ let (_dir, store) = fs_store().await;
+ suite::overwrite_replaces_content(&store).await;
+ }
+
+ #[tokio::test]
+ async fn empty_body_round_trips() {
+ let (_dir, store) = fs_store().await;
+ suite::empty_body_round_trips(&store).await;
+ }
+
+ #[tokio::test]
+ async fn list_with_and_without_prefix() {
+ let (_dir, store) = fs_store().await;
+ suite::list_with_and_without_prefix(&store).await;
+ }
+
+ #[tokio::test]
+ async fn prefix_matching_is_component_aligned() {
+ let (_dir, store) = fs_store().await;
+ suite::prefix_matching_is_component_aligned(&store).await;
+ }
+
+ #[tokio::test]
+ async fn delete_prefix_removes_and_counts() {
+ let (_dir, store) = fs_store().await;
+ suite::delete_prefix_removes_and_counts(&store).await;
+ }
+
+ #[tokio::test]
+ async fn missing_object_is_not_found() {
+ let (_dir, store) = fs_store().await;
+ suite::missing_object_is_not_found(&store).await;
+ }
+
+ #[tokio::test]
+ async fn invalid_keys_are_rejected() {
+ let (_dir, store) = fs_store().await;
+ suite::invalid_keys_are_rejected(&store).await;
+ }
+
+ #[tokio::test]
+ async fn body_larger_than_part_size() {
+ let (_dir, store) = fs_store().await;
+ suite::body_larger_than_part_size(&store).await;
+ }
+
+ #[tokio::test]
+ async fn failed_put_leaves_nothing() {
+ let (_dir, store) = fs_store().await;
+ suite::failed_put_leaves_nothing(&store).await;
+ // The rename-into-place discipline: a failed put must also leave no
+ // temporary file behind anywhere under the root.
+ let (dir, store) = fs_store().await;
+ let good = patterned_bytes(64 * 1024);
+ let _ = store.put("aborted/again.bin", failing_body(good)).await;
+ let mut stack = vec![dir.path().to_path_buf()];
+ while let Some(current) = stack.pop() {
+ for entry in std::fs::read_dir(¤t).expect("read_dir") {
+ let entry = entry.expect("entry");
+ if entry.file_type().expect("file_type").is_dir() {
+ stack.push(entry.path());
+ } else {
+ panic!("stray file after failed put: {}", entry.path().display());
+ }
+ }
+ }
+ }
+
+ #[tokio::test]
+ async fn validate_probes_the_root() {
+ let (_dir, store) = fs_store().await;
+ store.validate().await.expect("writable root validates");
+ }
+
+ #[tokio::test]
+ async fn validate_rejects_an_unwritable_root() {
+ use std::os::unix::fs::PermissionsExt;
+ let dir = tempfile::tempdir().expect("tempdir");
+ let store = FilesystemStore::open(dir.path()).await.expect("open");
+ std::fs::set_permissions(dir.path(), std::fs::Permissions::from_mode(0o500))
+ .expect("chmod");
+ let result = store.validate().await;
+ std::fs::set_permissions(dir.path(), std::fs::Permissions::from_mode(0o700))
+ .expect("chmod back");
+ let err = result.expect_err("unwritable root fails validation");
+ assert!(matches!(err, StoreError::PermissionDenied(_)), "{err}");
+ }
+
+ #[tokio::test]
+ async fn open_refuses_a_missing_root() {
+ let dir = tempfile::tempdir().expect("tempdir");
+ let missing = dir.path().join("does-not-exist");
+ assert!(FilesystemStore::open(&missing).await.is_err());
+ }
+
+ #[tokio::test]
+ async fn symlinked_directory_escape_is_refused() {
+ let outside = tempfile::tempdir().expect("outside dir");
+ std::fs::write(outside.path().join("secret.txt"), b"outside data").unwrap();
+
+ let (dir, store) = fs_store().await;
+ std::os::unix::fs::symlink(outside.path(), dir.path().join("esc")).unwrap();
+
+ let err = store
+ .get("esc/secret.txt")
+ .await
+ .err()
+ .expect("get refused");
+ assert!(matches!(err, StoreError::PermissionDenied(_)), "{err}");
+
+ let err = store
+ .head("esc/secret.txt")
+ .await
+ .expect_err("head refused");
+ assert!(matches!(err, StoreError::PermissionDenied(_)), "{err}");
+
+ let err = store
+ .put("esc/new-file", chunked_body(vec![1]))
+ .await
+ .expect_err("put refused");
+ assert!(matches!(err, StoreError::PermissionDenied(_)), "{err}");
+ assert!(
+ !outside.path().join("new-file").exists(),
+ "nothing may be written outside the root"
+ );
+ }
+
+ #[tokio::test]
+ async fn symlinked_file_escape_is_refused() {
+ let outside = tempfile::tempdir().expect("outside dir");
+ let target = outside.path().join("secret.txt");
+ std::fs::write(&target, b"outside data").unwrap();
+
+ let (dir, store) = fs_store().await;
+ std::os::unix::fs::symlink(&target, dir.path().join("leak.json")).unwrap();
+
+ let err = store.get("leak.json").await.err().expect("get refused");
+ assert!(matches!(err, StoreError::PermissionDenied(_)), "{err}");
+
+ let err = store
+ .put("leak.json", chunked_body(vec![1]))
+ .await
+ .expect_err("put through an escaping symlink refused");
+ assert!(matches!(err, StoreError::PermissionDenied(_)), "{err}");
+ assert_eq!(
+ std::fs::read(&target).unwrap(),
+ b"outside data",
+ "the symlink target must be untouched"
+ );
+ }
+
+ /// An in-root directory symlink aliases one key onto another: a put
+ /// through the alias would land under the victim's prefix while list and
+ /// `delete_prefix` on the alias see nothing. The store refuses the alias
+ /// on every key-resolving operation, so all operations agree on what a
+ /// key names.
+ #[cfg(unix)]
+ #[tokio::test]
+ async fn in_root_directory_symlink_is_refused() {
+ let (dir, store) = fs_store().await;
+ store
+ .put("victim/sub/obj", chunked_body(vec![1]))
+ .await
+ .unwrap();
+ std::os::unix::fs::symlink(dir.path().join("victim"), dir.path().join("alias")).unwrap();
+
+ let err = store
+ .put("alias/sub/obj2", chunked_body(vec![2]))
+ .await
+ .expect_err("put through an in-root symlink refused");
+ assert!(matches!(err, StoreError::PermissionDenied(_)), "{err}");
+ assert!(
+ !dir.path().join("victim/sub/obj2").exists(),
+ "nothing may land under the aliased prefix"
+ );
+
+ let err = store
+ .get("alias/sub/obj")
+ .await
+ .err()
+ .expect("get through an in-root symlink refused");
+ assert!(matches!(err, StoreError::PermissionDenied(_)), "{err}");
+
+ let err = store
+ .head("alias/sub/obj")
+ .await
+ .expect_err("head through an in-root symlink refused");
+ assert!(matches!(err, StoreError::PermissionDenied(_)), "{err}");
+
+ assert!(
+ list_keys(&store, "alias").await.is_empty(),
+ "the aliased prefix lists nothing"
+ );
+
+ let deleted = store.delete_prefix("alias").await.unwrap();
+ assert_eq!(deleted, 0, "delete through the alias removes nothing");
+ assert_eq!(
+ read_all(store.get("victim/sub/obj").await.unwrap()).await,
+ vec![1],
+ "the object under its real key survives"
+ );
+ }
+
+ /// An in-root file symlink planted at a key would redirect get and head
+ /// to another key's content, and a put over it would resolve through the
+ /// link. All three are refused.
+ #[cfg(unix)]
+ #[tokio::test]
+ async fn in_root_file_symlink_is_refused() {
+ let (dir, store) = fs_store().await;
+ store
+ .put("victim/data", chunked_body(vec![7]))
+ .await
+ .unwrap();
+ std::os::unix::fs::symlink(dir.path().join("victim/data"), dir.path().join("planted"))
+ .unwrap();
+
+ let err = store.get("planted").await.err().expect("get refused");
+ assert!(matches!(err, StoreError::PermissionDenied(_)), "{err}");
+
+ let err = store.head("planted").await.expect_err("head refused");
+ assert!(matches!(err, StoreError::PermissionDenied(_)), "{err}");
+
+ let err = store
+ .put("planted", chunked_body(vec![8]))
+ .await
+ .expect_err("put over an in-root file symlink refused");
+ assert!(matches!(err, StoreError::PermissionDenied(_)), "{err}");
+ assert_eq!(
+ read_all(store.get("victim/data").await.unwrap()).await,
+ vec![7],
+ "the symlink target must be untouched"
+ );
+ }
+
+ #[tokio::test]
+ async fn symlinks_are_invisible_to_list() {
+ let outside = tempfile::tempdir().expect("outside dir");
+ std::fs::write(outside.path().join("secret.txt"), b"x").unwrap();
+
+ let (dir, store) = fs_store().await;
+ store
+ .put("real/object", chunked_body(vec![1]))
+ .await
+ .unwrap();
+ std::os::unix::fs::symlink(outside.path(), dir.path().join("esc")).unwrap();
+
+ assert_eq!(list_keys(&store, "").await, vec!["real/object"]);
+ }
+
+ /// A put through a planted directory symlink: `create_dir_all` follows
+ /// symlinks in existing components, so a naive implementation creates
+ /// parent directories outside the root before its escape check refuses
+ /// the put. The store must check the deepest existing ancestor before
+ /// creating anything.
+ #[tokio::test]
+ async fn multi_level_put_under_directory_symlink_creates_nothing_outside() {
+ let outside = tempfile::tempdir().expect("outside dir");
+ let (dir, store) = fs_store().await;
+ std::os::unix::fs::symlink(outside.path(), dir.path().join("esc")).unwrap();
+
+ let err = store
+ .put("esc/sub/deeper/file", chunked_body(vec![1]))
+ .await
+ .expect_err("put through an escaping symlink refused");
+ assert!(matches!(err, StoreError::PermissionDenied(_)), "{err}");
+ assert!(
+ !outside.path().join("sub").exists(),
+ "no directory may be created outside the root"
+ );
+ assert_eq!(
+ std::fs::read_dir(outside.path()).unwrap().count(),
+ 0,
+ "the symlink target must be untouched"
+ );
+ }
+
+ /// Same shape with a file symlink at the parent position: the walk stops
+ /// at the symlink, which resolves outside the root, and nothing is
+ /// created or modified on the far side.
+ #[tokio::test]
+ async fn multi_level_put_under_file_symlink_creates_nothing_outside() {
+ let outside = tempfile::tempdir().expect("outside dir");
+ let target = outside.path().join("secret.txt");
+ std::fs::write(&target, b"outside data").unwrap();
+
+ let (dir, store) = fs_store().await;
+ std::os::unix::fs::symlink(&target, dir.path().join("leak")).unwrap();
+
+ let err = store
+ .put("leak/sub/file", chunked_body(vec![1]))
+ .await
+ .expect_err("put through an escaping file symlink refused");
+ assert!(matches!(err, StoreError::PermissionDenied(_)), "{err}");
+ assert_eq!(
+ std::fs::read(&target).unwrap(),
+ b"outside data",
+ "the symlink target must be untouched"
+ );
+ assert_eq!(
+ std::fs::read_dir(outside.path()).unwrap().count(),
+ 1,
+ "nothing may be created outside the root"
+ );
+ }
+
+ /// An in-root regular file at an ancestor of the key is a collision, not
+ /// an escape: the put fails with an I/O error and creates nothing.
+ #[tokio::test]
+ async fn put_under_an_existing_object_is_refused() {
+ let (_dir, store) = fs_store().await;
+ store.put("plain", chunked_body(vec![1])).await.unwrap();
+ let err = store
+ .put("plain/sub/file", chunked_body(vec![2]))
+ .await
+ .expect_err("ancestor collision refused");
+ assert!(matches!(err, StoreError::Io { .. }), "{err}");
+ assert_eq!(read_all(store.get("plain").await.unwrap()).await, vec![1]);
+ }
+}
+
+mod s3 {
+ use super::*;
+ use extenddb_backup_store::{S3Store, S3StoreConfig};
+
+ const SKIP: &str =
+ "skipping: EXTENDDB_TEST_S3_ENDPOINT is not set; see crates/backup-store/README.md";
+
+ /// The test harness captures `eprintln!`, so a skipped S3 suite would be
+ /// invisible in default output. Writing to the raw stderr handle bypasses
+ /// capture and keeps the skip visible.
+ fn announce_skip() {
+ use std::io::Write;
+ let _ = writeln!(std::io::stderr(), "{SKIP}");
+ }
+
+ /// Build a store against the local endpoint, or `None` (with a printed
+ /// reason) when the environment does not provide one. Each test gets a
+ /// unique bucket prefix so runs never collide.
+ async fn store() -> Option {
+ let Ok(endpoint) = std::env::var("EXTENDDB_TEST_S3_ENDPOINT") else {
+ announce_skip();
+ return None;
+ };
+ let bucket = std::env::var("EXTENDDB_TEST_S3_BUCKET")
+ .unwrap_or_else(|_| "extenddb-backup-store-tests".to_owned());
+ let region =
+ std::env::var("EXTENDDB_TEST_S3_REGION").unwrap_or_else(|_| "us-east-1".to_owned());
+ let config = S3StoreConfig {
+ bucket: bucket.clone(),
+ prefix: format!("conformance-{}/", uuid::Uuid::new_v4()),
+ region: Some(region),
+ endpoint: Some(endpoint),
+ force_path_style: true,
+ upload_part_size_bytes: PART_SIZE,
+ max_concurrent_uploads: 4,
+ };
+ let store = S3Store::open(&config).await.expect("S3Store::open");
+ ensure_bucket(&config).await;
+ Some(store)
+ }
+
+ /// Create the test bucket when it does not exist yet, with a raw client
+ /// so the check stays independent of the store under test.
+ async fn ensure_bucket(config: &S3StoreConfig) {
+ let client = raw_client(config).await;
+ match client.create_bucket().bucket(&config.bucket).send().await {
+ Ok(_) => {}
+ Err(err) => {
+ let service = err.as_service_error();
+ let exists = service.is_some_and(|e| {
+ e.is_bucket_already_owned_by_you() || e.is_bucket_already_exists()
+ });
+ assert!(exists, "create_bucket failed: {err}");
+ }
+ }
+ }
+
+ async fn raw_client(config: &S3StoreConfig) -> aws_sdk_s3::Client {
+ let mut loader = aws_config::defaults(aws_config::BehaviorVersion::latest());
+ if let Some(region) = &config.region {
+ loader = loader.region(aws_config::Region::new(region.clone()));
+ }
+ if let Some(endpoint) = &config.endpoint {
+ loader = loader.endpoint_url(endpoint.clone());
+ }
+ let shared = loader.load().await;
+ let builder = aws_sdk_s3::config::Builder::from(&shared).force_path_style(true);
+ aws_sdk_s3::Client::from_conf(builder.build())
+ }
+
+ #[tokio::test]
+ async fn round_trip() {
+ let Some(store) = store().await else { return };
+ suite::round_trip(&store).await;
+ }
+
+ #[tokio::test]
+ async fn overwrite_replaces_content() {
+ let Some(store) = store().await else { return };
+ suite::overwrite_replaces_content(&store).await;
+ }
+
+ #[tokio::test]
+ async fn empty_body_round_trips() {
+ let Some(store) = store().await else { return };
+ suite::empty_body_round_trips(&store).await;
+ }
+
+ #[tokio::test]
+ async fn list_with_and_without_prefix() {
+ let Some(store) = store().await else { return };
+ suite::list_with_and_without_prefix(&store).await;
+ }
+
+ #[tokio::test]
+ async fn prefix_matching_is_component_aligned() {
+ let Some(store) = store().await else { return };
+ suite::prefix_matching_is_component_aligned(&store).await;
+ }
+
+ #[tokio::test]
+ async fn delete_prefix_removes_and_counts() {
+ let Some(store) = store().await else { return };
+ suite::delete_prefix_removes_and_counts(&store).await;
+ }
+
+ #[tokio::test]
+ async fn missing_object_is_not_found() {
+ let Some(store) = store().await else { return };
+ suite::missing_object_is_not_found(&store).await;
+ }
+
+ #[tokio::test]
+ async fn invalid_keys_are_rejected() {
+ let Some(store) = store().await else { return };
+ suite::invalid_keys_are_rejected(&store).await;
+ }
+
+ #[tokio::test]
+ async fn body_larger_than_part_size() {
+ let Some(store) = store().await else { return };
+ suite::body_larger_than_part_size(&store).await;
+ }
+
+ #[tokio::test]
+ async fn failed_put_aborts_the_multipart_upload() {
+ let Ok(endpoint) = std::env::var("EXTENDDB_TEST_S3_ENDPOINT") else {
+ announce_skip();
+ return;
+ };
+ let bucket = std::env::var("EXTENDDB_TEST_S3_BUCKET")
+ .unwrap_or_else(|_| "extenddb-backup-store-tests".to_owned());
+ let region =
+ std::env::var("EXTENDDB_TEST_S3_REGION").unwrap_or_else(|_| "us-east-1".to_owned());
+ let prefix = format!("conformance-{}/", uuid::Uuid::new_v4());
+ let config = S3StoreConfig {
+ bucket: bucket.clone(),
+ prefix: prefix.clone(),
+ region: Some(region),
+ endpoint: Some(endpoint),
+ force_path_style: true,
+ upload_part_size_bytes: PART_SIZE,
+ max_concurrent_uploads: 4,
+ };
+ let store = S3Store::open(&config).await.expect("S3Store::open");
+ ensure_bucket(&config).await;
+
+ suite::failed_put_leaves_nothing(&store).await;
+
+ // The store-level check proved no object landed; this proves the
+ // multipart upload itself was aborted rather than left in progress.
+ let client = raw_client(&config).await;
+ let uploads = client
+ .list_multipart_uploads()
+ .bucket(&bucket)
+ .prefix(&prefix)
+ .send()
+ .await
+ .expect("list_multipart_uploads");
+ assert!(
+ uploads.uploads().is_empty(),
+ "a failed put must abort its multipart upload: {:?}",
+ uploads.uploads()
+ );
+ }
+
+ /// An interceptor that fails `CompleteMultipartUpload`, so the abort path
+ /// after a complete-time failure is exercised against a real endpoint.
+ #[derive(Debug)]
+ struct FailCompleteMultipartUpload;
+
+ impl aws_sdk_s3::config::Intercept for FailCompleteMultipartUpload {
+ fn name(&self) -> &'static str {
+ "FailCompleteMultipartUpload"
+ }
+
+ fn read_before_execution(
+ &self,
+ context: &aws_sdk_s3::config::interceptors::BeforeSerializationInterceptorContextRef<
+ '_,
+ >,
+ _cfg: &mut aws_sdk_s3::config::ConfigBag,
+ ) -> Result<(), aws_sdk_s3::error::BoxError> {
+ let is_complete = context
+ .input()
+ .downcast_ref::()
+ .is_some();
+ if is_complete {
+ return Err("injected CompleteMultipartUpload failure".into());
+ }
+ Ok(())
+ }
+ }
+
+ #[tokio::test]
+ async fn failed_complete_aborts_the_multipart_upload() {
+ let Ok(endpoint) = std::env::var("EXTENDDB_TEST_S3_ENDPOINT") else {
+ announce_skip();
+ return;
+ };
+ let bucket = std::env::var("EXTENDDB_TEST_S3_BUCKET")
+ .unwrap_or_else(|_| "extenddb-backup-store-tests".to_owned());
+ let region =
+ std::env::var("EXTENDDB_TEST_S3_REGION").unwrap_or_else(|_| "us-east-1".to_owned());
+ let prefix = format!("conformance-{}/", uuid::Uuid::new_v4());
+ let config = S3StoreConfig {
+ bucket: bucket.clone(),
+ prefix: prefix.clone(),
+ region: Some(region),
+ endpoint: Some(endpoint),
+ force_path_style: true,
+ upload_part_size_bytes: PART_SIZE,
+ max_concurrent_uploads: 4,
+ };
+ let store = S3Store::open_with_interceptor(&config, FailCompleteMultipartUpload)
+ .await
+ .expect("S3Store::open_with_interceptor");
+ ensure_bucket(&config).await;
+
+ let body = patterned_bytes(usize::try_from(PART_SIZE).unwrap() * 2 + 4096);
+ let err = store
+ .put("complete/fails.bin", chunked_body(body))
+ .await
+ .expect_err("injected complete failure surfaces");
+ assert!(
+ err.to_string().contains("injected") || matches!(err, StoreError::Transport { .. }),
+ "the injected failure must propagate: {err}"
+ );
+ assert!(
+ store.head("complete/fails.bin").await.unwrap().is_none(),
+ "no object may exist after a failed complete"
+ );
+
+ let client = raw_client(&config).await;
+ let uploads = client
+ .list_multipart_uploads()
+ .bucket(&bucket)
+ .prefix(&prefix)
+ .send()
+ .await
+ .expect("list_multipart_uploads");
+ assert!(
+ uploads.uploads().is_empty(),
+ "a failed complete must abort its multipart upload: {:?}",
+ uploads.uploads()
+ );
+ }
+
+ #[tokio::test]
+ async fn validate_calls_head_bucket() {
+ let Some(store) = store().await else { return };
+ store.validate().await.expect("existing bucket validates");
+ }
+
+ #[tokio::test]
+ async fn validate_names_a_missing_bucket() {
+ let Ok(endpoint) = std::env::var("EXTENDDB_TEST_S3_ENDPOINT") else {
+ announce_skip();
+ return;
+ };
+ let region =
+ std::env::var("EXTENDDB_TEST_S3_REGION").unwrap_or_else(|_| "us-east-1".to_owned());
+ let missing = format!("extenddb-no-such-bucket-{}", uuid::Uuid::new_v4());
+ let config = S3StoreConfig {
+ bucket: missing.clone(),
+ prefix: String::new(),
+ region: Some(region),
+ endpoint: Some(endpoint),
+ force_path_style: true,
+ upload_part_size_bytes: PART_SIZE,
+ max_concurrent_uploads: 4,
+ };
+ let store = S3Store::open(&config).await.expect("S3Store::open");
+ let err = store.validate().await.expect_err("missing bucket fails");
+ assert!(
+ err.to_string().contains(&missing),
+ "the message must name the bucket: {err}"
+ );
+ }
+}