//! Abstract code to manage a set of tunnels which has underlying circuit(s).

//!

//! This module implements the real logic for deciding when and how to

//! launch tunnels, and for which tunnels to hand out in response to

//! which requests.

//!

//! For testing and abstraction purposes, this module _does not_

//! actually know anything about tunnels _per se_.  Instead,

//! everything is handled using a set of traits that are internal to this

//! crate:

//!

//!  * [`AbstractTunnel`] is a view of a tunnel.

//!  * [`AbstractTunnelBuilder`] knows how to build an `AbstractCirc`.

//!

//! Using these traits, the [`AbstractTunnelMgr`] object manages a set of

//! tunnels , launching them as necessary, and keeping track of the

//! restrictions on their use.

// TODO:

// - Testing

//    - Error from prepare_action()

//    - Error reported by restrict_mut?

use crate::config::CircuitTiming;

use crate::usage::{SupportedTunnelUsage, TargetTunnelUsage};

use crate::{DirInfo, Error, PathConfig, Result, timeouts};

use retry_error::RetryError;

use tor_async_utils::mpsc_channel_no_memquota;

use tor_basic_utils::retry::RetryDelay;

use tor_config::MutCfg;

use tor_error::{AbsRetryTime, HasRetryTime, debug_report, info_report, internal, warn_report};

#[cfg(feature = "vanguards")]

use tor_guardmgr::vanguards::VanguardMgr;

use tor_linkspec::CircTarget;

use tor_proto::circuit::UniqId;

use tor_proto::client::circuit::{CircParameters, Path};

use tor_rtcompat::{Runtime, SleepProviderExt};

use async_trait::async_trait;

use futures::channel::mpsc;

use futures::future::{FutureExt, Shared};

use futures::stream::{FuturesUnordered, StreamExt};

use oneshot_fused_workaround as oneshot;

use std::collections::HashMap;

use std::fmt::Debug;

use std::hash::Hash;

use std::panic::AssertUnwindSafe;

use std::sync::{self, Arc, Weak};

use std::time::{Duration, Instant};

use tor_rtcompat::SpawnExt;

use tracing::{debug, instrument, trace, warn};

use weak_table::PtrWeakHashSet;

mod streams;

/// Description of how we got a tunnel.

#[non_exhaustive]

#[derive(Debug, Copy, Clone, Eq, PartialEq)]

pub(crate) enum TunnelProvenance {

    /// This channel was newly launched, or was in progress and finished while

    /// we were waiting.

    NewlyCreated,

    /// This channel already existed when we asked for it.

    Preexisting,

}

/// An error returned when we cannot apply circuit restriction.

#[derive(Clone, Debug, thiserror::Error)]

#[non_exhaustive]

pub enum RestrictionFailed {

    /// Tried to restrict a specification, but the tunnel didn't support the

    /// requested usage.

    #[error("Specification did not support desired usage")]

    NotSupported,

}

/// Minimal abstract view of a tunnel.

///

/// From this module's point of view, tunnels are simply objects

/// with unique identities, and a possible closed-state.

#[async_trait]

pub(crate) trait AbstractTunnel: Debug {

    /// Type for a unique identifier for tunnels.

    type Id: Clone + Debug + Hash + Eq + Send + Sync;

    /// Return the unique identifier for this tunnel.

    ///

    /// # Requirements

    ///

    /// The values returned by this function are unique for distinct

    /// tunnels.

    fn id(&self) -> Self::Id;

    /// Return true if this tunnel is usable for some purpose.

    ///

    /// Reasons a tunnel might be unusable include being closed.

    fn usable(&self) -> bool;

    /// Return a list of [`Path`] objects describing the only circuit in this tunnel.

    ///

    /// Returns an error if the tunnel has more than one tunnel.

    fn single_path(&self) -> tor_proto::Result<Arc<Path>>;

    /// Return the number of hops in this tunnel.

    ///

    /// Returns an error if the circuit is closed.

    ///

    /// NOTE: This function will currently return only the number of hops

    /// _currently_ in the tunnel. If there is an extend operation in progress,

    /// the currently pending hop may or may not be counted, depending on whether

    /// the extend operation finishes before this call is done.

    fn n_hops(&self) -> tor_proto::Result<usize>;

    /// Return true if this tunnel is closed and therefore unusable.

    fn is_closing(&self) -> bool;

    /// Return a process-unique identifier for this tunnel.

    fn unique_id(&self) -> UniqId;

    /// Extend the tunnel via the most appropriate handshake to a new `target` hop.

