diff --git a/Cargo.lock b/Cargo.lock index 6389ae4d..9f48a09f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -45,6 +45,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2c99f64d1e06488f620f932677e24bc6e2897582980441ae90a671415bd7ec2f" dependencies = [ "cfg-if", + "getrandom 0.2.10", "once_cell", "version_check", ] @@ -175,6 +176,12 @@ version = "1.0.71" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9c7d0618f0e0b7e8ff11427422b64564d5fb0be1940354bfe2e0529b18a9d9b8" +[[package]] +name = "arc-swap" +version = "1.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bddcadddf5e9015d310179a59bb28c4d4b9920ad0f11e8e14dbadf654890c9a6" + [[package]] name = "ascii" version = "1.1.0" @@ -386,6 +393,26 @@ dependencies = [ "cipher", ] +[[package]] +name = "bstr" +version = "1.6.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4c2f7349907b712260e64b0afe2f84692af14a454be26187d9df565c7f69266a" +dependencies = [ + "memchr", + "regex-automata 0.3.2", + "serde", +] + +[[package]] +name = "btoi" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9dd6407f73a9b8b6162d8a2ef999fe6afd7cc15902ebf42c5cd296addf17e0ad" +dependencies = [ + "num-traits", +] + [[package]] name = "bumpalo" version = "3.13.0" @@ -736,6 +763,12 @@ dependencies = [ "zeroize", ] +[[package]] +name = "either" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a39bffec1e2015c5d8a6773cb0cf48d0d758c842398f624c34969071f5499ea7" + [[package]] name = "elliptic-curve" version = "0.12.3" @@ -794,6 +827,15 @@ dependencies = [ "serde_json", ] +[[package]] +name = "faster-hex" +version = "0.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "239f7bfb930f820ab16a9cd95afc26f88264cf6905c960b340a615384aa3338a" +dependencies = [ + "serde", +] + [[package]] name = "fastrand" version = "2.0.0" @@ -940,6 +982,7 @@ version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ebb6549ddc63ba5722acb98c823b0eccb7f8b979407bd2a8fd616f581ae50982" dependencies = [ + "bstr", "serde", "thiserror", ] @@ -969,6 +1012,391 @@ dependencies = [ "url", ] +[[package]] +name = "gix-actor" +version = "0.23.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1969b77b9ee4cc1755c841987ec6f7622aaca95e952bcafb76973ae59d1b8716" +dependencies = [ + "bstr", + "btoi", + "gix-date", + "itoa", + "nom", + "thiserror", +] + +[[package]] +name = "gix-chunk" +version = "0.4.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5b42ea64420f7994000130328f3c7a2038f639120518870436d31b8bde704493" +dependencies = [ + "thiserror", +] + +[[package]] +name = "gix-command" +version = "0.2.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0f28f654184b5f725c5737c7e4f466cbd8f0102ac352d5257eeab19647ee4256" +dependencies = [ + "bstr", +] + +[[package]] +name = "gix-commitgraph" +version = "0.17.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed42baa50075d41c1a0931074ce1a97c5797c7c6fe7591d9f1f2dcd448532c26" +dependencies = [ + "bstr", + "gix-chunk", + "gix-features 0.31.1", + "gix-hash", + "memmap2", + "thiserror", +] + +[[package]] +name = "gix-config-value" +version = "0.12.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6e874f41437441c02991dcea76990b9058fadfc54b02ab4dd06ab2218af43897" +dependencies = [ + "bitflags 2.4.0", + "bstr", + "gix-path", + "libc", + "thiserror", +] + +[[package]] +name = "gix-credentials" +version = "0.16.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "75a75565e0e6e7f80cfa4eb1b05cc448c6846ddd48dcf413a28875fbc11ee9af" +dependencies = [ + "bstr", + "gix-command", + "gix-config-value", + "gix-path", + "gix-prompt", + "gix-sec", + "gix-url", + "thiserror", +] + +[[package]] +name = "gix-date" +version = "0.7.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0a825babda995d788e30d306a49dacd1e93d5f5d33d53c7682d0347cef40333c" +dependencies = [ + "bstr", + "itoa", + "thiserror", + "time", +] + +[[package]] +name = "gix-diff" +version = "0.32.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aaf5d9b9b521b284ebe53ee69eee33341835ec70edc314f36b2100ea81396121" +dependencies = [ + "gix-hash", + "gix-object", + "imara-diff", + "thiserror", +] + +[[package]] +name = "gix-features" +version = "0.31.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "06142d8cff5d17509399b04052b64d2f9b3a311d5cff0b1a32b220f62cd0d595" +dependencies = [ + "crc32fast", + "flate2", + "gix-hash", + "gix-trace", + "libc", + "prodash", + "sha1_smol", + "thiserror", + "walkdir", +] + +[[package]] +name = "gix-features" +version = "0.32.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "882695cccf38da4c3cc7ee687bdb412cf25e37932d7f8f2c306112ea712449f1" +dependencies = [ + "gix-hash", + "gix-trace", + "libc", +] + +[[package]] +name = "gix-fs" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4d5b6e9d34a2c61ea4a02bbca94c409ab6dbbca1348cbb67298cd7fed8758761" +dependencies = [ + "gix-features 0.32.1", +] + +[[package]] +name = "gix-hash" +version = "0.11.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4b422ff2ad9a0628baaad6da468cf05385bf3f5ab495ad5a33cce99b9f41092f" +dependencies = [ + "hex", + "thiserror", +] + +[[package]] +name = "gix-hashtable" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "385f4ce6ecf3692d313ca3aa9bd3b3d8490de53368d6d94bedff3af8b6d9c58d" +dependencies = [ + "gix-hash", + "hashbrown 0.14.0", + "parking_lot", +] + +[[package]] +name = "gix-object" +version = "0.32.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a953f3d7ffad16734aa3ab1d05807972c80e339d1bd9dde03e0198716b99e2a6" +dependencies = [ + "bstr", + "btoi", + "gix-actor", + "gix-date", + "gix-features 0.31.1", + "gix-hash", + "gix-validate", + "hex", + "itoa", + "nom", + "smallvec", + "thiserror", +] + +[[package]] +name = "gix-odb" +version = "0.49.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f6418cff00ecc2713b58c8e04bff30dda808fbba1a080e7248b299d069894a01" +dependencies = [ + "arc-swap", + "gix-date", + "gix-features 0.31.1", + "gix-hash", + "gix-object", + "gix-pack", + "gix-path", + "gix-quote", + "parking_lot", + "tempfile", + "thiserror", +] + +[[package]] +name = "gix-pack" +version = "0.39.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "414935138d90043ea5898de7a93f02c2558e52652492719470e203ef26a8fd0a" +dependencies = [ + "gix-chunk", + "gix-diff", + "gix-features 0.31.1", + "gix-hash", + "gix-hashtable", + "gix-object", + "gix-path", + "gix-tempfile", + "gix-traverse", + "memmap2", + "parking_lot", + "smallvec", + "thiserror", +] + +[[package]] +name = "gix-packetline" +version = "0.16.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2a374cb5eba089e3c123df4d996eb00da411bb90ec92cb35bffeeb2d22ee106a" +dependencies = [ + "bstr", + "faster-hex", + "thiserror", +] + +[[package]] +name = "gix-path" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "18609c8cbec8508ea97c64938c33cd305b75dfc04a78d0c3b78b8b3fd618a77c" +dependencies = [ + "bstr", + "gix-trace", + "home", + "once_cell", + "thiserror", +] + +[[package]] +name = "gix-prompt" +version = "0.5.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2c22decaf4a063ccae2b2108820c8630c01bd6756656df3fe464b32b8958a5ea" +dependencies = [ + "gix-command", + "gix-config-value", + "parking_lot", + "rustix", + "thiserror", +] + +[[package]] +name = "gix-protocol" +version = "0.35.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed7069fac7eb23b043b4bd7890df1e244cb370c3fe8b2ff482203d36b4fd4099" +dependencies = [ + "bstr", + "btoi", + "gix-credentials", + "gix-date", + "gix-features 0.31.1", + "gix-hash", + "gix-transport", + "maybe-async", + "nom", + "thiserror", +] + +[[package]] +name = "gix-quote" +version = "0.4.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "475c86a97dd0127ba4465fbb239abac9ea10e68301470c9791a6dd5351cdc905" +dependencies = [ + "bstr", + "btoi", + "thiserror", +] + +[[package]] +name = "gix-revwalk" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "028d50fcaf8326a8f79a359490d9ca9fb4e2b51ac9ac86503560d0bcc888d2eb" +dependencies = [ + "gix-commitgraph", + "gix-date", + "gix-hash", + "gix-hashtable", + "gix-object", + "smallvec", + "thiserror", +] + +[[package]] +name = "gix-sec" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9615cbd6b456898aeb942cd75e5810c382fbfc48dbbff2fa23ebd2d33dcbe9c7" +dependencies = [ + "bitflags 2.4.0", + "gix-path", + "libc", + "windows", +] + +[[package]] +name = "gix-tempfile" +version = "7.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fa28d567848cec8fdd77d36ad4f5f78ecfaba7d78f647d4f63c8ae1a2cec7243" +dependencies = [ + "gix-fs", + "libc", + "once_cell", + "parking_lot", + "tempfile", +] + +[[package]] +name = "gix-trace" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "96b6d623a1152c3facb79067d6e2ecdae48130030cf27d6eb21109f13bd7b836" + +[[package]] +name = "gix-transport" +version = "0.33.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0929bb80a07c04033edd4585091c4db9ea458cb932e883bf22efb146ebfbdc89" +dependencies = [ + "bstr", + "gix-command", + "gix-features 0.31.1", + "gix-packetline", + "gix-quote", + "gix-sec", + "gix-url", + "thiserror", +] + +[[package]] +name = "gix-traverse" +version = "0.29.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e3cdfd54598db4fae57d5ae6f52958422b2d13382d2745796bfe5c8015ffa86e" +dependencies = [ + "gix-commitgraph", + "gix-date", + "gix-hash", + "gix-hashtable", + "gix-object", + "gix-revwalk", + "smallvec", + "thiserror", +] + +[[package]] +name = "gix-url" +version = "0.20.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "beaede6dbc83f408b19adfd95bb52f1dbf01fb8862c3faf6c6243e2e67fcdfa1" +dependencies = [ + "bstr", + "gix-features 0.31.1", + "gix-path", + "home", + "thiserror", + "url", +] + +[[package]] +name = "gix-validate" +version = "0.7.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba9b3737b2cef3dcd014633485f0034b0f1a931ee54aeb7d8f87f177f3c89040" +dependencies = [ + "bstr", + "thiserror", +] + [[package]] name = "group" version = "0.12.1" @@ -1046,6 +1474,12 @@ version = "0.3.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "443144c8cdadd93ebf52ddb4056d257f5b52c04d3c804e657d19eb73fc33668b" +[[package]] +name = "hex" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7f24254aa9a54b5c858eaee2f5bccdb46aaf0e486a595ed5fd8f86ba55232a70" + [[package]] name = "hmac" version = "0.12.1" @@ -1055,6 +1489,15 @@ dependencies = [ "digest 0.10.7", ] +[[package]] +name = "home" +version = "0.5.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5444c27eef6923071f7ebcc33e3444508466a76f7a2b93da00ed6e19f30c1ddb" +dependencies = [ + "windows-sys", +] + [[package]] name = "http" version = "0.2.9" @@ -1152,6 +1595,16 @@ dependencies = [ "unicode-normalization", ] +[[package]] +name = "imara-diff" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e98c1d0ad70fc91b8b9654b1f33db55e59579d3b3de2bffdced0fdb810570cb8" +dependencies = [ + "ahash", + "hashbrown 0.12.3", +] + [[package]] name = "indexmap" version = "1.9.3" @@ -1337,6 +1790,16 @@ dependencies = [ "serde", ] +[[package]] +name = "lock_api" +version = "0.4.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c1cc9717a20b1bb222f333e6a92fd32f7d8a18ddc5a3191a11af45dcbf4dcd16" +dependencies = [ + "autocfg", + "scopeguard", +] + [[package]] name = "log" version = "0.4.20" @@ -1368,10 +1831,30 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b87248edafb776e59e6ee64a79086f65890d3510f2c656c000bf2a7e8a0aea40" [[package]] -name = "memchr" -version = "2.5.0" +name = "maybe-async" +version = "0.2.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2dffe52ecf27772e601905b7522cb4ef790d2cc203488bbd0e2fe85fcb74566d" +checksum = "0f1b8c13cb1f814b634a96b2c725449fe7ed464a7b8781de8688be5ffbd3f305" +dependencies = [ + "proc-macro2", + "quote", + "syn 1.0.109", +] + +[[package]] +name = "memchr" +version = "2.6.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f232d6ef707e1956a43342693d2a31e72989554d58299d7a88738cc95b0d35c" + +[[package]] +name = "memmap2" +version = "0.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f49388d20533534cd19360ad3d6a7dadc885944aa802ba3995040c5ec11288c6" +dependencies = [ + "libc", +] [[package]] name = "mime" @@ -1379,6 +1862,12 @@ version = "0.3.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" +[[package]] +name = "minimal-lexical" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a" + [[package]] name = "miniz_oxide" version = "0.7.1" @@ -1444,6 +1933,16 @@ dependencies = [ "cyphergraphy", ] +[[package]] +name = "nom" +version = "7.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d273983c5a657a70a3e8f2a01329822f3b8c8172b73826411a55751e404a0a4a" +dependencies = [ + "memchr", + "minimal-lexical", +] + [[package]] name = "nonempty" version = "0.5.0" @@ -1533,6 +2032,15 @@ dependencies = [ "libc", ] +[[package]] +name = "num_threads" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2819ce041d2ee131036f4fc9d6ae7ae125a3a40e97ba64d04fe799ad9dabbb44" +dependencies = [ + "libc", +] + [[package]] name = "numtoa" version = "0.1.0" @@ -1588,6 +2096,29 @@ dependencies = [ "sha2 0.10.7", ] +[[package]] +name = "parking_lot" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3742b2c103b9f06bc9fff0a37ff4912935851bee6d36f3c02bcc755bcfec228f" +dependencies = [ + "lock_api", + "parking_lot_core", +] + +[[package]] +name = "parking_lot_core" +version = "0.9.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93f00c865fe7cabf650081affecd3871070f26767e7b2070a3ffae14c654b447" +dependencies = [ + "cfg-if", + "libc", + "redox_syscall 0.3.5", + "smallvec", + "windows-targets", +] + [[package]] name = "pbkdf2" version = "0.11.0" @@ -1741,6 +2272,12 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "prodash" +version = "25.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d67eb4220992a4a052a4bb03cf776e493ecb1a3a36bab551804153d63486af7" + [[package]] name = "qcheck" version = "1.0.0" @@ -1919,6 +2456,26 @@ dependencies = [ "fastrand", ] +[[package]] +name = "radicle-fetch" +version = "0.1.0" +dependencies = [ + "bstr", + "either", + "gix-actor", + "gix-features 0.31.1", + "gix-hash", + "gix-odb", + "gix-pack", + "gix-protocol", + "gix-transport", + "log", + "nonempty 0.8.1", + "radicle", + "radicle-git-ext", + "thiserror", +] + [[package]] name = "radicle-git-ext" version = "0.6.0" @@ -2290,6 +2847,21 @@ dependencies = [ "cipher", ] +[[package]] +name = "same-file" +version = "1.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93fc1dc3aaa9bfed95e02e6eadabb4baf7e3078b0bd1b4d7b6b0b68378900502" +dependencies = [ + "winapi-util", +] + +[[package]] +name = "scopeguard" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" + [[package]] name = "scrypt" version = "0.10.0" @@ -2393,6 +2965,12 @@ dependencies = [ "digest 0.10.7", ] +[[package]] +name = "sha1_smol" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae1a47186c03a32177042e55dbc5fd5aee900b8e0069a8d70fba96a9375cd012" + [[package]] name = "sha2" version = "0.9.9" @@ -2728,6 +3306,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "59e399c068f43a5d116fedaf73b203fa4f9c519f17e2b34f63221d3792f81446" dependencies = [ "itoa", + "libc", + "num_threads", "serde", "time-core", "time-macros", @@ -3169,6 +3749,16 @@ version = "0.9.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "49874b5167b65d7193b8aba1567f5c7d93d001cafc34600cee003eda787e483f" +[[package]] +name = "walkdir" +version = "2.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d71d857dc86794ca4c280d616f7da00d2dbfd8cd788846559a6813e6aa4b54ee" +dependencies = [ + "same-file", + "winapi-util", +] + [[package]] name = "want" version = "0.3.1" @@ -3260,6 +3850,15 @@ version = "0.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ac3b87c63620426dd9b991e5ce0329eff545bccbbb34f3be09ff6fb6ab51b7b6" +[[package]] +name = "winapi-util" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "70ec6ce85bb158151cae5e5c87f95a8e97d2c0c4b001223f33a334e3ce5de178" +dependencies = [ + "winapi", +] + [[package]] name = "winapi-x86_64-pc-windows-gnu" version = "0.4.0" diff --git a/Cargo.toml b/Cargo.toml index 6814b72b..70b95855 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -7,6 +7,7 @@ members = [ "radicle-crdt", "radicle-crypto", "radicle-dag", + "radicle-fetch", "radicle-httpd", "radicle-node", "radicle-remote-helper", diff --git a/radicle-fetch/Cargo.toml b/radicle-fetch/Cargo.toml new file mode 100644 index 00000000..df9bc052 --- /dev/null +++ b/radicle-fetch/Cargo.toml @@ -0,0 +1,24 @@ +[package] +name = "radicle-fetch" +license = "MIT OR Apache-2.0" +version = "0.1.0" +authors = ["Fintan Halpenny "] +edition = "2021" + +[dependencies] +bstr = { version = "1.3" } +either = { version = "0" } +gix-actor = { version = "0.23.0" } +gix-features = { version = "0.31", features = ["progress"] } +gix-hash = { version = "0.11" } +gix-odb = { version = "0.49" } +gix-pack = { version = "0.39" } +gix-protocol = { version = "0.35", features = ["blocking-client"] } +gix-transport = { version = "0.33", features = ["blocking-client"] } +log = { version = "0.4.17", features = ["std"] } +nonempty = { version = "0.8.1" } +radicle-git-ext = { version = "0.6.0", features = ["bstr"] } +thiserror = { version = "1" } + +[dependencies.radicle] +path = "../radicle" diff --git a/radicle-fetch/src/git.rs b/radicle-fetch/src/git.rs new file mode 100644 index 00000000..b84df9a7 --- /dev/null +++ b/radicle-fetch/src/git.rs @@ -0,0 +1,22 @@ +pub(crate) mod mem; +pub(crate) mod repository; + +pub mod refs; + +pub(crate) mod oid { + //! Helper functions for converting to/from [`radicle::git::Oid`] and + //! [`ObjectId`]. + + use gix_hash::ObjectId; + use radicle::git::Oid; + + /// Convert from an [`ObjectId`] to an [`Oid`]. + pub fn to_oid(oid: ObjectId) -> Oid { + Oid::try_from(oid.as_bytes()).expect("invalid gix Oid") + } + + /// Convert from an [`Oid`] to an [`ObjectId`]. + pub fn to_object_id(oid: Oid) -> ObjectId { + ObjectId::try_from(oid.as_bytes()).expect("invalid git-ext Oid") + } +} diff --git a/radicle-fetch/src/git/mem.rs b/radicle-fetch/src/git/mem.rs new file mode 100644 index 00000000..df53dcb0 --- /dev/null +++ b/radicle-fetch/src/git/mem.rs @@ -0,0 +1,78 @@ +use std::collections::HashMap; + +use radicle::git::{Component, Oid, Qualified, RefString}; +use radicle::prelude::PublicKey; + +use super::refs::{Applied, RefUpdate, Update}; + +/// An in-memory reference store. +/// +/// It provides the same functionality as the [`super::Refdb`], but is +/// used to temporarily store reference names and objects. +#[derive(Clone, Debug, Default)] +pub struct Refdb(HashMap, Oid>); + +impl Refdb { + pub fn refname_to_id<'a, N>(&self, refname: N) -> Option + where + N: Into>, + { + let name = refname.into(); + self.0.get(&name).copied() + } + + pub fn references_of<'a>( + &'a self, + remote: &'a PublicKey, + ) -> impl Iterator + 'a { + self.0.iter().filter_map(move |(refname, oid)| { + let ns = refname.to_namespaced()?; + (ns.namespace() == Component::from(remote)) + .then(|| (ns.strip_namespace().to_ref_string(), *oid)) + }) + } + + pub fn update<'a, I>(&mut self, updates: I) -> Applied<'a> + where + I: IntoIterator>, + { + updates + .into_iter() + .fold(Applied::default(), |mut ap, update| match update { + Update::Direct { name, target, .. } => { + let name = name.into_qualified().into_owned(); + let prev = match self.0.insert(name.clone(), target) { + Some(prev) => prev, + None => radicle::git::raw::Oid::zero().into(), + }; + ap.updated.push(RefUpdate::Updated { + name: name.to_ref_string(), + old: prev, + new: target, + }); + ap + } + Update::Prune { name, .. } => { + let name = name.into_qualified().into_owned(); + if let Some((name, prev)) = self.0.remove_entry(&name) { + ap.updated.push(RefUpdate::Deleted { + name: name.to_ref_string(), + oid: prev, + }) + } + ap + } + }) + } + + #[allow(dead_code)] + pub(crate) fn print(&self) { + if self.0.is_empty() { + println!("Refdb is empty!"); + } else { + for (name, oid) in self.0.iter() { + println!("{name} -> {oid}"); + } + } + } +} diff --git a/radicle-fetch/src/git/refs.rs b/radicle-fetch/src/git/refs.rs new file mode 100644 index 00000000..ffa4d96d --- /dev/null +++ b/radicle-fetch/src/git/refs.rs @@ -0,0 +1,2 @@ +mod update; +pub use update::{Applied, Policy, RefUpdate, Update, Updates}; diff --git a/radicle-fetch/src/git/refs/update.rs b/radicle-fetch/src/git/refs/update.rs new file mode 100644 index 00000000..9bb58f85 --- /dev/null +++ b/radicle-fetch/src/git/refs/update.rs @@ -0,0 +1,145 @@ +//! The set of types that describe performing updates to a Git +//! repository. +//! +//! An [`Update`] describes a single update that can made to a Git +//! repository. **Note** that it currently does not support symbolic +//! references. +//! +//! A group of `Update`s is described by [`Updates`] which groups +//! those updates by each peer's namespace, i.e. their [`PublicKey`]. +//! +//! When an `Update` is successful the corresponding [`Updated`] is +//! expected to be produced. +//! +//! The final result of applying a set of [`Updates`] is captured in +//! the [`Applied`] type, which contains any rejected, but non-fatal, +//! [`Update`]s and successful [`Updated`] values. + +use std::collections::BTreeMap; + +use either::Either; +use radicle::git::{Namespaced, Oid, Qualified}; +use radicle::prelude::PublicKey; + +pub use radicle::storage::RefUpdate; + +/// The set of applied changes from a reference store update. +#[derive(Debug, Default)] +pub struct Applied<'a> { + /// Set of rejected updates if they did not meet the update + /// requirements, e.g. concurrent change to previous object id, + /// broke fast-forward policy, etc. + pub rejected: Vec>, + /// Set of successfully updated references. + pub updated: Vec, +} + +impl<'a> Applied<'a> { + pub fn append(&mut self, other: &mut Self) { + self.rejected.append(&mut other.rejected); + self.updated.append(&mut other.updated); + } +} + +/// A set of [`Update`]s that are grouped by which namespace they are +/// affecting. +#[derive(Clone, Default, Debug)] +pub struct Updates<'a> { + pub tips: BTreeMap>>, +} + +impl<'a> Updates<'a> { + pub fn build(updates: impl IntoIterator)>) -> Self { + let tips = updates.into_iter().fold( + BTreeMap::<_, Vec>>::new(), + |mut tips, (remote, up)| { + tips.entry(remote) + .and_modify(|ups| ups.push(up.clone())) + .or_insert(vec![up]); + tips + }, + ); + Self { tips } + } + + pub fn add(&mut self, remote: PublicKey, up: Update<'a>) { + self.tips + .entry(remote) + .and_modify(|ups| ups.push(up.clone())) + .or_insert(vec![up]); + } + + pub fn append(&mut self, remote: PublicKey, mut new: Vec>) { + self.tips + .entry(remote) + .and_modify(|ups| ups.append(&mut new)) + .or_insert(new); + } +} + +/// The policy to follow when an [`Update::Direct`] is not a +/// fast-forward. +#[derive(Clone, Copy, Debug)] +pub enum Policy { + /// Abort the entire transaction. + Abort, + /// Reject this update, but continue the transaction. + Reject, + /// Allow the update. + Allow, +} + +/// An update that can be applied to a Git repository. +#[derive(Clone, Debug)] +pub enum Update<'a> { + /// Update a direct reference, i.e. a reference that points to an + /// object. + Direct { + /// The name of the reference that is being updated. + name: Namespaced<'a>, + /// The resulting target of the reference that is being + /// updated. + target: Oid, + /// Policy to apply when an [`Update`] would not apply as a + /// fast-forward. + no_ff: Policy, + }, + /// Delete a reference. + Prune { + /// The name of the reference that is being deleted. + name: Namespaced<'a>, + /// The previous value of the reference. + /// + /// It can either be a direct reference pointing to an + /// [`Oid`], or a symbolic reference pointing to a + /// [`Qualified`] reference name. + prev: Either>, + }, +} + +impl<'a> Update<'a> { + pub fn refname(&self) -> &Namespaced<'a> { + match self { + Update::Direct { name, .. } => name, + Update::Prune { name, .. } => name, + } + } + + pub fn into_owned<'b>(self) -> Update<'b> { + match self { + Self::Direct { + name, + target, + no_ff, + } => Update::Direct { + name: name.into_owned(), + target, + no_ff, + }, + Self::Prune { name, prev } => Update::Prune { + name: name.into_owned(), + prev: prev.map_right(|q| q.into_owned()), + }, + } + } +} diff --git a/radicle-fetch/src/git/repository.rs b/radicle-fetch/src/git/repository.rs new file mode 100644 index 00000000..bdc6f83f --- /dev/null +++ b/radicle-fetch/src/git/repository.rs @@ -0,0 +1,184 @@ +pub mod error; + +use either::{ + Either, + Either::{Left, Right}, +}; +use radicle::git::{Namespaced, Oid, Qualified}; +use radicle::storage::git::Repository; +use radicle::storage::ReadRepository; + +use super::refs::{Applied, Policy, RefUpdate, Update}; + +pub fn contains(repo: &Repository, oid: Oid) -> Result { + repo.backend + .odb() + .map(|odb| odb.exists(oid.into())) + .map_err(error::Contains) +} + +pub fn is_in_ancestry_path(repo: &Repository, old: Oid, new: Oid) -> Result { + if !contains(repo, old)? || !contains(repo, new)? { + return Ok(false); + } + + if old == new { + return Ok(true); + } + + repo.is_ancestor_of(old, new) + .map_err(|err| error::Ancestry::Check { old, new, err }) +} + +pub fn refname_to_id<'a, N>(repo: &Repository, refname: N) -> Result, error::Resolve> +where + N: Into>, +{ + use radicle::git::raw::ErrorCode::NotFound; + + let refname = refname.into(); + match repo.backend.refname_to_id(refname.as_ref()) { + Ok(oid) => Ok(Some(oid.into())), + Err(e) if matches!(e.code(), NotFound) => Ok(None), + Err(err) => Err(error::Resolve { + name: refname.to_owned(), + err, + }), + } +} + +pub fn update<'a, I>(repo: &Repository, updates: I) -> Result, error::Update> +where + I: IntoIterator>, +{ + let mut applied = Applied::default(); + for up in updates.into_iter() { + match up { + Update::Direct { + name, + target, + no_ff, + } => match direct(repo, name, target, no_ff)? { + Left(r) => applied.rejected.push(r), + Right(u) => applied.updated.push(u), + }, + Update::Prune { name, prev } => match prune(repo, name, prev)? { + Left(r) => applied.rejected.push(r), + Right(u) => applied.updated.push(u), + }, + } + } + + Ok(applied) +} + +fn direct<'a>( + repo: &Repository, + name: Namespaced<'a>, + target: Oid, + no_ff: Policy, +) -> Result, RefUpdate>, error::Update> { + let tip = refname_to_id(repo, name.clone())?; + match tip { + Some(prev) => { + let is_ff = is_in_ancestry_path(repo, prev, target)?; + if !is_ff { + match no_ff { + Policy::Abort => { + return Err(error::Update::NonFF { + name: name.to_owned(), + new: target, + cur: prev, + }) + } + Policy::Reject => Ok(Left(Update::Direct { + name, + target, + no_ff, + })), + Policy::Allow => { + // N.b. the update is a non-fast-forward but + // we allow it, so we pass `force: true`. + repo.backend + .reference(name.as_ref(), target.into(), true, "radicle: update") + .map_err(|err| error::Update::Create { + name: name.to_owned(), + target, + err, + })?; + Ok(Right(RefUpdate::from(name.to_ref_string(), prev, target))) + } + } + } else { + // N.b. the update is a fast-forward so we can safely + // pass `force: true`. + repo.backend + .reference(name.as_ref(), target.into(), true, "radicle: update") + .map_err(|err| error::Update::Create { + name: name.to_owned(), + target, + err, + })?; + Ok(Right(RefUpdate::from(name.to_ref_string(), prev, target))) + } + } + None => { + // N.b. the reference didn't exist so we pass `force: + // false`. + repo.backend + .reference(name.as_ref(), target.into(), false, "radicle: create") + .map_err(|err| error::Update::Create { + name: name.to_owned(), + target, + err, + })?; + Ok(Right(RefUpdate::Created { + name: name.to_ref_string(), + oid: target, + })) + } + } +} + +fn prune<'a>( + repo: &Repository, + name: Namespaced<'a>, + prev: Either>, +) -> Result, RefUpdate>, error::Update> { + use radicle::git::raw::ObjectType; + + match find(repo, &name)? { + Some(mut r) => { + // N.b. peel this reference to whatever object it points to, + // presumably a commit, and get its Oid + let prev = r + .peel(ObjectType::Any) + .map_err(error::Update::Peel)? + .id() + .into(); + r.delete().map_err(|err| error::Update::Delete { + name: name.to_owned(), + err, + })?; + Ok(Right(RefUpdate::Deleted { + name: name.to_ref_string(), + oid: prev, + })) + } + None => Ok(Left(Update::Prune { name, prev })), + } +} + +fn find<'a>( + repo: &'a Repository, + name: &Namespaced<'_>, +) -> Result>, error::Update> { + match repo.backend.find_reference(name.as_ref()) { + Ok(r) => Ok(Some(r)), + Err(e) if matches!(e.code(), radicle::git::raw::ErrorCode::NotFound) => Ok(None), + Err(err) => Err(error::Update::Find { + name: name.clone().into_owned(), + err, + }), + } +} diff --git a/radicle-fetch/src/git/repository/error.rs b/radicle-fetch/src/git/repository/error.rs new file mode 100644 index 00000000..effb6ca9 --- /dev/null +++ b/radicle-fetch/src/git/repository/error.rs @@ -0,0 +1,70 @@ +use radicle::git::{ext, raw, Namespaced, Oid, Qualified}; +use thiserror::Error; + +#[derive(Debug, Error)] +#[error("could not open Git ODB")] +pub struct Contains(#[source] pub raw::Error); + +#[derive(Debug, Error)] +pub enum Ancestry { + #[error(transparent)] + Contains(#[from] Contains), + #[error("failed to check ancestry for {old} and {new}")] + Check { + old: Oid, + new: Oid, + #[source] + err: ext::Error, + }, +} + +#[derive(Debug, Error)] +#[error("failed to resolve {name} to its Oid")] +pub struct Resolve { + pub name: Qualified<'static>, + #[source] + pub err: raw::Error, +} + +#[derive(Debug, Error)] +#[error("failed to scan for refs matching {pattern}")] +pub struct Scan { + pub pattern: radicle::git::PatternString, + #[source] + pub err: ext::Error, +} + +#[derive(Debug, Error)] +pub enum Update { + #[error(transparent)] + Ancestry(#[from] Ancestry), + #[error("failed to create reference from {name} to {target}")] + Create { + name: Namespaced<'static>, + target: Oid, + #[source] + err: raw::Error, + }, + #[error("failed to delete reference {name}")] + Delete { + name: Namespaced<'static>, + #[source] + err: raw::Error, + }, + #[error("failed to find ref {name}")] + Find { + name: Namespaced<'static>, + #[source] + err: raw::Error, + }, + #[error("non-fast-forward update of {name} (current: {cur}, new: {new})")] + NonFF { + name: Namespaced<'static>, + new: Oid, + cur: Oid, + }, + #[error("failed to peel ref to object")] + Peel(#[source] raw::Error), + #[error(transparent)] + Resolve(#[from] Resolve), +} diff --git a/radicle-fetch/src/handle.rs b/radicle-fetch/src/handle.rs new file mode 100644 index 00000000..ddcff453 --- /dev/null +++ b/radicle-fetch/src/handle.rs @@ -0,0 +1,128 @@ +use std::sync::atomic::{self, AtomicBool}; +use std::sync::Arc; + +use bstr::BString; +use radicle::crypto::{PublicKey, Verified}; +use radicle::git::Oid; +use radicle::identity::DocError; +use radicle::prelude::Doc; +use radicle::storage::git::Repository; +use radicle::storage::ReadRepository; + +use crate::tracking::{BlockList, Tracked}; +use crate::transport::{ConnectionStream, Transport}; + +/// The handle used for pulling or cloning changes from a remote peer. +pub struct Handle { + pub(crate) local: PublicKey, + pub(crate) repo: Repository, + pub(crate) tracked: Tracked, + pub(crate) transport: Transport, + /// The set of keys we will ignore when fetching from a + /// remote. This set can be constructed using the tracking + /// `config`'s blocked node entries. + /// + /// Note that it's important to ignore the local peer's + /// key in [`crate::pull`], however, we choose to allow the local + /// peer's key in [`crate::clone`]. + pub(crate) blocked: BlockList, + // Signals to the pack writer to interrupt the process + pub(crate) interrupt: Arc, +} + +impl Handle { + pub fn new( + local: PublicKey, + repo: Repository, + tracked: Tracked, + blocked: BlockList, + connection: S, + ) -> Result + where + S: ConnectionStream, + { + let git_dir = repo.backend.path().to_path_buf(); + let transport = Transport::new(git_dir, BString::from(repo.id.canonical()), connection); + + Ok(Self { + local, + repo, + tracked, + transport, + blocked, + interrupt: Arc::new(AtomicBool::new(false)), + }) + } + + pub fn is_blocked(&self, key: &PublicKey) -> bool { + self.blocked.is_blocked(key) + } + + pub fn repository(&self) -> &Repository { + &self.repo + } + + pub fn repository_mut(&mut self) -> &mut Repository { + &mut self.repo + } + + pub fn local(&self) -> &PublicKey { + &self.local + } + + pub fn interrupt_pack_writer(&mut self) { + self.interrupt.store(true, atomic::Ordering::Relaxed); + } + + pub fn verified(&self, head: Oid) -> Result, DocError> { + Ok(self.repo.identity_doc_at(head)?.doc) + } + + pub fn tracked(&self) -> Tracked { + self.tracked.clone() + } +} + +pub mod error { + use std::io; + + use radicle::node::tracking; + use radicle::prelude::Id; + use radicle::{git, storage}; + use thiserror::Error; + + #[derive(Debug, Error)] + pub enum Init { + #[error(transparent)] + Io(#[from] io::Error), + #[error(transparent)] + Tracking(#[from] tracking::config::Error), + } + + #[derive(Debug, Error)] + pub enum Tracking { + #[error("failed to find tracking policy for {rid}")] + FailedPolicy { + rid: Id, + #[source] + err: tracking::store::Error, + }, + #[error("cannot fetch {rid} as it is not tracked")] + BlockedPolicy { rid: Id }, + #[error("failed to get tracking nodes for {rid}")] + FailedNodes { + rid: Id, + #[source] + err: tracking::store::Error, + }, + + #[error(transparent)] + Storage(#[from] storage::Error), + + #[error(transparent)] + Git(#[from] git::raw::Error), + + #[error(transparent)] + Refs(#[from] storage::refs::Error), + } +} diff --git a/radicle-fetch/src/lib.rs b/radicle-fetch/src/lib.rs new file mode 100644 index 00000000..5d77529e --- /dev/null +++ b/radicle-fetch/src/lib.rs @@ -0,0 +1,106 @@ +pub mod git; +pub mod handle; +pub mod tracking; +pub mod transport; + +pub(crate) mod sigrefs; + +mod refs; +mod stage; +mod state; + +pub use handle::Handle; +pub use state::{FetchLimit, FetchResult}; +pub use tracking::{BlockList, Scope, Tracked}; +pub use transport::Transport; + +use std::io; + +use radicle::crypto::PublicKey; +use state::FetchState; +use thiserror::Error; + +#[derive(Debug, Error)] +pub enum Error { + #[error("failed to perform fetch handshake")] + Handshake { + #[source] + err: io::Error, + }, + #[error("failed to load `rad/id`")] + Identity { + #[source] + err: Box, + }, + #[error(transparent)] + Protocol(#[from] state::error::Protocol), + #[error("missing `rad/id`")] + MissingRadId, + #[error("attempted to replicate from self")] + ReplicateSelf, +} + +/// Pull changes from the `remote`. +/// +/// It is expected that the local peer has a copy of the repository +/// and is pulling new changes. If the repository does not exist, then +/// [`clone`] should be used. +pub fn pull( + handle: &mut Handle, + limit: FetchLimit, + remote: PublicKey, +) -> Result +where + S: transport::ConnectionStream, +{ + let local = *handle.local(); + if local == remote { + return Err(Error::ReplicateSelf); + } + let state = FetchState::default(); + let handshake = handle + .transport + .handshake() + .map_err(|err| Error::Handshake { err })?; + // N.b. ensure that we ignore the local peer's key. + handle.blocked.extend([local]); + state + .run(handle, &handshake, limit, remote) + .map_err(Error::Protocol) +} + +/// Clone changes from the `remote`. +/// +/// It is expected that the local peer has an empty repository which +/// they want to populate with the `remote`'s view of the project. +pub fn clone( + handle: &mut Handle, + limit: FetchLimit, + remote: PublicKey, +) -> Result +where + S: transport::ConnectionStream, +{ + if *handle.local() == remote { + return Err(Error::ReplicateSelf); + } + let handshake = handle + .transport + .handshake() + .map_err(|err| Error::Handshake { err })?; + let mut state = FetchState::default(); + state + .run_stage( + handle, + &handshake, + &stage::CanonicalId { + remote, + limit: limit.special, + }, + ) + .map_err(|e| Error::from(state::error::Protocol::from(e)))?; + + state + .run(handle, &handshake, limit, remote) + .map_err(Error::Protocol) +} diff --git a/radicle-fetch/src/refs.rs b/radicle-fetch/src/refs.rs new file mode 100644 index 00000000..0c8d84e8 --- /dev/null +++ b/radicle-fetch/src/refs.rs @@ -0,0 +1,182 @@ +use bstr::{BString, ByteSlice}; +use either::Either; +use radicle::crypto::PublicKey; +use radicle::git::{self, Component, Namespaced, Oid, Qualified}; +use thiserror::Error; + +pub use radicle::git::refs::storage; +pub use radicle::git::refs::storage::Special; + +use crate::git::refs::{Policy, Update}; + +pub(crate) use radicle::git::refs::storage::IDENTITY_BRANCH as REFS_RAD_ID; + +#[derive(Debug, Error)] +#[non_exhaustive] +pub enum Error { + #[error("non-namespaced ref name '{0}' is not 'refs/rad/id'")] + NotCanonicalRadID(Qualified<'static>), + #[error("invalid remote peer id")] + PublicKey(#[from] radicle::crypto::PublicKeyError), + #[error(transparent)] + Ref(#[from] radicle::git::RefError), + #[error(transparent)] + Utf8(#[from] bstr::Utf8Error), +} + +pub(crate) fn unpack_ref<'a>( + r: gix_protocol::handshake::Ref, +) -> Result<(ReceivedRefname<'a>, Oid), Error> { + use crate::git::oid; + use gix_protocol::handshake::Ref; + + match r { + Ref::Peeled { + full_ref_name, + object, + .. + } + | Ref::Direct { + full_ref_name, + object, + } + | Ref::Symbolic { + full_ref_name, + object, + .. + } => ReceivedRefname::try_from(full_ref_name).map(|name| (name, oid::to_oid(object))), + Ref::Unborn { full_ref_name, .. } => { + unreachable!("BUG: unborn ref {}", full_ref_name) + } + } +} + +/// A reference name received during an exchange with another peer. The +/// expected references are either namespaced references in the form +/// of [`RemoteRef`] or the canonical `rad/id` reference. +#[derive(Debug)] +pub(crate) enum ReceivedRefname<'a> { + /// A reference name under a `remote` namespace. + /// + /// # Examples + /// + /// * `refs/namespaces//refs/rad/id` + /// * `refs/namespaces//refs/rad/sigrefs` + /// * `refs/namespaces//refs/heads/main` + /// * `refs/namespaces//refs/cobs/issue.rad.xyz` + Namespaced { + /// The namespace of the remote. + remote: PublicKey, + /// The reference is expected to either be a [`Special`] reference + /// or a generic reference name. + suffix: Either>, + }, + /// The canonical `refs/rad/id` reference + RadId, +} + +impl<'a> ReceivedRefname<'a> { + pub fn remote(remote: PublicKey, suffix: Qualified<'a>) -> Self { + Self::Namespaced { + remote, + suffix: Either::Right(suffix), + } + } + + pub fn to_qualified<'b>(&self) -> Qualified<'b> { + match &self { + Self::Namespaced { remote, suffix } => match suffix { + Either::Left(s) => Qualified::from(*s) + .with_namespace(Component::from(remote)) + .into(), + Either::Right(name) => { + Qualified::from(name.with_namespace(Component::from(remote))).to_owned() + } + }, + Self::RadId => REFS_RAD_ID.clone(), + } + } + + pub fn to_namespaced<'b>(&self) -> Option> { + match self { + Self::Namespaced { remote, suffix } => Some(match suffix { + Either::Left(special) => special.namespaced(remote), + Either::Right(refname) => { + refname.with_namespace(Component::from(remote)).to_owned() + } + }), + Self::RadId => None, + } + } +} + +impl TryFrom for ReceivedRefname<'_> { + type Error = Error; + + fn try_from(value: BString) -> Result { + match git::parse_ref::(value.to_str()?)? { + (None, name) => (name == *REFS_RAD_ID) + .then_some(ReceivedRefname::RadId) + .ok_or_else(|| Error::NotCanonicalRadID(name.to_owned())), + (Some(remote), name) => Ok(ReceivedRefname::Namespaced { + remote, + suffix: match Special::from_qualified(&name) { + None => Either::Right(name.to_owned()), + Some(special) => Either::Left(special), + }, + }), + } + } +} + +/// A reference name and the associated tip received during an +/// exchange with another peer. +#[derive(Debug)] +pub(crate) struct ReceivedRef { + pub tip: Oid, + pub name: ReceivedRefname<'static>, +} + +impl ReceivedRef { + pub fn new(tip: Oid, name: ReceivedRefname<'static>) -> Self { + Self { tip, name } + } + + pub fn to_qualified(&self) -> Qualified<'static> { + self.name.to_qualified() + } + + pub fn as_special_ref_update(&self, is_delegate: F) -> Option<(PublicKey, Update<'static>)> + where + F: Fn(&PublicKey) -> bool, + { + match &self.name { + ReceivedRefname::RadId => None, + ReceivedRefname::Namespaced { remote, suffix } => { + special_update(remote, suffix, self.tip, is_delegate).map(|up| (*remote, up)) + } + } + } +} + +pub(crate) fn special_update( + remote: &PublicKey, + suffix: &Either, + tip: Oid, + is_delegate: F, +) -> Option> +where + F: Fn(&PublicKey) -> bool, +{ + suffix.as_ref().left().map(|special| Update::Direct { + name: special.namespaced(remote).to_owned(), + target: tip, + // N.b. reject any updates if the remote is not a delegate, + // since this is not fatal. + no_ff: if is_delegate(remote) { + Policy::Abort + } else { + Policy::Reject + }, + }) +} diff --git a/radicle-fetch/src/sigrefs.rs b/radicle-fetch/src/sigrefs.rs new file mode 100644 index 00000000..7f0b41b0 --- /dev/null +++ b/radicle-fetch/src/sigrefs.rs @@ -0,0 +1,156 @@ +use std::collections::{BTreeMap, BTreeSet}; +use std::ops::{Deref, Not as _}; + +pub use radicle::storage::refs::SignedRefsAt; +pub use radicle::storage::{git::Validation, Validations}; +use radicle::{crypto::PublicKey, storage::ValidateRepository}; + +use crate::state::Cached; + +pub mod error { + use radicle::crypto::PublicKey; + use thiserror::Error; + + #[derive(Debug, Error)] + #[non_exhaustive] + pub enum RemoteRefs { + #[error("required sigrefs of {0} not found")] + NotFound(PublicKey), + #[error(transparent)] + Load(#[from] Load), + } + + pub type Load = radicle::storage::refs::Error; +} + +/// A data carrier that associates that data with whether a given +/// `PublicKey` is a delegate or a non-delegate. +/// +/// Construct a `DelegateStatus` via [`DelegateStatus::empty`], if no +/// data is required, or [`DelegateStatus::new`] if there is data to +/// associate. +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)] +pub(crate) enum DelegateStatus { + Delegate { remote: PublicKey, data: T }, + NonDelegate { remote: PublicKey, data: T }, +} + +impl DelegateStatus { + /// Construct a `DelegateStatus` without any data. + pub fn empty(remote: PublicKey, delegates: &BTreeSet) -> Self { + Self::new((), remote, delegates) + } +} + +impl DelegateStatus { + pub fn new(data: T, remote: PublicKey, delegates: &BTreeSet) -> Self { + if delegates.contains(&remote) { + Self::Delegate { remote, data } + } else { + Self::NonDelegate { remote, data } + } + } + + /// Construct a `DelegateStatus` with [`SignedRefsAt`] signed reference + /// data, if it can be found in `repo`. + pub fn load( + self, + cached: &Cached, + ) -> Result>, radicle::storage::refs::Error> { + let remote = *self.remote(); + self.traverse(|_| cached.load(&remote)) + } + + fn remote(&self) -> &PublicKey { + match self { + Self::Delegate { remote, .. } => remote, + Self::NonDelegate { remote, .. } => remote, + } + } + + fn traverse(self, f: impl FnOnce(T) -> Result) -> Result, E> { + match self { + Self::Delegate { remote, data } => Ok(DelegateStatus::Delegate { + remote, + data: f(data)?, + }), + Self::NonDelegate { remote, data } => Ok(DelegateStatus::NonDelegate { + remote, + data: f(data)?, + }), + } + } +} + +pub(crate) fn validate( + repo: &impl ValidateRepository, + SignedRefsAt { sigrefs, .. }: SignedRefsAt, +) -> Result, radicle::storage::Error> { + let remote = radicle::storage::Remote::::new(sigrefs); + let validations = repo.validate_remote(&remote)?; + Ok(validations.is_empty().not().then_some(validations)) +} + +/// The sigrefs found for each remote. +/// +/// Construct using [`RemoteRefs::load`]. +#[derive(Debug, Default)] +pub struct RemoteRefs(BTreeMap); + +impl RemoteRefs { + /// Load the sigrefs for the given `must` and `may` remotes. + /// + /// The `must` remotes have to be present, otherwise an error will + /// be returned. + /// + /// The `may` remotes do not have to be present and any missing + /// sigrefs for that remote will be ignored. + pub(crate) fn load( + cached: &Cached, + Select { must, may }: Select, + ) -> Result { + let must = must.iter().map(|id| { + cached + .load(id) + .map_err(error::RemoteRefs::from) + .and_then(|sr| match sr { + None => Err(error::RemoteRefs::NotFound(*id)), + Some(sr) => Ok((id, sr)), + }) + }); + let may = may.iter().filter_map(|id| match cached.load(id) { + Ok(None) => None, + Ok(Some(sr)) => Some(Ok((id, sr))), + Err(e) => Some(Err(e.into())), + }); + + must.chain(may) + .try_fold(RemoteRefs::default(), |mut acc, remote_refs| { + let (id, sigrefs) = remote_refs?; + acc.0.insert(*id, sigrefs); + Ok(acc) + }) + } +} + +impl Deref for RemoteRefs { + type Target = BTreeMap; + + fn deref(&self) -> &Self::Target { + &self.0 + } +} + +impl<'a> IntoIterator for &'a RemoteRefs { + type Item = <&'a BTreeMap as IntoIterator>::Item; + type IntoIter = <&'a BTreeMap as IntoIterator>::IntoIter; + + fn into_iter(self) -> Self::IntoIter { + self.0.iter() + } +} + +pub struct Select<'a> { + pub must: &'a BTreeSet, + pub may: &'a BTreeSet, +} diff --git a/radicle-fetch/src/stage.rs b/radicle-fetch/src/stage.rs new file mode 100644 index 00000000..c8662055 --- /dev/null +++ b/radicle-fetch/src/stage.rs @@ -0,0 +1,484 @@ +//! The Radicle fetch protocol can be split into two actions: `clone` +//! and `pull`. Each of these actions will interact with the server in +//! multiple stages, where each stage will perform a single roundtrip +//! of fetching. These stages are encapsulated in the +//! [`ProtocolStage`] trait. +//! +//! ### Clone +//! +//! A `clone` is split into three stages: +//! +//! 1. [`CanonicalId`]: fetches the canonical `refs/rad/id` to use +//! as an anchor for the rest of the fetch, i.e. provides initial +//! delegate data for the repository. +//! 2. [`SpecialRefs`]: fetches the special references, `rad/id` and +//! `rad/sigrefs`, for each configured namespace, i.e. tracked +//! and delegate peers if the scope is trusted and all peers is the +//! scope is all. +//! 3. [`DataRefs`]: fetches the `Oid`s for each reference listed in +//! the `rad/sigrefs` for each fetched peer in the +//! [`SpecialRefs`] stage. Additionally, any references that have +//! been removed from `rad/sigrefs` are marked for deletion. +//! +//! ### Pull +//! +//! A `pull` is split into two stages: +//! +//! 1. [`SpecialRefs`]: see above. +//! 2. [`DataRefs`]: see above. + +use std::collections::{BTreeMap, BTreeSet, HashSet}; + +use bstr::BString; +use either::Either; +use gix_protocol::handshake::Ref; +use nonempty::NonEmpty; +use radicle::crypto::PublicKey; +use radicle::git::{refname, Component, Namespaced, Qualified}; +use radicle::storage::git::Repository; +use radicle::storage::ReadRepository; + +use crate::git::refs::{Policy, Update, Updates}; +use crate::refs::{ReceivedRef, ReceivedRefname}; +use crate::sigrefs; +use crate::state::FetchState; +use crate::tracking::BlockList; +use crate::transport::WantsHaves; +use crate::{refs, tracking}; + +pub mod error { + use radicle::crypto::PublicKey; + use radicle::git::RefString; + use thiserror::Error; + + use crate::transport::WantsHavesError; + + #[derive(Debug, Error)] + pub enum Layout { + #[error("missing required refs: {0:?}")] + MissingRequiredRefs(Vec), + } + + #[derive(Debug, Error)] + pub enum Prepare { + #[error(transparent)] + References(#[from] radicle::storage::Error), + #[error("verification of rad/id for {remote} failed")] + Verification { + remote: PublicKey, + #[source] + err: Box, + }, + } + + #[derive(Debug, Error)] + pub enum WantsHaves { + #[error(transparent)] + WantsHavesAdd(#[from] WantsHavesError), + #[error("expected namespaced ref {0}")] + NotNamespaced(RefString), + } +} + +/// A [`ProtocolStage`] describes a single roundtrip with the Radicle +/// node that is serving the data. +/// +/// The stages are used as input for [`crate::FetchState::step`] and +/// are called in the order that they are listed here, .i.e: +/// +/// 1. `ls_refs`: asks the server for the provided reference +/// prefixes. +/// 2. `ref_filter`: filter the advertised refs to the set required +/// for inspection. +/// 3. `pre_validate`: before fetching the data, ensure the server +/// advertised the references that are required. +/// 4. `wants_haves`: build the set of `want`s and `have`s to send +/// to the server. +/// 5. `prepare_updates`: prepares the set of updates to update the +/// refdb (in-memory and production). +pub(crate) trait ProtocolStage { + /// If and how to perform `ls-refs`. + fn ls_refs(&self) -> Option>; + + /// Filter a remote-advertised [`Ref`]. + /// + /// Return `Some` if the ref should be considered, `None` otherwise. This + /// method may be called with the response of `ls-refs`, the `wanted-refs` + /// of a `fetch` response, or both. + fn ref_filter(&self, r: Ref) -> Option; + + /// Validate that all advertised refs conform to an expected layout. + /// + /// The supplied `refs` are `ls-ref`-advertised refs filtered + /// through [`ProtocolStage::ref_filter`]. + fn pre_validate(&self, refs: &[ReceivedRef]) -> Result<(), error::Layout>; + + /// Assemble the `want`s and `have`s for a `fetch`, retaining the refs which + /// would need updating after the `fetch` succeeds. + /// + /// The `refs` are the advertised refs from executing `ls-refs`, filtered + /// through [`ProtocolStage::ref_filter`]. + fn wants_haves( + &self, + refdb: &Repository, + refs: &[ReceivedRef], + ) -> Result { + let mut wants_haves = WantsHaves::default(); + wants_haves.add( + refdb, + refs.iter().map(|recv| (recv.to_qualified(), recv.tip)), + )?; + Ok(wants_haves) + } + + /// Prepare the [`Updates`] based on the received `refs`. + /// + /// These updates can then be used to update the refdb. + fn prepare_updates<'a>( + &self, + s: &FetchState, + repo: &Repository, + refs: &'a [ReceivedRef], + ) -> Result, error::Prepare>; +} + +/// The [`ProtocolStage`] for performing an initial clone from a `remote`. +/// +/// This step asks for the canonical `refs/rad/id` reference, which +/// allows us to use it as an anchor for the following steps. +#[derive(Debug)] +pub struct CanonicalId { + pub remote: PublicKey, + pub limit: u64, +} + +impl ProtocolStage for CanonicalId { + fn ls_refs(&self) -> Option> { + Some(NonEmpty::new(refs::REFS_RAD_ID.as_bstr().into())) + } + + fn ref_filter(&self, r: Ref) -> Option { + match refs::unpack_ref(r).ok()? { + ( + refname @ ReceivedRefname::Namespaced { + suffix: Either::Left(_), + .. + }, + tip, + ) => Some(ReceivedRef::new(tip, refname)), + (ReceivedRefname::RadId, tip) => Some(ReceivedRef::new(tip, ReceivedRefname::RadId)), + _ => None, + } + } + + fn pre_validate(&self, refs: &[ReceivedRef]) -> Result<(), error::Layout> { + // Ensures that we fetched the canonical 'refs/rad/id' + ensure_refs( + [BString::from(refs::REFS_RAD_ID.as_bstr())] + .into_iter() + .collect(), + refs.iter() + .map(|r| r.to_qualified().to_string().into()) + .collect(), + ) + } + + fn prepare_updates<'a>( + &self, + s: &FetchState, + repo: &Repository, + refs: &'a [ReceivedRef], + ) -> Result, error::Prepare> { + // SAFETY: checked by `pre_validate` that the `refs/rad/id` + // was received + let verified = repo + .identity_doc_at( + *s.canonical_rad_id() + .expect("ensure we got canonicdal 'rad/id' ref"), + ) + .map_err(|err| error::Prepare::Verification { + remote: self.remote, + err: Box::new(err), + })?; + if verified.delegates.contains(&self.remote.into()) { + let is_delegate = |remote: &PublicKey| verified.is_delegate(remote); + Ok(Updates::build( + refs.iter() + .filter_map(|r| r.as_special_ref_update(is_delegate)), + )) + } else { + Ok(Updates::default()) + } + } +} + +/// The [`ProtocolStage`] for fetching special refs from the set of +/// remotes in `tracked` and `delegates`. +/// +/// This step asks for all tracked and delegate remote's `rad/id` and +/// `rad/sigrefs`, iff the scope is +/// [`tracking::Scope::Trusted`]. Otherwise, it asks for all +/// namespaces. +/// +/// It ensures that all delegate refs were fetched. +#[derive(Debug)] +pub struct SpecialRefs { + /// The set of nodes that should be blocked from fetching. + pub blocked: BlockList, + /// The node that is being fetched from. + pub remote: PublicKey, + /// The set of nodes to be fetched. + pub tracked: tracking::Tracked, + /// The set of delegates to be fetched, with the local node + /// removed in the case of a `pull`. + pub delegates: BTreeSet, + /// The data limit for this stage of fetching. + pub limit: u64, +} + +impl ProtocolStage for SpecialRefs { + fn ls_refs(&self) -> Option> { + match &self.tracked { + tracking::Tracked::All => Some(NonEmpty::new("refs/namespaces".into())), + tracking::Tracked::Trusted { remotes } => NonEmpty::collect( + remotes + .iter() + .chain(self.delegates.iter()) + .flat_map(|remote| { + [ + BString::from(radicle::git::refs::storage::id(remote).to_string()), + BString::from(radicle::git::refs::storage::sigrefs(remote).to_string()), + ] + }), + ), + } + } + + fn ref_filter(&self, r: Ref) -> Option { + let (refname, tip) = refs::unpack_ref(r).ok()?; + match refname { + // N.b. ensure that any blocked peers are filtered since + // `Scope::All` can ls for them + ReceivedRefname::Namespaced { remote, .. } if self.blocked.is_blocked(&remote) => None, + ReceivedRefname::Namespaced { ref suffix, .. } if suffix.is_left() => { + Some(ReceivedRef::new(tip, refname)) + } + ReceivedRefname::Namespaced { .. } | ReceivedRefname::RadId => None, + } + } + + fn pre_validate(&self, refs: &[ReceivedRef]) -> Result<(), error::Layout> { + ensure_refs( + self.delegates + .iter() + .filter(|id| !self.blocked.is_blocked(id)) + .map(|id| { + // N.b. we asked for the rad/id but do not need to ensure it + BString::from(radicle::git::refs::storage::sigrefs(id).to_string()) + }) + .collect(), + refs.iter() + .filter_map(|r| r.name.to_namespaced()) + .map(|r| r.to_string().into()) + .collect(), + ) + } + + fn prepare_updates<'a>( + &self, + _s: &FetchState, + _repo: &Repository, + refs: &'a [ReceivedRef], + ) -> Result, error::Prepare> { + special_refs_updates(&self.delegates, &self.blocked, refs) + } +} + +/// The [`ProtocolStage`] for fetching data refs from the set of +/// remotes in `trusted`. +/// +/// All refs that are listed in the `remotes` sigrefs are checked +/// against our refdb/odb to build a set of `wants` and `haves`. The +/// `wants` will then be fetched from the server side to receive those +/// particular objects. +/// +/// Those refs and objects are then prepared for updating, removing +/// any that were found to exist before the latest fetch. +#[derive(Debug)] +pub struct DataRefs { + /// The node that is being fetched from. + pub remote: PublicKey, + /// The set of signed references from each remote that was + /// fetched. + pub remotes: sigrefs::RemoteRefs, + /// The data limit for this stage of fetching. + pub limit: u64, +} + +impl ProtocolStage for DataRefs { + // We don't need to ask for refs since we have all reference names + // and `Oid`s in `rad/sigrefs`. + fn ls_refs(&self) -> Option> { + None + } + + // Since we don't ask for refs, we don't need to filter them. + fn ref_filter(&self, _: Ref) -> Option { + None + } + + // Since we don't ask for refs, we don't need to validate them. + fn pre_validate(&self, _refs: &[ReceivedRef]) -> Result<(), error::Layout> { + Ok(()) + } + + // We ignore the `ReceivedRef`s since we are using the `remotes` + // as the source for refnames and `Oid`s. + fn wants_haves( + &self, + refdb: &Repository, + _refs: &[ReceivedRef], + ) -> Result { + let mut wants_haves = WantsHaves::default(); + + for (remote, loaded) in &self.remotes { + wants_haves.add( + refdb, + loaded.refs.iter().filter_map(|(refname, tip)| { + let refname = Qualified::from_refstr(refname) + .map(|refname| refname.with_namespace(Component::from(remote)))?; + Some((refname, *tip)) + }), + )?; + } + + Ok(wants_haves) + } + + fn prepare_updates<'a>( + &self, + _s: &FetchState, + repo: &Repository, + _refs: &'a [ReceivedRef], + ) -> Result, error::Prepare> { + let mut updates = Updates::default(); + + for (remote, refs) in &self.remotes { + let mut signed = HashSet::with_capacity(refs.refs.len()); + for (name, tip) in refs.iter() { + let tracking: Namespaced<'_> = Qualified::from_refstr(name) + .and_then(|q| refs::ReceivedRefname::remote(*remote, q).to_namespaced()) + .expect("we checked sigrefs well-formedness in wants_refs already"); + signed.insert(tracking.clone()); + updates.add( + *remote, + Update::Direct { + name: tracking, + target: *tip, + no_ff: Policy::Allow, + }, + ); + } + + // Prune refs not in signed + let prefix_rad = refname!("refs/rad"); + for (name, target) in repo.references_of(remote)? { + // 'rad/' refs are never subject to pruning + if name.starts_with(prefix_rad.as_str()) { + continue; + } + + let name = Qualified::from_refstr(name) + .expect("BUG: reference is guaranteed to be Qualified") + .with_namespace(Component::from(remote)); + + if !signed.contains(&name) { + updates.add( + *remote, + Update::Prune { + name, + prev: either::Left(target), + }, + ); + } + } + } + + Ok(updates) + } +} + +// N.b. the `delegates` are the delegates of the repository, with the +// potential removal of the local peer in the case of a `pull`. +fn special_refs_updates<'a>( + delegates: &BTreeSet, + blocked: &BlockList, + refs: &'a [ReceivedRef], +) -> Result, error::Prepare> { + use either::Either::*; + + let grouped = refs + .iter() + .filter_map(|r| match &r.name { + refs::ReceivedRefname::Namespaced { remote, suffix } => { + (!blocked.is_blocked(remote)).then_some((remote, r.tip, suffix.clone())) + } + refs::ReceivedRefname::RadId => None, + }) + .fold(BTreeMap::new(), |mut acc, (remote_id, tip, name)| { + acc.entry(*remote_id) + .or_insert_with(Vec::new) + .push((tip, name)); + acc + }); + + let mut updates = Updates::default(); + + for (remote_id, refs) in grouped { + let mut tips_inner = Vec::with_capacity(2); + for (tip, suffix) in &refs { + match &suffix { + Left(refs::Special::Id) => { + if let Some(u) = refs::special_update(&remote_id, suffix, *tip, |remote| { + delegates.contains(remote) + }) { + tips_inner.push(u); + } + } + + Left(refs::Special::SignedRefs) => { + if let Some(u) = refs::special_update(&remote_id, suffix, *tip, |remote| { + delegates.contains(remote) + }) { + tips_inner.push(u); + } + } + + Right(_) => continue, + } + } + + updates.append(remote_id, tips_inner); + } + + Ok(updates) +} + +fn ensure_refs(required: BTreeSet, wants: BTreeSet) -> Result<(), error::Layout> +where + T: Ord + ToString, +{ + if wants.is_empty() { + return Ok(()); + } + + let diff = required.difference(&wants).collect::>(); + + if diff.is_empty() { + Ok(()) + } else { + Err(error::Layout::MissingRequiredRefs( + diff.into_iter().map(|ns| ns.to_string()).collect(), + )) + } +} diff --git a/radicle-fetch/src/state.rs b/radicle-fetch/src/state.rs new file mode 100644 index 00000000..874b1c08 --- /dev/null +++ b/radicle-fetch/src/state.rs @@ -0,0 +1,532 @@ +use std::collections::{BTreeMap, BTreeSet}; + +use gix_protocol::handshake; +use radicle::crypto::PublicKey; +use radicle::git::{Oid, Qualified}; +use radicle::identity::{Doc, DocError}; + +use radicle::prelude::Verified; +use radicle::storage; +use radicle::storage::{ + git::Validation, Remote, RemoteId, RemoteRepository, Remotes, ValidateRepository, Validations, +}; + +use crate::git; +use crate::git::refs::{Applied, Update}; +use crate::git::repository; +use crate::sigrefs::SignedRefsAt; +use crate::stage; +use crate::stage::ProtocolStage; +use crate::{refs, sigrefs, transport, Handle}; + +/// The data size limit, 5Mb, while fetching the special refs, +/// i.e. `rad/id` and `rad/sigrefs`. +pub const DEFAULT_FETCH_SPECIAL_REFS_LIMIT: u64 = 1024 * 1024 * 5; +/// The data size limit, 5Gb, while fetching the data refs, +/// i.e. `refs/heads`, `refs/tags`, `refs/cobs`, etc. +pub const DEFAULT_FETCH_DATA_REFS_LIMIT: u64 = 1024 * 1024 * 1024 * 5; + +pub mod error { + use std::io; + + use thiserror::Error; + + use crate::{git, git::repository, handle, sigrefs, stage}; + + #[derive(Debug, Error)] + pub enum Step { + #[error(transparent)] + Io(#[from] io::Error), + #[error(transparent)] + Layout(#[from] stage::error::Layout), + #[error(transparent)] + Prepare(#[from] stage::error::Prepare), + #[error(transparent)] + WantsHaves(#[from] stage::error::WantsHaves), + } + + #[derive(Debug, Error)] + pub enum Protocol { + #[error(transparent)] + Canonical(#[from] Canonical), + #[error(transparent)] + Io(#[from] io::Error), + #[error("canonical 'refs/rad/id' is missing")] + MissingRadId, + #[error(transparent)] + RefdbUpdate(#[from] repository::error::Update), + #[error(transparent)] + Resolve(#[from] repository::error::Resolve), + #[error(transparent)] + Refs(#[from] radicle::storage::refs::Error), + #[error(transparent)] + RemoteRefs(#[from] sigrefs::error::RemoteRefs), + #[error(transparent)] + Step(#[from] Step), + #[error(transparent)] + Tracking(#[from] handle::error::Tracking), + #[error(transparent)] + Validation(#[from] radicle::storage::Error), + } + + #[derive(Debug, Error)] + pub enum Canonical { + #[error(transparent)] + Resolve(#[from] git::repository::error::Resolve), + #[error(transparent)] + Verified(#[from] radicle::identity::DocError), + } +} + +type IdentityTips = BTreeMap; +type SigrefTips = BTreeMap; + +#[derive(Clone, Copy, Debug)] +pub struct FetchLimit { + pub special: u64, + pub refs: u64, +} + +impl Default for FetchLimit { + fn default() -> Self { + Self { + special: DEFAULT_FETCH_SPECIAL_REFS_LIMIT, + refs: DEFAULT_FETCH_DATA_REFS_LIMIT, + } + } +} + +#[derive(Debug)] +pub enum FetchResult { + Success { + /// The set of applied changes to the reference store. + applied: Applied<'static>, + /// The set of namespaces that were fetched. + remotes: BTreeSet, + /// Validation errors that were found while fetching for + /// **non-delegate** remotes. + warnings: sigrefs::Validations, + }, + Failed { + /// Validation errors that were found while fetching for + /// **non-delegate** remotes. + warnings: sigrefs::Validations, + /// Validation errors that were found while fetching for + /// **delegate** remotes. + failures: sigrefs::Validations, + }, +} + +impl FetchResult { + pub fn rejected(&self) -> impl Iterator> { + match self { + Self::Success { applied, .. } => either::Either::Left(applied.rejected.iter()), + Self::Failed { .. } => either::Either::Right(std::iter::empty()), + } + } + + pub fn warnings(&self) -> impl Iterator { + match self { + Self::Success { warnings, .. } => warnings.iter(), + Self::Failed { warnings, .. } => warnings.iter(), + } + } + + pub fn is_success(&self) -> bool { + match self { + Self::Success { .. } => true, + Self::Failed { .. } => false, + } + } +} + +#[derive(Default)] +pub struct FetchState { + /// In-memory refdb used to keep track of new updates without + /// committing them to the real refdb until all validation has + /// occurred. + refs: git::mem::Refdb, + /// Have we seen the `rad/id` reference? + canonical_rad_id: Option, + /// Seen remote `rad/id` tips. + ids: IdentityTips, + /// Seen remote `rad/sigrefs` tips. + sigrefs: SigrefTips, + /// Seen reference tips, per remote. + tips: BTreeMap>>, +} + +impl FetchState { + /// Remove all tips associated with this `remote` in the + /// `FetchState`. + pub fn prune(&mut self, remote: &PublicKey) { + self.ids.remove(remote); + self.sigrefs.remove(remote); + self.tips.remove(remote); + } + + pub fn canonical_rad_id(&self) -> Option<&Oid> { + self.canonical_rad_id.as_ref() + } + + /// Update the in-memory refdb with the given updates while also + /// keeping track of the updates in [`FetchState::tips`]. + pub fn update_all<'a, I>(&mut self, other: I) -> Applied<'a> + where + I: IntoIterator>)>, + { + let mut ap = Applied::default(); + for (remote, ups) in other { + for up in &ups { + ap.append(&mut self.refs.update(Some(up.clone()))); + } + let mut ups = ups + .into_iter() + .map(|up| up.into_owned()) + .collect::>(); + self.tips + .entry(remote) + .and_modify(|tips| tips.append(&mut ups)) + .or_insert(ups); + } + ap + } + + pub(crate) fn as_cached<'a, S>(&'a mut self, handle: &'a mut Handle) -> Cached<'a, S> { + Cached { + handle, + state: self, + } + } +} + +impl FetchState { + /// Perform the ls-refs and fetch for the given `step`. The result + /// of these processes is kept track of in the internal state. + pub(super) fn run_stage( + &mut self, + handle: &mut Handle, + handshake: &handshake::Outcome, + step: &F, + ) -> Result, error::Step> + where + S: transport::ConnectionStream, + F: ProtocolStage, + { + let refs = match step.ls_refs() { + Some(refs) => handle + .transport + .ls_refs(refs.into(), handshake)? + .into_iter() + .filter_map(|r| step.ref_filter(r)) + .collect::>(), + None => vec![], + }; + log::trace!(target: "fetch", "Received refs {:?}", refs); + step.pre_validate(&refs)?; + + let wants_haves = step.wants_haves(&handle.repo, &refs)?; + if !wants_haves.wants.is_empty() { + handle + .transport + .fetch(wants_haves, handle.interrupt.clone(), handshake)?; + } else { + log::trace!(target: "fetch", "Nothing to fetch") + }; + + let mut fetched = BTreeSet::new(); + for r in &refs { + match &r.name { + refs::ReceivedRefname::Namespaced { remote, suffix } => { + fetched.insert(*remote); + if let Some(rad) = suffix.as_ref().left() { + match rad { + refs::Special::Id => { + self.ids.insert(*remote, r.tip); + } + + refs::Special::SignedRefs => { + self.sigrefs.insert(*remote, r.tip); + } + } + } + } + refs::ReceivedRefname::RadId => self.canonical_rad_id = Some(r.tip), + } + } + + let up = step.prepare_updates(self, &handle.repo, &refs)?; + self.update_all(up.tips.into_iter()); + + Ok(fetched) + } + + /// The finalization of the protocol exchange is as follows: + /// + /// 1. Load the canonical `rad/id` to use as the anchor for + /// getting the delegates of the identity. + /// 2. Calculate the trusted set of peers for fetching from. + /// 3. Fetch the special references, i.e. `rad/id` and `rad/sigrefs`. + /// 4. Load the signed references, where these signed references + /// must be cryptographically verified for delegates, + /// otherwise they are discarded for non-delegates. + /// 5. Fetch the data references, i.e. references found in + /// `rad/sigrefs`. + /// 6. Validate the fetched references for delegates and + /// non-delegates, pruning any invalid remotes from the set + /// of updating tips. + /// 7. Apply the valid tips, iff no delegates failed validation. + /// 8. Signal to the other side that the process has completed. + pub(super) fn run( + mut self, + handle: &mut Handle, + handshake: &handshake::Outcome, + limit: FetchLimit, + remote: PublicKey, + ) -> Result + where + S: transport::ConnectionStream, + { + // N.b. The error case here should not happen. In the case of + // a `clone` we have asked for refs/rad/id and ensured it was + // fetched. In the case of `pull` the repository should have + // the refs/rad/id set. + let anchor = self + .as_cached(handle) + .canonical()? + .ok_or(error::Protocol::MissingRadId)?; + + // TODO: not sure we should allow to block *any* peer from the + // delegate set. We could end up ignoring delegates. + let delegates = anchor + .delegates + .iter() + .filter(|id| !handle.is_blocked(id)) + .map(|did| PublicKey::from(*did)) + .collect::>(); + + log::trace!(target: "fetch", "Identity delegates {delegates:?}"); + + let tracked = handle.tracked(); + log::trace!(target: "fetch", "Tracked nodes {:?}", tracked); + + let special_refs = stage::SpecialRefs { + blocked: handle.blocked.clone(), + remote, + delegates: delegates.clone(), + tracked, + limit: limit.special, + }; + log::trace!(target: "fetch", "{special_refs:?}"); + let fetched = self.run_stage(handle, handshake, &special_refs)?; + + let signed_refs = sigrefs::RemoteRefs::load( + &self.as_cached(handle), + sigrefs::Select { + must: &delegates, + may: &fetched + .iter() + .filter(|id| !delegates.contains(id)) + .copied() + .collect(), + }, + )?; + log::trace!(target: "fetch", "{signed_refs:?}"); + + let data_refs = stage::DataRefs { + remote, + remotes: signed_refs, + limit: limit.refs, + }; + self.run_stage(handle, handshake, &data_refs)?; + + // Run validation of signed refs, pruning any offending + // remotes from the tips, thus not updating the production Git + // repository. + // N.b. any delegate validation errors are added to + // `failures`, while any non-delegate validation errors are + // added to `warnings`. + let mut warnings = sigrefs::Validations::default(); + let mut failures = sigrefs::Validations::default(); + let signed_refs = data_refs.remotes; + + for remote in signed_refs.keys() { + if handle.is_blocked(remote) { + continue; + } + + let remote = sigrefs::DelegateStatus::empty(*remote, &delegates); + match remote.load(&self.as_cached(handle))? { + sigrefs::DelegateStatus::NonDelegate { remote, data: None } => { + log::debug!(target: "fetch", "Pruning non-delegate {remote} tips, missing 'rad/sigrefs'"); + warnings.push(sigrefs::Validation::MissingRadSigRefs(remote)); + self.prune(&remote) + } + sigrefs::DelegateStatus::Delegate { remote, data: None } => { + log::warn!(target: "fetch", "Pruning delegate {remote} tips, missing 'rad/sigrefs'"); + failures.push(sigrefs::Validation::MissingRadSigRefs(remote)); + self.prune(&remote) + } + sigrefs::DelegateStatus::NonDelegate { + remote, + data: Some(sigrefs), + } => { + let cache = self.as_cached(handle); + if let Some(warns) = sigrefs::validate(&cache, sigrefs)?.as_mut() { + log::debug!( + target: "fetch", + "Pruning non-delegate {remote} tips, due to validation failures" + ); + self.prune(&remote); + warnings.append(warns); + } + } + sigrefs::DelegateStatus::Delegate { + remote, + data: Some(sigrefs), + } => { + let cache = self.as_cached(handle); + if let Some(fails) = sigrefs::validate(&cache, sigrefs)?.as_mut() { + log::warn!(target: "fetch", "Pruning delegate {remote} tips, due to validation failures"); + self.prune(&remote); + failures.append(fails) + } + } + } + } + + // N.b. signal to exit the upload-pack sequence + handle.transport.done()?; + + // N.b. only apply to Git repository if no delegates have failed verification. + if failures.is_empty() { + let applied = repository::update( + &handle.repo, + self.tips + .clone() + .into_values() + .flat_map(|ups| ups.into_iter()), + )?; + Ok(FetchResult::Success { + applied, + remotes: fetched, + warnings, + }) + } else { + Ok(FetchResult::Failed { warnings, failures }) + } + } +} + +/// A cached version of [`Handle`] by using the underlying +/// [`FetchState`]'s data for performing lookups. +pub(crate) struct Cached<'a, S> { + handle: &'a mut Handle, + state: &'a mut FetchState, +} + +impl<'a, S> Cached<'a, S> { + /// Resolves `refname` to its [`ObjectId`] by first looking at the + /// [`FetchState`] and falling back to the [`Handle::refdb`]. + pub fn refname_to_id<'b, N>( + &self, + refname: N, + ) -> Result, repository::error::Resolve> + where + N: Into>, + { + let refname = refname.into(); + match self.state.refs.refname_to_id(refname.clone()) { + None => repository::refname_to_id(&self.handle.repo, refname), + Some(oid) => Ok(Some(oid)), + } + } + + /// Get the `rad/id` found in the [`FetchState`]. + pub fn canonical_rad_id(&self) -> Option { + self.state.canonical_rad_id().copied() + } + + pub fn verified(&self, head: Oid) -> Result, DocError> { + self.handle.verified(head) + } + + pub fn canonical(&self) -> Result>, error::Canonical> { + let tip = self.refname_to_id(refs::REFS_RAD_ID.clone())?; + let cached_tip = self.canonical_rad_id(); + + tip.or(cached_tip) + .map(|tip| self.verified(tip).map_err(error::Canonical::from)) + .transpose() + } + + pub fn load(&self, remote: &PublicKey) -> Result, sigrefs::error::Load> { + match self.state.sigrefs.get(remote) { + None => SignedRefsAt::load(*remote, &self.handle.repo), + Some(tip) => SignedRefsAt::load_at(*tip, *remote, &self.handle.repo).map(Some), + } + } +} + +impl<'a, S> RemoteRepository for Cached<'a, S> { + fn remote(&self, remote: &RemoteId) -> Result { + // N.b. this is unused so we just delegate to the underlying + // repository for a correct implementation. + self.handle.repo.remote(remote) + } + + fn remotes(&self) -> Result, storage::refs::Error> { + self.state + .sigrefs + .keys() + .map(|id| self.remote(id).map(|remote| (*id, remote))) + .collect::>() + } +} + +impl<'a, S> ValidateRepository for Cached<'a, S> { + // N.b. we don't verify the `rad/id` of each remote since they may + // not have a reference to the COB if they have not interacted + // with it. + fn validate_remote(&self, remote: &Remote) -> Result { + // Contains a copy of the signed refs of this remote. + let mut signed = BTreeMap::from((*remote.refs).clone()); + let mut validations = Validations::default(); + let mut has_sigrefs = false; + + // Check all repository references, making sure they are present in the signed refs map. + for (refname, oid) in self.state.refs.references_of(&remote.id) { + // Skip validation of the signed refs branch, as it is not part of `Remote`. + if refname == storage::refs::SIGREFS_BRANCH.to_ref_string() { + has_sigrefs = true; + continue; + } + if let Some(signed_oid) = signed.remove(&refname) { + if oid != signed_oid { + validations.push(Validation::MismatchedRef { + refname, + expected: signed_oid, + actual: oid, + }); + } + } else { + validations.push(Validation::UnsignedRef(refname)); + } + } + + if !has_sigrefs { + validations.push(Validation::MissingRadSigRefs(remote.id)); + } + + // The refs that are left in the map, are ones that were signed, but are not + // in the repository. If any are left, bail. + for (name, _) in signed.into_iter() { + validations.push(Validation::MissingRef { + refname: name, + remote: remote.id, + }); + } + + Ok(validations) + } +} diff --git a/radicle-fetch/src/tracking.rs b/radicle-fetch/src/tracking.rs new file mode 100644 index 00000000..00fbb272 --- /dev/null +++ b/radicle-fetch/src/tracking.rs @@ -0,0 +1,108 @@ +use std::collections::HashSet; + +use radicle::crypto::PublicKey; +use radicle::node::tracking::config::Config; +use radicle::node::tracking::store::Read; +use radicle::prelude::Id; + +pub use radicle::node::tracking::{Policy, Scope}; + +#[derive(Clone, Debug)] +pub enum Tracked { + All, + Trusted { remotes: HashSet }, +} + +impl Tracked { + pub fn from_config(rid: Id, config: &Config) -> Result { + let entry = config + .repo_policy(&rid) + .map_err(|err| error::Tracking::FailedPolicy { rid, err })?; + match entry.policy { + Policy::Block => { + log::error!(target: "fetch", "Attempted to fetch untracked repo {rid}"); + Err(error::Tracking::BlockedPolicy { rid }) + } + Policy::Track => match entry.scope { + Scope::All => Ok(Self::All), + Scope::Trusted => { + let nodes = config + .node_policies() + .map_err(|err| error::Tracking::FailedNodes { rid, err })?; + let trusted: HashSet<_> = nodes + .filter_map(|node| (node.policy == Policy::Track).then_some(node.id)) + .collect(); + + Ok(Tracked::Trusted { remotes: trusted }) + } + }, + } + } +} + +/// A set of [`PublicKey`]s to ignore when fetching from a remote. +#[derive(Clone, Debug)] +pub struct BlockList(HashSet); + +impl FromIterator for BlockList { + fn from_iter>(iter: T) -> Self { + Self(iter.into_iter().collect()) + } +} + +impl Extend for BlockList { + fn extend>(&mut self, iter: T) { + self.0.extend(iter) + } +} + +impl BlockList { + pub fn is_blocked(&self, key: &PublicKey) -> bool { + self.0.contains(key) + } + + pub fn from_config(config: &Config) -> Result { + Ok(config + .node_policies()? + .filter_map(|entry| (entry.policy == Policy::Block).then_some(entry.id)) + .collect()) + } +} + +pub mod error { + use radicle::node::tracking; + use radicle::prelude::Id; + use radicle::storage; + use thiserror::Error; + + #[derive(Debug, Error)] + #[error(transparent)] + pub struct Blocked(#[from] tracking::config::Error); + + #[derive(Debug, Error)] + pub enum Tracking { + #[error("failed to find tracking policy for {rid}")] + FailedPolicy { + rid: Id, + #[source] + err: tracking::store::Error, + }, + #[error("cannot fetch {rid} as it is not tracked")] + BlockedPolicy { rid: Id }, + #[error("failed to get tracking nodes for {rid}")] + FailedNodes { + rid: Id, + #[source] + err: tracking::store::Error, + }, + + #[error(transparent)] + Storage(#[from] storage::Error), + + #[error(transparent)] + Git(#[from] radicle::git::raw::Error), + + #[error(transparent)] + Refs(#[from] storage::refs::Error), + } +} diff --git a/radicle-fetch/src/transport.rs b/radicle-fetch/src/transport.rs new file mode 100644 index 00000000..75cc7092 --- /dev/null +++ b/radicle-fetch/src/transport.rs @@ -0,0 +1,351 @@ +pub(crate) mod fetch; +pub(crate) mod ls_refs; + +use std::collections::BTreeSet; +use std::io; +use std::path::PathBuf; +use std::sync::atomic::AtomicBool; +use std::sync::Arc; + +use bstr::BString; +use gix_features::progress::prodash::progress; +use gix_protocol::handshake; +use gix_protocol::FetchConnection; +use gix_transport::client; +use gix_transport::client::TransportWithoutIO as _; +use gix_transport::Protocol; +use gix_transport::Service; +use radicle::git::Oid; +use radicle::git::Qualified; +use radicle::storage::git::Repository; +use thiserror::Error; + +use crate::git::oid; +use crate::git::repository; + +/// Open a reader and writer stream to pass to the ls-refs and fetch +/// processes for communicating during their respective protocols. +pub trait ConnectionStream { + type Read: io::Read; + type Write: io::Write + SignalEof; + type Error: std::error::Error + Send + Sync + 'static; + + fn open(&mut self) -> Result<(&mut Self::Read, &mut Self::Write), Self::Error>; +} + +/// The ability to signal EOF to the server side so that it can stop +/// serving for this fetch request. +pub trait SignalEof { + type Error: std::error::Error + Send + Sync + 'static; + + /// Since the git protocol is tunneled over an existing + /// connection, we can't signal the end of the protocol via the + /// usual means, which is to close the connection. Git also + /// doesn't have any special message we can send to signal the end + /// of the protocol. + /// + /// Hence, there's no other way for the server to know that we're + /// done sending requests than to send a special message outside + /// the git protocol. This message can then be processed by the + /// remote worker to end the protocol. We use the special "eof" + /// control message for this. + fn eof(&mut self) -> Result<(), Self::Error>; +} + +/// Configuration for running a Git `handshake`, `ls-refs`, or +/// `fetch`. +pub struct Transport { + git_dir: PathBuf, + repo: BString, + stream: S, +} + +impl Transport +where + S: ConnectionStream, +{ + pub fn new(git_dir: PathBuf, mut repo: BString, stream: S) -> Self { + let repo = if repo.starts_with(b"/") { + repo + } else { + let mut path = BString::new(b"/".to_vec()); + path.append(&mut repo); + path + }; + Self { + git_dir, + repo, + stream, + } + } + + /// Perform the handshake with the server side. + pub(crate) fn handshake(&mut self) -> io::Result { + log::trace!(target: "fetch", "Performing handshake for {}", self.repo); + let (read, write) = self.stream.open().map_err(io_other)?; + gix_protocol::fetch::handshake( + &mut Connection::new(read, write, FetchConnection::AllowReuse, self.repo.clone()), + |_| Ok(None), + vec![], + &mut progress::Discard, + ) + .map_err(io_other) + } + + /// Perform ls-refs with the server side. + pub(crate) fn ls_refs( + &mut self, + mut prefixes: Vec, + handshake: &handshake::Outcome, + ) -> io::Result> { + prefixes.sort(); + prefixes.dedup(); + let (read, write) = self.stream.open().map_err(io_other)?; + ls_refs::run( + ls_refs::Config { + prefixes, + extra_params: vec![], + repo: self.repo.clone(), + }, + handshake, + Connection::new(read, write, FetchConnection::AllowReuse, self.repo.clone()), + &mut progress::Discard, + ) + .map_err(io_other) + } + + /// Perform the fetch with the server side. + pub(crate) fn fetch( + &mut self, + wants_haves: WantsHaves, + interrupt: Arc, + handshake: &handshake::Outcome, + ) -> io::Result<()> { + log::trace!( + target: "fetch", + "Running fetch wants={:?}, haves={:?}", + wants_haves.wants, + wants_haves.haves + ); + let out = { + let (read, write) = self.stream.open().map_err(io_other)?; + fetch::run( + wants_haves.clone(), + fetch::PackWriter { + git_dir: self.git_dir.clone(), + interrupt, + }, + handshake, + Connection::new(read, write, FetchConnection::AllowReuse, self.repo.clone()), + &mut progress::Discard, + ) + .map_err(io_other)? + }; + let pack_path = out + .pack + .ok_or_else(|| { + io::Error::new( + io::ErrorKind::UnexpectedEof, + "empty or no packfile received", + ) + })? + .index_path + .expect("written packfile must have a path"); + + // Validate we got all requested tips in the pack + // + // N.b. the lookup is a binary search so is efficient for + // searching any given oid. + { + use gix_pack::index::File; + + let idx = File::at(&pack_path, gix_hash::Kind::Sha1).map_err(io_other)?; + for oid in wants_haves.wants { + if idx.lookup(oid::to_object_id(oid)).is_none() { + return Err(io::Error::new( + io::ErrorKind::NotFound, + format!("wanted {oid} not found in pack"), + )); + } + } + } + + Ok(()) + } + + /// Signal to the server side that we are done sending ls-refs and + /// fetch commands. + pub(crate) fn done(&mut self) -> io::Result<()> { + let (_, w) = self.stream.open().map_err(io_other)?; + w.eof().map_err(io_other) + } +} + +pub(crate) struct Connection { + inner: client::git::Connection, + mode: FetchConnection, +} + +impl Connection +where + R: io::Read, + W: io::Write, +{ + pub fn new(read: R, write: W, mode: FetchConnection, repo: BString) -> Self { + Self { + inner: client::git::Connection::new( + read, + write, + Protocol::V2, + repo, + None::<(String, Option)>, + client::git::ConnectMode::Daemon, + ), + mode, + } + } +} + +impl client::Transport for Connection +where + R: std::io::Read, + W: std::io::Write, +{ + fn handshake<'b>( + &mut self, + service: Service, + extra_parameters: &'b [(&'b str, Option<&'b str>)], + ) -> Result, client::Error> { + self.inner.handshake(service, extra_parameters) + } +} + +impl client::TransportWithoutIO for Connection +where + R: std::io::Read, + W: std::io::Write, +{ + fn request( + &mut self, + write_mode: client::WriteMode, + on_into_read: client::MessageKind, + ) -> Result, client::Error> { + self.inner.request(write_mode, on_into_read) + } + + fn to_url(&self) -> std::borrow::Cow<'_, bstr::BStr> { + self.inner.to_url() + } + + fn connection_persists_across_multiple_requests(&self) -> bool { + false + } + + fn configure( + &mut self, + config: &dyn std::any::Any, + ) -> Result<(), Box> { + self.inner.configure(config) + } + + fn supported_protocol_versions(&self) -> &[Protocol] { + &[Protocol::V2] + } +} + +fn indicate_end_of_interaction(transport: &mut Connection) -> Result<(), client::Error> +where + R: io::Read, + W: io::Write, +{ + // An empty request marks the (early) end of the interaction. Only relevant in stateful transports though. + if transport.connection_persists_across_multiple_requests() { + transport + .request(client::WriteMode::Binary, client::MessageKind::Flush)? + .into_read()?; + } + Ok(()) +} + +fn io_other(err: impl std::error::Error + Send + Sync + 'static) -> io::Error { + io::Error::new(io::ErrorKind::Other, err) +} + +#[derive(Debug, Error)] +pub enum WantsHavesError { + #[error(transparent)] + Ancestry(#[from] repository::error::Ancestry), + #[error(transparent)] + Contains(#[from] repository::error::Contains), + #[error(transparent)] + Resolve(#[from] repository::error::Resolve), +} + +#[derive(Clone, Default)] +pub(crate) struct WantsHaves { + pub wants: BTreeSet, + pub haves: BTreeSet, +} + +impl WantsHaves { + pub fn want(&mut self, oid: Oid) { + // N.b. if we have it, then we don't want it. + if !self.haves.contains(&oid) { + self.wants.insert(oid); + } + } + + pub fn have(&mut self, oid: Oid) { + // N.b. ensure that oid is not in wants + self.wants.remove(&oid); + self.haves.insert(oid); + } + + /// Add a set of references to the `wants` and `haves`. + /// + /// For each reference we want to build the range between its + /// current `Oid` and the advertised `Oid`. This allows the server + /// to send all objects between that range. + /// + /// If the reference exists, the range is given by marking the + /// existing `Oid` as a `have` and the tip as the `want`. If the + /// `tip`, however, is the same as the existing `Oid` or is in the + /// Odb, then there is no need to mark it as a `want`. + /// + /// If the reference does not exist, the range is simply marking + /// the tip as a `want`, iff it does not already exist in the Odb. + pub fn add<'a, N>( + &mut self, + repo: &Repository, + refs: impl IntoIterator, + ) -> Result<&mut Self, WantsHavesError> + where + N: Into>, + { + refs.into_iter().try_fold(self, |acc, (refname, tip)| { + match repository::refname_to_id(repo, refname)? { + Some(oid) => { + let want = oid != tip && !repository::contains(repo, tip)?; + acc.have(oid); + + if want { + acc.want(tip) + } + } + None => { + if !repository::contains(repo, tip)? { + acc.want(tip); + } + } + }; + Ok(acc) + }) + } +} + +fn agent_name() -> io::Result { + Ok(format!( + "git/{}", + radicle::git::version().map_err(io_other)? + )) +} diff --git a/radicle-fetch/src/transport/fetch.rs b/radicle-fetch/src/transport/fetch.rs new file mode 100644 index 00000000..4857154c --- /dev/null +++ b/radicle-fetch/src/transport/fetch.rs @@ -0,0 +1,308 @@ +use std::{ + borrow::Cow, + io::{self, BufRead}, + path::PathBuf, + sync::{atomic::AtomicBool, Arc}, +}; + +use gix_features::progress::Progress; +use gix_pack as pack; +use gix_protocol::{ + fetch::{self, Delegate, DelegateBlocking}, + handshake::{self, Ref}, + ls_refs, FetchConnection, +}; +use gix_transport::{bstr::BString, client, Protocol}; + +use super::{agent_name, indicate_end_of_interaction, Connection, WantsHaves}; + +pub type Error = gix_protocol::fetch::Error; + +pub mod error { + use std::io; + + use thiserror::Error; + + #[derive(Debug, Error)] + pub enum PackWriter { + #[error(transparent)] + Io(#[from] io::Error), + #[error(transparent)] + Write(#[from] gix_pack::bundle::write::Error), + } +} + +/// Configuration for writing a packfile. +pub struct PackWriter { + /// The repository path for writing the packfile to. Note this is + /// the root of the Git repository, e.g. the `.git` folder. + pub git_dir: PathBuf, + /// `interrupt` is checked regularly and when true, the whole + /// operation will stop. + pub interrupt: Arc, +} + +impl PackWriter { + /// Write the packfile read from `pack` to the `objects/pack` + /// directory. + pub fn write_pack( + &self, + pack: impl BufRead, + progress: impl Progress, + ) -> Result { + use gix_odb::FindExt as _; + + let options = pack::bundle::write::Options { + // N.b. use all cores. Can make configurable if needed + // later. + thread_limit: None, + iteration_mode: pack::data::input::Mode::Verify, + index_version: pack::index::Version::V2, + object_hash: gix_hash::Kind::Sha1, + }; + let odb_opts = gix_odb::store::init::Options { + slots: gix_odb::store::init::Slots::default(), + object_hash: gix_hash::Kind::Sha1, + use_multi_pack_index: true, + current_dir: Some(self.git_dir.clone()), + }; + let thickener = Arc::new(gix_odb::Store::at_opts( + self.git_dir.join("objects"), + [], + odb_opts, + )?); + let thickener = thickener.to_handle_arc(); + Ok(pack::Bundle::write_to_directory( + pack, + Some(self.git_dir.join("objects").join("pack")), + progress, + &self.interrupt, + Some(Box::new(move |oid, buf| thickener.find(oid, buf).ok())), + options, + )?) + } +} + +/// The fetch [`Delegate`] that negotiates the fetch with the +/// server-side. +pub struct Fetch { + wants_haves: WantsHaves, + pack_writer: PackWriter, + out: FetchOut, +} + +/// The result of running a fetch via [`run`]. +pub struct FetchOut { + pub refs: Vec, + pub pack: Option, +} + +// FIXME: the delegate pattern will be removed in the near future and +// we should look at the fetch code being used in gix to see how we +// can migrate to the proper form of fetching. +impl<'a> Delegate for &'a mut Fetch { + fn receive_pack( + &mut self, + input: impl io::BufRead, + progress: impl Progress, + _refs: &[handshake::Ref], + previous_response: &fetch::Response, + ) -> io::Result<()> { + self.out + .refs + .extend(previous_response.wanted_refs().iter().map( + |fetch::response::WantedRef { id, path }| Ref::Direct { + full_ref_name: path.clone(), + object: *id, + }, + )); + let pack = self + .pack_writer + .write_pack(input, progress) + .map_err(|err| io::Error::new(io::ErrorKind::Other, err))?; + self.out.pack = Some(pack); + Ok(()) + } +} + +impl<'a> DelegateBlocking for &'a mut Fetch { + fn negotiate( + &mut self, + _refs: &[handshake::Ref], + arguments: &mut fetch::Arguments, + _previous_response: Option<&fetch::Response>, + ) -> io::Result { + use crate::git::oid; + + for oid in &self.wants_haves.wants { + arguments.want(oid::to_object_id(*oid)); + } + + for oid in &self.wants_haves.haves { + arguments.have(oid::to_object_id(*oid)); + } + + // N.b. sends `done` packet + Ok(fetch::Action::Cancel) + } + + fn prepare_ls_refs( + &mut self, + _server: &client::Capabilities, + _arguments: &mut Vec, + _features: &mut Vec<(&str, Option>)>, + ) -> io::Result { + // N.b. we performed ls-refs before the fetch already. + Ok(ls_refs::Action::Skip) + } + + fn prepare_fetch( + &mut self, + _version: Protocol, + _server: &client::Capabilities, + _features: &mut Vec<(&str, Option>)>, + _refs: &[handshake::Ref], + ) -> io::Result { + if self.wants_haves.wants.is_empty() { + return Err(io::Error::new(io::ErrorKind::InvalidData, "empty fetch")); + } + Ok(fetch::Action::Continue) + } +} + +/// Run the fetch process using the provided `config` and +/// `pack_writer` configuration. +/// +/// It is expected that the `handshake` was run outside of this +/// process, since it should be reused across fetch processes. +#[allow(clippy::result_large_err)] +pub(crate) fn run( + wants_haves: WantsHaves, + pack_writer: PackWriter, + handshake: &handshake::Outcome, + mut conn: Connection, + progress: &mut P, +) -> Result +where + P: Progress, + P::SubProgress: 'static, + R: io::Read, + W: io::Write, +{ + log::trace!(target: "fetch", "Performing fetch"); + + let mut delegate = Fetch { + wants_haves, + pack_writer, + out: FetchOut { + refs: Vec::new(), + pack: None, + }, + }; + + let handshake::Outcome { + server_protocol_version: protocol, + refs: _refs, + capabilities, + } = handshake; + let agent = agent_name()?; + let fetch = gix_protocol::Command::Fetch; + + let mut features = fetch.default_features(*protocol, capabilities); + match (&mut delegate).prepare_fetch(*protocol, capabilities, &mut features, &[]) { + Ok(fetch::Action::Continue) => { + // FIXME: this is a private function in gitoxide + // fetch.validate_argument_prefixes_or_panic(protocol, &capabilities, &[], &features) + } + // N.b. we always return Action::Continue + Ok(fetch::Action::Cancel) => unreachable!(), + Err(err) => { + indicate_end_of_interaction(&mut conn)?; + return Err(err.into()); + } + } + + gix_protocol::fetch::Response::check_required_features(*protocol, &features)?; + let sideband_all = features.iter().any(|(n, _)| *n == "sideband-all"); + features.push(("agent", Some(Cow::Owned(agent)))); + let mut args = fetch::Arguments::new(*protocol, features); + + let mut previous_response = None::; + let mut round = 1; + 'negotiation: loop { + progress.step(); + progress.set_name(format!("negotiate (round {round})")); + round += 1; + let action = (&mut delegate).negotiate(&[], &mut args, previous_response.as_ref())?; + let mut reader = args.send(&mut conn, action == fetch::Action::Cancel)?; + if sideband_all { + setup_remote_progress(progress, &mut reader); + } + let response = fetch::Response::from_line_reader(*protocol, &mut reader, true)?; + previous_response = if response.has_pack() { + progress.step(); + progress.set_name("receiving pack"); + if !sideband_all { + setup_remote_progress(progress, &mut reader); + } + (&mut delegate).receive_pack(reader, progress, &[], &response)?; + break 'negotiation; + } else { + match action { + fetch::Action::Cancel => break 'negotiation, + fetch::Action::Continue => Some(response), + } + } + } + if matches!(protocol, Protocol::V2) + && matches!(conn.mode, FetchConnection::TerminateOnSuccessfulCompletion) + { + indicate_end_of_interaction(&mut conn)?; + } + + // N.b. the flush packet is never read in the packfile parser. To + // remedy this, we read from our connection until we see it ensure + // that there is no leftover data. + let (mut r, _) = conn.inner.into_inner(); + let mut buf = [0; 4096]; + + loop { + match r.read(&mut buf)? { + 0 => break, + n => match std::str::from_utf8(&buf[..n]) { + Ok(progress) => { + let lines = progress.split('\n'); + if lines.into_iter().any(|line| line == "0000") { + break; + } + } + Err(e) => { + return Err(std::io::Error::new( + std::io::ErrorKind::Other, + format!("found leftover packfile bytes: {e}"), + ) + .into()); + } + }, + } + } + + log::trace!(target: "fetch", "fetched refs: {:?}", delegate.out.refs); + Ok(delegate.out) +} + +fn setup_remote_progress

( + progress: &mut P, + reader: &mut Box, +) where + P: Progress, + P::SubProgress: 'static, +{ + reader.set_progress_handler(Some(Box::new({ + let mut remote_progress = progress.add_child("remote"); + move |is_err: bool, data: &[u8]| { + gix_protocol::RemoteProgress::translate_to_progress(is_err, data, &mut remote_progress); + gix_transport::packetline::read::ProgressAction::Continue + } + }) as gix_transport::client::HandleProgress)); +} diff --git a/radicle-fetch/src/transport/ls_refs.rs b/radicle-fetch/src/transport/ls_refs.rs new file mode 100644 index 00000000..c9cb636b --- /dev/null +++ b/radicle-fetch/src/transport/ls_refs.rs @@ -0,0 +1,175 @@ +use std::borrow::Cow; +use std::io::{self, BufRead}; + +use bstr::ByteSlice; +use gix_features::progress::Progress; +use gix_protocol::fetch::{self, Delegate, DelegateBlocking}; +use gix_protocol::handshake::{self, Ref}; +use gix_protocol::transport::Protocol; +use gix_protocol::{ls_refs, Command}; +use gix_transport::bstr::{BString, ByteVec}; +use gix_transport::client::{self, TransportV2Ext}; + +use super::{agent_name, indicate_end_of_interaction, Connection}; + +/// Configuration for running an ls-refs process. +/// +/// See [`run`]. +pub struct Config { + /// The repository name, i.e. `/`. + pub repo: BString, + /// Extra parameters to pass to the ls-refs process. + pub extra_params: Vec<(String, Option)>, + /// Ref prefixes for filtering the output of the ls-refs process. + pub prefixes: Vec, +} + +/// The Gitoxide delegate for running the ls-refs process. +struct LsRefs { + /// Configuration for the ls-refs process. + config: Config, + /// The resulting references returned by the ls-refs process. + refs: Vec, +} + +impl LsRefs { + fn new(config: Config) -> Self { + Self { + config, + refs: Vec::new(), + } + } +} + +// FIXME: the delegate pattern will be removed in the near future and +// we should look at the fetch code being used in gix to see how we +// can migrate to the proper form of fetching. +impl DelegateBlocking for LsRefs { + fn handshake_extra_parameters(&self) -> Vec<(String, Option)> { + self.config.extra_params.clone() + } + + fn prepare_ls_refs( + &mut self, + _caps: &client::Capabilities, + args: &mut Vec, + _: &mut Vec<(&str, Option>)>, + ) -> io::Result { + for prefix in &self.config.prefixes { + let mut arg = BString::from("ref-prefix "); + arg.push_str(prefix); + args.push(arg) + } + Ok(ls_refs::Action::Continue) + } + + fn prepare_fetch( + &mut self, + _: Protocol, + _: &client::Capabilities, + _: &mut Vec<(&str, Option>)>, + refs: &[Ref], + ) -> io::Result { + self.refs.extend_from_slice(refs); + Ok(fetch::Action::Cancel) + } + + fn negotiate( + &mut self, + _: &[Ref], + _: &mut fetch::Arguments, + _: Option<&fetch::Response>, + ) -> io::Result { + unreachable!("`negotiate` called even though no `fetch` command was sent") + } +} + +impl Delegate for LsRefs { + fn receive_pack( + &mut self, + _: impl BufRead, + _: impl Progress, + _: &[Ref], + _: &fetch::Response, + ) -> io::Result<()> { + unreachable!("`receive_pack` called even though no `fetch` command was sent") + } +} + +/// Run the ls-refs process using the provided `config`. +/// +/// It is expected that the `handshake` was run outside of this +/// process, since it should be reused across fetch processes. +/// +/// The resulting set of references are the ones returned by the +/// ls-refs process, filtered by any prefixes that were provided by +/// the `config`. +pub(crate) fn run( + config: Config, + handshake: &handshake::Outcome, + mut conn: Connection, + progress: &mut impl Progress, +) -> Result, ls_refs::Error> +where + R: io::Read, + W: io::Write, +{ + log::trace!(target: "fetch", "Performing ls-refs: {:?}", config.prefixes); + let mut delegate = LsRefs::new(config); + let handshake::Outcome { + server_protocol_version: protocol, + capabilities, + .. + } = handshake; + + if protocol != &Protocol::V2 { + return Err(ls_refs::Error::Io(io::Error::new( + io::ErrorKind::Other, + "expected protocol version 2", + ))); + } + + let ls = Command::LsRefs; + let mut features = ls.default_features(Protocol::V2, capabilities); + // N.b. copied from gitoxide + let mut args = vec![ + b"symrefs".as_bstr().to_owned(), + b"peel".as_bstr().to_owned(), + ]; + if capabilities + .capability("ls-refs") + .and_then(|cap| cap.supports("unborn")) + .unwrap_or_default() + { + args.push("unborn".into()); + } + let refs = match delegate.prepare_ls_refs(capabilities, &mut args, &mut features) { + Ok(ls_refs::Action::Skip) => Vec::new(), + Ok(ls_refs::Action::Continue) => { + // FIXME: this is a private function + // ls.validate_argument_prefixes_or_panic(Protocol::V2, capabilities, &args, &features); + + let agent = agent_name()?; + features.push(("agent", Some(Cow::Owned(agent)))); + + progress.step(); + progress.set_name("list refs"); + let mut remote_refs = conn.invoke( + ls.as_str(), + features.clone().into_iter(), + if args.is_empty() { + None + } else { + Some(args.into_iter()) + }, + )?; + handshake::refs::from_v2_refs(&mut remote_refs)? + } + Err(err) => { + indicate_end_of_interaction(&mut conn)?; + return Err(err.into()); + } + }; + + Ok(refs) +} diff --git a/radicle/src/storage/refs.rs b/radicle/src/storage/refs.rs index a455da76..e26b0ab3 100644 --- a/radicle/src/storage/refs.rs +++ b/radicle/src/storage/refs.rs @@ -383,7 +383,12 @@ impl SignedRefsAt { where S: ReadRepository, { - let at = repo.reference_oid(&remote, &SIGREFS_BRANCH)?; + let at = match repo.reference_oid(&remote, &SIGREFS_BRANCH) { + Ok(at) => at, + Err(git::ext::Error::NotFound(_)) => return Ok(None), + Err(git::ext::Error::Git(e)) if git::is_not_found_err(&e) => return Ok(None), + Err(e) => return Err(e.into()), + }; Self::load_at(at, remote, repo).map(Some) } @@ -396,6 +401,10 @@ impl SignedRefsAt { at, }) } + + pub fn iter(&self) -> impl Iterator { + self.sigrefs.refs.iter() + } } impl Deref for SignedRefsAt {