1use std::collections::{BTreeMap, BTreeSet};
2use std::time::Instant;
3
4use gix_protocol::Handshake;
5use radicle::crypto::PublicKey;
6use radicle::git::{Oid, fmt::Qualified};
7use radicle::identity::{Did, Doc, DocError};
8
9use radicle::storage;
10use radicle::storage::git::Repository;
11use radicle::storage::refs::{FeatureLevel, RefsAt};
12use radicle::storage::{
13 Remote, RemoteId, RemoteRepository, Remotes, ValidateRepository, Validations, git::Validation,
14};
15
16use crate::git;
17use crate::git::packfile::Keepfile;
18use crate::git::refs::{Applied, Update};
19use crate::git::repository;
20use crate::sigrefs::SignedRefs;
21use crate::stage;
22use crate::stage::ProtocolStage;
23use crate::{Handle, refs, sigrefs, transport};
24
25pub const DEFAULT_FETCH_SPECIAL_REFS_LIMIT: u64 = 1024 * 1024 * 5;
28pub const DEFAULT_FETCH_DATA_REFS_LIMIT: u64 = 1024 * 1024 * 1024 * 5;
31
32pub mod error {
33 use radicle::git::Oid;
34 use radicle::prelude::PublicKey;
35 use thiserror::Error;
36
37 use crate::{git, git::repository, handle, sigrefs, stage, transport};
38
39 #[derive(Debug, Error)]
40 pub enum Step {
41 #[error(transparent)]
42 Layout(#[from] stage::error::Layout),
43 #[error(transparent)]
44 Prepare(#[from] stage::error::Prepare),
45 #[error(transparent)]
46 WantsHaves(#[from] stage::error::WantsHaves),
47 #[error(transparent)]
48 Transport(#[from] transport::Error),
49 }
50
51 #[derive(Debug, Error)]
52 pub enum Protocol {
53 #[error(transparent)]
54 Ancestry(#[from] repository::error::Ancestry),
55 #[error(transparent)]
56 Canonical(#[from] Canonical),
57 #[error("delegate '{remote}' has diverged 'rad/sigrefs': {current} -> {received}")]
58 Diverged {
59 remote: PublicKey,
60 current: Oid,
61 received: Oid,
62 },
63 #[error("canonical 'refs/rad/id' is missing")]
64 MissingRadId,
65 #[error(transparent)]
66 RefdbUpdate(#[from] repository::error::Update),
67 #[error(transparent)]
68 Resolve(#[from] repository::error::Resolve),
69 #[error(transparent)]
70 Refs(#[from] radicle::storage::refs::sigrefs::read::error::Read),
71 #[error(transparent)]
72 RemoteRefs(#[from] sigrefs::error::RemoteRefs),
73 #[error("failed to get remote namespaces: {0}")]
74 RemoteIds(#[source] radicle::git::raw::Error),
75 #[error(transparent)]
76 Step(#[from] Step),
77 #[error(transparent)]
78 Tracking(#[from] handle::error::Tracking),
79 #[error(transparent)]
80 Validation(#[from] radicle::storage::Error),
81 }
82
83 #[derive(Debug, Error)]
84 pub enum Canonical {
85 #[error(transparent)]
86 Resolve(#[from] git::repository::error::Resolve),
87 #[error(transparent)]
88 Verified(#[from] radicle::identity::DocError),
89 #[error("failed to verify `refs/rad/id`: {0}")]
90 Graph(#[source] radicle::git::raw::Error),
91 }
92}
93
94type IdentityTips = BTreeMap<PublicKey, Oid>;
95type SigrefTips = BTreeMap<PublicKey, Oid>;
96
97#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
98pub struct FetchLimit {
99 pub special: u64,
100 pub refs: u64,
101}
102
103#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash)]
104pub struct Config {
105 pub limit: FetchLimit,
106 pub level_min: FeatureLevel,
107}
108
109impl Default for FetchLimit {
110 fn default() -> Self {
111 Self {
112 special: DEFAULT_FETCH_SPECIAL_REFS_LIMIT,
113 refs: DEFAULT_FETCH_DATA_REFS_LIMIT,
114 }
115 }
116}
117
118#[derive(Debug)]
119pub enum FetchResult {
120 Success {
121 applied: Applied<'static>,
123 remotes: BTreeSet<PublicKey>,
125 validations: sigrefs::Validations,
127 },
128 Failed {
129 threshold: usize,
131 delegates: BTreeSet<PublicKey>,
133 validations: sigrefs::Validations,
135 },
136}
137
138impl FetchResult {
139 pub fn rejected(&self) -> impl Iterator<Item = &Update<'static>> {
140 match self {
141 Self::Success { applied, .. } => either::Either::Left(applied.rejected.iter()),
142 Self::Failed { .. } => either::Either::Right(std::iter::empty()),
143 }
144 }
145
146 pub fn is_success(&self) -> bool {
147 std::matches!(self, Self::Success { .. })
148 }
149}
150
151#[derive(Default)]
152pub struct FetchState {
153 refs: git::mem::Refdb,
157 canonical_rad_id: Option<Oid>,
159 ids: IdentityTips,
161 sigrefs: SigrefTips,
163 tips: BTreeMap<PublicKey, Vec<Update<'static>>>,
165 keepfiles: Vec<Keepfile>,
169}
170
171impl FetchState {
172 pub fn prune(&mut self, remote: &PublicKey) {
175 self.ids.remove(remote);
176 self.sigrefs.remove(remote);
177 self.tips.remove(remote);
178 }
179
180 pub fn canonical_rad_id(&self) -> Option<&Oid> {
181 self.canonical_rad_id.as_ref()
182 }
183
184 pub fn update_all<'a, I>(&mut self, other: I) -> Applied<'a>
187 where
188 I: IntoIterator<Item = (PublicKey, Vec<Update<'a>>)>,
189 {
190 let mut ap = Applied::default();
191 for (remote, ups) in other {
192 for up in &ups {
193 ap.append(&mut self.refs.update(Some(up.clone())));
194 }
195 let mut ups = ups
196 .into_iter()
197 .map(|up| up.into_owned())
198 .collect::<Vec<_>>();
199 self.tips
200 .entry(remote)
201 .and_modify(|tips| tips.append(&mut ups))
202 .or_insert(ups);
203 }
204 ap
205 }
206
207 pub(crate) fn as_cached<'a, R, S>(
208 &'a mut self,
209 handle: &'a mut Handle<R, S>,
210 ) -> Cached<'a, R, S> {
211 Cached {
212 handle,
213 state: self,
214 }
215 }
216}
217
218impl FetchState {
219 pub(super) fn run_stage<R, S, F>(
222 &mut self,
223 handle: &mut Handle<R, S>,
224 handshake: &Handshake,
225 step: &F,
226 ) -> Result<BTreeSet<PublicKey>, error::Step>
227 where
228 R: AsRef<Repository>,
229 S: transport::ConnectionStream,
230 F: ProtocolStage,
231 {
232 let refs = match step.ls_refs() {
233 Some(refs) => handle
234 .transport
235 .ls_refs(refs, handshake)?
236 .into_iter()
237 .filter_map(|r| step.ref_filter(r))
238 .collect::<Vec<_>>(),
239 None => vec![],
240 };
241 log::trace!("Received refs {refs:#?}");
242 step.pre_validate(&refs)?;
243
244 let wants_haves = step.wants_haves(handle.repository(), &refs)?;
245 if !wants_haves.wants.is_empty() {
246 let keepfile =
247 handle
248 .transport
249 .fetch(wants_haves, handle.interrupt.clone(), handshake)?;
250 self.keepfiles.extend(keepfile);
251 } else {
252 log::trace!("Nothing to fetch")
253 };
254
255 let mut fetched = BTreeSet::new();
256 for r in &refs {
257 match &r.name {
258 refs::ReceivedRefname::Namespaced { remote, suffix } => {
259 fetched.insert(*remote);
260 if let Some(rad) = suffix.as_ref().left() {
261 match rad {
262 refs::Special::Id => {
263 self.ids.insert(*remote, r.tip);
264 }
265
266 refs::Special::SignedRefs => {
267 self.sigrefs.insert(*remote, r.tip);
268 }
269 }
270 }
271 }
272 refs::ReceivedRefname::RadId => self.canonical_rad_id = Some(r.tip),
273 }
274 }
275
276 let up = step.prepare_updates(self, handle.repository(), &refs)?;
277 self.update_all(up.tips);
278
279 Ok(fetched)
280 }
281
282 #[allow(clippy::too_many_arguments)]
295 fn run_special_refs<R, S>(
296 &mut self,
297 handle: &mut Handle<R, S>,
298 handshake: &Handshake,
299 delegates: BTreeSet<PublicKey>,
300 threshold: usize,
301 limit: &FetchLimit,
302 remote: PublicKey,
303 refs_at: Option<Vec<RefsAt>>,
304 ) -> Result<sigrefs::RemoteRefs, error::Protocol>
305 where
306 R: AsRef<Repository>,
307 S: transport::ConnectionStream,
308 {
309 let remotes: Vec<_> = match refs_at {
310 Some(refs_at) => {
311 let sigrefs_at = stage::SigrefsAt {
312 remote,
313 delegates: delegates.clone(),
314 refs_at: refs_at.clone(),
315 blocked: handle.blocked.clone(),
316 limit: limit.special,
317 };
318 log::trace!("{sigrefs_at:?}");
319 self.run_stage(handle, handshake, &sigrefs_at)?;
320
321 refs_at.iter().map(|r| &r.remote).cloned().collect()
322 }
323 None => {
324 let followed = handle.allowed();
325 log::trace!("Followed nodes {followed:?}");
326 let special_refs = stage::SpecialRefs {
327 blocked: handle.blocked.clone(),
328 remote,
329 delegates: delegates.clone(),
330 followed,
331 threshold,
332 limit: limit.special,
333 };
334 log::trace!("{special_refs:?}");
335 let fetched = self.run_stage(handle, handshake, &special_refs)?;
336
337 fetched.iter().chain(delegates.iter()).cloned().collect()
338 }
339 };
340
341 Ok(remotes
342 .into_iter()
343 .map(|remote| (remote, self.as_cached(handle).load(&remote)))
344 .collect())
345 }
346
347 pub(super) fn run<R, S>(
364 mut self,
365 handle: &mut Handle<R, S>,
366 handshake: &Handshake,
367 config: Config,
368 remote: PublicKey,
369 refs_at: Option<Vec<RefsAt>>,
370 ) -> Result<FetchResult, error::Protocol>
371 where
372 R: AsRef<Repository>,
373 S: transport::ConnectionStream,
374 {
375 let start = Instant::now();
376 self.run_stage(
380 handle,
381 handshake,
382 &stage::CanonicalId {
383 remote,
384 limit: config.limit.special,
385 },
386 )?;
387 log::debug!("Fetched rad/id ({}ms)", start.elapsed().as_millis());
388
389 let anchor = self
394 .as_cached(handle)
395 .canonical()?
396 .ok_or(error::Protocol::MissingRadId)?;
397
398 let is_delegate = anchor.is_delegate(&Did::from(handle.local()));
399 let delegates = anchor
402 .delegates()
403 .iter()
404 .filter(|id| !handle.is_blocked(id))
405 .map(|did| PublicKey::from(*did))
406 .collect::<BTreeSet<_>>();
407
408 log::trace!("Identity delegates {delegates:?}");
409
410 let threshold = if is_delegate {
413 anchor.threshold() - 1
414 } else {
415 anchor.threshold()
416 };
417 let signed_refs = self.run_special_refs(
418 handle,
419 handshake,
420 delegates.clone(),
421 threshold,
422 &config.limit,
423 remote,
424 refs_at,
425 )?;
426
427 log::debug!(
428 "Fetched data for {} remote(s) ({}ms)",
429 signed_refs.len(),
430 start.elapsed().as_millis()
431 );
432
433 let data_refs = stage::DataRefs::new(signed_refs);
434 let fetched = self.run_stage(handle, handshake, &data_refs)?;
435 log::debug!(
436 "Fetched data refs for {} remotes ({}ms)",
437 fetched.len(),
438 start.elapsed().as_millis()
439 );
440
441 match handle.transport.done() {
445 Ok(()) => log::debug!("Sent done signal to remote {remote}"),
446 Err(err) => {
447 log::debug!("Failed to signal EOF to {remote}: {err}")
448 }
449 }
450
451 let mut failures = sigrefs::Validations::default();
455
456 let signed_refs = data_refs.into_inner();
457
458 let mut remotes = BTreeSet::new();
461
462 let mut valid_delegates = handle
465 .repository()
466 .remote_ids()
467 .map_err(error::Protocol::RemoteIds)?
468 .filter_map(|id| id.ok())
469 .filter(|id| delegates.contains(id))
470 .collect::<BTreeSet<_>>();
471 let mut failed_delegates = BTreeSet::new();
472
473 for (remote, refs) in signed_refs {
476 if handle.is_blocked(&remote) {
477 log::trace!("Skipping blocked remote {remote}");
478 continue;
479 }
480
481 match (refs, delegates.contains(&remote)) {
482 (Ok(None), false) => {
483 log::debug!("Pruning non-delegate {remote} tips, missing 'rad/sigrefs'");
484 failures.push(sigrefs::Validation::MissingRadSigRefs(remote));
485 self.prune(&remote);
486 }
487 (Ok(None), true) => {
488 log::debug!("Pruning delegate {remote} tips, missing 'rad/sigrefs'");
489 failures.push(sigrefs::Validation::MissingRadSigRefs(remote));
490 self.prune(&remote);
491 valid_delegates.remove(&remote);
497 failed_delegates.insert(remote);
498 }
499 (Err(err), _) => {
500 log::debug!("Pruning {remote} tips due to: {err}");
501 self.prune(&remote);
502 valid_delegates.remove(&remote);
503 failed_delegates.insert(remote);
504 failures.push(sigrefs::Validation::Read {
505 remote,
506 source: err,
507 });
508 }
509 (Ok(Some(refs)), delegate) if refs.feature_level() < config.level_min => {
510 log::debug!(
511 "Pruning {remote} tips due to insufficient feature level '{}' < '{}'",
512 refs.feature_level(),
513 config.level_min
514 );
515
516 failures.push(sigrefs::Validation::InsufficientFeatureLevel {
517 remote,
518 actual: refs.feature_level(),
519 minimum: config.level_min,
520 });
521
522 if delegate {
523 valid_delegates.remove(&remote);
524 failed_delegates.insert(remote);
525 }
526
527 self.prune(&remote);
528 }
529 (Ok(Some(refs)), false) => {
530 let level_reachable = refs.feature_level();
531
532 match SignedRefs::load(remote, handle.repository()) {
533 Ok(Some(SignedRefs { at, .. })) => {
534 if matches!(
538 repository::ancestry(handle.repository(), at, refs.at)?,
539 repository::Ancestry::Behind | repository::Ancestry::Diverged
540 ) {
541 self.prune(&remote);
542 continue;
543 }
544 }
545 Err(radicle::storage::refs::sigrefs::read::error::Read::Downgrade {
546 levels,
547 actual,
548 ..
549 }) => {
550 let level_required = levels.max();
551 if level_reachable >= level_required {
552 log::info!(
553 "Non-delegate {remote} has downgraded history, currently stuck at '{actual}', expects to be upgraded to '{level_required}' and will be upgraded to '{level_reachable}'."
554 )
555 } else {
556 log::debug!(
557 "Non-delegate {remote} has downgraded history, currently stuck at '{actual}', expects to be upgraded to '{level_required}' but only level '{level_reachable}' was advertised."
558 );
559 self.prune(&remote);
560 continue;
561 }
562 }
563 Err(err) => return Err(error::Protocol::Refs(err)),
564 Ok(None) => {
565 }
567 }
568
569 let cache = self.as_cached(handle);
570 if let Some(warns) = sigrefs::validate(&cache, refs)?.as_mut() {
571 log::debug!(
572 "Pruning non-delegate {remote} tips, due to validation failures"
573 );
574 self.prune(&remote);
575 failures.append(warns);
576 } else {
577 remotes.insert(remote);
578 }
579 }
580 (Ok(Some(refs)), true) => {
581 let level_reachable = refs.feature_level();
582
583 match SignedRefs::load(remote, handle.repository()) {
584 Ok(Some(SignedRefs { at, .. })) => {
585 let ancestry = repository::ancestry(handle.repository(), at, refs.at)?;
586 if matches!(ancestry, repository::Ancestry::Behind) {
587 log::trace!(
588 "Advertised `rad/sigrefs` {} is behind {at} for {remote}",
589 refs.at
590 );
591 self.prune(&remote);
592 continue;
593 } else if matches!(ancestry, repository::Ancestry::Diverged) {
594 return Err(error::Protocol::Diverged {
595 remote,
596 current: at,
597 received: refs.at,
598 });
599 }
600 }
601 Err(radicle::storage::refs::sigrefs::read::error::Read::Downgrade {
602 levels,
603 actual,
604 ..
605 }) => {
606 let level_required = levels.max();
607 if level_reachable >= level_required {
608 log::info!(
609 "Delegate {remote} has downgraded history, currently stuck at '{actual}', expects to be upgraded to '{level_required}' and will be upgraded to '{level_reachable}'."
610 )
611 } else {
612 log::info!(
613 "Delegate {remote} has downgraded history, currently stuck at '{actual}', expects to be upgraded to '{level_required}' but only level '{level_reachable}' was advertised."
614 );
615 self.prune(&remote);
616 continue;
617 }
618 }
619 Err(err) => return Err(error::Protocol::Refs(err)),
620 Ok(None) => {
621 }
623 }
624
625 let cache = self.as_cached(handle);
626 let mut fails =
627 sigrefs::validate(&cache, refs)?.unwrap_or(Validations::default());
628 if !fails.is_empty() {
629 log::debug!("Pruning delegate {remote} tips, due to validation failures");
630 self.prune(&remote);
631 valid_delegates.remove(&remote);
632 failed_delegates.insert(remote);
633 failures.append(&mut fails)
634 } else {
635 valid_delegates.insert(remote);
636 remotes.insert(remote);
637 }
638
639 if level_reachable < FeatureLevel::LATEST {
640 log::warn!(
641 "Delegate {remote} is on feature level '{level_reachable}' which is lower than '{}', they should consider upgrading Radicle.",
642 FeatureLevel::LATEST
643 )
644 }
645 }
646 }
647 }
648 log::debug!(
649 "Validated {} remote(s) ({}ms)",
650 remotes.len(),
651 start.elapsed().as_millis()
652 );
653
654 if valid_delegates.len() >= threshold {
657 let applied = repository::update(
658 handle.repository(),
659 self.tips
660 .clone()
661 .into_values()
662 .flat_map(|ups| ups.into_iter()),
663 )?;
664 log::debug!("Applied updates ({}ms)", start.elapsed().as_millis());
665 Ok(FetchResult::Success {
666 applied,
667 remotes,
668 validations: failures,
669 })
670 } else {
671 log::debug!(
672 "Fetch failed: {} failure(s) ({}ms)",
673 failures.len(),
674 start.elapsed().as_millis()
675 );
676 Ok(FetchResult::Failed {
677 threshold,
678 delegates: failed_delegates,
679 validations: failures,
680 })
681 }
682 }
683}
684
685pub(crate) struct Cached<'a, R, S> {
688 handle: &'a mut Handle<R, S>,
689 state: &'a mut FetchState,
690}
691
692impl<R, S> Cached<'_, R, S>
693where
694 R: AsRef<Repository>,
695{
696 pub fn refname_to_id<'b, N>(
699 &self,
700 refname: N,
701 ) -> Result<Option<Oid>, repository::error::Resolve>
702 where
703 N: Into<Qualified<'b>>,
704 {
705 let refname = refname.into();
706 match self.state.refs.refname_to_id(refname.clone()) {
707 None => repository::refname_to_id(self.handle.repository(), refname),
708 Some(oid) => Ok(Some(oid)),
709 }
710 }
711
712 pub fn canonical_rad_id(&self) -> Option<Oid> {
714 self.state.canonical_rad_id().copied()
715 }
716
717 pub fn verified(&self, head: Oid) -> Result<Doc, DocError> {
718 self.handle.verified(head)
719 }
720
721 pub fn canonical(&self) -> Result<Option<Doc>, error::Canonical> {
736 let tip = self.refname_to_id(refs::REFS_RAD_ID.clone())?;
737 let cached_tip = self.canonical_rad_id();
738
739 let oid = match (tip, cached_tip) {
740 (None, None) => {
741 return Ok(None);
742 }
743 (Some(oid), None) | (None, Some(oid)) => oid,
744 (Some(repository), Some(cached)) => {
745 let repo = self.handle.repository();
746 match repo
747 .backend
748 .graph_ahead_behind(repository.into(), cached.into())
749 {
750 Ok((ahead, behind)) => match (ahead, behind) {
751 (0, _) => cached,
752 _ => repository,
753 },
754 Err(err) if err.code() == radicle::git::raw::ErrorCode::NotFound => repository,
755 Err(err) => {
756 return Err(error::Canonical::Graph(err));
757 }
758 }
759 }
760 };
761
762 self.verified(oid).map(Some).map_err(error::Canonical::from)
763 }
764
765 pub fn load(
766 &self,
767 remote: &PublicKey,
768 ) -> Result<Option<SignedRefs>, radicle::storage::refs::sigrefs::read::error::Read> {
769 match self.state.sigrefs.get(remote) {
770 None => SignedRefs::load(*remote, self.handle.repository()),
771 Some(tip) => SignedRefs::load_at(*tip, *remote, self.handle.repository()),
772 }
773 }
774
775 #[allow(dead_code)]
776 pub(crate) fn inspect(&self) {
777 self.state.refs.inspect()
778 }
779}
780
781impl<R, S> RemoteRepository for Cached<'_, R, S>
782where
783 R: AsRef<Repository>,
784{
785 fn remote(&self, remote: &RemoteId) -> Result<Remote, storage::refs::Error> {
786 self.handle.repository().remote(remote)
789 }
790
791 fn remotes(&self) -> Result<Remotes, storage::refs::Error> {
792 self.state
793 .sigrefs
794 .keys()
795 .map(|id| self.remote(id).map(|remote| (*id, remote)))
796 .collect::<Result<_, _>>()
797 }
798
799 fn remote_refs_at(&self) -> Result<Vec<RefsAt>, storage::refs::Error> {
800 self.handle.repository().remote_refs_at()
801 }
802}
803
804impl<R, S> ValidateRepository for Cached<'_, R, S>
805where
806 R: AsRef<Repository>,
807{
808 fn validate_remote(&self, remote: &Remote) -> Result<Validations, storage::Error> {
812 let mut signed = BTreeMap::from((*remote.refs).clone());
814 let mut validations = Validations::default();
815 let mut has_sigrefs = false;
816
817 for (refname, oid) in self.state.refs.references_of(&remote.id()) {
819 if refname == storage::refs::SIGREFS_BRANCH.to_ref_string() {
821 has_sigrefs = true;
822 continue;
823 }
824 if let Some(signed_oid) = signed.remove(&refname) {
825 if oid != signed_oid {
826 validations.push(Validation::MismatchedRef {
827 remote: remote.id(),
828 refname,
829 expected: signed_oid,
830 actual: oid,
831 });
832 }
833 } else {
834 validations.push(Validation::UnsignedRef(refname));
835 }
836 }
837
838 if !has_sigrefs {
839 validations.push(Validation::MissingRadSigRefs(remote.id()));
840 }
841
842 for (name, _) in signed.into_iter() {
845 validations.push(Validation::MissingRef {
846 refname: name,
847 remote: remote.id(),
848 });
849 }
850
851 Ok(validations)
852 }
853}