    async fn extend<T: CircTarget + Sync>(

        &self,

        target: &T,

        params: CircParameters,

    ) -> tor_proto::Result<()>;

    /// Return a time at which this tunnel is last known to be used,

    /// or None if it is in use right now (or has never been used).

    async fn last_known_to_be_used_at(&self) -> tor_proto::Result<Option<Instant>>;

}

/// A plan for an `AbstractCircBuilder` that can maybe be mutated by tests.

///

/// You should implement this trait using all default methods for all code that isn't test code.

pub(crate) trait MockablePlan {

    /// Add a reason string that was passed to `SleepProvider::block_advance()` to this object

    /// so that it knows what to pass to `::release_advance()`.

}

/// An object that knows how to build tunnels.

///

/// This creates tunnels in two phases. First, a plan is

/// made for how to build the tunnel. This planning phase should be

/// relatively fast, and must not suspend or block.  Its purpose is to

/// get an early estimate of which operations the tunnel will be able

/// to support when it's done.

///

/// Second, the tunnel is actually built, using the plan as input.

#[async_trait]

pub(crate) trait AbstractTunnelBuilder<R: Runtime>: Send + Sync {

    /// The tunnel type that this builder knows how to build.

    type Tunnel: AbstractTunnel + Send + Sync;

    /// An opaque type describing how a given tunnel will be built.

    /// It may represent some or all of a path-or it may not.

    //

    // TODO: It would be nice to have this parameterized on a lifetime,

    // and have that lifetime depend on the lifetime of the directory.

    // But I don't think that rust can do that.

    //

    // HACK(eta): I don't like the fact that `MockablePlan` is necessary here.

    type Plan: Send + Debug + MockablePlan;

    // TODO: I'd like to have a Dir type here to represent

    // create::DirInfo, but that would need to be parameterized too,

    // and would make everything complicated.

    /// Form a plan for how to build a new tunnel that supports `usage`.

    ///

    /// Return an opaque Plan object, and a new spec describing what

    /// the tunnel will actually support when it's built.  (For

    /// example, if the input spec requests a tunnel that connect to

    /// port 80, then "planning" the tunnel might involve picking an

    /// exit that supports port 80, and the resulting spec might be

    /// the exit's complete list of supported ports.)

    ///

    /// # Requirements

    ///

    /// The resulting Spec must support `usage`.

    fn plan_tunnel(

        &self,

        usage: &TargetTunnelUsage,

        dir: DirInfo<'_>,

    ) -> Result<(Self::Plan, SupportedTunnelUsage)>;

    /// Construct a tunnel according to a given plan.

    ///

    /// On success, return a spec describing what the tunnel can be used for,

    /// and the tunnel that was just constructed.

    ///

    /// This function should implement some kind of a timeout for

    /// tunnel that are taking too long.

    ///

    /// # Requirements

    ///

    /// The spec that this function returns _must_ support the usage

    /// that was originally passed to `plan_tunnel`.  It _must_ also

    /// contain the spec that was originally returned by

    /// `plan_tunnel`.

    async fn build_tunnel(&self, plan: Self::Plan) -> Result<(SupportedTunnelUsage, Self::Tunnel)>;

    /// Return a "parallelism factor" with which tunnels should be

    /// constructed for a given purpose.

    ///

    /// If this function returns N, then whenever we launch tunnels

    /// for this purpose, then we launch N in parallel.

    ///

    /// The default implementation returns 1.  The value of 0 is

    /// treated as if it were 1.

    /// Return a "parallelism factor" for which tunnels should be

    /// used for a given purpose.

    ///

    /// If this function returns N, then whenever we select among

    /// open tunnels for this purpose, we choose at random from the

    /// best N.

    ///

    /// The default implementation returns 1.  The value of 0 is

    /// treated as if it were 1.

    // TODO: Possibly this doesn't belong in this trait.

    /// Return true if we are currently attempting to learn tunnel

    /// timeouts by building testing tunnels.

    fn learning_timeouts(&self) -> bool;

    /// Flush state to the state manager if we own the lock.

    ///

    /// Return `Ok(true)` if we saved, and `Ok(false)` if we didn't hold the lock.

    fn save_state(&self) -> Result<bool>;

    /// Return this builder's [`PathConfig`].

    fn path_config(&self) -> Arc<PathConfig>;

    /// Replace this builder's [`PathConfig`].

    // TODO: This is dead_code because we only call this for the CircuitBuilder specialization of

    // CircMgr, not from the generic version, because this trait doesn't provide guardmgr, which is

