//! Abstract code to manage a set of circuits.

//!

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

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

//! which requests.

//!

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

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

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

//! crate:

//!

//!  * [`AbstractCirc`] is a view of a circuit.

//!  * [`AbstractSpec`] represents a circuit's possible usages.

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

//!

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

//! circuits, 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::{DirInfo, Error, Result};

use retry_error::RetryError;

use tor_config::MutCfg;

use tor_error::internal;

use tor_rtcompat::{Runtime, SleepProviderExt};

use async_trait::async_trait;

use futures::channel::{mpsc, oneshot};

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

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

use futures::task::SpawnExt;

use std::collections::HashMap;

use std::convert::TryInto;

use std::fmt::Debug;

use std::hash::Hash;

use std::panic::AssertUnwindSafe;

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

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

use tracing::{debug, info, warn};

use weak_table::PtrWeakHashSet;

mod streams;

/// Represents restrictions on circuit usage.

///

/// An `AbstractSpec` describes what a circuit can be used for.  Each

/// `AbstractSpec` type has an associated `Usage` type that

/// describes a _single_ operation that the circuit might support or

/// not.

///

/// (For example, an `AbstractSpec` can describe a set of ports

/// supported by the exit relay on a circuit.  In that case, its

/// `Usage` type could be a single port that a client wants to

/// connect to.)

///

/// If an `AbstractSpec` A allows every operation described in a

/// `Usage` B, we say that A "supports" B.

///

/// If one `AbstractSpec` A supports every operation supported by

/// another `AbstractSpec` B, we say that A "contains" B.

///

/// Some circuits can be used for either of two operations, but not both.

/// For example, a circuit that is used as a rendezvous point can't

/// be used as an introduction point.  To represent these transitions,

/// we use a `restrict` operation.  Every time a circuit is used for something

/// new, that new use "restricts" the circuit's spec, and narrows

/// what the circuit can be used for.

pub(crate) trait AbstractSpec: Clone + Debug {

    /// A type to represent the kind of usages that this circuit permits.

    type Usage: Clone + Debug + Send + Sync;

    /// Return true if this spec permits the usage described by `other`.

    ///

    /// If this function returns `true`, then it is okay to use a circuit

    /// with this spec for the target usage described by `other`.

    fn supports(&self, other: &Self::Usage) -> bool;

    /// Change the value of this spec based on the circuit having

    /// been used for `usage`.

    ///

    /// # Requirements

    ///

    /// Must return an error and make no changes to `self` if `usage`

    /// was not supported by this spec.

    ///

    /// If this function returns Ok, the resulting spec must be

    /// contained by the original spec, and must support `usage`.

    fn restrict_mut(&mut self, usage: &Self::Usage) -> Result<()>;

    /// Find all open circuits in `list` whose specifications permit

    /// `usage`.

    ///

    /// By default, this calls `abstract_spec_find_supported`.

}

/// Default implementation of `AbstractSpec::find_supported`; provided as a separate function

/// so it can be used in overridden implementations.

///

/// This returns the all circuits in `list` for which `circuit.spec.supports(usage)` returns

/// `true`.

/// Minimal abstract view of a circuit.

///

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

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

pub(crate) trait AbstractCirc: Clone + Debug {

    /// Type for a unique identifier for circuits.

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

    /// Return the unique identifier for this circuit.

    ///

    /// # Requirements

    ///

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

    /// circuits.

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

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

    ///

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

    fn usable(&self) -> bool;

}

/// 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 circuits.

///

/// AbstractCircBuilder creates circuits in two phases.  First, a plan is

/// made for how to build the circuit.  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 circuit will be able

/// to support when it's done.

///

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

#[async_trait]

pub(crate) trait AbstractCircBuilder: Send + Sync {

    /// The specification type describing what operations circuits can

    /// be used for.

    type Spec: AbstractSpec + Debug + Send + Sync;

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

    type Circ: AbstractCirc + Send + Sync;

    /// An opaque type describing how a given circuit 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 circuit that supports `usage`.

    ///

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

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

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

    /// port 80, then "planning" the circuit 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_circuit(

        &self,

        usage: &<Self::Spec as AbstractSpec>::Usage,

        dir: DirInfo<'_>,

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

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

    ///

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

    /// and the circuit that was just constructed.

    ///

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

    /// circuits that are taking too long.

    ///

    /// # Requirements

    ///

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

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

    /// contain the spec that was originally returned by

    /// `plan_circuit`.

    async fn build_circuit(&self, plan: Self::Plan) -> Result<(Self::Spec, Self::Circ)>;

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

    /// constructed for a given purpose.

    ///

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

    /// 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 circuits should be

    /// used for a given purpose.

    ///

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

    /// open circuits 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 circuit

    /// timeouts by building testing circuits.

    fn learning_timeouts(&self) -> bool;

}

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

///

/// A circuit 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 circuits 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.

enum ExpirationInfo {

    /// The circuit has never been used.

    Unused {

        /// A time when the circuit should expire.

        use_before: Instant,

    },

    /// The circuit has been used (or at least, restricted for use with a

    /// request) at least once.

