Skip to main content

radicle_fetch/
state.rs

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
25/// The data size limit, 5Mb, while fetching the special refs,
26/// i.e. `rad/id` and `rad/sigrefs`.
27pub const DEFAULT_FETCH_SPECIAL_REFS_LIMIT: u64 = 1024 * 1024 * 5;
28/// The data size limit, 5Gb, while fetching the data refs,
29/// i.e. `refs/heads`, `refs/tags`, `refs/cobs`, etc.
30pub 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        /// The set of applied changes to the reference store.
122        applied: Applied<'static>,
123        /// The set of namespaces that were fetched.
124        remotes: BTreeSet<PublicKey>,
125        /// Any validation errors that were found while fetching.
126        validations: sigrefs::Validations,
127    },
128    Failed {
129        /// The threshold that needed to be met.
130        threshold: usize,
131        /// The offending delegates.
132        delegates: BTreeSet<PublicKey>,
133        /// Validation errors that were found while fetching.
134        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    /// In-memory refdb used to keep track of new updates without
154    /// committing them to the real refdb until all validation has
155    /// occurred.
156    refs: git::mem::Refdb,
157    /// Have we seen the `rad/id` reference?
158    canonical_rad_id: Option<Oid>,
159    /// Seen remote `rad/id` tips.
160    ids: IdentityTips,
161    /// Seen remote `rad/sigrefs` tips.
162    sigrefs: SigrefTips,
163    /// Seen reference tips, per remote.
164    tips: BTreeMap<PublicKey, Vec<Update<'static>>>,
165    /// The `.keep` files created during packfile transfers. They are kept
166    /// within the state, so that when the state is dropped, it also attempts to
167    /// delete the files to release the locks on the packfiles.
168    keepfiles: Vec<Keepfile>,
169}
170
171impl FetchState {
172    /// Remove all tips associated with this `remote` in the
173    /// `FetchState`.
174    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    /// Update the in-memory refdb with the given updates while also
185    /// keeping track of the updates in [`FetchState::tips`].
186    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    /// Perform the ls-refs and fetch for the given `step`. The result
220    /// of these processes is kept track of in the internal state.
221    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    /// Fetch the set of special refs, depending on `refs_at`.
283    ///
284    /// If `refs_at` is `Some`, then run the [`SigrefsAt`] stage,
285    /// which specifically fetches `rad/sigrefs` which are listed in
286    /// `refs_at`.
287    ///
288    /// If `refs_at` is `None`, then run the [`SpecialRefs`] stage,
289    /// which fetches `rad/sigrefs` and `rad/id` from all tracked and
290    /// delegate peers (scope dependent).
291    ///
292    /// The resulting [`sigrefs::RemoteRefs`] will be the set of
293    /// `rad/sigrefs` of the fetched remotes.
294    #[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    /// The finalization of the protocol exchange is as follows:
348    ///
349    ///   1. Load the canonical `rad/id` to use as the anchor for
350    ///      getting the delegates of the identity.
351    ///   2. Calculate the trusted set of peers for fetching from.
352    ///   3. Fetch the special references, i.e. `rad/id` and `rad/sigrefs`.
353    ///   4. Load the signed references, where these signed references
354    ///      must be cryptographically verified for delegates,
355    ///      otherwise they are discarded for non-delegates.
356    ///   5. Fetch the data references, i.e. references found in
357    ///      `rad/sigrefs`.
358    ///   6. Validate the fetched references for delegates and
359    ///      non-delegates, pruning any invalid remotes from the set
360    ///      of updating tips.
361    ///   7. Apply the valid tips, iff no delegates failed validation.
362    ///   8. Signal to the other side that the process has completed.
363    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        // N.b. we always fetch the `rad/id` since our delegate set
377        // might be further ahead than theirs, e.g. we are the
378        // deciding vote on adding a delegate.
379        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        // N.b. The error case here should not happen. In the case of
390        // a `clone` we have asked for refs/rad/id and ensured it was
391        // fetched. In the case of `pull` the repository should have
392        // the refs/rad/id set.
393        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        // TODO: not sure we should allow to block *any* peer from the
400        // delegate set. We could end up ignoring delegates.
401        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        // The local peer does not need to count towards the threshold
411        // since they must be valid already.
412        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        // N.b. signal to exit the upload-pack sequence
442        // We're finished fetching on this side, and all that's left
443        // is validation.
444        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        // Run validation of signed refs, pruning any offending
452        // remotes from the tips, thus not updating the production Git
453        // repository.
454        let mut failures = sigrefs::Validations::default();
455
456        let signed_refs = data_refs.into_inner();
457
458        // We may prune fetched remotes, so we keep track of
459        // non-pruned, fetched remotes here.
460        let mut remotes = BTreeSet::new();
461
462        // The valid delegates start with all delegates that this peer
463        // currently has valid references for
464        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        // TODO(finto): this might read better if it got its own
474        // private function.
475        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                    // This delegate has removed their `rad/sigrefs`.
492                    // Technically, we can continue with their
493                    // previous `rad/sigrefs` but if this occurs with
494                    // enough delegates also failing validation we
495                    // would rather surface the issue and fail the fetch.
496                    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                            // Prune non-delegates if they're behind or
535                            // diverged. A diverged case is non-fatal for
536                            // delegates.
537                            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                            // We see signed references for this non-delegate for the first time.
566                        }
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                            // We see signed references for this delegate for the first time.
622                        }
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        // N.b. only apply to Git repository if there are enough valid
655        // delegates that pass the threshold.
656        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
685/// A cached version of [`Handle`] by using the underlying
686/// [`FetchState`]'s data for performing lookups.
687pub(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    /// Resolves `refname` to its [`ObjectId`] by first looking at the
697    /// [`FetchState`] and falling back to the [`Handle::refdb`].
698    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    /// Get the `rad/id` found in the [`FetchState`].
713    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    /// Resolve the verified [`Doc`], by choosing a `refs/rad/id` head to
722    /// resolve from.
723    ///
724    /// There are two candidate namespaces:
725    ///
726    ///   1. Of the fetching node.
727    ///   2. Of the node being fetched from.
728    ///
729    /// Both might be unset, in this case [`None`] is returned.
730    ///
731    /// If exactly one of the two is set, it is used.
732    ///
733    /// Otherwise, the ahead/behind relationship between the two candidates
734    /// is checked, and (2.) is used if it is ahead of (1.).
735    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        // N.b. this is unused so we just delegate to the underlying
787        // repository for a correct implementation.
788        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    // N.b. we don't verify the `rad/id` of each remote since they may
809    // not have a reference to the COB if they have not interacted
810    // with it.
811    fn validate_remote(&self, remote: &Remote) -> Result<Validations, storage::Error> {
812        // Contains a copy of the signed refs of this remote.
813        let mut signed = BTreeMap::from((*remote.refs).clone());
814        let mut validations = Validations::default();
815        let mut has_sigrefs = false;
816
817        // Check all repository references, making sure they are present in the signed refs map.
818        for (refname, oid) in self.state.refs.references_of(&remote.id()) {
819            // Skip validation of the signed refs branch, as it is not part of `Remote`.
820            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        // The refs that are left in the map, are ones that were signed, but are not
843        // in the repository. If any are left, bail.
844        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}