    // needed by the [`CircMgr::reconfigure`] function that would be the only caller of this. We

    // should add `guardmgr` to this trait, make [`CircMgr::reconfigure`] generic, and remove this

    // dead_code marking.

    #[allow(dead_code)]

    fn set_path_config(&self, new_config: PathConfig);

    /// Return a reference to this builder's timeout estimator.

    fn estimator(&self) -> &timeouts::Estimator;

    /// Return a reference to this builder's `VanguardMgr`.

    #[cfg(feature = "vanguards")]

    fn vanguardmgr(&self) -> &Arc<VanguardMgr<R>>;

    /// Replace our state with a new owning state, assuming we have

    /// storage permission.

    fn upgrade_to_owned_state(&self) -> Result<()>;

    /// Reload persistent state from disk, if we don't have storage permission.

    fn reload_state(&self) -> Result<()>;

    /// Return a reference to this builder's `GuardMgr`.

    fn guardmgr(&self) -> &tor_guardmgr::GuardMgr<R>;

    /// Reconfigure this builder using the latest set of network parameters.

    ///

    /// (NOTE: for now, this only affects tunnel timeout estimation.)

    fn update_network_parameters(&self, p: &tor_netdir::params::NetParameters);

}

/// Enumeration to track the expiration state of a tunnel.

///

/// A tunnel an either be unused (at which point it should expire if it is

/// _still unused_ by a certain time, or dirty (at which point it should

/// expire after a certain duration).

///

/// All tunnels start out "unused" and become "dirty" when their spec

/// is first restricted -- that is, when they are first handed out to be

/// used for a request.

#[derive(Debug, Clone, PartialEq, Eq)]

enum ExpirationInfo {

    /// The tunnel has never been used, and has never been restricted for use with a request.

    Unused {

        /// A time when the tunnel was created.

        created: Instant,

    },

    /// The tunnel is not-long-lived; we will expire by waiting until a certain amount of time

    /// after it was first used.

    Dirty {

        /// The time at which this tunnel's spec was first restricted.

        dirty_since: Instant,

    },

    /// The tunnel is long-lived; we will expire by waiting until it has passed

    /// a certain amount of time without having any streams attached to it.

    LongLived {

        /// Last time at which the tunnel was checked and found not to have any streams.

        ///

        /// (This is a bit complicated: We have to be vague here, since we need

        /// an async check to find out that a tunnel is used, or when it actually

        /// became disused.)

        last_known_to_be_used_at: Instant,

    },

}

impl ExpirationInfo {

    /// Return an ExpirationInfo for a newly created tunnel.

    /// Mark this ExpirationInfo as having been in-use at `now`.

    ///

    /// If `long_lived` is false, the associated tunnel should expire a certain amount of time

    /// after it was _first_ used.

    /// If `long_lived` is true, the associated tunnel should expire a certain amount of time

    /// after it was _last_ used.

            }

        }

    /// Return an internal error if this ExpirationInfo is not marked as long-lived.

        }

}

/// Settings to determine when circuits are expired.

#[derive(Clone, Debug)]

pub(crate) struct ExpirationParameters {

    /// Any unused circuit is expired this long after it was created.

    expire_unused_after: Duration,

    /// Any non long-lived dirty circuit is expired this long after it first becomes dirty.

    expire_dirty_after: Duration,

    /// Any long-lived circuit is expired after having been disused for this long.

    expire_disused_after: Duration,

}

/// An entry for an open tunnel held by an `AbstractTunnelMgr`.

#[derive(Debug, Clone)]

pub(crate) struct OpenEntry<T> {

    /// The supported usage for this tunnel.

    spec: SupportedTunnelUsage,

    /// The tunnel under management.

    tunnel: Arc<T>,

    /// When does this tunnel expire?

    ///

    /// (Note that expired tunnels are removed from the manager,

    /// which does not actually close them until there are no more

    /// references to them.)

    expiration: ExpirationInfo,

}

impl<T: AbstractTunnel> OpenEntry<T> {

    /// Make a new OpenEntry for a given tunnel and spec.

    /// Return true if the underlying tunnel can be used for `usage`.

    /// Change the underlying tunnel's permissible usage, based on its having

    /// been used for `usage` at time `now`.

    ///

    /// Return an error if the tunnel may not be used for `usage`.

    /// Find the "best" entry from a slice of OpenEntry for supporting

    /// a given `usage`.

    ///

    /// If `parallelism` is some N greater than 1, we pick randomly

    /// from the best `N` tunnels.

