radicle-fetch

The `radicle-fetch` crate introduces a method for fetching from a
server using Git protocol V2, which provides the implementation
control over performing a sequence of fetches and validation over
domain data.

The protocol is built on top of the suite of Gitoxide crates[[0]]. It
builds wrappers on top of Gitoxide's refdb, odb, and transport layers
-- as well as taking the ls-refs and fetch code straight from the code
base to simplify for the needs of this crate.

[0]: https://github.com/Byron/gitoxide

Signed-off-by: Fintan Halpenny <fintan.halpenny@gmail.com>
X-Clacks-Overhead: GNU Terry Pratchett
This commit is contained in:
Fintan Halpenny 2023-05-16 09:10:01 +01:00 committed by cloudhead
parent 87187d47e5
commit 981172aa4f
No known key found for this signature in database
20 changed files with 3668 additions and 4 deletions

605
Cargo.lock generated
View File

@ -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"

View File

@ -7,6 +7,7 @@ members = [
"radicle-crdt",
"radicle-crypto",
"radicle-dag",
"radicle-fetch",
"radicle-httpd",
"radicle-node",
"radicle-remote-helper",

24
radicle-fetch/Cargo.toml Normal file
View File

@ -0,0 +1,24 @@
[package]
name = "radicle-fetch"
license = "MIT OR Apache-2.0"
version = "0.1.0"
authors = ["Fintan Halpenny <fintan.halpenny@gmail.com>"]
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"

22
radicle-fetch/src/git.rs Normal file
View File

@ -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")
}
}

View File

@ -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<Qualified<'static>, Oid>);
impl Refdb {
pub fn refname_to_id<'a, N>(&self, refname: N) -> Option<Oid>
where
N: Into<Qualified<'a>>,
{
let name = refname.into();
self.0.get(&name).copied()
}
pub fn references_of<'a>(
&'a self,
remote: &'a PublicKey,
) -> impl Iterator<Item = (RefString, Oid)> + '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<Item = Update<'a>>,
{
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}");
}
}
}
}

View File

@ -0,0 +1,2 @@
mod update;
pub use update::{Applied, Policy, RefUpdate, Update, Updates};

View File

@ -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<Update<'a>>,
/// Set of successfully updated references.
pub updated: Vec<RefUpdate>,
}
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<PublicKey, Vec<Update<'a>>>,
}
impl<'a> Updates<'a> {
pub fn build(updates: impl IntoIterator<Item = (PublicKey, Update<'a>)>) -> Self {
let tips = updates.into_iter().fold(
BTreeMap::<_, Vec<Update<'a>>>::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<Update<'a>>) {
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<Oid, Qualified<'a>>,
},
}
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()),
},
}
}
}

View File

@ -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<bool, error::Contains> {
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<bool, error::Ancestry> {
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<Option<Oid>, error::Resolve>
where
N: Into<Qualified<'a>>,
{
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<Applied<'a>, error::Update>
where
I: IntoIterator<Item = Update<'a>>,
{
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<Either<Update<'a>, 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<Oid, Qualified<'a>>,
) -> Result<Either<Update<'a>, 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<Option<radicle::git::raw::Reference<'a>>, 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,
}),
}
}

View File

@ -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),
}

128
radicle-fetch/src/handle.rs Normal file
View File

@ -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<S> {
pub(crate) local: PublicKey,
pub(crate) repo: Repository,
pub(crate) tracked: Tracked,
pub(crate) transport: Transport<S>,
/// 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<AtomicBool>,
}
impl<S> Handle<S> {
pub fn new(
local: PublicKey,
repo: Repository,
tracked: Tracked,
blocked: BlockList,
connection: S,
) -> Result<Self, error::Init>
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<Doc<Verified>, 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),
}
}

106
radicle-fetch/src/lib.rs Normal file
View File

@ -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<dyn std::error::Error + Send + Sync + 'static>,
},
#[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<S>(
handle: &mut Handle<S>,
limit: FetchLimit,
remote: PublicKey,
) -> Result<FetchResult, Error>
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<S>(
handle: &mut Handle<S>,
limit: FetchLimit,
remote: PublicKey,
) -> Result<FetchResult, Error>
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)
}

182
radicle-fetch/src/refs.rs Normal file
View File

@ -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/<remote>/refs/rad/id`
/// * `refs/namespaces/<remote>/refs/rad/sigrefs`
/// * `refs/namespaces/<remote>/refs/heads/main`
/// * `refs/namespaces/<remote>/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<Special, Qualified<'a>>,
},
/// 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<Namespaced<'b>> {
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<BString> for ReceivedRefname<'_> {
type Error = Error;
fn try_from(value: BString) -> Result<Self, Self::Error> {
match git::parse_ref::<PublicKey>(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<F>(&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<F>(
remote: &PublicKey,
suffix: &Either<Special, Qualified>,
tip: Oid,
is_delegate: F,
) -> Option<Update<'static>>
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
},
})
}