    Dirty {

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

        dirty_since: Instant,

    },

}

impl ExpirationInfo {

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

    /// Mark this ExpirationInfo as dirty, if it is not already dirty.

}

/// An entry for an open circuit held by an `AbstractCircMgr`.

pub(crate) struct OpenEntry<S, C> {

    /// Current AbstractCircSpec for this circuit's permitted usages.

    spec: S,

    /// The circuit under management.

    circ: C,

    /// When does this circuit expire?

    ///

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

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

    /// references to them.)

    expiration: ExpirationInfo,

}

impl<S: AbstractSpec, C: AbstractCirc> OpenEntry<S, C> {

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

    /// Return true if this circuit can be used for `usage`.

    /// Change this circuit's permissible usage, based on its having

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

    ///

    /// Return an error if this circuit may not be used for `usage`.

    fn restrict_mut(&mut self, usage: &<S as AbstractSpec>::Usage, now: Instant) -> Result<()> {

    /// 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` circuits.

    ///

    /// # Requirements

    ///

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

    /// supports `spec`.

    /// Return true if this circuit has been marked as dirty before

    /// `dirty_cutoff`, or if it is an unused circuit set to expire before

    /// `unused_cutoff`.

        }

}

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

type PendResult<B> = Result<<<B as AbstractCircBuilder>::Circ as AbstractCirc>::Id>;

/// An in-progress circuit request tracked by an `AbstractCircMgr`.

///

/// (In addition to tracking circuits, `AbstractCircMgr` tracks

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

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

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

struct PendingRequest<B: AbstractCircBuilder> {

    /// Usage for the operation requested by this request

    usage: <B::Spec as AbstractSpec>::Usage,

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

    /// might like.

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

}

impl<B: AbstractCircBuilder> PendingRequest<B> {

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

}

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

/// an `AbstractCircMgr`.

struct PendingEntry<B: AbstractCircBuilder> {

    /// Specification that this circuit 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 circuit.

    ///

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

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

    tentative_assignment: sync::Mutex<B::Spec>,

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

    /// notification of this circuit's success.

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

}

impl<B: AbstractCircBuilder> PendingEntry<B> {

    /// 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 circuit.

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

    /// supports `usage`.

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

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

    ///

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

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

    fn tentative_restrict_mut(&self, usage: &<B::Spec as AbstractSpec>::Usage) -> Result<()> {

    /// 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.

}

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

/// circuit and constructing it.

struct CircBuildPlan<B: AbstractCircBuilder> {

    /// The Plan object returned by [`AbstractCircBuilder::plan_circuit`].

    plan: B::Plan,

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

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

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

    pending: Arc<PendingEntry<B>>,

}

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

struct CircList<B: AbstractCircBuilder> {

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

    /// open circuits.

    #[allow(clippy::type_complexity)]

    open_circs: HashMap<<B::Circ as AbstractCirc>::Id, OpenEntry<B::Spec, B::Circ>>,

    /// Weak-set of PendingEntry for circuits 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 circuit, this set's members are lazily removed

    /// after the circuit is either built or fails to build.

    pending_circs: PtrWeakHashSet<Weak<PendingEntry<B>>>,

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

    /// circuit 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 circuit to be built, this set's members are

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

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

}

impl<B: AbstractCircBuilder> CircList<B> {

    /// Make a new empty `CircList`

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

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

    ///

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

        } else {

        }

    /// Find an open circuit by ID.

    ///

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

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

    ///

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

    /// Remove circuits based on expiration times.

    ///

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

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

    /// since before `dirty_cutoff`.

    /// Remove the circuit with given `id`, if it is scheduled to

    /// expire now, according to the provided expiration times.

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

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

    ///

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

        } else {

        }

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

    ///

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

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

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

    /// for a circuit.

    ///

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

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

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

    /// that supports `circ_spec`.

    /// Clear all pending circuits and open circuits.

}

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

///

/// Currently taken from the network parameters.

struct UnusedTimings {

    /// Minimum lifetime of a circuit created while learning

    /// circuit timeouts.

    learning: Duration,

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

    /// circuit 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 {

}

/// Abstract implementation for circuit management.

///

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

///

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

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

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

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

/// request's needs.

///

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

/// numerical limit.

///

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

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

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

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

/// was waiting for.

pub(crate) struct AbstractCircMgr<B: AbstractCircBuilder, R: Runtime> {

    /// Builder used to construct circuits.

    builder: B,

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

    /// checking timeouts.

    runtime: R,

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

    /// pending circuits.

    circs: sync::Mutex<CircList<B>>,

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

    circuit_timing: MutCfg<CircuitTiming>,

    /// Minimum lifetime of an unused circuit.

    ///

    /// Derived from the network parameters.

    unused_timing: sync::Mutex<UnusedTimings>,

}

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

enum Action<B: AbstractCircBuilder> {

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

    Open(B::Circ),

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

    /// or all fail.

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

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

    /// to do so.

    Build(Vec<CircBuildPlan<B>>),

}