    ///

    /// # Requirements

    ///

    /// Requires that `ents` is nonempty, and that every element of `ents`

    /// supports `spec`.

        use rand::seq::IndexedMutRandom as _;

        // TODO: Actually look over the whole list to see which is better.

    /// Return true if this tunnel should be expired given that the current time is `now`,

    /// and the current settings are `params`.

            }

            }

            ExpirationInfo::LongLived {

            } => {

            }

        }

}

/// When should a tunnel expire?

///

/// Reflects possible uncertainty.

#[derive(Clone, Copy, Debug, Eq, PartialEq)]

enum ShouldExpire {

    /// The tunnel should expire now.

    Now,

    /// The circuit might expire now; we need to check.

    ///

    /// (This is the result we get when we know that this is a tunnel that should expire

    /// if it has gone for some duration D without having any streams on it,

    /// and that it definitely had a stream at time T.  It is now at least time T+D,

    /// but we don't know whether the tunnel has any streams in the intervening time.

    /// We need to call the async fn `last_known_to_be_used_at` to check.)

    PossiblyNow,

    /// The tunnel will not expire before the specified time.

    NotBefore(Instant),

}

impl ShouldExpire {

    /// Return a ShouldExpire reflecting an expiration that is known to be happening at `expiration`.

        } else {

        }

    /// Return a ShouldExpire reflecting an expiration that is known to be no sooner than `expiration`,

    /// but possibly later.

        } else {

        }

}

/// A result type whose "Ok" value is the Id for a tunnel from B.

type PendResult<B, R> = Result<<<B as AbstractTunnelBuilder<R>>::Tunnel as AbstractTunnel>::Id>;

/// An in-progress tunnel request tracked by an `AbstractTunnelMgr`.

///

/// (In addition to tracking tunnels, `AbstractTunnelMgr` tracks

/// _requests_ for tunnels.  The manager uses these entries if it

/// finds that some tunnel created _after_ a request first launched

/// might meet the request's requirements.)

struct PendingRequest<B: AbstractTunnelBuilder<R>, R: Runtime> {

    /// Usage for the operation requested by this request

    usage: TargetTunnelUsage,

    /// A channel to use for telling this request about tunnels that it

    /// might like.

    notify: mpsc::Sender<PendResult<B, R>>,

}

impl<B: AbstractTunnelBuilder<R>, R: Runtime> PendingRequest<B, R> {

    /// Return true if this request would be supported by `spec`.

}

/// An entry for an under-construction in-progress tunnel tracked by

/// an `AbstractTunnelMgr`.

#[derive(Debug)]

struct PendingEntry<B: AbstractTunnelBuilder<R>, R: Runtime> {

    /// Specification that this tunnel will support, if every pending

    /// request that is waiting for it is attached to it.

    ///

    /// This spec becomes more and more restricted as more pending

    /// requests are waiting for this tunnel.

    ///

    /// This spec is contained by circ_spec, and must support the usage

    /// of every pending request that's waiting for this tunnel.

    tentative_assignment: sync::Mutex<SupportedTunnelUsage>,

    /// A shared future for requests to use when waiting for

    /// notification of this tunnel's success.

    receiver: Shared<oneshot::Receiver<PendResult<B, R>>>,

}

impl<B: AbstractTunnelBuilder<R>, R: Runtime> PendingEntry<B, R> {

    /// Make a new PendingEntry that starts out supporting a given

    /// spec.  Return that PendingEntry, along with a Sender to use to

    /// report the result of building this tunnel.

    /// Return true if this tunnel's current tentative assignment

    /// supports `usage`.

    /// Try to change the tentative assignment of this tunnel by

    /// restricting it for use with `usage`.

    ///

    /// Return an error if the current tentative assignment didn't

    /// support `usage` in the first place.

    /// Find the best PendingEntry values from a slice for use with

    /// `usage`.

    ///

    /// # Requirements

    ///

    /// The `ents` slice must not be empty.  Every element of `ents`

    /// must support the given spec.

        // TODO: Actually look over the whole list to see which is better.

}

/// Wrapper type to represent the state between planning to build a

/// tunnel and constructing it.

#[derive(Debug)]

struct TunnelBuildPlan<B: AbstractTunnelBuilder<R>, R: Runtime> {

    /// The Plan object returned by [`AbstractTunnelBuilder::plan_tunnel`].

    plan: B::Plan,

    /// A sender to notify any pending requests when this tunnel is done.

    sender: oneshot::Sender<PendResult<B, R>>,

