diff --git a/CHANGELOG.md b/CHANGELOG.md index 947e468..f4fca5d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,30 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [0.2.0] - 2026-06-03 + +### Added + +- `TieredStorage`, a `CacheStorage` that layers a fast hot tier over a durable cold + tier — for example an `InMemoryStorage` working set in front of a `FileSystemStorage` durable + set, though any two backends compose. Lookups check the hot tier first; a cold hit is promoted + into the hot tier as it is read, so a hot tier emptied by a restart repopulates from cold on + demand. Writes populate the hot tier as the body streams and are flushed to the cold tier by a + background task, so evicting an entry from the hot tier only drops a fast-path copy while the + entry stays served from cold. Because the flush runs in the background, construct it with the + two backends and the runtime the surrounding server or client already uses: + `TieredStorage::new(hot, cold, runtime)`. + +- `FileSystemStorage`, a disk-backed `CacheStorage` that persists cached responses under a + root directory so they survive process restarts. Bodies stream to and from disk rather than + being buffered. Each response is stored as a compact rkyv-encoded metadata sidecar plus a + raw body file; the metadata is optimized for fast loading rather than being human-readable. + Enable it with the `fs` feature and select an async runtime with one of `smol`, `tokio`, or + `async-std`. A byte cap (1 GiB by default) bounds total stored body size, evicting + least-recently-used entries and deleting their files once it is reached; override it with + `with_max_capacity_bytes` or remove it with `unbounded`. The cap is enforced across + restarts and trims a directory that grew beyond it under an earlier configuration. + ## [0.1.1] - 2026-05-26 ### Fixed diff --git a/Cargo.lock b/Cargo.lock index 60c4e87..059e1a5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -86,6 +86,17 @@ version = "1.0.102" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c" +[[package]] +name = "async-channel" +version = "1.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "81953c529336010edd6d8e358f886d9581267795c61b19475b71314bffa46d35" +dependencies = [ + "concurrent-queue", + "event-listener 2.5.3", + "futures-core", +] + [[package]] name = "async-channel" version = "2.5.0" @@ -98,6 +109,19 @@ dependencies = [ "pin-project-lite", ] +[[package]] +name = "async-compat" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a1ba85bc55464dcbf728b56d97e119d673f4cf9062be330a9a26f3acf504a590" +dependencies = [ + "futures-core", + "futures-io", + "once_cell", + "pin-project-lite", + "tokio", +] + [[package]] name = "async-dup" version = "1.2.4" @@ -122,13 +146,39 @@ dependencies = [ "slab", ] +[[package]] +name = "async-fs" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8034a681df4aed8b8edbd7fbe472401ecf009251c8b40556b304567052e294c5" +dependencies = [ + "async-lock", + "blocking", + "futures-lite 2.6.1", +] + +[[package]] +name = "async-global-executor" +version = "2.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "05b1b633a2115cd122d73b955eadd9916c18c8f510ec9cd1686404c60ad1c29c" +dependencies = [ + "async-channel 2.5.0", + "async-executor", + "async-io", + "async-lock", + "blocking", + "futures-lite 2.6.1", + "once_cell", +] + [[package]] name = "async-global-executor" version = "3.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "13f937e26114b93193065fd44f507aa2e9169ad0cdabbb996920b1fe1ddea7ba" dependencies = [ - "async-channel", + "async-channel 2.5.0", "async-executor", "async-io", "async-lock", @@ -160,7 +210,7 @@ version = "3.4.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "290f7f2596bd5b78a9fec8088ccd89180d7f9f55b94b0576823bbbdc72ee8311" dependencies = [ - "event-listener", + "event-listener 5.4.1", "event-listener-strategy", "pin-project-lite", ] @@ -194,6 +244,32 @@ dependencies = [ "windows-sys", ] +[[package]] +name = "async-std" +version = "1.13.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2c8e079a4ab67ae52b7403632e4618815d6db36d2a010cfe41b02c1b1578f93b" +dependencies = [ + "async-channel 1.9.0", + "async-global-executor 2.4.1", + "async-io", + "async-lock", + "crossbeam-utils", + "futures-channel", + "futures-core", + "futures-io", + "futures-lite 2.6.1", + "gloo-timers", + "kv-log-macro", + "log", + "memchr", + "once_cell", + "pin-project-lite", + "pin-utils", + "slab", + "wasm-bindgen-futures", +] + [[package]] name = "async-task" version = "4.7.1" @@ -227,13 +303,22 @@ version = "2.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c4512299f36f043ab09a583e57bceb5a5aab7a73db1805848e8fef3c9e8c78b3" +[[package]] +name = "block-buffer" +version = "0.10.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71" +dependencies = [ + "generic-array", +] + [[package]] name = "blocking" version = "1.6.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e83f8d02be6967315521be875afa792a316e28d57b5a2d401897e2a7921b7f21" dependencies = [ - "async-channel", + "async-channel 2.5.0", "async-task", "futures-io", "futures-lite 2.6.1", @@ -246,6 +331,29 @@ version = "3.20.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649" +[[package]] +name = "bytecheck" +version = "0.8.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0caa33a2c0edca0419d15ac723dff03f1956f7978329b1e3b5fdaaaed9d3ca8b" +dependencies = [ + "bytecheck_derive", + "ptr_meta", + "rancor", + "simdutf8", +] + +[[package]] +name = "bytecheck_derive" +version = "0.8.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "89385e82b5d1821d2219e0b095efa2cc1f246cbf99080f3be46a1a85c0d392d9" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "bytes" version = "1.11.1" @@ -289,6 +397,15 @@ dependencies = [ "crossbeam-utils", ] +[[package]] +name = "cpufeatures" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "59ed5838eebb26a2bb2e58f6d5b5316989ae9d08bab10e0e6d103e656d1b0280" +dependencies = [ + "libc", +] + [[package]] name = "crossbeam-channel" version = "0.5.15" @@ -322,6 +439,16 @@ version = "0.8.21" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d0a5c400df2834b80a4c3327b3aad3a4c4cd4de0629063962b03235697506a28" +[[package]] +name = "crypto-common" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a" +dependencies = [ + "generic-array", + "typenum", +] + [[package]] name = "dashmap" version = "6.2.1" @@ -336,6 +463,16 @@ dependencies = [ "parking_lot_core", ] +[[package]] +name = "digest" +version = "0.10.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" +dependencies = [ + "block-buffer", + "crypto-common", +] + [[package]] name = "displaydoc" version = "0.2.5" @@ -405,6 +542,12 @@ dependencies = [ "xxhash-rust", ] +[[package]] +name = "event-listener" +version = "2.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0206175f82b8d6bf6652ff7d71a1e27fd2e4efde587fd368662814d6ec1d9ce0" + [[package]] name = "event-listener" version = "5.4.1" @@ -422,7 +565,7 @@ version = "0.5.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8be9f3dfaaffdae2972880079a491a1a8bb7cbed0b8dd7a347f668b4150a3b93" dependencies = [ - "event-listener", + "event-listener 5.4.1", "pin-project-lite", ] @@ -511,6 +654,15 @@ dependencies = [ "pin-project-lite", ] +[[package]] +name = "futures-channel" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "07bbe89c50d7a535e539b8c17bc0b49bdb77747034daa8087407d655f3f7cc1d" +dependencies = [ + "futures-core", +] + [[package]] name = "futures-core" version = "0.3.32" @@ -596,6 +748,16 @@ dependencies = [ "windows-result", ] +[[package]] +name = "generic-array" +version = "0.14.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a" +dependencies = [ + "typenum", + "version_check", +] + [[package]] name = "getrandom" version = "0.3.4" @@ -621,6 +783,18 @@ dependencies = [ "wasip3", ] +[[package]] +name = "gloo-timers" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bbb143cf96099802033e0d4f4963b19fd2e0b728bcf076cd9cf7f6634f092994" +dependencies = [ + "futures-channel", + "futures-core", + "js-sys", + "wasm-bindgen", +] + [[package]] name = "hashbrown" version = "0.14.5" @@ -839,16 +1013,24 @@ dependencies = [ [[package]] name = "js-sys" -version = "0.3.99" +version = "0.3.103" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "142bc4740e452c1e57ade0cbc129f139c9093e354346f0872ef985f4f5cf5f11" +checksum = "53b44bfcdb3f8d5837a46dae1ca9660a837176eee74a28b229bc626816589102" dependencies = [ "cfg-if", "futures-util", - "once_cell", "wasm-bindgen", ] +[[package]] +name = "kv-log-macro" +version = "1.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0de8b303297635ad57c9f5059fd9cee7a47f8e8daa09df0fcd07dd39fb22977f" +dependencies = [ + "log", +] + [[package]] name = "lazy_static" version = "1.5.0" @@ -904,6 +1086,9 @@ name = "log" version = "0.4.30" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "616ec5685824bcc94416c6d4a7a446eea774a31efd7062c8480ba6fd06d7a6e5" +dependencies = [ + "value-bag", +] [[package]] name = "loom" @@ -939,6 +1124,17 @@ version = "0.3.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" +[[package]] +name = "mio" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "02bd0af71c67b473010cbbc60715ee815645a4dc942899111f494b4b737d6fda" +dependencies = [ + "libc", + "wasi", + "windows-sys", +] + [[package]] name = "moka" version = "0.12.15" @@ -950,7 +1146,7 @@ dependencies = [ "crossbeam-epoch", "crossbeam-utils", "equivalent", - "event-listener", + "event-listener 5.4.1", "futures-util", "parking_lot", "portable-atomic", @@ -1041,6 +1237,12 @@ version = "0.2.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" +[[package]] +name = "pin-utils" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184" + [[package]] name = "piper" version = "0.2.5" @@ -1222,13 +1424,17 @@ name = "rend" version = "0.5.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cadadef317c2f20755a64d7fdc48f9e7178ee6b0e1f7fce33fa60f1d68a276e6" +dependencies = [ + "bytecheck", +] [[package]] name = "rkyv" -version = "0.8.16" +version = "0.8.17" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "73389e0c99e664f919275ab5b5b0471391fe9a8de61e1dff9b1eaf56a90f16e3" +checksum = "815cc8a37159a463064825246cadb07961e25cd9885908606f6d08a98d8f8874" dependencies = [ + "bytecheck", "bytes", "hashbrown 0.17.1", "indexmap", @@ -1243,9 +1449,9 @@ dependencies = [ [[package]] name = "rkyv_derive" -version = "0.8.16" +version = "0.8.17" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5d2ed0b54125315fb36bd021e82d314d1c126548f871634b483f46b31d13cac6" +checksum = "c0ed1a78a1b19d184b0daa629dd9a024573173ec7d485b287cb369fb3607cc1c" dependencies = [ "proc-macro2", "quote", @@ -1341,6 +1547,17 @@ dependencies = [ "zmij", ] +[[package]] +name = "sha2" +version = "0.10.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283" +dependencies = [ + "cfg-if", + "cpufeatures", + "digest", +] + [[package]] name = "sharded-slab" version = "0.1.7" @@ -1376,6 +1593,18 @@ dependencies = [ "libc", ] +[[package]] +name = "signal-hook-tokio" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e513e435a8898a0002270f29d0a708b7879708fb5c4d00e46983ca2d2d378cf0" +dependencies = [ + "futures-core", + "libc", + "signal-hook", + "tokio", +] + [[package]] name = "simdutf8" version = "0.1.5" @@ -1400,7 +1629,7 @@ version = "0.6.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "160b744a45e8261307bcfe03c98e2f8274502207d534c9a64b675c4db1b6bd58" dependencies = [ - "async-channel", + "async-channel 2.5.0", "futures-core", "futures-io", ] @@ -1431,6 +1660,16 @@ dependencies = [ "version_check", ] +[[package]] +name = "socket2" +version = "0.6.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "52d1cfed4120b4d927bf7c0f86d2087a4a7d6027c906d9f9d525a80573b9be51" +dependencies = [ + "libc", + "windows-sys", +] + [[package]] name = "sonic-number" version = "0.1.2" @@ -1494,7 +1733,7 @@ version = "0.3.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "59e547d6edfcd09568f663d6340c48478cec8cac1403db95e8072c9a7e513253" dependencies = [ - "event-listener", + "event-listener 5.4.1", "futures-core", "log", "pin-project-lite", @@ -1534,6 +1773,19 @@ version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7b2093cf4c8eb1e67749a6762251bc9cd836b6fc171623bd0a9d324d37af2417" +[[package]] +name = "tempfile" +version = "3.27.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" +dependencies = [ + "fastrand 2.4.1", + "getrandom 0.4.2", + "once_cell", + "rustix", + "windows-sys", +] + [[package]] name = "test-harness" version = "0.3.1" @@ -1599,6 +1851,31 @@ version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20" +[[package]] +name = "tokio" +version = "1.52.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8fc7f01b389ac15039e4dc9531aa973a135d7a4135281b12d7c1bc79fd57fffe" +dependencies = [ + "bytes", + "libc", + "mio", + "pin-project-lite", + "socket2", + "windows-sys", +] + +[[package]] +name = "tokio-stream" +version = "0.1.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32da49809aab5c3bc678af03902d4ccddea2a87d028d86392a4b1560c6906c70" +dependencies = [ + "futures-core", + "pin-project-lite", + "tokio", +] + [[package]] name = "tracing" version = "0.1.44" @@ -1662,20 +1939,29 @@ dependencies = [ [[package]] name = "trillium-cache" -version = "0.1.1" +version = "0.2.0" dependencies = [ + "async-compat", + "async-fs", + "async-std", + "cfg-if", "env_logger", "fieldwork", "futures-lite 2.6.1", "httpdate", "log", "moka", + "rkyv", + "sha2", + "tempfile", + "tokio", "trillium", "trillium-cache", "trillium-caching-headers", "trillium-client", "trillium-http", "trillium-proxy", + "trillium-server-common", "trillium-smol", "trillium-testing", "url", @@ -1726,23 +2012,23 @@ dependencies = [ [[package]] name = "trillium-http" -version = "1.3.3" +version = "1.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b326aef316f0d8944ef28d844553f963e88e9ef8c027f23ffa0c0890fbf50e2b" +checksum = "9d518b6617d4a5b40b32b226f28f5f78bbdb8349a6fbc059bf0ec58fd40792cb" dependencies = [ "atomic-waker", "encoding_rs", - "event-listener", + "event-listener 5.4.1", "fastrand 2.4.1", "fieldwork", "futures-lite 2.6.1", "hashbrown 0.17.1", - "httparse", "httpdate", "log", "memchr", "mime", "pin-project-lite", + "rkyv", "smallvec", "smartcow", "smartstring", @@ -1770,7 +2056,7 @@ version = "0.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4aa40533d6ac63a031d80b368525a34fa060617041b872cf45883fb7fb85a4cb" dependencies = [ - "event-listener", + "event-listener 5.4.1", "fastrand 2.4.1", "full-duplex-async-copy", "futures-lite 2.6.1", @@ -1791,7 +2077,7 @@ version = "0.7.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "dd3dd35f2a09ec85c5367962bf37fe697ebebf03d8453adce89fbe759d47cc62" dependencies = [ - "async-channel", + "async-channel 2.5.0", "async_cell", "fieldwork", "futures-lite 2.6.1", @@ -1811,7 +2097,7 @@ version = "0.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "351215c76e52a526d43d8b147bb0a494fb1be13570014a41e828365ed05e0886" dependencies = [ - "async-global-executor", + "async-global-executor 3.1.0", "async-io", "async-net", "async-signal", @@ -1831,7 +2117,7 @@ version = "0.10.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "50a2fbe24a29098c729ed1422038089a157f19598532ade563f829773d5b1a4c" dependencies = [ - "async-channel", + "async-channel 2.5.0", "async-dup", "cfg-if", "dashmap", @@ -1849,14 +2135,39 @@ dependencies = [ "trillium-http", "trillium-macros", "trillium-server-common", + "trillium-tokio", "url", ] +[[package]] +name = "trillium-tokio" +version = "0.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1b0d2e388eae1c568fb0ec5fe8d44253cbd338aa56139da9bb5663b22de01e2d" +dependencies = [ + "async-compat", + "log", + "signal-hook", + "signal-hook-tokio", + "tokio", + "tokio-stream", + "trillium", + "trillium-http", + "trillium-macros", + "trillium-server-common", +] + [[package]] name = "type-set" -version = "0.3.2" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a6d0bbc2be93a6be556bf72da67cab01b87e650f9dbd8a095e8b1a3d4712dfb" + +[[package]] +name = "typenum" +version = "1.20.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "845e18586b78ef401c37d9d2d2cca631c19816d5aa26af12647e828739bd6614" +checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" [[package]] name = "unicode-ident" @@ -1911,6 +2222,12 @@ version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" +[[package]] +name = "value-bag" +version = "1.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7ba6f5989077681266825251a52748b8c1d8a4ad098cc37e440103d0ea717fc0" + [[package]] name = "version_check" version = "0.9.5" @@ -1923,6 +2240,12 @@ version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "317211a0dc0ceedd78fb2ca9a44aed3d7b9b26f81870d485c07122b4350673b7" +[[package]] +name = "wasi" +version = "0.11.1+wasi-snapshot-preview1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" + [[package]] name = "wasip2" version = "1.0.3+wasi-0.2.9" @@ -1943,9 +2266,9 @@ dependencies = [ [[package]] name = "wasm-bindgen" -version = "0.2.122" +version = "0.2.126" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3ed04576f974d2b2fba0f38c51dbc5518011e38c36bf1143164be765528fd409" +checksum = "4b067c0c11094aef6b7a801c1e34a26affafdf3d051dba08456b868789aaf9a4" dependencies = [ "cfg-if", "once_cell", @@ -1954,11 +2277,21 @@ dependencies = [ "wasm-bindgen-shared", ] +[[package]] +name = "wasm-bindgen-futures" +version = "0.4.76" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c62df1340f32221cb9c54d6a27b030e3dba64361d4a95bed55f9aacb44da291d" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + [[package]] name = "wasm-bindgen-macro" -version = "0.2.122" +version = "0.2.126" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "916151b09da36bd82f6615cbf3a419e2f0ba23a03c6160e8e92eb6bd4aa1dec6" +checksum = "167ce5e579f6bcf889c4f7175a8a5a585de84e8ff93976ce393efa5f2837aab1" dependencies = [ "quote", "wasm-bindgen-macro-support", @@ -1966,9 +2299,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-macro-support" -version = "0.2.122" +version = "0.2.126" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "299047362ccbfce148b67ab7e73349f77748e00c8296f9542adfad2ad82c5c5e" +checksum = "f3997c7839262f4ef12cf90b818d6340c18e80f263f1a94bf157d0ec4420380e" dependencies = [ "bumpalo", "proc-macro2", @@ -1979,9 +2312,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-shared" -version = "0.2.122" +version = "0.2.126" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9a929b2c61f11ba3e9bc35b50c1f25cb38e0e892c0c231ae2b8cf78d5dad4437" +checksum = "dc1b4cb0cc549fcf58d7dfc081778139b3d283a081644e833e84682ad71cea24" dependencies = [ "unicode-ident", ] diff --git a/Cargo.toml b/Cargo.toml index bd4f28c..93e179d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "trillium-cache" -version = "0.1.1" +version = "0.2.0" edition = "2024" description = "http cache handler for trillium.rs" license = "MIT OR Apache-2.0" @@ -13,24 +13,41 @@ categories = ["caching", "web-programming::http-client"] development = ["trillium-cache"] [package.metadata.docs.rs] -features = ["client"] +features = ["client", "fs", "smol"] [features] client = ["dep:trillium-client"] +fs = ["dep:rkyv", "dep:sha2", "dep:cfg-if", "trillium-http/rkyv_08"] +# `tokio::fs` is reactor-bound, so the test harness must run on tokio to exercise the fs backend +# under this feature; forward it to trillium-testing. smol and async-std need no ambient runtime +# (their fs adapters run under trillium-testing's default runtimeless harness), so they don't +# forward — and forwarding more than one runtime would break `--all-features`. +tokio = ["fs", "dep:tokio", "dep:async-compat", "trillium-testing/tokio"] +async-std = ["fs", "dep:async-std"] +smol = ["fs", "dep:async-fs"] [dependencies] +async-compat = { version = "0.2.5", optional = true } +async-fs = { version = "2.2.0", optional = true } +async-std = { version = "1.13.2", optional = true } +cfg-if = { version = "1.0.4", optional = true } fieldwork = "0.5.2" httpdate = "1.0.3" log = "0.4.29" futures-lite = "2.6.1" -moka = { version = "0.12.15", features = ["future"] } +moka = { version = "0.12.15", features = ["future", "sync"] } +rkyv = { version = "0.8.17", optional = true } +sha2 = { version = "0.10.9", optional = true } +tokio = { version = "1.50.0", features = ["fs", "io-util"], optional = true } trillium = "1" trillium-caching-headers = "0.4" trillium-client = { version = "0.9", optional = true } -trillium-http = { version = "1", features = ["unstable"] } +trillium-http = { version = "1.4", features = ["unstable"] } +trillium-server-common = "0.7" url = "2.5.8" [dev-dependencies] +tempfile = "3" trillium-cache = { path = ".", features = ["client"] } trillium-smol = "0.6.1" trillium-testing = "0.10" @@ -45,3 +62,7 @@ required-features = ["client"] name = "conformance_proxy" required-features = ["client"] +[[example]] +name = "tiered_cache" +required-features = ["smol", "client"] + diff --git a/examples/tiered_cache.rs b/examples/tiered_cache.rs new file mode 100644 index 0000000..b8e9560 --- /dev/null +++ b/examples/tiered_cache.rs @@ -0,0 +1,115 @@ +//! Wiring a [`TieredStorage`] — an in-memory hot tier over a filesystem cold tier — into a +//! server [`Cache`], with a focus on where its background-flush runtime comes from. +//! +//! The server serves one cacheable route through the cache. The first request is a miss and +//! runs the wrapped handler; the second is served from the hot (in-memory) tier without +//! touching the handler, while the cold (filesystem) tier receives a durable copy from the +//! background write-back. +//! +//! Run with trace logging to watch the cache decisions: +//! +//! ```text +//! RUST_LOG=trillium_cache=trace cargo run --example tiered_cache --features smol,client +//! ``` +//! +//! [`TieredStorage`]: trillium_cache::TieredStorage +//! [`Cache`]: trillium_cache::Cache + +use std::{ + sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }, + time::Duration, +}; +use trillium::{Conn, KnownHeaderName}; +use trillium_cache::{ + Cache, CacheKey, CacheStorage, FileSystemStorage, InMemoryStorage, TieredStorage, +}; +use trillium_client::Client; +use trillium_http::Method; +use trillium_smol::{ClientConfig, SmolRuntime, async_global_executor, config}; + +fn main() { + env_logger::Builder::from_env( + env_logger::Env::default().default_filter_or("trillium_cache=trace"), + ) + .init(); + async_global_executor::block_on(run()); +} + +async fn run() { + // The cold tier lives on disk; use a fresh temp dir for the demo. + let cache_dir = std::env::temp_dir().join("trillium-cache-tiered-example"); + let _ = std::fs::remove_dir_all(&cache_dir); + + // TieredStorage flushes to its cold tier on a background task, so it needs the runtime that + // task runs on. On smol that is `SmolRuntime::default()` — a handle to the same global + // executor `trillium_smol` drives. On tokio it would be `TokioRuntime::default()` from + // inside your tokio context. On a *client* you would instead take it from the connector: + // `client.connector().runtime()` (see the `proxy_cache` example for the client-cache shape). + let storage = TieredStorage::new( + InMemoryStorage::new(), + FileSystemStorage::new(&cache_dir), + SmolRuntime::default(), + ); + + // Count how often the wrapped handler actually runs; a cache hit skips it. + let handler_runs = Arc::new(AtomicUsize::new(0)); + let runs = handler_runs.clone(); + + let server = config() + .with_port(0) + .with_host("127.0.0.1") + .without_signals() + .spawn((Cache::new(storage), move |conn: Conn| { + let runs = runs.clone(); + async move { + runs.fetch_add(1, Ordering::SeqCst); + conn.with_response_header(KnownHeaderName::CacheControl, "max-age=600") + .ok("hello from the cached handler") + } + })); + let addr = *server.info().await.tcp_socket_addr().unwrap(); + println!("server listening on {addr}\n"); + + let client = Client::new(ClientConfig::new()); + for label in ["request #1 (expect MISS)", "request #2 (expect HIT)"] { + let url = format!("http://{addr}/"); + let mut conn = client.get(url.as_str()).await.expect("request failed"); + let body = conn.response_body().read_string().await.expect("read body"); + println!("[{label}] {body:?}"); + } + + println!( + "\nhandler ran {} time(s) (expected 1 — the second request was served from the hot tier)", + handler_runs.load(Ordering::SeqCst) + ); + + // The cold-tier flush runs on a detached task, so poll a freshly opened FileSystemStorage + // over the same directory — the view a restarted process would get — until the write-back + // commits (or give up). This is the durability the hot tier alone can't offer. + let key = CacheKey::new(Method::Get, format!("http://{addr}/").parse().unwrap()); + let reopened_cold = FileSystemStorage::new(&cache_dir); + let mut committed = false; + for _ in 0..50 { + if !reopened_cold.get(&key).await.is_empty() { + committed = true; + break; + } + SmolRuntime::default() + .delay(Duration::from_millis(20)) + .await; + } + println!( + "cold tier under {} {} the entry after write-back", + cache_dir.display(), + if committed { + "durably holds" + } else { + "did not receive" + } + ); + + server.shut_down().await; +} diff --git a/src/client.rs b/src/client.rs index ff2b53a..859764a 100644 --- a/src/client.rs +++ b/src/client.rs @@ -643,13 +643,13 @@ mod tests { async fn run(&self, conn: ServerConn) -> ServerConn { let n = self.counter.fetch_add(1, Ordering::SeqCst); - if let Some(etag) = self.etag { - if conn.request_headers().get_str(KnownHeaderName::IfNoneMatch) == Some(etag) { - return conn - .with_status(Status::NotModified) - .with_response_header(KnownHeaderName::Etag, etag) - .halt(); - } + if let Some(etag) = self.etag + && conn.request_headers().get_str(KnownHeaderName::IfNoneMatch) == Some(etag) + { + return conn + .with_status(Status::NotModified) + .with_response_header(KnownHeaderName::Etag, etag) + .halt(); } let mut conn = conn diff --git a/src/fs.rs b/src/fs.rs new file mode 100644 index 0000000..a36cac0 --- /dev/null +++ b/src/fs.rs @@ -0,0 +1,873 @@ +//! Filesystem-backed [`CacheStorage`]. + +use crate::{ + CacheKey, CachePolicy, CacheStorage, PutHandle, StoredEntry, fs_shims, policy::PolicyRepr, +}; +use futures_lite::{AsyncRead, AsyncWrite, AsyncWriteExt}; +use moka::{notification::RemovalCause, sync::Cache}; +use sha2::{Digest, Sha256}; +use std::{ + fmt::{self, Debug, Formatter, Write as _}, + io, + path::{Path, PathBuf}, + pin::Pin, + sync::{ + Arc, + atomic::{AtomicU64, Ordering}, + }, + task::{Context, Poll}, +}; +use trillium_http::{Body, BodySource, Headers}; + +const META_SUFFIX: &str = ".meta"; +const BODY_SUFFIX: &str = ".body"; + +// Disk caches are cheap to grow relative to memory, so the default ceiling is larger than +// `InMemoryStorage`'s. +const DEFAULT_MAX_CAPACITY_BYTES: u64 = 1024 * 1024 * 1024; + +// Disambiguates concurrent temporary files under one directory. Process-local; on-disk +// temporaries from a previous run are never read (only committed files are). +static TEMP_COUNTER: AtomicU64 = AtomicU64::new(0); + +/// Filesystem-backed cache storage rooted at a directory. +/// +/// Persists cached responses under a root directory so they survive process restarts. Each +/// response is two files: a `.meta` sidecar holding the [`CachePolicy`] and any trailers +/// as an rkyv-encoded binary blob, and a `.body` holding the raw body bytes and nothing +/// else. Bodies stream in and out — [`put`] writes to a temporary file the caller feeds +/// incrementally, and [`open`] streams the stored body back without loading it into memory. The +/// metadata is not human-readable; it is optimized for compact, fast loading rather than +/// inspection. +/// +/// Defaults to a 1 GiB byte cap; override with +/// [`with_max_capacity_bytes`][Self::with_max_capacity_bytes] or remove it with +/// [`unbounded`][Self::unbounded]. +/// +/// `Clone` is cheap — clones share the same root and capacity index, and see each other's +/// writes. +/// +/// # Layout +/// +/// Entries live at `//.{meta,body}`. The key hash is a SHA-256 +/// of the request method and URL; the variant hash is a SHA-256 of the `Vary` signature, so +/// the multiple variants of one URL are sibling files in the same directory and [`get`] +/// enumerates them by reading that directory. Writing a variant that already exists replaces +/// it. +/// +/// # Durability +/// +/// Writes commit by renaming a fully-written temporary file into place, and the `.meta` is +/// written last — a reader treats it as the commit marker, so a half-written or abandoned entry +/// (a [`PutHandle`] dropped without [`finalize`]) is never visible to [`get`]. +/// +/// # Capacity +/// +/// A byte cap (1 GiB by default) bounds the total stored body size. When a write would push +/// the total past the cap, least-recently-used variants are evicted — their `.meta` and +/// `.body` files deleted — until the cache fits. The cap counts body bytes only, per variant, +/// matching the granularity of the on-disk layout. Reads count as use, so a frequently-served +/// variant outlives idle ones. Override with [`with_max_capacity_bytes`] or remove the cap with +/// [`unbounded`]. +/// +/// The cap is tracked in an in-memory index built by scanning the root at construction, so it +/// survives restarts (recency resets to whatever order the scan encounters). A directory that +/// grew past the current cap under an older, unbounded configuration is trimmed to fit on the +/// next construction. +/// +/// # Runtime +/// +/// Filesystem access goes through the runtime selected by the `smol`, `tokio`, or `async-std` +/// feature. Enabling `fs` without one of those compiles but panics on use. +/// +/// [`put`]: CacheStorage::put +/// [`get`]: CacheStorage::get +/// [`open`]: StoredEntry::open +/// [`finalize`]: PutHandle::finalize +/// [`with_max_capacity_bytes`]: FileSystemStorage::with_max_capacity_bytes +/// [`unbounded`]: FileSystemStorage::unbounded +#[derive(Clone)] +pub struct FileSystemStorage { + root: Arc, + index: Cache, + max_capacity_bytes: Option, +} + +impl Debug for FileSystemStorage { + fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { + f.debug_struct("FileSystemStorage") + .field("root", &self.root) + .field("weighted_size", &self.index.weighted_size()) + .field("max_capacity_bytes", &self.max_capacity_bytes) + .finish() + } +} + +impl FileSystemStorage { + /// Construct a storage rooted at `root` with a 1 GiB byte cap. The directory is created + /// on demand as entries are written; it need not exist yet. If it exists, it is scanned + /// to seed the capacity index, so previously stored entries count against the cap. + pub fn new(root: impl Into) -> Self { + let root = Arc::new(root.into()); + let max_capacity_bytes = Some(DEFAULT_MAX_CAPACITY_BYTES); + let index = build_index(Arc::clone(&root), max_capacity_bytes); + scan_root(&root, &index); + Self { + root, + index, + max_capacity_bytes, + } + } + + /// Set the maximum total stored body size, in bytes. Least-recently-used variants are + /// evicted — their files deleted — when a write would exceed this cap. Defaults to + /// 1 GiB. Re-scans the root, so a directory already over the new cap is trimmed to fit. + pub fn with_max_capacity_bytes(mut self, bytes: u64) -> Self { + self.max_capacity_bytes = Some(bytes); + self.rebuild(); + self + } + + /// Remove the size cap. Stored bytes grow without bound. Useful in tests and short-lived + /// processes; a cache living on shared disk should prefer the default capped + /// configuration. + pub fn unbounded(mut self) -> Self { + self.max_capacity_bytes = None; + self.rebuild(); + self + } + + /// Approximate total stored body size, in bytes, currently counted against the cap. + /// Eventually consistent — call [`run_pending_tasks`][Self::run_pending_tasks] first for + /// a settled value. + pub fn weighted_size(&self) -> u64 { + self.index.weighted_size() + } + + /// Approximate count of stored variants. Eventually consistent — call + /// [`run_pending_tasks`][Self::run_pending_tasks] first for a settled value. + pub fn entry_count(&self) -> u64 { + self.index.entry_count() + } + + /// Flush pending eviction bookkeeping, including deletion of files for evicted variants. + /// Call before reading [`weighted_size`][Self::weighted_size] or + /// [`entry_count`][Self::entry_count] when an exact value matters. + pub async fn run_pending_tasks(&self) { + self.index.run_pending_tasks(); + } + + // The capacity index has no resize API; rebuilding it and re-scanning the root applies a + // new cap while preserving on-disk entries (unlike the in-memory backend, disk data + // survives a reconfigure). + fn rebuild(&mut self) { + self.index = build_index(Arc::clone(&self.root), self.max_capacity_bytes); + scan_root(&self.root, &self.index); + } +} + +// Identity of one stored variant, sufficient to reconstruct its `.meta`/`.body` paths under +// a known root. Keys the capacity index. +#[derive(Clone, Hash, PartialEq, Eq)] +struct VariantId { + key_hash: String, + variant_hash: String, +} + +// Build the capacity index. The eviction listener deletes a variant's files when moka +// evicts it for size or expiry; replacement and explicit invalidation are handled at their +// call sites, so the listener ignores those causes. +fn build_index(root: Arc, max_capacity_bytes: Option) -> Cache { + let mut builder = Cache::::builder() + .weigher(|_key, &body_len| u32::try_from(body_len).unwrap_or(u32::MAX)) + .eviction_listener(move |id: Arc, _body_len, cause: RemovalCause| { + if cause.was_evicted() { + let dir = root.join(&id.key_hash); + let _ = std::fs::remove_file(dir.join(format!("{}{META_SUFFIX}", id.variant_hash))); + let _ = std::fs::remove_file(dir.join(format!("{}{BODY_SUFFIX}", id.variant_hash))); + } + }); + if let Some(cap) = max_capacity_bytes { + builder = builder.max_capacity(cap); + } + builder.build() +} + +// Seed the index from the root, counting each committed variant's body length against the +// cap. Runs on the calling thread with blocking IO — a one-time construction cost — and +// forces eviction so an over-cap directory is trimmed before the storage is used. +fn scan_root(root: &Path, index: &Cache) { + let Ok(key_dirs) = std::fs::read_dir(root) else { + return; + }; + for key_entry in key_dirs.flatten() { + let key_dir = key_entry.path(); + let Some(key_hash) = file_stem_string(&key_dir) else { + continue; + }; + let Ok(files) = std::fs::read_dir(&key_dir) else { + continue; + }; + for file in files.flatten() { + let path = file.path(); + let Some(variant_hash) = path + .file_name() + .and_then(|name| name.to_str()) + .and_then(|name| name.strip_suffix(META_SUFFIX)) + .map(str::to_string) + else { + continue; + }; + let body = key_dir.join(format!("{variant_hash}{BODY_SUFFIX}")); + let Ok(metadata) = std::fs::metadata(&body) else { + continue; + }; + index.insert( + VariantId { + key_hash: key_hash.clone(), + variant_hash, + }, + metadata.len(), + ); + } + } + index.run_pending_tasks(); +} + +fn file_stem_string(path: &Path) -> Option { + path.file_name() + .and_then(|name| name.to_str()) + .map(str::to_string) +} + +// The rkyv-encoded sidecar written alongside each body. `PolicyRepr` recomputes the derived +// cache-control fields on load, so only the directly-captured policy fields are stored. +#[derive(rkyv::Archive, rkyv::Serialize, rkyv::Deserialize)] +struct StoredMeta { + policy: PolicyRepr, + trailers: Option, +} + +impl CacheStorage for FileSystemStorage { + type PutHandle = FsPutHandle; + type StoredEntry = FsStoredEntry; + + async fn get(&self, key: &CacheKey) -> Vec { + let key_hash = key_hash(key); + let dir = self.root.join(&key_hash); + let Ok(paths) = fs_shims::read_dir_paths(&dir).await else { + return Vec::new(); + }; + + let mut entries = Vec::new(); + for path in paths { + let Some(variant_hash) = path + .file_name() + .and_then(|name| name.to_str()) + .and_then(|name| name.strip_suffix(META_SUFFIX)) + .map(str::to_string) + else { + continue; + }; + let Ok(bytes) = fs_shims::read(&path).await else { + continue; + }; + let Ok(meta) = deserialize_meta(&bytes) else { + continue; + }; + // Count the lookup as use so a frequently-served variant survives eviction. + self.index.get(&VariantId { + key_hash: key_hash.clone(), + variant_hash: variant_hash.clone(), + }); + entries.push(FsStoredEntry { + meta_path: path, + body_path: dir.join(format!("{variant_hash}{BODY_SUFFIX}")), + policy: meta.policy.into(), + trailers: meta.trailers, + }); + } + entries + } + + async fn put(&self, key: CacheKey, policy: CachePolicy) -> io::Result { + let key_hash = key_hash(&key); + let dir = self.root.join(&key_hash); + fs_shims::create_dir_all(&dir).await?; + + let variant_hash = variant_hash(&policy); + let n = TEMP_COUNTER.fetch_add(1, Ordering::Relaxed); + let body_tmp = dir.join(format!("{variant_hash}{BODY_SUFFIX}.tmp.{n}")); + let writer = fs_shims::create(&body_tmp).await?; + + Ok(FsPutHandle { + writer, + body_tmp, + body_final: dir.join(format!("{variant_hash}{BODY_SUFFIX}")), + meta_tmp: dir.join(format!("{variant_hash}{META_SUFFIX}.tmp.{n}")), + meta_final: dir.join(format!("{variant_hash}{META_SUFFIX}")), + policy, + index: self.index.clone(), + variant_id: VariantId { + key_hash, + variant_hash, + }, + written: 0, + committed: false, + }) + } + + async fn invalidate(&self, key: &CacheKey) { + let key_hash = key_hash(key); + let dir = self.root.join(&key_hash); + // Prune the index before removing files; the whole directory goes at once, so the + // per-variant eviction listener would be redundant (it skips explicit removals). + if let Ok(paths) = fs_shims::read_dir_paths(&dir).await { + for path in paths { + if let Some(variant_hash) = path + .file_name() + .and_then(|name| name.to_str()) + .and_then(|name| name.strip_suffix(META_SUFFIX)) + { + self.index.invalidate(&VariantId { + key_hash: key_hash.clone(), + variant_hash: variant_hash.to_string(), + }); + } + } + } + let _ = fs_shims::remove_dir_all(&dir).await; + } +} + +/// Streaming [`PutHandle`] for [`FileSystemStorage`]. +/// +/// Body bytes are written to a temporary file as they arrive; [`finalize`][Self::finalize] +/// renames the body into place and writes the metadata sidecar. Dropping without finalizing +/// removes the temporary body and stores nothing. +pub struct FsPutHandle { + writer: fs_shims::Writer, + body_tmp: PathBuf, + body_final: PathBuf, + meta_tmp: PathBuf, + meta_final: PathBuf, + policy: CachePolicy, + index: Cache, + variant_id: VariantId, + written: u64, + committed: bool, +} + +impl Debug for FsPutHandle { + fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { + f.debug_struct("FsPutHandle") + .field("body_final", &self.body_final) + .finish_non_exhaustive() + } +} + +impl AsyncWrite for FsPutHandle { + fn poll_write( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + let this = self.get_mut(); + let poll = Pin::new(&mut this.writer).poll_write(cx, buf); + if let Poll::Ready(Ok(n)) = &poll { + this.written += *n as u64; + } + poll + } + + fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.get_mut().writer).poll_flush(cx) + } + + fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.get_mut().writer).poll_close(cx) + } +} + +impl PutHandle for FsPutHandle { + async fn finalize(mut self, trailers: Option) -> io::Result<()> { + self.writer.close().await?; + fs_shims::rename(&self.body_tmp, &self.body_final).await?; + + let meta = StoredMeta { + policy: PolicyRepr::from(&self.policy), + trailers, + }; + let bytes = serialize_meta(&meta)?; + fs_shims::write(&self.meta_tmp, &bytes).await?; + fs_shims::rename(&self.meta_tmp, &self.meta_final).await?; + + // Account the committed body against the cap. Re-inserting the same variant replaces + // its prior weight; the eviction listener ignores the replacement. + self.index.insert(self.variant_id.clone(), self.written); + + self.committed = true; + Ok(()) + } +} + +impl Drop for FsPutHandle { + fn drop(&mut self) { + if !self.committed { + let _ = std::fs::remove_file(&self.body_tmp); + } + } +} + +/// One stored response returned by [`FileSystemStorage::get`]. +/// +/// Holds the metadata; the body stays on disk until [`open`][Self::open] streams it. `Clone` +/// copies the metadata and re-opens the body file on demand. +#[derive(Clone)] +pub struct FsStoredEntry { + meta_path: PathBuf, + body_path: PathBuf, + policy: CachePolicy, + trailers: Option, +} + +impl Debug for FsStoredEntry { + fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { + f.debug_struct("FsStoredEntry") + .field("body_path", &self.body_path) + .field("has_trailers", &self.trailers.is_some()) + .finish_non_exhaustive() + } +} + +impl StoredEntry for FsStoredEntry { + fn policy(&self) -> &CachePolicy { + &self.policy + } + + async fn refresh_policy(&mut self, new_policy: CachePolicy) -> io::Result<()> { + let meta = StoredMeta { + policy: PolicyRepr::from(&new_policy), + trailers: self.trailers.clone(), + }; + let bytes = serialize_meta(&meta)?; + let tmp = temp_sibling(&self.meta_path); + fs_shims::write(&tmp, &bytes).await?; + fs_shims::rename(&tmp, &self.meta_path).await?; + + self.policy = new_policy; + Ok(()) + } + + async fn open(self) -> io::Result { + let len = fs_shims::metadata_len(&self.body_path).await?; + let reader = fs_shims::open(&self.body_path).await?; + let source = FsBodySource { + reader, + trailers: self.trailers, + }; + Ok(Body::new_with_trailers(source, Some(len))) + } +} + +// BodySource over a stored body file. Reads stream straight from the file; trailers surface +// after EOF. +struct FsBodySource { + reader: fs_shims::Reader, + trailers: Option, +} + +impl AsyncRead for FsBodySource { + fn poll_read( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut [u8], + ) -> Poll> { + Pin::new(&mut self.get_mut().reader).poll_read(cx, buf) + } +} + +impl BodySource for FsBodySource { + fn trailers(self: Pin<&mut Self>) -> Option { + self.get_mut().trailers.take() + } +} + +fn hash_hex(bytes: &[u8]) -> String { + let mut hasher = Sha256::new(); + hasher.update(bytes); + finalize_hex(hasher) +} + +fn key_hash(key: &CacheKey) -> String { + hash_hex(key.to_string().as_bytes()) +} + +fn variant_hash(policy: &CachePolicy) -> String { + let mut hasher = Sha256::new(); + for (name, value) in &policy.vary_snapshot { + hasher.update(name.as_bytes()); + hasher.update([0]); + match value { + Some(value) => { + hasher.update([1]); + hasher.update(value.as_bytes()); + } + None => hasher.update([0]), + } + hasher.update([0]); + } + finalize_hex(hasher) +} + +fn finalize_hex(hasher: Sha256) -> String { + let digest = hasher.finalize(); + let mut out = String::with_capacity(digest.len() * 2); + for byte in digest { + write!(out, "{byte:02x}").expect("writing to a String cannot fail"); + } + out +} + +// A unique sibling temp path for atomically rewriting `path`. +fn temp_sibling(path: &Path) -> PathBuf { + let n = TEMP_COUNTER.fetch_add(1, Ordering::Relaxed); + let mut name = path.as_os_str().to_owned(); + name.push(format!(".tmp.{n}")); + PathBuf::from(name) +} + +fn serialize_meta(meta: &StoredMeta) -> io::Result { + rkyv::to_bytes::(meta) + .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error)) +} + +fn deserialize_meta(bytes: &[u8]) -> io::Result { + // A disk read lands in a buffer aligned only to 1, but rkyv's validated access requires + // the archived root to be aligned; copy into an `AlignedVec` before decoding. + let mut aligned = rkyv::util::AlignedVec::<16>::new(); + aligned.extend_from_slice(bytes); + rkyv::from_bytes::(&aligned) + .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error)) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::test_helpers::*; + use futures_lite::{AsyncReadExt, AsyncWriteExt}; + use std::time::{Duration, SystemTime}; + use tempfile::TempDir; + use trillium_client::Conn; + use trillium_http::{KnownHeaderName::*, Method, Status}; + use trillium_testing::{TestResult, harness, test}; + + fn key() -> CacheKey { + CacheKey::new(Method::Get, "http://example.com/".parse().unwrap()) + } + + fn new_storage() -> (TempDir, FileSystemStorage) { + let dir = tempfile::tempdir().unwrap(); + let storage = FileSystemStorage::new(dir.path()); + (dir, storage) + } + + async fn store_at(storage: &FileSystemStorage, url: &str, body: &[u8]) { + let key = CacheKey::new(Method::Get, url.parse().unwrap()); + let conn = exchange( + Method::Get, + &[], + Status::Ok, + &[(CacheControl, "max-age=600")], + ); + let policy = policy_from(&conn, SystemTime::now(), private_cache()); + let mut handle = storage.put(key, policy).await.unwrap(); + handle.write_all(body).await.unwrap(); + handle.finalize(None).await.unwrap(); + } + + async fn store(storage: &FileSystemStorage, key: CacheKey, conn: &Conn, body: &[u8]) { + let policy = policy_from(conn, SystemTime::now(), private_cache()); + let mut handle = storage.put(key, policy).await.unwrap(); + handle.write_all(body).await.unwrap(); + handle.finalize(None).await.unwrap(); + } + + async fn read_body(entry: FsStoredEntry) -> Vec { + let mut body = entry.open().await.unwrap(); + let mut buf = Vec::new(); + body.read_to_end(&mut buf).await.unwrap(); + buf + } + + #[test(harness)] + async fn get_missing_key_returns_empty() -> TestResult { + let (_dir, storage) = new_storage(); + assert!(storage.get(&key()).await.is_empty()); + Ok(()) + } + + #[test(harness)] + async fn put_then_get_round_trips_through_disk() -> TestResult { + let (_dir, storage) = new_storage(); + let conn = exchange( + Method::Get, + &[], + Status::Ok, + &[(CacheControl, "max-age=600")], + ); + store(&storage, key(), &conn, b"hello").await; + let result = storage.get(&key()).await; + assert_eq!(result.len(), 1); + assert_eq!(read_body(result[0].clone()).await, b"hello"); + Ok(()) + } + + #[test(harness)] + async fn put_with_same_vary_replaces() -> TestResult { + let (_dir, storage) = new_storage(); + let conn = exchange( + Method::Get, + &[(AcceptEncoding, "gzip")], + Status::Ok, + &[(CacheControl, "max-age=600"), (Vary, "Accept-Encoding")], + ); + store(&storage, key(), &conn, b"v1").await; + store(&storage, key(), &conn, b"v2").await; + let result = storage.get(&key()).await; + assert_eq!(result.len(), 1); + assert_eq!(read_body(result[0].clone()).await, b"v2"); + Ok(()) + } + + #[test(harness)] + async fn put_with_different_vary_appends() -> TestResult { + let (_dir, storage) = new_storage(); + let gzip = exchange( + Method::Get, + &[(AcceptEncoding, "gzip")], + Status::Ok, + &[(CacheControl, "max-age=600"), (Vary, "Accept-Encoding")], + ); + let br = exchange( + Method::Get, + &[(AcceptEncoding, "br")], + Status::Ok, + &[(CacheControl, "max-age=600"), (Vary, "Accept-Encoding")], + ); + store(&storage, key(), &gzip, b"gz").await; + store(&storage, key(), &br, b"br").await; + assert_eq!(storage.get(&key()).await.len(), 2); + Ok(()) + } + + #[test(harness)] + async fn invalidate_removes_all_entries_for_key() -> TestResult { + let (_dir, storage) = new_storage(); + let conn = exchange( + Method::Get, + &[], + Status::Ok, + &[(CacheControl, "max-age=600")], + ); + store(&storage, key(), &conn, b"x").await; + storage.invalidate(&key()).await; + assert!(storage.get(&key()).await.is_empty()); + Ok(()) + } + + #[test(harness)] + async fn invalidate_does_not_touch_other_keys() -> TestResult { + let (_dir, storage) = new_storage(); + let conn = exchange( + Method::Get, + &[], + Status::Ok, + &[(CacheControl, "max-age=600")], + ); + let key_a = CacheKey::new(Method::Get, "http://a.example/".parse().unwrap()); + let key_b = CacheKey::new(Method::Get, "http://b.example/".parse().unwrap()); + store(&storage, key_a.clone(), &conn, b"a").await; + store(&storage, key_b.clone(), &conn, b"b").await; + storage.invalidate(&key_a).await; + assert!(storage.get(&key_a).await.is_empty()); + assert_eq!(storage.get(&key_b).await.len(), 1); + Ok(()) + } + + #[test(harness)] + async fn drop_put_handle_without_finalize_discards() -> TestResult { + let (_dir, storage) = new_storage(); + let conn = exchange( + Method::Get, + &[], + Status::Ok, + &[(CacheControl, "max-age=600")], + ); + let policy = policy_from(&conn, SystemTime::now(), private_cache()); + let mut handle = storage.put(key(), policy).await.unwrap(); + handle.write_all(b"partial").await.unwrap(); + drop(handle); + assert!(storage.get(&key()).await.is_empty()); + Ok(()) + } + + #[test(harness)] + async fn refresh_policy_updates_meta_and_keeps_body() -> TestResult { + let (_dir, storage) = new_storage(); + let conn = exchange( + Method::Get, + &[], + Status::Ok, + &[(CacheControl, "max-age=600")], + ); + store(&storage, key(), &conn, b"body").await; + + let mut entries = storage.get(&key()).await; + let original_time = entries[0].policy().response_time; + let refreshed = exchange( + Method::Get, + &[], + Status::Ok, + &[(CacheControl, "max-age=1200")], + ); + let new_policy = policy_from( + &refreshed, + original_time + Duration::from_secs(100), + private_cache(), + ); + entries[0].refresh_policy(new_policy).await.unwrap(); + + let fresh = storage.get(&key()).await; + assert_eq!(fresh.len(), 1); + assert_ne!(fresh[0].policy().response_time, original_time); + assert_eq!(read_body(fresh[0].clone()).await, b"body"); + Ok(()) + } + + #[test(harness)] + async fn trailers_round_trip() -> TestResult { + let (_dir, storage) = new_storage(); + let conn = exchange( + Method::Get, + &[], + Status::Ok, + &[(CacheControl, "max-age=600")], + ); + let policy = policy_from(&conn, SystemTime::now(), private_cache()); + let mut handle = storage.put(key(), policy).await.unwrap(); + handle.write_all(b"data").await.unwrap(); + let mut trailers = Headers::new(); + trailers.insert("x-checksum", "abc123"); + handle.finalize(Some(trailers)).await.unwrap(); + + let entry = storage.get(&key()).await.remove(0); + let mut body = entry.open().await.unwrap(); + let mut buf = Vec::new(); + body.read_to_end(&mut buf).await.unwrap(); + assert_eq!(buf, b"data"); + let trailers = body + .trailers() + .expect("stored trailers should surface after EOF"); + assert_eq!(trailers.get_str("x-checksum"), Some("abc123")); + Ok(()) + } + + #[test(harness)] + async fn persists_across_new_storage_on_same_root() -> TestResult { + let dir = tempfile::tempdir().unwrap(); + let conn = exchange( + Method::Get, + &[], + Status::Ok, + &[(CacheControl, "max-age=600")], + ); + { + let storage = FileSystemStorage::new(dir.path()); + store(&storage, key(), &conn, b"persisted").await; + } + + // A brand-new storage over the same directory sees the prior instance's entry. + let reopened = FileSystemStorage::new(dir.path()); + let result = reopened.get(&key()).await; + assert_eq!(result.len(), 1); + assert_eq!(read_body(result[0].clone()).await, b"persisted"); + Ok(()) + } + + #[test(harness)] + async fn size_cap_evicts_and_deletes_files() -> TestResult { + // Cap at 1 KiB; write ten 600-byte bodies under distinct URLs. + let dir = tempfile::tempdir().unwrap(); + let storage = FileSystemStorage::new(dir.path()).with_max_capacity_bytes(1024); + let body = vec![b'x'; 600]; + for i in 0..10 { + store_at(&storage, &format!("http://example.com/{i}"), &body).await; + } + storage.run_pending_tasks().await; + assert!( + storage.weighted_size() <= 1024, + "weighted size {} should be within cap of 1024", + storage.weighted_size() + ); + + // A fresh unbounded scan of the same root reflects only the files still on disk, so + // the low total proves evicted variants' files were actually deleted, not just + // forgotten by the index. + let reopened = FileSystemStorage::new(dir.path()).unbounded(); + assert!( + reopened.weighted_size() <= 1024, + "on-disk bytes {} should be within cap of 1024", + reopened.weighted_size() + ); + Ok(()) + } + + #[test(harness)] + async fn rebuild_scan_trims_over_cap_directory() -> TestResult { + let dir = tempfile::tempdir().unwrap(); + let body = vec![b'x'; 600]; + { + let unbounded = FileSystemStorage::new(dir.path()).unbounded(); + for i in 0..10 { + store_at(&unbounded, &format!("http://example.com/{i}"), &body).await; + } + unbounded.run_pending_tasks().await; + assert_eq!(unbounded.entry_count(), 10); + } + + // Reopening with a cap trims the pre-existing directory to fit during construction. + let capped = FileSystemStorage::new(dir.path()).with_max_capacity_bytes(1024); + assert!( + capped.weighted_size() <= 1024, + "weighted size {} should be within cap of 1024", + capped.weighted_size() + ); + Ok(()) + } + + #[test(harness)] + async fn unbounded_keeps_all_entries() -> TestResult { + let dir = tempfile::tempdir().unwrap(); + let storage = FileSystemStorage::new(dir.path()).unbounded(); + let body = vec![b'x'; 600]; + for i in 0..10 { + store_at(&storage, &format!("http://example.com/{i}"), &body).await; + } + storage.run_pending_tasks().await; + assert_eq!(storage.entry_count(), 10); + assert_eq!(storage.weighted_size(), 6000); + Ok(()) + } + + #[test(harness)] + async fn replacing_a_variant_does_not_double_count() -> TestResult { + let (_dir, storage) = new_storage(); + store_at(&storage, "http://example.com/", &vec![b'x'; 600]).await; + store_at(&storage, "http://example.com/", &vec![b'y'; 300]).await; + storage.run_pending_tasks().await; + assert_eq!(storage.entry_count(), 1); + assert_eq!(storage.weighted_size(), 300); + Ok(()) + } +} diff --git a/src/fs_shims.rs b/src/fs_shims.rs new file mode 100644 index 0000000..ea2e03e --- /dev/null +++ b/src/fs_shims.rs @@ -0,0 +1,226 @@ +//! Runtime filesystem shims for [`FileSystemStorage`](crate::FileSystemStorage). +//! +//! Selects a concrete async filesystem implementation from the enabled runtime feature +//! (`smol`, `tokio`, or `async-std`) and exposes a small uniform surface: streaming +//! [`Reader`]/[`Writer`] handles plus whole-file and directory operations. With no runtime +//! feature enabled the surface still exists but every call panics, so the crate — and its +//! documentation — builds with `fs` alone. + +use std::{ + io, + path::{Path, PathBuf}, +}; + +cfg_if::cfg_if! { + if #[cfg(feature = "tokio")] { + // tokio's `File` speaks `tokio::io`; wrap it so callers see `futures_lite` traits. + pub(crate) type Reader = async_compat::Compat; + pub(crate) type Writer = async_compat::Compat; + + pub(crate) async fn open(path: &Path) -> io::Result { + Ok(async_compat::Compat::new(tokio::fs::File::open(path).await?)) + } + + pub(crate) async fn create(path: &Path) -> io::Result { + Ok(async_compat::Compat::new(tokio::fs::File::create(path).await?)) + } + + pub(crate) async fn read(path: &Path) -> io::Result> { + tokio::fs::read(path).await + } + + pub(crate) async fn write(path: &Path, contents: &[u8]) -> io::Result<()> { + tokio::fs::write(path, contents).await + } + + pub(crate) async fn create_dir_all(path: &Path) -> io::Result<()> { + tokio::fs::create_dir_all(path).await + } + + pub(crate) async fn rename(from: &Path, to: &Path) -> io::Result<()> { + tokio::fs::rename(from, to).await + } + + pub(crate) async fn remove_dir_all(path: &Path) -> io::Result<()> { + tokio::fs::remove_dir_all(path).await + } + + pub(crate) async fn metadata_len(path: &Path) -> io::Result { + Ok(tokio::fs::metadata(path).await?.len()) + } + + pub(crate) async fn read_dir_paths(path: &Path) -> io::Result> { + let mut read_dir = tokio::fs::read_dir(path).await?; + let mut paths = Vec::new(); + while let Some(entry) = read_dir.next_entry().await? { + paths.push(entry.path()); + } + Ok(paths) + } + } else if #[cfg(feature = "async-std")] { + pub(crate) type Reader = async_std::fs::File; + pub(crate) type Writer = async_std::fs::File; + + pub(crate) async fn open(path: &Path) -> io::Result { + async_std::fs::File::open(path).await + } + + pub(crate) async fn create(path: &Path) -> io::Result { + async_std::fs::File::create(path).await + } + + pub(crate) async fn read(path: &Path) -> io::Result> { + async_std::fs::read(path).await + } + + pub(crate) async fn write(path: &Path, contents: &[u8]) -> io::Result<()> { + async_std::fs::write(path, contents).await + } + + pub(crate) async fn create_dir_all(path: &Path) -> io::Result<()> { + async_std::fs::create_dir_all(path).await + } + + pub(crate) async fn rename(from: &Path, to: &Path) -> io::Result<()> { + async_std::fs::rename(from, to).await + } + + pub(crate) async fn remove_dir_all(path: &Path) -> io::Result<()> { + async_std::fs::remove_dir_all(path).await + } + + pub(crate) async fn metadata_len(path: &Path) -> io::Result { + Ok(async_std::fs::metadata(path).await?.len()) + } + + pub(crate) async fn read_dir_paths(path: &Path) -> io::Result> { + use futures_lite::StreamExt; + let mut read_dir = async_std::fs::read_dir(path).await?; + let mut paths = Vec::new(); + while let Some(entry) = read_dir.next().await { + paths.push(entry?.path().into()); + } + Ok(paths) + } + } else if #[cfg(feature = "smol")] { + pub(crate) type Reader = async_fs::File; + pub(crate) type Writer = async_fs::File; + + pub(crate) async fn open(path: &Path) -> io::Result { + async_fs::File::open(path).await + } + + pub(crate) async fn create(path: &Path) -> io::Result { + async_fs::File::create(path).await + } + + pub(crate) async fn read(path: &Path) -> io::Result> { + async_fs::read(path).await + } + + pub(crate) async fn write(path: &Path, contents: &[u8]) -> io::Result<()> { + async_fs::write(path, contents).await + } + + pub(crate) async fn create_dir_all(path: &Path) -> io::Result<()> { + async_fs::create_dir_all(path).await + } + + pub(crate) async fn rename(from: &Path, to: &Path) -> io::Result<()> { + async_fs::rename(from, to).await + } + + pub(crate) async fn remove_dir_all(path: &Path) -> io::Result<()> { + async_fs::remove_dir_all(path).await + } + + pub(crate) async fn metadata_len(path: &Path) -> io::Result { + Ok(async_fs::metadata(path).await?.len()) + } + + pub(crate) async fn read_dir_paths(path: &Path) -> io::Result> { + use futures_lite::StreamExt; + let mut read_dir = async_fs::read_dir(path).await?; + let mut paths = Vec::new(); + while let Some(entry) = read_dir.next().await { + paths.push(entry?.path()); + } + Ok(paths) + } + } else { + use std::{pin::Pin, task::{Context, Poll}}; + + const NO_RUNTIME: &str = + "enable the `smol`, `tokio`, or `async-std` feature to use FileSystemStorage"; + + #[derive(Debug)] + pub(crate) struct Reader; + + #[derive(Debug)] + pub(crate) struct Writer; + + impl futures_lite::AsyncRead for Reader { + fn poll_read( + self: Pin<&mut Self>, + _cx: &mut Context<'_>, + _buf: &mut [u8], + ) -> Poll> { + unimplemented!("{NO_RUNTIME}") + } + } + + impl futures_lite::AsyncWrite for Writer { + fn poll_write( + self: Pin<&mut Self>, + _cx: &mut Context<'_>, + _buf: &[u8], + ) -> Poll> { + unimplemented!("{NO_RUNTIME}") + } + + fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + unimplemented!("{NO_RUNTIME}") + } + + fn poll_close(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + unimplemented!("{NO_RUNTIME}") + } + } + + pub(crate) async fn open(_path: &Path) -> io::Result { + unimplemented!("{NO_RUNTIME}") + } + + pub(crate) async fn create(_path: &Path) -> io::Result { + unimplemented!("{NO_RUNTIME}") + } + + pub(crate) async fn read(_path: &Path) -> io::Result> { + unimplemented!("{NO_RUNTIME}") + } + + pub(crate) async fn write(_path: &Path, _contents: &[u8]) -> io::Result<()> { + unimplemented!("{NO_RUNTIME}") + } + + pub(crate) async fn create_dir_all(_path: &Path) -> io::Result<()> { + unimplemented!("{NO_RUNTIME}") + } + + pub(crate) async fn rename(_from: &Path, _to: &Path) -> io::Result<()> { + unimplemented!("{NO_RUNTIME}") + } + + pub(crate) async fn remove_dir_all(_path: &Path) -> io::Result<()> { + unimplemented!("{NO_RUNTIME}") + } + + pub(crate) async fn metadata_len(_path: &Path) -> io::Result { + unimplemented!("{NO_RUNTIME}") + } + + pub(crate) async fn read_dir_paths(_path: &Path) -> io::Result> { + unimplemented!("{NO_RUNTIME}") + } + } +} diff --git a/src/lib.rs b/src/lib.rs index cef4842..8844c32 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -49,15 +49,24 @@ mod server; mod storability; mod storage; mod tee; +mod tiered; mod validation; +#[cfg(feature = "fs")] +mod fs; +#[cfg(feature = "fs")] +mod fs_shims; + #[cfg(feature = "client")] pub mod client; #[cfg(test)] mod test_helpers; +#[cfg(feature = "fs")] +pub use fs::{FileSystemStorage, FsPutHandle, FsStoredEntry}; pub use memory::{InMemoryEntry, InMemoryPutHandle, InMemoryStorage}; pub use policy::{CacheOptions, CachePolicy}; pub use server::Cache; pub use storage::{CacheKey, CacheStorage, PutHandle, StoredEntry}; +pub use tiered::{TieredEntry, TieredPutHandle, TieredStorage}; diff --git a/src/memory.rs b/src/memory.rs index c76803c..54860e4 100644 --- a/src/memory.rs +++ b/src/memory.rs @@ -1,30 +1,4 @@ //! In-memory [`CacheStorage`]. -//! -//! [`InMemoryStorage`] is suitable for production reverse-proxy and -//! client-side caching: byte-aware size cap, scan-resistant -//! admission, and concurrent reads and writes on distinct keys -//! without contention. -//! -//! ## Granularity -//! -//! Eviction is coarse: the unit is one [`CacheKey`] (method + URL), -//! and all `Vary` variants stored under that key live and die together -//! during eviction. In typical traffic patterns variants of the same URL -//! are hot or cold together (a single `Accept-Encoding` is usually -//! dominant, etc.), so the cost is bounded — at worst we keep a few -//! cold variants resident alongside one hot variant. This is correct -//! per RFC 9111; the only consequence is slightly less efficient use -//! of memory than per-variant eviction would give. -//! -//! ## Sizing -//! -//! The byte cap is enforced over stored *body* bytes only (the -//! dominant cost); headers and other metadata are not counted. The -//! per-response cap on [`Cache::with_max_cacheable_size`] interacts -//! independently — that one bounds how large any single response may -//! be; the storage cap bounds total resident size across the cache. -//! -//! [`Cache::with_max_cacheable_size`]: crate::Cache::with_max_cacheable_size use crate::{CacheKey, CachePolicy, CacheStorage, PutHandle, StoredEntry}; use futures_lite::{AsyncRead, AsyncWrite}; @@ -63,6 +37,10 @@ impl Debug for Variant { /// Bounded in-memory cache storage. /// +/// Suitable for production reverse-proxy and client-side caching: byte-aware size cap, +/// scan-resistant admission, and concurrent reads and writes on distinct keys without +/// contention. +/// /// Defaults to a 256 MiB byte cap; override with /// [`with_max_capacity_bytes`][Self::with_max_capacity_bytes], /// [`unbounded`][Self::unbounded], @@ -72,6 +50,24 @@ impl Debug for Variant { /// construction, before the storage is populated or shared. /// /// `Clone` is cheap — clones share the same backing storage. +/// +/// # Granularity +/// +/// Eviction is coarse: the unit is one [`CacheKey`] (method + URL), and all `Vary` variants +/// stored under that key live and die together during eviction. In typical traffic patterns +/// variants of the same URL are hot or cold together (a single `Accept-Encoding` is usually +/// dominant, etc.), so the cost is bounded — at worst we keep a few cold variants resident +/// alongside one hot variant. This is correct per RFC 9111; the only consequence is slightly +/// less efficient use of memory than per-variant eviction would give. +/// +/// # Sizing +/// +/// The byte cap is enforced over stored *body* bytes only (the dominant cost); headers and +/// other metadata are not counted. The per-response cap on [`Cache::with_max_cacheable_size`] +/// interacts independently — that one bounds how large any single response may be; the storage +/// cap bounds total resident size across the cache. +/// +/// [`Cache::with_max_cacheable_size`]: crate::Cache::with_max_cacheable_size #[derive(Clone)] pub struct InMemoryStorage { cache: Cache, diff --git a/src/policy.rs b/src/policy.rs index 734f609..8b1d7be 100644 --- a/src/policy.rs +++ b/src/policy.rs @@ -37,6 +37,31 @@ pub(crate) fn effective_response_cache_control( (response_headers.cache_control(), false) } +// Derive the effective response Cache-Control and its targeted-field flag from a +// response's headers and caching options. This is a pure function of the two inputs. +fn derive_response_cache_control( + response_headers: &Headers, + options: &CacheOptions, +) -> (Option, bool) { + let (mut response_cache_control, targeted_cc_in_effect) = + effective_response_cache_control(response_headers, options); + + // RFC 9111 §5.4: when no Cache-Control is present, treat + // `Pragma: no-cache` as if `Cache-Control: no-cache` were set. This + // is suppressed when a targeted field took effect (Pragma is part of + // the Cache-Control / Expires family the targeted-field rule + // displaces). + if response_cache_control.is_none() + && response_headers + .get_str(KnownHeaderName::Pragma) + .is_some_and(|p| p.contains("no-cache")) + { + response_cache_control = Some(CacheControlHeader::from(CacheControlDirective::NoCache)); + } + + (response_cache_control, targeted_cc_in_effect) +} + /// RFC 9213 §2.1: targeted fields are Dictionary Structured Fields (RFC /// 8941 §3.2). A full SF parser is out of scope, but this catches the /// common "garbage trailing tokens" case (e.g. `max-age=10000, &&&&&`) by @@ -167,21 +192,8 @@ impl CachePolicy { response_time: SystemTime, options: CacheOptions, ) -> Self { - let (mut response_cache_control, targeted_cc_in_effect) = - effective_response_cache_control(&response_headers, &options); - - // RFC 9111 §5.4: when no Cache-Control is present, treat - // `Pragma: no-cache` as if `Cache-Control: no-cache` were set. This - // is suppressed when a targeted field took effect (Pragma is part of - // the Cache-Control / Expires family the targeted-field rule - // displaces). - if response_cache_control.is_none() - && response_headers - .get_str(KnownHeaderName::Pragma) - .is_some_and(|p| p.contains("no-cache")) - { - response_cache_control = Some(CacheControlHeader::from(CacheControlDirective::NoCache)); - } + let (response_cache_control, targeted_cc_in_effect) = + derive_response_cache_control(&response_headers, &options); let vary_snapshot = build_vary_snapshot(&response_headers, request_headers); @@ -198,6 +210,81 @@ impl CachePolicy { } } +// On-disk proxy for `CachePolicy`, used by the `FileSystemStorage` backend to persist a +// policy through rkyv. It carries only the fields captured directly from the exchange; +// `response_cache_control` and `targeted_cc_in_effect` are a pure function of the stored +// headers and options, so they are recomputed on load rather than serialized. `CacheOptions` +// is flattened into individual fields so no rkyv-archived type is generated for the public +// `CacheOptions`; destructuring it here makes a future added field a compile error until it +// is threaded through. +#[cfg(feature = "fs")] +#[derive(rkyv::Archive, rkyv::Serialize, rkyv::Deserialize)] +pub(crate) struct PolicyRepr { + request_method: Method, + vary_snapshot: Vec<(String, Option)>, + response_status: Status, + response_headers: Headers, + #[rkyv(with = rkyv::with::AsUnixTime)] + response_time: SystemTime, + shared: bool, + cache_heuristic: f32, + immutable_min_time_to_live: Duration, +} + +#[cfg(feature = "fs")] +impl From<&CachePolicy> for PolicyRepr { + fn from(policy: &CachePolicy) -> Self { + let CacheOptions { + shared, + cache_heuristic, + immutable_min_time_to_live, + } = policy.options; + Self { + request_method: policy.request_method, + vary_snapshot: policy.vary_snapshot.clone(), + response_status: policy.response_status, + response_headers: policy.response_headers.clone(), + response_time: policy.response_time, + shared, + cache_heuristic, + immutable_min_time_to_live, + } + } +} + +#[cfg(feature = "fs")] +impl From for CachePolicy { + fn from(repr: PolicyRepr) -> Self { + let PolicyRepr { + request_method, + vary_snapshot, + response_status, + response_headers, + response_time, + shared, + cache_heuristic, + immutable_min_time_to_live, + } = repr; + let options = CacheOptions { + shared, + cache_heuristic, + immutable_min_time_to_live, + }; + let (response_cache_control, targeted_cc_in_effect) = + derive_response_cache_control(&response_headers, &options); + Self { + request_method, + vary_snapshot, + response_status, + response_headers, + response_cache_control, + targeted_cc_in_effect, + response_time, + options, + } + } +} + fn build_vary_snapshot( response_headers: &Headers, request_headers: &Headers, @@ -325,4 +412,41 @@ mod tests { vec![("accept-encoding".to_string(), None)] ); } + + #[cfg(feature = "fs")] + #[test] + fn policy_round_trips_through_rkyv() { + let conn = exchange( + Method::Get, + &[(AcceptEncoding, "gzip")], + Status::Ok, + &[(CacheControl, "max-age=600"), (Vary, "Accept-Encoding")], + ); + let policy = policy_from(&conn, SystemTime::now(), private_cache()); + + let repr = PolicyRepr::from(&policy); + let bytes = rkyv::to_bytes::(&repr).unwrap(); + let restored: CachePolicy = rkyv::from_bytes::(&bytes) + .unwrap() + .into(); + + assert_eq!(restored.request_method, policy.request_method); + assert_eq!(restored.response_status, policy.response_status); + assert_eq!(restored.vary_snapshot, policy.vary_snapshot); + assert_eq!(restored.response_time, policy.response_time); + assert_eq!( + restored.response_headers.get_str(CacheControl), + policy.response_headers.get_str(CacheControl) + ); + assert_eq!( + restored.response_headers.get_str(Vary), + policy.response_headers.get_str(Vary) + ); + // recomputed from the stored headers + options, not serialized + assert_eq!(restored.targeted_cc_in_effect, policy.targeted_cc_in_effect); + assert_eq!( + restored.response_cache_control.is_some(), + policy.response_cache_control.is_some() + ); + } } diff --git a/src/tiered.rs b/src/tiered.rs new file mode 100644 index 0000000..35bce44 --- /dev/null +++ b/src/tiered.rs @@ -0,0 +1,516 @@ +//! Tiered [`CacheStorage`] composing a fast hot tier over a durable cold tier. + +use crate::{CacheKey, CachePolicy, CacheStorage, PutHandle, StoredEntry, tee::TeeingReader}; +use futures_lite::{AsyncReadExt, AsyncWrite, AsyncWriteExt}; +use std::{ + fmt::{self, Debug, Formatter}, + io, + pin::Pin, + task::{Context, Poll}, +}; +use trillium_http::{Body, Headers}; +use trillium_server_common::{Runtime, RuntimeTrait}; + +/// Two-tier cache storage: a fast hot tier over a durable cold tier. +/// +/// `TieredStorage` layers two backends: a `Hot` tier serving the working set from fast storage +/// and a `Cold` tier holding the larger, durable set. It is itself a [`CacheStorage`], so it +/// drops in wherever a single backend would go — the headline pairing is an [`InMemoryStorage`] +/// hot tier over a [`FileSystemStorage`] cold tier, but any two backends compose. +/// +/// `Clone` is available when both tiers are `Clone`, and shares their backing storage. +/// +/// # Runtime +/// +/// The write path finishes asynchronously (see below), so a `TieredStorage` is constructed with +/// the [`Runtime`] it spawns that background work on — and it must be the runtime actually +/// driving the process, or the flush never makes progress. Construct the adapter for your +/// runtime directly (for example `trillium_smol::SmolRuntime::default()` or +/// `trillium_tokio::TokioRuntime::default()`); on a client you can instead take it from the +/// connector with `client.connector().runtime()`. The `tiered_cache` example wires this up end +/// to end. +/// +/// # Read path +/// +/// [`get`] consults the hot tier first and, on a hit, serves from it alone. On a hot miss it +/// reads the cold tier; opening a cold entry *promotes* it, streaming the body to the reader +/// and into the hot tier at once (the same teeing used on the origin→user+storage path), so +/// the working set migrates into fast storage as it is served. A hot tier emptied by a restart +/// repopulates from cold as entries are read. +/// +/// The hot-first lookup assumes the hot tier evicts a whole [`CacheKey`] at once — all `Vary` +/// variants of a URL together — so a hot hit implies the full variant set for that key is +/// present. [`InMemoryStorage`] satisfies this. A hot tier that evicts individual variants +/// could leave siblings only in cold and hide them behind a hot hit; pair `TieredStorage` with +/// a whole-key-eviction hot tier. +/// +/// # Write path +/// +/// [`put`] writes the body into the hot tier as it streams, then finalizing the entry spawns a +/// background task that copies it into the cold tier — a write-back. The hot tier is populated +/// synchronously; cold durability follows shortly after, off the request path. A crash in that +/// window loses the not-yet-flushed entry, which for a cache means only an extra origin fetch. +/// Because cold ends up holding every stored entry, evicting from hot only drops a fast-path +/// copy — the entry stays served from cold and re-promotes on its next read. +/// +/// # Policy refresh +/// +/// A 304 revalidation refreshes the policy on whichever tier served the entry. After a hot +/// eviction a request may fall through to a cold copy carrying the pre-refresh policy and +/// revalidate once more; the content served is always correct. +/// +/// [`InMemoryStorage`]: crate::InMemoryStorage +/// [`FileSystemStorage`]: crate::FileSystemStorage +/// [`get`]: CacheStorage::get +/// [`put`]: CacheStorage::put +pub struct TieredStorage { + hot: Hot, + cold: Cold, + runtime: Runtime, +} + +impl TieredStorage { + /// Compose `hot` and `cold` into a tiered storage, spawning background write-back onto + /// `runtime`. Lookups and promotions favor `hot`; every stored entry is flushed through to + /// `cold`. + /// + /// Pass the runtime the surrounding server or client already runs on. + pub fn new(hot: Hot, cold: Cold, runtime: impl RuntimeTrait) -> Self { + Self { + hot, + cold, + runtime: runtime.into(), + } + } + + /// Borrow the hot tier. + pub fn hot(&self) -> &Hot { + &self.hot + } + + /// Borrow the cold tier. + pub fn cold(&self) -> &Cold { + &self.cold + } +} + +impl Clone for TieredStorage { + fn clone(&self) -> Self { + Self { + hot: self.hot.clone(), + cold: self.cold.clone(), + runtime: self.runtime.clone(), + } + } +} + +impl Debug for TieredStorage { + fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { + f.debug_struct("TieredStorage") + .field("hot", &self.hot) + .field("cold", &self.cold) + .finish_non_exhaustive() + } +} + +impl CacheStorage for TieredStorage +where + Hot: CacheStorage + Clone, + Cold: CacheStorage + Clone, +{ + type StoredEntry = TieredEntry; + type PutHandle = TieredPutHandle; + + async fn get(&self, key: &CacheKey) -> Vec { + let hot = self.hot.get(key).await; + if !hot.is_empty() { + return hot.into_iter().map(TieredEntry::Hot).collect(); + } + self.cold + .get(key) + .await + .into_iter() + .map(|entry| TieredEntry::Cold { + entry, + hot: self.hot.clone(), + key: key.clone(), + }) + .collect() + } + + async fn put(&self, key: CacheKey, policy: CachePolicy) -> io::Result { + let hot = self.hot.put(key.clone(), policy.clone()).await?; + Ok(TieredPutHandle { + hot, + hot_store: self.hot.clone(), + cold: self.cold.clone(), + runtime: self.runtime.clone(), + key, + policy, + }) + } + + async fn invalidate(&self, key: &CacheKey) { + self.hot.invalidate(key).await; + self.cold.invalidate(key).await; + } +} + +/// One stored response from a [`TieredStorage`], held in either tier. +/// +/// A cold-tier entry carries a handle to the hot tier and its key so that +/// [`open`][StoredEntry::open] can promote it — streaming the body to the reader and into the +/// hot tier at once. +pub enum TieredEntry { + /// An entry served from the hot tier. + Hot(Hot::StoredEntry), + /// An entry served from the cold tier, promoted into hot on open. + Cold { + /// The cold-tier entry. + entry: Cold::StoredEntry, + /// Hot tier to promote into. + hot: Hot, + /// Key the entry is stored under. + key: CacheKey, + }, +} + +impl Clone for TieredEntry +where + Hot: CacheStorage + Clone, + Cold: CacheStorage, +{ + fn clone(&self) -> Self { + match self { + Self::Hot(entry) => Self::Hot(entry.clone()), + Self::Cold { entry, hot, key } => Self::Cold { + entry: entry.clone(), + hot: hot.clone(), + key: key.clone(), + }, + } + } +} + +impl Debug for TieredEntry { + fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { + match self { + Self::Hot(entry) => f.debug_tuple("Hot").field(entry).finish(), + Self::Cold { entry, key, .. } => f + .debug_struct("Cold") + .field("entry", entry) + .field("key", key) + .finish_non_exhaustive(), + } + } +} + +impl StoredEntry for TieredEntry +where + Hot: CacheStorage + Clone, + Cold: CacheStorage, +{ + fn policy(&self) -> &CachePolicy { + match self { + Self::Hot(entry) => entry.policy(), + Self::Cold { entry, .. } => entry.policy(), + } + } + + async fn refresh_policy(&mut self, new_policy: CachePolicy) -> io::Result<()> { + match self { + Self::Hot(entry) => entry.refresh_policy(new_policy).await, + Self::Cold { entry, .. } => entry.refresh_policy(new_policy).await, + } + } + + async fn open(self) -> io::Result { + match self { + Self::Hot(entry) => entry.open().await, + Self::Cold { entry, hot, key } => { + let policy = entry.policy().clone(); + let cold_body = entry.open().await?; + let len = cold_body.len(); + match hot.put(key, policy).await { + Ok(put_handle) => { + let tee = TeeingReader::new(cold_body, put_handle, u64::MAX); + Ok(Body::new_with_trailers(tee, len)) + } + Err(e) => { + log::warn!("cache: promotion put failed: {e}, serving cold entry only"); + Ok(cold_body) + } + } + } + } + } +} + +/// Streaming [`PutHandle`] for [`TieredStorage`]. +/// +/// Body bytes stream into the hot tier; [`finalize`][PutHandle::finalize] commits the hot entry +/// and spawns a background task that copies it into the cold tier. Dropping without finalizing +/// aborts the hot write, and nothing reaches either tier. +pub struct TieredPutHandle { + hot: Hot::PutHandle, + hot_store: Hot, + cold: Cold, + runtime: Runtime, + key: CacheKey, + policy: CachePolicy, +} + +impl Debug for TieredPutHandle { + fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { + f.debug_struct("TieredPutHandle") + .field("key", &self.key) + .finish_non_exhaustive() + } +} + +// Only the hot `PutHandle` is ever polled through a pin (and `PutHandle: Unpin`); the storage +// handles and metadata are plain data, moved but never pin-projected. So the composite is +// `Unpin` regardless of whether the tier types are. +impl Unpin for TieredPutHandle {} + +impl AsyncWrite for TieredPutHandle { + fn poll_write( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + Pin::new(&mut self.get_mut().hot).poll_write(cx, buf) + } + + fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.get_mut().hot).poll_flush(cx) + } + + fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.get_mut().hot).poll_close(cx) + } +} + +impl PutHandle for TieredPutHandle +where + Hot: CacheStorage + Clone, + Cold: CacheStorage, +{ + async fn finalize(self, trailers: Option) -> io::Result<()> { + let Self { + hot, + hot_store, + cold, + runtime, + key, + policy, + } = self; + hot.finalize(trailers).await?; + + let log_key = key.clone(); + let _detached = runtime.spawn(async move { + if let Err(e) = flush_to_cold(hot_store, cold, key, policy).await { + log::warn!("cache: tiered background flush to cold failed for {log_key}: {e}"); + } + }); + Ok(()) + } +} + +// Copy the just-committed hot entry into the cold tier. Reads the entry back from hot (cheap +// when hot is in-memory) and streams it into a cold `put`, carrying over any trailers the hot +// body surfaces. A hot eviction between finalize and flush leaves nothing to copy — the entry +// is simply not yet durable in cold, which a later read re-promotes and re-flushes. +async fn flush_to_cold( + hot_store: Hot, + cold: Cold, + key: CacheKey, + policy: CachePolicy, +) -> io::Result<()> +where + Hot: CacheStorage, + Cold: CacheStorage, +{ + let Some(entry) = hot_store + .get(&key) + .await + .into_iter() + .find(|entry| entry.policy().same_variant_as(&policy)) + else { + return Ok(()); + }; + + let mut body = entry.open().await?; + let mut put = cold.put(key, policy).await?; + let mut buf = [0u8; 8192]; + loop { + let n = body.read(&mut buf).await?; + if n == 0 { + break; + } + put.write_all(&buf[..n]).await?; + } + let trailers = body.trailers(); + put.finalize(trailers).await +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{InMemoryStorage, test_helpers::*}; + use std::time::{Duration, SystemTime}; + use trillium_http::{KnownHeaderName::*, Method, Status}; + use trillium_testing::{TestResult, harness, runtime, test}; + + fn key() -> CacheKey { + CacheKey::new(Method::Get, "http://example.com/".parse().unwrap()) + } + + fn tiered() -> TieredStorage { + TieredStorage::new(InMemoryStorage::new(), InMemoryStorage::new(), runtime()) + } + + async fn store_into(storage: &impl CacheStorage, key: CacheKey, body: &[u8]) { + let conn = exchange( + Method::Get, + &[], + Status::Ok, + &[(CacheControl, "max-age=600")], + ); + let policy = policy_from(&conn, SystemTime::now(), private_cache()); + let mut handle = storage.put(key, policy).await.unwrap(); + handle.write_all(body).await.unwrap(); + handle.finalize(None).await.unwrap(); + } + + async fn read_body(entry: impl StoredEntry) -> Vec { + let mut body = entry.open().await.unwrap(); + let mut buf = Vec::new(); + body.read_to_end(&mut buf).await.unwrap(); + buf + } + + // Write-back finalizes cold on a spawned task; poll until it lands (or give up). + async fn cold_settles( + storage: &TieredStorage, + key: &CacheKey, + ) -> Vec + where + Hot: CacheStorage + Clone, + Cold: CacheStorage + Clone, + { + for _ in 0..200 { + let entries = storage.cold().get(key).await; + if !entries.is_empty() { + return entries; + } + storage.runtime.delay(Duration::from_millis(5)).await; + } + panic!("cold tier never populated"); + } + + #[test(harness)] + async fn hot_populated_synchronously_cold_written_back() -> TestResult { + let storage = tiered(); + store_into(&storage, key(), b"hello").await; + + // Hot is populated before finalize returns. + let entries = storage.get(&key()).await; + assert_eq!(entries.len(), 1); + assert!(matches!(entries[0], TieredEntry::Hot(_))); + assert_eq!(read_body(entries[0].clone()).await, b"hello"); + + // Cold catches up on the background task. + let cold = cold_settles(&storage, &key()).await; + assert_eq!(cold.len(), 1); + assert_eq!(read_body(cold[0].clone()).await, b"hello"); + Ok(()) + } + + #[test(harness)] + async fn cold_hit_promotes_into_hot() -> TestResult { + let storage = tiered(); + // Seed cold directly so hot starts empty — the post-restart / post-eviction shape. + store_into(storage.cold(), key(), b"promoted").await; + assert!(storage.hot().get(&key()).await.is_empty()); + + let entries = storage.get(&key()).await; + assert_eq!(entries.len(), 1); + assert!(matches!(entries[0], TieredEntry::Cold { .. })); + // Opening the cold entry streams it through into hot. + assert_eq!(read_body(entries[0].clone()).await, b"promoted"); + + let hot = storage.hot().get(&key()).await; + assert_eq!(hot.len(), 1); + assert_eq!(read_body(hot[0].clone()).await, b"promoted"); + Ok(()) + } + + #[test(harness)] + async fn invalidate_clears_both_tiers() -> TestResult { + let storage = tiered(); + store_into(&storage, key(), b"x").await; + cold_settles(&storage, &key()).await; + storage.invalidate(&key()).await; + assert!(storage.get(&key()).await.is_empty()); + assert!(storage.hot().get(&key()).await.is_empty()); + assert!(storage.cold().get(&key()).await.is_empty()); + Ok(()) + } + + #[test(harness)] + async fn drop_put_handle_without_finalize_stores_nothing() -> TestResult { + let storage = tiered(); + let conn = exchange( + Method::Get, + &[], + Status::Ok, + &[(CacheControl, "max-age=600")], + ); + let policy = policy_from(&conn, SystemTime::now(), private_cache()); + let mut handle = storage.put(key(), policy).await.unwrap(); + handle.write_all(b"partial").await.unwrap(); + drop(handle); + assert!(storage.hot().get(&key()).await.is_empty()); + assert!(storage.cold().get(&key()).await.is_empty()); + Ok(()) + } + + // The headline pairing: memory hot tier over a filesystem cold tier. A cold copy survives a + // fresh hot tier (the post-restart shape) and re-promotes into memory on read. + #[cfg(feature = "fs")] + #[test(harness)] + async fn memory_over_filesystem_promotes_from_disk() -> TestResult { + use crate::FileSystemStorage; + + let dir = tempfile::tempdir().unwrap(); + { + let storage = TieredStorage::new( + InMemoryStorage::new(), + FileSystemStorage::new(dir.path()), + runtime(), + ); + store_into(&storage, key(), b"on-disk").await; + cold_settles(&storage, &key()).await; + } + + // A fresh instance over the same directory: hot is empty, cold holds the entry on disk. + let reopened = TieredStorage::new( + InMemoryStorage::new(), + FileSystemStorage::new(dir.path()), + runtime(), + ); + assert!(reopened.hot().get(&key()).await.is_empty()); + + let entries = reopened.get(&key()).await; + assert_eq!(entries.len(), 1); + assert!(matches!(entries[0], TieredEntry::Cold { .. })); + assert_eq!(read_body(entries[0].clone()).await, b"on-disk"); + + // Promotion pulled it into memory. + let hot = reopened.hot().get(&key()).await; + assert_eq!(hot.len(), 1); + assert_eq!(read_body(hot[0].clone()).await, b"on-disk"); + Ok(()) + } +}