impl<B: AbstractCircBuilder + 'static, R: Runtime> AbstractCircMgr<B, R> {

    /// Construct a new AbstractCircMgr.

    /// 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 AbstractCircMgr.

            // How much time is remaining?

                None => {

                }

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

                        }

                        Err(_) => {

                        }

                    }

                }

                    // We couldn't pick the action! This is unusual; wait

                    // a little while before we try again.

                }

            };

        }

    /// 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.

    #[allow(dead_code)]

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

    /// 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`.

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

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

            // parallelism, perhaps we should launch more?

            // There are pending circuits 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.

                }

        }

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

    /// resulting circuit or error.

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

            }

            }

        };

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

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

        // other than those we are pending on.

                            Ok(()) => {

                                // Great, this will work.  We drop the

                                // pending request now explicitly to remove

                                // it from the list.

                            }

                                } else {

                                }

                            }

                        }

                }

                }

                Err(oneshot::Canceled) => {

                }

            }

            // TODO: Improve this log message; using :? here will make it

            // hard to understand.

        }

        // Nothing worked.  We drop the pending request now explicitly

        // to remove it from the list.  (We could just let it get dropped

        // implicitly, but that's a bit confusing.)

    /// Given a directory and usage, compute the necessary objects to

    /// build a circuit: A [`PendingEntry`] to keep track of the in-process

    /// circuit, and a [`CircBuildPlan`] that we'll give to the thread

    /// that will build the circuit.

    ///

    /// The caller should probably add the resulting `PendingEntry` to

    /// `self.circs`.

    ///

    /// This is an internal function that we call when we're pretty sure

    /// we want to build a circuit.

    /// Launch a managed circuit for a target usage, without checking

    /// whether one already exists or is pending.

    ///

    /// Return a listener that will be informed when the circuit is done.

    /// Spawn a background task to launch a circuit, and report its status.

    ///

    /// The `usage` argument is the usage from the original request that made

    /// us build this circuit.

                    }

                };

                // Tell anybody who was listening about it that this

                // circuit is now usable or failed.

                //

                // (We ignore any errors from `send`: That just means that nobody

                // was waiting for this circuit.)

                    // Wait briefly before we notify opportunistically.  This

                    // delay will give the circuits that were originally

                    // specifically intended for a request a little more time

                    // to finish, before we offer it this circuit instead.

                    };

    /// Run in the background to launch a circuit. Return a 2-tuple of the new

    /// circuit spec and the outcome that should be sent to the initiator.

                    } else {

                        // This circuit is no longer pending! It must have been cancelled.

                    }

                }

            }

        }

    /// Remove the circuit with a given `id` from this manager.

    ///

    /// After this function is called, that circuit will no longer be handed

    /// out to any future requests.

    ///

    /// Return None if we have no circuit with the given ID.

    /// Remove all circuits from this manager, to ensure they can't be given out for any more

    /// requests.

    /// Expire circuits according to the rules in `config` and the

    /// current time `now`.

    ///

    /// Expired circuits will not be automatically closed, but they will

    /// no longer be given out for new circuits.

    /// Consider expiring the circuit with given circuit `id`,

    /// according to the rules in `config` and the current time `now`.

    /// Return the number of open circuits held by this circuit manager.

    /// Return the number of pending circuits tracked by this circuit manager.

    #[cfg(test)]

    /// Get a reference to this manager's runtime.

    /// Get a reference to this manager's builder.

    /// Pick a duration by when a new circuit should expire from now

    /// if it has not yet been used

        } else {

            // TODO: In Tor, this calculation also depends on

            // stuff related to predicted ports and channel

            // padding.

            use rand::Rng;

        }

}

/// Spawn an expiration task that expires a circuit at given instant.

///

/// If given instant is earlier than now, expire the circuit immediately.

/// Otherwise, spawn a timer expiration task on given runtime.

///

/// When the timeout occurs, if the circuit manager is still present,

/// the task will ask the manager to expire the circuit, if the circuit

/// is ready to expire.

        } else {

            // Circuits manager has already been dropped, so are the references it held.

        };

    } else {

        // Spawn a timer expiration task with given expiration instant.

            } else {

            };

    }

#[cfg(test)]

mod test {

    #![allow(clippy::unwrap_used)]

    use super::*;

    use crate::usage::{ExitPolicy, SupportedCircUsage};

    use crate::{Error, StreamIsolation, TargetCircUsage, TargetPort};

    use std::collections::BTreeSet;

    use std::sync::atomic::{self, AtomicUsize};

    use tor_error::bad_api_usage;

    use tor_netdir::testnet;

    use tor_rtcompat::SleepProvider;

    use tor_rtmock::MockSleepRuntime;

    use tracing::trace;

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

    struct FakeId {

        id: usize,

    }

    static NEXT_FAKE_ID: AtomicUsize = AtomicUsize::new(0);

    impl FakeId {

        fn next() -> Self {

            let id = NEXT_FAKE_ID.fetch_add(1, atomic::Ordering::SeqCst);

            FakeId { id }

        }

    }

    #[derive(Debug, PartialEq, Clone)]

    struct FakeCirc {

        id: FakeId,

    }

    impl FakeCirc {

        fn eq(&self, other: &Self) -> bool {