    /// A strong entry to the PendingEntry for this tunnel build attempt.

    pending: Arc<PendingEntry<B, R>>,

}

/// The inner state of an [`AbstractTunnelMgr`].

struct TunnelList<B: AbstractTunnelBuilder<R>, R: Runtime> {

    /// A map from tunnel ID to [`OpenEntry`] values for all managed

    /// open tunnels.

    ///

    /// A tunnel is added here from [`AbstractTunnelMgr::do_launch`] when we find

    /// that it completes successfully, and has not been cancelled.

    /// When we decide that such a tunnel should no longer be handed out for

    /// any new requests, we "retire" the tunnel by removing it from this map.

    #[allow(clippy::type_complexity)]

    open_tunnels: HashMap<<B::Tunnel as AbstractTunnel>::Id, OpenEntry<B::Tunnel>>,

    /// Weak-set of PendingEntry for tunnels that are being built.

    ///

    /// Because this set only holds weak references, and the only strong

    /// reference to the PendingEntry is held by the task building the tunnel,

    /// this set's members are lazily removed after the tunnel is either built

    /// or fails to build.

    ///

    /// This set is used for two purposes:

    ///

    /// 1. When a tunnel request finds that there is no open tunnel for its

    ///    purposes, it checks here to see if there is a pending tunnel that it

    ///    could wait for.

    /// 2. When a pending tunnel finishes building, it checks here to make sure

    ///    that it has not been cancelled. (Removing an entry from this set marks

    ///    it as cancelled.)

    ///

    /// An entry is added here in [`AbstractTunnelMgr::prepare_action`] when we

    /// decide that a tunnel needs to be launched.

    ///

    /// Later, in [`AbstractTunnelMgr::do_launch`], once the tunnel has finished

    /// (or failed), we remove the entry (by pointer identity).

    /// If we cannot find the entry, we conclude that the request has been

    /// _cancelled_, and so we discard any tunnel that was created.

    pending_tunnels: PtrWeakHashSet<Weak<PendingEntry<B, R>>>,

    /// Weak-set of PendingRequest for requests that are waiting for a

    /// tunnel to be built.

    ///

    /// Because this set only holds weak references, and the only

    /// strong reference to the PendingRequest is held by the task

    /// waiting for the tunnel to be built, this set's members are

    /// lazily removed after the request succeeds or fails.

    pending_requests: PtrWeakHashSet<Weak<PendingRequest<B, R>>>,

}

impl<B: AbstractTunnelBuilder<R>, R: Runtime> TunnelList<B, R> {

    /// Make a new empty `CircList`

    /// Add `e` to the list of open tunnels.

    /// Find all the usable open tunnels that support `usage`.

    ///

    /// Return None if there are no such tunnels.

    /// Find an open tunnel by ID.

    ///

    /// Return None if no such tunnels exists in this list.

    /// Extract an open tunnel by ID, removing it from this list.

    ///

    /// Return None if no such tunnel exists in this list.

    /// Remove tunnels based on expiration times.

    ///

    /// We remove every unused tunnel that is set to expire by

    /// `unused_cutoff`, and every dirty tunnel that has been dirty

    /// since before `dirty_cutoff`.

    ///

    /// Return the next time at which anything will definitely expire,

    /// and a list of long-lived tunnels where we need to check their usage status

    /// before we can be sure if they are expired.

    #[must_use]

                // Expires now: Do not retain.

                // Will expire at `when`: keep, but update `earliest_expiration`.

                    };

                }

                // Need to check tunnel to see if/when it is disused.

                ShouldExpire::PossiblyNow => {

                }

    /// Return the time when the tunnel with given `id`, should expire.

    ///

    /// Return None if no such tunnel exists.

    /// Update the "last known to be in use" time of a long-lived tunnel with ID `id`,

    /// based on learning when it was last used.

    ///

    /// Expire the tunnel if appropriate.

    ///

    /// If the tunnel is still part of the map, return the next instant at which it might expire.

    ///

    /// Returns an error if the tunnel was present but was _not_ already marked as long-lived.

            // got an error looking up disused time: discard the circuit.

        };

            // Circuit isn't there. Return.

        };

            ShouldExpire::Now | ShouldExpire::PossiblyNow => {

            }

        }

    /// Add `pending` to the set of in-progress tunnels.

    /// Find all pending tunnels that support `usage`.

    ///

    /// If no such tunnels are currently being built, return None.

        } else {

        }

    /// Return true if `circ` is still pending.

    ///

    /// A tunnel will become non-pending when finishes (successfully or not), or when it's

    /// removed from this list via `clear_all_tunnels()`.

    /// Construct and add a new entry to the set of request waiting

    /// for a tunnel.

    ///

    /// Return the request, and a new receiver stream that it should

    /// use for notification of possible tunnels to use.

    /// Return all pending requests that would be satisfied by a tunnel

    /// that supports `circ_spec`.

    /// Clear all pending and open tunnels.

    ///

    /// Calling `clear_all_tunnels` ensures that any request that is answered _after

    /// this method runs_ will receive a tunnels that was launched _after this

    /// method runs_.

        // Note that removing entries from pending_circs will also cause the

        // tunnel tasks to realize that they are cancelled when they

        // go to tell anybody about their results.

}