View File

@ -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<T = ()> {
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<PublicKey>) -> Self {
Self::new((), remote, delegates)
}
}
impl<T> DelegateStatus<T> {
pub fn new(data: T, remote: PublicKey, delegates: &BTreeSet<PublicKey>) -> 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<S>(
self,
cached: &Cached<S>,
) -> Result<DelegateStatus<Option<SignedRefsAt>>, 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<U, E>(self, f: impl FnOnce(T) -> Result<U, E>) -> Result<DelegateStatus<U>, 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<Option<Validations>, radicle::storage::Error> {
let remote = radicle::storage::Remote::<radicle::crypto::Verified>::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<PublicKey, SignedRefsAt>);
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<S>(
cached: &Cached<S>,
Select { must, may }: Select,
) -> Result<Self, error::RemoteRefs> {
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<PublicKey, SignedRefsAt>;
fn deref(&self) -> &Self::Target {
&self.0
}
}
impl<'a> IntoIterator for &'a RemoteRefs {
type Item = <&'a BTreeMap<PublicKey, SignedRefsAt> as IntoIterator>::Item;
type IntoIter = <&'a BTreeMap<PublicKey, SignedRefsAt> as IntoIterator>::IntoIter;
fn into_iter(self) -> Self::IntoIter {
self.0.iter()
}
}
pub struct Select<'a> {
pub must: &'a BTreeSet<PublicKey>,
pub may: &'a BTreeSet<PublicKey>,
}

484
radicle-fetch/src/stage.rs Normal file
View File

