From 93d7724fbc4d8d60c75fa5d37ca54f856e0e2f9f Mon Sep 17 00:00:00 2001 From: Roland Groza Date: Fri, 21 Aug 2026 19:18:33 +0900 Subject: [PATCH 1/3] feat(lfs): cache git-LFS objects via batch proxy A client's git-lfs derives its LFS endpoint from the proxy URL, so the proxy must answer the batch API and object transfers or LFS-tracked files fail to check out. Proxy the batch API upstream, rewrite each download URL back to the proxy, and serve objects from a content-addressed cache: a miss fetches once (verifying sha256 == oid), a hit is served locally, and objects share the mirrors' cache cap and LRU eviction. Uploads are refused (read-only). The LFS transfer is in-process over rustls (no OpenSSL), keeping the binary static. Assisted-by: Claude:claude-opus-4-8 --- Cargo.lock | 944 +++++++++++++++++++++++++++++++++++++++++++- Cargo.toml | 8 + Dockerfile | 18 +- README.md | 64 ++- bench/README.md | 17 +- bench/lfs_origin.py | 70 ++++ bench/run.sh | 56 ++- deny.toml | 3 + src/evict.rs | 134 ++++++- src/lfs.rs | 553 ++++++++++++++++++++++++++ src/lib.rs | 1 + src/main.rs | 29 +- src/metrics.rs | 30 +- src/repo.rs | 106 ++++- src/server.rs | 173 ++++++++ tests/e2e.rs | 19 + tests/http.rs | 21 + tests/lfs.rs | 375 ++++++++++++++++++ 18 files changed, 2557 insertions(+), 64 deletions(-) create mode 100644 bench/lfs_origin.py create mode 100644 src/lfs.rs create mode 100644 tests/lfs.rs diff --git a/Cargo.lock b/Cargo.lock index 393318c..a849820 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -53,7 +53,7 @@ version = "1.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" dependencies = [ - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -64,7 +64,7 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" dependencies = [ "anstyle", "once_cell_polyfill", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -131,24 +131,72 @@ dependencies = [ "tracing", ] +[[package]] +name = "base64" +version = "0.22.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" + [[package]] name = "bitflags" version = "2.13.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b588b76d00fde79687d7646a9b5bdf3cc0f655e0bbd080335a95d7e96f3587da" +[[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 = "bumpalo" +version = "3.20.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649" + [[package]] name = "bytes" version = "1.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" +[[package]] +name = "cc" +version = "1.4.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ad534f4357a5264cce5019c989cf66a4f0dc4e0d1b1d15f8aacec0ff7360273" +dependencies = [ + "find-msvc-tools", + "shlex", +] + [[package]] name = "cfg-if" version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" +[[package]] +name = "cfg_aliases" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f079e83a288787bcd14a6aea84cee5c87a67c5a3e660c30f557a3d24761b3527" + +[[package]] +name = "chacha20" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81" +dependencies = [ + "cfg-if", + "cpufeatures 0.3.0", + "rand_core", +] + [[package]] name = "clap" version = "4.6.6" @@ -195,6 +243,40 @@ version = "1.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" +[[package]] +name = "core-foundation" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b2a6cd9ae233e7f62ba4e9353e81a88df7fc8a5987b8d445b4d90c879bd156f6" +dependencies = [ + "core-foundation-sys", + "libc", +] + +[[package]] +name = "core-foundation-sys" +version = "0.8.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" + +[[package]] +name = "cpufeatures" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "59ed5838eebb26a2bb2e58f6d5b5316989ae9d08bab10e0e6d103e656d1b0280" +dependencies = [ + "libc", +] + +[[package]] +name = "cpufeatures" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b2a41393f66f16b0823bb79094d54ac5fbd34ab292ddafb9a0456ac9f87d201" +dependencies = [ + "libc", +] + [[package]] name = "crc32fast" version = "1.5.0" @@ -219,12 +301,43 @@ version = "0.8.22" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "61803da095bee82a81bb1a452ecc25d3b2f1416d1897eb86430c6159ef717c17" +[[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 = "deranged" version = "0.5.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7cd812cc2bc1d69d4764bd80df88b4317eaef9e773c75226407d9bc0876b211c" +[[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.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c6232dd377dcc64799954cbd3a9bb882e9cdc1308ccd87b1c098f1fb2eaf82a8" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + [[package]] name = "errno" version = "0.3.14" @@ -232,7 +345,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -241,6 +354,12 @@ version = "2.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "da7c62ceae207dd37ea5b845da6a0696c799f85e97da1ab5b7910be3c1c80223" +[[package]] +name = "find-msvc-tools" +version = "0.1.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d45db016d36b838f563236e9193d0ee6ce38f3f68b6c94e914b4929c96bbb890" + [[package]] name = "flate2" version = "1.1.9" @@ -305,6 +424,29 @@ dependencies = [ "slab", ] +[[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.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff2abc00be7fca6ebc474524697ae276ad847ad0a6b3faa4bcb027e9a4614ad0" +dependencies = [ + "cfg-if", + "js-sys", + "libc", + "wasi", + "wasm-bindgen", +] + [[package]] name = "getrandom" version = "0.4.3" @@ -312,8 +454,11 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099" dependencies = [ "cfg-if", + "js-sys", "libc", "r-efi", + "rand_core", + "wasm-bindgen", ] [[package]] @@ -327,6 +472,9 @@ dependencies = [ "flate2", "lru", "prometheus", + "reqwest", + "serde_json", + "sha2", "subtle", "tempfile", "tokio", @@ -406,6 +554,23 @@ dependencies = [ "pin-project-lite", "smallvec", "tokio", + "want", +] + +[[package]] +name = "hyper-rustls" +version = "0.27.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "33ca68d021ef39cf6463ab54c1d0f5daf03377b70561305bb89a8f83aab66e0f" +dependencies = [ + "http", + "hyper", + "hyper-util", + "rustls", + "rustls-native-certs", + "tokio", + "tokio-rustls", + "tower-service", ] [[package]] @@ -414,15 +579,133 @@ version = "0.1.20" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "96547c2556ec9d12fb1578c4eaf448b04993e7fb79cbaad930a656880a6bdfa0" dependencies = [ + "base64", "bytes", + "futures-channel", + "futures-util", "http", "http-body", "hyper", + "ipnet", + "libc", + "percent-encoding", "pin-project-lite", + "socket2", "tokio", "tower-service", + "tracing", +] + +[[package]] +name = "icu_collections" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fa68d21081c4a05d5a901a1c62add574c77048b6a1c67be3b50ce0b60d4ca513" +dependencies = [ + "displaydoc", + "potential_utf", + "utf8_iter", + "yoke", + "zerofrom", + "zerovec", +] + +[[package]] +name = "icu_locale_core" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d56e28588da92eee5c3201a6eff33fabdd49b62269c8938d4ff050ce4d900deb" +dependencies = [ + "displaydoc", + "litemap", + "tinystr", + "writeable", + "zerovec", ] +[[package]] +name = "icu_normalizer" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "12f9cf5f235641ed274641dd81c3f28d870e276763d0797aeeab72317b1c646f" +dependencies = [ + "icu_collections", + "icu_normalizer_data", + "icu_properties", + "icu_provider", + "smallvec", + "zerovec", +] + +[[package]] +name = "icu_normalizer_data" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1563da1ed3e0b3bf3d74c9b85917ac9c56464d2f57242270c09c9e752f8021a0" + +[[package]] +name = "icu_properties" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7e7ca276ad3145661a65914e6daf131ca5120cd3dcee8f8f3214b8875184a148" +dependencies = [ + "displaydoc", + "icu_collections", + "icu_locale_core", + "icu_properties_data", + "icu_provider", + "zerotrie", + "zerovec", +] + +[[package]] +name = "icu_properties_data" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e590f038c1464a96894fd6d10127e90a8be4509f56ff7ecef851b15cee0b7caa" + +[[package]] +name = "icu_provider" +version = "2.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d27bbb9d3abbefac45d55f647c9de1d44aafcd1186eb91879afef17c396c3e73" +dependencies = [ + "displaydoc", + "icu_locale_core", + "writeable", + "yoke", + "zerofrom", + "zerotrie", + "zerovec", +] + +[[package]] +name = "idna" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3b0875f23caa03898994f6ddc501886a45c7d3d62d04d2d90788d47be1b1e4de" +dependencies = [ + "idna_adapter", + "smallvec", + "utf8_iter", +] + +[[package]] +name = "idna_adapter" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb68373c0d6620ef8105e855e7745e18b0d00d3bdb07fb532e434244cdb9a714" +dependencies = [ + "icu_normalizer", + "icu_properties", +] + +[[package]] +name = "ipnet" +version = "2.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6a756c3fac73139e83f14c2d742155dd2b78d3ee56597b419a0579b7bdd6dd78" + [[package]] name = "is_terminal_polyfill" version = "1.70.2" @@ -435,6 +718,17 @@ version = "1.0.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" +[[package]] +name = "js-sys" +version = "0.3.104" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0e0c1080212aad755ea003d18543e8768dd432c48819efd73a7bf1e39b7a5a3a" +dependencies = [ + "cfg-if", + "futures-util", + "wasm-bindgen", +] + [[package]] name = "lazy_static" version = "1.5.0" @@ -453,6 +747,12 @@ version = "0.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32a66949e030da00e8c7d4434b251670a91556f4144941d37452769c25d58a53" +[[package]] +name = "litemap" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47d9d19d1d6efa0109d2f65ff4c85cddd50bd572e5a00127ab10987290bcefae" + [[package]] name = "lock_api" version = "0.4.14" @@ -474,6 +774,12 @@ version = "0.18.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5d2f2f9b4ba7e6b24d95e7e899329d35be83bcded72c8540cdd5368932d1d90a" +[[package]] +name = "lru-slab" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "112b39cec0b298b6c1999fee3e31427f74f676e4cb9879ed1a121b43661a4154" + [[package]] name = "matchers" version = "0.2.0" @@ -519,7 +825,7 @@ checksum = "30d65c71f1ce40ab09135ce117d742b9f8a19ff91a41a8b57ed50bc2de59c427" dependencies = [ "libc", "wasi", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -528,7 +834,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -549,6 +855,12 @@ version = "1.70.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" +[[package]] +name = "openssl-probe" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7c87def4c32ab89d880effc9e097653c8da5d6ef28e6b539d313baaacfbafcbe" + [[package]] name = "parking_lot" version = "0.12.5" @@ -584,6 +896,15 @@ version = "0.2.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" +[[package]] +name = "potential_utf" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d83eb9bc6d8e5cf568e7a1101d60ee05e81ed50ea106026f3d18deeb046d7661" +dependencies = [ + "zerovec", +] + [[package]] name = "powerfmt" version = "0.2.0" @@ -613,6 +934,62 @@ dependencies = [ "thiserror", ] +[[package]] +name = "quinn" +version = "0.11.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c1a41e437b6bbd489372cd4971de128e85c855f56c57f283d20ff016cf7c0a8" +dependencies = [ + "bytes", + "cfg_aliases", + "pin-project-lite", + "quinn-proto", + "quinn-udp", + "rustc-hash", + "rustls", + "socket2", + "thiserror", + "tokio", + "tracing", + "web-time", +] + +[[package]] +name = "quinn-proto" +version = "0.11.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "04759210543be93709136e28212294a659ef5001836ff4eab4d663e4529bba83" +dependencies = [ + "bytes", + "getrandom 0.4.3", + "lru-slab", + "rand", + "rand_pcg", + "ring", + "rustc-hash", + "rustls", + "rustls-pki-types", + "slab", + "thiserror", + "tinyvec", + "tracing", + "web-time", +] + +[[package]] +name = "quinn-udp" +version = "0.5.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "35a133f956daabe89a61a685c2649f13d82d5aa4bd5d12d1277e1072a21c0694" +dependencies = [ + "cfg_aliases", + "libc", + "once_cell", + "socket2", + "tracing", + "windows-sys 0.61.2", +] + [[package]] name = "quote" version = "1.0.47" @@ -628,6 +1005,32 @@ version = "6.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" +[[package]] +name = "rand" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80" +dependencies = [ + "chacha20", + "getrandom 0.4.3", + "rand_core", +] + +[[package]] +name = "rand_core" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69" + +[[package]] +name = "rand_pcg" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "caa0f4137e1c0a72f4c651489402276c8e8e1cf081f3b0ba156d2cbeef09e86a" +dependencies = [ + "rand_core", +] + [[package]] name = "redox_syscall" version = "0.5.18" @@ -654,6 +1057,64 @@ version = "0.8.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4" +[[package]] +name = "reqwest" +version = "0.12.28" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eddd3ca559203180a307f12d114c268abf583f59b03cb906fd0b3ff8646c1147" +dependencies = [ + "base64", + "bytes", + "futures-core", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-rustls", + "hyper-util", + "js-sys", + "log", + "percent-encoding", + "pin-project-lite", + "quinn", + "rustls", + "rustls-native-certs", + "rustls-pki-types", + "serde", + "serde_json", + "serde_urlencoded", + "sync_wrapper", + "tokio", + "tokio-rustls", + "tower", + "tower-http", + "tower-service", + "url", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", +] + +[[package]] +name = "ring" +version = "0.17.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a4689e6c2294d81e88dc6261c768b63bc4fcdb852be6d1352498b114f61383b7" +dependencies = [ + "cc", + "cfg-if", + "getrandom 0.2.17", + "libc", + "untrusted", + "windows-sys 0.52.0", +] + +[[package]] +name = "rustc-hash" +version = "2.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6b1e7f9a428571be2dc5bc0505c13fb6bf936822b894ec87abf8a08a4e51742d" + [[package]] name = "rustix" version = "1.1.4" @@ -664,21 +1125,106 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys", + "windows-sys 0.61.2", +] + +[[package]] +name = "rustls" +version = "0.23.43" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0283386ce02abc0151e1761d08802dfe86c173b0b494af5cbc086574e453da06" +dependencies = [ + "once_cell", + "ring", + "rustls-pki-types", + "rustls-webpki", + "subtle", + "zeroize", +] + +[[package]] +name = "rustls-native-certs" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dab5152771c58876a2146916e53e35057e1a4dfa2b9df0f0305b07f611fdea4d" +dependencies = [ + "openssl-probe", + "rustls-pki-types", + "schannel", + "security-framework", +] + +[[package]] +name = "rustls-pki-types" +version = "1.15.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2f4925028c7eb5d1fcdaf196971378ed9d2c1c4efc7dc5d011256f76c99c0a96" +dependencies = [ + "web-time", + "zeroize", +] + +[[package]] +name = "rustls-webpki" +version = "0.103.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f3c3cf1d8b1e7d4927e2d154c3fcb02979afb9939629c62cd9048d4f07b60ac2" +dependencies = [ + "ring", + "rustls-pki-types", + "untrusted", ] +[[package]] +name = "rustversion" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf54715a573b99ac80df0bc206da022bcd442c974952c7b9720069370852e21f" + [[package]] name = "ryu" version = "1.0.23" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f" +[[package]] +name = "schannel" +version = "0.1.29" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "91c1b7e4904c873ef0710c1f407dde2e6287de2bebc1bbbf7d430bb7cbffd939" +dependencies = [ + "windows-sys 0.61.2", +] + [[package]] name = "scopeguard" version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" +[[package]] +name = "security-framework" +version = "3.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b7f4bc775c73d9a02cde8bf7b2ec4c9d12743edf609006c7facc23998404cd1d" +dependencies = [ + "bitflags", + "core-foundation", + "core-foundation-sys", + "libc", + "security-framework-sys", +] + +[[package]] +name = "security-framework-sys" +version = "2.17.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ce2691df843ecc5d231c0b14ece2acc3efb62c0a398c7e1d875f3983ce020e3" +dependencies = [ + "core-foundation-sys", + "libc", +] + [[package]] name = "serde" version = "1.0.229" @@ -744,6 +1290,17 @@ dependencies = [ "serde", ] +[[package]] +name = "sha2" +version = "0.10.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283" +dependencies = [ + "cfg-if", + "cpufeatures 0.2.17", + "digest", +] + [[package]] name = "sharded-slab" version = "0.1.7" @@ -753,6 +1310,12 @@ dependencies = [ "lazy_static", ] +[[package]] +name = "shlex" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" + [[package]] name = "signal-hook-registry" version = "1.4.8" @@ -788,9 +1351,15 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4" dependencies = [ "libc", - "windows-sys", + "windows-sys 0.61.2", ] +[[package]] +name = "stable_deref_trait" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596" + [[package]] name = "strsim" version = "0.11.1" @@ -836,6 +1405,20 @@ name = "sync_wrapper" version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0bf256ce5efdfa370213c1dabab5935a12e49f2c58d15e9eac2870d3b4f27263" +dependencies = [ + "futures-core", +] + +[[package]] +name = "synstructure" +version = "0.13.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "728a70f3dbaf5bab7f0c4b1ac8d7ae5ea60a4b5549c8a5914361c99147a709d2" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] [[package]] name = "tempfile" @@ -844,10 +1427,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom", + "getrandom 0.4.3", "once_cell", "rustix", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -909,6 +1492,31 @@ dependencies = [ "time-core", ] +[[package]] +name = "tinystr" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b1e27c91459209c2986af3dcf603a5a74a4368754ce37414f59acc971167f643" +dependencies = [ + "displaydoc", + "zerovec", +] + +[[package]] +name = "tinyvec" +version = "1.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bb4ebadaa0af04fab11ae01eb5f9fdb5f9c5b875506e210e71c07873528baa7f" +dependencies = [ + "tinyvec_macros", +] + +[[package]] +name = "tinyvec_macros" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20" + [[package]] name = "tokio" version = "1.53.1" @@ -922,7 +1530,7 @@ dependencies = [ "signal-hook-registry", "socket2", "tokio-macros", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -936,6 +1544,16 @@ dependencies = [ "syn 3.0.3", ] +[[package]] +name = "tokio-rustls" +version = "0.26.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1729aa945f29d91ba541258c8df89027d5792d85a8841fb65e8bf0f4ede4ef61" +dependencies = [ + "rustls", + "tokio", +] + [[package]] name = "tokio-util" version = "0.7.19" @@ -966,6 +1584,24 @@ dependencies = [ "tracing", ] +[[package]] +name = "tower-http" +version = "0.6.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4cfcf7e2740e6fc6d4d688b4ef00650406bb94adf4731e43c096c3a19fe40840" +dependencies = [ + "bitflags", + "bytes", + "futures-util", + "http", + "http-body", + "pin-project-lite", + "tower", + "tower-layer", + "tower-service", + "url", +] + [[package]] name = "tower-layer" version = "0.3.3" @@ -1066,12 +1702,48 @@ dependencies = [ "tracing-serde", ] +[[package]] +name = "try-lock" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" + +[[package]] +name = "typenum" +version = "1.20.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" + [[package]] name = "unicode-ident" version = "1.0.24" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" +[[package]] +name = "untrusted" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1" + +[[package]] +name = "url" +version = "2.5.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff67a8a4397373c3ef660812acab3268222035010ab8680ec4215f38ba3d0eed" +dependencies = [ + "form_urlencoded", + "idna", + "percent-encoding", + "serde", +] + +[[package]] +name = "utf8_iter" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be" + [[package]] name = "utf8parse" version = "0.2.2" @@ -1084,18 +1756,117 @@ version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" +[[package]] +name = "version_check" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" + +[[package]] +name = "want" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bfa7760aed19e106de2c7c0b581b509f2f25d3dacaf737cb82ac61bc6d760b0e" +dependencies = [ + "try-lock", +] + [[package]] name = "wasi" version = "0.11.1+wasi-snapshot-preview1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" +[[package]] +name = "wasm-bindgen" +version = "0.2.127" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1b70935747edd64d89de3efa29d73789b806c15798f8e7dca4d8ac356b50ce70" +dependencies = [ + "cfg-if", + "once_cell", + "rustversion", + "wasm-bindgen-macro", + "wasm-bindgen-shared", +] + +[[package]] +name = "wasm-bindgen-futures" +version = "0.4.77" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6b7777d5cc23d0e91404e53ce2d5e8ec7acae3026b16233dba62cd3246457950" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + +[[package]] +name = "wasm-bindgen-macro" +version = "0.2.127" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "77775f8f3f7217702089053b94958f8f54061a3f663417df76e19cbdcca29bc1" +dependencies = [ + "quote", + "wasm-bindgen-macro-support", +] + +[[package]] +name = "wasm-bindgen-macro-support" +version = "0.2.127" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e11d33f857dc2fb11b8bc75aee111aa9cbeb12cd9f25efd3d4c2a3dd4e235284" +dependencies = [ + "bumpalo", + "proc-macro2", + "quote", + "syn 2.0.119", + "wasm-bindgen-shared", +] + +[[package]] +name = "wasm-bindgen-shared" +version = "0.2.127" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7ef64dbcc55df09c7e5a46182d181c2cfa3e925f3da937ea764728b4bbb9dcbf" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "web-sys" +version = "0.3.104" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c435338968042f4f59a557f690a253676d47ce13ceb55d70100e7facf6620a30" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + +[[package]] +name = "web-time" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a6580f308b1fad9207618087a65c04e7a10bc77e02c8e84e9b00dd4b12fa0bb" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + [[package]] name = "windows-link" version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" +[[package]] +name = "windows-sys" +version = "0.52.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "282be5f36a8ce781fad8c8ae18fa3f9beff57ec1b52cb3de0789201425d9a33d" +dependencies = [ + "windows-targets", +] + [[package]] name = "windows-sys" version = "0.61.2" @@ -1105,6 +1876,159 @@ dependencies = [ "windows-link", ] +[[package]] +name = "windows-targets" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973" +dependencies = [ + "windows_aarch64_gnullvm", + "windows_aarch64_msvc", + "windows_i686_gnu", + "windows_i686_gnullvm", + "windows_i686_msvc", + "windows_x86_64_gnu", + "windows_x86_64_gnullvm", + "windows_x86_64_msvc", +] + +[[package]] +name = "windows_aarch64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3" + +[[package]] +name = "windows_aarch64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469" + +[[package]] +name = "windows_i686_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b" + +[[package]] +name = "windows_i686_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66" + +[[package]] +name = "windows_i686_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66" + +[[package]] +name = "windows_x86_64_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78" + +[[package]] +name = "windows_x86_64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d" + +[[package]] +name = "windows_x86_64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" + +[[package]] +name = "writeable" +version = "0.6.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3ad82d2a33cdc9674dc7465672f271e096168fcdbe0f799d9e6db8c5892679dc" + +[[package]] +name = "yoke" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "709fe23a0424b6a435d82152b1bd3fdfb0833487d5fa90d05d42762a9891fef5" +dependencies = [ + "stable_deref_trait", + "yoke-derive", + "zerofrom", +] + +[[package]] +name = "yoke-derive" +version = "0.8.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "de844c262c8848816172cef550288e7dc6c7b7814b4ee56b3e1553f275f1858e" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", + "synstructure", +] + +[[package]] +name = "zerofrom" +version = "0.1.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ec05a11813ea801ff6d75110ad09cd0824ddba17dfe17128ea0d5f68e6c5272" +dependencies = [ + "zerofrom-derive", +] + +[[package]] +name = "zerofrom-derive" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "11532158c46691caf0f2593ea8358fed6bbf68a0315e80aae9bd41fbade684a1" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", + "synstructure", +] + +[[package]] +name = "zeroize" +version = "1.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e13c156562582aa81c60cb29407084cdb54c4164760106ab78e6c5b0858cf64e" + +[[package]] +name = "zerotrie" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4ea269c3bd32f0a32c321907a2ae912ba6f4649bb0fc764a15627e99a7095a3f" +dependencies = [ + "displaydoc", + "yoke", + "zerofrom", +] + +[[package]] +name = "zerovec" +version = "0.11.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bb0464e17806c1d976d5cba29399c7f08e516e279e2ba493f63123b5fca67dd8" +dependencies = [ + "yoke", + "zerofrom", + "zerovec-derive", +] + +[[package]] +name = "zerovec-derive" +version = "0.11.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "34df6fc39dbd26ddc9c10e6a2984476e13acce22e64e4487636ef494369225da" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + [[package]] name = "zmij" version = "1.0.23" diff --git a/Cargo.toml b/Cargo.toml index 6d8c6de..5efb20a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -23,6 +23,14 @@ flate2 = "1" # foldhash / equivalent deps); the std `HashMap` backend is plenty for this. lru = { version = "0.18", default-features = false } prometheus = { version = "0.14", default-features = false } +# LFS objects use an HTTPS API; reqwest fetches them over rustls+ring (no native-tls, +# so the binary stays statically linked). serde_json rewrites the batch JSON; sha2 +# verifies a fetched object against its oid before caching. +reqwest = { version = "0.12", default-features = false, features = [ + "rustls-tls-native-roots", +] } +serde_json = "1" +sha2 = "0.10" subtle = "2" tokio = { version = "1", features = [ "macros", diff --git a/Dockerfile b/Dockerfile index bdcacdc..948e6db 100644 --- a/Dockerfile +++ b/Dockerfile @@ -6,16 +6,18 @@ # dependency of its own. The runtime layer is a bare Alpine that adds only # `git` + CA certs. # -# Why not a true "distroless" (gcr.io/distroless/static) image? The proxy -# delegates all wire-protocol work to the system `git` binary, so the runtime -# MUST contain git. distroless/static has no package manager and no git, so it -# cannot host this design as-is. Alpine is the smallest base that still ships a -# git package. A genuinely distroless (git-free) image only becomes possible if -# the git plumbing moves in-process to a Rust library (gitoxide/libgit2) - see -# the "no external git binary" item on the roadmap. +# Why not a true "distroless" (gcr.io/distroless/static) image? The proxy delegates +# the git wire protocol to the system `git` binary, so the runtime MUST contain git. +# distroless/static has no package manager and no git, so it cannot host this design +# as-is. Alpine is the smallest base that still ships a git package. The LFS HTTPS +# transfer, by contrast, is in-process (reqwest + rustls), so it needs no runtime +# tool - only CA certs. A genuinely git-free image only becomes possible if the git +# plumbing also moves in-process - see the "no external git binary" roadmap item. FROM rust:alpine AS build -RUN apk add --no-cache musl-dev +# musl-dev for the static libc; build-base gives the C toolchain `ring` (rustls' +# crypto provider) compiles its assembly with. +RUN apk add --no-cache build-base WORKDIR /src COPY . . # Default target on rust:alpine is x86_64-unknown-linux-musl (static). diff --git a/README.md b/README.md index 66c7a0b..ccc7755 100644 --- a/README.md +++ b/README.md @@ -69,21 +69,46 @@ anything git-receive-pack -> 403 (read-only) Concurrent clients for the same repo are serialized so a burst triggers a single upstream fetch; a short TTL coalesces repeated requests. +### git-LFS + +LFS objects use a different HTTP API from the git protocol, so they are cached +separately: + +``` +POST /info/lfs/objects/batch + -> forward to upstream, then rewrite each object's download URL back to this + proxy so the object fetch is cached here +GET /info/lfs/objects/ + -> serve from the on-disk cache, or on a miss fetch it from upstream once + (verify sha256 == oid, store), then serve +anything with operation=upload -> 403 (read-only) +``` + +Objects are content-addressed and immutable, so a cached object is never stale and +is shared across every repo that references the same oid; the first fetch across the +fleet pays the WAN cost, the rest are served locally. The upstream LFS transfer is +in-process over HTTPS (reqwest + rustls, no OpenSSL), so the binary stays statically +linked. Cached objects share the `--cache-max-mb` budget and LRU eviction with the +git mirrors. No configuration is needed: LFS is served on the same endpoints the git +client already routes through the proxy. + ## Benchmark -For a fleet of ephemeral clients cloning the same repo, only the first clone pays -the WAN cost; the rest are served from the local mirror. Cloning a 64 MB repo over -an emulated 20 Mbit/s, 60 ms-RTT link: +For a fleet of ephemeral clients fetching the same content, only the first fetch +pays the WAN cost; the rest are served locally. Over an emulated 20 Mbit/s, +60 ms-RTT link - cloning a 64 MB repo, and fetching a 32 MB git-LFS object: -| Scenario | Clone time | WAN bytes | -| ----------------------------- | ---------: | --------: | -| Direct clone (today) | 28.1 s | 64 MB | -| Via proxy, cold (runner 1) | 28.5 s | 64 MB | -| Via proxy, warm (runner 2..N) | 0.6 s | ~0 MB | +| Scenario | Time | WAN bytes | +| --------------------------------- | ------: | --------: | +| Clone, direct (today) | 28.1 s | 64 MB | +| Clone, via proxy cold (runner 1) | 28.5 s | 64 MB | +| Clone, via proxy warm (2..N) | 0.6 s | ~0 MB | +| LFS object, cold (runner 1) | 12.8 s | 32 MB | +| LFS object, warm (2..N) | 0.2 s | ~0 MB | -The first runner sees no penalty and every subsequent runner clones ~47x faster +The first runner sees no penalty and every subsequent runner is served in-region while nothing crosses the WAN; the saving scales with fleet size and link cost. -Reproduce or retune (`TOTAL_MB`, `RATE_MBIT`, `RTT_MS`) with +Reproduce or retune (`TOTAL_MB`, `LFS_MB`, `RATE_MBIT`, `RTT_MS`) with [`bench/run.sh`](./bench/run.sh) - see [`bench/README.md`](./bench/README.md) for the method and its caveats. @@ -151,8 +176,8 @@ Every flag has an environment-variable equivalent. | `--git-binary` | `GITCACHEPROXY_GIT_BINARY` | `git` | Path to git | Endpoints: `/healthz`, `/readyz`, `/metrics` (Prometheus - per-repo request and -upstream counters, cache-size gauges, and `*_duration_seconds` fetch/serve -latency histograms). +upstream counters, cache-size gauges, LFS object hit/miss counters, and +`*_duration_seconds` fetch/serve latency histograms). ## Auth model @@ -218,11 +243,12 @@ explicit before you expose it: ## Deploy The `Dockerfile` builds a statically linked (musl) binary and drops it onto a -minimal Alpine base. Because all wire-protocol work is delegated to the system -`git` binary, the runtime image must contain `git` - so it is Alpine-with-git -rather than a fully distroless/`FROM scratch` image. Removing that dependency -(and enabling a git-free image) means moving the git plumbing in-process to a -Rust library - see the roadmap below. +minimal Alpine base. Because the git wire protocol is delegated to the system `git` +binary, the runtime image must contain `git` - so it is Alpine-with-git rather than a +fully distroless/`FROM scratch` image. (The LFS transfer is in-process, so it adds no +runtime tool - only CA certs.) Removing the git dependency and enabling a git-free +image means moving the git plumbing in-process to a Rust library - see the roadmap +below. On Kubernetes, a Helm chart lives in [`chart/`](./chart) (single-writer Deployment, `/healthz`+`/readyz` probes, cache PVC, optional Ingress and @@ -240,7 +266,9 @@ See the [chart README](./chart/README.md) for the full values reference. Working and end-to-end tested against both Git wire protocol versions — the modern **v2** (`git-protocol` header, the default since Git 2.26) and the legacy **v0/v1** advertisement — covering full clone, incremental delta fetch, and -push rejection. +push rejection. **git-LFS** is cached too: the batch API is proxied and objects are +stored content-addressed on disk, covered by the LFS integration tests (miss/fetch/ +verify, cache hit, corrupted-object rejection, upload refusal). This is early, single-maintainer software: no independent review or wide deployment yet. Pin a version and try it against your own setup before you diff --git a/bench/README.md b/bench/README.md index 059adc6..ad4b03d 100644 --- a/bench/README.md +++ b/bench/README.md @@ -1,12 +1,13 @@ # Benchmark -`run.sh` measures what the proxy saves a fleet of ephemeral clients that clone -the same repo over a slow link. Everything runs on localhost, so it needs no -privileges and no Docker - just `git`, `python3`, and `cargo`. +`run.sh` measures what the proxy saves a fleet of ephemeral clients that fetch +the same content over a slow link - both git clones and git-LFS objects. +Everything runs on localhost, so it needs no privileges and no Docker - just +`git`, `curl`, `python3`, and `cargo`. ```sh ./bench/run.sh -# tunables (env): TOTAL_MB=64 CHUNK_MB=8 RATE_MBIT=20 RTT_MS=60 +# tunables (env): TOTAL_MB=64 CHUNK_MB=8 LFS_MB=32 RATE_MBIT=20 RTT_MS=60 ``` ## What it does @@ -20,6 +21,10 @@ privileges and no Docker - just `git`, `python3`, and `cargo`. - **A, direct** - client clones the origin through the WAN. Every runner pays this today. - **B, cold proxy** - client clones from the proxy, which fetches the origin through the WAN once (runner 1). - **C, warm proxy** - client clones from the proxy again; within the fetch TTL it serves from the local mirror, so ~0 bytes cross the WAN (runner 2..N). +4. Then, for git-LFS, points the shim at `lfs_origin.py` (a tiny batch-API + object + server) and fetches one object through the proxy with `curl`: + - **D, cold LFS** - the object is fetched from the origin through the WAN once (runner 1). + - **E, warm LFS** - the object is served from the content-addressed cache, so ~0 bytes cross the WAN (runner 2..N). ## Caveats @@ -33,4 +38,6 @@ saving, not a precise WAN emulator: on the emulation fidelity. - **Same WAN transport throughout** - the direct clone (A) and the proxy's upstream fetch (B) both cross the shim over `git://`; the client-to-proxy hop is local HTTP - that never crosses the shim, so the byte comparison is like-for-like. + that never crosses the shim, so the byte comparison is like-for-like. For LFS, the + batch response advertises the object href through the shim too, so the cold fetch + (D) crosses the WAN and the warm fetch (E) does not. diff --git a/bench/lfs_origin.py b/bench/lfs_origin.py new file mode 100644 index 0000000..76a454b --- /dev/null +++ b/bench/lfs_origin.py @@ -0,0 +1,70 @@ +#!/usr/bin/env python3 +"""A minimal git-LFS origin for the benchmark: batch API plus object storage. + +Serves two endpoints, enough to drive the proxy's LFS cache: + + POST /info/lfs/objects/batch -> a download action per requested object, + with an href of /lfs/. The advertise base is the shim, + so the object download crosses the same emulated WAN as the batch (and is + counted), exactly as a cold client fetch would. + GET /lfs/ -> streams the object bytes. + +Anonymous - the benchmark does not exercise auth. Single fixed object, since the +benchmark measures the cold-vs-warm transfer of one object, not fan-out. +""" + +import argparse +import json +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--port", type=int, required=True) + ap.add_argument("--advertise-base", required=True, help="e.g. http://127.0.0.1:") + ap.add_argument("--object-file", required=True) + args = ap.parse_args() + + with open(args.object_file, "rb") as f: + blob = f.read() + + class Handler(BaseHTTPRequestHandler): + def log_message(self, *_): # keep the benchmark output clean + pass + + def do_POST(self): + if not self.path.endswith("/info/lfs/objects/batch"): + self.send_error(404) + return + n = int(self.headers.get("Content-Length", 0)) + req = json.loads(self.rfile.read(n) or b"{}") + objects = [ + { + "oid": o["oid"], + "size": o["size"], + "actions": {"download": {"href": f"{args.advertise_base}/lfs/{o['oid']}"}}, + } + for o in req.get("objects", []) + ] + body = json.dumps({"transfer": "basic", "objects": objects}).encode() + self.send_response(200) + self.send_header("Content-Type", "application/vnd.git-lfs+json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def do_GET(self): + if "/lfs/" not in self.path: + self.send_error(404) + return + self.send_response(200) + self.send_header("Content-Type", "application/octet-stream") + self.send_header("Content-Length", str(len(blob))) + self.end_headers() + self.wfile.write(blob) + + ThreadingHTTPServer(("127.0.0.1", args.port), Handler).serve_forever() + + +if __name__ == "__main__": + main() diff --git a/bench/run.sh b/bench/run.sh index 9a6f44c..a0a7f81 100755 --- a/bench/run.sh +++ b/bench/run.sh @@ -21,23 +21,27 @@ ROOT="$(pwd)" TOTAL_MB="${TOTAL_MB:-64}" CHUNK_MB="${CHUNK_MB:-8}" +LFS_MB="${LFS_MB:-32}" RATE_MBIT="${RATE_MBIT:-20}" RTT_MS="${RTT_MS:-60}" DAEMON_PORT="${DAEMON_PORT:-9419}" SHIM_PORT="${SHIM_PORT:-9420}" PROXY_PORT="${PROXY_PORT:-8899}" +LFS_ORIGIN_PORT="${LFS_ORIGIN_PORT:-9421}" +PROXY_LFS_PORT="${PROXY_LFS_PORT:-8900}" WORK="$(mktemp -d "${TMPDIR:-/tmp}/git-cache-proxy-bench.XXXXXX")" ORIGIN="$WORK/origin" CACHE="$WORK/cache" +CACHE_LFS="$WORK/cache-lfs" COUNTER="$WORK/counter" BIN="$ROOT/target/release/git-cache-proxy" -DAEMON_PID="" PROXY_PID="" SHIM_PID="" +DAEMON_PID="" PROXY_PID="" SHIM_PID="" LFS_ORIGIN_PID="" PROXY_LFS_PID="" cleanup() { - [ -n "$SHIM_PID" ] && kill "$SHIM_PID" 2>/dev/null || true - [ -n "$PROXY_PID" ] && kill "$PROXY_PID" 2>/dev/null || true - [ -n "$DAEMON_PID" ] && kill "$DAEMON_PID" 2>/dev/null || true + for pid in "$SHIM_PID" "$PROXY_LFS_PID" "$LFS_ORIGIN_PID" "$PROXY_PID" "$DAEMON_PID"; do + [ -n "$pid" ] && kill "$pid" 2>/dev/null || true + done wait 2>/dev/null || true rm -rf "$WORK" } @@ -74,7 +78,7 @@ sleep 1 start_shim() { : > "$COUNTER" python3 "$ROOT/bench/shim.py" --listen-port "$SHIM_PORT" \ - --origin-port "$DAEMON_PORT" --rate-mbit "$RATE_MBIT" --rtt-ms "$RTT_MS" \ + --origin-port "${1:-$DAEMON_PORT}" --rate-mbit "$RATE_MBIT" --rtt-ms "$RTT_MS" \ --counter-file "$COUNTER" & SHIM_PID=$! sleep 1 @@ -121,10 +125,50 @@ stop_shim C_TIME=$(elapsed "$t0" "$t1"); C_MB=$(wan_mb) rm -rf "$WORK/c" +# --- LFS: cold vs warm object fetch through the proxy --------------------- +# The proxy also caches git-LFS objects: the batch API is proxied and the object +# is stored content-addressed. A cold fetch crosses the WAN once; every later +# fetch across the fleet is served locally. Driven with curl against the proxy's +# object endpoint (no git-lfs client needed). +echo ">> creating ${LFS_MB}MB LFS object + origin" +OBJECT="$WORK/object.bin" +dd if=/dev/urandom of="$OBJECT" bs=1048576 count="$LFS_MB" status=none +OID=$(python3 -c "import hashlib,sys; print(hashlib.sha256(open(sys.argv[1],'rb').read()).hexdigest())" "$OBJECT") +OSIZE=$(wc -c < "$OBJECT" | tr -d ' ') + +python3 "$ROOT/bench/lfs_origin.py" --port "$LFS_ORIGIN_PORT" \ + --advertise-base "http://127.0.0.1:$SHIM_PORT" --object-file "$OBJECT" & +LFS_ORIGIN_PID=$! +sleep 1 + +"$BIN" --bind "127.0.0.1:$PROXY_LFS_PORT" --cache-root "$CACHE_LFS" \ + --upstream "http://127.0.0.1:$SHIM_PORT" --fetch-ttl-seconds 3600 >/dev/null 2>&1 & +PROXY_LFS_PID=$! +for _ in $(seq 1 30); do + curl -fsS "http://127.0.0.1:$PROXY_LFS_PORT/readyz" >/dev/null 2>&1 && break || sleep 0.5 +done +LFS_URL="http://127.0.0.1:$PROXY_LFS_PORT/bench.git/info/lfs/objects/$OID?size=$OSIZE" + +# --- D: cold LFS object via proxy (runner #1) ---------------------------- +echo ">> D: cold LFS object via proxy (runner #1)" +start_shim "$LFS_ORIGIN_PORT" +t0=$(now); curl -fsS -o /dev/null "$LFS_URL" ; t1=$(now) +stop_shim +D_TIME=$(elapsed "$t0" "$t1"); D_MB=$(wan_mb) + +# --- E: warm LFS object via proxy (runner #2..N) ------------------------- +echo ">> E: warm LFS object via proxy (runner #2..N)" +start_shim "$LFS_ORIGIN_PORT" +t0=$(now); curl -fsS -o /dev/null "$LFS_URL" ; t1=$(now) +stop_shim +E_TIME=$(elapsed "$t0" "$t1"); E_MB=$(wan_mb) + # --- report -------------------------------------------------------------- echo -echo "repo=${TOTAL_MB}MB WAN=${RATE_MBIT}Mbit/s RTT=${RTT_MS}ms" +echo "repo=${TOTAL_MB}MB lfs-obj=${LFS_MB}MB WAN=${RATE_MBIT}Mbit/s RTT=${RTT_MS}ms" printf '%-28s %10s %12s\n' "scenario" "wall (s)" "WAN (MB)" printf '%-28s %10s %12s\n' "A direct (per runner)" "$A_TIME" "$A_MB" printf '%-28s %10s %12s\n' "B cold proxy (runner 1)" "$B_TIME" "$B_MB" printf '%-28s %10s %12s\n' "C warm proxy (runner 2+)" "$C_TIME" "$C_MB" +printf '%-28s %10s %12s\n' "D cold LFS obj (runner 1)" "$D_TIME" "$D_MB" +printf '%-28s %10s %12s\n' "E warm LFS obj (runner 2+)" "$E_TIME" "$E_MB" diff --git a/deny.toml b/deny.toml index 6e87729..967bd7a 100644 --- a/deny.toml +++ b/deny.toml @@ -19,6 +19,9 @@ allow = [ "MIT", "BSD-3-Clause", "Unicode-3.0", + # Permissive, OSI-approved. Pulled in by the rustls TLS stack used for the LFS + # HTTPS transfer: ring (Apache-2.0 AND ISC), rustls-webpki, and untrusted. + "ISC", ] confidence-threshold = 0.9 diff --git a/src/evict.rs b/src/evict.rs index 6688e8c..9f935b3 100644 --- a/src/evict.rs +++ b/src/evict.rs @@ -59,12 +59,14 @@ impl CacheIndex { /// tail), so recency roughly survives a restart. Blocking, but runs at startup /// before the server binds. pub fn new(cache_root: PathBuf, max_bytes: u64, metrics: Arc) -> Arc { - let mut mirrors = find_mirrors(&cache_root); - mirrors.sort_by_key(|m| m.mtime); // oldest first -> pushed to the LRU tail first + // Both bare mirrors and cached LFS objects share one byte budget and LRU. + let mut entries = find_mirrors(&cache_root); + entries.extend(find_lfs_blobs(&cache_root)); + entries.sort_by_key(|m| m.mtime); // oldest first -> pushed to the LRU tail first // Unbounded: the cap is enforced by byte total, not `LruCache`'s item count. let mut cache: LruCache = LruCache::unbounded(); let mut total = 0u64; - for m in mirrors { + for m in entries { total += m.size; cache.put(m.name, m.size); } @@ -96,6 +98,20 @@ impl CacheIndex { let _ = self.lock().cache.get(name); } + /// Record a cached LFS object with its exact size and wake the background task in + /// case the cache is now over cap. Unlike a mirror, an object is immutable and its + /// size is known at store time, so it needs no deferred measurement - it goes + /// straight into the index at its final size and is promoted to most-recently-used. + pub fn record_blob(&self, key: &str, size: u64) { + { + let mut inner = self.lock(); + let old = inner.cache.put(key.to_string(), size); // inserts and promotes to MRU + inner.total = inner.total - old.unwrap_or(0) + size; + self.set_gauges(&inner); + } + self.work.notify_one(); + } + /// Flag a mirror as changed after a clone/fetch: promote it (it was just used), /// schedule it for measurement, and wake the background task. O(1) bookkeeping /// only - the size walk happens off the request path. @@ -219,18 +235,37 @@ async fn maintain(cache: &GitCache, index: &CacheIndex) { } for (name, dir) in index.take_victims() { - match cache.evict(&name, &dir).await { + // An LFS object is a single immutable file: a plain unlink is enough (an open + // reader keeps the inode via POSIX unlink semantics), so it skips the mirror's + // rename-then-remove dance. A mirror goes through `GitCache::evict`, which + // serializes against an in-flight clone/fetch for that repo. + let result = if is_lfs_blob(&name) { + match tokio::fs::remove_file(&dir).await { + Ok(()) => Ok(()), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()), + Err(e) => Err(anyhow::Error::from(e)), + } + } else { + cache.evict(&name, &dir).await + }; + match result { Ok(()) => { index.metrics.record_eviction(); - tracing::info!(repo = %name, "evicted idle mirror"); + tracing::info!(entry = %name, "evicted idle cache entry"); } // The entry is already out of the index; a failed unlink just leaves an - // untracked dir on disk, which the next startup scan picks back up. - Err(e) => tracing::warn!(repo = %name, error = %e, "evict failed"), + // untracked file/dir on disk, which the next startup scan picks back up. + Err(e) => tracing::warn!(entry = %name, error = %e, "evict failed"), } } } +/// Whether a cache key names an LFS object (under the reserved store dir) rather than +/// a bare mirror. Determines how [`maintain`] removes an evicted entry. +fn is_lfs_blob(name: &str) -> bool { + name.split('/').next() == Some(crate::repo::LFS_OBJECTS_DIR) +} + /// A mirror found on disk during the startup scan. struct Scanned { name: String, @@ -271,8 +306,9 @@ fn find_mirrors(cache_root: &Path) -> Vec { let fname = fname.to_string_lossy(); if fname.ends_with(crate::repo::INCOMING_SUFFIX) || fname.ends_with(crate::repo::EVICTING_SUFFIX) + || fname == crate::repo::LFS_OBJECTS_DIR { - continue; + continue; // the LFS store is scanned separately by `find_lfs_blobs` } stack.push(entry.path()); } @@ -280,6 +316,41 @@ fn find_mirrors(cache_root: &Path) -> Vec { out } +/// Discover every cached LFS object under the reserved store (`/.__lfs__// +/// `, two levels deep). Each object is one immutable file whose size is known +/// from its metadata, so - unlike a mirror - it needs no later re-measurement. The +/// in-flight-download subdir is skipped. Startup-only; steady state uses `record_blob`. +fn find_lfs_blobs(cache_root: &Path) -> Vec { + let mut out = Vec::new(); + let lfs_root = cache_root.join(crate::repo::LFS_OBJECTS_DIR); + let Ok(shards) = std::fs::read_dir(&lfs_root) else { + return out; // no LFS store yet + }; + for shard in shards.flatten() { + let Ok(ft) = shard.file_type() else { continue }; + if !ft.is_dir() || shard.file_name().to_string_lossy() == crate::lfs::INCOMING_DIR { + continue; + } + let shard_name = shard.file_name().to_string_lossy().into_owned(); + let Ok(objects) = std::fs::read_dir(shard.path()) else { + continue; + }; + for object in objects.flatten() { + let Ok(md) = object.metadata() else { continue }; + if !md.is_file() { + continue; + } + let oid = object.file_name().to_string_lossy().into_owned(); + out.push(Scanned { + name: format!("{}/{shard_name}/{oid}", crate::repo::LFS_OBJECTS_DIR), + size: md.len(), + mtime: md.modified().unwrap_or(SystemTime::UNIX_EPOCH), + }); + } + } + out +} + /// A mirror's cache-key name: its path relative to the root, `/`-joined so it /// matches the key `resolve` produces regardless of the platform separator. fn rel_name(root: &Path, dir: &Path) -> String { @@ -487,6 +558,53 @@ mod tests { assert!(!dir.exists()); } + #[test] + fn record_blob_tracks_exact_size_in_place() { + let tmp = tempfile::tempdir().unwrap(); + let idx = CacheIndex::new(tmp.path().to_path_buf(), u64::MAX, Arc::new(Metrics::new())); + let key = format!("{}/ab/oid", crate::repo::LFS_OBJECTS_DIR); + idx.record_blob(&key, 500); + assert_eq!(idx.totals(), (500, 1)); + // Re-storing the same oid updates its size rather than double-counting. + idx.record_blob(&key, 700); + assert_eq!(idx.totals(), (700, 1)); + } + + #[tokio::test] + async fn lfs_blobs_are_scanned_at_startup_and_evicted_over_cap() { + let tmp = tempfile::tempdir().unwrap(); + let root = tmp.path(); + // A cached LFS object on disk under the reserved store, sharded by oid[..2]. + let oid = format!("ab{}", "c".repeat(62)); + let blob = root + .join(crate::repo::LFS_OBJECTS_DIR) + .join("ab") + .join(&oid); + std::fs::create_dir_all(blob.parent().unwrap()).unwrap(); + // A stray file in the in-flight-download dir must be ignored by the scan. + let incoming = root + .join(crate::repo::LFS_OBJECTS_DIR) + .join(crate::lfs::INCOMING_DIR); + std::fs::create_dir_all(&incoming).unwrap(); + write_file(&incoming.join("half"), &[b'x'; 10], None); + write_file(&blob, &vec![b'x'; 4096], None); + + let metrics = Arc::new(Metrics::new()); + // Cap below the blob: the startup scan tracks it, then it is over cap. + let idx = CacheIndex::new(root.to_path_buf(), 1000, metrics.clone()); + assert_eq!( + idx.totals(), + (4096, 1), + "only the blob is tracked, not the in-flight file" + ); + + let cache = GitCache::new(dummy_cfg(), metrics.clone(), Some(idx.clone())); + maintain(&cache, &idx).await; // over cap -> the blob file is unlinked + assert!(!blob.exists(), "over-cap LFS blob should be evicted"); + assert_eq!(idx.totals(), (0, 0)); + assert!(metrics.gather().contains("gitcacheproxy_evictions_total 1")); + } + fn dummy_cfg() -> GitConfig { // Eviction never shells out to git, so the binary is irrelevant. GitConfig { diff --git a/src/lfs.rs b/src/lfs.rs new file mode 100644 index 0000000..7ba9333 --- /dev/null +++ b/src/lfs.rs @@ -0,0 +1,553 @@ +// SPDX-License-Identifier: Apache-2.0 +//! git-LFS caching: proxy the batch API to upstream and cache objects on disk. +//! +//! git-cache-proxy serves the git wire protocol (see `git`), but LFS objects use a +//! different HTTP API: a JSON "batch" negotiation that hands back a per-object +//! download URL, then a content-addressed GET of the object itself. Because clones +//! are routed through the proxy, a client's git-lfs derives its LFS endpoint from the +//! proxy URL and talks LFS to us - so we must answer it, or every LFS-tracked file +//! fails to check out. +//! +//! Flow: +//! POST /info/lfs/objects/batch +//! -> forward to upstream, then rewrite every download href to point back here +//! (`/info/lfs/objects/`) so the object fetch is cached. +//! GET /info/lfs/objects/ +//! -> serve from the on-disk cache (content-addressed by oid), or on a miss fetch +//! it once (re-batch for a fresh authorized URL, download, verify sha256 == +//! oid, store) then serve. Objects are immutable, so a cached object is never +//! stale and is shared across every repo that references the same oid. + +use std::collections::HashMap; +use std::fmt::Write as _; +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Duration; + +use anyhow::{Context, Result, bail}; +use reqwest::header::{ACCEPT, CONTENT_TYPE, HeaderMap, HeaderName, HeaderValue}; +use serde_json::Value; +use tokio::io::AsyncWriteExt; +use tokio::sync::Mutex; + +use crate::evict::CacheIndex; +use crate::repo; + +/// The git-LFS batch API content type, sent and expected on the batch endpoint. +const LFS_CONTENT_TYPE: &str = "application/vnd.git-lfs+json"; + +/// Reserved subdir of the LFS store holding in-flight downloads before their atomic +/// rename into place. On the same filesystem as the final object so the rename is +/// atomic; skipped by the eviction scan. +pub const INCOMING_DIR: &str = ".incoming"; + +/// Whether an object request was served from cache or fetched from upstream. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Outcome { + Hit, + Miss, +} + +#[derive(Clone)] +pub struct LfsConfig { + /// Upstream base URL, trailing slash trimmed (same value the git side uses). + pub upstream_base: String, + pub cache_root: PathBuf, + /// Full upstream auth header line (e.g. `Authorization: Basic `), or `None` + /// for anonymous. + pub upstream_auth_header: Option, + /// The proxy's own serve token, if configured; embedded in each rewritten download + /// action so git-lfs re-presents it on the (auth-checked) object GET. + pub serve_token: Option, +} + +pub struct Lfs { + cfg: LfsConfig, + /// One shared client (connection pooling) using rustls+ring - no native-tls, so + /// the binary stays static and free of an OpenSSL dependency. + client: reqwest::Client, + /// Present only when a cache cap is configured; cached objects are recorded here + /// so they share the mirrors' byte budget and LRU eviction. + index: Option>, + /// Per-oid single-flight: collapses a burst of concurrent misses for one object + /// into a single upstream download. Only populated on a miss (a hit returns + /// before locking), so it holds at most one entry per distinct object fetched + /// since startup - each tiny, and never on the hot cached path. + slots: Mutex>>>, + /// Names the per-download temp file the object is streamed to before its atomic + /// rename into the content-addressed cache path. + tmp_counter: AtomicU64, +} + +impl Lfs { + pub fn new(cfg: LfsConfig, index: Option>) -> Self { + let client = reqwest::Client::builder() + .connect_timeout(Duration::from_secs(30)) + .build() + .expect("build reqwest client"); + Self { + cfg, + client, + index, + slots: Mutex::new(HashMap::new()), + tmp_counter: AtomicU64::new(0), + } + } + + /// Proxy a client batch request to upstream and rewrite the download hrefs so the + /// objects are fetched back through this proxy (and thus cached). `advertise_base` + /// is the proxy's own `scheme://host` as the client reached it. Returns the + /// rewritten batch JSON. + pub async fn batch(&self, repo: &str, body: &[u8], advertise_base: &str) -> Result> { + let url = format!("{}/{repo}/info/lfs/objects/batch", self.cfg.upstream_base); + let resp = self + .post_batch(&url, body) + .await + .context("upstream lfs batch")?; + let mut json: Value = serde_json::from_slice(&resp).context("parse lfs batch response")?; + rewrite_download_hrefs( + &mut json, + advertise_base, + repo, + self.cfg.serve_token.as_deref(), + ); + serde_json::to_vec(&json).context("serialize lfs batch response") + } + + /// Ensure the object identified by `oid` is on disk and return its path plus + /// whether it was a cache hit. On a miss `size` is required (the upstream batch + /// API needs it); it is not needed for a hit. + pub async fn ensure_object( + &self, + repo: &str, + oid: &str, + size: Option, + ) -> Result<(PathBuf, Outcome)> { + let path = repo::lfs_object_path(&self.cfg.cache_root, oid); + if self.cached(&path).await { + self.touch(oid); + return Ok((path, Outcome::Hit)); + } + // Serialize concurrent misses for the same oid onto one download. + let slot = self.slot(oid).await; + let _guard = slot.lock().await; + if self.cached(&path).await { + self.touch(oid); // another task fetched it while we waited on the lock + return Ok((path, Outcome::Hit)); + } + let size = size.context("cache miss without an object size")?; + self.fetch_object(repo, oid, size, &path).await?; + Ok((path, Outcome::Miss)) + } + + async fn cached(&self, path: &Path) -> bool { + tokio::fs::try_exists(path).await.unwrap_or(false) + } + + fn touch(&self, oid: &str) { + if let Some(idx) = &self.index { + idx.touch(&repo::lfs_object_key(oid)); + } + } + + /// Download one object from upstream into the cache: re-batch for a fresh, + /// authorized download URL (the batch JWT is short-lived, so it is fetched per + /// download, never cached), stream it to a temp file, verify the content hashes + /// to `oid`, then atomically move it into place. + async fn fetch_object( + &self, + repo: &str, + oid: &str, + size: u64, + final_path: &Path, + ) -> Result<()> { + let (href, headers) = self.download_action(repo, oid, size).await?; + let tmp = self.tmp_path().await?; + if let Err(e) = self.download_to_file(&href, headers, &tmp).await { + let _ = tokio::fs::remove_file(&tmp).await; + return Err(e); + } + let got = sha256_file(tmp.clone()).await?; + if got != oid { + let _ = tokio::fs::remove_file(&tmp).await; + bail!("lfs object {oid} failed integrity check (upstream returned {got})"); + } + if let Some(parent) = final_path.parent() { + tokio::fs::create_dir_all(parent) + .await + .context("create lfs shard dir")?; + } + let bytes = tokio::fs::metadata(&tmp) + .await + .map(|m| m.len()) + .unwrap_or(0); + tokio::fs::rename(&tmp, final_path) + .await + .context("store lfs object")?; + if let Some(idx) = &self.index { + idx.record_blob(&repo::lfs_object_key(oid), bytes); + } + Ok(()) + } + + /// Re-batch upstream for a single object and return its download href plus the + /// headers to send when fetching it (the batch response embeds a short-lived + /// authorization for the object URL). + async fn download_action( + &self, + repo: &str, + oid: &str, + size: u64, + ) -> Result<(String, HeaderMap)> { + let url = format!("{}/{repo}/info/lfs/objects/batch", self.cfg.upstream_base); + let req = serde_json::json!({ + "operation": "download", + "transfers": ["basic"], + "objects": [{ "oid": oid, "size": size }], + }); + let body = serde_json::to_vec(&req).context("build lfs re-batch request")?; + let resp = self + .post_batch(&url, &body) + .await + .context("upstream lfs re-batch")?; + let json: Value = serde_json::from_slice(&resp).context("parse lfs re-batch response")?; + parse_download_action(&json) + } + + /// POST `body` to an upstream LFS batch URL with the LFS content type and (if set) + /// the upstream credential, and return the response bytes. + async fn post_batch(&self, url: &str, body: &[u8]) -> Result> { + let mut req = self + .client + .post(url) + .header(CONTENT_TYPE, LFS_CONTENT_TYPE) + .header(ACCEPT, LFS_CONTENT_TYPE) + .body(body.to_vec()); + if let Some(line) = &self.cfg.upstream_auth_header + && let Some((name, value)) = parse_header_line(line) + { + req = req.header(name, value); + } + let resp = req + .send() + .await + .context("send lfs batch")? + .error_for_status() + .context("lfs batch http status")?; + Ok(resp.bytes().await.context("read lfs batch body")?.to_vec()) + } + + /// Stream an object href to `out`, sending the batch-supplied `headers`. reqwest + /// follows redirects (LFS hrefs commonly redirect to object storage) and drops the + /// auth header on a cross-host hop. + async fn download_to_file(&self, url: &str, headers: HeaderMap, out: &Path) -> Result<()> { + let mut resp = self + .client + .get(url) + .headers(headers) + .send() + .await + .context("send lfs download")? + .error_for_status() + .context("lfs download http status")?; + let mut file = tokio::fs::File::create(out) + .await + .context("create lfs temp file")?; + while let Some(chunk) = resp.chunk().await.context("read lfs object chunk")? { + file.write_all(&chunk) + .await + .context("write lfs object chunk")?; + } + file.flush().await.context("flush lfs object")?; + Ok(()) + } + + async fn tmp_path(&self) -> Result { + let dir = self + .cfg + .cache_root + .join(repo::LFS_OBJECTS_DIR) + .join(INCOMING_DIR); + tokio::fs::create_dir_all(&dir) + .await + .context("create lfs incoming dir")?; + let n = self.tmp_counter.fetch_add(1, Ordering::Relaxed); + Ok(dir.join(format!("{}-{n}", std::process::id()))) + } + + async fn slot(&self, oid: &str) -> Arc> { + self.slots + .lock() + .await + .entry(oid.to_string()) + .or_insert_with(|| Arc::new(Mutex::new(()))) + .clone() + } +} + +/// Rewrite each object's `download` href to point back at this proxy so the fetch is +/// cached, and replace the upstream authorization header with the proxy's serve token +/// (or drop it when the proxy serves anonymously) - the client authenticates to the +/// proxy, not upstream. Objects carrying an `error`, or an `upload` action, are left +/// untouched. Malformed entries are skipped rather than failing the batch. +fn rewrite_download_hrefs( + json: &mut Value, + advertise_base: &str, + repo: &str, + serve_token: Option<&str>, +) { + let Some(objects) = json.get_mut("objects").and_then(Value::as_array_mut) else { + return; + }; + for obj in objects { + let Some(oid) = obj.get("oid").and_then(Value::as_str).map(str::to_string) else { + continue; + }; + let size = obj.get("size").and_then(Value::as_u64).unwrap_or(0); + let Some(download) = obj + .get_mut("actions") + .and_then(|a| a.get_mut("download")) + .and_then(Value::as_object_mut) + else { + continue; + }; + download.insert( + "href".to_string(), + Value::String(format!( + "{advertise_base}/{repo}/info/lfs/objects/{oid}?size={size}" + )), + ); + match serve_token { + Some(token) => { + download.insert( + "header".to_string(), + serde_json::json!({ "Authorization": format!("Bearer {token}") }), + ); + } + None => { + download.remove("header"); + } + } + } +} + +/// Extract the download href and object-transfer headers from a batch response for a +/// single requested object. Errors if upstream reported the object missing or omitted +/// a usable download action. +fn parse_download_action(json: &Value) -> Result<(String, HeaderMap)> { + let obj = json + .get("objects") + .and_then(Value::as_array) + .and_then(|a| a.first()) + .context("lfs batch: no objects in response")?; + if let Some(err) = obj.get("error") { + bail!("lfs batch: upstream object error {err}"); + } + let download = obj + .get("actions") + .and_then(|a| a.get("download")) + .context("lfs batch: no download action")?; + let href = download + .get("href") + .and_then(Value::as_str) + .context("lfs batch: download action has no href")? + .to_string(); + let mut headers = HeaderMap::new(); + if let Some(map) = download.get("header").and_then(Value::as_object) { + for (k, v) in map { + if let (Ok(name), Some(val)) = (HeaderName::from_bytes(k.as_bytes()), v.as_str()) + && let Ok(value) = HeaderValue::from_str(val) + { + headers.insert(name, value); + } + } + } + Ok((href, headers)) +} + +/// Parse a full header line (`Name: value`) into a typed name/value pair, marking the +/// value sensitive so it is redacted from any debug output. `None` if either half is +/// not a valid header token. +fn parse_header_line(line: &str) -> Option<(HeaderName, HeaderValue)> { + let (name, value) = line.split_once(':')?; + let name = HeaderName::from_bytes(name.trim().as_bytes()).ok()?; + let mut value = HeaderValue::from_str(value.trim()).ok()?; + value.set_sensitive(true); + Some((name, value)) +} + +/// Hex-encoded sha256 of a file, computed on the blocking pool (the object can be +/// large, and this runs off the request's async path). +async fn sha256_file(path: PathBuf) -> Result { + tokio::task::spawn_blocking(move || -> Result { + use std::io::Read; + + use sha2::{Digest, Sha256}; + + let mut f = std::fs::File::open(&path).context("open lfs object to hash")?; + let mut hasher = Sha256::new(); + let mut buf = [0u8; 64 * 1024]; + loop { + let n = f.read(&mut buf).context("read lfs object to hash")?; + if n == 0 { + break; + } + hasher.update(&buf[..n]); + } + let mut hex = String::with_capacity(64); + for b in hasher.finalize() { + let _ = write!(hex, "{b:02x}"); + } + Ok(hex) + }) + .await + .context("join sha256 task")? +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn rewrites_download_href_and_strips_auth() { + let mut json = serde_json::json!({ + "transfer": "basic", + "objects": [{ + "oid": "abc123", + "size": 42, + "actions": { + "download": { + "href": "https://upstream.example/storage/abc123", + "header": { "Authorization": "Bearer secret" } + } + } + }], + }); + rewrite_download_hrefs(&mut json, "http://proxy:8080", "g/r.git", None); + let dl = &json["objects"][0]["actions"]["download"]; + assert_eq!( + dl["href"], + "http://proxy:8080/g/r.git/info/lfs/objects/abc123?size=42" + ); + assert!( + dl.get("header").is_none(), + "upstream auth header must be stripped when serving anonymously" + ); + } + + #[test] + fn embeds_serve_token_in_download_header() { + let mut json = serde_json::json!({ + "objects": [{ + "oid": "abc123", + "size": 1, + "actions": { "download": { + "href": "https://upstream/storage/abc123", + "header": { "Authorization": "Bearer upstream-secret" } + }} + }], + }); + rewrite_download_hrefs( + &mut json, + "http://proxy:8080", + "g/r.git", + Some("serve-secret"), + ); + let dl = &json["objects"][0]["actions"]["download"]; + // The upstream credential is replaced by the proxy's serve token so git-lfs + // re-presents it on the (auth-checked) object GET. + assert_eq!(dl["header"]["Authorization"], "Bearer serve-secret"); + } + + #[test] + fn leaves_error_and_upload_objects_untouched() { + let mut json = serde_json::json!({ + "objects": [ + { "oid": "bad", "size": 0, "error": { "code": 404, "message": "missing" } }, + { "oid": "up", "size": 1, "actions": { "upload": { "href": "https://upstream/put" } } }, + { "size": 2, "actions": { "download": { "href": "https://x" } } } // no oid -> skipped + ], + }); + let before = json.clone(); + rewrite_download_hrefs(&mut json, "http://proxy:8080", "g/r.git", None); + assert_eq!(json, before, "no download action -> nothing rewritten"); + } + + #[test] + fn parse_download_action_extracts_href_and_headers() { + let json = serde_json::json!({ + "objects": [{ + "oid": "abc", + "size": 3, + "actions": { "download": { + "href": "https://storage/abc", + "header": { "Authorization": "Bearer jwt", "X-Extra": "1" } + }} + }], + }); + let (href, headers) = parse_download_action(&json).unwrap(); + assert_eq!(href, "https://storage/abc"); + assert_eq!(headers.get("authorization").unwrap(), "Bearer jwt"); + assert_eq!(headers.get("x-extra").unwrap(), "1"); + } + + #[test] + fn parse_download_action_reports_upstream_and_shape_errors() { + // An object upstream flagged as an error propagates as an error. + let err_obj = serde_json::json!({ + "objects": [{ "oid": "x", "size": 0, "error": { "code": 404, "message": "gone" } }] + }); + assert!(parse_download_action(&err_obj).is_err()); + // No download action (e.g. an upload-only response) is an error. + let no_action = serde_json::json!({ "objects": [{ "oid": "x", "size": 0 }] }); + assert!(parse_download_action(&no_action).is_err()); + // A download action without an href is an error. + let no_href = serde_json::json!({ + "objects": [{ "oid": "x", "size": 0, "actions": { "download": {} } }] + }); + assert!(parse_download_action(&no_href).is_err()); + // An empty response has no first object. + assert!(parse_download_action(&serde_json::json!({ "objects": [] })).is_err()); + } + + #[test] + fn parse_download_action_tolerates_a_missing_header_map() { + let json = serde_json::json!({ + "objects": [{ "oid": "x", "size": 0, "actions": { "download": { "href": "https://s/x" } } }] + }); + let (href, headers) = parse_download_action(&json).unwrap(); + assert_eq!(href, "https://s/x"); + assert!(headers.is_empty()); + } + + #[test] + fn parse_header_line_splits_and_trims() { + let (name, value) = parse_header_line("Authorization: Basic abc123").unwrap(); + assert_eq!(name, "authorization"); + assert_eq!(value, "Basic abc123"); + assert!(value.is_sensitive()); + // A line with no colon is not a header. + assert!(parse_header_line("not a header").is_none()); + } + + #[tokio::test] + async fn sha256_matches_known_vectors() { + // The well-known empty and "abc" sha256 digests. + let dir = std::env::temp_dir(); + let empty = dir.join(format!("gcp-lfs-empty-{}", std::process::id())); + tokio::fs::write(&empty, b"").await.unwrap(); + assert_eq!( + sha256_file(empty.clone()).await.unwrap(), + "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" + ); + let abc = dir.join(format!("gcp-lfs-abc-{}", std::process::id())); + tokio::fs::write(&abc, b"abc").await.unwrap(); + assert_eq!( + sha256_file(abc.clone()).await.unwrap(), + "ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad" + ); + let _ = tokio::fs::remove_file(&empty).await; + let _ = tokio::fs::remove_file(&abc).await; + } +} diff --git a/src/lib.rs b/src/lib.rs index cd1b7d7..3a88ce0 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -8,6 +8,7 @@ pub mod config; pub mod evict; pub mod git; +pub mod lfs; pub mod metrics; pub mod repo; pub mod server; diff --git a/src/main.rs b/src/main.rs index 9f4ad80..2e21204 100644 --- a/src/main.rs +++ b/src/main.rs @@ -15,7 +15,7 @@ use clap::Parser; use tracing_subscriber::EnvFilter; use git_cache_proxy::config::{Config, LogFormat}; -use git_cache_proxy::{evict, git, metrics, server}; +use git_cache_proxy::{evict, git, lfs, metrics, repo, server}; #[tokio::main(flavor = "multi_thread")] async fn main() -> Result<()> { @@ -49,7 +49,7 @@ async fn main() -> Result<()> { let git_cfg = git::GitConfig { git_binary: cfg.git_binary.clone(), - upstream_auth_header, + upstream_auth_header: upstream_auth_header.clone(), fetch_ttl: Duration::from_secs(cfg.fetch_ttl_seconds), }; @@ -68,9 +68,32 @@ async fn main() -> Result<()> { }); let cache = Arc::new(git::GitCache::new(git_cfg, metrics.clone(), index.clone())); + + let upstream_base = cfg.upstream.trim_end_matches('/').to_string(); + + // Clear any in-flight LFS downloads left by a crash so they never linger; the + // objects themselves are content-addressed and re-fetched on demand. + let _ = tokio::fs::remove_dir_all( + cfg.cache_root + .join(repo::LFS_OBJECTS_DIR) + .join(lfs::INCOMING_DIR), + ) + .await; + + let lfs = Arc::new(lfs::Lfs::new( + lfs::LfsConfig { + upstream_base: upstream_base.clone(), + cache_root: cfg.cache_root.clone(), + upstream_auth_header, + serve_token: cfg.serve_token.clone(), + }, + index.clone(), + )); + let state = server::AppState { cache: cache.clone(), - upstream_base: cfg.upstream.trim_end_matches('/').to_string(), + lfs, + upstream_base, cache_root: cfg.cache_root.clone(), serve_token: cfg.serve_token.clone(), max_decoded_body: (cfg.max_decoded_body_mb as usize).saturating_mul(1024 * 1024), diff --git a/src/metrics.rs b/src/metrics.rs index a57cb27..be9e5f1 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -25,8 +25,9 @@ const DURATION_BUCKETS: &[f64] = &[ pub struct Metrics { pub registry: Registry, /// `requests_total{kind, result, repo}` - kind = info_refs | upload_pack | - /// auth | receive_pack; result = ok | error | upstream_error | unauthorized | - /// rejected; repo = the served repo path when result = ok, else `-`. + /// auth | receive_pack | lfs_batch | lfs_object; result = ok | error | + /// upstream_error | unauthorized | rejected; repo = the served repo path when + /// result = ok, else `-`. requests: IntCounterVec, /// `upstream_ops_total{op, result, repo}` - op = clone | fetch; result = ok | /// error; repo = the repo path when result = ok, else `-`. @@ -39,6 +40,10 @@ pub struct Metrics { cache_mirrors: IntGauge, /// `evictions_total` - idle mirrors evicted to keep the cache under the cap. evictions: IntCounter, + /// `lfs_objects_total{result}` - cached git-LFS object lookups; result = hit + /// (served from disk) | miss (fetched from upstream, then cached) | error. No + /// `repo` label: objects are content-addressed and shared across repos. + lfs_objects: IntCounterVec, /// `upstream_duration_seconds{op, repo}` - clone/fetch wall-clock, observed /// only on success (same bounded-`repo` discipline as the counters). upstream_duration: HistogramVec, @@ -79,6 +84,14 @@ impl Metrics { "Idle mirrors evicted to keep the cache under the configured cap", ) .expect("valid metric"); + let lfs_objects = IntCounterVec::new( + Opts::new( + "gitcacheproxy_lfs_objects_total", + "Cached git-LFS object lookups (hit/miss/error)", + ), + &["result"], + ) + .expect("valid metric"); let upstream_duration = HistogramVec::new( HistogramOpts::new( "gitcacheproxy_upstream_duration_seconds", @@ -112,6 +125,9 @@ impl Metrics { registry .register(Box::new(evictions.clone())) .expect("register evictions"); + registry + .register(Box::new(lfs_objects.clone())) + .expect("register lfs_objects"); registry .register(Box::new(upstream_duration.clone())) .expect("register upstream_duration"); @@ -125,6 +141,7 @@ impl Metrics { cache_bytes, cache_mirrors, evictions, + lfs_objects, upstream_duration, serve_duration, } @@ -157,6 +174,11 @@ impl Metrics { self.evictions.inc(); } + /// Record a cached LFS object lookup: `result` = `hit` | `miss` | `error`. + pub fn record_lfs(&self, result: &str) { + self.lfs_objects.with_label_values(&[result]).inc(); + } + /// Observe an upstream op's duration. Call only on success with the real repo, /// matching the counters' bounded-`repo` cardinality discipline. pub fn observe_upstream(&self, op: &str, repo: &str, seconds: f64) { @@ -202,6 +224,8 @@ mod tests { m.set_cache_size(2048, 3); m.record_eviction(); m.record_eviction(); + m.record_lfs("hit"); + m.record_lfs("miss"); m.observe_upstream("clone", "group/foo.git", 1.5); m.observe_serve("upload_pack", "group/foo.git", 2.0); @@ -209,6 +233,8 @@ mod tests { assert!(out.contains("gitcacheproxy_cache_bytes 2048")); assert!(out.contains("gitcacheproxy_cache_mirrors 3")); assert!(out.contains("gitcacheproxy_evictions_total 2")); + assert!(out.contains(r#"gitcacheproxy_lfs_objects_total{result="hit"} 1"#)); + assert!(out.contains(r#"gitcacheproxy_lfs_objects_total{result="miss"} 1"#)); assert!(out.contains( r#"gitcacheproxy_upstream_duration_seconds_count{op="clone",repo="group/foo.git"} 1"# )); diff --git a/src/repo.rs b/src/repo.rs index 898e0d4..1c8cc18 100644 --- a/src/repo.rs +++ b/src/repo.rs @@ -39,6 +39,56 @@ pub const INCOMING_SUFFIX: &str = ".__incoming__"; /// mid-eviction. pub const EVICTING_SUFFIX: &str = ".__evicting__"; +/// Reserved top-level directory under the cache root holding cached git-LFS objects +/// (content-addressed by oid, shared across repos - see `lfs_object_path`). Reserved +/// like the suffixes above so a client repo path can never resolve into the LFS store. +pub const LFS_OBJECTS_DIR: &str = ".__lfs__"; + +/// The path marker that identifies a git-LFS endpoint, `/info/lfs/objects/...`. +const LFS_MARKER: &str = "/info/lfs/objects/"; + +/// Repo name for the LFS batch endpoint (`/info/lfs/objects/batch`), or `None` +/// if the path is not a batch request. +pub fn lfs_batch_repo(path: &str) -> Option { + repo_name_from_path(path, &format!("{LFS_MARKER}batch")) +} + +/// Split an LFS object path (`/info/lfs/objects/`) into `(repo, oid)`. +/// Any query string must be stripped by the caller. `None` if the path is not an +/// object request or the oid is malformed. +pub fn lfs_object_from_path(path: &str) -> Option<(String, String)> { + let p = path.trim_start_matches('/'); + let idx = p.find(LFS_MARKER)?; + let repo = &p[..idx]; + let oid = &p[idx + LFS_MARKER.len()..]; + if repo.is_empty() || !valid_lfs_oid(oid) { + return None; + } + Some((repo.to_string(), oid.to_string())) +} + +/// An LFS oid is the lowercase-hex sha256 of the object's content: 64 hex digits. +/// Validated before use as both a filesystem path component and the cache key. +pub fn valid_lfs_oid(oid: &str) -> bool { + oid.len() == 64 + && oid + .bytes() + .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b)) +} + +/// Cache key of an LFS object: its path relative to the cache root, `/`-joined so it +/// matches the keys the eviction index uses for mirrors. Content-addressed and shared +/// across all repos (the oid is the content hash), sharded by the first two hex chars +/// so no single directory holds every object. Caller must pass a `valid_lfs_oid`. +pub fn lfs_object_key(oid: &str) -> String { + format!("{LFS_OBJECTS_DIR}/{}/{oid}", &oid[..2]) +} + +/// On-disk path of a cached LFS object (`cache_root` joined with `lfs_object_key`). +pub fn lfs_object_path(cache_root: &Path, oid: &str) -> PathBuf { + cache_root.join(lfs_object_key(oid)) +} + /// Validate a repo path (no traversal / absolute / NUL) and resolve it against /// the upstream base and cache root. pub fn resolve(name: &str, upstream_base: &str, cache_root: &Path) -> Result { @@ -51,9 +101,13 @@ pub fn resolve(name: &str, upstream_base: &str, cache_root: &Path) -> Result.__incoming__`) or eviction trash (`.__evicting__`) - // of another. + // of another, or the LFS object store (`.__lfs__`). assert!(resolve(&format!("foo{INCOMING_SUFFIX}"), "https://up", root).is_err()); assert!(resolve(&format!("a/b{INCOMING_SUFFIX}"), "https://up", root).is_err()); assert!(resolve(&format!("foo{EVICTING_SUFFIX}"), "https://up", root).is_err()); assert!(resolve(&format!("a/b{EVICTING_SUFFIX}"), "https://up", root).is_err()); + assert!(resolve(LFS_OBJECTS_DIR, "https://up", root).is_err()); + assert!(resolve(&format!("{LFS_OBJECTS_DIR}/ab/cd"), "https://up", root).is_err()); + } + + #[test] + fn parses_lfs_batch_and_object_paths() { + assert_eq!( + lfs_batch_repo("/group/foo.git/info/lfs/objects/batch").as_deref(), + Some("group/foo.git") + ); + assert_eq!(lfs_batch_repo("/group/foo.git/info/refs"), None); + + let oid = "a".repeat(64); + let (repo, got) = + lfs_object_from_path(&format!("/g/r.git/info/lfs/objects/{oid}")).unwrap(); + assert_eq!(repo, "g/r.git"); + assert_eq!(got, oid); + // A non-hex or wrong-length oid is not an object path. + assert!(lfs_object_from_path("/g/r.git/info/lfs/objects/NOTHEX").is_none()); + assert!(lfs_object_from_path("/g/r.git/info/lfs/objects/abc").is_none()); + // The batch endpoint is not an object (batch is not a valid oid). + assert!(lfs_object_from_path("/g/r.git/info/lfs/objects/batch").is_none()); + } + + #[test] + fn validates_oids_and_shards_the_object_path() { + assert!(valid_lfs_oid(&"0".repeat(64))); + assert!(valid_lfs_oid(&format!( + "{}{}", + "a".repeat(32), + "f".repeat(32) + ))); + assert!(!valid_lfs_oid(&"A".repeat(64))); // uppercase is not git-lfs's form + assert!(!valid_lfs_oid(&"a".repeat(63))); + assert!(!valid_lfs_oid(&"g".repeat(64))); // not hex + + let oid = format!("ab{}", "c".repeat(62)); + assert_eq!( + lfs_object_path(Path::new("/cache"), &oid), + Path::new("/cache") + .join(LFS_OBJECTS_DIR) + .join("ab") + .join(&oid) + ); } } diff --git a/src/server.rs b/src/server.rs index c8d743e..433359a 100644 --- a/src/server.rs +++ b/src/server.rs @@ -21,16 +21,20 @@ use subtle::ConstantTimeEq; use tower::limit::GlobalConcurrencyLimitLayer; use crate::git::GitCache; +use crate::lfs::{Lfs, Outcome}; use crate::metrics::Metrics; use crate::repo; const MAX_BODY: usize = 64 * 1024 * 1024; const UPLOAD_PACK: &str = "git-upload-pack"; const RECEIVE_PACK: &str = "git-receive-pack"; +const LFS_CONTENT_TYPE: &str = "application/vnd.git-lfs+json"; +const LFS_BATCH_SUFFIX: &str = "/info/lfs/objects/batch"; #[derive(Clone)] pub struct AppState { pub cache: Arc, + pub lfs: Arc, pub upstream_base: String, pub cache_root: PathBuf, pub serve_token: Option, @@ -168,9 +172,143 @@ async fn handle_git(State(st): State, req: Request) -> Response return upload_pack(st, &path, git_protocol.as_deref(), body).await; } + // git-LFS rides a separate HTTP API alongside the git endpoints. + if parts.method == Method::POST && path.ends_with(LFS_BATCH_SUFFIX) { + let body = match axum::body::to_bytes(body, MAX_BODY).await { + Ok(b) => b, + Err(_) => return err(StatusCode::BAD_REQUEST, "failed to read request body"), + }; + return lfs_batch(st, &path, &parts.headers, body).await; + } + if parts.method == Method::GET + && let Some((repo_name, oid)) = repo::lfs_object_from_path(&path) + { + return lfs_object(st, repo_name, oid, &query).await; + } + err(StatusCode::NOT_FOUND, "not a git smart-http endpoint") } +/// Proxy an LFS batch request to upstream, rewriting each object's download URL back +/// to this proxy so the object fetch is cached here. An `upload` batch is refused: +/// the proxy is read-only, like `git-receive-pack`. +async fn lfs_batch(st: AppState, path: &str, headers: &HeaderMap, body: Bytes) -> Response { + let Some(name) = repo::lfs_batch_repo(path) else { + st.metrics.record_request("lfs_batch", "error", "-"); + return err(StatusCode::NOT_FOUND, "bad lfs batch path"); + }; + if is_lfs_upload(&body) { + st.metrics.record_request("lfs_batch", "rejected", "-"); + return err( + StatusCode::FORBIDDEN, + "read-only proxy: lfs upload is not allowed", + ); + } + let advertise = advertise_base(headers); + match st.lfs.batch(&name, &body, &advertise).await { + Ok(json) => { + st.metrics.record_request("lfs_batch", "ok", &name); + Response::builder() + .header(header::CONTENT_TYPE, LFS_CONTENT_TYPE) + .header(header::CACHE_CONTROL, "no-cache") + .body(Body::from(json)) + .expect("valid response") + } + Err(e) => { + st.metrics + .record_request("lfs_batch", "upstream_error", "-"); + tracing::warn!(repo = %name, error = %e, "lfs batch failed"); + err(StatusCode::BAD_GATEWAY, "upstream lfs batch failed") + } + } +} + +/// Serve a cached LFS object, fetching and caching it from upstream on a miss. The +/// `repo` label is `-` because objects are content-addressed and shared across repos +/// (the cache hit/miss is tracked separately in `lfs_objects_total`). +async fn lfs_object(st: AppState, repo_name: String, oid: String, query: &str) -> Response { + let size = size_from_query(query); + match st.lfs.ensure_object(&repo_name, &oid, size).await { + Ok((path, outcome)) => { + st.metrics.record_lfs(match outcome { + Outcome::Hit => "hit", + Outcome::Miss => "miss", + }); + match lfs_file_response(&path).await { + Ok(resp) => { + st.metrics.record_request("lfs_object", "ok", "-"); + resp + } + Err(e) => { + st.metrics.record_request("lfs_object", "error", "-"); + tracing::warn!(oid = %oid, error = %e, "serve cached lfs object failed"); + err(StatusCode::INTERNAL_SERVER_ERROR, "serve lfs object failed") + } + } + } + Err(e) => { + st.metrics.record_lfs("error"); + st.metrics + .record_request("lfs_object", "upstream_error", "-"); + tracing::warn!(oid = %oid, error = %e, "lfs object fetch failed"); + err(StatusCode::BAD_GATEWAY, "upstream lfs object fetch failed") + } + } +} + +/// Stream a cached LFS object file to the client with its length. +async fn lfs_file_response(path: &Path) -> std::io::Result { + let file = tokio::fs::File::open(path).await?; + let len = file.metadata().await?.len(); + let stream = tokio_util::io::ReaderStream::new(file); + Ok(Response::builder() + .header(header::CONTENT_TYPE, "application/octet-stream") + .header(header::CONTENT_LENGTH, len) + .body(Body::from_stream(stream)) + .expect("valid response")) +} + +/// Whether an LFS batch request asks to upload (write). Best-effort: an unparsable +/// body is treated as not-upload and forwarded, letting upstream decide. +fn is_lfs_upload(body: &[u8]) -> bool { + let Ok(v) = serde_json::from_slice::(body) else { + return false; + }; + v.get("operation").and_then(serde_json::Value::as_str) == Some("upload") +} + +/// The `size` query parameter of an object request, which the proxy embeds in the +/// download URLs it advertises (the upstream batch API needs it on a cache miss). +fn size_from_query(query: &str) -> Option { + query + .split('&') + .find_map(|kv| kv.strip_prefix("size=")) + .and_then(|v| v.parse().ok()) +} + +/// The proxy's own base URL (`scheme://host`) as the client reached it, used to +/// rewrite LFS object download URLs back to this proxy. Honors `X-Forwarded-Proto` / +/// `X-Forwarded-Host` from a TLS-terminating ingress and otherwise falls back to the +/// request `Host` over plain http, which is what the proxy itself speaks. +fn advertise_base(headers: &HeaderMap) -> String { + let first = |v: &axum::http::HeaderValue| { + v.to_str() + .ok() + .map(|s| s.split(',').next().unwrap_or(s).trim().to_string()) + }; + let scheme = headers + .get("x-forwarded-proto") + .and_then(first) + .filter(|s| !s.is_empty()) + .unwrap_or_else(|| "http".to_string()); + let host = headers + .get("x-forwarded-host") + .or_else(|| headers.get(header::HOST)) + .and_then(first) + .unwrap_or_default(); + format!("{scheme}://{host}") +} + async fn info_refs(st: AppState, path: &str, git_protocol: Option<&str>) -> Response { let Some(name) = repo::repo_name_from_path(path, "/info/refs") else { st.metrics.record_request("info_refs", "error", "-"); @@ -411,4 +549,39 @@ mod tests { assert!(!token_matches("", "s3cret")); assert!(token_matches("", "")); // degenerate empty token } + + #[test] + fn advertise_base_uses_forwarded_headers_then_host() { + use axum::http::HeaderValue; + + let mut h = HeaderMap::new(); + h.insert(header::HOST, HeaderValue::from_static("svc.local:8080")); + assert_eq!(advertise_base(&h), "http://svc.local:8080"); + // A TLS-terminating ingress advertises via X-Forwarded-*. + h.insert("x-forwarded-proto", HeaderValue::from_static("https")); + h.insert( + "x-forwarded-host", + HeaderValue::from_static("proxy.example"), + ); + assert_eq!(advertise_base(&h), "https://proxy.example"); + // A comma-listed forwarded chain uses the first hop. + h.insert("x-forwarded-proto", HeaderValue::from_static("https, http")); + assert_eq!(advertise_base(&h), "https://proxy.example"); + } + + #[test] + fn size_from_query_parses_only_a_valid_size() { + assert_eq!(size_from_query("size=42"), Some(42)); + assert_eq!(size_from_query("a=1&size=7&b=2"), Some(7)); + assert_eq!(size_from_query(""), None); + assert_eq!(size_from_query("size=notanumber"), None); + } + + #[test] + fn is_lfs_upload_detects_the_operation() { + assert!(is_lfs_upload(br#"{"operation":"upload","objects":[]}"#)); + assert!(!is_lfs_upload(br#"{"operation":"download"}"#)); + assert!(!is_lfs_upload(b"not json")); // unparsable -> forwarded, not upload + assert!(!is_lfs_upload(b"{}")); + } } diff --git a/tests/e2e.rs b/tests/e2e.rs index 81ee038..f17235f 100644 --- a/tests/e2e.rs +++ b/tests/e2e.rs @@ -17,6 +17,7 @@ use axum::body::Body; use axum::http::{Request, StatusCode}; use git_cache_proxy::evict::CacheIndex; use git_cache_proxy::git::{GitCache, GitConfig}; +use git_cache_proxy::lfs::{Lfs, LfsConfig}; use git_cache_proxy::metrics::Metrics; use git_cache_proxy::server::{AppState, router}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; @@ -89,6 +90,15 @@ async fn upload_pack_decodes_gzip_encoded_request() { }; let state = AppState { cache: Arc::new(GitCache::new(cfg, metrics.clone(), None)), + lfs: Arc::new(Lfs::new( + LfsConfig { + upstream_base: format!("file://{}", up.path().display()), + cache_root: cache.path().to_path_buf(), + upstream_auth_header: None, + serve_token: None, + }, + None, + )), upstream_base: format!("file://{}", up.path().display()), cache_root: cache.path().to_path_buf(), serve_token: None, @@ -174,6 +184,15 @@ async fn clones_through_proxy_serves_all_refs_and_rejects_push() { let idx = CacheIndex::new(cache.path().to_path_buf(), u64::MAX, metrics.clone()); let state = AppState { cache: Arc::new(GitCache::new(cfg, metrics.clone(), Some(idx.clone()))), + lfs: Arc::new(Lfs::new( + LfsConfig { + upstream_base: format!("file://{}", up.path().display()), + cache_root: cache.path().to_path_buf(), + upstream_auth_header: None, + serve_token: None, + }, + Some(idx.clone()), + )), upstream_base: format!("file://{}", up.path().display()), cache_root: cache.path().to_path_buf(), serve_token: None, diff --git a/tests/http.rs b/tests/http.rs index 23129d8..98f1a6a 100644 --- a/tests/http.rs +++ b/tests/http.rs @@ -11,6 +11,7 @@ use std::time::Duration; use axum::body::Body; use axum::http::{Request, StatusCode, header}; use git_cache_proxy::git::{GitCache, GitConfig}; +use git_cache_proxy::lfs::{Lfs, LfsConfig}; use git_cache_proxy::metrics::Metrics; use git_cache_proxy::server::{AppState, router}; use tower::ServiceExt; // for `oneshot` @@ -25,8 +26,18 @@ fn state(serve_token: Option) -> AppState { upstream_auth_header: None, fetch_ttl: Duration::from_secs(10), }; + let lfs = Arc::new(Lfs::new( + LfsConfig { + upstream_base: "https://upstream.invalid".into(), + cache_root: cache_root.clone(), + upstream_auth_header: None, + serve_token: None, + }, + None, + )); AppState { cache: Arc::new(GitCache::new(cfg, metrics.clone(), None)), + lfs, upstream_base: "https://upstream.invalid".into(), cache_root, serve_token, @@ -202,8 +213,18 @@ async fn upstream_failure_returns_bad_gateway_and_records_error() { upstream_auth_header: None, fetch_ttl: Duration::from_secs(10), }; + let lfs = Arc::new(Lfs::new( + LfsConfig { + upstream_base: "file:///nonexistent/git-cache-proxy-upstream".into(), + cache_root: cache.path().to_path_buf(), + upstream_auth_header: None, + serve_token: None, + }, + None, + )); let st = AppState { cache: Arc::new(GitCache::new(cfg, metrics.clone(), None)), + lfs, upstream_base: "file:///nonexistent/git-cache-proxy-upstream".into(), cache_root: cache.path().to_path_buf(), serve_token: None, diff --git a/tests/lfs.rs b/tests/lfs.rs new file mode 100644 index 0000000..7f9380c --- /dev/null +++ b/tests/lfs.rs @@ -0,0 +1,375 @@ +// SPDX-License-Identifier: Apache-2.0 +//! End-to-end git-LFS tests: a mock upstream LFS server (batch API + object storage) +//! on a live socket, with the proxy driven in-process reaching it over real HTTP. +//! Verifies batch href rewriting, a cache miss (fetch + sha256 verify + store), a +//! cache hit served from disk without touching upstream, and rejection of a +//! corrupted object. + +use std::net::SocketAddr; +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::time::Duration; + +use axum::body::Body; +use axum::extract::State; +use axum::http::{Method, Request, StatusCode, header}; +use axum::response::{IntoResponse, Response}; +use axum::routing::any; +use git_cache_proxy::evict::CacheIndex; +use git_cache_proxy::git::{GitCache, GitConfig}; +use git_cache_proxy::lfs::{Lfs, LfsConfig}; +use git_cache_proxy::metrics::Metrics; +use git_cache_proxy::server::{AppState, router}; +use sha2::{Digest, Sha256}; +use tower::ServiceExt; // for `oneshot` + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn batch_rewrites_download_href_to_the_proxy() { + let (addr, _batches) = spawn_upstream(Mode::Ok).await; + let cache = tempfile::tempdir().unwrap(); + let metrics = Arc::new(Metrics::new()); + let state = proxy_state(addr, cache.path(), metrics); + let oid = oid_of(PAYLOAD); + + let resp = router(state) + .oneshot( + Request::post(format!("/{REPO}/info/lfs/objects/batch")) + .header(header::HOST, "proxy.test:8080") + .header(header::CONTENT_TYPE, "application/vnd.git-lfs+json") + .body(Body::from(batch_body(&oid, PAYLOAD.len()))) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + + let json: serde_json::Value = serde_json::from_slice(&body_bytes(resp).await).unwrap(); + let download = &json["objects"][0]["actions"]["download"]; + assert_eq!( + download["href"], + format!( + "http://proxy.test:8080/{REPO}/info/lfs/objects/{oid}?size={}", + PAYLOAD.len() + ) + ); + assert!( + download.get("header").is_none(), + "the upstream object auth header must not leak to the client" + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn object_miss_fetches_then_hit_serves_from_cache() { + let (addr, batches) = spawn_upstream(Mode::Ok).await; + let cache = tempfile::tempdir().unwrap(); + let metrics = Arc::new(Metrics::new()); + let state = proxy_state(addr, cache.path(), metrics.clone()); + let oid = oid_of(PAYLOAD); + let uri = format!("/{REPO}/info/lfs/objects/{oid}?size={}", PAYLOAD.len()); + + // First GET is a miss: the proxy re-batches upstream, downloads, verifies, caches. + let resp = router(state.clone()) + .oneshot(Request::get(&uri).body(Body::empty()).unwrap()) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + assert_eq!(body_bytes(resp).await, PAYLOAD); + let after_miss = batches.load(Ordering::SeqCst); + assert!(after_miss >= 1, "a miss must re-batch upstream"); + + // The object is now on disk, content-addressed by oid. + assert!( + git_cache_proxy::repo::lfs_object_path(cache.path(), &oid).exists(), + "object should be cached after a miss" + ); + + // Second GET is a hit: served from disk, no further upstream batch. + let resp = router(state) + .oneshot(Request::get(&uri).body(Body::empty()).unwrap()) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + assert_eq!(body_bytes(resp).await, PAYLOAD); + assert_eq!( + batches.load(Ordering::SeqCst), + after_miss, + "a cache hit must not contact upstream" + ); + + let scraped = metrics.gather(); + assert!(scraped.contains(r#"gitcacheproxy_lfs_objects_total{result="miss"} 1"#)); + assert!(scraped.contains(r#"gitcacheproxy_lfs_objects_total{result="hit"} 1"#)); + // The cached object counts toward the shared on-disk cache budget. + assert!( + scraped.contains(&format!("gitcacheproxy_cache_bytes {}", PAYLOAD.len())), + "cached object bytes should be accounted:\n{scraped}" + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn corrupted_object_fails_the_integrity_check() { + // Upstream hands back bytes that do not hash to the requested oid; the proxy must + // reject them (502) and cache nothing. + let (addr, _batches) = spawn_upstream(Mode::Corrupt).await; + let cache = tempfile::tempdir().unwrap(); + let metrics = Arc::new(Metrics::new()); + let state = proxy_state(addr, cache.path(), metrics.clone()); + let oid = oid_of(PAYLOAD); + let uri = format!("/{REPO}/info/lfs/objects/{oid}?size={}", PAYLOAD.len()); + + let resp = router(state) + .oneshot(Request::get(&uri).body(Body::empty()).unwrap()) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::BAD_GATEWAY); + assert!( + !git_cache_proxy::repo::lfs_object_path(cache.path(), &oid).exists(), + "a corrupted object must not be cached" + ); + assert!( + metrics + .gather() + .contains(r#"gitcacheproxy_lfs_objects_total{result="error"} 1"#) + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn object_download_failure_returns_bad_gateway() { + // The batch succeeds but the object GET 404s; the proxy surfaces a 502, records + // the error, and caches nothing (the temp download is cleaned up). + let (addr, _batches) = spawn_upstream(Mode::Missing).await; + let cache = tempfile::tempdir().unwrap(); + let metrics = Arc::new(Metrics::new()); + let state = proxy_state(addr, cache.path(), metrics.clone()); + let oid = oid_of(PAYLOAD); + let uri = format!("/{REPO}/info/lfs/objects/{oid}?size={}", PAYLOAD.len()); + + let resp = router(state) + .oneshot(Request::get(&uri).body(Body::empty()).unwrap()) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::BAD_GATEWAY); + assert!( + !git_cache_proxy::repo::lfs_object_path(cache.path(), &oid).exists(), + "a failed download must not leave a cached object" + ); + assert!( + metrics + .gather() + .contains(r#"gitcacheproxy_lfs_objects_total{result="error"} 1"#) + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn upload_batch_is_rejected() { + // A read-only proxy refuses an upload batch before any upstream contact. + let (addr, batches) = spawn_upstream(Mode::Ok).await; + let cache = tempfile::tempdir().unwrap(); + let metrics = Arc::new(Metrics::new()); + let state = proxy_state(addr, cache.path(), metrics); + + let body = serde_json::to_vec(&serde_json::json!({ + "operation": "upload", + "objects": [{ "oid": oid_of(PAYLOAD), "size": PAYLOAD.len() }], + })) + .unwrap(); + let resp = router(state) + .oneshot( + Request::post(format!("/{REPO}/info/lfs/objects/batch")) + .header(header::HOST, "proxy.test:8080") + .body(Body::from(body)) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::FORBIDDEN); + assert_eq!( + batches.load(Ordering::SeqCst), + 0, + "an upload must be refused before contacting upstream" + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn batch_upstream_failure_returns_bad_gateway() { + // The upstream is not listening, so the batch POST fails and the proxy 502s. + let cache = tempfile::tempdir().unwrap(); + let metrics = Arc::new(Metrics::new()); + let dead: SocketAddr = "127.0.0.1:1".parse().unwrap(); + let state = proxy_state(dead, cache.path(), metrics.clone()); + let resp = router(state) + .oneshot( + Request::post(format!("/{REPO}/info/lfs/objects/batch")) + .header(header::HOST, "proxy.test:8080") + .body(Body::from(batch_body(&oid_of(PAYLOAD), PAYLOAD.len()))) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::BAD_GATEWAY); + assert!(metrics.gather().contains( + r#"gitcacheproxy_requests_total{kind="lfs_batch",repo="-",result="upstream_error"} 1"# + )); +} + +// --- test helpers -------------------------------------------------------- + +const REPO: &str = "group/repo.git"; +const PAYLOAD: &[u8] = b"the quick brown fox jumps over the lazy dog\n"; +/// A header the mock puts in the batch download action; the storage endpoint asserts +/// the proxy forwards it, proving batch-supplied object auth is honored. +const OBJECT_AUTH: &str = "X-Object-Auth: let-me-in"; + +/// How the mock upstream serves object storage, exercising the success and failure +/// paths (a corrupted object fails the integrity check; a missing one fails the GET). +#[derive(Clone, Copy)] +enum Mode { + Ok, + Corrupt, + Missing, +} + +/// Shared state of the mock upstream: how many batch calls it has served (so a test +/// can prove a cache hit did not re-contact upstream) and how it serves objects. +#[derive(Clone)] +struct Upstream { + addr: SocketAddr, + batches: Arc, + mode: Mode, +} + +fn oid_of(bytes: &[u8]) -> String { + let mut h = Sha256::new(); + h.update(bytes); + let mut s = String::new(); + use std::fmt::Write as _; + for b in h.finalize() { + let _ = write!(s, "{b:02x}"); + } + s +} + +/// The whole mock upstream: `POST /info/lfs/objects/batch` returns a download +/// action per requested object pointing at `GET /storage/`, which serves +/// the object bytes (or corrupted bytes) after checking the batch-supplied header. +async fn upstream_handler(State(up): State, req: Request) -> Response { + let (parts, body) = req.into_parts(); + let path = parts.uri.path().to_string(); + + if parts.method == Method::POST && path.ends_with("/info/lfs/objects/batch") { + up.batches.fetch_add(1, Ordering::SeqCst); + let repo = path + .trim_start_matches('/') + .strip_suffix("/info/lfs/objects/batch") + .unwrap_or("") + .to_string(); + let bytes = axum::body::to_bytes(body, usize::MAX).await.unwrap(); + let req_json: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); + let objects: Vec = req_json["objects"] + .as_array() + .unwrap() + .iter() + .map(|o| { + let oid = o["oid"].as_str().unwrap(); + let size = o["size"].as_u64().unwrap(); + let (k, v) = OBJECT_AUTH.split_once(": ").unwrap(); + serde_json::json!({ + "oid": oid, + "size": size, + "actions": { + "download": { + "href": format!("http://{}/{repo}/storage/{oid}", up.addr), + "header": { k: v } + } + } + }) + }) + .collect(); + let resp = serde_json::json!({ "transfer": "basic", "objects": objects }); + return ( + [(header::CONTENT_TYPE, "application/vnd.git-lfs+json")], + serde_json::to_vec(&resp).unwrap(), + ) + .into_response(); + } + + if parts.method == Method::GET && path.contains("/storage/") { + // Prove the proxy forwarded the batch-supplied object authorization header. + let (k, v) = OBJECT_AUTH.split_once(": ").unwrap(); + if parts.headers.get(k).and_then(|h| h.to_str().ok()) != Some(v) { + return (StatusCode::UNAUTHORIZED, "missing object auth").into_response(); + } + return match up.mode { + Mode::Ok => (StatusCode::OK, PAYLOAD.to_vec()).into_response(), + Mode::Corrupt => (StatusCode::OK, b"tampered".to_vec()).into_response(), + Mode::Missing => (StatusCode::NOT_FOUND, "gone").into_response(), + }; + } + + (StatusCode::NOT_FOUND, "mock: unknown path").into_response() +} + +/// Bind the mock upstream on an ephemeral port and start serving; returns its addr +/// and the shared batch counter. +async fn spawn_upstream(mode: Mode) -> (SocketAddr, Arc) { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let batches = Arc::new(AtomicUsize::new(0)); + let state = Upstream { + addr, + batches: batches.clone(), + mode, + }; + let app = axum::Router::new() + .fallback(any(upstream_handler)) + .with_state(state); + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + (addr, batches) +} + +/// A proxy `AppState` whose LFS upstream is the mock at `addr`, cached under `cache`. +fn proxy_state(addr: SocketAddr, cache: &std::path::Path, metrics: Arc) -> AppState { + let upstream = format!("http://{addr}"); + let idx = CacheIndex::new(cache.to_path_buf(), u64::MAX, metrics.clone()); + let cfg = GitConfig { + git_binary: "git".into(), + upstream_auth_header: None, + fetch_ttl: Duration::from_secs(10), + }; + AppState { + cache: Arc::new(GitCache::new(cfg, metrics.clone(), Some(idx.clone()))), + lfs: Arc::new(Lfs::new( + LfsConfig { + upstream_base: upstream.clone(), + cache_root: cache.to_path_buf(), + upstream_auth_header: None, + serve_token: None, + }, + Some(idx), + )), + upstream_base: upstream, + cache_root: cache.to_path_buf(), + serve_token: None, + max_decoded_body: 512 * 1024 * 1024, + max_concurrent: 8, + metrics, + } +} + +fn batch_body(oid: &str, size: usize) -> Vec { + serde_json::to_vec(&serde_json::json!({ + "operation": "download", + "transfers": ["basic"], + "objects": [{ "oid": oid, "size": size }], + })) + .unwrap() +} + +async fn body_bytes(resp: Response) -> Vec { + axum::body::to_bytes(resp.into_body(), usize::MAX) + .await + .unwrap() + .to_vec() +} From 05d2d588513ab3b199f9ef5f9653df479a7fc398 Mon Sep 17 00:00:00 2001 From: Roland Groza Date: Mon, 24 Aug 2026 17:53:08 +0900 Subject: [PATCH 2/3] refactor(metrics): use enums for label values The record_*/observe_* helpers took plain &str for the kind/op/result labels, so a typo compiled and only surfaced as a wrong metric series. Type those arguments with enums (RequestKind, Status, UpstreamOp, ServeKind, LfsResult) that render to the label string, keeping the free-form repo path a &str. Assisted-by: Claude:claude-opus-4-8 --- src/git.rs | 35 ++++++++---- src/metrics.rs | 151 ++++++++++++++++++++++++++++++++++++++++--------- src/server.rs | 66 +++++++++++++-------- 3 files changed, 190 insertions(+), 62 deletions(-) diff --git a/src/git.rs b/src/git.rs index a1985b5..9627f46 100644 --- a/src/git.rs +++ b/src/git.rs @@ -28,7 +28,7 @@ use tokio::process::{ChildStdout, Command}; use tokio::sync::Mutex; use tokio_util::io::ReaderStream; -use crate::metrics::Metrics; +use crate::metrics::{Metrics, ServeKind, Status, UpstreamOp}; use crate::repo::RepoRef; /// What `ensure_fresh` did - for metrics. @@ -196,8 +196,11 @@ impl GitCache { String::from_utf8_lossy(&out.stderr) ); } - self.metrics - .observe_serve("info_refs", &repo.name, started.elapsed().as_secs_f64()); + self.metrics.observe_serve( + ServeKind::InfoRefs, + &repo.name, + started.elapsed().as_secs_f64(), + ); let mut body = pkt_line("# service=git-upload-pack\n"); body.extend_from_slice(b"0000"); // flush-pkt body.extend_from_slice(&out.stdout); @@ -300,15 +303,18 @@ impl GitCache { // `-` not the repo name: a failed clone must not mint a per-repo series // for an arbitrary client-supplied path (see `metrics`). The failing // repo is still named in the returned error, which the caller logs. - self.metrics.record_upstream("clone", "error", "-"); + self.metrics + .record_upstream(UpstreamOp::Clone, Status::Error, "-"); bail!("git clone --mirror failed for {}", repo.name); } let elapsed = started.elapsed().as_secs_f64(); tokio::fs::rename(&tmp, &repo.cache_dir) .await .context("rename mirror into place")?; - self.metrics.record_upstream("clone", "ok", &repo.name); - self.metrics.observe_upstream("clone", &repo.name, elapsed); + self.metrics + .record_upstream(UpstreamOp::Clone, Status::Ok, &repo.name); + self.metrics + .observe_upstream(UpstreamOp::Clone, &repo.name, elapsed); self.mark_changed(repo); Ok(()) } @@ -331,12 +337,17 @@ impl GitCache { .await .context("spawn git fetch")?; if !status.success() { - self.metrics.record_upstream("fetch", "error", "-"); + self.metrics + .record_upstream(UpstreamOp::Fetch, Status::Error, "-"); bail!("git fetch failed for {}", repo.name); } - self.metrics.record_upstream("fetch", "ok", &repo.name); self.metrics - .observe_upstream("fetch", &repo.name, started.elapsed().as_secs_f64()); + .record_upstream(UpstreamOp::Fetch, Status::Ok, &repo.name); + self.metrics.observe_upstream( + UpstreamOp::Fetch, + &repo.name, + started.elapsed().as_secs_f64(), + ); self.mark_changed(repo); Ok(()) } @@ -379,7 +390,11 @@ pub struct TimedReader { impl TimedReader { fn record(&mut self) { if let Some((metrics, started)) = self.recorder.take() { - metrics.observe_serve("upload_pack", &self.repo, started.elapsed().as_secs_f64()); + metrics.observe_serve( + ServeKind::UploadPack, + &self.repo, + started.elapsed().as_secs_f64(), + ); } } } diff --git a/src/metrics.rs b/src/metrics.rs index be9e5f1..618cb27 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -22,6 +22,102 @@ const DURATION_BUCKETS: &[f64] = &[ 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0, 120.0, 300.0, ]; +/// The `kind` label on `requests_total`: which client endpoint served the request. +#[derive(Debug, Clone, Copy)] +pub enum RequestKind { + InfoRefs, + UploadPack, + Auth, + ReceivePack, + LfsBatch, + LfsObject, +} + +/// The `result` label on `requests_total` / `upstream_ops_total`. +#[derive(Debug, Clone, Copy)] +pub enum Status { + Ok, + Error, + UpstreamError, + Unauthorized, + Rejected, +} + +/// The `op` label on `upstream_ops_total` / `upstream_duration_seconds`. +#[derive(Debug, Clone, Copy)] +pub enum UpstreamOp { + Clone, + Fetch, +} + +/// The `kind` label on `serve_duration_seconds`. +#[derive(Debug, Clone, Copy)] +pub enum ServeKind { + InfoRefs, + UploadPack, +} + +/// The `result` label on `lfs_objects_total`. +#[derive(Debug, Clone, Copy)] +pub enum LfsResult { + Hit, + Miss, + Error, +} + +impl RequestKind { + fn as_str(self) -> &'static str { + match self { + Self::InfoRefs => "info_refs", + Self::UploadPack => "upload_pack", + Self::Auth => "auth", + Self::ReceivePack => "receive_pack", + Self::LfsBatch => "lfs_batch", + Self::LfsObject => "lfs_object", + } + } +} + +impl Status { + fn as_str(self) -> &'static str { + match self { + Self::Ok => "ok", + Self::Error => "error", + Self::UpstreamError => "upstream_error", + Self::Unauthorized => "unauthorized", + Self::Rejected => "rejected", + } + } +} + +impl UpstreamOp { + fn as_str(self) -> &'static str { + match self { + Self::Clone => "clone", + Self::Fetch => "fetch", + } + } +} + +impl ServeKind { + fn as_str(self) -> &'static str { + match self { + Self::InfoRefs => "info_refs", + Self::UploadPack => "upload_pack", + } + } +} + +impl LfsResult { + fn as_str(self) -> &'static str { + match self { + Self::Hit => "hit", + Self::Miss => "miss", + Self::Error => "error", + } + } +} + pub struct Metrics { pub registry: Registry, /// `requests_total{kind, result, repo}` - kind = info_refs | upload_pack | @@ -149,18 +245,19 @@ impl Metrics { /// Record the outcome of a client request. `repo` is the served repo path on /// success and `-` on any failure, so unbounded client-supplied paths cannot - /// inflate label cardinality. `kind` is `info_refs`, `upload_pack`, `auth` or - /// `receive_pack`; `result` is `ok`, `error`, `upstream_error`, - /// `unauthorized` or `rejected`. - pub fn record_request(&self, kind: &str, result: &str, repo: &str) { - self.requests.with_label_values(&[kind, result, repo]).inc(); + /// inflate label cardinality. + pub fn record_request(&self, kind: RequestKind, result: Status, repo: &str) { + self.requests + .with_label_values(&[kind.as_str(), result.as_str(), repo]) + .inc(); } - /// Record an upstream clone/fetch. Errors are recorded too (`result = - /// "error"`); pass the repo path on success and `-` on error, matching - /// `record_request`. - pub fn record_upstream(&self, op: &str, result: &str, repo: &str) { - self.upstream.with_label_values(&[op, result, repo]).inc(); + /// Record an upstream clone/fetch. Errors are recorded too (`Status::Error`); + /// pass the repo path on success and `-` on error, matching `record_request`. + pub fn record_upstream(&self, op: UpstreamOp, result: Status, repo: &str) { + self.upstream + .with_label_values(&[op.as_str(), result.as_str(), repo]) + .inc(); } /// Refresh the cache-size gauges from the eviction index. @@ -174,24 +271,24 @@ impl Metrics { self.evictions.inc(); } - /// Record a cached LFS object lookup: `result` = `hit` | `miss` | `error`. - pub fn record_lfs(&self, result: &str) { - self.lfs_objects.with_label_values(&[result]).inc(); + /// Record a cached LFS object lookup (hit / miss / error). + pub fn record_lfs(&self, result: LfsResult) { + self.lfs_objects.with_label_values(&[result.as_str()]).inc(); } /// Observe an upstream op's duration. Call only on success with the real repo, /// matching the counters' bounded-`repo` cardinality discipline. - pub fn observe_upstream(&self, op: &str, repo: &str, seconds: f64) { + pub fn observe_upstream(&self, op: UpstreamOp, repo: &str, seconds: f64) { self.upstream_duration - .with_label_values(&[op, repo]) + .with_label_values(&[op.as_str(), repo]) .observe(seconds); } - /// Observe a client serve duration (`kind` = `info_refs` | `upload_pack`), - /// same cardinality discipline as `observe_upstream`. - pub fn observe_serve(&self, kind: &str, repo: &str, seconds: f64) { + /// Observe a client serve duration, same cardinality discipline as + /// `observe_upstream`. + pub fn observe_serve(&self, kind: ServeKind, repo: &str, seconds: f64) { self.serve_duration - .with_label_values(&[kind, repo]) + .with_label_values(&[kind.as_str(), repo]) .observe(seconds); } @@ -217,17 +314,17 @@ mod tests { #[test] fn gather_renders_recorded_series() { let m = Metrics::new(); - m.record_request("info_refs", "ok", "group/foo.git"); - m.record_request("upload_pack", "error", "group/bar.git"); - m.record_upstream("fetch", "ok", "group/foo.git"); - m.record_upstream("clone", "error", "group/bar.git"); + m.record_request(RequestKind::InfoRefs, Status::Ok, "group/foo.git"); + m.record_request(RequestKind::UploadPack, Status::Error, "group/bar.git"); + m.record_upstream(UpstreamOp::Fetch, Status::Ok, "group/foo.git"); + m.record_upstream(UpstreamOp::Clone, Status::Error, "group/bar.git"); m.set_cache_size(2048, 3); m.record_eviction(); m.record_eviction(); - m.record_lfs("hit"); - m.record_lfs("miss"); - m.observe_upstream("clone", "group/foo.git", 1.5); - m.observe_serve("upload_pack", "group/foo.git", 2.0); + m.record_lfs(LfsResult::Hit); + m.record_lfs(LfsResult::Miss); + m.observe_upstream(UpstreamOp::Clone, "group/foo.git", 1.5); + m.observe_serve(ServeKind::UploadPack, "group/foo.git", 2.0); let out = m.gather(); assert!(out.contains("gitcacheproxy_cache_bytes 2048")); diff --git a/src/server.rs b/src/server.rs index 433359a..3db5dd1 100644 --- a/src/server.rs +++ b/src/server.rs @@ -22,7 +22,7 @@ use tower::limit::GlobalConcurrencyLimitLayer; use crate::git::GitCache; use crate::lfs::{Lfs, Outcome}; -use crate::metrics::Metrics; +use crate::metrics::{LfsResult, Metrics, RequestKind, Status}; use crate::repo; const MAX_BODY: usize = 64 * 1024 * 1024; @@ -125,7 +125,8 @@ async fn handle_git(State(st): State, req: Request) -> Response .map(str::to_string); if let Some(resp) = check_auth(&st, &parts.headers) { - st.metrics.record_request("auth", "unauthorized", "-"); + st.metrics + .record_request(RequestKind::Auth, Status::Unauthorized, "-"); return resp; } @@ -133,7 +134,8 @@ async fn handle_git(State(st): State, req: Request) -> Response if path.ends_with(&format!("/{RECEIVE_PACK}")) || query.contains(&format!("service={RECEIVE_PACK}")) { - st.metrics.record_request("receive_pack", "rejected", "-"); + st.metrics + .record_request(RequestKind::ReceivePack, Status::Rejected, "-"); return err( StatusCode::FORBIDDEN, "read-only proxy: pushes are not allowed", @@ -142,7 +144,8 @@ async fn handle_git(State(st): State, req: Request) -> Response if parts.method == Method::GET && path.ends_with("/info/refs") { if !query.contains(&format!("service={UPLOAD_PACK}")) { - st.metrics.record_request("info_refs", "error", "-"); + st.metrics + .record_request(RequestKind::InfoRefs, Status::Error, "-"); return err( StatusCode::BAD_REQUEST, "only smart-http git-upload-pack is supported", @@ -172,7 +175,7 @@ async fn handle_git(State(st): State, req: Request) -> Response return upload_pack(st, &path, git_protocol.as_deref(), body).await; } - // git-LFS rides a separate HTTP API alongside the git endpoints. + // git-LFS uses a different HTTP API alongside the git endpoints. if parts.method == Method::POST && path.ends_with(LFS_BATCH_SUFFIX) { let body = match axum::body::to_bytes(body, MAX_BODY).await { Ok(b) => b, @@ -194,11 +197,13 @@ async fn handle_git(State(st): State, req: Request) -> Response /// the proxy is read-only, like `git-receive-pack`. async fn lfs_batch(st: AppState, path: &str, headers: &HeaderMap, body: Bytes) -> Response { let Some(name) = repo::lfs_batch_repo(path) else { - st.metrics.record_request("lfs_batch", "error", "-"); + st.metrics + .record_request(RequestKind::LfsBatch, Status::Error, "-"); return err(StatusCode::NOT_FOUND, "bad lfs batch path"); }; if is_lfs_upload(&body) { - st.metrics.record_request("lfs_batch", "rejected", "-"); + st.metrics + .record_request(RequestKind::LfsBatch, Status::Rejected, "-"); return err( StatusCode::FORBIDDEN, "read-only proxy: lfs upload is not allowed", @@ -207,7 +212,8 @@ async fn lfs_batch(st: AppState, path: &str, headers: &HeaderMap, body: Bytes) - let advertise = advertise_base(headers); match st.lfs.batch(&name, &body, &advertise).await { Ok(json) => { - st.metrics.record_request("lfs_batch", "ok", &name); + st.metrics + .record_request(RequestKind::LfsBatch, Status::Ok, &name); Response::builder() .header(header::CONTENT_TYPE, LFS_CONTENT_TYPE) .header(header::CACHE_CONTROL, "no-cache") @@ -216,7 +222,7 @@ async fn lfs_batch(st: AppState, path: &str, headers: &HeaderMap, body: Bytes) - } Err(e) => { st.metrics - .record_request("lfs_batch", "upstream_error", "-"); + .record_request(RequestKind::LfsBatch, Status::UpstreamError, "-"); tracing::warn!(repo = %name, error = %e, "lfs batch failed"); err(StatusCode::BAD_GATEWAY, "upstream lfs batch failed") } @@ -231,25 +237,27 @@ async fn lfs_object(st: AppState, repo_name: String, oid: String, query: &str) - match st.lfs.ensure_object(&repo_name, &oid, size).await { Ok((path, outcome)) => { st.metrics.record_lfs(match outcome { - Outcome::Hit => "hit", - Outcome::Miss => "miss", + Outcome::Hit => LfsResult::Hit, + Outcome::Miss => LfsResult::Miss, }); match lfs_file_response(&path).await { Ok(resp) => { - st.metrics.record_request("lfs_object", "ok", "-"); + st.metrics + .record_request(RequestKind::LfsObject, Status::Ok, "-"); resp } Err(e) => { - st.metrics.record_request("lfs_object", "error", "-"); + st.metrics + .record_request(RequestKind::LfsObject, Status::Error, "-"); tracing::warn!(oid = %oid, error = %e, "serve cached lfs object failed"); err(StatusCode::INTERNAL_SERVER_ERROR, "serve lfs object failed") } } } Err(e) => { - st.metrics.record_lfs("error"); + st.metrics.record_lfs(LfsResult::Error); st.metrics - .record_request("lfs_object", "upstream_error", "-"); + .record_request(RequestKind::LfsObject, Status::UpstreamError, "-"); tracing::warn!(oid = %oid, error = %e, "lfs object fetch failed"); err(StatusCode::BAD_GATEWAY, "upstream lfs object fetch failed") } @@ -311,13 +319,15 @@ fn advertise_base(headers: &HeaderMap) -> String { async fn info_refs(st: AppState, path: &str, git_protocol: Option<&str>) -> Response { let Some(name) = repo::repo_name_from_path(path, "/info/refs") else { - st.metrics.record_request("info_refs", "error", "-"); + st.metrics + .record_request(RequestKind::InfoRefs, Status::Error, "-"); return err(StatusCode::NOT_FOUND, "bad path"); }; let repo = match repo::resolve(&name, &st.upstream_base, &st.cache_root) { Ok(r) => r, Err(e) => { - st.metrics.record_request("info_refs", "error", "-"); + st.metrics + .record_request(RequestKind::InfoRefs, Status::Error, "-"); return err(StatusCode::BAD_REQUEST, &e.to_string()); } }; @@ -328,14 +338,15 @@ async fn info_refs(st: AppState, path: &str, git_protocol: Option<&str>) -> Resp // repo paths cannot inflate label cardinality (see `metrics`). if let Err(e) = st.cache.ensure_fresh(&repo, true).await { st.metrics - .record_request("info_refs", "upstream_error", "-"); + .record_request(RequestKind::InfoRefs, Status::UpstreamError, "-"); tracing::warn!(repo = %name, error = %e, "ensure_fresh failed"); return err(StatusCode::BAD_GATEWAY, "upstream fetch failed"); } match st.cache.advertise_refs(&repo, git_protocol).await { Ok(body) => { - st.metrics.record_request("info_refs", "ok", &name); + st.metrics + .record_request(RequestKind::InfoRefs, Status::Ok, &name); Response::builder() .header( header::CONTENT_TYPE, @@ -346,7 +357,8 @@ async fn info_refs(st: AppState, path: &str, git_protocol: Option<&str>) -> Resp .expect("valid response") } Err(e) => { - st.metrics.record_request("info_refs", "error", "-"); + st.metrics + .record_request(RequestKind::InfoRefs, Status::Error, "-"); tracing::warn!(repo = %name, error = %e, "advertise_refs failed"); err(StatusCode::INTERNAL_SERVER_ERROR, "advertise-refs failed") } @@ -360,13 +372,15 @@ async fn upload_pack( body: Bytes, ) -> Response { let Some(name) = repo::repo_name_from_path(path, &format!("/{UPLOAD_PACK}")) else { - st.metrics.record_request("upload_pack", "error", "-"); + st.metrics + .record_request(RequestKind::UploadPack, Status::Error, "-"); return err(StatusCode::NOT_FOUND, "bad path"); }; let repo = match repo::resolve(&name, &st.upstream_base, &st.cache_root) { Ok(r) => r, Err(e) => { - st.metrics.record_request("upload_pack", "error", "-"); + st.metrics + .record_request(RequestKind::UploadPack, Status::Error, "-"); return err(StatusCode::BAD_REQUEST, &e.to_string()); } }; @@ -375,14 +389,15 @@ async fn upload_pack( // present (a client could POST against a not-yet-cloned repo). if let Err(e) = st.cache.ensure_fresh(&repo, false).await { st.metrics - .record_request("upload_pack", "upstream_error", "-"); + .record_request(RequestKind::UploadPack, Status::UpstreamError, "-"); tracing::warn!(repo = %name, error = %e, "ensure mirror exists failed"); return err(StatusCode::BAD_GATEWAY, "upstream unavailable"); } match st.cache.upload_pack_rpc(&repo, git_protocol, body).await { Ok(stream) => { - st.metrics.record_request("upload_pack", "ok", &name); + st.metrics + .record_request(RequestKind::UploadPack, Status::Ok, &name); Response::builder() .header(header::CONTENT_TYPE, "application/x-git-upload-pack-result") .header(header::CACHE_CONTROL, "no-cache") @@ -390,7 +405,8 @@ async fn upload_pack( .expect("valid response") } Err(e) => { - st.metrics.record_request("upload_pack", "error", "-"); + st.metrics + .record_request(RequestKind::UploadPack, Status::Error, "-"); tracing::warn!(repo = %name, error = %e, "upload_pack_rpc failed"); err(StatusCode::INTERNAL_SERVER_ERROR, "upload-pack failed") } From 623f25397504ab95d7c36d5826da6276ae62343e Mon Sep 17 00:00:00 2001 From: Roland Groza Date: Mon, 24 Aug 2026 18:21:48 +0900 Subject: [PATCH 3/3] ci: enforce 90% patch coverage and no project drop Pin the Codecov patch bar at 90% and require project coverage to never drop (0% tolerance), so a change can only hold or raise the line - matching the floor ci.yml enforces via --fail-under-lines. Assisted-by: Claude:claude-opus-4-8 --- codecov.yml | 12 ++++++++++++ 1 file changed, 12 insertions(+) create mode 100644 codecov.yml diff --git a/codecov.yml b/codecov.yml new file mode 100644 index 0000000..d0506b0 --- /dev/null +++ b/codecov.yml @@ -0,0 +1,12 @@ +# Codecov status gates. Patch coverage must be at least 90%, and project coverage +# must never drop against the base (0% tolerance) - so a change can only hold or +# raise the line. ci.yml enforces a matching floor via `--fail-under-lines`. +coverage: + status: + project: + default: + target: auto + threshold: 0% + patch: + default: + target: 90%