/// Timing information for tunnels that have been built but never used.

///

/// Currently taken from the network parameters.

struct UnusedTimings {

    /// Minimum lifetime of a tunnel created while learning

    /// tunnel timeouts.

    learning: Duration,

    /// Minimum lifetime of a tunnel created while not learning

    /// tunnel timeouts.

    not_learning: Duration,

}

// This isn't really fallible, given the definitions of the underlying

// types.

#[allow(clippy::fallible_impl_from)]

impl From<&tor_netdir::params::NetParameters> for UnusedTimings {

        // These try_into() calls can't fail, so unwrap() can't panic.

        #[allow(clippy::unwrap_used)]

}

/// Abstract implementation for tunnel management.

///

/// The algorithm provided here is fairly simple. In its simplest form:

///

/// When somebody asks for a tunnel for a given operation: if we find

/// one open already, we return it.  If we find in-progress tunnels

/// that would meet our needs, we wait for one to finish (or for all

/// to fail).  And otherwise, we launch one or more tunnels to meet the

/// request's needs.

///

/// If this process fails, then we retry it, up to a timeout or a

/// numerical limit.

///

/// If a tunnel not previously considered for a given request

/// finishes before the request is satisfied, and if the tunnel would

/// satisfy the request, we try to give that tunnel as an answer to

/// that request even if it was not one of the tunnels that request

/// was waiting for.

pub(crate) struct AbstractTunnelMgr<B: AbstractTunnelBuilder<R>, R: Runtime> {

    /// Builder used to construct tunnels.

    builder: B,

    /// An asynchronous runtime to use for launching tasks and

    /// checking timeouts.

    runtime: R,

    /// A CircList to manage our list of tunnels, requests, and

    /// pending tunnels.

    tunnels: sync::Mutex<TunnelList<B, R>>,

    /// Configured information about when to expire tunnels and requests.

    circuit_timing: MutCfg<CircuitTiming>,

    /// Minimum lifetime of an unused tunnel.

    ///

    /// Derived from the network parameters.

    unused_timing: sync::Mutex<UnusedTimings>,

}

/// An action to take in order to satisfy a request for a tunnel.

enum Action<B: AbstractTunnelBuilder<R>, R: Runtime> {

    /// We found an open tunnel: return immediately.

    Open(Arc<B::Tunnel>),

    /// We found one or more pending tunnels: wait until one succeeds,

    /// or all fail.

    Wait(FuturesUnordered<Shared<oneshot::Receiver<PendResult<B, R>>>>),

    /// We should launch tunnels: here are the instructions for how

    /// to do so.

    Build(Vec<TunnelBuildPlan<B, R>>),

}