@ -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<String>),
}
#[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<dyn std::error::Error + Send + Sync + 'static>,
},
}
#[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<NonEmpty<BString>>;
/// 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<ReceivedRef>;
/// 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<WantsHaves, error::WantsHaves> {
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<Updates<'a>, 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<NonEmpty<BString>> {
Some(NonEmpty::new(refs::REFS_RAD_ID.as_bstr().into()))
}
fn ref_filter(&self, r: Ref) -> Option<ReceivedRef> {
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<Updates<'a>, 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<PublicKey>,
/// The data limit for this stage of fetching.
pub limit: u64,
}
impl ProtocolStage for SpecialRefs {
fn ls_refs(&self) -> Option<NonEmpty<BString>> {
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<ReceivedRef> {
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<Updates<'a>, 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<NonEmpty<BString>> {
None
}
// Since we don't ask for refs, we don't need to filter them.
fn ref_filter(&self, _: Ref) -> Option<ReceivedRef> {
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<WantsHaves, error::WantsHaves> {
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<Updates<'a>, 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<PublicKey>,
blocked: &BlockList,
refs: &'a [ReceivedRef],
) -> Result<Updates<'a>, 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<T>(required: BTreeSet<T>, wants: BTreeSet<T>) -> Result<(), error::Layout>
where
T: Ord + ToString,
{
if wants.is_empty() {
return Ok(());
}
let diff = required.difference(&wants).collect::<Vec<_>>();
if diff.is_empty() {
Ok(())
} else {
Err(error::Layout::MissingRequiredRefs(
diff.into_iter().map(|ns| ns.to_string()).collect(),
))
}
}

532
radicle-fetch/src/state.rs Normal file
View File

@ -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<PublicKey, Oid>;
type SigrefTips = BTreeMap<PublicKey, Oid>;
#[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<PublicKey>,
/// 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<Item = &Update<'static>> {
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<Item = &sigrefs::Validation> {
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<Oid>,
/// Seen remote `rad/id` tips.
ids: IdentityTips,
/// Seen remote `rad/sigrefs` tips.
sigrefs: SigrefTips,
/// Seen reference tips, per remote.
tips: BTreeMap<PublicKey, Vec<Update<'static>>>,
}
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<Item = (PublicKey, Vec<Update<'a>>)>,
{
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::<Vec<_>>();
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<S>) -> 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<S, F>(
&mut self,
handle: &mut Handle<S>,
handshake: &handshake::Outcome,
step: &F,
) -> Result<BTreeSet<PublicKey>, 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::<Vec<_>>(),
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<S>(
mut self,
handle: &mut Handle<S>,
handshake: &handshake::Outcome,
limit: FetchLimit,
remote: PublicKey,
) -> Result<FetchResult, error::Protocol>
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::<BTreeSet<_>>();
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<S>,
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<Option<Oid>, repository::error::Resolve>
where
N: Into<Qualified<'b>>,
{
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<Oid> {
self.state.canonical_rad_id().copied()
}
pub fn verified(&self, head: Oid) -> Result<Doc<Verified>, DocError> {
self.handle.verified(head)
}
pub fn canonical(&self) -> Result<Option<Doc<Verified>>, 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<Option<SignedRefsAt>, 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<Remote, storage::refs::Error> {
// 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<Remotes<Verified>, storage::refs::Error> {
self.state
.sigrefs
.keys()
.map(|id| self.remote(id).map(|remote| (*id, remote)))
.collect::<Result<_, _>>()
}
}
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<Validations, storage::Error> {
// 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)
}
}

View File

@ -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<PublicKey> },
}
impl Tracked {
pub fn from_config(rid: Id, config: &Config<Read>) -> Result<Self, error::Tracking> {
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<PublicKey>);
impl FromIterator<PublicKey> for BlockList {
fn from_iter<T: IntoIterator<Item = PublicKey>>(iter: T) -> Self {
Self(iter.into_iter().collect())
}
}
impl Extend<PublicKey> for BlockList {
fn extend<T: IntoIterator<Item = PublicKey>>(&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<Read>) -> Result<BlockList, error::Blocked> {
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),
}
}

View File

@ -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<S> {
git_dir: PathBuf,
repo: BString,
stream: S,
}
impl<S> Transport<S>
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<handshake::Outcome> {
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<BString>,
handshake: &handshake::Outcome,
) -> io::Result<Vec<handshake::Ref>> {
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<AtomicBool>,
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<R, W> {
inner: client::git::Connection<R, W>,
mode: FetchConnection,
}
impl<R, W> Connection<R, W>
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<u16>)>,
client::git::ConnectMode::Daemon,
),
mode,
}
}
}
impl<R, W> client::Transport for Connection<R, W>
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::SetServiceResponse<'_>, client::Error> {
self.inner.handshake(service, extra_parameters)
}
}
impl<R, W> client::TransportWithoutIO for Connection<R, W>
where
R: std::io::Read,
W: std::io::Write,
{
fn request(
&mut self,
write_mode: client::WriteMode,
on_into_read: client::MessageKind,
) -> Result<client::RequestWriter<'_>, 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<dyn std::error::Error + Send + Sync + 'static>> {
self.inner.configure(config)
}
fn supported_protocol_versions(&self) -> &[Protocol] {
&[Protocol::V2]
}
}
fn indicate_end_of_interaction<R, W>(transport: &mut Connection<R, W>) -> 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<Oid>,
pub haves: BTreeSet<Oid>,
}
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<Item = (N, Oid)>,
) -> Result<&mut Self, WantsHavesError>
where
N: Into<Qualified<'a>>,
{
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<String> {
Ok(format!(
"git/{}",
radicle::git::version().map_err(io_other)?
))
}

View File

@ -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<AtomicBool>,
}
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<pack::bundle::write::Outcome, error::PackWriter> {
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<Ref>,
pub pack: Option<pack::bundle::write::Outcome>,
}
// 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<fetch::Action> {
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<BString>,
_features: &mut Vec<(&str, Option<Cow<'_, str>>)>,
) -> io::Result<ls_refs::Action> {
// 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<Cow<'_, str>>)>,
_refs: &[handshake::Ref],
) -> io::Result<fetch::Action> {
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<P, R, W>(
wants_haves: WantsHaves,
pack_writer: PackWriter,
handshake: &handshake::Outcome,
mut conn: Connection<R, W>,
progress: &mut P,
) -> Result<FetchOut, Error>
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::<fetch::Response>;
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<P>(
progress: &mut P,
reader: &mut Box<dyn gix_transport::client::ExtendedBufRead + Unpin + '_>,
) 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));
}

View File

@ -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. `/<rid>`.
pub repo: BString,
/// Extra parameters to pass to the ls-refs process.
pub extra_params: Vec<(String, Option<String>)>,
/// Ref prefixes for filtering the output of the ls-refs process.
pub prefixes: Vec<BString>,
}
/// 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<Ref>,
}
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<String>)> {
self.config.extra_params.clone()
}
fn prepare_ls_refs(
&mut self,
_caps: &client::Capabilities,
args: &mut Vec<BString>,
_: &mut Vec<(&str, Option<Cow<'_, str>>)>,
) -> io::Result<ls_refs::Action> {
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<Cow<'_, str>>)>,
refs: &[Ref],
) -> io::Result<fetch::Action> {
self.refs.extend_from_slice(refs);
Ok(fetch::Action::Cancel)
}
fn negotiate(
&mut self,
_: &[Ref],
_: &mut fetch::Arguments,
_: Option<&fetch::Response>,
) -> io::Result<fetch::Action> {
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<R, W>(
config: Config,
handshake: &handshake::Outcome,
mut conn: Connection<R, W>,
progress: &mut impl Progress,
) -> Result<Vec<Ref>, 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)
}

View File

@ -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<Item = (&git::RefString, &Oid)> {
self.sigrefs.refs.iter()
}
}
impl Deref for SignedRefsAt {