diff --git a/Cargo.lock b/Cargo.lock index 2f35c5df..b543deba 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -20,7 +20,7 @@ version = "0.5.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d122413f284cf2d62fb1b7db97e02edb8cda96d769b16e443a4f6195e35662b0" dependencies = [ - "crypto-common", + "crypto-common 0.1.7", "generic-array", ] @@ -32,7 +32,7 @@ checksum = "b169f7a6d4742236a0a00c541b845991d0ac43e546831af1249753ab4c3aa3a0" dependencies = [ "cfg-if", "cipher", - "cpufeatures", + "cpufeatures 0.2.17", ] [[package]] @@ -132,7 +132,7 @@ checksum = "3c3610892ee6e0cbce8ae2700349fcf8f98adb0dbfbee85aec3c9179d29cc072" dependencies = [ "base64ct", "blake2", - "cpufeatures", + "cpufeatures 0.2.17", "password-hash", ] @@ -161,9 +161,9 @@ dependencies = [ [[package]] name = "async-compression" -version = "0.4.37" +version = "0.4.41" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d10e4f991a553474232bc0a31799f6d24b034a84c0971d80d2e2f78b2e576e40" +checksum = "d0f9ee0f6e02ffd7ad5816e9464499fba7b3effd01123b515c41d1697c43dad1" dependencies = [ "compression-codecs", "compression-core", @@ -264,8 +264,8 @@ dependencies = [ "aws-sdk-ssooidc", "aws-sdk-sts", "aws-smithy-async", - "aws-smithy-http 0.63.6", - "aws-smithy-json 0.62.5", + "aws-smithy-http", + "aws-smithy-json", "aws-smithy-runtime", "aws-smithy-runtime-api", "aws-smithy-types", @@ -326,7 +326,7 @@ dependencies = [ "aws-sigv4", "aws-smithy-async", "aws-smithy-eventstream", - "aws-smithy-http 0.63.6", + "aws-smithy-http", "aws-smithy-runtime", "aws-smithy-runtime-api", "aws-smithy-types", @@ -346,9 +346,9 @@ dependencies = [ [[package]] name = "aws-sdk-s3" -version = "1.119.0" +version = "1.129.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1d65fddc3844f902dfe1864acb8494db5f9342015ee3ab7890270d36fbd2e01c" +checksum = "6d4e8410fadbc0ee453145dd77a4958227b18b05bf67c2795d0a8b8596c9aa0f" dependencies = [ "aws-credential-types", "aws-runtime", @@ -356,8 +356,9 @@ dependencies = [ "aws-smithy-async", "aws-smithy-checksums", "aws-smithy-eventstream", - "aws-smithy-http 0.62.6", - "aws-smithy-json 0.61.9", + "aws-smithy-http", + "aws-smithy-json", + "aws-smithy-observability", "aws-smithy-runtime", "aws-smithy-runtime-api", "aws-smithy-types", @@ -369,11 +370,11 @@ dependencies = [ "hmac", "http 0.2.12", "http 1.4.0", - "http-body 0.4.6", + "http-body 1.0.1", "lru", "percent-encoding", "regex-lite", - "sha2", + "sha2 0.10.9", "tracing", "url", ] @@ -387,8 +388,8 @@ dependencies = [ "aws-credential-types", "aws-runtime", "aws-smithy-async", - "aws-smithy-http 0.63.6", - "aws-smithy-json 0.62.5", + "aws-smithy-http", + "aws-smithy-json", "aws-smithy-observability", "aws-smithy-runtime", "aws-smithy-runtime-api", @@ -411,8 +412,8 @@ dependencies = [ "aws-credential-types", "aws-runtime", "aws-smithy-async", - "aws-smithy-http 0.63.6", - "aws-smithy-json 0.62.5", + "aws-smithy-http", + "aws-smithy-json", "aws-smithy-observability", "aws-smithy-runtime", "aws-smithy-runtime-api", @@ -435,8 +436,8 @@ dependencies = [ "aws-credential-types", "aws-runtime", "aws-smithy-async", - "aws-smithy-http 0.63.6", - "aws-smithy-json 0.62.5", + "aws-smithy-http", + "aws-smithy-json", "aws-smithy-observability", "aws-smithy-query", "aws-smithy-runtime", @@ -459,7 +460,7 @@ checksum = "b0b660013a6683ab23797778e21f1f854744fdf05f68204b4cca4c8c04b5d1f4" dependencies = [ "aws-credential-types", "aws-smithy-eventstream", - "aws-smithy-http 0.63.6", + "aws-smithy-http", "aws-smithy-runtime-api", "aws-smithy-types", "bytes", @@ -472,7 +473,7 @@ dependencies = [ "p256 0.11.1", "percent-encoding", "ring", - "sha2", + "sha2 0.10.9", "subtle", "time", "tracing", @@ -492,21 +493,22 @@ dependencies = [ [[package]] name = "aws-smithy-checksums" -version = "0.63.12" +version = "0.64.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "87294a084b43d649d967efe58aa1f9e0adc260e13a6938eb904c0ae9b45824ae" +checksum = "6750f3dd509b0694a4377f0293ed2f9630d710b1cebe281fa8bac8f099f88bc6" dependencies = [ - "aws-smithy-http 0.62.6", + "aws-smithy-http", "aws-smithy-types", "bytes", "crc-fast", "hex", - "http 0.2.12", - "http-body 0.4.6", - "md-5", + "http 1.4.0", + "http-body 1.0.1", + "http-body-util", + "md-5 0.10.6", "pin-project-lite", "sha1", - "sha2", + "sha2 0.10.9", "tracing", ] @@ -521,34 +523,13 @@ dependencies = [ "crc32fast", ] -[[package]] -name = "aws-smithy-http" -version = "0.62.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "826141069295752372f8203c17f28e30c464d22899a43a0c9fd9c458d469c88b" -dependencies = [ - "aws-smithy-eventstream", - "aws-smithy-runtime-api", - "aws-smithy-types", - "bytes", - "bytes-utils", - "futures-core", - "futures-util", - "http 0.2.12", - "http 1.4.0", - "http-body 0.4.6", - "percent-encoding", - "pin-project-lite", - "pin-utils", - "tracing", -] - [[package]] name = "aws-smithy-http" version = "0.63.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ba1ab2dc1c2c3749ead27180d333c42f11be8b0e934058fb4b2258ee8dbe5231" dependencies = [ + "aws-smithy-eventstream", "aws-smithy-runtime-api", "aws-smithy-types", "bytes", @@ -594,15 +575,6 @@ dependencies = [ "tracing", ] -[[package]] -name = "aws-smithy-json" -version = "0.61.9" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "49fa1213db31ac95288d981476f78d05d9cbb0353d22cdf3472cc05bb02f6551" -dependencies = [ - "aws-smithy-types", -] - [[package]] name = "aws-smithy-json" version = "0.62.5" @@ -638,7 +610,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "028999056d2d2fd58a697232f9eec4a643cf73a71cf327690a7edad1d2af2110" dependencies = [ "aws-smithy-async", - "aws-smithy-http 0.63.6", + "aws-smithy-http", "aws-smithy-http-client", "aws-smithy-observability", "aws-smithy-runtime-api", @@ -724,9 +696,9 @@ dependencies = [ [[package]] name = "axum" -version = "0.8.8" +version = "0.8.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8b52af3cb4058c895d37317bb27508dccc8e5f2d39454016b297bf4a400597b8" +checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90" dependencies = [ "axum-core", "axum-macros", @@ -778,9 +750,9 @@ dependencies = [ [[package]] name = "axum-macros" -version = "0.5.0" +version = "0.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "604fde5e028fea851ce1d8570bbdc034bec850d157f7569d10f347d06808c05c" +checksum = "7aa268c23bfbbd2c4363b9cd302a4f504fb2a9dfe7e3451d66f35dd392e20aca" dependencies = [ "proc-macro2", "quote", @@ -810,7 +782,7 @@ dependencies = [ "rustc_version", "serde", "serde_json", - "sha2", + "sha2 0.10.9", "time", "tracing", "url", @@ -949,21 +921,21 @@ version = "0.10.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "46502ad458c9a52b69d4d4d32775c788b7a1b85e8bc9d482d92250fc0e3f8efe" dependencies = [ - "digest", + "digest 0.10.7", ] [[package]] name = "blake3" -version = "1.8.3" +version = "1.8.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2468ef7d57b3fb7e16b576e8377cdbde2320c60e1491e961d11da40fc4f02a2d" +checksum = "4d2d5991425dfd0785aed03aedcf0b321d61975c9b5b3689c774a2610ae0b51e" dependencies = [ "arrayref", "arrayvec", "cc", "cfg-if", "constant_time_eq", - "cpufeatures", + "cpufeatures 0.3.0", "memmap2", "rayon-core", ] @@ -977,6 +949,15 @@ dependencies = [ "generic-array", ] +[[package]] +name = "block-buffer" +version = "0.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cdd35008169921d80bc60d3d0ab416eecb028c4cd653352907921d95084790be" +dependencies = [ + "hybrid-array", +] + [[package]] name = "brotli" version = "8.0.2" @@ -1124,7 +1105,7 @@ version = "0.4.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "773f3b9af64447d2ce9850330c473515014aa235e6a783b02db81ff39e4a3dad" dependencies = [ - "crypto-common", + "crypto-common 0.1.7", "inout", ] @@ -1168,9 +1149,9 @@ dependencies = [ [[package]] name = "compression-codecs" -version = "0.4.36" +version = "0.4.37" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "00828ba6fd27b45a448e57dbfe84f1029d4c9f26b368157e9a448a5f49a2ec2a" +checksum = "eb7b51a7d9c967fc26773061ba86150f19c50c0d65c887cb1fbe295fd16619b7" dependencies = [ "brotli", "compression-core", @@ -1199,6 +1180,12 @@ version = "0.9.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2459377285ad874054d797f3ccebf984978aa39129f6eafde5cdc8315b612f8" +[[package]] +name = "const-oid" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a6ef517f0926dd24a1582492c791b6a4818a4d94e789a334894aa15b0d12f55c" + [[package]] name = "const-str" version = "0.3.2" @@ -1266,10 +1253,19 @@ dependencies = [ ] [[package]] -name = "crc" -version = "3.4.0" +name = "cpufeatures" +version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5eb8a2a1cd12ab0d987a5d5e825195d372001a4094a0376319d5a0ad71c1ba0d" +checksum = "8b2a41393f66f16b0823bb79094d54ac5fbd34ab292ddafb9a0456ac9f87d201" +dependencies = [ + "libc", +] + +[[package]] +name = "crc" +version = "3.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9710d3b3739c2e349eb44fe848ad0b7c8cb1e42bd87ee49371df2f7acaf3e675" dependencies = [ "crc-catalog", ] @@ -1282,15 +1278,14 @@ checksum = "19d374276b40fb8bbdee95aef7c7fa6b5316ec764510eb64b8dd0e2ed0d7e7f5" [[package]] name = "crc-fast" -version = "1.6.0" +version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6ddc2d09feefeee8bd78101665bd8645637828fa9317f9f292496dbbd8c65ff3" +checksum = "2fd92aca2c6001b1bf5ba0ff84ee74ec8501b52bbef0cac80bf25a6c1d87a83d" dependencies = [ "crc", - "digest", - "rand 0.9.2", - "regex", + "digest 0.10.7", "rustversion", + "spin 0.10.0", ] [[package]] @@ -1380,6 +1375,15 @@ dependencies = [ "typenum", ] +[[package]] +name = "crypto-common" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "77727bb15fa921304124b128af125e7e3b968275d1b108b379190264f4423710" +dependencies = [ + "hybrid-array", +] + [[package]] name = "cssparser" version = "0.33.0" @@ -1428,9 +1432,9 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "97fb8b7c4503de7d6ae7b42ab72a5a59857b4c937ec27a3d4539dba95b5ab2be" dependencies = [ "cfg-if", - "cpufeatures", + "cpufeatures 0.2.17", "curve25519-dalek-derive", - "digest", + "digest 0.10.7", "fiat-crypto", "rustc_version", "subtle", @@ -1496,7 +1500,7 @@ version = "0.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f1a467a65c5e759bce6e65eaf91cc29f466cdc57cb65777bd646872a8a1fd4de" dependencies = [ - "const-oid", + "const-oid 0.9.6", "zeroize", ] @@ -1506,7 +1510,7 @@ version = "0.7.10" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e7c1832837b905bbfb5101e07cc24c8deddf52f93225eee6ead5f4d63d53ddcb" dependencies = [ - "const-oid", + "const-oid 0.9.6", "pem-rfc7468", "zeroize", ] @@ -1527,12 +1531,23 @@ version = "0.10.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" dependencies = [ - "block-buffer", - "const-oid", - "crypto-common", + "block-buffer 0.10.4", + "const-oid 0.9.6", + "crypto-common 0.1.7", "subtle", ] +[[package]] +name = "digest" +version = "0.11.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4850db49bf08e663084f7fb5c87d202ef91a3907271aff24a94eb97ff039153c" +dependencies = [ + "block-buffer 0.12.0", + "const-oid 0.10.2", + "crypto-common 0.2.1", +] + [[package]] name = "displaydoc" version = "0.2.5" @@ -1608,7 +1623,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ee27f32b5c5292967d2d4a9d7f1e0b0aed2c15daded5a60300e4abb9d8020bca" dependencies = [ "der 0.7.10", - "digest", + "digest 0.10.7", "elliptic-curve 0.13.8", "rfc6979 0.4.0", "signature 2.2.0", @@ -1634,7 +1649,7 @@ dependencies = [ "curve25519-dalek", "ed25519", "serde", - "sha2", + "sha2 0.10.9", "subtle", "zeroize", ] @@ -1657,7 +1672,7 @@ dependencies = [ "base16ct 0.1.1", "crypto-bigint 0.4.9", "der 0.6.1", - "digest", + "digest 0.10.7", "ff 0.12.1", "generic-array", "group 0.12.1", @@ -1676,7 +1691,7 @@ checksum = "b5e6043086bf7973472e0c7dff2142ea0b680d30e18d9cc40f267efbf222bd47" dependencies = [ "base16ct 0.2.0", "crypto-bigint 0.5.5", - "digest", + "digest 0.10.7", "ff 0.13.1", "generic-array", "group 0.13.0", @@ -1764,6 +1779,12 @@ dependencies = [ "pin-project-lite", ] +[[package]] +name = "fastcdc" +version = "4.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b2613fe3f26b5958b5cd1cb2a07b58c1f3411adaa16abd6b0882279b1eb08c37" + [[package]] name = "fastrand" version = "1.9.0" @@ -1838,7 +1859,7 @@ checksum = "da0e4dd2a88388a1f4ccc7c9ce104604dab68d9f408dc34cd45823d5a9069095" dependencies = [ "futures-core", "futures-sink", - "spin", + "spin 0.9.8", ] [[package]] @@ -1853,6 +1874,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 = "foreign-types" version = "0.3.2" @@ -2197,7 +2224,7 @@ checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1" dependencies = [ "allocator-api2", "equivalent", - "foldhash", + "foldhash 0.1.5", ] [[package]] @@ -2205,6 +2232,17 @@ 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]] +name = "hashbrown" +version = "0.17.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4f467dd6dccf739c208452f8014c75c18bb8301b050ad1cfb27153803edb0f51" dependencies = [ "allocator-api2", ] @@ -2245,7 +2283,7 @@ version = "0.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6c49c37c09c17a53d937dfbb742eb3a961d65a994e6bcdcf37e7399d0cc8ab5e" dependencies = [ - "digest", + "digest 0.10.7", ] [[package]] @@ -2350,6 +2388,15 @@ version = "1.0.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" +[[package]] +name = "hybrid-array" +version = "0.4.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3944cf8cf766b40e2a1a333ee5e9b563f854d5fa49d6a8ca2764e97c6eddb214" +dependencies = [ + "typenum", +] + [[package]] name = "hyper" version = "0.14.32" @@ -2463,7 +2510,7 @@ dependencies = [ "libc", "percent-encoding", "pin-project-lite", - "socket2 0.6.2", + "socket2 0.5.10", "tokio", "tower-service", "tracing", @@ -2614,9 +2661,9 @@ dependencies = [ [[package]] name = "image" -version = "0.25.9" +version = "0.25.10" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e6506c6c10786659413faa717ceebcb8f70731c0a60cbae39795fdf114519c1a" +checksum = "85ab80394333c02fe689eaf900ab500fbd0c2213da414687ebf995a65d5a6104" dependencies = [ "bytemuck", "byteorder-lite", @@ -2769,16 +2816,16 @@ dependencies = [ "rsa", "serde", "serde_json", - "sha2", + "sha2 0.10.9", "signature 2.2.0", "simple_asn1", ] [[package]] name = "kamadak-exif" -version = "0.5.5" +version = "0.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ef4fc70d0ab7e5b6bafa30216a6b48705ea964cdfc29c050f2412295eba58077" +checksum = "1130d80c7374efad55a117d715a3af9368f0fa7a2c54573afc15a188cd984837" dependencies = [ "mutate_once", ] @@ -2789,7 +2836,7 @@ version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" dependencies = [ - "spin", + "spin 0.9.8", ] [[package]] @@ -2800,9 +2847,9 @@ checksum = "09edd9e8b54e49e587e4f6295a7d29c3ea94d469cb40ab8ca70b288248a81db2" [[package]] name = "libc" -version = "0.2.182" +version = "0.2.185" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6800badb6cb2082ffd7b6a67e6125bb39f18782f793520caee8cb8846be06112" +checksum = "52ff2c0fe9bc6cb6b14a0592c2ff4fa9ceb83eea9db979b0487cd054946a2b8f" [[package]] name = "libm" @@ -2843,9 +2890,9 @@ dependencies = [ [[package]] name = "lightningcss" -version = "1.0.0-alpha.70" +version = "1.0.0-alpha.71" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9efb6a77b2389e62735b0b8157be9cc10a159eb4d1c3b864e99db9f297ada1b0" +checksum = "cb6314c2f0590ac93c86099b98bb7ba8abcf759bfd89604ffca906472bb54937" dependencies = [ "ahash 0.8.12", "bitflags", @@ -2910,11 +2957,11 @@ checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897" [[package]] name = "lru" -version = "0.12.5" +version = "0.16.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "234cf4f4a04dc1f57e24b96cc0cd600cf2af460d4161ac5ecdd0af8e1f3b2a38" +checksum = "7f66e8d5d03f609abc3a39e6f08e4164ebf1447a732906d39eb9b99b7919ef39" dependencies = [ - "hashbrown 0.15.5", + "hashbrown 0.16.1", ] [[package]] @@ -2951,7 +2998,17 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d89e7ee0cfbedfc4da3340218492196241d89eefb6dab27de5df917a6d2e78cf" dependencies = [ "cfg-if", - "digest", + "digest 0.10.7", +] + +[[package]] +name = "md-5" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "69b6441f590336821bb897fb28fc622898ccceb1d6cea3fde5ea86b090c4de98" +dependencies = [ + "cfg-if", + "digest 0.11.2", ] [[package]] @@ -3006,9 +3063,9 @@ dependencies = [ [[package]] name = "mio" -version = "1.1.1" +version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a69bcab0ad47271a0234d9422b131806bf3968021e5dc9328caf2d4cd58557fc" +checksum = "50b7e5b27aa02a74bac8c3f23f448f8d87ff11f92d3aac1a6ed369ee08cc56c1" dependencies = [ "libc", "wasi 0.11.1+wasi-snapshot-preview1", @@ -3043,9 +3100,9 @@ dependencies = [ [[package]] name = "moka" -version = "0.12.13" +version = "0.12.15" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b4ac832c50ced444ef6be0767a008b02c106a909ba79d1d830501e94b96f6b7e" +checksum = "957228ad12042ee839f93c8f257b62b4c0ab5eaae1d4fa60de53b27c9d7c5046" dependencies = [ "async-lock", "crossbeam-channel", @@ -3063,9 +3120,9 @@ dependencies = [ [[package]] name = "moxcms" -version = "0.7.11" +version = "0.8.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ac9557c559cd6fc9867e122e20d2cbefc9ca29d80d027a8e39310920ed2f0a97" +checksum = "bb85c154ba489f01b25c0d36ae69a87e4a1c73a72631fc6c0eb6dde34a73e44b" dependencies = [ "num-traits", "pxfm", @@ -3093,7 +3150,7 @@ dependencies = [ "httparse", "memchr", "mime", - "spin", + "spin 0.9.8", "version_check", ] @@ -3286,9 +3343,9 @@ dependencies = [ [[package]] name = "oxc-miette" -version = "2.7.0" +version = "2.7.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "60a7ba54c704edefead1f44e9ef09c43e5cfae666bdc33516b066011f0e6ebf7" +checksum = "4356a61f2ed4c9b3610245215fbf48970eb277126919f87db9d0efa93a74245c" dependencies = [ "cfg-if", "owo-colors", @@ -3301,9 +3358,9 @@ dependencies = [ [[package]] name = "oxc-miette-derive" -version = "2.7.0" +version = "2.7.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d4faecb54d0971f948fbc1918df69b26007e6f279a204793669542e1e8b75eb3" +checksum = "b237422b014f8f8fff75bb9379e697d13f8d57551a22c88bebb39f073c1bf696" dependencies = [ "proc-macro2", "quote", @@ -3312,21 +3369,21 @@ dependencies = [ [[package]] name = "oxc_allocator" -version = "0.116.0" +version = "0.125.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "93d4e5a43018728a7f6aa14b1034c43a0869fc0d05002a181d5dacc698e488b9" +checksum = "fc35d43f3816c0ade3657e051959e0fbcef18bcb99fe9cde80df5b8ae4307d1d" dependencies = [ "allocator-api2", - "hashbrown 0.16.1", + "hashbrown 0.17.0", "oxc_data_structures", "rustc-hash", ] [[package]] name = "oxc_ast" -version = "0.116.0" +version = "0.125.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "12b7429c1035c3bf0de582ae4458d4f769566d574cfe71e170e328a123337a8d" +checksum = "dcd098e18245c41bb94b79bcd4bc1ae02b00cfb805fe6c345210f369d51359e7" dependencies = [ "bitflags", "oxc_allocator", @@ -3336,14 +3393,15 @@ dependencies = [ "oxc_estree", "oxc_regular_expression", "oxc_span", + "oxc_str", "oxc_syntax", ] [[package]] name = "oxc_ast_macros" -version = "0.116.0" +version = "0.125.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1d3d3eb4a7b45c9f407f96ecafddad0a88cbf085dca7d74f3e4191ba03d2d5f3" +checksum = "f8ce2f47de20a90bb085f523b23dddd35587aa3e58284d6eb7148559808e2c6d" dependencies = [ "phf 0.13.1", "proc-macro2", @@ -3353,9 +3411,9 @@ dependencies = [ [[package]] name = "oxc_ast_visit" -version = "0.116.0" +version = "0.125.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "87408c66c769e714cc74c24626403c63b0fd76f251d0971404a347611280668d" +checksum = "dfafe2434e80d214debb69f3c4f3c0b53b2850ae2b49a6c5f7722b9a4b0b1834" dependencies = [ "oxc_allocator", "oxc_ast", @@ -3365,9 +3423,9 @@ dependencies = [ [[package]] name = "oxc_codegen" -version = "0.116.0" +version = "0.125.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "db5daa86977bc78c58c19ecf96afad677eb3dbc3726901d6a02b19ad1cf77f9c" +checksum = "f645e560c9ee5518df28578cad12ebe9d259faf90c40deffca9c64725babbfa0" dependencies = [ "bitflags", "cow-utils", @@ -3380,15 +3438,16 @@ dependencies = [ "oxc_semantic", "oxc_sourcemap", "oxc_span", + "oxc_str", "oxc_syntax", "rustc-hash", ] [[package]] name = "oxc_compat" -version = "0.116.0" +version = "0.125.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4d7987ecf86a8b5c4232dba5cdbe3b639bc0e3618e8c6a7c0a4797447a2baf57" +checksum = "0e4cdbc5cf3e5a85471093b04b2535ffd9b4135a74691110b4843c4e992a60ea" dependencies = [ "cow-utils", "oxc-browserslist", @@ -3399,15 +3458,15 @@ dependencies = [ [[package]] name = "oxc_data_structures" -version = "0.116.0" +version = "0.125.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "af06d9ab1bfd8baff16ef7dcc6824ea1fea0938ba41b1e113b882738e285d8ee" +checksum = "758fb6faa0d257b6f7b456509428dcea9ac545bad87a922fadc2b2476346fe24" [[package]] name = "oxc_diagnostics" -version = "0.116.0" +version = "0.125.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "908b7dae769982c7ade68f6ba3b4edb32b5a5941800fe8118f9c96fe4cfa7982" +checksum = "c4c9abc156895d127776c015fbe7da9bf88e8722b49a2ac7c5ba15ef9c2fbfd5" dependencies = [ "cow-utils", "oxc-miette", @@ -3416,9 +3475,9 @@ dependencies = [ [[package]] name = "oxc_ecmascript" -version = "0.116.0" +version = "0.125.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "78e4efa8dd802bb4e15536c71b081540f7bf4fc046b7043bc2e650cea40f62ca" +checksum = "5dfee7c08f13589fec9f75b27f7ded98c3770c9788c046866281e3781a5abd0c" dependencies = [ "cow-utils", "num-bigint", @@ -3432,9 +3491,9 @@ dependencies = [ [[package]] name = "oxc_estree" -version = "0.116.0" +version = "0.125.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f5e33d1f099a1e5c8f820df64654b1657f0aaf744d4137377726902738bfba80" +checksum = "d60d5a46e1c8a3551693d72afb77bad9a4bbabe764df1d3f19ec77244735ea92" [[package]] name = "oxc_index" @@ -3448,9 +3507,9 @@ dependencies = [ [[package]] name = "oxc_mangler" -version = "0.116.0" +version = "0.125.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "24ea391da3da4ae3f218b9457af29eb3a245372cc80c48694a7c1a398e5df539" +checksum = "1d461e9a47651fbba7369b5d02a7ac21a9d7c8d3438705d8f159e90128923425" dependencies = [ "itertools 0.14.0", "oxc_allocator", @@ -3459,15 +3518,16 @@ dependencies = [ "oxc_index", "oxc_semantic", "oxc_span", + "oxc_str", "oxc_syntax", "rustc-hash", ] [[package]] name = "oxc_minifier" -version = "0.116.0" +version = "0.125.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "de969df0153fb14e363e1afe62bf6c1c4c01b8f8ab72b5763c569dd0215cb3e6" +checksum = "79dc6424f983a4346ddda5e97d69bb764c9e7b01b6c29911512fd8d4929edeab" dependencies = [ "cow-utils", "itoa", @@ -3491,9 +3551,9 @@ dependencies = [ [[package]] name = "oxc_parser" -version = "0.116.0" +version = "0.125.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fa019a6d043e2087d8506c6716a55ee680ffb3eb564ab52bed0328d4076afb33" +checksum = "1f5a07182628017d9ff227a759ceb848030b8d75d8da907633fa02087580bb92" dependencies = [ "bitflags", "cow-utils", @@ -3507,6 +3567,7 @@ dependencies = [ "oxc_ecmascript", "oxc_regular_expression", "oxc_span", + "oxc_str", "oxc_syntax", "rustc-hash", "seq-macro", @@ -3514,15 +3575,16 @@ dependencies = [ [[package]] name = "oxc_regular_expression" -version = "0.116.0" +version = "0.125.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0f28f7617dc3d0dd642ca4f9be4c8cf5365f114dc27a745d1ebfa82b8f729ef3" +checksum = "acdcbaf3009ca3acca6f9e43047839edfab2fc50a4685f47deedea7d8f6f5eea" dependencies = [ "bitflags", "oxc_allocator", "oxc_ast_macros", "oxc_diagnostics", "oxc_span", + "oxc_str", "phf 0.13.1", "rustc-hash", "unicode-id-start", @@ -3530,24 +3592,23 @@ dependencies = [ [[package]] name = "oxc_semantic" -version = "0.116.0" +version = "0.125.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c5ab63f46cfbf4ef58b5a78439202593fe4a8511281ba1de32303e2c28b0e5e8" +checksum = "cabf785c2f7ddd188ab5f664b969e7c57b4265687eca14b40fe8ed90b25d88af" dependencies = [ "itertools 0.14.0", "memchr", "oxc_allocator", "oxc_ast", "oxc_ast_visit", - "oxc_data_structures", "oxc_diagnostics", "oxc_ecmascript", "oxc_index", "oxc_span", + "oxc_str", "oxc_syntax", "rustc-hash", "self_cell", - "smallvec", ] [[package]] @@ -3565,9 +3626,9 @@ dependencies = [ [[package]] name = "oxc_span" -version = "0.116.0" +version = "0.125.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6813c0f28625a9a1c8906144396fb7ee22a2a7722d49b707a06d19f81be450ef" +checksum = "2d6563c4417d299d8d39221218b512ddfa4a05d1d3607091f72547e9b4f2f1e0" dependencies = [ "compact_str", "oxc-miette", @@ -3579,21 +3640,21 @@ dependencies = [ [[package]] name = "oxc_str" -version = "0.116.0" +version = "0.125.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b163ab61908f1d636704bb57425692c38a7addde8e7244ff0dd92394084a17fe" +checksum = "a6e67a133122ad525d7059bd8824d36acea375d198b11790f4cad5d406cdb23d" dependencies = [ "compact_str", - "hashbrown 0.16.1", + "hashbrown 0.17.0", "oxc_allocator", "oxc_estree", ] [[package]] name = "oxc_syntax" -version = "0.116.0" +version = "0.125.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "06c82e33172f5613c4b7cf502d90b4ea866cd9103cb5e818a938889aef847daf" +checksum = "fc45d59f9d82e49b63430ddf08cf84e127aa48211bab05975ddfa8fef5460fad" dependencies = [ "bitflags", "cow-utils", @@ -3604,15 +3665,16 @@ dependencies = [ "oxc_estree", "oxc_index", "oxc_span", + "oxc_str", "phf 0.13.1", "unicode-id-start", ] [[package]] name = "oxc_traverse" -version = "0.116.0" +version = "0.125.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5c36f3713080a74481e1d725dcfcec7a1e0ebce350df427f1babc0cd4c8865f1" +checksum = "7f32ee338e4c2e39ddb9107179e2941609cf252b3802413a01238d3d5b81fb5a" dependencies = [ "itoa", "oxc_allocator", @@ -3649,6 +3711,7 @@ dependencies = [ "chrono", "dashmap 6.1.0", "dotenvy", + "fastcdc", "flate2", "fs2", "futures", @@ -3663,7 +3726,8 @@ dependencies = [ "kamadak-exif", "lightningcss", "lru", - "md-5", + "md-5 0.11.0", + "memmap2", "mimalloc", "mime_guess", "mockall", @@ -3681,8 +3745,8 @@ dependencies = [ "reqwest", "serde", "serde_json", - "sha2", - "socket2 0.6.2", + "sha2 0.11.0", + "socket2 0.6.3", "sqlx", "tempfile", "thiserror 2.0.18", @@ -3706,7 +3770,7 @@ checksum = "51f44edd08f51e2ade572f141051021c5af22677e42b7dd28a88155151c33594" dependencies = [ "ecdsa 0.14.8", "elliptic-curve 0.12.3", - "sha2", + "sha2 0.10.9", ] [[package]] @@ -3718,7 +3782,7 @@ dependencies = [ "ecdsa 0.16.9", "elliptic-curve 0.13.8", "primeorder", - "sha2", + "sha2 0.10.9", ] [[package]] @@ -3730,7 +3794,7 @@ dependencies = [ "ecdsa 0.16.9", "elliptic-curve 0.13.8", "primeorder", - "sha2", + "sha2 0.10.9", ] [[package]] @@ -4030,7 +4094,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9d1fe60d06143b2430aa532c94cfe9e29783047f06c0d7fd359a9a51b729fa25" dependencies = [ "cfg-if", - "cpufeatures", + "cpufeatures 0.2.17", "opaque-debug", "universal-hash", ] @@ -4204,7 +4268,7 @@ dependencies = [ "quinn-udp", "rustc-hash", "rustls 0.23.36", - "socket2 0.6.2", + "socket2 0.5.10", "thiserror 2.0.18", "tokio", "tracing", @@ -4241,7 +4305,7 @@ dependencies = [ "cfg_aliases", "libc", "once_cell", - "socket2 0.6.2", + "socket2 0.5.10", "tracing", "windows-sys 0.60.2", ] @@ -4369,9 +4433,9 @@ dependencies = [ [[package]] name = "rayon" -version = "1.11.0" +version = "1.12.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "368f01d005bf8fd9b1206fb6fa653e6c4a81ceb1466406b81792d87c5677a58f" +checksum = "fb39b166781f92d482534ef4b4b1b2568f42613b53e5b6c160e24cfbfa30926d" dependencies = [ "either", "rayon-core", @@ -4563,8 +4627,8 @@ version = "0.9.10" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b8573f03f5883dcaebdfcf4725caa1ecb9c15b2ef50c43a07b816e06799bb12d" dependencies = [ - "const-oid", - "digest", + "const-oid 0.9.6", + "digest 0.10.7", "num-bigint-dig", "num-integer", "num-traits", @@ -4881,8 +4945,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e3bf829a2d51ab4a5ddf1352d8470c140cadc8301b2ae1789db023f01cedd6ba" dependencies = [ "cfg-if", - "cpufeatures", - "digest", + "cpufeatures 0.2.17", + "digest 0.10.7", ] [[package]] @@ -4892,8 +4956,19 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283" dependencies = [ "cfg-if", - "cpufeatures", - "digest", + "cpufeatures 0.2.17", + "digest 0.10.7", +] + +[[package]] +name = "sha2" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "446ba717509524cb3f22f17ecc096f10f4822d76ab5c0b9822c5f9c284e825f4" +dependencies = [ + "cfg-if", + "cpufeatures 0.3.0", + "digest 0.11.2", ] [[package]] @@ -4927,7 +5002,7 @@ version = "1.6.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "74233d3b3b2f6d4b006dc19dee745e73e2a6bfb6f93607cd3b02bd5b00797d7c" dependencies = [ - "digest", + "digest 0.10.7", "rand_core 0.6.4", ] @@ -4937,7 +5012,7 @@ version = "2.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "77549399552de45a898a580c1b41d445bf730df867cc44e6c0233bbc4b8329de" dependencies = [ - "digest", + "digest 0.10.7", "rand_core 0.6.4", ] @@ -5013,12 +5088,12 @@ dependencies = [ [[package]] name = "socket2" -version = "0.6.2" +version = "0.6.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "86f4aa3ad99f2088c990dfa82d367e19cb29268ed67c574d10d0a4bfe71f07e0" +checksum = "3a766e1110788c36f4fa1c2b71b387a7815aa65f88ce0229841826633d93723e" dependencies = [ "libc", - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -5030,6 +5105,12 @@ dependencies = [ "lock_api", ] +[[package]] +name = "spin" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d5fe4ccb98d9c292d56fec89a5e07da7fc4cf0dc11e156b41793132775d3e591" + [[package]] name = "spki" version = "0.6.0" @@ -5090,7 +5171,7 @@ dependencies = [ "rustls 0.23.36", "serde", "serde_json", - "sha2", + "sha2 0.10.9", "smallvec", "thiserror 2.0.18", "tokio", @@ -5129,7 +5210,7 @@ dependencies = [ "quote", "serde", "serde_json", - "sha2", + "sha2 0.10.9", "sqlx-core", "sqlx-mysql", "sqlx-postgres", @@ -5152,7 +5233,7 @@ dependencies = [ "bytes", "chrono", "crc", - "digest", + "digest 0.10.7", "dotenvy", "either", "futures-channel", @@ -5165,7 +5246,7 @@ dependencies = [ "hmac", "itoa", "log", - "md-5", + "md-5 0.10.6", "memchr", "once_cell", "percent-encoding", @@ -5173,7 +5254,7 @@ dependencies = [ "rsa", "serde", "sha1", - "sha2", + "sha2 0.10.9", "smallvec", "sqlx-core", "stringprep", @@ -5206,13 +5287,13 @@ dependencies = [ "home", "itoa", "log", - "md-5", + "md-5 0.10.6", "memchr", "once_cell", "rand 0.8.5", "serde", "serde_json", - "sha2", + "sha2 0.10.9", "smallvec", "sqlx-core", "stringprep", @@ -5469,25 +5550,25 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20" [[package]] name = "tokio" -version = "1.49.0" +version = "1.52.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "72a2903cd7736441aac9df9d7688bd0ce48edccaadf181c3b90be801e81d3d86" +checksum = "a91135f59b1cbf38c91e73cf3386fca9bb77915c45ce2771460c9d92f0f3d776" dependencies = [ "bytes", "libc", "mio", "pin-project-lite", "signal-hook-registry", - "socket2 0.6.2", + "socket2 0.6.3", "tokio-macros", "windows-sys 0.61.2", ] [[package]] name = "tokio-macros" -version = "2.6.0" +version = "2.7.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "af407857209536a95c8e56f8231ef2c2e2aff839b22e07a1ffcbc617e9db9fa5" +checksum = "385a6cb71ab9ab790c5fe8d67f1645e6c450a7ce006a33de03daa956cf70a496" dependencies = [ "proc-macro2", "quote", @@ -5750,7 +5831,7 @@ version = "0.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fc1de2c688dc15305988b563c3854064043356019f97a4b46276fe734c4f07ea" dependencies = [ - "crypto-common", + "crypto-common 0.1.7", "subtle", ] @@ -5812,9 +5893,9 @@ dependencies = [ [[package]] name = "uuid" -version = "1.21.0" +version = "1.23.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b672338555252d43fd2240c714dc444b8c6fb0a5c5335e65a07bba7742735ddb" +checksum = "5ac8b6f42ead25368cf5b098aeb3dc8a1a2c05a3eee8a9a1a68c640edbfc79d9" dependencies = [ "getrandom 0.4.1", "js-sys", diff --git a/Cargo.toml b/Cargo.toml index 591f331f..6d1f8624 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -7,8 +7,8 @@ default-run = "oxicloud" [dependencies] mimalloc = { version = "0.1.48", default-features = false } -axum = { version = "0.8.8", features = ["multipart", "http1", "http2", "tokio", "macros"] } -tokio = { version = "1.49.0", features = ["rt-multi-thread", "macros", "io-util", "net", "time", "sync", "fs"] } +axum = { version = "0.8.9", features = ["multipart", "http1", "http2", "tokio", "macros"] } +tokio = { version = "1.52.0", features = ["rt-multi-thread", "macros", "io-util", "net", "time", "sync", "fs"] } tokio-util = { version = "0.7.18", features = ["io", "codec", "compat"] } tokio-stream = { version = "0.1.18", features = ["fs"] } bytes = "1.11.1" @@ -25,7 +25,7 @@ serde_json = "1.0.149" futures = "0.3.32" async-stream = "0.3.6" mime_guess = "2.0.5" -uuid = { version = "1.21.0", features = ["v4", "serde"] } +uuid = { version = "1.23.0", features = ["v4", "serde"] } thiserror = "2.0.18" mockall = { version = "0.14.0", optional = true } @@ -35,37 +35,39 @@ argon2 = "0.5.3" rand_core = { version = "0.6", features = ["std", "getrandom"] } quick-xml = "0.39.2" dotenvy = "0.15.7" -moka = { version = "0.12", features = ["future", "sync"] } +moka = { version = "0.12.15", features = ["future", "sync"] } http-range-header = "0.4" -image = { version = "0.25", default-features = false, features = ["jpeg", "png", "gif", "webp"] } +image = { version = "0.25.10", default-features = false, features = ["jpeg", "png", "gif", "webp"] } id3 = "1.14" mp3-duration = "0.1" -kamadak-exif = "0.5" -md-5 = "0.10" -sha2 = "0.10.9" -blake3 = { version = "1.8.3", features = ["rayon", "mmap"] } +kamadak-exif = "0.6.1" +md-5 = "0.11.0" +sha2 = "0.11.0" +blake3 = { version = "1.8.4", features = ["rayon", "mmap"] } hex = "0.4.3" http-body-util = "0.1.3" -percent-encoding = "2.3" +percent-encoding = "2.3.2" reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls-webpki-roots"] } base64 = "0.22.1" fs2 = "0.4" -rayon = "1.11" +rayon = "1.12.0" infer = "0.19" -async-compression = { version = "0.4", features = ["tokio", "gzip"] } +async-compression = { version = "0.4.41", features = ["tokio", "gzip"] } async_zip = { version = "0.0.18", features = ["tokio", "deflate"] } -dashmap = "6" -socket2 = { version = "0.6.2", features = ["all"] } +dashmap = "6.1.0" +socket2 = { version = "0.6.3", features = ["all"] } urlencoding = "2.1.3" -utoipa = { version = "5", features = ["axum_extras", "uuid", "chrono"] } -aws-sdk-s3 = "1" -aws-config = { version = "1", features = ["behavior-version-latest"] } -aws-smithy-types = "1" +utoipa = { version = "5.4.0", features = ["axum_extras", "uuid", "chrono"] } +aws-sdk-s3 = "1.129.0" +aws-config = { version = "1.8.15", features = ["behavior-version-latest"] } +aws-smithy-types = "1.4.7" azure_core = "0.21" azure_storage = "0.21" azure_storage_blobs = "0.21" -aes-gcm = "0.10" -lru = "0.12" +aes-gcm = "0.10.3" +lru = "0.16.4" +fastcdc = "4.0.0" +memmap2 = "0.9.10" [features] default = [] @@ -80,12 +82,12 @@ name = "generate-openapi" path = "src/bin/generate-openapi.rs" [build-dependencies] -oxc_allocator = "0.116" -oxc_parser = "0.116" -oxc_span = "0.116" -oxc_codegen = "0.116" -oxc_minifier = "0.116" -lightningcss = "1.0.0-alpha.70" +oxc_allocator = "0.125.0" +oxc_parser = "0.125.0" +oxc_span = "0.125.0" +oxc_codegen = "0.125.0" +oxc_minifier = "0.125.0" +lightningcss = "1.0.0-alpha.71" [profile.release] lto = "fat" diff --git a/doc/deduplication.md b/doc/deduplication.md index 1e8499a9..74d917d0 100644 --- a/doc/deduplication.md +++ b/doc/deduplication.md @@ -1,89 +1,222 @@ # 06 - Deduplication -OxiCloud uses **content-addressable deduplication** via SHA-256 hashing. Uploaded file content is hashed and stored in a central blob store. Identical files share the same blob, tracked by a reference counter. Disk savings scale with the number of duplicates. +OxiCloud uses **content-defined chunking (CDC)** via FastCDC for sub-file deduplication. Files are split into variable-size chunks (64 KB – 1 MB, average 256 KB) using the FastCDC 2020 algorithm. Each chunk is individually BLAKE3-hashed and stored in a pluggable blob backend (local FS, S3, Azure). A PostgreSQL *manifest* maps the whole-file BLAKE3 hash to the ordered list of chunk hashes that compose it. Identical chunks across any files are stored once and reference-counted. -Deduplication is always enabled and non-fatal -- if dedup fails, file operations proceed normally with a warning log. +Deduplication is always enabled and non-fatal — if dedup fails, file operations proceed normally with a warning log. + +**Backward compatibility**: files uploaded before CDC (legacy whole-file blobs in `storage.blobs`) are served transparently. When no manifest row exists for a hash, the service falls back to direct blob reads. ## Architecture ``` -User Files (references) ──▶ Dedup Index (hash→metadata) ──▶ Blob Store (actual data) +┌─────────────────┐ ┌─────────────────────┐ ┌───────────────┐ +│ storage.files │────▶│ chunk_manifests │────▶│ storage.blobs │──▶ Blob Store +│ (references) │ │ (file→[chunk_hashes])│ │ (chunks) │ (Local/S3/Azure) +└─────────────────┘ └─────────────────────┘ └───────────────┘ ``` -### Storage Layout +### Database Tables -``` -/ - .blobs/ - 00/ .. ff/ ← 256 prefix directories (hex) for FS distribution - .blob ← actual blob files - .dedup_temp/ ← temp staging directory for atomic writes - .dedup_index.json ← persistent JSON index (hash → metadata) -``` +| Table | Schema | Purpose | +|---|---|---| +| `storage.chunk_manifests` | `storage` | Maps file_hash → ordered chunk_hashes[] + chunk_sizes[] + ref_count | +| `storage.blobs` | `storage` | Per-chunk metadata: hash (PK), size, ref_count, content_type | + +Defined in `migrations/20260414000000_chunk_manifests.sql`. Blobs table is part of the initial schema. ### Layer Placement | Layer | Component | File | |---|---|---| | Application Port | **DedupPort** trait + DTOs | `src/application/ports/dedup_ports.rs` | +| Application Port | **BlobStorageBackend** trait | `src/application/ports/blob_storage_ports.rs` | | Infrastructure | **DedupService** implementation | `src/infrastructure/services/dedup_service.rs` | +| Infrastructure | Blob backends (Local, S3, Azure, Retry, Encrypted, Cached, Migration) | `src/infrastructure/services/*_blob_backend.rs` | | Interfaces | **DedupHandler** REST endpoints | `src/interfaces/api/handlers/dedup_handler.rs` | -| Integration | **FileUploadService** (dedup on upload) | `src/application/services/file_upload_service.rs` | -| Integration | **FileManagementService** (ref-count on delete) | `src/application/services/file_management_service.rs` | +| Integration | **FileBlobWriteRepository** (dedup on upload) | `src/infrastructure/repositories/pg/file_blob_write_repository.rs` | +| Integration | **FileBlobReadRepository** (dedup reads) | `src/infrastructure/repositories/pg/file_blob_read_repository.rs` | ## Constants | Constant | Value | Description | |---|---|---| -| `HASH_CHUNK_SIZE` | 256 KB (`256 * 1024`) | Chunk size for streaming SHA-256 computation | -| `MIN_DEDUP_SIZE` | 4 KB (`4096`) | Files below this size skip deduplication | +| `CDC_MIN_CHUNK` | 64 KB (`65_536`) | Minimum CDC chunk size | +| `CDC_AVG_CHUNK` | 256 KB (`262_144`) | Average / target CDC chunk size | +| `CDC_MAX_CHUNK` | 1 MB (`1_048_576`) | Maximum CDC chunk size | +| `CHUNK_UPLOAD_CONCURRENCY` | 8 | Maximum parallel chunk uploads to blob backend | -Hardcoded in `dedup_service.rs`. No runtime configuration beyond **storage_path**. +Hardcoded in `dedup_service.rs`. + +## Write Path: `store_from_file` + +The core write operation follows a **write-first strategy** that never holds a PG connection during disk I/O: + +``` +store_from_file(source_path, content_type, pre_computed_hash) + │ + ├─ Fast path: pre_computed_hash provided? + │ └─ try_dedup_hit() → check manifest + legacy blob + │ └─ Hit? → bump ref_count, delete source, return ExistingBlob + │ + ├─ CDC analysis (single mmap pass, spawn_blocking): + │ ├─ Memory-map the file (memmap2) + │ ├─ FastCDC 2020 boundary detection → ChunkMeta[] + │ └─ BLAKE3 whole-file hash (concurrent with chunking) + │ + ├─ Second dedup check with computed hash (if no pre_computed_hash) + │ + ├─ store_chunks() — 3-phase pipeline: + │ │ + │ ├─ Phase 0: Batch-check existing chunks (single PG query) + │ │ SELECT hash FROM storage.blobs WHERE hash = ANY($1) + │ │ → HashSet of already-stored chunk hashes + │ │ + │ ├─ Phase 1: Selective disk read (sequential, one pass) + │ │ For each chunk: + │ │ existing? → skip read (None) + │ │ new? → seek + read_exact → Some(Bytes) + │ │ + │ └─ Phase 2: Parallel operations (buffer_unordered × 8) + │ new chunk: put_blob_from_bytes + INSERT ON CONFLICT + │ existing chunk: UPDATE ref_count + 1 (no disk I/O) + │ + ├─ INSERT manifest into storage.chunk_manifests + │ (file_hash, chunk_hashes[], chunk_sizes[], total_size, chunk_count) + │ + └─ Delete source file, return NewBlob { hash, size } +``` + +### Dedup Skip Optimization + +The **biggest I/O saving** for versioned files. Before reading any chunk from disk or uploading it to the blob backend, `store_chunks` batch-queries PG to discover which chunk hashes already exist: + +```sql +SELECT hash FROM storage.blobs WHERE hash = ANY($1) +``` + +This single round-trip returns all known chunks. For each existing chunk, the service skips: +- `seek()` + `read_exact()` from the source file (no disk I/O) +- `put_blob_from_bytes()` to the backend (no network I/O for S3/Azure) + +Only a lightweight `UPDATE ref_count + 1` is executed in PG (~0.1 ms per chunk). + +**Impact**: for a 100 MB versioned file where 95% of chunks are unchanged, only ~5 MB is read from disk and uploaded. The remaining 95% costs only PG ref-count bumps. + +### Parallel Chunk Storage + +Phase 2 of `store_chunks` uses `futures::stream::buffer_unordered(8)` to execute up to 8 concurrent chunk operations. This is a major win for S3/Azure backends where each PUT has 50-200 ms of network latency. + +Chunk order in the returned `(chunk_hashes, chunk_sizes)` is preserved by deriving both from the original `ChunkMeta` slice (CDC order), not from the unordered parallel results. + +### Full-File Dedup Hit (Fast Path) + +When a file with the exact same BLAKE3 hash already has a manifest, `try_dedup_hit` returns immediately: +- Bumps `chunk_manifests.ref_count` +- Deletes the source file +- Returns `ExistingBlob` — **zero chunk I/O** + +Also checks legacy whole-file blobs in `storage.blobs` for backward compatibility. + +## Read Path + +### Streaming Read (`read_blob_stream`) + +CDC-aware with legacy fallback: + +1. Query `chunk_manifests` for `chunk_hashes[]` +2. If found: stream chunks in order via `backend.get_blob_stream(chunk_hash)`, concatenated into a single byte stream with `buffered(1) + try_flatten` +3. If not found: fall back to `backend.get_blob_stream(hash)` for legacy blobs + +### Range Read (`read_blob_range_stream`) + +For HTTP Range requests (and WOPI/WebDAV partial reads): + +1. Query manifest for `chunk_hashes[]`, `chunk_sizes[]`, `total_size` +2. Calculate which chunks overlap `[start, end)` using cumulative offsets +3. For each overlapping chunk, compute the sub-range within that chunk +4. Stream only the relevant chunk portions via `backend.get_blob_range_stream()` + +### Blob Size (`blob_size`) + +Returns `total_size` from the manifest (O(1) PG lookup). Falls back to `backend.blob_size()` for legacy blobs. Used by HEAD requests for Content-Length. + +## Reference Counting + +### Adding References (`add_reference`) + +Manifest-aware with legacy fallback: +1. Try `UPDATE chunk_manifests SET ref_count = ref_count + 1 WHERE file_hash = $1` +2. If no rows affected, try `UPDATE storage.blobs SET ref_count + 1 WHERE hash = $1` +3. If neither exists, return NotFound error + +### Removing References (`remove_reference`) + +**CDC manifest path** (transactional): +1. Check `chunk_manifests` for the file hash +2. If `ref_count > 1`: decrement manifest ref_count → commit +3. If `ref_count == 1` (last reference): + - `SELECT ... FOR UPDATE` to lock the manifest row + - `DELETE FROM chunk_manifests` + - `UPDATE storage.blobs SET ref_count = ref_count - 1 WHERE hash = ANY(chunk_hashes)` + - `DELETE FROM storage.blobs WHERE hash = ANY(chunk_hashes) AND ref_count <= 0 RETURNING hash` + - Commit TX + - Delete orphaned chunk blob files from backend (after commit) + +**Legacy blob path** (transactional): +1. `SELECT ref_count, size FROM storage.blobs WHERE hash = $1 FOR UPDATE` +2. If `ref_count == 1`: `DELETE FROM storage.blobs` + delete blob file +3. If `ref_count > 1`: `UPDATE SET ref_count = ref_count - 1` ## Port: DedupPort Trait Defined in `src/application/ports/dedup_ports.rs`: ```rust -#[async_trait] pub trait DedupPort: Send + Sync + 'static { - /// Store content from bytes, returning dedup result - async fn store_bytes(&self, content: &[u8], content_type: Option) -> Result; + /// Store content with CDC deduplication (from file). + async fn store_from_file( + &self, + source_path: &Path, + content_type: Option, + pre_computed_hash: Option, + ) -> Result; - /// Store content from an existing file path - async fn store_from_file(&self, source_path: &Path, content_type: Option) -> Result; - - /// Check if a blob exists by hash + /// Check if a blob exists by hash (manifest or legacy). async fn blob_exists(&self, hash: &str) -> bool; - /// Get metadata for a blob + /// Get metadata for a blob. async fn get_blob_metadata(&self, hash: &str) -> Option; - /// Read blob content as Vec - async fn read_blob(&self, hash: &str) -> Result, DomainError>; + /// Stream blob content — CDC-aware with legacy fallback. + async fn read_blob_stream(&self, hash: &str) + -> Result> + Send>>, DomainError>; - /// Read blob content as Bytes - async fn read_blob_bytes(&self, hash: &str) -> Result; + /// Stream a byte range — CDC-aware with legacy fallback. + async fn read_blob_range_stream(&self, hash: &str, start: u64, end: Option) + -> Result> + Send>>, DomainError>; - /// Increment reference count for a blob + /// Get blob size without reading content. + async fn blob_size(&self, hash: &str) -> Result; + + /// Increment reference count (manifest-aware). async fn add_reference(&self, hash: &str) -> Result<(), DomainError>; - /// Decrement reference count; deletes blob if it reaches 0. Returns true if deleted. + /// Decrement reference count. Returns true if blob was deleted. async fn remove_reference(&self, hash: &str) -> Result; - /// Compute SHA-256 hash of bytes (synchronous) - fn hash_bytes(&self, content: &[u8]) -> String; - - /// Compute SHA-256 hash of file (streaming) + /// Calculate BLAKE3 hash of a file (mmap + rayon). async fn hash_file(&self, path: &Path) -> Result; - /// Get deduplication statistics + /// Get local filesystem path for a blob hash. + fn blob_path(&self, hash: &str) -> PathBuf; + + /// Get deduplication statistics (computed from PG). async fn get_stats(&self) -> DedupStatsDto; - /// Persist index to disk + /// Flush index to persistent storage (no-op for PG backend). async fn flush(&self) -> Result<(), DomainError>; - /// Verify integrity of all blobs (existence, hash, size) + /// Verify integrity of all stored blobs and manifests. async fn verify_integrity(&self) -> Result, DomainError>; } ``` @@ -91,22 +224,22 @@ pub trait DedupPort: Send + Sync + 'static { ### Port DTOs ```rust -/// Result of a dedup store operation +/// Result of a dedup store operation. pub enum DedupResultDto { - NewBlob { hash: String, size: u64, blob_path: PathBuf }, - ExistingBlob { hash: String, size: u64, blob_path: PathBuf, saved_bytes: u64 }, + NewBlob { hash: String, size: u64 }, + ExistingBlob { hash: String, size: u64, saved_bytes: u64 }, } -// Methods: hash(), size(), blob_path(), was_deduplicated() +// Methods: hash(), size(), was_deduplicated() -/// Metadata for a stored blob +/// Metadata for a stored blob. pub struct BlobMetadataDto { - pub hash: String, // SHA-256 hex string + pub hash: String, // BLAKE3 hex string pub size: u64, pub ref_count: u32, pub content_type: Option, } -/// Aggregate dedup statistics +/// Aggregate dedup statistics (computed from PG). pub struct DedupStatsDto { pub total_blobs: u64, pub total_bytes_stored: u64, @@ -125,11 +258,9 @@ Implemented in `src/infrastructure/services/dedup_service.rs`. ```rust pub struct DedupService { - blob_root: PathBuf, // /.blobs - temp_root: PathBuf, // /.dedup_temp - index: Arc>>, // in-memory index - index_path: PathBuf, // /.dedup_index.json - stats: Arc>, + backend: Arc, // Pluggable blob storage (Local/S3/Azure/...) + pool: Arc, // Primary pool (request-path operations) + maintenance_pool: Arc, // Isolated pool (verify_integrity, GC) } ``` @@ -137,28 +268,34 @@ pub struct DedupService { | Method | Description | |---|---| -| `new(storage_root: &Path)` | Constructs paths, initializes empty index | -| `initialize()` | Creates `.blobs/` (256 prefix dirs), `.dedup_temp/`, loads index JSON | -| `blob_path(hash: &str)` | Returns `//.blob` | -| `hash_bytes(content: &[u8])` | Static SHA-256 → hex string | -| `hash_file(path: &Path)` | Streaming SHA-256 of file (256 KB chunks) | -| `store_bytes(content, content_type)` | Hash → check existing → increment ref or write new blob atomically | -| `store_from_file(source_path, content_type)` | Hash file → dedup check → move file to blob store | -| `add_reference(hash)` | Increments **ref_count** + updates stats | -| `remove_reference(hash)` | Decrements **ref_count**. If 0, deletes blob file + removes from index | -| `read_blob(hash)` | Reads blob file content | -| `get_stats()` | Returns current dedup statistics | -| `flush()` | Saves index to JSON atomically (write to `.json.tmp` then rename) | -| `verify_integrity()` | Checks every blob: file exists, hash matches, size matches | -| `garbage_collect()` | Removes blobs with `ref_count == 0`. Returns `(deleted_count, deleted_bytes)` | +| `new(backend, pool, maintenance_pool)` | Construct — wires pluggable backend + dual PG pools | +| `initialize()` | Initialize backend + log blob/manifest counts from PG | +| `cdc_hash_and_chunk_file(path)` | Single mmap pass: BLAKE3 whole-file hash + FastCDC chunk boundaries + per-chunk BLAKE3 | +| `cdc_chunk_file(path)` | CDC without whole-file hash (when hash is pre-computed) | +| `hash_file(path)` | BLAKE3 hash via mmap + rayon parallelism | +| `store_from_file(path, ct, hash)` | CDC → store_chunks → manifest INSERT (main write path) | +| `try_dedup_hit(hash, path)` | Check manifest/legacy for full-file dedup hit | +| `store_chunks(path, chunks)` | 3-phase: batch-check → selective read → parallel upload | +| `blob_exists(hash)` | Check manifest + legacy blob existence | +| `user_owns_blob_reference(hash, user_id)` | Authorization: check file ownership | +| `get_blob_metadata(hash)` | Manifest-aware metadata with legacy fallback | +| `add_reference(hash)` | Manifest-aware ref_count increment | +| `remove_reference(hash)` | Manifest-aware ref_count decrement + cascade cleanup | +| `read_blob_stream(hash)` | CDC chunk-streaming with legacy fallback | +| `read_blob_range_stream(hash, start, end)` | CDC range-streaming with legacy fallback | +| `blob_size(hash)` | O(1) from manifest, fallback to backend | +| `get_stats()` | Compute stats from PG (blobs + manifests) | +| `verify_integrity()` | Verify manifests (counts, sizes) + blobs (existence, size, re-hash) | +| `garbage_collect()` | Batch-delete orphaned manifests/blobs (uses maintenance pool) | ### Key Behaviors -- **Atomic writes**: new blobs are written to `.dedup_temp/.tmp` then renamed into `.blobs//.blob` -- **Index persistence**: auto-saved every 100 new blobs, also saved explicitly via `flush()` -- **Small file bypass**: files < 4 KB skip deduplication -- **File move optimization**: `store_from_file` uses `fs::rename` to move the source file into the blob store (zero-copy on same filesystem) -- **Thread safety**: index and stats are protected by `Arc>` +- **CDC analysis in `spawn_blocking`**: mmap + FastCDC runs off the async runtime to avoid blocking the event loop +- **Dual PG pools**: request-path operations use the primary pool; `verify_integrity` and `garbage_collect` use the maintenance pool to prevent starvation +- **Pluggable blob backend**: all chunk I/O goes through `Arc` — works with local FS, S3, Azure, or any composed backend (retry, encryption, caching) +- **Atomic chunk storage**: `put_blob_from_bytes` is idempotent; `INSERT ON CONFLICT` handles concurrent uploads of the same chunk +- **Delete-after-commit**: blob files are deleted from the backend only after the PG transaction commits, preventing orphaned PG rows +- **Flush is no-op**: PG handles durability via WAL/commit — no explicit index persistence needed ## REST API Endpoints @@ -166,7 +303,7 @@ All routes under `/api/dedup`, authentication required. | Method | Path | Handler | Description | |---|---|---|---| -| `GET` | `/api/dedup/check/{hash}` | `DedupHandler::check_hash` | Check if a blob exists by SHA-256 hash | +| `GET` | `/api/dedup/check/{hash}` | `DedupHandler::check_hash` | Check if a blob exists by BLAKE3 hash | | `POST` | `/api/dedup/upload` | `DedupHandler::upload_with_dedup` | Multipart upload with automatic dedup | | `GET` | `/api/dedup/stats` | `DedupHandler::get_stats` | Get deduplication statistics | | `GET` | `/api/dedup/blob/{hash}` | `DedupHandler::get_blob` | Retrieve raw blob content by hash | @@ -184,8 +321,6 @@ All routes under `/api/dedup`, authentication required. "ref_count": 3 } ``` -- Validates 64-character hex format for the hash parameter -- `existing_size` and `ref_count` are omitted when `exists` is `false` **Dedup Upload** (`POST /api/dedup/upload`): ```json @@ -197,8 +332,6 @@ All routes under `/api/dedup`, authentication required. "ref_count": 2 } ``` -- Returns `201 Created` for new blobs, `200 OK` for deduplicated content -- Accepts multipart form data **Stats** (`GET /api/dedup/stats`): ```json @@ -213,70 +346,22 @@ All routes under `/api/dedup`, authentication required. } ``` -**Get Blob** (`GET /api/dedup/blob/{hash}`): -- Returns raw blob content with `Content-Type` from metadata -- Adds `X-Dedup-Hash` response header - -**Remove Reference** (`DELETE /api/dedup/blob/{hash}`): -```json -{ - "success": true, - "deleted": true, - "message": "Blob deleted (ref count reached 0)" -} -``` - -## Integration with File Upload - -**FileUploadService** holds `dedup: Option>`. - -During `smart_upload()`, dedup runs for **all upload tiers** (write-behind, buffered, streaming): - -```rust -// Inside smart_upload() — dedup runs after data is collected -{ - let dedup_data: Vec = { /* combine all chunks */ }; - self.run_dedup(&dedup_data, &content_type).await; -} -``` - -The private `run_dedup` method: - -```rust -async fn run_dedup(&self, data: &[u8], content_type: &str) { - let Some(dedup) = &self.dedup else { return }; - match dedup.store_bytes(data, Some(content_type.to_string())).await { - Ok(result) => { /* log new or dedup hit */ } - Err(e) => { warn!("DEDUP: Failed to store in blob store: {}", e); } - } -} -``` - -Dedup is non-fatal -- failures are only logged as warnings. The file upload always completes regardless of dedup outcome. - -## Integration with File Deletion - -**FileManagementService** holds `dedup_service: Option>`. - -In `delete_with_cleanup()`: - -1. **Compute content hash** -- reads file via **FileReadPort**, calls `dedup.hash_bytes(&content)` -2. **Delete file** -- tries trash (soft delete) first, falls back to permanent delete -3. **Decrement dedup ref-count** -- calls `dedup.remove_reference(hash)` which may delete the blob if **ref_count** reaches 0 - -```rust -// Private helpers in FileManagementService -async fn compute_content_hash(&self, id: &str) -> Option -async fn decrement_dedup_ref(&self, hash: &str) -``` - ## DI Wiring In `src/common/di.rs`: ```rust -// Initialization (in create_core_services) -let dedup_service = Arc::new(DedupService::new(&self.storage_path)); +// Blob backend is assembled from layered backends: +// LocalBlobBackend / S3BlobBackend / AzureBlobBackend +// → RetryBlobBackend → EncryptedBlobBackend → CachedBlobBackend +let blob_backend: Arc = /* ... */; + +// DedupService receives the composed blob backend + dual PG pools +let dedup_service = Arc::new(DedupService::new( + blob_backend, + db_pool.clone(), + maintenance_pool.clone(), +)); dedup_service.initialize().await?; // Stored in CoreServices as: @@ -285,36 +370,70 @@ pub struct CoreServices { // ... } -// Injected into blob repositories (which handle dedup internally): +// Injected into blob repositories: FileBlobReadRepository::new(pool, core.dedup_service.clone(), folder_repo) FileBlobWriteRepository::new(pool, core.dedup_service.clone(), folder_repo) - -// Also injected into FileManagementService for ref cleanup on delete: -FileManagementService::new_full(write, read, trash, core.dedup_service.clone()) ``` -## Persistence +## Maintenance -Deduplication uses a **file-based JSON index** (`/.dedup_index.json`), NOT a database table. The index is loaded into memory at startup and flushed to disk: +### Garbage Collection (`garbage_collect`) -- Automatically every 100 new blobs -- Explicitly via `flush()` -- Uses atomic write (write to `.json.tmp` then rename) for crash safety +Two-phase batch deletion using the maintenance pool: + +1. **Phase 1 — Orphaned manifests**: `DELETE FROM chunk_manifests WHERE ref_count <= 0` (batches of 500). For each deleted manifest, `UPDATE storage.blobs SET ref_count = ref_count - 1` for its chunks. +2. **Phase 2 — Orphaned blobs**: `DELETE FROM storage.blobs WHERE ref_count <= 0` (batches of 500). Deletes blob files from backend + thumbnail cleanup (best-effort). + +Uses `tokio::task::yield_now()` between batches to avoid starving other tasks. + +### Integrity Verification (`verify_integrity`) + +Phase 1 — Verify CDC manifests: +- `chunk_hashes.len() == chunk_sizes.len()` +- `SUM(chunk_sizes) == total_size` +- Every referenced chunk exists in the blob backend with correct size + +Phase 2 — Verify blobs (chunks + legacy): +- Blob file exists in backend +- Actual size matches PG record +- (Local backends only) Re-hash file content to verify BLAKE3 integrity +- Processes 16 blobs concurrently via `buffer_unordered` ## Tests -Located at the bottom of `src/infrastructure/services/dedup_service.rs`: +Located at the bottom of `src/infrastructure/services/dedup_service.rs` (12 tests): | Test | Description | |---|---| -| `test_dedup_identical_content` | Stores same content (>4KB) twice. Verifies second is deduplicated, hashes match, `stats.dedup_hits == 1` | -| `test_reference_counting` | Stores twice (ref_count=2), removes one ref (not deleted), removes second (blob deleted) | +| `test_cdc_deterministic_same_content` | Same content → same file hash + same chunk hashes/offsets/lengths | +| `test_cdc_empty_file` | Empty file → zero chunks, correct BLAKE3 empty hash | +| `test_cdc_small_file_single_chunk` | File below min chunk → single chunk covering entire file | +| `test_cdc_chunk_sizes_within_bounds` | All non-last chunks are within [64 KB, 1 MB] | +| `test_cdc_file_hash_matches_hash_file` | CDC whole-file hash matches standalone `hash_file()` | +| `test_cdc_chunk_hashes_are_correct` | Each chunk hash == BLAKE3 of that chunk's data | +| `test_cdc_reassembly_matches_original` | Concatenating chunks reproduces original file | +| `test_cdc_chunks_are_contiguous` | Chunks cover entire file with no gaps or overlaps | +| `test_cdc_similar_files_share_chunks` | Editing last 64 KB of 2 MB file → most chunks shared | +| `test_cdc_chunk_file_matches_full` | `cdc_chunk_file` produces same chunks as `cdc_hash_and_chunk_file` | +| `test_cdc_large_file_chunk_count` | 8 MB file produces 8-128 chunks (avg ~256 KB) | +| `test_cdc_insert_at_beginning_preserves_later_chunks` | 128 KB prefix insert → CDC resynchronizes, later chunks shared | + +## Performance Characteristics + +| Scenario | Behavior | +|---|---| +| **First upload of new file** | Single mmap pass (CDC + hash) → parallel chunk upload → manifest INSERT | +| **Re-upload of identical file** | `try_dedup_hit` → manifest ref_count bump → zero chunk I/O | +| **Upload of edited file (5% changed)** | CDC → batch-check finds 95% existing → reads only 5% → uploads 5% → ref-bumps 95% | +| **Range read (1 MB from 1 GB file)** | Manifest lookup → identify overlapping chunks → stream only those portions | +| **Delete last reference** | TX: delete manifest → batch-decrement chunks → delete zero-ref chunks → commit → delete blob files | +| **Garbage collection** | Maintenance pool, batches of 500, yields between batches | ## Client Usage Example ```bash # 1. Check if file already exists by hash -HASH=$(sha256sum myfile.txt | cut -d' ' -f1) +HASH=$(b3sum myfile.txt | cut -d' ' -f1) curl -H "Authorization: Bearer $TOKEN" \ "https://oxicloud.example.com/api/dedup/check/$HASH" diff --git a/migrations/20260414000000_chunk_manifests.sql b/migrations/20260414000000_chunk_manifests.sql new file mode 100644 index 00000000..65ffdc67 --- /dev/null +++ b/migrations/20260414000000_chunk_manifests.sql @@ -0,0 +1,24 @@ +-- Content-Defined Chunking (CDC) manifests for sub-file deduplication. +-- +-- Each file uploaded via CDC is split into variable-size chunks (FastCDC). +-- The manifest records the ordered list of chunk hashes that compose the file. +-- Individual chunks are stored in storage.blobs (shared across manifests). +-- +-- Legacy whole-file blobs (pre-CDC) remain in storage.blobs and are +-- accessed directly when no matching manifest row exists. + +CREATE TABLE IF NOT EXISTS storage.chunk_manifests ( + file_hash VARCHAR(64) PRIMARY KEY, + chunk_hashes TEXT[] NOT NULL, + chunk_sizes BIGINT[] NOT NULL, + total_size BIGINT NOT NULL, + chunk_count INTEGER NOT NULL, + content_type TEXT, + ref_count INTEGER NOT NULL DEFAULT 1 CHECK (ref_count >= 0), + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +-- Index for GC: find manifests with no references. +CREATE INDEX IF NOT EXISTS idx_chunk_manifests_ref_count_zero + ON storage.chunk_manifests (file_hash) + WHERE ref_count = 0; diff --git a/src/application/ports/blob_storage_ports.rs b/src/application/ports/blob_storage_ports.rs index a5decd58..391fe164 100644 --- a/src/application/ports/blob_storage_ports.rs +++ b/src/application/ports/blob_storage_ports.rs @@ -54,6 +54,12 @@ pub trait BlobStorageBackend: Send + Sync + 'static { /// without overwriting. Returns the number of bytes stored. fn put_blob(&self, hash: &str, source_path: &Path) -> BoxFut<'_, Result>; + /// Store a blob from in-memory bytes (used by CDC chunk storage). + /// + /// Must be **idempotent**: if the blob already exists the call succeeds + /// without overwriting. Returns the number of bytes stored. + fn put_blob_from_bytes(&self, hash: &str, data: Bytes) -> BoxFut<'_, Result>; + /// Stream the full blob content in chunks. fn get_blob_stream(&self, hash: &str) -> BoxFut<'_, Result>; diff --git a/src/application/ports/dedup_ports.rs b/src/application/ports/dedup_ports.rs index 6d6f493e..95f219ea 100644 --- a/src/application/ports/dedup_ports.rs +++ b/src/application/ports/dedup_ports.rs @@ -28,16 +28,11 @@ pub struct BlobMetadataDto { #[derive(Debug, Clone)] pub enum DedupResultDto { /// New content was stored (first occurrence). - NewBlob { - hash: String, - size: u64, - blob_path: PathBuf, - }, + NewBlob { hash: String, size: u64 }, /// Content already existed; a reference was added instead. ExistingBlob { hash: String, size: u64, - blob_path: PathBuf, saved_bytes: u64, }, } @@ -57,13 +52,6 @@ impl DedupResultDto { } } - pub fn blob_path(&self) -> &Path { - match self { - DedupResultDto::NewBlob { blob_path, .. } => blob_path, - DedupResultDto::ExistingBlob { blob_path, .. } => blob_path, - } - } - pub fn was_deduplicated(&self) -> bool { matches!(self, DedupResultDto::ExistingBlob { .. }) } diff --git a/src/infrastructure/services/azure_blob_backend.rs b/src/infrastructure/services/azure_blob_backend.rs index daadb8a6..5f949a89 100644 --- a/src/infrastructure/services/azure_blob_backend.rs +++ b/src/infrastructure/services/azure_blob_backend.rs @@ -115,6 +115,29 @@ impl BlobStorageBackend for AzureBlobBackend { }) } + fn put_blob_from_bytes( + &self, + hash: &str, + data: Bytes, + ) -> Pin> + Send + '_>> { + let hash = hash.to_owned(); + Box::pin(async move { + let client = self.blob_client(&hash); + let size = data.len() as u64; + + // Idempotent: skip if exists + if client.get_properties().await.is_ok() { + return Ok(size); + } + + client.put_block_blob(data.to_vec()).await.map_err(|e| { + DomainError::internal_error("Azure", format!("Failed to upload blob {hash}: {e}")) + })?; + + Ok(size) + }) + } + fn get_blob_stream( &self, hash: &str, diff --git a/src/infrastructure/services/cached_blob_backend.rs b/src/infrastructure/services/cached_blob_backend.rs index ad3a15ab..502edd4a 100644 --- a/src/infrastructure/services/cached_blob_backend.rs +++ b/src/infrastructure/services/cached_blob_backend.rs @@ -12,6 +12,7 @@ use std::pin::Pin; use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; +use bytes::Bytes; use lru::LruCache; use std::num::NonZeroUsize; use tokio::fs; @@ -151,6 +152,37 @@ impl BlobStorageBackend for CachedBlobBackend { }) } + fn put_blob_from_bytes( + &self, + hash: &str, + data: Bytes, + ) -> Pin> + Send + '_>> { + let inner = self.inner.clone(); + let hash = hash.to_string(); + let self_ref = CachedRef { + cache_dir: self.cache_dir.clone(), + max_cache_bytes: self.max_cache_bytes, + index: self.index.clone(), + current_size: self.current_size.clone(), + }; + Box::pin(async move { + let size = inner.put_blob_from_bytes(&hash, data.clone()).await?; + // Also cache locally (best-effort): write bytes to cache path + let dest = self_ref.cached_path(&hash); + if let Some(parent) = dest.parent() { + let _ = fs::create_dir_all(parent).await; + } + let _ = fs::write(&dest, &data).await; + let data_len = data.len() as u64; + let mut idx = self_ref.index.lock().await; + if let Some(old) = idx.put(hash, CacheEntry { size: data_len }) { + self_ref.current_size.fetch_sub(old.size, Ordering::Relaxed); + } + self_ref.current_size.fetch_add(data_len, Ordering::Relaxed); + Ok(size) + }) + } + fn get_blob_stream( &self, hash: &str, diff --git a/src/infrastructure/services/chunked_upload_service.rs b/src/infrastructure/services/chunked_upload_service.rs index 4d5a3353..127f6584 100644 --- a/src/infrastructure/services/chunked_upload_service.rs +++ b/src/infrastructure/services/chunked_upload_service.rs @@ -517,7 +517,8 @@ impl ChunkedUploadService { let data_clone = data.clone(); // Bytes::clone is O(1) — just an Arc increment let actual_checksum = tokio::task::spawn_blocking(move || { use md5::{Digest, Md5}; - format!("{:x}", Md5::digest(&data_clone)) + let hash = Md5::digest(&data_clone); + hash.iter().map(|b| format!("{b:02x}")).collect::() }) .await .map_err(|e| format!("MD5 checksum task failed: {e}"))?; diff --git a/src/infrastructure/services/dedup_service.rs b/src/infrastructure/services/dedup_service.rs index 88c55f07..9d1d4903 100644 --- a/src/infrastructure/services/dedup_service.rs +++ b/src/infrastructure/services/dedup_service.rs @@ -1,35 +1,37 @@ -//! Content-Addressable Storage with Deduplication (PostgreSQL-backed) +//! Content-Addressable Storage with CDC Deduplication (PostgreSQL-backed) //! -//! Implements hash-based deduplication to eliminate redundant file storage. -//! Files are stored by their BLAKE3 hash, and multiple references can point -//! to the same physical blob. +//! Implements sub-file deduplication using FastCDC (content-defined chunking). +//! Files are split into variable-size chunks (64 KB – 1 MB, avg 256 KB) +//! using the FastCDC 2020 algorithm. Each chunk is BLAKE3-hashed and stored +//! independently in the blob backend. A *manifest* in PostgreSQL maps the +//! whole-file hash to the ordered list of chunk hashes that compose it. //! //! Architecture: //! ```text -//! ┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ -//! │ storage.files │────▶│ storage.blobs │────▶│ Blob Store │ -//! │ (references) │ │ (PG dedup index)│ │ (.blobs/ on FS) │ -//! └─────────────────┘ └─────────────────┘ └─────────────────┘ +//! ┌─────────────────┐ ┌─────────────────────┐ ┌─────────────┐ +//! │ storage.files │────▶│ chunk_manifests │────▶│ storage.blobs│──▶ Blob Store +//! │ (references) │ │ (file→[chunk_hashes])│ │ (chunks) │ +//! └─────────────────┘ └─────────────────────┘ └─────────────┘ //! ``` //! -//! The dedup index lives in PostgreSQL (`storage.blobs`) — no in-memory -//! HashMap, no JSON file, no WAL. +//! **Backward compatibility**: files uploaded before CDC (legacy whole-file +//! blobs in `storage.blobs`) are served transparently — when no manifest +//! row exists for a hash, the service falls back to direct blob reads. //! //! **Write-first strategy** (store_from_file): -//! 1. Write/move the blob file to disk *before* touching PostgreSQL. -//! 2. Single `INSERT … ON CONFLICT … RETURNING ref_count` upsert -//! (~2-4 ms) — no explicit transaction, no `SELECT FOR UPDATE`. -//! 3. PG connection is never held during disk I/O. -//! -//! `remove_reference` retains `SELECT … FOR UPDATE` inside a short -//! transaction because it must atomically decide whether to delete the -//! row *and* the blob file. +//! 1. CDC-analyse the file (mmap → FastCDC boundaries + per-chunk BLAKE3). +//! 2. Batch-check which chunk hashes already exist in PG (dedup skip). +//! 3. Read + upload only *new* chunks to the blob backend (idempotent). +//! 4. Bump ref_count for existing chunks (no disk I/O). +//! 5. Single manifest INSERT (~few ms total). +//! 6. PG connection is never held during disk I/O. //! //! Benefits: +//! - Sub-file dedup: edited files share unchanged chunks //! - ACID durability — crash-safe, zero orphaned index entries //! - PG connections never blocked by disk I/O (write-first) -//! - 30-50% storage reduction typical -//! - Faster uploads for existing content (instant dedup) +//! - 60-80% storage reduction for versioned / edited files +//! - Faster uploads when chunks already exist use bytes::Bytes; use futures::stream::{self, StreamExt}; @@ -40,6 +42,7 @@ use std::path::{Path, PathBuf}; use std::pin::Pin; use std::sync::Arc; use tokio::fs; +use tokio::io::{AsyncReadExt, AsyncSeekExt}; use crate::application::ports::blob_storage_ports::BlobStorageBackend; use crate::application::ports::dedup_ports::{ @@ -47,11 +50,30 @@ use crate::application::ports::dedup_ports::{ }; use crate::domain::errors::{DomainError, ErrorKind}; -/// Content-Addressable Storage Service (PostgreSQL-backed) +// ── CDC Constants ──────────────────────────────────────────────────────────── + +/// Minimum CDC chunk size (64 KB). +const CDC_MIN_CHUNK: usize = 65_536; +/// Average CDC chunk size (256 KB). +const CDC_AVG_CHUNK: usize = 262_144; +/// Maximum CDC chunk size (1 MB). +const CDC_MAX_CHUNK: usize = 1_048_576; + +// ── CDC helper types ───────────────────────────────────────────────────────── + +/// Metadata for a single CDC chunk (offset + length + BLAKE3 hash). +struct ChunkMeta { + hash: String, + offset: usize, + length: usize, +} + +/// Content-Addressable Storage Service with CDC (PostgreSQL-backed) /// -/// Delegates all byte-level I/O to a [`BlobStorageBackend`] implementation -/// (local filesystem, S3, etc.) while keeping BLAKE3 hashing, ref-counting -/// and the PostgreSQL dedup index here. +/// Splits files into variable-size chunks via FastCDC, stores each chunk +/// in the [`BlobStorageBackend`], and maintains a manifest in PostgreSQL +/// mapping file_hash → \[chunk_hashes\]. BLAKE3 hashing, ref-counting +/// and the PostgreSQL dedup index all live here. pub struct DedupService { /// Pluggable blob storage backend (local FS, S3, …). backend: Arc, @@ -102,23 +124,29 @@ impl DedupService { pub async fn initialize(&self) -> Result<(), DomainError> { self.backend.initialize().await?; - // Log existing blob stats from PG - let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM storage.blobs") + let blob_count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM storage.blobs") .fetch_one(self.pool.as_ref()) .await .unwrap_or(0); - let total_bytes: i64 = + let blob_bytes: i64 = sqlx::query_scalar("SELECT COALESCE(SUM(size), 0) FROM storage.blobs") .fetch_one(self.pool.as_ref()) .await .unwrap_or(0); + let manifest_count: i64 = + sqlx::query_scalar("SELECT COUNT(*) FROM storage.chunk_manifests") + .fetch_one(self.pool.as_ref()) + .await + .unwrap_or(0); + tracing::info!( - "Dedup service initialized (backend={}): {} blobs, {} bytes stored", + "Dedup service initialized (backend={}, CDC): {} chunk blobs ({} bytes), {} manifests", self.backend.backend_type(), - count, - total_bytes + blob_count, + blob_bytes, + manifest_count, ); Ok(()) @@ -138,18 +166,86 @@ impl DedupService { .unwrap_or_else(|| PathBuf::from(format!("remote://{}", hash))) } + // ── CDC analysis ─────────────────────────────────────────── + + /// Single-pass CDC: compute whole-file BLAKE3 hash + chunk boundaries + per-chunk hashes. + /// + /// Memory-maps the file and runs FastCDC boundary detection + /// concurrently with BLAKE3 hashing — all in one pass. + async fn cdc_hash_and_chunk_file(path: &Path) -> std::io::Result<(String, Vec)> { + let path = path.to_path_buf(); + tokio::task::spawn_blocking(move || { + let file = std::fs::File::open(&path)?; + let file_size = file.metadata()?.len(); + + if file_size == 0 { + return Ok((blake3::hash(b"").to_hex().to_string(), vec![])); + } + + // SAFETY: file is opened read-only; no concurrent writers expected + // (source is a temp upload file owned exclusively by this request). + let mmap = unsafe { memmap2::Mmap::map(&file)? }; + let chunker = + fastcdc::v2020::FastCDC::new(&mmap, CDC_MIN_CHUNK, CDC_AVG_CHUNK, CDC_MAX_CHUNK); + + let mut file_hasher = blake3::Hasher::new(); + let mut chunks = Vec::new(); + + for chunk in chunker { + let data = &mmap[chunk.offset..chunk.offset + chunk.length]; + file_hasher.update(data); + chunks.push(ChunkMeta { + hash: blake3::hash(data).to_hex().to_string(), + offset: chunk.offset, + length: chunk.length, + }); + } + + Ok((file_hasher.finalize().to_hex().to_string(), chunks)) + }) + .await + .expect("cdc_hash_and_chunk_file: spawn_blocking panicked") + } + + /// CDC analysis without file-hash computation (when hash is pre-computed). + async fn cdc_chunk_file(path: &Path) -> std::io::Result> { + let path = path.to_path_buf(); + tokio::task::spawn_blocking(move || { + let file = std::fs::File::open(&path)?; + let file_size = file.metadata()?.len(); + + if file_size == 0 { + return Ok(vec![]); + } + + let mmap = unsafe { memmap2::Mmap::map(&file)? }; + let chunker = + fastcdc::v2020::FastCDC::new(&mmap, CDC_MIN_CHUNK, CDC_AVG_CHUNK, CDC_MAX_CHUNK); + + let chunks: Vec = chunker + .map(|chunk| { + let data = &mmap[chunk.offset..chunk.offset + chunk.length]; + ChunkMeta { + hash: blake3::hash(data).to_hex().to_string(), + offset: chunk.offset, + length: chunk.length, + } + }) + .collect(); + + Ok(chunks) + }) + .await + .expect("cdc_chunk_file: spawn_blocking panicked") + } + // ── Hash helpers ───────────────────────────────────────────── /// Calculate BLAKE3 hash of a file (~5× faster than SHA-256). /// - /// Runs entirely on `spawn_blocking` with synchronous I/O so the Tokio - /// worker threads are never blocked by CPU-bound hashing. - /// - /// Uses memory-mapped I/O (`update_mmap_rayon`) which avoids loading the - /// entire file into the heap. The OS pages in data on demand and BLAKE3 - /// parallelises the computation across all available cores via rayon. - /// Peak RAM for a 500 MB file is only a few MB of active pages instead - /// of the full 500 MB. + /// Uses memory-mapped I/O with rayon parallelism. Kept for callers + /// that only need the hash (e.g. upload handlers pre-computing the hash + /// before calling `store_from_file`). pub async fn hash_file(path: &Path) -> std::io::Result { let path = path.to_path_buf(); tokio::task::spawn_blocking(move || { @@ -163,75 +259,323 @@ impl DedupService { // ── Core store operations ──────────────────────────────────── - /// Store content with deduplication (streaming from file). + /// Store content with CDC deduplication (from file). /// - /// **Write-first strategy**: the source file is moved/uploaded to the - /// blob backend *before* touching PostgreSQL, so the PG connection is - /// never held during I/O. + /// **Fast path**: if `pre_computed_hash` is `Some`, the manifest / + /// legacy-blob index is checked *before* running CDC — returning + /// instantly on a full-file dedup hit. /// - /// If `pre_computed_hash` is `Some`, the file will NOT be re-read for - /// BLAKE3 — saving one full sequential read (the biggest I/O win). + /// **New-file path**: CDC-analyses the file (single mmap pass), + /// stores unique chunks via the blob backend, then inserts the + /// manifest in PostgreSQL. pub async fn store_from_file( &self, source_path: &Path, content_type: Option, pre_computed_hash: Option, ) -> Result { - // Use pre-computed hash if available, otherwise calculate (streaming) - let hash = match pre_computed_hash { - Some(h) => h, - None => Self::hash_file(source_path) + // ── Fast path: pre-computed hash → check before CDC ────── + if let Some(ref hash) = pre_computed_hash + && let Some(result) = self.try_dedup_hit(hash, source_path).await? + { + return Ok(result); + } + + // ── CDC analysis ───────────────────────────────────────── + let (file_hash, chunks) = if let Some(hash) = pre_computed_hash { + let chunks = Self::cdc_chunk_file(source_path) .await - .map_err(DomainError::from)?, + .map_err(DomainError::from)?; + (hash, chunks) + } else { + let (hash, chunks) = Self::cdc_hash_and_chunk_file(source_path) + .await + .map_err(DomainError::from)?; + // Check dedup with newly computed hash + if let Some(result) = self.try_dedup_hit(&hash, source_path).await? { + return Ok(result); + } + (hash, chunks) }; - // ── Phase 1: Place blob in backend (NO PG connection held) ─── - let file_size = self.backend.put_blob(&hash, source_path).await?; + let file_size = fs::metadata(source_path) + .await + .map_err(DomainError::from)? + .len(); - let blob_path = self.blob_path(&hash); + // ── Store chunks (write-first — no PG connection held) ─── + let (chunk_hashes, chunk_sizes) = self.store_chunks(source_path, &chunks).await?; - // ── Phase 2: Single atomic upsert (~2-4 ms, no explicit TX) ─ - let ref_count: i32 = sqlx::query_scalar( - "INSERT INTO storage.blobs (hash, size, ref_count, content_type) - VALUES ($1, $2, 1, $3) - ON CONFLICT (hash) DO UPDATE SET ref_count = storage.blobs.ref_count + 1 - RETURNING ref_count", + // ── Insert manifest ────────────────────────────────────── + sqlx::query( + "INSERT INTO storage.chunk_manifests + (file_hash, chunk_hashes, chunk_sizes, total_size, chunk_count, content_type, ref_count) + VALUES ($1, $2, $3, $4, $5, $6, 1)", ) - .bind(&hash) + .bind(&file_hash) + .bind(&chunk_hashes) + .bind(chunk_sizes.iter().map(|s| *s as i64).collect::>()) .bind(file_size as i64) + .bind(chunk_hashes.len() as i32) .bind(&content_type) - .fetch_one(self.pool.as_ref()) + .execute(self.pool.as_ref()) .await .map_err(|e| { - DomainError::internal_error("Dedup", format!("Failed to upsert blob: {}", e)) + DomainError::internal_error("Dedup", format!("Failed to insert manifest: {}", e)) })?; - if ref_count > 1 { + // ── Clean up source file ───────────────────────────────── + let _ = fs::remove_file(source_path).await; + + tracing::info!( + "NEW BLOB (CDC): {} ({} bytes, {} chunks)", + &file_hash[..12], + file_size, + chunk_hashes.len() + ); + + Ok(DedupResultDto::NewBlob { + hash: file_hash, + size: file_size, + }) + } + + /// Check manifest or legacy blob for a dedup hit. + /// + /// Returns `Some(ExistingBlob)` if the exact file was already stored. + /// Bumps the appropriate ref_count and removes the source file. + async fn try_dedup_hit( + &self, + hash: &str, + source_path: &Path, + ) -> Result, DomainError> { + // ── CDC manifest hit ───────────────────────────────────── + let manifest = sqlx::query_as::<_, (i64,)>( + "SELECT total_size FROM storage.chunk_manifests WHERE file_hash = $1", + ) + .bind(hash) + .fetch_optional(self.pool.as_ref()) + .await + .map_err(|e| { + DomainError::internal_error("Dedup", format!("Failed to check manifest: {}", e)) + })?; + + if let Some((total_size,)) = manifest { + sqlx::query( + "UPDATE storage.chunk_manifests SET ref_count = ref_count + 1 WHERE file_hash = $1", + ) + .bind(hash) + .execute(self.pool.as_ref()) + .await + .map_err(|e| { + DomainError::internal_error( + "Dedup", + format!("Failed to bump manifest ref_count: {}", e), + ) + })?; + + let _ = fs::remove_file(source_path).await; + tracing::info!( - "DEDUP HIT (file): {} ({} bytes saved)", + "DEDUP HIT (manifest): {} ({} bytes saved)", &hash[..12], - file_size + total_size ); - Ok(DedupResultDto::ExistingBlob { - hash, - size: file_size, - blob_path, - saved_bytes: file_size, - }) - } else { - tracing::info!("NEW BLOB (file): {} ({} bytes)", &hash[..12], file_size); - Ok(DedupResultDto::NewBlob { - hash, - size: file_size, - blob_path, - }) + return Ok(Some(DedupResultDto::ExistingBlob { + hash: hash.to_owned(), + size: total_size as u64, + saved_bytes: total_size as u64, + })); } + + // ── Legacy whole-file blob hit ─────────────────────────── + let legacy = sqlx::query_as::<_, (i64,)>("SELECT size FROM storage.blobs WHERE hash = $1") + .bind(hash) + .fetch_optional(self.pool.as_ref()) + .await + .map_err(|e| { + DomainError::internal_error("Dedup", format!("Failed to check legacy blob: {}", e)) + })?; + + if let Some((size,)) = legacy { + sqlx::query("UPDATE storage.blobs SET ref_count = ref_count + 1 WHERE hash = $1") + .bind(hash) + .execute(self.pool.as_ref()) + .await + .map_err(|e| { + DomainError::internal_error( + "Dedup", + format!("Failed to bump legacy ref_count: {}", e), + ) + })?; + + let _ = fs::remove_file(source_path).await; + tracing::info!( + "DEDUP HIT (legacy blob): {} ({} bytes saved)", + &hash[..12], + size + ); + return Ok(Some(DedupResultDto::ExistingBlob { + hash: hash.to_owned(), + size: size as u64, + saved_bytes: size as u64, + })); + } + + Ok(None) + } + + /// Maximum concurrent chunk uploads to the blob backend. + const CHUNK_UPLOAD_CONCURRENCY: usize = 8; + + /// Store CDC chunks via the blob backend + upsert in PG. + /// + /// Phase 0: Batch-queries PG to discover which chunk hashes already + /// exist in `storage.blobs`. + /// Phase 1: Reads only *new* chunks from the source file (the biggest + /// I/O saving for versioned files where most chunks are unchanged). + /// Phase 2: Parallel operations — uploads new chunks, bumps ref_count + /// for existing ones — with up to [`CHUNK_UPLOAD_CONCURRENCY`] in flight. + async fn store_chunks( + &self, + source_path: &Path, + chunks: &[ChunkMeta], + ) -> Result<(Vec, Vec), DomainError> { + let pool = &self.pool; + let backend = &self.backend; + + // ── Phase 0: Batch-check which chunks already exist ────── + let unique_hashes: Vec = { + let mut seen = std::collections::HashSet::new(); + chunks + .iter() + .filter_map(|c| { + if seen.insert(c.hash.as_str()) { + Some(c.hash.clone()) + } else { + None + } + }) + .collect() + }; + + let existing_hashes: std::collections::HashSet = + sqlx::query_scalar::<_, String>("SELECT hash FROM storage.blobs WHERE hash = ANY($1)") + .bind(&unique_hashes) + .fetch_all(pool.as_ref()) + .await + .map_err(|e| { + DomainError::internal_error( + "Dedup", + format!("Failed to check existing chunks: {}", e), + ) + })? + .into_iter() + .collect(); + + // ── Phase 1: Read only NEW chunks from disk ────────────── + let mut file = tokio::fs::File::open(source_path).await.map_err(|e| { + DomainError::internal_error("Dedup", format!("Failed to open source file: {}", e)) + })?; + + // (hash, Option, size) — None = existing chunk (skip I/O), + // Some = new chunk (needs upload). + let mut chunk_ops: Vec<(String, Option, u64)> = Vec::with_capacity(chunks.len()); + + for chunk in chunks { + let size = chunk.length as u64; + if existing_hashes.contains(&chunk.hash) { + chunk_ops.push((chunk.hash.clone(), None, size)); + } else { + file.seek(std::io::SeekFrom::Start(chunk.offset as u64)) + .await + .map_err(|e| { + DomainError::internal_error("Dedup", format!("Failed to seek: {}", e)) + })?; + let mut buf = vec![0u8; chunk.length]; + file.read_exact(&mut buf).await.map_err(|e| { + DomainError::internal_error("Dedup", format!("Failed to read chunk: {}", e)) + })?; + chunk_ops.push((chunk.hash.clone(), Some(Bytes::from(buf)), size)); + } + } + + // ── Phase 2: Parallel upload (new) / ref-bump (existing) ─ + let results: Vec> = stream::iter(chunk_ops) + .map(|(hash, data, size)| async move { + if let Some(bytes) = data { + // New chunk: upload to blob backend + INSERT/upsert + backend.put_blob_from_bytes(&hash, bytes).await?; + sqlx::query( + "INSERT INTO storage.blobs (hash, size, ref_count) + VALUES ($1, $2, 1) + ON CONFLICT (hash) DO UPDATE + SET ref_count = storage.blobs.ref_count + 1", + ) + .bind(&hash) + .bind(size as i64) + .execute(pool.as_ref()) + .await + .map_err(|e| { + DomainError::internal_error( + "Dedup", + format!("Failed to upsert chunk: {}", e), + ) + })?; + } else { + // Existing chunk: just bump ref_count (no I/O) + sqlx::query( + "UPDATE storage.blobs + SET ref_count = ref_count + 1 + WHERE hash = $1", + ) + .bind(&hash) + .execute(pool.as_ref()) + .await + .map_err(|e| { + DomainError::internal_error( + "Dedup", + format!("Failed to bump ref_count: {}", e), + ) + })?; + } + Ok(()) + }) + .buffer_unordered(Self::CHUNK_UPLOAD_CONCURRENCY) + .collect() + .await; + + // All operations must succeed. Order preservation is not needed + // here — chunk_hashes/chunk_sizes are derived from the input + // `chunks` slice which keeps the original CDC order. + for result in results { + result?; + } + + let chunk_hashes: Vec = chunks.iter().map(|c| c.hash.clone()).collect(); + let chunk_sizes: Vec = chunks.iter().map(|c| c.length as u64).collect(); + + Ok((chunk_hashes, chunk_sizes)) } // ── Reference counting ─────────────────────────────────────── - /// Check if a blob with the given hash exists in the PG index. + /// Check if a blob with the given hash exists (manifest or legacy). pub async fn blob_exists(&self, hash: &str) -> bool { + // Check manifest first + let manifest = sqlx::query_scalar::<_, bool>( + "SELECT EXISTS(SELECT 1 FROM storage.chunk_manifests WHERE file_hash = $1)", + ) + .bind(hash) + .fetch_one(self.pool.as_ref()) + .await + .unwrap_or(false); + + if manifest { + return true; + } + + // Legacy blob sqlx::query_scalar::<_, bool>("SELECT EXISTS(SELECT 1 FROM storage.blobs WHERE hash = $1)") .bind(hash) .fetch_one(self.pool.as_ref()) @@ -241,9 +585,6 @@ impl DedupService { /// Returns `true` if `user_id` owns at least one (non-trashed) file that /// references the blob identified by `hash`. - /// - /// Used by the dedup API handlers to enforce per-user access control on - /// the content-addressed blob store. pub async fn user_owns_blob_reference(&self, hash: &str, user_id: &str) -> bool { sqlx::query_scalar::<_, bool>( "SELECT EXISTS(SELECT 1 FROM storage.files WHERE blob_hash = $1 AND user_id = $2 AND NOT is_trashed)", @@ -255,8 +596,29 @@ impl DedupService { .unwrap_or(false) } - /// Get metadata for a blob from PostgreSQL. + /// Get metadata for a blob (manifest-aware with legacy fallback). pub async fn get_blob_metadata(&self, hash: &str) -> Option { + // Check manifest first + let manifest = sqlx::query_as::<_, (i64, i32, Option)>( + "SELECT total_size, ref_count, content_type + FROM storage.chunk_manifests WHERE file_hash = $1", + ) + .bind(hash) + .fetch_optional(self.pool.as_ref()) + .await + .ok() + .flatten(); + + if let Some((total_size, ref_count, content_type)) = manifest { + return Some(BlobMetadataDto { + hash: hash.to_owned(), + size: total_size as u64, + ref_count: ref_count as u32, + content_type, + }); + } + + // Legacy blob let row = sqlx::query_as::<_, (String, i64, i32, Option)>( "SELECT hash, size, ref_count, content_type FROM storage.blobs WHERE hash = $1", ) @@ -274,8 +636,25 @@ impl DedupService { }) } - /// Add a reference to a blob (increment ref_count). + /// Add a reference (manifest-aware with legacy fallback). pub async fn add_reference(&self, hash: &str) -> Result<(), DomainError> { + // Try manifest first + let manifest_affected = sqlx::query( + "UPDATE storage.chunk_manifests SET ref_count = ref_count + 1 WHERE file_hash = $1", + ) + .bind(hash) + .execute(self.pool.as_ref()) + .await + .map_err(|e| { + DomainError::internal_error("Dedup", format!("Failed to add manifest ref: {}", e)) + })? + .rows_affected(); + + if manifest_affected > 0 { + return Ok(()); + } + + // Legacy blob let rows_affected = sqlx::query("UPDATE storage.blobs SET ref_count = ref_count + 1 WHERE hash = $1") .bind(hash) @@ -300,12 +679,129 @@ impl DedupService { Ok(()) } - /// Remove a reference from a blob. + /// Remove a reference from a blob (manifest-aware with legacy fallback). /// - /// Uses a single transaction with `SELECT … FOR UPDATE` to atomically - /// decrement ref_count and delete the row + blob file if it reaches 0. - /// Returns `true` if the blob was deleted. + /// For CDC manifests: decrements manifest ref_count. When it reaches 0 + /// the manifest is deleted and all chunk ref_counts are decremented; + /// chunks that reach 0 are deleted from both PG and the blob backend. + /// + /// For legacy blobs: uses a single TX with `SELECT … FOR UPDATE`. pub async fn remove_reference(&self, hash: &str) -> Result { + // ── CDC manifest path ──────────────────────────────────── + let manifest = sqlx::query_as::<_, (i32, Vec)>( + "SELECT ref_count, chunk_hashes FROM storage.chunk_manifests WHERE file_hash = $1", + ) + .bind(hash) + .fetch_optional(self.pool.as_ref()) + .await + .map_err(|e| DomainError::internal_error("Dedup", format!("Manifest lookup: {}", e)))?; + + if let Some((ref_count, chunk_hashes)) = manifest { + return self + .remove_manifest_reference(hash, ref_count, &chunk_hashes) + .await; + } + + // ── Legacy whole-file blob path ────────────────────────── + self.remove_legacy_reference(hash).await + } + + /// Remove a manifest reference. Handles chunk cleanup when last ref is removed. + async fn remove_manifest_reference( + &self, + file_hash: &str, + _initial_ref_count: i32, + chunk_hashes: &[String], + ) -> Result { + let mut tx = self.pool.begin().await.map_err(|e| { + DomainError::internal_error("Dedup", format!("Failed to begin TX: {}", e)) + })?; + + // Lock manifest row + let current_rc = sqlx::query_scalar::<_, i32>( + "SELECT ref_count FROM storage.chunk_manifests WHERE file_hash = $1 FOR UPDATE", + ) + .bind(file_hash) + .fetch_optional(&mut *tx) + .await + .map_err(|e| DomainError::internal_error("Dedup", format!("Lock manifest: {}", e)))?; + + let Some(current_rc) = current_rc else { + tx.rollback().await.ok(); + return Ok(false); + }; + + if current_rc <= 1 { + // Last reference — delete manifest and decrement chunks + sqlx::query("DELETE FROM storage.chunk_manifests WHERE file_hash = $1") + .bind(file_hash) + .execute(&mut *tx) + .await + .map_err(|e| { + DomainError::internal_error("Dedup", format!("Delete manifest: {}", e)) + })?; + + // Batch decrement chunk ref_counts + sqlx::query("UPDATE storage.blobs SET ref_count = ref_count - 1 WHERE hash = ANY($1)") + .bind(chunk_hashes) + .execute(&mut *tx) + .await + .map_err(|e| { + DomainError::internal_error("Dedup", format!("Decrement chunks: {}", e)) + })?; + + // Find chunks that reached 0 + let zero_chunks: Vec = sqlx::query_scalar( + "DELETE FROM storage.blobs WHERE hash = ANY($1) AND ref_count <= 0 RETURNING hash", + ) + .bind(chunk_hashes) + .fetch_all(&mut *tx) + .await + .map_err(|e| { + DomainError::internal_error("Dedup", format!("Delete zero chunks: {}", e)) + })?; + + tx.commit() + .await + .map_err(|e| DomainError::internal_error("Dedup", format!("Commit: {}", e)))?; + + // Delete blob files AFTER commit + for chunk_hash in &zero_chunks { + if let Err(e) = self.backend.delete_blob(chunk_hash).await { + tracing::warn!("Failed to delete chunk blob {}: {}", chunk_hash, e); + } + } + + tracing::info!( + "MANIFEST DELETED: {} ({} chunks, {} orphan chunks removed)", + &file_hash[..12], + chunk_hashes.len(), + zero_chunks.len() + ); + Ok(true) + } else { + // Still has references — just decrement + sqlx::query( + "UPDATE storage.chunk_manifests SET ref_count = ref_count - 1 WHERE file_hash = $1", + ) + .bind(file_hash) + .execute(&mut *tx) + .await + .map_err(|e| { + DomainError::internal_error("Dedup", format!("Decrement manifest: {}", e)) + })?; + + tx.commit() + .await + .map_err(|e| DomainError::internal_error("Dedup", format!("Commit: {}", e)))?; + + tracing::debug!("Reference removed from manifest {}", &file_hash[..12]); + Ok(false) + } + } + + /// Remove a reference from a legacy whole-file blob. + async fn remove_legacy_reference(&self, hash: &str) -> Result { let mut tx = self.pool.begin().await.map_err(|e| { DomainError::internal_error("Dedup", format!("Failed to begin transaction: {}", e)) })?; @@ -380,16 +876,52 @@ impl DedupService { // ── Read operations ────────────────────────────────────────── - /// Stream blob content in chunks — constant memory usage. + /// Stream blob content — CDC-aware with legacy fallback. + /// + /// For CDC files: looks up the manifest, then streams chunks in order, + /// concatenating them into a single byte stream. + /// For legacy blobs: delegates directly to the backend. pub async fn read_blob_stream( &self, hash: &str, ) -> Result> + Send>>, DomainError> { - self.backend.get_blob_stream(hash).await + // Check manifest + let manifest = sqlx::query_scalar::<_, Vec>( + "SELECT chunk_hashes FROM storage.chunk_manifests WHERE file_hash = $1", + ) + .bind(hash) + .fetch_optional(self.pool.as_ref()) + .await + .map_err(|e| DomainError::internal_error("Dedup", format!("Manifest lookup: {}", e)))?; + + if let Some(chunk_hashes) = manifest { + // CDC file: stream chunks in order + let backend = self.backend.clone(); + let chunk_stream = stream::iter(chunk_hashes) + .map(move |chunk_hash| { + let backend = backend.clone(); + async move { + backend + .get_blob_stream(&chunk_hash) + .await + .map_err(|e| std::io::Error::other(e.to_string())) + } + }) + .buffered(1) + .try_flatten(); + + Ok(Box::pin(chunk_stream)) + } else { + // Legacy whole-file blob + self.backend.get_blob_stream(hash).await + } } - /// Stream a byte range of a blob — only reads the requested portion. + /// Stream a byte range — CDC-aware with legacy fallback. + /// + /// For CDC files: calculates which chunks overlap the requested range, + /// then streams only the relevant portions. pub async fn read_blob_range_stream( &self, hash: &str, @@ -397,32 +929,134 @@ impl DedupService { end: Option, ) -> Result> + Send>>, DomainError> { - self.backend.get_blob_range_stream(hash, start, end).await + // Check manifest + let manifest = sqlx::query_as::<_, (Vec, Vec, i64)>( + "SELECT chunk_hashes, chunk_sizes, total_size + FROM storage.chunk_manifests WHERE file_hash = $1", + ) + .bind(hash) + .fetch_optional(self.pool.as_ref()) + .await + .map_err(|e| DomainError::internal_error("Dedup", format!("Manifest lookup: {}", e)))?; + + if let Some((chunk_hashes, chunk_sizes, total_size)) = manifest { + let end = end.unwrap_or(total_size as u64); + + // Calculate which chunks overlap [start, end) + let mut offset: u64 = 0; + // (chunk_hash, range_start_within_chunk, range_end_within_chunk) + let mut selected: Vec<(String, u64, Option)> = Vec::new(); + + for (i, &chunk_size) in chunk_sizes.iter().enumerate() { + let chunk_size = chunk_size as u64; + let chunk_end = offset + chunk_size; + + if chunk_end > start && offset < end { + let range_start = start.saturating_sub(offset); + let range_end = if chunk_end > end { + Some(end - offset) + } else { + None + }; + selected.push((chunk_hashes[i].clone(), range_start, range_end)); + } + + offset += chunk_size; + if offset >= end { + break; + } + } + + // Stream selected chunks with ranges + let backend = self.backend.clone(); + let chunk_stream = stream::iter(selected) + .map(move |(chunk_hash, range_start, range_end)| { + let backend = backend.clone(); + async move { + backend + .get_blob_range_stream(&chunk_hash, range_start, range_end) + .await + .map_err(|e| std::io::Error::other(e.to_string())) + } + }) + .buffered(1) + .try_flatten(); + + Ok(Box::pin(chunk_stream)) + } else { + // Legacy whole-file blob + self.backend.get_blob_range_stream(hash, start, end).await + } } - /// Get the size of a blob without reading its content. + /// Get blob size — manifest-aware with legacy fallback. pub async fn blob_size(&self, hash: &str) -> Result { + // Check manifest first (O(1) from PG) + let manifest_size = sqlx::query_scalar::<_, i64>( + "SELECT total_size FROM storage.chunk_manifests WHERE file_hash = $1", + ) + .bind(hash) + .fetch_optional(self.pool.as_ref()) + .await + .map_err(|e| DomainError::internal_error("Dedup", format!("Manifest lookup: {}", e)))?; + + if let Some(size) = manifest_size { + return Ok(size as u64); + } + + // Legacy: delegate to backend self.backend.blob_size(hash).await } // ── Statistics (computed from PG) ──────────────────────────── - /// Get deduplication statistics by querying PostgreSQL. + /// Get deduplication statistics (CDC + legacy). pub async fn get_stats(&self) -> DedupStatsDto { - let row = sqlx::query_as::<_, (i64, i64, i64)>( - "SELECT - COUNT(*) AS total_blobs, - COALESCE(SUM(size), 0) AS total_bytes_stored, - COALESCE(SUM(size::BIGINT * ref_count), 0) AS total_bytes_referenced - FROM storage.blobs", + // Physical storage (all blobs = chunks + legacy) + let (total_blobs, total_bytes_stored): (i64, i64) = + sqlx::query_as("SELECT COUNT(*), COALESCE(SUM(size), 0) FROM storage.blobs") + .fetch_one(self.pool.as_ref()) + .await + .unwrap_or((0, 0)); + + // Referenced bytes from CDC manifests + let manifest_referenced: i64 = sqlx::query_scalar( + "SELECT COALESCE(SUM(total_size::BIGINT * ref_count), 0) FROM storage.chunk_manifests", ) .fetch_one(self.pool.as_ref()) .await - .unwrap_or((0, 0, 0)); + .unwrap_or(0); - let total_blobs = row.0 as u64; - let total_bytes_stored = row.1 as u64; - let total_bytes_referenced = row.2 as u64; + // Referenced bytes from legacy blobs (those not used as CDC chunks). + // A legacy blob has its hash directly in storage.files.blob_hash. + // We approximate by subtracting manifest-attributed storage. + let all_blob_referenced: i64 = sqlx::query_scalar( + "SELECT COALESCE(SUM(size::BIGINT * ref_count), 0) FROM storage.blobs", + ) + .fetch_one(self.pool.as_ref()) + .await + .unwrap_or(0); + + let manifest_count: i64 = + sqlx::query_scalar("SELECT COUNT(*) FROM storage.chunk_manifests") + .fetch_one(self.pool.as_ref()) + .await + .unwrap_or(0); + + // If manifests exist, use manifest-based referenced bytes; + // otherwise fall back to pure legacy calculation. + let total_bytes_referenced = if manifest_count > 0 { + // Legacy blobs that aren't chunks contribute directly; + // CDC manifests contribute total_size × ref_count. + // Approximation: all_blob_referenced overcounts chunk sharing, + // but manifest_referenced accounts for file-level dedup. + manifest_referenced.max(all_blob_referenced) as u64 + } else { + all_blob_referenced as u64 + }; + + let total_blobs = total_blobs as u64; + let total_bytes_stored = total_bytes_stored as u64; let bytes_saved = total_bytes_referenced.saturating_sub(total_bytes_stored); let dedup_ratio = if total_bytes_stored > 0 { total_bytes_referenced as f64 / total_bytes_stored as f64 @@ -435,30 +1069,77 @@ impl DedupService { total_bytes_stored, total_bytes_referenced, bytes_saved, - dedup_hits: 0, // Not tracked per-session — derive from SUM(ref_count - 1) + dedup_hits: 0, dedup_ratio, } } // ── Maintenance ────────────────────────────────────────────── - /// Verify integrity of all blobs (PG index vs filesystem). + /// Verify integrity of all stored data (manifests + blobs). /// - /// Uses a **streaming cursor** (`fetch()`) so memory stays O(batch) - /// instead of O(total_blobs). Blobs are verified in micro-batches - /// of `VERIFY_CONCURRENCY` using `buffer_unordered`. + /// For CDC manifests: verifies chunk count, total_size consistency, + /// and that every referenced chunk exists in the backend. + /// For blobs (chunks + legacy): verifies existence, size, and + /// (for local backends) re-hashes to confirm content integrity. pub async fn verify_integrity(&self) -> Result, DomainError> { - /// Max blobs verified concurrently. Each spawns a blocking - /// thread for BLAKE3 so this also caps blocking-pool pressure. const VERIFY_CONCURRENCY: usize = 16; + let mut issues = Vec::new(); + // ── Phase 1: Verify CDC manifests ──────────────────────── + let manifests: Vec<(String, Vec, Vec, i64)> = sqlx::query_as( + "SELECT file_hash, chunk_hashes, chunk_sizes, total_size + FROM storage.chunk_manifests", + ) + .fetch_all(self.maintenance_pool.as_ref()) + .await + .map_err(|e| DomainError::internal_error("Dedup", format!("List manifests: {}", e)))?; + + for (file_hash, chunk_hashes, chunk_sizes, total_size) in &manifests { + let label = &file_hash[..file_hash.len().min(12)]; + + if chunk_hashes.len() != chunk_sizes.len() { + issues.push(format!( + "Manifest {label}: chunk_hashes/chunk_sizes length mismatch" + )); + continue; + } + + let sum: i64 = chunk_sizes.iter().sum(); + if sum != *total_size { + issues.push(format!( + "Manifest {label}: total_size {total_size} != sum of chunk_sizes {sum}" + )); + } + + for (i, chunk_hash) in chunk_hashes.iter().enumerate() { + let chunk_label = &chunk_hash[..chunk_hash.len().min(12)]; + match self.backend.blob_size(chunk_hash).await { + Ok(actual_size) => { + if actual_size != chunk_sizes[i] as u64 { + issues.push(format!( + "Manifest {label} chunk {chunk_label}: size mismatch \ + (expected {}, actual {actual_size})", + chunk_sizes[i] + )); + } + } + Err(_) => { + issues.push(format!( + "Manifest {label} chunk {chunk_label}: missing in backend" + )); + } + } + } + } + + // ── Phase 2: Verify blobs (chunks + legacy) ────────────── let mut row_stream = sqlx::query_as::<_, (String, i64)>( "SELECT hash, size FROM storage.blobs ORDER BY hash", ) .fetch(self.maintenance_pool.as_ref()); let mut total = 0usize; - let mut corrupted = Vec::::new(); let mut batch = Vec::with_capacity(VERIFY_CONCURRENCY); loop { @@ -473,19 +1154,17 @@ impl DedupService { batch.push(row); } - // Flush when batch is full or we've exhausted the cursor if batch.len() >= VERIFY_CONCURRENCY || (is_done && !batch.is_empty()) { let backend = self.backend.clone(); let current_batch = std::mem::replace(&mut batch, Vec::with_capacity(VERIFY_CONCURRENCY)); - let issues: Vec = stream::iter(current_batch) + let blob_issues: Vec = stream::iter(current_batch) .map(move |(hash, expected_size)| { let backend = backend.clone(); async move { let mut issues = Vec::new(); - // Check existence + size via backend match backend.blob_size(&hash).await { Ok(actual_size) => { if actual_size != expected_size as u64 { @@ -501,7 +1180,6 @@ impl DedupService { } }; - // Verify hash — only possible for local backends if let Some(blob_path) = backend.local_blob_path(&hash) { match Self::hash_file(&blob_path).await { Ok(actual_hash) => { @@ -526,7 +1204,7 @@ impl DedupService { .collect() .await; - corrupted.extend(issues); + issues.extend(blob_issues); } if is_done { @@ -534,41 +1212,81 @@ impl DedupService { } } - if corrupted.is_empty() { - tracing::info!("Integrity check passed for {} blobs", total); + if issues.is_empty() { + tracing::info!( + "Integrity check passed ({} manifests, {} blobs)", + manifests.len(), + total + ); } else { - tracing::warn!("Integrity check found {} issues", corrupted.len()); + tracing::warn!("Integrity check found {} issues", issues.len()); } - Ok(corrupted) + Ok(issues) } - /// Garbage collect orphaned blobs (ref_count = 0). + /// Garbage collect orphaned manifests and blobs. /// - /// Deletes in small batches (BATCH_SIZE rows per TX) so that each - /// transaction lasts only a few milliseconds. This avoids: - /// - massive row-lock accumulation in PostgreSQL, - /// - WAL bloat from a single giant DELETE, - /// - blocking concurrent uploads that touch `storage.blobs`. - /// - /// Blob files are removed **after** each batch commits, so a crash - /// mid-GC only leaves a few orphan files on disk (reclaimed next run). + /// Phase 1: Delete manifests with ref_count = 0, then decrement + /// chunk ref_counts for their chunks. + /// Phase 2: Delete blobs (chunks + legacy) with ref_count = 0. pub async fn garbage_collect(&self) -> Result<(u64, u64), DomainError> { - /// Max rows deleted per mini-transaction. const BATCH_SIZE: i64 = 500; let mut total_deleted = 0u64; let mut total_bytes = 0u64; + // ── Phase 1: GC orphaned manifests ─────────────────────── + loop { + let batch: Vec<(String, Vec, i64)> = sqlx::query_as( + "DELETE FROM storage.chunk_manifests + WHERE ctid = ANY( + SELECT ctid FROM storage.chunk_manifests + WHERE ref_count <= 0 + LIMIT $1 + ) + RETURNING file_hash, chunk_hashes, total_size", + ) + .bind(BATCH_SIZE) + .fetch_all(self.maintenance_pool.as_ref()) + .await + .map_err(|e| DomainError::internal_error("Dedup", format!("GC manifests: {e}")))?; + + if batch.is_empty() { + break; + } + + for (file_hash, chunk_hashes, size) in &batch { + // Decrement chunk ref_counts + sqlx::query( + "UPDATE storage.blobs SET ref_count = ref_count - 1 WHERE hash = ANY($1)", + ) + .bind(chunk_hashes) + .execute(self.maintenance_pool.as_ref()) + .await + .map_err(|e| { + DomainError::internal_error("Dedup", format!("GC decrement chunks: {e}")) + })?; + + total_bytes += *size as u64; + tracing::debug!( + "GC: removed manifest {} ({} chunks)", + &file_hash[..file_hash.len().min(12)], + chunk_hashes.len() + ); + } + total_deleted += batch.len() as u64; + + tokio::task::yield_now().await; + } + + // ── Phase 2: GC orphaned blobs/chunks ──────────────────── loop { - // Each DELETE is its own implicit TX — short and bounded. - // The `ctid` sub-select is the canonical way to do - // `DELETE … LIMIT` in PostgreSQL. let batch: Vec<(String, i64)> = sqlx::query_as( "DELETE FROM storage.blobs WHERE ctid = ANY( SELECT ctid FROM storage.blobs - WHERE ref_count = 0 + WHERE ref_count <= 0 LIMIT $1 ) RETURNING hash, size", @@ -576,22 +1294,17 @@ impl DedupService { .bind(BATCH_SIZE) .fetch_all(self.maintenance_pool.as_ref()) .await - .map_err(|e| DomainError::internal_error("Dedup", format!("GC batch failed: {e}")))?; + .map_err(|e| DomainError::internal_error("Dedup", format!("GC blobs: {e}")))?; if batch.is_empty() { break; } - // Delete blob files OUTSIDE the TX (already committed). - // Also clean up any thumbnail files for these blob hashes - // (thumbnails are keyed by blob_hash and live under - // storage_root/.thumbnails/{icon,preview,large}/{hash}.jpg). - for (hash, size) in &batch { if let Err(e) = self.backend.delete_blob(hash).await { - tracing::warn!("Failed to delete orphan blob file {hash}: {e}"); + tracing::warn!("Failed to delete orphan blob {hash}: {e}"); } - // Remove associated thumbnail files (best-effort, always local) + // Clean up thumbnails (best-effort, only local backends) if let Some(blob_path) = self.backend.local_blob_path(hash) && let Some(storage_root) = blob_path.ancestors().nth(3) { @@ -605,12 +1318,11 @@ impl DedupService { } total_deleted += batch.len() as u64; - // Yield so uploads / other tasks are not starved. tokio::task::yield_now().await; } if total_deleted > 0 { - tracing::info!("GC: removed {total_deleted} blobs ({total_bytes} bytes)"); + tracing::info!("GC: removed {total_deleted} items ({total_bytes} bytes)"); } Ok((total_deleted, total_bytes)) @@ -691,3 +1403,349 @@ impl DedupPort for DedupService { self.verify_integrity().await } } + +// ─── Tests ─────────────────────────────────────────────────────────────────── + +#[cfg(test)] +mod tests { + use super::*; + use std::collections::HashSet; + use tempfile::NamedTempFile; + + /// Helper: write `data` to a temp file and return its path. + async fn write_temp_file(data: &[u8]) -> NamedTempFile { + let file = NamedTempFile::new().unwrap(); + tokio::fs::write(file.path(), data).await.unwrap(); + file + } + + // ── Determinism ────────────────────────────────────────────── + + #[tokio::test] + async fn test_cdc_deterministic_same_content() { + let data = vec![42u8; 512 * 1024]; // 512 KB of 0x2A + let f1 = write_temp_file(&data).await; + let f2 = write_temp_file(&data).await; + + let (hash1, chunks1) = DedupService::cdc_hash_and_chunk_file(f1.path()) + .await + .unwrap(); + let (hash2, chunks2) = DedupService::cdc_hash_and_chunk_file(f2.path()) + .await + .unwrap(); + + assert_eq!(hash1, hash2, "same content must produce same file hash"); + assert_eq!( + chunks1.len(), + chunks2.len(), + "same content must produce same chunk count" + ); + for (c1, c2) in chunks1.iter().zip(chunks2.iter()) { + assert_eq!(c1.hash, c2.hash); + assert_eq!(c1.offset, c2.offset); + assert_eq!(c1.length, c2.length); + } + } + + // ── Empty file ─────────────────────────────────────────────── + + #[tokio::test] + async fn test_cdc_empty_file() { + let f = write_temp_file(b"").await; + let (hash, chunks) = DedupService::cdc_hash_and_chunk_file(f.path()) + .await + .unwrap(); + + assert!(chunks.is_empty(), "empty file must produce zero chunks"); + assert_eq!(hash, blake3::hash(b"").to_hex().to_string()); + } + + // ── Small file (below min chunk) → single chunk ────────────── + + #[tokio::test] + async fn test_cdc_small_file_single_chunk() { + let data = b"Hello, OxiCloud CDC dedup!"; + let f = write_temp_file(data).await; + let (hash, chunks) = DedupService::cdc_hash_and_chunk_file(f.path()) + .await + .unwrap(); + + assert_eq!(chunks.len(), 1, "tiny file must be a single chunk"); + assert_eq!(chunks[0].offset, 0); + assert_eq!(chunks[0].length, data.len()); + assert_eq!(hash, blake3::hash(data).to_hex().to_string()); + } + + // ── Chunk sizes within CDC bounds ──────────────────────────── + + #[tokio::test] + async fn test_cdc_chunk_sizes_within_bounds() { + // 4 MB file of pseudo-random data (deterministic seed) + let data: Vec = (0..4 * 1024 * 1024) + .map(|i| ((i as u64).wrapping_mul(6364136223846793005).wrapping_add(1)) as u8) + .collect(); + let f = write_temp_file(&data).await; + + let (_, chunks) = DedupService::cdc_hash_and_chunk_file(f.path()) + .await + .unwrap(); + + assert!(chunks.len() > 1, "4 MB should produce multiple chunks"); + + // All non-last chunks must be within [min, max] + for (i, chunk) in chunks.iter().enumerate() { + let is_last = i == chunks.len() - 1; + if !is_last { + assert!( + chunk.length >= CDC_MIN_CHUNK as usize, + "non-last chunk {} too small: {} < {}", + i, + chunk.length, + CDC_MIN_CHUNK, + ); + } + assert!( + chunk.length <= CDC_MAX_CHUNK as usize, + "chunk {} too large: {} > {}", + i, + chunk.length, + CDC_MAX_CHUNK, + ); + } + } + + // ── File hash matches hash_file() ──────────────────────────── + + #[tokio::test] + async fn test_cdc_file_hash_matches_hash_file() { + let data: Vec = (0..1024 * 1024).map(|i| (i % 251) as u8).collect(); + let f = write_temp_file(&data).await; + + let (cdc_hash, _) = DedupService::cdc_hash_and_chunk_file(f.path()) + .await + .unwrap(); + let standalone_hash = DedupService::hash_file(f.path()).await.unwrap(); + + assert_eq!( + cdc_hash, standalone_hash, + "CDC file hash must match standalone hash_file()" + ); + } + + // ── Chunk hashes are correct BLAKE3 of chunk data ──────────── + + #[tokio::test] + async fn test_cdc_chunk_hashes_are_correct() { + let data: Vec = (0..2 * 1024 * 1024) + .map(|i| ((i as u64).wrapping_mul(2862933555777941757).wrapping_add(3)) as u8) + .collect(); + let f = write_temp_file(&data).await; + + let (_, chunks) = DedupService::cdc_hash_and_chunk_file(f.path()) + .await + .unwrap(); + + for chunk in &chunks { + let chunk_data = &data[chunk.offset..chunk.offset + chunk.length]; + let expected_hash = blake3::hash(chunk_data).to_hex().to_string(); + assert_eq!( + chunk.hash, expected_hash, + "chunk at offset {} has wrong hash", + chunk.offset + ); + } + } + + // ── Reassembly matches original ────────────────────────────── + + #[tokio::test] + async fn test_cdc_reassembly_matches_original() { + let data: Vec = (0..3 * 1024 * 1024) + .map(|i| ((i as u64).wrapping_mul(1103515245).wrapping_add(12345)) as u8) + .collect(); + let f = write_temp_file(&data).await; + + let (_, chunks) = DedupService::cdc_hash_and_chunk_file(f.path()) + .await + .unwrap(); + + // Reassemble from chunks + let mut reassembled = Vec::with_capacity(data.len()); + for chunk in &chunks { + reassembled.extend_from_slice(&data[chunk.offset..chunk.offset + chunk.length]); + } + + assert_eq!( + reassembled.len(), + data.len(), + "reassembled length must match" + ); + assert_eq!(reassembled, data, "reassembled content must match original"); + } + + // ── Chunks cover entire file (no gaps, no overlaps) ────────── + + #[tokio::test] + async fn test_cdc_chunks_are_contiguous() { + let data: Vec = (0..2 * 1024 * 1024).map(|i| (i % 199) as u8).collect(); + let f = write_temp_file(&data).await; + + let (_, chunks) = DedupService::cdc_hash_and_chunk_file(f.path()) + .await + .unwrap(); + + let mut expected_offset = 0usize; + for (i, chunk) in chunks.iter().enumerate() { + assert_eq!( + chunk.offset, expected_offset, + "chunk {} starts at {} but expected {}", + i, chunk.offset, expected_offset + ); + expected_offset += chunk.length; + } + assert_eq!(expected_offset, data.len(), "chunks must cover entire file"); + } + + // ── Sub-file dedup: similar files share chunks ─────────────── + + #[tokio::test] + async fn test_cdc_similar_files_share_chunks() { + // Create a base file of 2 MB with random-ish data + let base: Vec = (0..2 * 1024 * 1024) + .map(|i| ((i as u64).wrapping_mul(6364136223846793005).wrapping_add(1)) as u8) + .collect(); + + // Modified file: change only the last 64 KB + let mut modified = base.clone(); + let start = modified.len() - 64 * 1024; + for b in &mut modified[start..] { + *b = b.wrapping_add(1); + } + + let f_base = write_temp_file(&base).await; + let f_mod = write_temp_file(&modified).await; + + let (hash_base, chunks_base) = DedupService::cdc_hash_and_chunk_file(f_base.path()) + .await + .unwrap(); + let (hash_mod, chunks_mod) = DedupService::cdc_hash_and_chunk_file(f_mod.path()) + .await + .unwrap(); + + // File hashes must differ + assert_ne!( + hash_base, hash_mod, + "modified file must have different hash" + ); + + // Collect chunk hashes + let base_set: HashSet<&str> = chunks_base.iter().map(|c| c.hash.as_str()).collect(); + let mod_set: HashSet<&str> = chunks_mod.iter().map(|c| c.hash.as_str()).collect(); + + let shared = base_set.intersection(&mod_set).count(); + + // With only the last 64 KB changed, most chunks should be shared. + // The first ~1.9 MB of content is identical → expect significant overlap. + let min_expected_shared = chunks_base.len().min(chunks_mod.len()) / 2; + assert!( + shared >= min_expected_shared, + "expected at least {} shared chunks between similar files, got {} \ + (base: {} chunks, modified: {} chunks)", + min_expected_shared, + shared, + chunks_base.len(), + chunks_mod.len() + ); + } + + // ── cdc_chunk_file matches cdc_hash_and_chunk_file ─────────── + + #[tokio::test] + async fn test_cdc_chunk_file_matches_full() { + let data: Vec = (0..1024 * 1024) + .map(|i| (i as u8).wrapping_mul(7)) + .collect(); + let f = write_temp_file(&data).await; + + let (_, chunks_full) = DedupService::cdc_hash_and_chunk_file(f.path()) + .await + .unwrap(); + let chunks_only = DedupService::cdc_chunk_file(f.path()).await.unwrap(); + + assert_eq!(chunks_full.len(), chunks_only.len()); + for (a, b) in chunks_full.iter().zip(chunks_only.iter()) { + assert_eq!(a.hash, b.hash); + assert_eq!(a.offset, b.offset); + assert_eq!(a.length, b.length); + } + } + + // ── Large file produces expected chunk count ────────────────── + + #[tokio::test] + async fn test_cdc_large_file_chunk_count() { + // 8 MB should produce roughly 8MB / 256KB ≈ 32 chunks (±) + let data: Vec = (0..8 * 1024 * 1024) + .map(|i| ((i as u64).wrapping_mul(2862933555777941757).wrapping_add(3)) as u8) + .collect(); + let f = write_temp_file(&data).await; + + let (_, chunks) = DedupService::cdc_hash_and_chunk_file(f.path()) + .await + .unwrap(); + + // With 256KB avg, expect 20-60 chunks for 8MB + assert!( + chunks.len() >= 8 && chunks.len() <= 128, + "8 MB file should produce 8-128 chunks (avg 256KB), got {}", + chunks.len() + ); + + let total_size: usize = chunks.iter().map(|c| c.length).sum(); + assert_eq!( + total_size, + data.len(), + "total chunk sizes must equal file size" + ); + } + + // ── Prefix insert: CDC shifts only locally ─────────────────── + + #[tokio::test] + async fn test_cdc_insert_at_beginning_preserves_later_chunks() { + // Base file: 2 MB of deterministic data + let base: Vec = (0..2 * 1024 * 1024) + .map(|i| ((i as u64).wrapping_mul(6364136223846793005).wrapping_add(1)) as u8) + .collect(); + + // Insert 128 KB at the beginning (simulates a header change) + let prefix: Vec = (0..128 * 1024).map(|i| (i % 173) as u8).collect(); + let mut with_prefix = prefix; + with_prefix.extend_from_slice(&base); + + let f_base = write_temp_file(&base).await; + let f_prefix = write_temp_file(&with_prefix).await; + + let (_, chunks_base) = DedupService::cdc_hash_and_chunk_file(f_base.path()) + .await + .unwrap(); + let (_, chunks_prefix) = DedupService::cdc_hash_and_chunk_file(f_prefix.path()) + .await + .unwrap(); + + let base_set: HashSet<&str> = chunks_base.iter().map(|c| c.hash.as_str()).collect(); + let prefix_set: HashSet<&str> = chunks_prefix.iter().map(|c| c.hash.as_str()).collect(); + + // CDC's content-defined boundaries mean chunks after the insertion + // should resynchronize — we expect *some* shared chunks, proving + // CDC is better than fixed-size chunking (which would share zero). + let shared = base_set.intersection(&prefix_set).count(); + assert!( + shared > 0, + "CDC should resynchronize and share chunks after insertion \ + (base: {} chunks, with-prefix: {} chunks, shared: 0)", + chunks_base.len(), + chunks_prefix.len() + ); + } +} diff --git a/src/infrastructure/services/encrypted_blob_backend.rs b/src/infrastructure/services/encrypted_blob_backend.rs index 4b0bdcab..47a3997a 100644 --- a/src/infrastructure/services/encrypted_blob_backend.rs +++ b/src/infrastructure/services/encrypted_blob_backend.rs @@ -104,6 +104,31 @@ impl BlobStorageBackend for EncryptedBlobBackend { }) } + fn put_blob_from_bytes( + &self, + hash: &str, + data: Bytes, + ) -> Pin> + Send + '_>> { + let inner = self.inner.clone(); + let hash = hash.to_string(); + let cipher = self.cipher.clone(); + Box::pin(async move { + // Encrypt in memory: nonce || ciphertext (includes GCM tag) + let nonce = Aes256Gcm::generate_nonce(&mut OsRng); + let ciphertext = cipher.encrypt(&nonce, data.as_ref()).map_err(|e| { + DomainError::internal_error("Encryption", format!("encrypt failed: {e}")) + })?; + + let mut encrypted = Vec::with_capacity(NONCE_SIZE + ciphertext.len()); + encrypted.extend_from_slice(nonce.as_slice()); + encrypted.extend_from_slice(&ciphertext); + + inner + .put_blob_from_bytes(&hash, Bytes::from(encrypted)) + .await + }) + } + fn get_blob_stream( &self, hash: &str, diff --git a/src/infrastructure/services/local_blob_backend.rs b/src/infrastructure/services/local_blob_backend.rs index 86c7e536..1e896849 100644 --- a/src/infrastructure/services/local_blob_backend.rs +++ b/src/infrastructure/services/local_blob_backend.rs @@ -9,6 +9,8 @@ use tokio::fs::{self, File}; use tokio::io::AsyncSeekExt; use tokio_util::io::ReaderStream; +use bytes::Bytes; + use crate::application::ports::blob_storage_ports::{ BlobStorageBackend, BlobStream, StorageHealthStatus, }; @@ -146,6 +148,33 @@ impl BlobStorageBackend for LocalBlobBackend { }) } + fn put_blob_from_bytes( + &self, + hash: &str, + data: Bytes, + ) -> Pin> + Send + '_>> { + let hash = hash.to_owned(); + Box::pin(async move { + let blob_path = self.blob_path(&hash); + let size = data.len() as u64; + + // Idempotent: if blob already exists, skip + if fs::try_exists(&blob_path).await.unwrap_or(false) { + return Ok(size); + } + + // Write directly to blob path + fs::write(&blob_path, &data).await.map_err(|e| { + DomainError::internal_error( + "Blob", + format!("Failed to write blob from bytes: {}", e), + ) + })?; + + Ok(size) + }) + } + fn get_blob_stream( &self, hash: &str, diff --git a/src/infrastructure/services/migration_blob_backend.rs b/src/infrastructure/services/migration_blob_backend.rs index 1f302fe5..40858fb8 100644 --- a/src/infrastructure/services/migration_blob_backend.rs +++ b/src/infrastructure/services/migration_blob_backend.rs @@ -10,6 +10,7 @@ use std::path::{Path, PathBuf}; use std::pin::Pin; use std::sync::Arc; +use bytes::Bytes; use chrono::{DateTime, Utc}; use serde::Serialize; use tokio::sync::RwLock; @@ -114,6 +115,12 @@ impl BlobStorageBackend for MigrationBlobBackend { Box::pin(async move { self.target.put_blob(&hash, &path).await }) } + /// Writes bytes to **target** only. + fn put_blob_from_bytes(&self, hash: &str, data: Bytes) -> BoxFut<'_, Result> { + let hash = hash.to_string(); + Box::pin(async move { self.target.put_blob_from_bytes(&hash, data).await }) + } + /// Read from target first; fall back to source. fn get_blob_stream(&self, hash: &str) -> BoxFut<'_, Result> { let hash = hash.to_string(); diff --git a/src/infrastructure/services/retry_blob_backend.rs b/src/infrastructure/services/retry_blob_backend.rs index 8076d5cf..6467081a 100644 --- a/src/infrastructure/services/retry_blob_backend.rs +++ b/src/infrastructure/services/retry_blob_backend.rs @@ -14,6 +14,7 @@ use crate::application::ports::blob_storage_ports::{ BlobStorageBackend, BlobStream, StorageHealthStatus, }; use crate::domain::errors::DomainError; +use bytes::Bytes; // ── Retry policy ─────────────────────────────────────────────────── @@ -139,6 +140,25 @@ impl BlobStorageBackend for RetryBlobBackend { }) } + fn put_blob_from_bytes( + &self, + hash: &str, + data: Bytes, + ) -> Pin> + Send + '_>> { + let inner = self.inner.clone(); + let policy = self.policy.clone(); + let hash = hash.to_string(); + Box::pin(async move { + retry_async(&policy, &format!("put_blob_from_bytes({hash})"), || { + let inner = inner.clone(); + let hash = hash.clone(); + let data = data.clone(); + async move { inner.put_blob_from_bytes(&hash, data).await } + }) + .await + }) + } + fn get_blob_stream( &self, hash: &str, diff --git a/src/infrastructure/services/s3_blob_backend.rs b/src/infrastructure/services/s3_blob_backend.rs index 99285f1e..abf6630b 100644 --- a/src/infrastructure/services/s3_blob_backend.rs +++ b/src/infrastructure/services/s3_blob_backend.rs @@ -4,6 +4,7 @@ //! Wasabi, and any other service that implements the S3 API. use aws_sdk_s3::primitives::ByteStream; +use bytes::Bytes; use std::path::{Path, PathBuf}; use std::pin::Pin; use tokio::fs; @@ -157,6 +158,48 @@ impl BlobStorageBackend for S3BlobBackend { }) } + fn put_blob_from_bytes( + &self, + hash: &str, + data: Bytes, + ) -> Pin> + Send + '_>> { + let hash = hash.to_owned(); + Box::pin(async move { + let key = Self::object_key(&hash); + let size = data.len() as u64; + + // Idempotent: skip if already exists + if self + .client + .head_object() + .bucket(&self.bucket) + .key(&key) + .send() + .await + .is_ok() + { + return Ok(size); + } + + let body = ByteStream::from(data); + self.client + .put_object() + .bucket(&self.bucket) + .key(&key) + .body(body) + .send() + .await + .map_err(|e| { + DomainError::internal_error( + "S3", + format!("Failed to upload blob {}: {}", hash, e), + ) + })?; + + Ok(size) + }) + } + fn get_blob_stream( &self, hash: &str,