impl<B: AbstractTunnelBuilder<R> + 'static, R: Runtime> AbstractTunnelMgr<B, R> {

    /// Construct a new AbstractTunnelMgr.

    /// Reconfigure this manager using the latest set of network parameters.

    /// Return this manager's [`CircuitTiming`].

    /// Return this manager's [`CircuitTiming`].

    /// Return a circuit suitable for use with a given `usage`,

    /// creating that circuit if necessary, and restricting it

    /// under the assumption that it will be used for that spec.

    ///

    /// This is the primary entry point for AbstractTunnelMgr.

    #[allow(clippy::cognitive_complexity)] // TODO #2010: Refactor?

    #[instrument(level = "trace", skip_all)]

        /// Largest number of "resets" that we will accept in this attempt.

        ///

        /// A "reset" is an internally generated error that does not represent a

        /// real problem; only a "whoops, got to try again" kind of a situation.

        /// For example, if we reconfigure in the middle of an attempt and need

        /// to re-launch the circuit, that counts as a "reset", since there was

        /// nothing actually _wrong_ with the circuit we were building.

        ///

        /// We accept more resets than we do real failures. However,

        /// we don't accept an unlimited number: we don't want to inadvertently

        /// permit infinite loops here. If we ever bump against this limit, we

        /// should not automatically increase it: we should instead figure out

        /// why it is happening and try to make it not happen.

        const MAX_RESETS: usize = 8;

        let circuit_timing = self.circuit_timing();

        let timeout_at = self.runtime.now() + circuit_timing.request_timeout;

        let max_tries = circuit_timing.request_max_retries;

        // We compute the maximum number of failures by dividing the maximum

        // number of circuits to attempt by the number that will be launched in

        // parallel for each iteration.

        let max_failures = usize::div_ceil(

            max_tries as usize,

            std::cmp::max(1, self.builder.launch_parallelism(usage)),

        );

        let mut retry_schedule = RetryDelay::from_msec(100);

        let mut retry_err = RetryError::<Box<Error>>::in_attempt_to("find or build a tunnel");

        let mut n_failures = 0;

        let mut n_resets = 0;

        for attempt_num in 1.. {

            // How much time is remaining?

            let remaining = match timeout_at.checked_duration_since(self.runtime.now()) {

                None => {

                    retry_err.push_timed(

                        Error::RequestTimeout,

                        self.runtime.now(),

                        Some(self.runtime.wallclock()),

                    );

                    break;

                }

                Some(t) => t,

            };

            let error = match self.prepare_action(usage, dir, true) {

                Ok(action) => {

                    // We successfully found an action: Take that action.

                    let outcome = self

                        .runtime

                        .timeout(remaining, Arc::clone(self).take_action(action, usage))

                        .await;

                    match outcome {

                        Ok(Ok(circ)) => return Ok(circ),

                        Ok(Err(e)) => {

                            debug!("Circuit attempt {} failed.", attempt_num);

                            Error::RequestFailed(e)

                        }

                        Err(_) => {

                            // We ran out of "remaining" time; there is nothing

                            // more to be done.

                            warn!("All tunnel attempts failed due to timeout");

                            retry_err.push_timed(

                                Error::RequestTimeout,

                                self.runtime.now(),

                                Some(self.runtime.wallclock()),

                            );

                            break;

                        }

                    }

                }

                Err(e) => {

                    // We couldn't pick the action!

                    debug_report!(

                        &e,

                        "Couldn't pick action for tunnel attempt {}",

                        attempt_num,

                    );

                    e

                }

            };

            // There's been an error.  See how long we wait before we retry.

            let now = self.runtime.now();

            let retry_time =

            let (count, count_limit) = if error.is_internal_reset() {

                (&mut n_resets, MAX_RESETS)

            } else {

                (&mut n_failures, max_failures)

            };

            // Record the error, flattening it if needed.

            match error {

                // Flatten nested RetryError, using mockable time for each error

                Error::RequestFailed(e) => {

                    retry_err.extend_from_retry_error(e);

                }

                e => retry_err.push_timed(e, now, Some(self.runtime.wallclock())),

            }

            *count += 1;

            // If we have reached our limit of this kind of problem, we're done.

            if *count >= count_limit {

                warn!("Reached circuit build retry limit, exiting...");

                break;

            }

            // Wait, or not, as appropriate.

            match retry_time {

                AbsRetryTime::Immediate => {}

                AbsRetryTime::Never => break,

                AbsRetryTime::At(t) => {

                    let remaining = timeout_at.saturating_duration_since(now);

                    let delay = t.saturating_duration_since(now);

                    trace!(?delay, "Waiting to retry...");

                    self.runtime.sleep(std::cmp::min(delay, remaining)).await;

                }

            }

        }

        warn!("Request failed");

        Err(Error::RequestFailed(retry_err))

    /// Make sure a circuit exists, without actually asking for it.

    ///

    /// Make sure that there is a circuit (built or in-progress) that could be

    /// used for `usage`, and launch one or more circuits in a background task

    /// if there is not.

    // TODO: This should probably take some kind of parallelism parameter.

    #[cfg(test)]

    /// Choose which action we should take in order to provide a tunnel

    /// for a given `usage`.

    ///

    /// If `restrict_circ` is true, we restrict the spec of any

    /// circ we decide to use to mark that it _is_ being used for

    /// `usage`.

    #[instrument(level = "trace", skip_all)]

            // We have open tunnels that meet the spec: return the best one.

            // TODO: If we have fewer tunnels here than our select

            // parallelism, perhaps we should launch more?

            // There are pending tunnels that could meet the spec.

            // Restrict them under the assumption that they could all

            // be used for this, and then wait until one is ready (or

            // all have failed)

                    // TODO: Do we want to tentatively restrict _all_ of these?

                    // not clear to me.

                }

            // TODO: if we have fewer tunnels here than our launch

            // parallelism, we might want to launch more.

        // Okay, we need to launch tunnels here.

                }

            }

        }

        } else {

            // we didn't even try to plan anything!

        }

    /// Execute an action returned by pick-action, and return the

    /// resulting tunnel or error.

    #[allow(clippy::cognitive_complexity, clippy::type_complexity)] // TODO #2010: Refactor

    #[instrument(level = "trace", skip_all)]

        /// Store the error `err` into `retry_err`, as appropriate.

                // We don't care about this error, since it is from neither a tunnel we launched

                // nor one that we're waiting on.

        /// Return a string describing what it means, within the context of this

        /// function, to have gotten an answer from `source`.

            }

        // Get or make a stream of futures to wait on.

            Action::Open(c) => {

                // There's already a perfectly good open tunnel; we can return

                // it now.

                trace!("Returning existing tunnel.");

                return Ok((c, TunnelProvenance::Preexisting));

            }

            Action::Wait(f) => {

                // There is one or more pending tunnel that we're waiting for.

                // If any succeeds, we try to use it.  If they all fail, we

                // fail.

                trace!("Waiting for tunnel.");

                (false, f)

            }

            Action::Build(plans) => {

                // We're going to launch one or more tunnels in parallel.  We

                // report success if any succeeds, and failure of they all fail.

                trace!("Building new tunnel.");

                let futures = FuturesUnordered::new();

                for plan in plans {

                    let self_clone = Arc::clone(&self);

                    // (This is where we actually launch tunnels.)

                    futures.push(self_clone.spawn_launch(usage, plan));

                }

                (true, futures)

            }

        };

        // Insert ourself into the list of pending requests, and make a

        // stream for us to listen on for notification from pending tunnels

        // other than those we are pending on.

        let (pending_request, additional_stream) = {

            // We don't want this queue to participate in memory quota tracking.

            // There isn't any tunnel yet, so there wouldn't be anything to account it to.

            // If this queue has the oldest data, probably the whole system is badly broken.

            // Tearing down the whole tunnel manager won't help.

            let (send, recv) = mpsc_channel_no_memquota(8);

            let pending = Arc::new(PendingRequest {

                usage: usage.clone(),

                notify: send,

            });

            let mut list = self.tunnels.lock().expect("poisoned lock");

            list.add_pending_request(&pending);

            (pending, recv)

        };

        // We use our "select_biased" stream combiner here to ensure that:

        //   1) Circuits from wait_on_stream (the ones we're pending on) are

        //      preferred.

        //   2) We exit this function when those tunnels are exhausted.

        //   3) We still get notified about other tunnels that might meet our

        //      interests.

        //

        // The events from Left stream are the oes that we explicitly asked for,

        // so we'll treat errors there as real problems.  The events from the

        // Right stream are ones that we got opportunistically told about; it's

        // not a big deal if those fail.

        let mut incoming = streams::select_biased(wait_on_stream, additional_stream.map(Ok));

        let mut retry_error = RetryError::in_attempt_to("wait for tunnels");

        while let Some((src, id)) = incoming.next().await {

            match id {

                Ok(Ok(ref id)) => {

                    // Great, we have a tunnel . See if we can use it!

                    let mut list = self.tunnels.lock().expect("poisoned lock");

                    if let Some(ent) = list.get_open_mut(id) {

                        let now = self.runtime.now();

                        match ent.restrict_mut(usage, now) {

                            Ok(()) => {

                                // Great, this will work.  We drop the

                                // pending request now explicitly to remove

                                // it from the list.

                                drop(pending_request);

                                if matches!(ent.expiration, ExpirationInfo::Unused { .. }) {

                                    let try_to_expire_after = if ent.spec.is_long_lived() {

                                        self.circuit_timing().disused_circuit_timeout

                                    } else {

                                        self.circuit_timing().max_dirtiness

                                    };

                                    // Since this tunnel hasn't been used yet, schedule expiration

                                    // task after `max_dirtiness` from now.

                                    spawn_expiration_task(

                                        &self.runtime,

                                        Arc::downgrade(&self),

                                        ent.tunnel.id(),

                                        now + try_to_expire_after,

                                    );

                                }

                                return Ok((ent.tunnel.clone(), TunnelProvenance::NewlyCreated));

                            }

                            Err(e) => {